diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index 47bfa0433e9..c45a1d80366 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -970,6 +970,20 @@ export type SessionLogOutput = readonly reason: "auto" | "manual" readonly model?: Model.Ref | undefined readonly providerState?: SessionMessage.ProviderState | undefined + readonly providerContext?: + | { + readonly version: 1 + readonly provenance: { + readonly providerID: Provider.ID + readonly provider: string + readonly modelID: string + readonly route: string + readonly protocol: string + readonly endpoint: string + } + readonly messages: Schema.Json + } + | undefined readonly text: string readonly recent: string } diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index 02a49cae1fa..ac9ed1348ed 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -138,6 +138,15 @@ export type SessionMessageCompactionRunning = { recent: string } +export type SessionProviderContextProvenance = { + providerID: string + provider: string + modelID: string + route: string + protocol: string + endpoint: string +} + export type SessionActive = { type: "running" } export type SessionInboxDelivery = "steer" | "queue" @@ -512,19 +521,6 @@ export type SessionMessageAssistantReasoning = { time?: { created: number; completed?: number } } -export type SessionMessageCompactionCompleted = { - type: "compaction" - id: string - metadata?: { [x: string]: JsonValue } - time: { created: number } - status: "completed" - reason: "auto" | "manual" - model?: ModelRef - providerState?: SessionMessageProviderState - summary: string - recent: string -} - export type ToolContent = ToolTextContent | ToolFileContent export type SessionMessageAssistantRetry = { attempt: number; at: number; error: SessionStructuredError } @@ -539,6 +535,8 @@ export type SessionMessageCompactionFailed = { error: SessionStructuredError } +export type SessionProviderContext = { version: 1; provenance: SessionProviderContextProvenance; messages: JsonValue } + export type SessionInboxSynthetic = { id: string sessionID: string @@ -1345,23 +1343,6 @@ export type SessionToolCalled = { } } -export type SessionCompactionEnded = { - id: string - created: number - metadata?: { [x: string]: any } - type: "session.compaction.ended" - durable: { aggregateID: string; seq: number; version: 1 } - location?: LocationRef - data: { - sessionID: string - reason: "auto" | "manual" - model?: ModelRef - providerState?: SessionMessageProviderState1 - text: string - recent: string - } -} - export type SessionMessageAssistantText1 = { type: "text"; text: string; state?: SessionMessageProviderState1 } export type SessionMessageAssistantReasoning1 = { @@ -1742,10 +1723,37 @@ export type SessionMessageToolStateError = { metadata?: { [x: string]: JsonValue } } -export type SessionMessageCompaction = - | SessionMessageCompactionRunning - | SessionMessageCompactionCompleted - | SessionMessageCompactionFailed +export type SessionMessageCompactionCompleted = { + type: "compaction" + id: string + metadata?: { [x: string]: JsonValue } + time: { created: number } + status: "completed" + reason: "auto" | "manual" + model?: ModelRef + providerState?: SessionMessageProviderState + summary: string + recent: string + providerContext?: SessionProviderContext +} + +export type SessionCompactionEnded = { + id: string + created: number + metadata?: { [x: string]: any } + type: "session.compaction.ended" + durable: { aggregateID: string; seq: number; version: 1 } + location?: LocationRef + data: { + sessionID: string + reason: "auto" | "manual" + model?: ModelRef + providerState?: SessionMessageProviderState1 + providerContext?: SessionProviderContext + text: string + recent: string + } +} export type SessionForked = { id: string @@ -2085,6 +2093,11 @@ export type SessionMessageAssistantTool = { time: { created: number; ran?: number; completed?: number } } +export type SessionMessageCompaction = + | SessionMessageCompactionRunning + | SessionMessageCompactionCompleted + | SessionMessageCompactionFailed + export type SessionMessageAssistantTool1 = { type: "tool" id: string @@ -3080,6 +3093,18 @@ export type SessionImportInput = { readonly providerState?: { readonly [x: string]: JsonValue } readonly summary: string readonly recent: string + readonly providerContext?: { + readonly version: 1 + readonly provenance: { + readonly providerID: string + readonly provider: string + readonly modelID: string + readonly route: string + readonly protocol: string + readonly endpoint: string + } + readonly messages: JsonValue + } } | { readonly type: "compaction" @@ -3359,6 +3384,18 @@ export type SessionImportInput = { readonly providerState?: { readonly [x: string]: JsonValue } readonly summary: string readonly recent: string + readonly providerContext?: { + readonly version: 1 + readonly provenance: { + readonly providerID: string + readonly provider: string + readonly modelID: string + readonly route: string + readonly protocol: string + readonly endpoint: string + } + readonly messages: JsonValue + } } | { readonly type: "compaction" @@ -3638,6 +3675,18 @@ export type SessionImportInput = { readonly providerState?: { readonly [x: string]: JsonValue } readonly summary: string readonly recent: string + readonly providerContext?: { + readonly version: 1 + readonly provenance: { + readonly providerID: string + readonly provider: string + readonly modelID: string + readonly route: string + readonly protocol: string + readonly endpoint: string + } + readonly messages: JsonValue + } } | { readonly type: "compaction" diff --git a/packages/core/src/session/context.ts b/packages/core/src/session/context.ts index 54e20485055..81afec73872 100644 --- a/packages/core/src/session/context.ts +++ b/packages/core/src/session/context.ts @@ -18,6 +18,7 @@ import { SkillInstructions } from "../skill/instructions.js" import { Tool } from "../tool.js" import { AgentNotFoundError } from "./error.js" import { SessionHistory } from "./history.js" +import { SessionProviderContext } from "./provider-context.js" import { InstructionEntry } from "./instruction-entry.js" import { SessionMessage } from "./message.js" import { SessionModelRequest } from "./model-request.js" @@ -156,7 +157,12 @@ const layer = Layer.effect( const load = Effect.fn("SessionContext.load")(function* (selection: Selection) { const model = yield* resolveModel(selection.session) - const history = yield* SessionHistory.entriesForRunner(db, selection.session.id, selection.instructions) + const history = yield* SessionHistory.entriesForRunner( + db, + selection.session.id, + selection.instructions, + SessionProviderContext.provenance(model) ?? "local", + ) return { session: selection.session, agent: selection.agent, diff --git a/packages/core/src/session/generate.ts b/packages/core/src/session/generate.ts index 4dada9e8024..71694f8683d 100644 --- a/packages/core/src/session/generate.ts +++ b/packages/core/src/session/generate.ts @@ -9,6 +9,7 @@ import type { Instructions } from "../instructions/index.js" import { SessionContext } from "./context.js" import type { AgentNotFoundError } from "./error.js" import { SessionHistory } from "./history.js" +import { SessionProviderContext } from "./provider-context.js" import { SessionModelRequest } from "./model-request.js" import type { SessionRunnerModel } from "./runner/model.js" import type { SessionSchema } from "./schema.js" @@ -29,7 +30,12 @@ export const generate = Effect.fn("SessionGenerate.generate")(function* (input: const context = yield* SessionContext.Service const selection = yield* context.select(input.session.id) const model = yield* context.resolveModel(selection.session) - const history = yield* SessionHistory.preview(database.db, selection.session.id, selection.instructions) + const history = yield* SessionHistory.preview( + database.db, + selection.session.id, + selection.instructions, + SessionProviderContext.provenance(model) ?? "local", + ) const transcript = SessionModelRequest.baseTranscript({ agent: selection.agent.info, model, diff --git a/packages/core/src/session/history.ts b/packages/core/src/session/history.ts index b3df7847160..aeb7cfc9f81 100644 --- a/packages/core/src/session/history.ts +++ b/packages/core/src/session/history.ts @@ -1,4 +1,4 @@ -import { and, asc, desc, eq, gte, sql } from "drizzle-orm" +import { and, asc, desc, eq, gte, or, sql } from "drizzle-orm" import { Effect, Schema } from "effect" import { Database } from "../database/database.js" import { MessageDecodeError } from "./error.js" @@ -6,13 +6,30 @@ import { SessionMessage } from "./message.js" import { SessionSchema } from "./schema.js" import { Instructions } from "../instructions/index.js" import { InstructionState } from "./instruction-state.js" +import { SessionProviderContext } from "./provider-context.js" import { SessionMessageTable } from "./sql.js" type DatabaseService = Database.Interface["db"] const decode = Schema.decodeUnknownEffect(SessionMessage.Info) -export const latestCompaction = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) { +/** + * Which completed compactions bound a history read. Local summaries always do. Native + * windows do for model-neutral readers (`latest`), never for the original transcript + * (`local`), and only when the target model can replay them (a provenance). + */ +export type Boundary = "latest" | "local" | SessionProviderContext.Provenance + +const replayable = (message: SessionMessage.Info, boundary: Boundary) => + !SessionProviderContext.isCheckpoint(message) || + boundary === "latest" || + (boundary !== "local" && SessionProviderContext.compatible(message.providerContext.provenance, boundary)) + +export const latestCompaction = Effect.fnUntraced(function* ( + db: DatabaseService, + sessionID: SessionSchema.ID, + boundary: Boundary, +) { return yield* db .select({ seq: SessionMessageTable.seq }) .from(SessionMessageTable) @@ -21,6 +38,19 @@ export const latestCompaction = Effect.fnUntraced(function* (db: DatabaseService eq(SessionMessageTable.session_id, sessionID), eq(SessionMessageTable.type, "compaction"), sql`json_extract(${SessionMessageTable.data}, '$.status') = 'completed'`, + boundary === "latest" + ? undefined + : or( + sql`json_extract(${SessionMessageTable.data}, '$.providerContext') is null`, + boundary === "local" + ? undefined + : and( + ...Object.entries(boundary).map( + ([key, value]) => + sql`json_extract(${SessionMessageTable.data}, ${`$.providerContext.provenance.${key}`}) = ${value}`, + ), + ), + ), ), ) .orderBy(desc(SessionMessageTable.seq)) @@ -31,6 +61,11 @@ export const latestCompaction = Effect.fnUntraced(function* (db: DatabaseService export const decodeMessageRow = (row: typeof SessionMessageTable.$inferSelect) => decode({ ...row.data, id: row.id, type: row.type }).pipe( + Effect.tap((message) => + SessionProviderContext.isCheckpoint(message) + ? SessionProviderContext.validate(message.providerContext) + : Effect.void, + ), Effect.mapError( () => new MessageDecodeError({ @@ -40,8 +75,12 @@ export const decodeMessageRow = (row: typeof SessionMessageTable.$inferSelect) = ), ) -const messageEntries = Effect.fnUntraced(function* (db: DatabaseService, sessionID: SessionSchema.ID) { - const compaction = yield* latestCompaction(db, sessionID) +const messageEntries = Effect.fnUntraced(function* ( + db: DatabaseService, + sessionID: SessionSchema.ID, + boundary: Boundary, +) { + const compaction = yield* latestCompaction(db, sessionID, boundary) const rows = yield* db .select() .from(SessionMessageTable) @@ -54,24 +93,38 @@ const messageEntries = Effect.fnUntraced(function* (db: DatabaseService, session .orderBy(asc(SessionMessageTable.seq)) .all() .pipe(Effect.orDie) - return yield* Effect.forEach(rows, (row) => + const entries = yield* Effect.forEach(rows, (row) => decodeMessageRow(row).pipe(Effect.map((message) => ({ seq: row.seq, message }))), ) + // Re-expansion may cross a native checkpoint whose completion already advanced the instruction + // epoch: the baseline supersedes the chronological updates before it. Forks seed their baseline + // at sequence 0 but retain parent sequences, so the copied checkpoint still retires them. + const native = entries.findLast((entry) => SessionProviderContext.isCheckpoint(entry.message)) + // Skipped native checkpoints are not textual summaries. Their original transcript remains available. + return entries.filter( + (entry) => + !(entry.message.type === "system" && native && entry.seq < native.seq) && replayable(entry.message, boundary), + ) }) -export const load = Effect.fn("SessionHistory.load")(function* (db: DatabaseService, sessionID: SessionSchema.ID) { - return (yield* messageEntries(db, sessionID)).map((entry) => entry.message) +export const load = Effect.fn("SessionHistory.load")(function* ( + db: DatabaseService, + sessionID: SessionSchema.ID, + boundary: Boundary, +) { + return (yield* messageEntries(db, sessionID, boundary)).map((entry) => entry.message) }) export const entriesForRunner = Effect.fn("SessionHistory.entriesForRunner")(function* ( db: DatabaseService, sessionID: SessionSchema.ID, instructions: Instructions.List, + boundary: Boundary, ) { return yield* db .transaction(() => Effect.gen(function* () { - const messages = yield* messageEntries(db, sessionID) + const messages = yield* messageEntries(db, sessionID, boundary) return { initial: yield* InstructionState.initial(db, sessionID, instructions), entries: messages, @@ -85,12 +138,13 @@ export const preview = Effect.fn("SessionHistory.preview")(function* ( db: DatabaseService, sessionID: SessionSchema.ID, instructions: Instructions.List, + boundary: Boundary, ) { const observed = yield* Instructions.read(instructions) return yield* db .transaction(() => Effect.gen(function* () { - const messages = yield* messageEntries(db, sessionID) + const messages = yield* messageEntries(db, sessionID, boundary) // An active assistant may contain an unresolved tool call, so only preview the settled prefix. const unsettled = messages.findIndex( (entry) => entry.message.type === "assistant" && entry.message.time.completed === undefined, diff --git a/packages/core/src/session/message-updater.ts b/packages/core/src/session/message-updater.ts index 126e9d28b69..8675690f33b 100644 --- a/packages/core/src/session/message-updater.ts +++ b/packages/core/src/session/message-updater.ts @@ -413,6 +413,7 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) { model: event.data.model, providerState: event.data.providerState, summary: event.data.text, + providerContext: event.data.providerContext, recent: event.data.recent, }) return @@ -427,6 +428,7 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) { model: event.data.model, providerState: event.data.providerState, summary: event.data.text, + providerContext: event.data.providerContext, recent: event.data.recent, time: { created }, }), diff --git a/packages/core/src/session/model-request.ts b/packages/core/src/session/model-request.ts index 115aeeec632..ce097b4940b 100644 --- a/packages/core/src/session/model-request.ts +++ b/packages/core/src/session/model-request.ts @@ -15,6 +15,7 @@ import { PluginHooks } from "../plugin/hooks.js" import { QuestionTool } from "../tool/plugin/question.js" import { Tool } from "../tool.js" import { SessionModelTransport } from "./model-transport.js" +import { SessionProviderContext } from "./provider-context.js" import { SessionRunnerModel } from "./runner/model.js" import { SessionSchema } from "./schema.js" import { SessionSystemPrompt } from "./system-prompt.js" @@ -345,6 +346,21 @@ export const layer = Layer.effect( providerOptions: Object.keys(context.providerOptions).length === 0 ? undefined : context.providerOptions, }), ) + // History selects native windows against the catalog route before hooks run. A newly installed + // routing hook must not send an existing opaque window to another deployment; `prepare` has no + // error channel, so like hook failures this surfaces as a defect. + const selected = SessionProviderContext.provenance(resolved) + if ( + selected && + !SessionProviderContext.compatible( + selected, + SessionProviderContext.provenance({ model: request.model, ref: resolved.ref }), + ) && + request.messages.some((message) => message.content.some((part) => part.type === "compaction")) + ) + return yield* Effect.die( + new Error("Provider context is incompatible with the route selected by model request hooks"), + ) const hasHttpHooks = (yield* hooks.has("session", "http.request", resolved.ref.providerID)) || (yield* hooks.has("session", "http.response", resolved.ref.providerID)) diff --git a/packages/core/src/session/provider-context.ts b/packages/core/src/session/provider-context.ts new file mode 100644 index 00000000000..4880a45772e --- /dev/null +++ b/packages/core/src/session/provider-context.ts @@ -0,0 +1,69 @@ +export * as SessionProviderContext from "./provider-context.js" + +import { Message } from "@opencode-ai/ai" +import { SessionProviderContext } from "@opencode-ai/schema/session-provider-context" +import { Schema } from "effect" +import { isDeepStrictEqual } from "node:util" +import { Hash } from "@opencode-ai/util/hash" +import type { SessionMessage } from "./message.js" +import type { SessionRunnerModel } from "./runner/model.js" + +export type Provenance = SessionProviderContext.Provenance +export const Info = SessionProviderContext.Info +export type Info = SessionProviderContext.Info + +const messages = Schema.toCodecJson(Schema.Array(Message)) + +/** No guessed endpoints. Dynamic URL builders cannot establish a durable deployment identity here. */ +export function provenance(resolved: Pick): Provenance | undefined { + const model = resolved.model + const endpoint = model.route.endpoint + if (!endpoint.baseURL || typeof endpoint.path !== "string") return undefined + return { + providerID: resolved.ref.providerID, + provider: model.provider, + modelID: model.id, + route: model.route.id, + protocol: model.route.protocol, + endpoint: Hash.sha256( + JSON.stringify([ + endpoint.baseURL, + endpoint.path, + Object.entries(endpoint.query ?? {}).sort((a, b) => (a[0] < b[0] ? -1 : a[0] > b[0] ? 1 : 0)), + ]), + ), + } +} + +export const compatible = (source: Provenance, target: Provenance | undefined) => + target !== undefined && isDeepStrictEqual(source, target) + +/** A completed compaction that installed a native replacement window instead of a local summary. */ +export const isCheckpoint = ( + message: SessionMessage.Info, +): message is SessionMessage.CompactionCompleted & { readonly providerContext: Info } => + message.type === "compaction" && message.status === "completed" && message.providerContext !== undefined + +/** Stores the canonical replacement, not a local summary or transport continuation. + * Provider and attachment metadata can contain optional undefined entries. Use JSON's + * omission semantics, while preserving canonical binary media as equivalent base64. + */ +export const encode = (provenance: Provenance, replacement: ReadonlyArray): Info => ({ + version: 1, + provenance, + messages: Schema.decodeSync(Schema.fromJsonString(Schema.Json))( + JSON.stringify( + replacement.map((message) => ({ + ...message, + content: message.content.map((part) => + part.type === "media" && part.data instanceof Uint8Array + ? { ...part, data: Buffer.from(part.data).toString("base64") } + : part, + ), + })), + ), + ), +}) + +export const decode = (context: Info) => Schema.decodeUnknownSync(messages)(context.messages) +export const validate = (context: Info) => Schema.decodeUnknownEffect(messages)(context.messages) diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index c83b462aeed..a7ab0972ab3 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -11,6 +11,7 @@ import { SessionContext } from "../context.js" import { SessionEvent } from "../event.js" import { SessionInbox } from "../inbox.js" import { SessionHistory } from "../history.js" +import { SessionProviderContext } from "../provider-context.js" import { SessionModelRequest } from "../model-request.js" import { SessionModelTransport } from "../model-transport.js" import { SessionMessage } from "../message.js" @@ -113,7 +114,12 @@ const layer = Layer.effect( const selected = yield* context.select(session.id) const model = yield* context.resolveModel(selected.session) // Preview updates without admitting them after the already-delivered compaction marker. - const history = yield* SessionHistory.preview(db, session.id, selected.instructions) + const history = yield* SessionHistory.preview( + db, + session.id, + selected.instructions, + SessionProviderContext.provenance(model) ?? "local", + ) return { session: selected.session, agent: selected.agent, diff --git a/packages/core/src/session/runner/to-llm-message.ts b/packages/core/src/session/runner/to-llm-message.ts index 8b5d767f7f6..17fb423c5b4 100644 --- a/packages/core/src/session/runner/to-llm-message.ts +++ b/packages/core/src/session/runner/to-llm-message.ts @@ -3,6 +3,7 @@ import type { Model } from "@opencode-ai/schema/model" import { Option, Schema } from "effect" import { fileURLToPath } from "url" import { SessionMessage } from "../message.js" +import { SessionProviderContext } from "../provider-context.js" import type { FileAttachment } from "@opencode-ai/schema/prompt" const imageMimes = new Set(["image/png", "image/jpeg", "image/gif", "image/webp"]) @@ -274,6 +275,9 @@ function toLLMMessage(message: SessionMessage.Info, model: Model.Ref, providerMe return assistant(message, model, providerMetadataKey) case "compaction": if (message.status !== "completed") return [] + // History selection only keeps native windows the target model can replay. + if (SessionProviderContext.isCheckpoint(message)) + return [...SessionProviderContext.decode(message.providerContext)] return [ Message.make({ id: message.id, diff --git a/packages/core/src/session/store.ts b/packages/core/src/session/store.ts index e10c16e47e5..427d62601ba 100644 --- a/packages/core/src/session/store.ts +++ b/packages/core/src/session/store.ts @@ -172,7 +172,7 @@ const layer = Layer.effect( SessionHistory.decodeMessageRow, ) }), - context: Effect.fn("SessionStore.context")((sessionID) => SessionHistory.load(db, sessionID)), + context: Effect.fn("SessionStore.context")((sessionID) => SessionHistory.load(db, sessionID, "latest")), message: Effect.fn("SessionStore.message")(function* (messageID) { const row = yield* db .select() diff --git a/packages/core/test/session-provider-context.test.ts b/packages/core/test/session-provider-context.test.ts new file mode 100644 index 00000000000..02ede815735 --- /dev/null +++ b/packages/core/test/session-provider-context.test.ts @@ -0,0 +1,303 @@ +import { expect, test } from "bun:test" +import { CompactionPart, LanguageModel, Message, ToolCallPart } from "@opencode-ai/ai" +import { OpenAIResponses } from "@opencode-ai/ai/protocols" +import { Bus } from "@opencode-ai/core/bus" +import { Database } from "@opencode-ai/core/database/database" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { EventTable } from "@opencode-ai/core/event/sql" +import { Instructions } from "@opencode-ai/core/instructions/index" +import { Project } from "@opencode-ai/core/project" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { SessionEvent } from "@opencode-ai/core/session/event" +import { SessionHistory } from "@opencode-ai/core/session/history" +import { SessionInbox } from "@opencode-ai/core/session/inbox" +import { InstructionState } from "@opencode-ai/core/session/instruction-state" +import { SessionMessage } from "@opencode-ai/core/session/message" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { SessionProviderContext } from "@opencode-ai/core/session/provider-context" +import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model" +import { toLLMMessages } from "@opencode-ai/core/session/runner/to-llm-message" +import { SessionSchema } from "@opencode-ai/core/session/schema" +import { InstructionStateTable, SessionTable } from "@opencode-ai/core/session/sql" +import { SessionStore } from "@opencode-ai/core/session/store" +import { LayerNode } from "@opencode-ai/util/effect/layer-node" +import { Model } from "@opencode-ai/schema/model" +import { asc, eq } from "drizzle-orm" +import { Effect, Schema } from "effect" +import { testEffect } from "./lib/effect" + +const model = SessionRunnerModel.resolved( + LanguageModel.make({ id: "deployment", provider: "openai", route: OpenAIResponses.route }), + { + capabilities: { tools: true, input: ["text"], output: ["text"] }, + cost: [], + limit: { context: 128_000, output: 4096 }, + }, +) +const target = SessionProviderContext.provenance(model) +if (!target) throw new Error("Fixture must have a concrete endpoint") +const replacement = [ + Message.user("retained request"), + Message.assistant( + CompactionPart.make({ provider: model.model.provider, encrypted: "opaque-checkpoint", id: "cp_1" }), + ), +] +const providerContext = SessionProviderContext.encode(target, replacement) +const sessionID = SessionSchema.ID.make("ses_provider_context") + +const it = testEffect( + AppNodeBuilder.build( + LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionInbox.node, SessionStore.node]), + [Bus.node.replace(Bus.configured({ persist: true }))], + ), +) + +const setup = Effect.gen(function* () { + const database = yield* Database.Service + const bus = yield* Bus.Service + const inbox = yield* SessionInbox.Service + yield* database.db + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] }) + .run() + yield* bus.publish(SessionEvent.Created, { + sessionID, + projectID: Project.ID.global, + location: { directory: AbsolutePath.make("/project") }, + slug: "provider-context", + version: "test", + }) + const state = { value: "initial instructions" } + const instructions = Instructions.make({ + key: Instructions.Key.make("test/context"), + codec: Schema.toCodecJson(Schema.String), + read: Effect.sync(() => state.value), + render: { initial: String, changed: (_previous, value) => value, removed: () => "removed" }, + }) + const prepare = InstructionState.prepare(database.db, bus, instructions, sessionID) + const prompt = Effect.fnUntraced(function* (text: string) { + const id = SessionMessage.ID.create() + yield* inbox.admit({ id, sessionID, item: { type: "user", payload: { text }, delivery: "steer" } }) + yield* bus.publish(SessionEvent.InboxDelivered, { sessionID, inboxID: id }) + return id + }) + const compact = (context?: SessionProviderContext.Info) => + bus.publish(SessionEvent.Compaction.Ended, { + sessionID, + reason: "manual", + text: context ? "" : "local summary", + recent: "", + providerContext: context, + }) + const load = (boundary: SessionHistory.Boundary) => + SessionHistory.entriesForRunner(database.db, sessionID, instructions, boundary) + return { db: database.db, bus, state, instructions, prepare, prompt, compact, load } +}) + +test("canonical provider context round-trips tools, opaque checkpoints and binary media through JSON", () => { + const messages = [ + ...replacement, + Message.assistant(ToolCallPart.make({ id: "call_1", name: "read", input: { path: "file" } })), + Message.tool({ id: "call_1", name: "read", result: { text: "result" } }), + Message.user({ type: "media", mediaType: "image/png", data: new Uint8Array([1, 2, 3]) }), + ] + const context = SessionProviderContext.encode(providerContext.provenance, messages) + const stored = Schema.decodeUnknownSync(Schema.fromJsonString(SessionProviderContext.Info))(JSON.stringify(context)) + const decoded = SessionProviderContext.decode(stored) + expect(decoded.slice(0, -1)).toEqual(messages.slice(0, -1)) + expect(decoded.at(-1)?.content).toEqual([{ type: "media", mediaType: "image/png", data: "AQID" }]) + const optionalMetadata = SessionProviderContext.encode(providerContext.provenance, [ + Message.make({ + role: "user", + content: [ + { type: "text", text: "attachment", metadata: { attachment: { name: undefined, source: { type: "inline" } } } }, + ], + providerMetadata: { openai: { itemId: undefined, type: "message", status: undefined, phase: undefined } }, + }), + ]) + expect(SessionProviderContext.decode(optionalMetadata)[0]).toMatchObject({ + providerMetadata: { openai: { type: "message" } }, + content: [{ metadata: { attachment: { source: { type: "inline" } } } }], + }) + expect(() => + SessionProviderContext.decode({ + ...context, + messages: [{ role: "assistant", content: [{ type: "compaction", provider: "openai" }] }], + }), + ).toThrow() +}) + +test("compatibility uses the actual deployment and endpoint rather than a catalog alias or variant", () => { + expect( + SessionProviderContext.compatible( + providerContext.provenance, + SessionProviderContext.provenance({ + ...model, + ref: { ...model.ref, id: Model.ID.make("alias"), variant: Model.VariantID.make("high") }, + }), + ), + ).toBe(true) + for (const changed of [ + { ...model, model: LanguageModel.update(model.model, { id: "other-deployment" }) }, + { + ...model, + model: LanguageModel.update(model.model, { + route: model.model.route.with({ endpoint: { baseURL: "https://another.example/v1?api-key=secret" } }), + }), + }, + { ...model, model: LanguageModel.update(model.model, { route: model.model.route.with({ id: "other-route" }) }) }, + ]) + expect( + SessionProviderContext.compatible(providerContext.provenance, SessionProviderContext.provenance(changed)), + ).toBe(false) + const privateEndpoint = SessionProviderContext.provenance({ + ...model, + model: LanguageModel.update(model.model, { + route: model.model.route.with({ endpoint: { baseURL: "https://user:secret@example.com/v1?api-key=secret" } }), + }), + }) + expect(JSON.stringify(privateEndpoint)).not.toContain("secret") + expect( + SessionProviderContext.provenance({ + ...model, + model: LanguageModel.update(model.model, { + route: model.model.route.with({ endpoint: { path: () => "/dynamic" } }), + }), + }), + ).toBeUndefined() + expect(SessionProviderContext.compatible(providerContext.provenance, undefined)).toBe(false) +}) + +it.effect( + "advances the native instruction epoch and omits superseded chronological updates after durable replay and provider switches", + () => + Effect.gen(function* () { + const s = yield* setup + yield* s.prepare + yield* s.prompt("original request") + s.state.value = "changed instructions" + yield* s.prepare + yield* s.bus.publish(SessionEvent.Compaction.Started, { sessionID, reason: "manual", recent: "" }) + const completed = yield* s.compact(providerContext) + s.state.value = "newest instructions" + yield* s.prepare + yield* s.prompt("continue") + + const verify = Effect.gen(function* () { + expect( + yield* s.db.select().from(InstructionStateTable).where(eq(InstructionStateTable.session_id, sessionID)).get(), + ).toMatchObject({ + epoch_start: completed.durable.seq, + initial_values: { "test/context": Instructions.hash("changed instructions") }, + current_values: { "test/context": Instructions.hash("newest instructions") }, + }) + const native = yield* s.load(target) + expect(native.initial).toBe("changed instructions") + expect( + toLLMMessages( + native.entries.map((entry) => entry.message), + model.ref, + ), + ).toEqual([ + ...replacement, + Message.system("newest instructions"), + expect.objectContaining({ role: "user", content: [Message.text("continue")] }), + ]) + for (const incompatible of [ + "local" as const, + { ...providerContext.provenance, modelID: "other" }, + { ...providerContext.provenance, provider: "other" }, + ]) { + const expanded = yield* s.load(incompatible) + expect(expanded.initial).toBe("changed instructions") + expect( + toLLMMessages( + expanded.entries.map((entry) => entry.message), + model.ref, + ).map((message) => message.content), + ).toEqual([ + [Message.text("original request")], + [Message.text("newest instructions")], + [Message.text("continue")], + ]) + } + const preview = yield* SessionHistory.preview(s.db, sessionID, s.instructions, target) + expect(preview.initial).toBe("changed instructions") + expect(preview.messages).toEqual(native.entries.map((entry) => entry.message)) + const store = yield* SessionStore.Service + expect((yield* store.messages({ sessionID })).map((message) => message.type)).toEqual([ + "user", + "system", + "compaction", + "system", + "user", + ]) + }) + yield* verify + const recorded = yield* s.db + .select() + .from(EventTable) + .where(eq(EventTable.aggregate_id, sessionID)) + .orderBy(asc(EventTable.seq)) + .all() + expect(recorded.filter((event) => event.data.providerContext !== undefined)).toHaveLength(1) + yield* s.bus.remove(sessionID) + yield* s.db.delete(SessionTable).where(eq(SessionTable.id, sessionID)).run() + for (const event of recorded) + yield* s.bus.replay({ + id: event.id, + created: event.created, + aggregateID: event.aggregate_id, + seq: event.seq, + type: event.type, + data: event.data, + }) + yield* verify + }), +) + +it.effect("falls back to an earlier compatible native or local checkpoint", () => + Effect.gen(function* () { + const s = yield* setup + yield* s.prepare + yield* s.prompt("before local") + s.state.value = "local baseline" + yield* s.prepare + yield* s.compact() + yield* s.prompt("after local") + yield* s.compact(providerContext) + yield* s.prompt("after native") + s.state.value = "new native baseline" + yield* s.prepare + yield* s.compact({ ...providerContext, provenance: { ...providerContext.provenance, modelID: "other" } }) + s.state.value = "post-epoch update" + yield* s.prepare + const native = yield* s.load(target) + expect(native.initial).toBe("new native baseline") + expect(native.entries.map((entry) => entry.message.type)).toEqual(["compaction", "user", "system"]) + expect(native.entries[0]?.message).toMatchObject({ providerContext }) + expect( + toLLMMessages( + native.entries.map((entry) => entry.message), + model.ref, + ).filter((message) => message.role === "system"), + ).toEqual([Message.system("post-epoch update")]) + const local = yield* s.load("local") + expect(local.initial).toBe("new native baseline") + expect(local.entries.map((entry) => entry.message.type)).toEqual(["compaction", "user", "user", "system"]) + expect(local.entries[0]?.message).toMatchObject({ summary: "local summary" }) + }), +) + +it.effect("rejects malformed persisted native windows instead of silently dropping them", () => + Effect.gen(function* () { + const s = yield* setup + yield* s.compact({ ...providerContext, messages: [{ role: "invalid", content: [] }] }) + expect(yield* SessionHistory.load(s.db, sessionID, target).pipe(Effect.flip)).toMatchObject({ + _tag: "Session.MessageDecodeError", + }) + const store = yield* SessionStore.Service + expect(yield* store.context(sessionID).pipe(Effect.flip)).toMatchObject({ _tag: "Session.MessageDecodeError" }) + }), +) diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index b08deabda78..4d68351a182 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -1,6 +1,7 @@ import { describe, expect, test } from "bun:test" import { AIError, + CompactionPart, HttpContext, LLMEvent, LLMRequest, @@ -40,6 +41,7 @@ import { SessionCompaction } from "@opencode-ai/core/session/compaction" import { SessionInbox } from "@opencode-ai/core/session/inbox" import { SessionMessage } from "@opencode-ai/core/session/message" import { SessionModelTransport } from "@opencode-ai/core/session/model-transport" +import { SessionProviderContext } from "@opencode-ai/core/session/provider-context" import { Money } from "@opencode-ai/schema/money" import { SessionProjector } from "@opencode-ai/core/session/projector" import { SessionExecution } from "@opencode-ai/core/session/execution" @@ -1435,6 +1437,92 @@ describe("SessionRunnerLLM", () => { expect(yield* s.inbox).toEqual([]) }) + scenario( + "restores installed native context with auto disabled and preserves it across fork and revert", + function* (s) { + const compaction = yield* SessionCompaction.Service + yield* compaction.transform((editor) => editor.configure({ auto: false })) + yield* s.runPrompt("Original request") + s.systemBaseline = "Checkpoint instructions" + yield* s.runPrompt("Before checkpoint") + const target = SessionProviderContext.provenance({ + model: s.currentModel, + ref: Model.Ref.make({ + id: Model.ID.make(s.currentModel.id), + providerID: Provider.ID.make(s.currentModel.provider), + }), + }) + if (!target) throw new Error("Expected concrete fixture endpoint") + const replacement = [ + Message.assistant(CompactionPart.make({ provider: s.currentModel.provider, encrypted: "checkpoint" })), + ] + const providerContext = SessionProviderContext.encode(target, replacement) + yield* s.bus.publish(SessionEvent.Compaction.Ended, { + sessionID, + reason: "manual", + text: "", + recent: "", + providerContext, + }) + const checkpoint = (yield* s.messages).find((message) => message.type === "compaction") + if (!checkpoint) throw new Error("Expected checkpoint") + + s.systemBaseline = "Newest instructions" + const after = yield* s.runPrompt("After checkpoint") + const continued = s.requests.at(-1) + if (!continued) throw new Error("Expected continuation request") + expect(continued.messages[0]).toEqual(replacement[0]) + expect(continued.system.map((part) => part.text)).toContain("Checkpoint instructions") + expect(systemTexts(continued)).toEqual(["Newest instructions"]) + + const forked = yield* s.session.fork({ sessionID, boundary: { type: "before", messageID: after.id } }) + yield* s.session.prompt({ sessionID: forked.id, text: "Fork prompt", resume: false }) + yield* s.session.resume(forked.id) + expect(s.requests.at(-1)?.messages[0]).toEqual(replacement[0]) + expect(s.requests.at(-1)?.system.map((part) => part.text)).toContain("Newest instructions") + expect(s.requests.at(-1)?.messages.filter((message) => message.role === "system")).toEqual([ + Message.system("Newest instructions"), + ]) + expect( + (yield* s.session.messages({ sessionID: forked.id })).find((message) => message.type === "compaction"), + ).toMatchObject({ providerContext }) + + const original = s.currentModel + s.currentModel = LanguageModel.update(original, { id: "different-deployment" }) + yield* s.session.prompt({ sessionID: forked.id, text: "Switched fork", resume: false }) + yield* s.session.resume(forked.id) + expect(s.requests.at(-1)?.messages[0]?.content).toEqual([Message.text("Original request")]) + expect(s.requests.at(-1)?.messages.filter((message) => message.role === "system")).toEqual([ + Message.system("Newest instructions"), + ]) + s.currentModel = original + + yield* s.bus.publish(SessionEvent.RevertEvent.Committed, { sessionID, to: checkpoint.id }) + yield* s.runPrompt("After revert") + expect( + s.requests + .at(-1) + ?.messages.flatMap((message) => message.content) + .some((part) => part.type === "compaction"), + ).toBe(false) + expect(s.requests.at(-1)?.messages[0]?.content).toEqual([Message.text("Original request")]) + expect( + (yield* s.session.messages({ sessionID: forked.id })).find((message) => message.type === "compaction"), + ).toMatchObject({ providerContext }) + + const hooks = yield* PluginHooks.Service + yield* hooks.register("session", "model.request", (event) => + Effect.sync(() => { + event.baseURL = "https://another-deployment.example/v1" + }), + ) + const before = s.requests.length + yield* s.session.prompt({ sessionID: forked.id, text: "Changed route", resume: false }) + expect(yield* s.session.resume(forked.id).pipe(Effect.exit)).toMatchObject({ _tag: "Failure" }) + expect(s.requests).toHaveLength(before) + }, + ) + scenario("seeds a fork with the parent's newest instruction values", function* (s) { yield* s.runPrompt("First") s.systemBaseline = "Changed context" diff --git a/packages/protocol/openapi.json b/packages/protocol/openapi.json index 57a0c242de2..57b5435e8fd 100644 --- a/packages/protocol/openapi.json +++ b/packages/protocol/openapi.json @@ -18355,6 +18355,9 @@ }, "recent": { "type": "string" + }, + "providerContext": { + "$ref": "#/components/schemas/Session.ProviderContext" } }, "required": ["type", "id", "time", "status", "reason", "summary", "recent"], @@ -18903,6 +18906,46 @@ "Session.Metadata": { "type": "object" }, + "Session.ProviderContext": { + "type": "object", + "properties": { + "version": { + "type": "number", + "enum": [1] + }, + "provenance": { + "$ref": "#/components/schemas/Session.ProviderContext.Provenance" + }, + "messages": {} + }, + "required": ["version", "provenance", "messages"], + "additionalProperties": false + }, + "Session.ProviderContext.Provenance": { + "type": "object", + "properties": { + "providerID": { + "type": "string" + }, + "provider": { + "type": "string" + }, + "modelID": { + "type": "string" + }, + "route": { + "type": "string" + }, + "protocol": { + "type": "string" + }, + "endpoint": { + "type": "string" + } + }, + "required": ["providerID", "provider", "modelID", "route", "protocol", "endpoint"], + "additionalProperties": false + }, "Session.Revert": { "type": "object", "properties": { diff --git a/packages/schema/src/session-event.ts b/packages/schema/src/session-event.ts index 3d597ddd886..d82ad08a0aa 100644 --- a/packages/schema/src/session-event.ts +++ b/packages/schema/src/session-event.ts @@ -587,6 +587,7 @@ export namespace Compaction { reason: Started.data.fields.reason, model: SessionMessage.CompactionCompleted.fields.model, providerState: SessionMessage.CompactionCompleted.fields.providerState, + providerContext: SessionMessage.CompactionCompleted.fields.providerContext, text: Schema.String, recent: Schema.String, }, diff --git a/packages/schema/src/session-message.ts b/packages/schema/src/session-message.ts index c3f2ae30dd6..6bd37eb5174 100644 --- a/packages/schema/src/session-message.ts +++ b/packages/schema/src/session-message.ts @@ -1,6 +1,7 @@ export * as SessionMessage from "./session-message.js" import { Schema } from "effect" +import { SessionProviderContext } from "./session-provider-context.js" import { optional } from "./schema.js" import { Content } from "./tool.js" import { Location } from "./location.js" @@ -254,6 +255,7 @@ export const CompactionCompleted = Schema.Struct({ providerState: ProviderState.pipe(optional), summary: Schema.String, recent: Schema.String, + providerContext: SessionProviderContext.Info.pipe(optional), }).annotate({ identifier: "Session.Message.Compaction.Completed" }) export interface CompactionFailed extends Schema.Schema.Type {} diff --git a/packages/schema/src/session-provider-context.ts b/packages/schema/src/session-provider-context.ts new file mode 100644 index 00000000000..a1709fccd80 --- /dev/null +++ b/packages/schema/src/session-provider-context.ts @@ -0,0 +1,24 @@ +export * as SessionProviderContext from "./session-provider-context.js" + +import { Schema } from "effect" +import { Provider } from "./provider.js" + +/** Exact producing model/deployment and route identity, never credentials or a connection ID. */ +export interface Provenance extends Schema.Schema.Type {} +export const Provenance = Schema.Struct({ + providerID: Provider.ID, + provider: Schema.String, + modelID: Schema.String, + route: Schema.String, + protocol: Schema.String, + /** Digest of the configured endpoint; raw URLs and query values are not persisted. */ + endpoint: Schema.String, +}).annotate({ identifier: "Session.ProviderContext.Provenance" }) + +/** Core validates the versioned canonical AI Message[] payload on installation and replay. */ +export interface Info extends Schema.Schema.Type {} +export const Info = Schema.Struct({ + version: Schema.Literal(1), + provenance: Provenance, + messages: Schema.Json, +}).annotate({ identifier: "Session.ProviderContext" }) diff --git a/packages/schema/test/session-terminal.test.ts b/packages/schema/test/session-terminal.test.ts index ef5ef79ae8e..3659c17a5e4 100644 --- a/packages/schema/test/session-terminal.test.ts +++ b/packages/schema/test/session-terminal.test.ts @@ -48,3 +48,25 @@ test("failed steps only override the assistant finish for content filters", () = }) expect(() => decode({ ...input, finish: "stop" })).toThrow() }) + +test("provider compaction context is optional, versioned and JSON-only", () => { + const decode = Schema.decodeUnknownSync(SessionEvent.Compaction.Ended.data) + const encode = Schema.encodeSync(SessionEvent.Compaction.Ended.data) + const local = { sessionID: "ses_context", reason: "manual" as const, text: "summary", recent: "" } + expect(encode({ ...decode(local), providerContext: undefined })).toEqual(local) + const providerContext = { + version: 1 as const, + provenance: { + providerID: "openai", + provider: "openai", + modelID: "deployment", + route: "responses", + protocol: "responses", + endpoint: "digest", + }, + messages: [{ role: "assistant", content: [{ type: "compaction", provider: "openai", encrypted: "opaque" }] }], + } + expect(encode(decode({ ...local, providerContext }))).toEqual({ ...local, providerContext }) + expect(() => decode({ ...local, providerContext: { ...providerContext, version: 2 } })).toThrow() + expect(() => decode({ ...local, providerContext: { ...providerContext, messages: [() => "invalid"] } })).toThrow() +}) diff --git a/packages/www/openapi.json b/packages/www/openapi.json index 57a0c242de2..57b5435e8fd 100644 --- a/packages/www/openapi.json +++ b/packages/www/openapi.json @@ -18355,6 +18355,9 @@ }, "recent": { "type": "string" + }, + "providerContext": { + "$ref": "#/components/schemas/Session.ProviderContext" } }, "required": ["type", "id", "time", "status", "reason", "summary", "recent"], @@ -18903,6 +18906,46 @@ "Session.Metadata": { "type": "object" }, + "Session.ProviderContext": { + "type": "object", + "properties": { + "version": { + "type": "number", + "enum": [1] + }, + "provenance": { + "$ref": "#/components/schemas/Session.ProviderContext.Provenance" + }, + "messages": {} + }, + "required": ["version", "provenance", "messages"], + "additionalProperties": false + }, + "Session.ProviderContext.Provenance": { + "type": "object", + "properties": { + "providerID": { + "type": "string" + }, + "provider": { + "type": "string" + }, + "modelID": { + "type": "string" + }, + "route": { + "type": "string" + }, + "protocol": { + "type": "string" + }, + "endpoint": { + "type": "string" + } + }, + "required": ["providerID", "provider", "modelID", "route", "protocol", "endpoint"], + "additionalProperties": false + }, "Session.Revert": { "type": "object", "properties": { diff --git a/packages/www/public/openapi.json b/packages/www/public/openapi.json index 57a0c242de2..57b5435e8fd 100644 --- a/packages/www/public/openapi.json +++ b/packages/www/public/openapi.json @@ -18355,6 +18355,9 @@ }, "recent": { "type": "string" + }, + "providerContext": { + "$ref": "#/components/schemas/Session.ProviderContext" } }, "required": ["type", "id", "time", "status", "reason", "summary", "recent"], @@ -18903,6 +18906,46 @@ "Session.Metadata": { "type": "object" }, + "Session.ProviderContext": { + "type": "object", + "properties": { + "version": { + "type": "number", + "enum": [1] + }, + "provenance": { + "$ref": "#/components/schemas/Session.ProviderContext.Provenance" + }, + "messages": {} + }, + "required": ["version", "provenance", "messages"], + "additionalProperties": false + }, + "Session.ProviderContext.Provenance": { + "type": "object", + "properties": { + "providerID": { + "type": "string" + }, + "provider": { + "type": "string" + }, + "modelID": { + "type": "string" + }, + "route": { + "type": "string" + }, + "protocol": { + "type": "string" + }, + "endpoint": { + "type": "string" + } + }, + "required": ["providerID", "provider", "modelID", "route", "protocol", "endpoint"], + "additionalProperties": false + }, "Session.Revert": { "type": "object", "properties": {