From 408db283307db42380230411fe72420efb708434 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 2 Oct 2026 22:37:08 +0000 Subject: [PATCH] feat(inbox): show what specialists need from the user in the main conversation Specialists ask questions, request approvals and the computer screen in their own threads, and the user only finds out by opening each thread. With the main conversation as the single entry point, those requests have to reach the user there, with the specialist as the visible sender. - inbox.list returns, for the current space and the user's own bots, pending asks and approvals (bot messages with a pending ask block whose run waits for input), takeover requests, and unread specialist threads. Asks already in the main conversation are left out; takeovers stay because their controls live in the bot's own thread. Each item carries the bot, the thread target, run and message ids, and the ask block for asks and approvals. - The main conversation shows the items above the composer. Asks and approvals are answered in place through threads.answer with the source thread target; takeovers and unread threads open the bot's thread. The list refreshes when the bots' status, unread state or newest message changes, every 15 seconds, and after an answer. - The inbox shrinks and scrolls so it never pushes the composer out of view. The transcript now clips its scroller: with no room left in the column, the scroller's padding used to spill over whatever sat below it. - The ask block schema is extracted from MessageBlock so the inbox contract reuses it, and AskCard accepts a className. Tests: unit tests for the item builder and the component, a Postgres suite for the query (added to the harness list), inbox cases in the authorization suite, and an E2E spec in which a specialist asks in its own thread and the user answers from the main conversation. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01MRtGG9ARfqqhma3t85Ro2u --- apps/api/src/inbox.test.ts | 230 +++++++++++++++ apps/api/src/inbox.ts | 236 +++++++++++++++ apps/api/src/router.ts | 6 + apps/api/src/runs.ts | 2 +- apps/web/e2e/inbox.spec.ts | 85 ++++++ apps/web/src/components/AskCard.tsx | 6 +- apps/web/src/components/MainInbox.test.tsx | 219 ++++++++++++++ apps/web/src/components/MainInbox.tsx | 169 +++++++++++ apps/web/src/lib/main-inbox.test.ts | 40 +++ apps/web/src/lib/main-inbox.ts | 24 ++ apps/web/src/pages/Shell.tsx | 7 +- packages/contracts/src/events.ts | 45 +-- packages/contracts/src/inbox.ts | 50 ++++ packages/contracts/src/index.ts | 1 + packages/contracts/src/rpc.ts | 5 + packages/testkit/src/authorization.test.ts | 78 +++++ packages/testkit/src/cli/harness.ts | 1 + packages/testkit/src/inbox.test.ts | 317 +++++++++++++++++++++ 18 files changed, 1496 insertions(+), 25 deletions(-) create mode 100644 apps/api/src/inbox.test.ts create mode 100644 apps/api/src/inbox.ts create mode 100644 apps/web/e2e/inbox.spec.ts create mode 100644 apps/web/src/components/MainInbox.test.tsx create mode 100644 apps/web/src/components/MainInbox.tsx create mode 100644 apps/web/src/lib/main-inbox.test.ts create mode 100644 apps/web/src/lib/main-inbox.ts create mode 100644 packages/contracts/src/inbox.ts create mode 100644 packages/testkit/src/inbox.test.ts 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; +}