Continue on non-fatal reactor errors and skip empty checkpoints (#892)

* Continue on non-fatal reactor errors and skip empty checkpoints

reactor.error with fatal:false is a recoverable stream signal; treating it
as terminal idle/failed the TUI and run sink while the reactor kept going.
Empty managed checkpoints were also creating new commits and risking
session junk like partial.jsonl riding along — skip when no allowlisted
path differs from HEAD.

* Keep text-delta suppression across non-fatal reactor errors

* Keep unpublished rewrites out of the empty checkpoint skip

Empty managed checkpoints skip the git commit when no allowlisted
path differs from HEAD. A staged rewrite is still unpublished on
disk at that check, so skipping would publish it in memory without
writing or committing. Only skip when nothing is staged to land.

Author: TheGreatAxios

src/agent/reactor-events.test.ts

@@ -1,7 +1,11 @@
 import { describe, expect, test } from "bun:test";
 import type { ReactorEmittedEvent } from "@intx/inference";
 import type { ReactorInboundEvent } from "@intx/types/runtime";
-import { onReactorShutdown, onTurnBoundary } from "./reactor-events.js";
+import {
+  isReactorErrorFatal,
+  onReactorShutdown,
+  onTurnBoundary,
+} from "./reactor-events.js";
 
 // Bare `{ type: string }` literals only prove the string comparison works.
 // The generic exists so the guards narrow across both `ReactorInboundEvent`
@@ -115,3 +119,27 @@ describe("onReactorShutdown", () => {
     expect(shutdowns.length).toBe(1);
   });
 });
+
+describe("isReactorErrorFatal", () => {
+  test("fatal:false continues", () => {
+    expect(
+      isReactorErrorFatal({ error: "transient write", fatal: false }),
+    ).toBe(false);
+  });
+
+  test("fatal:true stays terminal", () => {
+    expect(isReactorErrorFatal({ error: "gave up", fatal: true })).toBe(true);
+  });
+
+  test("missing fatal stays terminal", () => {
+    expect(isReactorErrorFatal({ error: "gave up" })).toBe(true);
+  });
+
+  test("malformed payloads stay terminal", () => {
+    expect(isReactorErrorFatal(undefined)).toBe(true);
+    expect(isReactorErrorFatal(null)).toBe(true);
+    expect(isReactorErrorFatal("boom")).toBe(true);
+    expect(isReactorErrorFatal({ fatal: "false" })).toBe(true);
+    expect(isReactorErrorFatal({ fatal: 0 })).toBe(true);
+  });
+});

src/agent/reactor-events.ts

@@ -13,6 +13,8 @@
  * call sites without re-declaring the union here.
  */
 
+import { type } from "arktype";
+
 /** True when `event` is the turn boundary — fires once per turn, every turn. */
 export const onTurnBoundary = <E extends { type: string }>(
   event: E,
@@ -24,3 +26,19 @@ export const onReactorShutdown = <E extends { type: string }>(
   event: E,
 ): event is Extract<E, { type: "reactor.done" }> =>
   event.type === "reactor.done";
+
+/**
+ * Explicit non-fatal reactor.error payload. Only `fatal: false` continues;
+ * missing, malformed, or any other value stays terminal.
+ */
+const NonFatalReactorErrorData = type({
+  fatal: "false",
+});
+
+/**
+ * Whether a `reactor.error` payload should terminate the turn/shell.
+ * Returns false only when the payload explicitly carries `fatal: false`.
+ */
+export function isReactorErrorFatal(data: unknown): boolean {
+  return NonFatalReactorErrorData(data) instanceof type.errors;
+}

src/session/optimized-context-store.test.ts

@@ -710,6 +710,24 @@ describe("createOptimizedContextStore checkpoint", () => {
     }
     expect(gitSpawns).toEqual([]);
   });
+
+  test("skips commit when no managed path differs and never stages partial.jsonl", async () => {
+    const dir = tempDir();
+    const store = await createOptimizedContextStore(dir);
+    await store.writeMetadata(EMPTY_CHECKPOINT_METADATA);
+    const first = await store.commit({ message: "first managed checkpoint" });
+
+    fs.writeFileSync(path.join(dir, "partial.jsonl"), '{"reason":"abort"}\n');
+    fs.writeFileSync(path.join(dir, "untracked-junk.txt"), "session junk\n");
+    await store.writeMetadata(EMPTY_CHECKPOINT_METADATA);
+
+    const second = await store.commit({ message: "empty managed checkpoint" });
+    expect(second.hash).toBe(first.hash);
+
+    const tree = await gitLsTree(dir);
+    expect(tree).not.toContain("partial.jsonl");
+    expect(tree).not.toContain("untracked-junk.txt");
+  });
 });
 
 describe("createSessionStores", () => {

src/session/optimized-context-store.ts

@@ -489,6 +489,24 @@ function extraCommitPaths(paths: readonly string[]): string[] {
   return paths.filter((filepath) => !VENDOR_COMMIT_ROOT_FILES.has(filepath));
 }
 
+/**
+ * True when any allowlisted managed path differs between HEAD and the
+ * worktree. Callers must pass only managed paths — never "." and never
+ * session junk such as partial.jsonl.
+ */
+async function managedPathsDiffer(
+  dir: string,
+  filepaths: readonly string[],
+): Promise<boolean> {
+  if (filepaths.length === 0) return false;
+  const matrix = await git.statusMatrix({
+    fs,
+    dir,
+    filepaths: [...filepaths],
+  });
+  return matrix.some(([, head, workdir]) => head !== workdir);
+}
+
 export interface SessionStores {
   storage: ContextStore;
   audit: AuditStore;
@@ -724,6 +742,28 @@ export async function createSessionStores(
         const headBefore = stagedRewrite === null ? null : await headOid(dir);
         let extraPaths: string[] = [];
 
+        // Empty managed checkpoints must not create a new commit or stage
+        // session junk such as partial.jsonl. A staged unpublished rewrite
+        // is still unpublished on disk — skip would swallow the compact
+        // without writing it, so that path always goes through commit.
+        if (stagedRewrite === null) {
+          const managedFilepaths = [
+            ...VENDOR_COMMIT_ROOT_FILES,
+            TOOL_OUTPUT_DIR,
+            EVIDENCE_ARCHIVE_DIR,
+            ...pendingSegmentPaths,
+            ...pendingBlobFilepaths,
+          ];
+          if (!(await managedPathsDiffer(dir, managedFilepaths))) {
+            const [head] = await base.log(1);
+            if (head !== undefined) {
+              pendingBlobFilepaths.clear();
+              pendingSegmentPaths.clear();
+              return head;
+            }
+          }
+        }
+
         try {
           if (stagedRewrite !== null) {
             await writeSegmented(writeTurnsSegmented, stagedRewrite);

src/session/run-sink.test.ts

@@ -532,4 +532,47 @@ describe("createRunSink", () => {
       toolCallCount: 0,
     });
   });
+
+  test("non-fatal reactor.error does not sticky-fail the run", () => {
+    const runSink = createRunSink({
+      emitter: new EventEmitter(),
+      hookManager: stubHookManager([]),
+    });
+
+    runSink.sink(
+      event("reactor.error", {
+        error: "transient checkpoint write",
+        fatal: false,
+      }),
+    );
+
+    expect(runSink.getRunError()).toBeUndefined();
+    expect(runSink.getStatus()).not.toBe("failed");
+  });
+
+  test("fatal reactor.error sticky-fails the run", () => {
+    const runSink = createRunSink({
+      emitter: new EventEmitter(),
+      hookManager: stubHookManager([]),
+    });
+
+    runSink.sink(
+      event("reactor.error", { error: "reactor gave up", fatal: true }),
+    );
+
+    expect(runSink.getRunError()).toBe("reactor gave up");
+    expect(runSink.getStatus()).toBe("failed");
+  });
+
+  test("reactor.error without fatal sticky-fails the run", () => {
+    const runSink = createRunSink({
+      emitter: new EventEmitter(),
+      hookManager: stubHookManager([]),
+    });
+
+    runSink.sink(event("reactor.error", { error: "reactor gave up" }));
+
+    expect(runSink.getRunError()).toBe("reactor gave up");
+    expect(runSink.getStatus()).toBe("failed");
+  });
 });

src/session/run-sink.ts

@@ -2,7 +2,10 @@ import type { EventEmitter } from "node:events";
 import type { ReactorEmittedEvent } from "@intx/inference";
 import type { LastCycleSource, TokenUsage } from "@intx/types/runtime";
 import { createPerfReactorObserver } from "../perf/reactor-spans.js";
-import { onTurnBoundary } from "../agent/reactor-events.js";
+import {
+  isReactorErrorFatal,
+  onTurnBoundary,
+} from "../agent/reactor-events.js";
 import {
   createTurnContextCollector,
   type LifecycleHookManager,
@@ -191,7 +194,7 @@ export function createRunSink(args: RunSinkArgs): RunSink {
       runError = undefined;
       onTurnBoundarySnapshot?.();
     }
-    if (event.type === "reactor.error") {
+    if (event.type === "reactor.error" && isReactorErrorFatal(event.data)) {
       const data = event.data as { error: string };
       runError = data.error;
     }

src/tui/stream-event-map.test.ts

@@ -97,6 +97,39 @@ describe("mapProductionEvent", () => {
     ]);
   });
 
+  test("non-fatal reactor.error paints the error without idling", () => {
+    expect(
+      mapProductionEvent({
+        type: "reactor.error",
+        data: { error: "transient checkpoint write", fatal: false },
+      }),
+    ).toEqual([{ type: "error", message: "transient checkpoint write" }]);
+  });
+
+  test("fatal reactor.error paints the error and idles", () => {
+    expect(
+      mapProductionEvent({
+        type: "reactor.error",
+        data: { error: "gave up", fatal: true },
+      }),
+    ).toEqual([
+      { type: "error", message: "gave up" },
+      { type: "run", state: "idle" },
+    ]);
+  });
+
+  test("reactor.error without fatal paints the error and idles", () => {
+    expect(
+      mapProductionEvent({
+        type: "reactor.error",
+        data: { error: "gave up" },
+      }),
+    ).toEqual([
+      { type: "error", message: "gave up" },
+      { type: "run", state: "idle" },
+    ]);
+  });
+
   test("connector.reply after deltas is skipped (already painted)", () => {
     const ctx = createStreamMapContext();
     mapProductionEvent(
@@ -111,6 +144,29 @@ describe("mapProductionEvent", () => {
     ).toEqual([]);
   });
 
+  test("non-fatal reactor.error keeps connector.reply suppression after deltas", () => {
+    const ctx = createStreamMapContext();
+    mapProductionEvent(
+      { type: "inference.text.delta", data: { token: "partial" } },
+      ctx,
+    );
+    expect(
+      mapProductionEvent(
+        {
+          type: "reactor.error",
+          data: { error: "transient checkpoint write", fatal: false },
+        },
+        ctx,
+      ),
+    ).toEqual([{ type: "error", message: "transient checkpoint write" }]);
+    expect(
+      mapProductionEvent(
+        { type: "connector.reply", data: { content: "final answer" } },
+        ctx,
+      ),
+    ).toEqual([]);
+  });
+
   test("connector.reply without prior deltas becomes assistant", () => {
     expect(
       mapProductionEvent({

src/tui/stream-event-map.ts

@@ -9,6 +9,7 @@ import {
   splitPendingControlTail,
   stripTerminalControlSequences,
 } from "../util/control-char-strip.js";
+import { isReactorErrorFatal } from "../agent/reactor-events.js";
 import { terminalProviderFailureMessage } from "../inference-error-message.js";
 import {
   normalizeInferenceErrorForTerminal,
@@ -574,6 +575,11 @@ function mapEvent(
     case "reactor.error": {
       const error =
         typeof data.error === "string" ? data.error : "reactor error";
+      // Non-fatal errors don't end the turn: keep delta suppression so a
+      // later connector.reply doesn't repaint already-streamed text.
+      if (!isReactorErrorFatal(event.data)) {
+        return [{ type: "error", message: error }];
+      }
       if (ctx) ctx.hadTextDelta = false;
       return [
         ...disarmAttempt(ctx),

src/tui/turn-state.test.ts

@@ -508,4 +508,41 @@ describe("repetition tracking", () => {
     expect(restarted.repeatingSinceTokenCount).toBeNull();
     expect(restarted.streamText).toBe("");
   });
+
+  test("non-fatal reactor.error leaves the turn running", () => {
+    const running = fold([
+      { type: "inference.start" },
+      { type: "inference.text.delta", data: { token: "hi" } },
+    ]);
+    const continued = turnStateFromEvent(
+      running,
+      {
+        type: "reactor.error",
+        data: { error: "transient checkpoint write", fatal: false },
+      },
+      100,
+    );
+    expect(continued.status).toBe("running");
+    expect(continued.isProcessing).toBe(true);
+  });
+
+  test("fatal reactor.error fails the turn", () => {
+    const running = fold([{ type: "inference.start" }]);
+    const failed = turnStateFromEvent(
+      running,
+      { type: "reactor.error", data: { error: "gave up", fatal: true } },
+      100,
+    );
+    expect(failed.status).toBe("failed");
+  });
+
+  test("reactor.error without fatal fails the turn", () => {
+    const running = fold([{ type: "inference.start" }]);
+    const failed = turnStateFromEvent(
+      running,
+      { type: "reactor.error", data: { error: "gave up" } },
+      100,
+    );
+    expect(failed.status).toBe("failed");
+  });
 });

src/tui/turn-state.ts

@@ -11,6 +11,7 @@
 
 import { type } from "arktype";
 
+import { isReactorErrorFatal } from "../agent/reactor-events.js";
 import type { TurnStatus } from "./session-chrome.js";
 
 // Bound on the accumulated stream text kept for the current cycle. Comfortably
@@ -687,6 +688,7 @@ export function turnStateFromEvent(
       });
 
     case "reactor.error":
+      if (!isReactorErrorFatal(event.data)) return state;
       return carryBlockedGateCount(state, {
         ...initialTurnState(nowMs),
         status: "failed",