refactor(core): remove duplicated session projection code (#61178)

packages/core/src/session/inbox.ts

@@ -475,14 +475,7 @@ export const nextPromotable = Effect.fn("SessionInbox.nextPromotable")(function*
   const steer = (yield* pendingSteers(db, sessionID))[0]
   if (steer) return promotion(steer)
   if (promotable !== "input") return undefined
-  const queued = yield* db
-    .select()
-    .from(SessionInboxTable)
-    .where(and(eq(SessionInboxTable.session_id, sessionID), eq(SessionInboxTable.delivery, "queue")))
-    .orderBy(asc(SessionInboxTable.enqueued_seq))
-    .limit(1)
-    .get()
-    .pipe(Effect.orDie)
+  const queued = yield* firstQueued(db, sessionID)
   return queued ? promotion(queued) : undefined
 })
 
@@ -582,32 +575,36 @@ export const promote = Effect.fn("SessionInbox.promote")(function* (
     Effect.gen(function* () {
       const steers = yield* pendingSteers(db, sessionID)
       if (steers.length > 0 || scope === "steer") {
-        const control = steers.findIndex((row) => row.type === "compaction" || row.type === "move")
-        if (control === 0) return undefined
-        return yield* publish(db, bus, sessionID, control === -1 ? steers : steers.slice(0, control))
+        if (steers.length > 0 && isControl(steers[0])) return undefined
+        return yield* publish(db, bus, sessionID, beforeControl(steers))
       }
 
-      const queued = yield* db
-        .select()
-        .from(SessionInboxTable)
-        .where(and(eq(SessionInboxTable.session_id, sessionID), eq(SessionInboxTable.delivery, "queue")))
-        .orderBy(asc(SessionInboxTable.enqueued_seq))
-        .limit(1)
-        .get()
-        .pipe(Effect.orDie)
+      const queued = yield* firstQueued(db, sessionID)
       if (!queued) return 0
-      if (queued.type === "compaction" || queued.type === "move") return undefined
+      if (isControl(queued)) return undefined
       const promoted = yield* publish(db, bus, sessionID, [queued])
-      const arrivedSteers = yield* pendingSteers(db, sessionID)
-      const control = arrivedSteers.findIndex((row) => row.type === "compaction" || row.type === "move")
-      return (
-        promoted +
-        (yield* publish(db, bus, sessionID, control === -1 ? arrivedSteers : arrivedSteers.slice(0, control)))
-      )
+      return promoted + (yield* publish(db, bus, sessionID, beforeControl(yield* pendingSteers(db, sessionID))))
     }),
   )
 })
 
+const isControl = (row: typeof SessionInboxTable.$inferSelect) => row.type === "compaction" || row.type === "move"
+
+const beforeControl = (rows: Array<typeof SessionInboxTable.$inferSelect>) => {
+  const control = rows.findIndex(isControl)
+  return control === -1 ? rows : rows.slice(0, control)
+}
+
+const firstQueued = (db: DatabaseService, sessionID: SessionSchema.ID) =>
+  db
+    .select()
+    .from(SessionInboxTable)
+    .where(and(eq(SessionInboxTable.session_id, sessionID), eq(SessionInboxTable.delivery, "queue")))
+    .orderBy(asc(SessionInboxTable.enqueued_seq))
+    .limit(1)
+    .get()
+    .pipe(Effect.orDie)
+
 const pendingSteers = (db: DatabaseService, sessionID: SessionSchema.ID) =>
   db
     .select()
@@ -621,7 +618,7 @@ const pendingSteers = (db: DatabaseService, sessionID: SessionSchema.ID) =>
         // A move changes the context's Location: never pull compaction across it.
         // Within that boundary, compact before promoting even earlier steers so
         // their text stays verbatim after the checkpoint, not inside its summary.
-        const control = rows.findIndex((row) => row.type === "compaction" || row.type === "move")
+        const control = rows.findIndex(isControl)
         if (control > 0 && rows[control].type === "compaction") rows.unshift(...rows.splice(control, 1))
         return rows
       }),

packages/core/src/session/instruction-entry.ts

@@ -25,6 +25,10 @@ export const snapshot = Effect.fn("InstructionEntry.snapshot")(function* (
   db: DatabaseService,
   sessionID: SessionSchema.ID,
 ) {
+  return yield* rows(db, sessionID, true)
+})
+
+const rows = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID, includeRemoved: boolean) {
   return yield* db
     .select({
       key: InstructionEntryTable.key,
@@ -32,7 +36,12 @@ export const snapshot = Effect.fn("InstructionEntry.snapshot")(function* (
       removed: InstructionEntryTable.removed,
     })
     .from(InstructionEntryTable)
-    .where(eq(InstructionEntryTable.session_id, sessionID))
+    .where(
+      and(
+        eq(InstructionEntryTable.session_id, sessionID),
+        includeRemoved ? undefined : eq(InstructionEntryTable.removed, false),
+      ),
+    )
     .orderBy(asc(InstructionEntryTable.key))
     .all()
     .pipe(Effect.orDie)
@@ -108,27 +117,8 @@ const layer = Layer.effect(
   Effect.gen(function* () {
     const { db } = yield* Database.Service
 
-    const rows = Effect.fnUntraced(function* (sessionID: SessionSchema.ID, includeRemoved: boolean) {
-      return yield* db
-        .select({
-          key: InstructionEntryTable.key,
-          value: InstructionEntryTable.value,
-          removed: InstructionEntryTable.removed,
-        })
-        .from(InstructionEntryTable)
-        .where(
-          and(
-            eq(InstructionEntryTable.session_id, sessionID),
-            includeRemoved ? undefined : eq(InstructionEntryTable.removed, false),
-          ),
-        )
-        .orderBy(asc(InstructionEntryTable.key))
-        .all()
-        .pipe(Effect.orDie)
-    })
-
     const list = Effect.fn("InstructionEntry.list")(function* (sessionID: SessionSchema.ID) {
-      return (yield* rows(sessionID, false)).map((row) => ({ key: row.key, value: row.value }))
+      return (yield* rows(db, sessionID, false)).map((row) => ({ key: row.key, value: row.value }))
     })
 
     const put = Effect.fn("InstructionEntry.put")(function* (input: Parameters<Interface["put"]>[0]) {
@@ -171,7 +161,7 @@ const layer = Layer.effect(
     })
 
     const load = Effect.fn("InstructionEntry.load")(function* (sessionID: SessionSchema.ID) {
-      return Instructions.combine((yield* rows(sessionID, true)).map(source))
+      return Instructions.combine((yield* rows(db, sessionID, true)).map(source))
     })
 
     return Service.of({ list, put, remove, load })

packages/core/src/session/instruction-state.ts

@@ -94,20 +94,7 @@ export const apply = Effect.fn("InstructionState.apply")(function* (
 ) {
   const stored = yield* find(db, sessionID)
   const current = Instructions.applyHashDelta(stored?.current_values ?? {}, delta)
-  if (!stored) {
-    yield* db
-      .insert(InstructionStateTable)
-      .values({
-        session_id: sessionID,
-        epoch_start: seq,
-        through_seq: seq,
-        initial_values: current,
-        current_values: current,
-      })
-      .run()
-      .pipe(Effect.orDie)
-    return
-  }
+  if (!stored) return yield* initialize(db, sessionID, seq, current)
   yield* db
     .update(InstructionStateTable)
     .set({ through_seq: seq, current_values: current })

packages/core/src/session/projector.ts

@@ -16,6 +16,7 @@ import { SessionBlob } from "./blob.js"
 import { SessionInboxTable, SessionMessageTable, SessionPartTable, SessionTable, type SessionPartData } from "./sql.js"
 import { SessionStored } from "./stored.js"
 import { InstructionEntry } from "./instruction-entry.js"
+import type { SQLiteUpdateSetSource } from "drizzle-orm/sqlite-core/query-builders/update"
 import { Slug } from "../util/slug.js"
 import { FSUtil } from "@opencode/util/fs-util"
 import { Money } from "@opencode/schema/money"
@@ -52,30 +53,28 @@ const forkTitle = (value?: string) => {
   return `${value} (fork #1)`
 }
 
+function setSession(
+  db: DatabaseService,
+  sessionID: SessionSchema.ID,
+  values: SQLiteUpdateSetSource<typeof SessionTable>,
+) {
+  return db.update(SessionTable).set(values).where(eq(SessionTable.id, sessionID)).run().pipe(Effect.orDie)
+}
+
 function applyUsage(db: DatabaseService, sessionID: SessionSchema.ID, value: Usage, timeUpdated?: number) {
-  return db
-    .update(SessionTable)
-    .set({
-      cost: sql`${SessionTable.cost} + ${value.cost}`,
-      tokens_input: sql`${SessionTable.tokens_input} + ${value.tokens.input}`,
-      tokens_output: sql`${SessionTable.tokens_output} + ${value.tokens.output}`,
-      tokens_reasoning: sql`${SessionTable.tokens_reasoning} + ${value.tokens.reasoning}`,
-      tokens_cache_read: sql`${SessionTable.tokens_cache_read} + ${value.tokens.cache.read}`,
-      tokens_cache_write: sql`${SessionTable.tokens_cache_write} + ${value.tokens.cache.write}`,
-      time_updated: timeUpdated ?? sql`${SessionTable.time_updated}`,
-    })
-    .where(eq(SessionTable.id, sessionID))
-    .run()
-    .pipe(Effect.orDie)
+  return setSession(db, sessionID, {
+    cost: sql`${SessionTable.cost} + ${value.cost}`,
+    tokens_input: sql`${SessionTable.tokens_input} + ${value.tokens.input}`,
+    tokens_output: sql`${SessionTable.tokens_output} + ${value.tokens.output}`,
+    tokens_reasoning: sql`${SessionTable.tokens_reasoning} + ${value.tokens.reasoning}`,
+    tokens_cache_read: sql`${SessionTable.tokens_cache_read} + ${value.tokens.cache.read}`,
+    tokens_cache_write: sql`${SessionTable.tokens_cache_write} + ${value.tokens.cache.write}`,
+    time_updated: timeUpdated ?? sql`${SessionTable.time_updated}`,
+  })
 }
 
 function touch(db: DatabaseService, event: MessageEvent) {
-  return db
-    .update(SessionTable)
-    .set({ time_updated: event.created })
-    .where(eq(SessionTable.id, event.data.sessionID))
-    .run()
-    .pipe(Effect.orDie)
+  return setSession(db, event.data.sessionID, { time_updated: event.created })
 }
 
 const publishSessionUsage = Effect.fn("SessionProjector.publishUsage")(function* (
@@ -520,17 +519,12 @@ function projectIdle(
         : event.type === SessionEvent.Execution.Failed.type
           ? "failed"
           : "interrupted"
-    yield* db
-      .update(SessionTable)
-      .set({
-        // Unread uses a strict timestamp comparison, so every terminal must advance even within one millisecond.
-        time_idle: sql`max(${time}, coalesce(${SessionTable.time_idle} + 1, ${time}))`,
-        idle_outcome: outcome,
-        time_updated: sql`${SessionTable.time_updated}`,
-      })
-      .where(eq(SessionTable.id, event.data.sessionID))
-      .run()
-      .pipe(Effect.orDie)
+    yield* setSession(db, event.data.sessionID, {
+      // Unread uses a strict timestamp comparison, so every terminal must advance even within one millisecond.
+      time_idle: sql`max(${time}, coalesce(${SessionTable.time_idle} + 1, ${time}))`,
+      idle_outcome: outcome,
+      time_updated: sql`${SessionTable.time_updated}`,
+    })
   })
 }
 
@@ -569,18 +563,13 @@ const layer = Layer.effectDiscard(
     yield* bus.project(SessionEvent.Moved, (event) =>
       Effect.gen(function* () {
         yield* run(db, event)
-        yield* db
-          .update(SessionTable)
-          .set({
-            directory: event.data.location.directory,
-            path: event.data.subpath,
-            ...(event.data.projectID ? { project_id: event.data.projectID } : {}),
-            workspace_id: event.data.location.workspaceID ?? null,
-            time_updated: event.created,
-          })
-          .where(eq(SessionTable.id, event.data.sessionID))
-          .run()
-          .pipe(Effect.orDie)
+        yield* setSession(db, event.data.sessionID, {
+          directory: event.data.location.directory,
+          path: event.data.subpath,
+          ...(event.data.projectID ? { project_id: event.data.projectID } : {}),
+          workspace_id: event.data.location.workspaceID ?? null,
+          time_updated: event.created,
+        })
       }),
     )
     // Sessions whose ownership came from the directory's previous resolution
@@ -627,17 +616,12 @@ const layer = Layer.effectDiscard(
               ? row.canonical
               : AbsolutePath.make(path.resolve(row.directory))
             if (!FSUtil.contains(event.data.directory, directory)) return Effect.void
-            return db
-              .update(SessionTable)
-              .set({
-                project_id: event.data.projectID,
-                path: RelativePath.make(path.relative(event.data.directory, directory).replaceAll("\\", "/")),
-                // Self-assignment suppresses the column's $onUpdate: adoption is not activity.
-                time_updated: sql`${SessionTable.time_updated}`,
-              })
-              .where(eq(SessionTable.id, row.id))
-              .run()
-              .pipe(Effect.orDie)
+            return setSession(db, row.id, {
+              project_id: event.data.projectID,
+              path: RelativePath.make(path.relative(event.data.directory, directory).replaceAll("\\", "/")),
+              // Self-assignment suppresses the column's $onUpdate: adoption is not activity.
+              time_updated: sql`${SessionTable.time_updated}`,
+            })
           },
           { discard: true },
         )
@@ -652,78 +636,38 @@ const layer = Layer.effectDiscard(
     yield* bus.project(SessionEvent.AgentSelected, (event) =>
       Effect.gen(function* () {
         yield* run(db, event)
-        yield* db
-          .update(SessionTable)
-          .set({ agent: event.data.agent, time_updated: event.created })
-          .where(eq(SessionTable.id, event.data.sessionID))
-          .run()
-          .pipe(Effect.orDie)
+        yield* setSession(db, event.data.sessionID, { agent: event.data.agent, time_updated: event.created })
       }),
     )
     yield* bus.project(SessionEvent.ModelSelected, (event) =>
       Effect.gen(function* () {
         yield* run(db, event)
-        yield* db
-          .update(SessionTable)
-          .set({ model: event.data.model, time_updated: event.created })
-          .where(eq(SessionTable.id, event.data.sessionID))
-          .run()
-          .pipe(Effect.orDie)
+        yield* setSession(db, event.data.sessionID, { model: event.data.model, time_updated: event.created })
       }),
     )
     yield* bus.project(SessionEvent.Renamed, (event) =>
-      db
-        .update(SessionTable)
-        .set({ title: event.data.title, time_updated: event.created })
-        .where(eq(SessionTable.id, event.data.sessionID))
-        .run()
-        .pipe(Effect.orDie),
+      setSession(db, event.data.sessionID, { title: event.data.title, time_updated: event.created }),
     )
     yield* bus.project(SessionEvent.MetadataUpdated, (event) =>
-      db
-        .update(SessionTable)
-        .set({ metadata: event.data.metadata, time_updated: event.created })
-        .where(eq(SessionTable.id, event.data.sessionID))
-        .run()
-        .pipe(Effect.orDie),
+      setSession(db, event.data.sessionID, { metadata: event.data.metadata, time_updated: event.created }),
     )
     yield* bus.project(SessionEvent.Permissions, (event) =>
-      db
-        .update(SessionTable)
-        .set({ permission: event.data.permissions, time_updated: event.created })
-        .where(eq(SessionTable.id, event.data.sessionID))
-        .run()
-        .pipe(Effect.orDie),
+      setSession(db, event.data.sessionID, { permission: event.data.permissions, time_updated: event.created }),
     )
     yield* bus.project(SessionEvent.Viewed, (event) => {
       const idle = event.data.idle
-      return db
-        .update(SessionTable)
-        .set({
-          // Monotone watermark: a duplicate or stale view never regresses, and a terminal event
-          // committing after the viewer's observation keeps the newer idle transition unread.
-          time_viewed: sql`max(${idle}, coalesce(${SessionTable.time_viewed}, ${idle}))`,
-          time_updated: sql`${SessionTable.time_updated}`,
-        })
-        .where(eq(SessionTable.id, event.data.sessionID))
-        .run()
-        .pipe(Effect.orDie)
+      return setSession(db, event.data.sessionID, {
+        // Monotone watermark: a duplicate or stale view never regresses, and a terminal event
+        // committing after the viewer's observation keeps the newer idle transition unread.
+        time_viewed: sql`max(${idle}, coalesce(${SessionTable.time_viewed}, ${idle}))`,
+        time_updated: sql`${SessionTable.time_updated}`,
+      })
     })
     yield* bus.project(SessionEvent.Archived, (event) =>
-      db
-        .update(SessionTable)
-        .set({ time_archived: event.created, time_updated: event.created })
-        .where(eq(SessionTable.id, event.data.sessionID))
-        .run()
-        .pipe(Effect.orDie),
+      setSession(db, event.data.sessionID, { time_archived: event.created, time_updated: event.created }),
     )
     yield* bus.project(SessionEvent.Unarchived, (event) =>
-      db
-        .update(SessionTable)
-        .set({ time_archived: null, time_updated: event.created })
-        .where(eq(SessionTable.id, event.data.sessionID))
-        .run()
-        .pipe(Effect.orDie),
+      setSession(db, event.data.sessionID, { time_archived: null, time_updated: event.created }),
     )
     yield* bus.project(SessionEvent.MessageContentUpdated, (event) => run(db, event))
     yield* bus.project(SessionEvent.UsageRecorded, (event) => applyUsage(db, event.data.sessionID, event.data))
@@ -766,12 +710,7 @@ const layer = Layer.effectDiscard(
           item: event.data.item,
           timeCreated: event.created,
         })
-        yield* db
-          .update(SessionTable)
-          .set({ time_updated: event.created })
-          .where(eq(SessionTable.id, event.data.sessionID))
-          .run()
-          .pipe(Effect.orDie)
+        yield* touch(db, event)
       }),
     )
     yield* bus.project(SessionEvent.InboxCancelled, (event) =>
@@ -788,12 +727,10 @@ const layer = Layer.effectDiscard(
       }),
     )
     yield* bus.project(SessionEvent.Execution.Started, (event) =>
-      db
-        .update(SessionTable)
-        .set({ time_busy: event.created, time_updated: sql`${SessionTable.time_updated}` })
-        .where(eq(SessionTable.id, event.data.sessionID))
-        .run()
-        .pipe(Effect.orDie),
+      setSession(db, event.data.sessionID, {
+        time_busy: event.created,
+        time_updated: sql`${SessionTable.time_updated}`,
+      }),
     )
     yield* bus.project(SessionEvent.Execution.Succeeded, (event) => projectIdle(db, event))
     yield* bus.project(SessionEvent.Execution.Failed, (event) => projectIdle(db, event))
@@ -853,26 +790,10 @@ const layer = Layer.effectDiscard(
     )
     yield* bus.project(SessionEvent.Compaction.Failed, (event) => run(db, event))
     yield* bus.project(SessionEvent.RevertEvent.Staged, (event) =>
-      Effect.gen(function* () {
-        const revert = event.data.revert
-        yield* db
-          .update(SessionTable)
-          .set({
-            revert,
-            time_updated: event.created,
-          })
-          .where(eq(SessionTable.id, event.data.sessionID))
-          .run()
-          .pipe(Effect.orDie)
-      }),
+      setSession(db, event.data.sessionID, { revert: event.data.revert, time_updated: event.created }),
     )
     yield* bus.project(SessionEvent.RevertEvent.Cleared, (event) =>
-      db
-        .update(SessionTable)
-        .set({ revert: null, time_updated: event.created })
-        .where(eq(SessionTable.id, event.data.sessionID))
-        .run()
-        .pipe(Effect.orDie, Effect.asVoid),
+      setSession(db, event.data.sessionID, { revert: null, time_updated: event.created }),
     )
     yield* bus.project(SessionEvent.RevertEvent.Committed, (event) =>
       Effect.gen(function* () {
@@ -902,12 +823,7 @@ const layer = Layer.effectDiscard(
           )
           .run()
           .pipe(Effect.orDie)
-        yield* db
-          .update(SessionTable)
-          .set({ revert: null, time_updated: event.created })
-          .where(eq(SessionTable.id, event.data.sessionID))
-          .run()
-          .pipe(Effect.orDie)
+        yield* setSession(db, event.data.sessionID, { revert: null, time_updated: event.created })
         yield* InstructionState.rewind(db, event.data.sessionID, boundary.seq)
         yield* SessionBlob.releaseFrom(db, event.data.sessionID, boundary.seq)
       }),

packages/core/src/session/runner/publish-llm-event.ts

@@ -228,52 +228,32 @@ export const createLLMEventPublisher = (bus: Pick<Bus.Interface, "publish">, inp
     return { start, append, end, flush, publishPending, has: (id: string) => chunks.has(id) }
   }
 
-  const text = fragments(
-    "text",
-    (_textID, value, ordinal, native) =>
-      Effect.gen(function* () {
-        yield* bus.publish(SessionEvent.Text.Ended, {
-          sessionID: input.sessionID,
-          assistantMessageID: yield* currentAssistantMessageID(),
-          ordinal,
-          text: value,
-          native,
-        })
-      }),
-    (_textID, value, ordinal) =>
-      Effect.gen(function* () {
-        yield* bus.publish(SessionEvent.Text.Delta, {
-          sessionID: input.sessionID,
-          assistantMessageID: yield* currentAssistantMessageID(),
-          ordinal,
-          delta: value,
-        })
-      }),
-    true,
-  )
-  const reasoning = fragments(
-    "reasoning",
-    (_reasoningID, value, ordinal, native) =>
-      Effect.gen(function* () {
-        yield* bus.publish(SessionEvent.Reasoning.Ended, {
-          sessionID: input.sessionID,
-          assistantMessageID: yield* currentAssistantMessageID(),
-          ordinal,
-          text: value,
-          native,
-        })
-      }),
-    (_reasoningID, value, ordinal) =>
-      Effect.gen(function* () {
-        yield* bus.publish(SessionEvent.Reasoning.Delta, {
-          sessionID: input.sessionID,
-          assistantMessageID: yield* currentAssistantMessageID(),
-          ordinal,
-          delta: value,
-        })
-      }),
-    true,
-  )
+  const contentFragments = (name: string, events: typeof SessionEvent.Text | typeof SessionEvent.Reasoning) =>
+    fragments(
+      name,
+      (_id, value, ordinal, native) =>
+        Effect.gen(function* () {
+          yield* bus.publish(events.Ended, {
+            sessionID: input.sessionID,
+            assistantMessageID: yield* currentAssistantMessageID(),
+            ordinal,
+            text: value,
+            native,
+          })
+        }),
+      (_id, value, ordinal) =>
+        Effect.gen(function* () {
+          yield* bus.publish(events.Delta, {
+            sessionID: input.sessionID,
+            assistantMessageID: yield* currentAssistantMessageID(),
+            ordinal,
+            delta: value,
+          })
+        }),
+      true,
+    )
+  const text = contentFragments("text", SessionEvent.Text)
+  const reasoning = contentFragments("reasoning", SessionEvent.Reasoning)
   const toolInput = fragments("tool input", (id, value) =>
     Effect.gen(function* () {
       const tool = tools.get(id)
@@ -607,7 +587,6 @@ export const createLLMEventPublisher = (bus: Pick<Bus.Interface, "publish">, inp
     publishStepFailure,
     failUnsettledTools,
     failIncompleteTools,
-    hasProviderError: () => providerFailed,
     hasStarted: () => stepStarted,
     /** Immutable snapshot of everything recorded for this step so far. */
     record: (): StepRecord => ({

packages/core/src/session/runner/step.ts

@@ -110,7 +110,7 @@ export const make = Effect.gen(function* () {
     const providerStream = llm.stream(input.prepared.request, input.prepared.options).pipe(
       Stream.runForEach((event) =>
         Effect.gen(function* () {
-          if (publisher.hasProviderError()) return
+          if (publisher.record().providerFailed) return
           // Wait here, where cancellation still works, rather than inside the uninterruptible publish.
           if (!publisher.hasStarted()) yield* Fiber.join(pendingStartSnapshot)
           yield* publisher.publish(event)