Make routines act on their own and check every bot action
Scheduled runs are pre-approved: they carry out their brief instead of drafting for confirmation. A run that can't do its brief calls report_run_failed and is retried 5 and 10 minutes later; only then does the user get one "failed after N tries" message, and the routine keeps why in last_error. Every reply and task now gets ActionCheckNote: look for an existing record before creating one, and verify an action took effect before saying it's done. Memory and instructions saves are read back, and create_bot, create_schedule, start_task and delegate_task refuse duplicates. Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
Claude Opus 5.5
@@ -9,6 +9,8 @@ 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 recordCronJobFailure from "@/src/functions/backend/db/cron-jobs/record-cron-job-failure";
|
||||
import scheduledRunRetryAt from "@/src/utils/scheduled-run-retry-at";
|
||||
import {
|
||||
BotComputerBusyStatusLine,
|
||||
BotReplyJobTitle,
|
||||
@@ -56,7 +58,9 @@ type Params = {
|
||||
* 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.
|
||||
* dropped. A failed scheduled run (a routine) is retried with longer gaps
|
||||
* (`scheduledRunRetryAt`); once out of tries the user is told, and the
|
||||
* schedule keeps why (`recordCronJobFailure`).
|
||||
*/
|
||||
export default async function processBotJob({
|
||||
job,
|
||||
@@ -67,6 +71,8 @@ export default async function processBotJob({
|
||||
|
||||
const job_id = job.id;
|
||||
const bot_id = job.bot_id;
|
||||
const params = parseJobJson({ text: job.params }) || {};
|
||||
const schedule_id = Number(params.schedule_id) || null;
|
||||
|
||||
/**
|
||||
* Its face changes only for the user it works for.
|
||||
@@ -133,7 +139,7 @@ export default async function processBotJob({
|
||||
|
||||
return await runner({
|
||||
job,
|
||||
params: parseJobJson({ text: job.params }) || {},
|
||||
params,
|
||||
checkpoint: parseJobJson({ text: job.checkpoint }),
|
||||
signal: controller.signal,
|
||||
update,
|
||||
@@ -223,11 +229,17 @@ export default async function processBotJob({
|
||||
return;
|
||||
}
|
||||
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
const tries = Number(job.attempts) || 1;
|
||||
|
||||
const failed = await failBotJob({
|
||||
job_id,
|
||||
worker,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
error: message,
|
||||
fatal: error instanceof BotJobFatalError,
|
||||
failure_message: schedule_id
|
||||
? `My scheduled run "${job.title || "Scheduled task"}" failed after ${tries} ${tries == 1 ? "try" : "tries"}: ${message}`
|
||||
: undefined,
|
||||
}).catch(async (fail_error) => {
|
||||
await logError({
|
||||
title: `Bot job #${job_id} couldn't be failed`,
|
||||
@@ -240,6 +252,19 @@ export default async function processBotJob({
|
||||
await pushBotNotice({ bot_id, ...failed.notice });
|
||||
}
|
||||
|
||||
if (schedule_id && failed?.status == "failed") {
|
||||
await recordCronJobFailure({ cron_job_id: schedule_id, job_id, error: message }).catch(
|
||||
() => {},
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* A routine's retry waits long enough for an outage to pass.
|
||||
*/
|
||||
if (schedule_id && failed?.status == "pending") {
|
||||
await deferBotJob({ job_id, worker, until: scheduledRunRetryAt({ attempts: failed.attempts }) });
|
||||
}
|
||||
|
||||
/**
|
||||
* A retry no sooner than the provider said to come back.
|
||||
*/
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import type { BUN_SQLITE_COBBLE_BOTS } from "@/db/types/db";
|
||||
import { AppData } from "@/src/data/app-data";
|
||||
import {
|
||||
ActionCheckNote,
|
||||
ChatReplyNote,
|
||||
CoordinatorReplyNote,
|
||||
CoordinatorSoloNote,
|
||||
@@ -110,6 +111,7 @@ export default async function runBotReply({
|
||||
prompt,
|
||||
carry_over,
|
||||
ChatReplyNote,
|
||||
ActionCheckNote,
|
||||
workspace_id ? WorkspaceLearningNote : "",
|
||||
composeComputersNote({
|
||||
computers,
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import type { BUN_SQLITE_COBBLE_BOTS } from "@/db/types/db";
|
||||
import { TaskRunNote, WorkspaceLearningNote } from "@/src/data/harness-prompts";
|
||||
import { ActionCheckNote, ScheduledRunNote, TaskRunNote, WorkspaceLearningNote } from "@/src/data/harness-prompts";
|
||||
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";
|
||||
@@ -11,11 +11,13 @@ 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 searchComputersTool from "@/src/harness/tools/computer/search-computers-tool";
|
||||
import reportRunFailedTool from "@/src/harness/tools/bot/report-run-failed-tool";
|
||||
import computerRef from "@/src/utils/computer-ref";
|
||||
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 ScheduledRunFailedError from "./scheduled-run-failed-error";
|
||||
import type { BotJobContext, BotJobOutcome } from "./types";
|
||||
|
||||
type Params = BotJobContext & {
|
||||
@@ -34,6 +36,8 @@ type Params = BotJobContext & {
|
||||
* 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({
|
||||
@@ -77,6 +81,8 @@ export default async function runBotTask({
|
||||
signal,
|
||||
}).catch(() => "");
|
||||
const take_overs: string[] = [];
|
||||
const scheduled = Boolean(params.schedule_id);
|
||||
let failure = null as string | null;
|
||||
|
||||
/**
|
||||
* A take-over defaults to the computer the task runs on.
|
||||
@@ -112,6 +118,7 @@ export default async function runBotTask({
|
||||
}),
|
||||
]
|
||||
: []),
|
||||
...(scheduled ? [reportRunFailedTool({ onFail: ({ reason }) => (failure = reason) })] : []),
|
||||
...(computer?.tools || []),
|
||||
];
|
||||
|
||||
@@ -128,6 +135,8 @@ export default async function runBotTask({
|
||||
prompt,
|
||||
carry_over,
|
||||
TaskRunNote,
|
||||
scheduled ? ScheduledRunNote : "",
|
||||
ActionCheckNote,
|
||||
workspace_id ? WorkspaceLearningNote : "",
|
||||
bot.kind == "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.`
|
||||
@@ -167,6 +176,10 @@ export default async function runBotTask({
|
||||
report({ ...step, text: step.text && computer ? computer.redact(step.text) : step.text }),
|
||||
});
|
||||
|
||||
if (failure) {
|
||||
throw new ScheduledRunFailedError(computer ? computer.redact(failure) : failure);
|
||||
}
|
||||
|
||||
return withTakeOverRefs({
|
||||
text: computer ? computer.redact(result.text) : result.text,
|
||||
refs: take_overs,
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
/**
|
||||
* Thrown by a task runner when a scheduled run reported that it failed
|
||||
* (`report_run_failed`). Unlike `BotJobFatalError`, the queue retries it.
|
||||
*/
|
||||
export default class ScheduledRunFailedError extends Error {
|
||||
constructor(message: string) {
|
||||
super(message);
|
||||
this.name = "ScheduledRunFailedError";
|
||||
}
|
||||
}
|
||||
@@ -112,6 +112,11 @@ export const AppData = {
|
||||
BotQueueMaxAttempts: 3,
|
||||
BotQueueLockTimeoutMs: 10 * 60 * 1000,
|
||||
BotQueueBackoffMs: 30 * 1000,
|
||||
/**
|
||||
* A failed scheduled run is tried again after this, doubling each time
|
||||
* (`scheduledRunRetryAt`), so a short outage doesn't sink the run.
|
||||
*/
|
||||
ScheduledRunRetryMs: 5 * 60 * 1000,
|
||||
/**
|
||||
* The queue handler's lock name. Stable across restarts, so on startup
|
||||
* it can release the jobs it was running when the process stopped.
|
||||
|
||||
@@ -36,6 +36,19 @@ You are chatting with the user. Answer their latest messages directly and briefl
|
||||
|
||||
Be proactive. Follow where the conversation is heading and, once you've answered, offer the next step: one concrete suggestion or a short question about what to do next, e.g. "Want me to draft the reply as well?" or "Should I check the other two accounts the same way?". Base it on what the user is trying to get done, not on generic options. Offer one step, not a menu. Skip it when you've already asked the user something in this reply, when you've just started a task (its result will suggest what's next), or when the user is clearly wrapping up.`;
|
||||
|
||||
/**
|
||||
* Added to every run, chat replies and tasks alike: check before you
|
||||
* create, and check after you act.
|
||||
*/
|
||||
export const ActionCheckNote = `# Check before and after you act
|
||||
|
||||
Every action counts here: saving memory or instructions, creating or changing a bot, workspace or schedule, starting or handing over a task, and anything you do through other tools or on a computer (sending an email, writing a file, adding a record).
|
||||
|
||||
- Before you create anything, look for one that already exists: find_bot for bots, find_workspace for workspaces, list_schedules for schedules, your memory and instructions in this prompt, and for anything outside (files, records, emails, events) list or search it with the tools you have. If it's there, update it instead of making a second one. Create something new only once you've checked there's none.
|
||||
- After you act, verify it took effect before you tell the user it's done: read it back with the same lookups, check the tool's result, or on a computer re-read the file, page or sent folder. Only say "done" for what you verified.
|
||||
- If you couldn't verify it, say so plainly instead of claiming it's done. If it didn't take, try again once, then report what failed.
|
||||
- Starting or handing over a task isn't finishing it: say it's under way, not done.`;
|
||||
|
||||
/**
|
||||
* Added to the coordinator's chat prompt: it hands work to the crew.
|
||||
*/
|
||||
@@ -54,6 +67,18 @@ export const TaskRunNote = `# Working on a task
|
||||
|
||||
You are working through a task on your own; the user isn't watching. Use your tools until the task is done, then answer with the report for the user: what you did, what you found, and anything left for them. End it with the next step you'd suggest, as a short question the user can say yes to. Keep the report short and in markdown. If you're blocked on something only the user can give you, stop and ask for it in the report. If they need to do it on the computer themselves (log in, a captcha, a code), call ask_user_to_take_over first.`;
|
||||
|
||||
/**
|
||||
* Added after `TaskRunNote` when the task is a scheduled run (a routine):
|
||||
* the user approved its brief when they set it up, so it acts on its own.
|
||||
*/
|
||||
export const ScheduledRunNote = `# Running a routine
|
||||
|
||||
This task is a scheduled run of a routine the user set up. They already approved everything its brief asks for, so carry it out in full on your own, including actions that reach outside: sending emails or messages, posting, filing. Never draft something for their approval or ask whether to go ahead; do it, then report it.
|
||||
|
||||
- Only stop for what nobody but the user can give you (a login, a captcha, a code), and ask them to take over with ask_user_to_take_over.
|
||||
- If you couldn't do what the brief asks (a send failed, a service is down, a tool keeps erroring), call report_run_failed with why instead of writing a report. The run is tried again later, and the user hears about it only if every try fails.
|
||||
- Don't end the report with a question: say what you did and anything the user should know.`;
|
||||
|
||||
/**
|
||||
* Added to a bot's prompt when it works in a workspace: what it learns
|
||||
* there stays there unless it applies everywhere.
|
||||
|
||||
@@ -66,6 +66,7 @@ export const ReservedToolNames = [
|
||||
"delegate_task",
|
||||
"delete_schedule",
|
||||
"list_schedules",
|
||||
"report_run_failed",
|
||||
"search_computers",
|
||||
"set_bot_workspace_instructions",
|
||||
"set_workspace_computer",
|
||||
|
||||
@@ -24,8 +24,8 @@ type Params = {
|
||||
/**
|
||||
* Records a failed attempt. The job goes back to the queue with a backoff
|
||||
* delay, or, once out of attempts (or `fatal`), is marked failed and the
|
||||
* bot tells the user it needs them. Returns the job's new status and the
|
||||
* failure notice, if one was sent.
|
||||
* bot tells the user it needs them. Returns the job's new status, the
|
||||
* attempts it has used and the failure notice, if one was sent.
|
||||
*/
|
||||
export default async function failBotJob({
|
||||
job_id,
|
||||
@@ -58,9 +58,10 @@ export default async function failBotJob({
|
||||
}
|
||||
|
||||
const status = res.singleRes.status;
|
||||
const attempts = Number(res.singleRes.attempts || 0);
|
||||
|
||||
if (status != "failed") {
|
||||
return { status, notice: null };
|
||||
return { status, attempts, notice: null };
|
||||
}
|
||||
|
||||
const notice = await notifyBotJobFailed({
|
||||
@@ -68,5 +69,5 @@ export default async function failBotJob({
|
||||
failure_message,
|
||||
});
|
||||
|
||||
return { status, notice };
|
||||
return { status, attempts, notice };
|
||||
}
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
import BunSQLite from "@moduletrace/bun-sqlite";
|
||||
|
||||
type Params = {
|
||||
bot_id: number | string;
|
||||
session_id: number | string;
|
||||
title: string;
|
||||
};
|
||||
|
||||
/**
|
||||
* The bot's task in this session with this title that's still queued or
|
||||
* working, if any, so the same work isn't started twice.
|
||||
*/
|
||||
export default async function grabOpenBotTask({ bot_id, session_id, title }: Params) {
|
||||
const res = await BunSQLite.sql<{ id: number; status: string }>({
|
||||
sql: `SELECT id, status FROM bots_queue
|
||||
WHERE bot_id = ? AND session_id = ? AND kind = 'task'
|
||||
AND status IN ('pending', 'processing')
|
||||
AND lower(trim(title)) = lower(trim(?))
|
||||
ORDER BY id DESC LIMIT 1`,
|
||||
values: [bot_id, session_id, title],
|
||||
});
|
||||
|
||||
return res.payload?.[0] || null;
|
||||
}
|
||||
@@ -0,0 +1,47 @@
|
||||
import BunSQLite from "@moduletrace/bun-sqlite";
|
||||
|
||||
export type BotSavedTextKind =
|
||||
| "core_memory"
|
||||
| "core_instructions"
|
||||
| "user_memory"
|
||||
| "workspace_memory"
|
||||
| "workspace_instructions";
|
||||
|
||||
type Params = {
|
||||
kind: BotSavedTextKind;
|
||||
bot_id: number | string;
|
||||
/**
|
||||
* Whose memory, for `user_memory`.
|
||||
*/
|
||||
user_id?: number | string;
|
||||
/**
|
||||
* Which workspace, for the workspace kinds.
|
||||
*/
|
||||
workspace_id?: number | string | null;
|
||||
};
|
||||
|
||||
/**
|
||||
* Reads back one of a bot's memories or instructions as stored, so a save
|
||||
* can be checked. Empty when there's none. The caller checks access.
|
||||
*/
|
||||
export default async function grabBotSavedText({ kind, bot_id, user_id, workspace_id }: Params) {
|
||||
const query =
|
||||
kind == "core_memory"
|
||||
? { sql: `SELECT memory AS text FROM bots WHERE id = ?`, values: [bot_id] }
|
||||
: kind == "core_instructions"
|
||||
? { sql: `SELECT system_prompt AS text FROM bots WHERE id = ?`, values: [bot_id] }
|
||||
: kind == "user_memory"
|
||||
? {
|
||||
sql: `SELECT memory AS text FROM bot_user_memories WHERE bot_id = ? AND user_id = ?`,
|
||||
values: [bot_id, user_id ?? 0],
|
||||
}
|
||||
: {
|
||||
sql: `SELECT ${kind == "workspace_memory" ? "memory" : "system_prompt"} AS text
|
||||
FROM workspace_bots WHERE bot_id = ? AND workspace_id = ?`,
|
||||
values: [bot_id, workspace_id ?? 0],
|
||||
};
|
||||
|
||||
const res = await BunSQLite.sql<{ text: string | null }>({ ...query, sql: `${query.sql} LIMIT 1` });
|
||||
|
||||
return res.payload?.[0]?.text || "";
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
import BunSQLite from "@moduletrace/bun-sqlite";
|
||||
|
||||
type Params = {
|
||||
cron_job_id: number;
|
||||
/**
|
||||
* The run's job: only the schedule's latest run records its failure.
|
||||
*/
|
||||
job_id: number;
|
||||
error: string;
|
||||
};
|
||||
|
||||
/**
|
||||
* Keeps why a schedule's latest run failed in `last_error`, so Routines
|
||||
* shows it. The next run that's queued clears it.
|
||||
*/
|
||||
export default async function recordCronJobFailure({ cron_job_id, job_id, error }: Params) {
|
||||
await BunSQLite.sql({
|
||||
sql: `UPDATE cron_jobs SET last_error = ?, updated_at = ? WHERE id = ? AND last_job_id = ?`,
|
||||
values: [`Last run failed: ${error}`, Date.now(), cron_job_id, job_id],
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
import grabBotSavedText, {
|
||||
type BotSavedTextKind,
|
||||
} from "@/src/functions/backend/db/bots/grab-bot-saved-text";
|
||||
|
||||
type Params = {
|
||||
kind: BotSavedTextKind;
|
||||
bot_id: number | string;
|
||||
user_id?: number | string;
|
||||
workspace_id?: number | string | null;
|
||||
/**
|
||||
* What the bot asked to save.
|
||||
*/
|
||||
expected: string;
|
||||
};
|
||||
|
||||
/**
|
||||
* Reads a saved memory or instructions back and throws unless it holds
|
||||
* what the bot sent, so the bot never reports a save that didn't take.
|
||||
*/
|
||||
export default async function checkSavedText({ expected, ...params }: Params) {
|
||||
const stored = await grabBotSavedText(params);
|
||||
|
||||
if (stored.trim() != expected.trim()) {
|
||||
throw new Error(`The save didn't take: what's stored doesn't match what you sent. Try again.`);
|
||||
}
|
||||
}
|
||||
@@ -63,6 +63,19 @@ export default function createBotTool({
|
||||
);
|
||||
}
|
||||
|
||||
const twin = (
|
||||
await BunSQLite.sql<BUN_SQLITE_COBBLE_BOTS>({
|
||||
sql: `SELECT id, name FROM bots WHERE lower(trim(name)) = lower(trim(?)) LIMIT 1`,
|
||||
values: [args.name],
|
||||
})
|
||||
).payload?.[0];
|
||||
|
||||
if (twin?.id) {
|
||||
throw new Error(
|
||||
`There's already a bot called ${twin.name} (${botRef({ id: twin.id })}). Change it with update_bot, or pick another name if it's really a different job.`,
|
||||
);
|
||||
}
|
||||
|
||||
const hues =
|
||||
(
|
||||
await BunSQLite.sql<BUN_SQLITE_COBBLE_BOTS>({
|
||||
|
||||
@@ -2,6 +2,8 @@ import { z } from "zod";
|
||||
import type { BUN_SQLITE_COBBLE_COMPUTERS } from "@/db/types/db";
|
||||
import { AppData } from "@/src/data/app-data";
|
||||
import createCronJob from "@/src/functions/backend/db/cron-jobs/create-cron-job";
|
||||
import grabCronJobs from "@/src/functions/backend/db/cron-jobs/grab-cron-jobs";
|
||||
import findDuplicateCronJob from "@/src/utils/find-duplicate-cron-job";
|
||||
import describeCronJob from "@/src/utils/describe-cron-job";
|
||||
import type { HarnessTool } from "../../types";
|
||||
|
||||
@@ -44,7 +46,7 @@ export default function createScheduleTool({ user_id, bot_id, workspace_id, comp
|
||||
computers.length
|
||||
? `Runs use the computer you're on when they start unless you pick one: ${listed}.`
|
||||
: `You have no computer here, so runs can't use one.`
|
||||
} Write the instructions as a full brief for a future you with no memory of this chat.`,
|
||||
} Write the instructions as a full brief for a future you with no memory of this chat. Runs carry the brief out without asking the user, so state the action itself ("send the summary to ..."), not a draft for them to approve, unless they asked to approve it.`,
|
||||
schema: z.object({
|
||||
title: z.string().min(1).max(80).describe(`A short label, e.g. "Morning inbox digest"`),
|
||||
instructions: z.string().min(1).max(8000),
|
||||
@@ -53,6 +55,20 @@ export default function createScheduleTool({ user_id, bot_id, workspace_id, comp
|
||||
max_runs: z.number().int().positive().optional().describe("Stop after this many runs; leave out to run until stopped"),
|
||||
}),
|
||||
run: async ({ args }) => {
|
||||
const twin = findDuplicateCronJob({
|
||||
cron_jobs: await grabCronJobs({ user_id, bot_id }),
|
||||
title: args.title,
|
||||
instructions: args.instructions,
|
||||
schedule: args.schedule,
|
||||
workspace_id,
|
||||
});
|
||||
|
||||
if (twin) {
|
||||
throw new Error(
|
||||
`This looks like a schedule you already have: ${describeCronJob({ cron_job: twin })}. Change it with update_schedule instead, or give the new one a different title if it really is separate work.`,
|
||||
);
|
||||
}
|
||||
|
||||
const cron_job = await createCronJob({
|
||||
user_id,
|
||||
bot_id,
|
||||
@@ -64,7 +80,7 @@ export default function createScheduleTool({ user_id, bot_id, workspace_id, comp
|
||||
max_runs: args.max_runs,
|
||||
});
|
||||
|
||||
return `Schedule saved: ${describeCronJob({ cron_job })}. Tell the user when it runs, in their time, and that they can pause or change it under Routines.`;
|
||||
return `Schedule saved: ${describeCronJob({ cron_job })}. Check it with list_schedules before you confirm it to the user. Tell the user when it runs, in their time, and that they can pause or change it under Routines.`;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -2,6 +2,7 @@ import { z } from "zod";
|
||||
import type { BUN_SQLITE_COBBLE_BOTS } from "@/db/types/db";
|
||||
import pushBotNotice from "@/src/bots-queue/push-bot-notice";
|
||||
import enqueueBotJob from "@/src/functions/backend/db/bots-queue/enqueue-bot-job";
|
||||
import grabOpenBotTask from "@/src/functions/backend/db/bots-queue/grab-open-bot-task";
|
||||
import insertBotMessage from "@/src/functions/backend/db/bots/insert-bot-message";
|
||||
import grabCrewWorkers from "@/src/functions/backend/db/bots/grab-crew-workers";
|
||||
import grabBotComputers from "@/src/functions/backend/db/computers/grab-bot-computers";
|
||||
@@ -79,6 +80,17 @@ export default function delegateTaskTool({
|
||||
|
||||
const computers = await grabBotComputers({ bot_id: worker.id, workspace_id });
|
||||
const worker_session = await ensureBotSession({ user_id, bot_id: worker.id, workspace_id });
|
||||
const open = await grabOpenBotTask({
|
||||
bot_id: worker.id,
|
||||
session_id: Number(worker_session.id),
|
||||
title: args.title,
|
||||
});
|
||||
|
||||
if (open) {
|
||||
throw new Error(
|
||||
`${worker.name} already has task #${open.id} "${args.title}" ${open.status == "pending" ? "queued" : "under way"}. Don't hand it over again; it's reported back here when they finish.`,
|
||||
);
|
||||
}
|
||||
|
||||
const job_id = await enqueueBotJob({
|
||||
user_id,
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
import { describe, expect, test } from "bun:test";
|
||||
import reportRunFailedTool from "./report-run-failed-tool";
|
||||
|
||||
const signal = new AbortController().signal;
|
||||
|
||||
describe("reportRunFailedTool", () => {
|
||||
test("passes the reason on", async () => {
|
||||
const reasons: string[] = [];
|
||||
const tool = reportRunFailedTool({ onFail: ({ reason }) => reasons.push(reason) });
|
||||
|
||||
await tool.run({ args: { reason: " SMTP refused the login. " }, signal });
|
||||
|
||||
expect(reasons).toEqual(["SMTP refused the login."]);
|
||||
});
|
||||
|
||||
test("rejects an empty reason", () => {
|
||||
const tool = reportRunFailedTool({ onFail: () => {} });
|
||||
|
||||
expect(tool.schema?.safeParse({ reason: " " }).success).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,34 @@
|
||||
import { z } from "zod";
|
||||
import type { HarnessTool } from "../../types";
|
||||
|
||||
type Params = {
|
||||
/**
|
||||
* Gets why the run failed. The task runner throws it after the run, so
|
||||
* the queue retries the job.
|
||||
*/
|
||||
onFail: (params: { reason: string }) => void;
|
||||
};
|
||||
|
||||
type Args = {
|
||||
reason: string;
|
||||
};
|
||||
|
||||
/**
|
||||
* Lets a scheduled run say it couldn't do what its brief asks (a send
|
||||
* that failed, a service that's down), so it's retried later instead of
|
||||
* posting a report. Only given to scheduled runs.
|
||||
*/
|
||||
export default function reportRunFailedTool({ onFail }: Params): HarnessTool<Args> {
|
||||
return {
|
||||
name: "report_run_failed",
|
||||
description: `Call this when this scheduled run couldn't do what its brief asks (a tool or service failed, something it needs is down). The run is tried again later, and the user is told only if every try fails. Say why in one or two sentences. Don't call it for something only the user can do; ask them to take over instead.`,
|
||||
schema: z.object({
|
||||
reason: z.string().trim().min(1).max(1000).describe("Why the run failed, for the user"),
|
||||
}),
|
||||
run: async ({ args }) => {
|
||||
onFail({ reason: args.reason.trim() });
|
||||
|
||||
return `Recorded. Stop here: the run will be tried again later. Don't write a report.`;
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
import { z } from "zod";
|
||||
import type { BUN_SQLITE_COBBLE_COMPUTERS } from "@/db/types/db";
|
||||
import enqueueBotJob from "@/src/functions/backend/db/bots-queue/enqueue-bot-job";
|
||||
import grabOpenBotTask from "@/src/functions/backend/db/bots-queue/grab-open-bot-task";
|
||||
import type { HarnessTool } from "../../types";
|
||||
|
||||
type Params = {
|
||||
@@ -54,6 +55,14 @@ export default function startTaskTool({
|
||||
computer_id: z.number().int().optional(),
|
||||
}),
|
||||
run: async ({ args }) => {
|
||||
const open = await grabOpenBotTask({ bot_id, session_id, title: args.title });
|
||||
|
||||
if (open) {
|
||||
throw new Error(
|
||||
`Task #${open.id} "${args.title}" is already ${open.status == "pending" ? "queued" : "under way"}. Don't start it again; its result will be posted here.`,
|
||||
);
|
||||
}
|
||||
|
||||
const computer_id = args.computer_id ?? computers[0]?.id ?? undefined;
|
||||
|
||||
const job_id = await enqueueBotJob({
|
||||
|
||||
@@ -3,6 +3,7 @@ import { AppData } from "@/src/data/app-data";
|
||||
import updateBotSystemPrompt from "@/src/functions/backend/db/bots/update-bot-system-prompt";
|
||||
import updateWorkspaceBot from "@/src/functions/backend/db/workspaces/update-workspace-bot";
|
||||
import type { HarnessTool } from "../../types";
|
||||
import checkSavedText from "./check-saved-text";
|
||||
|
||||
type Params = {
|
||||
user_id: number | string;
|
||||
@@ -47,7 +48,13 @@ export default function updateInstructionsTool({
|
||||
bot_id,
|
||||
system_prompt: args.instructions,
|
||||
});
|
||||
return `Instructions for this workspace saved. They apply from your next run here.`;
|
||||
await checkSavedText({
|
||||
kind: "workspace_instructions",
|
||||
bot_id,
|
||||
workspace_id,
|
||||
expected: args.instructions,
|
||||
});
|
||||
return `Instructions for this workspace saved and checked. They apply from your next run here.`;
|
||||
}
|
||||
|
||||
await updateBotSystemPrompt({
|
||||
@@ -55,7 +62,8 @@ export default function updateInstructionsTool({
|
||||
bot_id,
|
||||
system_prompt: args.instructions,
|
||||
});
|
||||
return `Core instructions saved. They apply from your next run, everywhere.`;
|
||||
await checkSavedText({ kind: "core_instructions", bot_id, expected: args.instructions });
|
||||
return `Core instructions saved and checked. They apply from your next run, everywhere.`;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -4,6 +4,7 @@ import updateBotMemory from "@/src/functions/backend/db/bots/update-bot-memory";
|
||||
import updateBotUserMemory from "@/src/functions/backend/db/bots/update-bot-user-memory";
|
||||
import updateWorkspaceBot from "@/src/functions/backend/db/workspaces/update-workspace-bot";
|
||||
import type { HarnessTool } from "../../types";
|
||||
import checkSavedText from "./check-saved-text";
|
||||
|
||||
type Params = {
|
||||
/**
|
||||
@@ -52,16 +53,19 @@ export default function updateMemoryTool({
|
||||
run: async ({ args }) => {
|
||||
if (args.scope == "core") {
|
||||
await updateBotMemory({ user_id, bot_id, memory: args.memory });
|
||||
return `Core memory saved.`;
|
||||
await checkSavedText({ kind: "core_memory", bot_id, expected: args.memory });
|
||||
return `Core memory saved and checked.`;
|
||||
}
|
||||
|
||||
if (workspace_id && args.scope == "workspace") {
|
||||
await updateWorkspaceBot({ user_id, workspace_id, bot_id, memory: args.memory });
|
||||
return `Memory for this workspace saved.`;
|
||||
await checkSavedText({ kind: "workspace_memory", bot_id, workspace_id, expected: args.memory });
|
||||
return `Memory for this workspace saved and checked.`;
|
||||
}
|
||||
|
||||
await updateBotUserMemory({ user_id, bot_id, memory: args.memory });
|
||||
return `Memory about the user saved.`;
|
||||
await checkSavedText({ kind: "user_memory", bot_id, user_id, expected: args.memory });
|
||||
return `Memory about the user saved and checked.`;
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
import { describe, expect, test } from "bun:test";
|
||||
import type { BUN_SQLITE_COBBLE_CRON_JOBS } from "@/db/types/db";
|
||||
import findDuplicateCronJob from "./find-duplicate-cron-job";
|
||||
|
||||
const digest: BUN_SQLITE_COBBLE_CRON_JOBS = {
|
||||
id: 2,
|
||||
title: "Morning digest",
|
||||
instructions: "Email the news digest.",
|
||||
schedule: "0 6 * * *",
|
||||
workspace_id: undefined,
|
||||
};
|
||||
|
||||
const find = (params: Partial<Parameters<typeof findDuplicateCronJob>[0]>) =>
|
||||
findDuplicateCronJob({
|
||||
cron_jobs: [digest],
|
||||
title: "Other",
|
||||
instructions: "Something else.",
|
||||
schedule: "0 7 * * *",
|
||||
...params,
|
||||
});
|
||||
|
||||
describe("findDuplicateCronJob", () => {
|
||||
test("matches the same title, ignoring case and spacing", () => {
|
||||
expect(find({ title: " morning DIGEST " })?.id).toBe(2);
|
||||
});
|
||||
|
||||
test("matches the same timing and brief under another title", () => {
|
||||
expect(find({ schedule: "0 6 * * *", instructions: "Email the news digest." })?.id).toBe(2);
|
||||
});
|
||||
|
||||
test("doesn't match the same brief at another time", () => {
|
||||
expect(find({ instructions: "Email the news digest." })).toBeNull();
|
||||
});
|
||||
|
||||
test("only matches schedules in the same place", () => {
|
||||
expect(find({ title: "Morning digest", workspace_id: 4 })).toBeNull();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,39 @@
|
||||
import type { BUN_SQLITE_COBBLE_CRON_JOBS } from "@/db/types/db";
|
||||
|
||||
type Params<T extends BUN_SQLITE_COBBLE_CRON_JOBS> = {
|
||||
/**
|
||||
* The bot's schedules for this user.
|
||||
*/
|
||||
cron_jobs: T[];
|
||||
title: string;
|
||||
instructions: string;
|
||||
schedule: string;
|
||||
/**
|
||||
* Where the new one would run: only schedules there can be its twin.
|
||||
*/
|
||||
workspace_id?: number | string | null;
|
||||
};
|
||||
|
||||
const normalize = (text?: string | null) => (text || "").trim().replace(/\s+/g, " ").toLowerCase();
|
||||
|
||||
/**
|
||||
* The schedule a new one would duplicate, if any: same place and either
|
||||
* the same title, or the same timing and brief.
|
||||
*/
|
||||
export default function findDuplicateCronJob<T extends BUN_SQLITE_COBBLE_CRON_JOBS>({
|
||||
cron_jobs,
|
||||
title,
|
||||
instructions,
|
||||
schedule,
|
||||
workspace_id,
|
||||
}: Params<T>) {
|
||||
return (
|
||||
cron_jobs.find(
|
||||
(cron_job) =>
|
||||
Number(cron_job.workspace_id || 0) == Number(workspace_id || 0) &&
|
||||
(normalize(cron_job.title) == normalize(title) ||
|
||||
(normalize(cron_job.schedule) == normalize(schedule) &&
|
||||
normalize(cron_job.instructions) == normalize(instructions))),
|
||||
) || null
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
import { describe, expect, test } from "bun:test";
|
||||
import { AppData } from "@/src/data/app-data";
|
||||
import scheduledRunRetryAt from "./scheduled-run-retry-at";
|
||||
|
||||
describe("scheduledRunRetryAt", () => {
|
||||
test("waits the retry delay after the first failure", () => {
|
||||
expect(scheduledRunRetryAt({ attempts: 1, now: 0 })).toBe(AppData.ScheduledRunRetryMs);
|
||||
});
|
||||
|
||||
test("doubles after each failure", () => {
|
||||
expect(scheduledRunRetryAt({ attempts: 2, now: 0 })).toBe(AppData.ScheduledRunRetryMs * 2);
|
||||
expect(scheduledRunRetryAt({ attempts: 3, now: 0 })).toBe(AppData.ScheduledRunRetryMs * 4);
|
||||
});
|
||||
|
||||
test("treats no attempts as the first", () => {
|
||||
expect(scheduledRunRetryAt({ attempts: 0, now: 10 })).toBe(10 + AppData.ScheduledRunRetryMs);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,19 @@
|
||||
import { AppData } from "@/src/data/app-data";
|
||||
|
||||
type Params = {
|
||||
/**
|
||||
* Attempts the job has used so far, the failed one included.
|
||||
*/
|
||||
attempts: number;
|
||||
now?: number;
|
||||
};
|
||||
|
||||
/**
|
||||
* When a failed scheduled run is tried again: `AppData.ScheduledRunRetryMs`
|
||||
* after the first failure, doubling after each one after that.
|
||||
*/
|
||||
export default function scheduledRunRetryAt({ attempts, now = Date.now() }: Params) {
|
||||
const step = Math.min(Math.max(attempts - 1, 0), 10);
|
||||
|
||||
return now + AppData.ScheduledRunRetryMs * 2 ** step;
|
||||
}
|
||||
Reference in New Issue
Block a user