feat(beehive): broker identity-bearing ACP conversations under host ownership

Signed-off-by: Logan Johnson <loganj@squareup.com>

beehive/CHECKPOINT.md

@@ -1,91 +1,93 @@
 # Beehive executable checkpoint
 
 Worktree `/Users/loganj/.buzz/REPOS/beehive-cbfd9440`, branch `beehive/cbfd9440`.
-Base `051c3a270be9c73da9ab06700bcab7d5552fceaa`; previous reviewed milestone
+Base `051c3a270be9c73da9ab06700bcab7d5552fceaa`; this continuation starts at
-`6bd859479503832a331cb9f8eb0a495082c8f9d8`. This continuation is the commit
+`9ff9c6cf16f8c5d7cae0b95e4c01cb3820ff833e`. Candidate is the commit containing
-containing this checkpoint (`git rev-parse HEAD`). No push/PR/merge/release.
+this checkpoint (`git rev-parse HEAD`). No push/PR/merge/release.
 
-## Outcome and exact remaining critical gap
+## Implemented conversation boundary — not full delivery
 
-**The requested identity-bearing relay conversation is NOT delivered.** The
+**The host-owned TypeScript ACP broker is implemented and host Start is wired.**
-existing executable path is still a retained anonymous ACP greeting probe.
+`conversation-setup` still attaches to the existing persistent agent identity,
-Do not call its evidence a Buzz conversation, provider attestation or live-model
+never regenerating keys or initiating OAuth. External buzz-acp receives that key,
-proof. New `conversation-setup <existing-host-dir>` provisions the external
+explicit owner, separate conversation relay, optional local attestation and model.
-buzz-acp contract onto the SAME existing key, atomically under the host exclusion
+It retains all Buzz authentication/member/mention/dispatch/reply/tool semantics.
-lock. It does not create a key, initiate auth or connect to any conversation relay.
+There is no second Buzz chat runtime or remotely callable host API.
-Configured conversation Start fails BEFORE spawning or committing actual-run state.
-Inventory and CLI explicitly say blocked; Save/Stop remain credential-neutral.
 
-`src/conversation.ts` validates and snapshots the external executable, selected
+`src/broker.ts` opens a fresh private temporary directory/Unix socket with a random
-Buzz Agent launch, conversation relay, exact existing agent secret, owner pubkey,
+per-run capability. buzz-acp launches `acp-shim.ts` with that capability; the shim
-optional host-local opaque NIP-OA tag, exact model, owner-only inbound gate and
+is childless and only transports stdio. The actual harness is launched by the
-local default permission mode. Env is allowlisted; identity and relay are
+host via existing `spawnOwned`, NOT by the escaping upstream process group.
-explicitly authoritative. No owner secret, arbitrary env or remote paths/argv are
+Capabilities cannot cross runs or supply executable/argv/env/key choices. Session
-accepted. Upstream comma-delimited args reject lossy comma/NUL/empty elements.
+workspace is fixed; nonempty MCP provisioning is rejected (see delivery gap).
-Remote summary contains no credential/executable path. This is a tested launch
+Socket authentication deadlines, connection/process/session/request/frame bounds
-PLAN, not a launched identity-bearing session. CLI currently asks no attestation;
+and write-pressure failure prevent unbounded buffering. No raw diagnostics,
-community operator must admit the public key and owner to intended channels, or
+response text, keys, attestation or capability enter management observations.
-locally provision a valid NIP-OA tag. Neither that action nor live OAuth occurred.
 
-## Source-grounded runtime decision (not a second agent runtime)
+The proxy forwards bidirectional requests/responses/notifications, including
+cancellation, reverse client RPC and session/new/load/resume. Before releasing a
+created/loaded/resumed session, it injects `session/set_model` and requires exact
+session/model response. It rejects conflicting/rejected/unconfirmed model changes,
+including model config choices, and re-verifies after config changes. Prompts
+require a confirmed session; evidence hashes actual same-session response chunks
+and requires uncancelled `end_turn`. Evidence identifies agent pubkey, harness and
+external runtime executable hashes, session and model. This is trusted-executable
+protocol evidence, NOT provider attestation or relay delivery confirmation.
 
-Pinned upstream `051c3a2`:
+Start waits up to 30 seconds for an actual inbound conversation and completed
-- `desktop/src-tauri/src/managed_agents/runtime.rs::spawn_agent_child` passes
+response; it never manufactures a greeting on this path. Save stays selected-next;
-  BUZZ_PRIVATE_KEY, BUZZ_RELAY_URL, BUZZ_AUTH_TAG, external agent command/args,
+Stop/config remain credential-neutral. Missing admission/sign-in yields a bounded
-  model/provider/effort. No BUZZ_ACP_REQUIRED_MODEL contract exists here.
+Start failure, not a running agent claim. Generic non-conversation setups still
-- `crates/buzz-acp/src/config.rs::CliArgs` defines these external env/CLI inputs.
+run the explicitly labelled old anonymous ACP probe.
-- `crates/buzz-acp/src/lib.rs::resolve_agent_owner` verifies NIP-OA against the
-  agent pubkey or falls back to explicit owner. Startup uses HarnessRelay::connect
-  with that key/auth tag, then the actual inbound author/member gate.
-- `crates/buzz-acp/src/pool.rs::apply_model_switch` treats application rejection
-  as Rejected and proceeds with the default. Model env alone is insufficient.
-- **Critical containment seam:** `crates/buzz-acp/src/acp.rs::AcpClient::spawn`
-  calls `cmd.process_group(0)` on Unix. Replacing the probe with buzz-acp under
-  the current outer anchor would let the ACP subtree escape owned Stop (F1).
 
-Reuse existing buzz-acp for all Nostr/owner/membership/mention/reply/delivery
+## Owned teardown and review regressions
-semantics. Do not invent a chat protocol or second Buzz dispatcher. Smallest
-coherent next implementation: host-owned TypeScript ACP stdio broker/proxy so the
-actual adapter is spawned under a host-retained live anchor, not as an escaping
-buzz-acp descendant; the buzz-acp-launched TS shim only forwards stdio. Bound and
-close every shim/broker session on Stop, retain adapter ownership after parent
-exit, enforce session/model acknowledgement, and capture prompt completion on
-THAT identity-bearing conversation session. This broker is NOT implemented yet.
-The current guard deliberately prevents unsafe direct launch in the meantime.
 
-Management transport remains the dedicated loopback TypeScript ciphertext WS
+Stop/cancel/failed Start closes sockets (escaping childless shims exit on EOF),
-protocol, NOT a Nostr/Buzz conversation relay. No existing clients or Rust changed.
+then stops runtime and every harness through retained live group anchors. Shim
+PID checks are absence observations only, NEVER kill authority. Failure to confirm
+any group/shim exit fails closed. Existing supervisor retains ownership through
+500ms TERM -> own-group KILL, including host IPC loss; numeric journal PIDs grant
+no takeover/kill authority. Arbitrary executables that deliberately escape their
+owned group remain outside this POSIX containment claim.
 
-## Independent review integrated
+D1 resistant descendants remain tested. D2 replacement host conversation path now
+shares exact-authority/revision/body-admitted cancellation outside the mutation
+queue; wrong agent/revision/body cannot cancel. Host close interrupts Start. D3
+coalesced completion plus malformed tail cannot commit readiness. These regressions
+are exercised for both old probe and new conversation broker paths. No independent
+review of this delta has occurred; prior independent frozen-9ff9c6cf work was not
+used as approval of this new implementation.
 
-Read `WORK_LOGS/BEEHIVE_DESIGN_183FFAF0/EXECUTABLE_REVIEW_6BD85947.md`.
+## Actual executable evidence and precise remaining gap
-Earlier F1/F2/F3 remain covered. This continuation addresses its D1–D3:
-- D1: live supervisor ignores its own TERM, remains through a 500 ms graceful
-  interval, then sends KILL to its OWN still-owned group. Host only polls absence;
-  transient EPERM is not success and is retried within the same bounded wait.
-  IPC disconnect/send failure invokes the same cleanup rather than killing the
-  anchor prematurely. Resistant orphan descendants covered through remote Stop,
-  host close and parent IPC loss. Genuine anchor loss still fails closed.
-- D2: exact-authority/current-revision/body-valid Stop signals ACP cancellation
-  outside the serialized mutation queue. Close cancels before waiting. Cancelled
-  Start fails; queued Stop retains ordinary receipt/admission handling. Delayed
-  responses cannot commit evidence after failure. Wrong agent/revision/body does
-  not cancel. No blanket stale-revision bypass or recovered-PID kill added.
-- D3: request promise commit and evidence commit recheck session health; host
-  rechecks before running. Coalesced completed-response + malformed-tail regression
-  fails at both ACP and host production seams, rather than accepting then waiting
-  for heartbeat quarantine.
 
-This candidate has NOT received independent delta re-review. Report fixtures
+Automated host -> external runtime fixture -> detached/process_group(0)-style
-remain independent; no review artifacts modified.
+shim -> owned ACP harness passes using persistent identity, exact model, same
+conversation completion, bidirectional RPC, session loading, cancellation and
+complete runtime/shim/harness/TERM-resistant-descendant absence after Stop. The
+external runtime fixture models the process/ACP contract, not a Buzz relay.
 
-## Dependency route and validation
+Also exercised installed `/Applications/Buzz.app/Contents/MacOS/buzz-acp`, SHA256
+`10612d0025d1420bdd9e9afcc2e7129377da2414e75049a8b640e81d857441ea`, against an
+isolated NIP-42/REST/NIP-29 fixture relay and the owned ACP fixture harness. It
+signed verified AUTH with a fresh fixture agent key, discovered fixture membership,
+subscribed, consumed a signed owner mention, and completed the exact-model same
+conversation through the production broker. **Installed executable subscribes to
+kind 9 (plus 46010/40007), not kind 40002**; the fixture follows its actual kind 9
+subscription. This is evidence for that installed binary, not current Buzz relay
+compatibility/admission. Test never touches owner keys/cache/profile or providers.
 
-Read and applied `WORK_LOGS/BEEHIVE_DESIGN_183FFAF0/DEPENDENCY_INSTALLATION.md`.
+**No signed threaded response publication was observed.** ACP text completion is
-Public npm registry is host-policy blocked; approved Block mirror is the supported
+not automatic Buzz message delivery here. The current launch plan explicitly sets
-route, not a policy bypass. Borrowed ignored modules removed. A standalone
+`BUZZ_ACP_MCP_COMMAND=''`; no locally authorized MCP/tool executable plan is
-registry-agnostic `pnpm-lock.yaml` now pins all seven packages with integrity.
+provisioned. The broker deliberately rejects shim-supplied MCP commands/env rather
-No global/user npm/Git config or machine-path dependency committed. Use:
+than permitting arbitrary executable injection. Tool-backed conversation delivery
+therefore remains a concrete required engineering step, not just an OAuth gate.
+Do not claim the whole conversation/delivery slice or owner build is complete.
+
+## Validation / dependencies / next action
+
+Hermit Node24.15.0 / pinned pnpm11.4.0. Standalone registry-agnostic lockfile unchanged;
+no new dependencies or borrowed modules. Approved install route remains:
 
 ```
 . ./bin/activate-hermit
@@ -93,32 +95,27 @@ cd beehive
 pnpm install --ignore-workspace --frozen-lockfile --ignore-scripts \
   --registry https://global.block-artifacts.com/artifactory/api/npm/square-npm/
 npm run check
-npm test
+BEEHIVE_REAL_BUZZ_ACP=/Applications/Buzz.app/Contents/MacOS/buzz-acp npm test
 ```
 
-Fresh empty node_modules + fresh temporary pnpm store fetched all seven packages
+Full suite with installed opt-in: **11/11 top-level tests pass**. Without opt-in:
-successfully on pnpm 11.4.0 / Node 24.15.0. Strict TypeScript check passes;
+10 pass, installed test explicitly skipped. Strict TS and `git diff --check` pass.
-**npm test: 9/9 top-level tests pass** (13.2 seconds). Tests include original
+No repo-wide `just ci`, production service, owner OAuth or live provider call.
-ACP/WS/TUI regressions, D1–D3, and blocked conversation setup/launch contract.
+Commit uses configured Logan Johnson <loganj@squareup.com> author/committer/DCO,
-`git diff --check` passes. Local commit uses configured Logan Johnson
+without persistent config/signing changes. No independent broker review yet.
-<loganj@squareup.com> author/committer and DCO; no persistent Git config edits.
-No real Databricks OAuth, model call, conversation relay admission or external
-buzz-acp network session was exercised. Generic ACP fixtures are NOT Buzz relay
-parity. No repo-wide `just ci` or production readiness claim.
 
-## Next executable action / full scope
+**ONE next executable action:** implement a locally provisioned, fixed-authority
+MCP/Buzz-tool launch plan (including service-local tool executable discovery and
+owned teardown), then extend the installed isolated-relay test to require an actual
+signed threaded agent reply. Do not loosen the broker to accept shim executable/env
+provisioning or invent an alternate Buzz dispatcher. Request independent broker
+review against this usable artifact before broadening the launch surface.
 
-**Implement the host-owned ACP stdio broker described above, then remove the
+Still open: actual community admission/attestation and named-service-user normal
-conversation Start gate only when adapter/shim teardown and exact-model real
+Databricks v2 OAuth/live `databricks-claude-haiku-4-5` proof; catalog auth ambiguity;
-conversation correlation pass tests.** Do not spend another run rediscovering
+reconnect/ACKs and durable TUI intents; crash recovery/strong OS containment;
-upstream routing. Preserve the provisioned agent key and the two-relay distinction.
+multiple hosts/agents/setups, profiles, key import/revocation, Restart/Move; remaining
-
+Desktop harnesses/presets/providers/mesh/compute. S1 two-host identity unproven.
-Still open: actual Buzz-relay conversation/owner admission, live provider proof,
+Selected-next != actual-run; stopped identity remains assigned. Move must never
-catalog auth ambiguity, reconnect/ACKs and durable TUI intents, crash recovery and
+transfer credentials/workspace/session or permit unreachable takeover. Hash/spawn
-strong OS containment, multiple agents/hosts/setups, profiles/instructions/metadata,
+TOCTOU, hostile journals, rollback/clones and outbox pruning remain outside evidence.
-key import/revocation, Restart/Move, other Desktop harnesses/presets/providers,
-relay-mesh and compute. S1 two-host same identity is unproven. Source setup is one
-user-facing object but initial key setup is still coupled to first harness setup;
-conversation attachment now independently preserves keys. No remaining scope row
-is an owner-approved exclusion. Hash-to-spawn TOCTOU, hostile journals, rollback/
-clone protection and outbox pruning remain explicitly outside current evidence.

beehive/README.md

@@ -135,7 +135,7 @@ until Stop. Auth/protocol failures do not become accepted Start.
 
 **This is an ACP integration probe, NOT yet a conversational Buzz relay agent.**
 The persistent Beehive agent key binds host authority but is not supplied to this
-stdio-only probe, and there is no relay prompt/conversation bridge yet. Model
+stdio-only probe, and this unconfigured path does not use the conversation broker. Model
 acknowledgement is the trusted external executable's claim, not cryptographic
 provider/model attestation. Only deterministic external ACP fixtures have passed;
 no credentialed Databricks response, live account or installed binary ACP
@@ -158,27 +158,34 @@ seam remains missing in the external protocol; do not relabel it authenticated.
 An empty/filtered catalog does not prevent trying an explicitly saved exact model;
 only same-session acknowledgement plus completed response establishes probe evidence.
 
-## External conversation setup (launch blocked)
+## External conversation setup (broker wired; delivery incomplete)
 
 `node src/cli.ts conversation-setup /absolute/existing-host-directory` attaches
-an installed **buzz-acp** executable and separate Buzz conversation relay URL to
+an installed **buzz-acp** and separate conversation relay to the SAME existing
-an existing Buzz Agent setup. It never generates/replaces the provisioned agent
+agent key. Close the host first; setup uses its exclusion lock. It never reads
-identity, reads OAuth or initiates a connection. Close the host first; setup uses
+owner profiles, performs sign-in or regenerates an identity.
-the same atomic exclusion lock and leaves the key/auth context unchanged.
 
-**This does not enable a conversation yet.** A configured conversation Start is
+Start now uses a private per-run TypeScript ACP broker. External buzz-acp owns
-rejected before spawn; inventory explains why. Upstream `AcpClient::spawn` calls
+Buzz authentication, membership, mentions and dispatch; its detached childless
-`process_group(0)`, so putting buzz-acp inside the existing outer supervisor would
+stdio shim cannot choose executables/env/keys. The host separately owns the actual
-let its ACP adapter escape owned Stop. `apply_model_switch` can also proceed on
+harness process group. The proxy enforces exact-model acknowledgement at each
-a default model after rejection. A host-owned TS ACP stdio broker must retain
+conversation session boundary and hashes the actual completed response, never a
-adapter ownership and observe exact-model/session completions in actual Buzz
+separate greeting. Start waits up to 30 seconds for a real inbound conversation;
-conversations before this gate can be removed. Buzz-acp—not Beehive—will continue
+missing relay admission or local sign-in fails Start. Stop tears down all retained
-to own Nostr auth, membership, author gates, mentions, replies and delivery.
+runtime/harness groups and confirms shim exit. No raw text or credentials enter
+management observations. This is not provider attestation or delivery proof.
 
-The launch plan supplies the existing agent key, explicit owner, optional local
+Both external process-contract fixtures and the installed buzz-acp have exercised
-NIP-OA attestation and authoritative relay/model env. None has been exercised
+this broker. The installed-binary test uses fresh fixture identities and an
-against a real conversation relay. Community admission and host-service-user OAuth
+isolated NIP-42/NIP-29 relay fixture, not a real community or provider. That binary
-remain explicit operator actions; this package has not performed them.
+subscribes to legacy kind 9; this is not current Buzz relay compatibility evidence.
+**No signed threaded agent reply was observed.** Current setup disables MCP tools;
+locally provisioned, fixed-authority tool launch and reply delivery remain required.
+Untrusted shim-supplied MCP executables/env are rejected rather than executed.
+See CHECKPOINT.md for exact evidence and next executable action.
+
+Community admission/attestation and normal OAuth in the named host-service context
+remain explicit local/operator actions. No live Databricks request has occurred.
 
 ## Supported / remaining parity
 
@@ -188,7 +195,7 @@ remain explicit operator actions; this package has not performed them.
 | Host inventory | Real relay, fixed authority, freshness | multiple setups/agents, service install, reconnect/reconciliation |
 | Remote configuration | CAS model/workspace/default profile selection | reusable profiles, metadata, config history, setup revision pinning |
 | Lifecycle | fixture Start/Stop, UI-independent host | strict real launch, Restart, host-owned irreversible Move |
-| Buzz Agent Databricks v2 | Local setup + TypeScript ACP probe boundary | live OAuth/model proof, full conversation bridge and diagnostic catalog provenance |
+| Buzz Agent Databricks v2 | Local setup + TypeScript ACP probe boundary | fixed-authority tool/reply delivery, live OAuth/model proof and catalog provenance |
 | Other Buzz Agent providers | Not implemented | Anthropic, OpenAI-compatible, Databricks legacy, OpenRouter |
 | Goose / Claude Code / Codex | Not implemented | harness-specific local auth and adapters/catalog/launch |
 | Ten presets / custom ACP | Not implemented | explicit executable/env trust, preset-specific capability grounding |

beehive/TESTING.md

@@ -41,16 +41,26 @@ strict TypeScript checking passed on Node 24.15.0 / pnpm 11.4.0. Keep
   remote Stop and host close. `disconnect.test.ts` verifies parent IPC loss during
   runner life uses the same ownership-preserving teardown, including closed-IPC
   notification handling. Fixtures self-expire to bound failed regression runs.
-- `conversation.test.ts`: exact local external buzz-acp launch contract, preserved
+- `conversation.test.ts`: persistent identity and exact external env, plus real
-  provisioned identity, authoritative owner/relay/model env, no ambient provider
+  management WS Start -> external runtime fixture -> escaping detached shim ->
-  token, rejected lossy args and unsafe URLs, sanitized remote summary. Real
+  separately anchored harness, same-session completion and Stop absence checks.
-  management WS refuses unsafe conversation Start before any process/run state,
+- `broker.test.ts`: exact-model rejection/conflict, malformed coalesced completion,
-  preserves Save/Stop and keys, and exercises CLI exclusion-lock cleanup.
+  cancellation, reverse RPC, session/load, forbidden workspace/MCP provisioning,
+  and teardown including TERM-resistant descendants.
+- `cancellation.test.ts` also exercises the conversation replacement path for
+  admitted Stop, host close, non-authority attempts and malformed completions.
+- `installed-conversation.test.ts`: opt-in with
+  `BEEHIVE_REAL_BUZZ_ACP=/Applications/Buzz.app/Contents/MacOS/buzz-acp npm test`.
+  Uses fresh fixture identities/dirs and an isolated NIP-42/REST/NIP-29 relay
+  fixture. Verifies actual installed executable AUTH signatures, membership,
+  owner mention dispatch, and exact-model same-conversation ACP completion through
+  the broker. Its actual kind-9 subscription is respected. No provider is called.
+  Signed threaded reply is **not** observed; tools remain disabled/unprovisioned.
 
-Tests use isolated keys/directories/ports/processes only. Generic external ACP
+Tests never open existing owner keys/profiles/cache, production services or native
-fixtures are **not** Buzz conversation relay protocol fixtures. There is no
+client resources. With installed opt-in: 11/11 pass; without: 10 pass/1 explicit
-end-to-end identity-bearing Buzz conversation test yet. No real OAuth/model call,
+skip. The installed fixture relay does not establish actual community admission,
-owner credential store, production relay, native service or old app is touched.
+current relay compatibility, provider/model attestation or working reply delivery.
 
 ## Independent evidence and limits
 
@@ -65,4 +75,4 @@ No full repo `just ci`, multi-host Move, arbitrary escaped descendant containmen
 strong OS sandbox, hostile journal, log-full/reconnect/ACK, rollback, host SIGKILL
 recovery or real provider proof. Genuine ownership loss retains quarantine/lock;
 no stale numeric PID grants cleanup authority. See CHECKPOINT.md for exact next
-executable action and the intentionally blocked external conversation topology.
+executable action and the wired broker and remaining fixed-authority tool/delivery gap.

beehive/src/acp-shim.ts

@@ -0,0 +1,18 @@
+/** Private stdio transport only: no subprocesses, executable selection or env grants. */
+import { connect } from 'node:net';
+const [path, capability] = process.argv.slice(2);
+if (!path || !capability || process.argv.length !== 4) process.exit(2);
+const socket = connect(path);
+const timer = setTimeout(() => process.exit(2), 5000);
+socket.once('connect', () => {
+  clearTimeout(timer);
+  socket.write(JSON.stringify({ capability, pid: process.pid }) + '\n');
+  process.stdin.pipe(socket);
+  socket.pipe(process.stdout);
+});
+// Upstream starts this shim in a separate process group. It never spawns children;
+// EOF, disconnect and TERM all terminate it, independently of upstream lifetime.
+const exit = () => { socket.destroy(); process.exit(0); };
+socket.on('error', exit); socket.on('close', exit);
+process.stdin.on('end', exit); process.stdout.on('error', exit);
+process.on('SIGTERM', exit); process.on('SIGINT', exit);

beehive/src/acp.ts

@@ -9,7 +9,7 @@ export type AgentLaunch = Readonly<{ executable: string; args: readonly string[]
 /** ACP catalogs can be fallback data; even a nonempty result is NOT auth evidence. */
 export type Catalog = { state: 'reported' | 'empty' | 'filtered'; models: string[]; authentication: 'unverified' };
 /** Same child/session acknowledgement plus completed text response, not provider attestation. */
-export type Evidence = { session: string; model: string; responseHash: string; stopReason: 'end_turn'; source: 'external-acp-session'; executableHash: string };
+export type Evidence = { session: string; model: string; responseHash: string; stopReason: 'end_turn'; source: 'external-acp-session' | 'external-buzz-conversation'; executableHash: string; agentPublicKey?: string; runtimeExecutableHash?: string };
 const hash = (s: string | Buffer) => createHash('sha256').update(s).digest('hex');
 const record = (v: unknown): Record<string, unknown> => {
   if (!v || typeof v !== 'object' || Array.isArray(v)) throw Error('Invalid ACP response');

beehive/src/broker.ts

@@ -0,0 +1,262 @@
+import { setTimeout as delay } from 'node:timers/promises';
+import { createServer, type Socket } from 'node:net';
+import { mkdtempSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { fileURLToPath } from 'node:url';
+import { createHash, randomBytes } from 'node:crypto';
+import { spawnOwned, type OwnedProcess } from './owned.ts';
+import { prepareAgent, type AgentLaunch, type Evidence } from './acp.ts';
+import { prepareConversation, type ConversationSetup } from './conversation.ts';
+
+type RPC = { jsonrpc: string; id?: string | number; method?: string; params?: any; result?: any; error?: unknown };
+type Pending = { method: string; params: any };
+const validID = (s: unknown): s is string => typeof s === 'string' && /^[a-zA-Z0-9_.:/-]{1,200}$/.test(s);
+
+/** Run-local ACP transport. Authority comes only from the locally snapshotted plan. */
+export class ConversationSession {
+  private directory: string;
+  private capability = randomBytes(32).toString('hex');
+  private sockets = new Set<Socket>();
+  private harnesses: OwnedProcess[] = [];
+  private shimPids = new Set<number>();
+  private runtime?: OwnedProcess;
+  private cancellations: (() => void)[] = [];
+  private failed = false;
+  private stopping?: Promise<void>;
+  private resolve!: (e: Evidence) => void;
+  private reject!: (e: Error) => void;
+  private evidence = new Promise<Evidence>((resolve, reject) => { this.resolve = resolve; this.reject = reject; });
+  private server = createServer(socket => this.accept(socket));
+  private prepared;
+  private plan;
+  private timeout: ReturnType<typeof setTimeout>;
+  readonly ready: Promise<void>;
+  constructor(setup: ConversationSetup, launch: AgentLaunch, secret: string, owner: string, timeoutMs = 30_000) {
+    this.prepared = prepareAgent(launch);
+    this.plan = prepareConversation(setup, launch, secret, owner);
+    this.directory = mkdtempSync(join(tmpdir(), 'bh-'));
+    // Rejection can precede verify() while the local listener is opening.
+    void this.evidence.catch(() => {});
+    this.timeout = setTimeout(() => this.fail('Conversation response timed out; check relay admission and local sign-in'), timeoutMs);
+    this.server.on('error', () => this.fail('Private ACP broker unavailable'));
+    this.ready = new Promise<void>((resolve, reject) => {
+      this.server.once('error', reject);
+      this.server.listen(join(this.directory, 'acp'), () => {
+        if (this.failed) { reject(Error('Conversation Start cancelled')); return; }
+        const shim = fileURLToPath(new URL('./acp-shim.ts', import.meta.url));
+        const args = [shim, join(this.directory, 'acp'), this.capability];
+        if (args.some(a => a.includes(','))) { reject(Error('Unsupported local shim path')); return; }
+        this.runtime = spawnOwned(this.plan.executable, [], this.plan.workspace, {
+          ...this.plan.env, BUZZ_ACP_AGENT_COMMAND: process.execPath, BUZZ_ACP_AGENT_ARGS: args.join(','),
+        });
+        this.runtime.child.stdout.resume(); this.runtime.child.stderr.resume();
+        this.watch(this.runtime);
+        this.runtime.ready.then(resolve, reject);
+      });
+    });
+    void this.ready.catch(() => this.fail('Conversation runtime unavailable'));
+  }
+  get healthy() { return !this.failed; }
+  get exited() { return this.failed || !!this.runtime?.exited; }
+  get connected() { return !this.stopping || !this.failed; }
+  private watch(owned: OwnedProcess) {
+    owned.child.on('message', () => { if (owned.exited) this.fail('Conversation process exited'); });
+    owned.child.on('exit', () => this.fail('Conversation supervisor exited'));
+    owned.child.on('error', () => this.fail('Conversation process unavailable'));
+    void owned.ready.catch(() => this.fail('Conversation harness unavailable'));
+  }
+  private fail(reason: string) {
+    if (this.failed) return;
+    this.failed = true; clearTimeout(this.timeout); this.reject(Error(reason));
+    for (const socket of this.sockets) socket.destroy();
+    // Keep ownership handles until the caller verifies complete teardown.
+  }
+  /** Interrupt Start without allowing late ACP responses to commit readiness. */
+  cancel() {
+    if (!this.failed) for (const cancel of this.cancellations) cancel();
+    this.fail('Conversation Start cancelled');
+  }
+  /** Evidence is from the external runtime's actual prompt, never a synthetic probe. */
+  async verify() {
+    await this.ready;
+    const evidence = await this.evidence;
+    if (!this.healthy) throw Error('Conversation failed before evidence commit');
+    return evidence;
+  }
+  /** Close escaping, childless shims via EOF and stop every retained live group. */
+  stop(): Promise<void> {
+    if (this.stopping) return this.stopping;
+    this.cancel();
+    this.stopping = (async () => {
+      await this.ready.catch(() => {});
+      await new Promise<void>(resolve => this.server.close(() => resolve()));
+      const results = await Promise.allSettled([...this.harnesses, ...(this.runtime ? [this.runtime] : [])].map(p => p.stop()));
+      if (results.some(r => r.status === 'rejected')) throw Error('Conversation ownership teardown failed; reconcile locally');
+      // These numeric IDs are absence observations ONLY, never kill authority.
+      // A shim is childless and exits on EOF even though upstream gave it a new PGID.
+      for (let i = 0; i < 100 && this.shimPids.size; i++) {
+        for (const pid of this.shimPids) {
+          try { process.kill(pid, 0); }
+          catch (error) { if ((error as NodeJS.ErrnoException).code === 'ESRCH') this.shimPids.delete(pid); }
+        }
+        if (this.shimPids.size) await delay(25);
+      }
+      if (this.shimPids.size) throw Error('ACP shim exit unconfirmed; reconcile locally');
+      rmSync(this.directory, { recursive: true, force: true });
+    })();
+    return this.stopping;
+  }
+  private accept(socket: Socket) {
+    if (this.failed || this.sockets.size >= 8) { socket.destroy(); return; }
+    this.sockets.add(socket);
+    let authenticated = false;
+    let buffer = '';
+    let harness: OwnedProcess | undefined;
+    const timer = setTimeout(() => socket.destroy(), 2000);
+    socket.on('error', () => socket.destroy());
+    socket.on('close', () => {
+      clearTimeout(timer); this.sockets.delete(socket);
+      if (authenticated && !this.failed) this.fail('ACP shim disconnected');
+    });
+    socket.setEncoding('utf8');
+    socket.on('data', (chunk: string) => {
+      if (authenticated) return;
+      buffer += chunk;
+      if (buffer.length > 4096) { socket.destroy(); return; }
+      const end = buffer.indexOf('\n');
+      if (end < 0) return;
+      let hello: { capability?: string; pid?: number };
+      try { hello = JSON.parse(buffer.slice(0, end)); } catch { socket.destroy(); return; }
+      if (!hello || hello.capability !== this.capability || !Number.isSafeInteger(hello.pid) || hello.pid! <= 1 || this.harnesses.length >= 8) { socket.destroy(); return; }
+      this.shimPids.add(hello.pid!);
+      authenticated = true; clearTimeout(timer);
+      const { plan, env } = this.prepared;
+      // No data supplied by a shim can alter argv, workspace, identity or env.
+      harness = spawnOwned(plan.executable, plan.args, plan.workspace, {
+        ...env, BUZZ_PRIVATE_KEY: this.plan.env.BUZZ_PRIVATE_KEY,
+        BUZZ_RELAY_URL: this.plan.env.BUZZ_RELAY_URL,
+        ...(this.plan.env.BUZZ_AUTH_TAG ? { BUZZ_AUTH_TAG: this.plan.env.BUZZ_AUTH_TAG } : {}),
+      });
+      this.harnesses.push(harness); this.watch(harness);
+      this.proxy(socket, harness, buffer.slice(end + 1));
+      buffer = '';
+    });
+  }
+  private proxy(socket: Socket, harness: OwnedProcess, initial: string) {
+    const pending = new Map<string | number, Pending>();
+    const sessions = new Set<string>();
+    const modelConfigs = new Set<string>(['model']);
+    const prompts = new Map<string, { id: string | number; hash: ReturnType<typeof createHash>; text: boolean; cancelled: boolean }>();
+    this.cancellations.push(() => {
+      for (const sessionId of prompts.keys()) harness.child.stdin.write(JSON.stringify({ jsonrpc: '2.0', method: 'session/cancel', params: { sessionId } }) + '\n');
+    });
+    const injected = new Map<string, { original: RPC; session: string }>();
+    let sequence = 0;
+    const send = (target: NodeJS.WritableStream, msg: RPC) => {
+      if (this.failed) return;
+      if (!target.write(JSON.stringify(msg) + '\n')) this.fail('ACP transport backpressure limit');
+    };
+    const inspect = (msg: RPC, fromHarness: boolean) => {
+      if (msg.jsonrpc !== '2.0' || (msg.id !== undefined &&
+        !(typeof msg.id === 'string' && msg.id.length <= 200) &&
+        !(typeof msg.id === 'number' && Number.isSafeInteger(msg.id)))) throw Error();
+      if (msg.method !== undefined) {
+        if (typeof msg.method !== 'string' || !msg.method.length || msg.result !== undefined || msg.error !== undefined) throw Error();
+      } else if (msg.id === undefined || (Object.hasOwn(msg, 'result') === Object.hasOwn(msg, 'error'))) throw Error();
+      if (!fromHarness && msg.method) {
+        const p = msg.params;
+        // Session transport does not grant a shim authority to provision tools,
+        // workspaces or credentials. This slice has no locally approved MCP plan.
+        if (['session/new', 'session/load', 'session/resume'].includes(msg.method)) {
+          if (p?.cwd !== this.prepared.plan.workspace || !Array.isArray(p.mcpServers) || p.mcpServers.length) throw Error();
+          sessions.delete(p.sessionId);
+        }
+        if (p && ['env', 'environment', 'executable', 'command'].some(key => Object.hasOwn(p, key))) throw Error();
+        if (msg.id !== undefined) {
+          if (pending.size >= 128 || pending.has(msg.id) || String(msg.id).startsWith('beehive-model-')) throw Error();
+          pending.set(msg.id, { method: msg.method, params: p });
+        }
+        if (msg.method === 'session/set_model' && (!sessions.has(p?.sessionId) || p.modelId !== this.prepared.plan.model)) throw Error();
+        if (msg.method === 'session/set_config_option' && modelConfigs.has(p?.configId) && p.value !== this.prepared.plan.model) throw Error();
+        if (['session/set_model', 'session/set_config_option'].includes(msg.method)) sessions.delete(p?.sessionId);
+        if (msg.method === 'session/prompt') {
+          if (!sessions.has(p?.sessionId) || prompts.has(p.sessionId) || msg.id === undefined) throw Error();
+          prompts.set(p.sessionId, { id: msg.id, hash: createHash('sha256'), text: false, cancelled: false });
+        }
+        if (msg.method === 'session/cancel') {
+          const prompt = prompts.get(p?.sessionId); if (prompt) prompt.cancelled = true;
+        }
+      }
+      if (fromHarness && !msg.method && msg.id !== undefined) {
+        const internal = injected.get(String(msg.id));
+        if (internal) {
+          injected.delete(String(msg.id));
+          if (msg.error !== undefined || msg.result?.sessionId !== internal.session || msg.result?.modelId !== this.prepared.plan.model) throw Error();
+          sessions.add(internal.session); send(socket, internal.original); return;
+        }
+        const request = pending.get(msg.id);
+        if (!request) throw Error();
+        pending.delete(msg.id);
+        if (['session/new', 'session/load', 'session/resume', 'session/set_config_option'].includes(request.method) && msg.error === undefined) {
+          const session = msg.result?.sessionId ?? request.params?.sessionId;
+          if (!validID(session) || sessions.size >= 128 || injected.size >= 128) throw Error();
+          for (const option of msg.result?.configOptions ?? []) {
+            if (option.category === 'model' && typeof option.id === 'string') modelConfigs.add(option.id);
+            if (option.category === 'model' && typeof option.configId === 'string') modelConfigs.add(option.configId);
+          }
+          if (modelConfigs.size > 128) throw Error();
+          sessions.delete(session);
+          const id = `beehive-model-${++sequence}`;
+          injected.set(id, { original: msg, session });
+          send(harness.child.stdin, { jsonrpc: '2.0', id, method: 'session/set_model', params: { sessionId: session, modelId: this.prepared.plan.model } }); return;
+        }
+        if (request.method === 'session/set_model') {
+          if (msg.error !== undefined || msg.result?.sessionId !== request.params.sessionId || msg.result?.modelId !== this.prepared.plan.model) throw Error();
+          sessions.add(request.params.sessionId);
+        }
+        if (request.method === 'session/prompt') {
+          const session = request.params.sessionId; const prompt = prompts.get(session); prompts.delete(session);
+          if (msg.error === undefined && msg.result?.stopReason === 'end_turn' && prompt?.text && !prompt.cancelled) {
+            const evidence: Evidence = { session, model: this.prepared.plan.model, executableHash: this.prepared.executableHash, responseHash: prompt.hash.digest('hex'), stopReason: 'end_turn', source: 'external-buzz-conversation', agentPublicKey: this.plan.agentPublicKey, runtimeExecutableHash: this.plan.executableHash };
+            // Defer commit until all lines in this batch have been validated (D3).
+            queueMicrotask(() => { if (this.healthy) { clearTimeout(this.timeout); this.resolve(evidence); } });
+          }
+        }
+      }
+      if (fromHarness && msg.method === 'session/update') {
+        const p = msg.params; const u = p?.update;
+        if (u?.sessionUpdate === 'current_model_update' && u.currentModelId !== this.prepared.plan.model) throw Error();
+        if (u?.sessionUpdate === 'config_option_update' && Array.isArray(u.configOptions)) {
+          for (const option of u.configOptions) if (option.category === 'model' && option.currentValue !== this.prepared.plan.model) throw Error();
+        }
+        const prompt = prompts.get(p?.sessionId);
+        if (prompt && u?.sessionUpdate === 'agent_message_chunk' && u.content?.type === 'text' && typeof u.content.text === 'string') {
+          prompt.hash.update(u.content.text); prompt.text ||= !!u.content.text.trim();
+        }
+      }
+      send(fromHarness ? socket : harness.child.stdin, msg);
+    };
+    const reader = (fromHarness: boolean) => {
+      let buffer = '';
+      return (chunk: string) => {
+        if (this.failed) return;
+        buffer += chunk;
+        if (Buffer.byteLength(buffer) > 1024 * 1024) { this.fail('ACP frame limit exceeded'); return; }
+        try {
+          let end: number;
+          while ((end = buffer.indexOf('\n')) >= 0) {
+            const line = buffer.slice(0, end); buffer = buffer.slice(end + 1);
+            if (line.trim()) { const msg = JSON.parse(line); if (!msg || typeof msg !== 'object' || Array.isArray(msg)) throw Error(); inspect(msg, fromHarness); }
+          }
+        } catch { this.fail('Conversation ACP protocol/model verification failed (diagnostics withheld)'); }
+      };
+    };
+    const inbound = reader(false);
+    socket.on('data', inbound);
+    harness.child.stdout.setEncoding('utf8'); harness.child.stdout.on('data', reader(true));
+    harness.child.stdin.on('error', () => this.fail('ACP input failed'));
+    harness.child.stderr.resume();
+    if (initial) inbound(initial);
+  }
+}

beehive/src/conversation.ts

@@ -52,16 +52,7 @@ export function prepareConversation(input: ConversationSetup, agent: AgentLaunch
   });
 }
 
-/**
- * Do not regress owned Stop: upstream AcpClient::spawn uses process_group(0),
- * escaping the existing outer group anchor. A host-owned ACP stdio broker must
- * supervise the actual adapter before this plan is safe to execute.
- */
-export function requireConversationContainment(): never {
-  throw Error('Conversation Start unavailable: external buzz-acp creates a separate ACP process group; host-owned ACP bridge containment is not implemented. No process started.');
-}
-
 /** Safe remote summary. Credentials and executable paths are never included. */
 export function conversationSummary(_setup: ConversationSetup) {
-  return { transport: 'external-buzz-acp', admission: 'unverified', modelEvidence: 'not-observed', launch: 'blocked: ACP subtree containment required' };
+  return { transport: 'external-buzz-acp', admission: 'unverified', modelEvidence: 'not-observed', launch: 'host-owned ACP broker; awaiting conversation evidence' };
 }

beehive/src/host.ts

@@ -4,9 +4,10 @@ import { createHash } from 'node:crypto';
 import { connect, validateRelayURL } from './client.ts';
 import { digest, fields, message, object, publicKey, text, type Message } from './protocol.ts';
 import { readPrivate, writePrivate } from './storage.ts';
+import { ConversationSession } from './broker.ts';
 import { spawnOwned, type OwnedProcess } from './owned.ts';
 import { AgentSession, prepareAgent, type AgentLaunch, type Catalog, type Evidence } from './acp.ts';
-import { prepareConversation, requireConversationContainment, conversationSummary, type ConversationSetup } from './conversation.ts';
+import { prepareConversation, conversationSummary, type ConversationSetup } from './conversation.ts';
 
 export type Setup = { host: string; ownerSecret: string; agentSecret: string; runner: string; args: string[]; workspace: string; mode: 'fixture' | 'buzz-agent-databricks-v2'; databricksHost?: string; serviceHome?: string; configDirectory?: string; conversation?: ConversationSetup };
 type Selection = { model: string; workspace: string; profile: string };
@@ -33,15 +34,15 @@ export async function host(directory: string, url: string) {
   if (state.phase !== 'stopped') state.phase = 'quarantined';
   const save = () => writePrivate(path,state);
   try { save(); } catch (e) { rmdirSync(lock); throw e; }
-  let owned: OwnedProcess | undefined;
+  let owned: OwnedProcess | ConversationSession | undefined;
-  let acp: AgentSession | undefined;
+  let acp: AgentSession | ConversationSession | undefined;
   let catalog: Catalog | { state: 'not-probed' | 'failed'; authentication: 'unverified' } = { state: 'not-probed', authentication: 'unverified' };
   function inventory() {
     return message('inventory',setup.host,agent,state.revision, {
       assignedHost: setup.host, phase: state.phase, selectedNext: state.selected, actualRun: state.actual,
       ...(setup.conversation ? { conversation: conversationSummary(setup.conversation) } : {}),
       catalog, setup: setup.mode, models: [setup.mode === 'fixture' ? 'fixture-model' : 'databricks-claude-haiku-4-5'],
-      workspaces: [setup.workspace], profiles: ['default'], readiness: setup.conversation ? 'Conversation Start blocked: host-owned ACP subtree containment required; relay/auth/model unverified' : setup.mode === 'fixture' ? 'fixture-only' : (state.phase === 'running' && state.actual?.evidence ? 'ACP model acknowledged + same-session response (not provider attestation or Buzz relay agent)' : 'unverified: Start runs an ACP greeting probe; sign in locally as host service user if required'), observedAt: Date.now(),
+      workspaces: [setup.workspace], profiles: ['default'], readiness: setup.conversation ? (state.actual?.evidence ? 'Conversation session model acknowledged and response completed; relay delivery unverified' : 'Waiting for an admitted conversation and local provider sign-in; no synthetic prompt') : setup.mode === 'fixture' ? 'fixture-only' : (state.phase === 'running' && state.actual?.evidence ? 'ACP model acknowledged + same-session response (not provider attestation or Buzz relay agent)' : 'unverified: Start runs an ACP greeting probe; sign in locally as host service user if required'), observedAt: Date.now(),
     });
   }
   let closing = false;
@@ -72,7 +73,7 @@ export async function host(directory: string, url: string) {
     let result = 'accepted';
     if (m.agent !== agent) result = 'not-authority';
     else if (m.revision !== state.revision) result = 'revision-conflict';
-    else if ((state.phase === 'quarantined' || state.phase === 'transitioning') && !(m.type === 'stop' && owned?.child.connected)) result = 'quarantined';
+    else if ((state.phase === 'quarantined' || state.phase === 'transitioning') && !(m.type === 'stop' && (owned instanceof ConversationSession ? owned.connected : owned?.child.connected))) result = 'quarantined';
     else {
       // Reserve operation before any spawn/kill; interrupted admission never retries effects.
       const pending = message('receipt',setup.host,agent,state.revision,{ operation: m.id, result: 'interrupted-reconcile-locally' });
@@ -93,7 +94,6 @@ export async function host(directory: string, url: string) {
           if (setup.conversation) {
             if (!launch) throw Error('Conversation requires a Buzz Agent harness setup');
             prepareConversation(setup.conversation, launch, setup.agentSecret, state.binding.owner);
-            requireConversationContainment(); // Fail before journaling a run or spawning anything.
           }
           state.actual = { selection: structuredClone(state.selected), executableHash: createHash('sha256').update(readFileSync(setup.runner)).digest('hex'), run: m.id };
           state.phase = 'transitioning'; save();
@@ -101,6 +101,16 @@ export async function host(directory: string, url: string) {
             owned = spawnOwned(setup.runner, setup.args, state.actual.selection.workspace, { PATH: '/usr/bin:/bin', BEEHIVE_FIXTURE: '1' });
             owned.child.stdout.resume(); owned.child.stderr.resume();
             await owned.ready;
+          } else if (setup.conversation) {
+            const session = new ConversationSession(setup.conversation, launch!, setup.agentSecret, state.binding.owner);
+            acp = session; owned = session;
+            try {
+              state.actual.evidence = await session.verify();
+              if (!session.healthy) throw Error('Conversation failed before running commit');
+            } catch (e) {
+              await stopOwned(); state.phase = 'stopped'; state.actual = null;
+              throw e;
+            }
           } else {
             const session = new AgentSession(launch!); acp = session;
             owned = session.owned;

beehive/test/broker.test.ts

@@ -0,0 +1,41 @@
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import { mkdtempSync, realpathSync, writeFileSync, readFileSync, existsSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join, resolve } from 'node:path';
+import { setTimeout as delay } from 'node:timers/promises';
+import { ConversationSession } from '../src/broker.ts';
+import { newKey, publicKey } from '../src/protocol.ts';
+
+test('broker enforces model, completion health, cancellation and owned teardown across escaped shims', async () => {
+  for (const mode of ['ok', 'reject', 'wrong-model', 'conflict', 'bad-tail', 'delayed', 'cancel', 'foreign-workspace', 'injected-mcp']) {
+    const dir = realpathSync(mkdtempSync(join(tmpdir(), 'bh-broker-test-')));
+    const executable = realpathSync(process.execPath); const runtime = join(dir, 'runtime');
+    writeFileSync(runtime, `#!/bin/sh\nexec '${executable}' '${resolve('test/conversation-runtime-fixture.ts')}'\n`, { mode: 0o700 });
+    writeFileSync(join(dir, 'mode'), mode);
+    const session = new ConversationSession({ executable: runtime, relay: 'ws://127.0.0.1:1' }, {
+      executable, args: [resolve('test/conversation-harness-fixture.ts')], workspace: dir, home: dir, configDirectory: dir,
+      databricksHost: 'https://fixture.invalid', model: 'databricks-claude-haiku-4-5',
+    }, newKey(), publicKey(newKey()), 2000);
+    const result = session.verify().then(value => ({ value }), error => ({ error }));
+    try {
+      if (mode === 'delayed' || mode === 'cancel') {
+        for (let i = 0; i < 80 && !existsSync(join(dir, mode === 'cancel' ? 'cancel-observed' : 'prompt-started')); i++) await delay(20);
+        assert.ok(existsSync(join(dir, mode === 'cancel' ? 'cancel-observed' : 'prompt-started')));
+        session.cancel();
+      }
+      const r = await result;
+      if (mode === 'ok') {
+        assert.ok('value' in r); assert.equal(r.value.session, 'conversation-session');
+        for (let i = 0; i < 40 && !existsSync(join(dir, 'reverse-rpc')); i++) await delay(20);
+        assert.equal(readFileSync(join(dir, 'reverse-rpc'), 'utf8'), 'ok');
+      } else assert.ok('error' in r, `${mode} must not commit evidence`);
+    } finally {
+      await session.stop();
+      for (const file of ['runtime-pid', 'shim-pid', 'harness-pid', 'descendant-pid']) {
+        if (existsSync(join(dir, file))) assert.throws(() => process.kill(Number(readFileSync(join(dir, file), 'utf8')), 0), { code: 'ESRCH' }, `${mode}: ${file}`);
+      }
+      rmSync(dir, { recursive: true, force: true });
+    }
+  }
+});

beehive/test/cancellation.test.ts

@@ -1,6 +1,6 @@
 import { test } from 'node:test';
 import assert from 'node:assert/strict';
-import { mkdtempSync, realpathSync, rmSync, readFileSync, existsSync } from 'node:fs';
+import { mkdtempSync, realpathSync, rmSync, readFileSync, existsSync, writeFileSync } from 'node:fs';
 import { tmpdir } from 'node:os';
 import { join, resolve } from 'node:path';
 import { setTimeout as delay } from 'node:timers/promises';
@@ -11,10 +11,13 @@ import { message, newKey, publicKey, type Message } from '../src/protocol.ts';
 import { writePrivate } from '../src/storage.ts';
 
 test('in-flight ACP Start is cancellable by authorized Stop and host close; malformed completion cannot commit running', async () => {
-  for (const action of ['stop', 'close', 'bad-tail']) {
+  for (const target of ['probe', 'conversation']) for (const action of ['stop', 'close', 'bad-tail']) {
     const dir = realpathSync(mkdtempSync(join(tmpdir(), 'beehive-cancel-')));
     const secret = newKey(); const agentSecret = newKey(); const agent = publicKey(agentSecret);
-    writePrivate(join(dir, 'setup.json'), { host: 'cancel-host', ownerSecret: secret, agentSecret, runner: realpathSync(process.execPath), args: [resolve('test/acp-fixture.ts'), action === 'bad-tail' ? 'bad-tail' : 'delayed'], workspace: dir, mode: 'buzz-agent-databricks-v2', serviceHome: dir, configDirectory: dir, databricksHost: 'https://fixture.invalid' });
+    const runtime = join(dir, 'runtime');
+    writeFileSync(runtime, `#!/bin/sh\nexec '${realpathSync(process.execPath)}' '${resolve('test/conversation-runtime-fixture.ts')}'\n`, { mode: 0o700 });
+    writeFileSync(join(dir, 'mode'), action === 'bad-tail' ? 'bad-tail' : 'delayed');
+    writePrivate(join(dir, 'setup.json'), { host: 'cancel-host', ownerSecret: secret, agentSecret, runner: realpathSync(process.execPath), args: target === 'probe' ? [resolve('test/acp-fixture.ts'), action === 'bad-tail' ? 'bad-tail' : 'delayed'] : [resolve('test/conversation-harness-fixture.ts')], ...(target === 'conversation' ? { conversation: { executable: runtime, relay: 'ws://127.0.0.1:1' } } : {}), workspace: dir, mode: 'buzz-agent-databricks-v2', serviceHome: dir, configDirectory: dir, databricksHost: 'https://fixture.invalid' });
     const server = await relay(0, publicKey(secret), join(dir, 'relay.json'));
     const address = server.address(); assert.ok(address && typeof address !== 'string');
     const url = `ws://127.0.0.1:${address.port}`;

beehive/test/conversation-harness-fixture.ts

@@ -0,0 +1,37 @@
+/** Owned ACP harness fixture: exact session/model, bidirectional RPC, resistant child. */
+import { createInterface } from 'node:readline';
+import { readFileSync, writeFileSync } from 'node:fs';
+import { spawn } from 'node:child_process';
+import { publicKey } from '../src/protocol.ts';
+const mode = readFileSync('mode', 'utf8');
+const sessionId = 'conversation-session'; let selected = ''; let prompt: number | undefined;
+writeFileSync('harness-identity', publicKey(process.env.BUZZ_PRIVATE_KEY!));
+writeFileSync('harness-pid', String(process.pid));
+const child = spawn(process.execPath, ['-e', "process.on('SIGTERM',()=>{}); setTimeout(()=>{},15000)"], { stdio: 'ignore' });
+writeFileSync('descendant-pid', String(child.pid));
+process.on('SIGTERM', () => {});
+setTimeout(() => process.exit(0), 15000).unref();
+const send = (m: any) => process.stdout.write(JSON.stringify(m) + '\n');
+for await (const line of createInterface({ input: process.stdin })) {
+  const m = JSON.parse(line); let result: any;
+  if (!m.method) { if (m.result?.ok) writeFileSync('reverse-rpc', 'ok'); continue; }
+  if (m.method === 'initialize') result = { protocolVersion: 1, agentInfo: { name: 'buzz-agent' } };
+  else if (m.method === 'session/new') result = { sessionId, models: { currentModelId: 'default', availableModels: [] } };
+  else if (m.method === 'session/load') { selected = ''; result = {}; }
+  else if (m.method === 'session/set_model') {
+    if (mode === 'reject') { send({ jsonrpc: '2.0', id: m.id, error: { code: -1, message: 'secret-diagnostic' } }); continue; }
+    selected = m.params.modelId; result = { sessionId, modelId: mode === 'wrong-model' ? 'other' : selected };
+  } else if (m.method === 'session/cancel') {
+    writeFileSync('cancel-observed', 'yes');
+    if (prompt) send({ jsonrpc: '2.0', id: prompt, result: { stopReason: 'cancelled' } }); continue;
+  } else if (m.method === 'session/prompt') {
+    if (selected !== process.env.BUZZ_AGENT_MODEL || m.params.sessionId !== sessionId) process.exit(8);
+    writeFileSync('prompt-started', selected); prompt = m.id;
+    send({ jsonrpc: '2.0', id: 'reverse', method: 'client/fixture', params: {} });
+    if (mode === 'delayed' || mode === 'cancel') continue;
+    const update = { jsonrpc: '2.0', method: 'session/update', params: { sessionId, update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'Private conversation fixture response' } } } };
+    const reply = { jsonrpc: '2.0', id: m.id, result: { stopReason: 'end_turn' } };
+    process.stdout.write(JSON.stringify(update) + '\n' + JSON.stringify(reply) + '\n' + (mode === 'bad-tail' ? 'not-json\n' : '')); continue;
+  } else throw Error('Unexpected method');
+  send({ jsonrpc: '2.0', id: m.id, result });
+}

beehive/test/conversation-runtime-fixture.ts

@@ -0,0 +1,32 @@
+/** External runtime contract fixture, NOT a replacement Buzz relay implementation. */
+import { spawn } from 'node:child_process';
+import { createInterface } from 'node:readline';
+import { writeFileSync, readFileSync } from 'node:fs';
+import { publicKey } from '../src/protocol.ts';
+const mode = readFileSync('mode', 'utf8');
+writeFileSync('runtime-identity', publicKey(process.env.BUZZ_PRIVATE_KEY!));
+writeFileSync('runtime-pid', String(process.pid));
+const child = spawn(process.env.BUZZ_ACP_AGENT_COMMAND!, process.env.BUZZ_ACP_AGENT_ARGS!.split(','), {
+  // Reproduce AcpClient::spawn's process_group(0) escape, not naive inheritance.
+  detached: true, stdio: ['pipe', 'pipe', 'ignore'], env: { PATH: '/usr/bin:/bin' },
+});
+writeFileSync('shim-pid', String(child.pid));
+setTimeout(() => process.exit(0), 15000).unref();
+const pending = new Map<number, (m: any) => void>(); let seq = 0;
+const input = createInterface({ input: child.stdout });
+input.on('line', line => { const m = JSON.parse(line); if (m.method === 'client/fixture') child.stdin.write(JSON.stringify({ jsonrpc: '2.0', id: m.id, result: { ok: true } }) + '\n'); else pending.get(m.id)?.(m); });
+child.on('exit', () => process.exit(0));
+async function request(method: string, params: unknown) {
+  const id = ++seq;
+  return new Promise<any>(resolve => { pending.set(id, resolve); child.stdin.write(JSON.stringify({ jsonrpc: '2.0', id, method, params }) + '\n'); });
+}
+await request('initialize', { protocolVersion: 1, clientCapabilities: {} });
+const created = await request('session/new', { cwd: mode === 'foreign-workspace' ? '/' : process.cwd(), mcpServers: mode === 'injected-mcp' ? [{ name: 'bad', command: '/bin/sh', args: ['-c', 'exit 0'], env: [] }] : [] });
+const sessionId = created.result.sessionId;
+if (mode === 'conflict') await request('session/set_model', { sessionId, modelId: 'wrong' });
+else await request('session/set_model', { sessionId, modelId: process.env.BUZZ_ACP_MODEL });
+await request('session/load', { sessionId, cwd: process.cwd(), mcpServers: [] });
+const prompt = request('session/prompt', { sessionId, prompt: [{ type: 'text', text: 'External runtime conversation input' }] });
+if (mode === 'cancel') setTimeout(() => child.stdin.write(JSON.stringify({ jsonrpc: '2.0', method: 'session/cancel', params: { sessionId } }) + '\n'), 50);
+const response = await prompt;
+writeFileSync('runtime-completed', response.result.stopReason);

beehive/test/conversation.test.ts

@@ -1,6 +1,6 @@
 import { test } from 'node:test';
 import assert from 'node:assert/strict';
-import { mkdtempSync, realpathSync, rmSync, readFileSync, existsSync } from 'node:fs';
+import { mkdtempSync, realpathSync, rmSync, readFileSync, existsSync, writeFileSync } from 'node:fs';
 import { spawnSync } from 'node:child_process';
 import { tmpdir } from 'node:os';
 import { join, resolve } from 'node:path';
@@ -47,10 +47,13 @@ test('external buzz-acp launch contract preserves identity and rejects lossy/uns
   } finally { rmSync(dir, { recursive: true, force: true }); }
 });
 
-test('real management WS rejects unsafe conversation Start before spawn; Save/Stop and key binding survive', async () => {
+test('real management WS runs identity-bearing external conversation through escaped shim and owned harness; Save/Stop preserve keys', async () => {
   const dir = realpathSync(mkdtempSync(join(tmpdir(), 'beehive-conversation-host-')));
   const ownerSecret = newKey(); const agentSecret = newKey(); const key = publicKey(agentSecret);
-  const setup = { host: 'conversation-host', ownerSecret, agentSecret, runner: executable, args: agent(dir).args, workspace: dir, mode: 'buzz-agent-databricks-v2', serviceHome: dir, configDirectory: dir, databricksHost: 'https://fixture.invalid', conversation: { executable, relay: 'wss://conversation.invalid', authTag: 'fixture-attestation' } };
+  const runtime = join(dir, 'runtime');
+  writeFileSync(runtime, `#!/bin/sh\nexec '${executable}' '${resolve('test/conversation-runtime-fixture.ts')}'\n`, { mode: 0o700 });
+  writeFileSync(join(dir, 'mode'), 'ok');
+  const setup = { host: 'conversation-host', ownerSecret, agentSecret, runner: executable, args: [resolve('test/conversation-harness-fixture.ts')], workspace: dir, mode: 'buzz-agent-databricks-v2', serviceHome: dir, configDirectory: dir, databricksHost: 'https://fixture.invalid', conversation: { executable: runtime, relay: 'wss://conversation.invalid', authTag: 'fixture-attestation' } };
   writePrivate(join(dir, 'setup.json'), setup);
   const server = await relay(0, publicKey(ownerSecret), join(dir, 'relay.json'));
   const address = server.address(); assert.ok(address && typeof address !== 'string');
@@ -68,14 +71,22 @@ test('real management WS rejects unsafe conversation Start before spawn; Save/St
   }
   try {
     const start = await request('start', 0);
-    assert.match(String(start.body.result), /Conversation Start unavailable.*No process started/);
+    assert.equal(start.body.result, 'accepted');
     const state = JSON.parse(readFileSync(join(dir, 'journal.json'), 'utf8'));
-    assert.equal(state.phase, 'stopped'); assert.equal(state.actual, null); assert.equal(state.revision, 0);
+    assert.equal(state.phase, 'running'); assert.equal(state.actual.evidence.session, 'conversation-session'); assert.equal(state.actual.evidence.model, model); assert.equal(state.revision, 1);
-    assert.equal((await request('save', 0, { model, workspace: dir, profile: 'default' })).body.result, 'saved; running configuration unchanged');
+    assert.equal(readFileSync(join(dir, 'runtime-identity'), 'utf8'), key);
-    assert.equal((await request('stop', 1)).body.result, 'accepted');
+    assert.equal(readFileSync(join(dir, 'harness-identity'), 'utf8'), key);
+    assert.equal(readFileSync(join(dir, 'runtime-completed'), 'utf8'), 'end_turn');
+    assert.equal((await request('save', 1, { model, workspace: dir, profile: 'default' })).body.result, 'saved; running configuration unchanged');
+    assert.equal((await request('stop', 2)).body.result, 'accepted');
     assert.equal(JSON.parse(readFileSync(join(dir, 'setup.json'), 'utf8')).agentSecret, agentSecret);
     assert.ok(!JSON.stringify(seen).includes('fixture-attestation'));
     assert.ok(!JSON.stringify(seen).includes(agentSecret));
+    assert.ok(!JSON.stringify(seen).includes('Private conversation fixture response'));
+    for (const file of ['runtime-pid', 'shim-pid', 'harness-pid', 'descendant-pid']) {
+      const pid = Number(readFileSync(join(dir, file), 'utf8'));
+      assert.throws(() => process.kill(pid, 0), { code: 'ESRCH' });
+    }
     const cli = spawnSync(process.execPath, ['src/cli.ts', 'conversation-setup', dir], { encoding: 'utf8', timeout: 3000 });
     assert.notEqual(cli.status, 0); assert.ok(existsSync(join(dir, 'host.lock')));
   } finally {

beehive/test/installed-conversation.test.ts

@@ -0,0 +1,87 @@
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import { createServer } from 'node:http';
+import { WebSocketServer } from 'ws';
+import { mkdtempSync, realpathSync, writeFileSync, rmSync, existsSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join, resolve } from 'node:path';
+import { createHash } from 'node:crypto';
+import { schnorr } from '@noble/curves/secp256k1';
+import { setTimeout as delay } from 'node:timers/promises';
+import { ConversationSession } from '../src/broker.ts';
+import { newKey, publicKey } from '../src/protocol.ts';
+
+// Explicitly opt-in installed executable; never opens an owner profile or provider.
+test('installed buzz-acp authenticates and completes the same identity-bearing conversation through broker', { skip: !process.env.BEEHIVE_REAL_BUZZ_ACP }, async () => {
+  const dir = realpathSync(mkdtempSync(join(tmpdir(), 'bh-installed-')));
+  writeFileSync(join(dir, 'mode'), 'ok');
+  const secret = newKey(); const ownerSecret = newKey(); const agent = publicKey(secret); const owner = publicKey(ownerSecret);
+  const channel = 'aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee';
+  let authenticated = false; let subscriptions = 0; let delivered = false;
+  function observeReply(e: any) {
+    if (![9, 40002].includes(e.kind) || e.pubkey !== agent) return;
+    assert.ok(schnorr.verify(e.sig, e.id, agent));
+    assert.ok(e.tags.some((t: string[]) => t[0] === 'h' && t[1] === channel));
+    assert.ok(e.tags.some((t: string[]) => t[0] === 'e' && t[1] === inbound.id));
+    assert.ok(e.content.includes('Private conversation fixture response'));
+    delivered = true;
+  }
+  function event(kind: number, tags: string[][], content: string, key = ownerSecret) {
+    const pubkey = publicKey(key); const created_at = Math.floor(Date.now() / 1000);
+    const id = createHash('sha256').update(JSON.stringify([0, pubkey, created_at, kind, tags, content])).digest('hex');
+    return { id, pubkey, created_at, kind, tags, content, sig: Buffer.from(schnorr.sign(id, key)).toString('hex') };
+  }
+  const members = event(39002, [['d', channel], ['p', agent], ['p', owner]], '');
+  const metadata = event(39000, [['d', channel], ['name', 'fixture'], ['type', 'channel']], '');
+  const inbound = event(9, [['h', channel], ['p', agent]], 'Please respond to this isolated fixture conversation.');
+  const http = createServer(async (req, res) => {
+    let body = ''; for await (const chunk of req) body += chunk;
+    res.setHeader('Content-Type', 'application/json');
+    if (req.url?.includes('/query')) {
+      const filters = JSON.parse(body); const serialized = JSON.stringify(filters);
+      res.end(JSON.stringify(serialized.includes('39002') ? [members] : serialized.includes('39000') ? [metadata] : []));
+    } else if (req.url?.includes('/events')) {
+      const e = JSON.parse(body); observeReply(e);
+      res.end(JSON.stringify({ id: e.id, accepted: true }));
+    } else res.end('[]');
+  });
+  const ws = new WebSocketServer({ server: http });
+  ws.on('connection', socket => {
+    socket.send(JSON.stringify(['AUTH', 'beehive-isolated-challenge']));
+    socket.on('message', bytes => {
+      const m = JSON.parse(bytes.toString());
+      if (m[0] === 'AUTH') {
+        const e = m[1];
+        assert.equal(e.pubkey, agent); assert.equal(e.kind, 22242);
+        assert.ok(e.tags.some((t: string[]) => t[0] === 'challenge' && t[1] === 'beehive-isolated-challenge'));
+        const hash = createHash('sha256').update(JSON.stringify([0, e.pubkey, e.created_at, e.kind, e.tags, e.content])).digest('hex');
+        assert.equal(e.id, hash); assert.ok(schnorr.verify(e.sig, hash, e.pubkey));
+        authenticated = true; socket.send(JSON.stringify(['OK', e.id, true, '']));
+      } else if (m[0] === 'EVENT') { observeReply(m[1]); socket.send(JSON.stringify(['OK', m[1].id, true, ''])); }
+      else if (m[0] === 'REQ') {
+        subscriptions++; socket.send(JSON.stringify(['EOSE', m[1]]));
+        if (m.slice(2).some((f: any) => f['#h']?.includes(channel))) setTimeout(() => socket.send(JSON.stringify(['EVENT', m[1], inbound])), 100);
+      }
+    });
+  });
+  await new Promise<void>(resolve => http.listen(0, '127.0.0.1', resolve));
+  const address = http.address(); assert.ok(address && typeof address !== 'string');
+  const session = new ConversationSession({ executable: realpathSync(process.env.BEEHIVE_REAL_BUZZ_ACP!), relay: `ws://127.0.0.1:${address.port}` }, {
+    executable: realpathSync(process.execPath), args: [resolve('test/conversation-harness-fixture.ts')], workspace: dir, home: dir, configDirectory: dir,
+    databricksHost: 'https://fixture.invalid', model: 'databricks-claude-haiku-4-5',
+  }, secret, owner, 10_000);
+  try {
+    await session.ready;
+    for (let i = 0; i < 100 && !subscriptions; i++) await delay(50);
+    assert.ok(authenticated, 'installed executable must authenticate with the provisioned key');
+    assert.ok(subscriptions > 0, 'installed executable must enter subscription loop');
+    const evidence = await session.verify();
+    assert.equal(evidence.session, 'conversation-session');
+    console.log(`isolated installed runtime: NIP-42 verified, subscriptions=${subscriptions}, harnessSpawned=${existsSync(join(dir, 'harness-pid'))}, same-session completed=true, signed threaded reply observed=${delivered}`);
+  } finally {
+    await session.stop(); for (const socket of ws.clients) socket.terminate();
+    await new Promise<void>(resolve => ws.close(() => resolve()));
+    await new Promise<void>(resolve => http.close(() => resolve()));
+    rmSync(dir, { recursive: true, force: true });
+  }
+});