diff --git a/apps/api/src/inbox.test.ts b/apps/api/src/inbox.test.ts new file mode 100644 index 0000000..08360a1 --- /dev/null +++ b/apps/api/src/inbox.test.ts @@ -0,0 +1,230 @@ +import { describe, expect, it } from "vitest"; +import { + buildInboxItems, + type InboxMessageRow, + type InboxRunRow, + type InboxUnreadThreadRow, +} from "./inbox.js"; + +const scout = { id: "bot-scout", name: "Scout", color: "#3B82F6" }; +const writer = { id: "bot-writer", name: "Writer", color: "#D9508A" }; + +function at(minute: number): Date { + return new Date(Date.UTC(2026, 9, 2, 9, minute)); +} + +function run( + id: string, + status: "waiting_input" | "waiting_takeover", + thread: Partial = {}, + bot = scout, +): InboxRunRow { + return { + id, + status, + updatedAt: at(0), + bot, + thread: { botId: bot.id, groupId: null, group: null, ...thread }, + }; +} + +function message(id: string, runId: string, blocks: unknown[], minute: number): InboxMessageRow { + return { id, runId, blocks, createdAt: at(minute) }; +} + +const question = { kind: "ask", text: "Which city should I use?", status: "pending" } as const; +const approval = { + kind: "ask", + text: "Review before writing to notes", + approvalEffectId: "effect-1", + status: "pending", + actions: [ + { id: "allow", label: "Allow once" }, + { id: "always", label: "Always allow this tool" }, + { id: "deny", label: "Deny" }, + ], +} as const; + +function unreadThread( + bot: InboxUnreadThreadRow["bot"], + latest?: { id: string; text: string; minute: number }, +): InboxUnreadThreadRow { + return { + createdAt: at(0), + bot, + messages: latest + ? [ + { + id: latest.id, + blocks: [{ kind: "text", text: latest.text }], + createdAt: at(latest.minute), + }, + ] + : [], + }; +} + +describe("buildInboxItems", () => { + it("lists a pending ask with the asking bot as sender and its own thread as target", () => { + const items = buildInboxItems({ + runs: [run("run-1", "waiting_input")], + messages: [message("msg-1", "run-1", [{ kind: "text", text: "one moment" }, question], 5)], + unreadThreads: [], + }); + + expect(items).toEqual([ + { + kind: "ask", + bot: scout, + target: { botId: scout.id }, + runId: "run-1", + messageId: "msg-1", + ask: question, + createdAt: at(5).toISOString(), + }, + ]); + }); + + it("marks approval asks as approvals", () => { + const items = buildInboxItems({ + runs: [run("run-1", "waiting_input")], + messages: [message("msg-1", "run-1", [approval], 1)], + unreadThreads: [], + }); + + expect(items.map((item) => item.kind)).toEqual(["approval"]); + }); + + it("answers group asks through the group and leaves out ones it cannot answer from here", () => { + const items = buildInboxItems({ + runs: [ + run("in-main", "waiting_input", { + botId: null, + groupId: "group-main", + group: { kind: "main", archivedAt: null }, + }), + run("in-group", "waiting_input", { + botId: null, + groupId: "group-ops", + group: { kind: "group", archivedAt: null }, + }), + run("in-archived", "waiting_input", { + botId: null, + groupId: "group-old", + group: { kind: "group", archivedAt: at(0) }, + }), + run("in-external", "waiting_input", { botId: null }), + ], + messages: ["in-main", "in-group", "in-archived", "in-external"].map((runId, index) => + message(`msg-${runId}`, runId, [question], index), + ), + unreadThreads: [], + }); + + expect(items).toHaveLength(1); + expect(items[0]).toMatchObject({ runId: "in-group", target: { groupId: "group-ops" } }); + }); + + it("uses the newest pending ask and skips runs without one", () => { + const items = buildInboxItems({ + runs: [run("run-1", "waiting_input"), run("run-2", "waiting_input", {}, writer)], + messages: [ + message("newest", "run-1", [{ ...question, text: "And the date?" }], 9), + message("older", "run-1", [{ ...question, status: "answered", answer: "Paris" }], 4), + message("answered", "run-2", [{ ...question, status: "answered", answer: "Rome" }], 3), + message("malformed", "run-2", [{ kind: "ask" }], 8), + ], + unreadThreads: [], + }); + + expect(items).toHaveLength(1); + expect(items[0]).toMatchObject({ messageId: "newest", ask: { text: "And the date?" } }); + }); + + it("sends takeovers to the bot's own thread, including runs in the main conversation", () => { + const items = buildInboxItems({ + runs: [ + run("run-1", "waiting_takeover", { + botId: null, + groupId: "group-main", + group: { kind: "main", archivedAt: null }, + }), + run("run-2", "waiting_takeover", {}, writer), + ], + messages: [ + message( + "msg-1", + "run-1", + [{ kind: "computer", state: "Needs you", text: "Sign in to continue." }], + 2, + ), + ], + unreadThreads: [], + }); + + expect(items).toEqual([ + { + kind: "takeover", + bot: writer, + target: { botId: writer.id }, + runId: "run-2", + messageId: null, + text: "", + createdAt: at(0).toISOString(), + }, + { + kind: "takeover", + bot: scout, + target: { botId: scout.id }, + runId: "run-1", + messageId: "msg-1", + text: "Sign in to continue.", + createdAt: at(2).toISOString(), + }, + ]); + }); + + it("adds unread threads after waiting items unless the bot already waits in that thread", () => { + const items = buildInboxItems({ + runs: [ + run("run-1", "waiting_input"), + run( + "run-2", + "waiting_input", + { + botId: null, + groupId: "group-ops", + group: { kind: "group", archivedAt: null }, + }, + writer, + ), + ], + messages: [ + message("ask-1", "run-1", [question], 7), + message("ask-2", "run-2", [question], 8), + ], + unreadThreads: [ + unreadThread(scout, { id: "scout-latest", text: "covered by the ask", minute: 7 }), + unreadThread(writer, { id: "writer-latest", text: "Draft ready\nfor review", minute: 1 }), + unreadThread({ id: "bot-quiet", name: "Quiet", color: "#3EC5A8" }), + unreadThread(null, { id: "orphan", text: "no bot", minute: 1 }), + ], + }); + + expect(items.map((item) => [item.kind, item.bot.name])).toEqual([ + ["ask", "Scout"], + ["ask", "Writer"], + ["unread", "Quiet"], + ["unread", "Writer"], + ]); + expect(items[3]).toEqual({ + kind: "unread", + bot: writer, + target: { botId: writer.id }, + messageId: "writer-latest", + text: "Draft ready for review", + createdAt: at(1).toISOString(), + }); + expect(items[2]).toMatchObject({ messageId: null, text: "", createdAt: at(0).toISOString() }); + }); +}); diff --git a/apps/api/src/inbox.ts b/apps/api/src/inbox.ts new file mode 100644 index 0000000..549d5a6 --- /dev/null +++ b/apps/api/src/inbox.ts @@ -0,0 +1,236 @@ +import type { Actor, InboxItem, InboxThreadTarget } from "@rakazo/contracts"; +import { MessageBlock } from "@rakazo/contracts"; +import { isApprovalAskBlock } from "@rakazo/core"; +import type { PrismaClient } from "@rakazo/db"; +import { promptSnippet } from "./runs.js"; + +/** Caps per query; a space rarely holds more waiting runs or unread threads than this. */ +const INBOX_RUN_LIMIT = 50; +const INBOX_UNREAD_LIMIT = 50; +const INBOX_PREVIEW_LENGTH = 280; + +type AskBlock = Extract; +type ComputerBlock = Extract; +type InboxBot = { id: string; name: string; color: string }; + +export interface InboxRunRow { + id: string; + status: string; + updatedAt: Date; + bot: InboxBot; + thread: { + botId: string | null; + groupId: string | null; + group: { kind: string; archivedAt: Date | null } | null; + }; +} + +export interface InboxMessageRow { + id: string; + runId: string | null; + blocks: unknown; + createdAt: Date; +} + +export interface InboxUnreadThreadRow { + createdAt: Date; + bot: InboxBot | null; + /** The newest bot message, if any. */ + messages: Array>; +} + +function parseBlocks(blocks: unknown): MessageBlock[] { + const parsed = MessageBlock.array().safeParse(blocks); + return parsed.success ? parsed.data : []; +} + +function pendingAsk(blocks: MessageBlock[]): AskBlock | undefined { + return blocks.find( + (block): block is AskBlock => block.kind === "ask" && block.status !== "answered", + ); +} + +function takeoverRequest(blocks: MessageBlock[]): ComputerBlock | undefined { + return blocks.find((block): block is ComputerBlock => block.kind === "computer"); +} + +/** The newest message whose blocks hold what `pick` looks for. */ +function newest(messages: InboxMessageRow[], pick: (blocks: MessageBlock[]) => T | undefined) { + for (const message of messages) { + const block = pick(parseBlocks(message.blocks)); + if (block) return { message, block }; + } + return undefined; +} + +function previewText(blocks: unknown): string { + for (const block of parseBlocks(blocks)) { + if ("text" in block && typeof block.text === "string" && block.text.trim()) { + return promptSnippet(block.text, INBOX_PREVIEW_LENGTH); + } + } + return ""; +} + +/** + * Where an ask can be answered from. The main conversation already shows its own asks, and an + * archived group or an external conversation has no thread target to answer through. + */ +function askTarget(thread: InboxRunRow["thread"]): InboxThreadTarget | null { + if (thread.groupId) { + if (!thread.group || thread.group.kind === "main" || thread.group.archivedAt) return null; + return { groupId: thread.groupId }; + } + return thread.botId ? { botId: thread.botId } : null; +} + +function targetKey(target: InboxThreadTarget): string { + return "botId" in target ? `bot:${target.botId}` : `group:${target.groupId}`; +} + +function byCreatedAt(left: InboxItem, right: InboxItem): number { + return left.createdAt.localeCompare(right.createdAt); +} + +/** + * Turns waiting runs, their ask and takeover messages, and unread threads into inbox items: + * everything that waits on the user first, then unread threads not already covered, each + * oldest first. + */ +export function buildInboxItems(input: { + runs: InboxRunRow[]; + /** Ask and takeover messages of those runs, newest first. */ + messages: InboxMessageRow[]; + unreadThreads: InboxUnreadThreadRow[]; +}): InboxItem[] { + const messagesByRun = new Map(); + for (const message of input.messages) { + if (!message.runId) continue; + const list = messagesByRun.get(message.runId) ?? []; + list.push(message); + messagesByRun.set(message.runId, list); + } + + const waiting: InboxItem[] = []; + for (const run of input.runs) { + const messages = messagesByRun.get(run.id) ?? []; + if (run.status === "waiting_input") { + const target = askTarget(run.thread); + const pending = target ? newest(messages, pendingAsk) : undefined; + if (!target || !pending) continue; + waiting.push({ + kind: isApprovalAskBlock(pending.block) ? "approval" : "ask", + bot: run.bot, + target, + runId: run.id, + messageId: pending.message.id, + ask: pending.block, + createdAt: pending.message.createdAt.toISOString(), + }); + } else if (run.status === "waiting_takeover") { + // Takeover controls live in the bot's own thread, so the item stays even for a run that + // asked in the main conversation. + const request = newest(messages, takeoverRequest); + waiting.push({ + kind: "takeover", + bot: run.bot, + target: { botId: run.bot.id }, + runId: run.id, + messageId: request?.message.id ?? null, + text: request?.block.text ?? "", + createdAt: (request?.message.createdAt ?? run.updatedAt).toISOString(), + }); + } + } + + const covered = new Set(waiting.map((item) => targetKey(item.target))); + const unread: InboxItem[] = []; + for (const thread of input.unreadThreads) { + if (!thread.bot) continue; + const target = { botId: thread.bot.id }; + if (covered.has(targetKey(target))) continue; + const latest = thread.messages[0]; + unread.push({ + kind: "unread", + bot: thread.bot, + target, + messageId: latest?.id ?? null, + text: previewText(latest?.blocks), + createdAt: (latest?.createdAt ?? thread.createdAt).toISOString(), + }); + } + + return [...waiting.sort(byCreatedAt), ...unread.sort(byCreatedAt)]; +} + +const botSelect = { id: true, name: true, color: true } as const; + +/** The current space's inbox: only the actor's own bots, runs and threads. */ +export async function listInbox(prisma: PrismaClient, actor: Actor): Promise { + const ownBot = { userId: actor.userId, archivedAt: null }; + const [runs, unreadThreads] = await Promise.all([ + prisma.run.findMany({ + where: { + spaceId: actor.spaceId, + userId: actor.userId, + status: { in: ["waiting_input", "waiting_takeover"] }, + bot: ownBot, + thread: { userId: actor.userId }, + }, + select: { + id: true, + status: true, + threadId: true, + updatedAt: true, + bot: { select: botSelect }, + thread: { + select: { + botId: true, + groupId: true, + group: { select: { kind: true, archivedAt: true } }, + }, + }, + }, + orderBy: [{ updatedAt: "asc" }, { id: "asc" }], + take: INBOX_RUN_LIMIT, + }), + // Only specialists' own threads; group threads, the main conversation included, stay out. + prisma.thread.findMany({ + where: { + spaceId: actor.spaceId, + userId: actor.userId, + unread: true, + bot: { ...ownBot, kind: "specialist" }, + }, + select: { + createdAt: true, + bot: { select: botSelect }, + messages: { + where: { role: "bot" }, + orderBy: { seq: "desc" }, + take: 1, + select: { id: true, blocks: true, createdAt: true }, + }, + }, + orderBy: [{ createdAt: "asc" }, { id: "asc" }], + take: INBOX_UNREAD_LIMIT, + }), + ]); + const messages = + runs.length === 0 + ? [] + : await prisma.message.findMany({ + where: { + threadId: { in: [...new Set(runs.map((run) => run.threadId))] }, + runId: { in: runs.map((run) => run.id) }, + role: "bot", + OR: [ + { blocks: { array_contains: [{ kind: "ask" }] } }, + { blocks: { array_contains: [{ kind: "computer" }] } }, + ], + }, + orderBy: { seq: "desc" }, + select: { id: true, runId: true, blocks: true, createdAt: true }, + }); + return buildInboxItems({ runs, messages, unreadThreads }); +} diff --git a/apps/api/src/router.ts b/apps/api/src/router.ts index 7067820..f4e12b2 100644 --- a/apps/api/src/router.ts +++ b/apps/api/src/router.ts @@ -152,6 +152,7 @@ import { resolveBusyBotName, toComputerStatus, } from "./computer-status.js"; +import { listInbox } from "./inbox.js"; import { searchIntegrationCatalog } from "./integration-catalog.js"; import { buildMcpUpdateMaterial } from "./mcp-material.js"; import { @@ -4653,6 +4654,11 @@ export function createRouter(deps: RouterDeps) { runs: await listSpaceRuns(deps.prisma, context.actor, input.filter), })), }, + inbox: { + list: authed.inbox.list.handler(async ({ context }) => ({ + items: await listInbox(deps.prisma, context.actor), + })), + }, voice: { catalog: authed.voice.catalog.handler(async () => listVoiceCatalog()), status: authed.voice.status.handler(async ({ context }) => { diff --git a/apps/api/src/runs.ts b/apps/api/src/runs.ts index 5d22bff..1e8626c 100644 --- a/apps/api/src/runs.ts +++ b/apps/api/src/runs.ts @@ -5,7 +5,7 @@ import type { PrismaClient } from "@rakazo/db"; const RECENT_LIMIT = 20; const TERMINAL_STATUSES = ["completed", "failed", "cancelled"] as const; -function promptSnippet(prompt: string, max = 120): string { +export function promptSnippet(prompt: string, max = 120): string { const oneLine = prompt.replace(/\s+/g, " ").trim(); if (oneLine.length <= max) return oneLine; return `${oneLine.slice(0, max - 1)}…`; diff --git a/apps/web/e2e/inbox.spec.ts b/apps/web/e2e/inbox.spec.ts new file mode 100644 index 0000000..b31ea48 --- /dev/null +++ b/apps/web/e2e/inbox.spec.ts @@ -0,0 +1,85 @@ +import { expect, type Page, test } from "@playwright/test"; +import type { Group } from "@rakazo/contracts"; +import { captureScreenshot, routePatchRpc, rpc, signup, waitForChatComposer } from "./helpers"; + +test("answers a specialist's question from the main conversation", async ({ page }, testInfo) => { + const stamp = Date.now(); + await signup(page, `inbox-${stamp}@rakazo.test`, "password12", "Inbox"); + // Bots come from RPC: this spec is about the inbox, not the onboarding flow. + await page.waitForURL(/\/(onboarding|app)/, { timeout: 30_000 }); + const researcher = await createBot(page, "Researcher"); + + await rpc(page, "threads/send", { botId: researcher.id, text: "ask me which city to use" }); + await expect + .poll(async () => (await activeRun(page, researcher.id))?.status ?? null, { timeout: 60_000 }) + .toBe("waiting_input"); + + await openMainConversation(page, researcher.id); + const inbox = page.getByRole("region", { name: "Needs you" }); + const ask = inbox.locator("[data-kind='ask']"); + await expect(ask).toBeVisible({ timeout: 30_000 }); + await expect(ask).toContainText("Researcher"); + await expect(ask).toContainText("Which city should I use?"); + await expect(page.getByRole("combobox", { name: "Message" })).toBeInViewport(); + await captureScreenshot(page, testInfo, "inbox-ask"); + + await ask.getByRole("button", { name: "Edit first" }).click(); + await ask.getByRole("textbox", { name: "Answer" }).fill("Paris"); + await ask.getByRole("button", { name: "Send answer" }).click(); + await expect(ask).toHaveCount(0); + await expect.poll(async () => activeRun(page, researcher.id), { timeout: 60_000 }).toBeNull(); + + // The answer resumed the specialist in its own thread; its reply arrives as unread. + const reply = inbox.locator("[data-kind='unread']").filter({ hasText: "Researcher" }); + await expect(reply).toContainText("Paris", { timeout: 30_000 }); + await captureScreenshot(page, testInfo, "inbox-unread-reply"); + + await reply.getByRole("button").click(); + await page.waitForURL(new RegExp(`/app/${researcher.id}$`)); + await expect(page.getByText("Answered: Paris", { exact: true })).toBeVisible({ timeout: 30_000 }); +}); + +async function createBot(page: Page, name: string) { + return rpc<{ id: string }>(page, "bots/create", { + name, + title: "", + description: "", + notifyOnFinish: true, + computerMode: "team", + }); +} + +async function activeRun(page: Page, botId: string) { + const snapshot = await rpc<{ run?: { status: string } | null }>(page, "threads/get", { botId }); + return snapshot.run ?? null; +} + +/** + * Opens the user's main conversation. Until the app sets one up, a group with the specialist + * stands in for it and is reported to the client as the main conversation. + */ +async function openMainConversation(page: Page, specialistId: string) { + // The shell's first load is where the app sets up the main conversation. + await rpc(page, "bootstrap", {}); + const groups = await rpc(page, "groups/list", {}); + let main = groups.find((group) => group.kind === "main"); + if (!main) { + const lead = await createBot(page, "Lead"); + const stand = await rpc(page, "groups/create", { + name: "Main", + botIds: [lead.id, specialistId], + }); + const asMain = (list: unknown) => + Array.isArray(list) + ? list.map((group) => (group?.id === stand.id ? { ...group, kind: "main" } : group)) + : list; + await routePatchRpc(page, "bootstrap", (json) => ({ ...json, groups: asMain(json.groups) })); + await routePatchRpc(page, "spaces/list", (json) => { + const current = json.current as { groups?: unknown } | undefined; + return current ? { ...json, current: { ...current, groups: asMain(current.groups) } } : json; + }); + main = stand; + } + await page.goto(`/app/g/${main.id}`, { waitUntil: "domcontentloaded" }); + await waitForChatComposer(page); +} diff --git a/apps/web/src/components/AskCard.tsx b/apps/web/src/components/AskCard.tsx index 483b3e9..3b4f59d 100644 --- a/apps/web/src/components/AskCard.tsx +++ b/apps/web/src/components/AskCard.tsx @@ -3,7 +3,7 @@ import { Trans, useLingui } from "@lingui/react/macro"; import { ChatMarkdown } from "@rakazo/chat-ui/web"; import type { ThreadMessage } from "@rakazo/contracts"; import { isApprovalAskBlock, isSecretAskBlock, selectedAskActionLabel } from "@rakazo/core"; -import { Button, Input } from "@rakazo/ui-web"; +import { Button, cn, Input } from "@rakazo/ui-web"; import { useState } from "react"; export type AskBlock = Extract; @@ -49,10 +49,12 @@ export function AskCard({ block, canAnswer, onAnswer, + className, }: { block: AskBlock; canAnswer: boolean; onAnswer: (text: string) => Promise; + className?: string; }) { const { t } = useLingui(); const [editing, setEditing] = useState(false); @@ -86,7 +88,7 @@ export function AskCard({ return (
{block.text} diff --git a/apps/web/src/components/MainInbox.test.tsx b/apps/web/src/components/MainInbox.test.tsx new file mode 100644 index 0000000..0d6175d --- /dev/null +++ b/apps/web/src/components/MainInbox.test.tsx @@ -0,0 +1,219 @@ +// @vitest-environment jsdom + +import type { InboxItem } from "@rakazo/contracts"; +import type { ReactNode } from "react"; +import { act } from "react"; +import { createRoot, type Root } from "react-dom/client"; +import { afterEach, beforeEach, expect, it, vi } from "vitest"; + +const api = vi.hoisted(() => ({ list: vi.fn(), answer: vi.fn(), markRead: vi.fn() })); +vi.mock("../lib/rpc", () => ({ + rpc: { inbox: { list: api.list }, threads: { answer: api.answer, markRead: api.markRead } }, +})); +vi.mock("@lingui/react/macro", () => { + const t = (parts: TemplateStringsArray) => parts.join(""); + return { useLingui: () => ({ t }), Trans: ({ children }: { children: ReactNode }) => children }; +}); +vi.mock("@rakazo/ui-web", () => ({ + BotAvatar: () => , + parseBotAvatar: (color: string) => ({ color }), + resolvePersonaColorDef: (_id: string, color: string) => ({ light: color }), +})); +vi.mock("./AskCard", () => ({ + AskCard: ({ + block, + onAnswer, + }: { + block: { text: string }; + onAnswer: (text: string) => Promise; + }) => ( + // Like the real card, show a failed answer instead of rejecting. + + ), +})); + +import { MainInbox } from "./MainInbox"; + +const scout = { id: "bot-scout", name: "Scout", color: "#3B82F6" }; +const writer = { id: "bot-writer", name: "Writer", color: "#D9508A" }; +const createdAt = "2026-10-02T09:00:00.000Z"; + +const askInOwnThread: InboxItem = { + kind: "ask", + bot: scout, + target: { botId: scout.id }, + runId: "run-ask", + messageId: "msg-ask", + ask: { kind: "ask", text: "Which city should I use?", status: "pending" }, + createdAt, +}; +const approvalInGroup: InboxItem = { + kind: "approval", + bot: writer, + target: { groupId: "group-ops" }, + runId: "run-approval", + messageId: "msg-approval", + ask: { kind: "ask", text: "Review before writing to notes", status: "pending" }, + createdAt, +}; +const takeover: InboxItem = { + kind: "takeover", + bot: writer, + target: { botId: writer.id }, + runId: "run-takeover", + messageId: "msg-takeover", + text: "Sign in to continue.", + createdAt, +}; +const unread: InboxItem = { + kind: "unread", + bot: scout, + target: { botId: scout.id }, + messageId: "msg-unread", + text: "Draft ready for review", + createdAt, +}; + +const idleBots = [ + { id: scout.id, status: "idle", unread: false, preview: "" }, + { id: writer.id, status: "idle", unread: false, preview: "" }, +]; + +let container: HTMLDivElement; +let root: Root; +const onOpenBot = vi.fn(); +const onOpenGroup = vi.fn(); + +function render(bots = idleBots) { + return act(async () => + root.render(), + ); +} + +function itemOf(kind: InboxItem["kind"]) { + const found = container.querySelector( + `[data-testid='main-inbox-item'][data-kind='${kind}']`, + ); + if (!found) throw new Error(`Missing ${kind} item`); + return found; +} + +beforeEach(() => { + vi.stubGlobal("IS_REACT_ACT_ENVIRONMENT", true); + api.list.mockReset(); + api.answer.mockReset(); + api.markRead.mockReset().mockResolvedValue({ ok: true }); + onOpenBot.mockReset(); + onOpenGroup.mockReset(); + container = document.createElement("div"); + document.body.append(container); + root = createRoot(container); +}); + +afterEach(async () => { + await act(async () => root.unmount()); + container.remove(); + vi.unstubAllGlobals(); +}); + +it("renders nothing until a bot needs the user", async () => { + api.list.mockResolvedValue({ items: [] }); + await render(); + + expect(api.list).toHaveBeenCalledOnce(); + expect(container.querySelector("[data-testid='main-inbox']")).toBeNull(); +}); + +it("shows each item with its bot as the sender", async () => { + api.list.mockResolvedValue({ items: [askInOwnThread, approvalInGroup, takeover, unread] }); + await render(); + + const inbox = container.querySelector("[data-testid='main-inbox']"); + expect(inbox?.getAttribute("aria-label")).toBe("Needs you"); + expect(itemOf("ask").textContent).toContain("Scout"); + expect(itemOf("ask").textContent).toContain("Which city should I use?"); + expect(itemOf("approval").textContent).toContain("Writer"); + expect(itemOf("takeover").textContent).toContain("Sign in to continue."); + expect(itemOf("unread").textContent).toContain("Draft ready for review"); +}); + +it("answers through the thread the ask came from and drops only that item", async () => { + api.list + .mockResolvedValueOnce({ items: [askInOwnThread, approvalInGroup] }) + .mockReturnValue(new Promise(() => undefined)); + api.answer.mockResolvedValue({ ok: true }); + await render(); + + await act(async () => + itemOf("approval").querySelector("[data-answer]")?.click(), + ); + + expect(api.answer).toHaveBeenCalledWith({ + groupId: "group-ops", + runId: "run-approval", + messageId: "msg-approval", + answer: "Paris", + }); + expect(api.markRead).not.toHaveBeenCalled(); + expect(api.list).toHaveBeenCalledTimes(2); + expect(container.querySelectorAll("[data-testid='main-inbox-item']")).toHaveLength(1); + expect(itemOf("ask")).toBeTruthy(); +}); + +it("marks the bot's own thread read before answering, so the question does not return as unread", async () => { + api.list.mockResolvedValueOnce({ items: [askInOwnThread] }).mockResolvedValue({ items: [] }); + api.answer.mockResolvedValue({ ok: true }); + await render(); + + await act(async () => itemOf("ask").querySelector("[data-answer]")?.click()); + + expect(api.markRead).toHaveBeenCalledWith({ botId: scout.id }); + expect(api.markRead.mock.invocationCallOrder[0]).toBeLessThan( + api.answer.mock.invocationCallOrder[0]!, + ); + expect(container.querySelector("[data-testid='main-inbox']")).toBeNull(); +}); + +it("refreshes after a failed answer so a stale ask disappears", async () => { + api.list.mockResolvedValueOnce({ items: [askInOwnThread] }).mockResolvedValue({ items: [] }); + api.answer.mockRejectedValue(new Error("This prompt is no longer awaiting an answer")); + await render(); + + await act(async () => itemOf("ask").querySelector("[data-answer]")?.click()); + + expect(api.answer).toHaveBeenCalledWith({ + botId: scout.id, + runId: "run-ask", + messageId: "msg-ask", + answer: "Paris", + }); + expect(container.querySelector("[data-testid='main-inbox']")).toBeNull(); +}); + +it("opens the bot's thread for takeovers and unread threads, and the source thread for asks", async () => { + api.list.mockResolvedValue({ items: [approvalInGroup, takeover, unread] }); + await render(); + + await act(async () => itemOf("takeover").querySelector("button")?.click()); + await act(async () => itemOf("unread").querySelector("button")?.click()); + await act(async () => itemOf("approval").querySelector("button")?.click()); + + expect(onOpenBot.mock.calls).toEqual([[writer.id], [scout.id]]); + expect(onOpenGroup.mock.calls).toEqual([["group-ops"]]); +}); + +it("refreshes when a bot's state changes, not on every render", async () => { + api.list.mockResolvedValue({ items: [] }); + await render(); + await render([...idleBots]); + expect(api.list).toHaveBeenCalledOnce(); + + await render([{ ...idleBots[0]!, status: "waiting_input" }, idleBots[1]!]); + expect(api.list).toHaveBeenCalledTimes(2); +}); diff --git a/apps/web/src/components/MainInbox.tsx b/apps/web/src/components/MainInbox.tsx new file mode 100644 index 0000000..5f3007b --- /dev/null +++ b/apps/web/src/components/MainInbox.tsx @@ -0,0 +1,169 @@ +import { useLingui } from "@lingui/react/macro"; +import type { Bot, InboxItem, InboxThreadTarget } from "@rakazo/contracts"; +import { BotAvatar, parseBotAvatar, resolvePersonaColorDef } from "@rakazo/ui-web"; +import { Monitor } from "lucide-react"; +import { useCallback, useEffect, useMemo, useRef, useState } from "react"; +import { inboxItemKey, inboxRefreshSignal } from "../lib/main-inbox"; +import { rpc } from "../lib/rpc"; +import { AskCard } from "./AskCard"; + +/** Catches what the bots' state does not reveal, such as a second ask with the same text. */ +const INBOX_POLL_MS = 15_000; + +type PendingAskItem = Extract; + +/** + * What the user's bots need from them outside the main conversation, with each bot as the + * sender: asks and approvals are answered in place, takeovers and unread threads open the + * bot's thread. + */ +export function MainInbox({ + bots, + onOpenBot, + onOpenGroup, +}: { + /** The space's bots; a change in their status, unread flag or preview refreshes the inbox. */ + bots: readonly Pick[]; + onOpenBot: (botId: string) => void; + onOpenGroup: (groupId: string) => void; +}) { + const { t } = useLingui(); + const [items, setItems] = useState([]); + const latestRequest = useRef(0); + const signal = useMemo(() => inboxRefreshSignal(bots), [bots]); + + const refresh = useCallback(async () => { + const request = ++latestRequest.current; + try { + const listed = await rpc.inbox.list(); + if (request === latestRequest.current) setItems(listed.items); + } catch { + // Keep the last list; the next refresh retries. + } + }, []); + + // `signal` is the trigger: refetch whenever the bots' state changes. + useEffect(() => { + void refresh(); + }, [refresh, signal]); + + useEffect(() => { + const timer = window.setInterval(() => { + if (document.visibilityState === "visible") void refresh(); + }, INBOX_POLL_MS); + return () => { + window.clearInterval(timer); + // Ignore responses that arrive after unmount. + latestRequest.current += 1; + }; + }, [refresh]); + + const open = useCallback( + (target: InboxThreadTarget) => { + if ("botId" in target) onOpenBot(target.botId); + else onOpenGroup(target.groupId); + }, + [onOpenBot, onOpenGroup], + ); + + const answer = useCallback( + async (item: PendingAskItem, text: string) => { + try { + // Answering reads the ask, so the bot's thread should not come back as unread with the + // same question. Mark it first: the reply to this answer marks it unread again. + if ("botId" in item.target) { + await rpc.threads.markRead(item.target).catch(() => undefined); + } + await rpc.threads.answer({ + ...item.target, + runId: item.runId, + messageId: item.messageId, + answer: text, + }); + const key = inboxItemKey(item); + setItems((current) => current.filter((entry) => inboxItemKey(entry) !== key)); + } finally { + void refresh(); + } + }, + [refresh], + ); + + if (items.length === 0) return null; + + return ( + // Shrinks and scrolls before it can push the composer out of view. +
+
    + {items.map((item) => ( +
  • + {item.kind === "ask" || item.kind === "approval" ? ( +
    + + answer(item, text)} + className="max-w-none" + /> +
    + ) : ( + + )} +
    • + ))} +
+
+ ); +} + +function Sender({ bot }: { bot: InboxItem["bot"] }) { + const color = resolvePersonaColorDef(bot.id, parseBotAvatar(bot.color).color).light; + return ( + + + {bot.name} + + ); +} diff --git a/apps/web/src/lib/main-inbox.test.ts b/apps/web/src/lib/main-inbox.test.ts new file mode 100644 index 0000000..172de4c --- /dev/null +++ b/apps/web/src/lib/main-inbox.test.ts @@ -0,0 +1,40 @@ +import type { InboxItem } from "@rakazo/contracts"; +import { describe, expect, it } from "vitest"; +import { inboxItemKey, inboxRefreshSignal } from "./main-inbox"; + +const bot = { id: "bot-scout", name: "Scout", color: "#3B82F6" }; +const base = { bot, target: { botId: bot.id }, createdAt: "2026-10-02T09:00:00.000Z" }; +const ask = { kind: "ask", text: "Which city?", status: "pending" } as const; + +describe("inboxItemKey", () => { + it("keys asks by message, takeovers by run, and unread threads by bot", () => { + const items: InboxItem[] = [ + { ...base, kind: "ask", runId: "run-1", messageId: "msg-1", ask }, + { ...base, kind: "approval", runId: "run-2", messageId: "msg-2", ask }, + { ...base, kind: "takeover", runId: "run-3", messageId: null, text: "" }, + { ...base, kind: "unread", messageId: "msg-4", text: "hi" }, + ]; + + expect(items.map(inboxItemKey)).toEqual([ + "ask:msg-1", + "ask:msg-2", + "takeover:run-3", + `unread:${bot.id}`, + ]); + }); +}); + +describe("inboxRefreshSignal", () => { + const idle = { id: bot.id, status: "idle", unread: false, preview: "done" }; + + it("stays the same while nothing the inbox shows has changed", () => { + expect(inboxRefreshSignal([idle])).toBe(inboxRefreshSignal([{ ...idle }])); + }); + + it("changes with run status, unread state, or a new message", () => { + const signal = inboxRefreshSignal([idle]); + expect(inboxRefreshSignal([{ ...idle, status: "waiting_input" }])).not.toBe(signal); + expect(inboxRefreshSignal([{ ...idle, unread: true }])).not.toBe(signal); + expect(inboxRefreshSignal([{ ...idle, preview: "Which city?" }])).not.toBe(signal); + }); +}); diff --git a/apps/web/src/lib/main-inbox.ts b/apps/web/src/lib/main-inbox.ts new file mode 100644 index 0000000..7badda4 --- /dev/null +++ b/apps/web/src/lib/main-inbox.ts @@ -0,0 +1,24 @@ +import type { Bot, InboxItem } from "@rakazo/contracts"; + +/** One item per pending ask message, waiting takeover run, or unread bot thread. */ +export function inboxItemKey(item: InboxItem): string { + switch (item.kind) { + case "ask": + case "approval": + return `ask:${item.messageId}`; + case "takeover": + return `takeover:${item.runId}`; + case "unread": + return `unread:${item.bot.id}`; + } +} + +/** + * Changes when a bot's run status, unread flag or newest message changes, which is when the + * inbox may have changed. The shell already keeps these fresh. + */ +export function inboxRefreshSignal( + bots: readonly Pick[], +): string { + return bots.map((bot) => `${bot.id}:${bot.status}:${bot.unread}:${bot.preview}`).join("\n"); +} diff --git a/apps/web/src/pages/Shell.tsx b/apps/web/src/pages/Shell.tsx index 6d0ba02..a89b312 100644 --- a/apps/web/src/pages/Shell.tsx +++ b/apps/web/src/pages/Shell.tsx @@ -166,6 +166,7 @@ import { } from "../components/ComputersUnavailableHint"; import { ComputerUpdateProgress } from "../components/ComputerUpdateProgress"; import { DataTrustPanel } from "../components/DataTrustPanel"; +import { MainInbox } from "../components/MainInbox"; import { MessageHoverMetadata } from "../components/MessageHoverMetadata"; import { OoGroupStage, OoStage } from "../components/stage/OoStage"; import { TradingModeIndicator } from "../components/TradingModeIndicator"; @@ -4592,6 +4593,9 @@ export function ShellPage() { /> )} + {activeGroup?.kind === "main" && !showProjectTimeline ? ( + + ) : null} {recordingSkill ? (
Teaching in progress. Stop teaching before sending a new message. @@ -5965,7 +5969,8 @@ const Transcript = memo(function Transcript({ ); return ( -
+ // Clip the scroller's padding when the column leaves no room, e.g. beside the main inbox. +
; +/** A question or approval that pauses its run until the user answers. */ +export const AskBlockSchema = z.object({ + kind: z.literal("ask"), + text: z.string(), + approvalEffectId: Id.optional(), + detail: z.string().optional(), + input: z.enum(["text", "secret"]).optional(), + /** Why the secret is needed; drives field label on the masked card. */ + purpose: SecretAskPurpose.optional(), + credential: BotSecretDestination.optional(), + status: z.enum(["pending", "answered"]).optional(), + answer: z.string().optional(), + actions: z + .array( + z.object({ + id: z.string(), + label: z.string(), + outcome: z.enum(["created", "cancelled"]).optional(), + }), + ) + .optional(), +}); + export const MessageBlock = z.discriminatedUnion("kind", [ z.object({ kind: z.literal("text"), text: z.string() }), z.object({ kind: z.literal("card"), lines: z.array(z.object({ k: z.string(), v: z.string() })), }), - z.object({ - kind: z.literal("ask"), - text: z.string(), - approvalEffectId: Id.optional(), - detail: z.string().optional(), - input: z.enum(["text", "secret"]).optional(), - /** Why the secret is needed; drives field label on the masked card. */ - purpose: SecretAskPurpose.optional(), - credential: BotSecretDestination.optional(), - status: z.enum(["pending", "answered"]).optional(), - answer: z.string().optional(), - actions: z - .array( - z.object({ - id: z.string(), - label: z.string(), - outcome: z.enum(["created", "cancelled"]).optional(), - }), - ) - .optional(), - }), + AskBlockSchema, z.object({ kind: z.literal("choice"), question: z.string(), diff --git a/packages/contracts/src/inbox.ts b/packages/contracts/src/inbox.ts new file mode 100644 index 0000000..7894cd2 --- /dev/null +++ b/packages/contracts/src/inbox.ts @@ -0,0 +1,50 @@ +import * as z from "zod"; +import { AskBlockSchema } from "./events.js"; +import { Id } from "./ids.js"; + +/** The thread an item belongs to, in the shape the thread procedures accept. */ +export const InboxThreadTargetSchema = z.union([ + z.object({ botId: Id }), + z.object({ groupId: Id }), +]); +export type InboxThreadTarget = z.infer; + +const InboxItemBase = z.object({ + /** The bot that needs the user, shown as the sender. */ + bot: z.object({ id: Id, name: z.string(), color: z.string() }), + target: InboxThreadTargetSchema, + createdAt: z.string(), +}); + +const PendingAskItem = InboxItemBase.extend({ + runId: Id, + /** The bot message that holds the pending ask; answer it with threads.answer. */ + messageId: Id, + ask: AskBlockSchema, +}); + +/** + * What a bot needs from the user outside the main conversation: an answer, an approval, the + * computer screen, or a look at an unread thread. + */ +export const InboxItemSchema = z.discriminatedUnion("kind", [ + PendingAskItem.extend({ kind: z.literal("ask") }), + PendingAskItem.extend({ kind: z.literal("approval") }), + InboxItemBase.extend({ + kind: z.literal("takeover"), + runId: Id, + messageId: Id.nullable(), + /** Why the bot needs the screen. */ + text: z.string(), + }), + InboxItemBase.extend({ + kind: z.literal("unread"), + messageId: Id.nullable(), + /** Preview of the newest bot message. */ + text: z.string(), + }), +]); +export type InboxItem = z.infer; + +export const InboxListOutputSchema = z.object({ items: z.array(InboxItemSchema) }); +export type InboxListOutput = z.infer; diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index bd5ffd8..401c5c1 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -5,6 +5,7 @@ export * from "./desktop.js"; export * from "./domain.js"; export * from "./events.js"; export * from "./ids.js"; +export * from "./inbox.js"; export * from "./integration-settings.js"; export * from "./local-settings.js"; export * from "./mcp.js"; diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index b3a7091..179909b 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -76,6 +76,7 @@ import { } from "./domain.js"; import { ProductEventSchema } from "./events.js"; import { Id, IsoDate } from "./ids.js"; +import { InboxListOutputSchema } from "./inbox.js"; import { IntegrationProviderConfigSchema, IntegrationSetupStateSchema, @@ -784,6 +785,10 @@ export const appContract = { runs: { list: oc.input(z.object({ filter: z.enum(["active", "recent"]) })).output(RunsListOutputSchema), }, + /** Asks, approvals, takeovers and unread threads outside the main conversation. */ + inbox: { + list: oc.output(InboxListOutputSchema), + }, voice: { catalog: oc.output(z.array(VoiceCatalogEntrySchema)), status: oc.output(VoiceStatusSchema), diff --git a/packages/testkit/src/authorization.test.ts b/packages/testkit/src/authorization.test.ts index 6fe0dd9..6957761 100644 --- a/packages/testkit/src/authorization.test.ts +++ b/packages/testkit/src/authorization.test.ts @@ -6,6 +6,7 @@ import type { AiConsentStatus, appContract, Space, SpaceNavigation } from "@raka import { claimEmptySpaceDeletionForMember, computerHomeKey, + createThreadMessage, deleteEmptySpaceForMember, releaseSpaceDeletionClaim, renewSpaceDeletionClaim, @@ -184,6 +185,7 @@ describeWithDatabase("API authorization and resource isolation", () => { ["export/bot", { botId: "missing-bot" }], ["notifications/registerPush", { token: "ExponentPushToken[not-real]" }], ["search/query", { q: "anything" }], + ["inbox/list"], ["voice/catalog"], ["voice/status"], ["voice/credentials"], @@ -305,6 +307,13 @@ describeWithDatabase("API authorization and resource isolation", () => { trigger: "user", }, }); + await createThreadMessage(handles.prisma, { + threadId: ownerThread.id, + role: "bot", + botId: ownerBot.id, + runId: ownerRun.id, + blocks: [{ kind: "ask", text: "Owner-only question", status: "pending" }], + }); const botIdCalls: Array<[string, unknown]> = [ ["bots/get", { botId: ownerBot.id }], @@ -438,6 +447,13 @@ describeWithDatabase("API authorization and resource isolation", () => { expect( await rpc<{ hits: unknown[] }>(app, intruder, "search/query", { q: ownerBot.name }), ).toEqual({ hits: [] }); + expect( + (await rpc<{ items: Array<{ runId?: string }> }>(app, owner, "inbox/list")).items, + ).toContainEqual(expect.objectContaining({ kind: "ask", runId: ownerRun.id })); + const intruderInbox = JSON.stringify(await rpc(app, intruder, "inbox/list")); + expect(intruderInbox).not.toContain(ownerBot.id); + expect(intruderInbox).not.toContain("Owner-only question"); + await expectDenied(app, intruder, "inbox/list", {}, ownerActor.spaceId); // These endpoints are deliberately idempotent for unknown IDs. Success must not mutate // a row in a different workspace or disclose whether it exists. @@ -521,6 +537,68 @@ describeWithDatabase("API authorization and resource isolation", () => { expect(rejoined[0]?.role).toBe("member"); }); + it("keeps each shared-space member's inbox to their own bots", async () => { + const owner = await signup(app, `inbox-owner-${stamp}@rakazo.test`, "Inbox Owner"); + const member = await signup(app, `inbox-member-${stamp}@rakazo.test`, "Inbox Member"); + const ownerActor = await rpc(app, owner, "me"); + const memberActor = await rpc(app, member, "me"); + const sharedSpaceId = ownerActor.spaceId; + const sharedRow = await handles.prisma.space.findUniqueOrThrow({ + where: { id: sharedSpaceId }, + select: { organizationId: true }, + }); + await joinMemberOnSpace(handles.prisma, { + id: `inbox-org-member-${stamp}`, + organizationId: sharedRow.organizationId, + spaceId: sharedSpaceId, + userId: memberActor.userId, + }); + const ownerBot = await rpc( + app, + owner, + "bots/create", + botInput("Owner Inbox Bot"), + sharedSpaceId, + ); + const thread = await handles.prisma.thread.findUniqueOrThrow({ + where: { botId: ownerBot.id }, + }); + const task = await handles.prisma.task.create({ + data: { + spaceId: sharedSpaceId, + userId: ownerActor.userId, + botId: ownerBot.id, + threadId: thread.id, + prompt: "owner prompt", + status: "waiting_input", + }, + }); + const run = await handles.prisma.run.create({ + data: { + spaceId: sharedSpaceId, + userId: ownerActor.userId, + botId: ownerBot.id, + threadId: thread.id, + taskId: task.id, + status: "waiting_input", + trigger: "user", + }, + }); + await createThreadMessage(handles.prisma, { + threadId: thread.id, + role: "bot", + botId: ownerBot.id, + runId: run.id, + blocks: [{ kind: "ask", text: "Owner-only question", status: "pending" }], + }); + + expect( + (await rpc<{ items: Array<{ runId?: string }> }>(app, owner, "inbox/list", {}, sharedSpaceId)) + .items, + ).toContainEqual(expect.objectContaining({ kind: "ask", runId: run.id })); + expect(await rpc(app, member, "inbox/list", {}, sharedSpaceId)).toEqual({ items: [] }); + }); + it("prevents one shared-space member from hijacking another member's Team Computer takeover", async () => { const owner = await signup(app, `takeover-owner-${stamp}@rakazo.test`, "Takeover Owner"); const member = await signup(app, `takeover-member-${stamp}@rakazo.test`, "Takeover Member"); diff --git a/packages/testkit/src/cli/harness.ts b/packages/testkit/src/cli/harness.ts index 272712a..8a7357b 100644 --- a/packages/testkit/src/cli/harness.ts +++ b/packages/testkit/src/cli/harness.ts @@ -100,6 +100,7 @@ async function main() { "packages/testkit/src/attachments.test.ts", "packages/testkit/src/voice.test.ts", "packages/testkit/src/search.test.ts", + "packages/testkit/src/inbox.test.ts", "packages/testkit/src/executor-lifecycle.test.ts", "packages/testkit/src/connections.test.ts", "packages/testkit/src/bot-secrets.test.ts", diff --git a/packages/testkit/src/inbox.test.ts b/packages/testkit/src/inbox.test.ts new file mode 100644 index 0000000..2fe3d09 --- /dev/null +++ b/packages/testkit/src/inbox.test.ts @@ -0,0 +1,317 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import type { InboxItem, MessageBlock, ThreadSnapshot } from "@rakazo/contracts"; +import { + addMainConversationMembers, + createThreadMessage, + ensureMainConversation, + type PrismaClient, +} from "@rakazo/db"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { sessionCookieHeader } from "./index.js"; + +type App = { request: (input: string, init?: RequestInit) => Response | Promise }; + +process.env.WAKEUP_DRIVER = "memory"; +process.env.SANDBOX_PROVIDER = "fake"; +process.env.AGENT_RUNTIME = "scripted"; + +const hasDb = process.env.VERIFY_DATABASE === "1" && Boolean(process.env.DATABASE_URL); +const describeInbox = hasDb ? describe : describe.skip; + +describeInbox("inbox.list", () => { + let app: App; + let prisma: PrismaClient; + let stop: () => Promise; + const stamp = `${Date.now()}-${Math.random().toString(36).slice(2)}`; + const dataDir = mkdtempSync(path.join(tmpdir(), "rakazo-inbox-")); + + beforeAll(async () => { + const { createApp } = await import("../../../apps/api/src/app.ts"); + const handles = await createApp({ + databaseUrl: process.env.DATABASE_URL!, + dataDir, + sandboxProvider: "fake", + agentRuntime: "scripted", + wakeupDriver: "memory", + }); + app = handles.app; + prisma = handles.prisma; + stop = handles.stop; + }); + + afterAll(async () => { + await stop?.(); + rmSync(dataDir, { recursive: true, force: true }); + }); + + it("answers a specialist's question through the item and then shows its unread reply", async () => { + const cookie = await signup(app, `inbox-flow-${stamp}@rakazo.test`, "Inbox Flow"); + const scout = await createBot(app, cookie, "Scout"); + + await rpc(app, cookie, "threads/send", { botId: scout.id, text: "ask me which city to use" }); + const waiting = await waitForThread( + app, + cookie, + scout.id, + (snapshot) => snapshot.run?.status === "waiting_input", + ); + + const asked = await listInbox(app, cookie); + expect(asked).toEqual([ + expect.objectContaining({ + kind: "ask", + bot: { id: scout.id, name: "Scout", color: scout.color }, + target: { botId: scout.id }, + runId: waiting.run?.id, + ask: expect.objectContaining({ text: "Which city should I use?", status: "pending" }), + }), + ]); + const item = asked[0]; + if (item?.kind !== "ask") throw new Error("expected an ask item"); + + await rpc(app, cookie, "threads/answer", { + ...item.target, + runId: item.runId, + messageId: item.messageId, + answer: "Paris", + }); + // A finished run leaves the snapshot. + await waitForThread(app, cookie, scout.id, (snapshot) => !snapshot.run); + + const answered = await listInbox(app, cookie); + expect(answered).toEqual([ + expect.objectContaining({ + kind: "unread", + bot: expect.objectContaining({ id: scout.id }), + target: { botId: scout.id }, + text: expect.stringContaining("Paris"), + }), + ]); + + await rpc(app, cookie, "threads/markRead", { botId: scout.id }); + expect(await listInbox(app, cookie)).toEqual([]); + }); + + it("collects approvals, takeovers and unread threads of this space outside the main conversation", async () => { + const cookie = await signup(app, `inbox-mixed-${stamp}@rakazo.test`, "Inbox Mixed"); + const me = await rpc<{ userId: string; spaceId: string }>(app, cookie, "me"); + const scout = await createBot(app, cookie, "Scout"); + const writer = await createBot(app, cookie, "Writer"); + const pilot = await createBot(app, cookie, "Pilot"); + const retired = await createBot(app, cookie, "Retired"); + const ops = await rpc<{ id: string; threadId: string }>(app, cookie, "groups/create", { + name: "Ops", + botIds: [scout.id, writer.id], + }); + const main = await ensureMainConversation(prisma, me.userId); + await addMainConversationMembers(prisma, { userId: me.userId, botIds: [scout.id] }); + + const approval = await seedWaitingRun(prisma, { + botId: writer.id, + threadId: ops.threadId, + status: "waiting_input", + blocks: [ + { + kind: "ask", + text: "Review before writing to notes", + approvalEffectId: `effect-${stamp}`, + status: "pending", + actions: [ + { id: "allow", label: "Allow once" }, + { id: "always", label: "Always allow this tool" }, + { id: "deny", label: "Deny" }, + ], + }, + ], + }); + // Already visible where the inbox lives. + await seedWaitingRun(prisma, { + botId: scout.id, + threadId: main.threadId, + status: "waiting_input", + blocks: [{ kind: "ask", text: "Asked in the main conversation", status: "pending" }], + }); + const takeover = await seedWaitingRun(prisma, { + botId: pilot.id, + threadId: pilot.threadId, + status: "waiting_takeover", + blocks: [{ kind: "computer", state: "Needs you", text: "Sign in to continue." }], + }); + await seedWaitingRun(prisma, { + botId: retired.id, + threadId: retired.threadId, + status: "waiting_input", + blocks: [{ kind: "ask", text: "Asked by an archived bot", status: "pending" }], + }); + await prisma.bot.update({ where: { id: retired.id }, data: { archivedAt: new Date() } }); + await createThreadMessage(prisma, { + threadId: scout.threadId, + role: "bot", + botId: scout.id, + blocks: [{ kind: "text", text: "The report\nis ready." }], + }); + // The main bot's own thread is not a specialist thread. + await createThreadMessage(prisma, { + threadId: (await prisma.thread.findUniqueOrThrow({ where: { botId: main.leadBotId } })).id, + role: "bot", + botId: main.leadBotId, + blocks: [{ kind: "text", text: "Main bot note" }], + }); + + const remoteSpace = await rpc<{ id: string }>(app, cookie, "spaces/create", { name: "Remote" }); + const remote = await createBot(app, cookie, "Remote", remoteSpace.id); + const remoteAsk = await seedWaitingRun(prisma, { + botId: remote.id, + threadId: remote.threadId, + status: "waiting_input", + blocks: [{ kind: "ask", text: "Asked in another space", status: "pending" }], + }); + + const items = await listInbox(app, cookie, me.spaceId); + expect(items).toEqual([ + expect.objectContaining({ + kind: "approval", + bot: expect.objectContaining({ id: writer.id, name: "Writer" }), + target: { groupId: ops.id }, + runId: approval.runId, + messageId: approval.messageId, + }), + { + kind: "takeover", + bot: { id: pilot.id, name: "Pilot", color: pilot.color }, + target: { botId: pilot.id }, + runId: takeover.runId, + messageId: takeover.messageId, + text: "Sign in to continue.", + createdAt: expect.any(String), + }, + expect.objectContaining({ + kind: "unread", + bot: expect.objectContaining({ id: scout.id }), + target: { botId: scout.id }, + text: "The report is ready.", + }), + ]); + // The Pilot thread is unread too, but its takeover already covers it. + expect(items.filter((item) => item.bot.id === pilot.id)).toHaveLength(1); + + expect(await listInbox(app, cookie, remoteSpace.id)).toEqual([ + expect.objectContaining({ + kind: "ask", + bot: expect.objectContaining({ id: remote.id }), + target: { botId: remote.id }, + runId: remoteAsk.runId, + messageId: remoteAsk.messageId, + }), + ]); + }); +}); + +async function seedWaitingRun( + prisma: PrismaClient, + input: { + botId: string; + threadId: string; + status: "waiting_input" | "waiting_takeover"; + blocks: MessageBlock[]; + }, +) { + const thread = await prisma.thread.findUniqueOrThrow({ where: { id: input.threadId } }); + const task = await prisma.task.create({ + data: { + spaceId: thread.spaceId, + userId: thread.userId, + botId: input.botId, + threadId: thread.id, + prompt: "inbox fixture", + status: input.status, + }, + }); + const run = await prisma.run.create({ + data: { + spaceId: thread.spaceId, + userId: thread.userId, + botId: input.botId, + threadId: thread.id, + taskId: task.id, + status: input.status, + trigger: "user", + }, + }); + const message = await createThreadMessage(prisma, { + threadId: thread.id, + role: "bot", + botId: input.botId, + runId: run.id, + blocks: input.blocks, + }); + return { runId: run.id, messageId: message.id }; +} + +async function createBot(app: App, cookie: string, name: string, spaceId?: string) { + return rpc<{ id: string; color: string; threadId: string }>( + app, + cookie, + "bots/create", + { name, title: name, description: "", instructions: "", notifyOnFinish: true }, + spaceId, + ); +} + +async function listInbox(app: App, cookie: string, spaceId?: string) { + return (await rpc<{ items: InboxItem[] }>(app, cookie, "inbox/list", {}, spaceId)).items; +} + +async function waitForThread( + app: App, + cookie: string, + botId: string, + done: (snapshot: ThreadSnapshot) => boolean, +): Promise { + const start = Date.now(); + let last: ThreadSnapshot | null = null; + while (Date.now() - start < 20_000) { + last = await rpc(app, cookie, "threads/get", { botId }); + if (done(last)) return last; + await new Promise((resolve) => setTimeout(resolve, 100)); + } + throw new Error(`timeout waiting for thread: ${JSON.stringify(last)}`); +} + +async function signup(app: App, email: string, name: string) { + const response = await app.request("/api/auth/sign-up/email", { + method: "POST", + headers: { "content-type": "application/json", origin: "http://127.0.0.1:5173" }, + body: JSON.stringify({ email, password: "password12", name }), + }); + if (response.status >= 400) { + throw new Error(`signup failed ${response.status}: ${await response.text()}`); + } + return sessionCookieHeader(response); +} + +async function rpc( + app: App, + cookie: string, + procedure: string, + body: unknown = {}, + spaceId?: string, +): Promise { + const response = await app.request(`/rpc/${procedure}`, { + method: "POST", + headers: { + "content-type": "application/json", + cookie, + origin: "http://127.0.0.1:5173", + ...(spaceId ? { "x-rakazo-space-id": spaceId } : {}), + }, + body: JSON.stringify({ json: body }), + }); + const payload = (await response.json()) as { json?: T; error?: { message?: string } }; + if (response.status >= 400 || payload.error) { + throw new Error(`${procedure} ${response.status}: ${payload.error?.message ?? "failed"}`); + } + return payload.json as T; +}