diff --git a/.specs/features/orchestrator-live-tokens/design.md b/.specs/features/orchestrator-live-tokens/design.md new file mode 100644 index 0000000..1625be4 --- /dev/null +++ b/.specs/features/orchestrator-live-tokens/design.md @@ -0,0 +1,243 @@ +# Orchestrator live tokens design + +**Spec**: `.specs/features/orchestrator-live-tokens/spec.md` +**Status**: Draft + +--- + +## Architecture overview + +The statusline sends Claude's `session_id` and `transcript_path` through the existing `usage` command. The CLI forwards the transcript observation alongside the cost observation in one `usage.get` request. The daemon resolves and validates the path, reads at most 1 MiB, and keeps its live cursor and dedupe state in a `Map` keyed by native id. It submits cumulative token values through the existing usage ledger. Release-time transcript reconciliation uses the same source key and per-field high-water marks. If a run query fails while `CODEDECK_RUN_ID` is set, the statusline omits `tok` rather than showing the smaller context-window snapshot. + +```mermaid +graph TD + P[Claude statusline payload] -->|session_id, cost, transcript_path| S[plugin/statusline.sh] + S -->|usage --json --observe --transcript| C[src/cli/commands/usage.ts] + C -->|usage.get| I[src/daemon/protocol.ts] + I --> D[Daemon] + D -->|validated chunk, at most 1 MiB| R[Incremental reader] + R -->|candidate cursor state and cumulative totals| D + D -->|ledger transaction| L[UsageLedger] + D -->|cursor after ledger commit| M[Daemon memory Map keyed by native id] + Q[Release reconciler] -->|cost-state or deduped tokens| L + L --> A[aggregateRunUsage] + A -->|worker totalTokens + orchestrator totalTokens| S +``` + +The statusline gives the usage command a one-second timeout (`plugin/statusline.sh:193-199`), and `buildSettings` requests a two-second refresh (`src/open/launchers/claude.ts:121-139`). After a daemon restart, a large transcript is reread from byte zero in 1 MiB requests. A 30 MB transcript therefore takes about 30 refreshes, roughly one minute, to catch up. During that time the persisted usage ledger ignores lower partial token totals until each field passes its high-water mark, so the displayed total does not drop. Earlier native ids must not block the live request while their full transcripts are reconciled. + +## Code reuse analysis + +### Existing components to reuse + +| Component | Location | How to use | +| --- | --- | --- | +| Statusline usage invocation | `plugin/statusline.sh:180-225` | Keep the current `usage --json` call and cost observation. Add the transcript argument from the same payload. | +| Statusline token rendering | `plugin/statusline.sh:248-257` | Replace the worker-only total with the worker and orchestrator `totalTokens` fields. Keep the local context snapshot only when no run id is set. | +| Usage CLI parser | `src/cli/commands/usage.ts:15-45,157-163` | Preserve `--observe =` and add `--transcript =`. Forward both values on one `usage.get` request. | +| IPC request type | `src/daemon/protocol.ts:161-167` | Extend `GetUsageRequest.params` with `transcript?: { nativeId: string; path: string }`. | +| Daemon usage handler | `src/daemon/daemon.ts:1085-1127` | Reuse run lookup, open-row selection, cost observation, link creation, and final `aggregateRunUsage` call. Add validated incremental transcript ingestion before aggregation. | +| Usage ledger | `src/store/usage-ledger.ts:4-18,53-112` | Reuse independent cost, input, output, and cached high-water marks. Do not change its attribution formula. | +| Orchestrator source key | `src/core/usage-source.ts:20-22` | Use `openSourceKey(nativeId)`, which returns `claude-open:`, for both live and release observations. | +| Release transcript reconciliation | `src/daemon/daemon.ts:342-378` | Keep full transcript reads. Their values enter `UsageLedger.observe` with the same source key as live observations. | +| Full transcript parser | `src/core/claude-transcript.ts:88-167` | Reuse the existing assistant usage mapping and `message.id` plus `requestId` dedupe rule. Add a pure chunk consumer for live reads without changing release behavior. | +| Run usage summary | `src/core/run-usage.ts:15-35,79-97` | Use the merged branch's worker `totalTokens` and `orchestrator.totalTokens` fields. Do not add cached tokens a second time. | + +There is no transcript cursor store. The feature adds no SQLite tables or cursor-store module. Existing SQLite usage-ledger rows continue to hold high-water marks. + +### Integration points + +| System | Integration method | +| --- | --- | +| Claude statusline | Pass `--transcript =` with the existing cost observation when both payload fields are valid. The statusline uses an argument array, so the path does not pass through a shell. | +| Usage CLI and IPC | Parse `--transcript` by splitting at the first `=` so the rest of a path is preserved. Send `transcript: { nativeId, path }` and the cost observation separately in one `usage.get` request. | +| Claude projects directory | Resolve the supplied path and `~/.claude/projects` with `realpath`. Require the resolved file to be under the resolved projects root, have basename `.jsonl`, and be a regular file. A symlink is allowed when its resolved target passes these checks. Use a cheap stat of device, inode, and size to detect replacement or truncation. Do not reject symlink components individually, use no-follow open flags, or compare an opened descriptor's identity with the validated stat. | +| Live reader state | Store the byte offset, pending line, file device/inode, cumulative token totals, and seen message keys in daemon memory in a `Map` keyed by native id. Keep an oversize-line skip flag alongside that state. On daemon restart, initialize at byte zero and reread in capped chunks. | +| Cursor lifetime | Process live transcript requests only for active Claude `open` rows, rechecking status inside the per-native-id serializer. After a successful full release pass, queue cleanup for every linked native id through the same serializer. Remove a cursor only when no active Claude `open` row remains linked to it; retain all cursors if the pass fails and retain any shared active id. | +| Usage ledger | Send cumulative input, output, and cached tokens with `openSourceKey(nativeId)`. The ledger retains independent high-water values across requests and daemon restarts. | +| Release reconciliation | Continue to read the full transcript. Pass `cost-state` totals or fallback deduped assistant totals through the same `claude-open:` source. The ledger adds only any positive difference above its marks. | + +The daemon and statusline run as the same local user. A client able to send a malicious transcript path can already read that file. The realpath containment, basename, and regular-file checks keep a transcript observation within the expected Claude project tree. Per-component symlink rejection and no-follow descriptor verification do not add a meaningful privilege boundary for this local process pair. + +Live ingestion sums newly seen `assistant.message.usage` records. Release reconciliation uses `cost-state.modelUsage` when present and otherwise uses deduplicated assistant records (`src/core/claude-transcript.ts:42-67,113-167`). These snapshots can differ. Since the ledger tracks input, output, and cached tokens independently, the final row can contain the maximum observed value for each field even when those maxima came from different snapshots. The design prevents double counting and prevents a release value from lowering a live mark. It does not promise that the final field tuple matches one `cost-state` snapshot when the sources disagree. + +The existing statusline fallback remains when `CODEDECK_RUN_ID` is absent. For an active run whose CLI query fails, the statusline omits `tok`. The cost value still comes from `payload.cost.total_cost_usd`; storage and attribution follow the orchestrator usage specification. + +## Components + +### Statusline script + +- **Purpose**: send the transcript path with the live orchestrator observation and display the combined run token total. +- **Location**: `plugin/statusline.sh` +- **Interface**: `codedeck usage --json [--observe =] [--transcript =]` +- **Behavior**: pass `--transcript` only when `session_id` and `transcript_path` are non-empty and the session id matches the existing Claude native-id format. When both `summary.totalTokens` and `summary.orchestrator.totalTokens` are finite, non-negative numbers, render their sum. If either value is absent, nonnumeric, non-finite, or negative, omit `tok` for that refresh. With `CODEDECK_RUN_ID` set, also omit `tok` when the query fails or the summary fails validation. Use the local context token fallback only when no run id is set. +- **Dependencies**: Claude's statusline stdin payload and `CODEDECK_RUN_ID`. +- **Reuses**: current cost observation, one-second timeout, ANSI formatting, and local token fallback. + +### Usage CLI and IPC protocol + +- **Purpose**: carry a transcript observation from the statusline to the daemon without changing cost observation semantics. +- **Location**: `src/cli/commands/usage.ts` and `src/daemon/protocol.ts` +- **Request interface**: +```typescript +export interface GetUsageRequest { + method: "usage.get"; + params: { + runId: string; + observe?: { nativeId: string; costUsd: number }; + transcript?: { nativeId: string; path: string }; + }; +} +``` +- **Behavior**: parse `--transcript` by splitting at the first `=` so paths containing `=` remain intact. Forward transcript and cost observations separately in one request. The daemon rejects transcript ingestion when both observations have different native ids. +- **Dependencies**: existing `usage.get` request and `RunUsageSummary`. +- **Reuses**: `parseUsageObservation`, `IpcClient`, and the merged `RunUsageSummary` type. + +### Daemon transcript ingestion + +- **Purpose**: validate a transcript path, read a bounded chunk, and submit cumulative token totals before returning run usage. +- **Location**: `src/daemon/daemon.ts`, next to the `usage.get` handler. +- **In-memory state**: +```typescript +interface LiveTranscriptCursor extends IncrementalTranscriptState { + byteOffset: number; + device: number; + inode: number; +} + +private readonly liveTranscriptCursors = new Map(); +``` +- **Behavior**: + 1. Find the `origin = 'open'` Claude row for the requested run. Preserve existing cost-observation handling for that row, but process its transcript only while the row is active. + 2. Validate a present `observe` value using the existing strict behavior. A malformed cost observation returns `INVALID` before transcript processing. If `observe` is valid, apply it independently even if transcript processing is rejected; if it is absent, continue with transcript processing without a cost observation. + 3. If `transcript` is present but is not an object with non-empty string `nativeId` and `path` fields, ignore the transcript portion and continue the request. Do not return `INVALID` or suppress a valid cost observation. + 4. Reject a transcript observation whose native id differs from the cost observation's native id when both are present. + 5. Resolve the real projects root and the real transcript path. Require the resolved path to be a descendant of the root, its basename to equal `.jsonl`, and `stat` to report a regular file. Accept symlink paths only when their resolved target passes these checks. + 6. Open the resolved path without `O_NOFOLLOW` or a post-open device/inode comparison. If validation or open fails, leave the cursor and token observations unchanged. A valid cost observation remains applied. + 7. Serialize transcript processing by native id. Recheck that the open row is still active inside the serialized section before linking or reading. If it became terminal while the request waited, skip transcript ingestion and continue to aggregation. Link a valid native id before reading, even when the request has no cost observation. Do not create a link for a rejected path unless the independent cost path already created it. + 8. Read at most `TRANSCRIPT_CHUNK_LIMIT_BYTES` new bytes for the native id. Use the same per-id serializer for reads, cursor replacement, and release cleanup so a post-release request cannot recreate a cursor after cleanup and an in-flight read finishes before cleanup. + 9. Pass the bytes to the pure chunk consumer. Parse complete newline-terminated assistant records and retain the final partial line in memory. Skip invalid complete JSON lines and ignore `cost-state` during live token ingestion. + 10. Submit cumulative token fields through `UsageLedger.observe` with `openSourceKey(nativeId)` inside a SQLite transaction for the ledger writes. Commit the ledger transaction, then replace the in-memory cursor. The cursor is never part of a SQLite transaction. If ledger persistence fails, keep the previous cursor so the next request rereads the bytes. + 11. If linking the id returns earlier unreconciled ids, send the live response first and queue their full transcript reconciliation afterward. The durable link state lets release or startup retry if the daemon stops before the queued reads finish. + 12. Make `reconcileOpenUsageSafely` return a success boolean for the requested pass: `true` when it returns normally, including when a vanished transcript is marked `missing` or was already reconciled; `false` when reconciliation or ledger persistence throws and the error is logged (`src/daemon/daemon.ts:381-390`). Only the full `session.release` pass may use `true` to consider cursor cleanup, not the selected earlier-id pass from `usage.get`. After a successful full release pass, get every native id linked to the released session and queue each cursor cleanup through that id's serializer. Remove a cursor only when no active Claude `open` row remains linked to that id. If the full pass fails, retain all cursors, including those whose individual link was reconciled before the failure. Use `SessionStore.listActive` and `NativeLinkStore.linksFor` for this check (`src/store/sessions.ts:213-217`; `src/store/native-links.ts:67-77`). + 13. Return `aggregateRunUsage` even when transcript validation fails. Keep a valid cost observation from the same request. + +### Incremental transcript reader + +- **Purpose**: process new transcript bytes while preserving the existing token mapping and dedupe behavior. +- **Location**: `src/core/claude-transcript.ts` +- **Interface**: +```typescript +export const TRANSCRIPT_CHUNK_LIMIT_BYTES = 1_048_576; +export const TRANSCRIPT_PENDING_LINE_LIMIT_BYTES = 4_194_304; + +export interface IncrementalTranscriptState { + pendingLine: Uint8Array; + discardUntilNewline: boolean; + inputTokens: number; + outputTokens: number; + cachedTokens: number; + seenMessageKeys: Set; +} + +export function consumeTranscriptChunk( + state: IncrementalTranscriptState, + bytes: Uint8Array, +): IncrementalTranscriptState; +``` +- **Behavior**: + - Reject a chunk larger than `TRANSCRIPT_CHUNK_LIMIT_BYTES` with `RangeError` and leave the supplied state unchanged. + - Parse newline-terminated records only. Keep the final fragment until a later chunk completes the line. + - If a pending line grows beyond `TRANSCRIPT_PENDING_LINE_LIMIT_BYTES`, drop it and ignore bytes through its next newline. Continue parsing after that newline. + - Skip complete invalid JSON lines. + - Read `message.usage` from assistant records. Count input, output, cache-read, and cache-creation values using the existing mapping in `src/core/claude-transcript.ts:113-135`. + - Store `JSON.stringify([message.id, requestId])` when both ids are present. Ignore a keyed record already in `seenMessageKeys`. Records missing either id keep the existing one-pass behavior. + - Return a new state without mutating the input state or its `Set`. + +### Usage ledger and release reconciliation + +- **Purpose**: keep live and release transcript values on the same per-native-id high-water source. +- **Location**: existing `src/store/usage-ledger.ts` and the transcript reconciler in `src/daemon/daemon.ts`. +- **Behavior**: the ledger tracks cost, input, output, and cached marks independently. A lower or equal cumulative field adds nothing. A higher field adds only the positive difference to its linked row. Live and release observations both use `claude-open:`. +- **Reuses**: `UsageLedger.observe` and `openSourceKey`. + +## Data models + +```typescript +interface LiveTranscriptCursor extends IncrementalTranscriptState { + byteOffset: number; + device: number; + inode: number; +} + +const liveTranscriptCursors = new Map(); +``` + +The map key is the native id. The cursor stores the offset after every byte read, including bytes held in `pendingLine`. It keeps cumulative input, output, and cached totals plus the set of keyed assistant records already counted. The current file stat supplies `device`, `inode`, and `size`. The map stores device and inode; each request compares the current size with the saved offset. + +Live transcript ingestion runs only for an active Claude `open` row. The handler checks that status again inside the per-native-id serializer, before it links the id or reads. The serializer covers transcript reads, cursor replacement, and release cleanup. A request that waits behind release therefore sees a terminal row and skips ingestion; a request already reading completes before release cleanup. + +After a full release reconciliation returns successfully, queue cleanup for every native id linked to that session through its per-id serializer. `reconcileOpenUsageSafely` returns `true` when `reconcileOpenUsage` completes without throwing, including when a vanished transcript is durably marked `missing` or its link already has a reconciled state. It returns `false` when reconciliation or ledger persistence throws and the error is logged. If it returns `false`, retain all cursors even if some links were reconciled before the error. For each id after a `true` full pass, remove its cursor only when no active Claude `open` row remains linked to it. This frees pending bytes and seen keys for released sessions without making a shared live id rescan unnecessarily. + +The reader counts only complete assistant records. `cachedTokens` is cache-read plus cache-creation input tokens. The statusline gets the combined total from the run summary, where the pricing helper has already applied the harness cache-counting rule. + +On daemon restart, the map starts empty. The first live request starts at byte zero and reads at most 1 MiB. Lower partial observations do not change existing ledger marks. As the reread passes each mark, the ledger attributes only the new difference. A roughly 30 MB transcript therefore catches up over about one minute of two-second refreshes without lowering the displayed total. + +## Error handling + +| Error scenario | Handling | User impact | +| --- | --- | --- | +| `transcript_path` is missing | Skip transcript reading and return the stored run summary. | Existing totals remain visible. | +| Resolved path is outside the projects root, has the wrong basename, or is not a regular file | Do not read it or change the cursor. Continue a valid cost observation and return the current summary. | Tokens may remain stale for this refresh; the statusline still runs. | +| `transcript` has the wrong runtime shape | Ignore the transcript parameter, keep the cursor unchanged, apply a valid cost observation, and return the current summary. | A malformed transcript value does not reject the whole `usage.get` request. | +| `observe` has the wrong runtime shape while `transcript` is valid | Preserve the existing strict validation: return `INVALID` before transcript processing. | A malformed cost observation does not create a transcript link or advance a cursor. | +| A `usage.get` transcript request reaches a terminal Claude `open` row | Skip transcript linking and ingestion after checking status inside the per-id serializer. Preserve the current cost path and return the current summary. | Post-release statusline refreshes do not recreate a live cursor. | +| Opening the validated real path fails | Do not change the cursor or token observations. Keep a valid cost observation and return the current summary. | Tokens may remain stale for this refresh. | +| Transcript ends with an incomplete line of at most 4 MiB | Keep its bytes in memory and resume them on the next request. | Only complete records affect totals. | +| Incomplete line grows beyond 4 MiB | Drop the fragment and skip bytes through its next newline. | That live record is omitted; later records still count. | +| Complete line contains invalid JSON | Skip that line and advance the cursor. | Later valid lines still count. | +| A keyed assistant record repeats in a later chunk | Ignore the repeated `(message.id, requestId)` key. | The record contributes once while the daemon is live. | +| Transcript exceeds the per-request byte cap | Commit the ledger observation, then advance the in-memory cursor and continue at the next statusline request. | Large transcripts show partial totals that rise over later refreshes. | +| Daemon restarts during a live session | Start the reader at byte zero and replay in 1 MiB chunks; keep existing ledger high-water marks. | Displayed totals do not drop while the transcript catches up. | +| Ledger observation fails after parsing | Keep the previous in-memory cursor and retry those bytes on the next request. | The next request can process the uncommitted chunk. | +| Usage invocation reaches one second or fails during an active run | The statusline catches the failure, omits `tok`, and prints remaining local fields. | Claude's prompt remains usable without showing a lower context-window value as the run total. | +| A run summary has missing or invalid worker/orchestrator `totalTokens` | Omit `tok` for that refresh when `CODEDECK_RUN_ID` is set. | The statusline does not show a partial sum or fall back to the smaller context-window snapshot. | +| Release reconciliation sees values already observed live | Apply the same source high-water marks independently to each token field. | Equal or lower values add nothing; a higher release value adds only the difference. A final tuple can combine per-field maxima from live and release snapshots. | +| A new live native id has earlier unreconciled ids | Send the live response, then queue the earlier ids' full reads. | The statusline's one-second budget is not spent on an unbounded earlier transcript. | + +## Risks and limits + +| Concern | Location (file:line) | Impact | Handling | +| --- | --- | --- | --- | +| The statusline supplies a filesystem path to the daemon. | `plugin/statusline.sh:184-198`; `src/daemon/daemon.ts:1085-1127` | An unchecked path could make the daemon read an arbitrary local file. | Resolve the path and root, then require containment, the expected basename, and a regular file. The statusline and daemon run as the same local user, so per-component symlink rejection and post-open descriptor identity checks are unnecessary for this boundary. | +| A large transcript must be reread after daemon restart. | `src/core/claude-transcript.ts:88-95`; `plugin/statusline.sh:193-199` | A 30 MB transcript takes about one minute of two-second refreshes to catch up. | Keep each request to 1 MiB. The ledger high-water marks keep the displayed total from dropping. | +| The seen-key set lives in daemon memory. | `src/core/claude-transcript.ts:122-128` | A restart loses the cursor and dedupe set. | Reread from byte zero. Cumulative ledger marks ignore lower partial totals until reread catches up. | +| A cursor keeps state after its session ends. | `src/daemon/daemon.ts` release reconciliation | Pending bytes and seen keys would remain in memory after they are no longer useful. | After a successful full pass, serialize cleanup for every id linked to the released session and delete each cursor only when no active Claude `open` row remains linked to it. A failed pass retains all cursors. | +| A statusline request overlaps release or arrives after release. | The live reader and `session.release` handler | A late request could recreate a cursor after cleanup, or an in-flight read could finish after cleanup. | Recheck activity inside the per-id serializer. After a successful full release pass, queue cleanup for every linked id in that serializer. A failed pass retains every cursor. | +| An active transcript's seen-key set grows with its keyed assistant records. | Live `Map` cursor | Memory use grows with the transcript while an `open` row still uses the native id. | Keep the keys needed for dedupe while the id is active, then remove the cursor after the last linked active row is reconciled. | +| A same-device, same-inode rewrite can keep a size at least as large as the saved offset. | The in-memory cursor uses stat device, inode, and size. | Such a rewrite is not detected and may leave live totals stale. | Reset only when device/inode changes or size falls below the saved offset. Release reconciliation still reads the full transcript. | +| An incomplete JSONL line exceeds the 4 MiB cap. | Incremental transcript reader | The live reader drops that line and skips to its newline. | Bound retained memory; later complete records continue to count. | +| A request links an id with prior native sessions. | `src/daemon/daemon.ts:1113-1120` | Full reads of earlier transcripts can exceed the statusline budget. | Send the live response first and queue prior-id reconciliation afterward. | + +## Technical decisions + +| Decision | Choice | Rationale | +| --- | --- | --- | +| CLI argument for transcript path | `--transcript =` | Keeps the current cost flag stable and passes the path as an argument instead of through a shell. | +| Allowed path | A regular file below the real `~/.claude/projects` root whose resolved basename is `.jsonl` | Matches Claude's transcript layout and keeps observations inside the expected project tree. | +| Symlink handling | Resolve the root and file, then validate the resolved path. Do not reject each symlink component or compare an opened descriptor's identity. | The daemon and statusline run as the same local user; a malicious-path caller can already read the file. | +| Read bound | 1,048,576 new bytes per native id per `usage.get` request | Keeps each statusline refresh bounded and resumes on the next two-second refresh. | +| Live cursor and dedupe state | Store offset, partial line, device/inode, cumulative totals, and seen keys in daemon memory keyed by native id. | Restarting the daemon begins at byte zero; the persisted usage ledger preserves displayed high-water totals during reread. | +| Oversized partial line | Retain at most 4 MiB. Drop a fragment that grows beyond the cap and skip to its next newline. | Prevents one unterminated record from growing without bound in memory. | +| Live and release source | `claude-open:` for both paths | Existing per-field high-water marks absorb overlap without changing the orchestrator usage contract. | +| Statusline token total | `summary.totalTokens + summary.orchestrator.totalTokens` | The merged usage summary already applies the harness-specific cache-counting rule. | +| Token values disagree at release | Keep each field's high-water maximum across live and release observations. | This avoids double counting and prevents lower release values from reducing a live total. A final tuple can combine fields from different snapshots. | +| Prior native id reconciliation during `usage.get` | Queue full prior-id reads after sending the live response. | The statusline's one-second budget covers bounded live ingestion only; durable link state supports retry. | + +## Open questions + +| Question | Current behavior | Why it remains open | +| --- | --- | --- | +| Which Claude Code version first provides `transcript_path`? | Skip live reading when the field is absent. | The current statusline documentation lists the field but does not state the first supported version. | +| Should assistant usage lines missing `message.id` or `requestId` get a fallback dedupe key after a transcript rewrite? | Preserve the current rule and dedupe only when both fields are present. | The existing reader does not define a stable identity for those lines. | +| Is 1,048,576 bytes enough for each two-second refresh on slower disks? | Keep the fixed cap and let later refreshes continue. | No latency measurement was requested or run for this documentation task. | diff --git a/.specs/features/orchestrator-live-tokens/run-notes.md b/.specs/features/orchestrator-live-tokens/run-notes.md new file mode 100644 index 0000000..4ca4341 --- /dev/null +++ b/.specs/features/orchestrator-live-tokens/run-notes.md @@ -0,0 +1,29 @@ +# Run notes: orchestrator-live-tokens + +Append-only. One entry per decision or event. + +## Before autonomous activation (human-approved) + +- S0. Investigation: in `codedeck open` the statusline `tok` field counts worker tokens only; reproduced `ctx 60% · 0 tok · run $2.10` with a fake payload and a fake `codedeck usage`. +- S0. Found Codex cached tokens double counted in `tok` and `queryUsage` totals. Fixed by worker f830 (commit ab2bcbb), PR #110 opened with the human's approval before autonomous mode. +- S0. Human decisions: cache tokens count in `tok`; orchestrator live tokens via incremental transcript read; spec cuts approved: in-memory cursor and dedupe (no SQLite tables), simplified path validation, `CLAUDE_CONFIG_DIR` out of scope. +- S0. Spec v1 by worker e174 (commit a87d632). Worker 76f5 merged the fix branch (7a3f067) then died on a backend 503; its queued `send` never delivered because the session was `failed`. Superseded by worker 6ff1 with the same briefing. + +## Autonomous run + +- A1. Autonomous mode activated by the human (`/codedeck:autonomous`). Bucket 2 from here on: no push, no PR, no network writes. Implementation stays on local branches. +- A2. Spec v2 + tasks.md accepted from worker 6ff1 (commit 1ed891c): LIVE-09 removed, cursor/dedupe in memory, simplified path validation, CLAUDE_CONFIG_DIR out of scope. Worker 6ff1 retired. +- A3. Decision (bucket 1): run T4 in parallel with T1 and T2 instead of after T3. Reason: the statusline only needs the fixed `--transcript` flag name and the `totalTokens` fields already on `RunUsageSummary`; its tests use a fake `codedeck` binary. Files are disjoint. +- A4. Decision (bucket 1): workers symlink `/home/andreello/dev/codedeck/node_modules` instead of running `npm install` (network fetch is bucket 2 in autonomous mode). Dependencies are unchanged on these branches. +- A5. Dispatched T1 bf8b (ra/live-tokens-t1-reader-bf8b), T2 b09d (ra/live-tokens-t2-cli-b09d), T4 6d82 (ra/live-tokens-t4-statusline-6d82), all based on 1ed891c. T3 waits for T1 and T2. +- A6. T1 accepted: commit 6a548f0, 2 owned files, `npx vitest run tests/claude-transcript.test.ts` 11/11 (re-run by orchestrator), mutation (dedupe key on message.id only) killed. Worker bf8b retired. +- A7. T2 accepted: commit 5e4c4f5, 3 owned files, `npx vitest run tests/usage-cli.test.ts` 13/13 (re-run by orchestrator), mutation (split at last `=`) killed. Worker b09d retired. +- A8. Merged T1 and T2 into the integration branch (1318ecc). Dispatched T3 60bc (ra/live-tokens-t3-daemon-60bc) based on 1318ecc. +- A9. T4 accepted: commit f81683b, 3 owned files, `npx vitest run tests/statusline.test.ts tests/usage-statusline-contract.test.ts` 28/28 (re-run by orchestrator). The old `input+output+cached` fallback for a summary without `totalTokens` was removed on purpose (LIVE-11: omit `tok`). Worker 6d82 retired; merged into integration. +- A10. T3 accepted: commit 05517f6, 2 owned files, `npx vitest run tests/orchestrator-usage-daemon.test.ts tests/usage-ledger.test.ts` 40/40 (re-run by orchestrator), mutation (drop path containment check) killed per worker log. Worker 60bc retired; merged into integration (bca85ce). +- A11. Integration verification on bca85ce: batch 1 `npx vitest run tests/claude-transcript.test.ts tests/pricing.test.ts tests/usage.test.ts tests/usage-query.test.ts tests/usage-ledger.test.ts` 60/60; batch 2 `npx vitest run tests/orchestrator-usage-daemon.test.ts tests/usage-daemon.test.ts` 43/43; batch 3 `npx vitest run tests/statusline.test.ts tests/usage-statusline-contract.test.ts tests/usage-cli.test.ts` 41/41; `npx tsc --noEmit -p .` clean. +- A12. Dispatched final read-only review 2730 (reviewer role) over 6344eeb..bca85ce. +- A13. Final review 2730: no blocker or major; 137 tests across 9 scoped files passed in its run. Minor 1 (earlier-id reconciliations pile up concurrently on every usage.get): accepted as a fix slice, dispatched 0309. Minor 2 (keyless assistant lines recounted after inode change or shrink): left as the spec open question, recorded as known risk. Minor 3 (redundant outer active-row check): no change, behaviour correct. Reviewer retired. +- A14. Fix de71eea accepted: 2 owned files, `npx vitest run tests/orchestrator-usage-daemon.test.ts tests/usage-ledger.test.ts` 40/40 (re-run by orchestrator), mutation (remove in-flight filter) killed: 2 reconciliations instead of 1. Correction reviewed by the orchestrator (diff read). Worker 0309 retired; fast-forwarded integration to de71eea. +- A15. Decision (bucket 1): renamed the local integration branch `ra/orchestrator-live-tokens-spec-e174` to `feat/orchestrator-live-tokens`. Not pushed (bucket 2). +- A16. Post-merge check on de71eea: `npx vitest run tests/orchestrator-usage-daemon.test.ts tests/usage-daemon.test.ts` 43/43; `npx tsc --noEmit -p .` clean. All workers of run c18d retired. diff --git a/.specs/features/orchestrator-live-tokens/run-report.md b/.specs/features/orchestrator-live-tokens/run-report.md new file mode 100644 index 0000000..a398a34 --- /dev/null +++ b/.specs/features/orchestrator-live-tokens/run-report.md @@ -0,0 +1,60 @@ +# Run report: orchestrator-live-tokens + +Goal: in `codedeck open`, the Claude statusline `tok` field counted worker tokens only, so a run without workers showed `0 tok` while `run $` already included the orchestrator. This run makes `tok` equal worker `totalTokens` plus orchestrator `totalTokens`, with the orchestrator read live from its transcript, and fixes the Codex cached-token double count found on the way. + +## Done + +- Branch (local only, not pushed): [feat/orchestrator-live-tokens](https://github.com/4ndreello/codedeck/tree/feat/orchestrator-live-tokens) +- Commit: [de71eea1fc9a5e062524fddb773bfe6bf89ffe95](https://github.com/4ndreello/codedeck/commit/de71eea1fc9a5e062524fddb773bfe6bf89ffe95) +- The links resolve only after the branch is pushed. Base: `6344eeb`. Diff: 20 files, 2096 insertions, 49 deletions. +- The run notes and this report are committed on top of that commit. + +| Commit | Worker | Slice | Evidence checked by the orchestrator | +| --- | --- | --- | --- | +| ab2bcbb | f830 | `fix(usage)`: count Codex cached tokens once (`totalTokensFor`, `RunUsageSummary.totalTokens`). Also pushed as PR #110 before autonomous mode. | 6 files, 74/74 tests; mutation (always add cached) killed | +| a87d632, 1ed891c | e174, 6ff1 | spec, design, tasks (LIVE-01..15, LIVE-09 removed) | docs only | +| 6a548f0 | bf8b | T1: `consumeTranscriptChunk`, 1 MiB chunk cap, 4 MiB pending-line cap, dedupe by `message.id`+`requestId` across chunks | `tests/claude-transcript.test.ts` 11/11; mutation (key on `message.id` only) killed | +| 5e4c4f5 | b09d | T2: `--transcript =` CLI flag, `usage.get` `transcript` param | `tests/usage-cli.test.ts` 13/13; mutation (split at last `=`) killed | +| f81683b | 6d82 | T4: statusline forwards `transcript_path`, `tok` = worker + orchestrator `totalTokens`, omits `tok` when a run is active but the summary is unavailable | statusline + contract tests 28/28; mutation (drop orchestrator term) killed | +| 05517f6 | 60bc | T3: daemon live ingestion: in-memory cursor per native id, per-id lock, realpath containment under `~/.claude/projects`, `.jsonl` basename, regular file, reset on inode change or shrink, ledger commit before cursor swap, earlier-id reconciliation after the response | daemon + ledger tests 40/40; mutation (drop containment check) killed | +| de71eea | 0309 | review fix: skip duplicate in-flight earlier-id reconciliations | daemon + ledger tests 40/40; mutation (remove in-flight filter) killed | + +Integration checks on the merged branch (`bca85ce`, then `de71eea`): +- `npx vitest run tests/claude-transcript.test.ts tests/pricing.test.ts tests/usage.test.ts tests/usage-query.test.ts tests/usage-ledger.test.ts`: 5 files, 60 tests passed +- `npx vitest run tests/orchestrator-usage-daemon.test.ts tests/usage-daemon.test.ts`: 2 files, 43 tests passed (re-run after de71eea) +- `npx vitest run tests/statusline.test.ts tests/usage-statusline-contract.test.ts tests/usage-cli.test.ts`: 3 files, 41 tests passed +- `npx tsc --noEmit -p .`: clean + +The final read-only review (reviewer 2730, over `6344eeb..bca85ce`) found no blocker and no major issue. It ran 9 scoped files with 137 passing tests and found no drift outside each task's owned files. It reported three minor points: one was fixed in de71eea, and the other two are listed below. + +## Assumptions I made + +- The human approved these before autonomous mode: cache tokens count in `tok`; the orchestrator's live tokens come from an incremental transcript read; cursor and dedupe state stay in memory (no SQLite tables); path validation is simplified; `CLAUDE_CONFIG_DIR` is out of scope. +- Bucket 1: T4 ran in parallel with T1 and T2 instead of after T3. The statusline needs only the fixed flag name and the `totalTokens` fields, and its tests use a fake `codedeck` binary. +- Bucket 1: workers symlinked `/home/andreello/dev/codedeck/node_modules` instead of running `npm install`, because a network fetch is bucket 2. Dependencies are unchanged. +- Bucket 1: T4 removed the old `input + output + cached` fallback for a summary without `totalTokens`. Per LIVE-11, `tok` is omitted instead. The CLI and the plugin ship in the same package, so a version mismatch between them is not expected. +- Bucket 1: renamed the integration branch to `feat/orchestrator-live-tokens`. +- Bucket 1: the review fix for concurrent earlier-id reconciliations was applied without asking. + +## Deferred / waiting for you + +- Push `feat/orchestrator-live-tokens` and open its PR. This is bucket 2: publishing commits. +- Order against PR #110: the feature branch already contains `ab2bcbb`. Merge #110 first, or close it and ship everything through the feature PR. +- Delete the scratch directories that workers left in `/tmp`: `codedeck-t1-mutation.*`, `codedeck-t4-mutant.*`, `codedeck-usage-reconcile-probe.*`, `codedeck-token-mutation-*`. Their sandbox refused `rm -rf`. Deleting files is bucket 2. +- Open product questions that remain in the spec: + - Should keyless assistant lines get a fallback dedupe key? Today they can be recounted after an inode change or a shrink; this is the known risk in the next section. + - Is 1 MiB per refresh enough on slow disks? + - What pruning policy should in-memory cursors follow beyond release cleanup? + +## Blocked / failed + +- Worker 76f5 (spec v2) failed on a backend 503 (`Unable to verify Daybreak Blue access`). The merge it had already made (7a3f067) was kept. `codedeck send` to the failed session stayed queued and was never delivered. The worker was superseded by 6ff1, which completed. The undelivered queued `send` to a `failed` session may be a CodeDeck send-queue gap. It was not investigated. +- Known risk, not fixed (review minor 2): an assistant line without `message.id` or `requestId` is counted again when the transcript's inode changes or the file shrinks below the saved offset. A daemon restart is not affected, because the ledger high-water marks absorb the lower re-read totals. + +## Not covered + +- No end-to-end run against a real `codedeck open` session, real daemon and real transcript. Coverage comes from integration tests with fixture transcripts and a fake `codedeck` binary. +- The `dist/` gates (`scripts/pty-gate.sh`, theme and rename gates) were not run, and there was no `npm run build`. +- The full test suite was not run, by the machine constraint. Only the files listed above ran. +- Read latency of the 1 MiB cap on a slow disk was not measured. +- After a daemon restart, a transcript of about 30 MB takes about a minute of 2 s refreshes to catch up. The displayed total does not drop meanwhile. This behaviour is in the design and was not measured. diff --git a/.specs/features/orchestrator-live-tokens/spec.md b/.specs/features/orchestrator-live-tokens/spec.md new file mode 100644 index 0000000..76f3169 --- /dev/null +++ b/.specs/features/orchestrator-live-tokens/spec.md @@ -0,0 +1,140 @@ +# Orchestrator live tokens specification + +## Problem Statement + +In `codedeck open`, the Claude statusline shows worker tokens but omits the orchestrator's tokens. A run with no workers can therefore show `0 tok` while its cost already includes the orchestrator. The statusline cannot use `context_window.total_*_tokens` as a session total because Claude Code reports the current context window there. This feature reads the session transcript incrementally and adds its cumulative token usage to the run summary. + +The current statusline submits cost through `codedeck usage` and has a one-second command timeout, but it does not submit `transcript_path` or read transcript tokens (`plugin/statusline.sh:180-225`). Its aggregate token field is worker-only (`plugin/statusline.sh:248-256`). The existing transcript reader streams the full file at release and keeps message dedupe keys only for that read (`src/core/claude-transcript.ts:88-167`). + +## Relationship to existing specifications + +This feature extends `.specs/features/orchestrator-usage/spec.md`. It replaces that spec's "Live orchestrator token counts" out-of-scope entry. It also replaces the token-source clause in `.specs/features/run-usage-statusline/spec.md` acceptance criterion 5, which sources orchestrator tokens from the local context window, and the worker-only token total in acceptance criterion 9. When `CODEDECK_RUN_ID` is set but the run query is unavailable, it replaces the local-token fallback in criterion 9 and the fallback in criterion 11. When `CODEDECK_RUN_ID` is absent, the local-token fallback remains. The cost value still comes from `payload.cost.total_cost_usd`; `.specs/features/orchestrator-usage/spec.md` defines its storage and attribution. Other requirements in those specifications remain in force. + +## Goals + +- Show cumulative worker and orchestrator tokens in the Claude `open` statusline. +- Read a bounded amount of new transcript data per refresh and continue later without losing or double-counting token observations. +- Keep live transcript tokens and release-time transcript reconciliation on the same per-native-session high-water ledger. + +## Out of Scope + +| Feature | Reason | +| --- | --- | +| Input, output, or cache breakdown on the statusline | The agreed display is one total token count. | +| Live tokens in statuslines for Codex, OpenCode, or OMP | This feature covers the Claude orchestrator started by `codedeck open`. | +| Replacing transcript totals with `context_window` token values | Those values describe the current context window, not the cumulative session. | +| Durable cursor and dedupe storage | The daemon keeps live reader state in memory. Its usage ledger keeps the per-field high-water marks. | +| Non-default Claude config roots through `CLAUDE_CONFIG_DIR` | CodeDeck accepts only `~/.claude/projects` for this feature. The repository does not set or read `CLAUDE_CONFIG_DIR` in `src/` or `plugin/hooks/`. | + +## Assumptions & Open Questions + +| Decision | Chosen behavior | Rationale | +| --- | --- | --- | +| Run summary token fields | Use `RunUsageSummary.totalTokens` and `RunUsageSummary.orchestrator.totalTokens`, both present on this branch after commit `7a3f067`. | The statusline consumes the run aggregate instead of recalculating harness-specific cache rules. | +| Transcript read cap | Read at most 1,048,576 new bytes per native id per `usage.get` request. | A fixed byte bound limits work under the statusline's one-second timeout. | +| Restart after daemon exit | Keep the byte offset, pending line, cumulative totals, and seen keys in daemon memory only. After restart, begin at byte zero and reread in capped chunks. | The persisted per-field ledger ignores lower partial totals until the reread passes each high-water mark, so the displayed total does not drop. | +| Large transcript catch-up | A 30 MB transcript takes about 30 refreshes, or roughly one minute at the current two-second refresh interval, after a daemon restart. | The 1 MiB request cap remains in place; the existing high-water marks keep displayed totals stable during catch-up. | +| Transcript path | Resolve the projects root and supplied path with `realpath`. Accept only a regular file below the resolved root whose resolved basename is `.jsonl`. | The statusline and daemon run as the same local user. A caller able to submit a malicious path can already read that file. Canonical containment, basename, and file-type checks keep observations tied to the expected transcript tree without per-component symlink rejection or descriptor identity checks. | +| Rejected transcript path | Skip transcript reading and leave its cursor and token observations unchanged. A valid cost observation in the same request remains eligible for recording. | A bad transcript path must not suppress the existing cost path or make the statusline fail. | +| Partial JSONL line | Keep an incomplete line up to 4 MiB in memory. If it grows beyond that cap, drop the fragment and skip bytes through the next newline. | The read cap bounds new bytes per request. This also bounds retained partial-line memory. | +| Same-file rewrite detection | Compare the current stat device and inode with the cursor identity, and compare size with the saved offset. Restart at byte zero if the identity changes or the file shrinks below the offset. | These cheap checks catch replacement and truncation. A same-identity rewrite that remains at least as large as the saved offset is not detected. | + +**Open questions:** + +1. Which minimum Claude Code version must CodeDeck support for `transcript_path`? If the field is absent, live transcript reading is skipped. +2. Should assistant usage lines missing `message.id` or `requestId` get a fallback dedupe key after a transcript rewrite? The existing reader counts such lines once per read and this feature keeps that rule. +3. Is the 1,048,576-byte cap enough for each refresh on slower disks? The cap is fixed for this design; no latency measurement was requested. + +## User Stories + +### P1: Show live orchestrator tokens + +**User story**: As a person using `codedeck open`, I want the statusline token count to include the orchestrator so that `tok` and `run $` describe the same run. + +**Why P1**: The orchestrator can account for most of a run's usage, but the current statusline displays only worker tokens. + +**Acceptance criteria**: + +1. **LIVE-01**: WHEN the Claude statusline receives a valid `session_id` and a non-empty `transcript_path` THEN it SHALL pass both values in the same `codedeck usage --json` invocation. +2. **LIVE-02**: WHEN the usage CLI receives `--transcript =` THEN it SHALL forward the native id and path in the `usage.get` request, preserving every path character after the first `=`. +3. **LIVE-03**: WHEN the daemon validates a transcript observation THEN it SHALL resolve the real path of `~/.claude/projects` and the supplied path, require the supplied path's resolved target to be a descendant of the resolved projects root, require the resolved basename `.jsonl`, and require a regular file. Symlink components are allowed when the resolved target passes these checks. +4. **LIVE-04**: IF the transcript parameter has the wrong shape, transcript path validation fails, or open fails THEN the daemon SHALL skip transcript reading and leave the in-memory cursor map and transcript token observations unchanged. A valid cost observation in the same request remains eligible for recording. +5. **LIVE-05**: IF a request contains cost and transcript observations with different native ids THEN the daemon SHALL reject the transcript observation and still apply the valid cost observation. +6. **LIVE-06**: WHEN the daemon reads a transcript for one native id during a `usage.get` request THEN it SHALL consume no more than 1,048,576 new transcript bytes for that native id. +7. **LIVE-07**: WHEN the read cap is reached before end of file THEN the reader SHALL retain the byte offset and parser continuation state in daemon memory, including an incomplete trailing line or the oversize-line skip flag, return totals for complete lines processed so far, and resume at that offset on the next request. After a daemon restart, it begins at byte zero; the usage ledger SHALL ignore lower partial observations until each reread field passes its high-water mark. +8. **LIVE-08**: WHEN an assistant usage line has both `message.id` and `requestId` THEN the reader SHALL add its token usage only once per native id while its in-memory state is live, including when a duplicate appears in a later read batch. +9. **LIVE-09**: WHEN a transcript batch is accepted THEN the daemon SHALL submit its cumulative token totals through `UsageLedger.observe` and advance the in-memory cursor only after the ledger transaction commits. If the transaction fails, it SHALL retain the prior cursor so the next request rereads the bytes. +10. **LIVE-10**: WHEN the transcript file's device or inode changes, or its size falls below the saved offset, THEN the reader SHALL reset the offset, pending line, and oversize-line skip state while preserving cumulative totals and seen message keys. +11. **LIVE-11**: IF `CODEDECK_RUN_ID` is set and the run summary is unavailable THEN the statusline SHALL omit `tok` instead of showing the current `context_window` token snapshot. +12. **LIVE-12**: WHEN live transcript tokens and release-time transcript tokens are observed for the same native id THEN the system SHALL use the same `claude-open:` source and retain each field's highest cumulative value, attributing only a positive difference above its high-water mark. +13. **LIVE-13**: WHEN the usage CLI returns a run summary with finite, non-negative `totalTokens` values for both the worker and orchestrator THEN the Claude statusline SHALL render ` tok` from `summary.totalTokens + summary.orchestrator.totalTokens`. IF either value is absent, nonnumeric, non-finite, or negative THEN it SHALL omit `tok` for that refresh. +14. **LIVE-14**: IF the usage CLI fails or reaches its one-second timeout THEN the statusline SHALL render the remaining valid local fields and exit with status 0. +15. **LIVE-15**: WHEN a `usage.get` request links a native id that has earlier unreconciled ids THEN the daemon SHALL return the live request without waiting for full transcript reads of those earlier ids. + +**Independent test**: Feed the statusline a fixture payload with `session_id` and `transcript_path`, then serve a fixture transcript in chunks. Verify the returned statusline total includes worker and orchestrator `totalTokens`, a repeated assistant line in a later chunk contributes once, an oversized partial line is dropped through its next newline, and a timed-out or malformed active-run query omits `tok` instead of falling back to the smaller context snapshot. Verify invalid worker or orchestrator `totalTokens` values omit `tok`, while a statusline without `CODEDECK_RUN_ID` retains its local token fallback. Reconcile a `cost-state` fixture whose input total is below the live input total and whose output total is above the live output total; verify the ledger retains the per-field maxima and adds only the output difference. Also verify an in-root symlink resolving to the expected transcript is accepted, a path resolving outside the root is rejected, malformed transcript parameters and path rejection preserve a valid cost observation, a malformed cost observation returns `INVALID` without processing a valid transcript parameter, mismatched native ids reject transcript ingestion, a device/inode change resets the cursor while preserving totals and seen keys, a daemon restart rereads from byte zero without lowering displayed totals, a failed ledger observation leaves the in-memory cursor unchanged, post-release requests skip transcript ingestion, in-flight ingestion is serialized before cleanup, every linked id is cleaned after successful reconciliation only when no active row uses it, failed reconciliation retains all cursors, and earlier full transcript reads do not block the live request. + +## Edge cases + +- An absent or empty `transcript_path` leaves the stored run summary in place. +- An incomplete final JSONL line remains pending up to 4 MiB. If it exceeds the cap, the reader drops the fragment and skips through its next newline. +- A complete invalid JSON line is skipped and advances the byte offset. +- Paths outside the resolved projects root, paths whose resolved basename does not match the native id, and non-regular files are rejected. A symlink path is accepted only when its resolved target passes the same checks. +- A first read larger than the byte cap returns totals for processed complete lines; later refreshes continue from the in-memory offset. +- After a daemon restart, the in-memory cursor begins at zero. A 30 MB transcript takes about one minute of two-second refreshes to reread. The persisted ledger keeps the displayed total from dropping while partial totals catch up. +- A same-device, same-inode rewrite whose size remains at least the saved offset is not detected by the stat checks. +- Replaying a partial batch after a failed ledger observation starts from the prior in-memory offset and does not skip unread bytes. +- A malformed present `observe` parameter keeps the existing strict `INVALID` response and prevents transcript processing in that request. +- Live transcript ingestion runs only for active Claude `open` rows. Recheck row status inside the per-native-id serializer so a post-release request skips ingestion and an in-flight read finishes before cleanup. +- After a successful full release reconciliation, consider cleanup for every id linked to that released session. If any reconciliation step fails, retain all cursors, including ids reconciled earlier in the pass. + +## Coverage matrix + +These are the focused implementation test targets. They are not run as part of this documentation change. + +| Code layer | Test type | Test file | Scoped Vitest command | +| --- | --- | --- | --- | +| Statusline script | Integration: invocation arguments, token rendering, timeout and invalid-total fallback | `tests/statusline.test.ts`, `tests/usage-statusline-contract.test.ts` | `npx vitest run tests/statusline.test.ts tests/usage-statusline-contract.test.ts` | +| Usage CLI | Integration: option parsing and IPC request contract | `tests/usage-cli.test.ts` | `npx vitest run tests/usage-cli.test.ts` | +| IPC protocol | Type and request contract through the usage CLI and daemon seam | `tests/usage-cli.test.ts`, `tests/orchestrator-usage-daemon.test.ts` | `npx vitest run tests/usage-cli.test.ts tests/orchestrator-usage-daemon.test.ts` | +| Daemon | Integration: path validation, chunk ingestion, live-to-release high-water reconciliation, reset, and failure behavior | `tests/orchestrator-usage-daemon.test.ts` | `npx vitest run tests/orchestrator-usage-daemon.test.ts` | +| Usage ledger | Integration: per-field high-water attribution | `tests/usage-ledger.test.ts` | `npx vitest run tests/usage-ledger.test.ts` | +| Incremental transcript reader | Unit: byte boundaries, partial-line cap, token mapping, and dedupe | `tests/claude-transcript.test.ts` | `npx vitest run tests/claude-transcript.test.ts` | + +## Requirement Traceability + +| Requirement ID | Story | Phase | Status | +| --- | --- | --- | --- | +| LIVE-01 | P1: Show live orchestrator tokens | Design, Tasks | Pending | +| LIVE-02 | P1: Show live orchestrator tokens | Design, Tasks | Pending | +| LIVE-03 | P1: Show live orchestrator tokens | Design, Tasks | Pending | +| LIVE-04 | P1: Show live orchestrator tokens | Design, Tasks | Pending | +| LIVE-05 | P1: Show live orchestrator tokens | Design, Tasks | Pending | +| LIVE-06 | P1: Show live orchestrator tokens | Design, Tasks | Pending | +| LIVE-07 | P1: Show live orchestrator tokens | Design, Tasks | Pending | +| LIVE-08 | P1: Show live orchestrator tokens | Design, Tasks | Pending | +| LIVE-09 | P1: Show live orchestrator tokens | Design, Tasks | Pending, rewritten for in-memory cursor ordering | +| LIVE-10 | P1: Show live orchestrator tokens | Design, Tasks | Pending | +| LIVE-11 | P1: Show live orchestrator tokens | Design, Tasks | Pending | +| LIVE-12 | P1: Show live orchestrator tokens | Design, Tasks | Pending | +| LIVE-13 | P1: Show live orchestrator tokens | Design, Tasks | Pending | +| LIVE-14 | P1: Show live orchestrator tokens | Design, Tasks | Pending | +| LIVE-15 | P1: Show live orchestrator tokens | Design, Tasks | Pending | + +**Coverage**: 15 total, all 15 mapped to Design and Tasks. LIVE-09 keeps its ID with the in-memory equivalent; no acceptance-criterion IDs were removed. + +## Success criteria + +- [ ] With a fixture transcript and no workers, the statusline shows the orchestrator's cumulative `totalTokens` instead of `0 tok`. +- [ ] With workers and an orchestrator, the displayed token total equals worker `totalTokens` plus orchestrator `totalTokens`, with cached tokens counted once. +- [ ] Replaying a transcript chunk, restarting the daemon, or reconciling at release never lowers totals or counts a keyed assistant usage record twice. +- [ ] Every `usage.get` transcript read consumes no more than 1,048,576 new bytes per native id. +- [ ] A timeout during an active run never replaces the aggregate token total with the smaller local context-window snapshot. + +## External dependencies + +| Resource | Identifier | System | Verified | Evidence | +| --- | --- | --- | --- | --- | +| Claude statusline payload field | `transcript_path` | Claude Code | yes | [Official statusline documentation](https://code.claude.com/docs/en/statusline), "Available data" lists the field and the full JSON example includes it. | +| Claude transcript location | `~/.claude/projects//.jsonl` | Claude Code | yes | [Official application-data documentation](https://code.claude.com/docs/en/claude-directory), which lists project transcripts below `projects/`. | +| Claude projects root | `~/.claude/projects` | Claude Code | yes | [Official application-data documentation](https://code.claude.com/docs/en/claude-directory), which lists transcripts at `projects//.jsonl`. | +| Claude live run id environment | `CODEDECK_RUN_ID` | repo | yes | `src/cli/commands/open.ts:924-934` passes the run id to Claude. | diff --git a/.specs/features/orchestrator-live-tokens/tasks.md b/.specs/features/orchestrator-live-tokens/tasks.md new file mode 100644 index 0000000..9bb59f7 --- /dev/null +++ b/.specs/features/orchestrator-live-tokens/tasks.md @@ -0,0 +1,238 @@ +# Orchestrator live tokens tasks + +## Execution protocol + +Implement each task with the `tlc-spec-driven` skill. Keep each task's source changes and tests together, run its scoped gate, and make one commit per task. Do not start a task until its dependencies are complete. + +--- + +**Spec**: `.specs/features/orchestrator-live-tokens/spec.md` +**Design**: `.specs/features/orchestrator-live-tokens/design.md` +**Status**: Draft + +## Test Coverage Matrix + +> Generated from `CLAUDE.md`, `package.json`, `vitest.config.ts`, sampled tests, and the approved feature spec. The project uses Vitest files under `tests/**/*.test.ts`. Every gate scopes Vitest to named files and runs the TypeScript check. The full test suite is not part of these task gates. + +| Code layer | Required test type | Coverage expectation | Test file | Run command | +| ---------- | ------------------ | -------------------- | ---------- | ----------- | +| Pure incremental transcript reader | Unit | Chunk boundary, keyed dedupe, invalid lines, 4 MiB partial-line cap, state immutability, and the 1 MiB input bound | `tests/claude-transcript.test.ts` | `npx vitest run tests/claude-transcript.test.ts` | +| Usage CLI | Integration | Valid and malformed `--transcript` values, first-equals parsing, optional cost observation, and exact `usage.get` params | `tests/usage-cli.test.ts` | `npx vitest run tests/usage-cli.test.ts` | +| IPC protocol | Integration | `usage.get` accepts the typed transcript parameter while preserving the existing cost observation | `tests/usage-cli.test.ts`, `tests/orchestrator-usage-daemon.test.ts` | `npx vitest run tests/usage-cli.test.ts tests/orchestrator-usage-daemon.test.ts` | +| Daemon live ingestion | Integration | Runtime parameter shapes, active-row checks, path containment and file type, symlink resolution, cost/transcript id mismatch, read cap, cursor resume/reset/cleanup, live-to-release high-water reconciliation, ledger failure, release overlap, and deferred earlier-id reads | `tests/orchestrator-usage-daemon.test.ts` | `npx vitest run tests/orchestrator-usage-daemon.test.ts` | +| Usage ledger high-water marks | Integration | Lower fields do not move marks; each higher field adds only its positive difference across rows | `tests/usage-ledger.test.ts` | `npx vitest run tests/usage-ledger.test.ts` | +| Statusline script | Integration | Exact CLI arguments, worker plus orchestrator token total, invalid summary token fields, active-run timeout behavior, and local fallback without a run id | `tests/statusline.test.ts`, `tests/usage-statusline-contract.test.ts` | `npx vitest run tests/statusline.test.ts tests/usage-statusline-contract.test.ts` | + +## Gate Check Commands + +| Gate | When to use | Command | +| ---- | ----------- | ------- | +| Task gate | After each task | `npx vitest run && npx tsc --noEmit -p .` | + +## Execution Plan + +Phase 1 tasks have disjoint files and can run in parallel. Phase 2 starts after both finish. Phase 3 follows the daemon integration. + +| Task | Deliverable | LIVE IDs | Files owned | Depends on | Parallel | +| ---- | ----------- | -------- | ----------- | ---------- | -------- | +| T1 | Pure incremental transcript reader | LIVE-07, LIVE-08 | `src/core/claude-transcript.ts`, `tests/claude-transcript.test.ts` | None | Yes, with T2 | +| T2 | Transcript CLI option and IPC parameter | LIVE-02 | `src/daemon/protocol.ts`, `src/cli/commands/usage.ts`, `tests/usage-cli.test.ts` | None | Yes, with T1 | +| T3 | Bounded daemon ingestion | LIVE-03, LIVE-04, LIVE-05, LIVE-06, LIVE-07, LIVE-08, LIVE-09, LIVE-10, LIVE-12, LIVE-15 | `src/daemon/daemon.ts`, `tests/orchestrator-usage-daemon.test.ts` | T1, T2 | No | +| T4 | Statusline transcript forwarding and combined tokens | LIVE-01, LIVE-11, LIVE-13, LIVE-14 | `plugin/statusline.sh`, `tests/statusline.test.ts`, `tests/usage-statusline-contract.test.ts` | T2, T3 | No | + +```text +Phase 1: T1 T2 +Phase 2: T1 -> T3 + T2 -> T3 +Phase 3: T2 -> T4 + T3 -> T4 +``` + +## Task Breakdown + +### Phase 1: Independent foundations + +#### T1: Add the pure incremental transcript reader + +**What**: Add a pure chunk consumer for live assistant token records while preserving the existing full-file release reader. +**Where**: `src/core/claude-transcript.ts` +**Files owned**: `src/core/claude-transcript.ts`, `tests/claude-transcript.test.ts` +**Depends on**: None +**Can run in parallel with**: T2 +**Requirement**: LIVE-07, LIVE-08 + +**Exported interface**: + +```typescript +export const TRANSCRIPT_CHUNK_LIMIT_BYTES = 1_048_576; +export const TRANSCRIPT_PENDING_LINE_LIMIT_BYTES = 4_194_304; + +export interface IncrementalTranscriptState { + pendingLine: Uint8Array; + discardUntilNewline: boolean; + inputTokens: number; + outputTokens: number; + cachedTokens: number; + seenMessageKeys: Set; +} + +export function consumeTranscriptChunk( + state: IncrementalTranscriptState, + bytes: Uint8Array, +): IncrementalTranscriptState; +``` + +**Done when**: + +- [ ] The consumer rejects a chunk over `TRANSCRIPT_CHUNK_LIMIT_BYTES` with `RangeError` and does not mutate its input state. +- [ ] It parses only complete newline-terminated JSONL records, skips invalid JSON, and resumes a pending line when the next chunk completes it. +- [ ] Assistant input, output, cache-read, and cache-creation values accumulate with the existing token mapping. +- [ ] A `JSON.stringify([message.id, requestId])` key contributes once across chunks; records missing either id keep the existing one-pass behavior. +- [ ] A partial line that grows beyond 4 MiB is dropped through its next newline, and a valid following line is counted. +- [ ] Tests for these outcomes are added to `tests/claude-transcript.test.ts` in this task. Existing release-reader cases keep passing. +- [ ] Gate passes: `npx vitest run tests/claude-transcript.test.ts && npx tsc --noEmit -p .` + +**Tests**: Unit tests in `tests/claude-transcript.test.ts`, written in this task. +**Gate**: `npx vitest run tests/claude-transcript.test.ts && npx tsc --noEmit -p .` +**Mutation probe**: Change the dedupe key to use only `message.id`. The test with one message id and two request ids must fail. +**Commit**: `feat(usage): Add the incremental transcript reader` + +--- + +#### T2: Forward transcript observations through usage.get + +**What**: Add the `--transcript =` CLI option and the optional transcript parameter to the existing usage request. +**Where**: `usage.get` request contract and usage option parsing +**Files owned**: `src/daemon/protocol.ts`, `src/cli/commands/usage.ts`, `tests/usage-cli.test.ts` +**Depends on**: None +**Can run in parallel with**: T1 +**Requirement**: LIVE-02 + +**Exact request parameter shape**: + +```typescript +export interface GetUsageRequest { + method: "usage.get"; + params: { + runId: string; + observe?: { nativeId: string; costUsd: number }; + transcript?: { nativeId: string; path: string }; + }; +} +``` + +**Done when**: + +- [ ] `GetUsageRequest.params` has exactly the existing `runId` and `observe` fields plus optional `transcript` in the shape above. +- [ ] The CLI parses `--transcript` by splitting at its first `=`, preserving additional equals signs in the path. +- [ ] A valid transcript flag is sent in the same `usage.get` request as a valid cost observation. A transcript-only request is also forwarded. +- [ ] A malformed transcript flag is omitted without suppressing a valid cost observation or the returned run summary. +- [ ] Tests for the request shape and parsing behavior are added to `tests/usage-cli.test.ts` in this task. +- [ ] Gate passes: `npx vitest run tests/usage-cli.test.ts && npx tsc --noEmit -p .` + +**Tests**: Integration tests in `tests/usage-cli.test.ts`, written in this task. +**Gate**: `npx vitest run tests/usage-cli.test.ts && npx tsc --noEmit -p .` +**Mutation probe**: Split the flag at its last `=` instead of its first. A path containing `=` must make the test fail. +**Commit**: `feat(usage): Forward transcript observations through usage.get` + +--- + +### Phase 2: Daemon integration + +#### T3: Ingest bounded live transcript chunks in the daemon + +**What**: Validate and read transcript observations in `usage.get`, keep reader state in daemon memory, and submit cumulative token totals to the existing ledger. +**Where**: Daemon `usage.get` handler +**Files owned**: `src/daemon/daemon.ts`, `tests/orchestrator-usage-daemon.test.ts` +**Depends on**: T1, T2 +**Requirement**: LIVE-03, LIVE-04, LIVE-05, LIVE-06, LIVE-07, LIVE-08, LIVE-09, LIVE-10, LIVE-12, LIVE-15 + +**Done when**: + +- [ ] A `Map` stores each native id's byte offset, parser state, device, and inode. The parser state includes pending bytes, the oversize-line flag, cumulative token totals, and the seen-key `Set`. +- [ ] A transcript parameter with a malformed runtime shape is ignored without rejecting the request or suppressing a valid cost observation. +- [ ] Process transcript observations only for an active Claude `open` row. Check the row status inside the per-native-id serializer before linking or reading; a terminal row skips transcript ingestion while preserving valid cost handling and the summary response. +- [ ] The daemon resolves the projects root and supplied path with `realpath`, requires the resolved transcript to be under the root with basename `.jsonl`, and rejects non-regular files. An in-root symlink resolving to the expected file is accepted; a symlink resolving outside is rejected. +- [ ] Validation or open failure leaves the cursor and transcript token observations unchanged. A valid cost observation still applies when transcript validation fails or cost and transcript ids differ. +- [ ] A malformed present cost observation returns `INVALID` before processing a valid transcript parameter, preserving the existing strict daemon behavior. +- [ ] Each request reads at most 1 MiB for one native id. It resumes from the in-memory offset, stores partial parser state, and serializes transcript reads, cursor replacement, and release cleanup for that id. +- [ ] A device/inode change or file size below the saved offset resets offset, pending bytes, and oversize-line state while preserving cumulative totals and seen keys. +- [ ] The ledger receives cumulative fields on `openSourceKey(nativeId)`. Commit the ledger transaction before replacing the in-memory cursor. If ledger persistence fails, keep the old cursor so a retry rereads the chunk. +- [ ] A new daemon starts with an empty cursor map and reads from byte zero. Partial totals below the persisted ledger high-water marks do not lower displayed totals; later chunks catch up and add only new differences. +- [ ] Linking a new id does not wait for earlier full transcript reads. Send the current live response first, then queue earlier-id reconciliation. Existing release/startup reconciliation can retry from durable link state. +- [ ] Make `reconcileOpenUsageSafely` return `true` when the requested reconciliation pass completes without throwing, including a vanished transcript marked `missing` or an already reconciled link, and `false` when reconciliation or ledger persistence throws. Only the full `session.release` pass may use a `true` result for cursor cleanup, not a selected earlier-id pass from `usage.get`. After a `true` full pass, queue cleanup for every id linked to that released session through its per-id serializer and remove each cursor only when no active Claude `open` row remains linked to it. If the full pass returns `false`, retain all cursors even if some links were reconciled before the failure. +- [ ] Tests cover path handling, malformed transcript and cost params, terminal-row and post-release requests, chunking, reset, in-flight release overlap, cleanup for multiple ids after successful reconciliation, retaining all cursors after failed reconciliation, a missing transcript terminal mark, ledger failure, restart, deferred reconciliation, and a live observation followed by release `cost-state` values with lower input and higher output, verifying only the output difference is attributed. Add them to `tests/orchestrator-usage-daemon.test.ts` in this task. Existing isolated high-water cases in `tests/usage-ledger.test.ts` remain part of the gate. +- [ ] Gate passes: `npx vitest run tests/orchestrator-usage-daemon.test.ts tests/usage-ledger.test.ts && npx tsc --noEmit -p .` + +**Tests**: Integration tests in `tests/orchestrator-usage-daemon.test.ts`, written in this task. Run the existing `tests/usage-ledger.test.ts` cases for the persisted high-water behavior. +**Gate**: `npx vitest run tests/orchestrator-usage-daemon.test.ts tests/usage-ledger.test.ts && npx tsc --noEmit -p .` +**Mutation probe**: Remove the resolved-path containment check. The outside-project transcript test must fail while its same-request valid cost observation still succeeds. +**Commit**: `feat(usage): Ingest live transcript token chunks` + +--- + +### Phase 3: Statusline + +#### T4: Forward the transcript path and display combined run tokens + +**What**: Pass the statusline transcript path to the usage CLI and render worker plus orchestrator token totals. +**Where**: `plugin/statusline.sh` +**Files owned**: `plugin/statusline.sh`, `tests/statusline.test.ts`, `tests/usage-statusline-contract.test.ts` +**Depends on**: T2, T3 +**Requirement**: LIVE-01, LIVE-11, LIVE-13, LIVE-14 + +**Done when**: + +- [ ] When `session_id` and `transcript_path` are valid and non-empty, the statusline appends `--transcript =` to its existing `usage --json` argument array. +- [ ] With a valid run summary, the `tok` field uses `summary.totalTokens + summary.orchestrator.totalTokens` and does not add `cachedTokens` a second time. +- [ ] If `CODEDECK_RUN_ID` is set and the usage call fails, times out, returns an invalid summary, or has an absent, nonnumeric, non-finite, or negative `totalTokens` value at either level, the statusline omits `tok`, keeps other valid local fields, and exits 0 without using context-window tokens. +- [ ] If both summary `totalTokens` values are finite and non-negative, the `tok` value is their exact sum; tests cover invalid worker and orchestrator values independently. +- [ ] With no `CODEDECK_RUN_ID`, the current local context-token fallback remains. +- [ ] The run summary contract tests include the merged worker and orchestrator `totalTokens` fields. Statusline and contract tests are updated in this task. +- [ ] Gate passes: `npx vitest run tests/statusline.test.ts tests/usage-statusline-contract.test.ts && npx tsc --noEmit -p .` + +**Tests**: Integration tests in `tests/statusline.test.ts` and `tests/usage-statusline-contract.test.ts`, written in this task. +**Gate**: `npx vitest run tests/statusline.test.ts tests/usage-statusline-contract.test.ts && npx tsc --noEmit -p .` +**Mutation probe**: Render only `summary.totalTokens` and omit `summary.orchestrator.totalTokens`. The combined-total fixture must make the test fail. +**Commit**: `feat(statusline): Display live orchestrator token totals` + +--- + +## Phase Execution Map + +Phases run in order. T1 and T2 own disjoint files and can run in parallel. T3 waits for both interfaces; T4 waits for the CLI, protocol, and daemon behavior. + +```text +Phase 1: T1 T2 +Phase 2: T1 -> T3 + T2 -> T3 +Phase 3: T2 -> T4 + T3 -> T4 +``` + +## Task Granularity Check + +| Task | Scope | Status | +| ---- | ----- | ------ | +| T1 | Pure chunk consumer in one module, with co-located unit tests | Granular | +| T2 | One request contract across protocol and CLI files, with co-located CLI tests | Cohesive, as required by the approved task shape | +| T3 | One daemon request path and its integration tests | Granular | +| T4 | One statusline behavior and its contract tests | Granular | + +## Diagram-Definition Cross-Check + +| Task | Depends on | Diagram shows | Status | +| ---- | ---------- | ------------- | ------ | +| T1 | None | None | Match | +| T2 | None | None | Match | +| T3 | T1, T2 | T1 -> T3; T2 -> T3 | Match | +| T4 | T2, T3 | T2 -> T4; T3 -> T4 | Match | + +## Test Co-location Validation + +| Task | Code layer | Matrix requires | Task says | Status | +| ---- | ---------- | --------------- | --------- | ------ | +| T1 | Pure incremental transcript reader | Unit | Unit tests in T1 | Match | +| T2 | Usage CLI and IPC request contract | Integration | Integration tests in T2 | Match | +| T3 | Daemon ingestion and usage ledger high-water behavior | Integration | Daemon tests in T3; existing ledger cases in its gate | Match | +| T4 | Statusline script and summary contract | Integration | Statusline and contract tests in T4 | Match | diff --git a/plugin/statusline.sh b/plugin/statusline.sh index e56ee95..a682793 100755 --- a/plugin/statusline.sh +++ b/plugin/statusline.sh @@ -112,9 +112,8 @@ const compactTokens = (value) => { /** * Since Claude Code 2.1.132, these fields describe the latest context window, - * not cumulative session totals. The aggregate token field therefore stays - * worker-only instead of adding a repeated local window to the run total. - * Degraded mode may show this current local snapshot when it is available. + * not cumulative session totals. Use the local snapshot only without an active + * run summary, which already combines worker and orchestrator token totals. */ const localTokens = () => { const context = payload.context_window; @@ -157,18 +156,20 @@ const localCost = () => { return typeof total === "number" && Number.isFinite(total) ? total : undefined; }; +const runId = text(process.env.CODEDECK_RUN_ID); + const getRunUsage = () => { - const runId = text(process.env.CODEDECK_RUN_ID); if (!runId) return undefined; const args = ["usage", runId, "--json"]; const sessionId = payload.session_id; + const validSessionId = + typeof sessionId === "string" && /^[0-9a-fA-F-]{8,}$/.test(sessionId); const observeCost = nonNegativeNumber(payload.cost?.total_cost_usd); - if ( - typeof sessionId === "string" && - /^[0-9a-fA-F-]{8,}$/.test(sessionId) && - observeCost !== undefined - ) args.push("--observe", `${sessionId}=${observeCost}`); + if (validSessionId && observeCost !== undefined) args.push("--observe", `${sessionId}=${observeCost}`); + if (validSessionId && typeof payload.transcript_path === "string" && payload.transcript_path.length > 0) { + args.push("--transcript", `${sessionId}=${payload.transcript_path}`); + } try { const output = execFileSync("codedeck", args, { @@ -228,12 +229,15 @@ const getOrchestratorUsage = (usage) => { const local = localCost(); const runUsage = getRunUsage(); const orchestratorUsage = runUsage ? getOrchestratorUsage(runUsage) : undefined; -const workerTokens = runUsage - ? nonNegativeNumber(runUsage.totalTokens) ?? - runUsage.inputTokens + runUsage.outputTokens + runUsage.cachedTokens +const workerTokens = runUsage ? nonNegativeNumber(runUsage.totalTokens) : undefined; +const orchestratorTokens = runUsage && isObject(runUsage.orchestrator) + ? nonNegativeNumber(runUsage.orchestrator.totalTokens) + : undefined; +const runTokens = workerTokens !== undefined && orchestratorTokens !== undefined + ? nonNegativeNumber(workerTokens + orchestratorTokens) : undefined; const tokenField = () => { - const total = workerTokens ?? (runUsage ? undefined : localTokens()); + const total = runId ? runTokens : localTokens(); return total === undefined ? undefined : paint(MUTED, compactTokens(total) + " tok"); }; diff --git a/src/cli/commands/usage.ts b/src/cli/commands/usage.ts index b026be4..56154b6 100644 --- a/src/cli/commands/usage.ts +++ b/src/cli/commands/usage.ts @@ -32,6 +32,7 @@ export interface UsageCommandOptions { watch?: boolean; interval?: string; observe?: string; + transcript?: string; backfill?: boolean; web?: boolean; port?: string; @@ -55,6 +56,17 @@ function parseUsageObservation(value: string | undefined): { nativeId: string; c return { nativeId, costUsd }; } +function parseTranscriptObservation(value: string | undefined): { nativeId: string; path: string } | undefined { + if (value === undefined) return undefined; + const separator = value.indexOf("="); + if (separator <= 0 || separator === value.length - 1) return undefined; + + const nativeId = value.slice(0, separator); + const path = value.slice(separator + 1); + if (!SESSION_ID_PATTERN.test(nativeId)) return undefined; + return { nativeId, path }; +} + export function formatUsageSummary(summary: RunUsageSummary): string { const cost = `$${summary.costUsd.toFixed(2)}${summary.costComplete ? "" : "?"}`; return `Run ${summary.runId}: ${summary.sessionCount} sessions, ${summary.inputTokens} input / ${summary.outputTokens} output / ${summary.cachedTokens} cached tokens, cost ${cost}`; @@ -136,6 +148,7 @@ export function registerUsageCommand(program: Command, dependencies: UsageComman .option("-a, --agent ", "filter by agent harness (e.g. codex, claude)") .option("--by ", "group by dimension: day, repo, model, agent, run, origin") .option("--observe ", "report live orchestrator cost") + .option("--transcript ", "report live orchestrator transcript tokens") .option("--backfill", "import historical orchestrator usage") .option("--web", "open aggregate usage in the browser") .option("--port ", "port to listen on (default: 3100)", String(DEFAULT_WEB_PORT)) @@ -180,9 +193,11 @@ export function registerUsageCommand(program: Command, dependencies: UsageComman let summary: RunUsageSummary; try { const observe = parseUsageObservation(opts.observe); + const transcript = parseTranscriptObservation(opts.transcript); summary = await client.request("usage.get", { runId: targetRunId, ...(observe ? { observe } : {}), + ...(transcript ? { transcript } : {}), }); } catch (error) { console.error(error instanceof Error ? error.message : String(error)); diff --git a/src/core/claude-transcript.ts b/src/core/claude-transcript.ts index e9becad..880d3b4 100644 --- a/src/core/claude-transcript.ts +++ b/src/core/claude-transcript.ts @@ -4,6 +4,9 @@ import path from "node:path"; import { createInterface } from "node:readline"; import { computeSessionCost, type SessionCostUsage } from "./pricing.js"; +export const TRANSCRIPT_CHUNK_LIMIT_BYTES = 1_048_576; +export const TRANSCRIPT_PENDING_LINE_LIMIT_BYTES = 4_194_304; + export interface TranscriptUsage { state: "cost-state" | "tokens" | "no-price"; cost?: number; @@ -23,6 +26,15 @@ interface UsageTotals extends SessionCostUsage { cachedTokens: number; } +export interface IncrementalTranscriptState { + pendingLine: Uint8Array; + discardUntilNewline: boolean; + inputTokens: number; + outputTokens: number; + cachedTokens: number; + seenMessageKeys: Set; +} + function isRecord(value: unknown): value is JsonRecord { return typeof value === "object" && value !== null && !Array.isArray(value); } @@ -39,6 +51,96 @@ function emptyUsage(): UsageTotals { return { inputTokens: 0, outputTokens: 0, cachedTokens: 0 }; } +function appendBytes(left: Uint8Array, right: Uint8Array): Uint8Array { + const result = new Uint8Array(left.byteLength + right.byteLength); + result.set(left); + result.set(right, left.byteLength); + return result; +} + +function consumeAssistantLine(line: Uint8Array, state: IncrementalTranscriptState): void { + let parsed: unknown; + try { + parsed = JSON.parse(new TextDecoder().decode(line)); + } catch { + return; + } + if (!isRecord(parsed) || parsed.type !== "assistant" || !isRecord(parsed.message) || !isRecord(parsed.message.usage)) { + return; + } + + const usage = parsed.message.usage; + const inputTokens = tokenCount(usage.input_tokens); + const outputTokens = tokenCount(usage.output_tokens); + const cacheReadTokens = tokenCount(usage.cache_read_input_tokens); + const cacheCreationTokens = tokenCount(usage.cache_creation_input_tokens); + if (inputTokens === 0 && outputTokens === 0 && cacheReadTokens === 0 && cacheCreationTokens === 0) return; + + const messageId = parsed.message.id; + const requestId = parsed.requestId; + if (messageId != null && requestId != null) { + const key = JSON.stringify([messageId, requestId]); + if (state.seenMessageKeys.has(key)) return; + state.seenMessageKeys.add(key); + } + + state.inputTokens += inputTokens; + state.outputTokens += outputTokens; + state.cachedTokens += cacheReadTokens + cacheCreationTokens; +} + +export function consumeTranscriptChunk( + state: IncrementalTranscriptState, + bytes: Uint8Array, +): IncrementalTranscriptState { + if (bytes.byteLength > TRANSCRIPT_CHUNK_LIMIT_BYTES) { + throw new RangeError(`Transcript chunk exceeds ${TRANSCRIPT_CHUNK_LIMIT_BYTES} bytes`); + } + + const nextState: IncrementalTranscriptState = { + ...state, + pendingLine: state.pendingLine.slice(), + seenMessageKeys: new Set(state.seenMessageKeys), + }; + let offset = 0; + + while (offset < bytes.byteLength) { + if (nextState.discardUntilNewline) { + let newline = offset; + while (newline < bytes.byteLength && bytes[newline] !== 0x0a) newline += 1; + if (newline === bytes.byteLength) break; + nextState.discardUntilNewline = false; + offset = newline + 1; + continue; + } + + let newline = offset; + while (newline < bytes.byteLength && bytes[newline] !== 0x0a) newline += 1; + const complete = newline < bytes.byteLength; + const end = complete ? newline : bytes.byteLength; + const fragment = bytes.subarray(offset, end); + const lineLength = nextState.pendingLine.byteLength + fragment.byteLength; + + if (lineLength > TRANSCRIPT_PENDING_LINE_LIMIT_BYTES) { + nextState.pendingLine = new Uint8Array(); + if (!complete) nextState.discardUntilNewline = true; + } else if (complete) { + const line = nextState.pendingLine.byteLength === 0 + ? fragment + : appendBytes(nextState.pendingLine, fragment); + consumeAssistantLine(line, nextState); + nextState.pendingLine = new Uint8Array(); + } else { + nextState.pendingLine = appendBytes(nextState.pendingLine, fragment); + } + + if (!complete) break; + offset = newline + 1; + } + + return nextState; +} + function usageFromCostState(value: JsonRecord): TranscriptUsage { const totals = emptyUsage(); const modelUsage = isRecord(value.modelUsage) ? value.modelUsage : {}; diff --git a/src/daemon/daemon.ts b/src/daemon/daemon.ts index 6e8fd09..79fbc8b 100644 --- a/src/daemon/daemon.ts +++ b/src/daemon/daemon.ts @@ -11,7 +11,7 @@ import { getPaths, ensureDirs } from "../config/paths.js"; import { createIpcServer } from "./ipc.js"; import type { IpcRequest, IpcResponse, UsageQueryParams } from "./protocol.js"; import { getRegistry } from "../drivers/registry.js"; -import { isTerminalStatus, liveStatus, normalizeAgentId, type AgentId, type Session, type SessionStatus } from "../core/session.js"; +import { isActiveStatus, isTerminalStatus, liveStatus, normalizeAgentId, type AgentId, type Session, type SessionStatus } from "../core/session.js"; import { parseSandbox, type AgentDriver, type CodexSandbox, type DriverSession } from "../core/driver.js"; import { generateSessionId, generateBranchName } from "../core/session.js"; import { getGitInfo, getBaseCommit } from "../git/repository.js"; @@ -27,7 +27,13 @@ import { aggregateRunUsage } from "../core/run-usage.js"; import { UsageLedger } from "../store/usage-ledger.js"; import { openSourceKey, workerSourceKey } from "../core/usage-source.js"; import { NativeLinkStore } from "../store/native-links.js"; -import { findTranscript, readTranscriptUsage } from "../core/claude-transcript.js"; +import { + consumeTranscriptChunk, + findTranscript, + readTranscriptUsage, + TRANSCRIPT_CHUNK_LIMIT_BYTES, + type IncrementalTranscriptState, +} from "../core/claude-transcript.js"; // Daemon's view of power readiness for the doctor IPC result (field names // fixed by cross-worker contract; the CLI falls back to local detection @@ -60,6 +66,12 @@ function withLiveStatus(s: Session): Session { // Send-queue cap: bytes, matching MAX_SEND_BODY_BYTES on the web route. const MAX_SEND_MESSAGE_BYTES = 64 * 1024; +interface LiveTranscriptCursor extends IncrementalTranscriptState { + byteOffset: number; + device: number; + inode: number; +} + function validateSendMessage(raw: unknown): { message: string } | { code: string; error: string } { if (typeof raw !== "string") return { code: "INVALID", error: "message required" }; const message = raw.trim(); @@ -104,6 +116,167 @@ class Daemon { private inhibitChild: ChildProcess | null = null; private inhibitExitHookInstalled = false; private inFlightModels = new Map>(); + private inFlightOpenUsageReconciliations = new Map>(); + private liveTranscriptCursors = new Map(); + private liveTranscriptQueues = new Map>(); + + private async withLiveTranscriptLock(nativeId: string, operation: () => Promise): Promise { + const previous = this.liveTranscriptQueues.get(nativeId) ?? Promise.resolve(); + let unlock!: () => void; + const current = new Promise((resolve) => { unlock = resolve; }); + this.liveTranscriptQueues.set(nativeId, current); + + await previous; + try { + return await operation(); + } finally { + unlock(); + if (this.liveTranscriptQueues.get(nativeId) === current) { + this.liveTranscriptQueues.delete(nativeId); + } + } + } + + private async ingestLiveTranscript( + sessionId: string, + nativeId: string, + suppliedPath: string, + ): Promise { + return this.withLiveTranscriptLock(nativeId, async () => { + const isActiveClaudeOpenRow = () => { + const session = this.sessions.get(sessionId); + return session?.origin === "open" && session.agent === "claude" && isActiveStatus(session.status); + }; + if (!isActiveClaudeOpenRow()) return []; + + let resolvedPath: string; + let stat: fs.Stats; + try { + const projectsRoot = await fs.promises.realpath(path.join(os.homedir(), ".claude", "projects")); + resolvedPath = await fs.promises.realpath(suppliedPath); + const relativePath = path.relative(projectsRoot, resolvedPath); + if ( + relativePath === "" || + relativePath === ".." || + relativePath.startsWith(`..${path.sep}`) || + path.isAbsolute(relativePath) || + path.basename(resolvedPath) !== `${nativeId}.jsonl` + ) return []; + + stat = await fs.promises.stat(resolvedPath); + if (!stat.isFile()) return []; + } catch { + return []; + } + + let file: fs.promises.FileHandle; + try { + file = await fs.promises.open(resolvedPath, "r"); + } catch { + return []; + } + + try { + if (!isActiveClaudeOpenRow()) return []; + + let previous: string[] = []; + try { + previous = this.nativeLinks.link(sessionId, nativeId).previous; + + const saved = this.liveTranscriptCursors.get(nativeId); + const reset = saved !== undefined && + (saved.device !== stat.dev || saved.inode !== stat.ino || stat.size < saved.byteOffset); + const state: IncrementalTranscriptState = { + pendingLine: reset ? new Uint8Array() : saved?.pendingLine ?? new Uint8Array(), + discardUntilNewline: reset ? false : saved?.discardUntilNewline ?? false, + inputTokens: saved?.inputTokens ?? 0, + outputTokens: saved?.outputTokens ?? 0, + cachedTokens: saved?.cachedTokens ?? 0, + seenMessageKeys: saved?.seenMessageKeys ?? new Set(), + }; + const byteOffset = reset ? 0 : saved?.byteOffset ?? 0; + const bytesToRead = Math.min(TRANSCRIPT_CHUNK_LIMIT_BYTES, Math.max(0, stat.size - byteOffset)); + const buffer = Buffer.alloc(bytesToRead); + const { bytesRead } = bytesToRead === 0 + ? { bytesRead: 0 } + : await file.read(buffer, 0, bytesToRead, byteOffset); + const nextState = consumeTranscriptChunk(state, buffer.subarray(0, bytesRead)); + if (!isActiveClaudeOpenRow()) return previous; + + const db = this.db.getHandle(); + db.exec("BEGIN"); + try { + this.usageLedger.observe(sessionId, openSourceKey(nativeId), { + inputTokens: nextState.inputTokens, + outputTokens: nextState.outputTokens, + cachedTokens: nextState.cachedTokens, + }); + db.exec("COMMIT"); + } catch { + try { db.exec("ROLLBACK"); } catch {} + return previous; + } + this.liveTranscriptCursors.set(nativeId, { + ...nextState, + byteOffset: byteOffset + bytesRead, + device: stat.dev, + inode: stat.ino, + }); + return previous; + } catch { return previous; } + } finally { + await file.close().catch(() => {}); + } + }); + } + + private queueOpenUsageReconciliation(sessionId: string, nativeIds: readonly string[]): void { + const uniqueIds = [...new Set(nativeIds)]; + const keyFor = (nativeId: string) => JSON.stringify([sessionId, nativeId]); + const pendingIds = uniqueIds.filter((nativeId) => + !this.inFlightOpenUsageReconciliations.has(keyFor(nativeId))); + if (pendingIds.length === 0) return; + + const keys = pendingIds.map(keyFor); + const reconciliation = Promise.resolve().then(() => + this.reconcileOpenUsageSafely(sessionId, pendingIds)); + for (const key of keys) this.inFlightOpenUsageReconciliations.set(key, reconciliation); + + const clearInFlight = () => { + for (const key of keys) { + if (this.inFlightOpenUsageReconciliations.get(key) === reconciliation) { + this.inFlightOpenUsageReconciliations.delete(key); + } + } + }; + void reconciliation.then(clearInFlight, clearInFlight); + } + + private async cleanupReleasedTranscriptCursors(sessionId: string): Promise { + const nativeIds = new Set(this.nativeLinks.linksFor([sessionId]).map((link) => link.nativeId)); + await Promise.all([...nativeIds].map((nativeId) => this.withLiveTranscriptLock(nativeId, async () => { + const activeClaudeOpenIds = this.sessions.listActive() + .filter((session) => session.origin === "open" && session.agent === "claude") + .map((session) => session.id); + const stillActive = this.nativeLinks.linksFor(activeClaudeOpenIds) + .some((link) => link.nativeId === nativeId); + if (!stillActive) this.liveTranscriptCursors.delete(nativeId); + }))); + } + + private queueReleasedTranscriptCursorCleanup(sessionId: string): void { + queueMicrotask(() => { + void this.cleanupReleasedTranscriptCursors(sessionId).catch((error) => { + const detail = error instanceof Error ? error.message : String(error); + try { + fs.appendFileSync( + getPaths().daemonLog, + `[${new Date().toISOString()}] usage cursor cleanup failed session=${sessionId}: ${detail}\n`, + ); + } catch {} + }); + }); + } private async fetchModels(agent?: AgentId, refresh?: boolean): Promise { const key = `${agent || "all"}:${Boolean(refresh)}`; @@ -378,9 +551,10 @@ class Daemon { } } - private async reconcileOpenUsageSafely(sessionId: string, nativeIds?: readonly string[]): Promise { + private async reconcileOpenUsageSafely(sessionId: string, nativeIds?: readonly string[]): Promise { try { await this.reconcileOpenUsage(sessionId, nativeIds); + return true; } catch (error) { const detail = error instanceof Error ? error.message : String(error); try { @@ -389,6 +563,7 @@ class Daemon { `[${new Date().toISOString()}] usage reconcile failed session=${sessionId}: ${detail}\n`, ); } catch {} + return false; } } @@ -679,7 +854,8 @@ class Daemon { if (typeof p.nativeSessionId === "string" && p.nativeSessionId.length > 0) { this.nativeLinks.link(s.id, p.nativeSessionId); } - await this.reconcileOpenUsageSafely(s.id); + const reconciled = await this.reconcileOpenUsageSafely(s.id); + if (reconciled) this.queueReleasedTranscriptCursorCleanup(s.id); } const ev: AgentEvent = targetStatus === "failed" @@ -1085,14 +1261,23 @@ class Daemon { case "usage.get": { const p = (params || {}) as { runId?: unknown; - observe?: { nativeId?: unknown; costUsd?: unknown }; + observe?: unknown; + transcript?: unknown; }; if (typeof p.runId !== "string" || p.runId.length === 0) { send({ error: { code: "INVALID", message: "runId required" } }); return; } + + let observation: { nativeId: string; costUsd: number } | undefined; if (p.observe !== undefined) { - const { nativeId, costUsd } = p.observe; + const raw = p.observe; + const nativeId = typeof raw === "object" && raw !== null && !Array.isArray(raw) + ? (raw as { nativeId?: unknown }).nativeId + : undefined; + const costUsd = typeof raw === "object" && raw !== null && !Array.isArray(raw) + ? (raw as { costUsd?: unknown }).costUsd + : undefined; if ( typeof nativeId !== "string" || nativeId.length === 0 || @@ -1103,28 +1288,53 @@ class Daemon { send({ error: { code: "INVALID", message: "observe requires nativeId and a finite non-negative costUsd" } }); return; } - const session = this.sessions.getByRunId(p.runId) - .find((candidate) => candidate.id === p.runId && candidate.origin === "open"); - if (session) { - const db = this.db.getHandle(); - db.exec("BEGIN"); - let previous: string[] = []; - try { - previous = this.nativeLinks.link(session.id, nativeId).previous; - this.usageLedger.observe(session.id, openSourceKey(nativeId), { cost: costUsd }); - db.exec("COMMIT"); - } catch (error) { - try { db.exec("ROLLBACK"); } catch {} - throw error; - } - if (previous.length > 0) await this.reconcileOpenUsageSafely(session.id, previous); + observation = { nativeId, costUsd }; + } + + const session = this.sessions.getByRunId(p.runId) + .find((candidate) => candidate.id === p.runId && candidate.origin === "open"); + const earlierNativeIds = new Set(); + if (session && observation) { + const db = this.db.getHandle(); + db.exec("BEGIN"); + try { + const { previous } = this.nativeLinks.link(session.id, observation.nativeId); + previous.forEach((nativeId) => earlierNativeIds.add(nativeId)); + this.usageLedger.observe(session.id, openSourceKey(observation.nativeId), { cost: observation.costUsd }); + db.exec("COMMIT"); + } catch (error) { + try { db.exec("ROLLBACK"); } catch {} + throw error; } } + + const rawTranscript = p.transcript; + const transcript = typeof rawTranscript === "object" && rawTranscript !== null && !Array.isArray(rawTranscript) + ? rawTranscript as { nativeId?: unknown; path?: unknown } + : undefined; + if ( + session && + session.agent === "claude" && + isActiveStatus(session.status) && + transcript && + typeof transcript.nativeId === "string" && + transcript.nativeId.length > 0 && + typeof transcript.path === "string" && + transcript.path.length > 0 && + (!observation || observation.nativeId === transcript.nativeId) + ) { + const previous = await this.ingestLiveTranscript(session.id, transcript.nativeId, transcript.path); + previous.forEach((nativeId) => earlierNativeIds.add(nativeId)); + } + const sessions = this.sessions.getByRunId(p.runId); const sessionIds = sessions.map((session) => session.id); const attributions = this.usageLedger.attributionsFor(sessionIds); const linkStates = this.nativeLinks.linksFor(sessionIds).map(({ sessionId, state }) => ({ sessionId, state })); send({ result: aggregateRunUsage(p.runId, sessions, attributions, linkStates) }); + if (session && earlierNativeIds.size > 0) { + this.queueOpenUsageReconciliation(session.id, [...earlierNativeIds]); + } break; } diff --git a/src/daemon/protocol.ts b/src/daemon/protocol.ts index 83cf83c..6cc9752 100644 --- a/src/daemon/protocol.ts +++ b/src/daemon/protocol.ts @@ -163,6 +163,7 @@ export interface GetUsageRequest { params: { runId: string; observe?: { nativeId: string; costUsd: number }; + transcript?: { nativeId: string; path: string }; }; } diff --git a/tests/claude-transcript.test.ts b/tests/claude-transcript.test.ts index b4e0696..d054b58 100644 --- a/tests/claude-transcript.test.ts +++ b/tests/claude-transcript.test.ts @@ -5,7 +5,13 @@ import os from "node:os"; import path from "node:path"; import { fileURLToPath } from "node:url"; import { afterEach, describe, expect, it } from "vitest"; -import { findTranscript, readTranscriptUsage } from "../src/core/claude-transcript.js"; +import { + consumeTranscriptChunk, + findTranscript, + readTranscriptUsage, + TRANSCRIPT_CHUNK_LIMIT_BYTES, + type IncrementalTranscriptState, +} from "../src/core/claude-transcript.js"; const fixturePath = (name: string): string => fileURLToPath(new URL(`./fixtures/claude-transcript/${name}`, import.meta.url)); @@ -18,10 +24,164 @@ function makeTempDir(): string { return dir; } +function emptyIncrementalState(): IncrementalTranscriptState { + return { + pendingLine: new Uint8Array(), + discardUntilNewline: false, + inputTokens: 0, + outputTokens: 0, + cachedTokens: 0, + seenMessageKeys: new Set(), + }; +} + +function assistantLine(options: { + messageId?: string; + requestId?: string; + inputTokens?: number; + outputTokens?: number; + cacheReadTokens?: number; + cacheCreationTokens?: number; +}): string { + return `${JSON.stringify({ + type: "assistant", + ...(options.requestId === undefined ? {} : { requestId: options.requestId }), + message: { + ...(options.messageId === undefined ? {} : { id: options.messageId }), + usage: { + input_tokens: options.inputTokens ?? 0, + output_tokens: options.outputTokens ?? 0, + cache_read_input_tokens: options.cacheReadTokens ?? 0, + cache_creation_input_tokens: options.cacheCreationTokens ?? 0, + }, + }, + })}\n`; +} + afterEach(() => { for (const dir of tempDirs.splice(0)) fs.rmSync(dir, { recursive: true, force: true }); }); +describe("incremental Claude transcript usage", () => { + it("rejects an oversized chunk without mutating the input state", () => { + const pendingLine = new TextEncoder().encode("partial"); + const seenMessageKeys = new Set(["seen"]); + const state: IncrementalTranscriptState = { + pendingLine, + discardUntilNewline: true, + inputTokens: 4, + outputTokens: 5, + cachedTokens: 6, + seenMessageKeys, + }; + + expect(() => consumeTranscriptChunk(state, new Uint8Array(TRANSCRIPT_CHUNK_LIMIT_BYTES + 1))).toThrow( + RangeError, + ); + expect(state).toEqual({ + pendingLine: new TextEncoder().encode("partial"), + discardUntilNewline: true, + inputTokens: 4, + outputTokens: 5, + cachedTokens: 6, + seenMessageKeys: new Set(["seen"]), + }); + expect(state.pendingLine).toBe(pendingLine); + expect(state.seenMessageKeys).toBe(seenMessageKeys); + }); + + it("counts only complete lines, resumes pending JSON, and skips invalid JSON", () => { + const firstLine = assistantLine({ messageId: "message-1", requestId: "request-1", inputTokens: 2 }); + const secondLine = assistantLine({ messageId: "message-2", requestId: "request-2", inputTokens: 3 }); + const incompleteSecondLine = secondLine.slice(0, -4); + const initial = emptyIncrementalState(); + + const partial = consumeTranscriptChunk(initial, new TextEncoder().encode(firstLine + incompleteSecondLine)); + + expect(partial.inputTokens).toBe(2); + expect(new TextDecoder().decode(partial.pendingLine)).toBe(incompleteSecondLine); + expect(initial.inputTokens).toBe(0); + expect(initial.pendingLine).toEqual(new Uint8Array()); + + const completed = consumeTranscriptChunk( + partial, + new TextEncoder().encode(`${secondLine.slice(incompleteSecondLine.length)}not json\n`), + ); + + expect(completed.inputTokens).toBe(5); + expect(completed.pendingLine).toEqual(new Uint8Array()); + }); + + it("counts each message id and request id pair once across chunks", () => { + const firstRequest = assistantLine({ + messageId: "message-1", + requestId: "request-1", + inputTokens: 5, + outputTokens: 2, + cacheReadTokens: 4, + cacheCreationTokens: 3, + }); + const secondRequest = assistantLine({ + messageId: "message-1", + requestId: "request-2", + inputTokens: 7, + outputTokens: 4, + cacheReadTokens: 1, + cacheCreationTokens: 5, + }); + + const firstChunk = consumeTranscriptChunk(emptyIncrementalState(), new TextEncoder().encode(firstRequest)); + const totals = consumeTranscriptChunk( + firstChunk, + new TextEncoder().encode(firstRequest + secondRequest), + ); + + expect(totals).toMatchObject({ inputTokens: 12, outputTokens: 6, cachedTokens: 13 }); + expect(totals.seenMessageKeys).toEqual(new Set(['["message-1","request-1"]', '["message-1","request-2"]'])); + }); + + it("keeps counting assistant records when either dedupe id is missing", () => { + const withoutRequestId = assistantLine({ messageId: "message-1", inputTokens: 2 }); + const withoutMessageId = assistantLine({ requestId: "request-1", inputTokens: 3 }); + + const totals = consumeTranscriptChunk( + emptyIncrementalState(), + new TextEncoder().encode(withoutRequestId + withoutRequestId + withoutMessageId + withoutMessageId), + ); + + expect(totals.inputTokens).toBe(10); + expect(totals.seenMessageKeys.size).toBe(0); + }); + + it("drops an oversized partial line through its newline and counts the next line", () => { + let state = emptyIncrementalState(); + const largeFragment = new Uint8Array(TRANSCRIPT_CHUNK_LIMIT_BYTES).fill(0x78); + + for (let index = 0; index < 4; index += 1) { + state = consumeTranscriptChunk(state, largeFragment); + } + expect(state.pendingLine.byteLength).toBe(4 * TRANSCRIPT_CHUNK_LIMIT_BYTES); + + state = consumeTranscriptChunk(state, new TextEncoder().encode("x")); + expect(state).toMatchObject({ + discardUntilNewline: true, + inputTokens: 0, + pendingLine: new Uint8Array(), + }); + + state = consumeTranscriptChunk( + state, + new TextEncoder().encode(`\n${assistantLine({ messageId: "message-1", requestId: "request-1", inputTokens: 9 })}`), + ); + + expect(state).toMatchObject({ + discardUntilNewline: false, + inputTokens: 9, + pendingLine: new Uint8Array(), + }); + }); +}); + describe("Claude transcript usage", () => { it("uses the last cost-state and sums its model token totals", async () => { await expect(readTranscriptUsage(fixturePath("cost-state.jsonl"))).resolves.toEqual({ diff --git a/tests/orchestrator-usage-daemon.test.ts b/tests/orchestrator-usage-daemon.test.ts index d2c46d7..fbfea6b 100644 --- a/tests/orchestrator-usage-daemon.test.ts +++ b/tests/orchestrator-usage-daemon.test.ts @@ -88,6 +88,39 @@ function installTranscript(nativeId: string, fixture: string): void { ); } +function assistantLine( + messageId: string, + requestId: string, + inputTokens: number, + outputTokens: number, + cachedTokens = 0, + paddingBytes = 0, +): string { + return JSON.stringify({ + type: "assistant", + requestId, + message: { + id: messageId, + usage: { + input_tokens: inputTokens, + output_tokens: outputTokens, + cache_read_input_tokens: cachedTokens, + }, + }, + ...(paddingBytes > 0 ? { padding: "x".repeat(paddingBytes) } : {}), + }); +} + +function liveCursor(nativeId: string): Record | undefined { + return (daemon as any).liveTranscriptCursors.get(nativeId); +} + +function deferred(): { promise: Promise; resolve: () => void } { + let resolve!: () => void; + const promise = new Promise((done) => { resolve = done; }); + return { promise, resolve }; +} + describe("orchestrator usage daemon methods", () => { it("links multiple native ids to an open row and reports whether each link was created", async () => { seedOpen("open-link-row"); @@ -471,4 +504,502 @@ describe("orchestrator usage daemon methods", () => { expect(secondTotal).toBe(firstTotal); expect(secondTotal).toBe(6); }); + + it("accepts an in-project symlink and rejects an outside-project symlink", async () => { + seedOpen("row-path-validation"); + const insideId = "native-inside-link"; + const insidePath = transcriptFile(insideId); + fs.writeFileSync(insidePath, `${assistantLine("message-inside", "request-inside", 7, 3)}\n`); + const aliasPath = path.join(path.dirname(insidePath), "transcript-alias.jsonl"); + fs.symlinkSync(insidePath, aliasPath); + + const inside = await request("usage.get", { + runId: "row-path-validation", + transcript: { nativeId: insideId, path: aliasPath }, + }); + expect(inside.result.orchestrator).toMatchObject({ inputTokens: 7, outputTokens: 3, totalTokens: 10 }); + expect(liveCursor(insideId)?.byteOffset).toBeGreaterThan(0); + + const outsideId = "native-outside-link"; + const outsideDir = path.join(homeDir, "outside-projects"); + fs.mkdirSync(outsideDir, { recursive: true }); + const outsidePath = path.join(outsideDir, `${outsideId}.jsonl`); + fs.writeFileSync(outsidePath, `${assistantLine("message-outside", "request-outside", 90, 40)}\n`); + const outsideAlias = path.join(path.dirname(insidePath), "outside-transcript-alias.jsonl"); + fs.symlinkSync(outsidePath, outsideAlias); + const outside = await request("usage.get", { + runId: "row-path-validation", + observe: { nativeId: outsideId, costUsd: 1.5 }, + transcript: { nativeId: outsideId, path: outsideAlias }, + }); + + expect(outside.result.orchestrator.costUsd).toBe(1.5); + expect(outside.result.orchestrator.inputTokens).toBe(7); + expect(liveCursor(outsideId)).toBeUndefined(); + expect((daemon as any).usageLedger.attributionsFor(["row-path-validation"])) + .toContainEqual(expect.objectContaining({ sourceKey: `claude-open:${outsideId}`, cost: 1.5, inputTokens: 0 })); + + const wrongBasenameId = "native-wrong-basename"; + const wrongBasenamePath = path.join(path.dirname(insidePath), "wrong-basename.jsonl"); + fs.writeFileSync(wrongBasenamePath, `${assistantLine("message-wrong-basename", "request-wrong-basename", 60, 20)}\n`); + await request("usage.get", { + runId: "row-path-validation", + transcript: { nativeId: wrongBasenameId, path: wrongBasenamePath }, + }); + expect(liveCursor(wrongBasenameId)).toBeUndefined(); + expect((daemon as any).nativeLinks.linksFor(["row-path-validation"])) + .not.toContainEqual(expect.objectContaining({ nativeId: wrongBasenameId })); + + const directoryId = "native-directory-path"; + const directoryPath = transcriptFile(directoryId); + fs.mkdirSync(directoryPath); + await request("usage.get", { + runId: "row-path-validation", + transcript: { nativeId: directoryId, path: directoryPath }, + }); + expect(liveCursor(directoryId)).toBeUndefined(); + expect((daemon as any).nativeLinks.linksFor(["row-path-validation"])) + .not.toContainEqual(expect.objectContaining({ nativeId: directoryId })); + + const retryId = "native-open-retry"; + const retryPath = transcriptFile(retryId); + const firstLine = assistantLine("message-open-first", "request-open-first", 4, 2); + const secondLine = assistantLine("message-open-second", "request-open-second", 6, 3); + fs.writeFileSync(retryPath, `${firstLine}\n`); + await request("usage.get", { runId: "row-path-validation", transcript: { nativeId: retryId, path: retryPath } }); + const oldCursor = { ...liveCursor(retryId) }; + fs.appendFileSync(retryPath, `${secondLine}\n`); + + const originalOpen = fs.promises.open.bind(fs.promises); + fs.promises.open = (async (...args: Parameters) => { + if (args[0] === retryPath) throw new Error("injected open failure"); + return originalOpen(...args); + }) as typeof fs.promises.open; + try { + const failedOpen = await request("usage.get", { + runId: "row-path-validation", + transcript: { nativeId: retryId, path: retryPath }, + }); + expect(failedOpen.result.orchestrator).toMatchObject({ inputTokens: 11, outputTokens: 5 }); + expect(liveCursor(retryId)).toMatchObject({ + byteOffset: oldCursor.byteOffset, + inputTokens: oldCursor.inputTokens, + outputTokens: oldCursor.outputTokens, + }); + } finally { + fs.promises.open = originalOpen; + } + const retriedOpen = await request("usage.get", { + runId: "row-path-validation", + transcript: { nativeId: retryId, path: retryPath }, + }); + expect(retriedOpen.result.orchestrator).toMatchObject({ inputTokens: 17, outputTokens: 8 }); + }); + + it("ignores malformed transcripts, rejects malformed costs first, and rejects mismatched ids", async () => { + seedOpen("row-malformed-observations"); + const malformedTranscriptId = "native-malformed-transcript"; + const malformedTranscriptPath = transcriptFile(malformedTranscriptId); + fs.writeFileSync(malformedTranscriptPath, `${assistantLine("message-malformed", "request-malformed", 8, 4)}\n`); + + const costWithMalformedTranscript = await request("usage.get", { + runId: "row-malformed-observations", + observe: { nativeId: "native-cost-with-malformed-transcript", costUsd: 2.25 }, + transcript: { nativeId: malformedTranscriptId, path: 27 }, + }); + expect(costWithMalformedTranscript.result.orchestrator.costUsd).toBe(2.25); + expect(liveCursor(malformedTranscriptId)).toBeUndefined(); + + const invalidCost = await request("usage.get", { + runId: "row-malformed-observations", + observe: null, + transcript: { nativeId: malformedTranscriptId, path: malformedTranscriptPath }, + }); + expect(invalidCost.error?.code).toBe("INVALID"); + expect(liveCursor(malformedTranscriptId)).toBeUndefined(); + expect((daemon as any).nativeLinks.linksFor(["row-malformed-observations"])) + .not.toContainEqual(expect.objectContaining({ nativeId: malformedTranscriptId })); + + const mismatchId = "native-transcript-mismatch"; + const mismatchPath = transcriptFile(mismatchId); + fs.writeFileSync(mismatchPath, `${assistantLine("message-mismatch", "request-mismatch", 11, 9)}\n`); + const mismatch = await request("usage.get", { + runId: "row-malformed-observations", + observe: { nativeId: "native-cost-mismatch", costUsd: 3 }, + transcript: { nativeId: mismatchId, path: mismatchPath }, + }); + expect(mismatch.result.orchestrator.costUsd).toBe(5.25); + expect(liveCursor(mismatchId)).toBeUndefined(); + expect((daemon as any).nativeLinks.linksFor(["row-malformed-observations"])) + .toContainEqual(expect.objectContaining({ nativeId: "native-cost-mismatch" })); + }); + + it("skips transcript ingestion for terminal rows while keeping valid cost observations", async () => { + seedOpen("row-terminal-live"); + const nativeId = "native-terminal-live"; + const file = transcriptFile(nativeId); + fs.writeFileSync(file, `${assistantLine("message-terminal", "request-terminal", 14, 6)}\n`); + seam(daemon!).sessions.setStatus("row-terminal-live", "completed"); + + const response = await request("usage.get", { + runId: "row-terminal-live", + observe: { nativeId, costUsd: 4 }, + transcript: { nativeId, path: file }, + }); + + expect(response.result.orchestrator).toMatchObject({ costUsd: 4, inputTokens: 0, outputTokens: 0 }); + expect(liveCursor(nativeId)).toBeUndefined(); + }); + + it("does not recreate a cursor after release", async () => { + seedOpen("row-post-release-live"); + const nativeId = "native-post-release-live"; + const file = transcriptFile(nativeId); + fs.writeFileSync(file, `${assistantLine("message-post-release", "request-post-release", 12, 5)}\n`); + await request("usage.get", { + runId: "row-post-release-live", + transcript: { nativeId, path: file }, + }); + expect(liveCursor(nativeId)).toBeDefined(); + + await request("session.release", { id: "row-post-release-live" }); + await new Promise((resolve) => setImmediate(resolve)); + expect(liveCursor(nativeId)).toBeUndefined(); + + const response = await request("usage.get", { + runId: "row-post-release-live", + observe: { nativeId, costUsd: 2 }, + transcript: { nativeId, path: file }, + }); + expect(response.result.orchestrator).toMatchObject({ costUsd: 2, inputTokens: 12, outputTokens: 5 }); + expect(liveCursor(nativeId)).toBeUndefined(); + }); + + it("reads at most one MiB and carries an incomplete line to the next request", async () => { + seedOpen("row-live-chunk"); + const nativeId = "native-live-chunk"; + const file = transcriptFile(nativeId); + const line = assistantLine("message-chunk", "request-chunk", 21, 9, 4, 1_048_600); + fs.writeFileSync(file, `${line}\n`); + + const first = await request("usage.get", { + runId: "row-live-chunk", + transcript: { nativeId, path: file }, + }); + expect(first.result.orchestrator.totalTokens).toBe(0); + expect(liveCursor(nativeId)).toMatchObject({ + byteOffset: 1_048_576, + pendingLine: expect.any(Uint8Array), + inputTokens: 0, + outputTokens: 0, + }); + expect(liveCursor(nativeId)?.pendingLine.byteLength).toBe(1_048_576); + + const second = await request("usage.get", { + runId: "row-live-chunk", + transcript: { nativeId, path: file }, + }); + expect(second.result.orchestrator).toMatchObject({ inputTokens: 21, outputTokens: 9, cachedTokens: 4, totalTokens: 34 }); + expect(liveCursor(nativeId)?.byteOffset).toBe(Buffer.byteLength(`${line}\n`)); + }); + + it("resets pending parser state after truncation and preserves totals and seen keys", async () => { + seedOpen("row-live-reset"); + const nativeId = "native-live-reset"; + const file = transcriptFile(nativeId); + const duplicate = assistantLine("message-reset-duplicate", "request-reset-duplicate", 10, 4); + fs.writeFileSync(file, `${duplicate}\n${"x".repeat(1_100_000)}`); + await request("usage.get", { runId: "row-live-reset", transcript: { nativeId, path: file } }); + expect(liveCursor(nativeId)).toMatchObject({ byteOffset: 1_048_576, inputTokens: 10, outputTokens: 4 }); + + const added = assistantLine("message-reset-added", "request-reset-added", 5, 3); + const replacement = `${duplicate}\n${added}\n`; + fs.writeFileSync(file, replacement); + const afterTruncate = await request("usage.get", { + runId: "row-live-reset", + transcript: { nativeId, path: file }, + }); + expect(afterTruncate.result.orchestrator).toMatchObject({ inputTokens: 15, outputTokens: 7 }); + expect(liveCursor(nativeId)?.pendingLine.byteLength).toBe(0); + expect(liveCursor(nativeId)?.discardUntilNewline).toBe(false); + + const replaced = `${duplicate}\n${assistantLine("message-new-inode", "request-new-inode", 6, 2)}\n${"z".repeat(500)}`; + const replacementPath = `${file}.replacement`; + fs.writeFileSync(replacementPath, replaced); + fs.renameSync(replacementPath, file); + const afterReplace = await request("usage.get", { + runId: "row-live-reset", + transcript: { nativeId, path: file }, + }); + expect(afterReplace.result.orchestrator).toMatchObject({ inputTokens: 21, outputTokens: 9 }); + }); + + it("serializes an in-flight transcript read with release cursor cleanup", async () => { + seedOpen("row-release-overlap"); + const nativeId = "native-release-overlap"; + const file = transcriptFile(nativeId); + fs.writeFileSync(file, `${assistantLine("message-overlap-first", "request-overlap-first", 5, 2)}\n`); + await request("usage.get", { runId: "row-release-overlap", transcript: { nativeId, path: file } }); + fs.appendFileSync(file, `${assistantLine("message-overlap-second", "request-overlap-second", 7, 3)}\n`); + + const opened = deferred(); + const allowRead = deferred(); + const originalOpen = fs.promises.open.bind(fs.promises); + fs.promises.open = (async (...args: Parameters) => { + const handle = await originalOpen(...args); + if (args[0] === file) { + const originalRead = handle.read.bind(handle); + handle.read = (async (...readArgs: Parameters) => { + opened.resolve(); + await allowRead.promise; + return originalRead(...readArgs); + }) as typeof handle.read; + } + return handle; + }) as typeof fs.promises.open; + + try { + const liveRequest = request("usage.get", { + runId: "row-release-overlap", + transcript: { nativeId, path: file }, + }); + await opened.promise; + const releaseRequest = request("session.release", { id: "row-release-overlap" }); + expect(seam(daemon!).sessions.get("row-release-overlap")?.status).toBe("completed"); + const released = await releaseRequest; + expect(released.result.session.status).toBe("completed"); + allowRead.resolve(); + await liveRequest; + await new Promise((resolve) => setImmediate(resolve)); + } finally { + allowRead.resolve(); + fs.promises.open = originalOpen; + } + + expect(liveCursor(nativeId)).toBeUndefined(); + expect(seam(daemon!).sessions.get("row-release-overlap")?.usage) + .toMatchObject({ inputTokens: 12, outputTokens: 5 }); + }); + + it("cleans every released cursor after successful reconciliation and keeps a shared active id", async () => { + seedOpen("row-release-cleanup"); + seedOpen("row-shared-cleanup"); + const ids = ["native-cleanup-one", "native-cleanup-two", "native-cleanup-shared"]; + for (const [index, nativeId] of ids.entries()) { + const file = transcriptFile(nativeId); + fs.writeFileSync(file, `${assistantLine(`message-cleanup-${index}`, `request-cleanup-${index}`, 1, 1)}\n`); + await request("usage.get", { runId: "row-release-cleanup", transcript: { nativeId, path: file } }); + } + await request("session.linkNative", { id: "row-shared-cleanup", nativeId: ids[2] }); + + await request("session.release", { id: "row-release-cleanup" }); + await new Promise((resolve) => setImmediate(resolve)); + + expect(liveCursor(ids[0])).toBeUndefined(); + expect(liveCursor(ids[1])).toBeUndefined(); + expect(liveCursor(ids[2])).toBeDefined(); + }); + + it("retains every cursor when a full release reconciliation fails", async () => { + seedOpen("row-release-reconcile-failure"); + const ids = ["native-a-reconcile-good", "native-z-reconcile-fails"]; + for (const [index, nativeId] of ids.entries()) { + const file = transcriptFile(nativeId); + fs.writeFileSync(file, `${assistantLine(`message-failure-${index}`, `request-failure-${index}`, 2, 1)}\n`); + await request("usage.get", { runId: "row-release-reconcile-failure", transcript: { nativeId, path: file } }); + } + + const ledger = (daemon as any).usageLedger; + const originalObserve = ledger.observe.bind(ledger); + ledger.observe = (sessionId: string, sourceKey: string, observation: unknown, seedIntoIncoming?: boolean) => { + if (sourceKey === `claude-open:${ids[1]}`) throw new Error("injected ledger failure"); + return originalObserve(sessionId, sourceKey, observation, seedIntoIncoming); + }; + try { + await request("session.release", { id: "row-release-reconcile-failure" }); + } finally { + ledger.observe = originalObserve; + } + + expect(liveCursor(ids[0])).toBeDefined(); + expect(liveCursor(ids[1])).toBeDefined(); + expect((daemon as any).nativeLinks.unreconciled("row-release-reconcile-failure")).toHaveLength(1); + }); + + it("marks a vanished transcript missing and cleans its cursor after release", async () => { + seedOpen("row-missing-live-transcript"); + const nativeId = "native-missing-live-transcript"; + const file = transcriptFile(nativeId); + fs.writeFileSync(file, `${assistantLine("message-missing-live", "request-missing-live", 3, 2)}\n`); + await request("usage.get", { runId: "row-missing-live-transcript", transcript: { nativeId, path: file } }); + expect(liveCursor(nativeId)).toBeDefined(); + fs.unlinkSync(file); + + await request("session.release", { id: "row-missing-live-transcript" }); + await new Promise((resolve) => setImmediate(resolve)); + + expect((daemon as any).nativeLinks.linksFor(["row-missing-live-transcript"])) + .toMatchObject([{ nativeId, state: "missing" }]); + expect(liveCursor(nativeId)).toBeUndefined(); + }); + + it("keeps the cursor unchanged when ledger persistence fails, then rereads on retry", async () => { + seedOpen("row-live-ledger-failure"); + const earlierId = "native-earlier-ledger-failure"; + await request("session.linkNative", { id: "row-live-ledger-failure", nativeId: earlierId }); + const nativeId = "native-live-ledger-failure"; + const file = transcriptFile(nativeId); + fs.writeFileSync(file, `${assistantLine("message-ledger-failure", "request-ledger-failure", 13, 7)}\n`); + const ledger = (daemon as any).usageLedger; + const originalObserve = ledger.observe.bind(ledger); + const reconciliationQueued = deferred(); + let queuedSessionId: string | undefined; + let queuedNativeIds: readonly string[] | undefined; + const originalReconcile = (daemon as any).reconcileOpenUsageSafely.bind(daemon); + (daemon as any).reconcileOpenUsageSafely = async (sessionId: string, nativeIds?: readonly string[]) => { + queuedSessionId = sessionId; + queuedNativeIds = nativeIds; + reconciliationQueued.resolve(); + return true; + }; + ledger.observe = () => { throw new Error("injected live ledger failure"); }; + try { + const failed = await request("usage.get", { runId: "row-live-ledger-failure", transcript: { nativeId, path: file } }); + await reconciliationQueued.promise; + expect(failed.result.runId).toBe("row-live-ledger-failure"); + expect(liveCursor(nativeId)).toBeUndefined(); + expect((daemon as any).usageLedger.attributionsFor(["row-live-ledger-failure"])).toEqual([]); + expect(queuedSessionId).toBe("row-live-ledger-failure"); + expect(queuedNativeIds).toEqual([earlierId]); + } finally { + ledger.observe = originalObserve; + (daemon as any).reconcileOpenUsageSafely = originalReconcile; + } + + const retried = await request("usage.get", { runId: "row-live-ledger-failure", transcript: { nativeId, path: file } }); + expect(retried.result.orchestrator).toMatchObject({ inputTokens: 13, outputTokens: 7 }); + expect(liveCursor(nativeId)?.byteOffset).toBe(Buffer.byteLength(`${assistantLine("message-ledger-failure", "request-ledger-failure", 13, 7)}\n`)); + }); + + it("restarts at byte zero without lowering high-water totals and catches up later chunks", async () => { + seedOpen("row-live-restart"); + const nativeId = "native-live-restart"; + const file = transcriptFile(nativeId); + const firstLine = assistantLine("message-restart-first", "request-restart-first", 10, 4); + const secondLine = assistantLine("message-restart-second", "request-restart-second", 20, 8, 0, 1_048_600); + fs.writeFileSync(file, `${firstLine}\n${secondLine}\n`); + + await request("usage.get", { runId: "row-live-restart", transcript: { nativeId, path: file } }); + expect(seam(daemon!).sessions.get("row-live-restart")?.usage).toMatchObject({ inputTokens: 10, outputTokens: 4 }); + await request("usage.get", { runId: "row-live-restart", transcript: { nativeId, path: file } }); + expect(seam(daemon!).sessions.get("row-live-restart")?.usage).toMatchObject({ inputTokens: 30, outputTokens: 12 }); + + seam(daemon!).db.close(); + daemon = new Daemon(); + expect(liveCursor(nativeId)).toBeUndefined(); + const afterRestartFirst = await request("usage.get", { runId: "row-live-restart", transcript: { nativeId, path: file } }); + expect(afterRestartFirst.result.orchestrator).toMatchObject({ inputTokens: 30, outputTokens: 12 }); + expect(liveCursor(nativeId)?.byteOffset).toBe(1_048_576); + expect(seam(daemon!).sessions.get("row-live-restart")?.usage).toMatchObject({ inputTokens: 30, outputTokens: 12 }); + + await request("usage.get", { runId: "row-live-restart", transcript: { nativeId, path: file } }); + fs.appendFileSync(file, `${assistantLine("message-restart-later", "request-restart-later", 5, 2)}\n`); + const caughtUp = await request("usage.get", { runId: "row-live-restart", transcript: { nativeId, path: file } }); + expect(caughtUp.result.orchestrator).toMatchObject({ inputTokens: 35, outputTokens: 14 }); + expect(seam(daemon!).sessions.get("row-live-restart")?.usage).toMatchObject({ inputTokens: 35, outputTokens: 14 }); + }); + + it("deduplicates deferred earlier-id reconciliation while it is in flight", async () => { + seedOpen("row-deferred-reconcile"); + await request("session.linkNative", { id: "row-deferred-reconcile", nativeId: "native-earlier-unreconciled" }); + const liveId = "native-current-live"; + const file = transcriptFile(liveId); + fs.writeFileSync(file, `${assistantLine("message-current-live", "request-current-live", 9, 4)}\n`); + + const reconciliationStarted = deferred(); + const finishFirstReconciliation = deferred(); + const secondReconciliationStarted = deferred(); + const reconciliations: Array<{ sessionId: string; nativeIds?: readonly string[] }> = []; + const originalReconcile = (daemon as any).reconcileOpenUsageSafely.bind(daemon); + (daemon as any).reconcileOpenUsageSafely = async (sessionId: string, nativeIds?: readonly string[]) => { + reconciliations.push({ sessionId, nativeIds }); + if (reconciliations.length === 1) { + reconciliationStarted.resolve(); + await finishFirstReconciliation.promise; + } + if (reconciliations.length === 2) secondReconciliationStarted.resolve(); + return true; + }; + try { + const response = await request("usage.get", { + runId: "row-deferred-reconcile", + transcript: { nativeId: liveId, path: file }, + }); + expect(response.result.orchestrator).toMatchObject({ inputTokens: 9, outputTokens: 4 }); + await reconciliationStarted.promise; + expect(reconciliations[0]).toEqual({ + sessionId: "row-deferred-reconcile", + nativeIds: ["native-earlier-unreconciled"], + }); + + const secondResponse = await request("usage.get", { + runId: "row-deferred-reconcile", + transcript: { nativeId: liveId, path: file }, + }); + expect(secondResponse.result.orchestrator).toMatchObject({ inputTokens: 9, outputTokens: 4 }); + expect(reconciliations).toHaveLength(1); + + finishFirstReconciliation.resolve(); + await new Promise((resolve) => setImmediate(resolve)); + + const thirdResponse = await request("usage.get", { + runId: "row-deferred-reconcile", + transcript: { nativeId: liveId, path: file }, + }); + expect(thirdResponse.result.orchestrator).toMatchObject({ inputTokens: 9, outputTokens: 4 }); + await secondReconciliationStarted.promise; + expect(reconciliations).toHaveLength(2); + expect(reconciliations[1]).toEqual({ + sessionId: "row-deferred-reconcile", + nativeIds: ["native-earlier-unreconciled"], + }); + } finally { + finishFirstReconciliation.resolve(); + (daemon as any).reconcileOpenUsageSafely = originalReconcile; + } + }); + + it("attributes only release output growth above live input and output marks", async () => { + seedOpen("row-live-release-high-water"); + const nativeId = "native-live-release-high-water"; + const file = transcriptFile(nativeId); + fs.writeFileSync(file, `${assistantLine("message-live-high-water", "request-live-high-water", 100, 20)}\n`); + await request("usage.get", { runId: "row-live-release-high-water", transcript: { nativeId, path: file } }); + expect(seam(daemon!).sessions.get("row-live-release-high-water")?.usage) + .toMatchObject({ inputTokens: 100, outputTokens: 20 }); + + fs.writeFileSync(file, `${JSON.stringify({ + type: "cost-state", + totalCostUSD: 5, + modelUsage: { + "claude-sonnet-4-5": { + inputTokens: 80, + outputTokens: 45, + cacheReadInputTokens: 0, + cacheCreationInputTokens: 0, + costUSD: 5, + }, + }, + })}\n`); + await request("session.release", { id: "row-live-release-high-water" }); + + expect(seam(daemon!).sessions.get("row-live-release-high-water")?.usage) + .toMatchObject({ inputTokens: 100, outputTokens: 45 }); + expect((daemon as any).usageLedger.attributionsFor(["row-live-release-high-water"])) + .toContainEqual(expect.objectContaining({ + sourceKey: `claude-open:${nativeId}`, + inputTokens: 100, + outputTokens: 45, + })); + }); }); diff --git a/tests/statusline.test.ts b/tests/statusline.test.ts index 5c612fd..e251619 100644 --- a/tests/statusline.test.ts +++ b/tests/statusline.test.ts @@ -14,12 +14,23 @@ interface RenderOptions { payload: Record; runId?: string; usage?: Record; + usageOutput?: string; shimExitCode?: number; + shimDelaySeconds?: number; sessionId?: string; taskName?: string; } -async function render({ payload, runId, usage, shimExitCode = 0, sessionId, taskName }: RenderOptions): Promise<{ +async function render({ + payload, + runId, + usage, + usageOutput, + shimExitCode = 0, + shimDelaySeconds, + sessionId, + taskName, +}: RenderOptions): Promise<{ output: string; args: string[]; exitCode: number | null; @@ -33,7 +44,8 @@ async function render({ payload, runId, usage, shimExitCode = 0, sessionId, task [ "#!/bin/sh", 'printf "%s\\n" "$@" > "$CODEDECK_SHIM_ARGS"', - `printf '%s\\n' '${JSON.stringify(usage ?? {})}'`, + ...(shimDelaySeconds === undefined ? [] : [`sleep ${shimDelaySeconds}`]), + `printf '%s\\n' '${usageOutput ?? JSON.stringify(usage ?? {})}'`, `exit ${shimExitCode}`, "", ].join("\n"), @@ -103,11 +115,21 @@ describe("Claude statusline", () => { inputTokens: 1200, outputTokens: 800, cachedTokens: 300, + totalTokens: 2300, costUsd: 0.4, sessionCount: 2, activeSessionCount: 2, costComplete: true, sessionsWithoutCost: 0, + orchestrator: { + costUsd: 0, + costComplete: true, + inputTokens: 0, + outputTokens: 0, + cachedTokens: 0, + totalTokens: 0, + sources: [], + }, }, }); @@ -120,8 +142,9 @@ describe("Claude statusline", () => { it("reports the local orchestrator cost without counting its live source twice", async () => { const sessionId = "92d88cce-bdbc-46db-8573-916afd32f6f7"; const otherSourceId = "3f1f93b8-c484-43aa-8a11-32a486109e22"; + const transcriptPath = "/home/user/.claude/projects/project path/transcript=part.jsonl"; const result = await render({ - payload: { ...payload(1), session_id: sessionId }, + payload: { ...payload(1), session_id: sessionId, transcript_path: transcriptPath }, sessionId, runId: "run-example", usage: { @@ -129,6 +152,7 @@ describe("Claude statusline", () => { inputTokens: 1200, outputTokens: 800, cachedTokens: 300, + totalTokens: 120, costUsd: 0.5, sessionCount: 2, activeSessionCount: 2, @@ -140,6 +164,7 @@ describe("Claude statusline", () => { inputTokens: 0, outputTokens: 0, cachedTokens: 0, + totalTokens: 90, sources: [ { nativeId: otherSourceId, costUsd: 3 }, { nativeId: sessionId, costUsd: 0.8 }, @@ -149,6 +174,8 @@ describe("Claude statusline", () => { }, }); + expect(stripAnsi(result.output)).toContain("210 tok"); + expect(stripAnsi(result.output)).not.toContain("510 tok"); expect(stripAnsi(result.output)).toContain("run $4.50"); expect(result.args).toEqual([ "usage", @@ -156,6 +183,8 @@ describe("Claude statusline", () => { "--json", "--observe", `${sessionId}=1`, + "--transcript", + `${sessionId}=${transcriptPath}`, ]); }); @@ -167,12 +196,20 @@ describe("Claude statusline", () => { runId: "run-example", }); const invalidId = await render({ - payload: { ...payload(1), session_id: "ses_invalid" }, + payload: { + ...payload(1), + session_id: "ses_invalid", + transcript_path: "/tmp/ses_invalid.jsonl", + }, sessionId: "ses_invalid", runId: "run-example", }); const negativeCost = await render({ - payload: { ...payload(-1), session_id: sessionId }, + payload: { + ...payload(-1), + session_id: sessionId, + transcript_path: "/tmp/session.jsonl", + }, sessionId, runId: "run-example", }); @@ -185,7 +222,30 @@ describe("Claude statusline", () => { `${sessionId}=0`, ]); expect(invalidId.args).toEqual(["usage", "run-example", "--json"]); - expect(negativeCost.args).toEqual(["usage", "run-example", "--json"]); + expect(negativeCost.args).toEqual([ + "usage", + "run-example", + "--json", + "--transcript", + `${sessionId}=/tmp/session.jsonl`, + ]); + }); + + it("skips transcript forwarding when the transcript path is empty", async () => { + const sessionId = "92d88cce-bdbc-46db-8573-916afd32f6f7"; + const result = await render({ + payload: { ...payload(0), session_id: sessionId, transcript_path: "" }, + sessionId, + runId: "run-example", + }); + + expect(result.args).toEqual([ + "usage", + "run-example", + "--json", + "--observe", + `${sessionId}=0`, + ]); }); it("omits a present task name sidecar and starts with the role", async () => { @@ -244,16 +304,52 @@ describe("Claude statusline", () => { expect(stripAnsi(result.output)).toBe(`builder · ${project}/main · ctx 68% · 2k tok · $0.25`); }); - it("keeps the local token snapshot when the usage CLI fails", async () => { + it("omits run tokens when the usage CLI fails", async () => { const result = await render({ payload: payload(1, { total_input_tokens: 1_200, total_output_tokens: 800 }), runId: "run-unavailable", shimExitCode: 1, }); - expect(stripAnsi(result.output)).toBe(`builder · ${project}/main · ctx 68% · 2k tok · $1.00`); + expect(stripAnsi(result.output)).toBe(`builder · ${project}/main · ctx 68% · $1.00`); + expect(result.exitCode).toBe(0); expect(stripAnsi(result.output)).not.toContain(" · run "); - expect(stripAnsi(result.output)).not.toContain("agents"); + expect(stripAnsi(result.output)).not.toContain(" tok"); + }); + + it("omits run tokens when the usage CLI times out", async () => { + const result = await render({ + payload: payload(1, { total_input_tokens: 1_200, total_output_tokens: 800 }), + runId: "run-timeout", + shimDelaySeconds: 2, + }); + + expect(stripAnsi(result.output)).toBe(`builder · ${project}/main · ctx 68% · $1.00`); + expect(result.exitCode).toBe(0); + expect(stripAnsi(result.output)).not.toContain(" tok"); + }); + + it("omits run tokens when the usage summary is invalid", async () => { + const result = await render({ + payload: payload(1, { total_input_tokens: 1_200, total_output_tokens: 800 }), + runId: "run-invalid-summary", + usage: { + runId: "another-run", + inputTokens: 100, + outputTokens: 100, + cachedTokens: 0, + totalTokens: 200, + costUsd: 0.25, + sessionCount: 1, + activeSessionCount: 0, + costComplete: true, + sessionsWithoutCost: 0, + }, + }); + + expect(stripAnsi(result.output)).toBe(`builder · ${project}/main · ctx 68% · $1.00`); + expect(result.exitCode).toBe(0); + expect(stripAnsi(result.output)).not.toContain(" tok"); }); it("formats token totals compactly", async () => { @@ -265,11 +361,21 @@ describe("Claude statusline", () => { inputTokens: 1_000_000, outputTokens: 234_567, cachedTokens: 0, + totalTokens: 1_234_567, costUsd: 0, sessionCount: 1, activeSessionCount: 1, costComplete: true, sessionsWithoutCost: 0, + orchestrator: { + costUsd: 0, + costComplete: true, + inputTokens: 0, + outputTokens: 0, + cachedTokens: 0, + totalTokens: 0, + sources: [], + }, }, }); const thousands = await render({ @@ -284,7 +390,7 @@ describe("Claude statusline", () => { expect(stripAnsi(small.output)).toContain("980 tok"); }); - it("uses a valid run totalTokens value for the tok field", async () => { + it("adds worker and orchestrator totals without counting cached tokens again", async () => { const result = await render({ payload: payload(0), runId: "run-cached-tokens", @@ -292,40 +398,79 @@ describe("Claude statusline", () => { runId: "run-cached-tokens", inputTokens: 1_000, outputTokens: 100, - cachedTokens: 800, - totalTokens: 1_100, + cachedTokens: 80, + totalTokens: 110, costUsd: 0, sessionCount: 1, activeSessionCount: 1, costComplete: true, sessionsWithoutCost: 0, + orchestrator: { + costUsd: 0, + costComplete: true, + inputTokens: 0, + outputTokens: 0, + cachedTokens: 70, + totalTokens: 50, + sources: [], + }, }, }); - expect(stripAnsi(result.output)).toContain("1.1k tok"); - expect(stripAnsi(result.output)).not.toContain("1.9k tok"); + expect(stripAnsi(result.output)).toContain("160 tok"); + expect(stripAnsi(result.output)).not.toContain("310 tok"); }); - it("falls back to the token sum when totalTokens is invalid", async () => { - const result = await render({ - payload: payload(0), - runId: "run-invalid-token-total", - usage: { - runId: "run-invalid-token-total", - inputTokens: 1_000, - outputTokens: 100, - cachedTokens: 800, - totalTokens: -1, - costUsd: 0, - sessionCount: 1, - activeSessionCount: 1, - costComplete: true, - sessionsWithoutCost: 0, + const invalidRunTokenTotals = [ + { label: "missing", value: undefined }, + { label: "nonnumeric", value: "1200" }, + { label: "non-finite", value: "__statusline_non_finite__" }, + { label: "negative", value: -1 }, + ]; + + for (const level of ["worker", "orchestrator"] as const) { + it.each(invalidRunTokenTotals)( + `omits tok when the ${level} totalTokens value is $label`, + async ({ value }) => { + const usage: Record & { orchestrator: Record } = { + runId: "run-invalid-token-total", + inputTokens: 1_000, + outputTokens: 100, + cachedTokens: 800, + totalTokens: 1_100, + costUsd: 0, + sessionCount: 1, + activeSessionCount: 0, + costComplete: true, + sessionsWithoutCost: 0, + orchestrator: { + costUsd: 0, + costComplete: true, + inputTokens: 0, + outputTokens: 0, + cachedTokens: 700, + totalTokens: 500, + sources: [], + }, + }; + if (level === "worker") usage.totalTokens = value; + else usage.orchestrator.totalTokens = value; + const usageOutput = JSON.stringify(usage).replaceAll( + '"__statusline_non_finite__"', + "1e9999", + ); + const result = await render({ + payload: payload(1, { total_input_tokens: 1_200, total_output_tokens: 800 }), + runId: "run-invalid-token-total", + usageOutput, + }); + + expect(stripAnsi(result.output)).not.toContain(" tok"); + expect(stripAnsi(result.output)).toContain("run $1.00"); + expect(result.exitCode).toBe(0); }, - }); - - expect(stripAnsi(result.output)).toContain("1.9k tok"); - }); + ); + } it("marks a partial aggregate even when the local cost is zero", async () => { const result = await render({ @@ -336,11 +481,21 @@ describe("Claude statusline", () => { inputTokens: 0, outputTokens: 0, cachedTokens: 0, + totalTokens: 0, costUsd: 0.42, sessionCount: 1, activeSessionCount: 0, costComplete: false, sessionsWithoutCost: 1, + orchestrator: { + costUsd: 0, + costComplete: true, + inputTokens: 0, + outputTokens: 0, + cachedTokens: 0, + totalTokens: 0, + sources: [], + }, }, }); @@ -387,6 +542,7 @@ describe("Claude statusline", () => { inputTokens: 0, outputTokens: 0, cachedTokens: 0, + totalTokens: 120, costUsd: 0.4, sessionCount: 1, activeSessionCount: 0, @@ -398,11 +554,13 @@ describe("Claude statusline", () => { inputTokens: 0, outputTokens: 0, cachedTokens: 0, + totalTokens: 90, sources: [{ nativeId: sessionId, costUsd: "invalid" }], }, }, }); + expect(stripAnsi(result.output)).toContain("210 tok"); expect(stripAnsi(result.output)).toContain("run $0.65"); expect(stripAnsi(result.output)).not.toContain("run $0.65?"); }); diff --git a/tests/usage-cli.test.ts b/tests/usage-cli.test.ts index 1b838fe..37a9662 100644 --- a/tests/usage-cli.test.ts +++ b/tests/usage-cli.test.ts @@ -163,6 +163,76 @@ describe("usage CLI", () => { expect(JSON.parse(logs[0]!)).toEqual(runSummary); }); + it("forwards a transcript path containing equals alongside a valid cost observation", async () => { + request.mockResolvedValue(runSummary); + + await runProgram([ + "run-1", + "--observe", + "92d88cce-bdbc-46db-8573-916afd32f6f7=0.125", + "--transcript", + "92d88cce-bdbc-46db-8573-916afd32f6f7=/tmp/project=archive/session.jsonl", + "--json", + ]); + + expect(request).toHaveBeenCalledWith("usage.get", { + runId: "run-1", + observe: { nativeId: "92d88cce-bdbc-46db-8573-916afd32f6f7", costUsd: 0.125 }, + transcript: { + nativeId: "92d88cce-bdbc-46db-8573-916afd32f6f7", + path: "/tmp/project=archive/session.jsonl", + }, + }); + expect(JSON.parse(logs[0]!)).toEqual(runSummary); + }); + + it("forwards a transcript without a cost observation", async () => { + request.mockResolvedValue(runSummary); + + await runProgram([ + "run-1", + "--transcript", + "92d88cce-bdbc-46db-8573-916afd32f6f7=/tmp/session.jsonl", + ]); + + expect(request).toHaveBeenCalledWith("usage.get", { + runId: "run-1", + transcript: { + nativeId: "92d88cce-bdbc-46db-8573-916afd32f6f7", + path: "/tmp/session.jsonl", + }, + }); + expect(logs).toEqual([ + "Run run-1: 1 sessions, 100 input / 20 output / 5 cached tokens, cost $0.25", + ]); + }); + + it.each([ + ["invalid native id", "not-a-session-id=/tmp/session.jsonl"], + ["missing equals", "92d88cce-bdbc-46db-8573-916afd32f6f7"], + ["empty native id", "=/tmp/session.jsonl"], + ["empty path", "92d88cce-bdbc-46db-8573-916afd32f6f7="], + ])("ignores a transcript with %s while preserving a valid cost observation and summary", async (_label, transcript) => { + request.mockResolvedValue(runSummary); + + await runProgram([ + "run-1", + "--observe", + "92d88cce-bdbc-46db-8573-916afd32f6f7=0.125", + "--transcript", + transcript, + "--json", + ]); + + expect(request).toHaveBeenCalledWith("usage.get", { + runId: "run-1", + observe: { nativeId: "92d88cce-bdbc-46db-8573-916afd32f6f7", costUsd: 0.125 }, + }); + expect(JSON.parse(logs[0]!)).toEqual(runSummary); + expect(errors).toEqual([]); + expect(process.exitCode).toBeUndefined(); + }); + it.each([ ["malformed", "not-a-session-id=0.5"], ["negative", "92d88cce-bdbc-46db-8573-916afd32f6f7=-0.01"], diff --git a/tests/usage-statusline-contract.test.ts b/tests/usage-statusline-contract.test.ts index 1c68a93..7e087c9 100644 --- a/tests/usage-statusline-contract.test.ts +++ b/tests/usage-statusline-contract.test.ts @@ -3,17 +3,28 @@ import { formatUsageSummary } from "../src/cli/commands/usage.js"; import type { RunUsageSummary } from "../src/core/run-usage.js"; describe("usage statusline contract", () => { - it("ensures RunUsageSummary preserves the 9 contract keys required by statusline.sh", () => { + it("keeps worker and orchestrator totalTokens in the statusline summary contract", () => { const summary: RunUsageSummary = { runId: "run-test-123", inputTokens: 1000, outputTokens: 500, cachedTokens: 200, + totalTokens: 1500, costUsd: 0.15, sessionCount: 1, activeSessionCount: 0, costComplete: true, sessionsWithoutCost: 0, + orchestrator: { + costUsd: 0.05, + costComplete: true, + inputTokens: 100, + outputTokens: 50, + cachedTokens: 25, + totalTokens: 175, + sources: [], + }, + total: { costUsd: 0.2 }, }; const keys = Object.keys(summary).sort(); @@ -24,12 +35,17 @@ describe("usage statusline contract", () => { "costUsd", "inputTokens", "outputTokens", + "orchestrator", "runId", "sessionCount", "sessionsWithoutCost", + "total", + "totalTokens", ].sort(); expect(keys).toEqual(expectedKeys); + expect(summary.totalTokens).toBe(1500); + expect(summary.orchestrator.totalTokens).toBe(175); }); it("formats legacy summary string with cost and tokens", () => { @@ -38,11 +54,22 @@ describe("usage statusline contract", () => { inputTokens: 1200, outputTokens: 800, cachedTokens: 300, + totalTokens: 2000, costUsd: 0.42, sessionCount: 2, activeSessionCount: 0, costComplete: true, sessionsWithoutCost: 0, + orchestrator: { + costUsd: 0, + costComplete: true, + inputTokens: 0, + outputTokens: 0, + cachedTokens: 0, + totalTokens: 0, + sources: [], + }, + total: { costUsd: 0.42 }, }; const text = formatUsageSummary(summary); @@ -55,11 +82,22 @@ describe("usage statusline contract", () => { inputTokens: 1200, outputTokens: 800, cachedTokens: 300, + totalTokens: 2000, costUsd: 0.42, sessionCount: 2, activeSessionCount: 0, costComplete: false, sessionsWithoutCost: 1, + orchestrator: { + costUsd: 0, + costComplete: true, + inputTokens: 0, + outputTokens: 0, + cachedTokens: 0, + totalTokens: 0, + sources: [], + }, + total: { costUsd: 0.42 }, }; const text = formatUsageSummary(summary);