diff --git a/src/main/claude/claude-background-task-frames.ts b/src/main/claude/claude-background-task-frames.ts index a3c38d1a9db..7497b8ad149 100644 --- a/src/main/claude/claude-background-task-frames.ts +++ b/src/main/claude/claude-background-task-frames.ts @@ -29,6 +29,10 @@ export function taskId(message: Record): string | null { return typeof value === 'string' && isBoundedClaudeTaskId(value) ? value : null } +export function taskAliasId(value: unknown): string | undefined { + return typeof value === 'string' && isBoundedClaudeTaskId(value) ? value : undefined +} + function boundedTaskText(value: unknown): string | undefined { if (typeof value !== 'string') { return undefined diff --git a/src/main/claude/claude-background-task-row-journal.ts b/src/main/claude/claude-background-task-row-journal.ts new file mode 100644 index 00000000000..6a56a546c68 --- /dev/null +++ b/src/main/claude/claude-background-task-row-journal.ts @@ -0,0 +1,39 @@ +import type { + AgentJournalItemBody, + AgentJournalItemIdentity +} from '../../shared/agent-session-journal-types' +import { backgroundTaskFallbackText } from '../../shared/native-chat-background-task-row' +import type { NativeChatBackgroundTaskBlock } from '../../shared/native-chat-types' +import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import type { ClaudeBackgroundTaskRow } from './claude-background-task-row-lifecycle' + +export function claudeBackgroundTaskIdentity(taskId: string): AgentJournalItemIdentity { + return { provider: 'orca', clientMessageId: `claude-background-task:${taskId}` } +} + +export function claudeBackgroundTaskBody( + block: NativeChatBackgroundTaskBlock +): AgentJournalItemBody { + return { + kind: 'message', + role: 'system', + blocks: [{ type: 'text', text: backgroundTaskFallbackText(block) }, { ...block }] + } +} + +export function writeClaudeBackgroundTaskRow( + sink: StructuredAgentSessionEventSink, + id: string, + row: ClaudeBackgroundTaskRow +): void { + const body = claudeBackgroundTaskBody(row.block) + const serialized = JSON.stringify(body) + if (serialized === row.lastSerialized) { + return + } + row.lastSerialized = serialized + sink.appendItem(claudeBackgroundTaskIdentity(id), body, { + coalescingKey: `claude-background-task:${id}` + }) + sink.publish() +} diff --git a/src/main/claude/claude-background-task-row-lifecycle.ts b/src/main/claude/claude-background-task-row-lifecycle.ts new file mode 100644 index 00000000000..41c603245db --- /dev/null +++ b/src/main/claude/claude-background-task-row-lifecycle.ts @@ -0,0 +1,203 @@ +import { + canReplaceBackgroundTaskState, + isSettledBackgroundTaskState +} from '../../shared/native-chat-background-task-row' +import type { NativeChatBackgroundTaskBlock } from '../../shared/native-chat-types' +import { + classifyClaudeBackgroundTaskKind, + liveClaudeTaskRunState, + record, + taskAliasId, + taskDescription, + taskName, + taskText, + taskUsageTotalTokens, + terminalClaudeTaskRunState +} from './claude-background-task-frames' + +export type ClaudeBackgroundTaskRow = { + block: NativeChatBackgroundTaskBlock + lastSerialized: string | null + toolUseId?: string +} + +export type ClaudeBackgroundTaskChange = { + state?: NativeChatBackgroundTaskBlock['state'] | null + label?: string | undefined + kind?: NativeChatBackgroundTaskBlock['kind'] | undefined + summary?: string | undefined + error?: string | undefined + outputFile?: string | undefined + tokens?: number | undefined +} + +export function claudeBackgroundTaskToolUseId( + message: Record +): string | undefined { + const patch = record(message.patch) + return taskAliasId(message.tool_use_id) ?? taskAliasId(patch?.tool_use_id) +} + +export function claudeBackgroundTaskNotificationChange( + message: Record +): ClaudeBackgroundTaskChange { + return { + state: terminalClaudeTaskRunState(message.status) ?? 'done', + summary: taskText(message.summary), + error: taskText(message.error), + outputFile: taskText(message.output_file), + tokens: taskUsageTotalTokens(message) + } +} + +export function claudeBackgroundTaskPatchChange( + message: Record +): ClaudeBackgroundTaskChange { + const patch = record(message.patch) ?? message + const status = patch.status ?? message.status + const terminal = terminalClaudeTaskRunState(status) + return { + state: terminal ?? liveClaudeTaskRunState(status), + label: + message.subtype === 'task_progress' + ? undefined + : (taskDescription(patch.description) ?? taskName(patch)), + kind: 'task_type' in patch ? classifyClaudeBackgroundTaskKind(patch.task_type) : undefined, + error: taskText(patch.error), + tokens: taskUsageTotalTokens(message) + } +} + +export function newClaudeBackgroundTaskRow( + id: string, + message: Record, + now: number +): ClaudeBackgroundTaskRow { + const totalTokens = taskUsageTotalTokens(message) + const toolUseId = claudeBackgroundTaskToolUseId(message) + return { + lastSerialized: null, + ...(toolUseId === undefined ? {} : { toolUseId }), + block: { + type: 'background-task', + taskId: id, + kind: classifyClaudeBackgroundTaskKind(message.task_type), + label: taskDescription(message.description) ?? taskName(message) ?? '', + state: + terminalClaudeTaskRunState(message.status) ?? + liveClaudeTaskRunState(message.status) ?? + 'working', + startedAt: now, + ...(totalTokens === undefined ? {} : { tokens: totalTokens }) + } + } +} + +export function newClaudeBackgroundTaskTerminalRow( + id: string, + message: Record +): ClaudeBackgroundTaskRow { + const patch = record(message.patch) + const source = patch ?? message + const toolUseId = claudeBackgroundTaskToolUseId(message) + return { + lastSerialized: null, + ...(toolUseId === undefined ? {} : { toolUseId }), + block: { + type: 'background-task', + taskId: id, + kind: 'task_type' in source ? classifyClaudeBackgroundTaskKind(source.task_type) : 'unknown', + label: taskDescription(source.description) ?? taskName(source) ?? '', + state: 'working' + } + } +} + +export function reopenClaudeBackgroundTaskRow( + row: ClaudeBackgroundTaskRow, + id: string, + message: Record, + state: NativeChatBackgroundTaskBlock['state'], + now: number +): ClaudeBackgroundTaskRow { + const kind = + 'task_type' in message ? classifyClaudeBackgroundTaskKind(message.task_type) : row.block.kind + const totalTokens = taskUsageTotalTokens(message) + const toolUseId = claudeBackgroundTaskToolUseId(message) + return { + ...row, + ...(toolUseId === undefined ? {} : { toolUseId }), + block: { + type: 'background-task', + taskId: id, + kind: kind === 'unknown' ? row.block.kind : kind, + label: taskDescription(message.description) ?? taskName(message) ?? row.block.label, + state, + startedAt: now, + ...(totalTokens === undefined ? {} : { tokens: totalTokens }) + } + } +} + +export function shouldRestartClaudeBackgroundTaskRow( + row: ClaudeBackgroundTaskRow, + message: Record +): boolean { + if (!isSettledBackgroundTaskState(row.block.state) || row.block.startedAt === undefined) { + return false + } + const toolUseId = claudeBackgroundTaskToolUseId(message) + return toolUseId !== undefined && toolUseId !== row.toolUseId +} + +export function canReopenClaudeBackgroundTaskRowFromAggregate( + row: ClaudeBackgroundTaskRow, + state: NativeChatBackgroundTaskBlock['state'] | null +): state is NativeChatBackgroundTaskBlock['state'] { + return ( + state !== null && + !isSettledBackgroundTaskState(state) && + isSettledBackgroundTaskState(row.block.state) && + row.block.state !== 'blocked' + ) +} + +export function isClaudeBackgroundTranscriptTask( + message: Record, + kind: NativeChatBackgroundTaskBlock['kind'] +): boolean { + return message.is_backgrounded === true || kind === 'workflow' || kind === 'monitor' +} + +export function reviseClaudeBackgroundTaskRow( + row: ClaudeBackgroundTaskRow, + change: ClaudeBackgroundTaskChange, + now: number +): void { + const next: NativeChatBackgroundTaskBlock = { ...row.block } + if (change.label && !next.label) { + next.label = change.label + } + if (change.kind !== undefined && change.kind !== 'unknown') { + next.kind = change.kind + } + if (change.summary !== undefined) { + next.summary = change.summary + } + if (change.error !== undefined) { + next.error = change.error + } + if (change.outputFile !== undefined) { + next.outputFile = change.outputFile + } + if (change.tokens !== undefined) { + next.tokens = change.tokens + } + if (change.state && canReplaceBackgroundTaskState(next.state, change.state)) { + next.state = change.state + if (isSettledBackgroundTaskState(change.state)) { + next.settledAt = now + } + } + row.block = next +} diff --git a/src/main/claude/claude-background-task-rows.test.ts b/src/main/claude/claude-background-task-rows.test.ts index bb583d15132..98609b60034 100644 --- a/src/main/claude/claude-background-task-rows.test.ts +++ b/src/main/claude/claude-background-task-rows.test.ts @@ -155,6 +155,27 @@ describe('claude background task rows', () => { expect(items).toEqual([]) }) + it('leaves foreground commands to the ordinary transcript path', () => { + const { rows, items } = harness() + expect( + rows.observe({ + ...START_BASH, + task_id: 'foreground-1', + is_backgrounded: false + }) + ).toBe(true) + expect( + rows.observe({ + type: 'system', + subtype: 'task_notification', + task_id: 'foreground-1', + status: 'failed', + summary: 'foreground command failed' + }) + ).toBe(true) + expect(items).toEqual([]) + }) + it('revises in place rather than opening a row from a patch', () => { const { rows, items, latest } = harness() rows.observe({ @@ -176,6 +197,52 @@ describe('claude background task rows', () => { expect(latest()).toMatchObject({ label: 'Wait for the verification verdict', tokens: 1_200 }) }) + it('opens a row for a terminal update that arrives before the announcement', () => { + const { rows, latest, latestTwin } = harness() + expect( + rows.observe({ + type: 'system', + subtype: 'task_updated', + task_id: 'pre-journal', + patch: { status: 'failed', description: 'Check logs', error: 'boom' } + }) + ).toBe(true) + + expect(latest()).toMatchObject({ + taskId: 'pre-journal', + label: 'Check logs', + state: 'blocked', + error: 'boom' + }) + expect(latestTwin()).toBe('boom') + }) + + it('does not resurrect a task whose terminal edge arrived before its start', () => { + const { rows, items } = harness() + expect( + rows.observe({ + type: 'system', + subtype: 'task_notification', + task_id: 'done-before-start', + status: 'completed' + }) + ).toBe(true) + expect(rows.observe({ ...START_BASH, task_id: 'done-before-start' })).toBe(true) + expect(items).toEqual([]) + }) + + it('does not resurrect after a terminal update that arrived before start', () => { + const { rows, items } = harness() + rows.observe({ + type: 'system', + subtype: 'task_updated', + task_id: 'updated-before-start', + patch: { status: 'completed' } + }) + rows.observe({ ...START_BASH, task_id: 'updated-before-start' }) + expect(items).toEqual([]) + }) + it('latches a reported outcome against a later live tick', () => { const { rows, latest } = harness() rows.observe(START_BASH) @@ -188,6 +255,66 @@ describe('claude background task rows', () => { expect(latest()).toMatchObject({ state: 'blocked' }) }) + it('reopens a settled row when Claude re-announces the same task id with a new tool id', () => { + const { rows, latest } = harness() + rows.observe({ ...START_BASH, task_id: 'resume-1', tool_use_id: 'toolu_first' }) + rows.observe({ + type: 'system', + subtype: 'task_notification', + task_id: 'resume-1', + tool_use_id: 'toolu_first', + status: 'completed', + summary: 'first run finished' + }) + expect(latest()).toMatchObject({ state: 'done', summary: 'first run finished' }) + + rows.observe({ + ...START_BASH, + task_id: 'resume-1', + tool_use_id: 'toolu_second', + status: 'running', + description: 'Second run' + }) + expect(latest()).toMatchObject({ state: 'working', label: 'Second run' }) + expect(latest()).not.toHaveProperty('summary') + + rows.observe({ + type: 'system', + subtype: 'task_notification', + task_id: 'resume-1', + tool_use_id: 'toolu_second', + status: 'failed', + summary: 'second run failed' + }) + expect(latest()).toMatchObject({ state: 'blocked', summary: 'second run failed' }) + }) + + it('lets an aggregate live roster reopen a settled same-id row', () => { + const { rows, latest } = harness() + rows.observe({ ...START_BASH, task_id: 'aggregate-resume' }) + rows.observe({ + type: 'system', + subtype: 'task_notification', + task_id: 'aggregate-resume', + status: 'completed' + }) + + rows.observe({ + type: 'system', + subtype: 'background_tasks_changed', + tasks: [ + { + task_id: 'aggregate-resume', + status: 'running', + task_type: 'local_bash', + description: 'Resumed by roster' + } + ] + }) + + expect(latest()).toMatchObject({ state: 'working', label: 'Resumed by roster' }) + }) + it('never burns a revision on a duplicate delivery', () => { const { rows, items } = harness() rows.observe(START_BASH) @@ -212,6 +339,94 @@ describe('claude background task rows', () => { expect(latest()).toMatchObject({ taskId: 'byjnee2no', state: 'blocked' }) }) + it('rejects overlong tool-use aliases instead of clipping them into collisions', () => { + const { rows, items, latest } = harness() + const overlong = `${'x'.repeat(512)}A` + rows.observe({ ...START_BASH, task_id: 'task-a', tool_use_id: overlong }) + const afterStart = items.length + + expect( + rows.observe({ + type: 'system', + subtype: 'task_notification', + tool_use_id: overlong, + status: 'failed', + summary: 'misattributed failure' + }) + ).toBe(false) + expect(items).toHaveLength(afterStart) + expect(latest()).toMatchObject({ taskId: 'task-a', state: 'working' }) + }) + + it('evicts settled rows so the lifetime cap cannot drop a later failure', () => { + const { rows, latest } = harness() + for (let index = 0; index < 64; index += 1) { + rows.observe({ ...START_BASH, task_id: `settled-${index}` }) + rows.observe({ + type: 'system', + subtype: 'task_notification', + task_id: `settled-${index}`, + status: 'completed' + }) + } + + rows.observe({ ...START_BASH, task_id: 'overflow' }) + rows.observe({ + type: 'system', + subtype: 'task_notification', + task_id: 'overflow', + status: 'failed', + summary: 'overflow failed' + }) + + expect(latest()).toMatchObject({ + taskId: 'overflow', + state: 'blocked', + summary: 'overflow failed' + }) + }) + + it('declines coverage so fallback can report a failure when every row is live', () => { + const { rows, items } = harness() + for (let index = 0; index < 64; index += 1) { + rows.observe({ ...START_BASH, task_id: `live-${index}` }) + } + + expect( + rows.observe({ + type: 'system', + subtype: 'task_notification', + task_id: 'overflow-live', + status: 'failed', + summary: 'overflow failed' + }) + ).toBe(false) + expect( + items.some((item) => + item.identity.provider === 'orca' + ? item.identity.clientMessageId === 'claude-background-task:overflow-live' + : false + ) + ).toBe(false) + }) + + it('bounds foreign-owner memory for tasks rendered elsewhere', () => { + const { rows } = harness() + for (let index = 0; index < 600; index += 1) { + rows.observe({ + type: 'system', + subtype: 'task_started', + task_id: `agent-${index}`, + task_type: 'local_agent', + subagent_type: 'explorer' + }) + } + + const foreign = Reflect.get(rows, 'foreign') + expect(foreign).toBeInstanceOf(Map) + expect(foreign.size).toBeLessThanOrEqual(512) + }) + it('loses contact rather than claiming an outcome when the provider goes away', () => { const { rows, latest, latestTwin } = harness() rows.observe(START_BASH) diff --git a/src/main/claude/claude-background-task-rows.ts b/src/main/claude/claude-background-task-rows.ts index cb85db0f015..272836d41e8 100644 --- a/src/main/claude/claude-background-task-rows.ts +++ b/src/main/claude/claude-background-task-rows.ts @@ -1,51 +1,38 @@ -// The durable transcript row one Claude background task writes. -// -// Claude announces every task — subagent, workflow, monitor, backgrounded shell -// — on one `message:system:task_*` channel. The subagent roster claims the -// agents and excludes the rest, which left the rest with no typed row at all: -// their frames were catalogued as chrome, and the generic payload sniffer then -// promoted the failed ones to a red row whose visible text was the wire opcode. -// -// Suppressing those frames instead is not an option, and that is measured, not -// assumed: when the last background task settles, the tracker flushes it and -// the strip unmounts, `local_bash` is excluded from the roster, and the status -// feed publishes live tasks only. For a lone backgrounded command, this row is -// the ONLY place its failure is ever reported. -// -// So one row per `task_id`, opened by the announcement, revised in place by the -// lifecycle frames, closed by the notification — never one row per frame, which -// is what printed a single failure twice. +// One durable row per Claude background `task_id`, revised in place from the +// lifecycle frames so a failed command prints once with the provider sentence. -import type { - AgentJournalItemBody, - AgentJournalItemIdentity -} from '../../shared/agent-session-journal-types' -import { - backgroundTaskFallbackText, - canReplaceBackgroundTaskState, - isSettledBackgroundTaskState -} from '../../shared/native-chat-background-task-row' -import type { NativeChatBackgroundTaskBlock } from '../../shared/native-chat-types' +import { isSettledBackgroundTaskState } from '../../shared/native-chat-background-task-row' import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' import { classifyClaudeBackgroundTaskKind, - isBoundedClaudeTaskId, liveClaudeTaskRunState, record, taskDescription, taskId as readTaskId, taskName, - taskText, - taskUsageTotalTokens, terminalClaudeTaskRunState } from './claude-background-task-frames' +import { + canReopenClaudeBackgroundTaskRowFromAggregate, + claudeBackgroundTaskNotificationChange, + claudeBackgroundTaskPatchChange, + claudeBackgroundTaskToolUseId, + isClaudeBackgroundTranscriptTask, + newClaudeBackgroundTaskRow, + newClaudeBackgroundTaskTerminalRow, + reopenClaudeBackgroundTaskRow, + reviseClaudeBackgroundTaskRow, + shouldRestartClaudeBackgroundTaskRow, + type ClaudeBackgroundTaskChange, + type ClaudeBackgroundTaskRow +} from './claude-background-task-row-lifecycle' +import { writeClaudeBackgroundTaskRow } from './claude-background-task-row-journal' import { ClaudeSubagentIds } from './claude-subagent-id-aliases' import { isClaudeSubagentTask } from './claude-subagent-task-frames' -/** Rows kept per session. Bounds an event-accumulated map no provider snapshot - * prunes; a session running more concurrent background tasks than this gets no - * row for the overflow rather than an unbounded journal. */ const MAX_TASK_ROWS = 64 +const MAX_FOREIGN_TASK_ROWS = 512 +const MAX_TERMINAL_TASK_IDS = 512 const TASK_SUBTYPES: ReadonlySet = new Set([ 'task_started', @@ -54,34 +41,7 @@ const TASK_SUBTYPES: ReadonlySet = new Set([ 'task_notification' ]) -/** Who owns an id this module is not writing a row for. */ -type ForeignOwner = - /** The subagent roster announced it as an agent and renders it already. */ - | 'roster' - /** Housekeeping Claude runs for itself; the user never asked for it. */ - | 'ambient' - -type TaskRow = { block: NativeChatBackgroundTaskBlock; lastSerialized: string | null } - -/** Durable journal identity for a task's row — stable across revisions and - * across a restart, so replay finds the same row instead of appending one. */ -export function claudeBackgroundTaskIdentity(taskId: string): AgentJournalItemIdentity { - return { provider: 'orca', clientMessageId: `claude-background-task:${taskId}` } -} - -/** The row: the structured block plus the plain sentence a client without the - * block type renders in its place. A message whose only block is the new - * variant would reach such a client with nothing it can draw, and a new item - * KIND would reach it as nothing at all. */ -export function claudeBackgroundTaskBody( - block: NativeChatBackgroundTaskBlock -): AgentJournalItemBody { - return { - kind: 'message', - role: 'system', - blocks: [{ type: 'text', text: backgroundTaskFallbackText(block) }, { ...block }] - } -} +type ForeignOwner = 'roster' | 'ambient' | 'foreground' export type ClaudeBackgroundTaskRowsDeps = { sink: StructuredAgentSessionEventSink @@ -89,8 +49,9 @@ export type ClaudeBackgroundTaskRowsDeps = { } export class ClaudeBackgroundTaskRows { - private readonly rows = new Map() + private readonly rows = new Map() private readonly foreign = new Map() + private readonly terminalTaskIds = new Set() private readonly ids = new ClaudeSubagentIds() private readonly now: () => number @@ -98,14 +59,14 @@ export class ClaudeBackgroundTaskRows { this.now = deps.now ?? (() => Date.now()) } - /** Consume a background-task lifecycle frame. Returns false when it is not - * one. A `true` return means this module owns the frame, INCLUDING when it - * deliberately writes nothing for it. */ observe(message: Record): boolean { if (message.type !== 'system') { return false } if (message.subtype === 'background_tasks_changed') { + if (!Array.isArray(message.tasks)) { + return false + } this.observeAggregateRoster(message.tasks) return true } @@ -114,28 +75,20 @@ export class ClaudeBackgroundTaskRows { } const id = this.canonicalId(message) if (id === null) { - return true + return false } if (message.subtype === 'task_started') { - this.observeStart(id, message) - return true + return this.observeStart(id, message) } if (this.foreign.has(id)) { return true } if (message.subtype === 'task_notification') { - this.observeNotification(id, message) - return true + return this.observeNotification(id, message) } - // `task_updated` and `task_progress` are patches, not announcements: neither - // carries a `task_type`, so honouring one for an id nothing declared would - // row whatever else shares this channel. They revise, never create. - this.revise(id, this.patchChange(message)) - return true + return this.observePatch(id, message) } - /** Nothing more will arrive for any task, so every live row loses contact. - * That is not evidence it exited (docs/reference/ssh-execution-boundary.md). */ settleSession(): void { for (const [id, row] of this.rows) { if (!isSettledBackgroundTaskState(row.block.state)) { @@ -145,128 +98,109 @@ export class ClaudeBackgroundTaskRows { } dispose(): void { - // Teardown reaches here without an `ended` event, so a row still reporting - // live work would have nothing left to revise it. this.settleSession() this.rows.clear() this.foreign.clear() + this.terminalTaskIds.clear() this.ids.clear() } - /** The task id a frame names, with its tool id recorded as an alias: Claude - * re-announces a resumed task under a NEW `tool_use_id` while `task_id` - * stays put, so keying on the tool id would show the task twice. */ private canonicalId(message: Record): string | null { - const patch = record(message.patch) const declared = readTaskId(message) - const toolUseId = taskText(message.tool_use_id) ?? taskText(patch?.tool_use_id) + const toolUseId = claudeBackgroundTaskToolUseId(message) if (declared === null) { const aliased = toolUseId === undefined ? null : this.ids.canonical(toolUseId) return aliased !== null && aliased !== toolUseId ? aliased : null } - if (toolUseId !== undefined && isBoundedClaudeTaskId(toolUseId)) { + if (toolUseId !== undefined) { this.ids.alias(toolUseId, declared) } return declared } - private observeStart(id: string, message: Record): void { + private observeStart(id: string, message: Record): boolean { if (message.ambient === true || message.skip_transcript === true) { - this.foreign.set(id, 'ambient') - return + this.rememberForeign(id, 'ambient') + return true } if (isClaudeSubagentTask(message)) { - this.foreign.set(id, 'roster') - return + this.rememberForeign(id, 'roster') + return true + } + const kind = classifyClaudeBackgroundTaskKind(message.task_type) + if (!isClaudeBackgroundTranscriptTask(message, kind)) { + this.rememberForeign(id, 'foreground') + return true } this.foreign.delete(id) const existing = this.rows.get(id) if (existing) { - this.revise(id, this.patchChange(message)) - return - } - if (this.rows.size >= MAX_TASK_ROWS) { - return - } - const now = this.now() - this.rows.set(id, { - lastSerialized: null, - block: { - type: 'background-task', - taskId: id, - kind: classifyClaudeBackgroundTaskKind(message.task_type), - label: taskDescription(message.description) ?? taskName(message) ?? '', - state: liveClaudeTaskRunState(message.status) ?? 'working', - startedAt: now, - ...(taskUsageTotalTokens(message) === undefined - ? {} - : { tokens: taskUsageTotalTokens(message) }) + if (shouldRestartClaudeBackgroundTaskRow(existing, message)) { + this.rows.set(id, newClaudeBackgroundTaskRow(id, message, this.now())) + this.write(id) + } else { + this.revise(id, claudeBackgroundTaskPatchChange(message)) } - }) + return true + } + if (this.terminalTaskIds.has(id)) { + return true + } + if (!this.ensureRowSlot()) { + return false + } + this.rows.set(id, newClaudeBackgroundTaskRow(id, message, this.now())) this.write(id) + return true } - private observeNotification(id: string, message: Record): void { - // The notification is affirmative terminal evidence even when its status is - // unreadable, matching the liveness semantics this channel always had. - const state = terminalClaudeTaskRunState(message.status) ?? 'done' - const change: TaskChange = { - state, - summary: taskText(message.summary), - error: taskText(message.error), - outputFile: taskText(message.output_file), - tokens: taskUsageTotalTokens(message) - } + private observeNotification(id: string, message: Record): boolean { + const change = claudeBackgroundTaskNotificationChange(message) + const state = change.state ?? 'done' + this.rememberTerminalId(id) if (this.rows.has(id)) { this.revise(id, change) - return + return true } - // The first and last frame for a task this session never saw start — a - // resumed session, or an announcement that predates the journal. A failure - // here is the only report the user will ever get, so it opens a row of its - // own; a silent success is not worth one nobody asked for. if (state === 'done' && change.error === undefined) { - return + return true } - if (this.rows.size >= MAX_TASK_ROWS) { - return + if (!this.ensureRowSlot()) { + return false } this.rows.set(id, { - lastSerialized: null, - block: { - type: 'background-task', - taskId: id, - kind: 'unknown', - label: taskDescription(message.description) ?? taskName(message) ?? '', - state: 'working' - } + ...newClaudeBackgroundTaskTerminalRow(id, message) }) this.revise(id, change) + return true } - /** A `task_updated` patch or a `task_progress` tick, read as a row change. - * `task_progress` carries the CURRENT ACTIVITY in `description`, not the - * task's name, so only a row still missing a label takes one from it. */ - private patchChange(message: Record): TaskChange { - const patch = record(message.patch) ?? message - const status = patch.status ?? message.status - const terminal = terminalClaudeTaskRunState(status) - return { - state: terminal ?? liveClaudeTaskRunState(status), - label: - message.subtype === 'task_progress' - ? undefined - : (taskDescription(patch.description) ?? taskName(patch)), - kind: 'task_type' in patch ? classifyClaudeBackgroundTaskKind(patch.task_type) : undefined, - error: taskText(patch.error), - tokens: taskUsageTotalTokens(message) + private observePatch(id: string, message: Record): boolean { + const patch = record(message.patch) + if (patch?.is_backgrounded === false) { + this.rememberForeign(id, 'foreground') + return true } + const change = claudeBackgroundTaskPatchChange(message) + if (this.rows.has(id)) { + this.revise(id, change) + return true + } + if (change.state && isSettledBackgroundTaskState(change.state)) { + this.rememberTerminalId(id) + if (change.state === 'done' && change.error === undefined) { + return true + } + if (!this.ensureRowSlot()) { + return false + } + this.rows.set(id, newClaudeBackgroundTaskTerminalRow(id, message)) + this.revise(id, change) + return true + } + return change.error === undefined } - /** The aggregate roster enumerates BACKGROUND work only, so it is - * authoritative over the tasks it lists and silent about everything else. A - * task missing from it is not thereby finished — only its own terminal frame - * says that — so this revises listed rows and creates none. */ private observeAggregateRoster(value: unknown): void { if (!Array.isArray(value)) { return @@ -277,8 +211,15 @@ export class ClaudeBackgroundTaskRows { if (task === null || id === null || !this.rows.has(id)) { continue } + const state = terminalClaudeTaskRunState(task.status) ?? liveClaudeTaskRunState(task.status) + const row = this.rows.get(id) + if (row && canReopenClaudeBackgroundTaskRowFromAggregate(row, state)) { + this.rows.set(id, reopenClaudeBackgroundTaskRow(row, id, task, state, this.now())) + this.write(id) + continue + } this.revise(id, { - state: terminalClaudeTaskRunState(task.status) ?? liveClaudeTaskRunState(task.status), + state, label: taskDescription(task.description) ?? taskName(task), kind: task.task_type === undefined @@ -288,39 +229,54 @@ export class ClaudeBackgroundTaskRows { } } - private revise(id: string, change: TaskChange): void { + private revise(id: string, change: ClaudeBackgroundTaskChange): void { const row = this.rows.get(id) if (!row) { return } - const next: NativeChatBackgroundTaskBlock = { ...row.block } - if (change.label && !next.label) { - next.label = change.label + reviseClaudeBackgroundTaskRow(row, change, this.now()) + this.write(id) + } + + private ensureRowSlot(): boolean { + if (this.rows.size < MAX_TASK_ROWS) { + return true } - if (change.kind !== undefined && change.kind !== 'unknown') { - next.kind = change.kind - } - if (change.summary !== undefined) { - next.summary = change.summary - } - if (change.error !== undefined) { - next.error = change.error - } - if (change.outputFile !== undefined) { - next.outputFile = change.outputFile - } - if (change.tokens !== undefined) { - next.tokens = change.tokens - } - // Proven outcomes latch; lost contact can still receive a later verdict. - if (change.state && canReplaceBackgroundTaskState(next.state, change.state)) { - next.state = change.state - if (isSettledBackgroundTaskState(change.state)) { - next.settledAt = this.now() + for (const [id, row] of this.rows) { + if (isSettledBackgroundTaskState(row.block.state)) { + this.rows.delete(id) + return true } } - row.block = next - this.write(id) + return false + } + + private rememberForeign(id: string, owner: ForeignOwner): void { + if (this.foreign.has(id)) { + this.foreign.delete(id) + } + this.foreign.set(id, owner) + while (this.foreign.size > MAX_FOREIGN_TASK_ROWS) { + const oldest = this.foreign.keys().next() + if (oldest.done || oldest.value === id) { + break + } + this.foreign.delete(oldest.value) + } + } + + private rememberTerminalId(id: string): void { + if (this.terminalTaskIds.has(id)) { + this.terminalTaskIds.delete(id) + } + this.terminalTaskIds.add(id) + while (this.terminalTaskIds.size > MAX_TERMINAL_TASK_IDS) { + const oldest = this.terminalTaskIds.values().next() + if (oldest.done || oldest.value === id) { + break + } + this.terminalTaskIds.delete(oldest.value) + } } private write(id: string): void { @@ -328,28 +284,6 @@ export class ClaudeBackgroundTaskRows { if (!row) { return } - const body = claudeBackgroundTaskBody(row.block) - const serialized = JSON.stringify(body) - if (serialized === row.lastSerialized) { - // Nothing changed — a duplicate delivery must not burn a revision. - return - } - row.lastSerialized = serialized - this.deps.sink.appendItem(claudeBackgroundTaskIdentity(id), body, { - coalescingKey: `claude-background-task:${id}` - }) - // Publish keeps the sink's own coalescing slot: sharing the row's key makes - // each queued publish evict the append it was meant to flush. - this.deps.sink.publish() + writeClaudeBackgroundTaskRow(this.deps.sink, id, row) } } - -type TaskChange = { - state?: NativeChatBackgroundTaskBlock['state'] | null - label?: string | undefined - kind?: NativeChatBackgroundTaskBlock['kind'] | undefined - summary?: string | undefined - error?: string | undefined - outputFile?: string | undefined - tokens?: number | undefined -} diff --git a/src/main/claude/claude-structured-journal-translation-background-tasks.test.ts b/src/main/claude/claude-structured-journal-translation-background-tasks.test.ts index 3b0268957d3..0981a0813f4 100644 --- a/src/main/claude/claude-structured-journal-translation-background-tasks.test.ts +++ b/src/main/claude/claude-structured-journal-translation-background-tasks.test.ts @@ -122,4 +122,18 @@ describe('claude journal translation — background task rows', () => { // sniffer for the kinds nobody has modelled. expect(fallbackRows()).toEqual(['claude · message:system:future_event']) }) + + it('falls back visibly when a malformed task frame reports a failure', () => { + const { translator, fallbackRows, taskRowIds } = harness() + translator.handle( + systemFrame({ + subtype: 'task_notification', + status: 'failed', + summary: 'Background command "Wait" failed with exit code 1' + }) + ) + + expect(taskRowIds()).toEqual([]) + expect(fallbackRows()).toEqual(['Background command "Wait" failed with exit code 1']) + }) }) diff --git a/src/main/claude/claude-structured-journal-translation.ts b/src/main/claude/claude-structured-journal-translation.ts index 2b3c46f92bd..7615fb161f4 100644 --- a/src/main/claude/claude-structured-journal-translation.ts +++ b/src/main/claude/claude-structured-journal-translation.ts @@ -22,12 +22,8 @@ import { readClaudeMessageEnvelope, type ClaudeToolUse } from './claude-structured-item-translation' -import { - claudeApprovalItem, - claudePromptIdentity, - claudeQuestionItems -} from './claude-structured-prompt-items' import type { ClaudePromptRegistry } from './claude-structured-prompt-replies' +import { appendClaudePromptJournalItems } from './claude-structured-prompt-journal' import { claudeProviderFrameActivity } from '../native-chat/agent-session-wire/provider-frame-activity' import { appendUnmodeledClaudeContent, @@ -108,7 +104,6 @@ export function createClaudeJournalTranslator( const publishLifecycle = (turn: ClaudeCurrentTurn, end?: ClaudeTurnEnd): void => { const item = claudeTurnLifecycleItem(turn, end) deps.sink.appendItem(item.identity, item.body, item.options) - // Preserve first-work evidence when completion arrives before the journal drains. deps.sink.publish({ coalescingKey: item.publishCoalescingKey }) } @@ -146,7 +141,6 @@ export function createClaudeJournalTranslator( } const outputEnvelope = claudeOutputEnvelope(envelope) const body = claudeMessageBody(outputEnvelope) - // The final frame of a streamed block lands on the block's identity, not its own uuid. const identity = (body && envelope.role === 'assistant' ? streamedBlocks.reconcile(envelope) : null) ?? claudeMessageIdentity(envelope) @@ -173,9 +167,7 @@ export function createClaudeJournalTranslator( claudeToolIdentity(envelope.sessionId, result.toolUseId), claudeToolBody({ tool, result }) ) - // A spawn call's result is the parent turn's evidence its child finished. subagents.observeToolResult(result.toolUseId, result.failed) - // Tool inputs are only needed until their matching result arrives. tools.delete(result.toolUseId) changed = true } @@ -198,8 +190,6 @@ export function createClaudeJournalTranslator( message.parent_tool_use_id === null ) { if (currentTurn) { - // A new turn starting is the only end the previous one gets when its - // result never arrives; settling it later would sweep THIS turn. subagents.settleTurn(groupKeyOf(currentTurn)) publishLifecycle(currentTurn, { state: 'interrupted', completedAt: observedAt }) } @@ -207,7 +197,6 @@ export function createClaudeJournalTranslator( sessionId: envelope.sessionId, turnId: envelope.uuid, startedAt: observedAt, - // A user echo lands on its own message identity, so this is the user row's key. userItemId: agentJournalItemKey(identity) } publishLifecycle(currentTurn) @@ -220,25 +209,11 @@ export function createClaudeJournalTranslator( } const handlePrompt = (event: Extract): void => { - const identities: AgentJournalItemIdentity[] = [] - if (event.prompt.kind === 'question') { - for (const question of claudeQuestionItems({ - sessionId: event.sessionId, - prompt: event.prompt - })) { - identities.push(question.identity) - deps.sink.appendItem(question.identity, question.body) - deps.bindPromptItemId?.(agentJournalItemKey(question.identity), event.prompt.promptKey) - } - } else { - const identity = claudePromptIdentity({ - sessionId: event.sessionId, - promptKey: event.prompt.promptKey - }) - identities.push(identity) - deps.sink.appendItem(identity, claudeApprovalItem(event.prompt)) - deps.bindPromptItemId?.(agentJournalItemKey(identity), event.prompt.promptKey) - } + const identities = appendClaudePromptJournalItems({ + event, + sink: deps.sink, + ...(deps.bindPromptItemId ? { bindPromptItemId: deps.bindPromptItemId } : {}) + }) promptItems.set(event.prompt.promptKey, identities) deps.sink.publish() } @@ -247,10 +222,8 @@ export function createClaudeJournalTranslator( handle: (event) => { if (event.type === 'ended') { streamedText.flush() - // No event will ever settle a child once the provider is gone. subagents.settleSession() if (currentTurn) { - // The host saw the child end, so the turn's end is observed, not lost. publishLifecycle(currentTurn, { state: 'interrupted', completedAt: event.observedAt ?? Date.now() @@ -273,8 +246,6 @@ export function createClaudeJournalTranslator( promptItems.delete(event.promptKey) deps.sink.publish() } else if (event.type === 'message' && event.message.type === 'result') { - // The turn is over however it ended, so a foreground child still - // reported as working will never be settled by an event. subagents.settleTurn(groupKeyOf(currentTurn)) if (currentTurn) { publishLifecycle( @@ -284,27 +255,23 @@ export function createClaudeJournalTranslator( currentTurn = null } deps.sink.setActivity?.(null) - // The turn is over. A block still awaiting its final keeps the text the - // flush above journaled, but its live state goes: an interrupted turn - // would otherwise retain that text for the life of the session. streamedBlocks.clear() streamedText.settle() const kind = claudeProviderFrameKind(event.message) - // Ordinary turn bookkeeping stays suppressed; a reported failure never does. const failure = claudeResultFailure(event.message) if (failure || !isSettledClaudeResultKind(kind)) { providerFallback.append(kind, event.message, failure?.text) } } else if (event.type === 'message') { - // The roster claims the agent tasks and the row owner claims the rest; - // both kinds are covered, so the fallback below emits nothing for them. subagents.observeSystemFrame(event.message) - backgroundTasks.observe(event.message) + const backgroundTaskCovered = backgroundTasks.observe(event.message) const kind = claudeProviderFrameKind(event.message) if ( !handleMessage(event.message, event.startsTurn === true, event.observedAt ?? Date.now()) ) { - providerFallback.append(kind, event.message) + providerFallback.append(kind, event.message, undefined, { + coveredByTypedTranslator: backgroundTaskCovered + }) } publishActivity(kind, event.message) } else if (event.type === 'provider-frame') { @@ -322,9 +289,6 @@ export function createClaudeJournalTranslator( promptItems.clear() streamedBlocks.clear() subagents.dispose() - // Settles every live row: `ended` is delivered immediately before this on - // every provider-exit path, and this one also covers a teardown with no - // `ended` at all. backgroundTasks.dispose() } } diff --git a/src/main/claude/claude-structured-prompt-journal.ts b/src/main/claude/claude-structured-prompt-journal.ts new file mode 100644 index 00000000000..8eee098fe7b --- /dev/null +++ b/src/main/claude/claude-structured-prompt-journal.ts @@ -0,0 +1,37 @@ +import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key' +import type { AgentJournalItemIdentity } from '../../shared/agent-session-journal-types' +import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import type { ClaudeStructuredSessionEvent } from './claude-structured-session-state' +import { + claudeApprovalItem, + claudePromptIdentity, + claudeQuestionItems +} from './claude-structured-prompt-items' + +export function appendClaudePromptJournalItems(input: { + event: Extract + sink: StructuredAgentSessionEventSink + bindPromptItemId?: (journalItemId: string, promptKey: string, questionId?: string) => void +}): AgentJournalItemIdentity[] { + const identities: AgentJournalItemIdentity[] = [] + const { event, sink, bindPromptItemId } = input + if (event.prompt.kind === 'question') { + for (const question of claudeQuestionItems({ + sessionId: event.sessionId, + prompt: event.prompt + })) { + identities.push(question.identity) + sink.appendItem(question.identity, question.body) + bindPromptItemId?.(agentJournalItemKey(question.identity), event.prompt.promptKey) + } + return identities + } + const identity = claudePromptIdentity({ + sessionId: event.sessionId, + promptKey: event.prompt.promptKey + }) + identities.push(identity) + sink.appendItem(identity, claudeApprovalItem(event.prompt)) + bindPromptItemId?.(agentJournalItemKey(identity), event.prompt.promptKey) + return identities +} diff --git a/src/main/claude/claude-structured-provider-fallback.ts b/src/main/claude/claude-structured-provider-fallback.ts index 68aec07976b..ab54b49790b 100644 --- a/src/main/claude/claude-structured-provider-fallback.ts +++ b/src/main/claude/claude-structured-provider-fallback.ts @@ -5,6 +5,7 @@ import { } from '../native-chat/agent-session-journal/journal-payload-bounds' import { CLAUDE_STREAM_JSON_FRAME_KINDS } from '../native-chat/agent-session-wire/claude-stream-json-frame-schema' import { + type UnhandledProviderFrameJournalItemOptions, readableProviderFrameText, unhandledProviderFrameJournalItem } from '../native-chat/agent-session-wire/unhandled-provider-frame' @@ -106,13 +107,24 @@ export function createClaudeProviderFrameFallback( acquisitionId: string ): { /** `displayText` leads the row when Claude knows the sentence the frame itself does not name. */ - append: (kind: string, payload: unknown, displayText?: string | null) => void + append: ( + kind: string, + payload: unknown, + displayText?: string | null, + options?: UnhandledProviderFrameJournalItemOptions + ) => void } { let sequence = 0 return { - append: (kind, payload, displayText) => { + append: (kind, payload, displayText, options) => { sequence += 1 - const translated = unhandledProviderFrameJournalItem('claude', kind, payload) + const translated = unhandledProviderFrameJournalItem( + 'claude', + kind, + payload, + DEFAULT_JOURNAL_PAYLOAD_LIMITS, + options + ) if (!translated) { return } diff --git a/src/main/native-chat/agent-session-journal/journal-subagent-liveness.test.ts b/src/main/native-chat/agent-session-journal/journal-subagent-liveness.test.ts index 9d6887de61e..b0dff559b1a 100644 --- a/src/main/native-chat/agent-session-journal/journal-subagent-liveness.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-subagent-liveness.test.ts @@ -7,8 +7,12 @@ import type { AgentSessionJournalIdentity } from '../../../shared/agent-session-journal-types' import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' -import { isSubagentGroupBlock } from '../../../shared/native-chat-types' -import type { NativeChatSubagentEntry } from '../../../shared/native-chat-types' +import { backgroundTaskFallbackText } from '../../../shared/native-chat-background-task-row' +import { isBackgroundTaskBlock, isSubagentGroupBlock } from '../../../shared/native-chat-types' +import type { + NativeChatBackgroundTaskBlock, + NativeChatSubagentEntry +} from '../../../shared/native-chat-types' import { codexSubagentGroupBody, codexSubagentGroupIdentity @@ -55,6 +59,34 @@ function rosterRow(agents: NativeChatSubagentEntry[]) { } } +function backgroundTaskBlock( + overrides: Partial = {} +): NativeChatBackgroundTaskBlock { + return { + type: 'background-task', + taskId: 'task-1', + kind: 'command', + label: 'sleep 20', + state: 'working', + startedAt: 10, + ...overrides + } +} + +function backgroundTaskRow(block = backgroundTaskBlock()) { + return { + identity: { + provider: 'orca' as const, + clientMessageId: `claude-background-task:${block.taskId}` + }, + body: { + kind: 'message' as const, + role: 'system' as const, + blocks: [{ type: 'text' as const, text: backgroundTaskFallbackText(block) }, block] + } + } +} + function renderItem(agents: NativeChatSubagentEntry[]): AgentJournalRenderItem { const row = rosterRow(agents) return { @@ -70,6 +102,10 @@ function rosterOf(body: AgentJournalRenderItem['body']): NativeChatSubagentEntry return body.kind === 'message' ? (body.blocks.find(isSubagentGroupBlock)?.agents ?? []) : [] } +function taskOf(body: AgentJournalRenderItem['body']): NativeChatBackgroundTaskBlock | undefined { + return body.kind === 'message' ? body.blocks.find(isBackgroundTaskBlock) : undefined +} + function twinOf(body: AgentJournalRenderItem['body']): string | undefined { return body.kind === 'message' ? body.blocks.find((block) => block.type === 'text')?.text @@ -104,6 +140,23 @@ describe('staleSubagentRosterRevisions', () => { expect(twinOf(revisions[0]!.body)).toBe('Ran 2 subagents (1 unverifiable)') }) + it('settles a background task the previous host left live, and moves the twin with it', () => { + const row = backgroundTaskRow() + const revisions = staleSubagentRosterRevisions([ + { + itemId: agentJournalItemKey(row.identity), + revision: 1, + body: row.body, + sequence: 2, + observedAt: 1 + } + ]) + + expect(revisions).toHaveLength(1) + expect(taskOf(revisions[0]!.body)).toMatchObject({ taskId: 'task-1', state: 'unverifiable' }) + expect(twinOf(revisions[0]!.body)).toBe('Background command "sleep 20" stopped reporting') + }) + // The child stopped being observable at an unknown moment. A stamp taken now // would report the time the app was down as how long the child ran. it('records no terminal timestamp for a child whose run length is unknown', () => { @@ -114,6 +167,21 @@ describe('staleSubagentRosterRevisions', () => { expect(rosterOf(revisions[0]!.body)[0]).not.toHaveProperty('settledAt') }) + it('records no terminal timestamp for a background task whose run length is unknown', () => { + const row = backgroundTaskRow(backgroundTaskBlock({ settledAt: 20 })) + const revisions = staleSubagentRosterRevisions([ + { + itemId: agentJournalItemKey(row.identity), + revision: 1, + body: row.body, + sequence: 2, + observedAt: 1 + } + ]) + + expect(taskOf(revisions[0]!.body)).not.toHaveProperty('settledAt') + }) + it('owes nothing for a roster whose children all settled', () => { expect( staleSubagentRosterRevisions([ @@ -186,6 +254,22 @@ describe('journal reopen after the writing host is gone', () => { expect(reopened.snapshot().items.at(-1)?.revision).toBe(2) }) + it('settles a persisted working background task to unverifiable', async () => { + const live = await open() + const row = backgroundTaskRow() + await live.appendItem(row.identity, row.body, { fence: 0 }) + const beforeRestart = live.snapshot().items.at(-1)! + expect(taskOf(beforeRestart.body)).toMatchObject({ state: 'working' }) + expect(twinOf(beforeRestart.body)).toBe('Started background command "sleep 20"') + await live.close() + + const reopened = await open() + const afterRestart = reopened.snapshot().items.at(-1)! + expect(afterRestart.itemId).toBe(beforeRestart.itemId) + expect(taskOf(afterRestart.body)).toMatchObject({ state: 'unverifiable' }) + expect(twinOf(afterRestart.body)).toBe('Background command "sleep 20" stopped reporting') + }) + it('writes nothing on a second reopen once every child is settled', async () => { const live = await open() const row = rosterRow([{ id: 'a', label: 'read', state: 'working', startedAt: 10 }]) diff --git a/src/main/native-chat/agent-session-journal/journal-subagent-liveness.ts b/src/main/native-chat/agent-session-journal/journal-subagent-liveness.ts index 9b2724e9d1d..f3868eedcf4 100644 --- a/src/main/native-chat/agent-session-journal/journal-subagent-liveness.ts +++ b/src/main/native-chat/agent-session-journal/journal-subagent-liveness.ts @@ -27,8 +27,15 @@ import { subagentGroupFallbackText } from '../../../shared/native-chat-subagent-summary' import { + backgroundTaskFallbackText, + isSettledBackgroundTaskState, + normalizeBackgroundTaskState +} from '../../../shared/native-chat-background-task-row' +import { + isBackgroundTaskBlock, isSubagentGroupBlock, type NativeChatBlock, + type NativeChatBackgroundTaskBlock, type NativeChatSubagentGroupBlock } from '../../../shared/native-chat-types' @@ -45,7 +52,7 @@ export function staleSubagentRosterRevisions( const revisions: JournalSubagentLivenessRevision[] = [] for (const item of items) { const body = item.body - if (body.kind !== 'message' || !body.blocks.some(hasWorkingChild)) { + if (body.kind !== 'message' || !body.blocks.some(hasStaleLiveWork)) { continue } // A key that will not parse cannot be re-addressed, and appending under a @@ -59,30 +66,58 @@ export function staleSubagentRosterRevisions( return revisions } -function hasWorkingChild(block: NativeChatBlock): boolean { +function hasStaleLiveWork(block: NativeChatBlock): boolean { + return hasWorkingChild(block) || hasLiveBackgroundTask(block) +} + +function hasWorkingChild(block: NativeChatBlock): block is NativeChatSubagentGroupBlock { return ( isSubagentGroupBlock(block) && block.agents.some((agent) => normalizeSubagentState(agent.state) === 'working') ) } +function hasLiveBackgroundTask(block: NativeChatBlock): block is NativeChatBackgroundTaskBlock { + return ( + isBackgroundTaskBlock(block) && + !isSettledBackgroundTaskState(normalizeBackgroundTaskState(block.state)) + ) +} + /** No `settledAt`: the child stopped being observable at an unknown moment, and * stamping the reopen would report the time the app was down as how long it * ran. Readers already draw an unverifiable child with no stamp as having no * known run length. */ function settleBlocks(blocks: readonly NativeChatBlock[]): NativeChatBlock[] { - const settled = blocks.map((block) => - hasWorkingChild(block) ? settleGroup(block as NativeChatSubagentGroupBlock) : block + const backgroundTaskTwinText = new Map() + const settled = blocks.map((block) => { + if (hasWorkingChild(block)) { + return settleGroup(block) + } + if (hasLiveBackgroundTask(block)) { + const next = settleBackgroundTask(block) + backgroundTaskTwinText.set( + backgroundTaskFallbackText(block), + backgroundTaskFallbackText(next) + ) + return next + } + return block + }) + const withBackgroundTaskTwins = settled.map((block) => + block.type === 'text' + ? { ...block, text: backgroundTaskTwinText.get(block.text) ?? block.text } + : block ) - const rosters = settled.filter(isSubagentGroupBlock) + const rosters = withBackgroundTaskTwins.filter(isSubagentGroupBlock) const only = rosters.length === 1 ? rosters[0] : undefined if (!only) { - return settled + return withBackgroundTaskTwins } // The plain-text twin is all a client without the block type ever shows, so it // has to move with the block or the two would disagree about the same row. const twin = subagentGroupFallbackText(only.agents) - return settled.map((block) => + return withBackgroundTaskTwins.map((block) => block.type === 'text' && isSubagentGroupFallbackText(block.text) ? { ...block, text: twin } : block @@ -99,3 +134,8 @@ function settleGroup(block: NativeChatSubagentGroupBlock): NativeChatSubagentGro ) } } + +function settleBackgroundTask(block: NativeChatBackgroundTaskBlock): NativeChatBackgroundTaskBlock { + const { settledAt: _settledAt, ...withoutSettledAt } = block + return { ...withoutSettledAt, state: 'unverifiable' } +} diff --git a/src/main/native-chat/agent-session-wire/provider-frame-disposition.test.ts b/src/main/native-chat/agent-session-wire/provider-frame-disposition.test.ts index 3de9b68ba90..f0f41cb3768 100644 --- a/src/main/native-chat/agent-session-wire/provider-frame-disposition.test.ts +++ b/src/main/native-chat/agent-session-wire/provider-frame-disposition.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it } from 'vitest' +import { DEFAULT_JOURNAL_PAYLOAD_LIMITS } from '../agent-session-journal/journal-payload-bounds' import { CODEX_APP_SERVER_NOTIFICATION_METHODS } from '../../codex/codex-app-server-notification-schema' import { CLAUDE_STREAM_JSON_FRAME_KINDS } from './claude-stream-json-frame-schema' import { @@ -180,16 +181,34 @@ describe('typed translator coverage', () => { 'message:system:background_tasks_changed' ]) { expect( - unhandledProviderFrameJournalItem('claude', kind, { - task_id: 'byjnee2no', - status: 'failed', - summary: 'Background command "Wait" failed with exit code 1' - }), + unhandledProviderFrameJournalItem( + 'claude', + kind, + { + task_id: 'byjnee2no', + status: 'failed', + summary: 'Background command "Wait" failed with exit code 1' + }, + DEFAULT_JOURNAL_PAYLOAD_LIMITS, + { coveredByTypedTranslator: true } + ), kind ).toBeNull() } }) + it('keeps malformed covered-kind failures eligible for the generic fallback', () => { + expect( + unhandledProviderFrameJournalItem('claude', 'message:system:task_notification', { + status: 'failed', + summary: 'Background command "Wait" failed with exit code 1' + }) + ).toMatchObject({ + classification: 'error-surface', + body: { text: 'Background command "Wait" failed with exit code 1' } + }) + }) + it('covers Claude only — the same method name on another provider still falls back', () => { expect( unhandledProviderFrameJournalItem('codex', 'message:system:task_notification', { diff --git a/src/main/native-chat/agent-session-wire/structured-rewind-journal-body.test.ts b/src/main/native-chat/agent-session-wire/structured-rewind-journal-body.test.ts index e93aef352fe..ac13c75d480 100644 --- a/src/main/native-chat/agent-session-wire/structured-rewind-journal-body.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-rewind-journal-body.test.ts @@ -16,6 +16,26 @@ describe('rewind recovery of newer durable records', () => { blocks: [{ type: 'text', text: '{"type":"future-block"}' }] }) }) + + it('preserves background-task blocks across rewind recovery', () => { + const body = { + kind: 'message' as const, + role: 'system', + blocks: [ + { type: 'text' as const, text: 'Started background command "sleep 20"' }, + { + type: 'background-task' as const, + taskId: 'task-1', + kind: 'command', + label: 'sleep 20', + state: 'working' + } + ] + } + + expect(restoreRewindJournalBody(body)).toEqual(body) + }) + it('preserves unknown state as evidence rather than inventing success or pending work', () => { const body = { kind: 'tool-call' as const, diff --git a/src/main/native-chat/agent-session-wire/structured-rewind-journal-body.ts b/src/main/native-chat/agent-session-wire/structured-rewind-journal-body.ts index e23058a7d7b..db92d55e110 100644 --- a/src/main/native-chat/agent-session-wire/structured-rewind-journal-body.ts +++ b/src/main/native-chat/agent-session-wire/structured-rewind-journal-body.ts @@ -21,7 +21,12 @@ export function restoreRewindJournalBody(body: StoredBody): AgentJournalItemBody (block.type === 'text' && 'text' in block) || (block.type === 'tool-call' && 'name' in block && !('state' in block)) || (block.type === 'tool-result' && 'output' in block) || - block.type === 'image-ref' + block.type === 'image-ref' || + (block.type === 'background-task' && + 'taskId' in block && + 'kind' in block && + 'label' in block && + 'state' in block) ) { return block } diff --git a/src/main/native-chat/agent-session-wire/unhandled-provider-frame.ts b/src/main/native-chat/agent-session-wire/unhandled-provider-frame.ts index 454ed23f1fd..1a89e230a51 100644 --- a/src/main/native-chat/agent-session-wire/unhandled-provider-frame.ts +++ b/src/main/native-chat/agent-session-wire/unhandled-provider-frame.ts @@ -17,6 +17,11 @@ export type UnhandledProviderFrameJournalItem = { classification: 'timeline-substantive' | 'error-surface' } +export type UnhandledProviderFrameJournalItemOptions = { + /** A typed translator accepted this exact frame, not merely this frame kind. */ + coveredByTypedTranslator?: boolean +} + function serializeProviderPayload(payload: unknown): string { try { const serialized = JSON.stringify(payload) @@ -33,6 +38,7 @@ const MESSAGE_KEYS = [ 'text', 'warning', 'detail', + 'summary', 'description', 'reason', // `error` is how a failed dependency reports itself — an MCP server that could not start says @@ -79,14 +85,18 @@ export function unhandledProviderFrameJournalItem( provider: string, kind: string, payload: unknown, - limits: JournalPayloadLimits = DEFAULT_JOURNAL_PAYLOAD_LIMITS + limits: JournalPayloadLimits = DEFAULT_JOURNAL_PAYLOAD_LIMITS, + options: UnhandledProviderFrameJournalItemOptions = {} ): UnhandledProviderFrameJournalItem | null { // A kind a typed translator owns never degrades to its opcode here, in either // direction: "no row" is that translator's decision, not a gap this fallback // has to cover. Checked before classification, because the payload sniffer // inside it promotes a covered frame that reports a failure and would // otherwise print `${provider} · ${kind}` beside the typed row. - if (hasTypedProviderFrameTranslator(provider, kind)) { + if ( + options.coveredByTypedTranslator === true && + hasTypedProviderFrameTranslator(provider, kind) + ) { return null } const classification = classifyProviderFrame(provider, kind, payload) diff --git a/src/main/runtime/orchestration/worker-transcript-activity-block-bounds.ts b/src/main/runtime/orchestration/worker-transcript-activity-block-bounds.ts new file mode 100644 index 00000000000..6e4d682da74 --- /dev/null +++ b/src/main/runtime/orchestration/worker-transcript-activity-block-bounds.ts @@ -0,0 +1,63 @@ +import { + normalizeBackgroundTaskKind, + normalizeBackgroundTaskState +} from '../../../shared/native-chat-background-task-row' +import { normalizeSubagentState } from '../../../shared/native-chat-subagent-summary' +import type { NativeChatBlock, NativeChatSubagentState } from '../../../shared/native-chat-types' + +type WorkerTranscriptActivityBlock = Extract< + NativeChatBlock, + { type: 'subagent-group' | 'background-task' } +> + +export type WorkerTranscriptActivityBlockBounders = { + clipMetadata: (value: string) => string + clipText: (value: string) => string + boundEntryId: (value: string) => string + markClipped: (warning: string) => void +} + +const MAX_WORKER_TRANSCRIPT_SUBAGENTS = 64 + +export function boundWorkerTranscriptActivityBlock( + block: WorkerTranscriptActivityBlock, + bounders: WorkerTranscriptActivityBlockBounders +): NativeChatBlock { + if (block.type === 'subagent-group') { + const agents = block.agents.slice(0, MAX_WORKER_TRANSCRIPT_SUBAGENTS) + if (agents.length < block.agents.length) { + bounders.markClipped('Some subagents were omitted from oversized spawn groups.') + } + return { + ...block, + groupId: bounders.clipMetadata(block.groupId), + agents: agents.map((agent) => ({ + ...agent, + id: bounders.boundEntryId(agent.id), + label: bounders.clipMetadata(agent.label), + state: clipSubagentState(agent.state, bounders) + })) + } + } + const { outputFile, ...carried } = block + if (outputFile) { + bounders.markClipped('Background task output paths were omitted from transcript output.') + } + return { + ...carried, + taskId: bounders.boundEntryId(block.taskId), + kind: normalizeBackgroundTaskKind(bounders.clipMetadata(block.kind)), + label: bounders.clipMetadata(block.label), + state: normalizeBackgroundTaskState(bounders.clipMetadata(block.state)), + ...(block.summary ? { summary: bounders.clipText(block.summary) } : {}), + ...(block.error ? { error: bounders.clipText(block.error) } : {}) + } +} + +function clipSubagentState( + value: NativeChatSubagentState, + bounders: WorkerTranscriptActivityBlockBounders +): NativeChatSubagentState { + const clipped = bounders.clipMetadata(value) + return clipped === value ? value : normalizeSubagentState(clipped) +} diff --git a/src/main/runtime/orchestration/worker-transcript-payload.test.ts b/src/main/runtime/orchestration/worker-transcript-payload.test.ts index aaa3fdcfcc1..c843fe1cfa7 100644 --- a/src/main/runtime/orchestration/worker-transcript-payload.test.ts +++ b/src/main/runtime/orchestration/worker-transcript-payload.test.ts @@ -132,6 +132,37 @@ describe('worker transcript wire bounds', () => { expect(result.limited).toBe(true) }) + it('bounds a background-task kind and state a newer build wrote as open strings', () => { + const result = boundWorkerTranscriptMessages([ + JSON.parse( + JSON.stringify({ + id: 'message-task-state', + role: 'system', + timestamp: null, + source: 'transcript', + blocks: [ + { + type: 'background-task', + taskId: 'task-1', + kind: 'k'.repeat(900), + label: 'l'.repeat(900), + state: 's'.repeat(900) + } + ] + }) + ) + ]) + + const block = result.messages[0]?.blocks[0] + if (block?.type !== 'background-task') { + throw new Error('expected a background-task block') + } + expect(block.kind).toBe('unknown') + expect(block.label).toHaveLength(512) + expect(block.state).toBe('unverifiable') + expect(result.limited).toBe(true) + }) + it('keeps complete bounded messages unlimited', () => { const result = boundWorkerTranscriptMessages([ { diff --git a/src/main/runtime/orchestration/worker-transcript-payload.ts b/src/main/runtime/orchestration/worker-transcript-payload.ts index f53e9375b22..b4365e259c2 100644 --- a/src/main/runtime/orchestration/worker-transcript-payload.ts +++ b/src/main/runtime/orchestration/worker-transcript-payload.ts @@ -1,11 +1,7 @@ import { createHash } from 'node:crypto' -import { normalizeSubagentState } from '../../../shared/native-chat-subagent-summary' -import type { - NativeChatBlock, - NativeChatMessage, - NativeChatSubagentState -} from '../../../shared/native-chat-types' +import type { NativeChatBlock, NativeChatMessage } from '../../../shared/native-chat-types' import { boundSubagentEntryId } from '../../native-chat/subagent-entry-id-bounds' +import { boundWorkerTranscriptActivityBlock } from './worker-transcript-activity-block-bounds' export const DEFAULT_WORKER_TRANSCRIPT_MESSAGE_LIMIT = 40 export const MAX_WORKER_TRANSCRIPT_MESSAGE_LIMIT = 50 @@ -13,10 +9,6 @@ const MAX_WORKER_TRANSCRIPT_BLOCKS = 6 const MAX_WORKER_TRANSCRIPT_BLOCK_CHARS = 1_200 const MAX_WORKER_TRANSCRIPT_INPUT_ITEMS = 20 const MAX_WORKER_TRANSCRIPT_INPUT_NODES = 100 -// Matches the producer's per-group cap, so no group this build writes is clipped -// here. The bound stays because the journal schema declares no maximum and a -// remote host may run a build with a larger one. -const MAX_WORKER_TRANSCRIPT_SUBAGENTS = 64 // Message ids, turn ids, tool-call names and image urls, not only roster fields. // Equal to `MAX_SUBAGENT_FIELD_CHARS` today, kept a separate literal so a // roster-motivated change to that cap cannot silently move this one. @@ -142,40 +134,13 @@ function boundBlock(block: NativeChatBlock, state: TranscriptBoundState): Native input: boundToolInput(block.input, budget, 0, state) } } - if (block.type === 'subagent-group') { - const agents = block.agents.slice(0, MAX_WORKER_TRANSCRIPT_SUBAGENTS) - if (agents.length < block.agents.length) { - markClipped(state, 'Some subagents were omitted from oversized spawn groups.') - } - // Labels, ids and states come from provider-supplied strings, so they get the - // same redaction and clipping every other piece of transcript metadata gets. - return { - ...block, - groupId: clipMetadata(block.groupId, state), - agents: agents.map((agent) => ({ - ...agent, - id: boundEntryId(agent.id, state), - label: clipMetadata(agent.label, state), - state: clipSubagentState(agent.state, state) - })) - } - } - if (block.type === 'background-task') { - // Provider strings get the same redaction and clipping as every other piece - // of transcript metadata. `outputFile` is dropped outright for the reason - // the image branch below drops local paths: it names a file on the - // execution host, which no peer reading this payload can open. - const { outputFile, ...carried } = block - if (outputFile) { - markClipped(state, 'Background task output paths were omitted from transcript output.') - } - return { - ...carried, - taskId: boundEntryId(block.taskId, state), - label: clipMetadata(block.label, state), - ...(block.summary ? { summary: clipText(block.summary, state) } : {}), - ...(block.error ? { error: clipText(block.error, state) } : {}) - } + if (block.type === 'subagent-group' || block.type === 'background-task') { + return boundWorkerTranscriptActivityBlock(block, { + clipMetadata: (value) => clipMetadata(value, state), + clipText: (value) => clipText(value, state), + boundEntryId: (value) => boundEntryId(value, state), + markClipped: (warning) => markClipped(state, warning) + }) } if (block.path || (block.url && isLocalFileLocator(block.url))) { markClipped(state, 'Local image paths were omitted from transcript output.') @@ -233,17 +198,6 @@ function clipMetadata(value: string, state: TranscriptBoundState): string { return redacted.slice(0, MAX_WORKER_TRANSCRIPT_METADATA_CHARS) } -/** `state` is an open string on the wire, so it takes the same bound. A value - * that had to be redacted or clipped names no state any build knows, which is - * exactly what `unverifiable` records. */ -function clipSubagentState( - value: NativeChatSubagentState, - state: TranscriptBoundState -): NativeChatSubagentState { - const clipped = clipMetadata(value, state) - return clipped === value ? value : normalizeSubagentState(clipped) -} - function clipText(value: string, state: TranscriptBoundState): string { const redacted = redactSensitiveText(value, state.warnings) if (redacted.length <= MAX_WORKER_TRANSCRIPT_BLOCK_CHARS) {