Add computer switching, crew reporting, job steps and markdown chat
- Bots have a current computer (bots.computer_id), switched from the chat header beside the model button or by clicking a computer the bot names ([computer:ID]); tasks run on it by default - search_computers runs one read-only command on all of a bot's computers; chat replies know their computers (compose-computers-note) - The coordinator can use every computer and grants or revokes crew access with assign_computer - Delegated jobs record requested_by_bot_id; the worker's result or failure is reported back in the coordinator's conversation - bots_queue_updates records each model turn of a job, pushed live and shown in the working-on panel - Live relay: every process pushes new messages, job steps and bot changes from the database to its own sockets, so work done elsewhere shows live - Bot messages render as markdown (react-markdown + remark-gfm, no raw HTML) with twui components; rail previews are plain text - Fix header button overflow in the model and computer selects - Bump @moduletrace/twui to 0.1.2 Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
Claude Opus 5.5
@@ -0,0 +1,61 @@
|
||||
import BunSQLite from "@moduletrace/bun-sqlite";
|
||||
import { AppData } from "@/src/data/app-data";
|
||||
import { logError } from "@/src/functions/backend/logger";
|
||||
import grabConnectedUserIds from "@/src/websocket/(functions)/grab-connected-user-ids";
|
||||
import relayChangedBots from "./relay-changed-bots";
|
||||
import relayNewBotMessages from "./relay-new-bot-messages";
|
||||
import relayNewJobUpdates from "./relay-new-job-updates";
|
||||
import type { LiveRelayCursor } from "./types";
|
||||
|
||||
/**
|
||||
* Keeps this process's sockets current with work done anywhere: every
|
||||
* `AppData.LiveRelayPollMs` it reads new bot messages, new job steps and
|
||||
* changed bots from the database and pushes them to their owners'
|
||||
* sockets here. The process that did the work pushes at once
|
||||
* (`pushBotNotice`); this catches the rest, e.g. a job run by another
|
||||
* server on the same database. Clients ignore what they already have.
|
||||
* Runs in every server process, ungated.
|
||||
*/
|
||||
export default async function liveRelay() {
|
||||
const latest = async (table: "bot_messages" | "bots_queue_updates") =>
|
||||
Number(
|
||||
(
|
||||
await BunSQLite.sql<{ id: number }>({
|
||||
sql: `SELECT MAX(id) AS id FROM ${table}`,
|
||||
})
|
||||
).payload?.[0]?.id,
|
||||
) || 0;
|
||||
|
||||
const cursor: LiveRelayCursor = {
|
||||
message_id: await latest("bot_messages"),
|
||||
update_id: await latest("bots_queue_updates"),
|
||||
bots_at: Date.now(),
|
||||
};
|
||||
|
||||
let running = false;
|
||||
|
||||
setInterval(async () => {
|
||||
if (running) return;
|
||||
|
||||
running = true;
|
||||
|
||||
try {
|
||||
const users = grabConnectedUserIds();
|
||||
const now = Date.now();
|
||||
|
||||
cursor.message_id = await relayNewBotMessages({ after_id: cursor.message_id, users });
|
||||
cursor.update_id = await relayNewJobUpdates({ after_id: cursor.update_id, users });
|
||||
|
||||
/**
|
||||
* Overlap one interval, so a change written in the same
|
||||
* millisecond as the last look isn't missed.
|
||||
*/
|
||||
await relayChangedBots({ since: cursor.bots_at - AppData.LiveRelayPollMs, users });
|
||||
cursor.bots_at = now;
|
||||
} catch (error) {
|
||||
await logError({ title: `Live relay failed`, error }).catch(() => {});
|
||||
} finally {
|
||||
running = false;
|
||||
}
|
||||
}, AppData.LiveRelayPollMs);
|
||||
}
|
||||
@@ -0,0 +1,47 @@
|
||||
import BunSQLite from "@moduletrace/bun-sqlite";
|
||||
import type { BUN_SQLITE_COBBLE_BOTS } from "@/db/types/db";
|
||||
import grabActiveBotTask from "@/src/functions/backend/db/bots/grab-active-bot-task";
|
||||
import type { BotUpdatedData } from "@/src/types";
|
||||
import { sendToUser } from "@/src/websocket/(functions)/send-to-user";
|
||||
|
||||
type Params = {
|
||||
/**
|
||||
* Unix ms: bots updated since then are pushed.
|
||||
*/
|
||||
since: number;
|
||||
users: Set<string>;
|
||||
};
|
||||
|
||||
/**
|
||||
* Pushes `server:bot-updated` (face, status line, active task) for every
|
||||
* bot of a connected user that changed since `since`. Pushes are
|
||||
* idempotent, so overlapping windows only repeat them.
|
||||
*/
|
||||
export default async function relayChangedBots({ since, users }: Params) {
|
||||
if (!users.size) return;
|
||||
|
||||
const ids = [...users];
|
||||
|
||||
const bots =
|
||||
(
|
||||
await BunSQLite.sql<BUN_SQLITE_COBBLE_BOTS>({
|
||||
sql: `SELECT id, user_id, face, status_line, computer_id FROM bots
|
||||
WHERE updated_at > ? AND user_id IN (${ids.map(() => "?").join(", ")})`,
|
||||
values: [since, ...ids],
|
||||
})
|
||||
).payload || [];
|
||||
|
||||
for (const bot of bots) {
|
||||
if (!bot.id || !bot.user_id) continue;
|
||||
|
||||
const data: BotUpdatedData = {
|
||||
id: bot.id,
|
||||
face: bot.face,
|
||||
status_line: bot.status_line || "",
|
||||
task: await grabActiveBotTask({ bot_id: bot.id }),
|
||||
computer_id: bot.computer_id || null,
|
||||
};
|
||||
|
||||
sendToUser({ user_id: bot.user_id, data: { event: "server:bot-updated", data } });
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
import BunSQLite from "@moduletrace/bun-sqlite";
|
||||
import type { BUN_SQLITE_COBBLE_BOT_MESSAGES } from "@/db/types/db";
|
||||
import { sendToUser } from "@/src/websocket/(functions)/send-to-user";
|
||||
|
||||
type Params = {
|
||||
after_id: number;
|
||||
users: Set<string>;
|
||||
};
|
||||
|
||||
/**
|
||||
* Pushes `server:bot-message` for every message newer than `after_id` to
|
||||
* its owner's sockets on this process. Returns the newest id read.
|
||||
*/
|
||||
export default async function relayNewBotMessages({ after_id, users }: Params) {
|
||||
const rows =
|
||||
(
|
||||
await BunSQLite.sql<BUN_SQLITE_COBBLE_BOT_MESSAGES & { user_id: number }>({
|
||||
sql: `SELECT m.*, b.user_id FROM bot_messages m JOIN bots b ON b.id = m.bot_id
|
||||
WHERE m.id > ? ORDER BY m.id ASC LIMIT 500`,
|
||||
values: [after_id],
|
||||
})
|
||||
).payload || [];
|
||||
|
||||
for (const { user_id, ...message } of rows) {
|
||||
if (users.has(String(user_id))) {
|
||||
sendToUser({ user_id, data: { event: "server:bot-message", data: message } });
|
||||
}
|
||||
}
|
||||
|
||||
return Number(rows[rows.length - 1]?.id) || after_id;
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
import BunSQLite from "@moduletrace/bun-sqlite";
|
||||
import type { BotJobUpdateData } from "@/src/types";
|
||||
import { sendToUser } from "@/src/websocket/(functions)/send-to-user";
|
||||
|
||||
type Params = {
|
||||
after_id: number;
|
||||
users: Set<string>;
|
||||
};
|
||||
|
||||
/**
|
||||
* Pushes `server:bot-job-update` for every job step newer than `after_id`
|
||||
* to its owner's sockets on this process. Returns the newest id read.
|
||||
*/
|
||||
export default async function relayNewJobUpdates({ after_id, users }: Params) {
|
||||
const rows =
|
||||
(
|
||||
await BunSQLite.sql<BotJobUpdateData & { user_id: number }>({
|
||||
sql: `SELECT u.*, q.task_id, b.user_id FROM bots_queue_updates u
|
||||
JOIN bots_queue q ON q.id = u.bots_queue_id
|
||||
JOIN bots b ON b.id = u.bot_id
|
||||
WHERE u.id > ? ORDER BY u.id ASC LIMIT 500`,
|
||||
values: [after_id],
|
||||
})
|
||||
).payload || [];
|
||||
|
||||
for (const { user_id, ...update } of rows) {
|
||||
if (users.has(String(user_id))) {
|
||||
sendToUser({ user_id, data: { event: "server:bot-job-update", data: update } });
|
||||
}
|
||||
}
|
||||
|
||||
return Number(rows[rows.length - 1]?.id) || after_id;
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
/**
|
||||
* How far the relay has read: the newest message and job step it pushed,
|
||||
* and when it last looked for changed bots.
|
||||
*/
|
||||
export type LiveRelayCursor = {
|
||||
message_id: number;
|
||||
update_id: number;
|
||||
bots_at: number;
|
||||
};
|
||||
Reference in New Issue
Block a user