diff --git a/CHANGELOG.md b/CHANGELOG.md index 68c53822..4c03c4a4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,7 @@ opt-out. - CI is one workflow with two jobs: `ci` (type checks, lint, build, tests with coverage, docs, API and fallow gates, pack-smoke, size check, and a zod 4 type check) and `compat` (the newest peers, `ai` 7 with zod 4, plus the real-Docker sandbox tests). The separate Docker Engine, Agent Forge E2E and weekly registry-smoke workflows and the `ai` 6 / zod 3 matrix entries are gone; `npm run registry-smoke` and `npm run test:e2e:studio` still run locally. When the fallow gate fails, CI now prints its full health and duplication reports. ### Added +- `POST /chat` accepts an `Idempotency-Key` header: a resend with the same key replays the finished turn's reply (`Idempotent-Replayed: true`, nothing runs), resumes a turn a crash interrupted instead of appending the message again, and is a `409` (`LOUSHO_SESSION_BUSY`) while that turn still runs. The key is stored on the turn's user message (`metadata.idempotencyKey`), so it survives restarts; `session.pending()` reports it as `idempotencyKey`. See [Resending a turn](docs/deployment.md#resending-a-turn). - `fromAiSdkTools(record)` converts Vercel AI SDK tools (`tool({ ... })` or plain `{ description, inputSchema | parameters, execute, needsApproval? }` objects) into Lousho tools, keyed by name: zod and Standard Schemas are kept, `jsonSchema()` wrappers and plain JSON Schemas are converted (their own `validate` still runs), `needsApproval` (boolean or function) pauses the run for approval, and `execute` gets the call id and the run's abort signal. It imports nothing from `ai`, so tools built on `ai` 4 to 7 work with any installed version. New guide: [Migrating from the AI SDK](docs/migrating-from-ai-sdk.md). - `lousho eval` without vitest installed now says it runs `*.eval.ts` files through vitest, the directory it searched from, and the install command; [Evals](docs/evals.md) states the requirement up front. - `result.usage.delegated.toolErrors`: the tool calls that failed inside a run's sub-agents (and theirs), which the lead's own `result.toolErrors` does not count. `ToolCallContext.onDelegatedUsage` takes the child's `toolErrors` as an optional second argument. diff --git a/api/executor.api.md b/api/executor.api.md index 80e3247b..a0bfb414 100644 --- a/api/executor.api.md +++ b/api/executor.api.md @@ -1619,6 +1619,7 @@ interface PendingTurn { approvalId?: string; approvalIds?: string[]; approvalKind?: ApprovalKind; + idempotencyKey?: string; status: 'in-progress' | 'awaiting-approval'; } diff --git a/api/index.api.md b/api/index.api.md index ddfd113e..3330d14e 100644 --- a/api/index.api.md +++ b/api/index.api.md @@ -4571,6 +4571,7 @@ export interface PendingTurn { approvalId?: string; approvalIds?: string[]; approvalKind?: ApprovalKind; + idempotencyKey?: string; status: 'in-progress' | 'awaiting-approval'; } diff --git a/api/mcp.api.md b/api/mcp.api.md index 15d7115e..ab010ca6 100644 --- a/api/mcp.api.md +++ b/api/mcp.api.md @@ -1392,6 +1392,7 @@ interface PendingTurn { approvalId?: string; approvalIds?: string[]; approvalKind?: ApprovalKind; + idempotencyKey?: string; status: 'in-progress' | 'awaiting-approval'; } diff --git a/api/react.api.md b/api/react.api.md index a75e43e6..0bfde3a7 100644 --- a/api/react.api.md +++ b/api/react.api.md @@ -1028,6 +1028,7 @@ interface PendingTurn { approvalId?: string; approvalIds?: string[]; approvalKind?: ApprovalKind; + idempotencyKey?: string; status: 'in-progress' | 'awaiting-approval'; } diff --git a/api/svelte.api.md b/api/svelte.api.md index 64c4a206..97fe56e0 100644 --- a/api/svelte.api.md +++ b/api/svelte.api.md @@ -1046,6 +1046,7 @@ interface PendingTurn { approvalId?: string; approvalIds?: string[]; approvalKind?: ApprovalKind; + idempotencyKey?: string; status: 'in-progress' | 'awaiting-approval'; } diff --git a/api/tools.api.md b/api/tools.api.md index 47f39b89..c28d4b7c 100644 --- a/api/tools.api.md +++ b/api/tools.api.md @@ -1789,6 +1789,7 @@ interface PendingTurn { approvalId?: string; approvalIds?: string[]; approvalKind?: ApprovalKind; + idempotencyKey?: string; status: 'in-progress' | 'awaiting-approval'; } diff --git a/api/vue.api.md b/api/vue.api.md index 756e8e47..36b8b6e0 100644 --- a/api/vue.api.md +++ b/api/vue.api.md @@ -1034,6 +1034,7 @@ interface PendingTurn { approvalId?: string; approvalIds?: string[]; approvalKind?: ApprovalKind; + idempotencyKey?: string; status: 'in-progress' | 'awaiting-approval'; } diff --git a/api/worker.api.md b/api/worker.api.md index b15c2974..0fb0a6f1 100644 --- a/api/worker.api.md +++ b/api/worker.api.md @@ -2684,6 +2684,7 @@ interface PendingTurn { approvalId?: string; approvalIds?: string[]; approvalKind?: ApprovalKind; + idempotencyKey?: string; status: 'in-progress' | 'awaiting-approval'; } diff --git a/docs/deployment.md b/docs/deployment.md index 1554c957..6c361efc 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -113,14 +113,7 @@ 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. - -While a turn streams, the server writes an SSE comment line (`: ping`) after -every 15 seconds without an event, so a proxy (nginx closes an idle upstream -after 60s) or a client (Node's `fetch` gives up after 300s without a byte) -keeps the connection while a model or a tool is slow. SSE parsers ignore -comment lines. `createRouteHandler(agent, { heartbeatMs })` changes the -interval; `0` turns it off. With an auth list, a session belongs to the first principal that sends +new turn. 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 @@ -131,12 +124,51 @@ otherwise. In an agent directory, export `authorizeSession` and a manager decide other users' approvals): see [Who may decide an approval](./auth.md#who-may-decide-an-approval). +While a turn streams, the server writes an SSE comment line (`: ping`) after +every 15 seconds without an event, so a proxy (nginx closes an idle upstream +after 60s) or a client (Node's `fetch` gives up after 300s without a byte) +keeps the connection while a model or a tool is slow. SSE parsers ignore +comment lines. `createRouteHandler(agent, { heartbeatMs })` changes the +interval; `0` turns it off. + ```bash curl -N http://127.0.0.1:3000/chat \ -H 'Authorization: Bearer '"$LOUSHO_API_TOKEN" -H 'Content-Type: application/json' \ -d '{ "sessionId": "alice", "input": "Hello" }' ``` +### Resending a turn + +A `POST /chat` that the client sends again (a retry after a dropped +connection or a timeout, a double click, a client that restarted) is, without +more, a new turn: the user message is appended again and the model may call +the same tools again. Send an `Idempotency-Key` header (1-255 visible ASCII +characters, e.g. a UUID the client makes per message) and resend with the +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. | +| 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`). | + +A session paused on an approval answers `409` +(`LOUSHO_SESSION_AWAITING_APPROVAL`) to a resend as to any turn. The key is +stored with the turn in the session store (the transcript, or the turn's +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. + +```bash +curl -N http://127.0.0.1:3000/chat -H 'Content-Type: application/json' \ + -H 'Idempotency-Key: 6f1c2b0e-7d4a-4c8e-9a51-2f0d3b6e8c11' \ + -d '{ "sessionId": "alice", "input": "Refund order B-1" }' +``` + ### Auth Set `LOUSHO_API_TOKEN` (on the Worker target, as a secret: `npx wrangler secret diff --git a/docs/errors.md b/docs/errors.md index dc733fea..3a8b6286 100644 --- a/docs/errors.md +++ b/docs/errors.md @@ -634,7 +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. +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)). **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/docs/sessions.md b/docs/sessions.md index b6e5ad5b..3148b07e 100644 --- a/docs/sessions.md +++ b/docs/sessions.md @@ -30,7 +30,7 @@ console.log(session.id, session.messages.length); | `compact(options?)` | Compacts the transcript now - see [Compacting a session](./compaction.md#compacting-a-session). | | `clear()` | Empties the transcript and saves it empty (the store holds no messages for the id). Keeps the session id, store and options; deletes every turn's checkpoint and checkpoint history (`.turn-`, which hold copies of the transcript), so `agent.fork('.turn-')` cannot bring the conversation back; memory slots are cross-session and untouched. Emits `context.cleared` to `on()` listeners. Rejects with `LOUSHO_SESSION_BUSY` while a turn is running, and with `LOUSHO_SESSION_AWAITING_APPROVAL` while a durable turn waits on an approval. | | `on(listener)` | Listens for the session's events: every turn's events as they happen (`send()` and `stream()` turns, and a paused turn continued by `agent.approvals.resolve()`), plus `compact()` / `clear()` events (`compaction.start`, `compaction.done`, `context.cleared`); returns a function that removes the listener. A turn's events are the `AgentEvent`s its `stream()` yields. | -| `pending()` | In a [durable session](#durable-sessions), the turn that has not finished (`{ status, approvalId?, approvalKind? }`; `approvalKind` is `'question'` when the turn waits on an `ask_question`), or `null`. | +| `pending()` | In a [durable session](#durable-sessions), the turn that has not finished (`{ status, approvalId?, approvalKind?, idempotencyKey? }`; `approvalKind` is `'question'` when the turn waits on an `ask_question`; `idempotencyKey` is the turn's HTTP `Idempotency-Key`, see [Resending a turn](./deployment.md#resending-a-turn)), or `null`. | | `resume({ signal })` | In a durable session, finishes an interrupted turn and resolves with its result, or `null` when none is pending. | | `discardPending()` | In a durable session, drops an unfinished turn without running it, and withdraws the approvals it waited on (they are no longer listed, and resolving one fails with `LOUSHO_APPROVAL_NOT_FOUND`). Also drops a turn checkpoint that cannot be read (`LOUSHO_CHECKPOINT_CORRUPT`). | | `history()` | The steps of the transcript (one per model response, with its tool calls and results) - see [Forking a session](#forking-a-session). | diff --git a/llms-full.txt b/llms-full.txt index 34a530b0..6ae7afd0 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -7139,14 +7139,7 @@ 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. - -While a turn streams, the server writes an SSE comment line (`: ping`) after -every 15 seconds without an event, so a proxy (nginx closes an idle upstream -after 60s) or a client (Node's `fetch` gives up after 300s without a byte) -keeps the connection while a model or a tool is slow. SSE parsers ignore -comment lines. `createRouteHandler(agent, { heartbeatMs })` changes the -interval; `0` turns it off. With an auth list, a session belongs to the first principal that sends +new turn. 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 @@ -7157,12 +7150,51 @@ otherwise. In an agent directory, export `authorizeSession` and a manager decide other users' approvals): see [Who may decide an approval](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/auth.md#who-may-decide-an-approval). +While a turn streams, the server writes an SSE comment line (`: ping`) after +every 15 seconds without an event, so a proxy (nginx closes an idle upstream +after 60s) or a client (Node's `fetch` gives up after 300s without a byte) +keeps the connection while a model or a tool is slow. SSE parsers ignore +comment lines. `createRouteHandler(agent, { heartbeatMs })` changes the +interval; `0` turns it off. + ```bash curl -N http://127.0.0.1:3000/chat \ -H 'Authorization: Bearer '"$LOUSHO_API_TOKEN" -H 'Content-Type: application/json' \ -d '{ "sessionId": "alice", "input": "Hello" }' ``` +### Resending a turn + +A `POST /chat` that the client sends again (a retry after a dropped +connection or a timeout, a double click, a client that restarted) is, without +more, a new turn: the user message is appended again and the model may call +the same tools again. Send an `Idempotency-Key` header (1-255 visible ASCII +characters, e.g. a UUID the client makes per message) and resend with the +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. | +| 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`). | + +A session paused on an approval answers `409` +(`LOUSHO_SESSION_AWAITING_APPROVAL`) to a resend as to any turn. The key is +stored with the turn in the session store (the transcript, or the turn's +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. + +```bash +curl -N http://127.0.0.1:3000/chat -H 'Content-Type: application/json' \ + -H 'Idempotency-Key: 6f1c2b0e-7d4a-4c8e-9a51-2f0d3b6e8c11' \ + -d '{ "sessionId": "alice", "input": "Refund order B-1" }' +``` + ### Auth Set `LOUSHO_API_TOKEN` (on the Worker target, as a secret: `npx wrangler secret @@ -8766,7 +8798,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. +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)). **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 @@ -16123,7 +16157,7 @@ console.log(session.id, session.messages.length); | `compact(options?)` | Compacts the transcript now - see [Compacting a session](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/compaction.md#compacting-a-session). | | `clear()` | Empties the transcript and saves it empty (the store holds no messages for the id). Keeps the session id, store and options; deletes every turn's checkpoint and checkpoint history (`.turn-`, which hold copies of the transcript), so `agent.fork('.turn-')` cannot bring the conversation back; memory slots are cross-session and untouched. Emits `context.cleared` to `on()` listeners. Rejects with `LOUSHO_SESSION_BUSY` while a turn is running, and with `LOUSHO_SESSION_AWAITING_APPROVAL` while a durable turn waits on an approval. | | `on(listener)` | Listens for the session's events: every turn's events as they happen (`send()` and `stream()` turns, and a paused turn continued by `agent.approvals.resolve()`), plus `compact()` / `clear()` events (`compaction.start`, `compaction.done`, `context.cleared`); returns a function that removes the listener. A turn's events are the `AgentEvent`s its `stream()` yields. | -| `pending()` | In a [durable session](#durable-sessions), the turn that has not finished (`{ status, approvalId?, approvalKind? }`; `approvalKind` is `'question'` when the turn waits on an `ask_question`), or `null`. | +| `pending()` | In a [durable session](#durable-sessions), the turn that has not finished (`{ status, approvalId?, approvalKind?, idempotencyKey? }`; `approvalKind` is `'question'` when the turn waits on an `ask_question`; `idempotencyKey` is the turn's HTTP `Idempotency-Key`, see [Resending a turn](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/deployment.md#resending-a-turn)), or `null`. | | `resume({ signal })` | In a durable session, finishes an interrupted turn and resolves with its result, or `null` when none is pending. | | `discardPending()` | In a durable session, drops an unfinished turn without running it, and withdraws the approvals it waited on (they are no longer listed, and resolving one fails with `LOUSHO_APPROVAL_NOT_FOUND`). Also drops a turn checkpoint that cannot be read (`LOUSHO_CHECKPOINT_CORRUPT`). | | `history()` | The steps of the transcript (one per model response, with its tool calls and results) - see [Forking a session](#forking-a-session). | diff --git a/src/cli/devEvents.ts b/src/cli/devEvents.ts index 4bda0806..d8cc7f26 100644 --- a/src/cli/devEvents.ts +++ b/src/cli/devEvents.ts @@ -4,6 +4,7 @@ * (Approval continuations stream live from `agent.approvals.streamResolve()`.) */ import { AGENT_EVENT_SCHEMA_VERSION, type AgentEvent, type AgentEventPayload } from '../execution/agentEvents'; +import type { ExecutionFinishReason } from '../execution/AgentExecutor'; import { newId } from '../utils/id'; /** `payloads` as one run's events: a shared `runId`, `seq` 0, 1, 2..., `timestamp` and `v` filled in. */ @@ -14,6 +15,11 @@ function runEvents(payloads: AgentEventPayload[]): AgentEvent[] { ); } +/** A lone `run.done` with `text`, for a turn that already ran and is answered again (an idempotent resend). */ +export function replayEvents(text: string, finishReason: ExecutionFinishReason): AgentEvent[] { + return runEvents([{ type: 'run.done', finishReason, text }]); +} + /** `error` then `run.done`, for a request that failed while streaming. */ export function errorEvents(error: unknown): AgentEvent[] { const message = error instanceof Error ? error.message : String(error); diff --git a/src/server/fetchRoutes.ts b/src/server/fetchRoutes.ts index 46f6c19f..d934ddbe 100644 --- a/src/server/fetchRoutes.ts +++ b/src/server/fetchRoutes.ts @@ -4,7 +4,8 @@ * * GET /health `ok`, never behind the bearer token * POST /chat { sessionId, input } in, the turn streamed as SSE out - * (`data: ` frames, then `event: done`); + * (`data: ` frames, then `event: done`); an + * `Idempotency-Key` header makes a resend replay or resume its turn; * the deprecated { message } still returns the ExecutionResult * GET /chat/:sessionId the session's transcript and pending approvals * POST /chat/:sessionId/approvals/:id { approved, note? } or { answer }, the continuation streamed @@ -21,9 +22,11 @@ import type { AgentEvent, AgentEventError } from '../execution/agentEvents'; import { describeApproval, type PendingApproval } from '../execution/ApprovalGate'; import { samePrincipal } from '../execution/runPrincipal'; import { assertSessionId } from '../session/sessionStore'; -import type { AgentSession } from '../session/AgentSession'; +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 { errorEvents } from '../cli/devEvents'; +import { errorEvents, replayEvents } from '../cli/devEvents'; import type { AuthFn, Principal } from '../auth/types'; import { routeAuth } from '../auth/routeAuth'; import { apiToken } from '../auth/basic'; @@ -505,8 +508,180 @@ async function awaitingApproval(session: AgentSession): Promise `), per agent: a resend of one while it runs is a `409` + * (`LOUSHO_SESSION_BUSY`) instead of a second turn. + */ +const keysInFlight = new WeakMap>(); + +function keysOf(agent: SimpleAgent): Set { + let keys = keysInFlight.get(agent); + if (!keys) keysInFlight.set(agent, (keys = new Set())); + return keys; +} + +/** `input` as the turn's user messages, each marked with `key` (`metadata.idempotencyKey`), which the transcript keeps. */ +function keyedInput(input: AgentInput, key: string): Message[] { + return toMessages(input).map((message) => (message.role === 'user' ? { ...message, metadata: { ...message.metadata, idempotencyKey: key } } : message)); +} + +/** The messages of the turn whose user message carries `key` (up to the next turn's user message), or `undefined`. */ +function keyedTurn(transcript: readonly Message[], key: string): Message[] | undefined { + const start = transcript.findIndex((message) => idempotencyKeyOf(message) === key); + if (start < 0) return undefined; + let end = start + 1; + while (end < transcript.length && (transcript[end].role !== 'user' || idempotencyKeyOf(transcript[end]) === key)) end += 1; + return transcript.slice(start, end); +} + +function textOf(message: Message | undefined): string { + const content = message?.content; + if (typeof content === 'string') return content; + return (content ?? []).map((part) => (part.type === 'text' ? part.text : '')).join(''); +} + +/** + * A turn of `session` sent with an `Idempotency-Key`. The key is stored on the + * turn's user message (`metadata.idempotencyKey`), in the transcript and in the + * turn's checkpoint, so it outlives the process: + * + * - its turn is running in this process: `409` (`LOUSHO_SESSION_BUSY`); + * - the session is paused on an approval: `409` (`LOUSHO_SESSION_AWAITING_APPROVAL`), as without a key; + * - its turn was interrupted (a crash): that turn is resumed, not sent again, and streamed; + * - its turn is in the transcript: its reply is replayed as `run.done` + * (`Idempotent-Replayed: true`), and nothing runs; + * - else it runs as a new turn, its user message marked with the key. + */ +async function idempotentTurn(request: Request, ctx: ChatRoutesContext, principal: Principal | undefined, session: AgentSession, input: AgentInput, key: string): Promise { + 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' }); + } + keys.add(claim); + const release = () => void keys.delete(claim); + try { + const pending = await session.pending().catch(() => null); + if (pending?.status === 'awaiting-approval') { + release(); + 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)); + } + const turn = keyedTurn(await session.load(), key); + if (turn) { + release(); + const last = [...turn].reverse().find((message) => message.role === 'assistant'); + const replay = replayEvents(textOf(last), last?.metadata?.interrupted === true ? 'aborted' : 'stop'); + const response = sseResponse(request, ctx, async function* () { + yield* replay; + }); + 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)); + } catch (error) { + release(); + throw error; + } +} + +/** + * The events of `session.resume()`, the turn an earlier request started and a + * crash interrupted, as they happen (through `session.on()`); a resume that + * reports none ends with its result's `run.done`. + */ +async function* resumedTurn(session: AgentSession, signal: AbortSignal): AsyncGenerator { + const queue: AgentEvent[] = []; + let wake: (() => void) | undefined; + let settled = false; + let failure: { error: unknown } | undefined; + let result = null as Awaited>; + const stop = session.on((event) => { + queue.push(event); + wake?.(); + }); + void session + .resume({ signal }) + .then( + (resumed) => void (result = resumed), + (error: unknown) => void (failure = { error }) + ) + .finally(() => { + settled = true; + wake?.(); + }); + try { + let done = false; + for (;;) { + for (let event = queue.shift(); event; event = queue.shift()) { + done ||= event.type === 'run.done'; + yield event; + } + if (settled) break; + await new Promise((resolve) => (wake = resolve)); + wake = undefined; + } + if (failure) throw failure.error; + if (!done && result) yield* replayEvents(result.text, result.finishReason); + } finally { + stop(); + } +} + +/** `run`, calling `release` once when it ends, fails or is returned (also before it started). */ +function releasing(run: AsyncIterable, release: () => void): AsyncIterable { + return { + [Symbol.asyncIterator]() { + const iterator = run[Symbol.asyncIterator](); + let released = false; + const done = () => { + if (released) return; + released = true; + release(); + }; + return { + next: async () => { + try { + const next = await iterator.next(); + if (next.done) done(); + return next; + } catch (error) { + done(); + throw error; + } + }, + return: async () => { + try { + return (await iterator.return?.()) ?? { done: true, value: undefined }; + } finally { + done(); + } + }, + }; + }, + }; +} + type RouteHandler = (request: Request, ctx: ChatRoutesContext, params: string[], principal: Principal | undefined) => Promise; +/** A `{ sessionId, input }` turn whose input passed the body checks: the session checks, then the turn streamed. */ +async function sessionTurn(request: Request, ctx: ChatRoutesContext, principal: Principal | undefined, sessionId: string, input: AgentInput): Promise { + const key = request.headers.get('idempotency-key'); + if (key !== null && !IDEMPOTENCY_KEY.test(key)) return badRequest("The 'Idempotency-Key' header must be 1-255 visible ASCII characters"); + const session = openSession(ctx, sessionId); + if (session instanceof Response) return session; + 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 }))); +} + const runChat: RouteHandler = async (request, ctx, _params, principal) => { const { sessionId, input, message } = await readJson(request); if (typeof sessionId === 'string' && Array.isArray(input) && input.length === 0) return badRequest("Request body's 'input' is empty: send the user's message"); @@ -522,10 +697,7 @@ const runChat: RouteHandler = async (request, ctx, _params, principal) => { if (Array.isArray(messages) && messages.length === 0) { return badRequest("Request body's 'input' must contain at least one 'user' message"); } - const session = openSession(ctx, sessionId); - if (session instanceof Response) return session; - const forbidden = (await sessionForbidden(ctx, principal, sessionId, 'chat', session)) ?? (await awaitingApproval(session)); - return forbidden ?? sseResponse(request, ctx, (signal) => claimOnCommit(ctx, principal, session, session.stream(messages as AgentInput, { signal, principal }))); + return sessionTurn(request, ctx, principal, sessionId, messages as AgentInput); } if (typeof message === 'string' && message) { if (sessionId !== undefined && typeof sessionId !== 'string') return badRequest("Request body's 'sessionId', if present, must be a string"); @@ -663,20 +835,7 @@ const runApproval: RouteHandler = async (request, ctx, [sessionId, id], principa inFlight.delete(decision); return gate; } - const decided = gate; - return sseResponse( - request, - ctx, - () => - (async function* () { - try { - yield* decided; - } finally { - inFlight.delete(decision); - } - })(), - gate.controller - ); + return sseResponse(request, ctx, () => releasing(gate, () => void inFlight.delete(decision)), gate.controller); }; /** Binds a recovered pause to `session` and starts the decision's continuation, gated by {@link decisionGate}. */ diff --git a/src/server/idempotency.test.ts b/src/server/idempotency.test.ts new file mode 100644 index 00000000..44fcb7eb --- /dev/null +++ b/src/server/idempotency.test.ts @@ -0,0 +1,136 @@ +/** + * `POST /chat` with an `Idempotency-Key`: a client that resends a turn + * (a retry after a dropped connection, a double submit, a restart) gets that + * turn's outcome instead of a second turn with the same side effects. + */ +import { describe, expect, it } from 'vitest'; +import { z } from 'zod'; +import { createAgent } from '../createAgent'; +import { defineTool } from '../tools/defineTool'; +import { PropagatingToolError } from '../execution/AgentExecutor'; +import { mockModel, type MockTurn } from '../testing'; +import { memoryStore, type AgentStore } from '../storage/agentStore'; +import { createRouteHandler } from './routeHandler'; + +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), + }); + +async function events(response: Response): Promise & { type: string }>> { + return (await response.text()) + .split('\n\n') + .filter((frame) => frame.startsWith('data: ')) + .map((frame) => JSON.parse(frame.slice(6))); +} + +const doneText = async (response: Response) => (await events(response)).find((event) => event.type === 'run.done')?.text; + +function desk(script: readonly MockTurn[], store: AgentStore = memoryStore(), crashOnce = false) { + const refunds: string[] = []; + let runs = 0; + const refund = defineTool({ + name: 'issue_refund', + description: 'Refunds an order', + input: z.object({ orderId: z.string() }), + execute: async ({ orderId }) => { + runs += 1; + if (crashOnce && runs === 1) throw new PropagatingToolError('process died while refunding'); + refunds.push(orderId); + return 'refunded'; + }, + }); + const model = mockModel(script); + const agent = createAgent({ provider: model, tools: [refund], store }); + return { agent, model, refunds, store, ...createRouteHandler(agent, { basePath: '/api' }) }; +} + +const refundTurn: MockTurn[] = [{ toolCalls: [{ name: 'issue_refund', args: { orderId: 'B-1' } }] }, 'Refund issued.']; + +const userMessages = async (store: AgentStore, id: string) => ((await store.sessions!.load(id)) ?? []).filter((message) => message.role === 'user'); + +describe('POST /chat with an Idempotency-Key', () => { + it('replays a finished turn on a resend instead of running it again', async () => { + const { handler, model, refunds, store } = desk(refundTurn); + expect(await doneText(await handler(post({ sessionId: 'k-1', input: 'Refund B-1' }, 'req-1')))).toBe('Refund issued.'); + + const resend = await handler(post({ sessionId: 'k-1', input: 'Refund B-1' }, 'req-1')); + expect(resend.status).toBe(200); + expect(resend.headers.get('content-type')).toBe('text/event-stream'); + expect(resend.headers.get('idempotent-replayed')).toBe('true'); + const replayed = await events(resend); + expect(replayed.map((event) => event.type)).toEqual(['run.done']); + expect(replayed[0]).toMatchObject({ text: 'Refund issued.', finishReason: 'stop' }); + + expect(refunds).toEqual(['B-1']); + expect(model.calls).toHaveLength(2); + const users = await userMessages(store, 'k-1'); + expect(users).toHaveLength(1); + expect(users[0].metadata).toMatchObject({ idempotencyKey: 'req-1' }); + }); + + it('runs a new turn for a new key, and an unkeyed resend as before', async () => { + const { handler, store } = desk(['One.', 'Two.', 'Three.']); + expect(await doneText(await handler(post({ sessionId: 'k-2', input: 'hi' }, 'a')))).toBe('One.'); + expect(await doneText(await handler(post({ sessionId: 'k-2', input: 'hi' }, 'b')))).toBe('Two.'); + expect(await doneText(await handler(post({ sessionId: 'k-2', input: 'hi' })))).toBe('Three.'); + expect(await userMessages(store, 'k-2')).toHaveLength(3); + }); + + it('is a 409 while the same key is still running its turn', async () => { + let release!: () => void; + const slow = new Promise((resolve) => (release = resolve)); + const { handler, model } = desk([ + async () => { + await slow; + return 'Done.'; + }, + ]); + const first = await handler(post({ sessionId: 'k-3', input: 'hi' }, 'dup')); + const firstEvents = events(first); + + const double = await handler(post({ sessionId: 'k-3', input: 'hi' }, 'dup')); + expect(double.status).toBe(409); + expect(await double.json()).toMatchObject({ code: 'LOUSHO_SESSION_BUSY' }); + + release(); + expect((await firstEvents).find((event) => event.type === 'run.done')?.text).toBe('Done.'); + expect(model.calls).toHaveLength(1); + // Once it finished, the same key replays. + expect(await doneText(await handler(post({ sessionId: 'k-3', input: 'hi' }, 'dup')))).toBe('Done.'); + }); + + it('survives a restart: a fresh agent over the same store replays the turn', async () => { + const store = memoryStore(); + await events(await desk(['Stored.'], store).handler(post({ sessionId: 'k-4', input: 'hi' }, 'req-4'))); + + const restarted = desk([], store); + expect(await doneText(await restarted.handler(post({ sessionId: 'k-4', input: 'hi' }, 'req-4')))).toBe('Stored.'); + expect(restarted.model.calls).toHaveLength(0); + }); + + it('resumes a turn a crash interrupted, without appending the message again or re-running finished steps', async () => { + const store = memoryStore(); + const crashed = desk(refundTurn, store, true); + const failed = await events(await crashed.handler(post({ sessionId: 'k-5', input: 'Refund B-1' }, 'req-5'))); + expect(failed.some((event) => event.type === 'error')).toBe(true); + expect(await crashed.agent.session({ id: 'k-5' }).pending()).toMatchObject({ status: 'in-progress', idempotencyKey: 'req-5' }); + + const restarted = desk(['Refund issued.'], store); + const resumed = await events(await restarted.handler(post({ sessionId: 'k-5', input: 'Refund B-1' }, 'req-5'))); + expect(resumed.find((event) => event.type === 'run.done')).toMatchObject({ text: 'Refund issued.', finishReason: 'stop' }); + expect(restarted.refunds).toEqual(['B-1']); + expect(restarted.model.calls).toHaveLength(1); + expect(await userMessages(store, 'k-5')).toHaveLength(1); + expect(await restarted.agent.session({ id: 'k-5' }).pending()).toBeNull(); + }); + + it('rejects a malformed key', async () => { + const { handler } = desk(['x']); + const response = await handler(post({ sessionId: 'k-6', input: 'hi' }, 'has space')); + expect(response.status).toBe(400); + expect(await response.json()).toMatchObject({ code: 'LOUSHO_REQUEST_INVALID' }); + }); +}); diff --git a/src/session/AgentSession.ts b/src/session/AgentSession.ts index 32167ea9..472cd096 100644 --- a/src/session/AgentSession.ts +++ b/src/session/AgentSession.ts @@ -110,6 +110,12 @@ export interface PendingTurn { approvalIds?: string[]; /** `'question'` when the turn waits on an `ask_question` call; absent for a tool approval (or a turn paused before this field existed). */ approvalKind?: ApprovalKind; + /** + * The `metadata.idempotencyKey` of the turn's user message, when it was sent + * with one (the HTTP API's `Idempotency-Key` header), so a client retrying + * that request can tell its own unfinished turn from another. + */ + idempotencyKey?: string; } /** Where a checkpointed session's turn is checkpointed: pass both to `AgentExecutor.execute()`. */ @@ -260,6 +266,21 @@ function pauseApprovalIds(paused: InMemoryPause): string[] { return paused.approvalIds ?? (paused.approvalId ? [paused.approvalId] : []); } +/** @internal The `metadata.idempotencyKey` of `message`, when it is a user message that has one. */ +export function idempotencyKeyOf(message: Message): string | undefined { + const key = message.role === 'user' ? message.metadata?.idempotencyKey : undefined; + return typeof key === 'string' ? key : undefined; +} + +/** The idempotency key of a pending turn: the last one among its messages that the transcript does not hold yet. */ +function pendingIdempotencyKey(pending: readonly Message[], transcript: readonly Message[]): string | undefined { + const committed = new Set(transcript.map(idempotencyKeyOf)); + return pending + .map(idempotencyKeyOf) + .filter((key) => key !== undefined && !committed.has(key)) + .at(-1); +} + /** The `pending()` view of an in-process pause (no checkpoint store). */ function inMemoryPending(paused: InMemoryPause): PendingTurn { const approvalIds = pauseApprovalIds(paused); @@ -712,7 +733,14 @@ export class AgentSession { const status = checkpoint.status === 'awaiting-approval' ? 'awaiting-approval' : 'in-progress'; const { approvalId, approvalKind } = checkpoint; const approvalIds = status === 'awaiting-approval' ? pausedApprovalIds(checkpoint) : []; - return { status, approvalId, ...(approvalIds.length > 1 && { approvalIds }), ...(status === 'awaiting-approval' && approvalKind && { approvalKind }) }; + const idempotencyKey = pendingIdempotencyKey(checkpoint.messages, this.transcript); + return { + status, + approvalId, + ...(approvalIds.length > 1 && { approvalIds }), + ...(status === 'awaiting-approval' && approvalKind && { approvalKind }), + ...(idempotencyKey !== undefined && { idempotencyKey }), + }; }); }