fix(session): shard-aware getPart + sweep marker to skip redundant seed

Author: sjawhar

packages/opencode/src/session/index.ts

@@ -476,19 +476,17 @@ export namespace Session {
         }).pipe(Effect.withSpan("Session.updatePart"))
 
       const getPart: Interface["getPart"] = Effect.fn("Session.getPart")(function* (input) {
-        const row = Database.use((db) =>
-          db
-            .select()
-            .from(PartTable)
-            .where(
-              and(
-                eq(PartTable.session_id, input.sessionID),
-                eq(PartTable.message_id, input.messageID),
-                eq(PartTable.id, input.partID),
-              ),
-            )
-            .get(),
-        )
+        const row = Database.resolveSession(input.sessionID)
+          .select()
+          .from(PartTable)
+          .where(
+            and(
+              eq(PartTable.session_id, input.sessionID),
+              eq(PartTable.message_id, input.messageID),
+              eq(PartTable.id, input.partID),
+            ),
+          )
+          .get()
         if (!row) return
         return {
           ...row.data,

packages/opencode/src/storage/db.ts

@@ -69,6 +69,7 @@ export namespace Database {
 
   const cache = new Map<string, Entry>()
   const migrating = new Set<string>()
+  const swept = new Set<string>()
 
   const schema = [
     `CREATE TABLE IF NOT EXISTS message (
@@ -113,6 +114,7 @@ export namespace Database {
       origin TEXT
     )`,
   ]
+  const meta = `CREATE TABLE IF NOT EXISTS _meta (key TEXT PRIMARY KEY, value TEXT)`
 
   const tables = ["message", "part", "todo", "event_sequence", "event"]
 
@@ -325,6 +327,7 @@ export namespace Database {
 
     if (fresh) {
       for (const sql of schema) db.run(sql)
+      db.run(meta)
     }
 
     cache.set(id, { db, at: Date.now() })
@@ -444,6 +447,9 @@ export namespace Database {
         }
 
         dst.exec("COMMIT")
+        dst.run(meta)
+        const max = src.query("SELECT MAX(time_created) as t FROM message WHERE session_id = ?").get(root) as { t: number | null } | null
+        dst.query("INSERT OR REPLACE INTO _meta (key, value) VALUES ('swept', ?)").run(String(max?.t ?? 0))
         log.info("migrate.done", { root, sessions: ids.length })
       } catch (err) {
         dst.exec("ROLLBACK")
@@ -456,7 +462,17 @@ export namespace Database {
 
   export function ensureShard(id: string): string | undefined {
     const root = sessionRoot(id)
-    if (root) return root
+    if (root) {
+      if (!swept.has(root)) {
+        swept.add(root)
+        const db = session(root).$client
+        db.run(meta)
+        const marker = db.query("SELECT value FROM _meta WHERE key = 'swept'").get() as { value: string } | null
+        const max = Client().$client.query("SELECT MAX(time_created) as t FROM message WHERE session_id = ? LIMIT 1").get(id) as { t: number | null } | null
+        if (!marker || String(max?.t ?? 0) !== marker.value) seed(root)
+      }
+      return root
+    }
 
     const target = findRoot(id)
     if (!target) return undefined
@@ -478,6 +494,10 @@ export namespace Database {
     cache.delete(id)
   }
 
+  export function resetSwept() {
+    swept.clear()
+  }
+
   export function close() {
     for (const item of cache.values()) item.db.$client.close()
     cache.clear()

packages/opencode/test/session/resolve-routing.test.ts

@@ -482,3 +482,119 @@ describe("resolve() routing", () => {
     })
   })
 })
+
+describe("sweep marker", () => {
+  test("seed writes marker, second ensureShard skips seed", async () => {
+    await scope(async () => {
+      const session = await Session.create({})
+      const sid = session.id
+
+      // Write a message via Sync → goes to shard
+      await msg(sid, "shard-msg")
+
+      // Simulate orphan: insert directly into global
+      const src = Database.Client().$client
+      const now = Date.now()
+      src.query("INSERT INTO message (id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?)").run(
+        MessageID.ascending(), sid, now, now, JSON.stringify({ role: "user", text: "orphan" }),
+      )
+
+      // First ensureShard: no marker → should seed
+      Database.closeSession(sid)
+      const db1 = Database.ensureShard(sid)
+      expect(db1).toBe(sid)
+
+      // Verify marker was written
+      const shard = Database.session(sid).$client
+      const marker = shard.query("SELECT value FROM _meta WHERE key = 'swept'").get() as { value: string } | null
+      expect(marker).not.toBeNull()
+      expect(Number(marker!.value)).toBeGreaterThan(0)
+
+      // Close and reopen to reset in-memory swept set
+      Database.closeSession(sid)
+      Database.resetSwept()
+
+      // Second ensureShard: marker matches global MAX → should skip seed
+      const before = performance.now()
+      const db2 = Database.ensureShard(sid)
+      const elapsed = performance.now() - before
+      expect(db2).toBe(sid)
+      // Should be fast (< 50ms) because seed was skipped
+      expect(elapsed).toBeLessThan(50)
+    })
+  })
+
+  test("new orphan after sweep triggers re-seed", async () => {
+    await scope(async () => {
+      const session = await Session.create({})
+      const sid = session.id
+
+      // Write initial message and trigger first sweep
+      await msg(sid, "initial")
+      const src = Database.Client().$client
+      const now = Date.now()
+      src.query("INSERT INTO message (id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?)").run(
+        MessageID.ascending(), sid, now, now, JSON.stringify({ role: "user", text: "orphan-1" }),
+      )
+
+      Database.closeSession(sid)
+      Database.ensureShard(sid)
+
+      // Verify orphan was copied to shard
+      const shard1 = Database.session(sid)
+      const count1 = shard1
+        .select()
+        .from(MessageTable)
+        .where(eq(MessageTable.session_id, sid))
+        .all().length
+      expect(count1).toBe(2) // shard-msg + orphan-1
+
+      // Now add ANOTHER orphan with a newer timestamp
+      const later = now + 10000
+      src.query("INSERT INTO message (id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?)").run(
+        MessageID.ascending(), sid, later, later, JSON.stringify({ role: "user", text: "orphan-2" }),
+      )
+
+      // Reset swept set to simulate process restart
+      Database.closeSession(sid)
+      Database.resetSwept()
+
+      // ensureShard should detect stale marker and re-seed
+      Database.ensureShard(sid)
+
+      const shard2 = Database.session(sid)
+      const count2 = shard2
+        .select()
+        .from(MessageTable)
+        .where(eq(MessageTable.session_id, sid))
+        .all().length
+      expect(count2).toBe(3) // shard-msg + orphan-1 + orphan-2
+    })
+  })
+
+  test("marker survives shard cache eviction", async () => {
+    await scope(async () => {
+      const session = await Session.create({})
+      const sid = session.id
+
+      // Plant orphan and sweep
+      const src = Database.Client().$client
+      const now = Date.now()
+      src.query("INSERT INTO message (id, session_id, time_created, time_updated, data) VALUES (?, ?, ?, ?, ?)").run(
+        MessageID.ascending(), sid, now, now, JSON.stringify({ role: "user", text: "orphan" }),
+      )
+
+      Database.ensureShard(sid)
+
+      // Close the shard (simulates cache eviction)
+      Database.closeSession(sid)
+      Database.resetSwept()
+
+      // Reopen — marker should still be in the sqlite file
+      const shard = Database.session(sid).$client
+      const marker = shard.query("SELECT value FROM _meta WHERE key = 'swept'").get() as { value: string } | null
+      expect(marker).not.toBeNull()
+      expect(Number(marker!.value)).toBe(now)
+    })
+  })
+})