From 8ba5fa9cd474aab5ce48fa255b194f8d3173aec1 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Thu, 10 Sep 2026 00:10:11 -0400 Subject: [PATCH] feat(beehive): isolate agent slots under one host installation Signed-off-by: Logan Johnson --- beehive/CHECKPOINT.md | 90 ++++++++++++++++ beehive/README.md | 58 ++++++++-- beehive/TESTING.md | 30 ++++++ beehive/src/cli.ts | 41 ++++++-- beehive/src/host.ts | 92 ++++++++++------ beehive/src/intents.ts | 4 +- beehive/src/slots.ts | 79 ++++++++++++++ beehive/test/slots.test.ts | 209 +++++++++++++++++++++++++++++++++++++ 8 files changed, 557 insertions(+), 46 deletions(-) create mode 100644 beehive/src/slots.ts create mode 100644 beehive/test/slots.test.ts diff --git a/beehive/CHECKPOINT.md b/beehive/CHECKPOINT.md index a260e008b1..26faa6f1fd 100644 --- a/beehive/CHECKPOINT.md +++ b/beehive/CHECKPOINT.md @@ -5,6 +5,96 @@ Base `051c3a270be9c73da9ab06700bcab7d5552fceaa`; continuation starts at `023c9274767ef50fa0f5b37ef1883336f8be59fd`. Candidate is the commit containing this checkpoint (`git rev-parse HEAD`). No merge/release. +## Slot continuation e0a5 — working single-installation slice + +Started from verified published `ebc8b3cb24fe0eefe5b8f239a88555786eadbf14`. +The candidate is the commit containing this checkpoint. One installation now owns +ONE exclusion lock, owner identity, management WS connection and heartbeat. The +existing admission/receipt/run state is an explicit in-process slot per persistent +agent, not N daemons. Routing happens BEFORE operation-ID reservation/cancellation; +queues, retractions, receipts, revisions, assignment and actual-run ownership remain +per agent. Selected-next still never mutates actual-run. No validator relaxation. + +`src/slots.ts` provides an explicit stopped `migrate-slots` upgrade: atomically +replace setup.json with version-2 installation inventory (shared harness setups + +agent keys/setup references), leaving the original journal byte-identical in place. +No key generation, copied key, re-enrollment or wiped history. Legacy single-slot +hosts remain supported without silent upgrade. Local `add-agent` creates a new key +(or explicitly imports a matching public-root standby), reuses host-owned default +harness, writes its inert per-agent journal before activating the manifest entry, +and refuses partial leftovers. Maximum 32 slots. Shared auth/conversation setup is +host-owned. Multiple selectable harness inventories and key/profile CRUD remain +unfinished, not hidden remote setup APIs. + +`host.close` attempts every slot via allSettled, preserving successful siblings' +stopped truth while quarantining/reporting uncertain execution and retaining the +single host lock. Shared management client reconciliation now queries every +unresolved host/agent pair, not merely the last agent per host. All slots hydrate +before dialing: initial relay history coalesced with socket open is accepted before +the ready-promise continuation, rather than discarded. Existing receipt fingerprint, +policy-blocked unchanged retry, admission cancellation and no late revival retained. + +Production-bound `test/slots.test.ts` proves: +- ONE real host subprocess + ONE management socket hosts X,Y. Actual TUI subprocess + selects exact numbered rows, Save/Start both, Stop X; ambiguous host-name shortcut + rejected. Y's journal stays byte-identical across X Stop, including receipt, + revision and actual run. Same operation ID on different slots is independent. +- Both delayed external ACP Starts enter transitioning concurrently with a shared + operation ID. Stop X cancels X while Y remains pending and later completes. Replay + returns their distinct terminal results without either journal changing. +- Wrong-agent/host operations cannot affect a run; actual host WS loss/recovery + replays each slot truthfully; graceful service restart retains assignment and + reports stopped instead of phantom live execution. Same-key standby elsewhere + still cannot Start. No transfer or source-unreachable takeover. +- Explicit migration after real Start/Stop preserves revision, two operations, + outbox and assignment byte-for-byte. Unknown first slot makes close fail/retain + lock but does not skip teardown of a healthy running sibling. +- Actual management-client reopen/reconcile queries each unresolved slot; starting + the real host later consumes initial relay history and resolves both intents. + +Final executable candidate: strict TypeScript + installed-enabled full package +**25/25 pass, zero skips**, 17.29s. Logs in workspace +`WORK_LOGS/BEEHIVE_DESIGN_183FFAF0/SLOTS_E0A5_EVIDENCE/{check.log,full.log}`. +Single-process proof logged public identities +`d2c1dbd61931eed22800b45de0cf422bd3f5e70b7585ec6f789e8124f2f21a9e` and +`ab3c21202f172cbc9d5ae1e07d5d0021ed9a34c1d56130af24124220c3df8e83`. +Installed binaries freshly rehashed, both match retained pins below. Existing +installed signed three-conversation seam also passed; slots themselves use external +lifecycle/ACP fixtures, not simultaneous live-provider conversations. + +Iteration failures retained honestly: the new TUI test initially timed out (twice, +one instrumented) because the test driver assumed a prompt ended a stdout chunk; +async status coalesced after it. Fixed the driver to consume prompt positions, no +product wait/sleep or assertion weakened. New reconciliation test then exposed +initial-history loss (one failure plus diagnostic confirmation); fixed at host +hydration/connection owner. These are distinct from the earlier unresolved +`reply-tool.test.ts:68` descendant observation. No containment code/assertion changed; +this 25/25 pass does NOT resolve that observation or prove Move-safe exit. + +Consumed independent `RECOVERY_VERIFICATION_8A67B967.md`: old frozen R1/U1 both +verified fixed, original replay reproducer and conflict/later-run controls green, +actual TUI byte-identical once-only retry/outbox recovery green, strict + installed +19/19. This closes the old recovery review, not new slots review or descendant exit. +The separate bounded descendant investigator has not delivered its causal report +at this checkpoint. Trusted non-cloned/non-rollback installations and exclusive +supervision remain assumptions; shared-owner signatures are not host attestation. + +**Move is NOT implemented and remains gated. ONE next executable action:** consume +the bounded descendant-exit report and bind its causal finding (or remaining probe) +to the actual owned-containment seam before implementing the source-consumed Move +transaction below: target preflight, complete owned source exit, atomic one-way +revocation/exact grant+outbox, idempotent target accept/fresh launch. No initial-host +validator relaxation, timeout resurrection, key/workspace/session transfer or new +controller/HTTP lifecycle service. Request independent immutable slot UX/correctness +review alongside that continuation; do not repeat closed old R1/U1 review. + +Full remaining parity stays in README: Restart, profiles/keys/metadata/history, +selectable harness/provider/preset/mesh/compute inventory, live model/admission, +retention and recovery. No owner credential/profile/OAuth reads, production/native +or Rust edits, full repo CI/GitHub CI, PR, merge or release. Publication uses existing +approved GitHub branch, configured Logan Johnson author/committer + DCO, no invented +cryptographic signer. + ## Assignment foundation handoff — delegated fb23e116 Governing boundary: key possession is not assignment. Provisioning must create a diff --git a/beehive/README.md b/beehive/README.md index 84dd36a21d..16941f7a87 100644 --- a/beehive/README.md +++ b/beehive/README.md @@ -132,8 +132,8 @@ assumes non-cloned, non-rollback, exclusively supervised installations. TUI pending intents/reconnect are implemented with a local encrypted journal. Lost connection means unknown, not stopped or success; exhausted recovery needs operator attention. - Inventory has freshness, not liveness proof. Host starts no action based on peer - agent presence. Only its locally bound agent is manageable; one slot per installation remains a - limitation, not multi-agent-per-host support. + agent presence. Only its locally bound slots are manageable; each has independent authority, + revisions, receipts, admission and process ownership under one installation. ## Explicit initial assignment (Move prerequisite, not Move) @@ -171,16 +171,60 @@ A partial setup remains inert and requires local investigation, not automatic re `agents` groups observations by persistent public key; `hosts` numbers host/agent rows, shows reported assignment, and `select` routes to the exact row. A unique host name remains a backward-compatible shortcut. Observations are not consensus; an -unreachable source never makes its standby eligible. This candidate still has one -agent slot per installation. Tests use three distinct installations for two agent -identities; they do **not** disguise cloned daemons as multi-agent slots on one host. +unreachable source never makes its standby eligible. The original assignment test uses three distinct installations. The newer +`slots.test.ts` separately proves two identities in ONE host process/connection; +these are independent agent slots, not cloned daemons. **Not yet implemented:** destination preflight, source reservation/whole-tree Stop as a transfer, irreversible grant/outbox lineage, target acceptance/fresh launch, -and several independently managed agents in one host process. No `move` command +and source-consumed Move lineage. No `move` command is exposed until these boundaries are implemented and tested together. The root ID is an identity binding, not a lock service or partition/rollback solution. +## Several agents, one installation + +Upgrade an existing **stopped**, assignment-pinned installation explicitly, then +add an independent identity using the same locally trusted harness setup: + +```sh +node src/cli.ts migrate-slots /absolute/host-directory +node src/cli.ts add-agent /absolute/host-directory +node src/cli.ts host /absolute/host-directory ws://127.0.0.1:19481 +``` + +Both commands ask for confirmation and refuse a live installation lock. Migration +replaces only setup inventory atomically; the original journal stays byte-identical +with its assignment, revision, operation receipts and outbox. It never generates or +copies the existing key. Legacy single-slot hosts still run without silent upgrade. +Pre-assignment installations must use `migrate-assignment` first. Missing journals +and partial additions fail closed, not an invitation to re-enroll an identity. + +`add-agent` creates a NEW key, or accepts local key/root files for a standby import +using the same optional-file convention as `setup`. Host-owned `default` harness, +executable/auth context and conversation setup are shared; agent keys are not part +of that reusable harness. A version-2 owner-only `setup.json` holds one installation +identity, setup inventory and per-agent key/setup references. New authority journals +live under `agents//journal.json`; these directories cannot run daemons. +The original slot keeps its journal in place. Maximum 32 slots per installation. +Changing local conversation setup requires every slot stopped and updates the shared +harness; `auth-info` prints the shared service-user context without signing in. + +Use remote TUI `hosts` then `select ` to configure/start/stop each agent. +A host-name shortcut is rejected when ambiguous. `agents` groups persistent public +identities and labels reported assignment separately from the observing host. Each +slot retains independent selected-next/actual-run, revision, operation IDs/outbox and +admission queue. Stop X does not stop Y or change Y's receipt/revision. One host +process holds ONE installation lock, heartbeat and management connection. Reconnect +replays each slot's receipts; client reconciliation queries every unresolved +host/agent pair. Host shutdown attempts every slot even if one is uncertain, reports +failure and retains the installation lock rather than falsely declaring all stopped. + +Setup/key enrollment remains local and offline; ordinary selections and lifecycle +remain remote. Multiple selectable harness inventory entries, profile CRUD, key +revocation UX, Restart and Move are still future work. Standby X elsewhere still +cannot Start, even while source X is stopped/unreachable. This is trusted-installation +coordination, not physical host attestation, partition safety or exactly-once execution. + ## Real target: Buzz Agent + Databricks v2 Local setup option 2 now takes **buzz-agent**, not buzz-acp. It creates fresh @@ -291,7 +335,7 @@ remain explicit local/operator actions. No live Databricks request has occurred. | Surface | This slice | Required next work | |---|---|---| | Local owner/key setup | New owner + separately generated agent key | nsec import, saved-owner verification, revocation/key removal UX | -| Host inventory | Real relay, fixed authority, freshness | multiple setups/agents, service install, reconnect/reconciliation | +| Host inventory | Real relay, multiple independent slots, shared host harness, freshness | multiple selectable setups, service install, richer 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, production protocol compatibility and catalog provenance | diff --git a/beehive/TESTING.md b/beehive/TESTING.md index 4996cf5a54..cdbb463b79 100644 --- a/beehive/TESTING.md +++ b/beehive/TESTING.md @@ -164,3 +164,33 @@ migration preserves revision/history and refuses replacement. This is assignment bootstrap evidence, NOT Move or multiple agent slots on one host evidence. All existing fixture setups now explicitly provision initial authority; none relies on host startup self-assigning from key presence. + +## Single-installation slot continuation + +`slots.test.ts` uses the production host/client/relay and real external fixtures: +- ONE actual host subprocess exposes X and Y over exactly ONE management socket. + Actual TUI subprocess selects numbered rows, saves/starts both, stops X, rejects + ambiguous host-name selection and shows persistent identities. Y's journal stays + byte-identical across X Stop, including revision, actual run and receipt. +- A reused operation ID on another slot remains independent. Wrong-agent/host + operations cannot affect a run. Host socket loss replays both inventories/outbox; + graceful process restart retains assignments and truthful stopped state. Same-key + standby in another process never obtains Start authority. +- Concurrent delayed ACP Starts with the same operation ID reach transitioning + independently. Stop X cancels X while Y is still pending; Y completes. Replays + return distinct terminal receipts without mutating either journal. +- Explicit migration after real Start/Stop preserves the complete journal byte for + byte. An uncertain first slot makes host close reject/retain the lock while the + healthy sibling is nevertheless torn down and saved stopped. +- Actual management-client reopen/reconcile emits an inspect for EACH unresolved + host/agent pair. Starting the real host later consumes initial relay history and + resolves both retained intents. Slots are fully hydrated before dialing, so + history coalesced with socket open cannot be silently dropped awaiting ready. + +These are lifecycle/ACP fixture proofs, not simultaneous live provider/conversation +proof or Move acceptance. First new TUI harness iteration timed out because its +prompt parser assumed prompts ended stdout chunks; corrected to consume prompt +positions amid coalesced async status reports. A subsequent new reconciliation test +exposed the initial-history hydration gap above, fixed at the host connection owner. +No containment assertion, delay or existing test was weakened. The historical +reply-tool descendant-exit observation remains a separate open investigation. diff --git a/beehive/src/cli.ts b/beehive/src/cli.ts index 7362fcdf79..93140f5144 100644 --- a/beehive/src/cli.ts +++ b/beehive/src/cli.ts @@ -4,6 +4,7 @@ import { existsSync, mkdirSync, realpathSync, rmdirSync } from 'node:fs'; import { resolve, join, dirname } from 'node:path'; import { relay } from './relay.ts'; import { host, validateSetup, provision, migrateAssignment } from './host.ts'; +import { migrateSlots, addSlot, installationSlots, saveDefaultHarness } from './slots.ts'; import { managementClient } from './intents.ts'; import { message, newKey, publicKey, object, text, type Message } from './protocol.ts'; import { readPrivate, writePrivate } from './storage.ts'; @@ -15,6 +16,8 @@ const [command, ...args] = process.argv.slice(2); const help = `Beehive — isolated development preview (loopback relay only) identity Create a NEW local owner identity setup Guided local harness + key setup + migrate-slots Explicit stopped upgrade, preserves journal + add-agent New identity using shared local harness assignment-export Export public pinned genesis locally migrate-assignment Explicit stopped legacy enrollment auth-info Print exact local sign-in context (no login) @@ -22,7 +25,8 @@ const help = `Beehive — isolated development preview (loopback relay only) relay Dedicated ciphertext relay host Persistent foreground host tui Relay-connected terminal UI -setup accepts optional for standby import. +setup/add-agent accept optional for standby import. +assignment-export accepts agent public key after filename when several slots exist. No Move, provider login RPC or production relay support yet.`; async function main() { if (command === 'identity') { @@ -56,8 +60,30 @@ async function main() { console.log('Local setup saved. Start the host separately; the TUI never owns its lifetime.'); 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 === 'migrate-slots') { + const dir = resolve(text(args[0])); + const ui = createInterface({ input: stdin, output: stdout }); + try { + if (await ui.question('Upgrade this stopped installation to shared agent slots, preserving keys and journal? [yes/no]: ') !== 'yes') return; + migrateSlots(dir); console.log('Slots enabled; original assignment and lifecycle history retained.'); + } finally { ui.close(); } + } else if (command === 'add-agent') { + const dir = resolve(text(args[0])); + const first = installationSlots(dir)[0]!.setup; + const importing = args.length > 1; + const secret = importing ? text(object(readPrivate(resolve(text(args[1])))).secret) : newKey(); + const root = importing ? validateGenesis(readPrivate(resolve(text(args[2])))) : createGenesis(publicKey(first.ownerSecret), publicKey(secret), first.host); + if (importing && root.initialHost === first.host) throw Error('Import cannot recreate initial authority'); + const ui = createInterface({ input: stdin, output: stdout }); + try { + if (await ui.question(importing ? `Add standby identity assigned to ${root.initialHost}, reusing host harness default? [yes/no]: ` : 'Create independent NEW agent assigned here, reusing host harness default? [yes/no]: ') !== 'yes') return; + addSlot(dir, secret, root); console.log(`Agent ${publicKey(secret)} added; start/configure remotely after restarting the host.`); + } finally { ui.close(); } } else if (command === 'assignment-export') { - const state = object(readPrivate(join(resolve(text(args[0])), 'journal.json'))); + const entries = installationSlots(resolve(text(args[0]))); + const entry = args[2] ? entries.find(e => publicKey(e.setup.agentSecret) === args[2]) : entries.length === 1 ? entries[0] : undefined; + if (!entry) throw Error('Specify agent public key after export filename for multi-slot host'); + const state = object(readPrivate(entry.path)); const genesis = validateGenesis(object(state.assignment).genesis); writePrivate(resolve(text(args[1])), genesis, true); console.log('Public genesis exported; no keys or execution grant. Provision the matching key separately on the destination.'); @@ -75,8 +101,9 @@ async function main() { const lock = join(dir, 'host.lock'); mkdirSync(lock, { mode: 0o700 }); // Same atomic exclusion as host startup; never remove a competing lock. try { - if (existsSync(join(dir, 'journal.json')) && object(readPrivate(join(dir, 'journal.json'))).phase !== 'stopped') throw Error('Reconcile the prior run before local setup changes'); - const setup = validateSetup(readPrivate(join(dir, 'setup.json'))); + const entries = installationSlots(dir); + if (entries.some(e => object(readPrivate(e.path)).phase !== 'stopped')) throw Error('Reconcile every prior run before local setup changes'); + const setup = entries[0]!.setup; if (setup.mode !== 'buzz-agent-databricks-v2') throw Error('Provision Buzz Agent first; this action never creates or replaces agent keys'); const ui = createInterface({ input: stdin, output: stdout }); try { @@ -89,12 +116,12 @@ async function main() { const plan = prepareConversation(conversation, { executable: setup.runner, args: setup.args, workspace: setup.workspace, home: text(setup.serviceHome), configDirectory: text(setup.configDirectory), databricksHost: text(setup.databricksHost), model: 'databricks-claude-haiku-4-5' }, setup.agentSecret, publicKey(setup.ownerSecret)); if ((await ui.question('Save onto existing identity? Start waits for an admitted conversation and local provider sign-in. [yes/no]: ')) !== 'yes') return; // One atomic replacement; keys, assignment and provider auth context unchanged. - writePrivate(join(dir, 'setup.json'), { ...setup, conversation }); + saveDefaultHarness(dir, { ...setup, conversation }); console.log(`Conversation setup saved for existing agent ${plan.agentPublicKey}. No network connection, login or process started. The community operator must admit this public key and owner to the chosen relay/channels (or provision a valid owner attestation locally). Admission remains unverified. Start uses the host-owned ACP broker; the Buzz CLI tool, when enabled, uses agent membership authority across conversations without local thread configuration.`); } finally { ui.close(); } } finally { rmdirSync(lock); } } else if (command === 'auth-info') { - const setup = validateSetup(readPrivate(join(resolve(text(args[0])), 'setup.json'))); + const setup = installationSlots(resolve(text(args[0])))[0]!.setup; 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') { @@ -103,7 +130,7 @@ async function main() { process.once('SIGINT',() => { for (const c of server.clients) c.close(); server.close(); }); } else if (command === 'host') { const running = await host(resolve(text(args[0])),text(args[1])); - console.log(`Host online; agent ${running.agent}. Ctrl-C stops owned runner before host exits.`); + console.log(`Host online; agents ${running.agents.join(', ')}. Ctrl-C stops owned runner before host exits.`); process.once('SIGINT',() => { void running.close(); }); process.once('SIGTERM',() => { void running.close(); }); } else if (command === 'tui') { diff --git a/beehive/src/host.ts b/beehive/src/host.ts index f4a70b2fc8..10c92db30b 100644 --- a/beehive/src/host.ts +++ b/beehive/src/host.ts @@ -1,6 +1,7 @@ import { accessSync, constants, existsSync, mkdirSync, rmdirSync, realpathSync, statSync, readFileSync } from 'node:fs'; import { join, isAbsolute } from 'node:path'; import { createHash } from 'node:crypto'; +import { installationSlots } from './slots.ts'; import { validateGenesis, type Genesis } from './assignment.ts'; import { connect, validateRelayURL } from './client.ts'; import { digest, fields, message, object, publicKey, text, type Message } from './protocol.ts'; @@ -28,7 +29,8 @@ export function provision(directory: string, value: Setup, root: Genesis): void writePrivate(join(directory, 'setup.json'), setup, true); writePrivate(join(directory, 'journal.json'), initialState(setup, genesis), true); } -function initialState(setup: Setup, genesis: Genesis): State { +/** Initial stopped authority shared by legacy provisioning and slot enrollment. */ +export function initialState(setup: Setup, genesis: Genesis): State { return { assignment: { genesis, assignedHost: genesis.initialHost }, binding: { host: setup.host, owner: publicKey(setup.ownerSecret), agent: publicKey(setup.agentSecret) }, revision: 0, phase: 'stopped', selected: { model: setup.mode === 'fixture' ? 'fixture-model' : 'databricks-claude-haiku-4-5', workspace: setup.workspace, profile: 'default' }, actual: null, operations: {}, outbox: [], @@ -48,22 +50,17 @@ export function migrateAssignment(directory: string, genesis: Genesis): void { } finally { rmdirSync(lock); } } /** Hosts consult durable assignment, never key presence or relay inventory, for authority. */ -export async function host(directory: string, url: string) { - validateRelayURL(url); - const setup = validateSetup(readPrivate(join(directory,'setup.json'))); +function slot(setup: Setup, path: string, currentSetup: () => Setup, publish: (m: Message) => void) { const agent = publicKey(setup.agentSecret); - const lock = join(directory,'host.lock'); - const path = join(directory,'journal.json'); if (!existsSync(path)) throw Error('Authority journal missing; restore/reconcile locally, never re-enroll from key possession'); let state = readPrivate(path) as State; if (state.binding.host !== setup.host || state.binding.owner !== publicKey(setup.ownerSecret) || state.binding.agent !== agent) throw Error('Saved ownership/assignment mismatch'); if (!state.assignment) throw Error('Legacy assignment requires explicit local migrate-assignment while stopped'); const genesis = validateGenesis(state.assignment.genesis); if (genesis.owner !== state.binding.owner || genesis.agent !== agent || state.assignment.assignedHost !== genesis.initialHost) throw Error('Saved assignment mismatch'); - mkdirSync(lock, { mode: 0o700 }); // No PID-only stale-lock recovery. if (state.phase !== 'stopped') state.phase = 'quarantined'; const save = () => writePrivate(path,state); - try { save(); } catch (e) { rmdirSync(lock); throw e; } + save(); let owned: OwnedProcess | ConversationSession | undefined; let acp: AgentSession | ConversationSession | undefined; let catalog: Catalog | { state: 'not-probed' | 'failed'; authentication: 'unverified' } = { state: 'not-probed', authentication: 'unverified' }; @@ -77,7 +74,6 @@ export async function host(directory: string, url: string) { } let closing = false; let queue = Promise.resolve(); - let initialized = false; // Exact-fenced Stop retractions received ahead of their serialized handling, keyed // by that Stop's operation ID. One invariant covers every not-yet-committed Start // admission at the retracted revision, independent of callback/microtask timing: @@ -89,9 +85,7 @@ export async function host(directory: string, url: string) { const queuedOperations = new Set(); const retracted = new Map(); const retracting = (revision: number) => { for (const target of retracted.values()) if (target === revision) return true; return false; }; - let client: ReturnType; - try { client = connect(url,setup.ownerSecret,m => { - if (!initialized) return; + function receive(m: Message) { // Cancellation is signalled outside the serialized mutation queue. Admission is // still exact-authority/revision and never bypasses durable receipt processing. const operation = m.host === setup.host && ['save','start','stop'].includes(m.type); @@ -105,15 +99,11 @@ export async function host(directory: string, url: string) { if (state.phase === 'transitioning' && state.actual) acp?.cancel(); } queue = queue.then(() => handle(m)).catch(() => { state.phase = 'quarantined'; save(); }); - }, () => { - if (closing || !initialized) return; - // Replay committed receipts after transport recovery. Incoming relay history - // still traverses the fingerprint/revision fence; reconnect cannot repeat effects. + } + function replay() { for (const m of state.outbox) publish(m); publish(inventory()); - }); } - 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 (!owned) throw Error('Unknown ownership; quarantined'); await owned.stop(); owned = undefined; acp = undefined; @@ -150,7 +140,7 @@ export async function host(directory: string, url: string) { if (retracting(m.revision)) throw Error('Start cancelled by concurrent Stop'); // Lifecycle never restores deleted keys from the in-memory setup snapshot. try { - const current = validateSetup(readPrivate(join(directory, 'setup.json'))); + const current = currentSetup(); if (current.agentSecret !== setup.agentSecret || current.ownerSecret !== setup.ownerSecret || current.host !== setup.host) throw Error('changed'); } catch { throw Error('Local agent key/setup missing or changed; repair locally without resetting assignment'); } accessSync(setup.runner,constants.X_OK); @@ -215,23 +205,65 @@ 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()); } - 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(() => { + return { agent, receive, replay, heartbeat() { if (owned && (owned.exited || acp?.healthy === false) && state.phase === 'running') { state.phase = 'quarantined'; save(); } publish(inventory()); - },2000); - return { agent, async close() { + }, async close() { if (closing) return; closing = true; - client.close(); clearInterval(heartbeat); acp?.cancel(); await queue; + acp?.cancel(); await queue; 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. + save(); + if (state.phase !== 'stopped') throw Error(`Agent ${agent}: unknown execution; installation remains locked`); + } }; +} + +/** One installation owns every slot, lock and management transport. */ +export async function host(directory: string, url: string) { + validateRelayURL(url); + const lock = join(directory, 'host.lock'); + mkdirSync(lock, { mode: 0o700 }); // Never infer ownership from a recovered PID. + let client: ReturnType | undefined; + let initialized = false; + const slots = new Map>(); + const publish = (m: Message) => { try { client?.send(m); } catch { /* Slot outbox survives transport loss. */ } }; + try { + const entries = installationSlots(directory); + for (const entry of entries) { + slots.set(publicKey(entry.setup.agentSecret), slot(entry.setup, entry.path, () => { + const current = installationSlots(directory).find(e => publicKey(e.setup.agentSecret) === publicKey(entry.setup.agentSecret)); + if (!current) throw Error('Key removed'); + return current.setup; + }, publish)); + } + const setup = entries[0]!.setup; + // Every slot is hydrated before dialing. Initial WS history may arrive in the + // same event-loop turn as open, before the ready promise continuation. + initialized = true; + client = connect(url, setup.ownerSecret, m => { + if (!initialized || m.host !== setup.host) return; + const selected = slots.get(m.agent); + if (selected) selected.receive(m); + else if (m.type === 'inspect') { for (const s of slots.values()) s.replay(); } + else if (['save','start','stop'].includes(m.type)) publish(message('receipt',setup.host,m.agent,0,{ operation: m.id, fingerprint: digest(JSON.stringify(m)).toString('hex'), result: 'not-authority' })); + }, () => { if (initialized) for (const s of slots.values()) s.replay(); }); + await client.ready; + } catch (e) { client?.close(); rmdirSync(lock); throw e; } + initialized = true; + for (const s of slots.values()) s.replay(); + const heartbeat = setInterval(() => { for (const s of slots.values()) s.heartbeat(); }, 2000); + let closing: Promise | undefined; + return { agent: [...slots.keys()][0]!, agents: [...slots.keys()], close() { + return closing ??= (async () => { + initialized = false; client?.close(); clearInterval(heartbeat); + // allSettled starts every teardown and preserves each slot's durable truth. + const results = await Promise.allSettled([...slots.values()].map(s => s.close())); + const errors = results.filter((r): r is PromiseRejectedResult => r.status === 'rejected').map(r => r.reason); + if (errors.length) throw new AggregateError(errors, 'Incomplete owned teardown; installation remains locked'); + rmdirSync(lock); + })(); } }; } diff --git a/beehive/src/intents.ts b/beehive/src/intents.ts index 019cf6bc06..870a87b19e 100644 --- a/beehive/src/intents.ts +++ b/beehive/src/intents.ts @@ -57,8 +57,8 @@ export function managementClient(root: string, url: string, secret: string, rece try { transport.sendEnvelope(intent.envelope); } catch { break; } // Intent remains durable. } // Query host outbox too: relay history alone may lack a lost terminal receipt. - const hosts = new Map([...intents.values()].filter(i => !i.receipt).map(i => [i.request.host, i.request])); - for (const request of hosts.values()) { + const slots = new Map([...intents.values()].filter(i => !i.receipt).map(i => [JSON.stringify([i.request.host, i.request.agent]), i.request])); + for (const request of slots.values()) { try { transport.send(message('inspect', request.host, request.agent)); } catch { break; } } changed(); diff --git a/beehive/src/slots.ts b/beehive/src/slots.ts new file mode 100644 index 0000000000..d562d688f7 --- /dev/null +++ b/beehive/src/slots.ts @@ -0,0 +1,79 @@ +import { existsSync, mkdirSync, rmdirSync } from 'node:fs'; +import { join } from 'node:path'; +import { validateSetup, initialState, type Setup } from './host.ts'; +import { readPrivate, writePrivate } from './storage.ts'; +import { object, publicKey } from './protocol.ts'; +import { validateGenesis, type Genesis } from './assignment.ts'; + +type Harness = Omit; +type Installation = { version: 2; host: string; ownerSecret: string; setups: Record; agents: Record }; +function readInstallation(directory: string): Installation { + const i = object(readPrivate(join(directory, 'setup.json'))) as Installation; + if (i.version !== 2) throw Error('Explicit stopped migrate-slots required'); + const keys = Object.keys(object(i.agents)); + if (!keys.length || keys.length > 32) throw Error('Installation requires 1..32 slots'); + object(i.setups); + for (const key of keys) { + const a = i.agents[key]!; + if (publicKey(a.secret) !== key || !Object.hasOwn(i.setups, a.setup)) throw Error('Invalid slot binding'); + validateSetup({ ...i.setups[a.setup], host: i.host, ownerSecret: i.ownerSecret, agentSecret: a.secret }); + } + return i; +} +/** Resolve shared local harness inventory without copying agent keys into setups. */ +export function installationSlots(directory: string): { setup: Setup; path: string }[] { + const raw = object(readPrivate(join(directory, 'setup.json'))); + if (raw.version !== 2) return [{ setup: validateSetup(raw), path: join(directory, 'journal.json') }]; + const i = readInstallation(directory); + return Object.entries(i.agents).map(([key, a]) => ({ + setup: validateSetup({ ...i.setups[a.setup], host: i.host, ownerSecret: i.ownerSecret, agentSecret: a.secret }), + path: a.legacy ? join(directory, 'journal.json') : join(directory, 'agents', key, 'journal.json'), + })); +} +/** Explicit single stopped upgrade: one atomic setup replacement; journal stays byte-identical. */ +export function migrateSlots(directory: string): void { + const lock = join(directory, 'host.lock'); mkdirSync(lock, { mode: 0o700 }); + try { + const raw = object(readPrivate(join(directory, 'setup.json'))); + if (raw.version === 2) throw Error('Slots already enabled'); + const s = validateSetup(raw); + const state = object(readPrivate(join(directory, 'journal.json'))); + const binding = object(state.binding); const assignment = object(state.assignment); + const genesis = validateGenesis(assignment.genesis); + if (state.phase !== 'stopped' || state.actual !== null || binding.host !== s.host || binding.owner !== publicKey(s.ownerSecret) || binding.agent !== publicKey(s.agentSecret) || genesis.owner !== binding.owner || genesis.agent !== binding.agent || assignment.assignedHost !== genesis.initialHost) throw Error('Slot migration requires matching stopped assignment'); + const { host, ownerSecret, agentSecret, ...harness } = s; + const i: Installation = { version: 2, host, ownerSecret, setups: { default: harness }, agents: { [publicKey(agentSecret)]: { secret: agentSecret, setup: 'default', legacy: true } } }; + writePrivate(join(directory, 'setup.json'), i); + } finally { rmdirSync(lock); } +} +/** Local key enrollment reuses a host-owned harness, never another agent's key. */ +export function addSlot(directory: string, secret: string, genesis: Genesis, setupName = 'default'): void { + const lock = join(directory, 'host.lock'); mkdirSync(lock, { mode: 0o700 }); + try { + const i = readInstallation(directory); const key = publicKey(secret); + if (Object.keys(i.agents).length >= 32 || Object.hasOwn(i.agents, key)) throw Error('Slot exists or installation full'); + if (!Object.hasOwn(i.setups, setupName)) throw Error('Unknown host harness setup'); + const setup = validateSetup({ ...i.setups[setupName], host: i.host, ownerSecret: i.ownerSecret, agentSecret: secret }); + const slotDirectory = join(directory, 'agents', key); + if (existsSync(slotDirectory)) throw Error('Partial slot requires local reconciliation; cannot reset'); + // The journal is inert until the manifest activates it; never copy setup/keys. + provisionSlotJournal(slotDirectory, setup, genesis); + i.agents[key] = { secret, setup: setupName }; + writePrivate(join(directory, 'setup.json'), i); + } finally { rmdirSync(lock); } +} +function provisionSlotJournal(directory: string, setup: Setup, root: Genesis) { + const genesis = validateGenesis(root); + if (genesis.owner !== publicKey(setup.ownerSecret) || genesis.agent !== publicKey(setup.agentSecret)) throw Error('Genesis ownership mismatch'); + writePrivate(join(directory, 'journal.json'), initialState(setup, genesis), true); +} + +/** Caller holds the installation lock; update shared harness inventory without key copies. */ +export function saveDefaultHarness(directory: string, setup: Setup): void { + const raw = object(readPrivate(join(directory, 'setup.json'))); + if (raw.version !== 2) { writePrivate(join(directory, 'setup.json'), setup); return; } + const i = readInstallation(directory); + const { host: _host, ownerSecret: _owner, agentSecret: _agent, ...harness } = setup; + i.setups.default = harness; + writePrivate(join(directory, 'setup.json'), i); +} diff --git a/beehive/test/slots.test.ts b/beehive/test/slots.test.ts new file mode 100644 index 0000000000..112f9194aa --- /dev/null +++ b/beehive/test/slots.test.ts @@ -0,0 +1,209 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { spawn, type ChildProcess } from 'node:child_process'; +import { once } from 'node:events'; +import { mkdtempSync, realpathSync, rmSync, readFileSync, existsSync } from 'node:fs'; +import { join, resolve } from 'node:path'; +import { tmpdir } from 'node:os'; +import { setTimeout as delay } from 'node:timers/promises'; +import { createGenesis } from '../src/assignment.ts'; +import { provision, host, type Setup } from '../src/host.ts'; +import { migrateSlots, addSlot } from '../src/slots.ts'; +import { newKey, publicKey, message, type Message } from '../src/protocol.ts'; +import { writePrivate } from '../src/storage.ts'; +import { relay } from '../src/relay.ts'; +import { connect } from '../src/client.ts'; +import { managementClient } from '../src/intents.ts'; + +async function until(predicate: () => boolean) { + for (let i = 0; i < 400; i++) { if (predicate()) return; await delay(20); } + throw Error('Missing single-installation evidence'); +} +async function closeChild(child: ChildProcess) { + if (child.exitCode !== null || child.signalCode !== null) return; + const done = once(child, 'exit'); child.kill('SIGTERM'); await done; +} + +test('one real host process/connection: X and Y TUI lifecycle, independent receipts, reconnect, restart, standby denial', { timeout: 30000 }, async () => { + const dir = realpathSync(mkdtempSync(join(tmpdir(), 'beehive-slots-'))); + const ownerSecret = newKey(), x = newKey(), y = newKey(); const X = publicKey(x), Y = publicKey(y); + const source = join(dir, 'source'); const target = join(dir, 'target'); + const setup: Setup = { host: 'source', ownerSecret, agentSecret: x, runner: process.execPath, args: [resolve('test/runner.ts')], workspace: dir, mode: 'fixture' }; + const gx = createGenesis(publicKey(ownerSecret), X, 'source'); + provision(source, setup, gx); + const before = readFileSync(join(source, 'journal.json')); + migrateSlots(source); assert.deepEqual(readFileSync(join(source, 'journal.json')), before); + addSlot(source, y, createGenesis(publicKey(ownerSecret), Y, 'source')); + const manifest = JSON.parse(readFileSync(join(source, 'setup.json'), 'utf8')); + assert.equal(Object.keys(manifest.setups).length, 1); assert.equal(Object.keys(manifest.agents).length, 2); + assert.equal(existsSync(join(source, 'agents', Y, 'setup.json')), false, 'no second daemon setup/key copy'); + provision(target, { ...setup, host: 'standby' }, gx); + writePrivate(join(dir, 'identity.json'), { secret: ownerSecret }); + 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 seen: Message[] = []; const children: ChildProcess[] = []; + const ui = connect(url, ownerSecret, m => seen.push(m)); await ui.ready; + const journal = (agent: string) => JSON.parse(readFileSync(agent === X ? join(source, 'journal.json') : join(source, 'agents', agent, 'journal.json'), 'utf8')); + async function launch(path: string) { + const child = spawn(process.execPath, ['src/cli.ts', 'host', path, url], { stdio: ['ignore', 'pipe', 'pipe'] }); children.push(child); + let output = ''; child.stdout?.on('data', chunk => { output += chunk.toString(); }); child.stderr?.resume(); + await until(() => output.includes('Host online')); return child; + } + async function receipt(m: Message) { ui.send(m); await until(() => seen.some(r => r.type === 'receipt' && r.body.operation === m.id && r.agent === m.agent)); return seen.find(r => r.type === 'receipt' && r.body.operation === m.id && r.agent === m.agent)!; } + try { + let processHost = await launch(source); + await until(() => seen.some(m => m.type === 'inventory' && m.agent === Y)); + assert.equal(server.clients.size, 2, 'exactly ONE host management socket plus observation client'); + assert.equal(existsSync(join(source, 'host.lock')), true); + await assert.rejects(host(source, url), /EEXIST/); + assert.throws(() => addSlot(source, newKey(), gx), /EEXIST/); + // Actual remote TUI selects numbered slots, saves and starts both, then stops X. + const terminal = spawn(process.execPath, ['src/cli.ts', 'tui', join(dir, 'identity.json'), url], { stdio: ['pipe', 'pipe', 'pipe'] }); children.push(terminal); + let output = ''; let index = 0; + const steps: { prompt: string; value: string; gate?: () => boolean }[] = [ + { prompt: 'beehive> ', value: 'hosts' }, { prompt: 'beehive> ', value: 'select source' }, + { prompt: 'beehive> ', value: 'select 1' }, { prompt: 'beehive> ', value: 'save' }, + { prompt: 'Model: ', value: 'fixture-model' }, { prompt: 'Workspace: ', value: dir }, { prompt: 'Behavior profile: ', value: 'default' }, + { prompt: 'beehive> ', value: 'start', gate: () => journal(X).revision === 1 }, + { prompt: 'beehive> ', value: 'select 2', gate: () => journal(X).phase === 'running' }, + { prompt: 'beehive> ', value: 'save' }, { prompt: 'Model: ', value: 'fixture-model' }, { prompt: 'Workspace: ', value: dir }, { prompt: 'Behavior profile: ', value: 'default' }, + { prompt: 'beehive> ', value: 'start', gate: () => journal(Y).revision === 1 }, + { prompt: 'beehive> ', value: 'show', gate: () => journal(Y).phase === 'running' }, + { prompt: 'beehive> ', value: 'select 1' }, { prompt: 'beehive> ', value: 'stop' }, + { prompt: 'beehive> ', value: 'agents', gate: () => journal(X).phase === 'stopped' }, { prompt: 'beehive> ', value: 'quit' }, + ]; + let pending = false; let cursor = 0; + terminal.stdout.on('data', chunk => { + output += chunk.toString(); const step = steps[index]; + if (pending || !step || !output.slice(cursor).includes(step.prompt)) return; + pending = true; + void (async () => { if (step.gate) await until(step.gate); await delay(100); index++; cursor = output.length; pending = false; terminal.stdin.write(`${step.value}\n`); })(); + }); terminal.stderr.resume(); + const [code] = await once(terminal, 'exit'); assert.equal(code, 0); + assert.match(output, /Unknown or ambiguous host/); + assert.match(output, new RegExp(`Selected source agent ${X}`)); assert.match(output, new RegExp(`Selected source agent ${Y}`)); + assert.match(output, /"phase": "running"/); + assert.equal(journal(X).revision, 3); assert.equal(journal(Y).revision, 2); assert.equal(journal(Y).phase, 'running'); + const yBefore = readFileSync(join(source, 'agents', Y, 'journal.json')); + const previous = Object.values(journal(Y).operations).map((o: any) => o.reply).find((r: Message) => r.body.result === 'accepted') as Message; + assert.ok(previous); + // Same operation ID on a different slot is not a receipt reservation conflict. + const stopX = message('stop', 'source', X, 3); stopX.id = String(previous.body.operation); + assert.equal((await receipt(stopX)).body.result, 'accepted'); + assert.deepEqual(readFileSync(join(source, 'agents', Y, 'journal.json')), yBefore, 'Stop X cannot mutate Y receipt/revision/run'); + assert.equal((await receipt(message('stop', 'source', publicKey(newKey()), 0))).body.result, 'not-authority'); + ui.send(message('stop', 'wrong-host', Y, 2)); await delay(100); assert.equal(journal(Y).phase, 'running'); + // Drop only the host's socket. Both slots replay receipts through ONE replacement. + const hostSocket = [...server.clients][1]!; seen.length = 0; hostSocket.terminate(); + await until(() => seen.some(m => m.type === 'receipt' && m.body.operation === previous.body.operation)); + await until(() => seen.some(m => m.type === 'inventory' && m.agent === Y && m.body.phase === 'running')); + assert.equal(server.clients.size, 2); assert.deepEqual(readFileSync(join(source, 'agents', Y, 'journal.json')), yBefore); + await launch(target); + assert.equal((await receipt(message('start', 'standby', X, 0))).body.result, 'not-authority'); + await closeChild(processHost); assert.equal(journal(Y).phase, 'stopped'); assert.equal(journal(Y).revision, 2); + seen.length = 0; processHost = await launch(source); + await until(() => seen.some(m => m.type === 'inventory' && m.agent === Y && m.host === 'source' && m.body.phase === 'stopped')); + assert.equal(journal(X).assignment.assignedHost, 'source'); assert.equal(journal(Y).assignment.assignedHost, 'source'); + assert.equal((await receipt(message('start', 'source', Y, 2))).body.result, 'accepted'); + assert.equal((await receipt(message('start', 'standby', X, 0))).body.result, 'not-authority'); + console.log(`Single host PID ${processHost.pid}; identities ${X}, ${Y}; one connection, shared setup, independent journals`); + } finally { + for (const child of children) await closeChild(child); + ui.close(); for (const c of server.clients) c.terminate(); + await new Promise(resolve => server.close(() => resolve())); rmSync(dir, { recursive: true, force: true }); + } +}); + +test('per-slot concurrent ACP admission: cancelling X leaves pending Y independent and receipts replay per agent', { timeout: 20000 }, async () => { + const dir = realpathSync(mkdtempSync(join(tmpdir(), 'beehive-slots-cancel-'))); + const ownerSecret = newKey(), x = newKey(), y = newKey(); const X = publicKey(x), Y = publicKey(y); + const setup: Setup = { host: 'both', ownerSecret, agentSecret: x, runner: process.execPath, args: [resolve('test/acp-fixture.ts'), 'delayed'], workspace: dir, mode: 'buzz-agent-databricks-v2', serviceHome: dir, configDirectory: dir, databricksHost: 'https://fixture.invalid' }; + provision(dir, setup, createGenesis(publicKey(ownerSecret), X, 'both')); migrateSlots(dir); addSlot(dir, y, createGenesis(publicKey(ownerSecret), Y, 'both')); + 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(dir, url); const seen: Message[] = []; + const client = connect(url, ownerSecret, m => seen.push(m)); await client.ready; + const journal = (a: string) => JSON.parse(readFileSync(a === X ? join(dir, 'journal.json') : join(dir, 'agents', a, 'journal.json'), 'utf8')); + const startX = message('start', 'both', X, 0), startY = message('start', 'both', Y, 0); startY.id = startX.id; + try { + client.send(startX); client.send(startY); + await until(() => journal(X).phase === 'transitioning' && journal(Y).phase === 'transitioning'); + const stopX = message('stop', 'both', X, 0); client.send(stopX); + await until(() => seen.some(m => m.type === 'receipt' && m.body.operation === stopX.id)); + assert.equal(journal(X).phase, 'stopped'); assert.equal(journal(Y).phase, 'transitioning'); + await until(() => journal(Y).phase === 'running'); + assert.match(journal(X).operations[startX.id].reply.body.result, /cancelled/); + assert.equal(journal(Y).operations[startY.id].reply.body.result, 'accepted'); assert.equal(journal(Y).revision, 1); + const beforeX = readFileSync(join(dir, 'journal.json')), beforeY = readFileSync(join(dir, 'agents', Y, 'journal.json')); + seen.length = 0; client.send(startX); client.send(startY); + await until(() => seen.filter(m => m.type === 'receipt' && m.body.operation === startX.id).length >= 2); + assert.deepEqual(readFileSync(join(dir, 'journal.json')), beforeX); assert.deepEqual(readFileSync(join(dir, 'agents', Y, 'journal.json')), beforeY); + // Close tears down the remaining running sibling independently. + await h.close(); assert.equal(journal(Y).phase, 'stopped'); assert.equal(existsSync(join(dir, 'host.lock')), false); + } finally { + await h.close(); client.close(); for (const c of server.clients) c.terminate(); + await new Promise(resolve => server.close(() => resolve())); rmSync(dir, { recursive: true, force: true }); + } +}); + +test('explicit upgrade preserves real receipts; uncertain first slot cannot skip healthy sibling teardown', { timeout: 15000 }, async () => { + const dir = realpathSync(mkdtempSync(join(tmpdir(), 'beehive-slots-close-'))); + const ownerSecret = newKey(), x = newKey(), y = newKey(); const X = publicKey(x), Y = publicKey(y); + provision(dir, { host: 'close', ownerSecret, agentSecret: x, runner: process.execPath, args: [resolve('test/runner.ts')], workspace: dir, mode: 'fixture' }, createGenesis(publicKey(ownerSecret), X, 'close')); + 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}`; let h = await host(dir, url); const seen: Message[] = []; + const client = connect(url, ownerSecret, m => seen.push(m)); await client.ready; + async function request(m: Message) { client.send(m); await until(() => seen.some(r => r.type === 'receipt' && r.body.operation === m.id)); } + let expectedFailure = false; + try { + await request(message('start', 'close', X, 0)); await request(message('stop', 'close', X, 1)); await h.close(); + const journalPath = join(dir, 'journal.json'); const before = readFileSync(journalPath); + migrateSlots(dir); assert.deepEqual(readFileSync(journalPath), before); + const migrated = JSON.parse(before.toString()); assert.equal(migrated.revision, 2); assert.equal(Object.keys(migrated.operations).length, 2); assert.equal(migrated.outbox.length, 2); + assert.throws(() => migrateSlots(dir), /already enabled/); + addSlot(dir, y, createGenesis(publicKey(ownerSecret), Y, 'close')); + // Explicitly simulate recovered unknown execution, with no invented live PID. + writePrivate(journalPath, { ...migrated, phase: 'quarantined' }); + h = await host(dir, url); + await request(message('start', 'close', Y, 0)); + const yPath = join(dir, 'agents', Y, 'journal.json'); + assert.equal(JSON.parse(readFileSync(yPath, 'utf8')).phase, 'running'); + expectedFailure = true; await assert.rejects(h.close(), /Incomplete owned teardown/); + assert.equal(JSON.parse(readFileSync(journalPath, 'utf8')).phase, 'quarantined'); + assert.equal(JSON.parse(readFileSync(yPath, 'utf8')).phase, 'stopped'); + assert.equal(JSON.parse(readFileSync(yPath, 'utf8')).actual, null); + assert.equal(existsSync(join(dir, 'host.lock')), true, 'uncertain sibling retains installation fence'); + } finally { + if (!expectedFailure) await h.close(); + client.close(); for (const c of server.clients) c.terminate(); + await new Promise(resolve => server.close(() => resolve())); rmSync(dir, { recursive: true, force: true }); + } +}); + +test('management reconciliation queries every unresolved host/agent slot, then resolves both from real host', { timeout: 15000 }, async () => { + const dir = realpathSync(mkdtempSync(join(tmpdir(), 'beehive-slots-intents-'))); + const owner = newKey(), x = newKey(), y = newKey(); const X = publicKey(x), Y = publicKey(y); + const source = join(dir, 'host'); + provision(source, { host: 'shared', ownerSecret: owner, agentSecret: x, runner: process.execPath, args: [resolve('test/runner.ts')], workspace: dir, mode: 'fixture' }, createGenesis(publicKey(owner), X, 'shared')); + migrateSlots(source); addSlot(source, y, createGenesis(publicKey(owner), Y, 'shared')); + const server = await relay(0, publicKey(owner), join(dir, 'relay.json')); const address = server.address(); assert.ok(address && typeof address !== 'string'); + const url = `ws://127.0.0.1:${address.port}`; const seen: Message[] = []; + const observer = connect(url, owner, m => seen.push(m)); await observer.ready; + let client = managementClient(join(dir, 'intents'), url, owner, () => {}, () => {}); await client.ready; + let h: Awaited> | undefined; + try { + client.submit(message('start', 'shared', X, 0)); client.submit(message('start', 'shared', Y, 0)); + await until(() => seen.filter(m => m.type === 'start').length >= 2); + client.close(); seen.length = 0; + client = managementClient(join(dir, 'intents'), url, owner, () => {}, () => {}); await client.ready; + // Scope assertion is against the new connection's actual inspect publications, + // not historical relay replay: inspect IDs are fresh, unlike retained intents. + await delay(100); const prior = new Set(seen.map(m => m.id)); client.reconcile(); + await until(() => [X, Y].every(a => seen.some(m => m.type === 'inspect' && m.agent === a && !prior.has(m.id)))); + h = await host(source, url); + await until(() => client.status().every(i => i.state === 'completed')); + assert.equal(client.status().length, 2); + } finally { + await h?.close(); client.close(); observer.close(); for (const c of server.clients) c.terminate(); + await new Promise(resolve => server.close(() => resolve())); rmSync(dir, { recursive: true, force: true }); + } +});