diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index 25fdd1e35..b3a4ae3ac 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -950,6 +950,14 @@ type AutonomousRuntimeSnapshot = Pick< "continuationsUsed" | "gateAttempts" | "lastGateFailure" | "lastGateFailureSnapshot" >; +type RlmChildUpdateEvent = Extract; + +interface PendingRlmChildUpdate { + event: RlmChildUpdateEvent; + /** C01 assignment fence rechecked both at enqueue and publication. */ + isCurrent: () => boolean; +} + interface RlmChildRun { id: string; /** UUID attempt fence, minted before this run is visible to any host. */ @@ -1120,6 +1128,12 @@ export class AgentSession { followUps: [], }; private _agentEventQueue: Promise = Promise.resolve(); + // Presentation-only, per-parent coalescing. This never gates child admission or model work. + private _rlmChildUpdateFlushTimer: ReturnType | undefined; + private _rlmChildUpdateGeneration = 0; + private readonly _pendingRlmChildUpdates = new Map(); + private _observerFailureDiagnostics = 0; + private _afterToolHookFailureDiagnostics = 0; /** Session-owned actions. Items are never fed into Agent.steer/followUp. */ private readonly _actionStore = new ActionStore(); @@ -1475,25 +1489,31 @@ export class AgentSession { return undefined; } - const hookResult = await runner.emitToolResult({ - type: "tool_result", - toolName: toolCall.name, - toolCallId: toolCall.id, - input: args as Record, - content: result.content, - details: result.details, - isError, - }); + try { + const hookResult = await runner.emitToolResult({ + type: "tool_result", + toolName: toolCall.name, + toolCallId: toolCall.id, + input: args as Record, + content: result.content, + details: result.details, + isError, + }); - if (!hookResult) { + if (!hookResult) { + return undefined; + } + + return { + content: hookResult.content, + details: hookResult.details, + isError: hookResult.isError ?? isError, + }; + } catch (error) { + // Result hooks observe completed work; unlike beforeToolCall they are not a veto. + this._recordBoundedDiagnostic("after-tool-hook", error); return undefined; } - - return { - content: hookResult.content, - details: hookResult.details, - isError: hookResult.isError ?? isError, - }; }; } @@ -1510,18 +1530,95 @@ export class AgentSession { // Event Subscription // ========================================================================= + private _recordBoundedDiagnostic(kind: "observer" | "after-tool-hook", error: unknown): void { + const count = kind === "observer" ? ++this._observerFailureDiagnostics : ++this._afterToolHookFailureDiagnostics; + // Diagnostics deliberately exclude event, hook, tool, prompt, result, and stack content. + if (count <= 10) { + const errorClass = error instanceof Error ? error.constructor.name : typeof error; + console.warn(`AgentSession ${kind} failure (${errorClass})`); + } + } + /** Emit an event to all listeners */ private _emit(event: AgentSessionEvent): void { + const terminalChildUpdate = + event.type === "rlm_child_update" && + (event.child.status === "done" || event.child.status === "error" || event.child.status === "cancelled"); + if ((event.type !== "rlm_child_update" || terminalChildUpdate) && this._pendingRlmChildUpdates.size > 0) { + // Structural and terminal events cannot overtake a retained progress snapshot. + this._flushRlmChildUpdates(); + } for (const l of this._eventListeners) { try { l(event); - } catch { + } catch (error) { // A failing observer must not prevent other subscribers from // receiving lifecycle and persistence events. + this._recordBoundedDiagnostic("observer", error); } } } + /** + * Retain only the newest nonterminal snapshot per child until the next turn. + * The caller supplies C01's assignment fence so a replaced incarnation cannot + * be published from a stale timeout. + */ + private _queueRlmChildUpdate( + event: RlmChildUpdateEvent, + isCurrent: () => boolean, + publishSynchronously: boolean, + ): void { + if (this._disposed || !isCurrent()) return; + const { child } = event; + const terminal = child.status === "done" || child.status === "error" || child.status === "cancelled"; + if (terminal) { + // A terminal must not overtake the final retained activity for this child. + this._flushRlmChildUpdates(); + if (!this._disposed && isCurrent()) this._emit(event); + return; + } + if (publishSynchronously) { + // A status edge replaces any older activity for its own child, then + // preserves ordering with activity retained for other children. + this._pendingRlmChildUpdates.delete(child.id); + this._flushRlmChildUpdates(); + if (!this._disposed && isCurrent()) this._emit(event); + return; + } + + this._pendingRlmChildUpdates.set(child.id, { event, isCurrent }); + if (this._rlmChildUpdateFlushTimer !== undefined) return; + const generation = this._rlmChildUpdateGeneration; + this._rlmChildUpdateFlushTimer = setTimeout(() => { + if (generation !== this._rlmChildUpdateGeneration) return; + this._rlmChildUpdateFlushTimer = undefined; + if (this._disposed) return; + this._flushRlmChildUpdates(); + }, 0); + } + + private _flushRlmChildUpdates(): void { + if (this._rlmChildUpdateFlushTimer !== undefined) { + clearTimeout(this._rlmChildUpdateFlushTimer); + this._rlmChildUpdateFlushTimer = undefined; + } + const pending = [...this._pendingRlmChildUpdates.values()]; + this._pendingRlmChildUpdates.clear(); + for (const { event, isCurrent } of pending) { + if (!this._disposed && isCurrent()) this._emit(event); + } + } + + private _cancelRlmChildUpdateFlush(): void { + this._rlmChildUpdateGeneration++; + if (this._rlmChildUpdateFlushTimer !== undefined) { + clearTimeout(this._rlmChildUpdateFlushTimer); + this._rlmChildUpdateFlushTimer = undefined; + } + this._pendingRlmChildUpdates.clear(); + } + private _emitQueueUpdate(): void { const actions = this.getSessionActionSnapshot(); if (JSON.stringify(actions) === JSON.stringify(this._lastSessionActionSnapshot)) return; @@ -4000,6 +4097,7 @@ export class AgentSession { if (this._disposed) { return; } + this._cancelRlmChildUpdateFlush(); this._disposed = true; this._sessionActionCommitDisposeAbortController.abort(); try { @@ -6555,6 +6653,7 @@ export class AgentSession { const compactionOperation = this._compactionOperation; const branchSummaryOperation = this._branchSummaryOperation; this.requestAbort(); + this._cancelRlmChildUpdateFlush(); this._cancelActiveRlmChildRuns("Parent session aborted"); this._goalAbortInProgress = this._goalState.status === "active"; try { @@ -6577,6 +6676,7 @@ export class AgentSession { this._sessionInputPumpSuspended = true; this._cancelPostCompactionContinue(); this.abortRetry(); + this._cancelRlmChildUpdateFlush(); this._cancelActiveRlmChildRuns("Parent session aborted for update restart"); this._goalAbortInProgress = this._goalState.status === "active"; this.agent.abort(); @@ -9798,16 +9898,20 @@ export class AgentSession { if (run.status === "cancelled") throw new Error(run.error ?? "RLM child cancelled"); }; this._activeRlmChildRuns.set(run.id, run); - const emitChildUpdate = () => { - const activeOwner = - this._activeRlmChildRuns.get(run.id) === run && - this._activeRlmChildRuns.get(run.id)?.assignmentId === run.assignmentId; - const retainedOwner = - this._rlmChildSessions.get(run.id) === childSession && - this._rlmChildSessionAssignments.get(run.id) === run.assignmentId; - if (!activeOwner && !retainedOwner) return; + let runningPublished = false; + const emitChildUpdate = (structural = false) => { + const isCurrent = () => { + const activeOwner = + this._activeRlmChildRuns.get(run.id) === run && + this._activeRlmChildRuns.get(run.id)?.assignmentId === run.assignmentId; + const retainedOwner = + this._rlmChildSessions.get(run.id) === childSession && + this._rlmChildSessionAssignments.get(run.id) === run.assignmentId; + return activeOwner || retainedOwner; + }; + if (!isCurrent()) return; const childModel = childSession?.model ?? modelSelection.model; - this._emit({ + const event: RlmChildUpdateEvent = { type: "rlm_child_update", child: { id: childNodeId, @@ -9826,7 +9930,12 @@ export class AgentSession { repliedSinceTask: childSession?._repliedToParentSinceTask, error: run.error, }, - }); + }; + const terminal = run.status === "done" || run.status === "error" || run.status === "cancelled"; + const publishSynchronously = + structural || run.status === "queued" || (run.status === "running" && !runningPublished) || terminal; + this._queueRlmChildUpdate(event, isCurrent, publishSynchronously); + if (run.status === "running") runningPublished = true; }; run.emitUpdate = emitChildUpdate; emitChildUpdate(); @@ -9963,7 +10072,11 @@ export class AgentSession { runningToolCount = Math.max(0, runningToolCount - 1); if (runningToolCount === 0) activity = { kind: "waiting" }; emitChildUpdate(); - } else if (event.type === "session_info_changed" || event.type === "recap_update") { + } else if (event.type === "session_info_changed") { + // A child rename changes its public identity. Unlike recap/progress activity, + // it must be visible before another macrotask can replace the snapshot. + emitChildUpdate(true); + } else if (event.type === "recap_update") { emitChildUpdate(); } }); diff --git a/packages/coding-agent/src/modes/daemon/active-session-state.ts b/packages/coding-agent/src/modes/daemon/active-session-state.ts index fe062253c..e2c27e74a 100644 --- a/packages/coding-agent/src/modes/daemon/active-session-state.ts +++ b/packages/coding-agent/src/modes/daemon/active-session-state.ts @@ -15,6 +15,10 @@ export interface DaemonSocketClient { catchupPurposes?: Map; /** The single catch-up drain currently serving this client. */ catchupPromise?: Promise; + /** An attachment-local catch-up callback is queued for the next event-loop turn. */ + catchupDrainScheduled?: boolean; + /** Invalidates queued attachment-local catch-up callbacks during cleanup. */ + catchupGeneration?: number; /** Delayed retry after transient catch-up snapshot preparation failure. */ catchupRetryTimer?: NodeJS.Timeout; backpressured?: boolean; diff --git a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts index 4b2cd349b..ac0855b86 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-supervisor.ts @@ -661,6 +661,8 @@ export class DaemonSupervisor { private updateRestartPhase?: "draining" | "fencing" | "prepared"; private readonly mutationDrain = new MutationDrainLatch(); private readonly clients = new Set(); + /** Handles are private supervisor state: the wire/client descriptor remains unchanged. */ + private readonly catchupDrainTimers = new WeakMap(); private readonly protocolClientIds = new WeakMap(); private readonly workers = new Map(); private readonly openingWorkers = new Map>(); @@ -1316,6 +1318,7 @@ export class DaemonSupervisor { return; } cleaned = true; + this.cancelClientCatchup(client); client.detachInput(); this.clients.delete(client); this.cancelWaitingPromptAdmissionsForClient(client); @@ -1330,9 +1333,7 @@ export class DaemonSupervisor { socket.on("drain", () => { client.backpressured = false; if (!client.snapshotStreaming) { - void this.catchUpClient(client).catch((error) => - this.log(`Failed to catch up client ${client.id}: ${String(error)}`), - ); + this.scheduleClientCatchup(client); } }); } @@ -4459,9 +4460,7 @@ export class DaemonSupervisor { client.backpressured = false; } if (!client.snapshotStreaming && client.catchupActiveSessionIds?.size) { - void this.catchUpClient(client).catch((error) => - this.log(`Failed to catch up client ${client.id}: ${String(error)}`), - ); + this.scheduleClientCatchup(client); } }; } @@ -4803,9 +4802,7 @@ export class DaemonSupervisor { for (const client of this.clients) { if (!client.attachedActiveSessionIds.has(activeSessionId)) continue; this.queueCatchup(client, activeSessionId, snapshotPurpose === "replacement" ? "replacement" : "resync"); - void this.catchUpClient(client).catch((error) => - this.log(`Failed to catch up client ${client.id}: ${String(error)}`), - ); + this.scheduleClientCatchup(client); } } return; @@ -4842,9 +4839,7 @@ export class DaemonSupervisor { activeSessionId, snapshotPurpose === "replacement" ? "replacement" : "resync", ); - void this.catchUpClient(client).catch((error) => - this.log(`Failed to catch up client ${client.id}: ${String(error)}`), - ); + this.scheduleClientCatchup(client); } } } catch (error) { @@ -4979,12 +4974,65 @@ export class DaemonSupervisor { const clients = [...this.clients].filter((client) => client.attachedActiveSessionIds.has(activeSessionId)); for (const client of clients) { this.queueCatchup(client, activeSessionId); + this.scheduleClientCatchup(client); + } + // The per-client scheduler owns the deferred work. Do not leave a second, + // session-global callback alive solely to release this synchronous dedupe latch. + this.compactCatchupInProgress.delete(activeSessionId); + } + + /** + * Coalesce recovery triggers for a single socket attachment. This deliberately + * owns no provider/model work: it only starts the existing attachment catch-up + * latch after this event-loop turn. + */ + private scheduleClientCatchup(client: DaemonSocketClient): void { + if ( + client.catchupDrainScheduled || + client.socket.destroyed || + !this.clients.has(client) || + client.snapshotStreaming || + client.backpressured + ) { + return; } - void Promise.all(clients.map((client) => this.catchUpClient(client))) - .catch((error) => this.log(`Failed compact catch-up for ${activeSessionId}: ${String(error)}`)) - .finally(() => { - this.compactCatchupInProgress.delete(activeSessionId); + client.catchupDrainScheduled = true; + const generation = client.catchupGeneration ?? 0; + const timer = setImmediate(() => { + this.catchupDrainTimers.delete(client); + client.catchupDrainScheduled = false; + if ( + generation !== (client.catchupGeneration ?? 0) || + client.socket.destroyed || + !this.clients.has(client) || + client.snapshotStreaming || + client.backpressured + ) { + return; + } + void this.catchUpClient(client).catch((error) => { + const errorClass = error instanceof Error ? error.constructor.name : typeof error; + this.log(`Failed attachment catch-up for client ${client.id.slice(0, 64)} (${errorClass})`); }); + }); + this.catchupDrainTimers.set(client, timer); + } + + /** Make queued attachment-local work inert without affecting snapshot ownership. */ + private cancelClientCatchup(client: DaemonSocketClient): void { + client.catchupGeneration = (client.catchupGeneration ?? 0) + 1; + const timer = this.catchupDrainTimers.get(client); + if (timer) { + clearImmediate(timer); + this.catchupDrainTimers.delete(client); + } + client.catchupDrainScheduled = false; + client.catchupActiveSessionIds?.clear(); + client.catchupPurposes?.clear(); + if (client.catchupRetryTimer) { + clearTimeout(client.catchupRetryTimer); + client.catchupRetryTimer = undefined; + } } private queueCatchup( @@ -5661,6 +5709,7 @@ export class DaemonSupervisor { } }); for (const client of this.clients) { + this.cancelClientCatchup(client); client.attachedActiveSessionIds.clear(); await this.runCleanupStep(`daemon client input ${client.id}`, () => client.detachInput()); await this.runCleanupStep(`daemon client socket ${client.id}`, () => { @@ -5780,6 +5829,7 @@ export class DaemonSupervisor { } await this.catalog.stop(); for (const client of this.clients) { + this.cancelClientCatchup(client); client.detachInput(); client.socket.end(); } diff --git a/packages/coding-agent/src/modes/interactive/interactive-mode.ts b/packages/coding-agent/src/modes/interactive/interactive-mode.ts index 0ec1d8bcd..9b53b4b13 100644 --- a/packages/coding-agent/src/modes/interactive/interactive-mode.ts +++ b/packages/coding-agent/src/modes/interactive/interactive-mode.ts @@ -529,6 +529,8 @@ const DEAD_TERMINAL_ERROR_CODES = new Set(["EIO", "EPIPE", "ENOTCONN"]); // inline limit before storing, so this holds many recent pastes; the oldest are // evicted past the cap to keep a long session bounded. const MAX_PASTED_IMAGE_BYTES = 64 * 1024 * 1024; +/** Bound UI-only coalescing memory without limiting agent or provider work. */ +const MAX_PENDING_PROGRESS_EVENTS = 128; const INITIAL_TRANSCRIPT_RENDER_MESSAGE_LIMIT = 400; function initialRenderMessages(messages: AgentMessage[]): AgentMessage[] { @@ -928,6 +930,14 @@ export class InteractiveMode { // Serializes session event handling; see subscribeToAgent private sessionEventQueue: Promise = Promise.resolve(); private sessionEventGeneration = 0; + // Streaming progress is replaceable per UI entity: retain each entity's newest + // update until the next turn, but always put retained work back on the normal + // event tail before a structural transition. + private progressFlushTimer?: ReturnType; + private progressFlushGeneration = 0; + private pendingProgressEvents = new Map(); + private progressFlushStopped = false; + private readonly progressChildStatuses = new Map(); private fastModeToggleQueue: Promise = Promise.resolve(); // Tool execution tracking: toolCallId -> component @@ -5030,20 +5040,166 @@ export class InteractiveMode { }; } + private isReplaceableProgressEvent(event: AgentConnectionSessionEvent): boolean { + if (event.type === "message_update") return event.message?.role === "assistant"; + if (event.type === "tool_execution_update") return true; + if (event.type !== "rlm_child_update") return false; + + const { id, status } = event.child; + const previous = this.progressChildStatuses.get(id); + if (status !== "queued" && status !== "running") { + this.progressChildStatuses.delete(id); + return false; + } + this.progressChildStatuses.set(id, status); + // The first observation and queued/running edge affect the child summary; + // preserve them, and coalesce only later snapshots in the same state. + return previous === status; + } + + /** Return a stable key for each independently replaceable progress entity. */ + private getProgressEventKey(event: AgentConnectionSessionEvent): string | undefined { + if (event.type === "message_update" && event.message.role === "assistant") { + // responseId is the only message identity exposed by AssistantMessage. A + // session can have only one active assistant stream when it is absent. + return `assistant:${event.message.responseId ?? "active"}`; + } + if (event.type === "tool_execution_update") return `tool:${event.toolCallId}`; + if (event.type === "rlm_child_update") return `rlm-child:${event.child.id}`; + return undefined; + } + + /** Add work to the one UI-owned event tail, and keep that tail usable after errors. */ + private enqueueSessionEvent( + event: AgentConnectionSessionEvent, + generation: number, + options: { preserveAcrossReplacement?: boolean } = {}, + ): Promise { + const run = this.sessionEventQueue.then(() => { + if ( + this.progressFlushStopped || + (!options.preserveAcrossReplacement && generation !== this.sessionEventGeneration) + ) { + return; + } + return this.handleEvent(event); + }); + this.sessionEventQueue = run.catch(() => {}); + return run; + } + + private reportProgressFlushError(error: unknown): void { + // Keep the error surface consistent with the subscription's outer catch. + this.showError(error instanceof Error ? error.message : String(error)); + } + + private pendingProgress(): Map { + // Prototype-based focused harnesses deliberately only supply the state they + // exercise. Lazily initializing here also keeps this UI-only queue resilient. + if (!this.pendingProgressEvents) this.pendingProgressEvents = new Map(); + return this.pendingProgressEvents; + } + + private drainPendingProgressEvents(): AgentConnectionSessionEvent[] { + const pending = [...this.pendingProgress().values()]; + this.pendingProgress().clear(); + return pending; + } + + private enqueueFlushedProgress( + events: readonly AgentConnectionSessionEvent[], + generation: number, + options: { preserveAcrossReplacement?: boolean } = {}, + ): void { + for (const event of events) { + void this.enqueueSessionEvent(event, generation, options).catch((error) => + this.reportProgressFlushError(error), + ); + } + } + + /** + * Apply replaceable progress updates on a later turn. This is UI work only; + * it neither delays nor limits agent/provider work. + */ + private queueProgressEvent(event: AgentConnectionSessionEvent, sessionGeneration: number): void { + if (this.progressFlushStopped || sessionGeneration !== this.sessionEventGeneration) return; + const key = this.getProgressEventKey(event); + if (!key) return; + + const pending = this.pendingProgress(); + // A replacement is a later event, so move it to the tail to preserve event + // order among the newest retained snapshots. + if (pending.has(key)) { + pending.delete(key); + } else if (pending.size >= MAX_PENDING_PROGRESS_EVENTS) { + // Do not evict UI entities. Put this complete ordered batch onto the + // existing UI tail now, then retain the new entity for its scheduled flush. + // The batch was received before any later replacement, so it must retain + // that ordering even when the replacement advances the generation before + // this queued UI work gets a turn. This bounds only local coalescing + // memory, never provider/agent work. + this.enqueueFlushedProgress(this.drainPendingProgressEvents(), sessionGeneration, { + preserveAcrossReplacement: true, + }); + } + pending.set(key, event); + if (this.progressFlushTimer) return; + + const progressGeneration = this.progressFlushGeneration; + this.progressFlushTimer = setTimeout(() => { + this.progressFlushTimer = undefined; + if ( + this.progressFlushStopped || + progressGeneration !== this.progressFlushGeneration || + sessionGeneration !== this.sessionEventGeneration + ) { + return; + } + this.enqueueFlushedProgress(this.drainPendingProgressEvents(), sessionGeneration); + }, 0); + } + + /** Put retained progress ahead of the next structural event on the existing tail. */ + private flushPendingProgress(): void { + if (this.progressFlushTimer) { + clearTimeout(this.progressFlushTimer); + this.progressFlushTimer = undefined; + } + if (!this.progressFlushStopped) { + // A replacement advances sessionEventGeneration synchronously. These events + // were received by the old UI and are deliberately ordered before it. + this.enqueueFlushedProgress(this.drainPendingProgressEvents(), this.sessionEventGeneration, { + preserveAcrossReplacement: true, + }); + } + } + + /** Invalidate timers and retained updates before a replacement or UI teardown. */ + private cancelPendingProgress(): void { + this.progressFlushGeneration++; + if (this.progressFlushTimer) { + clearTimeout(this.progressFlushTimer); + this.progressFlushTimer = undefined; + } + this.pendingProgress()?.clear(); + this.progressChildStatuses?.clear(); + } + private subscribeToAgent(): void { this.unsubscribe = this.agentConnection.subscribe(async (event) => { try { if (event.type === "session_event") { - // Connection adapters dispatch without awaiting, so serialize events. - // Replacement advances the generation before entering this queue, which - // prevents already-queued source events from mutating the target UI. const generation = this.sessionEventGeneration; - const run = this.sessionEventQueue.then(() => - generation === this.sessionEventGeneration ? this.handleEvent(event.event) : undefined, - ); - this.sessionEventQueue = run.catch(() => {}); - await run; + if (this.isReplaceableProgressEvent(event.event)) { + this.queueProgressEvent(event.event, generation); + } else { + this.flushPendingProgress(); + await this.enqueueSessionEvent(event.event, generation); + } } else if (event.type === "session_replaced") { + this.flushPendingProgress(); + this.cancelPendingProgress(); const generation = ++this.sessionEventGeneration; const run = this.sessionEventQueue.then(async () => { if (generation !== this.sessionEventGeneration) return; @@ -5058,6 +5214,8 @@ export class InteractiveMode { this.sessionEventQueue = run.catch(() => {}); await run; } else if (event.type === "session_resynced") { + this.flushPendingProgress(); + this.cancelPendingProgress(); const generation = this.sessionEventGeneration; const run = this.sessionEventQueue.then(async () => { if (generation !== this.sessionEventGeneration) return false; @@ -9700,6 +9858,10 @@ ${interrupt ? `| \`${interrupt}\` | Interrupt current operation |\n` : ""}${shor } stop(options: { preserveAltScreen?: boolean } = {}): void { + // Cancel before detaching the connection or stopping the TUI: a delayed + // progress callback must never repaint or mutate a stopped UI. + this.progressFlushStopped = true; + this.cancelPendingProgress(); this.unregisterSignalHandlers(); this.clearCtrlCExitHint({ render: false }); this.clearEscapeRepeat(); diff --git a/packages/coding-agent/test/agent-session-recursion.test.ts b/packages/coding-agent/test/agent-session-recursion.test.ts index a31d3bcdc..b134db287 100644 --- a/packages/coding-agent/test/agent-session-recursion.test.ts +++ b/packages/coding-agent/test/agent-session-recursion.test.ts @@ -1697,7 +1697,7 @@ describe("AgentSession rlm recursion", () => { } }); - it("emits updated child session names after a retained child is renamed", async () => { + it("emits a retained child rename before the next macrotask", async () => { const root = createSession(); const events: unknown[] = []; root.subscribe((event) => events.push(event)); @@ -1707,18 +1707,32 @@ describe("AgentSession rlm recursion", () => { throw new Error("Missing child session directory"); } const childId = basename(result.session_dir); + await waitFor( + () => root.getRlmChildRunStatus(childId) === undefined && root.getRlmChildSession(childId) !== undefined, + ); const child = root.getRlmChildSession(childId); if (!child) { throw new Error("Missing retained child session"); } + // Keep replaceable activity pending when the retained child is renamed. The + // rename itself is structural and must be published synchronously, rather + // than waiting for (or being overwritten by) the next macrotask. + child.setCurrentRecap("pending replaceable activity"); child.setSessionName("renamed-worker"); - const childUpdates = events.filter( - (event): event is { type: "rlm_child_update"; child: { sessionName?: string } } => - typeof event === "object" && event !== null && (event as { type?: string }).type === "rlm_child_update", - ); - expect(childUpdates.at(-1)?.child.sessionName).toBe("renamed-worker"); + const childUpdates = () => + events.filter( + (event): event is { type: "rlm_child_update"; child: { sessionName?: string; recap?: string } } => + typeof event === "object" && event !== null && (event as { type?: string }).type === "rlm_child_update", + ); + expect(childUpdates().at(-1)?.child.sessionName).toBe("renamed-worker"); + const updatesAtRename = childUpdates().length; + await sleep(0); + expect(childUpdates()).toHaveLength(updatesAtRename); + expect(childUpdates().at(-1)).toMatchObject({ + child: { sessionName: "renamed-worker", recap: "pending replaceable activity" }, + }); }); it("surfaces a child's recap on its snapshot once the summarizer sets it", async () => { diff --git a/packages/coding-agent/test/agent-session-runtime-events.test.ts b/packages/coding-agent/test/agent-session-runtime-events.test.ts index 37ce0bcfd..a7f0d09b6 100644 --- a/packages/coding-agent/test/agent-session-runtime-events.test.ts +++ b/packages/coding-agent/test/agent-session-runtime-events.test.ts @@ -1,7 +1,7 @@ import { existsSync, mkdirSync, readdirSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { fauxAssistantMessage, registerFauxProvider } from "@earendil-works/pi-ai"; +import { type FauxResponseStep, fauxAssistantMessage, registerFauxProvider } from "@earendil-works/pi-ai"; import { afterEach, describe, expect, it, vi } from "vitest"; import { type CreateAgentSessionRuntimeFactory, @@ -119,3 +119,310 @@ describe("AgentSessionRuntime session lifecycle events", () => { expect(readdirSync(leaseRoot).filter((entry) => entry.endsWith(".lock"))).toHaveLength(1); }); }); + +describe("AgentSession RLM child update ownership", () => { + const cleanups: Array<() => Promise | void> = []; + + afterEach(async () => { + vi.useRealTimers(); + while (cleanups.length > 0) await cleanups.pop()?.(); + }); + + async function createSession(responses: FauxResponseStep[] = [fauxAssistantMessage("child response")]) { + const tempDir = join(tmpdir(), `pi-c02-runtime-events-${Date.now()}-${Math.random().toString(36).slice(2)}`); + mkdirSync(tempDir, { recursive: true }); + const faux = registerFauxProvider(); + faux.setResponses(responses); + const authStorage = AuthStorage.inMemory(); + authStorage.setRuntimeApiKey(faux.getModel().provider, "faux-key"); + const services = await createAgentSessionServices({ + agentDir: tempDir, + authStorage, + cwd: tempDir, + resourceLoaderOptions: { noSkills: true, noPromptTemplates: true, noThemes: true }, + }); + const session = await createAgentSessionFromServices({ + services, + sessionManager: SessionManager.create(tempDir, join(tempDir, "sessions")), + model: faux.getModel(), + rlmDepth: 0, + rlmMaxDepth: 2, + }); + cleanups.push(async () => { + session.session.dispose(); + faux.unregister(); + rmSync(tempDir, { recursive: true, force: true }); + }); + return session.session as any; + } + + async function waitFor(condition: () => boolean, description: string): Promise { + const deadline = Date.now() + 2_000; + while (!condition()) { + if (Date.now() >= deadline) throw new Error(`Timed out waiting for ${description}`); + await new Promise((resolve) => setTimeout(resolve, 1)); + } + } + + function afterMacrotask(): Promise { + return new Promise((resolve) => setTimeout(resolve, 0)); + } + + function update(id: string, status: "queued" | "running" | "done" | "error" | "cancelled", preview?: string) { + return { + type: "rlm_child_update" as const, + child: { id, label: `child-${id}`, status, sessionDir: `/tmp/${id}`, answerPreview: preview }, + }; + } + + it("coalesces 64 latest activity snapshots, retains status edges, and clears zero-delay work on dispose", async () => { + vi.useFakeTimers(); + const session = await createSession(); + const observed: Array<{ id: string; status: string; preview?: string }> = []; + session.subscribe((event: any) => { + if (event.type === "rlm_child_update") + observed.push({ id: event.child.id, status: event.child.status, preview: event.child.answerPreview }); + }); + for (let i = 0; i < 64; i++) session._queueRlmChildUpdate(update(String(i), "queued"), () => true, true); + for (let i = 0; i < 64; i++) session._queueRlmChildUpdate(update(String(i), "running"), () => true, true); + for (let i = 0; i < 64; i++) { + const id = String(i); + session._queueRlmChildUpdate(update(id, "running", "x".repeat(160)), () => true, false); + session._queueRlmChildUpdate(update(id, "running", `latest-${id}`), () => true, false); + } + expect(session._pendingRlmChildUpdates.size).toBe(64); + expect(vi.getTimerCount()).toBe(1); + await vi.advanceTimersByTimeAsync(0); + expect(session._pendingRlmChildUpdates.size).toBe(0); + expect(vi.getTimerCount()).toBe(0); + for (let i = 0; i < 64; i++) { + const child = observed.filter((entry) => entry.id === String(i)); + expect(child.map((entry) => entry.status)).toEqual(["queued", "running", "running"]); + expect(child.at(-1)?.preview).toBe(`latest-${i}`); + } + session._queueRlmChildUpdate(update("teardown", "running"), () => true, false); + expect(vi.getTimerCount()).toBe(1); + session.dispose(); + expect(session._pendingRlmChildUpdates.size).toBe(0); + expect(vi.getTimerCount()).toBe(0); + await vi.advanceTimersByTimeAsync(0); + expect(observed.some((entry) => entry.id === "teardown")).toBe(false); + }); + + it("rechecks the C01 assignment fence at enqueue and flush, so stale A cannot publish over B", async () => { + vi.useFakeTimers(); + const session = await createSession(); + const observed: string[] = []; + session.subscribe((event: any) => { + if (event.type === "rlm_child_update") observed.push(event.child.answerPreview); + }); + let aCurrent = true; + let bCurrent = true; + session._queueRlmChildUpdate(update("same-id", "running", "A"), () => aCurrent, false); + aCurrent = false; + session._queueRlmChildUpdate(update("same-id", "running", "B"), () => bCurrent, false); + await vi.advanceTimersByTimeAsync(0); + expect(observed).toEqual(["B"]); + bCurrent = false; + session._queueRlmChildUpdate(update("same-id", "running", "stale-B"), () => bCurrent, false); + await vi.advanceTimersByTimeAsync(0); + expect(observed).toEqual(["B"]); + }); + + it("uses the public child lifecycle for synchronous first-running, coalesced activity, and ordering", async () => { + let releaseAnswer!: () => void; + const answerGate = new Promise((resolve) => { + releaseAnswer = resolve; + }); + const session = await createSession([ + async () => { + await answerGate; + return fauxAssistantMessage("child completed"); + }, + ]); + let releaseRuntime!: () => void; + const runtimeGate = new Promise((resolve) => { + releaseRuntime = resolve; + }); + const createInlineRuntime = session._createInlineRlmSubagentRuntime.bind(session); + session.setSubagentRuntimeHost({ + assignmentIdentityFenced: true, + createRlmSubagentRuntime: async (options: any) => { + await runtimeGate; + return createInlineRuntime(options); + }, + deleteRlmSubagentRuntime: async (_id: string, child: any) => child?.disposeAsync(), + }); + const observed: Array<{ type: string; status?: string; recap?: string }> = []; + session.subscribe((event: any) => { + observed.push({ type: event.type, status: event.child?.status, recap: event.child?.recap }); + }); + + const spawned = await session.runRlmChild("hold for lifecycle assertions"); + // This C01 host gate makes the public admission boundary deterministic: + // queued is synchronous, then the first running edge is emitted immediately + // when the real runtime is published, before activity can be coalesced. + expect(observed.map((event) => event.status).filter(Boolean)).toEqual(["queued"]); + releaseRuntime(); + await waitFor( + () => observed.filter((event) => event.status === "running").length === 1, + "the first running child update", + ); + const child = session.getRlmChildSession(spawned.rlm_child_id); + expect(child).toBeDefined(); + + child?.setCurrentRecap("first active snapshot"); + child?.setCurrentRecap("latest active snapshot"); + expect(observed.filter((event) => event.status === "running")).toHaveLength(1); + await afterMacrotask(); + expect(observed.filter((event) => event.status === "running")).toHaveLength(2); + expect(observed.at(-1)).toMatchObject({ status: "running", recap: "latest active snapshot" }); + + child?.setCurrentRecap("before structural transition"); + session.setSessionName("structural transition"); + const structuralIndex = observed.findIndex((event) => event.type === "session_info_changed"); + expect(observed[structuralIndex - 1]).toMatchObject({ + status: "running", + recap: "before structural transition", + }); + + child?.setCurrentRecap("before terminal transition"); + releaseAnswer(); + await waitFor(() => observed.some((event) => event.status === "done"), "the terminal child update"); + const terminalIndex = observed.findIndex((event) => event.status === "done"); + const pendingBeforeTerminal = observed.findIndex( + (event) => event.status === "running" && event.recap === "before terminal transition", + ); + expect(pendingBeforeTerminal).toBeGreaterThanOrEqual(0); + expect(pendingBeforeTerminal).toBeLessThan(terminalIndex); + expect(observed.slice(terminalIndex + 1).some((event) => event.status === "running")).toBe(false); + }); + + it("uses the C01 assignment ownership fence when a real A child is replaced by B", async () => { + let releaseAnswer!: () => void; + const answerGate = new Promise((resolve) => { + releaseAnswer = resolve; + }); + const session = await createSession([ + async () => { + await answerGate; + return fauxAssistantMessage("A completed"); + }, + ]); + const observed: Array<{ status: string; recap?: string }> = []; + session.subscribe((event: any) => { + if (event.type === "rlm_child_update") observed.push({ status: event.child.status, recap: event.child.recap }); + }); + const spawned = await session.runRlmChild("A must not publish after replacement"); + await waitFor(() => observed.some((event) => event.status === "running"), "A running"); + const childA = session.getRlmChildSession(spawned.rlm_child_id); + expect(childA).toBeDefined(); + + // This is the C01-owned selector replacement state: B has the same public + // selector but a new immutable assignment and session. Trigger A through its + // public child event path, rather than calling the queue helper directly. + const childB = await createSession(); + const internals = session as any; + internals._activeRlmChildRuns.delete(spawned.rlm_child_id); + expect(session.registerRlmChildSession(spawned.rlm_child_id, childB, undefined, "assignment-B")).toBe(true); + const beforeAActivity = observed.length; + childA?.setCurrentRecap("stale A activity"); + await afterMacrotask(); + expect(observed).toHaveLength(beforeAActivity); + expect(internals._rlmChildSessions.get(spawned.rlm_child_id)).toBe(childB); + expect(internals._rlmChildSessionAssignments.get(spawned.rlm_child_id)).toBe("assignment-B"); + releaseAnswer(); + }); + + it("cancels real pending child activity on abort, update restart, and dispose", async () => { + for (const teardown of [ + { name: "abort", run: async (session: any) => session.abort() }, + { name: "update restart", run: async (session: any) => session.abortForUpdateRestart() }, + { name: "dispose", run: async (session: any) => session.dispose() }, + ]) { + let releaseAnswer!: () => void; + const answerGate = new Promise((resolve) => { + releaseAnswer = resolve; + }); + const session = await createSession([ + async () => { + await answerGate; + return fauxAssistantMessage("unreachable"); + }, + ]); + const observed: Array<{ status: string; recap?: string }> = []; + session.subscribe((event: any) => { + if (event.type === "rlm_child_update") + observed.push({ status: event.child.status, recap: event.child.recap }); + }); + const spawned = await session.runRlmChild(`pending ${teardown.name}`); + await waitFor(() => observed.some((event) => event.status === "running"), `${teardown.name} running`); + session.getRlmChildSession(spawned.rlm_child_id)?.setCurrentRecap(`stale ${teardown.name}`); + expect(session._pendingRlmChildUpdates.size).toBe(1); + await teardown.run(session); + await afterMacrotask(); + expect(session._pendingRlmChildUpdates.size, teardown.name).toBe(0); + expect(session._rlmChildUpdateFlushTimer, teardown.name).toBeUndefined(); + expect(observed.some((event) => event.status === "running" && event.recap === `stale ${teardown.name}`)).toBe( + false, + ); + releaseAnswer(); + } + }); + + it("bounds a real child assistant preview at 160 characters", async () => { + const session = await createSession([fauxAssistantMessage("x".repeat(200))]); + const previews: string[] = []; + session.subscribe((event: any) => { + if (event.type === "rlm_child_update" && event.child.answerPreview) previews.push(event.child.answerPreview); + }); + await session.runRlmChild("produce a long preview"); + await waitFor(() => previews.some((preview) => preview.endsWith("...")), "a compacted child preview"); + const preview = previews.find((entry) => entry.endsWith("...")); + expect(preview).toHaveLength(160); + expect(preview).toBe(`${"x".repeat(157)}...`); + }); + + it("isolates a throwing afterToolCall hook while preserving beforeToolCall vetoes", async () => { + const session = await createSession(); + const runner = session._extensionRunner; + vi.spyOn(runner as any, "hasHandlers").mockImplementation( + (...args: unknown[]) => args[0] === "tool_result" || args[0] === "tool_call", + ); + vi.spyOn(runner, "emitToolResult").mockRejectedValue(new Error("tool output must not leak")); + vi.spyOn(runner, "emitToolCall").mockRejectedValue(new Error("veto")); + const warning = vi.spyOn(console, "warn").mockImplementation(() => undefined); + const result = { content: [{ type: "text", text: "original" }], details: undefined }; + await expect( + session.agent.afterToolCall?.({ + toolCall: { id: "id", name: "tool", arguments: {} }, + args: {}, + result, + isError: false, + }), + ).resolves.toBeUndefined(); + expect(session._afterToolHookFailureDiagnostics).toBe(1); + expect(warning.mock.calls.join(" ")).not.toContain("tool output must not leak"); + await expect( + session.agent.beforeToolCall?.({ toolCall: { id: "id", name: "tool", arguments: {} }, args: {} }), + ).rejects.toThrow("veto"); + warning.mockRestore(); + }); + + it("isolates throwing observers without leaking content-bearing diagnostics", async () => { + const session = await createSession(); + const warning = vi.spyOn(console, "warn").mockImplementation(() => undefined); + const received: string[] = []; + session.subscribe(() => { + throw new Error("secret prompt must not be logged"); + }); + session.subscribe((event: any) => { + if (event.type === "rlm_child_update") received.push(event.child.id); + }); + session._queueRlmChildUpdate(update("terminal", "done"), () => true, true); + expect(received).toEqual(["terminal"]); + expect(session._observerFailureDiagnostics).toBe(1); + expect(warning.mock.calls.join(" ")).not.toContain("secret prompt"); + warning.mockRestore(); + }); +}); diff --git a/packages/coding-agent/test/daemon-socket.test.ts b/packages/coding-agent/test/daemon-socket.test.ts index e1c8e220d..4c12bb46d 100644 --- a/packages/coding-agent/test/daemon-socket.test.ts +++ b/packages/coding-agent/test/daemon-socket.test.ts @@ -1,16 +1,19 @@ import { spawn } from "node:child_process"; import { existsSync, mkdirSync, mkdtempSync, rmSync, unlinkSync } from "node:fs"; -import { createConnection, createServer } from "node:net"; +import { createConnection, createServer, type Socket } from "node:net"; import { tmpdir } from "node:os"; import { basename, dirname, join } from "node:path"; +import { PassThrough } from "node:stream"; import lockfile from "proper-lockfile"; -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; +import type { DaemonSocketClient } from "../src/modes/daemon/active-session-state.js"; import { cleanupDaemonSocketPath, defaultDaemonSocketPath, getDaemonSocketIdentity, prepareDaemonSocketPath, } from "../src/modes/daemon/daemon-socket.js"; +import { DaemonSupervisor } from "../src/modes/daemon/daemon-supervisor.js"; describe("defaultDaemonSocketPath", () => { it("uses a fixed Windows named pipe path", () => { @@ -246,3 +249,80 @@ describe("defaultDaemonSocketPath", () => { } }); }); + +function attachmentClient(id: string): DaemonSocketClient { + return { + id, + socket: new PassThrough() as unknown as Socket, + attachedActiveSessionIds: new Set(["active-c02"]), + catchupActiveSessionIds: new Set(), + detachInput: () => {}, + supportsExtensionUi: false, + capabilities: new Set(), + }; +} + +describe("attachment-local catch-up scheduling", () => { + it("coalesces drain and cache triggers while preserving replacement precedence", async () => { + const supervisor = new DaemonSupervisor("/tmp/c02-catchup.sock", { + defaultSessionConfig: { agentDir: "/tmp", cwd: "/tmp" }, + descriptorDir: "/tmp/c02-catchup-state", + }); + const client = attachmentClient("slow-c02"); + const catchUpClient = vi.fn(async () => {}); + const internals = supervisor as unknown as { + clients: Set; + queueCatchup(client: DaemonSocketClient, activeSessionId: string, purpose?: "replacement" | "resync"): void; + scheduleClientCatchup(client: DaemonSocketClient): void; + catchUpClient: typeof catchUpClient; + }; + internals.clients.add(client); + internals.catchUpClient = catchUpClient; + + internals.queueCatchup(client, "active-c02", "resync"); + internals.scheduleClientCatchup(client); + internals.queueCatchup(client, "active-c02", "replacement"); + internals.scheduleClientCatchup(client); + await new Promise((resolve) => setImmediate(resolve)); + + expect(catchUpClient).toHaveBeenCalledTimes(1); + expect(client.catchupPurposes?.get("active-c02")).toBe("replacement"); + expect(client.catchupDrainScheduled).toBe(false); + client.socket.destroy(); + }); + + it("makes a queued callback inert on close/cleanup without affecting a healthy peer", async () => { + const supervisor = new DaemonSupervisor("/tmp/c02-close.sock", { + defaultSessionConfig: { agentDir: "/tmp", cwd: "/tmp" }, + descriptorDir: "/tmp/c02-close-state", + }); + const closedClient = attachmentClient("closed-c02"); + const healthyClient = attachmentClient("healthy-c02"); + const catchUpClient = vi.fn(async () => {}); + const internals = supervisor as unknown as { + clients: Set; + queueCatchup(client: DaemonSocketClient, activeSessionId: string): void; + scheduleClientCatchup(client: DaemonSocketClient): void; + cancelClientCatchup(client: DaemonSocketClient): void; + catchUpClient: typeof catchUpClient; + }; + internals.clients.add(closedClient); + internals.clients.add(healthyClient); + internals.catchUpClient = catchUpClient; + internals.queueCatchup(closedClient, "active-c02"); + internals.scheduleClientCatchup(closedClient); + internals.queueCatchup(healthyClient, "active-c02"); + internals.scheduleClientCatchup(healthyClient); + + internals.cancelClientCatchup(closedClient); + closedClient.socket.destroy(); + await new Promise((resolve) => setImmediate(resolve)); + + expect(catchUpClient).toHaveBeenCalledTimes(1); + expect(catchUpClient).toHaveBeenCalledWith(healthyClient); + expect(closedClient.catchupDrainScheduled).toBe(false); + expect(closedClient.catchupActiveSessionIds?.size).toBe(0); + expect(closedClient.catchupPurposes?.size).toBe(0); + healthyClient.socket.destroy(); + }); +}); diff --git a/packages/coding-agent/test/interactive-mode-status.test.ts b/packages/coding-agent/test/interactive-mode-status.test.ts index 7098e2f08..7d67c5831 100644 --- a/packages/coding-agent/test/interactive-mode-status.test.ts +++ b/packages/coding-agent/test/interactive-mode-status.test.ts @@ -1102,6 +1102,7 @@ describe("InteractiveMode pending bash components", () => { }, } as unknown as InteractiveMode; + Object.setPrototypeOf(fakeThis, InteractiveMode.prototype); (InteractiveMode.prototype as unknown as { subscribeToAgent(this: unknown): void }).subscribeToAgent.call( fakeThis, ); @@ -1307,6 +1308,7 @@ describe("InteractiveMode connection events", () => { showError: vi.fn(), }; + Object.setPrototypeOf(fakeThis, InteractiveMode.prototype); (InteractiveMode.prototype as unknown as { subscribeToAgent(this: typeof fakeThis): void }).subscribeToAgent.call( fakeThis, ); @@ -1356,6 +1358,7 @@ describe("InteractiveMode connection events", () => { showError: vi.fn(), }; + Object.setPrototypeOf(fakeThis, InteractiveMode.prototype); (InteractiveMode.prototype as unknown as { subscribeToAgent(this: typeof fakeThis): void }).subscribeToAgent.call( fakeThis, ); @@ -1405,6 +1408,7 @@ describe("InteractiveMode connection events", () => { handleConnectionExtensionUiRequest: vi.fn(), showError: vi.fn(), }; + Object.setPrototypeOf(fakeThis, InteractiveMode.prototype); (InteractiveMode.prototype as unknown as { subscribeToAgent(this: typeof fakeThis): void }).subscribeToAgent.call( fakeThis, ); @@ -1587,6 +1591,7 @@ describe("InteractiveMode connection events", () => { ui: { requestRender: vi.fn() }, showError: vi.fn(), }; + Object.setPrototypeOf(fakeThis, InteractiveMode.prototype); (InteractiveMode.prototype as unknown as { subscribeToAgent(this: typeof fakeThis): void }).subscribeToAgent.call( fakeThis, ); @@ -1599,6 +1604,411 @@ describe("InteractiveMode connection events", () => { expect(fakeThis.handleEvent).not.toHaveBeenCalled(); expect(fakeThis.renderInitialMessages).toHaveBeenCalledOnce(); }); + + test("coalesces active progress and flushes its latest state before a terminal event", async () => { + vi.useFakeTimers(); + try { + type Event = { type: "session_event"; event: AgentConnectionSessionEvent }; + let listener: ((event: Event) => Promise) | undefined; + const handled: string[] = []; + const fakeThis = { + agentConnection: { + subscribe: (callback: (event: Event) => Promise) => { + listener = callback; + return vi.fn(); + }, + }, + sessionEventQueue: Promise.resolve(), + sessionEventGeneration: 0, + progressFlushGeneration: 0, + progressFlushStopped: false, + handleEvent: vi.fn(async (event: { type: string; partialResult?: { content: unknown[] } }) => { + handled.push( + event.type === "tool_execution_update" ? String(event.partialResult?.content[0]) : event.type, + ); + }), + showError: vi.fn(), + }; + Object.setPrototypeOf(fakeThis, InteractiveMode.prototype); + ( + InteractiveMode.prototype as unknown as { subscribeToAgent(this: typeof fakeThis): void } + ).subscribeToAgent.call(fakeThis); + + const progress = (value: string) => + listener?.({ + type: "session_event", + event: { + type: "tool_execution_update", + toolCallId: "tool-1", + partialResult: { content: [value] }, + }, + } as unknown as Event); + progress("first"); + progress("latest"); + expect(vi.getTimerCount()).toBe(1); + expect( + ( + fakeThis as unknown as { + pendingProgressEvents?: Map; + } + ).pendingProgressEvents?.get("tool:tool-1"), + ).toMatchObject({ partialResult: { content: ["latest"] } }); + + const terminal = listener?.({ + type: "session_event", + event: { type: "agent_end" } as AgentConnectionSessionEvent, + }); + await terminal; + expect(handled).toEqual(["latest", "agent_end"]); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + test("retains independent progress entities, coalesces by key, and flushes retained event order", async () => { + vi.useFakeTimers(); + try { + type Event = { type: "session_event"; event: AgentConnectionSessionEvent }; + let listener: ((event: Event) => Promise) | undefined; + const handled: string[] = []; + const fakeThis = { + agentConnection: { + subscribe: (callback: (event: Event) => Promise) => { + listener = callback; + return vi.fn(); + }, + }, + sessionEventQueue: Promise.resolve(), + sessionEventGeneration: 0, + progressFlushGeneration: 0, + progressFlushStopped: false, + handleEvent: vi.fn( + async (event: { type: string; toolCallId?: string; partialResult?: { content: string[] } }) => { + handled.push( + event.type === "tool_execution_update" + ? `${event.toolCallId}:${event.partialResult?.content[0]}` + : event.type, + ); + }, + ), + showError: vi.fn(), + }; + Object.setPrototypeOf(fakeThis, InteractiveMode.prototype); + ( + InteractiveMode.prototype as unknown as { subscribeToAgent(this: typeof fakeThis): void } + ).subscribeToAgent.call(fakeThis); + + const progress = (toolCallId: string, value: string) => + listener?.({ + type: "session_event", + event: { + type: "tool_execution_update", + toolCallId, + partialResult: { content: [value] }, + }, + } as unknown as Event); + progress("tool-a", "first"); + progress("tool-b", "only"); + progress("tool-a", "latest"); + + expect([ + ...(fakeThis as unknown as { pendingProgressEvents: Map }).pendingProgressEvents.keys(), + ]).toEqual(["tool:tool-b", "tool:tool-a"]); + await listener?.({ type: "session_event", event: { type: "agent_end" } as AgentConnectionSessionEvent }); + expect(handled).toEqual(["tool-b:only", "tool-a:latest", "agent_end"]); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + test("bounds distinct progress coalescing without dropping ordered newest updates", async () => { + vi.useFakeTimers(); + try { + type Event = { type: "session_event"; event: AgentConnectionSessionEvent }; + let listener: ((event: Event) => Promise) | undefined; + const attempted: string[] = []; + const showError = vi.fn(); + const fakeThis = { + agentConnection: { + subscribe: (callback: (event: Event) => Promise) => { + listener = callback; + return vi.fn(); + }, + }, + sessionEventQueue: Promise.resolve(), + sessionEventGeneration: 0, + progressFlushGeneration: 0, + progressFlushStopped: false, + handleEvent: vi.fn(async (event: { type: string; toolCallId?: string }) => { + const id = event.toolCallId ?? event.type; + attempted.push(id); + if (id === "tool-64") throw new Error("bounded flush failed"); + }), + showError, + }; + Object.setPrototypeOf(fakeThis, InteractiveMode.prototype); + ( + InteractiveMode.prototype as unknown as { subscribeToAgent(this: typeof fakeThis): void } + ).subscribeToAgent.call(fakeThis); + const progress = (toolCallId: string) => + listener?.({ + type: "session_event", + event: { type: "tool_execution_update", toolCallId, partialResult: { content: [] } }, + } as unknown as Event); + + let observedPendingHighWater = 0; + // 129 is one more than MAX_PENDING_PROGRESS_EVENTS. The 129th distinct + // key must synchronously drain the first ordered batch rather than evict it. + for (let index = 0; index <= 128; index++) { + await progress(`tool-${index}`); + observedPendingHighWater = Math.max( + observedPendingHighWater, + (fakeThis as unknown as { pendingProgressEvents: Map }).pendingProgressEvents.size, + ); + } + expect(observedPendingHighWater).toBe(128); + expect( + (fakeThis as unknown as { pendingProgressEvents: Map }).pendingProgressEvents.size, + ).toBe(1); + + await (fakeThis as unknown as { sessionEventQueue: Promise }).sessionEventQueue; + await vi.runAllTimersAsync(); + await (fakeThis as unknown as { sessionEventQueue: Promise }).sessionEventQueue; + expect(attempted).toEqual(Array.from({ length: 129 }, (_, index) => `tool-${index}`)); + expect(showError).toHaveBeenCalledWith("bounded flush failed"); + + // A failed entry never poisons the existing UI tail; a later scheduled + // flush still runs after the bounded batch has drained. + await progress("tool-recovered"); + await vi.runAllTimersAsync(); + await (fakeThis as unknown as { sessionEventQueue: Promise }).sessionEventQueue; + expect(attempted).toEqual([...Array.from({ length: 129 }, (_, index) => `tool-${index}`), "tool-recovered"]); + expect( + (fakeThis as unknown as { pendingProgressEvents: Map }).pendingProgressEvents.size, + ).toBe(0); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + test("preserves a cap-drained pre-replacement batch ahead of replacement while new generation progress remains live", async () => { + vi.useFakeTimers(); + try { + type Event = + | { type: "session_event"; event: AgentConnectionSessionEvent } + | { type: "session_replaced"; state: AgentConnectionState }; + let listener: ((event: Event) => Promise) | undefined; + const handled: string[] = []; + const fakeThis = { + agentConnection: { + subscribe: (callback: (event: Event) => Promise) => { + listener = callback; + return vi.fn(); + }, + }, + sessionEventQueue: Promise.resolve(), + sessionEventGeneration: 0, + progressFlushGeneration: 0, + progressFlushStopped: false, + handleEvent: vi.fn(async (event: { type: string; toolCallId?: string }) => { + handled.push(event.toolCallId ?? event.type); + }), + resetSideQuestion: vi.fn(), + resetExtensionUI: vi.fn(), + applyConnectionStateSnapshot: vi.fn(), + resetCurrentSessionRenderState: vi.fn(), + rebindCurrentSession: vi.fn(async () => {}), + renderInitialMessages: vi.fn(async () => { + handled.push("session_replaced"); + }), + ui: { requestRender: vi.fn() }, + showError: vi.fn(), + }; + Object.setPrototypeOf(fakeThis, InteractiveMode.prototype); + ( + InteractiveMode.prototype as unknown as { subscribeToAgent(this: typeof fakeThis): void } + ).subscribeToAgent.call(fakeThis); + const progress = (toolCallId: string) => + listener?.({ + type: "session_event", + event: { type: "tool_execution_update", toolCallId, partialResult: { content: [] } }, + } as Event); + + // The 129th distinct entity drains the first full batch. Immediately replace + // before any queued handler can run, reproducing the cap/replacement race. + for (let index = 0; index <= 128; index++) progress(`old-${index}`); + const replacement = listener?.({ type: "session_replaced", state: createConnectionState() }); + await replacement; + expect(handled).toEqual([...Array.from({ length: 129 }, (_, index) => `old-${index}`), "session_replaced"]); + + // The replacement still fences the old session while admitting new progress. + await progress("new-generation"); + await vi.runAllTimersAsync(); + await (fakeThis as unknown as { sessionEventQueue: Promise }).sessionEventQueue; + expect(handled).toEqual([ + ...Array.from({ length: 129 }, (_, index) => `old-${index}`), + "session_replaced", + "new-generation", + ]); + } finally { + vi.useRealTimers(); + } + }); + + test("surfaces timer and explicit progress flush failures while recovering the queue", async () => { + vi.useFakeTimers(); + try { + type Event = { type: "session_event"; event: AgentConnectionSessionEvent }; + let listener: ((event: Event) => Promise) | undefined; + const showError = vi.fn(); + const handled: string[] = []; + const fakeThis = { + agentConnection: { + subscribe: (callback: (event: Event) => Promise) => { + listener = callback; + return vi.fn(); + }, + }, + sessionEventQueue: Promise.resolve(), + sessionEventGeneration: 0, + progressFlushGeneration: 0, + progressFlushStopped: false, + handleEvent: vi.fn(async (event: { type: string; toolCallId?: string }) => { + if (event.toolCallId === "timer-failure") throw new Error("timer flush failed"); + if (event.toolCallId === "explicit-failure") throw new Error("explicit flush failed"); + handled.push(event.toolCallId ?? event.type); + }), + showError, + }; + Object.setPrototypeOf(fakeThis, InteractiveMode.prototype); + ( + InteractiveMode.prototype as unknown as { subscribeToAgent(this: typeof fakeThis): void } + ).subscribeToAgent.call(fakeThis); + const progress = (toolCallId: string) => + listener?.({ + type: "session_event", + event: { type: "tool_execution_update", toolCallId, partialResult: { content: [] } }, + } as unknown as Event); + + progress("timer-failure"); + await vi.runAllTimersAsync(); + await (fakeThis as unknown as { sessionEventQueue: Promise }).sessionEventQueue; + expect(showError).toHaveBeenCalledWith("timer flush failed"); + + progress("explicit-failure"); + await listener?.({ type: "session_event", event: { type: "agent_end" } as AgentConnectionSessionEvent }); + expect(showError).toHaveBeenCalledWith("explicit flush failed"); + expect(handled).toEqual(["agent_end"]); + + progress("recovered"); + await vi.runAllTimersAsync(); + await (fakeThis as unknown as { sessionEventQueue: Promise }).sessionEventQueue; + expect(handled).toEqual(["agent_end", "recovered"]); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + test("invalidates retained progress before UI stop so no stale callback mutates the UI", async () => { + vi.useFakeTimers(); + try { + const handleEvent = vi.fn(async () => {}); + const fakeThis = { + sessionEventQueue: Promise.resolve(), + sessionEventGeneration: 0, + progressFlushGeneration: 0, + progressFlushStopped: false, + handleEvent, + unregisterSignalHandlers: vi.fn(), + clearCtrlCExitHint: vi.fn(), + clearEscapeRepeat: vi.fn(), + settingsManager: { getShowTerminalProgress: () => false }, + ui: { terminal: { setProgress: vi.fn() } }, + stopWorkingLoader: vi.fn(), + endFeatureHintRun: vi.fn(), + stopWorkingPulse: vi.fn(), + stopGoalTrayTimer: vi.fn(), + closeHeartbeatManager: vi.fn(), + clearExtensionTerminalInputListeners: vi.fn(), + footer: { dispose: vi.fn() }, + footerDataProvider: { dispose: vi.fn() }, + isInitialized: false, + }; + Object.setPrototypeOf(fakeThis, InteractiveMode.prototype); + const queueProgressEvent = ( + InteractiveMode.prototype as unknown as { + queueProgressEvent(this: typeof fakeThis, event: AgentConnectionSessionEvent, generation: number): void; + } + ).queueProgressEvent; + queueProgressEvent.call(fakeThis, { type: "tool_execution_update" } as AgentConnectionSessionEvent, 0); + expect(vi.getTimerCount()).toBe(1); + (InteractiveMode.prototype as unknown as { stop(this: typeof fakeThis): void }).stop.call(fakeThis); + await vi.runAllTimersAsync(); + await fakeThis.sessionEventQueue; + expect(handleEvent).not.toHaveBeenCalled(); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + test("invalidates old-generation progress before a replacement and does not rearm it", async () => { + vi.useFakeTimers(); + try { + type Event = + | { type: "session_event"; event: AgentConnectionSessionEvent } + | { type: "session_replaced"; state: AgentConnectionState }; + let listener: ((event: Event) => Promise) | undefined; + const handled: string[] = []; + const fakeThis = { + agentConnection: { + subscribe: (callback: (event: Event) => Promise) => { + listener = callback; + return vi.fn(); + }, + }, + sessionEventQueue: Promise.resolve(), + sessionEventGeneration: 0, + progressFlushGeneration: 0, + progressFlushStopped: false, + handleEvent: vi.fn(async (event: { type: string; partialResult?: { content: unknown[] } }) => { + handled.push( + event.type === "tool_execution_update" ? String(event.partialResult?.content[0]) : event.type, + ); + }), + resetSideQuestion: vi.fn(), + resetExtensionUI: vi.fn(), + applyConnectionStateSnapshot: vi.fn(), + resetCurrentSessionRenderState: vi.fn(), + rebindCurrentSession: vi.fn(async () => {}), + renderInitialMessages: vi.fn(async () => {}), + ui: { requestRender: vi.fn() }, + showError: vi.fn(), + }; + Object.setPrototypeOf(fakeThis, InteractiveMode.prototype); + ( + InteractiveMode.prototype as unknown as { subscribeToAgent(this: typeof fakeThis): void } + ).subscribeToAgent.call(fakeThis); + listener?.({ + type: "session_event", + event: { type: "tool_execution_update", toolCallId: "tool-1", partialResult: { content: ["old"] } }, + } as unknown as Event); + expect(vi.getTimerCount()).toBe(1); + await listener?.({ type: "session_replaced", state: createConnectionState() }); + await vi.runAllTimersAsync(); + expect(handled).toEqual(["old"]); + expect( + (fakeThis as unknown as { pendingProgressEvents?: Map }).pendingProgressEvents?.size ?? 0, + ).toBe(0); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); }); describe("InteractiveMode connection extension UI", () => { diff --git a/packages/coding-agent/test/swarm/c02-event-loop-evidence.test.ts b/packages/coding-agent/test/swarm/c02-event-loop-evidence.test.ts new file mode 100644 index 000000000..5ed5f01a0 --- /dev/null +++ b/packages/coding-agent/test/swarm/c02-event-loop-evidence.test.ts @@ -0,0 +1,126 @@ +/** Authenticated C02 evidence from the integrated owner, daemon attachment, and UI seams. */ +import { generateKeyPairSync } from "node:crypto"; +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { cpus, tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, expect, test } from "vitest"; +import { C02_FANOUT, type C02IntegratedRepetition, runC02IntegratedRepetition } from "./c02-integrated-harness.js"; +import { + verifySignedProductionEvidence, + verifySignedProductionEvidenceFreshProcess, + writeSignedProductionEvidence, +} from "./production-evidence-adapter.js"; + +const WARMUP_REPETITIONS = 1; +const MEASURED_REPETITIONS = 3; +const cleanups: string[] = []; + +afterEach(async () => { + await Promise.all(cleanups.splice(0).map((path) => rm(path, { recursive: true, force: true }))); +}); + +function samples(repetitions: readonly C02IntegratedRepetition[]) { + const values = (key: Key) => + repetitions.map((repetition) => repetition[key]); + return { + c02ParentPendingHighWater: values("parentPendingHighWater"), + c02UiPendingHighWater: values("uiPendingHighWater"), + c02SlowCatchupPendingHighWater: values("slowCatchupPendingHighWater"), + c02SlowCatchupScheduleHighWater: values("slowCatchupScheduleHighWater"), + c02SlowCatchupPromiseHighWater: values("slowCatchupPromiseHighWater"), + c02TimersScheduled: values("timersScheduled"), + c02TimersCancelled: values("timersCancelled"), + c02TimersFired: values("timersFired"), + c02TerminalDeliveries: values("terminalDeliveries"), + c02HealthyAttachmentLive: values("healthyAttachmentLive"), + c02HookErrors: values("hookErrors"), + c02ObserverErrors: values("observerErrors"), + c02BeforeToolVetoes: values("beforeToolVetoes"), + c02DroppedReplaceableProgress: values("droppedReplaceableProgress"), + c02TeardownPending: values("teardownPending"), + c02DelayP50Milliseconds: values("delayP50Milliseconds"), + c02DelayP95Milliseconds: values("delayP95Milliseconds"), + c02DelayP99Milliseconds: values("delayP99Milliseconds"), + c02DelayMaxMilliseconds: values("delayMaxMilliseconds"), + }; +} + +describe("C02 integrated event-loop evidence", () => { + test("observes fresh owner, attachment, and UI lifecycle repetitions through B00B", async () => { + // Excluded warm-up owns a fresh AgentSession, supervisor, clients, UI harness, and delay monitor. + const warmup = await runC02IntegratedRepetition(false); + expect(warmup.terminalDeliveries).toBe(C02_FANOUT); + const repetitions: C02IntegratedRepetition[] = []; + for (let index = 0; index < MEASURED_REPETITIONS; index++) + repetitions.push(await runC02IntegratedRepetition(true)); + for (const repetition of repetitions) { + expect(repetition.parentPendingHighWater).toBe(C02_FANOUT); + expect(repetition.uiPendingHighWater).toBe(1); + expect(repetition.slowCatchupPendingHighWater).toBe(1); + expect(repetition.slowCatchupScheduleHighWater).toBe(1); + expect(repetition.slowCatchupPromiseHighWater).toBe(1); + expect(repetition.timersScheduled).toBe(repetition.timersCancelled + repetition.timersFired); + expect(repetition.terminalDeliveries).toBe(C02_FANOUT); + expect(repetition.healthyAttachmentLive).toBe(1); + expect(repetition.hookErrors).toBe(1); + expect(repetition.observerErrors).toBeGreaterThanOrEqual(1); + expect(repetition.beforeToolVetoes).toBe(1); + expect(repetition.teardownPending).toBe(0); + expect(repetition.delayP99Milliseconds).toBeLessThanOrEqual(50); + expect(repetition.delayMaxMilliseconds).toBeLessThanOrEqual(100); + } + + const artifactDirectory = await mkdtemp(join(tmpdir(), "c02-integrated-artifact-")); + const trustDirectory = await mkdtemp(join(tmpdir(), "c02-integrated-trust-")); + cleanups.push(artifactDirectory, trustDirectory); + const keys = generateKeyPairSync("ed25519"); + const publicKeyPem = keys.publicKey.export({ type: "spki", format: "pem" }).toString(); + const written = await writeSignedProductionEvidence( + artifactDirectory, + trustDirectory, + { + scenario: "c02-integrated-owner-attachment-ui", + attempts: Array.from({ length: C02_FANOUT }, (_, index) => ({ + requestId: `request-${String(index + 1).padStart(4, "0")}` as `request-${string}`, + attempt: 1, + requested: { provider: "b00b-scripted", model: "fixture-zero" }, + resolved: { + api: "b00b-scripted", + provider: "b00b-scripted", + model: "fixture-zero", + responseModel: "fixture-zero-resolved", + }, + terminal: "done" as const, + usage: { inputMicroTokens: 0, outputMicroTokens: 0, cacheReadMicroTokens: 0, cacheWriteMicroTokens: 0 }, + })), + priceCard: { + version: "c02-integrated-test-only", + inputMicroCurrencyPerMillionMicroTokens: 0, + outputMicroCurrencyPerMillionMicroTokens: 0, + }, + metadata: { + c02Fanout: C02_FANOUT, + c02WarmupRepetitions: WARMUP_REPETITIONS, + c02MeasuredRepetitions: MEASURED_REPETITIONS, + ...samples(repetitions), + c02EnvironmentNodeMajor: Number(process.versions.node.split(".")[0]), + c02EnvironmentProcessorCount: cpus().length, + c02EnvironmentPlatformKnown: true, + }, + }, + keys.privateKey, + ); + await expect( + verifySignedProductionEvidence(artifactDirectory, written.commitmentPath, publicKeyPem), + ).resolves.toBeUndefined(); + await expect( + verifySignedProductionEvidenceFreshProcess(artifactDirectory, written.commitmentPath, publicKeyPem), + ).resolves.toBeUndefined(); + await writeFile(join(artifactDirectory, "summary.json"), "{}\n"); + await expect( + verifySignedProductionEvidenceFreshProcess(artifactDirectory, written.commitmentPath, publicKeyPem), + ).rejects.toThrow("B00B_EVIDENCE_FRESH_VERIFY_FAILED"); + // The trust-root commitment is stored separately from the mutable artifact root. + expect(await readFile(written.commitmentPath, "utf8")).toContain(written.artifactBundleId); + }, 60_000); +}); diff --git a/packages/coding-agent/test/swarm/c02-integrated-harness.ts b/packages/coding-agent/test/swarm/c02-integrated-harness.ts new file mode 100644 index 000000000..46180a892 --- /dev/null +++ b/packages/coding-agent/test/swarm/c02-integrated-harness.ts @@ -0,0 +1,494 @@ +/** C02 evidence harness: drives integrated ownership, daemon attachment, and UI seams. */ + +import { once } from "node:events"; +import { mkdir, rm } from "node:fs/promises"; +import { connect, createServer, type Server, type Socket } from "node:net"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { monitorEventLoopDelay } from "node:perf_hooks"; +import { type FauxResponseStep, fauxAssistantMessage, registerFauxProvider } from "@earendil-works/pi-ai"; +import { createAgentSessionFromServices, createAgentSessionServices } from "../../src/core/agent-session-runtime.js"; +import { AuthStorage } from "../../src/core/auth-storage.js"; +import { SessionManager } from "../../src/core/session-manager.js"; +import type { DaemonSocketClient } from "../../src/modes/daemon/active-session-state.js"; +import { DaemonSupervisor } from "../../src/modes/daemon/daemon-supervisor.js"; +import { InteractiveMode } from "../../src/modes/interactive/interactive-mode.js"; + +export const C02_FANOUT = 64; +const OPERATION_TIMEOUT_MS = 10_000; + +export interface C02IntegratedRepetition { + parentPendingHighWater: number; + uiPendingHighWater: number; + slowCatchupPendingHighWater: number; + slowCatchupScheduleHighWater: number; + slowCatchupPromiseHighWater: number; + timersScheduled: number; + timersCancelled: number; + timersFired: number; + terminalDeliveries: number; + healthyAttachmentLive: number; + hookErrors: number; + observerErrors: number; + beforeToolVetoes: number; + droppedReplaceableProgress: number; + teardownPending: number; + delayP50Milliseconds: number; + delayP95Milliseconds: number; + delayP99Milliseconds: number; + delayMaxMilliseconds: number; +} + +const afterMacrotask = () => new Promise((resolve) => setTimeout(resolve, 0)); +const afterImmediate = () => new Promise((resolve) => setImmediate(resolve)); + +async function bounded(label: string, promise: Promise, timeoutMs = OPERATION_TIMEOUT_MS): Promise { + let timer: ReturnType | undefined; + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`C02_INTEGRATED_TIMEOUT:${label}`)), timeoutMs); + }), + ]); + } finally { + if (timer) clearTimeout(timer); + } +} + +function requireInvariant(condition: unknown, message: string): asserts condition { + if (!condition) throw new Error(`C02_INTEGRATED_INVARIANT:${message}`); +} + +async function closeServer(server: Server): Promise { + await new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve()))); +} + +/** + * Opens an actual local socket through DaemonSupervisor.handleConnection. This + * deliberately does not start a daemon worker: C02 owns attachment-local work, + * and the drain seam below is the only worker-dependent operation substituted. + */ +async function openDaemonAttachment(supervisor: DaemonSupervisor): Promise<{ + attachment: DaemonSocketClient; + peer: Socket; + close(): Promise; +}> { + let resolveAttachment!: (attachment: DaemonSocketClient) => void; + let rejectAttachment!: (error: Error) => void; + const accepted = new Promise((resolve, reject) => { + resolveAttachment = resolve; + rejectAttachment = reject; + }); + const internals = supervisor as unknown as { + handleConnection(socket: Socket): void; + clients: Set; + }; + const server = createServer((socket) => { + try { + internals.handleConnection(socket); + const attachment = [...internals.clients].find((candidate) => candidate.socket === socket); + if (!attachment) throw new Error("Daemon did not register local attachment"); + resolveAttachment(attachment); + } catch (error) { + rejectAttachment(error instanceof Error ? error : new Error(String(error))); + socket.destroy(); + } + }); + await bounded( + "daemon attachment listen", + new Promise((resolve, reject) => { + server.once("error", reject); + server.listen({ host: "127.0.0.1", port: 0 }, () => { + server.off("error", reject); + resolve(); + }); + }), + ); + const address = server.address(); + if (!address || typeof address === "string") { + server.close(); + throw new Error("C02_INTEGRATED_INVARIANT:local TCP address unavailable"); + } + const peer = connect({ host: "127.0.0.1", port: address.port }); + peer.once("error", rejectAttachment); + await bounded("daemon attachment connect", once(peer, "connect")); + const attachment = await bounded("daemon attachment registration", accepted); + let closed = false; + return { + attachment, + peer, + async close() { + if (closed) return; + closed = true; + const peerClosed = once(peer, "close").catch(() => undefined); + attachment.socket.destroy(); + await bounded("daemon attachment socket close", peerClosed); + await bounded("daemon attachment server close", closeServer(server)); + }, + }; +} + +/** Exercise real socket close cleanup plus real per-attachment scheduler/latch ownership. */ +async function exerciseAttachments(): Promise< + Pick< + C02IntegratedRepetition, + | "slowCatchupPendingHighWater" + | "slowCatchupScheduleHighWater" + | "slowCatchupPromiseHighWater" + | "healthyAttachmentLive" + | "timersScheduled" + | "timersCancelled" + | "timersFired" + > +> { + const supervisor = new DaemonSupervisor("/tmp/c02-integrated-evidence.sock", { + defaultSessionConfig: { agentDir: "/tmp", cwd: "/tmp" }, + descriptorDir: "/tmp/c02-integrated-evidence-state", + }); + const slowConnection = await openDaemonAttachment(supervisor); + const healthyConnection = await openDaemonAttachment(supervisor); + const slow = slowConnection.attachment; + const healthy = healthyConnection.attachment; + const internals = supervisor as unknown as { + queueCatchup(client: DaemonSocketClient, activeSessionId: string, purpose?: "replacement" | "resync"): void; + scheduleClientCatchup(client: DaemonSocketClient): void; + cancelClientCatchup(client: DaemonSocketClient): void; + catchUpClient(client: DaemonSocketClient): Promise; + drainClientCatchups(client: DaemonSocketClient): Promise; + catchupDrainTimers: Map>; + clients: Set; + }; + let releaseSlow!: () => void; + const slowGate = new Promise((resolve) => { + releaseSlow = resolve; + }); + let healthyDrains = 0; + let timersScheduled = 0; + let timersCancelled = 0; + let timersFired = 0; + const realSchedule = internals.scheduleClientCatchup.bind(supervisor); + const realCancel = internals.cancelClientCatchup.bind(supervisor); + const realDrain = internals.drainClientCatchups.bind(supervisor); + const realTimerDelete = internals.catchupDrainTimers.delete.bind(internals.catchupDrainTimers); + let cancellingTimer = false; + // Observe removal at the production timer map. A callback deletes itself before + // deciding whether its client is still live, while cancelClientCatchup deletes + // the same entry as part of real socket cleanup. This accounts for both paths + // exactly once, including an inert callback that observes a closing socket. + internals.catchupDrainTimers.delete = (client) => { + const removed = realTimerDelete(client); + if (removed) { + if (cancellingTimer) timersCancelled++; + else timersFired++; + } + return removed; + }; + internals.scheduleClientCatchup = (client) => { + const scheduledBefore = internals.catchupDrainTimers.has(client); + realSchedule(client); + if (!scheduledBefore && internals.catchupDrainTimers.has(client)) timersScheduled++; + }; + internals.cancelClientCatchup = (client) => { + cancellingTimer = true; + try { + realCancel(client); + } finally { + cancellingTimer = false; + } + }; + // Replace only the worker request body. Queue ownership, immediate callback, + // latch, backpressure conditions, and socket-close cleanup stay production code. + internals.drainClientCatchups = async (client) => { + client.catchupActiveSessionIds?.clear(); + client.catchupPurposes?.clear(); + if (client === slow) await slowGate; + else if (client === healthy) healthyDrains++; + else await realDrain(client); + }; + try { + let slowPendingHighWater = 0; + let scheduledHighWater = 0; + for (let index = 0; index < C02_FANOUT; index++) { + internals.queueCatchup(slow, "c02-active"); + internals.scheduleClientCatchup(slow); + internals.queueCatchup(healthy, "c02-active"); + internals.scheduleClientCatchup(healthy); + slowPendingHighWater = Math.max(slowPendingHighWater, slow.catchupActiveSessionIds?.size ?? 0); + scheduledHighWater = Math.max(scheduledHighWater, Number(internals.catchupDrainTimers.has(slow))); + } + requireInvariant(slowPendingHighWater === 1, "slow attachment retains one latest session"); + requireInvariant(scheduledHighWater === 1, "slow attachment owns one deferred scheduler callback"); + const latchDeadline = Date.now() + OPERATION_TIMEOUT_MS; + while (slow.catchupPromise === undefined && Date.now() < latchDeadline) await afterImmediate(); + const promiseHighWater = Number(slow.catchupPromise !== undefined); + requireInvariant(promiseHighWater === 1, "slow attachment owns one in-flight catch-up promise"); + requireInvariant(healthyDrains === 1, "healthy attachment drains while slow is pending"); + releaseSlow(); + await bounded("slow attachment settle", slow.catchupPromise ?? Promise.resolve()); + await afterMacrotask(); + requireInvariant(slow.catchupPromise === undefined, "slow catch-up latch clears after completion"); + + // Queue a real replacement callback, then use the actual socket close path. + // Its cleanup invalidates/clears that callback in one later macrotask. + internals.queueCatchup(slow, "c02-replacement", "replacement"); + internals.scheduleClientCatchup(slow); + requireInvariant(slow.catchupPurposes?.get("c02-replacement") === "replacement", "replacement queued"); + // The production socket cleanup invokes the instrumented cancelClientCatchup + // wrapper, which observes this callback's real removal exactly once. + await slowConnection.close(); + await afterMacrotask(); + requireInvariant(!internals.clients.has(slow), "closed attachment removed from daemon"); + requireInvariant(!internals.catchupDrainTimers.has(slow), "closed socket owns no callback"); + requireInvariant((slow.catchupActiveSessionIds?.size ?? 0) === 0, "closed socket retains no catch-up"); + requireInvariant((slow.catchupPurposes?.size ?? 0) === 0, "closed socket retains no replacement"); + requireInvariant(slow.catchupPromise === undefined, "closed socket retains no catch-up latch"); + await healthyConnection.close(); + await afterMacrotask(); + requireInvariant(internals.clients.size === 0, "all local socket attachments cleaned up"); + return { + slowCatchupPendingHighWater: slowPendingHighWater, + slowCatchupScheduleHighWater: scheduledHighWater, + slowCatchupPromiseHighWater: promiseHighWater, + healthyAttachmentLive: healthyDrains, + timersScheduled, + timersCancelled, + timersFired, + }; + } finally { + releaseSlow(); + internals.drainClientCatchups = realDrain; + internals.catchupDrainTimers.delete = realTimerDelete; + internals.scheduleClientCatchup = realSchedule; + internals.cancelClientCatchup = realCancel; + await slowConnection.close(); + await healthyConnection.close(); + } +} + +/** Drive real InteractiveMode progress ordering through its established connection seam. */ +async function exerciseInteractive(): Promise< + Pick +> { + type Event = { + type: "session_event"; + event: { type: string; toolCallId?: string; partialResult?: { content: string[] } }; + }; + let listener: ((event: Event) => Promise) | undefined; + const handled: string[] = []; + const harness = { + agentConnection: { + subscribe: (callback: (event: Event) => Promise) => { + listener = callback; + return () => {}; + }, + }, + sessionEventQueue: Promise.resolve(), + sessionEventGeneration: 0, + progressFlushGeneration: 0, + progressFlushStopped: false, + handleEvent: async (event: { type: string; partialResult?: { content: string[] } }) => { + handled.push(event.type === "tool_execution_update" ? String(event.partialResult?.content[0]) : event.type); + }, + showError: () => {}, + }; + Object.setPrototypeOf(harness, InteractiveMode.prototype); + (InteractiveMode.prototype as unknown as { subscribeToAgent(this: typeof harness): void }).subscribeToAgent.call( + harness, + ); + let pendingHighWater = 0; + for (let index = 0; index < C02_FANOUT; index++) { + await listener?.({ + type: "session_event", + event: { type: "tool_execution_update", toolCallId: "c02", partialResult: { content: [`progress-${index}`] } }, + }); + pendingHighWater = Math.max( + pendingHighWater, + (harness as typeof harness & { pendingProgressEvents?: Map }).pendingProgressEvents?.size ?? + 0, + ); + } + await listener?.({ type: "session_event", event: { type: "agent_end" } }); + const progressDelivered = handled.filter((entry) => entry.startsWith("progress-")).length; + requireInvariant( + handled.length === 2 && handled[0] === "progress-63" && handled[1] === "agent_end", + "UI flushes latest progress before terminal", + ); + requireInvariant( + ((harness as typeof harness & { pendingProgressEvents?: Map }).pendingProgressEvents?.size ?? + 0) === 0, + "UI retains no progress after terminal", + ); + return { uiPendingHighWater: pendingHighWater, droppedReplaceableProgress: C02_FANOUT - progressDelivered }; +} + +/** One fresh session/services/provider/socket/monitor repetition. */ +export async function runC02IntegratedRepetition(measured = true): Promise { + const delay = monitorEventLoopDelay({ resolution: 10 }); + const directory = join( + tmpdir(), + `pi-c02-integrated-${process.pid}-${Date.now()}-${Math.random().toString(36).slice(2)}`, + ); + await mkdir(directory, { recursive: true }); + const faux = registerFauxProvider(); + let releaseAnswers!: () => void; + const answerGate = new Promise((resolve) => { + releaseAnswers = resolve; + }); + faux.setResponses( + Array.from({ length: C02_FANOUT }, () => async () => { + await answerGate; + return fauxAssistantMessage("c02 terminal"); + }) as FauxResponseStep[], + ); + const authStorage = AuthStorage.inMemory(); + authStorage.setRuntimeApiKey(faux.getModel().provider, "faux-key"); + let session: any; + try { + const services = await bounded( + "create real session services", + createAgentSessionServices({ + agentDir: directory, + authStorage, + cwd: directory, + resourceLoaderOptions: { noSkills: true, noPromptTemplates: true, noThemes: true }, + }), + ); + const created = await bounded( + "create real session", + createAgentSessionFromServices({ + services, + sessionManager: SessionManager.create(directory, join(directory, "sessions")), + model: faux.getModel(), + rlmDepth: 0, + rlmMaxDepth: 2, + }), + ); + session = created.session; + const delivered = new Map(); + const observedBefore = session._observerFailureDiagnostics; + const hooksBefore = session._afterToolHookFailureDiagnostics; + let observerFailureInjected = false; + session.subscribe(() => { + // Exercise the production observer isolation path once without turning a + // metric run into a console-I/O benchmark. + if (!observerFailureInjected) { + observerFailureInjected = true; + throw new Error("C02 observer isolation"); + } + }); + session.subscribe((event: any) => { + if (event.type === "rlm_child_update" && event.child.status === "done") + delivered.set(event.child.id, (delivered.get(event.child.id) ?? 0) + 1); + }); + const handles = await bounded( + "admit independent real child runtimes", + Promise.all(Array.from({ length: C02_FANOUT }, (_, index) => session.runRlmChild(`C02 child ${index}`))), + ); + const deadline = Date.now() + OPERATION_TIMEOUT_MS; + while (handles.some((handle) => !session.getRlmChildSession(handle.rlm_child_id))) { + if (Date.now() > deadline) throw new Error("C02_INTEGRATED_TIMEOUT:real child lifecycle publication"); + await afterMacrotask(); + } + + // Begin a fresh measurement only after fixture/session construction and all + // 64 child admissions. Prime the new monitor for one setup-only turn, then + // reset immediately before the owner recap. From that reset through lifecycle + // teardown, do not reset it: every measured C02 owner, hook, terminal, + // attachment, InteractiveMode, and cleanup path is in the reported interval. + delay.enable(); + await afterMacrotask(); + delay.reset(); + for (const handle of handles) { + const child = session.getRlmChildSession(handle.rlm_child_id); + child.setCurrentRecap("replaceable-first"); + child.setCurrentRecap("replaceable-latest"); + } + const parentPendingHighWater = session._pendingRlmChildUpdates.size; + requireInvariant(parentPendingHighWater === C02_FANOUT, "owner retained one latest activity per real child"); + await afterMacrotask(); + requireInvariant(session._pendingRlmChildUpdates.size === 0, "owner flushes after one macrotask"); + const runner = session._extensionRunner; + const originalHasHandlers = runner.hasHandlers.bind(runner); + runner.hasHandlers = (type: string) => + type === "tool_result" || type === "tool_call" || originalHasHandlers(type); + runner.emitToolResult = async () => { + throw new Error("C02 after hook"); + }; + runner.emitToolCall = async () => { + throw new Error("C02 before veto"); + }; + await session.agent.afterToolCall?.({ + toolCall: { id: "c02", name: "c02", arguments: {} }, + args: {}, + result: { content: [] }, + isError: false, + }); + let beforeToolVetoes = 0; + try { + await session.agent.beforeToolCall?.({ toolCall: { id: "c02", name: "c02", arguments: {} }, args: {} }); + } catch { + beforeToolVetoes++; + } + releaseAnswers(); + while (delivered.size !== C02_FANOUT) { + if (Date.now() > deadline) throw new Error("C02_INTEGRATED_TIMEOUT:real child terminals"); + await afterMacrotask(); + } + requireInvariant( + [...delivered.values()].every((count) => count === 1), + "terminal delivered once to healthy observer", + ); + const attachments = await exerciseAttachments(); + const interactive = await exerciseInteractive(); + + // Exercise every real owner shutdown edge after terminals: explicit abort, + // update-restart abort, then disposal. All retained child state must be inert + // after one macrotask rather than leaking a timer or late update. This remains + // inside the delay measurement, rather than being treated as teardown noise. + await bounded("parent abort cleanup", session.abort()); + session.abortForUpdateRestart(); + await afterMacrotask(); + session.dispose(); + await afterMacrotask(); + requireInvariant( + session._pendingRlmChildUpdates.size === 0 && session._rlmChildUpdateFlushTimer === undefined, + "abort/restart/dispose leaves no owner callback", + ); + requireInvariant(session._activeRlmChildRuns.size === 0, "dispose releases real child runs"); + + // Allow the monitor's own sampling timer one final turn only after every + // measured path above has completed. This is not a second sampling window: + // the monitor has remained enabled and un-reset since before the owner recap. + await bounded("event-loop delay final sample", new Promise((resolve) => setTimeout(resolve, 25))); + const delayStats = measured + ? { + delayP50Milliseconds: delay.percentile(50) / 1_000_000, + delayP95Milliseconds: delay.percentile(95) / 1_000_000, + delayP99Milliseconds: delay.percentile(99) / 1_000_000, + delayMaxMilliseconds: delay.max / 1_000_000, + } + : { delayP50Milliseconds: 0, delayP95Milliseconds: 0, delayP99Milliseconds: 0, delayMaxMilliseconds: 0 }; + delay.disable(); + return { + parentPendingHighWater, + terminalDeliveries: delivered.size, + hookErrors: session._afterToolHookFailureDiagnostics - hooksBefore, + observerErrors: session._observerFailureDiagnostics - observedBefore, + beforeToolVetoes, + teardownPending: session._pendingRlmChildUpdates.size, + ...delayStats, + ...attachments, + ...interactive, + }; + } finally { + releaseAnswers(); + try { + session?.dispose(); + delay.disable(); + } finally { + faux.unregister(); + await rm(directory, { recursive: true, force: true }); + } + } +} diff --git a/packages/coding-agent/test/swarm/swarm-evidence.ts b/packages/coding-agent/test/swarm/swarm-evidence.ts index 2f7e04a17..1ac9be64b 100644 --- a/packages/coding-agent/test/swarm/swarm-evidence.ts +++ b/packages/coding-agent/test/swarm/swarm-evidence.ts @@ -347,6 +347,33 @@ const SAFE_EVIDENCE_KEYS = new Set([ "version", "inputPerMillionTokens", "outputPerMillionTokens", + // C02 test-only, content-free event-loop evidence fields. Their values are + // validated below rather than accepting arbitrary benchmark metadata. + "c02Fanout", + "c02WarmupRepetitions", + "c02MeasuredRepetitions", + "c02ParentPendingHighWater", + "c02UiPendingHighWater", + "c02SlowCatchupPendingHighWater", + "c02SlowCatchupScheduleHighWater", + "c02SlowCatchupPromiseHighWater", + "c02TimersScheduled", + "c02TimersCancelled", + "c02TimersFired", + "c02TerminalDeliveries", + "c02HealthyAttachmentLive", + "c02HookErrors", + "c02ObserverErrors", + "c02BeforeToolVetoes", + "c02DroppedReplaceableProgress", + "c02TeardownPending", + "c02DelayP50Milliseconds", + "c02DelayP95Milliseconds", + "c02DelayP99Milliseconds", + "c02DelayMaxMilliseconds", + "c02EnvironmentNodeMajor", + "c02EnvironmentProcessorCount", + "c02EnvironmentPlatformKnown", ]); function safeEvidenceString(value: string, key?: string): boolean { return ( @@ -385,7 +412,11 @@ export function redactEvidence(value: T, key?: string, redactObjectKeys = fal if (Array.isArray(value)) return value.map((item) => redactEvidence(item, undefined, redactObjectKeys)) as T; if (isRecord(value)) { const entries = Object.entries(value).map(([entryKey, item]) => { - const safeKey = !redactObjectKeys && SAFE_EVIDENCE_KEYS.has(entryKey); + // C02 is the sole test-only extension to the otherwise opaque metadata bag. + // Its fixed keys and numeric/status values are checked by verifyC02Metadata. + const safeKey = + (!redactObjectKeys && SAFE_EVIDENCE_KEYS.has(entryKey)) || + (key === "metadata" && C02_METADATA_KEYS.has(entryKey)); return [ safeKey ? entryKey : REDACTED, redactEvidence(item, safeKey ? entryKey : undefined, redactObjectKeys || key === "metadata"), @@ -515,7 +546,7 @@ function publicConfig(config: SwarmBenchmarkConfig): Omit = new Set([ + ...C02_NUMERIC_METADATA_KEYS, + ...C02_NUMERIC_ARRAY_METADATA_KEYS, + ...C02_BOOLEAN_METADATA_KEYS, +]); + +/** Keeps C02's test-only metrics named, numeric/status-only, and schema-closed. */ +function verifyC02Metadata(metadata: Record): void { + const present = Object.keys(metadata).filter((key) => key.startsWith("c02")); + if (present.length === 0) return; + assert(present.length === C02_METADATA_KEYS.size, "incomplete C02 event-loop metadata"); + assert( + present.every((key) => C02_METADATA_KEYS.has(key)), + "unknown C02 event-loop metadata", + ); + for (const key of C02_NUMERIC_METADATA_KEYS) + assert( + typeof metadata[key] === "number" && Number.isSafeInteger(metadata[key]) && (metadata[key] as number) >= 0, + `invalid C02 metric: ${key}`, + ); + for (const key of C02_NUMERIC_ARRAY_METADATA_KEYS) + assert( + Array.isArray(metadata[key]) && + metadata[key].length === 3 && + metadata[key].every((value) => typeof value === "number" && Number.isFinite(value) && value >= 0), + `invalid C02 metric samples: ${key}`, + ); + for (const key of C02_BOOLEAN_METADATA_KEYS) + assert(typeof metadata[key] === "boolean", `invalid C02 status: ${key}`); + + assert(metadata.c02Fanout === 64, "C02 evidence requires 64 streams"); + assert(metadata.c02WarmupRepetitions === 1, "C02 evidence requires one excluded warmup"); + assert(metadata.c02MeasuredRepetitions === 3, "C02 evidence requires three measured repetitions"); + const sample = (key: (typeof C02_NUMERIC_ARRAY_METADATA_KEYS)[number]) => metadata[key] as number[]; + for (let index = 0; index < 3; index++) { + assert(sample("c02ParentPendingHighWater")[index] === 64, "invalid C02 owner coalescer bound"); + assert(sample("c02UiPendingHighWater")[index] <= 1, "invalid C02 UI coalescer bound"); + assert(sample("c02SlowCatchupPendingHighWater")[index] <= 1, "invalid C02 slow attachment bound"); + assert(sample("c02SlowCatchupScheduleHighWater")[index] === 1, "invalid C02 slow scheduler bound"); + assert(sample("c02SlowCatchupPromiseHighWater")[index] === 1, "invalid C02 slow latch bound"); + assert( + sample("c02TimersScheduled")[index] === sample("c02TimersCancelled")[index] + sample("c02TimersFired")[index], + "invalid C02 timer accounting", + ); + assert(sample("c02TerminalDeliveries")[index] === 64, "invalid C02 terminal delivery count"); + assert(sample("c02HealthyAttachmentLive")[index] === 1, "invalid C02 healthy attachment delivery"); + assert(sample("c02HookErrors")[index] === 1, "invalid C02 after-hook isolation result"); + assert(sample("c02ObserverErrors")[index] >= 1, "invalid C02 observer isolation result"); + assert(sample("c02BeforeToolVetoes")[index] === 1, "invalid C02 before-hook veto result"); + assert(sample("c02DroppedReplaceableProgress")[index] === 63, "invalid C02 UI replacement count"); + assert(sample("c02TeardownPending")[index] === 0, "invalid C02 teardown result"); + assert(sample("c02DelayP99Milliseconds")[index] <= 50, "C02 p99 delay threshold exceeded"); + assert(sample("c02DelayMaxMilliseconds")[index] <= 100, "C02 max delay threshold exceeded"); + } +} + function requireManifest(manifest: unknown): asserts manifest is SwarmManifest & { artifacts: EvidenceArtifact[] } { assert(isRecord(manifest), "manifest must be an object"); assert( @@ -874,6 +990,7 @@ function requireManifest(manifest: unknown): asserts manifest is SwarmManifest & "invalid artifact bundle identity", ); assertContentFree(manifest); + verifyC02Metadata(manifest.metadata as Record); const attemptIds = new Set(); for (const assignment of manifest.assignments as Record[]) { assert(isRecord(assignment) && isRecord(assignment.requested), "invalid assignment provenance");