mirror of
https://github.com/block/buzz.git
synced 2026-10-29 05:21:31 +00:00
feat(beehive): isolate agent slots under one host installation
Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
Logan Johnson
@@ -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
|
||||
|
||||
@@ -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/<public-key>/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 <row-number>` 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 |
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 <new-directory> Create a NEW local owner identity
|
||||
setup <new-host-directory> <identity-file> Guided local harness + key setup
|
||||
migrate-slots <host-directory> Explicit stopped upgrade, preserves journal
|
||||
add-agent <host-directory> New identity using shared local harness
|
||||
assignment-export <host-directory> <new-file> Export public pinned genesis locally
|
||||
migrate-assignment <host-directory> Explicit stopped legacy enrollment
|
||||
auth-info <host-directory> Print exact local sign-in context (no login)
|
||||
@@ -22,7 +25,8 @@ const help = `Beehive — isolated development preview (loopback relay only)
|
||||
relay <port> <owner-public-key> <log-file> Dedicated ciphertext relay
|
||||
host <host-directory> <ws://127.0.0.1:port> Persistent foreground host
|
||||
tui <identity-file> <ws://127.0.0.1:port> Relay-connected terminal UI
|
||||
setup accepts optional <local-key-file> <public-genesis-file> for standby import.
|
||||
setup/add-agent accept optional <local-key-file> <public-genesis-file> 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') {
|
||||
|
||||
@@ -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<string>();
|
||||
const retracted = new Map<string, number>();
|
||||
const retracting = (revision: number) => { for (const target of retracted.values()) if (target === revision) return true; return false; };
|
||||
let client: ReturnType<typeof connect>;
|
||||
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<typeof connect> | undefined;
|
||||
let initialized = false;
|
||||
const slots = new Map<string, ReturnType<typeof slot>>();
|
||||
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<void> | 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);
|
||||
})();
|
||||
} };
|
||||
}
|
||||
|
||||
@@ -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();
|
||||
|
||||
79
beehive/src/slots.ts
Normal file
79
beehive/src/slots.ts
Normal file
@@ -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<Setup, 'host' | 'ownerSecret' | 'agentSecret'>;
|
||||
type Installation = { version: 2; host: string; ownerSecret: string; setups: Record<string, Harness>; agents: Record<string, { secret: string; setup: string; legacy?: true }> };
|
||||
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);
|
||||
}
|
||||
209
beehive/test/slots.test.ts
Normal file
209
beehive/test/slots.test.ts
Normal file
@@ -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<void>(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<void>(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<void>(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<ReturnType<typeof host>> | 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<void>(resolve => server.close(() => resolve())); rmSync(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
Reference in New Issue
Block a user