diff --git a/server/comms-visibility.ts b/server/comms-visibility.ts new file mode 100644 index 00000000..ac00087f --- /dev/null +++ b/server/comms-visibility.ts @@ -0,0 +1,94 @@ +// Bot⇄bot comms visibility: channel creation, message mirroring, and +// per-thread chips. Extracted from /api/internal/ask-bot so delegations +// (delegate_bot) and any future peer flow reuse the same UX without a copy. + +import type { BotRecord, GroupRecord, Message, Store } from "./store.ts"; + +/** 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. */ +export interface CommsBus { + store: Store; + /** SSE broadcast (kind: "message" envelope). */ + broadcast: (payload: Record) => void; + /** SSE broadcast (kind: "group" envelope) for a single group. */ + broadcastGroup: (groupId: string) => void; +} + +/** Find or create the bot⇄bot channel for the pair. The channel keeps + * the pair's full exchange, lives in the sidebar like any room, and the + * user can open it to chip in. */ +export function getOrCreateChannel(store: Store, from: BotRecord, target: BotRecord): GroupRecord { + return ( + store.dmGroup(from.id, target.id) ?? + store.createGroup(`${from.name} ⇄ ${target.name}`, [from.id, target.id], true) + ); +} + +/** Mirror `from`'s outgoing message into the channel, drop chips into + * both 1:1 threads linking to the channel, and bump the channel's unread + * count. The chips are what make bot-to-bot turns observable — those + * turns cost the user tokens, and a hidden exchange is exactly the kind + * of mistake peer coordination is supposed to avoid. */ +export function mirrorExchange( + bus: CommsBus, + from: BotRecord, + target: BotRecord, + message: string, + channel: GroupRecord | undefined, +): void { + const note = (threadId: string, m: Omit) => { + const message = bus.store.appendMessage(threadId, m); + bus.broadcast({ kind: "message", threadId, message }); + return message; + }; + if (channel) { + note(channel.threadId, { + role: "bot", + kind: "text", + text: message, + from: { botId: from.id, name: from.name, color: from.color }, + }); + } + note(from.threadId, { + role: "bot", + kind: "activity", + tool: { name: `Messaged @${target.name}` }, + comm: channel + ? { groupId: channel.id, withBotId: target.id, withName: target.name, withColor: target.color } + : undefined, + }); + note(target.threadId, { + role: "bot", + kind: "activity", + tool: { name: `Message from @${from.name}` }, + comm: channel + ? { groupId: channel.id, withBotId: from.id, withName: from.name, withColor: from.color } + : undefined, + }); + if (channel) { + bus.store.patchGroup(channel.id, { unread: true }); + bus.broadcastGroup(channel.id); + } +} + +/** Mirror `target`'s reply into the channel so the channel stays the + * single authoritative record of the exchange. The 1:1 threads already + * carry their own chips from `mirrorExchange`. */ +export function mirrorReply( + bus: CommsBus, + target: BotRecord, + reply: string, + channel: GroupRecord | undefined, +): void { + if (!channel || !reply.trim()) return; + const message = bus.store.appendMessage(channel.threadId, { + role: "bot", + kind: "text", + text: reply, + from: { botId: target.id, name: target.name, color: target.color }, + }); + bus.broadcast({ kind: "message", threadId: channel.threadId, message }); + bus.store.patchGroup(channel.id, { unread: true }); + bus.broadcastGroup(channel.id); +} \ No newline at end of file diff --git a/server/comms.test.ts b/server/comms.test.ts index 07c68052..353e56a9 100644 --- a/server/comms.test.ts +++ b/server/comms.test.ts @@ -99,11 +99,23 @@ describe("comms e2e (fake ACP fleet)", () => { join(home, ".openmausbot", "config.json"), JSON.stringify({ instances: { + // the ask-peer fleet: both bots run "ask-peer" so A can ask B + // synchronously (existing ask_bot e2e + the approval-gate e2e, + // which uses the same sync path under a human card). grok: { driver: "grokAgent", environment: { FAKE_ACP_MODE: "ask-peer" }, config: { cli: FAKE_CLI, fullAuto: true }, }, + // a separate asker instance for the async-handoff e2e. B can stay + // on `grok` because its depth-1 turn runs without the agents + // integration either way (the depth guard), so it just plays + // plain happy text. + askerDelegate: { + driver: "grokAgent", + environment: { FAKE_ACP_MODE: "delegate-peer" }, + config: { cli: FAKE_CLI, fullAuto: true }, + }, }, }), ); @@ -218,4 +230,343 @@ describe("comms e2e (fake ACP fleet)", () => { }, 40_000, ); + + // ── async peer handoff (delegate_bot) ─────────────────────────────── + // A's fake CLI uses FAKE_ACP_MODE=delegate-peer, which calls delegate_bot + // (returns immediately with "Delegation queued"). After A's turn + // settles, the harness fires B's depth-1 turn — B runs plain happy text + // because the depth guard refuses to inject the agents integration at + // depth >= MAX_COMMS_DEPTH. The exchange mirrors into a bot⇄bot channel + // and shows up as chips on both 1:1 threads, just like ask_bot. + it( + "hands the task off async via delegate_bot and lets B run after A's turn settles", + async () => { + const seeded = (await api("GET", "/api/bots")).body.bots[0]; + await api("PATCH", `/api/bots/${seeded.id}`, { hidden: true }); + const helperSelection = { instanceId: "grok", model: "fake-model" }; + const askerSelection = { instanceId: "askerDelegate", model: "fake-model" }; + const helper = (await api("POST", "/api/bots")).body.bot; + await api("PATCH", `/api/bots/${helper.id}`, { name: "Helper", modelSelection: helperSelection }); + const asker = (await api("POST", "/api/bots")).body.bot; + await api("PATCH", `/api/bots/${asker.id}`, { name: "Asker", modelSelection: askerSelection }); + + const send = await api("POST", `/api/bots/${asker.id}/messages`, { text: "hey @Helper please pick this up" }); + expect(send.status).toBe(202); + + // wait for A's turn to settle: it should NOT have a "peer says:" line + // (delegate_bot doesn't return the peer's reply to A) and the channel + // chip should be the "Messaged @Helper" kind, not the ask_bot one. + const deadline = Date.now() + 30_000; + let askerBot: any; + let helperBot: any; + let note: any; + for (;;) { + const state = (await api("GET", "/api/bots")).body; + askerBot = state.bots.find((b: any) => b.id === asker.id); + helperBot = state.bots.find((b: any) => b.id === helper.id); + // A's "Delegated to @Helper: followup" chip proves the queue ran. + // We also want B to have actually run (its busy flag is false and + // it has a bot text reply). + const askerDelegated = askerBot.messages.some( + (m: any) => m.kind === "activity" && m.tool?.name === "Delegated to @Helper: followup", + ); + note = askerBot.messages.find( + (m: any) => m.kind === "activity" && m.tool?.name === "Messaged @Helper", + ); + const helperReplied = helperBot.messages.some( + (m: any) => m.role === "bot" && m.kind === "text" && m.text?.includes("hello from fake acp"), + ); + if (askerDelegated && note && helperReplied && !helperBot.busy) break; + if (Date.now() > deadline) { + throw new Error( + `delegate handoff never settled. asker busy=${askerBot.busy} helper busy=${helperBot.busy}\n` + + `asker tail: ${JSON.stringify(askerBot.messages.slice(-8))}\n` + + `helper tail: ${JSON.stringify(helperBot.messages.slice(-6))}\n` + + `stderr: ${stderr.slice(-2000)}`, + ); + } + await new Promise((r) => setTimeout(r, 250)); + } + + // A's reply is the queue-ack text the proxy returns, NOT a peer reply + const askerFinal = askerBot.messages.findLast((m: any) => m.kind === "text" && m.role === "bot"); + expect(askerFinal.text).toContain("delegated:"); + expect(askerFinal.text).not.toContain("hello from fake acp"); + + // A's comm chip links to the channel, attributed to @Helper + expect(note).toBeTruthy(); + expect(note.comm?.groupId).toBeTruthy(); + expect(note.comm?.withName).toBe("Helper"); + + // B ran a depth-1 turn: the inbound user text carries the delegation + // prefix, B's reply is the happy-mode line (no agents integration). + const helperInbound = helperBot.messages.find( + (m: any) => m.role === "user" && m.kind === "text", + ); + expect(helperInbound.text).toContain("[Delegated by @Asker"); + expect(helperInbound.text).toContain("delegated task"); + expect(helperInbound.text).toContain("[Reason: followup]"); + const helperReply = helperBot.messages.findLast( + (m: any) => m.kind === "text" && m.role === "bot", + ); + expect(helperReply.text).toContain("hello from fake acp"); + + // channel mirrors both sides, attributed by bot + const state = (await api("GET", "/api/bots")).body; + const channel = state.groups.find((g: any) => g.id === note.comm.groupId); + expect(channel?.dm).toBe(true); + expect(channel.memberIds).toContain(asker.id); + expect(channel.memberIds).toContain(helper.id); + // A's outgoing task is mirrored into the channel attributed to A. + // The peer's reply is intentionally NOT mirrored (delegate_bot is + // fire-and-forget — the user opens B's thread to see what B did). + expect( + channel.messages.some((m: any) => m.from?.botId === asker.id && m.text?.includes("delegated task")), + ).toBe(true); + + // B's receive-side chip points at the same channel + const helperNote = helperBot.messages.find( + (m: any) => m.kind === "activity" && m.tool?.name === "Message from @Asker", + ); + expect(helperNote?.comm?.groupId).toBe(note.comm.groupId); + expect(helperBot.busy).toBeFalsy(); + expect(askerBot.busy).toBeFalsy(); + }, + 45_000, + ); + + // ── approval gate (approvePeerComms) ───────────────────────────────── + // When the SOURCE bot has approvePeerComms = true, an ask_bot call must + // not run the peer turn until the user clicks Allow on a card pushed to + // the source's thread. The card carries a requestId that + // /api/bots/:id/respond resolves BEFORE forwarding to the provider + // adapter, so the provider never sees the request. + it( + "blocks ask_bot behind a card and only runs B after the user allows", + async () => { + const seeded = (await api("GET", "/api/bots")).body.bots[0]; + await api("PATCH", `/api/bots/${seeded.id}`, { hidden: true }); + const selection = { instanceId: "grok", model: "fake-model" }; + const helper = (await api("POST", "/api/bots")).body.bot; + await api("PATCH", `/api/bots/${helper.id}`, { name: "Helper", modelSelection: selection }); + const asker = (await api("POST", "/api/bots")).body.bot; + // approvePeerComms on the SOURCE bot — the test's whole point + const approval = await api("PATCH", `/api/bots/${asker.id}`, { + name: "Asker", + modelSelection: selection, + approvePeerComms: true, + }); + expect(approval.status).toBe(200); + expect(approval.body.bot.approvePeerComms).toBe(true); + + const send = await api("POST", `/api/bots/${asker.id}/messages`, { text: "hey @Helper needs approval" }); + expect(send.status).toBe(202); + + // Wait for the options card to appear on A's thread. While the card + // is open, B MUST NOT have started: no inbound user message, not busy, + // no reply text. + let askerBot: any; + let helperBot: any; + let card: any; + const cardDeadline = Date.now() + 20_000; + for (;;) { + const state = (await api("GET", "/api/bots")).body; + askerBot = state.bots.find((b: any) => b.id === asker.id); + helperBot = state.bots.find((b: any) => b.id === helper.id); + card = askerBot.messages.find( + (m: any) => m.kind === "options" && m.card?.requestId && m.card?.tool === "ask_bot", + ); + const helperStarted = + helperBot.messages.some((m: any) => m.role === "user" && m.kind === "text") || + helperBot.busy; + if (card && !helperStarted) break; + if (Date.now() > cardDeadline) { + throw new Error( + `approval card never appeared or B started anyway\n` + + `asker tail: ${JSON.stringify(askerBot.messages.slice(-8))}\n` + + `helper tail: ${JSON.stringify(helperBot.messages.slice(-6))}\n` + + `stderr: ${stderr.slice(-2000)}`, + ); + } + await new Promise((r) => setTimeout(r, 200)); + } + + // The card carries the allowKey the Always-allow flow mirrors back + // into bot.alwaysAllow, and offers Allow / Deny / Always allow. + expect(card.card.allowKey).toBe("ask_bot:@Helper"); + expect(card.card.options).toEqual(["Allow", "Deny", "Always allow"]); + expect(card.card.title).toContain("@Asker"); + expect(card.card.title).toContain("@Helper"); + + // Allow → B runs, A's reply folds the peer reply, channel mirrors both + const allow = await api( + "POST", + `/api/bots/${asker.id}/respond`, + { requestId: card.card.requestId, behavior: "allow" }, + ); + expect(allow.status).toBe(200); + + const settledDeadline = Date.now() + 25_000; + let finalAsker: any; + let finalHelper: any; + for (;;) { + const state = (await api("GET", "/api/bots")).body; + finalAsker = state.bots.find((b: any) => b.id === asker.id); + finalHelper = state.bots.find((b: any) => b.id === helper.id); + const peerFolded = finalAsker.messages.findLast( + (m: any) => m.kind === "text" && m.role === "bot", + )?.text?.includes("peer says:"); + const helperReplied = finalHelper.messages.some( + (m: any) => m.role === "bot" && m.kind === "text" && m.text?.includes("hello from fake acp"), + ); + if (peerFolded && helperReplied && !finalHelper.busy && !finalAsker.busy) break; + if (Date.now() > settledDeadline) { + throw new Error( + `B never ran after allow\n` + + `asker tail: ${JSON.stringify(finalAsker.messages.slice(-8))}\n` + + `helper tail: ${JSON.stringify(finalHelper.messages.slice(-6))}\n` + + `stderr: ${stderr.slice(-2000)}`, + ); + } + await new Promise((r) => setTimeout(r, 250)); + } + }, + 60_000, + ); + + it("refuses ask_bot with a denial chip and never starts B when the user denies", async () => { + const seeded = (await api("GET", "/api/bots")).body.bots[0]; + await api("PATCH", `/api/bots/${seeded.id}`, { hidden: true }); + const selection = { instanceId: "grok", model: "fake-model" }; + const helper = (await api("POST", "/api/bots")).body.bot; + await api("PATCH", `/api/bots/${helper.id}`, { name: "Helper", modelSelection: selection }); + const asker = (await api("POST", "/api/bots")).body.bot; + await api("PATCH", `/api/bots/${asker.id}`, { + name: "Asker", + modelSelection: selection, + approvePeerComms: true, + }); + + const send = await api("POST", `/api/bots/${asker.id}/messages`, { text: "hey @Helper do not allow" }); + expect(send.status).toBe(202); + + let askerBot: any; + let helperBot: any; + let card: any; + const cardDeadline = Date.now() + 20_000; + for (;;) { + const state = (await api("GET", "/api/bots")).body; + askerBot = state.bots.find((b: any) => b.id === asker.id); + helperBot = state.bots.find((b: any) => b.id === helper.id); + card = askerBot.messages.find( + (m: any) => m.kind === "options" && m.card?.requestId && m.card?.tool === "ask_bot", + ); + if (card) break; + if (Date.now() > cardDeadline) { + throw new Error( + `deny test: card never appeared\nasker tail: ${JSON.stringify(askerBot.messages.slice(-6))}\nstderr: ${stderr.slice(-2000)}`, + ); + } + await new Promise((r) => setTimeout(r, 200)); + } + + const deny = await api( + "POST", + `/api/bots/${asker.id}/respond`, + { requestId: card.card.requestId, behavior: "deny" }, + ); + expect(deny.status).toBe(200); + + // The harness returns {error:"denied by user"} to the agents proxy, + // which wraps it as "Couldn't reach that bot: denied by user" and + // returns it to A's agent. A's final assistant text carries that + // signal — wait for A's turn to settle with it. + const settledDeadline = Date.now() + 20_000; + for (;;) { + const state = (await api("GET", "/api/bots")).body; + askerBot = state.bots.find((b: any) => b.id === asker.id); + helperBot = state.bots.find((b: any) => b.id === helper.id); + const finalText = askerBot.messages.findLast( + (m: any) => m.role === "bot" && m.kind === "text" && m.text, + ); + if (finalText?.text?.includes("denied by user") && !askerBot.busy) break; + if (Date.now() > settledDeadline) { + throw new Error( + `deny test: denial didn't reach A\nasker tail: ${JSON.stringify(askerBot.messages.slice(-8))}\nstderr: ${stderr.slice(-2000)}`, + ); + } + await new Promise((r) => setTimeout(r, 200)); + } + + // B must not have started: no inbound user message, no depth-1 reply. + // (Every bot's thread has a seeded greeting of role:"bot" kind:"text", + // so a generic text search would always match — check for the actual + // happy-turn reply text and the inbound user text from A's exchange.) + expect(helperBot.busy).toBeFalsy(); + expect( + helperBot.messages.some((m: any) => m.role === "user" && m.kind === "text"), + ).toBe(false); + expect( + helperBot.messages.some( + (m: any) => + m.role === "bot" && m.kind === "text" && m.text?.includes("hello from fake acp"), + ), + ).toBe(false); + }, 50_000); + + // ── depth guard regression ─────────────────────────────────────────── + // A bot invoked via ask_bot or delegate_bot runs at depth=1, which equals + // MAX_COMMS_DEPTH. The depth guard in startTurn must refuse to inject + // the agents integration, so B's CLI sees no agents mcpServer and falls + // through to its plain happy text — NOT a "one hop" error from a depth-1 + // ask_bot. If the guard were removed, B's fake (also in ask-peer mode) + // would call ask_bot, the harness would refuse recursion, and B's reply + // would contain "peer error: ... one hop". The absence of that error is + // the regression signal. + it("does not inject the agents integration into a depth-1 turn", async () => { + const seeded = (await api("GET", "/api/bots")).body.bots[0]; + await api("PATCH", `/api/bots/${seeded.id}`, { hidden: true }); + // Both bots run ask-peer: A delegates to B (still in ask-peer mode), + // so the regression signal is observable when the guard is broken. + const selection = { instanceId: "grok", model: "fake-model" }; + const askerSelection = { instanceId: "askerDelegate", model: "fake-model" }; + const helper = (await api("POST", "/api/bots")).body.bot; + await api("PATCH", `/api/bots/${helper.id}`, { name: "Helper", modelSelection: selection }); + const asker = (await api("POST", "/api/bots")).body.bot; + await api("PATCH", `/api/bots/${asker.id}`, { name: "Asker", modelSelection: askerSelection }); + + const send = await api("POST", `/api/bots/${asker.id}/messages`, { text: "delegate this to @Helper please" }); + expect(send.status).toBe(202); + + // Wait for B's depth-1 turn to settle and write its reply + const deadline = Date.now() + 30_000; + let helperBot: any; + for (;;) { + const state = (await api("GET", "/api/bots")).body; + helperBot = state.bots.find((b: any) => b.id === helper.id); + // Match the actual reply text — the greeting on the seeded bot + // matches `role:"bot" kind:"text"` too, so a generic text search + // would break before B even runs. + const reply = helperBot.messages.find( + (m: any) => m.role === "bot" && m.kind === "text" && m.text?.includes("hello from fake acp"), + ); + if (reply && !helperBot.busy) break; + if (Date.now() > deadline) { + throw new Error( + `B never replied. helper tail: ${JSON.stringify(helperBot.messages.slice(-6))}\nstderr: ${stderr.slice(-2000)}`, + ); + } + await new Promise((r) => setTimeout(r, 250)); + } + + const reply = helperBot.messages.findLast( + (m: any) => m.role === "bot" && m.kind === "text" && m.text?.includes("hello from fake acp"), + ); + // If the guard were broken, B would have called ask_bot at depth=1 and + // received "message chains are limited to one hop" back from the + // harness. That error text would surface here as `peer error: ... one hop`. + expect(reply.text).toContain("hello from fake acp"); + expect(reply.text).not.toContain("one hop"); + expect(reply.text).not.toContain("peer error"); + }, 45_000); }); diff --git a/server/delegations.test.ts b/server/delegations.test.ts new file mode 100644 index 00000000..b7f6664d --- /dev/null +++ b/server/delegations.test.ts @@ -0,0 +1,310 @@ +// Async peer handoff (`delegate_bot`) — pure logic. Each test stands up a +// real Store with throwaway bots, a fake comms-bus (records broadcasts), +// and a runTarget stub that captures the would-be turn so the test can +// 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 type { CommsBus } from "./comms-visibility.ts"; +import { DATA_DIR } from "./config.ts"; +import type { ModelSelection } from "./contracts.ts"; +import { + drainDelegations, + queueDelegation, + _pendingCount, +} 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" }); + +interface BusPair { + commsBus: CommsBus; + approvalBus: { store: Store; broadcast: (payload: unknown) => void }; + broadcasts: unknown[]; + groupBroadcasts: string[]; +} + +function setupBuses(store: Store): BusPair { + const broadcasts: unknown[] = []; + const groupBroadcasts: string[] = []; + const broadcast = (payload: unknown) => { + broadcasts.push(payload); + }; + const broadcastGroup = (id: string) => { + groupBroadcasts.push(id); + }; + const commsBus: CommsBus = { store, broadcast, broadcastGroup }; + const approvalBus = { store, broadcast }; + return { commsBus, approvalBus, broadcasts, groupBroadcasts }; +} + +/** Poll until `predicate` returns a defined value or `timeout` elapses. + * drainDelegations is fire-and-forget (processOne runs as a Promise) so + * tests need to wait for its async steps to land. */ +async function waitFor(predicate: () => T | undefined, timeout = 2_000): Promise { + const deadline = Date.now() + timeout; + for (;;) { + const v = predicate(); + if (v !== undefined) return v; + if (Date.now() > deadline) throw new Error("waitFor: timed out"); + await new Promise((r) => setTimeout(r, 25)); + } +} + +describe("queueDelegation", () => { + let store: Store; + let from: BotRecord; + let target: BotRecord; + let commsBus: CommsBus; + let broadcasts: unknown[]; + + 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; + broadcasts = buses.broadcasts; + }); + + it("rejects a self-delegation without queueing", () => { + const result = queueDelegation(commsBus, from, { + toBotId: from.id, + message: "self-talk", + depth: 0, + }, 1); + expect(result).toBe("self"); + expect(_pendingCount(from.threadId)).toBe(0); + }); + + it("rejects when the source turn is already at the depth cap", () => { + const result = queueDelegation(commsBus, from, { + toBotId: target.id, + message: "next task", + depth: 1, + }, 1); + expect(result).toBe("too_deep"); + expect(_pendingCount(from.threadId)).toBe(0); + }); + + it("rejects when the target bot does not exist", () => { + const result = queueDelegation(commsBus, from, { + toBotId: "ghost", + message: "where?", + depth: 0, + }, 1); + expect(result).toBe("no_target"); + expect(_pendingCount(from.threadId)).toBe(0); + }); + + it("queues, broadcasts, and drops a 'Delegated to @Target' chip on the source thread", () => { + const result = queueDelegation(commsBus, from, { + toBotId: target.id, + message: "do this", + reason: "followup", + depth: 0, + }, 1); + expect(result).toBe("ok"); + expect(_pendingCount(from.threadId)).toBe(1); + + const chip = store + .messagesFor(from.threadId) + .find((m) => m.kind === "activity" && m.tool?.name?.startsWith("Delegated to @")); + expect(chip?.tool?.name).toBe("Delegated to @Helper: followup"); + + // The chip is also broadcast over SSE so chat clients see it without + // polling /api/bots + const broadcast = broadcasts.find( + (b) => + typeof b === "object" && + b !== null && + (b as { kind?: string }).kind === "message" && + (b as { threadId?: string }).threadId === from.threadId, + ); + expect(broadcast).toBeTruthy(); + }); +}); + +describe("drainDelegations", () => { + let store: Store; + let from: BotRecord; + let target: BotRecord; + let commsBus: CommsBus; + let approvalBus: { store: Store; broadcast: (payload: unknown) => void }; + let runTargetCalls: Array<{ toBotId: string; message: string; commsDepth: number }>; + + 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; + runTargetCalls = []; + }); + + afterEach(() => { + // Unresolved approval requests carry a 15-min timer that would otherwise + // keep vitest's event loop alive long after the suite ends. None of the + // tests above leave one — they all resolve via resolvePeerComms — but + // double-check by counting the module's pending map: tests that didn't + // resolve should be re-examined if this ever fires. + void runTargetCalls; + }); + + it("runs the target's turn via runTarget and mirrors the exchange", async () => { + queueDelegation(commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1); + drainDelegations(commsBus, approvalBus, from.threadId, (toBotId, message, commsDepth) => { + runTargetCalls.push({ toBotId, message, commsDepth }); + }); + + await waitFor(() => runTargetCalls.length === 1); + const call = runTargetCalls[0]!; + expect(call.toBotId).toBe(target.id); + expect(call.commsDepth).toBe(1); + expect(call.message).toContain("Delegated by @"); + expect(call.message).toContain("do this"); + + // Both 1:1 threads picked up their comm chips, attributed to the + // source/target bot respectively, linking to the same channel. + const fromChips = store + .messagesFor(from.threadId) + .filter((m) => m.kind === "activity" && m.tool?.name === "Messaged @Helper"); + expect(fromChips).toHaveLength(1); + const targetChips = store + .messagesFor(target.threadId) + .filter((m) => m.kind === "activity" && m.tool?.name === `Message from @${from.name}`); + expect(targetChips).toHaveLength(1); + expect(fromChips[0]?.comm?.groupId).toBe(targetChips[0]?.comm?.groupId); + }); + + it("includes the reason line in the prefixed message when one is given", async () => { + queueDelegation( + commsBus, + from, + { toBotId: target.id, message: "do this", reason: "next step", depth: 0 }, + 1, + ); + drainDelegations(commsBus, approvalBus, from.threadId, (toBotId, message, commsDepth) => { + runTargetCalls.push({ toBotId, message, commsDepth }); + }); + await waitFor(() => runTargetCalls.length === 1); + expect(runTargetCalls[0]!.message).toContain("[Reason: next step]"); + }); + + 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); + store.deleteBot(target.id); + drainDelegations(commsBus, approvalBus, from.threadId, (toBotId, message, commsDepth) => { + runTargetCalls.push({ toBotId, message, commsDepth }); + }); + const chip = await waitFor(() => + store + .messagesFor(from.threadId) + .find((m) => m.kind === "activity" && (m.tool?.name ?? "").includes("no such bot")), + ); + expect(chip.tool?.ok).toBe(false); + expect(runTargetCalls).toEqual([]); + }); + + it("skips runTarget and emits a 'is busy' 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) => { + runTargetCalls.push({ toBotId, message, commsDepth }); + }); + const chip = await waitFor(() => + store + .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(runTargetCalls).toEqual([]); + }); + + it("asks for approval when approvePeerComms is on, then runs only on allow", async () => { + store.patchBot(from.id, { approvePeerComms: true }); + queueDelegation(commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1); + drainDelegations(commsBus, approvalBus, from.threadId, (toBotId, message, commsDepth) => { + runTargetCalls.push({ toBotId, message, commsDepth }); + }); + + // the source bot's thread shows the options card BEFORE runTarget fires + const card = await waitFor(() => + store.messagesFor(from.threadId).find((m) => m.card?.requestId), + ); + expect(card.card?.title).toContain("delegate to @Helper"); + expect(card.card?.tool).toBe("delegate_bot"); + expect(card.card?.allowKey).toBe(peerAllowKey("delegate_bot", "Helper")); + expect(card.card?.options).toEqual(["Allow", "Deny", "Always allow"]); + expect(runTargetCalls).toEqual([]); + + resolvePeerComms(approvalBus, card.card!.requestId!, "allow"); + await waitFor(() => runTargetCalls.length === 1); + expect(runTargetCalls[0]!.toBotId).toBe(target.id); + expect(runTargetCalls[0]!.commsDepth).toBe(1); + }); + + it("emits a denial chip and skips runTarget when the user denies", async () => { + store.patchBot(from.id, { approvePeerComms: true }); + queueDelegation(commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1); + drainDelegations(commsBus, approvalBus, from.threadId, (toBotId, message, commsDepth) => { + runTargetCalls.push({ toBotId, message, commsDepth }); + }); + + const card = await waitFor(() => + store.messagesFor(from.threadId).find((m) => m.card?.requestId), + ); + resolvePeerComms(approvalBus, card.card!.requestId!, "deny"); + + const chip = await waitFor(() => + store + .messagesFor(from.threadId) + .find((m) => m.kind === "activity" && (m.tool?.name ?? "").includes("denied by user")), + ); + expect(chip.tool?.ok).toBe(false); + expect(runTargetCalls).toEqual([]); + }); + + it("auto-allows when alwaysAllow already covers the pair (no card pushed)", async () => { + store.patchBot(from.id, { + approvePeerComms: true, + alwaysAllow: [peerAllowKey("delegate_bot", "Helper")], + }); + queueDelegation(commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1); + drainDelegations(commsBus, approvalBus, from.threadId, (toBotId, message, commsDepth) => { + runTargetCalls.push({ toBotId, message, commsDepth }); + }); + + await waitFor(() => runTargetCalls.length === 1); + expect(runTargetCalls[0]!.commsDepth).toBe(1); + const card = store + .messagesFor(from.threadId) + .find((m) => m.card?.requestId && m.card.tool === "delegate_bot"); + expect(card).toBeUndefined(); + }); + + it("no-ops when nothing is queued for the source thread", () => { + drainDelegations(commsBus, approvalBus, from.threadId, (toBotId, message, commsDepth) => { + runTargetCalls.push({ toBotId, message, commsDepth }); + }); + expect(runTargetCalls).toEqual([]); + }); + + it("no-ops when the source thread no longer resolves to a bot", () => { + queueDelegation(commsBus, from, { toBotId: target.id, message: "do this", depth: 0 }, 1); + store.deleteBot(from.id); + drainDelegations(commsBus, approvalBus, from.threadId, (toBotId, message, commsDepth) => { + runTargetCalls.push({ toBotId, message, commsDepth }); + }); + expect(runTargetCalls).toEqual([]); + }); +}); diff --git a/server/delegations.ts b/server/delegations.ts new file mode 100644 index 00000000..1e1a9ec0 --- /dev/null +++ b/server/delegations.ts @@ -0,0 +1,171 @@ +// 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. + +import { getOrCreateChannel, mirrorExchange, type CommsBus } from "./comms-visibility.ts"; +import { requestPeerApproval, type ApprovalBus } from "./peer-approval.ts"; +import type { BotRecord } from "./store.ts"; + +export interface DelegationItem { + toBotId: string; + 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. */ + depth: number; +} + +export type QueueResult = "ok" | "no_target" | "self" | "too_deep" | "too_many"; + +/** Per source-thread queue. Persisted nowhere — a server restart drops + * delegations the same way provider permissions drop, which is honest: + * nobody can answer for an unattended bot. */ +const pendingDelegations = new Map(); + +/** 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. */ +export function queueDelegation( + bus: CommsBus, + from: BotRecord, + item: DelegationItem, + maxDepth: 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 list = pendingDelegations.get(from.threadId) ?? []; + // 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); + pendingDelegations.set(from.threadId, list); + const label = `Delegated to @${target.name}${item.reason ? `: ${item.reason}` : ""}`; + const note = bus.store.appendMessage(from.threadId, { + role: "bot", + kind: "activity", + tool: { name: label }, + }); + bus.broadcast({ kind: "message", threadId: from.threadId, message: note }); + return "ok"; +} + +/** 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) => void, +): void { + const list = pendingDelegations.get(threadId); + if (!list?.length) return; + pendingDelegations.delete(threadId); + const from = bus.store.botByThread(threadId); + if (!from) return; + for (const item of list) { + void processOne(bus, approvalBus, from, item, runTarget); + } +} + +/** 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); + const from = bus.store.botByThread(threadId); + if (!from) return; + const note = bus.store.appendMessage(from.threadId, { + role: "bot", + kind: "activity", + tool: { name: `${list.length} queued delegation${list.length > 1 ? "s" : ""} dropped — the turn did not finish`, ok: false }, + }); + bus.broadcast({ kind: "message", threadId: from.threadId, message: note }); +} + +async function processOne( + bus: CommsBus, + approvalBus: ApprovalBus, + from: BotRecord, + item: DelegationItem, + runTarget: (toBotId: string, message: string, commsDepth: number) => void, +): Promise { + const target = bus.store.bot(item.toBotId); + if (!target) { + const note = bus.store.appendMessage(from.threadId, { + role: "bot", + kind: "activity", + tool: { name: `error: delegation to ${item.toBotId} failed — no such bot`, ok: false }, + }); + bus.broadcast({ kind: "message", threadId: from.threadId, message: note }); + return; + } + if (target.busy) { + const note = bus.store.appendMessage(from.threadId, { + role: "bot", + kind: "activity", + tool: { name: `Delegation to @${target.name} canceled — @${target.name} is busy`, ok: false }, + }); + bus.broadcast({ kind: "message", threadId: from.threadId, message: note }); + return; + } + if (from.approvePeerComms) { + const verdict = await requestPeerApproval(approvalBus, from, target, item.message, "delegate_bot"); + if (verdict !== "allow") { + const note = bus.store.appendMessage(from.threadId, { + role: "bot", + kind: "activity", + tool: { name: `Delegation to @${target.name} denied by user`, ok: false }, + }); + bus.broadcast({ kind: "message", threadId: from.threadId, message: note }); + 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 sender = bus.store.bot(from.id); + if (!current || !sender) return; + if (current.busy) { + const note = bus.store.appendMessage(from.threadId, { + role: "bot", + kind: "activity", + tool: { name: `Delegation to @${current.name} canceled — @${current.name} is busy`, ok: false }, + }); + bus.broadcast({ kind: "message", threadId: from.threadId, message: note }); + return; + } + } + const channel = getOrCreateChannel(bus.store, from, target); + mirrorExchange(bus, from, target, item.message, channel); + const reasonLine = item.reason ? `\n\n[Reason: ${item.reason}]` : ""; + const prefixed = `[Delegated by @${from.name}, another bot in this OpenMausBot workspace. Do the work and reply directly.]\n\n${item.message}${reasonLine}`; + runTarget(item.toBotId, prefixed, item.depth + 1); +} + +/** Test helper: how many items remain queued for a thread. */ +export function _pendingCount(threadId: string): number { + return pendingDelegations.get(threadId)?.length ?? 0; +} \ No newline at end of file diff --git a/server/drivers/agents-proxy.test.ts b/server/drivers/agents-proxy.test.ts index 6cd201f3..cfbcf6b3 100644 --- a/server/drivers/agents-proxy.test.ts +++ b/server/drivers/agents-proxy.test.ts @@ -97,11 +97,11 @@ afterAll(async () => { }); describe("agents-proxy MCP surface", () => { - it("answers the MCP handshake and lists both tools", async () => { + it("answers the MCP handshake and lists all three tools", async () => { const init = await rpc("initialize", { protocolVersion: "2024-11-05" }); expect(init.result.serverInfo.name).toContain("agents"); const list = await rpc("tools/list"); - expect(list.result.tools.map((t: { name: string }) => t.name)).toEqual(["list_bots", "ask_bot"]); + expect(list.result.tools.map((t: { name: string }) => t.name)).toEqual(["list_bots", "ask_bot", "delegate_bot"]); }); it("list_bots renders the roster and authenticates with the shared token", async () => { diff --git a/server/drivers/agents-proxy.ts b/server/drivers/agents-proxy.ts index 50b3c724..80f0d4c7 100644 --- a/server/drivers/agents-proxy.ts +++ b/server/drivers/agents-proxy.ts @@ -1,10 +1,15 @@ // Agent-to-agent comms MCP proxy — spawned as an MCP server inside a bot's -// agent process (via the "agents" integration). Exposes two tools that let -// one bot talk to another, routed back through the harness so the harness -// stays the single owner of turns, permissions, and recursion limits: +// agent process (via the "agents" integration). Exposes three tools that +// let one bot talk to another, routed back through the harness so the +// harness stays the single owner of turns, permissions, and recursion +// limits: // -// list_bots() → the other bots in this workspace + their status -// ask_bot(bot_id, msg) → send msg to that bot, wait, return its reply +// list_bots() → the other bots in this workspace + their status +// ask_bot(bot_id, msg) → send msg to that bot, wait, return its reply +// delegate_bot(bot_id, msg, reason?) → hand the task to a peer ASYNC: returns +// immediately, the peer runs after your +// current turn finishes, the user sees +// the peer's reply as its own turn // // Speaks raw JSON-RPC 2.0 over stdio (no MCP SDK — house style, matches // computer-proxy / permission-proxy). All state comes from env, injected by @@ -40,6 +45,20 @@ const TOOLS = [ required: ["bot_id", "message"], }, }, + { + name: "delegate_bot", + description: + "Hand a task to another bot ASYNCHRONOUSLY: returns immediately and the peer runs after your current turn finishes. Use this when you want to keep working or hand off a long-running subtask without waiting. The user sees the peer's reply as its own turn; you do NOT receive the reply inline.", + inputSchema: { + type: "object", + properties: { + bot_id: { type: "string", description: "The target bot's id (from list_bots)." }, + message: { type: "string", description: "What the peer should do / answer." }, + reason: { type: "string", description: "Optional one-line reason for the delegation (shown to the user as a chip)." }, + }, + required: ["bot_id", "message"], + }, + }, ]; type Json = Record; @@ -83,6 +102,19 @@ async function callTool(name: string, args: Json): Promise<{ text: string; isErr if (r.error) return { text: `Couldn't reach that bot: ${r.error}`, isError: true }; return { text: `${r.botName ?? "Bot"} replied:\n${r.text ?? "(no reply)"}` }; } + if (name === "delegate_bot") { + const toBotId = String(args.bot_id ?? "").trim(); + const message = String(args.message ?? "").trim(); + const reason = typeof args.reason === "string" ? args.reason.trim() : ""; + if (!toBotId || !message) return { text: "delegate_bot needs bot_id and message.", isError: true }; + const body: Record = { fromBotId: BOT_ID, toBotId, message, depth: DEPTH }; + if (reason) body.reason = reason; + 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." }; + } return { text: `Unknown tool: ${name}`, isError: true }; } diff --git a/server/index.ts b/server/index.ts index 630a3d61..ace6913d 100644 --- a/server/index.ts +++ b/server/index.ts @@ -26,8 +26,11 @@ import { buildNotification, type Notification } from "./notify.ts"; import type { RuntimeEvent } from "./contracts.ts"; import { BUILT_IN_DRIVERS } from "./drivers/builtIn.ts"; +import { getOrCreateChannel, mirrorExchange, mirrorReply, type CommsBus } from "./comms-visibility.ts"; +import { discardDelegations, drainDelegations, queueDelegation, type QueueResult } from "./delegations.ts"; import { EventBus } from "./harness/bus.ts"; import { ProviderRegistry } from "./harness/registry.ts"; +import { cancelPeerApprovalsFor, dismissStalePeerCards, requestPeerApproval, resolvePeerComms, type ApprovalBus } from "./peer-approval.ts"; import { mentionedBots, roomResponders, Store, type GroupDefaultResponder, type Message } from "./store.ts"; import * as tts from "./tts/index.ts"; import { narrateTool, toUtterances } from "./tts/speech-text.ts"; @@ -459,6 +462,36 @@ bus.subscribe((event: RuntimeEvent) => { } }); +// Drain queued delegations for a source thread after its turn settles. +// Run as a separate subscriber so the drain logic stays out of the main +// fold (which has its own switch/case noise) and its approval + startTurn +// calls never have to share locals with the fold's state machine. +bus.subscribe((event: RuntimeEvent) => { + if (event.type !== "turn.completed") return; + // A turn that failed or was interrupted drops its queue rather than + // firing it later: the user who hit Stop does not expect the delegations + // that turn queued to run anyway, minutes later, on an unrelated turn. + if (!event.ok) return void discardDelegations(commsBus, event.threadId); + drainDelegations(commsBus, approvalBus, event.threadId, (toBotId, text, commsDepth) => { + // 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. + void startTurn(toBotId, text, { commsDepth }).catch((err) => { + const bot = store.bot(toBotId); + const why = err instanceof Error ? err.message : String(err); + const source = store.botByThread(event.threadId); + if (!source) return; + const note = store.appendMessage(source.threadId, { + role: "bot", + kind: "activity", + tool: { name: `error: delegation to @${bot?.name ?? toBotId} could not start — ${why.slice(0, 120)}`, ok: false }, + }); + broadcast({ kind: "message", threadId: source.threadId, message: note }); + }); + }); +}); + // ── live screen: poll the bot's box while it works ──────────────────── // Frames stream to clients as SSE {kind:'screen'} (the "Bot's screen" // panel); the final frame is folded into the transcript on turn end. @@ -852,6 +885,24 @@ function broadcastGroup(groupId: string) { if (group) broadcast({ kind: "group", group }); } +// 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, broadcastGroup }; + +// 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 +// can call resolvePeerComms without holding a reference back to here. +const approvalBus: ApprovalBus = { store, broadcast }; + +// Approvals live only in memory, so any peer card still open on disk is one +// whose resolver died with the previous process. Left alone it can never be +// answered, and the composer stays disabled behind it — settle them at boot. +{ + const stale = dismissStalePeerCards(approvalBus); + if (stale) console.log(`peer approvals: dismissed ${stale} card(s) left by a previous run`); +} + async function runGroupMemberTurn( groupId: string, botId: string, @@ -1146,63 +1197,71 @@ const server = createServer(async (req, res) => { const target = store.bot(toBotId); if (!target) return json(res, 404, { error: "no such bot" }); if (target.busy) return json(res, 200, { busy: true }); + // An unknown sender used to fall through: no mirroring AND no + // approval, while still running the peer turn. That made an + // unresolvable id the cheapest way past the gate, so it is now a + // hard refusal — every peer turn has an accountable sender. const from = store.bot(fromBotId); - const fromName = from?.name ?? "another bot"; + if (!from) return json(res, 403, { error: "unknown sender" }); + const fromName = from.name; // the exchange is mirrored into a bot⇄bot channel: it shows up in // the sidebar like any room, keeps the pair's full history, and the - // user can open it and chip in - let channel = from ? store.dmGroup(from.id, target.id) : undefined; - if (from && !channel) { - channel = store.createGroup(`${from.name} ⇄ ${target.name}`, [from.id, target.id], true); - } - const mirror = (speaker: { id: string; name: string; color: string }, text: string) => { - if (!channel || !text.trim()) return; - const msg = store.appendMessage(channel.threadId, { - role: "bot", - kind: "text", - text, - from: { botId: speaker.id, name: speaker.name, color: speaker.color }, - }); - broadcast({ kind: "message", threadId: channel.threadId, message: msg }); - }; - // both 1:1 threads get a clickable chip that opens the channel, so - // bot-to-bot turns are never invisible (they cost the user tokens) - const chip = ( - threadId: string, - label: string, - withBot: { id: string; name: string; color: string }, - ) => { - const note = store.appendMessage(threadId, { - role: "bot", - kind: "activity", - tool: { name: label }, - comm: channel - ? { groupId: channel.id, withBotId: withBot.id, withName: withBot.name, withColor: withBot.color } - : undefined, - }); - broadcast({ kind: "message", threadId, message: note }); - }; - if (from) { - mirror(from, message); - chip(from.threadId, `Messaged @${target.name}`, target); - chip(target.threadId, `Message from @${from.name}`, from); - if (channel) { - store.patchGroup(channel.id, { unread: true }); - broadcastGroup(channel.id); - } + // user can open it and chip in. Both 1:1 threads get a clickable + // chip that opens the channel, so bot-to-bot turns are never + // invisible (they cost the user tokens). + // + // per-bot approval gate: a chief-of-staff bot without this on is + // free to coordinate; one with it on must wait for a human card + // (15-min timeout → deny) before its peer turn starts. The channel + // and the chips are created only AFTER the verdict, so a denied + // contact leaves no trace of an exchange that never happened. + if (from.approvePeerComms) { + const verdict = await requestPeerApproval(approvalBus, from, target, message, "ask_bot"); + if (verdict !== "allow") return json(res, 200, { error: "denied by user" }); + // the card may have been open for minutes — re-check the target + if (store.bot(toBotId)?.busy) return json(res, 200, { busy: true }); } + const channel = getOrCreateChannel(store, from, target); + mirrorExchange(commsBus, from, target, message, channel); const prefixed = `[Message from @${fromName}, another bot in this OpenMausBot workspace. Reply to them.]\n\n${message}`; const reply = await askBotAndWait(toBotId, prefixed, depth); - if (from) { - mirror(target, reply); - if (channel) { - store.patchGroup(channel.id, { unread: true }); - broadcastGroup(channel.id); - } - } + mirrorReply(commsBus, target, reply, channel); return json(res, 200, { botName: target.name, text: reply }); } + // 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). + if (method === "POST" && path === "/api/internal/delegate-bot") { + const body = await readBody(req); + const fromBotId = String(body.fromBotId ?? ""); + 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; + if (!toBotId || !message) return json(res, 400, { error: "toBotId and message required" }); + const from = store.bot(fromBotId); + if (!from) return json(res, 404, { error: "no such bot" }); + const result = queueDelegation(commsBus, from, { toBotId, message, reason, depth }, MAX_COMMS_DEPTH); + if (result !== "ok") { + // the agent reads this string — a bare enum ("too_deep") tells it + // nothing about what to do instead + const said: Record, string> = { + self: "a bot cannot delegate to itself", + too_deep: "delegation chains are limited to one hop — do this one yourself", + no_target: "no such bot", + too_many: "too many delegations queued on this turn — finish some first", + }; + return json(res, 400, { error: said[result] }); + } + const targetName = store.bot(toBotId)?.name ?? toBotId; + return json(res, 200, { + queued: true, + message: from.approvePeerComms + ? `Queued for review — @${targetName} will only pick it up if the user approves after your turn finishes.` + : `Delegation queued — @${targetName} will pick it up after your current turn finishes.`, + }); + } return json(res, 404, { error: "unknown internal endpoint" }); } @@ -1462,13 +1521,19 @@ const server = createServer(async (req, res) => { if (body.hidden === true && existing?.chiefOfStaff && body.chiefOfStaff !== false) { return json(res, 400, { error: "choose another Chief of Staff before hiding this bot" }); } - // the two permission fields decide what runs unattended, so they are + // the permission fields decide what runs unattended, so they are // type-checked rather than copied through: a string alwaysAllow would // still answer .includes() — with substring matches, not tool names if (body.autoApprove !== undefined) { if (typeof body.autoApprove !== "boolean") return json(res, 400, { error: "autoApprove must be true or false" }); patch.autoApprove = body.autoApprove; } + if (body.approvePeerComms !== undefined) { + if (typeof body.approvePeerComms !== "boolean") { + return json(res, 400, { error: "approvePeerComms must be true or false" }); + } + patch.approvePeerComms = body.approvePeerComms; + } if (body.alwaysAllow !== undefined) { if (!Array.isArray(body.alwaysAllow) || body.alwaysAllow.some((t: unknown) => typeof t !== "string")) { return json(res, 400, { error: "alwaysAllow must be a list of tool keys" }); @@ -1498,6 +1563,10 @@ const server = createServer(async (req, res) => { stopScreenPoller(bot.id); routines!.disableForBot(bot.id); lastReply.delete(bot.threadId); + // a peer approval naming this bot can never be meaningfully answered + // now, and its caller would otherwise wait out the 15-minute timeout + cancelPeerApprovalsFor(bot.id); + discardDelegations(commsBus, bot.threadId); store.deleteBot(bot.id); for (const dir of [EVENTS_DIR, NATIVE_DIR]) { try { @@ -1588,6 +1657,12 @@ const server = createServer(async (req, res) => { const bot = store.bot(m[1]); if (!bot) return json(res, 404, { error: "no such bot" }); const body = await readBody(req); + // peer-approval intercept: harness-native cards carry a requestId + // that lives in peer-approval's pending map. Resolve them here so + // the provider adapter never sees a request it didn't raise. + if (resolvePeerComms(approvalBus, String(body.requestId), body.behavior)) { + return json(res, 200, { ok: true }); + } const instance = registry.get(bot.modelSelection.instanceId); if (!instance) return json(res, 409, { error: "provider unavailable" }); await instance.adapter.respondToRequest(bot.threadId, String(body.requestId), { @@ -1606,6 +1681,10 @@ const server = createServer(async (req, res) => { const group = store.groupByThread(threadId); const owner = group ? (group.busyBotId ? store.bot(group.busyBotId) : undefined) : store.botByThread(threadId); if (!owner) return json(res, 404, { error: "nothing is waiting on an answer in this conversation" }); + // peer-approval intercept (see /api/bots/:id/respond above). + if (resolvePeerComms(approvalBus, String(body.requestId), body.behavior)) { + return json(res, 200, { ok: true }); + } const instance = registry.get(owner.modelSelection.instanceId); if (!instance) return json(res, 409, { error: "provider unavailable" }); await instance.adapter.respondToRequest(threadId, String(body.requestId), { diff --git a/server/peer-approval.test.ts b/server/peer-approval.test.ts new file mode 100644 index 00000000..1376616b --- /dev/null +++ b/server/peer-approval.test.ts @@ -0,0 +1,112 @@ +// The approval card's LIFECYCLE, as opposed to its verdict. A card that is +// raised but never settled keeps matching the client's "unanswered" filter, +// and the composer stays disabled behind it — so a gate that works +// perfectly can still make a thread unusable. These tests pin the settle. +import { rmSync } from "node:fs"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; + +import { DATA_DIR } from "./config.ts"; +import type { ModelSelection } from "./contracts.ts"; +import { + cancelPeerApprovalsFor, + dismissStalePeerCards, + requestPeerApproval, + resolvePeerComms, + type ApprovalBus, +} from "./peer-approval.ts"; +import { Store, type BotRecord } from "./store.ts"; + +const selection = (): ModelSelection => ({ instanceId: "claude", model: "fake-model" }); + +function pendingCard(store: Store, bot: BotRecord) { + return store + .messagesFor(bot.threadId) + .find((m) => m.kind === "options" && m.card?.requestId && !m.card.answered && !m.card.dismissed); +} + +describe("peer approval card lifecycle", () => { + let store: Store; + let bus: ApprovalBus; + let from: BotRecord; + let target: BotRecord; + + beforeEach(() => { + store = new Store(selection); + from = store.patchBot(store.createBot().id, { name: "Asker", approvePeerComms: true })!; + target = store.patchBot(store.createBot().id, { name: "Helper" })!; + bus = { store, broadcast: () => {} }; + }); + + afterEach(() => { + rmSync(DATA_DIR, { recursive: true, force: true }); + }); + + it("settles the card when the user allows, so the composer unblocks", async () => { + const verdict = requestPeerApproval(bus, from, target, "ping", "ask_bot"); + const card = pendingCard(store, from); + expect(card).toBeTruthy(); + + expect(resolvePeerComms(bus, card!.card!.requestId!, "allow")).toBe(true); + expect(await verdict).toBe("allow"); + + // the card the client renders must now be answered — this is the bit + // whose absence bricked the thread + const settled = store.messagesFor(from.threadId).find((m) => m.id === card!.id); + expect(settled?.card?.answered).toBe("allow"); + expect(settled?.card?.dismissed).toBe(false); + expect(pendingCard(store, from)).toBeUndefined(); + }); + + it("settles the card on deny too", async () => { + const verdict = requestPeerApproval(bus, from, target, "ping", "delegate_bot"); + const card = pendingCard(store, from)!; + resolvePeerComms(bus, card.card!.requestId!, "deny"); + expect(await verdict).toBe("deny"); + expect(store.messagesFor(from.threadId).find((m) => m.id === card.id)?.card?.answered).toBe("deny"); + }); + + it("answers an unknown requestId as not-ours, so provider cards still route", () => { + expect(resolvePeerComms(bus, "not-a-peer-request", "allow")).toBe(false); + }); + + it("denies and settles when the bot on either side is deleted", async () => { + const verdict = requestPeerApproval(bus, from, target, "ping", "ask_bot"); + const card = pendingCard(store, from)!; + + cancelPeerApprovalsFor(target.id); + + expect(await verdict).toBe("deny"); + const settled = store.messagesFor(from.threadId).find((m) => m.id === card.id); + expect(settled?.card?.answered).toBe("deny"); + expect(settled?.card?.dismissed).toBe(true); // not the user's answer + }); + + it("dismisses cards left by a previous run, which nothing can answer", () => { + // a card on disk whose in-memory approval died with the process + const orphan = store.appendMessage(from.threadId, { + role: "bot", + kind: "options", + card: { + title: "@Asker wants to contact @Helper", + subtitle: "ping", + options: ["Allow", "Deny"], + requestId: "from-a-dead-process", + tool: "ask_bot", + }, + }); + + expect(dismissStalePeerCards(bus)).toBe(1); + const settled = store.messagesFor(from.threadId).find((m) => m.id === orphan.id); + expect(settled?.card?.dismissed).toBe(true); + // and it is idempotent — a second boot must not re-dismiss or double count + expect(dismissStalePeerCards(bus)).toBe(0); + }); + + it("leaves a live card alone at boot", async () => { + void requestPeerApproval(bus, from, target, "ping", "ask_bot"); + expect(pendingCard(store, from)).toBeTruthy(); + expect(dismissStalePeerCards(bus)).toBe(0); + expect(pendingCard(store, from)).toBeTruthy(); + cancelPeerApprovalsFor(from.id); // don't leave a timer pending + }); +}); diff --git a/server/peer-approval.ts b/server/peer-approval.ts new file mode 100644 index 00000000..55bce2ff --- /dev/null +++ b/server/peer-approval.ts @@ -0,0 +1,196 @@ +// Harness-native peer-comm approval gate. +// +// A bot with `approvePeerComms = true` may not call ask_bot or +// delegate_bot without a human approving the specific contact. The +// approval rides on the same options-card flow provider permissions +// already use: a card pushed into the SOURCE bot's thread with a +// `requestId`, answered by the user via /api/bots/:id/respond (or +// /api/threads/:id/respond) which the harness intercepts via +// `resolvePeerComms` BEFORE forwarding to the provider adapter. That +// way nothing front-end has to learn about peer comms. +// +// "Always allow" rides on the existing per-bot `alwaysAllow` list. The +// card carries an `allowKey` (`ask_bot:@Name` / `delegate_bot:@Name`) and +// the user-facing Always-allow flow already mirrors that key back into +// `alwaysAllow`, so the two sides never disagree about what was granted. + +import { newId } from "./contracts.ts"; +import type { BotRecord, Message, Store } from "./store.ts"; + +/** What a peer-approval helper needs from the outside world: the store + * for thread append + persist, and the SSE broadcaster so the chat + * updates without waiting for a refresh. */ +export interface ApprovalBus { + store: Store; + /** SSE broadcast (kind: "message" envelope). */ + broadcast: (payload: Record) => void; +} + +interface Pending { + resolve: (result: "allow" | "deny") => void; + /** Frees the requestId if the user never answers. */ + timer: ReturnType; + fromBotId: string; + toBotId: string; + message: string; + /** Where the card lives, so answering it can settle it. A card that is + * never settled keeps matching the client's "unanswered" filter, and the + * composer stays disabled behind it — the thread is unusable from then on. */ + threadId: string; + messageId: string; + bus: ApprovalBus; +} + +/** Mark the card answered so the UI stops treating it as pending. Mirrors + * what the `request.resolved` fold does for provider cards; a harness-native + * card never emits that event, so it has to settle itself. */ +function settleCard(pending: Pending, behavior: string, source: "user" | "system"): void { + const existing = pending.bus.store + .messagesFor(pending.threadId) + .find((m) => m.id === pending.messageId); + if (!existing?.card || existing.card.answered) return; + const patched = pending.bus.store.patchMessage(pending.threadId, pending.messageId, { + card: { ...existing.card, answered: behavior, dismissed: source !== "user" }, + }); + if (patched) { + pending.bus.broadcast({ kind: "message.patch", threadId: pending.threadId, message: patched }); + } +} + +/** requestId → pending ask. Lives only in memory — restarting the + * server cancels every in-flight approval, like provider permissions do. */ +const pendingComms = new Map(); + +const APPROVAL_TIMEOUT_MS = 15 * 60_000; + +/** The narrow grant "always allow" remembers for a peer comm. Mirrored + * back into `bot.alwaysAllow` when the user picks "Always allow" on the + * card. */ +export function peerAllowKey(action: "ask_bot" | "delegate_bot", targetName: string): string { + return `${action}:@${targetName}`; +} + +function allowKeyAllowed(from: BotRecord, allowKey: string): boolean { + return from.alwaysAllow?.includes(allowKey) ?? false; +} + +function pushApprovalCard( + bus: ApprovalBus, + from: BotRecord, + target: BotRecord, + message: string, + action: "ask_bot" | "delegate_bot", + requestId: string, +): Message { + const subtitle = message.length > 200 ? `${message.slice(0, 200)}…` : message; + const note = bus.store.appendMessage(from.threadId, { + role: "bot", + kind: "options", + card: { + title: `@${from.name} wants to ${action === "ask_bot" ? "contact" : "delegate to"} @${target.name}`, + subtitle, + options: ["Allow", "Deny", "Always allow"], + requestId, + tool: action, + allowKey: peerAllowKey(action, target.name), + }, + }); + bus.broadcast({ kind: "message", threadId: from.threadId, message: note }); + return note; +} + +/** Ask the user (in `from`'s thread) whether `from` may `action` `target`. + * Resolves with `"allow"` or `"deny"`. If `from.alwaysAllow` already + * covers the (action, target) pair, returns `"allow"` immediately + * without a card. */ +export function requestPeerApproval( + bus: ApprovalBus, + from: BotRecord, + target: BotRecord, + message: string, + action: "ask_bot" | "delegate_bot", +): Promise<"allow" | "deny"> { + if (allowKeyAllowed(from, peerAllowKey(action, target.name))) { + return Promise.resolve("allow"); + } + return new Promise((resolve) => { + const requestId = newId(); + // the card has to exist before the entry, so a timeout or an answer can + // always find it to settle + const card = pushApprovalCard(bus, from, target, message, action, requestId); + const timer = setTimeout(() => { + // 15 minutes without an answer → deny. Keeps an unattended bot from + // stalling its own turn forever (matches the Claude broker timeout). + const pending = pendingComms.get(requestId); + if (!pending) return; + pendingComms.delete(requestId); + settleCard(pending, "deny", "system"); + resolve("deny"); + }, APPROVAL_TIMEOUT_MS); + timer.unref?.(); // a waiting card must never hold the process open + pendingComms.set(requestId, { + resolve, + timer, + fromBotId: from.id, + toBotId: target.id, + message, + threadId: from.threadId, + messageId: card.id, + bus, + }); + }); +} + +/** Called by the respond endpoints BEFORE forwarding to the provider + * adapter. Returns true if the requestId belonged to a pending peer + * approval (and resolves it); false if it was a provider request and + * the endpoint should keep going. */ +export function resolvePeerComms( + _bus: ApprovalBus, + requestId: string, + behavior: string | undefined, +): boolean { + const pending = pendingComms.get(requestId); + if (!pending) return false; + pendingComms.delete(requestId); + clearTimeout(pending.timer); + const allow = behavior === "allow"; + settleCard(pending, allow ? "allow" : "deny", "user"); + pending.resolve(allow ? "allow" : "deny"); + return true; +} + +/** Drop every approval waiting on a bot that no longer exists (or is being + * deleted), denying it so the caller's turn doesn't wait out the timeout. */ +export function cancelPeerApprovalsFor(botId: string): void { + for (const [requestId, pending] of [...pendingComms]) { + if (pending.fromBotId !== botId && pending.toBotId !== botId) continue; + pendingComms.delete(requestId); + clearTimeout(pending.timer); + settleCard(pending, "deny", "system"); + pending.resolve("deny"); + } +} + +/** Cards left on disk by a previous run can never be answered — their + * in-memory approval died with the process. Settle them at boot so a + * crashed run doesn't leave a thread with a permanently blocked composer. */ +export function dismissStalePeerCards(bus: ApprovalBus): number { + let dismissed = 0; + for (const bot of bus.store.bots) { + for (const message of bus.store.messagesFor(bot.threadId)) { + const card = message.card; + if (!card?.requestId || card.answered || card.dismissed) continue; + if (card.tool !== "ask_bot" && card.tool !== "delegate_bot") continue; + if (pendingComms.has(card.requestId)) continue; + const patched = bus.store.patchMessage(bot.threadId, message.id, { + card: { ...card, answered: "deny", dismissed: true }, + }); + if (patched) { + bus.broadcast({ kind: "message.patch", threadId: bot.threadId, message: patched }); + dismissed += 1; + } + } + } + return dismissed; +} \ No newline at end of file diff --git a/server/store.ts b/server/store.ts index 1530e41d..9432c9a9 100644 --- a/server/store.ts +++ b/server/store.ts @@ -167,6 +167,10 @@ export interface BotRecord { /** The single workspace-wide coordinator. The store enforces that at * most one bot owns this role, even if an older/corrupt file says more. */ chiefOfStaff?: boolean; + /** Pause for human approval before this bot talks to a peer (ask_bot, + * delegate_bot). Off by default: a chief-of-staff-style bot is most + * useful when it can coordinate without nagging. */ + approvePeerComms?: boolean; busy?: boolean; createdAt: number; } diff --git a/server/testing/fake-acp-cli.ts b/server/testing/fake-acp-cli.ts index 5c734a98..e23b0ebc 100755 --- a/server/testing/fake-acp-cli.ts +++ b/server/testing/fake-acp-cli.ts @@ -11,6 +11,8 @@ // | ask-peer (spawn the injected "agents" MCP server from // session/new's mcpServers, call list_bots + ask_bot on a // peer, and reply with what the peer said — the comms e2e) +// | delegate-peer (same as ask-peer but uses delegate_bot — +// returns immediately, the peer runs after our turn) // FAKE_ACP_DUMP path to write {argv, env} as JSON, so a test can assert // argv shape (agent/stdio flags) and env hygiene // FAKE_ACP_MODELS comma-separated model ids. Enables the opencode-shaped @@ -288,6 +290,32 @@ function handle(msg: any) { }); return; } + if (mode === "delegate-peer" && agentsMcp) { + // async peer-handoff e2e: queue the delegation and return + // immediately; the harness fires the peer's depth-1 turn after our + // turn settles. We don't need the peer's reply in our text — the + // comms e2e verifies the channel mirroring on its own. + void driveMcp(agentsMcp, [ + { name: "list_bots", args: () => ({}) }, + { + name: "delegate_bot", + args: (list) => ({ + bot_id: /id: ([\w-]+)/.exec(list)?.[1] ?? "", + message: "delegated task", + reason: "followup", + }), + }, + ]) + .then((reply) => { + out({ jsonrpc: "2.0", method: "session/update", params: { update: { sessionUpdate: "agent_message_chunk", content: { text: `delegated: ${reply}` } } } }); + complete(); + }) + .catch((e) => { + out({ jsonrpc: "2.0", method: "session/update", params: { update: { sessionUpdate: "agent_message_chunk", content: { text: `delegate error: ${(e as Error).message}` } } } }); + complete(); + }); + return; + } playTurn(); if (mode === "permission") { // ask the client to approve a tool, then complete once answered diff --git a/src/components/SettingsPanel.tsx b/src/components/SettingsPanel.tsx index 09e3b36f..1d5161c1 100644 --- a/src/components/SettingsPanel.tsx +++ b/src/components/SettingsPanel.tsx @@ -49,6 +49,7 @@ export function SettingsPanel({ bot }: { bot: Bot }) { | "speakReplies" | "voice" | "chiefOfStaff" + | "approvePeerComms" > >, ) => dispatch({ type: "updateBot", botId: bot.id, patch: p }); @@ -232,6 +233,38 @@ export function SettingsPanel({ bot }: { bot: Bot }) { +
+
+
+ Ask me before contacting other bots +
+
+ {bot.approvePeerComms + ? "This bot will stop and ask before it reaches out to another bot." + : "Let this bot talk to teammates on its own, without a confirmation step."} +
+
+ +
+
Model
diff --git a/src/state/store.tsx b/src/state/store.tsx index 0181306b..55cd8ef3 100644 --- a/src/state/store.tsx +++ b/src/state/store.tsx @@ -124,6 +124,9 @@ export interface Bot { hidden?: boolean; /** The workspace's one primary coordinator. */ chiefOfStaff?: boolean; + /** When this bot wants to talk to another bot (ask_bot/delegate_bot), + * pause and ask the user first. Off by default. */ + approvePeerComms?: boolean; messages: Message[]; /** leaf of the visible conversation branch (see visibleMessages) */ activeLeafId?: string | null; @@ -312,6 +315,7 @@ type Action = | "pinned" | "hidden" | "chiefOfStaff" + | "approvePeerComms" > >; };