From 9028c9b63b6c3ed1737c168ae5e1548bab064668 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Tue, 14 Jul 2026 05:35:34 +0800 Subject: [PATCH 1/2] fix: contain ACP update predicate failures --- packages/support/acp-snapshot/src/launcher.ts | 13 +++++++++++-- packages/support/acp-snapshot/tests/harness.spec.ts | 4 ++++ 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/packages/support/acp-snapshot/src/launcher.ts b/packages/support/acp-snapshot/src/launcher.ts index 635a02ca35..975b1bb852 100644 --- a/packages/support/acp-snapshot/src/launcher.ts +++ b/packages/support/acp-snapshot/src/launcher.ts @@ -107,6 +107,7 @@ export function launchAcpTestAgent(options: AcpTestLaunchOptions): LaunchedAcpTe const updateWaiters: { match: (update: SessionNotification['update']) => boolean resolve: (update: SessionNotification['update']) => void + reject: (reason: unknown) => void }[] = [] const stream = ndJsonStream( Writable.toWeb(child.stdin) as WritableStream, @@ -119,7 +120,15 @@ export function launchAcpTestAgent(options: AcpTestLaunchOptions): LaunchedAcpTe const waiter = updateWaiters[index] /* v8 ignore next 1 -- index is bounded by the array length */ if (waiter === undefined) continue - if (!waiter.match(params.update)) continue + let matches: boolean + try { + matches = waiter.match(params.update) + } catch (error: unknown) { + updateWaiters.splice(index, 1) + waiter.reject(error) + continue + } + if (!matches) continue updateWaiters.splice(index, 1) waiter.resolve(params.update) } @@ -136,7 +145,7 @@ export function launchAcpTestAgent(options: AcpTestLaunchOptions): LaunchedAcpTe updates, rawStdout: () => Buffer.concat(rawBuffers).toString('utf8'), stderr: () => stderrChunks.join(''), - waitForUpdate: match => new Promise(resolve => updateWaiters.push({ match, resolve })), + waitForUpdate: match => new Promise((resolve, reject) => updateWaiters.push({ match, resolve, reject })), async close(signal?: NodeJS.Signals): Promise { if (child.exitCode !== null || child.signalCode !== null) return if (signal === undefined) child.stdin.end() diff --git a/packages/support/acp-snapshot/tests/harness.spec.ts b/packages/support/acp-snapshot/tests/harness.spec.ts index ec1ed5cf6f..db715db90c 100644 --- a/packages/support/acp-snapshot/tests/harness.spec.ts +++ b/packages/support/acp-snapshot/tests/harness.spec.ts @@ -57,7 +57,11 @@ describe('runScenario', () => { await launched.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} }) const { sessionId } = await launched.client.newSession({ cwd: dir, mcpServers: [] }) const nextChunk = launched.waitForUpdate(update => update.sessionUpdate === 'agent_message_chunk') + const predicateFailure = new Error('predicate failed') + const failedPredicate = launched.waitForUpdate(() => { throw predicateFailure }) + .catch((error: unknown): unknown => error) await launched.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] }) + expect(await failedPredicate).toBe(predicateFailure) expect((await nextChunk).sessionUpdate).toBe('agent_message_chunk') expect(launched.updates.some(update => update.sessionUpdate === 'agent_message_chunk')).toBe(true) expect(launched.rawStdout()).toContain('permission:{\\"outcome\\":\\"cancelled\\"}') From 7d711c44a8a6202680f2801ba16b391858715163 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Tue, 14 Jul 2026 05:41:01 +0800 Subject: [PATCH 2/2] fix: keep remote subagent ids parent-scoped --- .../2026-06-20-unify-agent-and-session-id.md | 4 ++-- .../core/agent-loop/tests/scope-lifecycle.spec.ts | 2 +- packages/subagent/subagent-acp/README.md | 2 ++ packages/subagent/subagent-acp/src/run.ts | 15 ++++++++++----- .../subagent-acp/tests/subagent-acp.spec.ts | 10 ++++++++-- packages/subagent/subagent/src/types.ts | 2 +- 6 files changed, 24 insertions(+), 11 deletions(-) diff --git a/docs/rfc/implemented/simplification/2026-06-20-unify-agent-and-session-id.md b/docs/rfc/implemented/simplification/2026-06-20-unify-agent-and-session-id.md index be1052830a..032eae4f7d 100644 --- a/docs/rfc/implemented/simplification/2026-06-20-unify-agent-and-session-id.md +++ b/docs/rfc/implemented/simplification/2026-06-20-unify-agent-and-session-id.md @@ -14,7 +14,7 @@ Session itself repeated the same fact as `Session.id` and `Session.header.id`. C ## Decision -An agent's registry id equals its session id. `CreateAgentOptions` accepts one `sessionId` used for both final registry entries; resume registers the agent under `resumeSessionId`; in-process and ACP subagent creation use the child session id; and `Session.id` derives from `header.id`. The existing creation transaction, final-entry collision checks, and exact-entry detach semantics remain; maps and fields whose sole job was translating between the ids are gone. +An agent's registry id equals its session id. `CreateAgentOptions` accepts one `sessionId` used for both final registry entries; resume registers the agent under `resumeSessionId`; in-process subagent creation uses the child session id; and `Session.id` derives from `header.id`. A remote ACP run has no local agent/session pair: it keeps one parent-minted lifecycle id while the child server's wire-local session id remains private to ACP calls. The existing creation transaction, final-entry collision checks, and exact-entry detach semantics remain; maps and fields whose sole job was translating between local ids are gone. The config-driven path keeps `agents[].id` as a stable configuration label, not a live routing identity. A fresh start mints the combined id `${label}-session-${randomUUID()}` so durable restarts do not collide; `resumeSessionId` instead supplies the exact combined identity to load and register. Logs may use the stable label while all live and durable lookups use the one `SessionId`. @@ -28,7 +28,7 @@ The config-driven path keeps `agents[].id` as a stable configuration label, not - Agent create/resume and subagent creation carry one identity, and `Session` stores it in one place. - The creation transaction retains final-entry collision, exact-entry detach, rollback, and quiescence coverage without identity-specific lifecycle state. -- ACP, stdio, hooks, bash ownership, persistence, and lineage use the shared `SessionId` directly. The ACP subagent backend uses the child server's returned session id as its run id. +- ACP, stdio, hooks, bash ownership, persistence, and lineage use the shared `SessionId` directly. The ACP subagent backend mints its lifecycle id in the parent namespace because a child server's returned session id is only server-local. - The config-driven resume-or-create policy is explicit and covered across a durable restart. - A production listener search kept `agent/created`/`agent/disposed` and their publication semantics. - Typecheck, coverage, snapshots, doc-sync, module-graph verification, build, and hygiene pass. diff --git a/packages/core/agent-loop/tests/scope-lifecycle.spec.ts b/packages/core/agent-loop/tests/scope-lifecycle.spec.ts index c68d96e525..e29e73c393 100644 --- a/packages/core/agent-loop/tests/scope-lifecycle.spec.ts +++ b/packages/core/agent-loop/tests/scope-lifecycle.spec.ts @@ -249,7 +249,7 @@ describe('agent scope lifecycle', () => { }) await setupStarted.promise expect(ctx.agents.get(SessionId('atomic'))).toBeUndefined() - expect(ctx.sessions.get(SessionId('atomic-s'))).toBeUndefined() + expect(ctx.sessions.get(SessionId('atomic'))).toBeUndefined() expect(order).toEqual(['setup:start']) gate.resolve(undefined) const handle = await creating diff --git a/packages/subagent/subagent-acp/README.md b/packages/subagent/subagent-acp/README.md index c949685fcc..03b1a856ed 100644 --- a/packages/subagent/subagent-acp/README.md +++ b/packages/subagent/subagent-acp/README.md @@ -6,6 +6,8 @@ The ACP provider runs each subagent in a fresh subprocess and drives it as an Ag `start(request)` performs `spawn` → ACP `initialize` → `newSession` before it fulfills. Fulfillment therefore means a remote session is ready and ownership has transferred to the caller. A spawn, initialization, new-session, or pre-publication cancellation failure rejects only after the subprocess has been reaped. +The returned run id is minted in the parent namespace. The child server's session id remains private to ACP wire calls because ACP guarantees it only within that fresh child process; using it as the parent lifecycle id could collide with another remote run or a local agent. + After publication, the provider sends the prompt and collects streamed `agent_message_chunk` text into `SubagentResult.output`. A prompt/transport failure resolves with `stopReason: 'error'`, or `aborted` when the required request signal or disposal requested cancellation. `dispose()` is idempotent. It removes the signal listener, requests ACP cancellation when possible, closes stdin, waits `disposeEofGraceMs`, escalates to SIGTERM, waits `disposeGraceMs`, and finally uses SIGKILL if necessary. Every run uses a fresh process; process pooling is not implemented. diff --git a/packages/subagent/subagent-acp/src/run.ts b/packages/subagent/subagent-acp/src/run.ts index 1cc220204f..7d0c11cbe1 100644 --- a/packages/subagent/subagent-acp/src/run.ts +++ b/packages/subagent/subagent-acp/src/run.ts @@ -23,6 +23,7 @@ */ import { spawn } from 'node:child_process' +import { randomUUID } from 'node:crypto' import { Readable, Writable } from 'node:stream' import { ClientSideConnection, @@ -190,6 +191,10 @@ function toError(value: unknown): Error { */ export async function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpec): Promise { if (request.signal.aborted) throw new Error('subagent request was aborted before the ACP child started') + // ACP session ids are unique only within the child server. The lifecycle id + // is minted in the parent namespace so fresh processes cannot collide with + // each other or with a local agent that happens to use the same session id. + const id = SessionId(randomUUID()) // Spawn the child ACP agent. stdin = ACP request channel, stdout = ACP // response channel, stderr = INHERIT so the child's diagnostics surface on the @@ -257,7 +262,7 @@ export async function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpe ), ) - let sessionId: SessionId | undefined + let sessionId: string | undefined // Resolves when a cancel is requested, so `result` can settle `aborted` even // if the child never cooperates with `session/cancel` (it ignores the notify, // or the prompt wedges). The result path races this against the ACP drive: the @@ -306,7 +311,7 @@ export async function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpe clientCapabilities: {}, }) const session = await conn.newSession({ cwd: spec.cwd, mcpServers: [] }) - sessionId = SessionId(session.sessionId) + sessionId = session.sessionId if (flags.cancelled) throw new Error('subagent cancelled before the ACP session started') })(), spawnFailed.then((err): never => { throw err }), @@ -322,7 +327,7 @@ export async function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpe // guard keeps that cross-closure invariant explicit for TypeScript. /* v8 ignore next */ if (sessionId === undefined) throw new Error('ACP child published without a session id') - const runId = sessionId + const remoteSessionId = sessionId const result: Promise = (async (): Promise => { try { @@ -334,7 +339,7 @@ export async function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpe // succeeds, transport/process failure rejects the in-flight prompt RPC. const prompt = async (): Promise => { // The startup phase cannot fulfill without assigning the session id. - const promptResult = await conn.prompt({ sessionId: runId, prompt: toAcpPrompt(request.prompt) }) + const promptResult = await conn.prompt({ sessionId: remoteSessionId, prompt: toAcpPrompt(request.prompt) }) return { output: collectOutput(), stopReason: acpStopReason(promptResult.stopReason) } } return await Promise.race([ @@ -368,7 +373,7 @@ export async function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpe let disposal: Promise | undefined return { - id: runId, + id, result, dispose(): Promise { if (disposal !== undefined) return disposal diff --git a/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts b/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts index 115d5c3e97..e5fe05384e 100644 --- a/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts +++ b/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts @@ -121,16 +121,22 @@ describe('buildChildEnv', () => { }) describe('dsh-subagent-acp', () => { - it('drives a child process to completion and returns its streamed output', async () => { + it('drives child processes with parent-unique run ids and returns streamed output', async () => { const ctx = await setup({ MOCK_TEXT: 'hello from acp child', MOCK_STOP: 'end_turn', MOCK_SESSION_ID: 'acp-child-session' }) const run = await ctx.subagents.start('acp', request('do X')) - expect(run.id).toBe('acp-child-session') + expect(run.id).not.toBe('acp-child-session') const result = await run.result expect(result.stopReason).toBe('completed') expect(text(result.output)).toBe('hello from acp child') const disposal = run.dispose() expect(run.dispose()).toBe(disposal) await disposal + + const nextRun = await ctx.subagents.start('acp', request('do X again')) + expect(nextRun.id).not.toBe(run.id) + expect(nextRun.id).not.toBe('acp-child-session') + await nextRun.result + await nextRun.dispose() }) it('maps a max_tokens stop reason', async () => { diff --git a/packages/subagent/subagent/src/types.ts b/packages/subagent/subagent/src/types.ts index 8d360b1182..827c2db7a7 100644 --- a/packages/subagent/subagent/src/types.ts +++ b/packages/subagent/subagent/src/types.ts @@ -147,7 +147,7 @@ export interface SubagentResult { * presence of the method IS the capability — narrow before calling. */ export interface SubagentRun { - /** The child agent's id (local in-process runs are already published in `ctx.agents`; remote transports need not publish locally). */ + /** Parent-scoped run id. Local runs use the published child session id; remote providers mint an id unique in the parent namespace. */ readonly id: SessionId /** * Resolves with the child's terminal {@link SubagentResult} when the run