feat(chat): resumable turns survive deploys (durable runs, takeover + continue) (v0.3.59)

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"

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;

apps/web/src/app/api/v1/chat/runs/[id]/route.ts

@@ -1,23 +1,28 @@
-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";
 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/<id>[?from=<bytes>]: replay the turn's SSE, then follow it live. */
+/**
+ * GET /api/v1/chat/runs/<id>[?from=<bytes>]: 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 });
 }

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) });
 }

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);
+  }
+}

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;
+}

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;
+  };
+}