From e29d0f4d828367afbcf47be885f76053d196a869 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Mon, 19 Oct 2026 22:08:38 +0300 Subject: [PATCH 1/2] fix(channels): answer a busy session with 409 + Retry-After and report it as a failed turn A channel turn on a session another process or replica runs (its run lease held: LOUSHO_SESSION_BUSY) was a 500 reported to onError with stage 'parse'. channelCore now answers a turn that fails before the ack itself: a 409 with Retry-After and the coded error for a busy session, else the usual 500; both are reported with stage 'turn'. The hosts forward response headers. Co-Authored-By: Claude Opus 5.5 --- docs/channels.md | 9 ++- src/channels/channelCore.busy.test.ts | 89 +++++++++++++++++++++++++++ src/channels/channelCore.ts | 54 +++++++++++++--- src/channels/fetchChannels.ts | 20 +++--- src/channels/mountChannels.ts | 11 ++-- 5 files changed, 159 insertions(+), 24 deletions(-) create mode 100644 src/channels/channelCore.busy.test.ts diff --git a/docs/channels.md b/docs/channels.md index f1fd2217d..d614d873a 100644 --- a/docs/channels.md +++ b/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'; diff --git a/src/channels/channelCore.busy.test.ts b/src/channels/channelCore.busy.test.ts new file mode 100644 index 000000000..31f1fd9e6 --- /dev/null +++ b/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 { + 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; +} + +/** 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; body: string } = { status: 0, headers: {}, body: '' }; + const fakeRes = { + writeHead: (status: number, headers: Record = {}) => ((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; + 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' }); + }); +}); diff --git a/src/channels/channelCore.ts b/src/channels/channelCore.ts index cb54aed22..a17aee47f 100644 --- a/src/channels/channelCore.ts +++ b/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; sessionId: string; channel: string }): void | Promise; } +/** + * 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) => 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; + handle(channel: Channel, approvalId: string | undefined, req: ChannelRequest, respond: ChannelCoreRespond): Promise; /** * 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, c } } - function runTurn(turn: PausedTurn, respond: ChannelRespond): Promise { + /** + * 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 { + 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 { 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, 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, c await claiming(decision.id, () => continueTurn(turn, decision, respond)); } - async function handle(channel: Channel, approvalId: string | undefined, req: ChannelRequest, respond: ChannelRespond): Promise { + async function handle(channel: Channel, approvalId: string | undefined, req: ChannelRequest, respond: ChannelCoreRespond): Promise { /** `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).`; diff --git a/src/channels/fetchChannels.ts b/src/channels/fetchChannels.ts index e081958bb..fe9a58a35 100644 --- a/src/channels/fetchChannels.ts +++ b/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 { 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): 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 }; } /** @@ -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; + 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((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; diff --git a/src/channels/mountChannels.ts b/src/channels/mountChannels.ts index e49a61aeb..4f8965009 100644 --- a/src/channels/mountChannels.ts +++ b/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 { From 7beff3cdd3a247e32af023091c1ef4a431bcd9ba Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Mon, 19 Oct 2026 23:38:39 +0300 Subject: [PATCH 2/2] docs: regenerate llms-full.txt Co-Authored-By: Claude Opus 5.5 --- llms-full.txt | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/llms-full.txt b/llms-full.txt index 7f6e5d675..c757d70e6 100644 --- a/llms-full.txt +++ b/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';