refactor: organize existing storage migrations
26 unmodified lines27282930313233343527 unmodified lines636465666667686970712 unmodified lines26 unmodified lines
### `src/` — the daemon (Gleam)
- **`src/albedo/daemon/`** — session actors and durable state: submission/run/turn plumbing (`session_*.gleam`), the append-only transcript and conversation store (SQLite via sqlight), the event bus, projections, the family (agents, mail, bus), folders, quota, usage, the reaper, the image store, and `server.gleam`, the HTTP entry point.- **`src/albedo/daemon/migrations/`** — existing SQLite upgrades, with ordered startup data steps in `daemon/migrations.gleam`; schema additions remain called from owning initialisers (see `robot-docs/migrations.md`).- **`src/albedo/harness/`** — the model-facing half: `loop.gleam` (the tool loop), `tool.gleam`, `extension.gleam` and `extensions.gleam` (plugin interfaces and composition), `compaction.gleam`, `runtime.gleam` (boots the built-ins), plus credentials, oauth, rotation, settings, and the usage feed.- **`src/albedo/harness/extensions/`** — one directory per extension. Enabled by default: `python`, `run`, `work`, `mail`, `agents`, `schedule`, `paperclips`, `files`, `memory`, `instructions`, `commands`, `skills`, `models`, `openai`, `codex`, `antigravity`, `alibaba`, `claude`, `rolling`, `snapcompact-memory`, `lcm-memory`, `remote`. Installed but off until enabled: `mcp`, `view`, `proxy`, `webhooks`, `warm`, `snapcompact`, `lcm`. The provider extensions are ordered before `models` on purpose (their own model lists win), and the memory/compaction pairs are ordered so archives cover past folds.- **`src/albedo/openai_api/`** — the provider transport every extension shares: request/turn types, SSE framing, stream reducers, replay after restarts, chat/responses encodings. A provider with a foreign wire format encodes an `openai_api.Exchange` and supplies only its own reducer (antigravity does this); HTTP, retries, limits, and callbacks stay shared.
27 unmodified lines- **`native/render/`** — optional Rust renderer behind the `view` extension, which shows the model its changes as highlighted images for the final review pass. Install with `native/render/install.sh`; needs cargo.
### robot-docs/
`robot-docs/` is one model-facing doc per subsystem (agents, auth, cache-ttl, cache-warming, commands, compaction, context, extensions, files, mcp, models, paperclips, provider-requests, proxy, quota, skills, usage-feed, webhooks, workspaces). This file is the map; those are the details. When you change a subsystem's behavior, update its doc in the same change; when you need to understand one, read it there first.`robot-docs/` is one model-facing doc per subsystem (agents, auth, migrations, cache-ttl, cache-warming, commands, compaction, context, extensions, files, mcp, models, paperclips, provider-requests, proxy, quota, skills, usage-feed, webhooks, workspaces). This file is the map; those are the details. When you change a subsystem's behavior, update its doc in the same change; when you need to understand one, read it there first.
### Other directories
- **`test/`** — `e2e/` (default suite, shared harness), `harness/` (invariant and logic unit tests, Gleam and Python), `daemon/` (Gleam daemon suites), `openai_api/` (transport), `manual/` (opt-in)2 unmodified lines123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990# existing storage migrations
this is an organization of the migrations already shipped, not a new migrationframework or schema version. the implementations live under`src/albedo/daemon/migrations/`; `src/albedo/daemon/migrations.gleam` is theexplicit ordered entrypoint for startup data upgrades.
## startup order
`server.prepare_storage` still runs before sessions start:
1. `conversation.initialise` creates the core tables, then calls `conversation_columns.apply`: the existing session columns first, transcript columns second. it then runs its existing domain-local session recovery (missing activity/title from transcript), then creates `sessions_activity`.2. quota, mail, and family initialise their tables, unchanged.3. `migrations.run(ledger, backup)` runs `image_store.run`, then `cell_images.run`, returning `#(transcript_rows_moved, cell_results_moved)` for the existing server log messages. `image_store.run` first externalizes legacy transcript images, then converts image-table TEXT data to decoded BLOB.4. the existing best-effort filesystem credential migration runs; then provider assignment from legacy configuration runs, unchanged.
schema creation is still owned by each subsystem. work and paperclips upgradeonly when their extension initialisers run, not unconditionally at core startup:`work_cwd.apply` checks `PRAGMA table_info(work)`, adds the existing reserved`__albedo_legacy__` cwd default when absent, then creates `work_cwd_id`;`paperclips_title.apply` adds the existing non-null empty-default title column.these calls remain after each owner's table creation.
## compatibility and interruption
- the existing `migrations(name,applied_at)` table remains unchanged. the exact markers are `image_store` and `cell_images`, recorded with `unixepoch()`. transcript migration skips when its marker exists; cell migration skips when its marker exists, and does nothing without a cells table.- TEXT-to-BLOB migration deliberately has no marker. it checks `typeof(data)='text'` on every run and converts pending rows, even when the `image_store` marker already exists. invalid legacy base64 fails without replacing that row.- transcript and cell scans retain their original cursor order and 16-row pages; BLOB conversion retains its 16-row limit. each page uses the existing transaction helper. there is no new all-migrations transaction or transaction policy change. markers are written at the original completion points, and interrupted runs recheck remaining rows.- `backup.gleam` preserves the two existing backup call patterns: transcript and BLOB migration check for an existing file before directory creation and `VACUUM INTO`, retaining the `image store backup failed: ` error prefix; cell migration ensures the directory first and copies on the first page containing inline images, outside its transaction. the server still supplies one shared `home/backups/albedo-before-image-store-<timestamp>.sqlite` path. no backup is overwritten and backup triggering is unchanged (the transcript candidate check can include tool outputs without inline images).- live writes and transcript migration share `image_payloads.insert`, the unchanged decoded-BLOB insertion helper. image hashes, serialized formats, image validation, and legacy inline fallback reads are unchanged. the existing Erlang conversion and cell packing functions remain the FFI implementation.
## migration inventory
| previous implementation | current implementation || --- | --- || `conversation.initialise`: session/transcript `add_columns` | `migrations/conversation_columns.apply` || `images.migrate` / `migrate_legacy` / transcript pages | `migrations/image_store.run` / `migrate_legacy` / transcript pages || `images.migrate_blobs` / BLOB pages | `migrations/image_store.migrate_blobs` / BLOB pages || `images.backup_before_migration` | `migrations/backup.image_store` || `python/cells.migrate_images` / cell pages | `migrations/cell_images.run` / cell pages || `python/cells.backup_before_migration` | `migrations/backup.cell_images` || `work/ledger.initialise`: cwd alteration and index | `migrations/work_cwd.apply` || `paperclips/ledger.initialise`: title addition | `migrations/paperclips_title.apply` |
`images.migrate` and `cells.migrate_images` are removed rather than wrapped:wrappers would introduce import cycles between image/cell domain modules andmigration modules. direct callers use the migration modules instead.
## intentionally not extracted
- session recovery stays in `conversation`: it uses the domain's transcript decoding/title helpers and still runs at exactly the same point. core schema bootstrap and index creation remain with their owners.- rolling compaction's legacy item-count-to-user-cut conversion remains lazy in `rolling/extension.resume`. it needs the current prepared history and source, validates the old fingerprint (including legacy image payload fingerprints), and only writes a cut when the history matches and its tail begins with a user message. treating this as an unconditional startup migration would change its timing and semantics; its projections are untouched.- credentials migration is filesystem/auth, not SQLite. legacy configuration's provider assignment likewise remains in the existing server startup path.- no zstd, compression, reference index, schema-version, offline storage upgrade, new dependency, or unrelated transaction fix is part of this refactor.123456789134 unmodified linesExpand all14414514614714714814915015115215215315415515615717 unmodified lines1751761771781781791801811821831 unmodified line1851861871881881891901911921937 unmodified lines20120220320420420520620720820923 unmodified lines/// Image deduplication and legacy migration need database fixtures absent from E2E.import albedo/daemon/conversationimport albedo/daemon/imageimport albedo/daemon/imagesimport albedo/daemon/migrations/image_storeimport albedo/daemon/storeimport albedo/harness/runtimeimport albedo/openai_api/typesimport gleam/dynamic/decode134 unmodified linesExpand all let assert [types.UserImage("old", before)] = loaded(ledger, "s") types.image_data(before) |> should.equal(types.InlineData(png))
let backup = path <> ".backup/before.sqlite" images.migrate(ledger, backup) |> should.equal(Ok(1)) image_store.run(ledger, backup) |> should.equal(Ok(1)) exists(backup) |> should.be_true let assert [types.UserImage("old", after)] = loaded(ledger, "s") let assert types.StoredData(read: read, ..) = types.image_data(after) read() |> should.equal(Ok(png)) images.migrate(ledger, backup) |> should.equal(Ok(0)) image_store.run(ledger, backup) |> should.equal(Ok(0)) cleanup(path)}
pub fn migration_converts_text_images_without_changing_references_test() {17 unmodified lines |> should.equal(1) images.reader(ledger)(hash) |> should.equal(Ok(png))
let backup = path <> ".backup/before-blob.sqlite" images.migrate(ledger, backup) |> should.equal(Ok(0)) image_store.run(ledger, backup) |> should.equal(Ok(0)) exists(backup) |> should.be_true count(ledger, "SELECT count(*) FROM images WHERE typeof(data)='blob'") |> should.equal(1) count(ledger, "SELECT length(data) FROM images") |> should.equal(24)1 unmodified line let assert types.StoredData(migrated_hash, _, read) = types.image_data(migrated) migrated_hash |> should.equal(hash) read() |> should.equal(Ok(png)) images.migrate(ledger, backup) |> should.equal(Ok(0)) image_store.run(ledger, backup) |> should.equal(Ok(0)) cleanup(path)}
pub fn failed_backup_keeps_text_images_untouched_test() {7 unmodified lines conversation.Idle, ) let assert Ok(_) = store.write(ledger, "UPDATE images SET data=?", [sqlight.text(png)]) let assert Error(_) = images.migrate(ledger, path <> "/backup.sqlite") let assert Error(_) = image_store.run(ledger, path <> "/backup.sqlite") count(ledger, "SELECT count(*) FROM images WHERE typeof(data)='text'") |> should.equal(1) cleanup(path)}23 unmodified lines607 unmodified linesExpand all6086096106116126126136146156166177 unmodified lines62562662762862963063163263363499 unmodified lines607 unmodified linesExpand all raise AssertionError("second concurrent E2E daemon started") _daemon_pid = self.connection["pid"] daemon_boots += 1
def restart(self, *, crash=False): def restart(self, *, crash=False, prepare=None): """Restart the shared daemon; optionally prepare offline storage fixtures.""" global restart_seconds started = time.monotonic() if crash: os.kill(self.connection["pid"], signal.SIGKILL)7 unmodified lines break time.sleep(0.05) else: self._fail("daemon did not stop") if prepare is not None: prepare(self) self.cli("sessions") self._connection_refresh() restart_seconds += time.monotonic() - started
99 unmodified lines123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179"""Existing image migrations survive real startup, paging, backups, and restarts.
Legacy ETF fixtures cannot be written by the current daemon. They are seededwhile the shared daemon is stopped, then exercised through its normal readers."""
import base64import hashlibimport jsonimport sqlite3import unittest
from harness import Albedo, Provider, exclusive, python, text
PNG = "iVBORw0KGgoAAAANSUhEUgAAAA0AAAAD"# term_to_binary({1, {user_image, <<"legacy image">>, Image}}), where Image# is the legacy {image, <<"image/png">>, Base64, 13, 3, 24} tuple.LEGACY_TRANSCRIPT = base64.b64decode( "g2gCYQFoA3cKdXNlcl9pbWFnZW0AAAAMbGVnYWN5IGltYWdlaAZ3BWltYWdlbQAAAAlp" "bWFnZS9wbmdtAAAAIGlWQk9SdzBLR2dvQUFBQU5TVWhFVWdBQUFBMEFBQUFEYQ1hA2EY")# A successful outcome before duration was added, carrying the same inline image.LEGACY_CELL = base64.b64decode( "g2gCYQFoAncCb2toCHcHb3V0Y29tZW0AAAAObGVnYWN5LWNlbGwtMDB3CXN1Y2NlZWRl" "ZG0AAAANbGVnYWN5IG91dHB1dG0AAAAIJ2xlZ2FjeSd3BWZhbHNlbAAAAAFoBncFaW1h" "Z2VtAAAACWltYWdlL3BuZ20AAAAgaVZCT1J3MEtHZ29BQUFBTlNVaEVVZ0FBQUEwQUFB" "QURhDWEDYRhqag==")
def legacy_cell(cell_id): # Replace only this fixture's ETF binary field, including its length. original = b"legacy-cell-00" encoded = cell_id.encode() return LEGACY_CELL.replace( b"m" + len(original).to_bytes(4, "big") + original, b"m" + len(encoded).to_bytes(4, "big") + encoded, )
@exclusiveclass MigrationsTest(unittest.TestCase): def test_startup_migrates_legacy_images_once_with_a_consistent_backup(self): probe = ""
def reply(request): if request["messages"][-1].get("role") == "user" and probe: return python(probe) return text("done")
provider = Provider(reply) self.addCleanup(provider.close) with Albedo(provider) as app: session = app.session() database = app.home / "albedo.sqlite" image_hash = hashlib.sha256(PNG.encode()).hexdigest() cell_ids = [f"{session}-legacy-{n}" for n in range(20)] seqs = [] backups_before = set( app.home.glob("backups/albedo-before-image-store-*.sqlite") )
def prepare(_app): with sqlite3.connect(database) as db: db.execute( "DELETE FROM migrations WHERE name IN ('image_store','cell_images')" ) db.execute( "INSERT INTO images(hash,data) VALUES(?,?) " "ON CONFLICT(hash) DO UPDATE SET data=excluded.data", (image_hash, PNG), ) for _ in range(20): row = db.execute( "INSERT INTO transcript(session,payload) VALUES(?,?)", (session, LEGACY_TRANSCRIPT), ) seqs.append(row.lastrowid) for cell_id in cell_ids: db.execute( "INSERT INTO cells(id,session,source,status,started,payload) " "VALUES(?,?,'legacy source','finished',1,?)", (cell_id, session, legacy_cell(cell_id)), )
app.restart(prepare=prepare) backups = set(app.home.glob("backups/albedo-before-image-store-*.sqlite")) created = backups - backups_before self.assertEqual(len(created), 1) with sqlite3.connect(next(iter(created))) as backup: self.assertEqual( backup.execute("PRAGMA integrity_check").fetchone()[0], "ok" ) self.assertEqual( backup.execute( "SELECT data,typeof(data) FROM images WHERE hash=?", (image_hash,), ).fetchone(), (PNG, "text"), ) self.assertEqual( backup.execute( "SELECT payload FROM transcript WHERE seq=?", (seqs[0],) ).fetchone()[0], LEGACY_TRANSCRIPT, ) self.assertEqual( backup.execute( "SELECT payload FROM cells WHERE id=?", (cell_ids[0],) ).fetchone()[0], legacy_cell(cell_ids[0]), )
def migrated_state(): with sqlite3.connect(database) as db: self.assertEqual( db.execute("PRAGMA integrity_check").fetchone()[0], "ok" ) self.assertEqual( db.execute("PRAGMA foreign_key_check").fetchall(), [] ) self.assertEqual( db.execute( "SELECT data,typeof(data) FROM images WHERE hash=?", (image_hash,), ).fetchone(), (base64.b64decode(PNG), "blob"), ) transcript = db.execute( "SELECT seq,payload FROM transcript WHERE session=? ORDER BY seq", (session,), ).fetchall() self.assertEqual([row[0] for row in transcript], seqs) cells = db.execute( "SELECT id,source,payload FROM cells WHERE session=? ORDER BY id", (session,), ).fetchall() self.assertEqual([row[0] for row in cells], sorted(cell_ids)) self.assertTrue(all(row[1] == "legacy source" for row in cells)) for payload in [row[1] for row in transcript] + [ row[2] for row in cells ]: self.assertNotIn(PNG.encode(), payload) self.assertIn(image_hash.encode(), payload) markers = db.execute( "SELECT name,applied_at FROM migrations " "WHERE name IN ('image_store','cell_images') ORDER BY name" ).fetchall() self.assertEqual( [row[0] for row in markers], ["cell_images", "image_store"] ) return transcript, cells, markers
first = migrated_state() app.restart() self.assertEqual(migrated_state(), first) self.assertEqual( set(app.home.glob("backups/albedo-before-image-store-*.sqlite")), backups, ) # A normal provider request resolves the migrated transcript images. probe = ( f"info = await cells.info({cell_ids[0]!r})\n" "assert info['status'] == 'ok' and info['started']\n" "assert info['duration'] is None\n" f"assert 'legacy source' in await cells.read({cell_ids[0]!r})\n" "print('legacy cell readable')" ) app.prompt(session, "read the migrated history and cell").close() app.idle(session) serialized = json.dumps(provider.requests[-1]["request"]) self.assertIn("legacy image", serialized) self.assertIn(PNG, serialized) self.assertIn("legacy cell readable", serialized)
if __name__ == "__main__": unittest.main()123456789114 unmodified linesExpand all12412512612712712812913013113213313413513613613713713813914014114237 unmodified lines//// Kernel image wire decoding and old journal records must survive format//// changes; synthetic corrupted and legacy records cannot arise in E2E.
import albedo/daemon/imagesimport albedo/daemon/migrations/cell_imagesimport albedo/daemon/storeimport albedo/harness/extensions/python/cellsimport albedo/harness/extensions/python/kernel as pythonimport albedo/openai_api/types114 unmodified linesExpand all/// Legacy cell records are not producible by the current daemon wire format.pub fn old_cell_images_migrate_without_losing_their_readers_test() { let path = temporary_database() let assert Ok(ledger) = store.start(path, images.schema) cells.migrate_images(ledger, path <> ".missing-backup") |> should.equal(Ok(0)) cell_images.run(ledger, path <> ".missing-backup") |> should.equal(Ok(0)) let assert Ok(_) = cells.initialise(ledger) let assert Ok(_) = store.write( ledger, "INSERT INTO cells(id,session,source,status,payload) VALUES('cell','s','show_image(...)','finished',?)", [sqlight.blob(journaled_png(2))], ) let backup = path <> ".backup" cells.migrate_images(ledger, backup) |> should.equal(Ok(1)) cells.migrate_images(ledger, backup) |> should.equal(Ok(0)) cell_images.run(ledger, backup) |> should.equal(Ok(1)) cell_images.run(ledger, backup) |> should.equal(Ok(0)) let assert Ok(cell) = cells.get(ledger, "cell") let assert Some(Ok(outcome)) = cell.outcome let assert [stored] = outcome.images let assert types.StoredData(read: read, ..) = types.image_data(stored)37 unmodified lines1234567867 unmodified lines7677787979808182838485868788899091929394959697989910010110210310410510610710880818283841075 unmodified linesExpand allimport albedo/daemon/eventsimport albedo/daemon/imagesimport albedo/daemon/mailimport albedo/daemon/migrations/conversation_columnsimport albedo/daemon/noteimport albedo/daemon/noticeimport albedo/daemon/requestsimport albedo/daemon/store67 unmodified lines "CREATE TABLE IF NOT EXISTS sessions(id TEXT PRIMARY KEY,title TEXT NOT NULL DEFAULT 'new session',cwd TEXT NOT NULL,model TEXT NOT NULL,protocol TEXT NOT NULL,stage TEXT NOT NULL DEFAULT 'idle',provider TEXT,activity_seq INTEGER,last_assistant_at INTEGER,usage_model TEXT,usage_recorded_at INTEGER,usage_prompt_tokens INTEGER,usage_completion_tokens INTEGER,usage_cached_prompt_tokens INTEGER,usage_cache_creation_tokens INTEGER,effort TEXT); CREATE TABLE IF NOT EXISTS transcript(seq INTEGER PRIMARY KEY AUTOINCREMENT,session TEXT NOT NULL REFERENCES sessions(id),payload BLOB NOT NULL,timestamp INTEGER,provider TEXT,thought_ms INTEGER); CREATE INDEX IF NOT EXISTS transcript_session ON transcript(session,seq);" <> images.schema <> requests.schema, )) use _ <- result.try( store.add_columns(db, "sessions", [ #("provider", "TEXT"), #("title", "TEXT"), #("activity_seq", "INTEGER"), #("last_assistant_at", "INTEGER"), #("usage_model", "TEXT"), #("usage_recorded_at", "INTEGER"), #("usage_prompt_tokens", "INTEGER"), #("usage_completion_tokens", "INTEGER"), #("usage_cached_prompt_tokens", "INTEGER"), #("usage_cache_creation_tokens", "INTEGER"), #("usage_cache_write_5m_tokens", "INTEGER"), #("usage_cache_write_1h_tokens", "INTEGER"), #("usage_reasoning_tokens", "INTEGER"), #("usage_cache", "TEXT"), #("effort", "TEXT"), #("pinned_instructions", "TEXT"), #("pinned_context", "BLOB"), #("pinned_head", "INTEGER"), #("name", "TEXT"), ]), ) use _ <- result.try( store.add_columns(db, "transcript", [ #("timestamp", "INTEGER"), #("provider", "TEXT"), #("thought_ms", "INTEGER"), ]), ) use _ <- result.try(conversation_columns.apply(db)) use _ <- result.try(recover_sessions(db)) store.exec( db, "CREATE INDEX IF NOT EXISTS sessions_activity ON sessions(activity_seq DESC)",1075 unmodified linesExpand all1234567891011121314151617181920//// Shared payload insertion for live writes and legacy transcript migration.
import albedo/daemon/storeimport gleam/listimport sqlight
pub fn insert( db: sqlight.Connection, blobs: List(#(String, String)),) -> Result(Nil, String) { list.try_each(blobs, fn(blob) { store.run(db, "INSERT OR IGNORE INTO images(hash,data) VALUES(?,?)", [ sqlight.text(blob.0), sqlight.blob(decode_base64(blob.1)), ]) })}
@external(erlang, "albedo_images", "decode_base64")fn decode_base64(data: String) -> BitArray3 unmodified lines45678910111242 unmodified lines555657585859606162633 unmodified lines676869707071727374757576777879808182838485867677787934 unmodified lines114115116117129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251118119120121111 unmodified linesExpand all2332342352363713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234242372382392401 unmodified line2422432442454344354364374384394404414424434444454464474484494504512462473 unmodified lines//// in `images`. Loading a session therefore reads references, and a payload is//// read only while a request body is written (albedo_openai_transport.erl).//// Rows are content-addressed, so forks and repeated screenshots share one copy.
import albedo/daemon/image_payloadsimport albedo/daemon/storeimport albedo/openai_api/typesimport gleam/dynamic/decodeimport gleam/list42 unmodified lines input: types.Input, read: fn(String) -> Result(String, Nil),) -> Result(types.Input, String) { let #(stored, blobs) = split(input, read) use _ <- result.try(insert(db, blobs)) use _ <- result.try(image_payloads.insert(db, blobs)) Ok(stored)}
/// Stores a cell's inline images in the caller's transaction. The returned3 unmodified lines images: List(types.Image),) -> Result(List(types.Image), String) { let #(stored, blobs) = split(types.ToolOutput("", "", images), fn(_) { Error(Nil) }) use _ <- result.try(insert(db, blobs)) use _ <- result.try(image_payloads.insert(db, blobs)) let assert types.ToolOutput(_, _, images) = stored Ok(images)}
fn insert( db: sqlight.Connection, blobs: List(#(String, String)),) -> Result(Nil, String) { list.try_each(blobs, fn(blob) { store.run(db, "INSERT OR IGNORE INTO images(hash,data) VALUES(?,?)", [ sqlight.text(blob.0), sqlight.blob(decode_base64(blob.1)), ]) })}
/// Hashes the given session's rows reference, read before those rows are/// deleted so `release` can drop payloads nothing else needs.pub fn session_hashes( db: sqlight.Connection,34 unmodified lines ) })}
/// Moves inline images in existing transcript rows into `images`, once per/// database. The database is first copied to `backup` with VACUUM INTO. Rows/// are rewritten a page at a time, each page in its own transaction, so an/// interrupted run resumes where it stopped (a rewritten row has nothing inline).pub fn migrate(ledger: store.Store, backup: String) -> Result(Int, String) { use moved <- result.try(migrate_legacy(ledger, backup)) use _ <- result.try(migrate_blobs(ledger, backup)) Ok(moved)}
fn migrate_legacy(ledger: store.Store, backup: String) -> Result(Int, String) { let read = reader(ledger) use applied <- result.try( store.read( ledger, "SELECT 1 FROM migrations WHERE name='image_store'", [], decode.dynamic, ) |> result.map(fn(rows) { rows != [] }), ) case applied { True -> Ok(0) False -> { use pending <- result.try(store.read( ledger, "SELECT count(*) FROM transcript WHERE instr(payload,CAST('user_image' AS BLOB))>0 OR instr(payload,CAST('tool_output' AS BLOB))>0", [], decode.field(0, decode.int, decode.success), )) use _ <- result.try(case pending { [0] -> Ok(Nil) _ -> { backup_before_migration(ledger, backup) } }) use moved <- result.try(migrate_pages(ledger, read, -1, 0)) use _ <- result.try( store.write( ledger, "INSERT OR IGNORE INTO migrations(name,applied_at) VALUES('image_store',unixepoch())", [], ), ) Ok(moved) } }}
/// Existing installations used TEXT for the base64 data. Convert in bounded/// transactions after taking a copy of the database. SQLite's BLOB affinity/// does not convert an existing TEXT value on its own.fn migrate_blobs(ledger: store.Store, backup: String) -> Result(Nil, String) { use pending <- result.try(store.read( ledger, "SELECT count(*) FROM images WHERE typeof(data)='text'", [], decode.field(0, decode.int, decode.success), )) case pending { [0] -> Ok(Nil) _ -> { use _ <- result.try(backup_before_migration(ledger, backup)) migrate_blob_pages(ledger) } }}
fn backup_before_migration( ledger: store.Store, backup: String,) -> Result(Nil, String) { case backup_exists(backup) { True -> Ok(Nil) False -> { ensure_dir(backup) store.query(ledger, fn(db) { store.run(db, "VACUUM INTO ?", [sqlight.text(backup)]) }) |> result.map_error(fn(e) { "image store backup failed: " <> e }) } }}
fn migrate_blob_pages(ledger: store.Store) -> Result(Nil, String) { let page = store.query(ledger, fn(db) { use rows <- result.try( store.rows( db, "SELECT hash,data FROM images WHERE typeof(data)='text' LIMIT ?", [sqlight.int(migrate_page_rows)], { use hash <- decode.field(0, decode.string) use data <- decode.field(1, decode.string) decode.success(#(hash, data)) }, ), ) use _ <- result.try( store.transaction(db, fn() { list.try_each(rows, fn(row) { case decode_legacy_base64(row.1) { Ok(bytes) -> store.run( db, "UPDATE images SET data=? WHERE hash=? AND typeof(data)='text'", [sqlight.blob(bytes), sqlight.text(row.0)], ) Error(_) -> Error("invalid base64 image at hash " <> row.0) } }) }), ) Ok(list.length(rows)) }) case page { Ok(0) -> Ok(Nil) Ok(_) -> migrate_blob_pages(ledger) Error(e) -> Error("image blob migration failed: " <> e) }}
/// Drops the images of tool outputs a committed compaction left out of the/// prepared request: the row and its Python cell keep their text plus a/// marker, and payloads nothing else names are released. User uploads stay;/// their hashes are part of the saved compaction cut.111 unmodified linesExpand all
@external(erlang, "albedo_native", "elide_cell_images")fn elide_cell_images(payload: BitArray) -> Result(BitArray, Nil)
/// Rows per migration transaction; a page of screenshot rows is tens of MB.const migrate_page_rows = 16
fn migrate_pages( ledger: store.Store, read: fn(String) -> Result(String, Nil), after: Int, moved: Int,) -> Result(Int, String) { let page = store.query(ledger, fn(db) { use rows <- result.try( store.rows( db, "SELECT seq,payload FROM transcript WHERE seq>? AND (instr(payload,CAST('user_image' AS BLOB))>0 OR instr(payload,CAST('tool_output' AS BLOB))>0) ORDER BY seq LIMIT ?", [sqlight.int(after), sqlight.int(migrate_page_rows)], { use seq <- decode.field(0, decode.int) use payload <- decode.field(1, decode.bit_array) decode.success(#(seq, payload)) }, ), ) use count <- result.try( store.transaction(db, fn() { list.try_fold(rows, 0, fn(count, row) { case migrate_row(row.1, read) { Keep -> Ok(count) Rewrite(payload, blobs) -> { use _ <- result.try(insert(db, blobs)) store.run(db, "UPDATE transcript SET payload=? WHERE seq=?", [ sqlight.blob(payload), sqlight.int(row.0), ]) |> result.replace(count + 1) } } }) }), ) Ok(#(count, list.last(rows) |> result.map(fn(row) { row.0 }))) }) case page { Error(e) -> Error("image store migration failed: " <> e) Ok(#(count, Ok(last))) -> migrate_pages(ledger, read, last, moved + count) Ok(#(count, Error(_))) -> Ok(moved + count) }}
type Migration { Keep Rewrite(payload: BitArray, blobs: List(#(String, String)))}
@external(erlang, "albedo_images", "externalize")fn split( input: types.Input, read: fn(String) -> Result(String, Nil),1 unmodified line
@external(erlang, "albedo_images", "hashes")fn hashes(payload: BitArray) -> List(String)
@external(erlang, "albedo_images", "migrate")fn migrate_row( payload: BitArray, read: fn(String) -> Result(String, Nil),) -> Migration
@external(erlang, "albedo_images", "ensure_dir")fn ensure_dir(path: String) -> Nil
@external(erlang, "albedo_images", "backup_exists")fn backup_exists(path: String) -> Bool
@external(erlang, "albedo_images", "decode_base64")fn decode_base64(data: String) -> BitArray
@external(erlang, "albedo_images", "decode_legacy_base64")fn decode_legacy_base64(data: String) -> Result(BitArray, Nil)
@external(erlang, "albedo_images", "encode_base64")fn encode_base64(data: BitArray) -> String123456789101112131415//// Ordered data upgrades after core schema initialisation, before sessions start.//// Schema column upgrades remain in their owning initialisers.
import albedo/daemon/migrations/cell_imagesimport albedo/daemon/migrations/image_storeimport albedo/daemon/storeimport gleam/result
/// Transcript references, image TEXT to BLOB, then cell references. Both image/// steps share the startup backup path. Returns rewritten transcript/cell counts.pub fn run(ledger: store.Store, backup: String) -> Result(#(Int, Int), String) { use transcript_moved <- result.try(image_store.run(ledger, backup)) use cells_moved <- result.try(cell_images.run(ledger, backup)) Ok(#(transcript_moved, cells_moved))}4 unmodified lines56789910111213143 unmodified lines181920212222232425534 unmodified linesExpand all5605615625635655665645655665675681853 unmodified linesExpand all4 unmodified linesimport albedo/daemon/familyimport albedo/daemon/foldersimport albedo/daemon/historyimport albedo/daemon/imageimport albedo/daemon/imagesimport albedo/daemon/mailimport albedo/daemon/migrationsimport albedo/daemon/quotaimport albedo/daemon/reaperimport albedo/daemon/requestsimport albedo/daemon/session3 unmodified linesimport albedo/harness/cache_ttlimport albedo/harness/commandimport albedo/harness/credentialsimport albedo/harness/extensionimport albedo/harness/extensions/python/cells as journalimport albedo/harness/extensions/schedule/ledger as scheduleimport albedo/harness/oauthimport albedo/harness/runtimeimport albedo/harness/usage_feed534 unmodified linesExpand all config.home <> "/backups/albedo-before-image-store-" <> int.to_string(usage.now()) <> ".sqlite" use moved <- result.try(images.migrate(runtime.ledger(host), backup)) use cells_moved <- result.try(journal.migrate_images( use #(moved, cells_moved) <- result.try(migrations.run( runtime.ledger(host), backup, )) case cells_moved {1853 unmodified linesExpand all