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

refactor(core): resolve attachments durably at promotion

Browse source
This commit is contained in:
Kit Langton 2026-07-02 09:48:02 -04:00
commit 994f55423a
14 changed files with 252 additions and 252 deletions
Download patch file Download diff file Expand all files Collapse all files

8
packages/core/src/session/compaction.ts
View file

@ -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") {

22
packages/core/src/session/input.ts
View file

@ -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<ReadonlyArray<SessionEvent.AttachmentResolution>>
/** 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<typeof SessionInputTable.$inferSelect>,
) {
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))
})

6
packages/core/src/session/message-updater.ts
View file

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

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

@ -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<string, Materialized>
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<Materialized>({ 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<Materialized>({ 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))))

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

@ -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)) {

30
packages/core/src/session/runner/to-llm-message.ts
View file

@ -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: `<attachment-unavailable path=${JSON.stringify(file.name ?? file.uri)}>\nAttachment was not captured; read it with tools if needed.\n</attachment-unavailable>`,
}
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 } : {}),