Merge branch 'codex/simp-unify-agent-session-id' into codex/simp-ui-identity-residue

# Conflicts:
#	docs/rfc/implemented/simplification/2026-06-20-unify-agent-and-session-id.md

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

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.

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<SubagentRun> {
   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<SubagentResult> = (async (): Promise<SubagentResult> => {
     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<SubagentResult> => {
         // 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<void> | undefined
   return {
-    id: runId,
+    id,
     result,
     dispose(): Promise<void> {
       if (disposal !== undefined) return disposal

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 () => {

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

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<Uint8Array>,
@@ -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<void> {
       if (child.exitCode !== null || child.signalCode !== null) return
       if (signal === undefined) child.stdin.end()

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\\"}')