From 023c9274767ef50fa0f5b37ef1883336f8be59fd Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Wed, 9 Sep 2026 23:08:42 -0400 Subject: [PATCH] feat(beehive): provision owned fixed-scope Buzz reply tool Signed-off-by: Logan Johnson --- beehive/CHECKPOINT.md | 191 ++++++++++--------- beehive/README.md | 23 ++- beehive/TESTING.md | 38 +++- beehive/src/broker.ts | 35 +++- beehive/src/cli.ts | 12 +- beehive/src/conversation.ts | 9 +- beehive/src/reply-tool.ts | 136 +++++++++++++ beehive/src/supervisor.ts | 6 +- beehive/test/broker.test.ts | 10 +- beehive/test/conversation-harness-fixture.ts | 31 ++- beehive/test/conversation-runtime-fixture.ts | 4 + beehive/test/installed-conversation.test.ts | 26 ++- beehive/test/reply-tool.test.ts | 70 +++++++ 13 files changed, 461 insertions(+), 130 deletions(-) create mode 100644 beehive/src/reply-tool.ts create mode 100644 beehive/test/reply-tool.test.ts diff --git a/beehive/CHECKPOINT.md b/beehive/CHECKPOINT.md index a5205f75e5..ae7fd28c04 100644 --- a/beehive/CHECKPOINT.md +++ b/beehive/CHECKPOINT.md @@ -2,92 +2,108 @@ Worktree `/Users/loganj/.buzz/REPOS/beehive-cbfd9440`, branch `beehive/cbfd9440`. Base `051c3a270be9c73da9ab06700bcab7d5552fceaa`; this continuation starts at -`9ff9c6cf16f8c5d7cae0b95e4c01cb3820ff833e`. Candidate is the commit containing +`c4f30176765bdedd075e9634f33e4d4576ce3619`. Candidate is the commit containing this checkpoint (`git rev-parse HEAD`). No push/PR/merge/release. -## Implemented conversation boundary — not full delivery +## Implemented: actual signed reply in the isolated installed-binary path -**The host-owned TypeScript ACP broker is implemented and host Start is wired.** -`conversation-setup` still attaches to the existing persistent agent identity, -never regenerating keys or initiating OAuth. External buzz-acp receives that key, -explicit owner, separate conversation relay, optional local attestation and model. -It retains all Buzz authentication/member/mention/dispatch/reply/tool semantics. -There is no second Buzz chat runtime or remotely callable host API. +Host -> existing installed buzz-acp -> escaping childless ACP shim -> separately +host-owned ACP fixture harness -> escaping childless MCP shim -> host-resident +TypeScript MCP adapter -> separately host-owned **actual installed Buzz CLI**. +The CLI, not Beehive/test publisher code, resolves members/thread and signs/publishes +an actual reply. The isolated relay verifies canonical event hash, Schnorr signature, +persistent fixture agent pubkey, channel, parent and sole explicit owner recipient. +This is legacy kind-9 fixture evidence, NOT current production kind-40002 compatibility, +real community admission, live Databricks/provider evidence or full product completion. -`src/broker.ts` opens a fresh private temporary directory/Unix socket with a random -per-run capability. buzz-acp launches `acp-shim.ts` with that capability; the shim -is childless and only transports stdio. The actual harness is launched by the -host via existing `spawnOwned`, NOT by the escaping upstream process group. -Capabilities cannot cross runs or supply executable/argv/env/key choices. Session -workspace is fixed; nonempty MCP provisioning is rejected (see delivery gap). -Socket authentication deadlines, connection/process/session/request/frame bounds -and write-pressure failure prevent unbounded buffering. No raw diagnostics, -response text, keys, attestation or capability enter management observations. +`conversation-setup` optionally provisions a canonical installed Buzz CLI executable +and a **fixed single-thread channel/parent/owner-recipient grant**. Existing identity, +provider context and assignment are preserved. No remote executable/env/key/scope +policy is accepted. The local configuration is frozen at launch; the ACP caller must +still supply EMPTY MCP provisioning. The broker substitutes only its own fixed +stdio MCP descriptor, and rejects foreign workspace/executable/env/MCP requests. -The proxy forwards bidirectional requests/responses/notifications, including -cancellation, reverse client RPC and session/new/load/resume. Before releasing a -created/loaded/resumed session, it injects `session/set_model` and requires exact -session/model response. It rejects conflicting/rejected/unconfirmed model changes, -including model config choices, and re-verifies after config changes. Prompts -require a confirmed session; evidence hashes actual same-session response chunks -and requires uncancelled `end_turn`. Evidence identifies agent pubkey, harness and -external runtime executable hashes, session and model. This is trusted-executable -protocol evidence, NOT provider attestation or relay delivery confirmation. +`src/reply-tool.ts` exposes only `buzz_reply({content})`, during a model-confirmed +active prompt. No shell, arbitrary CLI argv, file upload or caller-selected destination. +CLI invocation is fixed `messages send --channel ... --reply-to ... --mention ... +--content -`; content goes through stdin. Mention syntax is conservatively rejected +so implicit CLI mention resolution cannot widen the sole provisioned recipient. +Host-local signer/relay/optional attestation override all context. HOME is isolated; +no owner cache/profile or ambient env inheritance. Raw CLI output is bounded and +withheld from management/tool diagnostics. MCP is newline stdio only, not HTTP/SSE. -Start waits up to 30 seconds for an actual inbound conversation and completed -response; it never manufactures a greeting on this path. Save stays selected-next; -Stop/config remain credential-neutral. Missing admission/sign-in yields a bounded -Start failure, not a running agent claim. Generic non-conversation setups still -run the explicitly labelled old anonymous ACP probe. +**Scope limitation is intentional and explicit:** the tool does not discover dynamic +per-turn authority. It only replies into the locally provisioned thread, with an +additional fail-closed check that the upstream prompt contains that channel and +`--reply-to` instruction. Text parsing cannot enlarge the local grant and is not +signed-event provenance. General automatic multi-thread routing still needs a real +request-local scope interface; do not market this one-thread provisioning UX as +Desktop parity. A prompt for another thread fails rather than silently replying there. -## Owned teardown and review regressions +## Source-grounded ownership and teardown -Stop/cancel/failed Start closes sockets (escaping childless shims exit on EOF), -then stops runtime and every harness through retained live group anchors. Shim -PID checks are absence observations only, NEVER kill authority. Failure to confirm -any group/shim exit fails closed. Existing supervisor retains ownership through -500ms TERM -> own-group KILL, including host IPC loss; numeric journal PIDs grant -no takeover/kill authority. Arbitrary executables that deliberately escape their -owned group remain outside this POSIX containment claim. +Bounded read at base source: +- `buzz-acp/src/lib.rs:5894` build_mcp_servers derives stdio executable and signer/relay/ + attestation env; `pool.rs:1732` adds git-origin hints, not authoritative reply scope. +- `buzz-acp/src/queue.rs:1383` and following render ordinary reply instructions into + prompts; the existing CLI owns actual member/thread resolution and signed publication + (`buzz-cli/src/commands/messages.rs:611` and following). +- **Two deliberate process-group escapes** preclude simply enabling dev MCP: + `buzz-agent/src/mcp.rs:758` detaches its MCP subprocess, and + `buzz-dev-mcp/src/shell.rs:677` detaches tool shell processes. Directly forwarding + dev-mcp would lose host ownership of shell groups on abrupt parent death. -D1 resistant descendants remain tested. D2 replacement host conversation path now -shares exact-authority/revision/body-admitted cancellation outside the mutation -queue; wrong agent/revision/body cannot cancel. Host close interrupts Start. D3 -coalesced completion plus malformed tail cannot commit readiness. These regressions -are exercised for both old probe and new conversation broker paths. No independent -review of this delta has occurred; prior independent frozen-9ff9c6cf work was not -used as approval of this new implementation. +Only the existing childless stdio shim can escape here. MCP adapter lives in the +host; every actual CLI invocation uses `spawnOwned`, with retained live group anchor. +Supervisor now forwards only numeric runner exit code so CLI failure is not confused +with successful publication. TERM-resistant in-group descendants are killed through +the living anchor; PID observations never grant kill authority. Stop, cancellation, +failed startup, CLI failure and MCP disconnect retain/await teardown promises. +Broker failure starts tool cleanup immediately and final Stop checks the same promise. +Connections, IDs, frames, calls, stdout/stderr bytes and deadlines are bounded. +Host IPC loss continues to use the existing anchored group cleanup. Arbitrary trusted +executables that deliberately escape groups remain outside this POSIX containment claim. -## Actual executable evidence and precise remaining gap +## Review reconciled: B1 optional-setting rejection -Automated host -> external runtime fixture -> detached/process_group(0)-style -shim -> owned ACP harness passes using persistent identity, exact model, same -conversation completion, bidirectional RPC, session loading, cancellation and -complete runtime/shim/harness/TERM-resistant-descendant absence after Stop. The -external runtime fixture models the process/ACP contract, not a Buzz relay. +Consumed independent `WORK_LOGS/BEEHIVE_DESIGN_183FFAF0/BROKER_REVIEW_C4F30176.md` +(pinned c4f30176; independent strict + 11/11 and installed replay). It found one +medium B1: optional non-model config rejection cleared model eligibility permanently. +Upstream optional effort/permission rejection intentionally falls back rather than +aborting (`pool.rs:1880–1930`, `1973–2035`). Broker now conservatively re-acknowledges +the exact session/model BEFORE forwarding the original optional-setting error. +It does not assume the error had no side effects. Rejected model-setting changes +remain fail-closed. Production broker tests cover accepted optional setting, +rejected optional setting then successful prompt, failed re-ack and rejected model +config sibling. Existing wrong-model/conflict/cancellation/malformed-tail guards remain. +New MCP code and this B1 fix have NOT received independent delta review yet. -Also exercised installed `/Applications/Buzz.app/Contents/MacOS/buzz-acp`, SHA256 -`10612d0025d1420bdd9e9afcc2e7129377da2414e75049a8b640e81d857441ea`, against an -isolated NIP-42/REST/NIP-29 fixture relay and the owned ACP fixture harness. It -signed verified AUTH with a fresh fixture agent key, discovered fixture membership, -subscribed, consumed a signed owner mention, and completed the exact-model same -conversation through the production broker. **Installed executable subscribes to -kind 9 (plus 46010/40007), not kind 40002**; the fixture follows its actual kind 9 -subscription. This is evidence for that installed binary, not current Buzz relay -compatibility/admission. Test never touches owner keys/cache/profile or providers. +## Actual executable evidence -**No signed threaded response publication was observed.** ACP text completion is -not automatic Buzz message delivery here. The current launch plan explicitly sets -`BUZZ_ACP_MCP_COMMAND=''`; no locally authorized MCP/tool executable plan is -provisioned. The broker deliberately rejects shim-supplied MCP commands/env rather -than permitting arbitrary executable injection. Tool-backed conversation delivery -therefore remains a concrete required engineering step, not just an OAuth gate. -Do not claim the whole conversation/delivery slice or owner build is complete. +Installed `/Applications/Buzz.app/Contents/MacOS/buzz-acp` SHA256: +`10612d0025d1420bdd9e9afcc2e7129377da2414e75049a8b640e81d857441ea`. +Installed `/Applications/Buzz.app/Contents/MacOS/buzz` SHA256: +`147cc2ccf276ddedc5b84d13f5399a95282066303c9dca566bb2af5a37b856c5`. +Binary digests are independent of the source base; neither binary is rebuilt/modified. -## Validation / dependencies / next action +One isolated run verified NIP-42, two subscriptions, exact-model same-session ACP +completion, and signed threaded publication: +- reply `47f814433a621434fc76c8d282f46e3d4c2311081823def7d4aabde281a9dfff` +- fixture agent `3daf86eecb494ae8db349ea57ca2551c55c02eb3b2b923d54bd653000dbf5522` +- parent `d5e3633b90efcc7d76f66a1e1f0079c6782c1287e4c7b9aae640414449f3b9be` +- channel `aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee` -Hermit Node24.15.0 / pinned pnpm11.4.0. Standalone registry-agnostic lockfile unchanged; -no new dependencies or borrowed modules. Approved install route remains: +Fixture keys/events change each run and the loopback relay is removed afterward. +This is reproducible test evidence, not an event on the production relay. Harness, +MCP shim and resistant descendant PID files are checked absent after Stop. Tests +never manufacture the expected agent reply; the fixture merely calls the real tool, +as an LLM would. No provider, OAuth or real-owner credential access occurs. + +## Validation / next action + +Hermit Node24.15.0 / pnpm11.4.0. Fresh node_modules and fresh store frozen install +fetched all seven locked packages from the approved mirror. No new dependencies, +borrowed modules, global config changes or lockfile delta. ``` . ./bin/activate-hermit @@ -98,24 +114,21 @@ npm run check BEEHIVE_REAL_BUZZ_ACP=/Applications/Buzz.app/Contents/MacOS/buzz-acp npm test ``` -Full suite with installed opt-in: **11/11 top-level tests pass**. Without opt-in: -10 pass, installed test explicitly skipped. Strict TS and `git diff --check` pass. -No repo-wide `just ci`, production service, owner OAuth or live provider call. -Commit uses configured Logan Johnson author/committer/DCO, -without persistent config/signing changes. No independent broker review yet. +Full package suite with installed opt-in: **12/12 pass** (including additional +B1/tool authority/cleanup cases within top-level tests); strict TS and diff check +pass. Exact candidate is re-run before commit. No repo-wide just ci, production +operations, owner OAuth, live provider or new independent MCP review. -**ONE next executable action:** implement a locally provisioned, fixed-authority -MCP/Buzz-tool launch plan (including service-local tool executable discovery and -owned teardown), then extend the installed isolated-relay test to require an actual -signed threaded agent reply. Do not loosen the broker to accept shim executable/env -provisioning or invent an alternate Buzz dispatcher. Request independent broker -review against this usable artifact before broadening the launch surface. +**ONE next executable step:** independently review/replay this pinned fixed-reply +MCP delta and B1 regression, including the installed isolated signed-reply test, +before expanding to automatically provisioned per-turn scopes. Still open: actual community admission/attestation and named-service-user normal -Databricks v2 OAuth/live `databricks-claude-haiku-4-5` proof; catalog auth ambiguity; -reconnect/ACKs and durable TUI intents; crash recovery/strong OS containment; -multiple hosts/agents/setups, profiles, key import/revocation, Restart/Move; remaining -Desktop harnesses/presets/providers/mesh/compute. S1 two-host identity unproven. -Selected-next != actual-run; stopped identity remains assigned. Move must never -transfer credentials/workspace/session or permit unreachable takeover. Hash/spawn -TOCTOU, hostile journals, rollback/clones and outbox pruning remain outside evidence. +Databricks v2 OAuth/live model proof; current Buzz protocol compatibility; catalog +auth ambiguity; general per-turn reply scope (not prompt parsing); reconnect/ACKs, +durable TUI intents and crash recovery; multiple hosts/agents/setups; profiles; +key import/revocation; Restart/Move; remaining Desktop harnesses/presets/providers/ +relay-mesh/compute. S1 two-host identity unproven. Selected-next != actual-run; +stopped identity remains assigned. No remote secrets, credential/workspace/session +transfer or unreachable-host takeover. Hash/spawn TOCTOU, hostile journals, +rollback/clones, outbox pruning and stronger OS containment remain outside evidence. diff --git a/beehive/README.md b/beehive/README.md index 1732941732..2201771dce 100644 --- a/beehive/README.md +++ b/beehive/README.md @@ -158,7 +158,7 @@ seam remains missing in the external protocol; do not relabel it authenticated. An empty/filtered catalog does not prevent trying an explicitly saved exact model; only same-session acknowledgement plus completed response establishes probe evidence. -## External conversation setup (broker wired; delivery incomplete) +## External conversation setup (fixed-thread reply tool; broader delivery incomplete) `node src/cli.ts conversation-setup /absolute/existing-host-directory` attaches an installed **buzz-acp** and separate conversation relay to the SAME existing @@ -179,10 +179,21 @@ Both external process-contract fixtures and the installed buzz-acp have exercise this broker. The installed-binary test uses fresh fixture identities and an isolated NIP-42/NIP-29 relay fixture, not a real community or provider. That binary subscribes to legacy kind 9; this is not current Buzz relay compatibility evidence. -**No signed threaded agent reply was observed.** Current setup disables MCP tools; -locally provisioned, fixed-authority tool launch and reply delivery remain required. -Untrusted shim-supplied MCP executables/env are rejected rather than executed. -See CHECKPOINT.md for exact evidence and next executable action. +**An actual signed threaded reply is now observed** through the installed Buzz CLI +in that isolated test, with a deterministic ACP harness (not a live model). Local +`conversation-setup` can optionally provision a canonical Buzz CLI path and one +fixed channel/parent/owner-recipient scope. The host supplies a narrow stdio MCP +`buzz_reply({content})` tool during confirmed prompts. It invokes the existing CLI, +which owns signing, membership and thread semantics. MCP/CLI argv and destinations +cannot come from relay/ACP callers; foreign provisioning is still rejected. + +This is deliberately **single-thread local provisioning**, not automatic per-turn +routing or final UX parity. Another thread fails closed. Mention syntax in content +is currently unsupported (the host supplies the sole recipient). There is no shell, +file upload, arbitrary CLI or HTTP/SSE MCP support. No server is launched by an +escaping subprocess: the MCP shim stays childless and each CLI group has its own +host-owned anchor. See CHECKPOINT.md for source trace, cleanup and signed evidence. +The new tool surface still needs independent review. Community admission/attestation and normal OAuth in the named host-service context remain explicit local/operator actions. No live Databricks request has occurred. @@ -195,7 +206,7 @@ remain explicit local/operator actions. No live Databricks request has occurred. | Host inventory | Real relay, fixed authority, freshness | multiple setups/agents, service install, reconnect/reconciliation | | Remote configuration | CAS model/workspace/default profile selection | reusable profiles, metadata, config history, setup revision pinning | | Lifecycle | fixture Start/Stop, UI-independent host | strict real launch, Restart, host-owned irreversible Move | -| Buzz Agent Databricks v2 | Local setup + TypeScript ACP probe boundary | fixed-authority tool/reply delivery, live OAuth/model proof and catalog provenance | +| Buzz Agent Databricks v2 | Local setup + TypeScript ACP probe boundary | general per-turn reply scope, live OAuth/model proof and catalog provenance | | Other Buzz Agent providers | Not implemented | Anthropic, OpenAI-compatible, Databricks legacy, OpenRouter | | Goose / Claude Code / Codex | Not implemented | harness-specific local auth and adapters/catalog/launch | | Ten presets / custom ACP | Not implemented | explicit executable/env trust, preset-specific capability grounding | diff --git a/beehive/TESTING.md b/beehive/TESTING.md index 0549054dbe..bff97adc28 100644 --- a/beehive/TESTING.md +++ b/beehive/TESTING.md @@ -46,7 +46,10 @@ strict TypeScript checking passed on Node 24.15.0 / pnpm 11.4.0. Keep separately anchored harness, same-session completion and Stop absence checks. - `broker.test.ts`: exact-model rejection/conflict, malformed coalesced completion, cancellation, reverse RPC, session/load, forbidden workspace/MCP provisioning, - and teardown including TERM-resistant descendants. + and teardown including TERM-resistant descendants. B1 now tests accepted and + rejected optional config changes, preserving the original application error only + after exact-model re-ack; failed re-ack and rejected model configs still fail. + Tool-enabled foreign MCP and wrong-thread prompts remain rejected. - `cancellation.test.ts` also exercises the conversation replacement path for admitted Stop, host close, non-authority attempts and malformed completions. - `installed-conversation.test.ts`: opt-in with @@ -55,19 +58,29 @@ strict TypeScript checking passed on Node 24.15.0 / pnpm 11.4.0. Keep fixture. Verifies actual installed executable AUTH signatures, membership, owner mention dispatch, and exact-model same-conversation ACP completion through the broker. Its actual kind-9 subscription is respected. No provider is called. - Signed threaded reply is **not** observed; tools remain disabled/unprovisioned. + An actual signed threaded reply is required through the installed sibling `buzz` + CLI and the fixed host MCP adapter. Verifies canonical hash/signature, same agent, + channel/parent and sole owner p-tag; absence after Stop includes the MCP shim. + Test code supplies the owner event and invokes the tool, never signs/publishes the + expected agent response. This does not test a live LLM deciding to invoke tools. +- `reply-tool.test.ts`: production stdio adapter under detached childless shim; + caller destination/mention expansion and inactive calls cannot spawn a CLI; + fixed argv/environment; success, cancellation, disconnect, failed startup and + CLI error all clean up the shim and any TERM-resistant in-group descendants. Tests never open existing owner keys/profiles/cache, production services or native -client resources. With installed opt-in: 11/11 pass; without: 10 pass/1 explicit +client resources. With installed opt-in: 12/12 pass; without: 11 pass/1 explicit skip. The installed fixture relay does not establish actual community admission, -current relay compatibility, provider/model attestation or working reply delivery. +current relay compatibility, provider/model attestation or general multi-thread reply delivery. ## Independent evidence and limits Prior reviews are in workspace `WORK_LOGS/BEEHIVE_DESIGN_183FFAF0/EXECUTABLE_REVIEW_{94D1DA0E,6BD85947}.md`. -This candidate integrates F1/F2/F3 and D1/D2/D3 fixes but has not received an -independent delta re-review. The generic ACP fixture and signed/encrypted WS +Prior c4f30176 broker independently passed strict/11 tests and installed replay in +`WORK_LOGS/BEEHIVE_DESIGN_183FFAF0/BROKER_REVIEW_C4F30176.md`. Its B1 optional-config +finding is fixed with production regressions here. This new fixed-reply MCP delta +and B1 fix have not yet received independent review. The generic ACP fixture and signed/encrypted WS management protocol are different seams; passing them does not prove current Buzz relay auth/member/owner semantics. @@ -75,4 +88,15 @@ No full repo `just ci`, multi-host Move, arbitrary escaped descendant containmen strong OS sandbox, hostile journal, log-full/reconnect/ACK, rollback, host SIGKILL recovery or real provider proof. Genuine ownership loss retains quarantine/lock; no stale numeric PID grants cleanup authority. See CHECKPOINT.md for exact next -executable action and the wired broker and remaining fixed-authority tool/delivery gap. +executable action and the wired broker, narrow fixed-thread tool, and remaining general per-turn scope gap. + +## Installed executable pins + +- buzz-acp: `10612d0025d1420bdd9e9afcc2e7129377da2414e75049a8b640e81d857441ea` +- buzz CLI: `147cc2ccf276ddedc5b84d13f5399a95282066303c9dca566bb2af5a37b856c5` + +Both are read-only `/Applications/Buzz.app/Contents/MacOS/` binaries. Their digests +are not a claim that they were built from the source base. The MCP adapter supports +only newline stdio, fixed content-only reply calls, and host-local single-thread +scope. Optional local attestation is forwarded using the normal CLI environment; +no live admission/delegation policy proof is claimed from the fixture relay. diff --git a/beehive/src/broker.ts b/beehive/src/broker.ts index 18691378c8..20a04db46c 100644 --- a/beehive/src/broker.ts +++ b/beehive/src/broker.ts @@ -7,6 +7,7 @@ import { fileURLToPath } from 'node:url'; import { createHash, randomBytes } from 'node:crypto'; import { spawnOwned, type OwnedProcess } from './owned.ts'; import { prepareAgent, type AgentLaunch, type Evidence } from './acp.ts'; +import { ReplyTool } from './reply-tool.ts'; import { prepareConversation, type ConversationSetup } from './conversation.ts'; type RPC = { jsonrpc: string; id?: string | number; method?: string; params?: any; result?: any; error?: unknown }; @@ -19,6 +20,7 @@ export class ConversationSession { private capability = randomBytes(32).toString('hex'); private sockets = new Set(); private harnesses: OwnedProcess[] = []; + private tools: ReplyTool[] = []; private shimPids = new Set(); private runtime?: OwnedProcess; private cancellations: (() => void)[] = []; @@ -70,6 +72,7 @@ export class ConversationSession { if (this.failed) return; this.failed = true; clearTimeout(this.timeout); this.reject(Error(reason)); for (const socket of this.sockets) socket.destroy(); + for (const tool of this.tools) void tool.stop().catch(() => {}); // Retained promise is checked by stop(). // Keep ownership handles until the caller verifies complete teardown. } /** Interrupt Start without allowing late ACP responses to commit readiness. */ @@ -91,7 +94,7 @@ export class ConversationSession { this.stopping = (async () => { await this.ready.catch(() => {}); await new Promise(resolve => this.server.close(() => resolve())); - const results = await Promise.allSettled([...this.harnesses, ...(this.runtime ? [this.runtime] : [])].map(p => p.stop())); + const results = await Promise.allSettled([...this.tools, ...this.harnesses, ...(this.runtime ? [this.runtime] : [])].map(p => p.stop())); if (results.some(r => r.status === 'rejected')) throw Error('Conversation ownership teardown failed; reconcile locally'); // These numeric IDs are absence observations ONLY, never kill authority. // A shim is childless and exits on EOF even though upstream gave it a new PGID. @@ -144,6 +147,12 @@ export class ConversationSession { }); } private proxy(socket: Socket, harness: OwnedProcess, initial: string) { + const tool = this.plan.replyTool ? new ReplyTool(join(this.directory, `tool-${this.tools.length}`), this.plan.replyTool, this.plan.workspace, { + PATH: '/usr/bin:/bin', HOME: this.prepared.plan.home, + BUZZ_PRIVATE_KEY: this.plan.env.BUZZ_PRIVATE_KEY, BUZZ_RELAY_URL: this.plan.env.BUZZ_RELAY_URL, + ...(this.plan.env.BUZZ_AUTH_TAG ? { BUZZ_AUTH_TAG: this.plan.env.BUZZ_AUTH_TAG } : {}), + }, () => this.fail('Scoped Buzz reply tool failed (diagnostics withheld)')) : undefined; + if (tool) this.tools.push(tool); const pending = new Map(); const sessions = new Set(); const modelConfigs = new Set(['model']); @@ -167,9 +176,11 @@ export class ConversationSession { if (!fromHarness && msg.method) { const p = msg.params; // Session transport does not grant a shim authority to provision tools, - // workspaces or credentials. This slice has no locally approved MCP plan. + // workspaces or credentials. Replace only empty upstream provisioning with + // the host-local fixed reply adapter; never forward a supplied command. if (['session/new', 'session/load', 'session/resume'].includes(msg.method)) { if (p?.cwd !== this.prepared.plan.workspace || !Array.isArray(p.mcpServers) || p.mcpServers.length) throw Error(); + if (tool) p.mcpServers = [tool.descriptor]; sessions.delete(p.sessionId); } if (p && ['env', 'environment', 'executable', 'command'].some(key => Object.hasOwn(p, key))) throw Error(); @@ -182,10 +193,19 @@ export class ConversationSession { if (['session/set_model', 'session/set_config_option'].includes(msg.method)) sessions.delete(p?.sessionId); if (msg.method === 'session/prompt') { if (!sessions.has(p?.sessionId) || prompts.has(p.sessionId) || msg.id === undefined) throw Error(); + if (tool) { + if (prompts.size) throw Error(); + // This is a locally pinned single-thread grant, NOT authority parsed + // out of user text. Refuse prompts lacking the expected upstream routing + // instruction; parsing cannot enlarge the host's fixed destination. + const text = JSON.stringify(p.prompt); + if (!text.includes(`--reply-to ${this.plan.replyTool!.parent}`) || !text.includes(this.plan.replyTool!.channel)) throw Error(); + } + tool?.setActive(true); prompts.set(p.sessionId, { id: msg.id, hash: createHash('sha256'), text: false, cancelled: false }); } if (msg.method === 'session/cancel') { - const prompt = prompts.get(p?.sessionId); if (prompt) prompt.cancelled = true; + const prompt = prompts.get(p?.sessionId); if (prompt) { prompt.cancelled = true; tool?.setActive(false); } } } if (fromHarness && !msg.method && msg.id !== undefined) { @@ -198,7 +218,12 @@ export class ConversationSession { const request = pending.get(msg.id); if (!request) throw Error(); pending.delete(msg.id); - if (['session/new', 'session/load', 'session/resume', 'session/set_config_option'].includes(request.method) && msg.error === undefined) { + // B1: optional-setting application errors are recoverable upstream. Do + // not assume rejection had no side effect: re-ack the exact model before + // forwarding that original error. Model-changing errors remain fail-closed. + const optionalConfig = request.method === 'session/set_config_option' && !modelConfigs.has(request.params?.configId); + if (request.method === 'session/set_config_option' && !optionalConfig && msg.error !== undefined) throw Error(); + if (['session/new', 'session/load', 'session/resume', 'session/set_config_option'].includes(request.method) && (msg.error === undefined || optionalConfig)) { const session = msg.result?.sessionId ?? request.params?.sessionId; if (!validID(session) || sessions.size >= 128 || injected.size >= 128) throw Error(); for (const option of msg.result?.configOptions ?? []) { @@ -216,7 +241,7 @@ export class ConversationSession { sessions.add(request.params.sessionId); } if (request.method === 'session/prompt') { - const session = request.params.sessionId; const prompt = prompts.get(session); prompts.delete(session); + const session = request.params.sessionId; const prompt = prompts.get(session); prompts.delete(session); tool?.setActive(false); if (msg.error === undefined && msg.result?.stopReason === 'end_turn' && prompt?.text && !prompt.cancelled) { const evidence: Evidence = { session, model: this.prepared.plan.model, executableHash: this.prepared.executableHash, responseHash: prompt.hash.digest('hex'), stopReason: 'end_turn', source: 'external-buzz-conversation', agentPublicKey: this.plan.agentPublicKey, runtimeExecutableHash: this.plan.executableHash }; // Defer commit until all lines in this batch have been validated (D3). diff --git a/beehive/src/cli.ts b/beehive/src/cli.ts index 8bd92f75d9..11cd4af6d7 100644 --- a/beehive/src/cli.ts +++ b/beehive/src/cli.ts @@ -60,12 +60,18 @@ async function main() { try { const executable = realpathSync(text(await ui.question('Absolute installed buzz-acp executable: '))); const relay = text(await ui.question('Buzz CONVERSATION relay URL (not the Beehive management relay): ')); - const conversation = { executable, relay }; + const replyTool = (await ui.question('Enable a fixed single-thread Buzz reply tool? This is not automatic per-turn routing. [yes/no]: ')) === 'yes' ? { + executable: realpathSync(text(await ui.question('Absolute installed buzz CLI executable: '))), + channel: text(await ui.question('Locally authorized reply channel UUID: ')), + parent: text(await ui.question('Locally authorized reply parent event hex ID: ')), + recipient: publicKey(setup.ownerSecret), + } : undefined; + const conversation = { executable, relay, ...(replyTool ? { replyTool } : {}) }; const plan = prepareConversation(conversation, { executable: setup.runner, args: setup.args, workspace: setup.workspace, home: text(setup.serviceHome), configDirectory: text(setup.configDirectory), databricksHost: text(setup.databricksHost), model: 'databricks-claude-haiku-4-5' }, setup.agentSecret, publicKey(setup.ownerSecret)); - if ((await ui.question('Save onto existing identity? Start remains BLOCKED pending safe ACP subtree containment. [yes/no]: ')) !== 'yes') return; + if ((await ui.question('Save onto existing identity? Start waits for an admitted conversation and local provider sign-in. [yes/no]: ')) !== 'yes') return; // One atomic replacement; keys, assignment and provider auth context unchanged. writePrivate(join(dir, 'setup.json'), { ...setup, conversation }); - console.log(`Conversation setup saved for existing agent ${plan.agentPublicKey}. No network connection, login or process started. The community operator must admit this public key and owner to the chosen relay/channels (or provision a valid owner attestation locally). Admission remains unverified. Start is blocked until host-owned ACP bridge containment is implemented.`); + console.log(`Conversation setup saved for existing agent ${plan.agentPublicKey}. No network connection, login or process started. The community operator must admit this public key and owner to the chosen relay/channels (or provision a valid owner attestation locally). Admission remains unverified. Start uses the host-owned ACP broker; fixed reply scope, when enabled, must match the incoming conversation. General automatic per-turn reply provisioning remains unsupported.`); } finally { ui.close(); } } finally { rmdirSync(lock); } } else if (command === 'auth-info') { diff --git a/beehive/src/conversation.ts b/beehive/src/conversation.ts index 2bde0a2918..5a47e728e7 100644 --- a/beehive/src/conversation.ts +++ b/beehive/src/conversation.ts @@ -2,10 +2,11 @@ import { accessSync, constants, readFileSync, realpathSync } from 'node:fs'; import { isAbsolute } from 'node:path'; import { createHash } from 'node:crypto'; import { prepareAgent, type AgentLaunch } from './acp.ts'; +import { prepareReplyTool, type ReplyToolSetup } from './reply-tool.ts'; import { publicKey } from './protocol.ts'; /** Host-local external transport configuration. Never accepted through management messages. */ -export type ConversationSetup = { executable: string; relay: string; authTag?: string }; +export type ConversationSetup = { executable: string; relay: string; authTag?: string; replyTool?: ReplyToolSetup }; /** Launch intent, not a claim of relay admission or a model-confirmed conversation. */ export type ConversationPlan = ReturnType; @@ -17,6 +18,7 @@ export type ConversationPlan = ReturnType; export function prepareConversation(input: ConversationSetup, agent: AgentLaunch, agentSecret: string, owner: string) { const prepared = prepareAgent(agent); const agentPublicKey = publicKey(agentSecret); + if (input.replyTool && input.replyTool.recipient !== owner) throw Error('This reply tool supports the explicit owner recipient only'); if (!/^[a-f0-9]{64}$/.test(owner)) throw Error('Invalid conversation owner public key'); if (!isAbsolute(input.executable) || realpathSync(input.executable) !== input.executable) throw Error('Conversation executable must be a canonical absolute path'); accessSync(input.executable, constants.X_OK); @@ -27,6 +29,7 @@ export function prepareConversation(input: ConversationSetup, agent: AgentLaunch if (prepared.plan.args.some(a => a.includes(',') || a.includes('\0') || !a.length)) throw Error('ACP arguments cannot contain commas, NUL or empty values'); if (input.authTag !== undefined && (typeof input.authTag !== 'string' || input.authTag.length > 16384 || !input.authTag.length)) throw Error('Invalid local owner attestation'); return Object.freeze({ + replyTool: input.replyTool ? prepareReplyTool(input.replyTool) : undefined, executable: input.executable, executableHash: createHash('sha256').update(readFileSync(input.executable)).digest('hex'), workspace: prepared.plan.workspace, @@ -53,6 +56,6 @@ export function prepareConversation(input: ConversationSetup, agent: AgentLaunch } /** Safe remote summary. Credentials and executable paths are never included. */ -export function conversationSummary(_setup: ConversationSetup) { - return { transport: 'external-buzz-acp', admission: 'unverified', modelEvidence: 'not-observed', launch: 'host-owned ACP broker; awaiting conversation evidence' }; +export function conversationSummary(setup: ConversationSetup) { + return { replyTool: setup.replyTool ? 'fixed-owner-thread' : 'disabled', transport: 'external-buzz-acp', admission: 'unverified', modelEvidence: 'not-observed', launch: 'host-owned ACP broker; awaiting conversation evidence' }; } diff --git a/beehive/src/reply-tool.ts b/beehive/src/reply-tool.ts new file mode 100644 index 0000000000..691d1f4634 --- /dev/null +++ b/beehive/src/reply-tool.ts @@ -0,0 +1,136 @@ +import { createServer, type Socket } from 'node:net'; +import { accessSync, constants, readFileSync, realpathSync, statSync } from 'node:fs'; +import { isAbsolute } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { randomBytes, createHash } from 'node:crypto'; +import { setTimeout as delay } from 'node:timers/promises'; +import { spawnOwned, type OwnedProcess } from './owned.ts'; + +/** Local-only, single-thread reply authority. No scope or executable comes from ACP. */ +export type ReplyToolSetup = { executable: string; channel: string; parent: string; recipient: string }; +/** Snapshot the explicitly provisioned CLI and reply destination before launching anything. */ +export function prepareReplyTool(input: ReplyToolSetup) { + if (!isAbsolute(input.executable) || realpathSync(input.executable) !== input.executable) throw Error('Reply CLI must be a canonical absolute path'); + accessSync(input.executable, constants.X_OK); + if (!statSync(input.executable).isFile()) throw Error('Reply CLI must be a regular file'); + if (!/^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/.test(input.channel) || !/^[a-f0-9]{64}$/.test(input.parent) || !/^[a-f0-9]{64}$/.test(input.recipient)) throw Error('Invalid local reply scope'); + return Object.freeze({ executable: input.executable, channel: input.channel, parent: input.parent, recipient: input.recipient, + executableHash: createHash('sha256').update(readFileSync(input.executable)).digest('hex') }); +} + +/** Fixed stdio MCP adapter for the existing Buzz CLI, not a message publisher. + * The ACP harness may detach only the childless stdio shim. Every CLI process is + * launched under a separately retained host group anchor, including on IPC loss. + */ +export class ReplyTool { + private capability = randomBytes(32).toString('hex'); + private sockets = new Set(); + private pids = new Set(); + private owned: OwnedProcess[] = []; + private stops = new Map>(); + private current?: OwnedProcess; + private closed = false; + private stopping?: Promise; + private calls = 0; + private busy = false; + private active = false; + private server = createServer(socket => this.accept(socket)); + readonly ready: Promise; + readonly descriptor; + private plan; private cwd; private env; private fail; + constructor(path: string, plan: ReturnType, cwd: string, env: NodeJS.ProcessEnv, fail: () => void) { + this.plan = plan; this.cwd = cwd; this.env = Object.freeze({ ...env }); this.fail = fail; + this.descriptor = Object.freeze({ name: 'beehive-buzz-reply', command: process.execPath, + args: [fileURLToPath(new URL('./acp-shim.ts', import.meta.url)), path, this.capability], env: [] }); + this.server.on('error', fail); + this.ready = new Promise((resolve, reject) => { this.server.once('error', reject); this.server.listen(path, resolve); }); + void this.ready.catch(fail); + } + /** Only a model-confirmed active ACP prompt can exercise the local reply grant. */ + setActive(active: boolean) { + this.active = active; + if (!active && this.current) void this.stopOwned(this.current).catch(this.fail); + } + private stopOwned(p: OwnedProcess) { + let stopped = this.stops.get(p); + if (!stopped) { stopped = p.stop(); this.stops.set(p, stopped); } + return stopped; + } + private accept(socket: Socket) { + if (this.closed || this.sockets.size >= 8) { socket.destroy(); return; } + this.sockets.add(socket); + let authenticated = false; let buffer = ''; let requests = 0; + const ids = new Set(); + const timer = setTimeout(() => socket.destroy(), 2000); + socket.setEncoding('utf8'); socket.on('error', () => socket.destroy()); + socket.on('close', () => { clearTimeout(timer); this.sockets.delete(socket); if (authenticated && !this.closed) this.fail(); }); + const send = (id: unknown, result: unknown) => { + if (!this.closed && !socket.destroyed && !socket.write(JSON.stringify({ jsonrpc: '2.0', id, result }) + '\n')) this.fail(); + }; + socket.on('data', (chunk: string) => { + buffer += chunk; + if (Buffer.byteLength(buffer) > 128 * 1024) { socket.destroy(); this.fail(); return; } + try { + let end: number; + while ((end = buffer.indexOf('\n')) >= 0) { + const m = JSON.parse(buffer.slice(0, end)); buffer = buffer.slice(end + 1); + if (!authenticated) { + if (m?.capability !== this.capability || !Number.isSafeInteger(m.pid) || m.pid <= 1 || this.pids.size >= 8) throw Error(); + authenticated = true; this.pids.add(m.pid); clearTimeout(timer); continue; + } + if (++requests > 128 || m?.jsonrpc !== '2.0' || typeof m.method !== 'string') throw Error(); + if (m.method === 'notifications/initialized' && m.id === undefined) continue; + if (!(typeof m.id === 'number' && Number.isSafeInteger(m.id)) && !(typeof m.id === 'string' && m.id.length <= 200)) throw Error(); + if (ids.has(m.id)) throw Error(); + ids.add(m.id); + if (m.method === 'initialize') send(m.id, { protocolVersion: '2024-11-05', capabilities: { tools: {} }, serverInfo: { name: 'beehive-buzz-reply', version: '0.0.1' } }); + else if (m.method === 'tools/list') send(m.id, { tools: [{ name: 'buzz_reply', description: 'Send an ordinary reply using the actual Buzz CLI to the locally authorized channel, parent and recipient. Destination is fixed by the host, not by tool arguments. Mention syntax (@, nostr URIs, npub/nprofile) is unsupported; the host supplies the sole explicit recipient.', inputSchema: { type: 'object', properties: { content: { type: 'string', minLength: 1, maxLength: 16384 } }, required: ['content'], additionalProperties: false } }] }); + else if (m.method === 'ping') send(m.id, {}); + else if (m.method === 'tools/call') { + const args = m.params?.arguments; + if (!this.active || this.busy || ++this.calls > 8 || m.params?.name !== 'buzz_reply' || !args || Object.keys(args).length !== 1 || typeof args.content !== 'string' || !args.content.trim() || Buffer.byteLength(args.content) > 16384 || /@|nostr:|npub1|nprofile1/i.test(args.content)) throw Error(); + this.busy = true; + void this.call(args.content).then(() => send(m.id, { content: [{ type: 'text', text: 'Buzz CLI accepted the scoped reply.' }] }), () => { send(m.id, { isError: true, content: [{ type: 'text', text: 'Buzz reply failed; diagnostics withheld.' }] }); this.fail(); }).finally(() => { this.busy = false; }); + } else throw Error(); + } + } catch { socket.destroy(); this.fail(); } + }); + } + private async call(content: string) { + if (this.closed || !this.active) throw Error(); + const p = spawnOwned(this.plan.executable, ['messages', 'send', '--channel', this.plan.channel, '--reply-to', this.plan.parent, '--mention', this.plan.recipient, '--content', '-'], this.cwd, this.env); + this.owned.push(p); this.current = p; + let bytes = 0; + const account = (chunk: Buffer) => { bytes += chunk.length; if (bytes > 128 * 1024) this.fail(); }; + p.child.stdout.on('data', account); p.child.stderr.on('data', account); + p.child.stdin.on('error', () => this.fail()); + // Supervisor forwards the actual runner exit code, not its own anchor exit. + const completion = new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(Error('Reply timed out')), 10_000); + p.child.on('message', (m: any) => { if (m?.type === 'runner-exit' || m?.type === 'runner-error') { clearTimeout(timer); m.type === 'runner-exit' && m.code === 0 ? resolve() : reject(Error('Reply CLI failed')); } }); + p.child.once('error', () => { clearTimeout(timer); reject(Error('Reply unavailable')); }); + }); + void completion.catch(() => {}); + await p.ready; p.child.stdin.end(content); + try { await completion; } finally { this.current = undefined; } + await this.stopOwned(p); + } + /** Retain all anchors through teardown; no recovered numeric PID is kill authority. */ + stop() { + if (this.stopping) return this.stopping; + this.closed = true; this.active = false; + for (const socket of this.sockets) socket.destroy(); + this.stopping = (async () => { + await this.ready.catch(() => {}); + await new Promise(resolve => this.server.close(() => resolve())); + const results = await Promise.allSettled(this.owned.map(p => this.stopOwned(p))); + if (results.some(r => r.status === 'rejected')) throw Error('Reply tool ownership teardown failed'); + for (let i = 0; i < 100 && this.pids.size; i++) { + for (const pid of this.pids) { try { process.kill(pid, 0); } catch (e) { if ((e as NodeJS.ErrnoException).code === 'ESRCH') this.pids.delete(pid); } } + if (this.pids.size) await delay(25); + } + if (this.pids.size) throw Error('Reply tool shim exit unconfirmed'); + })(); + return this.stopping; + } +} diff --git a/beehive/src/supervisor.ts b/beehive/src/supervisor.ts index 12cf8b0725..61c9989b9c 100644 --- a/beehive/src/supervisor.ts +++ b/beehive/src/supervisor.ts @@ -14,9 +14,9 @@ function stopGroup() { // Escalation comes from the STILL LIVING group leader, never a recovered PGID. setTimeout(() => process.kill(-process.pid, 'SIGKILL'), 500); } -function notify(type: string) { +function notify(type: string, code?: number | null) { if (!process.connected) { stopGroup(); return; } - process.send?.({ type }, (error: Error | null) => { if (error) stopGroup(); }); + process.send?.({ type, code }, (error: Error | null) => { if (error) stopGroup(); }); } process.on('message', (message: unknown) => { const m = message as { type?: string; executable?: string; args?: string[]; cwd?: string; env?: NodeJS.ProcessEnv }; @@ -28,7 +28,7 @@ process.on('message', (message: unknown) => { const runner = spawn(m.executable, m.args ?? [], { cwd: m.cwd, env: m.env, stdio: ['inherit', 'inherit', 'inherit'] }); runner.once('spawn', () => notify('spawned')); runner.once('error', () => notify('runner-error')); - runner.once('exit', () => notify('runner-exit')); + runner.once('exit', code => notify('runner-exit', code)); } }); // Parent loss tears down the known group, without interpreting a persisted PID. diff --git a/beehive/test/broker.test.ts b/beehive/test/broker.test.ts index 0511d9c4fc..72a681db39 100644 --- a/beehive/test/broker.test.ts +++ b/beehive/test/broker.test.ts @@ -8,15 +8,16 @@ import { ConversationSession } from '../src/broker.ts'; import { newKey, publicKey } from '../src/protocol.ts'; test('broker enforces model, completion health, cancellation and owned teardown across escaped shims', async () => { - for (const mode of ['ok', 'reject', 'wrong-model', 'conflict', 'bad-tail', 'delayed', 'cancel', 'foreign-workspace', 'injected-mcp']) { + for (const mode of ['ok', 'optional-ok', 'optional-reject', 'optional-reack-fails', 'model-config-reject', 'reject', 'wrong-model', 'conflict', 'bad-tail', 'delayed', 'cancel', 'foreign-workspace', 'injected-mcp', 'tool-scope-mismatch']) { const dir = realpathSync(mkdtempSync(join(tmpdir(), 'bh-broker-test-'))); + const owner = publicKey(newKey()); const executable = realpathSync(process.execPath); const runtime = join(dir, 'runtime'); writeFileSync(runtime, `#!/bin/sh\nexec '${executable}' '${resolve('test/conversation-runtime-fixture.ts')}'\n`, { mode: 0o700 }); writeFileSync(join(dir, 'mode'), mode); - const session = new ConversationSession({ executable: runtime, relay: 'ws://127.0.0.1:1' }, { + const session = new ConversationSession({ executable: runtime, relay: 'ws://127.0.0.1:1', ...(['injected-mcp', 'tool-scope-mismatch'].includes(mode) ? { replyTool: { executable, channel: 'aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee', parent: 'a'.repeat(64), recipient: owner } } : {}) }, { executable, args: [resolve('test/conversation-harness-fixture.ts')], workspace: dir, home: dir, configDirectory: dir, databricksHost: 'https://fixture.invalid', model: 'databricks-claude-haiku-4-5', - }, newKey(), publicKey(newKey()), 2000); + }, newKey(), owner, 2000); const result = session.verify().then(value => ({ value }), error => ({ error })); try { if (mode === 'delayed' || mode === 'cancel') { @@ -25,7 +26,8 @@ test('broker enforces model, completion health, cancellation and owned teardown session.cancel(); } const r = await result; - if (mode === 'ok') { + if (['ok', 'optional-ok', 'optional-reject'].includes(mode)) { + if (mode === 'optional-reject') assert.equal(readFileSync(join(dir, 'config-error-forwarded'), 'utf8'), 'yes'); assert.ok('value' in r); assert.equal(r.value.session, 'conversation-session'); for (let i = 0; i < 40 && !existsSync(join(dir, 'reverse-rpc')); i++) await delay(20); assert.equal(readFileSync(join(dir, 'reverse-rpc'), 'utf8'), 'ok'); diff --git a/beehive/test/conversation-harness-fixture.ts b/beehive/test/conversation-harness-fixture.ts index a393049586..d137207e07 100644 --- a/beehive/test/conversation-harness-fixture.ts +++ b/beehive/test/conversation-harness-fixture.ts @@ -4,6 +4,7 @@ import { readFileSync, writeFileSync } from 'node:fs'; import { spawn } from 'node:child_process'; import { publicKey } from '../src/protocol.ts'; const mode = readFileSync('mode', 'utf8'); +let tool: any; let configRejected = false; const sessionId = 'conversation-session'; let selected = ''; let prompt: number | undefined; writeFileSync('harness-identity', publicKey(process.env.BUZZ_PRIVATE_KEY!)); writeFileSync('harness-pid', String(process.pid)); @@ -16,11 +17,17 @@ for await (const line of createInterface({ input: process.stdin })) { const m = JSON.parse(line); let result: any; if (!m.method) { if (m.result?.ok) writeFileSync('reverse-rpc', 'ok'); continue; } if (m.method === 'initialize') result = { protocolVersion: 1, agentInfo: { name: 'buzz-agent' } }; - else if (m.method === 'session/new') result = { sessionId, models: { currentModelId: 'default', availableModels: [] } }; + else if (m.method === 'session/new') { tool = m.params.mcpServers[0]; result = { sessionId, models: { currentModelId: 'default', availableModels: [] } }; } else if (m.method === 'session/load') { selected = ''; result = {}; } else if (m.method === 'session/set_model') { - if (mode === 'reject') { send({ jsonrpc: '2.0', id: m.id, error: { code: -1, message: 'secret-diagnostic' } }); continue; } + if (mode === 'reject' || (mode === 'optional-reack-fails' && configRejected)) { send({ jsonrpc: '2.0', id: m.id, error: { code: -1, message: 'secret-diagnostic' } }); continue; } selected = m.params.modelId; result = { sessionId, modelId: mode === 'wrong-model' ? 'other' : selected }; + } else if (m.method === 'session/set_config_option') { + if (mode !== 'optional-ok') { + configRejected = true; + send({ jsonrpc: '2.0', id: m.id, error: { code: -32601, message: 'unsupported optional setting' } }); continue; + } + result = {}; } else if (m.method === 'session/cancel') { writeFileSync('cancel-observed', 'yes'); if (prompt) send({ jsonrpc: '2.0', id: prompt, result: { stopReason: 'cancelled' } }); continue; @@ -29,6 +36,26 @@ for await (const line of createInterface({ input: process.stdin })) { writeFileSync('prompt-started', selected); prompt = m.id; send({ jsonrpc: '2.0', id: 'reverse', method: 'client/fixture', params: {} }); if (mode === 'delayed' || mode === 'cancel') continue; + if (tool) { + // Like buzz-agent's MCP transport, only the shim deliberately escapes. + const mcp = spawn(tool.command, tool.args, { detached: true, stdio: ['pipe', 'pipe', 'pipe'], env: { PATH: '/usr/bin:/bin' } }); + writeFileSync('tool-shim-pid', String(mcp.pid)); mcp.stderr.resume(); + const lines = createInterface({ input: mcp.stdout }); + const pending = new Map void>(); + lines.on('line', line => { const r = JSON.parse(line); pending.get(r.id)?.(r); }); + let seq = 0; + const rpc = (method: string, params: any) => new Promise((resolve, reject) => { + const id = ++seq; const timer = setTimeout(() => reject(Error('MCP fixture timeout')), 8000); + pending.set(id, m => { clearTimeout(timer); pending.delete(id); m.error || m.result?.isError ? reject(Error('MCP fixture failed')) : resolve(m.result); }); + mcp.stdin.write(JSON.stringify({ jsonrpc: '2.0', id, method, params }) + '\n'); + }); + await rpc('initialize', { protocolVersion: '2024-11-05', capabilities: {}, clientInfo: { name: 'fixture', version: '1' } }); + mcp.stdin.write(JSON.stringify({ jsonrpc: '2.0', method: 'notifications/initialized' }) + '\n'); + const tools = await rpc('tools/list', {}); + if (tools.tools[0].name !== 'buzz_reply') throw Error('Missing scoped reply tool'); + await rpc('tools/call', { name: 'buzz_reply', arguments: { content: 'Private conversation fixture response' } }); + // Remain connected until host Stop; the MCP shim has no children. + } const update = { jsonrpc: '2.0', method: 'session/update', params: { sessionId, update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'Private conversation fixture response' } } } }; const reply = { jsonrpc: '2.0', id: m.id, result: { stopReason: 'end_turn' } }; process.stdout.write(JSON.stringify(update) + '\n' + JSON.stringify(reply) + '\n' + (mode === 'bad-tail' ? 'not-json\n' : '')); continue; diff --git a/beehive/test/conversation-runtime-fixture.ts b/beehive/test/conversation-runtime-fixture.ts index 6d9b3fe940..9bc7e11331 100644 --- a/beehive/test/conversation-runtime-fixture.ts +++ b/beehive/test/conversation-runtime-fixture.ts @@ -26,6 +26,10 @@ const sessionId = created.result.sessionId; if (mode === 'conflict') await request('session/set_model', { sessionId, modelId: 'wrong' }); else await request('session/set_model', { sessionId, modelId: process.env.BUZZ_ACP_MODEL }); await request('session/load', { sessionId, cwd: process.cwd(), mcpServers: [] }); +if (['optional-ok', 'optional-reject', 'optional-reack-fails', 'model-config-reject'].includes(mode)) { + const config = await request('session/set_config_option', { sessionId, configId: mode === 'model-config-reject' ? 'model' : 'mode', value: mode === 'model-config-reject' ? process.env.BUZZ_ACP_MODEL : 'default' }); + if (config.error) writeFileSync('config-error-forwarded', 'yes'); +} const prompt = request('session/prompt', { sessionId, prompt: [{ type: 'text', text: 'External runtime conversation input' }] }); if (mode === 'cancel') setTimeout(() => child.stdin.write(JSON.stringify({ jsonrpc: '2.0', method: 'session/cancel', params: { sessionId } }) + '\n'), 50); const response = await prompt; diff --git a/beehive/test/installed-conversation.test.ts b/beehive/test/installed-conversation.test.ts index f46273c570..63498db221 100644 --- a/beehive/test/installed-conversation.test.ts +++ b/beehive/test/installed-conversation.test.ts @@ -2,7 +2,7 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; import { createServer } from 'node:http'; import { WebSocketServer } from 'ws'; -import { mkdtempSync, realpathSync, writeFileSync, rmSync, existsSync } from 'node:fs'; +import { mkdtempSync, realpathSync, writeFileSync, rmSync, existsSync, readFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join, resolve } from 'node:path'; import { createHash } from 'node:crypto'; @@ -12,19 +12,22 @@ import { ConversationSession } from '../src/broker.ts'; import { newKey, publicKey } from '../src/protocol.ts'; // Explicitly opt-in installed executable; never opens an owner profile or provider. -test('installed buzz-acp authenticates and completes the same identity-bearing conversation through broker', { skip: !process.env.BEEHIVE_REAL_BUZZ_ACP }, async () => { +test('installed buzz-acp completes a signed scoped threaded reply through the actual Buzz CLI', { skip: !process.env.BEEHIVE_REAL_BUZZ_ACP }, async () => { const dir = realpathSync(mkdtempSync(join(tmpdir(), 'bh-installed-'))); writeFileSync(join(dir, 'mode'), 'ok'); const secret = newKey(); const ownerSecret = newKey(); const agent = publicKey(secret); const owner = publicKey(ownerSecret); const channel = 'aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee'; + let replyID = ''; let authenticated = false; let subscriptions = 0; let delivered = false; function observeReply(e: any) { if (![9, 40002].includes(e.kind) || e.pubkey !== agent) return; - assert.ok(schnorr.verify(e.sig, e.id, agent)); + const hash = createHash('sha256').update(JSON.stringify([0, e.pubkey, e.created_at, e.kind, e.tags, e.content])).digest('hex'); + assert.equal(e.id, hash); assert.ok(schnorr.verify(e.sig, hash, agent)); + assert.deepEqual(e.tags.filter((t: string[]) => t[0] === 'p').map((t: string[]) => t[1]), [owner]); assert.ok(e.tags.some((t: string[]) => t[0] === 'h' && t[1] === channel)); assert.ok(e.tags.some((t: string[]) => t[0] === 'e' && t[1] === inbound.id)); assert.ok(e.content.includes('Private conversation fixture response')); - delivered = true; + replyID = e.id; delivered = true; } function event(kind: number, tags: string[][], content: string, key = ownerSecret) { const pubkey = publicKey(key); const created_at = Math.floor(Date.now() / 1000); @@ -39,7 +42,7 @@ test('installed buzz-acp authenticates and completes the same identity-bearing c res.setHeader('Content-Type', 'application/json'); if (req.url?.includes('/query')) { const filters = JSON.parse(body); const serialized = JSON.stringify(filters); - res.end(JSON.stringify(serialized.includes('39002') ? [members] : serialized.includes('39000') ? [metadata] : [])); + res.end(JSON.stringify(serialized.includes('39002') ? [members] : serialized.includes('39000') ? [metadata] : serialized.includes(inbound.id) ? [inbound] : [])); } else if (req.url?.includes('/events')) { const e = JSON.parse(body); observeReply(e); res.end(JSON.stringify({ id: e.id, accepted: true })); @@ -66,7 +69,7 @@ test('installed buzz-acp authenticates and completes the same identity-bearing c }); await new Promise(resolve => http.listen(0, '127.0.0.1', resolve)); const address = http.address(); assert.ok(address && typeof address !== 'string'); - const session = new ConversationSession({ executable: realpathSync(process.env.BEEHIVE_REAL_BUZZ_ACP!), relay: `ws://127.0.0.1:${address.port}` }, { + const session = new ConversationSession({ executable: realpathSync(process.env.BEEHIVE_REAL_BUZZ_ACP!), relay: `ws://127.0.0.1:${address.port}`, replyTool: { executable: realpathSync(join(process.env.BEEHIVE_REAL_BUZZ_ACP!, '..', 'buzz')), channel, parent: inbound.id, recipient: owner } }, { executable: realpathSync(process.execPath), args: [resolve('test/conversation-harness-fixture.ts')], workspace: dir, home: dir, configDirectory: dir, databricksHost: 'https://fixture.invalid', model: 'databricks-claude-haiku-4-5', }, secret, owner, 10_000); @@ -77,9 +80,16 @@ test('installed buzz-acp authenticates and completes the same identity-bearing c assert.ok(subscriptions > 0, 'installed executable must enter subscription loop'); const evidence = await session.verify(); assert.equal(evidence.session, 'conversation-session'); - console.log(`isolated installed runtime: NIP-42 verified, subscriptions=${subscriptions}, harnessSpawned=${existsSync(join(dir, 'harness-pid'))}, same-session completed=true, signed threaded reply observed=${delivered}`); + assert.ok(delivered, 'the actual Buzz CLI must publish a signed threaded agent reply, not just return tool JSON'); + console.log(`isolated installed runtime: NIP-42 verified, subscriptions=${subscriptions}, harnessSpawned=${existsSync(join(dir, 'harness-pid'))}, same-session completed=true, signed threaded reply observed=${delivered}, reply=${replyID}, agent=${agent}, parent=${inbound.id}`); } finally { - await session.stop(); for (const socket of ws.clients) socket.terminate(); + await session.stop(); + for (const file of ['harness-pid', 'descendant-pid', 'tool-shim-pid']) { + if (!existsSync(join(dir, file))) continue; + const pid = Number(readFileSync(join(dir, file), 'utf8')); + assert.throws(() => process.kill(pid, 0), (e: any) => e.code === 'ESRCH', `${file} must be absent after owned Stop`); + } + for (const socket of ws.clients) socket.terminate(); await new Promise(resolve => ws.close(() => resolve())); await new Promise(resolve => http.close(() => resolve())); rmSync(dir, { recursive: true, force: true }); diff --git a/beehive/test/reply-tool.test.ts b/beehive/test/reply-tool.test.ts new file mode 100644 index 0000000000..d244279684 --- /dev/null +++ b/beehive/test/reply-tool.test.ts @@ -0,0 +1,70 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { spawn } from 'node:child_process'; +import { createInterface } from 'node:readline'; +import { mkdtempSync, realpathSync, writeFileSync, chmodSync, readFileSync, existsSync, rmSync } from 'node:fs'; +import { join } from 'node:path'; +import { tmpdir } from 'node:os'; +import { setTimeout as delay } from 'node:timers/promises'; +import { ReplyTool, prepareReplyTool } from '../src/reply-tool.ts'; + +const scope = { executable: realpathSync(process.execPath), channel: 'aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee', parent: 'a'.repeat(64), recipient: 'b'.repeat(64) }; +const absent = (pid: number) => { try { process.kill(pid, 0); return false; } catch (e) { return (e as NodeJS.ErrnoException).code === 'ESRCH'; } }; + +test('fixed reply MCP rejects caller authority and owns CLI descendants on success, cancel, disconnect and failure', async () => { + for (const mode of ['success', 'cancel', 'disconnect', 'failure', 'startup', 'destination', 'mention', 'inactive']) { + const dir = realpathSync(mkdtempSync(join(tmpdir(), 'bh-tool-'))); + // Node executes this extensionless JS-subset TypeScript fixture as its fixed + // "messages" program. Production passes the very same argv to real Buzz CLI. + writeFileSync(join(dir, 'messages'), ` + const fs = require('node:fs'); const {spawn} = require('node:child_process'); + fs.writeFileSync('argv', JSON.stringify(process.argv.slice(2))); + fs.writeFileSync('identity', process.env.BUZZ_PRIVATE_KEY); + const child = spawn(process.execPath, ['-e', "process.on('SIGTERM',()=>{});setTimeout(()=>{},15000)"], {stdio:'ignore'}); + fs.writeFileSync('descendant', String(child.pid)); + process.stdin.resume(); + process.stdin.on('end', () => { ${mode === 'success' ? 'setTimeout(()=>process.exit(0),100)' : mode === 'failure' ? 'process.exit(2)' : "process.on('SIGTERM',()=>{});setTimeout(()=>process.exit(0),15000)"} }); + `); + let plan = prepareReplyTool(scope); + if (mode === 'startup') { + const executable = join(dir, 'unavailable'); writeFileSync(executable, '', { mode: 0o700 }); + plan = prepareReplyTool({ ...scope, executable }); chmodSync(executable, 0o600); + } + let failed = false; + const tool = new ReplyTool(join(dir, 'mcp'), plan, dir, { PATH: '/usr/bin:/bin', BUZZ_PRIVATE_KEY: 'fixture-only' }, () => { failed = true; void tool.stop().catch(() => {}); }); + await tool.ready; + const shim = spawn(tool.descriptor.command, tool.descriptor.args, { detached: true, stdio: ['pipe', 'pipe', 'pipe'], env: { PATH: '/usr/bin:/bin' } }); + shim.stderr.resume(); + const responses = new Map void>(); + createInterface({ input: shim.stdout }).on('line', line => { const m = JSON.parse(line); responses.get(m.id)?.(m.result); }); + let sequence = 0; + const rpc = (method: string, params: any) => new Promise((resolve, reject) => { + const id = ++sequence; const timer = setTimeout(() => reject(Error('Fixture RPC timeout')), 3000); + responses.set(id, r => { clearTimeout(timer); responses.delete(id); resolve(r); }); + shim.stdin.write(JSON.stringify({ jsonrpc: '2.0', id, method, params }) + '\n'); + }); + try { + assert.equal((await rpc('initialize', {})).serverInfo.name, 'beehive-buzz-reply'); + assert.equal((await rpc('tools/list', {})).tools.length, 1); + tool.setActive(mode !== 'inactive'); + const call = rpc('tools/call', { name: 'buzz_reply', arguments: { content: mode === 'mention' ? '@other' : 'fixture', ...(mode === 'destination' ? { channel: 'arbitrary' } : {}) } }); + void call.catch(() => {}); + if (['destination', 'mention', 'inactive', 'startup'].includes(mode)) { + for (let i = 0; i < 100 && !failed; i++) await delay(10); + assert.ok(failed); assert.ok(!existsSync(join(dir, 'argv'))); + } else { + for (let i = 0; i < 100 && !existsSync(join(dir, 'descendant')); i++) await delay(10); + assert.ok(existsSync(join(dir, 'descendant'))); + if (mode === 'success') assert.ok(!(await call).isError); + else if (mode === 'cancel') tool.setActive(false); + else if (mode === 'disconnect') shim.stdin.end(); + else { for (let i = 0; i < 100 && !failed; i++) await delay(10); assert.ok(failed); } + assert.deepEqual(JSON.parse(readFileSync(join(dir, 'argv'), 'utf8')), ['send', '--channel', scope.channel, '--reply-to', scope.parent, '--mention', scope.recipient, '--content', '-']); + assert.equal(readFileSync(join(dir, 'identity'), 'utf8'), 'fixture-only'); + } + await tool.stop(); + assert.ok(absent(shim.pid!)); + if (existsSync(join(dir, 'descendant'))) assert.ok(absent(Number(readFileSync(join(dir, 'descendant'), 'utf8')))); + } finally { await tool.stop(); rmSync(dir, { recursive: true, force: true }); } + } +});