diff --git a/config/patches/node-pty@1.1.0.patch b/config/patches/node-pty@1.1.0.patch index 9e42c754598..31674490e56 100644 --- a/config/patches/node-pty@1.1.0.patch +++ b/config/patches/node-pty@1.1.0.patch @@ -1,5 +1,5 @@ diff --git a/binding.gyp b/binding.gyp -index 5f63978b07ab50aaf7523219a2170ec737a6b5db..bbd9e06136e8922f40b5779e35d4fc835f1479ab 100644 +index 5f63978b07ab50aaf7523219a2170ec737a6b5db..b4110a7c781d505ae581436cec68a10bf600734a 100644 --- a/binding.gyp +++ b/binding.gyp @@ -1,13 +1,18 @@ @@ -8,7 +8,7 @@ index 5f63978b07ab50aaf7523219a2170ec737a6b5db..bbd9e06136e8922f40b5779e35d4fc83 'dependencies': [ - " 0; count--) { +- close(low_fds[count]); + for (size_t i = 0; i <= count && i < 3; i++) { + if (low_fds[i] != -1) { + close(low_fds[i]); + } -+ } -+} -+#endif -+ + } + } + #endif + +#if defined(__APPLE__) +/** + * Orca: in-process libproc/sysctl process table @@ -792,7 +799,7 @@ index 7b4b9e1f990fbf95b51528bb56dc9717f5b87532..45197ddf4440b8599002b360bc50c4fe +struct orca_process_row { + struct kinfo_proc proc; + std::string command; -+ // argv was unreadable (another user's process): ps, being setuid root, would print more. ++ // argv was unreadable (another user's, exiting or mid-exec): setuid ps may print more. + bool command_incomplete; +}; + @@ -805,28 +812,24 @@ index 7b4b9e1f990fbf95b51528bb56dc9717f5b87532..45197ddf4440b8599002b360bc50c4fe + std::string ctype; +}; + ++static std::string ++orca_comm_command(const struct kinfo_proc& proc) { ++ return "(" + std::string(proc.kp_proc.p_comm, strnlen(proc.kp_proc.p_comm, sizeof(proc.kp_proc.p_comm))) + ")"; ++} ++ ++// Unreadable argv prints "(comm)" as ps does when KERN_PROCARGS2 fails; no path lookup, so a dead mount cannot hang it. +static void +orca_fill_command(const struct kinfo_proc& proc, std::vector& args_buf, orca_process_row* row) { -+ pid_t pid = proc.kp_proc.p_pid; + std::string raw; + if (proc.kp_proc.p_stat == SZOMB) { + raw = ""; -+ } else if (!orca_read_process_args(pid, args_buf, &raw)) { ++ } else if (!orca_read_process_args(proc.kp_proc.p_pid, args_buf, &raw)) { ++ row->command = orca_comm_command(proc); + row->command_incomplete = true; -+ char path[PROC_PIDPATHINFO_MAXSIZE]; -+ int length = proc_pidpath(pid, path, sizeof(path)); -+ if (length > 0 && (size_t)length < sizeof(path)) { -+ raw.assign(path, length); -+ } else { -+ raw = std::string("(") + -+ std::string(proc.kp_proc.p_comm, strnlen(proc.kp_proc.p_comm, sizeof(proc.kp_proc.p_comm))) + -+ ")"; -+ } ++ return; + } + if (!orca_vis_command(raw, &row->command)) { -+ row->command = "(" + -+ std::string(proc.kp_proc.p_comm, strnlen(proc.kp_proc.p_comm, sizeof(proc.kp_proc.p_comm))) + -+ ")"; ++ row->command = orca_comm_command(proc); + row->command_incomplete = true; + } +} @@ -886,9 +889,7 @@ index 7b4b9e1f990fbf95b51528bb56dc9717f5b87532..45197ddf4440b8599002b360bc50c4fe + } + return 0; +} - -- for (; count > 0; count--) { -- close(low_fds[count]); ++ +static Napi::Array +orca_process_rows_to_js(Napi::Env env, bool include_command, const std::vector& rows) { + Napi::Array result = Napi::Array::New(env, rows.size()); @@ -904,6 +905,7 @@ index 7b4b9e1f990fbf95b51528bb56dc9717f5b87532..45197ddf4440b8599002b360bc50c4fe + row.Set("status", Napi::Number::New(env, proc.kp_proc.p_stat)); + row.Set("flags", Napi::Number::New(env, proc.kp_proc.p_flag)); + row.Set("nice", Napi::Number::New(env, proc.kp_proc.p_nice)); ++ row.Set("uid", Napi::Number::New(env, proc.kp_eproc.e_ucred.cr_uid)); + row.Set("sessionLeader", Napi::Boolean::New(env, (proc.kp_eproc.e_flag & EPROC_SLEADER) != 0)); + row.Set("startSec", Napi::Number::New(env, (double)proc.kp_proc.p_starttime.tv_sec)); + row.Set("startUsec", Napi::Number::New(env, (double)proc.kp_proc.p_starttime.tv_usec)); @@ -913,7 +915,7 @@ index 7b4b9e1f990fbf95b51528bb56dc9717f5b87532..45197ddf4440b8599002b360bc50c4fe + row.Set("commandIncomplete", Napi::Boolean::New(env, entry.command_incomplete)); + } + result.Set(index++, row); - } ++ } + return result; +} + @@ -1000,41 +1002,82 @@ index 7b4b9e1f990fbf95b51528bb56dc9717f5b87532..45197ddf4440b8599002b360bc50c4fe + return promise; +} + -+Napi::Value OrcaProcessCwd(const Napi::CallbackInfo& info) { ++// PROC_PIDVNODEPATHINFO stats the cwd vnode and proc_pidpath may ask the filesystem for a path, so either can block ++// on a dead network mount: both run on a worker thread and JS bounds how many may be outstanding. ++class OrcaProcessPathWorker : public Napi::AsyncWorker { ++ public: ++ OrcaProcessPathWorker(Napi::Env env, pid_t pid, bool cwd) ++ : Napi::AsyncWorker(env), deferred_(Napi::Promise::Deferred::New(env)), pid_(pid), cwd_(cwd), found_(false) {} ++ ++ Napi::Promise Promise() { return deferred_.Promise(); } ++ ++ void Execute() override { ++ if (cwd_) { ++ struct proc_vnodepathinfo vnode; ++ int size = proc_pidinfo(pid_, PROC_PIDVNODEPATHINFO, 0, &vnode, sizeof(vnode)); ++ if (size == (int)sizeof(vnode) && vnode.pvi_cdir.vip_path[0] != '\0') { ++ path_.assign(vnode.pvi_cdir.vip_path, strnlen(vnode.pvi_cdir.vip_path, sizeof(vnode.pvi_cdir.vip_path))); ++ found_ = true; ++ } ++ return; ++ } ++ char path[PROC_PIDPATHINFO_MAXSIZE]; ++ int length = proc_pidpath(pid_, path, sizeof(path)); ++ if (length > 0 && (size_t)length < sizeof(path)) { ++ path_.assign(path, length); ++ found_ = true; ++ } ++ } ++ ++ void OnOK() override { ++ Napi::Env env = Env(); ++ deferred_.Resolve(found_ ? Napi::Value(Napi::String::New(env, path_)) : env.Null()); ++ } ++ ++ void OnError(const Napi::Error& error) override { ++ deferred_.Reject(error.Value()); ++ } ++ ++ private: ++ Napi::Promise::Deferred deferred_; ++ pid_t pid_; ++ bool cwd_; ++ bool found_; ++ std::string path_; ++}; ++ ++static Napi::Value ++orca_queue_path_worker(const Napi::CallbackInfo& info, bool cwd) { + Napi::Env env(info.Env()); + if (info.Length() < 1 || !info[0].IsNumber()) { -+ throw Napi::Error::New(env, "Usage: processCwd(pid)"); ++ throw Napi::TypeError::New(env, cwd ? "Usage: readProcessCwd(pid)" : "Usage: readProcessPath(pid)"); + } -+ struct proc_vnodepathinfo vnode; -+ int size = proc_pidinfo(info[0].As().Int32Value(), PROC_PIDVNODEPATHINFO, 0, &vnode, sizeof(vnode)); -+ if (size != (int)sizeof(vnode) || vnode.pvi_cdir.vip_path[0] == '\0') { -+ return env.Null(); -+ } -+ return Napi::String::New(env, vnode.pvi_cdir.vip_path, strnlen(vnode.pvi_cdir.vip_path, sizeof(vnode.pvi_cdir.vip_path))); ++ OrcaProcessPathWorker* worker = new OrcaProcessPathWorker(env, info[0].As().Int32Value(), cwd); ++ Napi::Promise promise = worker->Promise(); ++ worker->Queue(); ++ return promise; +} + -+Napi::Value OrcaProcessPath(const Napi::CallbackInfo& info) { -+ Napi::Env env(info.Env()); -+ if (info.Length() < 1 || !info[0].IsNumber()) { -+ throw Napi::Error::New(env, "Usage: processPath(pid)"); -+ } -+ char path[PROC_PIDPATHINFO_MAXSIZE]; -+ int length = proc_pidpath(info[0].As().Int32Value(), path, sizeof(path)); -+ if (length <= 0 || (size_t)length >= sizeof(path)) { -+ return env.Null(); -+ } -+ return Napi::String::New(env, path, length); - } - #endif - -@@ -793,6 +1307,11 @@ Napi::Object init(Napi::Env env, Napi::Object exports) { ++Napi::Value OrcaReadProcessCwd(const Napi::CallbackInfo& info) { ++ return orca_queue_path_worker(info, true); ++} ++ ++Napi::Value OrcaReadProcessPath(const Napi::CallbackInfo& info) { ++ return orca_queue_path_worker(info, false); ++} ++#endif ++ + /** + * Init + */ +@@ -793,6 +1342,11 @@ Napi::Object init(Napi::Env env, Napi::Object exports) { exports.Set("open", Napi::Function::New(env, PtyOpen)); exports.Set("resize", Napi::Function::New(env, PtyResize)); exports.Set("process", Napi::Function::New(env, PtyGetProc)); +#if defined(__APPLE__) + exports.Set("readProcessTable", Napi::Function::New(env, OrcaProcessTableAsync)); -+ exports.Set("processCwd", Napi::Function::New(env, OrcaProcessCwd)); -+ exports.Set("processPath", Napi::Function::New(env, OrcaProcessPath)); ++ exports.Set("readProcessCwd", Napi::Function::New(env, OrcaReadProcessCwd)); ++ exports.Set("readProcessPath", Napi::Function::New(env, OrcaReadProcessPath)); +#endif return exports; } diff --git a/config/process-host-direct-import-baseline.txt b/config/process-host-direct-import-baseline.txt index 9c277c2eb8a..75dcdf5d8c6 100644 --- a/config/process-host-direct-import-baseline.txt +++ b/config/process-host-direct-import-baseline.txt @@ -102,7 +102,7 @@ src/main/zsh-startup-hook-pty-harness.ts src/shared/ephemeral-vm-recipe-process.ts src/shared/ephemeral-vm-recipe-runner.ts src/shared/fish-binary-requirement.ts -src/shared/process-table-snapshot-reader.ts +src/shared/process-table-ps-capture.ts src/shared/pty-slave-line-discipline-echo.ts src/shared/shell-process-readiness.ts src/wsl-guest/preflight-handler.ts diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 1de453faf04..22dfb1c38a4 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -185,7 +185,7 @@ patchedDependencies: i18next-cli@1.74.2: 2ee32eef742615cd133e1a6940f027deb3723a8430bb7953bba3a6769b0aac47 lint-staged@16.4.0: 7333b3837f80a7fbd045964db6d76ba4fc118e49134bdbabb00585b6b7b60673 monaco-editor@0.55.1: 53b71f7d8a7c00b76ffdb73ce83d18e31692f193f5fafa6a390e48f5b3fdd636 - node-pty@1.1.0: b56279c3729be0e79c21f4d6ab207052ed949e9a792ae326fa9bd473e23347a1 + node-pty@1.1.0: f81acb44ac52265a6053af59e60128471f4655de39e64f8c47c916e0dc8c9416 oxlint-plugin-anti-slop@0.1.2: c10c871702365a0c2c59e9f1c5fb92df97cae5755d534c0a3d83c6368f6feafe pdfjs-dist@6.3.289: 424fde648e16c965ce45c55bdda49ecb3a4aa992c856900714c841c3f13f8415 sprintf-js@1.1.3: 05316a8b01f155fe19a1e0cb4042c81d43d798a76755c407684e1199a18b1e08 @@ -234,7 +234,7 @@ importers: version: 3.3.1 node-pty: specifier: ^1.1.0 - version: 1.1.0(patch_hash=b56279c3729be0e79c21f4d6ab207052ed949e9a792ae326fa9bd473e23347a1) + version: 1.1.0(patch_hash=f81acb44ac52265a6053af59e60128471f4655de39e64f8c47c916e0dc8c9416) posthog-node: specifier: ^5.52.5 version: 5.52.5 @@ -13756,7 +13756,7 @@ snapshots: node-int64@0.4.0: {} - node-pty@1.1.0(patch_hash=b56279c3729be0e79c21f4d6ab207052ed949e9a792ae326fa9bd473e23347a1): + node-pty@1.1.0(patch_hash=f81acb44ac52265a6053af59e60128471f4655de39e64f8c47c916e0dc8c9416): dependencies: node-addon-api: 7.1.1 diff --git a/src/main/daemon/pty-subprocess/spawn-file-foreground-process.ts b/src/main/daemon/pty-subprocess/spawn-file-foreground-process.ts index 46c2d5f4696..97d9b394dd1 100644 --- a/src/main/daemon/pty-subprocess/spawn-file-foreground-process.ts +++ b/src/main/daemon/pty-subprocess/spawn-file-foreground-process.ts @@ -9,7 +9,7 @@ import { getProcessTableIndex } from '../../../shared/process-table-index' import { hasControllingTty } from '../../../shared/posix-controlling-terminal' -import type { ProcessTableRow } from '../../../shared/process-table-snapshot' +import { unreadableArgvName, type ProcessTableRow } from '../../../shared/process-table-snapshot' import { getFreshProcessTableSnapshot, getProcessTableSnapshot @@ -61,7 +61,8 @@ export function resolveSpawnFileForegroundFromRows( .filter((row) => row.pgid === root.tpgid && row.tty === root.tty && !/[TZ]/.test(row.stat)) .sort((left, right) => right.depth - left.depth) const foreground = candidates[0] - if (!foreground) { + // A foreground process whose argv ps withheld could be the agent, so its absence is not proof. + if (!foreground || candidates.some((row) => unreadableArgvName(row.command) !== null)) { return { available: false, processName: null } } const name = getCommandTokenPathBasename(getFirstCommandToken(foreground.command)).replace( diff --git a/src/main/daemon/pty-subprocess/spawn-file-foreground-rejected-agents.test.ts b/src/main/daemon/pty-subprocess/spawn-file-foreground-rejected-agents.test.ts index d79dc806e5a..ffb2310eb2f 100644 --- a/src/main/daemon/pty-subprocess/spawn-file-foreground-rejected-agents.test.ts +++ b/src/main/daemon/pty-subprocess/spawn-file-foreground-rejected-agents.test.ts @@ -72,6 +72,15 @@ describe('POSIX static-name agent selection', () => { ).toEqual({ available: true, processName: null }) }) + it('reports a foreground whose argv ps withheld as unavailable, never as no agent', () => { + expect( + resolveSpawnFileForegroundFromRows( + [root, { ...root, pid: 101, ppid: 100, pgid: 101, stat: 'S+', command: '(node)' }], + 100 + ) + ).toEqual({ available: false, processName: null }) + }) + it.each(['vim', 'npm', 'sleep'])('retains the ordinary %s name', (command) => { expect( resolveSpawnFileForegroundFromRows( diff --git a/src/main/daemon/terminal-descendant-shutdown-libproc.test.ts b/src/main/daemon/terminal-descendant-shutdown-libproc.test.ts index 92d1798e648..92c72f7b837 100644 --- a/src/main/daemon/terminal-descendant-shutdown-libproc.test.ts +++ b/src/main/daemon/terminal-descendant-shutdown-libproc.test.ts @@ -28,8 +28,8 @@ beforeEach(() => { // A stalled libuv pool: native reads never finish. setLibprocAddonForTests({ readProcessTable: () => new Promise(() => {}), - processCwd: () => null, - processPath: () => null + readProcessCwd: async () => null, + readProcessPath: async () => null }) // A 200ms ps, killed at its timeout like execFile. execFile.mockImplementation((_file, _args, { timeout }, callback) => { diff --git a/src/main/providers/agent-foreground-process-batch.ts b/src/main/providers/agent-foreground-process-batch.ts index 9ff3d7dada0..da420e80ba0 100644 --- a/src/main/providers/agent-foreground-process-batch.ts +++ b/src/main/providers/agent-foreground-process-batch.ts @@ -15,7 +15,7 @@ import { type ProcessTableIndex, type ProcessTableIndexStats } from '../../shared/process-table-index' -import type { ProcessTableRow } from '../../shared/process-table-snapshot' +import { unreadableArgvName, type ProcessTableRow } from '../../shared/process-table-snapshot' import { getStrictProcessTableSnapshot } from '../../shared/process-table-snapshot-reader' import { resolveRemoteForegroundEvidenceFromRows } from './agent-foreground-process-remote-evidence' @@ -209,6 +209,14 @@ export function resolveAgentForegroundProcessesFromIndex( occupancy.rowsByProcessGroup.get(root.pgid) === 1 const allCandidates = rowsByOwner.get(root.pid) ?? [] const foregroundCandidates = allCandidates.filter((row) => row.pgid === root.tpgid) + // A foreground process whose argv ps withheld could be the agent: absence of a name proves nothing. + if (foregroundCandidates.some((row) => unreadableArgvName(row.command) !== null)) { + return { + available: false, + processName: request.fallbackProcess ?? null, + reason: 'argv_unreadable' + } + } const fallbackProcess = request.fallbackProcess const wrapperFallback = typeof fallbackProcess === 'string' && isAgentForegroundWrapperProcess(fallbackProcess) diff --git a/src/main/providers/agent-foreground-process-remote-evidence.test.ts b/src/main/providers/agent-foreground-process-remote-evidence.test.ts index 2e28a9c6107..ac3bd2b6416 100644 --- a/src/main/providers/agent-foreground-process-remote-evidence.test.ts +++ b/src/main/providers/agent-foreground-process-remote-evidence.test.ts @@ -62,10 +62,25 @@ describe('host-stamped remote foreground resolver', () => { }) }) + it('names one agent that holds two rows of the group (npm Codex spawning native codex)', () => { + const evidence = resolveRemoteForegroundEvidence( + { rootPid: 100, fallbackProcess: 'zsh' }, + metadata, + rowsFor(['node /usr/local/bin/codex', '/usr/local/lib/node_modules/@openai/codex/bin/codex']) + ) + expect(evidence).toMatchObject({ + verdict: 'live', + processName: 'codex', + fence: { process: { pid: 101, startTime: 'candidate-0' } } + }) + }) + it.each([ ['multiplexer_boundary', rowsFor(['tmux new-session'])], ['ambiguous_foreground_group', rowsFor(['node /opt/codex', 'node /opt/claude'])], - ['candidate_start_time_missing', rowsFor(['node /opt/codex'], { candidateStart: '' })] + ['candidate_start_time_missing', rowsFor(['node /opt/codex'], { candidateStart: '' })], + // An agent under sudo whose argv recovery failed or was withheld by the fork budget. + ['argv_unreadable', rowsFor(['(node)'])] ])('degrades to unverifiable for %s', (reason, rows) => { const evidence = resolveRemoteForegroundEvidence( { rootPid: 100, fallbackProcess: 'zsh' }, diff --git a/src/main/providers/agent-foreground-process-remote-evidence.ts b/src/main/providers/agent-foreground-process-remote-evidence.ts index d9fa7791498..4a76d98f247 100644 --- a/src/main/providers/agent-foreground-process-remote-evidence.ts +++ b/src/main/providers/agent-foreground-process-remote-evidence.ts @@ -85,7 +85,9 @@ export function resolveRemoteForegroundEvidenceFromRows( name: NonNullable> } => Boolean(entry.name) ) - if (recognized.length > 1) { + // One agent can hold two rows (npm Codex: node codex.js spawning native codex; Gemini relaunching itself in node): + // that names it, and the outermost row is its fence. Two different agents prove neither. + if (new Set(recognized.map((entry) => entry.name.agent)).size > 1) { return { ...metadata, verdict: 'unverifiable', reason: 'ambiguous_foreground_group' } } const candidate = recognized[0] diff --git a/src/main/providers/agent-foreground-process.ts b/src/main/providers/agent-foreground-process.ts index e517c1ed807..bc0eebca9a3 100644 --- a/src/main/providers/agent-foreground-process.ts +++ b/src/main/providers/agent-foreground-process.ts @@ -1,6 +1,6 @@ import { recognizeAgentProcessFromCommandLine } from '../../shared/agent-process-recognition' import { resolveOuterWrapperForegroundProcess } from '../../shared/foreground-wrapper-agent' -import type { ProcessTableRow } from '../../shared/process-table-snapshot' +import { unreadableArgvName, type ProcessTableRow } from '../../shared/process-table-snapshot' import { getFreshProcessTableSnapshot, getFreshShellForegroundSnapshot, @@ -164,6 +164,14 @@ export async function resolveAgentForegroundProcessWithAvailability( if (options.fresh && !getProcessTableIndex(rows).byPid.has(shellPid)) { return { available: false, processName: fallbackProcess } } + // A foreground descendant whose argv ps withheld could be the agent, so its absence is not proof. + if ( + collectDescendantsFromIndex(getProcessTableIndex(rows), shellPid).some( + (row) => row.stat.includes('+') && unreadableArgvName(row.command) !== null + ) + ) { + return { available: false, processName: fallbackProcess } + } return { available: true, processName: resolveAgentForegroundProcessFromPs(rows, shellPid) ?? fallbackProcess diff --git a/src/main/providers/process-cwd.test.ts b/src/main/providers/process-cwd.test.ts index fa8f5e55984..4f7a805d4c7 100644 --- a/src/main/providers/process-cwd.test.ts +++ b/src/main/providers/process-cwd.test.ts @@ -73,4 +73,54 @@ describe('resolveProcessCwd', () => { await expect(cwdPromise).resolves.toBe('') expect(killMock).toHaveBeenCalled() }) + + it('answers from lsof -b while in-process cwd lookups hang on a wedged mount, holding at most two pool threads', async () => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'performance'] }) + readlinkMock.mockRejectedValue(new Error('proc unavailable')) + execFileMock.mockImplementation( + ( + _file: string, + args: string[], + _options: unknown, + callback: (...result: unknown[]) => void + ) => { + callback(null, `p${args[4]}\nfcwd\nn/Volumes/dead/${args[4]}\n`) + return { kill: vi.fn() } + } + ) + const { setLibprocAddonForTests } = await import('../../shared/libproc-addon') + const readProcessCwd = vi.fn(() => new Promise(() => {})) + setLibprocAddonForTests({ + readProcessTable: async () => [], + readProcessCwd, + readProcessPath: async () => null + }) + const { resolveProcessCwd } = await import('./process-cwd') + + const cwds = Promise.all([1, 2, 3].map((pid) => resolveProcessCwd(pid))) + await vi.advanceTimersByTimeAsync(250) + + await expect(cwds).resolves.toEqual(['/Volumes/dead/1', '/Volumes/dead/2', '/Volumes/dead/3']) + expect(readProcessCwd).toHaveBeenCalledTimes(2) + setLibprocAddonForTests(undefined) + }) + + it('keeps one stuck lsof per pid instead of spawning another every poll', async () => { + // The result cache must expire between polls. + vi.restoreAllMocks() + vi.useFakeTimers() + readlinkMock.mockRejectedValue(new Error('proc unavailable')) + // SIGTERM cannot interrupt lsof waiting in the kernel on a dead mount, so it never exits. + execFileMock.mockImplementation(() => ({ kill: vi.fn() })) + const { resolveProcessCwd } = await import('./process-cwd') + + for (let poll = 0; poll < 3; poll += 1) { + const cwd = resolveProcessCwd(42) + await vi.advanceTimersByTimeAsync(1500) + await expect(cwd).resolves.toBe('') + // Past the result cache, so the next poll looks the pid up again. + await vi.advanceTimersByTimeAsync(1500) + } + expect(execFileMock).toHaveBeenCalledTimes(1) + }) }) diff --git a/src/main/providers/process-cwd.ts b/src/main/providers/process-cwd.ts index 3a4b78f27c0..e9f7a3289e8 100644 --- a/src/main/providers/process-cwd.ts +++ b/src/main/providers/process-cwd.ts @@ -1,7 +1,7 @@ import { execFile as execFileCb } from 'node:child_process' import { readlink } from 'node:fs/promises' import { LSOF_NONBLOCKING_ARGS } from '../../shared/lsof-nonblocking-args' -import { readLibprocProcessCwd } from '../../shared/libproc-addon' +import { readLibprocProcessFilePath } from '../../shared/libproc-addon' // Coalesce/cache per-PID cwd reads briefly so rapid pane actions cannot accumulate lsof subprocesses. const CACHE_TTL_MS = 1500 @@ -62,7 +62,7 @@ async function doResolve(pid: number): Promise { /* fall through */ } - const inProcess = readLibprocProcessCwd(pid) + const inProcess = await readLibprocProcessFilePath('cwd', pid) if (inProcess !== undefined) { return inProcess ?? '' } @@ -83,7 +83,13 @@ async function doResolve(pid: number): Promise { return '' } +// lsof stuck on a dead mount outlives its timeout (SIGTERM waits for the syscall); one per pid, not one per poll. +const liveLsofPids = new Set() + function readCwdWithLsof(pid: number): Promise { + if (liveLsofPids.has(pid)) { + return Promise.reject(new Error('lsof for this pid has not exited')) + } return new Promise((resolve, reject) => { let settled = false let child: ReturnType | undefined @@ -107,6 +113,7 @@ function readCwdWithLsof(pid: number): Promise { // execFile timeout only sends a signal; bound a missing callback so shared lookup cannot remain cached forever. try { + liveLsofPids.add(pid) child = execFileCb( 'lsof', [...LSOF_NONBLOCKING_ARGS, '-a', '-p', String(pid), '-d', 'cwd', '-Fn'], @@ -115,6 +122,8 @@ function readCwdWithLsof(pid: number): Promise { timeout: LSOF_TIMEOUT_MS }, (error, stdout) => { + // execFile reports only once the child has exited. + liveLsofPids.delete(pid) if (error) { settle(() => reject(error)) return @@ -123,6 +132,7 @@ function readCwdWithLsof(pid: number): Promise { } ) } catch (error) { + liveLsofPids.delete(pid) settle(() => reject(error)) } }) diff --git a/src/main/pty-descendant-fresh-table.ts b/src/main/pty-descendant-fresh-table.ts index 0248d51bd1b..85fabd26f83 100644 --- a/src/main/pty-descendant-fresh-table.ts +++ b/src/main/pty-descendant-fresh-table.ts @@ -38,7 +38,8 @@ async function readFreshProcessTableOnce( // Registered now, so no native read submitted before this request serves it; a stalled one leaves ps its full timeout. const inProcess = await runPsInProcess(DESCENDANT_PS_ARGS, { env, - timeoutMs: inProcessSliceMs(timeoutMs) + timeoutMs: inProcessSliceMs(timeoutMs), + callerRunsPs: true }) if (inProcess && !inProcess.timedOut) { return { rows: parseProcessTable(inProcess.stdout), capturedAtMs } diff --git a/src/main/pty-descendant-termination-libproc.test.ts b/src/main/pty-descendant-termination-libproc.test.ts index 446dce1b59e..0085cd0036a 100644 --- a/src/main/pty-descendant-termination-libproc.test.ts +++ b/src/main/pty-descendant-termination-libproc.test.ts @@ -79,8 +79,8 @@ describe('descendant teardown over the in-process process table', () => { .mockImplementation(async () => [row(1, 0), row(100, 1), row(200, 1), row(300, 100)]) setLibprocAddonForTests({ readProcessTable, - processCwd: () => null, - processPath: () => null + readProcessCwd: async () => null, + readProcessPath: async () => null }) expect(await captureDescendantSnapshot(100, { timeoutMs: 10 })).toBeNull() @@ -106,8 +106,8 @@ describe('descendant teardown over the in-process process table', () => { vi.useFakeTimers() setLibprocAddonForTests({ readProcessTable: () => new Promise(() => {}), - processCwd: () => null, - processPath: () => null + readProcessCwd: async () => null, + readProcessPath: async () => null }) // A 750ms ps, killed at its timeout like execFile. execFile.mockImplementation((_file, _args, { timeout }, callback) => { diff --git a/src/main/pty/posix-pty-foreground-group-libproc.test.ts b/src/main/pty/posix-pty-foreground-group-libproc.test.ts index fd222f48b91..d3053c00e0c 100644 --- a/src/main/pty/posix-pty-foreground-group-libproc.test.ts +++ b/src/main/pty/posix-pty-foreground-group-libproc.test.ts @@ -50,8 +50,8 @@ describe('foreground group signalling over the in-process reader', () => { it('gives an 80ms ps its whole timeout while a terminal naming pass is held', async () => { setLibprocAddonForTests({ readProcessTable: async () => [row(ROOT_PID)], - processCwd: () => null, - processPath: () => null + readProcessCwd: async () => null, + readProcessPath: async () => null }) claimDarwinTerminalNaming(() => new Promise(() => {})) // An 80ms ps, killed at its timeout like process-host's runProcess. diff --git a/src/main/pty/posix-pty-foreground-group.ts b/src/main/pty/posix-pty-foreground-group.ts index 77f8c7526ca..abc495b00c2 100644 --- a/src/main/pty/posix-pty-foreground-group.ts +++ b/src/main/pty/posix-pty-foreground-group.ts @@ -33,7 +33,8 @@ async function runPs(pid: number): Promise { // A native read stuck behind a busy pool leaves ps its full timeout before signalling. const inProcess = await runPsInProcess(args, { timeoutMs: inProcessSliceMs(PROCESS_TABLE_QUERY_TIMEOUT_MS), - maxOutputBytes: PROCESS_TABLE_MAX_BYTES + maxOutputBytes: PROCESS_TABLE_MAX_BYTES, + callerRunsPs: true }) const result = inProcess && !inProcess.timedOut diff --git a/src/main/runtime/launched-agent-foreground.test.ts b/src/main/runtime/launched-agent-foreground.test.ts index 2e35d7b9869..8dd67623752 100644 --- a/src/main/runtime/launched-agent-foreground.test.ts +++ b/src/main/runtime/launched-agent-foreground.test.ts @@ -2,6 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import type { RemoteForegroundEvidence } from '../../shared/foreground-process-evidence' import type { ProcessTableRow } from '../../shared/process-table-snapshot' import type * as TerminalForegroundGroup from './terminal-foreground-group' +import { resolveRemoteForegroundEvidence } from '../providers/agent-foreground-process-batch' import { readLaunchedAgentForeground } from './launched-agent-foreground' const paneTerminal = vi.hoisted(() => { @@ -74,6 +75,186 @@ describe('the relay’s foreground group as proof of a launched agent', () => { readLaunchedAgentForeground(controller, POSIX_SSH, 'pty-1', 'claude') ).resolves.toBe('shell') }) + + function unverifiable(reason: string): RemoteForegroundEvidence { + return { + verdict: 'unverifiable', + reason, + authorityGeneration: 'gen-1', + observationEpoch: 1, + capturedAgeMs: 0, + ptyId: 'pty-1', + ptyIncarnationId: 'inc-1' + } + } + + // A cached name completes only fresh evidence that saw an agent row; it never fills an observation gap. + it.each([ + ['ambiguous_foreground_group', 0, 'codex', 'codex', 'agent'], + ['multiplexer_boundary', 0, 'claude', 'claude', 'agent'], + // A busy relay's cached snapshot: up to one 500ms TTL old plus its capture. + ['multiplexer_boundary', 900, 'claude', 'claude', 'agent'], + ['candidate_start_time_missing', 0, 'claude', 'claude', 'agent'], + ['ambiguous_foreground_group', 1_500, 'codex', 'codex', 'unknown'], + // Another agent's name never completes the launched one's evidence. + ['multiplexer_boundary', 0, 'codex', 'claude', 'unknown'], + // ps failed (EAGAIN) after the tracker cached `claude` and the foreground became bash. + ['process_table_unreadable', 0, 'claude', 'claude', 'unknown'], + ['anchor_missing', 0, 'claude', 'claude', 'unknown'], + // `sudo -i` with use_pty: the root shell is on a new pty. + ['tty_boundary', 0, 'sudo', 'claude', 'unknown'], + ['fence_incomplete', 0, 'claude', 'claude', 'unknown'], + ['argv_unreadable', 0, 'claude', 'claude', 'unknown'], + ['a_reason_this_build_does_not_know', 0, 'claude', 'claude', 'unknown'], + // A relay before same-agent dedupe: npm Codex's node wrapper and native binary read ambiguous, named `node`. + ['ambiguous_foreground_group', 0, 'node', 'codex', 'agent'], + ['multiplexer_boundary', 0, 'node', 'codex', 'unknown'], + // goose is a native binary, so `node` is not its interpreter. + ['ambiguous_foreground_group', 0, 'node', 'goose', 'unknown'] + ] as const)( + '%s evidence %ims old with the relay name %s, launching %s: %s', + async (reason, capturedAgeMs, name, agent, found) => { + const controller = { + ...controllerAnswering(async () => ({ ...unverifiable(reason), capturedAgeMs })), + getForegroundProcess: async () => name, + confirmForegroundProcess: async () => name + } + + await expect( + readLaunchedAgentForeground(controller, POSIX_SSH, 'pty-1', agent) + ).resolves.toBe(found) + } + ) + + it('completes structural evidence with a name read now, never the tracker’s display cache', async () => { + // The tracker cached `codex` at 100ms; Codex exited at 101ms to a bash; the guard runs at 999ms. + const controller = { + ...controllerAnswering(async () => ({ + ...unverifiable('ambiguous_foreground_group'), + capturedAgeMs: 999 + })), + getForegroundProcess: async () => 'codex', + confirmForegroundProcess: async () => 'bash' + } + + await expect( + readLaunchedAgentForeground(controller, POSIX_SSH, 'pty-1', 'codex') + ).resolves.toBe('shell') + }) + + it('keeps a host without a fresh read on its name: an older relay’s npm Codex `node`', async () => { + const { confirmForegroundProcess: _unsupported, ...controller } = { + ...controllerAnswering(async () => unverifiable('ambiguous_foreground_group')), + getForegroundProcess: async () => 'node' + } + + await expect( + readLaunchedAgentForeground(controller, POSIX_SSH, 'pty-1', 'codex') + ).resolves.toBe('agent') + }) + + it.each([ + // Past one TTL the agent may have exited since the capture: the read made now decides. + [990, 'bash', 'shell', 1], + // A 500ms-cached snapshot plus capture on a near-zero-latency link, the agent still there. + [600, 'claude', 'agent', 1], + // Within one TTL the capture alone answers. + [400, 'bash', 'agent', 0], + [5_000, 'claude', 'unknown', 0] + ] as const)( + 'judges live Claude evidence %ims old with a fresh read of %s: %s', + async (capturedAgeMs, freshName, found, freshReads) => { + const confirmForegroundProcess = vi.fn(async () => freshName) + const controller = { + ...controllerAnswering(async () => claudeInGroup(capturedAgeMs)), + getForegroundProcess: async () => 'claude', + confirmForegroundProcess + } + + await expect( + readLaunchedAgentForeground(controller, POSIX_SSH, 'pty-1', 'claude') + ).resolves.toBe(found) + expect(confirmForegroundProcess).toHaveBeenCalledTimes(freshReads) + } + ) + + it('judges the name against the host’s capture age, which a slow response cannot shrink', async () => { + // Captured 400ms before the read was asked for, but 1,500ms old (past the 1s bound) by the time it lands. + const controller = { + ...controllerAnswering(async () => { + vi.advanceTimersByTime(1_100) + return { ...unverifiable('ambiguous_foreground_group'), capturedAgeMs: 1_500 } + }), + getForegroundProcess: async () => 'claude' + } + + await expect( + readLaunchedAgentForeground(controller, POSIX_SSH, 'pty-1', 'claude') + ).resolves.toBe('unknown') + }) + + it.each([ + [ + 'npm Codex', + 'codex', + ['node /usr/local/bin/codex', '/usr/local/lib/node_modules/@openai/codex/bin/codex'] + ], + [ + 'Gemini relaunching itself', + 'gemini', + ['node /usr/local/bin/gemini', 'node --max-old-space-size=8192 /usr/local/bin/gemini'] + ] + ] as const)( + 'proves %s from the relay’s two rows of one agent, even when the relay’s name is node', + async (_label, agent, commands) => { + const row = (pid: number, ppid: number, command: string): ProcessTableRow => ({ + pid, + ppid, + pgid: 40210, + tpgid: 40210, + stat: 'S+', + tty: 'pts/7', + startTime: `start-${pid}`, + command + }) + const evidence = resolveRemoteForegroundEvidence( + { rootPid: 40100, fallbackProcess: 'zsh' }, + { + ptyId: 'pty-1', + ptyIncarnationId: 'inc-1', + authorityGeneration: 'gen-1', + observationEpoch: 1, + capturedAgeMs: 0, + platform: 'linux' + }, + [ + { ...row(40100, 1, '-zsh'), pgid: 40100, stat: 'Ss' }, + row(40210, 40100, commands[0]), + row(40211, 40210, commands[1]) + ] + ) + const controller = { + ...controllerAnswering(async () => evidence), + getForegroundProcess: async () => 'node' + } + + await expect( + readLaunchedAgentForeground(controller, POSIX_SSH, 'pty-1', agent) + ).resolves.toBe('agent') + } + ) + + it('refuses a stale capture without letting the cached display name restore authority', async () => { + // zsh → su → root bash → Claude: the tracker cached `claude` just before Claude exited to the root bash. + const controller = { + ...controllerAnswering(async () => claudeInGroup(1_500)), + getForegroundProcess: async () => 'claude' + } + + await expect( + readLaunchedAgentForeground(controller, POSIX_SSH, 'pty-1', 'claude') + ).resolves.toBe('unknown') + }) }) // Guarded pastes must use terminal-local probes instead of slow whole-machine scans. diff --git a/src/main/runtime/launched-agent-foreground.ts b/src/main/runtime/launched-agent-foreground.ts index 14b31bc8515..cb24ac728f8 100644 --- a/src/main/runtime/launched-agent-foreground.ts +++ b/src/main/runtime/launched-agent-foreground.ts @@ -1,5 +1,6 @@ import { isExpectedAgentProcess, + isNodeInterpreterOfAgent, recognizeAgentProcess } from '../../shared/agent-process-recognition' import { PROCESS_TABLE_SNAPSHOT_MAX_STALENESS_MS } from '../../shared/process-table-snapshot' @@ -30,6 +31,28 @@ export function describeLaunchedAgentHost( return { remote, windows, shellAloneProvable: windows && !remote && pty?.isWsl === false } } +/** Unverifiable reasons that still saw an agent row in the pane's foreground group. Anything else (an unreadable + * table, a missing anchor, another terminal such as `sudo -i`'s pty, withheld argv, a reason this build does not know) + * proves nothing, so a cached name cannot fill the gap. */ +const NAME_COMPLETES_EVIDENCE: ReadonlySet = new Set([ + 'ambiguous_foreground_group', + 'multiplexer_boundary', + 'candidate_start_time_missing' +]) + +// A relay's cached snapshot is up to one TTL old plus its capture time, which a busy host keeps under another TTL. +const EVIDENCE_MAX_AGE_MS = 2 * PROCESS_TABLE_SNAPSHOT_MAX_STALENESS_MS + +function judgeFreshForeground( + processName: string | null, + agent: TuiAgent +): LaunchedAgentForeground { + if (processName && isLaunchedAgent(processName, agent)) { + return 'agent' + } + return processName && isLaunchShell(processName) ? 'shell' : 'unknown' +} + function isLaunchShell(processName: string): boolean { return isShellProcess(processName.replace(/^-/, '')) } @@ -47,7 +70,11 @@ function isLaunchedAgent(processName: string, agent: TuiAgent): boolean { export async function readLaunchedAgentForeground( controller: Pick< RuntimePtyController, - 'getForegroundProcess' | 'confirmShellForeground' | 'inspectProcess' | 'listProcesses' + | 'getForegroundProcess' + | 'confirmForegroundProcess' + | 'confirmShellForeground' + | 'inspectProcess' + | 'listProcesses' > | null, host: { remote: boolean; windows: boolean }, ptyId: string, @@ -73,20 +100,53 @@ export async function readLaunchedAgentForeground( if ( evidence?.verdict === 'live' && evidence.fence.platform === 'posix' && - Date.now() - evidence.capturedAgeMs >= askedAt - PROCESS_TABLE_SNAPSHOT_MAX_STALENESS_MS && evidence.processName && isLaunchedAgent(evidence.processName, agent) ) { - return 'agent' + // Begun no more than one TTL before the read was asked (so any age within one TTL), however long ps took. + if ( + Date.now() - evidence.capturedAgeMs >= + askedAt - PROCESS_TABLE_SNAPSHOT_MAX_STALENESS_MS + ) { + return 'agent' + } + // Up to two TTLs the agent may have exited since: a foreground read now decides. + if (evidence.capturedAgeMs <= EVIDENCE_MAX_AGE_MS && controller.confirmForegroundProcess) { + return judgeFreshForeground(await controller.confirmForegroundProcess(ptyId), agent) + } } - const foreground = await controller.getForegroundProcess(ptyId) + // Only evidence that saw an agent row it could not fence alone lets a name complete it. + const completable = + evidence?.verdict === 'unverifiable' && + evidence.capturedAgeMs <= EVIDENCE_MAX_AGE_MS && + NAME_COMPLETES_EVIDENCE.has(evidence.reason) + // The tracker's name can be its 1s display cache; for completion read it now where the host can. + const foreground = + completable && controller.confirmForegroundProcess + ? await controller.confirmForegroundProcess(ptyId) + : await controller.getForegroundProcess(ptyId) if (!foreground) { return 'unknown' } - return isLaunchShell(foreground) && + if ( + isLaunchShell(foreground) && !isExpectedAgentProcess(foreground, TUI_AGENT_CONFIG[agent].expectedProcess) - ? 'shell' - : 'agent' + ) { + return 'shell' + } + // Legacy: relays that send no evidence (older versions) keep the possibly cached name. + if (!evidence) { + return 'agent' + } + if (!completable) { + return 'unknown' + } + // Relays before same-agent dedupe call npm Codex's two rows ambiguous and name its `node` wrapper. + return isLaunchedAgent(foreground, agent) || + (evidence.reason === 'ambiguous_foreground_group' && + isNodeInterpreterOfAgent(foreground, agent)) + ? 'agent' + : 'unknown' } catch { return 'unknown' } diff --git a/src/main/runtime/terminal-foreground-group.test.ts b/src/main/runtime/terminal-foreground-group.test.ts index df1c3348378..8176050fbdf 100644 --- a/src/main/runtime/terminal-foreground-group.test.ts +++ b/src/main/runtime/terminal-foreground-group.test.ts @@ -52,6 +52,12 @@ describe('who holds a pane’s terminal, by its foreground process group', () => pane(200, [{ pid: 200, ppid: 101, pgid: 200, stat: 'S+', command: 'bash ./start-agent.sh' }]), 'shell' ], + // A pane looping short commands: our own child, already exiting, prints `(comm)` in ps too. + [ + 'the pane’s own exiting command', + pane(200, [{ pid: 200, ppid: 101, pgid: 200, stat: 'SE+', command: '(git)' }]), + 'agent' + ], [ 'an agent stopped in the background', pane(101, [{ pid: 200, ppid: 101, pgid: 200, stat: 'T', command: '/opt/bin/claude' }]), @@ -64,7 +70,12 @@ describe('who holds a pane’s terminal, by its foreground process group', () => it.each([ ['the root is not on the terminal', pane(101, []).slice(1)], ['the terminal has no foreground group', pane(0, [])], - ['no process is in the foreground group', pane(300, [])] + ['no process is in the foreground group', pane(300, [])], + // `su` in the pane: a root shell whose argv neither the in-process reader nor ps recovered. + [ + 'the foreground argv is unreadable', + pane(200, [{ pid: 200, ppid: 101, pgid: 200, stat: 'S+', command: '(bash)' }]) + ] ])('proves nothing when %s', (_label, rows) => { expect(judgeTerminalForeground(rows, 100, 'claude')).toBe('unknown') }) diff --git a/src/main/runtime/terminal-foreground-group.ts b/src/main/runtime/terminal-foreground-group.ts index decdb5aa3c7..66045b4f736 100644 --- a/src/main/runtime/terminal-foreground-group.ts +++ b/src/main/runtime/terminal-foreground-group.ts @@ -3,7 +3,11 @@ import { recognizeAgentProcessFromCommandLine } from '../../shared/agent-process-recognition' import { runProcess } from '@orca/process-host' import { runPsInProcess } from '../../shared/libproc-process-table' -import { parseShellForegroundRows, type ProcessTableRow } from '../../shared/process-table-snapshot' +import { + parseShellForegroundRows, + unreadableArgvName, + type ProcessTableRow +} from '../../shared/process-table-snapshot' import { psLocaleEnvironment } from '../../shared/ps-locale-environment' import { isShellProcess } from '../../shared/shell-process-detection' import type { TuiAgent } from '../../shared/tui-agent' @@ -90,7 +94,12 @@ export function judgeTerminalForeground( return 'unknown' } const front = rows.filter((row) => row.pgid === foregroundGroup) - if (front.length === 0) { + // An unrecovered root shell (`su`) prints `(bash)`; typing into it could run the text. Our own exiting `(git)` is no risk. + const unreadableShell = (row: ProcessTableRow): boolean => { + const name = unreadableArgvName(row.command) + return name !== null && isShellProcess(name.replace(/^-/, '')) + } + if (front.length === 0 || front.some(unreadableShell)) { return 'unknown' } return front.some( diff --git a/src/shared/agent-process-recognition.ts b/src/shared/agent-process-recognition.ts index 0e427070cae..d28d0d27bef 100644 --- a/src/shared/agent-process-recognition.ts +++ b/src/shared/agent-process-recognition.ts @@ -255,6 +255,18 @@ export function isAgentForegroundWrapperProcess(processName: string | null | und ) } +/** True when `processName` is the Node interpreter of an npm-installed agent (npm Codex's `node codex.js` wrapper). */ +export function isNodeInterpreterOfAgent( + processName: string | null | undefined, + agent: TerminalAgent +): boolean { + return ( + normalizeProcessName(processName) === 'node' && + (agent in NODE_PACKAGE_SCRIPT_ENTRYPOINTS || + EXACT_NODE_ENTRYPOINT_IDENTITIES.some((identity) => identity.agent === agent)) + ) +} + export function isRecognizedAgentType(agentType: AgentType | null | undefined): boolean { if (typeof agentType !== 'string') { return false diff --git a/src/shared/cheap-process-table-snapshot-reader.ts b/src/shared/cheap-process-table-snapshot-reader.ts index 11b91098e0b..3ffa8718031 100644 --- a/src/shared/cheap-process-table-snapshot-reader.ts +++ b/src/shared/cheap-process-table-snapshot-reader.ts @@ -20,7 +20,9 @@ const cheapProcessTableReader = createProcessTableSnapshotReader ({ runProcess: vi.fn() })) +vi.mock('@orca/process-host', () => ({ runProcess })) + import { captureDarwinProcessTable } from './darwin-process-table-capture' import { DARWIN_TERMINAL_NAMING_STALL_MS, @@ -10,6 +14,7 @@ import { monotonicNow } from './deadline-settle-notifier' import { setLibprocAddonForTests, type LibprocProcessRow } from './libproc-addon' import { resetNativeReadsForTests } from './libproc-native-reads' import { runPsInProcess } from './libproc-process-table' +import { resetPsFallbackForTests } from './libproc-ps-fallback' import { PS_ARGS } from './process-table-snapshot' const ROW: LibprocProcessRow = { @@ -40,6 +45,8 @@ beforeEach(() => { }) afterEach(() => { setLibprocAddonForTests(undefined) + resetPsFallbackForTests() + runProcess.mockReset() vi.useRealTimers() }) @@ -59,15 +66,19 @@ describe('terminal naming single flight', () => { holdNamingSlot() setLibprocAddonForTests({ readProcessTable: async () => [ROW], - processCwd: () => null, - processPath: () => null + readProcessCwd: async () => null, + readProcessPath: async () => null }) const args = ['-p', '4300', '-o', 'pid=,tty='] + const ps = { code: 0, signal: null, stdout: ' 4300 ttys004\n', stderr: '', timedOut: false } + runProcess.mockResolvedValue(ps) + const waiting = runPsInProcess(args, { timeoutMs: 5_000 }) await vi.advanceTimersByTimeAsync(DARWIN_TERMINAL_NAMING_STALL_MS) - expect(await waiting).toBeNull() - expect(await runPsInProcess(args, { timeoutMs: 5_000 })).toBeNull() + expect(await waiting).toEqual(ps) + expect(await runPsInProcess(args, { timeoutMs: 5_000 })).toEqual(ps) + expect(runProcess).toHaveBeenCalledWith(expect.objectContaining({ program: '/bin/ps', args })) }) it('keeps one reaction on a stuck pass however many waiters time out', async () => { diff --git a/src/shared/libproc-addon-worker-teardown.test.ts b/src/shared/libproc-addon-worker-teardown.test.ts new file mode 100644 index 00000000000..9b78fd7ef01 --- /dev/null +++ b/src/shared/libproc-addon-worker-teardown.test.ts @@ -0,0 +1,45 @@ +import { spawnSync } from 'node:child_process' +import { createRequire } from 'node:module' +import { dirname } from 'node:path' +import { describe, expect, it } from 'vitest' + +// A worker_thread calling into the libproc reads from a timer while it is terminated used to abort the whole +// process (uncaught Napi::Error from a call-in or completion during env teardown), so run it in a child. +const SCRIPT = String.raw` +const { Worker } = require('node:worker_threads') +const ptyDir = process.argv[1] +function once(run) { + return new Promise((resolve, reject) => { + const worker = new Worker( + 'const m = require(' + JSON.stringify(ptyDir + '/lib/utils') + ').loadNativeModule("pty").module;' + + 'setInterval(() => { m.readProcessTable({ command: false }); m.readProcessCwd(process.pid); m.readProcessPath(process.pid) }, 0);' + + 'require("node:worker_threads").parentPort.postMessage(0)', + { eval: true } + ) + worker.on('message', () => setTimeout(() => worker.terminate().then(resolve, reject), run % 7)) + worker.on('error', reject) + }) +} +;(async () => { + for (let run = 0; run < 100; run += 1) await once(run) + console.log('survived') +})() +` + +describe.runIf(process.platform === 'darwin')('libproc addon worker teardown', () => { + // 100 worker_thread starts and terminates in a child node take a few seconds, more under suite load. + it( + 'survives terminating a worker thread that keeps calling the native reads', + { timeout: 60_000 }, + () => { + const ptyDir = dirname(createRequire(__filename).resolve('node-pty/package.json')) + const child = spawnSync(process.execPath, ['-e', SCRIPT, ptyDir], { + encoding: 'utf8', + timeout: 50_000 + }) + expect(child.stderr).not.toContain('terminating due to uncaught exception') + expect(child.stdout.trim()).toBe('survived') + expect(child.status).toBe(0) + } + ) +}) diff --git a/src/shared/libproc-addon.ts b/src/shared/libproc-addon.ts index 35d26654d03..7aaeb5ad419 100644 --- a/src/shared/libproc-addon.ts +++ b/src/shared/libproc-addon.ts @@ -1,4 +1,5 @@ import { createRequire } from 'node:module' +import { DEADLINE_PASSED, beforeDeadline, monotonicNow } from './deadline-settle-notifier' /** One `kinfo_proc` from node-pty's patched macOS addon (config/patches/node-pty@1.1.0.patch). */ export type LibprocProcessRow = { @@ -13,6 +14,8 @@ export type LibprocProcessRow = { flags: number /** `p_nice`; absent from older builds. */ nice?: number + /** Effective uid (`cr_uid`), which decides whether KERN_PROCARGS2 may read argv. */ + uid?: number sessionLeader: boolean startSec: number /** `p_starttime.tv_usec`, telling apart processes a reused pid started within one second. */ @@ -20,7 +23,9 @@ export type LibprocProcessRow = { comm: string /** ps's command text (getproclline + strvis in the query's LC_CTYPE). */ command?: string - /** argv was unreadable (another user's process): `command` holds the executable path, where setuid ps prints argv. */ + /** Set by argv recovery, not the addon: monotonic ms no later than setuid ps read `command`, so ages never understate. */ + commandReadAtMs?: number + /** argv was unreadable (another user's, exiting or mid-exec): `command` holds `(comm)`, where setuid ps may print argv. */ commandIncomplete?: boolean } @@ -35,24 +40,26 @@ export type LibprocQuery = { export type LibprocAddon = { /** Reads on a libuv worker thread; argv for a whole table costs ~20ms of syscalls. */ readProcessTable: (query: LibprocQuery) => Promise - processCwd: (pid: number) => string | null - processPath: (pid: number) => string | null + /** PROC_PIDVNODEPATHINFO and proc_pidpath can block on a dead network mount, so both run on a worker thread. + * Absent from a #35987 pty.node, which still reads the table: cwd and path then come from lsof. */ + readProcessCwd?: (pid: number) => Promise + readProcessPath?: (pid: number) => Promise } const requireFromHere = createRequire(__filename) let cachedAddon: LibprocAddon | null | undefined let addonOverride: LibprocAddon | null | undefined +function hasAsyncPathReads(addon: LibprocAddon): boolean { + return typeof addon.readProcessCwd === 'function' && typeof addon.readProcessPath === 'function' +} + function isLibprocAddon(native: unknown): native is LibprocAddon { return ( typeof native === 'object' && native !== null && 'readProcessTable' in native && - typeof native.readProcessTable === 'function' && - 'processCwd' in native && - typeof native.processCwd === 'function' && - 'processPath' in native && - typeof native.processPath === 'function' + typeof native.readProcessTable === 'function' ) } @@ -77,8 +84,12 @@ export function loadLibprocAddon(): LibprocAddon | null { typeof utils.loadNativeModule === 'function' ? utils.loadNativeModule('pty')?.module : null - // An older pty.node exports positional processTable() instead; it must not match. + // An older pty.node exports positional processTable() instead; it must not match. Its synchronous processCwd and + // processPath are never called: they can block the event loop on a dead mount. cachedAddon = isLibprocAddon(native) ? native : null + if (cachedAddon && !hasAsyncPathReads(cachedAddon)) { + console.warn('[process-table] node-pty lacks the async cwd/path reader; using lsof for those') + } } catch { cachedAddon = null } @@ -93,27 +104,37 @@ export function setLibprocAddonForTests(addon: LibprocAddon | null | undefined): addonOverride = addon } -/** PROC_PIDVNODEPATHINFO cwd, or undefined when only lsof can answer. */ -export function readLibprocProcessCwd(pid: number): string | null | undefined { - const addon = loadLibprocAddon() - if (!addon) { - return undefined - } - try { - return addon.processCwd(pid) - } catch { - return undefined - } -} +// A lookup stuck on a dead mount pins its worker thread in the kernel; cap them so the rest of the libuv pool stays free. +const MAX_OUTSTANDING_PATH_READS = 2 +// Path lookups take microseconds; past this the mount is wedged and lsof -b answers instead. lsof can block in the same +// PROC_PIDVNODEPATHINFO, but in a child process (one per pid), not on this process's thread pool. +const PATH_READ_TIMEOUT_MS = 250 +let outstandingPathReads = 0 -/** proc_pidpath executable, or undefined when only lsof can answer. */ -export function readLibprocProcessPath(pid: number): string | null | undefined { +/** cwd or executable path read in process, or undefined when only lsof can answer (no addon, a slow or wedged lookup). */ +export async function readLibprocProcessFilePath( + which: 'cwd' | 'executable', + pid: number +): Promise { const addon = loadLibprocAddon() - if (!addon) { + const read = which === 'cwd' ? addon?.readProcessCwd : addon?.readProcessPath + if (!read || outstandingPathReads >= MAX_OUTSTANDING_PATH_READS) { return undefined } + let pending: Promise try { - return addon.processPath(pid) + pending = read(pid) + } catch { + return undefined + } + outstandingPathReads += 1 + const release = (): void => { + outstandingPathReads -= 1 + } + pending.then(release, release) + try { + const path = await beforeDeadline(pending, monotonicNow() + PATH_READ_TIMEOUT_MS) + return path === DEADLINE_PASSED ? undefined : path } catch { return undefined } diff --git a/src/shared/libproc-argv-completion.ts b/src/shared/libproc-argv-completion.ts index 57f7df54af6..bdf066d3986 100644 --- a/src/shared/libproc-argv-completion.ts +++ b/src/shared/libproc-argv-completion.ts @@ -1,13 +1,18 @@ -import { runProcess } from '@orca/process-host' import type { LibprocProcessRow } from './libproc-addon' -import { monotonicNow } from './deadline-settle-notifier' -import { NODEV } from './libproc-ps-format' +import { DEADLINE_PASSED, beforeDeadline, monotonicNow } from './deadline-settle-notifier' +import { NODEV, P_WEXIT, SIDL, SZOMB } from './libproc-ps-format' +import { runArgvRecoveryPs } from './libproc-ps-fallback' -const SZOMB_STATUS = 5 const MAX_ANCESTOR_DEPTH = 64 -/** A pane descendant: the unreadable chain above it starts on the terminal of one of the user's own (argv-readable) - * processes, e.g. `sudo` under the pane shell, and the root shell and agent `sudo -i` puts on its own pty (use_pty). +/** KERN_PROCARGS2 refuses only another euid's argv; our own fails just mid-exec or exiting, where setuid ps fails too. */ +function isOwnProcess(row: LibprocProcessRow): boolean { + const ownUid = process.geteuid?.() + return row.uid !== undefined && ownUid !== undefined ? row.uid === ownUid : !row.commandIncomplete +} + +/** A pane descendant: the foreign chain above it starts on the terminal of one of the user's own processes, e.g. + * `sudo` under the pane shell, and the root shell and agent `sudo -i` puts on its own pty (use_pty). * Root daemons descend only from launchd, and a pane's root `login` opens its own pty under the terminal host, so * neither costs a ps. */ function descendsFromUsersTerminal( @@ -20,7 +25,7 @@ function descendsFromUsersTerminal( if (!parent) { return false } - if (!parent.commandIncomplete) { + if (isOwnProcess(parent)) { return parent.tdev !== NODEV && parent.tdev === top.tdev } top = parent @@ -28,90 +33,174 @@ function descendsFromUsersTerminal( return false } -type AuthoritativeCommand = { command: string; cachedAtMs: number } +/** Which rows the caller asked ps for: `-p` pids, one `-t` terminal, or the whole table. */ +export type CommandSelection = 'pids' | 'tty' | 'all' -// exec can keep pid, start time, comm and executable (node → node with another script) and no cheap field shows it, -// so a recovered argv is trusted for 30s: a sudo pane costs one targeted ps per process per TTL, other users none. -const authoritativeCommands = new Map() -const AUTHORITATIVE_COMMAND_LIMIT = 512 -const AUTHORITATIVE_COMMAND_TTL_MS = 30_000 - -/** exec keeps pid and start time but changes comm and the executable path; microseconds split same-second pid reuse. */ -function commandKey(ctype: string, row: LibprocProcessRow): string { - return [ctype, row.pid, row.startSec, row.startUsec ?? '', row.comm, row.command ?? ''].join('\0') +function needsRecovery( + row: LibprocProcessRow, + selection: CommandSelection, + byPid: ReadonlyMap +): boolean { + return ( + row.commandIncomplete === true && + row.status !== SZOMB && + row.status !== SIDL && + (row.flags & P_WEXIT) === 0 && + !isOwnProcess(row) && + (selection === 'pids' || + // After `exec su` no process of ours is left on the terminal; its foreground group is what callers judge. + (selection === 'tty' && row.pgid === row.tpgid) || + descendsFromUsersTerminal(row, byPid)) + ) } -function rememberCommand(key: string, command: string, now: number): void { - authoritativeCommands.delete(key) - for (const oldest of authoritativeCommands.keys()) { - if (authoritativeCommands.size < AUTHORITATIVE_COMMAND_LIMIT) { +/** A positive answer ages from when its ps started; a failure from when it ended, however long ps took. */ +type Recovered = Answer & { settledAtMs: number } + +// exec can keep pid, start time and comm (node → node with another script) and no cheap field shows it, so recovered +// argv is reused for at most 1.5s (plus the 500ms snapshot cache, under the 2s evidence freshness gate), and every row +// carries when ps read it so evidence reports its oldest input's age. +const recovered = new Map() +/** A ps run's answer for one process: its argv (undefined when ps omitted it) and when that ps started. */ +type Answer = { command: string | undefined; readAtMs: number } +// null when ps timed out or the budget withheld it: that proves nothing. +const recovering = new Map }>() +const RECOVERED_LIMIT = 512 +const RECOVERED_TTL_MS = 1_500 +// A row ps answered without (exited, or unreadable to it too) waits this long before costing another ps. +const UNRECOVERED_TTL_MS = 5_000 + +/** exec keeps pid and start time; comm usually changes, and microseconds split same-second pid reuse. */ +function commandKey(ctype: string, row: LibprocProcessRow): string { + return [ctype, row.pid, row.startSec, row.startUsec ?? '', row.comm].join('\0') +} + +function remember(key: string, entry: Recovered): void { + // A recovery begun past an older one's window can finish first; the older answer must not replace it. + if ((recovered.get(key)?.readAtMs ?? Number.NEGATIVE_INFINITY) > entry.readAtMs) { + return + } + recovered.delete(key) + for (const oldest of recovered.keys()) { + if (recovered.size < RECOVERED_LIMIT) { break } - authoritativeCommands.delete(oldest) + recovered.delete(oldest) } - authoritativeCommands.set(key, { command, cachedAtMs: now }) + recovered.set(key, entry) } -/** Fill argv only setuid ps can read, for the rows this caller selected or that descend from a user's terminal. */ +async function readCommands( + pids: readonly number[], + env: NodeJS.ProcessEnv, + deadline: number, + background: boolean +): Promise<{ commands: ReadonlyMap; readAtMs: number } | null> { + const requestedAtMs = monotonicNow() + try { + const result = await runArgvRecoveryPs( + ['-ww', '-o', 'pid=,command=', '-p', pids.join(',')], + { env, maxOutputBytes: 1024 * 1024 }, + deadline, + background + ) + if (result.timedOut || result.outputTruncated || (result.code !== 0 && result.code !== 1)) { + return null + } + const commands = new Map() + for (const line of result.stdout.split('\n')) { + const match = /^\s*(\d+) (.*)$/.exec(line) + if (match) { + commands.set(Number(match[1]), match[2]) + } + } + // Reuse ages from when ps started, not from a wait for its fork token. + return { commands, readAtMs: result.spawnedAtMs ?? requestedAtMs } + } catch { + return null + } +} + +/** Fill argv only setuid ps can read, for the rows this caller selected or that descend from a user's terminal. + * Concurrent callers share one ps per process; `background` polls yield the recovery budget to interactive reads. */ export async function completeCommands( rows: LibprocProcessRow[], - selectedByPid: boolean, + selection: CommandSelection, ctype: string, env: NodeJS.ProcessEnv, - deadline: number + deadline: number, + background: boolean ): Promise { const byPid = new Map(rows.map((row) => [row.pid, row])) const now = monotonicNow() + const pending: { row: LibprocProcessRow; answer: Promise }[] = [] const missing: { row: LibprocProcessRow; key: string }[] = [] for (const row of rows) { - if ( - !row.commandIncomplete || - row.status === SZOMB_STATUS || - !(selectedByPid || descendsFromUsersTerminal(row, byPid)) - ) { + if (!needsRecovery(row, selection, byPid)) { continue } const key = commandKey(ctype, row) - const cached = authoritativeCommands.get(key) - if (cached && now - cached.cachedAtMs < AUTHORITATIVE_COMMAND_TTL_MS) { - row.command = cached.command + const known = recovered.get(key) + // A failure only withholds argv (the row reads unverifiable), never serves stale argv. + if (known && known.command === undefined && now - known.settledAtMs < UNRECOVERED_TTL_MS) { + continue + } + if (known?.command !== undefined && now - known.readAtMs < RECOVERED_TTL_MS) { + row.command = known.command + row.commandReadAtMs = known.readAtMs + continue + } + // A run begun too long ago reads argv older than this caller may reuse. + const inFlight = recovering.get(key) + if (inFlight && now - inFlight.startedAtMs < RECOVERED_TTL_MS) { + pending.push({ row, answer: inFlight.answer }) } else { missing.push({ row, key }) } } - const timeoutMs = Math.floor(deadline - monotonicNow()) - if (missing.length === 0 || timeoutMs <= 0) { + if (deadline <= monotonicNow()) { return } - let stdout: string - try { - const result = await runProcess({ - program: '/bin/ps', - args: ['-ww', '-o', 'pid=,command=', '-p', missing.map(({ row }) => row.pid).join(',')], + if (missing.length > 0) { + const startedAtMs = monotonicNow() + const read = readCommands( + missing.map(({ row }) => row.pid), env, - timeoutMs, - maxOutputBytes: 1024 * 1024 - }) - // Unreadable rows keep the executable path and retry next read. - if (result.timedOut || result.outputTruncated || (result.code !== 0 && result.code !== 1)) { - return + deadline, + background + ) + for (const { row, key } of missing) { + const recovery = { + startedAtMs, + answer: read.then((result) => + result ? { command: result.commands.get(row.pid), readAtMs: result.readAtMs } : null + ) + } + recovering.set(key, recovery) + void recovery.answer.then((answer) => { + // Only an answer is remembered: a ps that timed out, or a run the budget withheld, proved nothing. + if (answer) { + remember(key, { ...answer, settledAtMs: monotonicNow() }) + } + if (recovering.get(key) === recovery) { + recovering.delete(key) + } + }) + pending.push({ row, answer: recovery.answer }) } - stdout = result.stdout - } catch { + } + if (pending.length === 0) { return } - const commands = new Map() - for (const line of stdout.split('\n')) { - const match = /^\s*(\d+) (.*)$/.exec(line) - if (match) { - commands.set(Number(match[1]), match[2]) - } + const answers = await beforeDeadline(Promise.all(pending.map(({ answer }) => answer)), deadline) + if (answers === DEADLINE_PASSED) { + return } - for (const { row, key } of missing) { - const command = commands.get(row.pid) - if (command !== undefined) { - row.command = command - rememberCommand(key, command, now) + pending.forEach(({ row }, index) => { + const answer = answers[index] + if (answer?.command !== undefined) { + row.command = answer.command + row.commandReadAtMs = answer.readAtMs } - } + }) } diff --git a/src/shared/libproc-native-reads.ts b/src/shared/libproc-native-reads.ts index cbaffa8fec4..a4d66f429fd 100644 --- a/src/shared/libproc-native-reads.ts +++ b/src/shared/libproc-native-reads.ts @@ -26,6 +26,22 @@ const STALLED_POOL_COOL_DOWN_MS = 30_000 let unhealthyUntil = Number.NEGATIVE_INFINITY // Set when the cap outlived a caller's deadline; cleared once any read settles. let poolStalled = false +// A rejecting addon (failed sysctl) backs off 1s, 2s, 4s… up to the stalled cool-down instead of retrying every poll. +const REJECTION_BACKOFF_MS = 1_000 +let consecutiveRejections = 0 + +function noteSettled(ok: boolean): void { + if (ok) { + consecutiveRejections = 0 + return + } + const backoff = Math.min( + STALLED_POOL_COOL_DOWN_MS, + REJECTION_BACKOFF_MS * 2 ** Math.min(consecutiveRejections, 15) + ) + consecutiveRejections += 1 + unhealthyUntil = Math.max(unhealthyUntil, monotonicNow() + backoff) +} export function isNativeReaderHealthy(): boolean { return monotonicNow() >= unhealthyUntil @@ -69,7 +85,16 @@ function submitRead(addon: LibprocAddon, query: LibprocQuery, key: string): Nati } slotWaiters.clear() } - pending.then(release, release) + pending.then( + () => { + noteSettled(true) + release() + }, + () => { + noteSettled(false) + release() + } + ) return read } @@ -90,7 +115,8 @@ function waitForSlot(deadline: number): Promise { }) } -/** Rows from a read submitted after `registeredSequence` (null: -t named no terminal); 'stalled' sends the caller to ps. */ +/** Rows from a read submitted after `registeredSequence` (null: -t named no terminal); 'stalled' sends the caller to ps. + * Rejects when the addon does, which also backs the reader off. */ export async function readFreshNativeTable( addon: LibprocAddon, query: LibprocQuery, @@ -142,5 +168,6 @@ export function resetNativeReadsForTests(): void { slotWaiters.clear() outstandingReads = 0 poolStalled = false + consecutiveRejections = 0 unhealthyUntil = Number.NEGATIVE_INFINITY } diff --git a/src/shared/libproc-process-table.test.ts b/src/shared/libproc-process-table.test.ts index 7553f955e35..c0b3f3d0354 100644 --- a/src/shared/libproc-process-table.test.ts +++ b/src/shared/libproc-process-table.test.ts @@ -1,4 +1,4 @@ -import { afterEach, describe, expect, it, vi } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' const { runProcess } = vi.hoisted(() => ({ runProcess: vi.fn() })) vi.mock('@orca/process-host', () => ({ runProcess })) @@ -15,6 +15,8 @@ import { } from './libproc-addon' import { nativeReadWaitersForTests, resetNativeReadsForTests } from './libproc-native-reads' import { runPsInProcess } from './libproc-process-table' +import { resetPsFallbackForTests } from './libproc-ps-fallback' +import { createProcessTableSnapshotReader } from './process-table-snapshot-reader' import { parseCheapProcessTableRows, parseStrictProcessTableRows } from './process-table-snapshot' const TTYS004 = (16 << 24) | 4 @@ -107,17 +109,37 @@ function tableFor( function installAddon(overrides: Partial = {}): LibprocAddon { const addon: LibprocAddon = { readProcessTable: vi.fn(async (query: LibprocQuery) => tableFor(query)), - processCwd: vi.fn(() => null), - processPath: vi.fn(() => null), + readProcessCwd: vi.fn(async () => null), + readProcessPath: vi.fn(async () => null), ...overrides } setLibprocAddonForTests(addon) return addon } +// Orca's euid is pinned, so neither a root runner nor Windows (no geteuid) changes whose argv is "own". +const OWN_UID = 501 +const OTHER_UID = 502 +const realGeteuid = process.geteuid + +beforeEach(() => { + process.geteuid = () => OWN_UID +}) + +const PS_RESULT = { + code: 0, + signal: null, + stdout: ' 1\n', + stderr: '', + timedOut: false, + outputTruncated: false +} + afterEach(() => { + process.geteuid = realGeteuid setLibprocAddonForTests(undefined) resetNativeReadsForTests() + resetPsFallbackForTests() runProcess.mockReset() vi.useRealTimers() }) @@ -182,7 +204,7 @@ describe('runPsInProcess', () => { }) }) - it('keeps the real ps for invocations it cannot reproduce or a failing addon', async () => { + it('keeps the real ps for invocations it cannot reproduce, and answers a failing addon with /bin/ps', async () => { installAddon() expect(await runPsInProcess(['-axo', 'pid=PROCESS_ID'], BUDGET)).toBeNull() expect(await runPsInProcess(['-axo', 'pid=,uid='], BUDGET)).toBeNull() @@ -200,7 +222,11 @@ describe('runPsInProcess', () => { throw new Error('sysctl failed') } }) - expect(await runPsInProcess(['-axo', 'pid='], BUDGET)).toBeNull() + runProcess.mockResolvedValue(PS_RESULT) + expect(await runPsInProcess(['-axo', 'pid='], BUDGET)).toEqual(PS_RESULT) + expect(runProcess).toHaveBeenCalledWith( + expect.objectContaining({ program: '/bin/ps', args: ['-axo', 'pid='] }) + ) setLibprocAddonForTests(null) expect(await runPsInProcess(['-axo', 'pid='], BUDGET)).toBeNull() @@ -279,9 +305,10 @@ describe('runPsInProcess', () => { expect(readProcessTable).toHaveBeenCalledTimes(2) }) - it('queues past the read cap without retaining timed-out waiters, then hands a stalled pool to ps', async () => { + it('queues past the read cap without retaining timed-out waiters, then hands a stalled pool to budgeted ps', async () => { vi.useFakeTimers() - installAddon({ readProcessTable: vi.fn(() => new Promise(() => {})) }) + const addon = installAddon({ readProcessTable: vi.fn(() => new Promise(() => {})) }) + runProcess.mockResolvedValue(PS_RESULT) const read = (pid: number, timeoutMs = 50) => runPsInProcess(['-p', String(pid), '-o', 'pid='], { timeoutMs }) @@ -291,11 +318,18 @@ describe('runPsInProcess', () => { expect(nativeReadWaitersForTests()).toBe(300) await vi.advanceTimersByTimeAsync(50) expect(nativeReadWaitersForTests()).toBe(0) - expect(await Promise.all(overflow)).toEqual(Array(300).fill(null)) + // The stalled burst costs the fork budget, not one ps per caller; the rest time out like a killed ps. + await vi.advanceTimersByTimeAsync(50) + const results = await Promise.all(overflow) + expect(results.filter((result) => result?.timedOut === false)).toHaveLength(4) + expect(results.filter((result) => result?.timedOut === true)).toHaveLength(296) + expect(runProcess).toHaveBeenCalledTimes(4) - // The breaker stays open for its cool-down: callers run ps without queueing. - expect(await read(7000)).toBeNull() + // The breaker stays open for its cool-down: callers go to ps without queueing on the pool. + await vi.advanceTimersByTimeAsync(1_000) + expect(await read(7000)).toEqual(PS_RESULT) expect(nativeReadWaitersForTests()).toBe(0) + expect(addon.readProcessTable).toHaveBeenCalledTimes(8) await vi.advanceTimersByTimeAsync(10_000) expect(await Promise.all(capped)).toEqual( Array(8).fill(expect.objectContaining({ timedOut: true })) @@ -308,7 +342,7 @@ describe('runPsInProcess', () => { settled = result }) await vi.advanceTimersByTimeAsync(0) - expect(settled).toBeNull() + expect(settled).toEqual(PS_RESULT) }) it('recovers argv for a sudo use_pty tree on its own pty, but not for root daemons or GUI helpers', async () => { @@ -383,4 +417,295 @@ describe('runPsInProcess', () => { expect(results.map((result) => result?.stdout)).toEqual(Array(3).fill('4242 ttys004\n')) expect(runProcess).not.toHaveBeenCalled() }) + + it('leaves argv of the user’s own mid-exec processes and of exiting ones to (comm), asking ps only for other users', async () => { + const foreign = { ppid: 4242, tdev: TTYS004, uid: OTHER_UID, flags: 0x2 } + const table = [ + ...TABLE, + // A pane looping short commands: our own child between fork and exec reads like another user's. + row({ pid: 4800, ppid: 4242, tdev: TTYS004, uid: OWN_UID, comm: 'git', command: '(git)' }), + row({ ...foreign, pid: 4801, flags: 0x2 | 0x2000, comm: 'sudo', command: '(sudo)' }), + row({ ...foreign, pid: 4802, status: 1, comm: 'sudo', command: '(sudo)' }), + row({ ...foreign, pid: 4803, comm: 'node', command: '(node)' }) + ] + installAddon({ readProcessTable: vi.fn(async (query: LibprocQuery) => tableFor(query, table)) }) + runProcess.mockResolvedValue({ + code: 0, + signal: null, + stdout: ' 4803 node /opt/claude/cli.js\n', + stderr: '', + timedOut: false + }) + + const result = await runPsInProcess(['-axo', 'pid=,command='], BUDGET) + expect(result?.stdout).toContain('4800 (git)\n') + expect(result?.stdout).toContain('4803 node /opt/claude/cli.js\n') + expect(runProcess).toHaveBeenCalledTimes(1) + expect(runProcess.mock.calls[0]?.[0].args.at(-1)).toBe('4803') + }) + + it('shares one argv ps among concurrent readers and waits out a failed recovery before asking again', async () => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'performance'] }) + const sudo = row({ + pid: 4900, + ppid: 4242, + tdev: TTYS004, + uid: OTHER_UID, + comm: 'node', + command: '(node)' + }) + installAddon({ + readProcessTable: vi.fn(async (query: LibprocQuery) => tableFor(query, [...TABLE, sudo])) + }) + // ps omits a process that exited, or one it cannot read either. + runProcess.mockResolvedValue({ code: 1, signal: null, stdout: '', stderr: '', timedOut: false }) + const read = (): Promise => + runPsInProcess(['-o', 'pid=,command=', '-t', 'ttys004'], BUDGET) + + const concurrent = await Promise.all(Array.from({ length: 24 }, read)) + expect(concurrent).toEqual(Array(24).fill(expect.objectContaining({ timedOut: false }))) + expect(runProcess).toHaveBeenCalledTimes(1) + + await read() + expect(runProcess).toHaveBeenCalledTimes(1) + await vi.advanceTimersByTimeAsync(5_000) + await read() + expect(runProcess).toHaveBeenCalledTimes(2) + }) + + it('reuses recovered argv for at most 1.5s, so a same-comm re-exec is seen', async () => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'performance'] }) + const sudo = row({ + pid: 5100, + ppid: 4242, + tdev: TTYS004, + uid: OTHER_UID, + comm: 'node', + command: '(node)' + }) + installAddon({ + readProcessTable: vi.fn(async (query: LibprocQuery) => tableFor(query, [...TABLE, sudo])) + }) + // Same pid, start time and comm: node re-exec'd from Claude into an unrelated server. + let argv = 'node /opt/claude/cli.js' + runProcess.mockImplementation(async () => ({ + code: 0, + signal: null, + stdout: ` 5100 ${argv}\n`, + stderr: '', + timedOut: false + })) + const args = ['-p', '5100', '-o', 'command='] + + expect((await runPsInProcess(args, BUDGET))?.stdout).toBe(`node /opt/claude/cli.js\n`) + argv = 'node /srv/server.js' + expect((await runPsInProcess(args, BUDGET))?.stdout).toBe(`node /opt/claude/cli.js\n`) + await vi.advanceTimersByTimeAsync(1_500) + expect((await runPsInProcess(args, BUDGET))?.stdout).toBe('node /srv/server.js\n') + expect(runProcess).toHaveBeenCalledTimes(2) + }) + + it('waits out a failed recovery from when ps gave up, however long it ran', async () => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'performance'] }) + const sudo = row({ + pid: 5200, + ppid: 4242, + tdev: TTYS004, + uid: OTHER_UID, + comm: 'su', + command: '(su)' + }) + installAddon({ + readProcessTable: vi.fn(async (query: LibprocQuery) => tableFor(query, [...TABLE, sudo])) + }) + // A loaded host: ps takes 6s, then omits the process. + let finish: () => void = () => {} + runProcess.mockImplementation( + () => + new Promise((resolve) => { + finish = () => resolve({ code: 1, signal: null, stdout: '', stderr: '', timedOut: false }) + }) + ) + const read = () => runPsInProcess(['-p', '5200', '-o', 'command='], { timeoutMs: 10_000 }) + + const slow = read() + await vi.advanceTimersByTimeAsync(6_000) + finish() + await slow + await read() + expect(runProcess).toHaveBeenCalledTimes(1) + }) + + it('recovers a su pane’s argv within the recovery budget however often it is read', async () => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'performance'] }) + const su = row({ + pid: 5300, + ppid: 4242, + tdev: TTYS004, + uid: OTHER_UID, + comm: 'su', + command: '(su)' + }) + installAddon({ + readProcessTable: vi.fn(async (query: LibprocQuery) => tableFor(query, [...TABLE, su])) + }) + runProcess.mockResolvedValue({ + code: 0, + signal: null, + stdout: ' 5300 su -l\n', + stderr: '', + timedOut: false + }) + const args = ['-o', 'pid=,command=', '-t', 'ttys004'] + + const reads: ReturnType[] = [] + for (let poll = 0; poll < 20; poll += 1) { + // readTerminalProcessRows' budget: a read may wait for the next recovery token. + reads.push(runPsInProcess(args, { ...BUDGET, timeoutMs: 3_000 })) + await vi.advanceTimersByTimeAsync(750) + } + for (const read of await Promise.all(reads)) { + expect(read?.stdout).toContain('5300 su -l\n') + } + // The recovery budget: a burst of 2, then one per 2s over the ~16s the reads span. + expect(runProcess.mock.calls.length).toBeLessThanOrEqual(10) + }) + + it('asks again at once after a recovery ps that timed out, which proved nothing', async () => { + const su = row({ + pid: 5350, + ppid: 4242, + tdev: TTYS004, + uid: OTHER_UID, + comm: 'su', + command: '(su)' + }) + installAddon({ + readProcessTable: vi.fn(async (query: LibprocQuery) => tableFor(query, [...TABLE, su])) + }) + runProcess + .mockResolvedValueOnce({ + code: null, + signal: 'SIGTERM', + stdout: '', + stderr: '', + timedOut: true + }) + .mockResolvedValue({ + code: 0, + signal: null, + stdout: ' 5350 su -l\n', + stderr: '', + timedOut: false + }) + const read = () => runPsInProcess(['-p', '5350', '-o', 'command='], BUDGET) + + expect((await read())?.stdout).toBe('(su)\n') + expect((await read())?.stdout).toBe('su -l\n') + }) + + it('recovers the foreground of a terminal where no process of ours is left (exec su)', async () => { + const root = { tdev: TTYS004, uid: OTHER_UID, flags: 0x2, tpgid: 5400 } + const table = [ + row({ ...root, pid: 4200, ppid: 1, pgid: 4200, comm: 'login', command: '(login)' }), + row({ ...root, pid: 5400, ppid: 4200, pgid: 5400, comm: 'sh', command: '(sh)' }) + ] + installAddon({ readProcessTable: vi.fn(async (query: LibprocQuery) => tableFor(query, table)) }) + runProcess.mockResolvedValue({ + code: 0, + signal: null, + stdout: ' 5400 -sh\n', + stderr: '', + timedOut: false + }) + + const result = await runPsInProcess(['-o', 'pid=,command=', '-t', 'ttys004'], BUDGET) + expect(result?.stdout).toContain('5400 -sh\n') + // The pane's root login is not in the foreground group, so it never costs a ps. + expect(runProcess.mock.calls.map(([spec]) => spec.args.at(-1))).toEqual(['5400']) + }) + + describe('recovered argv age in strict evidence', () => { + // Each case its own pid: the argv cache is process-wide. + let pid = 5500 + const agent = () => + row({ + pid, + ppid: 4242, + tdev: TTYS004, + uid: OTHER_UID, + comm: 'node', + command: '(node)' + }) + const args = ['-o', 'pid=,command=', '-t', 'ttys004'] + // The production reader, stamping on the same monotonic clock the fake timers drive. + const strictReader = () => + createProcessTableSnapshotReader({ + runPs: (_fresh, inputAt) => + runPsInProcess(args, { ...BUDGET, timeoutMs: 10_000, onArgvReadAt: inputAt }).then( + (result) => result?.stdout + ), + now: () => performance.now(), + ttlMs: 0 + }) + const psAnswers = (argv: () => string) => + runProcess.mockImplementation(async () => ({ + code: 0, + signal: null, + stdout: ` ${pid} ${argv()}\n`, + stderr: '', + timedOut: false + })) + + beforeEach(() => { + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'performance'] }) + installAddon({ + readProcessTable: vi.fn(async (query: LibprocQuery) => tableFor(query, [...TABLE, agent()])) + }) + }) + + it('ages a capture by the cached argv it reused, so a same-comm re-exec cannot read as fresh', async () => { + let argv = 'node /opt/claude/cli.js' + psAnswers(() => argv) + const reader = strictReader() + await reader.getSnapshotWithAge() + // node re-execs into a server; same pid, start time and comm. + argv = 'node /srv/server.js' + await vi.advanceTimersByTimeAsync(1_400) + + const snapshot = await reader.getSnapshotWithAge() + expect(snapshot.value).toContain('claude') + expect(snapshot.capturedAgeMs).toBeGreaterThanOrEqual(1_400) + }) + + it('never answers a reader with a recovery begun more than 1.5s before it asked', async () => { + pid = 5600 + let finishSlow: (stdout: string) => void = () => {} + runProcess + .mockImplementationOnce( + () => + new Promise((resolve) => { + finishSlow = (stdout) => + resolve({ code: 0, signal: null, stdout, stderr: '', timedOut: false }) + }) + ) + .mockResolvedValue({ + code: 0, + signal: null, + stdout: ' 5600 node /srv/server.js\n', + stderr: '', + timedOut: false + }) + // Two captures (say the daemon's and main's), sharing one recovery that takes 6s. + void strictReader().getSnapshotWithAge() + await vi.advanceTimersByTimeAsync(5_900) + + const late = strictReader().getSnapshotWithAge() + await vi.advanceTimersByTimeAsync(100) + finishSlow(' 5600 node /opt/claude/cli.js\n') + const snapshot = await late + expect(snapshot.value).toContain('/srv/server.js') + expect(runProcess).toHaveBeenCalledTimes(2) + }) + }) }) diff --git a/src/shared/libproc-process-table.ts b/src/shared/libproc-process-table.ts index f39db95504a..ae707377973 100644 --- a/src/shared/libproc-process-table.ts +++ b/src/shared/libproc-process-table.ts @@ -11,6 +11,7 @@ import { registerNativeRequest } from './libproc-native-reads' import { completeCommands } from './libproc-argv-completion' +import { registerPsFallbackRequest, runPsFallback } from './libproc-ps-fallback' import { compareLikePs, formatDarwinDevice, @@ -21,12 +22,12 @@ import { /** Shaped like a process-host ProcessResult so call sites can substitute it for ps's. */ export type LibprocPsResult = { - code: 0 | 1 | null + code: number | null signal: NodeJS.Signals | null stdout: string - stderr: '' + stderr: string timedOut: boolean - outputTruncated: boolean + outputTruncated?: boolean } export type InProcessPsOptions = { @@ -35,6 +36,14 @@ export type InProcessPsOptions = { /** The caller's ps timeout: past it the read resolves timed out, as the killed ps would. */ timeoutMs: number maxOutputBytes?: number + /** A display poll: any ps it needs waits behind interactive and fresh readers in the fork budget. */ + background?: boolean + /** Called with when (monotonic ms) the result's oldest recovered argv was read, when that predates this request: + * evidence built on the result is as old as that input, not as the table read. */ + onArgvReadAt?: (monotonicMs: number) => void + /** The caller runs its own ps when the in-process reader is down, rather than share the fallback budget, which may + * time it out (signalling paths that must not lose their read). */ + callerRunsPs?: boolean } const TIMED_OUT: LibprocPsResult = { @@ -95,19 +104,33 @@ export function inProcessDeadlineAllowanceMs(timeoutMs: number): number { return loadLibprocAddon() && isNativeReaderHealthy() ? inProcessSliceMs(timeoutMs) : 0 } -/** ps output for `args`, read in process; null when the addon or the invocation is unsupported so callers run ps. */ +/** ps output for `args`, read in process, or by one budgeted /bin/ps while the reader is down; null when the addon or + * the invocation is unsupported (or `callerRunsPs`) so callers run ps. */ export async function runPsInProcess( args: readonly string[], options: InProcessPsOptions ): Promise { const addon = loadLibprocAddon() const query = addon ? parsePsQuery(args) : null - if (!addon || !query || !isNativeReaderHealthy()) { + if (!addon || !query) { return null } const registeredSequence = registerNativeRequest() - const deadline = monotonicNow() + options.timeoutMs + const fallbackSequence = registerPsFallbackRequest() + const requestedAtMs = monotonicNow() + const deadline = requestedAtMs + options.timeoutMs const env = options.env ?? process.env + // ps gets the caller's whole timeout from the moment the reader gives up, as the caller's own ps did. + const fallback = async (): Promise => + options.callerRunsPs + ? null + : runPsFallback( + args, + { env, maxOutputBytes: options.maxOutputBytes }, + fallbackSequence, + monotonicNow() + options.timeoutMs, + options.background + ) const utc = lstartZoneIsUtc(env) const needsLstart = query.columns.includes('lstart') // lstart is `%c`: only the C and en_US forms are reproduced. @@ -117,6 +140,9 @@ export async function runPsInProcess( ) { return null } + if (!isNativeReaderHealthy()) { + return fallback() + } const includeCommand = query.columns.includes('command') const ctype = localeCategory(env, 'LC_CTYPE') const nativeQuery: LibprocQuery = { @@ -130,7 +156,7 @@ export async function runPsInProcess( try { const read = await readFreshNativeTable(addon, nativeQuery, registeredSequence, deadline) if (read === 'stalled') { - return null + return fallback() } if (read === DEADLINE_PASSED) { return TIMED_OUT @@ -144,7 +170,7 @@ export async function runPsInProcess( if (devices.size > 0) { const named = await nameTerminals(devices, deadline) if (named === 'stalled') { - return null + return fallback() } if (named === DEADLINE_PASSED) { return TIMED_OUT @@ -153,14 +179,25 @@ export async function runPsInProcess( } } if (rows && includeCommand) { - await completeCommands(rows, query.pids !== null, ctype, env, deadline) + await completeCommands( + rows, + query.pids !== null ? 'pids' : query.tty !== null ? 'tty' : 'all', + ctype, + env, + deadline, + options.background === true + ) } } catch { - return null + return fallback() } if (monotonicNow() > deadline) { return TIMED_OUT } + const argvReadAt = Math.min(...(rows ?? []).map((row) => row.commandReadAtMs ?? Infinity)) + if (argvReadAt < requestedAtMs) { + options.onArgvReadAt?.(argvReadAt) + } // ps exits 1 when -t names no terminal or a -p/-t selection matched nothing. if (rows === null) { return { ...TIMED_OUT, code: 1, signal: null, timedOut: false } @@ -207,18 +244,12 @@ export async function runPsInProcess( } } -/** execFile-shaped: rejects on a nonzero exit or timeout like a failed ps child; null when ps must run. */ -export async function readPsStdoutInProcess( - args: readonly string[], - options: InProcessPsOptions -): Promise { - const result = await runPsInProcess(args, options) - if (!result) { - return null - } +function stdoutLikeExecFile(args: readonly string[], result: LibprocPsResult): string { if (result.timedOut || result.code !== 0) { throw Object.assign( - new Error(`Command failed: ps ${args.join(' ')}${result.timedOut ? ' (timed out)' : ''}`), + new Error( + `Command failed: ps ${args.join(' ')}${result.timedOut ? ' (timed out)' : ''}\n${result.stderr}` + ), { code: result.code, killed: result.timedOut, signal: result.signal } ) } @@ -229,3 +260,34 @@ export async function readPsStdoutInProcess( } return result.stdout } + +/** execFile-shaped: rejects on a nonzero exit or timeout like a failed ps child; null when ps must run. */ +export async function readPsStdoutInProcess( + args: readonly string[], + options: InProcessPsOptions +): Promise { + const result = await runPsInProcess(args, options) + return result ? stdoutLikeExecFile(args, result) : null +} + +/** execFile-shaped /bin/ps through the shared fork budget, for captures that must run ps (a held naming pass); + * only a ps spawned after this call answers it. */ +export async function readPsStdoutBudgeted( + args: readonly string[], + options: { + env: NodeJS.ProcessEnv + timeoutMs: number + maxOutputBytes?: number + background?: boolean + } +): Promise { + const registered = registerPsFallbackRequest() + const result = await runPsFallback( + args, + { env: options.env, maxOutputBytes: options.maxOutputBytes }, + registered, + monotonicNow() + options.timeoutMs, + options.background + ) + return stdoutLikeExecFile(args, result) +} diff --git a/src/shared/libproc-ps-fallback.test.ts b/src/shared/libproc-ps-fallback.test.ts new file mode 100644 index 00000000000..81442b516fb --- /dev/null +++ b/src/shared/libproc-ps-fallback.test.ts @@ -0,0 +1,185 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +const { runProcess, execFile } = vi.hoisted(() => ({ runProcess: vi.fn(), execFile: vi.fn() })) +vi.mock('@orca/process-host', () => ({ runProcess })) +vi.mock('node:child_process', () => ({ execFile })) + +import { setLibprocAddonForTests, type LibprocAddon } from './libproc-addon' +import { resetNativeReadsForTests } from './libproc-native-reads' +import { runPsInProcess } from './libproc-process-table' +import { resetPsFallbackForTests } from './libproc-ps-fallback' +import { captureProcessTable } from './process-table-ps-capture' +import { PS_ARGS } from './process-table-snapshot' + +const POLL_MS = 750 +const STREAMS = 13 +const ENV = { LC_ALL: 'C', LANG: 'C' } + +function installAddon(readProcessTable: LibprocAddon['readProcessTable']): void { + setLibprocAddonForTests({ + readProcessTable, + readProcessCwd: async () => null, + readProcessPath: async () => null + }) +} + +/** The call-site shape every poller uses: the in-process reader, else its own ps. True when ps answered. */ +async function poll(pid: number, background = false, timeoutMs = 2000): Promise { + const args = ['-o', 'pid=,ppid=,pgid=,lstart=', '-p', String(pid)] + const options = { env: ENV, timeoutMs, maxOutputBytes: 4096, background } + const result = + (await runPsInProcess(args, options)) ?? (await runProcess({ program: 'ps', args, ...options })) + return !result.timedOut +} + +async function overOneMinute( + streams = STREAMS, + background = false +): Promise<{ forks: number; servedPerStream: number[] }> { + const servedPerStream = Array(streams).fill(0) + const timers = Array.from({ length: streams }, (_, stream) => + setInterval( + () => + void poll(7000 + stream, background).then((served) => { + servedPerStream[stream] += Number(served) + }), + POLL_MS + ) + ) + await vi.advanceTimersByTimeAsync(60_000) + timers.forEach(clearInterval) + await vi.advanceTimersByTimeAsync(2_000) + return { forks: runProcess.mock.calls.length, servedPerStream } +} + +beforeEach(() => { + vi.useFakeTimers({ + toFake: ['setTimeout', 'clearTimeout', 'setInterval', 'clearInterval', 'performance'] + }) + runProcess.mockImplementation(async ({ args }: { args: string[] }) => ({ + code: 0, + signal: null, + stdout: ` ${args.at(-1)} 1 1 Thu Sep 3 16:02:01 2026\n`, + stderr: '', + timedOut: false, + outputTruncated: false + })) +}) + +afterEach(() => { + vi.useRealTimers() + runProcess.mockReset() + setLibprocAddonForTests(undefined) + resetNativeReadsForTests() + resetPsFallbackForTests() +}) + +describe('ps fallback budget', () => { + it('bounds ps forks across pollers while native reads keep rejecting', async () => { + const readProcessTable = vi.fn(async () => { + throw new Error('sysctl failed') + }) + installAddon(readProcessTable) + + const { forks, servedPerStream } = await overOneMinute() + expect(forks).toBeLessThanOrEqual(70) + // Invocations take turns, so no poller is starved while another wins every token. + expect(Math.min(...servedPerStream)).toBeGreaterThanOrEqual(4) + // Backoff: rejection stops being retried on every poll. + expect(readProcessTable.mock.calls.length).toBeLessThan(15) + }) + + it('bounds ps forks across pollers while the stalled-pool breaker is open', async () => { + installAddon(vi.fn(() => new Promise(() => {}))) + // Fill the cap with stalled reads and trip the breaker. + for (let index = 0; index < 9; index += 1) { + void runPsInProcess(['-p', String(5000 + index), '-o', 'pid='], { timeoutMs: 50 }) + } + await vi.advanceTimersByTimeAsync(50) + runProcess.mockClear() + + const { forks, servedPerStream } = await overOneMinute() + expect(forks).toBeLessThanOrEqual(70) + expect(Math.min(...servedPerStream)).toBeGreaterThanOrEqual(4) + }) + + it('never answers a request with a fallback ps spawned before it, but shares one spawned after', async () => { + installAddon(async () => { + throw new Error('sysctl failed') + }) + const finish: ((stdout: string) => void)[] = [] + runProcess.mockImplementation( + () => + new Promise((resolve) => + finish.push((stdout) => + resolve({ code: 0, signal: null, stdout, stderr: '', timedOut: false }) + ) + ) + ) + // pid= only: a descendant read that would signal whatever pids it returns. + const read = () => runPsInProcess(['-axo', 'pid='], { timeoutMs: 2000 }) + + const early = read() + await vi.advanceTimersByTimeAsync(0) + expect(runProcess).toHaveBeenCalledTimes(1) + const late = [read(), read()] + await vi.advanceTimersByTimeAsync(0) + expect(runProcess).toHaveBeenCalledTimes(2) + finish[0]?.(' 9999\n') + finish[1]?.(' 4242\n') + expect((await early)?.stdout).toBe(' 9999\n') + expect((await Promise.all(late)).map((result) => result?.stdout)).toEqual([ + ' 4242\n', + ' 4242\n' + ]) + }) + + it('sends forced subprocess captures (a held naming pass) through the same budget', async () => { + const platform = Object.getOwnPropertyDescriptor(process, 'platform')! + Object.defineProperty(process, 'platform', { configurable: true, value: 'darwin' }) + try { + execFile.mockImplementation((_file, _args, _options, callback) => + callback(null, { stdout: ' 1 0\n', stderr: '' }) + ) + let captured = 0 + // captureDarwinProcessTable's subprocessCapture while a naming pass is held, from five readers. + const timers = Array.from({ length: 5 }, () => + setInterval( + () => + void captureProcessTable(PS_ARGS, 2000, true).then( + () => (captured += 1), + () => {} + ), + POLL_MS + ) + ) + await vi.advanceTimersByTimeAsync(60_000) + timers.forEach(clearInterval) + expect(execFile).not.toHaveBeenCalled() + expect(runProcess.mock.calls.length).toBeLessThanOrEqual(65) + expect(captured).toBeGreaterThan(runProcess.mock.calls.length) + } finally { + Object.defineProperty(process, 'platform', platform) + } + }) + + it('serves interactive reads ahead of a storm of display polls, still fair among the polls', async () => { + installAddon(async () => { + throw new Error('sysctl failed') + }) + // A launch paste's terminal read (3s) every 5s, beside 20 steady-state display streams. + let interactive = 0 + let interactiveServed = 0 + const paste = setInterval(() => { + interactive += 1 + void poll(9000 + interactive, false, 3000).then((served) => { + interactiveServed += Number(served) + }) + }, 5_000) + const { servedPerStream } = await overOneMinute(20, true) + clearInterval(paste) + + expect(interactiveServed).toBe(interactive) + expect(Math.min(...servedPerStream)).toBeGreaterThanOrEqual(2) + }) +}) diff --git a/src/shared/libproc-ps-fallback.ts b/src/shared/libproc-ps-fallback.ts new file mode 100644 index 00000000000..c1722c952fd --- /dev/null +++ b/src/shared/libproc-ps-fallback.ts @@ -0,0 +1,248 @@ +import { runProcess } from '@orca/process-host' +import type { ProcessResult } from '@orca/process-host/process-spec' +import { + DEADLINE_PASSED, + createSettleNotifier, + monotonicNow, + type Settled +} from './deadline-settle-notifier' + +/** Every setuid /bin/ps the in-process reader still needs runs through a fork budget: fallbacks while the reader is + * down (stalled pool, rejecting addon), forced subprocess captures, and argv recovery for other users' processes. + * Without one, each poller forks its own ps per poll per stream, the rate #31783 removed. */ + +type Flight = { + /** Creation order; a flight serves only requests registered at or before it, so its ps started after them. */ + sequence: number + readonly submitted: boolean + wait: (deadline: number) => Promise | typeof DEADLINE_PASSED> +} + +type Spec = { args: readonly string[]; env: NodeJS.ProcessEnv; maxOutputBytes?: number } + +/** `spawnedAtMs` (monotonic): when this ps started, so what it read is no older; absent when it never ran. */ +export type BudgetedPsResult = ProcessResult & { spawnedAtMs?: number } + +/** Requests for one invocation waiting on a token; served together, in the order their invocation first queued. */ +type Queued = { spec: Spec; waiters: Map<() => void, number> } + +const TIMED_OUT: ProcessResult = { + code: null, + signal: 'SIGTERM', + stdout: '', + stderr: '', + timedOut: true, + outputTruncated: false +} + +const FLIGHT_LIMIT = 64 +let nextSequence = 0 + +/** Register before any in-process attempt: only a ps spawned from now on may answer this request. */ +export function registerPsFallbackRequest(): number { + return nextSequence +} + +// ps output follows its locale, zone and (without -ww) width, so only identically shaped invocations share a run. +function flightKey(spec: Spec): string { + const shape = ['LC_ALL', 'LC_CTYPE', 'LC_TIME', 'LANG', 'TZ', 'COLUMNS'].map( + (name) => spec.env[name] ?? null + ) + return JSON.stringify([spec.args, shape, spec.maxOutputBytes ?? null]) +} + +type PsForkBudget = { + /** `background` (a display poll) queues behind interactive and fresh readers, so polls cannot run them past deadline. */ + run: ( + spec: Spec, + registeredSequence: number, + deadline: number, + background: boolean + ) => Promise + reset: () => void +} + +function createPsForkBudget(burst: number, refillMs: number): PsForkBudget { + // Keeps each invocation's latest flight, finished or not, for requests that queued meanwhile. + const flights = new Map() + // [interactive, background]; within a queue insertion order is service order, so no invocation takes every token. + const queues = [new Map(), new Map()] as const + const queued = (): number => queues[0].size + queues[1].size + let tokens = burst + let refilledAtMs = Number.NEGATIVE_INFINITY + let grantTimer: ReturnType | undefined + + const refill = (now: number): void => { + tokens = Math.min(burst, tokens + Math.max(0, now - refilledAtMs) / refillMs) + refilledAtMs = now + } + + const submit = (key: string, spec: Spec, timeoutMs: number): void => { + const submission = { submitted: false } + const pending = new Promise((resolve, reject) => { + // A microtask later, so requests registered in the same turn share this ps. + queueMicrotask(() => { + submission.submitted = true + const spawnedAtMs = monotonicNow() + runProcess({ + program: '/bin/ps', + args: [...spec.args], + env: spec.env, + timeoutMs, + ...(spec.maxOutputBytes === undefined ? {} : { maxOutputBytes: spec.maxOutputBytes }) + }).then((result) => resolve({ ...result, spawnedAtMs }), reject) + }) + }) + flights.delete(key) + for (const oldest of flights.keys()) { + if (flights.size < FLIGHT_LIMIT) { + break + } + flights.delete(oldest) + } + flights.set(key, { + sequence: nextSequence++, + get submitted() { + return submission.submitted + }, + wait: createSettleNotifier(pending) + }) + } + + const grant = (): void => { + grantTimer = undefined + const now = monotonicNow() + refill(now) + for (const queue of queues) { + for (const [key, entry] of queue) { + if (tokens < 1) { + break + } + queue.delete(key) + const latest = Math.max(...entry.waiters.values()) + if (entry.waiters.size === 0 || latest <= now) { + continue + } + tokens -= 1 + submit(key, entry.spec, Math.floor(latest - now)) + for (const wake of entry.waiters.keys()) { + wake() + } + } + } + if (queued() > 0) { + grantTimer = setTimeout(grant, Math.ceil((1 - tokens) * refillMs)) + } + } + + const run = async ( + spec: Spec, + registeredSequence: number, + deadline: number, + background: boolean + ): Promise => { + const key = flightKey(spec) + const ahead = (): number => (background ? queued() : queues[0].size) + for (;;) { + const flight = flights.get(key) + if (flight && (!flight.submitted || flight.sequence >= registeredSequence)) { + const settled = await flight.wait(deadline) + if (settled === DEADLINE_PASSED) { + return TIMED_OUT + } + if (!settled.ok) { + throw settled.error + } + return settled.value + } + const now = monotonicNow() + if (deadline <= now) { + return TIMED_OUT + } + refill(now) + if (ahead() === 0 && tokens >= 1) { + tokens -= 1 + submit(key, spec, Math.floor(deadline - now)) + continue + } + const served = await new Promise((resolve) => { + // One entry per invocation: an interactive request promotes a queued poll's, a poll joins an interactive one. + const [interactive, polls] = queues + const entry = interactive.get(key) ?? polls.get(key) ?? { spec, waiters: new Map() } + if (!background) { + polls.delete(key) + interactive.set(key, entry) + } else if (!interactive.has(key)) { + polls.set(key, entry) + } + const timer = setTimeout( + () => { + entry.waiters.delete(wake) + resolve(false) + }, + Math.max(0, deadline - now) + ) + const wake = (): void => { + clearTimeout(timer) + resolve(true) + } + entry.waiters.set(wake, deadline) + grantTimer ??= setTimeout(grant, Math.ceil((1 - tokens) * refillMs)) + }) + if (!served) { + return TIMED_OUT + } + } + } + + return { + run, + reset: () => { + clearTimeout(grantTimer) + grantTimer = undefined + flights.clear() + queues.forEach((queue) => queue.clear()) + tokens = burst + refilledAtMs = Number.NEGATIVE_INFINITY + } + } +} + +// While the reader is down: a burst of 4 forks, then one a second, however many streams poll. +const fallbackBudget = createPsForkBudget(4, 1_000) +// argv only setuid ps can read (sudo/su in a pane): a burst of 2, then one every 2s. +const argvRecoveryBudget = createPsForkBudget(2, 2_000) + +/** /bin/ps for a request registered at `registeredSequence`, sharing a run begun after it; times out like a killed ps + * when neither a shared run nor a fork token arrives before `deadline`. */ +export function runPsFallback( + args: readonly string[], + options: { env: NodeJS.ProcessEnv; maxOutputBytes?: number }, + registeredSequence: number, + deadline: number, + background = false +): Promise { + return fallbackBudget + .run({ args, ...options }, registeredSequence, deadline, background) + .then(({ spawnedAtMs: _spawnedAtMs, ...result }) => result) +} + +/** `ps -o pid=,command= -p …` for argv the in-process reader could not read, under its own smaller budget. */ +export function runArgvRecoveryPs( + args: readonly string[], + options: { env: NodeJS.ProcessEnv; maxOutputBytes?: number }, + deadline: number, + background: boolean +): Promise { + return argvRecoveryBudget.run( + { args, ...options }, + registerPsFallbackRequest(), + deadline, + background + ) +} + +export function resetPsFallbackForTests(): void { + fallbackBudget.reset() + argvRecoveryBudget.reset() +} diff --git a/src/shared/libproc-ps-format.ts b/src/shared/libproc-ps-format.ts index a3da57b7cd1..8110fae6f41 100644 --- a/src/shared/libproc-ps-format.ts +++ b/src/shared/libproc-ps-format.ts @@ -1,13 +1,13 @@ import type { LibprocProcessRow } from './libproc-addon' -const SIDL = 1 +export const SIDL = 1 const SSTOP = 4 -const SZOMB = 5 +export const SZOMB = 5 const P_CONTROLT = 0x2 const P_PPWAIT = 0x10 const P_SYSTEM = 0x200 const P_TRACED = 0x800 -const P_WEXIT = 0x2000 +export const P_WEXIT = 0x2000 export const NODEV = -1 /** ps's state letters, except running vs sleeping: that needs another task's thread states (ps is setuid for it), so live processes read `S`. */ diff --git a/src/shared/process-table-ps-capture.ts b/src/shared/process-table-ps-capture.ts new file mode 100644 index 00000000000..76f877e16e6 --- /dev/null +++ b/src/shared/process-table-ps-capture.ts @@ -0,0 +1,57 @@ +import { execFile as execFileCb } from 'node:child_process' +import { promisify } from 'node:util' +import { readPsStdoutBudgeted, readPsStdoutInProcess } from './libproc-process-table' +import { PS_MAX_BUFFER_BYTES, ProcessTableCaptureError } from './process-table-snapshot' +import { psLocaleEnvironment } from './ps-locale-environment' + +const execFile = promisify(execFileCb) + +// command= costs 1.15s for 1,948 idle processes, 1.3–6s at load 27; 3s failed 6/20 captures. +// Allow 15s for identity reads on busy hosts while bounding wedged ps. +export const PS_TIMEOUT_MS = 15_000 + +/** Reject captures truncated at the subprocess ceiling or containing no rows. */ +export function assertWholeCapture(stdout: string): string { + if (Buffer.byteLength(stdout, 'utf-8') >= PS_MAX_BUFFER_BYTES) { + throw new ProcessTableCaptureError('capture_truncated') + } + if (!/\S/.test(stdout)) { + throw new ProcessTableCaptureError('empty_capture') + } + return stdout +} + +/** One ps-shaped capture: in process where possible; `background` (a display poll) yields the fork budget, and + * `onArgvReadAt` learns when the oldest recovered argv in it was read. */ +export async function captureProcessTable( + args: readonly string[], + timeout = PS_TIMEOUT_MS, + ps = false, + background = false, + onArgvReadAt?: (monotonicMs: number) => void +) { + const env = psLocaleEnvironment() + const options = { timeoutMs: timeout, maxOutputBytes: PS_MAX_BUFFER_BYTES, background } + let stdout = ps ? null : await readPsStdoutInProcess(args, { env, ...options, onArgvReadAt }) + try { + // A forced subprocess capture (held naming pass) shares the fork budget; elsewhere ps is the only reader. + stdout ??= + ps && process.platform === 'darwin' + ? await readPsStdoutBudgeted(args, { env: env ?? process.env, ...options }) + : ( + await execFile('ps', [...args], { + encoding: 'utf-8', + env, + timeout, + maxBuffer: PS_MAX_BUFFER_BYTES + }) + ).stdout + } catch (error) { + // A ceiling hit is truncation, not absence: name it in the domain vocabulary. + if ((error as { code?: unknown } | null)?.code === 'ERR_CHILD_PROCESS_STDIO_MAXBUFFER') { + throw new ProcessTableCaptureError('capture_truncated') + } + throw error + } + return assertWholeCapture(stdout) +} diff --git a/src/shared/process-table-snapshot-reader.ts b/src/shared/process-table-snapshot-reader.ts index 2233156d01f..5a5ba722659 100644 --- a/src/shared/process-table-snapshot-reader.ts +++ b/src/shared/process-table-snapshot-reader.ts @@ -1,8 +1,6 @@ -import { execFile as execFileCb } from 'node:child_process' import { readFile } from 'node:fs/promises' -import { promisify } from 'node:util' -import { psLocaleEnvironment } from './ps-locale-environment' -import { readPsStdoutInProcess } from './libproc-process-table' +import { monotonicNow } from './deadline-settle-notifier' +import { PS_TIMEOUT_MS, assertWholeCapture, captureProcessTable } from './process-table-ps-capture' import { captureDarwinProcessTable, resetDarwinProcessTableCaptureForTests @@ -19,18 +17,14 @@ import { type ProcessTableRow } from './process-table-snapshot' -export { PROCESS_TABLE_SNAPSHOT_MAX_STALENESS_MS, PS_ARGS, PS_MAX_BUFFER_BYTES } - -const execFile = promisify(execFileCb) - -// command= costs 1.15s for 1,948 idle processes, 1.3–6s at load 27; 3s failed 6/20 captures. -// Allow 15s for identity reads on busy hosts while bounding wedged ps. -export const PS_TIMEOUT_MS = 15_000 +export { PROCESS_TABLE_SNAPSHOT_MAX_STALENESS_MS, PS_ARGS, PS_MAX_BUFFER_BYTES, PS_TIMEOUT_MS } type Snapshot = { value: T; capturedAtMs: number; completedAtMs: number } type ProcessTableSnapshotReaderDeps = { - runPs: () => Promise + /** `fresh`: an authoritative read waiting on this capture; anything else is a display poll. `inputAt` takes the + * time (in `now()`'s clock) of any input older than the capture itself, such as recovered argv. */ + runPs: (fresh: boolean, inputAt: (atMs: number) => void) => Promise now: () => number ttlMs?: number } @@ -50,11 +44,13 @@ export function createProcessTableSnapshotReader( let sequence = 0 let freshQueued: { promise: Promise; startSequence: number | null } | null = null - async function runSnapshot(): Promise { - // Stamp evidence age at capture start, TTL at completion: slow captures must remain old yet still coalesce - // instead of spawning one whole-machine ps per caller on loaded hosts. - const capturedAtMs = deps.now() - const promise = deps.runPs() + async function runSnapshot(fresh = false): Promise { + // Stamp evidence age at capture start (or its oldest input), TTL at completion: slow captures must remain old + // yet still coalesce instead of spawning one whole-machine ps per caller on loaded hosts. + let capturedAtMs = deps.now() + const promise = deps.runPs(fresh, (atMs) => { + capturedAtMs = Math.min(capturedAtMs, atMs) + }) inFlight = promise try { const value = await promise @@ -83,7 +79,8 @@ export function createProcessTableSnapshotReader( async function getSnapshotWithAge(): Promise<{ value: T; capturedAgeMs: number }> { const value = await getSnapshot() const capturedAtMs = cached?.value === value ? cached.capturedAtMs : deps.now() - return { value, capturedAgeMs: Math.max(0, deps.now() - capturedAtMs) } + // Whole milliseconds, rounded older: the wire requires integers and an input's clock conversion is fractional. + return { value, capturedAgeMs: Math.max(0, Math.ceil(deps.now() - capturedAtMs)) } } function getFreshSnapshot(): Promise { @@ -112,7 +109,7 @@ export function createProcessTableSnapshotReader( if (entry.startSequence <= requestSequence) { throw new Error('fresh process snapshot did not start after request') } - return runSnapshot() + return runSnapshot(true) }) freshQueued = entry const clearQueued = (): void => { @@ -199,17 +196,6 @@ function createProcessTableCapture( } } -/** Reject captures truncated at the subprocess ceiling or containing no rows. */ -function assertWholeCapture(stdout: string): string { - if (Buffer.byteLength(stdout, 'utf-8') >= PS_MAX_BUFFER_BYTES) { - throw new ProcessTableCaptureError('capture_truncated') - } - if (!/\S/.test(stdout)) { - throw new ProcessTableCaptureError('empty_capture') - } - return stdout -} - /** Field 22 (`starttime`) of `/proc//stat`, read past the parenthesised comm. */ export function parseLinuxProcStatStartTime(stat: string): string | null { const closingParen = stat.lastIndexOf(')') @@ -243,33 +229,21 @@ async function readLinuxProcessStartTimes( return new Map(starts.filter((entry) => entry !== null)) } -async function captureProcessTable(args: readonly string[], timeout = PS_TIMEOUT_MS, ps = false) { - const env = psLocaleEnvironment() - let stdout = ps ? null : await readPsStdoutInProcess(args, { env, timeoutMs: timeout }) - try { - stdout ??= ( - await execFile('ps', [...args], { - encoding: 'utf-8', - env, - timeout, - maxBuffer: PS_MAX_BUFFER_BYTES - }) - ).stdout - } catch (error) { - // A ceiling hit is truncation, not absence: name it in the domain vocabulary. - if ((error as { code?: unknown } | null)?.code === 'ERR_CHILD_PROCESS_STDIO_MAXBUFFER') { - throw new ProcessTableCaptureError('capture_truncated') - } - throw error - } - return assertWholeCapture(stdout) +/** Recovered argv reports monotonic read times; the snapshot readers stamp ages on the wall clock. */ +function onArgvReadAt(inputAt: (atMs: number) => void): (monotonicMs: number) => void { + return (monotonicMs) => inputAt(Date.now() - (monotonicNow() - monotonicMs)) } const processTableReader = createProcessTableSnapshotReader({ - runPs: async () => { + runPs: async (fresh, inputAt) => { const stdout = process.platform === 'darwin' - ? await captureDarwinProcessTable(captureProcessTable, assertWholeCapture, PS_TIMEOUT_MS) + ? await captureDarwinProcessTable( + (args, timeout, ps) => + captureProcessTable(args, timeout, ps, !fresh, onArgvReadAt(inputAt)), + assertWholeCapture, + PS_TIMEOUT_MS + ) : await captureProcessTable(PS_ARGS) if (process.platform !== 'linux') { return createProcessTableCapture(stdout) @@ -282,7 +256,16 @@ const processTableReader = createProcessTableSnapshotReader // Separate shell reader avoids macOS terminal-name costs and waiting on full captures it cannot use. const shellForegroundReader = createProcessTableSnapshotReader({ - runPs: async () => parseShellForegroundRows(await captureProcessTable(SHELL_FOREGROUND_PS_ARGS)), + runPs: async (fresh, inputAt) => + parseShellForegroundRows( + await captureProcessTable( + SHELL_FOREGROUND_PS_ARGS, + PS_TIMEOUT_MS, + false, + !fresh, + onArgvReadAt(inputAt) + ) + ), now: () => Date.now() }) diff --git a/src/shared/process-table-snapshot.test.ts b/src/shared/process-table-snapshot.test.ts index 3dd1efc6552..134402aa975 100644 --- a/src/shared/process-table-snapshot.test.ts +++ b/src/shared/process-table-snapshot.test.ts @@ -141,6 +141,21 @@ describe('process-table-snapshot reader', () => { expect((await reader.getSnapshotWithAge()).capturedAgeMs).toBe(4_000) }) + it('reports an older input’s age in whole milliseconds, rounded older, for the integer-only wire', async () => { + const clock = 10_000 + const reader = createProcessTableSnapshotReader({ + // Recovered argv read 750.5ms before the capture, converted from the monotonic clock. + runPs: async (_fresh, inputAt) => { + inputAt(clock - 750.5) + return 'scan' + }, + now: () => clock, + ttlMs: 500 + }) + + expect((await reader.getSnapshotWithAge()).capturedAgeMs).toBe(751) + }) + it('does not cache failures and retries on the next call', async () => { let scans = 0 const reader = createProcessTableSnapshotReader({ diff --git a/src/shared/process-table-snapshot.ts b/src/shared/process-table-snapshot.ts index e7130098ab3..91903d7ffa4 100644 --- a/src/shared/process-table-snapshot.ts +++ b/src/shared/process-table-snapshot.ts @@ -38,6 +38,12 @@ export const SHELL_FOREGROUND_PS_ARGS = [ 'pid=,ppid=,pgid=,tpgid=,stat=,command=' ] as readonly string[] +/** The name inside ps's `(comm)`, printed when argv was unreadable (another user's process, an exiting one, or a + * recovery the fork budget withheld): its identity is unknown, so it never proves which program holds a terminal. */ +export function unreadableArgvName(command: string): string | null { + return /^\((.*)\)$/.exec(command.trim())?.[1] ?? null +} + /** Match shell-tier columns exactly: the full parser’s optional tty/start pair eats argv like python 3 app.py and misreads zombies. * Drop invalid rows, but reject empty captures so missing data cannot prove the shell exited. */ export function parseShellForegroundRows(stdout: string): ProcessTableRow[] { diff --git a/src/shared/shell-process-readiness.ts b/src/shared/shell-process-readiness.ts index d6b2518c599..165059055ac 100644 --- a/src/shared/shell-process-readiness.ts +++ b/src/shared/shell-process-readiness.ts @@ -3,8 +3,9 @@ import { constants } from 'node:fs' import { access, readlink, realpath, stat } from 'node:fs/promises' import { delimiter, isAbsolute, resolve } from 'node:path' import { promisify } from 'node:util' +import { runProcess } from '@orca/process-host' import { LSOF_NONBLOCKING_ARGS } from './lsof-nonblocking-args' -import { readLibprocProcessPath } from './libproc-addon' +import { readLibprocProcessFilePath } from './libproc-addon' import { readPsStdoutInProcess } from './libproc-process-table' import { PS_MAX_BUFFER_BYTES } from './process-table-snapshot' @@ -32,19 +33,20 @@ async function readExecutablePath(pid: number): Promise { if (process.platform !== 'darwin') { return null } - const inProcess = readLibprocProcessPath(pid) + const inProcess = await readLibprocProcessFilePath('executable', pid) if (inProcess !== undefined) { return inProcess } - const { stdout } = await execFile( - '/usr/sbin/lsof', - [...LSOF_NONBLOCKING_ARGS, '-a', '-p', String(pid), '-d', 'txt', '-Fn'], - { - encoding: 'utf8', - timeout: PROCESS_READINESS_TIMEOUT_MS - } - ) - return parseDarwinExecutablePath(stdout) + // execFile settles only once lsof exits, and SIGTERM cannot interrupt it in a dead mount; runProcess gives up. + const result = await runProcess({ + program: '/usr/sbin/lsof', + args: [...LSOF_NONBLOCKING_ARGS, '-a', '-p', String(pid), '-d', 'txt', '-Fn'], + timeoutMs: PROCESS_READINESS_TIMEOUT_MS + }) + if (result.timedOut || result.code !== 0) { + throw new Error(`lsof -p ${pid} failed${result.timedOut ? ' (timed out)' : ''}`) + } + return parseDarwinExecutablePath(result.stdout) } /** Rejects like execFile when ps exits nonzero, so callers keep their error paths. */