import { ORPCError } from "@orpc/server"; import type { JobPublisher } from "@rakazo/adapter-kit"; import { runContinueJob } from "@rakazo/adapter-kit"; import type { Actor, CardAct, CardBlock, MessageBlock } from "@rakazo/contracts"; import { SUGGESTION_TTL_MS } from "@rakazo/contracts"; import type { CardActResult } from "@rakazo/core"; import { applyCardAct, withSuggestionStatus } from "@rakazo/core"; import { approvalEffectKey } from "@rakazo/core/node/approval-effect-key"; import type { Prisma, PrismaClient, ThreadEvents } from "@rakazo/db"; import { IsolationError, updateCardInTransaction } from "@rakazo/db"; import { getLogger } from "@rakazo/logging"; import type { ThreadTarget } from "./thread-target.js"; import { sendThreadMessage } from "./thread-target.js"; type CardDeps = { prisma: PrismaClient; events: ThreadEvents; jobs: JobPublisher }; /** What an accepted suggestion runs: the stored call, by the bot it was meant for. */ type StoredSuggestion = { tool: string; args: Record; botId: string; threadId: string; card?: { threadId: string; messageId: string }; }; function storedSuggestion(value: unknown): StoredSuggestion | null { if (!value || typeof value !== "object") return null; const request = value as Partial; return typeof request.tool === "string" && typeof request.botId === "string" && typeof request.threadId === "string" && request.args && typeof request.args === "object" ? (request as StoredSuggestion) : null; } function resultObject(value: unknown): Record { return value && typeof value === "object" && !Array.isArray(value) ? (value as Record) : {}; } async function loadCard(prisma: PrismaClient, threadId: string, messageId: string, cardId: string) { const message = await prisma.message.findFirst({ where: { id: messageId, threadId, role: "bot" }, select: { blocks: true }, }); const blocks = (Array.isArray(message?.blocks) ? message.blocks : []) as MessageBlock[]; const card = blocks.find((block) => block.kind === "card" && block.id === cardId); return card?.kind === "card" ? card : null; } /** * Apply one tap on a bot's card. The card changes in place for everyone watching the thread; a * choice or an action then goes to the card's bot as the user's reply, through the normal send. * Accept or Dismiss on a suggestion runs or drops the stored call instead. */ export async function actOnThreadCard( deps: CardDeps, actor: Actor, target: ThreadTarget, input: { messageId: string; cardId: string; act: CardAct; clientNonce: string }, options: { requireModel?: () => Promise } = {}, ): Promise<{ ok: true }> { // A retried answer already landed as the user's reply; the card was answered with it. const replayed = await deps.prisma.message.findUnique({ where: { threadId_clientNonce: { threadId: target.threadId, clientNonce: input.clientNonce } }, select: { id: true }, }); if (replayed) return { ok: true }; const card = await loadCard(deps.prisma, target.threadId, input.messageId, input.cardId); if (!card) throw new IsolationError(); if (card.suggestion && input.act.type === "action") { return settleSuggestion(deps, actor, target, input, card.suggestion.effectId); } if (input.act.type !== "toggle") await options.requireModel?.(); let outcome: CardActResult | undefined; let botId: string | null = null; const updated = await deps.prisma.$transaction(async (tx) => { const message = await tx.message.findFirst({ where: { id: input.messageId, threadId: target.threadId }, select: { botId: true }, }); botId = message?.botId ?? null; return updateCardInTransaction( tx, { threadId: target.threadId, messageId: input.messageId, cardId: input.cardId }, (current) => { outcome = applyCardAct(current, input.act); return outcome.ok && outcome.changed ? outcome.block : null; }, ); }); if (!outcome) throw new IsolationError(); if (!outcome.ok) throw new ORPCError("CONFLICT", { message: outcome.error }); if (updated) await notify(deps.events, target.threadId, updated.seq); if (outcome.reply) { await sendThreadMessage(deps, actor, target, { text: outcome.reply, replyToMessageId: input.messageId, clientNonce: input.clientNonce, // In a group the answer goes to the bot that asked, not to whoever leads. ...(target.kind === "group" && botId ? { mentions: [{ kind: "bot" as const, id: botId }] } : {}), }); } return { ok: true }; } async function notify(events: ThreadEvents, threadId: string, seq: number) { await events.notify(threadId, seq).catch((error) => { getLogger().error("thread card realtime notification", error); }); } /** * Accept runs the stored call as its own run of the bot it was meant for, already approved: * accepting is the approval. Dismiss only closes the card. Neither needs a model. */ async function settleSuggestion( deps: CardDeps, actor: Actor, target: ThreadTarget, input: { messageId: string; cardId: string; act: CardAct; clientNonce: string }, effectId: string, ): Promise<{ ok: true }> { const settled = await deps.prisma.$transaction(async (tx) => { let outcome: CardActResult | undefined; const effect = await tx.externalEffect.findUnique({ where: { id: effectId } }); const request = storedSuggestion(effect?.request); if ( !effect || !request || effect.kind !== "suggestion" || effect.spaceId !== actor.spaceId || request.card?.messageId !== input.messageId ) { throw new IsolationError(); } // A retried tap that already went through. if (resultObject(effect.result).clientNonce === input.clientNonce) return null; const expired = Date.now() - effect.createdAt.getTime() > SUGGESTION_TTL_MS; const updated = await updateCardInTransaction( tx, { threadId: target.threadId, messageId: input.messageId, cardId: input.cardId }, (current: CardBlock) => { if (expired && current.suggestion?.status === "proposed") { return withSuggestionStatus(current, "expired"); } outcome = applyCardAct(current, input.act); return outcome.ok && outcome.changed ? outcome.block : null; }, ); if (expired) { await tx.externalEffect.updateMany({ where: { id: effect.id, status: "proposed" }, data: { status: "expired" }, }); return { seq: updated?.seq ?? null, runId: null }; } if (!outcome) throw new IsolationError(); if (!outcome.ok) throw new ORPCError("CONFLICT", { message: outcome.error }); if (effect.status !== "proposed") { throw new ORPCError("CONFLICT", { message: "This suggestion was already settled." }); } const runId = outcome.suggestion === "accept" ? await startAcceptedRun(tx, actor, effect.spaceId, request) : null; await tx.externalEffect.update({ where: { id: effect.id }, data: { status: runId ? "accepted" : "dismissed", result: { ...resultObject(effect.result), clientNonce: input.clientNonce, runId }, }, }); return { seq: updated?.seq ?? null, runId }; }); if (settled?.seq) await notify(deps.events, target.threadId, settled.seq); if (settled?.runId) { await deps.jobs.enqueue(runContinueJob(settled.runId)).catch((error) => { getLogger().error("accepted suggestion enqueue", error); }); } return { ok: true }; } async function startAcceptedRun( tx: Prisma.TransactionClient, actor: Actor, spaceId: string, request: StoredSuggestion, ): Promise { const thread = await tx.thread.findFirst({ where: { id: request.threadId, spaceId }, select: { id: true }, }); const bot = await tx.bot.findFirst({ where: { id: request.botId, spaceId }, select: { id: true }, }); if (!thread || !bot) { throw new ORPCError("CONFLICT", { message: "The bot for this suggestion is gone." }); } const task = await tx.task.create({ data: { spaceId, botId: bot.id, threadId: thread.id, userId: actor.userId, prompt: `Accepted suggestion: ${request.tool}`, status: "queued", }, }); const run = await tx.run.create({ data: { spaceId, botId: bot.id, threadId: thread.id, taskId: task.id, userId: actor.userId, status: "queued", trigger: "suggestion", }, }); // The run replays this exact call; the approval gate sees it as already approved. await tx.externalEffect.create({ data: { spaceId, runId: run.id, kind: request.tool, idempotencyKey: approvalEffectKey(run.id, request.tool, request.args), status: "approved", request: request.args as never, }, }); return run.id; }