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.
This commit is contained in:
Tianyi Cui committed
1 parent
3e2e7e0b2d
commit
8fd7ff69e3
12 files changed
+193
-16
No files matched your search
@@ -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) |
|
||||
|
||||
@@ -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) |
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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 的状态。
|
||||
|
||||
### 载荷语义
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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` 记录结果；否则等待请求会失败，即使错误发生在等待请求之前。终止事件已经记录的错误仍通过会话事件表达，后续成功的活动会清除较早的运行时错误。
|
||||
|
||||
### 关闭与退出
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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))
|
||||
})
|
||||
})
|
||||
@@ -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<boolean>()
|
||||
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<typeof server.wait> | undefined
|
||||
if (timing === 'during') {
|
||||
const waiting = Promise.withResolvers<undefined>()
|
||||
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()
|
||||
}
|
||||
})
|
||||
})
|
||||
@@ -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([
|
||||
|
||||
Reference in new issue
Block a user