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 every 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: "ask", bot: expect.objectContaining({ id: remote.id }), target: { botId: remote.id }, space: { id: remoteSpace.id, name: "Remote" }, runId: remoteAsk.runId, messageId: remoteAsk.messageId, }), 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); // Items of the current space carry no space. expect(items.filter((item) => item.space).map((item) => item.bot.id)).toEqual([remote.id]); // From the other space, the same items with the sides swapped, plus the ask in the home // main conversation: Remote's own main conversation does not show it. const fromRemote = await listInbox(app, cookie, remoteSpace.id); expect(fromRemote.map((item) => [item.kind, item.space?.id ?? null])).toEqual([ ["approval", me.spaceId], ["ask", me.spaceId], ["takeover", me.spaceId], ["ask", null], ["unread", me.spaceId], ]); expect(fromRemote[1]).toMatchObject({ bot: expect.objectContaining({ id: scout.id }), target: { groupId: main.groupId }, }); }); it("answers an item from another space only in that space and only for its owner", async () => { const cookie = await signup(app, `inbox-cross-${stamp}@rakazo.test`, "Inbox Cross"); const intruder = await signup(app, `inbox-intruder-${stamp}@rakazo.test`, "Inbox Intruder"); const me = await rpc<{ spaceId: string }>(app, cookie, "me"); const side = await rpc<{ id: string }>(app, cookie, "spaces/create", { name: "Side" }); const scout = await createBot(app, cookie, "Scout", side.id); await rpc( app, cookie, "threads/send", { botId: scout.id, text: "ask me which city to use" }, side.id, ); await waitForThread( app, cookie, scout.id, (snapshot) => snapshot.run?.status === "waiting_input", side.id, ); const [item] = await listInbox(app, cookie, me.spaceId); if (item?.kind !== "ask") throw new Error("expected an ask item"); expect(item).toMatchObject({ space: { id: side.id, name: "Side" }, target: { botId: scout.id }, }); const answer = { ...item.target, runId: item.runId, messageId: item.messageId, answer: "Paris", }; // The thread is not in the current space, and the intruder belongs to neither space. await expect(rpc(app, cookie, "threads/answer", answer, me.spaceId)).rejects.toThrow(); await expect(rpc(app, intruder, "threads/answer", answer, side.id)).rejects.toThrow(); await expect(rpc(app, intruder, "threads/answer", answer)).rejects.toThrow(); expect(await listInbox(app, intruder)).toEqual([]); expect((await listInbox(app, cookie, me.spaceId)).map((entry) => entry.kind)).toEqual(["ask"]); await rpc(app, cookie, "threads/answer", answer, item.space?.id); await waitForThread(app, cookie, scout.id, (snapshot) => !snapshot.run, side.id); expect(await listInbox(app, cookie, me.spaceId)).toEqual([ expect.objectContaining({ kind: "unread", target: { botId: scout.id }, space: { id: side.id, name: "Side" }, text: expect.stringContaining("Paris"), }), ]); await rpc(app, cookie, "threads/markRead", { botId: scout.id }, side.id); expect(await listInbox(app, cookie, me.spaceId)).toEqual([]); }); }); 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, spaceId?: string, ): Promise { const start = Date.now(); let last: ThreadSnapshot | null = null; while (Date.now() - start < 20_000) { last = await rpc(app, cookie, "threads/get", { botId }, spaceId); 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; }