mirror of
https://github.com/stablyai/orca.git
synced 2026-09-22 00:02:31 +00:00
feat(native-chat): show Claude subagent activity on the shared carrier (#18806)
* feat(native-chat): show Claude subagent activity on the shared carrier Claude's `message:system:task_*` frames are classified `status-chrome` and reach the transcript as nothing at all, so a turn that spawns subagents renders as an idle turn. The journal translator now reads them into the shared subagent-group carrier — no new UI, and the frames stay `status-chrome` so nothing prints a raw opcode row. `local_agent`, `local_workflow` and `local_bash` tasks share that channel and all carry a `tool_use_id`, so `task_type` is the discriminator and a backgrounded `sleep 20` stays out of the roster; `subagent_type` covers releases that predate `task_type`. `skip_transcript` tasks never render, `is_backgrounded` children survive the turn-end sweep, and a resumed task re-announced under a fresh tool id is aliased onto its `task_id` rather than duplicated. A child still reported as working when the turn — or the session — ends becomes `unverifiable`: contact was lost, which is not evidence it exited. * fix(native-chat): stop the Claude subagent roster dropping its own rows The roster published under the same coalescing key it appends the row with, and the sink queue replaces any queued operation sharing a key regardless of kind: once a write was in flight, each new append evicted the pending publish and the next publish evicted that append, so the body never reached the journal and `lastSerialized` had already moved past it. Publish now takes the sink's own slot, as the Codex streams do. A tombstoned row could never come back: the non-batch item-row builder derived its revision from `items` alone, so a re-add was built at revision 1 against a tombstone at 2 and the reducer discarded it forever. It now takes the same `max(items, tombstones)` the batch builder already used — reachable here because an announcement that reveals a `local_bash` task empties and tombstones the group row that a genuine subagent later in the turn reuses. `settleTurn` swept whatever group the key named at the time it ran, so children rostered before any turn key existed were never swept, and a turn whose result never arrives was left working forever. The ending turn's key is now an argument, a superseding turn start settles the turn it replaces, and every turn end also sweeps the outside-turn group. Teardown without an `ended` event, and eviction past the group bound, both lose contact instead of stranding a row at `working`. Label ordinals are a high-water mark now: releasing one on a re-label handed the next child an ordinal that was already on screen. * fix(native-chat): bound subagent-group blocks on every wire that carries one Adding a fifth arm to `NativeChatBlock` made every consumer that assumed four wrong. Two of them ended in `return block`, so they compiled while handing a roster straight through: the mobile RPC sanitizer shipped it unclipped past both mobile char caps, and the legacy transcript import stored an untrusted roster unbounded. Both now clip each label and cap the entry count the way they bound their other blocks. The remaining three sites did not compile at all. The worker transcript payload and the live-session benchmark get real arms rather than casts — a cast would have turned the transcript one into a third silent passthrough inside the wire byte budget — and the CLI worker output renders a roster with its shared summary instead of `[image omitted]`. The mobile sanitizer moves to a sibling module beside the image-block one: the file sat exactly on the max-lines bound, and the block bounds are a self-contained concern with their own caps. Also caps the roster's `subagent_type` label fallback, which reached the journal uncapped, and covers the new block type in the schema audit. * test(native-chat): cover the capped subagent_type label The roster stores the frame's label verbatim, so the cap on the `subagent_type` fallback is the only thing bounding it. * fix(native-chat): type the roster fixture so the suite typechecks The mobile-cap test built its entries with an inferred `state: string`, which is not a `NativeChatSubagentState` — the only typecheck failure on the branch. * fix(native-chat): stop child traffic rostering an id Claude never announced `observeChildActivity` minted a provisional row for any `parent_tool_use_id` outside the excluded set. An id that was never announced is never excluded, so a nested Task, a workflow child, or a grandchild parented to a tool id inside the sidechain each produced a permanently unlabelled `subagent` row that could only ever end `unverifiable`. The bounded exclusion set cannot cover an id no frame ever declared, and in a long session it can forget a genuine exclusion. Track instead whether this CLI announces tasks at all — set by ANY `task_started`, including one the subagent filter rejects. Once it has, an undeclared child is provably not a new subagent, so no row is created. The provisional path now serves only releases that announce no task frames, which is what its comment already said it was for. The label-ordinal test moves to an announcement-driven removal, the scenario that path now actually reaches; it still fails if `remove` releases the ordinal. * fix(native-chat): outrank the tombstone when building one too `buildJournalTombstoneRow` still built its revision from `items` alone, leaving it asymmetric with the item builder. It is correct today only because `upsertItem` clears the tombstone whenever a re-add wins — an invariant that lives in the reducer and was not pinned. Apply the same `Math.max`, and pin the invariant so the reducer cannot drop it silently. * fix(native-chat): stop an unrelated turn end settling an outside-turn child `settleTurn` swept the `outside-turn` group on every turn end, so a child Claude announced while no turn was live — a frame trailing the previous turn's result, or one that arrives before the first turn starts — was marked `unverifiable` by the next, unrelated turn ending. That state is terminal and latches, so the `task_updated: completed` that followed was discarded: loss of contact was recorded as the child's outcome on evidence that was never about that child. A turn end now sweeps exactly the group its key names. `outside-turn` belongs to no turn, so only an end with no key of its own reaches it, and what no turn end reaches `settleSession` does — reliably, since teardown without an `ended` event also routes through it. The cost is a child outside every turn showing `working` a little longer; the alternative prints a wrong outcome that nothing can revise. Also pins that a subagent announced after a task the filter rejected still rosters: the announcement path was never what the child-traffic gate closes. * fix(native-chat): bound a subagent entry's id, not just its label Every site that bounds a `subagent-group` block clipped the label and handed the id through whole. From the Claude producer the id is bounded upstream, but the legacy transcript import reads an untrusted file, so an oversized id survived into the journal and then out to every wire that replays it — 64 entries of it, since only the entry count was capped. Each site now clips the id with the helper it already uses for its other bounded fields: the journal's inline-text bound on import, the transcript payload's metadata clip, and the mobile char cap (renamed, since it is no longer a label-only cap). * fix(native-chat): surface an adverse subagent outcome in the fallback sentence The roster row's plain-text stand-in counted only `working`, so a fan-out whose children all latched `unverifiable` (or `failed`, or `stopped`) rendered as "Ran 3 subagents" — a completion claim. Mobile and paired web have no roster renderer, so that write-time-frozen sentence is the entire row there, and collapsing `unverifiable` into something that reads like success is exactly what the SSH execution boundary forbids. It now appends the worst adverse count, worst-first across failed/stopped/ unverifiable, and shows it even while siblings still work — matching the Codex lane's shared `subagentGroupFallbackText` verbatim so collapsing the two copies later is a deletion, not a behaviour change. Also bounds the provisional entry id. `observeChildActivity` wrote the `parent_tool_use_id` straight into the entry's durable id with no length cap, while the announced path already rejects an over-long id via `claudeTaskId`. Both now share `isBoundedClaudeTaskId`, and the provisional path rejects rather than truncates, as the announced one does. * fix(native-chat): stop a subagent label ordinal and a clipped roster key colliding - claimLabel probes the labels the group actually rendered instead of a per-base counter, so a generated `Audit 2` cannot duplicate a provider's own `Audit 2`. - Bound `NativeChatSubagentEntry.id` with a head plus a digest of the whole id at every site that bounds it. The id is the roster key: a prefix clip merged two distinct children onto one entry. - Correct a stale journal-reducer test comment: tombstone cleanup is a map-state invariant, no longer load-bearing for revision ordering. * fix(claude): merge duplicate unhandled-provider-frame imports * fix(claude): preserve subagent lifecycle and bounded invocation identity --------- Co-authored-by: Merge Sim <sim@local>
This commit is contained in:
Brennan Benson
co-authored by
Merge Sim
Merge Sim
parent
852495d35c
commit
72befaf360
@@ -117,7 +117,7 @@ function blockContent(message: NativeChatMessage): string {
|
||||
if (block.type === 'tool-result') {
|
||||
return block.output
|
||||
}
|
||||
return block.path ?? block.url ?? block.alt ?? ''
|
||||
return block.type === 'image-ref' ? (block.path ?? block.url ?? block.alt ?? '') : block.groupId
|
||||
}
|
||||
|
||||
function messageWeight(message: NativeChatMessage, content: string): number {
|
||||
|
||||
@@ -20,14 +20,22 @@ function record(value: unknown): Record<string, unknown> | null {
|
||||
return typeof value === 'object' && value !== null ? (value as Record<string, unknown>) : null
|
||||
}
|
||||
|
||||
function taskId(message: Record<string, unknown>): string | null {
|
||||
const value = message.task_id
|
||||
return typeof value === 'string' && value.length > 0 && value.length <= MAX_TASK_ID_LENGTH
|
||||
? value
|
||||
: null
|
||||
/** The bound every task id shares, wherever it enters. An id the roster stores
|
||||
* becomes a durable entry key, so a provisional one takes the same bound the
|
||||
* announced path applies — an over-long id is rejected, never truncated. */
|
||||
export function isBoundedClaudeTaskId(value: string): boolean {
|
||||
return value.length > 0 && value.length <= MAX_TASK_ID_LENGTH
|
||||
}
|
||||
|
||||
function taskDescription(value: unknown): string | undefined {
|
||||
/** The task's canonical, resume-stable id. Shared with the subagent roster so
|
||||
* both readers of this channel agree on what identifies a task. */
|
||||
export function claudeTaskId(message: Record<string, unknown>): string | null {
|
||||
const value = message.task_id
|
||||
return typeof value === 'string' && isBoundedClaudeTaskId(value) ? value : null
|
||||
}
|
||||
|
||||
/** A task's human label, collapsed and bounded. */
|
||||
export function claudeTaskDescription(value: unknown): string | undefined {
|
||||
if (typeof value !== 'string') {
|
||||
return undefined
|
||||
}
|
||||
@@ -107,7 +115,7 @@ export class ClaudeBackgroundTaskTracker {
|
||||
this.replaceAggregateRoster(message.tasks)
|
||||
return true
|
||||
}
|
||||
const id = taskId(message)
|
||||
const id = claudeTaskId(message)
|
||||
if (!id) {
|
||||
return false
|
||||
}
|
||||
@@ -126,13 +134,13 @@ export class ClaudeBackgroundTaskTracker {
|
||||
}
|
||||
const existing = this.tasks.get(id)
|
||||
if (
|
||||
(patch.is_backgrounded === true || taskDescription(patch.description)) &&
|
||||
(patch.is_backgrounded === true || claudeTaskDescription(patch.description)) &&
|
||||
(!this.aggregateRosterObserved || existing)
|
||||
) {
|
||||
this.upsert(id, {
|
||||
backgrounded: patch.is_backgrounded === true || existing?.backgrounded === true,
|
||||
kind: existing?.kind ?? 'unknown',
|
||||
description: taskDescription(patch.description) ?? existing?.description
|
||||
description: claudeTaskDescription(patch.description) ?? existing?.description
|
||||
})
|
||||
return true
|
||||
}
|
||||
@@ -152,7 +160,7 @@ export class ClaudeBackgroundTaskTracker {
|
||||
this.upsert(id, {
|
||||
backgrounded: message.is_backgrounded === true || kind === 'workflow' || kind === 'monitor',
|
||||
kind,
|
||||
description: taskDescription(message.description)
|
||||
description: claudeTaskDescription(message.description)
|
||||
})
|
||||
return true
|
||||
}
|
||||
@@ -172,14 +180,14 @@ export class ClaudeBackgroundTaskTracker {
|
||||
if (!task || task.ambient === true) {
|
||||
continue
|
||||
}
|
||||
const id = taskId(task)
|
||||
const id = claudeTaskId(task)
|
||||
if (!id) {
|
||||
continue
|
||||
}
|
||||
this.tasks.set(id, {
|
||||
backgrounded: true,
|
||||
kind: classifyClaudeBackgroundTaskKind(task.task_type),
|
||||
description: taskDescription(task.description)
|
||||
description: claudeTaskDescription(task.description)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -74,6 +74,18 @@ export function claudeMessageIdentity(
|
||||
return { provider: 'claude', sessionId: envelope.sessionId, uuid: envelope.uuid }
|
||||
}
|
||||
|
||||
/** User bubbles belong to the submitted message; SDK user frames carry echoes
|
||||
* and tool results, so a user envelope keeps only its tool results. */
|
||||
export function claudeOutputEnvelope(envelope: ClaudeMessageEnvelope): ClaudeMessageEnvelope {
|
||||
if (envelope.role !== 'user') {
|
||||
return envelope
|
||||
}
|
||||
return {
|
||||
...envelope,
|
||||
content: envelope.content.filter((part) => claudeRecord(part)?.type === 'tool_result')
|
||||
}
|
||||
}
|
||||
|
||||
function messageBlocks(envelope: ClaudeMessageEnvelope): NativeChatBlock[] {
|
||||
const blocks: NativeChatBlock[] = []
|
||||
for (const value of envelope.content) {
|
||||
|
||||
@@ -0,0 +1,259 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import type {
|
||||
AgentJournalItemBody,
|
||||
AgentJournalItemIdentity
|
||||
} from '../../shared/agent-session-journal-types'
|
||||
import type {
|
||||
NativeChatSubagentEntry,
|
||||
NativeChatSubagentGroupBlock
|
||||
} from '../../shared/native-chat-types'
|
||||
import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink'
|
||||
import { createClaudeJournalTranslator } from './claude-structured-journal-translation'
|
||||
|
||||
const GROUP_ITEM_ID = 'claude-subagents:claude-session:user-1'
|
||||
|
||||
/** The union's other arms carry no client message id, so reading one narrows. */
|
||||
function orcaClientMessageId(identity: AgentJournalItemIdentity): string | null {
|
||||
return identity.provider === 'orca' ? identity.clientMessageId : null
|
||||
}
|
||||
|
||||
function harness() {
|
||||
const items: { identity: AgentJournalItemIdentity; body: AgentJournalItemBody }[] = []
|
||||
const sink: StructuredAgentSessionEventSink = {
|
||||
appendItem: (identity, body) => items.push({ identity, body }),
|
||||
appendTombstone: vi.fn(),
|
||||
publish: vi.fn()
|
||||
}
|
||||
const translator = createClaudeJournalTranslator({ sink, fallbackIdPrefix: 'test' })
|
||||
const groupRows = () =>
|
||||
items.filter((item) => orcaClientMessageId(item.identity) === GROUP_ITEM_ID)
|
||||
const agentsOf = (body: AgentJournalItemBody | undefined): NativeChatSubagentEntry[] => {
|
||||
if (!body || body.kind !== 'message') {
|
||||
return []
|
||||
}
|
||||
const block = body.blocks.find(
|
||||
(candidate): candidate is NativeChatSubagentGroupBlock => candidate.type === 'subagent-group'
|
||||
)
|
||||
return block ? block.agents : []
|
||||
}
|
||||
/** The last roster row written for one group, so a test can read a group that
|
||||
* is no longer the live one. */
|
||||
const rosterIn = (groupId: string): NativeChatSubagentEntry[] =>
|
||||
agentsOf(
|
||||
items.findLast((item) => orcaClientMessageId(item.identity) === `claude-subagents:${groupId}`)
|
||||
?.body
|
||||
)
|
||||
const rosterOf = (turnUuid: string): NativeChatSubagentEntry[] =>
|
||||
rosterIn(`claude-session:${turnUuid}`)
|
||||
const roster = (): NativeChatSubagentEntry[] => agentsOf(groupRows().at(-1)?.body)
|
||||
const fallbackRows = (): AgentJournalItemBody[] =>
|
||||
items
|
||||
.filter((item) => (orcaClientMessageId(item.identity) ?? '').startsWith('provider-frame:'))
|
||||
.map((item) => item.body)
|
||||
return { translator, groupRows, roster, rosterIn, rosterOf, fallbackRows }
|
||||
}
|
||||
|
||||
function userTurn(uuid: string) {
|
||||
return {
|
||||
type: 'message' as const,
|
||||
sessionId: 'orca-session',
|
||||
startsTurn: true as const,
|
||||
message: {
|
||||
type: 'user',
|
||||
uuid,
|
||||
session_id: 'claude-session',
|
||||
parent_tool_use_id: null,
|
||||
message: { role: 'user', content: [{ type: 'text', text: 'go' }] }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function systemFrame(subtype: string, fields: Record<string, unknown>) {
|
||||
return {
|
||||
type: 'message' as const,
|
||||
sessionId: 'orca-session',
|
||||
message: { type: 'system', subtype, session_id: 'claude-session', ...fields }
|
||||
}
|
||||
}
|
||||
|
||||
function spawnResult(uuid: string, toolUseId: string) {
|
||||
return {
|
||||
type: 'message' as const,
|
||||
sessionId: 'orca-session',
|
||||
message: {
|
||||
type: 'user',
|
||||
uuid,
|
||||
session_id: 'claude-session',
|
||||
parent_tool_use_id: null,
|
||||
message: {
|
||||
role: 'user',
|
||||
content: [{ type: 'tool_result', tool_use_id: toolUseId, content: 'done' }]
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function resultFrame() {
|
||||
return {
|
||||
type: 'message' as const,
|
||||
sessionId: 'orca-session',
|
||||
message: {
|
||||
type: 'result',
|
||||
subtype: 'success',
|
||||
session_id: 'claude-session',
|
||||
uuid: 'result-1',
|
||||
result: 'ok'
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
describe('claude journal translation — subagents', () => {
|
||||
it('rosters a spawned subagent and settles it on the spawn call result', () => {
|
||||
const { translator, roster, fallbackRows } = harness()
|
||||
translator.handle(userTurn('user-1'))
|
||||
translator.handle(
|
||||
systemFrame('task_started', {
|
||||
task_id: 'task-1',
|
||||
tool_use_id: 'toolu_1',
|
||||
task_type: 'local_agent',
|
||||
subagent_type: 'explorer',
|
||||
description: 'Map the lane'
|
||||
})
|
||||
)
|
||||
expect(roster()).toEqual([
|
||||
expect.objectContaining({ id: 'task-1', label: 'Map the lane', state: 'working' })
|
||||
])
|
||||
// The task frames stay status-chrome, so none of them prints an opcode row.
|
||||
expect(fallbackRows()).toEqual([])
|
||||
translator.handle(spawnResult('user-2', 'toolu_1'))
|
||||
expect(roster()).toEqual([expect.objectContaining({ state: 'completed' })])
|
||||
})
|
||||
|
||||
it('marks a child still working at turn end unverifiable', () => {
|
||||
const { translator, roster } = harness()
|
||||
translator.handle(userTurn('user-1'))
|
||||
translator.handle(
|
||||
systemFrame('task_started', {
|
||||
task_id: 'task-1',
|
||||
task_type: 'local_agent',
|
||||
description: 'Map the lane'
|
||||
})
|
||||
)
|
||||
translator.handle(resultFrame())
|
||||
expect(roster()).toEqual([expect.objectContaining({ state: 'unverifiable' })])
|
||||
})
|
||||
|
||||
it('leaves a backgrounded child running past the end of its turn', () => {
|
||||
const { translator, roster } = harness()
|
||||
translator.handle(userTurn('user-1'))
|
||||
translator.handle(
|
||||
systemFrame('task_started', {
|
||||
task_id: 'task-1',
|
||||
tool_use_id: 'toolu_1',
|
||||
task_type: 'local_agent',
|
||||
description: 'Watch the build',
|
||||
is_backgrounded: true
|
||||
})
|
||||
)
|
||||
// A backgrounded spawn returns its tool result immediately; the child runs on.
|
||||
translator.handle(spawnResult('user-2', 'toolu_1'))
|
||||
translator.handle(resultFrame())
|
||||
expect(roster()).toEqual([expect.objectContaining({ state: 'working' })])
|
||||
translator.handle({ type: 'ended', sessionId: 'orca-session', reason: 'closed' })
|
||||
expect(roster()).toEqual([expect.objectContaining({ state: 'unverifiable' })])
|
||||
})
|
||||
|
||||
it('keeps a backgrounded shell task out of the roster entirely', () => {
|
||||
const { translator, groupRows } = harness()
|
||||
translator.handle(userTurn('user-1'))
|
||||
translator.handle(
|
||||
systemFrame('task_started', {
|
||||
task_id: 'task-bash',
|
||||
tool_use_id: 'toolu_bash',
|
||||
task_type: 'local_bash',
|
||||
description: 'sleep 20',
|
||||
is_backgrounded: true
|
||||
})
|
||||
)
|
||||
translator.handle(resultFrame())
|
||||
expect(groupRows()).toEqual([])
|
||||
})
|
||||
|
||||
it('shows a subagent whose release announces no task frames, from its child traffic', () => {
|
||||
const { translator, roster } = harness()
|
||||
translator.handle(userTurn('user-1'))
|
||||
translator.handle({
|
||||
type: 'message' as const,
|
||||
sessionId: 'orca-session',
|
||||
message: {
|
||||
type: 'assistant',
|
||||
uuid: 'child-1',
|
||||
session_id: 'claude-session',
|
||||
parent_tool_use_id: 'toolu_1',
|
||||
message: { role: 'assistant', content: [{ type: 'text', text: 'looking' }] }
|
||||
}
|
||||
})
|
||||
expect(roster()).toEqual([
|
||||
expect.objectContaining({ id: 'toolu_1', label: 'subagent', state: 'working' })
|
||||
])
|
||||
})
|
||||
|
||||
it('settles the turn a new turn superseded, and leaves the new one running', () => {
|
||||
const { translator, rosterOf } = harness()
|
||||
translator.handle(userTurn('user-1'))
|
||||
translator.handle(
|
||||
systemFrame('task_started', {
|
||||
task_id: 'task-1',
|
||||
task_type: 'local_agent',
|
||||
description: 'First turn'
|
||||
})
|
||||
)
|
||||
// A second turn starts with no result frame for the first: the first turn
|
||||
// ends here, and nothing else will ever name its group again.
|
||||
translator.handle(userTurn('user-2'))
|
||||
translator.handle(
|
||||
systemFrame('task_started', {
|
||||
task_id: 'task-2',
|
||||
task_type: 'local_agent',
|
||||
description: 'Second turn'
|
||||
})
|
||||
)
|
||||
expect(rosterOf('user-1')).toEqual([expect.objectContaining({ state: 'unverifiable' })])
|
||||
expect(rosterOf('user-2')).toEqual([expect.objectContaining({ state: 'working' })])
|
||||
})
|
||||
|
||||
it('does not let an unrelated turn end settle a child announced outside a turn', () => {
|
||||
const { translator, rosterIn } = harness()
|
||||
// No turn is live yet, so this child has no turn key to belong to.
|
||||
translator.handle(
|
||||
systemFrame('task_started', {
|
||||
task_id: 'task-early',
|
||||
task_type: 'local_agent',
|
||||
description: 'Before the turn'
|
||||
})
|
||||
)
|
||||
translator.handle(userTurn('user-1'))
|
||||
translator.handle(resultFrame())
|
||||
expect(rosterIn('outside-turn')).toEqual([expect.objectContaining({ state: 'working' })])
|
||||
// The outcome still lands, which a latched `unverifiable` would have lost.
|
||||
translator.handle(
|
||||
systemFrame('task_updated', { task_id: 'task-early', patch: { status: 'completed' } })
|
||||
)
|
||||
expect(rosterIn('outside-turn')).toEqual([expect.objectContaining({ state: 'completed' })])
|
||||
})
|
||||
|
||||
it('settles a child left outside every turn when the session ends', () => {
|
||||
const { translator, rosterIn } = harness()
|
||||
translator.handle(
|
||||
systemFrame('task_started', {
|
||||
task_id: 'task-early',
|
||||
task_type: 'local_agent',
|
||||
description: 'Before the turn'
|
||||
})
|
||||
)
|
||||
translator.handle(userTurn('user-1'))
|
||||
translator.handle(resultFrame())
|
||||
translator.handle({ type: 'ended', sessionId: 'orca-session', reason: 'closed' })
|
||||
expect(rosterIn('outside-turn')).toEqual([expect.objectContaining({ state: 'unverifiable' })])
|
||||
})
|
||||
})
|
||||
@@ -11,9 +11,8 @@ import {
|
||||
claudeMessageBody,
|
||||
claudeMessageIdentity,
|
||||
claudeHasReplayContent,
|
||||
claudeRecord,
|
||||
claudeOutputEnvelope,
|
||||
claudeStreamingMessageBody,
|
||||
claudeText,
|
||||
claudeThinkingIdentity,
|
||||
claudeThinkingText,
|
||||
claudeToolBody,
|
||||
@@ -29,16 +28,15 @@ import {
|
||||
claudeQuestionItems
|
||||
} from './claude-structured-prompt-items'
|
||||
import type { ClaudePromptRegistry } from './claude-structured-prompt-replies'
|
||||
import { readableProviderFrameText } from '../native-chat/agent-session-wire/unhandled-provider-frame'
|
||||
import { claudeProviderFrameActivity } from '../native-chat/agent-session-wire/provider-frame-activity'
|
||||
import {
|
||||
CLAUDE_UNRENDERABLE_CONTENT_TEXT,
|
||||
appendUnmodeledClaudeContent,
|
||||
claudeProviderFrameKind,
|
||||
claudeResultFailure,
|
||||
createClaudeProviderFrameFallback,
|
||||
isModeledClaudeContent,
|
||||
isSettledClaudeResultKind
|
||||
} from './claude-structured-provider-fallback'
|
||||
import { ClaudeSubagentRoster } from './claude-subagent-roster'
|
||||
import { createClaudeStreamedBlockRegistry } from './claude-streamed-block-identity'
|
||||
import { createClaudeStreamedTextCheckpoints } from './claude-streamed-text-checkpoints'
|
||||
|
||||
@@ -89,10 +87,16 @@ export function createClaudeJournalTranslator(
|
||||
const promptItems = new Map<string, AgentJournalItemIdentity[]>()
|
||||
const streamedBlocks = createClaudeStreamedBlockRegistry()
|
||||
let currentTurn: { sessionId: string; turnId: string } | null = null
|
||||
const groupKeyOf = (turn: { sessionId: string; turnId: string } | null): string | null =>
|
||||
turn ? `${turn.sessionId}:${turn.turnId}` : null
|
||||
const providerFallback = createClaudeProviderFrameFallback(
|
||||
deps.sink,
|
||||
deps.fallbackIdPrefix ?? 'acquisition'
|
||||
)
|
||||
const subagents = new ClaudeSubagentRoster({
|
||||
sink: deps.sink,
|
||||
currentGroupKey: () => groupKeyOf(currentTurn)
|
||||
})
|
||||
const streamedText = createClaudeStreamedTextCheckpoints({
|
||||
...(deps.coalesceMs === undefined ? {} : { coalesceMs: deps.coalesceMs }),
|
||||
...(deps.schedule ? { schedule: deps.schedule } : {}),
|
||||
@@ -144,14 +148,10 @@ export function createClaudeJournalTranslator(
|
||||
return false
|
||||
}
|
||||
let changed = false
|
||||
// User bubbles belong to the submitted message; SDK user frames carry echoes and tool results.
|
||||
const outputEnvelope =
|
||||
envelope.role === 'user'
|
||||
? {
|
||||
...envelope,
|
||||
content: envelope.content.filter((part) => claudeRecord(part)?.type === 'tool_result')
|
||||
}
|
||||
: envelope
|
||||
if (envelope.parentToolUseId) {
|
||||
subagents.observeChildActivity(envelope.parentToolUseId)
|
||||
}
|
||||
const outputEnvelope = claudeOutputEnvelope(envelope)
|
||||
const body = claudeMessageBody(outputEnvelope)
|
||||
// The final frame of a streamed block lands on the block's identity, not its own uuid.
|
||||
const identity =
|
||||
@@ -180,6 +180,8 @@ export function createClaudeJournalTranslator(
|
||||
claudeToolIdentity(envelope.sessionId, result.toolUseId),
|
||||
claudeToolBody({ tool, result })
|
||||
)
|
||||
// A spawn call's result is the parent turn's evidence its child finished.
|
||||
subagents.observeToolResult(result.toolUseId, result.failed)
|
||||
// Tool inputs are only needed until their matching result arrives.
|
||||
tools.delete(result.toolUseId)
|
||||
changed = true
|
||||
@@ -192,21 +194,7 @@ export function createClaudeJournalTranslator(
|
||||
})
|
||||
changed = true
|
||||
}
|
||||
const unhandledContent = outputEnvelope.content.filter((part) => !isModeledClaudeContent(part))
|
||||
for (const part of unhandledContent) {
|
||||
const partType = claudeText(claudeRecord(part)?.type) ?? 'unknown'
|
||||
providerFallback.append(
|
||||
`message:${envelope.role}:content:${partType}`,
|
||||
part,
|
||||
readableProviderFrameText(part) ?? CLAUDE_UNRENDERABLE_CONTENT_TEXT
|
||||
)
|
||||
changed = true
|
||||
}
|
||||
// An empty user frame is a replay with nothing to show, not an unknown kind.
|
||||
if (envelope.content.length === 0 && envelope.role === 'assistant') {
|
||||
providerFallback.append(`message:${envelope.role}:empty`, message)
|
||||
changed = true
|
||||
}
|
||||
changed = appendUnmodeledClaudeContent(providerFallback, outputEnvelope, message) || changed
|
||||
if (
|
||||
envelope.role === 'user' &&
|
||||
startsTurn &&
|
||||
@@ -214,6 +202,9 @@ export function createClaudeJournalTranslator(
|
||||
message.parent_tool_use_id === null
|
||||
) {
|
||||
if (currentTurn) {
|
||||
// A new turn starting is the only end the previous one gets when its
|
||||
// result never arrives; settling it later would sweep THIS turn.
|
||||
subagents.settleTurn(groupKeyOf(currentTurn))
|
||||
publishLifecycle(currentTurn.sessionId, currentTurn.turnId, false)
|
||||
}
|
||||
currentTurn = { sessionId: envelope.sessionId, turnId: envelope.uuid }
|
||||
@@ -254,6 +245,8 @@ export function createClaudeJournalTranslator(
|
||||
handle: (event) => {
|
||||
if (event.type === 'ended') {
|
||||
streamedText.flush()
|
||||
// No event will ever settle a child once the provider is gone.
|
||||
subagents.settleSession()
|
||||
if (currentTurn) {
|
||||
publishLifecycle(currentTurn.sessionId, currentTurn.turnId, false)
|
||||
currentTurn = null
|
||||
@@ -274,6 +267,9 @@ export function createClaudeJournalTranslator(
|
||||
promptItems.delete(event.promptKey)
|
||||
deps.sink.publish()
|
||||
} else if (event.type === 'message' && event.message.type === 'result') {
|
||||
// The turn is over however it ended, so a foreground child still
|
||||
// reported as working will never be settled by an event.
|
||||
subagents.settleTurn(groupKeyOf(currentTurn))
|
||||
if (currentTurn) {
|
||||
publishLifecycle(currentTurn.sessionId, currentTurn.turnId, false)
|
||||
currentTurn = null
|
||||
@@ -291,6 +287,9 @@ export function createClaudeJournalTranslator(
|
||||
providerFallback.append(kind, event.message, failure?.text)
|
||||
}
|
||||
} else if (event.type === 'message') {
|
||||
// These frames stay `status-chrome`: the roster reads them here, and the
|
||||
// fallback below still drops the raw frame instead of printing an opcode.
|
||||
subagents.observeSystemFrame(event.message)
|
||||
const kind = claudeProviderFrameKind(event.message)
|
||||
if (!handleMessage(event.message, event.startsTurn === true)) {
|
||||
providerFallback.append(kind, event.message)
|
||||
@@ -310,6 +309,7 @@ export function createClaudeJournalTranslator(
|
||||
tools.clear()
|
||||
promptItems.clear()
|
||||
streamedBlocks.clear()
|
||||
subagents.dispose()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,8 +4,15 @@ import {
|
||||
DEFAULT_JOURNAL_PAYLOAD_LIMITS
|
||||
} from '../native-chat/agent-session-journal/journal-payload-bounds'
|
||||
import { CLAUDE_STREAM_JSON_FRAME_KINDS } from '../native-chat/agent-session-wire/claude-stream-json-frame-schema'
|
||||
import { unhandledProviderFrameJournalItem } from '../native-chat/agent-session-wire/unhandled-provider-frame'
|
||||
import { claudeRecord, claudeText } from './claude-structured-item-translation'
|
||||
import {
|
||||
readableProviderFrameText,
|
||||
unhandledProviderFrameJournalItem
|
||||
} from '../native-chat/agent-session-wire/unhandled-provider-frame'
|
||||
import {
|
||||
claudeRecord,
|
||||
claudeText,
|
||||
type ClaudeMessageEnvelope
|
||||
} from './claude-structured-item-translation'
|
||||
|
||||
export function claudeProviderFrameKind(message: Record<string, unknown>): string {
|
||||
const type = claudeText(message.type) ?? 'unknown'
|
||||
@@ -123,3 +130,30 @@ export function createClaudeProviderFrameFallback(
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export type ClaudeProviderFrameFallback = ReturnType<typeof createClaudeProviderFrameFallback>
|
||||
|
||||
/** Journal each content part this build does not model, plus the empty assistant
|
||||
* frame a replay leaves behind (an empty USER frame is a replay with nothing to
|
||||
* show, not an unknown kind). Returns whether anything was appended. */
|
||||
export function appendUnmodeledClaudeContent(
|
||||
fallback: ClaudeProviderFrameFallback,
|
||||
envelope: ClaudeMessageEnvelope,
|
||||
message: Record<string, unknown>
|
||||
): boolean {
|
||||
let changed = false
|
||||
for (const part of envelope.content.filter((part) => !isModeledClaudeContent(part))) {
|
||||
const partType = claudeText(claudeRecord(part)?.type) ?? 'unknown'
|
||||
fallback.append(
|
||||
`message:${envelope.role}:content:${partType}`,
|
||||
part,
|
||||
readableProviderFrameText(part) ?? CLAUDE_UNRENDERABLE_CONTENT_TEXT
|
||||
)
|
||||
changed = true
|
||||
}
|
||||
if (envelope.content.length === 0 && envelope.role === 'assistant') {
|
||||
fallback.append(`message:${envelope.role}:empty`, message)
|
||||
changed = true
|
||||
}
|
||||
return changed
|
||||
}
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import type { NativeChatSubagentEntry } from '../../shared/native-chat-types'
|
||||
import { claudeSubagentGroupBody } from './claude-subagent-group-row'
|
||||
|
||||
function entry(id: string, state: NativeChatSubagentEntry['state']): NativeChatSubagentEntry {
|
||||
return { id, label: id, state, startedAt: 1 }
|
||||
}
|
||||
|
||||
/** The fallback sentence is the WHOLE row on mobile and paired web, which have
|
||||
* no roster renderer, so these assertions are the entire contract there. */
|
||||
function sentence(agents: readonly NativeChatSubagentEntry[]): string {
|
||||
const body = claudeSubagentGroupBody('turn-1', agents)
|
||||
const block = body.kind === 'message' ? body.blocks[0] : undefined
|
||||
return block && block.type === 'text' ? block.text : ''
|
||||
}
|
||||
|
||||
describe('claudeSubagentGroupBody fallback sentence', () => {
|
||||
it('reads as a plain completion when every child completed', () => {
|
||||
expect(sentence([entry('a', 'completed'), entry('b', 'completed')])).toBe('Ran 2 subagents')
|
||||
})
|
||||
|
||||
it('keeps the singular noun for a lone child', () => {
|
||||
expect(sentence([entry('a', 'completed')])).toBe('Ran 1 subagent')
|
||||
expect(sentence([entry('a', 'working')])).toBe('Kicked off 1 subagent')
|
||||
})
|
||||
|
||||
it('names an unverifiable child instead of claiming the group ran', () => {
|
||||
expect(sentence([entry('a', 'completed'), entry('b', 'unverifiable')])).toBe(
|
||||
'Ran 2 subagents (1 unverifiable)'
|
||||
)
|
||||
})
|
||||
|
||||
it('ranks the adverse outcome worst-first', () => {
|
||||
expect(
|
||||
sentence([entry('a', 'failed'), entry('b', 'unverifiable'), entry('c', 'completed')])
|
||||
).toBe('Ran 3 subagents (1 failed)')
|
||||
expect(sentence([entry('a', 'stopped'), entry('b', 'unverifiable')])).toBe(
|
||||
'Ran 2 subagents (1 stopped)'
|
||||
)
|
||||
})
|
||||
|
||||
it('shows the adverse outcome while a sibling still works', () => {
|
||||
expect(
|
||||
sentence([entry('a', 'working'), entry('b', 'working'), entry('c', 'unverifiable')])
|
||||
).toBe('Kicked off 3 subagents (1 unverifiable)')
|
||||
})
|
||||
|
||||
it('leaves a benign settled state out of the sentence', () => {
|
||||
expect(sentence([entry('a', 'idle'), entry('b', 'completed')])).toBe('Ran 2 subagents')
|
||||
})
|
||||
|
||||
it('counts every child holding the worst adverse state', () => {
|
||||
expect(sentence([entry('a', 'failed'), entry('b', 'failed'), entry('c', 'stopped')])).toBe(
|
||||
'Ran 3 subagents (2 failed)'
|
||||
)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,32 @@
|
||||
// The journal row one Claude spawn group writes: its durable identity and the
|
||||
// body it revises in place.
|
||||
|
||||
import type {
|
||||
AgentJournalItemBody,
|
||||
AgentJournalItemIdentity
|
||||
} from '../../shared/agent-session-journal-types'
|
||||
import { subagentGroupFallbackText } from '../../shared/native-chat-subagent-summary'
|
||||
import type { NativeChatSubagentEntry } from '../../shared/native-chat-types'
|
||||
|
||||
/** 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 claudeSubagentGroupIdentity(groupId: string): AgentJournalItemIdentity {
|
||||
return { provider: 'orca', clientMessageId: `claude-subagents:${groupId}` }
|
||||
}
|
||||
|
||||
/** 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 claudeSubagentGroupBody(
|
||||
groupId: string,
|
||||
agents: readonly NativeChatSubagentEntry[]
|
||||
): AgentJournalItemBody {
|
||||
return {
|
||||
kind: 'message',
|
||||
role: 'system',
|
||||
blocks: [
|
||||
{ type: 'text', text: subagentGroupFallbackText(agents) },
|
||||
{ type: 'subagent-group', groupId, agents: [...agents] }
|
||||
]
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { ClaudeSubagentIds } from './claude-subagent-id-aliases'
|
||||
|
||||
describe('ClaudeSubagentIds', () => {
|
||||
it('resolves an aliased tool id to its task, and an unaliased id to itself', () => {
|
||||
const ids = new ClaudeSubagentIds()
|
||||
ids.alias('toolu_1', 'task-1')
|
||||
expect(ids.canonical('toolu_1')).toBe('task-1')
|
||||
expect(ids.canonical('toolu_unknown')).toBe('toolu_unknown')
|
||||
})
|
||||
|
||||
it('remembers an exclusion under either of the ids that named it', () => {
|
||||
const ids = new ClaudeSubagentIds()
|
||||
ids.exclude('task-bash')
|
||||
expect(ids.isExcluded('toolu_bash', 'task-bash')).toBe(true)
|
||||
expect(ids.isExcluded(null, null)).toBe(false)
|
||||
expect(ids.isExcluded('task-agent')).toBe(false)
|
||||
})
|
||||
|
||||
it('drops the oldest alias past the bound and keeps the newest', () => {
|
||||
const ids = new ClaudeSubagentIds()
|
||||
for (let index = 0; index <= 512; index += 1) {
|
||||
ids.alias(`toolu_${index}`, `task-${index}`)
|
||||
}
|
||||
// Evicted: the id now stands only for itself.
|
||||
expect(ids.canonical('toolu_0')).toBe('toolu_0')
|
||||
expect(ids.canonical('toolu_512')).toBe('task-512')
|
||||
expect(ids.canonical('toolu_1')).toBe('task-1')
|
||||
})
|
||||
|
||||
it('drops the oldest exclusion past the bound and keeps the newest', () => {
|
||||
const ids = new ClaudeSubagentIds()
|
||||
for (let index = 0; index <= 512; index += 1) {
|
||||
ids.exclude(`task-${index}`)
|
||||
}
|
||||
expect(ids.isExcluded('task-0')).toBe(false)
|
||||
expect(ids.isExcluded('task-512')).toBe(true)
|
||||
expect(ids.isExcluded('task-1')).toBe(true)
|
||||
})
|
||||
|
||||
it('does not retain oversized aliases or exclusions', () => {
|
||||
const ids = new ClaudeSubagentIds()
|
||||
const oversized = 'x'.repeat(513)
|
||||
ids.alias(oversized, 'task-1')
|
||||
ids.alias('tool-1', oversized)
|
||||
ids.exclude(oversized)
|
||||
expect(ids.canonical(oversized)).toBe(oversized)
|
||||
expect(ids.canonical('tool-1')).toBe('tool-1')
|
||||
expect(ids.isExcluded(oversized)).toBe(false)
|
||||
})
|
||||
|
||||
it('forgets everything on clear', () => {
|
||||
const ids = new ClaudeSubagentIds()
|
||||
ids.alias('toolu_1', 'task-1')
|
||||
ids.exclude('task-1')
|
||||
ids.clear()
|
||||
expect(ids.canonical('toolu_1')).toBe('toolu_1')
|
||||
expect(ids.isExcluded('task-1')).toBe(false)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,63 @@
|
||||
// Which Claude ids name the same subagent, and which name no subagent at all.
|
||||
//
|
||||
// Claude re-announces a resumed task under a NEW `tool_use_id` while `task_id`
|
||||
// stays put, so tool ids are aliases of a canonical task id — a store keyed on
|
||||
// the tool id would show the child twice after every resume.
|
||||
//
|
||||
// The exclusions matter just as much: `task_updated` carries no `task_type` and
|
||||
// child traffic carries no task metadata at all, so the one announcement that
|
||||
// said "this is a backgrounded shell, not an agent" has to be remembered or a
|
||||
// later frame re-admits it.
|
||||
|
||||
import { isBoundedClaudeTaskId } from './claude-background-task-tracker'
|
||||
|
||||
/** Both maps are event-accumulated and nothing prunes them, so both are bounded. */
|
||||
const MAX_TOOL_USE_ALIASES = 512
|
||||
const MAX_EXCLUDED_IDS = 512
|
||||
|
||||
export class ClaudeSubagentIds {
|
||||
private readonly canonicalByToolUse = new Map<string, string>()
|
||||
private readonly excluded = new Set<string>()
|
||||
|
||||
/** The task id a tool id stands for, or the id itself when nothing aliases it. */
|
||||
canonical(id: string): string {
|
||||
return this.canonicalByToolUse.get(id) ?? id
|
||||
}
|
||||
|
||||
alias(toolUseId: string, taskId: string): void {
|
||||
if (!isBoundedClaudeTaskId(toolUseId) || !isBoundedClaudeTaskId(taskId)) {
|
||||
return
|
||||
}
|
||||
this.canonicalByToolUse.set(toolUseId, taskId)
|
||||
while (this.canonicalByToolUse.size > MAX_TOOL_USE_ALIASES) {
|
||||
const oldest = this.canonicalByToolUse.keys().next()
|
||||
if (oldest.done || oldest.value === toolUseId) {
|
||||
break
|
||||
}
|
||||
this.canonicalByToolUse.delete(oldest.value)
|
||||
}
|
||||
}
|
||||
|
||||
exclude(id: string): void {
|
||||
if (!isBoundedClaudeTaskId(id)) {
|
||||
return
|
||||
}
|
||||
this.excluded.add(id)
|
||||
while (this.excluded.size > MAX_EXCLUDED_IDS) {
|
||||
const oldest = this.excluded.values().next()
|
||||
if (oldest.done || oldest.value === id) {
|
||||
break
|
||||
}
|
||||
this.excluded.delete(oldest.value)
|
||||
}
|
||||
}
|
||||
|
||||
isExcluded(...ids: (string | null)[]): boolean {
|
||||
return ids.some((id) => id !== null && this.excluded.has(id))
|
||||
}
|
||||
|
||||
clear(): void {
|
||||
this.canonicalByToolUse.clear()
|
||||
this.excluded.clear()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,75 @@
|
||||
import type { AgentJournalItemIdentity } from '../../shared/agent-session-journal-types'
|
||||
import type { NativeChatSubagentEntry } from '../../shared/native-chat-types'
|
||||
import type { ClaudeSubagentTaskFrame } from './claude-subagent-task-frames'
|
||||
|
||||
const MAX_INVOCATIONS_PER_SUBAGENT = 16
|
||||
|
||||
export type TrackedEntry = {
|
||||
entry: NativeChatSubagentEntry
|
||||
/** The only signal separating a child that dies with its turn from one told to
|
||||
* outlive it. A turn-end sweep must leave a backgrounded child alone. */
|
||||
backgrounded: boolean
|
||||
toolUseId: string | null
|
||||
invocationIds: Set<string> | null
|
||||
/** Label before its ordinal suffix, so a later announcement can tell a
|
||||
* provisional row from one that already carries the provider's own name. */
|
||||
labelBase: string
|
||||
}
|
||||
|
||||
export type RosterGroup = {
|
||||
groupId: string
|
||||
identity: AgentJournalItemIdentity
|
||||
/** Insertion order is the display order; the map holds the state. */
|
||||
entries: Map<string, TrackedEntry>
|
||||
/** Lifetime admissions bound retained labels even when entries are removed. */
|
||||
admittedEntries: number
|
||||
/** Labels remain reserved after removal or provisional-name replacement. */
|
||||
claimedLabels: Set<string>
|
||||
/** Last body written, so an idempotent replay writes no new revision. */
|
||||
lastSerialized: string | null
|
||||
}
|
||||
|
||||
// Invocation history stays with the entry, independent of the evicting alias cache.
|
||||
export function applyClaudeSubagentInvocation(
|
||||
tracked: TrackedEntry,
|
||||
frame: ClaudeSubagentTaskFrame,
|
||||
now: () => number
|
||||
): boolean {
|
||||
if (tracked.invocationIds === null) {
|
||||
return false
|
||||
}
|
||||
const newInvocation =
|
||||
frame.announcement && frame.toolUseId !== null && !tracked.invocationIds.has(frame.toolUseId)
|
||||
if (newInvocation && frame.toolUseId) {
|
||||
if (tracked.invocationIds.size >= MAX_INVOCATIONS_PER_SUBAGENT) {
|
||||
tracked.invocationIds = null
|
||||
tracked.entry = { ...tracked.entry, state: 'unverifiable', settledAt: now() }
|
||||
return true
|
||||
}
|
||||
tracked.invocationIds.add(frame.toolUseId)
|
||||
if (tracked.toolUseId !== null && tracked.toolUseId !== frame.toolUseId) {
|
||||
tracked.backgrounded = frame.backgrounded ?? false
|
||||
tracked.entry = { ...tracked.entry, state: frame.state ?? 'working', settledAt: undefined }
|
||||
}
|
||||
tracked.toolUseId = frame.toolUseId
|
||||
} else if (tracked.toolUseId && frame.toolUseId && tracked.toolUseId !== frame.toolUseId) {
|
||||
return false
|
||||
}
|
||||
if (tracked.toolUseId === null) {
|
||||
tracked.toolUseId = frame.toolUseId
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
/** Two children can share a description; the ordinal keeps their rows apart
|
||||
* without inventing a name the provider never sent. The probe is over the
|
||||
* labels actually rendered, not a per-base counter: a generated `Audit 2`
|
||||
* must not collide with a provider that names its own child `Audit 2`. */
|
||||
export function claimClaudeSubagentLabel(group: RosterGroup, base: string): string {
|
||||
let candidate = base
|
||||
for (let ordinal = 2; group.claimedLabels.has(candidate); ordinal++) {
|
||||
candidate = `${base} ${ordinal}`
|
||||
}
|
||||
group.claimedLabels.add(candidate)
|
||||
return candidate
|
||||
}
|
||||
@@ -0,0 +1,602 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import type {
|
||||
AgentJournalItemBody,
|
||||
AgentJournalItemIdentity
|
||||
} from '../../shared/agent-session-journal-types'
|
||||
import type {
|
||||
NativeChatSubagentEntry,
|
||||
NativeChatSubagentGroupBlock
|
||||
} from '../../shared/native-chat-types'
|
||||
import type { AgentSessionJournal } from '../native-chat/agent-session-journal/journal-store'
|
||||
import {
|
||||
createDeferredStructuredAgentSessionEventSink,
|
||||
type StructuredAgentSessionEventSink
|
||||
} from '../native-chat/agent-session-wire/structured-agent-session-event-sink'
|
||||
import { ClaudeSubagentRoster } from './claude-subagent-roster'
|
||||
|
||||
const TURN_1 = 'claude-session:turn-1'
|
||||
|
||||
function agentsOf(body: AgentJournalItemBody | undefined): NativeChatSubagentEntry[] {
|
||||
if (!body || body.kind !== 'message') {
|
||||
return []
|
||||
}
|
||||
const block = body.blocks.find(
|
||||
(candidate): candidate is NativeChatSubagentGroupBlock => candidate.type === 'subagent-group'
|
||||
)
|
||||
return block ? block.agents : []
|
||||
}
|
||||
|
||||
function isGroupRow(identity: AgentJournalItemIdentity, groupId: string): boolean {
|
||||
return identity.provider === 'orca' && identity.clientMessageId === `claude-subagents:${groupId}`
|
||||
}
|
||||
|
||||
function harness(groupKey: string | null = TURN_1) {
|
||||
const items: { identity: AgentJournalItemIdentity; body: AgentJournalItemBody }[] = []
|
||||
const tombstones: AgentJournalItemIdentity[] = []
|
||||
const sink: StructuredAgentSessionEventSink = {
|
||||
appendItem: (identity, body) => items.push({ identity, body }),
|
||||
appendTombstone: (identity) => tombstones.push(identity),
|
||||
publish: vi.fn()
|
||||
}
|
||||
let clock = 1_000
|
||||
let key = groupKey
|
||||
const roster = new ClaudeSubagentRoster({
|
||||
sink,
|
||||
currentGroupKey: () => key,
|
||||
now: () => (clock += 1)
|
||||
})
|
||||
const roles = (): NativeChatSubagentEntry[] => agentsOf(items.at(-1)?.body)
|
||||
/** The last row written for one group, so a test can read a row that is no
|
||||
* longer the newest one. */
|
||||
const rolesIn = (groupId: string): NativeChatSubagentEntry[] =>
|
||||
agentsOf(items.findLast((item) => isGroupRow(item.identity, groupId))?.body)
|
||||
return {
|
||||
roster,
|
||||
items,
|
||||
tombstones,
|
||||
roles,
|
||||
rolesIn,
|
||||
setGroupKey: (next: string | null) => {
|
||||
key = next
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function system(subtype: string, fields: Record<string, unknown>): Record<string, unknown> {
|
||||
return { type: 'system', subtype, session_id: 'claude-session', ...fields }
|
||||
}
|
||||
|
||||
function started(fields: Record<string, unknown>): Record<string, unknown> {
|
||||
return system('task_started', { task_type: 'local_agent', ...fields })
|
||||
}
|
||||
|
||||
describe('ClaudeSubagentRoster', () => {
|
||||
it('builds the row from task_started, with the fallback sentence beside the block', () => {
|
||||
const { roster, items, roles } = harness()
|
||||
roster.observeSystemFrame(
|
||||
started({ task_id: 'task-1', tool_use_id: 'toolu_1', description: 'Review the diff' })
|
||||
)
|
||||
expect(items).toHaveLength(1)
|
||||
expect(items[0]?.identity).toEqual({
|
||||
provider: 'orca',
|
||||
clientMessageId: 'claude-subagents:claude-session:turn-1'
|
||||
})
|
||||
const body = items[0]?.body
|
||||
expect(body?.kind === 'message' && body.blocks[0]).toEqual({
|
||||
type: 'text',
|
||||
text: 'Kicked off 1 subagent'
|
||||
})
|
||||
expect(roles()).toEqual([
|
||||
expect.objectContaining({ id: 'task-1', label: 'Review the diff', state: 'working' })
|
||||
])
|
||||
})
|
||||
|
||||
it('keeps a backgrounded shell task out of the roster', () => {
|
||||
const { roster, items } = harness()
|
||||
roster.observeSystemFrame(
|
||||
system('task_started', {
|
||||
task_id: 'task-bash',
|
||||
tool_use_id: 'toolu_bash',
|
||||
task_type: 'local_bash',
|
||||
description: 'sleep 20',
|
||||
is_backgrounded: true
|
||||
})
|
||||
)
|
||||
roster.observeSystemFrame(
|
||||
system('task_updated', { task_id: 'task-bash', patch: { status: 'running' } })
|
||||
)
|
||||
// Its own frames carry a tool_use_id, so only the excluded-id memory stops it.
|
||||
roster.observeChildActivity('toolu_bash')
|
||||
expect(items).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('never renders a task marked skip_transcript', () => {
|
||||
const { roster, items } = harness()
|
||||
roster.observeSystemFrame(
|
||||
started({ task_id: 'task-a', tool_use_id: 'toolu_a', skip_transcript: true })
|
||||
)
|
||||
roster.observeSystemFrame(
|
||||
system('task_updated', { task_id: 'task-a', patch: { status: 'completed' } })
|
||||
)
|
||||
roster.observeChildActivity('toolu_a')
|
||||
expect(items).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('drops a provisional row once an announcement says the task is not a subagent', () => {
|
||||
const { roster, items, tombstones, roles } = harness()
|
||||
roster.observeChildActivity('toolu_bash')
|
||||
expect(roles()).toHaveLength(1)
|
||||
roster.observeSystemFrame(
|
||||
system('task_started', {
|
||||
task_id: 'task-bash',
|
||||
tool_use_id: 'toolu_bash',
|
||||
task_type: 'local_bash'
|
||||
})
|
||||
)
|
||||
expect(tombstones).toEqual([
|
||||
{ provider: 'orca', clientMessageId: 'claude-subagents:claude-session:turn-1' }
|
||||
])
|
||||
expect(items).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('does not duplicate a resumed task re-announced under a new tool_use_id', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeSystemFrame(
|
||||
started({ task_id: 'task-1', tool_use_id: 'toolu_first', description: 'Audit' })
|
||||
)
|
||||
roster.observeChildActivity('toolu_first')
|
||||
roster.observeSystemFrame(
|
||||
started({ task_id: 'task-1', tool_use_id: 'toolu_second', description: 'Audit' })
|
||||
)
|
||||
roster.observeChildActivity('toolu_second')
|
||||
expect(roles()).toEqual([
|
||||
expect.objectContaining({ id: 'task-1', label: 'Audit', state: 'working' })
|
||||
])
|
||||
})
|
||||
|
||||
it('adopts a row built from child traffic when the announcement finally names it', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeChildActivity('toolu_1')
|
||||
expect(roles()).toEqual([expect.objectContaining({ id: 'toolu_1', label: 'subagent' })])
|
||||
roster.observeSystemFrame(
|
||||
started({ task_id: 'task-1', tool_use_id: 'toolu_1', description: 'Explore' })
|
||||
)
|
||||
expect(roles()).toEqual([
|
||||
expect.objectContaining({ id: 'task-1', label: 'Explore', state: 'working' })
|
||||
])
|
||||
})
|
||||
|
||||
it('is idempotent: a repeated frame writes no new revision', () => {
|
||||
const { roster, items } = harness()
|
||||
const frame = started({ task_id: 'task-1', tool_use_id: 'toolu_1', description: 'Audit' })
|
||||
roster.observeSystemFrame(frame)
|
||||
roster.observeSystemFrame(frame)
|
||||
roster.observeSystemFrame(
|
||||
system('task_updated', { task_id: 'task-1', patch: { status: 'running' } })
|
||||
)
|
||||
expect(items).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('latches a terminal state against a later live report', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1', description: 'Audit' }))
|
||||
roster.observeSystemFrame(
|
||||
system('task_updated', { task_id: 'task-1', patch: { status: 'failed' } })
|
||||
)
|
||||
roster.observeSystemFrame(
|
||||
system('task_updated', { task_id: 'task-1', patch: { status: 'running' } })
|
||||
)
|
||||
expect(roles()).toEqual([expect.objectContaining({ state: 'failed' })])
|
||||
})
|
||||
|
||||
it('ignores an update for a task it never rostered', () => {
|
||||
const { roster, items } = harness()
|
||||
roster.observeSystemFrame(
|
||||
system('task_updated', { task_id: 'task-unknown', patch: { status: 'running' } })
|
||||
)
|
||||
expect(items).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('disambiguates children that share a description', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1', description: 'Explore' }))
|
||||
roster.observeSystemFrame(started({ task_id: 'task-2', description: 'Explore' }))
|
||||
expect(roles().map((agent) => agent.label)).toEqual(['Explore', 'Explore 2'])
|
||||
})
|
||||
|
||||
describe('turn end', () => {
|
||||
it('leaves a backgrounded child working and marks a foreground one unverifiable', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeSystemFrame(started({ task_id: 'task-fg', description: 'Foreground' }))
|
||||
roster.observeSystemFrame(
|
||||
started({ task_id: 'task-bg', description: 'Background', is_backgrounded: true })
|
||||
)
|
||||
roster.settleTurn(TURN_1)
|
||||
expect(roles()).toEqual([
|
||||
expect.objectContaining({ label: 'Foreground', state: 'unverifiable' }),
|
||||
expect.objectContaining({ label: 'Background', state: 'working' })
|
||||
])
|
||||
})
|
||||
|
||||
it('never re-settles a child that already reported an outcome', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1', description: 'Audit' }))
|
||||
roster.observeSystemFrame(
|
||||
system('task_updated', { task_id: 'task-1', patch: { status: 'completed' } })
|
||||
)
|
||||
roster.settleTurn(TURN_1)
|
||||
expect(roles()).toEqual([expect.objectContaining({ state: 'completed' })])
|
||||
})
|
||||
|
||||
it('sweeps backgrounded children only when the provider itself is gone', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeSystemFrame(
|
||||
started({ task_id: 'task-bg', description: 'Background', is_backgrounded: true })
|
||||
)
|
||||
roster.settleTurn(TURN_1)
|
||||
roster.settleSession()
|
||||
expect(roles()).toEqual([expect.objectContaining({ state: 'unverifiable' })])
|
||||
})
|
||||
})
|
||||
|
||||
describe('spawn tool result', () => {
|
||||
it('settles a foreground child', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1', tool_use_id: 'toolu_1' }))
|
||||
roster.observeToolResult('toolu_1', false)
|
||||
expect(roles()).toEqual([expect.objectContaining({ state: 'completed' })])
|
||||
})
|
||||
|
||||
it('reports a failed spawn as failed', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1', tool_use_id: 'toolu_1' }))
|
||||
roster.observeToolResult('toolu_1', true)
|
||||
expect(roles()).toEqual([expect.objectContaining({ state: 'failed' })])
|
||||
})
|
||||
|
||||
it('ignores the immediate result a backgrounded spawn returns', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeSystemFrame(
|
||||
started({ task_id: 'task-1', tool_use_id: 'toolu_1', is_backgrounded: true })
|
||||
)
|
||||
roster.observeToolResult('toolu_1', false)
|
||||
expect(roles()).toEqual([expect.objectContaining({ state: 'working' })])
|
||||
})
|
||||
|
||||
it('ignores results for tools that are not spawn calls', () => {
|
||||
const { roster, items } = harness()
|
||||
roster.observeToolResult('toolu_read', false)
|
||||
expect(items).toHaveLength(0)
|
||||
})
|
||||
})
|
||||
|
||||
describe('label ordinals', () => {
|
||||
it('never re-issues an ordinal a removed row gave up', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1', description: 'Audit' }))
|
||||
roster.observeSystemFrame(started({ task_id: 'task-2', description: 'Audit' }))
|
||||
// task-1 is re-announced as a shell task, so its row goes; reclaiming the
|
||||
// ordinal it held would print a second 'Audit 2' beside the one still shown.
|
||||
roster.observeSystemFrame(
|
||||
system('task_started', { task_id: 'task-1', task_type: 'local_bash' })
|
||||
)
|
||||
roster.observeSystemFrame(started({ task_id: 'task-3', description: 'Audit' }))
|
||||
expect(roles().map((agent) => agent.label)).toEqual(['Audit 2', 'Audit 3'])
|
||||
})
|
||||
|
||||
it('never generates a label a provider-supplied one already took', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1', description: 'Audit' }))
|
||||
roster.observeSystemFrame(started({ task_id: 'task-2', description: 'Audit' }))
|
||||
// The provider's own name for the third child is the label the ordinal just
|
||||
// generated for the second; a per-base counter would print it twice.
|
||||
roster.observeSystemFrame(started({ task_id: 'task-3', description: 'Audit 2' }))
|
||||
const labels = roles().map((agent) => agent.label)
|
||||
expect(labels).toEqual(['Audit', 'Audit 2', 'Audit 2 2'])
|
||||
expect(new Set(labels).size).toBe(labels.length)
|
||||
})
|
||||
})
|
||||
|
||||
describe('child traffic for an id the CLI never declared', () => {
|
||||
it('creates nothing once the CLI has announced any task at all', () => {
|
||||
const { roster, items } = harness()
|
||||
// A rejected announcement still proves this CLI declares what it spawns.
|
||||
roster.observeSystemFrame(
|
||||
system('task_started', { task_id: 'task-bash', task_type: 'local_bash' })
|
||||
)
|
||||
roster.observeChildActivity('toolu_never_announced')
|
||||
expect(items).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('rejects an over-long provisional id instead of storing it as an entry id', () => {
|
||||
const { roster, items } = harness()
|
||||
// The announced path drops an id past `claudeTaskId`'s bound; the
|
||||
// provisional one writes the same durable entry id, so it must too.
|
||||
roster.observeChildActivity(`toolu_${'x'.repeat(512)}`)
|
||||
expect(items).toHaveLength(0)
|
||||
roster.observeChildActivity(`toolu_${'x'.repeat(500)}`)
|
||||
expect(items).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('still rosters a subagent announced after a task the filter rejected', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeSystemFrame(
|
||||
system('task_started', { task_id: 'task-bash', task_type: 'local_bash' })
|
||||
)
|
||||
// The gate closes the child-traffic fallback, never the announcement path.
|
||||
roster.observeSystemFrame(
|
||||
started({ task_id: 'task-1', tool_use_id: 'toolu_1', description: 'Explore' })
|
||||
)
|
||||
roster.observeChildActivity('toolu_1')
|
||||
expect(roles()).toEqual([
|
||||
expect.objectContaining({ id: 'task-1', label: 'Explore', state: 'working' })
|
||||
])
|
||||
})
|
||||
|
||||
it('leaves a grandchild parented inside the sidechain out of the roster', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeSystemFrame(
|
||||
started({ task_id: 'task-1', tool_use_id: 'toolu_1', description: 'Explore' })
|
||||
)
|
||||
roster.observeChildActivity('toolu_1')
|
||||
// A tool the subagent itself ran: never announced, so never excluded either.
|
||||
roster.observeChildActivity('toolu_inner')
|
||||
expect(roles()).toEqual([
|
||||
expect.objectContaining({ id: 'task-1', label: 'Explore', state: 'working' })
|
||||
])
|
||||
})
|
||||
|
||||
it('still mints the provisional row for a release that announces no task', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeChildActivity('toolu_1')
|
||||
// Not an announcement: the fallback path stays open for this release.
|
||||
roster.observeSystemFrame(
|
||||
system('task_updated', { task_id: 'task-x', patch: { status: 'running' } })
|
||||
)
|
||||
roster.observeChildActivity('toolu_2')
|
||||
expect(roles().map((agent) => agent.label)).toEqual(['subagent', 'subagent 2'])
|
||||
})
|
||||
})
|
||||
|
||||
describe('groups that no later event can reach', () => {
|
||||
it('loses contact with a group evicted past the bound', () => {
|
||||
const { roster, rolesIn, setGroupKey } = harness('turn-0')
|
||||
for (let index = 0; index < 33; index += 1) {
|
||||
setGroupKey(`turn-${index}`)
|
||||
roster.observeSystemFrame(started({ task_id: `task-${index}`, description: 'Audit' }))
|
||||
}
|
||||
expect(rolesIn('turn-0')).toEqual([expect.objectContaining({ state: 'unverifiable' })])
|
||||
expect(rolesIn('turn-32')).toEqual([expect.objectContaining({ state: 'working' })])
|
||||
})
|
||||
|
||||
it('loses contact with a live child when the translator is disposed without an end', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeSystemFrame(
|
||||
started({ task_id: 'task-bg', description: 'Background', is_backgrounded: true })
|
||||
)
|
||||
roster.dispose()
|
||||
expect(roles()).toEqual([expect.objectContaining({ state: 'unverifiable' })])
|
||||
})
|
||||
|
||||
it('writes nothing on dispose when the session already settled', () => {
|
||||
const { roster, items } = harness()
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1', description: 'Audit' }))
|
||||
roster.settleSession()
|
||||
const written = items.length
|
||||
roster.dispose()
|
||||
expect(items).toHaveLength(written)
|
||||
})
|
||||
})
|
||||
|
||||
it('groups children outside any turn under their own row', () => {
|
||||
const { roster, items } = harness(null)
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1', description: 'Audit' }))
|
||||
expect(items[0]?.identity).toEqual({
|
||||
provider: 'orca',
|
||||
clientMessageId: 'claude-subagents:outside-turn'
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
describe('ClaudeSubagentRoster — the turn that is ending', () => {
|
||||
it('leaves a child announced outside any turn alone when an unrelated turn ends', () => {
|
||||
const { roster, rolesIn, setGroupKey } = harness(null)
|
||||
roster.observeSystemFrame(started({ task_id: 'task-early', description: 'Early' }))
|
||||
setGroupKey(TURN_1)
|
||||
roster.observeSystemFrame(started({ task_id: 'task-turn', description: 'In turn' }))
|
||||
roster.settleTurn(TURN_1)
|
||||
expect(rolesIn('outside-turn')).toEqual([expect.objectContaining({ state: 'working' })])
|
||||
expect(rolesIn(TURN_1)).toEqual([expect.objectContaining({ state: 'unverifiable' })])
|
||||
// `unverifiable` latches, so sweeping it above would have swallowed this.
|
||||
roster.observeSystemFrame(
|
||||
system('task_updated', { task_id: 'task-early', patch: { status: 'completed' } })
|
||||
)
|
||||
expect(rolesIn('outside-turn')).toEqual([expect.objectContaining({ state: 'completed' })])
|
||||
})
|
||||
|
||||
it('sweeps the outside-turn group when a turn with no key of its own ends', () => {
|
||||
const { roster, rolesIn } = harness(null)
|
||||
roster.observeSystemFrame(started({ task_id: 'task-early', description: 'Early' }))
|
||||
roster.settleTurn(null)
|
||||
expect(rolesIn('outside-turn')).toEqual([expect.objectContaining({ state: 'unverifiable' })])
|
||||
})
|
||||
|
||||
it('still settles an outside-turn child once the session itself ends', () => {
|
||||
const { roster, rolesIn, setGroupKey } = harness(null)
|
||||
roster.observeSystemFrame(started({ task_id: 'task-early', description: 'Early' }))
|
||||
setGroupKey(TURN_1)
|
||||
roster.observeSystemFrame(started({ task_id: 'task-turn', description: 'In turn' }))
|
||||
roster.settleTurn(TURN_1)
|
||||
roster.settleSession()
|
||||
expect(rolesIn('outside-turn')).toEqual([expect.objectContaining({ state: 'unverifiable' })])
|
||||
})
|
||||
|
||||
it('sweeps the turn that ended, not whichever turn is live now', () => {
|
||||
const { roster, rolesIn, setGroupKey } = harness()
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1', description: 'First turn' }))
|
||||
setGroupKey('claude-session:turn-2')
|
||||
roster.observeSystemFrame(started({ task_id: 'task-2', description: 'Second turn' }))
|
||||
// Turn 1's result lands after turn 2 has already begun.
|
||||
roster.settleTurn(TURN_1)
|
||||
expect(rolesIn(TURN_1)).toEqual([expect.objectContaining({ state: 'unverifiable' })])
|
||||
expect(rolesIn('claude-session:turn-2')).toEqual([
|
||||
expect.objectContaining({ state: 'working' })
|
||||
])
|
||||
})
|
||||
})
|
||||
|
||||
describe('ClaudeSubagentRoster — through the real sink queue', () => {
|
||||
it('lands every revision, not just the one that was already in flight', async () => {
|
||||
const appended: AgentJournalItemBody[] = []
|
||||
let published = 0
|
||||
const journal = {
|
||||
appendItem: async (_identity: AgentJournalItemIdentity, body: AgentJournalItemBody) => {
|
||||
appended.push(body)
|
||||
return { cursor: { epoch: 'e', sequence: appended.length } }
|
||||
},
|
||||
appendTombstone: async () => ({ epoch: 'e', sequence: 0 })
|
||||
} as unknown as AgentSessionJournal
|
||||
const deferred = createDeferredStructuredAgentSessionEventSink()
|
||||
deferred.bind({
|
||||
journal,
|
||||
fence: 1,
|
||||
publish: () => {
|
||||
published += 1
|
||||
}
|
||||
})
|
||||
const roster = new ClaudeSubagentRoster({ sink: deferred.sink, currentGroupKey: () => TURN_1 })
|
||||
|
||||
// The first append is in flight while the rest are submitted, so a publish
|
||||
// sharing the row's coalescing key would evict them.
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1', description: 'One' }))
|
||||
roster.observeSystemFrame(started({ task_id: 'task-2', description: 'Two' }))
|
||||
roster.observeSystemFrame(
|
||||
system('task_updated', { task_id: 'task-1', patch: { status: 'completed' } })
|
||||
)
|
||||
const drained = await deferred.drained()
|
||||
|
||||
expect(drained).toEqual({ ok: true })
|
||||
expect(agentsOf(appended.at(-1))).toEqual([
|
||||
expect.objectContaining({ id: 'task-1', label: 'One', state: 'completed' }),
|
||||
expect.objectContaining({ id: 'task-2', label: 'Two', state: 'working' })
|
||||
])
|
||||
expect(published).toBeGreaterThan(0)
|
||||
})
|
||||
})
|
||||
|
||||
describe('ClaudeSubagentRoster — authoritative outcomes and retained budgets', () => {
|
||||
it('accepts a notification after the foreground turn lost contact', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1' }))
|
||||
roster.settleTurn(TURN_1)
|
||||
roster.observeSystemFrame(
|
||||
system('task_notification', { task_id: 'task-1', status: 'completed' })
|
||||
)
|
||||
expect(roles()).toEqual([expect.objectContaining({ state: 'completed' })])
|
||||
})
|
||||
|
||||
it('settles a background child from its notification without a task_updated', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1', is_backgrounded: true }))
|
||||
roster.settleTurn(TURN_1)
|
||||
roster.observeSystemFrame(system('task_notification', { task_id: 'task-1', status: 'failed' }))
|
||||
expect(roles()).toEqual([expect.objectContaining({ state: 'failed' })])
|
||||
})
|
||||
|
||||
it('bounds lifetime admissions when reclassification repeatedly removes entries', () => {
|
||||
const { roster, items } = harness()
|
||||
for (let i = 0; i < 100; i++) {
|
||||
roster.observeSystemFrame(started({ task_id: `task-${i}`, description: `Agent ${i}` }))
|
||||
roster.observeSystemFrame(
|
||||
system('task_started', { task_id: `task-${i}`, task_type: 'local_bash' })
|
||||
)
|
||||
}
|
||||
expect(items).toHaveLength(64)
|
||||
})
|
||||
})
|
||||
|
||||
describe('ClaudeSubagentRoster — resumed invocation', () => {
|
||||
it('reopens one canonical child on a new announcement without replaying old results', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1', tool_use_id: 'first' }))
|
||||
roster.observeToolResult('first', false)
|
||||
roster.observeSystemFrame(
|
||||
started({ task_id: 'task-1', tool_use_id: 'resumed', is_backgrounded: true })
|
||||
)
|
||||
expect(roles()).toEqual([expect.objectContaining({ id: 'task-1', state: 'working' })])
|
||||
expect(roles()[0].settledAt).toBeUndefined()
|
||||
roster.observeSystemFrame(
|
||||
system('task_notification', { task_id: 'task-1', tool_use_id: 'first', status: 'completed' })
|
||||
)
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1', tool_use_id: 'first' }))
|
||||
expect(roles()[0].state).toBe('working')
|
||||
roster.observeSystemFrame(
|
||||
system('task_notification', {
|
||||
task_id: 'task-1',
|
||||
tool_use_id: 'resumed',
|
||||
status: 'completed'
|
||||
})
|
||||
)
|
||||
roster.observeSystemFrame(
|
||||
started({ task_id: 'task-1', tool_use_id: 'resumed', is_backgrounded: true })
|
||||
)
|
||||
expect(roles()[0].state).toBe('completed')
|
||||
})
|
||||
})
|
||||
|
||||
describe('ClaudeSubagentRoster — invocation fences', () => {
|
||||
it('ignores a previous invocation tool result even without a background flag', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1', tool_use_id: 'first' }))
|
||||
roster.observeToolResult('first', false)
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1', tool_use_id: 'next' }))
|
||||
roster.observeToolResult('first', true)
|
||||
expect(roles()[0].state).toBe('working')
|
||||
roster.observeToolResult('next', false)
|
||||
expect(roles()[0].state).toBe('completed')
|
||||
})
|
||||
|
||||
it('does not treat an evicted alias as a new invocation', () => {
|
||||
const { roster, rolesIn, setGroupKey } = harness()
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1', tool_use_id: 'first' }))
|
||||
roster.observeToolResult('first', false)
|
||||
setGroupKey('churn')
|
||||
for (let i = 0; i < 513; i++) {
|
||||
roster.observeSystemFrame(
|
||||
system('task_updated', { task_id: `other-${i}`, tool_use_id: `tool-${i}` })
|
||||
)
|
||||
}
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1', tool_use_id: 'first' }))
|
||||
expect(rolesIn(TURN_1)[0].state).toBe('completed')
|
||||
})
|
||||
|
||||
it('bounds invocation history and refuses to reopen beyond the retained budget', () => {
|
||||
const { roster, roles } = harness()
|
||||
for (let i = 0; i < 20; i++) {
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1', tool_use_id: `tool-${i}` }))
|
||||
if (i >= 16) {
|
||||
expect(roles()[0].state).toBe('unverifiable')
|
||||
}
|
||||
roster.observeToolResult(`tool-${i}`, false)
|
||||
}
|
||||
roster.observeSystemFrame(started({ task_id: 'task-1', tool_use_id: 'tool-0' }))
|
||||
expect(roles()[0].state).toBe('unverifiable')
|
||||
})
|
||||
})
|
||||
|
||||
it('merges an explicit foreground patch without clearing on absent metadata', () => {
|
||||
const { roster, roles } = harness()
|
||||
roster.observeSystemFrame(
|
||||
started({ task_id: 'task-1', tool_use_id: 'tool', is_backgrounded: true })
|
||||
)
|
||||
roster.observeSystemFrame(
|
||||
system('task_updated', { task_id: 'task-1', patch: { description: 'Audit' } })
|
||||
)
|
||||
roster.observeToolResult('tool', false)
|
||||
expect(roles()[0].state).toBe('working')
|
||||
roster.observeSystemFrame(
|
||||
system('task_updated', { task_id: 'task-1', patch: { is_backgrounded: false } })
|
||||
)
|
||||
roster.observeToolResult('tool', false)
|
||||
expect(roles()[0].state).toBe('completed')
|
||||
})
|
||||
@@ -0,0 +1,388 @@
|
||||
// The Claude subagent roster: one journal row per turn that spawned children.
|
||||
//
|
||||
// Entries are built from `task_started`, never from child traffic: a
|
||||
// BACKGROUNDED subagent emits no child frames at all, so a roster fed by
|
||||
// `parent_tool_use_id` alone would leave every one of them an unlabelled row
|
||||
// forever. Child traffic only creates an entry for CLI releases that announce
|
||||
// no task frames.
|
||||
//
|
||||
// Claude re-announces a resumed task under a NEW `tool_use_id`, so `task_id` is
|
||||
// the key and tool ids are aliases; keying on the tool id would duplicate the
|
||||
// child on every resume. Outcomes latch within an invocation; a new spawn
|
||||
// alias can reopen it, and authoritative evidence can correct lost contact.
|
||||
|
||||
import {
|
||||
canReplaceSubagentState,
|
||||
isTerminalSubagentState
|
||||
} from '../../shared/native-chat-subagent-summary'
|
||||
import type { NativeChatSubagentEntry } from '../../shared/native-chat-types'
|
||||
import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink'
|
||||
import { isBoundedClaudeTaskId } from './claude-background-task-tracker'
|
||||
import { claudeSubagentGroupBody, claudeSubagentGroupIdentity } from './claude-subagent-group-row'
|
||||
import { ClaudeSubagentIds } from './claude-subagent-id-aliases'
|
||||
import { readClaudeSubagentTaskFrame } from './claude-subagent-task-frames'
|
||||
import {
|
||||
applyClaudeSubagentInvocation,
|
||||
claimClaudeSubagentLabel,
|
||||
type RosterGroup,
|
||||
type TrackedEntry
|
||||
} from './claude-subagent-roster-state'
|
||||
|
||||
/** Spawn-group rows kept live per session, and children per row. Both bound an
|
||||
* event-accumulated map that no provider snapshot ever prunes. */
|
||||
const MAX_SUBAGENT_GROUPS = 32
|
||||
const MAX_SUBAGENTS_PER_GROUP = 64
|
||||
|
||||
/** The turn a group belongs to when Claude reports a task outside any turn. */
|
||||
const OUTSIDE_TURN = 'outside-turn'
|
||||
|
||||
const UNLABELLED_AGENT = 'subagent'
|
||||
|
||||
export type ClaudeSubagentRosterDeps = {
|
||||
sink: StructuredAgentSessionEventSink
|
||||
/** The turn that owns children spawned right now; null outside any turn. */
|
||||
currentGroupKey: () => string | null
|
||||
now?: () => number
|
||||
}
|
||||
|
||||
export class ClaudeSubagentRoster {
|
||||
private readonly groups = new Map<string, RosterGroup>()
|
||||
/** Canonical id → the group holding its entry, so a late update for a child
|
||||
* from an earlier turn revises that turn's row instead of the live one. */
|
||||
private readonly groupIdByEntry = new Map<string, string>()
|
||||
private readonly ids = new ClaudeSubagentIds()
|
||||
/** Set by ANY `task_started`, including one the subagent filter rejects. Once
|
||||
* this CLI has proven it declares its tasks, child traffic for an id it never
|
||||
* announced is a nested tool or a grandchild, not a subagent. */
|
||||
private announcesTasks = false
|
||||
private readonly now: () => number
|
||||
|
||||
constructor(private readonly deps: ClaudeSubagentRosterDeps) {
|
||||
this.now = deps.now ?? (() => Date.now())
|
||||
}
|
||||
|
||||
/** Consume a `message:system:task_*` frame. Returns false when it is not one. */
|
||||
observeSystemFrame(message: Record<string, unknown>): boolean {
|
||||
const frame = readClaudeSubagentTaskFrame(message)
|
||||
if (!frame) {
|
||||
return false
|
||||
}
|
||||
this.announcesTasks ||= frame.announcement
|
||||
if (frame.excluded) {
|
||||
// Child traffic may already have built a provisional row under the tool id;
|
||||
// the announcement is the first frame that says it is not a subagent.
|
||||
for (const id of [frame.taskId, frame.toolUseId]) {
|
||||
if (id !== null) {
|
||||
this.ids.exclude(id)
|
||||
this.remove(id)
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
if (this.ids.isExcluded(frame.taskId, frame.toolUseId)) {
|
||||
return true
|
||||
}
|
||||
if (frame.toolUseId) {
|
||||
this.ids.alias(frame.toolUseId, frame.taskId)
|
||||
}
|
||||
const located =
|
||||
this.locate(frame.taskId) ??
|
||||
(frame.toolUseId ? this.adopt(frame.toolUseId, frame.taskId) : null)
|
||||
if (!located) {
|
||||
if (frame.announcesSubagent) {
|
||||
this.create(
|
||||
frame.taskId,
|
||||
frame.label,
|
||||
frame.state ?? 'working',
|
||||
frame.backgrounded ?? false,
|
||||
frame.toolUseId
|
||||
)
|
||||
}
|
||||
return true
|
||||
}
|
||||
const tracked = located.group.entries.get(frame.taskId)
|
||||
if (tracked && !applyClaudeSubagentInvocation(tracked, frame, this.now)) {
|
||||
return true
|
||||
}
|
||||
this.revise(located.group, frame.taskId, {
|
||||
label: frame.label,
|
||||
state: frame.state,
|
||||
backgrounded: frame.backgrounded
|
||||
})
|
||||
return true
|
||||
}
|
||||
|
||||
/**
|
||||
* A frame carrying `parent_tool_use_id` — the child's own traffic. It refreshes
|
||||
* nothing on an announced child; it exists so a CLI release that sends no task
|
||||
* frames still shows the subagent it is running.
|
||||
*/
|
||||
observeChildActivity(parentToolUseId: string): void {
|
||||
const canonical = this.ids.canonical(parentToolUseId)
|
||||
if (this.ids.isExcluded(parentToolUseId, canonical)) {
|
||||
return
|
||||
}
|
||||
if (this.locate(canonical)) {
|
||||
return
|
||||
}
|
||||
if (this.announcesTasks) {
|
||||
// A nested Task, a workflow child, or a grandchild parented to a tool id
|
||||
// inside the sidechain all reach here. This CLI announces what it spawns,
|
||||
// so an id it never declared cannot be a subagent — and a row invented for
|
||||
// one is unlabelled forever and can only ever end `unverifiable`. The
|
||||
// bounded exclusion set cannot cover an id that was never announced.
|
||||
return
|
||||
}
|
||||
if (!isBoundedClaudeTaskId(canonical)) {
|
||||
// `claudeTaskId` rejects an over-long announced id rather than truncating
|
||||
// it; a provisional id becomes the same durable entry key, so it cannot
|
||||
// enter under a looser rule.
|
||||
return
|
||||
}
|
||||
this.create(canonical, null, 'working', false, parentToolUseId)
|
||||
}
|
||||
|
||||
/**
|
||||
* The parent turn's tool result for a spawn call. It settles a foreground
|
||||
* child, whose result IS the turn's evidence the child finished. A backgrounded
|
||||
* child's spawn call returns immediately while the child keeps running, so its
|
||||
* result proves nothing and is ignored.
|
||||
*/
|
||||
observeToolResult(toolUseId: string, failed: boolean): void {
|
||||
const canonical = this.ids.canonical(toolUseId)
|
||||
const located = this.locate(canonical)
|
||||
if (
|
||||
!located ||
|
||||
located.tracked.invocationIds === null ||
|
||||
located.tracked.backgrounded ||
|
||||
(located.tracked.toolUseId !== null && located.tracked.toolUseId !== toolUseId)
|
||||
) {
|
||||
return
|
||||
}
|
||||
this.revise(located.group, canonical, {
|
||||
label: null,
|
||||
state: failed ? 'failed' : 'completed',
|
||||
backgrounded: false
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* The parent turn ended. A foreground child still reported as working will
|
||||
* never be settled by an event, so it becomes `unverifiable`: contact was
|
||||
* lost, which is NOT evidence the child exited. A backgrounded child was
|
||||
* explicitly told to outlive the turn and is left alone.
|
||||
*/
|
||||
settleTurn(groupKey: string | null): void {
|
||||
// Only the group this key names. `OUTSIDE_TURN` belongs to no turn, so an
|
||||
// unrelated turn ending is no evidence about a child announced outside it.
|
||||
// `settleSession` reaches what no turn does.
|
||||
this.sweep(this.groups.get(groupKey ?? OUTSIDE_TURN), false)
|
||||
}
|
||||
|
||||
/** The provider is gone. Nothing more will arrive for any child, backgrounded
|
||||
* or not, so every one of them loses contact at once. */
|
||||
settleSession(): void {
|
||||
for (const group of this.groups.values()) {
|
||||
this.sweep(group, true)
|
||||
}
|
||||
}
|
||||
|
||||
dispose(): void {
|
||||
// Teardown paths reach here without an `ended` event, so a row still
|
||||
// reporting `working` would have nothing left to revise it. A session that
|
||||
// did settle first leaves every child terminal, so this writes nothing.
|
||||
this.settleSession()
|
||||
this.groups.clear()
|
||||
this.groupIdByEntry.clear()
|
||||
this.ids.clear()
|
||||
this.announcesTasks = false
|
||||
}
|
||||
|
||||
private sweep(group: RosterGroup | undefined, includeBackgrounded: boolean): void {
|
||||
if (!group) {
|
||||
return
|
||||
}
|
||||
let changed = false
|
||||
for (const [id, tracked] of group.entries) {
|
||||
if (isTerminalSubagentState(tracked.entry.state)) {
|
||||
continue
|
||||
}
|
||||
if (tracked.backgrounded && !includeBackgrounded) {
|
||||
continue
|
||||
}
|
||||
group.entries.set(id, {
|
||||
...tracked,
|
||||
entry: { ...tracked.entry, state: 'unverifiable', settledAt: this.now() }
|
||||
})
|
||||
changed = true
|
||||
}
|
||||
if (changed) {
|
||||
this.write(group)
|
||||
}
|
||||
}
|
||||
|
||||
private create(
|
||||
id: string,
|
||||
label: string | null,
|
||||
state: NativeChatSubagentEntry['state'],
|
||||
backgrounded: boolean,
|
||||
toolUseId: string | null
|
||||
): void {
|
||||
const group = this.groupFor()
|
||||
if (group.admittedEntries >= MAX_SUBAGENTS_PER_GROUP) {
|
||||
return
|
||||
}
|
||||
group.admittedEntries += 1
|
||||
const now = this.now()
|
||||
const labelBase = label ?? UNLABELLED_AGENT
|
||||
group.entries.set(id, {
|
||||
backgrounded,
|
||||
toolUseId,
|
||||
invocationIds: new Set(toolUseId ? [toolUseId] : []),
|
||||
labelBase,
|
||||
entry: {
|
||||
id,
|
||||
label: claimClaudeSubagentLabel(group, labelBase),
|
||||
state,
|
||||
startedAt: now,
|
||||
...(isTerminalSubagentState(state) ? { settledAt: now } : {})
|
||||
}
|
||||
})
|
||||
this.groupIdByEntry.set(id, group.groupId)
|
||||
this.write(group)
|
||||
}
|
||||
|
||||
private revise(
|
||||
group: RosterGroup,
|
||||
id: string,
|
||||
change: {
|
||||
label: string | null
|
||||
state: NativeChatSubagentEntry['state'] | null
|
||||
backgrounded: boolean | null
|
||||
}
|
||||
): void {
|
||||
const tracked = group.entries.get(id)
|
||||
if (!tracked) {
|
||||
return
|
||||
}
|
||||
const next: TrackedEntry = {
|
||||
...tracked,
|
||||
backgrounded: change.backgrounded ?? tracked.backgrounded,
|
||||
entry: { ...tracked.entry }
|
||||
}
|
||||
// A provisional row built from child traffic takes the real name the first
|
||||
// announcement carries; an announced row keeps the name it was given.
|
||||
if (
|
||||
change.label &&
|
||||
tracked.labelBase === UNLABELLED_AGENT &&
|
||||
change.label !== UNLABELLED_AGENT
|
||||
) {
|
||||
next.labelBase = change.label
|
||||
next.entry.label = claimClaudeSubagentLabel(group, change.label)
|
||||
}
|
||||
// Proven outcomes latch; lost contact can still receive a later verdict.
|
||||
if (change.state && canReplaceSubagentState(tracked.entry.state, change.state)) {
|
||||
next.entry.state = change.state
|
||||
if (isTerminalSubagentState(change.state)) {
|
||||
next.entry.settledAt = this.now()
|
||||
}
|
||||
}
|
||||
group.entries.set(id, next)
|
||||
this.write(group)
|
||||
}
|
||||
|
||||
/** Re-key a provisional entry from its tool id onto the canonical task id the
|
||||
* announcement finally named, so the child does not appear twice. */
|
||||
private adopt(toolUseId: string, taskId: string): { group: RosterGroup } | null {
|
||||
if (toolUseId === taskId) {
|
||||
return null
|
||||
}
|
||||
const located = this.locate(toolUseId)
|
||||
if (!located) {
|
||||
return null
|
||||
}
|
||||
located.group.entries.delete(toolUseId)
|
||||
located.group.entries.set(taskId, {
|
||||
...located.tracked,
|
||||
entry: { ...located.tracked.entry, id: taskId }
|
||||
})
|
||||
this.groupIdByEntry.delete(toolUseId)
|
||||
this.groupIdByEntry.set(taskId, located.group.groupId)
|
||||
return { group: located.group }
|
||||
}
|
||||
|
||||
private remove(id: string): void {
|
||||
const located = this.locate(id)
|
||||
if (!located) {
|
||||
return
|
||||
}
|
||||
located.group.entries.delete(id)
|
||||
this.groupIdByEntry.delete(id)
|
||||
this.write(located.group)
|
||||
}
|
||||
|
||||
private locate(id: string): { group: RosterGroup; tracked: TrackedEntry } | null {
|
||||
const groupId = this.groupIdByEntry.get(id)
|
||||
const group = groupId === undefined ? undefined : this.groups.get(groupId)
|
||||
const tracked = group?.entries.get(id)
|
||||
return group && tracked ? { group, tracked } : null
|
||||
}
|
||||
|
||||
private groupFor(): RosterGroup {
|
||||
const groupId = this.deps.currentGroupKey() ?? OUTSIDE_TURN
|
||||
const existing = this.groups.get(groupId)
|
||||
if (existing) {
|
||||
return existing
|
||||
}
|
||||
const group: RosterGroup = {
|
||||
groupId,
|
||||
identity: claudeSubagentGroupIdentity(groupId),
|
||||
entries: new Map(),
|
||||
admittedEntries: 0,
|
||||
claimedLabels: new Set(),
|
||||
lastSerialized: null
|
||||
}
|
||||
this.groups.set(groupId, group)
|
||||
while (this.groups.size > MAX_SUBAGENT_GROUPS) {
|
||||
const oldest = this.groups.keys().next()
|
||||
if (oldest.done || oldest.value === groupId) {
|
||||
break
|
||||
}
|
||||
const evicted = this.groups.get(oldest.value)
|
||||
// Once the group leaves the map nothing can reach its children again —
|
||||
// not even a session sweep — so contact is lost here.
|
||||
this.sweep(evicted, true)
|
||||
for (const id of evicted?.entries.keys() ?? []) {
|
||||
this.groupIdByEntry.delete(id)
|
||||
}
|
||||
this.groups.delete(oldest.value)
|
||||
}
|
||||
return group
|
||||
}
|
||||
|
||||
private write(group: RosterGroup): void {
|
||||
const agents = [...group.entries.values()].map((tracked) => tracked.entry)
|
||||
const options = { coalescingKey: `claude-subagents:${group.groupId}` }
|
||||
if (agents.length === 0) {
|
||||
// The row's last child turned out not to be a subagent. An empty roster is
|
||||
// not a roster of nothing, so the row goes rather than reading "Ran 0".
|
||||
if (group.lastSerialized !== null) {
|
||||
group.lastSerialized = null
|
||||
this.deps.sink.appendTombstone(group.identity, options)
|
||||
this.deps.sink.publish()
|
||||
}
|
||||
return
|
||||
}
|
||||
const body = claudeSubagentGroupBody(group.groupId, agents)
|
||||
const serialized = JSON.stringify(body)
|
||||
if (serialized === group.lastSerialized) {
|
||||
// Nothing changed — a duplicate delivery must not burn a revision.
|
||||
return
|
||||
}
|
||||
group.lastSerialized = serialized
|
||||
this.deps.sink.appendItem(group.identity, body, options)
|
||||
// Publish keeps the sink's own coalescing slot: sharing the row's key makes
|
||||
// each queued publish evict the append it was meant to flush.
|
||||
this.deps.sink.publish()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,201 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { readClaudeSubagentTaskFrame } from './claude-subagent-task-frames'
|
||||
|
||||
function system(subtype: string, fields: Record<string, unknown>): Record<string, unknown> {
|
||||
return { type: 'system', subtype, session_id: 'claude-session', ...fields }
|
||||
}
|
||||
|
||||
describe('readClaudeSubagentTaskFrame', () => {
|
||||
it('ignores frames that are not task frames', () => {
|
||||
expect(readClaudeSubagentTaskFrame({ type: 'assistant', subtype: 'task_started' })).toBeNull()
|
||||
expect(readClaudeSubagentTaskFrame(system('init', { task_id: 'task-1' }))).toBeNull()
|
||||
expect(readClaudeSubagentTaskFrame(system('task_started', {}))).toBeNull()
|
||||
expect(readClaudeSubagentTaskFrame(system('task_started', { task_id: '' }))).toBeNull()
|
||||
})
|
||||
|
||||
describe('task_type triage', () => {
|
||||
it('announces a local_agent task', () => {
|
||||
const frame = readClaudeSubagentTaskFrame(
|
||||
system('task_started', {
|
||||
task_id: 'task-1',
|
||||
tool_use_id: 'toolu_1',
|
||||
task_type: 'local_agent',
|
||||
subagent_type: 'code-reviewer',
|
||||
description: 'Review the diff'
|
||||
})
|
||||
)
|
||||
expect(frame).toMatchObject({
|
||||
taskId: 'task-1',
|
||||
toolUseId: 'toolu_1',
|
||||
label: 'Review the diff',
|
||||
announcesSubagent: true,
|
||||
excluded: false
|
||||
})
|
||||
})
|
||||
|
||||
it('excludes a backgrounded shell command even though it carries a tool_use_id', () => {
|
||||
const frame = readClaudeSubagentTaskFrame(
|
||||
system('task_started', {
|
||||
task_id: 'task-bash',
|
||||
tool_use_id: 'toolu_bash',
|
||||
task_type: 'local_bash',
|
||||
description: 'sleep 20',
|
||||
is_backgrounded: true
|
||||
})
|
||||
)
|
||||
expect(frame).toMatchObject({
|
||||
taskId: 'task-bash',
|
||||
toolUseId: 'toolu_bash',
|
||||
announcesSubagent: false,
|
||||
excluded: true
|
||||
})
|
||||
})
|
||||
|
||||
it('excludes workflows and monitors', () => {
|
||||
for (const taskType of ['local_workflow', 'monitor']) {
|
||||
expect(
|
||||
readClaudeSubagentTaskFrame(
|
||||
system('task_started', { task_id: `task-${taskType}`, task_type: taskType })
|
||||
)
|
||||
).toMatchObject({ announcesSubagent: false, excluded: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('caps a subagent_type label the way a description is capped', () => {
|
||||
const frame = readClaudeSubagentTaskFrame(
|
||||
system('task_started', { task_id: 'task-1', subagent_type: 'a'.repeat(900) })
|
||||
)
|
||||
// The roster stores this label verbatim, so nothing downstream bounds it.
|
||||
expect(frame?.label).toHaveLength(512)
|
||||
})
|
||||
|
||||
it('falls back to subagent_type only when the release sends no task_type', () => {
|
||||
expect(
|
||||
readClaudeSubagentTaskFrame(
|
||||
system('task_started', { task_id: 'task-old', subagent_type: 'explorer' })
|
||||
)
|
||||
).toMatchObject({ announcesSubagent: true, label: 'explorer' })
|
||||
expect(
|
||||
readClaudeSubagentTaskFrame(system('task_started', { task_id: 'task-bare' }))
|
||||
).toMatchObject({ announcesSubagent: false, excluded: true })
|
||||
// A type this build does not recognise is not an agent on subagent_type's word.
|
||||
expect(
|
||||
readClaudeSubagentTaskFrame(
|
||||
system('task_started', {
|
||||
task_id: 'task-new',
|
||||
task_type: 'local_something_new',
|
||||
subagent_type: 'explorer'
|
||||
})
|
||||
)
|
||||
).toMatchObject({ announcesSubagent: false, excluded: true })
|
||||
})
|
||||
|
||||
it('excludes ambient housekeeping tasks', () => {
|
||||
for (const suppression of [{ skip_transcript: true }, { ambient: true }]) {
|
||||
expect(
|
||||
readClaudeSubagentTaskFrame(
|
||||
system('task_started', {
|
||||
task_id: 'task-ambient',
|
||||
task_type: 'local_agent',
|
||||
subagent_type: 'watcher',
|
||||
...suppression
|
||||
})
|
||||
)
|
||||
).toMatchObject({ announcesSubagent: false, excluded: true })
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
describe('status', () => {
|
||||
it('collapses every in-flight status to working', () => {
|
||||
for (const status of ['pending', 'running', 'paused']) {
|
||||
expect(
|
||||
readClaudeSubagentTaskFrame(
|
||||
system('task_updated', { task_id: 'task-1', patch: { status } })
|
||||
)
|
||||
).toMatchObject({ state: 'working' })
|
||||
}
|
||||
})
|
||||
|
||||
it('maps the settled statuses onto the carrier vocabulary', () => {
|
||||
const mapped: [string, string][] = [
|
||||
['completed', 'completed'],
|
||||
['failed', 'failed'],
|
||||
['killed', 'stopped'],
|
||||
['stopped', 'stopped']
|
||||
]
|
||||
for (const [status, state] of mapped) {
|
||||
expect(
|
||||
readClaudeSubagentTaskFrame(
|
||||
system('task_updated', { task_id: 'task-1', patch: { status } })
|
||||
)
|
||||
).toMatchObject({ state })
|
||||
}
|
||||
})
|
||||
|
||||
it('reports no state for a status it cannot map', () => {
|
||||
for (const status of ['__proto__', 'toString', 'invented', 7, null]) {
|
||||
expect(
|
||||
readClaudeSubagentTaskFrame(
|
||||
system('task_updated', { task_id: 'task-1', patch: { status } })
|
||||
)
|
||||
).toMatchObject({ state: null })
|
||||
}
|
||||
})
|
||||
|
||||
it('treats progress as no lifecycle verdict', () => {
|
||||
for (const subtype of ['task_progress']) {
|
||||
expect(
|
||||
readClaudeSubagentTaskFrame(
|
||||
system(subtype, { task_id: 'task-1', status: 'completed', patch: { status: 'failed' } })
|
||||
)
|
||||
).toMatchObject({ state: null })
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
it('reads the notification verdict from its top-level status', () => {
|
||||
for (const state of ['completed', 'failed', 'stopped']) {
|
||||
expect(
|
||||
readClaudeSubagentTaskFrame(
|
||||
system('task_notification', {
|
||||
task_id: 'task-1',
|
||||
status: state,
|
||||
patch: { status: 'running' }
|
||||
})
|
||||
)
|
||||
).toMatchObject({ state })
|
||||
}
|
||||
})
|
||||
|
||||
it('reads the backgrounded flag from the frame or its patch', () => {
|
||||
expect(
|
||||
readClaudeSubagentTaskFrame(
|
||||
system('task_started', {
|
||||
task_id: 'task-1',
|
||||
task_type: 'local_agent',
|
||||
is_backgrounded: true
|
||||
})
|
||||
)
|
||||
).toMatchObject({ backgrounded: true })
|
||||
expect(
|
||||
readClaudeSubagentTaskFrame(
|
||||
system('task_updated', { task_id: 'task-1', patch: { is_backgrounded: true } })
|
||||
)
|
||||
).toMatchObject({ backgrounded: true })
|
||||
expect(
|
||||
readClaudeSubagentTaskFrame(system('task_updated', { task_id: 'task-1', patch: {} }))
|
||||
).toMatchObject({ backgrounded: null })
|
||||
})
|
||||
|
||||
it('collapses a multi-line description into one bounded label', () => {
|
||||
expect(
|
||||
readClaudeSubagentTaskFrame(
|
||||
system('task_updated', {
|
||||
task_id: 'task-1',
|
||||
patch: { description: ' audit\n the lockfile ' }
|
||||
})
|
||||
)
|
||||
).toMatchObject({ label: 'audit the lockfile' })
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,123 @@
|
||||
// Claude's declarative task protocol, read as subagent roster events.
|
||||
//
|
||||
// `local_agent`, `local_workflow` and `local_bash` tasks all arrive on the same
|
||||
// `message:system:task_*` channel and ALL carry a `tool_use_id`, so id presence
|
||||
// discriminates nothing: filtering on it alone puts a backgrounded `sleep 20` in
|
||||
// the subagent roster. `task_type` is the discriminator, with `subagent_type`
|
||||
// covering CLI releases that predate it.
|
||||
|
||||
import type { NativeChatSubagentState } from '../../shared/native-chat-types'
|
||||
import {
|
||||
classifyClaudeBackgroundTaskKind,
|
||||
claudeTaskDescription,
|
||||
claudeTaskId,
|
||||
isBoundedClaudeTaskId
|
||||
} from './claude-background-task-tracker'
|
||||
import { claudeRecord, claudeText } from './claude-structured-item-translation'
|
||||
|
||||
const TASK_SUBTYPES: ReadonlySet<string> = new Set([
|
||||
'task_started',
|
||||
'task_updated',
|
||||
'task_progress',
|
||||
'task_notification'
|
||||
])
|
||||
|
||||
/** Provider status → the carrier's vocabulary. `killed` and `stopped` both mean
|
||||
* the task was deliberately ended, which the carrier calls `stopped`; every
|
||||
* in-flight status collapses to `working`. A Map, not an object, so a payload
|
||||
* carrying `__proto__` as its status cannot resolve to an inherited value. */
|
||||
const TASK_STATES: ReadonlyMap<string, NativeChatSubagentState> = new Map([
|
||||
['pending', 'working'],
|
||||
['running', 'working'],
|
||||
['paused', 'working'],
|
||||
['completed', 'completed'],
|
||||
['failed', 'failed'],
|
||||
['killed', 'stopped'],
|
||||
['stopped', 'stopped']
|
||||
] satisfies [string, NativeChatSubagentState][])
|
||||
|
||||
export type ClaudeSubagentTaskFrame = {
|
||||
/** Canonical, resume-stable id — the roster key. */
|
||||
taskId: string
|
||||
/** Re-minted when Claude re-announces a resumed task, so it is only an alias. */
|
||||
toolUseId: string | null
|
||||
label: string | null
|
||||
/** null when the frame reported no lifecycle status. */
|
||||
state: NativeChatSubagentState | null
|
||||
backgrounded: boolean | null
|
||||
/** Any `task_started`, subagent or not. Proof this CLI declares its tasks. */
|
||||
announcement: boolean
|
||||
/** `task_started` for a task the roster should show. Only an announcement
|
||||
* creates an entry: an update carries no `task_type`, so honouring one for an
|
||||
* unknown id would roster whatever else shares this channel. */
|
||||
announcesSubagent: boolean
|
||||
/** Ambient housekeeping, or a task that is not a subagent at all. Its ids must
|
||||
* never reach the roster, by this frame or by later child traffic. */
|
||||
excluded: boolean
|
||||
}
|
||||
|
||||
/** True when the task Claude announced is a subagent rather than a backgrounded
|
||||
* shell command or a workflow. */
|
||||
export function isClaudeSubagentTask(message: Record<string, unknown>): boolean {
|
||||
if (classifyClaudeBackgroundTaskKind(message.task_type) === 'agent') {
|
||||
return true
|
||||
}
|
||||
// Releases predating `task_type` still name the child in `subagent_type`. A
|
||||
// task_type Orca does not recognise is NOT covered: it is a type this build
|
||||
// has no reason to believe is an agent.
|
||||
return (
|
||||
(message.task_type === undefined || message.task_type === null) &&
|
||||
claudeText(message.subagent_type) !== null
|
||||
)
|
||||
}
|
||||
|
||||
function taskState(value: unknown): NativeChatSubagentState | null {
|
||||
return typeof value === 'string' ? (TASK_STATES.get(value) ?? null) : null
|
||||
}
|
||||
|
||||
export function readClaudeSubagentTaskFrame(
|
||||
message: Record<string, unknown>
|
||||
): ClaudeSubagentTaskFrame | null {
|
||||
if (message.type !== 'system') {
|
||||
return null
|
||||
}
|
||||
const subtype = claudeText(message.subtype)
|
||||
if (!subtype || !TASK_SUBTYPES.has(subtype)) {
|
||||
return null
|
||||
}
|
||||
const taskId = claudeTaskId(message)
|
||||
if (!taskId) {
|
||||
return null
|
||||
}
|
||||
const patch = claudeRecord(message.patch)
|
||||
const toolUseId = claudeText(message.tool_use_id) ?? claudeText(patch?.tool_use_id)
|
||||
const announcement = subtype === 'task_started'
|
||||
// Housekeeping Claude runs for itself; the user never asked for it.
|
||||
const suppressed = message.ambient === true || message.skip_transcript === true
|
||||
const subagent = announcement && !suppressed && isClaudeSubagentTask(message)
|
||||
return {
|
||||
taskId,
|
||||
toolUseId: toolUseId && isBoundedClaudeTaskId(toolUseId) ? toolUseId : null,
|
||||
label:
|
||||
claudeTaskDescription(message.description) ??
|
||||
claudeTaskDescription(patch?.description) ??
|
||||
// Bounded like a description: the roster stores whatever this returns.
|
||||
(announcement ? (claudeTaskDescription(message.subagent_type) ?? null) : null),
|
||||
// Notifications carry terminal evidence; progress carries usage only.
|
||||
state:
|
||||
subtype === 'task_notification'
|
||||
? taskState(message.status)
|
||||
: announcement || subtype === 'task_updated'
|
||||
? taskState(patch?.status ?? message.status)
|
||||
: null,
|
||||
backgrounded:
|
||||
typeof patch?.is_backgrounded === 'boolean'
|
||||
? patch.is_backgrounded
|
||||
: typeof message.is_backgrounded === 'boolean'
|
||||
? message.is_backgrounded
|
||||
: null,
|
||||
announcement,
|
||||
announcesSubagent: subagent,
|
||||
excluded: announcement && !subagent
|
||||
}
|
||||
}
|
||||
@@ -432,6 +432,86 @@ describe('payload bounds on import', () => {
|
||||
expect(body.output.byteLength).toBe(64 * 1024)
|
||||
expect(body.output.head).toHaveLength(1_024)
|
||||
})
|
||||
|
||||
it('bounds an imported subagent roster by entry count, label and id', async () => {
|
||||
// The import reads an untrusted file: nothing upstream bounded either string.
|
||||
const oversized = 'z'.repeat(20 * 1024)
|
||||
const journal = await open('claude', CLAUDE_SESSION)
|
||||
await appendLegacyTranscriptMessages({
|
||||
journal,
|
||||
agent: 'claude',
|
||||
sessionId: CLAUDE_SESSION,
|
||||
fence: 1,
|
||||
messages: [
|
||||
{
|
||||
id: 'legacy-roster',
|
||||
role: 'assistant',
|
||||
timestamp: null,
|
||||
source: 'transcript',
|
||||
blocks: [
|
||||
{
|
||||
type: 'subagent-group',
|
||||
groupId: 'group-1',
|
||||
agents: Array.from({ length: 80 }, (_, index) => ({
|
||||
id: index === 0 ? oversized : `task-${index}`,
|
||||
label: index === 0 ? oversized : `label-${index}`,
|
||||
state: 'working' as const
|
||||
}))
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
})
|
||||
|
||||
const body = journal.snapshot().items[0]?.body
|
||||
const block = body?.kind === 'message' ? body.blocks[0] : undefined
|
||||
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]?.id.length).toBeLessThan(oversized.length)
|
||||
expect(block.agents[0]?.id.startsWith('z')).toBe(true)
|
||||
})
|
||||
|
||||
it('bounds a roster id in the shared format, keeping a shared prefix distinct', async () => {
|
||||
// The id is the roster key: it takes the same bounded-id format the wires
|
||||
// use, so a later wire bound is a no-op instead of a second, merging clip.
|
||||
const head = 'y'.repeat(512)
|
||||
const journal = await open('claude', CLAUDE_SESSION)
|
||||
await appendLegacyTranscriptMessages({
|
||||
journal,
|
||||
agent: 'claude',
|
||||
sessionId: CLAUDE_SESSION,
|
||||
fence: 1,
|
||||
messages: [
|
||||
{
|
||||
id: 'legacy-roster-collision',
|
||||
role: 'assistant',
|
||||
timestamp: null,
|
||||
source: 'transcript',
|
||||
blocks: [
|
||||
{
|
||||
type: 'subagent-group',
|
||||
groupId: 'group-1',
|
||||
agents: [
|
||||
{ id: `${head}-one`, label: 'Audit', state: 'working' as const },
|
||||
{ id: `${head}-two`, label: 'Audit', state: 'working' as const }
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
})
|
||||
|
||||
const body = journal.snapshot().items[0]?.body
|
||||
const block = body?.kind === 'message' ? body.blocks[0] : undefined
|
||||
if (block?.type !== 'subagent-group') {
|
||||
throw new Error('expected a subagent-group block')
|
||||
}
|
||||
expect(block.agents[0]?.id).not.toBe(block.agents[1]?.id)
|
||||
expect(block.agents[0]?.id).toHaveLength(512)
|
||||
})
|
||||
})
|
||||
|
||||
describe('import failures', () => {
|
||||
|
||||
@@ -28,6 +28,7 @@ import {
|
||||
decodeOmpTranscriptLine
|
||||
} from '../transcript-line-decoders'
|
||||
import { decodeTranscriptStream } from '../transcript-stream-lines'
|
||||
import { boundSubagentEntryId } from '../subagent-entry-id-bounds'
|
||||
import { createLegacyIdentityTracker } from './journal-legacy-identity'
|
||||
import type { JournalReplacementItem } from './journal-epoch-replacement'
|
||||
import {
|
||||
@@ -47,6 +48,8 @@ export type LegacyImportOptions = ResolveSessionFileOptions & {
|
||||
}
|
||||
|
||||
const MAX_LEGACY_IMPORT_SOURCE_BYTES = 16 * 1024 * 1024
|
||||
/** A roster is a status list; an imported one is as untrusted as any other block. */
|
||||
const MAX_LEGACY_IMPORT_SUBAGENTS = 64
|
||||
|
||||
export type LegacyImportResult =
|
||||
| {
|
||||
@@ -271,5 +274,17 @@ function boundBlock(block: NativeChatBlock, limits: JournalPayloadLimits): Nativ
|
||||
if (block.type === 'tool-call') {
|
||||
return { ...block, input: boundToolInput(block.input, limits) }
|
||||
}
|
||||
if (block.type === 'subagent-group') {
|
||||
return {
|
||||
...block,
|
||||
agents: block.agents.slice(0, MAX_LEGACY_IMPORT_SUBAGENTS).map((agent) => ({
|
||||
...agent,
|
||||
// The id is the roster key, so it is bounded with a digest rather than
|
||||
// clipped to a prefix that two distinct children could share.
|
||||
id: boundSubagentEntryId(agent.id),
|
||||
label: boundInlineText(agent.label, limits).text
|
||||
}))
|
||||
}
|
||||
}
|
||||
return block
|
||||
}
|
||||
|
||||
@@ -5,7 +5,10 @@ import {
|
||||
boundJournalKeyComponent,
|
||||
MAX_JOURNAL_KEY_COMPONENT_CHARS
|
||||
} from '../../../shared/agent-session-journal-item-key'
|
||||
import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types'
|
||||
import type {
|
||||
AgentJournalItemIdentity,
|
||||
AgentJournalMessageItem
|
||||
} from '../../../shared/agent-session-journal-types'
|
||||
import { structuredAgentSessionPayloadFingerprint } from '../../../shared/structured-agent-session-mutation'
|
||||
import {
|
||||
applyJournalRow,
|
||||
@@ -14,6 +17,7 @@ import {
|
||||
renderJournalState,
|
||||
type JournalReducerState
|
||||
} from './journal-reducer'
|
||||
import { buildJournalItemRow, buildJournalTombstoneRow } from './journal-row-builders'
|
||||
import type { JournalRow } from './journal-row-schema'
|
||||
|
||||
const EPOCH = 'epoch-1'
|
||||
@@ -444,3 +448,49 @@ describe('bounded item-key collisions', () => {
|
||||
])
|
||||
})
|
||||
})
|
||||
|
||||
describe('re-adding a tombstoned row', () => {
|
||||
it('builds the rebuilt row above the tombstone that removed it', () => {
|
||||
const identity: AgentJournalItemIdentity = { provider: 'orca', clientMessageId: 'roster' }
|
||||
const itemId = agentJournalItemKey(identity)
|
||||
const state = createJournalReducerState('session-1', EPOCH)
|
||||
applyJournalRow(
|
||||
state,
|
||||
buildJournalItemRow({ state, identity, body: text('first'), seq: 1, fence: 1, ts: 1_001 })
|
||||
)
|
||||
applyJournalRow(state, buildJournalTombstoneRow({ state, itemId, seq: 2, fence: 1, ts: 1_002 }))
|
||||
expect(renderJournalState(state).items).toEqual([])
|
||||
|
||||
// Same identity, re-added later in the session: a revision built only from
|
||||
// `items` would restart at 1 and lose to the tombstone forever.
|
||||
applyJournalRow(
|
||||
state,
|
||||
buildJournalItemRow({ state, identity, body: text('second'), seq: 3, fence: 1, ts: 1_003 })
|
||||
)
|
||||
expect(renderJournalState(state).items.map((item) => item.body)).toEqual([text('second')])
|
||||
})
|
||||
|
||||
// `upsertItem` clearing the tombstone on a re-add is a map-state invariant:
|
||||
// `items` and `tombstones` stay disjoint, so a re-added row is never both
|
||||
// present and removed. Revision ordering is now independent of it —
|
||||
// `buildJournalTombstoneRow` takes `max(itemRevision, tombstoneRevision) + 1`
|
||||
// — so what this pins is the map state itself, not the ranking.
|
||||
it('removes the row again after it was re-added', () => {
|
||||
const identity: AgentJournalItemIdentity = { provider: 'orca', clientMessageId: 'roster' }
|
||||
const itemId = agentJournalItemKey(identity)
|
||||
const state = createJournalReducerState('session-1', EPOCH)
|
||||
applyJournalRow(
|
||||
state,
|
||||
buildJournalItemRow({ state, identity, body: text('first'), seq: 1, fence: 1, ts: 1_001 })
|
||||
)
|
||||
applyJournalRow(state, buildJournalTombstoneRow({ state, itemId, seq: 2, fence: 1, ts: 1_002 }))
|
||||
applyJournalRow(
|
||||
state,
|
||||
buildJournalItemRow({ state, identity, body: text('second'), seq: 3, fence: 1, ts: 1_003 })
|
||||
)
|
||||
expect(state.tombstones.get(itemId)).toBeUndefined()
|
||||
|
||||
applyJournalRow(state, buildJournalTombstoneRow({ state, itemId, seq: 4, fence: 1, ts: 1_004 }))
|
||||
expect(renderJournalState(state).items).toEqual([])
|
||||
})
|
||||
})
|
||||
|
||||
@@ -148,7 +148,13 @@ export function buildJournalItemRow(input: {
|
||||
}): JournalItemRow {
|
||||
const itemId = agentJournalItemKey(input.identity)
|
||||
const resolved = input.state.aliases.get(itemId) ?? itemId
|
||||
const revision = (input.state.items.get(resolved)?.revision ?? 0) + 1
|
||||
// A tombstoned row keeps its revision in `tombstones`, and the reducer drops
|
||||
// any item at or below it — so a re-add has to outrank the tombstone too.
|
||||
const revision =
|
||||
Math.max(
|
||||
input.state.items.get(resolved)?.revision ?? 0,
|
||||
input.state.tombstones.get(resolved) ?? 0
|
||||
) + 1
|
||||
return {
|
||||
kind: 'item',
|
||||
itemId,
|
||||
@@ -170,7 +176,15 @@ export function buildJournalTombstoneRow(input: {
|
||||
return {
|
||||
kind: 'tombstone',
|
||||
itemId: input.itemId,
|
||||
revision: (input.state.items.get(resolved)?.revision ?? 0) + 1,
|
||||
// Symmetric with the item builder: `upsertItem` clearing the tombstone on a
|
||||
// re-add is what keeps the two maps disjoint, and that invariant lives in the
|
||||
// reducer. Outranking both here means a repeat removal cannot be dropped as a
|
||||
// stale revision if it ever stops holding.
|
||||
revision:
|
||||
Math.max(
|
||||
input.state.items.get(resolved)?.revision ?? 0,
|
||||
input.state.tombstones.get(resolved) ?? 0
|
||||
) + 1,
|
||||
...journalRowBase(input.state.epoch, input.seq, input.fence, input.ts)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { boundSubagentEntryId, MAX_SUBAGENT_ENTRY_ID_CHARS } from './subagent-entry-id-bounds'
|
||||
|
||||
const SHARED_HEAD = 'a'.repeat(MAX_SUBAGENT_ENTRY_ID_CHARS)
|
||||
|
||||
describe('boundSubagentEntryId', () => {
|
||||
it('leaves an id that already fits untouched', () => {
|
||||
const id = 'task-1'
|
||||
expect(boundSubagentEntryId(id)).toBe(id)
|
||||
expect(boundSubagentEntryId(SHARED_HEAD)).toBe(SHARED_HEAD)
|
||||
})
|
||||
|
||||
it('keeps two ids sharing the whole cap-length head distinct', () => {
|
||||
const first = boundSubagentEntryId(`${SHARED_HEAD}-one`)
|
||||
const second = boundSubagentEntryId(`${SHARED_HEAD}-two`)
|
||||
expect(first).not.toBe(second)
|
||||
expect(first).toHaveLength(MAX_SUBAGENT_ENTRY_ID_CHARS)
|
||||
expect(second).toHaveLength(MAX_SUBAGENT_ENTRY_ID_CHARS)
|
||||
})
|
||||
|
||||
it('is deterministic and a no-op on an already bounded id', () => {
|
||||
const bounded = boundSubagentEntryId(`${SHARED_HEAD}-one`)
|
||||
expect(boundSubagentEntryId(`${SHARED_HEAD}-one`)).toBe(bounded)
|
||||
expect(boundSubagentEntryId(bounded)).toBe(bounded)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,27 @@
|
||||
// Bounding a subagent entry's id — a KEY, not display text.
|
||||
//
|
||||
// `NativeChatSubagentEntry.id` keys the roster, so clipping it to a fixed prefix
|
||||
// merges two distinct children whose ids agree that far and associates one's
|
||||
// state with the other. An id long enough to need bounding only reaches us from
|
||||
// an imported legacy transcript, but a bound is still required, so keep a head
|
||||
// for readability plus a digest of the WHOLE id to keep distinct ids distinct.
|
||||
|
||||
import { createHash } from 'node:crypto'
|
||||
|
||||
/** Shared by every site that puts a roster entry on a wire, so a journal-bound
|
||||
* id survives the RPC and transcript bounds untouched instead of being clipped
|
||||
* a second time into a different string. */
|
||||
export const MAX_SUBAGENT_ENTRY_ID_CHARS = 512
|
||||
|
||||
const DIGEST_CHARS = 16
|
||||
|
||||
/** Returns `id` unchanged when it fits, else a head plus a digest suffix whose
|
||||
* total length is exactly the cap — so bounding a bounded id is a no-op. */
|
||||
export function boundSubagentEntryId(id: string): string {
|
||||
if (id.length <= MAX_SUBAGENT_ENTRY_ID_CHARS) {
|
||||
return id
|
||||
}
|
||||
const digest = createHash('sha256').update(id, 'utf8').digest('base64url').slice(0, DIGEST_CHARS)
|
||||
const suffix = `…#${digest}`
|
||||
return `${id.slice(0, MAX_SUBAGENT_ENTRY_ID_CHARS - suffix.length)}${suffix}`
|
||||
}
|
||||
@@ -146,6 +146,36 @@ describe('worker transcript wire bounds', () => {
|
||||
expect(result).toMatchObject({ limited: false, warnings: [] })
|
||||
})
|
||||
|
||||
it('keeps two roster ids sharing a 512-char prefix distinct', () => {
|
||||
// The id is the roster key: a plain prefix clip would merge the two children.
|
||||
const head = 'a'.repeat(512)
|
||||
const result = boundWorkerTranscriptMessages([
|
||||
{
|
||||
id: 'message-1',
|
||||
role: 'assistant',
|
||||
timestamp: null,
|
||||
source: 'transcript',
|
||||
blocks: [
|
||||
{
|
||||
type: 'subagent-group',
|
||||
groupId: 'g',
|
||||
agents: [
|
||||
{ id: `${head}-one`, label: 'Audit', state: 'working' },
|
||||
{ id: `${head}-two`, label: 'Audit', state: 'working' }
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
])
|
||||
|
||||
const block = result.messages[0]?.blocks[0]
|
||||
if (block?.type !== 'subagent-group') {
|
||||
throw new Error('expected a subagent-group block')
|
||||
}
|
||||
expect(block.agents[0]?.id).not.toBe(block.agents[1]?.id)
|
||||
expect(block.agents[0]?.id).toHaveLength(512)
|
||||
})
|
||||
|
||||
it('keeps fallback identifiers stable without exposing the transcript path', () => {
|
||||
const transcriptPath = 'C:\\Users\\worker\\.codex\\session.jsonl'
|
||||
const message = {
|
||||
|
||||
@@ -5,6 +5,7 @@ import type {
|
||||
NativeChatMessage,
|
||||
NativeChatSubagentState
|
||||
} from '../../../shared/native-chat-types'
|
||||
import { boundSubagentEntryId } from '../../native-chat/subagent-entry-id-bounds'
|
||||
|
||||
export const DEFAULT_WORKER_TRANSCRIPT_MESSAGE_LIMIT = 40
|
||||
export const MAX_WORKER_TRANSCRIPT_MESSAGE_LIMIT = 50
|
||||
@@ -153,7 +154,7 @@ function boundBlock(block: NativeChatBlock, state: TranscriptBoundState): Native
|
||||
groupId: clipMetadata(block.groupId, state),
|
||||
agents: agents.map((agent) => ({
|
||||
...agent,
|
||||
id: clipMetadata(agent.id, state),
|
||||
id: boundEntryId(agent.id, state),
|
||||
label: clipMetadata(agent.label, state),
|
||||
state: clipSubagentState(agent.state, state)
|
||||
}))
|
||||
@@ -194,6 +195,18 @@ function isLocalFileLocator(value: string): boolean {
|
||||
)
|
||||
}
|
||||
|
||||
/** A roster entry's id is the roster KEY, so it is redacted like other metadata
|
||||
* but bounded with a digest rather than clipped: two ids sharing a 512-char
|
||||
* head must not collapse onto one entry. */
|
||||
function boundEntryId(value: string, state: TranscriptBoundState): string {
|
||||
const redacted = redactSensitiveText(value, state.warnings)
|
||||
const bounded = boundSubagentEntryId(redacted)
|
||||
if (bounded !== redacted) {
|
||||
markClipped(state, 'Oversized transcript metadata was clipped.')
|
||||
}
|
||||
return bounded
|
||||
}
|
||||
|
||||
function clipMetadata(value: string, state: TranscriptBoundState): string {
|
||||
const redacted = redactSensitiveText(value, state.warnings)
|
||||
if (redacted.length <= MAX_WORKER_TRANSCRIPT_METADATA_CHARS) {
|
||||
|
||||
@@ -0,0 +1,35 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import type { NativeChatBlock } from '../../../../shared/native-chat-types'
|
||||
import { sanitizeNativeChatRpcBlock } from './native-chat-rpc-block-sanitize'
|
||||
|
||||
const SHARED_HEAD = 'a'.repeat(512)
|
||||
|
||||
function rosterBlock(ids: readonly string[]): NativeChatBlock {
|
||||
return {
|
||||
type: 'subagent-group',
|
||||
groupId: 'group-1',
|
||||
agents: ids.map((id) => ({ id, label: 'l'.repeat(900), state: 'working' as const }))
|
||||
}
|
||||
}
|
||||
|
||||
describe('mobile subagent roster bounds', () => {
|
||||
it('keeps two ids sharing a 512-char prefix distinct', () => {
|
||||
const block = sanitizeNativeChatRpcBlock(
|
||||
rosterBlock([`${SHARED_HEAD}-one`, `${SHARED_HEAD}-two`]),
|
||||
'mobile'
|
||||
)
|
||||
|
||||
if (block.type !== 'subagent-group') {
|
||||
throw new Error('expected a subagent-group block')
|
||||
}
|
||||
expect(block.agents[0]?.id).not.toBe(block.agents[1]?.id)
|
||||
expect(block.agents[0]?.id).toHaveLength(512)
|
||||
// The label is display text and still clips.
|
||||
expect(block.agents[0]?.label).toContain('… (truncated)')
|
||||
})
|
||||
|
||||
it('leaves a short id alone', () => {
|
||||
const block = sanitizeNativeChatRpcBlock(rosterBlock(['task-1']), 'mobile')
|
||||
expect(block.type === 'subagent-group' && block.agents[0]?.id).toBe('task-1')
|
||||
})
|
||||
})
|
||||
@@ -3,6 +3,7 @@ import {
|
||||
normalizeSubagentState
|
||||
} from '../../../../shared/native-chat-subagent-summary'
|
||||
import type { NativeChatBlock, NativeChatSubagentState } from '../../../../shared/native-chat-types'
|
||||
import { boundSubagentEntryId } from '../../../native-chat/subagent-entry-id-bounds'
|
||||
import type { RpcContext } from '../core'
|
||||
import { sanitizeNativeChatRpcImageBlock } from './native-chat-rpc-image-block'
|
||||
|
||||
@@ -63,7 +64,9 @@ export function sanitizeNativeChatRpcBlock(
|
||||
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),
|
||||
// The id is the roster KEY: a prefix clip would merge two children, so
|
||||
// it takes the shared digest bound the other wires use.
|
||||
id: boundSubagentEntryId(agent.id),
|
||||
label: clip(agent.label, MAX_SUBAGENT_FIELD_CHARS),
|
||||
state: clipSubagentState(agent.state)
|
||||
}))
|
||||
|
||||
@@ -1,5 +1,8 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import type { NativeChatMessage } from '../../../../shared/native-chat-types'
|
||||
import type {
|
||||
NativeChatMessage,
|
||||
NativeChatSubagentEntry
|
||||
} from '../../../../shared/native-chat-types'
|
||||
import type { RpcContext } from '../core'
|
||||
|
||||
// Stub the bounded tail reader so the handler returns a deterministic transcript with
|
||||
@@ -88,6 +91,7 @@ vi.mock('../../../native-chat/transcript-watch', () => ({
|
||||
}
|
||||
}))
|
||||
|
||||
import { boundSubagentEntryId } from '../../../native-chat/subagent-entry-id-bounds'
|
||||
import { NATIVE_CHAT_METHODS } from './native-chat'
|
||||
|
||||
function makeMessage(text: string): NativeChatMessage {
|
||||
@@ -191,6 +195,35 @@ describe('nativeChat.readSession clientKind truncation gating', () => {
|
||||
expect(block.text).toBe(text)
|
||||
})
|
||||
|
||||
it('bounds a subagent roster before it reaches mobile', async () => {
|
||||
const agents: NativeChatSubagentEntry[] = Array.from({ length: 100 }, (_, index) => ({
|
||||
id: `task-${index}-${'i'.repeat(600)}`,
|
||||
label: 'l'.repeat(600),
|
||||
state: 'working'
|
||||
}))
|
||||
cachedResult.value = {
|
||||
messages: [
|
||||
{
|
||||
...makeMessage(''),
|
||||
blocks: [{ type: 'subagent-group', groupId: 'g', agents }]
|
||||
}
|
||||
]
|
||||
}
|
||||
const result = await readSessionHandler()(
|
||||
{ agent: 'claude', sessionId: 's' },
|
||||
ctxWith('mobile')
|
||||
)
|
||||
const block = (result as { messages: NativeChatMessage[] }).messages[0].blocks[0] as {
|
||||
agents: { id: string; label: string }[]
|
||||
}
|
||||
expect(block.agents).toHaveLength(64)
|
||||
expect(block.agents[0].label).toBe(`${'l'.repeat(512)}\n… (truncated)`)
|
||||
// The id is as untrusted as the label on an imported roster, but it is the
|
||||
// roster key: it is bounded with a digest, never clipped to a bare prefix.
|
||||
expect(block.agents[0].id).toHaveLength(512)
|
||||
expect(block.agents[0].id).toBe(boundSubagentEntryId(`task-0-${'i'.repeat(600)}`))
|
||||
})
|
||||
|
||||
it('clips a pathological text block at the safety ceiling for mobile clients', async () => {
|
||||
const text = 'y'.repeat(70_000)
|
||||
cachedResult.value = { messages: [makeTextMessage(text)] }
|
||||
|
||||
@@ -33,7 +33,15 @@ const CANONICAL_BODIES: AgentJournalItemBody[] = [
|
||||
},
|
||||
{ type: 'tool-call', name: 'Read', input: { path: 'a' } },
|
||||
{ type: 'tool-result', output: 'ok', isError: false },
|
||||
{ type: 'image-ref', path: '/tmp/a.png', alt: 'screenshot' }
|
||||
{ type: 'image-ref', path: '/tmp/a.png', alt: 'screenshot' },
|
||||
{
|
||||
type: 'subagent-group',
|
||||
groupId: 'claude-session:turn-1',
|
||||
agents: [
|
||||
{ id: 'task-1', label: 'Explore', state: 'working', startedAt: 1_000 },
|
||||
{ id: 'task-2', label: 'Review', state: 'completed', tokens: 42, settledAt: 2_000 }
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
{ kind: 'tool-call', name: 'Read', input: undefined, state: 'running' },
|
||||
@@ -143,6 +151,41 @@ describe('nested corruption is rejected', () => {
|
||||
).toBe(false)
|
||||
})
|
||||
|
||||
it('rejects a subagent roster whose entries are malformed', () => {
|
||||
// A KNOWN block type stays a known block: it must not fall through to the
|
||||
// forward-tolerant arm just because its payload is wrong.
|
||||
expect(
|
||||
isAdmissibleAgentJournalItemBody({
|
||||
kind: 'message',
|
||||
role: 'system',
|
||||
blocks: [{ type: 'subagent-group', groupId: 'g', agents: [{ id: 'a', label: 'x' }] }]
|
||||
})
|
||||
).toBe(false)
|
||||
expect(
|
||||
isAdmissibleAgentJournalItemBody({
|
||||
kind: 'message',
|
||||
role: 'system',
|
||||
blocks: [{ type: 'subagent-group', groupId: 'g', agents: 'not-a-roster' }]
|
||||
})
|
||||
).toBe(false)
|
||||
})
|
||||
|
||||
it('keeps a state string a newer build might write admissible', () => {
|
||||
expect(
|
||||
isAdmissibleAgentJournalItemBody({
|
||||
kind: 'message',
|
||||
role: 'system',
|
||||
blocks: [
|
||||
{
|
||||
type: 'subagent-group',
|
||||
groupId: 'g',
|
||||
agents: [{ id: 'a', label: 'x', state: 'some-future-state' }]
|
||||
}
|
||||
]
|
||||
})
|
||||
).toBe(true)
|
||||
})
|
||||
|
||||
it('rejects shallow render items and submissions', () => {
|
||||
expect(
|
||||
isAdmissibleAgentJournalRenderItem({
|
||||
|
||||
Reference in New Issue
Block a user