Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion CLAUDE.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
59 changes: 48 additions & 11 deletions dist/hook.cjs
Original file line number Diff line number Diff line change
Expand Up @@ -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(/[/\\]/);
Expand Down Expand Up @@ -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 = {};
Expand Down Expand Up @@ -26633,7 +26647,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];
Expand All @@ -26648,13 +26662,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) {
Expand All @@ -26664,7 +26693,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;
}
Expand Down Expand Up @@ -26700,7 +26729,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;
Expand Down Expand Up @@ -26796,8 +26831,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})`);
Expand Down
22 changes: 21 additions & 1 deletion src/emit.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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";
Expand Down
2 changes: 2 additions & 0 deletions src/hook.ts
Original file line number Diff line number Diff line change
Expand Up @@ -118,8 +118,10 @@ async function main(): Promise<number> {
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})`);
Expand Down
63 changes: 51 additions & 12 deletions src/pipeline.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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";

Expand Down Expand Up @@ -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]!;
Expand All @@ -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<string, SubagentTranscript>,
flushDeferredAgentTurns = false
flushDeferredAgentTurns = false,
finalAssistantText = ""
): [Turn[], SessionState] {
let rows: Row[];
// At SessionEnd no more transcript bytes are coming, so a buffered final
Expand All @@ -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;
}
Expand Down Expand Up @@ -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<boolean>;
}

Expand All @@ -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;
Expand Down
Loading
Loading