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 <cynarAI@users.noreply.github.com>
This commit is contained in:
parent
4264ef682f
commit
57f3ee4958
@ -182,6 +182,7 @@ import {
|
|||||||
shouldForwardPeerThreadEvent,
|
shouldForwardPeerThreadEvent,
|
||||||
} from "./thread-message-pages.js";
|
} from "./thread-message-pages.js";
|
||||||
import {
|
import {
|
||||||
|
followUpGroupThreadMessage,
|
||||||
reactToThreadMessage,
|
reactToThreadMessage,
|
||||||
resolveThreadTarget,
|
resolveThreadTarget,
|
||||||
sendThreadMessage,
|
sendThreadMessage,
|
||||||
@ -1545,79 +1546,8 @@ export function createRouter(deps: RouterDeps) {
|
|||||||
}
|
}
|
||||||
return { ok: true as const };
|
return { ok: true as const };
|
||||||
}
|
}
|
||||||
const committed = await deps.prisma.$transaction(async (tx) => {
|
const committed = await followUpGroupThreadMessage(deps, context.actor, target, {
|
||||||
await lockOwnedGroup(tx, context.actor, target.groupId);
|
text: input.text,
|
||||||
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 };
|
|
||||||
});
|
});
|
||||||
await deps.events.notify(target.threadId, committed.eventSeq).catch((error) => {
|
await deps.events.notify(target.threadId, committed.eventSeq).catch((error) => {
|
||||||
getLogger().error("group follow-up realtime notification", error);
|
getLogger().error("group follow-up realtime notification", error);
|
||||||
|
|||||||
@ -4,7 +4,9 @@ import type { PrismaClient } from "@rakazo/db";
|
|||||||
import { describe, expect, it, vi } from "vitest";
|
import { describe, expect, it, vi } from "vitest";
|
||||||
import {
|
import {
|
||||||
cancelSupersededQueuedRuns,
|
cancelSupersededQueuedRuns,
|
||||||
|
followUpGroupThreadMessage,
|
||||||
reactToThreadMessage,
|
reactToThreadMessage,
|
||||||
|
resolveGroupFollowUpBotId,
|
||||||
sendThreadMessage,
|
sendThreadMessage,
|
||||||
stopThreadRuns,
|
stopThreadRuns,
|
||||||
type ThreadTarget,
|
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", () => {
|
describe("stopThreadRuns", () => {
|
||||||
it("snapshots every lease when group members share a team computer", async () => {
|
it("snapshots every lease when group members share a team computer", async () => {
|
||||||
const releaseScreen = vi.fn().mockResolvedValue(undefined);
|
const releaseScreen = vi.fn().mockResolvedValue(undefined);
|
||||||
|
|||||||
@ -949,6 +949,114 @@ export async function sendThreadMessage(
|
|||||||
return sendResult(committed.message, 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.
|
* 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.
|
* Clients render it beneath that message; the reaction itself does not start an AI turn.
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user