Share one crew across the install, keep user memory per user
Bots, workspaces and computers now belong to the install. Users reach workers only through bot_users (assigned by a super admin or admin) and workspace_users; the coordinator and main computer are one per install, and only the super admin sees the coordinator. Conversations are per user (bot_sessions.user_id), jobs run for the session's user, and live updates go only to who should see them. A bot's core memory is now general knowledge shared by every user it works for; what it learns about a user lives in bot_user_memories. One-off scripts move existing installs over: move-bot-memory-to-owners (run first), then share-crew. Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
@@ -10,8 +10,8 @@ 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
|
||||
* changed bots from the database and pushes them to the sockets here
|
||||
* of the users they're for. 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.
|
||||
|
||||
@@ -1,8 +1,6 @@
|
||||
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";
|
||||
import pushBotUpdated from "@/src/bots-queue/push-bot-updated";
|
||||
|
||||
type Params = {
|
||||
/**
|
||||
@@ -13,35 +11,23 @@ type Params = {
|
||||
};
|
||||
|
||||
/**
|
||||
* 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.
|
||||
* Pushes `server:bot-updated` (face, status line, the user's own active
|
||||
* task) for every bot that changed since `since` to the connected users
|
||||
* who see it (`pushBotUpdated`). 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],
|
||||
sql: `SELECT id, face, status_line, computer_id FROM bots WHERE updated_at > ?`,
|
||||
values: [since],
|
||||
})
|
||||
).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 } });
|
||||
await pushBotUpdated({ bot, users });
|
||||
}
|
||||
}
|
||||
|
||||
@@ -9,13 +9,13 @@ type Params = {
|
||||
|
||||
/**
|
||||
* Pushes `server:bot-message` for every message newer than `after_id` to
|
||||
* its owner's sockets on this process. Returns the newest id read.
|
||||
* the sockets on this process of the user whose conversation it's in. 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
|
||||
sql: `SELECT m.*, s.user_id FROM bot_messages m JOIN bot_sessions s ON s.id = m.session_id
|
||||
WHERE m.id > ? ORDER BY m.id ASC LIMIT 500`,
|
||||
values: [after_id],
|
||||
})
|
||||
|
||||
@@ -9,15 +9,16 @@ type Params = {
|
||||
|
||||
/**
|
||||
* 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.
|
||||
* to the sockets on this process of the user whose conversation the job
|
||||
* is in. 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, q.session_id, b.user_id FROM bots_queue_updates u
|
||||
sql: `SELECT u.*, q.task_id, q.session_id, s.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
|
||||
JOIN bot_sessions s ON s.id = q.session_id
|
||||
WHERE u.id > ? ORDER BY u.id ASC LIMIT 500`,
|
||||
values: [after_id],
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user