From 06549daa9b2cac5a89f3f7826cb5c914199ff222 Mon Sep 17 00:00:00 2001 From: jinye Date: Wed, 24 Jun 2026 13:59:59 +0800 Subject: [PATCH] fix(sdk): self-heal stale clientId on invalid_client_id prompts (#5797) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(sdk): self-heal stale clientId on invalid_client_id prompts After a daemon restart or session reload the daemon's in-memory client registration is wiped, so a prompt sent with our now-unknown clientId is rejected at admission with 400 invalid_client_id (PR #5784). That rejection happens before the turn registers, so the prompt never ran and retrying cannot double-execute. DaemonSessionClient.prompt() now wraps the admission call (both the blocking and non-blocking paths) in a self-heal: on invalid_client_id it re-registers via resumeSession to obtain a fresh clientId and retries the admission exactly once. A single-flight guard coalesces concurrent prompts so re-registration happens once; any other error (and a second invalid_client_id) propagates. Adds 6 tests covering both paths, the retry bound, the non-matching-error guard, reattach-failure propagation, and concurrent single-flight. Design: docs/superpowers/specs/2026-06-24-daemon-clientid-self-heal-design.md πŸ€– Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) * test(integration): cover real-daemon invalid_client_id contract for self-heal Adds an integration case to qwen-serve-routes.test.ts validating the three real-daemon behaviors DaemonSessionClient's clientId self-heal depends on: (1) an unregistered prompt clientId is rejected at admission with 400 invalid_client_id, (2) resume re-registers and mints a fresh clientId, and (3) retrying admission with that clientId is accepted (202). Model-free: prompt admission runs before any model call, so promptNonBlocking returns 202 on acceptance without reaching the model. πŸ€– Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) * fix(sdk): Adjust daemon browser bundle budget Co-authored-by: Qwen-Coder * test(sdk): Cover clientId self-heal review cases Co-authored-by: Qwen-Coder --------- Co-authored-by: Qwen-Coder --- ...-06-24-daemon-clientid-self-heal-design.md | 216 +++++++++++++ .../cli/qwen-serve-routes.test.ts | 55 ++++ packages/sdk-typescript/scripts/build.js | 3 +- .../src/daemon/DaemonSessionClient.ts | 81 ++++- .../test/unit/DaemonSessionClient.test.ts | 300 ++++++++++++++++++ 5 files changed, 644 insertions(+), 11 deletions(-) create mode 100644 docs/superpowers/specs/2026-06-24-daemon-clientid-self-heal-design.md diff --git a/docs/superpowers/specs/2026-06-24-daemon-clientid-self-heal-design.md b/docs/superpowers/specs/2026-06-24-daemon-clientid-self-heal-design.md new file mode 100644 index 0000000000..d54ddc2f13 --- /dev/null +++ b/docs/superpowers/specs/2026-06-24-daemon-clientid-self-heal-design.md @@ -0,0 +1,216 @@ +# Design: clientId self-heal on `invalid_client_id` (DaemonSessionClient) + +- **Date:** 2026-06-24 +- **Component:** `packages/sdk-typescript` β€” `DaemonSessionClient` +- **Depends on:** PR #5784 (`fix(daemon): Reject stale prompt client admission`) β€” **merged** (`84745d0f0`) +- **Status:** Implemented (built on the merged #5784 base) + +## Problem + +After a daemon restart (or session reload), the daemon's in-memory client +registration is wiped. A frontend that still holds an older server-assigned +`clientId` will send `POST /session/:id/prompt` with that stale id. The bridge's +`resolveTrustedClientId` does not recognize it and rejects the prompt with +`InvalidClientIdError`. + +Observed production incident (trace `a76a31fe…`, daemon log 15:24): the prompt +was sent by `client_d019b847` while the session had been (re)loaded under a +different id `client_ac36fac9`, so the prompt-sending client was never +registered. The UI stayed in "倄理中" indefinitely because the failure was never +surfaced as a terminal turn event. + +PR #5784 fixes the _surfacing_ half: `invalid_client_id` is now thrown at +**admission time** so `POST /session/:id/prompt` returns a synchronous +`400 invalid_client_id` (no `promptId`) instead of `202`-then-silent-async-fail. +This design adds the _self-heal_ half: when the SDK receives that `400`, it +re-registers to obtain a fresh `clientId` and retries the prompt once, so the +turn proceeds without the user having to manually resend. + +## Scope + +In scope (SDK only, `DaemonSessionClient`): + +- Detect `invalid_client_id` on the prompt admission call. +- Re-register the client against the (already-restored) session to get a fresh + server-assigned `clientId`. +- Retry the prompt **once** with the new `clientId`. + +Explicitly out of scope (YAGNI): + +- SSE stream reconnection β€” remains the app layer's existing responsibility + (the dataworks app already owns `reloadSession`/reconnect logic). `invalid_client_id` + only surfaces on the admission call, never on the SSE wait. +- Self-heal for other `clientId`-bearing methods (`btw`, `shell`, mid-turn + message, `cancel`, `heartbeat`). Only `prompt()` self-heals. +- Persisting `clientId` across daemon restarts. + +## Key invariants (verified against source) + +1. **Retry is safe because `invalid_client_id` is an admission-time rejection.** + `resolveTrustedClientId` runs inside `bridge.sendPrompt` _before_ the turn is + registered and before the route emits `202`. With PR #5784 this throws + synchronously β†’ `400` before acceptance β†’ the prompt **never executed**. + Retrying therefore cannot double-execute the user's message. This invariant is + the entire basis for the retry being safe; it depends on #5784. + +2. **`registerClient` never throws and always yields a valid id.** For an unknown + `requestedClientId` it falls through to `createClientId()` and returns a fresh + `client_`. Only `resolveTrustedClientId` (used by prompt/cancel/…) throws. + So a `load`/`resume` call always returns a usable `clientId`. + +3. **The restore response always carries the registered `clientId`.** Both the + existing-entry fast path and the cold-restore path set + `clientId: registerClient(entry, req.clientId)` in the response. (The "echoed + back only when the caller supplied a clientId" note in `types.ts` applies to + `HeartbeatResult`, not to restore.) + +4. **No net attach leak in the restart scenario, and `close()` correctness + improves.** `resumeSession` does `attachCount++`. The refcounted decrement is + `/detach` β†’ `detachClient` (`attachCount--` + `unregisterClient`). `close()` β†’ + `DELETE /session/:id` β†’ `closeSessionImpl` is **destroy-all**: it validates the + clientId via `resolveTrustedClientId` and then tears the session down + (`byId.delete`), discarding `attachCount` with it. A daemon restart wipes the + pre-restart attach; `reattach()` re-establishes exactly one attach, and a later + `close()`/restart tears it all down β€” no net leak. Note `closeSessionImpl` also + validates the clientId, so before this change a post-restart `close()` with a + stale id would itself throw `InvalidClientIdError`; after a prompt-triggered + `reattach()`, `this.clientId` is valid so `close()` succeeds. (`close()` is not + itself self-healed β€” out of scope β€” but benefits indirectly.) + +5. **The change is inert without PR #5784.** A pre-#5784 daemon returns + `202`-then-async-fail, never `400 invalid_client_id`, so the predicate never + matches and self-heal never triggers. Harmless no-op. + +## Design + +All changes are confined to +`packages/sdk-typescript/src/daemon/DaemonSessionClient.ts`. + +### 1. `isInvalidClientId(err): boolean` + +```ts +function isInvalidClientId(err: unknown): boolean { + return ( + err instanceof DaemonHttpError && + err.status === 400 && + typeof err.body === 'object' && + err.body !== null && + (err.body as { code?: unknown }).code === 'invalid_client_id' + ); +} +``` + +Requires importing `DaemonHttpError` from `./DaemonHttpError.js`. + +### 2. `reattach(): Promise` β€” single-flight + +```ts +private reattaching?: Promise; + +private async reattach(): Promise { + // Coalesce concurrent prompts that all observed invalid_client_id so we + // re-register exactly once (avoids orphaning extra clientIds / attachCount). + if (this.reattaching) return this.reattaching; + this.reattaching = (async () => { + // Pass no clientId so the bridge issues a fresh registration instead of + // validating the stale one. Pass workspaceCwd explicitly: restoreSession + // calls resolveWorkspaceKey(req.workspaceCwd) before the existing-entry + // fast path, and that helper throws on a non-absolute/undefined path. + const { clientId } = await this.client.resumeSession( + this.sessionId, + { workspaceCwd: this.workspaceCwd }, + undefined, + ); + this.session.clientId = clientId; // only refresh clientId; leave the SSE + // cursor (lastSeenEventId) and state alone + })(); + try { + await this.reattaching; + } finally { + this.reattaching = undefined; + } +} +``` + +`this.session` is a shallow copy and `DaemonSession.clientId` is not `readonly`, +so in-place mutation is valid. `resume` (not `load`) is used because we only need +re-registration, not history replay. + +### 3. `withClientIdSelfHeal(fn): Promise` + +```ts +private async withClientIdSelfHeal(fn: () => Promise): Promise { + try { + return await fn(); + } catch (err) { + if (!isInvalidClientId(err)) throw err; // non-invalid_client_id: propagate + await this.reattach(); // may throw β†’ propagate + return await fn(); // retry exactly once; if it throws + // again (incl. invalid_client_id), + // propagate β€” no loop + } +} +``` + +### 4. Wiring into `prompt()` + +Wrap only the admission network call on both paths; keep +`reservePromptSlot`/`releaseAdmission` outside the wrapper so the local slot is +reserved once and reused across the retry: + +- Blocking path (`!this.subscriptionActive`): + `return await this.withClientIdSelfHeal(() => this.client.prompt(this.sessionId, req, signal, this.clientId));` +- Non-blocking path: + `accepted = await this.withClientIdSelfHeal(() => this.client.promptNonBlocking(this.sessionId, req, signal, this.clientId));` + +`this.clientId` is read **inside** the closure so the retry picks up the +refreshed id. Everything after admission (the `_pendingPrompts` registration and +SSE turn-event matching by `promptId`) is unchanged; the SSE subscription is keyed +by `sessionId`, so it survives the `clientId` change. + +## Error handling + +- Non-`invalid_client_id` errors (e.g. `500`, `SessionNotFoundError`, + `DaemonPendingPromptLimitError`): propagated immediately, no `reattach`. +- `reattach()` failure (session truly gone, network): propagated β€” the user sees + a real error instead of a hang. +- Retry exhausted (retry also `invalid_client_id`): propagated; bounded to one + retry, no loop. +- `AbortSignal`: the wrapped `prompt`/`promptNonBlocking` call `throwIfAborted()` + at entry, so a retry after abort throws `AbortError`. (`resumeSession` has no + signal parameter; a `reattach` in flight is not abortable β€” acceptable, it is a + single short call.) + +## Known limitations + +- **Rare individual-eviction edge:** if a `clientId` is evicted while the session + stays alive in memory (leak-revocation / `client_evicted`), `reattach()` adds an + extra attach (`attachCount++`) with no matching `/detach`. Because `close()` is + destroy-all, the only leak window is a session that is abandoned without an + explicit `close()` and is then kept from idle-GC by the stuck `attachCount` + (bounded to one session). The realistic incident is the daemon-restart case, + which is clean. Documented rather than engineered around. + +## Testing (TDD) + +Use the existing `recordingFetch` harness in +`packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts`, intercepting by +URL through a real `DaemonClient` (exercises the real `failOnError` β†’ +`DaemonHttpError` mapping). + +1. **Non-blocking self-heal:** first `POST /session/s-1/prompt` β†’ `400 +{code:'invalid_client_id'}`; `POST /session/s-1/resume` β†’ fresh + `clientId: 'client-2'`; second prompt β†’ `202`. Assert: prompt resolves, the + second prompt request carries `x-qwen-client-id: client-2`, resume called once. +2. **Blocking self-heal** (`subscriptionActive` false): same, via the blocking + `prompt` path (`200`/`202`+turn-complete on retry). +3. **Retry bounded:** prompt β†’ `400 invalid_client_id` twice β†’ the error + propagates (assert resume called once, error is `DaemonHttpError` + invalid_client_id). +4. **Non-invalid error not retried:** prompt β†’ `500` β†’ propagates immediately, + `resume` **never** called. +5. **reattach failure propagates:** prompt β†’ `400 invalid_client_id`; resume β†’ + `404`/`500` β†’ that error propagates. +6. **Single-flight:** two concurrent `prompt()` calls both get + `400 invalid_client_id` β†’ `resume` called exactly once; both retries use the + new id. diff --git a/integration-tests/cli/qwen-serve-routes.test.ts b/integration-tests/cli/qwen-serve-routes.test.ts index 30dad436c6..e9cfa246ae 100644 --- a/integration-tests/cli/qwen-serve-routes.test.ts +++ b/integration-tests/cli/qwen-serve-routes.test.ts @@ -576,3 +576,58 @@ describe('qwen serve β€” PATCH /session/:id/metadata', () => { await client.closeSession(session.sessionId); }); }); + +describe('qwen serve β€” prompt clientId admission', () => { + // Validates the three real-daemon behaviors that DaemonSessionClient's + // clientId self-heal relies on (see + // docs/superpowers/specs/2026-06-24-daemon-clientid-self-heal-design.md). + // Model-free: prompt admission (where invalid_client_id is decided) runs + // before any model call, so promptNonBlocking returns 202 on acceptance + // without reaching the (unreachable, fake) model. + it('rejects an unregistered prompt clientId and re-registers via resume', async () => { + const session = await client.createOrAttachSession({ + workspaceCwd: REPO_ROOT, + sessionScope: 'thread', + }); + const prompt = { prompt: [{ type: 'text', text: 'hi' }] }; + + // (1) An unregistered clientId (e.g. one held across a daemon restart) is + // rejected at admission with 400 invalid_client_id β€” the exact signal + // the SDK self-heals on. + const rejected = await client + .promptNonBlocking( + session.sessionId, + prompt, + undefined, + 'client-never-registered', + ) + .catch((err: unknown) => err); + expect(rejected).toBeInstanceOf(DaemonHttpError); + expect((rejected as DaemonHttpError).status).toBe(400); + expect((rejected as DaemonHttpError).body).toMatchObject({ + code: 'invalid_client_id', + }); + + // (2) resume re-registers and mints a fresh, valid clientId. + const reattached = await client.resumeSession(session.sessionId, { + workspaceCwd: REPO_ROOT, + }); + expect(reattached.clientId).toBeTypeOf('string'); + expect(reattached.clientId).not.toBe('client-never-registered'); + + // (3) Retrying admission with the fresh clientId is accepted (202), + // proving reattach + retry recovers the turn end-to-end. + const accepted = await client.promptNonBlocking( + session.sessionId, + prompt, + undefined, + reattached.clientId, + ); + expect(accepted).toMatchObject({ promptId: expect.any(String) }); + + // The accepted turn dispatches to the unreachable fake model + // asynchronously; cancel so nothing lingers past the test. + await client.cancel(session.sessionId, reattached.clientId).catch(() => {}); + await client.closeSession(session.sessionId); + }); +}); diff --git a/packages/sdk-typescript/scripts/build.js b/packages/sdk-typescript/scripts/build.js index 26343ffcea..f917c2018e 100755 --- a/packages/sdk-typescript/scripts/build.js +++ b/packages/sdk-typescript/scripts/build.js @@ -29,7 +29,8 @@ const rootDir = join(__dirname, '..'); // (install/update/enable/disable/uninstall/refresh/check update endpoints). // Bumped from 122KB to 124KB for daemon fork-session APIs/events. // Bumped from 124KB to 125KB for rewind/branch transcript/session APIs. -const MAX_DAEMON_BROWSER_BUNDLE_BYTES = 125 * 1024; +// Bumped from 125KB to 127KB for prompt clientId self-heal. +const MAX_DAEMON_BROWSER_BUNDLE_BYTES = 127 * 1024; rmSync(join(rootDir, 'dist'), { recursive: true, force: true }); mkdirSync(join(rootDir, 'dist'), { recursive: true }); diff --git a/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts b/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts index 3c7cb4d829..c82b5a104b 100644 --- a/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts +++ b/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts @@ -5,6 +5,7 @@ */ import type { DaemonClient } from './DaemonClient.js'; +import { DaemonHttpError } from './DaemonHttpError.js'; import { isNonBlockingAccepted, matchTurnEvent, @@ -96,6 +97,8 @@ export class DaemonSessionClient { readonly replaySnapshot: DaemonReplaySnapshot; private lastSeenEventId: number | undefined; private subscriptionActive = false; + /** In-flight `reattach()` so concurrent prompts re-register only once. */ + private reattaching?: Promise; private readonly promptLimit: number; private readonly _pendingPrompts = new Map< string, @@ -245,11 +248,8 @@ export class DaemonSessionClient { ): Promise { signal?.throwIfAborted(); if (!this.subscriptionActive) { - return await this.client.prompt( - this.sessionId, - req, - signal, - this.clientId, + return await this.withClientIdSelfHeal(() => + this.client.prompt(this.sessionId, req, signal, this.clientId), ); } @@ -259,11 +259,13 @@ export class DaemonSessionClient { ); let accepted: NonBlockingPromptAccepted | PromptResult; try { - accepted = await this.client.promptNonBlocking( - this.sessionId, - req, - signal, - this.clientId, + accepted = await this.withClientIdSelfHeal(() => + this.client.promptNonBlocking( + this.sessionId, + req, + signal, + this.clientId, + ), ); if (!isNonBlockingAccepted(accepted)) { releaseAdmission(); @@ -308,6 +310,51 @@ export class DaemonSessionClient { }); } + /** + * Run a prompt-admission call, recovering from a stale `clientId`. + * + * A daemon restart (or session reload) wipes the daemon's in-memory client + * registration, so a prompt sent with our now-unknown `clientId` is rejected + * at admission with `400 invalid_client_id` (see PR #5784). That rejection + * happens before the turn is registered, so the prompt never ran β€” retrying + * cannot double-execute. We re-register to obtain a fresh `clientId` and + * retry the admission exactly once. Any other error (and a second + * `invalid_client_id`) propagates. + */ + private async withClientIdSelfHeal(fn: () => Promise): Promise { + try { + return await fn(); + } catch (err) { + if (!isInvalidClientId(err)) throw err; + await this.reattach(); + return await fn(); + } + } + + /** + * Re-register this client against the (already-restored) session to obtain a + * fresh daemon-assigned `clientId`. Concurrent callers coalesce onto a single + * in-flight `resume` so we never orphan extra registrations. + */ + private async reattach(): Promise { + if (this.reattaching) return this.reattaching; + // Send no clientId so the bridge issues a fresh registration rather than + // validating the stale one. Pass workspaceCwd explicitly: the daemon's + // restore path resolves the workspace key before its existing-session fast + // path, and that resolution rejects a missing/relative path. + this.reattaching = this.client + .resumeSession(this.sessionId, { workspaceCwd: this.workspaceCwd }) + .then((session) => { + // Refresh only the clientId; leave the SSE cursor and ACP state intact. + this.session.clientId = session.clientId; + }); + try { + await this.reattaching; + } finally { + this.reattaching = undefined; + } + } + async cancel(): Promise { await this.client.cancel(this.sessionId, this.clientId); } @@ -635,3 +682,17 @@ function validateLastEventId( } return lastEventId; } + +/** + * True for the daemon's `400 invalid_client_id` prompt-admission rejection + * (the stale-clientId signal a daemon restart / session reload produces). + */ +function isInvalidClientId(err: unknown): boolean { + return ( + err instanceof DaemonHttpError && + err.status === 400 && + typeof err.body === 'object' && + err.body !== null && + (err.body as { code?: unknown }).code === 'invalid_client_id' + ); +} diff --git a/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts b/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts index ccc6aa0cf4..d21af937cf 100644 --- a/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts +++ b/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts @@ -1266,3 +1266,303 @@ describe('DaemonSessionClient', () => { ); }); }); + +describe('DaemonSessionClient clientId self-heal', () => { + function invalidClientIdResponse(): Response { + return jsonResponse(400, { + code: 'invalid_client_id', + error: 'unknown client', + sessionId: 's-1', + clientId: 'client-1', + }); + } + + function newSession(client: DaemonClient): DaemonSessionClient { + return new DaemonSessionClient({ + client, + session: { + sessionId: 's-1', + workspaceCwd: '/work/a', + attached: true, + clientId: 'client-1', + }, + maxPendingPromptsPerSession: 10, + }); + } + + it('re-registers and retries once when the blocking prompt is rejected with invalid_client_id', async () => { + let promptCalls = 0; + let resumeCalls = 0; + const { fetch, calls } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/resume')) { + resumeCalls++; + return jsonResponse(200, { + sessionId: 's-1', + workspaceCwd: '/work/a', + attached: true, + clientId: 'client-2', + state: {}, + }); + } + if (req.url.endsWith('/session/s-1/prompt')) { + promptCalls++; + if (promptCalls === 1) return invalidClientIdResponse(); + return jsonResponse(200, { stopReason: 'end_turn' }); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + const session = newSession(client); + + await expect( + session.prompt({ prompt: [{ type: 'text', text: 'hi' }] }), + ).resolves.toEqual({ stopReason: 'end_turn' }); + + expect(resumeCalls).toBe(1); + expect(promptCalls).toBe(2); + // The retried prompt carries the freshly registered clientId. + const promptRequests = calls.filter((c) => + c.url.endsWith('/session/s-1/prompt'), + ); + expect(promptRequests[0]?.headers['x-qwen-client-id']).toBe('client-1'); + expect(promptRequests[1]?.headers['x-qwen-client-id']).toBe('client-2'); + expect(session.clientId).toBe('client-2'); + // resume re-registers without sending the stale clientId. + const resumeReq = calls.find((c) => c.url.endsWith('/session/s-1/resume')); + expect(resumeReq?.headers['x-qwen-client-id']).toBeUndefined(); + expect(resumeReq?.body).toBe(JSON.stringify({ cwd: '/work/a' })); + }); + + it('re-registers and retries once on the non-blocking prompt path', async () => { + let promptCalls = 0; + let resumeCalls = 0; + let eventsController: + | ReadableStreamDefaultController + | undefined; + const encoder = new TextEncoder(); + const { fetch, calls } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/resume')) { + resumeCalls++; + return jsonResponse(200, { + sessionId: 's-1', + workspaceCwd: '/work/a', + attached: true, + clientId: 'client-2', + state: {}, + }); + } + if (req.url.endsWith('/session/s-1/events')) { + return pendingSseResponse( + () => {}, + (controller) => { + eventsController = controller; + }, + ); + } + if (req.url.endsWith('/session/s-1/prompt')) { + promptCalls++; + if (promptCalls === 1) return invalidClientIdResponse(); + return jsonResponse(202, { promptId: 'p-2', lastEventId: 0 }); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + const session = newSession(client); + // Activate the SSE subscription so prompt() takes the non-blocking path. + const eventsAbort = new AbortController(); + const eventPump = (async () => { + for await (const _event of session.events({ + signal: eventsAbort.signal, + })) { + /* keep subscription active */ + } + })().catch(() => {}); + await vi.waitFor(() => { + expect(calls.filter((c) => c.url.endsWith('/events'))).toHaveLength(1); + }); + + const promptPromise = session + .prompt({ prompt: [{ type: 'text', text: 'hi' }] }) + .catch((err: unknown) => err); + // The retried prompt registers a pending entry under the new promptId. + await waitForPendingPrompt(session, 'p-2'); + eventsController?.enqueue(encoder.encode(turnCompleteFrame('p-2'))); + + await expect(promptPromise).resolves.toEqual({ stopReason: 'end_turn' }); + expect(resumeCalls).toBe(1); + expect(promptCalls).toBe(2); + const promptRequests = calls.filter((c) => + c.url.endsWith('/session/s-1/prompt'), + ); + expect(promptRequests[1]?.headers['x-qwen-client-id']).toBe('client-2'); + + eventsController?.close(); + eventsAbort.abort(); + await eventPump; + }); + + it('propagates the error when the retried prompt is also invalid_client_id (no loop)', async () => { + let promptCalls = 0; + let resumeCalls = 0; + const { fetch } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/resume')) { + resumeCalls++; + return jsonResponse(200, { + sessionId: 's-1', + workspaceCwd: '/work/a', + attached: true, + clientId: 'client-2', + state: {}, + }); + } + if (req.url.endsWith('/session/s-1/prompt')) { + promptCalls++; + return invalidClientIdResponse(); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + const session = newSession(client); + + await expect( + session.prompt({ prompt: [{ type: 'text', text: 'hi' }] }), + ).rejects.toMatchObject({ + status: 400, + body: { code: 'invalid_client_id' }, + }); + expect(resumeCalls).toBe(1); + expect(promptCalls).toBe(2); + }); + + it('does not re-register or retry on a non-invalid_client_id error', async () => { + let promptCalls = 0; + let resumeCalls = 0; + const { fetch } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/resume')) { + resumeCalls++; + return jsonResponse(200, { clientId: 'client-2' }); + } + if (req.url.endsWith('/session/s-1/prompt')) { + promptCalls++; + return jsonResponse(500, { error: 'boom' }); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + const session = newSession(client); + + await expect( + session.prompt({ prompt: [{ type: 'text', text: 'hi' }] }), + ).rejects.toThrow('POST /session/:id/prompt: boom'); + expect(promptCalls).toBe(1); + expect(resumeCalls).toBe(0); + expect(session.clientId).toBe('client-1'); + }); + + it('does not re-register or retry on 400 with a different error code', async () => { + let promptCalls = 0; + let resumeCalls = 0; + const { fetch } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/resume')) { + resumeCalls++; + return jsonResponse(200, { clientId: 'client-2' }); + } + if (req.url.endsWith('/session/s-1/prompt')) { + promptCalls++; + return jsonResponse(400, { + code: 'validation_error', + error: 'bad request', + }); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + const session = newSession(client); + + await expect( + session.prompt({ prompt: [{ type: 'text', text: 'hi' }] }), + ).rejects.toMatchObject({ + status: 400, + body: { code: 'validation_error' }, + }); + expect(promptCalls).toBe(1); + expect(resumeCalls).toBe(0); + expect(session.clientId).toBe('client-1'); + }); + + it('propagates a reattach failure and clears the in-flight guard', async () => { + let promptCalls = 0; + let resumeCalls = 0; + const { fetch } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/resume')) { + resumeCalls++; + if (resumeCalls === 1) { + return jsonResponse(404, { error: 'session gone' }); + } + return jsonResponse(200, { + sessionId: 's-1', + workspaceCwd: '/work/a', + attached: true, + clientId: 'client-2', + state: {}, + }); + } + if (req.url.endsWith('/session/s-1/prompt')) { + promptCalls++; + if (promptCalls <= 2) return invalidClientIdResponse(); + return jsonResponse(200, { stopReason: 'end_turn' }); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + const session = newSession(client); + + await expect( + session.prompt({ prompt: [{ type: 'text', text: 'hi' }] }), + ).rejects.toThrow('POST /session/:id/resume: session gone'); + expect(resumeCalls).toBe(1); + expect(session.clientId).toBe('client-1'); + + await expect( + session.prompt({ prompt: [{ type: 'text', text: 'retry' }] }), + ).resolves.toEqual({ stopReason: 'end_turn' }); + expect(resumeCalls).toBe(2); + expect(promptCalls).toBe(3); + expect(session.clientId).toBe('client-2'); + }); + + it('coalesces concurrent reattach into a single re-registration', async () => { + let promptCalls = 0; + let resumeCalls = 0; + const { fetch } = recordingFetch((req) => { + if (req.url.endsWith('/session/s-1/resume')) { + resumeCalls++; + return jsonResponse(200, { + sessionId: 's-1', + workspaceCwd: '/work/a', + attached: true, + clientId: 'client-2', + state: {}, + }); + } + if (req.url.endsWith('/session/s-1/prompt')) { + promptCalls++; + // First two concurrent prompts are rejected; retries succeed. + if (promptCalls <= 2) return invalidClientIdResponse(); + return jsonResponse(200, { stopReason: 'end_turn' }); + } + return jsonResponse(500, { error: `unexpected ${req.url}` }); + }); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + const session = newSession(client); + + const [a, b] = await Promise.all([ + session.prompt({ prompt: [{ type: 'text', text: 'a' }] }), + session.prompt({ prompt: [{ type: 'text', text: 'b' }] }), + ]); + expect(a).toEqual({ stopReason: 'end_turn' }); + expect(b).toEqual({ stopReason: 'end_turn' }); + expect(resumeCalls).toBe(1); + expect(session.clientId).toBe('client-2'); + }); +});