57f3ee4958
Group threads/followUp always steered to members[0], so a follow-up while a later member had waiting_input went to the wrong bot or started a fresh run on the first member instead of continuing the ask. Resolve the target bot like group sends: prefer waiting ask/takeover on any member, then the newest steerable active run, then @mentions, then the default first member. Extract followUpGroupThreadMessage and add regression tests. Co-authored-by: cynarAI <cynarAI@users.noreply.github.com>
1261 lines
43 KiB
TypeScript
1261 lines
43 KiB
TypeScript
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 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 { 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;
|
||
members: GroupMember[];
|
||
memberBotIds: string[];
|
||
};
|
||
|
||
/**
|
||
* 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(/ /g, " ")
|
||
.replace(/&/g, "&")
|
||
.replace(/</g, "<")
|
||
.replace(/>/g, ">")
|
||
.replace(/"/g, '"')
|
||
.replace(/�?39;|'/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" },
|
||
},
|
||
},
|
||
});
|
||
if (!group || group.members.length < GROUP_MEMBER_MIN) 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,
|
||
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,
|
||
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;
|
||
// 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,
|
||
});
|
||
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.
|
||
*/
|
||
export function resolveGroupFollowUpBotId(input: {
|
||
text: string;
|
||
members: Array<{ botId: string; name: string }>;
|
||
activeRuns: Array<{ botId: string; status: string; createdAt: Date }>;
|
||
}): 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: [],
|
||
})[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 });
|
||
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 : target.memberBotIds[0];
|
||
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();
|
||
}
|