From 6bd859479503832a331cb9f8eb0a495082c8f9d8 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Wed, 9 Sep 2026 22:35:53 -0400 Subject: [PATCH] feat(beehive): add external ACP probe and anchored host teardown Signed-off-by: Logan Johnson --- beehive/CHECKPOINT.md | 159 ++++++++++++++++++++------------- beehive/README.md | 77 +++++++++++----- beehive/TESTING.md | 27 ++++++ beehive/src/acp.ts | 154 +++++++++++++++++++++++++++++++ beehive/src/cli.ts | 20 +++-- beehive/src/client.ts | 8 +- beehive/src/host.ts | 81 ++++++++++------- beehive/src/owned.ts | 47 ++++++++++ beehive/src/supervisor.ts | 21 +++++ beehive/test/acp-fixture.ts | 25 ++++++ beehive/test/acp.test.ts | 95 ++++++++++++++++++++ beehive/test/orphan-fixture.ts | 12 +++ beehive/test/slice.test.ts | 42 ++++++++- 13 files changed, 644 insertions(+), 124 deletions(-) create mode 100644 beehive/src/acp.ts create mode 100644 beehive/src/owned.ts create mode 100644 beehive/src/supervisor.ts create mode 100644 beehive/test/acp-fixture.ts create mode 100644 beehive/test/acp.test.ts create mode 100644 beehive/test/orphan-fixture.ts diff --git a/beehive/CHECKPOINT.md b/beehive/CHECKPOINT.md index f7f2e8632d..71c38271ec 100644 --- a/beehive/CHECKPOINT.md +++ b/beehive/CHECKPOINT.md @@ -1,73 +1,110 @@ -# Beehive implementation checkpoint +# Beehive executable checkpoint -Scope: standalone TypeScript experimental package, no existing client changes. -Base origin/main 051c3a270be9c73da9ab06700bcab7d5552fceaa; branch beehive/cbfd9440. +Worktree `/Users/loganj/.buzz/REPOS/beehive-cbfd9440`, branch `beehive/cbfd9440`. +Base `051c3a270be9c73da9ab06700bcab7d5552fceaa`; prior reviewed milestone +`94d1da0e8113c27f73865233a18b1157f2ea8e4c`. This continuation is the commit +containing this checkpoint (resolve with `git rev-parse HEAD`); do not restart +repo/design discovery or create a second implementation worktree. -First slice uses a dedicated, isolated loopback TypeScript WebSocket relay. It is -NOT a Buzz/Nostr relay extension: no kind allocations, production policy changes, -or promise of current Buzz relay admission. Ciphertext and signed envelopes are -stored/forwarded by relay, never direct-host HTTP. One locally provisioned owner -Nostr identity is shared by trusted, non-cloned controller/host installations. -Signatures cannot identify individual hosts sharing that key. Local admission -journal is safety state (necessary exception to only runtime/keys local); remote -configuration and receipts are also on relay. No controller service. +## Implemented now -Implementation priorities: real network → validated contracts → durable host -admission/outbox → fixture runner lifecycle → line-oriented terminal UI. -Real Buzz Agent / Databricks v2 exact-model evidence is a separate gate; do not -claim fixture success proves authentication, model application or crash safety. -No owner credential stores may be accessed to remove that gate. +Standalone TypeScript package; dedicated loopback real WebSocket ciphertext relay, +signed/encrypted commands, owner-only atomic storage, fixed host/agent authority, +durable revision/fingerprint admission and receipts/outbox, selected-next separate +from captured actual run, assigned host retained stopped, local setup and remote +line-oriented TUI. Existing native clients/relay/harness source untouched. -Pending parity: guided vendor setup, real ACP catalog/exact-model bridge, -Goose/Claude Code/Codex, ten presets/custom ACP, other Buzz Agent providers, -mesh transport, compute deployment, profiles/metadata, Restart/Move, key import -with saved-owner verification/revocation, service installation, reconnect UX. -No exclusion from requested parity is treated as owner-approved. +**External harness boundary:** `src/acp.ts` implements bounded newline ACP RPC to +external buzz-agent, not buzz-acp. Grounded in pinned source initialize/session/new/ +session/set_model explicit {sessionId,modelId} ack, then session/prompt. Captures +local immutable launch inputs, scrubbed env with Databricks v2, dedicated HOME and +BUZZ_AGENT_CONFIG_DIR; requires exact model/session acknowledgement and completed +same-session text response. Stores response SHA-256, model, session and executable +SHA-256, never raw text/diagnostics. Host Start uses this production path; Save and +Stop do not consult auth. No unmerged BUZZ_ACP_REQUIRED_MODEL assumption. -## Implemented milestone (2026-09-10) +`setup` option 2 provisions buzz-agent plus fresh service-home/agent-config dirs; +`auth-info ` prints the exact shell-quoted local OAuth command/context, +host name and current uid, without starting login. Operator must use the host +service's OS account. No token broker, remote auth RPC, inherited provider secrets, +key import or historical cache access. Existing option-2 setup (buzz-acp) requires +explicit local reprovisioning, not automatic migration. -Executable code: `src/protocol.ts` (runtime codecs, owner Schnorr signatures, -private AES-GCM envelopes), `storage.ts` (0600 atomic/fsynced writes), `relay.ts` -(real loopback WebSocket ciphertext persistence/fanout), `client.ts`, `host.ts` -(fixed local authority binding, exclusive lock, revisioned admission/fingerprints, -outbox, immutable fixture launch, owned process-group Stop), `cli.ts` (local -identity/guided setup, relay/host entry points, line-oriented TUI). +**Honest narrow seam:** current upstream session/new may return fallback catalog +on OAuth/discovery failure. Thus not-probed/reported/empty/filtered/failed are +explicit, but authentication stays unverified. Authenticated-empty, expired, +denied vs missing auth cannot be inferred. Catalog success is not readiness. -Validation: `npm test` passes 2 tests including actual external runner and -in-group descendant, actual TUI process close/reopen twice, relay client replay, -wrong-agent/revision/fingerprint rejection, exclusive host lock and changed -saved-agent binding rejection, clean host restart retaining stopped assignment. -Node 26.8.1. Fixture success is NOT real harness/model/OAuth success. +**Not yet a Buzz conversation agent:** this boundary keeps an ACP session alive +following a greeting probe; persistent Beehive key binds control authority but is +not supplied to stdio probe. Relay conversation/prompt bridge and actual live +Databricks execution remain gates. An external executable's model acknowledgement +is not provider attestation. Fixtures are not live proof. -`node node_modules/typescript/bin/tsc --noEmit` was attempted and FAILS because -@types/node and @types/ws are unavailable. npm install from registry.npmjs.org -returned Block Cloudflare Dependency Confusion HTML; npm --offline ENOTCACHED. -Runtime tests use ignored symlinks to preinstalled public JS package dependencies -(documented in TESTING.md). No lockfile or clean-install claim. No policy bypass. -Repo-wide `just ci` was not run; no Rust/native/client files changed. +## Independent feedback consumed and corrected -The existing relay explicitly rejects unknown kinds (`handlers/ingest.rs:436`) -and p-gated REQ authorization is kind-specific (`handlers/req.rs:1319`). This -prototype does not presume arbitrary private-kind admission or prove that no -future standard envelope can work; hosted/current-Buzz relay integration remains -an explicit unsolved boundary. Dedicated dev relay is intentionally distinct. +Read UX_ACCEPTANCE.md and EXECUTABLE_REVIEW_94D1DA0E.md in +WORK_LOGS/BEEHIVE_DESIGN_183FFAF0 (review artifacts untouched). +- F1: ordinary runner exit must not disable teardown of owned descendants. + Added `owned.ts` + internal `supervisor.ts`: a live TypeScript group anchor + outlives the external runner, preserving group identity. Stop is an IPC request + to that living anchor, which signals ITS OWN group; host only polls group + absence. No kill to a possibly reused numeric PID/group after leader exit. + Quarantine retains verified-owned Stop/close; failed teardown retains lock. + Unexpected anchor loss still fails closed (no PID recovery). +- F2: validate URL before lock; unwind only newly acquired startup lock after + connect/save failure before accepting any operation. Malformed URL and refused + connection followed by successful retry covered; competing live lock preserved. +- F3: intentional Quit prints independent host lifetime, not UNKNOWN. Unexpected + disconnect scopes uncertainty to live reachability/unresolved requests and + does not erase received receipts. -Real target: installed buzz-agent and buzz-acp found under /Applications/Buzz.app; -only an isolated-env auth-help attempt was made, returning DATABRICKS_HOST -required. No OAuth cache, owner key/profile or historical private store read. -Local option 2 records Databricks v2 target and displays blocked readiness; -Start fails closed. README gives the actual local OAuth command and gate. +New candidate is NOT independently re-reviewed. S1 two-host/same-identity not +established; only single-host authority and stopped assignment tests exist. -Fresh-run next step: from `beehive/`, run `npm install && npm run check && npm test` -once approved npm access/types are available; fix all strict diagnostics and -create a lockfile before expanding the ACP bridge. If access remains blocked, -continue with cached-runtime tests and a targeted host-side catalog/exact-model -adapter, never weaken typing or claim model-ready based on spawn. +## Commands and evidence -Known unfinished safety: no crash recovery beyond quarantine/exclusive lock, -no host reconnect, no relay ACK/TUI durable pending-intent store, no outbox -pruning, no OS containment against escaping descendants, no immutable executable -handle/script pinning, no hostile-local-journal validation/migration. No Move, -Restart, key import/revocation, real ACP bridge, multi-agent host, full profile -workflow or vendor parity. README has the concrete parity matrix. This is a -reviewable development checkpoint, not system completion or merge/release-ready. +Activate Hermit from repo root (`. ./bin/activate-hermit`), then `cd beehive`. +`npm test`: **5/5 top-level tests pass** on Node 24.15.0; table cases include +wrong model/session, cancelled/rejected/no-response protocol, malformed JSON, +timeout/flood, catalog empty/filtered ambiguity and sanitized output. Real WS +host→external ACP fixture Start/Save/Stop, auth-directory removal before Save/Stop, +actual auth-info CLI, two TUI subprocess reopen/quit sessions, initial-connect +unwind, persistent retries/conflicts and group-anchor parent-exit regressions for +both remote Stop and host close. All test identities/processes/stores isolated. + +`node node_modules/typescript/bin/tsc --noEmit`: **FAIL exit 2**, missing +@types/node/@types/ws declarations (diagnostics also cascade into implicit-any). +`npm run check` cannot find the tsc CLI because ignored borrowed dependency +symlinks have no .bin entry. No strict-check pass, lockfile or clean install claim. +Dependency investigator a2b36832 is separate; no approved dependency result has +arrived/been applied at this checkpoint. Continue integration, do not pause work +for the registry. No dependency-investigator artifacts modified. + +Installed executable safe probe: fresh temp HOME/config, no credentials/env +inheritance, `/Applications/Buzz.app/Contents/MacOS/buzz-agent --help` exits 2 +with BUZZ_AGENT_PROVIDER required. This proves executable presence only, not +version/capabilities/auth/model. No OAuth, live provider call or owner store read. + +`git diff --check` passes. No repo-wide just ci, push or PR; no merge/release-ready +claim. Commit uses configured Logan Johnson and DCO. + +## Next executable step / remaining scope + +Next: add a relay-driven conversation/probe operation to the retained ACP session +with run-bound request/response receipts and strict cancellation (no remote paths, +argv, env or credentials). Test it through the real host and external ACP fixture, +then have a human provision the exact host-service auth context for live-model +acceptance. Integrate approved dependency result and fix strict diagnostics as soon +as provided; request independent review of this candidate, including the anchor. + +Full requested parity remains open: hosted/current-Buzz relay admission (dedicated +loopback protocol is NOT compatible), reconnect/ACKs and durable TUI pending +intents, crash recovery/strong OS containment, multi-agent/setups, profiles and +instructions/metadata, key import/revocation, Restart/Move (same key destination-local, +no unreachable takeover or secret/session/workspace transfer), all Desktop harnesses, +presets/providers, mesh and compute deployment. Agent key authority and host assignment +must remain durable; no scope exclusions are owner-approved. Executable/script/config +replacement between hashing and spawn is still trusted-local-operator territory, +not immutable file-handle pinning. No hostile-journal validation, outbox pruning, +rollback/clone protection or cryptographic deployment certification. diff --git a/beehive/README.md b/beehive/README.md index 2ff580f7f5..7debea2965 100644 --- a/beehive/README.md +++ b/beehive/README.md @@ -71,13 +71,17 @@ assumes non-cloned, non-rollback, exclusively supervised installations. - Exact operation ID/fingerprint and revision preconditions; durable admission before effects, persistent replies/outbox, idempotent replay. No exactly-once - claim. Successful Start means process spawned, **not model/conversation ready**. + claim. Fixture Start means process spawned; ACP Start additionally requires probe evidence, **not Buzz conversation readiness**. - Immutable captured selection/executable digest; executable and workspace prerequisites checked immediately before fixture launch. Executable replacement between check and spawn is outside this trusted-operator preview; full prepared launch must pin all setup/credential/script generations and close that gap. -- Stop uses an in-lifetime child handle plus process group exit confirmation. - PID alone is never used for recovery. Failed/unexpected exit quarantines. +- Stop uses a live TypeScript group anchor and private in-lifetime IPC channel. + The anchor outlives the external runner, preserving group identity if its leader + exits. Stop asks that anchor to signal its own group, then confirms group absence; + it never sends a kill to a recovered or possibly reused numeric ID. Quarantine + preserves this teardown route. A lost anchor fails closed and retains the lock. + This is POSIX in-group supervision, not containment of escaping descendants. - Interrupted operation replies report reconciliation required. Abrupt host death leaves an exclusive lock: restart is blocked, not an automatic PID-based takeover. Local recovery tooling and crash-injection coverage remain pending. @@ -95,28 +99,57 @@ assumes non-cloned, non-rollback, exclusively supervised installations. ## Real target: Buzz Agent + Databricks v2 -Local setup option 2 captures installed adapter path and HTTPS workspace URL, -creates a separate local agent identity, and advertises the target selection -`databricks-claude-haiku-4-5` as **blocked**, not discovered/model-ready. Start -fails closed because exact-model application evidence is not implemented. - -Under the intended dedicated host-service OS user (not your Desktop context), -use an isolated configuration directory and the installed executable: +Local setup option 2 now takes **buzz-agent**, not buzz-acp. It creates fresh +owner-only `service-home` and `agent-config` directories beneath the new host +directory. Existing option-2 setups need deliberate local reprovisioning of the +runner and these paths; there is no automatic migration or credential import. +Never substitute an old Desktop configuration/cache. ```sh -BUZZ_AGENT_CONFIG_DIR= \ -DATABRICKS_HOST=https:// buzz-agent auth databricks +node src/cli.ts auth-info /absolute/host-directory ``` -Buzz Agent owns its normal browser OAuth ceremony, workspace/client/scopes cache -and refresh. No static token is required. The actual workspace/account is a -human input; no login URL or successful authentication was observed in this -milestone. `buzz-acp models --json` is the next external catalog seam, but it -spawns an ACP session and does not prove exact-model application. Current main -has no strict `BUZZ_ACP_REQUIRED_MODEL` capability to assume. Neither owner nor -historical credential files were read/copied. An installed `buzz-agent auth -databricks --help` invocation in a scrubbed environment returned -`DATABRICKS_HOST required`; it was not a login/authentication test. +This prints the exact shell-quoted executable, HOME, BUZZ_AGENT_CONFIG_DIR and +DATABRICKS_HOST context, with the host name and current OS uid. **The operator +must run that command as the same OS user that runs the host service.** It does +not perform login. Buzz Agent owns normal browser OAuth, cache, client/scopes and +headless refresh; Beehive does not broker tokens or inherit DATABRICKS_TOKEN. +No actual account/workspace is preconfigured here. Save and Stop never log in. + +Start in this preview launches a bounded TypeScript ACP boundary directly to the +locally selected external executable. It negotiates ACP v1, creates one session, +reads the catalog, requires `session/set_model` to acknowledge the exact selected +`databricks-claude-haiku-4-5` **and session ID**, then requests a short greeting on +that same session/process. Only a completed `end_turn` with same-session text +produces evidence: session ID, exact model, executable SHA-256 and response hash. +Text, raw diagnostics, names/descriptions and credentials are not published. +Selected-next is independent of the running session; the process remains owned +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 +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 +capability negotiation was observed. Installed `/Applications/Buzz.app/Contents/ +MacOS/buzz-agent --help` with a fresh isolated HOME/config returned provider-required +(exit 2); that is executable discovery, not an ACP capability test. + +The narrow protocol seam is grounded in pinned main `051c3a2`: +`crates/buzz-agent/src/lib.rs` initialize (349), session/new (500–610), +session/set_model (633–682); `config.rs` BUZZ_AGENT_MODEL (642), Databricks token +optional (676), `databricks_v2` selector (939); `llm.rs` effective-model request +routing (85–147). No unmerged `BUZZ_ACP_REQUIRED_MODEL` flag is used. + +**Catalog uncertainty is explicit.** `not-probed`, `reported`, `empty`, `filtered` +and `failed` are distinct, but all currently carry authentication `unverified`: +main's session/new can return a configured-model fallback after an OAuth or +network discovery failure. A reported/empty catalog cannot establish authenticated +availability or distinguish refresh-expired/denied/no-credential. That diagnostic +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. ## Supported / remaining parity @@ -126,7 +159,7 @@ databricks --help` invocation in a scrubbed environment returned | 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 + explicit blocked target | normal OAuth/capability/catalog bridge, exact-model evidence | +| Buzz Agent Databricks v2 | Local setup + TypeScript ACP probe boundary | live OAuth/model proof, full conversation bridge and diagnostic catalog provenance | | Other Buzz Agent providers | Not implemented | Anthropic, OpenAI-compatible, Databricks legacy, OpenRouter | | Goose / Claude Code / Codex | Not implemented | harness-specific local auth and adapters/catalog/launch | | Ten presets / custom ACP | Not implemented | explicit executable/env trust, preset-specific capability grounding | diff --git a/beehive/TESTING.md b/beehive/TESTING.md index 039d848d4c..bd46eeb94b 100644 --- a/beehive/TESTING.md +++ b/beehive/TESTING.md @@ -31,3 +31,30 @@ with missing declaration diagnostics. This is not a passing typecheck or a reproducible clean install. Restore approved dependency access, generate a lock file, run strict check and package tests before proposing merge/production use. No policy bypass or alternate registry workaround was attempted. + +## ACP and independent-review regressions (continuation) + +`npm test` now runs five top-level tests (many table cases) under Hermit Node +24.15.0. `acp.test.ts` launches `acp-fixture.ts` through the actual TypeScript +supervisor and ACP boundary. Cases cover exact session/model acknowledgement, +response hashing, empty/filtered catalog provenance, wrong model/session, rejected +operation (secret-bearing error withheld), cancelled turn, malformed JSON, timeout +and output flood. The host integration uses real encrypted WS commands and proves +Save → Start → same-session fixture evidence → Save → remove isolated auth directory +→ Stop, retaining stopped host assignment. No fixtures call an LLM or read a cache. + +The review at pinned 94d1da0e is in the workspace's +WORK_LOGS/BEEHIVE_DESIGN_183FFAF0/EXECUTABLE_REVIEW_94D1DA0E.md. Its F1/F2/F3 are +addressed with production regressions in slice.test.ts: +- orphan-fixture.ts exits its runner but leaves a non-escaping descendant; a live + group anchor must remain, and both remote Stop and host close kill the group; +- malformed relay URL and initial connection refusal, then successful startup; + second-live-host lock rejection remains tested; +- two actual TUI sessions intentionally quit without printing UNKNOWN. + +The new anchor pins group identity while the external runner is gone. Cleanup is +an IPC request to that living anchor, which signals its own group. No negative-PID +kill from a stale parent child handle is used. Group absence polling is read-only. +Unexpected loss of the anchor still requires local reconciliation and is not a +claim of general OS containment or crash recovery. These fixes and the ACP slice +need independent review on their new head; the old review is not approval of them. diff --git a/beehive/src/acp.ts b/beehive/src/acp.ts new file mode 100644 index 0000000000..510c780459 --- /dev/null +++ b/beehive/src/acp.ts @@ -0,0 +1,154 @@ +import { type ChildProcessWithoutNullStreams } from 'node:child_process'; +import { accessSync, constants, readFileSync, realpathSync, statSync } from 'node:fs'; +import { isAbsolute } from 'node:path'; +import { createHash } from 'node:crypto'; +import { spawnOwned, type OwnedProcess } from './owned.ts'; + +/** Only local provisioning may supply these inputs; never accept remote env/argv. */ +export type AgentLaunch = Readonly<{ executable: string; args: readonly string[]; workspace: string; home: string; configDirectory: string; databricksHost: string; model: 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 }; +const hash = (s: string | Buffer) => createHash('sha256').update(s).digest('hex'); +const record = (v: unknown): Record => { + if (!v || typeof v !== 'object' || Array.isArray(v)) throw Error('Invalid ACP response'); + return v as Record; +}; +const identifier = (v: unknown): string => { + if (typeof v !== 'string' || !/^[a-zA-Z0-9_.:/-]{1,200}$/.test(v)) throw Error('Invalid ACP identifier'); + return v; +}; +/** Validate and snapshot local launch settings. No ambient provider credentials inherited. */ +export function prepareAgent(input: AgentLaunch) { + const plan = Object.freeze({ ...input, args: Object.freeze([...input.args]) }); + for (const p of [plan.executable, plan.workspace, plan.home, plan.configDirectory]) { + if (!isAbsolute(p) || realpathSync(p) !== p) throw Error('Harness setup requires canonical absolute paths'); + } + accessSync(plan.executable, constants.X_OK); + for (const p of [plan.workspace, plan.home, plan.configDirectory]) if (!statSync(p).isDirectory()) throw Error('Harness setup directory unavailable'); + const url = new URL(plan.databricksHost); + if (url.protocol !== 'https:' || url.username || url.password || url.search || url.hash || url.pathname !== '/') throw Error('Databricks workspace must be an HTTPS origin'); + identifier(plan.model); + return Object.freeze({ plan, executableHash: hash(readFileSync(plan.executable)), env: Object.freeze({ PATH: '/usr/bin:/bin', HOME: plan.home, BUZZ_AGENT_CONFIG_DIR: plan.configDirectory, BUZZ_AGENT_PROVIDER: 'databricks_v2', DATABRICKS_HOST: url.origin, BUZZ_AGENT_MODEL: plan.model }) }); +} + +/** Bounded newline JSON-RPC boundary to an external Buzz Agent ACP executable. */ +export class AgentSession { + readonly child: ChildProcessWithoutNullStreams; + readonly owned: OwnedProcess; + readonly prepared: ReturnType; + private sequence = 0; + private pending = new Map void; reject: (e: Error) => void; timer: ReturnType }>(); + private buffer = ''; + private bytes = 0; + private failed = false; + private session = ''; + private response = ''; + private prompting = false; + private timeoutMs: number; + constructor(input: AgentLaunch, timeoutMs = 30_000) { + this.timeoutMs = timeoutMs; + this.prepared = prepareAgent(input); + const { plan, env } = this.prepared; + this.owned = spawnOwned(plan.executable, plan.args, plan.workspace, env); + this.child = this.owned.child; + void this.owned.ready.catch(() => this.fail('Harness executable unavailable')); + this.child.on('message', () => { if (this.owned.exited) this.fail('Harness exited'); }); + this.child.stdin.on('error', () => this.fail('ACP input failed')); + this.child.on('error', () => this.fail('Harness executable unavailable')); + this.child.on('exit', () => this.fail('Harness exited')); + // Never retain or relay raw external diagnostics (may contain credentials). + this.child.stderr.on('data', (chunk: Buffer) => this.account(chunk.length)); + this.child.stdout.setEncoding('utf8'); + this.child.stdout.on('data', (chunk: string) => { + if (!this.account(Buffer.byteLength(chunk))) return; + this.buffer += chunk; + try { + let end: number; + while ((end = this.buffer.indexOf('\n')) >= 0) { + const line = this.buffer.slice(0, end); this.buffer = this.buffer.slice(end + 1); + if (line.trim()) this.receive(record(JSON.parse(line))); + } + } catch { this.fail('Malformed ACP protocol'); } + }); + } + private account(n: number) { + this.bytes += n; + if (this.bytes > 1024 * 1024) this.fail('ACP output limit exceeded'); + return !this.failed; + } + private fail(reason: string) { + if (this.failed) return; + this.failed = true; + // Caller retains the live group anchor for verified Stop, including protocol failures. + for (const p of this.pending.values()) { clearTimeout(p.timer); p.reject(Error(reason)); } + this.pending.clear(); + } + /** Protocol failures invalidate readiness even if the OS process has not exited yet. */ + get healthy() { return !this.failed; } + + private receive(msg: Record) { + if (msg.jsonrpc !== '2.0') throw Error('Invalid RPC version'); + if (typeof msg.method === 'string') { + if (msg.id !== undefined) { + // No filesystem, terminal, permission or other agent-to-client execution grants. + this.child.stdin.write(JSON.stringify({ jsonrpc: '2.0', id: msg.id, error: { code: -32601, message: 'Unsupported client operation' } }) + '\n'); + return; + } + if (msg.method === 'session/update') { + const p = record(msg.params); + if (p.sessionId !== this.session || !this.prompting) return; + const update = record(p.update); + if (update.sessionUpdate === 'agent_message_chunk') { + const content = record(update.content); + if (content.type === 'text' && typeof content.text === 'string') this.response += content.text; + } + } + return; + } + if (typeof msg.id !== 'number') throw Error('Invalid RPC id'); + const p = this.pending.get(msg.id); + if (!p) throw Error('Unexpected RPC response'); + this.pending.delete(msg.id); clearTimeout(p.timer); + if (msg.error !== undefined) p.reject(Error('ACP operation rejected; repair harness setup locally (external diagnostics withheld)')); + else p.resolve(msg.result); + } + private request(method: string, params: unknown): Promise { + if (this.failed) return Promise.reject(Error('ACP session unavailable')); + if (this.pending.size) return Promise.reject(Error('ACP operation already in progress')); + const id = ++this.sequence; + return new Promise((resolve, reject) => { + const timer = setTimeout(() => this.fail('ACP operation timed out'), this.timeoutMs); + this.pending.set(id, { resolve, reject, timer }); + this.child.stdin.write(JSON.stringify({ jsonrpc: '2.0', id, method, params }) + '\n'); + }); + } + /** Discover a session catalog, explicitly retaining its ambiguous authentication provenance. */ + async catalog(): Promise { + await this.owned.ready; + const init = record(await this.request('initialize', { protocolVersion: 1, clientCapabilities: {}, clientInfo: { name: 'beehive', version: '0.0.1' } })); + if (init.protocolVersion !== 1 || record(init.agentInfo).name !== 'buzz-agent') throw Error('Unsupported harness capabilities'); + const created = record(await this.request('session/new', { cwd: this.prepared.plan.workspace, mcpServers: [] })); + this.session = identifier(created.sessionId); + const models = record(created.models); + if (!Array.isArray(models.availableModels) || models.availableModels.length > 1000) throw Error('Invalid ACP catalog'); + // Never publish names/descriptions/raw errors supplied by an external executable. + const ids = [...new Set(models.availableModels.map(m => identifier(record(m).modelId)))]; + return { state: ids.length === 0 ? 'empty' : ids.includes(this.prepared.plan.model) ? 'reported' : 'filtered', models: ids, authentication: 'unverified' }; + } + /** Require explicit exact-model acknowledgement, then a response in the SAME session/process. */ + async verify(): Promise { + if (!this.session) throw Error('Create ACP session first'); + const model = this.prepared.plan.model; + const ack = record(await this.request('session/set_model', { sessionId: this.session, modelId: model })); + if (ack.sessionId !== this.session || ack.modelId !== model) throw Error('Exact-model acknowledgement mismatch'); + this.prompting = true; + this.response = ''; + try { + const reply = record(await this.request('session/prompt', { sessionId: this.session, prompt: [{ type: 'text', text: 'Reply with a short greeting only. Do not use tools or access files.' }] })); + if (reply.stopReason !== 'end_turn' || !this.response.trim()) throw Error('No completed same-session text response'); + return { session: this.session, model, responseHash: hash(this.response), stopReason: 'end_turn', source: 'external-acp-session', executableHash: this.prepared.executableHash }; + } finally { this.prompting = false; this.response = ''; } + } +} diff --git a/beehive/src/cli.ts b/beehive/src/cli.ts index bd8c8d5b4f..293be6ea6f 100644 --- a/beehive/src/cli.ts +++ b/beehive/src/cli.ts @@ -3,15 +3,17 @@ import { stdin, stdout } from 'node:process'; import { existsSync, mkdirSync, realpathSync } from 'node:fs'; import { resolve, join } from 'node:path'; import { relay } from './relay.ts'; -import { host } from './host.ts'; +import { host, validateSetup } from './host.ts'; import { connect } from './client.ts'; import { message, newKey, publicKey, object, text, type Message } from './protocol.ts'; import { readPrivate, writePrivate } from './storage.ts'; +const shellQuote = (s: string) => `'${s.replaceAll("'", "'\"'\"'")}'`; const [command, ...args] = process.argv.slice(2); const help = `Beehive — isolated development preview (loopback relay only) identity Create a NEW local owner identity setup Guided local harness + key setup + auth-info Print exact local sign-in context (no login) relay Dedicated ciphertext relay host Persistent foreground host tui Relay-connected terminal UI @@ -32,17 +34,22 @@ async function main() { const name = text(await ui.question('Host name: ')); const mode = await ui.question('Setup [1 deterministic fixture / 2 Buzz Agent + Databricks v2]: '); if (!['1','2'].includes(mode)) throw Error('Choose 1 or 2'); - const runner = realpathSync(text(await ui.question(mode === '1' ? 'Absolute fixture runner executable: ' : 'Absolute installed buzz-acp executable: '))); + const runner = realpathSync(text(await ui.question(mode === '1' ? 'Absolute fixture runner executable: ' : 'Absolute installed buzz-agent executable: '))); const workspace = realpathSync(text(await ui.question('Allowed workspace (absolute directory): '))); const extra = mode === '1' ? text(await ui.question('Absolute fixture TypeScript script: ')) : ''; const databricksHost = mode === '2' ? text(await ui.question('Databricks workspace HTTPS URL: ')) : undefined; if (databricksHost && new URL(databricksHost).protocol !== 'https:') throw Error('HTTPS workspace required'); if ((await ui.question('Create a NEW agent identity assigned exclusively to this host? [yes/no]: ')) !== 'yes') return; mkdirSync(dir,{ mode: 0o700 }); - writePrivate(join(dir,'setup.json'),{ host: name, ownerSecret: secret, agentSecret: newKey(), runner, args: mode === '1' ? [resolve(extra)] : [], workspace, mode: mode === '1' ? 'fixture' : 'buzz-agent-databricks-v2', ...(databricksHost ? { databricksHost } : {}) }); + if (mode === '2') { mkdirSync(join(dir,'service-home'),{ mode: 0o700 }); mkdirSync(join(dir,'agent-config'),{ mode: 0o700 }); } + writePrivate(join(dir,'setup.json'),{ host: name, ownerSecret: secret, agentSecret: newKey(), runner, args: mode === '1' ? [resolve(extra)] : [], workspace, mode: mode === '1' ? 'fixture' : 'buzz-agent-databricks-v2', ...(databricksHost ? { databricksHost, serviceHome: join(dir,'service-home'), configDirectory: join(dir,'agent-config') } : {}) }); console.log('Local setup saved. Start the host separately; the TUI never owns its lifetime.'); - if (mode === '2') console.log(`Not authenticated or model-ready. Under the dedicated service OS user, with an isolated BUZZ_AGENT_CONFIG_DIR, run:\nDATABRICKS_HOST=${JSON.stringify(databricksHost)} buzz-agent auth databricks\nBuzz Agent owns browser OAuth/cache/refresh. Start is blocked until exact-model evidence is implemented.`); + if (mode === '2') console.log(`Not authenticated. Run auth-info ${shellQuote(dir)} for the exact local host/service-user login command. Start uses an ACP greeting probe, not yet a Buzz relay conversation agent.`); } finally { ui.close(); } + } else if (command === 'auth-info') { + const setup = validateSetup(readPrivate(join(resolve(text(args[0])), 'setup.json'))); + if (setup.mode !== 'buzz-agent-databricks-v2') throw Error('This harness setup has no provider sign-in'); + console.log(`Sign in on host ${setup.host} as the same OS user running its host service (current uid ${process.getuid?.() ?? 'unknown'}), not your Desktop account. Use exactly this context:\nHOME=${shellQuote(text(setup.serviceHome))} BUZZ_AGENT_CONFIG_DIR=${shellQuote(text(setup.configDirectory))} DATABRICKS_HOST=${shellQuote(text(setup.databricksHost))} ${shellQuote(setup.runner)} auth databricks\nBuzz Agent owns OAuth/cache/refresh. No login was initiated; Save and Stop do not require auth.`); } else if (command === 'relay') { const server = await relay(Number(args[0]),text(args[1]),resolve(text(args[2]))); console.log(`Development relay listening ${JSON.stringify(server.address())}`); @@ -63,7 +70,8 @@ async function main() { if (m.type === 'receipt') console.log(`\nHost receipt (may be replayed) ${m.body.operation}: ${JSON.stringify(m.body)}`); }); await client.ready; - client.socket.on('close',() => console.log('\nRelay disconnected: live status and command outcomes UNKNOWN. Reopen TUI.')); + let intentionalClose = false; + client.socket.on('close',() => { if (!intentionalClose) console.log('\nRelay disconnected: live status and unresolved requests UNKNOWN. Received receipts remain valid. Reopen TUI.'); }); const ui = createInterface({ input: stdin, output: stdout }); console.log('Beehive | Hosts → assigned agent → selected-next / actual run\nCommands: hosts, select , show, save, start, stop, quit. Closing this UI does not stop hosts.'); let selected = ''; @@ -87,7 +95,7 @@ async function main() { client.send(request); console.log(`Requested ${line} (${request.id}); publication is NOT host acceptance.`); } - } finally { ui.close(); client.close(); } + } finally { intentionalClose = true; ui.close(); client.close(); console.log('UI closed; host lifetime is independent.'); } } else console.log(help); } main().catch(error => { console.error(error instanceof Error ? error.message : 'Beehive failed'); process.exitCode = 1; }); diff --git a/beehive/src/client.ts b/beehive/src/client.ts index 820e3e4832..0dafd8935f 100644 --- a/beehive/src/client.ts +++ b/beehive/src/client.ts @@ -1,9 +1,13 @@ import { WebSocket } from 'ws'; import { open, seal, type Message } from './protocol.ts'; -/** All UI/host communication uses the real relay, never a host API. */ -export function connect(url: string, secret: string, receive: (m: Message) => void) { +/** Validate transport before acquiring installation resources. */ +export function validateRelayURL(url: string) { const endpoint = new URL(url); if (endpoint.protocol !== 'ws:' || endpoint.hostname !== '127.0.0.1') throw Error('Development transport requires ws://127.0.0.1'); +} +/** All UI/host communication uses the real relay, never a host API. */ +export function connect(url: string, secret: string, receive: (m: Message) => void) { + validateRelayURL(url); const socket = new WebSocket(url, { maxPayload: 70000 }); socket.on('message', data => { try { receive(open(JSON.parse(data.toString()),secret)); } catch { /* Reject malformed/untrusted data, never print payloads. */ } diff --git a/beehive/src/host.ts b/beehive/src/host.ts index acc1068ad0..94016124a1 100644 --- a/beehive/src/host.ts +++ b/beehive/src/host.ts @@ -1,14 +1,15 @@ -import { spawn, type ChildProcess } from 'node:child_process'; import { accessSync, constants, existsSync, mkdirSync, rmdirSync, realpathSync, statSync, readFileSync } from 'node:fs'; import { join, isAbsolute } from 'node:path'; -import { setTimeout as delay } from 'node:timers/promises'; -import { connect } from './client.ts'; +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 { spawnOwned, type OwnedProcess } from './owned.ts'; +import { AgentSession, prepareAgent, type AgentLaunch, type Catalog, type Evidence } from './acp.ts'; -export type Setup = { host: string; ownerSecret: string; agentSecret: string; runner: string; args: string[]; workspace: string; mode: 'fixture' | 'buzz-agent-databricks-v2'; databricksHost?: string }; +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 }; type Selection = { model: string; workspace: string; profile: string }; -type State = { binding: { host: string; owner: string; agent: string }; revision: number; phase: 'stopped' | 'transitioning' | 'running' | 'quarantined'; selected: Selection; actual: null | { selection: Selection; executableHash: string; run: string }; operations: Record; outbox: Message[] }; +type State = { binding: { host: string; owner: string; agent: string }; revision: number; phase: 'stopped' | 'transitioning' | 'running' | 'quarantined'; selected: Selection; actual: null | { selection: Selection; executableHash: string; run: string; evidence?: Evidence }; operations: Record; outbox: Message[] }; export function validateSetup(value: unknown): Setup { const s = object(value); for (const key of ['host','ownerSecret','agentSecret','runner','workspace']) text(s[key]); @@ -18,6 +19,7 @@ export function validateSetup(value: unknown): Setup { } /** Fixed-authority first slice. Keys never constitute a remote assignment grant. */ export async function host(directory: string, url: string) { + validateRelayURL(url); const setup = validateSetup(readPrivate(join(directory,'setup.json'))); const agent = publicKey(setup.agentSecret); const lock = join(directory,'host.lock'); @@ -29,33 +31,27 @@ export async function host(directory: string, url: string) { mkdirSync(lock, { mode: 0o700 }); // No PID-only stale-lock recovery. if (state.phase !== 'stopped') state.phase = 'quarantined'; const save = () => writePrivate(path,state); - save(); - let child: ChildProcess | undefined; + try { save(); } catch (e) { rmdirSync(lock); throw e; } + let owned: OwnedProcess | undefined; + let acp: AgentSession | 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: setup.mode, models: [setup.mode === 'fixture' ? 'fixture-model' : 'databricks-claude-haiku-4-5'], - workspaces: [setup.workspace], profiles: ['default'], readiness: setup.mode === 'fixture' ? 'fixture-only' : 'blocked: exact-model application evidence not implemented', observedAt: Date.now(), + catalog, setup: setup.mode, models: [setup.mode === 'fixture' ? 'fixture-model' : 'databricks-claude-haiku-4-5'], + workspaces: [setup.workspace], profiles: ['default'], readiness: 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; let queue = Promise.resolve(); - const client = connect(url,setup.ownerSecret,m => { queue = queue.then(() => handle(m)).catch(() => { state.phase = 'quarantined'; save(); }); }); + let initialized = false; + let client: ReturnType; + try { client = connect(url,setup.ownerSecret,m => { if (initialized) queue = queue.then(() => handle(m)).catch(() => { state.phase = 'quarantined'; save(); }); }); } + catch (e) { rmdirSync(lock); throw e; } const publish = (m: Message) => { try { client.send(m); } catch { /* Durable outbox retained for reconnect/restart. */ } }; async function stopOwned() { - if (!child?.pid) throw Error('Unknown ownership; quarantined'); - const pid = child.pid; - // Child handle from this host lifetime is required; never recover ownership from PID. - const exited = child.exitCode !== null || child.signalCode !== null; - if (!exited) process.kill(-pid,'SIGTERM'); - for (let i = 0; i < 100; i++) { - try { process.kill(-pid,0); } catch (e) { - if ((e as NodeJS.ErrnoException).code === 'ESRCH') { child = undefined; return; } - throw e; - } - await delay(25); - } - throw Error('Owned process group did not exit; quarantined'); + if (!owned) throw Error('Unknown ownership; quarantined'); + await owned.stop(); owned = undefined; acp = undefined; } async function handle(m: Message) { if (closing || m.host !== setup.host || !['inspect','save','start','stop'].includes(m.type)) return; @@ -68,7 +64,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') result = 'quarantined'; + else if ((state.phase === 'quarantined' || state.phase === 'transitioning') && !(m.type === 'stop' && 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' }); @@ -82,13 +78,28 @@ export async function host(directory: string, url: string) { } else if (m.type === 'start') { fields(m.body,[]); if (state.phase !== 'stopped') throw Error('Already running'); - if (setup.mode !== 'fixture') throw Error('Exact-model application evidence unavailable'); accessSync(setup.runner,constants.X_OK); if (realpathSync(setup.workspace) !== setup.workspace || !statSync(setup.workspace).isDirectory()) throw Error('Workspace changed'); - state.actual = { selection: structuredClone(state.selected), executableHash: digest(readFileSync(setup.runner).toString('base64')).toString('hex'), run: m.id }; + const launch: AgentLaunch | undefined = setup.mode === 'fixture' ? undefined : { executable: setup.runner, args: setup.args, workspace: state.selected.workspace, home: text(setup.serviceHome), configDirectory: text(setup.configDirectory), databricksHost: text(setup.databricksHost), model: state.selected.model }; + if (launch) prepareAgent(launch); // Prerequisite failure preserves stopped state. + state.actual = { selection: structuredClone(state.selected), executableHash: createHash('sha256').update(readFileSync(setup.runner)).digest('hex'), run: m.id }; state.phase = 'transitioning'; save(); - child = spawn(setup.runner,setup.args,{ cwd: state.selected.workspace, detached: true, stdio: 'ignore', env: { PATH: '/usr/bin:/bin', BEEHIVE_FIXTURE: '1' } }); - await new Promise((resolve,reject) => { child!.once('spawn',resolve); child!.once('error',reject); }); + if (setup.mode === 'fixture') { + 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 { + const session = new AgentSession(launch!); acp = session; + owned = session.owned; + try { + catalog = await session.catalog(); + state.actual.evidence = await session.verify(); + } catch (e) { + catalog = { state: 'failed', authentication: 'unverified' }; + await stopOwned(); state.phase = 'stopped'; state.actual = null; + throw e; + } + } state.phase = 'running'; } else { fields(m.body,[]); @@ -106,17 +117,23 @@ export async function host(directory: string, url: string) { Object.defineProperty(state.operations,m.id,{ value: { fingerprint, reply }, enumerable: true, configurable: true, writable: true }); state.outbox.push(reply); save(); publish(reply); publish(inventory()); } - await client.ready; + try { await client.ready; } catch (e) { client.close(); rmdirSync(lock); throw e; } + initialized = true; for (const m of state.outbox) publish(m); publish(inventory()); const heartbeat = setInterval(() => { - if (child && (child.exitCode !== null || child.signalCode !== null) && state.phase === 'running') { state.phase = 'quarantined'; save(); } + if (owned && (owned.exited || acp?.healthy === false) && state.phase === 'running') { state.phase = 'quarantined'; save(); } publish(inventory()); },2000); return { agent, async close() { if (closing) return; closing = true; client.close(); clearInterval(heartbeat); await queue; - if (state.phase === 'running') { state.phase = 'transitioning'; save(); try { await stopOwned(); state.phase = 'stopped'; state.actual = null; } catch { state.phase = 'quarantined'; } } - save(); client.close(); rmdirSync(lock); + if (owned) { + state.phase = 'transitioning'; save(); + try { await stopOwned(); state.phase = 'stopped'; state.actual = null; } + catch (e) { state.phase = 'quarantined'; save(); throw e; } + } + save(); client.close(); + if (state.phase === 'stopped') rmdirSync(lock); // Failed teardown retains recovery fence. } }; } diff --git a/beehive/src/owned.ts b/beehive/src/owned.ts new file mode 100644 index 0000000000..6ced27ca02 --- /dev/null +++ b/beehive/src/owned.ts @@ -0,0 +1,47 @@ +import { spawn, type ChildProcessWithoutNullStreams } from 'node:child_process'; +import { fileURLToPath } from 'node:url'; +import { setTimeout as delay } from 'node:timers/promises'; + +/** In-lifetime group ownership: a live TS anchor retains the group ID until its own Stop. */ +export function spawnOwned(executable: string, args: readonly string[], cwd: string, env: NodeJS.ProcessEnv) { + const child = spawn(process.execPath, [fileURLToPath(new URL('./supervisor.ts', import.meta.url))], { + cwd, env: { PATH: '/usr/bin:/bin' }, detached: true, stdio: ['pipe', 'pipe', 'pipe', 'ipc'], + }) as ChildProcessWithoutNullStreams; + let runnerExited = false; + let spawned = false; + let rejectReady: (e: Error) => void = () => {}; + const ready = new Promise((resolve, reject) => { + rejectReady = reject; + child.once('error', () => reject(Error('Group supervisor unavailable'))); + child.once('exit', () => { runnerExited = true; reject(Error('Group supervisor exited')); }); + child.on('message', (m: unknown) => { + const type = (m as { type?: string }).type; + if (type === 'spawned') { spawned = true; resolve(); } + if (type === 'runner-error' || type === 'runner-exit') { + runnerExited = true; + if (!spawned) reject(Error('External harness failed to launch')); + } + }); + child.once('spawn', () => child.send({ type: 'launch', executable, args: [...args], cwd, env }, error => { if (error) reject(Error('Group launch channel failed')); })); + }); + // Readiness deadline bounds even a supervisor that never acknowledges launch. + const timer = setTimeout(() => rejectReady(Error('Group launch timed out')), 5000); + void ready.then(() => clearTimeout(timer), () => clearTimeout(timer)); + return { + child, ready, + get exited() { return runnerExited; }, + async stop() { + const pid = child.pid; + if (!pid || !child.connected) throw Error('Live group anchor unavailable; reconcile locally'); + await new Promise((resolve, reject) => child.send({ type: 'stop' }, error => error ? reject(Error('Owned Stop channel failed')) : resolve())); + for (let i = 0; i < 100; i++) { + try { process.kill(-pid, 0); } + catch (e) { if ((e as NodeJS.ErrnoException).code === 'ESRCH') return; throw e; } + await delay(25); + } + throw Error('Owned process group did not exit; quarantined'); + }, + }; +} +/** Only the returned in-memory handle authorizes Stop; no reconstruction from journal PIDs. */ +export type OwnedProcess = ReturnType; diff --git a/beehive/src/supervisor.ts b/beehive/src/supervisor.ts new file mode 100644 index 0000000000..1a8afbeddd --- /dev/null +++ b/beehive/src/supervisor.ts @@ -0,0 +1,21 @@ +/** Internal POSIX group anchor. Never invoke directly; the parent provisions one launch via IPC. */ +import { spawn } from 'node:child_process'; + +if (!process.send) throw Error('Supervisor requires a private IPC channel'); +let launched = false; +const keepalive = setInterval(() => {}, 1000); +process.on('message', (message: unknown) => { + const m = message as { type?: string; executable?: string; args?: string[]; cwd?: string; env?: NodeJS.ProcessEnv }; + if (m.type === 'stop') { + // The group leader signals ITS OWN group: never a recovered or reused numeric identity. + process.kill(-process.pid, 'SIGTERM'); + } else if (m.type === 'launch' && !launched && m.executable) { + launched = true; + const runner = spawn(m.executable, m.args ?? [], { cwd: m.cwd, env: m.env, stdio: ['inherit', 'inherit', 'inherit'] }); + runner.once('spawn', () => process.send?.({ type: 'spawned' })); + runner.once('error', () => process.send?.({ type: 'runner-error' })); + runner.once('exit', () => process.send?.({ type: 'runner-exit' })); + } +}); +// Parent loss tears down the known group, without interpreting a persisted PID. +process.once('disconnect', () => { clearInterval(keepalive); process.kill(-process.pid, 'SIGTERM'); }); diff --git a/beehive/test/acp-fixture.ts b/beehive/test/acp-fixture.ts new file mode 100644 index 0000000000..88aca365ad --- /dev/null +++ b/beehive/test/acp-fixture.ts @@ -0,0 +1,25 @@ +/** Deterministic EXTERNAL ACP fixture: never calls a provider or reads credentials. */ +import { createInterface } from 'node:readline'; +const mode = process.argv[2] ?? 'ok'; +let selected = ''; +const send = (v: unknown) => process.stdout.write(JSON.stringify(v) + '\n'); +const sessionId = 'fixture-session'; +for await (const line of createInterface({ input: process.stdin })) { + const m = JSON.parse(line); + if (mode === 'timeout') continue; + if (mode === 'malformed') { console.log('not-json'); continue; } + if (mode === 'flood') { console.log('x'.repeat(1024 * 1024 + 1)); continue; } + let result: unknown; + if (m.method === 'initialize') result = { protocolVersion: 1, agentInfo: { name: 'buzz-agent' } }; + else if (m.method === 'session/new') result = { sessionId, models: { currentModelId: 'ignored-default', availableModels: mode === 'empty' ? [] : [{ modelId: mode === 'filtered' ? 'other-model' : 'databricks-claude-haiku-4-5', name: 'DO-NOT-RELAY-SECRET' }] } }; + else if (m.method === 'session/set_model') { + selected = m.params.modelId; + if (mode === 'reject') { send({ jsonrpc: '2.0', id: m.id, error: { code: -1, message: 'DO-NOT-RELAY-SECRET' } }); continue; } + result = { sessionId, modelId: mode === 'wrong-model' ? 'other-model' : selected }; + } else if (m.method === 'session/prompt') { + if (selected !== process.env.BUZZ_AGENT_MODEL || m.params.sessionId !== sessionId || process.env.DATABRICKS_TOKEN || process.env.BUZZ_PRIVATE_KEY || process.env.BUZZ_AGENT_PROVIDER !== 'databricks_v2') process.exit(8); + send({ jsonrpc: '2.0', method: 'session/update', params: { sessionId: mode === 'wrong-session' ? 'other-session' : sessionId, update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'Fixture greeting, not live model proof.' } } } }); + result = { stopReason: mode === 'cancelled' ? 'cancelled' : 'end_turn' }; + } else process.exit(9); + send({ jsonrpc: '2.0', id: m.id, result }); +} diff --git a/beehive/test/acp.test.ts b/beehive/test/acp.test.ts new file mode 100644 index 0000000000..8f94e543ff --- /dev/null +++ b/beehive/test/acp.test.ts @@ -0,0 +1,95 @@ +import { test } from 'node:test'; +import { spawnSync } from 'node:child_process'; +import assert from 'node:assert/strict'; +import { mkdtempSync, realpathSync, rmSync, mkdirSync, readFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join, resolve } from 'node:path'; +import { setTimeout as delay } from 'node:timers/promises'; +import { AgentSession, prepareAgent, type AgentLaunch } from '../src/acp.ts'; +import { host } from '../src/host.ts'; +import { relay } from '../src/relay.ts'; +import { connect } from '../src/client.ts'; +import { message, newKey, publicKey, type Message } from '../src/protocol.ts'; +import { writePrivate } from '../src/storage.ts'; + +const fixture = resolve('test/acp-fixture.ts'); +function plan(dir: string, mode = 'ok'): AgentLaunch { + return { executable: realpathSync(process.execPath), args: [fixture, mode], workspace: dir, home: dir, configDirectory: dir, databricksHost: 'https://fixture.invalid', model: 'databricks-claude-haiku-4-5' }; +} +async function cleanup(s: AgentSession) { await s.owned.stop(); } +test('external ACP boundary: snapshot, catalog provenance, exact-model same-session evidence and failures', async () => { + const dir = realpathSync(mkdtempSync(join(tmpdir(), 'beehive-acp-'))); + try { + const input = plan(dir); const prepared = prepareAgent(input); + (input.args as string[]).push('mutated'); + assert.equal(prepared.plan.args.length, 2); + assert.equal(Object.hasOwn(prepared.env, 'DATABRICKS_TOKEN'), false); + assert.equal(Object.hasOwn(prepared.env, 'BUZZ_PRIVATE_KEY'), false); + assert.throws(() => prepareAgent({ ...plan(dir), databricksHost: 'https://user:secret@example.com' })); + for (const mode of ['ok', 'empty', 'filtered', 'wrong-model', 'wrong-session', 'cancelled', 'reject', 'malformed', 'timeout', 'flood']) { + const s = new AgentSession(plan(dir, mode), 1500); + try { + if (['malformed', 'timeout', 'flood'].includes(mode)) { await assert.rejects(s.catalog()); continue; } + const catalog = await s.catalog(); + assert.equal(catalog.authentication, 'unverified'); + assert.equal(catalog.state, mode === 'empty' ? 'empty' : mode === 'filtered' ? 'filtered' : 'reported'); + assert.ok(!JSON.stringify(catalog).includes('DO-NOT-RELAY')); + if (['wrong-model', 'wrong-session', 'cancelled', 'reject'].includes(mode)) { + await assert.rejects(s.verify(), e => e instanceof Error && !e.message.includes('DO-NOT-RELAY')); + } else { + const evidence = await s.verify(); + assert.equal(evidence.model, plan(dir).model); + assert.equal(evidence.session, 'fixture-session'); + assert.match(evidence.responseHash, /^[0-9a-f]{64}$/); + assert.equal(evidence.executableHash, s.prepared.executableHash); + } + } finally { await cleanup(s); } + } + } finally { rmSync(dir, { recursive: true, force: true }); } +}); + +test('production host Start uses external ACP evidence; Save and Stop do not need auth', async () => { + const dir = realpathSync(mkdtempSync(join(tmpdir(), 'beehive-acp-host-'))); + const hostDir = join(dir, 'host'); mkdirSync(hostDir); + const ownerSecret = newKey(); const agentSecret = newKey(); const agent = publicKey(agentSecret); + const p = plan(dir); + const authDir = join(dir, 'isolated-auth'); mkdirSync(authDir); + writePrivate(join(hostDir, 'setup.json'), { host: 'acp-host', ownerSecret, agentSecret, runner: p.executable, args: p.args, workspace: dir, mode: 'buzz-agent-databricks-v2', serviceHome: dir, configDirectory: authDir, databricksHost: p.databricksHost }); + const authInfo = spawnSync(process.execPath, ['src/cli.ts', 'auth-info', hostDir], { encoding: 'utf8', timeout: 3000 }); + assert.equal(authInfo.status, 0); assert.match(authInfo.stdout, /acp-host/); assert.match(authInfo.stdout, /auth databricks/); + assert.ok(!authInfo.stdout.includes(ownerSecret) && !authInfo.stdout.includes(agentSecret)); + const server = await relay(0, publicKey(ownerSecret), join(dir, 'relay.json')); + const address = server.address(); assert.ok(address && typeof address !== 'string'); + const url = `ws://127.0.0.1:${address.port}`; + const h = await host(hostDir, url); + const seen: Message[] = []; + const client = connect(url, ownerSecret, m => seen.push(m)); await client.ready; + async function request(type: 'start' | 'save' | 'stop', revision: number, body = {}) { + const m = message(type, 'acp-host', agent, revision, body); client.send(m); + for (let i = 0; i < 200; i++) { + const r = seen.find(r => r.type === 'receipt' && r.body.operation === m.id); + if (r) return r; + await delay(25); + } + throw Error('No host receipt'); + } + try { + assert.equal((await request('save', 0, { model: p.model, workspace: dir, profile: 'default' })).body.result, 'saved; running configuration unchanged'); + assert.equal((await request('start', 1)).body.result, 'accepted'); + const state = JSON.parse(readFileSync(join(hostDir, 'journal.json'), 'utf8')); + assert.equal(state.actual.evidence.model, p.model); + assert.equal(state.actual.evidence.session, 'fixture-session'); + assert.equal(state.actual.selection.model, p.model); + assert.equal(state.actual.executableHash, state.actual.evidence.executableHash); + rmSync(authDir, { recursive: true }); + assert.equal((await request('save', 2, { model: p.model, workspace: dir, profile: 'default' })).body.result, 'saved; running configuration unchanged'); + // The auth directory is already absent; both Save and Stop remain usable. + assert.equal((await request('stop', 3)).body.result, 'accepted'); + const stopped = JSON.parse(readFileSync(join(hostDir, 'journal.json'), 'utf8')); + assert.equal(stopped.binding.host, 'acp-host'); assert.equal(stopped.phase, 'stopped'); + assert.ok(!readFileSync(join(dir, 'relay.json'), 'utf8').includes('fixture-session')); + } finally { + await h.close(); client.close(); for (const c of server.clients) c.terminate(); + await new Promise(resolve => server.close(() => resolve())); rmSync(dir, { recursive: true, force: true }); + } +}); diff --git a/beehive/test/orphan-fixture.ts b/beehive/test/orphan-fixture.ts new file mode 100644 index 0000000000..6eaef3fc37 --- /dev/null +++ b/beehive/test/orphan-fixture.ts @@ -0,0 +1,12 @@ +/** Non-escaping child survives its parent, to exercise owned quarantine teardown. */ +import { spawn } from 'node:child_process'; +import { writeFileSync } from 'node:fs'; +import { join } from 'node:path'; +if (process.argv[2] === 'child') setInterval(() => {}, 1000); +else { + const child = spawn(process.execPath, [import.meta.filename, 'child'], { stdio: 'ignore' }); + writeFileSync(join(process.cwd(), 'descendant.pid'), String(child.pid)); + writeFileSync(join(process.cwd(), 'anchor.pid'), String(process.ppid)); + child.unref(); + setTimeout(() => process.exit(0), 500); +} diff --git a/beehive/test/slice.test.ts b/beehive/test/slice.test.ts index 61f8b2ce45..826159957b 100644 --- a/beehive/test/slice.test.ts +++ b/beehive/test/slice.test.ts @@ -30,6 +30,11 @@ test('real websocket relay, external runner, durable retry, UI close/reopen, hos const server = await relay(0,publicKey(secret),log); const address = server.address(); assert.ok(address && typeof address !== 'string'); const url = `ws://127.0.0.1:${address.port}`; + await assert.rejects(host(hostDir, 'http://127.0.0.1:1'), /Development transport/); + const unused = await relay(0,publicKey(secret),join(dir,'unused.json')); + const unavailable = unused.address(); assert.ok(unavailable && typeof unavailable !== 'string'); + await new Promise(resolve => unused.close(() => resolve())); + await assert.rejects(host(hostDir, `ws://127.0.0.1:${unavailable.port}`), /ECONNREFUSED/); const running = await host(hostDir,url); await assert.rejects(host(hostDir,url),/EEXIST/); let seen: Message[] = []; @@ -59,7 +64,7 @@ test('real websocket relay, external runner, durable retry, UI close/reopen, hos } }); const code = await new Promise(resolve => terminal.once('exit',resolve)); - assert.equal(code,0); assert.match(output,/actualRun/); assert.match(output,/running/); + assert.equal(code,0); assert.match(output,/actualRun/); assert.match(output,/running/); assert.doesNotMatch(output,/UNKNOWN/); } ui.close(); seen = []; ui = connect(url,secret,m => seen.push(m)); await ui.ready; @@ -89,3 +94,38 @@ test('real websocket relay, external runner, durable retry, UI close/reopen, hos rmSync(dir,{recursive:true,force:true}); } }); + +test('quarantined leader exit retains Stop and close teardown for owned descendants', async () => { + for (const action of ['stop', 'close']) { + const dir = realpathSync(mkdtempSync(join(tmpdir(), 'beehive-orphan-'))); + const secret = newKey(); const agentSecret = newKey(); const agent = publicKey(agentSecret); + const hd = join(dir, 'host'); mkdirSync(hd); + writePrivate(join(hd, 'setup.json'), { host: 'orphan-host', ownerSecret: secret, agentSecret, runner: process.execPath, args: [resolve('test/orphan-fixture.ts')], workspace: dir, mode: 'fixture' }); + 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}`; + const h = await host(hd, url); const seen: Message[] = []; + const ui = connect(url, secret, m => seen.push(m)); await ui.ready; + async function wait(predicate: (m: Message) => boolean) { + for (let i = 0; i < 200; i++) { const m = seen.find(predicate); if (m) return m; await delay(25); } + throw Error('Missing quarantine/teardown receipt'); + } + try { + ui.send(message('start', 'orphan-host', agent, 0)); + await wait(m => m.type === 'inventory' && m.body.phase === 'quarantined'); + const pid = Number(readFileSync(join(dir, 'descendant.pid'), 'utf8')); + process.kill(pid, 0); + const anchor = Number(readFileSync(join(dir, 'anchor.pid'), 'utf8')); + assert.notEqual(anchor, process.pid); process.kill(anchor, 0); // Live anchor pins the group after runner exit. + if (action === 'stop') { + const stop = message('stop', 'orphan-host', agent, 1); ui.send(stop); + const receipt = await wait(m => m.type === 'receipt' && m.body.operation === stop.id); + assert.equal(receipt.body.result, 'accepted'); + } else await h.close(); + assert.throws(() => process.kill(pid, 0), (e: unknown) => (e as NodeJS.ErrnoException).code === 'ESRCH'); + } finally { + await h.close(); ui.close(); for (const c of server.clients) c.terminate(); + await new Promise(resolve => server.close(() => resolve())); rmSync(dir, { recursive: true, force: true }); + } + } +});