From 38cc62a0c5c61819fbd12d17f06bd700e801bca9 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 07:02:44 +1000 Subject: [PATCH 01/17] feat(core): bound tool and admitted event payloads via session blobs Persist large tool settlement and prompt attachment bodies in session_tool_payload, publish thin durable events with payloadHash, and hydrate projectors plus TUI fetches so SSE and the event log stay within a 256 KiB event-data budget without changing ToolOutputStore media rules. Co-authored-by: Cursor --- .../client/src/promise/generated/types.ts | 4 +- packages/core/src/database/migration.gen.ts | 1 + .../20260718010000_session_tool_payload.ts | 19 +++ packages/core/src/database/schema.gen.ts | 9 ++ packages/core/src/session/pending.ts | 31 +++- packages/core/src/session/projector.ts | 48 ++++++- packages/core/src/session/runner/llm.ts | 29 +++- .../src/session/runner/publish-llm-event.ts | 35 ++++- packages/core/src/session/sql.ts | 14 ++ packages/core/src/session/tool-payload.ts | 135 ++++++++++++++++++ .../test/session-pending-attachments.test.ts | 118 +++++++++++++++ .../test/session-runner-tool-events.test.ts | 42 ++++-- packages/core/test/tool-payload.test.ts | 86 +++++++++++ packages/schema/src/session-event.ts | 7 + packages/schema/src/tool-payload.ts | 17 +++ packages/tui/src/context/data.tsx | 50 +++++++ specs/v2/event-stream-architecture.md | 4 + specs/v2/schema-changelog.md | 2 +- 18 files changed, 619 insertions(+), 32 deletions(-) create mode 100644 packages/core/src/database/migration/20260718010000_session_tool_payload.ts create mode 100644 packages/core/src/session/tool-payload.ts create mode 100644 packages/core/test/session-pending-attachments.test.ts create mode 100644 packages/core/test/tool-payload.test.ts create mode 100644 packages/schema/src/tool-payload.ts 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..a15a2b7181b5 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,26 @@ 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: "" })), + }, + } +} + export class LifecycleConflict extends Schema.TaggedErrorClass()( "SessionPending.LifecycleConflict", { @@ -126,15 +141,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 +173,18 @@ export const admit = Effect.fn("SessionPending.admit")(function* ( } const promoted = yield* promotedFromHistory(db, request.sessionID, request.id) if (promoted !== undefined) return promoted + const payloadHash = yield* ToolPayload.insertJson( + db, + request.sessionID, + encodeMessage(request.input) as Schema.Json, + ) + const thin = stripAttachmentBytes(request.input) return yield* events .publish(SessionEvent.InputAdmitted, { inputID: request.id, sessionID: request.sessionID, - input: request.input, + input: thin, + payloadHash, }) .pipe( Effect.flatMap((event) => { diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts index 62bd34ecc6c1..a4c78966f3e9 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,16 @@ 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) => + Effect.gen(function* () { + yield* run(db, yield* hydrateToolPayload(db, event)) + }), + ) + yield* events.project(SessionEvent.Tool.Success, (event) => + Effect.gen(function* () { + yield* run(db, yield* hydrateToolPayload(db, event)) + }), + ) 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..ff00966b7dbc 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" @@ -125,6 +126,7 @@ const layer = Layer.effect( providerMetadataKey: model.route.providerMetadataKey ?? model.provider, snapshot: startSnapshot, assistantMessageID, + persistToolPayload: (body) => ToolPayload.insert(db, session.id, body), }) const publication = Semaphore.makeUnsafe(1) // Durable publishes are serialized so tool fibers and step settlement never interleave @@ -161,12 +163,27 @@ 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* ToolPayload.insert(db, session.id, body) + const thin = ToolPayload.preview(body) + const data = { + sessionID: session.id, + assistantMessageID, + callID: event.id, + structured: thin.structured, + content: thin.content, + payloadHash, + } + const overBudget = yield* ToolPayload.assertEventDataBudget(data).pipe( + Effect.as(false), + Effect.catchTag("ToolPayload.OverBudgetError", () => Effect.succeed(true)), + ) + if (overBudget) return + yield* events.publish(SessionEvent.Tool.Progress, data) }), ), }), diff --git a/packages/core/src/session/runner/publish-llm-event.ts b/packages/core/src/session/runner/publish-llm-event.ts index 6a71f5d88872..bd734a330c3e 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 => @@ -382,15 +385,39 @@ export const createLLMEventPublisher = (events: Pick Effect.succeed(error)), + ) + if (overBudget) { + yield* events.publish(SessionEvent.Tool.Failed, { + sessionID: input.sessionID, + assistantMessageID: tool.assistantMessageID, + callID: event.id, + error: { type: "unknown", message: overBudget.message }, + executed, + resultState, + }) + return + } + yield* events.publish(SessionEvent.Tool.Success, data) return } case "tool-error": { diff --git a/packages/core/src/session/sql.ts b/packages/core/src/session/sql.ts index 21d0ab5f2675..6b4335725af6 100644 --- a/packages/core/src/session/sql.ts +++ b/packages/core/src/session/sql.ts @@ -172,6 +172,20 @@ export const InstructionBlobTable = sqliteTable("instruction_blob", { value: text({ mode: "json" }).$type(), }) +/** 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().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..1cbb19da3e9b --- /dev/null +++ b/packages/core/src/session/tool-payload.ts @@ -0,0 +1,135 @@ +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"] + +export function hash(value: Body): Hash { + return Hash.make(createHash("sha256").update(canonical(value as Schema.Json)).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, body as Schema.Json) +}) + +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(entry as Schema.Json)}`) + .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..d1fab96f6d6e --- /dev/null +++ b/packages/core/test/session-pending-attachments.test.ts @@ -0,0 +1,118 @@ +import { describe, expect } from "bun:test" +import { and, eq } from "drizzle-orm" +import { Effect } 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 { 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]))) + +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: "user", + delivery: "steer", + data: { + text: "see file", + files: [ + { + data: mega, + mime: "application/octet-stream", + source: { type: "uri", uri: "file:///tmp/huge.bin" }, + name: "huge.bin", + }, + ], + }, + }, + }) + + expect(admitted.type).toBe("user") + if (admitted.type === "user") expect(admitted.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 = rows[0]!.data as { + payloadHash?: string + input: { type: string; data: { files?: Array<{ data: string }> } } + } + expect(eventData.payloadHash).toBeDefined() + expect(eventData.input.data.files?.[0]?.data).toBe("") + expect(JSON.stringify(eventData).includes(mega)).toBe(false) + + const pending = yield* SessionPending.list(db, sessionID) + const row = pending.find((item) => item.id === inputID) + expect(row?.type).toBe("user") + if (row?.type === "user") expect(row.data.files?.[0]?.data).toBe(mega) + }), + ) + + it.effect("stripAttachmentBytes clears only file data fields", () => + Effect.sync(() => { + const stripped = SessionPending.stripAttachmentBytes({ + type: "user", + delivery: "steer", + data: { + text: "hi", + files: [ + { + data: "AAAA", + mime: "text/plain", + source: { type: "uri", uri: "file:///tmp/a.txt" }, + name: "a.txt", + }, + ], + }, + }) + expect(stripped.type).toBe("user") + if (stripped.type === "user") { + expect(stripped.data.text).toBe("hi") + expect(stripped.data.files?.[0]?.data).toBe("") + expect(stripped.data.files?.[0]?.name).toBe("a.txt") + } + }), + ) +}) diff --git a/packages/core/test/session-runner-tool-events.test.ts b/packages/core/test/session-runner-tool-events.test.ts index 93600c3058d9..9788b63759f6 100644 --- a/packages/core/test/session-runner-tool-events.test.ts +++ b/packages/core/test/session-runner-tool-events.test.ts @@ -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,38 @@ 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) + const data = success?.data as { + payloadHash?: string + content: ReadonlyArray<{ type: string; text?: string }> + } + expect(data.payloadHash).toBeDefined() + 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(data.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") + expect(success?.data).not.toHaveProperty("result") + const hash = (success?.data as { payloadHash: string }).payloadHash + expect(payloads.get(hash)?.result).toBeDefined() }) test("provider metadata is flattened using the route key", async () => { 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..3431490ac5f2 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -426,6 +426,34 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ }, ) }) + if (event.data.payloadHash) + void client.api.session.pending + .list({ sessionID: event.data.sessionID }) + .then((pending) => { + setStore("session", "pending", event.data.sessionID, reconcile(pending)) + const item = pending.find((entry) => entry.id === event.data.inputID) + if (!item || item.type === "compaction") return + message.update(event.data.sessionID, (draft, index) => { + const position = index.get(item.id) + const next = + item.type === "user" + ? { + id: item.id, + type: "user" as const, + ...item.data, + time: { created: item.timeCreated }, + } + : { + id: item.id, + type: "synthetic" as const, + ...item.data, + time: { created: item.timeCreated }, + } + 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 +640,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 +670,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/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 From 86f6d258c15e781830ce726698513764376804a1 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 07:11:14 +1000 Subject: [PATCH 02/17] refactor(core): fold tool payload budget into Effect pipes Drop boolean/as escape hatches; assert then publish, recover OverBudgetError in-band. Co-authored-by: Cursor --- packages/core/src/session/runner/llm.ts | 8 +++--- .../src/session/runner/publish-llm-event.ts | 27 +++++++++---------- 2 files changed, 15 insertions(+), 20 deletions(-) diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index ff00966b7dbc..d38809f2a785 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -178,12 +178,10 @@ const layer = Layer.effect( content: thin.content, payloadHash, } - const overBudget = yield* ToolPayload.assertEventDataBudget(data).pipe( - Effect.as(false), - Effect.catchTag("ToolPayload.OverBudgetError", () => Effect.succeed(true)), + yield* ToolPayload.assertEventDataBudget(data).pipe( + Effect.andThen(events.publish(SessionEvent.Tool.Progress, data)), + Effect.catchTag("ToolPayload.OverBudgetError", () => Effect.void), ) - if (overBudget) return - yield* events.publish(SessionEvent.Tool.Progress, data) }), ), }), diff --git a/packages/core/src/session/runner/publish-llm-event.ts b/packages/core/src/session/runner/publish-llm-event.ts index bd734a330c3e..90f8980ac84f 100644 --- a/packages/core/src/session/runner/publish-llm-event.ts +++ b/packages/core/src/session/runner/publish-llm-event.ts @@ -402,22 +402,19 @@ export const createLLMEventPublisher = (events: Pick Effect.succeed(error)), + yield* ToolPayload.assertEventDataBudget(data).pipe( + Effect.andThen(events.publish(SessionEvent.Tool.Success, data)), + Effect.catchTag("ToolPayload.OverBudgetError", (error) => + events.publish(SessionEvent.Tool.Failed, { + sessionID: input.sessionID, + assistantMessageID: tool.assistantMessageID, + callID: event.id, + error: { type: "unknown", message: error.message }, + executed, + resultState, + }), + ), ) - if (overBudget) { - yield* events.publish(SessionEvent.Tool.Failed, { - sessionID: input.sessionID, - assistantMessageID: tool.assistantMessageID, - callID: event.id, - error: { type: "unknown", message: overBudget.message }, - executed, - resultState, - }) - return - } - yield* events.publish(SessionEvent.Tool.Success, data) return } case "tool-error": { From b449d8442f022e434f07b74fd8d4fd576a91188f Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 07:16:42 +1000 Subject: [PATCH 03/17] refactor(core): polish bound payload Effect and admit paths Encode via Schema instead of casts, blob admits only when attachments carry bytes, treat over-budget progress as a defect, and tighten projector/SQL typing. Co-authored-by: Cursor --- packages/core/src/session/pending.ts | 17 ++++--- packages/core/src/session/projector.ts | 8 +--- packages/core/src/session/runner/llm.ts | 3 +- packages/core/src/session/sql.ts | 3 +- packages/core/src/session/tool-payload.ts | 19 ++++---- .../test/session-pending-attachments.test.ts | 48 +++++++++++++++++++ 6 files changed, 75 insertions(+), 23 deletions(-) diff --git a/packages/core/src/session/pending.ts b/packages/core/src/session/pending.ts index a15a2b7181b5..fd5ef0271773 100644 --- a/packages/core/src/session/pending.ts +++ b/packages/core/src/session/pending.ts @@ -52,6 +52,13 @@ export function stripAttachmentBytes(input: Message): Message { } } +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", { @@ -173,18 +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 payloadHash = yield* ToolPayload.insertJson( - db, - request.sessionID, - encodeMessage(request.input) as Schema.Json, - ) 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: thin, - payloadHash, + ...(payloadHash !== undefined ? { payloadHash } : {}), }) .pipe( Effect.flatMap((event) => { diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts index a4c78966f3e9..7520cc4bcc8b 100644 --- a/packages/core/src/session/projector.ts +++ b/packages/core/src/session/projector.ts @@ -732,14 +732,10 @@ const layer = Layer.effectDiscard( 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) => - Effect.gen(function* () { - yield* run(db, yield* hydrateToolPayload(db, event)) - }), + hydrateToolPayload(db, event).pipe(Effect.flatMap((hydrated) => run(db, hydrated))), ) yield* events.project(SessionEvent.Tool.Success, (event) => - Effect.gen(function* () { - yield* run(db, yield* hydrateToolPayload(db, 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)) diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index d38809f2a785..3d152518fea9 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -178,9 +178,10 @@ const layer = Layer.effect( 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.catchTag("ToolPayload.OverBudgetError", () => Effect.void), + Effect.orDie, ) }), ), diff --git a/packages/core/src/session/sql.ts b/packages/core/src/session/sql.ts index 6b4335725af6..7eea83b73590 100644 --- a/packages/core/src/session/sql.ts +++ b/packages/core/src/session/sql.ts @@ -12,6 +12,7 @@ import type { MessageID, PartID, SessionV1 } from "../v1/session" import { WorkspaceV2 } from "../workspace" import { Timestamps } from "../database/schema.sql" import type { Instruction } from "@opencode-ai/schema/instruction" +import type { ToolPayload } from "@opencode-ai/schema/tool-payload" import type { Session } from "@opencode-ai/schema/session" import type { SyntheticData, UserData } from "@opencode-ai/schema/session-pending" import type { RevertV1 } from "@opencode-ai/schema/session-revert" @@ -180,7 +181,7 @@ export const SessionToolPayloadTable = sqliteTable( .$type() .notNull() .references(() => SessionTable.id, { onDelete: "cascade" }), - hash: text().notNull(), + hash: text().$type().notNull(), value: text({ mode: "json" }).$type().notNull(), }, (table) => [primaryKey({ columns: [table.session_id, table.hash] })], diff --git a/packages/core/src/session/tool-payload.ts b/packages/core/src/session/tool-payload.ts index 1cbb19da3e9b..ef36b0b1b80a 100644 --- a/packages/core/src/session/tool-payload.ts +++ b/packages/core/src/session/tool-payload.ts @@ -3,16 +3,14 @@ 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 { Body, Hash } 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 +export { Hash, Body } +export type Hash = typeof Hash.Type +export type Body = typeof Body.Type /** UTF-8 byte ceiling for encoded durable tool event `data` after thinning. */ export const MaxEventDataBytes = 256 * 1024 @@ -35,8 +33,11 @@ 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(value as Schema.Json)).digest("hex")) + return Hash.make(createHash("sha256").update(canonical(decodeJson(encodeBody(value)))).digest("hex")) } export const insertJson = Effect.fn("ToolPayload.insertJson")(function* ( @@ -83,7 +84,7 @@ export const insert = Effect.fn("ToolPayload.insert")(function* ( sessionID: SessionSchema.ID, body: Body, ) { - return yield* insertJson(db, sessionID, body as Schema.Json) + return yield* insertJson(db, sessionID, decodeJson(encodeBody(body))) }) export const load = Effect.fn("ToolPayload.load")(function* ( @@ -129,7 +130,7 @@ function canonical(value: Schema.Json): string { 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(entry as Schema.Json)}`) + .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 index d1fab96f6d6e..06860e940bf1 100644 --- a/packages/core/test/session-pending-attachments.test.ts +++ b/packages/core/test/session-pending-attachments.test.ts @@ -115,4 +115,52 @@ describe("SessionPending attachment bounding", () => { } }), ) + + 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: "user", + 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 = rows[0]!.data as { payloadHash?: string; input: { data: { text: string } } } + expect(eventData.payloadHash).toBeUndefined() + expect(eventData.input.data.text).toBe("hello") + }), + ) }) From d6e84429303fcc35c526570df091122d00577ac7 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 07:17:13 +1000 Subject: [PATCH 04/17] fix(core): re-export ToolPayload Hash/Body without type conflicts Co-authored-by: Cursor --- packages/core/src/session/tool-payload.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/packages/core/src/session/tool-payload.ts b/packages/core/src/session/tool-payload.ts index ef36b0b1b80a..a4ffefb6fec7 100644 --- a/packages/core/src/session/tool-payload.ts +++ b/packages/core/src/session/tool-payload.ts @@ -9,8 +9,7 @@ import { SessionToolPayloadTable } from "./sql" import type { SessionSchema } from "./schema" export { Hash, Body } -export type Hash = typeof Hash.Type -export type Body = typeof Body.Type +export type { Hash, Body } /** UTF-8 byte ceiling for encoded durable tool event `data` after thinning. */ export const MaxEventDataBytes = 256 * 1024 From 81e2f873f0c991f63cc7479cd08e0794764f44fa Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 07:17:31 +1000 Subject: [PATCH 05/17] fix(core): restore ToolPayload schema re-export pattern Co-authored-by: Cursor --- packages/core/src/session/tool-payload.ts | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/packages/core/src/session/tool-payload.ts b/packages/core/src/session/tool-payload.ts index a4ffefb6fec7..7f036460ad30 100644 --- a/packages/core/src/session/tool-payload.ts +++ b/packages/core/src/session/tool-payload.ts @@ -3,13 +3,16 @@ export * as ToolPayload from "./tool-payload" import { createHash } from "crypto" import { Effect, Schema } from "effect" import { eq, and } from "drizzle-orm" -import { Body, Hash } from "@opencode-ai/schema/tool-payload" +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 { Hash, Body } -export type { Hash, Body } +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 From cc1e385a4734a4959824ac336f2c52aaba7177b7 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 07:23:39 +1000 Subject: [PATCH 06/17] test(core): verify tool payload hydrate and over-budget fail path Cover projector rebuild of thin tool.success from session blobs, and publisher fail-not-truncate when thin event data still exceeds budget. Co-authored-by: Cursor --- .../test/session-runner-tool-events.test.ts | 25 +++++ .../core/test/session-tool-progress.test.ts | 101 ++++++++++++++++++ 2 files changed, 126 insertions(+) diff --git a/packages/core/test/session-runner-tool-events.test.ts b/packages/core/test/session-runner-tool-events.test.ts index 9788b63759f6..3a12ae22c265 100644 --- a/packages/core/test/session-runner-tool-events.test.ts +++ b/packages/core/test/session-runner-tool-events.test.ts @@ -195,6 +195,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, 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, + ) + }), + ) }) From 783e3e59813c63f8c4d18d10508f961be338d24e Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 07:28:55 +1000 Subject: [PATCH 07/17] refactor(core): share tool payload persist binding in runner Use one local persistToolPayload for publisher settlement and progress. Co-authored-by: Cursor --- packages/core/src/session/runner/llm.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 3d152518fea9..cc42fa0727cc 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -117,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, @@ -126,7 +127,7 @@ const layer = Layer.effect( providerMetadataKey: model.route.providerMetadataKey ?? model.provider, snapshot: startSnapshot, assistantMessageID, - persistToolPayload: (body) => ToolPayload.insert(db, session.id, body), + persistToolPayload, }) const publication = Semaphore.makeUnsafe(1) // Durable publishes are serialized so tool fibers and step settlement never interleave @@ -168,7 +169,7 @@ const layer = Layer.effect( structured: { ...update.structured }, content: [...update.content], } - const payloadHash = yield* ToolPayload.insert(db, session.id, body) + const payloadHash = yield* persistToolPayload(body) const thin = ToolPayload.preview(body) const data = { sessionID: session.id, From 9d356b2fe0b9c936b1fd30ebe10c78c864209408 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 07:45:17 +1000 Subject: [PATCH 08/17] test(core): decode event fixtures instead of casting Use Schema.decodeUnknownSync for tool success and admitted event data. Co-authored-by: Cursor --- .../test/session-pending-attachments.test.ts | 14 +++++++------ .../test/session-runner-tool-events.test.ts | 21 +++++++++++-------- 2 files changed, 20 insertions(+), 15 deletions(-) diff --git a/packages/core/test/session-pending-attachments.test.ts b/packages/core/test/session-pending-attachments.test.ts index 06860e940bf1..8e5ac0089812 100644 --- a/packages/core/test/session-pending-attachments.test.ts +++ b/packages/core/test/session-pending-attachments.test.ts @@ -1,6 +1,6 @@ import { describe, expect } from "bun:test" import { and, eq } from "drizzle-orm" -import { Effect } from "effect" +import { Effect, 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" @@ -9,6 +9,7 @@ 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" @@ -75,11 +76,10 @@ describe("SessionPending attachment bounding", () => { .all() .pipe(Effect.orDie) expect(rows).toHaveLength(1) - const eventData = rows[0]!.data as { - payloadHash?: string - input: { type: string; data: { files?: Array<{ data: string }> } } - } + const eventData = Schema.decodeUnknownSync(SessionEvent.InputAdmitted.data)(rows[0]!.data) expect(eventData.payloadHash).toBeDefined() + expect(eventData.input.type).toBe("user") + if (eventData.input.type !== "user") throw new Error("Expected user input on admitted event") expect(eventData.input.data.files?.[0]?.data).toBe("") expect(JSON.stringify(eventData).includes(mega)).toBe(false) @@ -158,8 +158,10 @@ describe("SessionPending attachment bounding", () => { .all() .pipe(Effect.orDie) expect(rows).toHaveLength(1) - const eventData = rows[0]!.data as { payloadHash?: string; input: { data: { text: string } } } + const eventData = Schema.decodeUnknownSync(SessionEvent.InputAdmitted.data)(rows[0]!.data) expect(eventData.payloadHash).toBeUndefined() + expect(eventData.input.type).toBe("user") + if (eventData.input.type !== "user") throw new Error("Expected user input on admitted event") expect(eventData.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 3a12ae22c265..c153b395f05a 100644 --- a/packages/core/test/session-runner-tool-events.test.ts +++ b/packages/core/test/session-runner-tool-events.test.ts @@ -79,16 +79,15 @@ test("local tool success stores media in the payload blob and omits base64 from const success = published.find((event) => event.type === "session.tool.success.1") expect(success).toBeDefined() - const data = success?.data as { - payloadHash?: string - content: ReadonlyArray<{ type: string; text?: string }> - } + if (!success) throw new Error("Expected session.tool.success.1") + const data = Schema.decodeUnknownSync(SessionEvent.Tool.Success.data)(success.data) expect(data.payloadHash).toBeDefined() expect(JSON.stringify(success).includes(base64)).toBe(false) - expect(success?.data).not.toHaveProperty("result") + expect(success.data).not.toHaveProperty("result") expect(data.content).toEqual([{ type: "text", text: "Image read successfully" }]) - const body = payloads.get(data.payloadHash!) + if (!data.payloadHash) throw new Error("Expected payloadHash on thin success") + const body = payloads.get(data.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" }, @@ -101,9 +100,13 @@ test("provider-executed success keeps raw provider result in the payload blob on 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).not.toHaveProperty("result") - const hash = (success?.data as { payloadHash: string }).payloadHash - expect(payloads.get(hash)?.result).toBeDefined() + expect(success).toBeDefined() + if (!success) throw new Error("Expected session.tool.success.1") + expect(success.data).not.toHaveProperty("result") + const data = Schema.decodeUnknownSync(SessionEvent.Tool.Success.data)(success.data) + expect(data.payloadHash).toBeDefined() + if (!data.payloadHash) throw new Error("Expected payloadHash on thin success") + expect(payloads.get(data.payloadHash)?.result).toBeDefined() }) test("provider metadata is flattened using the route key", async () => { From 4c8df72e0e29994ec5bf76014a041e67ceba09c4 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 07:46:37 +1000 Subject: [PATCH 09/17] fix(core): omit undefined resultState on thin tool events optionalKey schemas reject explicit undefined; keep publishes and Schema.decodeUnknownSync tests aligned. Co-authored-by: Cursor --- packages/core/src/session/runner/publish-llm-event.ts | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/packages/core/src/session/runner/publish-llm-event.ts b/packages/core/src/session/runner/publish-llm-event.ts index 90f8980ac84f..cfe84a74b00f 100644 --- a/packages/core/src/session/runner/publish-llm-event.ts +++ b/packages/core/src/session/runner/publish-llm-event.ts @@ -381,7 +381,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, @@ -433,7 +434,7 @@ export const createLLMEventPublisher = (events: Pick Date: Sat, 18 Jul 2026 08:17:46 +1000 Subject: [PATCH 10/17] test(tui): lock payloadHash message and pending refetch Cover thin tool.success and input.admitted refetching full bodies, and assert events without payloadHash do not hit those endpoints. Co-authored-by: Cursor --- .../tui/test/cli/tui/payload-hash.test.tsx | 391 ++++++++++++++++++ 1 file changed, 391 insertions(+) create mode 100644 packages/tui/test/cli/tui/payload-hash.test.tsx 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() + } + }) +}) From cb892f27c1df01cbf7fc4d16bc264f21aab4db8e Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 09:26:55 +1000 Subject: [PATCH 11/17] test(core): narrow with assert helpers for linear test bodies Co-authored-by: Cursor --- packages/core/test/lib/assert.ts | 14 +++++++++++ .../test/session-pending-attachments.test.ts | 24 +++++++++---------- .../test/session-runner-tool-events.test.ts | 15 +++++------- 3 files changed, 31 insertions(+), 22 deletions(-) create mode 100644 packages/core/test/lib/assert.ts diff --git a/packages/core/test/lib/assert.ts b/packages/core/test/lib/assert.ts new file mode 100644 index 000000000000..d56b607f8967 --- /dev/null +++ b/packages/core/test/lib/assert.ts @@ -0,0 +1,14 @@ +import { expect } from "bun:test" + +/** Runtime expect + TypeScript narrow — keeps test bodies linear. */ +export function assertDefined(value: T | null | undefined): asserts value is T { + expect(value).toBeDefined() +} + +/** Narrow a `{ type }` discriminated union after asserting the tag. */ +export function assertTag( + value: A, + type: T, +): asserts value is Extract { + expect(value.type).toBe(type) +} diff --git a/packages/core/test/session-pending-attachments.test.ts b/packages/core/test/session-pending-attachments.test.ts index 8e5ac0089812..1613041458ba 100644 --- a/packages/core/test/session-pending-attachments.test.ts +++ b/packages/core/test/session-pending-attachments.test.ts @@ -15,6 +15,7 @@ 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 { assertDefined, assertTag } from "./lib/assert" import { testEffect } from "./lib/effect" const it = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node, SessionProjector.node]))) @@ -77,16 +78,16 @@ describe("SessionPending attachment bounding", () => { .pipe(Effect.orDie) expect(rows).toHaveLength(1) const eventData = Schema.decodeUnknownSync(SessionEvent.InputAdmitted.data)(rows[0]!.data) - expect(eventData.payloadHash).toBeDefined() - expect(eventData.input.type).toBe("user") - if (eventData.input.type !== "user") throw new Error("Expected user input on admitted event") + assertDefined(eventData.payloadHash) + assertTag(eventData.input, "user") expect(eventData.input.data.files?.[0]?.data).toBe("") expect(JSON.stringify(eventData).includes(mega)).toBe(false) const pending = yield* SessionPending.list(db, sessionID) const row = pending.find((item) => item.id === inputID) - expect(row?.type).toBe("user") - if (row?.type === "user") expect(row.data.files?.[0]?.data).toBe(mega) + assertDefined(row) + assertTag(row, "user") + expect(row.data.files?.[0]?.data).toBe(mega) }), ) @@ -107,12 +108,10 @@ describe("SessionPending attachment bounding", () => { ], }, }) - expect(stripped.type).toBe("user") - if (stripped.type === "user") { - expect(stripped.data.text).toBe("hi") - expect(stripped.data.files?.[0]?.data).toBe("") - expect(stripped.data.files?.[0]?.name).toBe("a.txt") - } + assertTag(stripped, "user") + expect(stripped.data.text).toBe("hi") + expect(stripped.data.files?.[0]?.data).toBe("") + expect(stripped.data.files?.[0]?.name).toBe("a.txt") }), ) @@ -160,8 +159,7 @@ describe("SessionPending attachment bounding", () => { expect(rows).toHaveLength(1) const eventData = Schema.decodeUnknownSync(SessionEvent.InputAdmitted.data)(rows[0]!.data) expect(eventData.payloadHash).toBeUndefined() - expect(eventData.input.type).toBe("user") - if (eventData.input.type !== "user") throw new Error("Expected user input on admitted event") + assertTag(eventData.input, "user") expect(eventData.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 c153b395f05a..c47c98a4ab4f 100644 --- a/packages/core/test/session-runner-tool-events.test.ts +++ b/packages/core/test/session-runner-tool-events.test.ts @@ -11,6 +11,7 @@ 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" +import { assertDefined } from "./lib/assert" const sessionID = SessionV2.ID.make("ses_tool_event_test") const base64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAAB" @@ -78,15 +79,13 @@ test("local tool success stores media in the payload blob and omits base64 from await Effect.runPromise(publisher.publish(result)) const success = published.find((event) => event.type === "session.tool.success.1") - expect(success).toBeDefined() - if (!success) throw new Error("Expected session.tool.success.1") + assertDefined(success) const data = Schema.decodeUnknownSync(SessionEvent.Tool.Success.data)(success.data) - expect(data.payloadHash).toBeDefined() + assertDefined(data.payloadHash) expect(JSON.stringify(success).includes(base64)).toBe(false) expect(success.data).not.toHaveProperty("result") expect(data.content).toEqual([{ type: "text", text: "Image read successfully" }]) - if (!data.payloadHash) throw new Error("Expected payloadHash on thin success") const body = payloads.get(data.payloadHash) expect(body?.content).toEqual([ { type: "text", text: "Image read successfully" }, @@ -100,12 +99,10 @@ test("provider-executed success keeps raw provider result in the payload blob on 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).toBeDefined() - if (!success) throw new Error("Expected session.tool.success.1") + assertDefined(success) expect(success.data).not.toHaveProperty("result") const data = Schema.decodeUnknownSync(SessionEvent.Tool.Success.data)(success.data) - expect(data.payloadHash).toBeDefined() - if (!data.payloadHash) throw new Error("Expected payloadHash on thin success") + assertDefined(data.payloadHash) expect(payloads.get(data.payloadHash)?.result).toBeDefined() }) @@ -268,7 +265,7 @@ test("content-filter finish retains failure evidence until step closeout", async finish: "content-filter", tokens: { input: 8, output: 2, reasoning: 1 }, }) - if (!settlement) throw new Error("Expected content-filter settlement") + assertDefined(settlement) await Effect.runPromise( publisher.publishStepFailure({ cost: Money.USD.make(1.25), From 84136940b4cd62b0f3bd390e8a55f0527a8dea05 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 09:27:39 +1000 Subject: [PATCH 12/17] Revert "test(core): narrow with assert helpers for linear test bodies" This reverts commit cb892f27c1df01cbf7fc4d16bc264f21aab4db8e. --- packages/core/test/lib/assert.ts | 14 ----------- .../test/session-pending-attachments.test.ts | 24 ++++++++++--------- .../test/session-runner-tool-events.test.ts | 15 +++++++----- 3 files changed, 22 insertions(+), 31 deletions(-) delete mode 100644 packages/core/test/lib/assert.ts diff --git a/packages/core/test/lib/assert.ts b/packages/core/test/lib/assert.ts deleted file mode 100644 index d56b607f8967..000000000000 --- a/packages/core/test/lib/assert.ts +++ /dev/null @@ -1,14 +0,0 @@ -import { expect } from "bun:test" - -/** Runtime expect + TypeScript narrow — keeps test bodies linear. */ -export function assertDefined(value: T | null | undefined): asserts value is T { - expect(value).toBeDefined() -} - -/** Narrow a `{ type }` discriminated union after asserting the tag. */ -export function assertTag( - value: A, - type: T, -): asserts value is Extract { - expect(value.type).toBe(type) -} diff --git a/packages/core/test/session-pending-attachments.test.ts b/packages/core/test/session-pending-attachments.test.ts index 1613041458ba..8e5ac0089812 100644 --- a/packages/core/test/session-pending-attachments.test.ts +++ b/packages/core/test/session-pending-attachments.test.ts @@ -15,7 +15,6 @@ 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 { assertDefined, assertTag } from "./lib/assert" import { testEffect } from "./lib/effect" const it = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node, SessionProjector.node]))) @@ -78,16 +77,16 @@ describe("SessionPending attachment bounding", () => { .pipe(Effect.orDie) expect(rows).toHaveLength(1) const eventData = Schema.decodeUnknownSync(SessionEvent.InputAdmitted.data)(rows[0]!.data) - assertDefined(eventData.payloadHash) - assertTag(eventData.input, "user") + expect(eventData.payloadHash).toBeDefined() + expect(eventData.input.type).toBe("user") + if (eventData.input.type !== "user") throw new Error("Expected user input on admitted event") expect(eventData.input.data.files?.[0]?.data).toBe("") expect(JSON.stringify(eventData).includes(mega)).toBe(false) const pending = yield* SessionPending.list(db, sessionID) const row = pending.find((item) => item.id === inputID) - assertDefined(row) - assertTag(row, "user") - expect(row.data.files?.[0]?.data).toBe(mega) + expect(row?.type).toBe("user") + if (row?.type === "user") expect(row.data.files?.[0]?.data).toBe(mega) }), ) @@ -108,10 +107,12 @@ describe("SessionPending attachment bounding", () => { ], }, }) - assertTag(stripped, "user") - expect(stripped.data.text).toBe("hi") - expect(stripped.data.files?.[0]?.data).toBe("") - expect(stripped.data.files?.[0]?.name).toBe("a.txt") + expect(stripped.type).toBe("user") + if (stripped.type === "user") { + expect(stripped.data.text).toBe("hi") + expect(stripped.data.files?.[0]?.data).toBe("") + expect(stripped.data.files?.[0]?.name).toBe("a.txt") + } }), ) @@ -159,7 +160,8 @@ describe("SessionPending attachment bounding", () => { expect(rows).toHaveLength(1) const eventData = Schema.decodeUnknownSync(SessionEvent.InputAdmitted.data)(rows[0]!.data) expect(eventData.payloadHash).toBeUndefined() - assertTag(eventData.input, "user") + expect(eventData.input.type).toBe("user") + if (eventData.input.type !== "user") throw new Error("Expected user input on admitted event") expect(eventData.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 c47c98a4ab4f..c153b395f05a 100644 --- a/packages/core/test/session-runner-tool-events.test.ts +++ b/packages/core/test/session-runner-tool-events.test.ts @@ -11,7 +11,6 @@ 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" -import { assertDefined } from "./lib/assert" const sessionID = SessionV2.ID.make("ses_tool_event_test") const base64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAAB" @@ -79,13 +78,15 @@ test("local tool success stores media in the payload blob and omits base64 from await Effect.runPromise(publisher.publish(result)) const success = published.find((event) => event.type === "session.tool.success.1") - assertDefined(success) + expect(success).toBeDefined() + if (!success) throw new Error("Expected session.tool.success.1") const data = Schema.decodeUnknownSync(SessionEvent.Tool.Success.data)(success.data) - assertDefined(data.payloadHash) + expect(data.payloadHash).toBeDefined() expect(JSON.stringify(success).includes(base64)).toBe(false) expect(success.data).not.toHaveProperty("result") expect(data.content).toEqual([{ type: "text", text: "Image read successfully" }]) + if (!data.payloadHash) throw new Error("Expected payloadHash on thin success") const body = payloads.get(data.payloadHash) expect(body?.content).toEqual([ { type: "text", text: "Image read successfully" }, @@ -99,10 +100,12 @@ test("provider-executed success keeps raw provider result in the payload blob on 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") - assertDefined(success) + expect(success).toBeDefined() + if (!success) throw new Error("Expected session.tool.success.1") expect(success.data).not.toHaveProperty("result") const data = Schema.decodeUnknownSync(SessionEvent.Tool.Success.data)(success.data) - assertDefined(data.payloadHash) + expect(data.payloadHash).toBeDefined() + if (!data.payloadHash) throw new Error("Expected payloadHash on thin success") expect(payloads.get(data.payloadHash)?.result).toBeDefined() }) @@ -265,7 +268,7 @@ test("content-filter finish retains failure evidence until step closeout", async finish: "content-filter", tokens: { input: 8, output: 2, reasoning: 1 }, }) - assertDefined(settlement) + if (!settlement) throw new Error("Expected content-filter settlement") await Effect.runPromise( publisher.publishStepFailure({ cost: Money.USD.make(1.25), From 17ac510d5bed2da78ab68acefa6a9da649723c49 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 09:28:57 +1000 Subject: [PATCH 13/17] test(core): share asserted event type strings in payload tests Co-authored-by: Cursor --- .../test/session-pending-attachments.test.ts | 14 ++++++----- .../test/session-runner-tool-events.test.ts | 24 +++++++++++-------- 2 files changed, 22 insertions(+), 16 deletions(-) diff --git a/packages/core/test/session-pending-attachments.test.ts b/packages/core/test/session-pending-attachments.test.ts index 8e5ac0089812..056732147160 100644 --- a/packages/core/test/session-pending-attachments.test.ts +++ b/packages/core/test/session-pending-attachments.test.ts @@ -78,15 +78,16 @@ describe("SessionPending attachment bounding", () => { expect(rows).toHaveLength(1) const eventData = Schema.decodeUnknownSync(SessionEvent.InputAdmitted.data)(rows[0]!.data) expect(eventData.payloadHash).toBeDefined() - expect(eventData.input.type).toBe("user") - if (eventData.input.type !== "user") throw new Error("Expected user input on admitted event") + const inputType = "user" + expect(eventData.input.type).toBe(inputType) + if (eventData.input.type !== inputType) throw new Error(`Expected ${inputType} input on admitted event`) expect(eventData.input.data.files?.[0]?.data).toBe("") expect(JSON.stringify(eventData).includes(mega)).toBe(false) const pending = yield* SessionPending.list(db, sessionID) const row = pending.find((item) => item.id === inputID) - expect(row?.type).toBe("user") - if (row?.type === "user") expect(row.data.files?.[0]?.data).toBe(mega) + expect(row?.type).toBe(inputType) + if (row?.type === inputType) expect(row.data.files?.[0]?.data).toBe(mega) }), ) @@ -160,8 +161,9 @@ describe("SessionPending attachment bounding", () => { expect(rows).toHaveLength(1) const eventData = Schema.decodeUnknownSync(SessionEvent.InputAdmitted.data)(rows[0]!.data) expect(eventData.payloadHash).toBeUndefined() - expect(eventData.input.type).toBe("user") - if (eventData.input.type !== "user") throw new Error("Expected user input on admitted event") + const inputType = "user" + expect(eventData.input.type).toBe(inputType) + if (eventData.input.type !== inputType) throw new Error(`Expected ${inputType} input on admitted event`) expect(eventData.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 c153b395f05a..5e36eb4381fc 100644 --- a/packages/core/test/session-runner-tool-events.test.ts +++ b/packages/core/test/session-runner-tool-events.test.ts @@ -77,17 +77,19 @@ test("local tool success stores media in the payload blob and omits base64 from await Effect.runPromise(publisher.publish(call)) await Effect.runPromise(publisher.publish(result)) - const success = published.find((event) => event.type === "session.tool.success.1") + const successType = "session.tool.success.1" + const success = published.find((event) => event.type === successType) expect(success).toBeDefined() - if (!success) throw new Error("Expected session.tool.success.1") + if (!success) throw new Error(`Expected ${successType}`) const data = Schema.decodeUnknownSync(SessionEvent.Tool.Success.data)(success.data) - expect(data.payloadHash).toBeDefined() + const payloadHash = data.payloadHash + expect(payloadHash).toBeDefined() + if (!payloadHash) throw 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" }]) - if (!data.payloadHash) throw new Error("Expected payloadHash on thin success") - const body = payloads.get(data.payloadHash) + 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" }, @@ -99,14 +101,16 @@ test("provider-executed success keeps raw provider result in the payload blob on 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") + const successType = "session.tool.success.1" + const success = published.find((event) => event.type === successType) expect(success).toBeDefined() - if (!success) throw new Error("Expected session.tool.success.1") + if (!success) throw new Error(`Expected ${successType}`) expect(success.data).not.toHaveProperty("result") const data = Schema.decodeUnknownSync(SessionEvent.Tool.Success.data)(success.data) - expect(data.payloadHash).toBeDefined() - if (!data.payloadHash) throw new Error("Expected payloadHash on thin success") - expect(payloads.get(data.payloadHash)?.result).toBeDefined() + const payloadHash = data.payloadHash + expect(payloadHash).toBeDefined() + if (!payloadHash) throw new Error("Expected payloadHash on thin success") + expect(payloads.get(payloadHash)?.result).toBeDefined() }) test("provider metadata is flattened using the route key", async () => { From 45662abd53b5b815cdbd1a72fe94d290fcbff085 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 09:32:33 +1000 Subject: [PATCH 14/17] test(core): narrow payload fixtures with Option.getOrThrowWith Co-authored-by: Cursor --- .../test/session-pending-attachments.test.ts | 52 +++++++++++++------ .../test/session-runner-tool-events.test.ts | 36 +++++++------ 2 files changed, 57 insertions(+), 31 deletions(-) diff --git a/packages/core/test/session-pending-attachments.test.ts b/packages/core/test/session-pending-attachments.test.ts index 056732147160..c6de7219b714 100644 --- a/packages/core/test/session-pending-attachments.test.ts +++ b/packages/core/test/session-pending-attachments.test.ts @@ -1,6 +1,6 @@ import { describe, expect } from "bun:test" import { and, eq } from "drizzle-orm" -import { Effect, Schema } from "effect" +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" @@ -79,15 +79,27 @@ describe("SessionPending attachment bounding", () => { const eventData = Schema.decodeUnknownSync(SessionEvent.InputAdmitted.data)(rows[0]!.data) expect(eventData.payloadHash).toBeDefined() const inputType = "user" - expect(eventData.input.type).toBe(inputType) - if (eventData.input.type !== inputType) throw new Error(`Expected ${inputType} input on admitted event`) - expect(eventData.input.data.files?.[0]?.data).toBe("") + const input = Option.getOrThrowWith( + Option.liftPredicate(eventData.input, (value): value is Extract => + value.type === inputType, + ), + () => new Error(`Expected ${inputType} input on admitted event`), + ) + expect(input.data.files?.[0]?.data).toBe("") expect(JSON.stringify(eventData).includes(mega)).toBe(false) const pending = yield* SessionPending.list(db, sessionID) - const row = pending.find((item) => item.id === inputID) - expect(row?.type).toBe(inputType) - if (row?.type === inputType) expect(row.data.files?.[0]?.data).toBe(mega) + const pendingRow = Option.getOrThrowWith( + Option.fromUndefinedOr(pending.find((item) => item.id === inputID)), + () => new Error(`Expected pending input ${inputID}`), + ) + const row = Option.getOrThrowWith( + Option.liftPredicate(pendingRow, (value): value is Extract<(typeof pending)[number], { type: "user" }> => + value.type === inputType, + ), + () => new Error(`Expected ${inputType} pending input`), + ) + expect(row.data.files?.[0]?.data).toBe(mega) }), ) @@ -108,12 +120,16 @@ describe("SessionPending attachment bounding", () => { ], }, }) - expect(stripped.type).toBe("user") - if (stripped.type === "user") { - expect(stripped.data.text).toBe("hi") - expect(stripped.data.files?.[0]?.data).toBe("") - expect(stripped.data.files?.[0]?.name).toBe("a.txt") - } + const inputType = "user" + const input = Option.getOrThrowWith( + Option.liftPredicate(stripped, (value): value is Extract => + value.type === inputType, + ), + () => new Error(`Expected ${inputType} input`), + ) + expect(input.data.text).toBe("hi") + expect(input.data.files?.[0]?.data).toBe("") + expect(input.data.files?.[0]?.name).toBe("a.txt") }), ) @@ -162,9 +178,13 @@ describe("SessionPending attachment bounding", () => { const eventData = Schema.decodeUnknownSync(SessionEvent.InputAdmitted.data)(rows[0]!.data) expect(eventData.payloadHash).toBeUndefined() const inputType = "user" - expect(eventData.input.type).toBe(inputType) - if (eventData.input.type !== inputType) throw new Error(`Expected ${inputType} input on admitted event`) - expect(eventData.input.data.text).toBe("hello") + const input = Option.getOrThrowWith( + Option.liftPredicate(eventData.input, (value): value is Extract => + 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 5e36eb4381fc..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" @@ -78,13 +78,15 @@ test("local tool success stores media in the payload blob and omits base64 from await Effect.runPromise(publisher.publish(result)) const successType = "session.tool.success.1" - const success = published.find((event) => event.type === successType) - expect(success).toBeDefined() - if (!success) throw new Error(`Expected ${successType}`) + 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 = data.payloadHash - expect(payloadHash).toBeDefined() - if (!payloadHash) throw new Error("Expected payloadHash on thin success") + 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" }]) @@ -102,14 +104,16 @@ test("provider-executed success keeps raw provider result in the payload blob on await Effect.runPromise(publisher.publish(LLMEvent.toolCall({ ...call, providerExecuted: true }))) await Effect.runPromise(publisher.publish(LLMEvent.toolResult({ ...result, providerExecuted: true }))) const successType = "session.tool.success.1" - const success = published.find((event) => event.type === successType) - expect(success).toBeDefined() - if (!success) throw new Error(`Expected ${successType}`) + 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 = data.payloadHash - expect(payloadHash).toBeDefined() - if (!payloadHash) throw new Error("Expected payloadHash on thin success") + const payloadHash = Option.getOrThrowWith( + Option.fromUndefinedOr(data.payloadHash), + () => new Error("Expected payloadHash on thin success"), + ) expect(payloads.get(payloadHash)?.result).toBeDefined() }) @@ -267,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), From 1700efc4ffaaab230057e82e690a9e55cd948500 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 10:19:38 +1000 Subject: [PATCH 15/17] test(core): hoist UserInput aliases for Option liftPredicate narrows Co-authored-by: Cursor --- .../test/session-pending-attachments.test.ts | 53 ++++++++++--------- 1 file changed, 27 insertions(+), 26 deletions(-) diff --git a/packages/core/test/session-pending-attachments.test.ts b/packages/core/test/session-pending-attachments.test.ts index c6de7219b714..24b1341056be 100644 --- a/packages/core/test/session-pending-attachments.test.ts +++ b/packages/core/test/session-pending-attachments.test.ts @@ -18,6 +18,8 @@ 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", () => @@ -50,7 +52,7 @@ describe("SessionPending attachment bounding", () => { id: inputID, sessionID, input: { - type: "user", + type: inputType, delivery: "steer", data: { text: "see file", @@ -66,8 +68,12 @@ describe("SessionPending attachment bounding", () => { }, }) - expect(admitted.type).toBe("user") - if (admitted.type === "user") expect(admitted.data.files?.[0]?.data).toBe(mega) + 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() @@ -78,26 +84,24 @@ describe("SessionPending attachment bounding", () => { expect(rows).toHaveLength(1) const eventData = Schema.decodeUnknownSync(SessionEvent.InputAdmitted.data)(rows[0]!.data) expect(eventData.payloadHash).toBeDefined() - const inputType = "user" + + type UserInput = Extract const input = Option.getOrThrowWith( - Option.liftPredicate(eventData.input, (value): value is Extract => - value.type === inputType, - ), + 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("") + expect(input.data.files?.[0]?.data).toBe(strippedBytes) expect(JSON.stringify(eventData).includes(mega)).toBe(false) const pending = yield* SessionPending.list(db, sessionID) - const pendingRow = Option.getOrThrowWith( - Option.fromUndefinedOr(pending.find((item) => item.id === inputID)), - () => new Error(`Expected pending input ${inputID}`), - ) + type UserPending = Extract<(typeof pending)[number], { type: typeof inputType }> const row = Option.getOrThrowWith( - Option.liftPredicate(pendingRow, (value): value is Extract<(typeof pending)[number], { type: "user" }> => - value.type === inputType, + 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`), + () => new Error(`Expected ${inputType} pending input ${inputID}`), ) expect(row.data.files?.[0]?.data).toBe(mega) }), @@ -106,7 +110,7 @@ describe("SessionPending attachment bounding", () => { it.effect("stripAttachmentBytes clears only file data fields", () => Effect.sync(() => { const stripped = SessionPending.stripAttachmentBytes({ - type: "user", + type: inputType, delivery: "steer", data: { text: "hi", @@ -120,15 +124,13 @@ describe("SessionPending attachment bounding", () => { ], }, }) - const inputType = "user" + type UserInput = Extract const input = Option.getOrThrowWith( - Option.liftPredicate(stripped, (value): value is Extract => - value.type === inputType, - ), + 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("") + expect(input.data.files?.[0]?.data).toBe(strippedBytes) expect(input.data.files?.[0]?.name).toBe("a.txt") }), ) @@ -162,7 +164,7 @@ describe("SessionPending attachment bounding", () => { id: inputID, sessionID, input: { - type: "user", + type: inputType, delivery: "steer", data: { text: "hello" }, }, @@ -177,11 +179,10 @@ describe("SessionPending attachment bounding", () => { expect(rows).toHaveLength(1) const eventData = Schema.decodeUnknownSync(SessionEvent.InputAdmitted.data)(rows[0]!.data) expect(eventData.payloadHash).toBeUndefined() - const inputType = "user" + + type UserInput = Extract const input = Option.getOrThrowWith( - Option.liftPredicate(eventData.input, (value): value is Extract => - value.type === inputType, - ), + 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") From 135d2d8e332775cd39322e92e53abed4b518fdc8 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 10:29:37 +1000 Subject: [PATCH 16/17] refactor(tui): flatten admitted payloadHash pending refetch Co-authored-by: Cursor --- packages/tui/src/context/data.tsx | 33 +++++++++++++------------------ 1 file changed, 14 insertions(+), 19 deletions(-) diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index 3431490ac5f2..9fb2a0010005 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -426,34 +426,29 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ }, ) }) - if (event.data.payloadHash) + if (event.data.payloadHash) { + const sessionID = event.data.sessionID + const inputID = event.data.inputID void client.api.session.pending - .list({ sessionID: event.data.sessionID }) + .list({ sessionID }) .then((pending) => { - setStore("session", "pending", event.data.sessionID, reconcile(pending)) - const item = pending.find((entry) => entry.id === event.data.inputID) + setStore("session", "pending", sessionID, reconcile(pending)) + const item = pending.find((entry) => entry.id === inputID) if (!item || item.type === "compaction") return - message.update(event.data.sessionID, (draft, index) => { + 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) - const next = - item.type === "user" - ? { - id: item.id, - type: "user" as const, - ...item.data, - time: { created: item.timeCreated }, - } - : { - id: item.id, - type: "synthetic" as const, - ...item.data, - time: { created: item.timeCreated }, - } 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 From b447bde2c7a78956f32d76cf7d981ed129a2b0d2 Mon Sep 17 00:00:00 2001 From: Arman Charan Date: Sat, 18 Jul 2026 10:40:52 +1000 Subject: [PATCH 17/17] style(core): wrap ToolPayload.hash for prettier Co-authored-by: Cursor --- packages/core/src/session/tool-payload.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/packages/core/src/session/tool-payload.ts b/packages/core/src/session/tool-payload.ts index 7f036460ad30..a293763ed26f 100644 --- a/packages/core/src/session/tool-payload.ts +++ b/packages/core/src/session/tool-payload.ts @@ -39,7 +39,11 @@ 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")) + return Hash.make( + createHash("sha256") + .update(canonical(decodeJson(encodeBody(value)))) + .digest("hex"), + ) } export const insertJson = Effect.fn("ToolPayload.insertJson")(function* (