From 730460bb60cc5d9fc2d248374763cee5ba2d891f Mon Sep 17 00:00:00 2001 From: Aiden Cline <63023139+rekram1-node@users.noreply.github.com> Date: Sun, 1 Nov 2026 21:16:04 -0600 Subject: [PATCH] refactor(core): remove duplicated session projection code (#61178) --- packages/core/src/session/inbox.ts | 51 +++-- .../core/src/session/instruction-entry.ts | 34 ++- .../core/src/session/instruction-state.ts | 15 +- packages/core/src/session/projector.ts | 202 +++++------------- .../src/session/runner/publish-llm-event.ts | 73 +++---- packages/core/src/session/runner/step.ts | 2 +- 6 files changed, 123 insertions(+), 254 deletions(-) diff --git a/packages/core/src/session/inbox.ts b/packages/core/src/session/inbox.ts index bbdef5def1e..3e5cb046ee5 100644 --- a/packages/core/src/session/inbox.ts +++ b/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) => { + 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 }), diff --git a/packages/core/src/session/instruction-entry.ts b/packages/core/src/session/instruction-entry.ts index 2a9c199169e..ec6840dea5f 100644 --- a/packages/core/src/session/instruction-entry.ts +++ b/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[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 }) diff --git a/packages/core/src/session/instruction-state.ts b/packages/core/src/session/instruction-state.ts index 0e040017393..7aa31118b76 100644 --- a/packages/core/src/session/instruction-state.ts +++ b/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 }) diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts index 920dda5787a..e8bdc4d0d6e 100644 --- a/packages/core/src/session/projector.ts +++ b/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, +) { + 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) }), diff --git a/packages/core/src/session/runner/publish-llm-event.ts b/packages/core/src/session/runner/publish-llm-event.ts index 700077776dd..49ef5c7e22e 100644 --- a/packages/core/src/session/runner/publish-llm-event.ts +++ b/packages/core/src/session/runner/publish-llm-event.ts @@ -228,52 +228,32 @@ export const createLLMEventPublisher = (bus: Pick, 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, inp publishStepFailure, failUnsettledTools, failIncompleteTools, - hasProviderError: () => providerFailed, hasStarted: () => stepStarted, /** Immutable snapshot of everything recorded for this step so far. */ record: (): StepRecord => ({ diff --git a/packages/core/src/session/runner/step.ts b/packages/core/src/session/runner/step.ts index 8190179e3e6..9243c6702b5 100644 --- a/packages/core/src/session/runner/step.ts +++ b/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)