From 95675a2f25e326f94a45b4f09e60b8d734198de4 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Thu, 12 Nov 2026 08:49:14 +0000 Subject: [PATCH 1/3] wip: durable runs --- .../src/app/api/v1/chat/completions/route.ts | 27 +- .../src/app/api/v1/chat/runs/[id]/route.ts | 19 +- apps/web/src/app/api/v1/chat/runs/route.ts | 4 +- apps/web/src/instrumentation.ts | 12 + apps/web/src/lib/chat/run-manager.ts | 26 + apps/web/src/lib/chat/run-opener.ts | 39 ++ apps/web/src/lib/chat/runs.ts | 549 ++++++++++++++---- apps/web/src/lib/chat/server.ts | 2 +- apps/web/src/lib/chat/sync.test.ts | 241 ++++++-- 9 files changed, 733 insertions(+), 186 deletions(-) create mode 100644 apps/web/src/instrumentation.ts create mode 100644 apps/web/src/lib/chat/run-manager.ts create mode 100644 apps/web/src/lib/chat/run-opener.ts diff --git a/apps/web/src/app/api/v1/chat/completions/route.ts b/apps/web/src/app/api/v1/chat/completions/route.ts index 9feab3a5..501532a2 100644 --- a/apps/web/src/app/api/v1/chat/completions/route.ts +++ b/apps/web/src/app/api/v1/chat/completions/route.ts @@ -1,7 +1,8 @@ import { forwardChat, parseChatRequest } from "@/lib/chat/completions"; import { hasEnabledPlugins } from "@/lib/chat/plugins"; import { pluginChatResponse } from "@/lib/chat/plugin-http"; -import { RunLimitError, SSE_HEADERS, runStream, startRun } from "@/lib/chat/runs"; +import { runManager } from "@/lib/chat/run-manager"; +import { RunLimitError, SSE_HEADERS } from "@/lib/chat/runs"; import { currentUser, infernetFor, json, rateLimit, sameOrigin, unauthorized } from "@/lib/chat/server"; export const dynamic = "force-dynamic"; @@ -38,26 +39,12 @@ export async function POST(req: Request) { const approvals = Array.isArray(raw?.approvals) ? raw.approvals : []; const withPlugins = raw?.plugins !== false && (approvals.length > 0 || (await hasEnabledPlugins(user.id))); - if (raw?.resumable === true && parsed.stream) { + if (raw?.resumable === true && parsed.stream && raw) { try { - const run = await startRun(user.id, async (signal) => { - if (withPlugins) { - const res = await pluginChatResponse({ userId: user.id, upstream: infernetFor(user.coinpay_sub), req: parsed, approvals, events: raw?.plugin_events === true, signal }); - return res.body; - } - const up = await forwardChat(infernetFor(user.coinpay_sub), parsed, signal); - const type = up.headers.get("content-type") ?? ""; - if (!up.ok || !type.includes("event-stream")) { - const text = await up.text().catch(() => ""); - let message = `the c0mpute network is restarting or busy (upstream HTTP ${up.status}); try again in a minute`; - try { - message = JSON.parse(text).error?.message ?? message; - } catch {} - throw new Error(message); - } - return up.body; - }); - return new Response(runStream(run), { headers: { ...SSE_HEADERS, "x-c0mpute-run": run.id } }); + // Kept (sealed) with the run so any server can continue it after a deploy. + const runs = runManager(); + const run = await runs.start({ userId: user.id, coinpaySub: user.coinpay_sub ?? null, body: raw, withPlugins }); + return new Response(runs.stream(run.id), { headers: { ...SSE_HEADERS, "x-c0mpute-run": run.id } }); } catch (e) { if (e instanceof RunLimitError) return err(429, e.message, { "retry-after": "10" }); throw e; diff --git a/apps/web/src/app/api/v1/chat/runs/[id]/route.ts b/apps/web/src/app/api/v1/chat/runs/[id]/route.ts index ddb99b37..f269fd0d 100644 --- a/apps/web/src/app/api/v1/chat/runs/[id]/route.ts +++ b/apps/web/src/app/api/v1/chat/runs/[id]/route.ts @@ -1,4 +1,5 @@ -import { SSE_HEADERS, cancelRun, getRun, runStream } from "@/lib/chat/runs"; +import { runManager } from "@/lib/chat/run-manager"; +import { SSE_HEADERS } from "@/lib/chat/runs"; import { currentUser, json, sameOrigin, unauthorized } from "@/lib/chat/server"; export const dynamic = "force-dynamic"; @@ -6,18 +7,22 @@ export const maxDuration = 300; type Ctx = { params: Promise<{ id: string }> }; const gone = (id: string) => - json({ error: { message: `no run ${id}: it finished more than 15 minutes ago, or the server restarted`, code: 404 } }, 404); + json({ error: { message: `no run ${id}: it finished more than 15 minutes ago, or it is not on this account`, code: 404 } }, 404); -/** GET /api/v1/chat/runs/[?from=]: replay the turn's SSE, then follow it live. */ +/** + * GET /api/v1/chat/runs/[?from=]: replay the turn's SSE, then follow + * it live, from whichever server is running it (or continued it after a deploy). + */ export async function GET(req: Request, { params }: Ctx) { const user = await currentUser(req); if (!user) return unauthorized(); const { id } = await params; - const run = getRun(user.id, id); + const runs = runManager(); + const run = await runs.get(user.id, id); if (!run) return gone(id); const from = Number(new URL(req.url).searchParams.get("from") ?? 0); - return new Response(runStream(run, Number.isFinite(from) ? from : 0), { - headers: { ...SSE_HEADERS, "x-c0mpute-run": run.id, "x-c0mpute-run-status": run.status }, + return new Response(runs.stream(id, Number.isFinite(from) ? from : 0), { + headers: { ...SSE_HEADERS, "x-c0mpute-run": id, "x-c0mpute-run-status": run.status }, }); } @@ -27,6 +32,6 @@ export async function DELETE(req: Request, { params }: Ctx) { if (!user) return unauthorized(); if (!sameOrigin(req)) return json({ error: { message: "cross-site request refused", code: 403 } }, 403); const { id } = await params; - if (!cancelRun(user.id, id)) return gone(id); + if (!(await runManager().cancel(user.id, id))) return gone(id); return json({ id, cancelled: true }); } diff --git a/apps/web/src/app/api/v1/chat/runs/route.ts b/apps/web/src/app/api/v1/chat/runs/route.ts index 94ea9116..b515c9a2 100644 --- a/apps/web/src/app/api/v1/chat/runs/route.ts +++ b/apps/web/src/app/api/v1/chat/runs/route.ts @@ -1,4 +1,4 @@ -import { listRuns } from "@/lib/chat/runs"; +import { runManager } from "@/lib/chat/run-manager"; import { currentUser, json, unauthorized } from "@/lib/chat/server"; export const dynamic = "force-dynamic"; @@ -7,5 +7,5 @@ export const dynamic = "force-dynamic"; export async function GET(req: Request) { const user = await currentUser(req); if (!user) return unauthorized(); - return json({ data: listRuns(user.id) }); + return json({ data: await runManager().list(user.id) }); } diff --git a/apps/web/src/instrumentation.ts b/apps/web/src/instrumentation.ts new file mode 100644 index 00000000..04040f3d --- /dev/null +++ b/apps/web/src/instrumentation.ts @@ -0,0 +1,12 @@ +// Runs once when this Next server starts. Starts the chat run manager so +// answers that a previous server (killed by a deploy) was still generating are +// taken over and finished here, without waiting for anyone to reconnect. +export async function register() { + if (process.env.NEXT_RUNTIME !== "nodejs") return; + try { + const { runManager } = await import("./lib/chat/run-manager"); + runManager(); + } catch (e) { + console.error("chat runs: could not start the run manager", e); + } +} diff --git a/apps/web/src/lib/chat/run-manager.ts b/apps/web/src/lib/chat/run-manager.ts new file mode 100644 index 00000000..db4decfd --- /dev/null +++ b/apps/web/src/lib/chat/run-manager.ts @@ -0,0 +1,26 @@ +// The one RunManager of this server process, shared by the route handlers and +// by instrumentation (which starts it at boot so dead servers' runs are taken +// over without waiting for a request). Kept on globalThis because Next can load +// this module more than once. + +import { convoKey } from "./convo-store"; +import { pluginKey } from "./plugin-store"; +import { chatRunOpener } from "./run-opener"; +import { createRunManager, type RunManager } from "./runs"; +import { cloudSql, env, infernetFor } from "./server"; + +const g = globalThis as unknown as { __c0RunManager?: RunManager }; + +export function runManager(): RunManager { + if (g.__c0RunManager) return g.__c0RunManager; + // Sealed with the conversation key (C0MPUTE_CHAT_KEY or derived from the + // plugin key); a run request holds the prompts. + const key = convoKey(pluginKey(env("C0MPUTE_CHAT_KEY")), pluginKey(env("C0MPUTE_PLUGIN_KEY"))); + const durable = !!key && !!env("SQLITE_CLOUD_URL"); + g.__c0RunManager = createRunManager({ + opener: chatRunOpener((subject) => infernetFor(subject)), + sql: durable ? cloudSql : null, + key, + }); + return g.__c0RunManager; +} diff --git a/apps/web/src/lib/chat/run-opener.ts b/apps/web/src/lib/chat/run-opener.ts new file mode 100644 index 00000000..034640fe --- /dev/null +++ b/apps/web/src/lib/chat/run-opener.ts @@ -0,0 +1,39 @@ +// How a resumable turn talks to the network, on whichever server runs it: the +// first time (from the completions route) or after a restart (recovery). + +import { forwardChat, parseChatRequest, type ChatMessage, type Upstream } from "./completions"; +import { pluginChatResponse } from "./plugin-http"; +import { CONTINUE_PROMPT, type RunOpener } from "./runs"; + +/** The messages for a continued answer: history, the partial answer, then "continue". */ +export function continuationMessages(messages: ChatMessage[], partial: string): ChatMessage[] { + if (!partial.trim()) return messages; + return [...messages, { role: "assistant", content: partial }, { role: "user", content: CONTINUE_PROMPT }]; +} + +export function chatRunOpener(upstreamFor: (subject: string | null) => Upstream): RunOpener { + return async (req, signal, cont) => { + const parsed = parseChatRequest(req.body); + if (typeof parsed === "string") throw new Error(parsed); + const upstream = upstreamFor(req.coinpaySub); + if (req.withPlugins) { + // A plugin turn runs again after a restart. Approvals are dropped then: an + // approved destructive call may already have run, so it asks again. + const approvals = cont ? [] : Array.isArray(req.body.approvals) ? req.body.approvals : []; + const res = await pluginChatResponse({ userId: req.userId, upstream, req: parsed, approvals, events: req.body.plugin_events === true, signal }); + return res.body; + } + const messages = cont ? continuationMessages(parsed.messages, cont.partial) : parsed.messages; + const up = await forwardChat(upstream, { ...parsed, messages }, signal); + const type = up.headers.get("content-type") ?? ""; + if (!up.ok || !type.includes("event-stream")) { + const text = await up.text().catch(() => ""); + let message = `the c0mpute network is restarting or busy (upstream HTTP ${up.status}); try again in a minute`; + try { + message = JSON.parse(text).error?.message ?? message; + } catch {} + throw new Error(message); + } + return up.body; + }; +} diff --git a/apps/web/src/lib/chat/runs.ts b/apps/web/src/lib/chat/runs.ts index 741d7380..61180aeb 100644 --- a/apps/web/src/lib/chat/runs.ts +++ b/apps/web/src/lib/chat/runs.ts @@ -1,11 +1,24 @@ // Resumable chat turns. A turn asked for with `resumable: true` is read to the // end by the server whatever happens to the connection, and its SSE output is -// buffered here under a run id. A client that drops (sleep, wifi, a closed -// tab, ssh dying) reattaches with GET /api/v1/chat/runs/: the replay -// starts at the beginning and then follows the turn live. +// kept under a run id. A client that drops (sleep, wifi, a closed tab, ssh +// dying) reattaches with GET /api/v1/chat/runs/: the replay starts at the +// beginning and then follows the turn live. // -// Runs live in this process's memory: a deploy or restart loses the ones in -// flight (the conversation itself is safe in the account's synced history). +// Durable across deploys: a run's request (sealed) and its output are written +// to the database about once a second, with a heartbeat. A deploy kills the +// process that was generating an answer (and the standby container that served +// the deploy window goes away too), so on boot, and every 15 s, each server +// claims runs whose owner stopped beating and continues them: a plain answer +// picks up from the saved partial text, a plugin turn runs again. Any process +// can replay any run from the database, so a reattach that lands on another +// container still follows it. +// +// Without a database (tests, no sealing key) runs live in memory only. + +import { hostname } from "node:os"; +import { randomUUID } from "node:crypto"; +import { seal, unseal } from "./plugin-store"; +import type { Row, Sql } from "./store"; const enc = new TextEncoder(); @@ -15,42 +28,98 @@ export const RUN_MAX_BYTES = 2 * 1024 * 1024; export const RUN_MAX_ACTIVE = 6; /** A turn that never ends is cut off after this. */ export const RUN_MAX_MS = 20 * 60_000; +/** A run whose owner has not written for this long is taken over. */ +export const RUN_STALE_MS = 20_000; +/** How many times a run is continued after its server died. */ +export const RUN_MAX_ATTEMPTS = 3; + +export const CONTINUE_PROMPT = + "Your previous answer was cut off by a server restart. Continue it from exactly where it stopped: do not repeat anything already written, do not start over, and do not mention the interruption."; export type RunStatus = "running" | "done" | "failed" | "cancelled"; +/** What it takes to (re)open a turn on another process: the request as the client sent it. */ +export type RunRequest = { userId: string; coinpaySub: string | null; body: Record; withPlugins: boolean }; + +/** + * Opens the upstream SSE for a run. `cont` is set when continuing a run whose + * server died: `partial` is the answer text so far (empty for a plugin turn, + * which runs again). + */ +export type RunOpener = (req: RunRequest, signal: AbortSignal, cont?: { partial: string }) => Promise | null>; + +export const RUN_SCHEMA = [ + `CREATE TABLE IF NOT EXISTS chat_runs ( + id TEXT PRIMARY KEY, + user_id TEXT NOT NULL, + status TEXT NOT NULL, + owner TEXT NOT NULL, + heartbeat INTEGER NOT NULL, + started INTEGER NOT NULL, + finished INTEGER, + attempts INTEGER NOT NULL DEFAULT 0, + request TEXT + )`, + `CREATE INDEX IF NOT EXISTS chat_runs_user ON chat_runs (user_id, started)`, + `CREATE INDEX IF NOT EXISTS chat_runs_status ON chat_runs (status, heartbeat)`, + `CREATE TABLE IF NOT EXISTS chat_run_chunks ( + run_id TEXT NOT NULL, + seq INTEGER NOT NULL, + data TEXT NOT NULL, + PRIMARY KEY (run_id, seq) + )`, +]; + +const sse = (o: unknown) => `data: ${JSON.stringify(o)}\n\n`; +const errorEvent = (message: string, code = 502) => sse({ error: { message, type: "c0mpute_error", code } }); + type Listener = (chunk: Uint8Array | null) => void; +/** A run this process owns: its output in memory, and what is not yet written. */ export class Run { status: RunStatus = "running"; finished: number | null = null; readonly chunks: Uint8Array[] = []; bytes = 0; readonly listeners = new Set(); - readonly abort = new AbortController(); + abort = new AbortController(); + /** chunks[0..flushed) are in the database. */ + flushed = 0; + /** seq of the next chunk row. */ + seq = 0; + /** The status in the database differs from this one. */ + dirty = true; + readonly decoder = new TextDecoder(); constructor( readonly id: string, readonly userId: string, readonly started: number, + readonly request: RunRequest | null, ) {} push(chunk: Uint8Array) { - if (this.status !== "running") return; + if (this.status !== "running" || !chunk.byteLength) return; if (this.bytes + chunk.byteLength > RUN_MAX_BYTES) { - this.push(enc.encode(`data: ${JSON.stringify({ error: { message: "the answer passed the 2 MiB resumable-turn cap and was cut off", type: "c0mpute_error", code: 413 } })}\n\ndata: [DONE]\n\n`).slice(0, Math.max(0, RUN_MAX_BYTES - this.bytes))); + this.append(enc.encode(errorEvent("the answer passed the 2 MiB resumable-turn cap and was cut off", 413) + "data: [DONE]\n\n")); this.end("failed"); this.abort.abort(); return; } + this.append(chunk); + } + + private append(chunk: Uint8Array) { this.chunks.push(chunk); this.bytes += chunk.byteLength; for (const l of this.listeners) l(chunk); } - end(status: Exclude) { + end(status: Exclude, at = Date.now()) { if (this.status !== "running") return; this.status = status; - this.finished = Date.now(); + this.finished = at; + this.dirty = true; for (const l of this.listeners) l(null); this.listeners.clear(); } @@ -60,130 +129,386 @@ export class Run { } } -type Registry = { runs: Map }; -// One registry per process, also across Next's module reloads in dev. -const g = globalThis as unknown as { __c0RunRegistry?: Registry }; -const reg: Registry = (g.__c0RunRegistry ??= { runs: new Map() }); - -function sweep(now = Date.now()) { - for (const [id, r] of reg.runs) { - if (r.finished !== null && now - r.finished > RUN_TTL_MS) reg.runs.delete(id); - else if (r.status === "running" && now - r.started > RUN_MAX_MS) { - r.abort.abort(); - r.end("failed"); - } +export class RunLimitError extends Error {} + +export type RunSummary = { id: string; status: RunStatus; started: number; finished: number | null }; + +/** The answer text in an SSE transcript (choices[0].delta.content). */ +export function answerText(transcript: string): string { + let out = ""; + for (const line of transcript.split("\n")) { + const t = line.trim(); + if (!t.startsWith("data:")) continue; + const data = t.slice(5).trim(); + if (data === "[DONE]") break; + try { + const j = JSON.parse(data); + const c = j?.choices?.[0]?.delta?.content; + if (typeof c === "string") out += c; + } catch {} } + return out; } -export class RunLimitError extends Error {} +export function createRunManager(opts: { + opener: RunOpener; + sql?: Sql | null; + /** Seals stored requests (they hold the prompts). No key: memory only. */ + key?: Buffer | null; + owner?: string; + now?: () => number; + flushMs?: number; + /** Start the flush/recovery timers (off in tests, which call tick()). */ + timers?: boolean; +}) { + const owner = opts.owner ?? `${hostname()}:${process.pid}:${randomUUID().slice(0, 8)}`; + const now = opts.now ?? Date.now; + const db = opts.sql && opts.key ? opts.sql : null; + const key = opts.key ?? null; + const runs = new Map(); + const aad = (userId: string, id: string) => `${userId}|${id}|run`; -/** - * Start buffering a turn. `open` gets the run's abort signal (Stop, the cap, - * the time limit) and returns the SSE body to read; the run reads it to the - * end on its own, so the client connection no longer matters. - */ -export async function startRun(userId: string, open: (signal: AbortSignal) => Promise | null>, now = Date.now()): Promise { - sweep(now); - const active = [...reg.runs.values()].filter((r) => r.userId === userId && r.status === "running").length; - if (active >= RUN_MAX_ACTIVE) throw new RunLimitError(`at most ${RUN_MAX_ACTIVE} answers running at once on one account`); - const id = `run_${crypto.randomUUID().replace(/-/g, "")}`; - const run = new Run(id, userId, now); - reg.runs.set(id, run); - run.push(enc.encode(`data: ${JSON.stringify({ object: "c0mpute.run", id })}\n\n`)); - const body = await open(run.abort.signal).catch((e) => { - run.push(enc.encode(`data: ${JSON.stringify({ error: { message: e instanceof Error ? e.message : String(e), type: "c0mpute_error", code: 502 } })}\n\ndata: [DONE]\n\n`)); - run.end(run.abort.signal.aborted ? "cancelled" : "failed"); - return null; - }); - if (body) void pump(run, body); - else if (run.status === "running") run.end("failed"); - return run; -} + let ready: Promise | null = null; + const init = () => + (ready ??= (async () => { + if (!db) return; + for (const q of RUN_SCHEMA) await db.run(q); + })().catch((e) => { + ready = null; + throw e; + })); -async function pump(run: Run, body: ReadableStream) { - const reader = body.getReader(); - const onAbort = () => reader.cancel().catch(() => {}); - run.abort.signal.addEventListener("abort", onAbort); - try { - for (;;) { - const { done, value } = await reader.read(); - if (done) break; - if (value?.byteLength) run.push(value); - if (run.status !== "running") break; + function sweepMemory(t = now()) { + for (const [id, r] of runs) { + if (r.finished !== null && t - r.finished > RUN_TTL_MS && r.flushed === r.chunks.length && !r.dirty) runs.delete(id); + else if (r.status === "running" && t - r.started > RUN_MAX_MS) { + r.abort.abort(); + r.push(enc.encode(errorEvent("the answer took longer than 20 minutes and was stopped") + "data: [DONE]\n\n")); + r.end("failed", t); + } } - run.end(run.abort.signal.aborted ? "cancelled" : "done"); - } catch { - if (run.abort.signal.aborted) { - run.push(enc.encode("data: [DONE]\n\n")); - run.end("cancelled"); - } else { - run.push(enc.encode(`data: ${JSON.stringify({ error: { message: "the serving node or the network dropped the answer", type: "c0mpute_error", code: 502 } })}\n\ndata: [DONE]\n\n`)); - run.end("failed"); + } + + async function pump(run: Run, body: ReadableStream, endOnDone = true) { + const reader = body.getReader(); + const onAbort = () => reader.cancel().catch(() => {}); + run.abort.signal.addEventListener("abort", onAbort); + try { + for (;;) { + const { done, value } = await reader.read(); + if (done) break; + if (value?.byteLength) run.push(value); + if (run.status !== "running") break; + } + if (endOnDone) run.end(run.abort.signal.aborted ? "cancelled" : "done", now()); + } catch { + if (run.abort.signal.aborted) { + run.push(enc.encode("data: [DONE]\n\n")); + run.end("cancelled", now()); + } else { + run.push(enc.encode(errorEvent("the serving node or the network dropped the answer") + "data: [DONE]\n\n")); + run.end("failed", now()); + } + } finally { + run.abort.signal.removeEventListener("abort", onAbort); } - } finally { - run.abort.signal.removeEventListener("abort", onAbort); + void flush(); } -} -/** The run, only for the account that started it. */ -export function getRun(userId: string, id: string): Run | null { - sweep(); - const r = reg.runs.get(id); - return r && r.userId === userId ? r : null; -} + async function open(run: Run, cont?: { partial: string }) { + if (!run.request) return; + const body = await opts.opener(run.request, run.abort.signal, cont).catch((e) => { + run.push(enc.encode(errorEvent(e instanceof Error ? e.message : String(e)) + "data: [DONE]\n\n")); + run.end(run.abort.signal.aborted ? "cancelled" : "failed", now()); + return null; + }); + if (body) void pump(run, body); + else if (run.status === "running") run.end("failed", now()); + } -export function listRuns(userId: string) { - sweep(); - return [...reg.runs.values()].filter((r) => r.userId === userId).sort((a, b) => b.started - a.started).map((r) => r.summary()); -} + // ── persistence ───────────────────────────────────────────────────── + + let flushing: Promise | null = null; + /** Write new output, status changes and heartbeats; pick up cancels from other servers. */ + function flush(): Promise { + if (!db) return Promise.resolve(); + return (flushing ??= (async () => { + try { + await init(); + const t = now(); + for (const r of [...runs.values()]) { + if (r.status === "running" || r.flushed < r.chunks.length || r.dirty) { + // A cancel or a takeover from another server since the last tick. + const [row] = await db.all(`SELECT status, owner FROM chat_runs WHERE id = ?`, [r.id]); + if (row && String(row.owner) !== owner) { + // Another server took this run over (we stalled): stop, it carries on. + r.abort.abort(); + runs.delete(r.id); + continue; + } + if (row && String(row.status) === "cancelled" && r.status === "running") { + r.abort.abort(); + r.push(enc.encode("data: [DONE]\n\n")); + r.end("cancelled", t); + } + const fresh = r.chunks.slice(r.flushed); + for (const c of fresh) { + const text = r.decoder.decode(c, { stream: true }); + if (text) await db.run(`INSERT INTO chat_run_chunks (run_id, seq, data) VALUES (?, ?, ?)`, [r.id, r.seq++, text]); + } + r.flushed += fresh.length; + await db.run(`UPDATE chat_runs SET status = ?, heartbeat = ?, finished = ? WHERE id = ? AND owner = ?`, [r.status, t, r.finished, r.id, owner]); + r.dirty = false; + } + } + } catch (e) { + console.error("chat runs: flush failed", e); + } finally { + flushing = null; + } + })()); + } -/** Stop a running turn (the upstream request is aborted). */ -export function cancelRun(userId: string, id: string): boolean { - const r = getRun(userId, id); - if (!r) return false; - if (r.status === "running") { - r.abort.abort(); - r.push(enc.encode("data: [DONE]\n\n")); - r.end("cancelled"); + async function storedChunks(id: string, afterSeq = -1): Promise<{ seq: number; data: string }[]> { + if (!db) return []; + const rows = await db.all(`SELECT seq, data FROM chat_run_chunks WHERE run_id = ? AND seq > ? ORDER BY seq`, [id, afterSeq]); + return rows.map((r) => ({ seq: Number(r.seq), data: String(r.data) })); } - return true; -} -/** The run's output from byte `from`, then live until it ends. */ -export function runStream(run: Run, from = 0): ReadableStream { - let listener: Listener | null = null; - return new ReadableStream({ - start(ctrl) { + function openRequest(r: Row): RunRequest | null { + if (!key || typeof r.request !== "string") return null; + const plain = unseal(key, r.request, aad(String(r.user_id), String(r.id))); + try { + return plain ? (JSON.parse(plain) as RunRequest) : null; + } catch { + return null; + } + } + + /** + * Take over runs whose server stopped beating, and continue them here. + * Also drops finished runs older than an hour. + */ + async function recover(): Promise { + if (!db) return []; + await init(); + const t = now(); + await db.run(`DELETE FROM chat_run_chunks WHERE run_id IN (SELECT id FROM chat_runs WHERE finished IS NOT NULL AND finished < ?)`, [t - 60 * 60_000]); + await db.run(`DELETE FROM chat_runs WHERE finished IS NOT NULL AND finished < ?`, [t - 60 * 60_000]); + const stale = await db.all(`SELECT * FROM chat_runs WHERE status = 'running' AND heartbeat < ? AND owner <> ?`, [t - RUN_STALE_MS, owner]); + const taken: string[] = []; + for (const r of stale) { + const id = String(r.id); + const userId = String(r.user_id); + const started = Number(r.started); + const attempts = Number(r.attempts ?? 0); + const req = openRequest(r); + const giveUp = !req || attempts >= RUN_MAX_ATTEMPTS || t - started > RUN_MAX_MS; + // Claim: only one server wins (the row must still name the dead owner). + await db.run(`UPDATE chat_runs SET owner = ?, heartbeat = ?, attempts = attempts + 1 WHERE id = ? AND owner = ? AND status = 'running'`, [owner, t, id, String(r.owner)]); + const [mine] = await db.all(`SELECT owner FROM chat_runs WHERE id = ?`, [id]); + if (!mine || String(mine.owner) !== owner) continue; + const chunks = await storedChunks(id); + const run = new Run(id, userId, started, req); + run.seq = chunks.length ? chunks[chunks.length - 1]!.seq + 1 : 0; + for (const c of chunks) { + const b = enc.encode(c.data); + run.chunks.push(b); + run.bytes += b.byteLength; + } + run.flushed = run.chunks.length; + runs.set(id, run); + taken.push(id); + if (giveUp) { + run.push(enc.encode(errorEvent("the server restarted while answering and the answer could not be continued; send it again") + "data: [DONE]\n\n")); + run.end("failed", t); + continue; + } + const transcript = chunks.map((c) => c.data).join(""); + await open(run, { partial: req!.withPlugins ? "" : answerText(transcript) }); + } + if (taken.length) await flush(); + return taken; + } + + let timers: ReturnType[] = []; + if (opts.timers !== false && db) { + const t1 = setInterval(() => void flush(), opts.flushMs ?? 1000); + const t2 = setInterval(() => void recover().catch((e) => console.error("chat runs: recover failed", e)), 15_000); + t1.unref?.(); + t2.unref?.(); + timers = [t1, t2]; + setTimeout(() => void recover().catch((e) => console.error("chat runs: recover failed", e)), 2000).unref?.(); + } + + return { + owner, + durable: !!db, + + /** Start a turn and keep it going whatever happens to the client. */ + async start(req: RunRequest): Promise { + sweepMemory(); + const t = now(); + const localActive = [...runs.values()].filter((r) => r.userId === req.userId && r.status === "running").length; + let active = localActive; + if (db) { + await init(); + const [n] = await db.all(`SELECT COUNT(*) AS n FROM chat_runs WHERE user_id = ? AND status = 'running' AND heartbeat > ?`, [req.userId, t - RUN_STALE_MS]); + active = Math.max(localActive, Number(n?.n ?? 0)); + } + if (active >= RUN_MAX_ACTIVE) throw new RunLimitError(`at most ${RUN_MAX_ACTIVE} answers running at once on one account`); + const id = `run_${randomUUID().replace(/-/g, "")}`; + const run = new Run(id, req.userId, t, req); + runs.set(id, run); + run.push(enc.encode(sse({ object: "c0mpute.run", id }))); + if (db && key) { + await db.run(`INSERT INTO chat_runs (id, user_id, status, owner, heartbeat, started, attempts, request) VALUES (?, ?, 'running', ?, ?, ?, 0, ?)`, [ + id, + req.userId, + owner, + t, + t, + seal(key, JSON.stringify(req), aad(req.userId, id)), + ]); + } + await open(run); + return run; + }, + + /** The run's state, for its owner only (here or in the database). */ + async get(userId: string, id: string): Promise { + sweepMemory(); + const r = runs.get(id); + if (r) return r.userId === userId ? r.summary() : null; + if (!db) return null; + await init(); + const [row] = await db.all(`SELECT id, user_id, status, started, finished FROM chat_runs WHERE id = ?`, [id]); + if (!row || String(row.user_id) !== userId) return null; + return { id, status: String(row.status) as RunStatus, started: Number(row.started), finished: row.finished == null ? null : Number(row.finished) }; + }, + + async list(userId: string): Promise { + sweepMemory(); + const out = new Map(); + if (db) { + await init(); + const rows = await db.all(`SELECT id, status, started, finished FROM chat_runs WHERE user_id = ? AND (finished IS NULL OR finished > ?) ORDER BY started DESC LIMIT 50`, [userId, now() - RUN_TTL_MS]); + for (const r of rows) out.set(String(r.id), { id: String(r.id), status: String(r.status) as RunStatus, started: Number(r.started), finished: r.finished == null ? null : Number(r.finished) }); + } + for (const r of runs.values()) if (r.userId === userId) out.set(r.id, r.summary()); + return [...out.values()].sort((a, b) => b.started - a.started); + }, + + /** Stop a running turn; another server's run stops at its next tick. */ + async cancel(userId: string, id: string): Promise { + const r = runs.get(id); + if (r) { + if (r.userId !== userId) return false; + if (r.status === "running") { + r.abort.abort(); + r.push(enc.encode("data: [DONE]\n\n")); + r.end("cancelled", now()); + void flush(); + } + return true; + } + if (!db) return false; + await init(); + const [row] = await db.all(`SELECT user_id, status FROM chat_runs WHERE id = ?`, [id]); + if (!row || String(row.user_id) !== userId) return false; + if (String(row.status) === "running") await db.run(`UPDATE chat_runs SET status = 'cancelled', finished = ? WHERE id = ? AND status = 'running'`, [now(), id]); + return true; + }, + + /** + * The run's output from byte `from`, then live until it ends: from memory + * when this server owns it, else from the database (polled). + */ + stream(id: string, from = 0, pollMs = 750): ReadableStream { + const run = runs.get(id); let skip = Math.max(0, from); - for (const c of run.chunks) { + const emit = (ctrl: ReadableStreamDefaultController, c: Uint8Array) => { if (skip >= c.byteLength) { skip -= c.byteLength; - continue; + return; } ctrl.enqueue(skip ? c.subarray(skip) : c); skip = 0; - } - if (run.status !== "running") { - ctrl.close(); - return; - } - listener = (chunk) => { - try { - if (chunk) ctrl.enqueue(chunk); - else ctrl.close(); - } catch {} }; - run.listeners.add(listener); + if (run) { + let listener: Listener | null = null; + return new ReadableStream({ + start(ctrl) { + for (const c of run.chunks) emit(ctrl, c); + if (run.status !== "running") return ctrl.close(); + listener = (chunk) => { + try { + if (chunk) ctrl.enqueue(chunk); + else ctrl.close(); + } catch {} + }; + run.listeners.add(listener); + }, + cancel() { + if (listener) run.listeners.delete(listener); + }, + }); + } + // Another server owns it (or owned it before a restart): follow the database. + let stopped = false; + return new ReadableStream({ + async start(ctrl) { + let last = -1; + const deadline = now() + RUN_MAX_MS; + try { + for (;;) { + for (const c of await storedChunks(id, last)) { + emit(ctrl, enc.encode(c.data)); + last = c.seq; + } + const [row] = db ? await db.all(`SELECT status FROM chat_runs WHERE id = ?`, [id]) : []; + const done = !row || String(row.status) !== "running"; + if (done) { + // A cancel recorded here never reaches the owner's stream: end it cleanly. + const tail = await storedChunks(id, last); + for (const c of tail) emit(ctrl, enc.encode(c.data)); + if (row && String(row.status) === "cancelled" && !tail.length) ctrl.enqueue(enc.encode("data: [DONE]\n\n")); + break; + } + if (stopped || now() > deadline) break; + await new Promise((r) => setTimeout(r, pollMs)); + if (stopped) break; + } + } catch {} + try { + ctrl.close(); + } catch {} + }, + cancel() { + stopped = true; + }, + }); + }, + + flush, + recover, + /** Tests: the run this process owns. */ + local(id: string): Run | undefined { + return runs.get(id); }, - cancel() { - if (listener) run.listeners.delete(listener); + /** Tests: die without a word, like a SIGKILL (no flush, no status change). */ + crash() { + for (const t of timers) clearInterval(t); + for (const r of runs.values()) r.abort.abort(); + runs.clear(); }, - }); + stop() { + for (const t of timers) clearInterval(t); + timers = []; + }, + }; } -export const SSE_HEADERS = { "content-type": "text/event-stream", "cache-control": "no-store", "x-accel-buffering": "no" }; +export type RunManager = ReturnType; -/** Tests only. */ -export function _resetRuns() { - reg.runs.clear(); -} +export const SSE_HEADERS = { "content-type": "text/event-stream", "cache-control": "no-store", "x-accel-buffering": "no" }; diff --git a/apps/web/src/lib/chat/server.ts b/apps/web/src/lib/chat/server.ts index e256c07e..27d0ed4a 100644 --- a/apps/web/src/lib/chat/server.ts +++ b/apps/web/src/lib/chat/server.ts @@ -31,7 +31,7 @@ async function query(q: string, params: unknown[] = []) { return await db().sql({ query: q, parameters: params as never[] }); } } -const cloudSql: Sql = { +export const cloudSql: Sql = { all: async (q, p) => { const r = await query(q, p); return Array.isArray(r) ? (r.map((row) => ({ ...row })) as Record[]) : []; diff --git a/apps/web/src/lib/chat/sync.test.ts b/apps/web/src/lib/chat/sync.test.ts index a34b4919..7290a854 100644 --- a/apps/web/src/lib/chat/sync.test.ts +++ b/apps/web/src/lib/chat/sync.test.ts @@ -1,7 +1,8 @@ -import { beforeEach, describe, expect, test } from "bun:test"; +import { describe, expect, test } from "bun:test"; import { Database } from "bun:sqlite"; import { ConvoError, convoKey, createConvoStore, parseConversation } from "./convo-store"; -import { _resetRuns, cancelRun, getRun, listRuns, runStream, startRun } from "./runs"; +import { answerText, CONTINUE_PROMPT, createRunManager, RUN_STALE_MS, type RunOpener, type RunRequest } from "./runs"; +import { continuationMessages } from "./run-opener"; import type { Sql } from "./store"; import { adopt, slimForSync, syncPlan } from "./sync-merge"; import { handleRpc } from "./mcp"; @@ -91,67 +92,219 @@ describe("conversation store", () => { const enc = new TextEncoder(); const text = async (s: ReadableStream) => new Response(s).text(); -describe("resumable runs", () => { - beforeEach(() => _resetRuns()); +/** A controllable upstream: push SSE text, end it, or see the abort. */ +function fakeUpstream() { + let push: (t: string) => void = () => {}; + let close: () => void = () => {}; + let aborted = false; + const body = (signal: AbortSignal) => + new ReadableStream({ + start(c) { + push = (t) => { + try { + c.enqueue(enc.encode(t)); + } catch {} + }; + close = () => { + try { + c.close(); + } catch {} + }; + signal.addEventListener("abort", () => { + aborted = true; + try { + c.error(new DOMException("aborted", "AbortError")); + } catch {} + }); + }, + }); + return { body, push: (t: string) => push(t), close: () => close(), aborted: () => aborted }; +} +const delta = (t: string) => `data: {"choices":[{"delta":{"content":${JSON.stringify(t)}}}]}\n\n`; +const req = (over: Partial = {}): RunRequest => ({ userId: "u", coinpaySub: "sub", body: { messages: [{ role: "user", content: "tell me a story" }], stream: true, resumable: true }, withPlugins: false, ...over }); +const tick = () => new Promise((r) => setTimeout(r, 5)); +describe("resumable runs (in memory)", () => { test("a turn keeps running with no reader; a late reader replays it all then follows live", async () => { - let push: (s: string) => void = () => {}; - let close: () => void = () => {}; - const run = await startRun("u", async () => - new ReadableStream({ - start(c) { - push = (s) => c.enqueue(enc.encode(s)); - close = () => c.close(); - }, - }), - ); - push('data: {"choices":[{"delta":{"content":"Hel"}}]}\n\n'); - await new Promise((r) => setTimeout(r, 5)); - // First reader drops at once (a closed tab). - const first = runStream(run).getReader(); + const up = fakeUpstream(); + const m = createRunManager({ opener: async (_r, signal) => up.body(signal), timers: false }); + const run = await m.start(req()); + up.push(delta("Hel")); + await tick(); + const first = m.stream(run.id).getReader(); await first.read(); await first.cancel(); - const late = text(runStream(run)); - push('data: {"choices":[{"delta":{"content":"lo"}}]}\n\ndata: [DONE]\n\n'); - close(); + const late = text(m.stream(run.id)); + up.push(delta("lo") + "data: [DONE]\n\n"); + up.close(); const body = await late; expect(body.startsWith(`data: {"object":"c0mpute.run","id":"${run.id}"}`)).toBe(true); - expect(body).toContain('"Hel"'); - expect(body).toContain('"lo"'); - expect(body.trim().endsWith("[DONE]")).toBe(true); + expect(answerText(body)).toBe("Hello"); expect(run.status).toBe("done"); - // Replay after the end, and from a byte offset. - expect(await text(runStream(run))).toBe(body); - expect(await text(runStream(run, 10))).toBe(body.slice(10)); + expect(await text(m.stream(run.id))).toBe(body); + expect(await text(m.stream(run.id, 10))).toBe(body.slice(10)); }); test("only the owner sees a run; cancel aborts the upstream", async () => { - let aborted = false; - const run = await startRun("u", async (signal) => { - signal.addEventListener("abort", () => (aborted = true)); - return new ReadableStream({ start() {} }); - }); - expect(getRun("other", run.id)).toBeNull(); - expect(cancelRun("other", run.id)).toBe(false); - expect(listRuns("u").map((r) => r.status)).toEqual(["running"]); - expect(cancelRun("u", run.id)).toBe(true); - expect(aborted).toBe(true); + const up = fakeUpstream(); + const m = createRunManager({ opener: async (_r, signal) => up.body(signal), timers: false }); + const run = await m.start(req()); + expect(await m.get("other", run.id)).toBeNull(); + expect(await m.cancel("other", run.id)).toBe(false); + expect((await m.list("u")).map((r) => r.status)).toEqual(["running"]); + expect(await m.cancel("u", run.id)).toBe(true); + await tick(); + expect(up.aborted()).toBe(true); expect(run.status).toBe("cancelled"); - expect((await text(runStream(run))).trim().endsWith("[DONE]")).toBe(true); + expect((await text(m.stream(run.id))).trim().endsWith("[DONE]")).toBe(true); }); test("an upstream that fails to open ends the run with an error a client can show", async () => { - const run = await startRun("u", async () => { - throw new Error("no provider node"); + const m = createRunManager({ + opener: async () => { + throw new Error("no provider node"); + }, + timers: false, }); + const run = await m.start(req()); expect(run.status).toBe("failed"); - expect(await text(runStream(run))).toContain("no provider node"); + expect(await text(m.stream(run.id))).toContain("no provider node"); }); test("too many running turns on one account is refused", async () => { - for (let i = 0; i < 6; i++) await startRun("busy", async () => new ReadableStream({ start() {} })); - expect(startRun("busy", async () => null)).rejects.toThrow(/at most 6/); - await startRun("someone-else", async () => new ReadableStream({ start() {} })); + const m = createRunManager({ opener: async (_r, signal) => fakeUpstream().body(signal), timers: false }); + for (let i = 0; i < 6; i++) await m.start(req({ userId: "busy" })); + expect(m.start(req({ userId: "busy" }))).rejects.toThrow(/at most 6/); + await m.start(req({ userId: "someone-else" })); + }); +}); + +describe("resumable runs survive a deploy (durable)", () => { + /** Two servers on one database, a clock both share. */ + function cluster() { + const { sql } = memSql(); + let t = 1_000_000; + const now = () => t; + const calls: { server: string; cont?: { partial: string }; req: RunRequest }[] = []; + const ups: Record[]> = { a: [], b: [] }; + const server = (name: "a" | "b") => + createRunManager({ + sql, + key: KEY, + owner: name, + now, + timers: false, + opener: (async (r, signal, cont) => { + calls.push({ server: name, cont, req: r }); + const u = fakeUpstream(); + ups[name].push(u); + return u.body(signal); + }) as RunOpener, + }); + return { sql, a: server("a"), b: server("b"), calls, ups, advance: (ms: number) => (t += ms) }; + } + + test("a server killed mid-answer: another takes the run over and continues from the partial text", async () => { + const c = cluster(); + const run = await c.a.start(req()); + c.ups.a[0]!.push(delta("Once upon a time, ")); + await tick(); + await c.a.flush(); + c.ups.a[0]!.push(delta("never saved")); + c.a.crash(); // SIGKILL: no flush, no status change + + // Not stale yet: nothing happens. + expect(await c.b.recover()).toEqual([]); + c.advance(RUN_STALE_MS + 1); + expect(await c.b.recover()).toEqual([run.id]); + // b continued it with the saved partial answer (not the unsaved bit). + expect(c.calls.at(-1)).toMatchObject({ server: "b", cont: { partial: "Once upon a time, " } }); + expect(continuationMessages([{ role: "user", content: "tell me a story" }], "Once upon a time, ")).toEqual([ + { role: "user", content: "tell me a story" }, + { role: "assistant", content: "Once upon a time, " }, + { role: "user", content: CONTINUE_PROMPT }, + ]); + // A client reattaching to b gets the whole answer, seamlessly. + const replay = text(c.b.stream(run.id)); + c.ups.b[0]!.push(delta("there was a node.") + "data: [DONE]\n\n"); + c.ups.b[0]!.close(); + const body = await replay; + expect(answerText(body)).toBe("Once upon a time, there was a node."); + await tick(); + await c.b.flush(); + expect((await c.b.get("u", run.id))?.status).toBe("done"); + // The request was stored sealed: no prompt text in the database. + const raw = (await c.sql.all("SELECT request FROM chat_runs WHERE id = ?", [run.id]))[0]!.request as string; + expect(raw).not.toContain("tell me a story"); + }); + + test("only one server wins the takeover", async () => { + const c = cluster(); + const run = await c.a.start(req()); + await c.a.flush(); + c.a.crash(); + c.advance(RUN_STALE_MS + 1); + const third = createRunManager({ sql: c.sql, key: KEY, owner: "c", now: () => 1_000_000 + RUN_STALE_MS + 1, timers: false, opener: async (_r, s) => fakeUpstream().body(s) }); + const [wonB, wonC] = await Promise.all([c.b.recover(), third.recover()]); + expect([...wonB, ...wonC]).toEqual([run.id]); + }); + + test("a reattach on a server that does not own the run follows it from the database", async () => { + const c = cluster(); + const run = await c.a.start(req()); + c.ups.a[0]!.push(delta("from a ")); + await tick(); + await c.a.flush(); + expect(await c.b.get("u", run.id)).toMatchObject({ status: "running" }); + expect(await c.b.get("intruder", run.id)).toBeNull(); + const follower = text(c.b.stream(run.id, 0, 5)); + c.ups.a[0]!.push(delta("and more") + "data: [DONE]\n\n"); + c.ups.a[0]!.close(); + await tick(); + await c.a.flush(); + expect(answerText(await follower)).toBe("from a and more"); + expect((await c.b.list("u")).map((r) => r.status)).toEqual(["done"]); + }); + + test("a cancel that lands on another server stops the owner at its next tick", async () => { + const c = cluster(); + const run = await c.a.start(req()); + await c.a.flush(); + expect(await c.b.cancel("u", run.id)).toBe(true); + await c.a.flush(); + await tick(); + expect(c.ups.a[0]!.aborted()).toBe(true); + expect(c.a.local(run.id)?.status).toBe("cancelled"); + }); + + test("a plugin turn runs again after a restart, without the approvals it was sent", async () => { + const c = cluster(); + const run = await c.a.start(req({ withPlugins: true, body: { messages: [{ role: "user", content: "delete it" }], stream: true, approvals: ["tok"] } })); + await c.a.flush(); + c.a.crash(); + c.advance(RUN_STALE_MS + 1); + await c.b.recover(); + expect(c.calls.at(-1)).toMatchObject({ server: "b", cont: { partial: "" } }); + expect(c.calls.at(-1)!.req.withPlugins).toBe(true); + expect(run.id).toBeTruthy(); + }); + + test("a run that keeps dying is given up with a message instead of looping", async () => { + const c = cluster(); + const run = await c.a.start(req()); + await c.a.flush(); + c.a.crash(); + let srv = c.b; + for (let i = 0; i < 3; i++) { + c.advance(RUN_STALE_MS + 1); + expect(await srv.recover()).toEqual([run.id]); + srv.crash(); + srv = createRunManager({ sql: c.sql, key: KEY, owner: `x${i}`, now: () => 0, timers: false, opener: async (_r, s) => fakeUpstream().body(s) }); + } + const last = createRunManager({ sql: c.sql, key: KEY, owner: "final", now: () => 1_000_000 + 10 * (RUN_STALE_MS + 1), timers: false, opener: async (_r, s) => fakeUpstream().body(s) }); + expect(await last.recover()).toEqual([run.id]); + expect(await text(last.stream(run.id))).toContain("could not be continued"); + expect((await last.get("u", run.id))?.status).toBe("failed"); }); }); From d19bd39b0a83e9cf86646162bbdc09767c20431b Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Thu, 12 Nov 2026 08:49:31 +0000 Subject: [PATCH 2/3] test: shared clock --- apps/web/src/lib/chat/sync.test.ts | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/apps/web/src/lib/chat/sync.test.ts b/apps/web/src/lib/chat/sync.test.ts index 7290a854..245a0c84 100644 --- a/apps/web/src/lib/chat/sync.test.ts +++ b/apps/web/src/lib/chat/sync.test.ts @@ -201,7 +201,7 @@ describe("resumable runs survive a deploy (durable)", () => { return u.body(signal); }) as RunOpener, }); - return { sql, a: server("a"), b: server("b"), calls, ups, advance: (ms: number) => (t += ms) }; + return { sql, a: server("a"), b: server("b"), calls, ups, now, advance: (ms: number) => (t += ms) }; } test("a server killed mid-answer: another takes the run over and continues from the partial text", async () => { @@ -299,9 +299,10 @@ describe("resumable runs survive a deploy (durable)", () => { c.advance(RUN_STALE_MS + 1); expect(await srv.recover()).toEqual([run.id]); srv.crash(); - srv = createRunManager({ sql: c.sql, key: KEY, owner: `x${i}`, now: () => 0, timers: false, opener: async (_r, s) => fakeUpstream().body(s) }); + srv = createRunManager({ sql: c.sql, key: KEY, owner: `x${i}`, now: c.now, timers: false, opener: async (_r, s) => fakeUpstream().body(s) }); } - const last = createRunManager({ sql: c.sql, key: KEY, owner: "final", now: () => 1_000_000 + 10 * (RUN_STALE_MS + 1), timers: false, opener: async (_r, s) => fakeUpstream().body(s) }); + c.advance(RUN_STALE_MS + 1); + const last = createRunManager({ sql: c.sql, key: KEY, owner: "final", now: c.now, timers: false, opener: async (_r, s) => fakeUpstream().body(s) }); expect(await last.recover()).toEqual([run.id]); expect(await text(last.stream(run.id))).toContain("could not be continued"); expect((await last.get("u", run.id))?.status).toBe("failed"); From 37430a66b1df164e36176858689377b800f5f733 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Thu, 12 Nov 2026 08:50:00 +0000 Subject: [PATCH 3/3] chore: v0.3.59 Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 34 +++++++++++++++++----------------- Cargo.toml | 2 +- 2 files changed, 18 insertions(+), 18 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e8eaa89a..14e46037 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -486,7 +486,7 @@ checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" [[package]] name = "c0mpute-api" -version = "0.3.58" +version = "0.3.59" dependencies = [ "anyhow", "c0mpute-proto", @@ -501,7 +501,7 @@ dependencies = [ [[package]] name = "c0mpute-bench" -version = "0.3.58" +version = "0.3.59" dependencies = [ "anyhow", "blake3", @@ -512,7 +512,7 @@ dependencies = [ [[package]] name = "c0mpute-cli" -version = "0.3.58" +version = "0.3.59" dependencies = [ "anyhow", "async-trait", @@ -549,7 +549,7 @@ dependencies = [ [[package]] name = "c0mpute-core" -version = "0.3.58" +version = "0.3.59" dependencies = [ "aes-gcm", "anyhow", @@ -603,7 +603,7 @@ dependencies = [ [[package]] name = "c0mpute-doctor" -version = "0.3.58" +version = "0.3.59" dependencies = [ "anyhow", "serde", @@ -615,7 +615,7 @@ dependencies = [ [[package]] name = "c0mpute-envelope" -version = "0.3.58" +version = "0.3.59" dependencies = [ "blake3", "bs58", @@ -628,7 +628,7 @@ dependencies = [ [[package]] name = "c0mpute-gateway" -version = "0.3.58" +version = "0.3.59" dependencies = [ "anyhow", "axum", @@ -650,7 +650,7 @@ dependencies = [ [[package]] name = "c0mpute-market" -version = "0.3.58" +version = "0.3.59" dependencies = [ "c0mpute-envelope", "serde", @@ -660,7 +660,7 @@ dependencies = [ [[package]] name = "c0mpute-net" -version = "0.3.58" +version = "0.3.59" dependencies = [ "anyhow", "async-trait", @@ -681,7 +681,7 @@ dependencies = [ [[package]] name = "c0mpute-placement" -version = "0.3.58" +version = "0.3.59" dependencies = [ "anyhow", "async-trait", @@ -702,7 +702,7 @@ dependencies = [ [[package]] name = "c0mpute-proto" -version = "0.3.58" +version = "0.3.59" dependencies = [ "blake3", "hex", @@ -714,7 +714,7 @@ dependencies = [ [[package]] name = "c0mpute-sealed" -version = "0.3.58" +version = "0.3.59" dependencies = [ "aes", "aes-gcm", @@ -736,7 +736,7 @@ dependencies = [ [[package]] name = "c0mpute-secure-chat" -version = "0.3.58" +version = "0.3.59" dependencies = [ "aes-gcm", "anyhow", @@ -759,7 +759,7 @@ dependencies = [ [[package]] name = "c0mpute-store" -version = "0.3.58" +version = "0.3.59" dependencies = [ "anyhow", "blake3", @@ -778,7 +778,7 @@ dependencies = [ [[package]] name = "c0mpute-transcode" -version = "0.3.58" +version = "0.3.59" dependencies = [ "anyhow", "c0mpute-proto", @@ -791,7 +791,7 @@ dependencies = [ [[package]] name = "c0mpute-update" -version = "0.3.58" +version = "0.3.59" dependencies = [ "anyhow", "hex", @@ -806,7 +806,7 @@ dependencies = [ [[package]] name = "c0mpute-verify" -version = "0.3.58" +version = "0.3.59" dependencies = [ "anyhow", "blake3", diff --git a/Cargo.toml b/Cargo.toml index dc22616d..b313a438 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,7 @@ members = [ ] [workspace.package] -version = "0.3.58" +version = "0.3.59" edition = "2024" license = "MIT" repository = "https://github.com/profullstack/c0mpute"