diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index ce59c9c12b7..3a65e5812ca 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -420,6 +420,7 @@ export type Endpoint5_26Output = readonly data: { readonly sessionID: Session.ID readonly delta: { readonly [x: string]: (string & Brand.Brand<"Instruction.Hash">) | "removed" } + readonly text?: string | undefined } } | { diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index a8f8d88b989..bb1da0aa2fe 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -676,7 +676,7 @@ export type SessionInstructionsUpdated = { type: "session.instructions.updated" durable: { aggregateID: string; seq: number; version: 2 } location?: LocationRef - data: { sessionID: string; delta: { [x: string]: string | "removed" } } + data: { sessionID: string; delta: { [x: string]: string | "removed" }; text?: string } } export type SessionSynthetic = { diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index e438e67bb26..c7628ec799d 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -410,7 +410,7 @@ const layer = Layer.effect( fork: Effect.fn("Session.fork")(function* (input) { const parent = yield* result.get(input.sessionID) const boundary = yield* db - .select({ id: SessionMessageTable.id, seq: SessionMessageTable.seq }) + .select({ id: SessionMessageTable.id }) .from(SessionMessageTable) .where( and( @@ -429,13 +429,14 @@ const layer = Layer.effect( }) if (!boundary) return yield* new ForkEmptyError({ sessionID: input.sessionID }) const sessionID = SessionSchema.ID.create() - const instructionThrough = - input.boundary.type === "before" ? boundary.seq - 1 : yield* Bus.latestSequence(db, parent.id) + // The fork adopts the parent's newest instruction values rather than the + // values in effect at the boundary; copied history may contain frozen + // instruction-update text the initial baseline already reflects. yield* bus.publish(SessionEvent.Forked, { sessionID, parentID: parent.id, boundary: { ...input.boundary, messageID: boundary.id }, - instructions: yield* InstructionState.valuesAt(db, parent.id, instructionThrough), + instructions: yield* InstructionState.current(db, parent.id), }) return yield* result.get(sessionID).pipe(Effect.orDie) }), diff --git a/packages/core/src/session/history.ts b/packages/core/src/session/history.ts index 85114d35886..8aacf032af3 100644 --- a/packages/core/src/session/history.ts +++ b/packages/core/src/session/history.ts @@ -80,10 +80,9 @@ export const entriesForRunner = Effect.fn("SessionHistory.entriesForRunner")(fun .transaction(() => Effect.gen(function* () { const messages = yield* messageEntries(db, sessionID) - const assembled = yield* InstructionState.assemble(db, sessionID, instructions) return { - initial: assembled.initial, - entries: [...messages, ...assembled.updates].toSorted((a, b) => a.seq - b.seq), + initial: yield* InstructionState.initial(db, sessionID, instructions), + entries: messages, } }), ) @@ -106,10 +105,9 @@ export const preview = Effect.fn("SessionHistory.preview")(function* ( ) const settled = unsettled === -1 ? messages : messages.slice(0, unsettled) const assembled = yield* InstructionState.preview(db, sessionID, instructions, observed) - const entries = [...settled, ...assembled.updates].toSorted((a, b) => a.seq - b.seq) return { initial: assembled.initial, - messages: entries.map((entry) => entry.message), + messages: settled.map((entry) => entry.message), instructionUpdate: assembled.update, } }), diff --git a/packages/core/src/session/instruction-state.ts b/packages/core/src/session/instruction-state.ts index 119bff62cae..c366b592591 100644 --- a/packages/core/src/session/instruction-state.ts +++ b/packages/core/src/session/instruction-state.ts @@ -1,25 +1,20 @@ export * as InstructionState from "./instruction-state" -import { and, asc, desc, eq, gt, inArray, lte, sql } from "drizzle-orm" -import { DateTime, Effect, Option, Schema } from "effect" +import { eq, inArray, sql } from "drizzle-orm" +import { Effect, Option, Schema } from "effect" import type { Database } from "../database/database" -import { Bus } from "../bus" -import { EventTable } from "../event/sql" +import type { Bus } from "../bus" import { Instructions } from "../instructions/index" import { SessionEvent } from "./event" -import { SessionMessage } from "./message" -import { Event } from "@opencode-ai/schema/event" import { SessionSchema } from "./schema" import { InstructionBlobTable, InstructionStateTable } from "./sql" type DatabaseService = Database.Interface["db"] -const decodeInstructionsUpdated = Schema.decodeUnknownSync(SessionEvent.InstructionsUpdated.data) -const decodeForked = Schema.decodeUnknownSync(SessionEvent.Forked.data) - export interface Observation extends Instructions.Admission { readonly sessionID: SessionSchema.ID readonly initial: boolean + readonly previous: Instructions.Values readonly current: Instructions.Values } @@ -28,13 +23,14 @@ export const observe = Effect.fn("InstructionState.observe")(function* ( instructions: Instructions.Instructions, sessionID: SessionSchema.ID, ): Effect.fn.Return { - const [observed, stored] = yield* Effect.all([Instructions.read(instructions), ensure(db, sessionID)], { + const [observed, stored] = yield* Effect.all([Instructions.read(instructions), find(db, sessionID)], { concurrency: "unbounded", }) const result = yield* observeAgainst(observed, stored?.current_values) return { sessionID, initial: !stored, + previous: stored?.current_values ?? {}, ...result, } }) @@ -42,12 +38,20 @@ export const observe = Effect.fn("InstructionState.observe")(function* ( export const commit = Effect.fn("InstructionState.commit")(function* ( db: DatabaseService, bus: Bus.Interface, + instructions: Instructions.Instructions, observation: Observation, ) { if (!observation.initial && Object.keys(observation.delta).length === 0) return + // The rendered text is frozen into the durable event: replaying it later would + // require the Location-scoped registry that produced it. + const text = observation.initial ? "" : yield* renderUpdateText(db, instructions, observation) yield* bus.publish( SessionEvent.InstructionsUpdated, - { sessionID: observation.sessionID, delta: observation.delta }, + { + sessionID: observation.sessionID, + delta: observation.delta, + ...(text.length > 0 ? { text } : {}), + }, { // Initial sync establishes the baseline; unlike later deltas it is not chronological history. ...(observation.initial ? { metadata: { instructions: { initial: true } } } : {}), @@ -56,13 +60,27 @@ export const commit = Effect.fn("InstructionState.commit")(function* ( ) }) +const renderUpdateText = Effect.fnUntraced(function* ( + db: DatabaseService, + instructions: Instructions.Instructions, + observation: Observation, +) { + const replaced = Object.entries(observation.previous).filter(([key]) => Object.hasOwn(observation.delta, key)) + const blobs = yield* loadBlobs(db, replaced.map(([, hash]) => hash)) + const previous = Object.fromEntries(replaced.map(([key, hash]) => [key, requireBlob(blobs, hash)])) + const admitted = new Map( + Object.entries(observation.blobs).map(([hash, value]) => [Instructions.Hash.make(hash), value]), + ) + return Instructions.renderUpdate(instructions, previous, dereferenceDelta(observation.delta, admitted)) +}) + export const prepare = Effect.fn("InstructionState.prepare")(function* ( db: DatabaseService, bus: Bus.Interface, instructions: Instructions.Instructions, sessionID: SessionSchema.ID, ) { - yield* commit(db, bus, yield* observe(db, instructions, sessionID)) + yield* commit(db, bus, instructions, yield* observe(db, instructions, sessionID)) }) export const apply = Effect.fn("InstructionState.apply")(function* ( @@ -140,79 +158,24 @@ export const reset = Effect.fn("InstructionState.reset")(function* (db: Database .pipe(Effect.orDie) }) -export const rebuild = Effect.fn("InstructionState.rebuild")(function* ( - db: DatabaseService, - sessionID: SessionSchema.ID, -) { - const state = yield* stateFromEvents(db, sessionID) - if (!state) { - yield* reset(db, sessionID) - return undefined - } - yield* db - .insert(InstructionStateTable) - .values(state) - .onConflictDoUpdate({ - target: InstructionStateTable.session_id, - set: { - epoch_start: state.epoch_start, - through_seq: state.through_seq, - initial_values: state.initial_values, - current_values: state.current_values, - }, - }) - .run() - .pipe(Effect.orDie) - return state -}) - -const assembleState = Effect.fnUntraced(function* ( - db: DatabaseService, - sessionID: SessionSchema.ID, - instructions: Instructions.Instructions, - state: typeof InstructionStateTable.$inferSelect, -) { - const rows = yield* instructionUpdatesAfter(db, sessionID, state.epoch_start) - const updates = rows.map((row) => ({ - row, - delta: decodeInstructionsUpdated(row.data).delta, - })) - const blobs = yield* loadBlobs(db, [ - ...Object.values(state.initial_values), - ...updates.flatMap((update) => - Object.values(update.delta).filter((hash): hash is Instructions.Hash => hash !== "removed"), - ), - ]) - const valuesAtStart = dereference(state.initial_values, blobs) - let values = valuesAtStart - const result: Array<{ readonly seq: number; readonly message: SessionMessage.System }> = [] - for (const update of updates) { - const delta = dereferenceDelta(update.delta, blobs) - const text = Instructions.renderUpdate(instructions, values, delta) - if (text.length > 0) - result.push({ - seq: update.row.seq, - message: SessionMessage.System.make({ - id: SessionMessage.ID.fromEvent(Event.ID.make(update.row.id)), - type: "system", - text, - time: { created: DateTime.makeUnsafe(update.row.created) }, - }), - }) - values = Instructions.applyDelta(values, delta) - } - return { initial: Instructions.renderInitial(instructions, valuesAtStart), updates: result, current: values } -}) - -export const assemble = Effect.fn("InstructionState.assemble")(function* ( +/** Renders the epoch baseline shown at the start of every model request. */ +export const initial = Effect.fn("InstructionState.initial")(function* ( db: DatabaseService, sessionID: SessionSchema.ID, instructions: Instructions.Instructions, ) { const state = yield* find(db, sessionID) if (!state) return yield* Effect.die(new Error(`Instruction state not found during assembly: ${sessionID}`)) - const assembled = yield* assembleState(db, sessionID, instructions, state) - return { initial: assembled.initial, updates: assembled.updates } + const blobs = yield* loadBlobs(db, Object.values(state.initial_values)) + return Instructions.renderInitial(instructions, dereference(state.initial_values, blobs)) +}) + +/** The current instruction values, used to seed a fork's baseline. */ +export const current = Effect.fn("InstructionState.current")(function* ( + db: DatabaseService, + sessionID: SessionSchema.ID, +) { + return (yield* find(db, sessionID))?.current_values }) export const preview = Effect.fn("InstructionState.preview")(function* ( @@ -221,20 +184,26 @@ export const preview = Effect.fn("InstructionState.preview")(function* ( instructions: Instructions.Instructions, observed: Instructions.ReadResult, ) { - const state = yield* readState(db, sessionID) + const state = yield* find(db, sessionID) const result = yield* observeAgainst(observed, state?.current_values) - const blobs = new Map( + const observedBlobs = new Map( Object.entries(result.blobs).map(([hash, value]) => [Instructions.Hash.make(hash), value]), ) if (!state) { - const values = dereference(result.current, blobs) - return { initial: Instructions.renderInitial(instructions, values), updates: [], update: "" } + const values = dereference(result.current, observedBlobs) + return { initial: Instructions.renderInitial(instructions, values), update: "" } } - const assembled = yield* assembleState(db, sessionID, instructions, state) + const stored = yield* loadBlobs(db, [ + ...Object.values(state.initial_values), + ...Object.values(state.current_values), + ]) return { - initial: assembled.initial, - updates: assembled.updates, - update: Instructions.renderUpdate(instructions, assembled.current, dereferenceDelta(result.delta, blobs)), + initial: Instructions.renderInitial(instructions, dereference(state.initial_values, stored)), + update: Instructions.renderUpdate( + instructions, + dereference(state.current_values, stored), + dereferenceDelta(result.delta, new Map([...stored, ...observedBlobs])), + ), } }) @@ -255,46 +224,6 @@ const find = Effect.fnUntraced(function* (db: DatabaseService, sessionID: Sessio .pipe(Effect.orDie) }) -const ensure = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) { - const stored = yield* find(db, sessionID) - if (!stored) return yield* rebuild(db, sessionID) - const latest = yield* latestRelevantSequence(db, sessionID) - if (!latest || latest.seq <= stored.through_seq) return stored - return yield* rebuild(db, sessionID) -}) - -const readState = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) { - const stored = yield* find(db, sessionID) - if (!stored) return yield* stateFromEvents(db, sessionID) - const latest = yield* latestRelevantSequence(db, sessionID) - if (!latest || latest.seq <= stored.through_seq) return stored - return yield* stateFromEvents(db, sessionID) -}) - -const stateFromEvents = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) { - const folded = fold(yield* instructionEvents(db, sessionID)) - return folded ? foldedState(sessionID, folded) : undefined -}) - -export const valuesAt = Effect.fn("InstructionState.valuesAt")(function* ( - db: DatabaseService, - sessionID: SessionSchema.ID, - through: number, -) { - return fold(yield* instructionEvents(db, sessionID, through))?.current -}) - -const latestRelevantSequence = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) { - return yield* db - .select({ seq: EventTable.seq }) - .from(EventTable) - .where(and(eq(EventTable.aggregate_id, sessionID), inArray(EventTable.type, relevantEventTypes))) - .orderBy(desc(EventTable.seq)) - .limit(1) - .get() - .pipe(Effect.orDie) -}) - const insertBlobs = Effect.fnUntraced(function* (db: DatabaseService, blobs: Readonly>) { const rows = Object.entries(blobs).map(([hash, value]) => ({ hash: Instructions.Hash.make(hash), value })) if (rows.length === 0) return @@ -339,106 +268,3 @@ function requireBlob(blobs: ReadonlyMap, hash: I if (value === undefined) throw new Error(`Instruction blob not found: ${hash}`) return value } - -const instructionEventType = Bus.versionedType( - SessionEvent.InstructionsUpdated.type, - SessionEvent.InstructionsUpdated.durable.version, -) -const compactionEventType = Bus.versionedType( - SessionEvent.Compaction.Ended.type, - SessionEvent.Compaction.Ended.durable.version, -) -const movedEventType = Bus.versionedType(SessionEvent.Moved.type, SessionEvent.Moved.durable.version) -const revertedEventType = Bus.versionedType( - SessionEvent.RevertEvent.Committed.type, - SessionEvent.RevertEvent.Committed.durable.version, -) -const forkedEventType = Bus.versionedType(SessionEvent.Forked.type, SessionEvent.Forked.durable.version) -const relevantEventTypes = [ - forkedEventType, - instructionEventType, - compactionEventType, - movedEventType, - revertedEventType, -] - -type InstructionEventRow = typeof EventTable.$inferSelect - -const instructionEvents = Effect.fnUntraced(function* ( - db: DatabaseService, - sessionID: SessionSchema.ID, - through?: number, -): Effect.fn.Return> { - return yield* eventRows(db, sessionID, relevantEventTypes, undefined, through) -}) - -const instructionUpdatesAfter = Effect.fnUntraced(function* ( - db: DatabaseService, - sessionID: SessionSchema.ID, - after: number, -) { - return yield* eventRows(db, sessionID, [instructionEventType], after) -}) - -const eventRows = Effect.fnUntraced(function* ( - db: DatabaseService, - sessionID: SessionSchema.ID, - types: ReadonlyArray, - after?: number, - through?: number, -): Effect.fn.Return> { - return yield* db - .select() - .from(EventTable) - .where( - and( - eq(EventTable.aggregate_id, sessionID), - inArray(EventTable.type, types), - after === undefined ? undefined : gt(EventTable.seq, after), - through === undefined ? undefined : lte(EventTable.seq, through), - ), - ) - .orderBy(asc(EventTable.seq)) - .all() - .pipe(Effect.orDie) -}) - -function fold(rows: ReadonlyArray) { - return rows.reduce< - | { - readonly epochStart: number - readonly throughSeq: number - readonly initial: Instructions.Values - readonly current: Instructions.Values - } - | undefined - >((state, row) => { - if (row.type === forkedEventType) { - const instructions = decodeForked(row.data).instructions - return instructions - ? { epochStart: row.seq, throughSeq: row.seq, initial: instructions, current: instructions } - : undefined - } - if (row.type === movedEventType || row.type === revertedEventType) return undefined - if (row.type === compactionEventType) - return state - ? { epochStart: row.seq, throughSeq: row.seq, initial: state.current, current: state.current } - : undefined - if (row.type !== instructionEventType) return state - const delta = decodeInstructionsUpdated(row.data).delta - const current = Instructions.applyHashDelta(state?.current ?? {}, delta) - return state - ? { ...state, throughSeq: row.seq, current } - : { epochStart: row.seq, throughSeq: row.seq, initial: current, current } - }, undefined) -} - -function foldedState(sessionID: SessionSchema.ID, folded: NonNullable>) { - return { - session_id: sessionID, - epoch_start: folded.epochStart, - through_seq: folded.throughSeq, - initial_values: folded.initial, - current_values: folded.current, - } -} diff --git a/packages/core/src/session/message-updater.ts b/packages/core/src/session/message-updater.ts index af1aa353365..95ca6298947 100644 --- a/packages/core/src/session/message-updater.ts +++ b/packages/core/src/session/message-updater.ts @@ -179,7 +179,18 @@ export function update(adapter: Adapter, event: SessionEvent.Event) { "session.execution.succeeded": () => clearCurrentRetry, "session.execution.failed": () => clearCurrentRetry, "session.execution.interrupted": () => clearCurrentRetry, - "session.instructions.updated": () => Effect.void, + "session.instructions.updated": (event) => { + if (event.data.text === undefined) return Effect.void + return adapter.appendMessage( + SessionMessage.System.make({ + id: SessionMessage.ID.fromEvent(event.id), + type: "system", + text: event.data.text, + metadata: event.metadata, + time: { created: event.created }, + }), + ) + }, "session.synthetic": (event) => { return adapter.appendMessage( SessionMessage.Synthetic.make({ diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts index 1f738ca9fb3..7c5ea023967 100644 --- a/packages/core/src/session/projector.ts +++ b/packages/core/src/session/projector.ts @@ -21,10 +21,7 @@ import { Money } from "@opencode-ai/schema/money" type DatabaseService = Database.Interface["db"] type CurrentDurableEvent = Extract -type MessageEvent = Exclude< - CurrentDurableEvent, - typeof SessionEvent.Forked.Type | typeof SessionEvent.Deleted.Type | typeof SessionEvent.InstructionsUpdated.Type -> +type MessageEvent = Exclude const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Info) const encodeMessage = Schema.encodeSync(SessionMessage.Info) @@ -638,7 +635,10 @@ const layer = Layer.effectDiscard( yield* bus.project(SessionEvent.Execution.Failed, (event) => run(db, event)) yield* bus.project(SessionEvent.Execution.Interrupted, (event) => run(db, event)) yield* bus.project(SessionEvent.InstructionsUpdated, (event) => - InstructionState.apply(db, event.data.sessionID, event.durable.seq, event.data.delta), + Effect.gen(function* () { + yield* run(db, event) + yield* InstructionState.apply(db, event.data.sessionID, event.durable.seq, event.data.delta) + }), ) yield* bus.project(SessionEvent.Synthetic, (event) => run(db, event)) yield* bus.project(SessionEvent.Skill.Activated, (event) => run(db, event)) diff --git a/packages/core/test/instruction-state.test.ts b/packages/core/test/instruction-state.test.ts index 4e75340d783..d63ea574973 100644 --- a/packages/core/test/instruction-state.test.ts +++ b/packages/core/test/instruction-state.test.ts @@ -14,7 +14,7 @@ import { AbsolutePath } from "@opencode-ai/core/schema" import { InstructionState } from "@opencode-ai/core/session/instruction-state" import { SessionProjector } from "@opencode-ai/core/session/projector" import { SessionSchema } from "@opencode-ai/core/session/schema" -import { InstructionBlobTable, InstructionStateTable, SessionTable } from "@opencode-ai/core/session/sql" +import { InstructionBlobTable, InstructionStateTable, SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql" import { testEffect } from "./lib/effect" const it = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SessionProjector.node]))) @@ -105,6 +105,7 @@ describe("InstructionState", () => { expect(observation).toEqual({ sessionID, initial: true, + previous: {}, current: { "test/first": Instructions.hash("first"), "test/second": Instructions.hash("second"), @@ -156,7 +157,7 @@ describe("InstructionState", () => { const initial = yield* InstructionState.observe(db, instructions, sessionID) expect(reads).toBe(2) - yield* InstructionState.commit(db, events, initial) + yield* InstructionState.commit(db, events, instructions, initial) expect(reads).toBe(2) current = "changed" @@ -166,6 +167,10 @@ describe("InstructionState", () => { expect(changed).toMatchObject({ sessionID, initial: false, + previous: { + "test/current": Instructions.hash("initial"), + "test/retired": Instructions.hash("retired"), + }, current: { "test/current": Instructions.hash("changed") }, delta: { "test/current": Instructions.hash("changed"), @@ -173,7 +178,7 @@ describe("InstructionState", () => { }, blobs: { [Instructions.hash("changed")]: "changed" }, }) - yield* InstructionState.commit(db, events, changed) + yield* InstructionState.commit(db, events, instructions, changed) expect(reads).toBe(4) yield* unsubscribe @@ -190,6 +195,11 @@ describe("InstructionState", () => { "test/retired": "removed", }, ]) + // The chronological update text is frozen into the event; the baseline has none. + expect((yield* instructionEvents(db, sessionID)).map((event) => event.data.text)).toEqual([ + undefined, + "changed\n\nRemoved retired", + ]) expect(yield* db.select().from(InstructionStateTable).get().pipe(Effect.orDie)).toMatchObject({ initial_values: { "test/current": Instructions.hash("initial"), @@ -222,18 +232,19 @@ describe("InstructionState", () => { expect(observation).toEqual({ sessionID, initial: false, + previous: { "test/context": Instructions.hash("unchanged") }, current: { "test/context": Instructions.hash("unchanged") }, delta: {}, blobs: {}, }) - yield* InstructionState.commit(db, events, observation) + yield* InstructionState.commit(db, events, instructions, observation) expect(yield* instructionEvents(db, sessionID)).toEqual(beforeEvents) expect(yield* db.select().from(InstructionBlobTable).all().pipe(Effect.orDie)).toEqual(beforeBlobs) }), ) - it.effect("assembles a fresh private update without repairing a missing cache", () => + it.effect("treats a missing state row as a fresh baseline without repairing it", () => Effect.gen(function* () { const sessionID = SessionSchema.ID.make("ses_instruction_generate") const { db, events } = yield* setup(sessionID) @@ -254,7 +265,7 @@ describe("InstructionState", () => { const assembled = yield* preview(db, sessionID, instructions) - expect(assembled).toEqual({ initial: "Initial context", updates: [], update: "Changed context" }) + expect(assembled).toEqual({ initial: "Changed context", update: "" }) expect(yield* instructionEvents(db, sessionID)).toEqual(beforeEvents) expect(yield* db.select().from(InstructionBlobTable).all().pipe(Effect.orDie)).toEqual(beforeBlobs) expect( @@ -268,7 +279,7 @@ describe("InstructionState", () => { }), ) - it.effect("reads through a stale cache without repairing it", () => + it.effect("trusts the projected state without consulting durable events", () => Effect.gen(function* () { const sessionID = SessionSchema.ID.make("ses_instruction_generate_stale") const { db, events } = yield* setup(sessionID) @@ -280,6 +291,7 @@ describe("InstructionState", () => { yield* InstructionState.prepare(db, events, instructions, sessionID) value = "Committed update" yield* InstructionState.prepare(db, events, instructions, sessionID) + // Tamper with the projected state; the authoritative row wins over event history. yield* db .update(InstructionStateTable) .set({ through_seq: 0, current_values: { "test/context": Instructions.hash("Initial context") } }) @@ -294,7 +306,6 @@ describe("InstructionState", () => { const assembled = yield* preview(db, sessionID, instructions) expect(assembled.initial).toBe("Initial context") - expect(assembled.updates.map((entry) => entry.message.text)).toEqual(["Committed update"]) expect(assembled.update).toBe("Private update") expect(yield* instructionEvents(db, sessionID)).toEqual(beforeEvents) expect(yield* db.select().from(InstructionBlobTable).all().pipe(Effect.orDie)).toEqual(beforeBlobs) @@ -302,6 +313,41 @@ describe("InstructionState", () => { }), ) + it.effect("persists chronological updates as system messages", () => + Effect.gen(function* () { + const sessionID = SessionSchema.ID.make("ses_instruction_messages") + const { db, events } = yield* setup(sessionID) + let value = "Initial context" + const instructions = source( + "test/context", + Effect.sync(() => value), + ) + const messages = () => + db + .select() + .from(SessionMessageTable) + .where(and(eq(SessionMessageTable.session_id, sessionID), eq(SessionMessageTable.type, "system"))) + .orderBy(asc(SessionMessageTable.seq)) + .all() + .pipe(Effect.orDie) + + // The initial baseline is not chronological history and produces no message. + yield* InstructionState.prepare(db, events, instructions, sessionID) + expect(yield* messages()).toEqual([]) + + value = "Changed context" + yield* InstructionState.prepare(db, events, instructions, sessionID) + const rows = yield* messages() + expect(rows).toHaveLength(1) + expect(rows[0]?.data).toMatchObject({ text: "Changed context" }) + expect(rows.map((row) => row.seq)).toEqual([(yield* instructionEvents(db, sessionID)).at(-1)!.seq]) + + // A no-op observation adds nothing. + yield* InstructionState.prepare(db, events, instructions, sessionID) + expect(yield* messages()).toHaveLength(1) + }), + ) + it.effect("assembles initial instructions without persisting a baseline", () => Effect.gen(function* () { const sessionID = SessionSchema.ID.make("ses_instruction_generate_initial") @@ -310,7 +356,6 @@ describe("InstructionState", () => { expect(yield* preview(db, sessionID, instructions)).toEqual({ initial: "Initial context", - updates: [], update: "", }) expect(yield* instructionEvents(db, sessionID)).toEqual([]) @@ -336,7 +381,6 @@ describe("InstructionState", () => { expect(yield* preview(db, sessionID, instructions)).toEqual({ initial: "Committed context", - updates: [], update: "", }) expect(yield* instructionEvents(db, sessionID)).toEqual(beforeEvents) @@ -388,7 +432,7 @@ describe("InstructionState", () => { for (const next of ["initial", "changed", "changed", Instructions.removed] as const) { value = next yield* InstructionState.observe(db, observedInstructions, observedSessionID).pipe( - Effect.flatMap((observation) => InstructionState.commit(db, events, observation)), + Effect.flatMap((observation) => InstructionState.commit(db, events, observedInstructions, observation)), ) yield* InstructionState.prepare(db, events, preparedInstructions, preparedSessionID) } diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index 2567aa34c1b..5ac96a6def1 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -1180,7 +1180,7 @@ describe("SessionRunnerLLM", () => { }), ) - it.effect("forks instruction values at the selected message instead of the parent's latest state", () => + it.effect("seeds a fork with the parent's newest instruction values", () => Effect.gen(function* () { const session = yield* setup yield* runPrompt(session, "First") @@ -1197,14 +1197,16 @@ describe("SessionRunnerLLM", () => { .where(eq(InstructionStateTable.session_id, forked.id)) .get(), ).toMatchObject({ - initial_values: { "test/context": Instructions.hash("Changed context") }, - current_values: { "test/context": Instructions.hash("Changed context") }, + initial_values: { "test/context": Instructions.hash("Latest context") }, + current_values: { "test/context": Instructions.hash("Latest context") }, }) yield* session.prompt({ sessionID: forked.id, text: "Forked", resume: false }) yield* session.resume(forked.id) - expect(requests.at(-1)?.system.map((part) => part.text)).toEqual([defaultSystem, "Changed context"]) - expect(systemTexts(requests.at(-1)!)).toContain("Latest context") + expect(requests.at(-1)?.system.map((part) => part.text)).toEqual([defaultSystem, "Latest context"]) + // Copied history keeps the frozen chronological update; no new update is emitted. + expect(systemTexts(requests.at(-1)!)).toContain("Changed context") + expect(systemTexts(requests.at(-1)!)).not.toContain("Latest context") const { db } = yield* Database.Service const bus = yield* Bus.Service @@ -1263,7 +1265,7 @@ describe("SessionRunnerLLM", () => { }), ) - it.effect("rebuilds a missing instruction cache without admitting another delta", () => + it.effect("re-establishes a fresh baseline when instruction state is missing", () => Effect.gen(function* () { const session = yield* setup const { db } = yield* Database.Service @@ -1277,13 +1279,15 @@ describe("SessionRunnerLLM", () => { expect(requests).toHaveLength(1) expect(requests[0]?.system.map((part) => part.text)).toEqual([defaultSystem, "Initial context"]) expect(messageRoles(requests[0])).toEqual(["user", "user"]) + // The projected row is authoritative: a missing row admits a fresh baseline + // instead of rebuilding from durable events. expect( yield* db - .select({ id: EventTable.id }) + .select({ data: EventTable.data }) .from(EventTable) .where(eq(EventTable.type, "session.instructions.updated.2")) .all(), - ).toHaveLength(1) + ).toHaveLength(2) expect(yield* db.select().from(InstructionStateTable).get()).toMatchObject({ initial_values: { "test/context": Instructions.hash("Initial context") }, current_values: { "test/context": Instructions.hash("Initial context") }, @@ -1310,7 +1314,10 @@ describe("SessionRunnerLLM", () => { ]) expect(messageRoles(requests[1])).toEqual(["user", "system", "user"]) expect(requests[1]?.messages.at(1)?.content).toEqual([{ type: "text", text: "Changed context" }]) - expect(yield* session.messages({ sessionID })).toHaveLength(2) + // The chronological update is a durable client-visible system message. + const messages = yield* session.messages({ sessionID }) + expect(messages).toHaveLength(3) + expect(messages[1]).toMatchObject({ type: "system", text: "Changed context" }) const { db } = yield* Database.Service const updates = yield* db .select({ data: EventTable.data }) @@ -1327,9 +1334,10 @@ describe("SessionRunnerLLM", () => { expect(updates[1]?.data).toEqual({ sessionID, delta: { "test/context": Instructions.hash("Changed context") }, + text: "Changed context", }) yield* replaySessionProjection(sessionID) - expect(yield* session.messages({ sessionID })).toHaveLength(2) + expect(yield* session.messages({ sessionID })).toHaveLength(3) }), ) @@ -1596,7 +1604,7 @@ describe("SessionRunnerLLM", () => { expect(requests[1]?.messages.at(1)?.content).toEqual([ { type: "text", text: "System context source removed: test/context" }, ]) - expect(yield* session.messages({ sessionID })).toHaveLength(2) + expect(yield* session.messages({ sessionID })).toHaveLength(3) }), ) @@ -1708,12 +1716,14 @@ describe("SessionRunnerLLM", () => { expect(requests[2]?.messages.filter((message) => message.role === "system")).toHaveLength(2) expect((yield* session.context(sessionID)).map((message) => message.type)).toEqual([ "user", + "system", "user", "model-switched", + "system", "user", ]) yield* replaySessionProjection(sessionID) - expect(yield* session.messages({ sessionID })).toHaveLength(4) + expect(yield* session.messages({ sessionID })).toHaveLength(6) yield* runPrompt(session, "Fourth") }), ) diff --git a/packages/schema/src/session-event.ts b/packages/schema/src/session-event.ts index f9cad0d1121..a194604e0d9 100644 --- a/packages/schema/src/session-event.ts +++ b/packages/schema/src/session-event.ts @@ -186,6 +186,11 @@ export const InstructionsUpdated = Event.durable({ schema: { ...Base, delta: Instruction.Delta, + /** + * The rendered chronological update shown to the model, frozen at emit time. + * Absent for the initial baseline observation and for deltas that render empty. + */ + text: Schema.String.pipe(optional), }, }) export type InstructionsUpdated = typeof InstructionsUpdated.Type