diff --git a/server/delegations.test.ts b/server/delegations.test.ts index 44a2808c5..2558fc260 100644 --- a/server/delegations.test.ts +++ b/server/delegations.test.ts @@ -12,8 +12,13 @@ import { DATA_DIR } from "./config.ts"; import type { ModelSelection } from "./contracts.ts"; import { drainDelegations, + findDelegationReceipt, + MAX_BUSY_ATTEMPTS, + pendingDelegationInfo, pendingDelegationSnapshot, queueDelegation, + recordDelegationReceipt, + threadsWaitingOn, _pendingCount, } from "./delegations.ts"; import { peerAllowKey, resolvePeerComms } from "./peer-approval.ts"; @@ -81,7 +86,7 @@ describe("queueDelegation", () => { message: "self-talk", depth: 0, }, 1); - expect(result).toBe("self"); + expect(result.result).toBe("self"); expect(_pendingCount(from.threadId)).toBe(0); }); @@ -91,7 +96,7 @@ describe("queueDelegation", () => { message: "next task", depth: 1, }, 1); - expect(result).toBe("too_deep"); + expect(result.result).toBe("too_deep"); expect(_pendingCount(from.threadId)).toBe(0); }); @@ -101,7 +106,7 @@ describe("queueDelegation", () => { message: "where?", depth: 0, }, 1); - expect(result).toBe("no_target"); + expect(result.result).toBe("no_target"); expect(_pendingCount(from.threadId)).toBe(0); }); @@ -112,7 +117,7 @@ describe("queueDelegation", () => { reason: "followup", depth: 0, }, 1); - expect(result).toBe("ok"); + expect(result.result).toBe("ok"); expect(_pendingCount(from.threadId)).toBe(1); const chip = store @@ -156,7 +161,7 @@ describe("queueDelegation", () => { routineTask.threadId, ); - expect(result).toBe("ok"); + expect(result.result).toBe("ok"); expect(_pendingCount(routineTask.threadId)).toBe(1); expect(_pendingCount(from.threadId)).toBe(0); expect( @@ -322,7 +327,7 @@ describe("drainDelegations", () => { expect(runTargetCalls).toEqual([]); }); - it("skips runTarget and emits a 'is busy' chip when the target is currently busy", async () => { + it("keeps the handoff queued with a 'waiting' chip when the target is currently busy", async () => { store.patchBot(target.id, { busy: true }); queueDelegation(commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1); drainDelegations(commsBus, approvalBus, from.threadId, (toBotId, message, commsDepth) => { @@ -331,11 +336,12 @@ describe("drainDelegations", () => { 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("waiting — they're busy")), ); - expect(chip.tool?.name).toBe("Delegation to @Helper canceled — @Helper is busy"); - expect(chip.tool?.ok).toBe(false); + expect(chip.tool?.name).toBe("Delegation to @Helper waiting — they're busy (retry 1/3 when they finish)"); expect(runTargetCalls).toEqual([]); + // retained for the retry drain the target's settling turn triggers + expect(_pendingCount(from.threadId)).toBe(1); }); it("asks for approval when approvePeerComms is on, then runs only on allow", async () => { @@ -440,7 +446,7 @@ 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({ result: "ok" }); expect(existsSync(file())).toBe(true); const onDisk = JSON.parse(readFileSync(file(), "utf8")) as Record; expect(onDisk[from.threadId]).toHaveLength(1); @@ -529,3 +535,104 @@ describe("delegations survive a restart", () => { expect(pendingThreads()).toEqual([]); }); }); + +describe("busy retries and receipts", () => { + let store: Store; + let from: BotRecord; + let target: BotRecord; + let commsBus: CommsBus; + let approvalBus: { store: Store; broadcast: (payload: unknown) => void }; + + beforeEach(() => { + rmSync(DATA_DIR, { recursive: true, force: true }); + store = new Store(selection); + from = store.createBot(); + target = store.createBot(); + store.patchBot(target.id, { name: "Helper" }); + const buses = setupBuses(store); + commsBus = buses.commsBus; + approvalBus = buses.approvalBus; + }); + + const chipCount = (needle: string) => + store.messagesFor(from.threadId).filter((m) => m.kind === "activity" && m.tool?.name?.includes(needle)).length; + + it("keeps a handoff queued while the target is busy and dispatches on the retry drain", async () => { + store.patchBot(target.id, { busy: true }); + const queued = queueDelegation(commsBus, from, { toBotId: target.id, message: "later", depth: 0 }, 1); + expect(queued.result).toBe("ok"); + const taskId = queued.id!; + + const dispatched: unknown[][] = []; + const runTarget = (...args: unknown[]) => void dispatched.push(args); + + drainDelegations(commsBus, approvalBus, from.threadId, runTarget); + await waitFor(() => chipCount("waiting — they're busy (retry 1/") === 1); + expect(dispatched).toHaveLength(0); + expect(_pendingCount(from.threadId)).toBe(1); + // this is the set a settling target turn re-drains + expect(threadsWaitingOn(target.id)).toEqual([from.threadId]); + expect(pendingDelegationInfo(taskId)).toMatchObject({ toBotId: target.id, attempts: 1 }); + + store.patchBot(target.id, { busy: false }); + drainDelegations(commsBus, approvalBus, from.threadId, runTarget); + await waitFor(() => dispatched.length === 1); + expect(_pendingCount(from.threadId)).toBe(0); + // the task id rides into the dispatched turn so the receipt can be keyed + expect(dispatched[0][5]).toBe(taskId); + expect(pendingDelegationInfo(taskId)).toBeNull(); + }); + + it("gives up after the bounded retries, with a receipt the delegator can read", async () => { + store.patchBot(target.id, { busy: true }); + const queued = queueDelegation(commsBus, from, { toBotId: target.id, message: "later", depth: 0 }, 1); + const taskId = queued.id!; + const runTarget = () => undefined; + for (let round = 1; round < MAX_BUSY_ATTEMPTS; round++) { + drainDelegations(commsBus, approvalBus, from.threadId, runTarget); + await waitFor(() => chipCount(`retry ${round}/`) === 1); + } + drainDelegations(commsBus, approvalBus, from.threadId, runTarget); + await waitFor(() => _pendingCount(from.threadId) === 0); + expect(chipCount("canceled — still busy after")).toBe(1); + expect(findDelegationReceipt(taskId)).toMatchObject({ + status: "busy_gave_up", + toBotName: "Helper", + sourceThreadId: from.threadId, + }); + }); + + it("persists receipts across a restart and prunes the drawer by count", () => { + recordDelegationReceipt({ + id: "task-one", + sourceThreadId: from.threadId, + toBotId: target.id, + toBotName: "Helper", + status: "done", + result: "the reply text", + }); + // a fresh process loads what the last one recorded + _loadPending(); + expect(findDelegationReceipt("task-one")).toMatchObject({ status: "done", result: "the reply text" }); + + for (let index = 0; index < 105; index++) { + recordDelegationReceipt({ + id: `bulk-${index}`, + sourceThreadId: from.threadId, + toBotId: target.id, + toBotName: "Helper", + status: "done", + }); + } + expect(findDelegationReceipt("bulk-104")).toBeTruthy(); + expect(findDelegationReceipt("bulk-3")).toBeNull(); // oldest pruned + }); + + it("writes a dropped receipt for every handoff a failed turn discards", async () => { + const queued = queueDelegation(commsBus, from, { toBotId: target.id, message: "never runs", depth: 0 }, 1); + const { discardDelegations } = await import("./delegations.ts"); + discardDelegations(commsBus, from.threadId); + expect(_pendingCount(from.threadId)).toBe(0); + expect(findDelegationReceipt(queued.id!)).toMatchObject({ status: "dropped" }); + }); +}); diff --git a/server/delegations.ts b/server/delegations.ts index 2308a4725..7fe696545 100644 --- a/server/delegations.ts +++ b/server/delegations.ts @@ -33,12 +33,40 @@ export interface DelegationItem { } interface PendingDelegationItem extends DelegationItem { - /** Stable acknowledgement key for crash-safe removal from the queue. */ + /** Stable acknowledgement key for crash-safe removal from the queue — + * and the task id the delegating bot uses with check/wait_delegation. */ id: string; + /** Busy-target retries so far. The item stays queued (not canceled) while + * the target is busy, and is retried when any of the target's turns + * settles — up to MAX_BUSY_ATTEMPTS. */ + attempts: number; +} + +export type DelegationOutcome = "done" | "failed" | "denied" | "busy_gave_up" | "dropped" | "error"; + +/** The durable terminal record of one handoff: what the delegating bot reads + * back with check_delegation / wait_delegation. Bounded and pruned — this is + * a receipt drawer, not a transcript. */ +export interface DelegationReceipt { + id: string; + sourceThreadId: string; + toBotId: string; + toBotName: string; + status: DelegationOutcome; + /** the peer's reply on success; the failure name otherwise (bounded) */ + result?: string; + finishedAt: number; } export type QueueResult = "ok" | "no_target" | "self" | "too_deep" | "too_many"; +/** What queueDelegation hands back: the verdict, and on success the task id + * the delegating bot can later read back with check/wait_delegation. */ +export interface QueuedDelegation { + result: QueueResult; + id?: string; +} + /** Per source-thread queue. Persisted to delegations.json on every change * and reloaded at boot: a handoff queued right before a restart runs after * it. (Provider PERMISSIONS still die with the process — nobody can answer @@ -46,7 +74,68 @@ export type QueueResult = "ok" | "no_target" | "self" | "too_deep" | "too_many"; * and approvePeerComms are re-checked at drain time as always.) */ const pendingDelegations = new Map(); const drainingThreads = new Set(); +/** Threads whose drain was requested WHILE a drain was already running. + * Dropping such a request loses real work: the waiting-on retry fires the + * moment a busy target settles, and that can land mid-drain. */ +const queuedRedrains = new Set(); const DELEGATIONS_FILE = join(DATA_DIR, "delegations.json"); +const RECEIPTS_FILE = join(DATA_DIR, "delegation-receipts.json"); +const MAX_RECEIPTS = 100; +const RECEIPT_MAX_AGE_MS = 48 * 60 * 60 * 1000; +const RESULT_MAX_CHARS = 4_000; +export const MAX_BUSY_ATTEMPTS = 3; + +let receipts: DelegationReceipt[] = []; + +function saveReceipts(): void { + try { + writeFileAtomic(RECEIPTS_FILE, JSON.stringify(receipts, null, 2), { mode: 0o600 }); + } catch (error) { + console.error("delegations: could not persist receipts", error); + } +} + +/** Record one terminal outcome. Newest first; pruned by count and age so the + * drawer can never grow without bound. */ +export function recordDelegationReceipt(receipt: Omit & { finishedAt?: number }): void { + const now = Date.now(); + const bounded: DelegationReceipt = { + id: receipt.id, + sourceThreadId: receipt.sourceThreadId, + toBotId: receipt.toBotId, + toBotName: receipt.toBotName, + status: receipt.status, + finishedAt: receipt.finishedAt ?? now, + }; + if (receipt.result !== undefined) bounded.result = receipt.result.slice(0, RESULT_MAX_CHARS); + receipts = [bounded, ...receipts.filter((existing) => existing.id !== bounded.id)] + .filter((existing) => now - existing.finishedAt <= RECEIPT_MAX_AGE_MS) + .slice(0, MAX_RECEIPTS); + saveReceipts(); +} + +export function findDelegationReceipt(id: string): DelegationReceipt | null { + return receipts.find((receipt) => receipt.id === id) ?? null; +} + +/** A still-queued task's routing info, or null once it dispatched/settled. */ +export function pendingDelegationInfo(id: string): { sourceThreadId: string; toBotId: string; attempts: number } | null { + for (const [sourceThreadId, items] of pendingDelegations) { + const item = items.find((candidate) => candidate.id === id); + if (item) return { sourceThreadId, toBotId: item.toBotId, attempts: item.attempts }; + } + return null; +} + +/** Source threads holding a handoff that already waited on this busy bot at + * least once — the set a target's settling turn re-drains. Fresh items + * (attempts 0) are excluded: they run when their SOURCE turn settles, and + * draining them early would start the peer before the delegator finished. */ +export function threadsWaitingOn(toBotId: string): string[] { + return [...pendingDelegations.entries()] + .filter(([, items]) => items.some((item) => item.toBotId === toBotId && item.attempts > 0)) + .map(([threadId]) => threadId); +} function savePending(): void { try { @@ -77,6 +166,7 @@ export function _loadPending(): void { message: item.message, ...(typeof item.reason === "string" ? { reason: item.reason } : {}), depth: Math.max(0, Math.trunc(item.depth!)), + attempts: Number.isFinite(item.attempts) ? Math.max(0, Math.trunc(item.attempts!)) : 0, }]; }); if (items.length) pendingDelegations.set(threadId, items); @@ -84,6 +174,32 @@ export function _loadPending(): void { } catch { /* fresh install, or unreadable — start empty */ } + receipts = []; + try { + const rawReceipts = JSON.parse(readFileSync(RECEIPTS_FILE, "utf8")); + if (Array.isArray(rawReceipts)) { + const now = Date.now(); + const loaded: DelegationReceipt[] = []; + for (const value of rawReceipts) { + if (!value || typeof value !== "object" || Array.isArray(value)) continue; + // SAFETY: the Partial view only names candidate fields; every one is + // narrowed below before a receipt is constructed from the narrowed + // locals, so nothing unvalidated survives into `receipts`. + const candidate = value as Partial; + const { id, sourceThreadId, toBotId, toBotName, status, result, finishedAt } = candidate; + if (typeof id !== "string" || !id) continue; + if (typeof sourceThreadId !== "string" || typeof toBotId !== "string") continue; + if (typeof toBotName !== "string" || typeof status !== "string") continue; + if (!Number.isFinite(finishedAt) || now - finishedAt! > RECEIPT_MAX_AGE_MS) continue; + const receipt: DelegationReceipt = { id, sourceThreadId, toBotId, toBotName, status, finishedAt: finishedAt! }; + if (typeof result === "string") receipt.result = result; + loaded.push(receipt); + } + receipts = loaded.slice(0, MAX_RECEIPTS); + } + } catch { + /* no receipts yet */ + } } /** Source threads with something queued — what a boot drain iterates. */ @@ -119,17 +235,18 @@ export function queueDelegation( item: DelegationItem, maxDepth: number, sourceThreadId = from.threadId, -): QueueResult { - if (item.toBotId === from.id) return "self"; - if (item.depth >= maxDepth) return "too_deep"; +): QueuedDelegation { + if (item.toBotId === from.id) return { result: "self" }; + if (item.depth >= maxDepth) return { result: "too_deep" }; const target = bus.store.bot(item.toBotId); - if (!target) return "no_target"; + if (!target) return { result: "no_target" }; const list = pendingDelegations.get(sourceThreadId) ?? []; // 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 { result: "too_many" }; + const id = newId(); + list.push({ ...item, id, attempts: 0 }); pendingDelegations.set(sourceThreadId, list); savePending(); const label = `Delegated to @${target.name}${item.reason ? `: ${item.reason}` : ""}`; @@ -138,7 +255,7 @@ export function queueDelegation( kind: "activity", tool: { name: label }, }); - return "ok"; + return { result: "ok", id }; } /** Drain queued delegations for a source thread (called on its @@ -156,10 +273,14 @@ export function drainDelegations( message: string, commsDepth: number, sourceThreadId: string, - channel?: GroupRecord, + channel: GroupRecord | undefined, + taskId: string, ) => void | Promise, ): void { - if (drainingThreads.has(threadId)) return; + if (drainingThreads.has(threadId)) { + queuedRedrains.add(threadId); + return; + } const list = pendingDelegations.get(threadId); if (!list?.length) return; const from = bus.store.botByThread(threadId); @@ -172,10 +293,19 @@ export function drainDelegations( drainingThreads.add(threadId); void (async () => { for (const item of snapshot) { + let outcome: "settled" | "requeued" = "settled"; try { - await processOne(bus, approvalBus, from, threadId, item, runTarget); + outcome = await processOne(bus, approvalBus, from, threadId, item, runTarget); } catch (error) { const why = error instanceof Error ? error.message : String(error); + recordDelegationReceipt({ + id: item.id, + sourceThreadId: threadId, + toBotId: item.toBotId, + toBotName: bus.store.bot(item.toBotId)?.name ?? item.toBotId, + status: "error", + result: why.slice(0, 200), + }); try { bus.store.appendMessage(threadId, { role: "bot", @@ -186,15 +316,21 @@ export function drainDelegations( console.error("delegation failed and could not be reported", reportError); } } finally { - acknowledgeDelegation(threadId, item.id); + // A requeued item (busy target, retries left) stays for the drain + // that the target's own settling turn will trigger. + if (outcome !== "requeued") acknowledgeDelegation(threadId, item.id); } } })().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) { + // waiting for approval. Only items OUTSIDE our snapshot warrant a fresh + // drain — re-draining a just-requeued item would burn its bounded busy + // retries in milliseconds instead of once per target settle. + const redrainRequested = queuedRedrains.delete(threadId); + const snapshotIds = new Set(snapshot.map((item) => item.id)); + const hasNewItems = pendingDelegations.get(threadId)?.some((item) => !snapshotIds.has(item.id)) ?? false; + if (redrainRequested || hasNewItems) { drainDelegations(bus, approvalBus, threadId, runTarget); } }); @@ -217,6 +353,16 @@ export function discardDelegations(bus: CommsBus, threadId: string): void { if (!list?.length) return; pendingDelegations.delete(threadId); savePending(); + for (const item of list) { + recordDelegationReceipt({ + id: item.id, + sourceThreadId: threadId, + toBotId: item.toBotId, + toBotName: bus.store.bot(item.toBotId)?.name ?? item.toBotId, + status: "dropped", + result: "the delegating turn did not finish", + }); + } const from = bus.store.botByThread(threadId); if (!from) return; bus.store.appendMessage(threadId, { @@ -231,32 +377,59 @@ async function processOne( approvalBus: ApprovalBus, from: BotRecord, sourceThreadId: string, - item: DelegationItem, + item: PendingDelegationItem, runTarget: ( toBotId: string, message: string, commsDepth: number, sourceThreadId: string, - channel?: GroupRecord, + channel: GroupRecord | undefined, + taskId: string, ) => void | Promise, -): Promise { +): Promise<"settled" | "requeued"> { let sender = from; let target = bus.store.bot(item.toBotId); if (!target) { + recordDelegationReceipt({ + id: item.id, + sourceThreadId, + toBotId: item.toBotId, + toBotName: item.toBotId, + status: "error", + result: "no such bot", + }); bus.store.appendMessage(sourceThreadId, { role: "bot", kind: "activity", tool: { name: `error: delegation to ${item.toBotId} failed — no such bot`, ok: false }, }); - return; + return "settled"; } if (target.busy) { + item.attempts += 1; + if (item.attempts < MAX_BUSY_ATTEMPTS) { + savePending(); + bus.store.appendMessage(sourceThreadId, { + role: "bot", + kind: "activity", + tool: { name: `Delegation to @${target.name} waiting — they're busy (retry ${item.attempts}/${MAX_BUSY_ATTEMPTS} when they finish)` }, + }); + return "requeued"; + } + recordDelegationReceipt({ + id: item.id, + sourceThreadId, + toBotId: target.id, + toBotName: target.name, + status: "busy_gave_up", + result: `@${target.name} stayed busy through ${MAX_BUSY_ATTEMPTS} retries`, + }); bus.store.appendMessage(sourceThreadId, { role: "bot", kind: "activity", - tool: { name: `Delegation to @${target.name} canceled — @${target.name} is busy`, ok: false }, + tool: { name: `Delegation to @${target.name} canceled — still busy after ${MAX_BUSY_ATTEMPTS} retries`, ok: false }, }); - return; + return "settled"; } if (sender.approvePeerComms) { const verdict = await requestPeerApproval( @@ -268,12 +441,20 @@ async function processOne( sourceThreadId, ); if (verdict !== "allow") { + recordDelegationReceipt({ + id: item.id, + sourceThreadId, + toBotId: target.id, + toBotName: target.name, + status: "denied", + result: "the user denied this handoff", + }); bus.store.appendMessage(sourceThreadId, { role: "bot", kind: "activity", tool: { name: `Delegation to @${target.name} denied by user`, ok: false }, }); - return; + return "settled"; } // 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 @@ -281,14 +462,32 @@ 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)) return "settled"; if (current.busy) { + item.attempts += 1; + if (item.attempts < MAX_BUSY_ATTEMPTS) { + savePending(); + bus.store.appendMessage(sourceThreadId, { + role: "bot", + kind: "activity", + tool: { name: `Delegation to @${current.name} waiting — they're busy (retry ${item.attempts}/${MAX_BUSY_ATTEMPTS} when they finish)` }, + }); + return "requeued"; + } + recordDelegationReceipt({ + id: item.id, + sourceThreadId, + toBotId: current.id, + toBotName: current.name, + status: "busy_gave_up", + result: `@${current.name} stayed busy through ${MAX_BUSY_ATTEMPTS} retries`, + }); bus.store.appendMessage(sourceThreadId, { role: "bot", kind: "activity", - tool: { name: `Delegation to @${current.name} canceled — @${current.name} is busy`, ok: false }, + tool: { name: `Delegation to @${current.name} canceled — still busy after ${MAX_BUSY_ATTEMPTS} retries`, ok: false }, }); - return; + return "settled"; } sender = currentSender; target = current; @@ -297,7 +496,8 @@ async function processOne( 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); + await runTarget(item.toBotId, prefixed, item.depth + 1, sourceThreadId, channel, item.id); + return "settled"; } /** Test helper: how many items remain queued for a thread. */ @@ -309,4 +509,6 @@ export function _pendingCount(threadId: string): number { export function _resetPending(): void { pendingDelegations.clear(); drainingThreads.clear(); + queuedRedrains.clear(); + receipts = []; } diff --git a/server/drivers/agents-proxy.test.ts b/server/drivers/agents-proxy.test.ts index 20b97db2d..53239b57c 100644 --- a/server/drivers/agents-proxy.test.ts +++ b/server/drivers/agents-proxy.test.ts @@ -19,6 +19,8 @@ let lastAuth: string | undefined; let lastAskBody: any = null; let askResponse: unknown = { botName: "Helper", text: "hi from helper" }; let lastDelegateBody: any = null; +let lastDelegationUrl: string | null = null; +let delegationStatusResponse: unknown = { status: "done", toBotName: "Helper", result: "All done." }; let delegateResponse: unknown = { queued: true, message: "Delegation queued." }; let lastCreateBody: any = null; let lastCredentialBody: any = null; @@ -89,6 +91,12 @@ beforeAll(async () => { }); return; } + if (req.method === "GET" && req.url?.startsWith("/api/internal/delegations/")) { + lastDelegationUrl = req.url; + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify(delegationStatusResponse)); + return; + } if (req.method === "POST" && req.url === "/api/internal/create-bot") { let data = ""; req.on("data", (c) => (data += c)); @@ -170,6 +178,8 @@ describe("agents-proxy MCP surface", () => { "list_bots", "ask_bot", "delegate_bot", + "check_delegation", + "wait_delegation", "create_bot", "request_credential", "list_routines", @@ -303,6 +313,46 @@ describe("agents-proxy MCP surface", () => { expect(lastCredentialBody).toBeNull(); }); + it("hands the delegator its task id and the tools to read the outcome", async () => { + delegateResponse = { + queued: true, + taskId: "task-abc123", + message: "Delegation queued — @Helper will pick it up after your current turn finishes.", + }; + const res = await callTool("delegate_bot", { bot_id: "bot-helper", message: "do the thing" }); + expect(res.result.content[0].text).toContain("Task id: task-abc123"); + expect(res.result.content[0].text).toContain("wait_delegation"); + delegateResponse = { queued: true, message: "Delegation queued." }; + }); + + it("check/wait_delegation: flat schemas, guided errors, and the read-back wire", async () => { + const list = await rpc("tools/list"); + for (const name of ["check_delegation", "wait_delegation"]) { + const tool = list.result.tools.find((t: { name: string }) => t.name === name); + expect(JSON.stringify(tool.inputSchema)).not.toMatch(/"(oneOf|anyOf|allOf|const|format)":/); + } + + lastDelegationUrl = null; + const bad = await callTool("check_delegation", { task_id: "!" }); + expect(bad.result.isError).toBe(true); + expect(bad.result.content[0].text).toContain('"task_id"'); + expect(lastDelegationUrl).toBeNull(); // guidance is free + + const done = await callTool("check_delegation", { task_id: "task-abc123" }); + expect(done.result.content[0].text).toContain("@Helper finished task task-abc123"); + expect(done.result.content[0].text).toContain("All done."); + expect(lastDelegationUrl).toContain("/api/internal/delegations/task-abc123?"); + expect(lastDelegationUrl).toContain("wait_ms=0"); + expect(lastDelegationUrl).toContain("fromBotId=bot-asker"); + + delegationStatusResponse = { status: "queued", toBotName: "Helper" }; + const waiting = await callTool("wait_delegation", { task_id: "task-abc123", timeout_seconds: 45 }); + expect(waiting.result.content[0].text).toContain("still queued"); + expect(waiting.result.content[0].text).toContain("after 45s"); + expect(lastDelegationUrl).toContain("wait_ms=45000"); + delegationStatusResponse = { status: "done", toBotName: "Helper", result: "All done." }; + }); + it("lists only the current bot's routines with authoritative time context", async () => { routinesResponse = { now: "2026-08-28T10:30:00.000Z", diff --git a/server/drivers/agents-proxy.ts b/server/drivers/agents-proxy.ts index 80c5607d4..dd18a6ac8 100644 --- a/server/drivers/agents-proxy.ts +++ b/server/drivers/agents-proxy.ts @@ -215,6 +215,31 @@ const TOOLS = [ required: ["bot_id", "message"], }, }, + { + name: "check_delegation", + description: + "Check what happened to a delegation you queued with delegate_bot, without waiting: still queued, running, or finished — and the peer's reply once it is done.", + inputSchema: { + type: "object", + properties: { + task_id: { type: "string", description: "The task id delegate_bot returned." }, + }, + required: ["task_id"], + }, + }, + { + name: "wait_delegation", + description: + "Wait until a delegation you queued finishes and return the peer's reply — ONE call instead of repeated checks. Use it after your own remaining work is done; it returns immediately if the task already finished.", + inputSchema: { + type: "object", + properties: { + task_id: { type: "string", description: "The task id delegate_bot returned." }, + timeout_seconds: { type: "integer", description: "give up waiting after this many seconds; default 60, max 240" }, + }, + required: ["task_id"], + }, + }, { name: "create_bot", description: @@ -382,8 +407,32 @@ async function callTool(name: string, args: Json): Promise<{ text: string; isErr const r = await api(`/api/internal/delegate-bot`, { method: "POST", body: JSON.stringify(body) }); if (r.error) return { text: `Couldn't queue the delegation: ${r.error}`, isError: true }; // Fire-and-forget by contract: the harness returns immediately, the - // peer turn runs after our current turn finishes. - return { text: typeof r.message === "string" ? r.message : "Delegation queued." }; + // peer turn runs after our current turn finishes. The task id is the + // bot's claim ticket for the outcome. + const note = typeof r.message === "string" ? r.message : "Delegation queued."; + const suffix = typeof r.taskId === "string" && r.taskId + ? ` Task id: ${r.taskId} — after your own work is done, read the outcome with check_delegation or block on it with wait_delegation.` + : ""; + return { text: `${note}${suffix}` }; + } + if (name === "check_delegation" || name === "wait_delegation") { + const taskId = String(args.task_id ?? "").trim(); + if (!/^[\w-]{4,64}$/.test(taskId)) { + return { text: `${name} needs the "task_id" that delegate_bot returned, e.g. {"task_id":"1f0c2f4e-..."}.`, isError: true }; + } + const timeout = Math.min(Math.max(Math.trunc(Number(args.timeout_seconds) || 60), 1), 240); + const waitMs = name === "wait_delegation" ? timeout * 1000 : 0; + const query = new URLSearchParams({ fromBotId: BOT_ID, fromThreadId: THREAD_ID, wait_ms: String(waitMs) }); + const r = await api(`/api/internal/delegations/${encodeURIComponent(taskId)}?${query.toString()}`); + const who = typeof r.toBotName === "string" && r.toBotName ? `@${r.toBotName}` : "the peer"; + if (r.status === "done") return { text: `${who} finished task ${taskId}:\n${String(r.result || "(no reply text)")}` }; + if (r.status === "queued") { + return { text: `Task ${taskId} is still queued — ${who} hasn't picked it up yet${waitMs ? ` after ${timeout}s` : ""}. Keep working and check again later.` }; + } + if (r.status === "running") { + return { text: `Task ${taskId} is running with ${who}${waitMs ? ` (still going after ${timeout}s)` : ""}. Check again shortly.` }; + } + return { text: `Task ${taskId} ended without a reply — ${String(r.status ?? "unknown")}${r.result ? `: ${String(r.result)}` : ""}.`, isError: true }; } if (name === "create_bot") { const botName = String(args.name ?? "").trim(); diff --git a/server/index.ts b/server/index.ts index 19af39cfd..16a7b9346 100644 --- a/server/index.ts +++ b/server/index.ts @@ -86,7 +86,7 @@ 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 { promptWithReply, transcriptText } from "./replies.ts"; -import { _loadPending, discardDelegations, drainDelegations, pendingDelegationSnapshot, pendingThreads, queueDelegation, type QueueResult } from "./delegations.ts"; +import { _loadPending, discardDelegations, drainDelegations, findDelegationReceipt, pendingDelegationInfo, pendingDelegationSnapshot, pendingThreads, queueDelegation, recordDelegationReceipt, threadsWaitingOn, type QueueResult } from "./delegations.ts"; import { cancelSteeredMessage, drainSteeredMessages, @@ -1427,7 +1427,7 @@ bus.subscribe((event: RuntimeEvent) => { // (target threadId → channel) lets the main fold mirror the delegated // turn's TERMINAL state into the A⇄B channel when it completes — the // channel stays the full record of the handoff, not just its request. -const delegationWatch = new Map(); +const delegationWatch = new Map(); /** Consume one delegated-turn watch and mirror exactly one terminal state. * Some harness paths settle a busy bot without a provider turn.completed @@ -1441,6 +1441,19 @@ function finalizeDelegationWatch( const watched = delegationWatch.get(threadId); if (!watched) return false; delegationWatch.delete(threadId); + // The receipt is written before any mirror short-circuits: the delegating + // bot's check/wait_delegation must see a terminal state even when the + // channel or target is gone. + if (watched.taskId && watched.sourceThreadId) { + recordDelegationReceipt({ + id: watched.taskId, + sourceThreadId: watched.sourceThreadId, + toBotId: watched.toBotId, + toBotName: store.bot(watched.toBotId)?.name ?? watched.toBotId, + status: ok ? "done" : "failed", + result: ok ? reply : failureName, + }); + } const target = store.bot(watched.toBotId); const channel = watched.channelId ? store.group(watched.channelId) : undefined; if (!target || !channel) return true; @@ -1484,13 +1497,13 @@ 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 runDelegatedTurn: Parameters[3] = (toBotId, text, commsDepth, sourceThreadId, channel, taskId) => { // 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 // child. Every delegation failure has to land as a chip instead. const targetThreadId = store.bot(toBotId)?.threadId; - if (targetThreadId) delegationWatch.set(targetThreadId, { channelId: channel?.id, toBotId }); + if (targetThreadId) delegationWatch.set(targetThreadId, { channelId: channel?.id, toBotId, taskId, sourceThreadId }); let failureReported = false; const reportStartFailure = (error: unknown) => { if (failureReported) return; @@ -1530,8 +1543,17 @@ bus.subscribe((event: RuntimeEvent) => { // 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) discardDelegations(commsBus, event.threadId); + else drainDelegations(commsBus, approvalBus, event.threadId, runDelegatedTurn); + // A settling bot frees itself as a delegation TARGET too: handoffs that + // found it busy earlier were kept queued (bounded retries) on their own + // source threads, and this is the moment they get their retry. + const settledBot = store.botByThread(event.threadId); + if (settledBot) { + for (const waitingThread of threadsWaitingOn(settledBot.id)) { + if (waitingThread !== event.threadId) drainDelegations(commsBus, approvalBus, waitingThread, runDelegatedTurn); + } + } }); // ── steer-queue drain: messages sent while the bot was busy ──────────── @@ -3559,6 +3581,40 @@ const server = createServer(async (req, res) => { // Async handoff: the source bot queues a task for a peer and goes // back to the user; the peer turn runs after the source's // turn.completed. Returns immediately (the caller does not wait). + const delegationMatch = method === "GET" ? path.match(/^\/api\/internal\/delegations\/([\w-]{4,64})$/) : null; + if (delegationMatch) { + const taskId = delegationMatch[1]; + const fromBotId = String(url.searchParams.get("fromBotId") ?? ""); + const fromThreadId = String(url.searchParams.get("fromThreadId") ?? ""); + const from = store.bot(fromBotId); + if (!from || !store.taskByThread(from.id, fromThreadId)) return json(res, 403, { error: "unknown sender" }); + const waitMs = Math.min(Math.max(Number(url.searchParams.get("wait_ms")) || 0, 0), 240_000); + const deadline = Date.now() + waitMs; + // Bounded long-poll: the delegating bot parks ONE cheap HTTP request + // here instead of burning a model inference per status check. + for (;;) { + const receipt = findDelegationReceipt(taskId); + if (receipt) { + if (receipt.sourceThreadId !== fromThreadId) { + return json(res, 403, { error: "that task belongs to a different conversation" }); + } + return json(res, 200, { status: receipt.status, toBotName: receipt.toBotName, result: receipt.result ?? "" }); + } + const stillQueued = pendingDelegationInfo(taskId); + const running = [...delegationWatch.values()].find((watch) => watch.taskId === taskId); + const owner = stillQueued?.sourceThreadId ?? running?.sourceThreadId; + if (!owner) return json(res, 404, { error: "unknown task id — delegation receipts are kept for about 48 hours" }); + if (owner !== fromThreadId) return json(res, 403, { error: "that task belongs to a different conversation" }); + if (Date.now() >= deadline) { + const toBotId = stillQueued?.toBotId ?? running?.toBotId ?? ""; + return json(res, 200, { + status: running ? "running" : "queued", + toBotName: store.bot(toBotId)?.name ?? toBotId, + }); + } + await new Promise((wake) => setTimeout(wake, 500)); + } + } if (method === "POST" && path === "/api/internal/delegate-bot") { const body = await readBody(req); const fromBotId = String(body.fromBotId ?? ""); @@ -3578,14 +3634,14 @@ const server = createServer(async (req, res) => { if (!store.taskByThread(from.id, fromThreadId)) { return json(res, 403, { error: "source thread does not belong to sender" }); } - const result = queueDelegation( + const queued = queueDelegation( commsBus, from, { toBotId, message, reason, depth }, MAX_COMMS_DEPTH, fromThreadId, ); - if (result !== "ok") { + if (queued.result !== "ok" || !queued.id) { // the agent reads this string — a bare enum ("too_deep") tells it // nothing about what to do instead const said: Record, string> = { @@ -3594,11 +3650,12 @@ const server = createServer(async (req, res) => { 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, { error: said[queued.result === "ok" ? "no_target" : queued.result] }); } const targetName = store.bot(toBotId)?.name ?? toBotId; return json(res, 200, { queued: true, + taskId: queued.id, 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.`,