ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(core): replace context epoch with a single-entry checkpoint

Browse source 
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.
This commit is contained in:
Kit Langton 2026-07-02 00:21:56 -04:00
commit 67b4b2894f
10 changed files with 193 additions and 229 deletions
Download patch file Download diff file Expand all files Collapse all files

10
packages/core/src/session.ts
View file

@ -108,7 +108,7 @@ export class OperationUnavailableError extends Schema.TaggedErrorClass<Operation
}, },
) {} ) {}
export { ContextSnapshotDecodeError, MessageDecodeError } from "./session/error" export { MessageDecodeError } from "./session/error"
export class PromptConflictError extends Schema.TaggedErrorClass<PromptConflictError>()("Session.PromptConflictError", { export class PromptConflictError extends Schema.TaggedErrorClass<PromptConflictError>()("Session.PromptConflictError", {
sessionID: SessionSchema.ID, sessionID: SessionSchema.ID,
@ -466,7 +466,9 @@ const layer = Layer.effect(
text: skill.content, text: skill.content,
}) })
if (input.resume !== false) 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) { switchAgent: Effect.fn("V2Session.switchAgent")(function* (input) {
yield* result.get(input.sessionID) yield* result.get(input.sessionID)
@ -550,7 +552,9 @@ const layer = Layer.effect(
description: input.description, description: input.description,
metadata: input.metadata, 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) => interrupt: Effect.fn("V2Session.interrupt")((sessionID) =>
Effect.uninterruptible(execution.interrupt(sessionID)), Effect.uninterruptible(execution.interrupt(sessionID)),

131
packages/core/src/session/context-checkpoint.ts Normal file
View file

@ -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<SystemContext.SystemContext>,
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")
})

172
packages/core/src/session/context-epoch.ts
View file

@ -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<SystemContext.SystemContext>,
sessionID: SessionSchema.ID,
): Effect.Effect<Prepared | undefined, SystemContext.InitializationBlocked> {
return initializeOnce(db, context, sessionID).pipe(Effect.withSpan("SessionContextEpoch.initialize"))
}
export function prepare(
db: DatabaseService,
events: EventV2.Interface,
context: Effect.Effect<SystemContext.SystemContext>,
sessionID: SessionSchema.ID,
): Effect.Effect<Prepared, SystemContext.InitializationBlocked | ContextSnapshotDecodeError> {
return prepareOnce(db, events, context, sessionID).pipe(Effect.withSpan("SessionContextEpoch.prepare"))
}
const prepareOnce = Effect.fnUntraced(function* (
db: DatabaseService,
events: EventV2.Interface,
context: Effect.Effect<SystemContext.SystemContext>,
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<SystemContext.SystemContext>,
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")
})

12
packages/core/src/session/error.ts
View file

@ -10,15 +10,3 @@ export class MessageDecodeError extends Schema.TaggedErrorClass<MessageDecodeErr
return `Failed to decode message ${this.messageID} in session ${this.sessionID}` return `Failed to decode message ${this.messageID} in session ${this.sessionID}`
} }
} }
export class ContextSnapshotDecodeError extends Schema.TaggedErrorClass<ContextSnapshotDecodeError>()(
"Session.ContextSnapshotDecodeError",
{
sessionID: SessionSchema.ID,
details: Schema.String,
},
) {
override get message() {
return `Failed to decode context snapshot for session ${this.sessionID}: ${this.details}`
}
}

11
packages/core/src/session/history.ts
View file

@ -4,7 +4,7 @@ import { Database } from "../database/database"
import { MessageDecodeError } from "./error" import { MessageDecodeError } from "./error"
import { SessionMessage } from "./message" import { SessionMessage } from "./message"
import { SessionSchema } from "./schema" import { SessionSchema } from "./schema"
import { SessionContextEpochTable, SessionMessageTable } from "./sql" import { SessionContextCheckpointTable, SessionMessageTable } from "./sql"
type DatabaseService = Database.Interface["db"] type DatabaseService = Database.Interface["db"]
@ -33,6 +33,9 @@ const messageRows = Effect.fnUntraced(function* (
.where( .where(
and( and(
eq(SessionMessageTable.session_id, sessionID), 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 compaction
? or( ? or(
gte(SessionMessageTable.seq, compaction.seq), 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( const [epoch, compaction] = yield* Effect.all(
[ [
db db
.select({ baselineSeq: SessionContextEpochTable.baseline_seq }) .select({ baselineSeq: SessionContextCheckpointTable.baseline_seq })
.from(SessionContextEpochTable) .from(SessionContextCheckpointTable)
.where(eq(SessionContextEpochTable.session_id, sessionID)) .where(eq(SessionContextCheckpointTable.session_id, sessionID))
.get() .get()
.pipe(Effect.orDie), .pipe(Effect.orDie),
latestCompaction(db, sessionID), latestCompaction(db, sessionID),

14
packages/core/src/session/projector.ts
View file

@ -12,7 +12,7 @@ import { SessionMessage } from "./message"
import { SessionMessageUpdater } from "./message-updater" import { SessionMessageUpdater } from "./message-updater"
import { SessionInput } from "./input" import { SessionInput } from "./input"
import { WorkspaceV2 } from "../workspace" import { WorkspaceV2 } from "../workspace"
import { SessionContextEpoch } from "./context-epoch" import { SessionContextCheckpoint } from "./context-checkpoint"
import { MessageTable, PartTable, SessionInputTable, SessionMessageTable, SessionTable } from "./sql" import { MessageTable, PartTable, SessionInputTable, SessionMessageTable, SessionTable } from "./sql"
import type { DeepMutable } from "../schema" import type { DeepMutable } from "../schema"
import { Slug } from "../util/slug" import { Slug } from "../util/slug"
@ -156,12 +156,16 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* (
.select({ seq: SessionMessageTable.seq }) .select({ seq: SessionMessageTable.seq })
.from(SessionMessageTable) .from(SessionMessageTable)
.where( .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() .get()
.pipe(Effect.orDie) .pipe(Effect.orDie)
: undefined : 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 const copied = yield* db
.select({ seq: SessionMessageTable.seq }) .select({ seq: SessionMessageTable.seq })
.from(SessionMessageTable) .from(SessionMessageTable)
@ -452,7 +456,7 @@ const layer = Layer.effectDiscard(
.where(eq(SessionTable.id, event.data.sessionID)) .where(eq(SessionTable.id, event.data.sessionID))
.run() .run()
.pipe(Effect.orDie) .pipe(Effect.orDie)
yield* SessionContextEpoch.reset(db, event.data.sessionID) yield* SessionContextCheckpoint.reset(db, event.data.sessionID)
}), }),
) )
yield* events.project(SessionV1.Event.Deleted, (event) => yield* events.project(SessionV1.Event.Deleted, (event) =>
@ -666,7 +670,7 @@ const layer = Layer.effectDiscard(
.where(eq(SessionTable.id, event.data.sessionID)) .where(eq(SessionTable.id, event.data.sessionID))
.run() .run()
.pipe(Effect.orDie) .pipe(Effect.orDie)
yield* SessionContextEpoch.reset(db, event.data.sessionID) yield* SessionContextCheckpoint.reset(db, event.data.sessionID)
}), }),
) )
}), }),

3
packages/core/src/session/runner/index.ts
View file

@ -3,7 +3,7 @@ export * as SessionRunner from "./index"
import type { LLMError } from "@opencode-ai/llm" import type { LLMError } from "@opencode-ai/llm"
import { Context, Effect } from "effect" import { Context, Effect } from "effect"
import { SessionSchema } from "../schema" import { SessionSchema } from "../schema"
import type { ContextSnapshotDecodeError, MessageDecodeError } from "../error" import type { MessageDecodeError } from "../error"
import { SessionRunnerModel } from "./model" import { SessionRunnerModel } from "./model"
import type { SystemContext } from "../../system-context/index" import type { SystemContext } from "../../system-context/index"
import type { ToolOutputStore } from "../../tool-output-store" import type { ToolOutputStore } from "../../tool-output-store"
@ -12,7 +12,6 @@ export type RunError =
| LLMError | LLMError
| SessionRunnerModel.Error | SessionRunnerModel.Error
| MessageDecodeError | MessageDecodeError
| ContextSnapshotDecodeError
| SystemContext.InitializationBlocked | SystemContext.InitializationBlocked
| ToolOutputStore.Error | ToolOutputStore.Error

19
packages/core/src/session/runner/llm.ts
View file

@ -24,7 +24,7 @@ import { ReferenceGuidance } from "../../reference/guidance"
import { McpGuidance } from "../../mcp/guidance" import { McpGuidance } from "../../mcp/guidance"
import { ToolRegistry } from "../../tool/registry" import { ToolRegistry } from "../../tool/registry"
import { ToolOutputStore } from "../../tool-output-store" import { ToolOutputStore } from "../../tool-output-store"
import { SessionContextEpoch } from "../context-epoch" import { SessionContextCheckpoint } from "../context-checkpoint"
import { SessionCompaction } from "../compaction" import { SessionCompaction } from "../compaction"
import { SessionEvent } from "../event" import { SessionEvent } from "../event"
import { SessionHistory } from "../history" import { SessionHistory } from "../history"
@ -168,11 +168,9 @@ const layer = Layer.effect(
if (session.location.directory !== location.directory || session.location.workspaceID !== location.workspaceID) if (session.location.directory !== location.directory || session.location.workspaceID !== location.workspaceID)
return yield* Effect.interrupt return yield* Effect.interrupt
const agent = yield* agents.select(session.agent) const agent = yield* agents.select(session.agent)
// Epoch initialization runs before promotion so a blocked System Context fails the // Establish what the model knows before admitting what the user said, so
// turn while prompts are still pending in the durable inbox; a retry re-drains them. // a blocked first turn leaves pending inputs untouched.
// Reconciliation (`prepare` below) runs after promotion so ContextUpdated messages const checkpoint = yield* SessionContextCheckpoint.prepare(db, events, loadSystemContext(agent), session.id)
// sequence after the promoted user input.
const initialized = yield* SessionContextEpoch.initialize(db, loadSystemContext(agent), session.id)
const toolFibers = yield* FiberSet.make<void, ToolOutputStore.Error>() const toolFibers = yield* FiberSet.make<void, ToolOutputStore.Error>()
let needsContinuation = false let needsContinuation = false
let currentStep = step let currentStep = step
@ -185,10 +183,8 @@ const layer = Layer.effect(
} }
if (promoted > 0) currentStep = 1 if (promoted > 0) currentStep = 1
} }
const system =
initialized ?? (yield* SessionContextEpoch.prepare(db, events, loadSystemContext(agent), session.id))
const model = yield* models.resolve(session) 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 context = entries.map((entry) => entry.message)
const isLastStep = agent.info?.steps !== undefined && currentStep >= agent.info.steps const isLastStep = agent.info?.steps !== undefined && currentStep >= agent.info.steps
const toolMaterialization = isLastStep const toolMaterialization = isLastStep
@ -198,7 +194,10 @@ const layer = Layer.effect(
const request = LLM.request({ const request = LLM.request({
model, model,
providerOptions: { openai: { promptCacheKey } }, 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) .filter((part): part is string => part !== undefined && part.length > 0)
.map(SystemPart.make), .map(SystemPart.make),
messages: [...toLLMMessages(context, model), ...(isLastStep ? [Message.assistant(MAX_STEPS_PROMPT)] : [])], messages: [...toLLMMessages(context, model), ...(isLastStep ? [Message.assistant(MAX_STEPS_PROMPT)] : [])],

2
packages/core/src/session/sql.ts
View file

@ -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() session_id: text()
.$type<SessionSchema.ID>() .$type<SessionSchema.ID>()
.primaryKey() .primaryKey()

48
packages/core/test/session-runner.test.ts
View file

@ -25,7 +25,6 @@ import { QuestionV2 } from "@opencode-ai/core/question"
import { AbsolutePath } from "@opencode-ai/core/schema" import { AbsolutePath } from "@opencode-ai/core/schema"
import { SessionV2 } from "@opencode-ai/core/session" import { SessionV2 } from "@opencode-ai/core/session"
import { Snapshot } from "@opencode-ai/core/snapshot" import { Snapshot } from "@opencode-ai/core/snapshot"
import { ContextSnapshotDecodeError } from "@opencode-ai/core/session/error"
import { SessionEvent } from "@opencode-ai/core/session/event" import { SessionEvent } from "@opencode-ai/core/session/event"
import { SessionCompaction } from "@opencode-ai/core/session/compaction" import { SessionCompaction } from "@opencode-ai/core/session/compaction"
import { SessionTitle } from "@opencode-ai/core/session/title" 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 { ConfigCompaction } from "@opencode-ai/core/config/compaction"
import { Tool } from "@opencode-ai/core/tool/tool" import { Tool } from "@opencode-ai/core/tool/tool"
import { import {
SessionContextEpochTable, SessionContextCheckpointTable,
SessionInputTable, SessionInputTable,
SessionMessageTable, SessionMessageTable,
SessionTable, SessionTable,
@ -708,8 +707,8 @@ describe("SessionRunnerLLM", () => {
expect( expect(
yield* db yield* db
.select() .select()
.from(SessionContextEpochTable) .from(SessionContextCheckpointTable)
.where(eq(SessionContextEpochTable.session_id, sessionID)) .where(eq(SessionContextCheckpointTable.session_id, sessionID))
.get(), .get(),
).toBeUndefined() ).toBeUndefined()
@ -741,8 +740,8 @@ describe("SessionRunnerLLM", () => {
expect( expect(
yield* db yield* db
.select() .select()
.from(SessionContextEpochTable) .from(SessionContextCheckpointTable)
.where(eq(SessionContextEpochTable.session_id, sessionID)) .where(eq(SessionContextCheckpointTable.session_id, sessionID))
.get(), .get(),
).toBeUndefined() ).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* () { Effect.gen(function* () {
yield* setup yield* setup
const session = yield* SessionV2.Service const session = yield* SessionV2.Service
@ -764,19 +763,28 @@ describe("SessionRunnerLLM", () => {
response = [] response = []
yield* session.resume(sessionID) yield* session.resume(sessionID)
yield* db yield* db
.update(SessionContextEpochTable) .update(SessionContextCheckpointTable)
.set({ snapshot: { invalid: { value: "bad" } } }) .set({ snapshot: { invalid: { value: "bad" } } })
.where(eq(SessionContextEpochTable.session_id, sessionID)) .where(eq(SessionContextCheckpointTable.session_id, sessionID))
.run() .run()
.pipe(Effect.orDie) .pipe(Effect.orDie)
yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Second" }), resume: false }) yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Second" }), resume: false })
requests.length = 0 requests.length = 0
const exit = yield* session.resume(sessionID).pipe(Effect.exit) yield* session.resume(sessionID)
expect(Exit.isFailure(exit)).toBe(true) // Comparison state was lost, so every source re-announces as new.
if (Exit.isFailure(exit)) expect(Cause.squash(exit.cause)).toBeInstanceOf(ContextSnapshotDecodeError) expect(requests).toHaveLength(1)
expect(requests).toHaveLength(0) 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"],
[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[1]?.messages.at(-1)?.content).toEqual([{ type: "text", text: "Changed context" }]) expect(requests[1]?.messages.at(1)?.content).toEqual([{ type: "text", text: "Changed context" }])
expect(yield* session.messages({ sessionID })).toHaveLength(3) expect(yield* session.messages({ sessionID })).toHaveLength(3)
const { db } = yield* Database.Service const { db } = yield* Database.Service
expect( expect(
@ -1059,8 +1067,8 @@ describe("SessionRunnerLLM", () => {
yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Second" }), resume: false }) yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Second" }), resume: false })
yield* session.resume(sessionID) yield* session.resume(sessionID)
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[1]?.messages.at(-1)?.content).toEqual([ expect(requests[1]?.messages.at(1)?.content).toEqual([
{ type: "text", text: "System context source removed: test/context" }, { type: "text", text: "System context source removed: test/context" },
]) ])
expect(yield* session.messages({ sessionID })).toHaveLength(3) expect(yield* session.messages({ sessionID })).toHaveLength(3)
@ -1095,15 +1103,15 @@ describe("SessionRunnerLLM", () => {
[defaultSystem, "Initial context"], [defaultSystem, "Initial context"],
[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(requests[2]?.messages.filter((message) => message.role === "system")).toHaveLength(2)
expect((yield* session.context(sessionID)).map((message) => message.type)).toEqual([ expect((yield* session.context(sessionID)).map((message) => message.type)).toEqual([
"user",
"user", "user",
"system", "system",
"user",
"model-switched", "model-switched",
"user",
"system", "system",
"user",
]) ])
yield* replaySessionProjection(sessionID) yield* replaySessionProjection(sessionID)
expect(yield* session.messages({ sessionID })).toHaveLength(6) expect(yield* session.messages({ sessionID })).toHaveLength(6)