Let a user stop a bot, and say the app is still alpha
The thinking row and the header menu both carry Stop while a bot has work open for the user anywhere: every pending or processing job of that bot in their conversations is failed as stopped, its task closed, a Stopped step recorded and one closing message posted per session, so the thread stops waiting and the bot goes idle. The handler running the job is aborted from this process through a registry, so a job stuck waiting on a model stops at once instead of on its next heartbeat, and a late face change after the abort is dropped. The release stage is now said in the app too: Settings > General reports "Cobble v0.1.5 (alpha)" next to development, and the README and RELEASING.md say what alpha means for installs.
This commit is contained in:
@@ -0,0 +1,69 @@
|
||||
import { Database } from "bun:sqlite";
|
||||
import { describe, expect, test } from "bun:test";
|
||||
import openBotJobsSql from "./open-bot-jobs-sql";
|
||||
import type { SqlCondition } from "@/src/utils/crew-access/types";
|
||||
|
||||
/**
|
||||
* What "stop the bot" pulls the plug on: a bot's open jobs for one user,
|
||||
* across every one of their conversations with it. The test runs the
|
||||
* condition against real SQLite, since a wrong join would either leave a
|
||||
* job running (a stuck bot stays stuck) or stop another user's work.
|
||||
*/
|
||||
const db = new Database(":memory:");
|
||||
|
||||
db.run(`CREATE TABLE bot_sessions (id INTEGER PRIMARY KEY, user_id INTEGER, bot_id INTEGER)`);
|
||||
db.run(
|
||||
`CREATE TABLE bots_queue (id INTEGER PRIMARY KEY, bot_id INTEGER, session_id INTEGER, status TEXT, kind TEXT, title TEXT)`,
|
||||
);
|
||||
db.run(`INSERT INTO bot_sessions VALUES (100, 7, 2), (101, 7, 2), (102, 8, 2), (103, 9, 3)`);
|
||||
db.run(`INSERT INTO bots_queue VALUES
|
||||
(1, 2, 100, 'pending', 'reply', 'Thinking'),
|
||||
(2, 2, 101, 'processing', 'task', 'The audit'),
|
||||
(3, 2, 102, 'processing', 'task', 'Another user''s job'),
|
||||
(4, 2, 100, 'completed', 'reply', 'Thinking'),
|
||||
(5, 2, 100, 'failed', 'task', 'An old failure'),
|
||||
(6, 3, 103, 'pending', 'reply', 'Thinking'),
|
||||
(7, 3, 103, 'processing', 'task', 'A finished bot''s job')`);
|
||||
|
||||
function ids(condition: SqlCondition) {
|
||||
return db
|
||||
.query(`SELECT q.id FROM bots_queue q WHERE ${condition.sql} ORDER BY q.id`)
|
||||
.all(...condition.values)
|
||||
.map((row) => (row as { id: number }).id);
|
||||
}
|
||||
|
||||
describe("openBotJobsSql", () => {
|
||||
test("takes every open job of the bot across the user's conversations, tasks and replies alike", () => {
|
||||
expect(ids(openBotJobsSql({ bot_id: 2, user_id: 7 }))).toEqual([1, 2]);
|
||||
});
|
||||
|
||||
test("never takes another user's job, even on the same bot", () => {
|
||||
expect(ids(openBotJobsSql({ bot_id: 2, user_id: 7 }))).not.toContain(3);
|
||||
});
|
||||
|
||||
test("leaves finished jobs alone, so the work they did still counts", () => {
|
||||
expect(ids(openBotJobsSql({ bot_id: 2, user_id: 7 }))).not.toContain(4);
|
||||
expect(ids(openBotJobsSql({ bot_id: 2, user_id: 7 }))).not.toContain(5);
|
||||
});
|
||||
|
||||
test("another bot's jobs are never in it", () => {
|
||||
expect(ids(openBotJobsSql({ bot_id: 2, user_id: 7 }))).not.toContain(6);
|
||||
expect(ids(openBotJobsSql({ bot_id: 3, user_id: 9 }))).toEqual([6, 7]);
|
||||
});
|
||||
|
||||
test("a user with nothing open stops nothing", () => {
|
||||
expect(ids(openBotJobsSql({ bot_id: 2, user_id: 8 }))).toEqual([3]);
|
||||
expect(ids(openBotJobsSql({ bot_id: 2, user_id: 99 }))).toEqual([]);
|
||||
});
|
||||
|
||||
test("reads a join alias when one is given", () => {
|
||||
const condition = openBotJobsSql({ bot_id: 2, user_id: 7, alias: "j" });
|
||||
|
||||
expect(
|
||||
db
|
||||
.query(`SELECT j.id FROM bots_queue j WHERE ${condition.sql} ORDER BY j.id`)
|
||||
.all(...condition.values)
|
||||
.map((row) => (row as { id: number }).id),
|
||||
).toEqual([1, 2]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,29 @@
|
||||
import type { SqlCondition } from "@/src/utils/crew-access/types";
|
||||
|
||||
type Params = {
|
||||
bot_id: number | string;
|
||||
/**
|
||||
* Only jobs in this user's conversations, never another user's.
|
||||
*/
|
||||
user_id: number | string;
|
||||
/**
|
||||
* The `bots_queue` alias the condition reads.
|
||||
*/
|
||||
alias?: string;
|
||||
};
|
||||
|
||||
/**
|
||||
* The SQL for a bot's jobs for one user that can still run or are running:
|
||||
* pending (including one held back with `available_at` until a model comes
|
||||
* back) and processing (including one waiting for its computer). A job's
|
||||
* session is what makes it the user's, so another user's job is never in
|
||||
* it. This is what "stop the bot" pulls the plug on: with none of these
|
||||
* left, nothing of the bot's starts again until the user asks.
|
||||
*/
|
||||
export default function openBotJobsSql({ bot_id, user_id, alias = "q" }: Params): SqlCondition {
|
||||
return {
|
||||
sql: `(${alias}.bot_id = ? AND ${alias}.status IN ('pending', 'processing')
|
||||
AND ${alias}.session_id IN (SELECT id FROM bot_sessions WHERE user_id = ?))`,
|
||||
values: [Number(bot_id), Number(user_id)],
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
import { Database } from "bun:sqlite";
|
||||
import { describe, expect, test } from "bun:test";
|
||||
import stopBotJobs from "./stop-bot-jobs";
|
||||
|
||||
/**
|
||||
* Stopping the bot from the chat stops every open job of it in the user's
|
||||
* conversations and nothing else: another user's work, another bot's work
|
||||
* and finished jobs all survive, and the bot's own handler must find out
|
||||
* (its lock is gone) so it drops what it was producing.
|
||||
*
|
||||
* The helper talks to bun-sqlite's process-wide client, so this suite has to
|
||||
* drive a real one. It's skipped where the client isn't set up rather than
|
||||
* failing the other suites; `open-bot-jobs-sql.test.ts` covers the
|
||||
* selection itself against real SQLite unconditionally.
|
||||
*/
|
||||
const sqlite_ready = (() => {
|
||||
try {
|
||||
new Database(":memory:").run(`SELECT 1`);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
})();
|
||||
|
||||
describe.skipIf(!sqlite_ready)("stopBotJobs", () => {
|
||||
test("a user with nothing open is a no-op", async () => {
|
||||
const stopped = await stopBotJobs({ bot_id: 999_999, user_id: 999_999 });
|
||||
|
||||
expect(stopped).toBeNull();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,130 @@
|
||||
import BunSQLite from "@moduletrace/bun-sqlite";
|
||||
import type { BUN_SQLITE_COBBLE_BOT_MESSAGES, BUN_SQLITE_COBBLE_BOTS_QUEUE } from "@/db/types/db";
|
||||
import { BotJobStoppedError, BotJobStoppedStep } from "@/src/dict/bot-queue-dict";
|
||||
import insertBotMessage from "../bots/insert-bot-message";
|
||||
import setBotFace, { type BotFaceChange } from "../bots/set-bot-face";
|
||||
import { abortRunningJob } from "@/src/bots-queue/job-abort-registry";
|
||||
import insertBotJobUpdate from "./insert-bot-job-update";
|
||||
import openBotJobsSql from "./open-bot-jobs-sql";
|
||||
import stoppedJobReport, { pickReportedJobs } from "./stopped-job-report";
|
||||
|
||||
type Params = {
|
||||
bot_id: number | string;
|
||||
/**
|
||||
* Whose conversations stop. Never another user's jobs.
|
||||
*/
|
||||
user_id: number | string;
|
||||
};
|
||||
|
||||
export type StoppedBotJobs = {
|
||||
/**
|
||||
* The jobs that were open and are now failed as stopped.
|
||||
*/
|
||||
jobs: BUN_SQLITE_COBBLE_BOTS_QUEUE[];
|
||||
/**
|
||||
* One closing message per session the bot worked in, for the live
|
||||
* push. It's what ends the thread's "still thinking" state: the last
|
||||
* word is the bot's again.
|
||||
*/
|
||||
messages: BUN_SQLITE_COBBLE_BOT_MESSAGES[];
|
||||
/**
|
||||
* The user's face change (idle), for the live push.
|
||||
*/
|
||||
bot: BotFaceChange;
|
||||
};
|
||||
|
||||
/**
|
||||
* Pulls the plug on a bot for one user: every job of it that could still
|
||||
* run or is running in their conversations (replies it still owes, tasks
|
||||
* it's on, across all of them) is failed as stopped, its task closed. A
|
||||
* running handler finds out on its next `updateBotJob` (which returns
|
||||
* false) and drops whatever it was producing. Each job gets a closing
|
||||
* step, so a thread's working row says why it went quiet even after a
|
||||
* reload, one message per session ends it out loud, and the bot goes
|
||||
* idle. It's not a bus: nothing restarts by itself, so a stopped bot
|
||||
* works again only when the user asks. A schedule the user set up still
|
||||
* comes round, since that is work they asked for.
|
||||
*/
|
||||
export default async function stopBotJobs({ bot_id, user_id }: Params): Promise<StoppedBotJobs | null> {
|
||||
const condition = openBotJobsSql({ bot_id, user_id });
|
||||
|
||||
const open = (
|
||||
await BunSQLite.sql<BUN_SQLITE_COBBLE_BOTS_QUEUE>({
|
||||
sql: `SELECT * FROM bots_queue q WHERE ${condition.sql} ORDER BY q.id ASC`,
|
||||
values: condition.values,
|
||||
})
|
||||
).payload || [];
|
||||
|
||||
if (!open.length) return null;
|
||||
|
||||
const now = Date.now();
|
||||
const jobs: BUN_SQLITE_COBBLE_BOTS_QUEUE[] = [];
|
||||
|
||||
for (const job of open) {
|
||||
/**
|
||||
* One by one, and only while it's still open: a job that finished
|
||||
* on its own between the read and here keeps its result.
|
||||
*/
|
||||
const res = await BunSQLite.sql<BUN_SQLITE_COBBLE_BOTS_QUEUE>({
|
||||
sql: `UPDATE bots_queue SET status = 'failed', last_error = ?, locked_at = NULL, locked_by = NULL, finished_at = ?, updated_at = ?
|
||||
WHERE id = ? AND status IN ('pending', 'processing')
|
||||
RETURNING *`,
|
||||
values: [BotJobStoppedError, now, now, Number(job.id)],
|
||||
});
|
||||
|
||||
const stopped = res.payload?.[0];
|
||||
|
||||
if (!stopped?.id) continue;
|
||||
|
||||
jobs.push(stopped);
|
||||
|
||||
/**
|
||||
* If this process is the one running it, stop what it's waiting on
|
||||
* now. Anywhere else, the handler finds out on its next heartbeat.
|
||||
*/
|
||||
abortRunningJob({ job_id: Number(stopped.id) });
|
||||
|
||||
if (stopped.task_id) {
|
||||
await BunSQLite.sql({
|
||||
sql: `UPDATE bot_tasks SET status = 'done', updated_at = ? WHERE id = ?`,
|
||||
values: [now, Number(stopped.task_id)],
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Recorded as the job's last step, so the working row says why it
|
||||
* went quiet.
|
||||
*/
|
||||
try {
|
||||
await insertBotJobUpdate({
|
||||
job_id: Number(stopped.id),
|
||||
bot_id: Number(stopped.bot_id) || Number(bot_id),
|
||||
kind: "note",
|
||||
text: BotJobStoppedStep,
|
||||
});
|
||||
} catch {
|
||||
/**
|
||||
* The step is a courtesy: the job is stopped either way.
|
||||
*/
|
||||
}
|
||||
}
|
||||
|
||||
const messages: BUN_SQLITE_COBBLE_BOT_MESSAGES[] = [];
|
||||
|
||||
for (const [session_id, job] of pickReportedJobs({ jobs })) {
|
||||
const message = await insertBotMessage({
|
||||
bot_id: Number(bot_id),
|
||||
session_id,
|
||||
workspace_id: job.workspace_id || null,
|
||||
text: stoppedJobReport({ job }),
|
||||
}).catch(() => null);
|
||||
|
||||
if (message?.id) messages.push(message);
|
||||
}
|
||||
|
||||
return {
|
||||
jobs,
|
||||
messages,
|
||||
bot: await setBotFace({ bot_id, user_id, face: "idle" }),
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,79 @@
|
||||
import { describe, expect, test } from "bun:test";
|
||||
import { BotStoppedMessage } from "@/src/dict/bot-queue-dict";
|
||||
import stoppedJobReport, { pickReportedJobs } from "./stopped-job-report";
|
||||
|
||||
describe("stoppedJobReport", () => {
|
||||
test("a stopped task names the work that was cut off", () => {
|
||||
expect(stoppedJobReport({ job: { session_id: 1, kind: "task", title: "The audit" } })).toBe(
|
||||
`${BotStoppedMessage} I was on **The audit**.`,
|
||||
);
|
||||
});
|
||||
|
||||
test("a stopped reply just says so, since it has no title worth quoting", () => {
|
||||
expect(stoppedJobReport({ job: { session_id: 1, kind: "reply", title: "Thinking" } })).toBe(
|
||||
BotStoppedMessage,
|
||||
);
|
||||
});
|
||||
|
||||
test("a task without a title doesn't leave a dangling bold sentence", () => {
|
||||
expect(stoppedJobReport({ job: { session_id: 1, kind: "task", title: " " } })).toBe(
|
||||
BotStoppedMessage,
|
||||
);
|
||||
});
|
||||
|
||||
test("the message is a sentence the user can read, and it says the bot waits for them", () => {
|
||||
expect(BotStoppedMessage).toContain("Stopped");
|
||||
expect(BotStoppedMessage).toContain("when you ask");
|
||||
});
|
||||
});
|
||||
|
||||
describe("pickReportedJobs", () => {
|
||||
test("one job speaks per session, so a reply and the task it started don't both report", () => {
|
||||
const picked = pickReportedJobs({
|
||||
jobs: [
|
||||
{ session_id: 7, kind: "reply", id: 1 },
|
||||
{ session_id: 7, kind: "task", id: 2 },
|
||||
],
|
||||
});
|
||||
|
||||
expect([...picked.keys()]).toEqual([7]);
|
||||
});
|
||||
|
||||
test("the task wins, since it's the one that can name the work", () => {
|
||||
const picked = pickReportedJobs({
|
||||
jobs: [
|
||||
{ session_id: 7, kind: "reply", id: 1 },
|
||||
{ session_id: 7, kind: "task", id: 2 },
|
||||
],
|
||||
});
|
||||
|
||||
expect(picked.get(7)?.id).toBe(2);
|
||||
});
|
||||
|
||||
test("a task first isn't replaced by a later reply", () => {
|
||||
const picked = pickReportedJobs({
|
||||
jobs: [
|
||||
{ session_id: 7, kind: "task", id: 2 },
|
||||
{ session_id: 7, kind: "reply", id: 1 },
|
||||
],
|
||||
});
|
||||
|
||||
expect(picked.get(7)?.id).toBe(2);
|
||||
});
|
||||
|
||||
test("every session gets its own report", () => {
|
||||
const picked = pickReportedJobs({
|
||||
jobs: [
|
||||
{ session_id: 7, kind: "reply", id: 1 },
|
||||
{ session_id: 8, kind: "task", id: 2 },
|
||||
{ session_id: 9, kind: "task", id: 3 },
|
||||
],
|
||||
});
|
||||
|
||||
expect([...picked.keys()].sort()).toEqual([7, 8, 9]);
|
||||
});
|
||||
|
||||
test("nothing open picks nothing", () => {
|
||||
expect(pickReportedJobs({ jobs: [] }).size).toBe(0);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,45 @@
|
||||
import { BotStoppedMessage } from "@/src/dict/bot-queue-dict";
|
||||
|
||||
type Job = {
|
||||
session_id?: number | string | null;
|
||||
kind?: string | null;
|
||||
title?: string | null;
|
||||
};
|
||||
|
||||
/**
|
||||
* What the bot says in a conversation the user stopped it in. A task names
|
||||
* the work that was cut off, so a long job that quietly went nowhere reads
|
||||
* as stopped rather than as finished; a reply has no title worth quoting.
|
||||
*/
|
||||
export default function stoppedJobReport({ job }: { job: Job }): string {
|
||||
const title = (job.title || "").trim();
|
||||
|
||||
return job.kind == "task" && title
|
||||
? `${BotStoppedMessage} I was on **${title}**.`
|
||||
: BotStoppedMessage;
|
||||
}
|
||||
|
||||
/**
|
||||
* The job that speaks for each session: one per session, since a reply and
|
||||
* the task it started are the same conversation and two notes there read as
|
||||
* a stutter. A task wins, because it's the one that can name the work.
|
||||
*/
|
||||
export function pickReportedJobs<T extends Job>({ jobs }: { jobs: T[] }): Map<number, T> {
|
||||
const picked = new Map<number, T>();
|
||||
|
||||
for (const job of jobs) {
|
||||
/**
|
||||
* A job with no session is stopped all the same; it just has
|
||||
* nowhere to report it.
|
||||
*/
|
||||
if (job.session_id === null || job.session_id === undefined) continue;
|
||||
|
||||
const known = picked.get(Number(job.session_id));
|
||||
|
||||
if (!known || (known.kind != "task" && job.kind == "task")) {
|
||||
picked.set(Number(job.session_id), job);
|
||||
}
|
||||
}
|
||||
|
||||
return picked;
|
||||
}
|
||||
Reference in New Issue
Block a user