codex_local adapter output inactivity monitor (#5017)
## 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 <b.nirajkumarsingh@hotmail.com> Co-authored-by: Devin Foley <devin@paperclip.ing> Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Neeraj Kumar Singh B
committed by
GitHub
GitHub
parent
5ebbe67ebd
commit
7aa212296e
@@ -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).
|
||||
|
||||
@@ -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<string, string>, 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<AdapterExec
|
||||
resolvedCommand,
|
||||
});
|
||||
|
||||
const monitorResolution = resolveCodexInactivityTimeout(config.outputInactivityTimeoutMs);
|
||||
if (monitorResolution.mode === "disabled") {
|
||||
await onLog(
|
||||
"stdout",
|
||||
`[paperclip] Codex output inactivity monitor is DISABLED via adapterConfig.outputInactivityTimeoutMs=null. Hung codex runs will only be detected by the platform-level silent-run safety net.\n`,
|
||||
);
|
||||
} else if (monitorResolution.mode === "default" && "reason" in monitorResolution) {
|
||||
await onLog(
|
||||
"stdout",
|
||||
`[paperclip] Ignoring non-positive adapterConfig.outputInactivityTimeoutMs; falling back to default ${monitorResolution.timeoutMs}ms.\n`,
|
||||
);
|
||||
}
|
||||
const runtimeSessionParams = parseObject(runtime.sessionParams);
|
||||
const runtimeSessionId = asString(runtimeSessionParams.sessionId, runtime.sessionId ?? "");
|
||||
const runtimeSessionCwd = asString(runtimeSessionParams.cwd, "");
|
||||
@@ -766,39 +807,153 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
});
|
||||
}
|
||||
|
||||
const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, {
|
||||
cwd,
|
||||
env,
|
||||
stdin: prompt,
|
||||
timeoutSec,
|
||||
graceSec,
|
||||
onSpawn,
|
||||
onLog: async (stream, chunk) => {
|
||||
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<typeof setTimeout> | null = null;
|
||||
let monitorLogPromise: Promise<unknown> | 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<typeof parseCodexJsonl> },
|
||||
attempt: {
|
||||
proc: { exitCode: number | null; signal: string | null; timedOut: boolean; stdout: string; stderr: string };
|
||||
rawStderr: string;
|
||||
parsed: ReturnType<typeof parseCodexJsonl>;
|
||||
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,
|
||||
|
||||
+100
@@ -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<typeof setTimeout> | 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<string, string>,
|
||||
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,
|
||||
);
|
||||
});
|
||||
@@ -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<number, { fireAt: number; cb: () => 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);
|
||||
});
|
||||
});
|
||||
@@ -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<typeof setTimeout>));
|
||||
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`;
|
||||
}
|
||||
Reference in New Issue
Block a user