Merge origin/main into cl-6909-thinking-only-replay-guard

Author: TheGreatAxios

CHANGELOG.md

@@ -37,6 +37,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

docs/IMPLEMENTATION.md

@@ -353,6 +353,7 @@ Session runtime state lives under the global projects tree (not in the repo):
 - `~/.corbits/projects/<project-key>/<session-id>/run.json` — `RunState`
 - `~/.corbits/projects/<project-key>/<session-id>/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/<session-id>/`, it is moved into the global tree on open/list
 - Atomic JSON writes with schema validation on load

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<ExecResult> {
   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<ExecResult> {
     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<ExecResult> {
     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<ExecResult> {
     // 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<ExecResult> {
     // sticky inference.error and would hide a real failure.
     let sendCompleted = false;
     let runError: string | undefined;
-    let sinkStatus: ReturnType<typeof runSink.getStatus> = "cancelled";
+    let sinkStatus: ReturnType<typeof liveSink.getStatus> = "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<ExecResult> {
       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,

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({

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 () => {

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<string, unknown>;
+  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<void> => {
-    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<string, unknown> = { 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<void> => {
+  const flush = async (
+    reason: PartialFlushReason,
+    error?: PartialInferenceError,
+  ): Promise<void> => {
     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));
     }
   };