From b256b0c2c01a3478316b1e1a81661d3081094cb2 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Tue, 20 Oct 2026 00:22:15 +0300 Subject: [PATCH] V3: Recover interrupted runs when the built server starts The node-server and docker targets now call agent.recover() once the server listens, so a run or session turn a crash left in-progress is finished from its last checkpoint without a client resending it. Runs paused on an approval are left alone; leased runs are skipped and retried. LOUSHO_RECOVER_ON_START=0 turns it off. Co-Authored-By: Claude Opus 5.5 --- docs/deployment.md | 27 ++++ llms-full.txt | 27 ++++ src/deploy/adapters/node-server.ts | 5 + src/deploy/nodeServer.ts | 62 +++++++- src/deploy/nodeServerRecover.test.ts | 216 +++++++++++++++++++++++++++ 5 files changed, 336 insertions(+), 1 deletion(-) create mode 100644 src/deploy/nodeServerRecover.test.ts diff --git a/docs/deployment.md b/docs/deployment.md index a70f6c9a..6eeaa6a6 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -214,6 +214,33 @@ directory's `auth.ts`, with `LOUSHO_API_TOKEN` appended to it: see SQLite is one file on one disk, so run a single instance per database file. +### Crash recovery + +When the `node-server` or `docker` server starts listening, it finishes the +runs that a crash or a kill left unfinished in the store. It calls +[`agent.recover()`](./durable-execution.md), which resumes every +`in-progress` run and session turn from its last checkpoint. Steps that +completed are not run again. A tool call that was in flight when the process +died runs again, so make such tools idempotent. The server logs +`lousho server: resuming N interrupted run(s)` when it finds any, and one +`recovered interrupted run ''` line per run it finishes. It then checks +the store again every 30 seconds until it shuts down. + +- Runs paused on an approval or a question are left alone. Decide them + through `POST /chat/:sessionId/approvals/:id`. +- A run whose [run lease](./durable-execution.md#run-leases) another live + process holds is skipped and retried on a later check. So is a run whose + crashed process's lease has not expired yet. +- With the default `memory` store there is nothing to recover after a + restart, and the check finds nothing. +- A resend of the same turn that arrives while the server is still resuming + it runs after the resumed turn, as a new turn. Send an `Idempotency-Key` + ([Resending a turn](#resending-a-turn)) to get the resumed turn's reply + instead. + +Set `LOUSHO_RECOVER_ON_START=0` to turn this off, for example when another +process already runs `agent.recover()` for the same store. + ## `node-server` `dist/server.js` is a single self-contained bundle (the SDK and its diff --git a/llms-full.txt b/llms-full.txt index fac0e9be..4263f97f 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -7263,6 +7263,33 @@ directory's `auth.ts`, with `LOUSHO_API_TOKEN` appended to it: see SQLite is one file on one disk, so run a single instance per database file. +### Crash recovery + +When the `node-server` or `docker` server starts listening, it finishes the +runs that a crash or a kill left unfinished in the store. It calls +[`agent.recover()`](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/durable-execution.md), which resumes every +`in-progress` run and session turn from its last checkpoint. Steps that +completed are not run again. A tool call that was in flight when the process +died runs again, so make such tools idempotent. The server logs +`lousho server: resuming N interrupted run(s)` when it finds any, and one +`recovered interrupted run ''` line per run it finishes. It then checks +the store again every 30 seconds until it shuts down. + +- Runs paused on an approval or a question are left alone. Decide them + through `POST /chat/:sessionId/approvals/:id`. +- A run whose [run lease](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/durable-execution.md#run-leases) another live + process holds is skipped and retried on a later check. So is a run whose + crashed process's lease has not expired yet. +- With the default `memory` store there is nothing to recover after a + restart, and the check finds nothing. +- A resend of the same turn that arrives while the server is still resuming + it runs after the resumed turn, as a new turn. Send an `Idempotency-Key` + ([Resending a turn](#resending-a-turn)) to get the resumed turn's reply + instead. + +Set `LOUSHO_RECOVER_ON_START=0` to turn this off, for example when another +process already runs `agent.recover()` for the same store. + ## `node-server` `dist/server.js` is a single self-contained bundle (the SDK and its diff --git a/src/deploy/adapters/node-server.ts b/src/deploy/adapters/node-server.ts index 211f6835..ee91f606 100644 --- a/src/deploy/adapters/node-server.ts +++ b/src/deploy/adapters/node-server.ts @@ -209,6 +209,11 @@ function serverSource(variant: ServerVariant): string { * LOUSHO_ALLOWED_HOSTS (comma-separated) and LOUSHO_REQUIRE_JSON_BODY=false * set the routes' cross-site, Host and body checks (docs/auth.md). * + * Crash recovery: once listening, the server resumes the runs a crash left + * in-progress in the store (agent.recover(): completed steps do not run again, + * runs paused on an approval are left alone) and logs how many it found. + * LOUSHO_RECOVER_ON_START=0 turns it off. + * * Binds to 127.0.0.1 by default; pass --host= (or HOST=) to opt in to * listening on other interfaces. Port: --port= / PORT, default 3000. */ diff --git a/src/deploy/nodeServer.ts b/src/deploy/nodeServer.ts index 0d4df8d5..c7951cc4 100644 --- a/src/deploy/nodeServer.ts +++ b/src/deploy/nodeServer.ts @@ -9,11 +9,14 @@ * `/health` requires it, including a channel route, unless that channel * authenticates its own requests (it has a `verify`: a signed platform channel, * a webhook with a secret, an `httpChannel({ verify })`). + * Once it listens it finishes the runs a crash left interrupted + * (`agent.recover()`), unless `LOUSHO_RECOVER_ON_START=0`. * * Node-only (the Worker target has its own runtime, runtime.worker.ts). */ import * as http from 'node:http'; import type { SimpleAgent } from '../createAgent'; +import type { RunRecovery } from '../agentRecover'; import { relayFetch } from '../server/chatRoutes'; import { serveFetch, type ChatRoutesAccess, type ChatRoutesContext, type ServeAuth } from '../server/fetchRoutes'; import { startSchedules, type StartSchedulesOptions } from '../schedules/startSchedules'; @@ -34,6 +37,8 @@ import { routeAuth } from '../auth/routeAuth'; const API_TOKEN_ENV = 'LOUSHO_API_TOKEN'; /** Environment variable choosing where sessions live: `memory` (default) or `sqlite:`. */ const STORE_ENV = 'LOUSHO_STORE'; +/** Environment variable that turns off crash recovery at start-up when `0` or `false`. */ +const RECOVER_ENV = 'LOUSHO_RECOVER_ON_START'; /** * `authorizeSession`, `authorizeApproval` and `exposeErrors` work as on @@ -58,6 +63,14 @@ export interface DeployedServerOptions extends ChatRoutesAccess { scheduler?: StartSchedulesOptions; /** Channels of the served agent directory (`resolveAgentDir()`'s `channels`): mounted under `/channels` next to the chat routes. */ channels?: readonly Channel[]; + /** + * Crash recovery: once the server listens, `agent.recover()` resumes the + * runs a crash left `in-progress` in the store (runs paused on an approval + * are left alone), and keeps sweeping every `intervalMs` (default 30 s) until + * the server shuts down. On by default; `false`, or `LOUSHO_RECOVER_ON_START=0` + * in `env`, turns it off. `intervalMs` is mainly for tests. + */ + recovery?: false | { intervalMs?: number }; } /** The store `LOUSHO_STORE` names: `memory` (the default, lost on restart) or `sqlite:` (Node >= 22). */ @@ -139,6 +152,38 @@ async function channelRequestAllowed(req: http.IncomingMessage, res: http.Server return false; } +/** Whether crash recovery runs at start-up: not with `recovery: false` or `LOUSHO_RECOVER_ON_START=0` / `false`. */ +function recoveryEnabled(options: DeployedServerOptions): boolean { + const value = (options.env ?? process.env)[RECOVER_ENV]?.trim().toLowerCase(); + return options.recovery !== false && value !== '0' && value !== 'false'; +} + +/** Codes of `agent.pending()` / `agent.recover()` for a store that keeps no listable checkpoints: nothing to recover. */ +const NOT_RECOVERABLE = new Set(['LOUSHO_CONFIG_MISSING_CHECKPOINT_STORE', 'LOUSHO_CONFIG_INVALID']); + +/** + * Starts `agent.recover()` (stopped by `signal`), after logging how many + * interrupted runs it found. Never throws: a store that cannot list its + * checkpoints (none, or a custom one without `list()`) has nothing to + * recover, and any other failure is logged without stopping the server. + */ +async function startRunRecovery(agent: SimpleAgent, signal: AbortSignal, intervalMs: number | undefined): Promise { + try { + const interrupted = (await agent.pending()).filter((run) => run.status === 'in-progress' && run.lastError?.retryable !== false); + if (signal.aborted) return undefined; + if (interrupted.length > 0) console.log(`lousho server: resuming ${interrupted.length} interrupted run(s)`); + return agent.recover({ + signal, + ...(intervalMs !== undefined && { intervalMs }), + onRecovered: (run) => console.log(`lousho server: recovered interrupted run '${run.sessionId}'`), + onError: (error, run) => console.error(`[lousho server] recovery${run ? ` of run '${run.sessionId}'` : ''} failed:`, error), + }); + } catch (error) { + if (!NOT_RECOVERABLE.has((error as { code?: string } | null)?.code ?? '')) console.error('[lousho server] recovery failed to start:', error); + return undefined; + } +} + /** Default bound of {@link DeployedServer.shutdown}'s wait for in-flight turns; `LOUSHO_SHUTDOWN_TIMEOUT_MS` overrides it in the generated server. */ const DEFAULT_SHUTDOWN_TIMEOUT_MS = 10_000; @@ -151,7 +196,9 @@ export interface DeployedServer { channelRoutes: { name: string; auth: ChannelRouteAuth }[]; /** * A graceful stop, for SIGTERM. Stops accepting connections - * (`server.close()`), stops the schedules (no new fires), then waits for + * (`server.close()`), stops the schedules (no new fires) and crash recovery + * (its resumes in flight are aborted; their checkpoints stay for the next + * start), then waits for * what is in flight - requests and their turns (a channel turn, a stream), * turns that outlived a disconnected client, and schedule fires - for at * most `timeoutMs` (default 10 s), and finally closes the connections left. @@ -216,6 +263,18 @@ export function createDeployedServer(agent: SimpleAgent, options: DeployedServer server.on('listening', () => (running = startSchedules(agent, options.schedules ?? [], options.scheduler))); server.on('close', () => void stopSchedules()); } + // Crash recovery: the runs a crash left in-progress are resumed once the server listens. + const recoveryAbort = new AbortController(); + let recovering: Promise | undefined; + const stopRecovery = async (): Promise => { + recoveryAbort.abort(); + await (await recovering)?.done; + }; + if (recoveryEnabled(options)) { + const intervalMs = options.recovery === false ? undefined : options.recovery?.intervalMs; + server.on('listening', () => (recovering ??= startRunRecovery(agent, recoveryAbort.signal, intervalMs))); + server.on('close', () => void stopRecovery()); + } const shutdown = async ({ timeoutMs = DEFAULT_SHUTDOWN_TIMEOUT_MS }: { timeoutMs?: number } = {}): Promise => { if (server.listening) server.close(); server.closeIdleConnections(); @@ -223,6 +282,7 @@ export function createDeployedServer(agent: SimpleAgent, options: DeployedServer const timedOut = new Promise<'timeout'>((done) => (timer = setTimeout(() => done('timeout'), timeoutMs))); const drained = (async () => { await stopSchedules(); + await stopRecovery(); // A request can start a detached turn while it ends: wait until nothing is left. while (inFlight.size > 0) await Promise.all(inFlight); })(); diff --git a/src/deploy/nodeServerRecover.test.ts b/src/deploy/nodeServerRecover.test.ts new file mode 100644 index 00000000..a7c7c457 --- /dev/null +++ b/src/deploy/nodeServerRecover.test.ts @@ -0,0 +1,216 @@ +/** + * The deployed node server finishes the runs a crash left interrupted once it + * listens (`agent.recover()`): only the in-flight step runs again, runs paused + * on an approval are left alone, and `LOUSHO_RECOVER_ON_START=0` turns it off. + */ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import type { AddressInfo } from 'node:net'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { z } from 'zod'; +import { createAgent } from '../createAgent'; +import { defineTool, type DefinedTool } from '../tools/defineTool'; +import { PropagatingToolError } from '../execution/AgentExecutor'; +import { fileStore } from '../storage/fileStore'; +import { memoryStore } from '../storage/agentStore'; +import { mockModel, type MockTurn } from '../testing'; +import { createDeployedServer, type DeployedServerOptions } from './nodeServer'; +import type { SimpleAgent } from '../createAgent'; + +const calling = (name: string): MockTurn => ({ toolCalls: [{ name, id: `call_${name}` }] }); +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +const closers: (() => Promise)[] = []; +const dirs: string[] = []; +afterEach(async () => { + for (const close of closers.splice(0).reverse()) await close(); + while (dirs.length) rmSync(dirs.pop() as string, { recursive: true, force: true }); + vi.restoreAllMocks(); +}); + +/** + * `charge` counts its calls (the step that completed before the crash); + * `crash` dies the first time it runs, as the process would mid-tool, and + * waits on `gate` after that; `send_email` needs an approval. + */ +function tools(runs: Record, gate: Promise = Promise.resolve()): DefinedTool[] { + runs.charge ??= 0; + runs.crash ??= 0; + runs.send_email ??= 0; + return [ + defineTool({ name: 'charge', description: 'charge', input: z.object({}), execute: async () => `charged ${++runs.charge}` }), + defineTool({ + name: 'crash', + description: 'crash', + input: z.object({}), + execute: async () => { + if (++runs.crash === 1) throw new PropagatingToolError('process died'); + await gate; + return 'ok'; + }, + }), + defineTool({ name: 'send_email', description: 'email', input: z.object({}), needsApproval: true, execute: async () => `sent ${++runs.send_email}` }), + ]; +} + +function tempDir(): string { + const dir = mkdtempSync(join(tmpdir(), 'lousho-server-recover-')); + dirs.push(dir); + return dir; +} + +/** Leaves session `id` of the store in `dir` with a turn interrupted mid-tool: `charge` ran, `crash` died. */ +async function interruptedTurn(dir: string, id: string, runs: Record): Promise { + const before = createAgent({ provider: mockModel([calling('charge'), calling('crash')]), tools: tools(runs), store: fileStore(dir) }); + await expect(before.session({ id }).send('charge, then finish')).rejects.toThrow('process died'); + expect(await before.pending()).toEqual([expect.objectContaining({ sessionId: id, kind: 'session', status: 'in-progress' })]); +} + +async function serve(agent: SimpleAgent, options: DeployedServerOptions) { + closers.push(() => agent.close()); + const deployed = createDeployedServer(agent, options); + closers.push(() => deployed.shutdown({ timeoutMs: 200 })); + await new Promise((done) => deployed.server.listen(0, '127.0.0.1', done)); + return { ...deployed, url: `http://127.0.0.1:${(deployed.server.address() as AddressInfo).port}` }; +} + +interface SessionView { + messages: { role: string; content?: unknown }[]; + pending: { status: string } | null; +} + +async function sessionView(url: string, id: string): Promise { + return (await (await fetch(`${url}/chat/${id}`)).json()) as SessionView; +} + +describe('createDeployedServer() crash recovery', () => { + it('resumes a turn interrupted mid-tool once the server listens, re-running only the in-flight step', async () => { + const dir = tempDir(); + const runs: Record = {}; + await interruptedTurn(dir, 'chat-1', runs); + const log = vi.spyOn(console, 'log').mockImplementation(() => undefined); + const error = vi.spyOn(console, 'error').mockImplementation(() => undefined); + + const agent = createAgent({ provider: mockModel(['all done']), tools: tools(runs), store: fileStore(dir) }); + const { url } = await serve(agent, { env: {}, recovery: { intervalMs: 20 } }); + + await vi.waitFor(async () => expect((await sessionView(url, 'chat-1')).pending).toBeNull(), { timeout: 3000 }); + const view = await sessionView(url, 'chat-1'); + // One turn: one user message, ending with the reply the resume produced. + expect(view.messages.filter((message) => message.role === 'user')).toHaveLength(1); + expect(view.messages.at(-1)).toMatchObject({ role: 'assistant', content: 'all done' }); + // The completed charge did not run again; only the call that was mid-flight did. + expect(runs).toMatchObject({ charge: 1, crash: 2 }); + expect(await agent.pending()).toEqual([]); + expect(log).toHaveBeenCalledWith('lousho server: resuming 1 interrupted run(s)'); + expect(log).toHaveBeenCalledWith("lousho server: recovered interrupted run 'chat-1'"); + expect(error).not.toHaveBeenCalled(); + }); + + it('leaves a turn paused on an approval alone', async () => { + const dir = tempDir(); + const runs: Record = {}; + const before = createAgent({ provider: mockModel([calling('send_email')]), tools: tools(runs), store: fileStore(dir) }); + const paused = await before.session({ id: 'chat-approval' }).send('email Sam'); + expect(paused.finishReason).toBe('awaiting-approval'); + const log = vi.spyOn(console, 'log').mockImplementation(() => undefined); + + const agent = createAgent({ provider: mockModel(['should not run']), tools: tools(runs), store: fileStore(dir) }); + const { url } = await serve(agent, { env: {}, recovery: { intervalMs: 10 } }); + await sleep(150); + + expect(runs.send_email).toBe(0); + expect((await sessionView(url, 'chat-approval')).pending).toMatchObject({ status: 'awaiting-approval' }); + expect(await agent.pending()).toEqual([expect.objectContaining({ sessionId: 'chat-approval', status: 'awaiting-approval', approvalId: paused.approvalId })]); + expect(log).not.toHaveBeenCalledWith(expect.stringContaining('interrupted run')); + }); + + it.each([ + ['LOUSHO_RECOVER_ON_START=0', { env: { LOUSHO_RECOVER_ON_START: '0' }, recovery: { intervalMs: 10 } }], + ['LOUSHO_RECOVER_ON_START=false', { env: { LOUSHO_RECOVER_ON_START: 'false' }, recovery: { intervalMs: 10 } }], + ['recovery: false', { env: {}, recovery: false }], + ] as const)('does nothing with %s', async (_name, options) => { + const dir = tempDir(); + const runs: Record = {}; + await interruptedTurn(dir, 'chat-off', runs); + + const agent = createAgent({ provider: mockModel(['all done']), tools: tools(runs), store: fileStore(dir) }); + const { url } = await serve(agent, options); + await sleep(150); + + expect(runs).toMatchObject({ charge: 1, crash: 1 }); + expect((await sessionView(url, 'chat-off')).pending).toMatchObject({ status: 'in-progress' }); + }); + + it('starts quietly with nothing to recover: an empty memory store, or no checkpoint store at all', async () => { + const log = vi.spyOn(console, 'log').mockImplementation(() => undefined); + const error = vi.spyOn(console, 'error').mockImplementation(() => undefined); + const withStore = createAgent({ provider: mockModel(['hi']), store: memoryStore() }); + const withoutStore = createAgent({ provider: mockModel(['hi']) }); + for (const agent of [withStore, withoutStore]) { + const { url } = await serve(agent, { env: {}, recovery: { intervalMs: 10 } }); + await sleep(50); + expect((await fetch(`${url}/health`)).status).toBe(200); + } + expect(log).not.toHaveBeenCalled(); + expect(error).not.toHaveBeenCalled(); + }); + + it('runs a resend that races the recovery after the recovered turn, so no tool of that turn runs twice', async () => { + const dir = tempDir(); + const runs: Record = {}; + await interruptedTurn(dir, 'chat-race', runs); + vi.spyOn(console, 'log').mockImplementation(() => undefined); + + let release!: () => void; + const gate = new Promise((done) => (release = done)); + const agent = createAgent({ provider: mockModel(['all done', 'resent reply']), tools: tools(runs, gate), store: fileStore(dir) }); + const { url } = await serve(agent, { env: {}, recovery: { intervalMs: 20 } }); + // Recovery is now mid-tool, resuming the interrupted turn. + await vi.waitFor(() => expect(runs.crash).toBe(2), { timeout: 3000 }); + + // The client resends without an Idempotency-Key: the session's turns of this process run one at a time. + const resending = fetch(`${url}/chat`, { method: 'POST', headers: { 'content-type': 'application/json' }, body: JSON.stringify({ sessionId: 'chat-race', input: 'charge, then finish' }) }); + await sleep(100); + release(); + const resend = await resending; + expect(resend.status).toBe(200); + expect(await resend.text()).toContain('resent reply'); + + expect(runs).toMatchObject({ charge: 1, crash: 2 }); + const { messages, pending } = await sessionView(url, 'chat-race'); + expect(pending).toBeNull(); + expect(messages.filter((message) => message.role === 'tool')).toHaveLength(2); + expect(messages.filter((message) => message.role === 'assistant' && message.content).map((message) => message.content)).toEqual(['all done', 'resent reply']); + }); + + it('answers a resend with the same Idempotency-Key that races the recovery without running the turn again', async () => { + const dir = tempDir(); + const runs: Record = {}; + const headers = { 'content-type': 'application/json', 'idempotency-key': 'msg-1' }; + const body = JSON.stringify({ sessionId: 'chat-key', input: 'charge, then finish' }); + const before = createAgent({ provider: mockModel([calling('charge'), calling('crash')]), tools: tools(runs), store: fileStore(dir) }); + const first = await serve(before, { env: {}, recovery: false }); + await (await fetch(`${first.url}/chat`, { method: 'POST', headers, body })).text(); + expect(runs).toMatchObject({ charge: 1, crash: 1 }); + vi.spyOn(console, 'log').mockImplementation(() => undefined); + + let release!: () => void; + const gate = new Promise((done) => (release = done)); + const agent = createAgent({ provider: mockModel(['all done', 'not used']), tools: tools(runs, gate), store: fileStore(dir) }); + const { url } = await serve(agent, { env: {}, recovery: { intervalMs: 20 } }); + await vi.waitFor(() => expect(runs.crash).toBe(2), { timeout: 3000 }); + + const resending = fetch(`${url}/chat`, { method: 'POST', headers, body }); + await sleep(100); + release(); + const resend = await resending; + expect(resend.status).toBe(200); + expect(await resend.text()).toContain('"text":"all done"'); + expect(runs).toMatchObject({ charge: 1, crash: 2 }); + const { messages, pending } = await sessionView(url, 'chat-key'); + expect(pending).toBeNull(); + expect(messages.filter((message) => message.role === 'user')).toHaveLength(1); + }); +});