From 994f55423a434f827e2b1dd5a75f2ac5d613a3cf Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Thu, 2 Jul 2026 09:48:02 -0400 Subject: [PATCH] refactor(core): resolve attachments durably at promotion --- packages/client/src/generated/types.ts | 6 + packages/core/src/session/compaction.ts | 8 +- packages/core/src/session/input.ts | 22 ++- packages/core/src/session/message-updater.ts | 6 +- .../core/src/session/runner/attachment.ts | 88 +++------ packages/core/src/session/runner/llm.ts | 59 +++--- .../core/src/session/runner/to-llm-message.ts | 30 +++- packages/core/test/session-create.test.ts | 20 ++- packages/core/test/session-prompt.test.ts | 10 +- .../test/session-runner-attachment.test.ts | 169 ++++++------------ packages/core/test/session-runner.test.ts | 35 +++- packages/schema/src/session-event.ts | 16 +- packages/schema/src/session-message.ts | 13 +- specs/v2/session.md | 2 +- 14 files changed, 242 insertions(+), 242 deletions(-) diff --git a/packages/client/src/generated/types.ts b/packages/client/src/generated/types.ts index 1999704bcc..c37beb6290 100644 --- a/packages/client/src/generated/types.ts +++ b/packages/client/src/generated/types.ts @@ -672,6 +672,7 @@ export type SessionContextOutput = { readonly name?: string readonly description?: string readonly source?: { readonly start: number; readonly end: number; readonly text: string } + readonly resolved?: string }> readonly agents?: ReadonlyArray<{ readonly name: string @@ -901,6 +902,7 @@ export type SessionHistoryOutput = { }> } readonly delivery: "steer" | "queue" + readonly resolutions?: ReadonlyArray<{ readonly uri: string; readonly resolved: string }> } } | { @@ -1396,6 +1398,7 @@ export type SessionEventsOutput = }> } readonly delivery: "steer" | "queue" + readonly resolutions?: ReadonlyArray<{ readonly uri: string; readonly resolved: string }> } } | { @@ -1837,6 +1840,7 @@ export type SessionMessageOutput = { readonly name?: string readonly description?: string readonly source?: { readonly start: number; readonly end: number; readonly text: string } + readonly resolved?: string }> readonly agents?: ReadonlyArray<{ readonly name: string @@ -2018,6 +2022,7 @@ export type MessageListOutput = { readonly name?: string readonly description?: string readonly source?: { readonly start: number; readonly end: number; readonly text: string } + readonly resolved?: string }> readonly agents?: ReadonlyArray<{ readonly name: string @@ -3499,6 +3504,7 @@ export type EventSubscribeOutput = }> } readonly delivery: "steer" | "queue" + readonly resolutions?: ReadonlyArray<{ readonly uri: string; readonly resolved: string }> } } | { diff --git a/packages/core/src/session/compaction.ts b/packages/core/src/session/compaction.ts index 99c45cbb52..c9d3c11cd8 100644 --- a/packages/core/src/session/compaction.ts +++ b/packages/core/src/session/compaction.ts @@ -108,7 +108,13 @@ export const serializeToolContent = (content: SessionMessage.ToolStateCompleted[ const serialize = (message: SessionMessage.Message) => { if (message.type === "user") { - const files = message.files?.map((file) => `[Attached ${file.mime}: ${file.name ?? file.uri}]`) ?? [] + // Resolved text attachments carry the content the model actually saw; media stays a placeholder. + const files = + message.files?.map((file) => + file.resolved !== undefined && !file.resolved.startsWith("data:") + ? truncate(file.resolved) + : `[Attached ${file.mime}: ${file.name ?? file.uri}]`, + ) ?? [] return [`[User]: ${message.text}`, ...files].join("\n") } if (message.type === "assistant") { diff --git a/packages/core/src/session/input.ts b/packages/core/src/session/input.ts index 14b613678d..e2c6135ad3 100644 --- a/packages/core/src/session/input.ts +++ b/packages/core/src/session/input.ts @@ -213,21 +213,35 @@ const matchesProjection = ( equivalent(input, expected) && DateTime.toEpochMillis(input.timeCreated) === DateTime.toEpochMillis(expected.timeCreated) +/** + * Captures model-visible attachment content while an input is promoted, so + * projection replay stays deterministic without filesystem access. Resolution + * must not fail; unreadable attachments resolve to a model-visible note. + */ +export type Resolver = (prompt: Prompt) => Effect.Effect> + +/** Promote without capturing attachment content, e.g. in tests without a Location filesystem. */ +export const unresolved: Resolver = () => Effect.succeed([]) + const publish = Effect.fn("SessionInput.publish")(function* ( db: DatabaseService, events: EventV2.Interface, sessionID: SessionSchema.ID, + resolve: Resolver, rows: ReadonlyArray, ) { for (const row of rows) { const id = SessionMessage.ID.make(row.id) + const prompt = decodePrompt(row.prompt) + const resolutions = yield* resolve(prompt) yield* events .publish(SessionEvent.Prompted, { sessionID, timestamp: DateTime.makeUnsafe(row.time_created), messageID: id, - prompt: decodePrompt(row.prompt), + prompt, delivery: row.delivery, + ...(resolutions.length === 0 ? {} : { resolutions }), }) .pipe( Effect.catchDefect((defect) => @@ -247,6 +261,7 @@ export const promoteSteers = Effect.fn("SessionInput.promoteSteers")(function* ( events: EventV2.Interface, sessionID: SessionSchema.ID, cutoff: number, + resolve: Resolver, ) { const rows = yield* db .select() @@ -262,13 +277,14 @@ export const promoteSteers = Effect.fn("SessionInput.promoteSteers")(function* ( .orderBy(asc(SessionInputTable.admitted_seq)) .all() .pipe(Effect.orDie) - return yield* publish(db, events, sessionID, rows) + return yield* publish(db, events, sessionID, resolve, rows) }) export const promoteNextQueued = Effect.fn("SessionInput.promoteNextQueued")(function* ( db: DatabaseService, events: EventV2.Interface, sessionID: SessionSchema.ID, + resolve: Resolver, ) { const row = yield* db .select() @@ -284,5 +300,5 @@ export const promoteNextQueued = Effect.fn("SessionInput.promoteNextQueued")(fun .limit(1) .get() .pipe(Effect.orDie) - return row === undefined ? false : yield* publish(db, events, sessionID, [row]).pipe(Effect.as(true)) + return row === undefined ? false : yield* publish(db, events, sessionID, resolve, [row]).pipe(Effect.as(true)) }) diff --git a/packages/core/src/session/message-updater.ts b/packages/core/src/session/message-updater.ts index e12c24162c..45171cca2f 100644 --- a/packages/core/src/session/message-updater.ts +++ b/packages/core/src/session/message-updater.ts @@ -126,13 +126,17 @@ export function update(adapter: Adapter, event: SessionEvent.Event) { "session.next.renamed": () => Effect.void, "session.next.forked": () => Effect.void, "session.next.prompted": (event) => { + const resolved = new Map(event.data.resolutions?.map((item) => [item.uri, item.resolved])) return adapter.appendMessage( SessionMessage.User.make({ id: event.data.messageID, type: "user", metadata: event.metadata, text: event.data.prompt.text, - files: event.data.prompt.files, + files: event.data.prompt.files?.map((file) => { + const content = resolved.get(file.uri) + return content === undefined ? file : { ...file, resolved: content } + }), agents: event.data.prompt.agents, time: { created: event.data.timestamp }, }), diff --git a/packages/core/src/session/runner/attachment.ts b/packages/core/src/session/runner/attachment.ts index 8a4e233c44..887753107c 100644 --- a/packages/core/src/session/runner/attachment.ts +++ b/packages/core/src/session/runner/attachment.ts @@ -5,8 +5,8 @@ import { Effect } from "effect" import { Image } from "../../image" import { AbsolutePath } from "../../schema" import { ReadToolFileSystem } from "../../tool/read-filesystem" -import { SessionMessage } from "../message" -import type { FileAttachment } from "../prompt" +import { SessionEvent } from "../event" +import type { FileAttachment, Prompt } from "../prompt" export interface Services { readonly reader: ReadToolFileSystem.Interface @@ -14,57 +14,29 @@ export interface Services { } /** - * One drain's attachment materialization results, keyed by message ID and URI. - * Reusing it across the drain's turns avoids re-reading attachments from disk - * and pins their content for the drain, so mid-drain file edits neither rewrite - * history the model already saw nor invalidate the provider prompt-cache prefix. - */ -export type Cache = Map - -interface Materialized { - readonly file?: FileAttachment - readonly expansion?: string -} - -/** - * Materialize local `file:` attachments during per-turn request assembly. + * Capture model-visible content for a prompt's local `file:` attachments at + * promotion time, so the durable user message snapshots what the user attached. * * Providers accept media content only for a narrow set of mimes, so lowering an * unresolved `file:` URI (or an `application/x-directory` attachment) as a media - * part fails the provider turn. Directories become an inline listing, text files - * become inline content, and images are re-encoded as normalized data URLs. - * Other URI schemes (data URLs, MCP resources) pass through unchanged, and - * unreadable attachments degrade to a model-visible note instead of failing the - * turn. The durable projected message is never modified. + * part fails the provider turn. Directories resolve to an inline listing, text + * files to inline content, and images to normalized data URLs. Every result is + * bounded: reads cap at `MAX_READ_BYTES`/`MAX_READ_LINES`, listings at + * `MAX_READ_LINES` entries, and images at the configured normalization limit. + * Other URI schemes (data URLs, MCP resources) are skipped, and unreadable + * attachments resolve to a model-visible note instead of failing promotion. */ -export const materialize = Effect.fn("SessionRunnerAttachment.materialize")(function* ( +export const resolutions = Effect.fn("SessionRunnerAttachment.resolutions")(function* ( services: Services, - cache: Cache, - messages: readonly SessionMessage.Message[], + prompt: Prompt, ) { - if (!messages.some((message) => message.type === "user" && message.files?.some(local))) return messages - return yield* Effect.forEach(messages, (message) => { - if (message.type !== "user" || !message.files?.some(local)) return Effect.succeed(message) - return Effect.forEach(message.files, (file) => { - const key = `${message.id}:${file.uri}` - const hit = cache.get(key) - if (hit) return Effect.succeed(hit) - return materializeFile(services, file).pipe(Effect.tap((result) => Effect.sync(() => cache.set(key, result)))) - }).pipe( - Effect.map( - (results): SessionMessage.User => ({ - ...message, - text: [ - message.text, - ...results.flatMap((result) => (result.expansion === undefined ? [] : [result.expansion])), - ] - .filter(Boolean) - .join("\n\n"), - files: results.flatMap((result) => (result.file === undefined ? [] : [result.file])), - }), - ), - ) - }) + const locals = (prompt.files ?? []).filter(local) + const unique = [...new Map(locals.map((file) => [file.uri, file])).values()] + return yield* Effect.forEach(unique, (file) => + resolve(services, file).pipe( + Effect.map((resolved) => SessionEvent.AttachmentResolution.make({ uri: file.uri, resolved })), + ), + ) }) const local = (file: FileAttachment) => file.uri.startsWith("file:") @@ -79,9 +51,8 @@ const pageFromRange = (url: URL) => { return { offset: start, ...(end >= start ? { limit: end - start + 1 } : {}) } } -const materializeFile = (services: Services, file: FileAttachment) => { - if (!local(file)) return Effect.succeed({ file }) - return Effect.gen(function* () { +const resolve = (services: Services, file: FileAttachment) => + Effect.gen(function* () { const { target, page } = yield* Effect.try({ try: () => { const url = new URL(file.uri) @@ -100,25 +71,18 @@ const materializeFile = (services: Services, file: FileAttachment) => { ...listing.entries.map((entry) => entry.path), ...(listing.truncated ? ["(listing truncated)"] : []), ] - return { expansion: wrap("attached-directory", display, lines.join("\n")) } satisfies Materialized + return wrap("attached-directory", display, lines.join("\n")) } const content = yield* services.reader.read(target, display, page) if (content instanceof ReadToolFileSystem.TextPage) { const truncated = content.truncated ? "\n(content truncated)" : "" - return { expansion: wrap("attached-file", display, content.content + truncated) } satisfies Materialized + return wrap("attached-file", display, content.content + truncated) } if (content.encoding === "base64") { const normalized = yield* services.image .normalize(display, { ...content, encoding: "base64" }) .pipe(Effect.catchTag("Image.ResizerUnavailableError", () => Effect.succeed(content))) - return { - file: { ...file, uri: `data:${normalized.mime};base64,${normalized.content}`, mime: normalized.mime }, - } satisfies Materialized + return `data:${normalized.mime};base64,${normalized.content}` } - return { expansion: wrap("attached-file", display, content.content) } satisfies Materialized - }).pipe( - Effect.catch((error) => - Effect.succeed({ expansion: wrap("attachment-unavailable", file.name ?? file.uri, error.message) }), - ), - ) -} + return wrap("attached-file", display, content.content) + }).pipe(Effect.catch((error) => Effect.succeed(wrap("attachment-unavailable", file.name ?? file.uri, error.message)))) diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index b382e22929..617c6ab5e3 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -106,6 +106,8 @@ const layer = Layer.effect( reader: yield* ReadToolFileSystem.Service, image: yield* Image.Service, } + const resolveAttachments: SessionInput.Resolver = (prompt) => + SessionRunnerAttachment.resolutions(attachments, prompt) const models = yield* SessionRunnerModel.Service const store = yield* SessionStore.Service const location = yield* Location.Service @@ -185,7 +187,6 @@ const layer = Layer.effect( sessionID: SessionSchema.ID, promotion: SessionInput.Delivery | undefined, step: number, - attachmentCache: SessionRunnerAttachment.Cache, recoverOverflow?: typeof compaction.compactAfterOverflow, ) { const session = yield* getSession(sessionID) @@ -199,10 +200,11 @@ const layer = Layer.effect( if (promotion) { const cutoff = yield* EventV2.latestSequence(db, session.id) let promoted = 0 - if (promotion === "steer") promoted = yield* SessionInput.promoteSteers(db, events, session.id, cutoff) + if (promotion === "steer") + promoted = yield* SessionInput.promoteSteers(db, events, session.id, cutoff, resolveAttachments) if (promotion === "queue") { - promoted += Number(yield* SessionInput.promoteNextQueued(db, events, session.id)) - promoted += yield* SessionInput.promoteSteers(db, events, session.id, cutoff) + promoted += Number(yield* SessionInput.promoteNextQueued(db, events, session.id, resolveAttachments)) + promoted += yield* SessionInput.promoteSteers(db, events, session.id, cutoff, resolveAttachments) } if (promoted > 0) currentStep = 1 } @@ -211,8 +213,6 @@ const layer = Layer.effect( const model = yield* models.resolve(session) const entries = yield* SessionHistory.entriesForRunner(db, session.id, system.baselineSeq) const context = entries.map((entry) => entry.message) - // Expand local file/directory attachments for this request only; durable history keeps the URIs. - const materialized = yield* SessionRunnerAttachment.materialize(attachments, attachmentCache, context) const isLastStep = agent.info?.steps !== undefined && currentStep >= agent.info.steps const toolMaterialization = isLastStep ? undefined @@ -224,11 +224,11 @@ const layer = Layer.effect( system: [agent.info?.system ? agent.info.system : SessionRunnerSystemPrompt.provider(model), system.baseline] .filter((part): part is string => part !== undefined && part.length > 0) .map(SystemPart.make), - messages: [...toLLMMessages(materialized, model), ...(isLastStep ? [Message.assistant(MAX_STEPS_PROMPT)] : [])], + messages: [...toLLMMessages(context, model), ...(isLastStep ? [Message.assistant(MAX_STEPS_PROMPT)] : [])], tools: toolMaterialization?.definitions ?? [], toolChoice: isLastStep ? "none" : undefined, }) - if (yield* compaction.compactIfNeeded({ sessionID: session.id, messages: materialized, request })) + if (yield* compaction.compactIfNeeded({ sessionID: session.id, messages: context, request })) return yield* Effect.die(continueAfterCompaction(currentStep)) const startSnapshot = yield* snapshots.capture() const publisher = createLLMEventPublisher(events, { @@ -301,7 +301,7 @@ const layer = Layer.effect( recoverOverflow && !publisher.hasAssistantStarted() && isContextOverflowFailure(overflowFailure ?? failure) && - (yield* restore(recoverOverflow({ sessionID: session.id, messages: materialized, request }))) + (yield* restore(recoverOverflow({ sessionID: session.id, messages: context, request }))) ) return yield* Effect.die(continueAfterOverflowCompaction(currentStep)) if (overflowFailure) yield* publish(overflowFailure) @@ -367,34 +367,31 @@ const layer = Layer.effect( sessionID: SessionSchema.ID, promotion: SessionInput.Delivery | undefined, step: number, - attachmentCache: SessionRunnerAttachment.Cache, ) => Effect.Effect<{ readonly needsContinuation: boolean; readonly step: number }, RunError> - const runAfterOverflowCompaction: RunTurn = Effect.fnUntraced( - function* (sessionID, promotion, step, attachmentCache) { - return yield* runTurnAttempt(sessionID, promotion, step, attachmentCache).pipe( - Effect.catchDefect( - Effect.fnUntraced(function* (defect) { - if (!(defect instanceof TurnTransitionError)) return yield* Effect.die(defect) - if (defect.transition._tag === "ContinueAfterOverflowCompaction") - return yield* Effect.die("Post-compaction provider attempt cannot recover another overflow") - yield* Effect.yieldNow - return yield* runAfterOverflowCompaction(sessionID, undefined, defect.transition.step, attachmentCache) - }), - ), - ) - }, - ) + const runAfterOverflowCompaction: RunTurn = Effect.fnUntraced(function* (sessionID, promotion, step) { + return yield* runTurnAttempt(sessionID, promotion, step).pipe( + Effect.catchDefect( + Effect.fnUntraced(function* (defect) { + if (!(defect instanceof TurnTransitionError)) return yield* Effect.die(defect) + if (defect.transition._tag === "ContinueAfterOverflowCompaction") + return yield* Effect.die("Post-compaction provider attempt cannot recover another overflow") + yield* Effect.yieldNow + return yield* runAfterOverflowCompaction(sessionID, undefined, defect.transition.step) + }), + ), + ) + }) - const runTurn: RunTurn = Effect.fnUntraced(function* (sessionID, promotion, step, attachmentCache) { - return yield* runTurnAttempt(sessionID, promotion, step, attachmentCache, compaction.compactAfterOverflow).pipe( + const runTurn: RunTurn = Effect.fnUntraced(function* (sessionID, promotion, step) { + return yield* runTurnAttempt(sessionID, promotion, step, compaction.compactAfterOverflow).pipe( Effect.catchDefect( Effect.fnUntraced(function* (defect) { if (!(defect instanceof TurnTransitionError)) return yield* Effect.die(defect) yield* Effect.yieldNow if (defect.transition._tag === "ContinueAfterOverflowCompaction") - return yield* runAfterOverflowCompaction(sessionID, undefined, defect.transition.step, attachmentCache) - return yield* runTurn(sessionID, undefined, defect.transition.step, attachmentCache) + return yield* runAfterOverflowCompaction(sessionID, undefined, defect.transition.step) + return yield* runTurn(sessionID, undefined, defect.transition.step) }), ), ) @@ -408,15 +405,13 @@ const layer = Layer.effect( const hasQueue = hasSteer ? false : yield* SessionInput.hasPending(db, input.sessionID, "queue") if (!input.force && !hasSteer && !hasQueue) return yield* failInterruptedTools(input.sessionID) - // One attachment snapshot per drain: repeated turns reuse materialized content. - const attachmentCache: SessionRunnerAttachment.Cache = new Map() let promotion: SessionInput.Delivery | undefined = hasSteer ? "steer" : hasQueue ? "queue" : undefined let shouldRun = input.force || hasSteer || hasQueue while (shouldRun) { let needsContinuation = true let step = 1 while (needsContinuation) { - const result = yield* runTurn(input.sessionID, promotion, step, attachmentCache) + const result = yield* runTurn(input.sessionID, promotion, step) // Steer/queue promotion inside runTurn has already made the pending input a visible // user message by this point, so the first-user-message check below is reliable. if (!titleAttempted.has(input.sessionID)) { diff --git a/packages/core/src/session/runner/to-llm-message.ts b/packages/core/src/session/runner/to-llm-message.ts index 8ddd7cce5c..9e1409f9ac 100644 --- a/packages/core/src/session/runner/to-llm-message.ts +++ b/packages/core/src/session/runner/to-llm-message.ts @@ -8,15 +8,27 @@ import { type ProviderMetadata, } from "@opencode-ai/llm" import { SessionMessage } from "../message" -import type { FileAttachment } from "../prompt" -const media = (file: FileAttachment): ContentPart => ({ - type: "media", - mediaType: file.mime, - data: file.uri, - filename: file.name, - metadata: file.description === undefined ? undefined : { description: file.description }, -}) +// Attachments carry promotion-time `resolved` content: a data URL for media, +// model-visible text otherwise. Unresolved `file:` URIs cannot be lowered as +// media (providers reject the mime or the non-data payload), so they degrade +// to a model-visible note instead of failing the provider turn. +const attachment = (file: SessionMessage.UserFile): ContentPart => { + if (file.resolved !== undefined && !file.resolved.startsWith("data:")) return { type: "text", text: file.resolved } + const uri = file.resolved ?? file.uri + if (uri.startsWith("file:")) + return { + type: "text", + text: `\nAttachment was not captured; read it with tools if needed.\n`, + } + return { + type: "media", + mediaType: uri.match(/^data:([^;,]+)[;,]/i)?.[1] ?? file.mime, + data: uri, + filename: file.name, + metadata: file.description === undefined ? undefined : { description: file.description }, + } +} const toolInput = (tool: SessionMessage.AssistantTool) => { if (tool.state.status !== "pending") return tool.state.input @@ -122,7 +134,7 @@ function toLLMMessage(message: SessionMessage.Message, model: Model): Message[] Message.make({ id: message.id, role: "user", - content: [{ type: "text", text: message.text }, ...(message.files ?? []).map(media)], + content: [{ type: "text", text: message.text }, ...(message.files ?? []).map(attachment)], metadata: { ...message.metadata, ...(message.agents?.length ? { agents: message.agents } : {}), diff --git a/packages/core/test/session-create.test.ts b/packages/core/test/session-create.test.ts index 5554f9782a..d6ca8c9108 100644 --- a/packages/core/test/session-create.test.ts +++ b/packages/core/test/session-create.test.ts @@ -135,7 +135,7 @@ describe("SessionV2.create", () => { prompt: Prompt.make({ text: "First" }), resume: false, }) - yield* SessionInput.promoteSteers(db, events, parent.id, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers(db, events, parent.id, Number.MAX_SAFE_INTEGER, SessionInput.unresolved) yield* events.publish(SessionEvent.Synthetic, { sessionID: parent.id, messageID: SessionMessage.ID.create(), @@ -167,9 +167,9 @@ describe("SessionV2.create", () => { }) yield* session.prompt({ sessionID: parent.id, prompt: Prompt.make({ text: "Parent changed" }), resume: false }) - yield* SessionInput.promoteSteers(db, events, parent.id, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers(db, events, parent.id, Number.MAX_SAFE_INTEGER, SessionInput.unresolved) yield* session.prompt({ sessionID: forked.id, prompt: Prompt.make({ text: "Child continues" }), resume: false }) - yield* SessionInput.promoteSteers(db, events, forked.id, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers(db, events, forked.id, Number.MAX_SAFE_INTEGER, SessionInput.unresolved) expect((yield* session.context(parent.id)).map((message) => message.type)).toEqual(["user", "synthetic", "user"]) expect((yield* session.context(forked.id)).map((message) => message.type)).toEqual(["user", "synthetic", "user"]) @@ -192,13 +192,13 @@ describe("SessionV2.create", () => { prompt: Prompt.make({ text: "First" }), resume: false, }) - yield* SessionInput.promoteSteers(db, events, parent.id, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers(db, events, parent.id, Number.MAX_SAFE_INTEGER, SessionInput.unresolved) const second = yield* session.prompt({ sessionID: parent.id, prompt: Prompt.make({ text: "Second" }), resume: false, }) - yield* SessionInput.promoteSteers(db, events, parent.id, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers(db, events, parent.id, Number.MAX_SAFE_INTEGER, SessionInput.unresolved) const forked = yield* session.fork({ sessionID: parent.id, messageID: second.id }) @@ -314,7 +314,7 @@ describe("SessionV2.create", () => { const { db } = yield* Database.Service const created = yield* session.create({ location }) yield* session.prompt({ sessionID: created.id, prompt: Prompt.make({ text: "Hello" }), resume: false }) - yield* SessionInput.promoteSteers(db, events, created.id, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers(db, events, created.id, Number.MAX_SAFE_INTEGER, SessionInput.unresolved) expect( Array.from(yield* session.events({ sessionID: created.id }).pipe(Stream.take(2), Stream.runCollect)), @@ -336,7 +336,13 @@ describe("SessionV2.create", () => { prompt: Prompt.make({ text: "Replay lifecycle" }), resume: false, }) - yield* SessionInput.promoteSteers(sourceDb, sourceEvents, created.id, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers( + sourceDb, + sourceEvents, + created.id, + Number.MAX_SAFE_INTEGER, + SessionInput.unresolved, + ) const serialized = (yield* sourceDb .select() .from(EventTable) diff --git a/packages/core/test/session-prompt.test.ts b/packages/core/test/session-prompt.test.ts index 8eb8fb4a27..2e37b97ec3 100644 --- a/packages/core/test/session-prompt.test.ts +++ b/packages/core/test/session-prompt.test.ts @@ -197,7 +197,7 @@ describe("SessionV2.prompt", () => { prompt: Prompt.make({ text: "boundary" }), resume: false, }) - yield* SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER, SessionInput.unresolved) const stale = SessionMessage.ID.make("msg_stale_assistant") yield* db.insert(SessionMessageTable).values(assistantRow(stale, 100)).run().pipe(Effect.orDie) yield* events.publish(SessionEvent.RevertEvent.Staged, { @@ -250,7 +250,7 @@ describe("SessionV2.prompt", () => { yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "First" }), resume: false }) yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Second" }), resume: false }) - yield* SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER, SessionInput.unresolved) const streamed = Array.from(yield* Fiber.join(fiber)) expect(streamed.map((event) => [event.durable?.seq, event.type])).toEqual([ @@ -425,8 +425,8 @@ describe("SessionV2.prompt", () => { yield* Effect.all( [ - SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER), - SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER), + SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER, SessionInput.unresolved), + SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER, SessionInput.unresolved), ], { concurrency: "unbounded" }, ) @@ -449,7 +449,7 @@ describe("SessionV2.prompt", () => { const cutoff = first.admittedSeq const second = yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "After cutoff" }), resume: false }) - yield* SessionInput.promoteSteers(db, events, sessionID, cutoff) + yield* SessionInput.promoteSteers(db, events, sessionID, cutoff, SessionInput.unresolved) expect(yield* admitted(first.id)).toHaveProperty("promotedSeq") expect(yield* admitted(second.id)).not.toHaveProperty("promotedSeq") diff --git a/packages/core/test/session-runner-attachment.test.ts b/packages/core/test/session-runner-attachment.test.ts index 8a66656ac7..84b6ae47a9 100644 --- a/packages/core/test/session-runner-attachment.test.ts +++ b/packages/core/test/session-runner-attachment.test.ts @@ -1,29 +1,17 @@ import { describe, expect } from "bun:test" import path from "path" import { pathToFileURL } from "url" -import { DateTime, Effect, FileSystem } from "effect" +import { Effect, FileSystem } from "effect" import { LayerNodePlatform } from "@opencode-ai/core/effect/app-node-platform" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { Image } from "@opencode-ai/core/image" -import { SessionV2 } from "@opencode-ai/core/session" -import { SessionMessage } from "@opencode-ai/core/session/message" -import { FileAttachment } from "@opencode-ai/core/session/prompt" +import { Prompt } from "@opencode-ai/core/session/prompt" import { SessionRunnerAttachment } from "@opencode-ai/core/session/runner/attachment" import { ReadToolFileSystem } from "@opencode-ai/core/tool/read-filesystem" import { testEffect } from "./lib/effect" const it = testEffect(LayerNode.compile(LayerNode.group([ReadToolFileSystem.node, LayerNodePlatform.filesystem]))) -const created = DateTime.makeUnsafe(0) -const user = (files: FileAttachment[]) => - SessionMessage.User.make({ - id: SessionMessage.ID.make("msg_user"), - type: "user", - text: "Look at this", - files, - time: { created }, - }) - // The resizer-unavailable stub exercises the raw-content fallback deterministically. const image = Image.Service.of({ normalize: () => Effect.fail(new Image.ResizerUnavailableError()) }) @@ -34,50 +22,42 @@ const fixture = Effect.gen(function* () { return { services, files, directory } }) -const requireUser = (message: SessionMessage.Message) => { - if (message.type !== "user") throw new Error(`Expected a user message, got ${message.type}`) - return message -} +const prompt = (files: NonNullable) => Prompt.make({ text: "Look at this", files }) -describe("SessionRunnerAttachment.materialize", () => { - it.effect("expands a directory attachment into a listing instead of media", () => +describe("SessionRunnerAttachment.resolutions", () => { + it.effect("resolves a directory attachment to a listing", () => Effect.gen(function* () { const { services, files, directory } = yield* fixture yield* files.makeDirectory(path.join(directory, "src")) yield* files.writeFileString(path.join(directory, "package.json"), "{}") - const attachment = FileAttachment.make({ - uri: pathToFileURL(directory + path.sep).href, - mime: "application/x-directory", - name: "project/", - }) + const uri = pathToFileURL(directory + path.sep).href - const result = yield* SessionRunnerAttachment.materialize(services, new Map(), [user([attachment])]) + const result = yield* SessionRunnerAttachment.resolutions( + services, + prompt([{ uri, mime: "application/x-directory", name: "project/" }]), + ) - const message = requireUser(result[0]) - expect(message.files).toEqual([]) - expect(message.text).toContain('') - expect(message.text).toContain("src/") - expect(message.text).toContain("package.json") + expect(result).toHaveLength(1) + expect(result[0].uri).toBe(uri) + expect(result[0].resolved).toContain('') + expect(result[0].resolved).toContain("src/") + expect(result[0].resolved).toContain("package.json") }), ) - it.effect("expands a text file attachment into inline content", () => + it.effect("resolves a text file attachment to inline content", () => Effect.gen(function* () { const { services, files, directory } = yield* fixture const file = path.join(directory, "notes.md") yield* files.writeFileString(file, "first line\nsecond line\nthird line\n") - const attachment = FileAttachment.make({ - uri: pathToFileURL(file).href, - mime: "text/markdown", - name: "notes.md", - }) - const result = yield* SessionRunnerAttachment.materialize(services, new Map(), [user([attachment])]) + const result = yield* SessionRunnerAttachment.resolutions( + services, + prompt([{ uri: pathToFileURL(file).href, mime: "text/markdown", name: "notes.md" }]), + ) - const message = requireUser(result[0]) - expect(message.files).toEqual([]) - expect(message.text).toContain('') - expect(message.text).toContain("second line") + expect(result[0].resolved).toContain('') + expect(result[0].resolved).toContain("second line") }), ) @@ -86,103 +66,68 @@ describe("SessionRunnerAttachment.materialize", () => { const { services, files, directory } = yield* fixture const file = path.join(directory, "notes.md") yield* files.writeFileString(file, "first line\nsecond line\nthird line\nfourth line\n") - const attachment = FileAttachment.make({ - uri: pathToFileURL(file).href + "?start=2&end=3", - mime: "text/markdown", - name: "notes.md#2-3", - }) - const result = yield* SessionRunnerAttachment.materialize(services, new Map(), [user([attachment])]) + const result = yield* SessionRunnerAttachment.resolutions( + services, + prompt([{ uri: pathToFileURL(file).href + "?start=2&end=3", mime: "text/markdown", name: "notes.md#2-3" }]), + ) - const message = requireUser(result[0]) - expect(message.text).toContain("second line") - expect(message.text).toContain("third line") - expect(message.text).not.toContain("first line") - expect(message.text).not.toContain("fourth line") + expect(result[0].resolved).toContain("second line") + expect(result[0].resolved).toContain("third line") + expect(result[0].resolved).not.toContain("first line") + expect(result[0].resolved).not.toContain("fourth line") }), ) - it.effect("re-encodes an image attachment as a data URL media part", () => + it.effect("resolves an image attachment to a data URL", () => Effect.gen(function* () { const { services, files, directory } = yield* fixture const file = path.join(directory, "pixel.png") const png = Uint8Array.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a, 1, 2, 3, 4]) yield* files.writeFile(file, png) - const attachment = FileAttachment.make({ - uri: pathToFileURL(file).href, - mime: "image/png", - name: "pixel.png", - }) - const result = yield* SessionRunnerAttachment.materialize(services, new Map(), [user([attachment])]) + const result = yield* SessionRunnerAttachment.resolutions( + services, + prompt([{ uri: pathToFileURL(file).href, mime: "image/png", name: "pixel.png" }]), + ) - const message = requireUser(result[0]) - expect(message.text).toBe("Look at this") - expect(message.files).toHaveLength(1) - expect(message.files![0].mime).toBe("image/png") - expect(message.files![0].uri).toBe(`data:image/png;base64,${Buffer.from(png).toString("base64")}`) + expect(result[0].resolved).toBe(`data:image/png;base64,${Buffer.from(png).toString("base64")}`) }), ) - it.effect("degrades unreadable attachments to a model-visible note instead of failing", () => + it.effect("resolves unreadable attachments to a model-visible note instead of failing", () => Effect.gen(function* () { const { services, directory } = yield* fixture - const attachment = FileAttachment.make({ - uri: pathToFileURL(path.join(directory, "missing.txt")).href, - mime: "text/plain", - name: "missing.txt", - }) - const result = yield* SessionRunnerAttachment.materialize(services, new Map(), [user([attachment])]) + const result = yield* SessionRunnerAttachment.resolutions( + services, + prompt([ + { uri: pathToFileURL(path.join(directory, "missing.txt")).href, mime: "text/plain", name: "missing.txt" }, + ]), + ) - const message = requireUser(result[0]) - expect(message.files).toEqual([]) - expect(message.text).toContain('') + expect(result[0].resolved).toContain('') }), ) - it.effect("reuses cached materialization for the life of a drain", () => + it.effect("skips data URLs and deduplicates repeated URIs", () => Effect.gen(function* () { const { services, files, directory } = yield* fixture const file = path.join(directory, "notes.md") - yield* files.writeFileString(file, "original content\n") - const attachment = FileAttachment.make({ - uri: pathToFileURL(file).href, - mime: "text/plain", - name: "notes.md", - }) - const cache: SessionRunnerAttachment.Cache = new Map() + yield* files.writeFileString(file, "content\n") + const uri = pathToFileURL(file).href - const first = yield* SessionRunnerAttachment.materialize(services, cache, [user([attachment])]) - yield* files.writeFileString(file, "changed content\n") - const second = yield* SessionRunnerAttachment.materialize(services, cache, [user([attachment])]) + const result = yield* SessionRunnerAttachment.resolutions( + services, + prompt([ + { uri: "data:image/png;base64,aGVsbG8=", mime: "image/png", name: "hello.png" }, + { uri, mime: "text/plain", name: "notes.md" }, + { uri, mime: "text/plain", name: "notes.md" }, + ]), + ) - expect(requireUser(second[0]).text).toBe(requireUser(first[0]).text) - expect(requireUser(second[0]).text).toContain("original content") - }), - ) - - it.effect("passes data URLs and non-user messages through unchanged", () => - Effect.gen(function* () { - const { services } = yield* fixture - const dataAttachment = FileAttachment.make({ - uri: "data:image/png;base64,aGVsbG8=", - mime: "image/png", - name: "hello.png", - }) - const original = user([dataAttachment]) - const synthetic = SessionMessage.Synthetic.make({ - id: SessionMessage.ID.make("msg_synthetic"), - type: "synthetic", - sessionID: SessionV2.ID.make("ses_translate"), - text: "Synthetic context", - time: { created }, - }) - - const result = yield* SessionRunnerAttachment.materialize(services, new Map(), [original, synthetic]) - - expect(result[0]).toBe(original) - expect(result[1]).toBe(synthetic) + expect(result).toHaveLength(1) + expect(result[0].uri).toBe(uri) }), ) }) diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index cf35dd6cb8..dc67776e9e 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -711,10 +711,13 @@ describe("SessionRunnerLLM", () => { const text = userTexts(requests[0]).join("\n") expect(text).toContain('') expect(text).toContain("nested.txt") - // Durable projection keeps the original attachment URI; only the request is expanded. - expect(yield* session.messages({ sessionID })).toMatchObject([ - { type: "user", files: [{ mime: "application/x-directory" }] }, - ]) + // The durable projection keeps the original URI and snapshots the resolved listing. + const messages = yield* session.messages({ sessionID }) + expect(messages).toMatchObject([{ type: "user", files: [{ mime: "application/x-directory" }] }]) + const stored = messages[0] + if (stored?.type !== "user") throw new Error("Expected a user message") + expect(stored.files?.[0]?.uri.startsWith("file:")).toBe(true) + expect(stored.files?.[0]?.resolved).toContain("nested.txt") }), ) @@ -2330,7 +2333,13 @@ describe("SessionRunnerLLM", () => { const session = yield* SessionV2.Service const events = yield* EventV2.Service yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Recover interrupted tool" }), resume: false }) - yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers( + (yield* Database.Service).db, + events, + sessionID, + Number.MAX_SAFE_INTEGER, + SessionInput.unresolved, + ) const assistantMessageID = SessionMessage.ID.create() yield* events.publish(SessionEvent.Step.Started, { sessionID, @@ -2394,7 +2403,13 @@ describe("SessionRunnerLLM", () => { prompt: Prompt.make({ text: "Recover interrupted hosted tool" }), resume: false, }) - yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers( + (yield* Database.Service).db, + events, + sessionID, + Number.MAX_SAFE_INTEGER, + SessionInput.unresolved, + ) const assistantMessageID = SessionMessage.ID.create() yield* events.publish(SessionEvent.Step.Started, { sessionID, @@ -2454,7 +2469,13 @@ describe("SessionRunnerLLM", () => { prompt: Prompt.make({ text: "Recover interrupted tool input" }), resume: false, }) - yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers( + (yield* Database.Service).db, + events, + sessionID, + Number.MAX_SAFE_INTEGER, + SessionInput.unresolved, + ) const assistantMessageID = SessionMessage.ID.create() yield* events.publish(SessionEvent.Step.Started, { sessionID, diff --git a/packages/schema/src/session-event.ts b/packages/schema/src/session-event.ts index 5013dddee9..a5c9f95eff 100644 --- a/packages/schema/src/session-event.ts +++ b/packages/schema/src/session-event.ts @@ -105,10 +105,24 @@ export const Forked = Event.define({ }) export type Forked = typeof Forked.Type +/** + * Model-visible content captured for one attachment URI at promotion time: + * a data URL for media, otherwise text. Recorded on the event so projection + * replay stays deterministic without filesystem access. + */ +export const AttachmentResolution = Schema.Struct({ + uri: Schema.String, + resolved: Schema.String, +}).annotate({ identifier: "session.next.event.attachment-resolution" }) +export interface AttachmentResolution extends Schema.Schema.Type {} + export const Prompted = Event.define({ type: "session.next.prompted", ...options, - schema: PromptFields, + schema: { + ...PromptFields, + resolutions: Schema.Array(AttachmentResolution).pipe(optional), + }, }) export type Prompted = typeof Prompted.Type diff --git a/packages/schema/src/session-message.ts b/packages/schema/src/session-message.ts index 457cf87206..0a3691211e 100644 --- a/packages/schema/src/session-message.ts +++ b/packages/schema/src/session-message.ts @@ -41,11 +41,22 @@ export const ModelSwitched = Schema.Struct({ model: Model.Ref, }).annotate({ identifier: "Session.Message.ModelSwitched" }) +/** + * A prompt attachment plus the model-visible content captured for it at + * promotion time: a data URL for media, otherwise text. The original `uri` + * and `mime` are preserved for provenance; only `resolved` is server-produced. + */ +export interface UserFile extends Schema.Schema.Type {} +export const UserFile = Schema.Struct({ + ...FileAttachment.fields, + resolved: Schema.String.pipe(optional), +}).annotate({ identifier: "Session.Message.UserFile" }) + export interface User extends Schema.Schema.Type {} export const User = Schema.Struct({ ...Base, text: Prompt.fields.text, - files: Prompt.fields.files, + files: Schema.Array(UserFile).pipe(optional), agents: Prompt.fields.agents, type: Schema.Literal("user"), }).annotate({ identifier: "Session.Message.User" }) diff --git a/specs/v2/session.md b/specs/v2/session.md index b0931fdb2f..33e6dec956 100644 --- a/specs/v2/session.md +++ b/specs/v2/session.md @@ -145,7 +145,7 @@ Status: `complete` is usable in the native V2 path, `partial` covers only part o | Per-turn request assembly | Automatic/context-pressure compaction | complete | V2 initiates automatic and overflow-triggered compaction, then rebuilds the baseline from the completed checkpoint. | | Prompt/reference expansion | Durable typed prompt attachments | complete | None. | | Prompt/reference expansion | Native template and `@` mention expansion | missing | Parse and resolve native V2 prompt input before durable admission. | -| Prompt/reference expansion | File, directory, media, and MCP-resource materialization | partial | Materialize and normalize sources instead of lowering unresolved attachment metadata. | +| Prompt/reference expansion | File, directory, media, and MCP-resource materialization | partial | Local file, directory, and image attachments resolve durably at promotion (`Prompted.resolutions`); MCP-resource capture remains. | | Prompt/reference expansion | Agent-reference expansion | missing | Produce permission-aware model-visible task guidance. | | Prompt/reference expansion | Configured-reference expansion | missing | Resolve aliases and emit durable model-visible reference context or failures. | | Prompt/reference expansion | Native synthetic expansion replay | partial | V2 replays synthetic messages but only the V1 compatibility path creates them. |