Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 29 additions & 1 deletion src/agent/reactor-events.test.ts
Original file line number Diff line number Diff line change
@@ -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`
Expand Down Expand Up @@ -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);
});
});
18 changes: 18 additions & 0 deletions src/agent/reactor-events.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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;
}
18 changes: 18 additions & 0 deletions src/session/optimized-context-store.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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", () => {
Expand Down
40 changes: 40 additions & 0 deletions src/session/optimized-context-store.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down
43 changes: 43 additions & 0 deletions src/session/run-sink.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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");
});
});
7 changes: 5 additions & 2 deletions src/session/run-sink.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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;
}
Expand Down
56 changes: 56 additions & 0 deletions src/tui/stream-event-map.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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({
Expand Down
6 changes: 6 additions & 0 deletions src/tui/stream-event-map.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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),
Expand Down
37 changes: 37 additions & 0 deletions src/tui/turn-state.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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");
});
});
2 changes: 2 additions & 0 deletions src/tui/turn-state.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -687,6 +688,7 @@ export function turnStateFromEvent(
});

case "reactor.error":
if (!isReactorErrorFatal(event.data)) return state;
return carryBlockedGateCount(state, {
...initialTurnState(nowMs),
status: "failed",
Expand Down
Loading