diff --git a/packages/adapters/src/group-handoff.ts b/packages/adapters/src/group-handoff.ts index 800068d..ba11416 100644 --- a/packages/adapters/src/group-handoff.ts +++ b/packages/adapters/src/group-handoff.ts @@ -13,6 +13,7 @@ import { lockOwnedGroup, type PrismaClient, touchGroupUpdatedAt, + withTransactionRetry, } from "@rakazo/db"; import { getLogger } from "@rakazo/logging"; import type { ExecutorDeps } from "./executor.js"; @@ -30,180 +31,185 @@ export async function handoffToGroupBot( input: { bot_id?: string; confirm_name?: string; message: string }, ) { const deliveryKey = `group-handoff:${run.id}`; - const committed = await deps.prisma.$transaction(async (tx) => { - try { - await lockOwnedGroup(tx, run, groupId); - } catch (error) { - if (error instanceof IsolationError) - return { error: "group is no longer available" } as const; - throw error; - } - const [group, activeSource] = await Promise.all([ - tx.chatGroup.findFirst({ - where: { id: groupId, archivedAt: null, thread: { id: run.threadId } }, - include: { - members: { - where: { bot: { archivedAt: null } }, - include: { bot: { select: { id: true, name: true } } }, + // Runs finishing on the same thread contend for its event sequence; a deadlock here used to + // fail a specialist whose work was already done. + const committed = await withTransactionRetry(() => + deps.prisma.$transaction(async (tx) => { + try { + await lockOwnedGroup(tx, run, groupId); + } catch (error) { + if (error instanceof IsolationError) + return { error: "group is no longer available" } as const; + throw error; + } + const [group, activeSource] = await Promise.all([ + tx.chatGroup.findFirst({ + where: { id: groupId, archivedAt: null, thread: { id: run.threadId } }, + include: { + members: { + where: { bot: { archivedAt: null } }, + include: { bot: { select: { id: true, name: true } } }, + orderBy: { createdAt: "asc" }, + }, + }, + }), + tx.run.findFirst({ + where: { + id: run.id, + spaceId: run.spaceId, + threadId: run.threadId, + botId: run.botId, + userId: run.userId, + status: "running", + }, + select: { id: true, sourceMessage: { select: { blocks: true } } }, + }), + ]); + if (!group || !activeSource) return { error: "source run is no longer active" } as const; + if (!group.members.some((member) => member.bot.id === run.botId)) { + return { error: "source bot is no longer a group member" } as const; + } + // In a group with a lead, members hand results back to the lead and the lead delegates + // the next stage, so the bounce rules below only apply between members. + const leadBotId = group.members.some((member) => member.bot.id === group.leadBotId) + ? group.leadBotId + : null; + + const existing = await tx.message.findUnique({ + where: { threadId_clientNonce: { threadId: run.threadId, clientNonce: deliveryKey } }, + select: { + sourceRuns: { orderBy: { createdAt: "asc" }, + take: 1, + select: { id: true, botId: true }, }, }, - }), - tx.run.findFirst({ - where: { - id: run.id, - spaceId: run.spaceId, - threadId: run.threadId, - botId: run.botId, - userId: run.userId, - status: "running", - }, - select: { id: true, sourceMessage: { select: { blocks: true } } }, - }), - ]); - if (!group || !activeSource) return { error: "source run is no longer active" } as const; - if (!group.members.some((member) => member.bot.id === run.botId)) { - return { error: "source bot is no longer a group member" } as const; - } - // In a group with a lead, members hand results back to the lead and the lead delegates - // the next stage, so the bounce rules below only apply between members. - const leadBotId = group.members.some((member) => member.bot.id === group.leadBotId) - ? group.leadBotId - : null; - - const existing = await tx.message.findUnique({ - where: { threadId_clientNonce: { threadId: run.threadId, clientNonce: deliveryKey } }, - select: { - sourceRuns: { - orderBy: { createdAt: "asc" }, - take: 1, - select: { id: true, botId: true }, - }, - }, - }); - if (existing) { - const nextRun = existing.sourceRuns[0]; - const event = await tx.event.findFirst({ - where: { threadId: run.threadId, runId: run.id, type: "group.handoff" }, - orderBy: { seq: "desc" }, - select: { seq: true }, }); - if (!nextRun || !event) return { error: "recorded handoff is incomplete" } as const; - return { - ok: true, - botId: nextRun.botId, - runId: nextRun.id, - eventSeq: event.seq, - toLead: nextRun.botId === leadBotId, - } as const; - } - - let targetId = input.bot_id?.trim(); - if (!targetId && input.confirm_name?.trim()) { - const name = input.confirm_name.trim().toLowerCase(); - targetId = group.members.find((member) => member.bot.name.toLowerCase() === name)?.bot.id; - } - if (!targetId) return { error: "handoff target bot is required" } as const; - if (targetId === run.botId) return { error: "cannot hand off to yourself" } as const; - if (!group.members.some((member) => member.bot.id === targetId)) { - return { error: "handoff target is not a group member" } as const; - } - - let sourceBlocks: MessageBlock[] = []; - if (activeSource.sourceMessage) { - const parsedSource = MessageBlock.array().safeParse(activeSource.sourceMessage.blocks); - if (!parsedSource.success) { - return { error: "cannot verify the group handoff chain" } as const; + if (existing) { + const nextRun = existing.sourceRuns[0]; + const event = await tx.event.findFirst({ + where: { threadId: run.threadId, runId: run.id, type: "group.handoff" }, + orderBy: { seq: "desc" }, + select: { seq: true }, + }); + if (!nextRun || !event) return { error: "recorded handoff is incomplete" } as const; + return { + ok: true, + botId: nextRun.botId, + runId: nextRun.id, + eventSeq: event.seq, + toLead: nextRun.botId === leadBotId, + } as const; } - sourceBlocks = parsedSource.data; - } - const sourceHandoff = sourceBlocks.find( - (block): block is Extract => block.kind === "handoff", - ); - const leadInvolved = leadBotId !== null && (targetId === leadBotId || run.botId === leadBotId); - if (sourceHandoff?.fromBotId === targetId && !leadInvolved) { - return { - error: - "do not hand this stage back to its sender; post the result in the shared thread instead", - } as const; - } - if ( - run.botId !== leadBotId && - sourceHandoff?.toBotId === run.botId && - isSameStageHandoffBounce(sourceHandoff.text, input.message) - ) { - return { - error: - "you own this handed-off stage; workshop with the user and post your concept here instead of re-delegating the same task", - } as const; - } - const hop = nextBotMessageHop(sourceHandoff?.hop); - if (botMessageHopExhausted(hop)) { - return { - error: - "group handoff limit reached for this chain; finish the current stage in the shared thread instead", - } as const; - } - const handoffBlock: MessageBlock = { - kind: "handoff", - fromBotId: run.botId, - toBotId: targetId, - text: input.message, - hop, - }; - const message = await createThreadMessageInTransaction(tx, { - threadId: run.threadId, - role: "bot", - blocks: [handoffBlock], - botId: run.botId, - runId: run.id, - clientNonce: deliveryKey, - }); - const task = await tx.task.create({ - data: { - spaceId: run.spaceId, - botId: targetId, - threadId: run.threadId, - userId: run.userId, - prompt: input.message, - status: "queued", - }, - }); - const nextRun = await tx.run.create({ - data: { - spaceId: run.spaceId, - botId: targetId, - threadId: run.threadId, - taskId: task.id, - userId: run.userId, - status: "queued", - trigger: "follow_up", - sourceMessageId: message.id, - }, - }); - const event = await appendEventInTransaction(tx, { - spaceId: run.spaceId, - threadId: run.threadId, - botId: run.botId, - type: "group.handoff", - runId: run.id, - payload: { - messageId: message.id, + let targetId = input.bot_id?.trim(); + if (!targetId && input.confirm_name?.trim()) { + const name = input.confirm_name.trim().toLowerCase(); + targetId = group.members.find((member) => member.bot.name.toLowerCase() === name)?.bot.id; + } + if (!targetId) return { error: "handoff target bot is required" } as const; + if (targetId === run.botId) return { error: "cannot hand off to yourself" } as const; + if (!group.members.some((member) => member.bot.id === targetId)) { + return { error: "handoff target is not a group member" } as const; + } + + let sourceBlocks: MessageBlock[] = []; + if (activeSource.sourceMessage) { + const parsedSource = MessageBlock.array().safeParse(activeSource.sourceMessage.blocks); + if (!parsedSource.success) { + return { error: "cannot verify the group handoff chain" } as const; + } + sourceBlocks = parsedSource.data; + } + const sourceHandoff = sourceBlocks.find( + (block): block is Extract => block.kind === "handoff", + ); + const leadInvolved = + leadBotId !== null && (targetId === leadBotId || run.botId === leadBotId); + if (sourceHandoff?.fromBotId === targetId && !leadInvolved) { + return { + error: + "do not hand this stage back to its sender; post the result in the shared thread instead", + } as const; + } + if ( + run.botId !== leadBotId && + sourceHandoff?.toBotId === run.botId && + isSameStageHandoffBounce(sourceHandoff.text, input.message) + ) { + return { + error: + "you own this handed-off stage; workshop with the user and post your concept here instead of re-delegating the same task", + } as const; + } + const hop = nextBotMessageHop(sourceHandoff?.hop); + if (botMessageHopExhausted(hop)) { + return { + error: + "group handoff limit reached for this chain; finish the current stage in the shared thread instead", + } as const; + } + + const handoffBlock: MessageBlock = { + kind: "handoff", fromBotId: run.botId, toBotId: targetId, text: input.message, - }, - }); - await touchGroupUpdatedAt(tx, groupId); - return { - ok: true, - botId: targetId, - runId: nextRun.id, - eventSeq: event.seq, - toLead: targetId === leadBotId, - } as const; - }); + hop, + }; + const message = await createThreadMessageInTransaction(tx, { + threadId: run.threadId, + role: "bot", + blocks: [handoffBlock], + botId: run.botId, + runId: run.id, + clientNonce: deliveryKey, + }); + const task = await tx.task.create({ + data: { + spaceId: run.spaceId, + botId: targetId, + threadId: run.threadId, + userId: run.userId, + prompt: input.message, + status: "queued", + }, + }); + const nextRun = await tx.run.create({ + data: { + spaceId: run.spaceId, + botId: targetId, + threadId: run.threadId, + taskId: task.id, + userId: run.userId, + status: "queued", + trigger: "follow_up", + sourceMessageId: message.id, + }, + }); + const event = await appendEventInTransaction(tx, { + spaceId: run.spaceId, + threadId: run.threadId, + botId: run.botId, + type: "group.handoff", + runId: run.id, + payload: { + messageId: message.id, + fromBotId: run.botId, + toBotId: targetId, + text: input.message, + }, + }); + await touchGroupUpdatedAt(tx, groupId); + return { + ok: true, + botId: targetId, + runId: nextRun.id, + eventSeq: event.seq, + toLead: targetId === leadBotId, + } as const; + }), + ); if ("error" in committed) return committed; await deps.events.notify(run.threadId, committed.eventSeq).catch((error) => { getLogger().error("group handoff realtime notification", error); diff --git a/packages/adapters/src/history-compaction.test.ts b/packages/adapters/src/history-compaction.test.ts index b130201..44e0442 100644 --- a/packages/adapters/src/history-compaction.test.ts +++ b/packages/adapters/src/history-compaction.test.ts @@ -241,6 +241,7 @@ function compactionHarness( defaultMemoryScope: string; } | null; timeZone?: string | null; + groupLeadBotId?: string | null; } = {}, ) { const messages: HarnessMessage[] = @@ -252,7 +253,8 @@ function compactionHarness( })); const thread = { id: "thread-1", - botId: "bot-1", + botId: options.groupLeadBotId === undefined ? ("bot-1" as string | null) : null, + group: options.groupLeadBotId === undefined ? null : { leadBotId: options.groupLeadBotId }, spaceId: "workspace-1", userId: "user-1", nextEventSeq: 0, @@ -390,6 +392,33 @@ function compactionHarness( } describe("compactHistory", () => { + it("summarizes a group thread with its lead and keeps the summary local", async () => { + const harness = compactionHarness({ + deploymentModelKey: "openrouter-key", + groupLeadBotId: "lead-bot", + }); + + await compactHistory(harness.deps, "thread-1"); + + const [request] = harness.runtime.run.mock.calls[0]!; + expect(request.botId).toBe("lead-bot"); + expect(harness.thread.historyCompactedUpToSeq).toBe(49); + expect(harness.thread.historyCompactionSummary).toBe("Summary of 50 messages."); + expect(harness.memoryProviders.resolve).not.toHaveBeenCalled(); + expect(harness.saveMemory).not.toHaveBeenCalled(); + }); + + it("skips a group thread without a lead", async () => { + const harness = compactionHarness({ + deploymentModelKey: "openrouter-key", + groupLeadBotId: null, + }); + + await compactHistory(harness.deps, "thread-1"); + + expect(harness.runtime.run).not.toHaveBeenCalled(); + }); + it("anchors the summarizer in the thread owner's time zone", async () => { const harness = compactionHarness({ deploymentModelKey: "openrouter-key", diff --git a/packages/adapters/src/history-compaction.ts b/packages/adapters/src/history-compaction.ts index 49970da..7356289 100644 --- a/packages/adapters/src/history-compaction.ts +++ b/packages/adapters/src/history-compaction.ts @@ -147,8 +147,14 @@ export interface CompactHistoryDeps { } export async function compactHistory(deps: CompactHistoryDeps, threadId: string): Promise { - const thread = await deps.prisma.thread.findUniqueOrThrow({ where: { id: threadId } }); - if (!thread.botId) return; + const thread = await deps.prisma.thread.findUniqueOrThrow({ + where: { id: threadId }, + include: { group: { select: { leadBotId: true } } }, + }); + // A group thread (a project's main conversation included) is summarized by its lead. Its + // summary stays local: the external memory store is keyed by one bot's own thread. + const summarizerBotId = thread.botId ?? thread.group?.leadBotId; + if (!summarizerBotId) return; const previousCursor = thread.historyCompactedUpToSeq; const previousGeneration = thread.historyCompactionGeneration; const previousSummary = thread.historyCompactionSummary?.trim() || null; @@ -267,7 +273,7 @@ export async function compactHistory(deps: CompactHistoryDeps, threadId: string) ? await deps.resolveModel({ userId: thread.userId, spaceId: thread.spaceId, - botId: thread.botId, + botId: summarizerBotId, }) : deps.deploymentModelKey ? { @@ -296,7 +302,7 @@ export async function compactHistory(deps: CompactHistoryDeps, threadId: string) let runtimeReportedFailure = false; for await (const event of deps.runtime.run( { - botId: thread.botId, + botId: summarizerBotId, threadId, runId: `compact:${threadId}:${fromSeqExclusive}`, prompt, @@ -353,12 +359,14 @@ export async function compactHistory(deps: CompactHistoryDeps, threadId: string) // Local compaction is first-party behavior and must not depend on an optional external store. // Saving only after the compare-and-set also prevents losing workers from creating duplicates. let semanticMemory: ConfiguredMemoryProvider | null = null; - try { - semanticMemory = await deps.memoryProviders.resolve( - memoryProviderResolveRequest(thread.spaceId, thread.userId), - ); - } catch (error) { - getLogger().error("Failed to load semantic memory provider for history compaction", error); + if (thread.botId) { + try { + semanticMemory = await deps.memoryProviders.resolve( + memoryProviderResolveRequest(thread.spaceId, thread.userId), + ); + } catch (error) { + getLogger().error("Failed to load semantic memory provider for history compaction", error); + } } let externalSaveAttempted = false; const memoryContext = { @@ -366,7 +374,7 @@ export async function compactHistory(deps: CompactHistoryDeps, threadId: string) traceId: `history-compact:${threadId}`, spaceId: thread.spaceId, userId: thread.userId, - botId: thread.botId, + botId: summarizerBotId, signal: new AbortController().signal, }; if (semanticMemory) { @@ -376,7 +384,7 @@ export async function compactHistory(deps: CompactHistoryDeps, threadId: string) { content: summary, scope: "isolated", - botId: thread.botId, + botId: summarizerBotId, source: { kind: "history", generation: previousGeneration }, }, memoryContext, @@ -407,7 +415,7 @@ export async function compactHistory(deps: CompactHistoryDeps, threadId: string) ) { try { const removed = await semanticMemory.provider.purgeHistory( - { botId: thread.botId, generations: [previousGeneration] }, + { botId: summarizerBotId, generations: [previousGeneration] }, memoryContext, ); if (!removed.ok) {