diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index a6f20cf2f985..33737e69db04 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -1856,6 +1856,7 @@ export type SessionToolProgress = { callID: string structured: { [x: string]: any } content: Array + payloadHash?: string } } @@ -1875,6 +1876,7 @@ export type SessionToolSuccess = { result?: any executed: boolean resultState?: SessionMessageProviderState6 + payloadHash?: string } } @@ -2162,7 +2164,7 @@ export type SessionInputAdmitted = { type: "session.input.admitted" durable: { aggregateID: string; seq: number; version: 1 } location?: LocationRef - data: { sessionID: string; inputID: string; input: SessionPendingMessage } + data: { sessionID: string; inputID: string; input: SessionPendingMessage; payloadHash?: string } } export type SessionMessageInfo = diff --git a/packages/core/src/database/migration.gen.ts b/packages/core/src/database/migration.gen.ts index c4067c4765ed..d76383169c80 100644 --- a/packages/core/src/database/migration.gen.ts +++ b/packages/core/src/database/migration.gen.ts @@ -55,5 +55,6 @@ export const migrations = ( import("./migration/20260709190621_session_pending_table"), import("./migration/20260710025429_instruction_sync"), import("./migration/20260716020354_kv"), + import("./migration/20260718010000_session_tool_payload"), ]) ).map((module) => module.default) satisfies DatabaseMigration.Migration[] diff --git a/packages/core/src/database/migration/20260718010000_session_tool_payload.ts b/packages/core/src/database/migration/20260718010000_session_tool_payload.ts new file mode 100644 index 000000000000..b24b0fb8d14e --- /dev/null +++ b/packages/core/src/database/migration/20260718010000_session_tool_payload.ts @@ -0,0 +1,19 @@ +import { Effect } from "effect" +import type { DatabaseMigration } from "../migration" + +export default { + id: "20260718010000_session_tool_payload", + up(tx) { + return Effect.gen(function* () { + yield* tx.run(` + CREATE TABLE \`session_tool_payload\` ( + \`session_id\` text NOT NULL, + \`hash\` text NOT NULL, + \`value\` text NOT NULL, + PRIMARY KEY(\`session_id\`, \`hash\`), + FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE cascade + ); + `) + }) + }, +} satisfies DatabaseMigration.Migration diff --git a/packages/core/src/database/schema.gen.ts b/packages/core/src/database/schema.gen.ts index b92b75e47b53..c9220b9f9f26 100644 --- a/packages/core/src/database/schema.gen.ts +++ b/packages/core/src/database/schema.gen.ts @@ -244,6 +244,15 @@ export default { CONSTRAINT \`fk_session_project_id_project_id_fk\` FOREIGN KEY (\`project_id\`) REFERENCES \`project\`(\`id\`) ON DELETE CASCADE ); `) + yield* tx.run(` + CREATE TABLE \`session_tool_payload\` ( + \`session_id\` text NOT NULL, + \`hash\` text NOT NULL, + \`value\` text NOT NULL, + PRIMARY KEY(\`session_id\`, \`hash\`), + FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE cascade + ); + `) yield* tx.run(` CREATE TABLE \`session_share\` ( \`session_id\` text PRIMARY KEY, diff --git a/packages/core/src/session/pending.ts b/packages/core/src/session/pending.ts index 70f0b34c5b36..fd5ef0271773 100644 --- a/packages/core/src/session/pending.ts +++ b/packages/core/src/session/pending.ts @@ -21,6 +21,7 @@ import { SessionEvent } from "./event" import { SessionMessage } from "./message" import { SessionSchema } from "./schema" import { SessionMessageTable, SessionPendingTable } from "./sql" +import { ToolPayload } from "./tool-payload" type DatabaseService = Database.Interface["db"] @@ -31,12 +32,33 @@ const encodeUser = Schema.encodeSync(UserData) const decodeSynthetic = Schema.decodeUnknownSync(SyntheticData) const encodeSynthetic = Schema.encodeSync(SyntheticData) const decodeAdmittedEvent = Schema.decodeUnknownOption(SessionEvent.InputAdmitted.data) +const encodeMessage = Schema.encodeSync(Message) +const decodeMessage = Schema.decodeUnknownSync(Message) const admittedEventType = Event.versionedType( SessionEvent.InputAdmitted.type, SessionEvent.InputAdmitted.durable.version, ) const inboxLocks = KeyedMutex.makeUnsafe() +/** Drop inline attachment bytes from the durable event projection; full body lives in the payload blob. */ +export function stripAttachmentBytes(input: Message): Message { + if (input.type !== "user" || !input.data.files?.length) return input + return { + ...input, + data: { + ...input.data, + files: input.data.files.map((file) => ({ ...file, data: "" })), + }, + } +} + +function hasAttachmentBytes(input: Message): boolean { + if (input.type !== "user" || !input.data.files?.length) return false + return input.data.files.some((file) => file.data.length > 0) +} + +const decodeJson = Schema.decodeUnknownSync(Schema.Json) + export class LifecycleConflict extends Schema.TaggedErrorClass()( "SessionPending.LifecycleConflict", { @@ -126,15 +148,16 @@ const promotedFromHistory = Effect.fn("SessionPending.promotedFromHistory")(func for (const row of rows) { const decoded = decodeAdmittedEvent(row.data) if (decoded._tag !== "Some" || decoded.value.inputID !== id) continue + const input = decoded.value.payloadHash + ? decodeMessage(yield* ToolPayload.loadJson(db, sessionID, decoded.value.payloadHash).pipe(Effect.orDie)) + : decoded.value.input const base = { admittedSeq: row.seq, id, sessionID, timeCreated: DateTime.makeUnsafe(row.created), } - return decoded.value.input.type === "user" - ? User.make({ ...base, ...decoded.value.input }) - : Synthetic.make({ ...base, ...decoded.value.input }) + return input.type === "user" ? User.make({ ...base, ...input }) : Synthetic.make({ ...base, ...input }) } // A projected message without an admitted event in this aggregate (for // example fork-copied history) is not a retryable admission. @@ -157,11 +180,16 @@ export const admit = Effect.fn("SessionPending.admit")(function* ( } const promoted = yield* promotedFromHistory(db, request.sessionID, request.id) if (promoted !== undefined) return promoted + const thin = stripAttachmentBytes(request.input) + const payloadHash = hasAttachmentBytes(request.input) + ? yield* ToolPayload.insertJson(db, request.sessionID, decodeJson(encodeMessage(request.input))) + : undefined return yield* events .publish(SessionEvent.InputAdmitted, { inputID: request.id, sessionID: request.sessionID, - input: request.input, + input: thin, + ...(payloadHash !== undefined ? { payloadHash } : {}), }) .pipe( Effect.flatMap((event) => { diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts index 62bd34ecc6c1..7520cc4bcc8b 100644 --- a/packages/core/src/session/projector.ts +++ b/packages/core/src/session/projector.ts @@ -12,8 +12,10 @@ import { WorkspaceTable } from "../control-plane/workspace.sql" import { SessionMessage } from "./message" import { SessionMessageUpdater } from "./message-updater" import { SessionPending } from "./pending" +import type { SessionSchema } from "./schema" import { WorkspaceV2 } from "../workspace" import { InstructionState } from "./instruction-state" +import { ToolPayload } from "./tool-payload" import { MessageTable, PartTable, SessionPendingTable, SessionMessageTable, SessionTable } from "./sql" import type { DeepMutable } from "../schema" import { Slug } from "../util/slug" @@ -318,6 +320,33 @@ const projectFork = Effect.fn("SessionProjector.projectFork")(function* ( yield* InstructionState.rebuild(db, event.data.sessionID) }) +function hydrateToolPayload< + E extends { + readonly data: { + readonly sessionID: SessionSchema.ID + readonly payloadHash?: ToolPayload.Hash + readonly structured: Record + readonly content: ToolPayload.Body["content"] + readonly result?: unknown + } + }, +>(db: DatabaseService, event: E) { + return Effect.gen(function* () { + const payloadHash = event.data.payloadHash + if (!payloadHash) return event + const body = yield* ToolPayload.load(db, event.data.sessionID, payloadHash).pipe(Effect.orDie) + return { + ...event, + data: { + ...event.data, + structured: body.structured, + content: body.content, + ...(body.result !== undefined ? { result: body.result } : {}), + }, + } + }) +} + function run(db: DatabaseService, event: MessageEvent) { return Effect.gen(function* () { const decodeRow = (row: typeof SessionMessageTable.$inferSelect) => @@ -647,11 +676,16 @@ const layer = Layer.effectDiscard( Effect.gen(function* () { if (event.durable === undefined) return yield* Effect.die(new Error("Durable Session event is missing aggregate sequence")) + const input = event.data.payloadHash + ? Schema.decodeUnknownSync(SessionPending.Message)( + yield* ToolPayload.loadJson(db, event.data.sessionID, event.data.payloadHash).pipe(Effect.orDie), + ) + : event.data.input yield* SessionPending.projectAdmitted(db, { admittedSeq: event.durable.seq, id: event.data.inputID, sessionID: event.data.sessionID, - input: event.data.input, + input, timeCreated: event.created, }) }), @@ -697,8 +731,12 @@ const layer = Layer.effectDiscard( yield* events.project(SessionEvent.Tool.Input.Started, (event) => run(db, event)) yield* events.project(SessionEvent.Tool.Input.Ended, (event) => run(db, event)) yield* events.project(SessionEvent.Tool.Called, (event) => run(db, event)) - yield* events.project(SessionEvent.Tool.Progress, (event) => run(db, event)) - yield* events.project(SessionEvent.Tool.Success, (event) => run(db, event)) + yield* events.project(SessionEvent.Tool.Progress, (event) => + hydrateToolPayload(db, event).pipe(Effect.flatMap((hydrated) => run(db, hydrated))), + ) + yield* events.project(SessionEvent.Tool.Success, (event) => + hydrateToolPayload(db, event).pipe(Effect.flatMap((hydrated) => run(db, hydrated))), + ) yield* events.project(SessionEvent.Tool.Failed, (event) => run(db, event)) yield* events.project(SessionEvent.Reasoning.Started, (event) => run(db, event)) yield* events.project(SessionEvent.Reasoning.Ended, (event) => run(db, event)) diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 3495cfdc4fb7..cc42fa0727cc 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -20,6 +20,7 @@ import { SessionStore } from "../store" import { SessionTitle } from "../title" import { Service } from "./index" import { createLLMEventPublisher } from "./publish-llm-event" +import { ToolPayload } from "../tool-payload" import { Snapshot } from "../../snapshot" import { makeLocationNode } from "../../effect/app-node" import { llmClient } from "../../effect/app-node-platform" @@ -116,6 +117,7 @@ const layer = Layer.effect( const ownedToolFibers: Array> = [] let needsContinuation = false const startSnapshot = yield* snapshots.capture() + const persistToolPayload = (body: ToolPayload.Body) => ToolPayload.insert(db, session.id, body) const publisher = createLLMEventPublisher(events, { sessionID: session.id, agent: agent.id, @@ -125,6 +127,7 @@ const layer = Layer.effect( providerMetadataKey: model.route.providerMetadataKey ?? model.provider, snapshot: startSnapshot, assistantMessageID, + persistToolPayload, }) const publication = Semaphore.makeUnsafe(1) // Durable publishes are serialized so tool fibers and step settlement never interleave @@ -161,12 +164,26 @@ const layer = Layer.effect( call: event, progress: (update) => serialized( - events.publish(SessionEvent.Tool.Progress, { - sessionID: session.id, - assistantMessageID, - callID: event.id, - structured: { ...update.structured }, - content: [...update.content], + Effect.gen(function* () { + const body: ToolPayload.Body = { + structured: { ...update.structured }, + content: [...update.content], + } + const payloadHash = yield* persistToolPayload(body) + const thin = ToolPayload.preview(body) + const data = { + sessionID: session.id, + assistantMessageID, + callID: event.id, + structured: thin.structured, + content: thin.content, + payloadHash, + } + // Thin progress frames should always fit; over-budget is a defect. + yield* ToolPayload.assertEventDataBudget(data).pipe( + Effect.andThen(events.publish(SessionEvent.Tool.Progress, data)), + Effect.orDie, + ) }), ), }), diff --git a/packages/core/src/session/runner/publish-llm-event.ts b/packages/core/src/session/runner/publish-llm-event.ts index 6a71f5d88872..cfe84a74b00f 100644 --- a/packages/core/src/session/runner/publish-llm-event.ts +++ b/packages/core/src/session/runner/publish-llm-event.ts @@ -10,6 +10,7 @@ import { Money } from "@opencode-ai/schema/money" import { AgentV2 } from "../../agent" import { Snapshot } from "../../snapshot" import { SessionUsage } from "../usage" +import { ToolPayload } from "../tool-payload" type Input = { readonly sessionID: SessionSchema.ID @@ -18,6 +19,8 @@ type Input = { readonly providerMetadataKey: string readonly snapshot?: Snapshot.ID readonly assistantMessageID?: SessionMessage.ID + /** Persist full tool settlement beside the durable log; required for thin tool events. */ + readonly persistToolPayload: (body: ToolPayload.Body) => Effect.Effect } const record = (value: unknown): Record => @@ -378,19 +381,40 @@ export const createLLMEventPublisher = (events: Pick + events.publish(SessionEvent.Tool.Failed, { + sessionID: input.sessionID, + assistantMessageID: tool.assistantMessageID, + callID: event.id, + error: { type: "unknown", message: error.message }, + executed, + ...(resultState !== undefined ? { resultState } : {}), + }), + ), + ) return } case "tool-error": { @@ -400,6 +424,7 @@ export const createLLMEventPublisher = (events: Pick ${event.name}`)) if (tool.settled) return yield* Effect.die(new Error(`Duplicate tool error: ${event.id}`)) tool.settled = true + const resultState = providerState(event.providerMetadata) yield* events.publish(SessionEvent.Tool.Failed, { sessionID: input.sessionID, assistantMessageID: tool.assistantMessageID, @@ -409,7 +434,7 @@ export const createLLMEventPublisher = (events: Pick(), }) +/** Session-durable tool settlement bodies for thin tool events. */ +export const SessionToolPayloadTable = sqliteTable( + "session_tool_payload", + { + session_id: text() + .$type() + .notNull() + .references(() => SessionTable.id, { onDelete: "cascade" }), + hash: text().$type().notNull(), + value: text({ mode: "json" }).$type().notNull(), + }, + (table) => [primaryKey({ columns: [table.session_id, table.hash] })], +) + export const InstructionStateTable = sqliteTable("instruction_state", { session_id: text() .$type() diff --git a/packages/core/src/session/tool-payload.ts b/packages/core/src/session/tool-payload.ts new file mode 100644 index 000000000000..a293763ed26f --- /dev/null +++ b/packages/core/src/session/tool-payload.ts @@ -0,0 +1,142 @@ +export * as ToolPayload from "./tool-payload" + +import { createHash } from "crypto" +import { Effect, Schema } from "effect" +import { eq, and } from "drizzle-orm" +import { ToolPayload as ToolPayloadSchema } from "@opencode-ai/schema/tool-payload" +import type { Database } from "../database/database" +import { SessionToolPayloadTable } from "./sql" +import type { SessionSchema } from "./schema" + +export const Hash = ToolPayloadSchema.Hash +export type Hash = ToolPayloadSchema.Hash + +export const Body = ToolPayloadSchema.Body +export type Body = ToolPayloadSchema.Body + +/** UTF-8 byte ceiling for encoded durable tool event `data` after thinning. */ +export const MaxEventDataBytes = 256 * 1024 + +export const PreviewTextBytes = 2 * 1024 + +export class OverBudgetError extends Schema.TaggedErrorClass()("ToolPayload.OverBudgetError", { + actualBytes: Schema.Number, + maxBytes: Schema.Number, + message: Schema.String, +}) {} + +export class MissingError extends Schema.TaggedErrorClass()("ToolPayload.MissingError", { + hash: Hash, + sessionID: Schema.String, + message: Schema.String, +}) {} + +export type Error = OverBudgetError | MissingError + +type DatabaseService = Database.Interface["db"] + +const encodeBody = Schema.encodeSync(Body) +const decodeJson = Schema.decodeUnknownSync(Schema.Json) + +export function hash(value: Body): Hash { + return Hash.make( + createHash("sha256") + .update(canonical(decodeJson(encodeBody(value)))) + .digest("hex"), + ) +} + +export const insertJson = Effect.fn("ToolPayload.insertJson")(function* ( + db: DatabaseService, + sessionID: SessionSchema.ID, + value: Schema.Json, +) { + const payloadHash = Hash.make(createHash("sha256").update(canonical(value)).digest("hex")) + yield* db + .insert(SessionToolPayloadTable) + .values({ + session_id: sessionID, + hash: payloadHash, + value, + }) + .onConflictDoNothing() + .run() + .pipe(Effect.orDie) + return payloadHash +}) + +export const loadJson = Effect.fn("ToolPayload.loadJson")(function* ( + db: DatabaseService, + sessionID: SessionSchema.ID, + payloadHash: Hash, +) { + const row = yield* db + .select() + .from(SessionToolPayloadTable) + .where(and(eq(SessionToolPayloadTable.session_id, sessionID), eq(SessionToolPayloadTable.hash, payloadHash))) + .get() + .pipe(Effect.orDie) + if (!row) + return yield* new MissingError({ + hash: payloadHash, + sessionID, + message: `Session payload blob not found: ${payloadHash}`, + }) + return row.value +}) + +export const insert = Effect.fn("ToolPayload.insert")(function* ( + db: DatabaseService, + sessionID: SessionSchema.ID, + body: Body, +) { + return yield* insertJson(db, sessionID, decodeJson(encodeBody(body))) +}) + +export const load = Effect.fn("ToolPayload.load")(function* ( + db: DatabaseService, + sessionID: SessionSchema.ID, + payloadHash: Hash, +) { + const value = yield* loadJson(db, sessionID, payloadHash) + return yield* Schema.decodeUnknownEffect(Body)(value).pipe(Effect.orDie) +}) + +/** Preview fields safe to put on the durable event / SSE frame. */ +export function preview(body: Body): { + readonly structured: Record + readonly content: Body["content"] +} { + const text = body.content + .filter((item): item is Extract => item.type === "text") + .map((item) => item.text) + .join("") + const truncated = + Buffer.byteLength(text, "utf-8") <= PreviewTextBytes + ? text + : text.slice(0, Math.max(0, PreviewTextBytes - 3)) + "..." + return { + structured: {}, + content: truncated ? [{ type: "text" as const, text: truncated }] : [], + } +} + +export function assertEventDataBudget(data: unknown): Effect.Effect { + const actualBytes = Buffer.byteLength(JSON.stringify(data), "utf-8") + if (actualBytes <= MaxEventDataBytes) return Effect.void + return new OverBudgetError({ + actualBytes, + maxBytes: MaxEventDataBytes, + message: `Tool event data is ${actualBytes} bytes; the limit is ${MaxEventDataBytes} bytes`, + }) +} + +function canonical(value: Schema.Json): string { + if (Array.isArray(value)) return `[${value.map(canonical).join(",")}]` + if (value !== null && typeof value === "object") + return `{${Object.entries(value) + .sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0)) + .map(([key, entry]) => `${JSON.stringify(key)}:${canonical(decodeJson(entry))}`) + .join(",")}}` + return JSON.stringify(value) +} diff --git a/packages/core/test/session-pending-attachments.test.ts b/packages/core/test/session-pending-attachments.test.ts new file mode 100644 index 000000000000..24b1341056be --- /dev/null +++ b/packages/core/test/session-pending-attachments.test.ts @@ -0,0 +1,191 @@ +import { describe, expect } from "bun:test" +import { and, eq } from "drizzle-orm" +import { Effect, Option, Schema } from "effect" +import { Database } from "@opencode-ai/core/database/database" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { EventV2 } from "@opencode-ai/core/event" +import { EventTable } from "@opencode-ai/core/event/sql" +import { Project } from "@opencode-ai/core/project" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { SessionEvent } from "@opencode-ai/core/session/event" +import { SessionPending } from "@opencode-ai/core/session/pending" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { SessionSchema } from "@opencode-ai/core/session/schema" +import { SessionMessage } from "@opencode-ai/core/session/message" +import { SessionTable } from "@opencode-ai/core/session/sql" +import { testEffect } from "./lib/effect" + +const it = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node, SessionProjector.node]))) +const inputType = "user" as const +const strippedBytes = "" + +describe("SessionPending attachment bounding", () => { + it.effect("admits a large attachment without putting base64 on the durable event", () => + Effect.gen(function* () { + const { db } = yield* Database.Service + const events = yield* EventV2.Service + const sessionID = SessionSchema.ID.make("ses_pending_attach") + const inputID = SessionMessage.ID.create() + const mega = Buffer.alloc(1024 * 1024, 1).toString("base64") + yield* db + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] }) + .onConflictDoNothing() + .run() + .pipe(Effect.orDie) + yield* db + .insert(SessionTable) + .values({ + id: sessionID, + project_id: Project.ID.global, + slug: "pending-attach", + directory: "/project", + title: "Pending attach", + version: "test", + }) + .run() + .pipe(Effect.orDie) + + const admitted = yield* SessionPending.admit(db, events, { + id: inputID, + sessionID, + input: { + type: inputType, + delivery: "steer", + data: { + text: "see file", + files: [ + { + data: mega, + mime: "application/octet-stream", + source: { type: "uri", uri: "file:///tmp/huge.bin" }, + name: "huge.bin", + }, + ], + }, + }, + }) + + type AdmittedUser = Extract + const admittedUser = Option.getOrThrowWith( + Option.liftPredicate(admitted, (value): value is AdmittedUser => value.type === inputType), + () => new Error(`Expected ${inputType} admitted input`), + ) + expect(admittedUser.data.files?.[0]?.data).toBe(mega) + + const rows = yield* db + .select() + .from(EventTable) + .where(and(eq(EventTable.aggregate_id, sessionID), eq(EventTable.type, "session.input.admitted.1"))) + .all() + .pipe(Effect.orDie) + expect(rows).toHaveLength(1) + const eventData = Schema.decodeUnknownSync(SessionEvent.InputAdmitted.data)(rows[0]!.data) + expect(eventData.payloadHash).toBeDefined() + + type UserInput = Extract + const input = Option.getOrThrowWith( + Option.liftPredicate(eventData.input, (value): value is UserInput => value.type === inputType), + () => new Error(`Expected ${inputType} input on admitted event`), + ) + expect(input.data.files?.[0]?.data).toBe(strippedBytes) + expect(JSON.stringify(eventData).includes(mega)).toBe(false) + + const pending = yield* SessionPending.list(db, sessionID) + type UserPending = Extract<(typeof pending)[number], { type: typeof inputType }> + const row = Option.getOrThrowWith( + Option.fromUndefinedOr(pending.find((item) => item.id === inputID)).pipe( + Option.flatMap((pendingRow) => + Option.liftPredicate(pendingRow, (value): value is UserPending => value.type === inputType), + ), + ), + () => new Error(`Expected ${inputType} pending input ${inputID}`), + ) + expect(row.data.files?.[0]?.data).toBe(mega) + }), + ) + + it.effect("stripAttachmentBytes clears only file data fields", () => + Effect.sync(() => { + const stripped = SessionPending.stripAttachmentBytes({ + type: inputType, + delivery: "steer", + data: { + text: "hi", + files: [ + { + data: "AAAA", + mime: "text/plain", + source: { type: "uri", uri: "file:///tmp/a.txt" }, + name: "a.txt", + }, + ], + }, + }) + type UserInput = Extract + const input = Option.getOrThrowWith( + Option.liftPredicate(stripped, (value): value is UserInput => value.type === inputType), + () => new Error(`Expected ${inputType} input`), + ) + expect(input.data.text).toBe("hi") + expect(input.data.files?.[0]?.data).toBe(strippedBytes) + expect(input.data.files?.[0]?.name).toBe("a.txt") + }), + ) + + it.effect("text-only admits omit payloadHash", () => + Effect.gen(function* () { + const { db } = yield* Database.Service + const events = yield* EventV2.Service + const sessionID = SessionSchema.ID.make("ses_pending_text") + const inputID = SessionMessage.ID.create() + yield* db + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] }) + .onConflictDoNothing() + .run() + .pipe(Effect.orDie) + yield* db + .insert(SessionTable) + .values({ + id: sessionID, + project_id: Project.ID.global, + slug: "pending-text", + directory: "/project", + title: "Pending text", + version: "test", + }) + .run() + .pipe(Effect.orDie) + + yield* SessionPending.admit(db, events, { + id: inputID, + sessionID, + input: { + type: inputType, + delivery: "steer", + data: { text: "hello" }, + }, + }) + + const rows = yield* db + .select() + .from(EventTable) + .where(and(eq(EventTable.aggregate_id, sessionID), eq(EventTable.type, "session.input.admitted.1"))) + .all() + .pipe(Effect.orDie) + expect(rows).toHaveLength(1) + const eventData = Schema.decodeUnknownSync(SessionEvent.InputAdmitted.data)(rows[0]!.data) + expect(eventData.payloadHash).toBeUndefined() + + type UserInput = Extract + const input = Option.getOrThrowWith( + Option.liftPredicate(eventData.input, (value): value is UserInput => value.type === inputType), + () => new Error(`Expected ${inputType} input on admitted event`), + ) + expect(input.data.text).toBe("hello") + }), + ) +}) diff --git a/packages/core/test/session-runner-tool-events.test.ts b/packages/core/test/session-runner-tool-events.test.ts index 93600c3058d9..4a6691f2f1df 100644 --- a/packages/core/test/session-runner-tool-events.test.ts +++ b/packages/core/test/session-runner-tool-events.test.ts @@ -1,5 +1,5 @@ import { expect, test } from "bun:test" -import { Effect, Schema } from "effect" +import { Effect, Option, Schema } from "effect" import { LLMEvent } from "@opencode-ai/ai" import { Money } from "@opencode-ai/schema/money" import { EventV2 } from "@opencode-ai/core/event" @@ -10,12 +10,14 @@ import { SessionV2 } from "@opencode-ai/core/session" import { ModelV2 } from "@opencode-ai/core/model" import { ProviderV2 } from "@opencode-ai/core/provider" import { createLLMEventPublisher } from "@opencode-ai/core/session/runner/publish-llm-event" +import { ToolPayload } from "@opencode-ai/core/session/tool-payload" const sessionID = SessionV2.ID.make("ses_tool_event_test") const base64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAAB" const capture = (providerMetadataKey = "anthropic") => { const published: Array<{ readonly type: string; readonly data: unknown }> = [] + const payloads = new Map() const events: Pick = { publish: (definition, data) => Effect.sync(() => { @@ -31,6 +33,7 @@ const capture = (providerMetadataKey = "anthropic") => { } return { published, + payloads, publisher: createLLMEventPublisher(events, { sessionID, agent: AgentV2.ID.make("build"), @@ -39,6 +42,12 @@ const capture = (providerMetadataKey = "anthropic") => { providerID: ProviderV2.ID.opencode, }, providerMetadataKey, + persistToolPayload: (body) => + Effect.sync(() => { + const hash = ToolPayload.hash(body) + payloads.set(hash, body) + return hash + }), }), } } @@ -63,31 +72,49 @@ const result = LLMEvent.toolResult({ }, }) -test("local tool success serializes media base64 once and reconstructs from structured content", async () => { - const { published, publisher } = capture() +test("local tool success stores media in the payload blob and omits base64 from the event", async () => { + const { published, payloads, publisher } = capture() await Effect.runPromise(publisher.publish(call)) await Effect.runPromise(publisher.publish(result)) - const success = published.find((event) => event.type === "session.tool.success.1") - expect(success).toBeDefined() - const serialized = JSON.stringify(success) - expect(serialized.split(base64)).toHaveLength(2) - expect(success?.data).not.toHaveProperty("result") + const successType = "session.tool.success.1" + const success = Option.getOrThrowWith( + Option.fromUndefinedOr(published.find((event) => event.type === successType)), + () => new Error(`Expected ${successType}`), + ) + const data = Schema.decodeUnknownSync(SessionEvent.Tool.Success.data)(success.data) + const payloadHash = Option.getOrThrowWith( + Option.fromUndefinedOr(data.payloadHash), + () => new Error("Expected payloadHash on thin success"), + ) + expect(JSON.stringify(success).includes(base64)).toBe(false) + expect(success.data).not.toHaveProperty("result") + expect(data.content).toEqual([{ type: "text", text: "Image read successfully" }]) - expect(success?.data).toMatchObject({ - content: [ - { type: "text", text: "Image read successfully" }, - { type: "file", uri: `data:image/png;base64,${base64}`, mime: "image/png" }, - ], - }) + const body = payloads.get(payloadHash) + expect(body?.content).toEqual([ + { type: "text", text: "Image read successfully" }, + { type: "file", uri: `data:image/png;base64,${base64}`, mime: "image/png", name: "pixel.png" }, + ]) + expect(Buffer.byteLength(JSON.stringify(data), "utf-8")).toBeLessThanOrEqual(ToolPayload.MaxEventDataBytes) }) -test("provider-executed success retains its raw provider result", async () => { - const { published, publisher } = capture() +test("provider-executed success keeps raw provider result in the payload blob only", async () => { + const { published, payloads, publisher } = capture() await Effect.runPromise(publisher.publish(LLMEvent.toolCall({ ...call, providerExecuted: true }))) await Effect.runPromise(publisher.publish(LLMEvent.toolResult({ ...result, providerExecuted: true }))) - const success = published.find((event) => event.type === "session.tool.success.1") - expect(success?.data).toHaveProperty("result") + const successType = "session.tool.success.1" + const success = Option.getOrThrowWith( + Option.fromUndefinedOr(published.find((event) => event.type === successType)), + () => new Error(`Expected ${successType}`), + ) + expect(success.data).not.toHaveProperty("result") + const data = Schema.decodeUnknownSync(SessionEvent.Tool.Success.data)(success.data) + const payloadHash = Option.getOrThrowWith( + Option.fromUndefinedOr(data.payloadHash), + () => new Error("Expected payloadHash on thin success"), + ) + expect(payloads.get(payloadHash)?.result).toBeDefined() }) test("provider metadata is flattened using the route key", async () => { @@ -179,6 +206,31 @@ test("binary failure emits no success event", async () => { expect(published.some((event) => event.type === "session.tool.failed.1")).toBe(true) }) +test("over-budget thin success fails settlement instead of publishing success", async () => { + const { published, publisher } = capture() + await Effect.runPromise(publisher.publish(call)) + await Effect.runPromise( + publisher.publish( + LLMEvent.toolResult({ + ...result, + providerMetadata: { + anthropic: { pad: "x".repeat(ToolPayload.MaxEventDataBytes) }, + }, + }), + ), + ) + expect(published.some((event) => event.type === "session.tool.success.1")).toBe(false) + const failed = published.find((event) => event.type === "session.tool.failed.1") + expect(failed).toBeDefined() + expect(failed?.data).toMatchObject({ + callID: call.id, + error: { + type: "unknown", + message: expect.stringContaining(String(ToolPayload.MaxEventDataBytes)), + }, + }) +}) + test("success event data can carry a provider-executed result", () => { const decoded = Schema.decodeUnknownSync(SessionEvent.Tool.Success.data)({ sessionID, @@ -219,12 +271,14 @@ test("content-filter finish retains failure evidence until step closeout", async ) expect(published.map((event) => event.type)).toEqual(["session.step.started.1"]) - const settlement = publisher.stepSettlement() + const settlement = Option.getOrThrowWith( + Option.fromUndefinedOr(publisher.stepSettlement()), + () => new Error("Expected content-filter settlement"), + ) expect(settlement).toMatchObject({ finish: "content-filter", tokens: { input: 8, output: 2, reasoning: 1 }, }) - if (!settlement) throw new Error("Expected content-filter settlement") await Effect.runPromise( publisher.publishStepFailure({ cost: Money.USD.make(1.25), diff --git a/packages/core/test/session-tool-progress.test.ts b/packages/core/test/session-tool-progress.test.ts index e61bb54b636c..3d889fcea1df 100644 --- a/packages/core/test/session-tool-progress.test.ts +++ b/packages/core/test/session-tool-progress.test.ts @@ -16,6 +16,7 @@ import { SessionEvent } from "@opencode-ai/core/session/event" import { SessionMessage } from "@opencode-ai/core/session/message" import { SessionProjector } from "@opencode-ai/core/session/projector" import { SessionTable, SessionMessageTable } from "@opencode-ai/core/session/sql" +import { ToolPayload } from "@opencode-ai/core/session/tool-payload" import { testEffect } from "./lib/effect" const it = testEffect(LayerNode.compile(LayerNode.group([Database.node, EventV2.node, SessionProjector.node]))) @@ -148,4 +149,104 @@ describe("Tool.Progress", () => { expect(rows.map((row) => row.type)).toContain(EventV2.versionedType(SessionEvent.Tool.Failed.type, 1)) }), ) + + it.effect("hydrates thin tool success from the session payload blob into session_message", () => + Effect.gen(function* () { + const { db } = yield* Database.Service + const service = yield* EventV2.Service + const sessionID = SessionV2.ID.make("ses_tool_payload_hydrate") + yield* db + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] }) + .onConflictDoNothing() + .run() + .pipe(Effect.orDie) + yield* db + .insert(SessionTable) + .values({ + id: sessionID, + project_id: Project.ID.global, + slug: "payload-hydrate", + directory: "/project", + title: "payload-hydrate", + version: "test", + }) + .run() + .pipe(Effect.orDie) + + const mega = "m".repeat(2 * 1024 * 1024) + const body: ToolPayload.Body = { + structured: { mime: "application/octet-stream", bytes: mega.length }, + content: [ + { type: "text", text: "ready" }, + { type: "file", uri: `data:application/octet-stream;base64,${mega}`, mime: "application/octet-stream" }, + ], + } + const payloadHash = yield* ToolPayload.insert(db, sessionID, body) + const thin = ToolPayload.preview(body) + const assistantMessageID = SessionMessage.ID.create() + + yield* service.publish(SessionEvent.Step.Started, { + sessionID, + assistantMessageID, + agent: AgentV2.ID.make("build"), + model, + }) + yield* service.publish(SessionEvent.Tool.Input.Started, { + sessionID, + assistantMessageID, + callID: "call-hydrate", + name: "read", + }) + yield* service.publish(SessionEvent.Tool.Called, { + sessionID, + assistantMessageID, + callID: "call-hydrate", + input: { path: "huge.bin" }, + executed: false, + }) + yield* service.publish(SessionEvent.Tool.Success, { + sessionID, + assistantMessageID, + callID: "call-hydrate", + structured: thin.structured, + content: thin.content, + executed: false, + payloadHash, + }) + + const row = yield* db + .select() + .from(SessionMessageTable) + .where(eq(SessionMessageTable.id, assistantMessageID)) + .get() + .pipe(Effect.orDie) + if (!row) return yield* Effect.die("Missing projected assistant") + const assistant = Schema.decodeUnknownSync(SessionMessage.Assistant)({ + ...row.data, + id: row.id, + type: row.type, + }) + expect(assistant.content[0]).toMatchObject({ + state: { + status: "completed", + structured: body.structured, + content: body.content, + }, + }) + + const eventRows = yield* db + .select() + .from(EventTable) + .where(eq(EventTable.aggregate_id, sessionID)) + .all() + .pipe(Effect.orDie) + const success = eventRows.find((item) => item.type === EventV2.versionedType(SessionEvent.Tool.Success.type, 1)) + expect(success).toBeDefined() + expect(JSON.stringify(success!.data).includes(mega)).toBe(false) + expect(Buffer.byteLength(JSON.stringify(success!.data), "utf-8")).toBeLessThanOrEqual( + ToolPayload.MaxEventDataBytes, + ) + }), + ) }) diff --git a/packages/core/test/tool-payload.test.ts b/packages/core/test/tool-payload.test.ts new file mode 100644 index 000000000000..c6b785d61db5 --- /dev/null +++ b/packages/core/test/tool-payload.test.ts @@ -0,0 +1,86 @@ +import { describe, expect } from "bun:test" +import { Effect, Exit } from "effect" +import { Database } from "@opencode-ai/core/database/database" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { EventV2 } from "@opencode-ai/core/event" +import { Project } from "@opencode-ai/core/project" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { SessionSchema } from "@opencode-ai/core/session/schema" +import { SessionTable } from "@opencode-ai/core/session/sql" +import { ToolPayload } from "@opencode-ai/core/session/tool-payload" +import { testEffect } from "./lib/effect" + +const it = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node, SessionProjector.node]))) + +const setup = (sessionID: SessionSchema.ID) => + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* db + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] }) + .onConflictDoNothing() + .run() + .pipe(Effect.orDie) + yield* db + .insert(SessionTable) + .values({ + id: sessionID, + project_id: Project.ID.global, + slug: "tool-payload-test", + directory: "/project", + title: "Tool payload test", + version: "test", + }) + .run() + .pipe(Effect.orDie) + return db + }) + +describe("ToolPayload", () => { + it.effect("round-trips a multi-megabyte body while keeping thin event data under budget", () => + Effect.gen(function* () { + const sessionID = SessionSchema.ID.make("ses_tool_payload_mega") + const db = yield* setup(sessionID) + const mega = "a".repeat(3 * 1024 * 1024) + const body: ToolPayload.Body = { + structured: { mime: "image/png", bytes: mega.length }, + content: [{ type: "file", uri: `data:image/png;base64,${mega}`, mime: "image/png", name: "huge.png" }], + } + const hash = yield* ToolPayload.insert(db, sessionID, body) + const loaded = yield* ToolPayload.load(db, sessionID, hash) + expect(loaded).toEqual(body) + + const thin = { + sessionID, + assistantMessageID: "msg_test", + callID: "call_test", + ...ToolPayload.preview(body), + executed: false, + payloadHash: hash, + } + expect(JSON.stringify(thin).includes(mega)).toBe(false) + expect(Buffer.byteLength(JSON.stringify(thin), "utf-8")).toBeLessThanOrEqual(ToolPayload.MaxEventDataBytes) + const ok = yield* ToolPayload.assertEventDataBudget(thin).pipe(Effect.exit) + expect(Exit.isSuccess(ok)).toBe(true) + }), + ) + + it.effect("fails the event-data budget when the thin payload is still oversized", () => + Effect.gen(function* () { + const oversized = { + sessionID: SessionSchema.ID.make("ses_tool_payload_budget"), + callID: "call", + assistantMessageID: "msg", + structured: { pad: "x".repeat(ToolPayload.MaxEventDataBytes) }, + content: [], + executed: false, + payloadHash: ToolPayload.Hash.make("a".repeat(64)), + } + const result = yield* ToolPayload.assertEventDataBudget(oversized).pipe(Effect.exit) + expect(Exit.isFailure(result)).toBe(true) + }), + ) +}) diff --git a/packages/schema/src/session-event.ts b/packages/schema/src/session-event.ts index aaeb6565c316..350b76feafb2 100644 --- a/packages/schema/src/session-event.ts +++ b/packages/schema/src/session-event.ts @@ -22,6 +22,7 @@ import { Snapshot } from "./snapshot.js" import { TokenUsage } from "./token-usage.js" import { SessionPending } from "./session-pending.js" import { Project } from "./project.js" +import { ToolPayload } from "./tool-payload.js" export { FileAttachment } @@ -150,6 +151,8 @@ export const InputAdmitted = Event.durable({ ...Base, inputID: SessionMessage.ID, input: SessionPending.Message, + /** When set, file attachment bytes live in the session payload blob; `input` is a descriptor preview. */ + payloadHash: optional(ToolPayload.Hash), }, }) export type InputAdmitted = typeof InputAdmitted.Type @@ -419,6 +422,8 @@ export namespace Tool { ...ToolBase, structured: Schema.Record(Schema.String, Schema.Unknown), content: Schema.Array(ToolContent), + /** When set, `structured`/`content` are previews; full body is in the session tool-payload blob. */ + payloadHash: optional(ToolPayload.Hash), }, }) export type Progress = typeof Progress.Type @@ -433,6 +438,8 @@ export namespace Tool { result: Schema.Unknown.pipe(optional), executed: Schema.Boolean, resultState: SessionMessage.ProviderState.pipe(optional), + /** When set, `structured`/`content`/`result` are previews; full body is in the session tool-payload blob. */ + payloadHash: optional(ToolPayload.Hash), }, }) export type Success = typeof Success.Type diff --git a/packages/schema/src/tool-payload.ts b/packages/schema/src/tool-payload.ts new file mode 100644 index 000000000000..cf05c78600c8 --- /dev/null +++ b/packages/schema/src/tool-payload.ts @@ -0,0 +1,17 @@ +export * as ToolPayload from "./tool-payload.js" + +import { Schema } from "effect" +import { ToolContent } from "./llm.js" +import { optional } from "./schema.js" + +/** SHA-256 hex of the canonical JSON tool payload body. */ +export const Hash = Schema.String.check(Schema.isPattern(/^[a-f0-9]{64}$/)).pipe(Schema.brand("ToolPayload.Hash")) +export type Hash = typeof Hash.Type + +/** Full tool settlement bytes stored beside the durable event log. */ +export const Body = Schema.Struct({ + structured: Schema.Record(Schema.String, Schema.Unknown), + content: Schema.Array(ToolContent), + result: Schema.Unknown.pipe(optional), +}).annotate({ identifier: "ToolPayload.Body" }) +export type Body = typeof Body.Type diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index 14990fb93373..9fb2a0010005 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -426,6 +426,29 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ }, ) }) + if (event.data.payloadHash) { + const sessionID = event.data.sessionID + const inputID = event.data.inputID + void client.api.session.pending + .list({ sessionID }) + .then((pending) => { + setStore("session", "pending", sessionID, reconcile(pending)) + const item = pending.find((entry) => entry.id === inputID) + if (!item || item.type === "compaction") return + const next = { + id: item.id, + type: item.type, + ...item.data, + time: { created: item.timeCreated }, + } + message.update(sessionID, (draft, index) => { + const position = index.get(item.id) + if (position === undefined) return message.append(draft, index, next) + draft[position] = next + }) + }) + .catch((error) => console.error("Failed to load admitted input with attachments", error)) + } break case "session.instructions.updated": const instructions = event.metadata?.instructions @@ -612,6 +635,17 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ match.state.structured = event.data.structured match.state.content = [...event.data.content] }) + if (event.data.payloadHash) + void client.api.session + .message({ sessionID: event.data.sessionID, messageID: event.data.assistantMessageID }) + .then((item) => { + message.update(event.data.sessionID, (draft, index) => { + const position = index.get(item.id) + if (position === undefined) return message.append(draft, index, item) + draft[position] = item + }) + }) + .catch((error) => console.error("Failed to load tool progress message", error)) break case "session.tool.success": message.update(event.data.sessionID, (draft, index) => { @@ -631,6 +665,17 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ match.providerResultState = event.data.resultState match.time.completed = event.created }) + if (event.data.payloadHash) + void client.api.session + .message({ sessionID: event.data.sessionID, messageID: event.data.assistantMessageID }) + .then((item) => { + message.update(event.data.sessionID, (draft, index) => { + const position = index.get(item.id) + if (position === undefined) return message.append(draft, index, item) + draft[position] = item + }) + }) + .catch((error) => console.error("Failed to load tool success message", error)) break case "session.tool.failed": message.update(event.data.sessionID, (draft, index) => { diff --git a/packages/tui/test/cli/tui/payload-hash.test.tsx b/packages/tui/test/cli/tui/payload-hash.test.tsx new file mode 100644 index 000000000000..04b73af3b8b3 --- /dev/null +++ b/packages/tui/test/cli/tui/payload-hash.test.tsx @@ -0,0 +1,391 @@ +/** @jsxImportSource @opentui/solid */ +import { describe, expect, test } from "bun:test" +import { testRender } from "@opentui/solid" +import type { OpenCodeEvent } from "@opencode-ai/client" +import { onMount, type ParentProps } from "solid-js" +import { createEffect } from "solid-js" +import { ClientProvider } from "../../../src/context/client" +import { DataProvider as DataProviderBase, useData } from "../../../src/context/data" +import { LocationProvider, useLocation } from "../../../src/context/location" +import { createApi, createEventStream, createFetch, directory, json } from "../../fixture/tui-client" +import { TestTuiContexts } from "../../fixture/tui-environment" + +const payloadHash = "a".repeat(64) + +async function wait(fn: () => boolean, timeout = 2000) { + const start = Date.now() + while (!fn()) { + if (Date.now() - start > timeout) throw new Error("timed out waiting for condition") + await Bun.sleep(10) + } +} + +function emitEvent(events: ReturnType, event: OpenCodeEvent) { + events.emit({ ...event, location: { directory } }) +} + +function durable(sessionID: string, seq = 0) { + return { aggregateID: sessionID, seq, version: 1 as const } +} + +function DataProvider(props: ParentProps) { + return ( + + + + {props.children} + + + ) +} + +function SyncLocation() { + const data = useData() + const location = useLocation() + createEffect(() => location.set(data.location.default())) + return null +} + +function ProjectProvider(props: ParentProps) { + return props.children +} + +describe("payloadHash refetch", () => { + test("tool.success with payloadHash refetches the assistant message", async () => { + const events = createEventStream() + const sessionID = "ses_payload_tool" + const assistantMessageID = "msg_assistant_tool" + const messageHits: string[] = [] + const fullContent = [{ type: "text" as const, text: "full-from-fetch" }] + const calls = createFetch((url) => { + if (url.pathname === `/api/session/${sessionID}/message`) return json({ data: [], cursor: {} }) + if (url.pathname === `/api/session/${sessionID}/message/${assistantMessageID}`) { + messageHits.push(url.pathname) + return json({ + data: { + id: assistantMessageID, + type: "assistant", + parentID: "msg_user", + agent: "build", + model: { id: "model", providerID: "provider" }, + content: [ + { + type: "tool", + callID: "call-1", + tool: "read", + state: { + status: "completed", + input: {}, + structured: { source: "fetch" }, + content: fullContent, + }, + time: { start: 1, completed: 2 }, + }, + ], + time: { created: 1 }, + }, + }) + } + return undefined + }, events) + + let data!: ReturnType + let ready!: () => void + const mounted = new Promise((resolve) => { + ready = resolve + }) + + function Probe() { + data = useData() + onMount(ready) + return + } + + const app = await testRender(() => ( + + + + + + + + + + )) + + try { + await mounted + emitEvent(events, { + id: "evt_step", + created: 1, + type: "session.step.started", + durable: durable(sessionID), + data: { + sessionID, + assistantMessageID, + agent: "build", + model: { id: "model", providerID: "provider" }, + }, + }) + emitEvent(events, { + id: "evt_input", + created: 2, + type: "session.tool.input.started", + durable: durable(sessionID, 1), + data: { sessionID, assistantMessageID, callID: "call-1", name: "read" }, + }) + emitEvent(events, { + id: "evt_called", + created: 3, + type: "session.tool.called", + durable: durable(sessionID, 2), + data: { sessionID, assistantMessageID, callID: "call-1", input: {}, executed: false }, + }) + await wait(() => { + const assistant = data.session.message.get(sessionID, assistantMessageID) + return assistant?.type === "assistant" && assistant.content[0]?.type === "tool" + }) + + emitEvent(events, { + id: "evt_success", + created: 4, + type: "session.tool.success", + durable: durable(sessionID, 3), + data: { + sessionID, + assistantMessageID, + callID: "call-1", + structured: {}, + content: [{ type: "text", text: "preview" }], + executed: false, + payloadHash, + }, + }) + + await wait(() => messageHits.length === 1) + await wait(() => { + const assistant = data.session.message.get(sessionID, assistantMessageID) + return ( + assistant?.type === "assistant" && + assistant.content[0]?.type === "tool" && + assistant.content[0].state.status === "completed" && + assistant.content[0].state.structured.source === "fetch" + ) + }) + expect(messageHits).toEqual([`/api/session/${sessionID}/message/${assistantMessageID}`]) + } finally { + app.renderer.destroy() + } + }) + + test("input.admitted with payloadHash refetches pending", async () => { + const events = createEventStream() + const sessionID = "ses_payload_admit" + const inputID = "msg_user_attach" + const pendingHits: string[] = [] + const calls = createFetch((url) => { + if (url.pathname === `/api/session/${sessionID}/message`) return json({ data: [], cursor: {} }) + if (url.pathname === `/api/session/${sessionID}/pending`) { + pendingHits.push(url.pathname) + return json({ + data: [ + { + id: inputID, + sessionID, + type: "user", + delivery: "steer", + admittedSeq: 1, + timeCreated: 1, + data: { + text: "see file", + files: [ + { + data: "full-bytes", + mime: "text/plain", + source: { type: "uri", uri: "file:///tmp/a.txt" }, + name: "a.txt", + }, + ], + }, + }, + ], + }) + } + return undefined + }, events) + + let data!: ReturnType + let ready!: () => void + const mounted = new Promise((resolve) => { + ready = resolve + }) + + function Probe() { + data = useData() + onMount(ready) + return + } + + const app = await testRender(() => ( + + + + + + + + + + )) + + try { + await mounted + emitEvent(events, { + id: "evt_admit", + created: 1, + type: "session.input.admitted", + durable: durable(sessionID, 1), + data: { + sessionID, + inputID, + payloadHash, + input: { + type: "user", + delivery: "steer", + data: { + text: "see file", + files: [ + { + data: "", + mime: "text/plain", + source: { type: "uri", uri: "file:///tmp/a.txt" }, + name: "a.txt", + }, + ], + }, + }, + }, + }) + + await wait(() => pendingHits.length === 1) + await wait(() => { + const message = data.session.message.get(sessionID, inputID) + return message?.type === "user" && message.files?.[0]?.data === "full-bytes" + }) + expect(pendingHits).toEqual([`/api/session/${sessionID}/pending`]) + } finally { + app.renderer.destroy() + } + }) + + test("thin events without payloadHash do not refetch message or pending", async () => { + const events = createEventStream() + const sessionID = "ses_payload_none" + const assistantMessageID = "msg_assistant_none" + const inputID = "msg_user_none" + const hits: string[] = [] + const calls = createFetch((url) => { + if (url.pathname === `/api/session/${sessionID}/message`) return json({ data: [], cursor: {} }) + if ( + url.pathname === `/api/session/${sessionID}/message/${assistantMessageID}` || + url.pathname === `/api/session/${sessionID}/pending` + ) { + hits.push(url.pathname) + } + return undefined + }, events) + + let data!: ReturnType + let ready!: () => void + const mounted = new Promise((resolve) => { + ready = resolve + }) + + function Probe() { + data = useData() + onMount(ready) + return + } + + const app = await testRender(() => ( + + + + + + + + + + )) + + try { + await mounted + emitEvent(events, { + id: "evt_admit", + created: 1, + type: "session.input.admitted", + durable: durable(sessionID, 1), + data: { + sessionID, + inputID, + input: { type: "user", delivery: "steer", data: { text: "hello" } }, + }, + }) + await wait(() => data.session.message.get(sessionID, inputID)?.type === "user") + + emitEvent(events, { + id: "evt_step", + created: 2, + type: "session.step.started", + durable: durable(sessionID, 2), + data: { + sessionID, + assistantMessageID, + agent: "build", + model: { id: "model", providerID: "provider" }, + }, + }) + emitEvent(events, { + id: "evt_input", + created: 3, + type: "session.tool.input.started", + durable: durable(sessionID, 3), + data: { sessionID, assistantMessageID, callID: "call-1", name: "read" }, + }) + emitEvent(events, { + id: "evt_called", + created: 4, + type: "session.tool.called", + durable: durable(sessionID, 4), + data: { sessionID, assistantMessageID, callID: "call-1", input: {}, executed: false }, + }) + emitEvent(events, { + id: "evt_success", + created: 5, + type: "session.tool.success", + durable: durable(sessionID, 5), + data: { + sessionID, + assistantMessageID, + callID: "call-1", + structured: {}, + content: [{ type: "text", text: "inline" }], + executed: false, + }, + }) + await wait(() => { + const assistant = data.session.message.get(sessionID, assistantMessageID) + return ( + assistant?.type === "assistant" && + assistant.content[0]?.type === "tool" && + assistant.content[0].state.status === "completed" + ) + }) + await Bun.sleep(50) + expect(hits).toEqual([]) + } finally { + app.renderer.destroy() + } + }) +}) diff --git a/specs/v2/event-stream-architecture.md b/specs/v2/event-stream-architecture.md index 34eb8f137cc1..1871c561f962 100644 --- a/specs/v2/event-stream-architecture.md +++ b/specs/v2/event-stream-architecture.md @@ -121,6 +121,10 @@ Tune capacity separately using observed: Add a byte budget only if measurements show event count is an inadequate memory safeguard. +### Producer event-data budget (tool settlement) + +Large tool settlement bodies are not carried on the durable event or SSE frame. Core persists the full body in the session-scoped `session_tool_payload` table and publishes a thin `session.tool.success` / `session.tool.progress` projection (IDs, optional text preview, `payloadHash`). Projectors hydrate `session_message` from that blob. Encoded thin event `data` must stay within **256 KiB** UTF-8; settlement fails rather than truncating opaque payloads. Clients that need the full body use `GET /api/session/:sessionID/message/:messageID`. This is separate from `ToolOutputStore`, which only bounds the provider-facing text channel and must not be used as a durable event sidecar. + ## Feed Lifecycle ### Server scope diff --git a/specs/v2/schema-changelog.md b/specs/v2/schema-changelog.md index e5ed230d5176..f253dd667dab 100644 --- a/specs/v2/schema-changelog.md +++ b/specs/v2/schema-changelog.md @@ -26,7 +26,7 @@ Compatibility: Compatibility: - `20260709190621_session_pending_table` drops `session_input` (including consumed ledger rows, any in-flight pending work, and whatever historical index variant the database carried) and creates the empty `session_pending` table. V2 storage is beta; no compatibility or data retention is attempted. -- Durable event names and payloads are unchanged; `session.input.admitted` still records the full admitted message including delivery. +- Durable event names are unchanged. `session.input.admitted` records the admitted message including delivery; large file attachment bytes are stored in the session-scoped payload blob (`payloadHash`) and omitted from the event projection. Pending rows and message fetches still expose full attachment bytes. - Promise, Effect, and legacy JavaScript SDK surfaces are regenerated; `SessionInput*` generated schema names become `SessionPending*` while event-derived names keep their `session.input.*` vocabulary. ## 2026-07-05: Rename Session Context Contracts To Instructions