Add the harness, envs, per-bot models and chat attachments
- Harness: providers, model catalog synced hourly, automatic free picks, cooldowns, tool loop with context trimming, MCP computer tools - Bots queue runs replies and tasks through the harness; jobs wait (attempt refunded) while every model is resting - Envs: Settings > Environment, scoped general/workspace/computer, written to computers and redacted from tool output - Models: Settings > Models, per-harness provider select defaulting to OpenRouter, and a models modal on each bot's chat (the coordinator's edits the defaults) - Chat: thinking row while a reply runs, file attachments (upload, paste, drop) that replies read as images, audio and document text - Fix media uploads failing after a user's first file, and private files being served from the public media path Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
Claude Opus 5.5
@@ -0,0 +1,68 @@
|
||||
import { describe, expect, test } from "bun:test";
|
||||
import botMessagesToHarness from "./bot-messages-to-harness";
|
||||
|
||||
const message = (author: "user" | "bot", text: string) => ({ author, text });
|
||||
|
||||
describe("botMessagesToHarness", () => {
|
||||
test("maps authors to roles", () => {
|
||||
expect(
|
||||
botMessagesToHarness({
|
||||
messages: [message("user", "hi"), message("bot", "hello")],
|
||||
}),
|
||||
).toEqual([
|
||||
{ role: "user", content: "hi" },
|
||||
{ role: "assistant", content: "hello" },
|
||||
]);
|
||||
});
|
||||
|
||||
test("joins back-to-back messages from one side", () => {
|
||||
expect(
|
||||
botMessagesToHarness({
|
||||
messages: [message("user", "one"), message("user", "two")],
|
||||
}),
|
||||
).toEqual([{ role: "user", content: "one\n\ntwo" }]);
|
||||
});
|
||||
|
||||
test("starts with the user and skips empty messages", () => {
|
||||
expect(
|
||||
botMessagesToHarness({
|
||||
messages: [
|
||||
message("bot", "welcome"),
|
||||
message("user", " "),
|
||||
message("user", "hi"),
|
||||
],
|
||||
}),
|
||||
).toEqual([{ role: "user", content: "hi" }]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("botMessagesToHarness with files", () => {
|
||||
const attachments = [{ id: 9, name: "shot.png", media_type: "image" as const, url: "/media/private/9" }];
|
||||
|
||||
test("puts a message's loaded files after its text", () => {
|
||||
const parts = new Map([[1, [{ type: "image" as const, url: "data:image/jpeg;base64,AA", name: "shot.png" }]]]);
|
||||
|
||||
expect(
|
||||
botMessagesToHarness({
|
||||
messages: [{ id: 1, author: "user", text: "What's this?", attachments }],
|
||||
parts,
|
||||
}),
|
||||
).toEqual([
|
||||
{
|
||||
role: "user",
|
||||
content: [
|
||||
{ type: "text", text: "What's this?" },
|
||||
{ type: "image", url: "data:image/jpeg;base64,AA", name: "shot.png" },
|
||||
],
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
test("names files it didn't load, and keeps a file-only message", () => {
|
||||
expect(
|
||||
botMessagesToHarness({
|
||||
messages: [{ id: 2, author: "user", text: "", attachments }],
|
||||
}),
|
||||
).toEqual([{ role: "user", content: `[Attached earlier: "shot.png"]` }]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,99 @@
|
||||
import type { HarnessContentPart, HarnessMessage } from "@/src/harness/types";
|
||||
import type { BotMessageWithAttachments } from "@/src/types";
|
||||
|
||||
type Params = {
|
||||
messages: Pick<BotMessageWithAttachments, "id" | "author" | "text" | "attachments">[];
|
||||
/**
|
||||
* Loaded files by message id (`loadAttachmentParts`). Files of other
|
||||
* messages are only named.
|
||||
*/
|
||||
parts?: Map<number, HarnessContentPart[]>;
|
||||
};
|
||||
|
||||
type Turn = { role: "user" | "assistant"; parts: HarnessContentPart[] };
|
||||
|
||||
/**
|
||||
* A bot's conversation (oldest first) as model turns: the user's messages
|
||||
* (with their files) as user turns, the bot's as its own. Back-to-back
|
||||
* messages from the same side are joined, since some providers reject
|
||||
* repeated roles. Leading bot messages are dropped so the conversation
|
||||
* starts with the user. A turn that is only text stays a string.
|
||||
*/
|
||||
export default function botMessagesToHarness({ messages, parts }: Params): HarnessMessage[] {
|
||||
const turns: Turn[] = [];
|
||||
|
||||
for (const message of messages) {
|
||||
const role = message.author == "user" ? "user" : "assistant";
|
||||
const content = messageParts({ message, role, parts });
|
||||
|
||||
if (!content.length) continue;
|
||||
if (!turns.length && role == "assistant") continue;
|
||||
|
||||
const last = turns[turns.length - 1];
|
||||
|
||||
if (last?.role == role) {
|
||||
last.parts.push(...content);
|
||||
} else {
|
||||
turns.push({ role, parts: content });
|
||||
}
|
||||
}
|
||||
|
||||
return turns.map(({ role, parts }) => {
|
||||
const merged = mergeText({ parts });
|
||||
|
||||
if (role == "assistant" || merged.every((part) => part.type == "text")) {
|
||||
return {
|
||||
role,
|
||||
content: merged.map((part) => (part.type == "text" ? part.text : "")).join("\n\n"),
|
||||
};
|
||||
}
|
||||
|
||||
return { role, content: merged };
|
||||
});
|
||||
}
|
||||
|
||||
function messageParts({
|
||||
message,
|
||||
role,
|
||||
parts,
|
||||
}: {
|
||||
message: Params["messages"][number];
|
||||
role: Turn["role"];
|
||||
parts?: Params["parts"];
|
||||
}): HarnessContentPart[] {
|
||||
const text = message.text?.trim();
|
||||
const content: HarnessContentPart[] = text ? [{ type: "text", text }] : [];
|
||||
|
||||
if (role == "assistant" || !message.attachments?.length) return content;
|
||||
|
||||
const loaded = parts?.get(Number(message.id));
|
||||
|
||||
if (loaded?.length) return [...content, ...loaded];
|
||||
|
||||
return [
|
||||
...content,
|
||||
{
|
||||
type: "text",
|
||||
text: `[Attached earlier: ${message.attachments.map((file) => `"${file.name}"`).join(", ")}]`,
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
/**
|
||||
* Joins neighbouring text parts with a blank line.
|
||||
*/
|
||||
function mergeText({ parts }: { parts: HarnessContentPart[] }) {
|
||||
const merged: HarnessContentPart[] = [];
|
||||
|
||||
for (const part of parts) {
|
||||
const last = merged[merged.length - 1];
|
||||
|
||||
if (part.type == "text" && last?.type == "text") {
|
||||
merged[merged.length - 1] = { type: "text", text: `${last.text}\n\n${part.text}` };
|
||||
} else {
|
||||
merged.push(part);
|
||||
}
|
||||
}
|
||||
|
||||
return merged;
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
import { describe, expect, test } from "bun:test";
|
||||
import composeJobBrief from "./compose-job-brief";
|
||||
|
||||
describe("composeJobBrief", () => {
|
||||
test("puts the title, instructions and parameters in order", () => {
|
||||
expect(
|
||||
composeJobBrief({
|
||||
title: "Tidy invoices",
|
||||
instructions: " Rename every PDF by date. ",
|
||||
params: { folder: "~/invoices" },
|
||||
}),
|
||||
).toBe(
|
||||
[
|
||||
"# Job: Tidy invoices",
|
||||
"Rename every PDF by date.",
|
||||
'## Parameters\n\n```json\n{\n "folder": "~/invoices"\n}\n```',
|
||||
"Work on this until it's done. Then reply to the user with the result: what you did, and anything they need to know or do.",
|
||||
].join("\n\n"),
|
||||
);
|
||||
});
|
||||
|
||||
test("leaves out empty parts", () => {
|
||||
expect(composeJobBrief({ params: {} })).toStartWith(
|
||||
"# Job: Untitled\n\nWork on this",
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,29 @@
|
||||
type Params = {
|
||||
title?: string | null;
|
||||
instructions?: string | null;
|
||||
params?: Record<string, unknown>;
|
||||
};
|
||||
|
||||
/**
|
||||
* The message a bot gets to start a queued job (markdown): the brief it
|
||||
* wrote for itself, its parameters, and how to finish.
|
||||
*/
|
||||
export default function composeJobBrief({ title, instructions, params }: Params) {
|
||||
const parts = [`# Job: ${title?.trim() || "Untitled"}`];
|
||||
|
||||
if (instructions?.trim()) {
|
||||
parts.push(instructions.trim());
|
||||
}
|
||||
|
||||
if (params && Object.keys(params).length) {
|
||||
parts.push(
|
||||
`## Parameters\n\n\`\`\`json\n${JSON.stringify(params, null, 2)}\n\`\`\``,
|
||||
);
|
||||
}
|
||||
|
||||
parts.push(
|
||||
`Work on this until it's done. Then reply to the user with the result: what you did, and anything they need to know or do.`,
|
||||
);
|
||||
|
||||
return parts.join("\n\n");
|
||||
}
|
||||
+46
-23
@@ -1,18 +1,23 @@
|
||||
import bunext from "@moduletrace/bunext";
|
||||
import { AppData } from "@/src/data/app-data";
|
||||
import type { BotJobKind } from "@/src/dict/bot-queue-dict";
|
||||
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";
|
||||
import { setBotsQueueWaker } from "./wake-bots-queue";
|
||||
|
||||
const worker = AppData.BotQueueWorker;
|
||||
let running = 0;
|
||||
const running = { task: 0, reply: 0 };
|
||||
let polling = false;
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
|
||||
/**
|
||||
* 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
|
||||
* up to `AppData.BotQueueConcurrency` tasks and, on their own slots,
|
||||
* `AppData.BotQueueReplyConcurrency` chat replies. Gated in production to the
|
||||
* `MAIN_SERVER` container, like cron.
|
||||
*/
|
||||
export default async function botsQueue() {
|
||||
@@ -28,38 +33,56 @@ export default async function botsQueue() {
|
||||
bunext.bunextLog.info(`Bots queue: released ${released} cut-off job(s)`);
|
||||
}
|
||||
|
||||
setBotsQueueWaker({ waker: wake });
|
||||
poll();
|
||||
}
|
||||
|
||||
/**
|
||||
* Polls now, unless a poll is already running (`wakeBotsQueue`).
|
||||
*/
|
||||
function wake() {
|
||||
if (polling) return;
|
||||
|
||||
clearTimeout(timer);
|
||||
poll();
|
||||
}
|
||||
|
||||
async function claimAndRun({ kind, slots }: { kind: BotJobKind; slots: number }) {
|
||||
const free = slots - running[kind];
|
||||
|
||||
if (free < 1) return;
|
||||
|
||||
for (const job of await claimBotJobs({ worker, limit: free, kind })) {
|
||||
running[kind]++;
|
||||
|
||||
processBotJob({ job, worker })
|
||||
.catch((error) =>
|
||||
logError({
|
||||
title: `Bot job #${job.id} crashed`,
|
||||
error,
|
||||
}).catch(() => {}),
|
||||
)
|
||||
.finally(() => {
|
||||
running[kind]--;
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async function poll() {
|
||||
polling = true;
|
||||
|
||||
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--;
|
||||
});
|
||||
}
|
||||
}
|
||||
await claimAndRun({ kind: "reply", slots: AppData.BotQueueReplyConcurrency });
|
||||
await claimAndRun({ kind: "task", slots: AppData.BotQueueConcurrency });
|
||||
} catch (error) {
|
||||
await logError({ title: "Bots queue poll failed", error }).catch(
|
||||
() => {},
|
||||
);
|
||||
} finally {
|
||||
setTimeout(poll, AppData.BotQueuePollMs);
|
||||
polling = false;
|
||||
clearTimeout(timer);
|
||||
timer = setTimeout(poll, AppData.BotQueuePollMs);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
import path from "node:path";
|
||||
import { AppData } from "@/src/data/app-data";
|
||||
import grabMessagesMedia, {
|
||||
type MessageMediaRow,
|
||||
} from "@/src/functions/backend/db/bots/grab-messages-media";
|
||||
import type { HarnessContentPart } from "@/src/harness/types";
|
||||
import type { BotMessageWithAttachments } from "@/src/types";
|
||||
import grabDirNames from "@/src/utils/grab-dir-names";
|
||||
|
||||
type Params = {
|
||||
/**
|
||||
* The conversation a reply reads, oldest first.
|
||||
*/
|
||||
messages: BotMessageWithAttachments[];
|
||||
};
|
||||
|
||||
/**
|
||||
* The files of the latest few messages that have any
|
||||
* (`AppData.BotReplyAttachmentMessages`), as model input by message id:
|
||||
* images and audio as themselves (a model that can't take them gets them
|
||||
* read first), documents as their text, videos as a note. Older files are
|
||||
* only named (`botMessagesToHarness`).
|
||||
*/
|
||||
export default async function loadAttachmentParts({ messages }: Params) {
|
||||
const recent = messages
|
||||
.filter((message) => message.attachments?.length)
|
||||
.slice(-AppData.BotReplyAttachmentMessages)
|
||||
.map((message) => Number(message.id));
|
||||
|
||||
const media = await grabMessagesMedia({ message_ids: recent });
|
||||
const parts = new Map<number, HarnessContentPart[]>();
|
||||
|
||||
for (const [message_id, files] of media) {
|
||||
parts.set(
|
||||
message_id,
|
||||
(await Promise.all(files.map((file) => fileToParts({ file })))).flat(),
|
||||
);
|
||||
}
|
||||
|
||||
return parts;
|
||||
}
|
||||
|
||||
async function fileToParts({ file }: { file: MessageMediaRow }): Promise<HarnessContentPart[]> {
|
||||
const name = file.title || `file ${file.id}`;
|
||||
|
||||
if (file.media_type == "document") {
|
||||
const text = file.text_content?.trim();
|
||||
|
||||
if (!text) {
|
||||
return [{ type: "text", text: `[Attached file "${name}": its text couldn't be read.]` }];
|
||||
}
|
||||
|
||||
const cut = text.length > AppData.BotAttachmentMaxTextChars;
|
||||
|
||||
return [
|
||||
{
|
||||
type: "text",
|
||||
text: `Attached file "${name}":\n\n${text.slice(0, AppData.BotAttachmentMaxTextChars)}${cut ? `\n\n[The rest of "${name}" was cut: it's ${text.length} characters long.]` : ""}`,
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
if (file.media_type == "video") {
|
||||
return [{ type: "text", text: `[Attached video "${name}": videos can't be read yet.]` }];
|
||||
}
|
||||
|
||||
const disk = Bun.file(path.join(grabDirNames().DATA_DIR, file.media_write_path || ""));
|
||||
|
||||
if (!file.media_write_path || !(await disk.exists())) {
|
||||
return [{ type: "text", text: `[Attached file "${name}" is missing on the server.]` }];
|
||||
}
|
||||
|
||||
const base64 = Buffer.from(await disk.arrayBuffer()).toString("base64");
|
||||
|
||||
if (file.media_type == "image") {
|
||||
return [
|
||||
{ type: "text", text: `[Attached image "${name}"]` },
|
||||
{ type: "image", url: `data:image/${file.mime_type || "jpeg"};base64,${base64}`, name },
|
||||
];
|
||||
}
|
||||
|
||||
return [
|
||||
{ type: "text", text: `[Attached audio "${name}"]` },
|
||||
{ type: "audio", data: base64, format: file.mime_type || "mp3", name },
|
||||
];
|
||||
}
|
||||
@@ -6,6 +6,9 @@ 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 HarnessError from "@/src/harness/harness-error";
|
||||
import deferBotJob from "@/src/functions/backend/db/bots-queue/defer-bot-job";
|
||||
import { BotWaitingStatusLine } from "@/src/dict/bot-queue-dict";
|
||||
import BotJobFatalError from "./bot-job-fatal-error";
|
||||
import pushBotNotice from "./push-bot-notice";
|
||||
import runBotJob from "./run-bot-job";
|
||||
@@ -17,6 +20,11 @@ import type { BotJobContext, BotJobRunner } from "./types";
|
||||
*/
|
||||
const computer_lock = createKeyLock();
|
||||
|
||||
/**
|
||||
* One reply per bot at a time, so replies never cross in the chat.
|
||||
*/
|
||||
const reply_lock = createKeyLock();
|
||||
|
||||
/**
|
||||
* Renew the lock well inside the timeout, so one slow write can't lose it.
|
||||
*/
|
||||
@@ -86,9 +94,12 @@ export default async function processBotJob({
|
||||
});
|
||||
};
|
||||
|
||||
const outcome = job.computer_id
|
||||
? await computer_lock.run({ key: job.computer_id, task: run })
|
||||
: await run();
|
||||
const outcome =
|
||||
job.kind == "reply"
|
||||
? await reply_lock.run({ key: bot_id, task: run })
|
||||
: job.computer_id
|
||||
? await computer_lock.run({ key: job.computer_id, task: run })
|
||||
: await run();
|
||||
|
||||
if (controller.signal.aborted) return;
|
||||
|
||||
@@ -99,11 +110,35 @@ export default async function processBotJob({
|
||||
worker,
|
||||
result: outcome.result,
|
||||
needs_you: outcome.needs_you,
|
||||
silent: outcome.silent,
|
||||
})),
|
||||
});
|
||||
} catch (error: unknown) {
|
||||
if (controller.signal.aborted) return;
|
||||
|
||||
/**
|
||||
* Every model is cooling down (rate limits): nothing was tried, so
|
||||
* wait for the first to come back without using an attempt.
|
||||
*/
|
||||
if (error instanceof HarnessError && error.resting) {
|
||||
await deferBotJob({
|
||||
job_id,
|
||||
worker,
|
||||
until: error.retry_at || Date.now() + AppData.HarnessCooldownMs,
|
||||
refund: true,
|
||||
});
|
||||
|
||||
await pushBotNotice({
|
||||
bot_id,
|
||||
bot: await setBotFace({
|
||||
bot_id,
|
||||
face: "asleep",
|
||||
status_line: BotWaitingStatusLine,
|
||||
}),
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const failed = await failBotJob({
|
||||
job_id,
|
||||
worker,
|
||||
@@ -120,6 +155,18 @@ export default async function processBotJob({
|
||||
if (failed?.notice) {
|
||||
await pushBotNotice({ bot_id, ...failed.notice });
|
||||
}
|
||||
|
||||
/**
|
||||
* A retry no sooner than the provider said to come back.
|
||||
*/
|
||||
if (
|
||||
failed?.status == "pending" &&
|
||||
error instanceof HarnessError &&
|
||||
error.retryable &&
|
||||
error.retry_at
|
||||
) {
|
||||
await deferBotJob({ job_id, worker, until: error.retry_at });
|
||||
}
|
||||
} finally {
|
||||
clearInterval(heartbeat);
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ import type {
|
||||
BUN_SQLITE_COBBLE_BOT_MESSAGES,
|
||||
BUN_SQLITE_COBBLE_BOTS,
|
||||
} from "@/db/types/db";
|
||||
import grabActiveBotTask from "@/src/functions/backend/db/bots/grab-active-bot-task";
|
||||
import pushToBotOwner from "./push-to-bot-owner";
|
||||
|
||||
type Params = {
|
||||
@@ -12,6 +13,7 @@ type Params = {
|
||||
|
||||
/**
|
||||
* Pushes a new bot message and/or face change to the owner's open tabs.
|
||||
* A face change carries the bot's active task, for the working-on panel.
|
||||
*/
|
||||
export default async function pushBotNotice({ bot_id, message, bot }: Params) {
|
||||
if (message) {
|
||||
@@ -24,7 +26,10 @@ export default async function pushBotNotice({ bot_id, message, bot }: Params) {
|
||||
if (bot) {
|
||||
await pushToBotOwner({
|
||||
bot_id,
|
||||
data: { event: "server:bot-updated", data: bot },
|
||||
data: {
|
||||
event: "server:bot-updated",
|
||||
data: { ...bot, task: await grabActiveBotTask({ bot_id }) },
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,14 +1,42 @@
|
||||
import type { BotJobRunner } from "./types";
|
||||
import grabBotById from "@/src/functions/backend/db/bots/grab-bot-by-id";
|
||||
import HarnessError from "@/src/harness/harness-error";
|
||||
import BotJobFatalError from "./bot-job-fatal-error";
|
||||
import runBotReply from "./run-bot-reply";
|
||||
import runBotTask from "./run-bot-task";
|
||||
import type { BotJobRunner } from "./types";
|
||||
|
||||
/**
|
||||
* 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.
|
||||
* Does a job's actual work on the harness: a `reply` answers the user in
|
||||
* chat (`runBotReply`), a `task` works through its brief (`runBotTask`).
|
||||
*
|
||||
* Harness failures that waiting can fix (rate limits, outages) are thrown
|
||||
* as they are, so the queue retries with backoff; the rest are fatal.
|
||||
*/
|
||||
const runBotJob: BotJobRunner = async () => {
|
||||
throw new BotJobFatalError(`Bots can't run queued work yet.`);
|
||||
const runBotJob: BotJobRunner = async (context) => {
|
||||
const { job } = context;
|
||||
const bot = job.bot_id ? await grabBotById({ bot_id: job.bot_id }) : null;
|
||||
|
||||
if (!bot?.id || !bot.user_id) {
|
||||
throw new BotJobFatalError(`The bot for this job no longer exists.`);
|
||||
}
|
||||
|
||||
const owned = { ...bot, id: bot.id, user_id: bot.user_id };
|
||||
|
||||
try {
|
||||
return job.kind == "reply"
|
||||
? await runBotReply({ ...context, bot: owned })
|
||||
: await runBotTask({ ...context, bot: owned });
|
||||
} catch (error: unknown) {
|
||||
if (
|
||||
error instanceof HarnessError &&
|
||||
error.code != "aborted" &&
|
||||
!error.retryable
|
||||
) {
|
||||
throw new BotJobFatalError(error.message);
|
||||
}
|
||||
|
||||
throw error;
|
||||
}
|
||||
};
|
||||
|
||||
export default runBotJob;
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
import type { BUN_SQLITE_COBBLE_BOTS } from "@/db/types/db";
|
||||
import { AppData } from "@/src/data/app-data";
|
||||
import { ChatReplyNote, CoordinatorReplyNote } from "@/src/data/harness-prompts";
|
||||
import grabBotMessages from "@/src/functions/backend/db/bots/grab-bot-messages";
|
||||
import grabBotPrompt from "@/src/functions/backend/db/bots/grab-bot-prompt";
|
||||
import runTextLoop from "@/src/harness/text/run-text-loop";
|
||||
import grabBotChatTools from "@/src/harness/tools/bot/grab-bot-chat-tools";
|
||||
import botMessagesToHarness from "./bot-messages-to-harness";
|
||||
import loadAttachmentParts from "./load-attachment-parts";
|
||||
import type { BotJobContext, BotJobOutcome } from "./types";
|
||||
|
||||
type Params = BotJobContext & {
|
||||
bot: BUN_SQLITE_COBBLE_BOTS & { id: number; user_id: number };
|
||||
};
|
||||
|
||||
/**
|
||||
* Answers the user in chat: the bot reads the recent conversation (and
|
||||
* the files of the latest messages, `loadAttachmentParts`) and
|
||||
* replies, with its chat tools (memory, instructions, start_task, and for
|
||||
* the coordinator delegate_task). Finishes silently when the last message
|
||||
* is already the bot's, i.e. an earlier reply answered everything.
|
||||
*/
|
||||
export default async function runBotReply({
|
||||
bot,
|
||||
job,
|
||||
signal,
|
||||
}: Params): Promise<BotJobOutcome> {
|
||||
const history = await grabBotMessages({
|
||||
bot_id: bot.id,
|
||||
limit: AppData.BotReplyHistoryMessages,
|
||||
});
|
||||
|
||||
if (history[history.length - 1]?.author != "user") {
|
||||
return { result: "", silent: true };
|
||||
}
|
||||
|
||||
const prompt = await grabBotPrompt({ bot_id: bot.id, user_id: bot.user_id });
|
||||
|
||||
const result = await runTextLoop({
|
||||
system: [
|
||||
prompt,
|
||||
ChatReplyNote,
|
||||
bot.kind == "coordinator" ? CoordinatorReplyNote : "",
|
||||
`Now: ${new Date().toUTCString()}.`,
|
||||
]
|
||||
.filter(Boolean)
|
||||
.join("\n\n"),
|
||||
messages: botMessagesToHarness({
|
||||
messages: history,
|
||||
parts: await loadAttachmentParts({ messages: history }),
|
||||
}),
|
||||
tools: await grabBotChatTools({
|
||||
bot,
|
||||
user_id: bot.user_id,
|
||||
message_id: job.message_id || undefined,
|
||||
}),
|
||||
max_steps: AppData.BotReplyMaxToolSteps,
|
||||
bot_id: bot.id,
|
||||
signal,
|
||||
});
|
||||
|
||||
return { result: result.text };
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
import { TaskRunNote } from "@/src/data/harness-prompts";
|
||||
import grabBotPrompt from "@/src/functions/backend/db/bots/grab-bot-prompt";
|
||||
import runTextLoop from "@/src/harness/text/run-text-loop";
|
||||
import openComputerTools from "@/src/harness/tools/computer/open-computer-tools";
|
||||
import grabBotTools from "@/src/harness/tools/bot/grab-bot-tools";
|
||||
import composeJobBrief from "./compose-job-brief";
|
||||
import type { BotJobContext, BotJobOutcome } from "./types";
|
||||
|
||||
type Params = BotJobContext & {
|
||||
bot: { id: number; user_id: number };
|
||||
};
|
||||
|
||||
/**
|
||||
* Works through a task: the bot follows its brief with its full prompt,
|
||||
* its own tools and, when the job has a computer, that computer's tools
|
||||
* (over MCP), until it answers. That answer is the report for the user.
|
||||
* Progress is saved after every model turn.
|
||||
*/
|
||||
export default async function runBotTask({
|
||||
bot,
|
||||
job,
|
||||
params,
|
||||
signal,
|
||||
update,
|
||||
}: Params): Promise<BotJobOutcome> {
|
||||
const computer = job.computer_id
|
||||
? await openComputerTools({ computer_id: job.computer_id, bot_id: bot.id, signal })
|
||||
: null;
|
||||
|
||||
try {
|
||||
const prompt = await grabBotPrompt({ bot_id: bot.id, user_id: bot.user_id });
|
||||
|
||||
const result = await runTextLoop({
|
||||
system: [
|
||||
prompt,
|
||||
TaskRunNote,
|
||||
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.`
|
||||
: `You have no computer for this task.`,
|
||||
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")}`
|
||||
: "",
|
||||
]
|
||||
.filter(Boolean)
|
||||
.join("\n\n"),
|
||||
messages: [
|
||||
{
|
||||
role: "user",
|
||||
content: composeJobBrief({
|
||||
title: job.title,
|
||||
instructions: job.instructions,
|
||||
params,
|
||||
}),
|
||||
},
|
||||
],
|
||||
tools: [
|
||||
...grabBotTools({ user_id: bot.user_id, bot_id: bot.id }),
|
||||
...(computer?.tools || []),
|
||||
],
|
||||
bot_id: bot.id,
|
||||
signal,
|
||||
onStep: async ({ text, tool_calls }) => {
|
||||
await update({
|
||||
progress: tool_calls.length
|
||||
? `Using ${tool_calls.map((call) => call.name).join(", ")}`
|
||||
: text?.slice(0, 500) || undefined,
|
||||
});
|
||||
},
|
||||
});
|
||||
|
||||
return { result: computer ? computer.redact(result.text) : result.text };
|
||||
} finally {
|
||||
await computer?.close();
|
||||
}
|
||||
}
|
||||
@@ -29,6 +29,10 @@ export type BotJobOutcome = {
|
||||
*/
|
||||
result: string;
|
||||
needs_you?: boolean;
|
||||
/**
|
||||
* Finish without posting `result` (a reply with nothing left to answer).
|
||||
*/
|
||||
silent?: boolean;
|
||||
};
|
||||
|
||||
export type BotJobRunner = (context: BotJobContext) => Promise<BotJobOutcome>;
|
||||
|
||||
@@ -0,0 +1,18 @@
|
||||
let wake: (() => void) | undefined;
|
||||
|
||||
/**
|
||||
* Lets the running queue handler register how to poll at once. Kept apart
|
||||
* from the handler so API routes can wake it without bundling it.
|
||||
*/
|
||||
export function setBotsQueueWaker({ waker }: { waker: () => void }) {
|
||||
wake = waker;
|
||||
}
|
||||
|
||||
/**
|
||||
* Polls now instead of at the next tick, so a chat reply starts at once.
|
||||
* Does nothing where the queue doesn't run (the next poll on the main
|
||||
* server picks the job up).
|
||||
*/
|
||||
export default function wakeBotsQueue() {
|
||||
wake?.();
|
||||
}
|
||||
Reference in New Issue
Block a user