From 8fd7ff69e382bd8ac4866b7eae24efb1262214df Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Tue, 6 Oct 2026 01:01:03 +0800 Subject: [PATCH] fix(sdk): reject unrepresented session activity failures A summary can fail while appending turn/start or turn/end and still leave the root idle with a stable Session sequence. The SDK wait operation then returned {}, letting the external provider treat the preceding completed terminal as the result of the entire task. A listener installed only for the wait would also miss failures between prompt admission and the wait request. Retain one latest live failure on each existing SDK Session record, scoped to the exact owned Agent. Record the Session's next log offset when the failure occurs. A turn/end from the exact owned Session clears it only when that terminal was committed at or after that offset. This preserves failures raised during terminal publication, keeps ordinary represented turn errors in the durable outcome, and permits later successful activity to recover. At stable root/descendant completion, session/wait rejects any remaining failure. Raw session.status and HarnessSession.run behavior are unchanged; no generic failure history or new notification is introduced. Add seven owner-local cases for later start/end failures before and during wait, subsequent successful prompts, represented errors, and publication ordering including an undefined thrown value. Six failure cases returned success before the fix. Strengthen listener teardown coverage to assert all actual subscriptions are disposed after a disposer fails. Extend the existing recorded subagent-continuable SDK scenario to exercise session/wait. It still asserts that run() returns DONE at parent turn 1, then requires child completion and parent turn 2/idle plus its final summary to be delivered before the wait response. Bypassing the wait made that ordering assertion fail. No recorded Session or golden output was invented or refreshed. Update the protocol/server bilingual READMEs and regenerate the event-consumer matrix for the new error listener. Validation: - pnpm exec vitest run packages/sdk/server/tests/wait.spec.ts before the fix: 6 failed, 1 passed (the represented-error case) - pnpm exec vitest run packages/sdk/server/tests/server.spec.ts packages/sdk/server/tests/wait.spec.ts (40 passed) - pnpm run test:e2e packages/subagent/subagent-dsh-sdk/tests/nested-delegation.e2e.ts packages/subagent/subagent-dsh-sdk/tests/loader-composition.e2e.ts (8 passed) - DSH_EXAMPLE_MODE=lib pnpm run test:e2e packages/subagent/subagent-dsh-sdk/tests/nested-delegation.e2e.ts packages/subagent/subagent-dsh-sdk/tests/loader-composition.e2e.ts (8 passed) - pnpm run test:snapshot snapshots/sdk/sdk.snapshot.ts -t 'replays subagent-continuable through dsh --profile sdk' (1 passed, 24 unselected) - pnpm run doc-sync: Host compile/bundle, doc-typecheck and 40 other gates passed; the generated event graph was stale, then regenerated and passed pnpm run verify-doc-graphs (all 7 graph docs current) - pnpm run lint:contracts-ready and focused lint of all changed SDK code - named translation pairing checks and git diff --check Independent SDK rereview and 7/7 focused tests found no remaining issue. The parent task owns unchanged Python advanced/recovery validation on the combined branch; this commit changes no Python or high-level SDK results. --- docs/event-producer-consumer.md | 2 +- docs/event-producer-consumer.zh.md | 2 +- packages/sdk/protocol/README.i18n.yaml | 4 +- packages/sdk/protocol/README.md | 2 +- packages/sdk/protocol/README.zh.md | 2 +- packages/sdk/server/README.i18n.yaml | 4 +- packages/sdk/server/README.md | 2 +- packages/sdk/server/README.zh.md | 2 +- packages/sdk/server/src/server.ts | 22 +++- packages/sdk/server/tests/server.spec.ts | 10 +- packages/sdk/server/tests/wait.spec.ts | 124 +++++++++++++++++++++++ snapshots/sdk/sdk.snapshot.ts | 33 ++++++ 12 files changed, 193 insertions(+), 16 deletions(-) create mode 100644 packages/sdk/server/tests/wait.spec.ts diff --git a/docs/event-producer-consumer.md b/docs/event-producer-consumer.md index a611ee1e7d..fac6fbf8a0 100644 --- a/docs/event-producer-consumer.md +++ b/docs/event-producer-consumer.md @@ -13,7 +13,7 @@ This matrix shows which packages dispatch each harness-owned event and which pac | `agent/assistant-stream` | `emit` | [`packages/core/agent/src/runtime-types.ts:363`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`headless`](../packages/bundle/headless), `session-controller` | | `agent/created` | `serial` | [`packages/core/agent/src/runtime-types.ts:261`](../packages/core/agent/src/runtime-types.ts) | [`agent`](../packages/core/agent) (`serial`) | `agent-team`, `browser-use-runtime`, `claude-code-mods`, [`file-reference-local`](../packages/context/file-reference-local), [`goal`](../packages/goal/goal), [`goal-round-driver`](../packages/goal/goal-round-driver), [`hooks-claude-code`](../packages/hooks/hooks-claude-code), [`hooks-codex`](../packages/hooks/hooks-codex), [`loader-smoke`](../packages/test-support/loader-smoke), `session-controller`, `tool-agent-team`, [`tool-subagent`](../packages/subagent/tool-subagent) | | `agent/disposed` | `emit` | [`packages/core/agent/src/runtime-types.ts:270`](../packages/core/agent/src/runtime-types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`agent-loop`](../packages/core/agent-loop), `claude-code-mods`, [`file-reference-local`](../packages/context/file-reference-local), [`goal-round-driver`](../packages/goal/goal-round-driver), `session-controller`, [`subagent`](../packages/subagent/subagent), `tool-agent-team`, [`tool-subagent`](../packages/subagent/tool-subagent) | -| `agent/error` | `emit` | [`packages/core/agent/src/runtime-types.ts:393`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`acp`](../packages/acp/acp), [`deepseek-account`](../packages/credentials/deepseek-account), [`goal-round-driver`](../packages/goal/goal-round-driver), `session-controller`, [`session-telemetry`](../packages/session/session-telemetry) | +| `agent/error` | `emit` | [`packages/core/agent/src/runtime-types.ts:393`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`acp`](../packages/acp/acp), [`deepseek-account`](../packages/credentials/deepseek-account), [`goal-round-driver`](../packages/goal/goal-round-driver), `server`, `session-controller`, [`session-telemetry`](../packages/session/session-telemetry) | | `agent/inbox/claimed` | `emit` | [`packages/core/agent/src/runtime-types.ts:299`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`acp`](../packages/acp/acp), [`goal-round-driver`](../packages/goal/goal-round-driver), [`subagent`](../packages/subagent/subagent), [`tool-jobs`](../packages/jobs/tool-jobs), [`user-questions`](../packages/interaction/user-questions) | | `agent/inbox/discarded` | `emit` | [`packages/core/agent/src/runtime-types.ts:307`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`goal-round-driver`](../packages/goal/goal-round-driver), [`subagent`](../packages/subagent/subagent), [`user-questions`](../packages/interaction/user-questions) | | `agent/inbox/inserted` | `emit` | [`packages/core/agent/src/runtime-types.ts:288`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`goal-round-driver`](../packages/goal/goal-round-driver) | diff --git a/docs/event-producer-consumer.zh.md b/docs/event-producer-consumer.zh.md index 576fb736d6..6098bdc251 100644 --- a/docs/event-producer-consumer.zh.md +++ b/docs/event-producer-consumer.zh.md @@ -15,7 +15,7 @@ | `agent/assistant-stream` | `emit` | [`packages/core/agent/src/runtime-types.ts:363`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`headless`](../packages/bundle/headless), `session-controller` | | `agent/created` | `serial` | [`packages/core/agent/src/runtime-types.ts:261`](../packages/core/agent/src/runtime-types.ts) | [`agent`](../packages/core/agent) (`serial`) | `agent-team`, `browser-use-runtime`, `claude-code-mods`, [`file-reference-local`](../packages/context/file-reference-local), [`goal`](../packages/goal/goal), [`goal-round-driver`](../packages/goal/goal-round-driver), [`hooks-claude-code`](../packages/hooks/hooks-claude-code), [`hooks-codex`](../packages/hooks/hooks-codex), [`loader-smoke`](../packages/test-support/loader-smoke), `session-controller`, `tool-agent-team`, [`tool-subagent`](../packages/subagent/tool-subagent) | | `agent/disposed` | `emit` | [`packages/core/agent/src/runtime-types.ts:270`](../packages/core/agent/src/runtime-types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`agent-loop`](../packages/core/agent-loop), `claude-code-mods`, [`file-reference-local`](../packages/context/file-reference-local), [`goal-round-driver`](../packages/goal/goal-round-driver), `session-controller`, [`subagent`](../packages/subagent/subagent), `tool-agent-team`, [`tool-subagent`](../packages/subagent/tool-subagent) | -| `agent/error` | `emit` | [`packages/core/agent/src/runtime-types.ts:393`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`acp`](../packages/acp/acp), [`deepseek-account`](../packages/credentials/deepseek-account), [`goal-round-driver`](../packages/goal/goal-round-driver), `session-controller`, [`session-telemetry`](../packages/session/session-telemetry) | +| `agent/error` | `emit` | [`packages/core/agent/src/runtime-types.ts:393`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`acp`](../packages/acp/acp), [`deepseek-account`](../packages/credentials/deepseek-account), [`goal-round-driver`](../packages/goal/goal-round-driver), `server`, `session-controller`, [`session-telemetry`](../packages/session/session-telemetry) | | `agent/inbox/claimed` | `emit` | [`packages/core/agent/src/runtime-types.ts:299`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`acp`](../packages/acp/acp), [`goal-round-driver`](../packages/goal/goal-round-driver), [`subagent`](../packages/subagent/subagent), [`tool-jobs`](../packages/jobs/tool-jobs), [`user-questions`](../packages/interaction/user-questions) | | `agent/inbox/discarded` | `emit` | [`packages/core/agent/src/runtime-types.ts:307`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`goal-round-driver`](../packages/goal/goal-round-driver), [`subagent`](../packages/subagent/subagent), [`user-questions`](../packages/interaction/user-questions) | | `agent/inbox/inserted` | `emit` | [`packages/core/agent/src/runtime-types.ts:288`](../packages/core/agent/src/runtime-types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | [`goal-round-driver`](../packages/goal/goal-round-driver) | diff --git a/packages/sdk/protocol/README.i18n.yaml b/packages/sdk/protocol/README.i18n.yaml index c67ed42b46..a1a77cccf8 100644 --- a/packages/sdk/protocol/README.i18n.yaml +++ b/packages/sdk/protocol/README.i18n.yaml @@ -21,8 +21,8 @@ en: be53c71505c0d1b4 zh: 6210dee577715539 /deepseek-ai-dsh-sdk-protocol/use-this-package/the-sdk-methods: - en: 7a5a44f73174e9d9 - zh: 8b3e65f769e59a1e + en: 6986df4ffbae4d52 + zh: d9a75ac5459dc5af /deepseek-ai-dsh-sdk-protocol/use-this-package/payload-semantics: en: 3cf5d4ffbaf37f69 zh: 23c4450e31c91d67 diff --git a/packages/sdk/protocol/README.md b/packages/sdk/protocol/README.md index bd7f8109fe..ed91bf5610 100644 --- a/packages/sdk/protocol/README.md +++ b/packages/sdk/protocol/README.md @@ -48,7 +48,7 @@ Both wire ends share one method set: four client-to-server requests and four ser `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. +`session/wait` accepts `{ sessionId }` for an existing SDK-owned session and returns `{}` after the root stays idle with no managed descendants. Unknown ids and live Agent failures without a newer committed terminal 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 diff --git a/packages/sdk/protocol/README.zh.md b/packages/sdk/protocol/README.zh.md index bdedc9ddd9..b1983b6987 100644 --- a/packages/sdk/protocol/README.zh.md +++ b/packages/sdk/protocol/README.zh.md @@ -48,7 +48,7 @@ kind: "package-library" `HarnessSdkRequestMap` 与 `HarnessSdkNotificationMap` 按方法名索引这些结构；包根与传输一起导出它们。 -`session/wait` 接受已有 SDK 会话的 `{ sessionId }`，在根 Agent 持续空闲且没有受管理后代时返回 `{}`。未知 id 会被拒绝。等待期间提交的通知在同一传输上先于响应到达；原始 `session.status` 转换仍表示整个 Agent 的状态。 +`session/wait` 接受已有 SDK 会话的 `{ sessionId }`，在根 Agent 持续空闲且没有受管理后代时返回 `{}`。未知 id，以及没有后续终止事件记录的 Agent 运行时错误，都会使请求失败。等待期间提交的通知在同一传输上先于响应到达；原始 `session.status` 转换仍表示整个 Agent 的状态。 ### 载荷语义 diff --git a/packages/sdk/server/README.i18n.yaml b/packages/sdk/server/README.i18n.yaml index fd35520dc5..30c8f63688 100644 --- a/packages/sdk/server/README.i18n.yaml +++ b/packages/sdk/server/README.i18n.yaml @@ -27,8 +27,8 @@ en: 784b68b5a498f2d5 zh: 2c3c5a1490845179 /deepseek-ai-dsh-sdk-jsonrpc-server/use-this-package/what-sdk-clients-can-do: - en: 39ed6a280988103b - zh: c69e2900d9641d76 + en: b6531381c4277e1d + zh: 4df79f9f8c74d555 /deepseek-ai-dsh-sdk-jsonrpc-server/use-this-package/shutdown-and-exit: en: 08c3858f53090f39 zh: dfae960e11515b06 diff --git a/packages/sdk/server/README.md b/packages/sdk/server/README.md index fec729e6d6..7ee5016906 100644 --- a/packages/sdk/server/README.md +++ b/packages/sdk/server/README.md @@ -47,7 +47,7 @@ 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. +`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. A live Agent failure remains associated with that exact session until a later committed `turn/end` accounts for the outcome; otherwise the wait rejects, including when the failure preceded the wait request. Errors already represented by a terminal remain in Session events, and later successful activity clears an earlier live failure. ### Shutdown and exit diff --git a/packages/sdk/server/README.zh.md b/packages/sdk/server/README.zh.md index e02bd02328..78ed20a0c3 100644 --- a/packages/sdk/server/README.zh.md +++ b/packages/sdk/server/README.zh.md @@ -47,7 +47,7 @@ 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 传输上先于响应到达。 +`session/wait` 观察已有 SDK 会话，直到受管理后代及其触发的根 Agent 后续轮次完成。它拒绝未知或已释放的根 Agent，不会创建会话，并在后代检查期间根 Agent 持续空闲后响应。完成前发出的会话通知在 stdio 传输上先于响应到达。Agent 的运行时错误会保留在对应的确切会话中，直到后续提交的 `turn/end` 记录结果；否则等待请求会失败，即使错误发生在等待请求之前。终止事件已经记录的错误仍通过会话事件表达，后续成功的活动会清除较早的运行时错误。 ### 关闭与退出 diff --git a/packages/sdk/server/src/server.ts b/packages/sdk/server/src/server.ts index 49cfa0f6eb..2e0066fe5b 100644 --- a/packages/sdk/server/src/server.ts +++ b/packages/sdk/server/src/server.ts @@ -12,7 +12,7 @@ import type { Agent, AgentHandle } from '@deepseek-ai/dsh-agent' import { admitEncodedImages, type EncodedImageAttachment, type ImageAttachmentRef } from '@deepseek-ai/dsh-attachment' import { createUserMessage, ReasoningEffortId, type ContentBlock, type LlmRuntime } from '@deepseek-ai/dsh-llm' import { carrierKeyOf, type Scoped } from '@deepseek-ai/dsh-scope' -import type { SessionId } from '@deepseek-ai/dsh-session' +import type { SessionId, SessionLogOffset } from '@deepseek-ai/dsh-session' import type SubagentRuntime from '@deepseek-ai/dsh-subagent' import type { SubagentRunEndInfo } from '@deepseek-ai/dsh-subagent' import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek-api-key' @@ -31,6 +31,8 @@ import type { interface SessionRecord { handle: AgentHandle + /** Latest live failure not superseded by a subsequently committed terminal. */ + failure: { readonly error: unknown; readonly atOffset: SessionLogOffset } | undefined } function encodedImage(block: SessionPromptParams['contentBlocks'][number]): block is SdkEncodedImageBlock { @@ -94,9 +96,19 @@ export class HarnessSdkJsonRpcServer { ) { const serverOptions = this.options this.disposers.push(ctx.on('session/event', (session, event) => { + const record = this.sessions.get(session.id) + if (event.type === 'turn/end' && record?.handle.agent.session === session + && record.failure !== undefined && event.seq >= record.failure.atOffset) { + record.failure = undefined + } const payload: SessionEventNotification = { sessionId: String(session.id), event } this.transport.notify('session.event', payload) })) + this.disposers.push(ctx.on('agent/error', ({ agent, error }) => { + const record = this.sessions.get(agent.session.id) + if (record?.handle.agent !== agent) return + record.failure = { error, atOffset: agent.session.seq } + })) this.disposers.push(ctx.on('agent/status', ({ agent, status }) => { this.transport.notify('session.status', { sessionId: String(agent.session.id), status }) })) @@ -198,6 +210,7 @@ export class HarnessSdkJsonRpcServer { /** * Wait for an existing owned Agent and its managed descendants to finish. * The response follows all Session notifications emitted during the wait. + * Rejects a live Agent failure that has no subsequent durable terminal. * @param params - the SDK-owned session to observe without creating one. * @returns an empty result after the root stays idle across the descendant check. */ @@ -213,7 +226,10 @@ export class HarnessSdkJsonRpcServer { 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 {} + if (!children && agent.status === 'idle' && agent.session.seq === idleSeq) { + if (rec.failure !== undefined) throw rec.failure.error + return {} + } } } @@ -316,7 +332,7 @@ export class HarnessSdkJsonRpcServer { ...this.maxTokens === undefined ? {} : { maxTokens: this.maxTokens }, }, }) - const rec: SessionRecord = { handle } + const rec: SessionRecord = { handle, failure: undefined } this.sessions.set(sessionId, rec) return rec } diff --git a/packages/sdk/server/tests/server.spec.ts b/packages/sdk/server/tests/server.spec.ts index a5e3656e28..2c420a2f95 100644 --- a/packages/sdk/server/tests/server.spec.ts +++ b/packages/sdk/server/tests/server.spec.ts @@ -1246,10 +1246,14 @@ describe('HarnessSdkJsonRpcServer', () => { it('continues teardown after a subscription disposer fails', async () => { let subscription = 0 + const disposed: number[] = [] const listenerFailure = new Error('listener teardown failed') const on = vi.fn(() => { - subscription += 1 - return subscription === 1 ? () => { throw listenerFailure } : () => undefined + const id = ++subscription + return () => { + disposed.push(id) + if (id === subscription) throw listenerFailure + } }) const ctx = { on, @@ -1259,6 +1263,6 @@ describe('HarnessSdkJsonRpcServer', () => { const server = new HarnessSdkJsonRpcServer(ctx, new FakeTransport()) await expect(server.shutdown()).rejects.toBe(listenerFailure) - expect(on).toHaveBeenCalledTimes(4) + expect(disposed.toSorted()).toEqual(Array.from({ length: subscription }, (_, index) => index + 1)) }) }) diff --git a/packages/sdk/server/tests/wait.spec.ts b/packages/sdk/server/tests/wait.spec.ts new file mode 100644 index 0000000000..3f2385df8a --- /dev/null +++ b/packages/sdk/server/tests/wait.spec.ts @@ -0,0 +1,124 @@ +/** SDK task completion retains live failures that have no durable terminal outcome. */ +import { Context } from '@deepseek-ai/cordis' +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { describe, expect, it, vi } from 'vitest' +import AgentLoop from '@deepseek-ai/dsh-agent-loop' +import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit' +import { createUserMessage } from '@deepseek-ai/dsh-llm' +import { SessionId } from '@deepseek-ai/dsh-session' +import JsonlSessionPersistence from '@deepseek-ai/dsh-session-persistence-jsonl' +import SubagentRuntime from '@deepseek-ai/dsh-subagent' +import { MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts' +import { HarnessSdkJsonRpcServer } from '../src/server.ts' + +async function setup(beforeServer?: (ctx: Context) => void) { + const root = await mkdtemp(join(tmpdir(), 'dsh-sdk-wait-')) + const ctx = new Context() + await mountAgentLoopTestDependencies(ctx) + await ctx.plugin(AgentLoop, { agents: [] }) + await ctx.plugin(SubagentRuntime) + await ctx.plugin(JsonlSessionPersistence, { root }) + ctx.llm.registerAdapter(['mock'], new MockAdapter(Array.from({ length: 5 }, () => textResponse('answer')))) + beforeServer?.(ctx) + const server = new HarnessSdkJsonRpcServer(ctx, { + request: () => Promise.reject(new Error('the server must not request the client')), + notify: () => {}, + }) + await server.initialize({ cwd: root, provider: 'mock', model: 'mock' }) + return { + ctx, server, + dispose: async () => { + await server.shutdown() + await ctx.fiber.dispose() + await rm(root, { recursive: true, force: true }) + }, + } +} + +describe('SDK session/wait failures', () => { + it.each([ + ['turn/start', 'before'], ['turn/start', 'during'], + ['turn/end', 'before'], ['turn/end', 'during'], + ] as const)('rejects a later %s failure %s waiting and allows successful recovery', async (failedEvent, timing) => { + const { ctx, server, dispose } = await setup() + const children = Promise.withResolvers() + try { + await server.prompt({ sessionId: 'main', contentBlocks: [{ type: 'text', text: 'first' }] }) + const parent = ctx.agents.get(SessionId('main'))! + await parent.whenIdle() + const failure = new Error(`summary ${failedEvent} unavailable`) + let completion: ReturnType | undefined + if (timing === 'during') { + const waiting = Promise.withResolvers() + vi.spyOn(ctx.subagents, 'waitForChildren').mockImplementationOnce(() => { + waiting.resolve(undefined) + return children.promise + }) + completion = server.wait({ sessionId: 'main' }) + void completion.catch(() => undefined) + await waiting.promise + } + const append = parent.session.append.bind(parent.session) + const appendSpy = vi.spyOn(parent.session, 'append').mockImplementation(((type: string, ...rest: never[]) => { + if (type === failedEvent) throw failure + return (append as (...args: never[]) => unknown)(type as never, ...rest) + }) as never) + try { + parent.followup(createUserMessage({ content: [{ type: 'text', text: 'child result' }], source: { kind: 'user' } })) + children.resolve(true) + if (timing === 'before') { + await parent.whenIdle() + completion = server.wait({ sessionId: 'main' }) + } + await expect(completion).rejects.toBe(failure) + } finally { + appendSpy.mockRestore() + } + await server.prompt({ sessionId: 'main', contentBlocks: [{ type: 'text', text: 'recover' }] }) + await expect(server.wait({ sessionId: 'main' })).resolves.toEqual({}) + expect(parent.session.snapshotEvents().findLast(event => event.type === 'turn/end')?.data.reason.kind) + .toBe('completed') + } finally { + children.resolve(true) + await dispose() + } + }) + + it('leaves a represented turn error in the durable outcome', async () => { + const { ctx, server, dispose } = await setup() + try { + ctx.on('agent/pre-step', () => { throw new Error('represented failure') }) + await server.prompt({ sessionId: 'main', contentBlocks: [{ type: 'text', text: 'fail normally' }] }) + await expect(server.wait({ sessionId: 'main' })).resolves.toEqual({}) + const parent = ctx.agents.get(SessionId('main'))! + expect(parent.session.snapshotEvents().findLast(event => event.type === 'turn/end')?.data.reason) + .toMatchObject({ kind: 'error', error: { message: 'represented failure' } }) + } finally { + await dispose() + } + }) + + it.each([new Error('terminal publication failed'), undefined])('retains an error emitted during terminal publication (%s)', async (failure) => { + let enabled = true + const { ctx, server, dispose } = await setup((ctx) => { + ctx.on('session/event', (session, event) => { + if (!enabled || event.type !== 'turn/end') return + const agent = ctx.agents.get(session.id) + if (agent === undefined) throw new Error('missing fixture Agent') + ctx.emit('agent/error', { agent, turn: event.data.turn, step: 0, error: failure }) + }) + }) + try { + await server.prompt({ sessionId: 'main', contentBlocks: [{ type: 'text', text: 'first' }] }) + await ctx.agents.get(SessionId('main'))!.whenIdle() + await expect(server.wait({ sessionId: 'main' })).rejects.toBe(failure) + enabled = false + await server.prompt({ sessionId: 'main', contentBlocks: [{ type: 'text', text: 'recover' }] }) + await expect(server.wait({ sessionId: 'main' })).resolves.toEqual({}) + } finally { + await dispose() + } + }) +}) diff --git a/snapshots/sdk/sdk.snapshot.ts b/snapshots/sdk/sdk.snapshot.ts index 67d89d5f34..7789348c15 100644 --- a/snapshots/sdk/sdk.snapshot.ts +++ b/snapshots/sdk/sdk.snapshot.ts @@ -634,6 +634,9 @@ async function runScenario(scenario: CorpusScenario): Promise<{ const liveSessions: (string | undefined)[] = [sessionId] await harness.start() const subscription = harness.client.subscribeSessionTree(sessionId) + const completionSubscription = scenario.name === 'subagent-continuable' + ? harness.client.subscribeSessionTree(sessionId) + : undefined const observe = (notification: HarnessNotification): void => { observedMethods.add(notification.method) if (notification.method !== 'subagent.started') return @@ -660,6 +663,35 @@ async function runScenario(scenario: CorpusScenario): Promise<{ }, }) results.push(result) + if (completionSubscription !== undefined) { + expect(result.finalResponse, 'run returns at the first parent idle').toBe('DONE') + expect(result.events.filter(event => event.type === 'turn/end')) + .toMatchObject([{ data: { turn: 1 } }]) + expect(await harness.client.request('session/wait', { sessionId })).toEqual({}) + const delivered: HarnessNotification[] = [] + let notification: HarnessNotification | undefined + while ((notification = completionSubscription.tryNext()) !== undefined) delivered.push(notification) + const completionOrder = delivered.flatMap((entry) => { + if (entry.method === 'subagent.finished') return ['child:finished'] + if (entry.params.sessionId !== sessionId) return [] + if (entry.method === 'session.status' && entry.params.status === 'idle') return ['parent:idle'] + const event = notificationEvent(entry) + if (event?.type !== 'turn/end') return [] + const data = event.data as JsonObject + expect(data.reason).toEqual({ kind: 'completed' }) + return [`parent:turn:${String(data.turn)}`] + }) + expect(completionOrder, 'completion notifications arrive before the session/wait response').toEqual([ + 'parent:turn:1', 'parent:idle', 'child:finished', 'parent:turn:2', 'parent:idle', + ]) + const parentMessages = delivered.flatMap((entry) => { + const event = entry.params.sessionId === sessionId ? notificationEvent(entry) : undefined + return event?.type === 'assistant/message' ? [event] : [] + }) + expect(parentMessages.at(-1)).toMatchObject({ + data: { turn: 2, message: { content: [{ type: 'text', text: 'SUBAGENT_SETTLED_NOTED' }] } }, + }) + } if (scenario.manifest.environment?.DSH_SNAPSHOT_FEEDBACK === '1') { const feedback = result.events.filter(event => event.type.startsWith('feedback/')) expect(feedback.map(event => event.type)).toEqual([ @@ -678,6 +710,7 @@ async function runScenario(scenario: CorpusScenario): Promise<{ } } finally { subscription.close() + completionSubscription?.close() } await harness.close() const logs = (await Promise.all([