DeepSeek/deepseek-harness
1
0
Fork 0
mirror of https://github.com/deepseek-ai/deepseek-harness synced 2026-08-15 21:04:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: enforce unified agent startup invariants

Browse Source
This commit is contained in:
Tianyi Cui
2026-07-14 07:39:28 +08:00
parent 5a9d5f58a5
commit f02005c832
7 changed files with 55 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
packages/core/agent/README.md
View File

@@ -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.

3
packages/core/agent/src/index.ts
View File

@@ -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.

12
packages/core/agent/tests/agent.spec.ts
View File

@@ -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)

8
packages/hooks/hooks-claude/tests/coverage.spec.ts
View File

@@ -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<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')))

11
packages/subagent/subagent-acp/src/run.ts
View File

@@ -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<SubagentResult> = (async (): Promise<SubagentResult> => {

3
packages/subagent/subagent-acp/tests/mock-acp-server.ts
View File

@@ -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> {

26
packages/subagent/subagent-acp/tests/subagent-acp.spec.ts
View File

@@ -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