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..e7d54e33e 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,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); 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..4444eb878 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( @@ -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({ diff --git a/src/tui/stream-event-map.ts b/src/tui/stream-event-map.ts index 77b9715c7..301e4c532 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, @@ -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), 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",