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

docs/cordis-catalog/services.md

@@ -35,6 +35,7 @@ register(agent: Agent): () => void
 enter(agent: Agent, owner: Agent | undefined): () => void
 announce(agent: Agent): void
 get(id: SessionId): Agent | undefined
+isOwnedBy(id: SessionId, owner: Agent): boolean
 list(): Agent[]
 roots(): Agent[]
 ```

packages/core/agent/README.md

@@ -13,6 +13,7 @@ The scoped-registration surface: `Agent.ctx` is the agent's scope context (`dsh-
 - `ctx.agents.register(agent: Agent): () => void` — record an **already-constructed** agent. Disposed with the calling fiber.
 - Advanced ordered lifecycle: `enter(agent, owner): () => void` enforces `agent.id === agent.session.id`, performs the authoritative ID collision check, and inserts without announcing; `owner` explicitly records the live creator-agent relation (or `undefined` for a root), independently of durable session lineage. `announce(agent)` emits `agent/created` exactly once. A detach requested synchronously by a creation listener is deferred until that dispatch unwinds, and every detach checks the captured entry object, so a stale capability cannot delete a later same-ID replacement. The async factory uses this split; ordinary plugins use `register()`.
 - `ctx.agents.get(id: SessionId): Agent | undefined`
+- `ctx.agents.isOwnedBy(id: SessionId, owner: Agent): boolean` — whether the exact live entry was created through that parent agent's scoped context; runtime ownership is independent of durable session lineage.
 - `ctx.agents.list(): Agent[]`
 - `ctx.agents.roots(): Agent[]` — live agents created without an owning agent context; a resumed lineage-bearing session can still be a runtime root.
 

packages/core/agent/src/index.ts

@@ -440,6 +440,18 @@ export class AgentRegistry extends Service {
     return this.store.get(id)?.agent
   }
 
+  /**
+   * Test whether a live agent was created through one exact parent agent's
+   * scoped context. Runtime ownership is independent of durable session
+   * lineage and remains unambiguous when unrelated providers reuse an id.
+   * @param id - the candidate child agent's shared agent/session id.
+   * @param owner - the expected runtime creator agent.
+   * @returns true only while the exact child entry is live under that owner.
+   */
+  isOwnedBy(id: SessionId, owner: Agent): boolean {
+    return this.store.get(id)?.owner === owner
+  }
+
   /**
    * All live agents, in registration order.
    * @returns a fresh array; mutating it does not affect the registry.

packages/core/agent/tests/agent.spec.ts

@@ -72,8 +72,12 @@ describe('AgentRegistry', () => {
 
     expect(ctx.agents.list()).toEqual([root, child])
     expect(ctx.agents.roots()).toEqual([root])
+    expect(ctx.agents.isOwnedBy(child.id, root)).toBe(true)
+    expect(ctx.agents.isOwnedBy(root.id, root)).toBe(false)
+    expect(ctx.agents.isOwnedBy(SessionId('missing'), root)).toBe(false)
 
     detachChild()
+    expect(ctx.agents.isOwnedBy(child.id, root)).toBe(false)
     detachRoot()
   })
 

packages/support/acp-snapshot/src/harness.ts

@@ -244,9 +244,9 @@ export async function runScenario(input: InputScript, opts: RunOptions): Promise
   )
 
   // Failure-safe teardown: wait for a still-running child, then attempt BOTH
-  // directory removals even when an earlier cleanup rejects. The main outcome
-  // wins over teardown noise so a step/harvest failure is never replaced; on a
-  // successful run, the first cleanup failure remains visible to the caller.
+  // directory removals even when an earlier cleanup rejects. Report every
+  // teardown failure alongside a scenario failure so neither orthogonal
+  // outcome hides the other.
   const cleanupResults: PromiseSettledResult<unknown>[] = []
   const cleanup = async (action: () => Promise<unknown>): Promise<void> => {
     cleanupResults.push(...await Promise.allSettled([action()]))
@@ -256,10 +256,18 @@ export async function runScenario(input: InputScript, opts: RunOptions): Promise
   await cleanup(() => rm(cwd, { recursive: true, force: true }))
   await cleanup(() => rm(sessionsRoot, { recursive: true, force: true }))
 
+  const cleanupFailures = cleanupResults
+    .filter((result): result is PromiseRejectedResult => result.status === 'rejected')
+    .map(result => result.reason as unknown)
+  if (cleanupFailures.length > 0) {
+    throw new AggregateError(
+      outcome.status === 'rejected' ? [outcome.error, ...cleanupFailures] : cleanupFailures,
+      outcome.status === 'rejected'
+        ? 'snapshot scenario and cleanup failed'
+        : 'snapshot cleanup failed',
+    )
+  }
   if (outcome.status === 'rejected') throw outcome.error
-  const cleanupFailure = cleanupResults.find((result): result is PromiseRejectedResult => result.status === 'rejected')
-  /* v8 ignore next 1 -- defensive OS cleanup failure after an otherwise successful real subprocess run */
-  if (cleanupFailure !== undefined) throw cleanupFailure.reason
   return outcome.value
 }
 

packages/support/acp-snapshot/tests/harness.spec.ts

@@ -3,11 +3,29 @@ import { once } from 'node:events'
 import { tmpdir } from 'node:os'
 import { delimiter, join } from 'node:path'
 import { fileURLToPath } from 'node:url'
-import { afterAll, describe, expect, it } from 'vitest'
+import { afterAll, describe, expect, it, vi } from 'vitest'
 import { PROTOCOL_VERSION } from '@agentclientprotocol/sdk'
 import { runScenario, type AgentUnderTest, type InputStep } from '../src/harness.ts'
 import { launchAcpTestAgent } from '../src/launcher.ts'
 
+const fsControl = vi.hoisted(() => ({ cleanupFailure: undefined as Error | undefined }))
+
+vi.mock('node:fs/promises', async (importOriginal) => {
+  const actual = await importOriginal<typeof import('node:fs/promises')>()
+  return {
+    ...actual,
+    async rm(...args: Parameters<typeof actual.rm>): Promise<void> {
+      if (String(args[0]).includes('acp-snap-cwd-') && fsControl.cleanupFailure !== undefined) {
+        const failure = fsControl.cleanupFailure
+        fsControl.cleanupFailure = undefined
+        await actual.rm(...args)
+        throw failure
+      }
+      await actual.rm(...args)
+    },
+  }
+})
+
 /**
  * Unit tests for the subprocess harness, driven through the REAL spawn path
  * (tsx loader, temp cwd, env plumbing) against the scripted fake ACP bin in
@@ -238,6 +256,24 @@ describe('runScenario', () => {
     )).rejects.toThrow(/expected the prompt to fail/)
   })
 
+  it('reports scenario and cleanup failures together', { timeout: 20_000 }, async () => {
+    const { fixtureFile } = await scenario({ prompt: 'respond' })
+    const cleanupFailure = new Error('cleanup failed')
+    fsControl.cleanupFailure = cleanupFailure
+
+    const failure = await runScenario(
+      { steps: [...boot, { op: 'promptExpectError', text: 'fine' }] },
+      { agent: AGENT, mode: 'replay', fixtureFile },
+    ).catch((error: unknown): unknown => error)
+
+    expect(failure).toBeInstanceOf(AggregateError)
+    const failures = (failure as AggregateError).errors as unknown[]
+    expect(failures).toHaveLength(2)
+    expect(failures[0]).toBeInstanceOf(Error)
+    expect((failures[0] as Error).message).toMatch(/expected the prompt to fail/)
+    expect(failures[1]).toBe(cleanupFailure)
+  })
+
   it('newSessionExpectError swallows the rejection, with and without extra dirs', { timeout: 20_000 }, async () => {
     const { fixtureFile } = await scenario({ rejectExtraDirs: true })
     const result = await runScenario(

packages/ui/jsonrpc/README.md

@@ -4,7 +4,7 @@ The **SDK server plugin** (`jsonrpc`): mounting it serves a stdio JSON-RPC serve
 
 ## Wiring
 
-`inject: ['agents']` — the server creates one agent per SDK `sessionId` (get-or-create on `session/prompt`). A local subagent's shared agent/session id supplies `subagent.finished.childSessionId` directly; the server counts local starts by provider/id and the exact delegating-parent carrier, because a continuation may reuse one child and the child may be disposed before a later `subagent/end`. The paired event carrier preserves parent correlation even when reused ids settle out of order. Runs from remote providers are not reported because they create no local agent. The LLM seam is read opportunistically via `ctx.get('llm')` (not injected): when `initialize.model` has no registered adapter, the plugin mounts `dsh-llm-deepseek` for it (credentials from `$DEEPSEEK_API_KEY` / `$DEEPSEEK_BASE_URL`) — a config-registered adapter for the model wins. Everything else — persistence, the tool stacks, the adapter set — comes from the surrounding `cordis.yml`.
+`inject: ['agents']` — the server creates one agent per SDK `sessionId` (get-or-create on `session/prompt`). A local subagent's shared agent/session id supplies `subagent.finished.childSessionId` directly; the server verifies that the live child is owned by the exact delegating parent, then counts local starts by provider/id and that parent carrier because a continuation may reuse one child and the child may be disposed before a later `subagent/end`. The paired event carrier preserves parent correlation even when reused ids settle out of order. Runs from remote providers are not reported even when their parent-scoped run id collides with an unrelated local agent. The LLM seam is read opportunistically via `ctx.get('llm')` (not injected): when `initialize.model` has no registered adapter, the plugin mounts `dsh-llm-deepseek` for it (credentials from `$DEEPSEEK_API_KEY` / `$DEEPSEEK_BASE_URL`) — a config-registered adapter for the model wins. Everything else — persistence, the tool stacks, the adapter set — comes from the surrounding `cordis.yml`.
 
 ## Config
 

packages/ui/jsonrpc/src/server.ts

@@ -109,8 +109,8 @@ export class HarnessSdkServer {
     // child disposal and reused ids need no settlement-order assumption.
     const localRuns = this.localRuns
     this.disposers.push(ctx.on('subagent/start', function (this: Scoped<SubagentService>, info: SubagentRunInfo) {
-      if (ctx.agents.get(info.id) === undefined) return
       const parent = subagentParentOf(this)
+      if (!ctx.agents.isOwnedBy(info.id, parent)) return
       const providerRuns = localRuns.get(info.provider) ?? new Map<SessionId, Map<Agent, number>>()
       const parentRuns = providerRuns.get(info.id) ?? new Map<Agent, number>()
       parentRuns.set(parent, (parentRuns.get(parent) ?? 0) + 1)
@@ -118,7 +118,6 @@ export class HarnessSdkServer {
       localRuns.set(info.provider, providerRuns)
     }))
     this.disposers.push(ctx.on('subagent/end', function (this: Scoped<SubagentService>, info: SubagentRunEndInfo) {
-      const agent = ctx.agents.get(info.id)
       const parent = subagentParentOf(this)
       const providerRuns = localRuns.get(info.provider)
       const parentRuns = providerRuns?.get(info.id)
@@ -130,10 +129,10 @@ export class HarnessSdkServer {
         if (providerRuns?.size === 0) localRuns.delete(info.provider)
       }
       // This protocol reports LOCAL child sessions. A lineage-bearing child
-      // has the session/created-driven start notification above; a parentless
-      // local provider still gets its terminal notification. A remote provider
-      // has neither a pending local start nor a live local agent and is ignored.
-      if (pendingCount === undefined && agent === undefined) return
+      // has the session/created-driven start notification above. A remote run
+      // has neither a cached owned start nor a live child owned by this exact
+      // parent; an unrelated local agent with the same id never makes it local.
+      if (pendingCount === undefined && !ctx.agents.isOwnedBy(info.id, parent)) return
       transport.notify('subagent.finished', {
         provider: info.provider,
         agentId: String(info.id),

packages/ui/jsonrpc/tests/server.spec.ts

@@ -277,12 +277,12 @@ describe('HarnessSdkServer', () => {
         meta: { cwd: storageDir },
         agentOptions: { model: 'deepseek' },
       })
-      const handle = await ctx.agents.create({
+      const handle = await parentHandle.agent.ctx.agents.create({
         sessionId: SessionId('child-session'),
         meta: { cwd: storageDir, parentSession: SessionId('main') },
         agentOptions: { model: 'deepseek' },
       })
-      const parentlessHandle = await ctx.agents.create({
+      const parentlessHandle = await parentHandle.agent.ctx.agents.create({
         sessionId: SessionId('parentless-child-session'),
         meta: { cwd: storageDir },
         agentOptions: { model: 'deepseek' },
@@ -331,6 +331,43 @@ describe('HarnessSdkServer', () => {
     }
   })
 
+  it('ignores a remote run id that collides with an unrelated local agent', async () => {
+    const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-subagent-remote-collision-'))
+    const ctx = await makeHarness(storageDir)
+    try {
+      const transport = new FakeTransport()
+      const server = new HarnessSdkServer(ctx, transport)
+      const parentHandle = await ctx.agents.create({
+        sessionId: SessionId('collision-parent'),
+        meta: { cwd: storageDir },
+        agentOptions: { model: 'deepseek' },
+      })
+      const unrelatedHandle = await ctx.agents.create({
+        sessionId: SessionId('remote-run-id'),
+        meta: { cwd: storageDir },
+        agentOptions: { model: 'deepseek' },
+      })
+
+      await settleSubagent(ctx, parentHandle.agent, {
+        provider: 'remote',
+        id: SessionId('remote-run-id'),
+        stopReason: 'completed',
+        lastAssistantMessage: [],
+      }, () => unrelatedHandle.dispose())
+
+      expect(transport.notifications.some(notification =>
+        notification.method === 'subagent.finished'
+        && notification.params?.agentId === 'remote-run-id',
+      )).toBe(false)
+
+      await parentHandle.dispose()
+      await server.shutdown()
+    } finally {
+      await ctx.fiber.dispose()
+      await rm(storageDir, { recursive: true, force: true })
+    }
+  })
+
   it('retains locality across continuation runs on one live child', async () => {
     const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-subagent-continuation-'))
     const ctx = await makeHarness(storageDir)
@@ -342,7 +379,7 @@ describe('HarnessSdkServer', () => {
         meta: { cwd: storageDir },
         agentOptions: { model: 'deepseek' },
       })
-      const childHandle = await ctx.agents.create({
+      const childHandle = await parentHandle.agent.ctx.agents.create({
         sessionId: SessionId('continuation-child'),
         meta: { cwd: storageDir, parentSession: SessionId('continuation-parent') },
         agentOptions: { model: 'deepseek' },
@@ -385,7 +422,7 @@ describe('HarnessSdkServer', () => {
         meta: { cwd: storageDir },
         agentOptions: { model: 'deepseek' },
       })
-      const oldChild = await ctx.agents.create({
+      const oldChild = await oldParent.agent.ctx.agents.create({
         sessionId: SessionId('reused-child'),
         meta: { cwd: storageDir, parentSession: SessionId('old-parent') },
         agentOptions: { model: 'deepseek' },
@@ -425,7 +462,7 @@ describe('HarnessSdkServer', () => {
         meta: { cwd: storageDir },
         agentOptions: { model: 'deepseek' },
       })
-      const newChild = await ctx.agents.create({
+      const newChild = await newParent.agent.ctx.agents.create({
         sessionId: SessionId('reused-child'),
         meta: { cwd: storageDir, parentSession: SessionId('new-parent') },
         agentOptions: { model: 'deepseek' },
@@ -483,12 +520,12 @@ describe('HarnessSdkServer', () => {
         meta: { cwd: storageDir },
         agentOptions: { model: 'deepseek' },
       })
-      handle = await ctx.agents.create({
+      handle = await parentHandle.agent.ctx.agents.create({
         sessionId: SessionId('fallback-child-session'),
         meta: { cwd: storageDir, parentSession: SessionId('fallback-parent') },
         agentOptions: { model: 'deepseek' },
       })
-      failedHandle = await ctx.agents.create({
+      failedHandle = await parentHandle.agent.ctx.agents.create({
         sessionId: SessionId('failed-child-session'),
         meta: { cwd: storageDir },
         agentOptions: { model: 'deepseek' },