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
15 changes: 15 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,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
Expand Down
1 change: 1 addition & 0 deletions docs/IMPLEMENTATION.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
20 changes: 14 additions & 6 deletions src/exec/runner.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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",
Expand All @@ -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}`,
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand All @@ -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) {
Expand Down Expand Up @@ -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,
Expand Down
25 changes: 25 additions & 0 deletions src/session/run-sink.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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({
Expand Down
34 changes: 33 additions & 1 deletion src/session/stream-journal.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -44,6 +50,7 @@ async function readPartialRecords(): Promise<
text: string;
thinkingText?: string;
thinkingChars?: number;
error?: { category?: string; message?: string; statusCode?: number };
},
);
}
Expand Down Expand Up @@ -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 () => {
Expand Down
41 changes: 37 additions & 4 deletions src/session/stream-journal.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand All @@ -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),
Expand All @@ -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 => {
Expand All @@ -130,7 +163,7 @@ export function createCycleTextRecorder(
return;
}
if (event.type === "inference.error") {
void flush("inference-error");
void flush("inference-error", inferenceErrorFromEvent(event));
}
};

Expand Down
Loading