V8: Return 409 with Retry-After from /chat when the session is busy

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' \

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

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

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 {

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<void>((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<void>((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;
+  });
+});

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<Response | undef
   });
 }
 
+/** The `409` for a session another process or replica (or an earlier request) is running a turn of. */
+function sessionBusy(error: string): Response {
+  return jsonResponse(409, { error, code: 'LOUSHO_SESSION_BUSY' }, { 'Retry-After': String(SESSION_BUSY_RETRY_AFTER_S) });
+}
+
+/** Whether a turn's first event, or the error that ended it, is a refusal because another process holds the session's lease. */
+function isBusyRefusal(first: IteratorResult<AgentEvent> | { 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<AgentEvent>): Promise<Response> {
+  const controller = new AbortController();
+  if (disconnectAborts(ctx)) request.signal.addEventListener('abort', () => controller.abort());
+  const iterator = events(controller.signal)[Symbol.asyncIterator]();
+  let head: IteratorResult<AgentEvent> | { 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<AgentEvent> = {
+    [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) => {