From f02005c832b450d65d372c935c612c7d83eec54d Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Tue, 14 Jul 2026 07:39:28 +0800 Subject: [PATCH] fix: enforce unified agent startup invariants --- packages/core/agent/README.md | 2 +- packages/core/agent/src/index.ts | 3 +++ packages/core/agent/tests/agent.spec.ts | 12 ++++++++- .../hooks/hooks-claude/tests/coverage.spec.ts | 8 +++--- packages/subagent/subagent-acp/src/run.ts | 11 ++++---- .../subagent-acp/tests/mock-acp-server.ts | 3 +++ .../subagent-acp/tests/subagent-acp.spec.ts | 26 +++++++++++++++++++ 7 files changed, 55 insertions(+), 10 deletions(-) diff --git a/packages/core/agent/README.md b/packages/core/agent/README.md index 643c15002b..a45603f191 100644 --- a/packages/core/agent/README.md +++ b/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. diff --git a/packages/core/agent/src/index.ts b/packages/core/agent/src/index.ts index b25d1e3cac..abafa38e45 100644 --- a/packages/core/agent/src/index.ts +++ b/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. diff --git a/packages/core/agent/tests/agent.spec.ts b/packages/core/agent/tests/agent.spec.ts index 5563d01cad..d2866cacb3 100644 --- a/packages/core/agent/tests/agent.spec.ts +++ b/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) diff --git a/packages/hooks/hooks-claude/tests/coverage.spec.ts b/packages/hooks/hooks-claude/tests/coverage.spec.ts index 16c48aa149..0b41fd0d10 100644 --- a/packages/hooks/hooks-claude/tests/coverage.spec.ts +++ b/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 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[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[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[0] + const childId = SessionId('child-y') + const child = { id: childId, inject: () => { throw new Error('inject boom') }, session: new Session(childId) } as unknown as Parameters[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'))) diff --git a/packages/subagent/subagent-acp/src/run.ts b/packages/subagent/subagent-acp/src/run.ts index 7d0c11cbe1..8ef49f0f7d 100644 --- a/packages/subagent/subagent-acp/src/run.ts +++ b/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 - // 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') + // The startup transaction validates the returned id before it can fulfill. + // This assertion carries that cross-closure invariant into TypeScript. + if (sessionId === undefined) throw new Error('unreachable: ACP startup fulfilled without a session id') const remoteSessionId = sessionId const result: Promise = (async (): Promise => { diff --git a/packages/subagent/subagent-acp/tests/mock-acp-server.ts b/packages/subagent/subagent-acp/tests/mock-acp-server.ts index 5145941526..f56404834b 100644 --- a/packages/subagent/subagent-acp/tests/mock-acp-server.ts +++ b/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 { diff --git a/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts b/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts index e5fe05384e..daa037ca20 100644 --- a/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts +++ b/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts @@ -195,6 +195,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