diff --git a/docs/superpowers/plans/2026-09-05-subagent-refresh-coalescing.md b/docs/superpowers/plans/2026-09-05-subagent-refresh-coalescing.md new file mode 100644 index 00000000..32d59e3e --- /dev/null +++ b/docs/superpowers/plans/2026-09-05-subagent-refresh-coalescing.md @@ -0,0 +1,51 @@ +# Subagent refresh coalescing + +Status: implemented; PR/CI review pending. Issue #802. Base: origin/main at 5d641845. + +## Outcome and boundaries + +Each child transcript byte range is folded once despite parent-record bursts, +timers, and manual refreshes. A trigger during I/O schedules a trailing pass; +late appends and parent completion are preserved. Stopping a tracker prevents +in-flight I/O from repopulating state or emitting. No provider process, renderer, +worktree reconciliation, or external control changes belong in this PR. + +## Implementation + +1. Add a small tracker-local serialized refresh owner, coalescing pending + requests into one trailing pass and containing background I/O failures. +2. Route Claude and Codex refresh triggers through it. Check stopped state + after awaited reads before committing offsets, metadata, or accumulators. +3. Keep transcript UTF-8/partial-line and parent completion semantics intact. +4. Port the audit burst reproduction into deterministic deferred-read tests for + both trackers, including appended bytes, completion during I/O, transient + errors, and stop during I/O. Keep fixtures synthetic and private-data-free. + +## Validation and review + +Run focused subagent tests, testing contract and typecheck. Inspect before/after +behavior with the same synthetic burst (one child record, 50 parent triggers); +report duplicate fold and read counts, not production CPU/FPS claims. Review the +final diff, open a complete Conventional PR with Fixes #802, synchronize the +issue, and address applicable CI/review feedback. Never merge without explicit +user confirmation. + +## Following work + +Issue #803 will use a separate branch/PR based on this branch because both touch +Codex refresh ownership. Remote #804 starts independently from main; remote #805 +will explicitly depend on that resynchronization contract. Complete all four. + +## Implementation evidence + +- Shared CoalescedRefresh now owns each tracker’s poll loop; post-await stop + guards prevent offset/accumulator revival. No discovery policy changes here. +- 34 focused tests pass (including 10 new lifecycle/concurrency cases). The + baseline fails both provider burst cases: up to 52 overlapping range reads + versus one after coalescing. A late append causes exactly two sequential reads. +- Typecheck and testing contract passed; final incremental typecheck and CI + review complete the validation. Evidence is synthetic, not production speedup. + +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/CoalescedRefresh.ts b/src/main/subagents/CoalescedRefresh.ts new file mode 100644 index 00000000..0da6ac38 --- /dev/null +++ b/src/main/subagents/CoalescedRefresh.ts @@ -0,0 +1,49 @@ +/** One byte-offset owner per tracker, even when parent events arrive in bursts. + * + * A boolean `busy` guard that simply drops requests loses appends/completion + * arriving during I/O. Queueing every request instead re-scans once per parent + * record. The pending bit means exactly one trailing pass for any such burst; + * requests during that pass can ask for another pass without concurrent reads. + */ +export class CoalescedRefresh { + private pending = false + private stopped = false + private inflight: Promise | null = null + + constructor(private readonly poll: () => Promise) {} + + request(): Promise { + if (this.stopped) return Promise.resolve() + this.pending = true + if (!this.inflight) { + // Defer entry until inflight is assigned, including a synchronous throw. + // Clear ownership INSIDE this continuation: clearing it in .finally() + // leaves a microtask gap in which a request can join an already finished + // drain and strand its pending bit until the next periodic timer. + this.inflight = Promise.resolve().then(async () => { + try { + while (this.pending && !this.stopped) { + this.pending = false + try { + await this.poll() + } catch { + // Polling observes externally created/rotated files. Failure is + // retryable on the next requested pass, never an unhandled + // rejection from a fire-and-forget parent event or timer. + } + } + } finally { + this.inflight = null + } + }) + } + return this.inflight + } + + stop(): void { + this.stopped = true + this.pending = false + // The poller still owns its open I/O. Its post-await stop guards prevent + // writes/emissions; retaining the promise lets existing callers drain it. + } +} diff --git a/src/main/subagents/SubAgentWatcher.ts b/src/main/subagents/SubAgentWatcher.ts index b391958c..de9b1d7f 100644 --- a/src/main/subagents/SubAgentWatcher.ts +++ b/src/main/subagents/SubAgentWatcher.ts @@ -9,6 +9,7 @@ import { } from './subagentState.js' import type { SubAgentAccumulator, SubAgentMeta } from './subagentState.js' import { readRange } from './shared.js' +import { CoalescedRefresh } from './CoalescedRefresh.js' // One poller per session, watching /subagents/. // @@ -76,6 +77,7 @@ export class SubAgentWatcher { private prunedAgents = new Set() private dirty = false private stopped = false + private readonly refreshLoop = new CoalescedRefresh(() => this.tick()) constructor( private readonly subagentsDir: string, @@ -84,15 +86,17 @@ export class SubAgentWatcher { ) {} start(): void { + if (this.stopped || this.timer) return // Kick once immediately so an already-populated dir surfaces fast, then // poll. The first tick also covers the common "dir created moments later" // case — readdir simply throws and we retry next tick. - void this.tick() - this.timer = setInterval(() => void this.tick(), POLL_MS) + void this.refreshLoop.request() + this.timer = setInterval(() => void this.refreshLoop.request(), POLL_MS) } stop(): void { this.stopped = true + this.refreshLoop.stop() if (this.timer) clearInterval(this.timer) this.timer = null this.offsets.clear() @@ -106,9 +110,10 @@ export class SubAgentWatcher { /** Force a re-emit (e.g. the parent transcript just produced a tool_result * that flips a subagent running→done). */ - refresh(): void { + refresh(): Promise { + if (this.stopped) return Promise.resolve() this.dirty = true - void this.tick() + return this.refreshLoop.request() } private async tick(): Promise { @@ -120,7 +125,7 @@ export class SubAgentWatcher { // tick. Any other transient FS error is also safe to retry. return } - if (this.dirty) { + if (!this.stopped && this.dirty) { this.dirty = false this.emit() } @@ -129,12 +134,14 @@ export class SubAgentWatcher { private async rescan(): Promise { const files = await readdir(this.subagentsDir) for (const f of files) { + if (this.stopped) return if (f.endsWith('.meta.json')) { const agentId = f.slice('agent-'.length, -'.meta.json'.length) if (this.prunedAgents.has(agentId)) continue // finding-16: reclaimed if (this.metaByAgent.has(agentId)) continue // meta is written once try { const raw = await readFile(join(this.subagentsDir, f), 'utf8') + if (this.stopped) return this.metaByAgent.set(agentId, JSON.parse(raw) as SubAgentMeta) this.dirty = true } catch { @@ -153,6 +160,7 @@ export class SubAgentWatcher { private async readAppended(agentId: string, path: string): Promise { const { size } = await stat(path) + if (this.stopped) return const from = this.offsets.get(agentId) ?? 0 if (size <= from) return @@ -166,6 +174,9 @@ export class SubAgentWatcher { // Reading only `[from, size)` keeps the watcher proportional to new bytes, // which is the actual invariant future code should preserve. const appended = await readRange(path, from, size) + // stop() clears all fold state while this read may still be in flight. + // Never repopulate an offset/accumulator or emit for a dead session. + if (this.stopped) return const text = (this.partialByAgent.get(agentId) ?? '') + appended.text const lastNl = text.lastIndexOf('\n') if (lastNl < 0) { diff --git a/src/main/subagents/codexSubagentState.ts b/src/main/subagents/codexSubagentState.ts index acc6d721..a5f5c6dd 100644 --- a/src/main/subagents/codexSubagentState.ts +++ b/src/main/subagents/codexSubagentState.ts @@ -2,6 +2,7 @@ import { basename, dirname, join } from 'node:path' import { readdir, stat } from 'node:fs/promises' import type { JsonlEntry, SubAgentState, SubAgentToolCall } from '@preload/api/types.js' import { asRecord } from '@shared/lib/asRecord.js' +import { CoalescedRefresh } from './CoalescedRefresh.js' import { capToolCalls, headlineFromInput, @@ -559,10 +560,12 @@ export class CodexSubAgentTracker { private parentFile: string | null = null private dirty = false private stopped = false + private readonly refreshLoop = new CoalescedRefresh(() => this.poll()) constructor(private readonly onChange: (subAgents: Record) => void) {} observeParentEntry(entry: JsonlEntry, file: string): void { + if (this.stopped) return this.parentFile = file const spawn = extractCodexSpawnCall(entry) if (spawn) { @@ -625,6 +628,7 @@ export class CodexSubAgentTracker { stop(): void { this.stopped = true + this.refreshLoop.stop() if (this.timer) clearInterval(this.timer) this.timer = null // Mirror SubAgentWatcher.stop() (PR #300): clearing the timer alone left @@ -645,7 +649,11 @@ export class CodexSubAgentTracker { this.childAccByAgentId.clear() } - async refresh(): Promise { + refresh(): Promise { + return this.refreshLoop.request() + } + + private async poll(): Promise { if (this.stopped || !this.parentFile) return await this.readKnownChildren() // POST-AWAIT stop guard (PR #317, race fix). The pre-await check above only @@ -693,6 +701,7 @@ export class CodexSubAgentTracker { private async readAppendedChild(agentId: string, path: string): Promise { const { size } = await stat(path) + if (this.stopped) return false let from = this.childOffsetByAgentId.get(agentId) ?? 0 if (size < from) { // Rollouts are append-only in normal Codex operation, but editors/tests can @@ -706,6 +715,9 @@ export class CodexSubAgentTracker { if (size <= from) return false 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 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 new file mode 100644 index 00000000..3ecf9766 --- /dev/null +++ b/src/main/subagents/refreshCoalescing.system.test.ts @@ -0,0 +1,194 @@ +import { appendFile, mkdir, mkdtemp, rm, 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 { JsonlEntry, SubAgentState } from '@preload/api/types.js' +import { SubAgentWatcher } from './SubAgentWatcher.js' +import { CodexSubAgentTracker } from './codexSubagentState.js' + +const reads = vi.hoisted(() => ({ + count: 0, + active: 0, + maxActive: 0, + afterRead: null as null | (() => Promise), +})) +// Resolve metadata operations through already-settled promises so every +// competing refresh reaches readRange without relying on filesystem timing. +// The fixtures and byte-range reader remain real; only scheduling is controlled. +vi.mock('node:fs/promises', async importOriginal => { + const original = await importOriginal() + const sync = await import('node:fs') + return { + ...original, + readdir: async (path: string) => sync.readdirSync(path), + stat: async (path: string) => sync.statSync(path), + readFile: async (path: string, encoding: BufferEncoding) => sync.readFileSync(path, encoding), + } +}) + +vi.mock('./shared.js', async importOriginal => { + const original = await importOriginal() + return { + ...original, + readRange: async (...args: Parameters) => { + reads.count++ + reads.maxActive = Math.max(reads.maxActive, ++reads.active) + const hook = reads.afterRead + reads.afterRead = null + try { + const range = await original.readRange(...args) + // Hold the old bytes AFTER the real range read. An append here must + // be picked up by a trailing pass, not smuggled into the first read. + await hook?.() + return range + } finally { + reads.active-- + } + }, + } +}) + +function deferred() { + let resolve!: () => void + const promise = new Promise(r => { resolve = r }) + return { promise, resolve } +} + +const cleanups: Array<() => Promise> = [] +afterEach(async () => { + for (const cleanup of cleanups.splice(0).reverse()) await cleanup() + Object.assign(reads, { count: 0, active: 0, maxActive: 0, afterRead: null }) +}) + +async function fixture(provider: 'claude' | 'codex') { + const root = await mkdtemp(join(tmpdir(), 'subagent-refresh-')) + const dir = join(root, 'sessions') + await mkdir(dir) + const file = join(dir, provider === 'claude' ? 'agent-child.jsonl' : 'rollout-child.jsonl') + const emitted: Record[] = [] + let done = false + const tracker = provider === 'claude' + ? new SubAgentWatcher(dir, () => ({ done, error: false }), value => emitted.push(value)) + : new CodexSubAgentTracker(value => emitted.push(value)) + cleanups.push(async () => { tracker.stop(); await rm(root, { recursive: true, force: true }) }) + const parent = (entry: JsonlEntry) => { + if (tracker instanceof CodexSubAgentTracker) tracker.observeParentEntry(entry, join(dir, 'parent.jsonl')) + } + await writeFile(join(dir, 'agent-child.meta.json'), JSON.stringify({ toolUseId: 'spawn' })) + const line = (id: number) => JSON.stringify(provider === 'claude' + ? { type: 'assistant', timestamp: new Date().toISOString(), message: { role: 'assistant', content: [{ type: 'tool_use', id: `tool-${id}`, name: 'Read', input: { file_path: `/synthetic-${id}` } }] } } + : { type: 'response_item', timestamp: new Date().toISOString(), payload: { type: 'message', role: 'assistant', content: [{ type: 'output_text', text: `synthetic-${id}` }] } }) + '\n' + await writeFile(file, line(1)) + return { + emitted, tracker, file, line, + start() { + if (tracker instanceof SubAgentWatcher) tracker.start() + else parent({ type: 'response_item', payload: { type: 'function_call_output', call_id: 'spawn', output: JSON.stringify({ agent_id: 'child' }) } }) + }, + burst() { + for (let i = 0; i < 50; i++) { + if (tracker instanceof SubAgentWatcher) void tracker.refresh() + else parent({ type: 'event_msg', payload: { type: 'token_count' } }) + } + }, + complete() { + done = true + if (tracker instanceof SubAgentWatcher) void tracker.refresh() + else parent({ type: 'response_item', payload: { type: 'message', role: 'user', content: [{ type: 'input_text', text: '\n{"agent_path":"child","status":"completed"}\n' }] } }) + }, + } +} + +for (const provider of ['claude', 'codex'] as const) { + describe(`${provider} serialized child refresh`, () => { + it('folds one record once across 50 refresh triggers during the range read', async () => { + const f = await fixture(provider) + const entered = deferred(), release = deferred() + reads.afterRead = async () => { entered.resolve(); await release.promise } + f.start() + await entered.promise + f.burst() + const drained = f.tracker.refresh() + // The legacy paths need at most readdir/meta/stat continuations before + // opening a duplicate range. Drain those microtasks while the first + // range is held; this deterministically exposes duplicate ownership. + for (let i = 0; i < 20; i++) await Promise.resolve() + release.resolve() + await drained + await vi.waitFor(() => { + expect(reads.active).toBe(0) + expect(f.emitted.length).toBeGreaterThan(0) + }) + expect(reads.maxActive).toBe(1) + expect(reads.count).toBe(1) + expect(f.emitted.at(-1)?.spawn.turnCount).toBe(1) + if (provider === 'claude') expect(f.emitted.at(-1)?.spawn.toolCalls).toHaveLength(1) + }) + + it('drains appended bytes and parent completion requested during I/O', async () => { + const f = await fixture(provider) + const entered = deferred(), release = deferred() + reads.afterRead = async () => { entered.resolve(); await release.promise } + f.start() + await entered.promise + await appendFile(f.file, f.line(2)) + f.burst() + f.complete() + const drained = f.tracker.refresh() + release.resolve() + await drained + expect(reads.maxActive).toBe(1) + expect(reads.count).toBe(2) + expect(f.emitted.at(-1)?.spawn).toMatchObject({ turnCount: 2, status: 'done' }) + }) + + it('cannot repopulate stopped state or emit after its blocked read finishes', async () => { + const f = await fixture(provider) + const entered = deferred(), release = deferred() + reads.afterRead = async () => { entered.resolve(); await release.promise } + f.start() + await entered.promise + const drained = f.tracker.refresh() + f.tracker.stop() + release.resolve() + await drained + f.burst() + await f.tracker.refresh() + expect(f.emitted).toEqual([]) + expect(reads.count).toBe(1) + // Silence alone is not sufficient: the old outer stop guard still let + // the reader pin an accumulator after stop had released all state. + const retained = Reflect.get(f.tracker, provider === 'claude' ? 'accByAgent' : 'childAccByAgentId') as Map + expect(retained.size).toBe(0) + }) + + it('keeps a split UTF-8 JSONL record unread until its remaining bytes arrive', async () => { + const f = await fixture(provider) + f.start() + await f.tracker.refresh() + const bytes = Buffer.from(f.line(2).replace('synthetic-2', 'synthetic-🌱')) + const split = bytes.indexOf(Buffer.from('🌱')) + 2 + await appendFile(f.file, bytes.subarray(0, split)) + await f.tracker.refresh() + expect(f.emitted.at(-1)?.spawn.turnCount).toBe(1) + await appendFile(f.file, bytes.subarray(split)) + f.burst() + await f.tracker.refresh() + expect(f.emitted.at(-1)?.spawn.turnCount).toBe(2) + if (provider === 'claude') expect(f.emitted.at(-1)?.spawn.toolCalls.at(-1)?.headline).toBe('/synthetic-🌱') + }) + + it('retries a failed read on a requested trailing pass', async () => { + const f = await fixture(provider) + const entered = deferred(), release = deferred() + reads.afterRead = async () => { entered.resolve(); await release.promise; throw new Error('synthetic transient read failure') } + f.start() + await entered.promise + const drained = f.tracker.refresh() + release.resolve() + await drained + expect(f.emitted.at(-1)?.spawn.turnCount).toBe(1) + expect(reads.count).toBe(2) + }) + }) +}