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)