import type { SandboxProvider } from "@rakazo/adapter-kit"; import type { Actor } from "@rakazo/contracts"; import type { PrismaClient } from "@rakazo/db"; import { describe, expect, it, vi } from "vitest"; import { cancelSupersededQueuedRuns, followUpGroupThreadMessage, groupDefaultBotId, reactToThreadMessage, resolveGroupFollowUpBotId, sendThreadMessage, stopThreadRuns, type ThreadTarget, threadHead, threadSnapshot, } from "./thread-target.js"; describe("threadHead", () => { it("returns the durable cursor without loading a snapshot", async () => { const findFirst = vi.fn().mockResolvedValue({ seq: 12 }); const prisma = { event: { findFirst } } as unknown as PrismaClient; const target = { threadId: "thread-1" } as ThreadTarget; await expect(threadHead(prisma, target)).resolves.toEqual({ threadId: "thread-1", cursor: 12, }); expect(findFirst).toHaveBeenCalledWith({ where: { threadId: "thread-1" }, orderBy: { seq: "desc" }, select: { seq: true }, }); }); }); describe("queued run supersession", () => { it("only cancels queued runs started by user messages or reactions", async () => { const tx = { run: { findMany: vi.fn().mockResolvedValue([{ id: "run-old", taskId: "task-old" }]), updateMany: vi.fn(), }, task: { updateMany: vi.fn() }, }; await cancelSupersededQueuedRuns(tx as never, { threadId: "thread-1", botIds: ["bot-1"], keepRunIds: ["run-new"], }); expect(tx.run.findMany).toHaveBeenCalledWith( expect.objectContaining({ where: expect.objectContaining({ OR: [{ trigger: "user", sourceMessage: { role: "user" } }, { trigger: "reaction" }], }), }), ); expect(tx.run.updateMany).toHaveBeenCalledWith( expect.objectContaining({ where: { id: { in: ["run-old"] } } }), ); expect(tx.task.updateMany).toHaveBeenCalledWith({ where: { id: { in: ["task-old"] } }, data: { status: "cancelled" }, }); }); }); describe("reaction messages", () => { it("appends repeated reactions as quiet replies and deduplicates retries", async () => { const messages = new Map(); let messageSeq = 0; let eventSeq = 0; const tx = { $queryRaw: vi.fn().mockResolvedValue([]), message: { findFirst: vi.fn().mockResolvedValue({ id: "parent" }), findUnique: vi.fn( async ({ where }: { where: { threadId_clientNonce: { clientNonce: string } } }) => messages.get(where.threadId_clientNonce.clientNonce) ?? null, ), create: vi.fn(async ({ data }: { data: { clientNonce: string } }) => { const message = { id: `reaction-${messages.size}`, ...data }; messages.set(data.clientNonce, message); return message; }), }, thread: { update: vi.fn(async ({ data }: { data: { nextMessageSeq?: unknown } }) => data.nextMessageSeq ? { nextMessageSeq: ++messageSeq } : { nextEventSeq: ++eventSeq }, ), }, event: { create: vi.fn(async ({ data }: { data: Record }) => ({ id: `event-${eventSeq}`, createdAt: new Date(), ...data, })), }, task: { create: vi.fn() }, run: { create: vi.fn() }, }; const prisma = { $transaction: vi.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), } as unknown as PrismaClient; const actor = { spaceId: "space-1", userId: "user-1" } as Actor; const target = { kind: "bot", botId: "bot-1", threadId: "thread-1" } as ThreadTarget; for (const [clientNonce, reaction] of [ ["first", "❤️"], ["second", "👍"], ["third", "❤️"], ["third", "❤️"], ] as const) { await reactToThreadMessage({ prisma }, actor, target, { messageId: "parent", reaction, clientNonce, }); } expect(tx.message.create).toHaveBeenCalledTimes(3); expect(tx.event.create).toHaveBeenCalledTimes(3); expect(tx.message.create).toHaveBeenLastCalledWith({ data: expect.objectContaining({ role: "user", blocks: [{ kind: "text", text: "❤️" }], replyToMessageId: "parent", clientNonce: "third", }), }); expect(tx.event.create).toHaveBeenLastCalledWith({ data: expect.objectContaining({ type: "thread.message.created", payload: { messageId: "reaction-2", role: "user", blocks: [{ kind: "text", text: "❤️" }], replyToMessageId: "parent", }, }), }); expect(tx.task.create).not.toHaveBeenCalled(); expect(tx.run.create).not.toHaveBeenCalled(); expect(tx.message.findFirst).toHaveBeenCalledWith({ where: { id: "parent", threadId: "thread-1" }, select: { id: true }, }); tx.message.findFirst.mockResolvedValueOnce(null); await expect( reactToThreadMessage({ prisma }, actor, target, { messageId: "elsewhere", reaction: "❤️", clientNonce: "fourth", }), ).rejects.toThrow(); expect(tx.message.create).toHaveBeenCalledTimes(3); }); }); describe("threadSnapshot", () => { it("reloads tool-only live messages for an active run", async () => { const run = { id: "run-1", botId: "bot-1", threadId: "thread-1", taskId: "task-1", status: "running", trigger: "user", modelProvider: null, modelId: null, error: null, startedAt: null, completedAt: null, createdAt: new Date("2026-08-23T00:00:00.000Z"), }; const findManyEvents = vi.fn().mockResolvedValue([ { id: "event-1", threadId: "thread-1", botId: "bot-1", seq: 4, type: "agent.tool.called", runId: "run-1", payload: { name: "SLACK_FIND_CHANNELS" }, createdAt: new Date("2026-08-23T00:00:00.000Z"), }, ]); const tx = { $queryRaw: threadQueries(), message: { findMany: vi.fn().mockResolvedValue([]) }, event: { findFirst: vi.fn().mockResolvedValue({ seq: 4 }), findMany: findManyEvents, }, run: { findFirst: botRunFindFirst([run]) }, }; const prisma = { $transaction: vi.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), } as unknown as PrismaClient; const target = { kind: "bot", botId: "bot-1", threadId: "thread-1", bot: { computer: null }, } as ThreadTarget; const snapshot = await threadSnapshot({ prisma }, target); expect(tx.$queryRaw).toHaveBeenCalledWith( expect.arrayContaining([expect.stringContaining("FOR SHARE")]), "thread-1", ); expect(findManyEvents).toHaveBeenCalledWith( expect.objectContaining({ where: expect.objectContaining({ type: { in: [ "thread.message.created", "thread.progress", "thread.subagent", "agent.tool.called", "agent.tool.completed", ], }, }), }), ); expect(snapshot.messages).toEqual([ expect.objectContaining({ id: "progress:run-1", botId: "bot-1", blocks: [ { kind: "steps", steps: [{ label: "Slack find channels", count: 1 }], }, ], }), ]); }); it("does not reload live narration that was already posted as a message", async () => { const run = { id: "run-1", botId: "bot-1", threadId: "thread-1", taskId: "task-1", status: "running", trigger: "user", modelProvider: null, modelId: null, error: null, startedAt: null, completedAt: null, createdAt: new Date("2026-08-23T00:00:00.000Z"), }; const at = new Date("2026-08-23T00:00:00.000Z"); const event = (seq: number, type: string, payload: Record) => ({ id: `event-${seq}`, threadId: "thread-1", botId: "bot-1", seq, type, runId: "run-1", payload, createdAt: at, }); const tx = { $queryRaw: threadQueries(), message: { findMany: vi.fn().mockResolvedValue([]) }, event: { findFirst: vi.fn().mockResolvedValue({ seq: 6 }), findMany: vi.fn().mockResolvedValue([ event(4, "thread.progress", { text: "Opening the page." }), event(5, "thread.message.created", { messageId: "message-1", role: "bot", blocks: [{ kind: "text", text: "Opening the page." }], }), event(6, "thread.progress", { text: "Writing the summary." }), ]), }, run: { findFirst: botRunFindFirst([run]) }, }; const prisma = { $transaction: vi.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), } as unknown as PrismaClient; const target = { kind: "bot", botId: "bot-1", threadId: "thread-1", bot: { computer: null }, } as ThreadTarget; const snapshot = await threadSnapshot({ prisma }, target); expect(snapshot.messages).toEqual([ expect.objectContaining({ id: "progress:run-1", blocks: [{ kind: "progress", text: "Writing the summary." }], }), ]); }); it("returns the latest failed run so the client can show its error", async () => { const run = { id: "run-failed", botId: "bot-1", threadId: "thread-1", taskId: "task-1", status: "failed", trigger: "user", modelProvider: "openrouter", modelId: "openrouter/unknown", error: "Provider is not configured: openrouter", startedAt: null, completedAt: new Date("2026-08-23T00:00:01.000Z"), createdAt: new Date("2026-08-23T00:00:00.000Z"), }; const findManyEvents = vi.fn(); const findFirstRun = botRunFindFirst([run]); const tx = { $queryRaw: threadQueries(), message: { findMany: vi.fn().mockResolvedValue([]) }, event: { findFirst: vi.fn().mockResolvedValue(null), findMany: findManyEvents, }, run: { findFirst: findFirstRun }, }; const prisma = { $transaction: vi.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), } as unknown as PrismaClient; const target = { kind: "bot", botId: "bot-1", threadId: "thread-1", bot: { computer: null }, } as ThreadTarget; const snapshot = await threadSnapshot({ prisma }, target); expect(findFirstRun).toHaveBeenCalledWith( expect.objectContaining({ where: expect.objectContaining({ botId: "bot-1", threadId: "thread-1", trigger: { not: "bot_message" }, status: { in: ["queued", "leased", "running", "waiting_input", "waiting_takeover", "failed"], }, }), }), ); expect(findFirstRun).toHaveBeenCalledWith( expect.objectContaining({ where: { botId: "bot-1", threadId: "thread-1", status: { in: ["waiting_input", "waiting_takeover"] }, }, }), ); expect(snapshot.run).toEqual( expect.objectContaining({ id: "run-failed", status: "failed", error: "Provider is not configured: openrouter", }), ); expect(findManyEvents).not.toHaveBeenCalled(); }); it("prefers a waiting peer ask over a concurrent user run", async () => { const waitingPeer = { id: "run-peer-waiting", botId: "bot-1", threadId: "thread-1", taskId: "task-peer", status: "waiting_input", trigger: "bot_message", modelProvider: null, modelId: null, error: null, startedAt: new Date("2026-08-23T00:00:02.000Z"), completedAt: null, createdAt: new Date("2026-08-23T00:00:02.000Z"), }; const olderUser = { id: "run-user", botId: "bot-1", threadId: "thread-1", taskId: "task-user", status: "running", trigger: "user", modelProvider: null, modelId: null, error: null, startedAt: new Date("2026-08-23T00:00:01.000Z"), completedAt: null, createdAt: new Date("2026-08-23T00:00:01.000Z"), }; const snapshot = await threadSnapshot( { prisma: { $transaction: vi.fn(async (callback: (client: unknown) => unknown) => callback({ $queryRaw: threadQueries(), message: { findMany: vi.fn().mockResolvedValue([]) }, event: { findFirst: vi.fn().mockResolvedValue(null), findMany: vi.fn().mockResolvedValue([]), }, run: { findFirst: botRunFindFirst([waitingPeer, olderUser]) }, }), ), } as unknown as PrismaClient, }, { kind: "bot", botId: "bot-1", threadId: "thread-1", bot: { computer: null }, } as ThreadTarget, ); expect(snapshot.run).toEqual( expect.objectContaining({ id: "run-peer-waiting", status: "waiting_input" }), ); }); it("drops a failed run once a newer run has finished", async () => { const failed = { id: "run-failed", botId: "bot-1", threadId: "thread-1", taskId: "task-1", status: "failed", trigger: "user", modelProvider: "openrouter", modelId: "openrouter/unknown", error: "This operation was aborted", startedAt: null, completedAt: new Date("2026-08-23T00:00:01.000Z"), createdAt: new Date("2026-08-23T00:00:00.000Z"), }; const completed = { id: "run-completed", botId: "bot-1", threadId: "thread-1", taskId: "task-2", status: "completed", trigger: "user", modelProvider: null, modelId: null, error: null, startedAt: null, completedAt: new Date("2026-08-23T00:00:03.000Z"), createdAt: new Date("2026-08-23T00:00:02.000Z"), }; const findFirstRun = botRunFindFirst([failed, completed]); const tx = { $queryRaw: threadQueries(), message: { findMany: vi.fn().mockResolvedValue([]) }, event: { findFirst: vi.fn().mockResolvedValue(null), findMany: vi.fn(), }, run: { findFirst: findFirstRun }, }; const prisma = { $transaction: vi.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), } as unknown as PrismaClient; const target = { kind: "bot", botId: "bot-1", threadId: "thread-1", bot: { computer: null }, } as ThreadTarget; const snapshot = await threadSnapshot({ prisma }, target); expect(findFirstRun).toHaveBeenCalledWith( expect.objectContaining({ where: expect.objectContaining({ trigger: { not: "bot_message" }, status: { in: ["failed", "completed", "cancelled"] }, }), orderBy: [{ createdAt: "desc" }, { id: "desc" }], }), ); expect(snapshot.run).toBeNull(); }); it("does not return a cancelled or completed run", async () => { const findManyEvents = vi.fn(); const findFirstRun = botRunFindFirst([]); const tx = { $queryRaw: threadQueries(), message: { findMany: vi.fn().mockResolvedValue([]) }, event: { findFirst: vi.fn().mockResolvedValue(null), findMany: findManyEvents, }, run: { findFirst: findFirstRun }, }; const prisma = { $transaction: vi.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), } as unknown as PrismaClient; const target = { kind: "bot", botId: "bot-1", threadId: "thread-1", bot: { computer: null }, } as ThreadTarget; const snapshot = await threadSnapshot({ prisma }, target); expect(findFirstRun).toHaveBeenCalledWith( expect.objectContaining({ where: expect.objectContaining({ status: { in: ["queued", "leased", "running", "waiting_input", "waiting_takeover", "failed"], }, }), }), ); expect(snapshot.run).toBeNull(); expect(findManyEvents).not.toHaveBeenCalled(); }); it("returns a group's latest failed run so a refresh keeps its error", async () => { const run = { id: "run-failed", botId: "bot-2", threadId: "thread-1", taskId: "task-1", status: "failed", trigger: "user", modelProvider: "openrouter", modelId: "openrouter/unknown", error: "member exploded", startedAt: null, completedAt: new Date("2026-08-23T00:00:01.000Z"), createdAt: new Date("2026-08-23T00:00:00.000Z"), }; const findManyRuns = groupRunFindMany({ terminals: [run] }); const snapshot = await threadSnapshot({ prisma: groupPrisma(findManyRuns) }, groupTarget()); expect(findManyRuns).toHaveBeenCalledWith( expect.objectContaining({ where: { threadId: "thread-1", trigger: { not: "bot_message" }, status: { in: ["failed", "completed", "cancelled"] }, }, orderBy: [{ updatedAt: "desc" }, { id: "desc" }], take: 50, }), ); expect(findManyRuns).toHaveBeenCalledWith( expect.objectContaining({ where: { threadId: "thread-1", status: { in: ["queued", "leased", "running", "waiting_input", "waiting_takeover"] }, OR: [ { trigger: { not: "bot_message" } }, { status: { in: ["waiting_input", "waiting_takeover"] } }, ], }, }), ); expect(snapshot.run).toEqual( expect.objectContaining({ id: "run-failed", status: "failed", error: "member exploded" }), ); expect(snapshot.activeRuns).toEqual([]); }); it("omits peer bot_message runs from group activeRuns and displayed terminal run", async () => { const peerActive = { id: "run-peer-active", botId: "bot-a", threadId: "thread-1", taskId: "task-peer", status: "running", trigger: "bot_message", modelProvider: null, modelId: null, error: null, startedAt: new Date("2026-08-23T00:00:05.000Z"), completedAt: null, createdAt: new Date("2026-08-23T00:00:05.000Z"), }; const peerFailed = { id: "run-peer-failed", botId: "bot-b", threadId: "thread-1", taskId: "task-peer-fail", status: "failed", trigger: "bot_message", modelProvider: null, modelId: null, error: "peer exploded", startedAt: new Date("2026-08-23T00:00:01.000Z"), completedAt: new Date("2026-08-23T00:00:02.000Z"), createdAt: new Date("2026-08-23T00:00:01.000Z"), }; const findManyRuns = groupRunFindMany({ active: [peerActive], terminals: [peerFailed], }); const snapshot = await threadSnapshot({ prisma: groupPrisma(findManyRuns) }, groupTarget()); expect(snapshot.activeRuns).toEqual([]); expect(snapshot.run).toBeNull(); expect(findManyRuns).toHaveBeenCalledWith( expect.objectContaining({ where: expect.objectContaining({ OR: [ { trigger: { not: "bot_message" } }, { status: { in: ["waiting_input", "waiting_takeover"] } }, ], status: { in: ["queued", "leased", "running", "waiting_input", "waiting_takeover"] }, }), }), ); expect(findManyRuns).toHaveBeenCalledWith( expect.objectContaining({ where: expect.objectContaining({ trigger: { not: "bot_message" }, status: { in: ["failed", "completed", "cancelled"] }, }), }), ); }); it("includes waiting peer bot_message runs in group activeRuns", async () => { const peerWaiting = { id: "run-peer-waiting", botId: "bot-a", threadId: "thread-1", taskId: "task-peer", status: "waiting_input", trigger: "bot_message", modelProvider: null, modelId: null, error: null, startedAt: new Date("2026-08-23T00:00:05.000Z"), completedAt: null, createdAt: new Date("2026-08-23T00:00:05.000Z"), }; const findManyRuns = groupRunFindMany({ active: [peerWaiting] }); const snapshot = await threadSnapshot({ prisma: groupPrisma(findManyRuns) }, groupTarget()); expect(snapshot.activeRuns).toEqual([ expect.objectContaining({ id: "run-peer-waiting", status: "waiting_input" }), ]); }); it("keeps a waiting peer ask as the primary run even when a newer busy run exists", async () => { // Real DB order is createdAt desc, so the newer busy run for bot-b comes // first here, ahead of the older waiting peer run for bot-a. const newerBusy = { id: "run-newer-busy", botId: "bot-b", threadId: "thread-1", taskId: "task-busy", status: "running", trigger: "user", modelProvider: null, modelId: null, error: null, startedAt: new Date("2026-08-23T00:00:10.000Z"), completedAt: null, createdAt: new Date("2026-08-23T00:00:10.000Z"), }; const olderWaiting = { id: "run-older-waiting", botId: "bot-a", threadId: "thread-1", taskId: "task-peer", status: "waiting_input", trigger: "bot_message", modelProvider: null, modelId: null, error: null, startedAt: new Date("2026-08-23T00:00:05.000Z"), completedAt: null, createdAt: new Date("2026-08-23T00:00:05.000Z"), }; const findManyRuns = groupRunFindMany({ active: [newerBusy, olderWaiting] }); const snapshot = await threadSnapshot({ prisma: groupPrisma(findManyRuns) }, groupTarget()); expect(snapshot.run).toEqual(expect.objectContaining({ id: "run-older-waiting" })); // activeRuns is unaffected by which one is chosen as primary. expect(snapshot.activeRuns.map((run) => run.id)).toEqual([ "run-newer-busy", "run-older-waiting", ]); }); it("does not revive an older group failure after a newer run completed", async () => { const failed = { id: "run-old-failed", botId: "bot-2", threadId: "thread-1", taskId: "task-1", status: "failed", trigger: "user", modelProvider: null, modelId: null, error: "old failure", startedAt: null, completedAt: new Date("2026-08-23T00:00:01.000Z"), createdAt: new Date("2026-08-23T00:00:00.000Z"), }; const completed = { id: "run-newer-completed", botId: "bot-1", threadId: "thread-1", taskId: "task-2", status: "completed", trigger: "user", modelProvider: null, modelId: null, error: null, startedAt: new Date("2026-08-23T00:00:02.000Z"), completedAt: new Date("2026-08-23T00:00:04.000Z"), createdAt: new Date("2026-08-23T00:00:02.000Z"), }; const snapshot = await threadSnapshot( { prisma: groupPrisma(groupRunFindMany({ terminals: [completed, failed] })) }, groupTarget(), ); expect(snapshot.run).toBeNull(); expect(snapshot.activeRuns).toEqual([]); }); it("does not revive a failure when a newer cancelled run has null completedAt", async () => { const failed = { id: "run-old-failed", botId: "bot-2", threadId: "thread-1", taskId: "task-1", status: "failed", trigger: "user", modelProvider: null, modelId: null, error: "old failure", startedAt: null, completedAt: new Date("2026-08-23T00:00:01.000Z"), createdAt: new Date("2026-08-23T00:00:00.000Z"), }; const cancelled = { id: "run-newer-cancelled", botId: "bot-1", threadId: "thread-1", taskId: "task-2", status: "cancelled", trigger: "user", modelProvider: null, modelId: null, error: null, startedAt: new Date("2026-08-23T00:00:02.000Z"), completedAt: null, createdAt: new Date("2026-08-23T00:00:03.000Z"), }; const snapshot = await threadSnapshot( { prisma: groupPrisma(groupRunFindMany({ terminals: [cancelled, failed] })) }, groupTarget(), ); expect(snapshot.run).toBeNull(); }); it("prefers a timestamped terminal over an older failure with null completedAt", async () => { const failed = { id: "run-old-failed", botId: "bot-2", threadId: "thread-1", taskId: "task-1", status: "failed", trigger: "user", modelProvider: null, modelId: null, error: "old failure", startedAt: null, completedAt: null, createdAt: new Date("2026-08-23T00:00:00.000Z"), }; const completed = { id: "run-completed", botId: "bot-1", threadId: "thread-1", taskId: "task-2", status: "completed", trigger: "user", modelProvider: null, modelId: null, error: null, startedAt: new Date("2026-08-23T00:00:02.000Z"), completedAt: new Date("2026-08-23T00:00:04.000Z"), createdAt: new Date("2026-08-23T00:00:02.000Z"), }; const snapshot = await threadSnapshot( { prisma: groupPrisma(groupRunFindMany({ terminals: [failed, completed] })) }, groupTarget(), ); expect(snapshot.run).toBeNull(); }); it("clamps a long persisted group failure error on refresh", async () => { const longError = "x".repeat(400); const run = { id: "run-failed", botId: "bot-2", threadId: "thread-1", taskId: "task-1", status: "failed", trigger: "user", modelProvider: "openrouter", modelId: "openrouter/unknown", error: longError, startedAt: null, completedAt: new Date("2026-08-23T00:00:01.000Z"), createdAt: new Date("2026-08-23T00:00:00.000Z"), }; const snapshot = await threadSnapshot( { prisma: groupPrisma(groupRunFindMany({ terminals: [run] })) }, groupTarget(), ); expect(snapshot.run).toEqual( expect.objectContaining({ id: "run-failed", status: "failed", error: `${"x".repeat(300)}…`, }), ); }); it("keeps a concurrent member failure in run while another member is still active", async () => { const active = { id: "run-active", botId: "bot-a", threadId: "thread-1", taskId: "task-a", status: "running", trigger: "user", modelProvider: null, modelId: null, error: null, startedAt: new Date("2026-08-23T00:00:00.000Z"), completedAt: null, createdAt: new Date("2026-08-23T00:00:00.000Z"), }; const failed = { id: "run-failed", botId: "bot-b", threadId: "thread-1", taskId: "task-b", status: "failed", trigger: "user", modelProvider: null, modelId: null, error: "member exploded", startedAt: new Date("2026-08-23T00:00:01.000Z"), completedAt: new Date("2026-08-23T00:00:02.000Z"), createdAt: new Date("2026-08-23T00:00:01.000Z"), }; const snapshot = await threadSnapshot( { prisma: groupPrisma(groupRunFindMany({ active: [active], terminals: [failed] })) }, groupTarget(), ); expect(snapshot.run).toEqual( expect.objectContaining({ id: "run-failed", status: "failed", error: "member exploded" }), ); expect(snapshot.activeRuns).toEqual([ expect.objectContaining({ id: "run-active", status: "running" }), ]); }); it("keeps a failure on refresh when another member starts after it", async () => { const lateActive = { id: "run-late", botId: "bot-a", threadId: "thread-1", taskId: "task-a", status: "running", trigger: "user", modelProvider: null, modelId: null, error: null, startedAt: new Date("2026-08-23T00:00:03.000Z"), completedAt: null, createdAt: new Date("2026-08-23T00:00:03.000Z"), }; const failed = { id: "run-failed", botId: "bot-b", threadId: "thread-1", taskId: "task-b", status: "failed", trigger: "user", modelProvider: null, modelId: null, error: "member exploded", startedAt: new Date("2026-08-23T00:00:01.000Z"), completedAt: new Date("2026-08-23T00:00:02.000Z"), createdAt: new Date("2026-08-23T00:00:01.000Z"), }; const snapshot = await threadSnapshot( { prisma: groupPrisma(groupRunFindMany({ active: [lateActive], terminals: [failed] })), }, groupTarget(), ); expect(snapshot.run).toEqual( expect.objectContaining({ id: "run-failed", status: "failed", error: "member exploded" }), ); expect(snapshot.activeRuns).toEqual([ expect.objectContaining({ id: "run-late", status: "running" }), ]); }); }); function isTerminalRunQuery(where: { status?: { in?: string[] } } | undefined) { const statuses = where?.status?.in; return Array.isArray(statuses) && statuses.includes("failed") && statuses.includes("completed"); } function matchesPeerActiveFilter( row: { trigger?: string; status?: string }, where: | { trigger?: { not?: string }; OR?: Array<{ trigger?: { not?: string }; status?: { in?: string[] } }>; } | undefined, ) { if (where?.trigger?.not === "bot_message") return row.trigger !== "bot_message"; if (!where?.OR) return true; return where.OR.some((clause) => { if (clause.trigger?.not === "bot_message") return row.trigger !== "bot_message"; if (clause.status?.in) return clause.status.in.includes(row.status ?? ""); return false; }); } function botRunFindFirst( rows: Array<{ id: string; status: string; trigger?: string; createdAt?: Date; }>, ) { return vi.fn().mockImplementation( async (args: { where?: { status?: { in?: string[] }; trigger?: { not?: string }; }; select?: { id?: boolean }; }) => { const statuses = args.where?.status?.in; const matched = rows .filter((row) => !statuses || statuses.includes(row.status)) .filter((row) => args.where?.trigger?.not === "bot_message" ? row.trigger !== "bot_message" : true, ) .sort((a, b) => { const byCreated = (b.createdAt?.getTime() ?? 0) - (a.createdAt?.getTime() ?? 0); return byCreated !== 0 ? byCreated : b.id.localeCompare(a.id); }); const row = matched[0] ?? null; if (!row) return null; return args.select?.id ? { id: row.id } : row; }, ); } function groupRunFindMany(input: { active?: unknown[]; terminals?: unknown[] }) { return vi.fn().mockImplementation( async (args: { where?: { status?: { in?: string[] }; trigger?: { not?: string }; OR?: Array<{ trigger?: { not?: string }; status?: { in?: string[] } }>; }; }) => { const rows = isTerminalRunQuery(args.where) ? (input.terminals ?? []) : (input.active ?? []); return rows.filter((row) => matchesPeerActiveFilter(row as { trigger?: string; status?: string }, args.where), ); }, ); } /** The snapshot's thread lock finds the thread; its message page query finds no rows. */ function threadQueries() { return vi.fn(async (sql: TemplateStringsArray) => sql.join("").includes("FOR SHARE") ? [{ id: "thread-1" }] : [], ); } function groupPrisma(findManyRuns: ReturnType) { const tx = { $queryRaw: threadQueries(), message: { findMany: vi.fn().mockResolvedValue([]) }, event: { findFirst: vi.fn().mockResolvedValue(null), findMany: vi.fn().mockResolvedValue([]), }, run: { findMany: findManyRuns }, }; return { $transaction: vi.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), } as unknown as PrismaClient; } function groupTarget() { return { kind: "group", groupId: "group-1", groupName: "Group", members: [], threadId: "thread-1", } as unknown as ThreadTarget; } describe("sendThreadMessage", () => { it("rejects a new bot message while a run is waiting on input", async () => { const tx = { thread: { update: vi.fn().mockResolvedValue({ nextMessageSeq: 2 }), }, message: { create: vi.fn().mockResolvedValue({ id: "msg-1", threadId: "thread-1", seq: 1, role: "user", blocks: [{ kind: "text", text: "hi" }], botId: null, replyToMessageId: null, runId: null, createdAt: new Date(), }), update: vi.fn(), }, run: { findMany: vi .fn() .mockResolvedValue([{ id: "run-waiting", taskId: "task-1", status: "waiting_input" }]), }, steeringMessage: { create: vi.fn() }, event: { create: vi.fn() }, task: { create: vi.fn() }, }; const prisma = { message: { findUnique: vi.fn().mockResolvedValue(null) }, $transaction: vi.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), } as unknown as PrismaClient; const actor = { spaceId: "workspace-1", userId: "user-1" } as Actor; const target = { kind: "bot", botId: "bot-1", threadId: "thread-1", bot: { computer: null }, } as ThreadTarget; await expect( sendThreadMessage( { prisma, events: { notify: vi.fn() } as never, jobs: { enqueue: vi.fn() } as never, }, actor, target, { text: "hi", clientNonce: "nonce-1", }, ), ).rejects.toMatchObject({ code: "CONFLICT", message: "Answer the pending ask first.", }); expect(tx.steeringMessage.create).not.toHaveBeenCalled(); }); it("steers a waiting-takeover run instead of refusing the message", async () => { const tx = { thread: { update: vi .fn() .mockResolvedValueOnce({ nextMessageSeq: 2 }) .mockResolvedValueOnce({ nextEventSeq: 3 }), }, message: { create: vi.fn().mockResolvedValue({ id: "msg-1", threadId: "thread-1", seq: 1, role: "user", blocks: [{ kind: "text", text: "skip that" }], botId: null, replyToMessageId: null, runId: null, createdAt: new Date(), }), update: vi.fn(), }, run: { findMany: vi .fn() .mockResolvedValue([{ id: "run-waiting", taskId: "task-1", status: "waiting_takeover" }]), findUnique: vi.fn().mockResolvedValue({ status: "waiting_takeover", startedAt: new Date(), }), }, steeringMessage: { create: vi.fn() }, event: { create: vi.fn().mockResolvedValue({ seq: 2 }) }, task: { create: vi.fn() }, }; const prisma = { message: { findUnique: vi.fn().mockResolvedValue(null) }, $transaction: vi.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), } as unknown as PrismaClient; const actor = { spaceId: "workspace-1", userId: "user-1" } as Actor; const target = { kind: "bot", botId: "bot-1", threadId: "thread-1", bot: { computer: null }, } as ThreadTarget; const enqueue = vi.fn().mockResolvedValue(undefined); await expect( sendThreadMessage( { prisma, events: { notify: vi.fn().mockResolvedValue(undefined) } as never, jobs: { enqueue } as never, }, actor, target, { text: "skip that", clientNonce: "nonce-takeover", }, ), ).resolves.toMatchObject({ runId: "run-waiting", taskId: "task-1", seq: 1, runIds: ["run-waiting"], }); expect(tx.steeringMessage.create).toHaveBeenCalledWith({ data: { messageId: "msg-1", botId: "bot-1", userId: "user-1", runId: "run-waiting", }, }); expect(tx.task.create).not.toHaveBeenCalled(); expect(enqueue).toHaveBeenCalledWith(expect.objectContaining({ name: "run.continue" })); }); it("rejects a quote excerpt without a reply target", async () => { const prisma = { message: { findUnique: vi.fn().mockResolvedValue(null) }, $transaction: vi.fn(), } as unknown as PrismaClient; const actor = { spaceId: "workspace-1", userId: "user-1" } as Actor; const target = { kind: "bot", botId: "bot-1", threadId: "thread-1" } as ThreadTarget; await expect( sendThreadMessage( { prisma, events: { notify: vi.fn() } as never, jobs: { enqueue: vi.fn() } as never, }, actor, target, { text: "hi", replyQuote: "just this span", clientNonce: "nonce-1" }, ), ).rejects.toMatchObject({ code: "BAD_REQUEST" }); expect(prisma.$transaction).not.toHaveBeenCalled(); }); it("persists the quote excerpt alongside the reply target", async () => { let messageSeq = 0; let eventSeq = 0; const tx = { thread: { update: vi.fn(async ({ data }: { data: { nextMessageSeq?: unknown } }) => data.nextMessageSeq ? { nextMessageSeq: ++messageSeq } : { nextEventSeq: ++eventSeq }, ), }, message: { findFirst: vi.fn().mockResolvedValue({ id: "parent", blocks: [{ kind: "text", text: "the parent says just this span inside it" }], }), update: vi.fn(), create: vi.fn().mockResolvedValue({ id: "msg-1", threadId: "thread-1", seq: 1, role: "user", blocks: [{ kind: "text", text: "why this?" }], botId: null, replyToMessageId: "parent", replyQuote: "just this span", runId: null, createdAt: new Date(), }), }, run: { findMany: vi.fn().mockResolvedValue([]), findUnique: vi.fn().mockResolvedValue({ status: "queued", startedAt: null }), create: vi.fn().mockResolvedValue({ id: "run-1", taskId: "task-1", status: "queued" }), }, task: { create: vi.fn().mockResolvedValue({ id: "task-1" }) }, event: { create: vi.fn().mockResolvedValue({ id: "event-1", seq: 1, createdAt: new Date() }), }, steeringMessage: { create: vi.fn() }, }; const prisma = { message: { findUnique: vi.fn().mockResolvedValue(null) }, $transaction: vi.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), } as unknown as PrismaClient; const actor = { spaceId: "workspace-1", userId: "user-1" } as Actor; const target = { kind: "bot", botId: "bot-1", threadId: "thread-1" } as ThreadTarget; const result = await sendThreadMessage( { prisma, events: { notify: vi.fn().mockResolvedValue(undefined) } as never, jobs: { enqueue: vi.fn().mockResolvedValue(undefined) } as never, }, actor, target, { text: "why this?", replyToMessageId: "parent", replyQuote: "just this span", clientNonce: "nonce-1", }, ); expect(result).toMatchObject({ runId: "run-1", taskId: "task-1" }); expect(tx.message.findFirst).toHaveBeenCalledWith({ where: { id: "parent", threadId: "thread-1" }, select: { id: true, blocks: true }, }); expect(tx.message.create).toHaveBeenCalledWith({ data: expect.objectContaining({ replyToMessageId: "parent", replyQuote: "just this span", }), }); expect(tx.event.create).toHaveBeenCalledWith({ data: expect.objectContaining({ payload: expect.objectContaining({ replyQuote: "just this span" }), }), }); }); it("drops a quote excerpt that doesn't match the parent's text", async () => { let messageSeq = 0; let eventSeq = 0; const tx = { thread: { update: vi.fn(async ({ data }: { data: { nextMessageSeq?: unknown } }) => data.nextMessageSeq ? { nextMessageSeq: ++messageSeq } : { nextEventSeq: ++eventSeq }, ), }, message: { findFirst: vi.fn().mockResolvedValue({ id: "parent", blocks: [{ kind: "text", text: "the parent says something else entirely" }], }), update: vi.fn(), create: vi.fn().mockResolvedValue({ id: "msg-1", threadId: "thread-1", seq: 1, role: "user", blocks: [{ kind: "text", text: "why this?" }], botId: null, replyToMessageId: "parent", replyQuote: null, runId: null, createdAt: new Date(), }), }, run: { findMany: vi.fn().mockResolvedValue([]), findUnique: vi.fn().mockResolvedValue({ status: "queued", startedAt: null }), create: vi.fn().mockResolvedValue({ id: "run-1", taskId: "task-1", status: "queued" }), }, task: { create: vi.fn().mockResolvedValue({ id: "task-1" }) }, event: { create: vi.fn().mockResolvedValue({ id: "event-1", seq: 1, createdAt: new Date() }), }, steeringMessage: { create: vi.fn() }, }; const prisma = { message: { findUnique: vi.fn().mockResolvedValue(null) }, $transaction: vi.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), } as unknown as PrismaClient; const actor = { spaceId: "workspace-1", userId: "user-1" } as Actor; const target = { kind: "bot", botId: "bot-1", threadId: "thread-1" } as ThreadTarget; const result = await sendThreadMessage( { prisma, events: { notify: vi.fn().mockResolvedValue(undefined) } as never, jobs: { enqueue: vi.fn().mockResolvedValue(undefined) } as never, }, actor, target, { text: "why this?", replyToMessageId: "parent", replyQuote: "words the parent never said", clientNonce: "nonce-1", }, ); // The send still lands as a plain reply — the fabricated excerpt is dropped. expect(result).toMatchObject({ runId: "run-1", taskId: "task-1" }); expect(tx.message.create).toHaveBeenCalledWith({ data: expect.objectContaining({ replyToMessageId: "parent", replyQuote: undefined, }), }); expect(tx.event.create).toHaveBeenCalledWith({ data: expect.objectContaining({ payload: expect.not.objectContaining({ replyQuote: expect.anything() }), }), }); }); it("accepts a quote excerpt that matches formatted parent text", async () => { let messageSeq = 0; let eventSeq = 0; const tx = { thread: { update: vi.fn(async ({ data }: { data: { nextMessageSeq?: unknown } }) => data.nextMessageSeq ? { nextMessageSeq: ++messageSeq } : { nextEventSeq: ++eventSeq }, ), }, message: { findFirst: vi.fn().mockResolvedValue({ id: "parent", // Rendered as "42% growth", stored with markdown source. blocks: [{ kind: "text", text: "we saw **42%** growth last week" }], }), update: vi.fn(), create: vi.fn().mockResolvedValue({ id: "msg-1", threadId: "thread-1", seq: 1, role: "user", blocks: [{ kind: "text", text: "why this?" }], botId: null, replyToMessageId: "parent", replyQuote: "42% growth", runId: null, createdAt: new Date(), }), }, run: { findMany: vi.fn().mockResolvedValue([]), findUnique: vi.fn().mockResolvedValue({ status: "queued", startedAt: null }), create: vi.fn().mockResolvedValue({ id: "run-1", taskId: "task-1", status: "queued" }), }, task: { create: vi.fn().mockResolvedValue({ id: "task-1" }) }, event: { create: vi.fn().mockResolvedValue({ id: "event-1", seq: 1, createdAt: new Date() }), }, steeringMessage: { create: vi.fn() }, }; const prisma = { message: { findUnique: vi.fn().mockResolvedValue(null) }, $transaction: vi.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), } as unknown as PrismaClient; const actor = { spaceId: "workspace-1", userId: "user-1" } as Actor; const target = { kind: "bot", botId: "bot-1", threadId: "thread-1" } as ThreadTarget; const result = await sendThreadMessage( { prisma, events: { notify: vi.fn().mockResolvedValue(undefined) } as never, jobs: { enqueue: vi.fn().mockResolvedValue(undefined) } as never, }, actor, target, { text: "why this?", replyToMessageId: "parent", replyQuote: "42% growth", clientNonce: "nonce-1", }, ); expect(result).toMatchObject({ runId: "run-1", taskId: "task-1" }); expect(tx.message.create).toHaveBeenCalledWith({ data: expect.objectContaining({ replyQuote: "42% growth" }), }); }); it.each([ ["table cells", "| Name | Value |\n|:-----|------:|\n| Alice | 5 |", "Alice 5"], ["indented code", " 2. restart()", "2. restart()"], ["fenced code", "```text\n2. restart()\n```", "2. restart()"], ["lists after code fences", "```text\n1. code\n```\n1. First\n2. Second", "First\nSecond"], ["tab-indented code", "\t2. restart()", "2. restart()"], ["three-space lists", " 1. First\n 2. Second", "First\nSecond"], [ "ordered list items", "1. Review the diff\n2. Run the tests", "Review the diff\nRun the tests", ], [ "parenthesized list items", "1) Review the diff\n2) Run the tests", "Review the diff\nRun the tests", ], [ "quoted list items", "> 1. Review the diff\n> 2. Run the tests", "Review the diff\nRun the tests", ], ])("accepts a quote excerpt spanning rendered %s", async (_name, parentText, replyQuote) => { let messageSeq = 0; let eventSeq = 0; const tx = { thread: { update: vi.fn(async ({ data }: { data: { nextMessageSeq?: unknown } }) => data.nextMessageSeq ? { nextMessageSeq: ++messageSeq } : { nextEventSeq: ++eventSeq }, ), }, message: { findFirst: vi.fn().mockResolvedValue({ id: "parent", blocks: [ { kind: "text", text: parentText, }, ], }), update: vi.fn(), create: vi.fn().mockResolvedValue({ id: "msg-1", threadId: "thread-1", seq: 1, role: "user", blocks: [{ kind: "text", text: "why this?" }], botId: null, replyToMessageId: "parent", replyQuote, runId: null, createdAt: new Date(), }), }, run: { findMany: vi.fn().mockResolvedValue([]), findUnique: vi.fn().mockResolvedValue({ status: "queued", startedAt: null }), create: vi.fn().mockResolvedValue({ id: "run-1", taskId: "task-1", status: "queued" }), }, task: { create: vi.fn().mockResolvedValue({ id: "task-1" }) }, event: { create: vi.fn().mockResolvedValue({ id: "event-1", seq: 1, createdAt: new Date() }), }, steeringMessage: { create: vi.fn() }, }; const prisma = { message: { findUnique: vi.fn().mockResolvedValue(null) }, $transaction: vi.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), } as unknown as PrismaClient; const actor = { spaceId: "workspace-1", userId: "user-1" } as Actor; const target = { kind: "bot", botId: "bot-1", threadId: "thread-1" } as ThreadTarget; const result = await sendThreadMessage( { prisma, events: { notify: vi.fn().mockResolvedValue(undefined) } as never, jobs: { enqueue: vi.fn().mockResolvedValue(undefined) } as never, }, actor, target, { text: "why this?", replyToMessageId: "parent", replyQuote, clientNonce: "nonce-1", }, ); expect(result).toMatchObject({ runId: "run-1", taskId: "task-1" }); expect(tx.message.create).toHaveBeenCalledWith({ data: expect.objectContaining({ replyQuote }), }); }); it("drops a quote excerpt that only matches after stripping punctuation", async () => { let messageSeq = 0; let eventSeq = 0; const tx = { thread: { update: vi.fn(async ({ data }: { data: { nextMessageSeq?: unknown } }) => data.nextMessageSeq ? { nextMessageSeq: ++messageSeq } : { nextEventSeq: ++eventSeq }, ), }, message: { findFirst: vi.fn().mockResolvedValue({ id: "parent", blocks: [ { kind: "text", text: "C++ is fast and key:value pairs; version 1.2 and 2. items\n 2. restart()\n```text\n1. alpha\n2. beta\n```\n~~~text\n1. gamma\n2. delta\n~~~\n```text\n> ```\n1. epsilon\n2. zeta\n```\n> ```text\n> > ```\n> 1. eta\n> 2. theta\n> ```\n# 1. heading\n# 2. another heading\n\n > 1. literal\n > 2. another literal", }, ], }), update: vi.fn(), create: vi.fn().mockResolvedValue({ id: "msg-1", threadId: "thread-1", seq: 1, role: "user", blocks: [{ kind: "text", text: "why this?" }], botId: null, replyToMessageId: "parent", replyQuote: null, runId: null, createdAt: new Date(), }), }, run: { findMany: vi.fn().mockResolvedValue([]), findUnique: vi.fn().mockResolvedValue({ status: "queued", startedAt: null }), create: vi.fn().mockResolvedValue({ id: "run-1", taskId: "task-1", status: "queued" }), }, task: { create: vi.fn().mockResolvedValue({ id: "task-1" }) }, event: { create: vi.fn().mockResolvedValue({ id: "event-1", seq: 1, createdAt: new Date() }), }, steeringMessage: { create: vi.fn() }, }; const prisma = { message: { findUnique: vi.fn().mockResolvedValue(null) }, $transaction: vi.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), } as unknown as PrismaClient; const actor = { spaceId: "workspace-1", userId: "user-1" } as Actor; const target = { kind: "bot", botId: "bot-1", threadId: "thread-1" } as ThreadTarget; for (const replyQuote of [ "C is fast", "key value pairs", "version 12", "and items", "3. restart()", " 3. restart()", "99. C++ is fast", "alpha\nbeta", "gamma\ndelta", "epsilon\nzeta", "eta\ntheta", "heading\nanother heading", "literal\nanother literal", ]) { await sendThreadMessage( { prisma, events: { notify: vi.fn().mockResolvedValue(undefined) } as never, jobs: { enqueue: vi.fn().mockResolvedValue(undefined) } as never, }, actor, target, { text: "why this?", replyToMessageId: "parent", replyQuote }, ); } for (const call of tx.message.create.mock.calls) { expect(call[0].data.replyQuote).toBeUndefined(); } }); it("persists the quote excerpt on a group send", async () => { let messageSeq = 0; let eventSeq = 0; const tx = { thread: { update: vi.fn(async ({ data }: { data: { nextMessageSeq?: unknown } }) => data.nextMessageSeq ? { nextMessageSeq: ++messageSeq } : { nextEventSeq: ++eventSeq }, ), }, message: { findFirst: vi.fn().mockResolvedValue({ id: "parent", blocks: [{ kind: "text", text: "the parent says just this span inside it" }], }), update: vi.fn(), create: vi.fn().mockResolvedValue({ id: "msg-1", threadId: "thread-1", seq: 1, role: "user", blocks: [{ kind: "text", text: "why this?" }], botId: null, replyToMessageId: "parent", replyQuote: "just this span", runId: null, createdAt: new Date(), }), }, run: { findMany: vi.fn().mockResolvedValue([]), findUnique: vi.fn().mockResolvedValue({ status: "queued", startedAt: null }), create: vi .fn() .mockResolvedValue({ id: "run-1", taskId: "task-1", botId: "bot-a", status: "queued" }), }, task: { create: vi.fn().mockResolvedValue({ id: "task-1" }) }, event: { create: vi.fn().mockResolvedValue({ id: "event-1", seq: 1, createdAt: new Date() }), }, steeringMessage: { create: vi.fn() }, chatGroup: { findFirst: vi.fn().mockResolvedValue({ id: "group-1", members: [ { bot: { id: "bot-a", name: "Alpha", color: null } }, { bot: { id: "bot-b", name: "Beta", color: null } }, ], }), update: vi.fn().mockResolvedValue({ id: "group-1" }), }, $queryRaw: vi.fn().mockResolvedValue([{ id: "group-1" }]), }; const prisma = { message: { findUnique: vi.fn().mockResolvedValue(null) }, $transaction: vi.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), } as unknown as PrismaClient; const actor = { spaceId: "workspace-1", userId: "user-1" } as Actor; const target = { kind: "group", groupId: "group-1", groupName: "Group", threadId: "thread-1", members: [], memberBotIds: ["bot-a", "bot-b"], } satisfies ThreadTarget; const result = await sendThreadMessage( { prisma, events: { notify: vi.fn().mockResolvedValue(undefined) } as never, jobs: { enqueue: vi.fn().mockResolvedValue(undefined) } as never, }, actor, target, { text: "why this?", replyToMessageId: "parent", replyQuote: "just this span", clientNonce: "nonce-1", }, ); expect(result).toMatchObject({ runId: "run-1", taskId: "task-1" }); expect(tx.message.create).toHaveBeenCalledWith({ data: expect.objectContaining({ replyToMessageId: "parent", replyQuote: "just this span", }), }); expect(tx.event.create).toHaveBeenCalledWith({ data: expect.objectContaining({ payload: expect.objectContaining({ replyQuote: "just this span" }), }), }); }); }); describe("resolveGroupFollowUpBotId", () => { const members = [ { botId: "bot-a", name: "Alpha" }, { botId: "bot-b", name: "Beta" }, ]; it("prefers a waiting ask on a later member over the first member's busy run", () => { expect( resolveGroupFollowUpBotId({ text: "continue", members, activeRuns: [ { botId: "bot-a", status: "running", createdAt: new Date("2026-09-22T12:00:00.000Z"), }, { botId: "bot-b", status: "waiting_input", createdAt: new Date("2026-09-22T11:00:00.000Z"), }, ], }), ).toBe("bot-b"); }); it("falls back to the default first member when no member has an active run", () => { expect( resolveGroupFollowUpBotId({ text: "continue", members, activeRuns: [], }), ).toBe("bot-a"); }); it("honors @mentions when no member has an active run", () => { expect( resolveGroupFollowUpBotId({ text: "@Beta keep going", members, activeRuns: [], }), ).toBe("bot-b"); }); it("sends an unaddressed follow-up to the lead instead of the first member", () => { expect( resolveGroupFollowUpBotId({ text: "continue", members, activeRuns: [], defaultBotId: "bot-b", }), ).toBe("bot-b"); }); }); describe("groupDefaultBotId", () => { const group = { kind: "group", groupId: "group-1", threadId: "thread-1", groupName: "Group", groupKind: "main", members: [], memberBotIds: ["bot-a", "bot-b"], } as const; it("prefers the lead while it is a member", () => { expect( groupDefaultBotId({ ...group, memberBotIds: [...group.memberBotIds], leadBotId: "bot-b" }), ).toBe("bot-b"); }); it("falls back to the first member without a lead or after the lead left", () => { expect( groupDefaultBotId({ ...group, memberBotIds: [...group.memberBotIds], leadBotId: null }), ).toBe("bot-a"); expect( groupDefaultBotId({ ...group, memberBotIds: [...group.memberBotIds], leadBotId: "gone" }), ).toBe("bot-a"); }); }); describe("followUpGroupThreadMessage", () => { it("steers the follow-up to a waiting member instead of the first member", async () => { let messageSeq = 0; let eventSeq = 0; const tx = { $queryRaw: vi.fn().mockResolvedValue([{ id: "group-1" }]), thread: { update: vi.fn(async ({ data }: { data: { nextMessageSeq?: unknown } }) => data.nextMessageSeq ? { nextMessageSeq: ++messageSeq } : { nextEventSeq: ++eventSeq }, ), }, chatGroup: { findFirst: vi.fn().mockResolvedValue({ id: "group-1", members: [ { bot: { id: "bot-a", name: "Alpha", color: null } }, { bot: { id: "bot-b", name: "Beta", color: null } }, ], }), update: vi.fn().mockResolvedValue({ id: "group-1" }), }, message: { create: vi.fn().mockResolvedValue({ id: "msg-1", threadId: "thread-1", seq: 1, role: "user", blocks: [{ kind: "text", text: "continue" }], botId: null, replyToMessageId: null, runId: null, createdAt: new Date(), }), update: vi.fn(), }, run: { findMany: vi.fn().mockResolvedValue([ { id: "run-a", botId: "bot-a", status: "running", createdAt: new Date("2026-09-22T12:00:00.000Z"), }, { id: "run-b", botId: "bot-b", status: "waiting_input", createdAt: new Date("2026-09-22T11:00:00.000Z"), }, ]), findUnique: vi.fn().mockResolvedValue({ status: "waiting_input", startedAt: new Date(), }), create: vi.fn(), }, task: { create: vi.fn() }, steeringMessage: { create: vi.fn() }, event: { create: vi.fn().mockResolvedValue({ id: "event-1", seq: 1, createdAt: new Date() }), }, }; const prisma = { $transaction: vi.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), } as unknown as PrismaClient; const actor = { spaceId: "workspace-1", userId: "user-1" } as Actor; const target = { kind: "group", groupId: "group-1", groupName: "Group", threadId: "thread-1", members: [], memberBotIds: ["bot-a", "bot-b"], } satisfies ThreadTarget; await expect( followUpGroupThreadMessage({ prisma }, actor, target, { text: "continue" }), ).resolves.toEqual({ runId: undefined, eventSeq: 1 }); expect(tx.steeringMessage.create).toHaveBeenCalledWith({ data: { messageId: "msg-1", botId: "bot-b", userId: "user-1", runId: "run-b", }, }); expect(tx.run.create).not.toHaveBeenCalled(); expect(tx.event.create).toHaveBeenCalledWith({ data: expect.objectContaining({ botId: "bot-b", runId: "run-b", }), }); }); }); describe("stopThreadRuns", () => { it("snapshots every lease when group members share a team computer", async () => { const releaseScreen = vi.fn().mockResolvedValue(undefined); const execute = vi.fn(async function* () { yield { type: "exit", code: 0 }; }); const transaction = { $queryRaw: vi.fn(), run: { updateManyAndReturn: vi.fn().mockResolvedValue([ { id: "run-a", botId: "bot-a" }, { id: "run-b", botId: "bot-b" }, ]), }, steeringMessage: { deleteMany: vi.fn().mockResolvedValue({ count: 0 }) }, computer: { // Production team ownership is lease-only: Computer.executionRunId stays null. findMany: vi.fn().mockImplementation(async ({ where }: { where: { OR?: unknown[] } }) => { expect(where.OR).toEqual( expect.arrayContaining([ { id: { in: ["computer-db-team"] } }, { executionRunId: { in: ["run-a", "run-b"] } }, ]), ); return [ { id: "computer-db-team", homeKey: "home-team", kind: "fake", providerRef: "computer-team", executionBotId: null, executionRunId: null, }, ]; }), }, computerExecutionLease: { findMany: vi.fn().mockResolvedValue([ { computerId: "computer-db-team", botId: "bot-a", runId: "run-a", fence: 2 }, { computerId: "computer-db-team", botId: "bot-b", runId: "run-b", fence: 4 }, ]), }, }; const prisma = { $transaction: vi.fn(async (callback: (client: typeof transaction) => unknown) => callback(transaction), ), // Simulate workers clearing leases / execution columns as soon as the // transaction commits. A post-commit lookup would now miss both sandboxes. computer: { findMany: vi.fn().mockResolvedValue([]), updateMany: vi.fn().mockResolvedValue({ count: 0 }), }, computerExecutionLease: { updateMany: vi.fn().mockResolvedValue({ count: 2 }), }, event: { deleteMany: vi.fn().mockResolvedValue({ count: 0 }) }, } as unknown as PrismaClient; const actor = { spaceId: "workspace-1", userId: "user-1", } as Actor; const target = { kind: "group", groupId: "group-1", groupName: "Test group", threadId: "thread-1", members: [], memberBotIds: ["bot-a", "bot-b"], } satisfies ThreadTarget; await stopThreadRuns( { prisma, sandbox: { releaseScreen, execute } as unknown as SandboxProvider }, actor, target, ); expect(transaction.computerExecutionLease.findMany).toHaveBeenCalledWith({ where: { runId: { in: ["run-a", "run-b"] } }, select: { computerId: true, botId: true, runId: true, fence: true }, }); expect(execute).toHaveBeenCalledTimes(2); expect(execute).toHaveBeenCalledWith( expect.objectContaining({ providerRef: "computer-team" }), expect.objectContaining({ argv: expect.arrayContaining(["rakazo-cancel-run-work", "computer-db-team", "run-a"]), }), expect.objectContaining({ cancelRunWork: true, runId: "run-a", botId: "bot-a" }), ); expect(execute).toHaveBeenCalledWith( expect.objectContaining({ providerRef: "computer-team" }), expect.objectContaining({ argv: expect.arrayContaining(["rakazo-cancel-run-work", "computer-db-team", "run-b"]), }), expect.objectContaining({ cancelRunWork: true, runId: "run-b", botId: "bot-b" }), ); expect(releaseScreen).toHaveBeenCalledTimes(2); expect(releaseScreen).toHaveBeenCalledWith( expect.objectContaining({ providerRef: "computer-team" }), expect.objectContaining({ spaceId: "workspace-1", userId: "user-1", botId: "bot-a", cancelRunWork: true, runId: "run-a", screenLeaseId: "run-a:2", }), ); expect(releaseScreen).toHaveBeenCalledWith( expect.objectContaining({ providerRef: "computer-team" }), expect.objectContaining({ spaceId: "workspace-1", userId: "user-1", botId: "bot-b", cancelRunWork: true, runId: "run-b", screenLeaseId: "run-b:4", }), ); expect(prisma.computer.findMany).not.toHaveBeenCalled(); expect(prisma.computerExecutionLease.updateMany).toHaveBeenCalledWith({ where: { runId: { in: ["run-a", "run-b"] } }, data: { expiresAt: new Date(0) }, }); expect(prisma.computer.updateMany).toHaveBeenCalledWith( expect.objectContaining({ where: { executionRunId: { in: ["run-a", "run-b"] } } }), ); }); it("does not tear down a stale legacy execution run when the lease owns a cancelled run", async () => { const releaseScreen = vi.fn().mockResolvedValue(undefined); const execute = vi.fn(async function* () { yield { type: "exit", code: 0 }; }); const transaction = { $queryRaw: vi.fn(), run: { updateManyAndReturn: vi.fn().mockResolvedValue([{ id: "run-a", botId: "bot-a" }]), }, steeringMessage: { deleteMany: vi.fn().mockResolvedValue({ count: 0 }) }, computer: { // Selected via lease for cancelled run A, but legacy columns still name // unrelated live run B. Legacy teardown must not cancel/release B. findMany: vi.fn().mockResolvedValue([ { id: "computer-db-a", homeKey: "home-a", kind: "fake", providerRef: "computer-a", executionBotId: "bot-b", executionRunId: "run-b", }, ]), }, computerExecutionLease: { findMany: vi .fn() .mockResolvedValue([ { computerId: "computer-db-a", botId: "bot-a", runId: "run-a", fence: 2 }, ]), }, }; const prisma = { $transaction: vi.fn(async (callback: (client: typeof transaction) => unknown) => callback(transaction), ), computer: { findMany: vi.fn().mockResolvedValue([]), updateMany: vi.fn().mockResolvedValue({ count: 0 }), }, computerExecutionLease: { updateMany: vi.fn().mockResolvedValue({ count: 1 }), }, event: { deleteMany: vi.fn().mockResolvedValue({ count: 0 }) }, } as unknown as PrismaClient; const actor = { spaceId: "workspace-1", userId: "user-1", } as Actor; const target = { kind: "group", groupId: "group-1", groupName: "Test group", threadId: "thread-1", members: [], memberBotIds: ["bot-a", "bot-b"], } satisfies ThreadTarget; await stopThreadRuns( { prisma, sandbox: { releaseScreen, execute } as unknown as SandboxProvider }, actor, target, ); expect(execute).toHaveBeenCalledTimes(1); expect(execute).toHaveBeenCalledWith( expect.objectContaining({ providerRef: "computer-a" }), expect.objectContaining({ argv: expect.arrayContaining(["rakazo-cancel-run-work", "computer-db-a", "run-a"]), }), expect.objectContaining({ cancelRunWork: true, runId: "run-a", botId: "bot-a" }), ); expect(execute).not.toHaveBeenCalledWith( expect.anything(), expect.objectContaining({ argv: expect.arrayContaining(["rakazo-cancel-run-work", "computer-db-a", "run-b"]), }), expect.anything(), ); expect(releaseScreen).toHaveBeenCalledTimes(1); expect(releaseScreen).toHaveBeenCalledWith( expect.objectContaining({ providerRef: "computer-a" }), expect.objectContaining({ botId: "bot-a", runId: "run-a", screenLeaseId: "run-a:2", }), ); expect(releaseScreen).not.toHaveBeenCalledWith( expect.anything(), expect.objectContaining({ runId: "run-b" }), ); }); });