From 57f3ee49589a7be4f8378e53e13889f7491c581f Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 22 Sep 2026 06:03:59 +0000 Subject: [PATCH] fix(api): route group follow-ups to the waiting member bot 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 --- apps/api/src/router.ts | 76 +-------------- apps/api/src/thread-target.test.ts | 147 +++++++++++++++++++++++++++++ apps/api/src/thread-target.ts | 108 +++++++++++++++++++++ 3 files changed, 258 insertions(+), 73 deletions(-) diff --git a/apps/api/src/router.ts b/apps/api/src/router.ts index 240298f..4fa82cb 100644 --- a/apps/api/src/router.ts +++ b/apps/api/src/router.ts @@ -182,6 +182,7 @@ import { shouldForwardPeerThreadEvent, } from "./thread-message-pages.js"; import { + followUpGroupThreadMessage, reactToThreadMessage, resolveThreadTarget, sendThreadMessage, @@ -1545,79 +1546,8 @@ export function createRouter(deps: RouterDeps) { } return { ok: true as const }; } - const committed = await deps.prisma.$transaction(async (tx) => { - await lockOwnedGroup(tx, context.actor, target.groupId); - const group = await tx.chatGroup.findFirst({ - where: { - id: target.groupId, - archivedAt: null, - thread: { id: target.threadId }, - }, - include: { members: { orderBy: { createdAt: "asc" } } }, - }); - const botId = group?.members[0]?.botId; - if (!botId) throw new IsolationError(); - const blocks = [{ kind: "text" as const, text: input.text }]; - const message = await createThreadMessageInTransaction(tx, { - threadId: target.threadId, - role: "user", - blocks, - }); - const active = await tx.run.findFirst({ - where: { - threadId: target.threadId, - botId, - status: { in: [...ACTIVE_RUN_STATUSES] }, - }, - select: { id: true }, - }); - let run: { id: string } | null = null; - if (!active) { - const task = await tx.task.create({ - data: { - spaceId: context.actor.spaceId, - botId, - threadId: target.threadId, - userId: context.actor.userId, - prompt: input.text, - status: "queued", - }, - }); - run = await tx.run.create({ - data: { - spaceId: context.actor.spaceId, - botId, - threadId: target.threadId, - taskId: task.id, - userId: context.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: context.actor.userId, - runId: active.id, - }, - }); - await tx.message.update({ where: { id: message.id }, data: { runId: active.id } }); - } - const event = await appendEventInTransaction(tx, { - spaceId: context.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 }; + const committed = await followUpGroupThreadMessage(deps, context.actor, target, { + text: input.text, }); await deps.events.notify(target.threadId, committed.eventSeq).catch((error) => { getLogger().error("group follow-up realtime notification", error); diff --git a/apps/api/src/thread-target.test.ts b/apps/api/src/thread-target.test.ts index cc0e949..14deefc 100644 --- a/apps/api/src/thread-target.test.ts +++ b/apps/api/src/thread-target.test.ts @@ -4,7 +4,9 @@ import type { PrismaClient } from "@rakazo/db"; import { describe, expect, it, vi } from "vitest"; import { cancelSupersededQueuedRuns, + followUpGroupThreadMessage, reactToThreadMessage, + resolveGroupFollowUpBotId, sendThreadMessage, stopThreadRuns, type ThreadTarget, @@ -1612,6 +1614,151 @@ describe("sendThreadMessage", () => { }); }); +describe("resolveGroupFollowUpBotId", () => { + const members = [ + { botId: "bot-a", name: "Alpha" }, + { botId: "bot-b", name: "Beta" }, + ]; + + it("prefers a waiting ask on a later member over the first member's busy run", () => { + expect( + resolveGroupFollowUpBotId({ + text: "continue", + members, + activeRuns: [ + { + botId: "bot-a", + status: "running", + createdAt: new Date("2026-09-22T12:00:00.000Z"), + }, + { + botId: "bot-b", + status: "waiting_input", + createdAt: new Date("2026-09-22T11:00:00.000Z"), + }, + ], + }), + ).toBe("bot-b"); + }); + + it("falls back to the default first member when no member has an active run", () => { + expect( + resolveGroupFollowUpBotId({ + text: "continue", + members, + activeRuns: [], + }), + ).toBe("bot-a"); + }); + + it("honors @mentions when no member has an active run", () => { + expect( + resolveGroupFollowUpBotId({ + text: "@Beta keep going", + members, + activeRuns: [], + }), + ).toBe("bot-b"); + }); +}); + +describe("followUpGroupThreadMessage", () => { + it("steers the follow-up to a waiting member instead of the first member", async () => { + let messageSeq = 0; + let eventSeq = 0; + const tx = { + $queryRaw: vi.fn().mockResolvedValue([{ id: "group-1" }]), + thread: { + update: vi.fn(async ({ data }: { data: { nextMessageSeq?: unknown } }) => + data.nextMessageSeq ? { nextMessageSeq: ++messageSeq } : { nextEventSeq: ++eventSeq }, + ), + }, + chatGroup: { + findFirst: vi.fn().mockResolvedValue({ + id: "group-1", + members: [ + { bot: { id: "bot-a", name: "Alpha", color: null } }, + { bot: { id: "bot-b", name: "Beta", color: null } }, + ], + }), + update: vi.fn().mockResolvedValue({ id: "group-1" }), + }, + message: { + create: vi.fn().mockResolvedValue({ + id: "msg-1", + threadId: "thread-1", + seq: 1, + role: "user", + blocks: [{ kind: "text", text: "continue" }], + botId: null, + replyToMessageId: null, + runId: null, + createdAt: new Date(), + }), + update: vi.fn(), + }, + run: { + findMany: vi.fn().mockResolvedValue([ + { + id: "run-a", + botId: "bot-a", + status: "running", + createdAt: new Date("2026-09-22T12:00:00.000Z"), + }, + { + id: "run-b", + botId: "bot-b", + status: "waiting_input", + createdAt: new Date("2026-09-22T11:00:00.000Z"), + }, + ]), + findUnique: vi.fn().mockResolvedValue({ + status: "waiting_input", + startedAt: new Date(), + }), + create: vi.fn(), + }, + task: { create: vi.fn() }, + steeringMessage: { create: vi.fn() }, + event: { + create: vi.fn().mockResolvedValue({ id: "event-1", seq: 1, createdAt: new Date() }), + }, + }; + const prisma = { + $transaction: vi.fn(async (callback: (client: typeof tx) => unknown) => callback(tx)), + } as unknown as PrismaClient; + const actor = { spaceId: "workspace-1", userId: "user-1" } as Actor; + const target = { + kind: "group", + groupId: "group-1", + groupName: "Group", + threadId: "thread-1", + members: [], + memberBotIds: ["bot-a", "bot-b"], + } satisfies ThreadTarget; + + await expect( + followUpGroupThreadMessage({ prisma }, actor, target, { text: "continue" }), + ).resolves.toEqual({ runId: undefined, eventSeq: 1 }); + + expect(tx.steeringMessage.create).toHaveBeenCalledWith({ + data: { + messageId: "msg-1", + botId: "bot-b", + userId: "user-1", + runId: "run-b", + }, + }); + expect(tx.run.create).not.toHaveBeenCalled(); + expect(tx.event.create).toHaveBeenCalledWith({ + data: expect.objectContaining({ + botId: "bot-b", + runId: "run-b", + }), + }); + }); +}); + describe("stopThreadRuns", () => { it("snapshots every lease when group members share a team computer", async () => { const releaseScreen = vi.fn().mockResolvedValue(undefined); diff --git a/apps/api/src/thread-target.ts b/apps/api/src/thread-target.ts index fb309e4..9acf19b 100644 --- a/apps/api/src/thread-target.ts +++ b/apps/api/src/thread-target.ts @@ -949,6 +949,114 @@ export async function sendThreadMessage( 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, + 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.