From cd812e2045e1825ca92596e8d7f5a2ddf34510a1 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Thu, 4 Jun 2026 21:17:05 -0400 Subject: [PATCH] fix(core): finalize v2 session context epochs --- CONTEXT.md | 11 +- .../migration.sql | 0 .../snapshot.json | 260 +++---- packages/core/src/database/migration.gen.ts | 2 +- ...605003541_add_session_context_snapshot.ts} | 2 +- packages/core/src/event.ts | 37 +- packages/core/src/instruction-context.ts | 27 +- packages/core/src/session/context-epoch.ts | 127 ++- packages/core/src/session/context.ts | 10 +- packages/core/src/session/message.ts | 11 +- packages/core/src/session/projector.ts | 4 +- packages/core/src/session/runner/index.ts | 8 +- packages/core/src/session/runner/llm.ts | 3 +- packages/core/src/system-context-builtins.ts | 2 +- packages/core/src/system-context-registry.ts | 6 +- packages/core/src/system-context.ts | 23 +- packages/core/test/database-migration.test.ts | 159 ++-- packages/core/test/event.test.ts | 7 +- .../core/test/instruction-context.test.ts | 14 +- .../core/test/session-runner-message.test.ts | 727 +++++++++--------- .../core/test/session-runner-recorded.test.ts | 5 +- packages/core/test/session-runner.test.ts | 109 ++- .../core/test/system-context-registry.test.ts | 6 +- packages/core/test/system-context.test.ts | 29 +- ...haiku-4-5-chronological-system-update.json | 8 +- ...2-5-flash-chronological-system-update.json | 8 +- ...t-4o-mini-chronological-system-update.json | 8 +- .../src/cli/cmd/tui/context/sync-v2.tsx | 2 +- .../opencode/test/cli/tui/sync-v2.test.tsx | 97 +++ packages/sdk/js/src/v2/gen/types.gen.ts | 53 ++ packages/sdk/openapi.json | 169 ++++ specs/v2/refreshable-context-sources.md | 94 ++- specs/v2/schema-changelog.md | 24 +- 33 files changed, 1253 insertions(+), 799 deletions(-) rename packages/core/migration/{20260604234609_add_session_context_snapshot => 20260605003541_add_session_context_snapshot}/migration.sql (100%) rename packages/core/migration/{20260604234609_add_session_context_snapshot => 20260605003541_add_session_context_snapshot}/snapshot.json (94%) rename packages/core/src/database/migration/{20260604234609_add_session_context_snapshot.ts => 20260605003541_add_session_context_snapshot.ts} (92%) diff --git a/CONTEXT.md b/CONTEXT.md index 562cd43dcf6..d5ccd6de6da 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -46,22 +46,23 @@ The point immediately before a provider call, after durable input promotion and - Changes from multiple **Context Sources** admitted at one safe boundary combine into one **Mid-Conversation System Message**. - Context changes are sampled and admitted lazily at a **Safe Provider-Turn Boundary**, never pushed asynchronously when their source changes. - At a **Safe Provider-Turn Boundary**, newly promoted user input or settled tool results precede any combined **Mid-Conversation System Message**. -- The first provider turn renders the latest **Baseline System Context** and initializes its **Context Snapshot** without emitting a redundant **Mid-Conversation System Message**. +- The first provider turn renders the latest complete **Baseline System Context** and initializes its **Context Snapshot** without emitting a redundant **Mid-Conversation System Message**; unavailable initial context blocks the turn instead of persisting an incomplete baseline. +- Initial **System Context** preparation precedes the first durable input promotion so an unavailable baseline leaves that input pending and retryable; ordinary reconciliation remains after promotion. - Compaction starts a new **Context Epoch** with a freshly rendered **Baseline System Context** and **Context Snapshot**; prior **Mid-Conversation System Messages** remain durable audit history but leave projected model history. - A newly registered core or plugin-defined **Context Source** absent from the current snapshot emits its baseline rendering once at the next **Safe Provider-Turn Boundary**. - **Context Source** keys are stable and namespaced; duplicate keys fail composition. `SystemContext.combine(...)` preserves caller order; the **System Context Registry** evaluates producers concurrently and combines them in stable contribution-key order so rendered context remains deterministic. - Each **Context Source** loader returns one coherent typed value. `SystemContext.make(...)` hides that value type so differently typed sources compose uniformly. Its codec compares and stores that value; its pure renderers produce model-visible baseline, update, and removal text only when needed. - `SystemContext.initialize(...)` observes a composed **System Context** once and produces a fresh **Baseline System Context** with its **Context Snapshot**. -- `SystemContext.reconcile(...)` observes a composed **System Context** once and returns exactly one next action: unchanged, updated, replaced, or replacement blocked. +- `SystemContext.reconcile(...)` observes a composed **System Context** once and returns exactly one next action: unchanged, updated, replacement ready, or replacement blocked. - `SystemContext.replace(...)` represents an explicit baseline-replacing transition such as compaction or model/provider switch; it either produces a fresh generation or reports that replacement is blocked by unavailable admitted context. +- Context Epoch preparation retries until stable after optimistic revision mismatches so concurrent replacement requests cannot terminate an otherwise valid safe-boundary run. - **Unavailable Context** uses stale-while-revalidate semantics and is distinct from a successfully loaded absence, which may emit removal text. - Ordinary **Context Source** loaders return values directly; loaders that intentionally use stale-while-revalidate may explicitly return **Unavailable Context**. -- Nested project instruction files discovered while reading join the effective instructions returned by the instruction service and are admitted durably at the next **Safe Provider-Turn Boundary**. -- A discovered nested project instruction remains active for the session while it stays in the same location and is folded into later **Baseline System Contexts** after compaction. +- Nested project instruction discovery after successful reads remains a follow-up; when implemented, discovered instructions must be admitted durably at the next **Safe Provider-Turn Boundary**. - Location-scoped services naturally re-resolve effective context when a moved session next runs in its destination location. - Instruction discovery, source identity, persistence, and file loading belong to the instruction service; the **System Context** abstraction only composes effectful producers and renders loaded values. - The first instruction-service slice observes global and upward project `AGENTS.md` files as one ordered aggregate **Context Source** at each **Safe Provider-Turn Boundary**. -- Built-in, instruction, and plugin-defined context producers register through the **System Context Registry** with stable contribution keys so plugin hot reload and Location-scope cleanup add and remove sources predictably. +- Built-in and instruction context producers register through the **System Context Registry** with stable contribution keys. Plugin-defined context registration and hot-reload lifecycle remain a follow-up built on the same scoped registry seam. - Context source changes never wake idle sessions; the next naturally scheduled **Safe Provider-Turn Boundary** loads and compares current values lazily. - Once admitted, a **Mid-Conversation System Message** remains durable even if the following provider attempt fails and is replayed unchanged on retry. - **Mid-Conversation System Messages** remain durable Session-message history; normal user-facing transcript surfaces may hide them. diff --git a/packages/core/migration/20260604234609_add_session_context_snapshot/migration.sql b/packages/core/migration/20260605003541_add_session_context_snapshot/migration.sql similarity index 100% rename from packages/core/migration/20260604234609_add_session_context_snapshot/migration.sql rename to packages/core/migration/20260605003541_add_session_context_snapshot/migration.sql diff --git a/packages/core/migration/20260604234609_add_session_context_snapshot/snapshot.json b/packages/core/migration/20260605003541_add_session_context_snapshot/snapshot.json similarity index 94% rename from packages/core/migration/20260604234609_add_session_context_snapshot/snapshot.json rename to packages/core/migration/20260605003541_add_session_context_snapshot/snapshot.json index 22c68919c59..6e1cce13613 100644 --- a/packages/core/migration/20260604234609_add_session_context_snapshot/snapshot.json +++ b/packages/core/migration/20260605003541_add_session_context_snapshot/snapshot.json @@ -1,10 +1,8 @@ { "version": "7", "dialect": "sqlite", - "id": "65b52d0f-2bbe-483f-a7ec-9d2a7fa29f57", - "prevIds": [ - "fc92fa34-8074-44c3-88f0-a5417f7fd92d" - ], + "id": "40f7b9b8-83b4-4ea0-a59f-76a489679d88", + "prevIds": ["84c6ad6c-6116-48e1-b973-6fee4593496b"], "ddl": [ { "name": "workspace", @@ -838,19 +836,9 @@ "entityType": "columns", "table": "session_context_epoch" }, - { - "type": "integer", - "notNull": false, - "autoincrement": true, - "default": null, - "generated": null, - "name": "seq", - "entityType": "columns", - "table": "session_input" - }, { "type": "text", - "notNull": true, + "notNull": false, "autoincrement": false, "default": null, "generated": null, @@ -888,6 +876,16 @@ "entityType": "columns", "table": "session_input" }, + { + "type": "integer", + "notNull": true, + "autoincrement": false, + "default": null, + "generated": null, + "name": "admitted_seq", + "entityType": "columns", + "table": "session_input" + }, { "type": "integer", "notNull": false, @@ -1399,13 +1397,9 @@ "table": "session_share" }, { - "columns": [ - "project_id" - ], + "columns": ["project_id"], "tableTo": "project", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1414,13 +1408,9 @@ "table": "workspace" }, { - "columns": [ - "active_account_id" - ], + "columns": ["active_account_id"], "tableTo": "account", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "SET NULL", "nameExplicit": false, @@ -1429,13 +1419,9 @@ "table": "account_state" }, { - "columns": [ - "aggregate_id" - ], + "columns": ["aggregate_id"], "tableTo": "event_sequence", - "columnsTo": [ - "aggregate_id" - ], + "columnsTo": ["aggregate_id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1444,13 +1430,9 @@ "table": "event" }, { - "columns": [ - "project_id" - ], + "columns": ["project_id"], "tableTo": "project", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1459,13 +1441,9 @@ "table": "permission" }, { - "columns": [ - "project_id" - ], + "columns": ["project_id"], "tableTo": "project", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1474,13 +1452,9 @@ "table": "project_directory" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1489,13 +1463,9 @@ "table": "message" }, { - "columns": [ - "message_id" - ], + "columns": ["message_id"], "tableTo": "message", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1504,13 +1474,9 @@ "table": "part" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1519,13 +1485,9 @@ "table": "session_context_epoch" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1534,13 +1496,9 @@ "table": "session_input" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1549,13 +1507,9 @@ "table": "session_message" }, { - "columns": [ - "project_id" - ], + "columns": ["project_id"], "tableTo": "project", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1564,13 +1518,9 @@ "table": "session" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1579,13 +1529,9 @@ "table": "todo" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "tableTo": "session", - "columnsTo": [ - "id" - ], + "columnsTo": ["id"], "onUpdate": "NO ACTION", "onDelete": "CASCADE", "nameExplicit": false, @@ -1594,165 +1540,126 @@ "table": "session_share" }, { - "columns": [ - "email", - "url" - ], + "columns": ["email", "url"], "nameExplicit": false, "name": "control_account_pk", "entityType": "pks", "table": "control_account" }, { - "columns": [ - "project_id", - "directory" - ], + "columns": ["project_id", "directory"], "nameExplicit": false, "name": "project_directory_pk", "entityType": "pks", "table": "project_directory" }, { - "columns": [ - "session_id", - "position" - ], + "columns": ["session_id", "position"], "nameExplicit": false, "name": "todo_pk", "entityType": "pks", "table": "todo" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "workspace_pk", "table": "workspace", "entityType": "pks" }, { - "columns": [ - "name" - ], + "columns": ["name"], "nameExplicit": false, "name": "data_migration_pk", "table": "data_migration", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "account_state_pk", "table": "account_state", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "account_pk", "table": "account", "entityType": "pks" }, { - "columns": [ - "aggregate_id" - ], + "columns": ["aggregate_id"], "nameExplicit": false, "name": "event_sequence_pk", "table": "event_sequence", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "event_pk", "table": "event", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "permission_pk", "table": "permission", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "project_pk", "table": "project", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "message_pk", "table": "message", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "part_pk", "table": "part", "entityType": "pks" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "nameExplicit": false, "name": "session_context_epoch_pk", "table": "session_context_epoch", "entityType": "pks" }, { - "columns": [ - "seq" - ], + "columns": ["id"], "nameExplicit": false, "name": "session_input_pk", "table": "session_input", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "session_message_pk", "table": "session_message", "entityType": "pks" }, { - "columns": [ - "id" - ], + "columns": ["id"], "nameExplicit": false, "name": "session_pk", "table": "session", "entityType": "pks" }, { - "columns": [ - "session_id" - ], + "columns": ["session_id"], "nameExplicit": false, "name": "session_share_pk", "table": "session_share", @@ -1769,7 +1676,7 @@ "isExpression": false } ], - "isUnique": false, + "isUnique": true, "where": null, "origin": "manual", "name": "event_aggregate_seq_idx", @@ -1889,7 +1796,7 @@ "isExpression": false }, { - "value": "seq", + "value": "admitted_seq", "isExpression": false } ], @@ -1900,6 +1807,42 @@ "entityType": "indexes", "table": "session_input" }, + { + "columns": [ + { + "value": "session_id", + "isExpression": false + }, + { + "value": "admitted_seq", + "isExpression": false + } + ], + "isUnique": true, + "where": null, + "origin": "manual", + "name": "session_input_session_admitted_seq_idx", + "entityType": "indexes", + "table": "session_input" + }, + { + "columns": [ + { + "value": "session_id", + "isExpression": false + }, + { + "value": "promoted_seq", + "isExpression": false + } + ], + "isUnique": true, + "where": null, + "origin": "manual", + "name": "session_input_session_promoted_seq_idx", + "entityType": "indexes", + "table": "session_input" + }, { "columns": [ { @@ -1911,7 +1854,7 @@ "isExpression": false } ], - "isUnique": false, + "isUnique": true, "where": null, "origin": "manual", "name": "session_message_session_seq_idx", @@ -2031,16 +1974,7 @@ "name": "todo_session_idx", "entityType": "indexes", "table": "todo" - }, - { - "columns": [ - "id" - ], - "nameExplicit": false, - "name": "session_input_id_unique", - "entityType": "uniques", - "table": "session_input" } ], "renames": [] -} \ No newline at end of file +} diff --git a/packages/core/src/database/migration.gen.ts b/packages/core/src/database/migration.gen.ts index 949521287d3..299ff846a5f 100644 --- a/packages/core/src/database/migration.gen.ts +++ b/packages/core/src/database/migration.gen.ts @@ -32,6 +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/20260604234609_add_session_context_snapshot"), + import("./migration/20260605003541_add_session_context_snapshot"), ]) ).map((module) => module.default) satisfies DatabaseMigration.Migration[] diff --git a/packages/core/src/database/migration/20260604234609_add_session_context_snapshot.ts b/packages/core/src/database/migration/20260605003541_add_session_context_snapshot.ts similarity index 92% rename from packages/core/src/database/migration/20260604234609_add_session_context_snapshot.ts rename to packages/core/src/database/migration/20260605003541_add_session_context_snapshot.ts index f035e53cbd4..d1648969dda 100644 --- a/packages/core/src/database/migration/20260604234609_add_session_context_snapshot.ts +++ b/packages/core/src/database/migration/20260605003541_add_session_context_snapshot.ts @@ -2,7 +2,7 @@ import { Effect } from "effect" import type { DatabaseMigration } from "../migration" export default { - id: "20260604234609_add_session_context_snapshot", + id: "20260605003541_add_session_context_snapshot", up(tx) { return Effect.gen(function* () { yield* tx.run(` diff --git a/packages/core/src/event.ts b/packages/core/src/event.ts index 8ae993967fe..1362bf1e417 100644 --- a/packages/core/src/event.ts +++ b/packages/core/src/event.ts @@ -155,7 +155,6 @@ export interface Interface { readonly aggregateID: string readonly after?: Cursor }) => Stream.Stream - readonly sequence: (aggregateID: string) => Effect.Effect readonly sync: (handler: Sync) => Effect.Effect readonly listen: (listener: Listener) => Effect.Effect readonly beforeCommit: (guard: CommitGuard) => Effect.Effect @@ -337,9 +336,6 @@ export const layerWith = (options?: LayerOptions) => yield* projector({ ...event, seq } as Payload) } if (commit) yield* commit(seq) - const encoded = syncRegistry - .get(versionedType(definition.type, sync.version))! - .encode(event.data) yield* db .insert(EventSequenceTable) .values([{ aggregate_id: aggregateID, seq, owner_id: input?.ownerID }]) @@ -390,7 +386,10 @@ export const layerWith = (options?: LayerOptions) => const durable = registry.get(event.type)?.sync !== undefined if (!durable && options?.commit) return yield* Effect.die( - new InvalidSyncEventError({ type: event.type, message: "Local commit hooks require a synchronized event" }), + new InvalidSyncEventError({ + type: event.type, + message: "Local commit hooks require a synchronized event", + }), ) if (durable) { const committed = yield* commitSyncEvent(event as Payload, undefined, options?.commit) @@ -438,14 +437,17 @@ export const layerWith = (options?: LayerOptions) => (serviceLocation ? { directory: serviceLocation.directory, workspaceID: serviceLocation.workspaceID } : undefined) - return yield* publishEvent({ - id: options?.id ?? ID.create(), - ...(options?.metadata ? { metadata: options.metadata } : {}), - type: definition.type, - ...(definition.sync === undefined ? {} : { version: definition.sync.version }), - ...(location ? { location } : {}), - data, - } as Payload, options) + return yield* publishEvent( + { + id: options?.id ?? ID.create(), + ...(options?.metadata ? { metadata: options.metadata } : {}), + type: definition.type, + ...(definition.sync === undefined ? {} : { version: definition.sync.version }), + ...(location ? { location } : {}), + data, + } as Payload, + options, + ) }) } @@ -534,14 +536,6 @@ export const layerWith = (options?: LayerOptions) => .pipe(Effect.orDie) } - const sequence = (aggregateID: string) => - db - .select({ seq: EventSequenceTable.seq }) - .from(EventSequenceTable) - .where(eq(EventSequenceTable.aggregate_id, aggregateID)) - .get() - .pipe(Effect.orDie, Effect.map((row) => row?.seq ?? -1)) - const subscribe = (definition: D): Stream.Stream> => Stream.unwrap(getOrCreate(definition).pipe(Effect.map((pubsub) => Stream.fromPubSub(pubsub)))).pipe( Stream.map((event) => event as Payload), @@ -669,7 +663,6 @@ export const layerWith = (options?: LayerOptions) => subscribe, all: streamAll, aggregateEvents: streamEvents, - sequence, sync, listen, beforeCommit, diff --git a/packages/core/src/instruction-context.ts b/packages/core/src/instruction-context.ts index c764c130673..be258e2a1ca 100644 --- a/packages/core/src/instruction-context.ts +++ b/packages/core/src/instruction-context.ts @@ -15,6 +15,7 @@ class File extends Schema.Class("InstructionContext.File")({ }) {} const Files = Schema.Array(File) +const key = SystemContext.Key.make("core/instructions") export const layer = Layer.effectDiscard( Effect.gen(function* () { @@ -25,7 +26,7 @@ export const layer = Layer.effectDiscard( const source = (value: ReadonlyArray | SystemContext.Unavailable) => SystemContext.make({ - key: SystemContext.Key.make("core/instructions"), + key, codec: Schema.toCodecJson(Files), load: Effect.succeed(value), baseline: render, @@ -43,29 +44,37 @@ export const layer = Layer.effectDiscard( const files = yield* Effect.forEach( paths, (path) => - fs.readFileStringSafe(path).pipe( - Effect.map((content) => (content === undefined ? undefined : new File({ path: AbsolutePath.make(path), content }))), - ), + fs + .readFileStringSafe(path) + .pipe( + Effect.map((content) => + content === undefined ? undefined : new File({ path: AbsolutePath.make(path), content }), + ), + ), { concurrency: "unbounded" }, ) - if (files.some((file, index) => file === undefined && discovered.has(paths[index]))) return SystemContext.unavailable + if (files.some((file, index) => file === undefined && discovered.has(paths[index]))) + return SystemContext.unavailable return files.filter((file): file is File => file !== undefined) }) yield* registry.contribute({ - key: "core/instructions", + key, load: observe().pipe( Effect.map((files) => - files === SystemContext.unavailable ? source(files) : files.length === 0 ? SystemContext.empty : source(files), + files === SystemContext.unavailable + ? source(files) + : files.length === 0 + ? SystemContext.empty + : source(files), ), Effect.catch(() => Effect.succeed(source(SystemContext.unavailable))), + Effect.catchDefect(() => Effect.succeed(source(SystemContext.unavailable))), ), }) }), ) -export const locationLayer = layer - function render(files: ReadonlyArray) { return files.map((file) => `Instructions from: ${file.path}\n${file.content}`).join("\n\n") } diff --git a/packages/core/src/session/context-epoch.ts b/packages/core/src/session/context-epoch.ts index 7374b58efff..06465771efb 100644 --- a/packages/core/src/session/context-epoch.ts +++ b/packages/core/src/session/context-epoch.ts @@ -7,13 +7,51 @@ import { EventV2 } from "../event" import { SystemContext } from "../system-context" import { SystemContextRegistry } from "../system-context-registry" import { SessionEvent } from "./event" +import { SessionInput } from "./input" import { SessionMessageID } from "./message-id" import { SessionSchema } from "./schema" import { SessionContextEpochTable } from "./sql" type DatabaseService = Database.Interface["db"] -export const prepare = Effect.fn("SessionContextEpoch.prepare")(function* ( +class RevisionMismatch extends Error {} + +const retryRevisionMismatch = (attempt: () => Effect.Effect): Effect.Effect => + attempt().pipe( + Effect.catchDefect((defect) => + defect instanceof RevisionMismatch + ? Effect.yieldNow.pipe(Effect.andThen(retryRevisionMismatch(attempt))) + : Effect.die(defect), + ), + ) + +interface Prepared { + readonly baseline: string + readonly baselineSeq: number +} + +export function initialize( + db: DatabaseService, + context: SystemContextRegistry.Interface, + sessionID: SessionSchema.ID, +): Effect.Effect { + return retryRevisionMismatch(() => initializeOnce(db, context, sessionID)).pipe( + Effect.withSpan("SessionContextEpoch.initialize"), + ) +} + +export function prepare( + db: DatabaseService, + events: EventV2.Interface, + context: SystemContextRegistry.Interface, + sessionID: SessionSchema.ID, +): Effect.Effect { + return retryRevisionMismatch(() => prepareOnce(db, events, context, sessionID)).pipe( + Effect.withSpan("SessionContextEpoch.prepare"), + ) +} + +const prepareOnce = Effect.fnUntraced(function* ( db: DatabaseService, events: EventV2.Interface, context: SystemContextRegistry.Interface, @@ -22,17 +60,19 @@ export const prepare = Effect.fn("SessionContextEpoch.prepare")(function* ( const [value, stored] = yield* Effect.all([context.load(), find(db, sessionID)], { concurrency: "unbounded" }) if (!stored) { const generation = yield* SystemContext.initialize(value) - const baselineSeq = yield* initialize(db, events, sessionID, generation) + const baselineSeq = yield* insert(db, sessionID, generation) return { baseline: generation.baseline, baselineSeq } } const snapshot = yield* Schema.decodeUnknownEffect(SystemContext.Snapshot)(stored.snapshot).pipe(Effect.orDie) const result = - stored.replacement_seq === null ? yield* SystemContext.reconcile(value, snapshot) : yield* SystemContext.replace(value, snapshot) + stored.replacement_seq === null + ? yield* SystemContext.reconcile(value, snapshot) + : yield* SystemContext.replace(value, snapshot) if (result._tag === "Unchanged" || result._tag === "ReplacementBlocked") return { baseline: stored.baseline, baselineSeq: stored.baseline_seq } - if (result._tag === "Replaced") { - const replacementSeq = stored.replacement_seq ?? (yield* events.sequence(sessionID)) + if (result._tag === "ReplacementReady") { + const replacementSeq = stored.replacement_seq ?? (yield* SessionInput.latestSeq(db, sessionID)) yield* replace(db, sessionID, stored.revision, replacementSeq, result.generation) return { baseline: result.generation.baseline, baselineSeq: replacementSeq } } @@ -45,6 +85,28 @@ export const prepare = Effect.fn("SessionContextEpoch.prepare")(function* ( return { baseline: stored.baseline, baselineSeq: stored.baseline_seq } }) +const initializeOnce = Effect.fnUntraced(function* ( + db: DatabaseService, + context: SystemContextRegistry.Interface, + sessionID: SessionSchema.ID, +) { + if (yield* exists(db, sessionID)) return + const generation = yield* context.load().pipe(Effect.flatMap(SystemContext.initialize)) + const baselineSeq = yield* insert(db, sessionID, generation) + return { baseline: generation.baseline, 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() @@ -73,9 +135,8 @@ export const requestReplacement = Effect.fn("SessionContextEpoch.requestReplacem .pipe(Effect.orDie) }) -const initialize = Effect.fnUntraced(function* ( +const insert = Effect.fnUntraced(function* ( db: DatabaseService, - events: EventV2.Interface, sessionID: SessionSchema.ID, generation: SystemContext.Generation, ) { @@ -83,7 +144,7 @@ const initialize = Effect.fnUntraced(function* ( .transaction( () => Effect.gen(function* () { - const baselineSeq = yield* events.sequence(sessionID) + const baselineSeq = yield* SessionInput.latestSeq(db, sessionID) yield* db .insert(SessionContextEpochTable) .values({ @@ -93,8 +154,13 @@ const initialize = Effect.fnUntraced(function* ( baseline_seq: baselineSeq, revision: 0, }) - .run() - .pipe(Effect.orDie) + .onConflictDoNothing() + .returning({ sessionID: SessionContextEpochTable.session_id }) + .get() + .pipe( + Effect.orDie, + Effect.flatMap((inserted) => (inserted ? Effect.void : Effect.die(new RevisionMismatch()))), + ) return baselineSeq }), { behavior: "immediate" }, @@ -109,33 +175,22 @@ const replace = Effect.fnUntraced(function* ( baselineSeq: number, generation: SystemContext.Generation, ) { - yield* db - .transaction( - () => - Effect.gen(function* () { - const updated = yield* db - .update(SessionContextEpochTable) - .set({ - baseline: generation.baseline, - snapshot: generation.snapshot, - baseline_seq: baselineSeq, - replacement_seq: null, - revision: expectedRevision + 1, - }) - .where( - and( - eq(SessionContextEpochTable.session_id, sessionID), - eq(SessionContextEpochTable.revision, expectedRevision), - ), - ) - .returning({ revision: SessionContextEpochTable.revision }) - .get() - .pipe(Effect.orDie) - if (!updated) return yield* Effect.die("Session context epoch revision mismatch") - }), - { behavior: "immediate" }, + const updated = yield* db + .update(SessionContextEpochTable) + .set({ + baseline: generation.baseline, + snapshot: generation.snapshot, + baseline_seq: baselineSeq, + replacement_seq: null, + revision: expectedRevision + 1, + }) + .where( + and(eq(SessionContextEpochTable.session_id, sessionID), eq(SessionContextEpochTable.revision, expectedRevision)), ) + .returning({ revision: SessionContextEpochTable.revision }) + .get() .pipe(Effect.orDie) + if (!updated) return yield* Effect.die(new RevisionMismatch()) }) const advance = Effect.fnUntraced(function* ( @@ -157,5 +212,5 @@ const advance = Effect.fnUntraced(function* ( .returning({ revision: SessionContextEpochTable.revision }) .get() .pipe(Effect.orDie) - if (!updated) return yield* Effect.die("Session context epoch revision mismatch") + if (!updated) return yield* Effect.die(new RevisionMismatch()) }) diff --git a/packages/core/src/session/context.ts b/packages/core/src/session/context.ts index 1b41c24fe42..3148dbb8fe6 100644 --- a/packages/core/src/session/context.ts +++ b/packages/core/src/session/context.ts @@ -75,10 +75,7 @@ export const load = Effect.fn("SessionContext.load")(function* (db: DatabaseServ ], { concurrency: "unbounded" }, ) - return yield* Effect.forEach( - yield* messageRows(db, sessionID, compaction, epoch?.baselineSeq), - decodeMessageRow, - ) + return yield* Effect.forEach(yield* messageRows(db, sessionID, compaction, epoch?.baselineSeq), decodeMessageRow) }) export const loadForRunner = Effect.fn("SessionContext.loadForRunner")(function* ( @@ -86,7 +83,10 @@ export const loadForRunner = Effect.fn("SessionContext.loadForRunner")(function* sessionID: SessionSchema.ID, baselineSeq: number, ) { - return yield* Effect.forEach(yield* messageRows(db, sessionID, yield* latestCompaction(db, sessionID), baselineSeq), decodeMessageRow) + return yield* Effect.forEach( + yield* messageRows(db, sessionID, yield* latestCompaction(db, sessionID), baselineSeq), + decodeMessageRow, + ) }) export * as SessionContext from "./context" diff --git a/packages/core/src/session/message.ts b/packages/core/src/session/message.ts index 3a379e72d31..98360cd9368 100644 --- a/packages/core/src/session/message.ts +++ b/packages/core/src/session/message.ts @@ -176,7 +176,16 @@ export class Compaction extends Schema.Class("Session.Message.Compac ...Base, }) {} -export const Message = Schema.Union([AgentSwitched, ModelSwitched, User, Synthetic, System, Shell, Assistant, Compaction]) +export const Message = Schema.Union([ + AgentSwitched, + ModelSwitched, + User, + Synthetic, + System, + Shell, + Assistant, + Compaction, +]) .pipe(Schema.toTaggedUnion("type")) .annotate({ identifier: "Session.Message" }) diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts index 51da5b245ec..4d7e07c90b3 100644 --- a/packages/core/src/session/projector.ts +++ b/packages/core/src/session/projector.ts @@ -422,7 +422,9 @@ export const layer = Layer.effectDiscard( ) yield* events.project(SessionEvent.ContextUpdated, (event) => { if (!event.replay || event.seq === undefined) return run(db, event) - return run(db, event).pipe(Effect.andThen(SessionContextEpoch.requestReplacement(db, event.data.sessionID, event.seq))) + return run(db, event).pipe( + Effect.andThen(SessionContextEpoch.requestReplacement(db, event.data.sessionID, event.seq)), + ) }) yield* events.project(SessionEvent.Synthetic, (event) => run(db, event)) yield* events.project(SessionEvent.Shell.Started, (event) => run(db, event)) diff --git a/packages/core/src/session/runner/index.ts b/packages/core/src/session/runner/index.ts index fe17284b8a5..cfdc6d89430 100644 --- a/packages/core/src/session/runner/index.ts +++ b/packages/core/src/session/runner/index.ts @@ -5,6 +5,7 @@ import { Context, Effect, Schema } from "effect" import { SessionSchema } from "../schema" import type { MessageDecodeError } from "../error" import { SessionRunnerModel } from "./model" +import type { SystemContext } from "../../system-context" export class StepLimitExceededError extends Schema.TaggedErrorClass()( "SessionRunner.StepLimitExceededError", @@ -14,7 +15,12 @@ export class StepLimitExceededError extends Schema.TaggedErrorClass() let needsContinuation = false @@ -150,7 +151,7 @@ export const layer = Layer.effect( yield* SessionInput.promoteSteers(db, events, session.id, cutoff) } } - const system = yield* SessionContextEpoch.prepare(db, events, systemContext, session.id) + const system = initialized ?? (yield* SessionContextEpoch.prepare(db, events, systemContext, session.id)) const context = yield* getRunnerContext(session.id, system.baselineSeq) const request = LLM.request({ model, diff --git a/packages/core/src/system-context-builtins.ts b/packages/core/src/system-context-builtins.ts index f1ff09bc913..8e74788e75d 100644 --- a/packages/core/src/system-context-builtins.ts +++ b/packages/core/src/system-context-builtins.ts @@ -36,7 +36,7 @@ const builtIns = Layer.effectDiscard( }), ]) - yield* registry.contribute({ key: "core/builtins", load: Effect.succeed(context) }) + yield* registry.contribute({ key: SystemContext.Key.make("core/builtins"), load: Effect.succeed(context) }) }), ) diff --git a/packages/core/src/system-context-registry.ts b/packages/core/src/system-context-registry.ts index 65c9b11a197..3e28f30c6e0 100644 --- a/packages/core/src/system-context-registry.ts +++ b/packages/core/src/system-context-registry.ts @@ -4,7 +4,7 @@ import { Context, Effect, Layer, Ref, Scope } from "effect" import { SystemContext } from "./system-context" export interface Contribution { - readonly key: string + readonly key: SystemContext.Key readonly load: Effect.Effect } @@ -36,7 +36,7 @@ export const layer = Layer.effect( ) }), load: Effect.fn("SystemContextRegistry.load")(function* () { - const current = (yield* Ref.get(contributions)).toSorted((a, b) => a.key.localeCompare(b.key)) + const current = (yield* Ref.get(contributions)).toSorted((a, b) => (a.key < b.key ? -1 : a.key > b.key ? 1 : 0)) return SystemContext.combine( yield* Effect.forEach(current, (contribution) => contribution.load, { concurrency: "unbounded" }), ) @@ -44,5 +44,3 @@ export const layer = Layer.effect( }) }), ) - -export const locationLayer = layer diff --git a/packages/core/src/system-context.ts b/packages/core/src/system-context.ts index b69053d9bd5..739305bfeba 100644 --- a/packages/core/src/system-context.ts +++ b/packages/core/src/system-context.ts @@ -67,8 +67,8 @@ export interface Updated { readonly snapshot: Snapshot } -export interface Replaced { - readonly _tag: "Replaced" +export interface ReplacementReady { + readonly _tag: "ReplacementReady" readonly generation: Generation } @@ -76,9 +76,14 @@ export interface ReplacementBlocked { readonly _tag: "ReplacementBlocked" } -export type ReplacementResult = Replaced | ReplacementBlocked +export type ReplacementResult = ReplacementReady | ReplacementBlocked export type ReconcileResult = { readonly _tag: "Unchanged" } | Updated | ReplacementResult +export class InitializationBlocked extends Schema.TaggedErrorClass()( + "SystemContext.InitializationBlocked", + { keys: Schema.Array(Key) }, +) {} + export class DuplicateKeyError extends Schema.TaggedErrorClass()("SystemContext.DuplicateKeyError", { key: Key, }) { @@ -186,8 +191,14 @@ const observe = (value: SystemContext) => ) /** Creates the immutable baseline and durable snapshot for a new generation. */ -export function initialize(value: SystemContext): Effect.Effect { - return observe(value).pipe(Effect.map(initializeObservation)) +export function initialize(value: SystemContext): Effect.Effect { + return observe(value).pipe( + Effect.flatMap((entries) => { + const unavailable = entries.flatMap((entry) => (entry._tag === "Unavailable" ? [entry.key] : [])) + if (unavailable.length > 0) return new InitializationBlocked({ keys: unavailable }) + return Effect.succeed(initializeObservation(entries)) + }), + ) } function initializeObservation(entries: ReadonlyArray): Generation { @@ -272,7 +283,7 @@ export function replace(value: SystemContext, previous: Snapshot): Effect.Effect function replaceObservation(entries: ReadonlyArray, previous: Snapshot): ReplacementResult { if (entries.some((entry) => entry._tag === "Unavailable" && getSnapshot(previous, entry.key) !== undefined)) return { _tag: "ReplacementBlocked" } - return { _tag: "Replaced", generation: initializeObservation(entries) } + return { _tag: "ReplacementReady", generation: initializeObservation(entries) } } function context(sources: ReadonlyArray): SystemContext { diff --git a/packages/core/test/database-migration.test.ts b/packages/core/test/database-migration.test.ts index d840c2c6d86..2cd5510be77 100644 --- a/packages/core/test/database-migration.test.ts +++ b/packages/core/test/database-migration.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, test } from "bun:test" +import { describe, expect } from "bun:test" import { $ } from "bun" import { fileURLToPath } from "url" import path from "path" @@ -7,6 +7,7 @@ import { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite" import { Effect, Layer } from "effect" import { eq, inArray, sql } from "drizzle-orm" import { DatabaseMigration } from "@opencode-ai/core/database/migration" +import { migrations } from "@opencode-ai/core/database/migration.gen" import sessionUsageMigration from "@opencode-ai/core/database/migration/20260510033149_session_usage" import normalizeStoragePathsMigration from "@opencode-ai/core/database/migration/20260601010001_normalize_storage_paths" import sessionMessageProjectionOrderMigration from "@opencode-ai/core/database/migration/20260603040000_session_message_projection_order" @@ -17,43 +18,45 @@ import { AbsolutePath } from "@opencode-ai/core/schema" import { SessionSchema } from "@opencode-ai/core/session/schema" import { SessionTable } from "@opencode-ai/core/session/sql" import sessionMetadataMigration from "@opencode-ai/core/database/migration/20260511173437_session-metadata" -import type { SqlClient as SqlClientService } from "effect/unstable/sql/SqlClient" import { Database } from "@opencode-ai/core/database/database" import { tmpdir } from "./fixture/tmpdir" - -const run = (effect: Effect.Effect) => - Effect.runPromise( - effect.pipe(Effect.provide(SqliteClient.layer({ filename: ":memory:", disableWAL: true })), Effect.scoped), - ) +import { testEffect } from "./lib/effect" const makeDb = EffectDrizzleSqlite.makeWithDefaults() +const it = testEffect(SqliteClient.layer({ filename: ":memory:", disableWAL: true })) describe("DatabaseMigration", () => { - test("serializes concurrent embedded initialization for one database path", async () => { - await using tmp = await tmpdir() - const filename = path.join(tmp.path, "embedded.sqlite") - const layers = [Database.layerFromPath(filename), Database.layerFromPath(filename)] + it.effect("serializes concurrent embedded initialization for one database path", () => + Effect.promise(async () => { + await using tmp = await tmpdir() + const filename = path.join(tmp.path, "embedded.sqlite") + const layers = [Database.layerFromPath(filename), Database.layerFromPath(filename)] - await Effect.runPromise( - Effect.all( - layers.map((layer) => Effect.scoped(Layer.build(layer))), - { concurrency: "unbounded" }, - ), - ) - }) + await Effect.runPromise( + Effect.all( + layers.map((layer) => Effect.scoped(Layer.build(layer))), + { concurrency: "unbounded" }, + ), + ) + }), + ) if (process.platform === "linux") { - test("declared schema has no ungenerated migrations", async () => { - const result = await $`bun ${fileURLToPath(new URL("../script/migration.ts", import.meta.url))} --check` - .quiet() - .nothrow() - expect(result.exitCode, result.stderr.toString()).toBe(0) - expect(result.stdout.toString()).toContain("No schema changes, nothing to migrate") - }, 30_000) + it.effect( + "declared schema has no ungenerated migrations", + () => + Effect.promise(async () => { + const result = await $`bun ${fileURLToPath(new URL("../script/migration.ts", import.meta.url))} --check` + .quiet() + .nothrow() + expect(result.exitCode, result.stderr.toString()).toBe(0) + expect(result.stdout.toString()).toContain("No schema changes, nothing to migrate") + }), + 30_000, + ) } - test("applies tracked migrations to an empty database", async () => { - await run( - Effect.gen(function* () { + it.effect("applies tracked migrations to an empty database", () => + Effect.gen(function* () { const db = yield* makeDb yield* DatabaseMigration.apply(db) @@ -66,10 +69,7 @@ describe("DatabaseMigration", () => { expect( yield* db.get(sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'session_context_epoch'`), ).toEqual({ name: "session_context_epoch" }) - 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: 31 }) + expect(yield* db.get(sql`SELECT count(*) as count FROM migration`)).toEqual({ count: migrations.length }) 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_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`, @@ -84,13 +84,11 @@ describe("DatabaseMigration", () => { { name: "session_message_session_time_created_id_idx" }, { name: "session_message_session_type_seq_idx" }, ]) - }), - ) - }) + }), + ) - test("resets beta history and rebuilds event-sourced Session input storage", async () => { - await run( - Effect.gen(function* () { + it.effect("resets beta history and rebuilds event-sourced Session input storage", () => + Effect.gen(function* () { const db = yield* makeDb yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY, workspace_id text)`) yield* db.run(sql`CREATE TABLE workspace (id text PRIMARY KEY)`) @@ -160,13 +158,11 @@ describe("DatabaseMigration", () => { expect.objectContaining({ name: "session_input_session_promoted_seq_idx", unique: 1 }), expect.objectContaining({ name: "session_input_session_admitted_seq_idx", unique: 1 }), ]) - }), - ) - }) + }), + ) - test("resets incompatible projected Session messages before adding sequence order", async () => { - await run( - Effect.gen(function* () { + it.effect("resets incompatible projected Session messages before adding sequence order", () => + Effect.gen(function* () { const db = yield* makeDb yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY)`) yield* db.run( @@ -215,13 +211,11 @@ describe("DatabaseMigration", () => { sql`INSERT INTO session_message (id, session_id, type, seq, time_created, time_updated, data) VALUES ('fresh_projection', 'session', 'user', 7, 2, 2, '{}')`, ) expect(yield* db.get(sql`SELECT id, seq FROM session_message`)).toEqual({ id: "fresh_projection", seq: 7 }) - }), - ) - }) + }), + ) - test("runs session usage backfill in order with schema changes", async () => { - await run( - Effect.gen(function* () { + it.effect("runs session usage backfill in order with schema changes", () => + Effect.gen(function* () { const db = yield* makeDb yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY, time_updated integer NOT NULL)`) yield* db.run(sql`CREATE TABLE message (id text PRIMARY KEY, session_id text NOT NULL, data text NOT NULL)`) @@ -244,13 +238,11 @@ describe("DatabaseMigration", () => { tokens_cache_read: 5, tokens_cache_write: 6, }) - }), - ) - }) + }), + ) - test("normalizes Windows storage paths and leaves POSIX paths untouched", async () => { - await run( - Effect.gen(function* () { + it.effect("normalizes Windows storage paths and leaves POSIX paths untouched", () => + Effect.gen(function* () { const db = yield* makeDb yield* db.run(sql`CREATE TABLE project (id text PRIMARY KEY, worktree text NOT NULL, sandboxes text NOT NULL)`) yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY, directory text NOT NULL, path text)`) @@ -295,14 +287,12 @@ describe("DatabaseMigration", () => { directory: "/home/me/we\\ird", path: "src\\weird", }) - }), - ) - }) + }), + ) - test("maps native Windows paths through database columns", async () => { - if (process.platform !== "win32") return - await run( - Effect.gen(function* () { + it.effect("maps native Windows paths through database columns", () => { + if (process.platform !== "win32") return Effect.void + return Effect.gen(function* () { const db = yield* makeDb yield* DatabaseMigration.apply(db) const projectID = ProjectV2.ID.make("codec_project") @@ -405,13 +395,11 @@ describe("DatabaseMigration", () => { expect(() => Effect.runSync(db.select().from(ProjectTable).where(eq(ProjectTable.id, projectID)).get()), ).toThrow() - }), - ) + }) }) - test("imports existing drizzle migration state", async () => { - await run( - Effect.gen(function* () { + it.effect("imports existing drizzle migration state", () => + Effect.gen(function* () { const db = yield* makeDb yield* db.run( sql`CREATE TABLE __drizzle_migrations (id INTEGER PRIMARY KEY, hash text NOT NULL, created_at numeric, name text, applied_at TEXT)`, @@ -424,13 +412,11 @@ describe("DatabaseMigration", () => { yield* DatabaseMigration.applyOnly(db, []) expect(yield* db.get(sql`SELECT id FROM migration`)).toEqual({ id: "20260127222353_familiar_lady_ursula" }) - }), - ) - }) + }), + ) - test("does not replay a migrated session metadata column", async () => { - await run( - Effect.gen(function* () { + it.effect("does not replay a migrated session metadata column", () => + Effect.gen(function* () { const db = yield* makeDb yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY, metadata text)`) yield* db.run( @@ -444,13 +430,11 @@ describe("DatabaseMigration", () => { yield* DatabaseMigration.applyOnly(db, [sessionMetadataMigration]) expect(yield* db.all(sql`SELECT id FROM migration`)).toEqual([{ id: "20260511173437_session-metadata" }]) - }), - ) - }) + }), + ) - test("accepts the temporary replacement session metadata migration id", async () => { - await run( - Effect.gen(function* () { + it.effect("accepts the temporary replacement session metadata migration id", () => + Effect.gen(function* () { const db = yield* makeDb yield* db.run(sql`CREATE TABLE session (id text PRIMARY KEY, metadata text)`) yield* db.run(sql`CREATE TABLE migration (id TEXT PRIMARY KEY, time_completed INTEGER NOT NULL)`) @@ -462,13 +446,11 @@ describe("DatabaseMigration", () => { { id: "20260511173437_session-metadata" }, { id: "20260530232709_lovely_romulus" }, ]) - }), - ) - }) + }), + ) - test("skips drizzle import when migration table already has state", async () => { - await run( - Effect.gen(function* () { + it.effect("skips drizzle import when migration table already has state", () => + Effect.gen(function* () { const db = yield* makeDb yield* db.run(sql`CREATE TABLE migration (id TEXT PRIMARY KEY, time_completed INTEGER NOT NULL)`) yield* db.run(sql`INSERT INTO migration (id, time_completed) VALUES ('existing', 1)`) @@ -483,7 +465,6 @@ describe("DatabaseMigration", () => { yield* DatabaseMigration.applyOnly(db, []) expect(yield* db.all(sql`SELECT id FROM migration ORDER BY id`)).toEqual([{ id: "existing" }]) - }), - ) - }) + }), + ) }) diff --git a/packages/core/test/event.test.ts b/packages/core/test/event.test.ts index 9b454333b4b..cd8ba69253f 100644 --- a/packages/core/test/event.test.ts +++ b/packages/core/test/event.test.ts @@ -194,11 +194,12 @@ describe("EventV2", () => { Effect.gen(function* () { const events = yield* EventV2.Service const received = new Array() + const aggregateID = EventV2.ID.create() yield* events.project(SyncMessage, () => Effect.sync(() => received.push("projector"))) yield* events.publish( SyncMessage, - { id: "one", text: "hello" }, + { id: aggregateID, text: "hello" }, { commit: (seq) => Effect.sync(() => received.push(`commit:${seq}`)) }, ) @@ -224,7 +225,9 @@ describe("EventV2", () => { expect(String(exit)).toContain("commit failed") expect(yield* db.all("SELECT value FROM event_commit_probe")).toEqual([]) expect(yield* db.select().from(EventTable).where(eq(EventTable.aggregate_id, aggregateID)).all()).toEqual([]) - expect(yield* db.select().from(EventSequenceTable).where(eq(EventSequenceTable.aggregate_id, aggregateID)).all()).toEqual([]) + expect( + yield* db.select().from(EventSequenceTable).where(eq(EventSequenceTable.aggregate_id, aggregateID)).all(), + ).toEqual([]) }), ) diff --git a/packages/core/test/instruction-context.test.ts b/packages/core/test/instruction-context.test.ts index 2eea3586fb4..9b1657caac9 100644 --- a/packages/core/test/instruction-context.test.ts +++ b/packages/core/test/instruction-context.test.ts @@ -117,7 +117,9 @@ describe("InstructionContext", () => { const failingFS = Layer.effect( FSUtil.Service, FSUtil.Service.pipe( - Effect.map((fs) => FSUtil.Service.of({ ...fs, up: () => Effect.fail(new FSUtil.FileSystemError({ method: "up" })) })), + Effect.map((fs) => + FSUtil.Service.of({ ...fs, up: () => Effect.fail(new FSUtil.FileSystemError({ method: "up" })) }), + ), ), ).pipe(Layer.provide(FSUtil.defaultLayer)) const context = yield* SystemContextRegistry.Service.pipe( @@ -126,10 +128,7 @@ describe("InstructionContext", () => { Effect.provide(failingFS), Effect.provide(Global.layerWith({ config: "/global" })), Effect.provide( - Layer.succeed( - Location.Service, - Location.Service.of(location({ directory: AbsolutePath.make("/repo") })), - ), + Layer.succeed(Location.Service, Location.Service.of(location({ directory: AbsolutePath.make("/repo") }))), ), ) @@ -165,10 +164,7 @@ describe("InstructionContext", () => { Effect.provide(racingFS), Effect.provide(Global.layerWith({ config: "/global" })), Effect.provide( - Layer.succeed( - Location.Service, - Location.Service.of(location({ directory: AbsolutePath.make("/repo") })), - ), + Layer.succeed(Location.Service, Location.Service.of(location({ directory: AbsolutePath.make("/repo") }))), ), ) diff --git a/packages/core/test/session-runner-message.test.ts b/packages/core/test/session-runner-message.test.ts index 854f24158f4..66c90583ab1 100644 --- a/packages/core/test/session-runner-message.test.ts +++ b/packages/core/test/session-runner-message.test.ts @@ -16,375 +16,388 @@ const id = (value: string) => SessionMessage.ID.make(`msg_${value}`) const model = Model.make({ id: "model", provider: "provider", route: OpenAIChat.route }) describe("toLLMMessages", () => { - it.effect("maps every top-level V2 Session message type", () => Effect.sync(() => { - const file = new FileAttachment({ uri: "data:image/png;base64,aGVsbG8=", mime: "image/png", name: "hello.png" }) - const reference = new ReferenceAttachment({ name: "docs", kind: "local", uri: "file:///docs" }) - const messages = toLLMMessages( - [ - new SessionMessage.AgentSwitched({ - id: id("agent"), - type: "agent-switched", - agent: "build", - time: { created }, - }), - new SessionMessage.ModelSwitched({ - id: id("model"), - type: "model-switched", - model: { id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") }, - time: { created }, - }), - new SessionMessage.User({ - id: id("user"), - type: "user", - text: "Inspect this image", - files: [file], - agents: [new AgentAttachment({ name: "build" })], - references: [reference], - time: { created }, - }), - new SessionMessage.Synthetic({ - id: id("synthetic"), - type: "synthetic", - sessionID: SessionV2.ID.make("ses_translate"), - text: "Synthetic context", - time: { created }, - }), - new SessionMessage.Shell({ - id: id("shell"), - type: "shell", - callID: "shell-1", - command: "pwd", - output: "/project", - time: { created, completed: created }, - }), - new SessionMessage.Compaction({ - id: id("compaction"), - type: "compaction", - reason: "auto", - summary: "Earlier work", - time: { created }, - }), - ], - model, - ) - - expect(messages.map((message) => message.role)).toEqual(["user", "user", "user", "user"]) - expect(messages[0]).toEqual( - Message.make({ - id: id("user"), - role: "user", - content: [ - { type: "text", text: "Inspect this image" }, - { type: "media", mediaType: "image/png", data: "data:image/png;base64,aGVsbG8=", filename: "hello.png" }, - ], - metadata: { agents: [{ name: "build" }], references: [reference] }, - }), - ) - expect(messages.slice(1).map((message) => message.content)).toEqual([ - [{ type: "text", text: "Synthetic context" }], - [{ type: "text", text: "Shell command: pwd\n\n/project" }], - [{ type: "text", text: "Summary of earlier conversation:\nEarlier work" }], - ]) - })) - - it.effect("maps durable Session system messages into chronological system messages", () => Effect.sync(() => { - expect( - toLLMMessages( + it.effect("maps every top-level V2 Session message type", () => + Effect.sync(() => { + const file = new FileAttachment({ uri: "data:image/png;base64,aGVsbG8=", mime: "image/png", name: "hello.png" }) + const reference = new ReferenceAttachment({ name: "docs", kind: "local", uri: "file:///docs" }) + const messages = toLLMMessages( [ - new SessionMessage.System({ id: id("system"), type: "system", text: "Updated context\n\nOther context", time: { created } }), + new SessionMessage.AgentSwitched({ + id: id("agent"), + type: "agent-switched", + agent: "build", + time: { created }, + }), + new SessionMessage.ModelSwitched({ + id: id("model"), + type: "model-switched", + model: { id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") }, + time: { created }, + }), + new SessionMessage.User({ + id: id("user"), + type: "user", + text: "Inspect this image", + files: [file], + agents: [new AgentAttachment({ name: "build" })], + references: [reference], + time: { created }, + }), + new SessionMessage.Synthetic({ + id: id("synthetic"), + type: "synthetic", + sessionID: SessionV2.ID.make("ses_translate"), + text: "Synthetic context", + time: { created }, + }), + new SessionMessage.Shell({ + id: id("shell"), + type: "shell", + callID: "shell-1", + command: "pwd", + output: "/project", + time: { created, completed: created }, + }), + new SessionMessage.Compaction({ + id: id("compaction"), + type: "compaction", + reason: "auto", + summary: "Earlier work", + time: { created }, + }), ], model, - ), - ).toEqual([ - Message.system("Updated context\n\nOther context"), - ]) - })) + ) - it.effect("expands assistant tool calls and settled outcomes into canonical tool messages", () => Effect.sync(() => { - const messages = toLLMMessages( - [ - new SessionMessage.Assistant({ - id: id("assistant"), - type: "assistant", - agent: "build", - model: { id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") }, + expect(messages.map((message) => message.role)).toEqual(["user", "user", "user", "user"]) + expect(messages[0]).toEqual( + Message.make({ + id: id("user"), + role: "user", content: [ - new SessionMessage.AssistantText({ type: "text", id: "text-1", text: "Checking" }), - new SessionMessage.AssistantReasoning({ - type: "reasoning", - id: "reasoning-1", - text: "Think", - providerMetadata: { anthropic: { signature: "sig_1" } }, - }), - new SessionMessage.AssistantTool({ - type: "tool", - id: "pending", - name: "read", - state: new SessionMessage.ToolStatePending({ status: "pending", input: '{"path":"README.md"}' }), + { type: "text", text: "Inspect this image" }, + { type: "media", mediaType: "image/png", data: "data:image/png;base64,aGVsbG8=", filename: "hello.png" }, + ], + metadata: { agents: [{ name: "build" }], references: [reference] }, + }), + ) + expect(messages.slice(1).map((message) => message.content)).toEqual([ + [{ type: "text", text: "Synthetic context" }], + [{ type: "text", text: "Shell command: pwd\n\n/project" }], + [{ type: "text", text: "Summary of earlier conversation:\nEarlier work" }], + ]) + }), + ) + + it.effect("maps durable Session system messages into chronological system messages", () => + Effect.sync(() => { + expect( + toLLMMessages( + [ + new SessionMessage.System({ + id: id("system"), + type: "system", + text: "Updated context\n\nOther context", time: { created }, }), - new SessionMessage.AssistantTool({ - type: "tool", - id: "running", - name: "read", - state: new SessionMessage.ToolStateRunning({ - status: "running", - input: { path: "README.md" }, - content: [], - structured: {}, - }), - time: { created }, - }), - new SessionMessage.AssistantTool({ - type: "tool", - id: "completed", - name: "read", - state: new SessionMessage.ToolStateCompleted({ - status: "completed", - input: { path: "README.md" }, - content: [ - new ToolOutput.TextContent({ type: "text", text: "Hello" }), - new ToolOutput.FileContent({ - type: "file", - source: { type: "data", data: "aGVsbG8=" }, - mime: "image/png", - name: "hello.png", - }), - ], - structured: {}, - }), - time: { created, completed: created }, - }), - new SessionMessage.AssistantTool({ - type: "tool", - id: "hosted", - name: "web_search", - provider: { - executed: true, - metadata: { fake: { continuation: "hosted-call" } }, - resultMetadata: { fake: { continuation: "hosted-result" } }, - }, - state: new SessionMessage.ToolStateCompleted({ - status: "completed", - input: { query: "Effect" }, - content: [new ToolOutput.TextContent({ type: "text", text: "Found it" })], - structured: {}, - }), - time: { created, completed: created }, - }), - new SessionMessage.AssistantTool({ - type: "tool", - id: "hosted-failed", - name: "write", - provider: { executed: true, metadata: { fake: { continuation: "failed" } } }, - state: new SessionMessage.ToolStateError({ - status: "error", - input: { path: "README.md" }, - content: [], - structured: {}, - error: { type: "unknown", message: "Denied" }, - }), - time: { created, completed: created }, - }), ], - time: { created, completed: created }, - }), - ], - model, - ) + model, + ), + ).toEqual([Message.system("Updated context\n\nOther context")]) + }), + ) - expect(messages.map((message) => message.role)).toEqual(["assistant", "tool"]) - expect(messages[0]?.content).toEqual([ - { type: "text", text: "Checking" }, - { type: "reasoning", text: "Think", providerMetadata: { anthropic: { signature: "sig_1" } } }, - { type: "tool-call", id: "pending", name: "read", input: { path: "README.md" } }, - { type: "tool-call", id: "running", name: "read", input: { path: "README.md" } }, - { - type: "tool-call", - id: "completed", - name: "read", - input: { path: "README.md" }, - }, - { - type: "tool-call", - id: "hosted", - name: "web_search", - input: { query: "Effect" }, - providerExecuted: true, - providerMetadata: { fake: { continuation: "hosted-call" } }, - }, - { - type: "tool-result", - id: "hosted", - name: "web_search", - providerExecuted: true, - providerMetadata: { fake: { continuation: "hosted-result" } }, - result: { type: "text", value: "Found it" }, - }, - { - type: "tool-call", - id: "hosted-failed", - name: "write", - input: { path: "README.md" }, - providerExecuted: true, - providerMetadata: { fake: { continuation: "failed" } }, - }, - { - type: "tool-result", - id: "hosted-failed", - name: "write", - providerExecuted: true, - providerMetadata: { fake: { continuation: "failed" } }, - result: { - type: "error", - value: { error: { type: "unknown", message: "Denied" }, content: [], structured: {} }, + it.effect("expands assistant tool calls and settled outcomes into canonical tool messages", () => + Effect.sync(() => { + const messages = toLLMMessages( + [ + new SessionMessage.Assistant({ + id: id("assistant"), + type: "assistant", + agent: "build", + model: { id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") }, + content: [ + new SessionMessage.AssistantText({ type: "text", id: "text-1", text: "Checking" }), + new SessionMessage.AssistantReasoning({ + type: "reasoning", + id: "reasoning-1", + text: "Think", + providerMetadata: { anthropic: { signature: "sig_1" } }, + }), + new SessionMessage.AssistantTool({ + type: "tool", + id: "pending", + name: "read", + state: new SessionMessage.ToolStatePending({ status: "pending", input: '{"path":"README.md"}' }), + time: { created }, + }), + new SessionMessage.AssistantTool({ + type: "tool", + id: "running", + name: "read", + state: new SessionMessage.ToolStateRunning({ + status: "running", + input: { path: "README.md" }, + content: [], + structured: {}, + }), + time: { created }, + }), + new SessionMessage.AssistantTool({ + type: "tool", + id: "completed", + name: "read", + state: new SessionMessage.ToolStateCompleted({ + status: "completed", + input: { path: "README.md" }, + content: [ + new ToolOutput.TextContent({ type: "text", text: "Hello" }), + new ToolOutput.FileContent({ + type: "file", + source: { type: "data", data: "aGVsbG8=" }, + mime: "image/png", + name: "hello.png", + }), + ], + structured: {}, + }), + time: { created, completed: created }, + }), + new SessionMessage.AssistantTool({ + type: "tool", + id: "hosted", + name: "web_search", + provider: { + executed: true, + metadata: { fake: { continuation: "hosted-call" } }, + resultMetadata: { fake: { continuation: "hosted-result" } }, + }, + state: new SessionMessage.ToolStateCompleted({ + status: "completed", + input: { query: "Effect" }, + content: [new ToolOutput.TextContent({ type: "text", text: "Found it" })], + structured: {}, + }), + time: { created, completed: created }, + }), + new SessionMessage.AssistantTool({ + type: "tool", + id: "hosted-failed", + name: "write", + provider: { executed: true, metadata: { fake: { continuation: "failed" } } }, + state: new SessionMessage.ToolStateError({ + status: "error", + input: { path: "README.md" }, + content: [], + structured: {}, + error: { type: "unknown", message: "Denied" }, + }), + time: { created, completed: created }, + }), + ], + time: { created, completed: created }, + }), + ], + model, + ) + + expect(messages.map((message) => message.role)).toEqual(["assistant", "tool"]) + expect(messages[0]?.content).toEqual([ + { type: "text", text: "Checking" }, + { type: "reasoning", text: "Think", providerMetadata: { anthropic: { signature: "sig_1" } } }, + { type: "tool-call", id: "pending", name: "read", input: { path: "README.md" } }, + { type: "tool-call", id: "running", name: "read", input: { path: "README.md" } }, + { + type: "tool-call", + id: "completed", + name: "read", + input: { path: "README.md" }, }, - }, - ]) - expect(messages[1]?.content).toEqual([ - { - type: "tool-result", - id: "completed", - name: "read", - result: { - type: "content", - value: [ - { type: "text", text: "Hello" }, - { type: "media", mediaType: "image/png", data: "aGVsbG8=", filename: "hello.png" }, - ], + { + type: "tool-call", + id: "hosted", + name: "web_search", + input: { query: "Effect" }, + providerExecuted: true, + providerMetadata: { fake: { continuation: "hosted-call" } }, }, - }, - ]) - })) + { + type: "tool-result", + id: "hosted", + name: "web_search", + providerExecuted: true, + providerMetadata: { fake: { continuation: "hosted-result" } }, + result: { type: "text", value: "Found it" }, + }, + { + type: "tool-call", + id: "hosted-failed", + name: "write", + input: { path: "README.md" }, + providerExecuted: true, + providerMetadata: { fake: { continuation: "failed" } }, + }, + { + type: "tool-result", + id: "hosted-failed", + name: "write", + providerExecuted: true, + providerMetadata: { fake: { continuation: "failed" } }, + result: { + type: "error", + value: { error: { type: "unknown", message: "Denied" }, content: [], structured: {} }, + }, + }, + ]) + expect(messages[1]?.content).toEqual([ + { + type: "tool-result", + id: "completed", + name: "read", + result: { + type: "content", + value: [ + { type: "text", text: "Hello" }, + { type: "media", mediaType: "image/png", data: "aGVsbG8=", filename: "hello.png" }, + ], + }, + }, + ]) + }), + ) - it.effect("restores OpenAI encrypted reasoning metadata", () => Effect.sync(() => { - const messages = toLLMMessages( - [ - new SessionMessage.Assistant({ - id: id("assistant-openai-reasoning"), - type: "assistant", - agent: "build", - model: { id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") }, - content: [ - new SessionMessage.AssistantReasoning({ - type: "reasoning", - id: "reasoning-openai", - text: "Think", - providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "encrypted-state" } }, - }), - ], - time: { created, completed: created }, - }), - ], - model, - ) - - expect(messages[0]?.content).toEqual([ - { - type: "reasoning", - text: "Think", - providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "encrypted-state" } }, - }, - ]) - })) - - it.effect("drops provider-native continuation metadata after a model switch", () => Effect.sync(() => { - const messages = toLLMMessages( - [ - new SessionMessage.Assistant({ - id: id("assistant-old-model"), - type: "assistant", - agent: "build", - model: { id: ModelV2.ID.make("old-model"), providerID: ProviderV2.ID.make("provider") }, - content: [ - new SessionMessage.AssistantReasoning({ - type: "reasoning", - id: "reasoning-old-model", - text: "Visible thought", - providerMetadata: { anthropic: { signature: "sig_old" } }, - }), - new SessionMessage.AssistantTool({ - type: "tool", - id: "hosted-old-model", - name: "web_search", - provider: { - executed: true, - metadata: { openai: { itemId: "hosted-old-model" } }, - resultMetadata: { openai: { itemId: "hosted-old-model" } }, - }, - state: new SessionMessage.ToolStateCompleted({ - status: "completed", - input: { query: "Effect" }, - content: [], - structured: {}, - result: { type: "json", value: { status: "completed" } }, + it.effect("restores OpenAI encrypted reasoning metadata", () => + Effect.sync(() => { + const messages = toLLMMessages( + [ + new SessionMessage.Assistant({ + id: id("assistant-openai-reasoning"), + type: "assistant", + agent: "build", + model: { id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") }, + content: [ + new SessionMessage.AssistantReasoning({ + type: "reasoning", + id: "reasoning-openai", + text: "Think", + providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "encrypted-state" } }, }), - time: { created, completed: created }, - }), - new SessionMessage.AssistantTool({ - type: "tool", - id: "local-old-model", - name: "read", - provider: { - executed: false, - metadata: { fake: { call: "old" } }, - resultMetadata: { fake: { result: "old" } }, - }, - state: new SessionMessage.ToolStateCompleted({ - status: "completed", - input: { path: "README.md" }, - content: [], - structured: { text: "Hello" }, - }), - time: { created, completed: created }, - }), - ], - time: { created, completed: created }, - }), - ], - model, - ) + ], + time: { created, completed: created }, + }), + ], + model, + ) - expect(messages[0]?.content).toEqual([ - { type: "text", text: "Visible thought" }, - { - type: "tool-call", - id: "hosted-old-model", - name: "web_search", - input: { query: "Effect" }, - providerExecuted: true, - providerMetadata: undefined, - }, - { - type: "tool-result", - id: "hosted-old-model", - name: "web_search", - result: { type: "json", value: { status: "completed" } }, - providerExecuted: true, - cache: undefined, - metadata: undefined, - providerMetadata: undefined, - }, - { - type: "tool-call", - id: "local-old-model", - name: "read", - input: { path: "README.md" }, - providerExecuted: false, - providerMetadata: undefined, - }, - ]) - expect(messages[1]?.content).toEqual([ - { - type: "tool-result", - id: "local-old-model", - name: "read", - result: { type: "json", value: { text: "Hello" } }, - providerExecuted: false, - cache: undefined, - metadata: undefined, - providerMetadata: undefined, - }, - ]) - })) + expect(messages[0]?.content).toEqual([ + { + type: "reasoning", + text: "Think", + providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "encrypted-state" } }, + }, + ]) + }), + ) + + it.effect("drops provider-native continuation metadata after a model switch", () => + Effect.sync(() => { + const messages = toLLMMessages( + [ + new SessionMessage.Assistant({ + id: id("assistant-old-model"), + type: "assistant", + agent: "build", + model: { id: ModelV2.ID.make("old-model"), providerID: ProviderV2.ID.make("provider") }, + content: [ + new SessionMessage.AssistantReasoning({ + type: "reasoning", + id: "reasoning-old-model", + text: "Visible thought", + providerMetadata: { anthropic: { signature: "sig_old" } }, + }), + new SessionMessage.AssistantTool({ + type: "tool", + id: "hosted-old-model", + name: "web_search", + provider: { + executed: true, + metadata: { openai: { itemId: "hosted-old-model" } }, + resultMetadata: { openai: { itemId: "hosted-old-model" } }, + }, + state: new SessionMessage.ToolStateCompleted({ + status: "completed", + input: { query: "Effect" }, + content: [], + structured: {}, + result: { type: "json", value: { status: "completed" } }, + }), + time: { created, completed: created }, + }), + new SessionMessage.AssistantTool({ + type: "tool", + id: "local-old-model", + name: "read", + provider: { + executed: false, + metadata: { fake: { call: "old" } }, + resultMetadata: { fake: { result: "old" } }, + }, + state: new SessionMessage.ToolStateCompleted({ + status: "completed", + input: { path: "README.md" }, + content: [], + structured: { text: "Hello" }, + }), + time: { created, completed: created }, + }), + ], + time: { created, completed: created }, + }), + ], + model, + ) + + expect(messages[0]?.content).toEqual([ + { type: "text", text: "Visible thought" }, + { + type: "tool-call", + id: "hosted-old-model", + name: "web_search", + input: { query: "Effect" }, + providerExecuted: true, + providerMetadata: undefined, + }, + { + type: "tool-result", + id: "hosted-old-model", + name: "web_search", + result: { type: "json", value: { status: "completed" } }, + providerExecuted: true, + cache: undefined, + metadata: undefined, + providerMetadata: undefined, + }, + { + type: "tool-call", + id: "local-old-model", + name: "read", + input: { path: "README.md" }, + providerExecuted: false, + providerMetadata: undefined, + }, + ]) + expect(messages[1]?.content).toEqual([ + { + type: "tool-result", + id: "local-old-model", + name: "read", + result: { type: "json", value: { text: "Hello" } }, + providerExecuted: false, + cache: undefined, + metadata: undefined, + providerMetadata: undefined, + }, + ]) + }), + ) }) diff --git a/packages/core/test/session-runner-recorded.test.ts b/packages/core/test/session-runner-recorded.test.ts index edd244e4328..99410e6efd9 100644 --- a/packages/core/test/session-runner-recorded.test.ts +++ b/packages/core/test/session-runner-recorded.test.ts @@ -57,14 +57,15 @@ const model = OpenAIChat.route }) .model({ id: "gpt-4o-mini" }) const models = SessionRunnerModel.layerWith(() => Effect.succeed(model)) +const systemContextKey = SystemContext.Key.make("test/context") const systemContext = Layer.effectDiscard( SystemContextRegistry.Service.pipe( Effect.flatMap((registry) => registry.contribute({ - key: "test/context", + key: systemContextKey, load: Effect.succeed( SystemContext.make({ - key: SystemContext.Key.make("test/context"), + key: systemContextKey, codec: Schema.toCodecJson(Schema.String), load: Effect.succeed("Recorded context"), baseline: String, diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index ef8b8617c8e..10a87130bef 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -32,13 +32,18 @@ import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model" import { ToolRegistry } from "@opencode-ai/core/tool/registry" import { ApplicationTools } from "@opencode-ai/core/tool/application-tools" import { NativeTool } from "@opencode-ai/core/tool/native" -import { SessionContextEpochTable, SessionInputTable, SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql" +import { + SessionContextEpochTable, + SessionInputTable, + SessionMessageTable, + SessionTable, +} from "@opencode-ai/core/session/sql" import { SessionStore } from "@opencode-ai/core/session/store" import { SystemContext } from "@opencode-ai/core/system-context" import { SystemContextRegistry } from "@opencode-ai/core/system-context-registry" import { ModelV2 } from "@opencode-ai/core/model" import { ProviderV2 } from "@opencode-ai/core/provider" -import { Cause, DateTime, Deferred, Effect, Fiber, Layer, Schema, Stream } from "effect" +import { Cause, DateTime, Deferred, Effect, Exit, Fiber, Layer, Schema, Stream } from "effect" import { asc, eq } from "drizzle-orm" import { testEffect } from "./lib/effect" @@ -145,11 +150,12 @@ const systemContextKey = SystemContext.Key.make("test/context") let systemBaseline = "Initial context" let systemRemoved = false let systemUnavailable = false +let systemLoadHook = Effect.void const systemContext = Layer.effectDiscard( SystemContextRegistry.Service.pipe( Effect.flatMap((registry) => registry.contribute({ - key: "test/context", + key: systemContextKey, load: Effect.sync(() => SystemContext.combine( systemRemoved @@ -158,7 +164,11 @@ const systemContext = Layer.effectDiscard( SystemContext.make({ key: systemContextKey, codec: Schema.toCodecJson(Schema.String), - load: Effect.sync(() => (systemUnavailable ? SystemContext.unavailable : systemBaseline)), + load: systemLoadHook.pipe( + Effect.andThen( + Effect.sync(() => (systemUnavailable ? SystemContext.unavailable : systemBaseline)), + ), + ), baseline: String, update: (_previous, current) => current, removed: () => "System context source removed: test/context", @@ -240,6 +250,7 @@ const setup = Effect.gen(function* () { systemBaseline = "Initial context" systemRemoved = false systemUnavailable = false + systemLoadHook = Effect.void responses = undefined streamFailure = undefined responseStream = undefined @@ -552,6 +563,39 @@ describe("SessionRunnerLLM", () => { }), ) + it.effect("retries the first provider turn after system context becomes available", () => + Effect.gen(function* () { + yield* setup + const session = yield* SessionV2.Service + const { db } = yield* Database.Service + const messageID = SessionMessage.ID.create() + systemUnavailable = true + yield* session.prompt({ id: messageID, sessionID, prompt: new Prompt({ text: "First" }), resume: false }) + requests.length = 0 + + const exit = yield* session.resume(sessionID).pipe(Effect.exit) + + expect(Exit.isFailure(exit)).toBe(true) + if (Exit.isFailure(exit)) expect(Cause.squash(exit.cause)).toBeInstanceOf(SystemContext.InitializationBlocked) + expect(requests).toHaveLength(0) + expect(yield* SessionInput.hasPending(db, sessionID, "steer")).toBe(true) + expect( + yield* db + .select() + .from(SessionContextEpochTable) + .where(eq(SessionContextEpochTable.session_id, sessionID)) + .get(), + ).toBeUndefined() + + systemUnavailable = false + yield* session.prompt({ id: messageID, sessionID, prompt: new Prompt({ text: "First" }) }) + yield* (yield* SessionRunCoordinator.Service).awaitIdle(sessionID) + + expect(requests).toHaveLength(1) + expect(requests[0]?.messages.map((message) => message.role)).toEqual(["user"]) + }), + ) + it.effect("reuses one durable baseline after the context producer changes", () => Effect.gen(function* () { yield* setup @@ -622,6 +666,7 @@ describe("SessionRunnerLLM", () => { yield* session.resume(sessionID) yield* events.publish(SessionEvent.ModelSwitched, { sessionID, + messageID: SessionMessage.ID.create(), timestamp: DateTime.makeUnsafe(1), model: { id: ModelV2.ID.make("replacement"), providerID: ProviderV2.ID.make("fake") }, }) @@ -661,6 +706,7 @@ describe("SessionRunnerLLM", () => { yield* session.resume(sessionID) yield* events.publish(SessionEvent.ModelSwitched, { sessionID, + messageID: SessionMessage.ID.create(), timestamp: DateTime.makeUnsafe(1), model: { id: ModelV2.ID.make("replacement"), providerID: ProviderV2.ID.make("fake") }, }) @@ -692,15 +738,17 @@ describe("SessionRunnerLLM", () => { yield* events.publish(SessionEvent.ModelSwitched, { sessionID, + messageID: SessionMessage.ID.create(), timestamp: DateTime.makeUnsafe(1), model: { id: ModelV2.ID.make("replacement-1"), providerID: ProviderV2.ID.make("fake") }, }) yield* events.publish(SessionEvent.ModelSwitched, { sessionID, + messageID: SessionMessage.ID.create(), timestamp: DateTime.makeUnsafe(2), model: { id: ModelV2.ID.make("replacement-2"), providerID: ProviderV2.ID.make("fake") }, }) - const latest = yield* events.sequence(sessionID) + const latest = yield* SessionInput.latestSeq(db, sessionID) expect( yield* db @@ -713,6 +761,40 @@ describe("SessionRunnerLLM", () => { }), ) + it.effect("retries epoch preparation until observation-time invalidations settle", () => + 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 }) + response = [] + yield* session.resume(sessionID) + + requests.length = 0 + systemBaseline = "Changed context" + let invalidations = 0 + systemLoadHook = Effect.suspend(() => { + if (invalidations === 4) return Effect.void + invalidations++ + return events + .publish(SessionEvent.ModelSwitched, { + sessionID, + messageID: SessionMessage.ID.create(), + timestamp: DateTime.makeUnsafe(invalidations), + model: { id: ModelV2.ID.make(`replacement-${invalidations}`), providerID: ProviderV2.ID.make("fake") }, + }) + .pipe(Effect.asVoid) + }) + yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Second" }), resume: false }) + + yield* session.resume(sessionID) + + expect(invalidations).toBe(4) + expect(requests).toHaveLength(1) + expect(requests[0]?.system.map((part) => part.text)).toEqual(["Changed context"]) + }), + ) + it.effect("replays retained context projections while replacement is pending", () => Effect.gen(function* () { yield* setup @@ -728,6 +810,7 @@ describe("SessionRunnerLLM", () => { yield* session.resume(sessionID) yield* events.publish(SessionEvent.ModelSwitched, { sessionID, + messageID: SessionMessage.ID.create(), timestamp: DateTime.makeUnsafe(1), model: { id: ModelV2.ID.make("replacement"), providerID: ProviderV2.ID.make("fake") }, }) @@ -752,6 +835,7 @@ describe("SessionRunnerLLM", () => { yield* session.resume(sessionID) yield* events.publish(SessionEvent.ModelSwitched, { sessionID, + messageID: SessionMessage.ID.create(), timestamp: DateTime.makeUnsafe(1), model: { id: ModelV2.ID.make("replacement-1"), providerID: ProviderV2.ID.make("fake") }, }) @@ -760,6 +844,7 @@ describe("SessionRunnerLLM", () => { yield* session.resume(sessionID) yield* events.publish(SessionEvent.ModelSwitched, { sessionID, + messageID: SessionMessage.ID.create(), timestamp: DateTime.makeUnsafe(2), model: { id: ModelV2.ID.make("replacement-2"), providerID: ProviderV2.ID.make("fake") }, }) @@ -784,6 +869,7 @@ describe("SessionRunnerLLM", () => { yield* session.resume(sessionID) yield* events.publish(SessionEvent.Compaction.Started, { sessionID, + messageID: SessionMessage.ID.create(), timestamp: DateTime.makeUnsafe(1), reason: "manual", }) @@ -821,6 +907,7 @@ describe("SessionRunnerLLM", () => { yield* session.resume(sessionID) yield* events.publish(SessionEvent.Compaction.Started, { sessionID, + messageID: SessionMessage.ID.create(), timestamp: DateTime.makeUnsafe(1), reason: "manual", }) @@ -834,7 +921,16 @@ describe("SessionRunnerLLM", () => { yield* session.resume(sessionID) expect(requests.at(-1)?.system.map((part) => part.text)).toEqual(["Initial context"]) - expect(requests.at(-1)?.messages.some((message) => message.role === "system" && message.content[0]?.type === "text" && message.content[0].text === "Changed context")).toBe(true) + expect( + requests + .at(-1) + ?.messages.some( + (message) => + message.role === "system" && + message.content[0]?.type === "text" && + message.content[0].text === "Changed context", + ), + ).toBe(true) }), ) @@ -1022,6 +1118,7 @@ describe("SessionRunnerLLM", () => { yield* Deferred.await(toolExecutionsStarted) yield* events.publish(SessionEvent.ModelSwitched, { sessionID, + messageID: SessionMessage.ID.create(), timestamp: DateTime.makeUnsafe(1), model: { id: ModelV2.ID.make("replacement"), providerID: ProviderV2.ID.make("fake") }, }) diff --git a/packages/core/test/system-context-registry.test.ts b/packages/core/test/system-context-registry.test.ts index b1191cb065e..9625276f1ef 100644 --- a/packages/core/test/system-context-registry.test.ts +++ b/packages/core/test/system-context-registry.test.ts @@ -5,7 +5,7 @@ import { SystemContextRegistry } from "@opencode-ai/core/system-context-registry import { testEffect } from "./lib/effect" const contribution = (key: string, text: string, sourceKey = key) => ({ - key, + key: SystemContext.Key.make(key), load: Effect.succeed( SystemContext.make({ key: SystemContext.Key.make(sourceKey), @@ -43,7 +43,7 @@ describe("SystemContextRegistry", () => { const registry = yield* SystemContextRegistry.Service let loads = 0 yield* registry.contribute({ - key: "test/dynamic", + key: SystemContext.Key.make("test/dynamic"), load: Effect.sync(() => { loads++ return SystemContext.empty @@ -61,7 +61,7 @@ describe("SystemContextRegistry", () => { Effect.gen(function* () { const registry = yield* SystemContextRegistry.Service const failure = new Error("contribution failed") - yield* registry.contribute({ key: "test/failure", load: Effect.die(failure) }) + yield* registry.contribute({ key: SystemContext.Key.make("test/failure"), load: Effect.die(failure) }) const exit = yield* registry.load().pipe(Effect.exit) diff --git a/packages/core/test/system-context.test.ts b/packages/core/test/system-context.test.ts index 6aa7cb1470f..a20f8977b3e 100644 --- a/packages/core/test/system-context.test.ts +++ b/packages/core/test/system-context.test.ts @@ -125,16 +125,21 @@ describe("SystemContext", () => { expect(yield* SystemContext.reconcile(context, previous)).toEqual({ _tag: "Unchanged" }) expect(yield* SystemContext.replace(context, previous)).toEqual({ _tag: "ReplacementBlocked" }) - expect(yield* SystemContext.replace(context, {})).toMatchObject({ _tag: "Replaced" }) + expect(yield* SystemContext.replace(context, {})).toMatchObject({ _tag: "ReplacementReady" }) }), ) - it.effect("omits unavailable sources from an initial baseline", () => + it.effect("blocks initialization while a source is unavailable", () => Effect.gen(function* () { - expect(yield* SystemContext.initialize(stringContext({ key: "core/remote", value: SystemContext.unavailable }))).toEqual({ - baseline: "", - snapshot: {}, - }) + const exit = yield* SystemContext.initialize( + stringContext({ key: "core/remote", value: SystemContext.unavailable }), + ).pipe(Effect.exit) + + expect(Exit.isFailure(exit)).toBe(true) + if (Exit.isFailure(exit)) + expect(Cause.squash(exit.cause)).toEqual( + new SystemContext.InitializationBlocked({ keys: [key("core/remote")] }), + ) }), ) @@ -154,8 +159,10 @@ describe("SystemContext", () => { it.effect("requests replacement when a source without removal text disappears", () => Effect.gen(function* () { - expect(yield* SystemContext.reconcile(SystemContext.empty, { "core/date": { value: "2026-06-04" } })).toMatchObject({ - _tag: "Replaced", + expect( + yield* SystemContext.reconcile(SystemContext.empty, { "core/date": { value: "2026-06-04" } }), + ).toMatchObject({ + _tag: "ReplacementReady", }) }), ) @@ -188,7 +195,7 @@ describe("SystemContext", () => { yield* SystemContext.reconcile(stringContext({ key: "core/date", value: "2026-06-04" }), { "core/date": { value: 42, removed: "Date removed" }, }), - ).toMatchObject({ _tag: "Replaced" }) + ).toMatchObject({ _tag: "ReplacementReady" }) }), ) @@ -207,7 +214,7 @@ describe("SystemContext", () => { }) expect(yield* SystemContext.reconcile(context, { "core/date": { value: 42 } })).toMatchObject({ - _tag: "Replaced", + _tag: "ReplacementReady", generation: { baseline: "2026-06-04" }, }) expect(loads).toBe(1) @@ -234,7 +241,7 @@ describe("SystemContext", () => { "core/date": { value: "2026-06-03" }, "core/location": { value: 42 }, }), - ).toMatchObject({ _tag: "Replaced" }) + ).toMatchObject({ _tag: "ReplacementReady" }) expect(updates).toBe(0) }), ) diff --git a/packages/llm/test/fixtures/recordings/anthropic-messages/anthropic-haiku-4-5-chronological-system-update.json b/packages/llm/test/fixtures/recordings/anthropic-messages/anthropic-haiku-4-5-chronological-system-update.json index 2708b2ec6f8..72e4d7b21b5 100644 --- a/packages/llm/test/fixtures/recordings/anthropic-messages/anthropic-haiku-4-5-chronological-system-update.json +++ b/packages/llm/test/fixtures/recordings/anthropic-messages/anthropic-haiku-4-5-chronological-system-update.json @@ -7,13 +7,7 @@ "route": "anthropic-messages", "transport": "http", "model": "claude-haiku-4-5-20251001", - "tags": [ - "prefix:anthropic-messages", - "provider:anthropic", - "system", - "chronological-system-update", - "golden" - ] + "tags": ["prefix:anthropic-messages", "provider:anthropic", "system", "chronological-system-update", "golden"] }, "interactions": [ { diff --git a/packages/llm/test/fixtures/recordings/gemini/gemini-2-5-flash-chronological-system-update.json b/packages/llm/test/fixtures/recordings/gemini/gemini-2-5-flash-chronological-system-update.json index e7bb189294b..c4fd142fc77 100644 --- a/packages/llm/test/fixtures/recordings/gemini/gemini-2-5-flash-chronological-system-update.json +++ b/packages/llm/test/fixtures/recordings/gemini/gemini-2-5-flash-chronological-system-update.json @@ -7,13 +7,7 @@ "route": "gemini", "transport": "http", "model": "gemini-2.5-flash", - "tags": [ - "prefix:gemini", - "provider:google", - "system", - "chronological-system-update", - "golden" - ] + "tags": ["prefix:gemini", "provider:google", "system", "chronological-system-update", "golden"] }, "interactions": [ { diff --git a/packages/llm/test/fixtures/recordings/openai-chat/openai-chat-gpt-4o-mini-chronological-system-update.json b/packages/llm/test/fixtures/recordings/openai-chat/openai-chat-gpt-4o-mini-chronological-system-update.json index 6d002e4b94e..4c01f328d7f 100644 --- a/packages/llm/test/fixtures/recordings/openai-chat/openai-chat-gpt-4o-mini-chronological-system-update.json +++ b/packages/llm/test/fixtures/recordings/openai-chat/openai-chat-gpt-4o-mini-chronological-system-update.json @@ -7,13 +7,7 @@ "route": "openai-chat", "transport": "http", "model": "gpt-4o-mini", - "tags": [ - "prefix:openai-chat", - "provider:openai", - "system", - "chronological-system-update", - "golden" - ] + "tags": ["prefix:openai-chat", "provider:openai", "system", "chronological-system-update", "golden"] }, "interactions": [ { diff --git a/packages/opencode/src/cli/cmd/tui/context/sync-v2.tsx b/packages/opencode/src/cli/cmd/tui/context/sync-v2.tsx index 57df7c39358..31ff7d74780 100644 --- a/packages/opencode/src/cli/cmd/tui/context/sync-v2.tsx +++ b/packages/opencode/src/cli/cmd/tui/context/sync-v2.tsx @@ -177,7 +177,7 @@ export const { use: useSyncV2, provider: SyncProviderV2 } = createSimpleContext( break case "session.next.context.updated": update(event.properties.sessionID, (draft) => { - draft.unshift({ + prepend(draft, { id: event.properties.messageID, type: "system", text: event.properties.text, diff --git a/packages/opencode/test/cli/tui/sync-v2.test.tsx b/packages/opencode/test/cli/tui/sync-v2.test.tsx index 1b995d29431..a408e38e74a 100644 --- a/packages/opencode/test/cli/tui/sync-v2.test.tsx +++ b/packages/opencode/test/cli/tui/sync-v2.test.tsx @@ -261,6 +261,55 @@ test("sync v2 renders a promoted prompt when admission was missed", async () => } }) +test("sync v2 projects live context updates with their message ID", async () => { + const events = createEventSource() + const calls = createFetch() + let sync!: ReturnType + let ready!: () => void + const mounted = new Promise((resolve) => { + ready = resolve + }) + + function Probe() { + sync = useSyncV2() + onMount(ready) + return + } + + const app = await testRender(() => ( + + + + + + + + )) + + try { + await mounted + emitTwice(events, { + id: "evt_context_1", + type: "session.next.context.updated", + properties: { + sessionID: "session-1", + messageID: "msg_context_1", + timestamp: 1, + text: "Updated context", + }, + }) + + await wait(() => sync.session.message.fromSession("session-1").length === 1) + expect(sync.session.message.fromSession("session-1")[0]).toMatchObject({ + id: "msg_context_1", + type: "system", + text: "Updated context", + }) + } finally { + app.renderer.destroy() + } +}) + test("sync v2 preserves live events while snapshot hydration is in flight", async () => { const events = createEventSource() const response = Promise.withResolvers() @@ -309,6 +358,54 @@ test("sync v2 preserves live events while snapshot hydration is in flight", asyn } }) +test("sync v2 deduplicates a buffered context update already present in the hydrated snapshot", async () => { + const events = createEventSource() + const response = Promise.withResolvers() + const calls = createFetch((url) => { + if (url.pathname === "/api/session/session-1/message") return response.promise + return undefined + }) + let sync!: ReturnType + let ready!: () => void + const mounted = new Promise((resolve) => { + ready = resolve + }) + + function Probe() { + sync = useSyncV2() + onMount(ready) + return + } + + const app = await testRender(() => ( + + + + + + + + )) + + try { + await mounted + const hydration = sync.session.message.sync("session-1") + emitTwice(events, { + id: "evt_context_1", + type: "session.next.context.updated", + properties: { sessionID: "session-1", messageID: "msg_context_1", timestamp: 1, text: "Updated context" }, + }) + response.resolve( + json({ data: [{ id: "msg_context_1", type: "system", text: "Updated context", time: { created: 1 } }] }), + ) + await hydration + + expect(sync.session.message.fromSession("session-1")).toHaveLength(1) + } finally { + app.renderer.destroy() + } +}) + test("sync v2 replaces stale cached rows while preserving in-flight live rows", async () => { const events = createEventSource() const response = Promise.withResolvers() diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index f3a9cca90aa..69ab21185fd 100644 --- a/packages/sdk/js/src/v2/gen/types.gen.ts +++ b/packages/sdk/js/src/v2/gen/types.gen.ts @@ -21,6 +21,7 @@ export type Event = | EventSessionNextPrompted | EventSessionNextPromptAdmitted | EventSessionNextPromptPromoted + | EventSessionNextContextUpdated | EventSessionNextSynthetic | EventSessionNextShellStarted | EventSessionNextShellEnded @@ -867,6 +868,16 @@ export type GlobalEvent = { timeCreated: number } } + | { + id: string + type: "session.next.context.updated" + properties: { + timestamp: number + sessionID: string + messageID: string + text: string + } + } | { id: string type: "session.next.synthetic" @@ -1615,6 +1626,7 @@ export type GlobalEvent = { | SyncEventSessionNextPrompted | SyncEventSessionNextPromptAdmitted | SyncEventSessionNextPromptPromoted + | SyncEventSessionNextContextUpdated | SyncEventSessionNextSynthetic | SyncEventSessionNextShellStarted | SyncEventSessionNextShellEnded @@ -3259,6 +3271,23 @@ export type SyncEventSessionNextPromptPromoted = { } } +export type SyncEventSessionNextContextUpdated = { + type: "sync" + id: string + syncEvent: { + type: "session.next.context.updated.1" + id: string + seq: number + aggregateID: string + data: { + timestamp: number + sessionID: string + messageID: string + text: string + } + } +} + export type SyncEventSessionNextSynthetic = { type: "sync" id: string @@ -3822,6 +3851,18 @@ export type SessionMessageSynthetic = { type: "synthetic" } +export type SessionMessageSystem = { + id: string + metadata?: { + [key: string]: unknown + } + time: { + created: number + } + type: "system" + text: string +} + export type SessionMessageShell = { id: string metadata?: { @@ -3980,6 +4021,7 @@ export type SessionMessage = | SessionMessageModelSwitched | SessionMessageUser | SessionMessageSynthetic + | SessionMessageSystem | SessionMessageShell | SessionMessageAssistant | SessionMessageCompaction @@ -4339,6 +4381,17 @@ export type EventSessionNextPromptPromoted = { } } +export type EventSessionNextContextUpdated = { + id: string + type: "session.next.context.updated" + properties: { + timestamp: number + sessionID: string + messageID: string + text: string + } +} + export type EventSessionNextSynthetic = { id: string type: "session.next.synthetic" diff --git a/packages/sdk/openapi.json b/packages/sdk/openapi.json index 0a3d5fac381..6e350a71e24 100644 --- a/packages/sdk/openapi.json +++ b/packages/sdk/openapi.json @@ -12026,6 +12026,9 @@ { "$ref": "#/components/schemas/EventSessionNextPromptPromoted" }, + { + "$ref": "#/components/schemas/EventSessionNextContextUpdated" + }, { "$ref": "#/components/schemas/EventSessionNextSynthetic" }, @@ -14621,6 +14624,42 @@ "required": ["id", "type", "properties"], "additionalProperties": false }, + { + "type": "object", + "properties": { + "id": { + "type": "string", + "pattern": "^evt_" + }, + "type": { + "type": "string", + "enum": ["session.next.context.updated"] + }, + "properties": { + "type": "object", + "properties": { + "timestamp": { + "type": "number" + }, + "sessionID": { + "type": "string", + "pattern": "^ses" + }, + "messageID": { + "type": "string", + "pattern": "^msg_" + }, + "text": { + "type": "string" + } + }, + "required": ["timestamp", "sessionID", "messageID", "text"], + "additionalProperties": false + } + }, + "required": ["id", "type", "properties"], + "additionalProperties": false + }, { "type": "object", "properties": { @@ -17142,6 +17181,9 @@ { "$ref": "#/components/schemas/SyncEventSessionNextPromptPromoted" }, + { + "$ref": "#/components/schemas/SyncEventSessionNextContextUpdated" + }, { "$ref": "#/components/schemas/SyncEventSessionNextSynthetic" }, @@ -21870,6 +21912,63 @@ "required": ["type", "id", "syncEvent"], "additionalProperties": false }, + "SyncEventSessionNextContextUpdated": { + "type": "object", + "properties": { + "type": { + "type": "string", + "enum": ["sync"] + }, + "id": { + "type": "string", + "pattern": "^evt_" + }, + "syncEvent": { + "type": "object", + "properties": { + "type": { + "type": "string", + "enum": ["session.next.context.updated.1"] + }, + "id": { + "type": "string", + "pattern": "^evt_" + }, + "seq": { + "type": "number" + }, + "aggregateID": { + "type": "string" + }, + "data": { + "type": "object", + "properties": { + "timestamp": { + "type": "number" + }, + "sessionID": { + "type": "string", + "pattern": "^ses" + }, + "messageID": { + "type": "string", + "pattern": "^msg_" + }, + "text": { + "type": "string" + } + }, + "required": ["timestamp", "sessionID", "messageID", "text"], + "additionalProperties": false + } + }, + "required": ["type", "id", "seq", "aggregateID", "data"], + "additionalProperties": false + } + }, + "required": ["type", "id", "syncEvent"], + "additionalProperties": false + }, "SyncEventSessionNextSynthetic": { "type": "object", "properties": { @@ -23634,6 +23733,37 @@ "required": ["id", "time", "sessionID", "text", "type"], "additionalProperties": false }, + "SessionMessageSystem": { + "type": "object", + "properties": { + "id": { + "type": "string", + "pattern": "^msg_" + }, + "metadata": { + "type": "object" + }, + "time": { + "type": "object", + "properties": { + "created": { + "type": "number" + } + }, + "required": ["created"], + "additionalProperties": false + }, + "type": { + "type": "string", + "enum": ["system"] + }, + "text": { + "type": "string" + } + }, + "required": ["id", "time", "type", "text"], + "additionalProperties": false + }, "SessionMessageShell": { "type": "object", "properties": { @@ -24071,6 +24201,9 @@ { "$ref": "#/components/schemas/SessionMessageSynthetic" }, + { + "$ref": "#/components/schemas/SessionMessageSystem" + }, { "$ref": "#/components/schemas/SessionMessageShell" }, @@ -25174,6 +25307,42 @@ "required": ["id", "type", "properties"], "additionalProperties": false }, + "EventSessionNextContextUpdated": { + "type": "object", + "properties": { + "id": { + "type": "string", + "pattern": "^evt_" + }, + "type": { + "type": "string", + "enum": ["session.next.context.updated"] + }, + "properties": { + "type": "object", + "properties": { + "timestamp": { + "type": "number" + }, + "sessionID": { + "type": "string", + "pattern": "^ses" + }, + "messageID": { + "type": "string", + "pattern": "^msg_" + }, + "text": { + "type": "string" + } + }, + "required": ["timestamp", "sessionID", "messageID", "text"], + "additionalProperties": false + } + }, + "required": ["id", "type", "properties"], + "additionalProperties": false + }, "EventSessionNextSynthetic": { "type": "object", "properties": { diff --git a/specs/v2/refreshable-context-sources.md b/specs/v2/refreshable-context-sources.md index 29a6ce4579c..7e21a245f0c 100644 --- a/specs/v2/refreshable-context-sources.md +++ b/specs/v2/refreshable-context-sources.md @@ -21,15 +21,15 @@ Watcher-backed caches are a later efficiency optimization for roots with proven ## Existing Pieces -| Existing piece | Responsibility | -| ----------------------------------- | -------------------------------------------------------------------------------------------------------------------------------- | -| `Watcher.locationLayer` | Publish advisory `file.watcher.updated` events for local filesystem changes. | -| `EventV2.subscribe(...)` | Expose advisory events as scoped Effect streams. | -| `State.create(...)` | Rebuild replayable plugin and config contribution state from scoped transforms. | -| `SynchronizedRef.modifyEffect(...)` | Serialize effectful state refresh and store the next value only after success. | -| `SystemContext` | Convert coherent source samples into one immutable baseline, chronological updates, unavailable state, and removal tombstones. | -| `SystemContextRegistry` | Assemble Location-scoped built-in, instruction, and plugin context producers in stable contribution-key order. | -| `LocationServiceMap` | Own and clean up Location-scoped services, watcher subscriptions, and observation caches together. | +| Existing piece | Responsibility | +| ----------------------------------- | ------------------------------------------------------------------------------------------------------------------------------ | +| `Watcher.locationLayer` | Publish advisory `file.watcher.updated` events for local filesystem changes. | +| `EventV2.subscribe(...)` | Expose advisory events as scoped Effect streams. | +| `State.create(...)` | Rebuild replayable plugin and config contribution state from scoped transforms. | +| `SynchronizedRef.modifyEffect(...)` | Serialize effectful state refresh and store the next value only after success. | +| `SystemContext` | Convert coherent source samples into one immutable baseline, chronological updates, unavailable state, and removal tombstones. | +| `SystemContextRegistry` | Assemble Location-scoped built-in, instruction, and plugin context producers in stable contribution-key order. | +| `LocationServiceMap` | Own and clean up Location-scoped services, watcher subscriptions, and observation caches together. | The missing reusable piece is deliberately small: retain the last successful value, mark it stale, and serialize refresh attempts. @@ -153,10 +153,11 @@ embedded skill Add a Location-scoped contributor to `SystemContextRegistry`: ```ts -yield* registry.contribute({ - key: "core/instructions", - load: loadAmbientInstructions(), -}) +yield * + registry.contribute({ + key: SystemContext.Key.make("core/instructions"), + load: loadAmbientInstructions(), + }) ``` `InstructionContext` owns instruction discovery, deterministic ordering, and source loading. `SystemContextRegistry` owns contributor composition and lifecycle. `SystemContext` remains unaware of files and URLs. @@ -213,15 +214,15 @@ temporary discovery or read failure -> aggregate SystemContext.unavailable ``` -| Observation | Source outcome | -| -------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------- | -| Local scan succeeds and discovers readable file | Include its exact contents in the available aggregate source. | -| Local scan succeeds and a previously discovered file is absent | Remove it from the aggregate value; remove the aggregate source when no instructions remain. | -| Local scan or file read fails transiently | Preserve the admitted aggregate source as `SystemContext.unavailable`; never emit mass removals. | -| Empty local file | Include the empty exact content in the available aggregate source. | -| URL returns `2xx` body | Available source with exact contents. | -| URL times out or returns transient failure | `SystemContext.unavailable`. | -| URL returns `404` or `410` | Decide the explicit removal contract before URL implementation. | +| Observation | Source outcome | +| -------------------------------------------------------------- | ------------------------------------------------------------------------------------------------ | +| Local scan succeeds and discovers readable file | Include its exact contents in the available aggregate source. | +| Local scan succeeds and a previously discovered file is absent | Remove it from the aggregate value; remove the aggregate source when no instructions remain. | +| Local scan or file read fails transiently | Preserve the admitted aggregate source as `SystemContext.unavailable`; never emit mass removals. | +| Empty local file | Include the empty exact content in the available aggregate source. | +| URL returns `2xx` body | Available source with exact contents. | +| URL times out or returns transient failure | `SystemContext.unavailable`. | +| URL returns `404` or `410` | Decide the explicit removal contract before URL implementation. | Aggregate instruction removal text must be model-meaningful: @@ -309,6 +310,18 @@ sequenceDiagram If coverage is not proven, bypass the cache and observe directly whenever the safe boundary naturally requests current state. This is safe-turn refresh, not a background polling loop. +When coverage is proven, cache each known candidate instruction path independently rather than invalidating one aggregate instruction cache: + +```text +candidate instruction path +-> one Refreshable +-> watcher event invalidates only the matching path +-> next safe provider boundary reloads only stale candidates +-> available candidates become ordered per-file Context Sources +``` + +Ambient candidates include the global `AGENTS.md` path and one `AGENTS.md` candidate in every applicable ancestor directory, including candidates that are currently absent so later additions are observable. + ## URL Sources URLs never share an observation cache with local discovery. @@ -372,31 +385,36 @@ Nested instructions discovered after successful read-tool activity remain a Sess - Idle Sessions are not woken by local edits, URL timers, or plugin changes. - Context Epoch admission remains serialized by the Session event transaction at the next naturally scheduled provider turn. -## Proposed Implementation Order +## Implementation Status And Follow-Up Order + +Implemented in the direct-observation slice: + +1. Add the Location-scoped `SystemContextRegistry` backed by stable-keyed scoped contributions. +2. Register built-in and ambient instruction producers with `SystemContextRegistry`. +3. Observe local instructions directly at each safe provider boundary. +4. Preserve admitted instructions after transient scan/read failure and block initial provider turns while context is unavailable. +5. Test ordering, edit, unlink, empty file, transient scan failure, discovered-then-missing races, durable restart behavior, and deterministic context admission. + +Follow-up order: 1. Add and unit-test `Refreshable.make(load)` with `get` and `invalidate`. -2. Add model-meaningful instruction removal rendering support before unlink lands. -3. Add the Location-scoped `SystemContextRegistry` backed by stable-keyed scoped contributions. -4. Register built-in and ambient instruction producers with `SystemContextRegistry`. -5. Observe local instructions directly at each safe provider boundary. -6. Test add, edit, unlink, empty file, transient scan failure, transient read failure, restart, and deterministic ordering. -7. Add configured local exact paths and globs. -8. Add configured URL observations with explicit `404` and `410` semantics. -9. Add root-specific watcher registration and watcher-backed `Refreshable` invalidation where coverage is proven. -10. Migrate local `SkillV2` directory observations to per-source refreshables after skill failure semantics are corrected. -11. Add durable Session-scoped nested read discovery. +2. Add truthful root-specific watcher registration. +3. Move ambient instructions from one directly observed aggregate to one watcher-invalidated Refreshable and Context Source per candidate file. +4. Add configured local exact paths and globs. +5. Add configured URL observations with explicit `404` and `410` semantics. +6. Migrate local `SkillV2` directory observations to per-source refreshables after skill failure semantics are corrected. +7. Add durable Session-scoped nested read discovery. ## Open Questions -1. Should the first local scan failure preserve prior discovered sources as unavailable, or fail the current provider turn until a coherent rescan succeeds? -2. Should configured URL sources treat `404` and `410` as confirmed removals? -3. What root-specific watcher API cleanly models ignore policy and callback health? -4. Should own-process file mutations publish an advisory invalidation event synchronously after commit? +1. Should configured URL sources treat `404` and `410` as confirmed removals? +2. What root-specific watcher API cleanly models ignore policy and callback health? +3. Should own-process file mutations publish an advisory invalidation event synchronously after commit? ## Compression Line ```text -State remembers what should be loaded. +SystemContextRegistry remembers which context producers participate. Refreshable remembers whether a successful observation needs loading again. Context Epoch remembers what the model was told. ``` diff --git a/specs/v2/schema-changelog.md b/specs/v2/schema-changelog.md index e9a2bc7f19f..39b00e5eb8c 100644 --- a/specs/v2/schema-changelog.md +++ b/specs/v2/schema-changelog.md @@ -711,7 +711,7 @@ Compatibility: Affected schema: -- Add synchronized `session.next.context.updated.1` Session events containing only exact combined model-visible text. +- Add synchronized `session.next.context.updated.1` Session events containing a durable System-message ID and only exact combined model-visible text. - Add `session_context_epoch.revision` for transactional structured-snapshot advancement. - Add the first-class `system` Session message projection for chronological context updates. @@ -727,7 +727,7 @@ Compatibility: - The synchronized event log retains only text actually shown to the model, not internal structured snapshots. - 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. +- Replacement epochs after compaction or model switches, skills guidance, and plugin-defined context remain follow-up slices. ## 2026-06-04: Replace Session Context Epochs Lazily @@ -746,4 +746,22 @@ Compatibility: - Baseline replacement is bounded operational state and does not add permanent synchronized events. - 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. +- Compaction execution, skills guidance, and plugin-defined context remain follow-up slices. + +## 2026-06-05: Register Ambient System Context Producers + +Affected schema: + +- No database schema changes. + +Change: + +- Replace the Session-specific context loader with a Location-scoped registry of stable-keyed scoped context producers. +- Register environment/date and ambient instruction producers independently, then evaluate producers concurrently in stable contribution-key order. +- Directly discover and read global plus upward project `AGENTS.md` files at each safe provider-turn boundary. +- Preserve admitted instructions across transient scan/read failures and block first-epoch initialization while any context source is unavailable. +- Retry Context Epoch preparation until stable after optimistic revision mismatches. + +Compatibility: + +- Watcher-backed per-file `Refreshable` instruction observations, configured sources, nested discovery, and plugin-defined context remain follow-up slices.