From 7aa212296eb1b9bcefed937d6a58620e5308cafc Mon Sep 17 00:00:00 2001 From: Neeraj Kumar Singh B <43318996+neerazz@users.noreply.github.com> Date: Sat, 20 Jun 2026 20:06:27 -0700 Subject: [PATCH] codex_local adapter output inactivity monitor (#5017) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip orchestrates AI agents for zero-human companies > - The `codex_local` adapter runs `codex exec` as a child process and streams JSONL events from its stdout > - NEE-79 caught a real-world `codex_local` orphan: codex sat in `read()` on stdin for 1h+, no JSON events emitted past startup, no LLM call in flight. The adapter had no inactivity timer; the only safety net was the platform-level 1h silent-run detector > - This is precisely the failure shape NEE-80 scoped: an adapter-detected fault that should be killed *by the adapter* and surfaced as `failed`, not waited out for an hour > - This pull request adds an output-inactivity watchdog inside the codex-local adapter that resets on every parsed JSONL event from stdout, kills the child via SIGTERM → 5s grace → SIGKILL when it fires, and resolves the run with a structured watchdog failure > - The benefit is that NEE-79-class hangs shorter than 1h stop reaching the platform-level safety net — they fail fast and visibly at the adapter layer, with diagnostic logs that don't require host shell access ## Linked Issues or Issue Description No existing GitHub issue covers this; describing the underlying bug in-PR per the bug report template (`.github/ISSUE_TEMPLATE/bug_report.yml`): - **What happened?** A `codex_local` run sat in `read()` on stdin for over an hour: codex emitted its startup JSONL events, then nothing — no further events, no LLM call in flight, no exit. The adapter kept the child alive indefinitely; the run was only reaped by the platform-level 1h silent-run safety net, an hour after it had effectively died. - **Expected behavior:** The adapter should detect that its child has stopped producing output long before the platform-level safety net, kill it, and surface the run as `failed` with a diagnostic that explains what happened. - **Steps to reproduce:** Run any `codex_local` issue where `codex exec` hangs after startup (e.g. codex blocks reading stdin and never emits another JSONL event). Observe the run stays alive until the 1h platform safety net fires. - **Paperclip version or commit:** reproduced on master prior to this branch. - **Agent adapter(s) involved:** codex_local. Related PRs found while searching for duplicates (none implement an event-aware inactivity watchdog inside `codex_local`): - #4004 — generic idle + wall watchdogs for `runChildProcess` in adapter-utils; complementary, operates below the JSONL parse layer and is not codex-event-aware - #4742 — fail-fast on codex-local *startup* hang; this PR covers the post-startup hang class - #6861 — zombie-run termination for codex-local; reaping after exit, not inactivity detection - #7811 — the analogous output-idle timeout for grok-local ## What Changed - `packages/adapters/codex-local/src/server/watchdog.ts` *(new)* — pure watchdog primitive with injectable timers/clock: `resolveCodexInactivityTimeout`, `createCodexInactivityWatchdog`, `formatWatchdogErrorMessage`. Default `7 * 60_000` ms; honors `null` as the disabled escape hatch - `packages/adapters/codex-local/src/server/execute.ts` — `runAttempt` now wraps `onSpawn` to capture `pid`/`processGroupId`, feeds stdout chunks through `noteStdoutChunk`, and on watchdog fire sends SIGTERM to the process group, schedules SIGKILL after 5s, and returns an `AdapterExecutionResult` with `exitCode: null`, `signal: SIGTERM|SIGKILL`, `errorMessage: "watchdog: no codex output for {N}m {S}s"`, `errorCode: "codex_output_inactivity_watchdog"`. With `errorMessage` set and `timedOut: false`, `heartbeat.ts:5860` maps the run to `outcome === "failed"` (not `cancelled`) - `packages/adapters/codex-local/src/index.ts` — `agentConfigurationDoc` documents `outputInactivityTimeoutMs` (number ms; `null` disables; non-positive falls back to default with a warning log at spawn) - `packages/adapters/codex-local/src/server/watchdog.test.ts` *(new)* — 13 tests covering acceptance criteria 2 and 3 plus supporting cases: fires after silence, no-fire across 12× (threshold − 1s) cycles, multi-event chunks, non-JSON ignoring, single-fire idempotency, formatter shape, full resolution table, `null` → disabled - `packages/adapters/codex-local/src/server/watchdog.integration.test.ts` *(new)* — real Node subprocess that prints one JSONL event then sleeps; `runChildProcess` reaps it within `threshold + 6s`; signal is SIGTERM or SIGKILL; `parsedEventCount === 1` (acceptance criteria 1 and 4) ## Verification ``` $ pnpm --filter @paperclipai/adapter-codex-local typecheck > tsc --noEmit # clean $ pnpm --filter @paperclipai/adapter-codex-local exec vitest run ✓ src/server/quota-spawn-error.test.ts (1 test) ✓ src/server/codex-home.test.ts (3 tests) ✓ src/server/codex-args.test.ts (3 tests) ✓ src/server/watchdog.test.ts (13 tests) ✓ src/server/parse.test.ts (9 tests) ✓ src/server/execute.remote.test.ts (4 tests) ✓ src/ui/parse-stdout.test.ts (3 tests) ✓ src/ui/build-config.test.ts (1 test) ✓ src/server/watchdog.integration.test.ts (1 test) Test Files 9 passed (9) Tests 38 passed (38) ``` Acceptance criteria check: 1. ✅ Simulated child emits one event then sleeps → killed at threshold; result `errorMessage` matches `watchdog: no codex output for {N}m {S}s` (`watchdog.integration.test.ts`) 2. ✅ Child emits events every (threshold − 1s) → not killed (`watchdog.test.ts: "does not fire when events arrive every (threshold - 1s)"`) 3. ✅ `outputInactivityTimeoutMs: null` disables the watchdog (`resolveCodexInactivityTimeout` returns `disabled`; `execute.ts` skips watchdog construction and logs a startup warning) 4. ✅ Real subprocess reaped well within `threshold + 6s` — 250 ms threshold, 290 ms wall clock in CI 5. ✅ Adapter-level fault → `outcome === "failed"` per `heartbeat.ts:5860`. NEE-79-class hangs <1h get caught at the adapter, not the platform-level safety net Post-rebase verification (head `2a8703fab`, rebased onto master `69a368ed5`): ``` $ pnpm --filter @paperclipai/adapter-codex-local typecheck # clean $ pnpm --filter @paperclipai/adapter-codex-local exec vitest run Test Files 11 passed (11) Tests 64 passed (64) ``` ## Risks - Low risk. Behind a default-on watchdog that only fires after 7m of zero parsed JSON events. Operators can disable it with `outputInactivityTimeoutMs: null` for known-slow tasks - The kill path reuses the same `process.kill(-pgid, signal)` pattern that `runChildProcess` already uses for its terminal-result cleanup, so signal semantics match the existing code path - `timedOut: false` is preserved on watchdog fire — the platform-level timeout outcome is unchanged, only the `failed`-vs-success classification flips. No behavioral shift for already-failing runs - Sandbox/SSH execution targets: the watchdog fires and emits the structured log, but the kill is best-effort because remote pids aren't owned by this process. The platform-level 1h safety net still applies. Out of scope for NEE-81 by design ## Model Used - Provider: Anthropic Claude - Model: claude-opus-4-7 - Mode: Claude Code (extended thinking, tool use) ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [ ] If this change affects the UI, I have included before/after screenshots (N/A — no UI changes) - [x] I have updated relevant documentation to reflect my changes (`agentConfigurationDoc` for `outputInactivityTimeoutMs`) - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green (re-running on the rebased head) - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups (P2 addressed; review threads resolved) - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Neeraj Kumar Singh Co-authored-by: Devin Foley Co-authored-by: Paperclip --- packages/adapters/codex-local/src/index.ts | 1 + .../codex-local/src/server/execute.ts | 207 ++++++++++++-- ...put-inactivity-monitor.integration.test.ts | 100 +++++++ .../server/output-inactivity-monitor.test.ts | 260 ++++++++++++++++++ .../src/server/output-inactivity-monitor.ts | 142 ++++++++++ 5 files changed, 684 insertions(+), 26 deletions(-) create mode 100644 packages/adapters/codex-local/src/server/output-inactivity-monitor.integration.test.ts create mode 100644 packages/adapters/codex-local/src/server/output-inactivity-monitor.test.ts create mode 100644 packages/adapters/codex-local/src/server/output-inactivity-monitor.ts diff --git a/packages/adapters/codex-local/src/index.ts b/packages/adapters/codex-local/src/index.ts index 944d0369..b2e9df7a 100644 --- a/packages/adapters/codex-local/src/index.ts +++ b/packages/adapters/codex-local/src/index.ts @@ -87,6 +87,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". Notes: - Prompts are piped via stdin (Codex receives "-" prompt argument). diff --git a/packages/adapters/codex-local/src/server/execute.ts b/packages/adapters/codex-local/src/server/execute.ts index e3c38d49..fc9e8417 100644 --- a/packages/adapters/codex-local/src/server/execute.ts +++ b/packages/adapters/codex-local/src/server/execute.ts @@ -57,6 +57,12 @@ import { prepareCodexRuntimeConfig } from "./runtime-config.js"; import { resolveCodexDesiredSkillNames } from "./skills.js"; import { buildCodexExecArgs } from "./codex-args.js"; import { SANDBOX_INSTALL_COMMAND } from "../index.js"; +import { + CODEX_OUTPUT_INACTIVITY_MONITOR_SIGTERM_GRACE_MS, + createCodexOutputInactivityMonitor, + formatOutputInactivityMonitorErrorMessage, + resolveCodexInactivityTimeout, +} from "./output-inactivity-monitor.js"; const __moduleDir = path.dirname(fileURLToPath(import.meta.url)); const CODEX_ROLLOUT_NOISE_RE = @@ -86,6 +92,29 @@ function firstNonEmptyLine(text: string): string { ); } +function signalCodexChild( + target: { pid: number | null; processGroupId: number | null }, + signal: NodeJS.Signals, +): boolean { + if (process.platform !== "win32" && target.processGroupId && target.processGroupId > 0) { + try { + process.kill(-target.processGroupId, signal); + return true; + } catch { + // Fall back to direct child signal if group signaling fails (e.g. group already gone). + } + } + if (target.pid && target.pid > 0) { + try { + process.kill(target.pid, signal); + return true; + } catch { + return false; + } + } + return false; +} + function hasNonEmptyEnvValue(env: Record, key: string): boolean { const raw = env[key]; return typeof raw === "string" && raw.trim().length > 0; @@ -585,6 +614,18 @@ export async function execute(ctx: AdapterExecutionContext): Promise { - if (stream !== "stderr") { - await onLog(stream, chunk); - return; - } - const cleaned = stripCodexRolloutNoise(chunk); - if (!cleaned.trim()) return; - await onLog(stream, cleaned); - }, - }); - const cleanedStderr = stripCodexRolloutNoise(proc.stderr); - return { - proc: { - ...proc, - stderr: cleanedStderr, - }, - rawStderr: proc.stderr, - parsed: parseCodexJsonl(proc.stdout), + let monitorFired = false; + let monitorTerminationSignal: NodeJS.Signals | null = null; + let monitorElapsedMs = 0; + let monitorTimeoutMs = 0; + let killTarget: { pid: number | null; processGroupId: number | null } | null = null; + let sigkillTimer: ReturnType | null = null; + let monitorLogPromise: Promise | null = null; + + const monitor = + monitorResolution.mode === "disabled" + ? null + : createCodexOutputInactivityMonitor({ + timeoutMs: monitorResolution.timeoutMs, + onFire: (state) => { + monitorFired = true; + monitorElapsedMs = (state.firedAt ?? Date.now()) - state.lastEventAt; + monitorTimeoutMs = monitorResolution.timeoutMs; + const message = formatOutputInactivityMonitorErrorMessage(monitorElapsedMs); + const elapsedSec = Math.round(monitorElapsedMs / 1000); + const timeoutSecLabel = Math.round(monitorResolution.timeoutMs / 1000); + const logLine = + `[paperclip] adapter.invoke ${message}; ` + + `timeoutMs=${monitorResolution.timeoutMs} elapsedSinceLastEventMs=${monitorElapsedMs} ` + + `parsedEvents=${state.parsedEventCount} (timeout=${timeoutSecLabel}s elapsed=${elapsedSec}s); ` + + `terminating codex child via SIGTERM (5s grace, then SIGKILL).\n`; + // Issue the log without awaiting on the kill hot path, but capture + // the promise so the surrounding try/finally can await flush before + // the run resolves. Without this the diagnostic that explains the + // kill could be dropped if the child exits faster than onLog flushes. + monitorLogPromise = Promise.resolve(onLog("stderr", logLine)).catch(() => {}); + const target = killTarget; + if (!target || (target.pid == null && target.processGroupId == null)) { + return; + } + const sentSig = signalCodexChild(target, "SIGTERM"); + if (sentSig) monitorTerminationSignal = "SIGTERM"; + sigkillTimer = setTimeout(() => { + sigkillTimer = null; + const stillSent = signalCodexChild(target, "SIGKILL"); + if (stillSent) monitorTerminationSignal = "SIGKILL"; + }, CODEX_OUTPUT_INACTIVITY_MONITOR_SIGTERM_GRACE_MS); + if (typeof (sigkillTimer as { unref?: () => void }).unref === "function") { + (sigkillTimer as { unref: () => void }).unref(); + } + }, + }); + + const wrappedOnSpawn = async (meta: { pid: number; processGroupId: number | null; startedAt: string }) => { + killTarget = { pid: meta.pid ?? null, processGroupId: meta.processGroupId }; + if (onSpawn) { + await onSpawn(meta); + } }; + + try { + const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, { + cwd, + env, + stdin: prompt, + timeoutSec, + graceSec, + onSpawn: wrappedOnSpawn, + onLog: async (stream, chunk) => { + if (stream === "stdout") { + monitor?.noteStdoutChunk(chunk); + await onLog(stream, chunk); + return; + } + const cleaned = stripCodexRolloutNoise(chunk); + if (!cleaned.trim()) return; + await onLog(stream, cleaned); + }, + }); + const cleanedStderr = stripCodexRolloutNoise(proc.stderr); + return { + proc: { + ...proc, + stderr: cleanedStderr, + }, + rawStderr: proc.stderr, + parsed: parseCodexJsonl(proc.stdout), + monitor: monitorFired + ? { + fired: true as const, + terminationSignal: monitorTerminationSignal, + elapsedMsSinceLastEvent: monitorElapsedMs, + timeoutMs: monitorTimeoutMs, + } + : { fired: false as const }, + }; + } finally { + monitor?.stop(); + if (sigkillTimer) { + clearTimeout(sigkillTimer); + sigkillTimer = null; + } + if (monitorLogPromise) { + await monitorLogPromise; + monitorLogPromise = null; + } + } }; const toResult = ( - attempt: { proc: { exitCode: number | null; signal: string | null; timedOut: boolean; stdout: string; stderr: string }; rawStderr: string; parsed: ReturnType }, + attempt: { + proc: { exitCode: number | null; signal: string | null; timedOut: boolean; stdout: string; stderr: string }; + rawStderr: string; + parsed: ReturnType; + monitor?: + | { fired: false } + | { fired: true; terminationSignal: NodeJS.Signals | null; elapsedMsSinceLastEvent: number; timeoutMs: number }; + }, clearSessionOnMissingSession = false, isRetry = false, ): AdapterExecutionResult => { + if (attempt.monitor?.fired) { + const errorMessage = formatOutputInactivityMonitorErrorMessage(attempt.monitor.elapsedMsSinceLastEvent); + return { + exitCode: null, + signal: attempt.monitor.terminationSignal ?? attempt.proc.signal, + timedOut: false, + errorMessage, + errorCode: "codex_output_inactivity_monitor", + errorFamily: null, + usage: attempt.parsed.usage, + sessionId: null, + sessionParams: null, + sessionDisplayId: null, + provider: "openai", + biller: resolveCodexBiller(effectiveEnv, billingType), + model, + billingType, + costUsd: null, + resultJson: { + stdout: attempt.proc.stdout, + stderr: attempt.proc.stderr, + outputInactivityMonitor: { + kind: "output_inactivity", + timeoutMs: attempt.monitor.timeoutMs, + elapsedMsSinceLastEvent: attempt.monitor.elapsedMsSinceLastEvent, + terminationSignal: attempt.monitor.terminationSignal, + }, + }, + summary: attempt.parsed.summary, + clearSession: clearSessionOnMissingSession, + }; + } if (attempt.proc.timedOut) { return { exitCode: attempt.proc.exitCode, 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 new file mode 100644 index 00000000..e6fadbdf --- /dev/null +++ b/packages/adapters/codex-local/src/server/output-inactivity-monitor.integration.test.ts @@ -0,0 +1,100 @@ +import { describe, expect, it } from "vitest"; +import { runChildProcess } from "@paperclipai/adapter-utils/server-utils"; +import { + CODEX_OUTPUT_INACTIVITY_MONITOR_SIGTERM_GRACE_MS, + createCodexOutputInactivityMonitor, + formatOutputInactivityMonitorErrorMessage, +} from "./output-inactivity-monitor.js"; + +const FAKE_CODEX_SCRIPT = ` +process.stdout.write(JSON.stringify({ type: "thread.started", thread_id: "abc" }) + "\\n"); +// Simulate a wedged codex: read stdin forever, never write again. +process.stdin.resume(); +process.stdin.on("data", () => {}); +setInterval(() => {}, 60_000); +`; + +describe("codex inactivity monitor (integration: real subprocess)", () => { + it( + "kills a codex child that goes silent after one event and surfaces a monitor failure", + async () => { + const runId = `monitor-integration-${Date.now()}`; + const timeoutMs = 250; + const logs: Array<{ stream: string; chunk: string }> = []; + let killTarget: { pid: number | null; processGroupId: number | null } | null = null; + let monitorFired = false; + let terminationSignal: NodeJS.Signals | null = null; + let sigkillTimer: ReturnType | null = null; + let elapsedMs = 0; + + const kill = (signal: NodeJS.Signals) => { + const target = killTarget; + if (!target) return false; + if (target.processGroupId && target.processGroupId > 0) { + try { + process.kill(-target.processGroupId, signal); + return true; + } catch { + /* fall through */ + } + } + if (target.pid && target.pid > 0) { + try { + process.kill(target.pid, signal); + return true; + } catch { + return false; + } + } + return false; + }; + + const monitor = createCodexOutputInactivityMonitor({ + timeoutMs, + onFire: (state) => { + monitorFired = true; + elapsedMs = (state.firedAt ?? Date.now()) - state.lastEventAt; + if (kill("SIGTERM")) terminationSignal = "SIGTERM"; + sigkillTimer = setTimeout(() => { + if (kill("SIGKILL")) terminationSignal = "SIGKILL"; + }, CODEX_OUTPUT_INACTIVITY_MONITOR_SIGTERM_GRACE_MS); + }, + }); + + try { + const proc = await runChildProcess(runId, process.execPath, ["-e", FAKE_CODEX_SCRIPT], { + cwd: process.cwd(), + env: process.env as Record, + timeoutSec: 30, + graceSec: 1, + onSpawn: async (meta) => { + killTarget = { pid: meta.pid, processGroupId: meta.processGroupId }; + }, + onLog: async (stream, chunk) => { + logs.push({ stream, chunk }); + if (stream === "stdout") { + monitor.noteStdoutChunk(chunk); + } + }, + }); + + expect(monitorFired, "monitor should fire when codex goes silent").toBe(true); + // Process was killed by our signal, not by hitting timeoutSec. + expect(proc.timedOut).toBe(false); + expect(["SIGTERM", "SIGKILL"]).toContain(proc.signal); + expect(["SIGTERM", "SIGKILL"]).toContain(terminationSignal); + // The errorMessage shape mirrors the AdapterExecutionResult that + // execute.ts will produce for this case. + expect(formatOutputInactivityMonitorErrorMessage(elapsedMs)).toMatch( + /^monitor: no codex output for \d+m \d+s$/, + ); + // We should have observed exactly one parsed JSONL event before silence. + expect(monitor.state().parsedEventCount).toBe(1); + } finally { + monitor.stop(); + if (sigkillTimer) clearTimeout(sigkillTimer); + } + }, + 15_000, + ); +}); 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 new file mode 100644 index 00000000..11d63327 --- /dev/null +++ b/packages/adapters/codex-local/src/server/output-inactivity-monitor.test.ts @@ -0,0 +1,260 @@ +import { describe, expect, it } from "vitest"; +import { + CODEX_OUTPUT_INACTIVITY_MONITOR_SIGTERM_GRACE_MS, + DEFAULT_CODEX_OUTPUT_INACTIVITY_TIMEOUT_MS, + createCodexOutputInactivityMonitor, + formatOutputInactivityMonitorErrorMessage, + resolveCodexInactivityTimeout, +} from "./output-inactivity-monitor.js"; + +class FakeClock { + private nowMs = 0; + private nextHandle = 1; + private timers = new Map void }>(); + + now(): number { + return this.nowMs; + } + + setTimer(cb: () => void, ms: number): number { + const handle = this.nextHandle++; + this.timers.set(handle, { fireAt: this.nowMs + ms, cb }); + return handle; + } + + clearTimer(handle: unknown): void { + if (typeof handle === "number") this.timers.delete(handle); + } + + advance(ms: number): void { + const targetMs = this.nowMs + ms; + while (true) { + let nextHandle: number | null = null; + let nextTimer: { fireAt: number; cb: () => void } | null = null; + for (const [h, timer] of this.timers) { + if (timer.fireAt <= targetMs && (!nextTimer || timer.fireAt < nextTimer.fireAt)) { + nextHandle = h; + nextTimer = timer; + } + } + if (!nextTimer || nextHandle == null) break; + this.timers.delete(nextHandle); + this.nowMs = nextTimer.fireAt; + nextTimer.cb(); + } + this.nowMs = targetMs; + } + + pendingTimerCount(): number { + return this.timers.size; + } +} + +describe("resolveCodexInactivityTimeout", () => { + it("uses default when value is unset", () => { + expect(resolveCodexInactivityTimeout(undefined)).toEqual({ + mode: "default", + timeoutMs: DEFAULT_CODEX_OUTPUT_INACTIVITY_TIMEOUT_MS, + }); + }); + + it("treats explicit null as disabled", () => { + expect(resolveCodexInactivityTimeout(null)).toEqual({ + mode: "disabled", + reason: "explicit_null", + }); + }); + + it("returns configured value for positive numbers", () => { + expect(resolveCodexInactivityTimeout(12_000)).toEqual({ + mode: "configured", + timeoutMs: 12_000, + }); + }); + + it("falls back to default for non-positive numbers", () => { + expect(resolveCodexInactivityTimeout(0)).toEqual({ + mode: "default", + timeoutMs: DEFAULT_CODEX_OUTPUT_INACTIVITY_TIMEOUT_MS, + reason: "non_positive", + }); + expect(resolveCodexInactivityTimeout(-100)).toEqual({ + mode: "default", + timeoutMs: DEFAULT_CODEX_OUTPUT_INACTIVITY_TIMEOUT_MS, + reason: "non_positive", + }); + }); + + it("falls back to default for non-number, non-null values", () => { + expect(resolveCodexInactivityTimeout("420000")).toEqual({ + mode: "default", + timeoutMs: DEFAULT_CODEX_OUTPUT_INACTIVITY_TIMEOUT_MS, + }); + }); +}); + +describe("formatOutputInactivityMonitorErrorMessage", () => { + it("formats minutes and seconds", () => { + expect(formatOutputInactivityMonitorErrorMessage(0)).toBe("monitor: no codex output for 0m 0s"); + expect(formatOutputInactivityMonitorErrorMessage(7 * 60 * 1000)).toBe("monitor: no codex output for 7m 0s"); + expect(formatOutputInactivityMonitorErrorMessage(7 * 60 * 1000 + 12_000)).toBe("monitor: no codex output for 7m 12s"); + expect(formatOutputInactivityMonitorErrorMessage(45_000)).toBe("monitor: no codex output for 0m 45s"); + }); +}); + +describe("createCodexOutputInactivityMonitor (acceptance criteria 1: fires)", () => { + it("fires after timeoutMs when child emits one event then goes silent", () => { + const clock = new FakeClock(); + const fires: Array<{ elapsed: number; parsedEventCount: number }> = []; + const monitor = createCodexOutputInactivityMonitor({ + timeoutMs: 7 * 60 * 1000, + now: () => clock.now(), + setTimer: (cb, ms) => clock.setTimer(cb, ms), + clearTimer: (handle) => clock.clearTimer(handle), + onFire: (state) => { + fires.push({ + elapsed: (state.firedAt ?? 0) - state.lastEventAt, + parsedEventCount: state.parsedEventCount, + }); + }, + }); + + // One event right after spawn. + clock.advance(50); + monitor.noteStdoutChunk('{"type":"thread.started","thread_id":"abc"}\n'); + expect(fires).toHaveLength(0); + expect(monitor.state().parsedEventCount).toBe(1); + + // Now go silent for 7 minutes; monitor should fire exactly at threshold. + clock.advance(7 * 60 * 1000 - 1); + expect(fires).toHaveLength(0); + clock.advance(1); + expect(fires).toHaveLength(1); + expect(fires[0].elapsed).toBe(7 * 60 * 1000); + expect(fires[0].parsedEventCount).toBe(1); + + // Stopping after fire is a no-op for the timer but returns final state. + const finalState = monitor.stop(); + expect(finalState.fired).toBe(true); + }); + + it("only fires once even if more silence elapses after firing", () => { + const clock = new FakeClock(); + let fireCount = 0; + const monitor = createCodexOutputInactivityMonitor({ + timeoutMs: 1_000, + now: () => clock.now(), + setTimer: (cb, ms) => clock.setTimer(cb, ms), + clearTimer: (handle) => clock.clearTimer(handle), + onFire: () => { + fireCount += 1; + }, + }); + clock.advance(2_000); + expect(fireCount).toBe(1); + clock.advance(10_000); + expect(fireCount).toBe(1); + monitor.stop(); + }); + + it("ignores non-JSON lines when resetting the timer", () => { + const clock = new FakeClock(); + let fireCount = 0; + const monitor = createCodexOutputInactivityMonitor({ + timeoutMs: 1_000, + now: () => clock.now(), + setTimer: (cb, ms) => clock.setTimer(cb, ms), + clearTimer: (handle) => clock.clearTimer(handle), + onFire: () => { + 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); + expect(fireCount).toBe(1); + monitor.stop(); + }); +}); + +describe("createCodexOutputInactivityMonitor (acceptance criteria 2: does not fire)", () => { + it("does not fire when events arrive every (threshold - 1s)", () => { + 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; + }, + }); + + // 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`); + expect(fireCount).toBe(0); + } + + // Final event lets us "complete" — total state shows 12 parsed events. + expect(monitor.state().parsedEventCount).toBe(12); + expect(fireCount).toBe(0); + + // Stop cleanly before the timer would have fired. + monitor.stop(); + expect(fireCount).toBe(0); + }); + + it("multiple events in one chunk all reset the timer", () => { + const clock = new FakeClock(); + let fireCount = 0; + const monitor = createCodexOutputInactivityMonitor({ + timeoutMs: 1_000, + now: () => clock.now(), + setTimer: (cb, ms) => clock.setTimer(cb, ms), + clearTimer: (handle) => clock.clearTimer(handle), + onFire: () => { + fireCount += 1; + }, + }); + clock.advance(500); + monitor.noteStdoutChunk( + '{"type":"thread.started","thread_id":"a"}\n{"type":"item.completed","item":{"type":"agent_message","text":"hi"}}\n', + ); + expect(monitor.state().parsedEventCount).toBe(2); + // Now wait 999ms — should still not fire. + clock.advance(999); + expect(fireCount).toBe(0); + // Wait one more ms — fires now. + clock.advance(1); + expect(fireCount).toBe(1); + monitor.stop(); + }); +}); + +describe("createCodexOutputInactivityMonitor (acceptance criteria 3: disabled)", () => { + it("resolveCodexInactivityTimeout returns disabled for null and the adapter creates no monitor", () => { + const resolution = resolveCodexInactivityTimeout(null); + expect(resolution.mode).toBe("disabled"); + // Sanity: when a caller (execute.ts) honors `disabled`, it must not + // construct a monitor at all. Verify the constructor would otherwise + // require a positive timeoutMs. + expect(() => + createCodexOutputInactivityMonitor({ + timeoutMs: 0, + onFire: () => {}, + }), + ).toThrow(/timeoutMs > 0/); + }); +}); + +describe("CODEX_OUTPUT_INACTIVITY_MONITOR_SIGTERM_GRACE_MS", () => { + it("matches the 5-second grace window required by NEE-81", () => { + expect(CODEX_OUTPUT_INACTIVITY_MONITOR_SIGTERM_GRACE_MS).toBe(5_000); + }); +}); diff --git a/packages/adapters/codex-local/src/server/output-inactivity-monitor.ts b/packages/adapters/codex-local/src/server/output-inactivity-monitor.ts new file mode 100644 index 00000000..93633909 --- /dev/null +++ b/packages/adapters/codex-local/src/server/output-inactivity-monitor.ts @@ -0,0 +1,142 @@ +import { parseJson } from "@paperclipai/adapter-utils/server-utils"; + +export const DEFAULT_CODEX_OUTPUT_INACTIVITY_TIMEOUT_MS = 7 * 60 * 1000; +export const CODEX_OUTPUT_INACTIVITY_MONITOR_SIGTERM_GRACE_MS = 5_000; + +export type CodexOutputInactivityMonitorResolution = + | { mode: "default"; timeoutMs: number } + | { mode: "configured"; timeoutMs: number } + | { mode: "disabled"; reason: "explicit_null" } + | { mode: "default"; timeoutMs: number; reason: "non_positive" }; + +/** + * Resolve the inactivity monitor timeout from raw adapter config. + * + * - `null` → disabled (explicit escape hatch). + * - missing/`undefined` → default 7m. + * - number > 0 → configured value. + * - number ≤ 0 → default 7m (and a `non_positive` note for logging). + */ +export function resolveCodexInactivityTimeout(rawValue: unknown): CodexOutputInactivityMonitorResolution { + if (rawValue === null) return { mode: "disabled", reason: "explicit_null" }; + if (typeof rawValue === "number" && Number.isFinite(rawValue)) { + if (rawValue > 0) return { mode: "configured", timeoutMs: rawValue }; + return { mode: "default", timeoutMs: DEFAULT_CODEX_OUTPUT_INACTIVITY_TIMEOUT_MS, reason: "non_positive" }; + } + return { mode: "default", timeoutMs: DEFAULT_CODEX_OUTPUT_INACTIVITY_TIMEOUT_MS }; +} + +export interface CodexOutputInactivityMonitorState { + fired: boolean; + spawnedAt: number; + lastEventAt: number; + firedAt: number | null; + parsedEventCount: number; +} + +export interface CodexOutputInactivityMonitorOptions { + timeoutMs: number; + onFire: (state: CodexOutputInactivityMonitorState) => void; + now?: () => number; + setTimer?: (cb: () => void, ms: number) => unknown; + clearTimer?: (handle: unknown) => void; + /** + * Per-line predicate. When omitted, any line that successfully parses as + * JSON via the codex JSONL parser counts as a heartbeat event. + */ + isHeartbeatLine?: (line: string) => boolean; +} + +export interface CodexOutputInactivityMonitorHandle { + noteStdoutChunk(chunk: string): void; + /** Returns the current state without stopping the timer. */ + state(): CodexOutputInactivityMonitorState; + /** Cancels any pending timer and returns the final state. */ + stop(): CodexOutputInactivityMonitorState; +} + +function defaultIsHeartbeatLine(line: string): boolean { + const trimmed = line.trim(); + if (!trimmed) return false; + return parseJson(trimmed) !== null; +} + +export function createCodexOutputInactivityMonitor( + options: CodexOutputInactivityMonitorOptions, +): CodexOutputInactivityMonitorHandle { + const now = options.now ?? (() => Date.now()); + const setTimer = options.setTimer ?? ((cb, ms) => setTimeout(cb, ms)); + const clearTimer = options.clearTimer ?? ((h) => clearTimeout(h as ReturnType)); + const isHeartbeatLine = options.isHeartbeatLine ?? defaultIsHeartbeatLine; + const timeoutMs = options.timeoutMs; + + if (!(timeoutMs > 0)) { + throw new Error(`createCodexOutputInactivityMonitor requires timeoutMs > 0 (got ${timeoutMs})`); + } + + const spawnedAt = now(); + const state: CodexOutputInactivityMonitorState = { + fired: false, + spawnedAt, + lastEventAt: spawnedAt, + firedAt: null, + parsedEventCount: 0, + }; + let timerHandle: unknown = null; + let stopped = false; + + const fire = () => { + if (state.fired || stopped) return; + state.fired = true; + state.firedAt = now(); + timerHandle = null; + options.onFire({ ...state }); + }; + + const arm = () => { + if (stopped || state.fired) return; + if (timerHandle != null) clearTimer(timerHandle); + timerHandle = setTimer(fire, timeoutMs); + }; + + 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; + } + } + if (sawHeartbeat) { + state.lastEventAt = now(); + arm(); + } + }, + state() { + return { ...state }; + }, + stop() { + stopped = true; + if (timerHandle != null) { + clearTimer(timerHandle); + timerHandle = null; + } + return { ...state }; + }, + }; +} + +/** + * Format the inactivity monitor error message in the canonical + * `monitor: no codex output for {N}m {S}s` shape consumed by NEE-81. + */ +export function formatOutputInactivityMonitorErrorMessage(elapsedMs: number): string { + const total = Math.max(0, Math.round(elapsedMs / 1000)); + const minutes = Math.floor(total / 60); + const seconds = total - minutes * 60; + return `monitor: no codex output for ${minutes}m ${seconds}s`; +}