mirror of
https://github.com/deepseek-ai/deepseek-harness
synced 2026-08-15 21:04:50 +00:00
Merge branch 'codex/simp-hide-concrete-agent-loop' into codex/simp-hide-subagent-internals
This commit is contained in:
Tianyi Cui
@@ -8,7 +8,7 @@ This matrix shows which packages dispatch each harness-owned event and which pac
|
||||
| Event | Mode | Declared in | Dispatchers | Listeners |
|
||||
| --- | --- | --- | --- | --- |
|
||||
| `agent/created` | `emit` | [`packages/core/agent/src/types.ts:303`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`jsonrpc`](../packages/ui/jsonrpc), [`stdio-agent`](../packages/ui/stdio-agent) |
|
||||
| `agent/disposed` | `emit` | [`packages/core/agent/src/types.ts:318`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`stdio-agent`](../packages/ui/stdio-agent) |
|
||||
| `agent/disposed` | `emit` | [`packages/core/agent/src/types.ts:318`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`jsonrpc`](../packages/ui/jsonrpc), [`stdio-agent`](../packages/ui/stdio-agent) |
|
||||
| `agent/error` | `emit` | [`packages/core/agent/src/types.ts:592`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | - |
|
||||
| `agent/pre-step` | `serial` | [`packages/core/agent/src/types.ts:425`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`compact-basic`](../packages/compact/compact-basic), [`user-approval`](../packages/ui/user-approval) |
|
||||
| `agent/prompt-submit` | `waterfall` | [`packages/core/agent/src/types.ts:443`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`acp`](../packages/ui/acp), [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex), [`repeat-tool-guard`](../packages/guard/repeat-tool-guard) |
|
||||
@@ -34,7 +34,7 @@ This matrix shows which packages dispatch each harness-owned event and which pac
|
||||
| `subagent/end` | `emit` | [`packages/subagent/subagent/src/index.ts:91`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`events.dispatch`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`jsonrpc`](../packages/ui/jsonrpc) |
|
||||
| `subagent/provider-added` | `emit` | [`packages/subagent/subagent/src/index.ts:67`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`emit`) | [`tool-subagent`](../packages/subagent/tool-subagent) |
|
||||
| `subagent/provider-removed` | `emit` | [`packages/subagent/subagent/src/index.ts:73`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`events.dispatch`) | [`tool-subagent`](../packages/subagent/tool-subagent) |
|
||||
| `subagent/start` | `emit` | [`packages/subagent/subagent/src/index.ts:83`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`events.dispatch`) | [`hooks-claude`](../packages/hooks/hooks-claude) |
|
||||
| `subagent/start` | `emit` | [`packages/subagent/subagent/src/index.ts:83`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`events.dispatch`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`jsonrpc`](../packages/ui/jsonrpc) |
|
||||
| `system-prompt/assemble` | `waterfall` | [`packages/core/system-prompt/src/index.ts:49`](../packages/core/system-prompt/src/index.ts) | [`system-prompt`](../packages/core/system-prompt) (`waterfall`) | - |
|
||||
| `system-prompt/change` | `emit` | [`packages/core/system-prompt/src/index.ts:59`](../packages/core/system-prompt/src/index.ts) | [`system-prompt`](../packages/core/system-prompt) (`emit`) | - |
|
||||
| `tools/change` | `emit` | [`packages/core/tools/src/index.ts:173`](../packages/core/tools/src/index.ts) | [`tools`](../packages/core/tools) (`emit`) | - |
|
||||
|
||||
@@ -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 caches runtime-local identity plus optional parent lineage because the child may be disposed before `subagent/end`, and the provider contract does not require lineage. 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 caches runtime-local identity plus optional parent lineage for the agent lifetime and snapshots it per run, because a continuation may reuse one child and the child may be disposed before a later `subagent/end`. 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`.
|
||||
|
||||
## Config
|
||||
|
||||
|
||||
@@ -17,7 +17,7 @@ import { resolve } from 'node:path'
|
||||
import type { ContentBlock } from '@deepseek-ai/dsh-llm'
|
||||
import type { AgentHandle } from '@deepseek-ai/dsh-agent'
|
||||
import { SessionId, type TurnEndReason } from '@deepseek-ai/dsh-session'
|
||||
import type { SubagentRunEndInfo } from '@deepseek-ai/dsh-subagent'
|
||||
import type { SubagentRunEndInfo, SubagentRunInfo } from '@deepseek-ai/dsh-subagent'
|
||||
import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek'
|
||||
import type { JsonRpcTransportPeer } from './transport.ts'
|
||||
|
||||
@@ -65,10 +65,11 @@ interface LocalAgentRecord {
|
||||
|
||||
/**
|
||||
* The SDK server over a booted harness context. Constructing it subscribes to
|
||||
* the context's `session/event`, `session/created`, `agent/created`, and
|
||||
* `subagent/end` events and forwards them to the host as notifications; the
|
||||
* subscriptions live until {@link shutdown}. One instance serves one transport
|
||||
* peer for the process lifetime — there is no re-`initialize`.
|
||||
* session, agent, and subagent lifecycle events, forwarding durable session
|
||||
* events and SDK-facing completion notifications while retaining local-run
|
||||
* identity across child disposal. The subscriptions live until
|
||||
* {@link shutdown}. One instance serves one transport peer for the process
|
||||
* lifetime — there is no re-`initialize`.
|
||||
*/
|
||||
export class HarnessSdkServer {
|
||||
private cwd = process.cwd()
|
||||
@@ -77,6 +78,7 @@ export class HarnessSdkServer {
|
||||
private readonly sessions = new Map<string, SessionRecord>()
|
||||
private readonly sessionCreations = new Map<string, Promise<SessionRecord>>()
|
||||
private readonly localAgents = new Map<SessionId, LocalAgentRecord>()
|
||||
private readonly localRuns = new Map<SessionId, LocalAgentRecord[]>()
|
||||
private readonly disposers: (() => void)[] = []
|
||||
private shutdownTask: Promise<Record<string, never>> | undefined
|
||||
private shuttingDown = false
|
||||
@@ -100,18 +102,38 @@ export class HarnessSdkServer {
|
||||
childSessionId: String(session.id),
|
||||
})
|
||||
}))
|
||||
// Cache runtime-local identity and optional lineage on creation: by the
|
||||
// time `subagent/end` fires the child agent may already be disposed and
|
||||
// gone from the registry. Parent lineage is not required by the provider
|
||||
// contract, so an empty record remains a load-bearing locality marker.
|
||||
// Cache runtime-local identity and optional lineage for each agent lifetime.
|
||||
// Parent lineage is not required by the provider contract, so an empty
|
||||
// record remains a load-bearing locality marker.
|
||||
this.disposers.push(ctx.on('agent/created', (agent) => {
|
||||
const parentSessionId = agent.session.header.parentSession
|
||||
this.localAgents.set(agent.id, parentSessionId === undefined ? {} : { parentSessionId })
|
||||
}))
|
||||
this.disposers.push(ctx.on('subagent/end', (info: SubagentRunEndInfo) => {
|
||||
this.disposers.push(ctx.on('agent/disposed', (agent) => {
|
||||
this.localAgents.delete(agent.id)
|
||||
}))
|
||||
// Snapshot locality per run. A provider may settle one run, continue the
|
||||
// same live child in another run, and dispose that child before the later
|
||||
// result settles. Consuming an agent-lifetime marker at the first end would
|
||||
// lose the later notification; this queue pairs each start with one end.
|
||||
this.disposers.push(ctx.on('subagent/start', (info: SubagentRunInfo) => {
|
||||
const agent = this.ctx.agents.get(info.id)
|
||||
const cachedLocalAgent = this.localAgents.get(info.id)
|
||||
this.localAgents.delete(info.id)
|
||||
const localAgent = cachedLocalAgent ?? (agent === undefined
|
||||
? undefined
|
||||
: agent.session.header.parentSession === undefined
|
||||
? {}
|
||||
: { parentSessionId: agent.session.header.parentSession })
|
||||
if (localAgent === undefined) return
|
||||
const runs = this.localRuns.get(info.id) ?? []
|
||||
runs.push(localAgent)
|
||||
this.localRuns.set(info.id, runs)
|
||||
}))
|
||||
this.disposers.push(ctx.on('subagent/end', (info: SubagentRunEndInfo) => {
|
||||
const agent = this.ctx.agents.get(info.id)
|
||||
const runs = this.localRuns.get(info.id)
|
||||
const cachedLocalAgent = runs?.shift()
|
||||
if (runs?.length === 0) this.localRuns.delete(info.id)
|
||||
// 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
|
||||
@@ -196,6 +218,7 @@ export class HarnessSdkServer {
|
||||
const records = [...this.sessions.values()]
|
||||
this.sessions.clear()
|
||||
this.localAgents.clear()
|
||||
this.localRuns.clear()
|
||||
const failures: unknown[] = []
|
||||
while (this.disposers.length > 0) {
|
||||
try {
|
||||
|
||||
@@ -11,7 +11,7 @@ import { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import * as agentCore from '@deepseek-ai/dsh-agent-core'
|
||||
import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
|
||||
import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek'
|
||||
import SubagentService, { type SubagentRunEndInfo } from '@deepseek-ai/dsh-subagent'
|
||||
import SubagentService, { type SubagentResult, type SubagentRunEndInfo } from '@deepseek-ai/dsh-subagent'
|
||||
import { HarnessSdkServer, type JsonRpcTransportPeer } from '../src/index.ts'
|
||||
|
||||
class FakeTransport implements JsonRpcTransportPeer {
|
||||
@@ -67,7 +67,13 @@ async function makeHarness(storageDir: string) {
|
||||
}
|
||||
|
||||
/** Drive the owning service so test lifecycle events carry the real parent scope. */
|
||||
async function settleSubagent(ctx: Context, parent: Agent, info: SubagentRunEndInfo): Promise<void> {
|
||||
async function settleSubagent(
|
||||
ctx: Context,
|
||||
parent: Agent,
|
||||
info: SubagentRunEndInfo,
|
||||
beforeSettle?: () => Promise<void>,
|
||||
): Promise<void> {
|
||||
const result = Promise.withResolvers<SubagentResult>()
|
||||
const disposeProvider = ctx.subagents.registerProvider({
|
||||
name: info.provider,
|
||||
capabilities: { outputSchema: false, depthLimit: false, toolFilter: false, persona: false },
|
||||
@@ -75,9 +81,7 @@ async function settleSubagent(ctx: Context, parent: Agent, info: SubagentRunEndI
|
||||
async start() {
|
||||
return {
|
||||
id: info.id,
|
||||
result: info.lastAssistantMessage === undefined
|
||||
? Promise.reject(new Error('synthetic infrastructure failure'))
|
||||
: Promise.resolve({ output: info.lastAssistantMessage, stopReason: info.stopReason }),
|
||||
result: result.promise,
|
||||
dispose: () => Promise.resolve(),
|
||||
}
|
||||
},
|
||||
@@ -88,6 +92,12 @@ async function settleSubagent(ctx: Context, parent: Agent, info: SubagentRunEndI
|
||||
prompt: [],
|
||||
signal: new AbortController().signal,
|
||||
})
|
||||
await beforeSettle?.()
|
||||
if (info.lastAssistantMessage === undefined) {
|
||||
result.reject(new Error('synthetic infrastructure failure'))
|
||||
} else {
|
||||
result.resolve({ output: info.lastAssistantMessage, stopReason: info.stopReason })
|
||||
}
|
||||
await run.result.then(() => undefined, () => undefined)
|
||||
await run.dispose()
|
||||
} finally {
|
||||
@@ -277,21 +287,17 @@ describe('HarnessSdkServer', () => {
|
||||
meta: { cwd: storageDir },
|
||||
agentOptions: { model: 'deepseek' },
|
||||
})
|
||||
// The backend may dispose the child before publishing its run outcome;
|
||||
// cached locality must survive with or without optional parent lineage.
|
||||
await handle.dispose()
|
||||
await parentlessHandle.dispose()
|
||||
await settleSubagent(ctx, parentHandle.agent, {
|
||||
provider: 'spawn',
|
||||
id: SessionId('child-session'),
|
||||
stopReason: 'completed',
|
||||
lastAssistantMessage: [{ type: 'text', text: 'child done' }],
|
||||
})
|
||||
}, () => handle.dispose())
|
||||
await settleSubagent(ctx, parentHandle.agent, {
|
||||
provider: 'spawn',
|
||||
id: SessionId('parentless-child-session'),
|
||||
stopReason: 'error',
|
||||
})
|
||||
}, () => parentlessHandle.dispose())
|
||||
|
||||
expect(transport.notifications).toContainEqual({
|
||||
method: 'subagent.finished',
|
||||
@@ -324,6 +330,49 @@ describe('HarnessSdkServer', () => {
|
||||
}
|
||||
})
|
||||
|
||||
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)
|
||||
try {
|
||||
const transport = new FakeTransport()
|
||||
const server = new HarnessSdkServer(ctx, transport)
|
||||
const parentHandle = await ctx.agents.create({
|
||||
sessionId: SessionId('continuation-parent'),
|
||||
meta: { cwd: storageDir },
|
||||
agentOptions: { model: 'deepseek' },
|
||||
})
|
||||
const childHandle = await ctx.agents.create({
|
||||
sessionId: SessionId('continuation-child'),
|
||||
meta: { cwd: storageDir, parentSession: SessionId('continuation-parent') },
|
||||
agentOptions: { model: 'deepseek' },
|
||||
})
|
||||
|
||||
await settleSubagent(ctx, parentHandle.agent, {
|
||||
provider: 'continuation',
|
||||
id: SessionId('continuation-child'),
|
||||
stopReason: 'completed',
|
||||
lastAssistantMessage: [{ type: 'text', text: 'first' }],
|
||||
})
|
||||
await settleSubagent(ctx, parentHandle.agent, {
|
||||
provider: 'continuation',
|
||||
id: SessionId('continuation-child'),
|
||||
stopReason: 'completed',
|
||||
lastAssistantMessage: [{ type: 'text', text: 'second' }],
|
||||
}, () => childHandle.dispose())
|
||||
|
||||
expect(transport.notifications.filter(notification =>
|
||||
notification.method === 'subagent.finished'
|
||||
&& notification.params?.childSessionId === 'continuation-child',
|
||||
)).toHaveLength(2)
|
||||
|
||||
await parentHandle.dispose()
|
||||
await server.shutdown()
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('falls back to live lineage and ignores runs without a local child session', async () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-subagent-fallback-'))
|
||||
const ctx = await makeHarness(storageDir)
|
||||
@@ -585,6 +634,6 @@ describe('HarnessSdkServer', () => {
|
||||
const server = new HarnessSdkServer(ctx, new FakeTransport())
|
||||
|
||||
await expect(server.shutdown()).rejects.toBe(listenerFailure)
|
||||
expect(on).toHaveBeenCalledTimes(4)
|
||||
expect(on).toHaveBeenCalledTimes(6)
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user