Clear finished jobs' steps a minute after they end
Job steps (bots_queue_updates) only feed the live working row, so the new clear-job-steps cron deletes them shortly after a job ends, however it ended, keeping their count on bots_queue.steps for the message details. Not done the moment a job ends: the live relay would miss the last ones. Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
@@ -68,7 +68,7 @@ Enables SQLite foreign keys (`enableForeignKeys`), `bunextInit()`, `ensureDocker
|
|||||||
|
|
||||||
### Cron
|
### Cron
|
||||||
|
|
||||||
`src/cron/index.ts` registers everything via `registerCronJob` (`src/cron/register-cron-job.ts`); one folder per task. In-process `Bun.cron`, UTC, never overlapping. Notable: `heartbeat`, `sync-harness-models` (hourly), `run-cron-jobs` (every minute, fires schedules), `sync-computers`, `sync-mcp-servers`, `prune-bots-queue`, `prune-chatrooms`.
|
`src/cron/index.ts` registers everything via `registerCronJob` (`src/cron/register-cron-job.ts`); one folder per task. In-process `Bun.cron`, UTC, never overlapping. Notable: `heartbeat`, `sync-harness-models` (hourly), `run-cron-jobs` (every minute, fires schedules), `sync-computers`, `sync-mcp-servers`, `clear-job-steps`, `prune-bots-queue`, `prune-chatrooms`.
|
||||||
|
|
||||||
### Admin CRUD API
|
### Admin CRUD API
|
||||||
|
|
||||||
@@ -113,7 +113,7 @@ Enables SQLite foreign keys (`enableForeignKeys`), `bunextInit()`, `ensureDocker
|
|||||||
|
|
||||||
- Kinds (`BotJobKinds`): `reply` (chat answer, no task row) and `task` (long work, usually on a computer). bun-sqlite `isQueue` table. Helpers in `src/functions/backend/db/bots-queue/`: `enqueueBotJob`, `claimBotJobs` (kind-filtered atomic claim), `updateBotJob` (false once lost/cancelled), `completeBotJob`, `failBotJob` (backoff or fatal), `deferBotJob` (all models resting: attempt refunded, face `asleep`), `cancelBotJob`, `stop-bot-jobs`.
|
- Kinds (`BotJobKinds`): `reply` (chat answer, no task row) and `task` (long work, usually on a computer). bun-sqlite `isQueue` table. Helpers in `src/functions/backend/db/bots-queue/`: `enqueueBotJob`, `claimBotJobs` (kind-filtered atomic claim), `updateBotJob` (false once lost/cancelled), `completeBotJob`, `failBotJob` (backoff or fatal), `deferBotJob` (all models resting: attempt refunded, face `asleep`), `cancelBotJob`, `stop-bot-jobs`.
|
||||||
- Handler `src/bots-queue/index.ts`: separate reply/task slots, heartbeat, per-attempt deadline (`AppData.BotJobAttemptTimeoutMs`), one task per computer and one reply per session (`createKeyLock`). Work in `run-bot-job.ts` -> `run-bot-reply.ts` / `run-bot-task.ts`. Throw `BotJobFatalError` for failures retrying can't fix.
|
- Handler `src/bots-queue/index.ts`: separate reply/task slots, heartbeat, per-attempt deadline (`AppData.BotJobAttemptTimeoutMs`), one task per computer and one reply per session (`createKeyLock`). Work in `run-bot-job.ts` -> `run-bot-reply.ts` / `run-bot-task.ts`. Throw `BotJobFatalError` for failures retrying can't fix.
|
||||||
- Every model turn is a step in `bots_queue_updates` (tool names, never arguments; `describe-job-step.ts`, labels in `JobStepToolLabels`), pushed as `server:bot-job-update`.
|
- Every model turn is a step in `bots_queue_updates` (tool names, never arguments; `describe-job-step.ts`, labels in `JobStepToolLabels`), pushed as `server:bot-job-update` by the live relay only. Steps feed the live working row alone: the `clear-job-steps` cron deletes them `AppData.BotJobStepsKeepMs` after the job ends, however it ended, keeping their count on `bots_queue.steps`. Never delete them the moment a job ends (the relay would miss the last ones).
|
||||||
- **A bot can't vouch for itself:** a run's tool calls (`tool_uses`, names and ok, never arguments) go to the wrap-up turn when steps run out and onto the result message as `bot_messages.tool_log`; history follows each recorded bot message with an `[App record: ...]` line (`botMessagesToHarness`), and copies of that line in an answer are stripped (`stripAppRecords`).
|
- **A bot can't vouch for itself:** a run's tool calls (`tool_uses`, names and ok, never arguments) go to the wrap-up turn when steps run out and onto the result message as `bot_messages.tool_log`; history follows each recorded bot message with an `[App record: ...]` line (`botMessagesToHarness`), and copies of that line in an answer are stripped (`stripAppRecords`).
|
||||||
- **`start_task`/`delegate_task` jobs are runs, not tasks:** their results say `Run #<job id>` and that `list_tasks` doesn't show them (bots sent users to the tracker for them). `start_task` refuses a second open run in its session on the same computer unless `separate` (it would only wait behind it) and says how many runs are ahead on that computer.
|
- **`start_task`/`delegate_task` jobs are runs, not tasks:** their results say `Run #<job id>` and that `list_tasks` doesn't show them (bots sent users to the tracker for them). `start_task` refuses a second open run in its session on the same computer unless `separate` (it would only wait behind it) and says how many runs are ahead on that computer.
|
||||||
- **Replies take in messages sent mid-run** (`incoming`, `IncomingMessagesNote`, redraft); separate requests are acknowledged with `send_message` (interim). A queued reply finishes silently only when a finished reply already read every user message (`hasUnansweredMessage`), never by who wrote last.
|
- **Replies take in messages sent mid-run** (`incoming`, `IncomingMessagesNote`, redraft); separate requests are acknowledged with `send_message` (interim). A queued reply finishes silently only when a finished reply already read every user message (`hasUnansweredMessage`), never by who wrote last.
|
||||||
|
|||||||
+6
-1
@@ -1106,6 +1106,11 @@ const schema: BUN_SQLITE_DatabaseSchemaType = {
|
|||||||
fieldDescription: `Unix timestamp of completion or final failure`,
|
fieldDescription: `Unix timestamp of completion or final failure`,
|
||||||
dataType: "INTEGER",
|
dataType: "INTEGER",
|
||||||
},
|
},
|
||||||
|
{
|
||||||
|
fieldName: "steps",
|
||||||
|
fieldDescription: `How many steps it recorded, written when its \`bots_queue_updates\` are cleared after it ends (null while they're still there)`,
|
||||||
|
dataType: "INTEGER",
|
||||||
|
},
|
||||||
],
|
],
|
||||||
indexes: [
|
indexes: [
|
||||||
{
|
{
|
||||||
@@ -1116,7 +1121,7 @@ const schema: BUN_SQLITE_DatabaseSchemaType = {
|
|||||||
},
|
},
|
||||||
{
|
{
|
||||||
tableName: "bots_queue_updates",
|
tableName: "bots_queue_updates",
|
||||||
tableDescription: `What a bot did while working through a queued job, one row per step, oldest first. Live in the working-on panel while the job runs; deleted with the job when it's pruned`,
|
tableDescription: `What a bot did while working through a queued job, one row per step, oldest first. Live in the working-on panel while the job runs; cleared shortly after the job ends (clearFinishedJobSteps), its count kept on the job`,
|
||||||
fields: [
|
fields: [
|
||||||
{
|
{
|
||||||
fieldName: "bots_queue_id",
|
fieldName: "bots_queue_id",
|
||||||
|
|||||||
+6
-2
@@ -505,11 +505,11 @@ export type BUN_SQLITE_COBBLE_TASKS = {
|
|||||||
*/
|
*/
|
||||||
updated_at?: number | "";
|
updated_at?: number | "";
|
||||||
/**
|
/**
|
||||||
* The bot the task belongs to; null for a workspace's task
|
* The bot that has the task: its own list's, or with `workspace_id` its task in that workspace; null for a workspace's shared task nobody has yet
|
||||||
*/
|
*/
|
||||||
bot_id?: number | "";
|
bot_id?: number | "";
|
||||||
/**
|
/**
|
||||||
* The workspace the task belongs to (its bots all see it); null for a bot's own
|
* The workspace the task is in; null for a bot's own list
|
||||||
*/
|
*/
|
||||||
workspace_id?: number | "";
|
workspace_id?: number | "";
|
||||||
title?: string;
|
title?: string;
|
||||||
@@ -679,6 +679,10 @@ export type BUN_SQLITE_COBBLE_BOTS_QUEUE = {
|
|||||||
* Unix timestamp of completion or final failure
|
* Unix timestamp of completion or final failure
|
||||||
*/
|
*/
|
||||||
finished_at?: number | "";
|
finished_at?: number | "";
|
||||||
|
/**
|
||||||
|
* How many steps it recorded, written when its `bots_queue_updates` are cleared after it ends (null while they're still there)
|
||||||
|
*/
|
||||||
|
steps?: number | "";
|
||||||
/**
|
/**
|
||||||
* The state of the job. (pending | processing | completed | failed)
|
* The state of the job. (pending | processing | completed | failed)
|
||||||
*/
|
*/
|
||||||
|
|||||||
@@ -0,0 +1,8 @@
|
|||||||
|
import clearFinishedJobSteps from "@/src/functions/backend/db/bots-queue/clear-finished-job-steps";
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Clears the steps of jobs that have ended (`clearFinishedJobSteps`).
|
||||||
|
*/
|
||||||
|
export default async function cronClearJobSteps() {
|
||||||
|
return await clearFinishedJobSteps();
|
||||||
|
}
|
||||||
@@ -1,6 +1,7 @@
|
|||||||
import bunext from "@moduletrace/bunext";
|
import bunext from "@moduletrace/bunext";
|
||||||
import cronHeartbeat from "./heartbeat";
|
import cronHeartbeat from "./heartbeat";
|
||||||
import pruneBotsQueue from "./prune-bots-queue";
|
import pruneBotsQueue from "./prune-bots-queue";
|
||||||
|
import cronClearJobSteps from "./clear-job-steps";
|
||||||
import pruneBotHookEvents from "./prune-bot-hook-events";
|
import pruneBotHookEvents from "./prune-bot-hook-events";
|
||||||
import cronPruneChatrooms from "./prune-chatrooms";
|
import cronPruneChatrooms from "./prune-chatrooms";
|
||||||
import syncNotifyWatchers from "./sync-notify-watchers";
|
import syncNotifyWatchers from "./sync-notify-watchers";
|
||||||
@@ -31,6 +32,13 @@ export default function cron() {
|
|||||||
task: pruneBotsQueue,
|
task: pruneBotsQueue,
|
||||||
});
|
});
|
||||||
|
|
||||||
|
registerCronJob({
|
||||||
|
title: "clear-job-steps",
|
||||||
|
schedule: "*/5 * * * *",
|
||||||
|
task: cronClearJobSteps,
|
||||||
|
quiet: true,
|
||||||
|
});
|
||||||
|
|
||||||
registerCronJob({
|
registerCronJob({
|
||||||
title: "prune-bot-hook-events",
|
title: "prune-bot-hook-events",
|
||||||
schedule: "@daily",
|
schedule: "@daily",
|
||||||
|
|||||||
@@ -414,6 +414,12 @@ export const AppData = {
|
|||||||
* working-on panel shows.
|
* working-on panel shows.
|
||||||
*/
|
*/
|
||||||
BotJobUpdatesShown: 12,
|
BotJobUpdatesShown: 12,
|
||||||
|
/**
|
||||||
|
* How long a finished job's steps stay before they're cleared
|
||||||
|
* (`clear-job-steps` cron). Long enough for every process's live relay
|
||||||
|
* to have pushed the last ones (`LiveRelayPollMs`).
|
||||||
|
*/
|
||||||
|
BotJobStepsKeepMs: 60 * 1000,
|
||||||
/**
|
/**
|
||||||
* How often each server process reads new bot messages, job steps and
|
* How often each server process reads new bot messages, job steps and
|
||||||
* bot changes from the database for its own sockets (`liveRelay`).
|
* bot changes from the database for its own sockets (`liveRelay`).
|
||||||
|
|||||||
@@ -0,0 +1,99 @@
|
|||||||
|
import { Database } from "bun:sqlite";
|
||||||
|
import { describe, expect, test } from "bun:test";
|
||||||
|
import clearFinishedJobStepsSql from "./clear-finished-job-steps-sql";
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Clearing a finished job's steps must keep their count on the job (the
|
||||||
|
* message details show it) and never touch a job that can still run: its
|
||||||
|
* working row reads them.
|
||||||
|
*/
|
||||||
|
function setup() {
|
||||||
|
const db = new Database(":memory:");
|
||||||
|
|
||||||
|
db.run(
|
||||||
|
`CREATE TABLE bots_queue (id INTEGER PRIMARY KEY, status TEXT, finished_at INTEGER, updated_at INTEGER, steps INTEGER)`,
|
||||||
|
);
|
||||||
|
db.run(`CREATE TABLE bots_queue_updates (id INTEGER PRIMARY KEY, bots_queue_id INTEGER)`);
|
||||||
|
db.run(`INSERT INTO bots_queue VALUES
|
||||||
|
(1, 'completed', 100, 100, NULL),
|
||||||
|
(2, 'failed', 100, 100, NULL),
|
||||||
|
(3, 'processing', NULL, 100, NULL),
|
||||||
|
(4, 'pending', NULL, 100, NULL),
|
||||||
|
(5, 'completed', 900, 900, NULL),
|
||||||
|
(6, 'failed', NULL, 100, NULL),
|
||||||
|
(7, 'completed', 100, 100, NULL)`);
|
||||||
|
db.run(`INSERT INTO bots_queue_updates (bots_queue_id) VALUES (1), (1), (1), (2), (3), (3), (4), (5), (6)`);
|
||||||
|
|
||||||
|
return db;
|
||||||
|
}
|
||||||
|
|
||||||
|
function run({ db, before }: { db: Database; before: number }) {
|
||||||
|
const { count, clear } = clearFinishedJobStepsSql({ before });
|
||||||
|
|
||||||
|
db.query(count.sql).run(...count.values);
|
||||||
|
db.query(clear.sql).run(...clear.values);
|
||||||
|
}
|
||||||
|
|
||||||
|
function stepsLeft({ db }: { db: Database }) {
|
||||||
|
return db
|
||||||
|
.query(`SELECT bots_queue_id AS id FROM bots_queue_updates ORDER BY id`)
|
||||||
|
.all()
|
||||||
|
.map((row) => (row as { id: number }).id);
|
||||||
|
}
|
||||||
|
|
||||||
|
function counts({ db }: { db: Database }) {
|
||||||
|
return Object.fromEntries(
|
||||||
|
db
|
||||||
|
.query(`SELECT id, steps FROM bots_queue ORDER BY id`)
|
||||||
|
.all()
|
||||||
|
.map((row) => [(row as { id: number }).id, (row as { steps: number | null }).steps]),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
describe("clearFinishedJobStepsSql", () => {
|
||||||
|
test("clears the steps of jobs that ended before the cutoff, completed or failed", () => {
|
||||||
|
const db = setup();
|
||||||
|
run({ db, before: 500 });
|
||||||
|
expect(stepsLeft({ db })).not.toContain(1);
|
||||||
|
expect(stepsLeft({ db })).not.toContain(2);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("keeps how many steps each job recorded", () => {
|
||||||
|
const db = setup();
|
||||||
|
run({ db, before: 500 });
|
||||||
|
expect(counts({ db })[1]).toBe(3);
|
||||||
|
expect(counts({ db })[2]).toBe(1);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("never touches a job that can still run", () => {
|
||||||
|
const db = setup();
|
||||||
|
run({ db, before: 500 });
|
||||||
|
expect(stepsLeft({ db })).toEqual([3, 3, 4, 5]);
|
||||||
|
expect(counts({ db })[3]).toBeNull();
|
||||||
|
expect(counts({ db })[4]).toBeNull();
|
||||||
|
});
|
||||||
|
|
||||||
|
test("leaves a job that ended after the cutoff for the next sweep", () => {
|
||||||
|
const db = setup();
|
||||||
|
run({ db, before: 500 });
|
||||||
|
expect(counts({ db })[5]).toBeNull();
|
||||||
|
run({ db, before: 1000 });
|
||||||
|
expect(stepsLeft({ db })).not.toContain(5);
|
||||||
|
expect(counts({ db })[5]).toBe(1);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("a job lost without a finish time is still cleared by when it last changed", () => {
|
||||||
|
const db = setup();
|
||||||
|
run({ db, before: 500 });
|
||||||
|
expect(stepsLeft({ db })).not.toContain(6);
|
||||||
|
expect(counts({ db })[6]).toBe(1);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("a job with no steps keeps a null count, and a second sweep changes nothing", () => {
|
||||||
|
const db = setup();
|
||||||
|
run({ db, before: 500 });
|
||||||
|
run({ db, before: 500 });
|
||||||
|
expect(counts({ db })[7]).toBeNull();
|
||||||
|
expect(counts({ db })[1]).toBe(3);
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -0,0 +1,29 @@
|
|||||||
|
type Params = {
|
||||||
|
/**
|
||||||
|
* Jobs that ended before this (unix ms) have their steps cleared.
|
||||||
|
*/
|
||||||
|
before: number;
|
||||||
|
};
|
||||||
|
|
||||||
|
/**
|
||||||
|
* The two statements that clear the steps (`bots_queue_updates`) of jobs
|
||||||
|
* that ended before `before`: first each job keeps how many it recorded
|
||||||
|
* (`bots_queue.steps`, for the message details), then the steps go. A job
|
||||||
|
* that's still open, or one retrying after a failed attempt, keeps its
|
||||||
|
* steps. Run in this order: the count reads the rows the delete removes.
|
||||||
|
*/
|
||||||
|
export default function clearFinishedJobStepsSql({ before }: Params) {
|
||||||
|
const finished = `SELECT id FROM bots_queue WHERE status IN ('completed', 'failed') AND COALESCE(finished_at, updated_at) < ?`;
|
||||||
|
|
||||||
|
return {
|
||||||
|
count: {
|
||||||
|
sql: `UPDATE bots_queue SET steps = COALESCE(steps, 0) + (SELECT COUNT(*) FROM bots_queue_updates u WHERE u.bots_queue_id = bots_queue.id)
|
||||||
|
WHERE id IN (${finished}) AND EXISTS (SELECT 1 FROM bots_queue_updates u WHERE u.bots_queue_id = bots_queue.id)`,
|
||||||
|
values: [before],
|
||||||
|
},
|
||||||
|
clear: {
|
||||||
|
sql: `DELETE FROM bots_queue_updates WHERE bots_queue_id IN (${finished})`,
|
||||||
|
values: [before],
|
||||||
|
},
|
||||||
|
};
|
||||||
|
}
|
||||||
@@ -0,0 +1,21 @@
|
|||||||
|
import BunSQLite from "@moduletrace/bun-sqlite";
|
||||||
|
import { AppData } from "@/src/data/app-data";
|
||||||
|
import clearFinishedJobStepsSql from "./clear-finished-job-steps-sql";
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Clears the steps of jobs that ended more than `AppData.BotJobStepsKeepMs`
|
||||||
|
* ago, however they ended (completed, failed, cancelled, stopped, lost).
|
||||||
|
* Steps only feed the live working row; once the job's message is in they
|
||||||
|
* have no reader, and the job keeps their count. Not done the moment a job
|
||||||
|
* ends: steps reach the page only through the live relay, which would
|
||||||
|
* miss the last ones.
|
||||||
|
*/
|
||||||
|
export default async function clearFinishedJobSteps() {
|
||||||
|
const { count, clear } = clearFinishedJobStepsSql({
|
||||||
|
before: Date.now() - AppData.BotJobStepsKeepMs,
|
||||||
|
});
|
||||||
|
|
||||||
|
await BunSQLite.sql(count);
|
||||||
|
|
||||||
|
return await BunSQLite.sql(clear);
|
||||||
|
}
|
||||||
@@ -105,7 +105,7 @@ async function grabJob({ job_id }: { job_id: number }): Promise<MessageDetailsJo
|
|||||||
const row = (
|
const row = (
|
||||||
await BunSQLite.sql<JobRow>({
|
await BunSQLite.sql<JobRow>({
|
||||||
sql: `SELECT q.id, q.kind, q.title, q.status, q.attempts, q.started_at, q.finished_at,
|
sql: `SELECT q.id, q.kind, q.title, q.status, q.attempts, q.started_at, q.finished_at,
|
||||||
(SELECT COUNT(*) FROM bots_queue_updates u WHERE u.bots_queue_id = q.id) AS steps
|
COALESCE(q.steps, 0) + (SELECT COUNT(*) FROM bots_queue_updates u WHERE u.bots_queue_id = q.id) AS steps
|
||||||
FROM bots_queue q WHERE q.id = ? LIMIT 1`,
|
FROM bots_queue q WHERE q.id = ? LIMIT 1`,
|
||||||
values: [job_id],
|
values: [job_id],
|
||||||
})
|
})
|
||||||
|
|||||||
+1
-1
@@ -1581,7 +1581,7 @@ export type MessageDetailsJob = {
|
|||||||
started_at: number | null;
|
started_at: number | null;
|
||||||
finished_at: number | null;
|
finished_at: number | null;
|
||||||
/**
|
/**
|
||||||
* How many model turns and notes it recorded (`bots_queue_updates`).
|
* How many model turns and notes it recorded (`bots_queue.steps` once cleared, plus any `bots_queue_updates` left).
|
||||||
*/
|
*/
|
||||||
steps: number;
|
steps: number;
|
||||||
};
|
};
|
||||||
|
|||||||
Reference in New Issue
Block a user