V7: Return 409 with Retry-After when a channel's session is busy

docs/channels.md

@@ -108,7 +108,14 @@ acknowledge the request. An error thrown by `verify` or `parse` answers 400
 for a `SyntaxError` (bad JSON) and 500 otherwise; a body over 1MB gets 413. A
 500 body is only `{ "error": "The request failed. The server log has the
 details.", "code"? }`, as on the session routes: the error itself goes to
-`onError` (stage `'parse'`), never to the caller.
+`onError` (stage `'parse'`), never to the caller. A turn that fails before the
+channel acknowledged the request goes to `onError` with stage `'turn'`; when
+another process or replica is running that conversation's turn
+([`LOUSHO_SESSION_BUSY`](errors.md#lousho_session_busy)) the answer is
+`409 { "error", "code": "LOUSHO_SESSION_BUSY" }` with a `Retry-After` header,
+so the sender can resend shortly. A channel that acknowledges in `parse` (Slack,
+Discord, Telegram, Teams, GitHub) keeps its own answer and tells the user the
+request failed in the conversation.
 
 ```ts
 import { createAgent, defineChannel, mountChannels } from '@lousho/build-ai-agent';

llms-full.txt

@@ -4096,7 +4096,14 @@ acknowledge the request. An error thrown by `verify` or `parse` answers 400
 for a `SyntaxError` (bad JSON) and 500 otherwise; a body over 1MB gets 413. A
 500 body is only `{ "error": "The request failed. The server log has the
 details.", "code"? }`, as on the session routes: the error itself goes to
-`onError` (stage `'parse'`), never to the caller.
+`onError` (stage `'parse'`), never to the caller. A turn that fails before the
+channel acknowledged the request goes to `onError` with stage `'turn'`; when
+another process or replica is running that conversation's turn
+([`LOUSHO_SESSION_BUSY`](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/errors.md#lousho_session_busy)) the answer is
+`409 { "error", "code": "LOUSHO_SESSION_BUSY" }` with a `Retry-After` header,
+so the sender can resend shortly. A channel that acknowledges in `parse` (Slack,
+Discord, Telegram, Teams, GitHub) keeps its own answer and tells the user the
+request failed in the conversation.
 
 ```ts
 import { createAgent, defineChannel, mountChannels } from '@lousho/build-ai-agent';

src/channels/channelCore.busy.test.ts

@@ -0,0 +1,89 @@
+/**
+ * A channel turn on a session another process or replica is running (its run
+ * lease is held: `LOUSHO_SESSION_BUSY`) is a `409` with `Retry-After`, not a
+ * `500`, and is reported as a failed turn, not a failed parse.
+ */
+import { describe, it, expect, vi } from 'vitest';
+import type * as http from 'node:http';
+import { Readable } from 'node:stream';
+import type { SimpleAgent } from '../createAgent';
+import { SDKError } from '../execution/errors';
+import { httpChannel } from './httpChannel';
+import { mountChannels } from './mountChannels';
+import { mountFetchChannels } from './fetchChannels';
+import type { Channel, ChannelErrorContext } from './defineChannel';
+
+/** An agent whose every turn fails as a run another replica holds the lease of. */
+function busyAgent(): Pick<SimpleAgent, 'session' | 'approvals'> {
+  const busy = new SDKError("The run checkpointed as 'http_ali' is running elsewhere", 'LOUSHO_SESSION_BUSY');
+  const session = () => ({
+    stream: () => {
+      throw busy;
+    },
+  });
+  return { session, approvals: { list: async () => [] } } as unknown as Pick<SimpleAgent, 'session' | 'approvals'>;
+}
+
+/** Drives a `mountChannels` handler with a fake `node:http` request and response pair. */
+async function postNode(channels: readonly Channel[], onError: (error: unknown, ctx: ChannelErrorContext) => void) {
+  const handler = mountChannels(busyAgent(), channels, { onError });
+  const req = Object.assign(Readable.from([Buffer.from(JSON.stringify({ sessionKey: 'ali', input: 'hi' }))]), {
+    method: 'POST',
+    url: '/channels/http',
+    headers: { 'content-type': 'application/json' },
+  });
+  const res: { status: number; headers: Record<string, string>; body: string } = { status: 0, headers: {}, body: '' };
+  const fakeRes = {
+    writeHead: (status: number, headers: Record<string, string> = {}) => ((res.status = status), (res.headers = headers), fakeRes),
+    end: (text?: string) => ((res.body = text ?? ''), fakeRes),
+  };
+  await handler(req as unknown as http.IncomingMessage, fakeRes as unknown as http.ServerResponse);
+  return res;
+}
+
+describe('channel turn on a busy session (LOUSHO_SESSION_BUSY)', () => {
+  it('mountChannels answers 409 with Retry-After and the coded error, and reports a failed turn', async () => {
+    const onError = vi.fn();
+    const res = await postNode([httpChannel()], onError);
+    expect(res.status).toBe(409);
+    expect(res.headers['Retry-After']).toMatch(/^\d+$/);
+    expect(JSON.parse(res.body)).toMatchObject({ code: 'LOUSHO_SESSION_BUSY' });
+    expect(onError).toHaveBeenCalledTimes(1);
+    expect(onError.mock.calls[0][1]).toMatchObject({ channel: 'http', stage: 'turn' });
+  });
+
+  it('mountFetchChannels answers 409 with Retry-After and the coded error, and reports a failed turn', async () => {
+    const onError = vi.fn();
+    const handler = mountFetchChannels(busyAgent(), [httpChannel()], { onError });
+    const request = new Request('http://worker/channels/http', {
+      method: 'POST',
+      headers: { 'content-type': 'application/json' },
+      body: JSON.stringify({ sessionKey: 'ali', input: 'hi' }),
+    });
+    const response = await handler(request);
+    expect(response?.status).toBe(409);
+    expect(response?.headers.get('Retry-After')).toMatch(/^\d+$/);
+    expect(await response?.json()).toMatchObject({ code: 'LOUSHO_SESSION_BUSY' });
+    expect(onError).toHaveBeenCalledTimes(1);
+    expect(onError.mock.calls[0][1]).toMatchObject({ channel: 'http', stage: 'turn' });
+  });
+
+  it('logs any other turn failure before the ack as a failed turn (still a 500 that hides the message)', async () => {
+    const onError = vi.fn();
+    const agent = {
+      session: () => ({
+        stream: () => {
+          throw new Error('connect ECONNREFUSED postgres://admin:hunter2@10.0.0.1:5432/prod');
+        },
+      }),
+      approvals: { list: async () => [] },
+    } as unknown as Pick<SimpleAgent, 'session' | 'approvals'>;
+    const handler = mountFetchChannels(agent, [httpChannel()], { onError });
+    const response = await handler(
+      new Request('http://worker/channels/http', { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ sessionKey: 'ali', input: 'hi' }) })
+    );
+    expect(response?.status).toBe(500);
+    expect(JSON.stringify(await response?.json())).not.toContain('hunter2');
+    expect(onError.mock.calls[0][1]).toMatchObject({ stage: 'turn' });
+  });
+});

src/channels/channelCore.ts

@@ -23,7 +23,7 @@ import {
   type ChannelRespond,
   type ChannelUser,
 } from './defineChannel';
-import { reportChannelError } from './channelSupport';
+import { channelFailureBody, reportChannelError } from './channelSupport';
 import type { Principal } from '../auth/types';
 import { describeApproval, type ApprovalStore, type PendingApproval, type ResolvedApproval } from '../execution/ApprovalGate';
 import { SDKError, SessionAwaitingApprovalError } from '../execution/errors';
@@ -44,14 +44,23 @@ export interface ChannelCoreOptions {
   onDecision?(event: { decision: ChannelApprovalDecision; approver: NonNullable<ChannelDecision['approver']>; sessionId: string; channel: string }): void | Promise<void>;
 }
 
+/**
+ * How a host answers a channel request: a channel's {@link ChannelRespond}, plus
+ * response headers the core may set (`Retry-After` on a busy session's `409`).
+ */
+export type ChannelCoreRespond = (status: number, body: unknown, headers?: Record<string, string>) => void;
+
 /** What a host's mount returns: the channel logic, request/response plumbing left to the host. */
 export interface ChannelCore {
   /**
    * Runs `channel`'s verify -> parse -> turn (or a decision on `approvalId`),
-   * answering through `respond`. Throws before the surface was acknowledged;
-   * failures after that are reported to `onError` (and a failure text replied).
+   * answering through `respond`. Throws before the surface was acknowledged,
+   * except a turn that fails then: it is reported to `onError` (stage `turn`) and
+   * answered here - `409` with `Retry-After` when another process or replica runs the
+   * session (`LOUSHO_SESSION_BUSY`), else a `500`. Failures after the ack are reported
+   * to `onError` (and a failure text replied).
    */
-  handle(channel: Channel, approvalId: string | undefined, req: ChannelRequest, respond: ChannelRespond): Promise<void>;
+  handle(channel: Channel, approvalId: string | undefined, req: ChannelRequest, respond: ChannelCoreRespond): Promise<void>;
   /**
    * Decides a pause a channel turn stopped on (e.g. from a surface's button
    * callback) and delivers the continuation through the same channel's
@@ -63,6 +72,18 @@ export interface ChannelCore {
 
 const FAILED_TEXT = 'Sorry, that request failed.';
 
+/**
+ * Seconds a client waits before resending a turn to a session another process or
+ * 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;
+
+/** Whether `error` is a run another process, replica or store object holds the lease of. */
+function isSessionBusy(error: unknown): boolean {
+  return error instanceof SDKError && error.code === 'LOUSHO_SESSION_BUSY';
+}
+
 /** The reply to a decision that leaves other calls of its step undecided. */
 function stillWaiting(approvals: readonly PendingApproval[]): string {
   const left = approvals.map((approval) => `${approval.toolName} (approval id: ${approval.id})`).join(', ');
@@ -258,9 +279,22 @@ export function channelCore(agent: Pick<SimpleAgent, 'session' | 'approvals'>, c
     }
   }
 
-  function runTurn(turn: PausedTurn, respond: ChannelRespond): Promise<void> {
+  /**
+   * A turn that failed before the surface was answered: reported as a failed turn and
+   * answered here (`409` + `Retry-After` for a session busy elsewhere, else a `500` that
+   * keeps only the error's code), so the host does not report it as a failed parse.
+   */
+  async function failTurn(turn: PausedTurn, error: unknown, respond: ChannelCoreRespond): Promise<void> {
+    if (answered.has(respond)) throw error;
+    await reportChannelError(turn.channel.onError ?? options.onError, error, { channel: turn.channel.name, stage: 'turn', sessionId: turn.sessionId });
+    if (!isSessionBusy(error)) return respond(500, channelFailureBody(500, error));
+    const body = { error: 'This conversation is running a turn elsewhere; retry shortly.', code: 'LOUSHO_SESSION_BUSY' };
+    respond(409, body, { 'Retry-After': String(SESSION_BUSY_RETRY_AFTER_S) });
+  }
+
+  function runTurn(turn: PausedTurn, respond: ChannelCoreRespond): Promise<void> {
     const { channel, inbound, sessionId } = turn;
-    return guard(turn, 'turn', respond, async () => {
+    const done = guard(turn, 'turn', respond, async () => {
       const run = agent.session({ id: sessionId, store }).stream(inbound.input, { principal: inbound.principal, metadata: inbound.metadata });
       const events: AgentEvent[] = [];
       let text = '';
@@ -272,6 +306,7 @@ export function channelCore(agent: Pick<SimpleAgent, 'session' | 'approvals'>, c
       }
       await finish(turn, await run.result, events, respond);
     });
+    return done.catch((error: unknown) => failTurn(turn, error, respond));
   }
 
   /**
@@ -391,11 +426,12 @@ export function channelCore(agent: Pick<SimpleAgent, 'session' | 'approvals'>, c
     await claiming(decision.id, () => continueTurn(turn, decision, respond));
   }
 
-  async function handle(channel: Channel, approvalId: string | undefined, req: ChannelRequest, respond: ChannelRespond): Promise<void> {
+  async function handle(channel: Channel, approvalId: string | undefined, req: ChannelRequest, respond: ChannelCoreRespond): Promise<void> {
     /** `respond`, marked in `answered` on its first call, so `guard` can tell a pre-ack failure from a post-ack one. */
-    const tracked: ChannelRespond = (status, body) => {
+    const tracked: ChannelCoreRespond = (status, body, headers) => {
       answered.add(tracked);
-      respond(status, body);
+      if (headers) respond(status, body, headers);
+      else respond(status, body);
     };
     if (crossSiteToOpenChannel(channel, req)) {
       const error = `Cross-site request refused: channel '${channel.name}' has no verify. To accept a page on another site, add a verify that authenticates its requests (e.g. a bearer token).`;

src/channels/fetchChannels.ts

@@ -20,7 +20,7 @@ import {
   type ChannelRespond,
 } from './defineChannel';
 import { channelFailureBody, reportChannelError } from './channelSupport';
-import { channelCore } from './channelCore';
+import { channelCore, type ChannelCoreRespond } from './channelCore';
 import { decodeParam } from '../server/requestGuard';
 import { SDKError } from '../execution/errors';
 
@@ -88,8 +88,8 @@ async function readRequestBytes(request: Request): Promise<Uint8Array> {
   return body;
 }
 
-function jsonResponse(status: number, value: unknown): Response {
-  return new Response(JSON.stringify(value), { status, headers: { 'Content-Type': 'application/json' } });
+function jsonResponse(status: number, value: unknown, headers?: Record<string, string>): Response {
+  return new Response(JSON.stringify(value), { status, headers: { 'Content-Type': 'application/json', ...headers } });
 }
 
 /** The error status for a failed handler: 413 over the size cap, 400 for bad JSON, else 500 (its message reported, not sent). */
@@ -101,7 +101,7 @@ async function failureResponse(error: unknown, report: (error: unknown) => Promi
 
 /** The answer a channel's first `respond` call wrote, once `acknowledged` settles. */
 interface PendingAnswer {
-  current?: { status: number; body: unknown };
+  current?: { status: number; body: unknown; headers?: Record<string, string> };
 }
 
 /**
@@ -129,18 +129,18 @@ async function settledResponse(
       warnHeld();
       await tail;
     }
-    const answered = answer.current as { status: number; body: unknown };
-    return jsonResponse(answered.status, answered.body);
+    const answered = answer.current as NonNullable<PendingAnswer['current']>;
+    return jsonResponse(answered.status, answered.body, answered.headers);
   }
   if (settled !== 'done') {
     // `handle` threw: acknowledged already -> report, else a JSON failure.
     if (answer.current !== undefined) {
       await report(settled.error);
-      return jsonResponse(answer.current.status, answer.current.body);
+      return jsonResponse(answer.current.status, answer.current.body, answer.current.headers);
     }
     return failureResponse(settled.error, report);
   }
-  return jsonResponse(answer.current?.status ?? 200, answer.current?.body ?? { ok: true });
+  return jsonResponse(answer.current?.status ?? 200, answer.current?.body ?? { ok: true }, answer.current?.headers);
 }
 
 /**
@@ -194,9 +194,9 @@ export function mountFetchChannels(
     const answer: PendingAnswer = {};
     let acknowledge!: () => void;
     const acknowledged = new Promise<void>((resolve) => (acknowledge = resolve));
-    const respond: ChannelRespond = (status, body) => {
+    const respond: ChannelCoreRespond = (status, body, headers) => {
       if (answer.current !== undefined) return;
-      answer.current = { status, body };
+      answer.current = { status, body, ...(headers && { headers }) };
       acknowledge();
     };
     let turn: Promise<void>;

src/channels/mountChannels.ts

@@ -13,7 +13,7 @@ import type * as http from 'node:http';
 import type { SimpleAgent } from '../createAgent';
 import type { SessionStore } from '../session/sessionStore';
 import type { SessionStores } from '../session/AgentSession';
-import { readRawBody, sendJson } from '../server/chatRoutes';
+import { readRawBody } from '../server/chatRoutes';
 import { PayloadTooLargeError } from '../server/fetchRoutes';
 import {
   toChannelRequest,
@@ -24,7 +24,7 @@ import {
   type ChannelRespond,
 } from './defineChannel';
 import { channelFailureBody, reportChannelError } from './channelSupport';
-import { channelCore, type ChannelCore } from './channelCore';
+import { channelCore, type ChannelCore, type ChannelCoreRespond } from './channelCore';
 import { decodeParam } from '../server/requestGuard';
 import { SDKError } from '../execution/errors';
 
@@ -89,8 +89,11 @@ export function mountChannels(
     if (!match) return false;
     const { channel, approvalId } = match;
     let responded = false;
-    const respond: ChannelRespond = (status, body) => {
-      if (!responded) sendJson(res, status, body);
+    const respond: ChannelCoreRespond = (status, body, headers) => {
+      if (!responded) {
+        res.writeHead(status, { 'Content-Type': 'application/json', ...headers });
+        res.end(JSON.stringify(body));
+      }
       responded = true;
     };
     try {