fix(codex): preserve silent active builds (#10153)
## Thinking Path > - Paperclip is the open source control plane people use to coordinate AI agents and their work. > - Local adapters are responsible for observing agent subprocesses and terminating runs that are genuinely stuck. > - The Codex local adapter currently treats output bytes as its only liveness signal for the 30-minute inactivity monitor. > - Healthy compile and test loops can consume CPU and perform disk I/O for longer than that without producing terminal output. > - Terminating those runs loses valid work, while removing the monitor entirely would allow truly wedged processes to run indefinitely. > - This pull request adds a Linux process-group activity probe that recognizes meaningful CPU, disk I/O, and child-process churn while retaining the existing timeout for idle processes. > - The benefit is that long silent builds can finish without weakening the adapter's hung-run safety net. ## Linked Issues or Issue Description No public GitHub issue was found in the duplicate/related search. **What happened** A healthy Codex local run executing a long compile/test loop could be terminated at the default 30-minute output-inactivity threshold when the child process emitted no stdout or stderr. **Expected behavior** The inactivity monitor should keep a silent run alive while its process group is doing meaningful work, but should still terminate a process group that is alive and idle. **Steps to reproduce** 1. Run Codex local with the default `outputInactivityTimeoutMs`. 2. Have the agent start a compile or test command that consumes CPU or disk I/O without terminal output for longer than the threshold. 3. Observe the adapter terminate the otherwise healthy process group as output-inactive. **Affected version / deployment mode** Observed on a local-process Paperclip deployment using the Codex local adapter with the 30-minute default inactivity monitor. ## What Changed - Added a Linux `/proc` process-group sampler that tracks meaningful CPU tick growth, disk I/O growth, and child-process membership changes. - Reset the existing Codex inactivity timer when that sampler observes real process work, while leaving remote and non-Linux behavior unchanged. - Added diagnostics for the number of process-activity resets and documented the expanded liveness semantics. - Added unit coverage for process-activity timer resets and subprocess regressions for both a long silent CPU build and a genuinely wedged child. ## Verification - `pnpm --filter @paperclipai/adapter-codex-local typecheck` - `pnpm exec vitest run packages/adapters/codex-local/src/server/process-activity-monitor.test.ts packages/adapters/codex-local/src/server/output-inactivity-monitor.test.ts packages/adapters/codex-local/src/server/output-inactivity-monitor.integration.test.ts` - Focused result: 23 tests passed, including a silent CPU-bound subprocess that runs four times beyond the simulated inactivity window and an idle subprocess that is still terminated. ## Risks - Low risk and Linux-scoped: the new probe reads `/proc` every 15 seconds only while a monitored local Codex child is running. - The CPU threshold requires sustained work rather than any single scheduler tick, reducing the risk that a nearly idle event loop is treated as productive. - If `/proc` sampling is unavailable or fails, the adapter falls back to the existing output-only behavior. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - OpenAI Codex with model `gpt-5.6-sol`, high reasoning effort, terminal/tool execution, code editing, and test execution. The runtime did not expose a context-window size. ## 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 not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
GitHub
parent
e4662b1f9d
commit
b3b07ae44e
|
|
@ -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 whenever the child emits stdout or stderr bytes, including non-JSON progress from long-running verification commands. Defaults to 30 * 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/stderr bytes or, on Linux, its process group shows meaningful CPU, disk I/O, or child-process churn during a silent build. Defaults to 30 * 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 activity (output or process) 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
|
||||
|
|
|
|||
|
|
@ -87,6 +87,11 @@ import {
|
|||
formatOutputInactivityMonitorErrorMessage,
|
||||
resolveCodexInactivityTimeout,
|
||||
} from "./output-inactivity-monitor.js";
|
||||
import {
|
||||
CODEX_PROCESS_ACTIVITY_POLL_INTERVAL_MS,
|
||||
createCodexProcessActivityMonitor,
|
||||
type CodexProcessActivityMonitorHandle,
|
||||
} from "./process-activity-monitor.js";
|
||||
import {
|
||||
createCodexAcpExecutor,
|
||||
formatCodexAcpFallbackMessage,
|
||||
|
|
@ -1051,6 +1056,8 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
let killTarget: { pid: number | null; processGroupId: number | null } | null = null;
|
||||
let sigkillTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
let monitorLogPromise: Promise<unknown> | null = null;
|
||||
const processActivityMonitor: { current: CodexProcessActivityMonitorHandle | null } = { current: null };
|
||||
const resolvedMonitorTimeoutMs = monitorResolution.mode === "disabled" ? null : monitorResolution.timeoutMs;
|
||||
|
||||
const monitor =
|
||||
monitorResolution.mode === "disabled"
|
||||
|
|
@ -1068,7 +1075,8 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
`[paperclip] adapter.invoke ${message}; ` +
|
||||
`timeoutMs=${monitorResolution.timeoutMs} elapsedSinceLastEventMs=${monitorElapsedMs} ` +
|
||||
`outputChunkCount=${state.outputChunkCount} outputBytes=${state.outputBytes} ` +
|
||||
`parsedEvents=${state.parsedEventCount} (timeout=${timeoutSecLabel}s elapsed=${elapsedSec}s); ` +
|
||||
`parsedEvents=${state.parsedEventCount} processActivityCount=${state.processActivityCount} ` +
|
||||
`(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
|
||||
|
|
@ -1094,6 +1102,17 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
|
||||
const wrappedOnSpawn = async (meta: { pid: number; processGroupId: number | null; startedAt: string }) => {
|
||||
killTarget = { pid: meta.pid ?? null, processGroupId: meta.processGroupId };
|
||||
if (monitor && resolvedMonitorTimeoutMs !== null && !executionTargetIsRemote) {
|
||||
processActivityMonitor.current = createCodexProcessActivityMonitor({
|
||||
pid: meta.pid,
|
||||
processGroupId: meta.processGroupId,
|
||||
intervalMs: Math.min(
|
||||
CODEX_PROCESS_ACTIVITY_POLL_INTERVAL_MS,
|
||||
Math.max(1_000, Math.floor(resolvedMonitorTimeoutMs / 4)),
|
||||
),
|
||||
onActivity: () => monitor.noteProcessActivity(),
|
||||
});
|
||||
}
|
||||
if (onSpawn) {
|
||||
await onSpawn(meta);
|
||||
}
|
||||
|
|
@ -1139,6 +1158,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
|||
: { fired: false as const },
|
||||
};
|
||||
} finally {
|
||||
processActivityMonitor.current?.stop();
|
||||
monitor?.stop();
|
||||
if (sigkillTimer) {
|
||||
clearTimeout(sigkillTimer);
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ import {
|
|||
createCodexOutputInactivityMonitor,
|
||||
formatOutputInactivityMonitorErrorMessage,
|
||||
} from "./output-inactivity-monitor.js";
|
||||
import { createCodexProcessActivityMonitor } from "./process-activity-monitor.js";
|
||||
|
||||
const FAKE_CODEX_SCRIPT = `
|
||||
process.stdout.write(JSON.stringify({ type: "thread.started", thread_id: "abc" }) + "\\n");
|
||||
|
|
@ -15,6 +16,56 @@ setInterval(() => {}, 60_000);
|
|||
`;
|
||||
|
||||
describe("codex inactivity monitor (integration: real subprocess)", () => {
|
||||
it.skipIf(process.platform !== "linux")(
|
||||
"allows a long silent build while the child process group is consuming CPU",
|
||||
async () => {
|
||||
const runId = `monitor-active-build-${Date.now()}`;
|
||||
const timeoutMs = 500;
|
||||
const processActivityMonitor: {
|
||||
current: ReturnType<typeof createCodexProcessActivityMonitor> | null;
|
||||
} = { current: null };
|
||||
let monitorFired = false;
|
||||
const monitor = createCodexOutputInactivityMonitor({
|
||||
timeoutMs,
|
||||
onFire: () => {
|
||||
monitorFired = true;
|
||||
},
|
||||
});
|
||||
|
||||
try {
|
||||
const proc = await runChildProcess(
|
||||
runId,
|
||||
process.execPath,
|
||||
["-e", "const end = Date.now() + 2_000; while (Date.now() < end) {}"],
|
||||
{
|
||||
cwd: process.cwd(),
|
||||
env: process.env as Record<string, string>,
|
||||
timeoutSec: 5,
|
||||
graceSec: 1,
|
||||
onSpawn: async (meta) => {
|
||||
processActivityMonitor.current = createCodexProcessActivityMonitor({
|
||||
pid: meta.pid,
|
||||
processGroupId: meta.processGroupId,
|
||||
intervalMs: 50,
|
||||
onActivity: () => monitor.noteProcessActivity(),
|
||||
});
|
||||
},
|
||||
onLog: async (stream, chunk) => monitor.noteOutputChunk(stream, chunk),
|
||||
},
|
||||
);
|
||||
|
||||
expect(proc.exitCode).toBe(0);
|
||||
expect(proc.timedOut).toBe(false);
|
||||
expect(monitorFired).toBe(false);
|
||||
expect(monitor.state().processActivityCount).toBeGreaterThan(0);
|
||||
} finally {
|
||||
processActivityMonitor.current?.stop();
|
||||
monitor.stop();
|
||||
}
|
||||
},
|
||||
10_000,
|
||||
);
|
||||
|
||||
it(
|
||||
"kills a codex child that goes silent after one event and surfaces a monitor failure",
|
||||
async () => {
|
||||
|
|
@ -25,6 +76,9 @@ describe("codex inactivity monitor (integration: real subprocess)", () => {
|
|||
let monitorFired = false;
|
||||
let terminationSignal: NodeJS.Signals | null = null;
|
||||
let sigkillTimer: ReturnType<typeof setTimeout> | null = null;
|
||||
const processActivityMonitor: {
|
||||
current: ReturnType<typeof createCodexProcessActivityMonitor> | null;
|
||||
} = { current: null };
|
||||
let elapsedMs = 0;
|
||||
|
||||
const kill = (signal: NodeJS.Signals) => {
|
||||
|
|
@ -69,6 +123,12 @@ describe("codex inactivity monitor (integration: real subprocess)", () => {
|
|||
graceSec: 1,
|
||||
onSpawn: async (meta) => {
|
||||
killTarget = { pid: meta.pid, processGroupId: meta.processGroupId };
|
||||
processActivityMonitor.current = createCodexProcessActivityMonitor({
|
||||
pid: meta.pid,
|
||||
processGroupId: meta.processGroupId,
|
||||
intervalMs: 25,
|
||||
onActivity: () => monitor.noteProcessActivity(),
|
||||
});
|
||||
},
|
||||
onLog: async (stream, chunk) => {
|
||||
logs.push({ stream, chunk });
|
||||
|
|
@ -84,11 +144,12 @@ describe("codex inactivity monitor (integration: real subprocess)", () => {
|
|||
// 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$/,
|
||||
/^monitor: no codex activity \(output or process\) for \d+m \d+s$/,
|
||||
);
|
||||
// We should have observed exactly one parsed JSONL event before silence.
|
||||
expect(monitor.state().parsedEventCount).toBe(1);
|
||||
} finally {
|
||||
processActivityMonitor.current?.stop();
|
||||
monitor.stop();
|
||||
if (sigkillTimer) clearTimeout(sigkillTimer);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -99,10 +99,16 @@ describe("resolveCodexInactivityTimeout", () => {
|
|||
|
||||
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");
|
||||
expect(formatOutputInactivityMonitorErrorMessage(0)).toBe("monitor: no codex activity (output or process) for 0m 0s");
|
||||
expect(formatOutputInactivityMonitorErrorMessage(7 * 60 * 1000)).toBe(
|
||||
"monitor: no codex activity (output or process) for 7m 0s",
|
||||
);
|
||||
expect(formatOutputInactivityMonitorErrorMessage(7 * 60 * 1000 + 12_000)).toBe(
|
||||
"monitor: no codex activity (output or process) for 7m 12s",
|
||||
);
|
||||
expect(formatOutputInactivityMonitorErrorMessage(45_000)).toBe(
|
||||
"monitor: no codex activity (output or process) for 0m 45s",
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
|
|
@ -186,6 +192,29 @@ describe("createCodexOutputInactivityMonitor (acceptance criteria 1: fires)", ()
|
|||
expect(fireCount).toBe(1);
|
||||
monitor.stop();
|
||||
});
|
||||
|
||||
it("resets on process activity without output", () => {
|
||||
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(900);
|
||||
monitor.noteProcessActivity();
|
||||
expect(monitor.state().processActivityCount).toBe(1);
|
||||
clock.advance(999);
|
||||
expect(fireCount).toBe(0);
|
||||
clock.advance(1);
|
||||
expect(fireCount).toBe(1);
|
||||
monitor.stop();
|
||||
});
|
||||
});
|
||||
|
||||
describe("createCodexOutputInactivityMonitor (acceptance criteria 2: does not fire)", () => {
|
||||
|
|
|
|||
|
|
@ -34,6 +34,7 @@ export interface CodexOutputInactivityMonitorState {
|
|||
outputChunkCount: number;
|
||||
outputBytes: number;
|
||||
parsedEventCount: number;
|
||||
processActivityCount: number;
|
||||
}
|
||||
|
||||
export interface CodexOutputInactivityMonitorOptions {
|
||||
|
|
@ -51,6 +52,7 @@ export interface CodexOutputInactivityMonitorOptions {
|
|||
|
||||
export interface CodexOutputInactivityMonitorHandle {
|
||||
noteOutputChunk(stream: "stdout" | "stderr", chunk: string): void;
|
||||
noteProcessActivity(): void;
|
||||
/** Returns the current state without stopping the timer. */
|
||||
state(): CodexOutputInactivityMonitorState;
|
||||
/** Cancels any pending timer and returns the final state. */
|
||||
|
|
@ -85,6 +87,7 @@ export function createCodexOutputInactivityMonitor(
|
|||
outputChunkCount: 0,
|
||||
outputBytes: 0,
|
||||
parsedEventCount: 0,
|
||||
processActivityCount: 0,
|
||||
};
|
||||
let timerHandle: unknown = null;
|
||||
let stopped = false;
|
||||
|
|
@ -120,6 +123,12 @@ export function createCodexOutputInactivityMonitor(
|
|||
state.lastEventAt = now();
|
||||
arm();
|
||||
},
|
||||
noteProcessActivity() {
|
||||
if (stopped || state.fired) return;
|
||||
state.processActivityCount += 1;
|
||||
state.lastEventAt = now();
|
||||
arm();
|
||||
},
|
||||
state() {
|
||||
return { ...state };
|
||||
},
|
||||
|
|
@ -136,11 +145,11 @@ export function createCodexOutputInactivityMonitor(
|
|||
|
||||
/**
|
||||
* Format the inactivity monitor error message in the canonical
|
||||
* `monitor: no codex output for {N}m {S}s` shape consumed by NEE-81.
|
||||
* `monitor: no codex activity (output or process) 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`;
|
||||
return `monitor: no codex activity (output or process) for ${minutes}m ${seconds}s`;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,95 @@
|
|||
import { describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
createCodexProcessActivityMonitor,
|
||||
type CodexProcessActivitySnapshot,
|
||||
} from "./process-activity-monitor.js";
|
||||
|
||||
class PollHarness {
|
||||
private callback: (() => void) | null = null;
|
||||
|
||||
setTimer = (callback: () => void) => {
|
||||
this.callback = callback;
|
||||
return callback;
|
||||
};
|
||||
|
||||
clearTimer = () => {
|
||||
this.callback = null;
|
||||
};
|
||||
|
||||
async poll(): Promise<void> {
|
||||
await vi.waitFor(() => expect(this.callback).not.toBeNull());
|
||||
const callback = this.callback;
|
||||
this.callback = null;
|
||||
callback?.();
|
||||
}
|
||||
}
|
||||
|
||||
function snapshot(cpuTicks: number, ioBytes: number, processIds = "100"): CodexProcessActivitySnapshot {
|
||||
return { cpuTicks, ioBytes, processIds };
|
||||
}
|
||||
|
||||
describe("createCodexProcessActivityMonitor", () => {
|
||||
it("requires a baseline and ignores sub-threshold CPU changes", async () => {
|
||||
const samples = [snapshot(100, 1_000), snapshot(114, 1_000)];
|
||||
const harness = new PollHarness();
|
||||
const onActivity = vi.fn();
|
||||
const monitor = createCodexProcessActivityMonitor({
|
||||
pid: 100,
|
||||
processGroupId: 100,
|
||||
intervalMs: 15_000,
|
||||
sample: async () => samples.shift() ?? null,
|
||||
setTimer: harness.setTimer,
|
||||
clearTimer: harness.clearTimer,
|
||||
onActivity,
|
||||
});
|
||||
|
||||
await harness.poll();
|
||||
await vi.waitFor(() => expect(onActivity).not.toHaveBeenCalled());
|
||||
monitor.stop();
|
||||
});
|
||||
|
||||
it.each([
|
||||
["CPU growth", snapshot(115, 1_000)],
|
||||
["I/O growth", snapshot(100, 1_001)],
|
||||
["process-group churn", snapshot(100, 1_000, "100,101")],
|
||||
])("reports %s as process activity", async (_label, activeSnapshot) => {
|
||||
const samples = [snapshot(100, 1_000), activeSnapshot];
|
||||
const harness = new PollHarness();
|
||||
const onActivity = vi.fn();
|
||||
const monitor = createCodexProcessActivityMonitor({
|
||||
pid: 100,
|
||||
processGroupId: 100,
|
||||
intervalMs: 15_000,
|
||||
sample: async () => samples.shift() ?? null,
|
||||
setTimer: harness.setTimer,
|
||||
clearTimer: harness.clearTimer,
|
||||
onActivity,
|
||||
});
|
||||
|
||||
await harness.poll();
|
||||
await vi.waitFor(() => expect(onActivity).toHaveBeenCalledTimes(1));
|
||||
monitor.stop();
|
||||
});
|
||||
|
||||
it("resets its comparison baseline after an unavailable sample", async () => {
|
||||
const samples = [snapshot(100, 1_000), null, snapshot(200, 2_000), snapshot(215, 2_000)];
|
||||
const harness = new PollHarness();
|
||||
const onActivity = vi.fn();
|
||||
const monitor = createCodexProcessActivityMonitor({
|
||||
pid: 100,
|
||||
processGroupId: 100,
|
||||
intervalMs: 15_000,
|
||||
sample: async () => samples.shift() ?? null,
|
||||
setTimer: harness.setTimer,
|
||||
clearTimer: harness.clearTimer,
|
||||
onActivity,
|
||||
});
|
||||
|
||||
await harness.poll();
|
||||
await harness.poll();
|
||||
await vi.waitFor(() => expect(onActivity).not.toHaveBeenCalled());
|
||||
await harness.poll();
|
||||
await vi.waitFor(() => expect(onActivity).toHaveBeenCalledTimes(1));
|
||||
monitor.stop();
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,129 @@
|
|||
import fs from "node:fs/promises";
|
||||
|
||||
export const CODEX_PROCESS_ACTIVITY_POLL_INTERVAL_MS = 15_000;
|
||||
|
||||
export interface CodexProcessActivitySnapshot {
|
||||
cpuTicks: number;
|
||||
ioBytes: number;
|
||||
processIds: string;
|
||||
}
|
||||
|
||||
export interface CodexProcessActivityMonitorOptions {
|
||||
pid: number;
|
||||
processGroupId: number | null;
|
||||
onActivity: () => void;
|
||||
intervalMs?: number;
|
||||
sample?: () => Promise<CodexProcessActivitySnapshot | null>;
|
||||
setTimer?: (cb: () => void, ms: number) => unknown;
|
||||
clearTimer?: (handle: unknown) => void;
|
||||
}
|
||||
|
||||
export interface CodexProcessActivityMonitorHandle {
|
||||
stop(): void;
|
||||
}
|
||||
|
||||
function parseProcStat(stat: string): { processGroupId: number; cpuTicks: number } | null {
|
||||
const commandEnd = stat.lastIndexOf(")");
|
||||
if (commandEnd < 0) return null;
|
||||
const fields = stat.slice(commandEnd + 2).trim().split(/\s+/);
|
||||
const processGroupId = Number(fields[2]);
|
||||
const userTicks = Number(fields[11]);
|
||||
const systemTicks = Number(fields[12]);
|
||||
if (![processGroupId, userTicks, systemTicks].every(Number.isFinite)) return null;
|
||||
return { processGroupId, cpuTicks: userTicks + systemTicks };
|
||||
}
|
||||
|
||||
function parseProcIo(io: string): number {
|
||||
let bytes = 0;
|
||||
for (const line of io.split("\n")) {
|
||||
const match = /^(?:read_bytes|write_bytes):\s+(\d+)$/.exec(line.trim());
|
||||
if (match) bytes += Number(match[1]);
|
||||
}
|
||||
return bytes;
|
||||
}
|
||||
|
||||
export async function sampleCodexProcessActivity(
|
||||
pid: number,
|
||||
processGroupId: number | null,
|
||||
): Promise<CodexProcessActivitySnapshot | null> {
|
||||
if (process.platform !== "linux") return null;
|
||||
const targetProcessGroupId = processGroupId && processGroupId > 0 ? processGroupId : null;
|
||||
const entries = targetProcessGroupId ? await fs.readdir("/proc") : [String(pid)];
|
||||
const processIds: number[] = [];
|
||||
let cpuTicks = 0;
|
||||
let ioBytes = 0;
|
||||
|
||||
await Promise.all(
|
||||
entries.map(async (entry) => {
|
||||
if (!/^\d+$/.test(entry)) return;
|
||||
try {
|
||||
const parsed = parseProcStat(await fs.readFile(`/proc/${entry}/stat`, "utf8"));
|
||||
if (!parsed) return;
|
||||
if (targetProcessGroupId !== null && parsed.processGroupId !== targetProcessGroupId) return;
|
||||
if (targetProcessGroupId === null && Number(entry) !== pid) return;
|
||||
const io = await fs.readFile(`/proc/${entry}/io`, "utf8").catch(() => "");
|
||||
processIds.push(Number(entry));
|
||||
cpuTicks += parsed.cpuTicks;
|
||||
ioBytes += parseProcIo(io);
|
||||
} catch {
|
||||
// Processes can exit between listing /proc and reading their stat file.
|
||||
}
|
||||
}),
|
||||
);
|
||||
|
||||
if (processIds.length === 0) return null;
|
||||
processIds.sort((left, right) => left - right);
|
||||
return { cpuTicks, ioBytes, processIds: processIds.join(",") };
|
||||
}
|
||||
|
||||
export function createCodexProcessActivityMonitor(
|
||||
options: CodexProcessActivityMonitorOptions,
|
||||
): CodexProcessActivityMonitorHandle {
|
||||
const intervalMs = options.intervalMs ?? CODEX_PROCESS_ACTIVITY_POLL_INTERVAL_MS;
|
||||
const sample = options.sample ?? (() => sampleCodexProcessActivity(options.pid, options.processGroupId));
|
||||
const setTimer = options.setTimer ?? ((cb, ms) => setTimeout(cb, ms));
|
||||
const clearTimer = options.clearTimer ?? ((handle) => clearTimeout(handle as ReturnType<typeof setTimeout>));
|
||||
const minimumCpuTickDelta = Math.max(1, Math.floor(intervalMs / 1_000));
|
||||
let previous: CodexProcessActivitySnapshot | null = null;
|
||||
let timer: unknown = null;
|
||||
let stopped = false;
|
||||
|
||||
const schedule = () => {
|
||||
if (stopped) return;
|
||||
timer = setTimer(() => {
|
||||
void poll();
|
||||
}, intervalMs);
|
||||
if (typeof (timer as { unref?: () => void }).unref === "function") {
|
||||
(timer as { unref: () => void }).unref();
|
||||
}
|
||||
};
|
||||
|
||||
const poll = async () => {
|
||||
if (stopped) return;
|
||||
const current = await sample().catch(() => null);
|
||||
if (stopped) return;
|
||||
if (
|
||||
current &&
|
||||
previous &&
|
||||
(current.cpuTicks - previous.cpuTicks >= minimumCpuTickDelta ||
|
||||
current.ioBytes > previous.ioBytes ||
|
||||
current.processIds !== previous.processIds)
|
||||
) {
|
||||
options.onActivity();
|
||||
}
|
||||
previous = current;
|
||||
schedule();
|
||||
};
|
||||
|
||||
void poll();
|
||||
|
||||
return {
|
||||
stop() {
|
||||
stopped = true;
|
||||
if (timer != null) {
|
||||
clearTimer(timer);
|
||||
timer = null;
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
Loading…
Reference in New Issue