Add workspaces, conversation sessions, app-managed computers and pictures

- Rail: Add icon with stacked form modals (bot, workspace, computer),
  coordinator, recent bots with "See all bots", Workspaces section, and
  a workspace mode listing only its crew
- Workspaces: page for context, bots (per-workspace instructions and
  memory), computers (default, link, unlink), models and delete;
  conversations scoped per bot per workspace
- Sessions: conversations split into searchable, reopenable sessions
  with carry-over summaries; sessions and new-session header icons
- Computers: created and provisioned over the docker socket, home dir
  kept at DATA_DIR/computers/<id> after delete; super admins only
- Models: workspace settings win over a bot's own, then the defaults
- Pictures for bots and workspaces; bots editable from the header
- Coordinator tools: update_workspace, set_bot_workspace_instructions

Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
2026-10-19 08:39:33 +01:00
co-authored by Claude Opus 5.5
parent 56daf0c462
commit 0ce61cdb49
223 changed files with 8166 additions and 585 deletions
@@ -0,0 +1,51 @@
import type {
BUN_SQLITE_COBBLE_BOT_SESSIONS,
BunSQLiteTables,
} from "@/db/types/db";
import BunSQLite from "@moduletrace/bun-sqlite";
type Params = {
bot_id: number | string;
workspace_id?: number | string | null;
/**
* The session this one carries on from.
*/
previous_session_id?: number | null;
};
/**
* Starts an active session in a conversation. Callers close the old one
* first (`startBotSession`).
*/
export default async function createBotSession({
bot_id,
workspace_id,
previous_session_id,
}: Params): Promise<BUN_SQLITE_COBBLE_BOT_SESSIONS> {
const now = Date.now();
const row: BUN_SQLITE_COBBLE_BOT_SESSIONS = {
bot_id: Number(bot_id),
workspace_id: workspace_id ? Number(workspace_id) : undefined,
previous_session_id: previous_session_id || undefined,
status: "active",
unread: 0,
last_activity_at: now,
};
const res = await BunSQLite.insert<
BUN_SQLITE_COBBLE_BOT_SESSIONS,
(typeof BunSQLiteTables)[number]
>({
table: "bot_sessions",
data: [row],
});
const id = res.singleRes?.id || res.postInsertReturn?.insertId;
if (!res.success || !id) {
throw new Error(`Couldn't start a session: ${res.msg}`);
}
return { ...row, id, created_at: now, updated_at: now };
}
@@ -0,0 +1,62 @@
import type { BUN_SQLITE_COBBLE_BOT_SESSIONS } from "@/db/types/db";
import BunSQLite from "@moduletrace/bun-sqlite";
import { AppData } from "@/src/data/app-data";
import createBotSession from "./create-bot-session";
import grabActiveBotSession from "./grab-active-bot-session";
import startBotSession from "./start-bot-session";
type Params = {
bot_id: number | string;
workspace_id?: number | string | null;
/**
* Start the next session when the active one is full
* (`AppData.BotSessionMaxMessages`). Set when the user writes, so a
* message never lands in a session that's about to be cut.
*/
rollover?: boolean;
};
/**
* The conversation's active session, started first when there is none
* (carrying on from its latest closed one) or, with `rollover`, when the
* active one is full.
*/
export default async function ensureBotSession({
bot_id,
workspace_id,
rollover,
}: Params): Promise<BUN_SQLITE_COBBLE_BOT_SESSIONS> {
const active = await grabActiveBotSession({ bot_id, workspace_id });
if (active?.id) {
if (!rollover) return active;
const count =
Number(
(
await BunSQLite.sql<{ n: number }>({
sql: `SELECT COUNT(*) AS n FROM bot_messages WHERE session_id = ?`,
values: [active.id],
})
).payload?.[0]?.n,
) || 0;
return count < AppData.BotSessionMaxMessages
? active
: await startBotSession({ bot_id, workspace_id });
}
const latest = (
await BunSQLite.sql<BUN_SQLITE_COBBLE_BOT_SESSIONS>({
sql: `SELECT id FROM bot_sessions WHERE bot_id = ? AND workspace_id IS ?
ORDER BY last_activity_at DESC, id DESC LIMIT 1`,
values: [bot_id, workspace_id ? Number(workspace_id) : null],
})
).payload?.[0];
return await createBotSession({
bot_id,
workspace_id,
previous_session_id: latest?.id || null,
});
}
@@ -0,0 +1,28 @@
import type { BUN_SQLITE_COBBLE_BOT_SESSIONS } from "@/db/types/db";
import BunSQLite from "@moduletrace/bun-sqlite";
type Params = {
bot_id: number | string;
/**
* Null or undefined: the bot's own chat.
*/
workspace_id?: number | string | null;
};
/**
* The active session of one conversation (a bot in its own chat or in one
* workspace), or null when it has none yet.
*/
export default async function grabActiveBotSession({
bot_id,
workspace_id,
}: Params): Promise<BUN_SQLITE_COBBLE_BOT_SESSIONS | null> {
const res = await BunSQLite.sql<BUN_SQLITE_COBBLE_BOT_SESSIONS>({
sql: `SELECT * FROM bot_sessions
WHERE bot_id = ? AND workspace_id IS ? AND status = 'active'
ORDER BY id DESC LIMIT 1`,
values: [bot_id, workspace_id ? Number(workspace_id) : null],
});
return res.payload?.[0] || null;
}
@@ -0,0 +1,59 @@
import BunSQLite from "@moduletrace/bun-sqlite";
import { AppData } from "@/src/data/app-data";
import type { BotSessionListItem } from "@/src/types";
type Params = {
bot_id: number | string;
workspace_id?: number | string | null;
/**
* Matches the session's title, summary or any of its messages.
*/
query?: string;
page?: number;
};
/**
* A conversation's sessions for the sessions modal, latest activity first,
* each with its message count and latest message. Empty sessions are left
* out unless active.
*/
export default async function grabBotSessions({
bot_id,
workspace_id,
query,
page = 1,
}: Params): Promise<BotSessionListItem[]> {
const search = query?.trim();
const like = search ? `%${search.replace(/[\\%_]/g, (c) => `\\${c}`)}%` : null;
const limit = AppData.BotSessionsPageSize;
let sql = ``;
sql += `SELECT s.*,\n`;
sql += ` (SELECT COUNT(*) FROM bot_messages m WHERE m.session_id = s.id) AS message_count,\n`;
sql += ` (SELECT m.text FROM bot_messages m WHERE m.session_id = s.id ORDER BY m.created_at DESC, m.id DESC LIMIT 1) AS last_message_text\n`;
sql += `FROM bot_sessions s\n`;
sql += `WHERE s.bot_id = ? AND s.workspace_id IS ?\n`;
sql += `AND (s.status = 'active' OR EXISTS (SELECT 1 FROM bot_messages m WHERE m.session_id = s.id))\n`;
if (like) {
sql += `AND (s.title LIKE ? ESCAPE '\\' OR s.summary LIKE ? ESCAPE '\\'\n`;
sql += ` OR EXISTS (SELECT 1 FROM bot_messages m WHERE m.session_id = s.id AND m.text LIKE ? ESCAPE '\\'))\n`;
}
sql += `ORDER BY s.status = 'active' DESC, s.last_activity_at DESC, s.id DESC\n`;
sql += `LIMIT ? OFFSET ?\n`;
const res = await BunSQLite.sql<BotSessionListItem>({
sql,
values: [
bot_id,
workspace_id ? Number(workspace_id) : null,
...(like ? [like, like, like] : []),
limit,
(Math.max(1, page) - 1) * limit,
],
});
return res.payload || [];
}
@@ -0,0 +1,89 @@
import type { PagePropsType } from "@/src/types";
import grabBots from "../bots/grab-bots";
import grabActiveBotTask from "../bots/grab-active-bot-task";
import grabBotMessages from "../bots/grab-bot-messages";
import grabBotComputers from "../computers/grab-bot-computers";
import grabBotHarnessOverview from "../harness/grab-bot-harness-overview";
import grabHarnessOverview from "../harness/grab-harness-overview";
import grabUserWorkspaces from "../workspaces/grab-user-workspaces";
import ensureBotSession from "./ensure-bot-session";
import grabUserConversation from "./grab-user-conversation";
import markBotSessionRead from "./mark-bot-session-read";
type Params = {
user_id: number;
bot_id: number | string;
/**
* The workspace the conversation is in; none for the bot's own chat.
*/
workspace_id?: number | string | null;
};
/**
* The page props of a conversation (`/admin/bots/<id>` or
* `/admin/workspaces/<id>/bots/<id>`): the rail's crew (the workspace's
* in a workspace) and workspaces, the bot, its active session (marked
* read) and its messages, its models, computers and active task there.
* Null when the bot isn't the user's or isn't in the workspace.
*/
export default async function grabConversationProps({
user_id,
bot_id,
workspace_id,
}: Params): Promise<PagePropsType | null> {
const conversation = await grabUserConversation({ user_id, bot_id, workspace_id }).catch(
() => null,
);
if (!conversation) return null;
const scope = conversation.workspace_id;
const session = await ensureBotSession({ bot_id: conversation.bot.id, workspace_id: scope });
if (session.unread) {
await markBotSessionRead({ session_id: session.id!, user_id });
session.unread = 0;
}
const [bots, workspaces] = await Promise.all([
grabBots({ user_id, workspace_id: scope }),
grabUserWorkspaces({ user_id }),
]);
const bot = bots.find((b) => b.id == conversation.bot.id);
if (!bot?.id) return null;
bot.unread = 0;
/**
* The header's models modal: a worker's own settings, or for the
* coordinator the defaults it runs on (without the providers). In a
* workspace the workspace's own models come first, so what runs now
* shows its pick.
*/
const [bot_messages, bot_harness_overview, bot_computers, bot_task] = await Promise.all([
grabBotMessages({ session_id: session.id! }),
(bot.kind == "coordinator" && !scope
? grabHarnessOverview().then(({ harnesses }) => ({ harnesses }))
: grabBotHarnessOverview({
bot_id: bot.kind == "coordinator" ? undefined : bot.id,
workspace_id: scope || undefined,
})
).catch(() => null),
grabBotComputers({ bot_id: bot.id, user_id, workspace_id: scope }),
grabActiveBotTask({ bot_id: bot.id, session_id: session.id }),
]);
return {
bots,
bot,
workspaces,
workspace: conversation.workspace,
bot_session: session,
bot_messages,
bot_harness_overview,
bot_computers,
bot_task,
};
}
@@ -0,0 +1,23 @@
import type { BUN_SQLITE_COBBLE_BOT_SESSIONS } from "@/db/types/db";
import BunSQLite from "@moduletrace/bun-sqlite";
type Params = {
session_id: number | string;
user_id: number | string;
};
/**
* A session only if its bot belongs to the given owner.
*/
export default async function grabUserBotSession({
session_id,
user_id,
}: Params): Promise<BUN_SQLITE_COBBLE_BOT_SESSIONS | null> {
const res = await BunSQLite.sql<BUN_SQLITE_COBBLE_BOT_SESSIONS>({
sql: `SELECT s.* FROM bot_sessions s JOIN bots b ON b.id = s.bot_id
WHERE s.id = ? AND b.user_id = ? LIMIT 1`,
values: [session_id, user_id],
});
return res.payload?.[0] || null;
}
@@ -0,0 +1,52 @@
import type { BUN_SQLITE_COBBLE_BOTS, BUN_SQLITE_COBBLE_WORKSPACES } from "@/db/types/db";
import AppError from "@/src/utils/app-error";
import grabUserBot from "../bots/grab-user-bot";
import grabWorkspaceBot from "../workspaces/grab-workspace-bot";
type Params = {
user_id: number | string;
bot_id: number | string;
/**
* Null or undefined: the bot's own chat.
*/
workspace_id?: number | string | null;
};
export type UserConversation = {
bot: BUN_SQLITE_COBBLE_BOTS & { id: number };
workspace: BUN_SQLITE_COBBLE_WORKSPACES | null;
workspace_id: number | null;
};
/**
* One of the user's conversations: their bot, in its own chat or in one of
* their workspaces it works in. Throws when the bot isn't theirs or isn't
* in that workspace.
*/
export default async function grabUserConversation({
user_id,
bot_id,
workspace_id,
}: Params): Promise<UserConversation> {
if (workspace_id) {
const found = await grabWorkspaceBot({ workspace_id, bot_id, user_id });
if (!found?.bot.id || !found.workspace.id) {
throw new AppError(`That bot isn't in this workspace.`);
}
return {
bot: { ...found.bot, id: found.bot.id },
workspace: found.workspace,
workspace_id: found.workspace.id,
};
}
const bot = await grabUserBot({ bot_id, user_id });
if (!bot?.id) {
throw new AppError(`Bot not found`);
}
return { bot: { ...bot, id: bot.id }, workspace: null, workspace_id: null };
}
@@ -0,0 +1,17 @@
import BunSQLite from "@moduletrace/bun-sqlite";
type Params = {
session_id: number | string;
user_id: number | string;
};
/**
* Clears a session's unread flag when its owner has the conversation open.
*/
export default async function markBotSessionRead({ session_id, user_id }: Params) {
await BunSQLite.sql({
sql: `UPDATE bot_sessions SET unread = 0
WHERE id = ? AND unread = 1 AND bot_id IN (SELECT id FROM bots WHERE user_id = ?)`,
values: [session_id, user_id],
});
}
@@ -0,0 +1,49 @@
import type { BUN_SQLITE_COBBLE_BOT_SESSIONS } from "@/db/types/db";
import BunSQLite from "@moduletrace/bun-sqlite";
import AppError from "@/src/utils/app-error";
import grabUserBotSession from "./grab-user-bot-session";
type Params = {
session_id: number | string;
user_id: number | string;
};
/**
* Makes a closed session of one of the user's conversations the active
* one again, so the bot carries on from where it left off. The session
* that was active is closed (an empty one is deleted).
*/
export default async function reopenBotSession({
session_id,
user_id,
}: Params): Promise<BUN_SQLITE_COBBLE_BOT_SESSIONS> {
const session = await grabUserBotSession({ session_id, user_id });
if (!session?.id) {
throw new AppError(`Session not found.`);
}
if (session.status == "active") return session;
const now = Date.now();
const workspace_id = session.workspace_id || null;
await BunSQLite.sql({
sql: `DELETE FROM bot_sessions WHERE bot_id = ? AND workspace_id IS ? AND status = 'active'
AND NOT EXISTS (SELECT 1 FROM bot_messages m WHERE m.session_id = bot_sessions.id)`,
values: [session.bot_id!, workspace_id],
});
await BunSQLite.sql({
sql: `UPDATE bot_sessions SET status = 'closed', unread = 0, updated_at = ?
WHERE bot_id = ? AND workspace_id IS ? AND status = 'active'`,
values: [now, session.bot_id!, workspace_id],
});
await BunSQLite.sql({
sql: `UPDATE bot_sessions SET status = 'active', updated_at = ? WHERE id = ?`,
values: [now, session.id],
});
return { ...session, status: "active", updated_at: now };
}
@@ -0,0 +1,43 @@
import type { BUN_SQLITE_COBBLE_BOT_SESSIONS } from "@/db/types/db";
import BunSQLite from "@moduletrace/bun-sqlite";
import createBotSession from "./create-bot-session";
import grabActiveBotSession from "./grab-active-bot-session";
type Params = {
bot_id: number | string;
workspace_id?: number | string | null;
};
/**
* Closes the conversation's active session and starts the next one, which
* carries on from it (its summary is written when the next reply runs).
* An active session with no messages yet is kept as it is.
*/
export default async function startBotSession({
bot_id,
workspace_id,
}: Params): Promise<BUN_SQLITE_COBBLE_BOT_SESSIONS> {
const active = await grabActiveBotSession({ bot_id, workspace_id });
if (active?.id) {
const has_messages = (
await BunSQLite.sql({
sql: `SELECT id FROM bot_messages WHERE session_id = ? LIMIT 1`,
values: [active.id],
})
).payload?.[0];
if (!has_messages) return active;
await BunSQLite.sql({
sql: `UPDATE bot_sessions SET status = 'closed', unread = 0, updated_at = ? WHERE id = ?`,
values: [Date.now(), active.id],
});
}
return await createBotSession({
bot_id,
workspace_id,
previous_session_id: active?.id || null,
});
}
@@ -5,6 +5,7 @@ import setBotFace from "../bots/set-bot-face";
import countOpenBotJobs from "./count-open-bot-jobs";
import grabLockedBotJob from "./grab-locked-bot-job";
import relayJobToRequester from "./relay-job-to-requester";
import jobSessionId from "./job-session-id";
type Params = {
job_id: number | string;
@@ -71,6 +72,8 @@ export default async function completeBotJob({
? null
: await insertBotMessage({
bot_id: job.bot_id,
session_id: await jobSessionId({ job }),
workspace_id: job.workspace_id || null,
text: result,
needs_you,
});
@@ -8,6 +8,7 @@ import type { BotJobKind } from "@/src/dict/bot-queue-dict";
import AppError from "@/src/utils/app-error";
import grabUserBot from "../bots/grab-user-bot";
import grabBotComputers from "../computers/grab-bot-computers";
import grabUserBotSession from "../bot-sessions/grab-user-bot-session";
type Params = {
/**
@@ -17,6 +18,11 @@ type Params = {
kind?: BotJobKind;
user_id: number | string;
bot_id: number | string;
/**
* The bot's session the result is posted in. The job runs in its
* workspace (context, computers, models), if it has one.
*/
session_id: number | string;
title: string;
/**
* The brief the bot writes for itself (markdown): what to do and how
@@ -34,7 +40,8 @@ type Params = {
*/
task_id?: number | string;
/**
* Must be one of the bot's computers (direct or through a workspace).
* Must be one of the bot's computers where the session is
* (`grabBotComputers`).
*/
computer_id?: number | string;
/**
@@ -46,6 +53,10 @@ type Params = {
* result or failure in its own conversation (`relayJobToRequester`).
*/
requested_by_bot_id?: number | string;
/**
* The requesting bot's session; its report goes to that conversation.
*/
requested_by_session_id?: number | string;
};
/**
@@ -58,6 +69,7 @@ export default async function enqueueBotJob({
kind = "task",
user_id,
bot_id,
session_id,
title,
instructions,
params,
@@ -66,6 +78,7 @@ export default async function enqueueBotJob({
computer_id,
delay,
requested_by_bot_id,
requested_by_session_id,
}: Params) {
const bot = await grabUserBot({ bot_id, user_id });
@@ -73,10 +86,26 @@ export default async function enqueueBotJob({
throw new AppError(`Bot not found.`);
}
const session = await grabUserBotSession({ session_id, user_id });
if (!session?.id || session.bot_id != bot.id) {
throw new AppError(`Session not found.`);
}
const workspace_id = session.workspace_id || null;
if (requested_by_bot_id && !(await grabUserBot({ bot_id: requested_by_bot_id, user_id }))?.id) {
throw new AppError(`Bot not found.`);
}
if (requested_by_session_id) {
const requested_from = await grabUserBotSession({ session_id: requested_by_session_id, user_id });
if (!requested_from?.id || requested_from.bot_id != Number(requested_by_bot_id)) {
throw new AppError(`Session not found.`);
}
}
if (message_id && !(await belongsToBot({ table: "bot_messages", id: message_id, bot_id: bot.id }))) {
throw new AppError(`Message not found.`);
}
@@ -86,7 +115,7 @@ export default async function enqueueBotJob({
}
if (computer_id) {
const computers = await grabBotComputers({ bot_id: bot.id, user_id });
const computers = await grabBotComputers({ bot_id: bot.id, user_id, workspace_id });
if (!computers.some((c) => String(c.id) == String(computer_id))) {
throw new AppError(`This bot has no access to that computer.`);
@@ -96,7 +125,8 @@ export default async function enqueueBotJob({
const final_task_id =
kind == "reply"
? undefined
: task_id || (await createJobTask({ bot_id: bot.id, title }));
: task_id ||
(await createJobTask({ bot_id: bot.id, session_id: session.id, workspace_id, title }));
const res = await BunSQLite.queue.add<
BUN_SQLITE_COBBLE_BOTS_QUEUE,
@@ -107,6 +137,8 @@ export default async function enqueueBotJob({
{
kind,
bot_id: bot.id,
session_id: session.id,
workspace_id: workspace_id || undefined,
title,
instructions,
params: params ? JSON.stringify(params) : undefined,
@@ -114,6 +146,9 @@ export default async function enqueueBotJob({
task_id: final_task_id ? Number(final_task_id) : undefined,
computer_id: computer_id ? Number(computer_id) : undefined,
requested_by_bot_id: requested_by_bot_id ? Number(requested_by_bot_id) : undefined,
requested_by_session_id: requested_by_session_id
? Number(requested_by_session_id)
: undefined,
},
],
delay,
@@ -147,9 +182,13 @@ async function belongsToBot({
async function createJobTask({
bot_id,
session_id,
workspace_id,
title,
}: {
bot_id: number;
session_id: number;
workspace_id: number | null;
title: string;
}) {
const res = await BunSQLite.insert<
@@ -157,7 +196,7 @@ async function createJobTask({
(typeof BunSQLiteTables)[number]
>({
table: "bot_tasks",
data: [{ bot_id, title, status: "active" }],
data: [{ bot_id, session_id, workspace_id: workspace_id || undefined, title, status: "active" }],
});
const task_id = res.singleRes?.id || res.postInsertReturn?.insertId;
@@ -6,6 +6,10 @@ import { BotReplyJobTitle } from "@/src/dict/bot-queue-dict";
type Params = {
user_id: number | string;
bot_id: number | string;
/**
* The session the user wrote in.
*/
session_id: number | string;
/**
* The user message to answer.
*/
@@ -13,21 +17,23 @@ type Params = {
};
/**
* Queues the bot's chat reply to the user's latest message(s). A reply
* reads the whole recent conversation when it runs, so a burst of
* messages shares one reply: while one is still pending, it is reused.
* Queues the bot's chat reply to the user's latest message(s) in a
* session. A reply reads the session's recent messages when it runs, so a
* burst of messages shares one reply: while one is still pending in the
* session, it is reused.
*/
export default async function enqueueBotReply({
user_id,
bot_id,
session_id,
message_id,
}: Params) {
const pending = (
await BunSQLite.sql<BUN_SQLITE_COBBLE_BOTS_QUEUE>({
sql: `SELECT id FROM bots_queue
WHERE bot_id = ? AND kind = 'reply' AND status = 'pending'
WHERE bot_id = ? AND session_id = ? AND kind = 'reply' AND status = 'pending'
ORDER BY id DESC LIMIT 1`,
values: [bot_id],
values: [bot_id, session_id],
})
).payload?.[0];
@@ -44,6 +50,7 @@ export default async function enqueueBotReply({
kind: "reply",
user_id,
bot_id,
session_id,
title: BotReplyJobTitle,
instructions: "Reply to the user's latest messages.",
message_id,
@@ -0,0 +1,21 @@
import type { BUN_SQLITE_COBBLE_BOTS_QUEUE } from "@/db/types/db";
import ensureBotSession from "../bot-sessions/ensure-bot-session";
type Params = {
job: Pick<BUN_SQLITE_COBBLE_BOTS_QUEUE, "bot_id" | "session_id" | "workspace_id">;
};
/**
* The session a job's result goes in: its own, or (for a job queued
* without one) the active session of its bot's conversation there.
*/
export default async function jobSessionId({ job }: Params) {
if (job.session_id) return Number(job.session_id);
const session = await ensureBotSession({
bot_id: job.bot_id!,
workspace_id: job.workspace_id || null,
});
return Number(session.id);
}
@@ -3,6 +3,7 @@ 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";
import relayJobToRequester from "./relay-job-to-requester";
import jobSessionId from "./job-session-id";
type Params = {
job: BUN_SQLITE_COBBLE_BOTS_QUEUE;
@@ -30,6 +31,8 @@ export default async function notifyBotJobFailed({
const message = await insertBotMessage({
bot_id: job.bot_id,
session_id: await jobSessionId({ job }),
workspace_id: job.workspace_id || null,
text:
failure_message ||
[
@@ -1,6 +1,11 @@
import BunSQLite from "@moduletrace/bun-sqlite";
import type { BUN_SQLITE_COBBLE_BOTS, BUN_SQLITE_COBBLE_BOTS_QUEUE } from "@/db/types/db";
import type {
BUN_SQLITE_COBBLE_BOT_SESSIONS,
BUN_SQLITE_COBBLE_BOTS,
BUN_SQLITE_COBBLE_BOTS_QUEUE,
} from "@/db/types/db";
import insertBotMessage from "../bots/insert-bot-message";
import ensureBotSession from "../bot-sessions/ensure-bot-session";
type Params = {
job: BUN_SQLITE_COBBLE_BOTS_QUEUE;
@@ -13,9 +18,11 @@ type Params = {
/**
* Reports a handed-over job back to the bot that handed it over
* (`requested_by_bot_id`, the coordinator): a message in its conversation
* with the worker's report, flagged needs-you when the job failed. Returns
* the message for the live push, or null when nobody handed the job over.
* (`requested_by_bot_id`, the coordinator): a message in the conversation
* it was handed over from (the active session there, which may have moved
* on since), with the worker's report, flagged needs-you when the job
* failed. Returns the message for the live push, or null when nobody
* handed the job over.
*/
export default async function relayJobToRequester({ job, text, failed }: Params) {
if (!job.requested_by_bot_id || !job.bot_id || job.requested_by_bot_id == job.bot_id) {
@@ -38,11 +45,25 @@ export default async function relayJobToRequester({ job, text, failed }: Params)
if (!requester?.id) return null;
const requested_from = job.requested_by_session_id
? (
await BunSQLite.sql<BUN_SQLITE_COBBLE_BOT_SESSIONS>({
sql: `SELECT workspace_id FROM bot_sessions WHERE id = ? AND bot_id = ? LIMIT 1`,
values: [job.requested_by_session_id, requester.id],
})
).payload?.[0]
: null;
const workspace_id = requested_from ? requested_from.workspace_id || null : job.workspace_id || null;
const session = await ensureBotSession({ bot_id: requester.id, workspace_id });
const name = worker?.name || "A crew member";
const title = job.title || "the job";
const message = await insertBotMessage({
bot_id: requester.id,
session_id: Number(session.id),
workspace_id,
text: failed
? `${name} couldn't finish **${title}**. ${text}`.trim()
: `${name} finished **${title}**. Their report:\n\n${text}`,
@@ -0,0 +1,51 @@
import BunSQLite from "@moduletrace/bun-sqlite";
import type { BUN_SQLITE_COBBLE_BOTS, BunSQLiteTables } from "@/db/types/db";
import type { BotHue } from "@/src/dict/bot-dict";
import assertPromptLength from "@/src/utils/assert-prompt-length";
type Params = {
user_id: number | string;
name: string;
role?: string;
hue: BotHue;
/**
* Its core instructions (markdown).
*/
system_prompt: string;
};
/**
* Adds a worker bot to the user's crew and returns its id. It has no
* computer of its own until it's given one; in a workspace it uses the
* workspace's.
*/
export default async function createBot({ user_id, name, role, hue, system_prompt }: Params) {
assertPromptLength({ label: "Instructions", text: system_prompt });
const res = await BunSQLite.insert<
BUN_SQLITE_COBBLE_BOTS,
(typeof BunSQLiteTables)[number]
>({
table: "bots",
data: [
{
user_id: Number(user_id),
name,
kind: "worker",
hue,
role: role || undefined,
face: "idle",
status_line: "",
system_prompt,
},
],
});
const bot_id = res.singleRes?.id || res.postInsertReturn?.insertId;
if (!res.success || !bot_id) {
throw new Error(`Couldn't add the bot: ${res.msg}`);
}
return Number(bot_id);
}
@@ -55,8 +55,6 @@ export default async function ensureCoordinatorBot({ user_id }: Params) {
face: "idle",
status_line: "",
system_prompt: CoordinatorSystemPrompt,
unread: 0,
last_activity_at: Date.now(),
},
],
});
@@ -8,18 +8,25 @@ import grabBotJobUpdates from "../bots-queue/grab-bot-job-updates";
type Params = {
bot_id: number | string;
/**
* Only a task started from this session.
*/
session_id?: number | string | null;
};
/**
* The bot's most recent active task, its steps and what its job has done
* so far, for the working-on panel.
* The bot's most recent active task (in one session, when given), its
* steps and what its job has done so far, for the working-on panel.
*/
export default async function grabActiveBotTask({
bot_id,
session_id,
}: Params): Promise<BotTaskWithSteps | null> {
const task_res = await BunSQLite.sql<BUN_SQLITE_COBBLE_BOT_TASKS>({
sql: `SELECT * FROM bot_tasks WHERE bot_id = ? AND status = 'active' ORDER BY created_at DESC, id DESC LIMIT 1`,
values: [bot_id],
sql: `SELECT * FROM bot_tasks WHERE bot_id = ? AND status = 'active'
${session_id ? `AND session_id = ?` : ""}
ORDER BY created_at DESC, id DESC LIMIT 1`,
values: session_id ? [bot_id, session_id] : [bot_id],
});
const task = task_res.singleRes || task_res.payload?.[0];
@@ -5,30 +5,29 @@ import grabMessagesMedia from "./grab-messages-media";
import mediaToAttachment from "./media-to-attachment";
type Params = {
bot_id: number | string;
session_id: number | string;
limit?: number;
};
/**
* The latest messages of a bot's conversation, oldest first, with their
* attachments.
* The latest messages of a session, oldest first, with their attachments.
*/
export default async function grabBotMessages({
bot_id,
session_id,
limit = 100,
}: Params): Promise<BotMessageWithAttachments[]> {
let sql = ``;
sql += `SELECT * FROM (\n`;
sql += ` SELECT * FROM bot_messages\n`;
sql += ` WHERE bot_id = ?\n`;
sql += ` WHERE session_id = ?\n`;
sql += ` ORDER BY created_at DESC, id DESC\n`;
sql += ` LIMIT ?\n`;
sql += `) ORDER BY created_at ASC, id ASC\n`;
const res = await BunSQLite.sql<BUN_SQLITE_COBBLE_BOT_MESSAGES>({
sql,
values: [bot_id, limit],
values: [session_id, limit],
});
const messages = res.payload || [];
@@ -1,39 +1,42 @@
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 grabWorkspaceBot from "../workspaces/grab-workspace-bot";
import grabUserBot from "./grab-user-bot";
type Params = {
bot_id: number | string;
user_id: number | string;
/**
* The workspace the bot is working in. Its context and the bot's
* instructions and memory there join the prompt.
*/
workspace_id?: number | string | null;
};
/**
* 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.
* The full prompt for one of the user's bots (`composeBotPrompt`): its
* core instructions and memory, and in a workspace that workspace's
* context and what the bot has for it.
*/
export default async function grabBotPrompt({ bot_id, user_id }: Params) {
export default async function grabBotPrompt({ bot_id, user_id, workspace_id }: Params) {
const bot = await grabUserBot({ bot_id, user_id });
if (!bot?.id) {
throw new AppError(`Bot not found.`);
}
const workspaces =
(
await BunSQLite.sql<BUN_SQLITE_COBBLE_WORKSPACES>({
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 || [];
const found = workspace_id
? await grabWorkspaceBot({ workspace_id, bot_id: bot.id, user_id })
: null;
if (workspace_id && !found) {
throw new AppError(`${bot.name || "This bot"} isn't in that workspace.`);
}
return composeBotPrompt({
system_prompt: bot.system_prompt,
memory: bot.memory,
workspaces,
workspace: found?.workspace,
workspace_bot: found?.membership,
});
}
+25 -4
View File
@@ -3,36 +3,57 @@ import BunSQLite from "@moduletrace/bun-sqlite";
type Params = {
user_id: number | string;
/**
* A workspace's crew instead: the coordinator and the bots in it, with
* their conversations in that workspace.
*/
workspace_id?: number | string | null;
};
/**
* A user's crew for the rail: the coordinator pinned first, then
* workers by latest activity. Each row carries its latest message.
* workers by latest activity. Each row carries its conversation's unread
* flag, latest activity and latest message: in its own chat, or in the
* workspace when one is given.
*/
export default async function grabBots({
user_id,
workspace_id,
}: Params): Promise<BotWithLastMessage[]> {
const scope = workspace_id ? Number(workspace_id) : null;
let sql = ``;
sql += `SELECT\n`;
sql += ` b.*,\n`;
sql += ` COALESCE(t.unread, 0) AS unread,\n`;
sql += ` t.last_activity_at AS last_activity_at,\n`;
sql += ` lm.text AS last_message_text,\n`;
sql += ` lm.created_at AS last_message_at\n`;
sql += `FROM bots b\n`;
sql += `LEFT JOIN (\n`;
sql += ` SELECT bot_id, MAX(unread) AS unread, MAX(last_activity_at) AS last_activity_at\n`;
sql += ` FROM bot_sessions WHERE workspace_id IS ? GROUP BY bot_id\n`;
sql += `) t ON t.bot_id = b.id\n`;
sql += `LEFT JOIN bot_messages lm ON lm.id = (\n`;
sql += ` SELECT id FROM bot_messages\n`;
sql += ` WHERE bot_id = b.id\n`;
sql += ` WHERE bot_id = b.id AND workspace_id IS ?\n`;
sql += ` ORDER BY created_at DESC, id DESC\n`;
sql += ` LIMIT 1\n`;
sql += `)\n`;
sql += `WHERE b.user_id = ?\n`;
if (scope) {
sql += `AND (b.kind = 'coordinator' OR b.id IN (SELECT bot_id FROM workspace_bots WHERE workspace_id = ?))\n`;
}
sql += `ORDER BY\n`;
sql += ` CASE WHEN b.kind = 'coordinator' THEN 0 ELSE 1 END,\n`;
sql += ` COALESCE(lm.created_at, b.last_activity_at, b.created_at) DESC\n`;
sql += ` COALESCE(lm.created_at, t.last_activity_at, b.created_at) DESC\n`;
const res = await BunSQLite.sql<BotWithLastMessage>({
sql,
values: [user_id],
values: [scope, scope, user_id, ...(scope ? [scope] : [])],
});
return res.payload || [];
@@ -6,16 +6,23 @@ import BunSQLite from "@moduletrace/bun-sqlite";
type Params = {
bot_id: number;
/**
* The session it goes in, and that session's workspace.
*/
session_id: number;
workspace_id?: number | null;
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.
* Appends a bot message to a session, marks the session unread and bumps
* its activity. Callers must check the bot and session first.
*/
export default async function insertBotMessage({
bot_id,
session_id,
workspace_id,
text,
needs_you,
}: Params): Promise<BUN_SQLITE_COBBLE_BOT_MESSAGES> {
@@ -27,7 +34,16 @@ export default async function insertBotMessage({
(typeof BunSQLiteTables)[number]
>({
table: "bot_messages",
data: [{ bot_id, author: "bot", text, needs_you: needs_you_flag }],
data: [
{
bot_id,
session_id,
workspace_id: workspace_id || undefined,
author: "bot",
text,
needs_you: needs_you_flag,
},
],
});
const id = res.postInsertReturn?.insertId;
@@ -37,13 +53,15 @@ export default async function insertBotMessage({
}
await BunSQLite.sql({
sql: `UPDATE bots SET last_activity_at = ?, unread = 1 WHERE id = ?`,
values: [now, bot_id],
sql: `UPDATE bot_sessions SET last_activity_at = ?, unread = 1 WHERE id = ?`,
values: [now, session_id],
});
return {
id,
bot_id,
session_id,
workspace_id: workspace_id || undefined,
author: "bot",
text,
needs_you: needs_you_flag,
@@ -9,6 +9,11 @@ import mediaToAttachment from "./media-to-attachment";
type Params = {
bot_id: number;
/**
* The session it goes in, and that session's workspace.
*/
session_id: number;
workspace_id?: number | null;
text: string;
/**
* Files to attach, already checked to be the user's own
@@ -18,12 +23,14 @@ type Params = {
};
/**
* Appends a user message (and its files) to a bot's conversation and
* bumps its activity. Callers must check the bot belongs to the session
* user first.
* Appends a user message (and its files) to a session and bumps its
* activity. The first one titles the session. Callers must check the bot
* and session belong to the session user first.
*/
export default async function insertUserMessage({
bot_id,
session_id,
workspace_id,
text,
media = [],
}: Params): Promise<BotMessageWithAttachments> {
@@ -34,7 +41,16 @@ export default async function insertUserMessage({
(typeof BunSQLiteTables)[number]
>({
table: "bot_messages",
data: [{ bot_id, author: "user", text, needs_you: 0 }],
data: [
{
bot_id,
session_id,
workspace_id: workspace_id || undefined,
author: "user",
text,
needs_you: 0,
},
],
});
const id = res.postInsertReturn?.insertId;
@@ -50,14 +66,20 @@ export default async function insertUserMessage({
});
}
const title = (text || media[0]?.title || "").replace(/\s+/g, " ").trim().slice(0, 80);
await BunSQLite.sql({
sql: `UPDATE bots SET last_activity_at = ? WHERE id = ?`,
values: [now, bot_id],
sql: `UPDATE bot_sessions SET last_activity_at = ?,
title = CASE WHEN COALESCE(title, '') = '' THEN ? ELSE title END
WHERE id = ?`,
values: [now, title, session_id],
});
return {
id,
bot_id,
session_id,
workspace_id: workspace_id || undefined,
author: "user",
text,
needs_you: 0,
@@ -1,16 +0,0 @@
import BunSQLite from "@moduletrace/bun-sqlite";
type Params = {
bot_id: number | string;
user_id: number | string;
};
/**
* Clears a bot's unread flag when its owner opens the conversation.
*/
export default async function markBotRead({ bot_id, user_id }: Params) {
await BunSQLite.sql({
sql: `UPDATE bots SET unread = 0 WHERE id = ? AND user_id = ? AND unread = 1`,
values: [bot_id, user_id],
});
}
@@ -0,0 +1,43 @@
import BunSQLite from "@moduletrace/bun-sqlite";
import type { BotHue } from "@/src/dict/bot-dict";
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;
name: string;
role?: string;
hue: BotHue;
system_prompt: string;
};
/**
* Saves one of the user's bots: name, job, hue and core instructions. The
* coordinator keeps its name and stays hue-less, so only its instructions
* change. Returns the saved bot.
*/
export default async function updateBot({ user_id, bot_id, name, role, hue, system_prompt }: Params) {
assertPromptLength({ label: "Instructions", text: system_prompt });
const bot = await grabUserBot({ bot_id, user_id });
if (!bot?.id) {
throw new AppError(`Bot not found.`);
}
if (bot.kind == "coordinator") {
await BunSQLite.sql({
sql: `UPDATE bots SET system_prompt = ?, updated_at = ? WHERE id = ?`,
values: [system_prompt, Date.now(), bot.id],
});
} else {
await BunSQLite.sql({
sql: `UPDATE bots SET name = ?, role = ?, hue = ?, system_prompt = ?, updated_at = ? WHERE id = ?`,
values: [name, role || null, hue, system_prompt, Date.now(), bot.id],
});
}
return (await grabUserBot({ bot_id, user_id }))!;
}
@@ -0,0 +1,96 @@
import BunSQLite from "@moduletrace/bun-sqlite";
import type {
BUN_SQLITE_COBBLE_COMPUTERS,
BUN_SQLITE_COBBLE_CONTAINERS,
BunSQLiteTables,
} from "@/db/types/db";
import { AppData } from "@/src/data/app-data";
import type { ComputerKind } from "@/src/dict/computer-dict";
import grabEnv from "../envs/grab-env";
import provisionComputer from "./provision-computer";
type Params = {
user_id: number | string;
name: string;
description?: string;
/**
* `main` only for the owner's one main computer (`ensureMainComputer`).
*/
kind?: ComputerKind;
};
/**
* Records a new computer for the user and its container
* (`<AppData.ComputerContainerPrefix>-<id>`), then sets the container up
* in the background (`provisionComputer`). Returns the computer at once,
* still `creating`, and the setup to await when the caller must (a CLI
* script). The main computer of a server that reaches it at
* `MAIN_COMPUTER_URL` (a dev server) gets no container of its own.
*/
export default async function createComputer({
user_id,
name,
description,
kind = "sandbox",
}: Params) {
const external = kind == "main" && Boolean((await grabEnv({ key: "MAIN_COMPUTER_URL" }))?.trim());
const container_res = await BunSQLite.insert<
BUN_SQLITE_COBBLE_CONTAINERS,
(typeof BunSQLiteTables)[number]
>({
table: "containers",
data: [
{
name: `${AppData.ComputerContainerPrefix}-pending-${Date.now()}`,
purpose: "computer",
network: AppData.DockerNetworkName,
port: AppData.ComputerPort,
image: AppData.ComputerImage,
status: external ? "unknown" : "creating",
},
],
});
const container_id = container_res.singleRes?.id || container_res.postInsertReturn?.insertId;
if (!container_res.success || !container_id) {
throw new Error(`Couldn't record the computer's container: ${container_res.msg}`);
}
const computer_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,
description: description || undefined,
kind,
},
],
});
const computer_id = computer_res.singleRes?.id || computer_res.postInsertReturn?.insertId;
if (!computer_res.success || !computer_id) {
await BunSQLite.sql({ sql: `DELETE FROM containers WHERE id = ?`, values: [container_id] });
throw new Error(`Couldn't create the computer: ${computer_res.msg}`);
}
const container_name = `${AppData.ComputerContainerPrefix}-${computer_id}`;
await BunSQLite.sql({
sql: `UPDATE containers SET name = ?, hostname = ? WHERE id = ?`,
values: [container_name, container_name, container_id],
});
const provisioning = external
? Promise.resolve()
: provisionComputer({ computer_id: Number(computer_id), fresh: true });
return { computer_id: Number(computer_id), provisioning };
}
@@ -0,0 +1,57 @@
import BunSQLite from "@moduletrace/bun-sqlite";
import type { BUN_SQLITE_COBBLE_CONTAINERS } from "@/db/types/db";
import removeDockerContainer from "@/src/functions/backend/docker/remove-docker-container";
import AppError from "@/src/utils/app-error";
import grabUserComputer from "./grab-user-computer";
type Params = {
user_id: number | string;
computer_id: number | string;
};
/**
* Deletes one of the user's computers: removes its container and its
* records (its access grants and workspace links go with them). Its home
* directory is kept, so what it left stays on disk. Workspaces it was the
* default of fall back to their oldest computer still linked. The main
* computer can't be deleted.
*/
export default async function deleteComputer({ user_id, computer_id }: Params) {
const computer = await grabUserComputer({ computer_id, user_id });
if (!computer?.id) {
throw new AppError(`Computer not found.`);
}
if (computer.kind == "main") {
throw new AppError(`The main computer can't be deleted.`);
}
const container = computer.container_id
? (
await BunSQLite.sql<BUN_SQLITE_COBBLE_CONTAINERS>({
sql: `SELECT * FROM containers WHERE id = ? LIMIT 1`,
values: [computer.container_id],
})
).payload?.[0]
: null;
if (container?.docker_id || container?.name) {
await removeDockerContainer({ container: container.docker_id || container.name! });
}
await BunSQLite.sql({ sql: `DELETE FROM computers WHERE id = ?`, values: [computer.id] });
if (container?.id) {
await BunSQLite.sql({ sql: `DELETE FROM containers WHERE id = ?`, values: [container.id] });
}
await BunSQLite.sql({
sql: `UPDATE workspaces SET updated_at = ?, computer_id = (
SELECT wc.computer_id FROM workspace_computers wc WHERE wc.workspace_id = workspaces.id ORDER BY wc.id ASC LIMIT 1
) WHERE computer_id = ?`,
values: [Date.now(), computer.id],
});
return computer;
}
@@ -1,60 +0,0 @@
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;
}
@@ -1,10 +1,13 @@
import BunSQLite from "@moduletrace/bun-sqlite";
import type {
BUN_SQLITE_COBBLE_COMPUTERS,
BUN_SQLITE_COBBLE_CONTAINERS,
BunSQLiteTables,
} from "@/db/types/db";
import { MainComputerName } from "@/src/dict/computer-dict";
import ensureMainComputerContainer from "./ensure-main-computer-container";
import grabEnv from "../envs/grab-env";
import createComputer from "./create-computer";
import provisionComputer from "./provision-computer";
type Params = {
user_id: number | string;
@@ -13,30 +16,26 @@ type Params = {
* access to the main computer.
*/
coordinator_id: number | string;
/**
* Wait for a container being set up (a CLI script, which would
* otherwise exit first).
*/
wait?: boolean;
};
/**
* 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.
* Returns the id of the user's main computer, creating it first if
* missing (`createComputer`, which sets its container up in the
* background), and makes sure the coordinator has access to it. A main
* computer recorded before the app managed containers (no docker id, no
* `MAIN_COMPUTER_URL`) gets its container set up now. Safe to call on
* every visit.
*/
export default async function ensureMainComputer({
user_id,
coordinator_id,
wait,
}: 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<Params, "user_id">) {
const existing = (
await BunSQLite.select<
BUN_SQLITE_COBBLE_COMPUTERS,
@@ -54,32 +53,39 @@ async function grabOrCreateMainComputer({
})
).payload?.[0];
if (existing?.id) {
return existing.id;
let computer_id = existing?.id ? Number(existing.id) : 0;
let provisioning: Promise<void> = Promise.resolve();
if (!computer_id) {
const created = await createComputer({ user_id, name: MainComputerName, kind: "main" });
computer_id = created.computer_id;
provisioning = created.provisioning;
} else if (await needsContainer({ container_id: existing?.container_id })) {
provisioning = provisionComputer({ computer_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",
},
],
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()],
});
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}`);
}
if (wait) await provisioning;
return computer_id;
}
async function needsContainer({ container_id }: { container_id?: number | "" }) {
if ((await grabEnv({ key: "MAIN_COMPUTER_URL" }))?.trim()) return false;
const container = container_id
? (
await BunSQLite.sql<BUN_SQLITE_COBBLE_CONTAINERS>({
sql: `SELECT docker_id, status FROM containers WHERE id = ? LIMIT 1`,
values: [container_id],
})
).payload?.[0]
: null;
return Boolean(container && !container.docker_id && container.status == "unknown");
}
@@ -4,36 +4,53 @@ import BunSQLite from "@moduletrace/bun-sqlite";
type Params = {
bot_id: number | string;
user_id: number | string;
/**
* The workspace the bot is working in: then its computers are the
* workspace's, the default first, and nothing else.
*/
workspace_id?: number | string | null;
};
/**
* 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`); the coordinator can use all of them. Only the
* owner's computers, each once. The
* one it works on now (`bots.computer_id`) comes first, else its main one,
* so `computers[0]` is where its tasks run by default.
* Every computer a bot can use where it's working. In a workspace, the
* workspace's computers (none linked: none at all), its default first. In
* its own chat, the ones it was given directly (`bot_computers`); the
* coordinator can use all of the owner's. There the one it works on now
* (`bots.computer_id`) comes first, else its main one, so `computers[0]`
* is where its tasks run by default.
*/
export default async function grabBotComputers({
bot_id,
user_id,
workspace_id,
}: Params): Promise<BUN_SQLITE_COBBLE_COMPUTERS[]> {
if (workspace_id) {
const res = await BunSQLite.sql<BUN_SQLITE_COBBLE_COMPUTERS>({
sql: `SELECT c.* FROM computers c
JOIN workspace_computers wc ON wc.computer_id = c.id AND wc.workspace_id = ?
JOIN workspaces w ON w.id = wc.workspace_id AND w.user_id = c.user_id AND w.archived = 0
WHERE c.user_id = ?
AND (
(SELECT kind FROM bots WHERE id = ? AND user_id = c.user_id) = 'coordinator'
OR ? IN (SELECT bot_id FROM workspace_bots WHERE workspace_id = w.id)
)
ORDER BY c.id = w.computer_id DESC, wc.id ASC`,
values: [workspace_id, user_id, bot_id, Number(bot_id)],
});
return res.payload || [];
}
const res = await BunSQLite.sql<BUN_SQLITE_COBBLE_COMPUTERS>({
sql: `SELECT c.* FROM computers c
WHERE c.user_id = ?
AND (
(SELECT kind FROM bots WHERE id = ? AND user_id = c.user_id) = 'coordinator'
OR 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.id = (SELECT computer_id FROM bots WHERE id = ?) DESC,
c.kind = 'main' DESC, c.id ASC`,
values: [user_id, bot_id, bot_id, bot_id, bot_id],
values: [user_id, bot_id, bot_id, bot_id],
});
return res.payload || [];
@@ -0,0 +1,17 @@
import path from "path";
import { AppData } from "@/src/data/app-data";
import grabDirNames from "@/src/utils/grab-dir-names";
type Params = {
computer_id: number | string;
};
/**
* Where a computer's home directory is kept, as this app sees it:
* `DATA_DIR/<AppData.ComputerDataDir>/<id>`.
*/
export default function grabComputerDataDir({ computer_id }: Params) {
const { DATA_DIR } = grabDirNames();
return path.join(DATA_DIR, AppData.ComputerDataDir, String(computer_id));
}
@@ -0,0 +1,44 @@
import BunSQLite from "@moduletrace/bun-sqlite";
import parseJobJson from "@/src/bots-queue/parse-job-json";
import type { ComputerWithStatus } from "@/src/types";
type Params = {
/**
* Every computer of this owner, main first.
*/
user_id?: number | string;
/**
* Or just this one.
*/
computer_id?: number | string;
};
/**
* Computers with their container's state, and why setting one up failed.
*/
export default async function grabUserComputers({
user_id,
computer_id,
}: Params): Promise<ComputerWithStatus[]> {
if (!user_id && !computer_id) return [];
const rows =
(
await BunSQLite.sql<ComputerWithStatus & { metadata?: string | null }>({
sql: `SELECT c.*, co.status AS status, co.metadata AS metadata FROM computers c
LEFT JOIN containers co ON co.id = c.container_id
WHERE ${computer_id ? `c.id = ?` : `c.user_id = ?`}
ORDER BY c.kind = 'main' DESC, c.id ASC`,
values: [computer_id || user_id!],
})
).payload || [];
return rows.map(({ metadata, ...computer }) => {
const error = parseJobJson({ text: metadata })?.error;
return {
...computer,
error: computer.status == "failed" && typeof error == "string" ? error : null,
};
});
}
@@ -0,0 +1,40 @@
import fs from "fs/promises";
import { AppData } from "@/src/data/app-data";
import grabComputerDataDir from "./grab-computer-data-dir";
type Params = {
computer_id: number | string;
/**
* A new computer: files already at its path (left by a deleted
* computer with the same id, e.g. after a restore) are moved aside to
* `<id>-<timestamp>` and kept, so it starts empty.
*/
fresh?: boolean;
};
/**
* Creates a computer's home directory, owned by the sandbox's user so it
* can write there, and returns its path. Docker would otherwise create a
* missing mount source owned by root.
*/
export default async function prepareComputerDataDir({ computer_id, fresh }: Params) {
const dir = grabComputerDataDir({ computer_id });
if (fresh) {
const existing = await fs.readdir(dir).catch(() => null);
if (existing?.length) {
await fs.rename(dir, `${dir}-${Date.now()}`);
}
}
await fs.mkdir(dir, { recursive: true });
/**
* Only root can hand a directory to another user; a dev server
* running as that user already owns it.
*/
await fs.chown(dir, AppData.ComputerUserUid, AppData.ComputerUserGid).catch(() => {});
return dir;
}
@@ -0,0 +1,81 @@
import BunSQLite from "@moduletrace/bun-sqlite";
import type { BUN_SQLITE_COBBLE_CONTAINERS } from "@/db/types/db";
import parseJobJson from "@/src/bots-queue/parse-job-json";
import { AppData } from "@/src/data/app-data";
import createComputerContainer from "@/src/functions/backend/docker/create-computer-container";
import { logError } from "@/src/functions/backend/logger";
import prepareComputerDataDir from "./prepare-computer-data-dir";
import pushComputerUpdate from "./push-computer-update";
type Params = {
computer_id: number;
/**
* A brand new computer: start from an empty home directory.
*/
fresh?: boolean;
};
/**
* Sets up a computer's container: prepares its home directory, pulls the
* sandbox image and starts the container, then records it (`containers`:
* docker id, state, and where to reach it outside docker). A failure is
* recorded as `failed` with its reason, which the computer lists show. The
* owner's tabs hear about it either way. Never throws.
*/
export default async function provisionComputer({ computer_id, fresh }: Params) {
const row = (
await BunSQLite.sql<BUN_SQLITE_COBBLE_CONTAINERS>({
sql: `SELECT co.* FROM containers co JOIN computers c ON c.container_id = co.id WHERE c.id = ? LIMIT 1`,
values: [computer_id],
})
).payload?.[0];
if (!row?.id || !row.name) return;
const metadata = parseJobJson({ text: row.metadata }) || {};
try {
await BunSQLite.sql({
sql: `UPDATE containers SET status = 'creating', updated_at = ? WHERE id = ?`,
values: [Date.now(), row.id],
});
const data_dir = await prepareComputerDataDir({ computer_id, fresh });
const created = await createComputerContainer({ computer_id, name: row.name, data_dir });
const { error: _error, mcp_url: _mcp_url, ...kept } = metadata;
await BunSQLite.sql({
sql: `UPDATE containers SET docker_id = ?, image = ?, network = ?, hostname = ?, port = ?,
status = ?, metadata = ?, last_seen_at = ?, updated_at = ? WHERE id = ?`,
values: [
created.docker_id,
AppData.ComputerImage,
created.network,
row.name,
AppData.ComputerPort,
created.status,
JSON.stringify({
...kept,
data_dir,
host_dir: created.host_dir,
...(created.mcp_url ? { mcp_url: created.mcp_url } : {}),
}),
Date.now(),
Date.now(),
row.id,
],
});
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
await BunSQLite.sql({
sql: `UPDATE containers SET status = 'failed', metadata = ?, updated_at = ? WHERE id = ?`,
values: [JSON.stringify({ ...metadata, error: message }), Date.now(), row.id],
}).catch(() => {});
await logError({ title: `Computer #${computer_id} couldn't be set up`, error }).catch(() => {});
}
await pushComputerUpdate({ computer_id }).catch(() => {});
}
@@ -0,0 +1,21 @@
import { sendToUser } from "@/src/websocket/(functions)/send-to-user";
import grabUserComputers from "./grab-user-computers";
type Params = {
computer_id: number | string;
};
/**
* Pushes a computer's current state (`server:computer-updated`) to its
* owner's open tabs. Best effort.
*/
export default async function pushComputerUpdate({ computer_id }: Params) {
const computer = (await grabUserComputers({ computer_id }))[0];
if (!computer?.user_id) return;
sendToUser({
user_id: computer.user_id,
data: { event: "server:computer-updated", data: computer },
});
}
@@ -10,8 +10,10 @@ type Params = {
};
/**
* Switches the computer a bot works on (`bots.computer_id`) to one of its
* own (`grabBotComputers`). Returns its computers, the new one first.
* Switches the computer a bot works on in its own chat
* (`bots.computer_id`) to one of its own (`grabBotComputers`). In a
* workspace the default computer is switched instead
* (`setWorkspaceComputer`). Returns its computers, the new one first.
*/
export default async function setBotComputer({ bot_id, user_id, computer_id }: Params) {
const bot = await grabUserBot({ bot_id, user_id });
@@ -5,32 +5,31 @@ import resolveScopedEnvs from "./resolve-scoped-envs";
type Params = {
computer_id: number | string;
/**
* The bot using the computer. Only workspaces it shares with the
* computer apply; without a bot, every workspace the computer is in.
* The workspace the run is in. Its envs apply when the computer is
* linked to it (and it isn't archived); outside a workspace only the
* computer's own and general envs do.
*/
bot_id?: number | string;
workspace_id?: number | string | null;
};
/**
* The envs a run on this computer gets (`resolveScopedEnvs`). Workspaces
* are non-archived ones, oldest first, so the oldest wins a tie.
* The envs a run on this computer gets (`resolveScopedEnvs`).
*/
export default async function grabComputerEnvs({ computer_id, bot_id }: Params) {
const workspaces =
(
await BunSQLite.sql<{ id: number }>({
sql: `SELECT w.id FROM workspaces w
export default async function grabComputerEnvs({ computer_id, workspace_id }: Params) {
const linked = workspace_id
? (
await BunSQLite.sql<{ id: number }>({
sql: `SELECT w.id FROM workspaces w
JOIN workspace_computers wc ON wc.workspace_id = w.id AND wc.computer_id = ?
WHERE w.archived = 0
${bot_id ? `AND w.id IN (SELECT workspace_id FROM workspace_bots WHERE bot_id = ?)` : ""}
ORDER BY w.id ASC`,
values: bot_id ? [computer_id, bot_id] : [computer_id],
})
).payload || [];
WHERE w.id = ? AND w.archived = 0 LIMIT 1`,
values: [computer_id, workspace_id],
})
).payload?.[0]
: null;
return resolveScopedEnvs({
envs: await grabAllEnvs(),
computer_id: Number(computer_id),
workspace_ids: workspaces.map((workspace) => Number(workspace.id)),
workspace_ids: linked?.id ? [Number(linked.id)] : [],
});
}
@@ -7,7 +7,7 @@ type Params = {
};
/**
* Removes a provider and its models. Harnesses (default and per bot) that
* Removes a provider and its models. Harnesses (default, per bot and per workspace) that
* picked one of its models, or were limited to it, go back to automatic
* on any provider.
*/
@@ -18,7 +18,7 @@ export default async function deleteHarnessProvider({ provider_id }: Params) {
throw new AppError(`Provider not found.`);
}
for (const table of ["harness_settings", "bot_harness_settings"]) {
for (const table of ["harness_settings", "bot_harness_settings", "workspace_harness_settings"]) {
await BunSQLite.sql({
sql: `UPDATE ${table} SET model_id = NULL, provider_id = NULL, updated_at = ?
WHERE provider_id = ? OR model_id IN (SELECT id FROM harness_models WHERE provider_id = ?)`,
@@ -8,23 +8,44 @@ type Params = {
/**
* A worker bot the caller has checked belongs to the session user.
*/
bot_id: number;
bot_id?: number;
/**
* A workspace the caller has checked belongs to the session user. Then
* the settings shown are the workspace's own, over what it inherits:
* the bot's (when given) or the defaults.
*/
workspace_id?: number;
};
/**
* A worker bot's model settings page: per harness, the bot's own setting
* (or the default it inherits), the models it can pick, what it runs on
* now, and what the default runs on.
* A worker bot's (or a workspace's) model settings: per harness, its own
* setting (or what it inherits), the models it can pick, what it runs on
* now, and what it would run on without its own setting.
*/
export default async function grabBotHarnessOverview({
bot_id,
workspace_id,
}: Params): Promise<BotHarnessOverviewPayload> {
/**
* In a workspace, "the default" is what a run there gets without the
* workspace's own setting: the bot's, else the defaults.
*/
const inherited_bot_id = workspace_id ? bot_id : undefined;
const [own, defaults] = await Promise.all([
Promise.all(
Harnesses.map((harness) => grabHarnessSetting({ harness: harness.value, bot_id })),
Harnesses.map((harness) =>
grabHarnessSetting({
harness: harness.value,
bot_id: workspace_id ? undefined : bot_id,
workspace_id,
}),
),
),
Promise.all(
Harnesses.map((harness) => grabHarnessSetting({ harness: harness.value })),
Harnesses.map((harness) =>
grabHarnessSetting({ harness: harness.value, bot_id: inherited_bot_id }),
),
),
]);
@@ -38,11 +59,18 @@ export default async function grabBotHarnessOverview({
harnesses: await Promise.all(
own.map(async (setting, index) => {
const [item, default_item] = await Promise.all([
grabHarnessOverviewItem({ harness: setting.harness, setting, entries, bot_id }),
grabHarnessOverviewItem({
harness: setting.harness,
setting,
entries,
bot_id,
workspace_id,
}),
grabHarnessOverviewItem({
harness: setting.harness,
setting: defaults[index]!,
entries,
bot_id: inherited_bot_id,
}),
]);
@@ -14,6 +14,7 @@ type Params = {
* Resolve what runs now for this bot instead of the default.
*/
bot_id?: number | string;
workspace_id?: number | string;
};
/**
@@ -25,10 +26,11 @@ export default async function grabHarnessOverviewItem({
setting,
entries,
bot_id,
workspace_id,
}: Params): Promise<HarnessOverview> {
const meta = Harnesses.find((h) => h.value == harness)!;
const candidates = setting.enabled
? await resolveHarnessCandidates({ harness, bot_id }).catch(() => [])
? await resolveHarnessCandidates({ harness, bot_id, workspace_id }).catch(() => [])
: [];
const first = candidates[0];
const serving = entries.filter(({ model }) => modelServesHarness({ model, harness }));
@@ -2,6 +2,7 @@ import BunSQLite from "@moduletrace/bun-sqlite";
import type {
BUN_SQLITE_COBBLE_BOT_HARNESS_SETTINGS,
BUN_SQLITE_COBBLE_HARNESS_SETTINGS,
BUN_SQLITE_COBBLE_WORKSPACE_HARNESS_SETTINGS,
} from "@/db/types/db";
import type { HarnessKind } from "@/src/dict/harness-dict";
import parseJobJson from "@/src/bots-queue/parse-job-json";
@@ -22,18 +23,24 @@ export type HarnessSetting = {
enabled: boolean;
options: Record<string, unknown>;
/**
* `bot` when a worker bot overrides the default for this harness.
* `workspace` when the call's workspace sets this harness, `bot` when a
* worker bot overrides the default for it.
*/
source: "default" | "bot";
source: "default" | "bot" | "workspace";
};
type Params = {
harness: HarnessKind;
/**
* The bot the call is for. A worker bot's own setting wins; the
* coordinator always uses the default.
* The bot the call is for. A worker bot's own setting wins over the
* default; the coordinator always uses the default.
*/
bot_id?: number | string;
/**
* The workspace the call is in. Its setting wins over the bot's, for
* the coordinator too.
*/
workspace_id?: number | string | null;
};
type SettingRow = Pick<
@@ -42,13 +49,23 @@ type SettingRow = Pick<
>;
/**
* A harness's setting for a call. With no row it is automatic on the
* default provider (OpenRouter), on, with fallback. Text is always on.
* A harness's setting for a call: the workspace's, else the bot's, else
* the default. With no row it is automatic on the default provider
* (OpenRouter), on, with fallback. Text is always on.
*/
export default async function grabHarnessSetting({
harness,
bot_id,
workspace_id,
}: Params): Promise<HarnessSetting> {
const workspace_row = workspace_id
? await grabWorkspaceOverride({ harness, workspace_id })
: undefined;
if (workspace_row) {
return toSetting({ harness, row: workspace_row, source: "workspace" });
}
const bot_row = bot_id
? await grabBotOverride({ harness, bot_id })
: undefined;
@@ -74,6 +91,21 @@ export default async function grabHarnessSetting({
return toSetting({ harness, row, source: "default" });
}
async function grabWorkspaceOverride({
harness,
workspace_id,
}: {
harness: HarnessKind;
workspace_id: number | string;
}) {
return (
await BunSQLite.sql<BUN_SQLITE_COBBLE_WORKSPACE_HARNESS_SETTINGS>({
sql: `SELECT * FROM workspace_harness_settings WHERE workspace_id = ? AND harness = ? LIMIT 1`,
values: [workspace_id, harness],
})
).payload?.[0];
}
async function grabBotOverride({
harness,
bot_id,
@@ -0,0 +1,87 @@
import BunSQLite from "@moduletrace/bun-sqlite";
import type { HarnessKind } from "@/src/dict/harness-dict";
import AppError from "@/src/utils/app-error";
import assertHarnessModel from "./assert-harness-model";
import assertHarnessProvider from "./assert-harness-provider";
type Params = {
/**
* A workspace the caller has checked belongs to the session user.
*/
workspace_id: number;
harness: HarnessKind;
/**
* Drop the workspace's setting: each bot uses its own or the default
* again.
*/
inherit: boolean;
model_id: number | null;
/**
* Limits the harness to one provider; null: any.
*/
provider_id?: number | null;
fallback: boolean;
enabled: boolean;
options?: Record<string, unknown>;
};
/**
* Saves (or, with `inherit`, removes) a workspace's setting for one
* harness, which wins over its bots' own. Same rules as the default: the
* model must be available and fit the job, and text can't be turned off.
*/
export default async function saveWorkspaceHarnessSetting({
workspace_id,
harness,
inherit,
model_id,
provider_id = null,
fallback,
enabled,
options,
}: Params) {
if (inherit) {
await BunSQLite.sql({
sql: `DELETE FROM workspace_harness_settings WHERE workspace_id = ? AND harness = ?`,
values: [workspace_id, harness],
});
return;
}
if (harness == "text" && !enabled) {
throw new AppError(`The text model can't be turned off.`);
}
if (provider_id) {
await assertHarnessProvider({ provider_id });
}
if (model_id) {
await assertHarnessModel({ harness, model_id, provider_id });
}
const now = Date.now();
const res = await BunSQLite.sql({
sql: `INSERT INTO workspace_harness_settings (created_at, updated_at, workspace_id, harness, model_id, provider_id, fallback, enabled, options)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(workspace_id, harness) DO UPDATE SET updated_at = excluded.updated_at,
model_id = excluded.model_id, provider_id = excluded.provider_id, fallback = excluded.fallback,
enabled = excluded.enabled, options = excluded.options`,
values: [
now,
now,
workspace_id,
harness,
model_id,
provider_id,
fallback ? 1 : 0,
enabled ? 1 : 0,
JSON.stringify(options || {}),
],
});
if (!res.success) {
throw new Error(`Couldn't save the workspace's model setting: ${res.error}`);
}
}
@@ -0,0 +1,83 @@
import BunSQLite from "@moduletrace/bun-sqlite";
import { AppData } from "@/src/data/app-data";
import grabUserBot from "@/src/functions/backend/db/bots/grab-user-bot";
import deleteMedia from "@/src/functions/backend/db/media/delete-media";
import uploadAndRecordMedia from "@/src/functions/backend/db/users/media/upload-and-record-media";
import grabUserWorkspace from "@/src/functions/backend/db/workspaces/grab-user-workspace";
import type { ProfileImageTarget } from "@/src/schemas/profile-image-schema";
import AppError from "@/src/utils/app-error";
import bufferFromBase64 from "@/src/utils/buffer-from-base-64";
type Params = {
user_id: number | string;
target: ProfileImageTarget;
id: number | string;
/**
* The new picture as a base64 data URL; null removes it.
*/
data_url: string | null;
};
const Targets = {
bot: { table: "bots", paradigm: "bot-image" },
workspace: { table: "workspaces", paradigm: "workspace-image" },
} as const;
/**
* Sets (or removes) the picture of one of the user's bots or workspaces:
* a private image, the old one deleted. Returns the new `image_media_id`.
*/
export default async function setProfileImage({ user_id, target, id, data_url }: Params) {
const record =
target == "bot"
? await grabUserBot({ bot_id: id, user_id })
: await grabUserWorkspace({ workspace_id: id, user_id });
if (!record?.id) {
throw new AppError(target == "bot" ? `Bot not found.` : `Workspace not found.`);
}
const { table, paradigm } = Targets[target];
let image_media_id: number | null = null;
if (data_url) {
if (!/^data:image\/(png|jpe?g|webp|gif)/i.test(data_url)) {
throw new AppError(`Pick a PNG, JPG, WebP or GIF image.`);
}
const { buffer, ext } = await bufferFromBase64(data_url);
if (buffer.length > AppData.ProfileImageMaxBytes) {
throw new AppError(
`That picture is too big. Pictures can be up to ${Math.round(AppData.ProfileImageMaxBytes / 1024 / 1024)} MB.`,
);
}
const res = await uploadAndRecordMedia({
data: buffer,
media_type: "image",
media_mime_type: ext,
media_paradigms: [paradigm],
media: { title: `${record.name || target} picture` },
is_private: true,
user: { id: Number(user_id) },
});
if (!res.singleRes?.id) {
throw new AppError(`Couldn't save the picture. Try another image.`);
}
image_media_id = Number(res.singleRes.id);
}
await BunSQLite.sql({
sql: `UPDATE ${table} SET image_media_id = ?, updated_at = ? WHERE id = ?`,
values: [image_media_id, Date.now(), record.id],
});
if (record.image_media_id) {
await deleteMedia({ id: record.image_media_id, user_id: Number(user_id) }).catch(() => {});
}
return { image_media_id };
}
@@ -10,8 +10,9 @@ type Params = {
};
/**
* 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.
* Links one of the user's computers to one of their workspaces, so every
* bot working there can use it. The first one linked becomes the
* workspace's default. Linking it twice is a no-op.
*/
export default async function addComputerToWorkspace({
user_id,
@@ -29,4 +30,11 @@ export default async function addComputerToWorkspace({
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()],
});
if (!workspace.computer_id) {
await BunSQLite.sql({
sql: `UPDATE workspaces SET computer_id = ?, updated_at = ? WHERE id = ?`,
values: [computer.id, Date.now(), workspace.id],
});
}
}
@@ -0,0 +1,36 @@
import deleteMedia from "@/src/functions/backend/db/media/delete-media";
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;
};
/**
* Deletes one of the user's workspaces with its conversations, its
* links and its models. Its bots and computers stay.
*/
export default async function deleteWorkspace({ user_id, workspace_id }: Params) {
const workspace = await grabUserWorkspace({ workspace_id, user_id });
if (!workspace?.id) {
throw new AppError(`Workspace not found.`);
}
const res = await BunSQLite.sql({
sql: `DELETE FROM workspaces WHERE id = ?`,
values: [workspace.id],
});
if (!res.success) {
throw new Error(`Couldn't delete the workspace: ${res.error}`);
}
if (workspace.image_media_id) {
await deleteMedia({ id: workspace.image_media_id, user_id: Number(user_id) }).catch(() => {});
}
return workspace;
}
@@ -0,0 +1,26 @@
import BunSQLite from "@moduletrace/bun-sqlite";
import type { WorkspaceWithActivity } from "@/src/types";
type Params = {
user_id: number | string;
};
/**
* The user's workspaces for the rail, not archived, oldest first, each
* unread while any conversation in it is.
*/
export default async function grabUserWorkspaces({
user_id,
}: Params): Promise<WorkspaceWithActivity[]> {
const res = await BunSQLite.sql<WorkspaceWithActivity>({
sql: `SELECT w.*,
COALESCE((SELECT MAX(s.unread) FROM bot_sessions s WHERE s.workspace_id = w.id), 0) AS unread,
(SELECT MAX(s.last_activity_at) FROM bot_sessions s WHERE s.workspace_id = w.id) AS last_activity_at
FROM workspaces w
WHERE w.user_id = ? AND w.archived = 0
ORDER BY w.id ASC`,
values: [user_id],
});
return res.payload || [];
}
@@ -0,0 +1,60 @@
import type {
BUN_SQLITE_COBBLE_BOTS,
BUN_SQLITE_COBBLE_WORKSPACE_BOTS,
BUN_SQLITE_COBBLE_WORKSPACES,
} from "@/db/types/db";
import BunSQLite from "@moduletrace/bun-sqlite";
type Params = {
workspace_id: number | string;
bot_id: number | string;
user_id: number | string;
};
export type WorkspaceBotMembership = {
workspace: BUN_SQLITE_COBBLE_WORKSPACES;
bot: BUN_SQLITE_COBBLE_BOTS;
/**
* Its `workspace_bots` row. Null for the coordinator, which works in
* every workspace without one until it's given instructions there.
*/
membership: BUN_SQLITE_COBBLE_WORKSPACE_BOTS | null;
};
/**
* One of the user's bots in one of their (non-archived) workspaces: a
* member, or the coordinator, which is in all of them. Null otherwise.
*/
export default async function grabWorkspaceBot({
workspace_id,
bot_id,
user_id,
}: Params): Promise<WorkspaceBotMembership | null> {
const workspace = (
await BunSQLite.sql<BUN_SQLITE_COBBLE_WORKSPACES>({
sql: `SELECT * FROM workspaces WHERE id = ? AND user_id = ? AND archived = 0 LIMIT 1`,
values: [workspace_id, user_id],
})
).payload?.[0];
const bot = (
await BunSQLite.sql<BUN_SQLITE_COBBLE_BOTS>({
sql: `SELECT * FROM bots WHERE id = ? AND user_id = ? LIMIT 1`,
values: [bot_id, user_id],
})
).payload?.[0];
if (!workspace?.id || !bot?.id) return null;
const membership =
(
await BunSQLite.sql<BUN_SQLITE_COBBLE_WORKSPACE_BOTS>({
sql: `SELECT * FROM workspace_bots WHERE workspace_id = ? AND bot_id = ? LIMIT 1`,
values: [workspace.id, bot.id],
})
).payload?.[0] || null;
if (!membership && bot.kind != "coordinator") return null;
return { workspace, bot, membership };
}
@@ -0,0 +1,37 @@
import BunSQLite from "@moduletrace/bun-sqlite";
import type { BUN_SQLITE_COBBLE_WORKSPACE_BOTS } from "@/db/types/db";
import type { WorkspaceBot } from "@/src/types";
import grabBots from "../bots/grab-bots";
type Params = {
user_id: number | string;
workspace_id: number | string;
};
/**
* A workspace's crew (the coordinator first, then its bots by latest
* activity there), each with its instructions and memory for the
* workspace.
*/
export default async function grabWorkspaceBots({
user_id,
workspace_id,
}: Params): Promise<WorkspaceBot[]> {
const [bots, rows] = await Promise.all([
grabBots({ user_id, workspace_id }),
BunSQLite.sql<BUN_SQLITE_COBBLE_WORKSPACE_BOTS>({
sql: `SELECT bot_id, system_prompt, memory FROM workspace_bots WHERE workspace_id = ?`,
values: [workspace_id],
}).then((res) => res.payload || []),
]);
return bots.map((bot) => {
const row = rows.find((r) => r.bot_id == bot.id);
return {
...bot,
workspace_system_prompt: row?.system_prompt || null,
workspace_memory: row?.memory || null,
};
});
}
@@ -0,0 +1,33 @@
import BunSQLite from "@moduletrace/bun-sqlite";
import type { ComputerWithStatus } from "@/src/types";
import parseJobJson from "@/src/bots-queue/parse-job-json";
type Params = {
workspace_id: number | string;
};
/**
* The computers linked to a workspace, with their container's state, its
* default first.
*/
export default async function grabWorkspaceComputers({
workspace_id,
}: Params): Promise<ComputerWithStatus[]> {
const res = await BunSQLite.sql<ComputerWithStatus & { metadata?: string | null }>({
sql: `SELECT c.*, co.status AS status, co.metadata AS metadata FROM computers c
JOIN workspace_computers wc ON wc.computer_id = c.id AND wc.workspace_id = ?
JOIN workspaces w ON w.id = wc.workspace_id AND w.user_id = c.user_id
LEFT JOIN containers co ON co.id = c.container_id
ORDER BY c.id = w.computer_id DESC, wc.id ASC`,
values: [workspace_id],
});
return (res.payload || []).map(({ metadata, ...computer }) => {
const error = parseJobJson({ text: metadata })?.error;
return {
...computer,
error: computer.status == "failed" && typeof error == "string" ? error : null,
};
});
}
@@ -9,8 +9,9 @@ type Params = {
};
/**
* Takes a computer off one of the user's workspaces. Its bots keep any
* access they were given directly.
* Unlinks a computer from one of the user's workspaces; the computer
* itself stays. When it was the default, the oldest computer still linked
* takes over (none: the workspace has no computer).
*/
export default async function removeComputerFromWorkspace({
user_id,
@@ -27,4 +28,13 @@ export default async function removeComputerFromWorkspace({
sql: `DELETE FROM workspace_computers WHERE workspace_id = ? AND computer_id = ?`,
values: [workspace.id, computer_id],
});
if (String(workspace.computer_id) == String(computer_id)) {
await BunSQLite.sql({
sql: `UPDATE workspaces SET updated_at = ?, computer_id = (
SELECT computer_id FROM workspace_computers WHERE workspace_id = ? ORDER BY id ASC LIMIT 1
) WHERE id = ?`,
values: [Date.now(), workspace.id, workspace.id],
});
}
}
@@ -0,0 +1,44 @@
import BunSQLite from "@moduletrace/bun-sqlite";
import AppError from "@/src/utils/app-error";
import grabWorkspaceComputers from "./grab-workspace-computers";
import grabUserWorkspace from "./grab-user-workspace";
type Params = {
user_id: number | string;
workspace_id: number | string;
computer_id: number | string;
};
/**
* Makes one of a workspace's computers its default, where its tasks run.
* Returns the workspace's computers, the new default first.
*/
export default async function setWorkspaceComputer({
user_id,
workspace_id,
computer_id,
}: Params) {
const workspace = await grabUserWorkspace({ workspace_id, user_id });
if (!workspace?.id) {
throw new AppError(`Workspace not found.`);
}
const linked = await grabWorkspaceComputers({ workspace_id: workspace.id });
const computer = linked.find((c) => String(c.id) == String(computer_id));
if (!computer?.id) {
throw new AppError(`That computer isn't in ${workspace.name || "this workspace"}.`);
}
await BunSQLite.sql({
sql: `UPDATE workspaces SET computer_id = ?, updated_at = ? WHERE id = ?`,
values: [computer.id, Date.now(), workspace.id],
});
return {
workspace: { ...workspace, computer_id: computer.id },
computer,
computers: [computer, ...linked.filter((c) => c.id != computer.id)],
};
}
@@ -0,0 +1,59 @@
import BunSQLite from "@moduletrace/bun-sqlite";
import AppError from "@/src/utils/app-error";
import assertPromptLength from "@/src/utils/assert-prompt-length";
import grabWorkspaceBot from "./grab-workspace-bot";
type Params = {
user_id: number | string;
workspace_id: number | string;
bot_id: number | string;
/**
* Only the fields given are changed: the bot's instructions and memory
* for this workspace (markdown, whole-text replace).
*/
system_prompt?: string;
memory?: string;
};
/**
* Rewrites what a bot has for one workspace alone: its instructions there
* and its memory there. The bot, the coordinator and the user all can. The
* coordinator gets its row on first write.
*/
export default async function updateWorkspaceBot({
user_id,
workspace_id,
bot_id,
system_prompt,
memory,
}: Params) {
assertPromptLength({ label: "Workspace instructions", text: system_prompt });
assertPromptLength({ label: "Workspace memory", text: memory });
const found = await grabWorkspaceBot({ workspace_id, bot_id, user_id });
if (!found) {
throw new AppError(`That bot isn't in this workspace.`);
}
const now = Date.now();
await BunSQLite.sql({
sql: `INSERT OR IGNORE INTO workspace_bots (created_at, updated_at, workspace_id, bot_id) VALUES (?, ?, ?, ?)`,
values: [now, now, found.workspace.id!, found.bot.id!],
});
if (system_prompt !== undefined) {
await BunSQLite.sql({
sql: `UPDATE workspace_bots SET system_prompt = ?, updated_at = ? WHERE workspace_id = ? AND bot_id = ?`,
values: [system_prompt, now, found.workspace.id!, found.bot.id!],
});
}
if (memory !== undefined) {
await BunSQLite.sql({
sql: `UPDATE workspace_bots SET memory = ?, updated_at = ? WHERE workspace_id = ? AND bot_id = ?`,
values: [memory, now, found.workspace.id!, found.bot.id!],
});
}
}
@@ -0,0 +1,93 @@
import { AppData } from "@/src/data/app-data";
import { ComputerMcpPath } from "@/src/dict/computer-dict";
import dockerRequest from "./docker-request";
import ensureDockerNetwork from "./ensure-docker-network";
import grabAppContainer from "./grab-app-container";
import grabFreePort from "./grab-free-port";
import inspectDockerContainer from "./inspect-docker-container";
import pullDockerImage from "./pull-docker-image";
import removeDockerContainer from "./remove-docker-container";
import resolveHostPath from "./resolve-host-path";
type Params = {
computer_id: number;
/**
* The container's name, also its hostname on the network.
*/
name: string;
/**
* The computer's home directory as this app sees it.
*/
data_dir: string;
};
/**
* Creates and starts an AIO sandbox container for a computer
* (`AppData.ComputerImage`) on the computers' network, with its home
* directory bind-mounted from `data_dir` and restarted with docker. A
* container left by an earlier try is replaced. When the app runs outside
* docker, the sandbox is also published on a localhost port, returned as
* `mcp_url`'s host.
*/
export default async function createComputerContainer({ computer_id, name, data_dir }: Params) {
const network = await ensureDockerNetwork();
const in_docker = Boolean(await grabAppContainer());
const host_dir = await resolveHostPath({ local_path: data_dir });
const port_key = `${AppData.ComputerPort}/tcp`;
const host_port = in_docker ? undefined : grabFreePort();
await pullDockerImage({ image: AppData.ComputerImage });
await removeDockerContainer({ container: name });
const created = await dockerRequest<{ Id: string }>({
method: "POST",
path: `/containers/create`,
query: { name },
body: {
Image: AppData.ComputerImage,
Hostname: name,
Labels: {
[`${AppData.DockerNetworkName}.computer_id`]: String(computer_id),
},
ExposedPorts: { [port_key]: {} },
HostConfig: {
Binds: [`${host_dir}:${AppData.ComputerHomeDir}`],
NetworkMode: network,
RestartPolicy: { Name: "unless-stopped" },
SecurityOpt: ["seccomp=unconfined"],
ShmSize: AppData.ComputerShmBytes,
Memory: AppData.ComputerMemoryBytes,
NanoCpus: AppData.ComputerCpus * 1e9,
PortBindings: host_port
? { [port_key]: [{ HostIp: "127.0.0.1", HostPort: String(host_port) }] }
: undefined,
},
NetworkingConfig: {
EndpointsConfig: { [network]: { Aliases: [name] } },
},
},
});
if (!created.ok || !created.data?.Id) {
throw new Error(`Couldn't create the container: ${created.message}`);
}
const started = await dockerRequest({
method: "POST",
path: `/containers/${created.data.Id}/start`,
});
if (!started.ok && started.status != 304) {
throw new Error(`Couldn't start the container: ${started.message}`);
}
const state = await inspectDockerContainer({ container: created.data.Id });
return {
docker_id: created.data.Id,
network,
status: state?.status || "unknown",
host_dir,
mcp_url: host_port ? `http://127.0.0.1:${host_port}${ComputerMcpPath}` : undefined,
};
}
@@ -0,0 +1,74 @@
import { AppData } from "@/src/data/app-data";
type Params = {
method?: "GET" | "POST" | "DELETE";
/**
* Docker engine API path, e.g. `/containers/json`.
*/
path: string;
query?: Record<string, string | number | boolean | undefined>;
body?: unknown;
timeout_ms?: number;
};
export type DockerResponse<T> = {
status: number;
ok: boolean;
data: T | null;
/**
* The engine's own message when the call failed.
*/
message?: string;
};
/**
* One call to the docker engine API over its unix socket
* (`AppData.DockerSocketPath`, or `DOCKER_SOCKET` in the process env).
* Never throws for an HTTP error; check `ok`. Throws when the engine
* can't be reached at all.
*/
export default async function dockerRequest<T = unknown>({
method = "GET",
path,
query,
body,
timeout_ms = 60 * 1000,
}: Params): Promise<DockerResponse<T>> {
const search = new URLSearchParams();
for (const [key, value] of Object.entries(query || {})) {
if (value !== undefined) search.set(key, String(value));
}
const url = `http://docker${path}${search.size ? `?${search}` : ""}`;
const res = await fetch(url, {
method,
unix: process.env.DOCKER_SOCKET || AppData.DockerSocketPath,
headers: body === undefined ? undefined : { "Content-Type": "application/json" },
body: body === undefined ? undefined : JSON.stringify(body),
signal: AbortSignal.timeout(timeout_ms),
}).catch((error) => {
throw new Error(
`Can't reach docker: ${error instanceof Error ? error.message : String(error)}. The app needs the docker socket to create computers.`,
);
});
const text = await res.text();
let data: unknown = null;
try {
data = text ? JSON.parse(text) : null;
} catch {
data = text;
}
return {
status: res.status,
ok: res.ok,
data: res.ok ? (data as T) : null,
message: res.ok
? undefined
: (data as { message?: string } | null)?.message || text || `HTTP ${res.status}`,
};
}
@@ -0,0 +1,45 @@
import { AppData } from "@/src/data/app-data";
import dockerRequest from "./docker-request";
import grabAppContainer from "./grab-app-container";
/**
* Makes sure the network computers run on (`AppData.DockerNetworkName`)
* exists and, when the app runs in docker, that its container is on it,
* so it reaches computers by name.
*/
export default async function ensureDockerNetwork() {
const name = AppData.DockerNetworkName;
const existing = await dockerRequest({ path: `/networks/${encodeURIComponent(name)}` });
if (existing.status == 404) {
const created = await dockerRequest({
method: "POST",
path: `/networks/create`,
body: { Name: name, Driver: "bridge", CheckDuplicate: true },
});
if (!created.ok && created.status != 409) {
throw new Error(`Couldn't create the ${name} network: ${created.message}`);
}
} else if (!existing.ok) {
throw new Error(`Couldn't read the ${name} network: ${existing.message}`);
}
const container = await grabAppContainer();
if (!container || container.networks.includes(name)) return name;
const connected = await dockerRequest({
method: "POST",
path: `/networks/${encodeURIComponent(name)}/connect`,
body: { Container: container.id },
});
if (!connected.ok && connected.status != 403) {
throw new Error(`Couldn't connect the app to the ${name} network: ${connected.message}`);
}
container.networks.push(name);
return name;
}
@@ -0,0 +1,41 @@
import fs from "fs";
import os from "os";
import dockerRequest from "./docker-request";
export type AppContainer = {
id: string;
mounts: { Source: string; Destination: string }[];
networks: string[];
};
let cached: Promise<AppContainer | null> | undefined;
/**
* The docker container this app runs in, or null when it runs straight on
* the host (a dev server). Read once per process: docker sets the
* container's hostname to its id.
*/
export default function grabAppContainer(): Promise<AppContainer | null> {
cached ??= inspectSelf();
return cached;
}
async function inspectSelf(): Promise<AppContainer | null> {
if (!fs.existsSync("/.dockerenv")) return null;
const res = await dockerRequest<{
Id: string;
Mounts?: AppContainer["mounts"];
NetworkSettings?: { Networks?: Record<string, unknown> };
}>({ path: `/containers/${encodeURIComponent(os.hostname())}/json` });
if (!res.ok || !res.data?.Id) {
throw new Error(`The app runs in docker but can't find its own container: ${res.message}`);
}
return {
id: res.data.Id,
mounts: res.data.Mounts || [],
networks: Object.keys(res.data.NetworkSettings?.Networks || {}),
};
}
@@ -0,0 +1,16 @@
/**
* A port nothing on this host listens on right now, for publishing a
* computer when the app runs outside docker and can't reach it by name.
*/
export default function grabFreePort() {
const listener = Bun.listen({
hostname: "127.0.0.1",
port: 0,
socket: { data() {} },
});
const port = listener.port;
listener.stop(true);
return port;
}
@@ -0,0 +1,47 @@
import type { ContainerStatus } from "@/src/dict/computer-dict";
import dockerRequest from "./docker-request";
type Params = {
/**
* The container's id or name.
*/
container: string;
};
export type DockerContainerState = {
id: string;
status: ContainerStatus;
/**
* Host ports published for each container port, e.g. `{ "8080/tcp": "8301" }`.
*/
ports: Record<string, string>;
};
/**
* A container's state, or null when docker has no such container.
*/
export default async function inspectDockerContainer({
container,
}: Params): Promise<DockerContainerState | null> {
const res = await dockerRequest<{
Id: string;
State?: { Status?: ContainerStatus };
NetworkSettings?: { Ports?: Record<string, { HostPort?: string }[] | null> };
}>({ path: `/containers/${encodeURIComponent(container)}/json` });
if (res.status == 404) return null;
if (!res.ok || !res.data) {
throw new Error(`Couldn't inspect ${container}: ${res.message}`);
}
return {
id: res.data.Id,
status: res.data.State?.Status || "unknown",
ports: Object.fromEntries(
Object.entries(res.data.NetworkSettings?.Ports || {}).flatMap(([port, bindings]) =>
bindings?.[0]?.HostPort ? [[port, bindings[0].HostPort]] : [],
),
),
};
}
@@ -0,0 +1,51 @@
import dockerRequest from "./docker-request";
type Params = {
/**
* e.g. `ghcr.io/agent-infra/sandbox:latest`.
*/
image: string;
};
/**
* Pulls an image unless docker already has it. Waits for the whole pull,
* which can take minutes for a large image.
*/
export default async function pullDockerImage({ image }: Params) {
const present = await dockerRequest({ path: `/images/${encodeURIComponent(image)}/json` });
if (present.ok) return;
const slash = image.lastIndexOf("/");
const colon = image.lastIndexOf(":");
const has_tag = colon > slash;
const res = await dockerRequest<string>({
method: "POST",
path: `/images/create`,
query: {
fromImage: has_tag ? image.slice(0, colon) : image,
tag: has_tag ? image.slice(colon + 1) : "latest",
},
timeout_ms: 30 * 60 * 1000,
});
/**
* The pull streams JSON lines; a failure mid-way is a line with
* `error`, under a 200.
*/
const failed = String(res.data || "")
.split("\n")
.map((line) => {
try {
return JSON.parse(line) as { error?: string };
} catch {
return null;
}
})
.find((line) => line?.error);
if (!res.ok || failed) {
throw new Error(`Couldn't pull ${image}: ${failed?.error || res.message}`);
}
}
@@ -0,0 +1,24 @@
import dockerRequest from "./docker-request";
type Params = {
/**
* The container's id or name.
*/
container: string;
};
/**
* Stops and removes a container. Its bind-mounted files stay on the host.
* A container docker doesn't have is already gone.
*/
export default async function removeDockerContainer({ container }: Params) {
const res = await dockerRequest({
method: "DELETE",
path: `/containers/${encodeURIComponent(container)}`,
query: { force: true },
});
if (!res.ok && res.status != 404) {
throw new Error(`Couldn't remove ${container}: ${res.message}`);
}
}
@@ -0,0 +1,36 @@
import path from "path";
import grabAppContainer from "./grab-app-container";
type Params = {
/**
* A path as this app sees it.
*/
local_path: string;
};
/**
* The same path as the docker host sees it, for a bind mount: docker
* resolves mount sources on the host, not in this app's container. In a
* container the path must sit under one of its mounts (the deepest one
* wins); on the host it's returned as it is.
*/
export default async function resolveHostPath({ local_path }: Params) {
const container = await grabAppContainer();
const target = path.resolve(local_path);
if (!container) return target;
const mount = container.mounts
.filter(
(m) => target == m.Destination || target.startsWith(`${m.Destination.replace(/\/+$/, "")}/`),
)
.sort((a, b) => b.Destination.length - a.Destination.length)[0];
if (!mount) {
throw new Error(
`${target} isn't on a volume, so computers can't keep their files there. Mount DATA_DIR into the app's container.`,
);
}
return path.join(mount.Source, path.relative(mount.Destination, target));
}