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

fix: make tool progress live-only (#38217)

Browse source
This commit is contained in:
Kit Langton 2026-07-22 10:40:07 -04:00 • committed by GitHub
commit 5a9ed4d350
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
24 changed files with 337 additions and 89 deletions
Download patch file Download diff file Expand all files Collapse all files

4
packages/cli/src/acp/event.ts
View file

@ -198,8 +198,8 @@ export async function streamTurn(input: {
toolCallId: event.data.callID,
toolName: current.name,
input: current.input,
structured: current.structured,
content: current.content,
structured: event.data.metadata ?? current.structured,
content: event.data.content ?? current.content,
error: event.data.error.message,
cwd: input.cwd,
}),

12
packages/cli/src/run/noninteractive.ts
View file

@ -398,6 +398,8 @@ export async function runNonInteractivePrompt(input: Input) {
const key = toolKey(event.data.assistantMessageID, event.data.callID)
const current = tools.get(key) ?? fallbackTool(event)
const error = event.data.error.message
const structured = event.data.metadata ?? current.structured
const content = event.data.content ?? current.content
const tool: SessionMessageAssistantTool = {
type: "tool",
id: event.data.callID,
@ -408,8 +410,8 @@ export async function runNonInteractivePrompt(input: Input) {
state: {
status: "error",
input: current.input,
structured: current.structured,
content: current.content,
structured,
content,
error: event.data.error,
result: event.data.result,
},
@ -439,14 +441,14 @@ export async function runNonInteractivePrompt(input: Input) {
renderedTools.add(key)
if (input.compatibility === "v1" && (permissionRejected || formCancelled)) continue
if (!emit("tool_use", time, { part })) {
if (toolOutputText(current.tool, current.content).trim())
if (toolOutputText(current.tool, content).trim())
await input.renderTool({
...tool,
state: {
status: "completed",
input: current.input,
structured: current.structured,
content: current.content,
structured,
content,
result: event.data.result,
},
})

6
packages/cli/test/acp/event-behavior.test.ts
View file

@ -214,7 +214,7 @@ describe("acp event behavior", () => {
}),
)
send(
durableEvent("session.tool.progress", {
ephemeralEvent("session.tool.progress", {
sessionID: "ses_tools",
assistantMessageID: "msg_tools",
callID: "call_ok",
@ -251,7 +251,7 @@ describe("acp event behavior", () => {
}),
)
send(
durableEvent("session.tool.progress", {
ephemeralEvent("session.tool.progress", {
sessionID: "ses_tools",
assistantMessageID: "msg_tools",
callID: "call_fail",
@ -265,6 +265,8 @@ describe("acp event behavior", () => {
assistantMessageID: "msg_tools",
callID: "call_fail",
error: { type: "tool.error", message: "not found" },
metadata: { bytes: 0 },
content: [{ type: "text", text: "opening" }],
executed: true,
}),
)

7
packages/cli/test/run/noninteractive.test.ts
View file

@ -130,7 +130,6 @@ function failedTool(inputID: string): V2Event[] {
id: "evt_failed_tool_progress",
created: 3,
type: "session.tool.progress",
durable: { aggregateID: "ses_1", seq: 3, version: 1 },
data: {
sessionID: "ses_1",
assistantMessageID: "msg_failed_tool",
@ -149,6 +148,8 @@ function failedTool(inputID: string): V2Event[] {
assistantMessageID: "msg_failed_tool",
callID: "call_failed_tool",
error: { type: "unknown", message: "tool failed" },
metadata: { checkpoint: 1 },
content: [{ type: "text", text: "partial output" }],
executed: true,
},
},
@ -440,11 +441,11 @@ describe("runNonInteractivePrompt", () => {
expect(output).toEqual({ stdout: "", stderr: "", exitCode: 0 })
})
test("renders native failed tool output before the terminal error", async () => {
test("renders a native terminal failure snapshot when live progress was missed", async () => {
const rendered: SessionMessageAssistantTool[] = []
const failed: SessionMessageAssistantTool[] = []
await capture({
turn: failedTool,
turn: (inputID) => failedTool(inputID).filter((event) => event.type !== "session.tool.progress"),
renderTool: (part) => {
rendered.push(part)
return Promise.resolve()

70
packages/client/src/promise/generated/types.ts
View file

@ -1387,24 +1387,6 @@ export type SessionToolCalled = {
}
}
export type SessionToolFailed = {
id: string
created: number
metadata?: { [x: string]: any }
type: "session.tool.failed"
durable: { aggregateID: string; seq: number; version: 1 }
location?: LocationRef
data: {
sessionID: string
assistantMessageID: string
callID: string
error: SessionStructuredError
result?: any
executed: boolean
resultState?: SessionMessageProviderState7
}
}
export type ModelCompatibility = { reasoningField?: ModelReasoningField }
export type ModelCost = {
@ -1851,22 +1833,6 @@ export type SessionMessageToolStateError = {
result?: JsonValue
}
export type SessionToolProgress = {
id: string
created: number
metadata?: { [x: string]: any }
type: "session.tool.progress"
durable: { aggregateID: string; seq: number; version: 1 }
location?: LocationRef
data: {
sessionID: string
assistantMessageID: string
callID: string
structured: { [x: string]: any }
content: Array<LLMToolContent>
}
}
export type SessionToolSuccess = {
id: string
created: number
@ -1886,6 +1852,41 @@ export type SessionToolSuccess = {
}
}
export type SessionToolFailed = {
id: string
created: number
metadata?: { [x: string]: any }
type: "session.tool.failed"
durable: { aggregateID: string; seq: number; version: 1 }
location?: LocationRef
data: {
sessionID: string
assistantMessageID: string
callID: string
error: SessionStructuredError
content?: [LLMToolContent, ...Array<LLMToolContent>]
metadata?: { [x: string]: any }
result?: any
executed: boolean
resultState?: SessionMessageProviderState7
}
}
export type SessionToolProgress = {
id: string
created: number
metadata?: { [x: string]: any }
type: "session.tool.progress"
location?: LocationRef
data: {
sessionID: string
assistantMessageID: string
callID: string
structured: { [x: string]: any }
content: Array<LLMToolContent>
}
}
export type SessionMessageCompaction =
| SessionMessageCompactionRunning
| SessionMessageCompactionCompleted
@ -2224,7 +2225,6 @@ export type SessionEventDurable =
| SessionToolInputStarted
| SessionToolInputEnded
| SessionToolCalled
| SessionToolProgress
| SessionToolSuccess
| SessionToolFailed
| SessionRetryScheduled

1
packages/core/src/database/migration.gen.ts generated
View file

@ -55,5 +55,6 @@ export const migrations = (
import("./migration/20260709190621_session_pending_table"),
import("./migration/20260710025429_instruction_sync"),
import("./migration/20260716020354_kv"),
import("./migration/20260722011141_delete_tool_progress_events"),
])
).map((module) => module.default) satisfies DatabaseMigration.Migration[]

11
packages/core/src/database/migration/20260722011141_delete_tool_progress_events.ts Normal file
View file

@ -0,0 +1,11 @@
import { Effect } from "effect"
import type { DatabaseMigration } from "../migration"
export default {
id: "20260722011141_delete_tool_progress_events",
up(tx) {
return Effect.gen(function* () {
yield* tx.run(`DELETE FROM \`event\` WHERE \`type\` = 'session.tool.progress.1';`)
})
},
} satisfies DatabaseMigration.Migration

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

@ -402,8 +402,8 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
status: "error",
error: event.data.error,
input: typeof match.state.input === "string" ? {} : match.state.input,
structured: match.state.status === "running" ? match.state.structured : {},
content: match.state.status === "running" ? match.state.content : [],
structured: event.data.metadata ?? (match.state.status === "running" ? match.state.structured : {}),
content: event.data.content ?? (match.state.status === "running" ? match.state.content : []),
result: event.data.result,
}),
)

1
packages/core/src/session/projector.ts
View file

@ -697,7 +697,6 @@ 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.Failed, (event) => run(db, event))
yield* events.project(SessionEvent.Reasoning.Started, (event) => run(db, event))

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

@ -163,16 +163,7 @@ const layer = Layer.effect(
agent: agent.id,
messageID: assistantMessageID,
call: event,
progress: (update) =>
serialized(
events.publish(SessionEvent.Tool.Progress, {
sessionID: session.id,
assistantMessageID,
callID: event.id,
structured: { ...update.structured },
content: [...update.content],
}),
),
progress: (update) => serialized(publisher.progress(event.id, update)),
}),
).pipe(
Effect.flatMap((settlement) =>

29
packages/core/src/session/runner/publish-llm-event.ts
View file

@ -11,6 +11,7 @@ import { AgentV2 } from "../../agent"
import { Snapshot } from "../../snapshot"
import { RelativePath } from "../../schema"
import { SessionUsage } from "../usage"
import type { ToolRegistry } from "../../tool/registry"
type Input = {
readonly sessionID: SessionSchema.ID
@ -54,8 +55,17 @@ export const createLLMEventPublisher = (events: Pick<EventV2.Interface, "publish
called: boolean
settled: boolean
providerExecuted: boolean
progress?: ToolRegistry.Progress
}
>()
const failureSnapshot = (tool: { readonly progress?: ToolRegistry.Progress }) => {
if (!tool.progress) return {}
const first = tool.progress.content[0]
return {
...(first === undefined ? {} : { content: [first, ...tool.progress.content.slice(1)] as const }),
metadata: tool.progress.structured,
}
}
let assistantMessageID = input.assistantMessageID
let stepStarted = false
let stepFailed = false
@ -232,6 +242,7 @@ export const createLLMEventPublisher = (events: Pick<EventV2.Interface, "publish
type: "tool.input-json",
message: "Tool call arguments were malformed JSON and were not executed. Retry with valid JSON.",
},
...failureSnapshot(tool),
executed: false,
})
})
@ -256,6 +267,7 @@ export const createLLMEventPublisher = (events: Pick<EventV2.Interface, "publish
assistantMessageID: tool.assistantMessageID,
callID,
error,
...failureSnapshot(tool),
executed: tool.providerExecuted,
})
}
@ -415,6 +427,7 @@ export const createLLMEventPublisher = (events: Pick<EventV2.Interface, "publish
assistantMessageID: tool.assistantMessageID,
callID: event.id,
error: result.error,
...failureSnapshot(tool),
result: event.result,
executed,
resultState,
@ -447,6 +460,7 @@ export const createLLMEventPublisher = (events: Pick<EventV2.Interface, "publish
event.message === `Unknown tool: ${event.name}`
? { type: "tool.unknown", message: event.message }
: { type: "tool.execution", message: event.message },
...failureSnapshot(tool),
executed: tool.providerExecuted,
resultState: providerState(event.providerMetadata),
})
@ -471,8 +485,23 @@ export const createLLMEventPublisher = (events: Pick<EventV2.Interface, "publish
}
})
const progress = Effect.fnUntraced(function* (callID: string, update: ToolRegistry.Progress) {
const tool = tools.get(callID)
if (!tool?.called || tool.settled)
return yield* Effect.die(new Error(`Tool progress outside running call: ${callID}`))
const current = { structured: { ...update.structured }, content: [...update.content] }
tool.progress = current
yield* events.publish(SessionEvent.Tool.Progress, {
sessionID: input.sessionID,
assistantMessageID: tool.assistantMessageID,
callID,
...current,
})
})
return {
publish,
progress,
flush,
failAssistant,
publishStepFailure,

15
packages/core/src/tool/shell.ts
View file

@ -256,13 +256,22 @@ export const Plugin = {
}
}
let previousProgress: { readonly output: string; readonly truncated: boolean } | undefined
const progress = yield* Effect.sleep("1 second").pipe(
Effect.andThen(
captureShell().pipe(
Effect.flatMap((capture) =>
context.progress({
structured: { truncated: capture.truncated },
content: [{ type: "text", text: capture.output }],
Effect.gen(function* () {
if (
previousProgress?.output === capture.output &&
previousProgress.truncated === capture.truncated
)
return
previousProgress = capture
yield* context.progress({
structured: { truncated: capture.truncated },
content: [{ type: "text", text: capture.output }],
})
}),
),
),

26
packages/core/test/database-migration.test.ts
View file

@ -24,6 +24,7 @@ import renameInstructionsMigration from "@opencode-ai/core/database/migration/20
import addSessionForkMigration from "@opencode-ai/core/database/migration/20260706223930_add-session-fork"
import timeSuspendedMigration from "@opencode-ai/core/database/migration/20260709163752_time_suspended"
import instructionSyncMigration from "@opencode-ai/core/database/migration/20260710025429_instruction_sync"
import deleteToolProgressEventsMigration from "@opencode-ai/core/database/migration/20260722011141_delete_tool_progress_events"
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
import { EventV2 } from "@opencode-ai/core/event"
@ -557,6 +558,31 @@ describe("DatabaseMigration", () => {
)
})
test("deletes durable tool progress without changing aggregate sequence watermarks", async () => {
await run(
Effect.gen(function* () {
const db = yield* makeDb
yield* db.run(sql`CREATE TABLE event_sequence (aggregate_id text PRIMARY KEY, seq integer NOT NULL)`)
yield* db.run(
sql`CREATE TABLE event (id text PRIMARY KEY, aggregate_id text NOT NULL, seq integer NOT NULL, type text NOT NULL, data text NOT NULL)`,
)
yield* db.run(sql`INSERT INTO event_sequence VALUES ('ses_test', 5)`)
yield* db.run(sql`INSERT INTO event VALUES ('evt_success', 'ses_test', 4, 'session.tool.success.1', '{}')`)
yield* db.run(sql`INSERT INTO event VALUES ('evt_progress', 'ses_test', 5, 'session.tool.progress.1', '{}')`)
yield* DatabaseMigration.applyOnly(db, [deleteToolProgressEventsMigration])
expect(yield* db.all(sql`SELECT id, seq, type, data FROM event ORDER BY seq`)).toEqual([
{ id: "evt_success", seq: 4, type: "session.tool.success.1", data: "{}" },
])
expect(yield* db.get(sql`SELECT aggregate_id, seq FROM event_sequence`)).toEqual({
aggregate_id: "ses_test",
seq: 5,
})
}),
)
})
test("records the authoritative parent sequence on existing forks", async () => {
await run(
Effect.gen(function* () {

42
packages/core/test/session-runner-tool-events.test.ts
View file

@ -1,5 +1,5 @@
import { expect, test } from "bun:test"
import { Effect, Schema } from "effect"
import { Cause, Effect, Exit, Schema } from "effect"
import { LLMEvent } from "@opencode-ai/ai"
import { Money } from "@opencode-ai/schema/money"
import { EventV2 } from "@opencode-ai/core/event"
@ -16,11 +16,11 @@ import { createLLMEventPublisher } from "@opencode-ai/core/session/runner/publis
const sessionID = SessionV2.ID.make("ses_tool_event_test")
const base64 = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAAB"
const capture = (providerMetadataKey = "anthropic") => {
const capture = (providerMetadataKey = "anthropic", options?: { readonly interruptProgress?: boolean }) => {
const published: Array<{ readonly type: string; readonly data: unknown }> = []
const events: Pick<EventV2.Interface, "publish"> = {
publish: (definition, data) =>
Effect.sync(() => {
publish: (definition, data) => {
const publish = Effect.sync(() => {
const event = { id: EventV2.ID.create(), type: definition.type, data } as EventV2.Payload<typeof definition>
published.push({
type: definition.durable
@ -29,7 +29,11 @@ const capture = (providerMetadataKey = "anthropic") => {
data,
})
return event
}),
})
return definition.type === SessionEvent.Tool.Progress.type && options?.interruptProgress
? publish.pipe(Effect.andThen(Effect.interrupt))
: publish
},
}
return {
published,
@ -92,6 +96,34 @@ test("provider-executed success retains its raw provider result", async () => {
expect(success?.data).toHaveProperty("result")
})
test("interrupted progress publication remains in the terminal failure snapshot", async () => {
const { published, publisher } = capture("anthropic", { interruptProgress: true })
await Effect.runPromise(publisher.publish(call))
const exit = await Effect.runPromiseExit(
publisher.progress(call.id, {
structured: { phase: "visible" },
content: [{ type: "text", text: "visible" }],
}),
)
expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBe(true)
await Effect.runPromise(publisher.failUnsettledTools({ type: "aborted", message: "interrupted" }))
expect(published.find((event) => event.type === "session.tool.failed.1")?.data).toMatchObject({
metadata: { phase: "visible" },
content: [{ type: "text", text: "visible" }],
})
})
test("failure before progress omits partial output fields", async () => {
const { published, publisher } = capture()
await Effect.runPromise(publisher.publish(call))
await Effect.runPromise(publisher.failUnsettledTools({ type: "aborted", message: "interrupted" }))
const failed = published.find((event) => event.type === "session.tool.failed.1")?.data
expect(failed).not.toHaveProperty("content")
expect(failed).not.toHaveProperty("metadata")
})
test("provider metadata is flattened using the route key", async () => {
const { published, publisher } = capture()
await Effect.runPromise(

46
packages/core/test/session-runner.test.ts
View file

@ -892,6 +892,52 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("persists the latest partial snapshot when a tool fails", () =>
Effect.gen(function* () {
const session = yield* setup
const registry = yield* ToolRegistry.Service
yield* registry.register({
failing_progress: Tool.make({
description: "Report progress and fail",
input: Schema.Struct({}),
output: Schema.Struct({}),
execute: (_, context) =>
Effect.gen(function* () {
yield* context.progress({
structured: { phase: "running" },
content: [{ type: "text", text: "before failure" }],
})
return yield* new ToolFailure({ message: "failed after progress" })
}),
}),
}, { codemode: false })
yield* admit(session, "Run failing progress")
responses = [reply.tool("call-failing-progress", "failing_progress", {}), reply.stop()]
yield* session.resume(sessionID)
expect(yield* session.context(sessionID)).toMatchObject([
{ type: "user", text: "Run failing progress" },
{
type: "assistant",
content: [
{
type: "tool",
id: "call-failing-progress",
state: {
status: "error",
structured: { phase: "running" },
content: [{ type: "text", text: "before failure" }],
error: { message: "failed after progress" },
},
},
],
},
{ type: "assistant", finish: "stop" },
])
}),
)
it.effect("executes the tool advertised before a registry reload", () =>
Effect.gen(function* () {
const session = yield* setup

13
packages/core/test/session-tool-progress.test.ts
View file

@ -22,10 +22,10 @@ const it = testEffect(LayerNode.compile(LayerNode.group([Database.node, EventV2.
const timestamp = DateTime.makeUnsafe(1)
const model = { id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") }
const content = (text: string) => [{ type: "text" as const, text }]
const content = (text: string) => [{ type: "text" as const, text }] as const
describe("Tool.Progress", () => {
it.effect("projects durable progress and keeps final settlements durable", () =>
it.effect("keeps progress live-only and terminal settlements durable", () =>
Effect.gen(function* () {
const { db } = yield* Database.Service
const service = yield* EventV2.Service
@ -87,7 +87,7 @@ describe("Tool.Progress", () => {
state: { status: "running", structured: {}, content: [] },
})
yield* service.publish(SessionEvent.Tool.Progress, {
const progress = yield* service.publish(SessionEvent.Tool.Progress, {
sessionID,
assistantMessageID,
callID: "call-success",
@ -95,7 +95,7 @@ describe("Tool.Progress", () => {
content: content("saved"),
})
expect((yield* readAssistant).content[0]).toMatchObject({
state: { status: "running", structured: { phase: "checkpoint" }, content: content("saved") },
state: { status: "running", structured: {}, content: [] },
})
const success = yield* service.publish(SessionEvent.Tool.Success, {
@ -123,6 +123,8 @@ describe("Tool.Progress", () => {
assistantMessageID,
callID: "call-failed",
error: { type: "unknown", message: "boom" },
metadata: { phase: "checkpoint" },
content: content("before failure"),
executed: false,
})
expect((yield* readAssistant).content[1]).toMatchObject({
@ -133,6 +135,7 @@ describe("Tool.Progress", () => {
error: { type: "unknown", message: "boom" },
},
})
expect(Schema.is(SessionEvent.Durable)(progress)).toBe(false)
expect(Schema.is(SessionEvent.Durable)(success)).toBe(true)
expect(Schema.is(SessionEvent.Durable)(failed)).toBe(true)
@ -143,7 +146,7 @@ describe("Tool.Progress", () => {
.orderBy(asc(EventTable.seq))
.all()
.pipe(Effect.orDie)
expect(rows.map((row) => row.type)).toContain(EventV2.versionedType(SessionEvent.Tool.Progress.type, 1))
expect(rows.map((row) => row.type)).not.toContain(EventV2.versionedType(SessionEvent.Tool.Progress.type, 1))
expect(rows.map((row) => row.type)).toContain(EventV2.versionedType(SessionEvent.Tool.Success.type, 1))
expect(rows.map((row) => row.type)).toContain(EventV2.versionedType(SessionEvent.Tool.Failed.type, 1))
}),

31
packages/core/test/tool-shell.test.ts
View file

@ -159,6 +159,9 @@ const mixedOutputCommand = isWindows
? "[Console]::Out.Write('stdout'); Start-Sleep -Milliseconds 50; [Console]::Error.Write('stderr'); Start-Sleep -Milliseconds 100"
: "printf stdout; sleep 0.05; printf stderr >&2"
const idleCommand = isWindows ? "Start-Sleep -Seconds 60" : "sleep 60"
const steadyProgressCommand = isWindows
? "[Console]::Out.Write('steady'); Start-Sleep -Milliseconds 3400"
: "printf steady; sleep 3.4"
const bodyExitCommand = isWindows
? "[Console]::Out.Write('body'); Start-Sleep -Milliseconds 100; exit 7"
: "printf body && exit 7"
@ -462,6 +465,34 @@ describe("ShellTool", () => {
{ timeout: 15_000 },
)
it.live(
"does not repeat unchanged shell progress",
() =>
Effect.acquireUseRelease(
Effect.promise(() => tmpdir()),
(tmp) => {
reset()
return withSession(tmp.path, (registry) =>
Effect.gen(function* () {
const updates: ToolRegistry.Progress[] = []
yield* settleTool(registry, {
...call({ command: steadyProgressCommand }, "call-steady-progress"),
progress: (update) => Effect.sync(() => updates.push(update)),
})
expect(updates).toEqual([
{
structured: { truncated: false },
content: [{ type: "text", text: "steady" }],
},
])
}),
)
},
(tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]().then(() => undefined)),
),
{ timeout: 10_000 },
)
it.live("returns a useful timeout settlement", () =>
Effect.acquireUseRelease(
Effect.promise(() => tmpdir()),

13
packages/schema/src/session-event.ts
View file

@ -3,8 +3,7 @@ export * as SessionEvent from "./session-event.js"
import { Schema } from "effect"
import { optional } from "./schema.js"
import { Event } from "./event.js"
import { ToolContent } from "./llm.js"
import { FinishReason } from "./llm.js"
import { FinishReason, ToolContent } from "./llm.js"
import { Model } from "./model.js"
import { NonNegativeInt, PositiveInt, RelativePath } from "./schema.js"
import { FileAttachment } from "./prompt.js"
@ -410,13 +409,9 @@ export namespace Tool {
})
export type Called = typeof Called.Type
/**
* Replayable bounded running-tool state. Tools should checkpoint semantic
* transitions or at a bounded cadence, not persist every stdout/stderr chunk.
*/
export const Progress = Event.durable({
/** Live replacement snapshot for a running tool. */
export const Progress = Event.ephemeral({
type: "session.tool.progress",
...options,
schema: {
...ToolBase,
structured: Schema.Record(Schema.String, Schema.Unknown),
@ -445,6 +440,8 @@ export namespace Tool {
schema: {
...ToolBase,
error: SessionError.Error,
content: Schema.NonEmptyArray(ToolContent).pipe(optional),
metadata: Schema.Record(Schema.String, Schema.Unknown).pipe(optional),
result: Schema.Unknown.pipe(optional),
executed: Schema.Boolean,
resultState: SessionMessage.ProviderState.pipe(optional),

3
packages/schema/test/event-manifest.test.ts
View file

@ -125,7 +125,6 @@ describe("public event manifest", () => {
"session.tool.input.started.1",
"session.tool.input.ended.1",
"session.tool.called.1",
"session.tool.progress.1",
"session.tool.success.1",
"session.tool.failed.1",
"session.reasoning.started.1",
@ -152,6 +151,8 @@ describe("public event manifest", () => {
expect(EventManifest.Latest.has("session.usage.recorded")).toBe(false)
expect(SessionEvent.UsageUpdated.durability).toBe("ephemeral")
expect(SessionEvent.Compaction.Delta.durability).toBe("ephemeral")
expect(SessionEvent.Tool.Progress.durability).toBe("ephemeral")
expect(EventManifest.Server.get("session.tool.progress")).toBe(SessionEvent.Tool.Progress)
expect(EventManifest.Durable.has("session.compaction.delta.1")).toBe(false)
expect(EventManifest.ServerDefinitions).toContain(SessionEvent.UsageUpdated)
expect(EventManifest.Definitions.every((definition) => definition.durability !== undefined)).toBe(true)

4
packages/tui/src/context/data.tsx
View file

@ -643,8 +643,8 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({
status: "error",
error: event.data.error,
input: typeof match.state.input === "string" ? {} : match.state.input,
structured: match.state.status === "running" ? match.state.structured : {},
content: match.state.status === "running" ? match.state.content : [],
structured: event.data.metadata ?? (match.state.status === "running" ? match.state.structured : {}),
content: event.data.content ?? (match.state.status === "running" ? match.state.content : []),
result: event.data.result,
}
match.executed = event.data.executed || match.executed === true

18
packages/tui/src/mini/stream-v2.subagent.ts
View file

@ -837,8 +837,9 @@ export function createSubagentTracker(input: SubagentTrackerInput): SubagentTrac
? {
status: "error",
input: part && part.state.status !== "streaming" ? part.state.input : {},
structured: part && part.state.status !== "streaming" ? part.state.structured : {},
content: part && part.state.status !== "streaming" ? part.state.content : [],
structured:
event.data.metadata ?? (part && part.state.status !== "streaming" ? part.state.structured : {}),
content: event.data.content ?? (part && part.state.status !== "streaming" ? part.state.content : []),
error: event.data.error,
result: event.data.result,
}
@ -957,15 +958,16 @@ export function createSubagentTracker(input: SubagentTrackerInput): SubagentTrac
if (pendingCalls.has(key)) pendingCalls.set(key, event.data.input)
return
}
if (event.type === "session.tool.failed") {
pendingCalls.delete(sourceKey(event.data.assistantMessageID, event.data.callID))
if (
event.type !== "session.tool.progress" &&
event.type !== "session.tool.success" &&
event.type !== "session.tool.failed"
)
return
}
if (event.type !== "session.tool.progress" && event.type !== "session.tool.success") return
const key = sourceKey(event.data.assistantMessageID, event.data.callID)
const pending = pendingCalls.get(key)
if (event.type === "session.tool.success") pendingCalls.delete(key)
const found = childSessionID(record(event.data.structured))
if (event.type !== "session.tool.progress") pendingCalls.delete(key)
const found = childSessionID(record(event.type === "session.tool.failed" ? event.data.metadata : event.data.structured))
if (!found) return
const child = admitChild(found.sessionID)
if (!child) return

5
packages/tui/src/mini/stream-v2.transport.ts
View file

@ -1084,8 +1084,9 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
? {
status: "error",
input: part && part.state.status !== "streaming" ? part.state.input : {},
structured: part && part.state.status !== "streaming" ? part.state.structured : {},
content: part && part.state.status !== "streaming" ? part.state.content : [],
structured:
event.data.metadata ?? (part && part.state.status !== "streaming" ? part.state.structured : {}),
content: event.data.content ?? (part && part.state.status !== "streaming" ? part.state.content : []),
error: event.data.error,
result: event.data.result,
}

1
packages/tui/test/cli/tui/data.test.tsx
View file

@ -2332,7 +2332,6 @@ test("settles pending tools when a live failure arrives", async () => {
id: "evt_progress_1",
created: 0,
type: "session.tool.progress",
durable: durable("session-1", 5),
data: {
sessionID: "session-1",
assistantMessageID: "msg_explicit_assistant_9",

71
packages/tui/test/mini/stream-v2.transport.test.ts
View file

@ -2044,7 +2044,6 @@ describe("V2 mini transport", () => {
id: "evt_progress",
created: 3,
type: "session.tool.progress",
durable: durable("ses_1", 2),
data: {
sessionID: "ses_1",
assistantMessageID: "msg_progress",
@ -2063,6 +2062,8 @@ describe("V2 mini transport", () => {
assistantMessageID: "msg_progress",
callID: "call_progress",
error: { type: "unknown", message: "boom" },
metadata: { checkpoint: 1 },
content: [{ type: "text", text: "partial" }],
executed: true,
},
})
@ -2844,6 +2845,70 @@ describe("V2 mini transport", () => {
await transport.close()
})
test("discovers a subagent from its terminal failure snapshot", async () => {
const events = feed()
events.push(connected())
const client = sdk({ streams: [events], messages: { ses_child_failed: [] } })
const ui = footer()
const transport = await createSessionTransport({
sdk: client,
sessionID: "ses_1",
thinking: false,
footer: ui.api,
})
const states = () => ui.events.flatMap((event) => (event.type === "stream.subagent" ? [event.state] : []))
events.push({
id: "evt_failed_subagent_input",
created: 1,
type: "session.tool.input.started",
durable: durable("ses_1"),
data: {
sessionID: "ses_1",
assistantMessageID: "msg_failed_subagent",
callID: "call_failed_subagent",
name: "subagent",
},
})
events.push({
id: "evt_failed_subagent_called",
created: 2,
type: "session.tool.called",
durable: durable("ses_1", 1),
data: {
sessionID: "ses_1",
assistantMessageID: "msg_failed_subagent",
callID: "call_failed_subagent",
input: { agent: "explore", description: "Inspect failure", prompt: "inspect" },
executed: true,
},
})
events.push({
id: "evt_failed_subagent",
created: 3,
type: "session.tool.failed",
durable: durable("ses_1", 2),
data: {
sessionID: "ses_1",
assistantMessageID: "msg_failed_subagent",
callID: "call_failed_subagent",
error: { type: "unknown", message: "subagent failed" },
metadata: { sessionID: "ses_child_failed", status: "running" },
executed: true,
},
})
while (!states().some((state) => state.tabs.some((tab) => tab.sessionID === "ses_child_failed")))
await Bun.sleep(0)
expect(states().at(-1)?.tabs).toMatchObject([
{
sessionID: "ses_child_failed",
label: "Explore",
description: "Inspect failure",
},
])
await transport.close()
})
test("discovers current subagents from progress and reduces descendant tool state", async () => {
const events = feed()
events.push(connected())
@ -2885,7 +2950,6 @@ describe("V2 mini transport", () => {
id: "evt_subagent_progress",
created: 3,
type: "session.tool.progress",
durable: durable("ses_1", 2),
data: {
sessionID: "ses_1",
assistantMessageID: "msg_subagent",
@ -2937,7 +3001,6 @@ describe("V2 mini transport", () => {
id: "evt_child_tool_progress",
created: 6,
type: "session.tool.progress",
durable: durable("ses_child_progress", 2),
data: {
sessionID: "ses_child_progress",
assistantMessageID: "msg_child_tool",
@ -2968,6 +3031,8 @@ describe("V2 mini transport", () => {
assistantMessageID: "msg_child_tool",
callID: "call_child_shell",
error: { type: "unknown", message: "child boom" },
metadata: { checkpoint: "child" },
content: [{ type: "text", text: "child partial" }],
executed: true,
},
})