Merge branch 'codex/simp-hide-concrete-agent-loop' into codex/simp-hide-subagent-internals

packages/core/agent/README.md

@@ -11,7 +11,7 @@ Tracks live agents so UI, hook, and orchestrator plugins can find them without i
 The scoped-registration surface: `Agent.ctx` is the agent's scope context (`dsh-scope`, key = the agent) — register tools/sections/variables/listeners through it for that agent alone, all unwound on disposal. `agentEvents(ctx, agent)` is the fused dispatcher for ordinary agent-subject operations (carrier + injected subject in one move); its notification mode invokes every listener and contains both synchronous throws and returned-promise rejections. The registry lifecycle pair reuses one stable routing carrier. `assembleContextFor(agent)` builds the per-agent assembly context (`agent` + `scope` together). `CreateAgentOptions.setup(agentCtx)` and `ResumeAgentOptions.setup(agentCtx)` compose a fresh or resumed agent's scoped world while both objects remain unpublished. Setup is trusted, composition-only same-process code: drive the agent only after creation resolves.
 
 - `ctx.agents.register(agent: Agent): () => void` — record an **already-constructed** agent. Disposed with the calling fiber.
-- Advanced ordered lifecycle: `enter(agent, owner): () => void` 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()`.
+- 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.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

@@ -329,6 +329,9 @@ export class AgentRegistry extends Service {
    */
   enter(agent: Agent, owner: Agent | undefined): () => void {
     const id = agent.id
+    if (id !== agent.session.id) {
+      throw new Error(`agent id "${id}" does not match session id "${agent.session.id}"`)
+    }
     const carrier = scopeTarget(agent, agent)
     // This is the authoritative collision boundary. Concurrent create/resume
     // operations may both prepare, but only one exact entry can publish.

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

@@ -11,7 +11,7 @@ function stubAgent(rawId: string): Agent {
   return {
     id,
     options: {},
-    session: new Session(SessionId(`${id}-session`)),
+    session: new Session(id),
     status: 'idle',
     ctx: new Context(),
     send() {},
@@ -50,6 +50,16 @@ describe('AgentRegistry', () => {
     expect(lifecycle).toEqual(['created:a1', 'disposed:a1'])
   })
 
+  it('rejects an agent whose registry and session identities differ', async () => {
+    const ctx = new Context()
+    await ctx.plugin(AgentRegistry)
+    const agent = { ...stubAgent('agent-id'), session: new Session(SessionId('session-id')) }
+
+    expect(() => ctx.agents.enter(agent, undefined))
+      .toThrow('agent id "agent-id" does not match session id "session-id"')
+    expect(ctx.agents.list()).toEqual([])
+  })
+
   it('tracks runtime creator ownership separately from registry order', async () => {
     const ctx = new Context()
     await ctx.plugin(AgentRegistry)

packages/hooks/hooks-claude/tests/coverage.spec.ts

@@ -4,7 +4,7 @@ import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { Context } from 'cordis'
 import LlmService from '@deepseek-ai/dsh-llm'
-import SessionStore, { SessionId, type SessionEvent } from '@deepseek-ai/dsh-session'
+import SessionStore, { Session, SessionId, type SessionEvent } from '@deepseek-ai/dsh-session'
 import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
 import ToolRegistry, { defineTool } from '@deepseek-ai/dsh-tools'
 import AgentRegistry, { type Agent } from '@deepseek-ai/dsh-agent'
@@ -209,7 +209,8 @@ describe('hooks-claude coverage — Stop continuation + subagent inject/catch',
     const ctx = await harness(path, new MockAdapter([]))
     // Register a fake child agent under the id the event carries.
     const injected: string[] = []
-    const child = { id: SessionId('child-x'), inject: (content: { type: string; text?: string }[]) => { injected.push(content.map(b => b.text ?? '').join('')) }, session: { header: { id: 'child-x' } } } as unknown as Parameters<typeof ctx.agents.register>[0]
+    const childId = SessionId('child-x')
+    const child = { id: childId, inject: (content: { type: string; text?: string }[]) => { injected.push(content.map(b => b.text ?? '').join('')) }, session: new Session(childId) } as unknown as Parameters<typeof ctx.agents.register>[0]
     ctx.agents.register(child)
     ctx.emit('subagent/start', { provider: 'p', id: SessionId('child-x') })
     await waitFor(() => injected.includes('child guidance'))
@@ -225,7 +226,8 @@ describe('hooks-claude coverage — Stop continuation + subagent inject/catch',
     const path = hooks(d, { SubagentStart: [{ hooks: [{ type: 'command', command: s }] }] })
     const ctx = await harness(path, new MockAdapter([]))
     const warn = vi.fn(); ctx.logger.warn = warn as never
-    const child = { id: SessionId('child-y'), inject: () => { throw new Error('inject boom') }, session: { header: { id: 'child-y' } } } as unknown as Parameters<typeof ctx.agents.register>[0]
+    const childId = SessionId('child-y')
+    const child = { id: childId, inject: () => { throw new Error('inject boom') }, session: new Session(childId) } as unknown as Parameters<typeof ctx.agents.register>[0]
     ctx.agents.register(child)
     ctx.emit('subagent/start', { provider: 'p', id: SessionId('child-y') })
     await waitFor(() => warn.mock.calls.some(c => String(c[0]).includes('SubagentStart hook failed')))

packages/subagent/subagent-acp/src/run.ts

@@ -311,7 +311,9 @@ export async function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpe
           clientCapabilities: {},
         })
         const session = await conn.newSession({ cwd: spec.cwd, mcpServers: [] })
-        sessionId = session.sessionId
+        const returnedSessionId: unknown = Reflect.get(session, 'sessionId')
+        if (typeof returnedSessionId !== 'string') throw new Error('ACP child published without a session id')
+        sessionId = returnedSessionId
         if (flags.cancelled) throw new Error('subagent cancelled before the ACP session started')
       })(),
       spawnFailed.then((err): never => { throw err }),
@@ -323,10 +325,9 @@ export async function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpe
     if (flags.cancelled) throw new Error('subagent request was aborted before the ACP child started')
     throw toError(error)
   }
-  // The startup race can fulfill only after newSession assigned the id; this
+  // The startup transaction validates the returned id before it can fulfill.
-  // guard keeps that cross-closure invariant explicit for TypeScript.
+  // This assertion carries that cross-closure invariant into TypeScript.
-  /* v8 ignore next */
+  if (sessionId === undefined) throw new Error('unreachable: ACP startup fulfilled without a session id')
-  if (sessionId === undefined) throw new Error('ACP child published without a session id')
   const remoteSessionId = sessionId
 
   const result: Promise<SubagentResult> = (async (): Promise<SubagentResult> => {

packages/subagent/subagent-acp/tests/mock-acp-server.ts

@@ -19,6 +19,8 @@
  *                        handler is in flight (it has streamed its chunk). A test
  *                        polls for this file to cancel on a CONDITION rather than
  *                        an arbitrary timeout (subprocess cold-start is variable).
+ * - `MOCK_MISSING_SESSION_ID` — if `1`, return a malformed empty `session/new`
+ *                        response to exercise startup rollback.
  * - `MOCK_FLUSH_ON_EOF` — if set, on stdin EOF the agent takes an async beat
  *                         (MOCK_FLUSH_DELAY_MS, default 150) simulating the real
  *                         acp-agent's EOF-driven quiesce+flush, then touches this
@@ -99,6 +101,7 @@ function makeAgent(conn: AgentSideConnection): Agent {
         writeFileSync(NEWSESSION_GATE.ready, 'at-newSession')
         while (!existsSync(NEWSESSION_GATE.go)) await new Promise(r => setTimeout(r, 10))
       }
+      if (process.env.MOCK_MISSING_SESSION_ID === '1') return {} as NewSessionResponse
       return { sessionId: process.env.MOCK_SESSION_ID ?? randomUUID() }
     },
     authenticate(_params: AuthenticateRequest): Promise<void> {

packages/subagent/subagent-acp/tests/subagent-acp.spec.ts

@@ -194,6 +194,32 @@ describe('dsh-subagent-acp', () => {
     }
   })
 
+  it('reaps a child whose session/new response omits the session id', async () => {
+    const tmp = mkdtempSync(join(tmpdir(), 'acp-malformed-session-'))
+    const flushed = join(tmp, 'flushed')
+    try {
+      await expect(startAcpRun(request(), {
+        command: process.execPath,
+        args: ['--import', tsxLoader, mockServer],
+        cwd: process.cwd(),
+        permission: 'reject',
+        env: {
+          MOCK_MISSING_SESSION_ID: '1',
+          MOCK_FLUSH_ON_EOF: flushed,
+          MOCK_FLUSH_DELAY_MS: '20',
+          TSX_TSCONFIG_PATH: repoTsconfig,
+        },
+        disposeEofGraceMs: 1000,
+        disposeGraceMs: 100,
+      })).rejects.toThrow('ACP child published without a session id')
+      // Startup rejects only after its private child reaches quiescence. The
+      // marker proves rollback closed stdin and allowed the child's EOF flush.
+      expect(existsSync(flushed)).toBe(true)
+    } finally {
+      rmSync(tmp, { recursive: true, force: true })
+    }
+  })
+
   it('dispose escalates SIGTERM → SIGKILL for a child that traps SIGTERM (bounded quiescence)', async () => {
     // The child traps SIGTERM and keeps its event loop alive, so a graceful
     // term alone would hang dispose forever. With a short grace, dispose must