From 00f9c62bbc59ff0971ba5d8dd09ca0a4f4240151 Mon Sep 17 00:00:00 2001 From: Rick Gao Date: Thu, 16 Jul 2026 23:33:44 -0700 Subject: [PATCH 1/3] refactor(daemon): type the hook dispatch against the agent SDK Cast the socket JSON once against the SDK's HookInput and dispatch on the discriminant, so handlers take typed inputs instead of re-casting every field (~40 payload[...] as X casts deleted). Fields the SDK types as required drop their invented fallbacks ('unknown'/'teammate'/'?'); the two payloads that carry more than the SDK declares (PreCompact summary/counts, reconstruction source/model) keep documented raw-record reads. Behavior notes: PostToolUseFailure now trusts the typed `error` field (drops the undocumented tool_response fallback), and InstructionsLoaded drops its missing-file_path branch (the type requires it; a bad path still lands in the unreadable-file catch). Co-Authored-By: Claude Fable 5 --- src/daemon.ts | 200 ++++++++++--------- tests/daemon-shutdown-finalizes-turn.test.ts | 6 +- tests/interleave-handlers.test.ts | 10 +- 3 files changed, 114 insertions(+), 102 deletions(-) diff --git a/src/daemon.ts b/src/daemon.ts index 86f3f8b..cb584d5 100644 --- a/src/daemon.ts +++ b/src/daemon.ts @@ -21,6 +21,22 @@ import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node'; import { BatchSpanProcessor } from '@opentelemetry/sdk-trace-base'; import { resourceFromAttributes } from '@opentelemetry/resources'; import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-proto'; +import type { + HookInput, + SessionStartHookInput, + InstructionsLoadedHookInput, + UserPromptSubmitHookInput, + PreToolUseHookInput, + PostToolUseHookInput, + PostToolUseFailureHookInput, + PermissionRequestHookInput, + SubagentStartHookInput, + SubagentStopHookInput, + TeammateIdleHookInput, + PreCompactHookInput, + StopHookInput, + SessionEndHookInput, +} from '@anthropic-ai/claude-agent-sdk'; import { loadSettings, VERSION } from './setup.js'; import { resolveDaemonConfig, daemonConfigFingerprint, missingConfig } from './config.js'; import { appendToLog, deepEqual } from './utils.js'; @@ -769,16 +785,16 @@ export class GlobalDaemon { // ── event routing ───────────────────────────────────────────────────────── private async routeEvent(payload: HookPayload): Promise { - const eventName = payload['hook_event_name'] as string | undefined; - const sessionId = payload['session_id'] as string | undefined; - const agentId = payload['agent_id'] as string | undefined; - + // Trust the raw hook JSON against the SDK's schema once here so dispatch + // and handlers work with typed, discriminated inputs. + const input = payload as HookInput; + const sessionId = input.session_id; if (!sessionId) { this.log('ERROR', 'Missing session_id in payload'); return; } - this.log('INFO', `${eventName ?? 'unknown'} session=${sessionId}${agentId ? ` agent=${agentId}` : ''}`); + this.log('INFO', `${input.hook_event_name} session=${sessionId}${input.agent_id ? ` agent=${input.agent_id}` : ''}`); // Activate the session's integration baggage for the whole event so every // span created while handling it inherits the integration identity (copied @@ -790,75 +806,69 @@ export class GlobalDaemon { ? propagation.setBaggage(otelContext.active(), session.integrationBaggage) : otelContext.active(); await otelContext.with(eventContext, () => - this.dispatchEvent(eventName, sessionId, agentId, payload), + this.dispatchEvent(input, sessionId), ); } - /** Run the handler for a single hook event. Split out from `routeEvent` so - * the latter can run it inside the session's baggage context. */ - private async dispatchEvent( - eventName: string | undefined, - sessionId: string, - agentId: string | undefined, - payload: HookPayload, - ): Promise { + /** Narrow `input` via the discriminant and run its handler (inside the + * session's baggage context installed by `routeEvent`). */ + private async dispatchEvent(input: HookInput, sessionId: string): Promise { try { - switch (eventName) { + switch (input.hook_event_name) { case 'SessionStart': - await this.handleSessionStart(sessionId, payload); + await this.handleSessionStart(sessionId, input); break; case 'InstructionsLoaded': - // Reads the instruction file synchronously off the hook's file_path; - // no async work to await. - this.handleInstructionsLoaded(sessionId, payload); + // Synchronous: reads the instruction file inline; nothing to await. + this.handleInstructionsLoaded(sessionId, input); break; case 'UserPromptSubmit': - await this.handleUserPromptSubmit(sessionId, payload); + await this.handleUserPromptSubmit(sessionId, input); break; case 'PreToolUse': - await this.handlePreToolUse(sessionId, agentId, payload); + await this.handlePreToolUse(sessionId, input); break; case 'PermissionRequest': - await this.handlePermissionRequest(sessionId, payload); + await this.handlePermissionRequest(sessionId, input); break; case 'PostToolUse': - await this.handlePostToolUse(sessionId, payload); + await this.handlePostToolUse(sessionId, input); break; case 'PostToolUseFailure': - await this.handlePostToolUseFailure(sessionId, payload); + await this.handlePostToolUseFailure(sessionId, input); break; case 'SubagentStart': - await this.handleSubagentStart(sessionId, payload); + await this.handleSubagentStart(sessionId, input); break; case 'SubagentStop': - await this.handleSubagentStop(sessionId, payload); + await this.handleSubagentStop(sessionId, input); break; case 'TeammateIdle': - await this.handleTeammateIdle(sessionId, payload); + await this.handleTeammateIdle(sessionId, input); break; case 'PreCompact': - await this.handlePreCompact(sessionId, payload); + await this.handlePreCompact(sessionId, input); break; case 'Stop': - await this.handleStop(sessionId, payload); + await this.handleStop(sessionId, input); break; case 'SessionEnd': - await this.handleSessionEnd(sessionId, payload); + await this.handleSessionEnd(sessionId, input); break; default: break; } } catch (err) { - this.log('ERROR', `Error handling ${eventName ?? 'unknown'}: ${err}`); + this.log('ERROR', `Error handling ${input.hook_event_name}: ${err}`); } } // ── event handlers ──────────────────────────────────────────────────────── - private async handleSessionStart(sessionId: string, payload: HookPayload): Promise { + private async handleSessionStart(sessionId: string, input: SessionStartHookInput): Promise { if (this.sessions.has(sessionId)) return; // idempotent - const rawPath = payload['transcript_path'] as string | undefined; + const rawPath = input.transcript_path; if (!rawPath) { this.log('ERROR', `Missing transcript_path for session ${sessionId}`); return; @@ -872,9 +882,9 @@ export class GlobalDaemon { return; } - const source = (payload['source'] as string | undefined) ?? 'unknown'; - const initialRequestModel = payload['model'] as string | undefined; - const cwd = (payload['cwd'] as string | undefined) ?? ''; + const source = input.source; + const initialRequestModel = input.model; + const cwd = input.cwd; const conversationId = await this.resolveConversationId(sessionId, transcript.resolvedPath, source); @@ -984,12 +994,12 @@ export class GlobalDaemon { */ private async getOrReconstructSession( sessionId: string, - payload: HookPayload, + input: HookInput, ): Promise { const existing = this.sessions.get(sessionId); if (existing) return existing; - const rawPath = payload['transcript_path'] as string | undefined; + const rawPath = input.transcript_path; if (!rawPath) return undefined; let transcript: TranscriptFile; @@ -1000,9 +1010,12 @@ export class GlobalDaemon { return undefined; } - const source = (payload['source'] as string | undefined) ?? 'reconstructed'; - const cwd = (payload['cwd'] as string | undefined) ?? ''; - const initialRequestModel = payload['model'] as string | undefined; + // source/model aren't on every hook variant (this reconstructs from a + // UserPromptSubmit), so read them best-effort off the raw record. + const raw = input as Record; + const source = (raw['source'] as string | undefined) ?? 'reconstructed'; + const cwd = input.cwd; + const initialRequestModel = raw['model'] as string | undefined; const conversationId = await this.resolveConversationId(sessionId, transcript.resolvedPath, source); // Seed the turn counter from the turns already on disk so numbering @@ -1050,14 +1063,8 @@ export class GlobalDaemon { * reconstructed after a restart starts empty and picks up only files that * (re)load afterward (e.g. load_reason=compact). */ - private handleInstructionsLoaded(sessionId: string, payload: HookPayload): void { - const filePath = payload['file_path'] as string | undefined; - const loadReason = (payload['load_reason'] as string | undefined) ?? 'unknown'; - if (!filePath) { - this.log('DEBUG', `InstructionsLoaded ignored (missing file_path): session=${sessionId}`); - return; - } - + private handleInstructionsLoaded(sessionId: string, input: InstructionsLoadedHookInput): void { + const filePath = input.file_path; let content: string; try { content = fs.readFileSync(filePath, 'utf8'); @@ -1078,7 +1085,7 @@ export class GlobalDaemon { } this.log( 'DEBUG', - `InstructionsLoaded: session=${sessionId} reason=${loadReason} file=${path.basename(filePath)} bytes=${content.length}${session ? '' : ' (buffered)'}`, + `InstructionsLoaded: session=${sessionId} reason=${input.load_reason} file=${path.basename(filePath)} bytes=${content.length}${session ? '' : ' (buffered)'}`, ); } @@ -1092,18 +1099,18 @@ export class GlobalDaemon { this.log('DEBUG', `Drained ${pending.length} buffered instruction file(s) into session ${session.sessionId}`); } - private async handleUserPromptSubmit(sessionId: string, payload: HookPayload): Promise { + private async handleUserPromptSubmit(sessionId: string, input: UserPromptSubmitHookInput): Promise { // Reconstruct the session if this daemon never saw its SessionStart (e.g. it // idled out mid-session and a fresh daemon took over) so the rest of the // session stays traced instead of dropping with "Unknown session". - const session = await this.getOrReconstructSession(sessionId, payload); + const session = await this.getOrReconstructSession(sessionId, input); if (!session) { this.log('ERROR', `Unknown session (no transcript_path to reconstruct): ${sessionId}`); return; } if (!this.tracer) return; - const prompt = (payload['prompt'] as string | undefined) ?? ''; + const prompt = input.prompt; this.log( 'DEBUG', `UserPromptSubmit: session=${sessionId} current_turn_span=${session.currentTurnSpan ? 'open' : 'none'} turn_number=${session.turnNumber} prompt=${promptSnippet(prompt, 120)}`, @@ -1140,15 +1147,17 @@ export class GlobalDaemon { ); } - private async handlePreToolUse(sessionId: string, agentId: string | undefined, payload: HookPayload): Promise { + private async handlePreToolUse(sessionId: string, input: PreToolUseHookInput): Promise { const session = this.sessions.get(sessionId); if (!session || !this.tracer) return; - const toolUseId = payload['tool_use_id'] as string | undefined; - const toolName = payload['tool_name'] as string | undefined; + const agentId = input.agent_id; + const toolUseId = input.tool_use_id; + const toolName = input.tool_name; if (!toolUseId || !toolName) return; - const toolInput = (payload['tool_input'] ?? {}) as Record; + // tool_input is per-tool JSON the SDK types as `unknown`; narrow to index it. + const toolInput = (input.tool_input ?? {}) as Record; // Parent: subagent's invoke_agent span if this PreToolUse comes from inside // a subagent, else the current turn span. @@ -1396,11 +1405,11 @@ export class GlobalDaemon { } } - private async handlePermissionRequest(sessionId: string, payload: HookPayload): Promise { + private async handlePermissionRequest(sessionId: string, input: PermissionRequestHookInput): Promise { const session = this.sessions.get(sessionId); if (!session) return; - const toolName = payload['tool_name'] as string | undefined; + const toolName = input.tool_name; if (!toolName) return; // Correlate to a pending tool call by tool_name + tool_input. Record the @@ -1408,7 +1417,7 @@ export class GlobalDaemon { // once we know whether it was approved. let pending: PendingToolCall | undefined; for (const call of session.pendingToolCalls.values()) { - if (call.toolName === toolName && !call.permissionRequested && deepEqual(call.toolInput, payload['tool_input'])) { + if (call.toolName === toolName && !call.permissionRequested && deepEqual(call.toolInput, input.tool_input)) { pending = call; break; } @@ -1420,18 +1429,18 @@ export class GlobalDaemon { pending.permissionRequested = true; addPermissionRequestEvent(pending.span, { - suggestions: payload['permission_suggestions'], + suggestions: input.permission_suggestions, timestamp: new Date(), }); this.log('DEBUG', `Permission request recorded for ${toolName}`); } - private async handlePostToolUse(sessionId: string, payload: HookPayload): Promise { + private async handlePostToolUse(sessionId: string, input: PostToolUseHookInput): Promise { const session = this.sessions.get(sessionId); if (!session) return; - const toolUseId = payload['tool_use_id'] as string | undefined; + const toolUseId = input.tool_use_id; if (!toolUseId) return; // Subagent dispatch: the matching span is the subagent's invoke_agent @@ -1445,7 +1454,7 @@ export class GlobalDaemon { // end empty before the teammate works. The team map owns it now. session.subagents.remove(subagentTracker); } else { - this.closeSubagentInvokeAgentSpan(subagentTracker, payload['tool_response'], /*failure*/ false); + this.closeSubagentInvokeAgentSpan(subagentTracker, input.tool_response, /*failure*/ false); session.subagents.remove(subagentTracker); } session.totalToolCalls += 1; @@ -1459,7 +1468,7 @@ export class GlobalDaemon { resolvePermissionIfPending(pending, true); - pending.span.setAttribute(ATTR.TOOL_CALL_RESULT, jsonStr(payload['tool_response'])); + pending.span.setAttribute(ATTR.TOOL_CALL_RESULT, jsonStr(input.tool_response)); pending.span.end(); session.pendingToolCalls.delete(toolUseId); @@ -1468,14 +1477,14 @@ export class GlobalDaemon { session.toolCounts[pending.toolName] = (session.toolCounts[pending.toolName] ?? 0) + 1; } - private async handlePostToolUseFailure(sessionId: string, payload: HookPayload): Promise { + private async handlePostToolUseFailure(sessionId: string, input: PostToolUseFailureHookInput): Promise { const session = this.sessions.get(sessionId); if (!session) return; - const toolUseId = payload['tool_use_id'] as string | undefined; + const toolUseId = input.tool_use_id; if (!toolUseId) return; - const error = payload['error'] ?? payload['tool_response']; + const error = input.error; // Subagent dispatch failed (rare). Close the invoke_agent span with ERROR // status; subagent chat spans, if any reached SubagentStop, are already @@ -1546,14 +1555,14 @@ export class GlobalDaemon { tracker.ended = true; } - private async handleSubagentStart(sessionId: string, payload: HookPayload): Promise { + private async handleSubagentStart(sessionId: string, input: SubagentStartHookInput): Promise { const session = this.sessions.get(sessionId); if (!session || !this.tracer) return; - const agentId = payload['agent_id'] as string | undefined; + const agentId = input.agent_id; if (!agentId) return; - const agentType = (payload['agent_type'] as string | undefined) ?? 'unknown'; + const agentType = input.agent_type; // Content-based deterministic correlation: SubagentStart carries no // pointer back to the parent's `tool_use_id`, so we read the subagent's @@ -1643,13 +1652,11 @@ export class GlobalDaemon { private recoverSubagentTracker( session: SessionState, agentId: string, - payload: HookPayload, + agentType: string, ): SubagentTracker | undefined { if (!this.tracer) return undefined; const turnSpan = this.getOrReconstructTurnSpan(session); if (!turnSpan) return undefined; - // Type comes from the SubagentStop payload's agent_type; 'unknown' if absent. - const agentType = (payload['agent_type'] as string | undefined) ?? 'unknown'; const invokeAgentSpan = startInvokeAgentSpan(this.tracer, turnSpan, { agentType, conversationId: session.conversationId, @@ -1673,17 +1680,17 @@ export class GlobalDaemon { return tracker; } - private async handleSubagentStop(sessionId: string, payload: HookPayload): Promise { + private async handleSubagentStop(sessionId: string, input: SubagentStopHookInput): Promise { // Reconstruct the session if a restart lost it (see getOrReconstructSession). - const session = await this.getOrReconstructSession(sessionId, payload); + const session = await this.getOrReconstructSession(sessionId, input); if (!session || !this.tracer) return; - const agentId = payload['agent_id'] as string | undefined; + const agentId = input.agent_id; if (!agentId) return; // No tracker: the subagent started under a since-restarted daemon. Recover it. const tracker = session.subagents.byAgentId(agentId) - ?? this.recoverSubagentTracker(session, agentId, payload); + ?? this.recoverSubagentTracker(session, agentId, input.agent_type); if (!tracker) { this.log('ERROR', `SubagentStop: no tracker for agentId=${agentId} and none recoverable`); return; @@ -1696,7 +1703,7 @@ export class GlobalDaemon { // Fall back to the stored or agentId-derived path when the payload omits it. const agentTranscriptPath = - (payload['agent_transcript_path'] as string | undefined) ?? + input.agent_transcript_path ?? tracker.transcriptPath ?? computeSubagentTranscriptPath(session.transcript.resolvedPath, agentId); let model: string | undefined; @@ -1762,7 +1769,7 @@ export class GlobalDaemon { } } - private async handleTeammateIdle(sessionId: string, payload: HookPayload): Promise { + private async handleTeammateIdle(sessionId: string, input: TeammateIdleHookInput): Promise { if (!this.tracer) return; // FAIL-OPEN on a missing session: in the agent-teams model this hook fires // under the TEAMMATE's session_id, which may not be registered with this @@ -1784,8 +1791,8 @@ export class GlobalDaemon { // // Note: CC docs incorrectly listed agent_id / agent_type — those fields do // not appear in practice. - const agentType = (payload['teammate_name'] as string | undefined) ?? 'teammate'; - const teamName = (payload['team_name'] as string | undefined) ?? '?'; + const agentType = input.teammate_name; + const teamName = input.team_name; // ── Cross-session team path (agent-teams / TeamCreate model) ───────── // The coordinator's PreToolUse(Agent, team_name) registered the invoke_agent @@ -1803,7 +1810,7 @@ export class GlobalDaemon { return; } member.emitted = true; - const idleTranscript = session?.transcript.resolvedPath ?? (payload['transcript_path'] as string | undefined); + const idleTranscript = session?.transcript.resolvedPath ?? input.transcript_path; const teammateTranscriptPath = this.resolveTeammateTranscript(member.coordinatorTranscriptPath, agentType, idleTranscript); this.emitTeammateTranscript(member.invokeAgentSpan, member.conversationId, teammateTranscriptPath); // Remove the consumed entry; drop the key once its queue drains. @@ -1957,14 +1964,20 @@ export class GlobalDaemon { invokeAgentSpan.end(); } - private async handlePreCompact(sessionId: string, payload: HookPayload): Promise { + private async handlePreCompact(sessionId: string, input: PreCompactHookInput): Promise { const session = this.sessions.get(sessionId); if (!session) return; + // Live CC payloads carry a summary + item counts the SDK type doesn't + // declare — read them off the raw record. + const raw = input as Record; + const summary = raw['summary'] ?? raw['compaction_summary']; + const itemsBefore = raw['items_before']; + const itemsAfter = raw['items_after']; const attrs: CompactionAttrs = { - summary: (payload['summary'] as string | undefined) ?? (payload['compaction_summary'] as string | undefined), - itemsBefore: typeof payload['items_before'] === 'number' ? (payload['items_before'] as number) : undefined, - itemsAfter: typeof payload['items_after'] === 'number' ? (payload['items_after'] as number) : undefined, + summary: typeof summary === 'string' ? summary : undefined, + itemsBefore: typeof itemsBefore === 'number' ? itemsBefore : undefined, + itemsAfter: typeof itemsAfter === 'number' ? itemsAfter : undefined, }; if (session.currentTurnSpan) { @@ -1977,14 +1990,13 @@ export class GlobalDaemon { } } - private async handleStop(sessionId: string, payload: HookPayload): Promise { + private async handleStop(sessionId: string, input: StopHookInput): Promise { const session = this.sessions.get(sessionId); if (!session?.currentTurnSpan || !this.tracer) return; // Pass last_assistant_message so the retry waits for the synthesis to // flush — otherwise the final chat span drops when the read races the writer. - const rawFinalMessage = payload['last_assistant_message']; - const finalAssistantMessage = typeof rawFinalMessage === 'string' ? rawFinalMessage : undefined; + const finalAssistantMessage = input.last_assistant_message; const parsedSession = await this.parseSessionFileWithRetry( session.transcript, finalAssistantMessage, @@ -1994,7 +2006,7 @@ export class GlobalDaemon { const transcriptTurns = parsedSession?.turns.length ?? 0; this.log( 'DEBUG', - `Stop: session=${sessionId} trace_id=${session.currentTurnSpan.spanContext().traceId} transcript_path=${session.transcript.resolvedPath} transcript_turns=${transcriptTurns} parsed_model=${model ?? 'unknown'} last_assistant_message_present=${Boolean(payload['last_assistant_message'])}`, + `Stop: session=${sessionId} trace_id=${session.currentTurnSpan.spanContext().traceId} transcript_path=${session.transcript.resolvedPath} transcript_turns=${transcriptTurns} parsed_model=${model ?? 'unknown'} last_assistant_message_present=${Boolean(input.last_assistant_message)}`, ); // Finalize the chat-span state machine for this turn. @@ -2018,7 +2030,7 @@ export class GlobalDaemon { } const parsedTexts = currentTurn?.textBlocks() ?? []; - const lastMessage = (payload['last_assistant_message'] as string | undefined) ?? ''; + const lastMessage = input.last_assistant_message ?? ''; const assistantMessages = parsedTexts.length > 0 ? parsedTexts : (lastMessage ? [lastMessage] : []); if (assistantMessages.length) { @@ -2045,7 +2057,7 @@ export class GlobalDaemon { this.log('INFO', `Finished turn ${session.turnNumber} (${session.turnToolCalls} tools)`); } - private async handleSessionEnd(sessionId: string, payload: HookPayload): Promise { + private async handleSessionEnd(sessionId: string, input: SessionEndHookInput): Promise { // Discard any never-drained instruction buffer (e.g. a session that emitted // InstructionsLoaded but never SessionStart) so the map can't leak. this.pendingInstructions.delete(sessionId); @@ -2054,7 +2066,7 @@ export class GlobalDaemon { this.log( 'DEBUG', - `SessionEnd: session=${sessionId} reason=${(payload['reason'] as string | undefined) ?? 'unknown'} transcript_path=${session.transcript.resolvedPath} turns=${session.turnNumber} total_tools=${session.totalToolCalls} pending_tools=${session.pendingToolCalls.size} open_subagents=${session.subagents.size()}`, + `SessionEnd: session=${sessionId} reason=${input.reason} transcript_path=${session.transcript.resolvedPath} turns=${session.turnNumber} total_tools=${session.totalToolCalls} pending_tools=${session.pendingToolCalls.size} open_subagents=${session.subagents.size()}`, ); this.finalizeSession(session, 'session_ended'); diff --git a/tests/daemon-shutdown-finalizes-turn.test.ts b/tests/daemon-shutdown-finalizes-turn.test.ts index bd2032f..44f48ea 100644 --- a/tests/daemon-shutdown-finalizes-turn.test.ts +++ b/tests/daemon-shutdown-finalizes-turn.test.ts @@ -56,7 +56,7 @@ function makeTranscript(sessionId: string): { file: string; append: (line: unkno interface Harness { handleSessionStart(s: string, p: Record): Promise; handleUserPromptSubmit(s: string, p: Record): Promise; - handlePreToolUse(s: string, a: string | undefined, p: Record): Promise; + handlePreToolUse(s: string, p: Record): Promise; handlePostToolUse(s: string, p: Record): Promise; handleSessionEnd(s: string, p: Record): Promise; drain(reason: string): Promise; @@ -77,7 +77,7 @@ async function openTurnWithOneCompletedTool(d: Harness, sid: string, append: (l: await d.handleUserPromptSubmit(sid, { prompt: 'do the thing' }); append(aLine('msgA', '2026-01-01T00:00:02.000Z', { type: 'text', text: 'reading' })); append(aLine('msgA', '2026-01-01T00:00:03.000Z', { type: 'tool_use', id: 'tool_1', name: 'Read', input: {} }, 'tool_use')); - await d.handlePreToolUse(sid, undefined, { tool_use_id: 'tool_1', tool_name: 'Read', tool_input: { file_path: '/foo' } }); + await d.handlePreToolUse(sid, { tool_use_id: 'tool_1', tool_name: 'Read', tool_input: { file_path: '/foo' } }); await d.handlePostToolUse(sid, { tool_use_id: 'tool_1', tool_response: 'ok' }); } @@ -124,7 +124,7 @@ test('daemon shutdown ends an open subagent invoke_agent span under the same tra // Agent tool with subagent_type opens a nested invoke_agent span that a // mid-flight shutdown would otherwise leave open. append(aLine('msgA', '2026-01-01T00:00:02.000Z', { type: 'tool_use', id: 'agent_1', name: 'Agent', input: { subagent_type: 'code-reviewer', prompt: 'review' } }, 'tool_use')); - await d.handlePreToolUse(sid, undefined, { tool_use_id: 'agent_1', tool_name: 'Agent', tool_input: { subagent_type: 'code-reviewer', prompt: 'review' } }); + await d.handlePreToolUse(sid, { tool_use_id: 'agent_1', tool_name: 'Agent', tool_input: { subagent_type: 'code-reviewer', prompt: 'review' } }); await d.drain('SIGTERM'); await provider.forceFlush(); diff --git a/tests/interleave-handlers.test.ts b/tests/interleave-handlers.test.ts index f58276e..9b93f0b 100644 --- a/tests/interleave-handlers.test.ts +++ b/tests/interleave-handlers.test.ts @@ -65,7 +65,7 @@ function makeTranscript(sessionId: string): { file: string; append: (line: unkno interface Handlers { handleSessionStart(s: string, p: Record): Promise; handleUserPromptSubmit(s: string, p: Record): Promise; - handlePreToolUse(s: string, a: string | undefined, p: Record): Promise; + handlePreToolUse(s: string, p: Record): Promise; handlePostToolUse(s: string, p: Record): Promise; handleStop(s: string, p: Record): Promise; handleSessionEnd(s: string, p: Record): Promise; @@ -103,7 +103,7 @@ test('handlers: PreToolUse opens the chat span, Stop finalizes; text + tool inte // before the tool's PreToolUse fires. append(aLine('msgA', '2026-01-01T00:00:02.000Z', { type: 'text', text: 'first I will edit' })); append(aLine('msgA', '2026-01-01T00:00:03.000Z', { type: 'tool_use', id: 'tool_1', name: 'Edit', input: {} }, 'tool_use')); - await d.handlePreToolUse(sid, undefined, { tool_use_id: 'tool_1', tool_name: 'Edit', tool_input: { file_path: '/foo.ts' } }); + await d.handlePreToolUse(sid, { tool_use_id: 'tool_1', tool_name: 'Edit', tool_input: { file_path: '/foo.ts' } }); await d.handlePostToolUse(sid, { tool_use_id: 'tool_1', tool_response: 'ok' }); // msgB: text-only (no tool_use → no PreToolUse; back-filled at Stop). @@ -145,14 +145,14 @@ test('handlers: a new response transitions and finalizes the previous chat span' append(aLine('msgA', '2026-01-01T00:00:02.000Z', { type: 'text', text: 'editing A' })); append(aLine('msgA', '2026-01-01T00:00:03.000Z', { type: 'tool_use', id: 'tool_A', name: 'Edit', input: {} }, 'tool_use')); - await d.handlePreToolUse(sid, undefined, { tool_use_id: 'tool_A', tool_name: 'Edit', tool_input: {} }); + await d.handlePreToolUse(sid, { tool_use_id: 'tool_A', tool_name: 'Edit', tool_input: {} }); await d.handlePostToolUse(sid, { tool_use_id: 'tool_A', tool_response: 'ok' }); // Second response with its own tool_use → PreToolUse(tool_B) must finalize // msgA's chat span (transition) before opening msgB's. append(aLine('msgB', '2026-01-01T00:00:05.000Z', { type: 'text', text: 'editing B' })); append(aLine('msgB', '2026-01-01T00:00:06.000Z', { type: 'tool_use', id: 'tool_B', name: 'Edit', input: {} }, 'tool_use')); - await d.handlePreToolUse(sid, undefined, { tool_use_id: 'tool_B', tool_name: 'Edit', tool_input: {} }); + await d.handlePreToolUse(sid, { tool_use_id: 'tool_B', tool_name: 'Edit', tool_input: {} }); await d.handlePostToolUse(sid, { tool_use_id: 'tool_B', tool_response: 'ok' }); await d.handleStop(sid, {}); @@ -185,7 +185,7 @@ test('handlers: SessionEnd finalizes a still-open chat span with its text + usag append(aLine('msgA', '2026-01-01T00:00:02.000Z', { type: 'text', text: 'first I will edit' })); append(aLine('msgA', '2026-01-01T00:00:03.000Z', { type: 'tool_use', id: 'tool_1', name: 'Edit', input: {} }, 'tool_use')); - await d.handlePreToolUse(sid, undefined, { tool_use_id: 'tool_1', tool_name: 'Edit', tool_input: {} }); + await d.handlePreToolUse(sid, { tool_use_id: 'tool_1', tool_name: 'Edit', tool_input: {} }); // No Stop — session ends mid-turn with the chat span still open. await d.handleSessionEnd(sid, { reason: 'clear' }); From b701c58925b0a21175126a4749732a2bbf506d53 Mon Sep 17 00:00:00 2001 From: Rick Gao Date: Thu, 16 Jul 2026 23:35:10 -0700 Subject: [PATCH 2/3] refactor(daemon): move session helper fns to sessionState.ts Verbatim move of the session-scoped pure helpers (hashPrompt, computeSubagentTranscriptPath, extractUserMessageContent, lastAssistantTextEndsWith, readSubagentFirstLineWithRetry) out of daemon.ts. No behavior change; shrinks daemon.ts ahead of the SDK migration so its diff stays on the pipeline. Co-Authored-By: Claude Fable 5 --- src/daemon.ts | 76 ++++--------------------------------------- src/sessionState.ts | 79 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 86 insertions(+), 69 deletions(-) create mode 100644 src/sessionState.ts diff --git a/src/daemon.ts b/src/daemon.ts index cb584d5..4721226 100644 --- a/src/daemon.ts +++ b/src/daemon.ts @@ -5,7 +5,6 @@ import * as net from 'net'; import * as fs from 'fs'; import * as path from 'path'; -import { createHash } from 'crypto'; import { Baggage, Span, @@ -39,6 +38,13 @@ import type { } from '@anthropic-ai/claude-agent-sdk'; import { loadSettings, VERSION } from './setup.js'; import { resolveDaemonConfig, daemonConfigFingerprint, missingConfig } from './config.js'; +import { + hashPrompt, + computeSubagentTranscriptPath, + extractUserMessageContent, + lastAssistantTextEndsWith, + readSubagentFirstLineWithRetry, +} from './sessionState.js'; import { appendToLog, deepEqual } from './utils.js'; import { parseSessionFd, @@ -123,12 +129,6 @@ function resolvePermissionIfPending(pending: PendingToolCall, approved: boolean) }); } -/** sha256 of the firing prompt — used to correlate an `Agent` PreToolUse with - * the subagent's SubagentStart by matching transcript content. */ -function hashPrompt(prompt: string): string { - return createHash('sha256').update(prompt, 'utf8').digest('hex'); -} - /** Stable identity for an assistant API call within a turn. Anthropic returns * a `message.id` on every response; that's the primary key. When it's * missing (legacy transcripts), fall back to the index, which is stable @@ -174,68 +174,6 @@ function parseIsoOrNow(ts: string | undefined): Date { return parseTimestamp(ts) ?? new Date(); } -/** - * Map a parent transcript path + subagent agent_id to the subagent's transcript - * file. Claude Code writes subagent transcripts as siblings of the parent in a - * `/subagents/` subdirectory: - * parent: /.jsonl - * subagent: //subagents/agent-.jsonl - */ -function computeSubagentTranscriptPath(parentTranscriptPath: string, agentId: string): string { - const projectDir = path.dirname(parentTranscriptPath); - const sessionDirName = path.basename(parentTranscriptPath, '.jsonl'); - return path.join(projectDir, sessionDirName, 'subagents', `agent-${agentId}.jsonl`); -} - -/** Pull the user-message content out of a transcript line. Returns the prompt - * string for `{type: 'user', message: {content: string|Array}}` lines, else - * undefined. Array-form content is joined across text blocks. */ -function extractUserMessageContent(line: Record | undefined): string | undefined { - if (!line || line['type'] !== 'user') return undefined; - const msg = line['message']; - if (!msg || typeof msg !== 'object') return undefined; - const content = (msg as Record)['content']; - if (typeof content === 'string') return content; - if (Array.isArray(content)) { - const parts: string[] = []; - for (const block of content) { - if (block && typeof block === 'object' && (block as Record)['type'] === 'text') { - const t = (block as Record)['text']; - if (typeof t === 'string') parts.push(t); - } - } - return parts.length > 0 ? parts.join('') : undefined; - } - return undefined; -} - -/** True if the last assistant call's joined text ends with `suffix`, - * ignoring trailing whitespace on either side. */ -function lastAssistantTextEndsWith( - result: NonNullable>, - suffix: string, -): boolean { - const call = result.turns.at(-1)?.assistantCalls().at(-1); - // Turn exists but parser saw no assistant calls (writer mid-flush). - if (!call) return false; - return extractAssistantTextBlocks(call.contentBlocks).join('\n').trimEnd().endsWith(suffix); -} - -/** Read the subagent transcript's first line, retrying briefly because Claude - * Code may not have flushed it yet when SubagentStart fires. Total wait - * bounded by the sum of `RETRY_DELAYS_MS`. */ -const SUBAGENT_TRANSCRIPT_RETRY_DELAYS_MS = [0, 50, 100, 150]; -async function readSubagentFirstLineWithRetry( - transcriptPath: string, -): Promise | undefined> { - for (const delay of SUBAGENT_TRANSCRIPT_RETRY_DELAYS_MS) { - if (delay > 0) await new Promise(r => setTimeout(r, delay)); - const line = readFirstTranscriptLine(transcriptPath); - if (line && line['type'] === 'user') return line; - } - return undefined; -} - /** * Tracks a subagent across hook events. Two shapes: * (a) Matched — created at PreToolUse when an Agent tool with subagent_type diff --git a/src/sessionState.ts b/src/sessionState.ts new file mode 100644 index 0000000..220b7db --- /dev/null +++ b/src/sessionState.ts @@ -0,0 +1,79 @@ +// SPDX-FileCopyrightText: 2026 CoreWeave, Inc. +// SPDX-License-Identifier: MIT +// SPDX-PackageName: weave-claude-code + +// Session-scoped helpers shared by the daemon's hook handlers. Moved verbatim +// from daemon.ts; no behavior change. + +import * as path from 'path'; +import { createHash } from 'crypto'; +import { parseSessionFd, extractAssistantTextBlocks } from './parser.js'; +import { readFirstTranscriptLine } from './transcriptFile.js'; + +/** sha256 of the firing prompt — used to correlate an `Agent` PreToolUse with + * the subagent's SubagentStart by matching transcript content. */ +export function hashPrompt(prompt: string): string { + return createHash('sha256').update(prompt, 'utf8').digest('hex'); +} + +/** + * Map a parent transcript path + subagent agent_id to the subagent's transcript + * file. Claude Code writes subagent transcripts as siblings of the parent in a + * `/subagents/` subdirectory: + * parent: /.jsonl + * subagent: //subagents/agent-.jsonl + */ +export function computeSubagentTranscriptPath(parentTranscriptPath: string, agentId: string): string { + const projectDir = path.dirname(parentTranscriptPath); + const sessionDirName = path.basename(parentTranscriptPath, '.jsonl'); + return path.join(projectDir, sessionDirName, 'subagents', `agent-${agentId}.jsonl`); +} + +/** Pull the user-message content out of a transcript line. Returns the prompt + * string for `{type: 'user', message: {content: string|Array}}` lines, else + * undefined. Array-form content is joined across text blocks. */ +export function extractUserMessageContent(line: Record | undefined): string | undefined { + if (!line || line['type'] !== 'user') return undefined; + const msg = line['message']; + if (!msg || typeof msg !== 'object') return undefined; + const content = (msg as Record)['content']; + if (typeof content === 'string') return content; + if (Array.isArray(content)) { + const parts: string[] = []; + for (const block of content) { + if (block && typeof block === 'object' && (block as Record)['type'] === 'text') { + const t = (block as Record)['text']; + if (typeof t === 'string') parts.push(t); + } + } + return parts.length > 0 ? parts.join('') : undefined; + } + return undefined; +} + +/** True if the last assistant call's joined text ends with `suffix`, + * ignoring trailing whitespace on either side. */ +export function lastAssistantTextEndsWith( + result: NonNullable>, + suffix: string, +): boolean { + const call = result.turns.at(-1)?.assistantCalls().at(-1); + // Turn exists but parser saw no assistant calls (writer mid-flush). + if (!call) return false; + return extractAssistantTextBlocks(call.contentBlocks).join('\n').trimEnd().endsWith(suffix); +} + +/** Read the subagent transcript's first line, retrying briefly because Claude + * Code may not have flushed it yet when SubagentStart fires. Total wait + * bounded by the sum of `RETRY_DELAYS_MS`. */ +const SUBAGENT_TRANSCRIPT_RETRY_DELAYS_MS = [0, 50, 100, 150]; +export async function readSubagentFirstLineWithRetry( + transcriptPath: string, +): Promise | undefined> { + for (const delay of SUBAGENT_TRANSCRIPT_RETRY_DELAYS_MS) { + if (delay > 0) await new Promise(r => setTimeout(r, delay)); + const line = readFirstTranscriptLine(transcriptPath); + if (line && line['type'] === 'user') return line; + } + return undefined; +} From 2d19b88b80e57cbf9096037107338697eae70ff0 Mon Sep 17 00:00:00 2001 From: rgao-coreweave Date: Fri, 17 Jul 2026 12:47:19 -0700 Subject: [PATCH 3/3] refactor(daemon): move session types + tracking to sessionState.ts (#127) Verbatim move of SessionState, the tracker types (PendingToolCall, SubagentTracker, TeamMember, LoadedInstruction), SubagentTracking, newSessionState, and resolvePermissionIfPending. Also dedupes the sha256-hex idiom into utils.sha256Hex now that it has two consumers (hashPrompt + the config fingerprint). No behavior change. Co-authored-by: Claude Fable 5 --- src/config.ts | 6 +- src/daemon.ts | 290 ++------------------------------------------ src/sessionState.ts | 247 ++++++++++++++++++++++++++++++++++++- src/utils.ts | 14 ++- 4 files changed, 267 insertions(+), 290 deletions(-) diff --git a/src/config.ts b/src/config.ts index 09314f8..572e43c 100644 --- a/src/config.ts +++ b/src/config.ts @@ -7,7 +7,7 @@ // import cycle (cli.ts imports the daemon entry point). import { DEFAULT_AGENT_NAME } from './genaiSpans.js'; -import { createHash } from 'crypto'; +import { sha256Hex } from './utils.js'; import type { Settings } from './setup.js'; /** Where a resolved value came from, for user-facing "source" reporting. */ @@ -131,8 +131,6 @@ const CONFIG_FINGERPRINT_LENGTH = 16; /** Short, stable hash of a daemon config. The API key is hashed, not exposed, * so the fingerprint is safe to send over the socket. */ export function daemonConfigFingerprint(c: DaemonConfig): string { - return createHash('sha256') - .update(JSON.stringify([c.weaveProject, c.apiKey, c.baseUrl, c.agentName, c.debug])) - .digest('hex') + return sha256Hex(JSON.stringify([c.weaveProject, c.apiKey, c.baseUrl, c.agentName, c.debug])) .slice(0, CONFIG_FINGERPRINT_LENGTH); } diff --git a/src/daemon.ts b/src/daemon.ts index 4721226..bcb0839 100644 --- a/src/daemon.ts +++ b/src/daemon.ts @@ -39,11 +39,21 @@ import type { import { loadSettings, VERSION } from './setup.js'; import { resolveDaemonConfig, daemonConfigFingerprint, missingConfig } from './config.js'; import { + resolvePermissionIfPending, hashPrompt, computeSubagentTranscriptPath, extractUserMessageContent, lastAssistantTextEndsWith, readSubagentFirstLineWithRetry, + newSessionState, + upsertInstruction, +} from './sessionState.js'; +import type { + PendingToolCall, + SubagentTracker, + TeamMember, + SessionState, + LoadedInstruction, } from './sessionState.js'; import { appendToLog, deepEqual } from './utils.js'; import { @@ -98,37 +108,6 @@ function isControlMessage(payload: unknown): payload is ControlMessage { return cmd === 'shutdown' || cmd === 'config-hash'; } -/** Stores the tool span opened at PreToolUse so PostToolUse can close it. */ -type PendingToolCall = { - span: Span; - toolName: string; - toolInput: Record; - /** True once a PermissionRequest event has been emitted for this tool. */ - permissionRequested?: boolean; -} - -/** Tracks the chat span currently open for a single assistant API response. - * Tool spans for that response parent here so the trace tree shows the - * model's interleaved text → tool_use → text order. The response's - * text/thinking children are emitted when the span is finalized (at the next - * response transition or at Stop), once all its split transcript lines are - * present. */ -type ActiveChatSpan = { - /** Response key (Anthropic `message.id`, or index fallback) this chat span - * represents; see `chatMessageKey`. */ - responseKey: string; - span: Span; -} - -/** Emit `weave.permission_resolved` on a pending tool call's span, if one was requested. */ -function resolvePermissionIfPending(pending: PendingToolCall, approved: boolean): void { - if (!pending.permissionRequested) return; - addPermissionResolvedEvent(pending.span, { - approved, - timestamp: new Date(), - }); -} - /** Stable identity for an assistant API call within a turn. Anthropic returns * a `message.id` on every response; that's the primary key. When it's * missing (legacy transcripts), fall back to the index, which is stable @@ -174,136 +153,6 @@ function parseIsoOrNow(ts: string | undefined): Date { return parseTimestamp(ts) ?? new Date(); } -/** - * Tracks a subagent across hook events. Two shapes: - * (a) Matched — created at PreToolUse when an Agent tool with subagent_type - * is detected; carries `toolUseId`, `promptHash`, and a reference to - * the subagent's `invoke_agent` span. `agentId` is filled in at - * SubagentStart via content-based correlation: sha256(firing prompt) + - * subagent_type. - * (b) Orphan — created at SubagentStart when no tracker matches the firing - * prompt (the parent's Agent PreToolUse never fired, or its prompt - * differs from the subagent transcript's line 1). The `invoke_agent` - * span is created at SubagentStart with the current turn span as - * parent and no input messages (the firing prompt is unavailable). - * - * The subagent is its own `invoke_agent ` span, child of the - * parent turn's `invoke_agent claude-code` span. Per the Weave Agents chat - * view (`weave/trace_server/agents/chat_view.py`), nested `invoke_agent` - * spans render as an `agent_start` lifecycle marker with the inner agent's - * own assistant text — distinct from an `execute_tool` tool-call event. - * The Agent tool call does NOT emit an `execute_tool` span; it emits this - * `invoke_agent` span directly. - */ -type SubagentTracker = { - subagentType: string; - detectedAt: Date; - toolUseId?: string; // tool_use_id of the spawning Agent tool (matched path only) - invokeAgentSpan?: Span; // subagent's `invoke_agent` span; subagent chat/tool spans parent here - agentId?: string; - /** sha256 of the prompt passed to the Agent tool; matched against the - * subagent's transcript line-1 user message at SubagentStart. */ - promptHash?: string; - /** True once the invoke_agent span has been ended. Guards against - * double-end when PostToolUse and SubagentStop both try to close it. */ - ended?: boolean; - /** Subagent transcript path — stored at SubagentStart so TeammateIdle can - * read all turns without relying on the payload's transcript_path (which - * CC sets to the coordinator's path, not the subagent's). */ - transcriptPath?: string; - /** Set on orphan trackers when SubagentStop fires before TeammateIdle. - * Suppresses span closure at SubagentStop so TeammateIdle can close it - * with full all-turns content. */ - pendingTeammateIdle?: boolean; - /** Set when this Agent tool spawn carried a `team_name` (agent-teams model). - * The teammate runs in its OWN session, so its TeammateIdle fires under a - * different session_id and the per-session lookup misses. The invoke_agent - * span is registered in GlobalDaemon.teamMembers and closed there (at the - * teammate's TeammateIdle), NOT at the coordinator's PostToolUse(Agent). */ - teamName?: string; -} - -/** Cross-session team correlation. In agent-teams (TeamCreate) a teammate is an - * independent Claude session whose TeammateIdle fires under the teammate's own - * session_id, not the coordinator's — so the per-session SubagentTracking - * lookup misses. The coordinator's PreToolUse(Agent, team_name) is the one - * reliable anchor; we record its invoke_agent span here keyed by - * `${team_name}::${name}`. - * - * Entries are stored as a FIFO queue per key (not a single value) because the - * SAME `${team}::${name}` can be spawned more than once in a run — e.g. the - * TARS triage flow re-spawns a specialist (Sonnet→Opus) for deeper work. Each - * spawn pushes its own TeamMember; each teammate's TeammateIdle consumes the - * oldest not-yet-emitted entry (FIFO), so re-spawns never overwrite a live span - * (which would leak it and mis-attribute the first teammate's transcript). This - * mirrors SubagentTracking.findPendingTeammateIdle for the per-session path. */ -type TeamMember = { - invokeAgentSpan: Span; - conversationId: string; - coordinatorTranscriptPath: string; - emitted: boolean; -} - -/** One instruction file surfaced by the `InstructionsLoaded` hook. Accumulated - * per session (deduped by path) and stamped as `gen_ai.system_instructions` on - * each turn root. */ -type LoadedInstruction = { filePath: string; content: string }; - -/** Append `item` to `list` in place, replacing any existing entry with the same - * filePath so a reloaded file (e.g. `load_reason=compact`) updates rather than - * duplicates. Preserves each file's first-seen position. */ -function upsertInstruction(list: LoadedInstruction[], item: LoadedInstruction): void { - const idx = list.findIndex((i) => i.filePath === item.filePath); - if (idx >= 0) list[idx] = item; - else list.push(item); -} - -type SessionState = { - sessionId: string; - /** Root ancestor's session id — used as `gen_ai.conversation.id` so resumed - * turns stitch with their pre-resume turns server-side. Equals `sessionId` - * for fresh (non-forked) sessions. Resolved once at SessionStart by - * walking `forkedFrom.sessionId` pointers across transcript files. */ - conversationId: string; - transcript: TranscriptFile; - cwd: string; - source: string; - initialRequestModel?: string; - /** Integration identity (name, version, meta.*) as OTel Baggage, built once - * at SessionStart. Activated for every event in `routeEvent` so each span - * inherits it via `IntegrationBaggageSpanProcessor`. */ - integrationBaggage: Baggage; - - currentTurnSpan?: Span; - - turnNumber: number; - totalToolCalls: number; - turnToolCalls: number; - toolCounts: Record; - - pendingToolCalls: Map; - subagents: SubagentTracking; - - /** Chat span currently open for an in-progress assistant API call. Tool - * spans from PreToolUse parent here; finalized at Stop or on transition - * to the next API call. Cleared at Stop. */ - activeChatSpan?: ActiveChatSpan; - /** Response keys (see `chatMessageKey`) in the current turn for which a chat - * span has been opened (open or already finalized). Stop uses this to - * identify responses that need a chat span emitted from scratch (responses - * with no tool_use blocks never triggered PreToolUse). Reset per turn. */ - emittedChatSpanResponseKeys: Set; - - /** Compaction attrs buffered while no turn span is open. Drained on next UserPromptSubmit. */ - pendingCompaction?: CompactionAttrs; - - /** Instruction files (global/project CLAUDE.md, .claude/rules, @-imports) - * captured from InstructionsLoaded, in load order, deduped by path. Stamped - * on every turn root as `gen_ai.system_instructions`. */ - systemInstructions: LoadedInstruction[]; - -} - // ───────────────────────────────────────────────────────────────────────────── // GlobalDaemon // ───────────────────────────────────────────────────────────────────────────── @@ -328,125 +177,6 @@ const CONNECTION_TIMEOUT_MS = 5_000; // 5 seconds per connection const MAX_SOCKET_PAYLOAD_BYTES = 4 * 1024 * 1024; // 4 MiB per message -/** - * Per-session container that tracks subagents from PreToolUse (when an Agent - * tool with subagent_type is detected) through SubagentStop. Single source of - * truth for the tracker list, with intent-revealing lookup methods. - */ -class SubagentTracking { - private trackers: SubagentTracker[] = []; - - /** Add a pending tracker at PreToolUse, before SubagentStart correlates an agent_id. */ - add(tracker: SubagentTracker): void { - this.trackers.push(tracker); - } - - /** - * Find the unmatched tracker (no agent_id yet) matching `(promptHash, - * subagentType)`. FIFO across ties: the oldest pending tracker wins, so two - * back-to-back identical Agent calls still correlate in dispatch order. - * Returns undefined if no candidate qualifies. - */ - findUnmatchedByContent(promptHash: string, subagentType: string): SubagentTracker | undefined { - let best: SubagentTracker | undefined; - for (const t of this.trackers) { - if (t.agentId) continue; - if (t.promptHash !== promptHash) continue; - if (t.subagentType !== subagentType) continue; - if (!best || t.detectedAt.getTime() < best.detectedAt.getTime()) best = t; - } - return best; - } - - byAgentId(agentId: string): SubagentTracker | undefined { - return this.trackers.find(t => t.agentId === agentId); - } - - /** Find a tracker awaiting TeammateIdle by its subagentType. Used to - * correlate TeammateIdle(teammate_name) with the orphan tracker created - * at SubagentStart. Returns the oldest pending match (FIFO). */ - findPendingTeammateIdle(subagentType: string): SubagentTracker | undefined { - let best: SubagentTracker | undefined; - for (const t of this.trackers) { - if (!t.pendingTeammateIdle) continue; - if (t.subagentType !== subagentType) continue; - if (!best || t.detectedAt.getTime() < best.detectedAt.getTime()) best = t; - } - return best; - } - - /** Lookup by spawning Agent tool's tool_use_id. Used at PostToolUse to find - * the subagent's `invoke_agent` span when the matching toolUseId is not - * in `pendingToolCalls` (because the Agent tool emits an invoke_agent - * span instead of an execute_tool span). */ - byToolUseId(toolUseId: string): SubagentTracker | undefined { - return this.trackers.find(t => t.toolUseId === toolUseId); - } - - remove(tracker: SubagentTracker): void { - const idx = this.trackers.indexOf(tracker); - if (idx >= 0) this.trackers.splice(idx, 1); - } - - size(): number { - return this.trackers.length; - } - - all(): SubagentTracker[] { - return [...this.trackers]; - } -} - -/** Options for {@link newSessionState}. `turnNumber` seeds the turn counter: 0 - * for a brand-new session, or the number of turns already on disk when - * reconstructing a session lost across a daemon restart (so the resumed turn - * keeps counting up instead of resetting to 1). */ -type NewSessionStateOptions = { - sessionId: string; - conversationId: string; - transcript: TranscriptFile; - cwd: string; - source: string; - initialRequestModel: string | undefined; - turnNumber: number; -}; - -/** Build a fresh SessionState. */ -function newSessionState(options: NewSessionStateOptions): SessionState { - const { sessionId, conversationId, transcript, cwd, source, initialRequestModel, turnNumber } = - options; - // Claude Code stamps its CLI version on each transcript line; capture it - // best-effort from the head line for the integration metadata. Absent when - // the writer hasn't flushed yet, the meta key is simply omitted. Built - // here (not at the SessionStart call site) so a session reconstructed after - // a daemon restart carries the same integration identity on its spans. - const headLine = readFirstTranscriptLine(transcript.resolvedPath); - const version = headLine?.['version']; - const claudeCodeAppVersion = typeof version === 'string' ? version : undefined; - const integrationBaggage = createIntegrationBaggage({ - version: VERSION, - meta: { claude_code_app_version: claudeCodeAppVersion }, - }); - - return { - sessionId, - conversationId, - transcript, - cwd, - source, - initialRequestModel, - integrationBaggage, - turnNumber, - totalToolCalls: 0, - turnToolCalls: 0, - toolCounts: {}, - pendingToolCalls: new Map(), - subagents: new SubagentTracking(), - emittedChatSpanResponseKeys: new Set(), - systemInstructions: [], - }; -} - /** Absolute real path of the daemon's own entry script, resolving the npm bin * symlink to the actual dist/cli.js (or src/cli.ts under tsx). Lets `status` * report which build the running daemon is executing. Falls back to the raw diff --git a/src/sessionState.ts b/src/sessionState.ts index 220b7db..eb31d89 100644 --- a/src/sessionState.ts +++ b/src/sessionState.ts @@ -6,14 +6,255 @@ // from daemon.ts; no behavior change. import * as path from 'path'; -import { createHash } from 'crypto'; +import type { Baggage } from '@opentelemetry/api'; +import type { Span } from '@opentelemetry/api'; +import { VERSION } from './setup.js'; import { parseSessionFd, extractAssistantTextBlocks } from './parser.js'; -import { readFirstTranscriptLine } from './transcriptFile.js'; +import { TranscriptFile, readFirstTranscriptLine } from './transcriptFile.js'; +import { addPermissionResolvedEvent, createIntegrationBaggage } from './genaiSpans.js'; +import type { CompactionAttrs } from './genaiSpans.js'; +import { sha256Hex } from './utils.js'; + +/** Stores the tool span opened at PreToolUse so PostToolUse can close it. */ +export type PendingToolCall = { + span: Span; + toolName: string; + toolInput: Record; + /** True once a PermissionRequest event has been emitted for this tool. */ + permissionRequested?: boolean; +} + +/** The chat span open for the in-flight assistant response; its tool spans + * parent here. Content lands when it is finalized (next response transition, + * or Stop), once all its transcript lines are flushed. */ +type ActiveChatSpan = { + /** Response key (Anthropic `message.id`, or index fallback) this chat span + * represents; see `chatMessageKey`. */ + responseKey: string; + span: Span; +} + +/** Emit `weave.permission_resolved` on a pending tool call's span, if one was requested. */ +export function resolvePermissionIfPending(pending: PendingToolCall, approved: boolean): void { + if (!pending.permissionRequested) return; + addPermissionResolvedEvent(pending.span, { + approved, + timestamp: new Date(), + }); +} + +/** + * Tracks a subagent across hook events. Matched trackers are created at + * PreToolUse(Agent) and correlated to an agent_id at SubagentStart by + * sha256(firing prompt) + type; orphans are created at SubagentStart when + * nothing matches. Either way the subagent is its own `invoke_agent` span + * under the turn (why a marker and not `execute_tool`: see the daemon's + * Agent-dispatch branch). + */ +export type SubagentTracker = { + subagentType: string; + detectedAt: Date; + toolUseId?: string; // tool_use_id of the spawning Agent tool (matched path only) + invokeAgentSpan?: Span; // subagent's `invoke_agent` span; subagent chat/tool spans parent here + agentId?: string; + /** sha256 of the prompt passed to the Agent tool; matched against the + * subagent's transcript line-1 user message at SubagentStart. */ + promptHash?: string; + /** True once the invoke_agent span has been ended. Guards against + * double-end when PostToolUse and SubagentStop both try to close it. */ + ended?: boolean; + /** Stored at SubagentStart; TeammateIdle's own transcript_path is the + * coordinator's, so this is the reliable copy. */ + transcriptPath?: string; + /** Orphan awaiting TeammateIdle: SubagentStop leaves the span open so + * TeammateIdle can close it with full all-turns content. */ + pendingTeammateIdle?: boolean; + /** Set for `team_name` spawns: the span is owned by + * GlobalDaemon.teamMembers and closed at the teammate's TeammateIdle, + * NOT at the coordinator's PostToolUse(Agent). */ + teamName?: string; +} + +/** One queued team-member spawn. A teammate is an independent session whose + * TeammateIdle fires under its OWN session_id, so the coordinator's + * PreToolUse(Agent, team_name) is the only reliable anchor: it queues the + * span in GlobalDaemon.teamMembers (FIFO per `${team}::${name}`; the same + * name can be re-spawned, and overwriting would leak the first, still-open + * span). */ +export type TeamMember = { + invokeAgentSpan: Span; + conversationId: string; + coordinatorTranscriptPath: string; + emitted: boolean; +} + +/** One instruction file surfaced by the `InstructionsLoaded` hook. Accumulated + * per session (deduped by path) and stamped as `gen_ai.system_instructions` on + * each turn root. */ +export type LoadedInstruction = { filePath: string; content: string }; + +/** Append `item` to `list` in place, replacing any existing entry with the same + * filePath so a reloaded file (e.g. `load_reason=compact`) updates rather than + * duplicates. Preserves each file's first-seen position. */ +export function upsertInstruction(list: LoadedInstruction[], item: LoadedInstruction): void { + const idx = list.findIndex((i) => i.filePath === item.filePath); + if (idx >= 0) list[idx] = item; + else list.push(item); +} + +export type SessionState = { + sessionId: string; + /** Root ancestor's session id — used as `gen_ai.conversation.id` so resumed + * turns stitch with their pre-resume turns server-side. Equals `sessionId` + * for fresh (non-forked) sessions. Resolved once at SessionStart by + * walking `forkedFrom.sessionId` pointers across transcript files. */ + conversationId: string; + transcript: TranscriptFile; + cwd: string; + source: string; + initialRequestModel?: string; + /** Integration identity (name, version, meta.*) as OTel Baggage, built once + * at SessionStart. Activated for every event in `routeEvent` so each span + * inherits it via `IntegrationBaggageSpanProcessor`. */ + integrationBaggage: Baggage; + + currentTurnSpan?: Span; + + turnNumber: number; + totalToolCalls: number; + turnToolCalls: number; + toolCounts: Record; + + pendingToolCalls: Map; + subagents: SubagentTracking; + + /** Chat span currently open for an in-progress assistant API call. Tool + * spans from PreToolUse parent here; finalized at Stop or on transition + * to the next API call. Cleared at Stop. */ + activeChatSpan?: ActiveChatSpan; + /** Response keys with a chat span already opened this turn; Stop emits + * fresh spans for the rest (tool-less responses never hit PreToolUse). */ + emittedChatSpanResponseKeys: Set; + + /** Compaction attrs buffered while no turn span is open. Drained on next UserPromptSubmit. */ + pendingCompaction?: CompactionAttrs; + + /** Instruction files from InstructionsLoaded, in load order, deduped by + * path; stamped on every turn root as `gen_ai.system_instructions`. */ + systemInstructions: LoadedInstruction[]; +} + +/** + * Per-session container that tracks subagents from PreToolUse (when an Agent + * tool with subagent_type is detected) through SubagentStop. Single source of + * truth for the tracker list, with intent-revealing lookup methods. + */ +export class SubagentTracking { + private trackers: SubagentTracker[] = []; + + /** Add a pending tracker at PreToolUse, before SubagentStart correlates an agent_id. */ + add(tracker: SubagentTracker): void { + this.trackers.push(tracker); + } + + /** Oldest unmatched tracker (no agent_id yet) for `(promptHash, type)`; + * FIFO so back-to-back identical Agent calls correlate in dispatch order. */ + findUnmatchedByContent(promptHash: string, subagentType: string): SubagentTracker | undefined { + let best: SubagentTracker | undefined; + for (const t of this.trackers) { + if (t.agentId) continue; + if (t.promptHash !== promptHash) continue; + if (t.subagentType !== subagentType) continue; + if (!best || t.detectedAt.getTime() < best.detectedAt.getTime()) best = t; + } + return best; + } + + byAgentId(agentId: string): SubagentTracker | undefined { + return this.trackers.find(t => t.agentId === agentId); + } + + /** Oldest tracker awaiting TeammateIdle for this subagentType (FIFO). */ + findPendingTeammateIdle(subagentType: string): SubagentTracker | undefined { + let best: SubagentTracker | undefined; + for (const t of this.trackers) { + if (!t.pendingTeammateIdle) continue; + if (t.subagentType !== subagentType) continue; + if (!best || t.detectedAt.getTime() < best.detectedAt.getTime()) best = t; + } + return best; + } + + /** Lookup by the spawning Agent tool's tool_use_id (PostToolUse settle). */ + byToolUseId(toolUseId: string): SubagentTracker | undefined { + return this.trackers.find(t => t.toolUseId === toolUseId); + } + + remove(tracker: SubagentTracker): void { + const idx = this.trackers.indexOf(tracker); + if (idx >= 0) this.trackers.splice(idx, 1); + } + + size(): number { + return this.trackers.length; + } + + all(): SubagentTracker[] { + return [...this.trackers]; + } +} + +/** Options for {@link newSessionState}. `turnNumber` seeds the turn counter: 0 + * for a brand-new session, or the number of turns already on disk when + * reconstructing a session lost across a daemon restart (so the resumed turn + * keeps counting up instead of resetting to 1). */ +type NewSessionStateOptions = { + sessionId: string; + conversationId: string; + transcript: TranscriptFile; + cwd: string; + source: string; + initialRequestModel: string | undefined; + turnNumber: number; +}; + +/** Build a fresh SessionState. */ +export function newSessionState(options: NewSessionStateOptions): SessionState { + const { sessionId, conversationId, transcript, cwd, source, initialRequestModel, turnNumber } = + options; + // Best-effort CC CLI version from the transcript head line; built here so a + // reconstructed session carries the same integration identity. + const headLine = readFirstTranscriptLine(transcript.resolvedPath); + const version = headLine?.['version']; + const claudeCodeAppVersion = typeof version === 'string' ? version : undefined; + const integrationBaggage = createIntegrationBaggage({ + version: VERSION, + meta: { claude_code_app_version: claudeCodeAppVersion }, + }); + + return { + sessionId, + conversationId, + transcript, + cwd, + source, + initialRequestModel, + integrationBaggage, + turnNumber, + totalToolCalls: 0, + turnToolCalls: 0, + toolCounts: {}, + pendingToolCalls: new Map(), + subagents: new SubagentTracking(), + emittedChatSpanResponseKeys: new Set(), + systemInstructions: [], + }; +} /** sha256 of the firing prompt — used to correlate an `Agent` PreToolUse with * the subagent's SubagentStart by matching transcript content. */ export function hashPrompt(prompt: string): string { - return createHash('sha256').update(prompt, 'utf8').digest('hex'); + return sha256Hex(prompt); } /** diff --git a/src/utils.ts b/src/utils.ts index e2fa616..2765fea 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -7,6 +7,12 @@ import * as net from 'net'; import * as path from 'path'; import * as readline from 'readline'; import { spawnSync } from 'child_process'; +import { createHash } from 'crypto'; + +/** Hex-encoded sha256 of `input`. */ +export function sha256Hex(input: string): string { + return createHash('sha256').update(input, 'utf8').digest('hex'); +} export function prompt(question: string): Promise { return new Promise((resolve) => { @@ -95,10 +101,12 @@ export function deepEqual(a: unknown, b: unknown): boolean { if (a === b) return true; if (a === null || b === null || typeof a !== 'object' || typeof b !== 'object') return false; if (Array.isArray(a) !== Array.isArray(b)) return false; - const keysA = Object.keys(a as object); - const keysB = Object.keys(b as object); + const ao = a as Record; + const bo = b as Record; + const keysA = Object.keys(ao); + const keysB = Object.keys(bo); if (keysA.length !== keysB.length) return false; - return keysA.every(k => deepEqual((a as Record)[k], (b as Record)[k])); + return keysA.every(k => deepEqual(ao[k], bo[k])); } /**