From cf2b9e211d82a78f4ef1f545d01311d860214b76 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Mon, 27 Jul 2026 17:48:07 +0800 Subject: [PATCH] fix(sdk-client): address ds-review-bot findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - api: resolve a relative workspace cwd to absolute before the handshake — the child spawns relative to the parent cwd, but the wire cwd is resolved again inside the child, so a relative value double-resolved (worker -> worker/worker). - api: make the documented handshake retry real — HarnessClient.close() is permanent, so a failed initialize now reaps the runtime and swaps in a fresh client; DeepSeekHarness.close() is terminal and stops the respawns. - api: validate session.event envelopes, assistant/message content, and session.finished reasons at the wire boundary — a malformed runtime surfaces as SdkProtocolError instead of type-invalid TurnResult data or a TypeError out of finalResponse. - client: a throwing subscribe() filter fails and detaches only its own subscription (normalized to Error); sibling fan-out and the transport read loop are undisturbed. - client: NotificationSubscription.close() drops its queued notifications, matching its documented contract; runtime-death fail() still leaves already-delivered items drainable. - client: subscribe() after close()/runtime death returns a born-failed subscription so next() rejects instead of parking forever. - client/transport: bounded requests abandon via AbortSignal — the transport drops the pending entry at timeout, so repeated bounded calls against a hung method retain no per-call state. One test per finding; per-file coverage stays 100% on both packages. --- packages/sdk/sdk-client/README.i18n.yaml | 6 +- packages/sdk/sdk-client/README.md | 2 +- packages/sdk/sdk-client/README.zh.md | 2 +- packages/sdk/sdk-client/src/api.ts | 90 ++++++++++-- packages/sdk/sdk-client/src/client.ts | 54 ++++--- packages/sdk/sdk-client/tests/fake-runtime.ts | 35 ++++- .../sdk/sdk-client/tests/sdk-client.spec.ts | 138 +++++++++++++++++- packages/sdk/sdk-protocol/src/transport.ts | 41 +++++- .../sdk/sdk-protocol/tests/transport.spec.ts | 23 +++ 9 files changed, 347 insertions(+), 44 deletions(-) diff --git a/packages/sdk/sdk-client/README.i18n.yaml b/packages/sdk/sdk-client/README.i18n.yaml index ebb5ca18d5..11cdef9d94 100644 --- a/packages/sdk/sdk-client/README.i18n.yaml +++ b/packages/sdk/sdk-client/README.i18n.yaml @@ -1,6 +1,6 @@ # Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each # side as of the last confirmed-consistent state. Both languages carry equal authority; # after editing either side, bring the other along and re-record with: -# pnpm run verify-translation-pairing --write -README.md: 82e344014ac01120986ee2b4e07ac21192506ff7 -README.zh.md: 6e67f57ab92a75be3600cbdd0ce6bf21832fd2c8 +# pnpm run verify-translation-pairing --write packages/sdk/sdk-client/README.md +README.md: 3945f911990fa3df362581b0fb37389110bdd386 +README.zh.md: 3814b88aab1b10c96fdf809565994f29b8e8026b diff --git a/packages/sdk/sdk-client/README.md b/packages/sdk/sdk-client/README.md index 82e344014a..3945f91199 100644 --- a/packages/sdk/sdk-client/README.md +++ b/packages/sdk/sdk-client/README.md @@ -20,7 +20,7 @@ const result = await harness.run('say hi') console.log(result.status, result.finalResponse) ``` -The subprocess starts lazily on first use and stays owned by the instance across `run()` calls; `close()` (or `await using`) is required so the child is always reaped. `start()` memoizes the `initialize` handshake (cwd + provider/model route); a failed handshake closes the runtime and resets, so a later call may retry. `session(id?)` opens a named or fresh session handle; `run(input, { sessionId?, onNotification? })` sends one prompt turn and settles when the paired `session.finished` arrives, returning a `TurnResult`: `status` (`ok`/`error` as the deployment maps it), the structured `reason` (`TurnEndReason`), `finalResponse` (last assistant message text), plus every `session.event` envelope and raw notification observed for that session tree, in wire order. Model-level failure is a `status: 'error'` result, never a rejection; rejections mean transport loss, timeout, or protocol violation. +The subprocess starts lazily on first use and stays owned by the instance across `run()` calls; `close()` (or `await using`) is required so the child is always reaped. `start()` memoizes the `initialize` handshake (the workspace cwd — resolved absolute before it crosses the wire — plus the provider/model route); a failed handshake reaps the runtime and swaps in a fresh client, so a later call retries with a new subprocess (until `close()`, which is terminal). `session(id?)` opens a named or fresh session handle; `run(input, { sessionId?, onNotification? })` sends one prompt turn and settles when the paired `session.finished` arrives, returning a `TurnResult`: `status` (`ok`/`error` as the deployment maps it), the structured `reason` (`TurnEndReason`), `finalResponse` (last assistant message text), plus every `session.event` envelope and raw notification observed for that session tree, in wire order. Model-level failure is a `status: 'error'` result, never a rejection; rejections mean transport loss, timeout, or protocol violation. ## HarnessClient diff --git a/packages/sdk/sdk-client/README.zh.md b/packages/sdk/sdk-client/README.zh.md index 6e67f57ab9..3814b88aab 100644 --- a/packages/sdk/sdk-client/README.zh.md +++ b/packages/sdk/sdk-client/README.zh.md @@ -20,7 +20,7 @@ const result = await harness.run('say hi') console.log(result.status, result.finalResponse) ``` -子进程在首次使用时惰性启动，并在多次 `run()` 之间持续归实例所有；必须 `close()`（或 `await using`），子进程才总能被收割。`start()` 记忆化 `initialize` 握手（cwd + provider/model 路由）；握手失败会关闭运行时并复位，后续调用可以重试。`session(id?)` 打开具名或全新的会话句柄；`run(input, { sessionId?, onNotification? })` 发送一个 prompt 回合，在配对的 `session.finished` 到达时尘埃落定，返回 `TurnResult`：`status`（按部署映射的 `ok`/`error`）、结构化 `reason`（`TurnEndReason`）、`finalResponse`（最后一条助手消息文本），以及该会话树内按线序观察到的全部 `session.event` 封套与原始通知。模型层失败是 `status: 'error'` 的结果，绝不是拒绝；拒绝意味着传输丢失、超时或协议违例。 +子进程在首次使用时惰性启动，并在多次 `run()` 之间持续归实例所有；必须 `close()`（或 `await using`），子进程才总能被收割。`start()` 记忆化 `initialize` 握手（工作区 cwd——在跨越线之前解析为绝对路径——加 provider/model 路由）；握手失败会收割运行时并换入全新客户端，后续调用用新子进程重试（直到终结性的 `close()`）。`session(id?)` 打开具名或全新的会话句柄；`run(input, { sessionId?, onNotification? })` 发送一个 prompt 回合，在配对的 `session.finished` 到达时尘埃落定，返回 `TurnResult`：`status`（按部署映射的 `ok`/`error`）、结构化 `reason`（`TurnEndReason`）、`finalResponse`（最后一条助手消息文本），以及该会话树内按线序观察到的全部 `session.event` 封套与原始通知。模型层失败是 `status: 'error'` 的结果，绝不是拒绝；拒绝意味着传输丢失、超时或协议违例。 ## HarnessClient diff --git a/packages/sdk/sdk-client/src/api.ts b/packages/sdk/sdk-client/src/api.ts index 433a80d8c7..095018b452 100644 --- a/packages/sdk/sdk-client/src/api.ts +++ b/packages/sdk/sdk-client/src/api.ts @@ -8,9 +8,10 @@ */ import { randomUUID } from 'node:crypto' +import { resolve } from 'node:path' import type { SessionEvent, TurnEndReason } from '@deepseek-ai/dsh-session' -import { HarnessClient } from './client.ts' -import type { ContentBlock, DeepSeekHarnessOptions, HarnessNotification, TurnResult } from './types.ts' +import { HarnessClient, isRecord, SdkProtocolError } from './client.ts' +import type { ContentBlock, DeepSeekHarnessOptions, HarnessClientOptions, HarnessNotification, TurnResult } from './types.ts' /** * Reusable SDK for running DeepSeek Harness agent turns in a runtime @@ -19,33 +20,52 @@ import type { ContentBlock, DeepSeekHarnessOptions, HarnessNotification, TurnRes * child is reaped. */ export class DeepSeekHarness implements AsyncDisposable { - /** The underlying JSON-RPC client (exposed for low-level access). */ - readonly client: HarnessClient + private clientInstance: HarnessClient + private readonly launch: HarnessClientOptions private readonly cwd: string private readonly provider: string private readonly model: string private initialized: Promise | undefined + private closed = false /** @param options - runtime launch spec plus the session route (cwd/provider/model). */ constructor(options: DeepSeekHarnessOptions) { - this.client = new HarnessClient(options.launch) - this.cwd = options.cwd ?? options.launch.cwd ?? process.cwd() + this.launch = options.launch + this.clientInstance = new HarnessClient(options.launch) + // Absolute before the handshake: the child spawns relative to THIS + // process's cwd, but the wire cwd is resolved again inside the child — a + // relative value would double-resolve (e.g. `worker` → `worker/worker`). + this.cwd = resolve(options.cwd ?? options.launch.cwd ?? process.cwd()) this.provider = options.provider ?? 'deepseek' this.model = options.model ?? 'deepseek-v4-flash' } /** - * Start the subprocess and perform the `initialize` handshake once. + * The underlying JSON-RPC client (exposed for low-level access). A failed + * handshake reaps its runtime and swaps in a fresh instance, so do not + * cache this across a failed {@link start}. + * @returns the client currently owning the runtime subprocess. + */ + get client(): HarnessClient { + return this.clientInstance + } + + /** + * Start the subprocess and perform the `initialize` handshake once. On + * failure the runtime is reaped and a fresh client replaces it + * (`HarnessClient.close` is permanent), so a later call retries with a new + * subprocess — unless {@link close} already ended this harness. * @returns settlement of the (memoized) handshake. */ start(): Promise { this.initialized ??= (async () => { try { - this.client.start() - await this.client.initialize({ cwd: this.cwd, provider: this.provider, model: this.model }) + this.clientInstance.start() + await this.clientInstance.initialize({ cwd: this.cwd, provider: this.provider, model: this.model }) } catch (error) { this.initialized = undefined - await this.client.close() + await this.clientInstance.close() + if (!this.closed) this.clientInstance = new HarnessClient(this.launch) throw error } })() @@ -73,11 +93,13 @@ export class DeepSeekHarness implements AsyncDisposable { } /** - * Shut down and reap the runtime subprocess. Idempotent. + * Shut down and reap the runtime subprocess. Idempotent and terminal — + * a closed harness no longer retries a failed handshake. * @returns settlement of the complete teardown. */ close(): Promise { - return this.client.close() + this.closed = true + return this.clientInstance.close() } /** @@ -128,16 +150,26 @@ export class HarnessSession { const subscription = client.subscribeSessionTree(this.id) const collect = (notification: HarnessNotification): void => { - notifications.push(notification) - options?.onNotification?.(notification) if (notification.method === 'session.event' && notification.params.sessionId === this.id) { - events.push(notification.params.event as SessionEvent) + // Wire boundary: the envelope feeds the typed TurnResult, so a + // malformed runtime surfaces as a protocol error, not as type-invalid + // data (or a TypeError out of finalResponse). + const event = validatedSessionEvent(notification.params.event) + notifications.push(notification) + options?.onNotification?.(notification) + events.push(event) + return } if (notification.method === 'session.finished' && notification.params.sessionId === this.id) { + reason = validatedTurnEndReason(notification.params.reason) + notifications.push(notification) + options?.onNotification?.(notification) status = notification.params.status === 'ok' ? 'ok' : 'error' - reason = notification.params.reason as TurnEndReason | undefined finished = true + return } + notifications.push(notification) + options?.onNotification?.(notification) } const accepted = client.prompt(this.id, contentBlocks) // Drain concurrently so observers see progress while the prompt request @@ -175,6 +207,32 @@ export function normalizeInput(input: string | ContentBlock[]): ContentBlock[] { return typeof input === 'string' ? [{ type: 'text', text: input }] : input } +/** Validate a wire `session.event` envelope to the shape the typed result exposes. */ +function validatedSessionEvent(value: unknown): SessionEvent { + if (!isRecord(value) || typeof value.type !== 'string') { + throw new SdkProtocolError(`session.event carried no event envelope: ${JSON.stringify(value)}`) + } + // The one variant this module reads into (finalResponse) must carry + // kind-tagged content blocks; other variants pass through under their + // envelope shape. + if (value.type === 'assistant/message') { + const content = isRecord(value.data) ? value.data.content : undefined + if (!Array.isArray(content) || !content.every(block => isRecord(block) && typeof block.type === 'string')) { + throw new SdkProtocolError(`assistant/message event carried malformed content: ${JSON.stringify(value)}`) + } + } + return value as unknown as SessionEvent +} + +/** Validate a wire `session.finished` reason (absent, or a kind-tagged record). */ +function validatedTurnEndReason(value: unknown): TurnEndReason | undefined { + if (value === undefined) return undefined + if (!isRecord(value) || typeof value.kind !== 'string') { + throw new SdkProtocolError(`session.finished carried a malformed reason: ${JSON.stringify(value)}`) + } + return value as unknown as TurnEndReason +} + /** * Extract the concatenated text of the last assistant message. * @param events - the turn's `session.event` payloads in wire order. diff --git a/packages/sdk/sdk-client/src/client.ts b/packages/sdk/sdk-client/src/client.ts index bf852b8eb1..447ea75d5a 100644 --- a/packages/sdk/sdk-client/src/client.ts +++ b/packages/sdk/sdk-client/src/client.ts @@ -81,8 +81,9 @@ export class NotificationSubscription implements AsyncIterable { const queued = this.state.queue.shift() @@ -104,11 +105,15 @@ export class NotificationSubscription implements AsyncIterable { + abandon.abort(new RequestTimeoutError(`${method} timed out after ${timeout}ms waiting for the DeepSeek Harness runtime`)) + }, timeout) try { - return await Promise.race([ - pending, - new Promise((_, reject) => { - timer = setTimeout(() => { - // The abandoned wire promise settles on close; keep it handled. - pending.catch(() => {}) - reject(new RequestTimeoutError(`${method} timed out after ${timeout}ms waiting for the DeepSeek Harness runtime`)) - }, timeout) - }), - ]) + return await transport.request(method, params ?? {}, abandon.signal) } finally { clearTimeout(timer) } @@ -303,12 +315,18 @@ export class HarnessClient { /** * Subscribe to server notifications. * @param filter - optional predicate; omitted means every notification. - * @returns the subscription handle; close it to stop delivery. + * @returns the subscription handle; close it to stop delivery. After + * {@link close} or runtime death the handle is born failed — there is no + * producer left, so `next()` rejects instead of waiting forever. */ subscribe(filter?: NotificationFilter): NotificationSubscription { const id = String(this.subscriptionSerial++) const state: SubscriptionState = { queue: [], waiters: [], filter, failure: undefined } const subscription = new NotificationSubscription(state, () => { this.subscriptions.delete(id) }) + if (this.closeTask !== undefined || this.exitCode !== undefined || this.spawnError !== undefined) { + subscription.fail(this.closedError('DeepSeek Harness runtime closed')) + return subscription + } this.subscriptions.set(id, subscription) return subscription } diff --git a/packages/sdk/sdk-client/tests/fake-runtime.ts b/packages/sdk/sdk-client/tests/fake-runtime.ts index 0c8a928c34..c07085f3dd 100644 --- a/packages/sdk/sdk-client/tests/fake-runtime.ts +++ b/packages/sdk/sdk-client/tests/fake-runtime.ts @@ -16,6 +16,16 @@ * - `FAKE_MALFORMED`: `initialize` returns `{}` (no serverInfo); `prompt` returns `{}` (no accepted). * - `FAKE_MALFORMED_PROMPT`: `initialize` is normal; only `prompt` returns `{}` (no accepted). * - `FAKE_INIT_ERROR`: `initialize` answers a JSON-RPC error response with code 7. + * - `FAKE_INIT_ERROR_ONCE_FILE`: fail `initialize` (code 7) only when this + * marker file does NOT exist yet, creating it — so the first runtime + * process fails the handshake and a respawned one succeeds (retry probe). + * - `FAKE_ECHO_CWD_IN_INIT`: reply `serverInfo.version` = this process's cwd + * (wire-visible spawn-cwd probe). + * - `FAKE_MALFORMED_EVENT`: the turn's `session.event` carries a number as + * the event; `FAKE_MALFORMED_MESSAGE`: assistant/message content is not an + * array; `FAKE_MESSAGE_WITHOUT_DATA`: assistant/message with no data + * member; `FAKE_MALFORMED_REASON`: `session.finished` reason is a bare + * string (wire-validation probes). * - `FAKE_HANG_INIT`: never answer `initialize` (mid-handshake cancel probe). * - `FAKE_INIT_READY` + `FAKE_INIT_GO`: touch the READY file when `initialize` * arrives, then poll for the GO file before answering (deterministic @@ -78,8 +88,20 @@ function assistantText(): string { function runTurn(sessionId: string): void { const text = assistantText() + if (env.FAKE_MALFORMED_EVENT !== undefined) { + notify('session.event', { sessionId, event: 42 }) + return + } event(sessionId, 'turn/start', { turn: 0 }) event(sessionId, 'assistant/chunk', { turn: 0, step: 0, chunk: { type: 'text-delta', index: 0, text } }) + if (env.FAKE_MALFORMED_MESSAGE !== undefined) { + event(sessionId, 'assistant/message', { turn: 0, step: 0, content: 'not-an-array' }) + return + } + if (env.FAKE_MESSAGE_WITHOUT_DATA !== undefined) { + notify('session.event', { sessionId, event: { type: 'assistant/message', seq: seq++, time: 0 } }) + return + } event(sessionId, 'assistant/message', { turn: 0, step: 0, @@ -110,7 +132,9 @@ function runTurn(sessionId: string): void { notify('session.finished', { sessionId, status: env.FAKE_STATUS ?? 'ok', - ...(reasonKind === 'none' ? {} : { reason: { kind: reasonKind } }), + ...(env.FAKE_MALFORMED_REASON !== undefined + ? { reason: 'not-a-record' } + : reasonKind === 'none' ? {} : { reason: { kind: reasonKind } }), }) } @@ -144,10 +168,19 @@ reader.on('line', (line) => { write({ jsonrpc: '2.0', id: frame.id, error: { code: 7, message: 'scripted init failure', data: { hint: 'fake' } } }) return } + if (env.FAKE_INIT_ERROR_ONCE_FILE !== undefined && !existsSync(env.FAKE_INIT_ERROR_ONCE_FILE)) { + writeFileSync(env.FAKE_INIT_ERROR_ONCE_FILE, 'failed-once\n') + write({ jsonrpc: '2.0', id: frame.id, error: { code: 7, message: 'scripted first-boot failure' } }) + return + } if (env.FAKE_MALFORMED !== undefined) { respond({}) return } + if (env.FAKE_ECHO_CWD_IN_INIT !== undefined) { + respond({ serverInfo: { name: 'deepseek-harness-sdk-runtime', version: process.cwd() } }) + return + } respond({ serverInfo: { name: 'deepseek-harness-sdk-runtime', version: '0.0.1' } }) return case 'session/prompt': { diff --git a/packages/sdk/sdk-client/tests/sdk-client.spec.ts b/packages/sdk/sdk-client/tests/sdk-client.spec.ts index 3feff693f4..171c0f0655 100644 --- a/packages/sdk/sdk-client/tests/sdk-client.spec.ts +++ b/packages/sdk/sdk-client/tests/sdk-client.spec.ts @@ -5,9 +5,9 @@ * and session-tree scoping, error surfaces, timeouts, and the dispose ladder. */ -import { mkdtemp, readFile, rm, stat } from 'node:fs/promises' +import { mkdir, mkdtemp, readFile, realpath, rm, stat } from 'node:fs/promises' import { tmpdir } from 'node:os' -import { join } from 'node:path' +import { isAbsolute, join, relative, resolve as resolvePath } from 'node:path' import { fileURLToPath } from 'node:url' import { afterEach, describe, expect, it } from 'vitest' import { @@ -122,6 +122,31 @@ describe('DeepSeekHarness', () => { expect(records).toEqual([{ cwd: dir, provider: 'custom-provider', model: 'custom-model' }]) }) + it('resolves a relative launch cwd to an absolute workspace before the handshake', async () => { + // vitest workers forbid chdir, so derive a RELATIVE path from the real + // process cwd to a temp worker dir; resolution is lexical either way. + const dir = await tempDir('sdk-client-relcwd-') + const recordFile = join(dir, 'init.jsonl') + const inner = join(dir, 'worker') + await mkdir(inner) + const relativeCwd = relative(process.cwd(), inner) + expect(isAbsolute(relativeCwd)).toBe(false) + const harness = new DeepSeekHarness({ + launch: fakeLaunch({ FAKE_RECORD_INIT: recordFile, FAKE_ECHO_CWD_IN_INIT: '1' }, { cwd: relativeCwd }), + }) + cleanups.push(() => harness.close()) + await harness.start() + const identity = await harness.client.initialize({ cwd: inner, provider: 'p', model: 'm' }) + await harness.close() + // The child spawned under the temp worker dir (its physical cwd)... + expect(identity.serverInfo.version).toBe(await realpath(inner)) + // ...and the handshake wire cwd went out ABSOLUTE, so the child cannot + // re-resolve a relative string into dir/worker/worker. + const records = (await readFile(recordFile, 'utf8')).trim().split('\n') + .map(line => (JSON.parse(line) as { cwd: string }).cwd) + expect(records).toEqual([resolvePath(relativeCwd), inner]) + }) + it('propagates a JSON-RPC error response from initialize and closes the runtime', async () => { const harness = harnessWith({ FAKE_INIT_ERROR: '1' }) const failure = await harness.run('boom').then( @@ -134,6 +159,23 @@ describe('DeepSeekHarness', () => { await expect(harness.run('later')).rejects.toThrow() }) + it('retries a failed handshake with a fresh runtime process', async () => { + const dir = await tempDir('sdk-client-retry-') + const marker = join(dir, 'first-boot-failed') + const harness = harnessWith({ FAKE_INIT_ERROR_ONCE_FILE: marker, FAKE_TEXT: 'second boot answer' }) + const firstClient = harness.client + // First start: the scripted runtime fails the handshake and is reaped. + await expect(harness.start()).rejects.toThrow('scripted first-boot failure') + // Retry spawns a NEW subprocess through a fresh client (close is permanent). + const result = await harness.run('again') + expect(harness.client).not.toBe(firstClient) + expect(result.status).toBe('ok') + expect(result.finalResponse).toBe('second boot answer') + await harness.close() + // close() is terminal: a handshake failure after it must not respawn. + await expect(harness.run('after-close')).rejects.toThrow(TransportClosedError) + }) + it('rejects a malformed initialize result as a protocol error', async () => { const harness = harnessWith({ FAKE_MALFORMED: '1' }) await expect(harness.run('bad')).rejects.toThrow(SdkProtocolError) @@ -162,6 +204,22 @@ describe('HarnessClient', () => { await client.close() }) + it('a timed-out request leaves no pending transport state', async () => { + const client = new HarnessClient(fakeLaunch({ FAKE_HANG_PROMPT: '1' })) + cleanups.push(() => client.close()) + await client.initialize({ cwd: process.cwd(), provider: 'p', model: 'm' }) + for (let round = 0; round < 3; round++) { + await expect(client.request('session/prompt', { sessionId: 's', contentBlocks: normalizeInput('x') }, 50)) + .rejects.toThrow(RequestTimeoutError) + } + // Abandonment removed each pending entry at its timeout; a hung method + // retains nothing per call. (Private map read is the observable here — + // no wire surface reports transport bookkeeping.) + const transport = (client as unknown as { transport: { pending: Map } }).transport + expect(transport.pending.size).toBe(0) + await client.close() + }) + it('applies the client-wide request timeout when no per-call bound is given', async () => { const client = new HarnessClient(fakeLaunch({ FAKE_HANG_PROMPT: '1' }, { requestTimeoutMs: 400 })) cleanups.push(() => client.close()) @@ -255,6 +313,10 @@ describe('HarnessClient', () => { expect(finished.method).toBe('session.finished') expect(finishedOnly.tryNext()).toBeUndefined() + // A bare unbounded request with omitted params sends `{}` on the wire. + const identity = await client.request('initialize') as { serverInfo: { name: string } } + expect(identity.serverInfo.name).toBe('deepseek-harness-sdk-runtime') + // Async iteration consumes queued items and then parks. const collected: string[] = [] for await (const notification of all) { @@ -269,6 +331,56 @@ describe('HarnessClient', () => { await client.close() }) + it('contains a throwing filter to its own subscription', async () => { + const client = new HarnessClient(fakeLaunch()) + cleanups.push(() => client.close()) + await client.initialize({ cwd: process.cwd(), provider: 'p', model: 'm' }) + + const broken = client.subscribe(() => { throw new Error('filter exploded') }) + // A non-Error throw is normalized rather than crashing dispatch. + const brokenNonError = client.subscribe(() => { throw 'string boom' }) + const healthy = client.subscribe(n => n.method === 'session.finished') + await client.prompt('filter-contain', normalizeInput('go')) + + // The sibling subscription and the read loop are undisturbed. + expect((await healthy.next()).method).toBe('session.finished') + // Each broken subscription failed with ITS OWN error and detached. + await expect(broken.next()).rejects.toThrow('filter exploded') + await expect(brokenNonError.next()).rejects.toThrow('string boom') + healthy.close() + await client.close() + }) + + it('close() drops queued notifications; runtime death keeps them drainable', async () => { + const client = new HarnessClient(fakeLaunch()) + await client.initialize({ cwd: process.cwd(), provider: 'p', model: 'm' }) + const closed = client.subscribe() + const drainable = client.subscribe() + await client.prompt('queue-drop', normalizeInput('go')) + expect(closed.tryNext()).toBeDefined() + closed.close() + // Manual close drops the rest of the queue outright. + expect(closed.tryNext()).toBeUndefined() + await expect(closed.next()).rejects.toThrow('notification subscription closed') + // Runtime teardown, by contrast, only stops FUTURE delivery: what was + // already delivered before close() stays drainable. + await client.close() + expect(drainable.tryNext()).toBeDefined() + }) + + it('subscriptions created after termination are born failed', async () => { + const client = new HarnessClient(fakeLaunch()) + await client.initialize({ cwd: process.cwd(), provider: 'p', model: 'm' }) + await client.close() + // No producer can ever feed this subscription; next() must not park forever. + await expect(client.subscribe().next()).rejects.toThrow(TransportClosedError) + + const dead = new HarnessClient(fakeLaunch({ FAKE_EXIT_BEFORE_INIT: '1' })) + cleanups.push(() => dead.close()) + await dead.initialize({ cwd: process.cwd(), provider: 'p', model: 'm' }).catch(() => {}) + await expect(dead.subscribe().next()).rejects.toThrow(TransportClosedError) + }) + it('closes subscriptions with the runtime and rejects parked waiters', async () => { const client = new HarnessClient(fakeLaunch()) await client.initialize({ cwd: process.cwd(), provider: 'p', model: 'm' }) @@ -310,6 +422,28 @@ describe('HarnessClient', () => { }) }) +describe('wire payload validation', () => { + it('rejects a non-object session.event envelope as a protocol error', async () => { + const harness = harnessWith({ FAKE_MALFORMED_EVENT: '1' }) + await expect(harness.run('bad-event')).rejects.toThrow(SdkProtocolError) + }) + + it('rejects an assistant/message without a content array as a protocol error', async () => { + const harness = harnessWith({ FAKE_MALFORMED_MESSAGE: '1' }) + await expect(harness.run('bad-message')).rejects.toThrow(SdkProtocolError) + }) + + it('rejects an assistant/message without a data member as a protocol error', async () => { + const harness = harnessWith({ FAKE_MESSAGE_WITHOUT_DATA: '1' }) + await expect(harness.run('no-data')).rejects.toThrow(SdkProtocolError) + }) + + it('rejects a malformed session.finished reason as a protocol error', async () => { + const harness = harnessWith({ FAKE_MALFORMED_REASON: '1' }) + await expect(harness.run('bad-reason')).rejects.toThrow(SdkProtocolError) + }) +}) + describe('stderr tail bound', () => { it('keeps only the newest lines up to the limit', async () => { const manyLines = Array.from({ length: 450 }, (_, i) => `line-${i}`).join('\n') diff --git a/packages/sdk/sdk-protocol/src/transport.ts b/packages/sdk/sdk-protocol/src/transport.ts index a1291f019b..36574f46bf 100644 --- a/packages/sdk/sdk-protocol/src/transport.ts +++ b/packages/sdk/sdk-protocol/src/transport.ts @@ -109,15 +109,47 @@ export class JsonRpcLineTransport implements JsonRpcTransportPeer { this.notificationHandler = handler } - request(method: string, params: object): Promise { + /** + * Send a request and await its response. + * @param method - the JSON-RPC method name. + * @param params - the request parameters object. + * @param signal - optional abandonment signal: aborting removes the pending + * entry (no state is retained for a response that may never come) and + * rejects with the signal's reason. + * @returns the result; rejects per {@link JsonRpcTransportPeer.request}. + */ + request(method: string, params: object, signal?: AbortSignal): Promise { const id = `req_${randomUUID().replaceAll('-', '')}` const message = { jsonrpc: '2.0', id, method, params } return new Promise((resolve, reject) => { - this.pending.set(id, { resolve, reject }) + let detach = (): void => {} + if (signal !== undefined) { + if (signal.aborted) { + reject(abortError(signal.reason)) + return + } + const onAbort = (): void => { + this.pending.delete(id) + reject(abortError(signal.reason)) + } + signal.addEventListener('abort', onAbort, { once: true }) + detach = () => { signal.removeEventListener('abort', onAbort) } + } + this.pending.set(id, { + resolve: (value) => { + detach() + resolve(value) + }, + reject: (error) => { + detach() + reject(error) + }, + }) try { this.write(message) } catch (error) { this.pending.delete(id) + detach() reject(error instanceof Error ? error : new Error(String(error))) } }) @@ -240,3 +272,8 @@ export class JsonRpcLineTransport implements JsonRpcTransportPeer { function objectParams(params: unknown): Record { return params && typeof params === 'object' && !Array.isArray(params) ? params as Record : {} } + +/** Normalize an abort reason into the rejection Error (a non-Error reason is stringified). */ +function abortError(reason: unknown): Error { + return reason instanceof Error ? reason : new Error(`JSON-RPC request aborted: ${String(reason)}`) +} diff --git a/packages/sdk/sdk-protocol/tests/transport.spec.ts b/packages/sdk/sdk-protocol/tests/transport.spec.ts index 7eedb12867..a07324a6fc 100644 --- a/packages/sdk/sdk-protocol/tests/transport.spec.ts +++ b/packages/sdk/sdk-protocol/tests/transport.spec.ts @@ -60,6 +60,29 @@ describe('JsonRpcLineTransport', () => { b.close() }) + it('rejects immediately on a pre-aborted signal without registering pending state', async () => { + const { b } = transportPair() + b.start() + const controller = new AbortController() + controller.abort(new Error('already gone')) + await expect(b.request('never-sent', {}, controller.signal)).rejects.toThrow('already gone') + expect((b as unknown as { pending: Map }).pending.size).toBe(0) + b.close() + }) + + it('abandons a pending request on abort, stringifying a non-Error reason', async () => { + const { b } = transportPair() + b.start() + const controller = new AbortController() + const pending = b.request('never-answered', {}, controller.signal) + controller.abort('plain-string-reason') + await expect(pending).rejects.toThrow('JSON-RPC request aborted: plain-string-reason') + // The abandonment removed the pending entry — nothing is retained for a + // response that may never come. + expect((b as unknown as { pending: Map }).pending.size).toBe(0) + b.close() + }) + it('preserves structured error data from an error response frame', async () => { const { aToB, bToA, b } = transportPair() b.start()