Add workspaces, computers, bot memory and the bots queue

- Schema: workspaces (+ workspace_bots, workspace_computers), containers,
  computers, bot_computers, bots.memory, workspaces.system_prompt, and
  bots_queue as a bun-sqlite isQueue table
- Main computer (and its container record) ensured with the coordinator on
  signup, seed and /admin/bots; the coordinator's access to it is locked
- Effective computer access: direct grants plus workspace computers
- Full bot prompt: core instructions, workspace prompts, then memory;
  bots can rewrite their memory and system prompt
- Queue helpers on the native BunSQLite.queue API, plus a handler that
  heartbeats, serializes per computer, recovers after restarts, reports
  lost jobs, supports cancelling and pushes outcomes over the WebSocket
- Enable SQLite foreign keys at startup so schema cascades fire
- Prune finished jobs daily

Co-Authored-By: Claude Opus 5.5 <[email protected]>

src/bots-queue/bot-job-fatal-error.ts

@@ -0,0 +1,10 @@
+/**
+ * Throw from a job runner when retrying can't help (bad parameters, no
+ * access). The job fails at once instead of using its remaining attempts.
+ */
+export default class BotJobFatalError extends Error {
+    constructor(message: string) {
+        super(message);
+        this.name = "BotJobFatalError";
+    }
+}

src/bots-queue/index.ts

@@ -0,0 +1,65 @@
+import bunext from "@moduletrace/bunext";
+import { AppData } from "@/src/data/app-data";
+import claimBotJobs from "@/src/functions/backend/db/bots-queue/claim-bot-jobs";
+import releaseWorkerBotJobs from "@/src/functions/backend/db/bots-queue/release-worker-bot-jobs";
+import { logError } from "@/src/functions/backend/logger";
+import processBotJob from "./process-bot-job";
+import reportLostBotJobs from "./report-lost-bot-jobs";
+
+const worker = AppData.BotQueueWorker;
+let running = 0;
+
+/**
+ * Starts the `bots_queue` handler: releases jobs a restart cut off, then
+ * polls every `AppData.BotQueuePollMs`, reporting lost jobs and claiming
+ * up to `AppData.BotQueueConcurrency` at once. Gated in production to the
+ * `MAIN_SERVER` container, like cron.
+ */
+export default async function botsQueue() {
+    if (process.env.NODE_ENV == "production" && !process.env.MAIN_SERVER) {
+        return;
+    }
+
+    bunext.bunextLog.watch(`Running Bots Queue ...`);
+
+    const released = await releaseWorkerBotJobs({ worker });
+
+    if (released) {
+        bunext.bunextLog.info(`Bots queue: released ${released} cut-off job(s)`);
+    }
+
+    poll();
+}
+
+async function poll() {
+    try {
+        await reportLostBotJobs();
+
+        const free = AppData.BotQueueConcurrency - running;
+
+        if (free > 0) {
+            const jobs = await claimBotJobs({ worker, limit: free });
+
+            for (const job of jobs) {
+                running++;
+
+                processBotJob({ job, worker })
+                    .catch((error) =>
+                        logError({
+                            title: `Bot job #${job.id} crashed`,
+                            error,
+                        }).catch(() => {}),
+                    )
+                    .finally(() => {
+                        running--;
+                    });
+            }
+        }
+    } catch (error) {
+        await logError({ title: "Bots queue poll failed", error }).catch(
+            () => {},
+        );
+    } finally {
+        setTimeout(poll, AppData.BotQueuePollMs);
+    }
+}

src/bots-queue/parse-job-json.test.ts

@@ -0,0 +1,16 @@
+import { describe, expect, test } from "bun:test";
+import parseJobJson from "./parse-job-json";
+
+describe("parseJobJson", () => {
+    test("reads a JSON object", () => {
+        expect(parseJobJson({ text: `{"step":2}` })).toEqual({ step: 2 });
+    });
+
+    test("is null for empty, invalid or non-object JSON", () => {
+        expect(parseJobJson({})).toBeNull();
+        expect(parseJobJson({ text: "" })).toBeNull();
+        expect(parseJobJson({ text: "{nope" })).toBeNull();
+        expect(parseJobJson({ text: "[1]" })).toBeNull();
+        expect(parseJobJson({ text: "3" })).toBeNull();
+    });
+});

src/bots-queue/parse-job-json.ts

@@ -0,0 +1,22 @@
+type Params = {
+    text?: string | null;
+};
+
+/**
+ * Reads a job's `params` or `checkpoint` column. Null when empty or not a
+ * JSON object.
+ */
+export default function parseJobJson({
+    text,
+}: Params): Record<string, unknown> | null {
+    if (!text) return null;
+
+    try {
+        const value = JSON.parse(text);
+        return value && typeof value == "object" && !Array.isArray(value)
+            ? value
+            : null;
+    } catch {
+        return null;
+    }
+}

src/bots-queue/process-bot-job.ts

@@ -0,0 +1,126 @@
+import type { BUN_SQLITE_COBBLE_BOTS_QUEUE } from "@/db/types/db";
+import { AppData } from "@/src/data/app-data";
+import setBotFace from "@/src/functions/backend/db/bots/set-bot-face";
+import completeBotJob from "@/src/functions/backend/db/bots-queue/complete-bot-job";
+import failBotJob from "@/src/functions/backend/db/bots-queue/fail-bot-job";
+import updateBotJob from "@/src/functions/backend/db/bots-queue/update-bot-job";
+import { logError } from "@/src/functions/backend/logger";
+import createKeyLock from "@/src/utils/create-key-lock";
+import BotJobFatalError from "./bot-job-fatal-error";
+import pushBotNotice from "./push-bot-notice";
+import runBotJob from "./run-bot-job";
+import parseJobJson from "./parse-job-json";
+import type { BotJobContext, BotJobRunner } from "./types";
+
+/**
+ * One job per computer at a time, across all running jobs.
+ */
+const computer_lock = createKeyLock();
+
+/**
+ * Renew the lock well inside the timeout, so one slow write can't lose it.
+ */
+const HEARTBEAT_MS = Math.floor(AppData.BotQueueLockTimeoutMs / 4);
+
+type Params = {
+    job: BUN_SQLITE_COBBLE_BOTS_QUEUE;
+    worker: string;
+    /**
+     * Defaults to `runBotJob`.
+     */
+    runner?: BotJobRunner;
+};
+
+/**
+ * Runs one claimed job to the end: keeps its lock alive, waits for its
+ * computer, runs it, then completes or fails it and pushes the outcome to
+ * the user. If the job is cancelled or lost mid-way, its result is
+ * dropped.
+ */
+export default async function processBotJob({
+    job,
+    worker,
+    runner = runBotJob,
+}: Params) {
+    if (!job.id || !job.bot_id) return;
+
+    const job_id = job.id;
+    const bot_id = job.bot_id;
+    const controller = new AbortController();
+
+    const update: BotJobContext["update"] = async (params) => {
+        if (controller.signal.aborted) return false;
+
+        const held = await updateBotJob({ job_id, worker, ...params });
+
+        if (!held) controller.abort();
+
+        return held;
+    };
+
+    const heartbeat = setInterval(() => {
+        update({}).catch(() => {});
+    }, HEARTBEAT_MS);
+
+    try {
+        /**
+         * The face changes once the job really starts, after any wait for
+         * its computer.
+         */
+        const run = async () => {
+            await pushBotNotice({
+                bot_id,
+                bot: await setBotFace({
+                    bot_id,
+                    face: "working",
+                    status_line: job.title || "",
+                }),
+            });
+
+            return await runner({
+                job,
+                params: parseJobJson({ text: job.params }) || {},
+                checkpoint: parseJobJson({ text: job.checkpoint }),
+                signal: controller.signal,
+                update,
+            });
+        };
+
+        const outcome = job.computer_id
+            ? await computer_lock.run({ key: job.computer_id, task: run })
+            : await run();
+
+        if (controller.signal.aborted) return;
+
+        await pushBotNotice({
+            bot_id,
+            ...(await completeBotJob({
+                job_id,
+                worker,
+                result: outcome.result,
+                needs_you: outcome.needs_you,
+            })),
+        });
+    } catch (error: unknown) {
+        if (controller.signal.aborted) return;
+
+        const failed = await failBotJob({
+            job_id,
+            worker,
+            error: error instanceof Error ? error.message : String(error),
+            fatal: error instanceof BotJobFatalError,
+        }).catch(async (fail_error) => {
+            await logError({
+                title: `Bot job #${job_id} couldn't be failed`,
+                error: fail_error,
+            }).catch(() => {});
+            return null;
+        });
+
+        if (failed?.notice) {
+            await pushBotNotice({ bot_id, ...failed.notice });
+        }
+    } finally {
+        clearInterval(heartbeat);
+    }
+}

src/bots-queue/push-bot-notice.ts

@@ -0,0 +1,30 @@
+import type {
+    BUN_SQLITE_COBBLE_BOT_MESSAGES,
+    BUN_SQLITE_COBBLE_BOTS,
+} from "@/db/types/db";
+import pushToBotOwner from "./push-to-bot-owner";
+
+type Params = {
+    bot_id: number | string;
+    message?: BUN_SQLITE_COBBLE_BOT_MESSAGES | null;
+    bot?: Pick<BUN_SQLITE_COBBLE_BOTS, "id" | "face" | "status_line"> | null;
+};
+
+/**
+ * Pushes a new bot message and/or face change to the owner's open tabs.
+ */
+export default async function pushBotNotice({ bot_id, message, bot }: Params) {
+    if (message) {
+        await pushToBotOwner({
+            bot_id,
+            data: { event: "server:bot-message", data: message },
+        });
+    }
+
+    if (bot) {
+        await pushToBotOwner({
+            bot_id,
+            data: { event: "server:bot-updated", data: bot },
+        });
+    }
+}

src/bots-queue/push-to-bot-owner.ts

@@ -0,0 +1,26 @@
+import BunSQLite from "@moduletrace/bun-sqlite";
+import type { BUN_SQLITE_COBBLE_BOTS } from "@/db/types/db";
+import type { WebSocketDataType } from "@/src/types";
+import { sendToUser } from "@/src/websocket/(functions)/send-to-user";
+
+type Params = {
+    bot_id: number | string;
+    data: WebSocketDataType;
+};
+
+/**
+ * Live update to the bot owner's open tabs. Best effort: nothing is queued
+ * for offline users, who see the change on their next page load.
+ */
+export default async function pushToBotOwner({ bot_id, data }: Params) {
+    const bot = (
+        await BunSQLite.sql<BUN_SQLITE_COBBLE_BOTS>({
+            sql: `SELECT user_id FROM bots WHERE id = ? LIMIT 1`,
+            values: [bot_id],
+        })
+    ).payload?.[0];
+
+    if (!bot?.user_id) return;
+
+    sendToUser({ user_id: bot.user_id, data });
+}

src/bots-queue/report-lost-bot-jobs.ts

@@ -0,0 +1,22 @@
+import grabUnreportedFailedBotJobs from "@/src/functions/backend/db/bots-queue/grab-unreported-failed-bot-jobs";
+import notifyBotJobFailed from "@/src/functions/backend/db/bots-queue/notify-bot-job-failed";
+import pushBotNotice from "./push-bot-notice";
+
+/**
+ * Tells the user about jobs that ran out of attempts by going silent
+ * (bun-sqlite fails those itself, without a message).
+ */
+export default async function reportLostBotJobs() {
+    const jobs = await grabUnreportedFailedBotJobs();
+
+    for (const job of jobs) {
+        if (!job.bot_id) continue;
+
+        const notice = await notifyBotJobFailed({
+            job,
+            failure_message: `I lost track of "${job.title || "the job"}" and couldn't finish it.`,
+        });
+
+        await pushBotNotice({ bot_id: job.bot_id, ...notice });
+    }
+}

src/bots-queue/run-bot-job.ts

@@ -0,0 +1,14 @@
+import type { BotJobRunner } from "./types";
+import BotJobFatalError from "./bot-job-fatal-error";
+
+/**
+ * Does a job's actual work: the bot follows `job.instructions` with its
+ * full prompt (`grabBotPrompt`), on `job.computer_id` if set, and returns
+ * the message for the user. This is where the model layer plugs in; until
+ * it exists every job fails at once and the user is told.
+ */
+const runBotJob: BotJobRunner = async () => {
+    throw new BotJobFatalError(`Bots can't run queued work yet.`);
+};
+
+export default runBotJob;

src/bots-queue/types.ts

@@ -0,0 +1,34 @@
+import type { BUN_SQLITE_COBBLE_BOTS_QUEUE } from "@/db/types/db";
+
+export type BotJobContext = {
+    job: BUN_SQLITE_COBBLE_BOTS_QUEUE;
+    params: Record<string, unknown>;
+    /**
+     * Saved by an earlier attempt, if the job was picked up again.
+     */
+    checkpoint: Record<string, unknown> | null;
+    /**
+     * Aborted when the job is cancelled or lost to another handler. Stop
+     * work and return; the result is discarded.
+     */
+    signal: AbortSignal;
+    /**
+     * Save progress and a checkpoint. The lock is renewed on a timer
+     * already, so call this for meaningful steps. False once the job is
+     * cancelled or lost.
+     */
+    update: (params: {
+        progress?: string;
+        checkpoint?: Record<string, unknown>;
+    }) => Promise<boolean>;
+};
+
+export type BotJobOutcome = {
+    /**
+     * The completion message for the user (markdown).
+     */
+    result: string;
+    needs_you?: boolean;
+};
+
+export type BotJobRunner = (context: BotJobContext) => Promise<BotJobOutcome>;