vrr/opencode
2
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-08-12 23:03:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

fix(core): harden v2 session context epochs

Browse source
This commit is contained in:
Kit Langton 2026-06-04 14:49:05 -04:00
parent 83916f667d
commit 03974ba0e6
33 changed files with 340 additions and 6461 deletions
Download patch file Download diff file Expand all files Collapse all files

7
packages/core/migration/20260604180746_add_session_context_epoch/migration.sql
View file

@ -1,7 +0,0 @@
CREATE TABLE `session_context_epoch` (
`session_id` text PRIMARY KEY,
`baseline` text NOT NULL,
`checkpoint` text NOT NULL,
`baseline_seq` integer NOT NULL,
CONSTRAINT `fk_session_context_epoch_session_id_session_id_fk` FOREIGN KEY (`session_id`) REFERENCES `session`(`id`) ON DELETE CASCADE
);

2026
packages/core/migration/20260604180746_add_session_context_epoch/snapshot.json generated
View file

File diff suppressed because it is too large Load diff

10
packages/core/migration/20260604181329_add_session_context_updates/migration.sql
View file

@ -1,10 +0,0 @@
CREATE TABLE `session_context_message` (
`session_id` text NOT NULL,
`seq` integer NOT NULL,
`parts` text NOT NULL,
CONSTRAINT `session_context_message_pk` PRIMARY KEY(`session_id`, `seq`),
CONSTRAINT `fk_session_context_message_session_id_session_id_fk` FOREIGN KEY (`session_id`) REFERENCES `session`(`id`) ON DELETE CASCADE
);
--> statement-breakpoint
ALTER TABLE `session_context_epoch` ADD `revision` integer DEFAULT 0 NOT NULL;--> statement-breakpoint
CREATE INDEX `session_context_message_session_seq_idx` ON `session_context_message` (`session_id`,`seq`);

1
packages/core/migration/20260604181706_add_session_context_replacement/migration.sql
View file

@ -1 +0,0 @@
ALTER TABLE `session_context_epoch` ADD `replacement_pending` integer DEFAULT false NOT NULL;

2123
packages/core/migration/20260604181706_add_session_context_replacement/snapshot.json generated
View file

File diff suppressed because it is too large Load diff

1
packages/core/migration/20260604181807_add_session_context_replacement_sequence/migration.sql
View file

@ -1 +0,0 @@
ALTER TABLE `session_context_epoch` ADD `replacement_seq` integer;

2133
packages/core/migration/20260604181807_add_session_context_replacement_sequence/snapshot.json generated
View file

File diff suppressed because it is too large Load diff

17
packages/core/migration/20260604184448_add_session_context_epoch/migration.sql Normal file
View file

@ -0,0 +1,17 @@
CREATE TABLE `session_context_epoch` (
`session_id` text PRIMARY KEY,
`baseline` text NOT NULL,
`checkpoint` text NOT NULL,
`baseline_seq` integer NOT NULL,
`replacement_seq` integer,
`revision` integer DEFAULT 0 NOT NULL,
CONSTRAINT `fk_session_context_epoch_session_id_session_id_fk` FOREIGN KEY (`session_id`) REFERENCES `session`(`id`) ON DELETE CASCADE
);
--> statement-breakpoint
CREATE TABLE `session_context_message` (
`session_id` text NOT NULL,
`seq` integer NOT NULL,
`parts` text NOT NULL,
CONSTRAINT `session_context_message_pk` PRIMARY KEY(`session_id`, `seq`),
CONSTRAINT `fk_session_context_message_session_id_session_id_fk` FOREIGN KEY (`session_id`) REFERENCES `session`(`id`) ON DELETE CASCADE
);

32
packages/core/migration/20260604181329_add_session_context_updates/snapshot.json → packages/core/migration/20260604184448_add_session_context_epoch/snapshot.json generated
View file

@ -1,9 +1,9 @@
{
"version": "7",
"dialect": "sqlite",
"id": "85667f6a-603f-4041-be9b-783a34eb74d4",
"id": "6fe30aa9-5772-49a6-b688-d022dbd28903",
"prevIds": [
"f08074bb-2b27-4365-924c-5a4eae9a5194"
"fc92fa34-8074-44c3-88f0-a5417f7fd92d"
],
"ddl": [
{
@ -822,6 +822,16 @@
"entityType": "columns",
"table": "session_context_epoch"
},
{
"type": "integer",
"notNull": false,
"autoincrement": false,
"default": null,
"generated": null,
"name": "replacement_seq",
"entityType": "columns",
"table": "session_context_epoch"
},
{
"type": "integer",
"notNull": true,
@ -1923,24 +1933,6 @@
"entityType": "indexes",
"table": "part"
},
{
"columns": [
{
"value": "session_id",
"isExpression": false
},
{
"value": "seq",
"isExpression": false
}
],
"isUnique": false,
"where": null,
"origin": "manual",
"name": "session_context_message_session_seq_idx",
"entityType": "indexes",
"table": "session_context_message"
},
{
"columns": [
{

5
packages/core/src/database/migration.gen.ts generated
View file

@ -32,9 +32,6 @@ export const migrations = (
import("./migration/20260603141458_session_input_inbox"),
import("./migration/20260603160727_jittery_ezekiel_stane"),
import("./migration/20260604172448_event_sourced_session_input"),
import("./migration/20260604180746_add_session_context_epoch"),
import("./migration/20260604181329_add_session_context_updates"),
import("./migration/20260604181706_add_session_context_replacement"),
import("./migration/20260604181807_add_session_context_replacement_sequence"),
import("./migration/20260604184448_add_session_context_epoch"),
])
).map((module) => module.default) satisfies DatabaseMigration.Migration[]

19
packages/core/src/database/migration/20260604180746_add_session_context_epoch.ts
View file

@ -1,19 +0,0 @@
import { Effect } from "effect"
import type { DatabaseMigration } from "../migration"
export default {
id: "20260604180746_add_session_context_epoch",
up(tx) {
return Effect.gen(function* () {
yield* tx.run(`
CREATE TABLE \`session_context_epoch\` (
\`session_id\` text PRIMARY KEY,
\`baseline\` text NOT NULL,
\`checkpoint\` text NOT NULL,
\`baseline_seq\` integer NOT NULL,
CONSTRAINT \`fk_session_context_epoch_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
);
`)
})
},
} satisfies DatabaseMigration.Migration

11
packages/core/src/database/migration/20260604181706_add_session_context_replacement.ts
View file

@ -1,11 +0,0 @@
import { Effect } from "effect"
import type { DatabaseMigration } from "../migration"
export default {
id: "20260604181706_add_session_context_replacement",
up(tx) {
return Effect.gen(function* () {
yield* tx.run(`ALTER TABLE \`session_context_epoch\` ADD \`replacement_pending\` integer DEFAULT false NOT NULL;`)
})
},
} satisfies DatabaseMigration.Migration

11
packages/core/src/database/migration/20260604181807_add_session_context_replacement_sequence.ts
View file

@ -1,11 +0,0 @@
import { Effect } from "effect"
import type { DatabaseMigration } from "../migration"
export default {
id: "20260604181807_add_session_context_replacement_sequence",
up(tx) {
return Effect.gen(function* () {
yield* tx.run(`ALTER TABLE \`session_context_epoch\` ADD \`replacement_seq\` integer;`)
})
},
} satisfies DatabaseMigration.Migration

17
packages/core/src/database/migration/20260604181329_add_session_context_updates.ts → packages/core/src/database/migration/20260604184448_add_session_context_epoch.ts
View file

@ -2,9 +2,20 @@ import { Effect } from "effect"
import type { DatabaseMigration } from "../migration"
export default {
id: "20260604181329_add_session_context_updates",
id: "20260604184448_add_session_context_epoch",
up(tx) {
return Effect.gen(function* () {
yield* tx.run(`
CREATE TABLE \`session_context_epoch\` (
\`session_id\` text PRIMARY KEY,
\`baseline\` text NOT NULL,
\`checkpoint\` text NOT NULL,
\`baseline_seq\` integer NOT NULL,
\`replacement_seq\` integer,
\`revision\` integer DEFAULT 0 NOT NULL,
CONSTRAINT \`fk_session_context_epoch_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
CREATE TABLE \`session_context_message\` (
\`session_id\` text NOT NULL,
@ -14,10 +25,6 @@ export default {
CONSTRAINT \`fk_session_context_message_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`ALTER TABLE \`session_context_epoch\` ADD \`revision\` integer DEFAULT 0 NOT NULL;`)
yield* tx.run(
`CREATE INDEX \`session_context_message_session_seq_idx\` ON \`session_context_message\` (\`session_id\`,\`seq\`);`,
)
})
},
} satisfies DatabaseMigration.Migration

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

@ -1,6 +1,6 @@
export * as SessionContextEpoch from "./context-epoch"
import { and, eq, sql } from "drizzle-orm"
import { and, eq, isNull, lt, sql } from "drizzle-orm"
import { DateTime, Effect, Schema } from "effect"
import type { Database } from "../database/database"
import { EventV2 } from "../event"
@ -24,28 +24,33 @@ export const prepare = Effect.fn("SessionContextEpoch.prepare")(function* (
const stored = yield* find(db, sessionID)
if (!stored) {
const initialized = SystemContext.initialize(snapshot)
yield* events.publish(SessionEvent.ContextInitialized, {
const event = yield* events.publish(SessionEvent.ContextInitialized, {
sessionID,
timestamp: yield* DateTime.now,
baseline: initialized.baseline,
checkpoint: initialized.checkpoint,
})
return initialized.baseline
if (event.seq === undefined) return yield* Effect.die("Synchronized Session event is missing aggregate sequence")
return { baseline: initialized.baseline, baselineSeq: event.seq }
}
if (stored.replacement_pending) {
if (stored.replacement_seq !== null) {
if (SystemContext.replacementBlocked(snapshot, stored.checkpoint))
return { baseline: stored.baseline, baselineSeq: stored.baseline_seq }
const initialized = SystemContext.initialize(snapshot)
yield* events.publish(SessionEvent.ContextReplaced, {
const event = yield* events.publish(SessionEvent.ContextReplaced, {
sessionID,
timestamp: yield* DateTime.now,
expectedRevision: stored.revision,
baseline: initialized.baseline,
checkpoint: initialized.checkpoint,
})
return initialized.baseline
if (event.seq === undefined) return yield* Effect.die("Synchronized Session event is missing aggregate sequence")
return { baseline: initialized.baseline, baselineSeq: event.seq }
}
const refreshed = SystemContext.refresh(snapshot, stored.checkpoint)
if (sameCheckpoint(refreshed.checkpoint, stored.checkpoint)) return stored.baseline
if (sameCheckpoint(refreshed.checkpoint, stored.checkpoint))
return { baseline: stored.baseline, baselineSeq: stored.baseline_seq }
yield* events.publish(SessionEvent.ContextUpdated, {
sessionID,
timestamp: yield* DateTime.now,
@ -53,7 +58,7 @@ export const prepare = Effect.fn("SessionContextEpoch.prepare")(function* (
parts: refreshed.changes,
checkpoint: refreshed.checkpoint,
})
return stored.baseline
return { baseline: stored.baseline, baselineSeq: stored.baseline_seq }
})
export const find = Effect.fn("SessionContextEpoch.find")(function* (db: DatabaseService, sessionID: SessionSchema.ID) {
@ -84,7 +89,6 @@ export const projectInitialized = Effect.fn("SessionContextEpoch.projectInitiali
baseline: event.data.baseline,
checkpoint: event.data.checkpoint,
baseline_seq: seq,
replacement_pending: false,
replacement_seq: null,
revision: 0,
})
@ -99,7 +103,9 @@ export const projectUpdated = Effect.fn("SessionContextEpoch.projectUpdated")(fu
) {
const stored = yield* find(db, event.data.sessionID)
if (!stored) return yield* Effect.die("Session context epoch is not initialized")
if (stored.replacement_pending) return yield* Effect.die("Session context epoch replacement is pending")
if (stored.baseline_seq > seq) return yield* Effect.void
if (stored.replacement_seq !== null && seq >= stored.replacement_seq)
return yield* Effect.die("Session context epoch replacement is pending")
if (stored.revision > event.data.expectedRevision) {
if (event.data.parts.length === 0) return yield* Effect.void
const projected = yield* db
@ -141,8 +147,9 @@ export const projectReplaced = Effect.fn("SessionContextEpoch.projectReplaced")(
) {
const stored = yield* find(db, event.data.sessionID)
if (!stored) return yield* Effect.die("Session context epoch is not initialized")
if (!stored.replacement_pending) {
if (stored.baseline_seq === seq && sameBaseline(stored.baseline, event.data.baseline)) return yield* Effect.void
if (stored.baseline_seq > seq) return yield* Effect.void
if (stored.baseline_seq === seq && sameBaseline(stored.baseline, event.data.baseline)) return yield* Effect.void
if (stored.replacement_seq === null) {
return yield* Effect.die("Session context epoch replacement was not requested")
}
const updated = yield* db
@ -151,7 +158,6 @@ export const projectReplaced = Effect.fn("SessionContextEpoch.projectReplaced")(
baseline: event.data.baseline,
checkpoint: event.data.checkpoint,
baseline_seq: seq,
replacement_pending: false,
replacement_seq: null,
revision: event.data.expectedRevision + 1,
})
@ -173,12 +179,16 @@ export const requestReplacement = Effect.fn("SessionContextEpoch.requestReplacem
sessionID: SessionSchema.ID,
seq: number,
) {
const stored = yield* find(db, sessionID)
if (!stored || stored.baseline_seq >= seq || stored.replacement_seq === seq) return yield* Effect.void
return yield* db
.update(SessionContextEpochTable)
.set({ replacement_pending: true, replacement_seq: seq, revision: sql`${SessionContextEpochTable.revision} + 1` })
.where(eq(SessionContextEpochTable.session_id, sessionID))
.set({ replacement_seq: seq, revision: sql`${SessionContextEpochTable.revision} + 1` })
.where(
and(
eq(SessionContextEpochTable.session_id, sessionID),
isNull(SessionContextEpochTable.replacement_seq),
lt(SessionContextEpochTable.baseline_seq, seq),
),
)
.run()
.pipe(Effect.orDie)
})

16
packages/core/src/session/context.ts
View file

@ -4,7 +4,7 @@ import { Database } from "../database/database"
import { MessageDecodeError } from "./error"
import { SessionMessage } from "./message"
import { SessionSchema } from "./schema"
import { SessionContextEpochTable, SessionContextMessageTable, SessionMessageTable } from "./sql"
import { SessionContextMessageTable, SessionMessageTable } from "./sql"
import type { SystemContext } from "../system-context"
type DatabaseService = Database.Interface["db"]
@ -65,24 +65,14 @@ export const load = Effect.fn("SessionContext.load")(function* (db: DatabaseServ
export const loadForRunner = Effect.fn("SessionContext.loadForRunner")(function* (
db: DatabaseService,
sessionID: SessionSchema.ID,
baselineSeq: number,
) {
const compaction = yield* latestCompaction(db, sessionID)
const messages = yield* messageRows(db, sessionID, compaction)
const epoch = yield* db
.select({ baselineSeq: SessionContextEpochTable.baseline_seq })
.from(SessionContextEpochTable)
.where(eq(SessionContextEpochTable.session_id, sessionID))
.get()
.pipe(Effect.orDie)
const updates = yield* db
.select()
.from(SessionContextMessageTable)
.where(
and(
eq(SessionContextMessageTable.session_id, sessionID),
epoch ? gt(SessionContextMessageTable.seq, epoch.baselineSeq) : undefined,
),
)
.where(and(eq(SessionContextMessageTable.session_id, sessionID), gt(SessionContextMessageTable.seq, baselineSeq)))
.orderBy(asc(SessionContextMessageTable.seq))
.all()
.pipe(Effect.orDie)

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

@ -98,8 +98,11 @@ export const layer = Layer.effect(
const getContext = Effect.fn("SessionRunner.getContext")(function* (sessionID: SessionSchema.ID) {
return yield* store.context(sessionID)
})
const getRunnerContext = Effect.fn("SessionRunner.getRunnerContext")(function* (sessionID: SessionSchema.ID) {
return yield* store.runnerContext(sessionID)
const getRunnerContext = Effect.fn("SessionRunner.getRunnerContext")(function* (
sessionID: SessionSchema.ID,
baselineSeq: number,
) {
return yield* store.runnerContext(sessionID, baselineSeq)
})
const failInterruptedTools = Effect.fn("SessionRunner.failInterruptedTools")(function* (
@ -132,9 +135,10 @@ export const layer = Layer.effect(
cause.reasons.some((reason) => Cause.isDieReason(reason) && reason.defect instanceof QuestionV2.RejectedError)
const runTurn = Effect.fn("SessionRunner.runTurn")(function* (
session: SessionSchema.Info,
sessionID: SessionSchema.ID,
promotion: "steer" | "queue" | undefined,
) {
const session = yield* getSession(sessionID)
const model = yield* models.resolve(session)
const toolFibers = yield* FiberSet.make<void, never>()
let needsContinuation = false
@ -146,12 +150,11 @@ export const layer = Layer.effect(
yield* SessionInput.promoteSteers(db, events, session.id, cutoff)
}
}
yield* failInterruptedTools(session.id)
const system = yield* SessionContextEpoch.prepare(db, events, systemContext, session.id)
const context = yield* getRunnerContext(session.id)
const context = yield* getRunnerContext(session.id, system.baselineSeq)
const request = LLM.request({
model,
system: system.map((part) => SystemPart.make(part.text)),
system: system.baseline.map((part) => SystemPart.make(part.text)),
messages: toLLMMessages(context, model),
tools: yield* tools.definitions(),
})
@ -251,12 +254,13 @@ export const layer = Layer.effect(
const hasSteer = yield* SessionInput.hasPending(db, input.sessionID, "steer")
const hasQueue = hasSteer ? false : yield* SessionInput.hasPending(db, input.sessionID, "queue")
if (input.force !== true && !hasSteer && !hasQueue) return
yield* failInterruptedTools(input.sessionID)
let promotion: "steer" | "queue" | undefined = hasSteer ? "steer" : hasQueue ? "queue" : undefined
let openActivity = input.force === true || hasSteer || hasQueue
while (openActivity) {
let needsContinuation = true
for (let step = 0; step < MAX_STEPS; step++) {
needsContinuation = yield* runTurn(session, promotion)
needsContinuation = yield* runTurn(session.id, promotion)
promotion = "steer"
if (!needsContinuation) needsContinuation = yield* SessionInput.hasPending(db, input.sessionID, "steer")
if (!needsContinuation) break

3
packages/core/src/session/runner/to-llm-message.ts
View file

@ -10,7 +10,6 @@ import {
import { SessionMessage } from "../message"
import type { FileAttachment } from "../prompt"
import { SessionContext } from "../context"
import { SystemContext } from "../../system-context"
const media = (file: FileAttachment): ContentPart => ({
type: "media",
@ -134,7 +133,7 @@ function toLLMMessage(message: SessionContext.RunnerMessage, model: Model): Mess
}),
]
case "system-context":
return [Message.system(SystemContext.render(message.parts))]
return [Message.system(message.parts.map((part) => ({ type: "text", text: part.text })))]
}
}

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

@ -171,7 +171,6 @@ export const SessionContextEpochTable = sqliteTable("session_context_epoch", {
baseline: text({ mode: "json" }).notNull().$type<ReadonlyArray<SystemContext.Part>>(),
checkpoint: text({ mode: "json" }).notNull().$type<SystemContext.Checkpoint>(),
baseline_seq: integer().notNull(),
replacement_pending: integer({ mode: "boolean" }).notNull().default(false),
replacement_seq: integer(),
revision: integer().notNull().default(0),
})
@ -186,8 +185,5 @@ export const SessionContextMessageTable = sqliteTable(
seq: integer().notNull(),
parts: text({ mode: "json" }).notNull().$type<ReadonlyArray<SystemContext.Part>>(),
},
(table) => [
primaryKey({ columns: [table.session_id, table.seq] }),
index("session_context_message_session_seq_idx").on(table.session_id, table.seq),
],
(table) => [primaryKey({ columns: [table.session_id, table.seq] })],
)

5
packages/core/src/session/store.ts
View file

@ -15,6 +15,7 @@ export interface Interface {
readonly context: (sessionID: SessionSchema.ID) => Effect.Effect<SessionMessage.Message[], MessageDecodeError>
readonly runnerContext: (
sessionID: SessionSchema.ID,
baselineSeq: number,
) => Effect.Effect<SessionContext.RunnerMessage[], MessageDecodeError>
readonly message: (
messageID: SessionMessage.ID,
@ -37,8 +38,8 @@ export const layer = Layer.effect(
context: Effect.fn("SessionStore.context")(function* (sessionID) {
return yield* SessionContext.load(db, sessionID)
}),
runnerContext: Effect.fn("SessionStore.runnerContext")(function* (sessionID) {
return yield* SessionContext.loadForRunner(db, sessionID)
runnerContext: Effect.fn("SessionStore.runnerContext")(function* (sessionID, baselineSeq) {
return yield* SessionContext.loadForRunner(db, sessionID, baselineSeq)
}),
message: Effect.fn("SessionStore.message")(function* (messageID) {
const row = yield* db

12
packages/core/src/system-context.ts
View file

@ -54,7 +54,7 @@ export const PartSchema = Schema.Struct({
text: Schema.String,
})
export const PartsSchema = Schema.Array(PartSchema)
export const CheckpointSchema = Schema.Record(Schema.String, Schema.String)
export const CheckpointSchema = Schema.Record(Key, Schema.String)
export type Checkpoint = Readonly<Record<string, string>>
@ -111,6 +111,7 @@ export function initialize(snapshot: Snapshot): Initialized {
}
export function refresh(snapshot: Snapshot, previous: Checkpoint): Refreshed {
const keys = new Set(snapshot.entries.map((entry) => entry.key))
return {
changes: [
...snapshot.entries.flatMap((entry) =>
@ -119,18 +120,15 @@ export function refresh(snapshot: Snapshot, previous: Checkpoint): Refreshed {
: [],
),
...Object.keys(previous).flatMap((key) =>
snapshot.entries.some((entry) => entry.key === key)
? []
: [{ key: Key.make(key), text: `System context component removed: ${key}` }],
keys.has(Key.make(key)) ? [] : [{ key: Key.make(key), text: `System context component removed: ${key}` }],
),
],
checkpoint: nextCheckpoint(snapshot, previous),
}
}
export function render(parts: ReadonlyArray<Part>) {
return parts.map((part) => part.text).join("\n\n")
}
export const replacementBlocked = (snapshot: Snapshot, previous: Checkpoint) =>
snapshot.entries.some((entry) => entry._tag === "Unavailable" && getCheckpoint(previous, entry.key) !== undefined)
function nextCheckpoint(snapshot: Snapshot, previous: Checkpoint) {
return Object.fromEntries(

5
packages/core/test/database-migration.test.ts
View file

@ -69,16 +69,15 @@ describe("DatabaseMigration", () => {
expect(
yield* db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'session_context_message'`),
).toEqual({ name: "session_context_message" })
expect(yield* db.get(sql`SELECT count(*) as count FROM migration`)).toEqual({ count: 34 })
expect(yield* db.get(sql`SELECT count(*) as count FROM migration`)).toEqual({ count: 31 })
expect(
yield* db.all(
sql`SELECT name FROM sqlite_master WHERE type = 'index' AND name IN ('event_aggregate_seq_idx', 'event_aggregate_type_seq_idx', 'session_context_message_session_seq_idx', 'session_input_session_pending_seq_idx', 'session_input_session_pending_delivery_seq_idx', 'session_input_session_admitted_seq_idx', 'session_input_session_promoted_seq_idx', 'session_message_session_idx', 'session_message_session_type_idx', 'session_message_session_seq_idx', 'session_message_session_type_seq_idx', 'session_message_session_time_created_id_idx') ORDER BY name`,
sql`SELECT name FROM sqlite_master WHERE type = 'index' AND name IN ('event_aggregate_seq_idx', 'event_aggregate_type_seq_idx', 'session_input_session_pending_seq_idx', 'session_input_session_pending_delivery_seq_idx', 'session_input_session_admitted_seq_idx', 'session_input_session_promoted_seq_idx', 'session_message_session_idx', 'session_message_session_type_idx', 'session_message_session_seq_idx', 'session_message_session_type_seq_idx', 'session_message_session_time_created_id_idx') ORDER BY name`,
),
).toEqual([
{ name: "event_aggregate_seq_idx" },
{ name: "event_aggregate_type_seq_idx" },
{ name: "session_input_session_admitted_seq_idx" },
{ name: "session_context_message_session_seq_idx" },
{ name: "session_input_session_pending_delivery_seq_idx" },
{ name: "session_input_session_promoted_seq_idx" },
{ name: "session_message_session_seq_idx" },

17
packages/core/test/session-runner-message.test.ts
View file

@ -90,10 +90,23 @@ describe("toLLMMessages", () => {
test("maps hidden Session context updates into chronological system messages", () => {
expect(
toLLMMessages(
[{ type: "system-context", parts: [{ key: SystemContext.Key.make("test/context"), text: "Updated context" }] }],
[
{
type: "system-context",
parts: [
{ key: SystemContext.Key.make("test/context"), text: "Updated context" },
{ key: SystemContext.Key.make("test/other"), text: "Other context" },
],
},
],
model,
),
).toEqual([Message.system("Updated context")])
).toEqual([
Message.system([
{ type: "text", text: "Updated context" },
{ type: "text", text: "Other context" },
]),
])
})
test("expands assistant tool calls and settled outcomes into canonical tool messages", () => {

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

@ -57,6 +57,7 @@ let streamStarted: Deferred.Deferred<void> | undefined
let streamFailure: LLMError | undefined
let toolExecutionGate: Deferred.Deferred<void> | undefined
let toolExecutionsStarted: Deferred.Deferred<void> | undefined
let toolExecutionsReady = 5
let activeToolExecutions = 0
let maxActiveToolExecutions = 0
const client = Layer.succeed(
@ -85,6 +86,7 @@ const client = Layer.succeed(
}),
)
const model = Model.make({ id: "fake-model", provider: "fake", route: OpenAIChat.route })
const replacementModel = Model.make({ id: "replacement", provider: "fake", route: OpenAIChat.route })
const authorizations: ToolRegistry.AuthorizeInput[] = []
const executions: string[] = []
const permission = Layer.succeed(
@ -118,7 +120,7 @@ const echo = Layer.effectDiscard(
executions.push(text)
activeToolExecutions++
maxActiveToolExecutions = Math.max(maxActiveToolExecutions, activeToolExecutions)
if (activeToolExecutions === 5 && toolExecutionsStarted) {
if (activeToolExecutions === toolExecutionsReady && toolExecutionsStarted) {
yield* Deferred.succeed(toolExecutionsStarted, undefined)
}
if (toolExecutionGate) yield* Deferred.await(toolExecutionGate)
@ -137,10 +139,13 @@ const echo = Layer.effectDiscard(
}),
),
).pipe(Layer.provide(registry))
const models = SessionRunnerModel.layerWith(() => Effect.succeed(model))
const models = SessionRunnerModel.layerWith((session) =>
Effect.succeed(session.model?.id === "replacement" ? replacementModel : model),
)
const systemContextKey = SystemContext.Key.make("test/context")
let systemBaseline = "Initial context"
let systemRemoved = false
let systemUnavailable = false
const systemContext = Layer.succeed(
SessionSystemContext.Service,
SessionSystemContext.Service.of({
@ -148,15 +153,17 @@ const systemContext = Layer.succeed(
Effect.sync(() => ({
entries: systemRemoved
? []
: [
{
_tag: "Available" as const,
key: systemContextKey,
baseline: systemBaseline,
update: systemBaseline,
hash: Hash.sha256(systemBaseline),
},
],
: systemUnavailable
? [{ _tag: "Unavailable" as const, key: systemContextKey }]
: [
{
_tag: "Available" as const,
key: systemContextKey,
baseline: systemBaseline,
update: systemBaseline,
hash: Hash.sha256(systemBaseline),
},
],
})),
}),
)
@ -229,6 +236,7 @@ const setup = Effect.gen(function* () {
response = []
systemBaseline = "Initial context"
systemRemoved = false
systemUnavailable = false
responses = undefined
streamFailure = undefined
responseStream = undefined
@ -236,6 +244,7 @@ const setup = Effect.gen(function* () {
streamStarted = undefined
toolExecutionGate = undefined
toolExecutionsStarted = undefined
toolExecutionsReady = 5
activeToolExecutions = 0
maxActiveToolExecutions = 0
yield* db
@ -656,6 +665,105 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("defers replacement while admitted context is temporarily unavailable", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
const events = yield* EventV2.Service
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "First" }), resume: false })
requests.length = 0
response = []
yield* session.resume(sessionID)
yield* events.publish(SessionEvent.ModelSwitched, {
sessionID,
timestamp: DateTime.makeUnsafe(1),
model: { id: ModelV2.ID.make("replacement"), providerID: ProviderV2.ID.make("fake") },
})
systemUnavailable = true
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Second" }), resume: false })
yield* session.resume(sessionID)
systemUnavailable = false
systemBaseline = "Replacement context"
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Third" }), resume: false })
yield* session.resume(sessionID)
expect(requests.map((request) => request.system.map((part) => part.text))).toEqual([
["Initial context"],
["Initial context"],
["Replacement context"],
])
const { db } = yield* Database.Service
expect(
yield* db
.select({ id: EventTable.id })
.from(EventTable)
.where(eq(EventTable.type, "session.next.context.replaced.1"))
.all()
.pipe(Effect.orDie),
).toHaveLength(1)
}),
)
it.effect("replays retained context projections while replacement is pending", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
const events = yield* EventV2.Service
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "First" }), resume: false })
requests.length = 0
response = []
yield* session.resume(sessionID)
systemBaseline = "Changed context"
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Second" }), resume: false })
yield* session.resume(sessionID)
yield* events.publish(SessionEvent.ModelSwitched, {
sessionID,
timestamp: DateTime.makeUnsafe(1),
model: { id: ModelV2.ID.make("replacement"), providerID: ProviderV2.ID.make("fake") },
})
yield* replaySessionProjection(sessionID)
systemBaseline = "Replacement context"
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Third" }), resume: false })
yield* session.resume(sessionID)
expect(requests.at(-1)?.system.map((part) => part.text)).toEqual(["Replacement context"])
}),
)
it.effect("replays retained context projections after multiple replacements", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
const events = yield* EventV2.Service
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "First" }), resume: false })
requests.length = 0
response = []
yield* session.resume(sessionID)
yield* events.publish(SessionEvent.ModelSwitched, {
sessionID,
timestamp: DateTime.makeUnsafe(1),
model: { id: ModelV2.ID.make("replacement-1"), providerID: ProviderV2.ID.make("fake") },
})
systemBaseline = "Replacement context 1"
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Second" }), resume: false })
yield* session.resume(sessionID)
yield* events.publish(SessionEvent.ModelSwitched, {
sessionID,
timestamp: DateTime.makeUnsafe(2),
model: { id: ModelV2.ID.make("replacement-2"), providerID: ProviderV2.ID.make("fake") },
})
systemBaseline = "Replacement context 2"
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Third" }), resume: false })
yield* session.resume(sessionID)
yield* replaySessionProjection(sessionID)
expect(requests.at(-1)?.system.map((part) => part.text)).toEqual(["Replacement context 2"])
}),
)
it.effect("replaces the baseline lazily after completed compaction without reopening replacement on replay", () =>
Effect.gen(function* () {
yield* setup
@ -863,6 +971,49 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("reloads a model switch before a tool-driven continuation turn", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
const events = yield* EventV2.Service
yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Echo this" }), resume: false })
requests.length = 0
responses = [
[
LLMEvent.stepStart({ index: 0 }),
LLMEvent.toolCall({ id: "call-echo", name: "echo", input: { text: "hello" } }),
LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }),
LLMEvent.finish({ reason: "tool-calls" }),
],
[
LLMEvent.stepStart({ index: 0 }),
LLMEvent.stepFinish({ index: 0, reason: "stop" }),
LLMEvent.finish({ reason: "stop" }),
],
]
toolExecutionGate = yield* Deferred.make<void>()
toolExecutionsStarted = yield* Deferred.make<void>()
toolExecutionsReady = 1
const run = yield* Effect.forkChild(session.resume(sessionID))
yield* Deferred.await(toolExecutionsStarted)
yield* events.publish(SessionEvent.ModelSwitched, {
sessionID,
timestamp: DateTime.makeUnsafe(1),
model: { id: ModelV2.ID.make("replacement"), providerID: ProviderV2.ID.make("fake") },
})
systemBaseline = "Replacement context"
yield* Deferred.succeed(toolExecutionGate, undefined)
yield* Fiber.join(run)
expect(requests.map((request) => request.model)).toEqual([model, replacementModel])
expect(requests.map((request) => request.system.map((part) => part.text))).toEqual([
["Initial context"],
["Replacement context"],
])
}),
)
it.effect("restores durable reasoning provider metadata in a second-turn request", () =>
Effect.gen(function* () {
yield* setup

27
packages/core/test/system-context.test.ts
View file

@ -69,7 +69,6 @@ describe("SystemContext", () => {
"core/skills": Hash.sha256("Available skills: effect"),
},
})
expect(SystemContext.render(refreshed.changes)).toBe("The current date is 2026-06-04.\n\nAvailable skills: effect")
})
test("omits unavailable initial context and admits it after its first successful load", async () => {
@ -112,6 +111,23 @@ describe("SystemContext", () => {
expect(refreshed).toEqual({ changes: [], checkpoint: previous })
})
test("blocks replacement while admitted context is unavailable", async () => {
const previous = { "core/remote-instructions": Hash.sha256("Remote instructions: old") }
const snapshot = await Effect.runPromise(
SystemContext.load(
SystemContext.struct({
remote: SystemContext.value({
key: key("core/remote-instructions"),
load: Effect.succeed(SystemContext.unavailable),
}),
}),
),
)
expect(SystemContext.replacementBlocked(snapshot, previous)).toBe(true)
expect(SystemContext.replacementBlocked(snapshot, {})).toBe(false)
})
test("emits tombstones and drops checkpoints for removed components", async () => {
const context = SystemContext.struct({
date: SystemContext.value({
@ -185,4 +201,13 @@ describe("SystemContext", () => {
expect(() => decode("date")).toThrow()
expect(() => decode("core/")).toThrow()
})
test("requires namespaced checkpoint keys", () => {
const decode = Schema.decodeUnknownSync(SystemContext.CheckpointSchema)
const valid = JSON.parse('{"core/date":"hash"}')
const invalid = JSON.parse('{"date":"hash"}')
expect(decode(valid)).toEqual(valid)
expect(() => decode(invalid)).toThrow()
})
})

2
packages/llm/src/protocols/anthropic-messages.ts
View file

@ -386,7 +386,7 @@ const lowerMessages = Effect.fn("AnthropicMessages.lowerMessages")(function* (
for (const [index, message] of request.messages.entries()) {
if (message.role === "system") {
yield* ProviderShared.guardSystemUpdatePlacement("Anthropic Messages", request.messages[index - 1])
yield* ProviderShared.guardSystemUpdatePlacement("Anthropic Messages", request.messages, index)
if (supportsNativeSystemUpdates(request) && canUseNativeSystemUpdate(request.messages, index)) {
messages.push(yield* lowerNativeSystemUpdate(message, breakpoints))
continue

2
packages/llm/src/protocols/bedrock-converse.ts
View file

@ -294,7 +294,7 @@ const lowerMessages = Effect.fn("BedrockConverse.lowerMessages")(function* (
for (const [index, message] of request.messages.entries()) {
if (message.role === "system") {
yield* ProviderShared.guardSystemUpdatePlacement("Bedrock Converse", request.messages[index - 1])
yield* ProviderShared.guardSystemUpdatePlacement("Bedrock Converse", request.messages, index)
const part = yield* ProviderShared.wrappedSystemUpdate("Bedrock Converse", message)
const content = textWithCache(breakpoints, part.text, part.cache)
const previous = messages.at(-1)

2
packages/llm/src/protocols/gemini.ts
View file

@ -202,7 +202,7 @@ const lowerMessages = Effect.fn("Gemini.lowerMessages")(function* (request: LLMR
for (const [index, message] of request.messages.entries()) {
if (message.role === "system") {
yield* ProviderShared.guardSystemUpdatePlacement("Gemini", request.messages[index - 1])
yield* ProviderShared.guardSystemUpdatePlacement("Gemini", request.messages, index)
const part = yield* ProviderShared.wrappedSystemUpdate("Gemini", message)
const previous = contents.at(-1)
if (previous?.role === "user")

2
packages/llm/src/protocols/openai-chat.ts
View file

@ -254,7 +254,7 @@ const lowerMessages = Effect.fn("OpenAIChat.lowerMessages")(function* (request:
const messages = [...system]
for (const [index, message] of request.messages.entries()) {
if (message.role === "system") {
yield* ProviderShared.guardSystemUpdatePlacement("OpenAI Chat", request.messages[index - 1])
yield* ProviderShared.guardSystemUpdatePlacement("OpenAI Chat", request.messages, index)
const part = yield* ProviderShared.wrappedSystemUpdate("OpenAI Chat", message)
const previous = messages.at(-1)
if (previous?.role === "user")

2
packages/llm/src/protocols/openai-responses.ts
View file

@ -340,7 +340,7 @@ const lowerMessages = Effect.fn("OpenAIResponses.lowerMessages")(function* (requ
for (const [index, message] of request.messages.entries()) {
if (message.role === "system") {
yield* ProviderShared.guardSystemUpdatePlacement("OpenAI Responses", request.messages[index - 1])
yield* ProviderShared.guardSystemUpdatePlacement("OpenAI Responses", request.messages, index)
const part = yield* ProviderShared.wrappedSystemUpdate("OpenAI Responses", message)
const previous = input.at(-1)
if (previous && "role" in previous && previous.role === "user")

16
packages/llm/src/protocols/shared.ts
View file

@ -179,12 +179,18 @@ export const wrappedSystemUpdate = Effect.fn("ProviderShared.wrappedSystemUpdate
export const guardSystemUpdatePlacement = Effect.fn("ProviderShared.guardSystemUpdatePlacement")(function* (
route: string,
previous: LLMRequest["messages"][number] | undefined,
messages: LLMRequest["messages"],
index: number,
) {
if (
previous?.role === "assistant" &&
previous.content.some((part) => part.type === "tool-call" && part.providerExecuted !== true)
)
const pending = new Set<string>()
for (const message of messages.slice(0, index)) {
for (const part of message.content) {
if (message.role === "assistant" && part.type === "tool-call" && part.providerExecuted !== true)
pending.add(part.id)
if (message.role === "tool" && part.type === "tool-result") pending.delete(part.id)
}
}
if (pending.size > 0)
return yield* invalidRequest(
`${route} chronological system updates cannot appear between a local tool call and its tool result`,
)

26
packages/llm/test/schema.test.ts
View file

@ -65,12 +65,28 @@ describe("llm schema", () => {
})
test("rejects chronological system updates between a local tool call and its result", async () => {
const previous = Message.assistant([
ToolCallPart.make({ id: "call_1", name: "lookup", input: {} }),
{ type: "text", text: "Waiting." },
])
const messages = [
Message.assistant([
ToolCallPart.make({ id: "call_1", name: "lookup", input: {} }),
{ type: "text", text: "Waiting." },
]),
]
await expect(Effect.runPromise(ProviderShared.guardSystemUpdatePlacement("Test", previous))).rejects.toThrow(
await expect(Effect.runPromise(ProviderShared.guardSystemUpdatePlacement("Test", messages, 1))).rejects.toThrow(
"Test chronological system updates cannot appear between a local tool call and its tool result",
)
})
test("rejects chronological system updates between results for multiple local tool calls", async () => {
const messages = [
Message.assistant([
ToolCallPart.make({ id: "call_1", name: "lookup", input: {} }),
ToolCallPart.make({ id: "call_2", name: "lookup", input: {} }),
]),
Message.tool({ id: "call_1", name: "lookup", result: "first" }),
]
await expect(Effect.runPromise(ProviderShared.guardSystemUpdatePlacement("Test", messages, 2))).rejects.toThrow(
"Test chronological system updates cannot appear between a local tool call and its tool result",
)
})

10
specs/v2/schema-changelog.md
View file

@ -703,7 +703,7 @@ Change:
Compatibility:
- This adds one database migration and one synchronized Session event type.
- The unpublished Context Epoch schema is consolidated into one database migration and this adds one synchronized Session event type.
- Existing experimental V2 Session databases remain disposable across incompatible pre-launch event-schema changes.
- Chronological context updates, replacement epochs after compaction or model switches, project instructions, skills guidance, and plugin transforms remain follow-up slices.
@ -719,13 +719,13 @@ Change:
- Refresh Location-scoped Context Components at each safe provider-turn boundary.
- Keep the stored baseline immutable while admitting changed component values as runner-private chronological `Message.system(...)` history.
- Advance component-hash checkpoints transactionally even when a removed component produces no visible update text.
- Emit an explicit tombstone update and advance component-hash checkpoints transactionally when a component is removed.
- Keep ordinary Session transcript APIs unchanged while runner history merges visible Session messages and hidden context updates by durable aggregate sequence.
- Reject chronological system updates that would split a local tool call from its result across provider protocols; use wrapped user fallback when Anthropic native system-update placement is unsupported.
Compatibility:
- This adds one database migration and one synchronized Session event type.
- The unpublished Context Epoch schema remains consolidated into one database migration and this adds one synchronized Session event type.
- Existing experimental V2 Session databases remain disposable across incompatible pre-launch event-schema changes.
- Replacement epochs after compaction or model switches, project instructions, skills guidance, and plugin transforms remain follow-up slices.
@ -734,7 +734,7 @@ Compatibility:
Affected schema:
- Add synchronized `session.next.context.replaced.1` Session events.
- Add `session_context_epoch.replacement_pending` and `session_context_epoch.replacement_seq` for idempotent lazy replacement requests.
- Add nullable `session_context_epoch.replacement_seq` for idempotent lazy replacement requests.
Change:
@ -745,6 +745,6 @@ Change:
Compatibility:
- This adds two additive database migrations and one synchronized Session event type.
- The unpublished Context Epoch schema remains consolidated into one database migration and this adds one synchronized Session event type.
- Existing experimental V2 Session databases remain disposable across incompatible pre-launch event-schema changes.
- Compaction execution, project instructions, skills guidance, and plugin transforms remain follow-up slices.