evOo-local/apps/api/src/thread-target.ts

import { ORPCError } from "@orpc/server";
import { type JobPublisher, runContinueJob, type SandboxProvider } from "@rakazo/adapter-kit";
import { cancelComputerRunWork, screenLeaseIdForRun, toComputerRef } from "@rakazo/adapters";
import {
  type Actor,
  GROUP_MEMBER_MIN,
  type GroupKind,
  type GroupMember,
  type MessageBlock,
  type MessageReaction,
  REPLY_QUOTE_MAX_LENGTH,
  type RunStatus,
  type ThreadSnapshot,
} from "@rakazo/contracts";
import {
  ACTIVE_RUN_STATUSES,
  blocksToAgentHistoryText,
  isActive,
  projectMessages,
  resolveGroupTargetBotIds,
  runFailureError,
} from "@rakazo/core";
import {
  appendEventInTransaction,
  createGroupRepos,
  createRepos,
  createThreadMessageInTransaction,
  expireComputerExecutionLeases,
  IsolationError,
  lockOwnedGroup,
  type Prisma,
  type PrismaClient,
  type ThreadEvents,
  touchGroupUpdatedAt,
} from "@rakazo/db";
import { getLogger } from "@rakazo/logging";
import {
  buildSendPrompt,
  buildUserMessageBlocks,
  resolveGroupSendAttachments,
  resolveSendAttachments,
} from "./artifacts.js";
import { resolveBusyBotName, toComputerStatus } from "./computer-status.js";
import { joinAddressedBots } from "./main-conversation.js";
import { withSerializableRetry } from "./serializable-retry.js";
import { loadMessagePage } from "./thread-message-pages.js";

export type ThreadTarget =
  | {
      kind: "bot";
      botId: string;
      threadId: string;
      bot: Awaited<ReturnType<ReturnType<typeof createRepos>["getBot"]>>;
    }
  | {
      kind: "group";
      groupId: string;
      threadId: string;
      groupName: string;
      groupKind: GroupKind;
      /** Answers when nobody is addressed; null for ordinary groups. */
      leadBotId: string | null;
      members: GroupMember[];
      memberBotIds: string[];
    };

/** The member that answers when nobody is addressed: the group's lead, else its first member. */
export function groupDefaultBotId(target: Extract<ThreadTarget, { kind: "group" }>) {
  return target.leadBotId && target.memberBotIds.includes(target.leadBotId)
    ? target.leadBotId
    : target.memberBotIds[0];
}

/**
 * Flatten text for excerpt comparison. The parent blocks hold markdown
 * source while the selection captures rendered text, so structural syntax
 * is normalized away: table delimiters and alignment rows, list markers,
 * heading and blockquote markers, link targets, emphasis characters.
 * Semantic punctuation (: + - . ! #) stays on both sides — otherwise
 * "C++ is fast" would accept a fabricated "C is fast".
 */
function flattenForQuoteMatch(text: string, markdownSource = false): string {
  let fence: { marker: string; quoteDepth: number } | undefined;
  return text
    .split("\n")
    .filter((line) => !/^\s*\|?[\s:|-]+\|?\s*$/.test(line))
    .map((line) => {
      const normalized = line
        .replace(/^\s*(?:>\s*)+/, "")
        .replace(/^\s*#{1,6}\s+/, "")
        .replace(/^\s*[-*+•]\s+/, "");
      if (!markdownSource) return normalized;
      if (fence) {
        let boundary = line;
        const quoteDepth = fence.quoteDepth;
        for (let depth = 0; depth < quoteDepth; depth++) {
          const prefix = /^ {0,3}>[ \t]?/.exec(boundary)?.[0];
          if (!prefix) {
            fence = undefined;
            break;
          }
          boundary = boundary.slice(prefix.length);
        }
        if (fence) {
          const closing = /^ {0,3}(`{3,}|~{3,})[ \t]*$/.exec(boundary)?.[1];
          if (closing && closing[0] === fence.marker[0] && closing.length >= fence.marker.length) {
            fence = undefined;
          }
          return normalized;
        }
      }
      let source = line;
      let quoteDepth = 0;
      for (
        let prefix = /^ {0,3}>[ \t]?/.exec(source)?.[0];
        prefix;
        prefix = /^ {0,3}>[ \t]?/.exec(source)?.[0]
      ) {
        source = source.slice(prefix.length);
        quoteDepth++;
      }
      const match = /^ {0,3}(`{3,}|~{3,})(.*)$/.exec(source);
      const marker = match?.[1];
      const suffix = match?.[2] ?? "";
      if (marker && (marker[0] !== "`" || !suffix.includes("`"))) {
        fence = { marker, quoteDepth };
        return normalized;
      }
      // Test source syntax before stripping headings or other visible containers.
      return /^ {0,3}\d{1,9}[.)][ \t]+/.test(source)
        ? normalized.replace(/^ {0,3}\d{1,9}[.)][ \t]+/, "")
        : normalized;
    })
    .join(" ")
    .replace(/!\[([^\]]*)\]\([^)]*\)/g, "$1")
    .replace(/\[([^\]]*)\]\([^)]*\)/g, "$1")
    .toLowerCase()
    .replace(/&nbsp;/g, " ")
    .replace(/&amp;/g, "&")
    .replace(/&lt;/g, "<")
    .replace(/&gt;/g, ">")
    .replace(/&quot;/g, '"')
    .replace(/&#0?39;|&apos;/g, "'")
    .replace(/[‘’]/g, "'")
    .replace(/[“”]/g, '"')
    .replace(/[\\`*_~|[\]()•]/g, " ")
    .replace(/\s+/g, " ")
    .trim();
}

function quoteAppearsInBlocks(quote: string, blocks: MessageBlock[]): boolean {
  const excerpt = flattenForQuoteMatch(quote);
  if (!excerpt) return false;
  const parent = blocksToAgentHistoryText(blocks);
  // Keep existing matches (including visible numbering in code) before removing list syntax.
  return (
    flattenForQuoteMatch(parent).includes(excerpt) ||
    flattenForQuoteMatch(parent, true).includes(excerpt)
  );
}

const THREAD_MESSAGE_PAGE_SIZE = 100;
const RUNS_NEEDING_CONTINUE = new Set(["queued", "waiting_takeover"]);

const STEERABLE_RUN_STATUSES = new Set(["queued", "leased", "running", "waiting_takeover"]);

type MentionTargetInput = string | { kind: "bot" | "group" | "routine" | "connector"; id: string };

function splitMentionTargets(mentions: MentionTargetInput[] | undefined) {
  const botMentionIds = new Set<string>();
  const groupMentionIds = new Set<string>();
  const routineMentionIds = new Set<string>();
  const connectorMentionIds = new Set<string>();
  for (const mention of mentions ?? []) {
    if (typeof mention === "string") {
      botMentionIds.add(mention);
      continue;
    }
    if (mention.kind === "bot") botMentionIds.add(mention.id);
    if (mention.kind === "group") groupMentionIds.add(mention.id);
    if (mention.kind === "routine") routineMentionIds.add(mention.id);
    if (mention.kind === "connector") connectorMentionIds.add(mention.id);
  }
  return {
    botMentionIds: [...botMentionIds],
    groupMentionIds: [...groupMentionIds],
    routineMentionIds: [...routineMentionIds],
    connectorMentionIds: [...connectorMentionIds],
  };
}

async function resolveOwnedConnectorDisplayNames(
  tx: Prisma.TransactionClient,
  actor: Actor,
  connectionIds: string[],
) {
  if (!connectionIds.length) return [];
  const rows = await tx.connection.findMany({
    where: {
      id: { in: connectionIds },
      spaceId: actor.spaceId,
      userId: actor.userId,
      status: "connected",
    },
    select: { id: true, displayName: true },
  });
  if (rows.length !== connectionIds.length) throw new IsolationError();
  const byId = new Map(rows.map((row) => [row.id, row.displayName]));
  return connectionIds.map((id) => byId.get(id) ?? "connector");
}

function sendRunClientNonce(
  clientNonce: string | undefined,
  messageId: string,
  botId?: string,
): string | undefined {
  if (!clientNonce) return undefined;
  return botId ? `send:${messageId}:${botId}` : `send:${messageId}`;
}

async function enqueueRunsNeedingContinue(
  jobs: JobPublisher,
  runs: Array<{ id: string; status: string }>,
) {
  await Promise.all(
    runs
      .filter((run) => RUNS_NEEDING_CONTINUE.has(run.status))
      .map((run) =>
        jobs.enqueue(runContinueJob(run.id)).catch((error) => {
          // The queued run is durable; the reconciler repairs a missed immediate wake.
          getLogger().error("thread send enqueue", error);
        }),
      ),
  );
}

async function findSendReceipt(prisma: PrismaClient, threadId: string, clientNonce: string) {
  return prisma.message.findUnique({
    where: { threadId_clientNonce: { threadId, clientNonce } },
    include: { sourceRuns: { orderBy: [{ createdAt: "asc" }, { id: "asc" }] } },
  });
}

async function replayExistingSend(
  deps: { prisma: PrismaClient; events: ThreadEvents; jobs: JobPublisher },
  threadId: string,
  clientNonce: string | undefined,
) {
  if (!clientNonce) return null;
  const message = await findSendReceipt(deps.prisma, threadId, clientNonce);
  if (!message) return null;
  const receiptEvent = await deps.prisma.event.findFirst({
    where: {
      threadId,
      type: "thread.message.created",
      payload: { path: ["messageId"], equals: message.id },
    },
    orderBy: { seq: "desc" },
    select: { payload: true },
  });
  const receiptRunIds = sendEventRunIds(receiptEvent?.payload);
  const receiptRuns = receiptRunIds.length
    ? await deps.prisma.run.findMany({ where: { id: { in: receiptRunIds } } })
    : [];
  const receiptRunById = new Map(receiptRuns.map((run) => [run.id, run]));
  const orderedReceiptRuns = receiptRunIds.flatMap((id) => {
    const run = receiptRunById.get(id);
    return run ? [run] : [];
  });
  const linkedRun =
    message.sourceRuns[0] ??
    (message.runId ? await deps.prisma.run.findUnique({ where: { id: message.runId } }) : null);
  if (!linkedRun && orderedReceiptRuns.length === 0) return null;
  const runs = orderedReceiptRuns.length
    ? orderedReceiptRuns
    : message.sourceRuns.length
      ? message.sourceRuns
      : [linkedRun!];
  await enqueueRunsNeedingContinue(deps.jobs, runs);
  const latestEvent = await deps.prisma.event.findFirst({
    where: { threadId },
    orderBy: { seq: "desc" },
    select: { seq: true },
  });
  if (latestEvent) {
    await deps.events.notify(threadId, latestEvent.seq).catch((error) => {
      // Subscribers catch up from the durable event cursor after a missed realtime wake.
      getLogger().error("thread send realtime notification", error);
    });
  }
  return sendResult(message, runs);
}

function sendEventRunIds(payload: Prisma.JsonValue | undefined): string[] {
  if (!payload || typeof payload !== "object" || Array.isArray(payload)) return [];
  const runIds = (payload as { runIds?: unknown }).runIds;
  return Array.isArray(runIds) ? runIds.filter((id): id is string => typeof id === "string") : [];
}

function sendResult(message: { seq: number }, runs: Array<{ id: string; taskId: string }>) {
  const first = runs[0];
  if (!first) throw new IsolationError("Send did not create a run");
  return {
    taskId: first.taskId,
    runId: first.id,
    seq: message.seq,
    runIds: runs.map((run) => run.id),
  };
}

export async function cancelSupersededQueuedRuns(
  tx: Prisma.TransactionClient,
  input: { threadId: string; botIds: string[]; keepRunIds: string[] },
) {
  const superseded = await tx.run.findMany({
    where: {
      threadId: input.threadId,
      botId: { in: input.botIds },
      status: "queued",
      OR: [{ trigger: "user", sourceMessage: { role: "user" } }, { trigger: "reaction" }],
      id: { notIn: input.keepRunIds },
    },
    select: { id: true, taskId: true },
  });
  if (superseded.length === 0) return;
  const now = new Date();
  await tx.run.updateMany({
    where: { id: { in: superseded.map((run) => run.id) } },
    data: { status: "cancelled", completedAt: now },
  });
  await tx.task.updateMany({
    where: { id: { in: superseded.map((run) => run.taskId) } },
    data: { status: "cancelled" },
  });
}

async function lockAndLoadGroupMembers(
  tx: Prisma.TransactionClient,
  actor: Actor,
  target: Extract<ThreadTarget, { kind: "group" }>,
) {
  await lockOwnedGroup(tx, actor, target.groupId);
  const group = await tx.chatGroup.findFirst({
    where: {
      id: target.groupId,
      spaceId: actor.spaceId,
      userId: actor.userId,
      archivedAt: null,
      thread: { id: target.threadId },
    },
    include: {
      members: {
        where: { bot: { archivedAt: null } },
        include: { bot: { select: { id: true, name: true, color: true } } },
        orderBy: { createdAt: "asc" },
      },
    },
  });
  // The main conversation starts with only its lead.
  const minimum = group?.kind === "main" ? 1 : GROUP_MEMBER_MIN;
  if (!group || group.members.length < minimum) throw new IsolationError();
  return group.members.map((member) => ({
    botId: member.bot.id,
    name: member.bot.name,
    color: member.bot.color,
  }));
}

export async function resolveThreadTarget(
  prisma: PrismaClient,
  actor: Actor,
  input: { botId?: string; groupId?: string },
): Promise<ThreadTarget> {
  const repos = createRepos(prisma);
  const groupRepos = createGroupRepos(prisma);
  if (input.botId) {
    const bot = await repos.getBot(actor, input.botId);
    if (!bot.thread) throw new IsolationError();
    return {
      kind: "bot",
      botId: bot.id,
      threadId: bot.thread.id,
      bot,
    };
  }
  if (input.groupId) {
    const group = await groupRepos.getGroupTarget(actor, input.groupId);
    if (!group.thread) throw new IsolationError();
    const members = group.members.map((member) => ({
      botId: member.bot.id,
      name: member.bot.name,
      color: member.bot.color,
      status: member.bot.runs[0]?.status ?? "idle",
    }));
    return {
      kind: "group",
      groupId: group.id,
      threadId: group.thread.id,
      groupName: group.name,
      groupKind: group.kind === "main" ? "main" : "group",
      leadBotId: group.leadBotId,
      members,
      memberBotIds: members.map((member) => member.botId),
    };
  }
  throw new IsolationError();
}

export async function threadHead(prisma: PrismaClient, target: ThreadTarget) {
  const latest = await prisma.event.findFirst({
    where: { threadId: target.threadId },
    orderBy: { seq: "desc" },
    select: { seq: true },
  });
  return { threadId: target.threadId, cursor: latest?.seq ?? -1 };
}

export async function threadSnapshot(
  deps: { prisma: PrismaClient },
  target: ThreadTarget,
): Promise<ThreadSnapshot> {
  // Lock the thread row so messages, the event cursor, active runs, and live
  // progress are read from one consistent commit. A torn Promise.all can
  // otherwise advance the client cursor past thread.message.created while the
  // ask message page still omits it — leaving waiting_input with no AskCard.
  if (target.kind === "bot") {
    const [busyBotName, core] = await Promise.all([
      resolveBusyBotName(deps.prisma, {
        computerId: target.bot.computer?.id,
        botId: target.botId,
        botName: target.bot.name,
      }),
      deps.prisma.$transaction(async (tx) => {
        await tx.$queryRaw`SELECT id FROM threads WHERE id = ${target.threadId} FOR SHARE`;
        const [messagePage, last, waitingRun, busyOrFailed] = await Promise.all([
          loadMessagePage(tx, target.threadId, undefined, THREAD_MESSAGE_PAGE_SIZE),
          tx.event.findFirst({
            where: { threadId: target.threadId },
            orderBy: { seq: "desc" },
            select: { seq: true },
          }),
          // Waiting asks win over a concurrent busy run (including peer bot_message).
          tx.run.findFirst({
            where: {
              botId: target.botId,
              threadId: target.threadId,
              status: { in: ["waiting_input", "waiting_takeover"] },
            },
            orderBy: [{ createdAt: "desc" }, { id: "desc" }],
          }),
          tx.run.findFirst({
            where: {
              botId: target.botId,
              threadId: target.threadId,
              // Hide peer bot_message busy/failed noise; waiting is handled above.
              trigger: { not: "bot_message" },
              status: { in: [...ACTIVE_RUN_STATUSES, "failed"] },
            },
            // The id tiebreak keeps ordering deterministic under equal
            // timestamps, matching the supersession probe below.
            orderBy: [{ createdAt: "desc" }, { id: "desc" }],
          }),
        ]);
        const run = waitingRun ?? busyOrFailed;
        // A failed run is only the thread's word while it is still the newest
        // terminal run; otherwise a stale failure would resurface in the
        // composer error strip on every load, forever. Instead of comparing
        // timestamps (equal createdAt values reverse under gt/gte), ask for
        // the newest terminal run under the same deterministic ordering and
        // check whether it is this failure.
        const newestTerminal =
          run?.status === "failed"
            ? await tx.run.findFirst({
                where: {
                  botId: target.botId,
                  threadId: target.threadId,
                  // Peer bot_message failures must not bury a user-visible failure.
                  trigger: { not: "bot_message" },
                  status: { in: ["failed", "completed", "cancelled"] },
                },
                orderBy: [{ createdAt: "desc" }, { id: "desc" }],
                select: { id: true },
              })
            : null;
        const currentRun = run?.status === "failed" && newestTerminal?.id !== run.id ? null : run;
        const liveEvents =
          currentRun && isActive(currentRun.status as RunStatus)
            ? await tx.event.findMany({
                where: {
                  threadId: target.threadId,
                  runId: currentRun.id,
                  type: {
                    in: [
                      "thread.progress",
                      "thread.subagent",
                      "agent.tool.called",
                      "agent.tool.completed",
                    ],
                  },
                },
                orderBy: { seq: "asc" },
              })
            : [];
        return { messagePage, last, run: currentRun, liveEvents };
      }),
    ]);
    return {
      botId: target.botId,
      threadId: target.threadId,
      cursor: core.last?.seq ?? -1,
      messages: messagesWithLiveEvents(core.messagePage.messages, core.liveEvents),
      olderCursor: core.messagePage.olderCursor,
      run: core.run ? mapRun(core.run) : null,
      computer: toComputerStatus(target.botId, target.bot.computer, busyBotName),
    };
  }

  const core = await deps.prisma.$transaction(async (tx) => {
    await tx.$queryRaw`SELECT id FROM threads WHERE id = ${target.threadId} FOR SHARE`;
    const [messagePage, last, activeRuns, recentTerminals] = await Promise.all([
      loadMessagePage(tx, target.threadId, undefined, THREAD_MESSAGE_PAGE_SIZE),
      tx.event.findFirst({
        where: { threadId: target.threadId },
        orderBy: { seq: "desc" },
        select: { seq: true },
      }),
      tx.run.findMany({
        where: {
          threadId: target.threadId,
          status: { in: [...ACTIVE_RUN_STATUSES] },
          // Include waiting peer runs so their ask cards stay answerable.
          OR: [
            { trigger: { not: "bot_message" } },
            { status: { in: ["waiting_input", "waiting_takeover"] } },
          ],
        },
        orderBy: { createdAt: "desc" },
      }),
      // Recently updated terminals (completion bumps updatedAt). pickLatestTerminalRun then
      // ranks by completedAt ?? createdAt so null timestamps cannot revive a stale failure.
      tx.run.findMany({
        where: {
          threadId: target.threadId,
          trigger: { not: "bot_message" },
          status: { in: ["failed", "completed", "cancelled"] },
        },
        orderBy: [{ updatedAt: "desc" }, { id: "desc" }],
        take: 50,
      }),
    ]);
    const liveEvents =
      activeRuns.length > 0
        ? await tx.event.findMany({
            where: {
              threadId: target.threadId,
              runId: { in: activeRuns.map((run) => run.id) },
              type: {
                in: [
                  "thread.progress",
                  "thread.subagent",
                  "agent.tool.called",
                  "agent.tool.completed",
                ],
              },
            },
            orderBy: { seq: "asc" },
          })
        : [];
    return {
      messagePage,
      last,
      activeRuns,
      terminalRun: pickLatestTerminalRun(recentTerminals),
      liveEvents,
    };
  });
  const primaryActiveRun = pickPrimaryActiveRun(core.activeRuns);
  return {
    groupId: target.groupId,
    groupName: target.groupName,
    members: target.members,
    ...(target.leadBotId ? { leadBotId: target.leadBotId } : {}),
    threadId: target.threadId,
    cursor: core.last?.seq ?? -1,
    messages: messagesWithLiveEvents(core.messagePage.messages, core.liveEvents),
    olderCursor: core.messagePage.olderCursor,
    // Match the live reducer: a failed latest terminal stays in run even while siblings are
    // still active or start late. A newer completed/cancelled terminal clears it.
    run:
      core.terminalRun?.status === "failed"
        ? mapRun(core.terminalRun)
        : primaryActiveRun
          ? mapRun(primaryActiveRun)
          : null,
    activeRuns: core.activeRuns.map(mapRun),
  };
}

/**
 * Prefer a waiting ask/takeover over a merely-busy run for the group's headline
 * `run` field, even when the busy run started more recently — createdAt-desc
 * ordering alone would let a fresh busy run for one bot bury an older waiting
 * card for another. `activeRuns` still carries every active run regardless of
 * which one is picked here, so nothing is hidden from the client, only the
 * single-run summary field.
 */
function pickPrimaryActiveRun<T extends { status: string }>(runs: readonly T[]): T | undefined {
  return (
    runs.find((run) => run.status === "waiting_input" || run.status === "waiting_takeover") ??
    runs[0]
  );
}

/** Latest terminal by end time (completedAt, else createdAt), then createdAt, then id. */
function pickLatestTerminalRun<T extends { id: string; createdAt: Date; completedAt: Date | null }>(
  runs: T[],
): T | null {
  if (runs.length === 0) return null;
  return runs.reduce((best, run) => {
    const bestEnd = (best.completedAt ?? best.createdAt).getTime();
    const runEnd = (run.completedAt ?? run.createdAt).getTime();
    if (runEnd !== bestEnd) return runEnd > bestEnd ? run : best;
    if (run.createdAt.getTime() !== best.createdAt.getTime()) {
      return run.createdAt > best.createdAt ? run : best;
    }
    return run.id > best.id ? run : best;
  });
}

function messagesWithLiveEvents(
  persisted: ThreadSnapshot["messages"],
  liveEvents: Parameters<typeof projectMessages>[0],
) {
  const live = projectMessages(liveEvents).filter((message) => {
    if (message.blocks.some((block) => block.kind === "progress" || block.kind === "steps")) {
      return true;
    }
    if (!message.id.startsWith("subagent:")) return false;
    return !persisted.some((row) =>
      row.blocks.some(
        (block) => block.kind === "subagent" && message.id === `subagent:${block.agentId}`,
      ),
    );
  });
  return [...persisted, ...live];
}

function mapRun(run: {
  id: string;
  botId: string;
  threadId: string;
  taskId: string;
  status: string;
  trigger: string;
  routineId: string | null;
  modelProvider: string | null;
  modelId: string | null;
  error: string | null;
  startedAt: Date | null;
  completedAt: Date | null;
  createdAt: Date;
}) {
  return {
    id: run.id,
    botId: run.botId,
    threadId: run.threadId,
    taskId: run.taskId,
    status: run.status as never,
    trigger: run.trigger as never,
    routineId: run.routineId ?? null,
    modelProvider: run.modelProvider,
    modelId: run.modelId,
    // Same display clamp as live run.failed events so a huge stored error cannot bypass it.
    error:
      run.status === "failed"
        ? runFailureError({ type: "run.failed", payload: { error: run.error } })
        : run.error,
    startedAt: run.startedAt?.toISOString() ?? null,
    completedAt: run.completedAt?.toISOString() ?? null,
    createdAt: run.createdAt.toISOString(),
  };
}

export async function sendThreadMessage(
  deps: {
    prisma: PrismaClient;
    events: ThreadEvents;
    jobs: JobPublisher;
  },
  actor: Actor,
  target: ThreadTarget,
  input: {
    text?: string;
    artifactIds?: string[];
    mentions?: MentionTargetInput[];
    replyToMessageId?: string;
    replyQuote?: string;
    clientNonce?: string;
  },
) {
  const existing = await replayExistingSend(deps, target.threadId, input.clientNonce);
  if (existing) return existing;
  if (target.kind === "group" && target.groupKind === "main") {
    await joinAddressedBots(deps.prisma, actor, {
      memberBotIds: target.memberBotIds,
      text: input.text ?? "",
      botMentionIds: splitMentionTargets(input.mentions).botMentionIds,
    });
  }
  // The excerpt is rendered text while blocks hold markdown source, so it
  // can't be substring-verified verbatim — enforce the cap here, then check
  // the flattened form against the parent inside the transaction.
  let replyQuote = input.replyQuote?.trim().slice(0, REPLY_QUOTE_MAX_LENGTH) || undefined;
  if (replyQuote && !input.replyToMessageId) {
    throw new ORPCError("BAD_REQUEST", { message: "replyQuote requires replyToMessageId." });
  }

  const commit = () =>
    deps.prisma.$transaction(async (tx) => {
      if (input.replyToMessageId) {
        const reply = await tx.message.findFirst({
          where: { id: input.replyToMessageId, threadId: target.threadId },
          select: { id: true, blocks: true },
        });
        if (!reply) throw new IsolationError();
        // Client-supplied excerpts are untrusted: drop a mismatch instead of
        // failing the send — the reply still lands, just without the quote.
        if (
          replyQuote &&
          !quoteAppearsInBlocks(
            replyQuote,
            Array.isArray(reply.blocks) ? (reply.blocks as MessageBlock[]) : [],
          )
        ) {
          replyQuote = undefined;
        }
      }

      if (target.kind === "bot") {
        const mentionTargets = splitMentionTargets(input.mentions);
        const { blocks: attachmentBlocks, artifacts } = await resolveSendAttachments(
          { prisma: tx },
          actor,
          target.botId,
          input.artifactIds,
        );
        const connectorNames = await resolveOwnedConnectorDisplayNames(
          tx,
          actor,
          mentionTargets.connectorMentionIds,
        );
        const blocks = buildUserMessageBlocks(input.text, attachmentBlocks);
        const message = await createThreadMessageInTransaction(tx, {
          threadId: target.threadId,
          role: "user",
          blocks,
          replyToMessageId: input.replyToMessageId,
          replyQuote,
          clientNonce: input.clientNonce,
        });
        const activeRuns = await tx.run.findMany({
          where: {
            threadId: target.threadId,
            botId: target.botId,
            status: { in: [...ACTIVE_RUN_STATUSES] },
          },
          select: { id: true, taskId: true, status: true },
        });
        if (activeRuns.some((run) => !STEERABLE_RUN_STATUSES.has(run.status))) {
          throw new ORPCError("CONFLICT", {
            message: "Answer the pending ask first.",
          });
        }
        const active = activeRuns[0];
        if (active) {
          await tx.steeringMessage.create({
            data: {
              messageId: message.id,
              botId: target.botId,
              userId: actor.userId,
              runId: active.id,
            },
          });
          await tx.message.update({ where: { id: message.id }, data: { runId: active.id } });
          const event = await appendEventInTransaction(tx, {
            spaceId: actor.spaceId,
            threadId: target.threadId,
            botId: target.botId,
            type: "thread.message.created",
            runId: active.id,
            payload: {
              messageId: message.id,
              role: "user",
              blocks,
              replyToMessageId: input.replyToMessageId,
              replyQuote,
            },
          });
          return { message, runs: [active], eventSeq: event.seq };
        }
        const task = await tx.task.create({
          data: {
            spaceId: actor.spaceId,
            botId: target.botId,
            threadId: target.threadId,
            userId: actor.userId,
            prompt: buildSendPrompt(input.text, artifacts, connectorNames),
            status: "queued",
          },
        });
        const run = await tx.run.create({
          data: {
            spaceId: actor.spaceId,
            botId: target.botId,
            threadId: target.threadId,
            taskId: task.id,
            userId: actor.userId,
            status: "queued",
            trigger: "user",
            clientNonce: sendRunClientNonce(input.clientNonce, message.id),
            sourceMessageId: message.id,
          },
        });
        await tx.message.update({ where: { id: message.id }, data: { runId: run.id } });
        await cancelSupersededQueuedRuns(tx, {
          threadId: target.threadId,
          botIds: [target.botId],
          keepRunIds: [run.id],
        });
        const event = await appendEventInTransaction(tx, {
          spaceId: actor.spaceId,
          threadId: target.threadId,
          botId: target.botId,
          type: "thread.message.created",
          runId: run.id,
          payload: {
            messageId: message.id,
            role: "user",
            blocks,
            runIds: [run.id],
            replyToMessageId: input.replyToMessageId,
            replyQuote,
          },
        });
        return { message, runs: [run], eventSeq: event.seq };
      }

      const members = await lockAndLoadGroupMembers(tx, actor, target);
      const memberBotIds = members.map((member) => member.botId);
      const mentionTargets = splitMentionTargets(input.mentions);
      const targetBotIds = resolveGroupTargetBotIds({
        text: input.text ?? "",
        members: members.map((member) => ({ id: member.botId, name: member.name })),
        explicitMentions: mentionTargets.botMentionIds,
        defaultBotId: target.leadBotId,
      });
      const { blocks: attachmentBlocks, artifacts } = await resolveGroupSendAttachments(
        { prisma: tx },
        actor,
        target.groupId,
        memberBotIds,
        input.artifactIds,
      );
      const connectorNames = await resolveOwnedConnectorDisplayNames(
        tx,
        actor,
        mentionTargets.connectorMentionIds,
      );
      const blocks = buildUserMessageBlocks(input.text, attachmentBlocks);
      const message = await createThreadMessageInTransaction(tx, {
        threadId: target.threadId,
        role: "user",
        blocks,
        replyToMessageId: input.replyToMessageId,
        replyQuote,
        clientNonce: input.clientNonce,
      });
      const activeRuns = await tx.run.findMany({
        where: {
          threadId: target.threadId,
          botId: { in: targetBotIds },
          status: { in: [...ACTIVE_RUN_STATUSES] },
        },
        select: { id: true, taskId: true, botId: true, status: true },
      });
      const activeByBotId = new Map<string, (typeof activeRuns)[number]>();
      for (const run of activeRuns) {
        if (!STEERABLE_RUN_STATUSES.has(run.status)) {
          throw new ORPCError("CONFLICT", {
            message: "Answer the pending ask first.",
          });
        }
        if (!activeByBotId.has(run.botId)) activeByBotId.set(run.botId, run);
      }
      const runs: Array<{ id: string; taskId: string; botId: string; status: string }> = [];
      for (const botId of targetBotIds) {
        const active = activeByBotId.get(botId);
        if (active) {
          await tx.steeringMessage.create({
            data: { messageId: message.id, botId, userId: actor.userId, runId: active.id },
          });
          runs.push(active);
          continue;
        }
        const task = await tx.task.create({
          data: {
            spaceId: actor.spaceId,
            botId,
            threadId: target.threadId,
            userId: actor.userId,
            prompt: buildSendPrompt(input.text, artifacts, connectorNames),
            status: "queued",
          },
        });
        const run = await tx.run.create({
          data: {
            spaceId: actor.spaceId,
            botId,
            threadId: target.threadId,
            taskId: task.id,
            userId: actor.userId,
            status: "queued",
            trigger: "user",
            clientNonce: sendRunClientNonce(input.clientNonce, message.id, botId),
            sourceMessageId: message.id,
          },
        });
        runs.push(run);
      }
      const firstRun = runs[0];
      const eventBotId = firstRun?.botId ?? targetBotIds[0];
      if (!eventBotId) throw new IsolationError("Group send did not resolve a target");
      if (firstRun) {
        await tx.message.update({ where: { id: message.id }, data: { runId: firstRun.id } });
        const createdRuns = runs.filter((run) => !activeByBotId.has(run.botId));
        if (createdRuns.length) {
          await cancelSupersededQueuedRuns(tx, {
            threadId: target.threadId,
            botIds: createdRuns.map((run) => run.botId),
            keepRunIds: createdRuns.map((run) => run.id),
          });
        }
      }
      await touchGroupUpdatedAt(tx, target.groupId);
      const event = await appendEventInTransaction(tx, {
        spaceId: actor.spaceId,
        threadId: target.threadId,
        botId: eventBotId,
        type: "thread.message.created",
        runId: firstRun?.id ?? activeRuns[0]?.id,
        payload: {
          messageId: message.id,
          role: "user",
          blocks,
          runIds: runs.map((run) => run.id),
          replyToMessageId: input.replyToMessageId,
          replyQuote,
        },
      });
      return { message, runs, eventSeq: event.seq };
    });

  const committed = await withSerializableRetry(commit).catch(async (error) => {
    const winner = await replayExistingSend(deps, target.threadId, input.clientNonce);
    if (winner) return { replay: winner } as const;
    throw error;
  });
  if ("replay" in committed) return committed.replay;
  await deps.events.notify(target.threadId, committed.eventSeq).catch((error) => {
    // Subscribers catch up from the durable event cursor after a missed realtime wake.
    getLogger().error("thread send realtime notification", error);
  });
  await enqueueRunsNeedingContinue(deps.jobs, committed.runs);
  return sendResult(committed.message, committed.runs);
}

/**
 * Pick the group member bot that should receive a follow-up. Prefer a waiting
 * ask/takeover on any member over the first member's merely-busy run — the
 * router's old members[0] shortcut let a fresh run on bot A bury an unanswered
 * ask on bot B. Unaddressed follow-ups go to the lead when the group has one.
 */
export function resolveGroupFollowUpBotId(input: {
  text: string;
  members: Array<{ botId: string; name: string }>;
  activeRuns: Array<{ botId: string; status: string; createdAt: Date }>;
  defaultBotId?: string | null;
}): string {
  const memberBotIds = new Set(input.members.map((member) => member.botId));
  const memberRuns = input.activeRuns
    .filter((run) => memberBotIds.has(run.botId))
    .sort((left, right) => right.createdAt.getTime() - left.createdAt.getTime());
  const waiting = memberRuns.find(
    (run) => run.status === "waiting_input" || run.status === "waiting_takeover",
  );
  if (waiting) return waiting.botId;
  const steerable = memberRuns.find((run) => STEERABLE_RUN_STATUSES.has(run.status));
  if (steerable) return steerable.botId;
  const mentioned = resolveGroupTargetBotIds({
    text: input.text,
    members: input.members.map((member) => ({ id: member.botId, name: member.name })),
    explicitMentions: [],
    defaultBotId: input.defaultBotId,
  })[0];
  if (mentioned) return mentioned;
  const fallback = input.members[0]?.botId;
  if (!fallback) throw new IsolationError();
  return fallback;
}

export async function followUpGroupThreadMessage(
  deps: { prisma: PrismaClient },
  actor: Actor,
  target: Extract<ThreadTarget, { kind: "group" }>,
  input: { text: string },
) {
  return deps.prisma.$transaction(async (tx) => {
    const members = await lockAndLoadGroupMembers(tx, actor, target);
    const memberBotIds = members.map((member) => member.botId);
    const activeRuns = await tx.run.findMany({
      where: {
        threadId: target.threadId,
        botId: { in: memberBotIds },
        status: { in: [...ACTIVE_RUN_STATUSES] },
      },
      select: { id: true, botId: true, status: true, createdAt: true },
      orderBy: { createdAt: "desc" },
    });
    const botId = resolveGroupFollowUpBotId({
      text: input.text,
      members,
      activeRuns,
      defaultBotId: target.leadBotId,
    });
    const blocks = [{ kind: "text" as const, text: input.text }];
    const message = await createThreadMessageInTransaction(tx, {
      threadId: target.threadId,
      role: "user",
      blocks,
    });
    const active = activeRuns.find((run) => run.botId === botId);
    let run: { id: string } | null = null;
    if (!active) {
      const task = await tx.task.create({
        data: {
          spaceId: actor.spaceId,
          botId,
          threadId: target.threadId,
          userId: actor.userId,
          prompt: input.text,
          status: "queued",
        },
      });
      run = await tx.run.create({
        data: {
          spaceId: actor.spaceId,
          botId,
          threadId: target.threadId,
          taskId: task.id,
          userId: actor.userId,
          status: "queued",
          trigger: "follow_up",
          sourceMessageId: message.id,
        },
        select: { id: true },
      });
      await tx.message.update({ where: { id: message.id }, data: { runId: run.id } });
    } else {
      await tx.steeringMessage.create({
        data: {
          messageId: message.id,
          botId,
          userId: actor.userId,
          runId: active.id,
        },
      });
      await tx.message.update({ where: { id: message.id }, data: { runId: active.id } });
    }
    const event = await appendEventInTransaction(tx, {
      spaceId: actor.spaceId,
      threadId: target.threadId,
      botId,
      type: "thread.message.created",
      runId: run?.id ?? active?.id,
      payload: { messageId: message.id, role: "user", blocks },
    });
    await touchGroupUpdatedAt(tx, target.groupId);
    return { runId: run?.id, eventSeq: event.seq };
  });
}

/**
 * Append an emoji reply to the conversation so the next AI turn sees its target.
 * Clients render it beneath that message; the reaction itself does not start an AI turn.
 */
export async function reactToThreadMessage(
  deps: { prisma: PrismaClient },
  actor: Actor,
  target: ThreadTarget,
  input: {
    messageId: string;
    reaction: MessageReaction;
    clientNonce: string;
  },
) {
  return deps.prisma.$transaction(async (tx) => {
    await tx.$queryRaw`SELECT id FROM threads WHERE id = ${target.threadId} FOR UPDATE`;
    const parent = await tx.message.findFirst({
      where: { id: input.messageId, threadId: target.threadId },
      select: { id: true },
    });
    if (!parent) throw new IsolationError();
    const existing = await tx.message.findUnique({
      where: {
        threadId_clientNonce: { threadId: target.threadId, clientNonce: input.clientNonce },
      },
      select: { id: true },
    });
    if (existing) return { eventSeq: null };
    const botId = target.kind === "bot" ? target.botId : groupDefaultBotId(target);
    if (!botId) throw new IsolationError();
    const blocks: MessageBlock[] = [{ kind: "text", text: input.reaction }];
    const message = await createThreadMessageInTransaction(tx, {
      threadId: target.threadId,
      role: "user",
      blocks,
      replyToMessageId: parent.id,
      clientNonce: input.clientNonce,
    });
    if (target.kind === "group") await touchGroupUpdatedAt(tx, target.groupId);
    const event = await appendEventInTransaction(tx, {
      spaceId: actor.spaceId,
      threadId: target.threadId,
      botId,
      type: "thread.message.created",
      payload: { messageId: message.id, role: "user", blocks, replyToMessageId: parent.id },
    });
    return { eventSeq: event.seq };
  });
}

export async function stopThreadRuns(
  deps: {
    prisma: PrismaClient;
    sandbox: SandboxProvider;
  },
  actor: Actor,
  target: ThreadTarget,
) {
  const { runIds, computers, leases } = await deps.prisma.$transaction(async (tx) => {
    await tx.$queryRaw`SELECT id FROM threads WHERE id = ${target.threadId} FOR UPDATE`;
    const cancelled = await tx.run.updateManyAndReturn({
      where: {
        threadId: target.threadId,
        status: { in: [...ACTIVE_RUN_STATUSES] },
      },
      data: { status: "cancelled", completedAt: new Date() },
      select: { id: true },
    });
    const ids = cancelled.map((run) => run.id);
    await tx.steeringMessage.deleteMany({
      where: {
        botId: { in: target.kind === "bot" ? [target.botId] : target.memberBotIds },
        message: { threadId: target.threadId },
      },
    });
    // Snapshot teardown coordinates before commit. Once cancellation becomes
    // visible, a worker can release its lease / execution columns immediately; a
    // later lookup would then miss the sandbox work this request must stop.
    // Team ownership lives on ComputerExecutionLease; Computer.executionRunId is
    // only a legacy secondary path and is not written by current acquisition.
    const leases = ids.length
      ? await tx.computerExecutionLease.findMany({
          where: { runId: { in: ids } },
          select: { computerId: true, botId: true, runId: true, fence: true },
        })
      : [];
    const leaseComputerIds = [...new Set(leases.map((lease) => lease.computerId))];
    const computers = ids.length
      ? await tx.computer.findMany({
          where:
            leaseComputerIds.length > 0
              ? {
                  OR: [{ id: { in: leaseComputerIds } }, { executionRunId: { in: ids } }],
                }
              : { executionRunId: { in: ids } },
          select: {
            id: true,
            homeKey: true,
            kind: true,
            providerRef: true,
            executionBotId: true,
            executionRunId: true,
          },
        })
      : [];
    return { runIds: ids, computers, leases };
  });
  // Keep the DB lease until after teardown so a replacement run cannot claim the
  // screen while we still need the cancelled run's screenLeaseId to release it.
  const computerById = new Map(computers.map((computer) => [computer.id, computer]));
  const teardownTargets: Array<{
    computer: (typeof computers)[number];
    botId: string;
    runId: string;
    lease: (typeof leases)[number] | null;
  }> = [];
  const seenTargets = new Set<string>();
  for (const lease of leases) {
    const computer = computerById.get(lease.computerId);
    if (!computer) continue;
    const key = `${computer.id}:${lease.runId}`;
    if (seenTargets.has(key)) continue;
    seenTargets.add(key);
    teardownTargets.push({
      computer,
      botId: lease.botId,
      runId: lease.runId,
      lease,
    });
  }
  const cancelledRunIds = new Set(runIds);
  for (const computer of computers) {
    if (!computer.executionBotId || !computer.executionRunId) continue;
    // Computers may enter the snapshot via a cancelled lease while
    // Computer.executionRunId still points at an unrelated live run. Only treat
    // the legacy columns as a teardown target when they belong to a run we
    // cancelled in this transaction.
    if (!cancelledRunIds.has(computer.executionRunId)) continue;
    const key = `${computer.id}:${computer.executionRunId}`;
    if (seenTargets.has(key)) continue;
    seenTargets.add(key);
    teardownTargets.push({
      computer,
      botId: computer.executionBotId,
      runId: computer.executionRunId,
      lease: null,
    });
  }
  await Promise.all(
    teardownTargets.map(async ({ computer, botId, runId, lease }) => {
      if (!computer.providerRef) return;
      const context = {
        operationId: "stop",
        traceId: "stop",
        spaceId: actor.spaceId,
        userId: actor.userId,
        botId,
        runId,
        screenLeaseId: screenLeaseIdForRun(lease, runId),
        cancelRunWork: true,
        signal: new AbortController().signal,
      };
      const ref = toComputerRef(computer);
      await cancelComputerRunWork(deps.sandbox, ref, computer.id, runId, context);
      await deps.sandbox.releaseScreen?.(ref, context).catch(() => undefined);
    }),
  );
  await expireComputerExecutionLeases(deps.prisma, { runId: { in: runIds } });
  await deps.prisma.computer.updateMany({
    where: { executionRunId: { in: runIds } },
    data: {
      executionRunId: null,
      executionBotId: null,
      executionLeaseExpiresAt: null,
    },
  });
  await deps.prisma.event.deleteMany({
    where: {
      type: "thread.progress",
      runId: { in: runIds },
    },
  });
}

export async function setThreadUnreadState(
  prisma: PrismaClient,
  actor: Actor,
  target: ThreadTarget,
  unread: boolean,
) {
  const result = await prisma.thread.updateMany({
    where: {
      id: target.threadId,
      spaceId: actor.spaceId,
      userId: actor.userId,
      unread: { not: unread },
    },
    data: { unread },
  });
  if (result.count > 1) throw new IsolationError();
}