From d49f524fe8ca370d6b1bfc5d857cbba92e224bbc Mon Sep 17 00:00:00 2001 From: Benjamin Toby Date: Sun, 18 Oct 2026 18:33:54 +0100 Subject: [PATCH] Add workspaces, computers, bot memory and the bots queue - Schema: workspaces (+ workspace_bots, workspace_computers), containers, computers, bot_computers, bots.memory, workspaces.system_prompt, and bots_queue as a bun-sqlite isQueue table - Main computer (and its container record) ensured with the coordinator on signup, seed and /admin/bots; the coordinator's access to it is locked - Effective computer access: direct grants plus workspace computers - Full bot prompt: core instructions, workspace prompts, then memory; bots can rewrite their memory and system prompt - Queue helpers on the native BunSQLite.queue API, plus a handler that heartbeats, serializes per computer, recovers after restarts, reports lost jobs, supports cancelling and pushes outcomes over the WebSocket - Enable SQLite foreign keys at startup so schema cascades fire - Prune finished jobs daily Co-Authored-By: Claude Opus 5.5 --- AGENTS.md | 16 +- db/schema.ts | 402 +++++++++++++++++- db/types/db.ts | 267 +++++++++++- src/bots-queue/bot-job-fatal-error.ts | 10 + src/bots-queue/index.ts | 65 +++ src/bots-queue/parse-job-json.test.ts | 16 + src/bots-queue/parse-job-json.ts | 22 + src/bots-queue/process-bot-job.ts | 126 ++++++ src/bots-queue/push-bot-notice.ts | 30 ++ src/bots-queue/push-to-bot-owner.ts | 26 ++ src/bots-queue/report-lost-bot-jobs.ts | 22 + src/bots-queue/run-bot-job.ts | 14 + src/bots-queue/types.ts | 34 ++ src/cron/index.ts | 7 + src/cron/prune-bots-queue/index.ts | 23 + src/data/app-data.ts | 37 ++ src/data/coordinator-system-prompt.ts | 4 +- src/dict/bot-dict.ts | 2 +- src/dict/bot-queue-dict.ts | 11 + src/dict/computer-dict.ts | 72 ++++ .../backend/db/bots-queue/cancel-bot-job.ts | 41 ++ .../backend/db/bots-queue/claim-bot-jobs.ts | 51 +++ .../backend/db/bots-queue/complete-bot-job.ts | 80 ++++ .../db/bots-queue/count-open-bot-jobs.ts | 17 + .../backend/db/bots-queue/enqueue-bot-job.ts | 148 +++++++ .../backend/db/bots-queue/fail-bot-job.ts | 72 ++++ .../db/bots-queue/grab-locked-bot-job.ts | 23 + .../grab-unreported-failed-bot-jobs.ts | 15 + .../db/bots-queue/notify-bot-job-failed.ts | 46 ++ .../db/bots-queue/release-worker-bot-jobs.ts | 21 + .../backend/db/bots-queue/update-bot-job.ts | 45 ++ .../backend/db/bots/ensure-coordinator-bot.ts | 2 +- .../backend/db/bots/grab-bot-prompt.ts | 39 ++ src/functions/backend/db/bots/grab-bots.ts | 2 +- .../backend/db/bots/grab-user-bot.ts | 2 +- .../backend/db/bots/insert-bot-message.ts | 53 +++ src/functions/backend/db/bots/set-bot-face.ts | 25 ++ .../backend/db/bots/update-bot-memory.ts | 36 ++ .../db/bots/update-bot-system-prompt.ts | 40 ++ .../db/computers/add-bot-to-computer.ts | 32 ++ .../ensure-main-computer-container.ts | 60 +++ .../db/computers/ensure-main-computer.ts | 85 ++++ .../db/computers/grab-bot-computers.ts | 35 ++ .../db/computers/grab-user-computer.ts | 22 + .../db/computers/remove-bot-from-computer.ts | 44 ++ .../backend/db/enable-foreign-keys.ts | 14 + .../db/workspaces/add-bot-to-workspace.ts | 32 ++ .../workspaces/add-computer-to-workspace.ts | 32 ++ .../backend/db/workspaces/create-workspace.ts | 49 +++ .../db/workspaces/grab-user-workspace.ts | 22 + .../workspaces/remove-bot-from-workspace.ts | 29 ++ .../remove-computer-from-workspace.ts | 30 ++ .../backend/db/workspaces/update-workspace.ts | 64 +++ src/pages/admin/bots/index.server.ts | 6 +- .../admin/crud/(tables)/bot_computers/del.ts | 72 ++++ .../crud/(tables)/bot_computers/index.ts | 42 ++ .../admin/crud/(tables)/bot_computers/put.ts | 11 + src/pages/api/admin/crud/[[...paths]].ts | 3 + src/pages/api/auth/signup.tsx | 23 +- src/scripts/seed-bots.ts | 2 +- src/scripts/seed-super-admin.tsx | 7 + src/server.ts | 6 +- src/types/index.ts | 9 + src/utils/assert-prompt-length.test.ts | 24 ++ src/utils/assert-prompt-length.ts | 18 + src/utils/compose-bot-prompt.test.ts | 38 ++ src/utils/compose-bot-prompt.ts | 41 ++ src/utils/create-key-lock.test.ts | 68 +++ src/utils/create-key-lock.ts | 32 ++ src/utils/is-locked-computer-access.test.ts | 32 ++ src/utils/is-locked-computer-access.ts | 16 + src/websocket/(functions)/send-to-user.ts | 28 ++ 72 files changed, 2961 insertions(+), 31 deletions(-) create mode 100644 src/bots-queue/bot-job-fatal-error.ts create mode 100644 src/bots-queue/index.ts create mode 100644 src/bots-queue/parse-job-json.test.ts create mode 100644 src/bots-queue/parse-job-json.ts create mode 100644 src/bots-queue/process-bot-job.ts create mode 100644 src/bots-queue/push-bot-notice.ts create mode 100644 src/bots-queue/push-to-bot-owner.ts create mode 100644 src/bots-queue/report-lost-bot-jobs.ts create mode 100644 src/bots-queue/run-bot-job.ts create mode 100644 src/bots-queue/types.ts create mode 100644 src/cron/prune-bots-queue/index.ts create mode 100644 src/dict/bot-queue-dict.ts create mode 100644 src/dict/computer-dict.ts create mode 100644 src/functions/backend/db/bots-queue/cancel-bot-job.ts create mode 100644 src/functions/backend/db/bots-queue/claim-bot-jobs.ts create mode 100644 src/functions/backend/db/bots-queue/complete-bot-job.ts create mode 100644 src/functions/backend/db/bots-queue/count-open-bot-jobs.ts create mode 100644 src/functions/backend/db/bots-queue/enqueue-bot-job.ts create mode 100644 src/functions/backend/db/bots-queue/fail-bot-job.ts create mode 100644 src/functions/backend/db/bots-queue/grab-locked-bot-job.ts create mode 100644 src/functions/backend/db/bots-queue/grab-unreported-failed-bot-jobs.ts create mode 100644 src/functions/backend/db/bots-queue/notify-bot-job-failed.ts create mode 100644 src/functions/backend/db/bots-queue/release-worker-bot-jobs.ts create mode 100644 src/functions/backend/db/bots-queue/update-bot-job.ts create mode 100644 src/functions/backend/db/bots/grab-bot-prompt.ts create mode 100644 src/functions/backend/db/bots/insert-bot-message.ts create mode 100644 src/functions/backend/db/bots/set-bot-face.ts create mode 100644 src/functions/backend/db/bots/update-bot-memory.ts create mode 100644 src/functions/backend/db/bots/update-bot-system-prompt.ts create mode 100644 src/functions/backend/db/computers/add-bot-to-computer.ts create mode 100644 src/functions/backend/db/computers/ensure-main-computer-container.ts create mode 100644 src/functions/backend/db/computers/ensure-main-computer.ts create mode 100644 src/functions/backend/db/computers/grab-bot-computers.ts create mode 100644 src/functions/backend/db/computers/grab-user-computer.ts create mode 100644 src/functions/backend/db/computers/remove-bot-from-computer.ts create mode 100644 src/functions/backend/db/enable-foreign-keys.ts create mode 100644 src/functions/backend/db/workspaces/add-bot-to-workspace.ts create mode 100644 src/functions/backend/db/workspaces/add-computer-to-workspace.ts create mode 100644 src/functions/backend/db/workspaces/create-workspace.ts create mode 100644 src/functions/backend/db/workspaces/grab-user-workspace.ts create mode 100644 src/functions/backend/db/workspaces/remove-bot-from-workspace.ts create mode 100644 src/functions/backend/db/workspaces/remove-computer-from-workspace.ts create mode 100644 src/functions/backend/db/workspaces/update-workspace.ts create mode 100644 src/pages/api/admin/crud/(tables)/bot_computers/del.ts create mode 100644 src/pages/api/admin/crud/(tables)/bot_computers/index.ts create mode 100644 src/pages/api/admin/crud/(tables)/bot_computers/put.ts create mode 100644 src/utils/assert-prompt-length.test.ts create mode 100644 src/utils/assert-prompt-length.ts create mode 100644 src/utils/compose-bot-prompt.test.ts create mode 100644 src/utils/compose-bot-prompt.ts create mode 100644 src/utils/create-key-lock.test.ts create mode 100644 src/utils/create-key-lock.ts create mode 100644 src/utils/is-locked-computer-access.test.ts create mode 100644 src/utils/is-locked-computer-access.ts create mode 100644 src/websocket/(functions)/send-to-user.ts diff --git a/AGENTS.md b/AGENTS.md index 54cf96e..e825436 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -14,12 +14,12 @@ This is **Cobble** (`cobble.moduletrace.org`): an open-source crew of small AI w ## Entrypoint -`src/server.ts` - runs `bunextInit()`, starts `Bun.serve`, then starts cron: +`src/server.ts` - turns on SQLite foreign keys (`enableForeignKeys`, so the schema's cascades fire; off by default per connection), runs `bunextInit()`, starts `Bun.serve`, then starts cron: - `/media/*` short-circuits to the media server handler - `/ws` upgrades only after `userAuth` succeeds (session frozen into the socket at upgrade time); the `websocket` handler object lives in `src/websocket/` - everything else goes to `bunext.bunextRequestHandler` -- `cron()` starts after the server is up; in production it is gated to the `MAIN_SERVER` container +- `cron()` and the bots queue handler (`botsQueue()`, `src/bots-queue/`) start after the server is up; in production both are gated to the `MAIN_SERVER` container ## Commands @@ -83,13 +83,19 @@ Inspect live data with `bun llm-code-assistant/query-db.ts ""` - a read-onl ### Bots (the chat shell) -- **Tables:** `bots` (per-user workspace; `kind` worker|coordinator, `hue`, `role`, `face`, `status_line`, `system_prompt` (the bot's core instructions, markdown), `unread`), `bot_messages`, `bot_tasks`, `bot_task_steps`. Enums in `src/dict/bot-dict.ts`. -- **Coordinator:** every workspace has exactly one `kind: coordinator` bot that coordinates the crew and delegates to the workers. It is created with the account (signup, `bun run seed`) and, as a fallback, by `/admin/bots`, all via `ensureCoordinatorBot` (`src/functions/backend/db/bots/ensure-coordinator-bot.ts`, idempotent). Name in `CoordinatorBotName` (`src/dict/bot-dict.ts`). Its default prompt is `CoordinatorSystemPrompt` (`src/data/coordinator-system-prompt.ts`): attached at creation, and backfilled onto a coordinator whose prompt is empty, never over an edited one. +- **Tables:** `bots` (per user; `kind` worker|coordinator, `hue`, `role`, `face`, `status_line`, `system_prompt` (the bot's core instructions, markdown), `unread`), `bot_messages`, `bot_tasks`, `bot_task_steps`. Enums in `src/dict/bot-dict.ts`. +- **Coordinator:** every user has exactly one `kind: coordinator` bot that coordinates the crew and delegates to the workers. It is created with the account (signup, `bun run seed`) and, as a fallback, by `/admin/bots`, all via `ensureCoordinatorBot` (`src/functions/backend/db/bots/ensure-coordinator-bot.ts`, idempotent). Name in `CoordinatorBotName` (`src/dict/bot-dict.ts`). Its default prompt is `CoordinatorSystemPrompt` (`src/data/coordinator-system-prompt.ts`): attached at creation, and backfilled onto a coordinator whose prompt is empty, never over an edited one. - **Routes:** `/admin/bots` ensures the coordinator and redirects to its chat (`/admin/bots/1` on a fresh install); `/admin/bots/[bot_id]` renders conversation + working-on panel. Both render in the full-bleed `src/layouts/bots/` layout (rail of conversations + user popover), which `__root.tsx` uses for every `/admin/*` page. - **Server:** `src/functions/backend/db/bots/` (`grab-bots`, `grab-bot-messages`, `grab-active-bot-task`, `grab-user-bot`, `mark-bot-read`, `insert-user-message`). `api/admin/bots/send-message.ts` posts a user message to a bot owned by the session user (zod: `src/schemas/bot-schema.ts`). - **Avatars:** `BotStone` (`src/components/general/bot-stone.tsx`) renders the stone inline from `src/data/stone-geometry.ts`, hue as a fill token, face (`idle|working|done|asleep|needs-you`) drawn as eyes. The coordinator is hue-less (ink body, paper eyes; inverts in dark). Never put a stone in a circle/badge or add outline/shadow. Wordmark is vector (`Wordmark`, `src/data/wordmark-paths.ts`), never live text. - **Copy helpers:** `src/utils/crew-status.ts` (`3 working ยท 1 needs you`, "Three bots are working. One needs you."), `src/utils/relative-time.ts` (relative timestamps only). -- **Not built yet:** add-bot flow, attachments, routines, help, live updates over WebSocket, the model layer. +- **Prompt and memory (self-learning):** a bot has `system_prompt` (core instructions) and a separate `memory` (markdown, what it must always remember). The bot rewrites both as it adapts to the user (`update-bot-system-prompt`, `update-bot-memory` in `src/functions/backend/db/bots/`; whole-text replace, capped at `AppData.PromptMaxLength`). `grab-bot-prompt` builds the full prompt via `composeBotPrompt` (`src/utils/compose-bot-prompt.ts`): core instructions, then each non-archived workspace's `system_prompt` under `# Workspace: `, then `# Memory`. +- **Queue:** long-running work (typically on a computer) goes through `bots_queue`, a bun-sqlite `isQueue` table worked with the native `BunSQLite.queue` API (`status` pending|processing|completed|failed, `attempts`, `available_at`, `locked_at`/`locked_by`, `last_error`; options in `AppData.BotQueue*`). Not every message needs a job: the bot replies right away that it's on it and `enqueueBotJob` queues the brief (`instructions`, `params`, optional `message_id`/`computer_id`, the computer checked against the bot's access; a `bot_tasks` row is created for the working-on panel unless `task_id` is given). DB helpers in `src/functions/backend/db/bots-queue/`: `claimBotJobs`, `updateBotJob` (renews the lock, saves `progress`/`checkpoint`; false once lost or cancelled), `completeBotJob` (result message, task done, face done/needs-you), `failBotJob` (backoff retry, or `fatal`; out of attempts posts a needs-you message via `notifyBotJobFailed`), `cancelBotJob` (failed with `BotJobCancelledError`, `src/dict/bot-queue-dict.ts`), `releaseWorkerBotJobs`. +- **Queue handler:** `src/bots-queue/index.ts`, started from `server.ts` after cron and gated the same way. It releases its own cut-off jobs on startup (stable worker `AppData.BotQueueWorker`), then polls every `BotQueuePollMs`: reports jobs bun-sqlite failed on a lock timeout (`report-lost-bot-jobs`), claims up to `BotQueueConcurrency`, and runs each with `processBotJob` (heartbeat at a quarter of the lock timeout, abort on cancel/loss, one job per computer via `createKeyLock`, face `working` once it starts, outcome pushed live). The actual work is `run-bot-job.ts` (`BotJobRunner`, context in `types.ts`): the model layer plugs in there; until then it throws `BotJobFatalError`. Throw `BotJobFatalError` for failures retrying can't fix. The `prune-bots-queue` cron deletes finished jobs after `BotQueueRetentionDays`. +- **Live updates:** server push only. `sendToUser` (`src/websocket/(functions)/send-to-user.ts`) sends to one user's sockets; the queue pushes `server:bot-message` and `server:bot-updated`. The client doesn't listen yet. +- **Workspaces:** `workspaces` (per user, `name`, `description`, `system_prompt` shared by every bot in it, `archived`), with bots assigned through `workspace_bots` and computers through `workspace_computers`. Helpers in `src/functions/backend/db/workspaces/` (`create-workspace`, `update-workspace`, `add/remove-bot-*`, `add/remove-computer-*`), all ownership-checked against the session user. +- **Computers:** a computer is an AIO sandbox in its own docker container on the `AppData.DockerNetworkName` network. `containers` records every docker container the app knows about except the main app container (`name` doubles as the hostname, plus `docker_id`, `image`, `network`, `port`, `status`, `metadata`). `computers` (per user, `kind` main|sandbox, `container_id`) and `bot_computers` (direct access). Each user has one main computer, recorded with the account (signup, `bun run seed`) and as a fallback by `/admin/bots` via `ensureMainComputer` (`src/functions/backend/db/computers/ensure-main-computer.ts`, idempotent), which also records its container (`AppData.MainComputerContainerName`, created by the installer) and grants the coordinator access. That one grant can't be removed (`isLockedComputerAccess`, enforced in `remove-bot-from-computer` and the `bot_computers` CRUD handler); every other grant can. A bot's effective computers (`grab-bot-computers`) are its direct grants plus every computer assigned to a non-archived workspace it's in. Enums in `src/dict/computer-dict.ts`. +- **Not built yet:** add-bot flow, attachments, routines, help, client handling of live WebSocket updates, the model layer (plugs into `src/bots-queue/run-bot-job.ts`). ## Admin Route Hierarchy diff --git a/db/schema.ts b/db/schema.ts index 4fd6bf3..f548557 100644 --- a/db/schema.ts +++ b/db/schema.ts @@ -8,6 +8,12 @@ import { BotTaskStatuses, BotTaskStepStates, } from "@/src/dict/bot-dict"; +import { + ComputerKinds, + ContainerPurposes, + ContainerStatuses, +} from "@/src/dict/computer-dict"; +import { AppData } from "@/src/data/app-data"; import type { BUN_SQLITE_DatabaseSchemaType } from "@moduletrace/bun-sqlite/types"; const schema: BUN_SQLITE_DatabaseSchemaType = { @@ -197,7 +203,7 @@ const schema: BUN_SQLITE_DatabaseSchemaType = { fields: [ { fieldName: "user_id", - fieldDescription: `Workspace owner`, + fieldDescription: `Bot owner`, dataType: "INTEGER", foreignKey: { cascadeDelete: true, @@ -213,7 +219,7 @@ const schema: BUN_SQLITE_DatabaseSchemaType = { }, { fieldName: "kind", - fieldDescription: `One coordinator per workspace; everything else is a worker`, + fieldDescription: `One coordinator per owner; everything else is a worker`, dataType: "TEXT", options: BotKinds.map((bk) => bk.value), defaultValue: "worker", @@ -243,7 +249,12 @@ const schema: BUN_SQLITE_DatabaseSchemaType = { }, { fieldName: "system_prompt", - fieldDescription: `The bot's core instructions, in markdown`, + fieldDescription: `The bot's core instructions, in markdown. The bot can rewrite it`, + dataType: "TEXT", + }, + { + fieldName: "memory", + fieldDescription: `What the bot must always remember, in markdown. The bot keeps it up to date`, dataType: "TEXT", }, { @@ -377,6 +388,391 @@ const schema: BUN_SQLITE_DatabaseSchemaType = { }, ], }, + { + tableName: "bots_queue", + isQueue: { + maxAttempts: AppData.BotQueueMaxAttempts, + lockTimeout: AppData.BotQueueLockTimeoutMs, + backoff: AppData.BotQueueBackoffMs, + }, + fields: [ + { + fieldName: "bot_id", + fieldDescription: `The bot doing the work`, + dataType: "INTEGER", + foreignKey: { + cascadeDelete: true, + destinationTableName: "bots", + destinationTableColumnName: "id", + destinationTableColumnType: "INTEGER", + foreignKeyName: "bots_queue_bot_fk", + }, + }, + { + fieldName: "message_id", + fieldDescription: `The user message that asked for the work, if any`, + dataType: "INTEGER", + foreignKey: { + cascadeDelete: true, + destinationTableName: "bot_messages", + destinationTableColumnName: "id", + destinationTableColumnType: "INTEGER", + foreignKeyName: "bots_queue_message_fk", + }, + }, + { + fieldName: "task_id", + fieldDescription: `The task shown in the working-on panel, if any`, + dataType: "INTEGER", + foreignKey: { + cascadeDelete: true, + destinationTableName: "bot_tasks", + destinationTableColumnName: "id", + destinationTableColumnType: "INTEGER", + foreignKeyName: "bots_queue_task_fk", + }, + }, + { + fieldName: "computer_id", + fieldDescription: `The computer to run on, if any. Checked against the bot's access when queued`, + dataType: "INTEGER", + }, + { + fieldName: "title", + fieldDescription: `Short label for the job`, + dataType: "TEXT", + }, + { + fieldName: "instructions", + fieldDescription: `The brief the bot wrote for itself, in markdown`, + dataType: "TEXT", + }, + { + fieldName: "params", + fieldDescription: `Structured parameters the handler needs`, + dataType: "TEXT", + json: true, + }, + { + fieldName: "progress", + fieldDescription: `Latest progress note from the bot, in markdown`, + dataType: "TEXT", + }, + { + fieldName: "checkpoint", + fieldDescription: `State to resume from if the job is picked up again`, + dataType: "TEXT", + json: true, + }, + { + fieldName: "result", + fieldDescription: `What the bot reported when it finished, in markdown`, + dataType: "TEXT", + }, + { + fieldName: "started_at", + fieldDescription: `Unix timestamp of the first claim`, + dataType: "INTEGER", + }, + { + fieldName: "finished_at", + fieldDescription: `Unix timestamp of completion or final failure`, + dataType: "INTEGER", + }, + ], + indexes: [ + { + indexName: "idx_bots_queue_bot_id_status", + indexTableFields: ["bot_id", "status"], + }, + ], + }, + { + tableName: "workspaces", + fields: [ + { + fieldName: "user_id", + fieldDescription: `Workspace owner`, + dataType: "INTEGER", + foreignKey: { + cascadeDelete: true, + destinationTableName: "users", + destinationTableColumnName: "id", + destinationTableColumnType: "INTEGER", + foreignKeyName: "workspace_user_fk", + }, + }, + { + fieldName: "name", + dataType: "TEXT", + }, + { + fieldName: "description", + dataType: "TEXT", + }, + { + fieldName: "system_prompt", + fieldDescription: `Shared context in markdown, appended to the prompt of every bot in the workspace`, + dataType: "TEXT", + }, + { + fieldName: "archived", + dataType: "INTEGER", + options: [0, 1], + defaultValue: 0, + }, + ], + indexes: [ + { + indexName: "idx_workspaces_user_id", + indexTableFields: ["user_id"], + }, + ], + }, + { + tableName: "workspace_bots", + fields: [ + { + fieldName: "workspace_id", + dataType: "INTEGER", + foreignKey: { + cascadeDelete: true, + destinationTableName: "workspaces", + destinationTableColumnName: "id", + destinationTableColumnType: "INTEGER", + foreignKeyName: "workspace_bot_workspace_fk", + }, + }, + { + fieldName: "bot_id", + dataType: "INTEGER", + foreignKey: { + cascadeDelete: true, + destinationTableName: "bots", + destinationTableColumnName: "id", + destinationTableColumnType: "INTEGER", + foreignKeyName: "workspace_bot_bot_fk", + }, + }, + ], + uniqueConstraints: [ + { + alias: "one_bot_per_workspace", + constraintName: "A bot joins a workspace once", + constraintTableFields: [ + { value: "workspace_id" }, + { value: "bot_id" }, + ], + }, + ], + indexes: [ + { + indexName: "idx_workspace_bots_bot_id", + indexTableFields: ["bot_id"], + }, + ], + }, + { + tableName: "containers", + fields: [ + { + fieldName: "name", + fieldDescription: `Docker container name, also its hostname on the network`, + dataType: "TEXT", + unique: true, + }, + { + fieldName: "docker_id", + fieldDescription: `Docker container id, once known`, + dataType: "TEXT", + }, + { + fieldName: "purpose", + dataType: "TEXT", + options: ContainerPurposes.map((cp) => cp.value), + defaultValue: "computer", + }, + { + fieldName: "image", + fieldDescription: `Image the container runs, e.g. ghcr.io/agent-infra/sandbox:latest`, + dataType: "TEXT", + }, + { + fieldName: "network", + fieldDescription: `Docker network shared with the main container`, + dataType: "TEXT", + }, + { + fieldName: "hostname", + fieldDescription: `Address the main container reaches it on`, + dataType: "TEXT", + }, + { + fieldName: "port", + fieldDescription: `Port the service listens on inside the container`, + dataType: "INTEGER", + }, + { + fieldName: "status", + fieldDescription: `Last known docker state`, + dataType: "TEXT", + options: ContainerStatuses.map((cs) => cs.value), + defaultValue: "unknown", + }, + { + fieldName: "metadata", + fieldDescription: `Extra docker details (ports, volumes, env names, labels)`, + dataType: "TEXT", + json: true, + }, + { + fieldName: "last_seen_at", + fieldDescription: `Unix timestamp of the last docker inspect`, + dataType: "INTEGER", + }, + ], + }, + { + tableName: "computers", + fields: [ + { + fieldName: "user_id", + fieldDescription: `Computer owner`, + dataType: "INTEGER", + foreignKey: { + cascadeDelete: true, + destinationTableName: "users", + destinationTableColumnName: "id", + destinationTableColumnType: "INTEGER", + foreignKeyName: "computer_user_fk", + }, + }, + { + fieldName: "container_id", + fieldDescription: `The container this computer runs in`, + dataType: "INTEGER", + foreignKey: { + cascadeDelete: true, + destinationTableName: "containers", + destinationTableColumnName: "id", + destinationTableColumnType: "INTEGER", + foreignKeyName: "computer_container_fk", + }, + }, + { + fieldName: "name", + dataType: "TEXT", + }, + { + fieldName: "kind", + fieldDescription: `One main computer per owner; everything else is a sandbox`, + dataType: "TEXT", + options: ComputerKinds.map((ck) => ck.value), + defaultValue: "sandbox", + }, + { + fieldName: "description", + dataType: "TEXT", + }, + ], + indexes: [ + { + indexName: "idx_computers_user_id_kind", + indexTableFields: ["user_id", "kind"], + }, + { + indexName: "idx_computers_container_id", + indexTableFields: ["container_id"], + }, + ], + }, + { + tableName: "bot_computers", + fields: [ + { + fieldName: "bot_id", + dataType: "INTEGER", + foreignKey: { + cascadeDelete: true, + destinationTableName: "bots", + destinationTableColumnName: "id", + destinationTableColumnType: "INTEGER", + foreignKeyName: "bot_computer_bot_fk", + }, + }, + { + fieldName: "computer_id", + dataType: "INTEGER", + foreignKey: { + cascadeDelete: true, + destinationTableName: "computers", + destinationTableColumnName: "id", + destinationTableColumnType: "INTEGER", + foreignKeyName: "bot_computer_computer_fk", + }, + }, + ], + uniqueConstraints: [ + { + alias: "one_access_per_bot_computer", + constraintName: "A bot gets access to a computer once", + constraintTableFields: [ + { value: "bot_id" }, + { value: "computer_id" }, + ], + }, + ], + indexes: [ + { + indexName: "idx_bot_computers_computer_id", + indexTableFields: ["computer_id"], + }, + ], + }, + { + tableName: "workspace_computers", + fields: [ + { + fieldName: "workspace_id", + fieldDescription: `Every bot in this workspace gets access to the computer`, + dataType: "INTEGER", + foreignKey: { + cascadeDelete: true, + destinationTableName: "workspaces", + destinationTableColumnName: "id", + destinationTableColumnType: "INTEGER", + foreignKeyName: "workspace_computer_workspace_fk", + }, + }, + { + fieldName: "computer_id", + dataType: "INTEGER", + foreignKey: { + cascadeDelete: true, + destinationTableName: "computers", + destinationTableColumnName: "id", + destinationTableColumnType: "INTEGER", + foreignKeyName: "workspace_computer_computer_fk", + }, + }, + ], + uniqueConstraints: [ + { + alias: "one_computer_per_workspace", + constraintName: "A computer joins a workspace once", + constraintTableFields: [ + { value: "workspace_id" }, + { value: "computer_id" }, + ], + }, + ], + indexes: [ + { + indexName: "idx_workspace_computers_computer_id", + indexTableFields: ["computer_id"], + }, + ], + }, { tableName: "logs", fields: [ diff --git a/db/types/db.ts b/db/types/db.ts index 5932f5a..16b8cb6 100644 --- a/db/types/db.ts +++ b/db/types/db.ts @@ -9,6 +9,13 @@ export const BunSQLiteTables = [ "bot_messages", "bot_tasks", "bot_task_steps", + "bots_queue", + "workspaces", + "workspace_bots", + "containers", + "computers", + "bot_computers", + "workspace_computers", "logs", ] as const @@ -143,12 +150,12 @@ export type BUN_SQLITE_COBBLE_BOTS = { */ updated_at?: number | ""; /** - * Workspace owner + * Bot owner */ user_id?: number | ""; name?: string; /** - * One coordinator per workspace; everything else is a worker + * One coordinator per owner; everything else is a worker */ kind?: "worker" | "coordinator" | ""; /** @@ -168,9 +175,13 @@ export type BUN_SQLITE_COBBLE_BOTS = { */ status_line?: string; /** - * The bot's core instructions, in markdown + * The bot's core instructions, in markdown. The bot can rewrite it */ system_prompt?: string; + /** + * What the bot must always remember, in markdown. The bot keeps it up to date + */ + memory?: string; unread?: 0 | 1 | ""; /** * Unix timestamp of the latest message, for rail ordering @@ -241,6 +252,254 @@ export type BUN_SQLITE_COBBLE_BOT_TASK_STEPS = { position?: number | ""; } +export type BUN_SQLITE_COBBLE_BOTS_QUEUE = { + /** + * The unique identifier of the record. + */ + id?: number | ""; + /** + * The time when the record was created. (Unix Timestamp) + */ + created_at?: number | ""; + /** + * The time when the record was updated. (Unix Timestamp) + */ + updated_at?: number | ""; + /** + * The bot doing the work + */ + bot_id?: number | ""; + /** + * The user message that asked for the work, if any + */ + message_id?: number | ""; + /** + * The task shown in the working-on panel, if any + */ + task_id?: number | ""; + /** + * The computer to run on, if any. Checked against the bot's access when queued + */ + computer_id?: number | ""; + /** + * Short label for the job + */ + title?: string; + /** + * The brief the bot wrote for itself, in markdown + */ + instructions?: string; + /** + * Structured parameters the handler needs + */ + params?: string; + /** + * Latest progress note from the bot, in markdown + */ + progress?: string; + /** + * State to resume from if the job is picked up again + */ + checkpoint?: string; + /** + * What the bot reported when it finished, in markdown + */ + result?: string; + /** + * Unix timestamp of the first claim + */ + started_at?: number | ""; + /** + * Unix timestamp of completion or final failure + */ + finished_at?: number | ""; + /** + * The state of the job. (pending | processing | completed | failed) + */ + status?: string; + /** + * The number of times the job has been claimed. + */ + attempts?: number | ""; + /** + * The time from which the job can be claimed. (Unix Timestamp) + */ + available_at?: number | ""; + /** + * The time when the job was claimed. (Unix Timestamp) + */ + locked_at?: number | ""; + /** + * The worker that claimed the job. + */ + locked_by?: string; + /** + * The error from the last failed attempt. + */ + last_error?: string; +} + +export type BUN_SQLITE_COBBLE_WORKSPACES = { + /** + * The unique identifier of the record. + */ + id?: number | ""; + /** + * The time when the record was created. (Unix Timestamp) + */ + created_at?: number | ""; + /** + * The time when the record was updated. (Unix Timestamp) + */ + updated_at?: number | ""; + /** + * Workspace owner + */ + user_id?: number | ""; + name?: string; + description?: string; + /** + * Shared context in markdown, appended to the prompt of every bot in the workspace + */ + system_prompt?: string; + archived?: 0 | 1 | ""; +} + +export type BUN_SQLITE_COBBLE_WORKSPACE_BOTS = { + /** + * The unique identifier of the record. + */ + id?: number | ""; + /** + * The time when the record was created. (Unix Timestamp) + */ + created_at?: number | ""; + /** + * The time when the record was updated. (Unix Timestamp) + */ + updated_at?: number | ""; + workspace_id?: number | ""; + bot_id?: number | ""; +} + +export type BUN_SQLITE_COBBLE_CONTAINERS = { + /** + * The unique identifier of the record. + */ + id?: number | ""; + /** + * The time when the record was created. (Unix Timestamp) + */ + created_at?: number | ""; + /** + * The time when the record was updated. (Unix Timestamp) + */ + updated_at?: number | ""; + /** + * Docker container name, also its hostname on the network + */ + name?: string; + /** + * Docker container id, once known + */ + docker_id?: string; + purpose?: "computer" | ""; + /** + * Image the container runs, e.g. ghcr.io/agent-infra/sandbox:latest + */ + image?: string; + /** + * Docker network shared with the main container + */ + network?: string; + /** + * Address the main container reaches it on + */ + hostname?: string; + /** + * Port the service listens on inside the container + */ + port?: number | ""; + /** + * Last known docker state + */ + status?: "created" | "running" | "paused" | "restarting" | "exited" | "removing" | "dead" | "unknown" | ""; + /** + * Extra docker details (ports, volumes, env names, labels) + */ + metadata?: string; + /** + * Unix timestamp of the last docker inspect + */ + last_seen_at?: number | ""; +} + +export type BUN_SQLITE_COBBLE_COMPUTERS = { + /** + * The unique identifier of the record. + */ + id?: number | ""; + /** + * The time when the record was created. (Unix Timestamp) + */ + created_at?: number | ""; + /** + * The time when the record was updated. (Unix Timestamp) + */ + updated_at?: number | ""; + /** + * Computer owner + */ + user_id?: number | ""; + /** + * The container this computer runs in + */ + container_id?: number | ""; + name?: string; + /** + * One main computer per owner; everything else is a sandbox + */ + kind?: "main" | "sandbox" | ""; + description?: string; +} + +export type BUN_SQLITE_COBBLE_BOT_COMPUTERS = { + /** + * The unique identifier of the record. + */ + id?: number | ""; + /** + * The time when the record was created. (Unix Timestamp) + */ + created_at?: number | ""; + /** + * The time when the record was updated. (Unix Timestamp) + */ + updated_at?: number | ""; + bot_id?: number | ""; + computer_id?: number | ""; +} + +export type BUN_SQLITE_COBBLE_WORKSPACE_COMPUTERS = { + /** + * The unique identifier of the record. + */ + id?: number | ""; + /** + * The time when the record was created. (Unix Timestamp) + */ + created_at?: number | ""; + /** + * The time when the record was updated. (Unix Timestamp) + */ + updated_at?: number | ""; + /** + * Every bot in this workspace gets access to the computer + */ + workspace_id?: number | ""; + computer_id?: number | ""; +} + export type BUN_SQLITE_COBBLE_LOGS = { /** * The unique identifier of the record. @@ -288,4 +547,4 @@ export type BUN_SQLITE_COBBLE_LOGS = { ip_address?: string; } -export type BUN_SQLITE_COBBLE_ALL_TYPEDEFS = BUN_SQLITE_COBBLE_USERS & BUN_SQLITE_COBBLE_USER_TYPES & BUN_SQLITE_COBBLE_SSO_LOGIN_CODES & BUN_SQLITE_COBBLE_USER_PASSWORDS & BUN_SQLITE_COBBLE_MEDIA & BUN_SQLITE_COBBLE_MEDIA_PARADIGMS & BUN_SQLITE_COBBLE_BOTS & BUN_SQLITE_COBBLE_BOT_MESSAGES & BUN_SQLITE_COBBLE_BOT_TASKS & BUN_SQLITE_COBBLE_BOT_TASK_STEPS & BUN_SQLITE_COBBLE_LOGS \ No newline at end of file +export type BUN_SQLITE_COBBLE_ALL_TYPEDEFS = BUN_SQLITE_COBBLE_USERS & BUN_SQLITE_COBBLE_USER_TYPES & BUN_SQLITE_COBBLE_SSO_LOGIN_CODES & BUN_SQLITE_COBBLE_USER_PASSWORDS & BUN_SQLITE_COBBLE_MEDIA & BUN_SQLITE_COBBLE_MEDIA_PARADIGMS & BUN_SQLITE_COBBLE_BOTS & BUN_SQLITE_COBBLE_BOT_MESSAGES & BUN_SQLITE_COBBLE_BOT_TASKS & BUN_SQLITE_COBBLE_BOT_TASK_STEPS & BUN_SQLITE_COBBLE_BOTS_QUEUE & BUN_SQLITE_COBBLE_WORKSPACES & BUN_SQLITE_COBBLE_WORKSPACE_BOTS & BUN_SQLITE_COBBLE_CONTAINERS & BUN_SQLITE_COBBLE_COMPUTERS & BUN_SQLITE_COBBLE_BOT_COMPUTERS & BUN_SQLITE_COBBLE_WORKSPACE_COMPUTERS & BUN_SQLITE_COBBLE_LOGS \ No newline at end of file diff --git a/src/bots-queue/bot-job-fatal-error.ts b/src/bots-queue/bot-job-fatal-error.ts new file mode 100644 index 0000000..e395d28 --- /dev/null +++ b/src/bots-queue/bot-job-fatal-error.ts @@ -0,0 +1,10 @@ +/** + * Throw from a job runner when retrying can't help (bad parameters, no + * access). The job fails at once instead of using its remaining attempts. + */ +export default class BotJobFatalError extends Error { + constructor(message: string) { + super(message); + this.name = "BotJobFatalError"; + } +} diff --git a/src/bots-queue/index.ts b/src/bots-queue/index.ts new file mode 100644 index 0000000..974af6f --- /dev/null +++ b/src/bots-queue/index.ts @@ -0,0 +1,65 @@ +import bunext from "@moduletrace/bunext"; +import { AppData } from "@/src/data/app-data"; +import claimBotJobs from "@/src/functions/backend/db/bots-queue/claim-bot-jobs"; +import releaseWorkerBotJobs from "@/src/functions/backend/db/bots-queue/release-worker-bot-jobs"; +import { logError } from "@/src/functions/backend/logger"; +import processBotJob from "./process-bot-job"; +import reportLostBotJobs from "./report-lost-bot-jobs"; + +const worker = AppData.BotQueueWorker; +let running = 0; + +/** + * Starts the `bots_queue` handler: releases jobs a restart cut off, then + * polls every `AppData.BotQueuePollMs`, reporting lost jobs and claiming + * up to `AppData.BotQueueConcurrency` at once. Gated in production to the + * `MAIN_SERVER` container, like cron. + */ +export default async function botsQueue() { + if (process.env.NODE_ENV == "production" && !process.env.MAIN_SERVER) { + return; + } + + bunext.bunextLog.watch(`Running Bots Queue ...`); + + const released = await releaseWorkerBotJobs({ worker }); + + if (released) { + bunext.bunextLog.info(`Bots queue: released ${released} cut-off job(s)`); + } + + poll(); +} + +async function poll() { + try { + await reportLostBotJobs(); + + const free = AppData.BotQueueConcurrency - running; + + if (free > 0) { + const jobs = await claimBotJobs({ worker, limit: free }); + + for (const job of jobs) { + running++; + + processBotJob({ job, worker }) + .catch((error) => + logError({ + title: `Bot job #${job.id} crashed`, + error, + }).catch(() => {}), + ) + .finally(() => { + running--; + }); + } + } + } catch (error) { + await logError({ title: "Bots queue poll failed", error }).catch( + () => {}, + ); + } finally { + setTimeout(poll, AppData.BotQueuePollMs); + } +} diff --git a/src/bots-queue/parse-job-json.test.ts b/src/bots-queue/parse-job-json.test.ts new file mode 100644 index 0000000..a66a71e --- /dev/null +++ b/src/bots-queue/parse-job-json.test.ts @@ -0,0 +1,16 @@ +import { describe, expect, test } from "bun:test"; +import parseJobJson from "./parse-job-json"; + +describe("parseJobJson", () => { + test("reads a JSON object", () => { + expect(parseJobJson({ text: `{"step":2}` })).toEqual({ step: 2 }); + }); + + test("is null for empty, invalid or non-object JSON", () => { + expect(parseJobJson({})).toBeNull(); + expect(parseJobJson({ text: "" })).toBeNull(); + expect(parseJobJson({ text: "{nope" })).toBeNull(); + expect(parseJobJson({ text: "[1]" })).toBeNull(); + expect(parseJobJson({ text: "3" })).toBeNull(); + }); +}); diff --git a/src/bots-queue/parse-job-json.ts b/src/bots-queue/parse-job-json.ts new file mode 100644 index 0000000..657f44f --- /dev/null +++ b/src/bots-queue/parse-job-json.ts @@ -0,0 +1,22 @@ +type Params = { + text?: string | null; +}; + +/** + * Reads a job's `params` or `checkpoint` column. Null when empty or not a + * JSON object. + */ +export default function parseJobJson({ + text, +}: Params): Record | null { + if (!text) return null; + + try { + const value = JSON.parse(text); + return value && typeof value == "object" && !Array.isArray(value) + ? value + : null; + } catch { + return null; + } +} diff --git a/src/bots-queue/process-bot-job.ts b/src/bots-queue/process-bot-job.ts new file mode 100644 index 0000000..88c0969 --- /dev/null +++ b/src/bots-queue/process-bot-job.ts @@ -0,0 +1,126 @@ +import type { BUN_SQLITE_COBBLE_BOTS_QUEUE } from "@/db/types/db"; +import { AppData } from "@/src/data/app-data"; +import setBotFace from "@/src/functions/backend/db/bots/set-bot-face"; +import completeBotJob from "@/src/functions/backend/db/bots-queue/complete-bot-job"; +import failBotJob from "@/src/functions/backend/db/bots-queue/fail-bot-job"; +import updateBotJob from "@/src/functions/backend/db/bots-queue/update-bot-job"; +import { logError } from "@/src/functions/backend/logger"; +import createKeyLock from "@/src/utils/create-key-lock"; +import BotJobFatalError from "./bot-job-fatal-error"; +import pushBotNotice from "./push-bot-notice"; +import runBotJob from "./run-bot-job"; +import parseJobJson from "./parse-job-json"; +import type { BotJobContext, BotJobRunner } from "./types"; + +/** + * One job per computer at a time, across all running jobs. + */ +const computer_lock = createKeyLock(); + +/** + * Renew the lock well inside the timeout, so one slow write can't lose it. + */ +const HEARTBEAT_MS = Math.floor(AppData.BotQueueLockTimeoutMs / 4); + +type Params = { + job: BUN_SQLITE_COBBLE_BOTS_QUEUE; + worker: string; + /** + * Defaults to `runBotJob`. + */ + runner?: BotJobRunner; +}; + +/** + * Runs one claimed job to the end: keeps its lock alive, waits for its + * computer, runs it, then completes or fails it and pushes the outcome to + * the user. If the job is cancelled or lost mid-way, its result is + * dropped. + */ +export default async function processBotJob({ + job, + worker, + runner = runBotJob, +}: Params) { + if (!job.id || !job.bot_id) return; + + const job_id = job.id; + const bot_id = job.bot_id; + const controller = new AbortController(); + + const update: BotJobContext["update"] = async (params) => { + if (controller.signal.aborted) return false; + + const held = await updateBotJob({ job_id, worker, ...params }); + + if (!held) controller.abort(); + + return held; + }; + + const heartbeat = setInterval(() => { + update({}).catch(() => {}); + }, HEARTBEAT_MS); + + try { + /** + * The face changes once the job really starts, after any wait for + * its computer. + */ + const run = async () => { + await pushBotNotice({ + bot_id, + bot: await setBotFace({ + bot_id, + face: "working", + status_line: job.title || "", + }), + }); + + return await runner({ + job, + params: parseJobJson({ text: job.params }) || {}, + checkpoint: parseJobJson({ text: job.checkpoint }), + signal: controller.signal, + update, + }); + }; + + const outcome = job.computer_id + ? await computer_lock.run({ key: job.computer_id, task: run }) + : await run(); + + if (controller.signal.aborted) return; + + await pushBotNotice({ + bot_id, + ...(await completeBotJob({ + job_id, + worker, + result: outcome.result, + needs_you: outcome.needs_you, + })), + }); + } catch (error: unknown) { + if (controller.signal.aborted) return; + + const failed = await failBotJob({ + job_id, + worker, + error: error instanceof Error ? error.message : String(error), + fatal: error instanceof BotJobFatalError, + }).catch(async (fail_error) => { + await logError({ + title: `Bot job #${job_id} couldn't be failed`, + error: fail_error, + }).catch(() => {}); + return null; + }); + + if (failed?.notice) { + await pushBotNotice({ bot_id, ...failed.notice }); + } + } finally { + clearInterval(heartbeat); + } +} diff --git a/src/bots-queue/push-bot-notice.ts b/src/bots-queue/push-bot-notice.ts new file mode 100644 index 0000000..55b4f91 --- /dev/null +++ b/src/bots-queue/push-bot-notice.ts @@ -0,0 +1,30 @@ +import type { + BUN_SQLITE_COBBLE_BOT_MESSAGES, + BUN_SQLITE_COBBLE_BOTS, +} from "@/db/types/db"; +import pushToBotOwner from "./push-to-bot-owner"; + +type Params = { + bot_id: number | string; + message?: BUN_SQLITE_COBBLE_BOT_MESSAGES | null; + bot?: Pick | null; +}; + +/** + * Pushes a new bot message and/or face change to the owner's open tabs. + */ +export default async function pushBotNotice({ bot_id, message, bot }: Params) { + if (message) { + await pushToBotOwner({ + bot_id, + data: { event: "server:bot-message", data: message }, + }); + } + + if (bot) { + await pushToBotOwner({ + bot_id, + data: { event: "server:bot-updated", data: bot }, + }); + } +} diff --git a/src/bots-queue/push-to-bot-owner.ts b/src/bots-queue/push-to-bot-owner.ts new file mode 100644 index 0000000..befd7bc --- /dev/null +++ b/src/bots-queue/push-to-bot-owner.ts @@ -0,0 +1,26 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import type { BUN_SQLITE_COBBLE_BOTS } from "@/db/types/db"; +import type { WebSocketDataType } from "@/src/types"; +import { sendToUser } from "@/src/websocket/(functions)/send-to-user"; + +type Params = { + bot_id: number | string; + data: WebSocketDataType; +}; + +/** + * Live update to the bot owner's open tabs. Best effort: nothing is queued + * for offline users, who see the change on their next page load. + */ +export default async function pushToBotOwner({ bot_id, data }: Params) { + const bot = ( + await BunSQLite.sql({ + sql: `SELECT user_id FROM bots WHERE id = ? LIMIT 1`, + values: [bot_id], + }) + ).payload?.[0]; + + if (!bot?.user_id) return; + + sendToUser({ user_id: bot.user_id, data }); +} diff --git a/src/bots-queue/report-lost-bot-jobs.ts b/src/bots-queue/report-lost-bot-jobs.ts new file mode 100644 index 0000000..0c47431 --- /dev/null +++ b/src/bots-queue/report-lost-bot-jobs.ts @@ -0,0 +1,22 @@ +import grabUnreportedFailedBotJobs from "@/src/functions/backend/db/bots-queue/grab-unreported-failed-bot-jobs"; +import notifyBotJobFailed from "@/src/functions/backend/db/bots-queue/notify-bot-job-failed"; +import pushBotNotice from "./push-bot-notice"; + +/** + * Tells the user about jobs that ran out of attempts by going silent + * (bun-sqlite fails those itself, without a message). + */ +export default async function reportLostBotJobs() { + const jobs = await grabUnreportedFailedBotJobs(); + + for (const job of jobs) { + if (!job.bot_id) continue; + + const notice = await notifyBotJobFailed({ + job, + failure_message: `I lost track of "${job.title || "the job"}" and couldn't finish it.`, + }); + + await pushBotNotice({ bot_id: job.bot_id, ...notice }); + } +} diff --git a/src/bots-queue/run-bot-job.ts b/src/bots-queue/run-bot-job.ts new file mode 100644 index 0000000..24168aa --- /dev/null +++ b/src/bots-queue/run-bot-job.ts @@ -0,0 +1,14 @@ +import type { BotJobRunner } from "./types"; +import BotJobFatalError from "./bot-job-fatal-error"; + +/** + * Does a job's actual work: the bot follows `job.instructions` with its + * full prompt (`grabBotPrompt`), on `job.computer_id` if set, and returns + * the message for the user. This is where the model layer plugs in; until + * it exists every job fails at once and the user is told. + */ +const runBotJob: BotJobRunner = async () => { + throw new BotJobFatalError(`Bots can't run queued work yet.`); +}; + +export default runBotJob; diff --git a/src/bots-queue/types.ts b/src/bots-queue/types.ts new file mode 100644 index 0000000..eb9b134 --- /dev/null +++ b/src/bots-queue/types.ts @@ -0,0 +1,34 @@ +import type { BUN_SQLITE_COBBLE_BOTS_QUEUE } from "@/db/types/db"; + +export type BotJobContext = { + job: BUN_SQLITE_COBBLE_BOTS_QUEUE; + params: Record; + /** + * Saved by an earlier attempt, if the job was picked up again. + */ + checkpoint: Record | null; + /** + * Aborted when the job is cancelled or lost to another handler. Stop + * work and return; the result is discarded. + */ + signal: AbortSignal; + /** + * Save progress and a checkpoint. The lock is renewed on a timer + * already, so call this for meaningful steps. False once the job is + * cancelled or lost. + */ + update: (params: { + progress?: string; + checkpoint?: Record; + }) => Promise; +}; + +export type BotJobOutcome = { + /** + * The completion message for the user (markdown). + */ + result: string; + needs_you?: boolean; +}; + +export type BotJobRunner = (context: BotJobContext) => Promise; diff --git a/src/cron/index.ts b/src/cron/index.ts index 05c62d7..7e9d45d 100644 --- a/src/cron/index.ts +++ b/src/cron/index.ts @@ -1,5 +1,6 @@ import bunext from "@moduletrace/bunext"; import cronHeartbeat from "./heartbeat"; +import pruneBotsQueue from "./prune-bots-queue"; import registerCronJob from "./register-cron-job"; export default function cron() { @@ -15,4 +16,10 @@ export default function cron() { task: cronHeartbeat, run_on_start: true, }); + + registerCronJob({ + title: "prune-bots-queue", + schedule: "@daily", + task: pruneBotsQueue, + }); } diff --git a/src/cron/prune-bots-queue/index.ts b/src/cron/prune-bots-queue/index.ts new file mode 100644 index 0000000..9aa97c6 --- /dev/null +++ b/src/cron/prune-bots-queue/index.ts @@ -0,0 +1,23 @@ +import bunext from "@moduletrace/bunext"; +import BunSQLite from "@moduletrace/bun-sqlite"; +import { AppData } from "@/src/data/app-data"; + +/** + * Deletes finished jobs (completed, failed or cancelled) older than + * `AppData.BotQueueRetentionDays`. Their messages stay in the conversation. + */ +export default async function pruneBotsQueue() { + const cutoff = + Date.now() - AppData.BotQueueRetentionDays * 24 * 60 * 60 * 1000; + + const res = await BunSQLite.sql({ + sql: `DELETE FROM bots_queue WHERE status IN ('completed', 'failed') AND finished_at IS NOT NULL AND finished_at < ? RETURNING id`, + values: [cutoff], + }); + + bunext.bunextLog.info( + `Cron: pruned ${res.payload?.length || 0} finished bot job(s)`, + ); + + return res; +} diff --git a/src/data/app-data.ts b/src/data/app-data.ts index 7a3f540..32029b7 100644 --- a/src/data/app-data.ts +++ b/src/data/app-data.ts @@ -26,4 +26,41 @@ export const AppData = { PasswordMinLength: 8, PasswordMaxLength: 128, PasswordAttemptLimit: 10, + /** + * Docker network shared by the main container and every computer. The + * installer creates it, and the main computer's container on it. + */ + DockerNetworkName: SiteData["SiteSlug"], + MainComputerContainerName: `${SiteData["SiteSlug"]}-computer`, + /** + * Port the AIO sandbox listens on inside its container. + */ + ComputerPort: 8080, + /** + * Cap on a bot's `system_prompt` and `memory`, and on a workspace's + * `system_prompt`, so self-updates can't grow a prompt without bound. + */ + PromptMaxLength: 20000, + /** + * `bots_queue` (bun-sqlite `isQueue`). A job that hasn't sent an update + * within the lock timeout is presumed lost and goes back to the queue, + * so long computer jobs must call `updateBotJob` more often than this. + */ + BotQueueMaxAttempts: 3, + BotQueueLockTimeoutMs: 10 * 60 * 1000, + BotQueueBackoffMs: 30 * 1000, + /** + * The queue handler's lock name. Stable across restarts, so on startup + * it can release the jobs it was running when the process stopped. + */ + BotQueueWorker: "main", + BotQueuePollMs: 3000, + /** + * Jobs run at once. Jobs on the same computer always run one at a time. + */ + BotQueueConcurrency: 3, + /** + * Finished jobs are deleted after this many days. + */ + BotQueueRetentionDays: 30, } as const; diff --git a/src/data/coordinator-system-prompt.ts b/src/data/coordinator-system-prompt.ts index d67e63d..4bcb0e0 100644 --- a/src/data/coordinator-system-prompt.ts +++ b/src/data/coordinator-system-prompt.ts @@ -2,8 +2,8 @@ import { SiteData } from "./site-data"; /** * The coordinator's default core instructions (markdown), attached when the - * coordinator is created. Stored per bot in `bots.system_prompt`, so a - * workspace can edit its own copy without touching this default. + * coordinator is created. Stored per bot in `bots.system_prompt`, so each + * coordinator can edit its own copy without touching this default. */ export const CoordinatorSystemPrompt = `# You are the Coordinator diff --git a/src/dict/bot-dict.ts b/src/dict/bot-dict.ts index ac8e5ed..e9ddfff 100644 --- a/src/dict/bot-dict.ts +++ b/src/dict/bot-dict.ts @@ -49,7 +49,7 @@ export const BotKinds = [ export type BotKind = (typeof BotKinds)[number]["value"]; /** - * Every workspace has exactly one coordinator, created with the account. It + * Every user has exactly one coordinator, created with the account. It * coordinates the crew and hands work to the other bots. */ export const CoordinatorBotName = "Coordinator"; diff --git a/src/dict/bot-queue-dict.ts b/src/dict/bot-queue-dict.ts new file mode 100644 index 0000000..b3e1e47 --- /dev/null +++ b/src/dict/bot-queue-dict.ts @@ -0,0 +1,11 @@ +/** + * `last_error` of a job the user cancelled. Cancelled jobs are `failed` + * (bun-sqlite queues have no cancelled status) with this error. + */ +export const BotJobCancelledError = "Cancelled"; + +/** + * `last_error` bun-sqlite sets when it releases a job whose handler went + * silent past the lock timeout on its last attempt. + */ +export const BotJobLockTimedOutError = "Lock timed out"; diff --git a/src/dict/computer-dict.ts b/src/dict/computer-dict.ts new file mode 100644 index 0000000..d9b4d3e --- /dev/null +++ b/src/dict/computer-dict.ts @@ -0,0 +1,72 @@ +/** + * A computer is an AIO sandbox running in its own docker container on the + * cobble network. The main computer is set up at install and recorded on + * signup; sandboxes are created later by the main container. + */ +export const ComputerKinds = [ + { + title: "Main", + value: "main", + }, + { + title: "Sandbox", + value: "sandbox", + }, +] as const; + +export type ComputerKind = (typeof ComputerKinds)[number]["value"]; + +export const MainComputerName = "Main computer"; + +/** + * Docker container state as reported by `docker inspect` (`State.Status`), + * plus `unknown` for a record that hasn't been inspected yet. + */ +export const ContainerStatuses = [ + { + title: "Created", + value: "created", + }, + { + title: "Running", + value: "running", + }, + { + title: "Paused", + value: "paused", + }, + { + title: "Restarting", + value: "restarting", + }, + { + title: "Exited", + value: "exited", + }, + { + title: "Removing", + value: "removing", + }, + { + title: "Dead", + value: "dead", + }, + { + title: "Unknown", + value: "unknown", + }, +] as const; + +export type ContainerStatus = (typeof ContainerStatuses)[number]["value"]; + +/** + * What a recorded container is for. The main app container is never recorded. + */ +export const ContainerPurposes = [ + { + title: "Computer", + value: "computer", + }, +] as const; + +export type ContainerPurpose = (typeof ContainerPurposes)[number]["value"]; diff --git a/src/functions/backend/db/bots-queue/cancel-bot-job.ts b/src/functions/backend/db/bots-queue/cancel-bot-job.ts new file mode 100644 index 0000000..9a72f60 --- /dev/null +++ b/src/functions/backend/db/bots-queue/cancel-bot-job.ts @@ -0,0 +1,41 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import type { BUN_SQLITE_COBBLE_BOTS_QUEUE } from "@/db/types/db"; +import { BotJobCancelledError } from "@/src/dict/bot-queue-dict"; +import AppError from "@/src/utils/app-error"; + +type Params = { + user_id: number | string; + job_id: number | string; +}; + +/** + * Cancels one of the user's pending or running jobs. It's marked failed + * with `BotJobCancelledError`; a running handler finds out on its next + * `updateBotJob` (which returns false) and stops. Its task is closed. + */ +export default async function cancelBotJob({ user_id, job_id }: Params) { + const now = Date.now(); + + const res = await BunSQLite.sql({ + 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') + AND bot_id IN (SELECT id FROM bots WHERE user_id = ?) + RETURNING *`, + values: [BotJobCancelledError, now, now, job_id, user_id], + }); + + const job = res.payload?.[0]; + + if (!job?.id) { + throw new AppError(`No pending or running job to cancel.`); + } + + if (job.task_id) { + await BunSQLite.sql({ + sql: `UPDATE bot_tasks SET status = 'done', updated_at = ? WHERE id = ?`, + values: [now, job.task_id], + }); + } + + return job; +} diff --git a/src/functions/backend/db/bots-queue/claim-bot-jobs.ts b/src/functions/backend/db/bots-queue/claim-bot-jobs.ts new file mode 100644 index 0000000..d3de15f --- /dev/null +++ b/src/functions/backend/db/bots-queue/claim-bot-jobs.ts @@ -0,0 +1,51 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import type { + BUN_SQLITE_COBBLE_BOTS_QUEUE, + BunSQLiteTables, +} from "@/db/types/db"; + +type Params = { + /** + * Unique id of the handler claiming, e.g. `main:`. Every later + * update, completion or failure must pass the same id. + */ + worker: string; + limit?: number; +}; + +/** + * Claims the next pending jobs for a queue handler. Jobs whose handler + * stopped sending updates are released back to the queue first, so a lost + * job is picked up again (resume from its `checkpoint`). + */ +export default async function claimBotJobs({ worker, limit = 1 }: Params) { + const res = await BunSQLite.queue.claim< + BUN_SQLITE_COBBLE_BOTS_QUEUE, + (typeof BunSQLiteTables)[number] + >({ + table: "bots_queue", + worker, + limit, + }); + + if (!res.success) { + throw new Error(`Couldn't claim jobs: ${res.msg || res.error}`); + } + + const jobs = res.payload || []; + + for (const job of jobs) { + if (job.id && !job.started_at) { + const now = Date.now(); + + await BunSQLite.sql({ + sql: `UPDATE bots_queue SET started_at = ? WHERE id = ?`, + values: [now, job.id], + }); + + job.started_at = now; + } + } + + return jobs; +} diff --git a/src/functions/backend/db/bots-queue/complete-bot-job.ts b/src/functions/backend/db/bots-queue/complete-bot-job.ts new file mode 100644 index 0000000..1be01d3 --- /dev/null +++ b/src/functions/backend/db/bots-queue/complete-bot-job.ts @@ -0,0 +1,80 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import AppError from "@/src/utils/app-error"; +import insertBotMessage from "../bots/insert-bot-message"; +import setBotFace from "../bots/set-bot-face"; +import countOpenBotJobs from "./count-open-bot-jobs"; +import grabLockedBotJob from "./grab-locked-bot-job"; + +type Params = { + job_id: number | string; + worker: string; + /** + * The completion message for the user (markdown). Also kept as the + * job's `result`. + */ + result: string; + /** + * Flag the message as needing the user. + */ + needs_you?: boolean; +}; + +/** + * Marks a job done, closes its task and sends the completion message back + * to the user in the bot's conversation. The bot shows done (or needs-you) + * unless it has other jobs open. Throws if this handler no longer holds + * the job. + */ +export default async function completeBotJob({ + job_id, + worker, + result, + needs_you, +}: Params) { + const job = await grabLockedBotJob({ job_id, worker }); + + if (!job?.id || !job.bot_id) { + throw new AppError(`This job is no longer held by ${worker}.`); + } + + const res = await BunSQLite.queue.complete({ + table: "bots_queue", + id: job.id, + }); + + if (!res.success) { + throw new Error(`Couldn't complete the job: ${res.msg || res.error}`); + } + + const now = Date.now(); + + await BunSQLite.sql({ + sql: `UPDATE bots_queue SET result = ?, finished_at = ? WHERE id = ?`, + values: [result, now, job.id], + }); + + if (job.task_id) { + await BunSQLite.sql({ + sql: `UPDATE bot_tasks SET status = 'done', updated_at = ? WHERE id = ?`, + values: [now, job.task_id], + }); + } + + const message = await insertBotMessage({ + bot_id: job.bot_id, + text: result, + needs_you, + }); + + const bot = needs_you + ? await setBotFace({ + bot_id: job.bot_id, + face: "needs-you", + status_line: "Waiting on you", + }) + : (await countOpenBotJobs({ bot_id: job.bot_id })) + ? null + : await setBotFace({ bot_id: job.bot_id, face: "done" }); + + return { message, bot }; +} diff --git a/src/functions/backend/db/bots-queue/count-open-bot-jobs.ts b/src/functions/backend/db/bots-queue/count-open-bot-jobs.ts new file mode 100644 index 0000000..b961070 --- /dev/null +++ b/src/functions/backend/db/bots-queue/count-open-bot-jobs.ts @@ -0,0 +1,17 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; + +type Params = { + bot_id: number | string; +}; + +/** + * How many of a bot's jobs are still pending or processing. + */ +export default async function countOpenBotJobs({ bot_id }: Params) { + const res = await BunSQLite.sql<{ open: number }>({ + sql: `SELECT COUNT(*) AS open FROM bots_queue WHERE bot_id = ? AND status IN ('pending', 'processing')`, + values: [bot_id], + }); + + return Number(res.payload?.[0]?.open || 0); +} diff --git a/src/functions/backend/db/bots-queue/enqueue-bot-job.ts b/src/functions/backend/db/bots-queue/enqueue-bot-job.ts new file mode 100644 index 0000000..d57a645 --- /dev/null +++ b/src/functions/backend/db/bots-queue/enqueue-bot-job.ts @@ -0,0 +1,148 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import type { + BUN_SQLITE_COBBLE_BOT_TASKS, + BUN_SQLITE_COBBLE_BOTS_QUEUE, + BunSQLiteTables, +} from "@/db/types/db"; +import AppError from "@/src/utils/app-error"; +import grabUserBot from "../bots/grab-user-bot"; +import grabBotComputers from "../computers/grab-bot-computers"; + +type Params = { + user_id: number | string; + bot_id: number | string; + title: string; + /** + * The brief the bot writes for itself (markdown): what to do and how + * it knows it's done. + */ + instructions: string; + params?: Record; + /** + * The user message that asked for this work. + */ + message_id?: number | string; + /** + * The task for the working-on panel. One is created from `title` when + * not given. + */ + task_id?: number | string; + /** + * Must be one of the bot's computers (direct or through a workspace). + */ + computer_id?: number | string; + /** + * Milliseconds to hold the job back before it can be claimed. + */ + delay?: number; +}; + +/** + * Queues long-running work for one of the user's bots and returns the job + * id. The bot replies to the user straight away (that it's on it); the + * queue handler does the work and `completeBotJob` posts the result. Not + * every message needs a job; quick answers are replied to directly. + */ +export default async function enqueueBotJob({ + user_id, + bot_id, + title, + instructions, + params, + message_id, + task_id, + computer_id, + delay, +}: Params) { + const bot = await grabUserBot({ bot_id, user_id }); + + if (!bot?.id) { + throw new AppError(`Bot not found.`); + } + + if (message_id && !(await belongsToBot({ table: "bot_messages", id: message_id, bot_id: bot.id }))) { + throw new AppError(`Message not found.`); + } + + if (task_id && !(await belongsToBot({ table: "bot_tasks", id: task_id, bot_id: bot.id }))) { + throw new AppError(`Task not found.`); + } + + if (computer_id) { + const computers = await grabBotComputers({ bot_id: bot.id, user_id }); + + if (!computers.some((c) => String(c.id) == String(computer_id))) { + throw new AppError(`This bot has no access to that computer.`); + } + } + + const final_task_id = task_id || (await createJobTask({ bot_id: bot.id, title })); + + const res = await BunSQLite.queue.add< + BUN_SQLITE_COBBLE_BOTS_QUEUE, + (typeof BunSQLiteTables)[number] + >({ + table: "bots_queue", + data: [ + { + bot_id: bot.id, + title, + instructions, + params: params ? JSON.stringify(params) : undefined, + message_id: message_id ? Number(message_id) : undefined, + task_id: Number(final_task_id), + computer_id: computer_id ? Number(computer_id) : undefined, + }, + ], + delay, + }); + + const job_id = res.singleRes?.id || res.postInsertReturn?.insertId; + + if (!res.success || !job_id) { + throw new Error(`Couldn't queue the job: ${res.msg || res.error}`); + } + + return job_id; +} + +async function belongsToBot({ + table, + id, + bot_id, +}: { + table: "bot_messages" | "bot_tasks"; + id: number | string; + bot_id: number; +}) { + const res = await BunSQLite.sql({ + sql: `SELECT id FROM ${table} WHERE id = ? AND bot_id = ? LIMIT 1`, + values: [id, bot_id], + }); + + return Boolean(res.payload?.[0]); +} + +async function createJobTask({ + bot_id, + title, +}: { + bot_id: number; + title: string; +}) { + const res = await BunSQLite.insert< + BUN_SQLITE_COBBLE_BOT_TASKS, + (typeof BunSQLiteTables)[number] + >({ + table: "bot_tasks", + data: [{ bot_id, title, status: "active" }], + }); + + const task_id = res.singleRes?.id || res.postInsertReturn?.insertId; + + if (!res.success || !task_id) { + throw new Error(`Couldn't create the job's task: ${res.msg}`); + } + + return task_id; +} diff --git a/src/functions/backend/db/bots-queue/fail-bot-job.ts b/src/functions/backend/db/bots-queue/fail-bot-job.ts new file mode 100644 index 0000000..cecc85d --- /dev/null +++ b/src/functions/backend/db/bots-queue/fail-bot-job.ts @@ -0,0 +1,72 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import type { BUN_SQLITE_COBBLE_BOTS_QUEUE } from "@/db/types/db"; +import { AppData } from "@/src/data/app-data"; +import AppError from "@/src/utils/app-error"; +import grabLockedBotJob from "./grab-locked-bot-job"; +import notifyBotJobFailed from "./notify-bot-job-failed"; + +type Params = { + job_id: number | string; + worker: string; + error: string; + /** + * Retrying won't help (bad parameters, missing access): fail now + * instead of using the remaining attempts. + */ + fatal?: boolean; + /** + * What to tell the user if the job is out of attempts. Defaults to a + * short note with the job title and error. + */ + failure_message?: string; +}; + +/** + * Records a failed attempt. The job goes back to the queue with a backoff + * delay, or, once out of attempts (or `fatal`), is marked failed and the + * bot tells the user it needs them. Returns the job's new status and the + * failure notice, if one was sent. + */ +export default async function failBotJob({ + job_id, + worker, + error, + fatal, + failure_message, +}: Params) { + const job = await grabLockedBotJob({ job_id, worker }); + + if (!job?.id || !job.bot_id) { + throw new AppError(`This job is no longer held by ${worker}.`); + } + + if (fatal) { + await BunSQLite.sql({ + sql: `UPDATE bots_queue SET attempts = MAX(attempts, ?) WHERE id = ?`, + values: [AppData.BotQueueMaxAttempts, job.id], + }); + } + + const res = await BunSQLite.queue.fail({ + table: "bots_queue", + id: job.id, + error, + }); + + if (!res.success || !res.singleRes) { + throw new Error(`Couldn't fail the job: ${res.msg || res.error}`); + } + + const status = res.singleRes.status; + + if (status != "failed") { + return { status, notice: null }; + } + + const notice = await notifyBotJobFailed({ + job: res.singleRes, + failure_message, + }); + + return { status, notice }; +} diff --git a/src/functions/backend/db/bots-queue/grab-locked-bot-job.ts b/src/functions/backend/db/bots-queue/grab-locked-bot-job.ts new file mode 100644 index 0000000..08c9714 --- /dev/null +++ b/src/functions/backend/db/bots-queue/grab-locked-bot-job.ts @@ -0,0 +1,23 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import type { BUN_SQLITE_COBBLE_BOTS_QUEUE } from "@/db/types/db"; + +type Params = { + job_id: number | string; + worker: string; +}; + +/** + * A job only while the given handler still holds it. Null once its lock + * timed out and it went back to the queue (or another handler took it). + */ +export default async function grabLockedBotJob({ + job_id, + worker, +}: Params): Promise { + const res = await BunSQLite.sql({ + sql: `SELECT * FROM bots_queue WHERE id = ? AND status = 'processing' AND locked_by = ? LIMIT 1`, + values: [job_id, worker], + }); + + return res.payload?.[0] || null; +} diff --git a/src/functions/backend/db/bots-queue/grab-unreported-failed-bot-jobs.ts b/src/functions/backend/db/bots-queue/grab-unreported-failed-bot-jobs.ts new file mode 100644 index 0000000..ffe7430 --- /dev/null +++ b/src/functions/backend/db/bots-queue/grab-unreported-failed-bot-jobs.ts @@ -0,0 +1,15 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import type { BUN_SQLITE_COBBLE_BOTS_QUEUE } from "@/db/types/db"; + +/** + * Failed jobs the user hasn't been told about: the ones bun-sqlite failed + * itself after a lock timeout on their last attempt. Jobs failed through + * `failBotJob` or cancelled already have `finished_at`. + */ +export default async function grabUnreportedFailedBotJobs() { + const res = await BunSQLite.sql({ + sql: `SELECT * FROM bots_queue WHERE status = 'failed' AND finished_at IS NULL ORDER BY id ASC`, + }); + + return res.payload || []; +} diff --git a/src/functions/backend/db/bots-queue/notify-bot-job-failed.ts b/src/functions/backend/db/bots-queue/notify-bot-job-failed.ts new file mode 100644 index 0000000..0f20200 --- /dev/null +++ b/src/functions/backend/db/bots-queue/notify-bot-job-failed.ts @@ -0,0 +1,46 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import type { BUN_SQLITE_COBBLE_BOTS_QUEUE } from "@/db/types/db"; +import insertBotMessage from "../bots/insert-bot-message"; +import setBotFace from "../bots/set-bot-face"; + +type Params = { + job: BUN_SQLITE_COBBLE_BOTS_QUEUE; + failure_message?: string; +}; + +/** + * Wraps up a job that is out of attempts: stamps `finished_at` (so it's + * reported once), tells the user in the bot's conversation and sets the + * bot to needs-you. Its task stays active, since the work isn't done. + */ +export default async function notifyBotJobFailed({ + job, + failure_message, +}: Params) { + if (!job.id || !job.bot_id) { + throw new Error(`Job has no id or bot.`); + } + + await BunSQLite.sql({ + sql: `UPDATE bots_queue SET finished_at = ? WHERE id = ?`, + values: [Date.now(), job.id], + }); + + const message = await insertBotMessage({ + bot_id: job.bot_id, + text: + failure_message || + [`I couldn't finish "${job.title || "the job"}".`, job.last_error] + .filter(Boolean) + .join(" "), + needs_you: true, + }); + + const bot = await setBotFace({ + bot_id: job.bot_id, + face: "needs-you", + status_line: "Waiting on you", + }); + + return { message, bot }; +} diff --git a/src/functions/backend/db/bots-queue/release-worker-bot-jobs.ts b/src/functions/backend/db/bots-queue/release-worker-bot-jobs.ts new file mode 100644 index 0000000..4fbe823 --- /dev/null +++ b/src/functions/backend/db/bots-queue/release-worker-bot-jobs.ts @@ -0,0 +1,21 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; + +type Params = { + worker: string; +}; + +/** + * Puts the jobs a handler was running back in the queue straight away, + * instead of waiting out the lock timeout. For startup, after a restart + * cut those jobs off. The cut-off attempt isn't counted. + */ +export default async function releaseWorkerBotJobs({ worker }: Params) { + const now = Date.now(); + + const res = await BunSQLite.sql({ + sql: `UPDATE bots_queue SET status = 'pending', attempts = MAX(attempts - 1, 0), available_at = ?, locked_at = NULL, locked_by = NULL, updated_at = ? WHERE status = 'processing' AND locked_by = ? RETURNING id`, + values: [now, now, worker], + }); + + return res.payload?.length || 0; +} diff --git a/src/functions/backend/db/bots-queue/update-bot-job.ts b/src/functions/backend/db/bots-queue/update-bot-job.ts new file mode 100644 index 0000000..a503ab8 --- /dev/null +++ b/src/functions/backend/db/bots-queue/update-bot-job.ts @@ -0,0 +1,45 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; + +type Params = { + job_id: number | string; + worker: string; + /** + * Latest progress note (markdown), e.g. "Installed deps, running tests". + */ + progress?: string; + /** + * Whatever the bot needs to resume if the job is picked up again. + */ + checkpoint?: Record; +}; + +/** + * The bot's "still working" update while it processes a job. Renews the + * lock (so the job isn't presumed lost) and saves progress and a + * checkpoint. Call it more often than `AppData.BotQueueLockTimeoutMs`. + * + * Returns false if this handler no longer holds the job: stop working on + * it, since it has gone back to the queue. + */ +export default async function updateBotJob({ + job_id, + worker, + progress, + checkpoint, +}: Params) { + const now = Date.now(); + + const res = await BunSQLite.sql({ + sql: `UPDATE bots_queue SET locked_at = ?, updated_at = ?, progress = COALESCE(?, progress), checkpoint = COALESCE(?, checkpoint) WHERE id = ? AND status = 'processing' AND locked_by = ? RETURNING id`, + values: [ + now, + now, + progress ?? null, + checkpoint ? JSON.stringify(checkpoint) : null, + job_id, + worker, + ], + }); + + return Boolean(res.payload?.[0]); +} diff --git a/src/functions/backend/db/bots/ensure-coordinator-bot.ts b/src/functions/backend/db/bots/ensure-coordinator-bot.ts index 7c5916b..29cb9f5 100644 --- a/src/functions/backend/db/bots/ensure-coordinator-bot.ts +++ b/src/functions/backend/db/bots/ensure-coordinator-bot.ts @@ -9,7 +9,7 @@ type Params = { /** * Returns the id of the user's coordinator, creating it first (with the - * default `CoordinatorSystemPrompt`) if the workspace has none. A + * default `CoordinatorSystemPrompt`) if the user has none. A * coordinator without a prompt gets the default; an edited prompt is never * overwritten. Safe to call on every visit. */ diff --git a/src/functions/backend/db/bots/grab-bot-prompt.ts b/src/functions/backend/db/bots/grab-bot-prompt.ts new file mode 100644 index 0000000..7b276d2 --- /dev/null +++ b/src/functions/backend/db/bots/grab-bot-prompt.ts @@ -0,0 +1,39 @@ +import type { BUN_SQLITE_COBBLE_WORKSPACES } from "@/db/types/db"; +import BunSQLite from "@moduletrace/bun-sqlite"; +import composeBotPrompt from "@/src/utils/compose-bot-prompt"; +import AppError from "@/src/utils/app-error"; +import grabUserBot from "./grab-user-bot"; + +type Params = { + bot_id: number | string; + user_id: number | string; +}; + +/** + * The full prompt for one of the user's bots: its core instructions, the + * system prompt of every non-archived workspace it's in, and its memory. + */ +export default async function grabBotPrompt({ bot_id, user_id }: Params) { + const bot = await grabUserBot({ bot_id, user_id }); + + if (!bot?.id) { + throw new AppError(`Bot not found.`); + } + + const workspaces = + ( + await BunSQLite.sql({ + sql: `SELECT w.name, w.system_prompt FROM workspaces w + JOIN workspace_bots wb ON wb.workspace_id = w.id + WHERE wb.bot_id = ? AND w.user_id = ? AND w.archived = 0 + ORDER BY w.id ASC`, + values: [bot.id, user_id], + }) + ).payload || []; + + return composeBotPrompt({ + system_prompt: bot.system_prompt, + memory: bot.memory, + workspaces, + }); +} diff --git a/src/functions/backend/db/bots/grab-bots.ts b/src/functions/backend/db/bots/grab-bots.ts index f4171e7..41744ea 100644 --- a/src/functions/backend/db/bots/grab-bots.ts +++ b/src/functions/backend/db/bots/grab-bots.ts @@ -6,7 +6,7 @@ type Params = { }; /** - * A workspace's crew for the rail: the coordinator pinned first, then + * A user's crew for the rail: the coordinator pinned first, then * workers by latest activity. Each row carries its latest message. */ export default async function grabBots({ diff --git a/src/functions/backend/db/bots/grab-user-bot.ts b/src/functions/backend/db/bots/grab-user-bot.ts index fa0398b..b96628e 100644 --- a/src/functions/backend/db/bots/grab-user-bot.ts +++ b/src/functions/backend/db/bots/grab-user-bot.ts @@ -7,7 +7,7 @@ type Params = { }; /** - * A bot only if it belongs to the given workspace owner. + * A bot only if it belongs to the given owner. */ export default async function grabUserBot({ bot_id, diff --git a/src/functions/backend/db/bots/insert-bot-message.ts b/src/functions/backend/db/bots/insert-bot-message.ts new file mode 100644 index 0000000..5f2a269 --- /dev/null +++ b/src/functions/backend/db/bots/insert-bot-message.ts @@ -0,0 +1,53 @@ +import type { + BUN_SQLITE_COBBLE_BOT_MESSAGES, + BunSQLiteTables, +} from "@/db/types/db"; +import BunSQLite from "@moduletrace/bun-sqlite"; + +type Params = { + bot_id: number; + text: string; + needs_you?: boolean; +}; + +/** + * Appends a bot message to its conversation, marks the bot unread and + * bumps its activity. Callers must check the bot first. + */ +export default async function insertBotMessage({ + bot_id, + text, + needs_you, +}: Params): Promise { + const now = Date.now(); + const needs_you_flag = needs_you ? 1 : 0; + + const res = await BunSQLite.insert< + BUN_SQLITE_COBBLE_BOT_MESSAGES, + (typeof BunSQLiteTables)[number] + >({ + table: "bot_messages", + data: [{ bot_id, author: "bot", text, needs_you: needs_you_flag }], + }); + + const id = res.postInsertReturn?.insertId; + + if (!res.success || !id) { + throw new Error(`Failed to save message: ${res.msg}`); + } + + await BunSQLite.sql({ + sql: `UPDATE bots SET last_activity_at = ?, unread = 1 WHERE id = ?`, + values: [now, bot_id], + }); + + return { + id, + bot_id, + author: "bot", + text, + needs_you: needs_you_flag, + created_at: now, + updated_at: now, + }; +} diff --git a/src/functions/backend/db/bots/set-bot-face.ts b/src/functions/backend/db/bots/set-bot-face.ts new file mode 100644 index 0000000..9ff3b96 --- /dev/null +++ b/src/functions/backend/db/bots/set-bot-face.ts @@ -0,0 +1,25 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import type { BotFace } from "@/src/dict/bot-dict"; + +type Params = { + bot_id: number | string; + face: BotFace; + status_line?: string; +}; + +/** + * Sets a bot's face (its status) and status line. Callers must check the + * bot first. + */ +export default async function setBotFace({ + bot_id, + face, + status_line = "", +}: Params) { + await BunSQLite.sql({ + sql: `UPDATE bots SET face = ?, status_line = ?, updated_at = ? WHERE id = ?`, + values: [face, status_line, Date.now(), bot_id], + }); + + return { id: Number(bot_id), face, status_line }; +} diff --git a/src/functions/backend/db/bots/update-bot-memory.ts b/src/functions/backend/db/bots/update-bot-memory.ts new file mode 100644 index 0000000..6242ca0 --- /dev/null +++ b/src/functions/backend/db/bots/update-bot-memory.ts @@ -0,0 +1,36 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import AppError from "@/src/utils/app-error"; +import assertPromptLength from "@/src/utils/assert-prompt-length"; +import grabUserBot from "./grab-user-bot"; + +type Params = { + user_id: number | string; + bot_id: number | string; + /** + * The whole new memory (markdown), replacing the old one. + */ + memory: string; +}; + +/** + * Replaces a bot's memory. The bot calls this as it learns what it must + * keep in mind; the user can edit it too. + */ +export default async function updateBotMemory({ + user_id, + bot_id, + memory, +}: Params) { + assertPromptLength({ label: "Memory", text: memory }); + + const bot = await grabUserBot({ bot_id, user_id }); + + if (!bot?.id) { + throw new AppError(`Bot not found.`); + } + + await BunSQLite.sql({ + sql: `UPDATE bots SET memory = ?, updated_at = ? WHERE id = ?`, + values: [memory, Date.now(), bot.id], + }); +} diff --git a/src/functions/backend/db/bots/update-bot-system-prompt.ts b/src/functions/backend/db/bots/update-bot-system-prompt.ts new file mode 100644 index 0000000..815b461 --- /dev/null +++ b/src/functions/backend/db/bots/update-bot-system-prompt.ts @@ -0,0 +1,40 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import AppError from "@/src/utils/app-error"; +import assertPromptLength from "@/src/utils/assert-prompt-length"; +import grabUserBot from "./grab-user-bot"; + +type Params = { + user_id: number | string; + bot_id: number | string; + /** + * The whole new core instructions (markdown), replacing the old ones. + */ + system_prompt: string; +}; + +/** + * Replaces a bot's core instructions. The bot calls this to adapt how it + * works to the user's workflow; the user can edit it too. + */ +export default async function updateBotSystemPrompt({ + user_id, + bot_id, + system_prompt, +}: Params) { + if (!system_prompt.trim()) { + throw new AppError(`A bot's core instructions can't be empty.`); + } + + assertPromptLength({ label: "System prompt", text: system_prompt }); + + const bot = await grabUserBot({ bot_id, user_id }); + + if (!bot?.id) { + throw new AppError(`Bot not found.`); + } + + await BunSQLite.sql({ + sql: `UPDATE bots SET system_prompt = ?, updated_at = ? WHERE id = ?`, + values: [system_prompt, Date.now(), bot.id], + }); +} diff --git a/src/functions/backend/db/computers/add-bot-to-computer.ts b/src/functions/backend/db/computers/add-bot-to-computer.ts new file mode 100644 index 0000000..2ef3681 --- /dev/null +++ b/src/functions/backend/db/computers/add-bot-to-computer.ts @@ -0,0 +1,32 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import AppError from "@/src/utils/app-error"; +import grabUserBot from "../bots/grab-user-bot"; +import grabUserComputer from "./grab-user-computer"; + +type Params = { + user_id: number | string; + bot_id: number | string; + computer_id: number | string; +}; + +/** + * Gives one of the user's bots access to one of their computers. Adding a + * bot that already has access is a no-op. + */ +export default async function addBotToComputer({ + user_id, + bot_id, + computer_id, +}: Params) { + const bot = await grabUserBot({ bot_id, user_id }); + const computer = await grabUserComputer({ computer_id, user_id }); + + if (!bot?.id || !computer?.id) { + throw new AppError(`Bot or computer not found.`); + } + + await BunSQLite.sql({ + sql: `INSERT OR IGNORE INTO bot_computers (bot_id, computer_id, created_at, updated_at) VALUES (?, ?, ?, ?)`, + values: [bot.id, computer.id, Date.now(), Date.now()], + }); +} diff --git a/src/functions/backend/db/computers/ensure-main-computer-container.ts b/src/functions/backend/db/computers/ensure-main-computer-container.ts new file mode 100644 index 0000000..b423b89 --- /dev/null +++ b/src/functions/backend/db/computers/ensure-main-computer-container.ts @@ -0,0 +1,60 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import type { + BUN_SQLITE_COBBLE_CONTAINERS, + BunSQLiteTables, +} from "@/db/types/db"; +import { AppData } from "@/src/data/app-data"; + +/** + * Returns the id of the main computer's container record, creating it if + * missing. The container itself is created at install, named + * `AppData.MainComputerContainerName` on the `AppData.DockerNetworkName` + * network; this only records it. Safe to call on every visit. + */ +export default async function ensureMainComputerContainer() { + const name = AppData.MainComputerContainerName; + + const existing = ( + await BunSQLite.select< + BUN_SQLITE_COBBLE_CONTAINERS, + (typeof BunSQLiteTables)[number] + >({ + table: "containers", + query: { + query: { + name: { value: name }, + }, + limit: 1, + }, + }) + ).payload?.[0]; + + if (existing?.id) { + return existing.id; + } + + const res = await BunSQLite.insert< + BUN_SQLITE_COBBLE_CONTAINERS, + (typeof BunSQLiteTables)[number] + >({ + table: "containers", + data: [ + { + name, + purpose: "computer", + network: AppData.DockerNetworkName, + hostname: name, + port: AppData.ComputerPort, + status: "unknown", + }, + ], + }); + + const container_id = res.singleRes?.id || res.postInsertReturn?.insertId; + + if (!res.success || !container_id) { + throw new Error(`Couldn't record the main computer's container: ${res.msg}`); + } + + return container_id; +} diff --git a/src/functions/backend/db/computers/ensure-main-computer.ts b/src/functions/backend/db/computers/ensure-main-computer.ts new file mode 100644 index 0000000..006f089 --- /dev/null +++ b/src/functions/backend/db/computers/ensure-main-computer.ts @@ -0,0 +1,85 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import type { + BUN_SQLITE_COBBLE_COMPUTERS, + BunSQLiteTables, +} from "@/db/types/db"; +import { MainComputerName } from "@/src/dict/computer-dict"; +import ensureMainComputerContainer from "./ensure-main-computer-container"; + +type Params = { + user_id: number | string; + /** + * The owner's coordinator (from `ensureCoordinatorBot`). It always has + * access to the main computer. + */ + coordinator_id: number | string; +}; + +/** + * Returns the id of the user's main computer, recording it (and its + * container) first if missing, and makes sure the coordinator has access + * to it. Safe to call on every visit. + */ +export default async function ensureMainComputer({ + user_id, + coordinator_id, +}: Params) { + const computer_id = await grabOrCreateMainComputer({ user_id }); + + await BunSQLite.sql({ + sql: `INSERT OR IGNORE INTO bot_computers (bot_id, computer_id, created_at, updated_at) VALUES (?, ?, ?, ?)`, + values: [coordinator_id, computer_id, Date.now(), Date.now()], + }); + + return computer_id; +} + +async function grabOrCreateMainComputer({ + user_id, +}: Pick) { + const existing = ( + await BunSQLite.select< + BUN_SQLITE_COBBLE_COMPUTERS, + (typeof BunSQLiteTables)[number] + >({ + table: "computers", + query: { + query: { + user_id: { value: user_id }, + kind: { value: "main" }, + }, + order: { field: "id", strategy: "ASC" }, + limit: 1, + }, + }) + ).payload?.[0]; + + if (existing?.id) { + return existing.id; + } + + const container_id = await ensureMainComputerContainer(); + + const res = await BunSQLite.insert< + BUN_SQLITE_COBBLE_COMPUTERS, + (typeof BunSQLiteTables)[number] + >({ + table: "computers", + data: [ + { + user_id: Number(user_id), + container_id: Number(container_id), + name: MainComputerName, + kind: "main", + }, + ], + }); + + const computer_id = res.singleRes?.id || res.postInsertReturn?.insertId; + + if (!res.success || !computer_id) { + throw new Error(`Couldn't create the main computer: ${res.msg}`); + } + + return computer_id; +} diff --git a/src/functions/backend/db/computers/grab-bot-computers.ts b/src/functions/backend/db/computers/grab-bot-computers.ts new file mode 100644 index 0000000..aa87564 --- /dev/null +++ b/src/functions/backend/db/computers/grab-bot-computers.ts @@ -0,0 +1,35 @@ +import type { BUN_SQLITE_COBBLE_COMPUTERS } from "@/db/types/db"; +import BunSQLite from "@moduletrace/bun-sqlite"; + +type Params = { + bot_id: number | string; + user_id: number | string; +}; + +/** + * Every computer a bot can use: the ones it was given directly + * (`bot_computers`) plus the ones assigned to any workspace it's in + * (`workspace_computers`). Only the owner's computers, each once. + */ +export default async function grabBotComputers({ + bot_id, + user_id, +}: Params): Promise { + const res = await BunSQLite.sql({ + sql: `SELECT c.* FROM computers c + WHERE c.user_id = ? + AND ( + c.id IN (SELECT computer_id FROM bot_computers WHERE bot_id = ?) + OR c.id IN ( + SELECT wc.computer_id FROM workspace_computers wc + JOIN workspace_bots wb ON wb.workspace_id = wc.workspace_id + JOIN workspaces w ON w.id = wc.workspace_id + WHERE wb.bot_id = ? AND w.archived = 0 + ) + ) + ORDER BY c.kind = 'main' DESC, c.id ASC`, + values: [user_id, bot_id, bot_id], + }); + + return res.payload || []; +} diff --git a/src/functions/backend/db/computers/grab-user-computer.ts b/src/functions/backend/db/computers/grab-user-computer.ts new file mode 100644 index 0000000..4eb5710 --- /dev/null +++ b/src/functions/backend/db/computers/grab-user-computer.ts @@ -0,0 +1,22 @@ +import type { BUN_SQLITE_COBBLE_COMPUTERS } from "@/db/types/db"; +import BunSQLite from "@moduletrace/bun-sqlite"; + +type Params = { + computer_id: number | string; + user_id: number | string; +}; + +/** + * A computer only if it belongs to the given owner. + */ +export default async function grabUserComputer({ + computer_id, + user_id, +}: Params): Promise { + const res = await BunSQLite.sql({ + sql: `SELECT * FROM computers WHERE id = ? AND user_id = ? LIMIT 1`, + values: [computer_id, user_id], + }); + + return res.payload?.[0] || null; +} diff --git a/src/functions/backend/db/computers/remove-bot-from-computer.ts b/src/functions/backend/db/computers/remove-bot-from-computer.ts new file mode 100644 index 0000000..a495b0d --- /dev/null +++ b/src/functions/backend/db/computers/remove-bot-from-computer.ts @@ -0,0 +1,44 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import AppError from "@/src/utils/app-error"; +import isLockedComputerAccess from "@/src/utils/is-locked-computer-access"; +import grabUserBot from "../bots/grab-user-bot"; +import grabUserComputer from "./grab-user-computer"; + +type Params = { + user_id: number | string; + bot_id: number | string; + computer_id: number | string; +}; + +/** + * Takes a bot's access to a computer away. The coordinator's access to the + * main computer can't be removed. + */ +export default async function removeBotFromComputer({ + user_id, + bot_id, + computer_id, +}: Params) { + const bot = await grabUserBot({ bot_id, user_id }); + const computer = await grabUserComputer({ computer_id, user_id }); + + if (!bot?.id || !computer?.id) { + throw new AppError(`Bot or computer not found.`); + } + + if ( + isLockedComputerAccess({ + bot_kind: bot.kind, + computer_kind: computer.kind, + }) + ) { + throw new AppError( + `The coordinator always has access to the main computer.`, + ); + } + + await BunSQLite.sql({ + sql: `DELETE FROM bot_computers WHERE bot_id = ? AND computer_id = ?`, + values: [bot.id, computer.id], + }); +} diff --git a/src/functions/backend/db/enable-foreign-keys.ts b/src/functions/backend/db/enable-foreign-keys.ts new file mode 100644 index 0000000..a7e9a4d --- /dev/null +++ b/src/functions/backend/db/enable-foreign-keys.ts @@ -0,0 +1,14 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; + +/** + * SQLite leaves foreign keys off per connection, so the schema's + * `ON DELETE CASCADE` rules never fire unless this runs first. bun-sqlite + * reuses one connection per process, so once at startup is enough. + */ +export default async function enableForeignKeys() { + const res = await BunSQLite.sql({ sql: `PRAGMA foreign_keys = ON` }); + + if (!res.success) { + throw new Error(`Couldn't enable foreign keys: ${res.msg}`); + } +} diff --git a/src/functions/backend/db/workspaces/add-bot-to-workspace.ts b/src/functions/backend/db/workspaces/add-bot-to-workspace.ts new file mode 100644 index 0000000..be3f278 --- /dev/null +++ b/src/functions/backend/db/workspaces/add-bot-to-workspace.ts @@ -0,0 +1,32 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import AppError from "@/src/utils/app-error"; +import grabUserBot from "../bots/grab-user-bot"; +import grabUserWorkspace from "./grab-user-workspace"; + +type Params = { + user_id: number | string; + workspace_id: number | string; + bot_id: number | string; +}; + +/** + * Assigns one of the user's bots to one of their workspaces. Assigning a + * bot that is already there is a no-op. + */ +export default async function addBotToWorkspace({ + user_id, + workspace_id, + bot_id, +}: Params) { + const bot = await grabUserBot({ bot_id, user_id }); + const workspace = await grabUserWorkspace({ workspace_id, user_id }); + + if (!bot?.id || !workspace?.id) { + throw new AppError(`Bot or workspace not found.`); + } + + await BunSQLite.sql({ + sql: `INSERT OR IGNORE INTO workspace_bots (workspace_id, bot_id, created_at, updated_at) VALUES (?, ?, ?, ?)`, + values: [workspace.id, bot.id, Date.now(), Date.now()], + }); +} diff --git a/src/functions/backend/db/workspaces/add-computer-to-workspace.ts b/src/functions/backend/db/workspaces/add-computer-to-workspace.ts new file mode 100644 index 0000000..fda5307 --- /dev/null +++ b/src/functions/backend/db/workspaces/add-computer-to-workspace.ts @@ -0,0 +1,32 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import AppError from "@/src/utils/app-error"; +import grabUserComputer from "../computers/grab-user-computer"; +import grabUserWorkspace from "./grab-user-workspace"; + +type Params = { + user_id: number | string; + workspace_id: number | string; + computer_id: number | string; +}; + +/** + * Assigns one of the user's computers to one of their workspaces, so every + * bot in the workspace gets access to it. Assigning it twice is a no-op. + */ +export default async function addComputerToWorkspace({ + user_id, + workspace_id, + computer_id, +}: Params) { + const computer = await grabUserComputer({ computer_id, user_id }); + const workspace = await grabUserWorkspace({ workspace_id, user_id }); + + if (!computer?.id || !workspace?.id) { + throw new AppError(`Computer or workspace not found.`); + } + + await BunSQLite.sql({ + sql: `INSERT OR IGNORE INTO workspace_computers (workspace_id, computer_id, created_at, updated_at) VALUES (?, ?, ?, ?)`, + values: [workspace.id, computer.id, Date.now(), Date.now()], + }); +} diff --git a/src/functions/backend/db/workspaces/create-workspace.ts b/src/functions/backend/db/workspaces/create-workspace.ts new file mode 100644 index 0000000..6f17152 --- /dev/null +++ b/src/functions/backend/db/workspaces/create-workspace.ts @@ -0,0 +1,49 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import type { + BUN_SQLITE_COBBLE_WORKSPACES, + BunSQLiteTables, +} from "@/db/types/db"; +import assertPromptLength from "@/src/utils/assert-prompt-length"; + +type Params = { + user_id: number | string; + name: string; + description?: string; + system_prompt?: string; +}; + +/** + * Creates a workspace owned by the user and returns its id. + */ +export default async function createWorkspace({ + user_id, + name, + description, + system_prompt, +}: Params) { + assertPromptLength({ label: "System prompt", text: system_prompt }); + + const res = await BunSQLite.insert< + BUN_SQLITE_COBBLE_WORKSPACES, + (typeof BunSQLiteTables)[number] + >({ + table: "workspaces", + data: [ + { + user_id: Number(user_id), + name, + description, + system_prompt, + archived: 0, + }, + ], + }); + + const workspace_id = res.singleRes?.id || res.postInsertReturn?.insertId; + + if (!res.success || !workspace_id) { + throw new Error(`Couldn't create the workspace: ${res.msg}`); + } + + return workspace_id; +} diff --git a/src/functions/backend/db/workspaces/grab-user-workspace.ts b/src/functions/backend/db/workspaces/grab-user-workspace.ts new file mode 100644 index 0000000..4348af5 --- /dev/null +++ b/src/functions/backend/db/workspaces/grab-user-workspace.ts @@ -0,0 +1,22 @@ +import type { BUN_SQLITE_COBBLE_WORKSPACES } from "@/db/types/db"; +import BunSQLite from "@moduletrace/bun-sqlite"; + +type Params = { + workspace_id: number | string; + user_id: number | string; +}; + +/** + * A workspace only if it belongs to the given owner. + */ +export default async function grabUserWorkspace({ + workspace_id, + user_id, +}: Params): Promise { + const res = await BunSQLite.sql({ + sql: `SELECT * FROM workspaces WHERE id = ? AND user_id = ? LIMIT 1`, + values: [workspace_id, user_id], + }); + + return res.payload?.[0] || null; +} diff --git a/src/functions/backend/db/workspaces/remove-bot-from-workspace.ts b/src/functions/backend/db/workspaces/remove-bot-from-workspace.ts new file mode 100644 index 0000000..91debfc --- /dev/null +++ b/src/functions/backend/db/workspaces/remove-bot-from-workspace.ts @@ -0,0 +1,29 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import AppError from "@/src/utils/app-error"; +import grabUserWorkspace from "./grab-user-workspace"; + +type Params = { + user_id: number | string; + workspace_id: number | string; + bot_id: number | string; +}; + +/** + * Removes a bot from one of the user's workspaces. + */ +export default async function removeBotFromWorkspace({ + user_id, + workspace_id, + bot_id, +}: Params) { + const workspace = await grabUserWorkspace({ workspace_id, user_id }); + + if (!workspace?.id) { + throw new AppError(`Workspace not found.`); + } + + await BunSQLite.sql({ + sql: `DELETE FROM workspace_bots WHERE workspace_id = ? AND bot_id = ?`, + values: [workspace.id, bot_id], + }); +} diff --git a/src/functions/backend/db/workspaces/remove-computer-from-workspace.ts b/src/functions/backend/db/workspaces/remove-computer-from-workspace.ts new file mode 100644 index 0000000..abdc09d --- /dev/null +++ b/src/functions/backend/db/workspaces/remove-computer-from-workspace.ts @@ -0,0 +1,30 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import AppError from "@/src/utils/app-error"; +import grabUserWorkspace from "./grab-user-workspace"; + +type Params = { + user_id: number | string; + workspace_id: number | string; + computer_id: number | string; +}; + +/** + * Takes a computer off one of the user's workspaces. Its bots keep any + * access they were given directly. + */ +export default async function removeComputerFromWorkspace({ + user_id, + workspace_id, + computer_id, +}: Params) { + const workspace = await grabUserWorkspace({ workspace_id, user_id }); + + if (!workspace?.id) { + throw new AppError(`Workspace not found.`); + } + + await BunSQLite.sql({ + sql: `DELETE FROM workspace_computers WHERE workspace_id = ? AND computer_id = ?`, + values: [workspace.id, computer_id], + }); +} diff --git a/src/functions/backend/db/workspaces/update-workspace.ts b/src/functions/backend/db/workspaces/update-workspace.ts new file mode 100644 index 0000000..562748d --- /dev/null +++ b/src/functions/backend/db/workspaces/update-workspace.ts @@ -0,0 +1,64 @@ +import BunSQLite from "@moduletrace/bun-sqlite"; +import type { + BUN_SQLITE_COBBLE_WORKSPACES, + BunSQLiteTables, +} from "@/db/types/db"; +import AppError from "@/src/utils/app-error"; +import assertPromptLength from "@/src/utils/assert-prompt-length"; +import grabUserWorkspace from "./grab-user-workspace"; +import _ from "lodash"; + +type Params = { + user_id: number | string; + workspace_id: number | string; + /** + * Only the fields given are changed. + */ + data: Pick< + BUN_SQLITE_COBBLE_WORKSPACES, + "name" | "description" | "system_prompt" | "archived" + >; +}; + +/** + * Updates one of the user's workspaces. Its `system_prompt` is shared + * context for every bot in it. + */ +export default async function updateWorkspace({ + user_id, + workspace_id, + data, +}: Params) { + assertPromptLength({ label: "System prompt", text: data.system_prompt }); + + const workspace = await grabUserWorkspace({ workspace_id, user_id }); + + if (!workspace?.id) { + throw new AppError(`Workspace not found.`); + } + + const res = await BunSQLite.update< + BUN_SQLITE_COBBLE_WORKSPACES, + (typeof BunSQLiteTables)[number] + >({ + table: "workspaces", + targetId: workspace.id, + /** + * `BunSQLite.update` writes every key it gets, so a field left out + * here would be set to null. + */ + data: _.omitBy( + { + name: data.name, + description: data.description, + system_prompt: data.system_prompt, + archived: data.archived, + }, + _.isUndefined, + ), + }); + + if (!res.success) { + throw new Error(`Couldn't update the workspace: ${res.msg}`); + } +} diff --git a/src/pages/admin/bots/index.server.ts b/src/pages/admin/bots/index.server.ts index d7f1d81..43804f0 100644 --- a/src/pages/admin/bots/index.server.ts +++ b/src/pages/admin/bots/index.server.ts @@ -1,11 +1,12 @@ import userAuth from "@/src/functions/backend/auth/user-auth"; import ensureCoordinatorBot from "@/src/functions/backend/db/bots/ensure-coordinator-bot"; +import ensureMainComputer from "@/src/functions/backend/db/computers/ensure-main-computer"; import type { PagePropsType } from "@/src/types"; import type { BunextPageServerFn } from "@moduletrace/bunext/types"; /** - * Opens the coordinator's conversation, creating the coordinator first if - * this workspace doesn't have one yet. + * Opens the coordinator's conversation, creating the coordinator and the + * main computer first if this account doesn't have them yet. */ export const server: BunextPageServerFn = async ({ req }) => { const { user } = await userAuth({ req }); @@ -19,6 +20,7 @@ export const server: BunextPageServerFn = async ({ req }) => { } const coordinator_id = await ensureCoordinatorBot({ user_id: user.id }); + await ensureMainComputer({ user_id: user.id, coordinator_id }); return { redirect: { diff --git a/src/pages/api/admin/crud/(tables)/bot_computers/del.ts b/src/pages/api/admin/crud/(tables)/bot_computers/del.ts new file mode 100644 index 0000000..f2e8617 --- /dev/null +++ b/src/pages/api/admin/crud/(tables)/bot_computers/del.ts @@ -0,0 +1,72 @@ +import type { BUN_SQLITE_COBBLE_BOT_COMPUTERS } from "@/db/types/db"; +import type { AdminCrudAPIParams } from "@/src/types"; +import isLockedComputerAccess from "@/src/utils/is-locked-computer-access"; +import BunSQLite from "@moduletrace/bun-sqlite"; +import type { APIResponseObject } from "@moduletrace/bunext/types"; +import _ from "lodash"; + +type AccessRow = BUN_SQLITE_COBBLE_BOT_COMPUTERS & { + bot_kind?: string; + computer_kind?: string; +}; + +/** + * Deletes the matching access rows, skipping the coordinator's access to + * the main computer. + */ +export default async function ( + params: AdminCrudAPIParams, +): Promise { + const { table, body, id, query } = params; + + const records_to_delete = + await BunSQLite.select({ + table, + query: _.merge(query?.sql_query, body?.sql_query) || {}, + targetId: id, + }); + + if (!records_to_delete.payload?.[0]) { + return { + success: false, + msg: `No Records to Delete`, + }; + } + + let skipped = 0; + + for (let i = 0; i < records_to_delete.payload.length; i++) { + const record = records_to_delete.payload[i]; + + if (!record?.id) continue; + + const access = ( + await BunSQLite.sql({ + sql: `SELECT bc.id, b.kind AS bot_kind, c.kind AS computer_kind FROM bot_computers bc JOIN bots b ON b.id = bc.bot_id JOIN computers c ON c.id = bc.computer_id WHERE bc.id = ? LIMIT 1`, + values: [record.id], + }) + ).payload?.[0]; + + if ( + isLockedComputerAccess({ + bot_kind: access?.bot_kind, + computer_kind: access?.computer_kind, + }) + ) { + skipped++; + continue; + } + + await BunSQLite.delete({ + table, + targetId: record.id, + }); + } + + return { + success: true, + msg: skipped + ? `The coordinator always has access to the main computer.` + : undefined, + }; +} diff --git a/src/pages/api/admin/crud/(tables)/bot_computers/index.ts b/src/pages/api/admin/crud/(tables)/bot_computers/index.ts new file mode 100644 index 0000000..52167aa --- /dev/null +++ b/src/pages/api/admin/crud/(tables)/bot_computers/index.ts @@ -0,0 +1,42 @@ +import type { AdminCrudAPIParams } from "@/src/types"; +import type { APIResponseObject } from "@moduletrace/bunext/types"; +import checkUserAccess from "@/src/functions/backend/auth/check-user-access"; +import get from "../../(functions)/get"; +import post from "../../(functions)/post"; +import put from "./put"; +import del from "./del"; + +/** + * Generic CRUD (super admin only), except that the coordinator's access to + * the main computer can't be changed or removed. + */ +export default async function ( + params: AdminCrudAPIParams, +): Promise { + const { req, user_types } = params; + + const is_super_admin = checkUserAccess({ user_types }); + + if (!is_super_admin.success) { + return { + success: false, + msg: `Unauthorized`, + }; + } + + switch (req.method) { + case "GET": + return await get(params); + case "POST": + return await post(params); + case "PUT": + return await put(params); + case "DELETE": + return await del(params); + + default: + return { + success: false, + }; + } +} diff --git a/src/pages/api/admin/crud/(tables)/bot_computers/put.ts b/src/pages/api/admin/crud/(tables)/bot_computers/put.ts new file mode 100644 index 0000000..4e65173 --- /dev/null +++ b/src/pages/api/admin/crud/(tables)/bot_computers/put.ts @@ -0,0 +1,11 @@ +import type { AdminCrudAPIParams } from "@/src/types"; +import type { APIResponseObject } from "@moduletrace/bunext/types"; + +export default async function ( + params: AdminCrudAPIParams, +): Promise { + return { + success: false, + msg: `Can't update computer access. Can only delete and insert.`, + }; +} diff --git a/src/pages/api/admin/crud/[[...paths]].ts b/src/pages/api/admin/crud/[[...paths]].ts index 8e58c85..a359ced 100644 --- a/src/pages/api/admin/crud/[[...paths]].ts +++ b/src/pages/api/admin/crud/[[...paths]].ts @@ -2,6 +2,7 @@ import _ from "lodash"; import users from "./(tables)/users"; import userTypes from "./(tables)/user_types"; import media from "./(tables)/media"; +import botComputers from "./(tables)/bot_computers"; import get from "./(functions)/get"; import put from "./(functions)/put"; import del from "./(functions)/delete"; @@ -83,6 +84,8 @@ export const handler: BunextAPIRouteHandler = async ( return await userTypes(crud_params); case "media": return await media(crud_params); + case "bot_computers": + return await botComputers(crud_params); default: break; diff --git a/src/pages/api/auth/signup.tsx b/src/pages/api/auth/signup.tsx index 4062f2b..603aced 100644 --- a/src/pages/api/auth/signup.tsx +++ b/src/pages/api/auth/signup.tsx @@ -3,6 +3,7 @@ import setUserPassword from "@/src/functions/backend/auth/set-user-password"; import grabHasUsers from "@/src/functions/backend/auth/grab-has-users"; import loginUser from "@/src/functions/backend/auth/login-user"; import ensureCoordinatorBot from "@/src/functions/backend/db/bots/ensure-coordinator-bot"; +import ensureMainComputer from "@/src/functions/backend/db/computers/ensure-main-computer"; import validatePassword from "@/src/utils/validate-password"; import logEvent, { grabRequestIP, logError } from "@/src/functions/backend/logger"; import { AppData } from "@/src/data/app-data"; @@ -74,16 +75,20 @@ export const handler: BunextAPIRouteHandler = async ({ /** * The account exists now, so a failure here mustn't fail signup - * (which is one-shot). `/admin/bots` creates it on the next visit. + * (which is one-shot). `/admin/bots` creates both on the next visit. */ - await ensureCoordinatorBot({ user_id: new_user_id }).catch((error) => - logError({ - title: "Coordinator bot not created at signup", - error, - user_id: new_user_id, - ip_address: grabRequestIP(req), - }).catch(() => {}), - ); + await ensureCoordinatorBot({ user_id: new_user_id }) + .then((coordinator_id) => + ensureMainComputer({ user_id: new_user_id, coordinator_id }), + ) + .catch((error) => + logError({ + title: "Coordinator or main computer not created at signup", + error, + user_id: new_user_id, + ip_address: grabRequestIP(req), + }).catch(() => {}), + ); await logEvent({ title: "Super admin created", diff --git a/src/scripts/seed-bots.ts b/src/scripts/seed-bots.ts index 2928b94..5650eb2 100644 --- a/src/scripts/seed-bots.ts +++ b/src/scripts/seed-bots.ts @@ -17,7 +17,7 @@ import ensureCoordinatorBot from "@/src/functions/backend/db/bots/ensure-coordin /** * Seeds the sample crew from the approved chat design into one user's - * workspace. The coordinator already exists (it's made with the account), + * crew. The coordinator already exists (it's made with the account), * so its sample messages go onto it. Idempotent: skips a user who already * has worker bots. * diff --git a/src/scripts/seed-super-admin.tsx b/src/scripts/seed-super-admin.tsx index b8a5600..424e689 100644 --- a/src/scripts/seed-super-admin.tsx +++ b/src/scripts/seed-super-admin.tsx @@ -7,6 +7,7 @@ import type { import setUserPassword from "@/src/functions/backend/auth/set-user-password"; import validatePassword from "@/src/utils/validate-password"; import ensureCoordinatorBot from "@/src/functions/backend/db/bots/ensure-coordinator-bot"; +import ensureMainComputer from "@/src/functions/backend/db/computers/ensure-main-computer"; const FIRST_NAME = process.env.SEED_SUPER_ADMIN_FIRST_NAME || "Super"; const LAST_NAME = process.env.SEED_SUPER_ADMIN_LAST_NAME || "Admin"; @@ -97,6 +98,12 @@ async function seedSuperAdmin() { const coordinator_id = await ensureCoordinatorBot({ user_id }); console.log(`Coordinator bot is #${coordinator_id}.`); + const main_computer_id = await ensureMainComputer({ + user_id, + coordinator_id, + }); + console.log(`Main computer is #${main_computer_id}.`); + if (has_super_admin) { console.log(`User #${user_id} already has the super_admin type.`); return; diff --git a/src/server.ts b/src/server.ts index 4d530e8..deb9f81 100644 --- a/src/server.ts +++ b/src/server.ts @@ -5,6 +5,8 @@ import { websocket } from "./websocket"; import userAuth from "./functions/backend/auth/user-auth"; import type { WebSocketSession } from "./websocket/types"; import { logError, grabRequestIP } from "./functions/backend/logger"; +import botsQueue from "./bots-queue"; +import enableForeignKeys from "./functions/backend/db/enable-foreign-keys"; declare global {} @@ -13,6 +15,7 @@ const WEBSOCKET_ENDPOINT = "/ws"; const development = process.env.NODE_ENV == "development"; const port = process.env.PORT || 7227; +await enableForeignKeys(); await bunext.bunextInit(); const server = Bun.serve({ @@ -74,4 +77,5 @@ bunext.bunextLog.info( `Server running on http://${server.hostname}:${server.port} ...`, ); -cron(); \ No newline at end of file +cron(); +botsQueue(); \ No newline at end of file diff --git a/src/types/index.ts b/src/types/index.ts index a0278a6..2b54562 100755 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -218,6 +218,15 @@ export const WebSocketEvents = [ * @description Events sent from Server to Client */ "server:pong", + /** + * A new message in one of the user's bot conversations (`data`: the + * `bot_messages` row) + */ + "server:bot-message", + /** + * A bot's face or status line changed (`data`: `{ id, face, status_line }`) + */ + "server:bot-updated", ] as const; export type WebSocketDataType = { diff --git a/src/utils/assert-prompt-length.test.ts b/src/utils/assert-prompt-length.test.ts new file mode 100644 index 0000000..d79b61f --- /dev/null +++ b/src/utils/assert-prompt-length.test.ts @@ -0,0 +1,24 @@ +import { describe, expect, test } from "bun:test"; +import { AppData } from "@/src/data/app-data"; +import assertPromptLength from "./assert-prompt-length"; + +describe("assertPromptLength", () => { + test("allows text up to the cap", () => { + expect(() => + assertPromptLength({ + label: "Memory", + text: "a".repeat(AppData.PromptMaxLength), + }), + ).not.toThrow(); + expect(() => assertPromptLength({ label: "Memory" })).not.toThrow(); + }); + + test("rejects text over the cap", () => { + expect(() => + assertPromptLength({ + label: "Memory", + text: "a".repeat(AppData.PromptMaxLength + 1), + }), + ).toThrow("Memory is too long"); + }); +}); diff --git a/src/utils/assert-prompt-length.ts b/src/utils/assert-prompt-length.ts new file mode 100644 index 0000000..6d596aa --- /dev/null +++ b/src/utils/assert-prompt-length.ts @@ -0,0 +1,18 @@ +import { AppData } from "@/src/data/app-data"; +import AppError from "./app-error"; + +type Params = { + label: string; + text?: string | null; +}; + +/** + * Throws if a prompt or memory is longer than `AppData.PromptMaxLength`. + */ +export default function assertPromptLength({ label, text }: Params) { + if ((text?.length || 0) > AppData.PromptMaxLength) { + throw new AppError( + `${label} is too long (max ${AppData.PromptMaxLength} characters).`, + ); + } +} diff --git a/src/utils/compose-bot-prompt.test.ts b/src/utils/compose-bot-prompt.test.ts new file mode 100644 index 0000000..e91e6e5 --- /dev/null +++ b/src/utils/compose-bot-prompt.test.ts @@ -0,0 +1,38 @@ +import { describe, expect, test } from "bun:test"; +import composeBotPrompt from "./compose-bot-prompt"; + +describe("composeBotPrompt", () => { + test("orders core instructions, workspaces, then memory", () => { + expect( + composeBotPrompt({ + system_prompt: "# You are Scout\n\nFind things.", + memory: "- The user prefers short replies.", + workspaces: [ + { name: "Launch", system_prompt: "Ship by Friday." }, + { name: "Ops", system_prompt: "Servers are on Debian." }, + ], + }), + ).toBe( + [ + "# You are Scout\n\nFind things.", + "# Workspace: Launch\n\nShip by Friday.", + "# Workspace: Ops\n\nServers are on Debian.", + "# Memory\n\n- The user prefers short replies.", + ].join("\n\n"), + ); + }); + + test("leaves out empty parts", () => { + expect( + composeBotPrompt({ + system_prompt: "Core.", + memory: " ", + workspaces: [ + { name: "Empty", system_prompt: "" }, + { name: null, system_prompt: "Shared." }, + ], + }), + ).toBe("Core.\n\n# Workspace: Untitled\n\nShared."); + expect(composeBotPrompt({})).toBe(""); + }); +}); diff --git a/src/utils/compose-bot-prompt.ts b/src/utils/compose-bot-prompt.ts new file mode 100644 index 0000000..cf4ae86 --- /dev/null +++ b/src/utils/compose-bot-prompt.ts @@ -0,0 +1,41 @@ +type WorkspacePrompt = { + name?: string | null; + system_prompt?: string | null; +}; + +type Params = { + system_prompt?: string | null; + memory?: string | null; + workspaces?: WorkspacePrompt[]; +}; + +/** + * The full prompt a bot runs with (markdown): its own core instructions, + * then the shared context of each workspace it's in, then its memory. + * Empty parts are left out. + */ +export default function composeBotPrompt({ + system_prompt, + memory, + workspaces = [], +}: Params) { + const parts: string[] = []; + + if (system_prompt?.trim()) { + parts.push(system_prompt.trim()); + } + + for (const workspace of workspaces) { + if (!workspace.system_prompt?.trim()) continue; + + parts.push( + `# Workspace: ${workspace.name?.trim() || "Untitled"}\n\n${workspace.system_prompt.trim()}`, + ); + } + + if (memory?.trim()) { + parts.push(`# Memory\n\n${memory.trim()}`); + } + + return parts.join("\n\n"); +} diff --git a/src/utils/create-key-lock.test.ts b/src/utils/create-key-lock.test.ts new file mode 100644 index 0000000..12e9d8b --- /dev/null +++ b/src/utils/create-key-lock.test.ts @@ -0,0 +1,68 @@ +import { describe, expect, test } from "bun:test"; +import createKeyLock from "./create-key-lock"; + +const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +describe("createKeyLock", () => { + test("runs tasks with the same key one at a time, in order", async () => { + const lock = createKeyLock(); + const events: string[] = []; + + const job = (name: string, ms: number) => + lock.run({ + key: 1, + task: async () => { + events.push(`${name}:start`); + await wait(ms); + events.push(`${name}:end`); + return name; + }, + }); + + const results = await Promise.all([job("a", 20), job("b", 1)]); + + expect(results).toEqual(["a", "b"]); + expect(events).toEqual(["a:start", "a:end", "b:start", "b:end"]); + expect(lock.size()).toBe(0); + }); + + test("runs different keys side by side", async () => { + const lock = createKeyLock(); + const events: string[] = []; + + await Promise.all([ + lock.run({ + key: "x", + task: async () => { + events.push("x:start"); + await wait(20); + events.push("x:end"); + }, + }), + lock.run({ + key: "y", + task: async () => { + events.push("y:start"); + events.push("y:end"); + }, + }), + ]); + + expect(events.indexOf("y:end")).toBeLessThan(events.indexOf("x:end")); + }); + + test("a failing task doesn't block the next one", async () => { + const lock = createKeyLock(); + + const failed = lock.run({ + key: 1, + task: async () => { + throw new Error("boom"); + }, + }); + const next = lock.run({ key: 1, task: async () => "ok" }); + + expect(failed).rejects.toThrow("boom"); + expect(await next).toBe("ok"); + }); +}); diff --git a/src/utils/create-key-lock.ts b/src/utils/create-key-lock.ts new file mode 100644 index 0000000..073cf98 --- /dev/null +++ b/src/utils/create-key-lock.ts @@ -0,0 +1,32 @@ +type RunParams = { + key: string | number; + task: () => Promise; +}; + +/** + * An in-process lock per key: tasks with the same key run one at a time in + * call order, tasks with different keys run side by side. Used so two jobs + * never drive the same computer at once. + */ +export default function createKeyLock() { + const tails = new Map>(); + + async function run({ key, task }: RunParams): Promise { + const id = String(key); + const previous = tails.get(id) || Promise.resolve(); + const current = previous.catch(() => {}).then(task); + const tail = current.catch(() => {}); + + tails.set(id, tail); + + try { + return await current; + } finally { + if (tails.get(id) === tail) { + tails.delete(id); + } + } + } + + return { run, size: () => tails.size }; +} diff --git a/src/utils/is-locked-computer-access.test.ts b/src/utils/is-locked-computer-access.test.ts new file mode 100644 index 0000000..dced216 --- /dev/null +++ b/src/utils/is-locked-computer-access.test.ts @@ -0,0 +1,32 @@ +import { describe, expect, test } from "bun:test"; +import isLockedComputerAccess from "./is-locked-computer-access"; + +describe("isLockedComputerAccess", () => { + test("locks the coordinator to the main computer", () => { + expect( + isLockedComputerAccess({ + bot_kind: "coordinator", + computer_kind: "main", + }), + ).toBe(true); + }); + + test("leaves every other pair removable", () => { + expect( + isLockedComputerAccess({ bot_kind: "worker", computer_kind: "main" }), + ).toBe(false); + expect( + isLockedComputerAccess({ + bot_kind: "coordinator", + computer_kind: "sandbox", + }), + ).toBe(false); + expect( + isLockedComputerAccess({ + bot_kind: "worker", + computer_kind: "sandbox", + }), + ).toBe(false); + expect(isLockedComputerAccess({})).toBe(false); + }); +}); diff --git a/src/utils/is-locked-computer-access.ts b/src/utils/is-locked-computer-access.ts new file mode 100644 index 0000000..e50ca95 --- /dev/null +++ b/src/utils/is-locked-computer-access.ts @@ -0,0 +1,16 @@ +type Params = { + bot_kind?: string | null; + computer_kind?: string | null; +}; + +/** + * The coordinator always has the main computer. That one access is granted + * with the account and can never be removed; every other bot/computer pair + * can be added and removed freely. + */ +export default function isLockedComputerAccess({ + bot_kind, + computer_kind, +}: Params) { + return bot_kind == "coordinator" && computer_kind == "main"; +} diff --git a/src/websocket/(functions)/send-to-user.ts b/src/websocket/(functions)/send-to-user.ts new file mode 100644 index 0000000..151a31d --- /dev/null +++ b/src/websocket/(functions)/send-to-user.ts @@ -0,0 +1,28 @@ +import type { WebSocketDataType } from "@/src/types"; +import { getAllRooms } from "../(data)/rooms"; + +type Params = { + user_id: number | string; + data: WebSocketDataType; +}; + +/** + * Sends to every open socket of one user. Every socket joins `public` on + * open, so its members are all connected clients. + */ +export function sendToUser({ user_id, data }: Params) { + const members = getAllRooms().get("public"); + + if (!members) return; + + const message = JSON.stringify(data); + + for (const ws of members) { + if ( + String(ws.data.user?.id) == String(user_id) && + ws.readyState === WebSocket.OPEN + ) { + ws.send(message); + } + } +}