Summarize project conversations and retry contended handoffs
Group threads, which hold every project's main conversation, were never compacted. Runs there saw only the last 200 raw messages, so the original brief and earlier decisions fell out of context. The lead bot now summarizes them like a bot thread; the summary stays local. A group handoff now retries when its transaction hits a deadlock, so a specialist that finished its work no longer fails at the hand-back. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_011VfFvSskgrByEU75QLyreE
This commit is contained in:
Claude
parent
aa989f50b3
commit
8f56426ea7
@ -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<MessageBlock, { kind: "handoff" }> => 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<MessageBlock, { kind: "handoff" }> => 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);
|
||||
|
||||
@ -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",
|
||||
|
||||
@ -147,8 +147,14 @@ export interface CompactHistoryDeps {
|
||||
}
|
||||
|
||||
export async function compactHistory(deps: CompactHistoryDeps, threadId: string): Promise<void> {
|
||||
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) {
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user