feat(activity): correlate capa sh traces to provider conversations

src/cli/commands/activity-ingest.ts

@@ -10,7 +10,6 @@
 import { CANONICAL_HOOK_EVENTS, type CanonicalHookEvent } from "../../types/hooks";
 import { loadSettings } from "../../shared/config";
 import { normalizeActivityHookPayload } from "../../shared/agent-activity-normalize";
-import { getServerStatus } from "../utils/server-manager";
 
 /** Cursor (and similar) gate events that require a permission decision on stdout. */
 const PERMISSION_GATE_EVENTS = new Set<CanonicalHookEvent>([
@@ -21,6 +20,9 @@ const PERMISSION_GATE_EVENTS = new Set<CanonicalHookEvent>([
 	"subagentStart",
 ]);
 
+/** Max time to wait for hook stdin before giving up (fail-open). */
+const STDIN_READ_TIMEOUT_MS = 2_000;
+
 /**
  * Stdout payload for provider gate hooks. Observational events return null
  * (empty stdout is fine). Must stay valid JSON — empty string is not.
@@ -40,15 +42,15 @@ export function activityIngestGateStdout(
 
 export async function activityIngestCommand(args: string[]): Promise<void> {
 	const parsed = parseArgs(args);
-	// Always exit 0 from the process wrapper — this function may throw only
-	// for programmer errors; soft failures return normally.
+	// Emit gate JSON first so Cursor never blocks while we read stdin / POST.
+	const gate = activityIngestGateStdout(parsed.event);
+	if (gate) writeGateStdout(`${gate}\n`);
+
+	// Soft failures only — never throw out of the hook subprocess.
 	try {
 		await runIngest(parsed);
 	} catch {
 		/* fail-open */
-	} finally {
-		const gate = activityIngestGateStdout(parsed.event);
-		if (gate) writeGateStdout(`${gate}\n`);
 	}
 }
 
@@ -84,18 +86,13 @@ async function runIngest(parsed: {
 	const normalized = normalizeActivityHookPayload(event, raw, provider);
 	if (normalized.skip) return;
 
-	const status = await getServerStatus();
-	if (!status.running || !status.url) {
-		const settings = await loadSettings();
-		const fallback = clientConnectOrigin(
-			settings.server.host,
-			settings.server.port,
-		);
-		await postEvent(fallback, projectId, normalized);
-		return;
-	}
-
-	await postEvent(status.url, projectId, normalized);
+	const settings = await loadSettings();
+	// Always derive a client-routable origin (0.0.0.0 / :: are not fetchable).
+	const serverUrl = clientConnectOrigin(
+		settings.server.host,
+		settings.server.port,
+	);
+	await postEvent(serverUrl, projectId, normalized);
 }
 
 async function postEvent(
@@ -158,7 +155,13 @@ function parseArgs(args: string[]): {
 async function readStdin(): Promise<string> {
 	try {
 		if (process.stdin.isTTY) return "";
-		return await Bun.stdin.text();
+		// Hook hosts usually write stdin then close; don't hang forever if they don't.
+		return await Promise.race([
+			Bun.stdin.text(),
+			new Promise<string>((resolve) => {
+				setTimeout(() => resolve(""), STDIN_READ_TIMEOUT_MS);
+			}),
+		]);
 	} catch {
 		return "";
 	}

src/cli/index.ts

@@ -54,8 +54,7 @@ if (process.argv[2] === '__server__') {
       process.exit(1);
     });
 } else {
-  (async () => {
-    try {
+  try {
     // Start version check in the background while the command runs
     const subcommand = cliSubcommandFromArgv(process.argv);
     // Hook subprocesses (activity-ingest) must own stdout — only valid JSON for gate events.
@@ -448,10 +447,9 @@ if (process.argv[2] === '__server__') {
       console.log(`\n  A new version of capa is available: ${updateInfo.latestVersion} (current: ${updateInfo.currentVersion})`);
       console.log('  Run "capa upgrade" to update.\n');
     }
-    } catch (err) {
-      const message = err instanceof Error ? err.message : String(err);
-      error(message);
-      process.exit(ExitCode.SYSTEM_ERROR);
-    }
-  })();
+  } catch (err) {
+    const message = err instanceof Error ? err.message : String(err);
+    error(message);
+    process.exit(ExitCode.SYSTEM_ERROR);
+  }
 }

src/db/database.ts

@@ -409,6 +409,32 @@ export class CapaDatabase {
 		return this.toolCalls.get(id);
 	}
 
+	patchToolCallCorrelation(
+		id: string,
+		correlation: {
+			conversation_id: string;
+			generation_id: string | null;
+		},
+	): ToolCallRecord | null {
+		return this.toolCalls.patchCorrelation(id, correlation);
+	}
+
+	findUncorrelatedCapaShellTracesInWindow(input: {
+		projectId: string;
+		since: number;
+		until: number;
+	}): ToolCallRecord[] {
+		return this.toolCalls.findUncorrelatedCapaShellTracesInWindow(input);
+	}
+
+	findProviderCapaShHooksInWindow(input: {
+		projectId: string;
+		since: number;
+		until: number;
+	}): ToolCallRecord[] {
+		return this.toolCalls.findProviderCapaShHooksInWindow(input);
+	}
+
 	findLatestActivityCorrelation(
 		projectId: string,
 		withinMs?: number,

src/db/tool-calls.ts

@@ -122,7 +122,8 @@ export class ToolCallsRepo {
 		this.db.run(
 			`UPDATE tool_calls
        SET status = ?, duration_ms = ?, result_preview = ?, result_bytes = ?,
-           result_tokens = ?, input_tokens = COALESCE(?, input_tokens),
+           result_tokens = ?,
+           input_tokens = COALESCE(?, input_tokens),
            output_tokens = COALESCE(?, output_tokens),
            cache_read_tokens = COALESCE(?, cache_read_tokens),
            cache_write_tokens = COALESCE(?, cache_write_tokens),
@@ -145,6 +146,69 @@ export class ToolCallsRepo {
 		return this.get(id);
 	}
 
+	patchCorrelation(
+		id: string,
+		correlation: {
+			conversation_id: string;
+			generation_id: string | null;
+		},
+	): ToolCallRecord | null {
+		this.db.run(
+			`UPDATE tool_calls
+       SET conversation_id = ?, generation_id = ?
+       WHERE id = ?`,
+			[correlation.conversation_id, correlation.generation_id, id],
+		);
+		return this.get(id);
+	}
+
+	findUncorrelatedCapaShellTracesInWindow(input: {
+		projectId: string;
+		since: number;
+		until: number;
+	}): ToolCallRecord[] {
+		return this.db
+			.query(
+				`SELECT * FROM tool_calls
+         WHERE project_id = ?
+           AND source = 'shell'
+           AND (conversation_id IS NULL OR TRIM(conversation_id) = '')
+           AND started_at >= ?
+           AND started_at <= ?
+         ORDER BY started_at DESC, id DESC`,
+			)
+			.all(
+				input.projectId,
+				input.since,
+				input.until,
+			) as ToolCallRecord[];
+	}
+
+	findProviderCapaShHooksInWindow(input: {
+		projectId: string;
+		since: number;
+		until: number;
+	}): ToolCallRecord[] {
+		return this.db
+			.query(
+				`SELECT * FROM tool_calls
+         WHERE project_id = ?
+           AND kind = 'shell'
+           AND source IS NOT NULL
+           AND source NOT IN ('shell', 'mcp')
+           AND conversation_id IS NOT NULL
+           AND TRIM(conversation_id) <> ''
+           AND started_at >= ?
+           AND started_at <= ?
+           AND (
+             INSTR(LOWER(tool_name), 'capa sh') > 0
+             OR INSTR(LOWER(COALESCE(args_json, '')), 'capa sh') > 0
+           )
+         ORDER BY started_at ASC, id ASC`,
+			)
+			.all(input.projectId, input.since, input.until) as ToolCallRecord[];
+	}
+
 	get(id: string): ToolCallRecord | null {
 		return this.db
 			.query("SELECT * FROM tool_calls WHERE id = ?")

src/server/__tests__/tool-call-tracer.test.ts

@@ -402,7 +402,45 @@ describe("ToolCallTracer", () => {
 		expect(notified[1].record.status).toBe("ok");
 	});
 
-	it("inherits conversation/generation from latest provider activity", () => {
+	it("inherits conversation for capa MCP traces at start", () => {
+		db.insertToolCall({
+			id: "hook-1",
+			project_id: "proj-1",
+			session_id: null,
+			started_at: Date.now() - 1_000,
+			duration_ms: 10,
+			status: "ok",
+			source: "cursor",
+			kind: "tool",
+			tool_name: "Read",
+			meta_tool: null,
+			args_json: "{}",
+			result_preview: null,
+			result_bytes: null,
+			result_tokens: null,
+			error_message: null,
+			agent_id: null,
+			conversation_id: "conv-cursor",
+			generation_id: "gen-turn-2",
+		});
+
+		const tracer = new ToolCallTracer(db);
+		const id = tracer.start({
+			projectId: "proj-1",
+			sessionId: "mcp-sess",
+			source: "mcp",
+			kind: "tool",
+			toolName: "slack.search_public_and_private",
+			metaTool: "call_tool",
+			args: { query: "hi" },
+		});
+
+		const row = db.getToolCall(id);
+		expect(row?.conversation_id).toBe("conv-cursor");
+		expect(row?.generation_id).toBe("gen-turn-2");
+	});
+
+	it("does not inherit conversation for capa shell traces at start", () => {
 		db.insertToolCall({
 			id: "hook-1",
 			project_id: "proj-1",
@@ -436,8 +474,51 @@ describe("ToolCallTracer", () => {
 		});
 
 		const row = db.getToolCall(id);
-		expect(row?.conversation_id).toBe("conv-claude");
-		expect(row?.generation_id).toBe("gen-turn-1");
+		expect(row?.conversation_id).toBeNull();
+		expect(row?.generation_id).toBeNull();
+	});
+
+	it("links capa shell trace to provider hook by capa sh command when output differs", () => {
+		const started = Date.now();
+
+		db.insertToolCall({
+			id: "hook-pd",
+			project_id: "proj-1",
+			session_id: null,
+			started_at: started + 900,
+			duration_ms: 10,
+			status: "ok",
+			source: "cursor",
+			kind: "shell",
+			tool_name: "capa sh pagerduty list-incidents --statuses triggered",
+			meta_tool: null,
+			args_json: null,
+			result_preview: "Traceback from shell wrapper",
+			result_bytes: null,
+			result_tokens: null,
+			error_message: null,
+			agent_id: null,
+			conversation_id: "conv-pager",
+			generation_id: "gen-pd",
+		});
+
+		const tracer = new ToolCallTracer(db);
+		const id = tracer.start({
+			projectId: "proj-1",
+			source: "shell",
+			kind: "tool",
+			toolName: "pagerduty.list_incidents",
+			metaTool: "call_tool",
+			args: { statuses: "triggered" },
+		});
+
+		const finished = tracer.finish(id, {
+			status: "error",
+			resultPreview: "Error executing tool list_incidents: validation failed",
+		});
+
+		expect(finished?.conversation_id).toBe("conv-pager");
+		expect(finished?.generation_id).toBe("gen-pd");
 	});
 
 	it("does not inherit correlation when inheritCorrelation is false", () => {
@@ -583,8 +664,8 @@ describe("notifyToolCall SSE framing", () => {
 			result_tokens: 1,
 			error_message: null,
 			agent_id: null,
-			conversation_id: null,
-			generation_id: null,
+			conversation_id: "conv-test",
+			generation_id: "gen-test",
 			model: null,
 			attributes_json: null,
 			input_tokens: null,
@@ -599,4 +680,38 @@ describe("notifyToolCall SSE framing", () => {
 		expect(text.startsWith("event: tool-call\n")).toBe(true);
 		expect(text).toContain('"tool_name":"echo"');
 	});
+
+	it("does not push uncorrelated capa shell traces", () => {
+		const chunks: Uint8Array[] = [];
+		const clients = new Map<string, Set<(chunk: Uint8Array) => void>>();
+		clients.set("proj-1", new Set([(chunk) => chunks.push(chunk)]));
+
+		notifyToolCall(clients, "proj-1", {
+			id: "abc",
+			project_id: "proj-1",
+			session_id: null,
+			started_at: 1,
+			duration_ms: 10,
+			status: "ok",
+			source: "shell",
+			kind: "tool",
+			tool_name: "echo",
+			meta_tool: null,
+			args_json: "{}",
+			result_preview: "hi",
+			result_bytes: 2,
+			result_tokens: 1,
+			error_message: null,
+			agent_id: null,
+			conversation_id: null,
+			generation_id: null,
+			model: null,
+			attributes_json: null,
+			input_tokens: null,
+			output_tokens: null,
+			cache_read_tokens: null,
+			cache_write_tokens: null,
+		});
+		expect(chunks).toHaveLength(0);
+	});
 });

src/server/activity-routes.ts

@@ -1,6 +1,7 @@
 import { serializeActivityAttributes } from "../shared/activity-attributes";
 import { isAgentActivityEnabled } from "../shared/agent-activity";
 import { syncSystemActivityHooks } from "../shared/agent-activity-sync";
+import { tryLinkCapaShellTraceFromProviderHook } from "../shared/activity-trace-correlate";
 import { parseCapabilitiesFile } from "../shared/capabilities";
 import { detectCapabilitiesFile } from "../shared/paths";
 import { resolveProvidersForClean } from "../shared/providers/resolve";
@@ -135,7 +136,7 @@ export async function handlePostProjectActivityEvent(
 	});
 
 	// Tracer finish already notifies SSE via ToolCallTracer constructor notify.
-	deps.toolCallTracer.finish(id, {
+	const finished = deps.toolCallTracer.finish(id, {
 		status: status === "running" ? "ok" : status,
 		resultPreview: body.resultPreview,
 		errorMessage: body.errorMessage ?? null,
@@ -145,6 +146,11 @@ export async function handlePostProjectActivityEvent(
 		cacheWriteTokens: body.tokenUsage?.cache_write_tokens ?? null,
 	});
 
+	if (finished && kind === "shell") {
+		const linkedCapa = tryLinkCapaShellTraceFromProviderHook(deps.db, finished);
+		if (linkedCapa) deps.toolCallTracer.notifyRecord(linkedCapa);
+	}
+
 	return new Response(JSON.stringify({ ok: true, id }), {
 		status: 201,
 		headers: JSON_HEADERS,