diff --git a/docs/deployment.md b/docs/deployment.md index 6eeaa6a6..488be1af 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -113,7 +113,10 @@ invalid session id is `LOUSHO_SESSION_ID_INVALID`). Every error body is JSON `409` with `code: 'LOUSHO_SESSION_AWAITING_APPROVAL'`, `approvalId` and `approvalIds` (every approval still to decide): decide them through the approvals route, which streams the paused turn's continuation, then send the -new turn. With an auth list, a session belongs to the first principal that sends +new turn. A turn sent to a session that another process or replica is running a +turn of (it holds the session's run lease) is `409` with +`code: 'LOUSHO_SESSION_BUSY'` and a `Retry-After: 2` header, answered before +any stream starts; resend after the wait. With an auth list, a session belongs to the first principal that sends it a turn, and a stored session whose owner the server no longer remembers (after a restart) is refused to everyone until `authorizeSession` says whose it is; with only `LOUSHO_API_TOKEN` there is one caller and no owner check @@ -149,7 +152,7 @@ same key: | The turn of that key in that session | The resend gets | | ------------------------------------ | --------------- | | finished (it is in the transcript) | `200` SSE with one `run.done` carrying that turn's final reply (`text`, `finishReason`), and the header `Idempotent-Replayed: true`. Nothing runs. Read `GET /chat/:sessionId` for the whole turn. | -| still running on this server process | `409` (`LOUSHO_SESSION_BUSY`). Retry later and get the replay. | +| still running, on this server process or on another replica that shares the store | `409` (`LOUSHO_SESSION_BUSY`) with `Retry-After: 2`. Retry after the wait and get the replay. | | interrupted by a crash (a checkpointed turn left `in-progress`) | the turn resumed from its last checkpoint (`session.resume()`: finished steps and tool calls are not run again), streamed as SSE. | | none | a new turn, as without the header; its user message keeps the key (`metadata.idempotencyKey`). | @@ -160,8 +163,9 @@ checkpoint until it finishes), so a replay works after a restart and on another replica that shares the store; `session.pending()` reports an interrupted turn's key as `idempotencyKey`. A key is looked up only in its own session, and only while its turn is in the transcript (a compaction that summarizes it -away forgets it). The `409` for a running turn holds within one process; two -replicas that receive the same key at the same instant can both run it. +away forgets it). Across replicas the `409` rests on the session's run lease, so it needs a +store that supports leases; two replicas that receive the same key at the same +instant on a store without them can both run it. ```bash curl -N http://127.0.0.1:3000/chat -H 'Content-Type: application/json' \ diff --git a/docs/errors.md b/docs/errors.md index 375db980..e5d824a3 100644 --- a/docs/errors.md +++ b/docs/errors.md @@ -634,9 +634,9 @@ kept: the turn up to their results is added after the other writer's turn. Also checkpointed run (or session turn) is running in another process, replica or store object right now, which holds its run lease (see [Durable execution](./durable-execution.md#run-leases)); nothing ran here, and a resolved -approval stays pending. Over HTTP, a `409` from `POST /chat`: a request with the same -`Idempotency-Key` is still running its turn of that session on this server (see -[Resending a turn](./deployment.md#resending-a-turn)). +approval stays pending. Over HTTP, a `409` with `Retry-After` from `POST /chat`: another process or replica holds +the session's lease, or a request with the same `Idempotency-Key` is still running its turn of +that session (see [Resending a turn](./deployment.md#resending-a-turn)). **Fix:** await the turn's `send()` (or abort it), then call again. After a `send()` that failed this way, send again: the session reads the current transcript at the start of every call. A diff --git a/llms-full.txt b/llms-full.txt index 4263f97f..3225539a 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -7162,7 +7162,10 @@ invalid session id is `LOUSHO_SESSION_ID_INVALID`). Every error body is JSON `409` with `code: 'LOUSHO_SESSION_AWAITING_APPROVAL'`, `approvalId` and `approvalIds` (every approval still to decide): decide them through the approvals route, which streams the paused turn's continuation, then send the -new turn. With an auth list, a session belongs to the first principal that sends +new turn. A turn sent to a session that another process or replica is running a +turn of (it holds the session's run lease) is `409` with +`code: 'LOUSHO_SESSION_BUSY'` and a `Retry-After: 2` header, answered before +any stream starts; resend after the wait. With an auth list, a session belongs to the first principal that sends it a turn, and a stored session whose owner the server no longer remembers (after a restart) is refused to everyone until `authorizeSession` says whose it is; with only `LOUSHO_API_TOKEN` there is one caller and no owner check @@ -7198,7 +7201,7 @@ same key: | The turn of that key in that session | The resend gets | | ------------------------------------ | --------------- | | finished (it is in the transcript) | `200` SSE with one `run.done` carrying that turn's final reply (`text`, `finishReason`), and the header `Idempotent-Replayed: true`. Nothing runs. Read `GET /chat/:sessionId` for the whole turn. | -| still running on this server process | `409` (`LOUSHO_SESSION_BUSY`). Retry later and get the replay. | +| still running, on this server process or on another replica that shares the store | `409` (`LOUSHO_SESSION_BUSY`) with `Retry-After: 2`. Retry after the wait and get the replay. | | interrupted by a crash (a checkpointed turn left `in-progress`) | the turn resumed from its last checkpoint (`session.resume()`: finished steps and tool calls are not run again), streamed as SSE. | | none | a new turn, as without the header; its user message keeps the key (`metadata.idempotencyKey`). | @@ -7209,8 +7212,9 @@ checkpoint until it finishes), so a replay works after a restart and on another replica that shares the store; `session.pending()` reports an interrupted turn's key as `idempotencyKey`. A key is looked up only in its own session, and only while its turn is in the transcript (a compaction that summarizes it -away forgets it). The `409` for a running turn holds within one process; two -replicas that receive the same key at the same instant can both run it. +away forgets it). Across replicas the `409` rests on the session's run lease, so it needs a +store that supports leases; two replicas that receive the same key at the same +instant on a store without them can both run it. ```bash curl -N http://127.0.0.1:3000/chat -H 'Content-Type: application/json' \ @@ -8852,9 +8856,9 @@ kept: the turn up to their results is added after the other writer's turn. Also checkpointed run (or session turn) is running in another process, replica or store object right now, which holds its run lease (see [Durable execution](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/durable-execution.md#run-leases)); nothing ran here, and a resolved -approval stays pending. Over HTTP, a `409` from `POST /chat`: a request with the same -`Idempotency-Key` is still running its turn of that session on this server (see -[Resending a turn](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/deployment.md#resending-a-turn)). +approval stays pending. Over HTTP, a `409` with `Retry-After` from `POST /chat`: another process or replica holds +the session's lease, or a request with the same `Idempotency-Key` is still running its turn of +that session (see [Resending a turn](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/deployment.md#resending-a-turn)). **Fix:** await the turn's `send()` (or abort it), then call again. After a `send()` that failed this way, send again: the session reads the current transcript at the start of every call. A diff --git a/src/channels/channelCore.ts b/src/channels/channelCore.ts index a17aee47..cf81eeac 100644 --- a/src/channels/channelCore.ts +++ b/src/channels/channelCore.ts @@ -77,7 +77,7 @@ const FAILED_TEXT = 'Sorry, that request failed.'; * replica is running (`Retry-After` of the `409`). The other run holds the lease * until its turn ends; a short wait lets a typical turn finish without hammering it. */ -const SESSION_BUSY_RETRY_AFTER_S = 2; +export const SESSION_BUSY_RETRY_AFTER_S = 2; /** Whether `error` is a run another process, replica or store object holds the lease of. */ function isSessionBusy(error: unknown): boolean { diff --git a/src/server/chatBusy.test.ts b/src/server/chatBusy.test.ts new file mode 100644 index 00000000..eefe1cf4 --- /dev/null +++ b/src/server/chatBusy.test.ts @@ -0,0 +1,94 @@ +/** + * `POST /chat` for a session another process or replica is running a turn of + * (it holds the session's run lease): `409` with `Retry-After` and a coded body, + * the same answer a channel gives, before any stream starts. + */ +import { afterEach, describe, expect, it } from 'vitest'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { createAgent } from '../createAgent'; +import { mockModel } from '../testing'; +import { fileStore } from '../storage/fileStore'; +import { createRouteHandler } from './routeHandler'; + +const dirs: string[] = []; +afterEach(() => { + while (dirs.length) rmSync(dirs.pop()!, { recursive: true, force: true }); +}); + +const post = (body: unknown, key?: string) => + new Request('http://app.test/api/chat', { + method: 'POST', + headers: { 'content-type': 'application/json', ...(key !== undefined && { 'idempotency-key': key }) }, + body: JSON.stringify(body), + }); + +/** Two "replicas" over one directory: `a` holds a turn open on session `s1` until `release()`. */ +function replicas() { + const dir = mkdtempSync(join(tmpdir(), 'lousho-chat-busy-')); + dirs.push(dir); + let release!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + const a = createRouteHandler( + createAgent({ + provider: mockModel([async () => (await gate, 'A done')]), + store: fileStore(dir), + }), + { basePath: '/api' } + ); + const b = createRouteHandler(createAgent({ provider: mockModel(['B hello'], { onExhausted: 'repeat-last' }), store: fileStore(dir) }), { basePath: '/api' }); + return { a: a.handler, b: b.handler, release }; +} + +describe('POST /chat when another replica runs the session', () => { + it('answers 409 + Retry-After + a coded body instead of a 200 stream', async () => { + const { a, b, release } = replicas(); + const first = await a(post({ sessionId: 's1', input: 'hi' })); + expect(first.status).toBe(200); + const firstText = first.text(); + + const busy = await b(post({ sessionId: 's1', input: 'hello' })); + expect(busy.status).toBe(409); + expect(busy.headers.get('retry-after')).toBe('2'); + expect(busy.headers.get('content-type')).toContain('application/json'); + expect(await busy.json()).toMatchObject({ code: 'LOUSHO_SESSION_BUSY' }); + + release(); + expect(await firstText).toContain('A done'); + // Once the turn is over the other replica takes turns again. + expect((await b(post({ sessionId: 's1', input: 'hello' }))).status).toBe(200); + }); + + it('answers an Idempotency-Key resend the same way while the original turn still runs elsewhere', async () => { + const { a, b, release } = replicas(); + const first = await a(post({ sessionId: 's2', input: 'hi' }, 'req-1')); + const firstText = first.text(); + + const resend = await b(post({ sessionId: 's2', input: 'hi' }, 'req-1')); + expect(resend.status).toBe(409); + expect(resend.headers.get('retry-after')).toBe('2'); + expect(await resend.json()).toMatchObject({ code: 'LOUSHO_SESSION_BUSY' }); + + release(); + await firstText; + // The client retries as told and gets the finished turn replayed. + const retry = await b(post({ sessionId: 's2', input: 'hi' }, 'req-1')); + expect(retry.status).toBe(200); + expect(retry.headers.get('idempotent-replayed')).toBe('true'); + expect(await retry.text()).toContain('A done'); + }); + + it('a resend within the same process gets Retry-After too', async () => { + let release!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + const { handler } = createRouteHandler(createAgent({ provider: mockModel([async () => (await gate, 'Done.')]) }), { basePath: '/api' }); + const first = await handler(post({ sessionId: 's3', input: 'hi' }, 'dup')); + const text = first.text(); + const double = await handler(post({ sessionId: 's3', input: 'hi' }, 'dup')); + expect(double.status).toBe(409); + expect(double.headers.get('retry-after')).toBe('2'); + release(); + await text; + }); +}); diff --git a/src/server/fetchRoutes.ts b/src/server/fetchRoutes.ts index d934ddbe..86d7fd03 100644 --- a/src/server/fetchRoutes.ts +++ b/src/server/fetchRoutes.ts @@ -26,6 +26,7 @@ import { idempotencyKeyOf, type AgentSession } from '../session/AgentSession'; import type { Message } from '../providers/llm'; import { toMessages } from '../providers/content'; import { SDKError, SessionAwaitingApprovalError } from '../execution/errors'; +import { SESSION_BUSY_RETRY_AFTER_S } from '../channels/channelCore'; import { errorEvents, replayEvents } from '../cli/devEvents'; import type { AuthFn, Principal } from '../auth/types'; import { routeAuth } from '../auth/routeAuth'; @@ -508,6 +509,54 @@ async function awaitingApproval(session: AgentSession): Promise | { failure: unknown }): boolean { + const error = 'failure' in first ? first.failure : first.done || first.value.type !== 'error' ? undefined : first.value.error; + return typeof error === 'object' && error !== null && (error as { code?: unknown }).code === 'LOUSHO_SESSION_BUSY'; +} + +/** + * {@link sseResponse} for a turn, except that a turn refused because another process + * or replica holds the session's lease (`LOUSHO_SESSION_BUSY`, which fails the turn + * before it runs anything) is answered with a `409` and `Retry-After` instead of a + * `200` stream whose only events are the error. The turn's first event decides, so + * the response starts once the turn has; a failure later in the turn is still in-band. + */ +async function turnResponse(request: Request, ctx: ChatRoutesContext, events: (signal: AbortSignal) => AsyncIterable): Promise { + const controller = new AbortController(); + if (disconnectAborts(ctx)) request.signal.addEventListener('abort', () => controller.abort()); + const iterator = events(controller.signal)[Symbol.asyncIterator](); + let head: IteratorResult | { failure: unknown }; + try { + head = await iterator.next(); + } catch (failure) { + head = { failure }; + } + if (isBusyRefusal(head)) { + await iterator.return?.(); + return sessionBusy('This session is running a turn elsewhere; retry shortly.'); + } + let unread: typeof head | undefined = head; + const replay: AsyncIterable = { + [Symbol.asyncIterator]: () => ({ + next: async () => { + const first = unread; + unread = undefined; + if (!first) return iterator.next(); + if ('failure' in first) throw first.failure; + return first; + }, + return: async () => (await iterator.return?.()) ?? { done: true, value: undefined }, + }), + }; + return sseResponse(request, ctx, () => replay, controller); +} + /** What an `Idempotency-Key` header may hold: 1-255 visible ASCII characters (a UUID, a ULID, a hash). */ const IDEMPOTENCY_KEY = /^[\x21-\x7e]{1,255}$/; @@ -560,7 +609,7 @@ async function idempotentTurn(request: Request, ctx: ChatRoutesContext, principa const keys = keysOf(ctx.agent()); const claim = `${session.id} ${key}`; if (keys.has(claim)) { - return jsonResponse(409, { error: `A request with this Idempotency-Key is still running its turn of session '${session.id}'`, code: 'LOUSHO_SESSION_BUSY' }); + return sessionBusy(`A request with this Idempotency-Key is still running its turn of session '${session.id}'`); } keys.add(claim); const release = () => void keys.delete(claim); @@ -571,7 +620,7 @@ async function idempotentTurn(request: Request, ctx: ChatRoutesContext, principa return (await awaitingApproval(session))!; } if (pending?.status === 'in-progress' && pending.idempotencyKey === key) { - return sseResponse(request, ctx, (signal) => releasing(claimOnCommit(ctx, principal, session, resumedTurn(session, signal)), release)); + return turnResponse(request, ctx, (signal) => releasing(claimOnCommit(ctx, principal, session, resumedTurn(session, signal)), release)); } const turn = keyedTurn(await session.load(), key); if (turn) { @@ -584,7 +633,7 @@ async function idempotentTurn(request: Request, ctx: ChatRoutesContext, principa response.headers.set('Idempotent-Replayed', 'true'); return response; } - return sseResponse(request, ctx, (signal) => releasing(claimOnCommit(ctx, principal, session, session.stream(keyedInput(input, key), { signal, principal })), release)); + return turnResponse(request, ctx, (signal) => releasing(claimOnCommit(ctx, principal, session, session.stream(keyedInput(input, key), { signal, principal })), release)); } catch (error) { release(); throw error; @@ -679,7 +728,7 @@ async function sessionTurn(request: Request, ctx: ChatRoutesContext, principal: const forbidden = await sessionForbidden(ctx, principal, sessionId, 'chat', session); if (forbidden) return forbidden; if (key !== null) return idempotentTurn(request, ctx, principal, session, input, key); - return (await awaitingApproval(session)) ?? sseResponse(request, ctx, (signal) => claimOnCommit(ctx, principal, session, session.stream(input, { signal, principal }))); + return (await awaitingApproval(session)) ?? turnResponse(request, ctx, (signal) => claimOnCommit(ctx, principal, session, session.stream(input, { signal, principal }))); } const runChat: RouteHandler = async (request, ctx, _params, principal) => {