From 9026f42ccddb4d8f0ae00d0fe346eca25e5e4043 Mon Sep 17 00:00:00 2001 From: Olzhas Nurpeisov Date: Fri, 9 Oct 2026 17:24:33 +0100 Subject: [PATCH 1/2] fix: hold a turn until its final assistant row is in the transcript Claude Code fires Stop before it writes the turn's final assistant row; in interactive sessions often only after the hook exits. A turn that used tools was emitted at Stop without its final answer or last LLM call, and the late row was then dropped as an assistant row with no prompt in front of it. Hold the trailing turn when its last chat row is a tool_result, or when none of its assistant rows carries the end of the Stop payload's last_assistant_message. The next hook run emits it whole. Co-Authored-By: Claude Opus 5.5 --- dist/hook.cjs | 43 ++++++++++++++++++++------- src/hook.ts | 2 ++ src/pipeline.ts | 63 +++++++++++++++++++++++++++++++-------- tests/otlp-emit.test.ts | 66 +++++++++++++++++++++++++++++++++++++++++ 4 files changed, 152 insertions(+), 22 deletions(-) diff --git a/dist/hook.cjs b/dist/hook.cjs index d8f79a3..7726bed 100644 --- a/dist/hook.cjs +++ b/dist/hook.cjs @@ -26633,7 +26633,7 @@ function emitReadyTurns(emitter, config, sessionId, transcriptPath, turnsToEmit, } return emitted; } -function splitTrailingIncompleteTurn(rows) { +function splitTrailingIncompleteTurn(rows, finalAssistantText = "") { let lastUserIdx = -1; for (let i = 0; i < rows.length; i++) { const row = rows[i]; @@ -26648,13 +26648,28 @@ function splitTrailingIncompleteTurn(rows) { return [rows, []]; } const tail = rows.slice(lastUserIdx); - const tailHasAssistant = tail.some((r) => getUserOrAssistantRoleFromRow(r) === "assistant"); - if (tailHasAssistant) { - return [rows, []]; - } - return [rows.slice(0, lastUserIdx), tail]; + const hold = [rows.slice(0, lastUserIdx), tail]; + const assistantRows = []; + for (const row of tail) { + if (isToolResult(row)) { + assistantRows.length = 0; + } else if (getUserOrAssistantRoleFromRow(row) === "assistant") { + assistantRows.push(row); + } + } + if (assistantRows.length === 0) { + return hold; + } + const want = finalAssistantText.trim(); + if (want && !assistantRows.some((row) => { + const text = extractTextFromContent(getContentFromRow(row)).trim(); + return text !== "" && want.endsWith(text); + })) { + return hold; + } + return [rows, []]; } -function getNewTurnsFromTranscript(transcriptPath, sessionState, subagentMap, flushDeferredAgentTurns = false) { +function getNewTurnsFromTranscript(transcriptPath, sessionState, subagentMap, flushDeferredAgentTurns = false, finalAssistantText = "") { let rows; [rows, sessionState] = readNewJsonl(transcriptPath, sessionState, flushDeferredAgentTurns); if (sessionState.pendingTurnRows.length > 0) { @@ -26664,7 +26679,7 @@ function getNewTurnsFromTranscript(transcriptPath, sessionState, subagentMap, fl const taskIdToToolUseId = getTaskIdToToolUseId(subagentMap); let [deferredTurnRowLists, remainingRows] = resolveDeferredAgentTurns(rows, sessionState, taskIdToToolUseId); if (!flushDeferredAgentTurns) { - const [keep, hold] = splitTrailingIncompleteTurn(remainingRows); + const [keep, hold] = splitTrailingIncompleteTurn(remainingRows, finalAssistantText); sessionState.pendingTurnRows = hold; remainingRows = keep; } @@ -26700,7 +26715,13 @@ async function emitNewTurnsFromTranscript(emitter, config, sessionId, transcript debug(`Discovered ${Object.keys(subagentMap).length} subagent transcript(s)`); } let turns; - [turns, sessionState] = getNewTurnsFromTranscript(transcriptPath, sessionState, subagentMap, flushDeferredAgentTurns); + [turns, sessionState] = getNewTurnsFromTranscript( + transcriptPath, + sessionState, + subagentMap, + flushDeferredAgentTurns, + opts.finalAssistantText ?? "" + ); if (turns.length === 0) { saveSessionState(state, key, sessionState); return 0; @@ -26796,8 +26817,10 @@ async function main() { const flushDeferredAgentTurns = isSessionEndHookPayload(payload); const emitter = new TraceEmitter(config); try { + const lastAssistantMessage = payload.last_assistant_message; const emitted = await emitNewTurnsFromTranscript(emitter, config, sessionId, transcriptPath, { - flushDeferredAgentTurns + flushDeferredAgentTurns, + finalAssistantText: typeof lastAssistantMessage === "string" ? lastAssistantMessage : "" }); const dur = (Date.now() - start) / 1e3; info(`Processed ${emitted} turns in ${dur.toFixed(2)}s (session=${sessionId})`); diff --git a/src/hook.ts b/src/hook.ts index 9fb8c96..093e4dc 100644 --- a/src/hook.ts +++ b/src/hook.ts @@ -118,8 +118,10 @@ async function main(): Promise { const emitter = new TraceEmitter(config); try { + const lastAssistantMessage = payload.last_assistant_message; const emitted = await emitNewTurnsFromTranscript(emitter, config, sessionId, transcriptPath, { flushDeferredAgentTurns, + finalAssistantText: typeof lastAssistantMessage === "string" ? lastAssistantMessage : "", }); const dur = (Date.now() - start) / 1000; info(`Processed ${emitted} turns in ${dur.toFixed(2)}s (session=${sessionId})`); diff --git a/src/pipeline.ts b/src/pipeline.ts index b06eca0..79582aa 100644 --- a/src/pipeline.ts +++ b/src/pipeline.ts @@ -25,7 +25,13 @@ import { } from "./state.js"; import { getSubagentTranscriptsByToolUseId, type SubagentTranscript } from "./subagents.js"; import { exportWithTimeout, TraceEmitter } from "./tracer.js"; -import { getUserOrAssistantRoleFromRow, isToolResult, readNewJsonl } from "./transcript.js"; +import { + extractTextFromContent, + getContentFromRow, + getUserOrAssistantRoleFromRow, + isToolResult, + readNewJsonl, +} from "./transcript.js"; import { buildTurns, type Turn } from "./turns.js"; import type { Row } from "./types.js"; @@ -57,11 +63,16 @@ export function emitReadyTurns( } /** - * Split off an incomplete trailing turn — a user prompt not yet followed by any - * assistant row — so it is held for the next run instead of dropped. Returns - * [rowsToProcessNow, rowsToHold]. + * Split off an incomplete trailing turn so it is held for the next run instead + * of emitted without its ending. A trailing turn is incomplete when its user + * prompt has no assistant row yet, when its last chat row is a tool_result (the + * assistant's reply to it isn't written yet), or when none of its assistant + * rows carries the end of `finalAssistantText` — the text the Stop payload says + * the turn ended with. Claude Code fires Stop before it writes that final row, + * and in interactive sessions often only after the hook exits, so waiting for + * it inside the hook doesn't help. Returns [rowsToProcessNow, rowsToHold]. */ -function splitTrailingIncompleteTurn(rows: Row[]): [Row[], Row[]] { +function splitTrailingIncompleteTurn(rows: Row[], finalAssistantText = ""): [Row[], Row[]] { let lastUserIdx = -1; for (let i = 0; i < rows.length; i++) { const row = rows[i]!; @@ -76,18 +87,38 @@ function splitTrailingIncompleteTurn(rows: Row[]): [Row[], Row[]] { return [rows, []]; } const tail = rows.slice(lastUserIdx); - const tailHasAssistant = tail.some((r) => getUserOrAssistantRoleFromRow(r) === "assistant"); - if (tailHasAssistant) { - return [rows, []]; + const hold: [Row[], Row[]] = [rows.slice(0, lastUserIdx), tail]; + const assistantRows: Row[] = []; + for (const row of tail) { + if (isToolResult(row)) { + assistantRows.length = 0; // only replies written after the last tool_result count + } else if (getUserOrAssistantRoleFromRow(row) === "assistant") { + assistantRows.push(row); + } + } + if (assistantRows.length === 0) { + return hold; + } + // The final message can span several rows; its last text row ends the text. + const want = finalAssistantText.trim(); + if ( + want && + !assistantRows.some((row) => { + const text = extractTextFromContent(getContentFromRow(row)).trim(); + return text !== "" && want.endsWith(text); + }) + ) { + return hold; } - return [rows.slice(0, lastUserIdx), tail]; + return [rows, []]; } export function getNewTurnsFromTranscript( transcriptPath: string, sessionState: SessionState, subagentMap?: Record, - flushDeferredAgentTurns = false + flushDeferredAgentTurns = false, + finalAssistantText = "" ): [Turn[], SessionState] { let rows: Row[]; // At SessionEnd no more transcript bytes are coming, so a buffered final @@ -107,7 +138,7 @@ export function getNewTurnsFromTranscript( // Hold back an incomplete trailing turn (except at SessionEnd, which flushes // everything) so its user row is re-read with the assistant response next run. if (!flushDeferredAgentTurns) { - const [keep, hold] = splitTrailingIncompleteTurn(remainingRows); + const [keep, hold] = splitTrailingIncompleteTurn(remainingRows, finalAssistantText); sessionState.pendingTurnRows = hold; remainingRows = keep; } @@ -141,6 +172,8 @@ export function getNewTurnsFromTranscript( export interface EmitNewTurnsOptions { flushDeferredAgentTurns?: boolean; + // Stop payload's last_assistant_message: the text the turn ended with. + finalAssistantText?: string; exportFn?: (emitter: TraceEmitter) => Promise; } @@ -165,7 +198,13 @@ export async function emitNewTurnsFromTranscript( } let turns: Turn[]; - [turns, sessionState] = getNewTurnsFromTranscript(transcriptPath, sessionState, subagentMap, flushDeferredAgentTurns); + [turns, sessionState] = getNewTurnsFromTranscript( + transcriptPath, + sessionState, + subagentMap, + flushDeferredAgentTurns, + opts.finalAssistantText ?? "" + ); if (turns.length === 0) { saveSessionState(state, key, sessionState); return 0; diff --git a/tests/otlp-emit.test.ts b/tests/otlp-emit.test.ts index 177de07..8dc7cec 100644 --- a/tests/otlp-emit.test.ts +++ b/tests/otlp-emit.test.ts @@ -254,6 +254,72 @@ describe("incomplete trailing turn (flush race)", () => { assert.equal(saved().turnCount, 1); assert.equal(saved().pendingTurnRows.length, 0); }); + + it("holds a turn ending in a tool_result on Stop and emits it with its final answer", async () => { + dir = fs.mkdtempSync(path.join(os.tmpdir(), "lmnr-race-")); + process.env.CC_LMNR_STATE_DIR = dir; + const transcript = path.join(dir, "session.jsonl"); + const key = getSessionStateKey("sess", transcript); + const saved = () => getSessionState(loadHookState(), key); + const runHook = async (event: "Stop" | "SessionEnd") => { + const emitter = makeEmitter(); + const n = await emitNewTurnsFromTranscript(emitter, emitter.config, "sess", transcript, { + flushDeferredAgentTurns: event === "SessionEnd", + exportFn: async () => true, + }); + return [n, emitter] as const; + }; + + // Stop fires after the tool ran but before the final answer is written. + const rows = [ + userRow("read it"), + assistantRow([{ type: "tool_use", id: "t1", name: "Read", input: { file_path: "a" } }], { msgId: "m1" }), + toolResultRow("t1", "contents"), + ]; + fs.writeFileSync(transcript, rows.map((r) => JSON.stringify(r) + "\n").join("")); + const [heldCount] = await runHook("Stop"); + assert.equal(heldCount, 0); + assert.equal(saved().pendingTurnRows.length, 3); // held, not emitted without its ending + + // The final answer lands; the next run emits the whole turn. + fs.appendFileSync(transcript, JSON.stringify(assistantRow([{ type: "text", text: "done" }], { msgId: "m2", ts: "2026-07-08T10:00:12.000Z" })) + "\n"); + const [emittedCount, emitter] = await runHook("SessionEnd"); + assert.equal(emittedCount, 1); + const spans = spansByName(emitter.spans); + assert.ok(spans["LLM Call 2"]); + assert.match(String(attrs(spans["Claude Code - Turn 1 (sess)"]!)["lmnr.span.output"]), /done/); + }); + + it("holds a turn on Stop until the text the payload names is written", async () => { + dir = fs.mkdtempSync(path.join(os.tmpdir(), "lmnr-race-")); + process.env.CC_LMNR_STATE_DIR = dir; + const transcript = path.join(dir, "session.jsonl"); + const key = getSessionStateKey("sess", transcript); + const saved = () => getSessionState(loadHookState(), key); + const runStop = async (finalAssistantText: string) => { + const emitter = makeEmitter(); + const n = await emitNewTurnsFromTranscript(emitter, emitter.config, "sess", transcript, { + finalAssistantText, + exportFn: async () => true, + }); + return [n, emitter] as const; + }; + + // Stop fires with the final message's thinking row written but not its text row. + const rows = [userRow("hi"), assistantRow([{ type: "thinking", thinking: "hmm" }], { msgId: "m1" })]; + fs.writeFileSync(transcript, rows.map((r) => JSON.stringify(r) + "\n").join("")); + const [heldCount] = await runStop("All done."); + assert.equal(heldCount, 0); + assert.equal(saved().pendingTurnRows.length, 2); + + // The text row lands; the next Stop (whose payload names the next turn's + // text) emits the held turn whole. + fs.appendFileSync(transcript, JSON.stringify(assistantRow([{ type: "text", text: "All done." }], { msgId: "m1" })) + "\n"); + const [emittedCount, emitter] = await runStop("All done."); + assert.equal(emittedCount, 1); + assert.match(String(attrs(spansByName(emitter.spans)["Claude Code - Turn 1 (sess)"]!)["lmnr.span.output"]), /All done\./); + assert.equal(saved().pendingTurnRows.length, 0); + }); }); describe("emitTurn", () => { From 561dc75a16cc303f84a643c2f245f02892aea245 Mon Sep 17 00:00:00 2001 From: Olzhas Nurpeisov Date: Fri, 9 Oct 2026 17:24:53 +0100 Subject: [PATCH 2/2] fix: take a background subagent's output from its SubagentHandback report A background subagent ends with a SubagentHandback tool call instead of a final text message, so its span's output was empty. Fall back to the call's message. Also document the trailing-turn hold rules in CLAUDE.md. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 2 +- dist/hook.cjs | 16 +++++++++++++++- src/emit.ts | 22 +++++++++++++++++++++- tests/otlp-emit.test.ts | 33 +++++++++++++++++++++++++++++++++ 4 files changed, 70 insertions(+), 3 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 25e79d5..c3fb283 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -46,6 +46,6 @@ Laminar observability plugin for Claude Code. Stop/SessionEnd hooks parse the se - Per-session state (`~/.claude/state/lmnr_state.json`, or under `CC_LMNR_STATE_DIR`) stores the transcript byte offset; only advance it AFTER a successful export (`exportFn` returns true from the `ExportResultCode.SUCCESS` signal), so failed exports retry next Stop. A per-turn `emitTurn` exception is different: the turn is skipped (not counted in `turnCount`, no retry) because a malformed turn would fail identically on every retry and poison the session. Delivery is at-least-once by design: if the state write itself fails after a successful export (`saveHookState` is fail-open and only logs), the next run re-reads and re-exports the same turns as duplicate traces — exports can't be rolled back, and duplicating beats silently losing data. - Turns with async agent launches (`toolUseResult.status == "async_launched"`) are deferred until the task-notification row arrives, or flushed at SessionEnd. - **Unterminated final line flush:** `readNewJsonl` holds the segment after the last newline in `sessionState.buffer` as a partial line. At SessionEnd (`flushBuffer=true`), a buffered segment that parses as complete JSON is flushed as a row — otherwise a transcript ending without a trailing newline would strand its final row in the buffer forever (offset is already at EOF, so later runs read zero bytes). A genuinely partial line always fails `JSON.parse` and stays buffered. Pinned by the "complete unterminated final line flushed at session end" test. -- **Incomplete trailing turn hold (`pendingTurnRows`):** if a batch ends with a user prompt that has no assistant row yet, its raw rows are held in state and replayed (prepended) on the next run instead of being dropped — the offset still advances, but the rows are re-processed from state. This guards against the transcript being flushed between the `Stop` hook firing and the assistant row landing (observed with headless `claude -p`, where the user row would otherwise be consumed on `Stop` and the assistant orphaned on `SessionEnd`, losing the turn). SessionEnd never holds — it flushes everything. A turn with an assistant row is emitted immediately (no one-turn delay in the common interactive case). Pinned by the "incomplete trailing turn (flush race)" test. +- **Incomplete trailing turn hold (`pendingTurnRows`):** Claude Code fires `Stop` before it writes the turn's final assistant row — measured 60–80 ms later in headless `claude -p`, and in interactive sessions often only after the hook exits, so waiting inside the hook doesn't help. `splitTrailingIncompleteTurn` therefore holds the trailing turn's raw rows in state, replayed (prepended) on the next run, when (a) its user prompt has no assistant row yet, (b) its last chat row is a `tool_result` (the reply to it isn't written yet), or (c) none of its assistant rows carries the end of the Stop payload's `last_assistant_message` (passed as `finalAssistantText`; the final message can span rows, so a row's text must be a suffix). The offset still advances; the rows are re-processed from state. When the row is late, the turn is emitted one hook run later (next `Stop`, or `SessionEnd`, which never holds and flushes everything). Without the hold, a tool-using turn was emitted without its final answer and the late row was dropped as an assistant row with no prompt in front of it. Pinned by the "incomplete trailing turn (flush race)" tests. - The state file (`~/.claude/state/lmnr_state.json`, or under `CC_LMNR_STATE_DIR`) is purely local plugin state — no external consumer reads it — so its keys use the same camelCase convention as the rest of the code. The only fields serialized verbatim are raw Claude Code transcript rows (inside `pendingAgentTurns[].rows` / `pendingTurnRows` / `pendingTaskNotifications`); everything the code owns is a typed shape (`SessionState`, `PendingAgentTurn`, `ToolResultEntry`). Wire/OTLP names are the exception and stay in their required forms: `gen_ai.usage.*` keys, OpenAI-style `tool_call_id`/`name` in tool messages, `lmnr.*` attributes. - Keep `emitReadyTurns` / `emitNewTurnsFromTranscript`'s test seams (`emitTurnFn` / `exportFn` injection, both in `pipeline.ts`): ESM can't monkeypatch imported bindings, so these are the injection points the tests use to simulate emit/export failure. diff --git a/dist/hook.cjs b/dist/hook.cjs index 7726bed..99daa8f 100644 --- a/dist/hook.cjs +++ b/dist/hook.cjs @@ -26137,6 +26137,20 @@ function getToolResultForObservation(toolResultEntry) { const finalResultTimestamp = parseTimestamp(toolResultEntry.finalTimestamp); return { output, resultTimestamp, finalOutput, finalResultTimestamp }; } +function getSubagentReport(lastAssistant) { + const content = lastAssistant ? getContentFromRow(lastAssistant) : ""; + const text = extractTextFromContent(content); + if (text) { + return text; + } + for (const toolUse of getToolUseBlocks(content)) { + const message = toolUse.input?.message; + if (toolUse.name === "SubagentHandback" && typeof message === "string") { + return message; + } + } + return ""; +} function getShortTranscriptPathForMetadata(p) { if (typeof p === "string" && p) { const parts = p.split(/[/\\]/); @@ -26292,7 +26306,7 @@ function emitSubagentObservations(emitter, parentSpan, subagent, startTimestamp) const [subagentInputText] = truncateText(extractTextFromContent(getContentFromRow(firstTurn.userMsg))); const lastTurn = turns[turns.length - 1]; const lastAssistant = lastTurn.assistantMsgs[lastTurn.assistantMsgs.length - 1]; - const [subagentOutputText] = truncateText(extractTextFromContent(lastAssistant ? getContentFromRow(lastAssistant) : "")); + const [subagentOutputText] = truncateText(getSubagentReport(lastAssistant)); const description = subagent.description; const subagentName = typeof description === "string" && description ? `Subagent: ${description}` : "Subagent"; const subagentAttrs = {}; diff --git a/src/emit.ts b/src/emit.ts index fa52f44..e8a0cb8 100644 --- a/src/emit.ts +++ b/src/emit.ts @@ -123,6 +123,26 @@ function getToolResultForObservation(toolResultEntry: ToolResultEntry | null | u return { output, resultTimestamp, finalOutput, finalResultTimestamp }; } +/** + * The subagent's final report: its last message's text or, for a background + * subagent that hands its report back through a SubagentHandback tool call + * instead, that call's `message`. + */ +function getSubagentReport(lastAssistant: Row | undefined): string { + const content = lastAssistant ? getContentFromRow(lastAssistant) : ""; + const text = extractTextFromContent(content); + if (text) { + return text; + } + for (const toolUse of getToolUseBlocks(content)) { + const message = toolUse.input?.message; + if (toolUse.name === "SubagentHandback" && typeof message === "string") { + return message; + } + } + return ""; +} + function getShortTranscriptPathForMetadata(p: unknown): string | null { if (typeof p === "string" && p) { // Return the basename, mirroring Path(path).name. @@ -339,7 +359,7 @@ function emitSubagentObservations( const lastTurn = turns[turns.length - 1]!; const lastAssistant = lastTurn.assistantMsgs[lastTurn.assistantMsgs.length - 1]; - const [subagentOutputText] = truncateText(extractTextFromContent(lastAssistant ? getContentFromRow(lastAssistant) : "")); + const [subagentOutputText] = truncateText(getSubagentReport(lastAssistant)); const description = subagent.description; const subagentName = typeof description === "string" && description ? `Subagent: ${description}` : "Subagent"; diff --git a/tests/otlp-emit.test.ts b/tests/otlp-emit.test.ts index 8dc7cec..9ef33d5 100644 --- a/tests/otlp-emit.test.ts +++ b/tests/otlp-emit.test.ts @@ -452,6 +452,39 @@ describe("emitTurn", () => { } }); + it("subagent output falls back to its SubagentHandback report", () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), "lmnr-sub-")); + try { + const subJsonl = path.join(dir, "agent-hb.jsonl"); + const subRows = [ + userRow("count words", "2026-07-08T10:00:06.000Z"), + assistantRow([{ type: "tool_use", id: "hb", name: "SubagentHandback", input: { message: "14 words" } }], { + msgId: "sm1", + ts: "2026-07-08T10:00:08.000Z", + }), + toolResultRow("hb", "Report delivered to your caller.", "2026-07-08T10:00:08.500Z"), + ]; + fs.writeFileSync(subJsonl, subRows.map((r) => JSON.stringify(r)).join("\n") + "\n"); + + const subagents: Record = { + tu_task: { path: subJsonl, agentId: "hb", agentType: "general-purpose", description: "count" }, + }; + const emitter = emit( + [ + userRow("delegate"), + assistantRow([{ type: "tool_use", id: "tu_task", name: "Agent", input: { prompt: "go" } }]), + toolResultRow("tu_task", "14 words", "2026-07-08T10:00:09.000Z"), + assistantRow([{ type: "text", text: "14" }], { msgId: "m2", ts: "2026-07-08T10:00:10.000Z" }), + ], + subagents + ); + const subSpan = spansByName(emitter.spans)["Subagent: count"]!; + assert.equal(JSON.parse(attrs(subSpan)["lmnr.span.output"]).content, "14 words"); + } finally { + fs.rmSync(dir, { recursive: true, force: true }); + } + }); + it("async subagent + async tool result render on the resolving generation", () => { const dir = fs.mkdtempSync(path.join(os.tmpdir(), "lmnr-async-")); try {