fix(capture): correlate Stop turn meta with the flushed transcript record

scripts/trace-model-usage-e2e.mjs

@@ -38,7 +38,7 @@ const TABLE = process.env.HIVEMIND_SESSIONS_TABLE ?? "sessions_modeltest";
 const RUN_TAG = "modeltest-" + Date.now().toString(36);
 const MAX_FILES_PER_AGENT = Number(process.env.MAX_FILES ?? 300);
 
-const { parseClaudeTurnMeta, parseCodexTurnMeta, sdkTurnMeta } = await import(
+const { parseClaudeTurnMeta, parseCodexTurnMeta, scanClaudeTurnForCapture, sdkTurnMeta } = await import(
   join(ROOT, "dist/src/notifications/model-usage.js")
 );
 const { loadConfig } = await import(join(ROOT, "dist/src/config.js"));
@@ -143,6 +143,50 @@ function runHook(entry, payload) {
   return r.status === 0;
 }
 
+/**
+ * The transcript's newest non-sidechain assistant text — what a live Stop
+ * hook would receive as `last_assistant_message` for that transcript.
+ */
+function lastClaudeAssistantText(file) {
+  let raw;
+  try {
+    raw = readFileSync(file, "utf-8");
+  } catch {
+    return null;
+  }
+  const lines = raw.split("\n");
+  for (let i = lines.length - 1; i >= 0; i--) {
+    const t = lines[i].trim();
+    if (!t) continue;
+    let e;
+    try { e = JSON.parse(t); } catch { continue; }
+    // Mirror the live scanner: the newest text-bearing, non-sidechain
+    // assistant record — with NO usage requirement, because that is the
+    // record the hook will correlate against.
+    if (e?.type !== "assistant" || e.isSidechain === true) continue;
+    const c = e.message?.content;
+    const text = typeof c === "string"
+      ? c
+      : Array.isArray(c) ? c.map((b) => (b?.type === "text" && typeof b.text === "string" ? b.text : "")).join("") : "";
+    if (text.trim()) return text;
+  }
+  return null;
+}
+
+/**
+ * Live-faithful Claude parse for transcript selection: only pick transcripts
+ * whose FINAL turn would actually enrich under the hook's correlation
+ * semantics (newest text-bearing record matches + carries usage). Replaying a
+ * transcript whose final turn has no usage would correctly produce an
+ * unenriched row — useless for this harness's per-model assertions.
+ */
+function parseClaudeLive(file) {
+  const text = lastClaudeAssistantText(file);
+  if (!text) return null;
+  const scan = scanClaudeTurnForCapture(file, text);
+  return scan.status === "match" ? scan.meta : null;
+}
+
 async function main() {
   const config = loadConfig();
   if (!config) {
@@ -153,7 +197,7 @@ async function main() {
 
   const claudeModels = pickOnePerModel(
     listTranscripts(join(homedir(), ".claude", "projects"), MAX_FILES_PER_AGENT),
-    parseClaudeTurnMeta,
+    parseClaudeLive,
   );
   const codexModels = pickOnePerModel(
     listTranscripts(join(homedir(), ".codex", "sessions"), MAX_FILES_PER_AGENT),
@@ -176,10 +220,15 @@ async function main() {
   let n = 0;
   for (const [model, file] of claudeModels) {
     const sid = `${RUN_TAG}-cc-${n++}`;
+    // The capture hook correlates last_assistant_message with the transcript's
+    // newest assistant record (off-by-one guard), so a synthetic message would
+    // make every enrichment come back null — replay the transcript's REAL
+    // final assistant text.
+    const lastText = lastClaudeAssistantText(file);
     jobs.push({ agent: "claude_code", model, run: () =>
       runHook("dist/src/hooks/capture.js", {
         session_id: sid, transcript_path: file, cwd: ROOT,
-        hook_event_name: "Stop", last_assistant_message: `e2e trace for ${model}`,
+        hook_event_name: "Stop", last_assistant_message: lastText ?? `e2e trace for ${model}`,
       }) });
   }
   for (const [model, file] of codexModels) {

src/hooks/capture.ts

@@ -15,7 +15,7 @@ import { DeeplakeApi } from "../deeplake-api.js";
 import { projectNameFromCwd } from "../utils/project-name.js";
 import { log as _log } from "../utils/debug.js";
 import { buildSessionPath } from "../utils/session-path.js";
-import { parseClaudeTurnMeta } from "../notifications/model-usage.js";
+import { parseClaudeTurnMetaLive } from "../notifications/model-usage.js";
 import {
   bumpTotalCount,
   loadTriggerConfig,
@@ -139,7 +139,23 @@ async function main(): Promise<void> {
     // last assistant turn (best-effort; null on any read/parse failure). On
     // SubagentStop, last_assistant_message belongs to the subagent transcript;
     // transcript_path points at the parent session, so prefer the agent one.
-    const modelMeta = parseClaudeTurnMeta(input.agent_transcript_path ?? input.transcript_path);
+    // The Stop event races the transcript writer (the turn's final assistant
+    // record may not be flushed yet), so this re-reads across a short backoff
+    // and correlates the record with last_assistant_message — never attributing
+    // a PREVIOUS turn's tokens to this one. A sidechain record is a subagent's
+    // on a main-session Stop, but IS the turn on SubagentStop.
+    // An empty last_assistant_message can't be correlated with a transcript
+    // record — skipping the correlation would silently accept the PREVIOUS
+    // turn's record, so skip the enrichment instead.
+    const expectText = typeof input.last_assistant_message === "string" ? input.last_assistant_message.trim() : "";
+    if (!expectText) log("empty last_assistant_message — skipping turn-meta enrichment");
+    const modelMeta = expectText
+      ? await parseClaudeTurnMetaLive(
+          input.agent_transcript_path ?? input.transcript_path,
+          expectText,
+          Boolean(input.agent_transcript_path),
+        )
+      : null;
     entry = {
       id: crypto.randomUUID(),
       ...meta,

src/notifications/model-usage.ts

@@ -272,7 +272,8 @@ interface ClaudeUsage {
 interface ClaudeLine {
   type?: string;
   cwd?: unknown;
-  message?: { model?: unknown; usage?: ClaudeUsage; stop_reason?: unknown };
+  isSidechain?: unknown;
+  message?: { model?: unknown; usage?: ClaudeUsage; stop_reason?: unknown; content?: unknown };
 }
 
 /**
@@ -399,6 +400,165 @@ export function parseClaudeTurnMeta(transcriptPath?: string): TraceModelMeta | n
   }
 }
 
+/** Flatten a Claude message `content` (string or block array) to its text. */
+function flattenClaudeText(content: unknown): string {
+  if (typeof content === "string") return content;
+  if (!Array.isArray(content)) return "";
+  return content
+    .map((b) => (b && typeof b === "object" && (b as { type?: unknown }).type === "text" && typeof (b as { text?: unknown }).text === "string" ? (b as { text: string }).text : ""))
+    .join("");
+}
+
+/**
+ * Whether the transcript's assistant text plausibly IS the hook's
+ * `last_assistant_message`. Exact match after trimming, or an ANCHORED
+ * prefix with enough shared length that a coincidental hit is implausible —
+ * an unanchored substring test would let a short reply ("Done") match the
+ * PREVIOUS turn ("Done — details…") and resurrect the off-by-one this
+ * correlation exists to prevent.
+ */
+const MIN_TRUNCATION_MATCH_LEN = 32;
+
+function textMatches(entryText: string, expect: string): boolean {
+  const a = entryText.trim();
+  const b = expect.trim();
+  if (!a || !b) return false;
+  if (a === b) return true;
+  const shorter = a.length <= b.length ? a : b;
+  const longer = a.length <= b.length ? b : a;
+  return shorter.length >= MIN_TRUNCATION_MATCH_LEN && longer.startsWith(shorter);
+}
+
+/**
+ * One point-in-time scan of a Claude transcript for the CURRENT turn's
+ * assistant metadata.
+ *
+ * `"match"` — the newest relevant assistant record is complete (has usage) and,
+ * when `expectText` is given, its text corresponds to the captured
+ * `last_assistant_message`. `meta` is set.
+ *
+ * `"retry"` — the transcript hasn't caught up with the turn being captured:
+ * the file is missing/unreadable, ends in a partial JSONL record, has no
+ * relevant assistant record yet, or its newest relevant assistant record lacks
+ * usage or doesn't correspond to `expectText`. The caller should re-read after
+ * a short backoff. Crucially this NEVER falls back to an older usage-bearing
+ * record — that would silently attribute the previous turn's tokens to this
+ * one (the off-by-one this scan exists to prevent).
+ *
+ * `includeSidechain` — on SubagentStop the agent transcript IS the sidechain,
+ * so its records must not be filtered; on a main-session Stop, sidechain
+ * records belong to subagents and must be skipped.
+ */
+export interface ClaudeTurnScan {
+  status: "match" | "retry";
+  meta?: TraceModelMeta;
+}
+
+export function scanClaudeTurnForCapture(
+  transcriptPath: string | undefined,
+  expectText?: string,
+  includeSidechain = false,
+): ClaudeTurnScan {
+  if (!transcriptPath) return { status: "retry" };
+  const r = openTranscript(transcriptPath);
+  if (!r) return { status: "retry" };
+  try {
+    // The tail window can slice the newest records away only when a single
+    // record exceeds 1 MiB — scan the whole snapshot when the tail finds
+    // nothing relevant.
+    return scanClaudeTurn(r.lines, expectText, includeSidechain) ??
+      scanClaudeTurn(r.readWhole(), expectText, includeSidechain) ??
+      { status: "retry" };
+  } finally {
+    r.close();
+  }
+}
+
+/** Reverse-scan for the newest relevant assistant record. Null = nothing relevant in these lines. */
+function scanClaudeTurn(
+  lines: string[] | null,
+  expectText: string | undefined,
+  includeSidechain: boolean,
+): ClaudeTurnScan | null {
+  if (!lines) return null;
+  let sawNonBlank = false;
+  for (let i = lines.length - 1; i >= 0; i--) {
+    const trimmed = lines[i].trim();
+    if (!trimmed) continue;
+    let entry: ClaudeLine;
+    try {
+      entry = JSON.parse(trimmed) as ClaudeLine;
+    } catch {
+      // A malformed FINAL record is an in-progress append — wait for it to
+      // complete. Malformed records further up are just skipped.
+      if (!sawNonBlank) {
+        log("trailing partial record — retry");
+        return { status: "retry" };
+      }
+      sawNonBlank = true;
+      continue;
+    }
+    sawNonBlank = true;
+    if (!entry || typeof entry !== "object") continue;
+    const msg = entry.message;
+    if (entry.type !== "assistant" || !msg) continue;
+    if (!includeSidechain && entry.isSidechain === true) continue;
+    if (expectText !== undefined) {
+      // Only a TEXT-bearing record can be the one behind
+      // last_assistant_message — a turn's trailing tool_use/thinking records
+      // have no text and must not decide the correlation.
+      const text = flattenClaudeText(msg.content);
+      if (!text.trim()) continue;
+      if (!textMatches(text, expectText)) {
+        log("newest assistant record does not match last_assistant_message — retry");
+        return { status: "retry" };
+      }
+    }
+    // Newest relevant assistant record found — it alone decides the outcome.
+    if (!msg.usage) {
+      log("newest assistant record has no usage — retry");
+      return { status: "retry" };
+    }
+    const meta: TraceModelMeta = {
+      model: typeof msg.model === "string" ? msg.model : undefined,
+      reasoning_effort: readClaudeEffortLevel(typeof entry.cwd === "string" ? entry.cwd : undefined) ?? null,
+      token_usage: normalizeClaudeUsage(msg.usage),
+    };
+    if (typeof msg.stop_reason === "string") meta.stop_reason = msg.stop_reason;
+    const extra = claudeUsageExtra(msg.usage);
+    if (extra) meta.usage_extra = extra;
+    return { status: "match", meta };
+  }
+  return null;
+}
+
+/**
+ * Live-capture variant of {@link parseClaudeTurnMeta}: the Stop hook often
+ * fires before Claude has flushed the turn's final assistant record to the
+ * transcript, so a single read races the writer (observed in real sessions:
+ * turn 1 captured with null model/usage while the record landed ~ms later).
+ * Re-opens and re-scans the file across a short bounded backoff and returns
+ * null — capture proceeds unenriched — if the record never materializes.
+ */
+export async function parseClaudeTurnMetaLive(
+  transcriptPath: string | undefined,
+  expectText?: string,
+  includeSidechain = false,
+  backoffMs: readonly number[] = [25, 50, 100],
+): Promise<TraceModelMeta | null> {
+  let scan = scanClaudeTurnForCapture(transcriptPath, expectText, includeSidechain);
+  for (const delay of backoffMs) {
+    if (scan.status === "match") break;
+    await new Promise((resolve) => setTimeout(resolve, delay));
+    scan = scanClaudeTurnForCapture(transcriptPath, expectText, includeSidechain);
+  }
+  if (scan.status !== "match") {
+    log(`turn meta unavailable after ${backoffMs.length + 1} reads — writing unenriched`);
+    return null;
+  }
+  return scan.meta ?? null;
+}
+
 // ---------------------------------------------------------------------------
 // Codex
 // ---------------------------------------------------------------------------

tests/claude-code/capture-hook.test.ts

@@ -262,6 +262,9 @@ describe("capture hook — event-type branches", () => {
       last_assistant_message: "reply text",
     });
     await runHook();
+    // The Stop path re-reads the (here: absent) transcript across a bounded
+    // backoff before inserting unenriched — wait for the INSERT, not a tick.
+    await vi.waitFor(() => expect(queryMock).toHaveBeenCalled(), { timeout: 2000 });
     const sql = queryMock.mock.calls[0][0] as string;
     expect(sql).toContain('"type":"assistant_message"');
     expect(sql).toContain('"content":"reply text"');
@@ -277,6 +280,7 @@ describe("capture hook — event-type branches", () => {
       agent_transcript_path: "/tmp/agent.jsonl",
     });
     await runHook();
+    await vi.waitFor(() => expect(queryMock).toHaveBeenCalled(), { timeout: 2000 });
     const sql = queryMock.mock.calls[0][0] as string;
     expect(sql).toContain('"agent_transcript_path":"/tmp/agent.jsonl"');
   });