Keep bot jobs and harness calls from hanging

src/bots-queue/index.ts

@@ -22,15 +22,34 @@ let timer: ReturnType<typeof setTimeout> | undefined;
  */
 export default async function botsQueue() {
     if (process.env.NODE_ENV == "production" && !process.env.MAIN_SERVER) {
+        /**
+         * Said out loud: a production-mode process without `MAIN_SERVER`
+         * answers requests but never replies, because nothing runs this
+         * queue. `bun run start` sets it.
+         */
+        bunext.bunextLog.info(
+            `Bots queue is off in this process: neither MAIN_SERVER nor development mode is set.`,
+        );
         return;
     }
 
     bunext.bunextLog.watch(`Running Bots Queue ...`);
 
-    const released = await releaseWorkerBotJobs({ worker });
+    /**
+     * Releasing the jobs a restart cut off must never stop the queue: if
+     * it fails, polling still starts, and the lock timeout releases them.
+     */
+    try {
+        const released = await releaseWorkerBotJobs({ worker });
 
-    if (released) {
-        bunext.bunextLog.info(`Bots queue: released ${released} cut-off job(s)`);
+        if (released) {
+            bunext.bunextLog.info(`Bots queue: released ${released} cut-off job(s)`);
+        }
+    } catch (error) {
+        await logError({
+            title: "Bots queue couldn't release cut-off jobs",
+            error,
+        }).catch(() => {});
     }
 
     setBotsQueueWaker({ waker: wake });

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

@@ -7,6 +7,7 @@ import updateBotJob from "@/src/functions/backend/db/bots-queue/update-bot-job";
 import insertBotJobUpdate from "@/src/functions/backend/db/bots-queue/insert-bot-job-update";
 import { logError } from "@/src/functions/backend/logger";
 import createKeyLock from "@/src/utils/create-key-lock";
+import withTimeout from "@/src/utils/with-timeout";
 import HarnessError from "@/src/harness/harness-error";
 import deferBotJob from "@/src/functions/backend/db/bots-queue/defer-bot-job";
 import recordCronJobFailure from "@/src/functions/backend/db/cron-jobs/record-cron-job-failure";
@@ -85,6 +86,13 @@ export default async function processBotJob({
         await pushBotNotice({ bot_id, bot: await setBotFace({ bot_id, user_id, ...face }) });
     };
     const controller = new AbortController();
+    /**
+     * The deadline is the one abort that is a failure, not a cancellation:
+     * the job's error is then the timeout's, not the "Cancelled." an
+     * aborted model request reports.
+     */
+    const timeout_message = `"${job.title || "The job"}" took too long and was stopped.`;
+    let timed_out = false;
 
     const update: BotJobContext["update"] = async (params) => {
         if (controller.signal.aborted) return false;
@@ -137,13 +145,27 @@ export default async function processBotJob({
                 status_line: job.kind == "reply" ? BotReplyJobTitle : BotTaskStatusLine,
             });
 
-            return await runner({
-                job,
-                params,
-                checkpoint: parseJobJson({ text: job.checkpoint }),
+            /**
+             * A deadline per attempt: the heartbeat keeps the lock alive
+             * for as long as this run takes, so without one a call that
+             * never answers would hold the slot forever.
+             */
+            return await withTimeout({
+                timeout_ms: AppData.BotJobAttemptTimeoutMs,
                 signal: controller.signal,
-                update,
-                report,
+                message: timeout_message,
+                on_timeout: () => {
+                    timed_out = true;
+                },
+                task: (run_signal) =>
+                    runner({
+                        job,
+                        params,
+                        checkpoint: parseJobJson({ text: job.checkpoint }),
+                        signal: run_signal,
+                        update,
+                        report,
+                    }),
             });
         };
 
@@ -229,7 +251,11 @@ export default async function processBotJob({
             return;
         }
 
-        const message = error instanceof Error ? error.message : String(error);
+        const message = timed_out
+            ? timeout_message
+            : error instanceof Error
+              ? error.message
+              : String(error);
         const tries = Number(job.attempts) || 1;
 
         const failed = await failBotJob({

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

@@ -1,5 +1,6 @@
 import type { BUN_SQLITE_COBBLE_BOT_MESSAGES } from "@/db/types/db";
 import type { BotFaceChange } from "@/src/functions/backend/db/bots/set-bot-face";
+import { logError } from "@/src/functions/backend/logger";
 import notifyBotMessage from "@/src/push/notify-bot-message";
 import notifyBotFace from "@/src/push/notify-bot-face";
 import pushBotUpdated from "./push-bot-updated";
@@ -23,9 +24,22 @@ type Params = {
  * Pushes a new bot message to the open tabs of the user whose
  * conversation it's in, and a face change to the user it's for
  * (`pushBotUpdated`). A relayed report
- * goes to the conversation it was posted in.
+ * goes to the conversation it was posted in. Best effort: a push that
+ * fails is logged and never fails the job that made the change.
  */
 export default async function pushBotNotice({ bot_id, message, bot, relay }: Params) {
+    try {
+        await push({ bot_id, message, bot, relay });
+    } catch (error) {
+        await logError({
+            title: "Bot notice wasn't pushed",
+            error,
+            metadata: { bot_id, message_id: message?.id || null },
+        }).catch(() => {});
+    }
+}
+
+async function push({ bot_id, message, bot, relay }: Params) {
     if (message) {
         await pushToSessionUser({
             session_id: message.session_id,

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

@@ -1,4 +1,5 @@
 import grabBotAudience from "@/src/functions/backend/db/access/grab-bot-audience";
+import { logError } from "@/src/functions/backend/logger";
 import grabActiveBotTask from "@/src/functions/backend/db/bots/grab-active-bot-task";
 import grabBotFace from "@/src/functions/backend/db/bots/grab-bot-face";
 import { sendToUser } from "@/src/websocket/(functions)/send-to-user";
@@ -29,7 +30,8 @@ type Params = {
 /**
  * Pushes `server:bot-updated` to one user, or everyone who sees the bot
  * (`grabBotAudience`), each with their own face, status line and active
- * task with it, never another user's.
+ * task with it, never another user's. Best effort: a push that fails is
+ * logged and never fails the change that caused it.
  */
 export default async function pushBotUpdated({
     bot_id,
@@ -40,6 +42,24 @@ export default async function pushBotUpdated({
 }: Params) {
     if (!bot_id) return;
 
+    try {
+        await push({ bot_id, user_id, users, computer_id, workspace_ids });
+    } catch (error) {
+        await logError({
+            title: "Bot update wasn't pushed",
+            error,
+            metadata: { bot_id },
+        }).catch(() => {});
+    }
+}
+
+async function push({
+    bot_id,
+    user_id,
+    users,
+    computer_id,
+    workspace_ids,
+}: Params) {
     const audience = user_id ? [user_id] : await grabBotAudience({ bot_id });
 
     for (const viewer_id of audience) {

src/data/app-data.ts

@@ -122,6 +122,14 @@ export const AppData = {
     BotQueueMaxAttempts: 3,
     BotQueueLockTimeoutMs: 10 * 60 * 1000,
     BotQueueBackoffMs: 30 * 1000,
+    /**
+     * One attempt of a job (one model run or task, tools included) gives up
+     * after this long and counts as a failed try, so a call that hangs
+     * without ever answering can't hold the job, its computer or the
+     * conversation forever. The lock is outlived by the heartbeat, so this
+     * is the only thing that stops a live but stuck run.
+     */
+    BotJobAttemptTimeoutMs: 60 * 60 * 1000,
     /**
      * A failed scheduled run is tried again after this, doubling each time
      * (`scheduledRunRetryAt`), so a short outage doesn't sink the run.
@@ -170,6 +178,13 @@ export const AppData = {
      */
     HarnessMaxCandidates: 5,
     HarnessRequestTimeoutMs: 3 * 60 * 1000,
+    /**
+     * One tool call gives up after this long, whatever the tool is (bot
+     * tools, MCP servers, computers, custom tools), so a tool that ignores
+     * its own limits can't stall the run. Its result comes back as an error
+     * the model can work around.
+     */
+    HarnessToolTimeoutMs: 10 * 60 * 1000,
     /**
      * A model that rate limits or fails is skipped for this long, unless
      * the provider says when to come back (capped at the max).

src/functions/backend/docker/ensure-docker-network.ts

@@ -2,12 +2,38 @@ import { AppData } from "@/src/data/app-data";
 import dockerRequest from "./docker-request";
 import grabAppContainer from "./grab-app-container";
 
+/**
+ * Docker can still be coming up when the app container starts (a host
+ * reboot, especially), so a transient failure is retried before the app
+ * gives up on networking.
+ */
+const Attempts = 3;
+const RetryMs = 2000;
+
 /**
  * Makes sure the network computers run on (`AppData.DockerNetworkName`)
  * exists and, when the app runs in docker, that its container is on it,
  * so it reaches computers by name.
  */
 export default async function ensureDockerNetwork() {
+    let last_error: unknown;
+
+    for (let attempt = 1; attempt <= Attempts; attempt++) {
+        try {
+            return await connectNetwork();
+        } catch (error: unknown) {
+            last_error = error;
+
+            if (attempt < Attempts) {
+                await new Promise((resolve) => setTimeout(resolve, RetryMs));
+            }
+        }
+    }
+
+    throw last_error;
+}
+
+async function connectNetwork() {
     const name = AppData.DockerNetworkName;
     const existing = await dockerRequest({ path: `/networks/${encodeURIComponent(name)}` });
 

src/harness/providers/openai-compatible/generate-image.ts

@@ -1,3 +1,4 @@
+import { AppData } from "@/src/data/app-data";
 import HarnessError from "../../harness-error";
 import type { HarnessAdapter, HarnessImageOutput } from "../../types";
 import grabProviderPreset from "../grab-provider-preset";
@@ -54,7 +55,12 @@ const generateImage: HarnessAdapter["generateImage"] = async ({
                 media_type: item.media_type || "image/png",
             });
         } else if (typeof item?.url == "string") {
-            const download = await fetch(item.url, { signal });
+            const download = await fetch(item.url, {
+                signal: AbortSignal.any([
+                    signal,
+                    AbortSignal.timeout(AppData.HarnessRequestTimeoutMs),
+                ]),
+            });
 
             if (download.ok) {
                 images.push({

src/harness/providers/openai-compatible/list-models.ts

@@ -1,4 +1,5 @@
 import guessModelModalities from "../../catalog/guess-model-modalities";
+import HarnessError from "../../harness-error";
 import type { HarnessAdapter, HarnessCatalogModel } from "../../types";
 import grabProviderPreset from "../grab-provider-preset";
 import openRouterModelToCatalog, {
@@ -27,6 +28,18 @@ const listModels: HarnessAdapter["listModels"] = async ({
     });
 
     const json = await res.json().catch(() => null);
+
+    /**
+     * An unreadable list is a failed sync, not an empty catalog: keeping
+     * the models already known beats marking them all unavailable.
+     */
+    if (json === null) {
+        throw new HarnessError({
+            message: `${provider.name || preset.title} returned an unreadable model list.`,
+            retryable: true,
+        });
+    }
+
     const items: any[] = Array.isArray(json?.data)
         ? json.data
         : Array.isArray(json?.models)

src/harness/providers/openai-compatible/transcribe.ts

@@ -1,3 +1,4 @@
+import HarnessError from "../../harness-error";
 import type { HarnessAdapter } from "../../types";
 import providerRequest from "./provider-request";
 
@@ -36,6 +37,19 @@ const transcribe: HarnessAdapter["transcribe"] = async ({
 
     const json = await res.json().catch(() => null);
 
+    /**
+     * A cut-off or malformed answer is a failed call, not silence: saying
+     * it failed lets the harness try the next model instead of handing the
+     * user an empty transcript.
+     */
+    if (json === null) {
+        throw new HarnessError({
+            message: `${candidate.provider.name || "The provider"} returned an unreadable transcript.`,
+            retryable: true,
+            candidate_key: candidate.key,
+        });
+    }
+
     return { text: String(json?.text || "").trim() };
 };
 

src/harness/tools/run-tool-call.ts

@@ -1,5 +1,7 @@
 import { z } from "zod";
 import { AppData } from "@/src/data/app-data";
+import TaskTimeoutError from "@/src/utils/task-timeout-error";
+import withTimeout from "@/src/utils/with-timeout";
 import HarnessError from "../harness-error";
 import type { HarnessTool, HarnessToolCall } from "../types";
 
@@ -40,14 +42,27 @@ export default async function runToolCall({ call, tools, signal }: Params) {
     }
 
     try {
-        return truncate({
-            text: await tool.run({ args: parsed ? parsed.data : raw, signal }),
+        const text = await withTimeout({
+            timeout_ms: AppData.HarnessToolTimeoutMs,
+            signal,
+            message: `${call.name} took too long and was stopped.`,
+            task: (tool_signal) =>
+                tool.run({
+                    args: parsed ? parsed.data : raw,
+                    signal: tool_signal,
+                }),
         });
+
+        return truncate({ text });
     } catch (error: unknown) {
         if (signal.aborted) {
             throw new HarnessError({ message: `Cancelled.`, code: "aborted" });
         }
 
+        if (error instanceof TaskTimeoutError) {
+            return `Error: ${error.message}`;
+        }
+
         return `Error: ${error instanceof Error ? error.message : String(error)}`;
     }
 }

src/live-relay/index.ts

@@ -20,14 +20,29 @@ import type { LiveRelayCursor } from "./types";
  * Runs in every server process, ungated.
  */
 export default async function liveRelay() {
-    const latest = async (table: "bot_messages" | "bots_queue_updates") =>
-        Number(
-            (
-                await BunSQLite.sql<{ id: number }>({
-                    sql: `SELECT MAX(id) AS id FROM ${table}`,
-                })
-            ).payload?.[0]?.id,
-        ) || 0;
+    /**
+     * A failed read starts the relay from the beginning instead of not
+     * starting it: clients drop what they already have.
+     */
+    const latest = async (table: "bot_messages" | "bots_queue_updates") => {
+        try {
+            return (
+                Number(
+                    (
+                        await BunSQLite.sql<{ id: number }>({
+                            sql: `SELECT MAX(id) AS id FROM ${table}`,
+                        })
+                    ).payload?.[0]?.id,
+                ) || 0
+            );
+        } catch (error) {
+            await logError({
+                title: `Live relay couldn't read ${table}`,
+                error,
+            }).catch(() => {});
+            return 0;
+        }
+    };
 
     const cursor: LiveRelayCursor = {
         message_id: await latest("bot_messages"),

src/server.ts

@@ -200,6 +200,19 @@ if (tls) {
     }
 }
 
-cron();
-botsQueue();
-liveRelay();
+/**
+ * Background services start after the server is up, and one failing to
+ * start never takes the server (or the others) down with it.
+ */
+try {
+    cron();
+} catch (error) {
+    logError({ title: "Cron couldn't start", error }).catch(() => {});
+}
+
+botsQueue().catch((error) =>
+    logError({ title: "Bots queue couldn't start", error }).catch(() => {}),
+);
+liveRelay().catch((error) =>
+    logError({ title: "Live relay couldn't start", error }).catch(() => {}),
+);

src/utils/task-timeout-error.ts

@@ -0,0 +1,10 @@
+/**
+ * A task was stopped because it ran past its deadline (`withTimeout`).
+ * Its message is safe to show: it says what ran too long, not internals.
+ */
+export default class TaskTimeoutError extends Error {
+    constructor({ message }: { message: string }) {
+        super(message);
+        this.name = "TaskTimeoutError";
+    }
+}

src/utils/with-timeout.test.ts

@@ -0,0 +1,100 @@
+import { describe, expect, test } from "bun:test";
+import TaskTimeoutError from "./task-timeout-error";
+import withTimeout from "./with-timeout";
+
+const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
+
+describe("withTimeout", () => {
+    test("returns the task's result", async () => {
+        const result = await withTimeout({
+            timeout_ms: 100,
+            task: async () => "done",
+        });
+
+        expect(result).toBe("done");
+    });
+
+    test("stops a task that runs past its deadline", async () => {
+        const error = await withTimeout({
+            timeout_ms: 10,
+            message: "Too slow.",
+            task: () => wait(200),
+        }).catch((e) => e);
+
+        expect(error).toBeInstanceOf(TaskTimeoutError);
+        expect(error instanceof TaskTimeoutError && error.message).toBe("Too slow.");
+    });
+
+    test("ends the wait even when the task ignores its signal", async () => {
+        const error = await withTimeout({
+            timeout_ms: 10,
+            task: () => wait(200),
+        }).catch((e) => e);
+
+        expect(error).toBeInstanceOf(TaskTimeoutError);
+    });
+
+    test("aborts the task's signal on the deadline", async () => {
+        let aborted = false;
+
+        await withTimeout({
+            timeout_ms: 10,
+            task: async (task_signal) => {
+                task_signal.addEventListener("abort", () => {
+                    aborted = true;
+                });
+
+                await wait(50);
+            },
+        }).catch(() => {});
+
+        expect(aborted).toBe(true);
+    });
+
+    test("passes the caller's cancellation through", async () => {
+        const controller = new AbortController();
+        let aborted = false;
+        let started: () => void = () => {};
+        const running_task = new Promise<void>((resolve) => {
+            started = resolve;
+        });
+
+        const running = withTimeout({
+            timeout_ms: 1000,
+            signal: controller.signal,
+            task: async (task_signal) => {
+                task_signal.addEventListener("abort", () => {
+                    aborted = true;
+                });
+
+                started();
+                await wait(200);
+            },
+        });
+
+        await running_task;
+        controller.abort();
+        const error = await running.catch((e) => e);
+
+        expect(error).toBeInstanceOf(Error);
+        expect(aborted).toBe(true);
+    });
+
+    test("doesn't start a task when already cancelled", async () => {
+        const controller = new AbortController();
+        let started = false;
+
+        controller.abort();
+
+        const error = await withTimeout({
+            timeout_ms: 100,
+            signal: controller.signal,
+            task: async () => {
+                started = true;
+            },
+        }).catch((e) => e);
+
+        expect(error).toBeInstanceOf(Error);
+        expect(started).toBe(false);
+    });
+});

src/utils/with-timeout.ts

@@ -0,0 +1,85 @@
+import TaskTimeoutError from "./task-timeout-error";
+
+type Params<T> = {
+    /**
+     * The work. It gets a signal that aborts when the deadline passes or
+     * the caller cancels, so its requests stop too.
+     */
+    task: (signal: AbortSignal) => Promise<T>;
+    timeout_ms: number;
+    /**
+     * The caller's own cancellation, e.g. a job that lost its lock.
+     */
+    signal?: AbortSignal;
+    /**
+     * Shown when the deadline passes.
+     */
+    message?: string;
+    /**
+     * Called once the deadline passes, before the task is stopped. Use it
+     * to tell a deadline apart from the caller's own cancellation.
+     */
+    on_timeout?: () => void;
+};
+
+/**
+ * Runs a task with a deadline, resolving or rejecting with whatever the
+ * task does. When the deadline passes (or the caller's signal aborts) the
+ * task's signal is aborted and the wait ends at once, even if the task
+ * never settles, so a hung call can't hold a queue slot or a lock forever.
+ * Late results and errors from the task are dropped.
+ */
+export default function withTimeout<T>({
+    task,
+    timeout_ms,
+    signal,
+    message = "The task took too long.",
+    on_timeout,
+}: Params<T>): Promise<T> {
+    if (signal?.aborted) {
+        return Promise.reject(reasonOf({ signal }));
+    }
+
+    const controller = new AbortController();
+    const parent = signal;
+    const on_abort = () => controller.abort(reasonOf({ signal: parent! }));
+
+    parent?.addEventListener("abort", on_abort, { once: true });
+
+    const timer = setTimeout(() => {
+        on_timeout?.();
+        controller.abort(new TaskTimeoutError({ message }));
+    }, timeout_ms);
+
+    const done = new Promise<T>((resolve, reject) => {
+        controller.signal.addEventListener(
+            "abort",
+            () => reject(controller.signal.reason),
+            { once: true },
+        );
+
+        /**
+         * `Promise.resolve().then` turns a synchronous throw from the task
+         * into a rejection instead of escaping the executor. A task
+         * cancelled before it starts is never started.
+         */
+        Promise.resolve()
+            .then(() => {
+                if (controller.signal.aborted) {
+                    throw reasonOf({ signal: controller.signal });
+                }
+
+                return task(controller.signal);
+            })
+            .then(resolve, reject);
+    });
+
+    return done.finally(() => {
+        clearTimeout(timer);
+        parent?.removeEventListener("abort", on_abort);
+    });
+}
+
+function reasonOf({ signal }: { signal: AbortSignal }) {
+    return signal.reason instanceof Error ? signal.reason : new Error("Cancelled.");
+}