mirror of
https://github.com/block/buzz.git
synced 2026-11-09 02:41:57 +00:00
feat(beehive): add external ACP probe and anchored host teardown
Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
@@ -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 <host-dir>` 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 <loganj@squareup.com> 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.
|
||||
|
||||
@@ -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=<isolated-service-config-dir> \
|
||||
DATABRICKS_HOST=https://<your-workspace> 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 |
|
||||
|
||||
@@ -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.
|
||||
|
||||
154
beehive/src/acp.ts
Normal file
154
beehive/src/acp.ts
Normal file
@@ -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<string, unknown> => {
|
||||
if (!v || typeof v !== 'object' || Array.isArray(v)) throw Error('Invalid ACP response');
|
||||
return v as Record<string, unknown>;
|
||||
};
|
||||
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<typeof prepareAgent>;
|
||||
private sequence = 0;
|
||||
private pending = new Map<number, { resolve: (v: unknown) => void; reject: (e: Error) => void; timer: ReturnType<typeof setTimeout> }>();
|
||||
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<string, unknown>) {
|
||||
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<unknown> {
|
||||
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<Catalog> {
|
||||
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<Evidence> {
|
||||
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 = ''; }
|
||||
}
|
||||
}
|
||||
@@ -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 <new-directory> Create a NEW local owner identity
|
||||
setup <new-host-directory> <identity-file> Guided local harness + key setup
|
||||
auth-info <host-directory> Print exact local sign-in context (no login)
|
||||
relay <port> <owner-public-key> <log-file> Dedicated ciphertext relay
|
||||
host <host-directory> <ws://127.0.0.1:port> Persistent foreground host
|
||||
tui <identity-file> <ws://127.0.0.1:port> 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 <host>, 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; });
|
||||
|
||||
@@ -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. */ }
|
||||
|
||||
@@ -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<string, { fingerprint: string; reply: Message }>; 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<string, { fingerprint: string; reply: Message }>; 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<typeof connect>;
|
||||
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<void>((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.
|
||||
} };
|
||||
}
|
||||
|
||||
47
beehive/src/owned.ts
Normal file
47
beehive/src/owned.ts
Normal file
@@ -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<void>((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<void>((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<typeof spawnOwned>;
|
||||
21
beehive/src/supervisor.ts
Normal file
21
beehive/src/supervisor.ts
Normal file
@@ -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'); });
|
||||
25
beehive/test/acp-fixture.ts
Normal file
25
beehive/test/acp-fixture.ts
Normal file
@@ -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 });
|
||||
}
|
||||
95
beehive/test/acp.test.ts
Normal file
95
beehive/test/acp.test.ts
Normal file
@@ -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<void>(resolve => server.close(() => resolve())); rmSync(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
12
beehive/test/orphan-fixture.ts
Normal file
12
beehive/test/orphan-fixture.ts
Normal file
@@ -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);
|
||||
}
|
||||
@@ -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<void>(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<void>(resolve => server.close(() => resolve())); rmSync(dir, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user