export interface ObservedToolResult { readonly toolCallId: string; readonly toolName: string; readonly input: Record; readonly content: readonly unknown[]; readonly details: unknown; readonly isError: boolean; readonly usage?: unknown } export interface WaveCall { readonly id: string; readonly toolName: string; readonly args: unknown; readonly sourceIndex: number; readonly potentiallyMutating: boolean; terminal: boolean; result?: ObservedToolResult } export interface CompletedWave { readonly id: string; readonly turnIndex: number; readonly calls: readonly WaveCall[]; readonly carrierId?: string; readonly potentiallyMutating: boolean } export type WavePatches = ReadonlyMap; interface ActiveWave { id: string; turnIndex: number; calls: WaveCall[]; finalized: boolean; timer?: NodeJS.Timeout; promise: Promise; resolve(value: WavePatches): void } const READ_ONLY = new Set(["read", "grep", "find", "ls", "pi_hooks_waive_verification"]); export class ExecutionWaveCoordinator { private wave?: ActiveWave; private serial = 0; private turnIndex = 0; private stopped = false; constructor(private readonly finalize: (wave: CompletedWave) => Promise, private readonly onUncertain: (message: string) => void, private readonly timeoutMs = 30_000) {} turnStarted(turnIndex: number): void { this.turnIndex = turnIndex; } start(toolCallId: string, toolName: string, args: unknown): { waveId: string; isNew: boolean; potentiallyMutating: boolean } | undefined { if (this.stopped) return undefined; const isNew = !this.wave || this.wave.calls.every((call) => call.terminal); if (isNew) this.wave = this.createWave(); const wave = this.wave; if (!wave) return undefined; if (wave.calls.some((call) => call.id === toolCallId)) return { waveId: wave.id, isNew: false, potentiallyMutating: !READ_ONLY.has(toolName) }; const potentiallyMutating = !READ_ONLY.has(toolName); wave.calls.push({ id: toolCallId, toolName, args, sourceIndex: wave.calls.length, potentiallyMutating, terminal: false }); return { waveId: wave.id, isNew, potentiallyMutating }; } async result(result: ObservedToolResult): Promise<{ readonly content?: readonly unknown[]; readonly details?: unknown; readonly isError?: boolean; readonly usage?: unknown } | undefined> { const wave = this.wave; if (!wave || this.stopped) return undefined; const call = wave.calls.find((item) => item.id === result.toolCallId); if (!call) { this.onUncertain(`tool_result without start: ${result.toolCallId}`); return undefined; } if (!call.result) call.result = result; call.terminal = true; this.maybeFinalize(wave); return (await wave.promise).get(result.toolCallId); } end(toolCallId: string): void { const wave = this.wave; if (!wave || this.stopped) return; const call = wave.calls.find((item) => item.id === toolCallId); if (!call) { this.onUncertain(`tool_execution_end without start: ${toolCallId}`); return; } if (!call.terminal) call.terminal = true; this.maybeFinalize(wave); } turnEnded(): void { if (this.wave && !this.wave.finalized && this.wave.calls.some((call) => !call.terminal)) this.releaseUncertain(this.wave, "turn ended with unterminated tool calls"); } shutdown(): void { this.stopped = true; if (this.wave && !this.wave.finalized) this.releaseUncertain(this.wave, "session shutdown"); } private createWave(): ActiveWave { let resolve!: (value: WavePatches) => void; const promise = new Promise((done) => { resolve = done; }); const wave: ActiveWave = { id: `wave-${++this.serial}`, turnIndex: this.turnIndex, calls: [], finalized: false, promise, resolve }; wave.timer = setTimeout(() => this.releaseUncertain(wave, "execution-wave barrier timeout"), this.timeoutMs); wave.timer.unref?.(); return wave; } private maybeFinalize(wave: ActiveWave): void { if (wave.finalized || wave.calls.length === 0 || wave.calls.some((call) => !call.terminal)) return; wave.finalized = true; if (wave.timer) clearTimeout(wave.timer); const mutatingResults = wave.calls.filter((call) => call.potentiallyMutating && call.result); const participating = wave.calls.filter((call) => call.result); const carrier = (mutatingResults.at(-1) ?? participating.at(-1))?.id; const completed = Object.freeze({ id: wave.id, turnIndex: wave.turnIndex, calls: Object.freeze([...wave.calls]), ...(carrier ? { carrierId: carrier } : {}), potentiallyMutating: wave.calls.some((call) => call.potentiallyMutating) }); this.finalize(completed).then(wave.resolve, (error) => { this.onUncertain(`wave finalization failed: ${error instanceof Error ? error.message : String(error)}`); wave.resolve(new Map()); }); } private releaseUncertain(wave: ActiveWave, message: string): void { if (wave.finalized) return; wave.finalized = true; if (wave.timer) clearTimeout(wave.timer); this.onUncertain(message); wave.resolve(new Map()); } }