diff --git a/AGENTS.md b/AGENTS.md index 6ed0761b897..8880407a78d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -133,6 +133,7 @@ const table = sqliteTable("session", { - Avoid mocks as much as possible - Test actual implementation, do not duplicate logic into tests +- In `packages/core/test`, define tests with the shared `it.effect` or `testEffect(...)` helpers from `test/lib/effect.ts`; do not use raw `test(...)`. Wrap synchronous assertions in `Effect.sync(...)`. - Tests cannot run from repo root (guard: `do-not-run-tests-from-root`); run from package dirs like `packages/opencode`. ## Type Checking diff --git a/CONTEXT.md b/CONTEXT.md index a391bf95cc9..2758a944cf0 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -8,12 +8,12 @@ OpenCode sessions preserve durable conversational history while assembling the r The structured collection of contextual facts presented to the model as initial instructions and chronological updates. _Avoid_: System prompt -**Context Component**: -One independently loaded fact within the **System Context**, represented by a stable key and one effectfully loaded baseline/update rendering. +**Context Source**: +One independently observed typed value within the **System Context**, represented by a stable key, JSON codec, infallible loader, pure baseline/update renderers, and an optional removal renderer for dynamic sources. _Avoid_: Prompt fragment **Mid-Conversation System Message**: -A durable chronological instruction that tells the model the newly effective state of a changed **Context Component**. +A durable chronological instruction that tells the model the newly effective state of a changed **Context Source**. _Avoid_: System update, system notification, raw text diff **Context Epoch**: @@ -23,44 +23,47 @@ The span during which one initially rendered **System Context** remains immutabl The full **System Context** rendered at the start of a **Context Epoch**. _Avoid_: Live system prompt -**Context Checkpoint**: -The durable model-hidden comparison state used to detect which **Context Components** changed since context was last admitted to a provider turn. +**Context Snapshot**: +The overwriteable model-hidden JSON state used to compare each **Context Source** with the value last admitted to a provider turn. **Unavailable Context**: -An expected temporary inability to load a **Context Component** value; the runtime retains its prior effective state and emits no update, or omits it until first successfully loaded. +An expected temporary inability to observe a **Context Source** value; the runtime retains its prior effective state and emits no update, or omits it until first successfully loaded. **Safe Provider-Turn Boundary**: The point immediately before a provider call, after durable input promotion and any required tool settlement, where context changes may be admitted chronologically. ## Relationships -- A **System Context** contains one or more **Context Components**. -- A changed **Context Component** may produce one **Mid-Conversation System Message** containing its newly effective state. -- A **Mid-Conversation System Message** persists its originating **Context Component** key and the exact rendered text sent to the model. -- A **Context Checkpoint** advances atomically with the corresponding durable **Mid-Conversation System Message**. -- A **Context Checkpoint** stores one rendered-content hash per stable **Context Component** key so core and plugin-defined components can evolve independently. -- Changes from multiple **Context Components** admitted at one safe boundary combine into one **Mid-Conversation System Message**. +- A **System Context** is an opaque carrier composed from zero or more **Context Sources**. +- A changed **Context Source** may produce one **Mid-Conversation System Message** containing its newly effective state. +- A **Mid-Conversation System Message** persists the exact combined rendered text sent to the model. +- The current **Context Snapshot** advances atomically with the corresponding durable **Mid-Conversation System Message**. +- A **Context Snapshot** stores one codec-encoded JSON value and, for removable dynamic sources, a pre-rendered removal message per stable **Context Source** key. +- 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 Checkpoint** without emitting a redundant **Mid-Conversation System Message**. -- Compaction starts a new **Context Epoch** with a freshly rendered **Baseline System Context** and **Context Checkpoint**; prior **Mid-Conversation System Messages** remain durable audit history but leave projected model history. -- A **Context Checkpoint** is an evolvable component map; a newly registered core or plugin-defined **Context Component** absent from an existing checkpoint emits its current state once at the next **Safe Provider-Turn Boundary**. -- **Context Component** keys are stable and namespaced; duplicate keys fail assembly. Built-in components preserve declaration order and plugin-defined components append in lexicographic key order so rendered context is deterministic. -- Each **Context Component** loader returns its model-visible baseline string and absolute current-state update string from one coherent sample; the update string is hashed for change detection. +- The first provider turn renders the latest **Baseline System Context** and initializes its **Context Snapshot** without emitting a redundant **Mid-Conversation System Message**. +- 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; future plugin-source assembly must append plugin-defined sources in lexicographic 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.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. - **Unavailable Context** uses stale-while-revalidate semantics and is distinct from a successfully loaded absence, which may emit removal text. -- Ordinary **Context Component** loaders return values directly; loaders that intentionally use stale-while-revalidate may explicitly return **Unavailable Context**. +- 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. - 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. -- Plugin-defined **Context Components** register through a scoped replayable registry so plugin hot reload adds and removes components predictably. +- Plugin-defined **Context Sources** register through a scoped replayable registry so plugin hot reload adds and removes sources predictably. - 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 model-projection history but are hidden from normal user-facing transcript surfaces. -- The date **Context Component** initially preserves host-local calendar-date behavior; a configured user timezone may replace that default later. +- **Mid-Conversation System Messages** remain durable Session-message history; normal user-facing transcript surfaces may hide them. +- The date **Context Source** initially preserves host-local calendar-date behavior; a configured user timezone may replace that default later. - A **Context Epoch** begins with one immutable **Baseline System Context**. - A **Baseline System Context** is stored durably and reused verbatim across process restarts within its **Context Epoch**. -- A **Baseline System Context** durably preserves deterministic keyed top-level component strings rather than eagerly joining all text; request assembly lowers them into canonical LLM system parts. +- A **Baseline System Context** durably preserves the exact joined text used for the active provider-cache prefix. - Compaction or a model/provider switch starts a new **Context Epoch** because the baseline can be replaced without preserving the prior provider cache. - A model/provider switch always starts a new **Context Epoch** while preserving chronological conversation history. - A **Mid-Conversation System Message** lowers to the provider's native chronological instruction role when supported and to a wrapped chronological fallback otherwise. @@ -73,5 +76,5 @@ The point immediately before a provider call, after durable input promotion and ## Flagged ambiguities -- Legacy `experimental.chat.system.transform` can mutate the assembled baseline system prompt arbitrarily, but V2 plugins do not yet expose an equivalent hook. Decide separately whether to port it, replace dynamic uses with plugin-defined **Context Components**, or narrow its semantics. +- Legacy `experimental.chat.system.transform` can mutate the assembled baseline system prompt arbitrarily, but V2 plugins do not yet expose an equivalent hook. Decide separately whether to port it, replace dynamic uses with plugin-defined **Context Sources**, or narrow its semantics. - A location change likely starts a new **Context Epoch** so location-dependent instructions and discovery can be rebuilt cleanly, but implementation should verify whether an append-only update is sufficient and meaningfully preserves cache. diff --git a/packages/core/migration/20260604184448_add_session_context_epoch/migration.sql b/packages/core/migration/20260604184448_add_session_context_epoch/migration.sql deleted file mode 100644 index a10df47539c..00000000000 --- a/packages/core/migration/20260604184448_add_session_context_epoch/migration.sql +++ /dev/null @@ -1,17 +0,0 @@ -CREATE TABLE `session_context_epoch` ( - `session_id` text PRIMARY KEY, - `baseline` text NOT NULL, - `checkpoint` text NOT NULL, - `baseline_seq` integer NOT NULL, - `replacement_seq` integer, - `revision` integer DEFAULT 0 NOT NULL, - CONSTRAINT `fk_session_context_epoch_session_id_session_id_fk` FOREIGN KEY (`session_id`) REFERENCES `session`(`id`) ON DELETE CASCADE -); ---> statement-breakpoint -CREATE TABLE `session_context_message` ( - `session_id` text NOT NULL, - `seq` integer NOT NULL, - `parts` text NOT NULL, - CONSTRAINT `session_context_message_pk` PRIMARY KEY(`session_id`, `seq`), - CONSTRAINT `fk_session_context_message_session_id_session_id_fk` FOREIGN KEY (`session_id`) REFERENCES `session`(`id`) ON DELETE CASCADE -); diff --git a/packages/core/migration/20260604234609_add_session_context_snapshot/migration.sql b/packages/core/migration/20260604234609_add_session_context_snapshot/migration.sql new file mode 100644 index 00000000000..ec98751154f --- /dev/null +++ b/packages/core/migration/20260604234609_add_session_context_snapshot/migration.sql @@ -0,0 +1,9 @@ +CREATE TABLE `session_context_epoch` ( + `session_id` text PRIMARY KEY, + `baseline` text NOT NULL, + `snapshot` text NOT NULL, + `baseline_seq` integer NOT NULL, + `replacement_seq` integer, + `revision` integer DEFAULT 0 NOT NULL, + CONSTRAINT `fk_session_context_epoch_session_id_session_id_fk` FOREIGN KEY (`session_id`) REFERENCES `session`(`id`) ON DELETE CASCADE +); diff --git a/packages/core/migration/20260604184448_add_session_context_epoch/snapshot.json b/packages/core/migration/20260604234609_add_session_context_snapshot/snapshot.json similarity index 96% rename from packages/core/migration/20260604184448_add_session_context_epoch/snapshot.json rename to packages/core/migration/20260604234609_add_session_context_snapshot/snapshot.json index 4715fa89fe0..22c68919c59 100644 --- a/packages/core/migration/20260604184448_add_session_context_epoch/snapshot.json +++ b/packages/core/migration/20260604234609_add_session_context_snapshot/snapshot.json @@ -1,7 +1,7 @@ { "version": "7", "dialect": "sqlite", - "id": "6fe30aa9-5772-49a6-b688-d022dbd28903", + "id": "65b52d0f-2bbe-483f-a7ec-9d2a7fa29f57", "prevIds": [ "fc92fa34-8074-44c3-88f0-a5417f7fd92d" ], @@ -58,10 +58,6 @@ "name": "session_context_epoch", "entityType": "tables" }, - { - "name": "session_context_message", - "entityType": "tables" - }, { "name": "session_input", "entityType": "tables" @@ -808,7 +804,7 @@ "autoincrement": false, "default": null, "generated": null, - "name": "checkpoint", + "name": "snapshot", "entityType": "columns", "table": "session_context_epoch" }, @@ -842,36 +838,6 @@ "entityType": "columns", "table": "session_context_epoch" }, - { - "type": "text", - "notNull": true, - "autoincrement": false, - "default": null, - "generated": null, - "name": "session_id", - "entityType": "columns", - "table": "session_context_message" - }, - { - "type": "integer", - "notNull": true, - "autoincrement": false, - "default": null, - "generated": null, - "name": "seq", - "entityType": "columns", - "table": "session_context_message" - }, - { - "type": "text", - "notNull": true, - "autoincrement": false, - "default": null, - "generated": null, - "name": "parts", - "entityType": "columns", - "table": "session_context_message" - }, { "type": "integer", "notNull": false, @@ -1552,21 +1518,6 @@ "entityType": "fks", "table": "session_context_epoch" }, - { - "columns": [ - "session_id" - ], - "tableTo": "session", - "columnsTo": [ - "id" - ], - "onUpdate": "NO ACTION", - "onDelete": "CASCADE", - "nameExplicit": false, - "name": "fk_session_context_message_session_id_session_id_fk", - "entityType": "fks", - "table": "session_context_message" - }, { "columns": [ "session_id" @@ -1662,16 +1613,6 @@ "entityType": "pks", "table": "project_directory" }, - { - "columns": [ - "session_id", - "seq" - ], - "nameExplicit": false, - "name": "session_context_message_pk", - "entityType": "pks", - "table": "session_context_message" - }, { "columns": [ "session_id", diff --git a/packages/core/src/database/migration.gen.ts b/packages/core/src/database/migration.gen.ts index e0d8d04bfec..949521287d3 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/20260604184448_add_session_context_epoch"), + import("./migration/20260604234609_add_session_context_snapshot"), ]) ).map((module) => module.default) satisfies DatabaseMigration.Migration[] diff --git a/packages/core/src/database/migration/20260604184448_add_session_context_epoch.ts b/packages/core/src/database/migration/20260604234609_add_session_context_snapshot.ts similarity index 55% rename from packages/core/src/database/migration/20260604184448_add_session_context_epoch.ts rename to packages/core/src/database/migration/20260604234609_add_session_context_snapshot.ts index 45ff9e53358..f035e53cbd4 100644 --- a/packages/core/src/database/migration/20260604184448_add_session_context_epoch.ts +++ b/packages/core/src/database/migration/20260604234609_add_session_context_snapshot.ts @@ -2,29 +2,20 @@ import { Effect } from "effect" import type { DatabaseMigration } from "../migration" export default { - id: "20260604184448_add_session_context_epoch", + id: "20260604234609_add_session_context_snapshot", up(tx) { return Effect.gen(function* () { yield* tx.run(` CREATE TABLE \`session_context_epoch\` ( \`session_id\` text PRIMARY KEY, \`baseline\` text NOT NULL, - \`checkpoint\` text NOT NULL, + \`snapshot\` text NOT NULL, \`baseline_seq\` integer NOT NULL, \`replacement_seq\` integer, \`revision\` integer DEFAULT 0 NOT NULL, CONSTRAINT \`fk_session_context_epoch_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE ); `) - yield* tx.run(` - CREATE TABLE \`session_context_message\` ( - \`session_id\` text NOT NULL, - \`seq\` integer NOT NULL, - \`parts\` text NOT NULL, - CONSTRAINT \`session_context_message_pk\` PRIMARY KEY(\`session_id\`, \`seq\`), - CONSTRAINT \`fk_session_context_message_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE - ); - `) }) }, } satisfies DatabaseMigration.Migration diff --git a/packages/core/src/event.ts b/packages/core/src/event.ts index 57dd587decd..8ae993967fe 100644 --- a/packages/core/src/event.ts +++ b/packages/core/src/event.ts @@ -45,6 +45,8 @@ export type Payload = { readonly version?: number readonly location?: Location.Ref readonly metadata?: Record + /** Internal replay marker for projectors that own non-replicated operational state. */ + readonly replay?: boolean } export type Projector = (event: Payload) => Effect.Effect @@ -137,6 +139,8 @@ export interface PublishOptions { readonly id?: ID readonly metadata?: Record readonly location?: Location.Ref + /** Local operational projection committed atomically with a new synchronized event. Not replayed or serialized. */ + readonly commit?: (seq: number) => Effect.Effect } export interface Interface { @@ -151,6 +155,7 @@ 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 @@ -215,6 +220,7 @@ export const layerWith = (options?: LayerOptions) => readonly ownerID?: string readonly strictOwner?: boolean }, + commit?: (seq: number) => Effect.Effect, ) { return Effect.gen(function* () { const definition = registry.get(event.type) @@ -330,6 +336,10 @@ export const layerWith = (options?: LayerOptions) => for (const projector of list) { 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 }]) @@ -375,11 +385,15 @@ export const layerWith = (options?: LayerOptions) => }) } - function publishEvent(event: Payload) { + function publishEvent(event: Payload, options?: PublishOptions) { return Effect.gen(function* () { 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" }), + ) if (durable) { - const committed = yield* commitSyncEvent(event as Payload) + const committed = yield* commitSyncEvent(event as Payload, undefined, options?.commit) if (committed) { event = { ...event, seq: committed.seq } yield* Effect.forEach(syncHandlers, (sync) => observe(event as Payload, "sync", sync), { discard: true }) @@ -431,7 +445,7 @@ export const layerWith = (options?: LayerOptions) => ...(definition.sync === undefined ? {} : { version: definition.sync.version }), ...(location ? { location } : {}), data, - } as Payload) + } as Payload, options) }) } @@ -451,6 +465,7 @@ export const layerWith = (options?: LayerOptions) => type: definition.type, version: definition.sync.version, data: definition.decode(event.data), + replay: true, } as Payload const committed = yield* commitSyncEvent(payload, { seq: event.seq, @@ -519,6 +534,14 @@ 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), @@ -646,6 +669,7 @@ export const layerWith = (options?: LayerOptions) => subscribe, all: streamAll, aggregateEvents: streamEvents, + sequence, sync, listen, beforeCommit, diff --git a/packages/core/src/session-system-context.ts b/packages/core/src/session-system-context.ts index 0ae80c66e3f..1b66da721e1 100644 --- a/packages/core/src/session-system-context.ts +++ b/packages/core/src/session-system-context.ts @@ -1,11 +1,11 @@ export * as SessionSystemContext from "./session-system-context" -import { Context, DateTime, Effect, Layer } from "effect" +import { Context, DateTime, Effect, Layer, Schema } from "effect" import { Location } from "./location" import { SystemContext } from "./system-context" export interface Interface { - readonly load: () => Effect.Effect + readonly load: () => Effect.Effect } export class Service extends Context.Service()("@opencode/v2/SessionSystemContext") {} @@ -22,32 +22,25 @@ export const layer = Layer.effect( ` Platform: ${process.platform}`, "", ].join("\n") - const context = SystemContext.struct({ - environment: SystemContext.value({ + const context = SystemContext.combine([ + SystemContext.make({ key: SystemContext.Key.make("core/environment"), - load: Effect.succeed({ - baseline: ["Here is some useful information about the environment you are running in:", environment].join( - "\n", - ), - update: ["The environment you are running in is now:", environment].join("\n"), - }), + codec: Schema.toCodecJson(Schema.String), + load: Effect.succeed(environment), + baseline: (environment) => + ["Here is some useful information about the environment you are running in:", environment].join("\n"), + update: (_previous, environment) => ["The environment you are running in is now:", environment].join("\n"), }), - date: SystemContext.value({ + SystemContext.make({ key: SystemContext.Key.make("core/date"), - load: DateTime.nowAsDate.pipe( - Effect.map((date) => ({ - baseline: `Today's date: ${date.toDateString()}`, - update: `Today's date is now: ${date.toDateString()}`, - })), - ), + codec: Schema.toCodecJson(Schema.String), + load: DateTime.nowAsDate.pipe(Effect.map((date) => date.toDateString())), + baseline: (date) => `Today's date: ${date}`, + update: (_previous, date) => `Today's date is now: ${date}`, }), - }) + ]) - return Service.of({ - load: Effect.fn("SessionSystemContext.load")(function* () { - return yield* SystemContext.load(context) - }), - }) + return Service.of({ load: () => Effect.succeed(context) }) }), ) diff --git a/packages/core/src/session/context-epoch.ts b/packages/core/src/session/context-epoch.ts index dec7e564d20..2b9fdfb9f3f 100644 --- a/packages/core/src/session/context-epoch.ts +++ b/packages/core/src/session/context-epoch.ts @@ -1,18 +1,17 @@ export * as SessionContextEpoch from "./context-epoch" -import { and, eq, isNull, lt, sql } from "drizzle-orm" +import { and, eq, isNull, lt, or, sql } from "drizzle-orm" import { DateTime, Effect, Schema } from "effect" import type { Database } from "../database/database" import { EventV2 } from "../event" import { SessionSystemContext } from "../session-system-context" import { SystemContext } from "../system-context" import { SessionEvent } from "./event" +import { SessionMessageID } from "./message-id" import { SessionSchema } from "./schema" -import { SessionContextEpochTable, SessionContextMessageTable } from "./sql" +import { SessionContextEpochTable } from "./sql" type DatabaseService = Database.Interface["db"] -const sameBaseline = Schema.toEquivalence(SystemContext.PartsSchema) -const sameCheckpoint = Schema.toEquivalence(SystemContext.CheckpointSchema) export const prepare = Effect.fn("SessionContextEpoch.prepare")(function* ( db: DatabaseService, @@ -20,48 +19,33 @@ export const prepare = Effect.fn("SessionContextEpoch.prepare")(function* ( context: SessionSystemContext.Interface, sessionID: SessionSchema.ID, ) { - const snapshot = yield* context.load() - const stored = yield* find(db, sessionID) + const [value, stored] = yield* Effect.all([context.load(), find(db, sessionID)], { concurrency: "unbounded" }) if (!stored) { - const initialized = SystemContext.initialize(snapshot) - const event = yield* events.publish(SessionEvent.ContextInitialized, { - sessionID, - timestamp: yield* DateTime.now, - baseline: initialized.baseline, - checkpoint: initialized.checkpoint, - }) - if (event.seq === undefined) return yield* Effect.die("Synchronized Session event is missing aggregate sequence") - return { baseline: initialized.baseline, baselineSeq: event.seq } - } - if (stored.replacement_seq !== null) { - if (SystemContext.replacementBlocked(snapshot, stored.checkpoint)) - return { baseline: stored.baseline, baselineSeq: stored.baseline_seq } - const initialized = SystemContext.initialize(snapshot) - const event = yield* events.publish(SessionEvent.ContextReplaced, { - sessionID, - timestamp: yield* DateTime.now, - expectedRevision: stored.revision, - baseline: initialized.baseline, - checkpoint: initialized.checkpoint, - }) - if (event.seq === undefined) return yield* Effect.die("Synchronized Session event is missing aggregate sequence") - return { baseline: initialized.baseline, baselineSeq: event.seq } + const generation = yield* SystemContext.initialize(value) + const baselineSeq = yield* initialize(db, events, sessionID, generation) + return { baseline: generation.baseline, baselineSeq } } - const refreshed = SystemContext.refresh(snapshot, stored.checkpoint) - if (sameCheckpoint(refreshed.checkpoint, stored.checkpoint)) + 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) + if (result._tag === "Unchanged" || result._tag === "ReplacementBlocked") return { baseline: stored.baseline, baselineSeq: stored.baseline_seq } - yield* events.publish(SessionEvent.ContextUpdated, { - sessionID, - timestamp: yield* DateTime.now, - expectedRevision: stored.revision, - parts: refreshed.changes, - checkpoint: refreshed.checkpoint, - }) + if (result._tag === "Replaced") { + const replacementSeq = stored.replacement_seq ?? (yield* events.sequence(sessionID)) + yield* replace(db, sessionID, stored.revision, replacementSeq, result.generation) + return { baseline: result.generation.baseline, baselineSeq: replacementSeq } + } + + yield* events.publish( + SessionEvent.ContextUpdated, + { sessionID, messageID: SessionMessageID.ID.create(), timestamp: yield* DateTime.now, text: result.text }, + { commit: () => advance(db, sessionID, stored.revision, result.snapshot).pipe(Effect.orDie) }, + ) return { baseline: stored.baseline, baselineSeq: stored.baseline_seq } }) -export const find = Effect.fn("SessionContextEpoch.find")(function* (db: DatabaseService, sessionID: SessionSchema.ID) { +const find = Effect.fn("SessionContextEpoch.find")(function* (db: DatabaseService, sessionID: SessionSchema.ID) { return yield* db .select() .from(SessionContextEpochTable) @@ -70,110 +54,6 @@ export const find = Effect.fn("SessionContextEpoch.find")(function* (db: Databas .pipe(Effect.orDie) }) -export const projectInitialized = Effect.fn("SessionContextEpoch.projectInitialized")(function* ( - db: DatabaseService, - event: SessionEvent.ContextInitialized, - seq: number, -) { - const stored = yield* find(db, event.data.sessionID) - if (stored) { - if (stored.baseline_seq > seq) return yield* Effect.void - if (stored.baseline_seq !== seq || !sameBaseline(stored.baseline, event.data.baseline)) - return yield* Effect.die("Session context epoch initialization conflicts with stored baseline") - return yield* Effect.void - } - return yield* db - .insert(SessionContextEpochTable) - .values({ - session_id: event.data.sessionID, - baseline: event.data.baseline, - checkpoint: event.data.checkpoint, - baseline_seq: seq, - replacement_seq: null, - revision: 0, - }) - .run() - .pipe(Effect.orDie) -}) - -export const projectUpdated = Effect.fn("SessionContextEpoch.projectUpdated")(function* ( - db: DatabaseService, - event: SessionEvent.ContextUpdated, - seq: number, -) { - const stored = yield* find(db, event.data.sessionID) - if (!stored) return yield* Effect.die("Session context epoch is not initialized") - if (stored.baseline_seq > seq) return yield* Effect.void - if (stored.replacement_seq !== null && seq >= stored.replacement_seq) - return yield* Effect.die("Session context epoch replacement is pending") - if (stored.revision > event.data.expectedRevision) { - if (event.data.parts.length === 0) return yield* Effect.void - const projected = yield* db - .select({ parts: SessionContextMessageTable.parts }) - .from(SessionContextMessageTable) - .where( - and(eq(SessionContextMessageTable.session_id, event.data.sessionID), eq(SessionContextMessageTable.seq, seq)), - ) - .get() - .pipe(Effect.orDie) - if (projected && sameBaseline(projected.parts, event.data.parts)) return yield* Effect.void - return yield* Effect.die("Session context update conflicts with stored projection") - } - const updated = yield* db - .update(SessionContextEpochTable) - .set({ checkpoint: event.data.checkpoint, revision: event.data.expectedRevision + 1 }) - .where( - and( - eq(SessionContextEpochTable.session_id, event.data.sessionID), - eq(SessionContextEpochTable.revision, event.data.expectedRevision), - ), - ) - .returning({ revision: SessionContextEpochTable.revision }) - .get() - .pipe(Effect.orDie) - if (!updated) return yield* Effect.die("Session context epoch revision mismatch") - if (event.data.parts.length === 0) return yield* Effect.void - return yield* db - .insert(SessionContextMessageTable) - .values({ session_id: event.data.sessionID, seq, parts: event.data.parts }) - .run() - .pipe(Effect.orDie) -}) - -export const projectReplaced = Effect.fn("SessionContextEpoch.projectReplaced")(function* ( - db: DatabaseService, - event: SessionEvent.ContextReplaced, - seq: number, -) { - const stored = yield* find(db, event.data.sessionID) - if (!stored) return yield* Effect.die("Session context epoch is not initialized") - if (stored.baseline_seq > seq) return yield* Effect.void - if (stored.baseline_seq === seq && sameBaseline(stored.baseline, event.data.baseline)) return yield* Effect.void - if (stored.replacement_seq === null) { - return yield* Effect.die("Session context epoch replacement was not requested") - } - const updated = yield* db - .update(SessionContextEpochTable) - .set({ - baseline: event.data.baseline, - checkpoint: event.data.checkpoint, - baseline_seq: seq, - replacement_seq: null, - revision: event.data.expectedRevision + 1, - }) - .where( - and( - eq(SessionContextEpochTable.session_id, event.data.sessionID), - eq(SessionContextEpochTable.revision, event.data.expectedRevision), - ), - ) - .returning({ revision: SessionContextEpochTable.revision }) - .get() - .pipe(Effect.orDie) - if (!updated) return yield* Effect.die("Session context epoch revision mismatch") - return yield* Effect.void -}) - export const requestReplacement = Effect.fn("SessionContextEpoch.requestReplacement")(function* ( db: DatabaseService, sessionID: SessionSchema.ID, @@ -185,10 +65,97 @@ export const requestReplacement = Effect.fn("SessionContextEpoch.requestReplacem .where( and( eq(SessionContextEpochTable.session_id, sessionID), - isNull(SessionContextEpochTable.replacement_seq), lt(SessionContextEpochTable.baseline_seq, seq), + or(isNull(SessionContextEpochTable.replacement_seq), lt(SessionContextEpochTable.replacement_seq, seq)), ), ) .run() .pipe(Effect.orDie) }) + +const initialize = Effect.fnUntraced(function* ( + db: DatabaseService, + events: EventV2.Interface, + sessionID: SessionSchema.ID, + generation: SystemContext.Generation, +) { + return yield* db + .transaction( + () => + Effect.gen(function* () { + const baselineSeq = yield* events.sequence(sessionID) + yield* db + .insert(SessionContextEpochTable) + .values({ + session_id: sessionID, + baseline: generation.baseline, + snapshot: generation.snapshot, + baseline_seq: baselineSeq, + revision: 0, + }) + .run() + .pipe(Effect.orDie) + return baselineSeq + }), + { behavior: "immediate" }, + ) + .pipe(Effect.orDie) +}) + +const replace = Effect.fnUntraced(function* ( + db: DatabaseService, + sessionID: SessionSchema.ID, + expectedRevision: number, + 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" }, + ) + .pipe(Effect.orDie) +}) + +const advance = Effect.fnUntraced(function* ( + db: DatabaseService, + sessionID: SessionSchema.ID, + expectedRevision: number, + snapshot: SystemContext.Snapshot, +) { + const updated = yield* db + .update(SessionContextEpochTable) + .set({ snapshot, revision: expectedRevision + 1 }) + .where( + and( + eq(SessionContextEpochTable.session_id, sessionID), + eq(SessionContextEpochTable.revision, expectedRevision), + isNull(SessionContextEpochTable.replacement_seq), + ), + ) + .returning({ revision: SessionContextEpochTable.revision }) + .get() + .pipe(Effect.orDie) + if (!updated) return yield* Effect.die("Session context epoch revision mismatch") +}) diff --git a/packages/core/src/session/context.ts b/packages/core/src/session/context.ts index d4fa8769bba..1b41c24fe42 100644 --- a/packages/core/src/session/context.ts +++ b/packages/core/src/session/context.ts @@ -1,22 +1,18 @@ -import { and, asc, desc, eq, gt, gte, or } from "drizzle-orm" +import { and, asc, desc, eq, gt, gte, ne, or } from "drizzle-orm" import { Effect, Schema } from "effect" import { Database } from "../database/database" import { MessageDecodeError } from "./error" import { SessionMessage } from "./message" import { SessionSchema } from "./schema" -import { SessionContextMessageTable, SessionMessageTable } from "./sql" -import type { SystemContext } from "../system-context" +import { SessionContextEpochTable, SessionMessageTable } from "./sql" type DatabaseService = Database.Interface["db"] const decode = Schema.decodeUnknownEffect(SessionMessage.Message) -export type RunnerMessage = - | SessionMessage.Message - | { readonly type: "system-context"; readonly parts: ReadonlyArray } const latestCompaction = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) { return yield* db - .select() + .select({ seq: SessionMessageTable.seq }) .from(SessionMessageTable) .where(and(eq(SessionMessageTable.session_id, sessionID), eq(SessionMessageTable.type, "compaction"))) .orderBy(desc(SessionMessageTable.seq)) @@ -28,7 +24,8 @@ const latestCompaction = Effect.fnUntraced(function* (db: DatabaseService, sessi const messageRows = Effect.fnUntraced(function* ( db: DatabaseService, sessionID: SessionSchema.ID, - compaction: typeof SessionMessageTable.$inferSelect | undefined, + compaction: { readonly seq: number } | undefined, + baselineSeq?: number, ) { return yield* db .select() @@ -36,7 +33,17 @@ const messageRows = Effect.fnUntraced(function* ( .where( and( eq(SessionMessageTable.session_id, sessionID), - compaction ? or(gte(SessionMessageTable.seq, compaction.seq)) : undefined, + compaction + ? or( + gte(SessionMessageTable.seq, compaction.seq), + baselineSeq === undefined + ? undefined + : and(eq(SessionMessageTable.type, "system"), gt(SessionMessageTable.seq, baselineSeq)), + ) + : undefined, + baselineSeq === undefined + ? undefined + : or(ne(SessionMessageTable.type, "system"), gt(SessionMessageTable.seq, baselineSeq)), ), ) .orderBy(asc(SessionMessageTable.seq)) @@ -56,8 +63,20 @@ const decodeMessageRow = (row: typeof SessionMessageTable.$inferSelect) => ) export const load = Effect.fn("SessionContext.load")(function* (db: DatabaseService, sessionID: SessionSchema.ID) { + const [epoch, compaction] = yield* Effect.all( + [ + db + .select({ baselineSeq: SessionContextEpochTable.baseline_seq }) + .from(SessionContextEpochTable) + .where(eq(SessionContextEpochTable.session_id, sessionID)) + .get() + .pipe(Effect.orDie), + latestCompaction(db, sessionID), + ], + { concurrency: "unbounded" }, + ) return yield* Effect.forEach( - yield* messageRows(db, sessionID, yield* latestCompaction(db, sessionID)), + yield* messageRows(db, sessionID, compaction, epoch?.baselineSeq), decodeMessageRow, ) }) @@ -67,44 +86,7 @@ export const loadForRunner = Effect.fn("SessionContext.loadForRunner")(function* sessionID: SessionSchema.ID, baselineSeq: number, ) { - const compaction = yield* latestCompaction(db, sessionID) - const messages = yield* messageRows(db, sessionID, compaction) - const updates = yield* db - .select() - .from(SessionContextMessageTable) - .where(and(eq(SessionContextMessageTable.session_id, sessionID), gt(SessionContextMessageTable.seq, baselineSeq))) - .orderBy(asc(SessionContextMessageTable.seq)) - .all() - .pipe(Effect.orDie) - return yield* Effect.forEach( - merge( - messages.map((row) => ({ type: "message" as const, seq: row.seq, row })), - updates.map((row) => ({ type: "system-context" as const, seq: row.seq, row })), - ), - (item): Effect.Effect => - item.type === "message" - ? decodeMessageRow(item.row) - : Effect.succeed({ type: "system-context", parts: item.row.parts }), - ) + return yield* Effect.forEach(yield* messageRows(db, sessionID, yield* latestCompaction(db, sessionID), baselineSeq), decodeMessageRow) }) -function merge( - left: ReadonlyArray, - right: ReadonlyArray, -): Array { - const result: Array = [] - let leftIndex = 0 - let rightIndex = 0 - while (leftIndex < left.length || rightIndex < right.length) { - if (rightIndex >= right.length || (leftIndex < left.length && left[leftIndex].seq < right[rightIndex].seq)) { - result.push(left[leftIndex]) - leftIndex++ - continue - } - result.push(right[rightIndex]) - rightIndex++ - } - return result -} - export * as SessionContext from "./context" diff --git a/packages/core/src/session/event.ts b/packages/core/src/session/event.ts index f471d2bf783..048ec6c6246 100644 --- a/packages/core/src/session/event.ts +++ b/packages/core/src/session/event.ts @@ -10,7 +10,6 @@ import { SessionSchema } from "./schema" import { Location } from "../location" import { RelativePath } from "../schema" import { SessionMessageID } from "./message-id" -import { SystemContext } from "../system-context" export { FileAttachment } @@ -120,41 +119,17 @@ export namespace PromptLifecycle { export type Promoted = typeof Promoted.Type } -export const ContextInitialized = EventV2.define({ - type: "session.next.context.initialized", - ...options, - schema: { - ...Base, - baseline: SystemContext.PartsSchema, - checkpoint: SystemContext.CheckpointSchema, - }, -}) -export type ContextInitialized = typeof ContextInitialized.Type - export const ContextUpdated = EventV2.define({ type: "session.next.context.updated", ...options, schema: { ...Base, - expectedRevision: NonNegativeInt, - parts: SystemContext.PartsSchema, - checkpoint: SystemContext.CheckpointSchema, + messageID: SessionMessageID.ID, + text: Schema.String, }, }) export type ContextUpdated = typeof ContextUpdated.Type -export const ContextReplaced = EventV2.define({ - type: "session.next.context.replaced", - ...options, - schema: { - ...Base, - expectedRevision: NonNegativeInt, - baseline: SystemContext.PartsSchema, - checkpoint: SystemContext.CheckpointSchema, - }, -}) -export type ContextReplaced = typeof ContextReplaced.Type - export const Synthetic = EventV2.define({ type: "session.next.synthetic", ...options, @@ -480,9 +455,7 @@ const DurableDefinitions = [ Prompted, PromptLifecycle.Admitted, PromptLifecycle.Promoted, - ContextInitialized, ContextUpdated, - ContextReplaced, Synthetic, Shell.Started, Shell.Ended, diff --git a/packages/core/src/session/message-updater.ts b/packages/core/src/session/message-updater.ts index 6e4c2280829..38f52c21c4c 100644 --- a/packages/core/src/session/message-updater.ts +++ b/packages/core/src/session/message-updater.ts @@ -159,9 +159,15 @@ export function update(adapter: Adapter, event: SessionEvent.Event) { }, "session.next.prompt.admitted": () => Effect.void, "session.next.prompt.promoted": () => Effect.void, - "session.next.context.initialized": () => Effect.void, - "session.next.context.updated": () => Effect.void, - "session.next.context.replaced": () => Effect.void, + "session.next.context.updated": (event) => + adapter.appendMessage( + new SessionMessage.System({ + id: event.data.messageID, + type: "system", + text: event.data.text, + time: { created: event.data.timestamp }, + }), + ), "session.next.synthetic": (event) => { return adapter.appendMessage( new SessionMessage.Synthetic({ diff --git a/packages/core/src/session/message.ts b/packages/core/src/session/message.ts index 7f62236befc..3a379e72d31 100644 --- a/packages/core/src/session/message.ts +++ b/packages/core/src/session/message.ts @@ -51,6 +51,12 @@ export class Synthetic extends Schema.Class("Session.Message.Syntheti type: Schema.Literal("synthetic"), }) {} +export class System extends Schema.Class("Session.Message.System")({ + ...Base, + type: Schema.Literal("system"), + text: SessionEvent.ContextUpdated.data.fields.text, +}) {} + export class Shell extends Schema.Class("Session.Message.Shell")({ ...Base, type: Schema.Literal("shell"), @@ -170,7 +176,7 @@ export class Compaction extends Schema.Class("Session.Message.Compac ...Base, }) {} -export const Message = Schema.Union([AgentSwitched, ModelSwitched, User, Synthetic, 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 ca997c689ba..51da5b245ec 100644 --- a/packages/core/src/session/projector.ts +++ b/packages/core/src/session/projector.ts @@ -420,17 +420,9 @@ export const layer = Layer.effectDiscard( ) }), ) - yield* events.project(SessionEvent.ContextInitialized, (event) => { - if (event.seq === undefined) return Effect.die("Synchronized Session event is missing aggregate sequence") - return SessionContextEpoch.projectInitialized(db, event, event.seq) - }) yield* events.project(SessionEvent.ContextUpdated, (event) => { - if (event.seq === undefined) return Effect.die("Synchronized Session event is missing aggregate sequence") - return SessionContextEpoch.projectUpdated(db, event, event.seq) - }) - yield* events.project(SessionEvent.ContextReplaced, (event) => { - if (event.seq === undefined) return Effect.die("Synchronized Session event is missing aggregate sequence") - return SessionContextEpoch.projectReplaced(db, event, event.seq) + 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))) }) 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/llm.ts b/packages/core/src/session/runner/llm.ts index 087fe0a3287..417bf8fd156 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -154,7 +154,7 @@ export const layer = Layer.effect( const context = yield* getRunnerContext(session.id, system.baselineSeq) const request = LLM.request({ model, - system: system.baseline.map((part) => SystemPart.make(part.text)), + system: system.baseline.length > 0 ? [SystemPart.make(system.baseline)] : [], messages: toLLMMessages(context, model), tools: yield* tools.definitions(), }) diff --git a/packages/core/src/session/runner/to-llm-message.ts b/packages/core/src/session/runner/to-llm-message.ts index 82020c85385..cca399790fd 100644 --- a/packages/core/src/session/runner/to-llm-message.ts +++ b/packages/core/src/session/runner/to-llm-message.ts @@ -9,7 +9,6 @@ import { } from "@opencode-ai/llm" import { SessionMessage } from "../message" import type { FileAttachment } from "../prompt" -import { SessionContext } from "../context" const media = (file: FileAttachment): ContentPart => ({ type: "media", @@ -92,7 +91,7 @@ const assistant = (message: SessionMessage.Assistant, model: Model) => { return [Message.make({ id: message.id, role: "assistant", content, metadata: message.metadata }), ...results] } -function toLLMMessage(message: SessionContext.RunnerMessage, model: Model): Message[] { +function toLLMMessage(message: SessionMessage.Message, model: Model): Message[] { switch (message.type) { case "agent-switched": case "model-switched": @@ -112,6 +111,8 @@ function toLLMMessage(message: SessionContext.RunnerMessage, model: Model): Mess ] case "synthetic": return [Message.make({ id: message.id, role: "user", content: message.text, metadata: message.metadata })] + case "system": + return [Message.system(message.text)] case "shell": return [ Message.make({ @@ -132,11 +133,9 @@ function toLLMMessage(message: SessionContext.RunnerMessage, model: Model): Mess metadata: message.metadata, }), ] - case "system-context": - return [Message.system(message.parts.map((part) => ({ type: "text", text: part.text })))] } } /** Translate projected V2 Session history into canonical @opencode-ai/llm context. */ -export const toLLMMessages = (messages: readonly SessionContext.RunnerMessage[], model: Model) => +export const toLLMMessages = (messages: readonly SessionMessage.Message[], model: Model) => messages.flatMap((message) => toLLMMessage(message, model)) diff --git a/packages/core/src/session/sql.ts b/packages/core/src/session/sql.ts index 5564610d188..7cd9fbf9af1 100644 --- a/packages/core/src/session/sql.ts +++ b/packages/core/src/session/sql.ts @@ -168,22 +168,9 @@ export const SessionContextEpochTable = sqliteTable("session_context_epoch", { .$type() .primaryKey() .references(() => SessionTable.id, { onDelete: "cascade" }), - baseline: text({ mode: "json" }).notNull().$type>(), - checkpoint: text({ mode: "json" }).notNull().$type(), + baseline: text().notNull(), + snapshot: text({ mode: "json" }).notNull().$type(), baseline_seq: integer().notNull(), replacement_seq: integer(), revision: integer().notNull().default(0), }) - -export const SessionContextMessageTable = sqliteTable( - "session_context_message", - { - session_id: text() - .$type() - .notNull() - .references(() => SessionTable.id, { onDelete: "cascade" }), - seq: integer().notNull(), - parts: text({ mode: "json" }).notNull().$type>(), - }, - (table) => [primaryKey({ columns: [table.session_id, table.seq] })], -) diff --git a/packages/core/src/session/store.ts b/packages/core/src/session/store.ts index ad60c56d416..6eb7052539c 100644 --- a/packages/core/src/session/store.ts +++ b/packages/core/src/session/store.ts @@ -16,7 +16,7 @@ export interface Interface { readonly runnerContext: ( sessionID: SessionSchema.ID, baselineSeq: number, - ) => Effect.Effect + ) => Effect.Effect readonly message: ( messageID: SessionMessage.ID, ) => Effect.Effect<{ readonly sessionID: SessionSchema.ID; readonly message: SessionMessage.Message } | undefined> diff --git a/packages/core/src/system-context.ts b/packages/core/src/system-context.ts index 67aa6ad75d0..b69053d9bd5 100644 --- a/packages/core/src/system-context.ts +++ b/packages/core/src/system-context.ts @@ -1,73 +1,84 @@ export * as SystemContext from "./system-context" -import { Effect, Schema } from "effect" -import { Hash } from "./util/hash" +import { Effect, Option, Schema } from "effect" +/** + * Models privileged system context as independently refreshable typed sources. + * + * `Source` describes how to observe, compare, and render one value. `make` + * closes over `A`, producing an opaque `SystemContext` that composes uniformly + * with contexts built from other value types. Interpreters observe the composed + * context once, then produce a durable structured + * `Snapshot` alongside the exact model-visible baseline or update text. + * + * Returning `unavailable` means observation failed temporarily. It differs from + * removing a source from the context: refresh preserves the admitted snapshot, + * and replacement waits rather than silently constructing an incomplete baseline. + * + * @module + */ + +/** Stable namespaced identity for one independently refreshable context source. */ export const Key = Schema.String.check(Schema.isPattern(/^[a-z0-9][a-z0-9._-]*\/[a-z0-9][a-z0-9._/-]*$/)).pipe( Schema.brand("SystemContext.Key"), ) export type Key = typeof Key.Type +/** Indicates that a source could not be observed without treating it as removed. */ export const unavailable = Symbol.for("@opencode/SystemContext.Unavailable") export type Unavailable = typeof unavailable -export interface Value { - /** Full component text rendered into a new epoch baseline. */ - readonly baseline: string - /** Absolute current-state text emitted when this component changes. */ - readonly update: string -} - -export interface Component { +/** Defines one typed source before its value type is hidden by `make`. */ +export interface Source { readonly key: Key - readonly load: Effect.Effect + readonly codec: Schema.Codec + readonly load: Effect.Effect + readonly baseline: (current: A) => string + readonly update: (previous: A, current: A) => string + readonly removed?: (previous: A) => string } -export interface SystemContext { - readonly components: ReadonlyArray> +const ContextTypeId: unique symbol = Symbol.for("@opencode/SystemContext") + +/** Opaque carrier for composable system context sources. */ +export interface SystemContext { + readonly [ContextTypeId]: ReadonlyArray } -export interface AvailableEntry extends Value { - readonly _tag: "Available" - readonly key: Key - readonly hash: string -} - -export interface UnavailableEntry { - readonly _tag: "Unavailable" - readonly key: Key -} - -export type Entry = AvailableEntry | UnavailableEntry - -export interface Snapshot { - readonly entries: ReadonlyArray -} - -export interface Part { - readonly key: Key - readonly text: string -} - -export const PartSchema = Schema.Struct({ - key: Key, - text: Schema.String, +/** Durable comparison state for one admitted source. */ +export const SourceSnapshot = Schema.Struct({ + value: Schema.Json, + removed: Schema.optional(Schema.NonEmptyString), }) -export const PartsSchema = Schema.Array(PartSchema) -export const CheckpointSchema = Schema.Record(Key, Schema.String) +export type SourceSnapshot = typeof SourceSnapshot.Type -export type Checkpoint = Readonly> +/** Durable structured comparison state for one active context generation. */ +export const Snapshot = Schema.Record(Key, SourceSnapshot) +export type Snapshot = Readonly> -export interface Initialized { - readonly baseline: ReadonlyArray - readonly checkpoint: Checkpoint +export interface Generation { + readonly baseline: string + readonly snapshot: Snapshot } -export interface Refreshed { - readonly changes: ReadonlyArray - readonly checkpoint: Checkpoint +export interface Updated { + readonly _tag: "Updated" + readonly text: string + readonly snapshot: Snapshot } +export interface Replaced { + readonly _tag: "Replaced" + readonly generation: Generation +} + +export interface ReplacementBlocked { + readonly _tag: "ReplacementBlocked" +} + +export type ReplacementResult = Replaced | ReplacementBlocked +export type ReconcileResult = { readonly _tag: "Unchanged" } | Updated | ReplacementResult + export class DuplicateKeyError extends Schema.TaggedErrorClass()("SystemContext.DuplicateKeyError", { key: Key, }) { @@ -76,78 +87,219 @@ export class DuplicateKeyError extends Schema.TaggedErrorClass(component: Component): Component => component - -export function struct(components: Readonly>>): SystemContext { - const values = Object.values(components) - assertUniqueKeys(values) - return { components: values } +interface PackedSource { + readonly key: Key + readonly load: Effect.Effect } -export const load = (context: SystemContext) => - Effect.sync(() => assertUniqueKeys(context.components)).pipe( - Effect.andThen( - Effect.forEach(context.components, (component) => - component.load.pipe( - Effect.map( - (result): Entry => - result === unavailable - ? { _tag: "Unavailable", key: component.key } - : { _tag: "Available", key: component.key, ...result, hash: Hash.sha256(result.update) }, - ), +interface Loaded { + readonly baseline: () => Rendered + readonly compare: (previous: Schema.Json) => Compared +} + +interface Rendered { + readonly text: string + readonly snapshot: SourceSnapshot +} + +type Compared = + | { readonly _tag: "Incompatible" } + | { readonly _tag: "Unchanged" } + | { readonly _tag: "Updated"; readonly render: () => Rendered } + +interface AvailableEntry extends Loaded { + readonly _tag: "Available" + readonly key: Key +} + +interface UnavailableEntry { + readonly _tag: "Unavailable" + readonly key: Key +} + +type Entry = AvailableEntry | UnavailableEntry + +/** The identity context. */ +export const empty = context([]) + +/** Closes a typed source into a context that composes with differently typed sources. */ +export function make(source: Source): SystemContext { + const decode = Schema.decodeUnknownOption(source.codec) + const encode = Schema.encodeSync(source.codec) + const equivalent = Schema.toEquivalence(source.codec) + return context([ + { + key: source.key, + load: source.load.pipe( + Effect.map((value) => { + if (isUnavailable(value)) return value + const snapshot = (): SourceSnapshot => ({ + value: encode(value), + ...(source.removed ? { removed: requireText(source.key, "removal", source.removed(value)) } : {}), + }) + return { + baseline: (): Rendered => ({ + text: requireText(source.key, "baseline", source.baseline(value)), + snapshot: snapshot(), + }), + compare: (previous): Compared => + Option.match(decode(previous), { + onNone: (): Compared => ({ _tag: "Incompatible" }), + onSome: (decoded): Compared => + equivalent(decoded, value) + ? { _tag: "Unchanged" } + : { + _tag: "Updated", + render: () => ({ + text: requireText(source.key, "update", source.update(decoded, value)), + snapshot: snapshot(), + }), + }, + }), + } + }), + ), + }, + ]) +} + +/** Combines contexts in order and rejects duplicate source keys immediately. */ +export function combine(values: ReadonlyArray): SystemContext { + const sources = values.flatMap((value) => value[ContextTypeId]) + assertUniqueKeys(sources) + return context(sources) +} + +const observe = (value: SystemContext) => + Effect.forEach( + value[ContextTypeId], + (source) => + source.load.pipe( + Effect.map( + (result): Entry => + result === unavailable + ? { _tag: "Unavailable", key: source.key } + : { _tag: "Available", key: source.key, ...result }, ), ), - ), - Effect.map((entries): Snapshot => ({ entries })), + { concurrency: "unbounded" }, ) -export function initialize(snapshot: Snapshot): Initialized { +/** 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)) +} + +function initializeObservation(entries: ReadonlyArray): Generation { + const available = entries.filter((entry): entry is AvailableEntry => entry._tag === "Available") + const rendered = available.map((entry) => [entry.key, entry.baseline()] as const) return { - baseline: snapshot.entries.flatMap((entry) => - entry._tag === "Available" ? [{ key: entry.key, text: entry.baseline }] : [], - ), - checkpoint: nextCheckpoint(snapshot, {}), + baseline: render(rendered.map(([, result]) => result.text)), + snapshot: Object.fromEntries(rendered.map(([key, result]) => [key, result.snapshot])), } } -export function refresh(snapshot: Snapshot, previous: Checkpoint): Refreshed { - const keys = new Set(snapshot.entries.map((entry) => entry.key)) - return { - changes: [ - ...snapshot.entries.flatMap((entry) => - entry._tag === "Available" && getCheckpoint(previous, entry.key) !== entry.hash - ? [{ key: entry.key, text: entry.update }] - : [], - ), - ...Object.keys(previous).flatMap((key) => - keys.has(Key.make(key)) ? [] : [{ key: Key.make(key), text: `System context component removed: ${key}` }], - ), - ], - checkpoint: nextCheckpoint(snapshot, previous), - } -} - -export const replacementBlocked = (snapshot: Snapshot, previous: Checkpoint) => - snapshot.entries.some((entry) => entry._tag === "Unavailable" && getCheckpoint(previous, entry.key) !== undefined) - -function nextCheckpoint(snapshot: Snapshot, previous: Checkpoint) { - return Object.fromEntries( - snapshot.entries.flatMap((entry) => { - if (entry._tag === "Available") return [[entry.key, entry.hash]] - const hash = getCheckpoint(previous, entry.key) - return hash === undefined ? [] : [[entry.key, hash]] +/** Reconciles current source values with one active generation. */ +export function reconcile(value: SystemContext, previous: Snapshot): Effect.Effect { + return observe(value).pipe( + Effect.map((entries): ReconcileResult => { + const result = reconcileObservation(entries, previous) + if (result._tag === "Unchanged" || result._tag === "Updated") return result + return replaceObservation(entries, previous) }), ) } -function getCheckpoint(checkpoint: Checkpoint, key: Key) { - return Object.hasOwn(checkpoint, key) ? checkpoint[key] : undefined +function reconcileObservation( + entries: ReadonlyArray, + previous: Snapshot, +): { readonly _tag: "Unchanged" } | Updated | { readonly _tag: "Replace" } { + const keys = new Set(entries.map((entry) => entry.key)) + const comparisons = new Map() + for (const entry of entries) { + if (entry._tag === "Unavailable") continue + const stored = getSnapshot(previous, entry.key) + if (!stored) continue + const compared = entry.compare(stored.value) + if (compared._tag === "Incompatible") return { _tag: "Replace" } + comparisons.set(entry.key, compared) + } + for (const key of Object.keys(previous).sort()) { + if (keys.has(Key.make(key))) continue + if (previous[key].removed === undefined) return { _tag: "Replace" } + } + + const snapshot: Record = {} + const updates: string[] = [] + for (const entry of entries) { + const stored = getSnapshot(previous, entry.key) + if (entry._tag === "Unavailable") { + if (stored) snapshot[entry.key] = stored + continue + } + if (!stored) { + const rendered = entry.baseline() + updates.push(rendered.text) + snapshot[entry.key] = rendered.snapshot + continue + } + const compared = comparisons.get(entry.key) + if (!compared || compared._tag === "Incompatible") + throw new Error(`Missing comparison for system context source ${entry.key}`) + if (compared._tag === "Unchanged") { + snapshot[entry.key] = stored + continue + } + const rendered = compared.render() + updates.push(rendered.text) + snapshot[entry.key] = rendered.snapshot + } + for (const key of Object.keys(previous).sort()) { + if (keys.has(Key.make(key))) continue + const removed = previous[key].removed + if (removed === undefined) throw new Error(`Missing removal rendering for system context source ${key}`) + updates.push(removed) + } + if (updates.length === 0) return { _tag: "Unchanged" } + return { _tag: "Updated", text: render(updates), snapshot } } -function assertUniqueKeys(components: ReadonlyArray>) { +/** Creates a complete replacement generation or blocks while admitted context is unavailable. */ +export function replace(value: SystemContext, previous: Snapshot): Effect.Effect { + return observe(value).pipe(Effect.map((entries) => replaceObservation(entries, previous))) +} + +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) } +} + +function context(sources: ReadonlyArray): SystemContext { + return { [ContextTypeId]: sources } +} + +function render(parts: ReadonlyArray) { + return parts.join("\n\n") +} + +function getSnapshot(snapshot: Snapshot, key: Key) { + return Object.hasOwn(snapshot, key) ? snapshot[key] : undefined +} + +function isUnavailable(value: unknown): value is Unavailable { + return value === unavailable +} + +function requireText(key: Key, kind: string, text: string) { + if (text.length === 0) throw new Error(`System context source ${key} rendered an empty ${kind}`) + return text +} + +function assertUniqueKeys(sources: ReadonlyArray) { const keys = new Set() - for (const component of components) { - if (keys.has(component.key)) throw new DuplicateKeyError({ key: component.key }) - keys.add(component.key) + for (const source of sources) { + if (keys.has(source.key)) throw new DuplicateKeyError({ key: source.key }) + keys.add(source.key) } } diff --git a/packages/core/test/event.test.ts b/packages/core/test/event.test.ts index c91d34fcfa3..9b454333b4b 100644 --- a/packages/core/test/event.test.ts +++ b/packages/core/test/event.test.ts @@ -190,6 +190,53 @@ describe("EventV2", () => { }), ) + it.effect("commits local operational state inside a new synchronized event transaction", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const received = new Array() + yield* events.project(SyncMessage, () => Effect.sync(() => received.push("projector"))) + + yield* events.publish( + SyncMessage, + { id: "one", text: "hello" }, + { commit: (seq) => Effect.sync(() => received.push(`commit:${seq}`)) }, + ) + + expect(received).toEqual(["projector", "commit:0"]) + }), + ) + + it.effect("rolls back the synchronized event and projector when the local commit fails", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const { db } = yield* Database.Service + const aggregateID = EventV2.ID.create() + yield* db.run("CREATE TABLE IF NOT EXISTS event_commit_probe (value text NOT NULL)") + yield* db.run("DELETE FROM event_commit_probe") + yield* events.project(SyncMessage, () => + db.run("INSERT INTO event_commit_probe (value) VALUES ('projected')").pipe(Effect.orDie, Effect.asVoid), + ) + + const exit = yield* events + .publish(SyncMessage, { id: aggregateID, text: "hello" }, { commit: () => Effect.die("commit failed") }) + .pipe(Effect.exit) + + 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([]) + }), + ) + + it.effect("rejects local commit hooks on live-only events", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const exit = yield* events.publish(Message, { text: "hello" }, { commit: () => Effect.void }).pipe(Effect.exit) + + expect(String(exit)).toContain("Local commit hooks require a synchronized event") + }), + ) + it.effect("runs projectors before publishing to streams", () => Effect.gen(function* () { const events = yield* EventV2.Service diff --git a/packages/core/test/session-runner-message.test.ts b/packages/core/test/session-runner-message.test.ts index e00346cd1ee..854f24158f4 100644 --- a/packages/core/test/session-runner-message.test.ts +++ b/packages/core/test/session-runner-message.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, test } from "bun:test" +import { describe, expect } from "bun:test" import { Message, Model } from "@opencode-ai/llm" import * as OpenAIChat from "@opencode-ai/llm/protocols/openai-chat" import { ModelV2 } from "@opencode-ai/core/model" @@ -8,15 +8,15 @@ import { AgentAttachment, FileAttachment, ReferenceAttachment } from "@opencode- import { toLLMMessages } from "@opencode-ai/core/session/runner/to-llm-message" import { SessionV2 } from "@opencode-ai/core/session" import { ToolOutput } from "@opencode-ai/core/tool-output" -import { DateTime } from "effect" -import { SystemContext } from "@opencode-ai/core/system-context" +import { DateTime, Effect } from "effect" +import { it } from "./lib/effect" const created = DateTime.makeUnsafe(0) const id = (value: string) => SessionMessage.ID.make(`msg_${value}`) const model = Model.make({ id: "model", provider: "provider", route: OpenAIChat.route }) describe("toLLMMessages", () => { - test("maps every top-level V2 Session message type", () => { + 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( @@ -85,31 +85,22 @@ describe("toLLMMessages", () => { [{ type: "text", text: "Shell command: pwd\n\n/project" }], [{ type: "text", text: "Summary of earlier conversation:\nEarlier work" }], ]) - }) + })) - test("maps hidden Session context updates into chronological system messages", () => { + it.effect("maps durable Session system messages into chronological system messages", () => Effect.sync(() => { expect( toLLMMessages( [ - { - type: "system-context", - parts: [ - { key: SystemContext.Key.make("test/context"), text: "Updated context" }, - { key: SystemContext.Key.make("test/other"), text: "Other context" }, - ], - }, + new SessionMessage.System({ id: id("system"), type: "system", text: "Updated context\n\nOther context", time: { created } }), ], model, ), ).toEqual([ - Message.system([ - { type: "text", text: "Updated context" }, - { type: "text", text: "Other context" }, - ]), + Message.system("Updated context\n\nOther context"), ]) - }) + })) - test("expands assistant tool calls and settled outcomes into canonical tool messages", () => { + it.effect("expands assistant tool calls and settled outcomes into canonical tool messages", () => Effect.sync(() => { const messages = toLLMMessages( [ new SessionMessage.Assistant({ @@ -264,9 +255,9 @@ describe("toLLMMessages", () => { }, }, ]) - }) + })) - test("restores OpenAI encrypted reasoning metadata", () => { + it.effect("restores OpenAI encrypted reasoning metadata", () => Effect.sync(() => { const messages = toLLMMessages( [ new SessionMessage.Assistant({ @@ -295,9 +286,9 @@ describe("toLLMMessages", () => { providerMetadata: { openai: { itemId: "rs_1", reasoningEncryptedContent: "encrypted-state" } }, }, ]) - }) + })) - test("drops provider-native continuation metadata after a model switch", () => { + it.effect("drops provider-native continuation metadata after a model switch", () => Effect.sync(() => { const messages = toLLMMessages( [ new SessionMessage.Assistant({ @@ -395,5 +386,5 @@ describe("toLLMMessages", () => { providerMetadata: undefined, }, ]) - }) + })) }) diff --git a/packages/core/test/session-runner-recorded.test.ts b/packages/core/test/session-runner-recorded.test.ts index edeb04ff6ba..c9252b86bba 100644 --- a/packages/core/test/session-runner-recorded.test.ts +++ b/packages/core/test/session-runner-recorded.test.ts @@ -21,10 +21,9 @@ import { SessionTable } from "@opencode-ai/core/session/sql" import { SessionStore } from "@opencode-ai/core/session/store" import { SessionSystemContext } from "@opencode-ai/core/session-system-context" import { SystemContext } from "@opencode-ai/core/system-context" -import { Hash } from "@opencode-ai/core/util/hash" import { describe, expect } from "bun:test" import { eq } from "drizzle-orm" -import { Effect, Layer } from "effect" +import { Effect, Layer, Schema } from "effect" import path from "node:path" import { testEffect } from "./lib/effect" @@ -62,17 +61,16 @@ const systemContext = Layer.succeed( SessionSystemContext.Service, SessionSystemContext.Service.of({ load: () => - Effect.succeed({ - entries: [ - { - _tag: "Available" as const, - key: SystemContext.Key.make("test/context"), - baseline: "Recorded context", - update: "Recorded context", - hash: Hash.sha256("Recorded context"), - }, - ], - }), + Effect.succeed( + SystemContext.make({ + key: SystemContext.Key.make("test/context"), + codec: Schema.toCodecJson(Schema.String), + load: Effect.succeed("Recorded context"), + baseline: String, + update: (_previous, current) => current, + removed: () => "Recorded context removed", + }), + ), }), ) const runner = SessionRunnerLLM.defaultLayer.pipe( @@ -167,7 +165,6 @@ describe("SessionRunnerLLM recorded", () => { ).toEqual([ "session.next.prompt.admitted.1", "session.next.prompt.promoted.1", - "session.next.context.initialized.1", "session.next.step.started.1", "session.next.text.started.1", "session.next.text.ended.1", diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index 7b31b913580..a960061f455 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -32,11 +32,10 @@ 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 { 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 { SessionSystemContext } from "@opencode-ai/core/session-system-context" import { SystemContext } from "@opencode-ai/core/system-context" -import { Hash } from "@opencode-ai/core/util/hash" 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" @@ -150,21 +149,22 @@ const systemContext = Layer.succeed( SessionSystemContext.Service, SessionSystemContext.Service.of({ load: () => - Effect.sync(() => ({ - entries: systemRemoved - ? [] - : systemUnavailable - ? [{ _tag: "Unavailable" as const, key: systemContextKey }] + Effect.succeed( + SystemContext.combine( + systemRemoved + ? [] : [ - { - _tag: "Available" as const, + SystemContext.make({ key: systemContextKey, - baseline: systemBaseline, - update: systemBaseline, - hash: Hash.sha256(systemBaseline), - }, + codec: Schema.toCodecJson(Schema.String), + load: Effect.sync(() => (systemUnavailable ? SystemContext.unavailable : systemBaseline)), + baseline: String, + update: (_previous, current) => current, + removed: () => "System context source removed: test/context", + }), ], - })), + ), + ), }), ) const runner = SessionRunnerLLM.layer.pipe( @@ -568,16 +568,8 @@ describe("SessionRunnerLLM", () => { ]) expect(requests[1]?.messages.map((message) => message.role)).toEqual(["user", "user", "system"]) expect(requests[1]?.messages.at(-1)?.content).toEqual([{ type: "text", text: "Changed context" }]) - expect(yield* session.messages({ sessionID })).toHaveLength(2) + expect(yield* session.messages({ sessionID })).toHaveLength(3) const { db } = yield* Database.Service - expect( - yield* db - .select({ id: EventTable.id }) - .from(EventTable) - .where(eq(EventTable.type, "session.next.context.initialized.1")) - .all() - .pipe(Effect.orDie), - ).toHaveLength(1) expect( yield* db .select({ id: EventTable.id }) @@ -587,11 +579,11 @@ describe("SessionRunnerLLM", () => { .pipe(Effect.orDie), ).toHaveLength(1) yield* replaySessionProjection(sessionID) - expect(yield* session.messages({ sessionID })).toHaveLength(2) + expect(yield* session.messages({ sessionID })).toHaveLength(3) }), ) - it.effect("admits removed context as a hidden chronological tombstone", () => + it.effect("admits removed context as a chronological System message", () => Effect.gen(function* () { yield* setup const session = yield* SessionV2.Service @@ -606,13 +598,13 @@ describe("SessionRunnerLLM", () => { expect(requests[1]?.messages.map((message) => message.role)).toEqual(["user", "user", "system"]) expect(requests[1]?.messages.at(-1)?.content).toEqual([ - { type: "text", text: "System context component removed: test/context" }, + { type: "text", text: "System context source removed: test/context" }, ]) - expect(yield* session.messages({ sessionID })).toHaveLength(2) + expect(yield* session.messages({ sessionID })).toHaveLength(3) }), ) - it.effect("replaces the baseline lazily after a model switch and drops prior hidden updates", () => + it.effect("replaces the baseline lazily after a model switch and drops prior System updates", () => Effect.gen(function* () { yield* setup const session = yield* SessionV2.Service @@ -641,27 +633,16 @@ describe("SessionRunnerLLM", () => { ]) expect(requests[1]?.messages.map((message) => message.role)).toEqual(["user", "user", "system"]) expect(requests[2]?.messages.map((message) => message.role)).toEqual(["user", "user", "user"]) - const { db } = yield* Database.Service - expect( - yield* db - .select({ id: EventTable.id }) - .from(EventTable) - .where(eq(EventTable.type, "session.next.context.replaced.1")) - .all() - .pipe(Effect.orDie), - ).toHaveLength(1) + expect((yield* session.context(sessionID)).map((message) => message.type)).toEqual([ + "user", + "user", + "model-switched", + "user", + ]) yield* replaySessionProjection(sessionID) - expect(yield* session.messages({ sessionID })).toHaveLength(4) + expect(yield* session.messages({ sessionID })).toHaveLength(5) yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Fourth" }), resume: false }) yield* session.resume(sessionID) - expect( - yield* db - .select({ id: EventTable.id }) - .from(EventTable) - .where(eq(EventTable.type, "session.next.context.replaced.1")) - .all() - .pipe(Effect.orDie), - ).toHaveLength(1) }), ) @@ -693,15 +674,39 @@ describe("SessionRunnerLLM", () => { ["Initial context"], ["Replacement context"], ]) + }), + ) + + it.effect("advances a pending replacement to the latest invalidation boundary", () => + Effect.gen(function* () { + yield* setup + const session = yield* SessionV2.Service + const events = yield* EventV2.Service const { db } = yield* Database.Service + yield* session.prompt({ sessionID, prompt: new Prompt({ text: "First" }), resume: false }) + response = [] + yield* session.resume(sessionID) + + yield* events.publish(SessionEvent.ModelSwitched, { + sessionID, + timestamp: DateTime.makeUnsafe(1), + model: { id: ModelV2.ID.make("replacement-1"), providerID: ProviderV2.ID.make("fake") }, + }) + yield* events.publish(SessionEvent.ModelSwitched, { + sessionID, + timestamp: DateTime.makeUnsafe(2), + model: { id: ModelV2.ID.make("replacement-2"), providerID: ProviderV2.ID.make("fake") }, + }) + const latest = yield* events.sequence(sessionID) + expect( yield* db - .select({ id: EventTable.id }) - .from(EventTable) - .where(eq(EventTable.type, "session.next.context.replaced.1")) - .all() + .select({ replacementSeq: SessionContextEpochTable.replacement_seq }) + .from(SessionContextEpochTable) + .where(eq(SessionContextEpochTable.session_id, sessionID)) + .get() .pipe(Effect.orDie), - ).toHaveLength(1) + ).toEqual({ replacementSeq: latest }) }), ) @@ -792,26 +797,41 @@ describe("SessionRunnerLLM", () => { ["Initial context"], ["Replacement context"], ]) - const { db } = yield* Database.Service - expect( - yield* db - .select({ id: EventTable.id }) - .from(EventTable) - .where(eq(EventTable.type, "session.next.context.replaced.1")) - .all() - .pipe(Effect.orDie), - ).toHaveLength(1) yield* replaySessionProjection(sessionID) yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Third" }), resume: false }) yield* session.resume(sessionID) - expect( - yield* db - .select({ id: EventTable.id }) - .from(EventTable) - .where(eq(EventTable.type, "session.next.context.replaced.1")) - .all() - .pipe(Effect.orDie), - ).toHaveLength(1) + }), + ) + + it.effect("preserves effective System updates while compaction replacement is blocked", () => + Effect.gen(function* () { + yield* setup + const session = yield* SessionV2.Service + const events = yield* EventV2.Service + yield* session.prompt({ sessionID, prompt: new Prompt({ text: "First" }), resume: false }) + + requests.length = 0 + response = [] + yield* session.resume(sessionID) + systemBaseline = "Changed context" + yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Second" }), resume: false }) + yield* session.resume(sessionID) + yield* events.publish(SessionEvent.Compaction.Started, { + sessionID, + timestamp: DateTime.makeUnsafe(1), + reason: "manual", + }) + yield* events.publish(SessionEvent.Compaction.Ended, { + sessionID, + timestamp: DateTime.makeUnsafe(2), + text: "summary", + }) + systemUnavailable = true + yield* session.prompt({ sessionID, prompt: new Prompt({ text: "Third" }), resume: false }) + yield* session.resume(sessionID) + + expect(requests.at(-1)?.system.map((part) => part.text)).toEqual(["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) }), ) diff --git a/packages/core/test/session-system-context.test.ts b/packages/core/test/session-system-context.test.ts index a2c979d0698..d0fb03afa7b 100644 --- a/packages/core/test/session-system-context.test.ts +++ b/packages/core/test/session-system-context.test.ts @@ -30,41 +30,48 @@ describe("SessionSystemContext", () => { Effect.gen(function* () { yield* TestClock.setTime(timestamp) const context = yield* SessionSystemContext.Service - const initialized = SystemContext.initialize(yield* context.load()) + const initialized = yield* SystemContext.initialize(yield* context.load()) - expect(initialized.baseline).toEqual([ - { - key: SystemContext.Key.make("core/environment"), - text: [ - "Here is some useful information about the environment you are running in:", - "", - ` Working directory: ${directory}`, - ` Workspace root folder: ${projectDirectory}`, - " Is directory a git repo: yes", - ` Platform: ${process.platform}`, - "", - ].join("\n"), - }, - { key: SystemContext.Key.make("core/date"), text: `Today's date: ${localDate(timestamp)}` }, - ]) + expect(initialized.baseline).toBe( + [ + "Here is some useful information about the environment you are running in:", + "", + ` Working directory: ${directory}`, + ` Workspace root folder: ${projectDirectory}`, + " Is directory a git repo: yes", + ` Platform: ${process.platform}`, + "", + "", + `Today's date: ${localDate(timestamp)}`, + ].join("\n"), + ) }), ) - it.effect("refreshes the date without repeating unchanged environment context", () => + it.effect("reconciles the date without repeating unchanged environment context", () => Effect.gen(function* () { yield* TestClock.setTime(timestamp) const context = yield* SessionSystemContext.Service - const initialized = SystemContext.initialize(yield* context.load()) + const initialized = yield* SystemContext.initialize(yield* context.load()) yield* TestClock.setTime(timestamp + 24 * 60 * 60 * 1000) - const refreshed = SystemContext.refresh(yield* context.load(), initialized.checkpoint) + const refreshed = yield* SystemContext.reconcile(yield* context.load(), initialized.snapshot) - expect(refreshed.changes).toEqual([ - { - key: SystemContext.Key.make("core/date"), - text: `Today's date is now: ${localDate(timestamp + 24 * 60 * 60 * 1000)}`, - }, - ]) + expect(refreshed).toMatchObject({ + _tag: "Updated", + text: `Today's date is now: ${localDate(timestamp + 24 * 60 * 60 * 1000)}`, + }) + }), + ) + + it.effect("does not update again within the same local calendar day", () => + Effect.gen(function* () { + yield* TestClock.setTime(timestamp) + const context = yield* SessionSystemContext.Service + const initialized = yield* SystemContext.initialize(yield* context.load()) + + yield* TestClock.setTime(timestamp + 60 * 60 * 1000) + expect(yield* SystemContext.reconcile(yield* context.load(), initialized.snapshot)).toEqual({ _tag: "Unchanged" }) }), ) }) diff --git a/packages/core/test/system-context.test.ts b/packages/core/test/system-context.test.ts index 92acf5ad30f..6aa7cb1470f 100644 --- a/packages/core/test/system-context.test.ts +++ b/packages/core/test/system-context.test.ts @@ -1,213 +1,300 @@ -import { describe, expect, test } from "bun:test" -import { Effect, Schema } from "effect" +import { describe, expect } from "bun:test" +import { Cause, Effect, Exit, Schema } from "effect" import { SystemContext } from "@opencode-ai/core/system-context" -import { Hash } from "@opencode-ai/core/util/hash" +import { it } from "./lib/effect" const key = SystemContext.Key.make +const stringContext = (input: { + key: string + value: string | SystemContext.Unavailable + baseline?: (value: string) => string + update?: (previous: string, current: string) => string + removed?: (value: string) => string +}) => + SystemContext.make({ + key: key(input.key), + codec: Schema.toCodecJson(Schema.String), + load: Effect.succeed(input.value), + baseline: input.baseline ?? String, + update: input.update ?? ((_previous, current) => current), + removed: input.removed, + }) describe("SystemContext", () => { - test("loads one coherent sample and initializes a deterministic baseline", async () => { - let loads = 0 - const context = SystemContext.struct({ - date: SystemContext.value({ + it.effect("stores the canonical JSON encoding of the loaded value", () => + Effect.gen(function* () { + const context = SystemContext.make({ key: key("core/date"), + codec: Schema.toCodecJson(Schema.DateFromString), + load: Effect.succeed(new Date("2026-06-03T12:00:00.000Z")), + baseline: (date) => date.toISOString(), + update: (_previous, date) => date.toISOString(), + removed: () => "Date removed", + }) + + expect((yield* SystemContext.initialize(context)).snapshot["core/date"].value).toBe("2026-06-03T12:00:00.000Z") + }), + ) + + it.effect("loads once and initializes a baseline with a structured snapshot", () => + Effect.gen(function* () { + let loads = 0 + const context = SystemContext.combine([ + SystemContext.make({ + key: key("core/date"), + codec: Schema.toCodecJson(Schema.String), + load: Effect.sync(() => { + loads++ + return "2026-06-03" + }), + baseline: (date) => `Today's date is ${date}.`, + update: (previous, current) => `The date changed from ${previous} to ${current}.`, + removed: () => "The date was removed.", + }), + stringContext({ key: "core/location", value: "/repo", baseline: (value) => `Directory: ${value}` }), + ]) + + expect(yield* SystemContext.initialize(context)).toEqual({ + baseline: "Today's date is 2026-06-03.\n\nDirectory: /repo", + snapshot: { + "core/date": { value: "2026-06-03", removed: "The date was removed." }, + "core/location": { value: "/repo" }, + }, + }) + expect(loads).toBe(1) + }), + ) + + it.effect("renders updates only after a structured value changes", () => + Effect.gen(function* () { + const previous = { + "core/date": { value: "2026-06-03", removed: "The date was removed." }, + "core/location": { value: "/repo", removed: "Removed: /repo" }, + } + const changed = SystemContext.combine([ + stringContext({ + key: "core/date", + value: "2026-06-04", + update: (before, current) => `The date changed from ${before} to ${current}.`, + removed: () => "The date was removed.", + }), + stringContext({ key: "core/location", value: "/repo" }), + ]) + + expect(yield* SystemContext.reconcile(changed, previous)).toEqual({ + _tag: "Updated", + text: "The date changed from 2026-06-03 to 2026-06-04.", + snapshot: { + "core/date": { value: "2026-06-04", removed: "The date was removed." }, + "core/location": { value: "/repo", removed: "Removed: /repo" }, + }, + }) + + expect( + yield* SystemContext.reconcile( + SystemContext.combine([ + stringContext({ key: "core/date", value: "2026-06-03", removed: () => "The date was removed." }), + stringContext({ key: "core/location", value: "/repo" }), + ]), + previous, + ), + ).toEqual({ _tag: "Unchanged" }) + }), + ) + + it.effect("uses the baseline for a newly added source", () => + Effect.gen(function* () { + const context = stringContext({ + key: "core/skills", + value: "effect", + baseline: (skill) => `Available skill: ${skill}`, + }) + + expect(yield* SystemContext.reconcile(context, {})).toEqual({ + _tag: "Updated", + text: "Available skill: effect", + snapshot: { "core/skills": { value: "effect" } }, + }) + }), + ) + + it.effect("retains admitted snapshots while a source is temporarily unavailable", () => + Effect.gen(function* () { + const previous = { "core/remote": { value: "instructions", removed: "Instructions removed" } } + const context = stringContext({ key: "core/remote", value: SystemContext.unavailable }) + + 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" }) + }), + ) + + it.effect("omits unavailable sources from an initial baseline", () => + Effect.gen(function* () { + expect(yield* SystemContext.initialize(stringContext({ key: "core/remote", value: SystemContext.unavailable }))).toEqual({ + baseline: "", + snapshot: {}, + }) + }), + ) + + it.effect("emits the previously stored removal message", () => + Effect.gen(function* () { + expect( + yield* SystemContext.reconcile(SystemContext.empty, { + "core/instructions": { value: "contents", removed: "Instructions removed; stop applying them." }, + }), + ).toEqual({ + _tag: "Updated", + text: "Instructions removed; stop applying them.", + snapshot: {}, + }) + }), + ) + + 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", + }) + }), + ) + + it.effect("renders multiple removals in stable key order", () => + Effect.gen(function* () { + expect( + yield* SystemContext.reconcile(SystemContext.empty, { + "core/z": { value: "z", removed: "Removed z" }, + "core/a": { value: "a", removed: "Removed a" }, + }), + ).toMatchObject({ _tag: "Updated", text: "Removed a\n\nRemoved z" }) + }), + ) + + it.effect("rejects empty model-visible renderings", () => + Effect.gen(function* () { + const exit = yield* SystemContext.initialize( + stringContext({ key: "core/empty", value: "value", baseline: () => "" }), + ).pipe(Effect.exit) + + expect(Exit.isFailure(exit)).toBe(true) + if (Exit.isFailure(exit)) expect(Cause.pretty(exit.cause)).toContain("rendered an empty baseline") + }), + ) + + it.effect("requests replacement when a stored value no longer decodes", () => + Effect.gen(function* () { + expect( + yield* SystemContext.reconcile(stringContext({ key: "core/date", value: "2026-06-04" }), { + "core/date": { value: 42, removed: "Date removed" }, + }), + ).toMatchObject({ _tag: "Replaced" }) + }), + ) + + it.effect("replaces from one coherent source observation", () => + Effect.gen(function* () { + let loads = 0 + const context = SystemContext.make({ + key: key("core/date"), + codec: Schema.toCodecJson(Schema.String), load: Effect.sync(() => { loads++ - return { baseline: "Today's date is 2026-06-03.", update: "The current date is 2026-06-03." } + return "2026-06-04" }), - }), - location: SystemContext.value({ - key: key("core/location"), - load: Effect.succeed({ baseline: "Working directory: /repo", update: "The working directory is /repo." }), - }), - }) + baseline: String, + update: (_previous, current) => current, + }) - const initialized = SystemContext.initialize(await Effect.runPromise(SystemContext.load(context))) + expect(yield* SystemContext.reconcile(context, { "core/date": { value: 42 } })).toMatchObject({ + _tag: "Replaced", + generation: { baseline: "2026-06-04" }, + }) + expect(loads).toBe(1) + }), + ) - expect(loads).toBe(1) - expect(initialized).toEqual({ - baseline: [ - { key: key("core/date"), text: "Today's date is 2026-06-03." }, - { key: key("core/location"), text: "Working directory: /repo" }, - ], - checkpoint: { - "core/date": Hash.sha256("The current date is 2026-06-03."), - "core/location": Hash.sha256("The working directory is /repo."), - }, - }) - }) - - test("emits changed and newly registered components in declaration order", async () => { - const context = SystemContext.struct({ - date: SystemContext.value({ - key: key("core/date"), - load: Effect.succeed({ baseline: "Today's date is 2026-06-04.", update: "The current date is 2026-06-04." }), - }), - location: SystemContext.value({ - key: key("core/location"), - load: Effect.succeed({ baseline: "Working directory: /repo", update: "The working directory is /repo." }), - }), - skills: SystemContext.value({ - key: key("core/skills"), - load: Effect.succeed({ baseline: "Available skills: effect", update: "Available skills: effect" }), - }), - }) - - const refreshed = SystemContext.refresh(await Effect.runPromise(SystemContext.load(context)), { - "core/date": Hash.sha256("The current date is 2026-06-03."), - "core/location": Hash.sha256("The working directory is /repo."), - }) - - expect(refreshed).toEqual({ - changes: [ - { key: key("core/date"), text: "The current date is 2026-06-04." }, - { key: key("core/skills"), text: "Available skills: effect" }, - ], - checkpoint: { - "core/date": Hash.sha256("The current date is 2026-06-04."), - "core/location": Hash.sha256("The working directory is /repo."), - "core/skills": Hash.sha256("Available skills: effect"), - }, - }) - }) - - test("omits unavailable initial context and admits it after its first successful load", async () => { - let available = false - const context = SystemContext.struct({ - remote: SystemContext.value({ - key: key("core/remote-instructions"), - load: Effect.sync(() => - available - ? { baseline: "Remote instructions: available", update: "Remote instructions are now available." } - : SystemContext.unavailable, - ), - }), - }) - - const initialized = SystemContext.initialize(await Effect.runPromise(SystemContext.load(context))) - available = true - const refreshed = SystemContext.refresh( - await Effect.runPromise(SystemContext.load(context)), - initialized.checkpoint, - ) - - expect(initialized).toEqual({ baseline: [], checkpoint: {} }) - expect(refreshed.changes).toEqual([ - { key: key("core/remote-instructions"), text: "Remote instructions are now available." }, - ]) - }) - - test("retains an existing checkpoint while context is unavailable", async () => { - const previous = { "core/remote-instructions": Hash.sha256("Remote instructions: old") } - const context = SystemContext.struct({ - remote: SystemContext.value({ - key: key("core/remote-instructions"), - load: Effect.succeed(SystemContext.unavailable), - }), - }) - - const refreshed = SystemContext.refresh(await Effect.runPromise(SystemContext.load(context)), previous) - - expect(refreshed).toEqual({ changes: [], checkpoint: previous }) - }) - - test("blocks replacement while admitted context is unavailable", async () => { - const previous = { "core/remote-instructions": Hash.sha256("Remote instructions: old") } - const snapshot = await Effect.runPromise( - SystemContext.load( - SystemContext.struct({ - remote: SystemContext.value({ - key: key("core/remote-instructions"), - load: Effect.succeed(SystemContext.unavailable), - }), + it.effect("does not render discarded updates while replacing", () => + Effect.gen(function* () { + let updates = 0 + const context = SystemContext.combine([ + stringContext({ + key: "core/date", + value: "2026-06-04", + update: () => { + updates++ + return "updated" + }, }), - ), - ) + stringContext({ key: "core/location", value: "/repo" }), + ]) - expect(SystemContext.replacementBlocked(snapshot, previous)).toBe(true) - expect(SystemContext.replacementBlocked(snapshot, {})).toBe(false) - }) + expect( + yield* SystemContext.reconcile(context, { + "core/date": { value: "2026-06-03" }, + "core/location": { value: 42 }, + }), + ).toMatchObject({ _tag: "Replaced" }) + expect(updates).toBe(0) + }), + ) - test("emits tombstones and drops checkpoints for removed components", async () => { - const context = SystemContext.struct({ - date: SystemContext.value({ - key: key("core/date"), - load: Effect.succeed({ baseline: "Today's date is 2026-06-03.", update: "The current date is 2026-06-03." }), - }), - }) + it.effect("blocks an incompatible replacement while another admitted source is unavailable", () => + Effect.gen(function* () { + const previous = { + "core/date": { value: 42, removed: "Date removed" }, + "core/remote": { value: "instructions", removed: "Instructions removed" }, + } + const context = SystemContext.combine([ + stringContext({ key: "core/date", value: "2026-06-04" }), + stringContext({ key: "core/remote", value: SystemContext.unavailable }), + ]) - const refreshed = SystemContext.refresh(await Effect.runPromise(SystemContext.load(context)), { - "core/date": Hash.sha256("The current date is 2026-06-03."), - "plugin/removed": Hash.sha256("Removed plugin context"), - }) + expect(yield* SystemContext.reconcile(context, previous)).toEqual({ _tag: "ReplacementBlocked" }) + expect(yield* SystemContext.replace(context, previous)).toEqual({ _tag: "ReplacementBlocked" }) + }), + ) - expect(refreshed).toEqual({ - changes: [{ key: key("plugin/removed"), text: "System context component removed: plugin/removed" }], - checkpoint: { "core/date": Hash.sha256("The current date is 2026-06-03.") }, - }) - }) + it.effect("rejects duplicate source keys", () => + Effect.sync(() => { + expect(() => + SystemContext.combine([ + stringContext({ key: "core/date", value: "one" }), + stringContext({ key: "core/date", value: "two" }), + ]), + ).toThrow(new SystemContext.DuplicateKeyError({ key: key("core/date") })) + }), + ) - test("ignores inherited checkpoint properties", async () => { - const context = SystemContext.struct({ - date: SystemContext.value({ - key: key("core/date"), - load: Effect.succeed({ baseline: "Today's date is 2026-06-03.", update: "The current date is 2026-06-03." }), - }), - }) - const previous = Object.create({ - "core/date": Hash.sha256("The current date is 2026-06-03."), - }) as SystemContext.Checkpoint + it.effect("combines contexts in order", () => + Effect.gen(function* () { + expect( + (yield* SystemContext.initialize( + SystemContext.combine([ + stringContext({ key: "core/date", value: "date" }), + stringContext({ key: "core/location", value: "location" }), + ]), + )).baseline, + ).toBe("date\n\nlocation") + }), + ) - const refreshed = SystemContext.refresh(await Effect.runPromise(SystemContext.load(context)), previous) + it.effect("requires namespaced source keys", () => + Effect.sync(() => { + const decodeKey = Schema.decodeUnknownSync(SystemContext.Key) - expect(refreshed.changes).toEqual([{ key: key("core/date"), text: "The current date is 2026-06-03." }]) - expect(Object.hasOwn(refreshed.checkpoint, "core/date")).toBe(true) - }) + expect(decodeKey("core/date")).toBe(key("core/date")) + expect(() => decodeKey("date")).toThrow() + }), + ) - test("preserves unexpected loader failures", async () => { - const context = SystemContext.struct({ - broken: SystemContext.value({ - key: key("plugin/broken"), - load: Effect.fail("broken loader"), - }), - }) + it.effect("requires namespaced durable snapshot keys", () => + Effect.sync(() => { + const decodeSnapshot = Schema.decodeUnknownSync(SystemContext.Snapshot) - await expect(Effect.runPromise(SystemContext.load(context))).rejects.toBe("broken loader") - }) - - test("rejects duplicate component keys", () => { - expect(() => - SystemContext.struct({ - one: SystemContext.value({ key: key("core/date"), load: Effect.succeed({ baseline: "one", update: "one" }) }), - two: SystemContext.value({ key: key("core/date"), load: Effect.succeed({ baseline: "two", update: "two" }) }), - }), - ).toThrow(new SystemContext.DuplicateKeyError({ key: key("core/date") })) - }) - - test("rejects duplicate component keys at the interpreter boundary", async () => { - const component = SystemContext.value({ - key: key("core/date"), - load: Effect.succeed({ baseline: "date", update: "date" }), - }) - const context: SystemContext.SystemContext = { components: [component, component] } - - await expect(Effect.runPromise(SystemContext.load(context))).rejects.toBeInstanceOf(SystemContext.DuplicateKeyError) - }) - - test("requires namespaced component keys", () => { - const decode = Schema.decodeUnknownSync(SystemContext.Key) - - expect(decode("core/date")).toBe(key("core/date")) - expect(() => decode("date")).toThrow() - expect(() => decode("core/")).toThrow() - }) - - test("requires namespaced checkpoint keys", () => { - const decode = Schema.decodeUnknownSync(SystemContext.CheckpointSchema) - const valid = JSON.parse('{"core/date":"hash"}') - const invalid = JSON.parse('{"date":"hash"}') - - expect(decode(valid)).toEqual(valid) - expect(() => decode(invalid)).toThrow() - }) + expect(Object.keys(decodeSnapshot({ "core/date": { value: "date" } }))).toEqual(["core/date"]) + expect(() => decodeSnapshot({ date: { value: "date" } })).toThrow() + expect(() => decodeSnapshot({ "core/date": { value: "date", removed: "" } })).toThrow() + }), + ) }) 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 be13d2007dc..57df7c39358 100644 --- a/packages/opencode/src/cli/cmd/tui/context/sync-v2.tsx +++ b/packages/opencode/src/cli/cmd/tui/context/sync-v2.tsx @@ -175,6 +175,16 @@ export const { use: useSyncV2, provider: SyncProviderV2 } = createSimpleContext( }) }) break + case "session.next.context.updated": + update(event.properties.sessionID, (draft) => { + draft.unshift({ + id: event.properties.messageID, + type: "system", + text: event.properties.text, + time: { created: event.properties.timestamp }, + }) + }) + break case "session.next.synthetic": update(event.properties.sessionID, (draft) => { prepend(draft, { diff --git a/packages/opencode/src/cli/cmd/tui/feature-plugins/system/session-v2.tsx b/packages/opencode/src/cli/cmd/tui/feature-plugins/system/session-v2.tsx index 9f2008d2b5c..2e54a7365d8 100644 --- a/packages/opencode/src/cli/cmd/tui/feature-plugins/system/session-v2.tsx +++ b/packages/opencode/src/cli/cmd/tui/feature-plugins/system/session-v2.tsx @@ -104,6 +104,9 @@ function View(props: { api: TuiPluginApi; sessionID: string }) { <> + + <> + diff --git a/specs/v2/refreshable-context-sources.md b/specs/v2/refreshable-context-sources.md index fd664996651..d1d8ff70eea 100644 --- a/specs/v2/refreshable-context-sources.md +++ b/specs/v2/refreshable-context-sources.md @@ -27,7 +27,7 @@ Watcher-backed caches are a later efficiency optimization for roots with proven | `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 immutable baseline parts, chronological updates, unavailable state, and removal tombstones. | +| `SystemContext` | Convert coherent source samples into one immutable baseline, chronological updates, unavailable state, and removal tombstones. | | `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. @@ -128,7 +128,7 @@ sequenceDiagram ## Observation Units -Compose refreshables around coherent observations that share one invalidation policy. Do not create one uniformly per rendered Context Component or one aggregate cache for unrelated source kinds. +Compose refreshables around coherent observations that share one invalidation policy. Do not create one uniformly per rendered Context Source or one aggregate cache for unrelated source kinds. ```text local built-in discovery @@ -153,13 +153,13 @@ Add a Location-scoped service: ```ts export interface InstructionContext.Interface { - readonly loadAmbient: () => Effect.Effect> + readonly loadAmbient: () => Effect.Effect } ``` `InstructionContext` owns instruction discovery, stable source identity, deterministic ordering, and source loading. `SystemContext` remains unaware of files and URLs. -Each effective source becomes one independently keyed component: +Each effective instruction becomes one independently keyed `SystemContext.Source` closed into the aggregate context with `SystemContext.make(...)`: ```text core/instructions/file/ @@ -192,7 +192,7 @@ sequenceDiagram Runner->>Instructions: loadAmbient Instructions->>Files: discover and read AGENTS.md files Files-->>Instructions: coherent current observation - Instructions-->>Runner: independently keyed components + Instructions-->>Runner: composed SystemContext Runner->>Epoch: compare and durably admit changes ``` @@ -208,19 +208,19 @@ observation -> what bytes or temporary failure does each identity currently produce? ``` -| Observation | Component outcome | +| Observation | Source outcome | | -------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------- | -| Local scan succeeds and discovers readable file | Available component with exact contents. | -| Local scan succeeds and a previously discovered file is absent | Remove component so `SystemContext` emits a tombstone. | +| Local scan succeeds and discovers readable file | Available source with exact contents. | +| Local scan succeeds and a previously discovered file is absent | Remove source so `SystemContext` emits a tombstone. | | Local scan fails transiently | Preserve the domain-owned prior source graph as unavailable or fail the current turn; never emit mass removals. | -| Known local file read fails transiently | Preserve the component as `SystemContext.unavailable`. | +| Known local file read fails transiently | Preserve the source as `SystemContext.unavailable`. | | Known local file read reports not-found after discovery | Invalidate discovery and report unavailable until a coherent rescan confirms removal. | | Empty local file | Available exact content, not absence. | -| URL returns `2xx` body | Available component with exact contents. | +| 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. | -Instruction removal text must be model-meaningful. If instruction component keys hash source identities, add source-specific removal rendering before unlink support is considered complete: +Instruction removal text must be model-meaningful. If instruction source keys hash source identities, add source-specific removal rendering before unlink support is considered complete: ```text Instructions removed: /repo/packages/core/AGENTS.md @@ -234,7 +234,7 @@ Implement only: ```text global config AGENTS.md + upward project AGENTS.md ancestors -+ one keyed component per file ++ one keyed source per file + direct safe-turn observation ``` @@ -263,7 +263,7 @@ empty file transient scan failure transient file-read failure deterministic ordering -restart with durable checkpoint +restart with durable structured snapshots ``` ## Future Watcher Optimization @@ -316,7 +316,7 @@ Start with direct safe-turn loading: ```text safe provider-turn boundary -> fetch URL --> emit available or unavailable component +-> emit available or unavailable source ``` If measurements show excessive requests, add a URL-specific invalidation policy later: @@ -375,7 +375,7 @@ Nested instructions discovered after successful read-tool activity remain a Sess 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 Location-scoped `InstructionContext` for global and upward project `AGENTS.md` only. -4. Compose instruction components into `SessionSystemContext.load()`. +4. Compose instruction sources into `SessionSystemContext.load()`. 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. @@ -386,11 +386,10 @@ Nested instructions discovered after successful read-tool activity remain a Sess ## Open Questions -1. Should source-specific removal rendering extend `SystemContext.Component`, or should `InstructionContext` retain removal metadata in a separate component registry? -2. Should the first local scan failure preserve prior discovered sources as unavailable, or fail the current provider turn until a coherent rescan succeeds? -3. Should configured URL sources treat `404` and `410` as confirmed removals? -4. What root-specific watcher API cleanly models ignore policy and callback health? -5. Should own-process file mutations publish an advisory invalidation event synchronously after commit? +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? ## Compression Line diff --git a/specs/v2/schema-changelog.md b/specs/v2/schema-changelog.md index 441435eb5dc..e9a2bc7f19f 100644 --- a/specs/v2/schema-changelog.md +++ b/specs/v2/schema-changelog.md @@ -687,23 +687,23 @@ Compatibility: - Foreground V2 bash execution is unchanged. - Reintroduce background bash only with durable status observation, completion delivery, and explicit cancellation semantics. -## 2026-06-04: Initialize Durable Session Context Epochs +## 2026-06-04: Add Durable Session Context Snapshots Affected schema: -- Add synchronized `session.next.context.initialized.1` Session events. -- Add `session_context_epoch` for one active immutable keyed baseline, component-hash checkpoint, and baseline sequence per Session. +- Add `session_context_epoch` for one active immutable baseline string, structured JSON snapshot, and baseline sequence per Session. Change: -- Lazily initialize one durable Context Epoch at the first safe provider-turn boundary. -- Lower its exact keyed baseline parts through `LLMRequest.system` for every provider turn in the epoch. +- Lazily initialize one durable Context Epoch snapshot at the first safe provider-turn boundary. +- Lower its exact baseline string through `LLMRequest.system` for every provider turn in the epoch. - Reuse the stored baseline verbatim after restart or producer changes instead of resampling privileged initial context. -- Keep ordinary Session transcript APIs unchanged. +- Compare later observations against an overwriteable codec-encoded structured snapshot rather than rendered-text hashes. +- Expose admitted chronological context as first-class `system` Session messages while keeping the active baseline in bounded context state. Compatibility: -- The unpublished Context Epoch schema is consolidated into one database migration and this adds one synchronized Session event type. +- The unpublished Context Epoch schema is consolidated into one database migration; baseline and structured snapshots are operational state rather than synchronized event history. - Existing experimental V2 Session databases remain disposable across incompatible pre-launch event-schema changes. - Chronological context updates, replacement epochs after compaction or model switches, project instructions, skills guidance, and plugin transforms remain follow-up slices. @@ -711,21 +711,21 @@ Compatibility: Affected schema: -- Add synchronized `session.next.context.updated.1` Session events. -- Add `session_context_epoch.revision` for transactional checkpoint advancement. -- Add `session_context_message` for hidden chronological keyed context updates ordered by Session aggregate sequence. +- Add synchronized `session.next.context.updated.1` Session events containing 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. Change: -- Refresh Location-scoped Context Components at each safe provider-turn boundary. -- Keep the stored baseline immutable while admitting changed component values as runner-private chronological `Message.system(...)` history. -- Emit an explicit tombstone update and advance component-hash checkpoints transactionally when a component is removed. -- Keep ordinary Session transcript APIs unchanged while runner history merges visible Session messages and hidden context updates by durable aggregate sequence. +- Reconcile Location-scoped Context Sources at each safe provider-turn boundary using one coherent observation. +- Keep the stored baseline immutable while admitting changed source renderings as chronological `Message.system(...)` history. +- Advance the overwriteable structured snapshot atomically with the rendered System-message event. +- Emit the previously stored model-meaningful removal rendering when a source is removed. - Reject chronological system updates that would split a local tool call from its result across provider protocols; use wrapped user fallback when Anthropic native system-update placement is unsupported. Compatibility: -- The unpublished Context Epoch schema remains consolidated into one database migration and this adds one synchronized Session event type. +- 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. @@ -733,18 +733,17 @@ Compatibility: Affected schema: -- Add synchronized `session.next.context.replaced.1` Session events. - Add nullable `session_context_epoch.replacement_seq` for idempotent lazy replacement requests. Change: - Mark the active Context Epoch for replacement after a model switch or completed compaction projection. - Persist the triggering aggregate sequence so same-target replay cannot reopen an already-settled replacement. -- Render and persist the fresh immutable baseline lazily at the next safe provider-turn boundary. -- Exclude hidden chronological updates from earlier epochs when assembling active provider history. +- Render and overwrite the fresh immutable baseline and structured snapshot lazily at the next safe provider-turn boundary. +- Exclude chronological System messages from earlier epochs when assembling active provider history. Compatibility: -- The unpublished Context Epoch schema remains consolidated into one database migration and this adds one synchronized Session event type. +- 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.