From 6dd9e86b75a18a13a7139631a679fb25593645d1 Mon Sep 17 00:00:00 2001 From: Seth Date: Tue, 11 Aug 2026 10:07:03 -0700 Subject: [PATCH 01/17] feat(acp): preserve causal prompt correlation --- .../coding-agent/src/core/agent-session.ts | 48 ++ .../coding-agent/src/modes/acp/acp-meta.ts | 35 ++ .../coding-agent/src/modes/acp/acp-mode.ts | 387 +++++++++--- .../daemon-agent-connection.ts | 17 + .../src/modes/agent-connection/snapshot.ts | 1 + .../test/agent-connection-daemon.test.ts | 36 ++ .../test/agent-connection-in-process.test.ts | 1 + .../fixtures/acp-correlation-transcripts.json | 86 +++ .../coding-agent/test/suite/acp-mode.test.ts | 571 +++++++++++++++++- 9 files changed, 1083 insertions(+), 99 deletions(-) create mode 100644 packages/coding-agent/test/fixtures/acp-correlation-transcripts.json diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index c55a8c22c..201a4172c 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -943,6 +943,7 @@ interface RlmChildRun { prompt: string; sessionName: string; sessionDir: string; + model: Model; status: RlmChildAgentStatus; error?: string; abort: () => void; @@ -9563,6 +9564,52 @@ export class AgentSession { return unsubscribe; } + /** + * Live recursive child roster for connection snapshots. + * + * The run registry is authoritative while a child is queued or running; retained + * sessions preserve completed children and expose any nested work that outlives + * their direct parent run. This deliberately does not reconstruct state from + * observer events, which can predate a newly attached connection. + */ + getRlmChildSnapshots(): RlmChildAgentSnapshot[] { + const snapshots: RlmChildAgentSnapshot[] = []; + const recorded = new Set(); + for (const run of this._activeRlmChildRuns.values()) { + if (run.detachedDeletion || this._deletingRlmChildren.has(run.id) || this._deletedRlmChildIds.has(run.id)) { + continue; + } + const child = run.session; + snapshots.push({ + id: run.id, + parentId: this._rlmParentNodeId, + sessionName: child?.sessionName ?? run.sessionName, + model: `${(child?.model ?? run.model).provider}/${(child?.model ?? run.model).id}`, + label: rlmChildLabel(run.prompt), + status: run.status, + sessionDir: run.sessionDir, + }); + recorded.add(run.id); + if (child) snapshots.push(...child.getRlmChildSnapshots()); + } + for (const [childId, child] of this._rlmChildSessions) { + if (recorded.has(childId) || this._deletingRlmChildren.has(childId) || this._deletedRlmChildIds.has(childId)) { + continue; + } + snapshots.push({ + id: childId, + parentId: this._rlmParentNodeId, + sessionName: child.sessionName, + model: child.model ? `${child.model.provider}/${child.model.id}` : undefined, + label: child.sessionName ?? "child agent", + status: "done", + sessionDir: child._rlmSessionDir ?? child.sessionManager.getSessionDir(), + }); + snapshots.push(...child.getRlmChildSnapshots()); + } + return snapshots; + } + /** True when any direct or nested subagent is still running or queued. */ hasRunningRlmChildren(): boolean { for (const run of this._activeRlmChildRuns.values()) { @@ -9760,6 +9807,7 @@ export class AgentSession { prompt, sessionName, sessionDir: childSessionDir, + model: modelSelection.model, status: "queued", settled: false, abort: noopRlmChildAbort, diff --git a/packages/coding-agent/src/modes/acp/acp-meta.ts b/packages/coding-agent/src/modes/acp/acp-meta.ts index fbe1ff211..86ec98447 100644 --- a/packages/coding-agent/src/modes/acp/acp-meta.ts +++ b/packages/coding-agent/src/modes/acp/acp-meta.ts @@ -59,6 +59,13 @@ export interface PrimeAgentRefinementMeta { error?: string; } +export interface PrimeAgentQuiescenceMeta { + /** Subagents that have not reached a terminal state at the observation point. */ + outstandingSubagents: number; + /** Autonomous continuation slots still available at the observation point. */ + remainingAutonomousContinuations: number; +} + export interface PrimeAgentAgentMessageMeta { toolCallId: string; target?: string; @@ -72,7 +79,33 @@ export interface PrimeAgentCwdMeta { actual: string; } +/** + * Producer-side ordering and causality for ACP updates. + * + * `promptTurnId` is allocated when ACP accepts a prompt, never inferred from + * whichever prompt happens to be running when an update is delivered. `0` + * means a session-scoped event with no prompt origin (for example a heartbeat + * change before the first prompt). `eventSequence` is connection-wide and + * strictly increases for every update Prime Agent publishes. + */ +export type PrimeAgentEventPhase = "event" | "responseBoundary" | "terminalQuiescence"; + +/** The outcome carried by a correlated response boundary and terminal envelope. */ +export type PrimeAgentResponseOutcome = "result" | "error"; + export interface PrimeAgentSessionMeta { + /** Monotonically increasing ACP prompt turn which caused this update. */ + promptTurnId?: number; + /** Strictly increasing producer sequence, across all ACP updates. */ + eventSequence?: number; + /** Whether this is ordinary work, the prompt response boundary, or final quiescence. */ + phase?: PrimeAgentEventPhase; + /** + * The boundary/terminal outcome. This deliberately has only `result` and + * `error`: ACP's transport stop reasons (including `end_turn`) are never a + * causal completion signal. + */ + outcome?: PrimeAgentResponseOutcome; /** Present when a client-requested cwd differs from the agent's real cwd. */ cwd?: PrimeAgentCwdMeta; /** Set when the session's heartbeat or cron schedule changed. */ @@ -86,6 +119,8 @@ export interface PrimeAgentSessionMeta { compaction?: { tokensBefore?: number; summary?: string }; subagents?: PrimeAgentSubagentMeta[]; autonomous?: PrimeAgentAutonomousMeta; + /** Exact zero counters carried by a scoreable terminal-quiescence boundary. */ + quiescence?: PrimeAgentQuiescenceMeta; ipython?: PrimeAgentIpythonMeta; } diff --git a/packages/coding-agent/src/modes/acp/acp-mode.ts b/packages/coding-agent/src/modes/acp/acp-mode.ts index eed228ae0..2a207143b 100644 --- a/packages/coding-agent/src/modes/acp/acp-mode.ts +++ b/packages/coding-agent/src/modes/acp/acp-mode.ts @@ -10,10 +10,14 @@ import type { AgentSessionRuntime } from "../../core/agent-session-runtime.js"; import type { AgentAutonomousStatus } from "../../core/autonomous.js"; import { takeOverStdout, writeRawStdout } from "../../core/output-guard.js"; import { InProcessAgentConnection } from "../agent-connection/in-process-agent-connection.js"; -import type { AgentConnection } from "../agent-connection/types.js"; +import type { + AgentConnection, + AgentConnectionRlmChildAgentSnapshot, + AgentConnectionSessionEvent, +} from "../agent-connection/types.js"; import { latestAutonomousGateAttempt } from "../headless-completion.js"; import { type AcpEventMappingState, acpUpdatesForSessionEvent } from "./acp-events.js"; -import { primeAgentMeta } from "./acp-meta.js"; +import { PRIME_AGENT_META_NAMESPACE, type PrimeAgentAutonomousMeta, primeAgentMeta } from "./acp-meta.js"; import { type AcpStopReason, acpStopReason } from "./acp-stop-reason.js"; /** @@ -94,12 +98,119 @@ export interface AcpModeOptions { stream?: ReturnType; /** Skip claiming stdout when the caller supplies its own transport. */ ownStdout?: boolean; + /** Test seam for deterministically gating a serialized outbound update. */ + beforeAcpUpdatePublish?: (update: Record) => Promise | void; } interface AcpSessionEntry { id: string; abort: AbortController | undefined; unsubscribe: (() => void) | undefined; + producer: AcpUpdateProducer; +} + +/** + * The sole producer of ACP session updates for one ACP session. + * + * ACP notifications are asynchronous, so assigning an id at each call site is + * insufficient: detached calls can be observed out of order. This producer + * serializes publication and stamps the *delivered* order. Its phase/outcome + * fields are application metadata, deliberately independent of ACP stop + * reasons such as `end_turn`. + */ +class AcpUpdateProducer { + private eventSequence = 0; + private nextPromptTurnId = 0; + private activePromptTurnId = 0; + private tail: Promise = Promise.resolve(); + private readonly childOriginTurnIds = new Map(); + private readonly terminalTurns = new Set(); + + constructor( + private readonly sessionId: string, + private readonly client: { notify(method: unknown, params: unknown): Promise }, + private readonly beforePublish?: (update: Record) => Promise | void, + ) {} + + beginPrompt(): number { + this.activePromptTurnId = ++this.nextPromptTurnId; + return this.activePromptTurnId; + } + + finishPrompt(turnId: number): void { + if (this.activePromptTurnId === turnId) this.activePromptTurnId = 0; + } + + /** + * Cut a scoreable terminal boundary before it is queued. A subscription + * callback after this point is connection-scoped, never appended to a turn + * that an evaluator may treat as terminal. + */ + sealTerminal(turnId: number): void { + this.terminalTurns.add(turnId); + this.finishPrompt(turnId); + } + + isTerminalSealed(turnId: number): boolean { + return this.terminalTurns.has(turnId); + } + + turnForEvent(event: AgentConnectionSessionEvent): number { + if (event.type === "rlm_child_update") { + const known = this.childOriginTurnIds.get(event.child.id); + if (known !== undefined) return this.terminalTurns.has(known) ? 0 : known; + // A child is first observed while its parent prompt is producing it. + // Remember that origin so a later child update cannot be relabelled by + // a subsequent prompt. + if (this.activePromptTurnId !== 0) this.childOriginTurnIds.set(event.child.id, this.activePromptTurnId); + return this.activePromptTurnId; + } + return this.activePromptTurnId; + } + + publish( + update: Record, + turnId: number, + phase: "event" | "responseBoundary" | "terminalQuiescence", + outcome?: "result" | "error", + ): Promise { + const eventSequence = ++this.eventSequence; + const priorMeta = (update._meta && typeof update._meta === "object" ? update._meta : {}) as Record< + string, + unknown + >; + const priorPrimeMeta = + priorMeta[PRIME_AGENT_META_NAMESPACE] && typeof priorMeta[PRIME_AGENT_META_NAMESPACE] === "object" + ? (priorMeta[PRIME_AGENT_META_NAMESPACE] as Record) + : {}; + const correlatedUpdate = { + ...update, + _meta: { + ...priorMeta, + [PRIME_AGENT_META_NAMESPACE]: { + ...priorPrimeMeta, + promptTurnId: turnId, + eventSequence, + phase, + ...(outcome ? { outcome } : {}), + }, + }, + }; + // Keep the chain alive after a disconnect, while preserving the order of + // every later notification and allowing callers to await its drain. + this.tail = this.tail.then(async () => { + await this.beforePublish?.(correlatedUpdate); + await this.client + .notify(acp.methods.client.session.update, { sessionId: this.sessionId, update: correlatedUpdate }) + .then(() => undefined) + .catch(() => undefined); + }); + return this.tail; + } + + drain(): Promise { + return this.tail; + } } /** @@ -137,18 +248,32 @@ function promptContent(blocks: readonly unknown[]): { text: string; images: Imag return { text: texts.join("\n"), images }; } -function autonomousMeta(status: AgentAutonomousStatus | undefined): Record | undefined { +function autonomousMeta(status: AgentAutonomousStatus | undefined): PrimeAgentAutonomousMeta | undefined { if (!status?.enabled) return undefined; - return primeAgentMeta({ - autonomous: { - enabled: status.enabled, - continuationsUsed: status.continuationsUsed, - turnsUsed: status.turnsUsed, - tokensUsed: status.tokensUsed, - gateAttempt: latestAutonomousGateAttempt(status) || undefined, - gateFailure: status.lastGateFailure?.exitText, - }, - }); + return { + enabled: status.enabled, + continuationsUsed: status.continuationsUsed, + turnsUsed: status.turnsUsed, + tokensUsed: status.tokensUsed, + gateAttempt: latestAutonomousGateAttempt(status) || undefined, + gateFailure: status.lastGateFailure?.exitText, + }; +} + +function outstandingSubagentCount(children: readonly AgentConnectionRlmChildAgentSnapshot[] | undefined): number { + return (children ?? []).filter((child) => child.status === "queued" || child.status === "running").length; +} + +function quiescenceMeta( + status: AgentAutonomousStatus, + children: readonly AgentConnectionRlmChildAgentSnapshot[] | undefined, +): { outstandingSubagents: number; remainingAutonomousContinuations: number } { + return { + outstandingSubagents: outstandingSubagentCount(children), + remainingAutonomousContinuations: status.enabled + ? Math.max(0, status.limits.maxContinuations - status.continuationsUsed) + : 0, + }; } /** @@ -248,6 +373,7 @@ export async function runAcpModeWithConnection( // session keeps every event unambiguously attributable; a second session/new // is refused rather than silently sharing conversation state, cwd, and queues. let session: AcpSessionEntry | undefined; + let sessionNewInFlight = false; let bound = false; const stream = @@ -270,115 +396,184 @@ export async function runAcpModeWithConnection( _meta: primeAgentMeta({}), })) .onRequest("session/new", async (ctx: any) => { - if (!bound) { - // Only latch after a successful bind: a rejected bind must not leave - // extensions permanently unavailable for the rest of the process. - await options.bindHeadlessExtensions?.(); - bound = true; - } - if (session) { + // Reserve the single-session slot before the first await. Otherwise two + // concurrent requests can both pass the empty-slot check while cwd or + // snapshot reads are in flight, then overwrite each other's session. + if (session || sessionNewInFlight) { throw new Error( "prime-agent ACP mode hosts one session per connection; " + "start another prime-agent process for a second session", ); } - // prime-agent's cwd is fixed at startup by the session it was launched - // with, so a client-supplied cwd cannot be adopted after the fact. - // Report the real cwd back in `_meta` rather than failing the request or - // letting the client assume a directory the agent is not using. - const requestedCwd = (ctx.params as { cwd?: unknown } | undefined)?.cwd; - let cwdMismatch: { requested: string; actual: string } | undefined; - if (typeof requestedCwd === "string" && requestedCwd.length > 0) { - const actual = await connection - .getState() - .then((state) => state.cwd) - .catch(() => undefined); - if (actual && !sameCwd(requestedCwd, actual)) { - cwdMismatch = { requested: requestedCwd, actual }; + sessionNewInFlight = true; + try { + if (!bound) { + // Only latch after a successful bind: a rejected bind must not leave + // extensions permanently unavailable for the rest of the process. + await options.bindHeadlessExtensions?.(); + bound = true; } - } - const sessionId = randomUUID(); - const entry: AcpSessionEntry = { id: sessionId, abort: undefined, unsubscribe: undefined }; - // Subscribe for the session lifetime, not per prompt turn: prime-agent - // subagents are fire-and-forget and keep reporting after the spawning turn - // ends, so a turn-scoped subscription would drop their updates. One - // mapping state per session keeps streaming bash output correlated with - // the run that produced it. - const mappingState: AcpEventMappingState = {}; - const unsubscribe = connection.subscribe((event) => { - const notify = (update: Record) => - void ctx.client.notify(acp.methods.client.session.update, { sessionId, update }).catch(() => undefined); - // Heartbeats and cron schedules are connection-level rather than - // session events, but they drive the long-running work an ACP client - // most needs to observe. - if (event.type === "heartbeats_changed") { - notify({ sessionUpdate: "session_info_update", _meta: primeAgentMeta({ heartbeatsChanged: true }) }); - return; + // prime-agent's cwd is fixed at startup by the session it was launched + // with, so a client-supplied cwd cannot be adopted after the fact. + // Report the real cwd back in `_meta` rather than failing the request or + // letting the client assume a directory the agent is not using. + const requestedCwd = (ctx.params as { cwd?: unknown } | undefined)?.cwd; + let cwdMismatch: { requested: string; actual: string } | undefined; + if (typeof requestedCwd === "string" && requestedCwd.length > 0) { + const actual = await connection + .getState() + .then((state) => state.cwd) + .catch(() => undefined); + if (actual && !sameCwd(requestedCwd, actual)) { + cwdMismatch = { requested: requestedCwd, actual }; + } } - if (event.type !== "session_event") return; - for (const update of acpUpdatesForSessionEvent(event.event, mappingState)) { - notify(update); + const sessionId = randomUUID(); + // Install the listener before fetching the snapshot. Child updates can arrive + // while the snapshot request is in flight; the connection remains the + // authoritative source used when quiescence is emitted below. + const producer = new AcpUpdateProducer(sessionId, ctx.client, options.beforeAcpUpdatePublish); + const entry: AcpSessionEntry = { id: sessionId, abort: undefined, unsubscribe: undefined, producer }; + // Subscribe for the session lifetime, not per prompt turn: prime-agent + // subagents are fire-and-forget and keep reporting after the spawning turn + // ends, so a turn-scoped subscription would drop their updates. One + // mapping state per session keeps streaming bash output correlated with + // the run that produced it. + const mappingState: AcpEventMappingState = {}; + const unsubscribe = connection.subscribe((event) => { + // Heartbeats are connection-scoped, including if one races a prompt. + // They therefore intentionally use origin turn 0. + if (event.type === "heartbeats_changed") { + void producer.publish( + { sessionUpdate: "session_info_update", _meta: primeAgentMeta({ heartbeatsChanged: true }) }, + 0, + "event", + ); + return; + } + if (event.type !== "session_event") return; + const turnId = producer.turnForEvent(event.event); + for (const update of acpUpdatesForSessionEvent(event.event, mappingState)) { + void producer.publish(update, turnId, "event"); + } + }); + try { + // Reconcile after subscribing so updates cannot be lost while the snapshot + // request is in flight. Do not turn a failed read into an empty roster. + await connection.getInitialSnapshot(); + } catch (error) { + // A failed setup never claims the session slot, but it must still release + // the listener installed above. + unsubscribe(); + throw error; } - }); - // Claim the single-session slot only once the subscription exists, so a - // failed subscribe cannot leave the slot occupied and unusable. - entry.unsubscribe = unsubscribe; - session = entry; - return { - sessionId, - ...(cwdMismatch ? { _meta: primeAgentMeta({ cwd: cwdMismatch }) } : {}), - }; + // Claim the single-session slot only once the subscription and snapshot are + // ready, so a failed setup cannot leave it occupied and unusable. + entry.unsubscribe = unsubscribe; + session = entry; + return { + sessionId, + ...(cwdMismatch ? { _meta: primeAgentMeta({ cwd: cwdMismatch }) } : {}), + }; + } finally { + sessionNewInFlight = false; + } }) .onRequest("session/prompt", async (ctx: any) => { const params = ctx.params as { sessionId: string; prompt: readonly unknown[] }; const entry = session?.id === params.sessionId ? session : undefined; if (!entry) throw new Error(`Unknown ACP session: ${params.sessionId}`); + if (entry.abort) throw new Error("A prompt turn is already running for this ACP session"); - // ACP allows one turn at a time per session. Refuse a concurrent prompt - // rather than overwriting the running turn's controller, which would make - // the live turn uncancellable and let the loser's cleanup clear it. - if (entry.abort) { - throw new Error("A prompt turn is already running for this ACP session"); - } const abort = new AbortController(); entry.abort = abort; - + // Allocate the causal turn before the first await, not when an update is + // delivered. This prevents late producer events becoming the next turn. + const promptTurnId = entry.producer.beginPrompt(); + let responseBoundaryEmitted = false; try { const { text, images } = promptContent(params.prompt); - // Only this turn's messages may decide its outcome, and compaction can - // rebuild the transcript mid-turn, so record the pre-turn messages - // themselves rather than how many there were. const priorMessages = turnBoundary(await connection.getMessages()); + if (abort.signal.aborted) return { stopReason: "cancelled" satisfies AcpStopReason }; await connection.promptAndWait(text, images.length > 0 ? { images } : undefined); - // Autonomous gates continue inside this same prompt turn: the turn is - // only over once the gate loop settles. - const status = await connection.waitForHeadlessCompletion(); - const meta = autonomousMeta(status); - if (meta) { - await ctx.client - .notify(acp.methods.client.session.update, { - sessionId: params.sessionId, - update: { sessionUpdate: "session_info_update", _meta: meta }, - }) - .catch(() => undefined); + if (abort.signal.aborted) { + await entry.producer.drain(); + return { stopReason: "cancelled" satisfies AcpStopReason }; } - // A turn that failed (provider error, auth, no usable model) must not be - // reported as a clean end_turn. Print mode surfaces - // `stopReason: "error"` with its errorMessage; ACP previously dropped - // that and answered end_turn with no updates at all, which reads to a - // client as a successful but empty turn. const failure = await turnFailure(connection, priorMessages); - if (failure && !abort.signal.aborted) { - throw new Error(`prime-agent turn failed: ${failure}`); + if (abort.signal.aborted) { + await entry.producer.drain(); + return { stopReason: "cancelled" satisfies AcpStopReason }; } - return { stopReason: acpStopReason({ cancelled: abort.signal.aborted, autonomous: status }) }; - } catch (error) { - // Cancellation is a normal ACP prompt outcome, not a JSON-RPC error. + // Both successful and failed model turns must establish quiescence from + // the same authoritative sources before making a terminal claim. + const status = await connection.waitForHeadlessCompletion(); + if (abort.signal.aborted) { + await entry.producer.drain(); + return { stopReason: "cancelled" satisfies AcpStopReason }; + } + const autonomous = autonomousMeta(status); + const liveSnapshot = await connection.getInitialSnapshot(); + if (abort.signal.aborted) { + await entry.producer.drain(); + return { stopReason: "cancelled" satisfies AcpStopReason }; + } + const outcome = failure ? "error" : "result"; + const terminalQuiescence = quiescenceMeta(status, liveSnapshot.children); + const terminal = + terminalQuiescence.outstandingSubagents === 0 && + terminalQuiescence.remainingAutonomousContinuations === 0; + // The zero-terminal completion pair linearizes here, before either + // notification is queued. A cancellation before this cut produces no + // boundary/terminal; a cancellation after it cannot relabel a durable + // response+terminal pair as a cancelled prompt. if (abort.signal.aborted) return { stopReason: "cancelled" satisfies AcpStopReason }; + if (terminal) entry.producer.sealTerminal(promptTurnId); + await entry.producer.publish( + { sessionUpdate: "session_info_update", _meta: primeAgentMeta({}) }, + promptTurnId, + "responseBoundary", + outcome, + ); + responseBoundaryEmitted = true; + await entry.producer.publish( + { + sessionUpdate: "session_info_update", + _meta: primeAgentMeta({ ...(autonomous ? { autonomous } : {}), quiescence: terminalQuiescence }), + }, + promptTurnId, + terminal ? "terminalQuiescence" : "event", + terminal ? outcome : undefined, + ); + await entry.producer.drain(); + if (failure) throw new Error(`prime-agent turn failed: ${failure}`); + // A cancellation after the zero-terminal cut cannot relabel the durable + // boundary/terminal pair as cancellation. + return { + stopReason: acpStopReason({ + cancelled: abort.signal.aborted && !entry.producer.isTerminalSealed(promptTurnId), + autonomous: status, + }), + }; + } catch (error) { + if (abort.signal.aborted && !entry.producer.isTerminalSealed(promptTurnId)) { + await entry.producer.drain(); + return { stopReason: "cancelled" satisfies AcpStopReason }; + } + // Failed prompt/snapshot admission gets one correlated error boundary; + // it never gets an invented terminal-quiescence update. + if (!responseBoundaryEmitted) { + await entry.producer.publish( + { sessionUpdate: "session_info_update", _meta: primeAgentMeta({}) }, + promptTurnId, + "responseBoundary", + "error", + ); + } + await entry.producer.drain(); throw error; } finally { - // Only clear our own controller: a later turn must not be cleared by - // an earlier one unwinding. + entry.producer.finishPrompt(promptTurnId); if (entry.abort === abort) entry.abort = undefined; } }) diff --git a/packages/coding-agent/src/modes/agent-connection/daemon-agent-connection.ts b/packages/coding-agent/src/modes/agent-connection/daemon-agent-connection.ts index 016a05327..7afef59de 100644 --- a/packages/coding-agent/src/modes/agent-connection/daemon-agent-connection.ts +++ b/packages/coding-agent/src/modes/agent-connection/daemon-agent-connection.ts @@ -65,6 +65,7 @@ import type { AgentConnectionQueueMode, AgentConnectionQueueState, AgentConnectionResourceSnapshot, + AgentConnectionRlmChildAgentSnapshot, AgentConnectionSavedSessionInfo, AgentConnectionSavedSessionScope, AgentConnectionScopedModel, @@ -1488,6 +1489,9 @@ export class DaemonAgentConnection implements AgentConnection { if (message.event.type !== "refine_complete" && message.event.type !== "refine_failed") { this.observeStreamingMessage(message.event); } + if (message.event.type === "rlm_child_update") { + this.observeRlmChildUpdate(message.event.child); + } this.latestSnapshotIsFresh = false; await this.emit({ type: "session_event", event: message.event }); return; @@ -1914,6 +1918,19 @@ export class DaemonAgentConnection implements AgentConnection { } } + private observeRlmChildUpdate(child: AgentConnectionRlmChildAgentSnapshot): void { + if (!this.latestSnapshot) return; + const children = this.latestSnapshot.children ?? []; + const index = children.findIndex((candidate) => candidate.id === child.id); + const updatedChildren = [...children]; + if (index === -1) { + updatedChildren.push(child); + } else { + updatedChildren[index] = child; + } + this.latestSnapshot = { ...this.latestSnapshot, children: updatedChildren }; + } + private observeStreamingMessage(event: AgentSessionEvent): void { if (!this.latestSnapshot) { return; diff --git a/packages/coding-agent/src/modes/agent-connection/snapshot.ts b/packages/coding-agent/src/modes/agent-connection/snapshot.ts index 2c397645e..556fb20d5 100644 --- a/packages/coding-agent/src/modes/agent-connection/snapshot.ts +++ b/packages/coding-agent/src/modes/agent-connection/snapshot.ts @@ -74,6 +74,7 @@ export function createAgentConnectionSnapshot( tree: sessionManager.getTree(), leafId: sessionManager.getLeafId(), }, + children: session.getRlmChildSnapshots(), }; } diff --git a/packages/coding-agent/test/agent-connection-daemon.test.ts b/packages/coding-agent/test/agent-connection-daemon.test.ts index 7ff21e88c..572ce3020 100644 --- a/packages/coding-agent/test/agent-connection-daemon.test.ts +++ b/packages/coding-agent/test/agent-connection-daemon.test.ts @@ -600,6 +600,7 @@ interface CreateAttachResultOptions { omitSessionContext?: boolean; sessionTree?: DaemonAttachResult["snapshot"]["sessionTree"]; parent?: DaemonAttachResult["snapshot"]["parent"]; + children?: DaemonAttachResult["snapshot"]["children"]; replay?: DaemonAttachResult["replay"]; } @@ -654,6 +655,7 @@ function createAttachResult( lastEventSequence, lastEventCursor, ...(options.parent ? { parent: options.parent } : {}), + ...(options.children ? { children: options.children } : {}), }, replay: options.replay ?? { status: "complete", @@ -2139,6 +2141,40 @@ describe("DaemonAgentConnection", () => { }); }); + it("keeps the live child roster after an empty attach snapshot", async () => { + const fakeClient = new FakeDaemonClient(); + fakeClient.attachResultFactory = (command) => + createAttachResult(command.activeSessionId, command.clientId, command.capabilities, 12, { children: [] }); + const connection = new DaemonAgentConnection(asDaemonClient(fakeClient), "active-1"); + + await connection.attach(); + fakeClient.emitMessage({ + type: "session_event", + activeSessionId: "active-1", + event: { + type: "rlm_child_update", + child: { + id: "child-live", + label: "live child", + status: "running", + sessionDir: "/tmp/child-live", + }, + }, + meta: { + id: "active-1:13", + protocol: DAEMON_PROTOCOL_INFO, + activeSessionId: "active-1", + sequence: 13, + cursor: { generation: "generation-active-1", sequence: 13 }, + emittedAt: "2026-01-01T00:00:00.000Z", + }, + }); + + await expect(connection.getInitialSnapshot()).resolves.toMatchObject({ + children: [expect.objectContaining({ id: "child-live", status: "running" })], + }); + }); + it("refreshes initial snapshots after live events make the cached snapshot stale", async () => { const fakeClient = new FakeDaemonClient(); fakeClient.attachResultFactory = (command) => diff --git a/packages/coding-agent/test/agent-connection-in-process.test.ts b/packages/coding-agent/test/agent-connection-in-process.test.ts index be9054a09..beb868cb0 100644 --- a/packages/coding-agent/test/agent-connection-in-process.test.ts +++ b/packages/coding-agent/test/agent-connection-in-process.test.ts @@ -108,6 +108,7 @@ function createFakeSession(id: string, messages: AgentMessage[]): FakeSessionCon getActiveToolNames: () => ["ipython"], getContextUsage: () => undefined, cancelRlmChildRun: (childId: string) => childId === "child-1", + getRlmChildSnapshots: () => [], getToolDefinition: (toolName: string) => ({ name: toolName, label: toolName, diff --git a/packages/coding-agent/test/fixtures/acp-correlation-transcripts.json b/packages/coding-agent/test/fixtures/acp-correlation-transcripts.json new file mode 100644 index 000000000..208e9ca1e --- /dev/null +++ b/packages/coding-agent/test/fixtures/acp-correlation-transcripts.json @@ -0,0 +1,86 @@ +{ + "schema": "ai.primeintellect.prime-agent/v1", + "notes": "Exact JSON rendering of V3 acp_correlation_transcripts fixture; end_turn is deliberately absent because transport stop reasons are never causal.", + "cases": { + "success": [ + { + "promptTurnId": 1, + "eventSequence": 11, + "phase": "event", + "kind": "progress" + }, + { + "promptTurnId": 1, + "eventSequence": 12, + "phase": "responseBoundary", + "outcome": "result" + }, + { + "promptTurnId": 1, + "eventSequence": 13, + "phase": "terminalQuiescence", + "outcome": "result", + "quiescence": { + "outstandingSubagents": 0, + "remainingAutonomousContinuations": 0 + } + } + ], + "error_terminal": [ + { + "promptTurnId": 1, + "eventSequence": 21, + "phase": "responseBoundary", + "outcome": "error" + }, + { + "promptTurnId": 1, + "eventSequence": 22, + "phase": "terminalQuiescence", + "outcome": "error", + "quiescence": { + "outstandingSubagents": 0, + "remainingAutonomousContinuations": 0 + } + } + ], + "error_incomplete": [ + { + "promptTurnId": 1, + "eventSequence": 31, + "phase": "responseBoundary", + "outcome": "error" + } + ], + "cancelled": [], + "late_child": [ + { + "promptTurnId": 1, + "eventSequence": 41, + "phase": "event", + "child": { + "id": "late", + "status": "done" + } + } + ], + "global_sequence_turn_two": [ + { + "promptTurnId": 2, + "eventSequence": 51, + "phase": "responseBoundary", + "outcome": "result" + }, + { + "promptTurnId": 2, + "eventSequence": 52, + "phase": "terminalQuiescence", + "outcome": "result", + "quiescence": { + "outstandingSubagents": 0, + "remainingAutonomousContinuations": 0 + } + } + ] + } +} diff --git a/packages/coding-agent/test/suite/acp-mode.test.ts b/packages/coding-agent/test/suite/acp-mode.test.ts index 61b1c6252..b2eca01f8 100644 --- a/packages/coding-agent/test/suite/acp-mode.test.ts +++ b/packages/coding-agent/test/suite/acp-mode.test.ts @@ -1,6 +1,8 @@ +import { readFileSync } from "node:fs"; +import { resolve } from "node:path"; import * as acp from "@agentclientprotocol/sdk"; import { fauxAssistantMessage } from "@earendil-works/pi-ai"; -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import type { AgentSessionRuntime } from "../../src/core/agent-session-runtime.js"; import { PRIME_AGENT_META_NAMESPACE } from "../../src/modes/acp/acp-meta.js"; import { runAcpModeWithConnection } from "../../src/modes/acp/index.js"; @@ -29,7 +31,69 @@ interface ClientHarness { close: () => void; } -function connectAcpClient(connection: any): ClientHarness { +function fakeAcpConnection( + options: { + initialSnapshot?: () => Promise; + finalSnapshot?: () => Promise; + onInitialSnapshot?: (subscribed: boolean) => void; + onPromptAndWait?: () => void | Promise; + onWaitForHeadlessCompletion?: () => void | Promise; + onFinalSnapshot?: () => void | Promise; + onUnsubscribe?: () => void; + } = {}, +): any { + let listener: ((event: any) => void) | undefined; + let subscribed = false; + const snapshot = { state: { cwd: process.cwd() }, messages: [] }; + return { + subscribe(callback: (event: any) => void) { + subscribed = true; + listener = callback; + return () => { + listener = undefined; + options.onUnsubscribe?.(); + }; + }, + getState: async () => snapshot.state, + getMessages: async () => [], + getInitialSnapshot: async () => { + if (options.onInitialSnapshot) options.onInitialSnapshot(subscribed); + if (options.initialSnapshot) { + const result = await options.initialSnapshot(); + options.initialSnapshot = undefined; + return result; + } + if (options.finalSnapshot) { + await options.onFinalSnapshot?.(); + return options.finalSnapshot(); + } + return snapshot; + }, + promptAndWait: async () => { + await options.onPromptAndWait?.(); + }, + dispose: async () => {}, + abort: async () => {}, + waitForHeadlessCompletion: async () => { + await options.onWaitForHeadlessCompletion?.(); + return { + enabled: false, + continuationsUsed: 0, + turnsUsed: 0, + tokensUsed: 0, + limits: { maxContinuations: 0 }, + }; + }, + emitChild(child: any) { + listener?.({ type: "session_event", event: { type: "rlm_child_update", child } }); + }, + emitHeartbeat() { + listener?.({ type: "heartbeats_changed" }); + }, + }; +} + +function connectAcpClient(connection: any, options: Record = {}): ClientHarness { // Two web streams crossed over: agent's stdout is the client's stdin. const toAgent = new TransformStream(); const toClient = new TransformStream(); @@ -38,7 +102,7 @@ function connectAcpClient(connection: any): ClientHarness { const clientStream = acp.ndJsonStream(toAgent.writable, toClient.readable); const updates: any[] = []; - void runAcpModeWithConnection(connection, { stream: agentStream } as any); + void runAcpModeWithConnection(connection, { stream: agentStream, ...options } as any); const handle = acp .client({ name: "test-client" }) @@ -84,4 +148,505 @@ describe("ACP mode end to end", () => { harness.cleanup(); }, 30_000); + + it("emits score-safe quiescence metadata with outstanding work and budget", async () => { + const harness = await createHarness(); + harness.setResponses([fauxAssistantMessage("done")]); + const connection = new InProcessAgentConnection(runtimeHostFor(harness.session)); + const { client, updates } = connectAcpClient(connection); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const session = await client.request("session/new", { cwd: harness.tempDir, mcpServers: [] }); + await client.request("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "finish" }], + }); + const correlated = updates + .map((update) => update.update?._meta?.[PRIME_AGENT_META_NAMESPACE]) + .filter((meta) => meta?.promptTurnId === 1); + expect(correlated.map((meta) => meta.eventSequence)).toEqual( + [...correlated.map((meta) => meta.eventSequence)].sort((a, b) => a - b), + ); + expect(new Set(correlated.map((meta) => meta.eventSequence)).size).toBe(correlated.length); + expect(correlated.filter((meta) => meta.phase === "responseBoundary")).toEqual([ + expect.objectContaining({ outcome: "result" }), + ]); + const terminalIndex = correlated.findIndex((meta) => meta.phase === "terminalQuiescence"); + expect(terminalIndex).toBeGreaterThan(correlated.findIndex((meta) => meta.phase === "responseBoundary")); + expect(correlated[terminalIndex].quiescence).toEqual({ + outstandingSubagents: 0, + remainingAutonomousContinuations: 0, + }); + harness.cleanup(); + }, 30_000); + + it("reports a live in-process child that spawned after ACP attached", async () => { + const harness = await createHarness({ rlmDepth: 0, rlmMaxDepth: 1 }); + let releaseChild!: () => void; + const childReleased = new Promise((resolve) => { + releaseChild = resolve; + }); + harness.setResponses([ + async () => { + await childReleased; + return fauxAssistantMessage("child done"); + }, + ]); + const connection = new InProcessAgentConnection(runtimeHostFor(harness.session)); + const { client, updates } = connectAcpClient(connection); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const session = await client.request("session/new", { cwd: harness.tempDir, mcpServers: [] }); + // The child is admitted after ACP attaches, so an attach-time snapshot was + // empty. Its run stays live while the independent parent prompt completes. + await harness.session.runRlmChild("continue in the background"); + harness.appendResponses([fauxAssistantMessage("parent done")]); + await client.request("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "finish the parent turn" }], + }); + const quiescence = updates.find((update) => update.update?._meta?.[PRIME_AGENT_META_NAMESPACE]?.quiescence); + const meta = quiescence?.update?._meta?.[PRIME_AGENT_META_NAMESPACE]; + expect(meta?.quiescence.outstandingSubagents).toBe(1); + // Outstanding child work is truthful progress, never scoreable terminal quiescence. + expect(meta?.phase).toBe("event"); + + releaseChild(); + await vi.waitFor(() => expect(harness.session.getRlmChildSnapshots()[0]?.status).toBe("done")); + harness.cleanup(); + }, 30_000); + + it("fails session creation when the initial roster snapshot fails", async () => { + const connection = fakeAcpConnection({ + initialSnapshot: async () => { + throw new Error("snapshot unavailable"); + }, + }); + const { client, close } = connectAcpClient(connection); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + await expect(client.request("session/new", { cwd: process.cwd(), mcpServers: [] })).rejects.toThrow(); + close(); + }); + + it("releases the subscription when the initial roster snapshot fails", async () => { + let unsubscribeCount = 0; + const connection = fakeAcpConnection({ + initialSnapshot: async () => { + throw new Error("snapshot unavailable"); + }, + onUnsubscribe: () => { + unsubscribeCount += 1; + }, + }); + const { client, close } = connectAcpClient(connection); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + await expect(client.request("session/new", { cwd: process.cwd(), mcpServers: [] })).rejects.toThrow(); + expect(unsubscribeCount).toBe(1); + close(); + }); + + it("rejects a concurrent session creation while the first snapshot is pending", async () => { + let enteredSnapshot!: () => void; + let releaseSnapshot!: () => void; + const snapshotEntered = new Promise((resolve) => { + enteredSnapshot = resolve; + }); + const snapshotReleased = new Promise((resolve) => { + releaseSnapshot = resolve; + }); + const connection = fakeAcpConnection({ + initialSnapshot: async () => { + enteredSnapshot(); + await snapshotReleased; + return { state: { cwd: process.cwd() }, messages: [], children: [] }; + }, + }); + const { client, close } = connectAcpClient(connection); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const first = client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + await snapshotEntered; + await expect(client.request("session/new", { cwd: process.cwd(), mcpServers: [] })).rejects.toThrow(); + releaseSnapshot(); + await expect(first).resolves.toMatchObject({ sessionId: expect.any(String) }); + close(); + }); + + it("subscribes before taking the initial roster snapshot", async () => { + let subscribedAtSnapshot: boolean | undefined; + const connection = fakeAcpConnection({ + onInitialSnapshot: (subscribed) => { + subscribedAtSnapshot = subscribed; + }, + }); + const { client, close } = connectAcpClient(connection); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + await client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + expect(subscribedAtSnapshot).toBe(true); + close(); + }); + + it("propagates a failed roster read at quiescence emission", async () => { + // A snapshot failure while emitting must not degrade to outstandingSubagents: 0. + // Reporting a fabricated zero is the false-quiescent answer this metadata + // exists to prevent: a consumer would score a turn whose children still run. + const connection = fakeAcpConnection({ + // `initialSnapshot` serves session/new and is then consumed; `finalSnapshot` + // serves the emission-time read, which is the one under test here. + initialSnapshot: async () => ({ state: { cwd: process.cwd() }, messages: [], children: [] }), + finalSnapshot: async () => { + throw new Error("roster unavailable"); + }, + }); + const { client, updates, close } = connectAcpClient(connection); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const session = await client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + await expect( + client.request("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "finish" }], + }), + ).rejects.toThrow(); + const metadata = updates.map((u) => u.update?._meta?.[PRIME_AGENT_META_NAMESPACE]).filter(Boolean); + expect(metadata).toContainEqual( + expect.objectContaining({ + promptTurnId: 1, + phase: "responseBoundary", + outcome: "error", + }), + ); + expect(metadata.find((meta) => meta.phase === "terminalQuiescence")).toBeUndefined(); + close(); + }); + + it("counts the authoritative live roster at quiescence emission", async () => { + const child = { id: "child-1", label: "child", status: "running", sessionDir: "/tmp/child" }; + const connection = fakeAcpConnection({ + initialSnapshot: async () => ({ state: { cwd: process.cwd() }, messages: [], children: [] }), + finalSnapshot: async () => ({ state: { cwd: process.cwd() }, messages: [], children: [child] }), + }); + const { client, updates, close } = connectAcpClient(connection); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const session = await client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + await client.request("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "finish" }], + }); + const quiescence = updates.find((u) => u.update?._meta?.[PRIME_AGENT_META_NAMESPACE]?.quiescence); + expect(quiescence.update._meta[PRIME_AGENT_META_NAMESPACE].quiescence.outstandingSubagents).toBe(1); + close(); + }); + + it("keeps global sequences and causal turn ids across sequential prompts", async () => { + const connection = fakeAcpConnection(); + const { client, updates, close } = connectAcpClient(connection); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const session = await client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + await client.request("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "first" }], + }); + await client.request("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "second" }], + }); + const metadata = updates.map((u) => u.update?._meta?.[PRIME_AGENT_META_NAMESPACE]).filter(Boolean); + const sequences = metadata.map((meta) => meta.eventSequence); + expect(sequences).toEqual([...sequences].sort((a, b) => a - b)); + expect(metadata.filter((meta) => meta.phase === "responseBoundary").map((meta) => meta.promptTurnId)).toEqual([ + 1, 2, + ]); + expect(metadata.filter((meta) => meta.phase === "terminalQuiescence").map((meta) => meta.promptTurnId)).toEqual([ + 1, 2, + ]); + close(); + }); + + it("moves a retained child update after terminal to safe turn zero", async () => { + const child = { id: "child-1", label: "child", status: "running", sessionDir: "/tmp/child" }; + let connection: any; + connection = fakeAcpConnection({ + onPromptAndWait: () => connection.emitChild(child), + }); + const { client, updates, close } = connectAcpClient(connection); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const session = await client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + await client.request("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "first" }], + }); + await client.request("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "second" }], + }); + connection.emitChild({ ...child, status: "done" }); + await vi.waitFor(() => + expect( + updates.some( + (u) => + u.update?.sessionUpdate === "session_info_update" && + u.update?._meta?.[PRIME_AGENT_META_NAMESPACE]?.subagents, + ), + ).toBe(true), + ); + const childUpdates = updates + .map((u) => u.update?._meta?.[PRIME_AGENT_META_NAMESPACE]) + .filter((meta) => meta?.subagents); + expect(childUpdates.map((meta) => meta.promptTurnId)).toEqual([1, 0]); + expect(childUpdates.map((meta) => meta.phase)).toEqual(["event", "event"]); + close(); + }); + + it("makes a child first observed after terminal connection-scoped", async () => { + const connection = fakeAcpConnection(); + const { client, updates, close } = connectAcpClient(connection); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const session = await client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + await client.request("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "complete" }], + }); + connection.emitChild({ id: "late", label: "late", status: "running", sessionDir: "/tmp/late" }); + await vi.waitFor(() => + expect(updates.some((u) => u.update?._meta?.[PRIME_AGENT_META_NAMESPACE]?.subagents)).toBe(true), + ); + const late = updates + .map((u) => u.update?._meta?.[PRIME_AGENT_META_NAMESPACE]) + .find((meta) => meta?.subagents?.[0]?.id === "late"); + expect(late).toMatchObject({ promptTurnId: 0, phase: "event" }); + close(); + }); + + it("cancels after prompt completion without emitting a result or terminal boundary", async () => { + let entered!: () => void; + let release!: () => void; + const enteredPrompt = new Promise((resolve) => { + entered = resolve; + }); + const releasePrompt = new Promise((resolve) => { + release = resolve; + }); + const connection = fakeAcpConnection({ + onPromptAndWait: async () => { + entered(); + await releasePrompt; + }, + }); + const { client, updates, close } = connectAcpClient(connection); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const session = await client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + const pending = client.request("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "cancel" }], + }); + await enteredPrompt; + await client.notify("session/cancel", { sessionId: session.sessionId }); + release(); + await expect(pending).resolves.toMatchObject({ stopReason: "cancelled" }); + const metadata = updates.map((u) => u.update?._meta?.[PRIME_AGENT_META_NAMESPACE]).filter(Boolean); + expect(metadata.some((meta) => meta.phase === "responseBoundary" && meta.outcome === "result")).toBe(false); + expect(metadata.some((meta) => meta.phase === "terminalQuiescence")).toBe(false); + close(); + }); + + it("cancels during headless completion without any completion envelope", async () => { + let entered!: () => void; + let release!: () => void; + const enteredWait = new Promise((resolve) => { + entered = resolve; + }); + const releaseWait = new Promise((resolve) => { + release = resolve; + }); + const connection = fakeAcpConnection({ + onWaitForHeadlessCompletion: async () => { + entered(); + await releaseWait; + }, + }); + const { client, updates, close } = connectAcpClient(connection); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const session = await client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + const pending = client.request("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "cancel in headless" }], + }); + await enteredWait; + await client.notify("session/cancel", { sessionId: session.sessionId }); + release(); + await expect(pending).resolves.toMatchObject({ stopReason: "cancelled" }); + const metadata = updates.map((u) => u.update?._meta?.[PRIME_AGENT_META_NAMESPACE]).filter(Boolean); + expect(metadata.some((meta) => meta.phase === "responseBoundary" || meta.phase === "terminalQuiescence")).toBe( + false, + ); + close(); + }); + + it("cancels during the authoritative final snapshot without any completion envelope", async () => { + let entered!: () => void; + let release!: () => void; + const enteredSnapshot = new Promise((resolve) => { + entered = resolve; + }); + const releaseSnapshot = new Promise((resolve) => { + release = resolve; + }); + const connection = fakeAcpConnection({ + initialSnapshot: async () => ({ state: { cwd: process.cwd() }, messages: [], children: [] }), + finalSnapshot: async () => ({ state: { cwd: process.cwd() }, messages: [], children: [] }), + onFinalSnapshot: async () => { + entered(); + await releaseSnapshot; + }, + }); + const { client, updates, close } = connectAcpClient(connection); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const session = await client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + const pending = client.request("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "cancel in snapshot" }], + }); + await enteredSnapshot; + await client.notify("session/cancel", { sessionId: session.sessionId }); + release(); + await expect(pending).resolves.toMatchObject({ stopReason: "cancelled" }); + const metadata = updates.map((u) => u.update?._meta?.[PRIME_AGENT_META_NAMESPACE]).filter(Boolean); + expect(metadata.some((meta) => meta.phase === "responseBoundary" || meta.phase === "terminalQuiescence")).toBe( + false, + ); + close(); + }); + + it("linearizes a cancellation during the response-boundary publish as a completed pair", async () => { + let entered!: () => void; + let release!: () => void; + const enteredBoundary = new Promise((resolve) => { + entered = resolve; + }); + const releaseBoundary = new Promise((resolve) => { + release = resolve; + }); + const connection = fakeAcpConnection(); + const { client, updates, close } = connectAcpClient(connection, { + beforeAcpUpdatePublish: async (update: any) => { + if (update._meta?.[PRIME_AGENT_META_NAMESPACE]?.phase === "responseBoundary") { + entered(); + await releaseBoundary; + } + }, + }); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const session = await client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + const pending = client.request("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "gate boundary" }], + }); + await enteredBoundary; + await client.notify("session/cancel", { sessionId: session.sessionId }); + release(); + const result = await pending; + expect(result.stopReason).not.toBe("cancelled"); + const meta = updates.map((u) => u.update?._meta?.[PRIME_AGENT_META_NAMESPACE]).filter(Boolean); + expect(meta.filter((item) => item.phase === "responseBoundary")).toEqual([ + expect.objectContaining({ outcome: "result" }), + ]); + expect(meta.filter((item) => item.phase === "terminalQuiescence")).toEqual([ + expect.objectContaining({ + outcome: "result", + quiescence: { outstandingSubagents: 0, remainingAutonomousContinuations: 0 }, + }), + ]); + close(); + }); + + it("linearizes a cancellation during terminal publish as a completed pair", async () => { + let entered!: () => void; + let release!: () => void; + const enteredTerminal = new Promise((resolve) => { + entered = resolve; + }); + const releaseTerminal = new Promise((resolve) => { + release = resolve; + }); + const connection = fakeAcpConnection(); + const { client, updates, close } = connectAcpClient(connection, { + beforeAcpUpdatePublish: async (update: any) => { + if (update._meta?.[PRIME_AGENT_META_NAMESPACE]?.phase === "terminalQuiescence") { + entered(); + await releaseTerminal; + } + }, + }); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const session = await client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + const pending = client.request("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "gate terminal" }], + }); + await enteredTerminal; + await client.notify("session/cancel", { sessionId: session.sessionId }); + release(); + const result = await pending; + expect(result.stopReason).not.toBe("cancelled"); + const meta = updates.map((u) => u.update?._meta?.[PRIME_AGENT_META_NAMESPACE]).filter(Boolean); + expect(meta.filter((item) => item.phase === "responseBoundary")).toEqual([ + expect.objectContaining({ outcome: "result" }), + ]); + expect(meta.filter((item) => item.phase === "terminalQuiescence")).toEqual([ + expect.objectContaining({ + outcome: "result", + quiescence: { outstandingSubagents: 0, remainingAutonomousContinuations: 0 }, + }), + ]); + close(); + }); + + it("matches the exact canonical offline transcript fixture consumed by V3", () => { + const fixture = JSON.parse( + readFileSync(resolve(import.meta.dirname, "../fixtures/acp-correlation-transcripts.json"), "utf8"), + ) as { cases: Record>> }; + expect(Object.keys(fixture.cases)).toEqual([ + "success", + "error_terminal", + "error_incomplete", + "cancelled", + "late_child", + "global_sequence_turn_two", + ]); + for (const [name, records] of Object.entries(fixture.cases)) { + for (let index = 1; index < records.length; index++) { + expect(records[index].eventSequence, `${name} sequence`).toBeGreaterThan( + records[index - 1].eventSequence as number, + ); + } + } + for (const name of ["success", "error_terminal", "global_sequence_turn_two"]) { + const records = fixture.cases[name]; + const boundary = records.find((record) => record.phase === "responseBoundary"); + const terminal = records.find((record) => record.phase === "terminalQuiescence"); + expect(boundary).toBeDefined(); + expect(terminal).toMatchObject({ + promptTurnId: boundary?.promptTurnId, + outcome: boundary?.outcome, + quiescence: { outstandingSubagents: 0, remainingAutonomousContinuations: 0 }, + }); + } + expect(fixture.cases.error_incomplete).toHaveLength(1); + expect(fixture.cases.cancelled).toEqual([]); + expect(fixture.cases.late_child).toEqual([ + expect.objectContaining({ promptTurnId: 1, phase: "event", child: { id: "late", status: "done" } }), + ]); + }); + + it("correlates connection-scoped heartbeats to turn zero", async () => { + const connection = fakeAcpConnection(); + const { client, updates, close } = connectAcpClient(connection); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + await client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + connection.emitHeartbeat(); + await vi.waitFor(() => expect(updates).toHaveLength(1)); + expect(updates[0].update._meta[PRIME_AGENT_META_NAMESPACE]).toMatchObject({ + promptTurnId: 0, + phase: "event", + eventSequence: 1, + }); + close(); + }); }); From c2c370a1de2d573fd19322e546a68ca00268b592 Mon Sep 17 00:00:00 2001 From: Seth Date: Tue, 11 Aug 2026 10:39:35 -0700 Subject: [PATCH 02/17] fix(acp): preserve terminal and admission lifecycle truth --- .../coding-agent/src/core/agent-session.ts | 4 +- .../coding-agent/src/modes/acp/acp-mode.ts | 118 ++++++++++++++++-- .../test/agent-session-recursion.test.ts | 41 ++++++ .../coding-agent/test/suite/acp-mode.test.ts | 65 ++++++++++ 4 files changed, 220 insertions(+), 8 deletions(-) diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index 201a4172c..6c9d4f0ee 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -9602,7 +9602,9 @@ export class AgentSession { sessionName: child.sessionName, model: child.model ? `${child.model.provider}/${child.model.id}` : undefined, label: child.sessionName ?? "child agent", - status: "done", + // A failed delete retains the session solely for cleanup retry. Preserve + // its cancellation truth in snapshots rather than reviving it as done. + status: this._rlmChildCleanupFailures.has(childId) ? "cancelled" : "done", sessionDir: child._rlmSessionDir ?? child.sessionManager.getSessionDir(), }); snapshots.push(...child.getRlmChildSnapshots()); diff --git a/packages/coding-agent/src/modes/acp/acp-mode.ts b/packages/coding-agent/src/modes/acp/acp-mode.ts index 2a207143b..7a8dc8864 100644 --- a/packages/coding-agent/src/modes/acp/acp-mode.ts +++ b/packages/coding-agent/src/modes/acp/acp-mode.ts @@ -68,6 +68,19 @@ function canonicalCwd(path: string): string { return normalizeWindowsDriveLetter(canonical); } +/** A response cannot be confused with a peer request that reuses a JSON-RPC id. */ +function isJsonRpcResponse(message: unknown, requestId: unknown): boolean { + if (typeof message !== "object" || message === null) return false; + const record = message as Record; + const has = (key: string): boolean => Object.prototype.hasOwnProperty.call(record, key); + return ( + record.jsonrpc === "2.0" && + record.id === requestId && + !has("method") && + (has("result") !== has("error")) + ); +} + function sameCwd(left: string, right: string): boolean { const canonicalLeft = canonicalCwd(left); const canonicalRight = canonicalCwd(right); @@ -125,12 +138,36 @@ class AcpUpdateProducer { private tail: Promise = Promise.resolve(); private readonly childOriginTurnIds = new Map(); private readonly terminalTurns = new Set(); + private readonly admissionReady: Promise; + private releaseAdmission!: () => void; + private admissionOpen = false; + private admissionClosed = false; constructor( private readonly sessionId: string, private readonly client: { notify(method: unknown, params: unknown): Promise }, private readonly beforePublish?: (update: Record) => Promise | void, - ) {} + ) { + // Subscribe before the initial snapshot, but do not let that subscription + // publish a session-bound update before session/new has replied. + this.admissionReady = new Promise((resolve) => { + this.releaseAdmission = resolve; + }); + } + + /** Open the initial update gate only after session/new reached the transport. */ + commitSessionNewResponse(): void { + if (this.admissionClosed) return; + this.admissionOpen = true; + this.releaseAdmission(); + } + + /** Settle a failed admission without allowing its buffered updates to publish. */ + failSessionNewAdmission(): void { + if (this.admissionOpen || this.admissionClosed) return; + this.admissionClosed = true; + this.releaseAdmission(); + } beginPrompt(): number { this.activePromptTurnId = ++this.nextPromptTurnId; @@ -199,6 +236,8 @@ class AcpUpdateProducer { // Keep the chain alive after a disconnect, while preserving the order of // every later notification and allowing callers to await its drain. this.tail = this.tail.then(async () => { + await this.admissionReady; + if (!this.admissionOpen) return; await this.beforePublish?.(correlatedUpdate); await this.client .notify(acp.methods.client.session.update, { sessionId: this.sessionId, update: correlatedUpdate }) @@ -376,8 +415,68 @@ export async function runAcpModeWithConnection( let sessionNewInFlight = false; let bound = false; - const stream = + const baseStream = options.stream ?? acp.ndJsonStream(rawStdoutSink(), Readable.toWeb(process.stdin) as ReadableStream); + // ACP's public request handler only returns a response; it has no response + // commit callback. Observe the outgoing response at the supplied stream + // boundary instead. The SDK serializes every write, so opening the producer + // after this write resolves puts buffered notifications strictly behind it. + let pendingSessionNewResponse: + | { requestId: unknown; producer: AcpUpdateProducer } + | undefined; + const failPendingSessionNewResponse = (): void => { + const admission = pendingSessionNewResponse; + pendingSessionNewResponse = undefined; + admission?.producer.failSessionNewAdmission(); + }; + const stream: typeof baseStream = { + readable: baseStream.readable, + writable: new WritableStream({ + async write(message) { + let writer: WritableStreamDefaultWriter | undefined; + try { + writer = baseStream.writable.getWriter(); + await writer.write(message); + } catch (error) { + failPendingSessionNewResponse(); + throw error; + } finally { + writer?.releaseLock(); + } + if (pendingSessionNewResponse && isJsonRpcResponse(message, pendingSessionNewResponse.requestId)) { + const admission = pendingSessionNewResponse; + pendingSessionNewResponse = undefined; + admission.producer.commitSessionNewResponse(); + } + }, + async close() { + let writer: WritableStreamDefaultWriter | undefined; + try { + writer = baseStream.writable.getWriter(); + await writer.close(); + } catch (error) { + failPendingSessionNewResponse(); + throw error; + } finally { + writer?.releaseLock(); + } + failPendingSessionNewResponse(); + }, + async abort(reason) { + let writer: WritableStreamDefaultWriter | undefined; + try { + writer = baseStream.writable.getWriter(); + await writer.abort(reason); + } catch (error) { + failPendingSessionNewResponse(); + throw error; + } finally { + writer?.releaseLock(); + } + failPendingSessionNewResponse(); + }, + }), + }; const handle = acp .agent({ name: "prime-agent" }) @@ -463,7 +562,8 @@ export async function runAcpModeWithConnection( await connection.getInitialSnapshot(); } catch (error) { // A failed setup never claims the session slot, but it must still release - // the listener installed above. + // the listener installed above and settle buffered producers. + producer.failSessionNewAdmission(); unsubscribe(); throw error; } @@ -471,10 +571,14 @@ export async function runAcpModeWithConnection( // ready, so a failed setup cannot leave it occupied and unusable. entry.unsubscribe = unsubscribe; session = entry; - return { + const response = { sessionId, ...(cwdMismatch ? { _meta: primeAgentMeta({ cwd: cwdMismatch }) } : {}), }; + // The stream wrapper commits this gate after this exact response has + // written. Buffered subscription updates retain producer order. + pendingSessionNewResponse = { requestId: ctx.requestId, producer: entry.producer }; + return response; } finally { sessionNewInFlight = false; } @@ -520,9 +624,9 @@ export async function runAcpModeWithConnection( } const outcome = failure ? "error" : "result"; const terminalQuiescence = quiescenceMeta(status, liveSnapshot.children); - const terminal = - terminalQuiescence.outstandingSubagents === 0 && - terminalQuiescence.remainingAutonomousContinuations === 0; + // Lifecycle completion is authoritative here: unused autonomous + // continuation capacity is telemetry, not outstanding work. + const terminal = terminalQuiescence.outstandingSubagents === 0; // The zero-terminal completion pair linearizes here, before either // notification is queued. A cancellation before this cut produces no // boundary/terminal; a cancellation after it cannot relabel a durable diff --git a/packages/coding-agent/test/agent-session-recursion.test.ts b/packages/coding-agent/test/agent-session-recursion.test.ts index 8003bba40..692d5a220 100644 --- a/packages/coding-agent/test/agent-session-recursion.test.ts +++ b/packages/coding-agent/test/agent-session-recursion.test.ts @@ -41,6 +41,7 @@ import { SettingsManager, type SettingsStorage } from "../src/core/settings-mana import type { Skill } from "../src/core/skills.js"; import { createSyntheticSourceInfo } from "../src/core/source-info.js"; import { type ActiveSessionState, resolveActiveSessionState } from "../src/modes/daemon/active-session-state.js"; +import { InProcessAgentConnection } from "../src/modes/agent-connection/in-process-agent-connection.js"; import { AgentDaemon } from "../src/modes/daemon/daemon-mode.js"; import { invokeHostRequest } from "./host-request-context.js"; import { createTestExtensionsResult, createTestResourceLoader } from "./utilities.js"; @@ -656,6 +657,46 @@ describe("AgentSession rlm recursion", () => { }); }); + it("keeps a failed runtime deletion cancelled in direct and connection snapshots until retry", async () => { + const childId = "cancelled-cleanup-child"; + const child = createSession({ rlmSessionDir: join(tempDir, childId) }); + child.setSessionName("cancelled-cleanup-worker"); + let deleteAttempts = 0; + const root = createSession({ + subagentRuntimeHost: { + createRlmSubagentRuntime: async () => ({ session: child }), + deleteRlmSubagentRuntime: async (_id, session) => { + if (++deleteAttempts === 1) throw new Error("runtime deletion failed"); + await session?.disposeAsync(); + }, + }, + }); + expect(root.registerRlmChildSession(childId, child)).toBe(true); + const events: AgentSessionEvent[] = []; + root.subscribe((event) => events.push(event)); + await expect(root.deleteRlmSubagent("cancelled-cleanup-worker")).rejects.toThrow("runtime deletion failed"); + expect(events).toContainEqual( + expect.objectContaining({ type: "rlm_child_update", child: { id: childId, status: "cancelled" } }), + ); + expect(root.getRlmChildSnapshots()).toEqual([ + expect.objectContaining({ id: childId, status: "cancelled" }), + ]); + const connection = new InProcessAgentConnection({ + session: root, + setRebindSession() {}, + setBeforeSessionInvalidate() {}, + dispose: async () => {}, + } as unknown as AgentSessionRuntime); + await expect(connection.getInitialSnapshot()).resolves.toMatchObject({ + children: [expect.objectContaining({ id: childId, status: "cancelled" })], + }); + await expect(root.deleteRlmSubagent("cancelled-cleanup-worker")).resolves.toMatchObject({ + subagent: { rlm_child_id: childId }, + }); + expect(deleteAttempts).toBe(2); + expect(root.getRlmChildSnapshots()).toEqual([]); + }); + it("makes an orchestrator-chosen name override a custom runtime's preexisting name", async () => { const hostedChild = createSession(); hostedChild.setSessionName("factory-assigned-name"); diff --git a/packages/coding-agent/test/suite/acp-mode.test.ts b/packages/coding-agent/test/suite/acp-mode.test.ts index b2eca01f8..306a0cb0c 100644 --- a/packages/coding-agent/test/suite/acp-mode.test.ts +++ b/packages/coding-agent/test/suite/acp-mode.test.ts @@ -38,6 +38,7 @@ function fakeAcpConnection( onInitialSnapshot?: (subscribed: boolean) => void; onPromptAndWait?: () => void | Promise; onWaitForHeadlessCompletion?: () => void | Promise; + headlessStatus?: Record; onFinalSnapshot?: () => void | Promise; onUnsubscribe?: () => void; } = {}, @@ -82,6 +83,7 @@ function fakeAcpConnection( turnsUsed: 0, tokensUsed: 0, limits: { maxContinuations: 0 }, + ...options.headlessStatus, }; }, emitChild(child: any) { @@ -179,6 +181,31 @@ describe("ACP mode end to end", () => { harness.cleanup(); }, 30_000); + it("treats unused autonomous capacity as terminal lifecycle telemetry", async () => { + const connection = fakeAcpConnection({ + headlessStatus: { + enabled: true, + continuationsUsed: 1, + limits: { maxContinuations: 3 }, + }, + }); + const { client, updates, close } = connectAcpClient(connection); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const session = await client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + await client.request("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "finish" }], + }); + const meta = updates.map((item) => item.update?._meta?.[PRIME_AGENT_META_NAMESPACE]).filter(Boolean); + expect(meta.filter((item) => item.phase === "responseBoundary")).toHaveLength(1); + expect(meta.filter((item) => item.phase === "terminalQuiescence")).toEqual([ + expect.objectContaining({ + quiescence: { outstandingSubagents: 0, remainingAutonomousContinuations: 2 }, + }), + ]); + close(); + }); + it("reports a live in-process child that spawned after ACP attached", async () => { const harness = await createHarness({ rlmDepth: 0, rlmMaxDepth: 1 }); let releaseChild!: () => void; @@ -269,6 +296,44 @@ describe("ACP mode end to end", () => { close(); }); + it("buffers subscription updates until the session/new response commits", async () => { + let emitChild: (child: any) => void = () => {}; + let releaseSnapshot!: () => void; + let snapshotEventEmitted!: () => void; + const snapshotReleased = new Promise((resolve) => { + releaseSnapshot = resolve; + }); + const snapshotEvent = new Promise((resolve) => { + snapshotEventEmitted = resolve; + }); + const connection = fakeAcpConnection({ + initialSnapshot: async () => { + emitChild({ id: "during-snapshot", label: "during snapshot", status: "running", sessionDir: "/tmp/child" }); + snapshotEventEmitted(); + await snapshotReleased; + return { state: { cwd: process.cwd() }, messages: [], children: [] }; + }, + }); + const originalSubscribe = connection.subscribe.bind(connection); + connection.subscribe = (listener: (event: any) => void) => { + emitChild = (child) => listener({ type: "session_event", event: { type: "rlm_child_update", child } }); + return originalSubscribe(listener); + }; + const { client, updates, close } = connectAcpClient(connection); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const pending = client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + await snapshotEvent; + expect(updates).toHaveLength(0); + releaseSnapshot(); + const session = await pending; + await vi.waitFor(() => expect(updates).toHaveLength(1)); + expect(updates[0]).toMatchObject({ + sessionId: session.sessionId, + update: { _meta: { [PRIME_AGENT_META_NAMESPACE]: { eventSequence: 1, promptTurnId: 0 } } }, + }); + close(); + }); + it("subscribes before taking the initial roster snapshot", async () => { let subscribedAtSnapshot: boolean | undefined; const connection = fakeAcpConnection({ From 452916536edd8573a313546448186b2ba2514389 Mon Sep 17 00:00:00 2001 From: Seth Date: Tue, 11 Aug 2026 10:42:12 -0700 Subject: [PATCH 03/17] test(acp): type lifecycle regression fixtures --- .../coding-agent/test/agent-session-recursion.test.ts | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/packages/coding-agent/test/agent-session-recursion.test.ts b/packages/coding-agent/test/agent-session-recursion.test.ts index 692d5a220..9e2b7e499 100644 --- a/packages/coding-agent/test/agent-session-recursion.test.ts +++ b/packages/coding-agent/test/agent-session-recursion.test.ts @@ -19,7 +19,8 @@ import { createAgentSessionMessage, isAgentSessionMessage, } from "../src/core/agent-messages.js"; -import { AgentSession } from "../src/core/agent-session.js"; +import { AgentSession, type AgentSessionEvent } from "../src/core/agent-session.js"; +import type { AgentSessionRuntime } from "../src/core/agent-session-runtime.js"; import { AuthStorage } from "../src/core/auth-storage.js"; import type { LoadExtensionsResult } from "../src/core/extensions/index.js"; import { @@ -676,7 +677,10 @@ describe("AgentSession rlm recursion", () => { root.subscribe((event) => events.push(event)); await expect(root.deleteRlmSubagent("cancelled-cleanup-worker")).rejects.toThrow("runtime deletion failed"); expect(events).toContainEqual( - expect.objectContaining({ type: "rlm_child_update", child: { id: childId, status: "cancelled" } }), + expect.objectContaining({ + type: "rlm_child_update", + child: { id: childId, status: "cancelled" }, + }), ); expect(root.getRlmChildSnapshots()).toEqual([ expect.objectContaining({ id: childId, status: "cancelled" }), From 46613b4f8cab3d749ff9c844f43b8be5605efc93 Mon Sep 17 00:00:00 2001 From: Seth Date: Tue, 11 Aug 2026 10:46:08 -0700 Subject: [PATCH 04/17] fix(acp): stabilize lifecycle regressions --- packages/coding-agent/src/modes/acp/acp-mode.ts | 13 +++---------- .../test/agent-session-recursion.test.ts | 8 +++----- packages/coding-agent/test/suite/acp-mode.test.ts | 1 + 3 files changed, 7 insertions(+), 15 deletions(-) diff --git a/packages/coding-agent/src/modes/acp/acp-mode.ts b/packages/coding-agent/src/modes/acp/acp-mode.ts index 7a8dc8864..8a2c89292 100644 --- a/packages/coding-agent/src/modes/acp/acp-mode.ts +++ b/packages/coding-agent/src/modes/acp/acp-mode.ts @@ -72,13 +72,8 @@ function canonicalCwd(path: string): string { function isJsonRpcResponse(message: unknown, requestId: unknown): boolean { if (typeof message !== "object" || message === null) return false; const record = message as Record; - const has = (key: string): boolean => Object.prototype.hasOwnProperty.call(record, key); - return ( - record.jsonrpc === "2.0" && - record.id === requestId && - !has("method") && - (has("result") !== has("error")) - ); + const has = (key: string): boolean => Object.hasOwn(record, key); + return record.jsonrpc === "2.0" && record.id === requestId && !has("method") && has("result") !== has("error"); } function sameCwd(left: string, right: string): boolean { @@ -421,9 +416,7 @@ export async function runAcpModeWithConnection( // commit callback. Observe the outgoing response at the supplied stream // boundary instead. The SDK serializes every write, so opening the producer // after this write resolves puts buffered notifications strictly behind it. - let pendingSessionNewResponse: - | { requestId: unknown; producer: AcpUpdateProducer } - | undefined; + let pendingSessionNewResponse: { requestId: unknown; producer: AcpUpdateProducer } | undefined; const failPendingSessionNewResponse = (): void => { const admission = pendingSessionNewResponse; pendingSessionNewResponse = undefined; diff --git a/packages/coding-agent/test/agent-session-recursion.test.ts b/packages/coding-agent/test/agent-session-recursion.test.ts index 9e2b7e499..12e8c4844 100644 --- a/packages/coding-agent/test/agent-session-recursion.test.ts +++ b/packages/coding-agent/test/agent-session-recursion.test.ts @@ -41,8 +41,8 @@ import { SessionManager } from "../src/core/session-manager.js"; import { SettingsManager, type SettingsStorage } from "../src/core/settings-manager.js"; import type { Skill } from "../src/core/skills.js"; import { createSyntheticSourceInfo } from "../src/core/source-info.js"; -import { type ActiveSessionState, resolveActiveSessionState } from "../src/modes/daemon/active-session-state.js"; import { InProcessAgentConnection } from "../src/modes/agent-connection/in-process-agent-connection.js"; +import { type ActiveSessionState, resolveActiveSessionState } from "../src/modes/daemon/active-session-state.js"; import { AgentDaemon } from "../src/modes/daemon/daemon-mode.js"; import { invokeHostRequest } from "./host-request-context.js"; import { createTestExtensionsResult, createTestResourceLoader } from "./utilities.js"; @@ -679,7 +679,7 @@ describe("AgentSession rlm recursion", () => { expect(events).toContainEqual( expect.objectContaining({ type: "rlm_child_update", - child: { id: childId, status: "cancelled" }, + child: expect.objectContaining({ id: childId, status: "cancelled" }), }), ); expect(root.getRlmChildSnapshots()).toEqual([ @@ -691,9 +691,7 @@ describe("AgentSession rlm recursion", () => { setBeforeSessionInvalidate() {}, dispose: async () => {}, } as unknown as AgentSessionRuntime); - await expect(connection.getInitialSnapshot()).resolves.toMatchObject({ - children: [expect.objectContaining({ id: childId, status: "cancelled" })], - }); + await expect(connection.getInitialSnapshot()).resolves.toMatchObject({ children: [expect.objectContaining({ id: childId, status: "cancelled" })] }); await expect(root.deleteRlmSubagent("cancelled-cleanup-worker")).resolves.toMatchObject({ subagent: { rlm_child_id: childId }, }); diff --git a/packages/coding-agent/test/suite/acp-mode.test.ts b/packages/coding-agent/test/suite/acp-mode.test.ts index 306a0cb0c..0eae9d4bc 100644 --- a/packages/coding-agent/test/suite/acp-mode.test.ts +++ b/packages/coding-agent/test/suite/acp-mode.test.ts @@ -186,6 +186,7 @@ describe("ACP mode end to end", () => { headlessStatus: { enabled: true, continuationsUsed: 1, + gateAttempts: {}, limits: { maxContinuations: 3 }, }, }); From c4d710c88995c53e520bbeae4d99abc475f73ff3 Mon Sep 17 00:00:00 2001 From: Seth Date: Tue, 11 Aug 2026 10:54:56 -0700 Subject: [PATCH 05/17] style(acp): align lifecycle regression fixture --- .../coding-agent/test/agent-session-recursion.test.ts | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/packages/coding-agent/test/agent-session-recursion.test.ts b/packages/coding-agent/test/agent-session-recursion.test.ts index 12e8c4844..74fe50af8 100644 --- a/packages/coding-agent/test/agent-session-recursion.test.ts +++ b/packages/coding-agent/test/agent-session-recursion.test.ts @@ -682,16 +682,16 @@ describe("AgentSession rlm recursion", () => { child: expect.objectContaining({ id: childId, status: "cancelled" }), }), ); - expect(root.getRlmChildSnapshots()).toEqual([ - expect.objectContaining({ id: childId, status: "cancelled" }), - ]); + expect(root.getRlmChildSnapshots()).toEqual([expect.objectContaining({ id: childId, status: "cancelled" })]); const connection = new InProcessAgentConnection({ session: root, setRebindSession() {}, setBeforeSessionInvalidate() {}, dispose: async () => {}, } as unknown as AgentSessionRuntime); - await expect(connection.getInitialSnapshot()).resolves.toMatchObject({ children: [expect.objectContaining({ id: childId, status: "cancelled" })] }); + await expect(connection.getInitialSnapshot()).resolves.toMatchObject({ + children: [expect.objectContaining({ id: childId, status: "cancelled" })], + }); await expect(root.deleteRlmSubagent("cancelled-cleanup-worker")).resolves.toMatchObject({ subagent: { rlm_child_id: childId }, }); From 7698c7d1e7d5fefb1d46e79cc788824057868c48 Mon Sep 17 00:00:00 2001 From: Seth Date: Tue, 11 Aug 2026 11:07:48 -0700 Subject: [PATCH 06/17] fix(acp): preserve connection-scoped child origins --- .../coding-agent/src/modes/acp/acp-mode.ts | 10 ++++---- .../coding-agent/test/suite/acp-mode.test.ts | 24 +++++++++++++++++++ 2 files changed, 29 insertions(+), 5 deletions(-) diff --git a/packages/coding-agent/src/modes/acp/acp-mode.ts b/packages/coding-agent/src/modes/acp/acp-mode.ts index 8a2c89292..e7ae7f039 100644 --- a/packages/coding-agent/src/modes/acp/acp-mode.ts +++ b/packages/coding-agent/src/modes/acp/acp-mode.ts @@ -191,11 +191,11 @@ class AcpUpdateProducer { if (event.type === "rlm_child_update") { const known = this.childOriginTurnIds.get(event.child.id); if (known !== undefined) return this.terminalTurns.has(known) ? 0 : known; - // A child is first observed while its parent prompt is producing it. - // Remember that origin so a later child update cannot be relabelled by - // a subsequent prompt. - if (this.activePromptTurnId !== 0) this.childOriginTurnIds.set(event.child.id, this.activePromptTurnId); - return this.activePromptTurnId; + // Remember its initial origin, including connection scope, so a later + // child update cannot be relabelled by a subsequent prompt. + const originTurnId = this.activePromptTurnId !== 0 ? this.activePromptTurnId : 0; + this.childOriginTurnIds.set(event.child.id, originTurnId); + return originTurnId; } return this.activePromptTurnId; } diff --git a/packages/coding-agent/test/suite/acp-mode.test.ts b/packages/coding-agent/test/suite/acp-mode.test.ts index 0eae9d4bc..e2817e060 100644 --- a/packages/coding-agent/test/suite/acp-mode.test.ts +++ b/packages/coding-agent/test/suite/acp-mode.test.ts @@ -460,6 +460,30 @@ describe("ACP mode end to end", () => { close(); }); + it("preserves a child first observed between prompts as connection-scoped", async () => { + const child = { id: "between-prompts", label: "child", status: "running", sessionDir: "/tmp/child" }; + let connection: any; + connection = fakeAcpConnection({ + onPromptAndWait: () => connection.emitChild({ ...child, status: "done" }), + }); + const { client, updates, close } = connectAcpClient(connection); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const session = await client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + connection.emitChild(child); + await vi.waitFor(() => + expect(updates.some((u) => u.update?._meta?.[PRIME_AGENT_META_NAMESPACE]?.subagents)).toBe(true), + ); + await client.request("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "complete" }], + }); + const childUpdates = updates + .map((u) => u.update?._meta?.[PRIME_AGENT_META_NAMESPACE]) + .filter((meta) => meta?.subagents?.[0]?.id === child.id); + expect(childUpdates.map((meta) => meta.promptTurnId)).toEqual([0, 0]); + close(); + }); + it("makes a child first observed after terminal connection-scoped", async () => { const connection = fakeAcpConnection(); const { client, updates, close } = connectAcpClient(connection); From 86526a3878d6287f720728ceb7a994d75f4cdcdb Mon Sep 17 00:00:00 2001 From: Seth Date: Tue, 11 Aug 2026 11:50:51 -0700 Subject: [PATCH 07/17] fix(acp): settle terminal producer lifecycle --- .../coding-agent/src/modes/acp/acp-mode.ts | 12 +- .../coding-agent/test/suite/acp-mode.test.ts | 114 +++++++++++++++++- 2 files changed, 121 insertions(+), 5 deletions(-) diff --git a/packages/coding-agent/src/modes/acp/acp-mode.ts b/packages/coding-agent/src/modes/acp/acp-mode.ts index e7ae7f039..550439b64 100644 --- a/packages/coding-agent/src/modes/acp/acp-mode.ts +++ b/packages/coding-agent/src/modes/acp/acp-mode.ts @@ -591,13 +591,11 @@ export async function runAcpModeWithConnection( try { const { text, images } = promptContent(params.prompt); const priorMessages = turnBoundary(await connection.getMessages()); - if (abort.signal.aborted) return { stopReason: "cancelled" satisfies AcpStopReason }; - await connection.promptAndWait(text, images.length > 0 ? { images } : undefined); if (abort.signal.aborted) { await entry.producer.drain(); return { stopReason: "cancelled" satisfies AcpStopReason }; } - const failure = await turnFailure(connection, priorMessages); + await connection.promptAndWait(text, images.length > 0 ? { images } : undefined); if (abort.signal.aborted) { await entry.producer.drain(); return { stopReason: "cancelled" satisfies AcpStopReason }; @@ -609,6 +607,11 @@ export async function runAcpModeWithConnection( await entry.producer.drain(); return { stopReason: "cancelled" satisfies AcpStopReason }; } + const failure = await turnFailure(connection, priorMessages); + if (abort.signal.aborted) { + await entry.producer.drain(); + return { stopReason: "cancelled" satisfies AcpStopReason }; + } const autonomous = autonomousMeta(status); const liveSnapshot = await connection.getInitialSnapshot(); if (abort.signal.aborted) { @@ -691,6 +694,9 @@ export async function runAcpModeWithConnection( closing.abort.abort(); await connection.abort().catch(() => undefined); } + // Settle queued notifications before returning the close response, so no + // update can arrive after the client observes this session as closed. + await closing.producer.drain(); return {}; }) .onNotification("session/cancel", async (ctx: any) => { diff --git a/packages/coding-agent/test/suite/acp-mode.test.ts b/packages/coding-agent/test/suite/acp-mode.test.ts index e2817e060..80e3685a1 100644 --- a/packages/coding-agent/test/suite/acp-mode.test.ts +++ b/packages/coding-agent/test/suite/acp-mode.test.ts @@ -45,7 +45,8 @@ function fakeAcpConnection( ): any { let listener: ((event: any) => void) | undefined; let subscribed = false; - const snapshot = { state: { cwd: process.cwd() }, messages: [] }; + const messages: any[] = []; + const snapshot = { state: { cwd: process.cwd() }, messages }; return { subscribe(callback: (event: any) => void) { subscribed = true; @@ -56,7 +57,7 @@ function fakeAcpConnection( }; }, getState: async () => snapshot.state, - getMessages: async () => [], + getMessages: async () => messages, getInitialSnapshot: async () => { if (options.onInitialSnapshot) options.onInitialSnapshot(subscribed); if (options.initialSnapshot) { @@ -92,6 +93,7 @@ function fakeAcpConnection( emitHeartbeat() { listener?.({ type: "heartbeats_changed" }); }, + messages, }; } @@ -349,6 +351,37 @@ describe("ACP mode end to end", () => { close(); }); + it("captures an autonomous continuation error after headless completion", async () => { + let connection: any; + connection = fakeAcpConnection({ + onWaitForHeadlessCompletion: () => { + connection.messages.push({ + role: "assistant", + timestamp: Date.now(), + stopReason: "error", + errorMessage: "autonomous continuation failed", + }); + }, + }); + const { client, updates, close } = connectAcpClient(connection); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const session = await client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + await expect( + client.request("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "finish" }], + }), + ).rejects.toThrow("autonomous continuation failed"); + const metadata = updates.map((u) => u.update?._meta?.[PRIME_AGENT_META_NAMESPACE]).filter(Boolean); + expect(metadata.filter((meta) => meta.phase === "responseBoundary")).toEqual([ + expect.objectContaining({ outcome: "error" }), + ]); + expect(metadata.filter((meta) => meta.phase === "terminalQuiescence")).toEqual([ + expect.objectContaining({ outcome: "error" }), + ]); + close(); + }); + it("propagates a failed roster read at quiescence emission", async () => { // A snapshot failure while emitting must not degrade to outstandingSubagents: 0. // Reporting a fabricated zero is the false-quiescent answer this metadata @@ -504,6 +537,83 @@ describe("ACP mode end to end", () => { close(); }); + it("drains an earlier backpressured update before cancellation after the initial transcript read", async () => { + let entered!: () => void; + let release!: () => void; + const enteredPublish = new Promise((resolve) => { + entered = resolve; + }); + const releasePublish = new Promise((resolve) => { + release = resolve; + }); + const connection = fakeAcpConnection(); + const { client, updates, close } = connectAcpClient(connection, { + beforeAcpUpdatePublish: async () => { + entered(); + await releasePublish; + }, + }); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const session = await client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + connection.emitHeartbeat(); + await enteredPublish; + connection.getMessages = async () => { + await client.notify("session/cancel", { sessionId: session.sessionId }); + return []; + }; + const pending = client.request("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "cancel" }], + }); + await Promise.resolve(); + let settled = false; + void pending.then(() => { + settled = true; + }); + await Promise.resolve(); + expect(settled).toBe(false); + release(); + await expect(pending).resolves.toMatchObject({ stopReason: "cancelled" }); + expect(updates.map((update) => update.update?.sessionUpdate)).toEqual(["session_info_update"]); + close(); + }); + + it("settles queued updates before close resolves without a post-close notification", async () => { + let entered!: () => void; + let release!: () => void; + const enteredPublish = new Promise((resolve) => { + entered = resolve; + }); + const releasePublish = new Promise((resolve) => { + release = resolve; + }); + const connection = fakeAcpConnection(); + const { client, updates, close } = connectAcpClient(connection, { + beforeAcpUpdatePublish: async () => { + entered(); + await releasePublish; + }, + }); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const session = await client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + connection.emitHeartbeat(); + await enteredPublish; + const closing = client.request("session/close", { sessionId: session.sessionId }); + await Promise.resolve(); + let closed = false; + void closing.then(() => { + closed = true; + }); + await Promise.resolve(); + expect(closed).toBe(false); + release(); + await closing; + expect(updates).toHaveLength(1); + await Promise.resolve(); + expect(updates).toHaveLength(1); + close(); + }); + it("cancels after prompt completion without emitting a result or terminal boundary", async () => { let entered!: () => void; let release!: () => void; From 656dd21f17d03118c4fb3924168089c574a4cb27 Mon Sep 17 00:00:00 2001 From: Seth Date: Tue, 11 Aug 2026 11:55:00 -0700 Subject: [PATCH 08/17] test(acp): assert sanitized terminal failure --- packages/coding-agent/test/suite/acp-mode.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/coding-agent/test/suite/acp-mode.test.ts b/packages/coding-agent/test/suite/acp-mode.test.ts index 80e3685a1..e1bf3d8f9 100644 --- a/packages/coding-agent/test/suite/acp-mode.test.ts +++ b/packages/coding-agent/test/suite/acp-mode.test.ts @@ -371,7 +371,7 @@ describe("ACP mode end to end", () => { sessionId: session.sessionId, prompt: [{ type: "text", text: "finish" }], }), - ).rejects.toThrow("autonomous continuation failed"); + ).rejects.toThrow("Internal error"); const metadata = updates.map((u) => u.update?._meta?.[PRIME_AGENT_META_NAMESPACE]).filter(Boolean); expect(metadata.filter((meta) => meta.phase === "responseBoundary")).toEqual([ expect.objectContaining({ outcome: "error" }), From 6f81455b9c5c36995f00f67086e93c411d7be9ee Mon Sep 17 00:00:00 2001 From: Seth Date: Wed, 12 Aug 2026 04:11:20 -0700 Subject: [PATCH 09/17] fix(acp): recover update queue after publish failures --- .../coding-agent/src/modes/acp/acp-mode.ts | 22 +++++++++++-------- .../coding-agent/test/suite/acp-mode.test.ts | 20 +++++++++++++++++ 2 files changed, 33 insertions(+), 9 deletions(-) diff --git a/packages/coding-agent/src/modes/acp/acp-mode.ts b/packages/coding-agent/src/modes/acp/acp-mode.ts index 550439b64..1a6acbe3b 100644 --- a/packages/coding-agent/src/modes/acp/acp-mode.ts +++ b/packages/coding-agent/src/modes/acp/acp-mode.ts @@ -228,16 +228,20 @@ class AcpUpdateProducer { }, }, }; - // Keep the chain alive after a disconnect, while preserving the order of - // every later notification and allowing callers to await its drain. + // Keep the chain alive after a failed hook or notification, while preserving + // the order of every later notification and allowing callers to await its drain. this.tail = this.tail.then(async () => { - await this.admissionReady; - if (!this.admissionOpen) return; - await this.beforePublish?.(correlatedUpdate); - await this.client - .notify(acp.methods.client.session.update, { sessionId: this.sessionId, update: correlatedUpdate }) - .then(() => undefined) - .catch(() => undefined); + try { + await this.admissionReady; + if (!this.admissionOpen) return; + await this.beforePublish?.(correlatedUpdate); + await this.client.notify(acp.methods.client.session.update, { + sessionId: this.sessionId, + update: correlatedUpdate, + }); + } catch { + // Drop only this update; a rejected queue tail would strand later updates. + } }); return this.tail; } diff --git a/packages/coding-agent/test/suite/acp-mode.test.ts b/packages/coding-agent/test/suite/acp-mode.test.ts index e1bf3d8f9..7f4088c9f 100644 --- a/packages/coding-agent/test/suite/acp-mode.test.ts +++ b/packages/coding-agent/test/suite/acp-mode.test.ts @@ -578,6 +578,26 @@ describe("ACP mode end to end", () => { close(); }); + it("recovers the update queue after a failed publish hook", async () => { + let rejectOnce = true; + const connection = fakeAcpConnection(); + const { client, updates, close } = connectAcpClient(connection, { + beforeAcpUpdatePublish: () => { + if (!rejectOnce) return; + rejectOnce = false; + throw new Error("publish hook failed"); + }, + }); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const session = await client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + connection.emitHeartbeat(); + connection.emitHeartbeat(); + await vi.waitFor(() => expect(updates).toHaveLength(1)); + expect(updates[0].update?._meta?.[PRIME_AGENT_META_NAMESPACE]).toMatchObject({ eventSequence: 2 }); + await expect(client.request("session/close", { sessionId: session.sessionId })).resolves.toEqual({}); + close(); + }); + it("settles queued updates before close resolves without a post-close notification", async () => { let entered!: () => void; let release!: () => void; From 28b42c2eb3afc6b1881020a8d9c27fd6457bfb2d Mon Sep 17 00:00:00 2001 From: Seth Date: Wed, 12 Aug 2026 17:57:25 -0700 Subject: [PATCH 10/17] fix(acp): retain nested work through parent deletion --- .../coding-agent/src/core/agent-session.ts | 60 ++++++------ .../test/agent-session-recursion.test.ts | 95 +++++++++++++++++++ .../coding-agent/test/suite/acp-mode.test.ts | 32 +++++++ 3 files changed, 160 insertions(+), 27 deletions(-) diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index 0091956b5..15cd45a68 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -9547,38 +9547,44 @@ export class AgentSession { getRlmChildSnapshots(): RlmChildAgentSnapshot[] { const snapshots: RlmChildAgentSnapshot[] = []; const recorded = new Set(); + const traversed = new Set(); for (const run of this._activeRlmChildRuns.values()) { - if (run.detachedDeletion || this._deletingRlmChildren.has(run.id) || this._deletedRlmChildIds.has(run.id)) { - continue; - } + const hidden = + run.detachedDeletion || this._deletingRlmChildren.has(run.id) || this._deletedRlmChildIds.has(run.id); const child = run.session; - snapshots.push({ - id: run.id, - parentId: this._rlmParentNodeId, - sessionName: child?.sessionName ?? run.sessionName, - model: `${(child?.model ?? run.model).provider}/${(child?.model ?? run.model).id}`, - label: rlmChildLabel(run.prompt), - status: run.status, - sessionDir: run.sessionDir, - }); - recorded.add(run.id); - if (child) snapshots.push(...child.getRlmChildSnapshots()); + if (!hidden) { + snapshots.push({ + id: run.id, + parentId: this._rlmParentNodeId, + sessionName: child?.sessionName ?? run.sessionName, + model: `${(child?.model ?? run.model).provider}/${(child?.model ?? run.model).id}`, + label: rlmChildLabel(run.prompt), + status: run.status, + sessionDir: run.sessionDir, + }); + recorded.add(run.id); + } + if (child) { + traversed.add(run.id); + snapshots.push(...child.getRlmChildSnapshots()); + } } for (const [childId, child] of this._rlmChildSessions) { - if (recorded.has(childId) || this._deletingRlmChildren.has(childId) || this._deletedRlmChildIds.has(childId)) { - continue; + if (recorded.has(childId) || traversed.has(childId)) continue; + const hidden = this._deletingRlmChildren.has(childId) || this._deletedRlmChildIds.has(childId); + if (!hidden) { + snapshots.push({ + id: childId, + parentId: this._rlmParentNodeId, + sessionName: child.sessionName, + model: child.model ? `${child.model.provider}/${child.model.id}` : undefined, + label: child.sessionName ?? "child agent", + // A failed delete retains the session solely for cleanup retry. Preserve + // its cancellation truth in snapshots rather than reviving it as done. + status: this._rlmChildCleanupFailures.has(childId) ? "cancelled" : "done", + sessionDir: child._rlmSessionDir ?? child.sessionManager.getSessionDir(), + }); } - snapshots.push({ - id: childId, - parentId: this._rlmParentNodeId, - sessionName: child.sessionName, - model: child.model ? `${child.model.provider}/${child.model.id}` : undefined, - label: child.sessionName ?? "child agent", - // A failed delete retains the session solely for cleanup retry. Preserve - // its cancellation truth in snapshots rather than reviving it as done. - status: this._rlmChildCleanupFailures.has(childId) ? "cancelled" : "done", - sessionDir: child._rlmSessionDir ?? child.sessionManager.getSessionDir(), - }); snapshots.push(...child.getRlmChildSnapshots()); } return snapshots; diff --git a/packages/coding-agent/test/agent-session-recursion.test.ts b/packages/coding-agent/test/agent-session-recursion.test.ts index 72815fb5b..95c52d742 100644 --- a/packages/coding-agent/test/agent-session-recursion.test.ts +++ b/packages/coding-agent/test/agent-session-recursion.test.ts @@ -112,7 +112,10 @@ interface CapturedCommReply { interface InspectableRlmRun { id: string; + prompt: string; + sessionName: string; sessionDir: string; + model: typeof model; abort: () => void; status: string; settled: boolean; @@ -133,6 +136,7 @@ interface InspectableRlmSession { _rlmChildCleanupFailures: Map>["subagents"][number]>; _rlmChildSessions: Map; _rlmChildUnsubscribes: Map void>; + _deletedRlmChildIds: Set; _createKernelHostHandlers(): HostRequestHandlers; _reapDeletedRlmSubagentRuntimesAfterCompaction(): Promise; } @@ -693,6 +697,97 @@ describe("AgentSession rlm recursion", () => { expect(root.getRlmChildSnapshots()).toEqual([]); }); + it("keeps live nested children visible when their direct parent is hidden by deletion", () => { + const root = createSession(); + const rootInternals = root as unknown as InspectableRlmSession; + const hiddenParents = [ + { id: "deleted-parent", hiding: "deleted" as const }, + { id: "deleting-parent", hiding: "deleting" as const }, + { id: "detached-parent", hiding: "detached" as const }, + ]; + const parentInternalsToClear: InspectableRlmSession[] = []; + + for (const [index, { id, hiding }] of hiddenParents.entries()) { + const parent = createSession({ rlmSessionDir: join(tempDir, id) }); + const parentInternals = parent as unknown as InspectableRlmSession; + parentInternalsToClear.push(parentInternals); + const nestedId = `${id}-live-grandchild`; + parentInternals._activeRlmChildRuns.set(nestedId, { + id: nestedId, + prompt: "still working", + sessionName: nestedId, + sessionDir: join(tempDir, nestedId), + model, + abort: () => {}, + status: index === 1 ? "queued" : "running", + settled: false, + }); + + if (hiding === "detached") { + rootInternals._activeRlmChildRuns.set(id, { + id, + prompt: "hidden parent", + sessionName: id, + sessionDir: join(tempDir, id), + model, + abort: () => {}, + status: "cancelled", + settled: false, + detachedDeletion: { + rlm_child_id: id, + active_session_id: null, + session_id: null, + session_name: id, + session_dir: join(tempDir, id), + status: "running", + }, + session: parent, + }); + // The same child can be visible in both lifecycle registries while + // deletion settles; it must be traversed exactly once and remain hidden. + rootInternals._rlmChildSessions.set(id, parent); + } else { + rootInternals._rlmChildSessions.set(id, parent); + if (hiding === "deleted") { + rootInternals._deletedRlmChildIds.add(id); + } else { + rootInternals._deletingRlmChildren.set(id, { + subagent: { + rlm_child_id: id, + active_session_id: null, + session_id: null, + session_name: id, + session_dir: join(tempDir, id), + status: "completed", + }, + promise: Promise.resolve({ + subagent: { + rlm_child_id: id, + active_session_id: null, + session_id: null, + session_name: id, + session_dir: join(tempDir, id), + status: "completed", + }, + }), + }); + } + } + } + + const snapshots = root.getRlmChildSnapshots(); + expect(snapshots.map((snapshot) => snapshot.id).sort()).toEqual( + hiddenParents.map(({ id }) => `${id}-live-grandchild`).sort(), + ); + expect(snapshots.map((snapshot) => snapshot.status).sort()).toEqual(["queued", "running", "running"]); + // These are deliberately minimal lifecycle records; remove them before + // fixture teardown asks real runs to settle. + rootInternals._activeRlmChildRuns.clear(); + rootInternals._rlmChildSessions.clear(); + for (const parentInternals of parentInternalsToClear) parentInternals._activeRlmChildRuns.clear(); + root.dispose(); + }); + it("makes an orchestrator-chosen name override a custom runtime's preexisting name", async () => { const hostedChild = createSession(); hostedChild.setSessionName("factory-assigned-name"); diff --git a/packages/coding-agent/test/suite/acp-mode.test.ts b/packages/coding-agent/test/suite/acp-mode.test.ts index 7f4088c9f..5b1c9082e 100644 --- a/packages/coding-agent/test/suite/acp-mode.test.ts +++ b/packages/coding-agent/test/suite/acp-mode.test.ts @@ -209,6 +209,38 @@ describe("ACP mode end to end", () => { close(); }); + it("does not emit terminal quiescence when a lifecycle-filtered parent leaves a live grandchild", async () => { + const connection = fakeAcpConnection({ + finalSnapshot: async () => ({ + state: { cwd: process.cwd() }, + messages: [], + // getRlmChildSnapshots omits the deleting direct parent but retains this + // nested live child, which must keep ACP from claiming terminal quiescence. + children: [ + { + id: "live-grandchild", + parentId: "deleted-parent", + sessionName: "live-grandchild", + label: "still working", + status: "running", + sessionDir: "/tmp/live-grandchild", + }, + ], + }), + }); + const { client, updates, close } = connectAcpClient(connection); + await client.request("initialize", { protocolVersion: acp.PROTOCOL_VERSION, clientCapabilities: {} }); + const session = await client.request("session/new", { cwd: process.cwd(), mcpServers: [] }); + await client.request("session/prompt", { + sessionId: session.sessionId, + prompt: [{ type: "text", text: "finish" }], + }); + const metadata = updates.map((update) => update.update?._meta?.[PRIME_AGENT_META_NAMESPACE]).filter(Boolean); + expect(metadata.some((meta) => meta.phase === "terminalQuiescence")).toBe(false); + expect(metadata.find((meta) => meta.quiescence)?.quiescence.outstandingSubagents).toBe(1); + close(); + }); + it("reports a live in-process child that spawned after ACP attached", async () => { const harness = await createHarness({ rlmDepth: 0, rlmMaxDepth: 1 }); let releaseChild!: () => void; From dd035db108ad90a27ad5679052ddb130988ba9be Mon Sep 17 00:00:00 2001 From: Seth Date: Wed, 12 Aug 2026 20:01:07 -0700 Subject: [PATCH 11/17] fix(agent): add intercept relay identity --- packages/agent/src/agent-loop.ts | 13 ++++ packages/agent/test/agent-loop.test.ts | 89 ++++++++++++++++++++++++++ 2 files changed, 102 insertions(+) diff --git a/packages/agent/src/agent-loop.ts b/packages/agent/src/agent-loop.ts index 0d3d7376d..72dd14185 100644 --- a/packages/agent/src/agent-loop.ts +++ b/packages/agent/src/agent-loop.ts @@ -3,6 +3,7 @@ * Transforms to Message[] only at the LLM call boundary. */ +import { randomBytes } from "node:crypto"; import { type AssistantMessage, type AssistantMessageEvent, @@ -26,6 +27,17 @@ import type { export type AgentEventSink = (event: AgentEvent) => Promise | void; const ABORT_ERROR_MESSAGE = "Request was aborted"; +const PRIME_AGENT_RELAY_ID_HEADER = "X-Prime-Agent-Relay-ID"; + +function withPrimeAgentRelayId(headers: Record | undefined): Record { + const filteredHeaders = Object.fromEntries( + Object.entries(headers ?? {}).filter( + ([name]) => name.toLowerCase() !== PRIME_AGENT_RELAY_ID_HEADER.toLowerCase(), + ), + ); + return { ...filteredHeaders, [PRIME_AGENT_RELAY_ID_HEADER]: randomBytes(16).toString("hex") }; +} + const EMPTY_USAGE: AssistantMessage["usage"] = { input: 0, output: 0, @@ -516,6 +528,7 @@ async function streamAssistantResponse( ...config, apiKey: resolvedApiKey, signal, + ...(config.model.provider === "intercept" ? { headers: withPrimeAgentRelayId(config.headers) } : {}), }), signal, ); diff --git a/packages/agent/test/agent-loop.test.ts b/packages/agent/test/agent-loop.test.ts index f8745d0a0..b010d887d 100644 --- a/packages/agent/test/agent-loop.test.ts +++ b/packages/agent/test/agent-loop.test.ts @@ -110,6 +110,95 @@ function identityConverter(messages: AgentMessage[]): Message[] { return messages.filter((m) => m.role === "user" || m.role === "assistant" || m.role === "toolResult") as Message[]; } +describe("Agent relay identity", () => { + it("generates a distinct protected relay ID per intercept completion and keeps it through downstream retries", async () => { + const seenRelayIds: string[] = []; + const retryRelayIds: string[] = []; + const eventPayloads: string[] = []; + const streamFn = vi.fn((_model: Model, _context, options) => { + const relayId = options?.headers?.["X-Prime-Agent-Relay-ID"]; + if (!relayId) throw new Error("missing relay ID"); + seenRelayIds.push(relayId); + + // The provider SDK owns retries; it receives this one logical request's unchanged headers. + for (let attempt = 0; attempt < 2; attempt++) { + retryRelayIds.push(options?.headers?.["X-Prime-Agent-Relay-ID"] ?? ""); + } + + const stream = new MockAssistantStream(); + queueMicrotask(() => { + stream.push({ + type: "done", + reason: "stop", + message: createAssistantMessage([{ type: "text", text: "Response" }]), + }); + }); + return stream; + }); + const config: AgentLoopConfig = { + model: { ...createModel(), provider: "intercept" }, + convertToLlm: identityConverter, + headers: { + "X-Extra-Auth": "preserved", + "x-prime-agent-relay-id": "caller-must-not-control-this", + }, + }; + const run = async () => { + const events: AgentEvent[] = []; + await runAgentLoop( + [createUserMessage("Hello")], + { systemPrompt: "You are helpful.", messages: [], tools: [] }, + config, + (event) => { + events.push(event); + }, + undefined, + streamFn, + ); + eventPayloads.push(JSON.stringify(events)); + }; + + await run(); + await run(); + + expect(streamFn).toHaveBeenCalledTimes(2); + expect(seenRelayIds).toHaveLength(2); + expect(seenRelayIds[0]).toMatch(/^[0-9a-f]{32}$/); + expect(seenRelayIds[1]).toMatch(/^[0-9a-f]{32}$/); + expect(seenRelayIds[0]).not.toBe(seenRelayIds[1]); + expect(retryRelayIds).toEqual([seenRelayIds[0], seenRelayIds[0], seenRelayIds[1], seenRelayIds[1]]); + expect(eventPayloads.join("\n")).not.toContain(seenRelayIds[0]!); + expect(eventPayloads.join("\n")).not.toContain(seenRelayIds[1]!); + }); + + it("does not add a relay ID for non-intercept providers", async () => { + let headers: Record | undefined; + const stream = new MockAssistantStream(); + const streamFn = vi.fn((_model: Model, _context, options) => { + headers = options?.headers; + queueMicrotask(() => { + stream.push({ + type: "done", + reason: "stop", + message: createAssistantMessage([{ type: "text", text: "Response" }]), + }); + }); + return stream; + }); + + await runAgentLoop( + [createUserMessage("Hello")], + { systemPrompt: "You are helpful.", messages: [], tools: [] }, + { model: createModel(), convertToLlm: identityConverter, headers: { "X-Extra-Auth": "preserved" } }, + () => {}, + undefined, + streamFn, + ); + + expect(headers).toEqual({ "X-Extra-Auth": "preserved" }); + }); +}); + describe("agentLoop with AgentMessage", () => { it("should preserve a terminal response when abort fires after done", async () => { const context: AgentContext = { From f61d6091af83a26716ecfee40e44b3249e6fa962 Mon Sep 17 00:00:00 2001 From: Seth Date: Wed, 12 Aug 2026 20:12:13 -0700 Subject: [PATCH 12/17] fix(coding-agent): avoid duplicate intercepted retries --- .../coding-agent/src/core/agent-session.ts | 7 ++++++ .../suite/agent-session-retry-events.test.ts | 23 +++++++++++++++++-- 2 files changed, 28 insertions(+), 2 deletions(-) diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index 15cd45a68..12cc472a2 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -10115,6 +10115,13 @@ export class AgentSession { return false; } + // The intercept provider adds an idempotency key at the logical request boundary. + // Retrying here would call agent.continue() and mint a new key; provider retries + // remain below that boundary and retain the original key. + if (message.provider === "intercept") { + return false; + } + if (this._isAgentLifecycleFailure(message)) { return false; } diff --git a/packages/coding-agent/test/suite/agent-session-retry-events.test.ts b/packages/coding-agent/test/suite/agent-session-retry-events.test.ts index 6e0863264..526f06ea6 100644 --- a/packages/coding-agent/test/suite/agent-session-retry-events.test.ts +++ b/packages/coding-agent/test/suite/agent-session-retry-events.test.ts @@ -1,7 +1,7 @@ import type { AgentEvent, AgentTool } from "@earendil-works/pi-agent-core"; import { type AssistantMessage, fauxAssistantMessage, fauxThinking, fauxToolCall } from "@earendil-works/pi-ai"; import { Type } from "typebox"; -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { createHarness, type Harness } from "./harness.js"; function normalizeEventOrder(events: Harness["events"]): string[] { @@ -249,7 +249,26 @@ describe("AgentSession retry and event characterization", () => { }); } - it("retries generic provider errors", async () => { + it("does not outer-retry intercept provider errors", async () => { + const harness = await createHarness({ + provider: "intercept", + settings: { retry: { enabled: true, maxRetries: 3, baseDelayMs: 1 } }, + }); + harnesses.push(harness); + const continueSpy = vi.spyOn(harness.session.agent, "continue"); + harness.setResponses([ + fauxAssistantMessage("", { stopReason: "error", errorMessage: "unknown after provider" }), + fauxAssistantMessage("must not make a second logical request"), + ]); + + await harness.session.prompt("test"); + + expect(harness.faux.state.callCount).toBe(1); + expect(harness.eventsOfType("auto_retry_start")).toEqual([]); + expect(continueSpy).not.toHaveBeenCalled(); + }); + + it("retries generic errors from non-intercept providers", async () => { const harness = await createHarness({ settings: { retry: { enabled: true, maxRetries: 3, baseDelayMs: 1 } } }); harnesses.push(harness); harness.setResponses([ From b588c5b5cec50cda4c66cd41a5564650f63e669f Mon Sep 17 00:00:00 2001 From: Seth Date: Wed, 12 Aug 2026 20:42:21 -0700 Subject: [PATCH 13/17] fix(agent): type relay stream test callbacks --- packages/agent/test/agent-loop.test.ts | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/packages/agent/test/agent-loop.test.ts b/packages/agent/test/agent-loop.test.ts index b010d887d..e001a04b3 100644 --- a/packages/agent/test/agent-loop.test.ts +++ b/packages/agent/test/agent-loop.test.ts @@ -9,7 +9,14 @@ import { import { Type } from "typebox"; import { describe, expect, it, vi } from "vitest"; import { agentLoop, agentLoopContinue, runAgentLoop } from "../src/agent-loop.js"; -import type { AgentContext, AgentEvent, AgentLoopConfig, AgentMessage, AgentTool } from "../src/types.js"; +import type { + AgentContext, + AgentEvent, + AgentLoopConfig, + AgentMessage, + AgentTool, + StreamFn, +} from "../src/types.js"; // Mock stream for testing - mimics MockAssistantStream class MockAssistantStream extends EventStream { @@ -115,7 +122,7 @@ describe("Agent relay identity", () => { const seenRelayIds: string[] = []; const retryRelayIds: string[] = []; const eventPayloads: string[] = []; - const streamFn = vi.fn((_model: Model, _context, options) => { + const streamFn: StreamFn = vi.fn((_model, _context, options) => { const relayId = options?.headers?.["X-Prime-Agent-Relay-ID"]; if (!relayId) throw new Error("missing relay ID"); seenRelayIds.push(relayId); @@ -174,7 +181,7 @@ describe("Agent relay identity", () => { it("does not add a relay ID for non-intercept providers", async () => { let headers: Record | undefined; const stream = new MockAssistantStream(); - const streamFn = vi.fn((_model: Model, _context, options) => { + const streamFn: StreamFn = vi.fn((_model, _context, options) => { headers = options?.headers; queueMicrotask(() => { stream.push({ From 69983012ebee94db40e0613cf5a391904e9314ea Mon Sep 17 00:00:00 2001 From: Seth Date: Wed, 12 Aug 2026 20:44:12 -0700 Subject: [PATCH 14/17] fix(coding-agent): contain intercept auth retries --- packages/coding-agent/src/core/agent-session.ts | 4 +++- .../test/suite/agent-session-retry-events.test.ts | 7 +++++-- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index 12cc472a2..3d2f422a6 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -3568,7 +3568,9 @@ export class AgentSession { // Check for retryable errors first (overloaded, rate limit, server errors) const concreteAuthFailure = this._isConcreteProviderAuthFailure(msg); const retryConcreteAuthFailure = - concreteAuthFailure && !this._isStructuredPermanentProviderRetryExhausted(msg); + msg.provider !== "intercept" && + concreteAuthFailure && + !this._isStructuredPermanentProviderRetryExhausted(msg); if (this._isRetryableError(msg) || retryConcreteAuthFailure) { if (retryConcreteAuthFailure) { this._captureRetryAuthFailureSource(msg); diff --git a/packages/coding-agent/test/suite/agent-session-retry-events.test.ts b/packages/coding-agent/test/suite/agent-session-retry-events.test.ts index 526f06ea6..b96c82f17 100644 --- a/packages/coding-agent/test/suite/agent-session-retry-events.test.ts +++ b/packages/coding-agent/test/suite/agent-session-retry-events.test.ts @@ -249,7 +249,10 @@ describe("AgentSession retry and event characterization", () => { }); } - it("does not outer-retry intercept provider errors", async () => { + it.each([ + "unknown after provider", + "401 status code: unauthorized API key", + ])("does not outer-retry intercept provider error: %s", async (errorMessage) => { const harness = await createHarness({ provider: "intercept", settings: { retry: { enabled: true, maxRetries: 3, baseDelayMs: 1 } }, @@ -257,7 +260,7 @@ describe("AgentSession retry and event characterization", () => { harnesses.push(harness); const continueSpy = vi.spyOn(harness.session.agent, "continue"); harness.setResponses([ - fauxAssistantMessage("", { stopReason: "error", errorMessage: "unknown after provider" }), + fauxAssistantMessage("", { stopReason: "error", errorMessage }), fauxAssistantMessage("must not make a second logical request"), ]); From 58f9a052731bbb13177cda6e079307ee03ddb325 Mon Sep 17 00:00:00 2001 From: Seth Date: Wed, 12 Aug 2026 21:00:14 -0700 Subject: [PATCH 15/17] test(coding-agent): await process teardown before cleanup --- .../coding-agent/test/acp-cold-cli.test.ts | 4 ++- .../test/daemon-supervisor-process.test.ts | 34 +++++++++++++++++-- 2 files changed, 35 insertions(+), 3 deletions(-) diff --git a/packages/coding-agent/test/acp-cold-cli.test.ts b/packages/coding-agent/test/acp-cold-cli.test.ts index 38a3760e1..fa1b35da5 100644 --- a/packages/coding-agent/test/acp-cold-cli.test.ts +++ b/packages/coding-agent/test/acp-cold-cli.test.ts @@ -27,7 +27,9 @@ afterEach(async () => { await new Promise((done) => server.close(() => done())); } for (const dir of tempDirs.splice(0)) { - rmSync(dir, { recursive: true, force: true }); + // The cold CLI may finish while an inherited uv/Python helper is releasing + // files in its private HOME. Retry ENOTEMPTY rather than racing teardown. + rmSync(dir, { recursive: true, force: true, maxRetries: 50, retryDelay: 100 }); } }); diff --git a/packages/coding-agent/test/daemon-supervisor-process.test.ts b/packages/coding-agent/test/daemon-supervisor-process.test.ts index c474c5b28..26e8ff892 100644 --- a/packages/coding-agent/test/daemon-supervisor-process.test.ts +++ b/packages/coding-agent/test/daemon-supervisor-process.test.ts @@ -50,13 +50,15 @@ afterEach(async () => { } } daemonSockets.clear(); - for (const child of children) { + const trackedChildren = [...children]; + for (const child of trackedChildren) { if (child.exitCode === null && child.signalCode === null) { child.kill("SIGTERM"); } } children.clear(); - for (const pid of workerPids) { + const trackedWorkerPids = [...workerPids]; + for (const pid of trackedWorkerPids) { try { process.kill(pid, "SIGCONT"); } catch { @@ -73,6 +75,34 @@ afterEach(async () => { } } workerPids.clear(); + await Promise.all( + trackedChildren.map(async (child) => { + try { + await waitForExit(child); + } catch { + child.kill("SIGKILL"); + await waitForExit(child); + } + }), + ); + await Promise.all( + trackedWorkerPids.map(async (pid) => { + try { + await waitForProcessGone(pid); + } catch { + try { + process.kill(-pid, "SIGKILL"); + } catch { + try { + process.kill(pid, "SIGKILL"); + } catch { + return; + } + } + await waitForProcessGone(pid); + } + }), + ); for (const directory of tempDirs.splice(0)) { // Detached workers can release kernel/snapshot files just after their process group exits on macOS. rmSync(directory, { recursive: true, force: true, maxRetries: 50, retryDelay: 100 }); From 60e5994d6e3188944972f23d840529fb120c4846 Mon Sep 17 00:00:00 2001 From: Seth Date: Wed, 12 Aug 2026 21:21:08 -0700 Subject: [PATCH 16/17] fix(release): exclude local Python caches --- scripts/pack-prime-agent-release.mjs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/scripts/pack-prime-agent-release.mjs b/scripts/pack-prime-agent-release.mjs index 9460443d1..1417db3b7 100644 --- a/scripts/pack-prime-agent-release.mjs +++ b/scripts/pack-prime-agent-release.mjs @@ -145,9 +145,14 @@ function requireBuiltPackage(packageDir) { } } +function isReleaseCopyable(source) { + const name = basename(source); + return name !== "__pycache__" && name !== ".DS_Store" && !name.endsWith(".pyc"); +} + function copyIfExists(source, target) { if (existsSync(source)) { - cpSync(source, target, { recursive: true }); + cpSync(source, target, { recursive: true, filter: isReleaseCopyable }); } } From cbdff0bd8072a65934d60780c5a896f9e5a1b320 Mon Sep 17 00:00:00 2001 From: Seth Date: Thu, 13 Aug 2026 11:43:23 -0700 Subject: [PATCH 17/17] feat(cli): add exact socket daemon shutdown --- .../coding-agent/src/cli/command-registry.ts | 13 +++++-- .../coding-agent/src/cli/daemon-launch.ts | 31 +++++++++++++++ .../coding-agent/src/cli/public-command.ts | 33 +++++++++++++++- .../coding-agent/test/daemon-launch.test.ts | 38 +++++++++++++++++++ .../coding-agent/test/public-command.test.ts | 27 +++++++++++++ 5 files changed, 137 insertions(+), 5 deletions(-) diff --git a/packages/coding-agent/src/cli/command-registry.ts b/packages/coding-agent/src/cli/command-registry.ts index dff1ee4e3..0079fd111 100644 --- a/packages/coding-agent/src/cli/command-registry.ts +++ b/packages/coding-agent/src/cli/command-registry.ts @@ -87,10 +87,15 @@ export const COMMAND_SPECS: readonly CommandSpec[] = [ }, { path: ["shutdown"], - usage: "shutdown [--force] [--json]", - summary: "Stop every agent and background service", - description: "Without --force, an interactive confirmation is required. --force also kills unresponsive workers.", - options: ["--force Skip confirmation and kill unresponsive processes", "--json Print JSON"], + usage: "shutdown [--force] [--json] [--daemon-socket ]", + summary: "Stop agents and background services", + description: + "Without --daemon-socket, stops every service after confirmation. With --daemon-socket, stops only the verified daemon at that exact path and cannot be combined with --force.", + options: [ + "--force Skip confirmation and kill unresponsive processes", + "--daemon-socket Stop only the verified daemon at this exact socket", + "--json Print JSON", + ], }, { path: ["package"], diff --git a/packages/coding-agent/src/cli/daemon-launch.ts b/packages/coding-agent/src/cli/daemon-launch.ts index 6de9285e9..f0848d556 100644 --- a/packages/coding-agent/src/cli/daemon-launch.ts +++ b/packages/coding-agent/src/cli/daemon-launch.ts @@ -265,6 +265,37 @@ export async function shutdownDaemonAndWait(socketPath: string, timeoutMs = 5000 } } +/** + * Shut down exactly one current daemon addressed by `socketPath`. + * + * Unlike the machine-wide shutdown/reap commands this performs no discovery. It + * also refuses stale, foreign, or same-version/different-build daemons before + * sending a shutdown request. Callers use this for rollout-owned daemon sockets. + */ +export async function shutdownExactDaemonAndWait(socketPath: string, timeoutMs = 10_000): Promise { + const client = new DaemonClient(socketPath); + try { + await client.connect(1000); + const hello = await client.waitForHello(2000); + const expectedRuntime = getDaemonRuntimeIdentity(); + const compatible = + hello.protocol.version === DAEMON_PROTOCOL_VERSION && + hello.schemaId === DAEMON_SCHEMA_ID && + hello.appVersion === VERSION && + hello.runtime?.buildId === expectedRuntime.buildId; + if (!compatible) { + throw new Error(`Refusing exact shutdown of an unverified daemon on ${socketPath}`); + } + return await shutdownConnectedDaemonAndWait(client, socketPath, timeoutMs, hello); + } catch (error) { + client.close(); + if (!existsSync(socketPath)) { + return true; + } + throw error; + } +} + // activeSessions is undefined when the daemon is reachable but its sessions couldn't // be listed — callers must treat that as "possibly busy", not idle. export type RunningDaemonProbe = diff --git a/packages/coding-agent/src/cli/public-command.ts b/packages/coding-agent/src/cli/public-command.ts index 020498e1d..e746d1d90 100644 --- a/packages/coding-agent/src/cli/public-command.ts +++ b/packages/coding-agent/src/cli/public-command.ts @@ -13,6 +13,7 @@ import { REMOVED_COMMAND_NAMES, } from "./command-registry.js"; import { handleDaemonCommand } from "./daemon-command.js"; +import { shutdownExactDaemonAndWait } from "./daemon-launch.js"; import { runPs, runReap, runShutdownAll } from "./daemon-ps.js"; import { DAEMON_UPDATE_RESTART_COORDINATOR_FLAG } from "./daemon-update-restart.js"; @@ -259,8 +260,38 @@ async function runDoctor(args: string[]): Promise { } async function runShutdown(args: string[]): Promise { - const options = parseBooleanOptions(args, new Set(["--force", "--json"]), "shutdown"); + let socketPath: string | undefined; + const booleanOptions: string[] = []; + for (let index = 0; index < args.length; index++) { + const arg = args[index]!; + if (arg === "--daemon-socket") { + const value = args[++index]; + if (!value || value.startsWith("-")) { + return fail("--daemon-socket requires a path"); + } + if (socketPath !== undefined) { + return fail("--daemon-socket may only be supplied once"); + } + socketPath = value; + continue; + } + booleanOptions.push(arg); + } + const options = parseBooleanOptions(booleanOptions, new Set(["--force", "--json"]), "shutdown"); if (!options) return HANDLED; + if (socketPath !== undefined) { + if (options.has("--force")) { + return fail("--force cannot be combined with exact --daemon-socket shutdown"); + } + const stopped = await shutdownExactDaemonAndWait(socketPath); + if (!stopped) { + throw new Error(`Exact daemon shutdown was not confirmed for ${socketPath}`); + } + if (options.has("--json")) { + console.log(JSON.stringify({ socketPath, stopped: true })); + } + return HANDLED; + } await runShutdownAll(options.has("--json"), options.has("--force")); return HANDLED; } diff --git a/packages/coding-agent/test/daemon-launch.test.ts b/packages/coding-agent/test/daemon-launch.test.ts index 1a649d6cc..78bf1a323 100644 --- a/packages/coding-agent/test/daemon-launch.test.ts +++ b/packages/coding-agent/test/daemon-launch.test.ts @@ -10,6 +10,7 @@ import { probeRunningDaemonSessions, shouldStartDaemonEarly, shutdownDaemonAndWait, + shutdownExactDaemonAndWait, } from "../src/cli/daemon-launch.js"; import { ENV_AGENT_DIR, getDaemonLogPath, VERSION } from "../src/config.js"; import { DAEMON_PROTOCOL_VERSION, DAEMON_SCHEMA_ID } from "../src/modes/daemon/daemon-protocol.js"; @@ -26,6 +27,7 @@ interface FakeDaemonOptions { appVersion?: string; schemaId?: string; serverCapabilities?: string[]; + runtimeBuildId?: string; onCommand?: (command: { type: string }) => void; } @@ -50,6 +52,7 @@ async function startFakeDaemon(options: FakeDaemonOptions = {}): Promise { }); }); +describe("shutdownExactDaemonAndWait", () => { + const cleanups: Array<() => Promise> = []; + afterEach(async () => { + await Promise.all(cleanups.splice(0).map((fn) => fn())); + }); + + it("shuts down only a current same-build daemon at the supplied socket", async () => { + const commands: string[] = []; + const daemon = await startFakeDaemon({ + protocolVersion: DAEMON_PROTOCOL_VERSION, + appVersion: VERSION, + schemaId: DAEMON_SCHEMA_ID, + runtimeBuildId: `release-${VERSION}`, + onCommand: (command) => commands.push(command.type), + }); + cleanups.push(daemon.close); + await expect(shutdownExactDaemonAndWait(daemon.socketPath)).resolves.toBe(true); + expect(commands).toEqual(["shutdown"]); + }); + + it("refuses a same-version daemon with a different build identity", async () => { + const commands: string[] = []; + const daemon = await startFakeDaemon({ + protocolVersion: DAEMON_PROTOCOL_VERSION, + appVersion: VERSION, + schemaId: DAEMON_SCHEMA_ID, + runtimeBuildId: "unrelated-build", + onCommand: (command) => commands.push(command.type), + }); + cleanups.push(daemon.close); + await expect(shutdownExactDaemonAndWait(daemon.socketPath)).rejects.toThrow("Refusing exact shutdown"); + expect(commands).not.toContain("shutdown"); + }); +}); + describe("shouldStartDaemonEarly", () => { it.each([ ["interactive", []], diff --git a/packages/coding-agent/test/public-command.test.ts b/packages/coding-agent/test/public-command.test.ts index b24b115df..713ba9f82 100644 --- a/packages/coding-agent/test/public-command.test.ts +++ b/packages/coding-agent/test/public-command.test.ts @@ -7,6 +7,8 @@ const mocks = vi.hoisted(() => ({ psCalls: [] as boolean[], reapCalls: [] as Array<[boolean, boolean]>, shutdownCalls: [] as Array<[boolean, boolean]>, + exactShutdownCalls: [] as string[], + exactShutdownResult: true, })); vi.mock("../src/cli/daemon-command.js", () => ({ @@ -16,6 +18,13 @@ vi.mock("../src/cli/daemon-command.js", () => ({ }, })); +vi.mock("../src/cli/daemon-launch.js", () => ({ + shutdownExactDaemonAndWait: async (socketPath: string) => { + mocks.exactShutdownCalls.push(socketPath); + return mocks.exactShutdownResult; + }, +})); + vi.mock("../src/package-manager-cli.js", () => ({ handlePackageCommand: async (args: string[]) => { mocks.packageCommands.push(args); @@ -48,6 +57,8 @@ describe("public command routing", () => { mocks.psCalls.length = 0; mocks.reapCalls.length = 0; mocks.shutdownCalls.length = 0; + mocks.exactShutdownCalls.length = 0; + mocks.exactShutdownResult = true; process.exitCode = undefined; vi.spyOn(console, "log").mockImplementation(() => {}); vi.spyOn(console, "error").mockImplementation(() => {}); @@ -241,6 +252,22 @@ describe("public command routing", () => { ]); }); + it("shuts down only the explicitly addressed verified daemon", async () => { + await handlePublicCommand(["shutdown", "--daemon-socket", "/tmp/rollout/daemon.sock", "--json"]); + expect(mocks.exactShutdownCalls).toEqual(["/tmp/rollout/daemon.sock"]); + expect(mocks.shutdownCalls).toEqual([]); + expect(console.log).toHaveBeenCalledWith( + JSON.stringify({ socketPath: "/tmp/rollout/daemon.sock", stopped: true }), + ); + }); + + it("rejects force with exact-socket shutdown", async () => { + await handlePublicCommand(["shutdown", "--daemon-socket", "/tmp/rollout/daemon.sock", "--force"]); + expect(mocks.exactShutdownCalls).toEqual([]); + expect(mocks.shutdownCalls).toEqual([]); + expect(process.exitCode).toBe(1); + }); + it("routes doctor fixes through the safe cleanup path", async () => { await handlePublicCommand(["doctor", "--fix", "--json"]); expect(mocks.reapCalls).toEqual([[true, false]]);