fix(relay): stop three CPU growth terms in a long-running remote session

pty.resize gated only on `managed.disposed`, which is bookkeeping rather than
liveness. A shell that exits without node-pty's `onExit` leaves an undisposed
entry holding a closed master fd, and UnixTerminal.resize has no fd guard, so
the ioctl threw `ioctl(2) failed, EBADF` into the dispatcher's generic
parse-error catch. Nothing retired the entry, so it stayed advertised and kept
activePtyCount above zero -- which is what stops a relay with an unlimited
grace from reaching its idle-no-ptys exit (#12423). Probe liveness with the
same helper attach/listProcesses use, retire a provably dead pid, and contain
an ioctl failure over a live-or-unverifiable process.

processHasChildren forked `pgrep -P` per pane per inspection poll, uncached.
procps-ng opens six procfs files per process to resolve one ppid, so each call
cost O(host process count). Answer from the TTL-cached `ps` table the same RPC
already captured for the foreground lookup (#13537).

The remote AI Vault scanner had no parse cache at all, so every forced rescan
re-read and re-parsed the whole transcript corpus, including files untouched
for a month. Give it the mtime+size keyed memo the local scanner has (#13753).
This commit is contained in:
Neil
2026-09-02 14:08:57 -07:00
parent 817827be5b
commit e9da1f631e
7 changed files with 540 additions and 19 deletions
@@ -0,0 +1,180 @@
import { beforeEach, describe, expect, it } from 'vitest'
import type { FileReadResult } from '../providers/types'
import { getRemoteHostPlatform } from '../ssh/ssh-remote-platform'
import { resetRemoteSessionParseCacheForTests } from './remote-session-parse-cache'
import { scanRemoteAiVaultSessions } from './remote-session-scanner'
import { MemoryRemoteProvider, jsonLines } from './remote-session-scanner-test-fixtures'
/**
* Counts whole-transcript reads, which is the cost #13753 is about. Codex's
* per-scan `session_index.jsonl` title lookup is one small file and is not part
* of the corpus term, so it is excluded rather than asserted on.
*/
class CountingRemoteProvider extends MemoryRemoteProvider {
readonly readFilePaths: string[] = []
override async readFile(filePath: string): Promise<FileReadResult> {
if (filePath.includes('/sessions/')) {
this.readFilePaths.push(filePath)
}
return await super.readFile(filePath)
}
}
function transcript(sessionId: string, title: string, timestamp: string): string {
return jsonLines([
{
timestamp,
type: 'session_meta',
payload: { id: sessionId, cwd: '/home/ada/repo' }
},
{
timestamp,
type: 'response_item',
payload: { type: 'message', role: 'user', content: [{ type: 'text', text: title }] }
}
])
}
function scan(provider: CountingRemoteProvider): ReturnType<typeof scanRemoteAiVaultSessions> {
return scanRemoteAiVaultSessions({
provider,
executionHostId: 'ssh:dev-box',
remoteHome: '/home/ada',
hostPlatform: getRemoteHostPlatform('linux-x64')
})
}
describe('remote AI Vault transcript re-reads', () => {
beforeEach(() => {
resetRemoteSessionParseCacheForTests()
})
it('does not re-read an unchanged corpus on the next scan', async () => {
const provider = new CountingRemoteProvider()
for (const day of ['07/07', '07/25', '08/10']) {
provider.addFile(
`/home/ada/.codex/sessions/2026/${day}/rollout-${day.replace('/', '')}.jsonl`,
transcript(
`session-${day.replace('/', '')}`,
`Work from ${day}`,
'2026-07-07T01:00:00.000Z'
),
1_000
)
}
const first = await scan(provider)
expect(first.sessions).toHaveLength(3)
expect(provider.readFilePaths).toHaveLength(3)
provider.readFilePaths.length = 0
const second = await scan(provider)
// Historical transcripts are immutable; a second pass must cost zero reads.
expect(provider.readFilePaths).toEqual([])
expect(second.sessions.map((session) => session.title)).toEqual(
first.sessions.map((session) => session.title)
)
})
it('re-reads a transcript that actually changed', async () => {
const provider = new CountingRemoteProvider()
const path = '/home/ada/.codex/sessions/2026/08/31/rollout-live.jsonl'
provider.addFile(
path,
transcript('live-session', 'First prompt', '2026-08-31T01:00:00.000Z'),
1_000
)
await scan(provider)
provider.readFilePaths.length = 0
provider.addFile(
path,
transcript('live-session', 'Second prompt', '2026-08-31T02:00:00.000Z'),
2_000
)
const result = await scan(provider)
expect(provider.readFilePaths).toEqual([path])
expect(result.sessions[0]?.title).toBe('Second prompt')
})
it('re-reads when only the size changed under an unchanged mtime', async () => {
const provider = new CountingRemoteProvider()
const path = '/home/ada/.codex/sessions/2026/08/31/rollout-grown.jsonl'
provider.addFile(path, transcript('grown-session', 'Short', '2026-08-31T01:00:00.000Z'), 1_000)
await scan(provider)
provider.readFilePaths.length = 0
provider.addFile(
path,
transcript(
'grown-session',
'A much longer first prompt than before',
'2026-08-31T01:00:00.000Z'
),
1_000
)
const result = await scan(provider)
expect(provider.readFilePaths).toEqual([path])
expect(result.sessions[0]?.title).toBe('A much longer first prompt than before')
})
it('does not serve a cached parse to a different execution host', async () => {
const provider = new CountingRemoteProvider()
const path = '/home/ada/.codex/sessions/2026/08/31/rollout-host.jsonl'
provider.addFile(
path,
transcript('host-session', 'Host scoped', '2026-08-31T01:00:00.000Z'),
1_000
)
await scan(provider)
provider.readFilePaths.length = 0
const other = await scanRemoteAiVaultSessions({
provider,
executionHostId: 'ssh:other-box',
remoteHome: '/home/ada',
hostPlatform: getRemoteHostPlatform('linux-x64')
})
expect(provider.readFilePaths).toEqual([path])
expect(other.sessions[0]?.executionHostId).toBe('ssh:other-box')
})
it('does not cache a read that failed', async () => {
const provider = new CountingRemoteProvider()
const path = '/home/ada/.codex/sessions/2026/08/31/rollout-flaky.jsonl'
provider.addFile(
path,
transcript('flaky-session', 'Recovered', '2026-08-31T01:00:00.000Z'),
1_000
)
let failNextRead = true
const originalReadFile = provider.readFile.bind(provider)
provider.readFile = async (filePath: string): Promise<FileReadResult> => {
if (failNextRead && filePath === path) {
failNextRead = false
provider.readFilePaths.push(filePath)
throw new Error('EIO: transient relay read failure')
}
return await originalReadFile(filePath)
}
const failed = await scan(provider)
expect(failed.sessions).toEqual([])
expect(failed.issues).toHaveLength(1)
provider.readFilePaths.length = 0
const recovered = await scan(provider)
expect(provider.readFilePaths).toEqual([path])
expect(recovered.sessions[0]?.title).toBe('Recovered')
})
})
@@ -0,0 +1,99 @@
import type { AiVaultSession } from '../../shared/ai-vault-types'
import type { RemoteScannerContext, RemoteSessionCandidate } from './remote-session-scanner-types'
// Matches the local scanner's cap. The relay sidecar is forked with
// --max-old-space-size=384, and a retained session row is a title, a preview
// window and counters — orders of magnitude smaller than the transcript it was
// parsed from, which is what the cache stops us re-reading.
const MAX_CACHE_ENTRIES = 4096
type RemoteSessionParseCacheEntry = {
mtimeMs: number
sizeBytes: number | null
hostKey: string
session: AiVaultSession | null
}
// Module scope so it outlives one scan: the sidecar is retired only after 10
// idle minutes, so it spans many passes of a 30s cadence.
const cache = new Map<string, RemoteSessionParseCacheEntry>()
export type RemoteSessionParseStats = { reused: number; parsed: number }
export function createRemoteSessionParseStats(): RemoteSessionParseStats {
return { reused: 0, parsed: 0 }
}
export function resetRemoteSessionParseCacheForTests(): void {
cache.clear()
}
/** Identity of the host a parse result belongs to; a result is not portable across either field. */
export function remoteSessionParseHostKey(context: RemoteScannerContext): string {
return `${context.executionHostId}\u0000${context.hostPlatform.relayPlatform}`
}
function storeEntry(path: string, entry: RemoteSessionParseCacheEntry): void {
cache.delete(path)
cache.set(path, entry)
if (cache.size > MAX_CACHE_ENTRIES) {
const oldest = cache.keys().next()
if (!oldest.done) {
cache.delete(oldest.value)
}
}
}
/**
* Parse a remote transcript, reusing the previous result when the file is
* provably unchanged.
*
* Why this exists: the remote scanner had no cache of any kind, so every pass
* re-read and re-parsed the whole corpus — up to 3000 whole-file reads, GBs of
* JSONL, including July transcripts that had not changed in a month — which is
* what pegged the relay host on the renderer's 30s forced-rescan cadence
* (#13753). The local scanner has had `parseAgentSessionFileCached` for exactly
* this reason; this is its remote counterpart.
*
* `(mtimeMs, sizeBytes)` is a sound validity key here because discovery already
* folds a source's `contentDependencyPath` stat into both fields
* (remote-session-scanner-discovery.ts), so a metadata-only transcript whose
* companion file changed still looks changed.
*
* Only a completed parse is stored. A read that threw stays uncached so a
* transient filesystem failure cannot pin a wrong answer for the corpus's life.
*/
export async function parseRemoteSessionFileCached(args: {
candidate: RemoteSessionCandidate
hostKey: string
parse: () => Promise<AiVaultSession | null>
stats?: RemoteSessionParseStats
}): Promise<AiVaultSession | null> {
const { file } = args.candidate
const entry = cache.get(file.path)
const unchanged =
entry !== undefined &&
entry.hostKey === args.hostKey &&
entry.mtimeMs === file.mtimeMs &&
(entry.sizeBytes === null || file.sizeBytes === undefined || entry.sizeBytes === file.sizeBytes)
if (unchanged) {
if (args.stats) {
args.stats.reused++
}
// Refresh recency without re-parsing so the LRU evicts cold paths first.
storeEntry(file.path, entry)
return entry.session
}
const session = await args.parse()
if (args.stats) {
args.stats.parsed++
}
storeEntry(file.path, {
mtimeMs: file.mtimeMs,
sizeBytes: file.sizeBytes ?? null,
hostKey: args.hostKey,
session
})
return session
}
+18 -6
View File
@@ -12,6 +12,10 @@ import {
dedupeCodexRolloutFileAliases,
dedupeCodexSessionsBySessionId
} from './codex-session-root-dedup'
import {
parseRemoteSessionFileCached,
remoteSessionParseHostKey
} from './remote-session-parse-cache'
import { discoverRemoteSourceCandidates } from './remote-session-scanner-discovery'
import { remoteSessionSources } from './remote-session-scanner-sources'
import type {
@@ -224,12 +228,20 @@ async function parseRemoteSessionCandidate(
): Promise<AiVaultSession | null> {
try {
throwIfAiVaultScanCancelled(context.signal)
const read = await context.provider.readFile(candidate.file.path)
throwIfAiVaultScanCancelled(context.signal)
if (read.isBinary) {
return null
}
const session = await candidate.source.parse(candidate.file, read.content, context)
// The read is inside the cached parse: an unchanged transcript must not be
// pulled off the remote disk at all, which is the whole cost of #13753.
const session = await parseRemoteSessionFileCached({
candidate,
hostKey: remoteSessionParseHostKey(context),
parse: async () => {
const read = await context.provider.readFile(candidate.file.path)
throwIfAiVaultScanCancelled(context.signal)
if (read.isBinary) {
return null
}
return await candidate.source.parse(candidate.file, read.content, context)
}
})
throwIfAiVaultScanCancelled(context.signal)
// Mirror the local rule: every session carries its sibling subagent
// transcript count (row badge; recoverable signal at zero turns). The
@@ -0,0 +1,112 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
const { mockPtySpawn, mockPtyInstance, mockCreateShellPromptReadinessProbe } = vi.hoisted(() => ({
mockPtySpawn: vi.fn(),
mockCreateShellPromptReadinessProbe: vi.fn(),
mockPtyInstance: {
pid: process.pid,
onData: vi.fn(),
onExit: vi.fn(),
write: vi.fn(),
resize: vi.fn(),
kill: vi.fn(),
clear: vi.fn(),
pause: vi.fn(),
resume: vi.fn()
}
}))
vi.mock('node-pty', () => ({
spawn: mockPtySpawn
}))
vi.mock('../main/pty/posix-pty-process-groups', () => ({
forceKillPosixPtyProcessGroups: vi.fn((_pid: number, fallback: () => void) => fallback())
}))
vi.mock('../main/shell-prompt-readiness-probe', () => ({
createShellPromptReadinessProbe: mockCreateShellPromptReadinessProbe
}))
import type { PtyHandler } from './pty-handler'
import * as ptyShellUtils from './pty-shell-utils'
import { beginPtyHandlerTest, endPtyHandlerTest, testPtyId } from './pty-handler-test-harness'
import type { MockDispatcher } from './pty-handler-test-harness'
const PTY_1 = testPtyId(1)
const STALE_PID = 424_242
/** node-pty's native `pty.resize` error when the master fd is already closed. */
function ebadfResize(): never {
throw new Error('ioctl(2) failed, EBADF')
}
describe('PtyHandler.resize against a stale PTY handle', () => {
let dispatcher: MockDispatcher
let handler: PtyHandler
let originalPlatform: PropertyDescriptor | undefined
let resize: ReturnType<typeof vi.fn>
beforeEach(async () => {
;({ dispatcher, handler, originalPlatform } = beginPtyHandlerTest({
mockPtySpawn,
mockPtyInstance,
mockCreateShellPromptReadinessProbe
}))
resize = vi.fn()
// A shell that exited without node-pty producing `onExit`: the record is
// still in the pool and undisposed, but the master fd behind it is closed.
mockPtySpawn.mockReturnValue({ ...mockPtyInstance, pid: STALE_PID, resize })
await dispatcher.callRequest('pty.spawn', {})
expect(handler.activePtyCount).toBe(1)
})
afterEach(async () => {
await endPtyHandlerTest(handler, originalPlatform)
})
it('retires an entry whose pid the host proves is gone, instead of issuing the ioctl', () => {
vi.spyOn(ptyShellUtils, 'isProcessAlive').mockReturnValue(false)
resize.mockImplementation(ebadfResize)
expect(() =>
dispatcher.callNotification('pty.resize', { id: PTY_1, cols: 120, rows: 40 })
).not.toThrow()
expect(resize).not.toHaveBeenCalled()
// The record must leave the pool: while it stays, the relay keeps
// advertising a dead shell and `activePtyCount` never reaches zero, so a
// relay configured with an unlimited grace never reaches its idle exit.
expect(handler.activePtyCount).toBe(0)
})
it('contains an ioctl failure whose process is still live, and keeps the record', () => {
vi.spyOn(ptyShellUtils, 'isProcessAlive').mockReturnValue(true)
resize.mockImplementation(ebadfResize)
const stderr = vi.spyOn(process.stderr, 'write').mockReturnValue(true)
expect(() =>
dispatcher.callNotification('pty.resize', { id: PTY_1, cols: 120, rows: 40 })
).not.toThrow()
// Repeats must stay contained too — this is the notification the client
// re-sends on every reconnect and every window resize.
expect(() =>
dispatcher.callNotification('pty.resize', { id: PTY_1, cols: 90, rows: 30 })
).not.toThrow()
// Loss of an fd is not evidence the shell exited, so the claim is retained.
expect(handler.activePtyCount).toBe(1)
expect(stderr.mock.calls.map(([line]) => String(line)).join('')).toContain(
'ioctl(2) failed, EBADF'
)
})
it('still resizes a live PTY, with the clamped geometry', () => {
vi.spyOn(ptyShellUtils, 'isProcessAlive').mockReturnValue(true)
dispatcher.callNotification('pty.resize', { id: PTY_1, cols: 4_000, rows: 40 })
expect(resize).toHaveBeenCalledWith(500, 40)
expect(handler.activePtyCount).toBe(1)
})
})
+52 -7
View File
@@ -1963,8 +1963,7 @@ export class PtyHandler {
}
// Why: verify liveness because shells can exit without node-pty onExit.
if (managed.pty.pid && !isProcessAlive(managed.pty.pid)) {
this.reapExitedPty(managed)
if (this.reapPtyProvenExited(managed)) {
throw new Error(`PTY "${id}" not found`)
}
@@ -2082,8 +2081,34 @@ export class PtyHandler {
const cols = Math.max(1, Math.min(500, Math.floor(Number(params.cols) || 80)))
const rows = Math.max(1, Math.min(500, Math.floor(Number(params.rows) || 24)))
const managed = this.ptys.get(id)
if (managed && !managed.disposed) {
if (!managed || managed.disposed) {
return
}
// Why probe first (same probe attach() and listProcesses() run): a shell
// that exited without node-pty's `onExit` leaves this entry holding a closed
// master fd, and `UnixTerminal.resize` has no fd guard — the ioctl throws
// `ioctl(2) failed, EBADF` straight out of this notification handler into the
// dispatcher's generic parse-error catch, where it is logged and nothing
// else. Nothing retired the entry, so it kept being advertised as live and
// kept holding `activePtyCount` above zero, which is what stops a relay with
// `relayGracePeriodSeconds: 0` from ever reaching its idle-no-ptys exit
// (#12423).
if (this.reapPtyProvenExited(managed)) {
return
}
try {
managed.pty.resize(cols, rows)
} catch (err) {
// A failed ioctl observed the handle, not the host's process table, so on
// its own it is `unverifiable`. Re-probe: a now-absent pid retires the
// entry, anything else keeps it and is contained here rather than
// escaping as a parse error on every later resize.
if (this.reapPtyProvenExited(managed)) {
return
}
process.stderr.write(
`[pty-handler] resize failed for PTY ${id} whose process is still live or unverifiable: ${err instanceof Error ? err.message : String(err)}\n`
)
}
}
@@ -2183,9 +2208,7 @@ export class PtyHandler {
if (this.ptys.get(managed.id) !== managed || managed.disposed) {
return
}
const pid = managed.pty.pid
if (pid && !isProcessAlive(pid)) {
this.reapExitedPty(managed)
if (this.reapPtyProvenExited(managed)) {
return
}
if (attemptsRemaining <= 0) {
@@ -2225,6 +2248,25 @@ export class PtyHandler {
this.clearPtyFlowState(managed.id)
}
/**
* Retire this entry when the host proves its pid is gone; report whether it was.
*
* `managed.disposed` is bookkeeping, not liveness: it says we tore the record
* down, not that the shell died. A shell can exit without node-pty producing
* `onExit`, which leaves a non-disposed entry holding a handle whose master fd
* is already closed. Only `isProcessAlive` (ESRCH, from the host that owns the
* process) is positive evidence of absence; every other outcome is
* `unverifiable` and keeps its record and owner claim
* (docs/reference/ssh-execution-boundary.md).
*/
private reapPtyProvenExited(managed: ManagedPty): boolean {
if (!managed.pty.pid || isProcessAlive(managed.pty.pid)) {
return false
}
this.reapExitedPty(managed)
return true
}
private async sendSignal(params: Record<string, unknown>): Promise<void> {
const id = params.id as string
const signal = params.signal as string
@@ -2422,10 +2464,13 @@ export class PtyHandler {
}
}
for (const [entryIndex, [id, managed]] of managedEntries.entries()) {
if (managed.disposed || (managed.pty.pid && !isProcessAlive(managed.pty.pid))) {
if (managed.disposed) {
this.reapExitedPty(managed)
continue
}
if (this.reapPtyProvenExited(managed)) {
continue
}
// Reuse batched correlation; per-PTY tree scans recreate O(PTY × rows) work.
const title =
(evidenceRows
+62
View File
@@ -17,6 +17,7 @@ import { resetProcessTableSnapshotForTests } from '../shared/process-table-snaps
import {
getForegroundProcessName,
isProcessAlive,
processHasChildren,
resolveDefaultCwd,
resolveWindowsDefaultShell
} from './pty-shell-utils'
@@ -525,3 +526,64 @@ describe('getForegroundProcessName', () => {
await expect(getForegroundProcessName(100)).resolves.toBe('bash')
})
})
describe('processHasChildren', () => {
// Why these assert on argv, not just the answer: the defect in #13537 was the
// cost of the answer. `pgrep -P` forks per pane per poll and opens six procfs
// files per host process to resolve one ppid, so the contract worth pinning is
// "no fork of its own, and share the foreground lookup's cached table".
const PS_TABLE = ['100 1 Ss bash', '101 100 S+ node /opt/codex', '200 1 Ss zsh'].join('\n')
it('answers from the shared process table without forking pgrep', async () => {
await withProcessPlatform('linux', async () => {
mockExecFile((_command, args) => {
if (args[0] === '-axo') {
return { stdout: PS_TABLE }
}
return new Error('unexpected command')
})
await expect(processHasChildren(100)).resolves.toBe(true)
await expect(processHasChildren(200)).resolves.toBe(false)
expect(execFileMock.mock.calls.map((call) => call[0])).not.toContain('pgrep')
})
})
it('shares one process-table capture across a burst of panes', async () => {
await withProcessPlatform('linux', async () => {
mockExecFile((_command, args) => {
if (args[0] === '-axo') {
return { stdout: PS_TABLE }
}
return new Error('unexpected command')
})
const answers = await Promise.all([
processHasChildren(100),
processHasChildren(100),
processHasChildren(200),
getForegroundProcessName(100, 'bash')
])
expect(answers).toEqual([true, true, false, 'codex'])
expect(execFileMock).toHaveBeenCalledTimes(1)
})
})
it('reports no children when the process table is unreadable', async () => {
await withProcessPlatform('linux', async () => {
mockExecFile(() => new Error('ps table unavailable'))
await expect(processHasChildren(100)).resolves.toBe(false)
})
})
it('spawns nothing on Windows, where the answer was always false', async () => {
await withProcessPlatform('win32', async () => {
await expect(processHasChildren(100)).resolves.toBe(false)
expect(execFileMock).not.toHaveBeenCalled()
})
})
})
+17 -6
View File
@@ -170,15 +170,26 @@ export async function resolveProcessCwd(pid: number, fallbackCwd: string): Promi
}
/**
* Check whether a process has child processes (via pgrep).
* Check whether a process has child processes.
*
* Why the shared snapshot and not `pgrep -P`: this answers one field of
* `pty.inspectProcess`, which every tracked pane polls on a 750ms/2000ms
* cadence, and the fork was neither cached nor coalesced. procps-ng opens six
* procfs files per process to resolve a ppid — including a `/proc/<pid>/ctty`
* that never exists on Linux — so one call cost O(host process count) syscalls,
* ~4k opens per pgrep on a 690-process host, at up to 8 forks/sec (#13537).
* `getForegroundProcessName` in the same RPC already captured the TTL-cached
* `ps` table, whose index carries the parent/child map, so the answer is free.
*/
export async function processHasChildren(pid: number): Promise<boolean> {
// Windows has no `ps`; the previous `pgrep` fork always failed here too, so
// this keeps the same answer without spawning anything to reach it.
if (process.platform === 'win32') {
return false
}
try {
const { stdout } = await execFile('pgrep', ['-P', String(pid)], {
encoding: 'utf-8',
timeout: 3000
})
return stdout.trim().length > 0
const rows = await getProcessTableSnapshot()
return (getProcessTableIndex(rows).childrenByPpid.get(pid)?.length ?? 0) > 0
} catch {
return false
}