diff --git a/CHANGELOG.md b/CHANGELOG.md index f676aed6..8100eec1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,6 +35,21 @@ parallel copies under `docs/` or `scripts/notes/`. At cut time: rename unset; values must still be integers ≥1. `task(maxTurns)`, profile `maxTurns`, and `settings.subagentMaxTurns` may exceed 100 for long jobs. +### Internal + +- **`inference.error` partials keep the provider error.** `partial.jsonl` + records for `inference-error` now include `error` (`category`, `message`, + `statusCode` when present) even when the cycle streamed no text. +- **Exec `turnsUsed` follows the run-sink.** Mid-run and terminal `run.json` + snapshots use `getTurnCount()` the same way the TUI does, instead of + writing the initial zero until send finishes. + +### Docs + +- **`latest` is a symlink, not a session.** Naive globs of a project + sessions directory double-count unless they skip `latest` (`listSessions` + already does). + ## [0.2.104] - 2026-08-23 ### TUI diff --git a/docs/IMPLEMENTATION.md b/docs/IMPLEMENTATION.md index b2d8e908..3915c788 100644 --- a/docs/IMPLEMENTATION.md +++ b/docs/IMPLEMENTATION.md @@ -353,6 +353,7 @@ Session runtime state lives under the global projects tree (not in the repo): - `~/.corbits/projects///run.json` — `RunState` - `~/.corbits/projects///context/` — git-backed conversation context (`@intx/storage-isogit`) - Project key: slug + short hash of this checkout's git toplevel (from `--show-toplevel`, so linked worktrees have distinct keys; workspace realpath when not a git tree) +- `latest` is a symlink in that same project sessions directory (not a session of its own). Naive globs over the directory double-count unless they skip `latest` (as `listSessions` does). - Migration: if a session exists only under in-repo `.agent-state//`, it is moved into the global tree on open/list - Atomic JSON writes with schema validation on load diff --git a/src/exec/runner.ts b/src/exec/runner.ts index b95fe8cf..be03c5d6 100644 --- a/src/exec/runner.ts +++ b/src/exec/runner.ts @@ -77,7 +77,7 @@ import { sessionDir, } from "../session/index.js"; import { saveState, type ConnectedMcpServer } from "../session/state.js"; -import { createRunSink, resolveExecRunStatus } from "../session/run-sink.js"; +import { createRunSink, resolveExecRunStatus, type RunSink } from "../session/run-sink.js"; import { createLifecycleHookManager, createRunSummary, @@ -244,6 +244,7 @@ export async function runExec(config: Config): Promise { let textOut = ""; let finalized = false; let turnsUsed = 0; + let runSink: RunSink | null = null; const persist = async ( status: "running" | "done" | "failed" | "cancelled", @@ -253,7 +254,7 @@ export async function runExec(config: Config): Promise { if (status !== "running") finalized = true; await saveState(config.cwd, sessionId, { status, - turnsUsed, + turnsUsed: runSink?.getTurnCount() ?? turnsUsed, task, startedAt, model: `${config.providerName}:${config.model}`, @@ -645,7 +646,14 @@ export async function runExec(config: Config): Promise { const hookManager = createLifecycleHookManager({ hooks: await discoverLifecycleHooks(hookDirectories(config.cwd)), }); - const runSink = createRunSink({ emitter, hookManager }); + const liveSink = createRunSink({ + emitter, + hookManager, + onTurnBoundarySnapshot: () => { + void persist("running"); + }, + }); + runSink = liveSink; currentAgent = await buildAgent(); agent = currentAgent; @@ -680,7 +688,7 @@ export async function runExec(config: Config): Promise { // its partial output in partial.jsonl instead of vanishing. const cycleRecorder = createCycleTextRecorder(() => workdir); const sink = (event: ReactorEmittedEvent): void => { - runSink.sink(event); + liveSink.sink(event); cycleRecorder.handleEvent(event); if (event.type === "inference.text.delta") { const token = (event.data as { token?: string }).token; @@ -701,7 +709,7 @@ export async function runExec(config: Config): Promise { // sticky inference.error and would hide a real failure. let sendCompleted = false; let runError: string | undefined; - let sinkStatus: ReturnType = "cancelled"; + let sinkStatus: ReturnType = "cancelled"; try { // Final OAuth refresh immediately before send (token may have aged during MCP). if (initialCodexProfile !== undefined) { @@ -845,7 +853,7 @@ export async function runExec(config: Config): Promise { error: message, status: "failed", durationMs: Date.now() - startedAt, - turnsUsed, + turnsUsed: runSink?.getTurnCount() ?? turnsUsed, toolCallCount: 0, tokenUsage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, thinking: 0 }, provider: config.providerName, diff --git a/src/session/run-sink.test.ts b/src/session/run-sink.test.ts index 46f7319f..f23da7ab 100644 --- a/src/session/run-sink.test.ts +++ b/src/session/run-sink.test.ts @@ -87,6 +87,31 @@ describe("createRunSink", () => { }); }); + test("onTurnBoundarySnapshot reads getTurnCount after the turn, not the initial zero", () => { + // Exec persist now snapshots from this callback (same as TUI). A closed-over + // turnsUsed: 0 would write run.json as still-zero mid-run. + const snapshots: number[] = []; + const runSink = createRunSink({ + emitter: new EventEmitter(), + hookManager: stubHookManager([]), + onTurnBoundarySnapshot: () => { + snapshots.push(runSink.getTurnCount()); + }, + }); + + expect(runSink.getTurnCount()).toBe(0); + runSink.sink( + event("inference.done", { + turn: { role: "assistant", content: [], model: "test", timestamp: 0 }, + usage: { input: 1, output: 1, cacheRead: 0, cacheWrite: 0, thinking: 0 }, + source: { provider: "test", model: "test" }, + }), + ); + + expect(snapshots).toEqual([1]); + expect(runSink.getTurnCount()).toBe(1); + }); + test("reports the in-flight turn to onTurnFailed when a turn errors instead of completing", () => { const failures: { turnIndex: number; error: string }[] = []; const runSink = createRunSink({ diff --git a/src/session/stream-journal.test.ts b/src/session/stream-journal.test.ts index b2642bd1..3f96d4d9 100644 --- a/src/session/stream-journal.test.ts +++ b/src/session/stream-journal.test.ts @@ -31,7 +31,13 @@ function thinkingDelta(token: string): ReactorEmittedEvent { } async function readPartialRecords(): Promise< - { reason: string; text: string; thinkingText?: string; thinkingChars?: number }[] + { + reason: string; + text: string; + thinkingText?: string; + thinkingChars?: number; + error?: { category?: string; message?: string; statusCode?: number }; + }[] > { const raw = await readFile(join(dir, PARTIAL_FILE), "utf8"); return raw @@ -44,6 +50,7 @@ async function readPartialRecords(): Promise< text: string; thinkingText?: string; thinkingChars?: number; + error?: { category?: string; message?: string; statusCode?: number }; }, ); } @@ -90,6 +97,31 @@ describe("createCycleTextRecorder", () => { const records = await readPartialRecords(); expect(records[0]?.reason).toBe("inference-error"); expect(records[0]?.text).toBe("partial before failure"); + expect(records[0]?.error?.category).toBe("aborted"); + expect(records[0]?.error?.message).toBe("aborted"); + }); + + test("inference.error with empty cycle text still writes a partial with the error payload", async () => { + // Observed live: ~20 unattributable episodes had inference.error with no + // streamed text. The partial must still land so category/message survive. + const recorder = createCycleTextRecorder(() => dir); + recorder.handleEvent({ + type: "inference.error", + data: { + error: { category: "rate_limit", message: "429 too many requests", statusCode: 429 }, + }, + } as unknown as ReactorEmittedEvent); + + await Bun.sleep(20); + const records = await readPartialRecords(); + expect(records).toHaveLength(1); + expect(records[0]?.reason).toBe("inference-error"); + expect(records[0]?.text).toBe(""); + expect(records[0]?.error).toEqual({ + category: "rate_limit", + message: "429 too many requests", + statusCode: 429, + }); }); test("dispose flushes the entry snapshot with the given reason and returns it", async () => { diff --git a/src/session/stream-journal.ts b/src/session/stream-journal.ts index 3124cfe9..1e0ec1eb 100644 --- a/src/session/stream-journal.ts +++ b/src/session/stream-journal.ts @@ -43,6 +43,33 @@ export type PartialFlushReason = | "send-failed" | "inference-error"; +/** Fields copied from `inference.error` `data.error` onto a partial.jsonl record. */ +export interface PartialInferenceError { + category?: string; + message?: string; + statusCode?: number; +} + +function inferenceErrorFromEvent(event: ReactorEmittedEvent): PartialInferenceError | undefined { + const data = event.data as { error?: unknown } | undefined; + if (data === undefined || typeof data !== "object" || data === null) return undefined; + const raw = data.error; + if (raw === undefined || typeof raw !== "object" || raw === null) return undefined; + const rec = raw as Record; + const error: PartialInferenceError = {}; + if (typeof rec.category === "string") error.category = rec.category; + if (typeof rec.message === "string") error.message = rec.message; + if (typeof rec.statusCode === "number") error.statusCode = rec.statusCode; + if ( + error.category === undefined && + error.message === undefined && + error.statusCode === undefined + ) { + return undefined; + } + return error; +} + export interface CycleTextRecorder { /** Feed every stream event; buffers deltas, resets on done, flushes on error. */ handleEvent: (event: ReactorEmittedEvent) => void; @@ -80,8 +107,10 @@ export function createCycleTextRecorder( reason: PartialFlushReason, text: string, thinkingText: string, + error?: PartialInferenceError, ): Promise => { - if (text.trim().length === 0 && thinkingText.trim().length === 0) return; + const hasErrorPayload = reason === "inference-error" && error !== undefined; + if (text.trim().length === 0 && thinkingText.trim().length === 0 && !hasErrorPayload) return; const record: Record = { reason, chars: text.length, text }; // Omitted when empty: a text-only abort (the common case) keeps the // existing record shape, and diagnosing a thinking-loop abort needs the @@ -90,6 +119,7 @@ export function createCycleTextRecorder( record.thinkingChars = thinkingText.length; record.thinkingText = thinkingText; } + if (error !== undefined) record.error = error; try { await appendFile( join(resolveContextDir(), PARTIAL_FILE), @@ -104,12 +134,15 @@ export function createCycleTextRecorder( } }; - const flush = async (reason: PartialFlushReason): Promise => { + const flush = async ( + reason: PartialFlushReason, + error?: PartialInferenceError, + ): Promise => { const text = cycleText; const thinkingText = cycleThinkingText; cycleText = ""; cycleThinkingText = ""; - await writeRecord(reason, text, thinkingText); + await writeRecord(reason, text, thinkingText, error); }; const handleEvent = (event: ReactorEmittedEvent): void => { @@ -130,7 +163,7 @@ export function createCycleTextRecorder( return; } if (event.type === "inference.error") { - void flush("inference-error"); + void flush("inference-error", inferenceErrorFromEvent(event)); } };