From fcecbdc535b58a41d0a9b0c1c44c68f7492731e7 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 10 Sep 2026 16:06:04 -0700 Subject: [PATCH 1/3] Continue on non-fatal reactor errors and skip empty checkpoints MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- src/agent/reactor-events.test.ts | 30 +++++++++++++- src/agent/reactor-events.ts | 18 +++++++++ src/session/optimized-context-store.test.ts | 18 +++++++++ src/session/optimized-context-store.ts | 37 ++++++++++++++++++ src/session/run-sink.test.ts | 43 +++++++++++++++++++++ src/session/run-sink.ts | 7 +++- src/tui/stream-event-map.test.ts | 33 ++++++++++++++++ src/tui/stream-event-map.ts | 4 ++ src/tui/turn-state.test.ts | 37 ++++++++++++++++++ src/tui/turn-state.ts | 2 + 10 files changed, 226 insertions(+), 3 deletions(-) diff --git a/src/agent/reactor-events.test.ts b/src/agent/reactor-events.test.ts index d2eedf901..72fa32ce3 100644 --- a/src/agent/reactor-events.test.ts +++ b/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); + }); +}); diff --git a/src/agent/reactor-events.ts b/src/agent/reactor-events.ts index eaa2ec548..e26c316e1 100644 --- a/src/agent/reactor-events.ts +++ b/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 = ( event: E, @@ -24,3 +26,19 @@ export const onReactorShutdown = ( event: E, ): event is Extract => 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; +} diff --git a/src/session/optimized-context-store.test.ts b/src/session/optimized-context-store.test.ts index 8443f8b38..e98ed5096 100644 --- a/src/session/optimized-context-store.test.ts +++ b/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", () => { diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index 2c0590525..d74b82030 100644 --- a/src/session/optimized-context-store.ts +++ b/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 { + 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,25 @@ export async function createSessionStores( const headBefore = stagedRewrite === null ? null : await headOid(dir); let extraPaths: string[] = []; + const managedFilepaths = [ + ...VENDOR_COMMIT_ROOT_FILES, + TOOL_OUTPUT_DIR, + EVIDENCE_ARCHIVE_DIR, + ...extraPaths, + ]; + if (!(await managedPathsDiffer(dir, managedFilepaths))) { + const [head] = await base.log(1); + if (head !== undefined) { + pendingBlobFilepaths.clear(); + pendingSegmentPaths.clear(); + if (stagedRewrite !== null) { + liveTurnRefs = stagedRewrite; + unpublishedRewrite = null; + } + return head; + } + } + try { if (stagedRewrite !== null) { await writeSegmented(writeTurnsSegmented, stagedRewrite); diff --git a/src/session/run-sink.test.ts b/src/session/run-sink.test.ts index 2da3cbd4b..872da240b 100644 --- a/src/session/run-sink.test.ts +++ b/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"); + }); }); diff --git a/src/session/run-sink.ts b/src/session/run-sink.ts index dd7ea00ca..efd3c85fa 100644 --- a/src/session/run-sink.ts +++ b/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; } diff --git a/src/tui/stream-event-map.test.ts b/src/tui/stream-event-map.test.ts index b053f7485..4f1bb8566 100644 --- a/src/tui/stream-event-map.test.ts +++ b/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( diff --git a/src/tui/stream-event-map.ts b/src/tui/stream-event-map.ts index 77b9715c7..ba3410deb 100644 --- a/src/tui/stream-event-map.ts +++ b/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, @@ -575,6 +576,9 @@ function mapEvent( const error = typeof data.error === "string" ? data.error : "reactor error"; if (ctx) ctx.hadTextDelta = false; + if (!isReactorErrorFatal(event.data)) { + return [{ type: "error", message: error }]; + } return [ ...disarmAttempt(ctx), { type: "error", message: error }, diff --git a/src/tui/turn-state.test.ts b/src/tui/turn-state.test.ts index eff159e20..dd966fa29 100644 --- a/src/tui/turn-state.test.ts +++ b/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"); + }); }); diff --git a/src/tui/turn-state.ts b/src/tui/turn-state.ts index ce9d4a423..de139f821 100644 --- a/src/tui/turn-state.ts +++ b/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", From 8af755bd102d6e87fc0f23fe1f0ad9074560458b Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Thu, 10 Sep 2026 19:32:15 -0700 Subject: [PATCH 2/3] Keep text-delta suppression across non-fatal reactor errors --- src/tui/stream-event-map.test.ts | 23 +++++++++++++++++++++++ src/tui/stream-event-map.ts | 4 +++- 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/src/tui/stream-event-map.test.ts b/src/tui/stream-event-map.test.ts index 4f1bb8566..4444eb878 100644 --- a/src/tui/stream-event-map.test.ts +++ b/src/tui/stream-event-map.test.ts @@ -144,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({ diff --git a/src/tui/stream-event-map.ts b/src/tui/stream-event-map.ts index ba3410deb..301e4c532 100644 --- a/src/tui/stream-event-map.ts +++ b/src/tui/stream-event-map.ts @@ -575,10 +575,12 @@ function mapEvent( case "reactor.error": { const error = typeof data.error === "string" ? data.error : "reactor error"; - if (ctx) ctx.hadTextDelta = false; + // 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), { type: "error", message: error }, From c1e8eab55a9dcb84a0a82ee09c39be7b079ea11b Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Thu, 10 Sep 2026 20:29:11 -0700 Subject: [PATCH 3/3] 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. --- src/session/optimized-context-store.ts | 33 ++++++++++++++------------ 1 file changed, 18 insertions(+), 15 deletions(-) diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index d74b82030..e7d54e33e 100644 --- a/src/session/optimized-context-store.ts +++ b/src/session/optimized-context-store.ts @@ -742,22 +742,25 @@ export async function createSessionStores( const headBefore = stagedRewrite === null ? null : await headOid(dir); let extraPaths: string[] = []; - const managedFilepaths = [ - ...VENDOR_COMMIT_ROOT_FILES, - TOOL_OUTPUT_DIR, - EVIDENCE_ARCHIVE_DIR, - ...extraPaths, - ]; - if (!(await managedPathsDiffer(dir, managedFilepaths))) { - const [head] = await base.log(1); - if (head !== undefined) { - pendingBlobFilepaths.clear(); - pendingSegmentPaths.clear(); - if (stagedRewrite !== null) { - liveTurnRefs = stagedRewrite; - unpublishedRewrite = null; + // 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; } - return head; } }