fix(subagent): await nested task completion across SDK and ACP
The unified delegation tool returns as soon as an activation accepts its prompt. SDK and ACP child runtimes could therefore reach root idle while an admitted fork was still running. Both external providers interpreted that idle edge as successful task completion and disposed the runtime, returning a waiting message while cancelling the nested work. Add the opt-in SDK session/wait request for existing SDK-owned sessions. It waits for root idle, managed descendants, and any resulting root turns, then requires stable idle state and Session sequence across the final check. Preserve raw session.status notifications and HarnessSession.run's receipt-to-next-idle API; only the task-owning SDK provider uses the new completion operation. Subscribe before prompt admission, validate and fold queued root events through the FIFO wait response, and retain accepted partial output on cancellation or transport failure. Keep ACP prompt ownership through descendant completion and later root summary turns, including their stop reasons and output-delivery failures. Race child waiting against the existing prompt cancellation controller; cancellation releases the prompt without permanently closing descendant admission or repinning the original model selection onto later turns. Session close retains child-first descendant teardown. Add real Loader/profile regressions with a gated nested fork for SDK and ACP completion, cancellation, and final-summary failure. Add focused coverage for unknown/malformed wait requests, SDK root summary ordering, ACP cancellation followed by another prompt, and later-summary image output draining/failure. Update the five owning bilingual README pairs. Validation: - pnpm run build - pnpm run typecheck - pnpm exec vitest run packages/subagent/subagent-dsh-sdk/tests/subagent-dsh-sdk.spec.ts packages/sdk/server/tests/server.spec.ts packages/acp/acp/tests/turns.spec.ts (123 tests at initial fix) - pnpm exec vitest run packages/acp/acp/tests/turns.spec.ts packages/sdk/client/tests/sdk-client.spec.ts (80 tests) - pnpm exec vitest run packages/acp/acp/tests/turns.spec.ts packages/sdk/server/tests/server.spec.ts (70 tests) - pnpm exec vitest run packages/acp/acp/tests (144 tests) - pnpm run test:e2e packages/subagent/subagent-dsh-sdk/tests/nested-delegation.e2e.ts (6 tests) - DSH_EXAMPLE_MODE=lib pnpm run test:e2e packages/subagent/subagent-dsh-sdk/tests/nested-delegation.e2e.ts (6 tests) - pnpm run test:docs (21 gates) - pnpm run doc-sync (42 gates) - pnpm run lint:contracts-ready - git diff --check The unchanged SDK next-idle API remains covered by its existing client tests; no raw notification or persisted Session event format changes.
This commit is contained in:
Tianyi Cui committed
1 parent
6f0174b5a2
commit
50da6f0b16
28 files changed
+428
-43
No files matched your search
@@ -27,8 +27,8 @@
|
||||
en: 4ed8e37834267edb
|
||||
zh: a32d5dedb4e8e919
|
||||
/deepseek-ai-dsh-acp/use-this-package/protocol-contract:
|
||||
en: 804b87d8ebed9f8d
|
||||
zh: 1b0c1c7269cdf040
|
||||
en: d1bb30b0dec7c57f
|
||||
zh: 836e5219e77cc83f
|
||||
/deepseek-ai-dsh-acp/understand-the-implementation:
|
||||
en: ea8afe89dbdd1934
|
||||
zh: 6567979f92f34afd
|
||||
|
||||
@@ -68,11 +68,13 @@ One connection can run several sessions at once, each independent. The calls a c
|
||||
| `session/resume` | A persisted inactive session whose canonical workspace is verified before composition; its log is restored without replaying old updates. |
|
||||
| `session/close` | Quiescent cancellation, update draining, descendant disposal, persistence flush, and disposal of only the addressed Agent scope. |
|
||||
| `session/set_config_option` | A serialized update to the advertised `model` or `reasoning_effort`, returning the complete resulting state. |
|
||||
| `session/prompt` | Ordered text, resource links, and supported images, one prompt at a time per session; settlement follows Agent idle and ordered update delivery. |
|
||||
| `session/prompt` | Ordered text, resource links, and supported images, one prompt at a time per session; settlement follows root and descendant completion plus ordered update delivery. |
|
||||
| `session/cancel` / `$/cancel_request` | The prompt-owned cancellation path; without an ACP prompt in flight it cancels autonomous work, while unknown session ids are no-ops. |
|
||||
| `session/update` | Committed assistant messages and thoughts, generic tool lifecycle, configuration changes, and context usage, serialized per session. |
|
||||
| `session/request_permission` | A permission prompt with one-shot allow/reject choices; your client can answer automatically. |
|
||||
|
||||
A prompt remains in flight while managed descendants run and while their results trigger later root turns. Its completion includes the latest root stop reason and any assistant output delivery failure; only the original prompt turn pins the model selection captured at admission. Cancellation stops the root and releases the prompt wait promptly, leaving the session usable for later prompts. Closing the session also drains its descendants.
|
||||
|
||||
Session configuration offers opaque provider/model choices from the live LLM service catalog and a `reasoning_effort` selector when the exact model declares one. A prompt snapshots that selection before asynchronous image admission and pins it across every model step in that turn; a concurrent option change applies to the next turn. ACP clients are trusted controllers: stdio MCP entries authorize their absolute commands and environment, HTTP entries authorize their absolute HTTP(S) URLs and headers, and any initial connection or discovery failure rolls back the unpublished Agent. Unsupported surfaces are omitted or rejected: `session/load`, deletion, fork, additional directories, SSE or ACP-transport MCP, modes, commands, plans, terminals, client filesystem operations, and elicitation.
|
||||
|
||||
-----
|
||||
|
||||
@@ -68,11 +68,13 @@ kind: "package-reference"
|
||||
| `session/resume` | 恢复一个已持久化且非活跃的会话；组合前校验其规范工作区，并恢复日志但不回放旧更新。 |
|
||||
| `session/close` | 停稳式取消、更新 drain、后代释放、持久化 flush，并且只释放指定 Agent 作用域。 |
|
||||
| `session/set_config_option` | 串行更新公布的 `model` 或 `reasoning_effort`，并返回完整结果状态。 |
|
||||
| `session/prompt` | 有序文本、资源链接与受支持图片，每个会话一次一个提示词；Agent 空闲且有序更新交付后才结算。 |
|
||||
| `session/prompt` | 有序文本、资源链接与受支持图片，每个会话一次一个提示词；根 Agent 与后代完成且有序更新交付后才结算。 |
|
||||
| `session/cancel` / `$/cancel_request` | 提示词所拥有的取消路径；没有进行中的 ACP 提示词时取消自主工作，未知会话 id 则为空操作。 |
|
||||
| `session/update` | 已提交 assistant 消息与 thought、通用工具生命周期、配置变化与上下文用量，按会话串行交付。 |
|
||||
| `session/request_permission` | 带一次性允许／拒绝选项的权限提示；你的客户端可以自动回答。 |
|
||||
|
||||
受管理后代运行期间，以及它们的结果触发根 Agent 后续轮次期间，提示词保持进行中。最终结果包含最新根轮次的停止原因与所有助手输出交付错误；只有原始提示词轮次固定使用接受时捕获的模型选择。取消会停止根 Agent 并及时释放提示词等待，会话仍可接受后续提示词。关闭会话还会释放其后代。
|
||||
|
||||
会话配置从实时 LLM（大语言模型）服务目录提供不透明的提供方／模型选项，并在确切模型声明推理选项时提供 `reasoning_effort`。提示词会在异步图片准入前快照该选择，并在该轮次的每个模型步骤中固定它；并发选项变更从下一轮次开始生效。ACP 客户端是受信控制器：stdio MCP 条目授权其绝对命令与环境，HTTP 条目授权其绝对 HTTP(S) URL 与 header；初始连接或发现失败会回滚尚未发布的 Agent。不支持的界面会被省略或拒绝：`session/load`、删除、fork、附加目录、SSE（Server-Sent Events）或 ACP 传输 MCP、mode、命令、计划、终端、客户端文件系统操作与 elicitation。
|
||||
|
||||
-----
|
||||
|
||||
@@ -19,10 +19,12 @@ import { mountAcpMcpServers } from './mcp.ts'
|
||||
import { AcpModelControl } from './model-control.ts'
|
||||
import { assistantUpdates, toolCallUpdate, toolResultUpdate } from './updates.ts'
|
||||
|
||||
/** The subagent teardown used without depending on the subagent package. */
|
||||
interface SubagentDrain {
|
||||
/** Subagent completion and teardown without depending on the subagent package. */
|
||||
interface SubagentLifecycle {
|
||||
/** Dispose subagent descendants below exact host-owned parents child-first. */
|
||||
drainDescendants(parents: readonly Agent[]): Promise<void>
|
||||
/** Observe currently managed descendants without closing future admission. */
|
||||
waitForChildren(parent: Agent): Promise<boolean>
|
||||
}
|
||||
|
||||
/** Inputs shared by fresh and resumed ACP session construction. */
|
||||
@@ -234,7 +236,7 @@ export class AcpSession {
|
||||
}
|
||||
|
||||
/**
|
||||
* Admit, enqueue, and settle one prompt at whole-Agent quiescence.
|
||||
* Admit one prompt and settle after the root, descendants, and updates finish.
|
||||
* @param params - standard ACP prompt request for this session.
|
||||
* @param imageEnabled - connection capability advertised at initialization.
|
||||
* @param requestSignal - JSON-RPC request cancellation signal.
|
||||
@@ -343,6 +345,10 @@ export class AcpSession {
|
||||
* @param event - committed durable event.
|
||||
*/
|
||||
onSessionEvent(session: Session, event: SessionEvent): void {
|
||||
if (event.type === 'turn/start' && this.inflight?.turn !== undefined) {
|
||||
this.inflight.turn = event.data.turn
|
||||
this.inflight.endReason = undefined
|
||||
}
|
||||
try {
|
||||
if (event.type === 'assistant/message') {
|
||||
const inflight = this.inflight?.turn === event.data.turn ? this.inflight : undefined
|
||||
@@ -439,7 +445,7 @@ export class AcpSession {
|
||||
} catch (error: unknown) {
|
||||
failures.push(new Error('ACP session activity drain failed', { cause: error }))
|
||||
}
|
||||
const subagents = this.ctx.get('subagents') as SubagentDrain | undefined
|
||||
const subagents = this.ctx.get('subagents') as SubagentLifecycle | undefined
|
||||
try {
|
||||
await subagents?.drainDescendants([this.agent])
|
||||
} catch (error: unknown) {
|
||||
@@ -486,7 +492,23 @@ export class AcpSession {
|
||||
void (async () => {
|
||||
await inflight.admissionDone
|
||||
if (inflight.messageQueued) {
|
||||
await this.agent.whenIdle()
|
||||
const subagents = this.ctx.get('subagents') as SubagentLifecycle | undefined
|
||||
const signal = inflight.admissionController.signal
|
||||
const cancelled = Promise.withResolvers<false>()
|
||||
const onAbort = (): void => { cancelled.resolve(false) }
|
||||
signal.addEventListener('abort', onAbort, { once: true })
|
||||
if (signal.aborted) onAbort()
|
||||
try {
|
||||
while (true) {
|
||||
await this.agent.whenIdle()
|
||||
if (inflight.cancelRequested) break
|
||||
const idleSeq = this.agent.session.seq
|
||||
const children = await Promise.race([subagents?.waitForChildren(this.agent), cancelled.promise])
|
||||
if (!children && this.agent.status === 'idle' && this.agent.session.seq === idleSeq) break
|
||||
}
|
||||
} finally {
|
||||
signal.removeEventListener('abort', onAbort)
|
||||
}
|
||||
await this.outputTail
|
||||
}
|
||||
/* v8 ignore next -- this prompt owns the slot until this exact settlement clears it. */
|
||||
|
||||
@@ -525,6 +525,90 @@ describe('ACP prompt lifecycle', () => {
|
||||
await vi.waitFor(() => { expect(messageText(harness!)).toBe('recovered') })
|
||||
})
|
||||
|
||||
it.each(['completed', 'error'] as const)('waits for descendants and reports the later %s summary turn', async (outcome) => {
|
||||
harness = await makeBridgeHarness({ script: [
|
||||
textResponse('waiting'),
|
||||
outcome === 'completed' ? textResponse('summary') : errorResponse('summary failed'),
|
||||
] })
|
||||
const sessionId = await newSession(harness)
|
||||
const parent = harness.ctx.agents.get(SessionId(sessionId))!
|
||||
const waiting = Promise.withResolvers<undefined>()
|
||||
const children = Promise.withResolvers<boolean>()
|
||||
const waitForChildren = vi.fn()
|
||||
.mockImplementationOnce(() => { waiting.resolve(undefined); return children.promise })
|
||||
.mockResolvedValue(false)
|
||||
harness.ctx.provide('subagents', { waitForChildren, drainDescendants: vi.fn() } as never)
|
||||
let settled = false
|
||||
const prompt = harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'delegate' }] })
|
||||
const observed = prompt.then(() => { settled = true }, () => { settled = true })
|
||||
await waiting.promise
|
||||
expect(settled).toBe(false)
|
||||
parent.followup(createUserMessage({ content: [{ type: 'text', text: 'child result' }], source: { kind: 'test' } }))
|
||||
children.resolve(true)
|
||||
if (outcome === 'completed') {
|
||||
await expect(prompt).resolves.toEqual({ stopReason: 'end_turn' })
|
||||
expect(messageText(harness)).toBe('waitingsummary')
|
||||
} else {
|
||||
await expect(prompt).rejects.toThrow('summary failed')
|
||||
}
|
||||
await observed
|
||||
})
|
||||
|
||||
it('cancels the child wait without closing the session to its next prompt', async () => {
|
||||
harness = await makeBridgeHarness({ script: [textResponse('waiting'), textResponse('next')] })
|
||||
const sessionId = await newSession(harness)
|
||||
const waiting = Promise.withResolvers<undefined>()
|
||||
const children = Promise.withResolvers<boolean>()
|
||||
const waitForChildren = vi.fn()
|
||||
.mockImplementationOnce(() => { waiting.resolve(undefined); return children.promise })
|
||||
.mockResolvedValue(false)
|
||||
const drainDescendants = vi.fn()
|
||||
harness.ctx.provide('subagents', { waitForChildren, drainDescendants } as never)
|
||||
const first = harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'delegate' }] })
|
||||
await waiting.promise
|
||||
await harness.client.cancel({ sessionId })
|
||||
await expect(first).resolves.toEqual({ stopReason: 'cancelled' })
|
||||
expect(drainDescendants).not.toHaveBeenCalled()
|
||||
children.resolve(true)
|
||||
await expect(harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'again' }] }))
|
||||
.resolves.toEqual({ stopReason: 'end_turn' })
|
||||
expect(messageText(harness)).toBe('waitingnext')
|
||||
})
|
||||
|
||||
it.each([false, true])('waits for the later summary output and reports its delivery failure (%s)', async (fail) => {
|
||||
const script: StreamChunk[][] = [textResponse('waiting')]
|
||||
harness = await makeBridgeHarness({ script })
|
||||
const ref = await harness.attachments!.saveImage({ data: Uint8Array.of(4), mediaType: 'image/png' })
|
||||
script.push([
|
||||
{ type: 'block-start', index: 0, blockType: 'image' },
|
||||
{ type: 'block-end', index: 0, block: { type: 'image', attachment: ref } },
|
||||
{ type: 'finish', reason: { kind: 'stop' } },
|
||||
])
|
||||
const sessionId = await newSession(harness)
|
||||
const parent = harness.ctx.agents.get(SessionId(sessionId))!
|
||||
const reading = Promise.withResolvers<undefined>()
|
||||
const delivery = Promise.withResolvers<undefined>()
|
||||
harness.attachments!.beforeRead = async () => {
|
||||
reading.resolve(undefined)
|
||||
await delivery.promise
|
||||
if (fail) throw new Error('summary image unavailable')
|
||||
}
|
||||
const waitForChildren = vi.fn().mockImplementationOnce(() => {
|
||||
parent.followup(createUserMessage({ content: [{ type: 'text', text: 'child result' }], source: { kind: 'test' } }))
|
||||
return Promise.resolve(true)
|
||||
}).mockResolvedValue(false)
|
||||
harness.ctx.provide('subagents', { waitForChildren, drainDescendants: vi.fn() } as never)
|
||||
let settled = false
|
||||
const prompt = harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'delegate' }] })
|
||||
const observed = prompt.then(() => { settled = true }, () => { settled = true })
|
||||
await reading.promise
|
||||
expect(settled).toBe(false)
|
||||
delivery.resolve(undefined)
|
||||
if (fail) await expect(prompt).rejects.toThrow('assistant output delivery failed')
|
||||
else await expect(prompt).resolves.toEqual({ stopReason: 'end_turn' })
|
||||
await observed
|
||||
})
|
||||
|
||||
it('a failed turn with no retry still rejects', async () => {
|
||||
harness = await makeBridgeHarness({ script: [errorResponse('terminal boom')] })
|
||||
let offered = 0
|
||||
|
||||
@@ -21,8 +21,8 @@
|
||||
en: 01c58dddd7e62de2
|
||||
zh: 6dbd46e244f3f135
|
||||
/deepseek-ai-dsh-sdk-client/use-this-package/lower-level-control-with-harnessclient:
|
||||
en: 8e7eb3a9b3f91404
|
||||
zh: 277a722586dcb549
|
||||
en: 8778901987da6e59
|
||||
zh: 3d09fe4b0b393a15
|
||||
/deepseek-ai-dsh-sdk-client/understand-the-implementation:
|
||||
en: 8773f50c51da85f0
|
||||
zh: 4649ebc69357deb0
|
||||
|
||||
@@ -51,6 +51,8 @@ The subprocess starts lazily on first use and stays owned by the instance across
|
||||
|
||||
`HarnessClient` is the protocol client under the run API: explicit `start()`, `initialize()`, `prompt()`, `request()`, and `close()`, plus notification subscriptions. `prompt()` returns the queued message id as soon as the runtime accepts it and never waits for agent activity. `subscribe(filter?)` returns a `NotificationSubscription` (awaitable `next()`, non-blocking `tryNext()`, async iteration); `subscribeSessionTree(id)` scopes to one session and the descendants discovered from `subagent.started` lineage edges — the runtime notifies for every session in its context, and scoping is client-side, exactly like the Python SDK.
|
||||
|
||||
Task-owning consumers can subscribe before `prompt()`, then call `request('session/wait', { sessionId })` and drain queued notifications with `tryNext()`. This waits for managed descendants and their root summary turns; `HarnessSession.run()` retains its next-idle interval. `validatedSessionEvent()` validates the event envelope and the assistant/turn fields consumed by SDK readers before they enter an output fold.
|
||||
|
||||
The client exports typed errors for every failure mode: `JsonRpcResponseError` (a wire error response, code and data preserved), `RequestTimeoutError` (a configured bound elapsed), `SdkProtocolError` (a response outside the documented protocol), and `TransportClosedError` (the runtime is gone — the message carries the exit code and a bounded stderr tail). `close()` requests protocol `shutdown` (bounded by `shutdownTimeoutMs`, default 1000 ms), then walks a stdin-EOF → SIGTERM → SIGKILL ladder until the process has exited; it is idempotent, and a closed client refuses reuse. `HarnessClientOptions.env` replaces the child environment entirely when given (`undefined` inherits the parent's); callers own credential policy — `scrubbedParentEnv` from `dsh-subprocess` is the shared scrub base for isolation-minded launches.
|
||||
|
||||
-----
|
||||
|
||||
@@ -51,6 +51,8 @@ console.log(result.finalResponse)
|
||||
|
||||
`HarnessClient` 是运行 API 之下的协议客户端：显式 `start()`、`initialize()`、`prompt()`、`request()` 与 `close()`，外加通知订阅。`prompt()` 在运行时接受排队消息后立即返回该消息的 id，绝不等待 agent 活动。`subscribe(filter?)` 返回 `NotificationSubscription`（可等待的 `next()`、非阻塞 `tryNext()`、异步迭代）；`subscribeSessionTree(id)` 把范围限定到一个会话及从 `subagent.started` 血缘边发现的后代——运行时对上下文内每个会话都发通知，范围限定在客户端完成，与 Python SDK 完全一致。
|
||||
|
||||
拥有完整任务的消费者可以先订阅，再调用 `prompt()`，随后调用 `request('session/wait', { sessionId })`，并通过 `tryNext()` 取完已排队的通知。这会等待受管理后代及其触发的根 Agent 摘要轮次；`HarnessSession.run()` 仍以首次空闲作为活动区间终点。`validatedSessionEvent()` 在事件进入输出折叠前，校验事件封装及 SDK 读取方使用的 assistant／turn 字段。
|
||||
|
||||
本客户端为每种失败模式导出类型化错误：`JsonRpcResponseError`（协议错误响应，保留 code 与 data）、`RequestTimeoutError`（配置的时限已到）、`SdkProtocolError`（响应超出文档化协议）、`TransportClosedError`（运行时已消失——消息携带退出码与有界 stderr 尾部）。`close()` 先请求协议 `shutdown`（受 `shutdownTimeoutMs` 约束，默认 1000 毫秒），然后走 stdin-EOF → SIGTERM → SIGKILL 阶梯直到进程退出；幂等，已关闭的客户端拒绝复用。`HarnessClientOptions.env` 给定时整体替换子进程环境（`undefined` 原样继承父进程环境）；凭据策略归调用方——`dsh-subprocess` 的 `scrubbedParentEnv` 是面向隔离启动的共享擦除基底。
|
||||
|
||||
-----
|
||||
|
||||
@@ -260,8 +260,12 @@ function validatedTurnEndReason(value: unknown): TurnEndReason {
|
||||
return value as TurnEndReason
|
||||
}
|
||||
|
||||
/** Validate the fields in a wire `session.event` envelope before returning the typed result. */
|
||||
function validatedSessionEvent(value: unknown): SessionEvent {
|
||||
/**
|
||||
* Validate the event envelope and assistant/turn fields consumed by SDK readers.
|
||||
* @param value - the raw `session.event` event payload.
|
||||
* @returns the event, or throws a protocol error for malformed consumed fields.
|
||||
*/
|
||||
export function validatedSessionEvent(value: unknown): SessionEvent {
|
||||
if (!isRecord(value) || typeof value.type !== 'string') {
|
||||
throw new SdkProtocolError(`session.event carried no event envelope: ${JSON.stringify(value)}`)
|
||||
}
|
||||
|
||||
@@ -9,7 +9,7 @@
|
||||
* @module @deepseek-ai/dsh-sdk-client
|
||||
*/
|
||||
|
||||
export { DeepSeekHarness, HarnessSession } from './api.ts'
|
||||
export { DeepSeekHarness, HarnessSession, validatedSessionEvent } from './api.ts'
|
||||
export type { RunOptions } from './api.ts'
|
||||
export {
|
||||
HarnessClient,
|
||||
|
||||
@@ -221,6 +221,7 @@ function sessionIdOf(params: Record<string, unknown> | undefined): string {
|
||||
}
|
||||
|
||||
const reader = createInterface({ input: process.stdin })
|
||||
const settledSessions = new Set<string>()
|
||||
reader.on('line', (line) => {
|
||||
if (line.trim().length === 0) return
|
||||
const frame = JSON.parse(line) as { id?: string | number; method?: string; params?: Record<string, unknown> }
|
||||
@@ -308,9 +309,13 @@ reader.on('line', (line) => {
|
||||
}
|
||||
runTurn(sessionId)
|
||||
notify('session.status', { sessionId, status: 'idle' })
|
||||
settledSessions.add(sessionId)
|
||||
respond({ messageId })
|
||||
return
|
||||
}
|
||||
case 'session/wait':
|
||||
if (settledSessions.has(sessionIdOf(frame.params))) respond({})
|
||||
return
|
||||
case 'shutdown':
|
||||
respond({})
|
||||
// An EOF-ignoring fake also refuses the protocol exit, so the client's
|
||||
|
||||
@@ -21,8 +21,8 @@
|
||||
en: be53c71505c0d1b4
|
||||
zh: 6210dee577715539
|
||||
/deepseek-ai-dsh-sdk-protocol/use-this-package/the-sdk-methods:
|
||||
en: 8cfc9efc99c7198d
|
||||
zh: ac00d94406fc2388
|
||||
en: 7a5a44f73174e9d9
|
||||
zh: 8b3e65f769e59a1e
|
||||
/deepseek-ai-dsh-sdk-protocol/use-this-package/payload-semantics:
|
||||
en: 3cf5d4ffbaf37f69
|
||||
zh: 23c4450e31c91d67
|
||||
|
||||
@@ -33,12 +33,13 @@ Wire one JSON-RPC 2.0 message per `\n`-terminated line over byte streams you own
|
||||
|
||||
### The SDK methods
|
||||
|
||||
Both wire ends share one method set: three client-to-server requests and four server-to-client notifications.
|
||||
Both wire ends share one method set: four client-to-server requests and four server-to-client notifications.
|
||||
|
||||
| Direction | Method | Payload types |
|
||||
|---|---|---|
|
||||
| client→server | `initialize` | `InitializeParams` → `InitializeResult` |
|
||||
| client→server | `session/prompt` | `SessionPromptParams` → `SessionPromptResult` (durable enqueue receipt) |
|
||||
| client→server | `session/wait` | `SessionWaitParams` → `{}` |
|
||||
| client→server | `shutdown` | no params → `{}` |
|
||||
| server→client | `session.event` | `SessionEventNotification` (every session in the runtime, unfiltered) |
|
||||
| server→client | `session.status` | `SessionStatusNotification` (whole-agent `running`/`idle` transition) |
|
||||
@@ -47,6 +48,8 @@ Both wire ends share one method set: three client-to-server requests and four se
|
||||
|
||||
`HarnessSdkRequestMap` and `HarnessSdkNotificationMap` index these shapes by method name; the package root exports them together with the transport.
|
||||
|
||||
`session/wait` accepts `{ sessionId }` for an existing SDK-owned session and returns `{}` after the root stays idle with no managed descendants. Unknown ids are rejected. Notifications committed during the wait precede its response on the same transport; raw `session.status` transitions retain their whole-agent meaning.
|
||||
|
||||
### Payload semantics
|
||||
|
||||
`SessionPromptResult.messageId` identifies the queued user message; it does not identify a later assistant message, turn ending, or prompt result. `SdkPromptContentBlock` accepts ordinary durable content plus `SdkEncodedImageBlock { type: "image", data, mimeType }`; the server converts encoded images to durable references before enqueue. `InitializeParams.reasoningEffort` is an optional non-empty adapter-owned identifier for the selected provider/model route; omission preserves that model's default. `InitializeParams.maxTokens` is an optional positive safe integer that caps each conversation-model output for SDK-created agents and their in-process descendants; omission lets the selected adapter's exact-model default apply. The server resolves the exact route during initialization and rejects `session/prompt` until that handshake succeeds, so a missing adapter, unavailable model, or unsupported effort cannot fall back to constructor defaults. `SubagentFinishedNotification.lastAssistantMessage` carries the child's last non-empty assistant message, or its accumulated assistant text when no such message exists; the field is absent when the child produced neither. `serverInfo.name` stays the wire-stable `deepseek-harness-sdk-runtime`. Notification payloads depend on `SessionEvent` (`dsh-session`), `ContentBlock` (`dsh-llm`), and `SubagentStopReason` (`dsh-subagent`), so the session vocabulary is part of the wire contract.
|
||||
|
||||
@@ -33,12 +33,13 @@ kind: "package-library"
|
||||
|
||||
### SDK 方法
|
||||
|
||||
两个协议端共享同一套方法：三个客户端到服务端请求与四个服务端到客户端通知。
|
||||
两个协议端共享同一套方法：四个客户端到服务端请求与四个服务端到客户端通知。
|
||||
|
||||
| 方向 | 方法 | 载荷类型 |
|
||||
|---|---|---|
|
||||
| client→server | `initialize` | `InitializeParams` → `InitializeResult` |
|
||||
| client→server | `session/prompt` | `SessionPromptParams` → `SessionPromptResult`（持久入队回执） |
|
||||
| client→server | `session/wait` | `SessionWaitParams` → `{}` |
|
||||
| client→server | `shutdown` | 无参数 → `{}` |
|
||||
| server→client | `session.event` | `SessionEventNotification`（运行时内每个会话，不过滤） |
|
||||
| server→client | `session.status` | `SessionStatusNotification`（整个 agent（智能体）的 `running`/`idle` 转换） |
|
||||
@@ -47,6 +48,8 @@ kind: "package-library"
|
||||
|
||||
`HarnessSdkRequestMap` 与 `HarnessSdkNotificationMap` 按方法名索引这些结构；包根与传输一起导出它们。
|
||||
|
||||
`session/wait` 接受已有 SDK 会话的 `{ sessionId }`，在根 Agent 持续空闲且没有受管理后代时返回 `{}`。未知 id 会被拒绝。等待期间提交的通知在同一传输上先于响应到达；原始 `session.status` 转换仍表示整个 Agent 的状态。
|
||||
|
||||
### 载荷语义
|
||||
|
||||
`SessionPromptResult.messageId` 标识已排队的用户消息；它不标识后续的助手消息、轮次结束或提示词结果。`SdkPromptContentBlock` 接受普通持久内容以及 `SdkEncodedImageBlock { type: "image", data, mimeType }`；服务器在入队前把编码图像转换为持久引用。`InitializeParams.reasoningEffort` 是所选提供方／模型路由可选的非空适配器自有标识符；省略时保留该模型的默认值。`InitializeParams.maxTokens` 是可选的正安全整数，用于限制 SDK 创建的 agent 及其进程内后代的每次对话模型输出；省略时应用所选适配器的确切模型默认值。服务器会在初始化期间解析确切路由，并在握手成功前拒绝 `session/prompt`，因此缺少适配器、模型不可用或推理强度不受支持时，不会回退到构造期默认值。`SubagentFinishedNotification.lastAssistantMessage` 携带子 agent 最后一条非空 assistant 消息；若不存在这类消息，则携带其累积的 assistant 文本；子 agent 两种输出均未产生时，该字段缺省。`serverInfo.name` 的协议值固定为 `deepseek-harness-sdk-runtime`。通知载荷依赖 `SessionEvent`（`dsh-session`）、`ContentBlock`（`dsh-llm`）与 `SubagentStopReason`（`dsh-subagent`），因此会话词汇是协议格式约定的一部分。
|
||||
|
||||
@@ -22,6 +22,7 @@ export type {
|
||||
SessionStatusNotification,
|
||||
SessionPromptParams,
|
||||
SessionPromptResult,
|
||||
SessionWaitParams,
|
||||
SubagentFinishedNotification,
|
||||
SubagentStartedNotification,
|
||||
} from './types.ts'
|
||||
@@ -1,5 +1,5 @@
|
||||
/**
|
||||
* Named wire types for the DeepSeek Harness SDK runtime protocol: the three
|
||||
* Named wire types for the DeepSeek Harness SDK runtime protocol: the four
|
||||
* request/result pairs and the four server-to-client notification payloads
|
||||
* exchanged over the newline-delimited JSON-RPC stdio transport. The server
|
||||
* plugin (`@deepseek-ai/dsh-sdk-jsonrpc-server`) and SDK clients share these shapes;
|
||||
@@ -58,6 +58,12 @@ export interface SessionPromptResult {
|
||||
messageId: string
|
||||
}
|
||||
|
||||
/** Wait for an existing SDK session and its managed descendants to finish. */
|
||||
export interface SessionWaitParams {
|
||||
/** An SDK-owned session id; waiting never creates a session. */
|
||||
sessionId: string
|
||||
}
|
||||
|
||||
/** Deployment-mapped SDK outcome: `ok` for an accepted result, `error` otherwise. */
|
||||
export type SdkRunStatus = 'ok' | 'error'
|
||||
|
||||
@@ -115,5 +121,6 @@ export interface HarnessSdkNotificationMap {
|
||||
export interface HarnessSdkRequestMap {
|
||||
'initialize': { params: InitializeParams; result: InitializeResult }
|
||||
'session/prompt': { params: SessionPromptParams; result: SessionPromptResult }
|
||||
'session/wait': { params: SessionWaitParams; result: Record<string, never> }
|
||||
'shutdown': { params: undefined; result: Record<string, never> }
|
||||
}
|
||||
@@ -27,8 +27,8 @@
|
||||
en: 784b68b5a498f2d5
|
||||
zh: 2c3c5a1490845179
|
||||
/deepseek-ai-dsh-sdk-jsonrpc-server/use-this-package/what-sdk-clients-can-do:
|
||||
en: 2b1b52a7467c557e
|
||||
zh: 31f58d32469f9d50
|
||||
en: 39ed6a280988103b
|
||||
zh: c69e2900d9641d76
|
||||
/deepseek-ai-dsh-sdk-jsonrpc-server/use-this-package/shutdown-and-exit:
|
||||
en: 08c3858f53090f39
|
||||
zh: dfae960e11515b06
|
||||
|
||||
@@ -47,6 +47,8 @@ Stdout carries only JSON-RPC frames, so clients can parse every byte; diagnostic
|
||||
|
||||
`initialize` is the runtime-readiness boundary: when the server is mounted by a Loader composition, it waits for the current plugin tree to settle before replying, so async sibling capabilities such as initial MCP tool discovery are visible to the first prompt. The handshake returns the wire-stable identity `deepseek-harness-sdk-runtime`. The server validates the provider/model route and optional non-empty `reasoningEffort` through the selected adapter before it stores them; omission stores no effort, so the model retains its own default. An optional positive `maxTokens` becomes the request output cap of each SDK-created agent and its in-process descendants, while omission applies the selected adapter or provider route default. JSON-RPC requests may dispatch concurrently, so `session/prompt` rejects until one `initialize` has completed successfully; clients must await the handshake before sending prompts. An accepted prompt queues one identified user message and immediately returns `{ messageId }`; the server then streams every durable fact as `session.event` and every whole-agent lifecycle transition as `session.status`. It does not assign an assistant message or `turn/end` to a prompt, and independent requests may enqueue more work on the same session. Persistence roots and persona come from the surrounding composition.
|
||||
|
||||
`session/wait` observes an existing SDK-owned session through managed descendant completion and any resulting root turns. It rejects unknown or retired roots, never creates a session, and responds after the root stays idle across the descendant check. Session notifications emitted before completion precede the response on the stdio transport.
|
||||
|
||||
### Shutdown and exit
|
||||
|
||||
The plugin answers `shutdown`, flushes the response, disposes the root context so SDK-owned agents, subscriptions, and persistence reach quiescence, then exits 0. EOF and signal exits belong to the app bin, which also disposes the root context. Unloading only this plugin stops serving without exiting the process.
|
||||
|
||||
@@ -47,6 +47,8 @@ Stdout 只承载 JSON-RPC 帧，客户端可以逐字节解析；诊断信息应
|
||||
|
||||
`initialize` 是运行时就绪边界：服务器由 Loader 组合挂载时，会等待当前插件树完成所有加载任务后再响应，因此首次提示词能够看到 MCP 初始工具发现等异步同级能力。握手返回协议稳定标识 `deepseek-harness-sdk-runtime`。服务器会通过所选适配器校验提供方／模型路由与可选的非空 `reasoningEffort`，再保存这些值；省略时不会保存推理强度，因此模型保留自身默认值。可选的正整数 `maxTokens` 会成为每个 SDK 创建的 agent 及其进程内后代的请求输出上限，省略时则应用所选适配器或提供方路由的默认值。JSON-RPC 请求可能并发分派，因此在一次 `initialize` 成功完成之前，`session/prompt` 会拒绝；客户端必须等待握手完成后再发送提示词。已接受的提示词会把一条带标识的用户消息排入队列，并立即返回 `{ messageId }`；服务器随后把每个持久事实作为 `session.event`、把整个 agent 生命周期的每次状态转换作为 `session.status` 流式发出。它不会把某条助手消息或 `turn/end` 归属于某个提示词，同一会话上的独立请求可以继续排入更多工作。持久化根目录与 persona 来自外围组合。
|
||||
|
||||
`session/wait` 观察已有 SDK 会话，直到受管理后代及其触发的根 Agent 后续轮次完成。它拒绝未知或已释放的根 Agent，不会创建会话，并在后代检查期间根 Agent 持续空闲后响应。完成前发出的会话通知在 stdio 传输上先于响应到达。
|
||||
|
||||
### 关闭与退出
|
||||
|
||||
插件应答 `shutdown`，刷新响应并 dispose 根上下文，使 SDK 持有的 agent、订阅与持久化达到完全停稳，然后以 0 退出。EOF 与信号退出归 app bin 负责，后者也会 dispose 根上下文。仅卸载此插件会停止服务，但不会退出进程。
|
||||
|
||||
@@ -23,6 +23,7 @@ import type {
|
||||
SessionEventNotification,
|
||||
SessionPromptParams,
|
||||
SessionPromptResult,
|
||||
SessionWaitParams,
|
||||
SdkEncodedImageBlock,
|
||||
SubagentFinishedNotification,
|
||||
SubagentStartedNotification,
|
||||
@@ -194,6 +195,28 @@ export class HarnessSdkJsonRpcServer {
|
||||
return { messageId: message.id }
|
||||
}
|
||||
|
||||
/**
|
||||
* Wait for an existing owned Agent and its managed descendants to finish.
|
||||
* The response follows all Session notifications emitted during the wait.
|
||||
* @param params - the SDK-owned session to observe without creating one.
|
||||
* @returns an empty result after the root stays idle across the descendant check.
|
||||
*/
|
||||
async wait(params: SessionWaitParams): Promise<Record<string, never>> {
|
||||
if (!this.initialized) throw new Error('SDK server is not initialized')
|
||||
const rec = this.sessions.get(params.sessionId)
|
||||
if (rec === undefined) throw new Error(`unknown SDK session: ${params.sessionId}`)
|
||||
const agent = rec.handle.agent
|
||||
const subagents = this.ctx.get('subagents')
|
||||
while (true) {
|
||||
this.assertLiveAgent(rec, params.sessionId)
|
||||
await agent.whenIdle()
|
||||
const idleSeq = agent.session.seq
|
||||
const children = await subagents?.waitForChildren(agent)
|
||||
this.assertLiveAgent(rec, params.sessionId)
|
||||
if (!children && agent.status === 'idle' && agent.session.seq === idleSeq) return {}
|
||||
}
|
||||
}
|
||||
|
||||
private assertLiveAgent(rec: SessionRecord, sessionId: string): void {
|
||||
if (this.ctx.agents.get(rec.handle.agent.id) !== rec.handle.agent) {
|
||||
throw new Error(`session agent was disposed outside the server: ${sessionId}`)
|
||||
@@ -251,6 +274,11 @@ export class HarnessSdkJsonRpcServer {
|
||||
return this.initialize(params as unknown as InitializeParams)
|
||||
case 'session/prompt':
|
||||
return this.prompt(params as unknown as SessionPromptParams)
|
||||
case 'session/wait': {
|
||||
const sessionId = params?.sessionId
|
||||
if (typeof sessionId !== 'string') throw new TypeError('session/wait requires a sessionId string')
|
||||
return this.wait({ sessionId })
|
||||
}
|
||||
case 'shutdown':
|
||||
return this.shutdown()
|
||||
default:
|
||||
|
||||
@@ -14,7 +14,7 @@ import AgentRegistry, { type Agent, type AgentHandle } from '@deepseek-ai/dsh-ag
|
||||
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
|
||||
import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit'
|
||||
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import SessionStore, { SessionId, type SessionEvent } from '@deepseek-ai/dsh-session'
|
||||
import JsonlSessionPersistence from '@deepseek-ai/dsh-session-persistence-jsonl'
|
||||
import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek-api-key'
|
||||
import SubagentRuntime, { SubagentRunId, type SubagentStartRequest, type SubagentRun, type SubagentResult, type SubagentRunEndInfo } from '@deepseek-ai/dsh-subagent'
|
||||
@@ -127,6 +127,55 @@ async function settleSubagent(
|
||||
}
|
||||
|
||||
describe('HarnessSdkJsonRpcServer', () => {
|
||||
it('waits only for an existing SDK-owned session and includes a child-triggered root turn', async () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-wait-'))
|
||||
const ctx = await makeHarness(storageDir)
|
||||
const transport = new FakeTransport()
|
||||
const server = new HarnessSdkJsonRpcServer(ctx, transport)
|
||||
class Adapter extends LlmAdapter {
|
||||
override resolveModel(provider: string, model: string): Promise<LlmResolvedModelInfo> {
|
||||
return Promise.resolve({ provider, id: model, name: model })
|
||||
}
|
||||
async * stream(): AsyncIterable<StreamChunk> {
|
||||
yield { type: 'block-start', index: 0, blockType: 'text' }
|
||||
yield { type: 'text-delta', index: 0, text: 'answer' }
|
||||
yield { type: 'block-end', index: 0, block: { type: 'text', text: 'answer' } }
|
||||
yield { type: 'finish', reason: { kind: 'stop' } }
|
||||
}
|
||||
}
|
||||
ctx.llm.registerAdapter(['mock'], new Adapter())
|
||||
try {
|
||||
await expect(server.handleRequest('session/wait', { sessionId: 'missing' })).rejects.toThrow('not initialized')
|
||||
await server.initialize({ cwd: storageDir, provider: 'mock', model: 'mock' })
|
||||
await expect(server.handleRequest('session/wait', {})).rejects.toThrow('sessionId string')
|
||||
await expect(server.handleRequest('session/wait', { sessionId: 'missing' })).rejects.toThrow('unknown SDK session')
|
||||
expect(ctx.agents.get(SessionId('missing'))).toBeUndefined()
|
||||
await server.prompt({ sessionId: 'main', contentBlocks: [{ type: 'text', text: 'delegate' }] })
|
||||
const parent = ctx.agents.get(SessionId('main'))!
|
||||
const observed = Promise.withResolvers<undefined>()
|
||||
const children = Promise.withResolvers<boolean>()
|
||||
vi.spyOn(ctx.subagents, 'waitForChildren')
|
||||
.mockImplementationOnce(() => { observed.resolve(undefined); return children.promise })
|
||||
.mockResolvedValue(false)
|
||||
let settled = false
|
||||
const waiting = server.handleRequest('session/wait', { sessionId: 'main' }).then((result) => { settled = true; return result })
|
||||
await observed.promise
|
||||
expect(settled).toBe(false)
|
||||
parent.followup(createUserMessage({ content: [{ type: 'text', text: 'child result' }], source: { kind: 'user' } }))
|
||||
children.resolve(true)
|
||||
await expect(waiting).resolves.toEqual({})
|
||||
expect(transport.notifications.filter(notification => notification.method === 'session.event'
|
||||
&& notification.params?.sessionId === 'main'
|
||||
&& (notification.params.event as SessionEvent).type === 'turn/end')).toHaveLength(2)
|
||||
await server.shutdown()
|
||||
await expect(server.handleRequest('session/wait', { sessionId: 'main' })).rejects.toThrow('unknown SDK session')
|
||||
} finally {
|
||||
await server.shutdown()
|
||||
await ctx.fiber.dispose()
|
||||
await rm(storageDir, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
|
||||
it('creates a harness agent and calls the configured OpenAI-compatible endpoint', { timeout: 15_000 }, async () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-'))
|
||||
const llmServer = await mockCompletionServer()
|
||||
|
||||
@@ -39,8 +39,8 @@
|
||||
en: 1bed6c1577997df6
|
||||
zh: 8c99f132377f6ca2
|
||||
/deepseek-ai-dsh-subagent-dsh-sdk/understand-the-implementation/run-flow:
|
||||
en: 4f1b563c99d838d8
|
||||
zh: abdbf2cd61a7895d
|
||||
en: 6c50e4e8dc5e0698
|
||||
zh: 9d0011e28341bbde
|
||||
/deepseek-ai-dsh-subagent-dsh-sdk/understand-the-implementation/stop-reason-mapping:
|
||||
en: e040a57202916f91
|
||||
zh: 196bf38f0f0db310
|
||||
|
||||
@@ -104,7 +104,7 @@ This section explains how the backend drives a child Harness runtime and where t
|
||||
|
||||
### Run flow
|
||||
|
||||
A start resolves the child's working directory and one process-wide SDK route before spawning. Each declared `request.agentOptions` field (`provider`, `model`, `reasoningEffort`, or `maxTokens`) overrides the matching provider-instance default; omission preserves the configured provider/model and optional cap, while reasoning effort remains absent unless the request supplies it. The provider spawns the runtime through the SDK client and completes the `initialize` handshake, including exact-model and effort validation, before it fulfills. A route, spawn, handshake, or pre-publication cancellation failure rejects only after the subprocess is reaped; a working-directory resolution failure rejects before spawning. After publication the provider owns one SDK activity and reads the child's answer from its session events: the last complete non-empty `assistant/message` (an empty-content message that records usage is skipped), or the accumulated `text-delta` stream when no such message exists. Disposal is idempotent: it settles the result locally as `aborted`, sends a bounded protocol `shutdown` request, then escalates through stdin EOF → SIGTERM → SIGKILL to actual exit. The activation manager owns that run after acceptance; the caller's startup signal no longer cancels the child. Each activation executes once, and its receipt cancels execution and awaits cleanup.
|
||||
A start resolves the child's working directory and one process-wide SDK route before spawning. Each declared `request.agentOptions` field (`provider`, `model`, `reasoningEffort`, or `maxTokens`) overrides the matching provider-instance default; omission preserves the configured provider/model and optional cap, while reasoning effort remains absent unless the request supplies it. The provider spawns the runtime through the SDK client and completes the `initialize` handshake, including exact-model and effort validation, before it fulfills. A route, spawn, handshake, or pre-publication cancellation failure rejects only after the subprocess is reaped; a working-directory resolution failure rejects before spawning. After publication the provider subscribes to the child Session, queues its prompt, and waits through `session/wait` for managed descendants and subsequent root turns. It reads the child's answer from the ordered session events: the last complete non-empty `assistant/message` (an empty-content message that records usage is skipped), or the accumulated `text-delta` stream when no such message exists. Disposal is idempotent: it settles the result locally as `aborted`, sends a bounded protocol `shutdown` request, then escalates through stdin EOF → SIGTERM → SIGKILL to actual exit. The activation manager owns that run after acceptance; the caller's startup signal no longer cancels the child. Each activation executes once, and its receipt cancels execution and awaits cleanup.
|
||||
|
||||
### Stop-reason mapping
|
||||
|
||||
|
||||
@@ -104,7 +104,7 @@ kind: "package-reference"
|
||||
|
||||
### 运行流程
|
||||
|
||||
一次启动会在 spawn 前解析子进程工作目录与一条进程级 SDK 路由。`request.agentOptions` 中每个已声明字段（`provider`、`model`、`reasoningEffort` 或 `maxTokens`）都会覆盖对应的提供方实例默认值；省略时保留已配置的提供方／模型与可选上限，而推理强度只有在请求提供时才会出现。随后，提供方通过 SDK 客户端 spawn 运行时，并在履行前完成 `initialize` 握手，其中包括确切模型与推理强度校验。路由、spawn、握手或发布前取消失败时，只会在子进程被回收后拒绝；工作目录解析失败则会在尚未 spawn 任何内容时拒绝。发布后，提供方拥有一段 SDK 活动，并从子会话事件中读取答案：最后一条完整且非空的 `assistant/message`（记录 usage 的空内容消息会被跳过）；若没有这类消息，则取累积的 `text-delta` 流。dispose 是幂等的：先在本地把结果确定为 `aborted`，发出有界的协议 `shutdown` 请求，再经 stdin EOF → SIGTERM → SIGKILL 升级到实际退出。任务接受后由 activation 管理器拥有该运行；此后调用方的启动信号不再取消子任务。每个 activation 只执行一次，其凭据用于取消执行并等待清理。
|
||||
一次启动会在 spawn 前解析子进程工作目录与一条进程级 SDK 路由。`request.agentOptions` 中每个已声明字段（`provider`、`model`、`reasoningEffort` 或 `maxTokens`）都会覆盖对应的提供方实例默认值；省略时保留已配置的提供方／模型与可选上限，而推理强度只有在请求提供时才会出现。随后，提供方通过 SDK 客户端 spawn 运行时，并在履行前完成 `initialize` 握手，其中包括确切模型与推理强度校验。路由、spawn、握手或发布前取消失败时，只会在子进程被回收后拒绝；工作目录解析失败则会在尚未 spawn 任何内容时拒绝。发布后，提供方先订阅子会话，再排入提示词，并通过 `session/wait` 等待受管理后代与根 Agent 后续轮次。它从有序会话事件中读取答案：最后一条完整且非空的 `assistant/message`（记录 usage 的空内容消息会被跳过）；若没有这类消息，则取累积的 `text-delta` 流。dispose 是幂等的：先在本地把结果确定为 `aborted`，发出有界的协议 `shutdown` 请求，再经 stdin EOF → SIGTERM → SIGKILL 升级到实际退出。任务接受后由 activation 管理器拥有该运行；此后调用方的启动信号不再取消子任务。每个 activation 只执行一次，其凭据用于取消执行并等待清理。
|
||||
|
||||
### 停止原因映射
|
||||
|
||||
|
||||
@@ -15,13 +15,13 @@ import { brandString } from '@deepseek-ai/dsh-brand'
|
||||
import {
|
||||
DeepSeekHarness,
|
||||
type DeepSeekHarnessOptions,
|
||||
type HarnessNotification,
|
||||
JsonRpcResponseError,
|
||||
SdkProtocolError,
|
||||
TransportClosedError,
|
||||
validatedSessionEvent,
|
||||
} from '@deepseek-ai/dsh-sdk-client'
|
||||
import type { ContentBlock, ReasoningEffortId } from '@deepseek-ai/dsh-llm'
|
||||
import type { SessionEvent, SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
|
||||
import type { SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
|
||||
import type { SubagentResult, SubagentRun, SubagentStartRequest, SubagentStopReason } from '@deepseek-ai/dsh-subagent'
|
||||
import { AssistantOutputFold, settleRunResult, subprocessRunHandle } from '@deepseek-ai/dsh-subagent'
|
||||
import { scrubbedParentEnv } from '@deepseek-ai/dsh-subprocess'
|
||||
@@ -299,9 +299,16 @@ export async function startSdkRun(request: SubagentStartRequest, spec: SdkRunSpe
|
||||
// The child's final answer under the seam's canonical selection rule
|
||||
// (`AssistantOutputFold`); a partial answer survives cancel and error paths.
|
||||
const fold = new AssistantOutputFold()
|
||||
const observe = (notification: HarnessNotification): void => {
|
||||
if (notification.method !== 'session.event' || notification.params.sessionId !== childSessionId) return
|
||||
fold.push(notification.params.event as SessionEvent)
|
||||
const subscription = harness.client.subscribe(notification =>
|
||||
notification.method === 'session.event' && notification.params.sessionId === childSessionId)
|
||||
let lastEnd: TurnEndReason | undefined
|
||||
const drainNotifications = (): void => {
|
||||
let notification
|
||||
while ((notification = subscription.tryNext()) !== undefined) {
|
||||
const event = validatedSessionEvent(notification.params.event)
|
||||
fold.push(event)
|
||||
if (event.type === 'turn/end') lastEnd = event.data.reason
|
||||
}
|
||||
}
|
||||
const collectOutput = (): readonly ContentBlock[] => fold.collect() ?? []
|
||||
const teardown = async (): Promise<void> => {
|
||||
@@ -313,21 +320,29 @@ export async function startSdkRun(request: SubagentStartRequest, spec: SdkRunSpe
|
||||
}
|
||||
}
|
||||
|
||||
// Race the child turn against local cancellation; the shared settlement
|
||||
// Race the child task against local cancellation; the shared settlement
|
||||
// flattens failures under the seam's never-reject contract.
|
||||
let diagnostic: string | undefined
|
||||
let accepted = false
|
||||
const result: Promise<SubagentResult> = settleRunResult({
|
||||
attempt: async () => {
|
||||
try {
|
||||
const turn = await Promise.race([
|
||||
harness.session(childSessionId).run(request.prompt, { onNotification: observe }),
|
||||
cancelSettled.then(() => 'cancelled' as const),
|
||||
])
|
||||
if (turn === 'cancelled') return { output: collectOutput(), stopReason: 'aborted' }
|
||||
const lastEnd = turn.events.findLast(
|
||||
(event): event is Extract<SessionEvent, { type: 'turn/end' }> => event.type === 'turn/end',
|
||||
)
|
||||
const outcome = sdkChildOutcome(lastEnd?.data.reason)
|
||||
try {
|
||||
await Promise.race([
|
||||
(async () => {
|
||||
await harness.client.prompt(childSessionId, request.prompt)
|
||||
accepted = true
|
||||
await harness.client.request('session/wait', { sessionId: childSessionId })
|
||||
})(),
|
||||
cancelSettled,
|
||||
])
|
||||
} finally {
|
||||
// The wait response follows committed notifications on the same pipe.
|
||||
// Cancellation and transport failure retain every event already received.
|
||||
if (accepted) drainNotifications()
|
||||
}
|
||||
if (flags.cancelled) return { output: collectOutput(), stopReason: 'aborted' }
|
||||
const outcome = sdkChildOutcome(lastEnd)
|
||||
diagnostic = outcome.diagnostic
|
||||
return {
|
||||
output: collectOutput(),
|
||||
@@ -336,6 +351,8 @@ export async function startSdkRun(request: SubagentStartRequest, spec: SdkRunSpe
|
||||
} catch (error: unknown) {
|
||||
diagnostic = failureDiagnostic(sdkFailure(error, 'session-run').facts)
|
||||
throw error
|
||||
} finally {
|
||||
subscription.close()
|
||||
}
|
||||
},
|
||||
collectOutput,
|
||||
|
||||
@@ -0,0 +1,53 @@
|
||||
/** Gated nested fork for the SDK and ACP task-completion profile tests. */
|
||||
import { existsSync, writeFileSync } from 'node:fs'
|
||||
import { setTimeout } from 'node:timers/promises'
|
||||
import type { Context } from '@deepseek-ai/cordis'
|
||||
import { LlmAdapter, ToolCallId, type GenerateOptions, type LlmResolvedModelInfo, type StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
|
||||
class NestedAdapter extends LlmAdapter {
|
||||
override resolveModel(provider: string, model: string): Promise<LlmResolvedModelInfo> {
|
||||
return Promise.resolve({ provider, id: model, name: model })
|
||||
}
|
||||
|
||||
async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
|
||||
const messages = options.messages.flatMap(message => message.content
|
||||
.filter(block => block.type === 'text').map(block => block.text)).join('\n')
|
||||
let answer: string
|
||||
if (messages.includes('nested-child-task')) {
|
||||
writeFileSync(process.env.DSH_TEST_CHILD_READY!, 'ready\n')
|
||||
while (!existsSync(process.env.DSH_TEST_CHILD_RELEASE!)) {
|
||||
await setTimeout(10, undefined, { signal: options.signal })
|
||||
}
|
||||
answer = 'nested child result'
|
||||
} else if (!options.messages.some(message => message.role === 'tool')) {
|
||||
const args = JSON.stringify({ description: 'nested fork', prompt: 'nested-child-task' })
|
||||
const id = ToolCallId('nested-fork')
|
||||
yield { type: 'block-start', index: 0, blockType: 'tool-call' }
|
||||
yield { type: 'tool-call-delta', index: 0, id, name: 'subagent_fork', argumentsDelta: args }
|
||||
yield { type: 'block-end', index: 0, block: { type: 'tool-call', id, name: 'subagent_fork', arguments: args } }
|
||||
yield { type: 'finish', reason: { kind: 'tool-calls' } }
|
||||
return
|
||||
} else {
|
||||
answer = messages.includes('nested child result') ? 'parent summary after child' : 'waiting for child'
|
||||
}
|
||||
yield { type: 'block-start', index: 0, blockType: 'text' }
|
||||
yield { type: 'text-delta', index: 0, text: answer }
|
||||
yield { type: 'block-end', index: 0, block: { type: 'text', text: answer } }
|
||||
yield answer === 'parent summary after child' && process.env.DSH_TEST_SUMMARY_ERROR === '1'
|
||||
? { type: 'finish', reason: { kind: 'error', failure: { code: 'SUMMARY_ERROR', message: 'summary failed' } } }
|
||||
: { type: 'finish', reason: { kind: 'stop' } }
|
||||
}
|
||||
}
|
||||
|
||||
export const name = 'nested-mock-llm'
|
||||
export const inject = ['llm', 'agents']
|
||||
|
||||
/** Register the gated adapter and observe the root's first idle edge. */
|
||||
export function apply(ctx: Context): void {
|
||||
ctx.llm.registerAdapter(['mock'], new NestedAdapter())
|
||||
ctx.on('agent/status', ({ agent, status }) => {
|
||||
if (status === 'idle' && agent.session.header.parentSession === undefined) {
|
||||
writeFileSync(process.env.DSH_TEST_ROOT_IDLE!, 'idle\n')
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
/** A full child profile must finish its nested fork before its external task settles. */
|
||||
import { Context } from '@deepseek-ai/cordis'
|
||||
import { existsSync } from 'node:fs'
|
||||
import { mkdtemp, rm, writeFile } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { resolveExampleLaunch } from '@deepseek-ai/dsh-loader-smoke'
|
||||
import { createProcessDeepSeekHarness } from '../../../sdk/client/src/api.ts'
|
||||
import { spawnSubprocess } from '../../../subprocess/subprocess-local/src/spawn.ts'
|
||||
import { startAcpRun } from '../../subagent-acp/src/run.ts'
|
||||
import { externalTestParent } from '../../subagent/tests/external-activation-helpers.ts'
|
||||
import { internals, startSdkRun } from '../src/run.ts'
|
||||
|
||||
const sourceBin = fileURLToPath(new URL('../../../../apps/cli/src/bin.ts', import.meta.url))
|
||||
const tsconfig = fileURLToPath(new URL('../../../../tsconfig.json', import.meta.url))
|
||||
const fixture = fileURLToPath(new URL('./fixtures/loader/nested-mock-llm.ts', import.meta.url))
|
||||
|
||||
describe('external tasks with a nested fork', () => {
|
||||
it.each([
|
||||
['sdk', 'complete'], ['acp', 'complete'], ['sdk', 'cancel'], ['acp', 'cancel'],
|
||||
['sdk', 'error'], ['acp', 'error'],
|
||||
] as const)('%s owns the nested task through %s', { timeout: 60_000, retry: 0 }, async (profile, outcome) => {
|
||||
const root = await mkdtemp(join(tmpdir(), `dsh-${profile}-nested-`))
|
||||
const ctx = new Context()
|
||||
const originalFactory = internals.createHarness.bind(internals)
|
||||
const release = join(root, 'release')
|
||||
const ready = join(root, 'ready')
|
||||
const idle = join(root, 'idle')
|
||||
const patch = join(root, 'nested.patch.yml')
|
||||
const env = {
|
||||
DSH_HOME: join(root, 'home'), DSH_TELEMETRY_DISABLED: '1',
|
||||
DSH_TEST_CHILD_READY: ready, DSH_TEST_CHILD_RELEASE: release, DSH_TEST_ROOT_IDLE: idle,
|
||||
DSH_TEST_SUMMARY_ERROR: outcome === 'error' ? '1' : '0',
|
||||
}
|
||||
const launch = resolveExampleLaunch({
|
||||
srcBin: sourceBin, tsconfigPath: tsconfig, sourceImport: 'tsx/esm',
|
||||
configArgs: ['--profile', profile, '--patch', patch], env,
|
||||
})
|
||||
let run: Awaited<ReturnType<typeof startSdkRun>> | undefined
|
||||
try {
|
||||
await writeFile(patch, JSON.stringify([
|
||||
{ id: 'llm-deepseek', disabled: true },
|
||||
{ id: 'agent-instructions', disabled: true },
|
||||
{ id: 'skill-filesystem', config: { includeDefaultRoots: false } },
|
||||
{ id: 'typert-loader', disabled: true },
|
||||
...profile === 'acp' ? [{ id: 'acp', config: { provider: 'mock', model: 'mock' } }] : [],
|
||||
{ insert: [{ id: 'nested-mock-llm', name: fixture }] },
|
||||
]))
|
||||
const parent = await externalTestParent(ctx, root)
|
||||
const controller = new AbortController()
|
||||
const signal = AbortSignal.any([controller.signal, AbortSignal.timeout(45_000)])
|
||||
const request = { parent, prompt: [{ type: 'text' as const, text: 'delegate with a fork' }], signal }
|
||||
if (profile === 'sdk') {
|
||||
internals.createHarness = options => createProcessDeepSeekHarness({
|
||||
command: launch.command, args: launch.args, cwd: root,
|
||||
environment: () => ({ ...process.env, ...launch.env }),
|
||||
description: 'nested SDK profile', initializeTimeoutMs: 30_000,
|
||||
}, options)
|
||||
run = await startSdkRun(request, {
|
||||
profile, patches: [patch], dshHome: env.DSH_HOME, cwd: root, provider: 'mock', model: 'mock', env,
|
||||
shutdownTimeoutMs: 1_000, disposeEofGraceMs: 6_000, disposeGraceMs: 3_000,
|
||||
})
|
||||
} else {
|
||||
run = await startAcpRun(request, {
|
||||
command: launch.command, args: launch.args, cwd: root, permission: 'reject',
|
||||
env: Object.fromEntries(Object.entries(launch.env).filter((entry): entry is [string, string] => entry[1] !== undefined)),
|
||||
disposeEofGraceMs: 6_000, disposeGraceMs: 3_000, spawn: spec => spawnSubprocess(spec),
|
||||
})
|
||||
}
|
||||
let settled = false
|
||||
void run.result.then(() => { settled = true })
|
||||
await vi.waitFor(() => { expect(existsSync(ready) && existsSync(idle)).toBe(true) }, { timeout: 30_000 })
|
||||
expect(settled).toBe(false)
|
||||
if (outcome === 'cancel') {
|
||||
controller.abort()
|
||||
expect((await run.result).stopReason).toBe('aborted')
|
||||
return
|
||||
}
|
||||
await writeFile(release, 'release\n')
|
||||
const result = await run.result
|
||||
if (outcome === 'error') {
|
||||
expect(result.stopReason).toBe('error')
|
||||
return
|
||||
}
|
||||
expect(result.stopReason).toBe('completed')
|
||||
const text = result.output.filter(block => block.type === 'text').map(block => block.text).join('')
|
||||
expect(text).toContain('parent summary after child')
|
||||
} finally {
|
||||
await writeFile(release, 'release\n')
|
||||
await run?.dispose()
|
||||
internals.createHarness = originalFactory
|
||||
await ctx.fiber.dispose()
|
||||
await rm(root, { recursive: true, force: true })
|
||||
}
|
||||
})
|
||||
})
|
||||
@@ -19,7 +19,6 @@ import type { Agent, AgentOptions } from '@deepseek-ai/dsh-agent'
|
||||
import {
|
||||
DeepSeekHarness,
|
||||
HarnessClient,
|
||||
HarnessSession,
|
||||
SdkProtocolError,
|
||||
} from '@deepseek-ai/dsh-sdk-client'
|
||||
import { createProcessDeepSeekHarness } from '../../../sdk/client/src/api.ts'
|
||||
@@ -609,7 +608,7 @@ describe('dsh-subagent-dsh-sdk provider', () => {
|
||||
|
||||
it('uses a fixed unknown category for an untyped SDK exception', async () => {
|
||||
const rawMessage = 'unknown SDK failure at /private/path SECRET_TOKEN'
|
||||
const spy = vi.spyOn(HarnessSession.prototype, 'run')
|
||||
const spy = vi.spyOn(HarnessClient.prototype, 'prompt')
|
||||
.mockRejectedValue(new Error(rawMessage))
|
||||
try {
|
||||
const ctx = await setup()
|
||||
|
||||
Reference in new issue
Block a user