From 6bd75aedadccac36753c8c3778a4d17631cbe25b Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Thu, 2 Jul 2026 08:46:16 -0400 Subject: [PATCH] refactor(core): normalize attachment images and cache materialization per drain --- .../core/src/session/runner/attachment.ts | 111 +++++++++++------- packages/core/src/session/runner/llm.ts | 55 +++++---- .../test/session-runner-attachment.test.ts | 53 ++++++--- 3 files changed, 139 insertions(+), 80 deletions(-) diff --git a/packages/core/src/session/runner/attachment.ts b/packages/core/src/session/runner/attachment.ts index 8c7546a99f..8a4e233c44 100644 --- a/packages/core/src/session/runner/attachment.ts +++ b/packages/core/src/session/runner/attachment.ts @@ -2,36 +2,64 @@ export * as SessionRunnerAttachment from "./attachment" import { fileURLToPath } from "url" 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" +export interface Services { + readonly reader: ReadToolFileSystem.Interface + readonly image: Image.Interface +} + +/** + * 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. * * 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 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. + * 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. */ export const materialize = Effect.fn("SessionRunnerAttachment.materialize")(function* ( - reader: ReadToolFileSystem.Interface, + services: Services, + cache: Cache, messages: readonly SessionMessage.Message[], ) { + 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) => materializeFile(reader, file)).pipe( - Effect.map((results) => - SessionMessage.User.make({ + 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])), - ].join("\n\n"), + ] + .filter(Boolean) + .join("\n\n"), files: results.flatMap((result) => (result.file === undefined ? [] : [result.file])), }), ), @@ -41,61 +69,56 @@ export const materialize = Effect.fn("SessionRunnerAttachment.materialize")(func const local = (file: FileAttachment) => file.uri.startsWith("file:") -interface Materialized { - readonly file?: FileAttachment - readonly expansion?: string -} - const wrap = (tag: string, path: string, body: string) => `<${tag} path=${JSON.stringify(path)}>\n${body}\n` -const unavailable = (path: string, reason: string): Materialized => ({ - expansion: wrap("attachment-unavailable", path, reason), -}) - // Mirror V1's `?start`/`?end` line-range attachment parameters. const pageFromRange = (url: URL) => { - const start = url.searchParams.get("start") - if (start === null) return undefined - const offset = Math.max(parseInt(start, 10) || 1, 1) - const end = url.searchParams.get("end") - const parsedEnd = end === null ? Number.NaN : parseInt(end, 10) - return { offset, ...(parsedEnd >= offset ? { limit: parsedEnd - offset + 1 } : {}) } + const start = parseInt(url.searchParams.get("start") ?? "", 10) + if (!Number.isInteger(start) || start < 1) return undefined + const end = parseInt(url.searchParams.get("end") ?? "", 10) + return { offset: start, ...(end >= start ? { limit: end - start + 1 } : {}) } } -const materializeFile = (reader: ReadToolFileSystem.Interface, file: FileAttachment) => { +const materializeFile = (services: Services, file: FileAttachment) => { if (!local(file)) return Effect.succeed({ file }) - const resolved = Effect.try({ - try: () => { - const url = new URL(file.uri) - const page = pageFromRange(url) - url.search = "" - url.hash = "" - return { target: AbsolutePath.make(fileURLToPath(url)), page } - }, - catch: (error) => (error instanceof Error ? error : new Error(String(error))), - }) return Effect.gen(function* () { - const { target, page } = yield* resolved + const { target, page } = yield* Effect.try({ + try: () => { + const url = new URL(file.uri) + const page = pageFromRange(url) + url.search = "" + url.hash = "" + return { target: AbsolutePath.make(fileURLToPath(url)), page } + }, + catch: (error) => (error instanceof Error ? error : new Error(String(error))), + }) const display = file.name ?? target - const kind = yield* reader.inspect(target) + const kind = yield* services.reader.inspect(target) if (kind === "directory") { - const listing = yield* reader.list(target) + const listing = yield* services.reader.list(target) const lines = [ ...listing.entries.map((entry) => entry.path), ...(listing.truncated ? ["(listing truncated)"] : []), ] return { expansion: wrap("attached-directory", display, lines.join("\n")) } satisfies Materialized } - const content = yield* reader.read(target, display, page) - if ("encoding" in content && content.encoding === "base64") + 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 + } + 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:${content.mime};base64,${content.content}`, mime: content.mime }, + file: { ...file, uri: `data:${normalized.mime};base64,${normalized.content}`, mime: normalized.mime }, } satisfies Materialized - const truncated = "truncated" in content && content.truncated ? "\n(content truncated)" : "" - return { expansion: wrap("attached-file", display, content.content + truncated) } satisfies Materialized + } + return { expansion: wrap("attached-file", display, content.content) } satisfies Materialized }).pipe( Effect.catch((error) => - Effect.succeed(unavailable(file.name ?? file.uri, error instanceof Error ? error.message : String(error))), + Effect.succeed({ expansion: 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 f7f3b420af..b382e22929 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -12,6 +12,7 @@ import { Cause, DateTime, Effect, FiberSet, Layer, Option, Semaphore, Stream } f import { AgentV2 } from "../../agent" import { Config } from "../../config" import { Database } from "../../database/database" +import { Image } from "../../image" import { EventV2 } from "../../event" import { Location } from "../../location" import { ModelV2 } from "../../model" @@ -101,7 +102,10 @@ const layer = Layer.effect( const llm = yield* LLMClient.Service const agents = yield* AgentV2.Service const tools = yield* ToolRegistry.Service - const reader = yield* ReadToolFileSystem.Service + const attachments: SessionRunnerAttachment.Services = { + reader: yield* ReadToolFileSystem.Service, + image: yield* Image.Service, + } const models = yield* SessionRunnerModel.Service const store = yield* SessionStore.Service const location = yield* Location.Service @@ -181,6 +185,7 @@ 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) @@ -207,7 +212,7 @@ const layer = Layer.effect( 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(reader, context) + const materialized = yield* SessionRunnerAttachment.materialize(attachments, attachmentCache, context) const isLastStep = agent.info?.steps !== undefined && currentStep >= agent.info.steps const toolMaterialization = isLastStep ? undefined @@ -223,7 +228,7 @@ const layer = Layer.effect( tools: toolMaterialization?.definitions ?? [], toolChoice: isLastStep ? "none" : undefined, }) - if (yield* compaction.compactIfNeeded({ sessionID: session.id, messages: context, request })) + if (yield* compaction.compactIfNeeded({ sessionID: session.id, messages: materialized, request })) return yield* Effect.die(continueAfterCompaction(currentStep)) const startSnapshot = yield* snapshots.capture() const publisher = createLLMEventPublisher(events, { @@ -296,7 +301,7 @@ const layer = Layer.effect( recoverOverflow && !publisher.hasAssistantStarted() && isContextOverflowFailure(overflowFailure ?? failure) && - (yield* restore(recoverOverflow({ sessionID: session.id, messages: context, request }))) + (yield* restore(recoverOverflow({ sessionID: session.id, messages: materialized, request }))) ) return yield* Effect.die(continueAfterOverflowCompaction(currentStep)) if (overflowFailure) yield* publish(overflowFailure) @@ -362,31 +367,34 @@ 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) { - 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 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 runTurn: RunTurn = Effect.fnUntraced(function* (sessionID, promotion, step) { - return yield* runTurnAttempt(sessionID, promotion, step, compaction.compactAfterOverflow).pipe( + const runTurn: RunTurn = Effect.fnUntraced(function* (sessionID, promotion, step, attachmentCache) { + return yield* runTurnAttempt(sessionID, promotion, step, attachmentCache, 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) - return yield* runTurn(sessionID, undefined, defect.transition.step) + return yield* runAfterOverflowCompaction(sessionID, undefined, defect.transition.step, attachmentCache) + return yield* runTurn(sessionID, undefined, defect.transition.step, attachmentCache) }), ), ) @@ -400,13 +408,15 @@ 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) + const result = yield* runTurn(input.sessionID, promotion, step, attachmentCache) // 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)) { @@ -438,6 +448,7 @@ export const node = makeLocationNode({ AgentV2.node, ToolRegistry.node, ReadToolFileSystem.node, + Image.node, SessionRunnerModel.node, SessionStore.node, Location.node, diff --git a/packages/core/test/session-runner-attachment.test.ts b/packages/core/test/session-runner-attachment.test.ts index ccc7e69f3e..8a66656ac7 100644 --- a/packages/core/test/session-runner-attachment.test.ts +++ b/packages/core/test/session-runner-attachment.test.ts @@ -4,6 +4,7 @@ import { pathToFileURL } from "url" import { DateTime, 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" @@ -23,11 +24,14 @@ const user = (files: FileAttachment[]) => time: { created }, }) +// The resizer-unavailable stub exercises the raw-content fallback deterministically. +const image = Image.Service.of({ normalize: () => Effect.fail(new Image.ResizerUnavailableError()) }) + const fixture = Effect.gen(function* () { - const reader = yield* ReadToolFileSystem.Service + const services = { reader: yield* ReadToolFileSystem.Service, image } const files = yield* FileSystem.FileSystem const directory = yield* files.makeTempDirectoryScoped() - return { reader, files, directory } + return { services, files, directory } }) const requireUser = (message: SessionMessage.Message) => { @@ -38,7 +42,7 @@ const requireUser = (message: SessionMessage.Message) => { describe("SessionRunnerAttachment.materialize", () => { it.effect("expands a directory attachment into a listing instead of media", () => Effect.gen(function* () { - const { reader, files, directory } = yield* fixture + 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({ @@ -47,7 +51,7 @@ describe("SessionRunnerAttachment.materialize", () => { name: "project/", }) - const result = yield* SessionRunnerAttachment.materialize(reader, [user([attachment])]) + const result = yield* SessionRunnerAttachment.materialize(services, new Map(), [user([attachment])]) const message = requireUser(result[0]) expect(message.files).toEqual([]) @@ -59,7 +63,7 @@ describe("SessionRunnerAttachment.materialize", () => { it.effect("expands a text file attachment into inline content", () => Effect.gen(function* () { - const { reader, files, directory } = yield* fixture + 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({ @@ -68,7 +72,7 @@ describe("SessionRunnerAttachment.materialize", () => { name: "notes.md", }) - const result = yield* SessionRunnerAttachment.materialize(reader, [user([attachment])]) + const result = yield* SessionRunnerAttachment.materialize(services, new Map(), [user([attachment])]) const message = requireUser(result[0]) expect(message.files).toEqual([]) @@ -79,7 +83,7 @@ describe("SessionRunnerAttachment.materialize", () => { it.effect("honors ?start/?end line-range parameters", () => Effect.gen(function* () { - const { reader, files, directory } = yield* fixture + 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({ @@ -88,7 +92,7 @@ describe("SessionRunnerAttachment.materialize", () => { name: "notes.md#2-3", }) - const result = yield* SessionRunnerAttachment.materialize(reader, [user([attachment])]) + const result = yield* SessionRunnerAttachment.materialize(services, new Map(), [user([attachment])]) const message = requireUser(result[0]) expect(message.text).toContain("second line") @@ -100,7 +104,7 @@ describe("SessionRunnerAttachment.materialize", () => { it.effect("re-encodes an image attachment as a data URL media part", () => Effect.gen(function* () { - const { reader, files, directory } = yield* fixture + 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) @@ -110,7 +114,7 @@ describe("SessionRunnerAttachment.materialize", () => { name: "pixel.png", }) - const result = yield* SessionRunnerAttachment.materialize(reader, [user([attachment])]) + const result = yield* SessionRunnerAttachment.materialize(services, new Map(), [user([attachment])]) const message = requireUser(result[0]) expect(message.text).toBe("Look at this") @@ -122,14 +126,14 @@ describe("SessionRunnerAttachment.materialize", () => { it.effect("degrades unreadable attachments to a model-visible note instead of failing", () => Effect.gen(function* () { - const { reader, directory } = yield* fixture + 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(reader, [user([attachment])]) + const result = yield* SessionRunnerAttachment.materialize(services, new Map(), [user([attachment])]) const message = requireUser(result[0]) expect(message.files).toEqual([]) @@ -137,9 +141,30 @@ describe("SessionRunnerAttachment.materialize", () => { }), ) + it.effect("reuses cached materialization for the life of a drain", () => + 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() + + 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])]) + + 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 { reader } = yield* fixture + const { services } = yield* fixture const dataAttachment = FileAttachment.make({ uri: "data:image/png;base64,aGVsbG8=", mime: "image/png", @@ -154,7 +179,7 @@ describe("SessionRunnerAttachment.materialize", () => { time: { created }, }) - const result = yield* SessionRunnerAttachment.materialize(reader, [original, synthetic]) + const result = yield* SessionRunnerAttachment.materialize(services, new Map(), [original, synthetic]) expect(result[0]).toBe(original) expect(result[1]).toBe(synthetic)