From f78c8919c7527f00e801b3a8d4deea6a5332dc1b Mon Sep 17 00:00:00 2001 From: gus <42593099+lightcloud00@users.noreply.github.com> Date: Tue, 25 Aug 2026 17:36:26 -0400 Subject: [PATCH] fix: queue busy bot delegations --- server/comms.test.ts | 29 ++++ server/decision-log.ts | 11 +- server/delegations.test.ts | 158 ++++++++++++++++-- server/delegations.ts | 322 +++++++++++++++++++++++++++++++------ server/index.ts | 105 ++++++++---- 5 files changed, 536 insertions(+), 89 deletions(-) diff --git a/server/comms.test.ts b/server/comms.test.ts index d07e35673..0ff944a6d 100644 --- a/server/comms.test.ts +++ b/server/comms.test.ts @@ -352,6 +352,35 @@ describe("comms e2e (fake ACP fleet)", () => { (m: any) => m.kind === "activity" && m.tool?.name === "Message from @Asker", ); expect(helperNote?.comm?.groupId).toBe(note.comm.groupId); + + // Queue and dispatch share one stable task id in the existing + // decision ledger. The row intentionally names only the target id and + // attempt count: prompts and provider output belong in the thread, + // never this fleet-wide audit stream. + const ledgerDeadline = Date.now() + 5_000; + let delegationRows: any[] = []; + for (;;) { + const decisions = await api("GET", "/api/decisions"); + expect(decisions.status).toBe(200); + delegationRows = decisions.body.decisions.filter( + (row: any) => row.threadId === asker.threadId && row.tool === "delegate_bot", + ); + if ( + delegationRows.some((row: any) => row.decision === "delegation-queued") && + delegationRows.some((row: any) => row.decision === "delegation-completed") + ) break; + if (Date.now() > ledgerDeadline) { + throw new Error(`delegation ledger never settled: ${JSON.stringify(delegationRows)}`); + } + await new Promise((resolve) => setTimeout(resolve, 50)); + } + const queuedRow = delegationRows.find((row: any) => row.decision === "delegation-queued"); + const completedRow = delegationRows.find((row: any) => row.decision === "delegation-completed"); + expect(completedRow.requestId).toBe(queuedRow.requestId); + expect(delegationRows.every((row: any) => row.source === "delegation")).toBe(true); + expect(delegationRows.every((row: any) => /^target:[^;]+;attempts:\d+$/.test(row.summary))).toBe(true); + expect(JSON.stringify(delegationRows)).not.toContain("delegated task"); + expect(JSON.stringify(delegationRows)).not.toContain("hello from fake acp"); expect(helperBot.busy).toBeFalsy(); expect(askerBot.busy).toBeFalsy(); }, diff --git a/server/decision-log.ts b/server/decision-log.ts index 3326cbfd1..9ba4236c8 100644 --- a/server/decision-log.ts +++ b/server/decision-log.ts @@ -26,13 +26,20 @@ import { join } from "node:path"; import type { AutoVerdictSource } from "./auto-approve.ts"; import { redactSecrets } from "./redact.ts"; -export type DecisionKind = "auto-approved" | "card-shown" | "user-approved" | "user-denied"; +export type DecisionKind = + | "auto-approved" + | "card-shown" + | "user-approved" + | "user-denied" + | "delegation-queued" + | "delegation-completed" + | "delegation-failed"; /** Who or what produced the decision. The AutoVerdictSource values carry * straight through from auto-approve.ts; `question` marks the cards a rule * may never answer, `auto-fallback` a card shown because an auto-approval * could not be delivered, and `user` the human's answer to a card. */ -export type DecisionSource = AutoVerdictSource | "question" | "auto-fallback" | "user"; +export type DecisionSource = AutoVerdictSource | "question" | "auto-fallback" | "user" | "delegation"; export interface DecisionRow { at: string; diff --git a/server/delegations.test.ts b/server/delegations.test.ts index 7996ccfc6..374cc3d89 100644 --- a/server/delegations.test.ts +++ b/server/delegations.test.ts @@ -11,9 +11,15 @@ import type { CommsBus } from "./comms-visibility.ts"; import { DATA_DIR } from "./config.ts"; import type { ModelSelection } from "./contracts.ts"; import { + _loadPending, + _resetPending, drainDelegations, + drainReadyDelegations, + discardDelegations, + pendingThreads, queueDelegation, _pendingCount, + type DelegationOutcome, } from "./delegations.ts"; import { peerAllowKey, resolvePeerComms } from "./peer-approval.ts"; import { Store, type BotRecord } from "./store.ts"; @@ -80,7 +86,7 @@ describe("queueDelegation", () => { message: "self-talk", depth: 0, }, 1); - expect(result).toBe("self"); + expect(result).toEqual({ state: "failed", reason: "self" }); expect(_pendingCount(from.threadId)).toBe(0); }); @@ -90,7 +96,7 @@ describe("queueDelegation", () => { message: "next task", depth: 1, }, 1); - expect(result).toBe("too_deep"); + expect(result).toEqual({ state: "failed", reason: "too_deep" }); expect(_pendingCount(from.threadId)).toBe(0); }); @@ -100,7 +106,7 @@ describe("queueDelegation", () => { message: "where?", depth: 0, }, 1); - expect(result).toBe("no_target"); + expect(result).toEqual({ state: "failed", reason: "no_target" }); expect(_pendingCount(from.threadId)).toBe(0); }); @@ -111,7 +117,7 @@ describe("queueDelegation", () => { reason: "followup", depth: 0, }, 1); - expect(result).toBe("ok"); + expect(result).toMatchObject({ state: "queued", duplicate: false }); expect(_pendingCount(from.threadId)).toBe(1); const chip = store @@ -131,6 +137,20 @@ describe("queueDelegation", () => { expect(broadcast).toBeTruthy(); }); + it("deduplicates repeated identical requests by one stable task id", () => { + const outcomes: DelegationOutcome[] = []; + const item = { toBotId: target.id, message: "do this once", reason: "same task", depth: 0 }; + const first = queueDelegation(commsBus, from, item, 1, from.threadId, (event) => outcomes.push(event)); + const retry = queueDelegation(commsBus, from, item, 1, from.threadId, (event) => outcomes.push(event)); + + expect(first).toMatchObject({ state: "queued", duplicate: false }); + expect(retry).toEqual({ state: "queued", taskId: first.state === "queued" ? first.taskId : "", duplicate: true }); + expect(_pendingCount(from.threadId)).toBe(1); + expect(outcomes).toHaveLength(1); + expect(outcomes[0]).toMatchObject({ state: "queued", reason: "source_turn_active" }); + expect(store.messagesFor(from.threadId).filter((m) => m.tool?.name?.startsWith("Delegated to @"))).toHaveLength(1); + }); + it("keys detached routine delegations to their real source thread", async () => { const routineTask = store.createTask(from.id, "Routine run", false)!; const result = queueDelegation( @@ -141,7 +161,7 @@ describe("queueDelegation", () => { routineTask.threadId, ); - expect(result).toBe("ok"); + expect(result).toMatchObject({ state: "queued", duplicate: false }); expect(_pendingCount(routineTask.threadId)).toBe(1); expect(_pendingCount(from.threadId)).toBe(0); expect( @@ -307,20 +327,58 @@ describe("drainDelegations", () => { expect(runTargetCalls).toEqual([]); }); - it("skips runTarget and emits a 'is busy' chip when the target is currently busy", async () => { + it("keeps one busy-target item queued and drains it once when the target becomes idle", async () => { + const outcomes: DelegationOutcome[] = []; store.patchBot(target.id, { busy: true }); - queueDelegation(commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1); + const item = { toBotId: target.id, message: "do this", depth: 0 }; + const first = queueDelegation( + commsBus, + from, + item, + 1, + from.threadId, + (event) => outcomes.push(event), + ); + const duplicate = queueDelegation( + commsBus, + from, + item, + 1, + from.threadId, + (event) => outcomes.push(event), + ); + expect(first).toMatchObject({ state: "queued", duplicate: false }); + expect(duplicate).toEqual({ + state: "queued", + taskId: first.state === "queued" ? first.taskId : "", + duplicate: true, + }); + expect(_pendingCount(from.threadId)).toBe(1); drainDelegations(commsBus, approvalBus, from.threadId, (toBotId, message, commsDepth) => { runTargetCalls.push({ toBotId, message, commsDepth }); - }); + }, (event) => outcomes.push(event)); const chip = await waitFor(() => store .messagesFor(from.threadId) - .find((m) => m.kind === "activity" && (m.tool?.name ?? "").includes("is busy")), + .find((m) => m.kind === "activity" && (m.tool?.name ?? "").includes("[target_busy]")), ); - expect(chip.tool?.name).toBe("Delegation to @Helper canceled — @Helper is busy"); - expect(chip.tool?.ok).toBe(false); + expect(chip.tool?.name).toContain("will pick it up when the active turn completes"); + expect(_pendingCount(from.threadId)).toBe(1); expect(runTargetCalls).toEqual([]); + await waitFor(() => outcomes.length === 2); + await new Promise((resolve) => setTimeout(resolve, 0)); + + store.patchBot(target.id, { busy: false }); + const runTarget = (toBotId: string, message: string, commsDepth: number) => { + runTargetCalls.push({ toBotId, message, commsDepth }); + }; + drainReadyDelegations(commsBus, approvalBus, runTarget, (event) => outcomes.push(event)); + await waitFor(() => runTargetCalls.length === 1 && _pendingCount(from.threadId) === 0); + drainReadyDelegations(commsBus, approvalBus, runTarget, (event) => outcomes.push(event)); + + expect(runTargetCalls).toHaveLength(1); + expect(outcomes.map((event) => event.state)).toEqual(["queued", "queued", "completed"]); + expect(outcomes[1]).toMatchObject({ reason: "target_busy", attempts: 1 }); }); it("asks for approval when approvePeerComms is on, then runs only on allow", async () => { @@ -402,10 +460,8 @@ describe("drainDelegations", () => { }); }); -import { existsSync, readFileSync } from "node:fs"; +import { existsSync, readFileSync, writeFileSync } from "node:fs"; import { join } from "node:path"; -import { _loadPending, _resetPending, discardDelegations, pendingThreads } from "./delegations.ts"; - describe("delegations survive a restart", () => { let store: Store; let from: BotRecord; @@ -425,7 +481,10 @@ describe("delegations survive a restart", () => { afterEach(() => _resetPending()); it("writes the queue to disk on queue, and clears it on drain and discard", async () => { - expect(queueDelegation(buses.commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1)).toBe("ok"); + expect(queueDelegation(buses.commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1)).toMatchObject({ + state: "queued", + duplicate: false, + }); expect(existsSync(file())).toBe(true); const onDisk = JSON.parse(readFileSync(file(), "utf8")) as Record; expect(onDisk[from.threadId]).toHaveLength(1); @@ -503,6 +562,75 @@ describe("delegations survive a restart", () => { expect(pendingThreads()).toEqual([]); }); + it("preserves a busy-target wait across restart and dispatches it once", async () => { + store.patchBot(target.id, { busy: true }); + queueDelegation(buses.commsBus, from, { toBotId: target.id, message: "after restart", depth: 0 }, 1); + drainDelegations(buses.commsBus, buses.approvalBus, from.threadId, () => { + throw new Error("busy work must not start"); + }); + await waitFor(() => _pendingCount(from.threadId) === 1); + + _resetPending(); + _loadPending(); + expect(_pendingCount(from.threadId)).toBe(1); + store.patchBot(target.id, { busy: false }); + const ran: string[] = []; + const runTarget = async (_to: string, message: string) => { + ran.push(message); + }; + drainReadyDelegations(buses.commsBus, buses.approvalBus, runTarget); + await waitFor(() => ran.length === 1 && _pendingCount(from.threadId) === 0); + drainReadyDelegations(buses.commsBus, buses.approvalBus, runTarget); + expect(ran).toHaveLength(1); + }); + + it("expires an aged busy-target item with a machine-readable reason", async () => { + const outcomes: DelegationOutcome[] = []; + store.patchBot(target.id, { busy: true }); + queueDelegation(buses.commsBus, from, { toBotId: target.id, message: "too old", depth: 0 }, 1); + drainDelegations(buses.commsBus, buses.approvalBus, from.threadId, () => {}); + await waitFor(() => _pendingCount(from.threadId) === 1); + + const onDisk = JSON.parse(readFileSync(file(), "utf8")) as Record>>; + onDisk[from.threadId]![0]!.queuedAt = "2000-01-01T00:00:00.000Z"; + writeFileSync(file(), JSON.stringify(onDisk, null, 2)); + _resetPending(); + _loadPending(); + + drainReadyDelegations(buses.commsBus, buses.approvalBus, () => {}, (event) => outcomes.push(event)); + await waitFor(() => outcomes.some((event) => event.reason === "expired")); + expect(_pendingCount(from.threadId)).toBe(0); + expect(outcomes.at(-1)).toMatchObject({ state: "failed", reason: "expired" }); + expect(store.messagesFor(from.threadId).some((m) => m.tool?.name.includes("[expired]"))).toBe(true); + }); + + it("bounds repeated busy races and reports retry_limit", async () => { + const outcomes: DelegationOutcome[] = []; + queueDelegation(buses.commsBus, from, { toBotId: target.id, message: "racy target", depth: 0 }, 1); + let calls = 0; + const busyRace = () => { + calls += 1; + throw Object.assign(new Error("the bot is already working"), { status: 409 }); + }; + + drainDelegations(buses.commsBus, buses.approvalBus, from.threadId, busyRace, (event) => outcomes.push(event)); + await waitFor(() => outcomes.length === 1); + await new Promise((resolve) => setTimeout(resolve, 0)); + drainReadyDelegations(buses.commsBus, buses.approvalBus, busyRace, (event) => outcomes.push(event)); + await waitFor(() => outcomes.length === 2); + await new Promise((resolve) => setTimeout(resolve, 0)); + drainReadyDelegations(buses.commsBus, buses.approvalBus, busyRace, (event) => outcomes.push(event)); + await waitFor(() => outcomes.length === 3); + + expect(calls).toBe(3); + expect(_pendingCount(from.threadId)).toBe(0); + expect(outcomes.map((event) => [event.state, event.reason, event.attempts])).toEqual([ + ["queued", "target_busy", 1], + ["queued", "target_busy", 2], + ["failed", "retry_limit", 3], + ]); + }); + it("tolerates a missing or corrupt file", () => { _resetPending(); _loadPending(); // no file diff --git a/server/delegations.ts b/server/delegations.ts index 2d6b717f8..7f8f1cee3 100644 --- a/server/delegations.ts +++ b/server/delegations.ts @@ -11,13 +11,13 @@ // time, never at queue time, because the user might have just turned // approvePeerComms on between queueing and draining. +import { createHash } from "node:crypto"; import { readFileSync } from "node:fs"; import { join } from "node:path"; import { writeFileAtomic } from "./atomic.ts"; import { getOrCreateChannel, mirrorExchange, type CommsBus } from "./comms-visibility.ts"; import { DATA_DIR } from "./config.ts"; -import { newId } from "./contracts.ts"; import { requestPeerApproval, type ApprovalBus } from "./peer-approval.ts"; import type { BotRecord, GroupRecord } from "./store.ts"; @@ -32,12 +32,44 @@ export interface DelegationItem { depth: number; } +type PendingPhase = "awaiting_source" | "waiting_target"; + interface PendingDelegationItem extends DelegationItem { - /** Stable acknowledgement key for crash-safe removal from the queue. */ + /** Stable task key used for both crash-safe acknowledgement and dedup. */ id: string; + queuedAt: string; + attempts: number; + phase: PendingPhase; + lastReason?: DelegationReason; +} + +export type QueueFailureReason = "no_target" | "self" | "too_deep" | "too_many"; +export type DelegationReason = + | "source_turn_active" + | "duplicate" + | "target_busy" + | "expired" + | "retry_limit" + | "target_missing" + | "source_missing" + | "source_turn_failed" + | "user_denied" + | "dispatch_failed"; + +export interface DelegationOutcome { + state: "completed" | "queued" | "failed"; + taskId: string; + sourceThreadId: string; + toBotId: string; + attempts: number; + reason?: DelegationReason; } -export type QueueResult = "ok" | "no_target" | "self" | "too_deep" | "too_many"; +export type DelegationRecorder = (outcome: DelegationOutcome) => void; + +export type QueueResult = + | { state: "queued"; taskId: string; duplicate: boolean } + | { state: "failed"; reason: QueueFailureReason }; /** Per source-thread queue. Persisted to delegations.json on every change * and reloaded at boot: a handoff queued right before a restart runs after @@ -47,6 +79,34 @@ export type QueueResult = "ok" | "no_target" | "self" | "too_deep" | "too_many"; const pendingDelegations = new Map(); const drainingThreads = new Set(); const DELEGATIONS_FILE = join(DATA_DIR, "delegations.json"); +const MAX_TARGET_BUSY_ATTEMPTS = 3; +const MAX_DELEGATION_AGE_MS = 15 * 60 * 1000; + +function taskIdFor(sourceThreadId: string, item: DelegationItem): string { + const digest = createHash("sha256") + .update(JSON.stringify([ + sourceThreadId, + item.toBotId, + item.message, + item.reason ?? "", + Math.max(0, Math.trunc(item.depth)), + ])) + .digest("hex"); + return `delegation-${digest.slice(0, 32)}`; +} + +function recordOutcome(record: DelegationRecorder | undefined, outcome: DelegationOutcome): void { + try { + record?.(outcome); + } catch (error) { + console.error("delegations: could not record outcome", error); + } +} + +function isExpired(item: PendingDelegationItem, now = Date.now()): boolean { + const queuedAt = Date.parse(item.queuedAt); + return !Number.isFinite(queuedAt) || now - queuedAt >= MAX_DELEGATION_AGE_MS; +} function savePending(): void { try { @@ -63,6 +123,7 @@ export function _loadPending(): void { const raw = JSON.parse(readFileSync(DELEGATIONS_FILE, "utf8")) as Record; for (const [threadId, list] of Object.entries(raw)) { if (!Array.isArray(list)) continue; + const seen = new Set(); const items = list.flatMap((value): PendingDelegationItem[] => { if (!value || typeof value !== "object") return []; const item = value as Partial; @@ -71,12 +132,29 @@ export function _loadPending(): void { typeof item.message !== "string" || !Number.isFinite(item.depth) ) return []; - return [{ - id: typeof item.id === "string" && item.id ? item.id : newId(), + const normalized: DelegationItem = { toBotId: item.toBotId, message: item.message, ...(typeof item.reason === "string" ? { reason: item.reason } : {}), depth: Math.max(0, Math.trunc(item.depth!)), + }; + // Older files used a random acknowledgement id. Re-derive every id + // from the task itself so the first retry after an upgrade dedupes too. + const id = taskIdFor(threadId, normalized); + if (seen.has(id)) return []; + seen.add(id); + const queuedAt = + typeof item.queuedAt === "string" && Number.isFinite(Date.parse(item.queuedAt)) + ? item.queuedAt + : new Date().toISOString(); + const phase: PendingPhase = item.phase === "waiting_target" ? "waiting_target" : "awaiting_source"; + return [{ + ...normalized, + id, + queuedAt, + attempts: Number.isFinite(item.attempts) ? Math.max(0, Math.trunc(item.attempts!)) : 0, + phase, + ...(typeof item.lastReason === "string" ? { lastReason: item.lastReason as DelegationReason } : {}), }]; }); if (items.length) pendingDelegations.set(threadId, items); @@ -103,17 +181,29 @@ export function queueDelegation( item: DelegationItem, maxDepth: number, sourceThreadId = from.threadId, + record?: DelegationRecorder, ): QueueResult { - if (item.toBotId === from.id) return "self"; - if (item.depth >= maxDepth) return "too_deep"; + if (item.toBotId === from.id) return { state: "failed", reason: "self" }; + if (item.depth >= maxDepth) return { state: "failed", reason: "too_deep" }; const target = bus.store.bot(item.toBotId); - if (!target) return "no_target"; + if (!target) return { state: "failed", reason: "no_target" }; const list = pendingDelegations.get(sourceThreadId) ?? []; + const id = taskIdFor(sourceThreadId, item); + if (list.some((pending) => pending.id === id)) { + return { state: "queued", taskId: id, duplicate: true }; + } // Async handoff removes the backpressure that ask_bot got for free by // making the caller wait. Without a cap, one turn can queue unboundedly // and fan out into as many real turns on the next settle. - if (list.length >= MAX_QUEUED_PER_THREAD) return "too_many"; - list.push({ ...item, id: newId() }); + if (list.length >= MAX_QUEUED_PER_THREAD) return { state: "failed", reason: "too_many" }; + const pending: PendingDelegationItem = { + ...item, + id, + queuedAt: new Date().toISOString(), + attempts: 0, + phase: "awaiting_source", + }; + list.push(pending); pendingDelegations.set(sourceThreadId, list); savePending(); const label = `Delegated to @${target.name}${item.reason ? `: ${item.reason}` : ""}`; @@ -122,7 +212,15 @@ export function queueDelegation( kind: "activity", tool: { name: label }, }); - return "ok"; + recordOutcome(record, { + state: "queued", + taskId: id, + sourceThreadId, + toBotId: item.toBotId, + attempts: 0, + reason: "source_turn_active", + }); + return { state: "queued", taskId: id, duplicate: false }; } /** Drain queued delegations for a source thread (called on its @@ -142,22 +240,66 @@ export function drainDelegations( sourceThreadId: string, channel?: GroupRecord, ) => void | Promise, + record?: DelegationRecorder, +): void { + drainMatching(bus, approvalBus, threadId, runTarget, record, () => true); +} + +/** Drain busy-target work only when an event says some turn completed. This + * scans persisted waits for targets that are now idle; it never polls and it + * never touches work whose source turn has not settled yet. */ +export function drainReadyDelegations( + bus: CommsBus, + approvalBus: ApprovalBus, + runTarget: Parameters[3], + record?: DelegationRecorder, +): void { + const now = Date.now(); + for (const [threadId, list] of pendingDelegations) { + const ready = list.some((item) => + item.phase === "waiting_target" && + (isExpired(item, now) || !bus.store.bot(item.toBotId)?.busy), + ); + if (!ready) continue; + drainMatching( + bus, + approvalBus, + threadId, + runTarget, + record, + (item) => item.phase === "waiting_target" && + (isExpired(item, now) || !bus.store.bot(item.toBotId)?.busy), + ); + } +} + +function drainMatching( + bus: CommsBus, + approvalBus: ApprovalBus, + threadId: string, + runTarget: Parameters[3], + record: DelegationRecorder | undefined, + include: (item: PendingDelegationItem) => boolean, ): void { if (drainingThreads.has(threadId)) return; const list = pendingDelegations.get(threadId); if (!list?.length) return; + const snapshot = list.filter(include); + if (!snapshot.length) return; const from = bus.store.botByThread(threadId); if (!from) { - pendingDelegations.delete(threadId); - savePending(); + for (const item of snapshot) { + acknowledgeDelegation(threadId, item.id); + recordOutcome(record, outcome(item, threadId, "failed", "source_missing")); + } return; } - const snapshot = [...list]; drainingThreads.add(threadId); void (async () => { for (const item of snapshot) { + let result: DelegationOutcome; try { - await processOne(bus, approvalBus, from, threadId, item, runTarget); + result = await processOne(bus, approvalBus, from, threadId, item, runTarget); } catch (error) { const why = error instanceof Error ? error.message : String(error); try { @@ -169,21 +311,40 @@ export function drainDelegations( } catch (reportError) { console.error("delegation failed and could not be reported", reportError); } - } finally { + result = outcome(item, threadId, "failed", "dispatch_failed"); + } + if (result.state !== "queued") { acknowledgeDelegation(threadId, item.id); } + recordOutcome(record, result); } })().finally(() => { drainingThreads.delete(threadId); // A later turn may have queued and settled while this thread was // waiting for approval. Its items were not in our snapshot, so start a // fresh drain instead of leaving them parked until another restart. - if (pendingDelegations.get(threadId)?.length) { - drainDelegations(bus, approvalBus, threadId, runTarget); + if (pendingDelegations.get(threadId)?.some((item) => item.phase === "awaiting_source")) { + drainDelegations(bus, approvalBus, threadId, runTarget, record); } }); } +function outcome( + item: PendingDelegationItem, + sourceThreadId: string, + state: DelegationOutcome["state"], + reason?: DelegationReason, +): DelegationOutcome { + return { + state, + taskId: item.id, + sourceThreadId, + toBotId: item.toBotId, + attempts: item.attempts, + ...(reason ? { reason } : {}), + }; +} + /** Remove one terminal handoff only after approval/dispatch has settled. */ function acknowledgeDelegation(threadId: string, itemId: string): void { const current = pendingDelegations.get(threadId); @@ -196,18 +357,31 @@ function acknowledgeDelegation(threadId: string, itemId: string): void { /** Drop a thread's queued handoffs without running them, telling the user * they were dropped. Used when the queueing turn failed or was interrupted. */ -export function discardDelegations(bus: CommsBus, threadId: string): void { +export function discardDelegations( + bus: CommsBus, + threadId: string, + record?: DelegationRecorder, + includeWaiting = false, +): void { const list = pendingDelegations.get(threadId); if (!list?.length) return; - pendingDelegations.delete(threadId); + const discarded = includeWaiting ? list : list.filter((item) => item.phase === "awaiting_source"); + if (!discarded.length) return; + const discardedIds = new Set(discarded.map((item) => item.id)); + const retained = list.filter((item) => !discardedIds.has(item.id)); + if (retained.length) pendingDelegations.set(threadId, retained); + else pendingDelegations.delete(threadId); savePending(); const from = bus.store.botByThread(threadId); if (!from) return; bus.store.appendMessage(threadId, { role: "bot", kind: "activity", - tool: { name: `${list.length} queued delegation${list.length > 1 ? "s" : ""} dropped — the turn did not finish`, ok: false }, + tool: { name: `${discarded.length} queued delegation${discarded.length > 1 ? "s" : ""} dropped — the turn did not finish`, ok: false }, }); + for (const item of discarded) { + recordOutcome(record, outcome(item, threadId, "failed", "source_turn_failed")); + } } async function processOne( @@ -215,7 +389,7 @@ async function processOne( approvalBus: ApprovalBus, from: BotRecord, sourceThreadId: string, - item: DelegationItem, + item: PendingDelegationItem, runTarget: ( toBotId: string, message: string, @@ -223,24 +397,19 @@ async function processOne( sourceThreadId: string, channel?: GroupRecord, ) => void | Promise, -): Promise { +): Promise { + if (isExpired(item)) { + reportFailure(bus, sourceThreadId, item.toBotId, "expired", "the queued delegation expired"); + return outcome(item, sourceThreadId, "failed", "expired"); + } let sender = from; let target = bus.store.bot(item.toBotId); if (!target) { - bus.store.appendMessage(sourceThreadId, { - role: "bot", - kind: "activity", - tool: { name: `error: delegation to ${item.toBotId} failed — no such bot`, ok: false }, - }); - return; + reportFailure(bus, sourceThreadId, item.toBotId, "target_missing", "no such bot"); + return outcome(item, sourceThreadId, "failed", "target_missing"); } if (target.busy) { - bus.store.appendMessage(sourceThreadId, { - role: "bot", - kind: "activity", - tool: { name: `Delegation to @${target.name} canceled — @${target.name} is busy`, ok: false }, - }); - return; + return deferForBusyTarget(bus, sourceThreadId, item, target.name); } if (sender.approvePeerComms) { const verdict = await requestPeerApproval( @@ -257,7 +426,7 @@ async function processOne( kind: "activity", tool: { name: `Delegation to @${target.name} denied by user`, ok: false }, }); - return; + return outcome(item, sourceThreadId, "failed", "user_denied"); } // The approval could have been sitting for up to 15 minutes. Everything // checked above is a stale snapshot now: re-read both bots and re-check @@ -265,23 +434,86 @@ async function processOne( // and mirror a "Messaged @X" chip for an exchange that never happens. const current = bus.store.bot(item.toBotId); const currentSender = bus.store.bot(from.id); - if (!current || !currentSender || !bus.store.taskByThread(currentSender.id, sourceThreadId)) return; + if (!current || !currentSender || !bus.store.taskByThread(currentSender.id, sourceThreadId)) { + reportFailure(bus, sourceThreadId, item.toBotId, "source_missing", "the source task no longer exists"); + return outcome(item, sourceThreadId, "failed", "source_missing"); + } if (current.busy) { - bus.store.appendMessage(sourceThreadId, { - role: "bot", - kind: "activity", - tool: { name: `Delegation to @${current.name} canceled — @${current.name} is busy`, ok: false }, - }); - return; + return deferForBusyTarget(bus, sourceThreadId, item, current.name); } sender = currentSender; target = current; } const channel = getOrCreateChannel(bus.store, sender, target); - mirrorExchange(bus, sender, target, item.message, channel, sourceThreadId); const reasonLine = item.reason ? `\n\n[Reason: ${item.reason}]` : ""; const prefixed = `[Delegated by @${sender.name}, another bot in this OpenMausBot workspace. Do the work and reply directly.]\n\n${item.message}${reasonLine}`; - await runTarget(item.toBotId, prefixed, item.depth + 1, sourceThreadId, channel); + try { + await runTarget(item.toBotId, prefixed, item.depth + 1, sourceThreadId, channel); + } catch (error) { + if (isBusyError(error)) return deferForBusyTarget(bus, sourceThreadId, item, target.name); + const why = error instanceof Error ? error.message : String(error); + // A permanent start failure is still part of the handoff record: show + // the attempted request beside its terminal failure. Busy races are the + // exception above because they remain eligible and must not look sent. + mirrorExchange(bus, sender, target, item.message, channel, sourceThreadId); + reportFailure(bus, sourceThreadId, target.name, "dispatch_failed", `could not start — ${why.slice(0, 120)}`); + return outcome(item, sourceThreadId, "failed", "dispatch_failed"); + } + // Do not mirror a handoff until the target has actually accepted the turn. + // A busy race must remain retryable without showing a false exchange. + mirrorExchange(bus, sender, target, item.message, channel, sourceThreadId); + return outcome(item, sourceThreadId, "completed"); +} + +function isBusyError(error: unknown): boolean { + const message = error instanceof Error ? error.message : String(error); + // 409 is also used for permanent configuration failures (for example a + // missing provider instance), so status alone must never make work retry. + return /already working|\bis busy\b/i.test(message); +} + +function deferForBusyTarget( + bus: CommsBus, + sourceThreadId: string, + item: PendingDelegationItem, + targetName: string, +): DelegationOutcome { + const attempts = item.attempts + 1; + if (attempts >= MAX_TARGET_BUSY_ATTEMPTS) { + reportFailure(bus, sourceThreadId, targetName, "retry_limit", "the target stayed busy across the retry limit"); + return { ...outcome(item, sourceThreadId, "failed", "retry_limit"), attempts }; + } + const firstWait = item.phase !== "waiting_target" || item.lastReason !== "target_busy"; + item.phase = "waiting_target"; + item.attempts = attempts; + item.lastReason = "target_busy"; + savePending(); + if (firstWait) { + bus.store.appendMessage(sourceThreadId, { + role: "bot", + kind: "activity", + tool: { name: `Delegation queued [target_busy] — @${targetName} will pick it up when the active turn completes` }, + }); + } + return outcome(item, sourceThreadId, "queued", "target_busy"); +} + +function reportFailure( + bus: CommsBus, + sourceThreadId: string, + targetName: string, + reason: DelegationReason, + detail: string, +): void { + try { + bus.store.appendMessage(sourceThreadId, { + role: "bot", + kind: "activity", + tool: { name: `error: delegation failed [${reason}] — @${targetName}: ${detail}`, ok: false }, + }); + } catch { + /* the source task may have been deleted; the structured recorder remains */ + } } /** Test helper: how many items remain queued for a thread. */ diff --git a/server/index.ts b/server/index.ts index e06b1c917..b52ed6e75 100644 --- a/server/index.ts +++ b/server/index.ts @@ -66,7 +66,16 @@ import { EFFORT_LEVELS, isEffortLevel, type ModelSelection, type RequestOutcome, import { BUILT_IN_DRIVERS } from "./drivers/builtIn.ts"; import { getOrCreateChannel, mirrorActivity, mirrorExchange, mirrorReply, type CommsBus } from "./comms-visibility.ts"; import { searchMessages } from "./message-db.ts"; -import { _loadPending, discardDelegations, drainDelegations, pendingThreads, queueDelegation, type QueueResult } from "./delegations.ts"; +import { + _loadPending, + discardDelegations, + drainDelegations, + drainReadyDelegations, + pendingThreads, + queueDelegation, + type DelegationRecorder, + type QueueFailureReason, +} from "./delegations.ts"; import { drainSteeredMessages, queueSteeredMessage } from "./steer-queue.ts"; import { EventBus } from "./harness/bus.ts"; import { ProviderRegistry } from "./harness/registry.ts"; @@ -1125,7 +1134,29 @@ bus.subscribe((event: RuntimeEvent) => { /** How a drained delegation becomes a real turn on the target. Shared by * the settle-time drain and the boot-time drain of what a previous process * left queued. */ -const runDelegatedTurn: Parameters[3] = (toBotId, text, commsDepth, sourceThreadId, channel) => { +const recordDelegationOutcome: DelegationRecorder = (outcome) => { + const source = store.botByThread(outcome.sourceThreadId); + const decision = outcome.state === "queued" + ? "delegation-queued" + : outcome.state === "completed" + ? "delegation-completed" + : "delegation-failed"; + appendDecision(DATA_DIR, { + threadId: outcome.sourceThreadId, + requestId: outcome.taskId, + botId: source?.id, + botName: source?.name, + tool: "delegate_bot", + summary: `target:${outcome.toBotId};attempts:${outcome.attempts}`, + decision, + source: "delegation", + rule: outcome.reason, + unattended: source ? isUnattended(source.id) : undefined, + }); +}; + +const runDelegatedTurn: Parameters[3] = (toBotId, text, commsDepth, sourceThreadId, channel) => + new Promise((resolve, reject) => { // startTurn REJECTS on an ordinary condition — busy target, deleted bot, // unavailable provider. Unhandled, that rejection is fatal to the // harness (Node's default), which in the packaged app kills the server @@ -1133,10 +1164,10 @@ const runDelegatedTurn: Parameters[3] = (toBotId, text, const targetThreadId = store.bot(toBotId)?.threadId; if (targetThreadId) delegationWatch.set(targetThreadId, { channelId: channel?.id, toBotId }); let failureReported = false; + let settled = false; const reportStartFailure = (error: unknown) => { - if (failureReported) return; + if (failureReported || settled) return; failureReported = true; - const bot = store.bot(toBotId); const why = error instanceof Error ? error.message : String(error); if (targetThreadId) { finalizeDelegationWatch( @@ -1146,33 +1177,43 @@ const runDelegatedTurn: Parameters[3] = (toBotId, text, `Delegated turn could not start — ${why.slice(0, 120)}`, ); } - const source = store.botByThread(sourceThreadId); - if (!source) return; - store.appendMessage(sourceThreadId, { - role: "bot", - kind: "activity", - tool: { name: `error: delegation to @${bot?.name ?? toBotId} could not start — ${why.slice(0, 120)}`, ok: false }, - }); + if (!settled) { + settled = true; + const failure = error instanceof Error ? error : new Error(why); + reject(failure); + } + }; + const reportStarted = () => { + if (settled) return; + settled = true; + resolve(); }; - return startTurn(toBotId, text, { + void startTurn(toBotId, text, { commsDepth, unattended: isUnattended(store.botByThread(sourceThreadId)?.id), // startTurn schedules provider/integration setup after marking the bot // busy. Those asynchronous setup failures do not emit turn.completed, // so clear the watch and report them through this callback too. - onDispatchError: reportStartFailure, - }).catch((err) => { - reportStartFailure(err); - }); -}; + onDispatchError: (message) => reportStartFailure(new Error(message)), + onDispatched: reportStarted, + }).catch(reportStartFailure); + }); bus.subscribe((event: RuntimeEvent) => { if (event.type !== "turn.completed") return; // A turn that failed or was interrupted drops its queue rather than // firing it later: the user who hit Stop does not expect the delegations // that turn queued to run anyway, minutes later, on an unrelated turn. - if (!event.ok) return void discardDelegations(commsBus, event.threadId); - drainDelegations(commsBus, approvalBus, event.threadId, runDelegatedTurn); + if (!event.ok) return void discardDelegations(commsBus, event.threadId, recordDelegationOutcome); + drainDelegations(commsBus, approvalBus, event.threadId, runDelegatedTurn, recordDelegationOutcome); +}); + +// A busy target is a temporary state, not a terminal delegation failure. +// The main runtime fold runs before this subscriber and has already changed +// the completed bot to idle, so one event can drain each eligible wait once. +bus.subscribe((event: RuntimeEvent) => { + if (event.type !== "turn.completed") return; + drainReadyDelegations(commsBus, approvalBus, runDelegatedTurn, recordDelegationOutcome); }); // ── steer-queue drain: messages sent while the bot was busy ──────────── @@ -1340,6 +1381,7 @@ async function startTurn( * masquerading as another message authored by the user. */ connectorContinuation?: boolean; onDispatchError?: (message: string) => void; + onDispatched?: () => void; }, ) { const bot = store.bot(botId); @@ -1715,6 +1757,7 @@ async function startTurn( if (rewound) store.patchBot(bot.id, { rewound: false, resumeCursors: {} }); // and this engine now owns the thread's most recent turn store.markTaskDispatched(bot.id, threadId, instanceId); + opts?.onDispatched?.(); // a turn can settle before dispatch returns, and a poller started // after its own turn.completed would never be torn down — it would // keep polling the box forever, carrying dead per-turn state. busy @@ -1861,7 +1904,9 @@ _loadPending(); { const leftover = pendingThreads(); if (leftover.length) console.log(`delegations: ${leftover.length} thread(s) with queued handoffs from a previous run — draining`); - for (const threadId of leftover) drainDelegations(commsBus, approvalBus, threadId, runDelegatedTurn); + for (const threadId of leftover) { + drainDelegations(commsBus, approvalBus, threadId, runDelegatedTurn, recordDelegationOutcome); + } } async function runGroupMemberTurn( @@ -2594,24 +2639,30 @@ const server = createServer(async (req, res) => { { toBotId, message, reason, depth }, MAX_COMMS_DEPTH, fromThreadId, + recordDelegationOutcome, ); - if (result !== "ok") { + if (result.state === "failed") { // the agent reads this string — a bare enum ("too_deep") tells it // nothing about what to do instead - const said: Record, string> = { + const said: Record = { self: "a bot cannot delegate to itself", too_deep: "delegation chains are limited to one hop — do this one yourself", no_target: "no such bot", too_many: "too many delegations queued on this turn — finish some first", }; - return json(res, 200, { error: said[result] }); + return json(res, 200, { state: "failed", error: said[result.reason], reason: result.reason }); } const targetName = store.bot(toBotId)?.name ?? toBotId; return json(res, 200, { + state: "queued", queued: true, - message: from.approvePeerComms - ? `Queued for review — @${targetName} will only pick it up if the user approves after your turn finishes.` - : `Delegation queued — @${targetName} will pick it up after your current turn finishes.`, + taskId: result.taskId, + duplicate: result.duplicate, + message: result.duplicate + ? `Delegation already queued — @${targetName} will receive the existing task once eligible.` + : from.approvePeerComms + ? `Queued for review — @${targetName} will only pick it up if the user approves after your turn finishes.` + : `Delegation queued — @${targetName} will pick it up after your current turn finishes.`, }); } if (method === "POST" && path === "/api/internal/connectors/mcp") { @@ -3633,7 +3684,7 @@ const server = createServer(async (req, res) => { // a peer approval naming this bot can never be meaningfully answered // now, and its caller would otherwise wait out the 15-minute timeout cancelPeerApprovalsFor(bot.id); - discardDelegations(commsBus, bot.threadId); + discardDelegations(commsBus, bot.threadId, recordDelegationOutcome, true); computerControl.forget(bot.id); const target = perBotLocalVmTarget(bot.id); localVmIdles.get(target.key)?.cancel();