mirror of
https://github.com/block/buzz.git
synced 2026-11-06 01:11:33 +00:00
feat(beehive): provision owned fixed-scope Buzz reply tool
Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
Logan Johnson
@@ -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 <loganj@squareup.com> 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.
|
||||
|
||||
@@ -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 |
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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<Socket>();
|
||||
private harnesses: OwnedProcess[] = [];
|
||||
private tools: ReplyTool[] = [];
|
||||
private shimPids = new Set<number>();
|
||||
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<void>(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<string | number, Pending>();
|
||||
const sessions = new Set<string>();
|
||||
const modelConfigs = new Set<string>(['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).
|
||||
|
||||
@@ -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') {
|
||||
|
||||
@@ -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<typeof prepareConversation>;
|
||||
|
||||
@@ -17,6 +18,7 @@ export type ConversationPlan = ReturnType<typeof prepareConversation>;
|
||||
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' };
|
||||
}
|
||||
|
||||
136
beehive/src/reply-tool.ts
Normal file
136
beehive/src/reply-tool.ts
Normal file
@@ -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<Socket>();
|
||||
private pids = new Set<number>();
|
||||
private owned: OwnedProcess[] = [];
|
||||
private stops = new Map<OwnedProcess, Promise<void>>();
|
||||
private current?: OwnedProcess;
|
||||
private closed = false;
|
||||
private stopping?: Promise<void>;
|
||||
private calls = 0;
|
||||
private busy = false;
|
||||
private active = false;
|
||||
private server = createServer(socket => this.accept(socket));
|
||||
readonly ready: Promise<void>;
|
||||
readonly descriptor;
|
||||
private plan; private cwd; private env; private fail;
|
||||
constructor(path: string, plan: ReturnType<typeof prepareReplyTool>, 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<void>((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<string | number>();
|
||||
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<void>((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<void>(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;
|
||||
}
|
||||
}
|
||||
@@ -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.
|
||||
|
||||
@@ -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');
|
||||
|
||||
@@ -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<number, (m: any) => 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<any>((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;
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<void>(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<void>(resolve => ws.close(() => resolve()));
|
||||
await new Promise<void>(resolve => http.close(() => resolve()));
|
||||
rmSync(dir, { recursive: true, force: true });
|
||||
|
||||
70
beehive/test/reply-tool.test.ts
Normal file
70
beehive/test/reply-tool.test.ts
Normal file
@@ -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<number, (r: any) => 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<any>((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 }); }
|
||||
}
|
||||
});
|
||||
Reference in New Issue
Block a user