mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-07-09 17:19:02 +00:00
fix(sdk): self-heal stale clientId on invalid_client_id prompts (#5797)
* 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 <qwen-coder@alibabacloud.com> * test(sdk): Cover clientId self-heal review cases Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com> --------- Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
This commit is contained in:
jinye
GitHub
parent
ebf29f1187
commit
06549daa9b
5 changed files with 644 additions and 11 deletions
|
|
@ -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_<uuid>`. 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<void>` — single-flight
|
||||
|
||||
```ts
|
||||
private reattaching?: Promise<void>;
|
||||
|
||||
private async reattach(): Promise<void> {
|
||||
// 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<T>(fn): Promise<T>`
|
||||
|
||||
```ts
|
||||
private async withClientIdSelfHeal<T>(fn: () => Promise<T>): Promise<T> {
|
||||
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.
|
||||
|
|
@ -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);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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 });
|
||||
|
|
|
|||
|
|
@ -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<void>;
|
||||
private readonly promptLimit: number;
|
||||
private readonly _pendingPrompts = new Map<
|
||||
string,
|
||||
|
|
@ -245,11 +248,8 @@ export class DaemonSessionClient {
|
|||
): Promise<PromptResult> {
|
||||
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<T>(fn: () => Promise<T>): Promise<T> {
|
||||
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<void> {
|
||||
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<void> {
|
||||
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'
|
||||
);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<Uint8Array>
|
||||
| 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');
|
||||
});
|
||||
});
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue