From 81b28439327e0485d3d297a0f4a5c75b34321776 Mon Sep 17 00:00:00 2001 From: Rick Gao Date: Thu, 23 Jul 2026 15:45:05 -0700 Subject: [PATCH] fix(trace): harden Agent Team correlation --- src/daemon.ts | 3 +- src/hookHandler.ts | 158 ++++- src/parser.ts | 15 +- src/teamCoordinator.ts | 544 ++++++++++++++---- src/teamTranscripts.ts | 387 ++++++++++++- src/transcriptFile.ts | 65 ++- tests/agent-team-test-helpers.ts | 175 ++++++ tests/agent-teams-core.test.ts | 93 +-- tests/agent-teams-correlation.test.ts | 314 ++++++++++ tests/agent-teams-lifecycle.test.ts | 478 +++++++++++++++ tests/agent-teams-ordering.test.ts | 462 +++++++++++++++ tests/agent-teams-session-end.test.ts | 112 ++++ tests/agent-teams-transcript-progress.test.ts | 399 +++++++++++++ tests/parser.test.ts | 18 + 14 files changed, 2976 insertions(+), 247 deletions(-) create mode 100644 tests/agent-team-test-helpers.ts create mode 100644 tests/agent-teams-correlation.test.ts create mode 100644 tests/agent-teams-lifecycle.test.ts create mode 100644 tests/agent-teams-ordering.test.ts create mode 100644 tests/agent-teams-session-end.test.ts create mode 100644 tests/agent-teams-transcript-progress.test.ts diff --git a/src/daemon.ts b/src/daemon.ts index 81f408e..b5433bb 100644 --- a/src/daemon.ts +++ b/src/daemon.ts @@ -295,8 +295,9 @@ export class Daemon { return; } - socket.end(); void this.routeEvent(payload); + // Capture receipt-time transcript evidence before acknowledging the hook. + socket.end(); }); socket.on('error', (err: Error) => { diff --git a/src/hookHandler.ts b/src/hookHandler.ts index 8c9760e..367ab39 100644 --- a/src/hookHandler.ts +++ b/src/hookHandler.ts @@ -50,6 +50,8 @@ import type { SpanParent } from './genaiSpans.js'; import { parseSessionFd } from './parser.js'; import { Session } from './session.js'; import { TeamCoordinator } from './teamCoordinator.js'; +import type { TeamCompletion } from './teamCoordinator.js'; +import type { TeamTranscriptSnapshot } from './teamTranscripts.js'; import { TranscriptFile, readSubagentPrompt, @@ -67,6 +69,8 @@ type RecoverCallHookInput = HookInputFor< 'PermissionDenied' | 'PostToolUse' | 'PostToolUseFailure' >; +const MAX_RECENT_TRANSCRIPTS = 512; + function mergeSubagentOutput(transcriptText?: string, lastMessage?: string): string | undefined { const transcript = transcriptText?.trim(); const latest = lastMessage?.trim(); @@ -122,6 +126,9 @@ export class HookHandler { private readonly sessions = new Map(); private eventQueue = Promise.resolve(); private eventSequence = 0; + /** Hooks can wait in the queue before their sessions are reconstructed. Keep + * recently observed roots available for receipt-time Team snapshots. */ + private readonly recentTranscripts = new Set(); /** InstructionsLoaded can arrive before SessionStart. */ private readonly pendingInstructions = new Map>(); private readonly teams = new TeamCoordinator(); @@ -139,25 +146,60 @@ export class HookHandler { } const sequence = ++this.eventSequence; - const next = this.eventQueue.then(() => this.route(input, sequence)); + this.rememberTranscript(input); + const transcriptSnapshots = input.hook_event_name === 'TeammateIdle' + ? this.teams.snapshotTranscripts( + input.session_id, + input.transcript_path, + this.sessions.values(), + this.recentTranscripts.values(), + ) + : undefined; + const next = this.eventQueue.then(() => + this.route(input, sequence, transcriptSnapshots)); this.eventQueue = next; await next; } - private async route(input: HookInput, sequence: number): Promise { + private rememberTranscript(input: HookInput): void { + const transcriptPath = input.transcript_path; + if (typeof transcriptPath !== 'string') return; + try { + const resolvedPath = new TranscriptFile(transcriptPath).resolvedPath; + this.recentTranscripts.delete(resolvedPath); + this.recentTranscripts.add(resolvedPath); + if (this.recentTranscripts.size > MAX_RECENT_TRANSCRIPTS) { + const oldest = this.recentTranscripts.values().next().value; + if (oldest !== undefined) this.recentTranscripts.delete(oldest); + } + } catch { + // Event processing reports invalid transcript paths. + } + } + + private async route( + input: HookInput, + sequence: number, + transcriptSnapshots?: TeamTranscriptSnapshot[], + ): Promise { const sessionId = input.session_id; this.log( 'INFO', `${input.hook_event_name} session=${sessionId}${input.agent_id ? ` agent=${input.agent_id}` : ''}`, ); try { - await weave.runIsolated(() => this.dispatchEvent(input, sequence)); + await weave.runIsolated(() => + this.dispatchEvent(input, sequence, transcriptSnapshots)); } catch (err) { this.log('ERROR', `Error handling ${input.hook_event_name}: ${err}`); } } - private async dispatchEvent(input: HookInput, sequence: number): Promise { + private async dispatchEvent( + input: HookInput, + sequence: number, + transcriptSnapshots?: TeamTranscriptSnapshot[], + ): Promise { const sessionId = input.session_id; switch (input.hook_event_name) { case 'SessionStart': @@ -183,13 +225,18 @@ export class HookHandler { await this.handlePostToolResult(sessionId, input, sequence); return; case 'SubagentStart': - await this.handleSubagentStart(sessionId, input); + await this.handleSubagentStart(sessionId, input, sequence); return; case 'SubagentStop': - await this.handleSubagentStop(sessionId, input); + await this.handleSubagentStop(sessionId, input, sequence); return; case 'TeammateIdle': - await this.handleTeammateIdle(input, sequence); + await this.handleTeammateIdle( + sessionId, + input, + sequence, + transcriptSnapshots, + ); return; case 'PreCompact': this.handlePreCompact(sessionId, input); @@ -383,7 +430,13 @@ export class HookHandler { } const call = beginCall(session.calls, parent, descriptor); if (call?.kind === 'agent') { - this.teams.registerDispatch(session, call, sequence); + const team = this.teams.registerDispatch(session, call, sequence); + if (team) { + this.log( + 'INFO', + `Team member registered: ${team.teamName ?? 'implicit'}::${team.memberName} (queue depth ${team.depth})`, + ); + } } if (call && !input.agent_id) call.root.phase = 'active'; } @@ -461,12 +514,13 @@ export class HookHandler { const call = existingCall ?? await this.recoverCall(session, input, descriptor!); if (!existingCall && call?.kind === 'agent') { - this.teams.registerDispatch(session, call, sequence); + this.teams.registerDispatch(session, call, sequence, true); } if (call?.kind === 'agent') { - const update = this.teams.postOutcome(call, result); + const update = await this.teams.postOutcome(call, result); + this.settleTeamCompletions(update.completions); if (update.handled) { - this.settleTeamCompletions(update.completions, session); + if (!update.completions.length) this.settleSession(session); return; } } @@ -548,7 +602,7 @@ export class HookHandler { const call = existingCall ?? await this.recoverCall(session, input, descriptor!); const completions = call?.kind === 'agent' - ? this.teams.deny(call, input.reason) + ? await this.teams.deny(call, input.reason) : undefined; if (!completions) denyCall(session.calls, input.tool_use_id, input.reason); this.settleTeamCompletions(completions ?? [], session); @@ -557,6 +611,7 @@ export class HookHandler { private async handleSubagentStart( sessionId: string, input: SubagentStartHookInput, + sequence: number, ): Promise { const session = await this.getOrReconstructSession(sessionId, input); if (!session @@ -574,10 +629,24 @@ export class HookHandler { return; } + const teamLifecycle = this.teams.classifyLifecycle( + session, + input.agent_type, + transcriptPath, + ); + if ( + match.kind === 'missing' + && (teamLifecycle === 'dispatch' || teamLifecycle === 'ambiguous') + ) { + return; + } + + let lifecycle: TracedAgent; if (match.kind === 'found') { bindAgent(session.calls, match, input.agent_id, input.agent_type); + lifecycle = match.call; } else { - this.recoverAgent( + lifecycle = this.recoverAgent( session, input.agent_id, input.agent_type, @@ -586,6 +655,15 @@ export class HookHandler { 'SubagentStart', ); } + if (lifecycle.toolUseId === undefined && teamLifecycle === 'idle') { + this.teams.registerIdle( + session, + lifecycle, + input.agent_type, + transcriptPath, + sequence, + ); + } this.log('INFO', `Subagent started: agentId=${input.agent_id} type=${input.agent_type}`); } @@ -631,6 +709,7 @@ export class HookHandler { private async handleSubagentStop( sessionId: string, input: SubagentStopHookInput, + sequence: number, ): Promise { const session = await this.getOrReconstructSession(sessionId, input); if (!session || session.calls.agentTombstones.has(input.agent_id)) return; @@ -655,6 +734,18 @@ export class HookHandler { ); } + const teamLifecycle = this.teams.classifyLifecycle( + session, + input.agent_type, + transcriptPath, + ); + if ( + match.kind === 'missing' + && (teamLifecycle === 'dispatch' || teamLifecycle === 'ambiguous') + ) { + return; + } + const turn = session.turnForPrompt(input.prompt_id); const recovered = match.kind === 'missing' ? this.recoverAgent( @@ -667,12 +758,29 @@ export class HookHandler { ) : undefined; const lifecycle = match.kind === 'found' ? match.call : recovered; - if (lifecycle && this.teams.has(lifecycle)) { - this.log( - 'DEBUG', - `Subagent stopped: agentId=${input.agent_id} awaiting TeammateIdle`, + if ( + lifecycle + && lifecycle.toolUseId === undefined + && teamLifecycle === 'idle' + ) { + this.teams.registerIdle( + session, + lifecycle, + input.agent_type, + transcriptPath, + sequence, ); - return; + } + if (lifecycle) { + const update = await this.teams.stop(lifecycle, transcriptPath); + this.settleTeamCompletions(update.completions); + if (update.handled) { + this.log( + 'DEBUG', + `Subagent stopped: agentId=${input.agent_id} awaiting TeammateIdle`, + ); + return; + } } const parent = match.kind === 'found' ? match.call.span @@ -723,14 +831,18 @@ export class HookHandler { } private async handleTeammateIdle( + sessionId: string, input: TeammateIdleHookInput, sequence: number, + transcriptSnapshots?: TeamTranscriptSnapshot[], ): Promise { const result = await this.teams.recordIdle({ sequence, + sessionId, teamName: input.team_name, memberName: input.teammate_name, - transcriptPath: input.transcript_path, + idleTranscriptPath: input.transcript_path, + transcriptSnapshots, }); if (!result.completions.length) { this.log( @@ -798,9 +910,15 @@ export class HookHandler { } private settleTeamCompletions( - completions: Array<{ owner: Session }>, + completions: TeamCompletion[], fallback?: Session, ): void { + for (const completion of completions) { + this.log( + 'DEBUG', + `Team completed: ${completion.teamName ?? 'unknown'}::${completion.memberName} (${completion.mode})`, + ); + } const owners = new Set(completions.map(completion => completion.owner)); if (!owners.size && fallback) owners.add(fallback); for (const owner of owners) this.settleSession(owner); diff --git a/src/parser.ts b/src/parser.ts index cee8fb9..78f9b44 100644 --- a/src/parser.ts +++ b/src/parser.ts @@ -38,8 +38,8 @@ export interface ParsedSession { turns: ParsedTurn[]; } -export function parseSessionFd(fd: number): ParsedSession | null { - return parseSessionReader(() => readUtf8FromFd(fd)); +export function parseSessionFd(fd: number, maxBytes?: number): ParsedSession | null { + return parseSessionReader(() => readUtf8FromFd(fd, maxBytes)); } function parseSessionReader(read: () => string): ParsedSession | null { @@ -55,8 +55,12 @@ function parseSessionReader(read: () => string): ParsedSession | null { } } -function readUtf8FromFd(fd: number): string { - const size = fs.fstatSync(fd).size; +function readUtf8FromFd(fd: number, maxBytes?: number): string { + const fileSize = fs.fstatSync(fd).size; + if (maxBytes !== undefined && fileSize < maxBytes) { + throw new Error('transcript shortened before bounded read'); + } + const size = maxBytes ?? fileSize; if (size === 0) return ''; const buffer = Buffer.allocUnsafe(size); @@ -66,6 +70,9 @@ function readUtf8FromFd(fd: number): string { if (count === 0) break; bytesRead += count; } + if (maxBytes !== undefined && bytesRead !== size) { + throw new Error('transcript shortened during bounded read'); + } return buffer.toString('utf8', 0, bytesRead); } diff --git a/src/teamCoordinator.ts b/src/teamCoordinator.ts index bcdbf5e..be926ad 100644 --- a/src/teamCoordinator.ts +++ b/src/teamCoordinator.ts @@ -2,186 +2,492 @@ // SPDX-License-Identifier: MIT // SPDX-PackageName: weave-claude-code +import * as path from 'path'; import type * as weave from 'weave'; -import { emitChatSpans } from './chatSpans.js'; -import { - deferAgentOutcome, - denyCall, - finishAgentCall, -} from './callLifecycle.js'; +import { deferAgentOutcome, denyCall, finishAgentCall } from './callLifecycle.js'; import type { ToolResult, TracedAgent } from './callLifecycle.js'; +import { emitChatSpans } from './chatSpans.js'; import { ATTR, assistantOutputMessages, parseTimestamp } from './genaiSpans.js'; import type { ParsedTurn } from './parser.js'; -import { VERSION } from './setup.js'; -import { readTeammateTurns } from './teamTranscripts.js'; import type { Session } from './session.js'; +import { VERSION } from './setup.js'; +import * as teamTranscripts from './teamTranscripts.js'; +import type { + TeamTranscriptEvidenceContext, + TeamTranscriptProgress, + TeamTranscriptSnapshot, +} from './teamTranscripts.js'; -type PendingTeam = { - session: Session; - call: TracedAgent; - teamName: string; - memberName: string; - sequence: number; +type Idle = { + sequence: number; sessionId: string; teamName: string; memberName: string; + transcriptPath?: string; transcriptSnapshots: TeamTranscriptSnapshot[]; + receiptFingerprint?: string; selectedPath?: string; fingerprint?: string; + agentType?: string; }; - +type Reservation = { idle: Idle; transcriptPath: string }; +type PendingBase = { + session: Session; call: TracedAgent; + memberName: string; teamName?: string; sequence: number; + /** A Post hook recovered this dispatch without observing its earlier Pre. */ + recoveredWithoutPre: boolean; + lifecyclePath?: string; + reservation?: Reservation; +}; +type Pending = + | PendingBase & { kind: 'dispatch' } + | PendingBase & { kind: 'lifecycle' }; +type IdleHistory = { idle: Idle; completed: boolean }; +type Match = [Pending, string] | 'missing' | 'ambiguous'; export type TeamCompletion = { + mode: 'cross-session' | 'same-session'; owner: Session; - teamName: string; + teamName?: string; memberName: string; }; - -export type TeamUpdate = { - handled: boolean; - completions: TeamCompletion[]; -}; - -export type TeamIdleUpdate = { - status: 'missing' | 'ambiguous' | 'completed'; +type TeamPostUpdate = { handled: boolean; completions: TeamCompletion[] }; +type TeamStopUpdate = { handled: boolean; completions: TeamCompletion[] }; +type TeamIdleUpdate = { + status: 'duplicate' | 'missing' | 'ambiguous' | 'buffered' | 'retry' | 'completed'; completions: TeamCompletion[]; }; -const text = (value: unknown) => - typeof value === 'string' && value.trim() ? value.trim() : undefined; +const SEP = '\0'; +const MAX_CALLS = 256; +const MAX_IDLE_HISTORY = 512; +const MAX_PROGRESS_CURSORS = 1024; +const isDispatch = ( + pending: Pending, +): pending is Extract => pending.kind === 'dispatch'; +const text = (value: unknown) => typeof value === 'string' && value.trim() ? value.trim() : undefined; +const idleKey = (idle: Pick) => + JSON.stringify([idle.teamName, idle.memberName, idle.sessionId]); +/** Resolve only a transcript tied to this Agent type and idle session. */ +function transcriptFor( + pending: Pending, + idle: Idle, + context: TeamTranscriptEvidenceContext, +): { path: string; strong: boolean } | 'ambiguous' | undefined { + const exact = (candidate?: string) => teamTranscripts.matchTranscript( + candidate, + idle.sessionId, + pending.call.declaredAgentType, + context, + ); + const hook = exact(idle.transcriptPath); + if (hook === 'ambiguous') return 'ambiguous'; + const sameSession = pending.session.sessionId === idle.sessionId; + const metadata = teamTranscripts.findMetadata( + pending.session.transcript.resolvedPath, + pending.call.declaredAgentType, + idle.sessionId, + context, + ); + if (metadata === 'ambiguous') { + return hook && !sameSession ? { path: hook, strong: false } : 'ambiguous'; + } + // Cross-session hooks may carry the freshest teammate output. For a + // same-session lifecycle, the hook path is the coordinator transcript, so + // its lifecycle/metadata transcript remains authoritative. + if (metadata.length === 1) { + return { path: sameSession ? metadata[0] : hook ?? metadata[0], strong: true }; + } + const lifecycle = exact(pending.lifecyclePath); + if (lifecycle === 'ambiguous') return 'ambiguous'; + if (lifecycle) { + return { path: sameSession ? lifecycle : hook ?? lifecycle, strong: true }; + } + return hook ? { path: hook, strong: false } : undefined; +} function emitTeammate( - conversation: weave.Conversation, - memberName: string, - turns: ParsedTurn[], + conversation: weave.Conversation, memberName: string, turns: ParsedTurn[], ): { model?: string; text?: string } { const responses = turns.flatMap(turn => turn.responses); const model = turns.filter(turn => turn.model).at(-1)?.model; const span = conversation.startTurn({ - agentName: memberName, - agentVersion: VERSION, - model, + agentName: memberName, agentVersion: VERSION, model, userMessage: turns[0]?.userText, startTime: parseTimestamp(turns[0]?.startTime ?? responses[0]?.startTime), }); try { emitChatSpans(span, responses, { agentName: memberName }); const output = turns.flatMap(turn => turn.text); - if (output.length) { - span.setAttributes({ [ATTR.OUTPUT_MESSAGES]: assistantOutputMessages(output) }); - } + if (output.length) span.setAttributes({ [ATTR.OUTPUT_MESSAGES]: assistantOutputMessages(output) }); if (model) span.record({ model }); return { model, text: turns.at(-1)?.text.join('\n') || undefined }; } finally { span.end({ endTime: parseTimestamp(responses.at(-1)?.endTime) ?? new Date() }); } } - -/** Correlates explicit Agent Team dispatches with their cross-session idle - * event. Recovery and weak-evidence matching are deliberately separate. */ +/** The bounded join state missing from Claude's cross-session team hooks. */ export class TeamCoordinator { - private readonly pending = new Set(); + private calls: Pending[] = []; + private idleHistory: IdleHistory[] = []; + /** Retain logical teammate cursors for this daemon's lifetime. Once full, + * existing cursors continue advancing while new identities fail closed. */ + private progress = new Map(); + private progressOwners = new Map(); + + /** Capture every output path that current correlation evidence could select. */ + snapshotTranscripts( + sessionId: unknown, + transcriptPath: unknown, + sessions: Iterable, + recentTranscripts: Iterable, + ): TeamTranscriptSnapshot[] { + const pending = this.calls; + function* transcriptRoots() { + yield* recentTranscripts; + for (const session of sessions) yield session.transcript.resolvedPath; + for (const call of pending) yield call.session.transcript.resolvedPath; + } + return teamTranscripts.snapshotTranscripts({ + sessionId, + transcriptPath, + lifecyclePaths: this.calls.map(pending => pending.lifecyclePath), + transcriptRoots: transcriptRoots(), + }); + } + static isDispatchInput(input: Record): boolean { + return text(input['name']) !== undefined || text(input['team_name']) !== undefined; + } registerDispatch( session: Session, call: TracedAgent, sequence: number, - ): void { - if (!call.toolUseId || this.has(call)) return; - const teamName = text(call.input['team_name']); - if (!teamName) return; - this.pending.add({ - session, - call, - teamName, - memberName: text(call.input['name']) ?? call.agentType, - sequence, - }); + recoveredWithoutPre = false, + ) { + if (!call.toolUseId || !TeamCoordinator.isDispatchInput(call.input)) return undefined; + let pending = this.find(call); + if (!pending) { + if (this.calls.length >= MAX_CALLS) return undefined; + pending = { + kind: 'dispatch', session, call, sequence, recoveredWithoutPre, + memberName: text(call.input['name']) ?? call.agentType, + teamName: text(call.input['team_name']), + }; + this.calls.push(pending); + } + const depth = this.calls.filter(call => isDispatch(call) + && call.memberName === pending.memberName && call.teamName === pending.teamName).length; + return { teamName: pending.teamName, memberName: pending.memberName, depth }; } + private find(call: TracedAgent) { return this.calls.find(pending => pending.call === call); } + has(call: TracedAgent): boolean { return this.find(call) !== undefined; } + async postOutcome(call: TracedAgent, outcome: ToolResult): Promise { + const pending = this.find(call); + if (!pending || !isDispatch(pending)) return { handled: false, completions: [] }; - has(call: TracedAgent): boolean { - return [...this.pending].some(candidate => candidate.call === call); - } + if (!outcome.ok) { + if (pending.teamName === undefined) { + this.remove(pending); + return { handled: false, completions: (await this.reconcile()).completions }; + } + finishAgentCall(pending.session.calls, call, { outcome }); + const completion = this.completed(pending); + return { + handled: true, + completions: [completion, ...(await this.reconcile()).completions], + }; + } - postOutcome(call: TracedAgent, outcome: ToolResult): TeamUpdate { + deferAgentOutcome(call, outcome); + const confirmed = pending.teamName !== undefined; + const completions = (await this.reconcile()).completions; + return { + // A candidate completed from verified transcript progress is already + // tombstoned; otherwise ordinary Agent Stop/Post remains authoritative. + handled: confirmed || !this.find(call), + completions, + }; + } + classifyLifecycle( + session: Session, + memberName: string, + transcriptPath: string, + ): 'dispatch' | 'idle' | 'ambiguous' | undefined { + if (!teamTranscripts.isAgentSetting(transcriptPath)) return undefined; + // An enclosing teammate session can start ordinary child Agents. Require + // team evidence in this lifecycle's own transcript before suppressing or + // deferring anything. + const team = teamTranscripts.teamName(transcriptPath); + if (!team) return undefined; + const correlationContext = teamTranscripts.newEvidenceContext(); + const candidates = this.calls.filter(pending => isDispatch(pending) + && (!pending.teamName || pending.teamName === team)); + const evidence = candidates.map(pending => ({ + pending, + paths: teamTranscripts.findMetadata( + pending.session.transcript.resolvedPath, + pending.call.declaredAgentType, + session.sessionId, + correlationContext, + ), + })); + if (evidence.some(candidate => candidate.paths === 'ambiguous')) return 'ambiguous'; + const inferred = evidence.flatMap(candidate => + candidate.paths !== 'ambiguous' && candidate.paths.length === 1 + ? [candidate.pending] + : []); + if (inferred.length) { + return new Set(inferred.map(pending => pending.call.root)).size === 1 + && new Set(inferred.map(pending => pending.call.parent)).size === 1 + ? 'dispatch' : 'ambiguous'; + } + if (this.idleHistory.some(entry => entry.completed + && entry.idle.teamName === team + && entry.idle.sessionId === session.sessionId + && entry.idle.agentType === memberName)) return 'dispatch'; + return 'idle'; + } + registerIdle( + session: Session, + call: TracedAgent, + memberName: string, + transcriptPath: string, + sequence: number, + ): void { + if (this.has(call) || this.calls.length >= MAX_CALLS) return; + deferAgentOutcome(call, { ok: true, output: null }); + this.calls.push({ + kind: 'lifecycle', session, call, memberName, sequence, + recoveredWithoutPre: false, + teamName: teamTranscripts.teamName(transcriptPath), + lifecyclePath: transcriptPath, + }); + } + async stop(call: TracedAgent, transcriptPath: string): Promise { const pending = this.find(call); if (!pending) return { handled: false, completions: [] }; - if (outcome.ok) { - deferAgentOutcome(call, outcome); - return { handled: true, completions: [] }; + const teamName = teamTranscripts.teamName(transcriptPath); + if (pending.teamName === undefined && !teamName) { + this.remove(pending); + return { handled: false, completions: (await this.reconcile()).completions }; } - - finishAgentCall(pending.session.calls, call, { outcome }); - this.pending.delete(pending); - return { handled: true, completions: [this.completed(pending)] }; + pending.lifecyclePath = transcriptPath; + pending.teamName ??= teamName; + return { handled: true, completions: (await this.reconcile()).completions }; } - async recordIdle(input: { - sequence: number; - teamName: string; - memberName: string; - transcriptPath?: string; + sequence: number; sessionId: string; teamName: string; + memberName: string; idleTranscriptPath?: string; + transcriptSnapshots?: TeamTranscriptSnapshot[]; }): Promise { - const candidates = [...this.pending].filter(candidate => - candidate.sequence < input.sequence - && candidate.teamName === input.teamName - && candidate.memberName === input.memberName - && candidate.call.outcome?.ok); - if (candidates.length !== 1) { - return { - status: candidates.length ? 'ambiguous' : 'missing', - completions: [], - }; - } + const transcriptSnapshots = input.transcriptSnapshots ?? []; + const idle: Idle = { + sequence: input.sequence, sessionId: input.sessionId, + teamName: input.teamName, memberName: input.memberName, + transcriptPath: input.idleTranscriptPath, + transcriptSnapshots, + receiptFingerprint: teamTranscripts.receiptFingerprint(transcriptSnapshots), + }; + const key = idleKey(idle); + const buffered = idle.receiptFingerprint + ? this.idleHistory.find(entry => + !entry.completed && idleKey(entry.idle) === key + && entry.idle.receiptFingerprint === idle.receiptFingerprint) + : undefined; + const occurrence = buffered?.idle ?? idle; + if (!buffered) this.rememberIdle({ idle: occurrence, completed: false }); - const [pending] = candidates; - const turns = input.transcriptPath - ? readTeammateTurns(input.transcriptPath) - : []; - if (!turns.length) return { status: 'missing', completions: [] }; + const reconciled = await this.reconcile(); + if (reconciled.completions.length) { + return { status: 'completed', completions: reconciled.completions }; + } + if (reconciled.duplicates.has(occurrence)) return { status: 'duplicate', completions: [] }; + if (reconciled.retried.has(occurrence)) return { status: 'retry', completions: [] }; + if (this.calls.some(pending => pending.reservation?.idle === occurrence)) { + return { status: 'buffered', completions: [] }; + } + const match = this.match(occurrence); + return { + status: typeof match === 'string' ? match : 'buffered', + completions: [], + }; + } - const emitted = emitTeammate( - pending.session.conversation, - pending.memberName, - turns, + /** Match every buffered idle made unambiguous by the current transition. + * A transcript retry rolls back its reservation for a later event. */ + private async reconcile(): Promise<{ + completions: TeamCompletion[]; + retried: Set; + duplicates: Set; + }> { + const completions: TeamCompletion[] = []; + const attempted = new Set(); + const retried = new Set(); + const duplicates = new Set(); + const correlationContext = teamTranscripts.newEvidenceContext(); + for (;;) { + let pending = this.calls.find(candidate => candidate.reservation + && candidate.call.outcome && !attempted.has(candidate.reservation.idle)); + if (!pending) { + for (const entry of this.idleHistory) { + if (entry.completed || attempted.has(entry.idle) + || this.calls.some(candidate => candidate.reservation?.idle === entry.idle)) continue; + const match = this.match(entry.idle, correlationContext); + if (typeof match === 'string') continue; + const [matched, transcriptPath] = match; + const snapshot = teamTranscripts.snapshotFor( + transcriptPath, + entry.idle.transcriptSnapshots, + ); + const fingerprint = teamTranscripts.snapshotFingerprint(snapshot); + const selectedPath = snapshot?.path ?? path.resolve(transcriptPath); + if (fingerprint && this.idleHistory.some(previous => previous.completed + && idleKey(previous.idle) === idleKey(entry.idle) + && previous.idle.selectedPath === selectedPath + && previous.idle.fingerprint === fingerprint)) { + attempted.add(entry.idle); + duplicates.add(entry.idle); + this.idleHistory = this.idleHistory.filter(previous => previous !== entry); + continue; + } + entry.idle.selectedPath = selectedPath; + entry.idle.fingerprint = fingerprint; + pending = matched; + pending.reservation = { idle: entry.idle, transcriptPath }; + break; + } + } + const reservation = pending?.reservation; + if (!pending || !reservation) break; + const idle = reservation.idle; + attempted.add(idle); + const outcome = pending.call.outcome; + if (!outcome) continue; + const completion = await this.finish(pending, reservation, outcome); + if (completion) completions.push(completion); + else retried.add(idle); + } + return { completions, retried, duplicates }; + } + private match( + idle: Idle, + correlationContext = teamTranscripts.newEvidenceContext(), + ): Match { + let candidates = this.calls.filter(pending => !pending.reservation + && pending.memberName === idle.memberName + && (!pending.teamName || pending.teamName === idle.teamName) + && (isDispatch(pending) + ? pending.recoveredWithoutPre || pending.sequence < idle.sequence + : pending.session.sessionId === idle.sessionId && pending.sequence < idle.sequence)); + const exactTeam = candidates.filter(pending => pending.teamName === idle.teamName); + if (exactTeam.length) candidates = exactTeam; + const resolved: Array<[Pending, { path: string; strong: boolean }]> = []; + for (const pending of candidates) { + const transcript = transcriptFor(pending, idle, correlationContext); + if (transcript === 'ambiguous') return 'ambiguous'; + if (transcript) resolved.push([pending, transcript]); + } + if (!resolved.length) return 'missing'; + const strong = resolved.filter(([, transcript]) => transcript.strong); + const pool = strong.length ? strong : resolved; + if (pool.some(([pending]) => !isDispatch(pending))) { + if (pool.length !== 1) return 'ambiguous'; + } else if (new Set(pool.map(([pending]) => pending.call.root)).size !== 1 + || new Set(pool.map(([pending]) => pending.call.parent)).size !== 1) { + return 'ambiguous'; + } + pool.sort(([a], [b]) => { + if (a.recoveredWithoutPre !== b.recoveredWithoutPre) { + return a.recoveredWithoutPre ? -1 : 1; + } + return a.sequence - b.sequence; + }); + return [pool[0][0], pool[0][1].path]; + } + private async finish( + pending: Pending, + reservation: Reservation, + outcome: ToolResult, + ): Promise { + const { idle, transcriptPath } = reservation; + const snapshot = teamTranscripts.snapshotFor( + transcriptPath, + idle.transcriptSnapshots, + ); + if (!snapshot) { + pending.reservation = undefined; + return undefined; + } + const cursorKey = idleKey(idle); + const identityKey = [ + snapshot.device, + snapshot.inode, + snapshot.created, + ].join(SEP); + const identityOwner = this.progressOwners.get(identityKey); + if (identityOwner && identityOwner !== cursorKey) { + pending.reservation = undefined; + return undefined; + } + const prior = this.progress.get(cursorKey); + if (!prior && this.progress.size >= MAX_PROGRESS_CURSORS) { + pending.reservation = undefined; + return undefined; + } + const parsed = await teamTranscripts.readNewTurns( + transcriptPath, + prior, + snapshot, ); - if (emitted.model) { - pending.call.span.setAttributes({ [ATTR.RESPONSE_MODEL]: emitted.model }); + if (!parsed) { + pending.reservation = undefined; + return undefined; } - const original = pending.call.outcome; + const emitted = emitTeammate(pending.session.conversation, pending.memberName, parsed[0]); + this.progress.set(cursorKey, parsed[1]); + this.progressOwners.set(identityKey, cursorKey); + if (emitted.model) pending.call.span.setAttributes({ [ATTR.RESPONSE_MODEL]: emitted.model }); + const output = outcome.ok ? outcome.output : null; finishAgentCall(pending.session.calls, pending.call, { - outcome: { - ok: true, - output: emitted.text ?? (original?.ok ? original.output : null), - }, + outcome: { ok: true, output: emitted.text ?? output }, }); - this.pending.delete(pending); - return { status: 'completed', completions: [this.completed(pending)] }; + this.idleHistory = this.idleHistory.filter(entry => entry.idle !== idle); + if (idle.fingerprint) { + idle.agentType = teamTranscripts.agentSetting(transcriptPath) + ?? pending.call.declaredAgentType; + this.rememberIdle({ idle, completed: true }); + } + return this.completed(pending); } - - deny(call: TracedAgent, reason: string): TeamCompletion[] | undefined { + async deny(call: TracedAgent, reason: string): Promise { const pending = this.find(call); - if (!pending || !call.toolUseId) return undefined; + if (!pending || !isDispatch(pending) || !call.toolUseId) return undefined; denyCall(pending.session.calls, call.toolUseId, reason); - this.pending.delete(pending); - return [this.completed(pending)]; - } - - orphanSession(sessionId: string, reason: string, endTime: Date): void { - for (const pending of [...this.pending]) { - if (pending.session.sessionId !== sessionId) continue; - finishAgentCall( - pending.session.calls, - pending.call, - { orphanReason: reason }, - endTime, - ); - this.pending.delete(pending); - } + this.remove(pending); + return (await this.reconcile()).completions; } - - private find(call: TracedAgent): PendingTeam | undefined { - return [...this.pending].find(candidate => candidate.call === call); - } - - private completed(pending: PendingTeam): TeamCompletion { + private completed(pending: Pending): TeamCompletion { + this.remove(pending); return { + mode: isDispatch(pending) ? 'cross-session' : 'same-session', owner: pending.session, teamName: pending.teamName, memberName: pending.memberName, }; } + private remove(pending: Pending): void { + this.calls = this.calls.filter(candidate => candidate !== pending); + } + private rememberIdle(entry: IdleHistory): void { + this.idleHistory = this.idleHistory.filter(previous => { + if (!entry.completed) return previous.idle !== entry.idle; + return !previous.completed + || idleKey(previous.idle) !== idleKey(entry.idle) + || previous.idle.selectedPath !== entry.idle.selectedPath + || previous.idle.fingerprint !== entry.idle.fingerprint; + }); + this.idleHistory.push(entry); + if (this.idleHistory.length > MAX_IDLE_HISTORY) this.idleHistory.shift(); + } + orphanSession(sessionId: string, reason: string, endTime: Date): void { + for (const pending of [...this.calls]) { + if (pending.session.sessionId !== sessionId) continue; + finishAgentCall(pending.session.calls, pending.call, { orphanReason: reason }, endTime); + this.remove(pending); + } + } } diff --git a/src/teamTranscripts.ts b/src/teamTranscripts.ts index 9949e98..df87c99 100644 --- a/src/teamTranscripts.ts +++ b/src/teamTranscripts.ts @@ -2,21 +2,384 @@ // SPDX-License-Identifier: MIT // SPDX-PackageName: weave-claude-code -import { parseSessionFd } from './parser.js'; -import type { ParsedTurn } from './parser.js'; -import { TranscriptFile } from './transcriptFile.js'; +import * as fs from 'fs'; +import * as path from 'path'; +import { extractAssistantTextBlocks, parseSessionFd, type ParsedTurn } from './parser.js'; +import { + readFirstTranscriptLine, + readTranscriptPrefix, + subagentsDirectory, + TRANSCRIPT_SCAN_LIMIT_BYTES, + TranscriptFile, + type ReadBudget, +} from './transcriptFile.js'; +import { sha256Hex } from './utils.js'; -/** Read the transcript explicitly named by TeammateIdle. Correlation through - * metadata and neighboring transcript discovery belongs to the recovery layer. */ -export function readTeammateTurns(transcriptPath: string): ParsedTurn[] { +const SEP = '\0'; + +// Receipt-time discovery runs before the hook is acknowledged. Exceeding any +// scan budget discards the evidence so correlation fails closed. +const MAX_TRANSCRIPT_ROOTS = 1024; +const MAX_DIRECTORY_ENTRIES = 512; +const MAX_TRANSCRIPT_CANDIDATES = 64; +const MAX_CORRELATION_BYTES = 16 * 1024 * 1024; +const MAX_METADATA_BYTES = 1024 * 1024; +export const MAX_TEAM_TRANSCRIPT_BYTES = TRANSCRIPT_SCAN_LIMIT_BYTES; + +export type TeamTranscriptSnapshot = { + path: string; device: number; inode: number; created: number; size: number; + modified: number; completeLine: boolean; +}; + +export type TeamTranscriptProgress = { + device: number; inode: number; created: number; + bytes: number; responses: number; content: number; + lastResponseId?: string; contentPrefixHash: string; +}; + +export type TeamTranscriptEvidenceContext = { + metadata: Map; + firstLines: Map | null>; + firstLineBudget: ReadBudget; + entries: number; candidates: number; metadataBytes: number; +}; + +export const newEvidenceContext = (): TeamTranscriptEvidenceContext => ({ + metadata: new Map(), + firstLines: new Map(), + firstLineBudget: { remaining: MAX_CORRELATION_BYTES }, + entries: 0, + candidates: 0, + metadataBytes: 0, +}); + +/** Capture the append-only transcript boundary represented by an idle hook. */ +export function snapshot(transcriptPath: unknown): TeamTranscriptSnapshot | undefined { + if (typeof transcriptPath !== 'string') return undefined; let transcript: TranscriptFile | undefined; try { transcript = new TranscriptFile(transcriptPath); - return parseSessionFd(transcript.getFd())?.turns - .filter(turn => turn.responses.length) ?? []; - } catch { - return []; - } finally { - transcript?.close(); + const fd = transcript.getFd(); + const stat = fs.fstatSync(fd); + if (stat.size > MAX_TEAM_TRANSCRIPT_BYTES) return undefined; + let completeLine = stat.size === 0; + if (stat.size) { + const last = Buffer.allocUnsafe(1); + completeLine = fs.readSync(fd, last, 0, 1, stat.size - 1) === 1 && last[0] === 0x0a; + } + return { + path: transcript.resolvedPath, device: stat.dev, inode: stat.ino, + created: stat.birthtimeMs, size: stat.size, modified: stat.mtimeMs, completeLine, + }; + } catch { return undefined; } finally { transcript?.close(); } +} + +export function snapshotFingerprint(snapshot?: TeamTranscriptSnapshot): string | undefined { + return snapshot && [ + snapshot.path, snapshot.device, snapshot.inode, snapshot.created, + snapshot.size, snapshot.modified, + ].join(SEP); +} + +export function receiptFingerprint(snapshots: TeamTranscriptSnapshot[]): string | undefined { + if (!snapshots.length) return undefined; + return sha256Hex(snapshots.map(value => snapshotFingerprint(value) as string) + .sort() + .join(`${SEP}${SEP}`)); +} + +export function snapshotFor( + transcriptPath: string | undefined, + snapshots: TeamTranscriptSnapshot[], +): TeamTranscriptSnapshot | undefined { + if (!transcriptPath) return undefined; + const resolved = path.resolve(transcriptPath); + return snapshots.find(snapshot => snapshot.path === resolved); +} + +type SnapshotDiscoveryInput = { + sessionId: unknown; + transcriptPath: unknown; + lifecyclePaths: Iterable; + transcriptRoots: Iterable; +}; + +/** Capture every output path that current correlation evidence could select. */ +export function snapshotTranscripts( + input: SnapshotDiscoveryInput, +): TeamTranscriptSnapshot[] { + const direct = snapshot(input.transcriptPath); + const directOnly = () => direct ? [direct] : []; + const paths = new Set(); + const examined = new Set(); + const firstLineBudget = { remaining: MAX_CORRELATION_BYTES }; + if (direct) examined.add(direct.path); + + const addIfOwned = (candidate: string) => { + const resolved = path.resolve(candidate); + if (examined.has(resolved)) return true; + if (examined.size >= MAX_TRANSCRIPT_CANDIDATES) return false; + examined.add(resolved); + const first = readFirstTranscriptLine( + resolved, + TRANSCRIPT_SCAN_LIMIT_BYTES, + firstLineBudget, + ); + if (!first && firstLineBudget.remaining === 0) return false; + if (first?.['sessionId'] === input.sessionId) paths.add(resolved); + return true; + }; + + if (typeof input.sessionId === 'string') { + for (const lifecyclePath of input.lifecyclePaths) { + if (lifecyclePath && !addIfOwned(lifecyclePath)) return directOnly(); + } + const owners = new Set(); + let roots = 0; + for (const owner of input.transcriptRoots) { + if (++roots > MAX_TRANSCRIPT_ROOTS) return directOnly(); + owners.add(owner); + } + let entries = 0; + for (const owner of owners) { + let directory: fs.Dir | undefined; + try { + directory = fs.opendirSync(subagentsDirectory(owner)); + for (;;) { + const entry = directory.readSync(); + if (!entry) break; + if (++entries > MAX_DIRECTORY_ENTRIES) return directOnly(); + if (entry.name.endsWith('.jsonl') + && !addIfOwned(path.join(directory.path, entry.name))) return directOnly(); + } + } catch { /* colocated transcripts are optional */ } finally { + try { directory?.closeSync(); } catch { /* already closed */ } + } + } + } + + if (paths.size + (direct ? 1 : 0) > MAX_TRANSCRIPT_CANDIDATES) return directOnly(); + return [...directOnly(), ...[...paths].flatMap(candidate => { + const value = snapshot(candidate); + return value ? [value] : []; + })]; +} + +function firstLineFor( + transcriptPath: string, + context: TeamTranscriptEvidenceContext, +): Record | 'ambiguous' | undefined { + const resolved = path.resolve(transcriptPath); + const cached = context.firstLines.get(resolved); + if (cached !== undefined) return cached ?? undefined; + if (context.firstLineBudget.remaining === 0) return 'ambiguous'; + const first = readFirstTranscriptLine( + resolved, + TRANSCRIPT_SCAN_LIMIT_BYTES, + context.firstLineBudget, + ); + if (!first && context.firstLineBudget.remaining === 0) return 'ambiguous'; + context.firstLines.set(resolved, first ?? null); + return first; +} + +export function matchTranscript( + transcriptPath: string | undefined, + sessionId: string, + declaredAgentType: string | undefined, + context: TeamTranscriptEvidenceContext, +): string | 'ambiguous' | undefined { + if (!transcriptPath) return undefined; + const first = firstLineFor(transcriptPath, context); + if (first === 'ambiguous') return 'ambiguous'; + const setting = first?.['agentSetting']; + return first?.['sessionId'] === sessionId + && (!declaredAgentType + || typeof setting !== 'string' + || setting === declaredAgentType) + ? transcriptPath + : undefined; +} + +export function findMetadata( + rootTranscriptPath: string, + declaredAgentType: string | undefined, + sessionId: string, + context: TeamTranscriptEvidenceContext, +): string[] | 'ambiguous' { + const key = [rootTranscriptPath, declaredAgentType ?? '', sessionId].join(SEP); + const cached = context.metadata.get(key); + if (cached !== undefined) return cached; + const ambiguous = () => { + context.metadata.set(key, 'ambiguous'); + return 'ambiguous' as const; + }; + const metadata: string[] = []; + let directory: fs.Dir | undefined; + try { + directory = fs.opendirSync(subagentsDirectory(rootTranscriptPath)); + for (;;) { + const entry = directory.readSync(); + if (!entry) break; + if (++context.entries > MAX_DIRECTORY_ENTRIES) return ambiguous(); + if (!entry.name.endsWith('.meta.json')) continue; + const metaPath = path.join(directory.path, entry.name); + let fd: number | undefined; + try { + fd = fs.openSync(metaPath, fs.constants.O_RDONLY | fs.constants.O_NOFOLLOW); + const stat = fs.fstatSync(fd); + if (!stat.isFile()) continue; + context.metadataBytes += stat.size; + if (context.metadataBytes > MAX_METADATA_BYTES) return ambiguous(); + const buffer = Buffer.allocUnsafe(stat.size); + let read = 0; + while (read < stat.size) { + const count = fs.readSync(fd, buffer, read, stat.size - read, read); + if (!count) break; + read += count; + } + if (read !== stat.size) continue; + const meta = JSON.parse(buffer.toString('utf8', 0, read)) as Record; + if (declaredAgentType && meta['agentType'] !== declaredAgentType) continue; + if (++context.candidates > MAX_TRANSCRIPT_CANDIDATES) return ambiguous(); + const candidate = metaPath.replace(/\.meta\.json$/, '.jsonl'); + const first = firstLineFor(candidate, context); + if (first === 'ambiguous') return ambiguous(); + if (first?.['sessionId'] === sessionId) metadata.push(candidate); + if (metadata.length > 1) return ambiguous(); + } catch { /* ignore incomplete metadata */ } finally { + try { if (fd !== undefined) fs.closeSync(fd); } catch { /* already closed */ } + } + } + } catch { /* metadata is optional */ } finally { + try { directory?.closeSync(); } catch { /* already closed */ } } + context.metadata.set(key, metadata); + return metadata; +} + +export function isAgentSetting(transcriptPath: string): boolean { + return readFirstTranscriptLine(transcriptPath)?.['type'] === 'agent-setting'; +} + +/** Return the exact lifecycle type recorded by Claude for this transcript. */ +export function agentSetting(transcriptPath: string): string | undefined { + const first = readFirstTranscriptLine(transcriptPath); + const setting = first?.['agentSetting']; + return first?.['type'] === 'agent-setting' && typeof setting === 'string' && setting + ? setting + : undefined; +} + +/** Agent-team transcripts carry their team on a later user record rather than + * necessarily on the initial agent-setting record. */ +export function teamName(transcriptPath: string): string | undefined { + const prefix = readTranscriptPrefix(transcriptPath); + if (!prefix) return undefined; + for (const raw of prefix.split('\n')) { + if (!raw.trim()) continue; + try { + const teamName = (JSON.parse(raw) as Record)['teamName']; + if (typeof teamName === 'string' && teamName) return teamName; + } catch { + // A partial/malformed record does not invalidate the bounded scan. + } + } + return undefined; +} + +/** A partial JSONL record may finish after receipt, but later records belong + * to later hooks. Return only through that partial record's first newline. */ +function snapshotReadLimit( + fd: number, + snapshot: TeamTranscriptSnapshot, +): number | 'invalid' | undefined { + const stat = fs.fstatSync(fd); + if (stat.dev !== snapshot.device || stat.ino !== snapshot.inode + || stat.birthtimeMs !== snapshot.created || stat.size < snapshot.size + || snapshot.size > MAX_TEAM_TRANSCRIPT_BYTES) { + return 'invalid'; + } + if (snapshot.completeLine) return snapshot.size; + const chunk = Buffer.allocUnsafe(64 * 1024); + let offset = snapshot.size; + const scanLimit = Math.min(stat.size, MAX_TEAM_TRANSCRIPT_BYTES); + while (offset < scanLimit) { + const count = fs.readSync(fd, chunk, 0, Math.min(chunk.length, scanLimit - offset), offset); + if (!count) break; + const newline = chunk.indexOf(0x0a, 0); + if (newline >= 0 && newline < count) return offset + newline + 1; + offset += count; + } + return stat.size > MAX_TEAM_TRANSCRIPT_BYTES ? 'invalid' : undefined; +} + +/** Retry partial snapshots and advance only after new provider content. The + * cursor includes the final response's content offset because Claude can append + * another assistant record to the same normalized response and parsed turn. */ +export async function readNewTurns( + transcriptPath: string, progress?: TeamTranscriptProgress, snapshot?: TeamTranscriptSnapshot, +): Promise<[ParsedTurn[], TeamTranscriptProgress] | undefined> { + let transcript: TranscriptFile | undefined; + try { + transcript = new TranscriptFile(transcriptPath); + const fd = transcript.getFd(); + for (let attempt = 0; attempt < 5; attempt++) { + const stat = fs.fstatSync(fd); + const boundary = snapshot?.path === transcript.resolvedPath + ? snapshotReadLimit(fd, snapshot) + : stat.size <= MAX_TEAM_TRANSCRIPT_BYTES ? stat.size : 'invalid'; + if (boundary === 'invalid') return undefined; + const turns = boundary === undefined + ? [] + : parseSessionFd(fd, boundary)?.turns.filter(turn => turn.responses.length) ?? []; + const all = turns.flatMap(turn => turn.responses); + const sameFile = progress?.device === stat.dev && progress.inode === stat.ino + && progress.created === stat.birthtimeMs; + if (progress && (!sameFile || (boundary !== undefined && boundary < progress.bytes))) { + return undefined; + } + const priorLast = sameFile && progress.responses ? all[progress.responses - 1] : undefined; + if (sameFile && (progress.responses > all.length + || !priorLast + || priorLast.id !== progress.lastResponseId + || progress.content > priorLast.content.length + || sha256Hex(JSON.stringify(priorLast.content.slice(0, progress.content))) + !== progress.contentPrefixHash)) return undefined; + const cursor = sameFile ? progress : undefined; + let responseIndex = 0; + const fresh = turns.flatMap(turn => { + const turnStart = responseIndex; + const responses = turn.responses.flatMap(response => { + const index = responseIndex++; + if (cursor && index < cursor.responses - 1) return []; + if (cursor && index === cursor.responses - 1) { + const content = response.content.slice(cursor.content); + return content.length ? [{ ...response, content }] : []; + } + return [response]; + }); + if (!responses.length) return []; + return [{ + ...turn, + userText: !cursor || turnStart >= cursor.responses ? turn.userText : undefined, + responses, + text: responses.flatMap(response => extractAssistantTextBlocks(response.content)), + model: responses.filter(response => response.model).at(-1)?.model ?? turn.model, + }]; + }); + if (fresh.length) { + const last = all.at(-1); + return [fresh, { + device: stat.dev, inode: stat.ino, created: stat.birthtimeMs, bytes: boundary as number, + responses: all.length, content: last?.content.length ?? 0, + lastResponseId: last?.id, + contentPrefixHash: sha256Hex(JSON.stringify(last?.content ?? [])), + }]; + } + if (boundary !== undefined) return undefined; + if (attempt < 4) await new Promise(resolve => setTimeout(resolve, 50)); + } + } catch { /* a later idle can retry */ } finally { transcript?.close(); } + return undefined; } diff --git a/src/transcriptFile.ts b/src/transcriptFile.ts index 18ce8b9..5dd9b72 100644 --- a/src/transcriptFile.ts +++ b/src/transcriptFile.ts @@ -9,8 +9,10 @@ import { isPathWithinBase } from './utils.js'; const O_RDONLY_NOFOLLOW = fs.constants.O_RDONLY | fs.constants.O_NOFOLLOW; // Injected subagent context can make the first JSONL record much larger than a -// normal prompt. Bound the synchronous work used to correlate an Agent hook. -const TRANSCRIPT_SCAN_LIMIT_BYTES = 8 * 1024 * 1024; +// normal prompt. Eight MiB comfortably covers that preamble while bounding the +// synchronous work and memory used to correlate an id-less lifecycle hook. +export const TRANSCRIPT_SCAN_LIMIT_BYTES = 8 * 1024 * 1024; +export type ReadBudget = { remaining: number }; export type TranscriptHead = Record & { version?: string; @@ -73,19 +75,22 @@ export class TranscriptFile { /** Read a bounded transcript snapshot without leaving an fd open. * If the bound cuts a record, exclude that partial JSONL line. */ -function readTranscriptPrefix(transcriptPath: string): string | undefined { +export function readTranscriptPrefix( + transcriptPath: string, + limitBytes = TRANSCRIPT_SCAN_LIMIT_BYTES, +): string | undefined { let transcript: TranscriptFile | undefined; try { transcript = new TranscriptFile(transcriptPath); const fd = transcript.getFd(); const fileSize = fs.fstatSync(fd).size; - const want = Math.min(fileSize, TRANSCRIPT_SCAN_LIMIT_BYTES); - if (!want) return undefined; + const want = Math.min(fileSize, limitBytes); + if (want === 0) return undefined; const buffer = Buffer.allocUnsafe(want); let read = 0; while (read < want) { const count = fs.readSync(fd, buffer, read, want - read, read); - if (!count) break; + if (count === 0) break; read += count; } if (read < fileSize) { @@ -101,20 +106,60 @@ function readTranscriptPrefix(transcriptPath: string): string | undefined { } /** Read the first transcript line as JSON without leaving an fd open. */ -export function readFirstTranscriptLine(transcriptPath: string): TranscriptHead | undefined { - const line = readTranscriptPrefix(transcriptPath)?.split('\n', 1)[0]; - if (!line?.trim()) return undefined; +export function readFirstTranscriptLine( + transcriptPath: string, + limitBytes = TRANSCRIPT_SCAN_LIMIT_BYTES, + budget?: ReadBudget, +): TranscriptHead | undefined { + let transcript: TranscriptFile | undefined; try { + transcript = new TranscriptFile(transcriptPath); + const fd = transcript.getFd(); + const fileSize = fs.fstatSync(fd).size; + const want = Math.min( + fileSize, + limitBytes, + Math.max(0, budget?.remaining ?? limitBytes), + ); + if (!want) return undefined; + const chunk = Buffer.allocUnsafe(Math.min(64 * 1024, want)); + const parts: Buffer[] = []; + let offset = 0; + let length = 0; + let complete = false; + while (offset < want) { + const count = fs.readSync(fd, chunk, 0, Math.min(chunk.length, want - offset), offset); + if (!count) break; + offset += count; + if (budget) budget.remaining -= count; + const newline = chunk.subarray(0, count).indexOf(0x0a); + const keep = newline < 0 ? count : newline; + parts.push(Buffer.from(chunk.subarray(0, keep))); + length += keep; + if (newline >= 0) { + complete = true; + break; + } + } + if (!complete && offset < fileSize) return undefined; + const line = Buffer.concat(parts, length).toString('utf8'); + if (!line.trim()) return undefined; return JSON.parse(line) as TranscriptHead; } catch { return undefined; + } finally { + transcript?.close(); } } export function subagentTranscriptPath(parentTranscriptPath: string, agentId: string): string { + return path.join(subagentsDirectory(parentTranscriptPath), `agent-${agentId}.jsonl`); +} + +export function subagentsDirectory(parentTranscriptPath: string): string { const projectDir = path.dirname(parentTranscriptPath); const sessionId = path.basename(parentTranscriptPath, '.jsonl'); - return path.join(projectDir, sessionId, 'subagents', `agent-${agentId}.jsonl`); + return path.join(projectDir, sessionId, 'subagents'); } /** Read the dispatch prompt used to join an id-less lifecycle hook to Agent. */ diff --git a/tests/agent-team-test-helpers.ts b/tests/agent-team-test-helpers.ts new file mode 100644 index 0000000..62e09a4 --- /dev/null +++ b/tests/agent-team-test-helpers.ts @@ -0,0 +1,175 @@ +// SPDX-FileCopyrightText: 2026 CoreWeave, Inc. +// SPDX-License-Identifier: MIT +// SPDX-PackageName: weave-claude-code + +import { test, type TestContext } from 'node:test'; +import assert from 'node:assert/strict'; +import * as fs from 'node:fs'; +import { ATTR } from '../src/genaiSpans.ts'; +import { TeamCoordinator } from '../src/teamCoordinator.ts'; +import { + assistantEntry, + childrenOf, + flushWeave, + initWeaveInMemory, + makeGenaiDaemon, + makeTranscript, + spanParentId, + userEntry, +} from './helpers.ts'; + +export { + test, + assert, + fs, + ATTR, + TeamCoordinator, + assistantEntry, + childrenOf, + flushWeave, + initWeaveInMemory, + makeGenaiDaemon, + makeTranscript, + spanParentId, + userEntry, +}; + +export const TEAM = 'review-team'; +export const MEMBER = 'reviewer'; + +export function isTeammateTurn( + span: Parameters[0], + memberName = MEMBER, +): boolean { + return span.attributes[ATTR.AGENT_NAME] === memberName + && spanParentId(span) === undefined; +} + +export function teammateEntries( + sessionId: string, + text: string, + responseId: string, + agentSetting = MEMBER, +) { + return [ + { type: 'agent-setting', agentSetting, sessionId }, + { + type: 'user', + teamName: TEAM, + message: { role: 'user', content: `task: ${text}` }, + }, + assistantEntry(responseId, { type: 'text', text }), + ]; +} + +export async function coordinator( + t: TestContext, + label: string, + promptId?: string, +) { + const exporter = await initWeaveInMemory(); + exporter.reset(); + const sid = `team-${label}`; + const transcript = makeTranscript(t, sid, label); + transcript.append(userEntry('delegate reviews')); + const daemon = makeGenaiDaemon(); + await daemon.routeEvent({ + hook_event_name: 'SessionStart', + session_id: sid, + transcript_path: transcript.file, + source: 'startup', + cwd: '/x', + }); + await daemon.routeEvent({ + hook_event_name: 'UserPromptSubmit', + session_id: sid, + transcript_path: transcript.file, + prompt: 'delegate reviews', + ...(promptId ? { prompt_id: promptId } : {}), + }); + return { exporter, daemon, sid, transcript }; +} + +export async function dispatch( + daemon: ReturnType, + sid: string, + toolUseId: string, + prompt: string, + toolInput: Record = { + subagent_type: MEMBER, + prompt, + team_name: TEAM, + name: MEMBER, + }, +) { + await preDispatch(daemon, sid, toolUseId, toolInput); + await postDispatch(daemon, sid, toolUseId, toolInput); +} + +export async function preDispatch( + daemon: ReturnType, + sid: string, + toolUseId: string, + toolInput: Record, +) { + await daemon.routeEvent({ + hook_event_name: 'PreToolUse', + session_id: sid, + tool_use_id: toolUseId, + tool_name: 'Agent', + tool_input: toolInput, + }); +} + +export async function postDispatch( + daemon: ReturnType, + sid: string, + toolUseId: string, + toolInput: Record, +) { + await daemon.routeEvent({ + hook_event_name: 'PostToolUse', + session_id: sid, + tool_use_id: toolUseId, + tool_name: 'Agent', + tool_input: toolInput, + tool_response: 'dispatched', + }); +} + +export function writeMetadata(transcriptPath: string, agentType = MEMBER) { + fs.writeFileSync( + transcriptPath.replace(/\.jsonl$/, '.meta.json'), + JSON.stringify({ agentType }), + ); +} + +export async function startQueueBlocker( + t: TestContext, + daemon: ReturnType, + label: string, +): Promise<{ blocking: Promise }> { + const sessionId = `${label}-blocker`; + const transcript = makeTranscript(t, sessionId, sessionId); + transcript.append(userEntry('block queue')); + await daemon.routeEvent({ + hook_event_name: 'SessionStart', + session_id: sessionId, + transcript_path: transcript.file, + source: 'startup', + cwd: '/x', + }); + await daemon.routeEvent({ + hook_event_name: 'UserPromptSubmit', + session_id: sessionId, + prompt: 'block queue', + }); + return { + blocking: daemon.routeEvent({ + hook_event_name: 'SubagentStart', + session_id: sessionId, + agent_id: 'missing-transcript', + agent_type: 'general-purpose', + }), + }; +} diff --git a/tests/agent-teams-core.test.ts b/tests/agent-teams-core.test.ts index 1ac5548..a36f6b4 100644 --- a/tests/agent-teams-core.test.ts +++ b/tests/agent-teams-core.test.ts @@ -2,53 +2,26 @@ // SPDX-License-Identifier: MIT // SPDX-PackageName: weave-claude-code -import { test, type TestContext } from 'node:test'; -import assert from 'node:assert/strict'; -import { ATTR } from '../src/genaiSpans.ts'; +import type { TestContext } from 'node:test'; import { + ATTR, + MEMBER, + TEAM, + assert, assistantEntry, + coordinator, + dispatch, flushWeave, initWeaveInMemory, makeGenaiDaemon, makeTranscript, + postDispatch, + preDispatch, spanParentId, + teammateEntries, + test, userEntry, -} from './helpers.ts'; - -const TEAM = 'review-team'; -const MEMBER = 'reviewer'; - -function teammateEntries(sessionId: string, text: string, responseId: string) { - return [ - { type: 'agent-setting', agentSetting: MEMBER, sessionId }, - { type: 'user', teamName: TEAM, message: { role: 'user', content: `task: ${text}` } }, - assistantEntry(responseId, { type: 'text', text }), - ]; -} - -async function coordinator(t: TestContext, label: string, promptId?: string) { - const exporter = await initWeaveInMemory(); - exporter.reset(); - const sid = `team-${label}`; - const transcript = makeTranscript(t, sid, label); - transcript.append(userEntry('delegate reviews')); - const daemon = makeGenaiDaemon(); - await daemon.routeEvent({ - hook_event_name: 'SessionStart', - session_id: sid, - transcript_path: transcript.file, - source: 'startup', - cwd: '/x', - }); - await daemon.routeEvent({ - hook_event_name: 'UserPromptSubmit', - session_id: sid, - transcript_path: transcript.file, - prompt: 'delegate reviews', - ...(promptId ? { prompt_id: promptId } : {}), - }); - return { exporter, daemon, sid, transcript }; -} +} from './agent-team-test-helpers.ts'; const teamInput = (prompt: string): Record => ({ subagent_type: MEMBER, @@ -57,48 +30,6 @@ const teamInput = (prompt: string): Record => ({ name: MEMBER, }); -async function preDispatch( - daemon: ReturnType, - sid: string, - toolUseId: string, - input: Record, -) { - await daemon.routeEvent({ - hook_event_name: 'PreToolUse', - session_id: sid, - tool_use_id: toolUseId, - tool_name: 'Agent', - tool_input: input, - }); -} - -async function postDispatch( - daemon: ReturnType, - sid: string, - toolUseId: string, - input: Record, -) { - await daemon.routeEvent({ - hook_event_name: 'PostToolUse', - session_id: sid, - tool_use_id: toolUseId, - tool_name: 'Agent', - tool_input: input, - tool_response: 'dispatched', - }); -} - -async function dispatch( - daemon: ReturnType, - sid: string, - toolUseId: string, - prompt: string, -) { - const input = teamInput(prompt); - await preDispatch(daemon, sid, toolUseId, input); - await postDispatch(daemon, sid, toolUseId, input); -} - async function idle( t: TestContext, daemon: ReturnType, diff --git a/tests/agent-teams-correlation.test.ts b/tests/agent-teams-correlation.test.ts new file mode 100644 index 0000000..9923419 --- /dev/null +++ b/tests/agent-teams-correlation.test.ts @@ -0,0 +1,314 @@ +// SPDX-FileCopyrightText: 2026 CoreWeave, Inc. +// SPDX-License-Identifier: MIT +// SPDX-PackageName: weave-claude-code + +import { + ATTR, MEMBER, TEAM, TeamCoordinator, assert, assistantEntry, coordinator, dispatch, + flushWeave, fs, initWeaveInMemory, makeGenaiDaemon, makeTranscript, postDispatch, + preDispatch, startQueueBlocker, teammateEntries, test, userEntry, writeMetadata, +} from './agent-team-test-helpers.ts'; + +test('receipt snapshot discovery fails closed at its candidate limit', (t) => { + const owner = makeTranscript(t, 'bounded-snapshot-owner', 'bounded-snapshot'); + for (let i = 0; i < 65; i++) { + owner.subagent(`candidate-${i}`, { + type: 'agent-setting', agentSetting: MEMBER, sessionId: 'bounded-snapshot-member', + }); + } + assert.deepEqual( + new TeamCoordinator().snapshotTranscripts( + 'bounded-snapshot-member', undefined, [], [owner.file], + ), + [], + ); + + const direct = makeTranscript(t, 'bounded-snapshot-member', 'bounded-snapshot-direct'); + direct.append({ + type: 'agent-setting', agentSetting: MEMBER, sessionId: 'bounded-snapshot-member', + }); + const directSnapshots = new TeamCoordinator().snapshotTranscripts( + 'bounded-snapshot-member', direct.file, [], [owner.file], + ); + assert.equal(directSnapshots.length, 1); + assert.equal(directSnapshots[0].path, direct.file); +}); + +test('receipt discovery reads large first records without charging trailing transcript bytes', (t) => { + const owner = makeTranscript(t, 'large-snapshot-owner', 'large-snapshot'); + const teammate = owner.subagent('large-candidate', { + type: 'agent-setting', agentSetting: MEMBER, sessionId: 'large-snapshot-member', + injectedContext: 'x'.repeat(300 * 1024), + }); + for (let i = 0; i < 3; i++) { + owner.subagent( + `large-irrelevant-${i}`, + { type: 'agent-setting', agentSetting: MEMBER, sessionId: `irrelevant-${i}` }, + { type: 'progress', data: 'x'.repeat(8 * 1024 * 1024) }, + ); + } + const snapshots = new TeamCoordinator().snapshotTranscripts( + 'large-snapshot-member', undefined, [], [owner.file], + ); + assert.equal(snapshots.length, 1); + assert.equal(snapshots[0].path, teammate); +}); + +test('metadata correlation accepts a large injected first record', async (t) => { + const { exporter, daemon, sid, transcript } = await coordinator(t, 'large-metadata'); + await dispatch(daemon, sid, 'large-metadata-call', 'review'); + const teammate = transcript.subagent( + 'large-metadata-candidate', + { + type: 'agent-setting', agentSetting: MEMBER, sessionId: 'large-metadata-member', + injectedContext: 'x'.repeat(300 * 1024), + }, + { type: 'user', teamName: TEAM, message: { role: 'user', content: 'task' } }, + assistantEntry('large-metadata-msg', { type: 'text', text: 'large result' }), + ); + writeMetadata(teammate); + await daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: 'large-metadata-member', + team_name: TEAM, teammate_name: MEMBER, + }); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await flushWeave(); + + const agent = exporter.getFinishedSpans().find(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'large-metadata-call'); + assert.ok(agent); + assert.equal( + agent.attributes[ATTR.OUTPUT_MESSAGES], + JSON.stringify([{ role: 'assistant', content: 'large result' }]), + ); +}); + +test('an undeclared team alias does not swallow an unrelated lifecycle without session evidence', async (t) => { + const { exporter, daemon, sid } = await coordinator(t, 'unrelated-lifecycle'); + await dispatch(daemon, sid, 'pending-team-call', 'review it', { + description: 'review code', prompt: 'review it', name: MEMBER, + }); + + const unrelated = makeTranscript(t, 'unrelated-session', 'unrelated-session'); + const agentId = 'unrelated-agent'; + unrelated.subagent( + agentId, + ...teammateEntries('unrelated-session', 'unrelated result', 'unrelated-msg', 'general-purpose'), + ); + await daemon.routeEvent({ + hook_event_name: 'SessionStart', session_id: 'unrelated-session', + transcript_path: unrelated.file, source: 'startup', cwd: '/x', + }); + await daemon.routeEvent({ + hook_event_name: 'SubagentStart', session_id: 'unrelated-session', + agent_id: agentId, agent_type: 'general-purpose', + }); + await daemon.drain('SIGTERM'); + await flushWeave(); + + assert.equal(exporter.getFinishedSpans().filter(span => + span.attributes[ATTR.AGENT_ID] === agentId).length, 1); +}); + +test('a matching prompt alone does not swallow an unrelated lifecycle', async (t) => { + const { exporter, daemon, sid } = await coordinator(t, 'alias-lifecycle'); + const input = { description: 'review code', prompt: 'exact review task', name: MEMBER }; + await dispatch(daemon, sid, 'alias-team-call', 'exact review task', input); + + const teammate = makeTranscript(t, 'alias-member', 'alias-member'); + teammate.append(...teammateEntries( + 'alias-member', 'alias result', 'alias-team-msg', 'general-purpose', + )); + const agentId = 'alias-external-agent'; + const subPath = teammate.subagent( + agentId, + { type: 'agent-setting', agentSetting: 'general-purpose', sessionId: 'alias-member' }, + userEntry('exact review task'), + ); + await daemon.routeEvent({ + hook_event_name: 'SessionStart', session_id: 'alias-member', + transcript_path: teammate.file, source: 'startup', cwd: '/x', + }); + await daemon.routeEvent({ + hook_event_name: 'SubagentStart', session_id: 'alias-member', + agent_id: agentId, agent_type: 'general-purpose', + }); + await daemon.routeEvent({ + hook_event_name: 'SubagentStop', session_id: 'alias-member', + agent_id: agentId, agent_type: 'general-purpose', agent_transcript_path: subPath, + }); + await daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: 'alias-member', + transcript_path: teammate.file, team_name: TEAM, teammate_name: MEMBER, + }); + await daemon.drain('SIGTERM'); + await flushWeave(); + + const spans = exporter.getFinishedSpans(); + assert.equal(spans.some(span => span.attributes[ATTR.AGENT_ID] === agentId), true); + assert.ok(spans.some(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'alias-team-call')); +}); + +test('idle receipt snapshots a metadata-selected transcript', async (t) => { + const exporter = await initWeaveInMemory(); + exporter.reset(); + const daemon = makeGenaiDaemon(); + const sid = 'team-metadata-boundary'; + const transcript = makeTranscript(t, sid, 'metadata-boundary'); + transcript.append(userEntry('delegate reviews')); + const firstInput = { + subagent_type: MEMBER, prompt: 'first', team_name: TEAM, name: MEMBER, + }; + const secondInput = { + subagent_type: MEMBER, prompt: 'second', team_name: TEAM, name: MEMBER, + }; + const linked = transcript.subagent( + 'metadata-boundary-agent', + ...teammateEntries( + 'metadata-boundary-member', 'first result', 'metadata-boundary-msg-1', + ), + ); + writeMetadata(linked); + + const { blocking } = await startQueueBlocker(t, daemon, 'metadata-boundary'); + const coordinatorStart = daemon.routeEvent({ + hook_event_name: 'SessionStart', session_id: sid, + transcript_path: transcript.file, source: 'startup', cwd: '/x', + }); + const coordinatorPrompt = daemon.routeEvent({ + hook_event_name: 'UserPromptSubmit', session_id: sid, + transcript_path: transcript.file, prompt: 'delegate reviews', + }); + const queuedDispatch = [ + preDispatch(daemon, sid, 'metadata-boundary-call-1', firstInput), + postDispatch(daemon, sid, 'metadata-boundary-call-1', firstInput), + preDispatch(daemon, sid, 'metadata-boundary-call-2', secondInput), + postDispatch(daemon, sid, 'metadata-boundary-call-2', secondInput), + ]; + const idle = { + hook_event_name: 'TeammateIdle', session_id: 'metadata-boundary-member', + team_name: TEAM, teammate_name: MEMBER, + }; + const firstIdle = daemon.routeEvent(idle); + fs.appendFileSync(linked, [ + JSON.stringify({ + type: 'user', teamName: TEAM, + message: { role: 'user', content: 'task: second' }, + }), + JSON.stringify(assistantEntry( + 'metadata-boundary-msg-2', + { type: 'text', text: 'second result' }, + )), + '', + ].join('\n')); + const secondIdle = daemon.routeEvent(idle); + await Promise.all([ + blocking, coordinatorStart, coordinatorPrompt, + ...queuedDispatch, firstIdle, secondIdle, + ]); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await flushWeave(); + + const agents = exporter.getFinishedSpans().filter(span => + String(span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID]) + .startsWith('metadata-boundary-call-')); + assert.deepEqual(Object.fromEntries(agents.map(agent => [ + agent.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID], + agent.attributes[ATTR.OUTPUT_MESSAGES], + ])), { + 'metadata-boundary-call-1': JSON.stringify([{ role: 'assistant', content: 'first result' }]), + 'metadata-boundary-call-2': JSON.stringify([{ role: 'assistant', content: 'second result' }]), + }); +}); + +test('ambiguous exact transcript evidence cannot fall through to a weaker owner', async (t) => { + const exporter = await initWeaveInMemory(); + exporter.reset(); + const daemon = makeGenaiDaemon(); + const owners = [ + { sid: 'ambiguous-owner-a', transcript: makeTranscript(t, 'ambiguous-owner-a', 'a') }, + { sid: 'ambiguous-owner-b', transcript: makeTranscript(t, 'ambiguous-owner-b', 'b') }, + ]; + for (const owner of owners) { + owner.transcript.append(userEntry('delegate reviews')); + await daemon.routeEvent({ + hook_event_name: 'SessionStart', session_id: owner.sid, + transcript_path: owner.transcript.file, source: 'startup', cwd: '/x', + }); + await daemon.routeEvent({ + hook_event_name: 'UserPromptSubmit', session_id: owner.sid, prompt: 'delegate reviews', + }); + await dispatch(daemon, owner.sid, `${owner.sid}-call`, 'same review'); + } + for (const agentId of ['ambiguous-meta-a', 'ambiguous-meta-b']) { + const candidate = owners[0].transcript.subagent( + agentId, + ...teammateEntries('ambiguous-idle-member', 'ambiguous result', `${agentId}-msg`), + ); + writeMetadata(candidate); + } + const teammate = makeTranscript(t, 'ambiguous-idle-member', 'ambiguous-idle-member'); + teammate.append(...teammateEntries( + 'ambiguous-idle-member', 'weak result', 'ambiguous-weak-msg', + )); + await daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: 'ambiguous-idle-member', + transcript_path: teammate.file, team_name: TEAM, teammate_name: MEMBER, + }); + await daemon.drain('SIGTERM'); + await flushWeave(); + + const spans = exporter.getFinishedSpans(); + assert.equal(spans.some(span => span.attributes[ATTR.RESPONSE_ID] === 'ambiguous-weak-msg'), false); + const agents = spans.filter(span => + String(span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID]) + .startsWith('ambiguous-owner-')); + assert.equal(agents.length, 2); + for (const agent of agents) { + assert.equal(agent.attributes[ATTR.WEAVE_ORPHAN_REASON], 'daemon_shutdown'); + } +}); + +test('completed lifecycle history remains scoped to its exact team', async (t) => { + const { exporter, daemon, sid } = await coordinator(t, 'team-history'); + const firstInput = { + subagent_type: MEMBER, prompt: 'team x task', team_name: 'team-x', name: MEMBER, + }; + await dispatch(daemon, sid, 'team-x-call', 'team x task', firstInput); + const member = makeTranscript(t, 'shared-team-session', 'shared-team-session'); + member.append(...teammateEntries( + 'shared-team-session', 'team x result', 'team-x-msg', + ).map(entry => 'teamName' in entry ? { ...entry, teamName: 'team-x' } : entry)); + await daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: 'shared-team-session', + transcript_path: member.file, team_name: 'team-x', teammate_name: MEMBER, + }); + + await daemon.routeEvent({ + hook_event_name: 'SessionStart', session_id: 'shared-team-session', + transcript_path: member.file, source: 'startup', cwd: '/x', + }); + const agentId = 'team-y-lifecycle'; + const teamYPath = member.subagent( + agentId, + { type: 'agent-setting', agentSetting: MEMBER, sessionId: 'shared-team-session' }, + { type: 'user', teamName: 'team-y', message: { role: 'user', content: 'team y task' } }, + assistantEntry('team-y-msg', { type: 'text', text: 'team y result' }), + ); + await daemon.routeEvent({ + hook_event_name: 'SubagentStop', session_id: 'shared-team-session', + agent_id: agentId, agent_type: MEMBER, agent_transcript_path: teamYPath, + }); + await daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: 'shared-team-session', + transcript_path: teamYPath, team_name: 'team-y', teammate_name: MEMBER, + }); + await daemon.drain('SIGTERM'); + await flushWeave(); + + const spans = exporter.getFinishedSpans(); + assert.ok(spans.some(span => span.attributes[ATTR.RESPONSE_ID] === 'team-y-msg')); + const teamY = spans.find(span => span.attributes[ATTR.AGENT_ID] === agentId); + assert.ok(teamY); + assert.equal(teamY.attributes[ATTR.WEAVE_ORPHAN_REASON], undefined); +}); diff --git a/tests/agent-teams-lifecycle.test.ts b/tests/agent-teams-lifecycle.test.ts new file mode 100644 index 0000000..f1bd7c6 --- /dev/null +++ b/tests/agent-teams-lifecycle.test.ts @@ -0,0 +1,478 @@ +// SPDX-FileCopyrightText: 2026 CoreWeave, Inc. +// SPDX-License-Identifier: MIT +// SPDX-PackageName: weave-claude-code + +import { + ATTR, MEMBER, TEAM, assert, assistantEntry, childrenOf, coordinator, dispatch, + flushWeave, fs, initWeaveInMemory, isTeammateTurn, makeGenaiDaemon, + makeTranscript, postDispatch, preDispatch, spanParentId, teammateEntries, test, + userEntry, writeMetadata, +} from './agent-team-test-helpers.ts'; + +test('same-name respawns consume distinct transcripts and duplicate idle is idempotent', async (t) => { + const { exporter, daemon, sid, transcript } = await coordinator(t, 'fifo'); + await dispatch(daemon, sid, 'team-call-1', 'first', { + description: 'first review', prompt: 'first', name: MEMBER, + }); + await dispatch(daemon, sid, 'team-call-2', 'second', { + description: 'second review', prompt: 'second', name: MEMBER, + }); + + const firstPath = transcript.subagent( + 'first', ...teammateEntries('teammate-1', 'first result', 'team-msg-1', 'general-purpose')); + const secondPath = transcript.subagent( + 'second', ...teammateEntries('teammate-2', 'second result', 'team-msg-2', 'general-purpose')); + writeMetadata(firstPath, 'general-purpose'); + writeMetadata(secondPath, 'general-purpose'); + + // A teammate session can also emit SubagentStart. The queued coordinator + // dispatch owns it, so this must not manufacture a second Agent marker. + const teammateSession = makeTranscript(t, 'teammate-1', 'team-external'); + teammateSession.append(...teammateEntries( + 'teammate-1', 'first result', 'team-msg-1', 'general-purpose', + )); + const secondTeammateSession = makeTranscript(t, 'teammate-2', 'team-external-second'); + secondTeammateSession.append(...teammateEntries( + 'teammate-2', 'second result', 'team-msg-2', 'general-purpose', + )); + const externalId = 'external-lifecycle'; + teammateSession.subagent( + externalId, + ...teammateEntries('teammate-1', 'first result', 'external-msg', 'general-purpose'), + ); + await daemon.routeEvent({ + hook_event_name: 'SessionStart', session_id: 'teammate-1', + transcript_path: teammateSession.file, source: 'startup', cwd: '/x', + }); + await daemon.routeEvent({ + hook_event_name: 'SubagentStart', session_id: 'teammate-1', + agent_id: externalId, agent_type: 'general-purpose', + }); + + const firstIdle = { + hook_event_name: 'TeammateIdle', session_id: 'teammate-1', + transcript_path: teammateSession.file, team_name: TEAM, teammate_name: MEMBER, + }; + await daemon.routeEvent(firstIdle); + fs.appendFileSync(firstPath, `${JSON.stringify(assistantEntry( + 'mutated-after-idle', + { type: 'text', text: 'late mutation' }, + ))}\n`); + await daemon.routeEvent(firstIdle); // duplicate must not consume call 2 + await daemon.routeEvent({ + ...firstIdle, + session_id: 'teammate-2', + transcript_path: secondTeammateSession.file, + }); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await flushWeave(); + + const spans = exporter.getFinishedSpans(); + const agents = spans.filter(span => + ['team-call-1', 'team-call-2'].includes( + String(span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID]), + )); + assert.equal(agents.length, 2); + assert.deepEqual( + Object.fromEntries(agents.map(span => [ + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID], + span.attributes[ATTR.OUTPUT_MESSAGES], + ])), + { + 'team-call-1': JSON.stringify([{ role: 'assistant', content: 'first result' }]), + 'team-call-2': JSON.stringify([{ role: 'assistant', content: 'second result' }]), + }, + ); + assert.equal(spans.some(span => span.attributes[ATTR.AGENT_ID] === externalId), false); + assert.deepEqual( + spans.filter(span => isTeammateTurn(span)) + .flatMap(turn => childrenOf(spans, turn)) + .map(span => span.attributes[ATTR.RESPONSE_ID]) + .sort(), + ['team-msg-1', 'team-msg-2'], + ); +}); + +test('same-session agent-setting lifecycle waits for TeammateIdle', async (t) => { + const exporter = await initWeaveInMemory(); + exporter.reset(); + const sid = 'team-same-session'; + const transcript = makeTranscript(t, sid, 'same-session'); + transcript.append( + { type: 'last-prompt', sessionId: sid }, + userEntry('delegate reviews'), + assistantEntry('same-session-root-msg', { + type: 'text', text: 'root coordinator output', + }), + ); + const daemon = makeGenaiDaemon(); + await daemon.routeEvent({ + hook_event_name: 'SessionStart', session_id: sid, + transcript_path: transcript.file, source: 'startup', cwd: '/x', + }); + await daemon.routeEvent({ + hook_event_name: 'UserPromptSubmit', session_id: sid, + transcript_path: transcript.file, prompt: 'delegate reviews', + }); + const agentId = 'same-session-agent'; + const subPath = transcript.subagent( + agentId, + ...teammateEntries(sid, 'same-session result', 'same-session-msg'), + ); + await daemon.routeEvent({ + hook_event_name: 'SubagentStart', session_id: sid, + agent_id: agentId, agent_type: MEMBER, + }); + await daemon.routeEvent({ + hook_event_name: 'SubagentStop', session_id: sid, + agent_id: agentId, agent_type: MEMBER, agent_transcript_path: subPath, + }); + assert.equal(exporter.getFinishedSpans().some(span => span.attributes[ATTR.AGENT_ID] === agentId), false); + + await daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: sid, + transcript_path: transcript.file, team_name: TEAM, teammate_name: MEMBER, + }); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await flushWeave(); + + const spans = exporter.getFinishedSpans(); + const agent = spans.find(span => span.attributes[ATTR.AGENT_ID] === agentId); + const teammateTurn = spans.find(span => isTeammateTurn(span)); + const chat = spans.find(span => span.attributes[ATTR.RESPONSE_ID] === 'same-session-msg'); + assert.ok(agent && teammateTurn && chat); + assert.equal(spanParentId(chat), teammateTurn.spanContext().spanId); + assert.notEqual(spanParentId(chat), agent.spanContext().spanId); + assert.equal( + agent.attributes[ATTR.OUTPUT_MESSAGES], + JSON.stringify([{ role: 'assistant', content: 'same-session result' }]), + ); +}); + +test('current generic named Agent payload is traced as an implicit-team dispatch', async (t) => { + const { exporter, daemon, sid } = await coordinator(t, 'implicit'); + const input = { description: 'review code', prompt: 'inspect it', name: MEMBER }; + await dispatch(daemon, sid, 'implicit-call', 'inspect it', input); + + const teammate = makeTranscript(t, 'implicit-member', 'implicit-member'); + teammate.append( + { type: 'agent-setting', agentSetting: 'general-purpose', sessionId: 'implicit-member' }, + { type: 'user', teamName: TEAM, message: { role: 'user', content: 'task: implicit' } }, + assistantEntry('implicit-msg', { type: 'text', text: 'implicit result' }), + ); + await daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: 'implicit-member', + transcript_path: teammate.file, team_name: TEAM, teammate_name: MEMBER, + }); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await flushWeave(); + + const spans = exporter.getFinishedSpans(); + const agent = spans.find(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'implicit-call'); + assert.ok(agent); + assert.equal(agent.attributes[ATTR.AGENT_NAME], MEMBER); + assert.equal(agent.attributes[ATTR.OUTPUT_MESSAGES], JSON.stringify([ + { role: 'assistant', content: 'implicit result' }, + ])); +}); + +test('a tentative named Agent failure still accepts its ordinary Stop transcript', async (t) => { + const { exporter, daemon, sid, transcript } = await coordinator(t, 'named-failure'); + const input = { description: 'background work', prompt: 'ordinary failure', name: 'worker' }; + const agentId = 'ordinary-failed-agent'; + const subPath = transcript.subagent( + agentId, + userEntry('ordinary failure'), + assistantEntry('ordinary-failed-msg', { type: 'text', text: 'partial result' }), + ); + await preDispatch(daemon, sid, 'ordinary-failed-call', input); + await daemon.routeEvent({ + hook_event_name: 'SubagentStart', session_id: sid, + agent_id: agentId, agent_type: 'general-purpose', + }); + await daemon.routeEvent({ + hook_event_name: 'PostToolUseFailure', session_id: sid, + tool_use_id: 'ordinary-failed-call', tool_name: 'Agent', tool_input: input, + error: 'AgentError: failed after partial output', + }); + await daemon.routeEvent({ + hook_event_name: 'SubagentStop', session_id: sid, + agent_id: agentId, agent_type: 'general-purpose', agent_transcript_path: subPath, + }); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await flushWeave(); + + const agents = exporter.getFinishedSpans().filter(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'ordinary-failed-call'); + assert.equal(agents.length, 1); + assert.equal(agents[0].attributes[ATTR.ERROR_TYPE], 'AgentError'); + assert.equal(agents[0].attributes[ATTR.WEAVE_ORPHAN_REASON], undefined); + assert.ok(exporter.getFinishedSpans().some(span => + span.attributes[ATTR.RESPONSE_ID] === 'ordinary-failed-msg')); +}); + +test('agent-setting without a team remains an ordinary recovered lifecycle', async (t) => { + const { exporter, daemon, sid, transcript } = await coordinator(t, 'ordinary-agent-setting'); + const agentId = 'ordinary-setting-agent'; + const subPath = transcript.subagent( + agentId, + { type: 'agent-setting', agentSetting: 'general-purpose', sessionId: sid }, + userEntry('ordinary recovered task'), + assistantEntry('ordinary-setting-msg', { type: 'text', text: 'ordinary recovered result' }), + ); + await daemon.routeEvent({ + hook_event_name: 'SubagentStop', session_id: sid, + agent_id: agentId, agent_type: 'general-purpose', agent_transcript_path: subPath, + }); + await daemon.drain('SIGTERM'); + await flushWeave(); + + const agent = exporter.getFinishedSpans().find(span => + span.attributes[ATTR.AGENT_ID] === agentId); + assert.ok(agent); + assert.equal(agent.attributes[ATTR.WEAVE_ORPHAN_REASON], undefined); + assert.ok(exporter.getFinishedSpans().some(span => + span.attributes[ATTR.RESPONSE_ID] === 'ordinary-setting-msg')); +}); + +test('an ordinary child inside a teammate session remains an ordinary Agent', async (t) => { + const exporter = await initWeaveInMemory(); + exporter.reset(); + const sid = 'teammate-with-child'; + const daemon = makeGenaiDaemon(); + const coordinatorTranscript = makeTranscript(t, 'child-coordinator', 'child-coordinator'); + coordinatorTranscript.append(userEntry('delegate reviews')); + await daemon.routeEvent({ + hook_event_name: 'SessionStart', session_id: 'child-coordinator', + transcript_path: coordinatorTranscript.file, source: 'startup', cwd: '/x', + }); + await daemon.routeEvent({ + hook_event_name: 'UserPromptSubmit', session_id: 'child-coordinator', + prompt: 'delegate reviews', + }); + await dispatch(daemon, 'child-coordinator', 'child-coordinator-call', 'parent team task'); + const linked = coordinatorTranscript.subagent( + 'linked-parent', ...teammateEntries(sid, 'parent result', 'linked-parent-msg'), + ); + writeMetadata(linked); + + const transcript = makeTranscript(t, sid, 'teammate-with-child'); + transcript.append(...teammateEntries(sid, 'parent result', 'parent-team-msg')); + const agentId = 'ordinary-team-child'; + const subPath = transcript.subagent( + agentId, + { type: 'agent-setting', agentSetting: 'general-purpose', sessionId: sid }, + userEntry('ordinary nested task'), + assistantEntry('ordinary-team-child-msg', { type: 'text', text: 'child result' }), + ); + await daemon.routeEvent({ + hook_event_name: 'SessionStart', session_id: sid, + transcript_path: transcript.file, source: 'startup', cwd: '/x', + }); + await daemon.routeEvent({ + hook_event_name: 'SubagentStop', session_id: sid, + agent_id: agentId, agent_type: 'general-purpose', agent_transcript_path: subPath, + }); + await daemon.drain('SIGTERM'); + await flushWeave(); + + const agent = exporter.getFinishedSpans().find(span => + span.attributes[ATTR.AGENT_ID] === agentId); + assert.ok(agent); + assert.equal(agent.attributes[ATTR.WEAVE_ORPHAN_REASON], undefined); + assert.ok(exporter.getFinishedSpans().some(span => + span.attributes[ATTR.RESPONSE_ID] === 'ordinary-team-child-msg')); +}); + +test('name-only Team alias remains the display name when lifecycle type differs', async (t) => { + const { exporter, daemon, sid, transcript } = await coordinator(t, 'alias-type'); + const input = { name: 'instance-alias', prompt: 'research it' }; + const agentId = 'alias-type-agent'; + const subPath = transcript.subagent( + agentId, + { type: 'agent-setting', agentSetting: 'general-purpose', sessionId: sid }, + userEntry('research it'), + { type: 'user', teamName: TEAM, message: { role: 'user', content: 'task: research it' } }, + assistantEntry('alias-type-msg', { type: 'text', text: 'researched' }), + ); + + await preDispatch(daemon, sid, 'alias-type-call', input); + await daemon.routeEvent({ + hook_event_name: 'SubagentStart', session_id: sid, + agent_id: agentId, agent_type: 'general-purpose', + }); + await daemon.routeEvent({ + hook_event_name: 'SubagentStop', session_id: sid, + agent_id: agentId, agent_type: 'general-purpose', agent_transcript_path: subPath, + }); + await postDispatch(daemon, sid, 'alias-type-call', input); + await daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: sid, + transcript_path: transcript.file, team_name: TEAM, teammate_name: 'instance-alias', + }); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await flushWeave(); + + const spans = exporter.getFinishedSpans(); + const agent = spans.find(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'alias-type-call'); + assert.ok(agent); + assert.equal(agent.attributes[ATTR.AGENT_NAME], 'instance-alias'); + assert.equal(agent.attributes[ATTR.AGENT_ID], agentId); + assert.ok(spans.some(span => span.attributes[ATTR.RESPONSE_ID] === 'alias-type-msg')); +}); + +test('restart-first name-only Team learns lifecycle type and completes on idle', async (t) => { + const exporter = await initWeaveInMemory(); + exporter.reset(); + const sid = 'team-restart-name-only'; + const agentId = 'restart-name-only-agent'; + const transcript = makeTranscript(t, sid, sid); + transcript.append(userEntry('delegate it')); + const subPath = transcript.subagent( + agentId, + { type: 'agent-setting', agentSetting: 'general-purpose', sessionId: sid }, + userEntry('research it'), + { type: 'user', teamName: TEAM, message: { role: 'user', content: 'task: research it' } }, + assistantEntry('restart-name-msg', { type: 'text', text: 'restart result' }), + ); + const daemon = makeGenaiDaemon(); + const input = { name: 'instance-alias', prompt: 'research it' }; + + await daemon.routeEvent({ + hook_event_name: 'PostToolUse', session_id: sid, + transcript_path: transcript.file, cwd: '/x', + tool_use_id: 'restart-name-call', tool_name: 'Agent', + tool_input: input, tool_response: 'dispatched', + }); + await daemon.routeEvent({ + hook_event_name: 'SubagentStart', session_id: sid, + agent_id: agentId, agent_type: 'general-purpose', + }); + await daemon.routeEvent({ + hook_event_name: 'SubagentStop', session_id: sid, + agent_id: agentId, agent_type: 'general-purpose', agent_transcript_path: subPath, + }); + await daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: sid, + transcript_path: transcript.file, team_name: TEAM, teammate_name: 'instance-alias', + }); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await flushWeave(); + + const spans = exporter.getFinishedSpans(); + const agent = spans.find(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'restart-name-call'); + assert.ok(agent); + assert.equal(agent.attributes[ATTR.AGENT_NAME], 'instance-alias'); + assert.equal(agent.attributes[ATTR.AGENT_ID], agentId); + assert.ok(spans.some(span => span.attributes[ATTR.RESPONSE_ID] === 'restart-name-msg')); +}); + +test('completed Team alias suppresses a delayed lifecycle with its recorded type', async (t) => { + const { exporter, daemon, sid } = await coordinator(t, 'delayed-alias-type'); + const input = { name: 'instance-alias', prompt: 'research it' }; + await dispatch(daemon, sid, 'delayed-alias-call', 'research it', input); + + const teammateSessionId = 'delayed-alias-member'; + const teammate = makeTranscript(t, teammateSessionId, teammateSessionId); + teammate.append( + { + type: 'agent-setting', + agentSetting: 'general-purpose', + sessionId: teammateSessionId, + }, + { + type: 'user', + teamName: TEAM, + message: { role: 'user', content: 'task: research it' }, + }, + assistantEntry('delayed-alias-msg', { type: 'text', text: 'researched' }), + ); + await daemon.routeEvent({ + hook_event_name: 'TeammateIdle', + session_id: teammateSessionId, + transcript_path: teammate.file, + team_name: TEAM, + teammate_name: 'instance-alias', + }); + + const delayedAgentId = 'delayed-alias-lifecycle'; + const delayedPath = teammate.subagent( + delayedAgentId, + { + type: 'agent-setting', + agentSetting: 'general-purpose', + sessionId: teammateSessionId, + }, + { + type: 'user', + teamName: TEAM, + message: { role: 'user', content: 'task: research it' }, + }, + assistantEntry('delayed-lifecycle-msg', { type: 'text', text: 'duplicate result' }), + ); + await daemon.routeEvent({ + hook_event_name: 'SessionStart', + session_id: teammateSessionId, + transcript_path: teammate.file, + source: 'startup', + cwd: '/x', + }); + await daemon.routeEvent({ + hook_event_name: 'SubagentStart', + session_id: teammateSessionId, + agent_id: delayedAgentId, + agent_type: 'general-purpose', + }); + await daemon.routeEvent({ + hook_event_name: 'SubagentStop', + session_id: teammateSessionId, + agent_id: delayedAgentId, + agent_type: 'general-purpose', + agent_transcript_path: delayedPath, + }); + await daemon.drain('SIGTERM'); + await flushWeave(); + + const spans = exporter.getFinishedSpans(); + assert.equal(spans.some(span => span.attributes[ATTR.AGENT_ID] === delayedAgentId), false); + assert.equal(spans.filter(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] + === 'delayed-alias-call').length, 1); + assert.equal(spans.filter(span => + span.attributes[ATTR.RESPONSE_ID] === 'delayed-alias-msg').length, 1); + assert.equal(spans.some(span => + span.attributes[ATTR.RESPONSE_ID] === 'delayed-lifecycle-msg'), false); +}); + +test('PermissionDenied closes an explicit Team and its deferred root once', async (t) => { + const { exporter, daemon, sid } = await coordinator(t, 'permission-denied'); + const input = { + subagent_type: MEMBER, + prompt: 'review denied work', + team_name: TEAM, + name: MEMBER, + }; + await preDispatch(daemon, sid, 'denied-team-call', input); + await daemon.routeEvent({ + hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear', + }); + await daemon.routeEvent({ + hook_event_name: 'PermissionDenied', session_id: sid, + tool_use_id: 'denied-team-call', tool_name: 'Agent', + tool_input: input, reason: 'auto mode denied', + }); + await flushWeave(); + + const spans = exporter.getFinishedSpans(); + const agents = spans.filter(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'denied-team-call'); + const roots = spans.filter(span => + span.attributes[ATTR.AGENT_NAME] === 'claude-code'); + assert.equal(agents.length, 1); + assert.equal(roots.length, 1); + assert.equal(agents[0].attributes[ATTR.ERROR_TYPE], 'permission_denied'); + assert.equal(spans.some(span => isTeammateTurn(span)), false); +}); diff --git a/tests/agent-teams-ordering.test.ts b/tests/agent-teams-ordering.test.ts new file mode 100644 index 0000000..df3e164 --- /dev/null +++ b/tests/agent-teams-ordering.test.ts @@ -0,0 +1,462 @@ +// SPDX-FileCopyrightText: 2026 CoreWeave, Inc. +// SPDX-License-Identifier: MIT +// SPDX-PackageName: weave-claude-code + +import { + ATTR, MEMBER, TEAM, assert, assistantEntry, coordinator, dispatch, flushWeave, fs, + initWeaveInMemory, isTeammateTurn, makeGenaiDaemon, makeTranscript, postDispatch, + preDispatch, startQueueBlocker, teammateEntries, test, userEntry, writeMetadata, +} from './agent-team-test-helpers.ts'; + +test('restart-first receipt stages metadata before queued reconstruction', async (t) => { + const exporter = await initWeaveInMemory(); + exporter.reset(); + const daemon = makeGenaiDaemon(); + const sid = 'team-restart-receipt-owner'; + const transcript = makeTranscript(t, sid, 'restart-receipt'); + transcript.append(userEntry('delegate reviews')); + const teammatePath = transcript.subagent( + 'restart-receipt-agent', + ...teammateEntries( + 'restart-receipt-member', 'restart result', 'restart-receipt-msg', + ), + ); + writeMetadata(teammatePath); + const { blocking } = await startQueueBlocker(t, daemon, 'restart-receipt'); + const input = { + subagent_type: MEMBER, prompt: 'review', team_name: TEAM, name: MEMBER, + }; + const post = daemon.routeEvent({ + hook_event_name: 'PostToolUse', session_id: sid, + transcript_path: transcript.file, tool_use_id: 'restart-receipt-call', + tool_name: 'Agent', tool_input: input, tool_response: 'dispatched', + }); + const idle = daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: 'restart-receipt-member', + team_name: TEAM, teammate_name: MEMBER, + }); + + await Promise.all([blocking, post, idle]); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await flushWeave(); + + const agent = exporter.getFinishedSpans().find(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'restart-receipt-call'); + assert.ok(agent); + assert.equal(agent.attributes[ATTR.WEAVE_ORPHAN_REASON], undefined); + assert.equal( + agent.attributes[ATTR.OUTPUT_MESSAGES], + JSON.stringify([{ role: 'assistant', content: 'restart result' }]), + ); +}); + +test('queued duplicate SessionStart paths preserve the first owner root', async (t) => { + const exporter = await initWeaveInMemory(); + exporter.reset(); + const daemon = makeGenaiDaemon(); + const sid = 'team-conflicting-start-owner'; + const first = makeTranscript(t, sid, 'conflicting-start-first'); + first.append(userEntry('delegate reviews')); + const teammatePath = first.subagent( + 'conflicting-start-agent', + ...teammateEntries( + 'conflicting-start-member', 'first-root result', 'conflicting-start-msg', + ), + ); + writeMetadata(teammatePath); + const duplicate = makeTranscript(t, sid, 'conflicting-start-duplicate'); + duplicate.append(userEntry('wrong duplicate root')); + const { blocking } = await startQueueBlocker(t, daemon, 'conflicting-start'); + const input = { + subagent_type: MEMBER, prompt: 'review', team_name: TEAM, name: MEMBER, + }; + const queued = [ + daemon.routeEvent({ + hook_event_name: 'SessionStart', session_id: sid, + transcript_path: first.file, source: 'startup', cwd: '/x', + }), + daemon.routeEvent({ + hook_event_name: 'UserPromptSubmit', session_id: sid, + transcript_path: first.file, prompt: 'delegate reviews', + }), + daemon.routeEvent({ + hook_event_name: 'SessionStart', session_id: sid, + transcript_path: duplicate.file, source: 'startup', cwd: '/x', + }), + preDispatch(daemon, sid, 'conflicting-start-call', input), + postDispatch(daemon, sid, 'conflicting-start-call', input), + daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: 'conflicting-start-member', + team_name: TEAM, teammate_name: MEMBER, + }), + ]; + + await Promise.all([blocking, ...queued]); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await flushWeave(); + + const agent = exporter.getFinishedSpans().find(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'conflicting-start-call'); + assert.ok(agent); + assert.equal( + agent.attributes[ATTR.OUTPUT_MESSAGES], + JSON.stringify([{ role: 'assistant', content: 'first-root result' }]), + ); +}); + +test('an idle older than a later normal Pre cannot consume that future dispatch', async (t) => { + const { exporter, daemon, sid } = await coordinator(t, 'stale-idle'); + const input = { description: 'review code', prompt: 'later review', name: MEMBER }; + const teammate = makeTranscript(t, 'stale-member', 'stale-member'); + teammate.append(...teammateEntries('stale-member', 'stale result', 'stale-msg')); + const route = daemon as unknown as { + routeEvent(payload: Record, sequence: number): Promise; + }; + await route.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: 'stale-member', + transcript_path: teammate.file, team_name: TEAM, teammate_name: MEMBER, + }, 1); + await route.routeEvent({ + hook_event_name: 'PreToolUse', session_id: sid, tool_use_id: 'later-call', + tool_name: 'Agent', tool_input: input, + }, 2); + await route.routeEvent({ + hook_event_name: 'PostToolUse', session_id: sid, tool_use_id: 'later-call', + tool_name: 'Agent', tool_input: input, tool_response: 'dispatched', + }, 3); + await flushWeave(); + assert.equal(exporter.getFinishedSpans().some(span => + span.attributes[ATTR.RESPONSE_ID] === 'stale-msg'), false); + + await daemon.drain('SIGTERM'); + await flushWeave(); + const agent = exporter.getFinishedSpans().find(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'later-call'); + assert.ok(agent); + assert.equal(agent.attributes[ATTR.WEAVE_ORPHAN_REASON], 'daemon_shutdown'); +}); + +test('restart-first Agent Post registers its dispatch and consumes an earlier idle', async (t) => { + const { exporter, daemon, sid } = await coordinator(t, 'restart-post'); + const teammate = makeTranscript(t, 'restart-member', 'restart-member'); + teammate.append(...teammateEntries('restart-member', 'restart result', 'restart-team-msg')); + await daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: 'restart-member', + transcript_path: teammate.file, team_name: TEAM, teammate_name: MEMBER, + }); + await daemon.routeEvent({ + hook_event_name: 'PostToolUse', session_id: sid, tool_use_id: 'restart-team-call', + tool_name: 'Agent', + tool_input: { description: 'review code', prompt: 'restart review', name: MEMBER }, + tool_response: 'dispatched', + }); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await flushWeave(); + + const agent = exporter.getFinishedSpans().find(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'restart-team-call'); + assert.ok(agent); + assert.equal(agent.attributes[ATTR.WEAVE_ORPHAN_REASON], undefined); + assert.equal(agent.attributes[ATTR.OUTPUT_MESSAGES], JSON.stringify([ + { role: 'assistant', content: 'restart result' }, + ])); +}); + +test('partial teammate transcript retries without consuming the dispatch', async (t) => { + const { exporter, daemon, sid } = await coordinator(t, 'partial'); + await dispatch(daemon, sid, 'partial-team-call', 'partial review'); + const teammate = makeTranscript(t, 'partial-member', 'partial-member'); + const [setting, user] = teammateEntries('partial-member', 'partial result', 'partial-msg'); + teammate.append(setting, user); + const assistant = JSON.stringify(assistantEntry( + 'partial-msg', + { type: 'text', text: 'partial result' }, + )); + const split = assistant.length - 2; + fs.appendFileSync(teammate.file, assistant.slice(0, split)); + const idle = { + hook_event_name: 'TeammateIdle', session_id: 'partial-member', + transcript_path: teammate.file, team_name: TEAM, teammate_name: MEMBER, + }; + + await daemon.routeEvent(idle); + await flushWeave(); + assert.equal(exporter.getFinishedSpans().some(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'partial-team-call'), false); + + fs.appendFileSync(teammate.file, `${assistant.slice(split)}\n`); + await daemon.routeEvent(idle); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await flushWeave(); + assert.ok(exporter.getFinishedSpans().some(span => + span.attributes[ATTR.RESPONSE_ID] === 'partial-msg')); +}); + +test('concurrent idle and duplicate Post emit one teammate response', async (t) => { + const { exporter, daemon, sid } = await coordinator(t, 'concurrent-completion'); + const input = { + subagent_type: MEMBER, prompt: 'concurrent review', team_name: TEAM, name: MEMBER, + }; + await dispatch(daemon, sid, 'concurrent-team-call', 'concurrent review', input); + const teammate = makeTranscript(t, 'concurrent-member', 'concurrent-member'); + const [setting, user] = teammateEntries( + 'concurrent-member', 'concurrent result', 'concurrent-team-msg', + ); + teammate.append(setting, user); + const assistant = JSON.stringify(assistantEntry( + 'concurrent-team-msg', + { type: 'text', text: 'concurrent result' }, + )); + const split = assistant.length - 2; + fs.appendFileSync(teammate.file, assistant.slice(0, split)); + const idle = { + hook_event_name: 'TeammateIdle', session_id: 'concurrent-member', + transcript_path: teammate.file, team_name: TEAM, teammate_name: MEMBER, + }; + const finishWrite = new Promise(resolve => setTimeout(() => { + fs.appendFileSync(teammate.file, `${assistant.slice(split)}\n`); + resolve(); + }, 25)); + await Promise.all([ + daemon.routeEvent(idle), + postDispatch(daemon, sid, 'concurrent-team-call', input), + finishWrite, + ]); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await flushWeave(); + + const spans = exporter.getFinishedSpans(); + assert.equal(spans.filter(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'concurrent-team-call').length, 1); + assert.equal(spans.filter(span => + span.attributes[ATTR.RESPONSE_ID] === 'concurrent-team-msg').length, 1); + assert.equal(spans.filter(span => isTeammateTurn(span)).length, 1); +}); + +test('a complete idle queued behind a partial idle is reconsidered automatically', async (t) => { + const { exporter, daemon, sid } = await coordinator(t, 'concurrent-idles'); + await dispatch(daemon, sid, 'concurrent-idles-call', 'review once'); + const partial = makeTranscript(t, 'partial-idle-member', 'partial-idle-member'); + const [setting, user] = teammateEntries( + 'partial-idle-member', 'partial result', 'partial-idle-msg', + ); + partial.append(setting, user); + fs.appendFileSync(partial.file, '{"type":"assistant"'); + const complete = makeTranscript(t, 'complete-idle-member', 'complete-idle-member'); + complete.append(...teammateEntries( + 'complete-idle-member', 'complete result', 'complete-idle-msg', + )); + + await Promise.all([ + daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: 'partial-idle-member', + transcript_path: partial.file, team_name: TEAM, teammate_name: MEMBER, + }), + daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: 'complete-idle-member', + transcript_path: complete.file, team_name: TEAM, teammate_name: MEMBER, + }), + ]); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await flushWeave(); + + const agent = exporter.getFinishedSpans().find(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'concurrent-idles-call'); + assert.ok(agent); + assert.equal(agent.attributes[ATTR.WEAVE_ORPHAN_REASON], undefined); + assert.ok(exporter.getFinishedSpans().some(span => + span.attributes[ATTR.RESPONSE_ID] === 'complete-idle-msg')); +}); + +test('cross-session idles commit in global receipt order', async (t) => { + const { exporter, daemon, sid } = await coordinator(t, 'idle-receipt-order'); + await dispatch(daemon, sid, 'ordered-team-call-1', 'first task'); + await dispatch(daemon, sid, 'ordered-team-call-2', 'second task'); + + const first = makeTranscript(t, 'ordered-member-1', 'ordered-member-1'); + const [setting, user] = teammateEntries( + 'ordered-member-1', 'first result', 'ordered-team-msg-1', + ); + first.append(setting, user); + const assistant = JSON.stringify(assistantEntry( + 'ordered-team-msg-1', + { type: 'text', text: 'first result' }, + )); + const split = assistant.length - 2; + fs.appendFileSync(first.file, assistant.slice(0, split)); + const second = makeTranscript(t, 'ordered-member-2', 'ordered-member-2'); + second.append(...teammateEntries( + 'ordered-member-2', 'second result', 'ordered-team-msg-2', + )); + + const firstIdle = daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: 'ordered-member-1', + transcript_path: first.file, team_name: TEAM, teammate_name: MEMBER, + }); + const secondIdle = daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: 'ordered-member-2', + transcript_path: second.file, team_name: TEAM, teammate_name: MEMBER, + }); + await new Promise(resolve => setTimeout(() => { + fs.appendFileSync(first.file, `${assistant.slice(split)}\n`); + resolve(); + }, 25)); + await Promise.all([firstIdle, secondIdle]); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await flushWeave(); + + const agents = exporter.getFinishedSpans().filter(span => + String(span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID]) + .startsWith('ordered-team-call-')); + assert.deepEqual(Object.fromEntries(agents.map(agent => [ + agent.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID], + agent.attributes[ATTR.OUTPUT_MESSAGES], + ])), { + 'ordered-team-call-1': JSON.stringify([{ role: 'assistant', content: 'first result' }]), + 'ordered-team-call-2': JSON.stringify([{ role: 'assistant', content: 'second result' }]), + }); +}); + +test('removing an ordinary candidate re-evaluates a buffered ambiguous idle', async (t) => { + const exporter = await initWeaveInMemory(); + exporter.reset(); + const daemon = makeGenaiDaemon(); + const teamOwner = makeTranscript(t, 'reconcile-team-owner', 'reconcile-team-owner'); + const ordinaryOwner = makeTranscript(t, 'reconcile-ordinary-owner', 'reconcile-ordinary-owner'); + for (const [sid, transcript] of [ + ['reconcile-team-owner', teamOwner], + ['reconcile-ordinary-owner', ordinaryOwner], + ] as const) { + transcript.append(userEntry('delegate reviews')); + await daemon.routeEvent({ + hook_event_name: 'SessionStart', session_id: sid, + transcript_path: transcript.file, source: 'startup', cwd: '/x', + }); + await daemon.routeEvent({ + hook_event_name: 'UserPromptSubmit', session_id: sid, prompt: 'delegate reviews', + }); + } + + const teamInput = { description: 'team work', prompt: 'team task', name: MEMBER }; + await dispatch(daemon, 'reconcile-team-owner', 'reconcile-team-call', 'team task', teamInput); + const ordinaryInput = { description: 'ordinary work', prompt: 'ordinary task', name: MEMBER }; + const ordinaryId = 'reconcile-ordinary-agent'; + const ordinaryPath = ordinaryOwner.subagent( + ordinaryId, + userEntry('ordinary task'), + assistantEntry('reconcile-ordinary-msg', { type: 'text', text: 'ordinary result' }), + ); + await preDispatch(daemon, 'reconcile-ordinary-owner', 'reconcile-ordinary-call', ordinaryInput); + await daemon.routeEvent({ + hook_event_name: 'SubagentStart', session_id: 'reconcile-ordinary-owner', + agent_id: ordinaryId, agent_type: MEMBER, + }); + await postDispatch( + daemon, 'reconcile-ordinary-owner', 'reconcile-ordinary-call', ordinaryInput, + ); + + const teammate = makeTranscript(t, 'reconcile-member', 'reconcile-member'); + teammate.append(...teammateEntries( + 'reconcile-member', 'team result', 'reconcile-team-msg', + )); + await daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: 'reconcile-member', + transcript_path: teammate.file, team_name: TEAM, teammate_name: MEMBER, + }); + assert.equal(exporter.getFinishedSpans().some(span => + span.attributes[ATTR.RESPONSE_ID] === 'reconcile-team-msg'), false); + + await daemon.routeEvent({ + hook_event_name: 'SubagentStop', session_id: 'reconcile-ordinary-owner', + agent_id: ordinaryId, agent_type: MEMBER, agent_transcript_path: ordinaryPath, + }); + await daemon.routeEvent({ + hook_event_name: 'SessionEnd', session_id: 'reconcile-team-owner', reason: 'clear', + }); + await daemon.routeEvent({ + hook_event_name: 'SessionEnd', session_id: 'reconcile-ordinary-owner', reason: 'clear', + }); + await flushWeave(); + + const spans = exporter.getFinishedSpans(); + assert.ok(spans.some(span => span.attributes[ATTR.RESPONSE_ID] === 'reconcile-team-msg')); + assert.ok(spans.some(span => span.attributes[ATTR.RESPONSE_ID] === 'reconcile-ordinary-msg')); + for (const toolUseId of ['reconcile-team-call', 'reconcile-ordinary-call']) { + const agent = spans.find(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === toolUseId); + assert.ok(agent); + assert.equal(agent.attributes[ATTR.WEAVE_ORPHAN_REASON], undefined); + } +}); + +test('one disambiguation drains every ready buffered idle', async (t) => { + const exporter = await initWeaveInMemory(); + exporter.reset(); + const daemon = makeGenaiDaemon(); + const teamOwner = makeTranscript(t, 'batch-team-owner', 'batch-team-owner'); + const blockerOwner = makeTranscript(t, 'batch-blocker-owner', 'batch-blocker-owner'); + for (const [sid, transcript] of [ + ['batch-team-owner', teamOwner], + ['batch-blocker-owner', blockerOwner], + ] as const) { + transcript.append(userEntry('delegate reviews')); + await daemon.routeEvent({ + hook_event_name: 'SessionStart', session_id: sid, + transcript_path: transcript.file, source: 'startup', cwd: '/x', + }); + await daemon.routeEvent({ + hook_event_name: 'UserPromptSubmit', session_id: sid, prompt: 'delegate reviews', + }); + } + + for (let index = 0; index < 9; index++) { + const input = { description: 'team work', prompt: `batch task ${index}`, name: MEMBER }; + await dispatch(daemon, 'batch-team-owner', `batch-team-call-${index}`, input.prompt, input); + } + const blockerInput = { description: 'ordinary work', prompt: 'blocker task', name: MEMBER }; + const blockerId = 'batch-blocker-agent'; + const blockerPath = blockerOwner.subagent( + blockerId, + userEntry('blocker task'), + assistantEntry('batch-blocker-msg', { type: 'text', text: 'ordinary result' }), + ); + await preDispatch(daemon, 'batch-blocker-owner', 'batch-blocker-call', blockerInput); + await daemon.routeEvent({ + hook_event_name: 'SubagentStart', session_id: 'batch-blocker-owner', + agent_id: blockerId, agent_type: MEMBER, + }); + await postDispatch(daemon, 'batch-blocker-owner', 'batch-blocker-call', blockerInput); + + for (let index = 0; index < 9; index++) { + const memberSession = `batch-member-${index}`; + const teammate = makeTranscript(t, memberSession, memberSession); + teammate.append(...teammateEntries( + memberSession, `batch result ${index}`, `batch-team-msg-${index}`, + )); + await daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: memberSession, + transcript_path: teammate.file, team_name: TEAM, teammate_name: MEMBER, + }); + } + assert.equal(exporter.getFinishedSpans().filter(span => + String(span.attributes[ATTR.RESPONSE_ID]).startsWith('batch-team-msg-')).length, 0); + + await daemon.routeEvent({ + hook_event_name: 'SubagentStop', session_id: 'batch-blocker-owner', + agent_id: blockerId, agent_type: MEMBER, agent_transcript_path: blockerPath, + }); + await daemon.routeEvent({ + hook_event_name: 'SessionEnd', session_id: 'batch-team-owner', reason: 'clear', + }); + await daemon.routeEvent({ + hook_event_name: 'SessionEnd', session_id: 'batch-blocker-owner', reason: 'clear', + }); + await flushWeave(); + + const spans = exporter.getFinishedSpans(); + assert.equal(spans.filter(span => + String(span.attributes[ATTR.RESPONSE_ID]).startsWith('batch-team-msg-')).length, 9); + assert.equal(spans.filter(span => + String(span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID]) + .startsWith('batch-team-call-')).length, 9); +}); diff --git a/tests/agent-teams-session-end.test.ts b/tests/agent-teams-session-end.test.ts new file mode 100644 index 0000000..23fef3c --- /dev/null +++ b/tests/agent-teams-session-end.test.ts @@ -0,0 +1,112 @@ +// SPDX-FileCopyrightText: 2026 CoreWeave, Inc. +// SPDX-License-Identifier: MIT +// SPDX-PackageName: weave-claude-code + +import { + ATTR, MEMBER, TEAM, assert, assistantEntry, coordinator, dispatch, flushWeave, + isTeammateTurn, makeTranscript, postDispatch, preDispatch, spanParentId, + teammateEntries, test, userEntry, +} from './agent-team-test-helpers.ts'; + +test('SessionEnd retains an exact team call despite optional metadata overflow', async (t) => { + const promptId = 'session-end-prompt'; + const { exporter, daemon, sid, transcript } = await coordinator(t, 'session-end', promptId); + await dispatch(daemon, sid, 'deferred-team-call', 'inspect'); + for (let i = 0; i < 513; i++) { + transcript.subagent(`unrelated-${i}`, { + type: 'agent-setting', agentSetting: MEMBER, sessionId: `unrelated-${i}`, + }); + } + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await daemon.routeEvent({ + hook_event_name: 'UserPromptSubmit', session_id: sid, + transcript_path: transcript.file, prompt: 'delegate reviews', prompt_id: promptId, + }); + + const internals = daemon as unknown as { hasInFlightWork(): boolean }; + assert.equal(internals.hasInFlightWork(), true, 'deferred team work pins inactivity'); + assert.equal(exporter.getFinishedSpans().some(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'deferred-team-call'), false); + + const teammate = makeTranscript(t, 'fallback-teammate', 'team-fallback'); + teammate.append(...teammateEntries('fallback-teammate', 'fallback result', 'fallback-msg')); + await daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: 'fallback-teammate', + transcript_path: teammate.file, team_name: TEAM, teammate_name: MEMBER, + }); + await flushWeave(); + + const spans = exporter.getFinishedSpans(); + const agent = spans.find(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'deferred-team-call'); + const root = spans.find(span => span.attributes[ATTR.AGENT_NAME] === 'claude-code'); + assert.ok(agent && root); + assert.equal( + agent.attributes[ATTR.OUTPUT_MESSAGES], + JSON.stringify([{ role: 'assistant', content: 'fallback result' }]), + ); + assert.equal(agent.attributes[ATTR.WEAVE_ORPHAN_REASON], undefined); + assert.equal(spanParentId(agent), root.spanContext().spanId); + assert.ok(spans.some(span => span.attributes[ATTR.RESPONSE_ID] === 'fallback-msg')); + assert.equal(internals.hasInFlightWork(), false, 'a duplicate prompt does not cancel SessionEnd'); +}); + +test('generic implicit team work survives SessionEnd until TeammateIdle', async (t) => { + const { exporter, daemon, sid } = await coordinator(t, 'implicit-session-end'); + const input = { description: 'review code', prompt: 'late review', name: MEMBER }; + await dispatch(daemon, sid, 'implicit-late-call', 'late review', input); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + assert.equal(exporter.getFinishedSpans().some(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'implicit-late-call'), false); + + const teammate = makeTranscript(t, 'implicit-late-member', 'implicit-late-member'); + teammate.append(...teammateEntries( + 'implicit-late-member', 'late result', 'implicit-late-msg', 'general-purpose', + )); + await daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: 'implicit-late-member', + transcript_path: teammate.file, team_name: TEAM, teammate_name: MEMBER, + }); + await flushWeave(); + + const agent = exporter.getFinishedSpans().find(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'implicit-late-call'); + assert.ok(agent); + assert.equal(agent.attributes[ATTR.WEAVE_ORPHAN_REASON], undefined); +}); + +test('ambiguous same-session markers fail closed and shutdown orphans all team markers', async (t) => { + const { exporter, daemon, sid, transcript } = await coordinator(t, 'ambiguous'); + for (const agentId of ['idle-a', 'idle-b']) { + transcript.subagent( + agentId, + { type: 'agent-setting', agentSetting: MEMBER, sessionId: sid }, + { type: 'user', teamName: 'local-team', message: { role: 'user', content: 'local task' } }, + ); + await daemon.routeEvent({ + hook_event_name: 'SubagentStart', session_id: sid, + agent_id: agentId, agent_type: MEMBER, + }); + } + await daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: sid, + transcript_path: transcript.file, team_name: 'local-team', teammate_name: MEMBER, + }); + await dispatch(daemon, sid, 'shutdown-dispatch', 'remote task'); + await daemon.drain('SIGTERM'); + await flushWeave(); + + const spans = exporter.getFinishedSpans(); + assert.equal(spans.some(span => isTeammateTurn(span)), false); + const root = spans.find(span => span.attributes[ATTR.AGENT_NAME] === 'claude-code'); + const agents = spans.filter(span => + ['idle-a', 'idle-b'].includes(String(span.attributes[ATTR.AGENT_ID])) + || span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'shutdown-dispatch'); + assert.ok(root); + assert.equal(agents.length, 3); + for (const agent of agents) { + assert.equal(agent.attributes[ATTR.WEAVE_ORPHAN_REASON], 'daemon_shutdown'); + assert.equal(spanParentId(agent), root.spanContext().spanId); + assert.deepEqual(agent.endTime, root.endTime); + } +}); diff --git a/tests/agent-teams-transcript-progress.test.ts b/tests/agent-teams-transcript-progress.test.ts new file mode 100644 index 0000000..2496b5d --- /dev/null +++ b/tests/agent-teams-transcript-progress.test.ts @@ -0,0 +1,399 @@ +// SPDX-FileCopyrightText: 2026 CoreWeave, Inc. +// SPDX-License-Identifier: MIT +// SPDX-PackageName: weave-claude-code + +import { + ATTR, MEMBER, TEAM, assert, assistantEntry, coordinator, dispatch, flushWeave, fs, + makeTranscript, postDispatch, preDispatch, teammateEntries, test, userEntry, +} from './agent-team-test-helpers.ts'; +import { + MAX_TEAM_TRANSCRIPT_BYTES, + readNewTurns, + snapshot, +} from '../src/teamTranscripts.ts'; + +test('oversized teammate snapshots and partial-line extensions fail closed', async (t) => { + const oversized = makeTranscript(t, 'oversized-team', 'oversized-team'); + fs.writeFileSync(oversized.file, Buffer.alloc(MAX_TEAM_TRANSCRIPT_BYTES + 1, 0x20)); + assert.equal(snapshot(oversized.file), undefined); + assert.equal(await readNewTurns(oversized.file), undefined); + + const partial = makeTranscript(t, 'partial-team', 'partial-team'); + fs.writeFileSync(partial.file, '{"type":"assistant"'); + const receipt = snapshot(partial.file); + assert.ok(receipt); + fs.appendFileSync( + partial.file, + Buffer.alloc(MAX_TEAM_TRANSCRIPT_BYTES - receipt.size + 1, 0x20), + ); + fs.appendFileSync(partial.file, '\n'); + assert.equal(await readNewTurns(partial.file, undefined, receipt), undefined); +}); + +test('same-inode transcript regression cannot reset and replay progress', async (t) => { + const transcript = makeTranscript(t, 'regressed-team', 'regressed-team'); + transcript.append(...teammateEntries( + 'regressed-team', 'first result', 'regressed-team-msg', + )); + const firstSnapshot = snapshot(transcript.file); + assert.ok(firstSnapshot); + const first = await readNewTurns(transcript.file, undefined, firstSnapshot); + assert.ok(first); + + fs.truncateSync(transcript.file, 0); + transcript.append( + { type: 'agent-setting', agentSetting: MEMBER, sessionId: 'regressed-team' }, + { + type: 'user', + teamName: TEAM, + message: { role: 'user', content: 'replacement task' }, + }, + assistantEntry('rewritten-team-msg', [ + { type: 'text', text: 'replacement prefix' }, + { type: 'text', text: 'replacement tail' }, + ]), + ); + const regressedSnapshot = snapshot(transcript.file); + assert.ok(regressedSnapshot); + assert.equal(regressedSnapshot.inode, firstSnapshot.inode); + assert.equal( + await readNewTurns(transcript.file, first[1], regressedSnapshot), + undefined, + ); + + const replacement = `${transcript.file}.replacement`; + fs.writeFileSync( + replacement, + teammateEntries( + 'regressed-team', + 'replacement result', + 'replacement-team-msg', + ).map(entry => JSON.stringify(entry)).join('\n') + '\n', + ); + fs.renameSync(replacement, transcript.file); + const replacedSnapshot = snapshot(transcript.file); + assert.ok(replacedSnapshot); + assert.notEqual(replacedSnapshot.inode, firstSnapshot.inode); + assert.equal( + await readNewTurns(transcript.file, first[1], replacedSnapshot), + undefined, + ); +}); + +test('provider progress advances within one turn and ignores non-provider growth', async (t) => { + const { exporter, daemon, sid } = await coordinator(t, 'response-progress'); + await dispatch(daemon, sid, 'response-progress-1', 'first'); + await dispatch(daemon, sid, 'response-progress-2', 'second'); + const teammate = makeTranscript(t, 'response-progress-member', 'response-progress-member'); + teammate.append(...teammateEntries( + 'response-progress-member', 'first result', 'response-progress-msg-1', + )); + const idle = { + hook_event_name: 'TeammateIdle', session_id: 'response-progress-member', + transcript_path: teammate.file, team_name: TEAM, teammate_name: MEMBER, + }; + await daemon.routeEvent(idle); + teammate.append({ type: 'progress', message: { role: 'system', content: 'still working' } }); + await daemon.routeEvent(idle); + teammate.append(assistantEntry( + 'response-progress-msg-2', + { type: 'text', text: 'second result' }, + )); + await daemon.routeEvent(idle); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await flushWeave(); + + const agents = exporter.getFinishedSpans().filter(span => + String(span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID]) + .startsWith('response-progress-')); + assert.equal(agents.length, 2); + assert.ok(exporter.getFinishedSpans().some(span => + span.attributes[ATTR.RESPONSE_ID] === 'response-progress-msg-2')); +}); + +test('a stale idle with only non-provider growth cannot confirm the next named Agent', async (t) => { + const { exporter, daemon, sid, transcript } = await coordinator(t, 'stale-idle'); + await dispatch(daemon, sid, 'stale-team-call', 'first team task'); + const teammate = makeTranscript(t, 'stale-team-member', 'stale-team-member'); + teammate.append(...teammateEntries( + 'stale-team-member', 'team result', 'stale-team-msg', + )); + const idle = { + hook_event_name: 'TeammateIdle', session_id: 'stale-team-member', + transcript_path: teammate.file, team_name: TEAM, teammate_name: MEMBER, + }; + await daemon.routeEvent(idle); + + const input = { description: 'ordinary work', prompt: 'ordinary next task', name: MEMBER }; + const agentId = 'ordinary-after-stale-idle'; + const subPath = transcript.subagent( + agentId, + userEntry('ordinary next task'), + assistantEntry('ordinary-after-stale-msg', { type: 'text', text: 'ordinary result' }), + ); + await preDispatch(daemon, sid, 'ordinary-after-stale-call', input); + await daemon.routeEvent({ + hook_event_name: 'SubagentStart', session_id: sid, + agent_id: agentId, agent_type: 'general-purpose', + }); + await postDispatch(daemon, sid, 'ordinary-after-stale-call', input); + teammate.append({ type: 'progress', message: { role: 'system', content: 'bookkeeping' } }); + await daemon.routeEvent(idle); + await daemon.routeEvent({ + hook_event_name: 'SubagentStop', session_id: sid, + agent_id: agentId, agent_type: 'general-purpose', agent_transcript_path: subPath, + }); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await flushWeave(); + + const ordinary = exporter.getFinishedSpans().find(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'ordinary-after-stale-call'); + assert.ok(ordinary); + assert.equal(ordinary.attributes[ATTR.WEAVE_ORPHAN_REASON], undefined); + assert.ok(exporter.getFinishedSpans().some(span => + span.attributes[ATTR.RESPONSE_ID] === 'ordinary-after-stale-msg')); +}); + +test('idle history churn cannot evict persistent transcript progress', async (t) => { + const { exporter, daemon, sid } = await coordinator(t, 'progress-eviction'); + t.after(() => daemon.drain('test cleanup')); + await dispatch(daemon, sid, 'progress-eviction-call-1', 'first'); + const teammate = makeTranscript(t, 'progress-eviction-member', 'progress-eviction-member'); + teammate.append(...teammateEntries( + 'progress-eviction-member', 'first result', 'progress-eviction-msg-1', + )); + const idle = { + hook_event_name: 'TeammateIdle', session_id: 'progress-eviction-member', + transcript_path: teammate.file, team_name: TEAM, teammate_name: MEMBER, + }; + await daemon.routeEvent(idle); + + // Overflow the independent idle-event history without touching this + // persistent transcript. Its provider cursor must remain intact. + for (let i = 0; i < 511; i++) { + await daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: `unmatched-${i}`, + team_name: `unmatched-team-${i}`, teammate_name: `unmatched-member-${i}`, + }); + } + + await dispatch(daemon, sid, 'progress-eviction-call-2', 'second'); + teammate.append({ type: 'progress', message: { role: 'system', content: 'no provider output' } }); + await daemon.routeEvent(idle); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await flushWeave(); + + assert.equal(exporter.getFinishedSpans().some(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'progress-eviction-call-2'), false); +}); + +test('file aliases cannot replay persistent transcript output', async (t) => { + const { exporter, daemon, sid } = await coordinator(t, 'progress-alias'); + t.after(() => daemon.drain('test cleanup')); + await dispatch(daemon, sid, 'progress-alias-call-1', 'first'); + const teammate = makeTranscript(t, 'progress-alias-member', 'progress-alias-member'); + teammate.append(...teammateEntries( + 'progress-alias-member', 'first result', 'progress-alias-msg', + )); + const idle = { + hook_event_name: 'TeammateIdle', session_id: 'progress-alias-member', + transcript_path: teammate.file, team_name: TEAM, teammate_name: MEMBER, + }; + await daemon.routeEvent(idle); + await dispatch(daemon, sid, 'progress-alias-call-2', 'second'); + const alias = teammate.file.replace(/\.jsonl$/, '-alias.jsonl'); + fs.linkSync(teammate.file, alias); + await daemon.routeEvent({ ...idle, transcript_path: alias }); + await flushWeave(); + + assert.equal(exporter.getFinishedSpans().some(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] === 'progress-alias-call-2'), false); + assert.equal(exporter.getFinishedSpans().filter(span => + span.attributes[ATTR.RESPONSE_ID] === 'progress-alias-msg').length, 1); +}); + +test('a physical transcript cannot be relabeled to replay its output', async (t) => { + const { exporter, daemon, sid } = await coordinator(t, 'progress-relabel'); + t.after(() => daemon.drain('test cleanup')); + const input = { name: MEMBER, prompt: 'first' }; + await dispatch(daemon, sid, 'progress-relabel-call-1', 'first', input); + const teammate = makeTranscript(t, 'progress-relabel-member', 'progress-relabel-member'); + teammate.append(...teammateEntries( + 'progress-relabel-member', 'first result', 'progress-relabel-msg', + 'general-purpose', + )); + const idle = { + hook_event_name: 'TeammateIdle', session_id: 'progress-relabel-member', + transcript_path: teammate.file, team_name: TEAM, teammate_name: MEMBER, + }; + await daemon.routeEvent(idle); + + await dispatch(daemon, sid, 'progress-relabel-call-2', 'second', { + name: MEMBER, + prompt: 'second', + }); + await daemon.routeEvent({ ...idle, team_name: 'renamed-team' }); + await flushWeave(); + + assert.equal(exporter.getFinishedSpans().some(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] + === 'progress-relabel-call-2'), false); + assert.equal(exporter.getFinishedSpans().filter(span => + span.attributes[ATTR.RESPONSE_ID] === 'progress-relabel-msg').length, 1); +}); + +test('persistent progress survives many other teammate transcripts', async (t) => { + const { exporter, daemon, sid } = await coordinator(t, 'progress-capacity'); + t.after(() => daemon.drain('test cleanup')); + await dispatch(daemon, sid, 'progress-capacity-persistent-1', 'first'); + const persistent = makeTranscript(t, 'progress-capacity-member', 'progress-capacity-member'); + persistent.append(...teammateEntries( + 'progress-capacity-member', 'old result', 'progress-capacity-msg', + )); + const persistentIdle = { + hook_event_name: 'TeammateIdle', session_id: 'progress-capacity-member', + transcript_path: persistent.file, team_name: TEAM, teammate_name: MEMBER, + }; + await daemon.routeEvent(persistentIdle); + + for (let i = 0; i < 512; i++) { + const memberSession = `progress-capacity-churn-${i}`; + await dispatch(daemon, sid, `progress-capacity-call-${i}`, `task ${i}`); + const teammate = makeTranscript(t, memberSession, memberSession); + teammate.append(...teammateEntries( + memberSession, `result ${i}`, `progress-capacity-msg-${i}`, + )); + await daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: memberSession, + transcript_path: teammate.file, team_name: TEAM, teammate_name: MEMBER, + }); + } + + await dispatch(daemon, sid, 'progress-capacity-persistent-2', 'second'); + await daemon.routeEvent(persistentIdle); + await flushWeave(); + + assert.equal(exporter.getFinishedSpans().some(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] + === 'progress-capacity-persistent-2'), false); + + persistent.append( + { type: 'user', teamName: TEAM, message: { role: 'user', content: 'new task' } }, + assistantEntry('progress-capacity-resumed-msg', { + type: 'text', text: 'resumed result', + }), + ); + await daemon.routeEvent(persistentIdle); + await flushWeave(); + const resumed = exporter.getFinishedSpans().find(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] + === 'progress-capacity-persistent-2'); + assert.ok(resumed); + assert.equal( + resumed.attributes[ATTR.OUTPUT_MESSAGES], + JSON.stringify([{ role: 'assistant', content: 'resumed result' }]), + ); + + await dispatch(daemon, sid, 'progress-capacity-new-call', 'new task'); + const fresh = makeTranscript(t, 'progress-capacity-new-member', 'progress-capacity-new'); + fresh.append(...teammateEntries( + 'progress-capacity-new-member', 'new result', 'progress-capacity-new-msg', + )); + await daemon.routeEvent({ + hook_event_name: 'TeammateIdle', session_id: 'progress-capacity-new-member', + transcript_path: fresh.file, team_name: TEAM, teammate_name: MEMBER, + }); + await flushWeave(); + const freshAgent = exporter.getFinishedSpans().find(span => + span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID] + === 'progress-capacity-new-call'); + assert.ok(freshAgent); + assert.equal( + freshAgent.attributes[ATTR.OUTPUT_MESSAGES], + JSON.stringify([{ role: 'assistant', content: 'new result' }]), + ); +}); + +test('one persistent teammate session completes twice only after transcript growth', async (t) => { + const { exporter, daemon, sid } = await coordinator(t, 'persistent'); + await dispatch(daemon, sid, 'persistent-1', 'first'); + await dispatch(daemon, sid, 'persistent-2', 'second'); + const teammate = makeTranscript(t, 'persistent-member', 'persistent-member'); + teammate.append(...teammateEntries('persistent-member', 'first result', 'persistent-msg-1')); + const idle = { + hook_event_name: 'TeammateIdle', session_id: 'persistent-member', + transcript_path: teammate.file, team_name: TEAM, teammate_name: MEMBER, + }; + + await daemon.routeEvent(idle); + await daemon.routeEvent(idle); + teammate.append( + { type: 'user', teamName: TEAM, message: { role: 'user', content: 'task: second' } }, + assistantEntry('persistent-msg-2', { type: 'text', text: 'second result' }), + ); + await daemon.routeEvent(idle); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await flushWeave(); + + const agents = exporter.getFinishedSpans().filter(span => + String(span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID]).startsWith('persistent-')); + assert.equal(agents.length, 2); + assert.deepEqual(agents.map(agent => agent.attributes[ATTR.OUTPUT_MESSAGES]).sort(), [ + JSON.stringify([{ role: 'assistant', content: 'first result' }]), + JSON.stringify([{ role: 'assistant', content: 'second result' }]), + ]); +}); + +test('each idle reads only the persistent transcript state it observed', async (t) => { + const { exporter, daemon, sid } = await coordinator(t, 'persistent-receipt-boundary'); + await dispatch(daemon, sid, 'boundary-call-1', 'first'); + await dispatch(daemon, sid, 'boundary-call-2', 'second'); + const teammate = makeTranscript(t, 'boundary-member', 'boundary-member'); + const [setting, user] = teammateEntries( + 'boundary-member', 'first result', 'boundary-msg-1', + ); + teammate.append(setting, user); + const firstResponse = JSON.stringify(assistantEntry( + 'boundary-msg-1', + { type: 'text', text: 'first result' }, + )); + const split = firstResponse.length - 2; + fs.appendFileSync(teammate.file, firstResponse.slice(0, split)); + const idle = { + hook_event_name: 'TeammateIdle', session_id: 'boundary-member', + transcript_path: teammate.file, team_name: TEAM, teammate_name: MEMBER, + }; + + const firstIdle = daemon.routeEvent(idle); + await new Promise(resolve => setTimeout(resolve, 25)); + fs.appendFileSync(teammate.file, [ + `${firstResponse.slice(split)}\n`, + `${JSON.stringify({ + type: 'user', teamName: TEAM, + message: { role: 'user', content: 'task: second' }, + })}\n`, + `${JSON.stringify(assistantEntry( + 'boundary-msg-2', + { type: 'text', text: 'second result' }, + ))}\n`, + ].join('')); + const secondIdle = daemon.routeEvent(idle); + await Promise.all([firstIdle, secondIdle]); + await daemon.routeEvent({ hook_event_name: 'SessionEnd', session_id: sid, reason: 'clear' }); + await flushWeave(); + + const agents = exporter.getFinishedSpans().filter(span => + String(span.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID]) + .startsWith('boundary-call-')); + assert.deepEqual(Object.fromEntries(agents.map(agent => [ + agent.attributes[ATTR.WEAVE_SUBAGENT_SPAWNING_TOOL_CALL_ID], + agent.attributes[ATTR.OUTPUT_MESSAGES], + ])), { + 'boundary-call-1': JSON.stringify([{ role: 'assistant', content: 'first result' }]), + 'boundary-call-2': JSON.stringify([{ role: 'assistant', content: 'second result' }]), + }); + assert.deepEqual(agents.map(agent => agent.attributes[ATTR.WEAVE_ORPHAN_REASON]), [ + undefined, + undefined, + ]); +}); diff --git a/tests/parser.test.ts b/tests/parser.test.ts index a998057..6616973 100644 --- a/tests/parser.test.ts +++ b/tests/parser.test.ts @@ -29,6 +29,24 @@ function parseLines(lines: unknown[]): ParsedSession { } } +test('rejects a transcript that shrinks below its captured boundary', () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'weave-parser-boundary-')); + const file = path.join(dir, 'session.jsonl'); + fs.writeFileSync(file, `${JSON.stringify({ + type: 'user', + message: { role: 'user', content: 'captured' }, + })}\n`); + const capturedBytes = fs.statSync(file).size; + const fd = fs.openSync(file, 'r'); + try { + fs.truncateSync(file, capturedBytes - 1); + assert.equal(parseSessionFd(fd, capturedBytes), null); + } finally { + fs.closeSync(fd); + fs.rmSync(dir, { recursive: true, force: true }); + } +}); + function assistant( id: string | undefined, timestamp: string,