cobble/src/bots-queue/run-bot-task.ts

import grabBotRunPowers from "@/src/functions/backend/db/bot-types/grab-bot-run-powers";
import { grabSkillBodyShownChars } from "@/src/functions/backend/db/app-settings/grab-skill-limits";
import grabWorkspaceAdminTools from "@/src/harness/tools/bot/grab-workspace-admin-tools";
import workspaceAdminsNote from "./workspace-admins-note";
import type { BUN_SQLITE_COBBLE_BOTS } from "@/db/types/db";
import {
    ActionCheckNote,
    botFilesNote,
    ComputerFilesNote,
    ScheduledRunNote,
    TasksNote,
    TaskRunNote,
    WorkspaceLearningNote,
} from "@/src/data/harness-prompts";
import { grabHarnessMaxToolSteps } from "@/src/functions/backend/db/app-settings/grab-run-limits";
import grabBotComputers from "@/src/functions/backend/db/computers/grab-bot-computers";
import grabBotPrompt from "@/src/functions/backend/db/bots/grab-bot-prompt";
import grabUserWorkspaces from "@/src/functions/backend/db/workspaces/grab-user-workspaces";
import runTextLoop from "@/src/harness/text/run-text-loop";
import openComputerTools from "@/src/harness/tools/computer/open-computer-tools";
import openBotExtensions, { type OpenBotExtensions } from "@/src/harness/tools/extensions/open-bot-extensions";
import openBotSkills, { type OpenBotSkills } from "@/src/harness/tools/skills/open-bot-skills";
import grabBotTools from "@/src/harness/tools/bot/grab-bot-tools";
import createWorkspaceTool from "@/src/harness/tools/bot/create-workspace-tool";
import updateWorkspaceTool from "@/src/harness/tools/bot/update-workspace-tool";
import askToTakeOverTool from "@/src/harness/tools/computer/ask-to-take-over-tool";
import importFileTool from "@/src/harness/tools/computer/import-file-tool";
import searchComputersTool from "@/src/harness/tools/computer/search-computers-tool";
import grabBotFileTools from "@/src/harness/tools/storage/grab-bot-file-tools";
import reportRunFailedTool from "@/src/harness/tools/bot/report-run-failed-tool";
import computerRef from "@/src/utils/computer-ref";
import grabRunScopes from "@/src/functions/backend/db/bot-files/grab-run-scopes";
import composeJobBrief from "./compose-job-brief";
import grabSessionCarryOver from "./grab-session-carry-over";
import jobSessionId from "@/src/functions/backend/db/bots-queue/job-session-id";
import withTakeOverRefs from "./with-take-over-refs";
import stripAppRecords from "@/src/utils/strip-app-records";
import ScheduledRunFailedError from "./scheduled-run-failed-error";
import type { BotJobContext, BotJobOutcome } from "./types";
import computerRunDirs from "@/src/utils/computer-work-dirs";

type Params = BotJobContext & {
    bot: Pick<BUN_SQLITE_COBBLE_BOTS, "kind"> & { id: number };
    /**
     * The user whose conversation the job is in.
     */
    user_id: number;
};

/**
 * Works through a task in its session's workspace (if any): the bot
 * follows its brief with its full prompt,
 * its own tools, search_computers across all its computers, its MCP
 * servers' and custom tools (`openBotExtensions`) and, when the job has a
 * computer, that computer's tools (over MCP), until it answers. That answer is the report for the user.
 * When only the user can go on (a login, a captcha), the bot asks them to
 * take over (`ask_user_to_take_over`) and the report needs them.
 * A scheduled run acts on its own (`ScheduledRunNote`); one that reports
 * it failed (`report_run_failed`) throws, so the queue retries it.
 * Every model turn is reported as a step (`report`).
 */
export default async function runBotTask({
    bot,
    user_id,
    job,
    params,
    signal,
    report,
}: Params): Promise<BotJobOutcome> {
    const workspace_id = job.workspace_id || null;
    /**
     * Where the run works on a computer: the user's folder in the bot's.
     */
    const dirs = computerRunDirs({ bot_id: bot.id, bot_kind: bot.kind, user_id, workspace_id });

    const computer = job.computer_id
        ? await openComputerTools({
              computer_id: job.computer_id,
              bot_id: bot.id,
              workspace_id,
              dirs,
              signal,
          })
        : null;

    let extensions: OpenBotExtensions | undefined;
    let skills: OpenBotSkills | undefined;

    try {
        const session_id = await jobSessionId({ job });
        const prompt = await grabBotPrompt({ bot_id: bot.id, user_id, workspace_id });
        const computers = await grabBotComputers({ bot_id: bot.id, workspace_id });

        /**
         * What the bot's types let it do here. A coordinator's task can
         * also change workspaces, and a workspace admin's its workspace:
         * the findings it was asked to record go there, not only in the
         * report.
         */
        const powers = await grabBotRunPowers({ bot_id: bot.id, workspace_id });
        const coordinator_workspaces = powers.is_coordinator
            ? (await grabUserWorkspaces({ user_id })).filter((w) => !w.archived)
            : [];
        const carry_over = await grabSessionCarryOver({
            session_id,
            bot_id: bot.id,
            workspace_id,
            signal,
        }).catch(() => "");
        const take_overs: string[] = [];
        /**
         * Files the bot wrote and asked to send with its report.
         */
        const attachments: number[] = [];
        const scheduled = Boolean(params.schedule_id);
        let failure = null as string | null;

        /**
         * A take-over defaults to the computer the task runs on.
         */
        const take_over_computers = [
            ...computers.filter((c) => c.id == job.computer_id),
            ...computers.filter((c) => c.id != job.computer_id),
        ];
        /**
         * The files this run reaches, for its file tools, an import and the
         * pictures it may set: one answer, where `grabRunScopes` decides.
         */
        const scopes = await grabRunScopes({ bot, user_id, workspace_id });

        /**
         * A workspace admin's task keeps the workspace's context and its
         * members' instructions; it hands nothing over (no session to report
         * back to mid-task).
         */
        const admin_tools =
            !powers.is_coordinator && workspace_id && powers.is_workspace_admin
                ? (await grabWorkspaceAdminTools({ bot, user_id, workspace_id, powers, scopes })).tools
                : [];

        const tools = [
            ...(await grabBotTools({
                user_id,
                bot_id: bot.id,
                workspace_id,
                computers,
                session_id,
                is_coordinator: powers.is_coordinator,
                admin_workspace_ids: powers.admin_workspace_ids,
                scopes,
            })),
            ...admin_tools,
            ...(powers.is_coordinator
                ? [
                      createWorkspaceTool({ user_id, workspaces: coordinator_workspaces }),
                      ...(coordinator_workspaces.length
                          ? [
                                updateWorkspaceTool({
                                    user_id,
                                    workspaces: coordinator_workspaces,
                                    workspace_id,
                                    scopes,
                                }),
                            ]
                          : []),
                  ]
                : []),
            ...(computers.length
                ? [
                      searchComputersTool({ bot_id: bot.id, computers, workspace_id, dirs }),
                      askToTakeOverTool({
                          computers: take_over_computers,
                          in_task: true,
                          onAsk: ({ ref }) => take_overs.push(ref),
                      }),
                  ]
                : []),
            ...(scheduled ? [reportRunFailedTool({ onFail: ({ reason }) => (failure = reason) })] : []),
            ...(computer?.tools || []),
            ...(job.computer_id
                ? [
                      importFileTool({
                          computer_id: job.computer_id,
                          scopes,
                          bot_id: bot.id,
                          user_id,
                          dirs,
                      }),
                  ]
                : []),
            ...(await grabBotFileTools({
                bot,
                user_id,
                workspace_id,
                session_id,
                scopes,
                onAttach: ({ media_id }) => attachments.push(media_id),
            })),
        ];

        extensions = await openBotExtensions({
            bot_id: bot.id,
            workspace_id,
            computer_id: job.computer_id || null,
            dirs,
            taken: tools.map((tool) => tool.name),
            signal,
        });

        skills = openBotSkills({
            skills: extensions.skills,
            shown_chars: await grabSkillBodyShownChars(),
            taken: [...tools, ...extensions.tools].map((tool) => tool.name),
        });

        const result = await runTextLoop({
            system: [
                prompt,
                carry_over,
                TaskRunNote,
                scheduled ? ScheduledRunNote : "",
                botFilesNote({ bot_id: bot.id }),
                ComputerFilesNote({ computer_id: job.computer_id || 0, dirs }),
                TasksNote,
                ActionCheckNote,
                workspace_id ? WorkspaceLearningNote : "",
                powers.is_coordinator
                    ? `If this brief asked you to record something in a workspace (its context, or a new one), do it yourself with update_workspace or create_workspace before you report: the report alone changes nothing.`
                    : admin_tools.some((tool) => tool.name == "update_workspace")
                      ? `If this brief asked you to record something in this workspace's context, do it yourself with update_workspace before you report: the report alone changes nothing.`
                      : "",
                powers.is_coordinator || powers.is_workspace_admin
                    ? ""
                    : await workspaceAdminsNote({ bot_id: bot.id, workspace_id }),
                computer
                    ? `You're working on the computer "${computer.name}" (a Linux sandbox with a shell, files, code runners and a browser). Use its tools; don't ask the user to run anything. When only the user can go on (a login, a captcha, a code sent to them), ask them to take over with ask_user_to_take_over.`
                    : workspace_id
                      ? `You have no computer for this task: this workspace has none. If the task needs one, stop and say so in the report; the user can add a computer to the workspace.`
                      : `You have no computer for this task.`,
                computers.length > 1
                    ? `search_computers runs a read-only command on every computer you have. When you name one in the report, write its ref, e.g. ${computerRef({ id: computers[0]!.id || 0 })}: the user can click it to switch you to that computer.`
                    : "",
                computer?.envs.length
                    ? `Environment variables set on this computer (use them by name, e.g. $KEY; never print their values):\n${computer.envs
                          .map((env) => `- ${env.key}${env.description ? `: ${env.description}` : ""}`)
                          .join("\n")}`
                    : "",
                skills.note,
                extensions.note,
            ]
                .filter(Boolean)
                .join("\n\n"),
            messages: [
                {
                    role: "user",
                    content: composeJobBrief({
                        title: job.title,
                        instructions: job.instructions,
                        params,
                    }),
                },
            ],
            tools: [...tools, ...extensions.tools, ...skills.tools],
            max_steps: await grabHarnessMaxToolSteps(),
            bot_id: bot.id,
            workspace_id,
            signal,
            onStep: (step) =>
                report({ ...step, text: step.text && computer ? computer.redact(step.text) : step.text }),
        });

        if (failure) {
            throw new ScheduledRunFailedError(computer ? computer.redact(failure) : failure);
        }

        const text = stripAppRecords({ text: result.text });

        return {
            ...withTakeOverRefs({
                text: computer ? computer.redact(text) : text,
                refs: take_overs,
            }),
            attachments: attachments.length ? attachments : undefined,
            tool_log: result.tool_uses,
        };
    } finally {
        await extensions?.close();
        await computer?.close();
    }
}