diff --git a/docs/superpowers/plans/2026-09-05-codex-child-discovery.md b/docs/superpowers/plans/2026-09-05-codex-child-discovery.md new file mode 100644 index 00000000..6287c292 --- /dev/null +++ b/docs/superpowers/plans/2026-09-05-codex-child-discovery.md @@ -0,0 +1,50 @@ +# Bounded Codex child discovery + +Status: implemented; local validation complete; PR/CI review pending. Issue #803. Depends on #809 (serialized refreshes), +base fix/subagent-refresh-coalescing at 689b110e. Retarget to main only after +that dependency merges; this PR must not reintroduce concurrent offset owners. + +## Outcome + +Missing child rollouts cause at most one directory traversal per five-second +retry window per tracker, shared by all its unresolved children. Unrelated +parent records do not schedule filesystem work. A file created after a miss +is found within five seconds plus the existing 1.2-second poll and scan time. + +## Implementation + +- Replace per-child recursive stat walks with one Dirent traversal for the set + of unresolved child ids, skipping symlinks and checking stop during traversal. +- Keep only paths for tracked children; no archive-sized retained filename index. + Bound misses with a five-second retry deadline; new ids may wait for that same + window so spawning many children cannot bypass the bound. +- Preserve parent correlation/completion emissions. Continue byte-offset reads + of found children on the existing timer independently of discovery cooldown. +- Invalidate missing cached paths without letting one vanished child block the + rest. Changing the sessions root resets path/offset state and discovery age. + +## Validation + +Port the synthetic 1,000-file fixture into behavioral tests: many unrelated +parent entries and several missing children share one scan with zero per-file +stats; new files after a miss are eventually found; known children still tail; +symlink loops and deletion do not break progress. Re-run #802 regressions and +focused subagent tests, typecheck and test contract. Record operation counts, +not inferred production speedups. Review the diff, synchronize #803/#809, open a +complete Conventional dependent PR and complete CI/review. Do not merge. + +## Evidence and scope decision + +39 focused subagent tests pass. The 1,000-file/three-missing-child fixture uses +11 directory reads and zero file stats for discovery; 1,000 unrelated parent +records and three sequential retries add no scans in the same window. Late +creation, positive-path tailing, deletion/replacement, symlink cycles and root +changes are covered. The budget is per tracker, not a global all-parent cache: +this avoids retaining an archive-sized shared index and keeps lifetime ownership +with the tracker. Cross-parent scan sharing can be measured separately. + +Typecheck, testing contract and diff checks pass. + +Refreshed base: rebased onto main f7507980 after toolkit #812 and MCP repair +#818 merged. Their files do not overlap this implementation; rerun checks on +the refreshed PR head before requesting merge approval. diff --git a/src/main/subagents/codexChildDiscovery.system.test.ts b/src/main/subagents/codexChildDiscovery.system.test.ts new file mode 100644 index 00000000..62d848b4 --- /dev/null +++ b/src/main/subagents/codexChildDiscovery.system.test.ts @@ -0,0 +1,125 @@ +import { appendFile, mkdir, mkdtemp, rm, symlink, unlink, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it, vi } from 'vitest' +import type { SubAgentState } from '@preload/api/types.js' +import { CODEX_CHILD_DISCOVERY_RETRY_MS, CodexSubAgentTracker } from './codexSubagentState.js' + +const io = vi.hoisted(() => ({ readdir: 0, stat: 0 })) +vi.mock('node:fs/promises', async importOriginal => { + const actual = await importOriginal() + return { + ...actual, + readdir: (...args: Parameters) => { io.readdir++; return actual.readdir(...args) }, + stat: (...args: Parameters) => { io.stat++; return actual.stat(...args) }, + } +}) +const cleanups: Array<() => Promise> = [] +afterEach(async () => { + for (const cleanup of cleanups.splice(0).reverse()) await cleanup() + io.readdir = io.stat = 0 +}) + +async function fixture() { + const root = await mkdtemp(join(tmpdir(), 'codex-discovery-')) + const sessions = join(root, 'sessions') + await mkdir(sessions) + let clock = 1000 + const emitted: Record[] = [] + const tracker = new CodexSubAgentTracker(state => emitted.push(state), () => clock) + cleanups.push(async () => { tracker.stop(); await rm(root, { recursive: true, force: true }) }) + const parent = join(sessions, 'parent.jsonl') + return { + root, sessions, tracker, emitted, + advance: (ms: number) => { clock += ms }, + child(id: string) { + tracker.observeParentEntry({ type: 'response_item', payload: { type: 'function_call_output', call_id: id, output: JSON.stringify({ agent_id: id }) } }, parent) + }, + unrelated() { tracker.observeParentEntry({ type: 'event_msg', payload: { type: 'token_count' } }, parent) }, + } +} +function line(text: string) { + return JSON.stringify({ type: 'response_item', timestamp: new Date().toISOString(), payload: { type: 'message', role: 'assistant', content: [{ type: 'output_text', text }] } }) + '\n' +} + +describe('bounded missing Codex child discovery', () => { + it('shares one scan across missing children and ignores 1,000 unrelated records in the retry window', async () => { + const f = await fixture() + for (let d = 0; d < 10; d++) { + const dir = join(f.sessions, String(d)) + await mkdir(dir) + await Promise.all(Array.from({ length: 100 }, (_, i) => writeFile(join(dir, `unrelated-${i}.jsonl`), ''))) + } + f.child('missing-a'); f.child('missing-b'); f.child('missing-c') + await f.tracker.refresh() + expect(io).toEqual({ readdir: 11, stat: 0 }) + for (let i = 0; i < 1000; i++) f.unrelated() + // Explicit sequential refreshes protect against merely hiding the work + // behind #802 coalescing: a quiet missing child must keep its negative TTL. + await f.tracker.refresh(); await f.tracker.refresh(); await f.tracker.refresh() + expect(io).toEqual({ readdir: 11, stat: 0 }) + f.child('missing-d') + await f.tracker.refresh() + expect(io.readdir).toBe(11) // New ids cannot bypass the scan budget. + }) + + it('finds a child created after a miss and keeps tailing it during the discovery cooldown', async () => { + const f = await fixture() + f.child('late-child') + await f.tracker.refresh() + const file = join(f.sessions, 'rollout-late-child.jsonl') + await writeFile(file, line('first')) + f.advance(CODEX_CHILD_DISCOVERY_RETRY_MS - 1) + await f.tracker.refresh() + expect(f.emitted.at(-1)?.['late-child'].turnCount).toBe(0) + f.advance(1) + await f.tracker.refresh() + expect(f.emitted.at(-1)?.['late-child'].turnCount).toBe(1) + const scanned = io.readdir + await appendFile(file, line('second')) + await f.tracker.refresh() + expect(f.emitted.at(-1)?.['late-child'].turnCount).toBe(2) + expect(io.readdir).toBe(scanned) + }) + + it('does not follow symlink cycles or linked foreign rollouts', async () => { + const f = await fixture() + await symlink(f.sessions, join(f.sessions, 'cycle')) + const foreign = join(f.root, 'foreign.jsonl') + await writeFile(foreign, line('not this archive')) + await symlink(foreign, join(f.sessions, 'rollout-linked-child.jsonl')) + f.child('linked-child') + await f.tracker.refresh() + expect(io).toEqual({ readdir: 1, stat: 0 }) + expect(f.emitted.at(-1)?.['linked-child'].turnCount).toBe(0) + }) + + it('resets cached paths and offsets when a parent moves to another sessions root', async () => { + const f = await fixture() + await writeFile(join(f.sessions, 'rollout-child-a.jsonl'), line('old one') + line('old two')) + f.child('child-a') + await f.tracker.refresh() + expect(f.emitted.at(-1)?.['child-a'].turnCount).toBe(2) + const next = join(f.root, 'another', 'sessions') + await mkdir(next, { recursive: true }) + await writeFile(join(next, 'rollout-child-a.jsonl'), line('new archive')) + f.tracker.observeParentEntry({ type: 'event_msg', payload: { type: 'token_count' } }, join(next, 'parent.jsonl')) + await f.tracker.refresh() + expect(f.emitted.at(-1)?.['child-a'].turnCount).toBe(1) + }) + + it('keeps later children progressing after a cached file vanishes, then rediscovers it', async () => { + const f = await fixture() + const a = join(f.sessions, 'rollout-child-a.jsonl'), b = join(f.sessions, 'rollout-child-b.jsonl') + await writeFile(a, line('a')); await writeFile(b, line('b')) + f.child('child-a'); f.child('child-b') + await f.tracker.refresh() + await unlink(a); await appendFile(b, line('still running')) + await f.tracker.refresh() + expect(f.emitted.at(-1)?.['child-b'].turnCount).toBe(2) + await writeFile(join(f.sessions, 'replacement-child-a.jsonl'), line('replacement')) + f.advance(CODEX_CHILD_DISCOVERY_RETRY_MS) + await f.tracker.refresh() + expect(f.emitted.at(-1)?.['child-a'].turnCount).toBe(1) + }) +}) diff --git a/src/main/subagents/codexSubagentState.ts b/src/main/subagents/codexSubagentState.ts index a5f5c6dd..06f9e09d 100644 --- a/src/main/subagents/codexSubagentState.ts +++ b/src/main/subagents/codexSubagentState.ts @@ -511,32 +511,43 @@ function sessionsRootFromRolloutPath(path: string): string | null { } } -async function findFileContaining(root: string, needle: string): Promise { - async function walk(dir: string): Promise { - let entries: string[] +// Bound negative discovery independently of the 1.2 s activity poll. A new +// child may wait up to this interval + one poll + scan time, but parent bursts +// (including bursts of newly spawned ids) cannot defeat the scan budget. +export const CODEX_CHILD_DISCOVERY_RETRY_MS = 5000 + +async function findChildRollouts( + root: string, + missing: Set, + stopped: () => boolean, +): Promise> { + const found = new Map() + async function walk(dir: string): Promise { + if (stopped() || found.size === missing.size) return + let entries try { - entries = await readdir(dir) + entries = await readdir(dir, { withFileTypes: true }) } catch { - return null + return // A newly created/moved date directory is retried next window. } for (const entry of entries) { - const path = join(dir, entry) - let info - try { - info = await stat(path) - } catch { - continue - } - if (info.isDirectory()) { - const found = await walk(path) - if (found) return found - } else if (entry.endsWith('.jsonl') && entry.includes(needle)) { - return path + if (stopped() || found.size === missing.size) return + const path = join(dir, entry.name) + if (entry.isDirectory()) { + await walk(path) + } else if (entry.isFile() && entry.name.endsWith('.jsonl')) { + // Directory-entry types remove one stat per archived file. In + // particular, never follow symlinks: a linked ancestor can recurse + // forever, and a linked foreign archive is not this root's discovery + // authority. Retain only matches for currently tracked children. + for (const id of missing) { + if (!found.has(id) && entry.name.includes(id)) found.set(id, path) + } } } - return null } - return walk(root) + await walk(root) + return found } export class CodexSubAgentTracker { @@ -561,16 +572,32 @@ export class CodexSubAgentTracker { private dirty = false private stopped = false private readonly refreshLoop = new CoalescedRefresh(() => this.poll()) + private nextDiscoveryAt = 0 - constructor(private readonly onChange: (subAgents: Record) => void) {} + constructor( + private readonly onChange: (subAgents: Record) => void, + private readonly now: () => number = Date.now, + ) {} observeParentEntry(entry: JsonlEntry, file: string): void { if (this.stopped) return + const oldRoot = this.parentFile ? sessionsRootFromRolloutPath(this.parentFile) : null + const rootChanged = oldRoot !== sessionsRootFromRolloutPath(file) + if (rootChanged) { + // Offsets are meaningful only in their original files. A root change + // must not attach a new provider archive to cached old-path fold state. + this.childPathByAgentId.clear() + this.childOffsetByAgentId.clear() + this.childPartialByAgentId.clear() + this.childAccByAgentId.clear() + this.nextDiscoveryAt = 0 + } this.parentFile = file + let changed = rootChanged const spawn = extractCodexSpawnCall(entry) if (spawn) { this.spawnsByCallId.set(spawn.callId, spawn) - this.dirty = true + changed = true } const output = extractCodexSpawnOutput(entry) if (output) { @@ -580,7 +607,7 @@ export class CodexSubAgentTracker { // at the same moment, but emit() still needs to rebuild the record with the // parent spawn/output metadata. Mark dirty without touching the byte offset; // the accumulator remains the source of truth for child-derived fields. - this.dirty = true + changed = true } const waitCallId = isCodexWaitAgentCall(entry) if (waitCallId) this.waitCallIds.add(waitCallId) @@ -602,17 +629,20 @@ export class CodexSubAgentTracker { // here is the exact key to drop. const consumedCallId = stringField(asRecord(entry.payload), 'call_id') if (consumedCallId) this.waitCallIds.delete(consumedCallId) - this.dirty = true + changed = true } const notification = extractCodexSubagentNotification(entry) if (notification) { this.notificationsByAgentId.set(notification.agentId, notification) // Notification status is parent-rollout metadata, not child bytes. A // completion notice must repaint even when the child file is quiescent. - this.dirty = true + changed = true } if (this.knownAgentIds().length > 0) this.ensureTimer() - void this.refresh() + if (changed) { + this.dirty = true + void this.refresh() + } } private ensureTimer(): void { @@ -681,27 +711,43 @@ export class CodexSubAgentTracker { private async readKnownChildren(): Promise { const root = this.parentFile ? sessionsRootFromRolloutPath(this.parentFile) : null if (!root) return - for (const agentId of this.knownAgentIds()) { - // Per-iteration stop guard (PR #317). This loop awaits IO between every - // child; stop() can land mid-loop and clear the maps. Bail immediately so - // we never write derived state back into a tracker that has been torn down. + const ids = this.knownAgentIds() + const missing = new Set(ids.filter(id => !this.childPathByAgentId.has(id))) + if (missing.size > 0 && this.now() >= this.nextDiscoveryAt) { + // One traversal serves ALL unresolved children, including misses from a + // previous pass. No archive-sized filename cache is retained and no new + // child bypasses the cooldown. #802 supplies the single in-flight owner. + this.nextDiscoveryAt = this.now() + CODEX_CHILD_DISCOVERY_RETRY_MS + const found = await findChildRollouts(root, missing, () => this.stopped) + if (this.stopped || sessionsRootFromRolloutPath(this.parentFile!) !== root) return + for (const [id, path] of found) this.childPathByAgentId.set(id, path) + } + for (const agentId of ids) { if (this.stopped) return - let path = this.childPathByAgentId.get(agentId) ?? null - if (!path) { - path = await findFileContaining(root, agentId) + const path = this.childPathByAgentId.get(agentId) + if (!path) continue + try { + const changed = await this.readAppendedChild(agentId, path) + if (this.stopped) return + if (changed) this.dirty = true + } catch (error) { if (this.stopped) return - if (path) this.childPathByAgentId.set(agentId, path) + // A removed rollout must not poison every later child's activity + // poll. Rediscover it in the next bounded window, with fresh offsets. + const code = (error as NodeJS.ErrnoException).code + if ((code === 'ENOENT' || code === 'ENOTDIR') && this.childPathByAgentId.get(agentId) === path) { + this.childPathByAgentId.delete(agentId) + this.childOffsetByAgentId.delete(agentId) + this.childPartialByAgentId.delete(agentId) + this.childAccByAgentId.delete(agentId) + } } - if (!path) continue - const changed = await this.readAppendedChild(agentId, path) - if (this.stopped) return - if (changed) this.dirty = true } } private async readAppendedChild(agentId: string, path: string): Promise { const { size } = await stat(path) - if (this.stopped) return false + if (this.stopped || this.childPathByAgentId.get(agentId) !== path) return false let from = this.childOffsetByAgentId.get(agentId) ?? 0 if (size < from) { // Rollouts are append-only in normal Codex operation, but editors/tests can @@ -717,7 +763,7 @@ export class CodexSubAgentTracker { const appended = await readRange(path, from, size) // readKnownChildren's outer guard is too late: these maps must not be // repopulated after stop(), even if an open range read finishes afterwards. - if (this.stopped) return false + if (this.stopped || this.childPathByAgentId.get(agentId) !== path) return false const text = (this.childPartialByAgentId.get(agentId) ?? '') + appended.text const lastNl = text.lastIndexOf('\n') this.childOffsetByAgentId.set(agentId, appended.nextOffset) diff --git a/src/main/subagents/refreshCoalescing.system.test.ts b/src/main/subagents/refreshCoalescing.system.test.ts index 3ecf9766..8103bcf9 100644 --- a/src/main/subagents/refreshCoalescing.system.test.ts +++ b/src/main/subagents/refreshCoalescing.system.test.ts @@ -20,7 +20,7 @@ vi.mock('node:fs/promises', async importOriginal => { const sync = await import('node:fs') return { ...original, - readdir: async (path: string) => sync.readdirSync(path), + readdir: async (path: string, options?: { withFileTypes: true }) => options ? sync.readdirSync(path, options) : sync.readdirSync(path), stat: async (path: string) => sync.statSync(path), readFile: async (path: string, encoding: BufferEncoding) => sync.readFileSync(path, encoding), }