diff --git a/packages/adapters/codex-local/src/index.ts b/packages/adapters/codex-local/src/index.ts index 607667b6a1..8b25aa5343 100644 --- a/packages/adapters/codex-local/src/index.ts +++ b/packages/adapters/codex-local/src/index.ts @@ -95,7 +95,7 @@ Core fields: Operational fields: - timeoutSec (number, optional): run timeout in seconds - graceSec (number, optional): SIGTERM grace period in seconds -- outputInactivityTimeoutMs (number | null, optional): inactivity monitor around the codex child. Resets on every parsed JSONL event from stdout. Defaults to 7 * 60_000 ms when unset or non-positive. Set to \`null\` to disable the monitor entirely (only do this for known-slow tasks; the platform-level 1h silent-run safety net still applies). On fire, the adapter sends SIGTERM to the process group, waits 5s, then SIGKILL, and surfaces the run as failed with errorMessage "monitor: no codex output for {N}m {S}s". +- outputInactivityTimeoutMs (number | null, optional): inactivity monitor around the codex child. Resets whenever the child emits stdout or stderr bytes, including non-JSON progress from long-running verification commands. Defaults to 7 * 60_000 ms when unset or non-positive. Set to \`null\` to disable the monitor entirely (only do this for known-slow tasks; the platform-level 1h silent-run safety net still applies). On fire, the adapter sends SIGTERM to the process group, waits 5s, then SIGKILL, and surfaces the run as failed with errorMessage "monitor: no codex output for {N}m {S}s". - agentCommand (string, optional): ACP server command override used only when engine="acp"; defaults to the package-local codex-acp binary - mode (string, optional): ACP session mode when engine="acp"; persistent or oneshot - nonInteractivePermissions (string, optional): ACP non-interactive permission fallback when engine="acp"; deny or fail diff --git a/packages/adapters/codex-local/src/server/execute.ts b/packages/adapters/codex-local/src/server/execute.ts index c62eba981f..a6f47ba3ea 100644 --- a/packages/adapters/codex-local/src/server/execute.ts +++ b/packages/adapters/codex-local/src/server/execute.ts @@ -1037,6 +1037,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise { + monitor?.noteOutputChunk(stream, chunk); if (stream === "stdout") { - monitor?.noteStdoutChunk(chunk); await onLog(stream, chunk); return; } diff --git a/packages/adapters/codex-local/src/server/output-inactivity-monitor.integration.test.ts b/packages/adapters/codex-local/src/server/output-inactivity-monitor.integration.test.ts index e6fadbdff6..fda2aded6a 100644 --- a/packages/adapters/codex-local/src/server/output-inactivity-monitor.integration.test.ts +++ b/packages/adapters/codex-local/src/server/output-inactivity-monitor.integration.test.ts @@ -72,9 +72,7 @@ describe("codex inactivity monitor (integration: real subprocess)", () => { }, onLog: async (stream, chunk) => { logs.push({ stream, chunk }); - if (stream === "stdout") { - monitor.noteStdoutChunk(chunk); - } + monitor.noteOutputChunk(stream, chunk); }, }); diff --git a/packages/adapters/codex-local/src/server/output-inactivity-monitor.test.ts b/packages/adapters/codex-local/src/server/output-inactivity-monitor.test.ts index 11d6332748..71c25df94e 100644 --- a/packages/adapters/codex-local/src/server/output-inactivity-monitor.test.ts +++ b/packages/adapters/codex-local/src/server/output-inactivity-monitor.test.ts @@ -121,7 +121,7 @@ describe("createCodexOutputInactivityMonitor (acceptance criteria 1: fires)", () // One event right after spawn. clock.advance(50); - monitor.noteStdoutChunk('{"type":"thread.started","thread_id":"abc"}\n'); + monitor.noteOutputChunk("stdout", '{"type":"thread.started","thread_id":"abc"}\n'); expect(fires).toHaveLength(0); expect(monitor.state().parsedEventCount).toBe(1); @@ -157,7 +157,7 @@ describe("createCodexOutputInactivityMonitor (acceptance criteria 1: fires)", () monitor.stop(); }); - it("ignores non-JSON lines when resetting the timer", () => { + it("resets on non-JSON stdout bytes", () => { const clock = new FakeClock(); let fireCount = 0; const monitor = createCodexOutputInactivityMonitor({ @@ -169,17 +169,52 @@ describe("createCodexOutputInactivityMonitor (acceptance criteria 1: fires)", () fireCount += 1; }, }); - // Plain stderr-ish text should NOT reset the monitor. clock.advance(500); - monitor.noteStdoutChunk("loading model...\n"); - expect(monitor.state().parsedEventCount).toBe(0); - clock.advance(600); + monitor.noteOutputChunk("stdout", "loading model...\n"); + expect(monitor.state()).toMatchObject({ + outputChunkCount: 1, + outputBytes: Buffer.byteLength("loading model...\n", "utf8"), + parsedEventCount: 0, + }); + clock.advance(999); + expect(fireCount).toBe(0); + clock.advance(1); expect(fireCount).toBe(1); monitor.stop(); }); }); describe("createCodexOutputInactivityMonitor (acceptance criteria 2: does not fire)", () => { + it("keeps long verification alive while non-JSON stdout and stderr bytes continue", () => { + const clock = new FakeClock(); + let fireCount = 0; + const timeoutMs = 7 * 60 * 1000; + const monitor = createCodexOutputInactivityMonitor({ + timeoutMs, + now: () => clock.now(), + setTimer: (cb, ms) => clock.setTimer(cb, ms), + clearTimer: (handle) => clock.clearTimer(handle), + onFire: () => { + fireCount += 1; + }, + }); + + clock.advance(timeoutMs - 1_000); + monitor.noteOutputChunk("stdout", "packages/server: typecheck passed\n"); + clock.advance(timeoutMs - 1_000); + monitor.noteOutputChunk("stderr", "packages/ui: build still running\n"); + clock.advance(timeoutMs - 1_000); + monitor.noteOutputChunk("stdout", "packages/ui: build passed\n"); + + expect(fireCount).toBe(0); + expect(monitor.state()).toMatchObject({ + outputChunkCount: 3, + parsedEventCount: 0, + fired: false, + }); + monitor.stop(); + }); + it("does not fire when events arrive every (threshold - 1s)", () => { const clock = new FakeClock(); let fireCount = 0; @@ -197,7 +232,7 @@ describe("createCodexOutputInactivityMonitor (acceptance criteria 2: does not fi // Pump events at threshold-1s intervals for 12 cycles (~84 minutes). for (let i = 0; i < 12; i += 1) { clock.advance(timeoutMs - 1_000); - monitor.noteStdoutChunk(`{"type":"item.completed","item":{"type":"agent_message","text":"tick ${i}"}}\n`); + monitor.noteOutputChunk("stdout", `{"type":"item.completed","item":{"type":"agent_message","text":"tick ${i}"}}\n`); expect(fireCount).toBe(0); } @@ -223,7 +258,8 @@ describe("createCodexOutputInactivityMonitor (acceptance criteria 2: does not fi }, }); clock.advance(500); - monitor.noteStdoutChunk( + monitor.noteOutputChunk( + "stdout", '{"type":"thread.started","thread_id":"a"}\n{"type":"item.completed","item":{"type":"agent_message","text":"hi"}}\n', ); expect(monitor.state().parsedEventCount).toBe(2); diff --git a/packages/adapters/codex-local/src/server/output-inactivity-monitor.ts b/packages/adapters/codex-local/src/server/output-inactivity-monitor.ts index 9363390946..814e06d94d 100644 --- a/packages/adapters/codex-local/src/server/output-inactivity-monitor.ts +++ b/packages/adapters/codex-local/src/server/output-inactivity-monitor.ts @@ -31,6 +31,8 @@ export interface CodexOutputInactivityMonitorState { spawnedAt: number; lastEventAt: number; firedAt: number | null; + outputChunkCount: number; + outputBytes: number; parsedEventCount: number; } @@ -48,7 +50,7 @@ export interface CodexOutputInactivityMonitorOptions { } export interface CodexOutputInactivityMonitorHandle { - noteStdoutChunk(chunk: string): void; + noteOutputChunk(stream: "stdout" | "stderr", chunk: string): void; /** Returns the current state without stopping the timer. */ state(): CodexOutputInactivityMonitorState; /** Cancels any pending timer and returns the final state. */ @@ -80,6 +82,8 @@ export function createCodexOutputInactivityMonitor( spawnedAt, lastEventAt: spawnedAt, firedAt: null, + outputChunkCount: 0, + outputBytes: 0, parsedEventCount: 0, }; let timerHandle: unknown = null; @@ -102,19 +106,19 @@ export function createCodexOutputInactivityMonitor( arm(); return { - noteStdoutChunk(chunk: string) { - if (stopped || state.fired) return; - let sawHeartbeat = false; - for (const rawLine of chunk.split(/\r?\n/)) { - if (isHeartbeatLine(rawLine)) { - sawHeartbeat = true; - state.parsedEventCount += 1; + noteOutputChunk(stream: "stdout" | "stderr", chunk: string) { + if (stopped || state.fired || chunk.length === 0) return; + state.outputChunkCount += 1; + state.outputBytes += Buffer.byteLength(chunk, "utf8"); + if (stream === "stdout") { + for (const rawLine of chunk.split(/\r?\n/)) { + if (isHeartbeatLine(rawLine)) { + state.parsedEventCount += 1; + } } } - if (sawHeartbeat) { - state.lastEventAt = now(); - arm(); - } + state.lastEventAt = now(); + arm(); }, state() { return { ...state };