mirror of
https://github.com/block/buzz.git
synced 2026-10-28 21:11:48 +00:00
feat(beehive): broker identity-bearing ACP conversations under host ownership
Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
Logan Johnson
@@ -1,91 +1,93 @@
|
||||
# Beehive executable checkpoint
|
||||
|
||||
Worktree `/Users/loganj/.buzz/REPOS/beehive-cbfd9440`, branch `beehive/cbfd9440`.
|
||||
Base `051c3a270be9c73da9ab06700bcab7d5552fceaa`; previous reviewed milestone
|
||||
`6bd859479503832a331cb9f8eb0a495082c8f9d8`. This continuation is the commit
|
||||
containing this checkpoint (`git rev-parse HEAD`). No push/PR/merge/release.
|
||||
Base `051c3a270be9c73da9ab06700bcab7d5552fceaa`; this continuation starts at
|
||||
`9ff9c6cf16f8c5d7cae0b95e4c01cb3820ff833e`. Candidate is the commit containing
|
||||
this checkpoint (`git rev-parse HEAD`). No push/PR/merge/release.
|
||||
|
||||
## Outcome and exact remaining critical gap
|
||||
## Implemented conversation boundary — not full delivery
|
||||
|
||||
**The requested identity-bearing relay conversation is NOT delivered.** The
|
||||
existing executable path is still a retained anonymous ACP greeting probe.
|
||||
Do not call its evidence a Buzz conversation, provider attestation or live-model
|
||||
proof. New `conversation-setup <existing-host-dir>` provisions the external
|
||||
buzz-acp contract onto the SAME existing key, atomically under the host exclusion
|
||||
lock. It does not create a key, initiate auth or connect to any conversation relay.
|
||||
Configured conversation Start fails BEFORE spawning or committing actual-run state.
|
||||
Inventory and CLI explicitly say blocked; Save/Stop remain credential-neutral.
|
||||
**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.
|
||||
|
||||
`src/conversation.ts` validates and snapshots the external executable, selected
|
||||
Buzz Agent launch, conversation relay, exact existing agent secret, owner pubkey,
|
||||
optional host-local opaque NIP-OA tag, exact model, owner-only inbound gate and
|
||||
local default permission mode. Env is allowlisted; identity and relay are
|
||||
explicitly authoritative. No owner secret, arbitrary env or remote paths/argv are
|
||||
accepted. Upstream comma-delimited args reject lossy comma/NUL/empty elements.
|
||||
Remote summary contains no credential/executable path. This is a tested launch
|
||||
PLAN, not a launched identity-bearing session. CLI currently asks no attestation;
|
||||
community operator must admit the public key and owner to intended channels, or
|
||||
locally provision a valid NIP-OA tag. Neither that action nor live OAuth occurred.
|
||||
`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.
|
||||
|
||||
## Source-grounded runtime decision (not a second agent runtime)
|
||||
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.
|
||||
|
||||
Pinned upstream `051c3a2`:
|
||||
- `desktop/src-tauri/src/managed_agents/runtime.rs::spawn_agent_child` passes
|
||||
BUZZ_PRIVATE_KEY, BUZZ_RELAY_URL, BUZZ_AUTH_TAG, external agent command/args,
|
||||
model/provider/effort. No BUZZ_ACP_REQUIRED_MODEL contract exists here.
|
||||
- `crates/buzz-acp/src/config.rs::CliArgs` defines these external env/CLI inputs.
|
||||
- `crates/buzz-acp/src/lib.rs::resolve_agent_owner` verifies NIP-OA against the
|
||||
agent pubkey or falls back to explicit owner. Startup uses HarnessRelay::connect
|
||||
with that key/auth tag, then the actual inbound author/member gate.
|
||||
- `crates/buzz-acp/src/pool.rs::apply_model_switch` treats application rejection
|
||||
as Rejected and proceeds with the default. Model env alone is insufficient.
|
||||
- **Critical containment seam:** `crates/buzz-acp/src/acp.rs::AcpClient::spawn`
|
||||
calls `cmd.process_group(0)` on Unix. Replacing the probe with buzz-acp under
|
||||
the current outer anchor would let the ACP subtree escape owned Stop (F1).
|
||||
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.
|
||||
|
||||
Reuse existing buzz-acp for all Nostr/owner/membership/mention/reply/delivery
|
||||
semantics. Do not invent a chat protocol or second Buzz dispatcher. Smallest
|
||||
coherent next implementation: host-owned TypeScript ACP stdio broker/proxy so the
|
||||
actual adapter is spawned under a host-retained live anchor, not as an escaping
|
||||
buzz-acp descendant; the buzz-acp-launched TS shim only forwards stdio. Bound and
|
||||
close every shim/broker session on Stop, retain adapter ownership after parent
|
||||
exit, enforce session/model acknowledgement, and capture prompt completion on
|
||||
THAT identity-bearing conversation session. This broker is NOT implemented yet.
|
||||
The current guard deliberately prevents unsafe direct launch in the meantime.
|
||||
## Owned teardown and review regressions
|
||||
|
||||
Management transport remains the dedicated loopback TypeScript ciphertext WS
|
||||
protocol, NOT a Nostr/Buzz conversation relay. No existing clients or Rust changed.
|
||||
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.
|
||||
|
||||
## Independent review integrated
|
||||
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.
|
||||
|
||||
Read `WORK_LOGS/BEEHIVE_DESIGN_183FFAF0/EXECUTABLE_REVIEW_6BD85947.md`.
|
||||
Earlier F1/F2/F3 remain covered. This continuation addresses its D1–D3:
|
||||
- D1: live supervisor ignores its own TERM, remains through a 500 ms graceful
|
||||
interval, then sends KILL to its OWN still-owned group. Host only polls absence;
|
||||
transient EPERM is not success and is retried within the same bounded wait.
|
||||
IPC disconnect/send failure invokes the same cleanup rather than killing the
|
||||
anchor prematurely. Resistant orphan descendants covered through remote Stop,
|
||||
host close and parent IPC loss. Genuine anchor loss still fails closed.
|
||||
- D2: exact-authority/current-revision/body-valid Stop signals ACP cancellation
|
||||
outside the serialized mutation queue. Close cancels before waiting. Cancelled
|
||||
Start fails; queued Stop retains ordinary receipt/admission handling. Delayed
|
||||
responses cannot commit evidence after failure. Wrong agent/revision/body does
|
||||
not cancel. No blanket stale-revision bypass or recovered-PID kill added.
|
||||
- D3: request promise commit and evidence commit recheck session health; host
|
||||
rechecks before running. Coalesced completed-response + malformed-tail regression
|
||||
fails at both ACP and host production seams, rather than accepting then waiting
|
||||
for heartbeat quarantine.
|
||||
## Actual executable evidence and precise remaining gap
|
||||
|
||||
This candidate has NOT received independent delta re-review. Report fixtures
|
||||
remain independent; no review artifacts modified.
|
||||
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.
|
||||
|
||||
## Dependency route and validation
|
||||
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.
|
||||
|
||||
Read and applied `WORK_LOGS/BEEHIVE_DESIGN_183FFAF0/DEPENDENCY_INSTALLATION.md`.
|
||||
Public npm registry is host-policy blocked; approved Block mirror is the supported
|
||||
route, not a policy bypass. Borrowed ignored modules removed. A standalone
|
||||
registry-agnostic `pnpm-lock.yaml` now pins all seven packages with integrity.
|
||||
No global/user npm/Git config or machine-path dependency committed. Use:
|
||||
**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.
|
||||
|
||||
## Validation / dependencies / next action
|
||||
|
||||
Hermit Node24.15.0 / pinned pnpm11.4.0. Standalone registry-agnostic lockfile unchanged;
|
||||
no new dependencies or borrowed modules. Approved install route remains:
|
||||
|
||||
```
|
||||
. ./bin/activate-hermit
|
||||
@@ -93,32 +95,27 @@ cd beehive
|
||||
pnpm install --ignore-workspace --frozen-lockfile --ignore-scripts \
|
||||
--registry https://global.block-artifacts.com/artifactory/api/npm/square-npm/
|
||||
npm run check
|
||||
npm test
|
||||
BEEHIVE_REAL_BUZZ_ACP=/Applications/Buzz.app/Contents/MacOS/buzz-acp npm test
|
||||
```
|
||||
|
||||
Fresh empty node_modules + fresh temporary pnpm store fetched all seven packages
|
||||
successfully on pnpm 11.4.0 / Node 24.15.0. Strict TypeScript check passes;
|
||||
**npm test: 9/9 top-level tests pass** (13.2 seconds). Tests include original
|
||||
ACP/WS/TUI regressions, D1–D3, and blocked conversation setup/launch contract.
|
||||
`git diff --check` passes. Local commit uses configured Logan Johnson
|
||||
<loganj@squareup.com> author/committer and DCO; no persistent Git config edits.
|
||||
No real Databricks OAuth, model call, conversation relay admission or external
|
||||
buzz-acp network session was exercised. Generic ACP fixtures are NOT Buzz relay
|
||||
parity. No repo-wide `just ci` or production readiness claim.
|
||||
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.
|
||||
|
||||
## Next executable action / full scope
|
||||
**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.
|
||||
|
||||
**Implement the host-owned ACP stdio broker described above, then remove the
|
||||
conversation Start gate only when adapter/shim teardown and exact-model real
|
||||
conversation correlation pass tests.** Do not spend another run rediscovering
|
||||
upstream routing. Preserve the provisioned agent key and the two-relay distinction.
|
||||
|
||||
Still open: actual Buzz-relay conversation/owner admission, live provider proof,
|
||||
catalog auth ambiguity, reconnect/ACKs and durable TUI intents, crash recovery and
|
||||
strong OS containment, multiple agents/hosts/setups, profiles/instructions/metadata,
|
||||
key import/revocation, Restart/Move, other Desktop harnesses/presets/providers,
|
||||
relay-mesh and compute. S1 two-host same identity is unproven. Source setup is one
|
||||
user-facing object but initial key setup is still coupled to first harness setup;
|
||||
conversation attachment now independently preserves keys. No remaining scope row
|
||||
is an owner-approved exclusion. Hash-to-spawn TOCTOU, hostile journals, rollback/
|
||||
clone protection and outbox pruning remain explicitly outside current evidence.
|
||||
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.
|
||||
|
||||
@@ -135,7 +135,7 @@ until Stop. Auth/protocol failures do not become accepted Start.
|
||||
|
||||
**This is an ACP integration probe, NOT yet a conversational Buzz relay agent.**
|
||||
The persistent Beehive agent key binds host authority but is not supplied to this
|
||||
stdio-only probe, and there is no relay prompt/conversation bridge yet. Model
|
||||
stdio-only probe, and this unconfigured path does not use the conversation broker. Model
|
||||
acknowledgement is the trusted external executable's claim, not cryptographic
|
||||
provider/model attestation. Only deterministic external ACP fixtures have passed;
|
||||
no credentialed Databricks response, live account or installed binary ACP
|
||||
@@ -158,27 +158,34 @@ 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 (launch blocked)
|
||||
## External conversation setup (broker wired; delivery incomplete)
|
||||
|
||||
`node src/cli.ts conversation-setup /absolute/existing-host-directory` attaches
|
||||
an installed **buzz-acp** executable and separate Buzz conversation relay URL to
|
||||
an existing Buzz Agent setup. It never generates/replaces the provisioned agent
|
||||
identity, reads OAuth or initiates a connection. Close the host first; setup uses
|
||||
the same atomic exclusion lock and leaves the key/auth context unchanged.
|
||||
an installed **buzz-acp** and separate conversation relay to the SAME existing
|
||||
agent key. Close the host first; setup uses its exclusion lock. It never reads
|
||||
owner profiles, performs sign-in or regenerates an identity.
|
||||
|
||||
**This does not enable a conversation yet.** A configured conversation Start is
|
||||
rejected before spawn; inventory explains why. Upstream `AcpClient::spawn` calls
|
||||
`process_group(0)`, so putting buzz-acp inside the existing outer supervisor would
|
||||
let its ACP adapter escape owned Stop. `apply_model_switch` can also proceed on
|
||||
a default model after rejection. A host-owned TS ACP stdio broker must retain
|
||||
adapter ownership and observe exact-model/session completions in actual Buzz
|
||||
conversations before this gate can be removed. Buzz-acp—not Beehive—will continue
|
||||
to own Nostr auth, membership, author gates, mentions, replies and delivery.
|
||||
Start now uses a private per-run TypeScript ACP broker. External buzz-acp owns
|
||||
Buzz authentication, membership, mentions and dispatch; its detached childless
|
||||
stdio shim cannot choose executables/env/keys. The host separately owns the actual
|
||||
harness process group. The proxy enforces exact-model acknowledgement at each
|
||||
conversation session boundary and hashes the actual completed response, never a
|
||||
separate greeting. Start waits up to 30 seconds for a real inbound conversation;
|
||||
missing relay admission or local sign-in fails Start. Stop tears down all retained
|
||||
runtime/harness groups and confirms shim exit. No raw text or credentials enter
|
||||
management observations. This is not provider attestation or delivery proof.
|
||||
|
||||
The launch plan supplies the existing agent key, explicit owner, optional local
|
||||
NIP-OA attestation and authoritative relay/model env. None has been exercised
|
||||
against a real conversation relay. Community admission and host-service-user OAuth
|
||||
remain explicit operator actions; this package has not performed them.
|
||||
Both external process-contract fixtures and the installed buzz-acp have exercised
|
||||
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.
|
||||
|
||||
Community admission/attestation and normal OAuth in the named host-service context
|
||||
remain explicit local/operator actions. No live Databricks request has occurred.
|
||||
|
||||
## Supported / remaining parity
|
||||
|
||||
@@ -188,7 +195,7 @@ remain explicit operator actions; this package has not performed them.
|
||||
| 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 | live OAuth/model proof, full conversation bridge and diagnostic catalog provenance |
|
||||
| Buzz Agent Databricks v2 | Local setup + TypeScript ACP probe boundary | fixed-authority tool/reply delivery, 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 |
|
||||
|
||||
@@ -41,16 +41,26 @@ strict TypeScript checking passed on Node 24.15.0 / pnpm 11.4.0. Keep
|
||||
remote Stop and host close. `disconnect.test.ts` verifies parent IPC loss during
|
||||
runner life uses the same ownership-preserving teardown, including closed-IPC
|
||||
notification handling. Fixtures self-expire to bound failed regression runs.
|
||||
- `conversation.test.ts`: exact local external buzz-acp launch contract, preserved
|
||||
provisioned identity, authoritative owner/relay/model env, no ambient provider
|
||||
token, rejected lossy args and unsafe URLs, sanitized remote summary. Real
|
||||
management WS refuses unsafe conversation Start before any process/run state,
|
||||
preserves Save/Stop and keys, and exercises CLI exclusion-lock cleanup.
|
||||
- `conversation.test.ts`: persistent identity and exact external env, plus real
|
||||
management WS Start -> external runtime fixture -> escaping detached shim ->
|
||||
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.
|
||||
- `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
|
||||
`BEEHIVE_REAL_BUZZ_ACP=/Applications/Buzz.app/Contents/MacOS/buzz-acp npm test`.
|
||||
Uses fresh fixture identities/dirs and an isolated NIP-42/REST/NIP-29 relay
|
||||
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.
|
||||
|
||||
Tests use isolated keys/directories/ports/processes only. Generic external ACP
|
||||
fixtures are **not** Buzz conversation relay protocol fixtures. There is no
|
||||
end-to-end identity-bearing Buzz conversation test yet. No real OAuth/model call,
|
||||
owner credential store, production relay, native service or old app is touched.
|
||||
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
|
||||
skip. The installed fixture relay does not establish actual community admission,
|
||||
current relay compatibility, provider/model attestation or working reply delivery.
|
||||
|
||||
## Independent evidence and limits
|
||||
|
||||
@@ -65,4 +75,4 @@ 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 intentionally blocked external conversation topology.
|
||||
executable action and the wired broker and remaining fixed-authority tool/delivery gap.
|
||||
|
||||
18
beehive/src/acp-shim.ts
Normal file
18
beehive/src/acp-shim.ts
Normal file
@@ -0,0 +1,18 @@
|
||||
/** Private stdio transport only: no subprocesses, executable selection or env grants. */
|
||||
import { connect } from 'node:net';
|
||||
const [path, capability] = process.argv.slice(2);
|
||||
if (!path || !capability || process.argv.length !== 4) process.exit(2);
|
||||
const socket = connect(path);
|
||||
const timer = setTimeout(() => process.exit(2), 5000);
|
||||
socket.once('connect', () => {
|
||||
clearTimeout(timer);
|
||||
socket.write(JSON.stringify({ capability, pid: process.pid }) + '\n');
|
||||
process.stdin.pipe(socket);
|
||||
socket.pipe(process.stdout);
|
||||
});
|
||||
// Upstream starts this shim in a separate process group. It never spawns children;
|
||||
// EOF, disconnect and TERM all terminate it, independently of upstream lifetime.
|
||||
const exit = () => { socket.destroy(); process.exit(0); };
|
||||
socket.on('error', exit); socket.on('close', exit);
|
||||
process.stdin.on('end', exit); process.stdout.on('error', exit);
|
||||
process.on('SIGTERM', exit); process.on('SIGINT', exit);
|
||||
@@ -9,7 +9,7 @@ export type AgentLaunch = Readonly<{ executable: string; args: readonly string[]
|
||||
/** ACP catalogs can be fallback data; even a nonempty result is NOT auth evidence. */
|
||||
export type Catalog = { state: 'reported' | 'empty' | 'filtered'; models: string[]; authentication: 'unverified' };
|
||||
/** Same child/session acknowledgement plus completed text response, not provider attestation. */
|
||||
export type Evidence = { session: string; model: string; responseHash: string; stopReason: 'end_turn'; source: 'external-acp-session'; executableHash: string };
|
||||
export type Evidence = { session: string; model: string; responseHash: string; stopReason: 'end_turn'; source: 'external-acp-session' | 'external-buzz-conversation'; executableHash: string; agentPublicKey?: string; runtimeExecutableHash?: string };
|
||||
const hash = (s: string | Buffer) => createHash('sha256').update(s).digest('hex');
|
||||
const record = (v: unknown): Record<string, unknown> => {
|
||||
if (!v || typeof v !== 'object' || Array.isArray(v)) throw Error('Invalid ACP response');
|
||||
|
||||
262
beehive/src/broker.ts
Normal file
262
beehive/src/broker.ts
Normal file
@@ -0,0 +1,262 @@
|
||||
import { setTimeout as delay } from 'node:timers/promises';
|
||||
import { createServer, type Socket } from 'node:net';
|
||||
import { mkdtempSync, rmSync } from 'node:fs';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join } from 'node:path';
|
||||
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 { prepareConversation, type ConversationSetup } from './conversation.ts';
|
||||
|
||||
type RPC = { jsonrpc: string; id?: string | number; method?: string; params?: any; result?: any; error?: unknown };
|
||||
type Pending = { method: string; params: any };
|
||||
const validID = (s: unknown): s is string => typeof s === 'string' && /^[a-zA-Z0-9_.:/-]{1,200}$/.test(s);
|
||||
|
||||
/** Run-local ACP transport. Authority comes only from the locally snapshotted plan. */
|
||||
export class ConversationSession {
|
||||
private directory: string;
|
||||
private capability = randomBytes(32).toString('hex');
|
||||
private sockets = new Set<Socket>();
|
||||
private harnesses: OwnedProcess[] = [];
|
||||
private shimPids = new Set<number>();
|
||||
private runtime?: OwnedProcess;
|
||||
private cancellations: (() => void)[] = [];
|
||||
private failed = false;
|
||||
private stopping?: Promise<void>;
|
||||
private resolve!: (e: Evidence) => void;
|
||||
private reject!: (e: Error) => void;
|
||||
private evidence = new Promise<Evidence>((resolve, reject) => { this.resolve = resolve; this.reject = reject; });
|
||||
private server = createServer(socket => this.accept(socket));
|
||||
private prepared;
|
||||
private plan;
|
||||
private timeout: ReturnType<typeof setTimeout>;
|
||||
readonly ready: Promise<void>;
|
||||
constructor(setup: ConversationSetup, launch: AgentLaunch, secret: string, owner: string, timeoutMs = 30_000) {
|
||||
this.prepared = prepareAgent(launch);
|
||||
this.plan = prepareConversation(setup, launch, secret, owner);
|
||||
this.directory = mkdtempSync(join(tmpdir(), 'bh-'));
|
||||
// Rejection can precede verify() while the local listener is opening.
|
||||
void this.evidence.catch(() => {});
|
||||
this.timeout = setTimeout(() => this.fail('Conversation response timed out; check relay admission and local sign-in'), timeoutMs);
|
||||
this.server.on('error', () => this.fail('Private ACP broker unavailable'));
|
||||
this.ready = new Promise<void>((resolve, reject) => {
|
||||
this.server.once('error', reject);
|
||||
this.server.listen(join(this.directory, 'acp'), () => {
|
||||
if (this.failed) { reject(Error('Conversation Start cancelled')); return; }
|
||||
const shim = fileURLToPath(new URL('./acp-shim.ts', import.meta.url));
|
||||
const args = [shim, join(this.directory, 'acp'), this.capability];
|
||||
if (args.some(a => a.includes(','))) { reject(Error('Unsupported local shim path')); return; }
|
||||
this.runtime = spawnOwned(this.plan.executable, [], this.plan.workspace, {
|
||||
...this.plan.env, BUZZ_ACP_AGENT_COMMAND: process.execPath, BUZZ_ACP_AGENT_ARGS: args.join(','),
|
||||
});
|
||||
this.runtime.child.stdout.resume(); this.runtime.child.stderr.resume();
|
||||
this.watch(this.runtime);
|
||||
this.runtime.ready.then(resolve, reject);
|
||||
});
|
||||
});
|
||||
void this.ready.catch(() => this.fail('Conversation runtime unavailable'));
|
||||
}
|
||||
get healthy() { return !this.failed; }
|
||||
get exited() { return this.failed || !!this.runtime?.exited; }
|
||||
get connected() { return !this.stopping || !this.failed; }
|
||||
private watch(owned: OwnedProcess) {
|
||||
owned.child.on('message', () => { if (owned.exited) this.fail('Conversation process exited'); });
|
||||
owned.child.on('exit', () => this.fail('Conversation supervisor exited'));
|
||||
owned.child.on('error', () => this.fail('Conversation process unavailable'));
|
||||
void owned.ready.catch(() => this.fail('Conversation harness unavailable'));
|
||||
}
|
||||
private fail(reason: string) {
|
||||
if (this.failed) return;
|
||||
this.failed = true; clearTimeout(this.timeout); this.reject(Error(reason));
|
||||
for (const socket of this.sockets) socket.destroy();
|
||||
// Keep ownership handles until the caller verifies complete teardown.
|
||||
}
|
||||
/** Interrupt Start without allowing late ACP responses to commit readiness. */
|
||||
cancel() {
|
||||
if (!this.failed) for (const cancel of this.cancellations) cancel();
|
||||
this.fail('Conversation Start cancelled');
|
||||
}
|
||||
/** Evidence is from the external runtime's actual prompt, never a synthetic probe. */
|
||||
async verify() {
|
||||
await this.ready;
|
||||
const evidence = await this.evidence;
|
||||
if (!this.healthy) throw Error('Conversation failed before evidence commit');
|
||||
return evidence;
|
||||
}
|
||||
/** Close escaping, childless shims via EOF and stop every retained live group. */
|
||||
stop(): Promise<void> {
|
||||
if (this.stopping) return this.stopping;
|
||||
this.cancel();
|
||||
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()));
|
||||
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.
|
||||
for (let i = 0; i < 100 && this.shimPids.size; i++) {
|
||||
for (const pid of this.shimPids) {
|
||||
try { process.kill(pid, 0); }
|
||||
catch (error) { if ((error as NodeJS.ErrnoException).code === 'ESRCH') this.shimPids.delete(pid); }
|
||||
}
|
||||
if (this.shimPids.size) await delay(25);
|
||||
}
|
||||
if (this.shimPids.size) throw Error('ACP shim exit unconfirmed; reconcile locally');
|
||||
rmSync(this.directory, { recursive: true, force: true });
|
||||
})();
|
||||
return this.stopping;
|
||||
}
|
||||
private accept(socket: Socket) {
|
||||
if (this.failed || this.sockets.size >= 8) { socket.destroy(); return; }
|
||||
this.sockets.add(socket);
|
||||
let authenticated = false;
|
||||
let buffer = '';
|
||||
let harness: OwnedProcess | undefined;
|
||||
const timer = setTimeout(() => socket.destroy(), 2000);
|
||||
socket.on('error', () => socket.destroy());
|
||||
socket.on('close', () => {
|
||||
clearTimeout(timer); this.sockets.delete(socket);
|
||||
if (authenticated && !this.failed) this.fail('ACP shim disconnected');
|
||||
});
|
||||
socket.setEncoding('utf8');
|
||||
socket.on('data', (chunk: string) => {
|
||||
if (authenticated) return;
|
||||
buffer += chunk;
|
||||
if (buffer.length > 4096) { socket.destroy(); return; }
|
||||
const end = buffer.indexOf('\n');
|
||||
if (end < 0) return;
|
||||
let hello: { capability?: string; pid?: number };
|
||||
try { hello = JSON.parse(buffer.slice(0, end)); } catch { socket.destroy(); return; }
|
||||
if (!hello || hello.capability !== this.capability || !Number.isSafeInteger(hello.pid) || hello.pid! <= 1 || this.harnesses.length >= 8) { socket.destroy(); return; }
|
||||
this.shimPids.add(hello.pid!);
|
||||
authenticated = true; clearTimeout(timer);
|
||||
const { plan, env } = this.prepared;
|
||||
// No data supplied by a shim can alter argv, workspace, identity or env.
|
||||
harness = spawnOwned(plan.executable, plan.args, plan.workspace, {
|
||||
...env, 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.harnesses.push(harness); this.watch(harness);
|
||||
this.proxy(socket, harness, buffer.slice(end + 1));
|
||||
buffer = '';
|
||||
});
|
||||
}
|
||||
private proxy(socket: Socket, harness: OwnedProcess, initial: string) {
|
||||
const pending = new Map<string | number, Pending>();
|
||||
const sessions = new Set<string>();
|
||||
const modelConfigs = new Set<string>(['model']);
|
||||
const prompts = new Map<string, { id: string | number; hash: ReturnType<typeof createHash>; text: boolean; cancelled: boolean }>();
|
||||
this.cancellations.push(() => {
|
||||
for (const sessionId of prompts.keys()) harness.child.stdin.write(JSON.stringify({ jsonrpc: '2.0', method: 'session/cancel', params: { sessionId } }) + '\n');
|
||||
});
|
||||
const injected = new Map<string, { original: RPC; session: string }>();
|
||||
let sequence = 0;
|
||||
const send = (target: NodeJS.WritableStream, msg: RPC) => {
|
||||
if (this.failed) return;
|
||||
if (!target.write(JSON.stringify(msg) + '\n')) this.fail('ACP transport backpressure limit');
|
||||
};
|
||||
const inspect = (msg: RPC, fromHarness: boolean) => {
|
||||
if (msg.jsonrpc !== '2.0' || (msg.id !== undefined &&
|
||||
!(typeof msg.id === 'string' && msg.id.length <= 200) &&
|
||||
!(typeof msg.id === 'number' && Number.isSafeInteger(msg.id)))) throw Error();
|
||||
if (msg.method !== undefined) {
|
||||
if (typeof msg.method !== 'string' || !msg.method.length || msg.result !== undefined || msg.error !== undefined) throw Error();
|
||||
} else if (msg.id === undefined || (Object.hasOwn(msg, 'result') === Object.hasOwn(msg, 'error'))) throw Error();
|
||||
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.
|
||||
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();
|
||||
sessions.delete(p.sessionId);
|
||||
}
|
||||
if (p && ['env', 'environment', 'executable', 'command'].some(key => Object.hasOwn(p, key))) throw Error();
|
||||
if (msg.id !== undefined) {
|
||||
if (pending.size >= 128 || pending.has(msg.id) || String(msg.id).startsWith('beehive-model-')) throw Error();
|
||||
pending.set(msg.id, { method: msg.method, params: p });
|
||||
}
|
||||
if (msg.method === 'session/set_model' && (!sessions.has(p?.sessionId) || p.modelId !== this.prepared.plan.model)) throw Error();
|
||||
if (msg.method === 'session/set_config_option' && modelConfigs.has(p?.configId) && p.value !== this.prepared.plan.model) throw Error();
|
||||
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();
|
||||
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;
|
||||
}
|
||||
}
|
||||
if (fromHarness && !msg.method && msg.id !== undefined) {
|
||||
const internal = injected.get(String(msg.id));
|
||||
if (internal) {
|
||||
injected.delete(String(msg.id));
|
||||
if (msg.error !== undefined || msg.result?.sessionId !== internal.session || msg.result?.modelId !== this.prepared.plan.model) throw Error();
|
||||
sessions.add(internal.session); send(socket, internal.original); return;
|
||||
}
|
||||
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) {
|
||||
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 ?? []) {
|
||||
if (option.category === 'model' && typeof option.id === 'string') modelConfigs.add(option.id);
|
||||
if (option.category === 'model' && typeof option.configId === 'string') modelConfigs.add(option.configId);
|
||||
}
|
||||
if (modelConfigs.size > 128) throw Error();
|
||||
sessions.delete(session);
|
||||
const id = `beehive-model-${++sequence}`;
|
||||
injected.set(id, { original: msg, session });
|
||||
send(harness.child.stdin, { jsonrpc: '2.0', id, method: 'session/set_model', params: { sessionId: session, modelId: this.prepared.plan.model } }); return;
|
||||
}
|
||||
if (request.method === 'session/set_model') {
|
||||
if (msg.error !== undefined || msg.result?.sessionId !== request.params.sessionId || msg.result?.modelId !== this.prepared.plan.model) throw Error();
|
||||
sessions.add(request.params.sessionId);
|
||||
}
|
||||
if (request.method === 'session/prompt') {
|
||||
const session = request.params.sessionId; const prompt = prompts.get(session); prompts.delete(session);
|
||||
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).
|
||||
queueMicrotask(() => { if (this.healthy) { clearTimeout(this.timeout); this.resolve(evidence); } });
|
||||
}
|
||||
}
|
||||
}
|
||||
if (fromHarness && msg.method === 'session/update') {
|
||||
const p = msg.params; const u = p?.update;
|
||||
if (u?.sessionUpdate === 'current_model_update' && u.currentModelId !== this.prepared.plan.model) throw Error();
|
||||
if (u?.sessionUpdate === 'config_option_update' && Array.isArray(u.configOptions)) {
|
||||
for (const option of u.configOptions) if (option.category === 'model' && option.currentValue !== this.prepared.plan.model) throw Error();
|
||||
}
|
||||
const prompt = prompts.get(p?.sessionId);
|
||||
if (prompt && u?.sessionUpdate === 'agent_message_chunk' && u.content?.type === 'text' && typeof u.content.text === 'string') {
|
||||
prompt.hash.update(u.content.text); prompt.text ||= !!u.content.text.trim();
|
||||
}
|
||||
}
|
||||
send(fromHarness ? socket : harness.child.stdin, msg);
|
||||
};
|
||||
const reader = (fromHarness: boolean) => {
|
||||
let buffer = '';
|
||||
return (chunk: string) => {
|
||||
if (this.failed) return;
|
||||
buffer += chunk;
|
||||
if (Buffer.byteLength(buffer) > 1024 * 1024) { this.fail('ACP frame limit exceeded'); return; }
|
||||
try {
|
||||
let end: number;
|
||||
while ((end = buffer.indexOf('\n')) >= 0) {
|
||||
const line = buffer.slice(0, end); buffer = buffer.slice(end + 1);
|
||||
if (line.trim()) { const msg = JSON.parse(line); if (!msg || typeof msg !== 'object' || Array.isArray(msg)) throw Error(); inspect(msg, fromHarness); }
|
||||
}
|
||||
} catch { this.fail('Conversation ACP protocol/model verification failed (diagnostics withheld)'); }
|
||||
};
|
||||
};
|
||||
const inbound = reader(false);
|
||||
socket.on('data', inbound);
|
||||
harness.child.stdout.setEncoding('utf8'); harness.child.stdout.on('data', reader(true));
|
||||
harness.child.stdin.on('error', () => this.fail('ACP input failed'));
|
||||
harness.child.stderr.resume();
|
||||
if (initial) inbound(initial);
|
||||
}
|
||||
}
|
||||
@@ -52,16 +52,7 @@ export function prepareConversation(input: ConversationSetup, agent: AgentLaunch
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Do not regress owned Stop: upstream AcpClient::spawn uses process_group(0),
|
||||
* escaping the existing outer group anchor. A host-owned ACP stdio broker must
|
||||
* supervise the actual adapter before this plan is safe to execute.
|
||||
*/
|
||||
export function requireConversationContainment(): never {
|
||||
throw Error('Conversation Start unavailable: external buzz-acp creates a separate ACP process group; host-owned ACP bridge containment is not implemented. No process started.');
|
||||
}
|
||||
|
||||
/** 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: 'blocked: ACP subtree containment required' };
|
||||
return { transport: 'external-buzz-acp', admission: 'unverified', modelEvidence: 'not-observed', launch: 'host-owned ACP broker; awaiting conversation evidence' };
|
||||
}
|
||||
|
||||
@@ -4,9 +4,10 @@ import { createHash } from 'node:crypto';
|
||||
import { connect, validateRelayURL } from './client.ts';
|
||||
import { digest, fields, message, object, publicKey, text, type Message } from './protocol.ts';
|
||||
import { readPrivate, writePrivate } from './storage.ts';
|
||||
import { ConversationSession } from './broker.ts';
|
||||
import { spawnOwned, type OwnedProcess } from './owned.ts';
|
||||
import { AgentSession, prepareAgent, type AgentLaunch, type Catalog, type Evidence } from './acp.ts';
|
||||
import { prepareConversation, requireConversationContainment, conversationSummary, type ConversationSetup } from './conversation.ts';
|
||||
import { prepareConversation, conversationSummary, type ConversationSetup } from './conversation.ts';
|
||||
|
||||
export type Setup = { host: string; ownerSecret: string; agentSecret: string; runner: string; args: string[]; workspace: string; mode: 'fixture' | 'buzz-agent-databricks-v2'; databricksHost?: string; serviceHome?: string; configDirectory?: string; conversation?: ConversationSetup };
|
||||
type Selection = { model: string; workspace: string; profile: string };
|
||||
@@ -33,15 +34,15 @@ export async function host(directory: string, url: string) {
|
||||
if (state.phase !== 'stopped') state.phase = 'quarantined';
|
||||
const save = () => writePrivate(path,state);
|
||||
try { save(); } catch (e) { rmdirSync(lock); throw e; }
|
||||
let owned: OwnedProcess | undefined;
|
||||
let acp: AgentSession | undefined;
|
||||
let owned: OwnedProcess | ConversationSession | undefined;
|
||||
let acp: AgentSession | ConversationSession | undefined;
|
||||
let catalog: Catalog | { state: 'not-probed' | 'failed'; authentication: 'unverified' } = { state: 'not-probed', authentication: 'unverified' };
|
||||
function inventory() {
|
||||
return message('inventory',setup.host,agent,state.revision, {
|
||||
assignedHost: setup.host, phase: state.phase, selectedNext: state.selected, actualRun: state.actual,
|
||||
...(setup.conversation ? { conversation: conversationSummary(setup.conversation) } : {}),
|
||||
catalog, setup: setup.mode, models: [setup.mode === 'fixture' ? 'fixture-model' : 'databricks-claude-haiku-4-5'],
|
||||
workspaces: [setup.workspace], profiles: ['default'], readiness: setup.conversation ? 'Conversation Start blocked: host-owned ACP subtree containment required; relay/auth/model unverified' : setup.mode === 'fixture' ? 'fixture-only' : (state.phase === 'running' && state.actual?.evidence ? 'ACP model acknowledged + same-session response (not provider attestation or Buzz relay agent)' : 'unverified: Start runs an ACP greeting probe; sign in locally as host service user if required'), observedAt: Date.now(),
|
||||
workspaces: [setup.workspace], profiles: ['default'], readiness: setup.conversation ? (state.actual?.evidence ? 'Conversation session model acknowledged and response completed; relay delivery unverified' : 'Waiting for an admitted conversation and local provider sign-in; no synthetic prompt') : setup.mode === 'fixture' ? 'fixture-only' : (state.phase === 'running' && state.actual?.evidence ? 'ACP model acknowledged + same-session response (not provider attestation or Buzz relay agent)' : 'unverified: Start runs an ACP greeting probe; sign in locally as host service user if required'), observedAt: Date.now(),
|
||||
});
|
||||
}
|
||||
let closing = false;
|
||||
@@ -72,7 +73,7 @@ export async function host(directory: string, url: string) {
|
||||
let result = 'accepted';
|
||||
if (m.agent !== agent) result = 'not-authority';
|
||||
else if (m.revision !== state.revision) result = 'revision-conflict';
|
||||
else if ((state.phase === 'quarantined' || state.phase === 'transitioning') && !(m.type === 'stop' && owned?.child.connected)) result = 'quarantined';
|
||||
else if ((state.phase === 'quarantined' || state.phase === 'transitioning') && !(m.type === 'stop' && (owned instanceof ConversationSession ? owned.connected : owned?.child.connected))) result = 'quarantined';
|
||||
else {
|
||||
// Reserve operation before any spawn/kill; interrupted admission never retries effects.
|
||||
const pending = message('receipt',setup.host,agent,state.revision,{ operation: m.id, result: 'interrupted-reconcile-locally' });
|
||||
@@ -93,7 +94,6 @@ export async function host(directory: string, url: string) {
|
||||
if (setup.conversation) {
|
||||
if (!launch) throw Error('Conversation requires a Buzz Agent harness setup');
|
||||
prepareConversation(setup.conversation, launch, setup.agentSecret, state.binding.owner);
|
||||
requireConversationContainment(); // Fail before journaling a run or spawning anything.
|
||||
}
|
||||
state.actual = { selection: structuredClone(state.selected), executableHash: createHash('sha256').update(readFileSync(setup.runner)).digest('hex'), run: m.id };
|
||||
state.phase = 'transitioning'; save();
|
||||
@@ -101,6 +101,16 @@ export async function host(directory: string, url: string) {
|
||||
owned = spawnOwned(setup.runner, setup.args, state.actual.selection.workspace, { PATH: '/usr/bin:/bin', BEEHIVE_FIXTURE: '1' });
|
||||
owned.child.stdout.resume(); owned.child.stderr.resume();
|
||||
await owned.ready;
|
||||
} else if (setup.conversation) {
|
||||
const session = new ConversationSession(setup.conversation, launch!, setup.agentSecret, state.binding.owner);
|
||||
acp = session; owned = session;
|
||||
try {
|
||||
state.actual.evidence = await session.verify();
|
||||
if (!session.healthy) throw Error('Conversation failed before running commit');
|
||||
} catch (e) {
|
||||
await stopOwned(); state.phase = 'stopped'; state.actual = null;
|
||||
throw e;
|
||||
}
|
||||
} else {
|
||||
const session = new AgentSession(launch!); acp = session;
|
||||
owned = session.owned;
|
||||
|
||||
41
beehive/test/broker.test.ts
Normal file
41
beehive/test/broker.test.ts
Normal file
@@ -0,0 +1,41 @@
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { mkdtempSync, realpathSync, writeFileSync, readFileSync, existsSync, rmSync } from 'node:fs';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join, resolve } from 'node:path';
|
||||
import { setTimeout as delay } from 'node:timers/promises';
|
||||
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']) {
|
||||
const dir = realpathSync(mkdtempSync(join(tmpdir(), 'bh-broker-test-')));
|
||||
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' }, {
|
||||
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);
|
||||
const result = session.verify().then(value => ({ value }), error => ({ error }));
|
||||
try {
|
||||
if (mode === 'delayed' || mode === 'cancel') {
|
||||
for (let i = 0; i < 80 && !existsSync(join(dir, mode === 'cancel' ? 'cancel-observed' : 'prompt-started')); i++) await delay(20);
|
||||
assert.ok(existsSync(join(dir, mode === 'cancel' ? 'cancel-observed' : 'prompt-started')));
|
||||
session.cancel();
|
||||
}
|
||||
const r = await result;
|
||||
if (mode === 'ok') {
|
||||
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');
|
||||
} else assert.ok('error' in r, `${mode} must not commit evidence`);
|
||||
} finally {
|
||||
await session.stop();
|
||||
for (const file of ['runtime-pid', 'shim-pid', 'harness-pid', 'descendant-pid']) {
|
||||
if (existsSync(join(dir, file))) assert.throws(() => process.kill(Number(readFileSync(join(dir, file), 'utf8')), 0), { code: 'ESRCH' }, `${mode}: ${file}`);
|
||||
}
|
||||
rmSync(dir, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
});
|
||||
@@ -1,6 +1,6 @@
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { mkdtempSync, realpathSync, rmSync, readFileSync, existsSync } from 'node:fs';
|
||||
import { mkdtempSync, realpathSync, rmSync, readFileSync, existsSync, writeFileSync } from 'node:fs';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join, resolve } from 'node:path';
|
||||
import { setTimeout as delay } from 'node:timers/promises';
|
||||
@@ -11,10 +11,13 @@ import { message, newKey, publicKey, type Message } from '../src/protocol.ts';
|
||||
import { writePrivate } from '../src/storage.ts';
|
||||
|
||||
test('in-flight ACP Start is cancellable by authorized Stop and host close; malformed completion cannot commit running', async () => {
|
||||
for (const action of ['stop', 'close', 'bad-tail']) {
|
||||
for (const target of ['probe', 'conversation']) for (const action of ['stop', 'close', 'bad-tail']) {
|
||||
const dir = realpathSync(mkdtempSync(join(tmpdir(), 'beehive-cancel-')));
|
||||
const secret = newKey(); const agentSecret = newKey(); const agent = publicKey(agentSecret);
|
||||
writePrivate(join(dir, 'setup.json'), { host: 'cancel-host', ownerSecret: secret, agentSecret, runner: realpathSync(process.execPath), args: [resolve('test/acp-fixture.ts'), action === 'bad-tail' ? 'bad-tail' : 'delayed'], workspace: dir, mode: 'buzz-agent-databricks-v2', serviceHome: dir, configDirectory: dir, databricksHost: 'https://fixture.invalid' });
|
||||
const runtime = join(dir, 'runtime');
|
||||
writeFileSync(runtime, `#!/bin/sh\nexec '${realpathSync(process.execPath)}' '${resolve('test/conversation-runtime-fixture.ts')}'\n`, { mode: 0o700 });
|
||||
writeFileSync(join(dir, 'mode'), action === 'bad-tail' ? 'bad-tail' : 'delayed');
|
||||
writePrivate(join(dir, 'setup.json'), { host: 'cancel-host', ownerSecret: secret, agentSecret, runner: realpathSync(process.execPath), args: target === 'probe' ? [resolve('test/acp-fixture.ts'), action === 'bad-tail' ? 'bad-tail' : 'delayed'] : [resolve('test/conversation-harness-fixture.ts')], ...(target === 'conversation' ? { conversation: { executable: runtime, relay: 'ws://127.0.0.1:1' } } : {}), workspace: dir, mode: 'buzz-agent-databricks-v2', serviceHome: dir, configDirectory: dir, databricksHost: 'https://fixture.invalid' });
|
||||
const server = await relay(0, publicKey(secret), join(dir, 'relay.json'));
|
||||
const address = server.address(); assert.ok(address && typeof address !== 'string');
|
||||
const url = `ws://127.0.0.1:${address.port}`;
|
||||
|
||||
37
beehive/test/conversation-harness-fixture.ts
Normal file
37
beehive/test/conversation-harness-fixture.ts
Normal file
@@ -0,0 +1,37 @@
|
||||
/** Owned ACP harness fixture: exact session/model, bidirectional RPC, resistant child. */
|
||||
import { createInterface } from 'node:readline';
|
||||
import { readFileSync, writeFileSync } from 'node:fs';
|
||||
import { spawn } from 'node:child_process';
|
||||
import { publicKey } from '../src/protocol.ts';
|
||||
const mode = readFileSync('mode', 'utf8');
|
||||
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));
|
||||
const child = spawn(process.execPath, ['-e', "process.on('SIGTERM',()=>{}); setTimeout(()=>{},15000)"], { stdio: 'ignore' });
|
||||
writeFileSync('descendant-pid', String(child.pid));
|
||||
process.on('SIGTERM', () => {});
|
||||
setTimeout(() => process.exit(0), 15000).unref();
|
||||
const send = (m: any) => process.stdout.write(JSON.stringify(m) + '\n');
|
||||
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/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; }
|
||||
selected = m.params.modelId; result = { sessionId, modelId: mode === 'wrong-model' ? 'other' : selected };
|
||||
} else if (m.method === 'session/cancel') {
|
||||
writeFileSync('cancel-observed', 'yes');
|
||||
if (prompt) send({ jsonrpc: '2.0', id: prompt, result: { stopReason: 'cancelled' } }); continue;
|
||||
} else if (m.method === 'session/prompt') {
|
||||
if (selected !== process.env.BUZZ_AGENT_MODEL || m.params.sessionId !== sessionId) process.exit(8);
|
||||
writeFileSync('prompt-started', selected); prompt = m.id;
|
||||
send({ jsonrpc: '2.0', id: 'reverse', method: 'client/fixture', params: {} });
|
||||
if (mode === 'delayed' || mode === 'cancel') continue;
|
||||
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;
|
||||
} else throw Error('Unexpected method');
|
||||
send({ jsonrpc: '2.0', id: m.id, result });
|
||||
}
|
||||
32
beehive/test/conversation-runtime-fixture.ts
Normal file
32
beehive/test/conversation-runtime-fixture.ts
Normal file
@@ -0,0 +1,32 @@
|
||||
/** External runtime contract fixture, NOT a replacement Buzz relay implementation. */
|
||||
import { spawn } from 'node:child_process';
|
||||
import { createInterface } from 'node:readline';
|
||||
import { writeFileSync, readFileSync } from 'node:fs';
|
||||
import { publicKey } from '../src/protocol.ts';
|
||||
const mode = readFileSync('mode', 'utf8');
|
||||
writeFileSync('runtime-identity', publicKey(process.env.BUZZ_PRIVATE_KEY!));
|
||||
writeFileSync('runtime-pid', String(process.pid));
|
||||
const child = spawn(process.env.BUZZ_ACP_AGENT_COMMAND!, process.env.BUZZ_ACP_AGENT_ARGS!.split(','), {
|
||||
// Reproduce AcpClient::spawn's process_group(0) escape, not naive inheritance.
|
||||
detached: true, stdio: ['pipe', 'pipe', 'ignore'], env: { PATH: '/usr/bin:/bin' },
|
||||
});
|
||||
writeFileSync('shim-pid', String(child.pid));
|
||||
setTimeout(() => process.exit(0), 15000).unref();
|
||||
const pending = new Map<number, (m: any) => void>(); let seq = 0;
|
||||
const input = createInterface({ input: child.stdout });
|
||||
input.on('line', line => { const m = JSON.parse(line); if (m.method === 'client/fixture') child.stdin.write(JSON.stringify({ jsonrpc: '2.0', id: m.id, result: { ok: true } }) + '\n'); else pending.get(m.id)?.(m); });
|
||||
child.on('exit', () => process.exit(0));
|
||||
async function request(method: string, params: unknown) {
|
||||
const id = ++seq;
|
||||
return new Promise<any>(resolve => { pending.set(id, resolve); child.stdin.write(JSON.stringify({ jsonrpc: '2.0', id, method, params }) + '\n'); });
|
||||
}
|
||||
await request('initialize', { protocolVersion: 1, clientCapabilities: {} });
|
||||
const created = await request('session/new', { cwd: mode === 'foreign-workspace' ? '/' : process.cwd(), mcpServers: mode === 'injected-mcp' ? [{ name: 'bad', command: '/bin/sh', args: ['-c', 'exit 0'], env: [] }] : [] });
|
||||
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: [] });
|
||||
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;
|
||||
writeFileSync('runtime-completed', response.result.stopReason);
|
||||
@@ -1,6 +1,6 @@
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import { mkdtempSync, realpathSync, rmSync, readFileSync, existsSync } from 'node:fs';
|
||||
import { mkdtempSync, realpathSync, rmSync, readFileSync, existsSync, writeFileSync } from 'node:fs';
|
||||
import { spawnSync } from 'node:child_process';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { join, resolve } from 'node:path';
|
||||
@@ -47,10 +47,13 @@ test('external buzz-acp launch contract preserves identity and rejects lossy/uns
|
||||
} finally { rmSync(dir, { recursive: true, force: true }); }
|
||||
});
|
||||
|
||||
test('real management WS rejects unsafe conversation Start before spawn; Save/Stop and key binding survive', async () => {
|
||||
test('real management WS runs identity-bearing external conversation through escaped shim and owned harness; Save/Stop preserve keys', async () => {
|
||||
const dir = realpathSync(mkdtempSync(join(tmpdir(), 'beehive-conversation-host-')));
|
||||
const ownerSecret = newKey(); const agentSecret = newKey(); const key = publicKey(agentSecret);
|
||||
const setup = { host: 'conversation-host', ownerSecret, agentSecret, runner: executable, args: agent(dir).args, workspace: dir, mode: 'buzz-agent-databricks-v2', serviceHome: dir, configDirectory: dir, databricksHost: 'https://fixture.invalid', conversation: { executable, relay: 'wss://conversation.invalid', authTag: 'fixture-attestation' } };
|
||||
const runtime = join(dir, 'runtime');
|
||||
writeFileSync(runtime, `#!/bin/sh\nexec '${executable}' '${resolve('test/conversation-runtime-fixture.ts')}'\n`, { mode: 0o700 });
|
||||
writeFileSync(join(dir, 'mode'), 'ok');
|
||||
const setup = { host: 'conversation-host', ownerSecret, agentSecret, runner: executable, args: [resolve('test/conversation-harness-fixture.ts')], workspace: dir, mode: 'buzz-agent-databricks-v2', serviceHome: dir, configDirectory: dir, databricksHost: 'https://fixture.invalid', conversation: { executable: runtime, relay: 'wss://conversation.invalid', authTag: 'fixture-attestation' } };
|
||||
writePrivate(join(dir, 'setup.json'), setup);
|
||||
const server = await relay(0, publicKey(ownerSecret), join(dir, 'relay.json'));
|
||||
const address = server.address(); assert.ok(address && typeof address !== 'string');
|
||||
@@ -68,14 +71,22 @@ test('real management WS rejects unsafe conversation Start before spawn; Save/St
|
||||
}
|
||||
try {
|
||||
const start = await request('start', 0);
|
||||
assert.match(String(start.body.result), /Conversation Start unavailable.*No process started/);
|
||||
assert.equal(start.body.result, 'accepted');
|
||||
const state = JSON.parse(readFileSync(join(dir, 'journal.json'), 'utf8'));
|
||||
assert.equal(state.phase, 'stopped'); assert.equal(state.actual, null); assert.equal(state.revision, 0);
|
||||
assert.equal((await request('save', 0, { model, workspace: dir, profile: 'default' })).body.result, 'saved; running configuration unchanged');
|
||||
assert.equal((await request('stop', 1)).body.result, 'accepted');
|
||||
assert.equal(state.phase, 'running'); assert.equal(state.actual.evidence.session, 'conversation-session'); assert.equal(state.actual.evidence.model, model); assert.equal(state.revision, 1);
|
||||
assert.equal(readFileSync(join(dir, 'runtime-identity'), 'utf8'), key);
|
||||
assert.equal(readFileSync(join(dir, 'harness-identity'), 'utf8'), key);
|
||||
assert.equal(readFileSync(join(dir, 'runtime-completed'), 'utf8'), 'end_turn');
|
||||
assert.equal((await request('save', 1, { model, workspace: dir, profile: 'default' })).body.result, 'saved; running configuration unchanged');
|
||||
assert.equal((await request('stop', 2)).body.result, 'accepted');
|
||||
assert.equal(JSON.parse(readFileSync(join(dir, 'setup.json'), 'utf8')).agentSecret, agentSecret);
|
||||
assert.ok(!JSON.stringify(seen).includes('fixture-attestation'));
|
||||
assert.ok(!JSON.stringify(seen).includes(agentSecret));
|
||||
assert.ok(!JSON.stringify(seen).includes('Private conversation fixture response'));
|
||||
for (const file of ['runtime-pid', 'shim-pid', 'harness-pid', 'descendant-pid']) {
|
||||
const pid = Number(readFileSync(join(dir, file), 'utf8'));
|
||||
assert.throws(() => process.kill(pid, 0), { code: 'ESRCH' });
|
||||
}
|
||||
const cli = spawnSync(process.execPath, ['src/cli.ts', 'conversation-setup', dir], { encoding: 'utf8', timeout: 3000 });
|
||||
assert.notEqual(cli.status, 0); assert.ok(existsSync(join(dir, 'host.lock')));
|
||||
} finally {
|
||||
|
||||
87
beehive/test/installed-conversation.test.ts
Normal file
87
beehive/test/installed-conversation.test.ts
Normal file
@@ -0,0 +1,87 @@
|
||||
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 { tmpdir } from 'node:os';
|
||||
import { join, resolve } from 'node:path';
|
||||
import { createHash } from 'node:crypto';
|
||||
import { schnorr } from '@noble/curves/secp256k1';
|
||||
import { setTimeout as delay } from 'node:timers/promises';
|
||||
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 () => {
|
||||
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 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));
|
||||
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;
|
||||
}
|
||||
function event(kind: number, tags: string[][], content: string, key = ownerSecret) {
|
||||
const pubkey = publicKey(key); const created_at = Math.floor(Date.now() / 1000);
|
||||
const id = createHash('sha256').update(JSON.stringify([0, pubkey, created_at, kind, tags, content])).digest('hex');
|
||||
return { id, pubkey, created_at, kind, tags, content, sig: Buffer.from(schnorr.sign(id, key)).toString('hex') };
|
||||
}
|
||||
const members = event(39002, [['d', channel], ['p', agent], ['p', owner]], '');
|
||||
const metadata = event(39000, [['d', channel], ['name', 'fixture'], ['type', 'channel']], '');
|
||||
const inbound = event(9, [['h', channel], ['p', agent]], 'Please respond to this isolated fixture conversation.');
|
||||
const http = createServer(async (req, res) => {
|
||||
let body = ''; for await (const chunk of req) body += chunk;
|
||||
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] : []));
|
||||
} else if (req.url?.includes('/events')) {
|
||||
const e = JSON.parse(body); observeReply(e);
|
||||
res.end(JSON.stringify({ id: e.id, accepted: true }));
|
||||
} else res.end('[]');
|
||||
});
|
||||
const ws = new WebSocketServer({ server: http });
|
||||
ws.on('connection', socket => {
|
||||
socket.send(JSON.stringify(['AUTH', 'beehive-isolated-challenge']));
|
||||
socket.on('message', bytes => {
|
||||
const m = JSON.parse(bytes.toString());
|
||||
if (m[0] === 'AUTH') {
|
||||
const e = m[1];
|
||||
assert.equal(e.pubkey, agent); assert.equal(e.kind, 22242);
|
||||
assert.ok(e.tags.some((t: string[]) => t[0] === 'challenge' && t[1] === 'beehive-isolated-challenge'));
|
||||
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, e.pubkey));
|
||||
authenticated = true; socket.send(JSON.stringify(['OK', e.id, true, '']));
|
||||
} else if (m[0] === 'EVENT') { observeReply(m[1]); socket.send(JSON.stringify(['OK', m[1].id, true, ''])); }
|
||||
else if (m[0] === 'REQ') {
|
||||
subscriptions++; socket.send(JSON.stringify(['EOSE', m[1]]));
|
||||
if (m.slice(2).some((f: any) => f['#h']?.includes(channel))) setTimeout(() => socket.send(JSON.stringify(['EVENT', m[1], inbound])), 100);
|
||||
}
|
||||
});
|
||||
});
|
||||
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}` }, {
|
||||
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);
|
||||
try {
|
||||
await session.ready;
|
||||
for (let i = 0; i < 100 && !subscriptions; i++) await delay(50);
|
||||
assert.ok(authenticated, 'installed executable must authenticate with the provisioned key');
|
||||
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}`);
|
||||
} finally {
|
||||
await session.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 });
|
||||
}
|
||||
});
|
||||
Reference in New Issue
Block a user