From 238e1903df8b94a2593d22015710d3ca3e3efc01 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Fri, 21 Aug 2026 15:27:21 -0400 Subject: [PATCH] refactor(session): tighten persisted state (#43969) --- packages/core/src/session.ts | 17 +-- packages/core/src/session/history.ts | 2 +- packages/core/src/session/inbox.ts | 25 ++-- packages/core/src/session/message-updater.ts | 26 ++-- packages/core/src/session/sql.ts | 3 +- packages/core/src/session/store.ts | 5 +- packages/core/test/session-projector.test.ts | 129 ++++++++++++++++++- packages/core/test/session-sql.types.ts | 25 ++++ 8 files changed, 187 insertions(+), 45 deletions(-) create mode 100644 packages/core/test/session-sql.types.ts diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 366d954d9e..037261b9e4 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -53,6 +53,7 @@ import { Shell as ShellSchema } from "@opencode-ai/schema/shell" import { KeyedMutex } from "./effect/keyed-mutex.js" import { fileURLToPath } from "url" import { SessionEnvironment } from "./session/environment.js" +import { SessionHistory } from "./session/history.js" // get project -> project.locations // @@ -324,19 +325,8 @@ const layer = Layer.effect( Effect.provide(locations.get(location)), ) }) - const decodeMessage = Schema.decodeUnknownEffect(SessionMessage.Info) const isDurableSessionEvent = Schema.is(SessionEvent.Durable) const persistProject = (project: Project.Resolved) => upsertProject(db, project).pipe(Effect.orDie) - const decode = (row: typeof SessionMessageTable.$inferSelect) => - decodeMessage({ ...row.data, id: row.id, type: row.type }).pipe( - Effect.mapError( - () => - new MessageDecodeError({ - sessionID: SessionSchema.ID.make(row.session_id), - messageID: SessionMessage.ID.make(row.id), - }), - ), - ) const pendingConflict = Effect.fn("Session.pendingConflict")(function* (input: InboxItemRef) { yield* result.get(input.sessionID) @@ -543,7 +533,10 @@ const layer = Layer.effect( const rows = yield* (input.limit === undefined ? query.all() : query.limit(input.limit).all()).pipe( Effect.orDie, ) - return yield* Effect.forEach(direction === "previous" ? rows.toReversed() : rows, decode) + return yield* Effect.forEach( + direction === "previous" ? rows.toReversed() : rows, + SessionHistory.decodeMessageRow, + ) }), message: Effect.fn("Session.message")(function* (input) { const stored = yield* store.message(input.messageID) diff --git a/packages/core/src/session/history.ts b/packages/core/src/session/history.ts index 14b36d4e4a..19383000d6 100644 --- a/packages/core/src/session/history.ts +++ b/packages/core/src/session/history.ts @@ -29,7 +29,7 @@ export const latestCompaction = Effect.fnUntraced(function* (db: DatabaseService .pipe(Effect.orDie) }) -const decodeMessageRow = (row: typeof SessionMessageTable.$inferSelect) => +export const decodeMessageRow = (row: typeof SessionMessageTable.$inferSelect) => decode({ ...row.data, id: row.id, type: row.type }).pipe( Effect.mapError( () => diff --git a/packages/core/src/session/inbox.ts b/packages/core/src/session/inbox.ts index 1ebf86712d..f7ec3330b8 100644 --- a/packages/core/src/session/inbox.ts +++ b/packages/core/src/session/inbox.ts @@ -459,13 +459,7 @@ export const promote = Effect.fn("SessionInbox.promote")(function* ( return yield* serialized( sessionID, Effect.gen(function* () { - const steers = yield* db - .select() - .from(SessionInboxTable) - .where(and(eq(SessionInboxTable.session_id, sessionID), eq(SessionInboxTable.delivery, "steer"))) - .orderBy(asc(SessionInboxTable.enqueued_seq)) - .all() - .pipe(Effect.orDie) + const steers = yield* pendingSteers(db, sessionID) if (steers.length > 0 || scope === "steer") { const control = steers.findIndex((row) => row.type === "compaction" || row.type === "move") return yield* publish(db, bus, sessionID, control === -1 ? steers : steers.slice(0, control)) @@ -481,13 +475,7 @@ export const promote = Effect.fn("SessionInbox.promote")(function* ( .pipe(Effect.orDie) if (!queued) return 0 const promoted = yield* publish(db, bus, sessionID, [queued]) - const arrivedSteers = yield* db - .select() - .from(SessionInboxTable) - .where(and(eq(SessionInboxTable.session_id, sessionID), eq(SessionInboxTable.delivery, "steer"))) - .orderBy(asc(SessionInboxTable.enqueued_seq)) - .all() - .pipe(Effect.orDie) + const arrivedSteers = yield* pendingSteers(db, sessionID) const control = arrivedSteers.findIndex((row) => row.type === "compaction" || row.type === "move") return ( promoted + @@ -496,3 +484,12 @@ export const promote = Effect.fn("SessionInbox.promote")(function* ( }), ) }) + +const pendingSteers = (db: DatabaseService, sessionID: SessionSchema.ID) => + db + .select() + .from(SessionInboxTable) + .where(and(eq(SessionInboxTable.session_id, sessionID), eq(SessionInboxTable.delivery, "steer"))) + .orderBy(asc(SessionInboxTable.enqueued_seq)) + .all() + .pipe(Effect.orDie) diff --git a/packages/core/src/session/message-updater.ts b/packages/core/src/session/message-updater.ts index 767f0ac45f..bf7d70caea 100644 --- a/packages/core/src/session/message-updater.ts +++ b/packages/core/src/session/message-updater.ts @@ -21,8 +21,18 @@ export interface Adapter { readonly appendMessage: (message: SessionMessage.Info) => Effect.Effect } +type DraftAssistant = WritableDraft + +const projectTerminalSnapshot = (draft: DraftAssistant, event: SessionEvent.Step.Ended | SessionEvent.Step.Failed) => { + if (event.data.snapshot || event.data.files) + draft.snapshot = { + ...draft.snapshot, + end: event.data.snapshot, + files: event.data.files ? Array.from(event.data.files) : undefined, + } +} + export function update(adapter: Adapter, event: SessionEvent.DurableEvent) { - type DraftAssistant = WritableDraft type DraftTool = WritableDraft type DraftText = WritableDraft type DraftReasoning = WritableDraft @@ -229,12 +239,7 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) { draft.providerState = castDraft(event.data.providerState) draft.cost = event.data.cost draft.tokens = event.data.tokens - if (event.data.snapshot || event.data.files) - draft.snapshot = { - ...draft.snapshot, - end: event.data.snapshot, - files: event.data.files ? Array.from(event.data.files) : undefined, - } + projectTerminalSnapshot(draft, event) }) }, "session.step.failed": (event) => { @@ -249,12 +254,7 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) { draft.cost = event.data.cost draft.tokens = castDraft(event.data.tokens) } - if (event.data.snapshot || event.data.files) - draft.snapshot = { - ...draft.snapshot, - end: event.data.snapshot, - files: event.data.files ? Array.from(event.data.files) : undefined, - } + projectTerminalSnapshot(draft, event) }) }, "session.text.started": (event) => { diff --git a/packages/core/src/session/sql.ts b/packages/core/src/session/sql.ts index 96ebc3feab..5b77b15c5a 100644 --- a/packages/core/src/session/sql.ts +++ b/packages/core/src/session/sql.ts @@ -16,7 +16,8 @@ import type { CompactionPayload, MovePayload, SyntheticPayload, UserPayload } fr import type { RevertV1 } from "@opencode-ai/schema/session-revert" import type { Schema } from "effect" -type SessionMessageData = Omit<(typeof SessionMessage.Info)["Encoded"], "type" | "id"> +type DistributiveOmit = T extends unknown ? Omit : never +type SessionMessageData = DistributiveOmit<(typeof SessionMessage.Info)["Encoded"], "type" | "id"> export const SessionTable = sqliteTable( "session_v2", diff --git a/packages/core/src/session/store.ts b/packages/core/src/session/store.ts index d62e2dd073..908170c1ab 100644 --- a/packages/core/src/session/store.ts +++ b/packages/core/src/session/store.ts @@ -1,7 +1,7 @@ export * as SessionStore from "./store.js" import { and, eq, isNotNull, isNull, sql } from "drizzle-orm" -import { Context, Effect, Layer, Schema } from "effect" +import { Context, Effect, Layer } from "effect" import { Database } from "../database/database.js" import { makeGlobalNode } from "@opencode-ai/util/effect/app-node" import { SessionHistory } from "./history.js" @@ -51,7 +51,6 @@ const layer = Layer.effect( Service, Effect.gen(function* () { const { db } = yield* Database.Service - const decodeMessage = Schema.decodeUnknownEffect(SessionMessage.Info) return Service.of({ get: Effect.fnUntraced(function* (sessionID) { @@ -69,7 +68,7 @@ const layer = Layer.effect( return row ? { sessionID: Session.ID.make(row.session_id), - message: yield* decodeMessage({ ...row.data, id: row.id, type: row.type }).pipe(Effect.orDie), + message: yield* SessionHistory.decodeMessageRow(row).pipe(Effect.orDie), } : undefined }), diff --git a/packages/core/test/session-projector.test.ts b/packages/core/test/session-projector.test.ts index 1aed7f170b..6cc9caa571 100644 --- a/packages/core/test/session-projector.test.ts +++ b/packages/core/test/session-projector.test.ts @@ -12,7 +12,7 @@ import { Model } from "@opencode-ai/core/model" import { Project } from "@opencode-ai/core/project" import { ProjectTable } from "@opencode-ai/core/project/sql" import { Provider } from "@opencode-ai/core/provider" -import { AbsolutePath } from "@opencode-ai/core/schema" +import { AbsolutePath, RelativePath } from "@opencode-ai/core/schema" import { Session } from "@opencode-ai/core/session" import { SessionEvent } from "@opencode-ai/core/session/event" import { SessionMessage } from "@opencode-ai/core/session/message" @@ -298,6 +298,50 @@ describe("SessionProjector", () => { }).pipe(Effect.provide(sessionsLayer)), ) + it.effect("maps malformed persisted rows consistently while single-message lookup defects", () => + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* db + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] }) + .run() + .pipe(Effect.orDie) + yield* db + .insert(SessionTable) + .values({ + id: sessionID, + project_id: Project.ID.global, + slug: "test", + directory: "/project", + title: "test", + version: "test", + }) + .run() + .pipe(Effect.orDie) + const messageID = SessionMessage.ID.make("msg_malformed") + yield* db + .insert(SessionMessageTable) + .values({ + id: messageID, + session_id: sessionID, + type: "user", + seq: 0, + data: { text: "valid before corruption", time: { created: 0 } }, + }) + .run() + .pipe(Effect.orDie) + yield* db.run(sql`update session_message set data = '{"time":{"created":0}}' where id = ${messageID}`) + + const sessions = yield* Session.Service + const expected = { _tag: "Session.MessageDecodeError", sessionID, messageID } + expect(yield* sessions.messages({ sessionID }).pipe(Effect.flip)).toMatchObject(expected) + expect(yield* sessions.context(sessionID).pipe(Effect.flip)).toMatchObject(expected) + expect(yield* sessions.message({ sessionID, messageID }).pipe(Effect.catchDefect(Effect.succeed))).toMatchObject( + expected, + ) + }).pipe(Effect.provide(sessionsLayer)), + ) + it.effect("consumes the pending row and projects the message at promotion", () => Effect.gen(function* () { const { db } = yield* Database.Service @@ -711,6 +755,89 @@ describe("SessionProjector", () => { }), ) + it.effect("projects ended and failed step terminal state", () => + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* db + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] }) + .run() + .pipe(Effect.orDie) + yield* db + .insert(SessionTable) + .values({ + id: sessionID, + project_id: Project.ID.global, + slug: "test", + directory: "/project", + title: "test", + version: "test", + }) + .run() + .pipe(Effect.orDie) + const endedID = SessionMessage.ID.make("msg_ended") + const failedID = SessionMessage.ID.make("msg_failed") + yield* db + .insert(SessionMessageTable) + .values([assistantRow(endedID, 0), assistantRow(failedID, 1)]) + .run() + .pipe(Effect.orDie) + + const service = yield* Bus.Service + yield* service.publish(SessionEvent.Step.Ended, { + sessionID, + assistantMessageID: endedID, + finish: "stop", + rawFinish: "stop_sequence", + providerState: { response: "ended" }, + cost: Money.USD.make(1), + tokens: { input: 2, output: 3, reasoning: 4, cache: { read: 5, write: 6 } }, + snapshot: Snapshot.ID.make("snap_ended"), + files: [RelativePath.make("src/ended.ts")], + }) + yield* service.publish(SessionEvent.Step.Failed, { + sessionID, + assistantMessageID: failedID, + finish: "content-filter", + rawFinish: "blocked", + providerState: { response: "failed" }, + error: { type: "provider.invalid-request", message: "Failed" }, + snapshot: Snapshot.ID.make("snap_failed"), + files: [RelativePath.make("src/failed.ts")], + }) + + const rows = yield* db + .select() + .from(SessionMessageTable) + .where(eq(SessionMessageTable.session_id, sessionID)) + .orderBy(asc(SessionMessageTable.seq)) + .all() + .pipe(Effect.orDie) + const messages = rows.map((row) => + Schema.decodeUnknownSync(SessionMessage.Info)({ ...row.data, id: row.id, type: row.type }), + ) + expect(messages[0]).toMatchObject({ + type: "assistant", + finish: "stop", + rawFinish: "stop_sequence", + providerState: { response: "ended" }, + cost: Money.USD.make(1), + tokens: { input: 2, output: 3, reasoning: 4, cache: { read: 5, write: 6 } }, + snapshot: { end: "snap_ended", files: ["src/ended.ts"] }, + time: { completed: created }, + }) + expect(messages[1]).toMatchObject({ + type: "assistant", + finish: "content-filter", + rawFinish: "blocked", + providerState: { response: "failed" }, + error: { type: "provider.invalid-request", message: "Failed" }, + snapshot: { end: "snap_failed", files: ["src/failed.ts"] }, + time: { completed: created }, + }) + }), + ) + it.effect("does not revive a stale incomplete assistant projection", () => Effect.gen(function* () { const { db } = yield* Database.Service diff --git a/packages/core/test/session-sql.types.ts b/packages/core/test/session-sql.types.ts new file mode 100644 index 0000000000..dccb39cde9 --- /dev/null +++ b/packages/core/test/session-sql.types.ts @@ -0,0 +1,25 @@ +import { SessionMessageTable } from "@opencode-ai/core/session/sql" + +type MessageData = (typeof SessionMessageTable.$inferInsert)["data"] + +const user = { + text: "Hello", + time: { created: 0 }, +} satisfies MessageData + +const assistant = { + agent: "build", + model: { id: "model", providerID: "provider" }, + content: [], + time: { created: 0 }, +} satisfies MessageData + +const invalid = { + // @ts-expect-error Persisted message variants retain their field types. + text: 42, + time: { created: 0 }, +} satisfies MessageData + +void user +void assistant +void invalid