diff --git a/beehive/CHECKPOINT.md b/beehive/CHECKPOINT.md index 26faa6f1fd..d1ec2fce59 100644 --- a/beehive/CHECKPOINT.md +++ b/beehive/CHECKPOINT.md @@ -5,6 +5,92 @@ Base `051c3a270be9c73da9ab06700bcab7d5552fceaa`; continuation starts at `023c9274767ef50fa0f5b37ef1883336f8be59fd`. Candidate is the commit containing this checkpoint (`git rev-parse HEAD`). No merge/release. +## RECOVERED 9ff7 — executable fixture transaction, partial product handoff + +Recovered actual HEAD and actual remote both at +`8ba5fa9cd474aab5ce48fa255b194f8d3173aec1`, with no prior transaction commit/push. +Preserved dirty checkpoint/cli/host/intents/protocol and untracked handoff/move test. +A RECOVERED entry was written before implementation or long validation. The commit +containing this section is the new executable candidate (publication recorded in the +terminal result). Historical sections below are superseded by this current state. + +**Recovered:** source reservation, target preparation, source Stop, durable one-way +consumed authority + exact grant/outbox, validated predecessor chain, idempotent target +accept/fresh Start, TUI Move intent, three initial tests. **New:** validated Stop/Save +before cancelling Move; exact pending Move retry republishes prepare rather than a +terminal interrupted receipt; non-fixture/conversation Move explicitly refused before +source effects because metadata preparation is not exact-model ready. Shared status, +rows and retry confirmations now identify host + full agent + operation; Move includes +endpoints. Source Start never resumes after consumption, including service restart. +No source-validator relaxation without successor chain; no authority rollback. + +Observable tests: actual two-host subprocess + actual TUI Move X leaves source Y's +journal byte-identical; standby denied before grant; wrong target revision preserves +running source; reverse successor rejects historical grant. Fault-injected relay +loses grant and source receipt before receipt storage; inspect recovers outbox once. +Target key deletion and prepared setup change after grant both leave target assigned +stopped and source irrevocably denied, including after source service restart. Delayed +prepared replies cover exact pending retry, valid Stop/Save cancellation, malformed +Stop/unsupported Save inertness and late Start denial. Target admission failure here +is not a claim of a live-provider launch failure or moved conversation response. + +Consumed `ASSIGNMENT_SLOTS_REVIEW_8BA5FA9C.md`: F1 fixed, actual reopened two-agent +same-host/action/revision blocked-operation TUI regression checks each row and retry +confirmation. Declining either sends neither; immutable intent/receipt fences retained. +Consumed `RECOVERY_VERIFICATION_8A67B967.md` findings remain closed, not pending. + +Consumed new `DESCENDANT_EXIT_EBC8B3CB.md` report during this run. It reproduces the +invalid fixture PID failure signature and reports group/member teardown probes green; +historical raw PID/errno were not captured, so invalid PID versus PID reuse remains +unproved. **Containment gate remains OPEN, not cleared by these runs.** At the exact +reply-tool absence assertion, added raw PID, errno, spawn/error provenance and bounded +per-PID state diagnostics on recurrence, retaining ESRCH requirement and no extra +sleep. No owned.ts/supervisor teardown change or reconstructed PID kill authority. +Existing live anchors remain the only teardown authority; stronger full-lifetime +provenance and the historical observation still require reconciliation. Do not call +this preview safe Move or an exactly-once/partition-safe transfer. + +Validation evidence: `WORK_LOGS/BEEHIVE_DESIGN_183FFAF0/MOVE_9FF7_EVIDENCE/`. +Recovered strict + Move **3/3** passed; new targeted TUI/Move **5/5** passed, expanded +Move **4/4** passed. First candidate full installed-enabled **30/30**, zero skips, +19.90s (`check.log`, `full.log`). Late self-review then corrected pending Move retry +semantics; targeted **4/4** passed (`pending-retry.log`). Final-source strict/full +**30/30 pass, zero skips**, 19.12s (`check-final.log`, `full-final.log`). No failing executable run occurred in this +recovery; earlier worker test history is unknown. Initial discovery directory scan +timed out; no state mutation. Prior historical failing suite remains documented below. +Installed binary hashes are separate from source identity; both freshly match the +retained pins (`installed-binaries.sha256`). + +**ONE next executable action:** run the instrumented installed-enabled full suite at +this immutable candidate under the original contention configuration, retaining raw +failure capture if it recurs, to reconcile the still-open source-exit observation +before promoting fixture Move or implementing exact-ready conversation/provider Move. +Do not repeat resolved slots-label or R1/U1 reviews. Broader gate also needs independent +review of this new authority transaction; no installed moved-identity reply attempted. + +Scope: dedicated trusted non-cloned/non-rollback installations/exclusive supervision; +shared-owner signatures bind trusted protocol traffic, not physical host attestation. +Root/assignment persists stopped. TUI sends intents only; hosts share existing +management relay. No keys/workspaces/sessions/credentials transferred or regenerated, +no unreachable-source takeover, new controller/registry/HTTP route, provider/login, +production kind-40002/admission/OAuth proof, Rust/native or owner-store access. +README parity matrix retains profiles/key CRUD/Restart/harness/provider/preset/mesh/ +compute/real UI/live acceptance. No repo-wide CI, PR, merge or release claimed. + +## Move continuation 84e0 — implementation in progress + +Starting exact published `8ba5fa9cd474aab5ce48fa255b194f8d3173aec1`. +Implementing per-slot durable reservation → target prepared → source exit → +outgoing-consumed (grant and outbox in one journal write) → incoming-accepted → +fresh ordinary Start. Genesis stays pinned; only prefix-extending validated grant +lineage can replace initial-host authority. Preparation is not authority. Save/Stop +invalidate an outstanding source reservation; post-grant failures never undo it. +Next code path: host slot relay exchanges and immutable local preparation digest, +then actual two-host execution tests and TUI destination selection. +The descendant-exit report is not yet present at its supplied public path. Existing +25/25 is not resolution; final safe/conversation Move acceptance remains gated on +that investigation and causal regression. No containment changes in this stage. + ## Slot continuation e0a5 — working single-installation slice Started from verified published `ebc8b3cb24fe0eefe5b8f239a88555786eadbf14`. diff --git a/beehive/README.md b/beehive/README.md index 16941f7a87..a9f8dce426 100644 --- a/beehive/README.md +++ b/beehive/README.md @@ -96,7 +96,7 @@ arbitrary Nostr kind is not an admission solution. One dedicated owner shares signing/decryption authority with fully trusted host installations. Host IDs route messages; they are NOT cryptographic host proof. Agent identities are generated separately and only locally. Standby key import -attaches to an existing public genesis; Move remains disabled. Setup refuses an existing host directory; saved journal +attaches to an existing public genesis; only explicit experimental fixture Move can transfer assignment. Setup refuses an existing host directory; saved journal bindings prevent owner/agent/host replacement from resetting admission. An owner can still bypass software by cloning/recreating installations: safety assumes non-cloned, non-rollback, exclusively supervised installations. @@ -175,11 +175,10 @@ unreachable source never makes its standby eligible. The original assignment tes `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 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. +**Experimental fixture path implemented below:** destination local preflight, source +reservation/Stop, consumed grant/outbox lineage and target acceptance/fresh launch. +Safe-exit acceptance and provider exact-ready preflight remain gated. The root ID +is an identity binding, not a lock service or partition/rollback solution. ## Several agents, one installation @@ -221,8 +220,8 @@ failure and retains the installation lock rather than falsely declaring all stop 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 +revocation UX, Restart and provider/conversation Move are still future work. Standby +X elsewhere cannot Start without a grant, 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 @@ -337,7 +336,7 @@ remain explicit local/operator actions. No live Databricks request has occurred. | Local owner/key setup | New owner + separately generated agent key | nsec import, saved-owner verification, revocation/key removal UX | | 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 | +| Lifecycle | fixture Start/Stop and experimental source-consumed Move, UI-independent host | containment acceptance, exact-ready provider Move, Restart | | Buzz Agent Databricks v2 | Local setup + TypeScript ACP probe boundary | live OAuth/model proof, production protocol compatibility and catalog provenance | | Other Buzz Agent providers | Not implemented | Anthropic, OpenAI-compatible, Databricks legacy, OpenRouter | | Goose / Claude Code / Codex | Not implemented | harness-specific local auth and adapters/catalog/launch | @@ -348,3 +347,29 @@ remain explicit local/operator actions. No live Databricks request has occurred. None of the remaining rows is an owner-approved exclusion from the requested full parity. Review actual executable UX/security/lifecycle before expanding. + +## Experimental fixture Move (recovered continuation) + +On two separately provisioned, trusted installations with the SAME agent key and +pinned public genesis, use `hosts`, select the source agent row, then `move` and +confirm the destination. Hosts exchange preparation/grants over the existing relay; +the TUI sends intent only. Target selection uses its locally configured allowed +model/workspace/profile. No key, workspace, credential or conversation session moves. +The source receipt means **source consumed + grant durably queued**, NOT destination +process exit/start evidence. Observe target inventory separately. + +The source retains an irreversible grant chain/outbox after Stop. A destination +accepts an exact successor once before fresh Start; missing/changed prepared inputs +leave it assigned but stopped. Retry/inspect replays the grant without restoring the +source. A stopped or unreachable source does not authorize a standby by itself. +The experimental path is restricted in code to fixture harnesses: provider/catalog +metadata is not exact-model readiness, and conversation/provider Move is refused +before source effects. Ordinary Start retains same-session exact-model enforcement. + +**Containment acceptance remains OPEN.** The historical descendant-PID failure is +not classified between invalid fixture PID and PID reuse. Recovered probe evidence +supports the group teardown contract but does not close that historical gate; the +original absence requirement now records discriminating failure diagnostics. No +safe Move, exactly-once, partition-safe or moved-identity conversation claim. +Trust requires dedicated non-cloned/non-rollback installations, exclusive supervision +and local provisioning; shared owner signatures do not attest physical hosts. diff --git a/beehive/TESTING.md b/beehive/TESTING.md index cdbb463b79..a2da82992e 100644 --- a/beehive/TESTING.md +++ b/beehive/TESTING.md @@ -194,3 +194,25 @@ positions amid coalesced async status reports. A subsequent new reconciliation t 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. + +## Recovered fixture Move and recovery labels + +`move.test.ts` executes two real host subprocesses plus actual TUI and relay: Move X, +Y byte-identical, standby Start denied, wrong target revision preserves running source, +reverse successor and historical grant replay. Fault-injected real host/relay tests +lose grant and source receipt before receipt storage, replay outbox, remove target key +or change prepared setup, and verify assigned/stopped target and revoked source even +after source service restart. Delayed prepared replies exercise valid Stop/Save +cancellation, invalid Stop/Save inertness and late Start fencing. These are fixture +launch/admission tests, not provider readiness or destination conversation replies. + +`policy-tui.test.ts` reopens two same-host/action/revision blocked operations and checks +both agent identities and operation IDs in numbered rows AND each retry confirmation; +declining sends neither. Existing immutable-envelope policy-retry tests remain. + +Consumed `DESCENDANT_EXIT_EBC8B3CB.md`: exact historical cause remains unproved +(invalid fixture PID vs reuse); the group probes are evidence, not safe-Move approval. +`reply-tool.test.ts` now captures raw PID, actual probe errno, fixture spawn/error +provenance, and bounded per-PID process state on failure. It does not wait longer or +weaken ESRCH, and does not kill any PID obtained from observations. Full-suite green +results without recurrence do not classify the historical failure. diff --git a/beehive/src/cli.ts b/beehive/src/cli.ts index 93140f5144..21d38cb150 100644 --- a/beehive/src/cli.ts +++ b/beehive/src/cli.ts @@ -11,6 +11,7 @@ import { readPrivate, writePrivate } from './storage.ts'; import { createGenesis, validateGenesis } from './assignment.ts'; import { prepareConversation } from './conversation.ts'; +const operationLabel = (m: Message) => `${m.host} ${m.type}${m.type === 'move' ? ` → ${String(m.body.target)}` : ''} | agent ${m.agent} | operation ${m.id}`; const shellQuote = (s: string) => `'${s.replaceAll("'", "'\"'\"'")}'`; const [command, ...args] = process.argv.slice(2); const help = `Beehive — isolated development preview (loopback relay only) @@ -27,7 +28,7 @@ const help = `Beehive — isolated development preview (loopback relay only) tui Relay-connected terminal UI 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.`; +Move is fixture-only experimental; containment acceptance remains gated. No provider login RPC or production relay support.`; async function main() { if (command === 'identity') { const dir = resolve(text(args[0])); @@ -144,18 +145,18 @@ async function main() { } }, () => { console.log(client.connected ? '\nManagement relay connected.' : '\nRelay disconnected: pending results UNKNOWN; automatic reconnect is bounded (disabled on policy refusal). Use reconcile after checking relay policy.'); - for (const operation of client.status()) console.log(`${operation.request.host} ${operation.request.type}: ${operation.state} | ${operation.result ?? operation.publication}`); + for (const operation of client.status()) console.log(`${operationLabel(operation.request)}: ${operation.state} | ${operation.result ?? operation.publication}`); }); try { await client.ready; } catch (error) { client.close(); throw error; } const ui = createInterface({ input: stdin, output: stdout }); - console.log('Beehive | Hosts → assigned agent → selected-next / actual run\nCommands: operations, reconcile, retry , hosts, agents, select , show, save, start, stop, quit. Closing this UI does not stop hosts.'); + console.log('Beehive | Hosts → assigned agent → selected-next / actual run\nCommands: operations, reconcile, retry , hosts, agents, select , show, save, start, stop, move, quit. Closing this UI does not stop hosts.'); let selected = ''; try { for (;;) { const line = (await ui.question('beehive> ')).trim(); if (line === 'quit') break; if (line === 'operations') { - client.status().forEach((o, index) => console.log(`${index + 1}. ${o.request.host} ${o.request.type} at revision ${o.request.revision}: ${o.state} | ${o.result ?? o.publication}${o.retryAvailable ? ` | reconcile, then retry ${index + 1}` : ''}`)); + client.status().forEach((o, index) => console.log(`${index + 1}. ${operationLabel(o.request)} at revision ${o.request.revision}: ${o.state} | ${o.result ?? o.publication}${o.retryAvailable ? ` | reconcile, then retry ${index + 1}` : ''}`)); console.log('Historical results are not current host state. Reconcile queries receipts without retrying blocked work.'); continue; } if (line === 'reconcile') { client.reconcile(); console.log('Reconnecting/querying host results; blocked work stays blocked. Connection alone proves neither policy repair nor completion.'); continue; } @@ -163,7 +164,7 @@ async function main() { const number = line.slice(6); const operation = /^[1-9][0-9]*$/.test(number) ? client.status()[Number(number) - 1] : undefined; if (!operation?.retryAvailable) { console.log('Use operations to choose an unresolved policy-blocked operation.'); continue; } - console.log(`Retry original ${operation.request.host} ${operation.request.type} at revision ${operation.request.revision} ONCE, with unchanged signature, ID and preconditions. It may already have executed. Reconcile first. Repair relay policy before retry; expired/invalid signed events cannot be repaired here. This does not cancel remote work or enable automatic retries.`); + console.log(`Retry original ${operationLabel(operation.request)} at revision ${operation.request.revision} ONCE, with unchanged signature, ID and preconditions. It may already have executed. Reconcile first. Repair relay policy before retry; expired/invalid signed events cannot be repaired here. This does not cancel remote work or enable automatic retries.`); if ((await ui.question('Attempt unchanged operation after policy repair? [yes/no]: ')) === 'yes') { try { client.retry(operation.request.id); console.log('Unchanged retry attempted; result UNKNOWN until host receipt. Automatic retry stays disabled.'); } catch (error) { console.log(error instanceof Error ? error.message : 'Retry not sent'); } @@ -186,16 +187,24 @@ async function main() { const current = inventory.get(selected); if (!current) { console.log('Select an advertised host first.'); continue; } if (line === 'show') { console.log(JSON.stringify(current,null,2)); continue; } - if (!['save','start','stop'].includes(line)) { console.log('Use operations/reconcile/retry /hosts/select/show/save/start/stop/quit.'); continue; } + if (!['save','start','stop','move'].includes(line)) { console.log('Use operations/reconcile/retry /hosts/select/show/save/start/stop/move/quit.'); continue; } if (Date.now()-Number(current.body.observedAt) > 6000) { console.log('Host stale: status unknown; no action sent.'); continue; } let body: Record = {}; if (line === 'save') { console.log(`Allowed models: ${JSON.stringify(current.body.models)}; workspaces: ${JSON.stringify(current.body.workspaces)}; behavior profiles: ${JSON.stringify(current.body.profiles)}`); body = { model: await ui.question('Model: '), workspace: await ui.question('Workspace: '), profile: await ui.question('Behavior profile: ') }; } - const request = message(line as 'save' | 'start' | 'stop',current.host,current.agent,current.revision,body); + if (line === 'move') { + const destinations = [...inventory.values()].filter(m => m.agent === current.agent && m.host !== current.host); + destinations.forEach((m, i) => console.log(`${i + 1}. ${m.host} | ${JSON.stringify(m.body.selectedNext)} | locally provisioned key is NOT Start authority`)); + const target = destinations[Number(await ui.question('Destination number: ')) - 1]; + if (!target || Date.now()-Number(target.body.observedAt) > 6000) { console.log('Destination unknown/stale; repair local key/setup/auth and host connection. No Move sent.'); continue; } + if (await ui.question(`Move agent ${current.agent} from ${current.host} to ${target.host}, fresh execution, no workspace/session/credentials transferred; source cannot resume after grant. Confirm [yes/no]: `) !== 'yes') continue; + body = { target: target.host, targetRevision: target.revision, selection: target.body.selectedNext }; + } + const request = message(line as 'save' | 'start' | 'stop' | 'move',current.host,current.agent,current.revision,body); try { client.submit(request); } catch (error) { console.log(error instanceof Error ? error.message : 'Operation not submitted'); continue; } - console.log(`Durably pending ${line} (${request.id}); publication is NOT host acceptance.`); + console.log(`Durably pending ${line}: ${operationLabel(request)}; publication is NOT host acceptance.`); } } finally { ui.close(); client.close(); console.log('UI closed; host lifetime is independent.'); } } else console.log(help); diff --git a/beehive/src/handoff.ts b/beehive/src/handoff.ts new file mode 100644 index 0000000000..61e637d28a --- /dev/null +++ b/beehive/src/handoff.ts @@ -0,0 +1,31 @@ +import { digest, fields, object, text, type Message, parseMessage } from './protocol.ts'; +import { validateGenesis, type Genesis } from './assignment.ts'; + +/** Public exact launch selection; credentials and executable choices remain local. */ +export type Selection = { model: string; workspace: string; profile: string }; +/** A source-consumed successor. Outer management authentication assumes trusted hosts. */ +export type Grant = { root: string; predecessor: string; source: string; target: string; agent: string; operation: Message; prepared: string; selection: Selection; targetRevision: number; sourceRun: string | null }; +export type Assignment = { genesis: Genesis; assignedHost: string; chain?: Grant[] }; +export function hash(value: unknown): string { return digest(JSON.stringify(value)).toString('hex'); } +export function selection(value: unknown): Selection { + const s = object(value); fields(s, ['model','workspace','profile']); + return { model: text(s.model), workspace: text(s.workspace), profile: text(s.profile) }; +} +/** Validate every predecessor, routing and immutable operation binding, not timestamps. */ +export function validateAssignment(a: Assignment): void { + const root = validateGenesis(a.genesis); let holder = root.initialHost, predecessor = hash(root); + if (!Array.isArray(a.chain ?? []) || (a.chain?.length ?? 0) > 24) throw Error('Assignment chain limit'); + for (const g of a.chain ?? []) { + fields(object(g), ['root','predecessor','source','target','agent','operation','prepared','selection','targetRevision','sourceRun']); + const op = parseMessage(g.operation); fields(op.body, ['target','targetRevision','selection']); + selection(g.selection); text(g.prepared); + if (g.root !== hash(root) || g.predecessor !== predecessor || g.source !== holder || g.agent !== root.agent || g.target === holder || g.target !== op.body.target || op.host !== holder || op.agent !== root.agent || op.type !== 'move' || g.targetRevision !== op.body.targetRevision || hash(g.selection) !== hash(op.body.selection) || !Number.isSafeInteger(g.targetRevision) || g.targetRevision < 0 || (g.sourceRun !== null && typeof g.sourceRun !== 'string')) throw Error('Invalid consumed grant chain'); + text(g.target); holder = g.target; predecessor = hash(g); + } + if (a.assignedHost !== holder) throw Error('Saved assignment mismatch'); +} +/** A recipient may only extend its durable knowledge, never accept historical authority. */ +export function extendsAssignment(current: Assignment, next: Assignment): boolean { + validateAssignment(next); + return hash(current.genesis) === hash(next.genesis) && (next.chain?.length ?? 0) > (current.chain?.length ?? 0) && (current.chain ?? []).every((g, i) => hash(g) === hash(next.chain![i])); +} diff --git a/beehive/src/host.ts b/beehive/src/host.ts index 10c92db30b..3092d9c606 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 { hash, selection, validateAssignment, extendsAssignment, type Assignment, type Grant } from './handoff.ts'; import { installationSlots } from './slots.ts'; import { validateGenesis, type Genesis } from './assignment.ts'; import { connect, validateRelayURL } from './client.ts'; @@ -13,7 +14,7 @@ import { prepareConversation, conversationSummary, type ConversationSetup } from export type Setup = { host: string; ownerSecret: string; agentSecret: string; runner: string; args: string[]; workspace: string; mode: 'fixture' | 'buzz-agent-databricks-v2'; databricksHost?: string; serviceHome?: string; configDirectory?: string; conversation?: ConversationSetup }; type Selection = { model: string; workspace: string; profile: string }; -type State = { assignment: { genesis: Genesis; assignedHost: string }; binding: { host: string; owner: string; agent: string }; revision: number; phase: 'stopped' | 'transitioning' | 'running' | 'quarantined'; selected: Selection; actual: null | { selection: Selection; executableHash: string; run: string; evidence?: Evidence }; operations: Record; outbox: Message[] }; +type State = { assignment: Assignment; move?: { request: Message; prepare: Message }; preparations?: Record; incoming?: Record; binding: { host: string; owner: string; agent: string }; revision: number; phase: 'stopped' | 'transitioning' | 'running' | 'quarantined'; selected: Selection; actual: null | { selection: Selection; executableHash: string; run: string; evidence?: Evidence }; operations: Record; outbox: Message[] }; export function validateSetup(value: unknown): Setup { const s = object(value); for (const key of ['host','ownerSecret','agentSecret','runner','workspace']) text(s[key]); @@ -57,16 +58,20 @@ function slot(setup: Setup, path: string, currentSetup: () => Setup, publish: (m 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'); + if (genesis.owner !== state.binding.owner || genesis.agent !== agent) throw Error('Saved assignment mismatch'); + validateAssignment(state.assignment); if (state.phase !== 'stopped') state.phase = 'quarantined'; - const save = () => writePrivate(path,state); + let persistenceFailed = false; + const save = () => { try { writePrivate(path,state); } catch (e) { persistenceFailed = true; throw e; } }; + // Preparation is tied to this process lifetime. Never launch from a recovered token. + const livePreparations = new Set(); 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' }; function inventory() { return message('inventory',setup.host,agent,state.revision, { - assignedHost: state.assignment.assignedHost, genesis: state.assignment.genesis, executionAuthority: state.assignment.assignedHost === setup.host, phase: state.phase, selectedNext: state.selected, actualRun: state.actual, + move: state.move ? 'destination preflight pending; source still assigned' : undefined, assignmentChain: state.assignment.chain ?? [], assignedHost: state.assignment.assignedHost, genesis: state.assignment.genesis, executionAuthority: state.assignment.assignedHost === setup.host, phase: state.phase, selectedNext: state.selected, actualRun: state.actual, ...(setup.conversation ? { conversation: conversationSummary(setup.conversation) } : {}), catalog, setup: setup.mode, models: [setup.mode === 'fixture' ? 'fixture-model' : 'databricks-claude-haiku-4-5'], workspaces: [setup.workspace], profiles: ['default'], readiness: setup.conversation ? (state.actual?.evidence ? 'Conversation session model acknowledged and response completed; relay delivery unverified' : 'Waiting for an admitted conversation and local provider sign-in; no synthetic prompt') : setup.mode === 'fixture' ? 'fixture-only' : (state.phase === 'running' && state.actual?.evidence ? 'ACP model acknowledged + same-session response (not provider attestation or Buzz relay agent)' : 'unverified: Start runs an ACP greeting probe; sign in locally as host service user if required'), observedAt: Date.now(), @@ -88,7 +93,7 @@ function slot(setup: Setup, path: string, currentSetup: () => Setup, publish: (m 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); + const operation = m.host === setup.host && ['save','start','stop','move'].includes(m.type); const first = operation && !queuedOperations.has(m.id); if (operation) queuedOperations.add(m.id); // Reserve receive order too: a conflicting queued ID is not a Stop authority. @@ -98,29 +103,144 @@ function slot(setup: Setup, path: string, currentSetup: () => Setup, publish: (m // and the pre-commit recheck consume the retraction inside handle(). if (state.phase === 'transitioning' && state.actual) acp?.cancel(); } + if (first && state.move && m.type === 'save' && m.revision === state.revision && state.assignment.assignedHost === setup.host && !Object.hasOwn(state.operations, m.id)) { + try { const v = selection(m.body); if (v.model === state.selected.model && v.workspace === setup.workspace && v.profile === 'default') retracted.set(m.id, m.revision); } catch { /* Invalid Save cannot cancel. */ } + } queue = queue.then(() => handle(m)).catch(() => { state.phase = 'quarantined'; save(); }); } function replay() { for (const m of state.outbox) publish(m); + if (state.move) publish(state.move.prepare); publish(inventory()); } async function stopOwned() { if (!owned) throw Error('Unknown ownership; quarantined'); await owned.stop(); owned = undefined; acp = undefined; } + function prepareLocal(selected: Selection) { + selection(selected); + if (selected.model !== (setup.mode === 'fixture' ? 'fixture-model' : 'databricks-claude-haiku-4-5') || selected.workspace !== setup.workspace || selected.profile !== 'default') throw Error('Unsupported destination selection'); + try { + if (hash(currentSetup()) !== hash(setup)) throw Error('changed'); + } catch { throw Error('Local agent key/setup missing or changed; repair locally without resetting assignment'); } + accessSync(setup.runner, constants.X_OK); + if (realpathSync(setup.workspace) !== setup.workspace || !statSync(setup.workspace).isDirectory()) throw Error('Workspace changed'); + const launch: AgentLaunch | undefined = setup.mode === 'fixture' ? undefined : { executable: setup.runner, args: setup.args, workspace: selected.workspace, home: text(setup.serviceHome), configDirectory: text(setup.configDirectory), databricksHost: text(setup.databricksHost), model: selected.model }; + const prepared = launch ? prepareAgent(launch) : undefined; + const conversation = setup.conversation && launch ? prepareConversation(setup.conversation, launch, setup.agentSecret, state.binding.owner) : undefined; + // Local hashes only; no setup or secret values leave the host. Include existing + // script inputs and directory identity so replacement invalidates preparation. + const scripts = setup.args.filter(a => isAbsolute(a) && existsSync(a)).map(a => createHash('sha256').update(readFileSync(a)).digest('hex')); + const workspace = statSync(setup.workspace); + return { launch, token: hash({ setup, selected, prepared, conversation, scripts, executable: createHash('sha256').update(readFileSync(setup.runner)).digest('hex'), workspace: [workspace.dev, workspace.ino] }) }; + } + function finishMove(result: string) { + const pending = state.move; + if (!pending) return; + const request = pending.request, fingerprint = hash(request); + const reply = message('receipt', setup.host, agent, state.revision, { operation: request.id, fingerprint, result, assignedHost: state.assignment.assignedHost, transfer: result === 'accepted' ? 'source consumed; grant queued; destination launch observed separately' : 'no grant' }); + state.operations[request.id] = { fingerprint, reply }; state.outbox.push(reply); + delete state.move; save(); publish(reply); publish(inventory()); + } + async function exchange(m: Message) { + try { + if (m.type === 'prepare') { + fields(m.body, ['assignment','operation']); + if (setup.mode !== 'fixture' || setup.conversation) throw Error('Move exact-model preflight is not implemented for conversation/provider harnesses'); + const assignment = m.body.assignment as Assignment; validateAssignment(assignment); + const op = m.body.operation as Message; + // Validate the proposed grant shape through the same chain validator. + const g: Grant = { root: hash(assignment.genesis), predecessor: hash(assignment.chain?.at(-1) ?? assignment.genesis), source: op.host, target: setup.host, agent, operation: op, prepared: 'pending', selection: selection(op.body.selection), targetRevision: m.revision, sourceRun: null }; + validateAssignment({ ...assignment, assignedHost: setup.host, chain: [...(assignment.chain ?? []), g] }); + if (hash(assignment.genesis) !== hash(genesis) || !((state.assignment.chain ?? []).every((g, i) => hash(g) === hash(assignment.chain?.[i]))) || state.phase !== 'stopped' || state.assignment.assignedHost === setup.host || m.revision !== state.revision) throw Error('Destination authority/revision conflict'); + const prior = state.preparations?.[op.id]; + if (prior) { + if (hash(prior.request) !== hash(m)) throw Error('Preparation ID conflict'); + if (livePreparations.has(op.id)) { publish(prior.reply); return; } + } + if (!prior && Object.keys(state.preparations ?? {}).length >= 1000) throw Error('Preparation journal full; local reconciliation needed'); + const token = prepareLocal(g.selection).token; + const reply = message('prepared', op.host, agent, op.revision, { prepare: m, token }); + state.preparations ??= {}; state.preparations[op.id] = { request: m, token, reply }; livePreparations.add(op.id); save(); publish(reply); + } else if (m.type === 'prepared') { + const pending = state.move; + if (!pending || hash(m.body.prepare) !== hash(pending.prepare) || m.revision !== pending.request.revision || state.revision !== m.revision || state.assignment.assignedHost !== setup.host) return; + if (m.body.error) { finishMove(String(m.body.error)); return; } + fields(m.body, ['prepare','token']); const token = text(m.body.token); + if (state.phase !== 'stopped') { + if (state.phase !== 'running' || !owned) throw Error('Source ownership unknown; no grant'); + state.phase = 'transitioning'; save(); await stopOwned(); + } + // Stop/Save admitted during the await retract exactly this source revision. + state.phase = 'stopped'; const sourceRun = state.actual?.run ?? null; state.actual = null; + if (retracting(m.revision) || closing) { finishMove('Move cancelled after source Stop; no grant'); return; } + const op = pending.request; + const grant: Grant = { root: hash(genesis), predecessor: hash(state.assignment.chain?.at(-1) ?? genesis), source: setup.host, target: text(op.body.target), agent, operation: op, prepared: token, selection: selection(op.body.selection), targetRevision: Number(op.body.targetRevision), sourceRun }; + const assignment: Assignment = { genesis, assignedHost: grant.target, chain: [...(state.assignment.chain ?? []), grant] }; validateAssignment(assignment); + const delivery = message('grant', grant.target, agent, grant.targetRevision, { assignment }); + // Irreversible authority consumption + exact grant outbox precede publication. + state.assignment = assignment; state.revision++; state.outbox.push(delivery); + finishMove('accepted'); publish(delivery); + } else { + fields(m.body, ['assignment']); const next = m.body.assignment as Assignment; + validateAssignment(next); const g = next.chain?.at(-1); + if (!g || g.target !== setup.host || g.agent !== agent || m.revision !== g.targetRevision) return; + const prior = state.incoming?.[hash(g)]; + if (prior) { publish(prior); return; } + if (!extendsAssignment(state.assignment, next) || state.assignment.assignedHost === setup.host || state.phase !== 'stopped') return; + const prep = state.preparations?.[g.operation.id]; + if (!prep || hash(prep.request.body.operation) !== hash(g.operation) || prep.token !== g.prepared || state.revision !== g.targetRevision) return; + if (hash(prep.request.body.assignment) !== hash({ genesis: next.genesis, assignedHost: g.source, ...(next.chain!.length > 1 ? { chain: next.chain!.slice(0,-1) } : {}) })) { + // Genesis-only journals may explicitly retain an empty chain. + const before = prep.request.body.assignment as Assignment; + if (hash(before.genesis) !== hash(next.genesis) || before.assignedHost !== g.source || hash(before.chain ?? []) !== hash(next.chain!.slice(0,-1))) return; + } + // Accept before validating launch again: a stale/missing prerequisite can + // fail launch but must never return authority to the consumed source. + state.assignment = next; state.selected = g.selection; + const receipt = message('receipt', setup.host, agent, state.revision, { operation: m.id, fingerprint: hash(m), result: 'destination assigned; launch pending' }); + state.incoming ??= {}; state.incoming[hash(g)] = receipt; state.outbox.push(receipt); save(); publish(receipt); + let outcome: unknown; + try { + if (!livePreparations.has(g.operation.id) || prepareLocal(g.selection).token !== g.prepared) throw Error('Destination preparation invalidated; assigned here, stopped; repair local setup/key/auth then explicit Start'); + const start = message('start', setup.host, agent, state.revision); start.id = g.operation.id; + await handle(start); + outcome = state.operations[start.id]?.reply.body.result; + } catch (e) { outcome = e instanceof Error ? e.message : 'Destination launch failed'; } + livePreparations.delete(g.operation.id); + const completed = message('receipt', setup.host, agent, state.revision, { operation: m.id, fingerprint: hash(m), result: outcome, assignedHost: setup.host, transfer: 'source consumed; destination assigned' }); + state.incoming[hash(g)] = completed; state.outbox.push(completed); save(); publish(completed); publish(inventory()); + } + } catch (e) { + if (persistenceFailed) throw e; + if (m.type === 'prepare') { + const op = m.body.operation as Message; + if (op?.host) publish(message('prepared', op.host, agent, op.revision, { prepare: m, error: 'Destination preflight failed: locally provision matching key, allowed workspace and harness/auth setup; no source Stop' })); + } else if (m.type === 'prepared') { + if (state.phase === 'transitioning') state.phase = 'quarantined'; + finishMove(e instanceof Error ? e.message : 'Move failed; reconcile ownership'); + } // Invalid grants are inert; they cannot change local authority. + } + } async function handle(m: Message) { - if (m.host === setup.host && ['save','start','stop'].includes(m.type)) queuedOperations.delete(m.id); - if (closing || m.host !== setup.host || !['inspect','save','start','stop'].includes(m.type)) return; + if (m.host === setup.host && ['save','start','stop','move'].includes(m.type)) queuedOperations.delete(m.id); + if (closing || persistenceFailed || m.host !== setup.host || !['inspect','save','start','stop','move','prepare','prepared','grant'].includes(m.type)) return; + if (['prepare','prepared','grant'].includes(m.type)) { await exchange(m); return; } // Serialized Stop processing durably resolves that Stop's own retraction. - if (m.type === 'stop') retracted.delete(m.id); - if (m.type === 'inspect') { for (const receipt of state.outbox) publish(receipt); publish(inventory()); return; } + if (m.type === 'stop' || m.type === 'save') retracted.delete(m.id); + if (m.type === 'inspect') { if (state.move) publish(state.move.prepare); for (const receipt of state.outbox) publish(receipt); publish(inventory()); return; } const fingerprint = digest(JSON.stringify(m)).toString('hex'); const previous = Object.hasOwn(state.operations,m.id) ? state.operations[m.id] : undefined; if (previous) { + // Move awaits a relay reply outside this queue: reservation is not terminal. + if (state.move?.request.id === m.id && previous.fingerprint === fingerprint) { + publish(state.move.prepare); publish(inventory()); return; + } publish(previous.fingerprint === fingerprint ? previous.reply : message('receipt',setup.host,m.agent,state.revision,{ operation: m.id, fingerprint, result: 'operation-id-conflict' })); return; } let result = 'accepted'; if (m.agent !== agent || state.assignment.assignedHost !== setup.host) result = 'not-authority'; + else if (state.move && ['start','move'].includes(m.type)) result = 'Move reservation busy'; else if (m.revision !== state.revision) result = 'revision-conflict'; else if ((state.phase === 'quarantined' || state.phase === 'transitioning') && !(m.type === 'stop' && (owned instanceof ConversationSession ? owned.connected : owned?.child.connected))) result = 'quarantined'; else { @@ -128,35 +248,32 @@ function slot(setup: Setup, path: string, currentSetup: () => Setup, publish: (m const pending = message('receipt',setup.host,agent,state.revision,{ operation: m.id, fingerprint, result: 'interrupted-reconcile-locally' }); state.operations[m.id] = { fingerprint, reply: pending }; save(); try { - if (m.type === 'save') { + if (state.move && m.type === 'stop') { fields(m.body, []); finishMove('Move cancelled before grant by stop'); } + if (m.type === 'move') { + fields(m.body, ['target','targetRevision','selection']); + if (setup.mode !== 'fixture' || setup.conversation) throw Error('Move is fixture-only until exact-model preflight and containment acceptance are complete'); + const target = text(m.body.target); selection(m.body.selection); + if (target === setup.host || !Number.isSafeInteger(m.body.targetRevision) || Number(m.body.targetRevision) < 0 || (state.assignment.chain?.length ?? 0) >= 24) throw Error('Invalid destination or chain limit'); + const prepare = message('prepare', target, agent, Number(m.body.targetRevision), { assignment: state.assignment, operation: m }); + state.move = { request: m, prepare }; save(); publish(prepare); publish(inventory()); return; + } else if (m.type === 'save') { fields(m.body,['model','workspace','profile']); const selected = { model: text(m.body.model), workspace: text(m.body.workspace), profile: text(m.body.profile) }; if (selected.model !== (setup.mode === 'fixture' ? 'fixture-model' : 'databricks-claude-haiku-4-5') || selected.workspace !== setup.workspace || selected.profile !== 'default') throw Error('Unsupported selection'); + if (state.move) finishMove('Move cancelled before grant by save'); state.selected = selected; result = 'saved; running configuration unchanged'; } else if (m.type === 'start') { fields(m.body,[]); if (state.phase !== 'stopped') throw Error('Already running'); // A retracted admission never begins effects, including any spawn. 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 = 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); - if (realpathSync(setup.workspace) !== setup.workspace || !statSync(setup.workspace).isDirectory()) throw Error('Workspace changed'); - const launch: AgentLaunch | undefined = setup.mode === 'fixture' ? undefined : { executable: setup.runner, args: setup.args, workspace: state.selected.workspace, home: text(setup.serviceHome), configDirectory: text(setup.configDirectory), databricksHost: text(setup.databricksHost), model: state.selected.model }; - if (launch) prepareAgent(launch); // Prerequisite failure preserves stopped state. - if (setup.conversation) { - if (!launch) throw Error('Conversation requires a Buzz Agent harness setup'); - prepareConversation(setup.conversation, launch, setup.agentSecret, state.binding.owner); - } + const { launch } = prepareLocal(state.selected); state.actual = { selection: structuredClone(state.selected), executableHash: createHash('sha256').update(readFileSync(setup.runner)).digest('hex'), run: m.id }; state.phase = 'transitioning'; save(); if (setup.mode === 'fixture') { owned = spawnOwned(setup.runner, setup.args, state.actual.selection.workspace, { PATH: '/usr/bin:/bin', BEEHIVE_FIXTURE: '1' }); owned.child.stdout.resume(); owned.child.stderr.resume(); - await owned.ready; + try { await owned.ready; } catch (e) { await stopOwned(); state.phase = 'stopped'; state.actual = null; throw e; } } else if (setup.conversation) { const session = new ConversationSession(setup.conversation, launch!, setup.agentSecret, state.binding.owner); acp = session; owned = session; @@ -197,6 +314,7 @@ function slot(setup: Setup, path: string, currentSetup: () => Setup, publish: (m } state.revision++; } catch (error) { + if (persistenceFailed) throw error; if (state.phase === 'transitioning') state.phase = 'quarantined'; result = error instanceof Error ? error.message : 'Operation failed'; } @@ -248,7 +366,7 @@ export async function host(directory: string, url: string) { 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' })); + else if (['save','start','stop','move'].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; } diff --git a/beehive/src/intents.ts b/beehive/src/intents.ts index 870a87b19e..a8da9fc943 100644 --- a/beehive/src/intents.ts +++ b/beehive/src/intents.ts @@ -33,7 +33,7 @@ export function managementClient(root: string, url: string, secret: string, rece const value = object(readPrivate(path)); fields(value, ['scope', 'envelope']); if (value.scope !== scope) throw Error('Wrong journal scope'); const request = open(value.envelope, secret); - if (!['save','start','stop'].includes(request.type) || name !== `${request.id}.intent`) throw Error('Invalid intent'); + if (!['save','start','stop','move'].includes(request.type) || name !== `${request.id}.intent`) throw Error('Invalid intent'); const intent: Intent = { envelope: value.envelope as Envelope, request, published: false }; const receiptPath = join(dir, `${request.id}.receipt`); if (existsSync(receiptPath)) { @@ -97,7 +97,7 @@ export function managementClient(root: string, url: string, secret: string, rece get socket() { return transport.socket; }, submit(request: Message) { if (closed) throw Error('UI closed'); - if (!['save','start','stop'].includes(request.type)) throw Error('Invalid operation'); + if (!['save','start','stop','move'].includes(request.type)) throw Error('Invalid operation'); if (intents.size >= 1000) throw Error('Management journal full; no operation submitted'); if (intents.has(request.id) || existsSync(join(dir, `${request.id}.intent`))) throw Error('Operation ID already prepared'); const envelope = seal(request, secret); diff --git a/beehive/src/protocol.ts b/beehive/src/protocol.ts index 6c81df68d4..7a26cfc872 100644 --- a/beehive/src/protocol.ts +++ b/beehive/src/protocol.ts @@ -2,7 +2,7 @@ import { createHash, createCipheriv, createDecipheriv, randomBytes, randomUUID } import { schnorr } from '@noble/curves/secp256k1'; /** Experimental private wire protocol; deliberately not a Nostr kind allocation. */ -export type Message = { v: 1; id: string; host: string; agent: string; type: 'inspect' | 'save' | 'start' | 'stop' | 'inventory' | 'receipt'; revision: number; body: Record }; +export type Message = { v: 1; id: string; host: string; agent: string; type: 'inspect' | 'save' | 'start' | 'stop' | 'move' | 'prepare' | 'prepared' | 'grant' | 'inventory' | 'receipt'; revision: number; body: Record }; export type Envelope = { v: 1; owner: string; nonce: string; ciphertext: string; tag: string; signature: string }; /** Reject unknown envelope/message fields, oversized values and ambiguous revisions. */ export function object(value: unknown): Record { @@ -19,7 +19,7 @@ export function fields(value: Record, names: string[]) { export function parseMessage(value: unknown): Message { const m = object(value); fields(m, ['v','id','host','agent','type','revision','body']); - if (m.v !== 1 || !['inspect','save','start','stop','inventory','receipt'].includes(String(m.type)) || !Number.isSafeInteger(m.revision) || Number(m.revision) < 0) throw Error('Invalid message'); + if (m.v !== 1 || !['inspect','save','start','stop','move','prepare','prepared','grant','inventory','receipt'].includes(String(m.type)) || !Number.isSafeInteger(m.revision) || Number(m.revision) < 0) throw Error('Invalid message'); if (!/^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/.test(text(m.id))) throw Error('Invalid operation ID'); text(m.host); text(m.agent); object(m.body); return m as Message; diff --git a/beehive/test/move.test.ts b/beehive/test/move.test.ts new file mode 100644 index 0000000000..94c24823a4 --- /dev/null +++ b/beehive/test/move.test.ts @@ -0,0 +1,195 @@ +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, unlinkSync } 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, open, type Message } from '../src/protocol.ts'; +import { writePrivate } from '../src/storage.ts'; +import { relay } from '../src/relay.ts'; +import { connect } from '../src/client.ts'; + +test('two real hosts: source-consumed Move, sibling isolation, preflight refusal, historical replay', { timeout: 30000 }, async () => { + const dir = realpathSync(mkdtempSync(join(tmpdir(), 'beehive-move-'))); + const ownerSecret = newKey(), x = newKey(), y = newKey(), agent = publicKey(x); + const genesis = createGenesis(publicKey(ownerSecret), agent, 'source'); + const setup = (host: string): Setup => ({ host, ownerSecret, agentSecret: x, runner: process.execPath, args: [resolve('test/runner.ts')], workspace: dir, mode: 'fixture' }); + provision(join(dir, 'source'), setup('source'), genesis); provision(join(dir, 'target'), setup('target'), genesis); + migrateSlots(join(dir, 'source')); addSlot(join(dir, 'source'), y, createGenesis(publicKey(ownerSecret), publicKey(y), 'source')); + 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}`, seen: Message[] = [], children: ChildProcess[] = []; + const ui = connect(url, ownerSecret, m => seen.push(m)); await ui.ready; + async function wait(predicate: (m: Message) => boolean) { + for (let i = 0; i < 500; i++) { const m = seen.find(predicate); if (m) return m; await delay(10); } + throw Error('Missing Move evidence'); + } + const journal = (name: string) => JSON.parse(readFileSync(join(dir, name, 'journal.json'), 'utf8')); + async function request(type: Message['type'], host: string, revision: number, body = {}, key = agent) { + const m = message(type, host, key, revision, body); ui.send(m); + return wait(r => r.type === 'receipt' && r.body.operation === m.id && r.host === host); + } + try { + for (const name of ['source','target']) { + const child = spawn(process.execPath, ['src/cli.ts','host',join(dir,name),url], { stdio: ['ignore','pipe','pipe'] }); + children.push(child); child.stdout?.resume(); child.stderr?.resume(); await wait(m => m.type === 'inventory' && m.host === name); + } + assert.equal((await request('start','source',0)).body.result,'accepted'); + assert.equal((await request('start','source',0,{},publicKey(y))).body.result,'accepted'); + const siblingPath = join(dir,'source','agents',publicKey(y),'journal.json'); + const sibling = readFileSync(siblingPath,'utf8'); + assert.equal((await request('start','target',0)).body.result,'not-authority'); + const selected = { model: 'fixture-model', workspace: dir, profile: 'default' }; + const failed = await request('move','source',1,{ target: 'target', targetRevision: 9, selection: selected }); + assert.match(String(failed.body.result),/preflight failed/); + assert.equal(journal('source').phase,'running'); assert.equal(journal('source').revision,1); + writePrivate(join(dir,'identity.json'),{ secret: ownerSecret }); + const terminal = spawn(process.execPath,['src/cli.ts','tui',join(dir,'identity.json'),url],{ stdio:['pipe','pipe','pipe'] }); children.push(terminal); + let output = '', cursor = 0, index = 0; + const steps = [ + { prompt:'beehive> ', value:() => 'hosts' }, + { prompt:'beehive> ', value:() => { const row = output.match(new RegExp(`(\\d+)\\. source \\| agent ${agent}`)); assert.ok(row); return `select ${row[1]}`; } }, + { prompt:'beehive> ', value:() => 'move' }, + { prompt:'Destination number: ', value:() => '1' }, + { prompt:'Confirm [yes/no]: ', value:() => 'yes' }, + { prompt:'beehive> ', value:() => 'quit' }, + ]; + terminal.stdout.on('data',chunk => { + output += chunk.toString(); const step = steps[index]; + if (!step) return; + const position = output.indexOf(step.prompt, cursor); if (position < 0) return; + index++; cursor = position + step.prompt.length; + setImmediate(() => terminal.stdin.write(`${step.value()}\n`)); + }); terminal.stderr.resume(); + const [code] = await once(terminal,'exit'); assert.equal(code,0); assert.match(output,/Durably pending move/); + const move = await wait(m => m.type === 'move' && m.host === 'source' && m.body.targetRevision === 0); + + const result = await wait(m => m.type === 'receipt' && m.body.operation === move.id && m.host === 'source'); assert.equal(result.body.result,'accepted'); + await wait(m => m.type === 'inventory' && m.host === 'target' && m.body.phase === 'running'); + assert.equal(journal('source').assignment.assignedHost,'target'); assert.equal(journal('source').phase,'stopped'); assert.equal(journal('source').actual,null); + assert.equal(journal('target').assignment.assignedHost,'target'); assert.equal(journal('target').actual.run,move.id); + assert.equal(readFileSync(siblingPath,'utf8'),sibling); + const grant = await wait(m => m.type === 'grant'); const before = readFileSync(join(dir,'target','journal.json'),'utf8'); + ui.send(move); ui.send(grant); await delay(100); assert.equal(readFileSync(join(dir,'target','journal.json'),'utf8'),before); + assert.equal((await request('start','source',2)).body.result,'not-authority'); + // Move back is a NEW successor; historical forward grant cannot seize it. + assert.equal((await request('move','target',1,{ target: 'source', targetRevision: 2, selection: selected })).body.result,'accepted'); + await wait(m => m.type === 'inventory' && m.host === 'source' && m.revision === 3 && m.body.phase === 'running'); + ui.send(grant); await delay(100); + assert.equal(journal('target').assignment.assignedHost,'source'); assert.equal(journal('target').phase,'stopped'); + assert.equal(journal('source').assignment.chain.length,2); + assert.equal((await request('stop','source',3)).body.result,'accepted'); + assert.equal((await request('stop','source',1,{},publicKey(y))).body.result,'accepted'); + } finally { + for (const child of children) { if (child.exitCode === null && child.signalCode === null) { const exit = once(child,'exit'); child.kill('SIGTERM'); await exit; } } + ui.close(); for (const c of server.clients) c.terminate(); await new Promise(resolve => server.close(() => resolve())); rmSync(dir,{ recursive:true,force:true }); + } +}); + + +for (const failure of ['missing-key', 'changed-prepared-input'] as const) test(`dropped grant/receipt recovery; post-grant ${failure} leaves target assigned stopped`, { timeout: 20000 }, async () => { + const dir = realpathSync(mkdtempSync(join(tmpdir(), 'beehive-move-loss-'))); + const ownerSecret = newKey(), agentSecret = newKey(), agent = publicKey(agentSecret); + const genesis = createGenesis(publicKey(ownerSecret), agent, 'source'); + for (const name of ['source','target']) provision(join(dir,name), { host: name, ownerSecret, agentSecret, runner: process.execPath, args: [resolve('test/runner.ts')], workspace: dir, mode: 'fixture' },genesis); + 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 source = await host(join(dir,'source'),url); const target = await host(join(dir,'target'),url); + const sourceSocket = [...server.clients][0]!; + let lostOperation = '', loseReceipt = true; + const handlers = sourceSocket.listeners('message'); sourceSocket.removeAllListeners('message'); + sourceSocket.on('message', (raw, binary) => { + const m = open(JSON.parse(String(raw)),ownerSecret); + if (loseReceipt && m.type === 'receipt' && m.body.operation === lostOperation) return; + for (const handler of handlers) Reflect.apply(handler, sourceSocket, [raw,binary]); + }); + const targetSocket = [...server.clients][1]!; const send = targetSocket.send.bind(targetSocket); let drop = true; + targetSocket.send = ((data: Parameters[0], ...args: unknown[]) => { + const m = open(JSON.parse(String(data)),ownerSecret); + if (drop && m.type === 'grant') return; + return (send as Function)(data,...args); + }) as typeof targetSocket.send; + const seen: Message[] = []; const ui = connect(url,ownerSecret,m => seen.push(m)); await ui.ready; + const journal = (name: string) => JSON.parse(readFileSync(join(dir,name,'journal.json'),'utf8')); + async function until(predicate: () => boolean) { for (let i=0;i<500;i++) { if (predicate()) return; await delay(10); } throw Error('Missing failure/recovery evidence'); } + try { + const start = message('start','source',agent,0); ui.send(start); await until(() => journal('source').phase === 'running'); + const move = message('move','source',agent,1,{ target:'target',targetRevision:0,selection:{ model:'fixture-model',workspace:dir,profile:'default' } }); lostOperation = move.id; ui.send(move); + await until(() => journal('source').assignment.assignedHost === 'target'); + assert.equal(journal('source').phase,'stopped'); assert.equal(journal('target').assignment.assignedHost,'source'); + assert.equal(seen.some(m => m.type === 'receipt' && m.body.operation === move.id),false, 'source receipt lost before relay storage'); + const targetSetup = join(dir,'target','setup.json'); + if (failure === 'missing-key') unlinkSync(targetSetup); + else writePrivate(targetSetup, { ...JSON.parse(readFileSync(targetSetup,'utf8')), args: [resolve('test/runner.ts'), 'changed-after-preparation'] }); + drop = false; loseReceipt = false; + ui.send(message('inspect','source',agent)); + await until(() => seen.some(m => m.type === 'receipt' && m.body.operation === move.id && m.body.result === 'accepted')); + await until(() => journal('target').assignment.assignedHost === 'target'); + await until(() => seen.some(m => m.host === 'target' && m.type === 'receipt' && String(m.body.result).includes('Local agent key/setup missing'))); + assert.equal(journal('target').phase,'stopped'); assert.equal(journal('target').actual,null); + const denied = message('start','source',agent,2); ui.send(denied); + await until(() => seen.some(m => m.body.operation === denied.id && m.body.result === 'not-authority')); + const state = journal('target'); + for (const grant of seen.filter(m => m.type === 'grant')) ui.send(grant); + await delay(100); assert.deepEqual(journal('target'),state); + await source.close(); source = await host(join(dir,'source'),url); + const restarted = message('start','source',agent,2); ui.send(restarted); + await until(() => seen.some(m => m.body.operation === restarted.id && m.body.result === 'not-authority')); + assert.equal(journal('source').assignment.assignedHost,'target', 'service restart cannot resurrect consumed source'); + } finally { + await source.close(); await target.close(); ui.close(); for (const c of server.clients) c.terminate(); + await new Promise(resolve => server.close(() => resolve())); rmSync(dir,{ recursive:true,force:true }); + } +}); + +test('Stop and Save retract exact source reservation while destination preflight reply is delayed', { timeout: 20000 }, async () => { + const dir = realpathSync(mkdtempSync(join(tmpdir(),'beehive-move-cancel-'))); + const ownerSecret = newKey(), agentSecret = newKey(), agent = publicKey(agentSecret); + const genesis = createGenesis(publicKey(ownerSecret),agent,'source'); + for (const name of ['source','target']) provision(join(dir,name), { host:name, ownerSecret,agentSecret,runner:process.execPath,args:[resolve('test/runner.ts')],workspace:dir,mode:'fixture' },genesis); + 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 source = await host(join(dir,'source'),url), target = await host(join(dir,'target'),url); + const socket = [...server.clients][0]!, send = socket.send.bind(socket); const held: Parameters[0][] = []; + socket.send = ((data: Parameters[0], ...args: unknown[]) => { + if (open(JSON.parse(String(data)),ownerSecret).type === 'prepared') { held.push(data); return; } + return (send as Function)(data,...args); + }) as typeof socket.send; + const seen: Message[] = []; const ui = connect(url,ownerSecret,m => seen.push(m)); await ui.ready; + const journal = () => JSON.parse(readFileSync(join(dir,'source','journal.json'),'utf8')); + async function until(predicate: () => boolean) { for(let i=0;i<500;i++) { if(predicate()) return; await delay(10); } throw Error('Missing cancellation evidence'); } + const selected = { model:'fixture-model',workspace:dir,profile:'default' }; + try { + ui.send(message('start','source',agent)); await until(() => journal().phase === 'running'); + const actual = journal().actual; + for (const [revision,action] of [[1,'save'],[2,'stop']] as const) { + const move = message('move','source',agent,revision,{ target:'target',targetRevision:0,selection:selected }); ui.send(move); + await until(() => held.length > 0); + ui.send(move); await until(() => held.length > 1); + assert.equal(seen.some(m => m.type === 'receipt' && m.body.operation === move.id),false, 'exact pending retry must not publish a terminal interrupted receipt'); + const invalid = message('stop','source',agent,revision+20); ui.send(invalid); + await until(() => seen.some(m => m.body.operation === invalid.id && m.body.result === 'revision-conflict')); + assert.equal(journal().move.request.id,move.id); + for (const invalid of [message('stop','source',agent,revision,{ unexpected:true }), message('save','source',agent,revision,{ ...selected, model:'unsupported' }), message('start','source',agent,revision)]) { + ui.send(invalid); await until(() => seen.some(m => m.body.operation === invalid.id && m.type === 'receipt')); + assert.equal(journal().move.request.id,move.id, 'invalid Stop/Save and late Start must not displace reservation'); + assert.deepEqual(journal().actual,actual); + } + const cancel = message(action,'source',agent,revision,action === 'save' ? selected : {}); ui.send(cancel); + await until(() => seen.some(m => m.body.operation === move.id && String(m.body.result).includes('cancelled before grant'))); + for (const data of held.splice(0)) send(data); await delay(100); + assert.equal(journal().assignment.assignedHost,'source'); assert.equal(journal().assignment.chain,undefined); + if(action === 'save') assert.deepEqual(journal().actual,actual); + else await until(() => journal().phase === 'stopped'); + } + assert.equal(seen.some(m => m.type === 'grant'),false); + } finally { + await source.close(); await target.close(); ui.close(); for(const c of server.clients) c.terminate(); + await new Promise(resolve => server.close(() => resolve())); rmSync(dir,{ recursive:true,force:true }); + } +}); diff --git a/beehive/test/policy-tui.test.ts b/beehive/test/policy-tui.test.ts index eb9feafbcd..d6cf75b327 100644 --- a/beehive/test/policy-tui.test.ts +++ b/beehive/test/policy-tui.test.ts @@ -9,6 +9,7 @@ import { setTimeout as delay } from 'node:timers/promises'; import { WebSocket, WebSocketServer } from 'ws'; import { host } from '../src/host.ts'; import { relay } from '../src/relay.ts'; +import { managementClient } from '../src/intents.ts'; import { connect } from '../src/client.ts'; import { message, newKey, publicKey, open } from '../src/protocol.ts'; import { writePrivate } from '../src/storage.ts'; @@ -118,12 +119,12 @@ test('actual TUI: policy repair, bounded unchanged retry, lost committed receipt assert.equal(attempts.length, beforeReopen, 'reopen never republishes blocked intent'); await ui.command('operations'); if (loss === 'before-publication') { - assert.match(ui.output, /1\. policy-host stop at revision 1: unknown/); + assert.match(ui.output, new RegExp(`1\\. policy-host stop \\| agent ${agent} \\| operation ${operation.id} at revision 1: unknown`)); await ui.command('reconcile'); await delay(150); assert.equal(attempts.length, beforeReopen, 'query is not a retry'); await ui.command('retry 1', '[yes/no]: '); await ui.command('yes'); } - await until(() => ui.output.includes('policy-host stop: completed | accepted')); + await until(() => ui.output.includes(`policy-host stop | agent ${agent} | operation ${operation.id}: completed | accepted`)); assert.equal(journal().revision, 2); assert.equal(journal().phase, 'stopped'); assert.equal(Object.keys(journal().operations).length, 2, 'one Start and original Stop only'); assert.equal(journal().operations[operation.id].reply.body.result, 'accepted'); @@ -146,3 +147,50 @@ test('actual TUI: policy repair, bounded unchanged retry, lost committed receipt } } }); + +test('actual reopened TUI identifies both same-host/action/revision blocked agents in rows and retry confirmations', async () => { + const dir = realpathSync(mkdtempSync(join(tmpdir(), 'beehive-two-agent-recovery-'))); + const secret = newKey(), agents = [publicKey(newKey()), publicKey(newKey())]; + writePrivate(join(dir, 'identity.json'), { secret }); + const server = new WebSocketServer({ host: '127.0.0.1', port: 0 }); + await new Promise(resolve => server.once('listening', resolve)); + const address = server.address(); assert.ok(address && typeof address !== 'string'); + const url = `ws://127.0.0.1:${address.port}`; + let attempts = 0; + server.on('connection', socket => socket.on('message', raw => { + if (open(JSON.parse(String(raw)), secret).type === 'stop') { attempts++; socket.close(1008, 'Fixture policy'); } + })); + const client = managementClient(join(dir, 'management-intents'), url, secret, () => {}); + let child: ReturnType | undefined; + try { + await client.ready; + const requests = agents.map(agent => message('stop', 'shared', agent, 0)); + for (const request of requests) client.submit(request); + await until(() => client.status().every(o => o.retryAvailable)); client.close(); + const before = attempts; + child = spawn(process.execPath, ['src/cli.ts', 'tui', join(dir, 'identity.json'), url], { stdio: ['pipe','pipe','pipe'] }); + let output = ''; child.stdout!.on('data', d => { output += String(d); }); child.stderr!.on('data', d => { output += String(d); }); + await until(() => output.includes('beehive> ')); + async function command(line: string, prompt = 'beehive> ') { + const offset = output.length; child!.stdin!.write(`${line}\n`); + await until(() => output.slice(offset).includes(prompt)); return output.slice(offset); + } + const rows = await command('operations'); + for (const request of requests) { + const label = `shared stop | agent ${request.agent} | operation ${request.id}`; + const row = rows.split('\n').find(line => line.includes(label)); assert.ok(row, rows); + assert.ok(row.includes('at revision 0: unknown')); + const number = row.match(/^(\d+)\./)?.[1]; assert.ok(number); + const warning = await command(`retry ${number}`, '[yes/no]: '); + assert.ok(warning.includes(`Retry original ${label} at revision 0 ONCE`), warning); + await command('no'); + } + child.stdin!.write('quit\n'); await until(() => child!.exitCode !== null); + assert.equal(child.exitCode, 0, output); assert.equal(attempts, before, 'declined retry never publishes either agent'); + } finally { + client.close(); + if (child && child.exitCode === null) { child.kill(); await new Promise(resolve => child!.once('exit', () => resolve())); } + for (const socket of server.clients) socket.terminate(); + await new Promise(resolve => server.close(() => resolve())); rmSync(dir, { recursive: true, force: true }); + } +}); diff --git a/beehive/test/reply-tool.test.ts b/beehive/test/reply-tool.test.ts index 2e622acc43..9356694e83 100644 --- a/beehive/test/reply-tool.test.ts +++ b/beehive/test/reply-tool.test.ts @@ -1,6 +1,6 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; -import { spawn } from 'node:child_process'; +import { spawn, spawnSync } from 'node:child_process'; import { createInterface } from 'node:readline'; import { mkdtempSync, realpathSync, writeFileSync, chmodSync, readFileSync, existsSync, rmSync } from 'node:fs'; import { join } from 'node:path'; @@ -22,6 +22,8 @@ test('fixed Buzz CLI MCP rejects caller authority and owns CLI descendants on su fs.writeFileSync('identity', process.env.BUZZ_PRIVATE_KEY); const child = spawn(process.execPath, ['-e', "process.on('SIGTERM',()=>{});setTimeout(()=>{},15000)"], {stdio:'ignore'}); fs.writeFileSync('descendant', String(child.pid)); + child.once('error', error => fs.writeFileSync('descendant-spawn-error', String(error.code))); + child.once('spawn', () => fs.writeFileSync('descendant-spawned', JSON.stringify({ pid: child.pid, parent: process.pid }))); process.stdin.resume(); process.stdin.on('end', () => { ${mode === 'success' ? 'setTimeout(()=>process.exit(0),100)' : mode === 'failure' ? 'process.exit(2)' : "process.on('SIGTERM',()=>{});setTimeout(()=>process.exit(0),15000)"} }); `); @@ -65,7 +67,20 @@ test('fixed Buzz CLI MCP rejects caller authority and owns CLI descendants on su } await tool.stop(); assert.ok(absent(shim.pid!)); - if (existsSync(join(dir, 'descendant'))) assert.ok(absent(Number(readFileSync(join(dir, 'descendant'), 'utf8')))); + if (existsSync(join(dir, 'descendant'))) { + const raw = readFileSync(join(dir, 'descendant'), 'utf8'), pid = Number(raw); + // Preserve the original assertion. A recurrence must distinguish an invalid + // fixture PID from a visible PID; numeric identity alone is not ownership. + let errno = 'visible'; + try { process.kill(pid, 0); } catch (error) { errno = String((error as NodeJS.ErrnoException).code); } + if (errno !== 'ESRCH') { + const read = (name: string) => existsSync(join(dir, name)) ? readFileSync(join(dir, name), 'utf8') : null; + const processState = Number.isSafeInteger(pid) && pid > 0 + ? spawnSync('/bin/ps', ['-o','state=,pgid=,ppid=,comm=','-p',String(pid)], { encoding:'utf8' }).stdout : null; + console.error('Descendant exit observation', JSON.stringify({ mode, raw, pid, errno, spawned: read('descendant-spawned'), spawnError: read('descendant-spawn-error'), processState })); + } + assert.equal(errno, 'ESRCH', 'owned descendant must be absent after Stop (see failure capture)'); + } } finally { await tool.stop(); rmSync(dir, { recursive: true, force: true }); } } });