Merge branch 'codex/simp-ui-identity-residue' into codex/simp-hide-concrete-agent-loop

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; the ACP bridge verifies exact `Agent` ownership from the forward session map; and JSON-RPC caches only local disposable-child parent lineage while leaving remote runs outside its local-session notification pair.
+- 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 ACP bridge verifies exact `Agent` ownership from the forward session map; and JSON-RPC caches only local disposable-child parent lineage while leaving remote runs outside its local-session notification pair.
 - 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.

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