diff --git a/src/cli/handlers/orchestration/worker-output.test.ts b/src/cli/handlers/orchestration/worker-output.test.ts index 44da2d67f93..895e9909d05 100644 --- a/src/cli/handlers/orchestration/worker-output.test.ts +++ b/src/cli/handlers/orchestration/worker-output.test.ts @@ -1,5 +1,11 @@ import { describe, expect, it } from 'vitest' import type { OrchestrationFleetWorker } from '../../../shared/orchestration-fleet-projection' +import { subagentGroupFallbackText } from '../../../shared/native-chat-subagent-summary' +import type { + NativeChatBlock, + NativeChatMessage, + NativeChatSubagentEntry +} from '../../../shared/native-chat-types' import type { OrchestrationWorkerReadResult } from '../../../shared/orchestration-worker-output' import { formatWorkerRead, formatWorkerStart } from './worker-output' @@ -288,3 +294,207 @@ function workerReadResult( type WorkerReadResultWithoutContext = T extends unknown ? Omit : never + +function transcriptRead( + blocks: NativeChatBlock[], + role: NativeChatMessage['role'] = 'assistant' +): OrchestrationWorkerReadResult { + const message: NativeChatMessage = { + id: 'm1', + role, + blocks, + timestamp: 1, + source: 'transcript' + } + return { + dispatchId: 'd1', + source: 'transcript', + sourceIdentity: 'pane:1', + provider: 'codex', + transcript: { messages: [message], nextCursor: '1', limited: false, returnedMessageCount: 1 }, + cursor: '1', + status: { worker: 'running', terminal: 'running' }, + fallbackReason: null, + warnings: [] + } +} + +const ROSTER: readonly NativeChatSubagentEntry[] = [ + { id: 'child-1', label: 'read', state: 'working' }, + { id: 'child-2', label: 'edit', state: 'failed' } +] + +function occurrences(haystack: string, needle: string): number { + return haystack.split(needle).length - 1 +} + +describe('formatWorkerRead', () => { + // The replay case this row is durable for: SQLite-backed, re-sent on every + // reconnect, and read here by a client that draws no roster block, runs no + // reconciliation, and cannot re-check whether those children still exist. A + // sentence frozen mid-flight outlives the process that wrote it, so it must + // not keep asserting a liveness only that process could have observed — + // `docs/reference/ssh-execution-boundary.md` calls that loss of contact + // reported as a live state. + it('replays a mid-flight roster row without claiming a child is still working', () => { + const midFlight: readonly NativeChatSubagentEntry[] = [ + { id: 'child-1', label: 'read', state: 'working' }, + { id: 'child-2', label: 'search', state: 'working' }, + { id: 'child-3', label: 'edit', state: 'failed' } + ] + + const output = formatWorkerRead( + transcriptRead([ + { type: 'text', text: subagentGroupFallbackText(midFlight) }, + { type: 'subagent-group', groupId: 'thread:turn-1', agents: [...midFlight] } + ]) + ) + + expect(output).toContain('[assistant] Kicked off 3 subagents (1 failed)') + expect(output).not.toMatch(/\bworking\b/) + }) + + // The body `codexSubagentGroupBody` actually writes: the plain-text twin, then + // the block it stands in for. The twin exists for clients that cannot draw the + // block, so a client printing the block must not print the twin beside it — + // the renderer drops the twin for the same reason, from the other side. + it('prints the roster sentence once for the two-block row the producer writes', () => { + const sentence = subagentGroupFallbackText(ROSTER) + const output = formatWorkerRead( + transcriptRead( + [ + { type: 'text', text: sentence }, + { type: 'subagent-group', groupId: 'thread:turn-1', agents: [...ROSTER] } + ], + 'system' + ) + ) + + expect(output).toContain(`[system] ${sentence}`) + expect(occurrences(output, sentence)).toBe(1) + }) + + // Suppression is per twin, not per message. One twin beside two roster blocks + // silenced BOTH groups and printed one sentence, so the second roster vanished + // with no marker — the same silent drop the missing-twin case above avoids. + it('stands in for the second roster block when only one twin accompanies two', () => { + const other: readonly NativeChatSubagentEntry[] = [ + { id: 'child-3', label: 'plan', state: 'completed' } + ] + const output = formatWorkerRead( + transcriptRead( + [ + { type: 'text', text: subagentGroupFallbackText(ROSTER) }, + { type: 'subagent-group', groupId: 'thread:turn-1', agents: [...ROSTER] }, + { type: 'subagent-group', groupId: 'thread:turn-2', agents: [...other] } + ], + 'system' + ) + ) + + expect(occurrences(output, subagentGroupFallbackText(ROSTER))).toBe(1) + expect(output).toContain(`[subagents] ${subagentGroupFallbackText(other)}`) + }) + + // Which group a lone twin belongs to is decided by its TEXT, not its position. + // Claiming positionally silenced whichever group came first, so a twin + // belonging to a LATER group erased the earlier group's roster and printed the + // later one's sentence twice — the same silent drop, one permutation over. + it('claims a lone twin for the group it names, not the first group in the message', () => { + const other: readonly NativeChatSubagentEntry[] = [ + { id: 'child-3', label: 'plan', state: 'completed' } + ] + const second = subagentGroupFallbackText(other) + const output = formatWorkerRead( + transcriptRead( + [ + { type: 'text', text: second }, + { type: 'subagent-group', groupId: 'thread:turn-1', agents: [...ROSTER] }, + { type: 'subagent-group', groupId: 'thread:turn-2', agents: [...other] } + ], + 'system' + ) + ) + + expect(occurrences(output, second)).toBe(1) + expect(output).toContain(`[subagents] ${subagentGroupFallbackText(ROSTER)}`) + }) + + // The same claim, with the twin written after both blocks: nothing about the + // ORDER of a twin and its group is guaranteed by the block schema. + it('claims a trailing twin for the group it names', () => { + const other: readonly NativeChatSubagentEntry[] = [ + { id: 'child-3', label: 'plan', state: 'completed' } + ] + const second = subagentGroupFallbackText(other) + const output = formatWorkerRead( + transcriptRead( + [ + { type: 'subagent-group', groupId: 'thread:turn-1', agents: [...ROSTER] }, + { type: 'subagent-group', groupId: 'thread:turn-2', agents: [...other] }, + { type: 'text', text: second } + ], + 'system' + ) + ) + + expect(occurrences(output, second)).toBe(1) + expect(output).toContain(`[subagents] ${subagentGroupFallbackText(ROSTER)}`) + }) + + // A group with no twin beside it is a shape the block schema admits and no + // producer writes. Dropping it would lose the roster entirely, so the block + // itself carries the sentence when nothing else does. + it('stands in for a roster block that arrived without its twin', () => { + const output = formatWorkerRead( + transcriptRead([{ type: 'subagent-group', groupId: 'thread:turn-1', agents: [...ROSTER] }]) + ) + + expect(output).toContain(`[assistant] [subagents] ${subagentGroupFallbackText(ROSTER)}`) + }) + + // A roster from a newer build holds a state this build does not know, which + // `summarizeSubagentGroup` reads as `unverifiable`. Recomputing the sentence + // to compare it against the frozen twin therefore produced a DIFFERENT string, + // and the CLI printed the roster twice: the twin's own wording plus a + // `[subagents]` line contradicting it. + it('prints the roster once when the twin names a state this build cannot reproduce', () => { + const frozenTwin = 'Ran 2 subagents (1 cancelled)' + const output = formatWorkerRead( + transcriptRead( + [ + { type: 'text', text: frozenTwin }, + { + type: 'subagent-group', + groupId: 'thread:turn-1', + agents: [ + { id: 'child-1', label: 'read', state: 'completed' }, + { id: 'child-2', label: 'edit', state: 'cancelled' } + ] as unknown as NativeChatSubagentEntry[] + } + ], + 'system' + ) + ) + + expect(output).toContain(`[system] ${frozenTwin}`) + expect(output).not.toContain('[subagents]') + expect(output).not.toContain('unverifiable') + }) + + // The journal admits block types this build does not know, and `client.call` + // casts the RPC result rather than validating it — so a newer remote host's + // block reaches this formatter as-is. Reading fields off it threw a TypeError + // and took down the whole `worker read`. + it('degrades an unknown block type from a newer host instead of throwing', () => { + const output = formatWorkerRead( + transcriptRead([ + { type: 'text', text: 'before' }, + { type: 'plan-step', title: 'ship it' } as unknown as NativeChatBlock, + { type: 'text', text: 'after' } + ]) + ) + + expect(output).toContain('[assistant] before\n[unsupported block]\nafter') + }) +}) diff --git a/src/main/codex/codex-structured-item-translation.test.ts b/src/main/codex/codex-structured-item-translation.test.ts index 2558f4b60de..0c47919f59d 100644 --- a/src/main/codex/codex-structured-item-translation.test.ts +++ b/src/main/codex/codex-structured-item-translation.test.ts @@ -784,7 +784,7 @@ describe('codex item bodies', () => { } }) - it('leaves subagent items on the generic row until a real renderer exists', () => { + it('drops the raw subagent item now the roster row renders it', () => { expect( codexJournalItem({ type: 'subAgentActivity', @@ -793,10 +793,7 @@ describe('codex item bodies', () => { agentThreadId: 'thread-child', agentPath: '/root/list_directory' }) - ).toMatchObject({ - handled: false, - body: { kind: 'status', providerFrame: { kind: 'item:subAgentActivity' } } - }) + ).toMatchObject({ handled: true, body: null }) }) it('drops the sleep item, which codex itself renders as nothing', () => { diff --git a/src/main/codex/codex-structured-journal-limits.ts b/src/main/codex/codex-structured-journal-limits.ts index d741a9e86d2..5137ea8dd16 100644 --- a/src/main/codex/codex-structured-journal-limits.ts +++ b/src/main/codex/codex-structured-journal-limits.ts @@ -7,3 +7,10 @@ export const MAX_CODEX_PENDING_PROMPTS = 128 export const MAX_CODEX_IDENTITY_ENTRIES = 512 export const MAX_CODEX_DETAIL_ENTRIES = 512 export const MAX_CODEX_DETAIL_BYTES = 64 * 1024 +/** Spawn-group rows kept live per session, and children per row. Both bound an + * event-accumulated map that no provider snapshot ever prunes. */ +export const MAX_CODEX_SUBAGENT_GROUPS = 32 +export const MAX_CODEX_SUBAGENTS_PER_GROUP = 64 +/** Threads whose latest token total is retained. Usage frames arrive for + * threads that are not yet (or never become) roster children. */ +export const MAX_CODEX_TOKEN_USAGE_THREADS = 256 diff --git a/src/main/codex/codex-structured-journal-translation-frames.ts b/src/main/codex/codex-structured-journal-translation-frames.ts new file mode 100644 index 00000000000..22dc516b210 --- /dev/null +++ b/src/main/codex/codex-structured-journal-translation-frames.ts @@ -0,0 +1,43 @@ +/** + * The translator's provider-frame arms. + * + * Each returns null for a frame it does not own, which is the translator's + * signal to keep looking. Split out so the translator reads as routing rather + * than as the shape checks each arm performs. + */ + +import type { CodexJournalTranslationAdmission } from './codex-structured-journal-contracts' +import { settleCodexOversizedNotification } from './codex-structured-journal-settlement' +import { + readCodexJournalRecord, + readCodexJournalString +} from './codex-structured-journal-translation-values' + +type OversizedInput = Parameters[0] + +/** A notification the transport refused to carry whole: settle whatever it + * opened rather than leaving the item mid-flight. */ +export function settleCodexOversizedNotificationFrame(input: { + sessionId: string + threadId: string + kind: string + payload: unknown + sink: OversizedInput['sink'] + streams: OversizedInput['streams'] + activeItems: OversizedInput['activeItems'] +}): CodexJournalTranslationAdmission | null { + if (input.kind !== 'frame:oversized-notification') { + return null + } + const method = readCodexJournalString(readCodexJournalRecord(input.payload), 'method') + return method + ? settleCodexOversizedNotification({ + sessionId: input.sessionId, + threadId: input.threadId, + method, + sink: input.sink, + streams: input.streams, + activeItems: input.activeItems + }) + : null +} diff --git a/src/main/codex/codex-structured-journal-translation-subagents.test.ts b/src/main/codex/codex-structured-journal-translation-subagents.test.ts new file mode 100644 index 00000000000..bf5cdffa5a9 --- /dev/null +++ b/src/main/codex/codex-structured-journal-translation-subagents.test.ts @@ -0,0 +1,168 @@ +import { describe, expect, it } from 'vitest' +import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key' +import type { AgentSessionTurnActivity } from '../../shared/agent-session-wire' +import type { + AgentJournalItemBody, + AgentJournalItemIdentity +} from '../../shared/agent-session-journal-types' +import { isSubagentGroupBlock } from '../../shared/native-chat-types' +import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import { createCodexJournalTranslator } from './codex-structured-journal-translation' +import type { CodexStructuredSessionEvent } from './codex-structured-session-adapter' + +const SESSION_ID = 'session-1' +const THREAD_ID = 'thread-abc' +const TURN_ID = 'turn-1' + +type Row = { key: string; body: AgentJournalItemBody } + +function harness() { + const rows: Row[] = [] + const activities: (AgentSessionTurnActivity | null)[] = [] + const sink: StructuredAgentSessionEventSink = { + appendItem: (identity: AgentJournalItemIdentity, body) => + rows.push({ key: agentJournalItemKey(identity), body }), + appendTombstone: () => {}, + publish: () => {}, + setActivity: (activity) => activities.push(activity) + } + const translator = createCodexJournalTranslator({ + sink, + primaryThreadId: () => THREAD_ID, + schedule: (run: () => void) => { + run() + return () => {} + } + }) + return { translator, rows, activities } +} + +function notification(method: string, params: unknown): CodexStructuredSessionEvent { + return { type: 'notification', sessionId: SESSION_ID, threadId: THREAD_ID, method, params } +} + +function subagentItem(kind: string, agentThreadId: string, agentPath: string): unknown { + return { + turnId: TURN_ID, + item: { + type: 'subAgentActivity', + id: `item-${agentThreadId}-${kind}`, + kind, + agentThreadId, + agentPath + } + } +} + +/** Every activity item reaches the wire twice. */ +function deliverActivity( + translator: ReturnType, + params: unknown +): void { + translator.handle(notification('item/started', params)) + translator.handle(notification('item/completed', params)) +} + +function rosterAgents(rows: Row[]): { id: string; state: string; tokens?: number }[] { + const body = rows.findLast((row) => row.key.startsWith('orca:codex-subagents'))?.body + if (!body || body.kind !== 'message') { + return [] + } + return body.blocks.find(isSubagentGroupBlock)?.agents ?? [] +} + +describe('codex journal translation — subagents', () => { + it('renders a spawn group as one roster row and no opcode-shaped duplicate', () => { + const { translator, rows } = harness() + + translator.handle(notification('turn/started', { turn: { id: TURN_ID } })) + deliverActivity(translator, subagentItem('started', 'child-1', '/root/list_directory')) + deliverActivity(translator, subagentItem('interacted', 'child-1', '/root/list_directory')) + + expect(rosterAgents(rows)).toMatchObject([ + { id: 'child-1', label: 'list_directory', state: 'working' } + ]) + // Four wire deliveries (two items, each sent twice) collapse to ONE roster + // row, and none of the gray `codex · item:subAgentActivity` rows survive. + const providerFrameKinds = rows.flatMap((row) => + row.body.kind === 'status' && row.body.providerFrame ? [row.body.providerFrame.kind] : [] + ) + expect(providerFrameKinds).toEqual([]) + expect(rows.filter((row) => row.key.startsWith('orca:codex-subagents'))).toHaveLength(1) + }) + + // The roster claims the item, but claiming it must not take the turn tail with + // it: the activity table is reached only through the publish arm, so a bare + // return leaves the tail stuck on whatever the previous frame said. + it('still publishes the turn tail for an item the roster claims', () => { + const { translator, activities } = harness() + + translator.handle(notification('turn/started', { turn: { id: TURN_ID } })) + activities.length = 0 + deliverActivity(translator, subagentItem('started', 'child-1', '/root/read')) + + expect(activities.at(-1)).toEqual({ + turnId: TURN_ID, + text: 'Coordinating with another agent' + }) + }) + + it('consumes thread/tokenUsage/updated instead of swallowing it as chrome', () => { + const { translator, rows } = harness() + + translator.handle(notification('turn/started', { turn: { id: TURN_ID } })) + deliverActivity(translator, subagentItem('started', 'child-1', '/root/read')) + translator.handle( + notification('thread/tokenUsage/updated', { + threadId: 'child-1', + tokenUsage: { total: { totalTokens: 40661 } } + }) + ) + + expect(rosterAgents(rows)).toMatchObject([{ id: 'child-1', tokens: 40661 }]) + }) + + // The QA scenario this row got wrong: three `spawn_agent` children were still + // running when a mid-turn correction ended their turn and opened a new one. + // They reported `completed` 57-87s later, so a turn boundary is a fact about + // the turn and never evidence that contact with a child was lost. + it('leaves children working when their turn ends and a newer turn opens', () => { + const { translator, rows } = harness() + + translator.handle(notification('turn/started', { turn: { id: TURN_ID } })) + deliverActivity(translator, subagentItem('started', 'child-1', '/root/read_readme')) + deliverActivity(translator, subagentItem('started', 'child-2', '/root/read_package')) + translator.handle(notification('turn/completed', { turn: { id: TURN_ID } })) + translator.handle(notification('turn/started', { turn: { id: 'turn-2' } })) + + expect(rosterAgents(rows)).toMatchObject([ + { id: 'child-1', state: 'working' }, + { id: 'child-2', state: 'working' } + ]) + + // And the verdict a child reports after its turn ended still lands on the row. + deliverActivity(translator, subagentItem('completed', 'child-1', '/root/read_readme')) + + expect(rosterAgents(rows)).toMatchObject([ + { id: 'child-1', state: 'completed' }, + { id: 'child-2', state: 'working' } + ]) + }) + + it('sweeps every group when the provider ends', () => { + const { translator, rows } = harness() + + translator.handle(notification('turn/started', { turn: { id: TURN_ID } })) + deliverActivity(translator, subagentItem('started', 'child-1', '/root/read')) + translator.handle({ + type: 'ended', + sessionId: SESSION_ID, + reason: 'provider exited', + cause: 'unexpected-exit', + fence: 1, + acquisitionGeneration: 'gen-1' + } as CodexStructuredSessionEvent) + + expect(rosterAgents(rows)).toMatchObject([{ id: 'child-1', state: 'unverifiable' }]) + }) +}) diff --git a/src/main/codex/codex-structured-journal-translation.ts b/src/main/codex/codex-structured-journal-translation.ts index c0c103bddff..c8a6fe9f158 100644 --- a/src/main/codex/codex-structured-journal-translation.ts +++ b/src/main/codex/codex-structured-journal-translation.ts @@ -1,4 +1,10 @@ import { createCodexProviderActivityReader } from '../native-chat/agent-session-wire/provider-frame-activity' +import { + CODEX_TOKEN_USAGE_METHOD, + readCodexNotificationThreadItem +} from './codex-subagent-activity' +import { CodexSubagentRoster } from './codex-subagent-roster' +import { readCodexThreadItem } from './codex-structured-item-translation' import { CodexJournalGenericFrames } from './codex-structured-journal-generic-frames' import { CodexJournalItems } from './codex-structured-journal-items' import { CodexJournalPrompts } from './codex-structured-journal-prompts' @@ -10,16 +16,12 @@ import { } from './codex-structured-journal-contracts' import { settleCodexJournalSession, - settleCodexJournalTurn, - settleCodexOversizedNotification + settleCodexJournalTurn } from './codex-structured-journal-settlement' +import { settleCodexOversizedNotificationFrame } from './codex-structured-journal-translation-frames' import { restoreCodexJournalThread } from './codex-structured-journal-translation-restore' import { CodexJournalActiveTurns } from './codex-structured-journal-translation-turn-state' import { publishCodexTurnLifecycle } from './codex-structured-journal-translation-turns' -import { - readCodexJournalRecord, - readCodexJournalString -} from './codex-structured-journal-translation-values' import { readCodexTurnId } from './codex-structured-thread-facts' import type { CodexStructuredSessionEvent } from './codex-structured-session-adapter' @@ -55,6 +57,11 @@ export function createCodexJournalTranslator( const prompts = new CodexJournalPrompts(deps, (threadId, itemId) => items.detailFor(threadId, itemId) ) + const subagents = new CodexSubagentRoster({ + sink: deps.sink, + primaryThreadId: () => deps.primaryThreadId?.() ?? null, + activeTurn: (threadId) => activeTurns.current(threadId) + }) const flushStreams = (): CodexJournalTranslationAdmission => items.streams.flush() ? CODEX_JOURNAL_ADMITTED : { accepted: false, reason: 'backpressure' } let readActivity = createCodexProviderActivityReader() @@ -118,6 +125,11 @@ export function createCodexJournalTranslator( if (!admission.accepted) { return admission } + // No event will ever settle a child once the provider is gone. + const sweep = subagents.settleSession() + if (!sweep.accepted) { + return sweep + } readActivity = createCodexProviderActivityReader() deps.sink.setActivity?.(null) items.activeItems.clear() @@ -159,7 +171,30 @@ export function createCodexJournalTranslator( if (event.method === 'turn/completed') { return completeTurn(event) } + if (event.method === CODEX_TOKEN_USAGE_METHOD) { + // Classified `status-chrome`, so the generic-frame path swallows it + // before the journal. The roster consumes it as a typed notification. + const admission = subagents.handleTokenUsage(event.params) + if (admission) { + return admission + } + } if (event.method === 'item/started' || event.method === 'item/completed') { + const subagentItem = readCodexNotificationThreadItem(event.params, readCodexThreadItem) + // Null means the roster did not claim it; fall through to normal item + // handling. Returning here unconditionally swallows every other item. + const subagentAdmission = subagentItem + ? subagents.handleItem({ + threadId: event.threadId, + turnId: readCodexTurnId(event.params) ?? activeTurns.current(event.threadId), + item: subagentItem + }) + : null + if (subagentAdmission) { + // Not a bare return: the roster claiming the item must not skip the + // turn-tail arm, which is the only publisher of its activity copy. + return publishActivity(event, subagentAdmission) + } const translated = items.handle(event) return publishActivity( event, @@ -186,30 +221,25 @@ export function createCodexJournalTranslator( items.dispose() prompts.dispose() genericFrames.dispose() + subagents.dispose() activeTurns.clear() } } + /** Settles the item a notification the transport refused to carry left + * mid-flight; null when the frame is not one. */ function settleOversizedNotification(event: { sessionId: string threadId: string kind: string payload: unknown }): CodexJournalTranslationAdmission | null { - if (event.kind !== 'frame:oversized-notification') { - return null - } - const method = readCodexJournalString(readCodexJournalRecord(event.payload), 'method') - return method - ? settleCodexOversizedNotification({ - sessionId: event.sessionId, - threadId: event.threadId, - method, - sink: deps.sink, - streams: items.streams, - activeItems: items.activeItems - }) - : null + return settleCodexOversizedNotificationFrame({ + ...event, + sink: deps.sink, + streams: items.streams, + activeItems: items.activeItems + }) } function startTurn(event: { @@ -255,6 +285,12 @@ export function createCodexJournalTranslator( if (!turnId) { return CODEX_JOURNAL_ADMITTED } + // The roster is deliberately NOT swept here. `spawn_agent` children outlive + // the turn that spawned them and go on reporting into the same group, so a + // turn boundary is no evidence contact was lost — and `turn/completed` is + // the only turn-end notification Codex sends, so an abort cannot be told + // apart from a clean finish either. Only `settleSession` may write + // `unverifiable`. const admission = settleCodexJournalTurn({ sink: deps.sink, sessionId: event.sessionId, diff --git a/src/main/codex/codex-subagent-activity.ts b/src/main/codex/codex-subagent-activity.ts new file mode 100644 index 00000000000..f12e9dfb1b3 --- /dev/null +++ b/src/main/codex/codex-subagent-activity.ts @@ -0,0 +1,140 @@ +// Reading Codex's subagent wire shapes. +// +// Established by a live probe against `codex app-server` 0.152.1, not inferred: +// * `subAgentActivity` items carry `{kind, agentThreadId, agentPath}`, and each +// one arrives TWICE — via `item/started` and again via `item/completed`. +// * `agentPath` is a tree path (`/root`, `/root/list_directory`); the trailing +// segment is a semantic task name and the only label available. There is no +// `thread/started` for a child, so nickname/role/depth do not exist. +// * `agentsStates` on `collabAgentToolCall` arrived empty (`{}`) throughout the +// probe, so nothing here reads it — state comes from `kind` alone. +// * `thread/tokenUsage/updated` reports a per-thread RUNNING TOTAL, so the +// latest frame replaces the previous one — it is never accumulated. + +import type { NativeChatSubagentState } from '../../shared/native-chat-types' +import type { CodexThreadItem } from './codex-structured-item-translation' + +export const CODEX_SUBAGENT_ITEM_TYPE = 'subAgentActivity' +export const CODEX_TOKEN_USAGE_METHOD = 'thread/tokenUsage/updated' + +export type CodexSubagentActivity = { + kind: string + agentThreadId: string + agentPath: string | null +} + +function nonEmptyString(value: unknown): string | null { + return typeof value === 'string' && value.length > 0 ? value : null +} + +function record(value: unknown): Record | null { + return typeof value === 'object' && value !== null && !Array.isArray(value) + ? (value as Record) + : null +} + +export function readCodexSubagentActivity(item: CodexThreadItem): CodexSubagentActivity | null { + if (item.type !== CODEX_SUBAGENT_ITEM_TYPE) { + return null + } + const agentThreadId = nonEmptyString(item.agentThreadId) + if (!agentThreadId) { + return null + } + return { + kind: nonEmptyString(item.kind) ?? '', + agentThreadId, + agentPath: nonEmptyString(item.agentPath) + } +} + +/** + * The state a `kind` implies for the child it names. + * + * An unrecognized kind means "this child exists and reported something we + * cannot classify" — `working`, which the session sweep will later settle to + * `unverifiable` if nothing better ever arrives. Claiming a terminal state from + * an unknown kind would assert an outcome the wire never gave us. + */ +export function codexSubagentStateForKind(kind: string): NativeChatSubagentState { + if (kind === 'completed') { + return 'completed' + } + if (kind === 'interrupted') { + return 'stopped' + } + return 'working' +} + +/** Path segments, empty ones dropped: `/root/list_directory` → 2 segments. */ +export function codexSubagentPathSegments(agentPath: string | null): string[] { + return agentPath === null ? [] : agentPath.split('/').filter((part) => part.length > 0) +} + +/** The one path segment that names the parent turn itself rather than a child. + * Compared after the same normalization the label uses, not against the raw + * string: `/root/` and `/root//` are the same node as `/root`, and a check that + * disagreed with `codexSubagentPathSegments` would let one path be both the + * turn and a child of it — a phantom row labelled `root` inflating the group. + * Only this segment is the root; `/morpheus` is single-segment too but IS a + * child. */ +const CODEX_ROOT_AGENT_SEGMENT = 'root' + +/** + * Whether an activity item describes the ROOT of the agent tree rather than a + * spawned child. Counting the root would make the parent turn report itself as + * its own subagent. + * + * A path-less item cannot be placed in the tree at all, so it is treated as a + * child: dropping it would lose a real spawn, while an extra row is visible and + * self-correcting. + */ +export function isCodexRootAgentActivity(activity: CodexSubagentActivity): boolean { + const segments = codexSubagentPathSegments(activity.agentPath) + return segments.length === 1 && segments[0] === CODEX_ROOT_AGENT_SEGMENT +} + +/** Row label: the agent path's trailing segment, trimmed. A segment with nothing + * visible in it survives the empty-segment filter but would draw a nameless row, + * so it reads as no label and the caller's placeholder takes over. Trimmed + * because the caller keys its collision ordinals on this string: ` read ` and + * `read` render identically and must therefore collide. */ +export function codexSubagentLabel(activity: CodexSubagentActivity): string | null { + const trailing = codexSubagentPathSegments(activity.agentPath).at(-1)?.trim() + return trailing !== undefined && trailing.length > 0 ? trailing : null +} + +export type CodexThreadTokenTotal = { threadId: string; totalTokens: number } + +/** `{threadId, tokenUsage: {total: {totalTokens}}}`. Older builds put the total + * on the envelope, so both shapes are accepted. */ +export function readCodexThreadTokenTotal(params: unknown): CodexThreadTokenTotal | null { + const root = record(params) + if (!root) { + return null + } + const threadId = nonEmptyString(root.threadId) ?? nonEmptyString(record(root.thread)?.id) + if (!threadId) { + return null + } + const usage = record(root.tokenUsage) + const total = record(usage?.total)?.totalTokens ?? usage?.totalTokens ?? root.totalTokens + return typeof total === 'number' && Number.isFinite(total) && total >= 0 + ? { threadId, totalTokens: total } + : null +} + +/** Pull the `subAgentActivity` item out of a raw notification payload. + * + * Lives beside the readers rather than in the translator: the translator's job + * is routing, and this is the shape check that decides whether a frame is one + * of ours at all. Returns null for anything that is not a thread item, which is + * the translator's signal to keep looking. */ +export function readCodexNotificationThreadItem( + params: unknown, + read: (value: unknown) => CodexThreadItem | null +): CodexThreadItem | null { + const record = + typeof params === 'object' && params !== null ? (params as Record) : {} + return read(record.item) +} diff --git a/src/main/codex/codex-subagent-roster.test.ts b/src/main/codex/codex-subagent-roster.test.ts new file mode 100644 index 00000000000..2f20c9df9bf --- /dev/null +++ b/src/main/codex/codex-subagent-roster.test.ts @@ -0,0 +1,769 @@ +import { describe, expect, it } from 'vitest' +import { isAdmissibleAgentJournalItemBody } from '../../shared/agent-session-journal-schemas' +import type { + AgentJournalItemBody, + AgentJournalItemIdentity +} from '../../shared/agent-session-journal-types' +import { MAX_SUBAGENT_FIELD_CHARS } from '../../shared/native-chat-subagent-summary' +import { isSubagentGroupBlock, type NativeChatSubagentEntry } from '../../shared/native-chat-types' +import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import { + CodexSubagentRoster, + codexSubagentGroupIdentity, + codexSubagentGroupId +} from './codex-subagent-roster' +import type { CodexThreadItem } from './codex-structured-item-translation' +import { + MAX_CODEX_SUBAGENT_GROUPS, + MAX_CODEX_SUBAGENTS_PER_GROUP, + MAX_CODEX_TOKEN_USAGE_THREADS +} from './codex-structured-journal-limits' + +const THREAD = 'thread-parent' +const TURN = 'turn-1' + +type Appended = { identity: AgentJournalItemIdentity; body: AgentJournalItemBody } + +function createHarness(options: { threadId?: string | null } = {}): { + roster: CodexSubagentRoster + appended: Appended[] + agents: () => NativeChatSubagentEntry[] + latest: () => Appended | undefined +} { + const appended: Appended[] = [] + let clock = 1_000 + const sink: StructuredAgentSessionEventSink = { + appendItem: () => {}, + appendTombstone: () => {}, + publish: () => {}, + tryAppendItem: (identity, body) => { + appended.push({ identity, body }) + return { accepted: true } + }, + tryPublish: () => ({ accepted: true }) + } + const roster = new CodexSubagentRoster({ + sink, + primaryThreadId: () => (options.threadId === undefined ? THREAD : options.threadId), + activeTurn: () => TURN, + now: () => (clock += 1) + }) + const agents = (): NativeChatSubagentEntry[] => { + const body = appended.at(-1)?.body + if (!body || body.kind !== 'message') { + return [] + } + const block = body.blocks.find(isSubagentGroupBlock) + return block ? block.agents : [] + } + return { roster, appended, agents, latest: () => appended.at(-1) } +} + +function latestIdentity(appended: Appended[]): AgentJournalItemIdentity | undefined { + return appended.at(-1)?.identity +} + +function activity(input: { + id?: string + kind: string + agentThreadId: string + agentPath: string | null +}): CodexThreadItem { + return { + type: 'subAgentActivity', + id: input.id ?? `item-${input.agentThreadId}-${input.kind}`, + kind: input.kind, + agentThreadId: input.agentThreadId, + agentPath: input.agentPath + } +} + +function deliver( + roster: CodexSubagentRoster, + item: CodexThreadItem, + turnId: string | null = TURN +): void { + // Every activity item reaches the wire twice: item/started, then item/completed. + roster.handleItem({ threadId: THREAD, turnId, item }) + roster.handleItem({ threadId: THREAD, turnId, item }) +} + +/** + * A sink that coalesces the way the real queue does: by `coalescingKey` ALONE, + * with no op-kind check, and only draining when released. A fake that ignores + * the key cannot see an append being spliced out by its own publish. + */ +function createCoalescingHarness(): { + roster: CodexSubagentRoster + appended: Appended[] + drain: () => void +} { + const appended: Appended[] = [] + const queue: { key?: string; run: () => void }[] = [] + let clock = 1_000 + const submit = (key: string | undefined, run: () => void): void => { + const at = key === undefined ? -1 : queue.findIndex((queued) => queued.key === key) + if (at >= 0) { + queue.splice(at, 1) + } + queue.push(key === undefined ? { run } : { key, run }) + } + const sink: StructuredAgentSessionEventSink = { + appendItem: () => {}, + appendTombstone: () => {}, + publish: () => {}, + tryAppendItem: (identity, body, options) => { + submit(options?.coalescingKey, () => appended.push({ identity, body })) + return { accepted: true } + }, + tryPublish: (options) => { + submit(options?.coalescingKey ?? 'publish', () => {}) + return { accepted: true } + } + } + const roster = new CodexSubagentRoster({ + sink, + primaryThreadId: () => THREAD, + activeTurn: () => TURN, + now: () => (clock += 1) + }) + return { + roster, + appended, + drain: () => { + while (queue.length > 0) { + queue.shift()?.run() + } + } + } +} + +describe('CodexSubagentRoster', () => { + it('does not let its own publish evict the still-queued roster append', () => { + const { roster, appended, drain } = createCoalescingHarness() + + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + drain() + + // Sharing the append's coalescing key with the publish spliced the append + // out of the queue, and `lastSerialized` then suppressed every retry. + expect(appended).toHaveLength(1) + }) + + it('counts a /morpheus agent as a child — only /root is the turn itself', () => { + const { roster, agents } = createHarness() + + deliver(roster, activity({ kind: 'started', agentThreadId: 'child-m', agentPath: '/morpheus' })) + + expect(agents()).toMatchObject([{ id: 'child-m', label: 'morpheus', state: 'working' }]) + }) + + // `codexSubagentPathSegments` already defines what a path means for the label, + // and the root check has to agree with it: a path that normalizes to the same + // node must classify the same way, or one string is both the turn itself and a + // child of it — a phantom row labelled `root` inflating the group by one. + it('reads a root path with a trailing or doubled separator as the turn itself', () => { + for (const agentPath of ['/root/', '/root//', '//root']) { + const { roster, appended } = createHarness() + + deliver(roster, activity({ kind: 'started', agentThreadId: THREAD, agentPath })) + + expect(appended).toEqual([]) + } + }) + + it('keeps a doubled separator inside a child path off the label', () => { + const { roster, agents } = createHarness() + + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root//read/' }) + ) + + expect(agents()).toMatchObject([{ id: 'child-1', label: 'read' }]) + }) + + // An all-whitespace trailing segment survives the empty-segment filter and + // would draw a row with no visible name at all. + it('falls back to the placeholder when the trailing segment has nothing to show', () => { + const { roster, agents } = createHarness() + + deliver(roster, activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/ ' })) + + expect(agents()).toMatchObject([{ id: 'child-1', label: 'subagent' }]) + }) + + // The collision ordinal keys on the label, so two segments that render + // identically must collide rather than both draw as `read`. + it('collides labels that differ only in surrounding whitespace', () => { + const { roster, agents } = createHarness() + + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-2', agentPath: '/root/ read ' }) + ) + + expect(agents().map((agent) => agent.label)).toEqual(['read', 'read 2']) + }) + + it('ignores the root node so a turn is not its own subagent', () => { + const { roster, appended } = createHarness() + + deliver(roster, activity({ kind: 'started', agentThreadId: THREAD, agentPath: '/root' })) + + expect(appended).toEqual([]) + }) + + it('writes an admissible journal body carrying a plain-text fallback block', () => { + const { roster, latest } = createHarness() + + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/list_directory' }) + ) + + const body = latest()?.body + expect(body?.kind).toBe('message') + expect(isAdmissibleAgentJournalItemBody(body)).toBe(true) + expect(body?.kind === 'message' ? body.blocks.map((block) => block.type) : []).toEqual([ + 'text', + 'subagent-group' + ]) + expect( + body?.kind === 'message' && body.blocks[0]?.type === 'text' ? body.blocks[0].text : '' + ).toBe('Kicked off 1 subagent') + }) + + it('keys the durable identity by the parent turn so a revision lands on one row', () => { + const { roster, appended } = createHarness() + + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + deliver( + roster, + activity({ kind: 'completed', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + + const expected = codexSubagentGroupIdentity(codexSubagentGroupId(THREAD, TURN)) + expect(new Set(appended.map((entry) => JSON.stringify(entry.identity)))).toEqual( + new Set([JSON.stringify(expected)]) + ) + }) + + it('rule 1 — a duplicate delivery writes no second revision', () => { + const { roster, appended } = createHarness() + + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + + expect(appended).toHaveLength(1) + }) + + it('rule 2 — a first event of any kind creates the entry in the state it implies', () => { + const { roster, agents } = createHarness() + + deliver( + roster, + activity({ kind: 'completed', agentThreadId: 'child-late', agentPath: '/root/search' }) + ) + + expect(agents()).toMatchObject([{ id: 'child-late', label: 'search', state: 'completed' }]) + }) + + it('rule 3 — a terminal state latches against a late or duplicate start', () => { + const { roster, agents } = createHarness() + + deliver( + roster, + activity({ kind: 'completed', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + deliver( + roster, + activity({ kind: 'interacted', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + + expect(agents()).toMatchObject([{ state: 'completed' }]) + }) + + it('rule 4 — the session sweep settles a lost child as unverifiable, not exited', () => { + const { roster, agents } = createHarness() + + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + deliver( + roster, + activity({ kind: 'completed', agentThreadId: 'child-2', agentPath: '/root/search' }) + ) + roster.settleSession() + + expect(agents()).toMatchObject([ + { id: 'child-1', state: 'unverifiable' }, + { id: 'child-2', state: 'completed' } + ]) + }) + + it('lets a swept child still report what it actually did', () => { + const { roster, agents } = createHarness() + + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + roster.settleSession() + expect(agents()[0]?.state).toBe('unverifiable') + + // Contact can return — a reconnected provider replays the child's own + // verdict. Latching the sweep would report a child that finished as one we + // never saw finish. + deliver( + roster, + activity({ kind: 'completed', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + expect(agents()[0]?.state).toBe('completed') + }) + + it('refuses to put a swept child back to working', () => { + const { roster, agents } = createHarness() + + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + roster.settleSession() + // A straggler progress tick after we gave up must not re-light the row. + deliver( + roster, + activity({ kind: 'interacted', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + expect(agents()[0]?.state).toBe('unverifiable') + }) + + it('keeps a real verdict when a later frame disagrees', () => { + const { roster, agents } = createHarness() + + deliver( + roster, + activity({ kind: 'completed', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + deliver( + roster, + activity({ kind: 'interrupted', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + expect(agents()[0]?.state).toBe('completed') + }) + + it('rule 4 — the session sweep settles every group and never un-terminals one', () => { + const { roster, agents, appended } = createHarness() + + deliver( + roster, + activity({ kind: 'interacted', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + roster.settleSession() + const afterFirstSweep = appended.length + roster.settleSession() + + expect(agents()).toMatchObject([{ state: 'unverifiable' }]) + expect(appended).toHaveLength(afterFirstSweep) + }) + + it('rule 5 — the whole roster is persisted in the carrier, not just a count', () => { + const { roster, agents } = createHarness() + + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + roster.handleTokenUsage({ threadId: 'child-1', tokenUsage: { total: { totalTokens: 40661 } } }) + + expect(agents()).toMatchObject([ + { id: 'child-1', label: 'read', state: 'working', tokens: 40661 } + ]) + }) + + it('rule 6 — the group id names the parent turn, or says there was none', () => { + const { roster, appended } = createHarness() + + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-2', agentPath: '/root/search' }), + null + ) + + expect(appended.map((entry) => entry.identity)).toEqual([ + { provider: 'orca', clientMessageId: `codex-subagents:${THREAD}:${TURN}` }, + { provider: 'orca', clientMessageId: `codex-subagents:${THREAD}:outside-turn` } + ]) + }) + + it('disambiguates two children that share a trailing path segment', () => { + const { roster, agents } = createHarness() + + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-2', agentPath: '/root/read' }) + ) + + expect(agents().map((agent) => agent.label)).toEqual(['read', 'read 2']) + }) + + it('takes the latest token snapshot per child and never accumulates updates', () => { + const { roster, agents } = createHarness() + + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + roster.handleTokenUsage({ threadId: 'child-1', tokenUsage: { total: { totalTokens: 100 } } }) + roster.handleTokenUsage({ threadId: 'child-1', tokenUsage: { total: { totalTokens: 250 } } }) + + expect(agents()).toMatchObject([{ tokens: 250 }]) + }) + + it('retains a usage frame that arrives before the child is known', () => { + const { roster, agents } = createHarness() + + roster.handleTokenUsage({ threadId: 'child-1', tokenUsage: { total: { totalTokens: 900 } } }) + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + + expect(agents()).toMatchObject([{ tokens: 900 }]) + }) + + it('never attributes the parent thread its own usage', () => { + const { roster, agents, appended } = createHarness() + + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + const beforeParentUsage = appended.length + roster.handleTokenUsage({ threadId: THREAD, tokenUsage: { total: { totalTokens: 26099 } } }) + + expect(appended).toHaveLength(beforeParentUsage) + expect(agents()).toHaveLength(1) + expect(agents()[0]).not.toHaveProperty('tokens') + }) + + // The row is durable and both readers clip these fields to the same cap, so + // writing more than that is bytes replayed on every reconnect and then thrown + // away. The marker is an ellipsis, not the tool-output truncation sentence: + // `id` is the roster key and the renderer's React key. + it('bounds the provider strings the roster row carries into the journal', () => { + const { roster, agents, latest } = createHarness() + const oversized = 'a'.repeat(20 * 1024) + + deliver( + roster, + activity({ kind: 'started', agentThreadId: oversized, agentPath: `/root/${oversized}` }) + ) + + const entry = agents()[0] + expect(entry?.label.length).toBeLessThanOrEqual(MAX_SUBAGENT_FIELD_CHARS) + expect(entry?.label).toMatch(/…~0$/) + expect(entry?.id.length).toBeLessThanOrEqual(MAX_SUBAGENT_FIELD_CHARS) + expect(entry?.id).toMatch(/…~0$/) + expect(JSON.stringify(latest()?.body)).not.toContain('output truncated') + expect(isAdmissibleAgentJournalItemBody(latest()?.body)).toBe(true) + }) + + // The clip cuts UTF-16 code units, so a boundary landing inside a surrogate + // pair left a LONE high surrogate in a durable row — malformed, and replaced + // with U+FFFD through any non-JSON UTF-8 hop. + it('never clips a provider string mid surrogate pair', () => { + const { roster, agents } = createHarness() + const astral = '😀'.repeat(400) + + deliver( + roster, + activity({ kind: 'started', agentThreadId: astral, agentPath: `/root/${astral}` }) + ) + + const entry = agents()[0] + expect(entry?.id.length).toBeLessThanOrEqual(MAX_SUBAGENT_FIELD_CHARS) + expect(Buffer.from(entry?.id ?? '', 'utf8').toString('utf8')).toBe(entry?.id) + expect(Buffer.from(entry?.label ?? '', 'utf8').toString('utf8')).toBe(entry?.label) + }) + + // The clip removes exactly the tail that told two children apart: `id` is the + // renderer's React key, and `claimLabel` writes its repeat ordinal at the end. + // Two clipped children collapsing to one key drew two rows under one identity. + it('keeps clipped ids and labels distinct between children', () => { + const { roster, agents } = createHarness() + const prefix = 'p'.repeat(MAX_SUBAGENT_FIELD_CHARS) + const sharedPath = `/root/${'q'.repeat(640)}` + + deliver( + roster, + activity({ kind: 'started', agentThreadId: `${prefix}AAAA`, agentPath: sharedPath }) + ) + deliver( + roster, + activity({ kind: 'started', agentThreadId: `${prefix}BBBB`, agentPath: sharedPath }) + ) + + const entries = agents() + expect(entries).toHaveLength(2) + expect(new Set(entries.map((agent) => agent.id)).size).toBe(2) + expect(new Set(entries.map((agent) => agent.label)).size).toBe(2) + for (const agent of entries) { + expect(agent.id.length).toBeLessThanOrEqual(MAX_SUBAGENT_FIELD_CHARS) + expect(agent.label.length).toBeLessThanOrEqual(MAX_SUBAGENT_FIELD_CHARS) + } + }) + + it('caps the children one spawn group admits', () => { + const { roster, agents, appended } = createHarness() + for (let index = 0; index < MAX_CODEX_SUBAGENTS_PER_GROUP; index++) { + deliver( + roster, + activity({ kind: 'started', agentThreadId: `child-${index}`, agentPath: '/root/read' }) + ) + } + const atCap = appended.length + + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-over-cap', agentPath: '/root/read' }) + ) + + expect(agents()).toHaveLength(MAX_CODEX_SUBAGENTS_PER_GROUP) + expect(agents().map((agent) => agent.id)).not.toContain('child-over-cap') + // Refusing the child must not burn a revision either. + expect(appended).toHaveLength(atCap) + }) + + // The eviction is the KNOWN LIMITATION the module documents: `groups` is never + // seeded from the journal, so the evicted group's next child rebuilds its + // durable row from that one child. Pinned so the boundary cannot move silently. + it('caps live spawn groups, and an evicted group rebuilds its row from one child', () => { + const { roster, appended, agents } = createHarness() + for (let index = 0; index <= MAX_CODEX_SUBAGENT_GROUPS; index++) { + deliver( + roster, + activity({ kind: 'started', agentThreadId: `child-${index}`, agentPath: '/root/read' }), + `turn-${index}` + ) + } + const evicted = codexSubagentGroupIdentity(codexSubagentGroupId(THREAD, 'turn-0')) + const rowsFor = (identity: AgentJournalItemIdentity): Appended[] => + appended.filter((entry) => JSON.stringify(entry.identity) === JSON.stringify(identity)) + expect(rowsFor(evicted)).toHaveLength(1) + + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-late', agentPath: '/root/search' }), + 'turn-0' + ) + + expect(latestIdentity(appended)).toEqual(evicted) + expect(agents().map((agent) => agent.id)).toEqual(['child-late']) + }) + + it('keeps a token count a later thread-map eviction would otherwise retract', () => { + const { roster, agents } = createHarness() + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + roster.handleTokenUsage({ threadId: 'child-1', tokenUsage: { total: { totalTokens: 4242 } } }) + expect(agents()).toMatchObject([{ tokens: 4242 }]) + + for (let index = 0; index < MAX_CODEX_TOKEN_USAGE_THREADS; index++) { + roster.handleTokenUsage({ + threadId: `other-${index}`, + tokenUsage: { total: { totalTokens: index } } + }) + } + deliver( + roster, + activity({ kind: 'completed', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + + expect(agents()).toMatchObject([{ state: 'completed', tokens: 4242 }]) + }) + + it('caps retained usage threads, so a frame evicted before its child is dropped', () => { + const { roster, agents } = createHarness() + roster.handleTokenUsage({ threadId: 'child-1', tokenUsage: { total: { totalTokens: 900 } } }) + for (let index = 0; index < MAX_CODEX_TOKEN_USAGE_THREADS; index++) { + roster.handleTokenUsage({ + threadId: `other-${index}`, + tokenUsage: { total: { totalTokens: index } } + }) + } + + deliver( + roster, + activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/read' }) + ) + + expect(agents()[0]).not.toHaveProperty('tokens') + }) + + it('declines a payload that is not a subagent item or a usage frame', () => { + const { roster } = createHarness() + + expect( + roster.handleItem({ + threadId: THREAD, + turnId: TURN, + item: { type: 'commandExecution', id: 'item-9' } + }) + ).toBeNull() + expect(roster.handleTokenUsage({ threadId: 'child-1' })).toBeNull() + }) + + // A refusal must never advance the duplicate-suppression state: an identical + // replay would short-circuit and the revision would never be retried. The + // append and the publish are the two ways to be refused, so both are covered. + it.each([{ refuse: 'append' as const }, { refuse: 'publish' as const }])( + 'retries the same revision after the $refuse is refused', + ({ refuse }) => { + let refusing = true + const appended: Appended[] = [] + const published: number[] = [] + const refusal = { accepted: false, reason: 'backpressure' } as const + const roster = new CodexSubagentRoster({ + sink: { + appendItem: () => {}, + appendTombstone: () => {}, + publish: () => {}, + tryAppendItem: (identity, body) => { + if (refusing && refuse === 'append') { + return refusal + } + appended.push({ identity, body }) + return { accepted: true } + }, + tryPublish: () => { + if (refusing && refuse === 'publish') { + return refusal + } + published.push(1) + return { accepted: true } + } + }, + primaryThreadId: () => THREAD, + activeTurn: () => TURN, + now: () => 1_000 + }) + const item = activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/read' }) + + expect(roster.handleItem({ threadId: THREAD, turnId: TURN, item })).toEqual(refusal) + + // The wire redelivers the very same item; nothing about the roster changed, + // so only a cleared suppression state can get the revision out. + refusing = false + expect(roster.handleItem({ threadId: THREAD, turnId: TURN, item })).toEqual({ + accepted: true + }) + // The retry re-appends when the publish was the half that failed; the real + // queue coalesces those two by the group key into one journal write. What + // must not happen is the revision never being published at all. + expect(published).toHaveLength(1) + const body = appended.at(-1)?.body + expect( + body?.kind === 'message' ? body.blocks.filter(isSubagentGroupBlock) : [] + ).toMatchObject([{ agents: [{ id: 'child-1', state: 'working' }] }]) + } + ) + + // The sweep is the last event a group ever gets. A refusal there, left + // unretried, strands the settled roster's final revision — the exact "row + // stays stale forever" this row exists to prevent. + it('republishes the settled roster when the sweep publish was refused', () => { + let refusing = false + const appended: Appended[] = [] + const published: number[] = [] + const roster = new CodexSubagentRoster({ + sink: { + appendItem: () => {}, + appendTombstone: () => {}, + publish: () => {}, + tryAppendItem: (identity, body) => { + appended.push({ identity, body }) + return { accepted: true } + }, + tryPublish: () => { + if (refusing) { + return { accepted: false, reason: 'backpressure' } + } + published.push(1) + return { accepted: true } + } + }, + primaryThreadId: () => THREAD, + activeTurn: () => TURN, + now: () => 1_000 + }) + roster.handleItem({ + threadId: THREAD, + turnId: TURN, + item: activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/read' }) + }) + const publishedBeforeSweep = published.length + + refusing = true + expect(roster.settleSession()).toEqual({ accepted: false, reason: 'backpressure' }) + + // The retry sweep flips no state — every child already latched — so only a + // cleared suppression state can carry the unverifiable roster out. + refusing = false + expect(roster.settleSession()).toEqual({ accepted: true }) + expect(published.length).toBe(publishedBeforeSweep + 1) + const body = appended.at(-1)?.body + expect(body?.kind === 'message' ? body.blocks.filter(isSubagentGroupBlock) : []).toMatchObject([ + { agents: [{ id: 'child-1', state: 'unverifiable' }] } + ]) + }) + + it('propagates sink backpressure instead of reporting the row as written', () => { + const roster = new CodexSubagentRoster({ + sink: { + appendItem: () => {}, + appendTombstone: () => {}, + publish: () => {}, + tryAppendItem: () => ({ accepted: false, reason: 'backpressure' }), + tryPublish: () => ({ accepted: true }) + }, + primaryThreadId: () => THREAD, + activeTurn: () => TURN + }) + + expect( + roster.handleItem({ + threadId: THREAD, + turnId: TURN, + item: activity({ kind: 'started', agentThreadId: 'child-1', agentPath: '/root/read' }) + }) + ).toEqual({ accepted: false, reason: 'backpressure' }) + }) +}) diff --git a/src/main/codex/codex-subagent-roster.ts b/src/main/codex/codex-subagent-roster.ts new file mode 100644 index 00000000000..257fe705764 --- /dev/null +++ b/src/main/codex/codex-subagent-roster.ts @@ -0,0 +1,347 @@ +// The Codex subagent roster: one journal row per spawn group, revised in place. +// +// There is no snapshot to read. `agentsStates` arrived empty in the live probe +// and children get no `thread/started`, so the roster is +// accumulated purely from `subAgentActivity` items — each of which arrives TWICE +// (`item/started` and `item/completed`). Every transition here is therefore +// idempotent, and a terminal state latches: duplicate and out-of-order delivery +// must not resurrect a settled child. +// +// KNOWN LIMITATION: `groups` is process-local and is never seeded from the +// journal, while the row's identity is keyed on the group id alone. So once a +// group leaves the map its row stays, and the next activity item rebuilds that +// row from one child — rewriting N down to one. Two ways in: eviction past +// MAX_CODEX_SUBAGENT_GROUPS, which drops the oldest-inserted group in-process +// even while it is live, and skips the sweep so its children never latch +// `unverifiable`; and a restart on `threadId:outside-turn`, the one group id +// that outlives the process — `thread/resume` is verified to return the same +// thread, and a real turn id is assumed freshly minted per turn. Seeding from +// the journal is the fix. + +import type { + AgentJournalItemBody, + AgentJournalItemIdentity +} from '../../shared/agent-session-journal-types' +import { + canReplaceSubagentState, + isTerminalSubagentState, + MAX_SUBAGENT_FIELD_CHARS, + subagentGroupFallbackText +} from '../../shared/native-chat-subagent-summary' +import type { NativeChatSubagentEntry } from '../../shared/native-chat-types' +import type { + StructuredAgentSessionEventSink, + StructuredAgentSessionSinkAdmission +} from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import { + codexSubagentLabel, + codexSubagentStateForKind, + isCodexRootAgentActivity, + readCodexSubagentActivity, + readCodexThreadTokenTotal +} from './codex-subagent-activity' +import type { CodexThreadItem } from './codex-structured-item-translation' +import { + MAX_CODEX_SUBAGENT_GROUPS, + MAX_CODEX_SUBAGENTS_PER_GROUP, + MAX_CODEX_TOKEN_USAGE_THREADS +} from './codex-structured-journal-limits' + +const ADMITTED: StructuredAgentSessionSinkAdmission = { accepted: true } + +/** The turn a group belongs to when Codex reports activity outside any turn. + * Mirrors the generic-frame bucket name so the two read alike in the journal. */ +const OUTSIDE_TURN = 'outside-turn' + +const UNLABELLED_AGENT = 'subagent' + +type RosterGroup = { + groupId: string + identity: AgentJournalItemIdentity + /** Insertion order is the display order; the map holds the state. */ + entries: Map + /** Times each label has been claimed, so a repeat gets an ordinal suffix. */ + labelCounts: Map + /** Last body written, so an idempotent replay writes no new revision. */ + lastSerialized: string | null +} + +/** Group identity: the parent turn that spawned the children. `agentPath` is a + * tree rooted at the parent thread, so every child of one turn shares a row + * no matter which thread's stream carried its activity item. */ +export function codexSubagentGroupId(threadId: string, turnId: string | null): string { + return `${threadId}:${turnId ?? OUTSIDE_TURN}` +} + +/** Durable journal identity for the group's row — stable across revisions and + * across a restart, so replay finds the same row instead of appending a new one. */ +export function codexSubagentGroupIdentity(groupId: string): AgentJournalItemIdentity { + return { provider: 'orca', clientMessageId: `codex-subagents:${groupId}` } +} + +export type CodexSubagentRosterDeps = { + sink: StructuredAgentSessionEventSink + /** The thread that owns the agent tree; falls back to the event's thread. */ + primaryThreadId: () => string | null + activeTurn: (threadId: string) => string | null + now?: () => number +} + +export class CodexSubagentRoster { + private readonly groups = new Map() + /** Latest reported total per thread, kept regardless of roster membership: a + * usage frame can arrive before the child's first activity item, and filtering + * at receipt would lose it permanently. Children are selected at write time; + * the map itself is LRU-capped in `handleTokenUsage`. */ + private readonly tokensByThread = new Map() + private readonly now: () => number + + constructor(private readonly deps: CodexSubagentRosterDeps) { + this.now = deps.now ?? (() => Date.now()) + } + + /** Consume a `subAgentActivity` item. Returns null when the item is not one. */ + handleItem(input: { + threadId: string + turnId: string | null + item: CodexThreadItem + }): StructuredAgentSessionSinkAdmission | null { + const activity = readCodexSubagentActivity(input.item) + if (!activity) { + return null + } + // The root node is the parent turn itself, not a child it spawned. + if (isCodexRootAgentActivity(activity)) { + return ADMITTED + } + const group = this.groupFor(input.threadId, input.turnId) + const existing = group.entries.get(activity.agentThreadId) + const state = codexSubagentStateForKind(activity.kind) + if (!existing) { + // Rule: the first event for a child may be ANY kind. An `interacted` or + // `completed` with no prior `started` creates the entry in the state its + // kind implies rather than being dropped for lacking a roster row. + if (group.entries.size >= MAX_CODEX_SUBAGENTS_PER_GROUP) { + return ADMITTED + } + const now = this.now() + group.entries.set(activity.agentThreadId, { + id: activity.agentThreadId, + label: this.claimLabel(group, codexSubagentLabel(activity)), + state, + startedAt: now, + ...(isTerminalSubagentState(state) ? { settledAt: now } : {}) + }) + } else if (canReplaceSubagentState(existing.state, state)) { + // A child's own verdict latches. Re-applying the same non-terminal state + // is a no-op, which is what makes the duplicate `item/started` + + // `item/completed` delivery idempotent. `unverifiable` does not latch: a + // child swept when contact was lost can still report what it actually did + // if contact returns. + group.entries.set(activity.agentThreadId, { + ...existing, + state, + ...(isTerminalSubagentState(state) ? { settledAt: this.now() } : {}) + }) + } + return this.write(group) + } + + /** Consume `thread/tokenUsage/updated`. Returns null when the params are not one. */ + handleTokenUsage(params: unknown): StructuredAgentSessionSinkAdmission | null { + const usage = readCodexThreadTokenTotal(params) + if (!usage) { + return null + } + // A running total: the newest frame REPLACES the previous one. Summing + // updates would multiply a single child's usage by its frame count. + // Re-insert so the eviction scan below sees recency: `set` on an existing + // key keeps its original position, which would age out an active thread. + this.tokensByThread.delete(usage.threadId) + this.tokensByThread.set(usage.threadId, usage.totalTokens) + while (this.tokensByThread.size > MAX_CODEX_TOKEN_USAGE_THREADS) { + const oldest = this.tokensByThread.keys().next().value + if (typeof oldest !== 'string') { + break + } + this.tokensByThread.delete(oldest) + } + for (const group of this.groups.values()) { + if (!group.entries.has(usage.threadId)) { + continue + } + const admission = this.write(group) + if (!admission.accepted) { + return admission + } + } + return ADMITTED + } + + /** + * The provider is gone, so any child still reported as working will never be + * settled by an event: it becomes `unverifiable` — contact was lost, which is + * NOT evidence the child exited. + * + * This is the ONLY sweep. A turn ending is not one: `spawn_agent` children + * routinely outlive their turn and keep reporting into the same group. + */ + settleSession(): StructuredAgentSessionSinkAdmission { + for (const group of this.groups.values()) { + const admission = this.sweep(group) + if (!admission.accepted) { + return admission + } + } + return ADMITTED + } + + dispose(): void { + this.groups.clear() + this.tokensByThread.clear() + } + + private sweep(group: RosterGroup | undefined): StructuredAgentSessionSinkAdmission { + if (!group) { + return ADMITTED + } + let changed = false + for (const [id, entry] of group.entries) { + if (isTerminalSubagentState(entry.state)) { + continue + } + group.entries.set(id, { ...entry, state: 'unverifiable', settledAt: this.now() }) + changed = true + } + // A null `lastSerialized` means the previous write was refused part-way, so + // the settled roster's last revision is queued but never published. Nothing + // is guaranteed to write this group again, so retry here even when the sweep + // itself changed nothing. + return changed || group.lastSerialized === null ? this.write(group) : ADMITTED + } + + private groupFor(threadId: string, turnId: string | null): RosterGroup { + const ownerThreadId = this.deps.primaryThreadId() ?? threadId + const ownerTurnId = + ownerThreadId === threadId ? turnId : (this.deps.activeTurn(ownerThreadId) ?? turnId) + const groupId = codexSubagentGroupId(ownerThreadId, ownerTurnId) + const existing = this.groups.get(groupId) + if (existing) { + return existing + } + const group: RosterGroup = { + groupId, + identity: codexSubagentGroupIdentity(groupId), + entries: new Map(), + labelCounts: new Map(), + lastSerialized: null + } + this.groups.set(groupId, group) + while (this.groups.size > MAX_CODEX_SUBAGENT_GROUPS) { + const oldest = this.groups.keys().next().value + if (typeof oldest !== 'string' || oldest === groupId) { + break + } + this.groups.delete(oldest) + } + return group + } + + /** Two children can share a trailing path segment; the ordinal keeps their + * rows apart without inventing a name the provider never sent. */ + private claimLabel(group: RosterGroup, label: string | null): string { + const base = label ?? UNLABELLED_AGENT + const seen = group.labelCounts.get(base) ?? 0 + group.labelCounts.set(base, seen + 1) + return seen === 0 ? base : `${base} ${seen + 1}` + } + + private write(group: RosterGroup): StructuredAgentSessionSinkAdmission { + const agents = [...group.entries].map(([id, entry]) => { + const tokens = this.tokensByThread.get(id) + if (typeof tokens !== 'number' || tokens === entry.tokens) { + return entry + } + // Persisted, not merely read: the thread map is LRU-capped, and reading it + // afresh each write would retract a count this row has already shown. + const merged = { ...entry, tokens } + group.entries.set(id, merged) + return merged + }) + const body = codexSubagentGroupBody(group.groupId, agents) + const serialized = JSON.stringify(body) + if (serialized === group.lastSerialized) { + // Nothing changed — a duplicate delivery must not burn a revision. + return ADMITTED + } + group.lastSerialized = serialized + // The append coalesces per group so a burst collapses to the latest roster. + // The publish must NOT reuse that key: the queue coalesces by key alone, + // with no op-kind check, so a publish carrying it would splice out the + // still-queued append and the row would never reach the journal. + const options = { coalescingKey: `codex-subagents:${group.groupId}` } + const admission = this.deps.sink.tryAppendItem + ? this.deps.sink.tryAppendItem(group.identity, body, options) + : (this.deps.sink.appendItem(group.identity, body, options), ADMITTED) + if (!admission.accepted) { + group.lastSerialized = null + return admission + } + const published = this.deps.sink.tryPublish + ? this.deps.sink.tryPublish() + : (this.deps.sink.publish(), ADMITTED) + if (!published.accepted) { + // Symmetric with the append refusal above: the suppression state may only + // advance once the revision is both queued AND published. Left set, an + // identical replay short-circuits and the last revision of a settled + // roster stays queued but never reaches the renderer. + group.lastSerialized = null + } + return published + } +} + +/** The roster row: the structured block plus the plain sentence an older client + * renders in its place. A message whose only block is the new variant would + * reach such a client with nothing it can draw. */ +export function codexSubagentGroupBody( + groupId: string, + agents: readonly NativeChatSubagentEntry[] +): AgentJournalItemBody { + const bounded = agents.map((agent, index) => ({ + ...agent, + id: boundSubagentField(agent.id, index), + label: boundSubagentField(agent.label, index) + })) + return { + kind: 'message', + role: 'system', + blocks: [ + { type: 'text', text: subagentGroupFallbackText(bounded) }, + { type: 'subagent-group', groupId, agents: bounded } + ] + } +} + +/** `id` and `label` are provider strings, so they take the bound both readers of + * this row already clip them to. A plain length check, not the tool-output + * bound: that one digests the whole value before it checks the length, and this + * runs twice per child on every streamed token-usage frame. + * + * A clip is not identity-preserving, so a clipped value carries the child's + * index: two ids sharing a long prefix collapse to one React key, and + * `claimLabel` writes its ordinal at the very tail the clip removes. The index + * is reserved out of the bound, not appended to it, because both readers + * re-clip to the same cap and would cut a suffix that overflowed it. */ +function boundSubagentField(value: string, index: number): string { + if (value.length <= MAX_SUBAGENT_FIELD_CHARS) { + return value + } + const suffix = `…~${index}` + const keep = MAX_SUBAGENT_FIELD_CHARS - suffix.length + // Slicing UTF-16 units can split a surrogate pair; a lone surrogate is + // malformed in a durable row and lossy through any non-JSON UTF-8 hop. + const last = value.charCodeAt(keep - 1) + const end = last >= 0xd800 && last <= 0xdbff ? keep - 1 : keep + return `${value.slice(0, end)}${suffix}` +} diff --git a/src/main/native-chat/agent-session-journal/journal-store-open.ts b/src/main/native-chat/agent-session-journal/journal-store-open.ts index 721e5f4ba7f..7b5b6d0dff8 100644 --- a/src/main/native-chat/agent-session-journal/journal-store-open.ts +++ b/src/main/native-chat/agent-session-journal/journal-store-open.ts @@ -1,4 +1,8 @@ import { mkdir } from 'node:fs/promises' +import type { + AgentJournalItemBody, + AgentJournalItemIdentity +} from '../../../shared/agent-session-journal-types' import type { AgentType } from '../../../shared/agent-status-types' import { findJournalFileFormatRemnant, @@ -6,6 +10,7 @@ import { } from './journal-file-format-remnant' import type { JournalLoad } from './journal-open' import { journalRepairDisclosure, type JournalRepairDisclosure } from './journal-repair-disclosure' +import { staleSubagentRosterRevisions } from './journal-subagent-liveness' /** What any of this file's disclosures hands the store — a repair's, or the * pre-SQLite notice's. Same shape, and neither is only a repair. */ @@ -36,9 +41,9 @@ export async function openJournalStoreState(input: { adopt: (loaded: JournalLoad) => void /** Republishes an anchor row for an epoch a repair emptied. */ publishRepairEpoch: () => void - appendDisclosure: ( - identity: JournalRepairDisclosure['identity'], - body: JournalRepairDisclosure['body'], + appendItem: ( + identity: AgentJournalItemIdentity, + body: AgentJournalItemBody, fence: number ) => Promise agent: AgentType @@ -68,8 +73,9 @@ export async function openJournalStoreState(input: { } if (input.malformedRows() > 0 && !input.readOnly()) { const disclosure = journalRepairDisclosure({ malformedRows: input.malformedRows() }) - await input.appendDisclosure(disclosure.identity, disclosure.body, input.highestFence()) + await input.appendItem(disclosure.identity, disclosure.body, input.highestFence()) } + await settleStaleSubagentRosters(input, loaded) // Founding the epoch and appending the row are two transactions, and a // committed epoch sends every later open down this branch instead. Anything // that interrupts between them — a quit during startup restore, a failed @@ -92,7 +98,7 @@ export async function openJournalStoreState(input: { async function discloseFileFormatRemnant(input: { journalDir: string agent: AgentType - appendDisclosure: ( + appendItem: ( identity: JournalDisclosure['identity'], body: JournalDisclosure['body'], fence: number @@ -108,5 +114,32 @@ async function discloseFileFormatRemnant(input: { return } const disclosure = journalFileFormatRemnantDisclosure({ transcriptPath, agent: input.agent }) - await input.appendDisclosure(disclosure.identity, disclosure.body, input.highestFence()) + await input.appendItem(disclosure.identity, disclosure.body, input.highestFence()) +} + +/** + * Retires a `working` subagent roster the previous host never got to settle. + * + * Skipped on a corrupt load: that journal is still owed a rebuild from provider + * history, and content written past the repair's free sequence retires the + * demand for it. + */ +async function settleStaleSubagentRosters( + input: { + appendItem: ( + identity: AgentJournalItemIdentity, + body: AgentJournalItemBody, + fence: number + ) => Promise + highestFence: () => number + readOnly: () => boolean + }, + loaded: JournalLoad +): Promise { + if (input.readOnly() || loaded.corrupt) { + return + } + for (const revision of staleSubagentRosterRevisions(loaded.state.items.values())) { + await input.appendItem(revision.identity, revision.body, input.highestFence()) + } } diff --git a/src/main/native-chat/agent-session-journal/journal-store-restore.ts b/src/main/native-chat/agent-session-journal/journal-store-restore.ts index 3a5d3c7ac6e..fc69dd339d8 100644 --- a/src/main/native-chat/agent-session-journal/journal-store-restore.ts +++ b/src/main/native-chat/agent-session-journal/journal-store-restore.ts @@ -39,8 +39,7 @@ export function restoreJournalStore( publishRepairEpoch: () => collaborators.epochController.start('unreconcilable_prefix', host.state().highestFence), adopt: host.adopt, - appendDisclosure: (identity, body, fence) => - host.journal().appendItem(identity, body, { fence }), + appendItem: (identity, body, fence) => host.journal().appendItem(identity, body, { fence }), agent: host.identity.agent, highestFence: () => host.state().highestFence, malformedRows: host.malformedRows, 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 new file mode 100644 index 00000000000..9d6887de61e --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-subagent-liveness.test.ts @@ -0,0 +1,202 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import type { + AgentJournalRenderItem, + 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 { + codexSubagentGroupBody, + codexSubagentGroupIdentity +} from '../../codex/codex-subagent-roster' +import type { openAgentSessionJournal } from './journal-store-factory' +import { createTrackedJournalOpener } from './journal-store-test-open' +import { staleSubagentRosterRevisions } from './journal-subagent-liveness' + +const IDENTITY: AgentSessionJournalIdentity = { + sessionId: 'session-1', + workspaceId: 'ws-1', + hostId: 'host-1', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: 'thread-1' } +} + +const GROUP_ID = 'thread-1:turn-1' + +let root: string +let clock = 1_000 + +function tick(): number { + clock += 1 + return clock +} + +const journals = createTrackedJournalOpener() + +async function open(overrides: Partial[0]> = {}) { + return journals.open({ + identity: IDENTITY, + journalDir: root, + now: tick, + mintEpoch: () => `epoch-${clock}`, + ...overrides + }) +} + +/** The row as the producer writes it: the structured block plus its twin. */ +function rosterRow(agents: NativeChatSubagentEntry[]) { + return { + identity: codexSubagentGroupIdentity(GROUP_ID), + body: codexSubagentGroupBody(GROUP_ID, agents) + } +} + +function renderItem(agents: NativeChatSubagentEntry[]): AgentJournalRenderItem { + const row = rosterRow(agents) + return { + itemId: agentJournalItemKey(row.identity), + revision: 1, + body: row.body, + sequence: 2, + observedAt: 1 + } +} + +function rosterOf(body: AgentJournalRenderItem['body']): NativeChatSubagentEntry[] { + return body.kind === 'message' ? (body.blocks.find(isSubagentGroupBlock)?.agents ?? []) : [] +} + +function twinOf(body: AgentJournalRenderItem['body']): string | undefined { + return body.kind === 'message' + ? body.blocks.find((block) => block.type === 'text')?.text + : undefined +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-journal-subagents-')) + clock = 1_000 +}) + +afterEach(async () => { + await journals.closeAll() + await rm(root, { recursive: true, force: true }) +}) + +describe('staleSubagentRosterRevisions', () => { + it('settles a child the previous host left working, and moves the twin with it', () => { + const revisions = staleSubagentRosterRevisions([ + renderItem([ + { id: 'a', label: 'read_readme', state: 'working', startedAt: 10 }, + { id: 'b', label: 'read_package', state: 'completed', startedAt: 10, settledAt: 20 } + ]) + ]) + + expect(revisions).toHaveLength(1) + expect(rosterOf(revisions[0]!.body)).toMatchObject([ + { id: 'a', state: 'unverifiable' }, + { id: 'b', state: 'completed' } + ]) + // Mobile reads only this sentence, so it may not go on saying `Kicked off`. + expect(twinOf(revisions[0]!.body)).toBe('Ran 2 subagents (1 unverifiable)') + }) + + // 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', () => { + const revisions = staleSubagentRosterRevisions([ + renderItem([{ id: 'a', label: 'read', state: 'working', startedAt: 10 }]) + ]) + + expect(rosterOf(revisions[0]!.body)[0]).not.toHaveProperty('settledAt') + }) + + it('owes nothing for a roster whose children all settled', () => { + expect( + staleSubagentRosterRevisions([ + renderItem([{ id: 'a', label: 'read', state: 'completed', settledAt: 20 }]) + ]) + ).toEqual([]) + }) + + it('leaves rows that carry no roster alone', () => { + expect( + staleSubagentRosterRevisions([ + { + itemId: 'orca:plain', + revision: 1, + body: { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'hi' }] }, + sequence: 2, + observedAt: 1 + } + ]) + ).toEqual([]) + }) + + // Appending under a fresh identity would add a second row rather than revise + // the one on disk, so an unaddressable key is left exactly as it is. + it('skips a row whose key cannot be parsed back to its identity', () => { + expect( + staleSubagentRosterRevisions([ + { ...renderItem([{ id: 'a', label: 'r', state: 'working' }]), itemId: 'not-a-key' } + ]) + ).toEqual([]) + }) +}) + +describe('journal reopen after the writing host is gone', () => { + it('settles a persisted working roster to unverifiable, while the live row still reads working', async () => { + const live = await open() + const row = rosterRow([ + { id: 'a', label: 'read_readme', state: 'working', startedAt: 10 }, + { id: 'b', label: 'read_package', state: 'working', startedAt: 10 } + ]) + await live.appendItem(row.identity, row.body, { fence: 0 }) + + // Still the writing host: it can see the children, so the row says so. + const beforeRestart = live.snapshot().items.at(-1)! + expect(rosterOf(beforeRestart.body)).toMatchObject([{ state: 'working' }, { state: 'working' }]) + expect(twinOf(beforeRestart.body)).toBe('Kicked off 2 subagents') + + // The host dies without ever settling them — no `ended`, so no session sweep. + await live.close() + + const reopened = await open() + const afterRestart = reopened.snapshot().items.at(-1)! + expect(afterRestart.itemId).toBe(beforeRestart.itemId) + expect(rosterOf(afterRestart.body)).toMatchObject([ + { id: 'a', state: 'unverifiable' }, + { id: 'b', state: 'unverifiable' } + ]) + expect(twinOf(afterRestart.body)).toBe('Ran 2 subagents (2 unverifiable)') + }) + + it('revises the row in place rather than appending a second one', async () => { + const live = await open() + const row = rosterRow([{ id: 'a', label: 'read', state: 'working', startedAt: 10 }]) + await live.appendItem(row.identity, row.body, { fence: 0 }) + const before = live.snapshot().items.length + await live.close() + + const reopened = await open() + expect(reopened.snapshot().items).toHaveLength(before) + expect(reopened.snapshot().items.at(-1)?.revision).toBe(2) + }) + + 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 }]) + await live.appendItem(row.identity, row.body, { fence: 0 }) + await live.close() + + const once = await open() + const revision = once.snapshot().items.at(-1)?.revision + await once.close() + + const twice = await open() + expect(twice.snapshot().items.at(-1)?.revision).toBe(revision) + }) +}) 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 new file mode 100644 index 00000000000..9b2724e9d1d --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-subagent-liveness.ts @@ -0,0 +1,101 @@ +// A roster row left claiming live children by a host that is gone. +// +// The writing host revises its `subagent-group` rows in place while it can see +// the children, and sweeps whatever is still `working` when the provider goes +// away. A host that DIED — crash, quit, force-restart — does neither: its last +// revision goes on saying `working`, and nothing replays those children, so no +// later event can ever settle them. Opening the journal is the one moment a new +// host can state the truth about the old one: contact was lost. That is +// `unverifiable`, never a synthesized exit — see +// `docs/reference/ssh-execution-boundary.md`. +// +// Reconciles JOURNAL ROWS, not roster state: nothing here seeds the producer's +// in-process group map, so the roster's known limitation is untouched. + +import { + agentJournalItemKey, + parseAgentJournalItemKey +} from '../../../shared/agent-session-journal-item-key' +import type { + AgentJournalItemBody, + AgentJournalItemIdentity, + AgentJournalRenderItem +} from '../../../shared/agent-session-journal-types' +import { + isSubagentGroupFallbackText, + normalizeSubagentState, + subagentGroupFallbackText +} from '../../../shared/native-chat-subagent-summary' +import { + isSubagentGroupBlock, + type NativeChatBlock, + type NativeChatSubagentGroupBlock +} from '../../../shared/native-chat-types' + +export type JournalSubagentLivenessRevision = { + identity: AgentJournalItemIdentity + body: AgentJournalItemBody +} + +/** The revisions a reopened journal owes: one per row still claiming a live + * child. Empty — the common case — when nothing was left mid-flight. */ +export function staleSubagentRosterRevisions( + items: Iterable +): JournalSubagentLivenessRevision[] { + const revisions: JournalSubagentLivenessRevision[] = [] + for (const item of items) { + const body = item.body + if (body.kind !== 'message' || !body.blocks.some(hasWorkingChild)) { + continue + } + // A key that will not parse cannot be re-addressed, and appending under a + // fresh identity would duplicate the row rather than revise it. + const identity = parseAgentJournalItemKey(item.itemId) + if (!identity || agentJournalItemKey(identity) !== item.itemId) { + continue + } + revisions.push({ identity, body: { ...body, blocks: settleBlocks(body.blocks) } }) + } + return revisions +} + +function hasWorkingChild(block: NativeChatBlock): boolean { + return ( + isSubagentGroupBlock(block) && + block.agents.some((agent) => normalizeSubagentState(agent.state) === 'working') + ) +} + +/** 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 rosters = settled.filter(isSubagentGroupBlock) + const only = rosters.length === 1 ? rosters[0] : undefined + if (!only) { + return settled + } + // 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) => + block.type === 'text' && isSubagentGroupFallbackText(block.text) + ? { ...block, text: twin } + : block + ) +} + +function settleGroup(block: NativeChatSubagentGroupBlock): NativeChatSubagentGroupBlock { + return { + ...block, + agents: block.agents.map((agent) => + normalizeSubagentState(agent.state) === 'working' + ? { ...agent, state: 'unverifiable' as const } + : agent + ) + } +} diff --git a/src/main/native-chat/agent-session-wire/provider-frame-activity.test.ts b/src/main/native-chat/agent-session-wire/provider-frame-activity.test.ts index 79d9e4205cf..40504873282 100644 --- a/src/main/native-chat/agent-session-wire/provider-frame-activity.test.ts +++ b/src/main/native-chat/agent-session-wire/provider-frame-activity.test.ts @@ -28,6 +28,16 @@ describe('provider frame activity', () => { expect(codexProviderFrameActivity('item/reasoning/summaryPartAdded', {})).toBeNull() }) + it('names a fan-out from either Codex item type that reports one', () => { + for (const type of ['collabAgentToolCall', 'subAgentActivity']) { + expect( + codexProviderFrameActivity('item/started', { + item: { type, kind: 'started', agentThreadId: 'child-1', agentPath: '/root/read' } + }) + ).toBe('Coordinating with another agent') + } + }) + it('uses Claude descriptions and safe semantic status without exposing tool labels', () => { expect( claudeProviderFrameActivity('message:system:task_started', { 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 9860aaa81d8..d4726a9b602 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 @@ -6,6 +6,7 @@ import { isDeltaShapedProviderFrameKind, PROVIDER_FRAME_CLASSIFICATIONS } from './provider-frame-disposition' +import { unhandledProviderFrameJournalItem } from './unhandled-provider-frame' describe('provider frame classification catalog', () => { it('classifies every pinned Codex app-server notification method', () => { @@ -124,7 +125,7 @@ describe('provider frame classification catalog', () => { ) }) - it('keeps subagent items visible — the only evidence a spawned agent is working', () => { + it('suppresses subAgentActivity once the roster renders it, but never collabAgentToolCall', () => { expect( classifyProviderFrame('codex', 'item:subAgentActivity', { id: 'a-1', @@ -132,7 +133,9 @@ describe('provider frame classification catalog', () => { agentThreadId: 'thread-child', agentPath: '/root/list_directory' }) - ).toBe('timeline-substantive') + // The spawn-group roster row renders this now, so a raw gray row beside it + // would duplicate it. Suppressing it was gated on that renderer existing. + ).toBe('status-chrome') expect( classifyProviderFrame('codex', 'item:collabAgentToolCall', { id: 'c-1', @@ -161,3 +164,45 @@ describe('provider frame classification catalog', () => { } }) }) + +describe('codex subagent item disposition', () => { + it('keeps subagent lifecycle out of the transcript now that it renders as a roster row', () => { + expect( + classifyProviderFrame('codex', 'item:subAgentActivity', { + type: 'subAgentActivity', + kind: 'started', + agentThreadId: 'child-1', + agentPath: '/root/read' + }) + ).toBe('status-chrome') + }) + + it('leaves collab tool calls substantive — they may be the only subagent signal', () => { + // A session that reports no `subAgentActivity` gets no roster row, so + // suppressing this too would render its fan-out blank. + expect( + classifyProviderFrame('codex', 'item:collabAgentToolCall', { + type: 'collabAgentToolCall', + agentsStates: {} + }) + ).not.toBe('status-chrome') + }) + + it('journals no fallback row for subagent activity', () => { + expect( + unhandledProviderFrameJournalItem('codex', 'item:subAgentActivity', { + kind: 'completed', + agentThreadId: 'child-1' + }) + ).toBeNull() + }) + + it('still surfaces a subagent frame that reports a failure', () => { + expect( + classifyProviderFrame('codex', 'item:collabAgentToolCall', { + type: 'collabAgentToolCall', + status: 'failed' + }) + ).toBe('error-surface') + }) +}) diff --git a/src/main/native-chat/agent-session-wire/provider-frame-disposition.ts b/src/main/native-chat/agent-session-wire/provider-frame-disposition.ts index f05f4cd4c6c..35223a1971f 100644 --- a/src/main/native-chat/agent-session-wire/provider-frame-disposition.ts +++ b/src/main/native-chat/agent-session-wire/provider-frame-disposition.ts @@ -1,4 +1,5 @@ import type { CodexAppServerNotificationMethod } from '../../codex/codex-app-server-notification-schema' +import { CODEX_SUBAGENT_ITEM_TYPE } from '../../codex/codex-subagent-activity' import type { ClaudeStreamJsonFrameKind } from './claude-stream-json-frame-schema' export type ProviderFrameClassification = @@ -198,10 +199,21 @@ const CODEX_ITEM_CLASSIFICATIONS: Record = // The `thread/compacted` notification is already chrome; its item form is the // same event and must not read as a mysterious opcode row. contextCompaction: 'status-chrome', + // Subagent lifecycle renders as the spawn-group roster row, so its raw items + // must not print a gray `codex · item:` row beside it. The live + // notification path intercepts them before this catalog is reached; + // `restoreThread` replays them straight through `items.handle`, which is where + // the classification earns its keep. + // + // `collabAgentToolCall` is deliberately NOT suppressed with it. Nothing + // guarantees a session reports subagent work as `subAgentActivity` at all; one + // that only ever emits the collab tool call gets no roster row, and suppressing + // that too would leave its fan-out showing nothing. + [CODEX_SUBAGENT_ITEM_TYPE]: 'status-chrome', // `{id, durationMs}` and nothing else — Codex's own transcript renders it as // nothing at all. Every other item type this build does not model carries text - // a user would want (review output, an image path, hook prompt text, subagent - // progress), so those keep their visible fallback row. + // a user would want (review output, an image path, hook prompt text), so those + // keep their visible fallback row. sleep: 'status-chrome' } diff --git a/src/main/runtime/orchestration/worker-transcript-payload.test.ts b/src/main/runtime/orchestration/worker-transcript-payload.test.ts index 7899a63fb73..f47a46605e1 100644 --- a/src/main/runtime/orchestration/worker-transcript-payload.test.ts +++ b/src/main/runtime/orchestration/worker-transcript-payload.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it } from 'vitest' +import { MAX_CODEX_SUBAGENTS_PER_GROUP } from '../../codex/codex-structured-journal-limits' import { boundWorkerTranscriptMessages, redactWorkerTerminalLines @@ -57,6 +58,80 @@ describe('worker transcript wire bounds', () => { ) }) + // The bound matches the producer's per-group cap, so nothing this build writes + // is clipped here. It stays because the journal schema declares no maximum and + // a remote host may run a build with a larger one — the transport's own + // invariant that no single block is huge. + it('caps and redacts a spawn group the way every other collection is capped', () => { + const result = boundWorkerTranscriptMessages([ + { + id: 'message-roster', + role: 'system', + timestamp: null, + source: 'transcript', + blocks: [ + { + type: 'subagent-group', + groupId: 'thread-1:turn-1', + agents: Array.from({ length: 80 }, (_unused, index) => ({ + id: `child-${index}`, + label: index === 0 ? `dcap_${'A'.repeat(24)}` : 'read', + state: 'working' as const + })) + } + ] + } + ]) + + const block = result.messages[0]?.blocks[0] + expect(block?.type).toBe('subagent-group') + expect(block?.type === 'subagent-group' ? block.agents : []).toHaveLength( + MAX_CODEX_SUBAGENTS_PER_GROUP + ) + expect(JSON.stringify(result.messages)).not.toContain('dcap_') + expect(result.limited).toBe(true) + expect(result.warnings).toEqual( + expect.arrayContaining([ + 'Some subagents were omitted from oversized spawn groups.', + 'Dispatch capability tokens were redacted from transcript output.' + ]) + ) + }) + + it('bounds a spawn-group state a newer build wrote as an oversized open string', () => { + const result = boundWorkerTranscriptMessages([ + { + id: 'message-roster-state', + role: 'system', + timestamp: null, + source: 'transcript', + blocks: [ + { + type: 'subagent-group', + groupId: 'g'.repeat(900), + agents: [ + { + id: 'i'.repeat(900), + label: 'l'.repeat(900), + state: 's'.repeat(900) as 'working' + } + ] + } + ] + } + ]) + + const block = result.messages[0]?.blocks[0] + const agent = block?.type === 'subagent-group' ? block.agents[0] : undefined + expect(block?.type === 'subagent-group' ? block.groupId.length : 0).toBe(512) + expect(agent?.id.length).toBe(512) + expect(agent?.label.length).toBe(512) + // A clipped state names no state any build knows, which is what + // `unverifiable` records — a 512-character fragment is not a state at all. + expect(agent?.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 bcfc7cb0b75..f9d83e20c62 100644 --- a/src/main/runtime/orchestration/worker-transcript-payload.ts +++ b/src/main/runtime/orchestration/worker-transcript-payload.ts @@ -1,5 +1,10 @@ import { createHash } from 'node:crypto' -import type { NativeChatBlock, NativeChatMessage } from '../../../shared/native-chat-types' +import { normalizeSubagentState } from '../../../shared/native-chat-subagent-summary' +import type { + NativeChatBlock, + NativeChatMessage, + NativeChatSubagentState +} from '../../../shared/native-chat-types' export const DEFAULT_WORKER_TRANSCRIPT_MESSAGE_LIMIT = 40 export const MAX_WORKER_TRANSCRIPT_MESSAGE_LIMIT = 50 @@ -7,6 +12,14 @@ 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. +const MAX_WORKER_TRANSCRIPT_METADATA_CHARS = 512 const MAX_WORKER_TRANSCRIPT_RESPONSE_BYTES = 512 * 1024 const TRUNCATION_MARKER = '\n… (truncated)' const DISPATCH_CAPABILITY_PATTERN = /\bdcap_[A-Za-z0-9_-]{20,}\b/g @@ -128,6 +141,24 @@ 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: clipMetadata(agent.id, state), + label: clipMetadata(agent.label, state), + state: clipSubagentState(agent.state, state) + })) + } + } if (block.path || (block.url && isLocalFileLocator(block.url))) { markClipped(state, 'Local image paths were omitted from transcript output.') return { @@ -165,11 +196,22 @@ function isLocalFileLocator(value: string): boolean { function clipMetadata(value: string, state: TranscriptBoundState): string { const redacted = redactSensitiveText(value, state.warnings) - if (redacted.length <= 512) { + if (redacted.length <= MAX_WORKER_TRANSCRIPT_METADATA_CHARS) { return redacted } markClipped(state, 'Oversized transcript metadata was clipped.') - return redacted.slice(0, 512) + 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 { diff --git a/src/main/runtime/rpc/methods/native-chat-rpc-block-sanitize.ts b/src/main/runtime/rpc/methods/native-chat-rpc-block-sanitize.ts new file mode 100644 index 00000000000..fc19273d5ca --- /dev/null +++ b/src/main/runtime/rpc/methods/native-chat-rpc-block-sanitize.ts @@ -0,0 +1,132 @@ +import { + MAX_SUBAGENT_FIELD_CHARS, + normalizeSubagentState +} from '../../../../shared/native-chat-subagent-summary' +import type { NativeChatBlock, NativeChatSubagentState } from '../../../../shared/native-chat-types' +import type { RpcContext } from '../core' +import { sanitizeNativeChatRpcImageBlock } from './native-chat-rpc-image-block' + +// Why: the mobile-only payload diet. Inline image bytes are kept off every RPC +// transport; everything below that only applies to `mobile` clients, whose +// renderer previews block bodies rather than showing them whole. + +// Why: a single tool result (a big file read, a long diff) can be hundreds of KB. +// The mobile view only previews tool block bodies, so truncate them on the wire +// to keep the payload small; the marker tells the user content was clipped. +const MOBILE_BLOCK_CHAR_CAP = 4000 +// Why: text blocks are the message body itself, rendered in full by the chat +// view — a preview-sized cap cut long assistant replies mid-sentence with no way +// to read on (STA-3230). Keep only a generous safety ceiling: a transcript +// record can legally reach 2MB, and shipping that much markdown in one block +// would freeze the phone. +const MOBILE_TEXT_BLOCK_CHAR_CAP = 64_000 +const MOBILE_TOOL_INPUT_ITEMS_CAP = 20 +const MOBILE_TOOL_INPUT_NODE_CAP = 100 +// Why: a spawn group's roster is metadata, not a body — provider-supplied agent +// paths and an open-string lifecycle whose schema declares no maximum, so a +// journal from a newer build can carry more children and longer strings than +// this build ever writes. +const MOBILE_SUBAGENT_CAP = 64 +const TRUNCATION_MARKER = '\n… (truncated)' + +function clip(text: string, cap: number): string { + return text.length > cap ? text.slice(0, cap) + TRUNCATION_MARKER : text +} + +export function sanitizeNativeChatRpcBlock( + block: NativeChatBlock, + clientKind: RpcContext['clientKind'] +): NativeChatBlock { + if (block.type === 'image-ref') { + return sanitizeNativeChatRpcImageBlock(block) + } + if (clientKind !== 'mobile') { + return block + } + if (block.type === 'text') { + return block.text.length > MOBILE_TEXT_BLOCK_CHAR_CAP + ? { ...block, text: clip(block.text, MOBILE_TEXT_BLOCK_CHAR_CAP) } + : block + } + if (block.type === 'tool-result') { + return block.output.length > MOBILE_BLOCK_CHAR_CAP + ? { ...block, output: clip(block.output, MOBILE_BLOCK_CHAR_CAP) } + : block + } + if (block.type === 'tool-call') { + const budget = { remaining: MOBILE_BLOCK_CHAR_CAP, nodes: MOBILE_TOOL_INPUT_NODE_CAP } + return { ...block, input: sanitizeToolInput(block.input, budget, 0) } + } + if (block.type === 'subagent-group') { + return { + ...block, + groupId: clip(block.groupId, MAX_SUBAGENT_FIELD_CHARS), + agents: block.agents.slice(0, MOBILE_SUBAGENT_CAP).map((agent) => ({ + ...agent, + id: clip(agent.id, MAX_SUBAGENT_FIELD_CHARS), + label: clip(agent.label, MAX_SUBAGENT_FIELD_CHARS), + state: clipSubagentState(agent.state) + })) + } + } + return block +} + +/** A state too long to be one this build knows names no state at all, which is + * what `unverifiable` records — clipping it would ship a truncated word. */ +function clipSubagentState(value: NativeChatSubagentState): NativeChatSubagentState { + return value.length > MAX_SUBAGENT_FIELD_CHARS ? normalizeSubagentState(value) : value +} + +function sanitizeToolInput( + value: unknown, + budget: { remaining: number; nodes: number }, + depth: number +): unknown { + budget.nodes-- + if (budget.nodes < 0 || budget.remaining <= 0) { + return '… (truncated)' + } + if (typeof value === 'string') { + const length = Math.min(value.length, budget.remaining) + budget.remaining -= length + return length < value.length ? `${value.slice(0, length)}… (truncated)` : value + } + if (!value || typeof value !== 'object' || depth >= 5) { + return value && typeof value === 'object' ? '… (truncated)' : value + } + if (Array.isArray(value)) { + const result = value + .slice(0, MOBILE_TOOL_INPUT_ITEMS_CAP) + .map((item) => sanitizeToolInput(item, budget, depth + 1)) + if (value.length > MOBILE_TOOL_INPUT_ITEMS_CAP) { + result.push('… (truncated)') + } + return result + } + const result: Record = {} + let count = 0 + for (const key in value) { + if (!Object.hasOwn(value, key)) { + continue + } + if (count >= MOBILE_TOOL_INPUT_ITEMS_CAP || budget.remaining <= 0) { + result['…'] = 'truncated' + break + } + let boundedKey = key.slice(0, Math.min(key.length, budget.remaining, 128)) + // Why: sibling keys sharing a >=128-char (or budget-truncated) prefix collapse + // to the same bounded key; suffix collisions so neither field is silently lost. + if (Object.hasOwn(result, boundedKey)) { + boundedKey = `${boundedKey}~${count}` + } + budget.remaining -= boundedKey.length + result[boundedKey] = sanitizeToolInput( + (value as Record)[key], + budget, + depth + 1 + ) + count++ + } + return result +} diff --git a/src/main/runtime/rpc/methods/native-chat.test.ts b/src/main/runtime/rpc/methods/native-chat.test.ts index 65bb525e798..1417e716770 100644 --- a/src/main/runtime/rpc/methods/native-chat.test.ts +++ b/src/main/runtime/rpc/methods/native-chat.test.ts @@ -266,6 +266,39 @@ describe('nativeChat.readSession clientKind truncation gating', () => { expect(JSON.stringify(input)).toContain('truncated') }) + // The roster block reached mobile through a bare fall-through, uncapped, on the + // one path that exists to keep the payload off the phone. + it('bounds a spawn-group roster before sending it to mobile', async () => { + cachedResult.value = { + messages: [ + { + ...makeMessage('ignored'), + blocks: [ + { + type: 'subagent-group', + groupId: 'thread-1:turn-1', + agents: Array.from({ length: 80 }, (_unused, index) => ({ + id: `child-${index}`, + label: index === 0 ? OVERSIZED : 'read', + state: index === 0 ? (OVERSIZED as 'working') : ('working' as const) + })) + } + ] + } + ] + } + + const result = await readSessionHandler()({ agent: 'codex', sessionId: 's' }, ctxWith('mobile')) + const block = (result as { messages: NativeChatMessage[] }).messages[0].blocks[0] + if (block.type !== 'subagent-group') { + throw new Error('expected a subagent-group block') + } + + expect(block.agents).toHaveLength(64) + expect(block.agents[0].label.length).toBeLessThan(OVERSIZED.length) + expect(block.agents[0].state).toBe('unverifiable') + }) + it('preserves AskUserQuestion option objects at the supported nesting depth', async () => { cachedResult.value = { messages: [ diff --git a/src/main/runtime/rpc/methods/native-chat.ts b/src/main/runtime/rpc/methods/native-chat.ts index 8fc86bf695a..e1a92dd52db 100644 --- a/src/main/runtime/rpc/methods/native-chat.ts +++ b/src/main/runtime/rpc/methods/native-chat.ts @@ -1,9 +1,5 @@ import { z } from 'zod' -import type { - NativeChatBlock, - NativeChatMessage, - AgentType -} from '../../../../shared/native-chat-types' +import type { NativeChatMessage, AgentType } from '../../../../shared/native-chat-types' import { readNativeChatTranscriptTail, subscribeNativeChatTranscript, @@ -11,7 +7,7 @@ import { type SubscribeNativeChatTranscriptArgs } from '../../../native-chat/transcript-watch' import { defineMethod, defineStreamingMethod, type RpcAnyMethod, type RpcContext } from '../core' -import { sanitizeNativeChatRpcImageBlock } from './native-chat-rpc-image-block' +import { sanitizeNativeChatRpcBlock } from './native-chat-rpc-block-sanitize' // Why: native chat renders an agent's own transcript (Claude/Codex JSONL). The // desktop reaches the readers via Electron IPC; mobile/web clients reach the @@ -68,109 +64,15 @@ const NativeChatUnsubscribe = z.object({ // older history as the user scrolls back. const MOBILE_NATIVE_CHAT_DEFAULT_WINDOW = 40 const MOBILE_NATIVE_CHAT_MAX_WINDOW = 2000 -// Why: a single tool result (a big file read, a long diff) can be hundreds of KB. -// The mobile view only previews tool block bodies, so truncate them on the wire -// to keep the payload small; the marker tells the user content was clipped. -const MOBILE_BLOCK_CHAR_CAP = 4000 -// Why: text blocks are the message body itself, rendered in full by the chat -// view — a preview-sized cap cut long assistant replies mid-sentence with no way -// to read on (STA-3230). Keep only a generous safety ceiling: a transcript -// record can legally reach 2MB, and shipping that much markdown in one block -// would freeze the phone. -const MOBILE_TEXT_BLOCK_CHAR_CAP = 64_000 -const MOBILE_TOOL_INPUT_ITEMS_CAP = 20 -const MOBILE_TOOL_INPUT_NODE_CAP = 100 -const TRUNCATION_MARKER = '\n… (truncated)' - -function clip(text: string, cap: number): string { - return text.length > cap ? text.slice(0, cap) + TRUNCATION_MARKER : text -} - -function sanitizeBlock( - block: NativeChatBlock, - clientKind: RpcContext['clientKind'] -): NativeChatBlock { - if (block.type === 'image-ref') { - return sanitizeNativeChatRpcImageBlock(block) - } - if (clientKind !== 'mobile') { - return block - } - if (block.type === 'text') { - return block.text.length > MOBILE_TEXT_BLOCK_CHAR_CAP - ? { ...block, text: clip(block.text, MOBILE_TEXT_BLOCK_CHAR_CAP) } - : block - } - if (block.type === 'tool-result') { - return block.output.length > MOBILE_BLOCK_CHAR_CAP - ? { ...block, output: clip(block.output, MOBILE_BLOCK_CHAR_CAP) } - : block - } - if (block.type === 'tool-call') { - const budget = { remaining: MOBILE_BLOCK_CHAR_CAP, nodes: MOBILE_TOOL_INPUT_NODE_CAP } - return { ...block, input: sanitizeToolInput(block.input, budget, 0) } - } - return block -} - -function sanitizeToolInput( - value: unknown, - budget: { remaining: number; nodes: number }, - depth: number -): unknown { - budget.nodes-- - if (budget.nodes < 0 || budget.remaining <= 0) { - return '… (truncated)' - } - if (typeof value === 'string') { - const length = Math.min(value.length, budget.remaining) - budget.remaining -= length - return length < value.length ? `${value.slice(0, length)}… (truncated)` : value - } - if (!value || typeof value !== 'object' || depth >= 5) { - return value && typeof value === 'object' ? '… (truncated)' : value - } - if (Array.isArray(value)) { - const result = value - .slice(0, MOBILE_TOOL_INPUT_ITEMS_CAP) - .map((item) => sanitizeToolInput(item, budget, depth + 1)) - if (value.length > MOBILE_TOOL_INPUT_ITEMS_CAP) { - result.push('… (truncated)') - } - return result - } - const result: Record = {} - let count = 0 - for (const key in value) { - if (!Object.hasOwn(value, key)) { - continue - } - if (count >= MOBILE_TOOL_INPUT_ITEMS_CAP || budget.remaining <= 0) { - result['…'] = 'truncated' - break - } - let boundedKey = key.slice(0, Math.min(key.length, budget.remaining, 128)) - // Why: sibling keys sharing a >=128-char (or budget-truncated) prefix collapse - // to the same bounded key; suffix collisions so neither field is silently lost. - if (Object.hasOwn(result, boundedKey)) { - boundedKey = `${boundedKey}~${count}` - } - budget.remaining -= boundedKey.length - result[boundedKey] = sanitizeToolInput( - (value as Record)[key], - budget, - depth + 1 - ) - count++ - } - return result -} function sanitizeMessage( message: NativeChatMessage, clientKind: RpcContext['clientKind'] ): NativeChatMessage { - return { ...message, blocks: message.blocks.map((block) => sanitizeBlock(block, clientKind)) } + return { + ...message, + blocks: message.blocks.map((block) => sanitizeNativeChatRpcBlock(block, clientKind)) + } } function sanitizeAppendForClient( diff --git a/src/renderer/src/components/native-chat/NativeChatMessageList.test.tsx b/src/renderer/src/components/native-chat/NativeChatMessageList.test.tsx index 5b71136b86f..cb9ad6932f0 100644 --- a/src/renderer/src/components/native-chat/NativeChatMessageList.test.tsx +++ b/src/renderer/src/components/native-chat/NativeChatMessageList.test.tsx @@ -4,6 +4,11 @@ import '@testing-library/jest-dom/vitest' import { cleanup, fireEvent, render, screen } from '@testing-library/react' import { afterEach, describe, expect, it, vi } from 'vitest' +import { subagentGroupFallbackText } from '../../../../shared/native-chat-subagent-summary' +import type { + NativeChatMessage, + NativeChatSubagentEntry +} from '../../../../shared/native-chat-types' import type { NativeChatLiveSession } from './use-native-chat-live-session' import { NativeChatMessageList } from './NativeChatMessageList' @@ -560,3 +565,293 @@ describe('NativeChatMessageList assistant messages', () => { ) }) }) + +// List-level, because every defect this feature has shipped so far lived in the +// assembly between rows — the roster is its own `role: 'system'` journal row, and +// what reaches the DOM depends on `foldToolMessages`, the turn-key mapping and the +// disclosure state the list owns. Rendering `NativeChatToolRun` in isolation +// supplies those by hand and agrees with whatever the caller was asked to assume. +describe('NativeChatMessageList spawn-group roster', () => { + const ROSTER: NativeChatSubagentEntry[] = [ + { id: 'a', label: 'read', state: 'completed' }, + { id: 'b', label: 'search', state: 'failed' } + ] + + /** The exact two-block row `codexSubagentGroupBody` writes: the structured + * block plus the plain-text twin a client without the block type reads. */ + function rosterMessage(agents: NativeChatSubagentEntry[], at: number): NativeChatMessage { + return { + id: 'roster-1', + role: 'system', + blocks: [ + { type: 'text', text: subagentGroupFallbackText(agents) }, + { type: 'subagent-group', groupId: 'thread-1:turn-1', agents } + ], + timestamp: at, + source: 'transcript' + } + } + + // Explicit ascending timestamps: the list re-sorts by (timestamp, id), so rows + // sharing a millisecond tie-break alphabetically and the user turn can land + // last — which would strand the roster outside its own turn. + function rosterSession( + agents: NativeChatSubagentEntry[], + startedAt: number + ): NativeChatLiveSession { + return { + ...session, + status: 'ready', + messages: [ + { + id: 'user-fanout', + role: 'user', + blocks: [{ type: 'text', text: 'Fan this out' }], + timestamp: startedAt, + source: 'transcript' + }, + { + id: 'assistant-fanout', + role: 'assistant', + blocks: [ + { type: 'tool-call', name: 'shell', input: { command: 'pwd' }, state: 'completed' }, + { type: 'tool-result', output: '/repo' } + ], + timestamp: startedAt + 1, + source: 'transcript' + }, + rosterMessage(agents, startedAt + 2) + ] + } + } + + // A settled turn with its activity collapsed is the resting state of the whole + // transcript, so this is the roster's normal appearance, not an edge case. The + // completed-turn disclosure guard used to swallow it here — the compact row the + // feature exists to leave behind vanished the moment its turn ended. + it('leaves the roster row behind on a settled turn whose activity is collapsed', () => { + const startedAt = Date.now() - 3000 + render( + + ) + + expect(screen.getByRole('button', { name: 'Toggle turn details' })).toHaveAttribute( + 'aria-expanded', + 'false' + ) + expect(screen.getByRole('button', { name: /Ran 2 subagents/ })).toHaveTextContent('1 failed') + // The twin is the roster written out for clients that cannot draw the block. + // This one draws it, so printing the sentence too would say it all twice. + expect(screen.queryByText('Ran 2 subagents (1 failed)')).toBeNull() + }) + + // The block is provider-agnostic — the Claude lane feeds it too — so a lane + // that folds a roster into a message carrying real prose is a live shape. The + // filter used to drop EVERY text block once a roster was present, so that + // prose vanished on desktop while mobile, which reads the raw blocks, kept it. + it('keeps prose beside a roster block and drops only the twin', () => { + const startedAt = Date.now() - 3000 + const twin = subagentGroupFallbackText(ROSTER) + render( + + ) + + expect(screen.getByText('Handing the audit to two children.')).toBeInTheDocument() + expect(screen.getByRole('button', { name: /Ran 2 subagents/ })).toBeInTheDocument() + expect(screen.queryByText(twin)).toBeNull() + }) + + // The reordering that kept the roster visible must not have let TOOL activity + // out from behind the same disclosure: a failed child command reading as live + // on a finished turn is what put that guard there. + it('keeps tool activity behind the disclosure the roster now bypasses', () => { + const startedAt = Date.now() - 3000 + render( + + ) + + expect(screen.queryByRole('button', { name: /1× shell/ })).toBeNull() + fireEvent.click(screen.getByRole('button', { name: 'Toggle turn details' })) + expect(screen.getByRole('button', { name: /1× shell/ })).toBeInTheDocument() + // Expanding must reveal the tools beside the roster, never a second copy of it. + expect(screen.getAllByRole('button', { name: /Ran 2 subagents/ })).toHaveLength(1) + }) + + it('reads as a live spawn while the turn is still working', () => { + render( + + ) + + expect(screen.getByRole('button', { name: /Kicked off 2 subagents/ })).toHaveTextContent( + '2 working' + ) + }) + + // The QA defect, at the seam that produced it. A mid-turn correction opens a + // NEW turn, so `isCurrentTurn` goes false for the fan-out's row and the list + // passes `activeTurnIsWorking={false}` down to the roster. The row used to + // relabel every live child `unverifiable` and flip its headline to "Ran" — + // claiming both that contact was lost and that the fan-out had finished, while + // the three real children were still running and completed 57-87s later. + it('keeps live children working after a newer turn supersedes their own', () => { + const startedAt = Date.now() - 3000 + const live = rosterSession( + [ + { id: 'a', label: 'read_readme', state: 'working', startedAt }, + { id: 'b', label: 'read_package', state: 'working', startedAt } + ], + startedAt + ) + render( + + ) + + const roster = screen.getByRole('button', { name: /Kicked off 2 subagents/ }) + expect(roster).toHaveTextContent('2 working') + expect(roster).not.toHaveTextContent('unverifiable') + expect(screen.queryByRole('button', { name: /Ran 2 subagents/ })).toBeNull() + }) +}) + +// The block schema admits `agents: []`, so a childless spawn group is a shape the +// wire allows even though no producer writes one. It draws nothing, so the row +// must not be mounted on its account: "counts as renderable" and "actually draws" +// have to answer the same. A row that passes the first and fails the second is an +// invisible div that still consumes one `gap-5` slot of the transcript. +describe('NativeChatMessageList childless spawn group', () => { + const NO_AGENTS: NativeChatSubagentEntry[] = [] + + function rosterSession(blocks: NativeChatMessage['blocks'], at: number): NativeChatLiveSession { + return { + ...session, + status: 'ready', + messages: [ + { + id: 'user-fanout', + role: 'user', + blocks: [{ type: 'text', text: 'Fan this out' }], + timestamp: at, + source: 'transcript' + }, + { id: 'roster-1', role: 'system', blocks, timestamp: at + 1, source: 'transcript' } + ] + } + } + + /** Every slot the transcript column lays out — one per row that mounted. */ + function emptySlots(container: HTMLElement): Element[] { + const column = container.querySelector('.max-w-4xl') + expect(column).not.toBeNull() + return Array.from(column!.children).filter((slot) => slot.textContent === '') + } + + it('mounts no row for a bare spawn group with no children', () => { + const startedAt = Date.now() - 3000 + const { container } = render( + + ) + + expect(screen.getByText('Fan this out')).toBeInTheDocument() + expect(emptySlots(container)).toEqual([]) + }) + + it('falls back to the plain-text twin when the block it stands in for cannot draw', () => { + const startedAt = Date.now() - 3000 + const { container } = render( + + ) + + // The twin is dropped only because the block draws the roster instead. This + // one cannot, so suppressing it too would leave the row with nothing at all. + expect(screen.getByText(subagentGroupFallbackText(NO_AGENTS))).toBeInTheDocument() + expect(emptySlots(container)).toEqual([]) + }) +}) diff --git a/src/renderer/src/components/native-chat/NativeChatMessageRow.tsx b/src/renderer/src/components/native-chat/NativeChatMessageRow.tsx index 07ea51b5a62..64f489a1b48 100644 --- a/src/renderer/src/components/native-chat/NativeChatMessageRow.tsx +++ b/src/renderer/src/components/native-chat/NativeChatMessageRow.tsx @@ -4,7 +4,11 @@ import CommentMarkdown, { } from '@/components/sidebar/CommentMarkdown' import { cn } from '@/lib/utils' import { translate } from '@/i18n/i18n' -import type { NativeChatMessage } from '../../../../shared/native-chat-types' +import { + isSubagentGroupFallbackText, + subagentGroupBlocks +} from '../../../../shared/native-chat-subagent-summary' +import { isSubagentGroupBlock, type NativeChatMessage } from '../../../../shared/native-chat-types' import { splitNativeChatBlocks } from './native-chat-tool-fold' import { NativeChatToolRun } from './NativeChatToolRun' import { nativeChatProseToMarkdown } from './native-chat-prose' @@ -47,12 +51,28 @@ export const MessageRow = memo(function MessageRow({ const rowRef = useRef(null) // One pass per block set: a streaming turn re-renders this row on every frame, and these // derivations used to re-run each time even though `message.blocks` had not changed. - const { hasImages, markdown, prose, tools } = useMemo(() => { + const { hasImages, markdown, prose, subagentGroups, tools } = useMemo(() => { const split = splitNativeChatBlocks(message.blocks) + const groups = subagentGroupBlocks(split.prose) + // A spawn-group row carries a plain-text twin so a client without the block + // type still reads the roster. This one draws the block, so the twin is + // dropped rather than printed beside it — only the twin, never the prose + // beside it: the block is provider-agnostic, so a lane that folds a roster + // into a message with real text must not lose that text here. + const prose = + groups.length === 0 + ? split.prose + : split.prose.filter( + (block) => + !isSubagentGroupBlock(block) && + !(block.type === 'text' && isSubagentGroupFallbackText(block.text)) + ) return { - ...split, - markdown: nativeChatProseToMarkdown(split.prose), - hasImages: split.prose.some((block) => block.type === 'image-ref') + tools: split.tools, + prose, + subagentGroups: groups, + markdown: nativeChatProseToMarkdown(prose), + hasImages: prose.some((block) => block.type === 'image-ref') } }, [message.blocks]) const isUser = message.role === 'user' @@ -69,7 +89,7 @@ export const MessageRow = memo(function MessageRow({ // Skip rows with nothing renderable so the transcript shows no empty/ghost // bubble. // After all hooks, so hook order stays unconditional. - if (markdown.length === 0 && !hasImages && tools.length === 0) { + if (markdown.length === 0 && !hasImages && tools.length === 0 && subagentGroups.length === 0) { return null } @@ -151,9 +171,10 @@ export const MessageRow = memo(function MessageRow({ linkifyFilePaths={onLinkClick !== undefined} /> ) : null} - {tools.length > 0 ? ( + {tools.length > 0 || subagentGroups.length > 0 ? ( { + it('reads as a live spawn while children work', () => { + render( + + ) + + expect(screen.getByText('Kicked off 2 subagents')).toBeInTheDocument() + expect(screen.getByRole('button')).toHaveTextContent('1 working') + expect(screen.getByRole('button')).toHaveTextContent('40.7k tokens') + }) + + it('switches to Ran once every child completed', () => { + render( + + ) + + expect(screen.getByText('Ran 2 subagents')).toBeInTheDocument() + expect(screen.getByRole('button')).toHaveTextContent('completed') + }) + + it('shows the worst settled verdict, not the count of finished children', () => { + render( + + ) + + expect(screen.getByRole('button')).toHaveTextContent('2 failed') + }) + + it('surfaces a failed child while its siblings still work', () => { + const { container } = render( + + ) + + const row = screen.getByRole('button') + expect(row).toHaveTextContent('3 working') + expect(row).toHaveTextContent('+1 failed') + // The dot carries the failure; the pulse still says the group is in flight. + expect(container.querySelector('.bg-destructive.animate-pulse')).not.toBeNull() + }) + + it('leaves the dot neutral when nothing has gone wrong', () => { + const { container } = render( + + ) + + expect(screen.getByRole('button')).not.toHaveTextContent('failed') + expect(container.querySelector('.bg-destructive')).toBeNull() + }) + + // The QA defect: a mid-turn correction opened a new turn while three real + // children were still running, and the row relabelled every one of them + // `unverifiable` and flipped its headline to `Ran`. The children completed + // 57-87s later. A turn boundary says nothing about a child. + it('keeps a working child working once its turn is no longer the current one', () => { + render() + + const row = screen.getByRole('button') + expect(row).toHaveTextContent('working') + expect(row).not.toHaveTextContent('unverifiable') + expect(screen.getByText('Kicked off 1 subagent')).toBeInTheDocument() + }) + + it('reports the verdict a child lands after its turn ended', () => { + render( + + ) + + expect(screen.getByText('Ran 1 subagent')).toBeInTheDocument() + expect(screen.getByRole('button')).toHaveTextContent('completed') + }) + + // Only the writing host may claim loss of contact, and it writes that verdict + // into the row itself. The renderer draws it, and never infers it. + it('draws the unverifiable verdict the host recorded', () => { + render( + + ) + + expect(screen.getByRole('button')).toHaveTextContent('unverifiable') + }) + + it('leads with the bot glyph, decorative beside the word that names the group', () => { + const { container } = render( + + ) + + const glyph = container.querySelector('.lucide-bot') + expect(glyph).not.toBeNull() + expect(glyph).toHaveAttribute('aria-hidden', 'true') + // Never icon-only: the word is what carries the accessible name. + expect(screen.getByRole('button')).toHaveAccessibleName(/Kicked off 1 subagent/) + }) + + it('keeps the same glyph in every state, so a settling row never changes identity', () => { + const states: NativeChatSubagentState[] = [ + 'working', + 'idle', + 'completed', + 'failed', + 'stopped', + 'unverifiable' + ] + + for (const state of states) { + const { container } = render( + + ) + + expect(container.querySelectorAll('.lucide-bot')).toHaveLength(1) + expect(container.querySelector('.lucide-check')).toBeNull() + expect(container.querySelector('.lucide-users')).toBeNull() + cleanup() + } + }) + + // The only aria-hidden span carrying text is the elapsed-clock wrapper: the + // glyph's Bot is an and the status dots render empty. + function hiddenTextSpans(container: HTMLElement): Element[] { + return [...container.querySelectorAll('span[aria-hidden="true"]')].filter( + (element) => (element.textContent ?? '').trim().length > 0 + ) + } + + it('keeps the ticking clock out of the live region until it stops moving', () => { + const { container } = render( + + ) + + const row = screen.getByRole('button') + expect(row).toHaveAttribute('aria-live', 'polite') + // A clock that reticks every second would announce a new duration every + // second and bury the state changes the live region exists to report. + expect(hiddenTextSpans(container)).toHaveLength(1) + }) + + it('reads the elapsed time out once it has stopped moving', () => { + const { container } = render( + + ) + + // Settled: the duration is fixed, so hiding it would cost a reader real + // information for no announcement churn. + expect(hiddenTextSpans(container)).toHaveLength(0) + expect(screen.getByRole('button')).toHaveTextContent('4s') + }) + + it('shows no duration for a child whose run length was never recorded', () => { + render( + + ) + + const row = screen.getByRole('button') + expect(row).toHaveTextContent('unverifiable') + // `unverifiable` with no terminal timestamp has no known run length, so the + // clock would measure to `now` and report the time since we lost sight of + // the child as how long it ran — on a row that is not even counting. + expect(row.textContent).not.toContain('·') + }) + + // A partial sweep leaves one child settled and one whose fate is unknown. The + // group's clock would then report the settled sibling's duration as the + // group's run length while the other child is still unaccounted for. + it('shows no duration while one child settled and another is unaccounted for', () => { + render( + + ) + + const row = screen.getByRole('button') + expect(row).toHaveTextContent('unverifiable') + expect(row.textContent).not.toContain('·') + }) +}) + +describe('NativeChatToolRun with a spawn group', () => { + it('renders a roster with no tool calls without inventing a tool count', () => { + render( + + ) + + expect(screen.getByText('Kicked off 1 subagent')).toBeInTheDocument() + expect(screen.queryByText('1 tool call')).toBeNull() + }) + + // Every settled turn sits here by default: the list passes + // `expandOverride={expandedTurnIds.has(turnKey)}` — false until the reader + // opens that turn — and `activeTurnIsWorking={false}`. The completed-turn + // guard above bailed before the roster branch, so the one row this feature + // exists to draw vanished the moment its turn finished, and the message row + // that kept itself alive for it rendered an empty ghost bubble. + it('keeps the roster visible on a completed turn whose activity is collapsed', () => { + render( + + ) + + expect(screen.getByText('Ran 1 subagent')).toBeInTheDocument() + }) + + // The roster-only branch returns a `mt-3` wrapper whenever it has rows, so a + // group that draws nothing must not count as one — that wrapper would be the + // empty bubble with a margin that the message row refuses to emit. + it('draws nothing at all for a spawn group that carries no children', () => { + const { container } = render( + + ) + + expect(container).toBeEmptyDOMElement() + }) + + // The roster-only escape above is keyed on `blocks.length === 0`, so a group + // sharing its message with tool calls falls through to the settled-turn guard + // — which returned bare null and took the roster with it. + it('keeps a roster that shares its message with tool calls on a collapsed turn', () => { + render( + + ) + + expect(screen.getByText('Ran 1 subagent')).toBeInTheDocument() + expect(screen.queryByText('shell ls')).toBeNull() + }) + + it('renders the roster alongside the tool activity of its turn', () => { + render( + + ) + + expect(screen.getByText('Ran 1 subagent')).toBeInTheDocument() + expect(screen.getByText('shell ls')).toBeInTheDocument() + }) +}) diff --git a/src/renderer/src/components/native-chat/NativeChatSubagentRun.tsx b/src/renderer/src/components/native-chat/NativeChatSubagentRun.tsx new file mode 100644 index 00000000000..af3bf468011 --- /dev/null +++ b/src/renderer/src/components/native-chat/NativeChatSubagentRun.tsx @@ -0,0 +1,277 @@ +import { useMemo, useState } from 'react' +import { Bot, ChevronRight } from 'lucide-react' +import { cn } from '@/lib/utils' +import { translate } from '@/i18n/i18n' +import { useNow } from '@/hooks/use-now' +import { + normalizeSubagentState, + summarizeSubagentGroup +} from '../../../../shared/native-chat-subagent-summary' +import type { + NativeChatSubagentGroupBlock, + NativeChatSubagentState +} from '../../../../shared/native-chat-types' +import { formatNativeChatDuration } from './NativeChatWorkingStatus' + +/** Compact token counts: the row shows scale, not an exact ledger. */ +function formatSubagentTokens(tokens: number): string { + if (tokens < 1_000) { + return String(Math.round(tokens)) + } + const scaled = tokens < 1_000_000 ? tokens / 1_000 : tokens / 1_000_000 + const suffix = tokens < 1_000_000 ? 'k' : 'M' + return `${scaled.toFixed(1).replace(/\.0$/, '')}${suffix}` +} + +/** The group's one-line verdict. A single-child group reads as a bare word; any + * larger group always carries the count, because "working" alone would not say + * how many of the children it covers. `completed` never takes one: every child + * finishing is the whole group finishing. */ +function subagentStateLabel( + state: NativeChatSubagentState, + count: number, + groupTotal: number +): string { + if (state === 'completed') { + return translate('components.native-chat.subagents.state.completed', 'completed') + } + if (groupTotal <= 1) { + switch (state) { + case 'working': + return translate('components.native-chat.subagents.state.working', 'working') + case 'idle': + return translate('components.native-chat.subagents.state.idle', 'idle') + case 'failed': + return translate('components.native-chat.subagents.state.failed', 'failed') + case 'stopped': + return translate('components.native-chat.subagents.state.stopped', 'stopped') + case 'unverifiable': + return translate('components.native-chat.subagents.state.unverifiable', 'unverifiable') + } + } + switch (state) { + case 'working': + return translate( + 'components.native-chat.subagents.state.workingCount', + '{{value0}} working', + { + value0: count + } + ) + case 'idle': + return translate('components.native-chat.subagents.state.idleCount', '{{value0}} idle', { + value0: count + }) + case 'failed': + return translate('components.native-chat.subagents.state.failedCount', '{{value0}} failed', { + value0: count + }) + case 'stopped': + return translate( + 'components.native-chat.subagents.state.stoppedCount', + '{{value0}} stopped', + { + value0: count + } + ) + case 'unverifiable': + return translate( + 'components.native-chat.subagents.state.unverifiableCount', + '{{value0}} unverifiable', + { value0: count } + ) + } +} + +const STATE_DOT_CLASS: Record = { + working: 'bg-foreground/70', + idle: 'bg-muted-foreground/40', + completed: 'bg-muted-foreground/60', + failed: 'bg-destructive', + stopped: 'bg-muted-foreground', + unverifiable: 'bg-muted-foreground' +} + +/** + * The group's identity glyph, fixed across every state — a settling row must not + * appear to change identity. State is carried by {@link StatusDot} and the tone + * of the words beside it. + * + * SWAP POINT: once the shared category-icon component lands (PR #18760), this + * whole component becomes that component asked for the `bot` category, which is + * the same glyph the individual `subAgentActivity` rows use. + */ +function SubagentGlyph(): React.JSX.Element { + return ( + + + ) +} + +/** `pulsing` is separate from `state` so a group that is still working can show + * a failed sibling's colour without losing its in-flight cue. */ +function StatusDot({ + state, + pulsing = false +}: { + state: NativeChatSubagentState + pulsing?: boolean +}): React.JSX.Element { + return ( +