From 45380d99e3838e8f35f2ad7fb8bdb6308c720015 Mon Sep 17 00:00:00 2001 From: gus <42593099+lightcloud00@users.noreply.github.com> Date: Wed, 26 Aug 2026 13:00:18 -0400 Subject: [PATCH 1/3] fix(delegations): persist busy one-hop handoffs --- server/comms-visibility.ts | 16 + server/contracts.ts | 10 + server/delegations.test.ts | 383 +++++++++-- server/delegations.ts | 964 ++++++++++++++++++++++------ server/drivers/agents-proxy.test.ts | 2 + server/drivers/agents-proxy.ts | 3 + server/index.ts | 144 ++++- server/thread-events.test.ts | 44 ++ server/thread-events.ts | 11 + src/lib/inspector.test.ts | 25 + src/lib/inspector.ts | 5 + 11 files changed, 1346 insertions(+), 261 deletions(-) diff --git a/server/comms-visibility.ts b/server/comms-visibility.ts index a46efc2ff..3011fa583 100644 --- a/server/comms-visibility.ts +++ b/server/comms-visibility.ts @@ -4,6 +4,20 @@ import { sectionKey, type BotRecord, type GroupRecord, type Message, type Store } from "./store.ts"; +/** Machine-readable delegation lifecycle evidence. The payload is deliberately + * metadata-only: task and routing ids plus bounded enum-like state. Prompts, + * tool output, credentials, and approval-card text never enter the turn log. */ +export interface DelegationAuditEvent { + type: "delegation.status"; + threadId: string; + taskId: string; + targetBotId: string; + state: "completed" | "queued" | "failed"; + reason?: string; + attemptCount: number; + duplicate?: boolean; +} + /** What a peer-exchange helper needs from the outside world: * the store (for persisted messages + groups) and the SSE broadcasters * so chat clients see the change without waiting for a refresh. */ @@ -11,6 +25,8 @@ export interface CommsBus { store: Store; /** SSE broadcast (kind: "message" envelope). */ broadcast: (payload: Record) => void; + /** Optional bridge into the canonical per-thread event ledger. */ + recordDelegation?: (event: DelegationAuditEvent) => void; /** SSE broadcast (kind: "group" envelope) for a single group. */ } diff --git a/server/contracts.ts b/server/contracts.ts index d28aafcab..97e197e11 100644 --- a/server/contracts.ts +++ b/server/contracts.ts @@ -132,6 +132,16 @@ export type RuntimeEvent = RuntimeEventBase & // `setup: true` marks a failure the user fixes by installing or // configuring something, not by retrying — the UI offers setup instead. | { type: "runtime.error"; message: string; setup?: boolean } + | { + /** Metadata-only lifecycle evidence for a durable async handoff. */ + type: "delegation.status"; + taskId: string; + targetBotId: string; + state: "completed" | "queued" | "failed"; + reason?: string; + attemptCount: number; + duplicate?: boolean; + } ); export type RuntimeEventListener = (event: RuntimeEvent) => void; diff --git a/server/delegations.test.ts b/server/delegations.test.ts index 44a2808c5..4cf3cf93c 100644 --- a/server/delegations.test.ts +++ b/server/delegations.test.ts @@ -4,31 +4,64 @@ // assert what would have been dispatched to the harness. The harness itself // stays out of these — the integration happens in comms.test.ts (the full // e2e through the agents proxy + fake ACP CLI). -import { rmSync } from "node:fs"; -import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { existsSync, mkdirSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import type { CommsBus } from "./comms-visibility.ts"; +import type { CommsBus, DelegationAuditEvent } from "./comms-visibility.ts"; import { DATA_DIR } from "./config.ts"; import type { ModelSelection } from "./contracts.ts"; import { + MAX_DELEGATION_AGE_MS, + MAX_DELEGATION_ATTEMPTS, + _loadPending, drainDelegations, + drainReadyDelegations, + discardDelegations, pendingDelegationSnapshot, - queueDelegation, + pendingThreads, + queueDelegation as queueDelegationForRun, _pendingCount, + _pendingItems, + _resetPending, + _terminalOutcome, + type DelegationItem, } from "./delegations.ts"; import { peerAllowKey, resolvePeerComms } from "./peer-approval.ts"; import { Store, type BotRecord } from "./store.ts"; const selection = (): ModelSelection => ({ instanceId: "claude", model: "fake-model" }); +const TEST_SOURCE_RUN_ID = "source-run-test-0001"; + +function queueDelegation( + bus: CommsBus, + from: BotRecord, + item: DelegationItem, + maxDepth: number, + sourceThreadId = from.threadId, + options: { nowMs?: number } = {}, +) { + return queueDelegationForRun( + bus, + from, + item, + maxDepth, + TEST_SOURCE_RUN_ID, + sourceThreadId, + options, + ); +} interface BusPair { commsBus: CommsBus; approvalBus: { store: Store; broadcast: (payload: unknown) => void }; broadcasts: unknown[]; + delegationEvents: DelegationAuditEvent[]; } function setupBuses(store: Store): BusPair { const broadcasts: unknown[] = []; + const delegationEvents: DelegationAuditEvent[] = []; const broadcast = (payload: unknown) => { broadcasts.push(payload); }; @@ -39,9 +72,13 @@ function setupBuses(store: Store): BusPair { broadcasts.push({ kind: change.type, threadId: change.threadId, message: change.message }); } }); - const commsBus: CommsBus = { store, broadcast }; + const commsBus: CommsBus = { + store, + broadcast, + recordDelegation: (event) => delegationEvents.push(event), + }; const approvalBus = { store, broadcast }; - return { commsBus, approvalBus, broadcasts }; + return { commsBus, approvalBus, broadcasts, delegationEvents }; } /** Poll until `predicate` returns a truthy value or `timeout` elapses. @@ -66,6 +103,7 @@ describe("queueDelegation", () => { beforeEach(() => { rmSync(DATA_DIR, { recursive: true, force: true }); + _resetPending(); store = new Store(selection); from = store.createBot(); target = store.createBot(); @@ -81,7 +119,7 @@ describe("queueDelegation", () => { message: "self-talk", depth: 0, }, 1); - expect(result).toBe("self"); + expect(result).toMatchObject({ state: "failed", reason: "self", duplicate: false }); expect(_pendingCount(from.threadId)).toBe(0); }); @@ -91,7 +129,17 @@ describe("queueDelegation", () => { message: "next task", depth: 1, }, 1); - expect(result).toBe("too_deep"); + expect(result).toMatchObject({ state: "failed", reason: "too_deep", duplicate: false }); + expect(_pendingCount(from.threadId)).toBe(0); + }); + + it("rejects a forged negative depth instead of weakening the one-hop cap", () => { + const result = queueDelegation(commsBus, from, { + toBotId: target.id, + message: "pretend this is a root turn", + depth: -1, + }, 1); + expect(result).toMatchObject({ state: "failed", reason: "too_deep", duplicate: false }); expect(_pendingCount(from.threadId)).toBe(0); }); @@ -101,7 +149,7 @@ describe("queueDelegation", () => { message: "where?", depth: 0, }, 1); - expect(result).toBe("no_target"); + expect(result).toMatchObject({ state: "failed", reason: "no_target", duplicate: false }); expect(_pendingCount(from.threadId)).toBe(0); }); @@ -112,7 +160,7 @@ describe("queueDelegation", () => { reason: "followup", depth: 0, }, 1); - expect(result).toBe("ok"); + expect(result).toMatchObject({ state: "queued", reason: "accepted", duplicate: false }); expect(_pendingCount(from.threadId)).toBe(1); const chip = store @@ -156,7 +204,7 @@ describe("queueDelegation", () => { routineTask.threadId, ); - expect(result).toBe("ok"); + expect(result).toMatchObject({ state: "queued", reason: "accepted", duplicate: false }); expect(_pendingCount(routineTask.threadId)).toBe(1); expect(_pendingCount(from.threadId)).toBe(0); expect( @@ -166,6 +214,49 @@ describe("queueDelegation", () => { store.messagesFor(from.threadId).some((m) => m.tool?.name === "Delegated to @Helper"), ).toBe(false); }); + + it("deduplicates retries within one source run but keeps a later turn distinct", () => { + const item = { toBotId: target.id, message: "repeatable task", depth: 0 }; + const first = queueDelegationForRun( + commsBus, + from, + item, + 1, + "source-run-same-0001", + ); + const retry = queueDelegationForRun( + commsBus, + from, + item, + 1, + "source-run-same-0001", + ); + const laterTurn = queueDelegationForRun( + commsBus, + from, + item, + 1, + "source-run-later-001", + ); + + expect(retry).toEqual({ ...first, duplicate: true }); + expect(laterTurn).toMatchObject({ state: "queued", duplicate: false }); + expect(laterTurn.taskId).not.toBe(first.taskId); + expect(_pendingCount(from.threadId)).toBe(2); + }); + + it("rejects a malformed source run id before persistence", () => { + const result = queueDelegationForRun( + commsBus, + from, + { toBotId: target.id, message: "do this", depth: 0 }, + 1, + "short", + ); + + expect(result).toMatchObject({ state: "failed", reason: "source_run_invalid" }); + expect(_pendingCount(from.threadId)).toBe(0); + }); }); describe("drainDelegations", () => { @@ -175,9 +266,11 @@ describe("drainDelegations", () => { let commsBus: CommsBus; let approvalBus: { store: Store; broadcast: (payload: unknown) => void }; let runTargetCalls: Array<{ toBotId: string; message: string; commsDepth: number; sourceThreadId?: string }>; + let delegationEvents: DelegationAuditEvent[]; beforeEach(() => { rmSync(DATA_DIR, { recursive: true, force: true }); + _resetPending(); store = new Store(selection); from = store.createBot(); target = store.createBot(); @@ -185,6 +278,7 @@ describe("drainDelegations", () => { const buses = setupBuses(store); commsBus = buses.commsBus; approvalBus = buses.approvalBus; + delegationEvents = buses.delegationEvents; runTargetCalls = []; }); @@ -269,7 +363,7 @@ describe("drainDelegations", () => { }); it("contains a rejected delegation worker and reports it on the source thread", async () => { - queueDelegation(commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1); + const queued = queueDelegation(commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1); drainDelegations(commsBus, approvalBus, from.threadId, () => { throw new Error("target runner exploded"); }); @@ -277,15 +371,75 @@ describe("drainDelegations", () => { const failure = await waitFor(() => store .messagesFor(from.threadId) - .find((m) => m.tool?.ok === false && m.tool.name.includes("target runner exploded")), + .find((m) => m.tool?.ok === false && m.tool.name.includes("failed to start")), + ); + expect(failure.tool?.name).not.toContain("target runner exploded"); + expect(_terminalOutcome(queued.taskId)).toMatchObject({ state: "failed", reason: "dispatch_failed" }); + }); + + it("terminalizes an unexpected drain failure and continues the remaining batch", async () => { + const secondTarget = store.createBot(); + store.patchBot(secondTarget.id, { name: "Second helper" }); + const first = queueDelegation( + commsBus, + from, + { toBotId: target.id, message: "first", depth: 0 }, + 1, + ); + const second = queueDelegation( + commsBus, + from, + { toBotId: secondTarget.id, message: "second", depth: 0 }, + 1, + ); + vi.spyOn(store, "createGroup").mockImplementationOnce(() => { + throw new Error("channel store unavailable"); + }); + + drainDelegations(commsBus, approvalBus, from.threadId, (toBotId) => { + runTargetCalls.push({ toBotId, message: "dispatched", commsDepth: 1 }); + return { state: "completed", reason: "dispatch_accepted" }; + }); + + await waitFor(() => _terminalOutcome(first.taskId) && _terminalOutcome(second.taskId)); + expect(_terminalOutcome(first.taskId)).toMatchObject({ + state: "failed", + reason: "dispatch_failed", + }); + expect(_terminalOutcome(second.taskId)).toMatchObject({ + state: "completed", + reason: "dispatch_accepted", + }); + expect(runTargetCalls.map((call) => call.toBotId)).toEqual([secondTarget.id]); + }); + + it("keeps an accepted dispatch completed when visibility mirroring fails", async () => { + const queued = queueDelegation( + commsBus, + from, + { toBotId: target.id, message: "accepted work", depth: 0 }, + 1, ); - expect(failure.tool?.name).toContain("delegation failed"); + vi.spyOn(store, "appendMessage").mockImplementationOnce(() => { + throw new Error("message store unavailable"); + }); + + drainDelegations(commsBus, approvalBus, from.threadId, () => ({ + state: "completed", + reason: "dispatch_accepted", + })); + + await waitFor(() => _terminalOutcome(queued.taskId)); + expect(_terminalOutcome(queued.taskId)).toMatchObject({ + state: "completed", + reason: "dispatch_accepted", + }); }); it("reports an asynchronous target-start rejection on a detached source thread", async () => { const activeThreadId = from.threadId; const routineTask = store.createTask(from.id, "Routine run", false)!; - queueDelegation( + const queued = queueDelegation( commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, @@ -299,16 +453,22 @@ describe("drainDelegations", () => { const failure = await waitFor(() => store .messagesFor(routineTask.threadId) - .find((m) => m.tool?.ok === false && m.tool.name.includes("provider disappeared")), + .find((m) => m.tool?.ok === false && m.tool.name.includes("failed to start")), ); - expect(failure.tool?.name).toContain("delegation failed"); + expect(failure.tool?.name).not.toContain("provider disappeared"); + expect(_terminalOutcome(queued.taskId)).toMatchObject({ state: "failed", reason: "dispatch_failed" }); expect( - store.messagesFor(activeThreadId).some((m) => m.tool?.name.includes("provider disappeared")), + store.messagesFor(activeThreadId).some((m) => m.tool?.name.includes("failed to start")), ).toBe(false); }); it("skips runTarget and emits a 'no such bot' chip when the target was deleted", async () => { - queueDelegation(commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1); + const queued = queueDelegation( + commsBus, + from, + { toBotId: target.id, message: "do this", depth: 0 }, + 1, + ); store.deleteBot(target.id); drainDelegations(commsBus, approvalBus, from.threadId, (toBotId, message, commsDepth) => { runTargetCalls.push({ toBotId, message, commsDepth }); @@ -316,13 +476,17 @@ describe("drainDelegations", () => { const chip = await waitFor(() => store .messagesFor(from.threadId) - .find((m) => m.kind === "activity" && (m.tool?.name ?? "").includes("no such bot")), + .find((m) => m.kind === "activity" && (m.tool?.name ?? "").includes("no longer exists")), ); expect(chip.tool?.ok).toBe(false); expect(runTargetCalls).toEqual([]); + expect(_terminalOutcome(queued.taskId)).toMatchObject({ + state: "failed", + reason: "target_missing", + }); }); - it("skips runTarget and emits a 'is busy' chip when the target is currently busy", async () => { + it("keeps one durable queued item 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) => { @@ -333,8 +497,127 @@ describe("drainDelegations", () => { .messagesFor(from.threadId) .find((m) => m.kind === "activity" && (m.tool?.name ?? "").includes("is 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 queued — target is busy; it will retry after that turn completes"); + expect(chip.tool?.ok).toBeUndefined(); + expect(runTargetCalls).toEqual([]); + expect(_pendingCount(from.threadId)).toBe(1); + expect(_pendingItems(from.threadId)[0]).toMatchObject({ waitingForTarget: true, lastReason: "target_busy" }); + }); + + it("deduplicates repeated busy-target requests by stable task id without copying task content into the ledger", async () => { + store.patchBot(target.id, { busy: true }); + const item = { + toBotId: target.id, + message: "raw-tool-output-marker: private delegated payload", + reason: "same work", + depth: 0, + }; + + const first = queueDelegation(commsBus, from, item, 1); + const duplicate = queueDelegation(commsBus, from, item, 1); + drainDelegations(commsBus, approvalBus, from.threadId, () => { + throw new Error("busy work must not dispatch"); + }); + + await waitFor(() => _pendingItems(from.threadId)[0]?.waitingForTarget === true); + expect(first).toMatchObject({ state: "queued", duplicate: false }); + expect(duplicate).toEqual({ ...first, duplicate: true }); + expect(_pendingCount(from.threadId)).toBe(1); + const ledger = JSON.stringify(delegationEvents); + expect(ledger).toContain(first.taskId); + expect(ledger).toContain('"duplicate":true'); + expect(ledger).not.toContain("raw-tool-output-marker"); + expect(ledger).not.toContain("private delegated payload"); + }); + + it("drains a busy item once on the target's synthetic turn.completed wake", async () => { + store.patchBot(target.id, { busy: true }); + const queued = queueDelegation( + commsBus, + from, + { toBotId: target.id, message: "run after idle", depth: 0 }, + 1, + ); + const runTarget = (toBotId: string, message: string, commsDepth: number) => { + runTargetCalls.push({ toBotId, message, commsDepth }); + }; + drainDelegations(commsBus, approvalBus, from.threadId, runTarget); + await waitFor(() => _pendingItems(from.threadId)[0]?.waitingForTarget === true); + + store.patchBot(target.id, { busy: false }); + drainReadyDelegations(commsBus, approvalBus, runTarget, target.id); + drainReadyDelegations(commsBus, approvalBus, runTarget, target.id); + + await waitFor(() => runTargetCalls.length === 1 && _pendingCount(from.threadId) === 0); + drainReadyDelegations(commsBus, approvalBus, runTarget, target.id); + await new Promise((resolve) => setTimeout(resolve, 25)); + expect(runTargetCalls).toHaveLength(1); + expect(_terminalOutcome(queued.taskId)).toMatchObject({ + state: "completed", + reason: "dispatch_accepted", + }); + expect( + delegationEvents.filter( + (event) => event.taskId === queued.taskId && event.state === "completed", + ), + ).toHaveLength(1); + }); + + it("expires an over-age queued item with a machine-readable reason", async () => { + const base = Date.now(); + const queued = queueDelegation( + commsBus, + from, + { toBotId: target.id, message: "too old", depth: 0 }, + 1, + from.threadId, + { nowMs: base }, + ); + drainDelegations( + commsBus, + approvalBus, + from.threadId, + (toBotId, message, commsDepth) => { + runTargetCalls.push({ toBotId, message, commsDepth }); + }, + { now: () => base + MAX_DELEGATION_AGE_MS + 1 }, + ); + + await waitFor(() => _terminalOutcome(queued.taskId)); + expect(_terminalOutcome(queued.taskId)).toMatchObject({ + state: "failed", + reason: "max_age_exceeded", + }); + expect(runTargetCalls).toEqual([]); + }); + + it("bounds repeated busy wakes and records a permanent retry reason", async () => { + const base = Date.now(); + store.patchBot(target.id, { busy: true }); + const queued = queueDelegation( + commsBus, + from, + { toBotId: target.id, message: "bounded retry", depth: 0 }, + 1, + from.threadId, + { nowMs: base }, + ); + const runTarget = (toBotId: string, message: string, commsDepth: number) => { + runTargetCalls.push({ toBotId, message, commsDepth }); + }; + + drainDelegations(commsBus, approvalBus, from.threadId, runTarget, { now: () => base }); + await waitFor(() => _pendingItems(from.threadId)[0]?.attemptCount === 1); + drainReadyDelegations(commsBus, approvalBus, runTarget, target.id, { now: () => base + 1 }); + await waitFor(() => _pendingItems(from.threadId)[0]?.attemptCount === 2); + drainReadyDelegations(commsBus, approvalBus, runTarget, target.id, { now: () => base + 2 }); + + await waitFor(() => _terminalOutcome(queued.taskId)); + expect(_terminalOutcome(queued.taskId)).toMatchObject({ + state: "failed", + reason: "target_busy_retry_limit", + attemptCount: MAX_DELEGATION_ATTEMPTS, + }); expect(runTargetCalls).toEqual([]); }); @@ -417,10 +700,6 @@ describe("drainDelegations", () => { }); }); -import { existsSync, readFileSync } 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; @@ -440,14 +719,17 @@ 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); - expect(onDisk[from.threadId][0]).toMatchObject({ toBotId: target.id, message: "do this" }); + const onDisk = JSON.parse(readFileSync(file(), "utf8")) as { pending: Record }; + expect(onDisk.pending[from.threadId]).toHaveLength(1); + expect(onDisk.pending[from.threadId][0]).toMatchObject({ toBotId: target.id, message: "do this" }); discardDelegations(buses.commsBus, from.threadId); - expect(JSON.parse(readFileSync(file(), "utf8"))[from.threadId]).toBeUndefined(); + expect(JSON.parse(readFileSync(file(), "utf8")).pending[from.threadId]).toBeUndefined(); queueDelegation(buses.commsBus, from, { toBotId: target.id, message: "again", depth: 0 }, 1); const ran: string[] = []; @@ -455,7 +737,7 @@ describe("delegations survive a restart", () => { ran.push(message); }); await waitFor(() => ran.length === 1 && pendingThreads().length === 0); - expect(JSON.parse(readFileSync(file(), "utf8"))[from.threadId]).toBeUndefined(); + expect(JSON.parse(readFileSync(file(), "utf8")).pending[from.threadId]).toBeUndefined(); }); it("keeps a handoff durable until its approval and dispatch path settles", async () => { @@ -472,11 +754,11 @@ describe("delegations survive a restart", () => { await waitFor(() => started); expect(pendingThreads()).toEqual([from.threadId]); - expect(JSON.parse(readFileSync(file(), "utf8"))[from.threadId]).toHaveLength(1); + expect(JSON.parse(readFileSync(file(), "utf8")).pending[from.threadId]).toHaveLength(1); release(); await waitFor(() => pendingThreads().length === 0); - expect(JSON.parse(readFileSync(file(), "utf8"))[from.threadId]).toBeUndefined(); + expect(JSON.parse(readFileSync(file(), "utf8")).pending[from.threadId]).toBeUndefined(); }); it("drains work queued by a later settled turn while an earlier handoff is waiting", async () => { @@ -503,12 +785,16 @@ describe("delegations survive a restart", () => { }); it("a fresh process loads what the last one queued, and can drain it", async () => { - queueDelegation(buses.commsBus, from, { toBotId: target.id, message: "left over", depth: 0 }, 1); + const item = { toBotId: target.id, message: "left over", depth: 0 }; + const first = queueDelegation(buses.commsBus, from, item, 1); + expect(queueDelegation(buses.commsBus, from, item, 1)).toEqual({ ...first, duplicate: true }); // "restart": forget memory, reload from disk _resetPending(); expect(pendingThreads()).toEqual([]); _loadPending(); expect(pendingThreads()).toEqual([from.threadId]); + expect(queueDelegation(buses.commsBus, from, item, 1)).toEqual({ ...first, duplicate: true }); + expect(_pendingCount(from.threadId)).toBe(1); const ran: string[] = []; drainDelegations(buses.commsBus, buses.approvalBus, from.threadId, async (_to, message) => { ran.push(message); @@ -518,11 +804,34 @@ describe("delegations survive a restart", () => { expect(pendingThreads()).toEqual([]); }); + it("reloads a terminal receipt so a caller retry cannot replay completed work", async () => { + const item = { toBotId: target.id, message: "exactly once", depth: 0 }; + const queued = queueDelegation(buses.commsBus, from, item, 1); + const ran: string[] = []; + const runTarget = async (_to: string, message: string) => { + ran.push(message); + }; + drainDelegations(buses.commsBus, buses.approvalBus, from.threadId, runTarget); + await waitFor(() => _terminalOutcome(queued.taskId)?.state === "completed"); + expect(ran).toHaveLength(1); + + _resetPending(); + _loadPending(); + expect(queueDelegation(buses.commsBus, from, item, 1)).toEqual({ + state: "completed", + taskId: queued.taskId, + duplicate: true, + reason: "dispatch_accepted", + }); + drainDelegations(buses.commsBus, buses.approvalBus, from.threadId, runTarget); + await new Promise((resolve) => setTimeout(resolve, 25)); + expect(ran).toHaveLength(1); + }); + it("tolerates a missing or corrupt file", () => { _resetPending(); _loadPending(); // no file expect(pendingThreads()).toEqual([]); - const { mkdirSync, writeFileSync } = require("node:fs") as typeof import("node:fs"); mkdirSync(DATA_DIR, { recursive: true }); writeFileSync(file(), "{not json"); _loadPending(); diff --git a/server/delegations.ts b/server/delegations.ts index 2308a4725..a39aea703 100644 --- a/server/delegations.ts +++ b/server/delegations.ts @@ -1,23 +1,18 @@ -// Async peer handoff (delegate_bot). +// Durable async peer handoff (delegate_bot). // -// A bot that finishes one task can hand the NEXT task to a peer without -// blocking its own turn — the source bot's turn.completed fires after it -// settles, and the queued delegation runs then. The peer gets a fresh -// depth-1 turn (depth cap still blocks A→B→C chains, see index.ts). -// -// Visiblity rides on the same comms-visibility helpers ask_bot uses -// (channel mirror + 1:1 chips) so a delegated exchange looks like an -// exchanged one. The optional approval gate (A2) is checked at drain -// time, never at queue time, because the user might have just turned -// approvePeerComms on between queueing and draining. +// A delegation waits for the source turn to settle, then dispatches once the +// chosen target is idle. Busy capacity is an event-driven queue condition, +// not a cancellation: target turn.completed events wake eligible records. +// Stable task ids and terminal receipts make retries idempotent across both +// HTTP retries and process restarts. +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"; @@ -26,69 +21,359 @@ export interface DelegationItem { message: string; reason?: string; /** The source bot's comms depth (0 for a user-initiated turn). The - * delegated-to bot runs at `depth + 1`, which equals MAX_COMMS_DEPTH - * (= 1) for a user turn — so the peer has no agents integration, and - * recursive delegation is structurally impossible. */ + * delegated-to bot runs at depth + 1. The caller still enforces the + * one-hop ceiling before anything is persisted. */ depth: number; } +export type DelegationState = "completed" | "queued" | "failed"; +export type DelegationReason = + | "accepted" + | "dispatch_accepted" + | "self" + | "too_deep" + | "no_target" + | "too_many" + | "queue_persist_failed" + | "source_run_invalid" + | "source_missing" + | "target_missing" + | "target_busy" + | "target_busy_retry_limit" + | "retry_limit_exceeded" + | "max_age_exceeded" + | "approval_denied" + | "approval_unavailable" + | "dispatch_failed" + | "source_turn_failed" + | "dispatch_outcome_unknown_after_restart"; + +export interface QueueResult { + state: DelegationState; + taskId: string; + duplicate: boolean; + reason?: DelegationReason; +} + +export interface DelegationDispatchOutcome { + state: "completed" | "queued" | "failed"; + reason?: DelegationReason; +} + +export type RunDelegatedTarget = ( + toBotId: string, + message: string, + commsDepth: number, + sourceThreadId: string, + channel?: GroupRecord, +) => void | DelegationDispatchOutcome | Promise; + +type PendingStatus = "waiting_source" | "queued" | "dispatching"; + interface PendingDelegationItem extends DelegationItem { - /** Stable acknowledgement key for crash-safe removal from the queue. */ - id: string; + taskId: string; + sourceBotId: string; + sourceThreadId: string; + sourceRunId: string; + createdAt: string; + updatedAt: string; + attemptCount: number; + maxDepth: number; + status: PendingStatus; + waitingForTarget: boolean; + lastReason?: DelegationReason; +} + +interface TerminalDelegationOutcome { + taskId: string; + sourceThreadId: string; + targetBotId: string; + state: "completed" | "failed"; + reason: DelegationReason; + attemptCount: number; + completedAt: string; +} + +interface DelegationStoreV2 { + schema: typeof STORE_SCHEMA; + pending: Record; + outcomes: Record; } -export type QueueResult = "ok" | "no_target" | "self" | "too_deep" | "too_many"; +interface DrainOptions { + now?: () => number; +} + +const STORE_SCHEMA = "openmaus.delegations.v2" as const; +const DELEGATIONS_FILE = join(DATA_DIR, "delegations.json"); +const TASK_ID = /^[0-9a-f]{64}$/; +const SOURCE_RUN_ID = /^[A-Za-z0-9_-]{16,120}$/; +const MAX_QUEUED_PER_THREAD = 4; +export const MAX_DELEGATION_ATTEMPTS = 3; +export const MAX_DELEGATION_AGE_MS = 24 * 60 * 60 * 1_000; +const TERMINAL_RETENTION_MS = 7 * 24 * 60 * 60 * 1_000; +const MAX_TERMINAL_OUTCOMES = 2_048; +const DELEGATION_REASONS = new Set([ + "accepted", + "dispatch_accepted", + "self", + "too_deep", + "no_target", + "too_many", + "queue_persist_failed", + "source_run_invalid", + "source_missing", + "target_missing", + "target_busy", + "target_busy_retry_limit", + "retry_limit_exceeded", + "max_age_exceeded", + "approval_denied", + "approval_unavailable", + "dispatch_failed", + "source_turn_failed", + "dispatch_outcome_unknown_after_restart", +]); -/** 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 - * for an unattended bot — but queued work is not a permission; the target - * and approvePeerComms are re-checked at drain time as always.) */ const pendingDelegations = new Map(); +const terminalOutcomes = new Map(); const drainingThreads = new Set(); -const DELEGATIONS_FILE = join(DATA_DIR, "delegations.json"); +const pendingCandidates = new Map>(); +const activeTargets = new Set(); + +const nowIso = (nowMs: number) => new Date(nowMs).toISOString(); +const isDelegationReason = (value: unknown): value is DelegationReason => + typeof value === "string" && DELEGATION_REASONS.has(value as DelegationReason); + +function stableTaskId( + sourceBotId: string, + sourceThreadId: string, + sourceRunId: string, + item: DelegationItem, +): string { + const identity = JSON.stringify({ + sourceBotId, + sourceThreadId, + sourceRunId, + toBotId: item.toBotId, + message: item.message, + reason: item.reason ?? "", + depth: item.depth, + }); + return createHash("sha256").update(`openmaus-delegation-v1:${identity}`).digest("hex"); +} + +function pruneTerminal(nowMs: number): void { + for (const [taskId, outcome] of terminalOutcomes) { + const completed = Date.parse(outcome.completedAt); + if (!Number.isFinite(completed) || nowMs - completed > TERMINAL_RETENTION_MS) { + terminalOutcomes.delete(taskId); + } + } + const overflow = terminalOutcomes.size - MAX_TERMINAL_OUTCOMES; + if (overflow <= 0) return; + const oldest = [...terminalOutcomes.values()] + .sort((left, right) => left.completedAt.localeCompare(right.completedAt)) + .slice(0, overflow); + for (const outcome of oldest) terminalOutcomes.delete(outcome.taskId); +} + +function saveState(nowMs = Date.now()): void { + pruneTerminal(nowMs); + const payload: DelegationStoreV2 = { + schema: STORE_SCHEMA, + pending: Object.fromEntries( + [...pendingDelegations].filter(([, items]) => items.length > 0), + ), + outcomes: Object.fromEntries(terminalOutcomes), + }; + writeFileAtomic(DELEGATIONS_FILE, JSON.stringify(payload, null, 2), { mode: 0o600 }); +} -function savePending(): void { +function record( + bus: CommsBus, + item: Pick, + state: DelegationState, + reason: DelegationReason, + duplicate = false, +): void { try { - writeFileAtomic(DELEGATIONS_FILE, JSON.stringify(Object.fromEntries(pendingDelegations), null, 2), { mode: 0o600 }); - } catch (error) { - console.error("delegations: could not persist queue", error); + bus.recordDelegation?.({ + type: "delegation.status", + threadId: item.sourceThreadId, + taskId: item.taskId, + targetBotId: item.toBotId, + state, + reason, + attemptCount: item.attemptCount, + ...(duplicate ? { duplicate: true } : {}), + }); + } catch { + /* observability never changes the queue decision */ + } +} + +function appendActivity(bus: CommsBus, threadId: string, name: string, ok?: boolean): void { + try { + bus.store.appendMessage(threadId, { + role: "bot", + kind: "activity", + tool: { name, ...(ok === undefined ? {} : { ok }) }, + }); + } catch { + /* the durable outcome and turn ledger remain authoritative */ } } -/** Load what a previous process left queued. Missing or corrupt → empty. */ +function pendingItem(threadId: string, taskId: string): PendingDelegationItem | undefined { + return pendingDelegations.get(threadId)?.find((item) => item.taskId === taskId); +} + +function validationFailure( + bus: CommsBus, + sourceBotId: string, + sourceThreadId: string, + sourceRunId: string, + item: DelegationItem, + reason: DelegationReason, +): QueueResult { + const taskId = stableTaskId(sourceBotId, sourceThreadId, sourceRunId, item); + record(bus, { taskId, sourceThreadId, toBotId: item.toBotId, attemptCount: 0 }, "failed", reason); + return { state: "failed", taskId, duplicate: false, reason }; +} + +/** Load the crash-safe queue and its bounded idempotency receipts. Legacy + * per-thread maps are migrated in memory. A record that was already marked + * dispatching is never replayed after a crash: whether the child accepted it + * is unknowable, so the safe terminal receipt is an explicit failure. */ export function _loadPending(): void { pendingDelegations.clear(); + terminalOutcomes.clear(); + let raw: unknown; try { - const raw = JSON.parse(readFileSync(DELEGATIONS_FILE, "utf8")) as Record; - for (const [threadId, list] of Object.entries(raw)) { - if (!Array.isArray(list)) continue; - const items = list.flatMap((value): PendingDelegationItem[] => { - if (!value || typeof value !== "object") return []; - const item = value as Partial; - if ( - typeof item.toBotId !== "string" || - typeof item.message !== "string" || - !Number.isFinite(item.depth) - ) return []; - return [{ - id: typeof item.id === "string" && item.id ? item.id : newId(), - toBotId: item.toBotId, - message: item.message, - ...(typeof item.reason === "string" ? { reason: item.reason } : {}), - depth: Math.max(0, Math.trunc(item.depth!)), - }]; + raw = JSON.parse(readFileSync(DELEGATIONS_FILE, "utf8")); + } catch { + return; + } + if (!raw || typeof raw !== "object" || Array.isArray(raw)) return; + const root = raw as Record; + const v2 = root.schema === STORE_SCHEMA; + const rawPending = v2 && root.pending && typeof root.pending === "object" && !Array.isArray(root.pending) + ? root.pending as Record + : root; + const rawOutcomes = v2 && root.outcomes && typeof root.outcomes === "object" && !Array.isArray(root.outcomes) + ? root.outcomes as Record + : {}; + const loadedAt = Date.now(); + let migrated = !v2; + + for (const [taskId, value] of Object.entries(rawOutcomes)) { + if (!TASK_ID.test(taskId) || !value || typeof value !== "object" || Array.isArray(value)) continue; + const outcome = value as Partial; + if ( + (outcome.state !== "completed" && outcome.state !== "failed") || + !isDelegationReason(outcome.reason) || + typeof outcome.sourceThreadId !== "string" || + typeof outcome.targetBotId !== "string" || + typeof outcome.completedAt !== "string" + ) continue; + terminalOutcomes.set(taskId, { + taskId, + sourceThreadId: outcome.sourceThreadId, + targetBotId: outcome.targetBotId, + state: outcome.state, + reason: outcome.reason, + attemptCount: Number.isFinite(outcome.attemptCount) ? Math.max(0, Math.trunc(outcome.attemptCount!)) : 0, + completedAt: outcome.completedAt, + }); + } + + for (const [threadId, value] of Object.entries(rawPending)) { + if (threadId === "schema" || threadId === "pending" || threadId === "outcomes" || !Array.isArray(value)) continue; + const items: PendingDelegationItem[] = []; + for (const candidate of value) { + if (!candidate || typeof candidate !== "object" || Array.isArray(candidate)) continue; + const item = candidate as Partial & { id?: unknown }; + if ( + typeof item.toBotId !== "string" || + typeof item.message !== "string" || + !Number.isSafeInteger(item.depth) || + item.depth! < 0 + ) continue; + const normalized: DelegationItem = { + toBotId: item.toBotId, + message: item.message, + ...(typeof item.reason === "string" ? { reason: item.reason } : {}), + depth: item.depth!, + }; + const sourceBotId = typeof item.sourceBotId === "string" ? item.sourceBotId : ""; + const sourceRunId = typeof item.sourceRunId === "string" && SOURCE_RUN_ID.test(item.sourceRunId) + ? item.sourceRunId + : `legacy_${createHash("sha256").update(`${sourceBotId}:${threadId}:${item.taskId ?? ""}`).digest("hex").slice(0, 32)}`; + const computedTaskId = stableTaskId(sourceBotId, threadId, sourceRunId, normalized); + const taskId = typeof item.taskId === "string" && TASK_ID.test(item.taskId) && item.taskId === computedTaskId + ? item.taskId + : computedTaskId; + if (item.sourceRunId !== sourceRunId || item.taskId !== taskId) migrated = true; + if (terminalOutcomes.has(taskId) || items.some((existing) => existing.taskId === taskId)) { + migrated = true; + continue; + } + const createdAt = typeof item.createdAt === "string" && Number.isFinite(Date.parse(item.createdAt)) + ? item.createdAt + : nowIso(loadedAt); + const status: PendingStatus = item.status === "queued" || item.status === "dispatching" + ? item.status + : "waiting_source"; + const attemptCount = Number.isFinite(item.attemptCount) + ? Math.max(0, Math.trunc(item.attemptCount!)) + : 0; + const maxDepth = Number.isSafeInteger(item.maxDepth) && item.maxDepth! > 0 + ? item.maxDepth! + : 1; + if (status === "dispatching") { + terminalOutcomes.set(taskId, { + taskId, + sourceThreadId: threadId, + targetBotId: normalized.toBotId, + state: "failed", + reason: "dispatch_outcome_unknown_after_restart", + attemptCount, + completedAt: nowIso(loadedAt), + }); + migrated = true; + continue; + } + items.push({ + ...normalized, + taskId, + sourceBotId, + sourceThreadId: threadId, + sourceRunId, + createdAt, + updatedAt: typeof item.updatedAt === "string" ? item.updatedAt : createdAt, + attemptCount, + maxDepth, + status, + waitingForTarget: item.waitingForTarget === true, + ...(isDelegationReason(item.lastReason) ? { lastReason: item.lastReason } : {}), }); - if (items.length) pendingDelegations.set(threadId, items); } - } catch { - /* fresh install, or unreadable — start empty */ + if (items.length) pendingDelegations.set(threadId, items); + } + pruneTerminal(loadedAt); + if (migrated) { + try { + saveState(loadedAt); + } catch (error) { + console.error("delegations: could not persist migrated queue", error); + } } } -/** Source threads with something queued — what a boot drain iterates. */ +/** Source threads with active queued work. */ export function pendingThreads(): string[] { - return [...pendingDelegations.keys()]; + return [...pendingDelegations].filter(([, items]) => items.length > 0).map(([threadId]) => threadId); } /** Read-only metadata for the local Team Map. Task prompts stay private; @@ -107,206 +392,479 @@ export function pendingDelegationSnapshot(): Array<{ ); } -/** How many handoffs one turn may queue. Small on purpose: this is the only - * thing standing between a confused bot and a fan-out of real turns. */ -const MAX_QUEUED_PER_THREAD = 4; - -/** Validate and enqueue a delegation. Pushes a "Delegated to @B: reason" - * chip to the source thread so the user can see what was queued. */ +/** Validate and enqueue one stable task. Repeated identical requests return + * the existing queued or terminal receipt and never append another item. */ export function queueDelegation( bus: CommsBus, from: BotRecord, item: DelegationItem, maxDepth: number, + sourceRunId: string, sourceThreadId = from.threadId, + options: { nowMs?: number } = {}, ): QueueResult { - if (item.toBotId === from.id) return "self"; - if (item.depth >= maxDepth) return "too_deep"; - const target = bus.store.bot(item.toBotId); - if (!target) return "no_target"; + const taskId = stableTaskId(from.id, sourceThreadId, sourceRunId, item); + if (!SOURCE_RUN_ID.test(sourceRunId)) { + return validationFailure(bus, from.id, sourceThreadId, sourceRunId, item, "source_run_invalid"); + } + if (item.toBotId === from.id) { + return validationFailure(bus, from.id, sourceThreadId, sourceRunId, item, "self"); + } + if ( + !Number.isSafeInteger(item.depth) || + item.depth < 0 || + !Number.isSafeInteger(maxDepth) || + maxDepth < 1 || + item.depth >= maxDepth + ) return validationFailure(bus, from.id, sourceThreadId, sourceRunId, item, "too_deep"); + + const terminal = terminalOutcomes.get(taskId); + if (terminal) { + record( + bus, + { taskId, sourceThreadId, toBotId: item.toBotId, attemptCount: terminal.attemptCount }, + terminal.state, + terminal.reason, + true, + ); + return { state: terminal.state, taskId, duplicate: true, reason: terminal.reason }; + } 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() }); - pendingDelegations.set(sourceThreadId, list); - savePending(); - const label = `Delegated to @${target.name}${item.reason ? `: ${item.reason}` : ""}`; - bus.store.appendMessage(sourceThreadId, { - role: "bot", - kind: "activity", - tool: { name: label }, - }); - return "ok"; -} + const existing = list.find((candidate) => candidate.taskId === taskId); + if (existing) { + record(bus, existing, "queued", existing.lastReason ?? "accepted", true); + return { + state: "queued", + taskId, + duplicate: true, + reason: existing.lastReason ?? "accepted", + }; + } + if (!bus.store.bot(item.toBotId)) { + return validationFailure(bus, from.id, sourceThreadId, sourceRunId, item, "no_target"); + } + if (list.length >= MAX_QUEUED_PER_THREAD) { + return validationFailure(bus, from.id, sourceThreadId, sourceRunId, item, "too_many"); + } -/** Drain queued delegations for a source thread (called on its - * turn.completed). Each item is processed independently: a deny, a busy - * target, or an error in one does not stop the rest. The actual start - * of the target turn is delegated to `runTarget` so delegations.ts - * stays free of harness-level concerns (commsDepth is the only thing - * the caller needs). */ -export function drainDelegations( - bus: CommsBus, - approvalBus: ApprovalBus, - threadId: string, - runTarget: ( - toBotId: string, - message: string, - commsDepth: number, - sourceThreadId: string, - channel?: GroupRecord, - ) => void | Promise, -): void { - if (drainingThreads.has(threadId)) return; - const list = pendingDelegations.get(threadId); - if (!list?.length) return; - const from = bus.store.botByThread(threadId); - if (!from) { - pendingDelegations.delete(threadId); - savePending(); - return; + const nowMs = options.nowMs ?? Date.now(); + const createdAt = nowIso(nowMs); + const queued: PendingDelegationItem = { + ...item, + taskId, + sourceBotId: from.id, + sourceThreadId, + sourceRunId, + createdAt, + updatedAt: createdAt, + attemptCount: 0, + maxDepth, + status: "waiting_source", + waitingForTarget: false, + }; + list.push(queued); + pendingDelegations.set(sourceThreadId, list); + try { + saveState(nowMs); + } catch { + const remaining = list.filter((candidate) => candidate.taskId !== taskId); + if (remaining.length) pendingDelegations.set(sourceThreadId, remaining); + else pendingDelegations.delete(sourceThreadId); + return validationFailure(bus, from.id, sourceThreadId, sourceRunId, item, "queue_persist_failed"); } - const snapshot = [...list]; - drainingThreads.add(threadId); - void (async () => { - for (const item of snapshot) { - try { - await processOne(bus, approvalBus, from, threadId, item, runTarget); - } catch (error) { - const why = error instanceof Error ? error.message : String(error); - try { - bus.store.appendMessage(threadId, { - role: "bot", - kind: "activity", - tool: { name: `error: delegation failed — ${why.slice(0, 120)}`, ok: false }, - }); - } catch (reportError) { - console.error("delegation failed and could not be reported", reportError); - } - } finally { - 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) { - drainDelegations(bus, approvalBus, threadId, runTarget); - } - }); + + const target = bus.store.bot(item.toBotId)!; + appendActivity( + bus, + sourceThreadId, + `Delegated to @${target.name}${item.reason ? `: ${item.reason}` : ""}`, + ); + record(bus, queued, "queued", "accepted"); + return { state: "queued", taskId, duplicate: false, reason: "accepted" }; } -/** Remove one terminal handoff only after approval/dispatch has settled. */ -function acknowledgeDelegation(threadId: string, itemId: string): void { +function removePending(threadId: string, taskId: string): void { const current = pendingDelegations.get(threadId); if (!current) return; - const remaining = current.filter((item) => item.id !== itemId); + const remaining = current.filter((item) => item.taskId !== taskId); if (remaining.length) pendingDelegations.set(threadId, remaining); else pendingDelegations.delete(threadId); - savePending(); } -/** 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 { - const list = pendingDelegations.get(threadId); - if (!list?.length) return; - 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 }, +function settle( + bus: CommsBus, + item: PendingDelegationItem, + state: "completed" | "failed", + reason: DelegationReason, + nowMs: number, +): void { + removePending(item.sourceThreadId, item.taskId); + terminalOutcomes.set(item.taskId, { + taskId: item.taskId, + sourceThreadId: item.sourceThreadId, + targetBotId: item.toBotId, + state, + reason, + attemptCount: item.attemptCount, + completedAt: nowIso(nowMs), }); + try { + saveState(nowMs); + } catch (error) { + // Keep the in-process receipt so a caller retry still deduplicates. The + // last durable state is dispatching, which load converts to an explicit + // unknown-outcome failure rather than replaying it. + console.error("delegations: could not persist terminal receipt", error); + } + record(bus, item, state, reason); +} + +function deferBusy(bus: CommsBus, item: PendingDelegationItem, targetName: string, nowMs: number): void { + if (item.attemptCount >= MAX_DELEGATION_ATTEMPTS) { + settle(bus, item, "failed", "target_busy_retry_limit", nowMs); + appendActivity( + bus, + item.sourceThreadId, + `Delegation to @${targetName} failed — busy retry limit reached`, + false, + ); + return; + } + const firstDeferral = item.lastReason !== "target_busy"; + item.status = "queued"; + item.waitingForTarget = true; + item.lastReason = "target_busy"; + item.updatedAt = nowIso(nowMs); + try { + saveState(nowMs); + } catch { + settle(bus, item, "failed", "queue_persist_failed", nowMs); + return; + } + record(bus, item, "queued", "target_busy"); + if (firstDeferral) { + appendActivity( + bus, + item.sourceThreadId, + `Delegation to @${targetName} queued — target is busy; it will retry after that turn completes`, + ); + } +} + +function dispatchOutcome(value: unknown): DelegationDispatchOutcome { + if (!value || typeof value !== "object") return { state: "completed", reason: "dispatch_accepted" }; + const candidate = value as Partial; + if (candidate.state === "queued") return { state: "queued", reason: candidate.reason ?? "target_busy" }; + if (candidate.state === "failed") return { state: "failed", reason: candidate.reason ?? "dispatch_failed" }; + return { state: "completed", reason: "dispatch_accepted" }; } async function processOne( bus: CommsBus, approvalBus: ApprovalBus, - from: BotRecord, - sourceThreadId: string, - item: DelegationItem, - runTarget: ( - toBotId: string, - message: string, - commsDepth: number, - sourceThreadId: string, - channel?: GroupRecord, - ) => void | Promise, + threadId: string, + taskId: string, + runTarget: RunDelegatedTarget, + clock: () => number, ): Promise { - let sender = from; + const item = pendingItem(threadId, taskId); + if (!item || item.status === "dispatching") return; + const nowMs = clock(); + const created = Date.parse(item.createdAt); + if (!Number.isFinite(created) || nowMs - created > MAX_DELEGATION_AGE_MS) { + settle(bus, item, "failed", "max_age_exceeded", nowMs); + appendActivity(bus, threadId, "Delegation expired — maximum queue age exceeded", false); + return; + } + if (item.depth >= item.maxDepth) { + settle(bus, item, "failed", "too_deep", nowMs); + return; + } + if (item.attemptCount >= MAX_DELEGATION_ATTEMPTS) { + settle(bus, item, "failed", "retry_limit_exceeded", nowMs); + appendActivity(bus, threadId, "Delegation failed — retry limit reached", false); + return; + } + const from = bus.store.botByThread(threadId); + if ( + !from || + (item.sourceBotId && from.id !== item.sourceBotId) || + !bus.store.taskByThread(from.id, threadId) + ) { + settle(bus, item, "failed", "source_missing", nowMs); + return; + } 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 }, - }); + settle(bus, item, "failed", "target_missing", nowMs); + appendActivity(bus, threadId, "Delegation failed — target bot no longer exists", false); return; } - if (target.busy) { - bus.store.appendMessage(sourceThreadId, { - role: "bot", - kind: "activity", - tool: { name: `Delegation to @${target.name} canceled — @${target.name} is busy`, ok: false }, - }); + + item.attemptCount += 1; + item.updatedAt = nowIso(nowMs); + try { + saveState(nowMs); + } catch { + settle(bus, item, "failed", "queue_persist_failed", nowMs); + return; + } + if (target.busy || activeTargets.has(target.id)) { + deferBusy(bus, item, target.name, nowMs); return; } + + let sender = from; if (sender.approvePeerComms) { - const verdict = await requestPeerApproval( - approvalBus, - sender, - target, - item.message, - "delegate_bot", - sourceThreadId, - ); + let verdict: "allow" | "deny"; + try { + verdict = await requestPeerApproval( + approvalBus, + sender, + target, + item.message, + "delegate_bot", + threadId, + ); + } catch { + settle(bus, item, "failed", "approval_unavailable", clock()); + return; + } if (verdict !== "allow") { - bus.store.appendMessage(sourceThreadId, { - role: "bot", - kind: "activity", - tool: { name: `Delegation to @${target.name} denied by user`, ok: false }, - }); + settle(bus, item, "failed", "approval_denied", clock()); + appendActivity(bus, threadId, `Delegation to @${target.name} denied by user`, false); return; } - // 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 - // busy, or an allow can start a second turn on a bot that is mid-turn — - // 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.busy) { - bus.store.appendMessage(sourceThreadId, { - role: "bot", - kind: "activity", - tool: { name: `Delegation to @${current.name} canceled — @${current.name} is busy`, ok: false }, - }); + if (!current || !currentSender || !bus.store.taskByThread(currentSender.id, threadId)) { + settle(bus, item, "failed", current ? "source_missing" : "target_missing", clock()); return; } sender = currentSender; target = current; + if (target.busy || activeTargets.has(target.id)) { + deferBusy(bus, item, target.name, clock()); + return; + } } - 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); + + activeTargets.add(target.id); + try { + item.status = "dispatching"; + item.waitingForTarget = false; + item.updatedAt = nowIso(clock()); + try { + saveState(clock()); + } catch { + settle(bus, item, "failed", "queue_persist_failed", clock()); + return; + } + const channel = getOrCreateChannel(bus.store, sender, target); + 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}`; + let outcome: DelegationDispatchOutcome; + try { + outcome = dispatchOutcome( + await runTarget(item.toBotId, prefixed, item.depth + 1, threadId, channel), + ); + } catch (error) { + const status = (error as { status?: unknown } | null)?.status; + const message = error instanceof Error ? error.message : String(error); + outcome = status === 409 && /busy|already working/i.test(message) + ? { state: "queued", reason: "target_busy" } + : { state: "failed", reason: "dispatch_failed" }; + } + if (outcome.state === "queued") { + deferBusy(bus, item, target.name, clock()); + return; + } + if (outcome.state === "failed") { + settle(bus, item, "failed", outcome.reason ?? "dispatch_failed", clock()); + appendActivity(bus, threadId, `Delegation to @${target.name} failed to start`, false); + return; + } + try { + mirrorExchange(bus, sender, target, item.message, channel, threadId); + } catch (error) { + // Provider submission and harness ownership are already accepted. + // A visibility write must not downgrade that durable dispatch receipt. + console.error("delegations: could not mirror accepted dispatch", error); + } + settle(bus, item, "completed", "dispatch_accepted", clock()); + } finally { + activeTargets.delete(target.id); + } +} + +function scheduleDrain( + bus: CommsBus, + approvalBus: ApprovalBus, + threadId: string, + taskIds: Iterable, + runTarget: RunDelegatedTarget, + options: DrainOptions, +): void { + const candidates = pendingCandidates.get(threadId) ?? new Set(); + for (const taskId of taskIds) candidates.add(taskId); + if (!candidates.size) return; + pendingCandidates.set(threadId, candidates); + if (drainingThreads.has(threadId)) return; + drainingThreads.add(threadId); + const clock = options.now ?? Date.now; + void (async () => { + for (;;) { + const pending = pendingCandidates.get(threadId); + if (!pending?.size) break; + const batch = [...pending]; + pending.clear(); + for (const taskId of batch) { + try { + await processOne(bus, approvalBus, threadId, taskId, runTarget, clock); + } catch (error) { + // A store/channel/observer failure must be scoped to this exact + // task. Keep draining the batch and consume the rejection so the + // fire-and-forget scheduler can never become an unhandled promise. + const item = pendingItem(threadId, taskId); + if (item) { + let failedAt: number; + try { + failedAt = clock(); + } catch { + failedAt = Date.now(); + } + settle(bus, item, "failed", "dispatch_failed", failedAt); + appendActivity(bus, threadId, "Delegation failed during queue dispatch", false); + } + console.error("delegations: contained drain failure", error); + } + } + } + })().finally(() => { + drainingThreads.delete(threadId); + if (!pendingCandidates.get(threadId)?.size) pendingCandidates.delete(threadId); + else scheduleDrain(bus, approvalBus, threadId, [], runTarget, options); + }); } -/** Test helper: how many items remain queued for a thread. */ +/** Mark one source turn's handoffs ready and attempt every active item once. + * A busy result remains durable and is not re-added to this drain cycle. */ +export function drainDelegations( + bus: CommsBus, + approvalBus: ApprovalBus, + threadId: string, + runTarget: RunDelegatedTarget, + options: DrainOptions = {}, +): void { + const list = pendingDelegations.get(threadId); + if (!list?.length) return; + const nowMs = (options.now ?? Date.now)(); + let changed = false; + for (const item of list) { + if (item.status === "waiting_source") { + item.status = "queued"; + item.updatedAt = nowIso(nowMs); + changed = true; + } + } + if (changed) { + try { + saveState(nowMs); + } catch { + for (const item of [...list]) settle(bus, item, "failed", "queue_persist_failed", nowMs); + return; + } + } + scheduleDrain( + bus, + approvalBus, + threadId, + list.filter((item) => item.status === "queued").map((item) => item.taskId), + runTarget, + options, + ); +} + +/** Wake busy-deferred work from a target's turn.completed transition. With + * an exact target id the event is authoritative; without one (room turns do + * not retain their speaker after folding) every now-idle target is scanned. + * This is event-driven only: there is no timer or polling loop. */ +export function drainReadyDelegations( + bus: CommsBus, + approvalBus: ApprovalBus, + runTarget: RunDelegatedTarget, + targetBotId?: string, + options: DrainOptions = {}, +): void { + const byThread = new Map(); + for (const [threadId, items] of pendingDelegations) { + for (const item of items) { + if (item.status !== "queued" || !item.waitingForTarget) continue; + if (targetBotId && item.toBotId !== targetBotId) continue; + if (!targetBotId && bus.store.bot(item.toBotId)?.busy) continue; + const ids = byThread.get(threadId) ?? []; + ids.push(item.taskId); + byThread.set(threadId, ids); + } + } + for (const [threadId, taskIds] of byThread) { + scheduleDrain(bus, approvalBus, threadId, taskIds, runTarget, options); + } +} + +/** A failed/interrupted source turn terminalizes its own not-yet-dispatched + * fan-out. User words queued from other source threads are untouched. */ +export function discardDelegations(bus: CommsBus, threadId: string): void { + const list = [...(pendingDelegations.get(threadId) ?? [])]; + if (!list.length) return; + const nowMs = Date.now(); + for (const item of list) { + removePending(threadId, item.taskId); + terminalOutcomes.set(item.taskId, { + taskId: item.taskId, + sourceThreadId: threadId, + targetBotId: item.toBotId, + state: "failed", + reason: "source_turn_failed", + attemptCount: item.attemptCount, + completedAt: nowIso(nowMs), + }); + record(bus, item, "failed", "source_turn_failed"); + } + try { + saveState(nowMs); + } catch (error) { + console.error("delegations: could not persist discarded queue", error); + } + appendActivity( + bus, + threadId, + `${list.length} queued delegation${list.length > 1 ? "s" : ""} dropped — the source turn did not finish`, + false, + ); +} + +/** Test helper: how many active items remain for a source thread. */ export function _pendingCount(threadId: string): number { return pendingDelegations.get(threadId)?.length ?? 0; } -/** Test helper: forget the in-memory queue (a simulated restart). */ +/** Test helper: a copy of the persisted active record. */ +export function _pendingItems(threadId: string): ReadonlyArray> { + return (pendingDelegations.get(threadId) ?? []).map((item) => ({ ...item })); +} + +/** Test helper: read an idempotency receipt without exposing task content. */ +export function _terminalOutcome(taskId: string): Readonly | undefined { + const outcome = terminalOutcomes.get(taskId); + return outcome ? { ...outcome } : undefined; +} + +/** Test helper: simulate a fresh process. */ export function _resetPending(): void { pendingDelegations.clear(); + terminalOutcomes.clear(); + pendingCandidates.clear(); drainingThreads.clear(); + activeTargets.clear(); } diff --git a/server/drivers/agents-proxy.test.ts b/server/drivers/agents-proxy.test.ts index b686cc754..8cf17781f 100644 --- a/server/drivers/agents-proxy.test.ts +++ b/server/drivers/agents-proxy.test.ts @@ -106,6 +106,7 @@ beforeAll(async () => { OMB_HARNESS_URL: `http://127.0.0.1:${stubPort}`, OMB_BOT_ID: "bot-asker", OMB_THREAD_ID: "thread-asker-routine", + OMB_SOURCE_RUN_ID: "source-run-proxy-0001", OMB_COMMS_TOKEN: TOKEN, OMB_TURN_DEPTH: "0", }, @@ -192,6 +193,7 @@ describe("agents-proxy MCP surface", () => { expect(lastDelegateBody).toMatchObject({ fromBotId: "bot-asker", fromThreadId: "thread-asker-routine", + sourceRunId: "source-run-proxy-0001", toBotId: "bot-helper", message: "take this", reason: "follow-up", diff --git a/server/drivers/agents-proxy.ts b/server/drivers/agents-proxy.ts index c949db741..cddf655ae 100644 --- a/server/drivers/agents-proxy.ts +++ b/server/drivers/agents-proxy.ts @@ -19,6 +19,7 @@ // the harness when it builds the integration: // OMB_HARNESS_URL base URL of the harness (http://127.0.0.1:8799) // OMB_BOT_ID the calling bot's id (excluded from list_bots; sender) +// OMB_SOURCE_RUN_ID harness-generated identity for this exact source turn // OMB_COMMS_TOKEN shared secret for the localhost-only internal endpoints // OMB_TURN_DEPTH this turn's comms depth (the harness refuses recursion) import readline from "node:readline"; @@ -28,6 +29,7 @@ import { CREDENTIAL_TARGETS, isCredentialTargetId } from "../../shared/credentia const HARNESS = process.env.OMB_HARNESS_URL ?? "http://127.0.0.1:8799"; const BOT_ID = process.env.OMB_BOT_ID ?? ""; const THREAD_ID = process.env.OMB_THREAD_ID ?? ""; +const SOURCE_RUN_ID = process.env.OMB_SOURCE_RUN_ID ?? ""; const TOKEN = process.env.OMB_COMMS_TOKEN ?? ""; const DEPTH = Number(process.env.OMB_TURN_DEPTH ?? "0") || 0; const MAX_CREATED_PER_TURN = 4; @@ -152,6 +154,7 @@ async function callTool(name: string, args: Json): Promise<{ text: string; isErr const body: Record = { fromBotId: BOT_ID, fromThreadId: THREAD_ID, + sourceRunId: SOURCE_RUN_ID, toBotId, message, depth: DEPTH, diff --git a/server/index.ts b/server/index.ts index a758aa409..53e240139 100644 --- a/server/index.ts +++ b/server/index.ts @@ -77,7 +77,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, drainReadyDelegations, pendingDelegationSnapshot, pendingThreads, queueDelegation } from "./delegations.ts"; import { drainSteeredMessages, queueSteeredMessage } from "./steer-queue.ts"; import { EventBus } from "./harness/bus.ts"; import { ProviderRegistry } from "./harness/registry.ts"; @@ -205,6 +205,7 @@ function authorizedComms(header: string | string[] | undefined): boolean { // A→B is allowed but B→C (and A→B→A loops) never start. const MAX_COMMS_DEPTH = 1; const MAX_WORKSPACE_BOTS = 100; +const sourceRunIdSchema = z.string().min(16).max(120).regex(/^[A-Za-z0-9_-]+$/); // Resolved from the server root — see server/proxy-paths.ts. This descending // path happened to survive bundling, but it goes through the same anchor so // there is exactly one way proxies are located. @@ -213,7 +214,7 @@ const phoneProxyPath = SPAWNED_PROXIES.phone; // in the packaged app process.execPath is Electron — run the proxy as node const AGENTS_NODE_FLAG = { ELECTRON_RUN_AS_NODE: "1" }; -function agentsIntegration(botId: string, threadId: string, depth: number) { +function agentsIntegration(botId: string, threadId: string, depth: number, sourceRunId: string) { return { command: process.execPath, args: [agentsProxyPath], @@ -222,6 +223,7 @@ function agentsIntegration(botId: string, threadId: string, depth: number) { OMB_HARNESS_URL: `http://127.0.0.1:${PORT}`, OMB_BOT_ID: botId, OMB_THREAD_ID: threadId, + OMB_SOURCE_RUN_ID: sourceRunId, OMB_COMMS_TOKEN: COMMS_TOKEN, OMB_TURN_DEPTH: String(depth), }, @@ -1191,15 +1193,36 @@ const runDelegatedTurn: Parameters[3] = (toBotId, text, tool: { name: `error: delegation to @${bot?.name ?? toBotId} could not start — ${why.slice(0, 120)}`, ok: false }, }); }; - return 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); + return new Promise((resolve) => { + let resolved = false; + const finish = (outcome: { state: "completed" | "queued" | "failed"; reason: "dispatch_accepted" | "target_busy" | "dispatch_failed" }) => { + if (resolved) return; + resolved = true; + resolve(outcome); + }; + startTurn(toBotId, text, { + commsDepth, + unattended: isUnattended(store.botByThread(sourceThreadId)?.id), + // The queue becomes terminal only after the provider accepted the + // turn and the harness recorded ownership for this task. + onDispatchAccepted: () => finish({ state: "completed", reason: "dispatch_accepted" }), + // Asynchronous integration/provider setup failures do not emit + // turn.completed, so report and settle them through this callback. + onDispatchError: (message) => { + reportStartFailure(message); + finish({ state: "failed", reason: "dispatch_failed" }); + }, + }).catch((err) => { + const status = (err as { status?: unknown } | null)?.status; + const why = err instanceof Error ? err.message : String(err); + if (status === 409 && /busy|already working/i.test(why)) { + if (targetThreadId) delegationWatch.delete(targetThreadId); + finish({ state: "queued", reason: "target_busy" }); + return; + } + reportStartFailure(err); + finish({ state: "failed", reason: "dispatch_failed" }); + }); }); }; @@ -1212,6 +1235,20 @@ bus.subscribe((event: RuntimeEvent) => { drainDelegations(commsBus, approvalBus, event.threadId, runDelegatedTurn); }); +// Busy-target handoffs wake only from a real turn transition. Direct turns +// retain an exact bot id; room turns have already released their speaker by +// this subscriber, so the fallback scans all now-idle targets once. There is +// no timer and no retry polling. +bus.subscribe((event: RuntimeEvent) => { + if (event.type !== "turn.completed") return; + drainReadyDelegations( + commsBus, + approvalBus, + runDelegatedTurn, + store.botByThread(event.threadId)?.id, + ); +}); + // ── steer-queue drain: messages sent while the bot was busy ──────────── // Runs on ANY turn.completed rather than resolving the settling thread: a // bot busy in a room settles on the room's thread, and by the time this @@ -1378,6 +1415,8 @@ async function startTurn( cardContinuation?: boolean; /** Earlier text message this user turn is replying to. */ replyTo?: Message; + /** Provider submission and task ownership both succeeded. */ + onDispatchAccepted?: () => void; onDispatchError?: (message: string) => void; }, ) { @@ -1390,6 +1429,10 @@ async function startTurn( } if (bot.busy) throw Object.assign(new Error("the bot is already working — interrupt it first"), { status: 409 }); const threadId = opts?.threadId ?? bot.threadId; + // Harness-owned identity for this exact source turn. Agent tool retries + // reuse the mounted proxy and therefore this id; a later turn receives a + // new id even when it delegates identical text to the same target. + const sourceRunId = randomUUID(); // a webhook turn, or one inherited from a bot already running unattended if (opts?.automationSource === "webhook" || opts?.unattended) markUnattended(bot.id); // a person typing into this bot ends the unattended window immediately @@ -1715,7 +1758,7 @@ async function startTurn( commsDepth < MAX_COMMS_DEPTH && instance.adapter.capabilities.agentsMcp === true ) { - integrations.agents = agentsIntegration(bot.id, threadId, commsDepth); + integrations.agents = agentsIntegration(bot.id, threadId, commsDepth, sourceRunId); } // @mentions in the user's message (the composer's tagging UI) become // an explicit delegation nudge — the agent still does the ask_bot call @@ -1795,6 +1838,12 @@ 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); + try { + opts?.onDispatchAccepted?.(); + } catch { + // Dispatch ownership is already committed; a receipt callback must + // never convert an accepted provider turn into an apparent failure. + } // 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 @@ -1914,7 +1963,18 @@ function serializeRoomContext(threadId: string, userName: string): string { // comms bus: passed into the visibility helpers in comms-visibility.ts so // they can mirror messages + chips without re-deriving SSE plumbing. Same // shape every comms entry point uses (ask_bot, delegate_bot). -const commsBus: CommsBus = { store, broadcast }; +const commsBus: CommsBus = { + store, + broadcast, + recordDelegation: (event) => { + bus.publish({ + eventId: randomUUID(), + provider: "openmausbot", + createdAt: new Date().toISOString(), + ...event, + }); + }, +}; // approval bus: peer-approval.ts only needs to push cards and broadcast // them — its pending map lives in the module so the two respond endpoints @@ -2752,10 +2812,15 @@ const server = createServer(async (req, res) => { const fromBotId = String(body.fromBotId ?? ""); const toBotId = String(body.toBotId ?? ""); const message = String(body.message ?? "").trim(); - const depth = Number(body.depth ?? 0) || 0; + const depth = body.depth === undefined ? 0 : body.depth; if (!toBotId || !message) return json(res, 400, { error: "toBotId and message required" }); if (toBotId === fromBotId) return json(res, 400, { error: "a bot cannot message itself" }); - if (depth >= MAX_COMMS_DEPTH) return json(res, 200, { error: "message chains are limited to one hop" }); + if (!Number.isSafeInteger(depth) || (depth as number) < 0) { + return json(res, 400, { error: "depth must be a non-negative safe integer" }); + } + if ((depth as number) >= MAX_COMMS_DEPTH) { + return json(res, 200, { error: "message chains are limited to one hop" }); + } const target = store.bot(toBotId); if (!target) return json(res, 404, { error: "no such bot" }); if (target.busy) return json(res, 200, { busy: true }); @@ -2814,7 +2879,7 @@ const server = createServer(async (req, res) => { const channel = getOrCreateChannel(store, currentFrom, currentTarget); mirrorExchange(commsBus, currentFrom, currentTarget, message, channel, fromThreadId); const prefixed = `[Message from @${currentFrom.name}, another bot in this OpenMausBot workspace. Reply to them.]\n\n${message}`; - const reply = await askBotAndWait(toBotId, prefixed, depth, fromBotId); + const reply = await askBotAndWait(toBotId, prefixed, depth as number, fromBotId); mirrorReply(commsBus, currentTarget, reply, channel); return json(res, 200, { botName: currentTarget.name, text: reply }); } @@ -2827,8 +2892,13 @@ const server = createServer(async (req, res) => { const toBotId = String(body.toBotId ?? ""); const message = String(body.message ?? "").trim(); const reason = typeof body.reason === "string" && body.reason.trim() ? body.reason.trim() : undefined; - const depth = Number(body.depth ?? 0) || 0; + const depth = body.depth === undefined ? 0 : body.depth; + const sourceRun = sourceRunIdSchema.safeParse(body.sourceRunId); if (!toBotId || !message) return json(res, 400, { error: "toBotId and message required" }); + if (!sourceRun.success) return json(res, 400, { error: "sourceRunId is invalid" }); + if (!Number.isSafeInteger(depth) || (depth as number) < 0) { + return json(res, 400, { error: "depth must be a non-negative safe integer" }); + } const from = store.bot(fromBotId); if (!from) return json(res, 404, { error: "no such bot" }); const target = store.bot(toBotId); @@ -2843,23 +2913,55 @@ const server = createServer(async (req, res) => { const result = queueDelegation( commsBus, from, - { toBotId, message, reason, depth }, + { toBotId, message, reason, depth: depth as number }, MAX_COMMS_DEPTH, + sourceRun.data, fromThreadId, ); - 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", + queue_persist_failed: "the delegation queue could not be persisted", + source_run_invalid: "the source turn identity is invalid", + source_missing: "the source task no longer exists", + target_missing: "the target bot no longer exists", + target_busy_retry_limit: "the target stayed busy until the retry limit expired", + retry_limit_exceeded: "the delegation retry limit expired", + max_age_exceeded: "the delegation expired before it could run", + approval_denied: "the user denied this delegation", + approval_unavailable: "the delegation approval channel became unavailable", + dispatch_failed: "the target could not start the delegated turn", + source_turn_failed: "the source turn failed before delegation", + dispatch_outcome_unknown_after_restart: "the prior dispatch outcome is unknown after restart", }; - return json(res, 200, { error: said[result] }); + return json(res, 200, { + state: "failed", + taskId: result.taskId, + duplicate: result.duplicate, + reason: result.reason, + error: said[result.reason ?? ""] ?? "delegation failed", + }); } const targetName = store.bot(toBotId)?.name ?? toBotId; + if (result.state === "completed") { + return json(res, 200, { + state: "completed", + taskId: result.taskId, + duplicate: result.duplicate, + queued: false, + message: `Delegation already dispatched to @${targetName}.`, + }); + } return json(res, 200, { + state: "queued", + taskId: result.taskId, + duplicate: result.duplicate, + reason: result.reason, queued: true, message: from.approvePeerComms ? `Queued for review — @${targetName} will only pick it up if the user approves after your turn finishes.` diff --git a/server/thread-events.test.ts b/server/thread-events.test.ts index 87d20b5e1..6b762701e 100644 --- a/server/thread-events.test.ts +++ b/server/thread-events.test.ts @@ -118,6 +118,50 @@ describe("readThreadEvents", () => { expect(page.entries.map((entry) => (entry.data as { eventId: string }).eventId)).toEqual(["valid-retry"]); }); + it("retains metadata-only delegation lifecycle events and rejects malformed states", () => { + const eventsDir = tmp(); + const nativeDir = tmp(); + writeFileSync( + join(eventsDir, "t1.ndjson"), + line(runtime({ + eventId: "queued", + createdAt: "2026-08-17T10:00:00.000Z", + type: "delegation.status", + taskId: "a".repeat(64), + targetBotId: "helper", + state: "queued", + reason: "target_busy", + attemptCount: 1, + })) + + line(runtime({ + eventId: "bad", + createdAt: "2026-08-17T10:00:01.000Z", + type: "delegation.status", + taskId: "b".repeat(64), + targetBotId: "helper", + state: "waiting", + attemptCount: 1, + })) + + line(runtime({ + eventId: "completed", + createdAt: "2026-08-17T10:00:02.000Z", + type: "delegation.status", + taskId: "a".repeat(64), + targetBotId: "helper", + state: "completed", + reason: "dispatch_accepted", + attemptCount: 2, + })), + ); + + const page = readThreadEvents({ eventsDir, nativeDir, threadId: "t1" }); + expect(page.entries.map((entry) => (entry.data as { eventId: string }).eventId)).toEqual([ + "queued", + "completed", + ]); + expect(page.total.runtime).toBe(3); + }); + it("keeps walking backward when a corrupt tail record would otherwise consume the limit", () => { const eventsDir = tmp(); const nativeDir = tmp(); diff --git a/server/thread-events.ts b/server/thread-events.ts index fce043ecc..68c95b285 100644 --- a/server/thread-events.ts +++ b/server/thread-events.ts @@ -228,6 +228,17 @@ function isRuntimeEvent(value: unknown): value is RuntimeEvent { return typeof value.input === "number" && typeof value.output === "number"; case "runtime.error": return typeof value.message === "string" && (value.setup === undefined || typeof value.setup === "boolean"); + case "delegation.status": + return ( + typeof value.taskId === "string" && + typeof value.targetBotId === "string" && + (value.state === "completed" || value.state === "queued" || value.state === "failed") && + stringOrMissing(value.reason) && + typeof value.attemptCount === "number" && + Number.isInteger(value.attemptCount) && + value.attemptCount >= 0 && + (value.duplicate === undefined || typeof value.duplicate === "boolean") + ); default: return false; } diff --git a/src/lib/inspector.test.ts b/src/lib/inspector.test.ts index fd69971c5..f01782b6a 100644 --- a/src/lib/inspector.test.ts +++ b/src/lib/inspector.test.ts @@ -25,6 +25,31 @@ describe("summarizeRuntime", () => { expect(summary.length).toBeLessThanOrEqual("assistant: ".length + 120); expect(summary).not.toContain("\n"); }); + + it("summarizes durable delegation states without task content", () => { + const taskId = "abcdef1234567890"; + expect(summarizeRuntime({ + ...base, + type: "delegation.status", + taskId, + targetBotId: "helper", + state: "queued", + reason: "target_busy", + attemptCount: 1, + })).toEqual({ + summary: "delegation queued · abcdef12 · target_busy", + tone: "plain", + }); + expect(summarizeRuntime({ + ...base, + type: "delegation.status", + taskId, + targetBotId: "helper", + state: "failed", + reason: "max_age_exceeded", + attemptCount: 3, + }).tone).toBe("error"); + }); }); describe("summarizeNative", () => { diff --git a/src/lib/inspector.ts b/src/lib/inspector.ts index 4645d9098..620b7333c 100644 --- a/src/lib/inspector.ts +++ b/src/lib/inspector.ts @@ -95,6 +95,11 @@ export function summarizeRuntime(e: RuntimeEvent): { summary: string; tone: Insp return { summary: `tokens in ${e.input} · out ${e.output}`, tone: "plain" }; case "runtime.error": return { summary: `${e.setup ? "setup: " : ""}${clip(oneLine(e.message))}`, tone: "error" }; + case "delegation.status": + return { + summary: `delegation ${e.state} · ${e.taskId.slice(0, 8)}${e.reason ? ` · ${e.reason}` : ""}`, + tone: e.state === "failed" ? "error" : e.state === "completed" ? "boundary" : "plain", + }; default: return { summary: (e as { type: string }).type, tone: "plain" }; } From a7956a9da4f774db20e3bf65e5a69f9f5f2da26b Mon Sep 17 00:00:00 2001 From: gus <42593099+lightcloud00@users.noreply.github.com> Date: Wed, 26 Aug 2026 13:19:47 -0400 Subject: [PATCH 2/3] fix(delegations): identify group source runs --- server/index.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/server/index.ts b/server/index.ts index 53e240139..c50d4b5ea 100644 --- a/server/index.ts +++ b/server/index.ts @@ -2042,9 +2042,10 @@ async function runGroupMemberTurn( onDispatchError?.(message); return true; } + const sourceRunId = randomUUID(); const integrations: NonNullable[0]["integrations"]> = {}; if (hop < MAX_COMMS_DEPTH && instance.adapter.capabilities.agentsMcp === true) { - integrations.agents = agentsIntegration(bot.id, group.threadId, hop); + integrations.agents = agentsIntegration(bot.id, group.threadId, hop, sourceRunId); } const selectedSkills = selectBundledSkills( serializeRoomContext(group.threadId, userName), From 2cd0bf3d06f525282e14bc02a270a31b9cb516be Mon Sep 17 00:00:00 2001 From: gus <42593099+lightcloud00@users.noreply.github.com> Date: Wed, 26 Aug 2026 20:59:46 -0400 Subject: [PATCH 3/3] fix(delegations): link failed dispatch channels --- server/delegations.ts | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/server/delegations.ts b/server/delegations.ts index a39aea703..f55566e1f 100644 --- a/server/delegations.ts +++ b/server/delegations.ts @@ -680,6 +680,15 @@ async function processOne( return; } if (outcome.state === "failed") { + try { + // A non-busy dispatch attempt still belongs in the shared channel: + // runTarget may already have mirrored its terminal failure there, + // and the source needs the channel link to make that record visible. + // Busy races return above without producing a misleading exchange. + mirrorExchange(bus, sender, target, item.message, channel, threadId); + } catch (error) { + console.error("delegations: could not mirror failed dispatch", error); + } settle(bus, item, "failed", outcome.reason ?? "dispatch_failed", clock()); appendActivity(bus, threadId, `Delegation to @${target.name} failed to start`, false); return;