From 67b4b2894f73fd8ef5a7cc7d2c20ffdb3488f1c0 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Thu, 2 Jul 2026 00:21:56 -0400 Subject: [PATCH] refactor(core): replace context epoch with a single-entry checkpoint One prepare call per turn, before input promotion: it creates the checkpoint when missing, rebaselines after completed compaction, and otherwise narrates drift as a chronological update. The initialize/ prepare two-step and its redundant queries are gone, and context update messages now precede the promoted user message. An undecodable stored applied record heals by treating every source as new instead of failing the drain; ContextSnapshotDecodeError is deleted. Renames: SessionContextEpoch -> SessionContextCheckpoint. The SQL table name and wire event type are unchanged. --- packages/core/src/session.ts | 10 +- .../core/src/session/context-checkpoint.ts | 131 +++++++++++++ packages/core/src/session/context-epoch.ts | 172 ------------------ packages/core/src/session/error.ts | 12 -- packages/core/src/session/history.ts | 11 +- packages/core/src/session/projector.ts | 14 +- packages/core/src/session/runner/index.ts | 3 +- packages/core/src/session/runner/llm.ts | 19 +- packages/core/src/session/sql.ts | 2 +- packages/core/test/session-runner.test.ts | 48 +++-- 10 files changed, 193 insertions(+), 229 deletions(-) create mode 100644 packages/core/src/session/context-checkpoint.ts delete mode 100644 packages/core/src/session/context-epoch.ts diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index bbd6e2cc4b..9af15f7208 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -108,7 +108,7 @@ export class OperationUnavailableError extends Schema.TaggedErrorClass()("Session.PromptConflictError", { sessionID: SessionSchema.ID, @@ -466,7 +466,9 @@ const layer = Layer.effect( text: skill.content, }) if (input.resume !== false) - yield* execution.resume(input.sessionID).pipe(Effect.ignore, Effect.forkIn(scope, { startImmediately: true }), Effect.asVoid) + yield* execution + .resume(input.sessionID) + .pipe(Effect.ignore, Effect.forkIn(scope, { startImmediately: true }), Effect.asVoid) }), switchAgent: Effect.fn("V2Session.switchAgent")(function* (input) { yield* result.get(input.sessionID) @@ -550,7 +552,9 @@ const layer = Layer.effect( description: input.description, metadata: input.metadata, }) - yield* execution.resume(input.sessionID).pipe(Effect.ignore, Effect.forkIn(scope, { startImmediately: true }), Effect.asVoid) + yield* execution + .resume(input.sessionID) + .pipe(Effect.ignore, Effect.forkIn(scope, { startImmediately: true }), Effect.asVoid) }), interrupt: Effect.fn("V2Session.interrupt")((sessionID) => Effect.uninterruptible(execution.interrupt(sessionID)), diff --git a/packages/core/src/session/context-checkpoint.ts b/packages/core/src/session/context-checkpoint.ts new file mode 100644 index 0000000000..c2036e726a --- /dev/null +++ b/packages/core/src/session/context-checkpoint.ts @@ -0,0 +1,131 @@ +export * as SessionContextCheckpoint from "./context-checkpoint" + +import { eq } from "drizzle-orm" +import { DateTime, Effect, Option, Schema } from "effect" +import type { Database } from "../database/database" +import { EventV2 } from "../event" +import { SystemContext } from "../system-context/index" +import { SessionEvent } from "./event" +import { SessionHistory } from "./history" +import { SessionMessage } from "./message" +import { SessionSchema } from "./schema" +import { SessionContextCheckpointTable } from "./sql" + +type DatabaseService = Database.Interface["db"] + +const decodeApplied = Schema.decodeUnknownOption(SystemContext.Applied) + +/** + * Loads or creates the session's durable context checkpoint, narrating any + * drift since the model was last told as a chronological update. Completed + * compaction rebaselines; nothing else rewrites the baseline. Runs before + * input promotion so a blocked first turn leaves pending inputs untouched. + */ +export const prepare = Effect.fn("SessionContextCheckpoint.prepare")(function* ( + db: DatabaseService, + events: EventV2.Interface, + context: Effect.Effect, + sessionID: SessionSchema.ID, +) { + const [value, stored, compaction] = yield* Effect.all( + [context, find(db, sessionID), SessionHistory.latestCompaction(db, sessionID)], + { concurrency: "unbounded" }, + ) + if (!stored) { + const baseline = yield* SystemContext.initialize(value) + const baselineSeq = yield* insert(db, sessionID, baseline) + return { baseline: baseline.text, baselineSeq } + } + + // The applied record is comparison state only; an undecodable one heals by + // treating every source as new, re-announcing baselines as updates. + const applied = Option.getOrElse(decodeApplied(stored.snapshot), () => ({})) + if (compaction !== undefined && compaction.seq > stored.baseline_seq) { + const baseline = yield* SystemContext.rebaseline(value, applied) + yield* rewrite(db, sessionID, compaction.seq, baseline) + return { baseline: baseline.text, baselineSeq: compaction.seq } + } + const result = yield* SystemContext.reconcile(value, applied) + if (result._tag === "Unchanged") return { baseline: stored.baseline, baselineSeq: stored.baseline_seq } + + yield* events.publish( + SessionEvent.ContextUpdated, + { sessionID, messageID: SessionMessage.ID.create(), timestamp: yield* DateTime.now, text: result.text }, + { commit: () => advance(db, sessionID, result.applied).pipe(Effect.orDie) }, + ) + return { baseline: stored.baseline, baselineSeq: stored.baseline_seq } +}) + +export const reset = Effect.fn("SessionContextCheckpoint.reset")(function* ( + db: DatabaseService, + sessionID: SessionSchema.ID, +) { + yield* db + .delete(SessionContextCheckpointTable) + .where(eq(SessionContextCheckpointTable.session_id, sessionID)) + .run() + .pipe(Effect.orDie) +}) + +const find = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) { + return yield* db + .select() + .from(SessionContextCheckpointTable) + .where(eq(SessionContextCheckpointTable.session_id, sessionID)) + .get() + .pipe(Effect.orDie) +}) + +const insert = Effect.fnUntraced(function* ( + db: DatabaseService, + sessionID: SessionSchema.ID, + baseline: SystemContext.Baseline, +) { + const baselineSeq = yield* EventV2.latestSequence(db, sessionID) + yield* db + .insert(SessionContextCheckpointTable) + .values({ + session_id: sessionID, + baseline: baseline.text, + snapshot: baseline.applied, + baseline_seq: baselineSeq, + }) + .run() + .pipe(Effect.orDie) + return baselineSeq +}) + +const rewrite = Effect.fnUntraced(function* ( + db: DatabaseService, + sessionID: SessionSchema.ID, + baselineSeq: number, + baseline: SystemContext.Baseline, +) { + const updated = yield* db + .update(SessionContextCheckpointTable) + .set({ + baseline: baseline.text, + snapshot: baseline.applied, + baseline_seq: baselineSeq, + }) + .where(eq(SessionContextCheckpointTable.session_id, sessionID)) + .returning({ sessionID: SessionContextCheckpointTable.session_id }) + .get() + .pipe(Effect.orDie) + if (!updated) return yield* Effect.die("Context checkpoint not found") +}) + +const advance = Effect.fnUntraced(function* ( + db: DatabaseService, + sessionID: SessionSchema.ID, + applied: SystemContext.Applied, +) { + const updated = yield* db + .update(SessionContextCheckpointTable) + .set({ snapshot: applied }) + .where(eq(SessionContextCheckpointTable.session_id, sessionID)) + .returning({ sessionID: SessionContextCheckpointTable.session_id }) + .get() + .pipe(Effect.orDie) + if (!updated) return yield* Effect.die("Context checkpoint not found") +}) diff --git a/packages/core/src/session/context-epoch.ts b/packages/core/src/session/context-epoch.ts deleted file mode 100644 index 404b3eb2f3..0000000000 --- a/packages/core/src/session/context-epoch.ts +++ /dev/null @@ -1,172 +0,0 @@ -export * as SessionContextEpoch from "./context-epoch" - -import { eq } from "drizzle-orm" -import { DateTime, Effect, Schema } from "effect" -import type { Database } from "../database/database" -import { EventV2 } from "../event" -import { SystemContext } from "../system-context/index" -import { ContextSnapshotDecodeError } from "./error" -import { SessionEvent } from "./event" -import { SessionHistory } from "./history" -import { SessionMessage } from "./message" -import { SessionSchema } from "./schema" -import { SessionContextEpochTable } from "./sql" - -type DatabaseService = Database.Interface["db"] - -interface Prepared { - readonly baseline: string - readonly baselineSeq: number -} - -export function initialize( - db: DatabaseService, - context: Effect.Effect, - sessionID: SessionSchema.ID, -): Effect.Effect { - return initializeOnce(db, context, sessionID).pipe(Effect.withSpan("SessionContextEpoch.initialize")) -} - -export function prepare( - db: DatabaseService, - events: EventV2.Interface, - context: Effect.Effect, - sessionID: SessionSchema.ID, -): Effect.Effect { - return prepareOnce(db, events, context, sessionID).pipe(Effect.withSpan("SessionContextEpoch.prepare")) -} - -const prepareOnce = Effect.fnUntraced(function* ( - db: DatabaseService, - events: EventV2.Interface, - context: Effect.Effect, - sessionID: SessionSchema.ID, -) { - const [value, stored, compaction] = yield* Effect.all( - [context, find(db, sessionID), SessionHistory.latestCompaction(db, sessionID)], - { concurrency: "unbounded" }, - ) - if (!stored) return yield* insertInitial(db, sessionID, value) - - const applied = yield* Schema.decodeUnknownEffect(SystemContext.Applied)(stored.snapshot).pipe( - Effect.mapError((error) => new ContextSnapshotDecodeError({ sessionID, details: String(error) })), - ) - if (compaction !== undefined && compaction.seq > stored.baseline_seq) { - const generation = yield* SystemContext.rebaseline(value, applied) - yield* replace(db, sessionID, compaction.seq, generation) - return { baseline: generation.text, baselineSeq: compaction.seq } - } - const result = yield* SystemContext.reconcile(value, applied) - if (result._tag === "Unchanged") return { baseline: stored.baseline, baselineSeq: stored.baseline_seq } - - yield* events.publish( - SessionEvent.ContextUpdated, - { sessionID, messageID: SessionMessage.ID.create(), timestamp: yield* DateTime.now, text: result.text }, - { commit: () => advance(db, sessionID, result.applied).pipe(Effect.orDie) }, - ) - return { baseline: stored.baseline, baselineSeq: stored.baseline_seq } -}) - -const initializeOnce = Effect.fnUntraced(function* ( - db: DatabaseService, - context: Effect.Effect, - sessionID: SessionSchema.ID, -) { - if (yield* exists(db, sessionID)) return undefined - return yield* insertInitial(db, sessionID, yield* context) -}) - -const insertInitial = Effect.fnUntraced(function* ( - db: DatabaseService, - sessionID: SessionSchema.ID, - value: SystemContext.SystemContext, -) { - const generation = yield* SystemContext.initialize(value) - const baselineSeq = yield* insert(db, sessionID, generation) - return { baseline: generation.text, baselineSeq } -}) - -const exists = Effect.fn("SessionContextEpoch.exists")(function* (db: DatabaseService, sessionID: SessionSchema.ID) { - return ( - (yield* db - .select({ sessionID: SessionContextEpochTable.session_id }) - .from(SessionContextEpochTable) - .where(eq(SessionContextEpochTable.session_id, sessionID)) - .get() - .pipe(Effect.orDie)) !== undefined - ) -}) - -const find = Effect.fn("SessionContextEpoch.find")(function* (db: DatabaseService, sessionID: SessionSchema.ID) { - return yield* db - .select() - .from(SessionContextEpochTable) - .where(eq(SessionContextEpochTable.session_id, sessionID)) - .get() - .pipe(Effect.orDie) -}) - -export const reset = Effect.fn("SessionContextEpoch.reset")(function* ( - db: DatabaseService, - sessionID: SessionSchema.ID, -) { - yield* db - .delete(SessionContextEpochTable) - .where(eq(SessionContextEpochTable.session_id, sessionID)) - .run() - .pipe(Effect.orDie) -}) - -const insert = Effect.fnUntraced(function* ( - db: DatabaseService, - sessionID: SessionSchema.ID, - generation: SystemContext.Baseline, -) { - const baselineSeq = yield* EventV2.latestSequence(db, sessionID) - yield* db - .insert(SessionContextEpochTable) - .values({ - session_id: sessionID, - baseline: generation.text, - snapshot: generation.applied, - baseline_seq: baselineSeq, - }) - .run() - .pipe(Effect.orDie) - return baselineSeq -}) - -const replace = Effect.fnUntraced(function* ( - db: DatabaseService, - sessionID: SessionSchema.ID, - baselineSeq: number, - generation: SystemContext.Baseline, -) { - const updated = yield* db - .update(SessionContextEpochTable) - .set({ - baseline: generation.text, - snapshot: generation.applied, - baseline_seq: baselineSeq, - }) - .where(eq(SessionContextEpochTable.session_id, sessionID)) - .returning({ sessionID: SessionContextEpochTable.session_id }) - .get() - .pipe(Effect.orDie) - if (!updated) return yield* Effect.die("Context Epoch not found") -}) - -const advance = Effect.fnUntraced(function* ( - db: DatabaseService, - sessionID: SessionSchema.ID, - applied: SystemContext.Applied, -) { - const updated = yield* db - .update(SessionContextEpochTable) - .set({ snapshot: applied }) - .where(eq(SessionContextEpochTable.session_id, sessionID)) - .returning({ sessionID: SessionContextEpochTable.session_id }) - .get() - .pipe(Effect.orDie) - if (!updated) return yield* Effect.die("Context Epoch not found") -}) diff --git a/packages/core/src/session/error.ts b/packages/core/src/session/error.ts index 158e46dd07..68f4fd32a4 100644 --- a/packages/core/src/session/error.ts +++ b/packages/core/src/session/error.ts @@ -10,15 +10,3 @@ export class MessageDecodeError extends Schema.TaggedErrorClass()( - "Session.ContextSnapshotDecodeError", - { - sessionID: SessionSchema.ID, - details: Schema.String, - }, -) { - override get message() { - return `Failed to decode context snapshot for session ${this.sessionID}: ${this.details}` - } -} diff --git a/packages/core/src/session/history.ts b/packages/core/src/session/history.ts index f72ff91ab8..9d84859231 100644 --- a/packages/core/src/session/history.ts +++ b/packages/core/src/session/history.ts @@ -4,7 +4,7 @@ import { Database } from "../database/database" import { MessageDecodeError } from "./error" import { SessionMessage } from "./message" import { SessionSchema } from "./schema" -import { SessionContextEpochTable, SessionMessageTable } from "./sql" +import { SessionContextCheckpointTable, SessionMessageTable } from "./sql" type DatabaseService = Database.Interface["db"] @@ -33,6 +33,9 @@ const messageRows = Effect.fnUntraced(function* ( .where( and( eq(SessionMessageTable.session_id, sessionID), + // Keep system updates visible in the gap between a completed compaction + // and the next prepared turn's rebaseline, when their content is not yet + // folded into a new baseline. compaction ? or( gte(SessionMessageTable.seq, compaction.seq), @@ -67,9 +70,9 @@ export const load = Effect.fn("SessionHistory.load")(function* (db: DatabaseServ const [epoch, compaction] = yield* Effect.all( [ db - .select({ baselineSeq: SessionContextEpochTable.baseline_seq }) - .from(SessionContextEpochTable) - .where(eq(SessionContextEpochTable.session_id, sessionID)) + .select({ baselineSeq: SessionContextCheckpointTable.baseline_seq }) + .from(SessionContextCheckpointTable) + .where(eq(SessionContextCheckpointTable.session_id, sessionID)) .get() .pipe(Effect.orDie), latestCompaction(db, sessionID), diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts index ce6ebcd4ec..37e931efb1 100644 --- a/packages/core/src/session/projector.ts +++ b/packages/core/src/session/projector.ts @@ -12,7 +12,7 @@ import { SessionMessage } from "./message" import { SessionMessageUpdater } from "./message-updater" import { SessionInput } from "./input" import { WorkspaceV2 } from "../workspace" -import { SessionContextEpoch } from "./context-epoch" +import { SessionContextCheckpoint } from "./context-checkpoint" import { MessageTable, PartTable, SessionInputTable, SessionMessageTable, SessionTable } from "./sql" import type { DeepMutable } from "../schema" import { Slug } from "../util/slug" @@ -156,12 +156,16 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* ( .select({ seq: SessionMessageTable.seq }) .from(SessionMessageTable) .where( - and(eq(SessionMessageTable.session_id, event.data.parentID), eq(SessionMessageTable.id, event.data.messageID)), + and( + eq(SessionMessageTable.session_id, event.data.parentID), + eq(SessionMessageTable.id, event.data.messageID), + ), ) .get() .pipe(Effect.orDie) : undefined - if (event.data.messageID && !boundary) return yield* Effect.die(`Fork boundary message not found: ${event.data.messageID}`) + if (event.data.messageID && !boundary) + return yield* Effect.die(`Fork boundary message not found: ${event.data.messageID}`) const copied = yield* db .select({ seq: SessionMessageTable.seq }) .from(SessionMessageTable) @@ -452,7 +456,7 @@ const layer = Layer.effectDiscard( .where(eq(SessionTable.id, event.data.sessionID)) .run() .pipe(Effect.orDie) - yield* SessionContextEpoch.reset(db, event.data.sessionID) + yield* SessionContextCheckpoint.reset(db, event.data.sessionID) }), ) yield* events.project(SessionV1.Event.Deleted, (event) => @@ -666,7 +670,7 @@ const layer = Layer.effectDiscard( .where(eq(SessionTable.id, event.data.sessionID)) .run() .pipe(Effect.orDie) - yield* SessionContextEpoch.reset(db, event.data.sessionID) + yield* SessionContextCheckpoint.reset(db, event.data.sessionID) }), ) }), diff --git a/packages/core/src/session/runner/index.ts b/packages/core/src/session/runner/index.ts index 634075dd91..11141c3447 100644 --- a/packages/core/src/session/runner/index.ts +++ b/packages/core/src/session/runner/index.ts @@ -3,7 +3,7 @@ export * as SessionRunner from "./index" import type { LLMError } from "@opencode-ai/llm" import { Context, Effect } from "effect" import { SessionSchema } from "../schema" -import type { ContextSnapshotDecodeError, MessageDecodeError } from "../error" +import type { MessageDecodeError } from "../error" import { SessionRunnerModel } from "./model" import type { SystemContext } from "../../system-context/index" import type { ToolOutputStore } from "../../tool-output-store" @@ -12,7 +12,6 @@ export type RunError = | LLMError | SessionRunnerModel.Error | MessageDecodeError - | ContextSnapshotDecodeError | SystemContext.InitializationBlocked | ToolOutputStore.Error diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 66760e1cbb..7b34f1aedb 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -24,7 +24,7 @@ import { ReferenceGuidance } from "../../reference/guidance" import { McpGuidance } from "../../mcp/guidance" import { ToolRegistry } from "../../tool/registry" import { ToolOutputStore } from "../../tool-output-store" -import { SessionContextEpoch } from "../context-epoch" +import { SessionContextCheckpoint } from "../context-checkpoint" import { SessionCompaction } from "../compaction" import { SessionEvent } from "../event" import { SessionHistory } from "../history" @@ -168,11 +168,9 @@ const layer = Layer.effect( if (session.location.directory !== location.directory || session.location.workspaceID !== location.workspaceID) return yield* Effect.interrupt const agent = yield* agents.select(session.agent) - // Epoch initialization runs before promotion so a blocked System Context fails the - // turn while prompts are still pending in the durable inbox; a retry re-drains them. - // Reconciliation (`prepare` below) runs after promotion so ContextUpdated messages - // sequence after the promoted user input. - const initialized = yield* SessionContextEpoch.initialize(db, loadSystemContext(agent), session.id) + // Establish what the model knows before admitting what the user said, so + // a blocked first turn leaves pending inputs untouched. + const checkpoint = yield* SessionContextCheckpoint.prepare(db, events, loadSystemContext(agent), session.id) const toolFibers = yield* FiberSet.make() let needsContinuation = false let currentStep = step @@ -185,10 +183,8 @@ const layer = Layer.effect( } if (promoted > 0) currentStep = 1 } - const system = - initialized ?? (yield* SessionContextEpoch.prepare(db, events, loadSystemContext(agent), session.id)) const model = yield* models.resolve(session) - const entries = yield* SessionHistory.entriesForRunner(db, session.id, system.baselineSeq) + const entries = yield* SessionHistory.entriesForRunner(db, session.id, checkpoint.baselineSeq) const context = entries.map((entry) => entry.message) const isLastStep = agent.info?.steps !== undefined && currentStep >= agent.info.steps const toolMaterialization = isLastStep @@ -198,7 +194,10 @@ const layer = Layer.effect( const request = LLM.request({ model, providerOptions: { openai: { promptCacheKey } }, - system: [agent.info?.system ? agent.info.system : SessionRunnerSystemPrompt.provider(model), system.baseline] + system: [ + agent.info?.system ? agent.info.system : SessionRunnerSystemPrompt.provider(model), + checkpoint.baseline, + ] .filter((part): part is string => part !== undefined && part.length > 0) .map(SystemPart.make), messages: [...toLLMMessages(context, model), ...(isLastStep ? [Message.assistant(MAX_STEPS_PROMPT)] : [])], diff --git a/packages/core/src/session/sql.ts b/packages/core/src/session/sql.ts index 63413bf40a..5ff659be22 100644 --- a/packages/core/src/session/sql.ts +++ b/packages/core/src/session/sql.ts @@ -165,7 +165,7 @@ export const SessionInputTable = sqliteTable( ], ) -export const SessionContextEpochTable = sqliteTable("session_context_epoch", { +export const SessionContextCheckpointTable = sqliteTable("session_context_epoch", { session_id: text() .$type() .primaryKey() diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index c8c9340aff..f822fbe9c6 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -25,7 +25,6 @@ import { QuestionV2 } from "@opencode-ai/core/question" import { AbsolutePath } from "@opencode-ai/core/schema" import { SessionV2 } from "@opencode-ai/core/session" import { Snapshot } from "@opencode-ai/core/snapshot" -import { ContextSnapshotDecodeError } from "@opencode-ai/core/session/error" import { SessionEvent } from "@opencode-ai/core/session/event" import { SessionCompaction } from "@opencode-ai/core/session/compaction" import { SessionTitle } from "@opencode-ai/core/session/title" @@ -46,7 +45,7 @@ import { Config } from "@opencode-ai/core/config" import { ConfigCompaction } from "@opencode-ai/core/config/compaction" import { Tool } from "@opencode-ai/core/tool/tool" import { - SessionContextEpochTable, + SessionContextCheckpointTable, SessionInputTable, SessionMessageTable, SessionTable, @@ -708,8 +707,8 @@ describe("SessionRunnerLLM", () => { expect( yield* db .select() - .from(SessionContextEpochTable) - .where(eq(SessionContextEpochTable.session_id, sessionID)) + .from(SessionContextCheckpointTable) + .where(eq(SessionContextCheckpointTable.session_id, sessionID)) .get(), ).toBeUndefined() @@ -741,8 +740,8 @@ describe("SessionRunnerLLM", () => { expect( yield* db .select() - .from(SessionContextEpochTable) - .where(eq(SessionContextEpochTable.session_id, sessionID)) + .from(SessionContextCheckpointTable) + .where(eq(SessionContextCheckpointTable.session_id, sessionID)) .get(), ).toBeUndefined() @@ -755,7 +754,7 @@ describe("SessionRunnerLLM", () => { }), ) - it.effect("fails gracefully when a stored context snapshot cannot be decoded", () => + it.effect("heals an undecodable stored applied record by re-announcing context", () => Effect.gen(function* () { yield* setup const session = yield* SessionV2.Service @@ -764,19 +763,28 @@ describe("SessionRunnerLLM", () => { response = [] yield* session.resume(sessionID) yield* db - .update(SessionContextEpochTable) + .update(SessionContextCheckpointTable) .set({ snapshot: { invalid: { value: "bad" } } }) - .where(eq(SessionContextEpochTable.session_id, sessionID)) + .where(eq(SessionContextCheckpointTable.session_id, sessionID)) .run() .pipe(Effect.orDie) yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Second" }), resume: false }) requests.length = 0 - const exit = yield* session.resume(sessionID).pipe(Effect.exit) + yield* session.resume(sessionID) - expect(Exit.isFailure(exit)).toBe(true) - if (Exit.isFailure(exit)) expect(Cause.squash(exit.cause)).toBeInstanceOf(ContextSnapshotDecodeError) - expect(requests).toHaveLength(0) + // Comparison state was lost, so every source re-announces as new. + expect(requests).toHaveLength(1) + expect(requests[0]?.system.map((part) => part.text)).toEqual([defaultSystem, "Initial context"]) + expect(requests[0]?.messages.map((message) => message.role)).toEqual(["user", "system", "user"]) + expect(requests[0]?.messages.at(1)?.content).toEqual([{ type: "text", text: "Initial context" }]) + const healed = yield* db + .select({ snapshot: SessionContextCheckpointTable.snapshot }) + .from(SessionContextCheckpointTable) + .where(eq(SessionContextCheckpointTable.session_id, sessionID)) + .get() + .pipe(Effect.orDie) + expect(healed?.snapshot).toEqual({ "test/context": { value: "Initial context", removed: expect.any(String) } }) }), ) @@ -797,8 +805,8 @@ describe("SessionRunnerLLM", () => { [defaultSystem, "Initial context"], [defaultSystem, "Initial context"], ]) - expect(requests[1]?.messages.map((message) => message.role)).toEqual(["user", "user", "system"]) - expect(requests[1]?.messages.at(-1)?.content).toEqual([{ type: "text", text: "Changed context" }]) + expect(requests[1]?.messages.map((message) => message.role)).toEqual(["user", "system", "user"]) + expect(requests[1]?.messages.at(1)?.content).toEqual([{ type: "text", text: "Changed context" }]) expect(yield* session.messages({ sessionID })).toHaveLength(3) const { db } = yield* Database.Service expect( @@ -1059,8 +1067,8 @@ describe("SessionRunnerLLM", () => { yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Second" }), resume: false }) yield* session.resume(sessionID) - expect(requests[1]?.messages.map((message) => message.role)).toEqual(["user", "user", "system"]) - expect(requests[1]?.messages.at(-1)?.content).toEqual([ + expect(requests[1]?.messages.map((message) => message.role)).toEqual(["user", "system", "user"]) + expect(requests[1]?.messages.at(1)?.content).toEqual([ { type: "text", text: "System context source removed: test/context" }, ]) expect(yield* session.messages({ sessionID })).toHaveLength(3) @@ -1095,15 +1103,15 @@ describe("SessionRunnerLLM", () => { [defaultSystem, "Initial context"], [defaultSystem, "Initial context"], ]) - expect(requests[1]?.messages.map((message) => message.role)).toEqual(["user", "user", "system"]) + expect(requests[1]?.messages.map((message) => message.role)).toEqual(["user", "system", "user"]) expect(requests[2]?.messages.filter((message) => message.role === "system")).toHaveLength(2) expect((yield* session.context(sessionID)).map((message) => message.type)).toEqual([ - "user", "user", "system", + "user", "model-switched", - "user", "system", + "user", ]) yield* replaySessionProjection(sessionID) expect(yield* session.messages({ sessionID })).toHaveLength(6)