ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(core): normalize attachment images and cache materialization per drain

Browse source
This commit is contained in:
Kit Langton 2026-07-02 08:46:16 -04:00
commit 6bd75aedad
3 changed files with 139 additions and 80 deletions
Download patch file Download diff file Expand all files Collapse all files

111
packages/core/src/session/runner/attachment.ts
View file

@ -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<string, Materialized>
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</${tag}>`
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<Materialized>({ 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<Materialized>({ expansion: wrap("attachment-unavailable", file.name ?? file.uri, error.message) }),
),
)
}

55
packages/core/src/session/runner/llm.ts
View file

@ -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,