diff --git a/AGENTS.md b/AGENTS.md index 17b2ff2..e2d3e57 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -68,7 +68,7 @@ Enables SQLite foreign keys (`enableForeignKeys`), `bunextInit()`, `ensureDocker ### 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 @@ -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`. - 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`). - **`start_task`/`delegate_task` jobs are runs, not tasks:** their results say `Run #` 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. diff --git a/db/schema.ts b/db/schema.ts index cbf105f..a9477fb 100644 --- a/db/schema.ts +++ b/db/schema.ts @@ -1106,6 +1106,11 @@ const schema: BUN_SQLITE_DatabaseSchemaType = { fieldDescription: `Unix timestamp of completion or final failure`, 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: [ { @@ -1116,7 +1121,7 @@ const schema: BUN_SQLITE_DatabaseSchemaType = { }, { 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: [ { fieldName: "bots_queue_id", diff --git a/db/types/db.ts b/db/types/db.ts index 64f8bd0..32b41bd 100644 --- a/db/types/db.ts +++ b/db/types/db.ts @@ -505,11 +505,11 @@ export type BUN_SQLITE_COBBLE_TASKS = { */ 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 | ""; /** - * 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 | ""; title?: string; @@ -679,6 +679,10 @@ export type BUN_SQLITE_COBBLE_BOTS_QUEUE = { * Unix timestamp of completion or final failure */ 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) */ diff --git a/src/cron/clear-job-steps/index.ts b/src/cron/clear-job-steps/index.ts new file mode 100644 index 0000000..3ae1c95 --- /dev/null +++ b/src/cron/clear-job-steps/index.ts @@ -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(); +} diff --git a/src/cron/index.ts b/src/cron/index.ts index 2c672c7..c3377c7 100644 --- a/src/cron/index.ts +++ b/src/cron/index.ts @@ -1,6 +1,7 @@ import bunext from "@moduletrace/bunext"; import cronHeartbeat from "./heartbeat"; import pruneBotsQueue from "./prune-bots-queue"; +import cronClearJobSteps from "./clear-job-steps"; import pruneBotHookEvents from "./prune-bot-hook-events"; import cronPruneChatrooms from "./prune-chatrooms"; import syncNotifyWatchers from "./sync-notify-watchers"; @@ -31,6 +32,13 @@ export default function cron() { task: pruneBotsQueue, }); + registerCronJob({ + title: "clear-job-steps", + schedule: "*/5 * * * *", + task: cronClearJobSteps, + quiet: true, + }); + registerCronJob({ title: "prune-bot-hook-events", schedule: "@daily", diff --git a/src/data/app-data.ts b/src/data/app-data.ts index 1249d39..7ea2d42 100644 --- a/src/data/app-data.ts +++ b/src/data/app-data.ts @@ -414,6 +414,12 @@ export const AppData = { * working-on panel shows. */ 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 * bot changes from the database for its own sockets (`liveRelay`). diff --git a/src/functions/backend/db/bots-queue/clear-finished-job-steps-sql.test.ts b/src/functions/backend/db/bots-queue/clear-finished-job-steps-sql.test.ts new file mode 100644 index 0000000..28e4ee8 --- /dev/null +++ b/src/functions/backend/db/bots-queue/clear-finished-job-steps-sql.test.ts @@ -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); + }); +}); diff --git a/src/functions/backend/db/bots-queue/clear-finished-job-steps-sql.ts b/src/functions/backend/db/bots-queue/clear-finished-job-steps-sql.ts new file mode 100644 index 0000000..4075e48 --- /dev/null +++ b/src/functions/backend/db/bots-queue/clear-finished-job-steps-sql.ts @@ -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], + }, + }; +} diff --git a/src/functions/backend/db/bots-queue/clear-finished-job-steps.ts b/src/functions/backend/db/bots-queue/clear-finished-job-steps.ts new file mode 100644 index 0000000..225eb0f --- /dev/null +++ b/src/functions/backend/db/bots-queue/clear-finished-job-steps.ts @@ -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); +} diff --git a/src/functions/backend/db/usage/grab-message-details.ts b/src/functions/backend/db/usage/grab-message-details.ts index 8ed34d0..188f6d4 100644 --- a/src/functions/backend/db/usage/grab-message-details.ts +++ b/src/functions/backend/db/usage/grab-message-details.ts @@ -105,7 +105,7 @@ async function grabJob({ job_id }: { job_id: number }): Promise({ 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`, values: [job_id], }) diff --git a/src/types/index.ts b/src/types/index.ts index 3f917e4..043ff15 100755 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -1581,7 +1581,7 @@ export type MessageDetailsJob = { started_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; };