From d12ed7b4e08501e2e5829da80689c9759b32bd49 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Wed, 9 Sep 2026 23:29:39 -0400 Subject: [PATCH] feat(beehive): persist client intents and replay across TUI reconnect Signed-off-by: Logan Johnson --- beehive/CHECKPOINT.md | 72 +++++++++++++++++-- beehive/TESTING.md | 11 +++ beehive/src/cli.ts | 23 +++--- beehive/src/client.ts | 14 ++-- beehive/src/host.ts | 6 +- beehive/src/intents.ts | 112 +++++++++++++++++++++++++++++ beehive/src/storage.ts | 8 ++- beehive/test/intents.test.ts | 132 +++++++++++++++++++++++++++++++++++ 8 files changed, 350 insertions(+), 28 deletions(-) create mode 100644 beehive/src/intents.ts create mode 100644 beehive/test/intents.test.ts diff --git a/beehive/CHECKPOINT.md b/beehive/CHECKPOINT.md index fd9a2faa0e..60d95844b5 100644 --- a/beehive/CHECKPOINT.md +++ b/beehive/CHECKPOINT.md @@ -5,6 +5,68 @@ Base `051c3a270be9c73da9ab06700bcab7d5552fceaa`; continuation starts at `023c9274767ef50fa0f5b37ef1883336f8be59fd`. Candidate is the commit containing this checkpoint (`git rev-parse HEAD`). No merge/release. +## Local management-client journal continuation 90950bdd + +Started verified published HEAD `6f893f51d20d93ff25770b66f3b52b50bcff151e`. +`src/intents.ts` is LOCAL TUI/client state, not a controller process/service. +No dependency, private-host-key, ordinary conversation permission or runtime change. +TUI automatically creates owner-only `management-intents//` +beside the existing operator identity. Each submitted operation is encrypted/signed +ONCE, fsynced with an exclusive immutable intent file before any send, and replayed +on bounded WS reconnect or ordinary reopen. A duplicate operation ID cannot prepare +a replacement. Receipts are encrypted in separate exclusive files; stale publication +observations cannot erase a terminal result. 1,000 intents maximum; no pruning yet. + +Host receipts now include the existing request fingerprint. Matching requires the +actual owner decryption/signature, exact relay journal scope, host, agent, operation +ID and fingerprint. Host outbox/operations remain lifecycle truth; no second admission +model. Relay echo is publication evidence, not host admission/completion. The UI +shows pending/completed/failed/unknown, with `operations` for retained details, and +prints host/action/result rather than requiring operation IDs. Quit cancels transport +reconnect only; it never cancels an operation or stops a host. Save remains selected-next. + +`test/intents.test.ts` proves actual socket loss before publication, reopen applying +the original Start once, retained terminal results, unrelated/duplicate receipts not +closing another intent, permanent host rejection, and relay-side socket termination +when delivering a Stop receipt AFTER the host journal has committed stopped. Automatic +client reconnect recovers that receipt. A separate actual Node subprocess exits 23 +without running the network event loop or graceful close after preparing an intent; +reopen publishes byte-identical envelope, verified against the relay's stored log. +Tampered signature fails closed. Relay policy close 1008 remains UNKNOWN and durably +blocks blind automatic replay across reopen; no fabricated host failure. Normal +network loss retries eight times; exhausted/offline pending work remains durable. +File writes propagate errors instead of being swallowed as malformed network input. + +Evidence: strict TypeScript + full installed-enabled package suite **16/16** passed. +After human-readable connection/status display refinement, strict + targeted real +TUI/client suites **5/5** passed; final source strict + full default suite +**15 pass / 1 installed opt-in skip** passed (the unchanged installed path was not +repeated). Prior installed runtime evidence retained; this run +also produced three verified fixture replies from agent +`b727bb8ae1e277a522f8dbad00aae15cc7b71e39c57d43e7f5d532529e8087aa`. +No real provider/admission calls, production activity, full repo CI or PR/merge/release. +Independent preceding broad-tool/reconnect review still in progress (evidence directory +exists; no completed report received). Existing B1/D1-D3 conclusions retained, not +reopened or treated as approval of this new journal delta. + +Limits: relay URL is the community scope in this development topology, not a stable +production community identifier. Publication echo observation is session-local; terminal +host results persist. There is no relay per-event rejection/ACK protocol: policy failure +is conservatively UNKNOWN/blocked, not claimed failed; explicit guided recovery and +journal retention remain future UX work. Pre-upgrade host receipts lack fingerprints +and cannot complete new journal intents. Arbitrary disk rollback, malicious local +operator, simultaneous UI action coordination, log-full recovery and host SIGKILL +recovery are not proved. Single-operation exclusive files prevent conflicting-ID +replacement and receipt overwrite, not distributed UI serialization. Crash test is +process exit without cleanup, not machine power-loss proof. Stop-during-Start remains +covered by unchanged full-suite regressions; no claim that this new test automates the +entire interactive crash workflow. + +**ONE next executable step:** independent review and real interactive TUI recovery +acceptance of this client journal, especially policy-failure UNKNOWN recovery and +receipt ordering, before expanding to the two-host same-identity assignment/Move slice. +Full remaining parity scope below remains active, not declared complete. + ## Recovery 4200a3ab Recovered clean HEAD `f8f161ffc10f274e13b9e1cd984ca8ffc2897cfd`, not @@ -181,12 +243,10 @@ cryptographic signing key configured, so no invented signer/signature claim. ## Remaining product scope / ONE next executable step -**Next:** add a durable controller pending-intent journal and receipt matching, -then wire TUI reconnect/replay so a command saved locally before socket loss -converges using its original operation ID even if it never reached the relay. -Use a dropped-before-publication and dropped-after-commit test; retain UNKNOWN -until a matching host receipt. Host-side recovery is now implemented above. -Independent multi-tool and reconnect delta review can proceed against this artifact. +**Next:** independently review and exercise the local client-journal recovery +implemented above, including policy-failure UNKNOWN handling, then proceed to +two-host identity assignment/Move. The bounded journal is not full recovery/parity +completion. Independent multi-tool and reconnect review remains pending. README retains the full parity matrix: reconnect/ACKs/crash recovery; multiple hosts/agents/setups and S1 two-host identity; reusable profiles/metadata/history; diff --git a/beehive/TESTING.md b/beehive/TESTING.md index e6e28488b2..b28ed8a534 100644 --- a/beehive/TESTING.md +++ b/beehive/TESTING.md @@ -17,6 +17,17 @@ A fresh-store/fresh-node_modules frozen install fetched all seven packages; strict TypeScript checking passed on Node 24.15.0 / pnpm 11.4.0. Keep `--ignore-workspace` so installation does not expand to unrelated client packages. +## Client journal continuation + +`intents.test.ts`: real WS loss before publication, original-envelope reopen, +post-host-commit receipt loss and automatic reconnect, retained terminal rejection, +unrelated/duplicate receipt fencing, abrupt child-process exit before network effects, +exact ciphertext/signature relay storage comparison, tampered file rejection, and +policy close 1008 blocking blind retries across reopen. Strict + installed-enabled +full suite passed 16/16; after status wording refinement strict + targeted TUI/client +5/5 passed, then final source strict/full default 15 pass + 1 opt-in skip. These fixtures do not establish production community admission, hostile +local storage/power loss or a polished manual recovery UX. See current CHECKPOINT. + ## Automated production seams - `slice.test.ts`: malformed/wrong-owner/tampered encryption envelopes; real diff --git a/beehive/src/cli.ts b/beehive/src/cli.ts index 68e1a048a4..ed6a60576a 100644 --- a/beehive/src/cli.ts +++ b/beehive/src/cli.ts @@ -1,10 +1,10 @@ import { createInterface } from 'node:readline/promises'; import { stdin, stdout } from 'node:process'; import { existsSync, mkdirSync, realpathSync, rmdirSync } from 'node:fs'; -import { resolve, join } from 'node:path'; +import { resolve, join, dirname } from 'node:path'; import { relay } from './relay.ts'; import { host, validateSetup } from './host.ts'; -import { connect } from './client.ts'; +import { managementClient } from './intents.ts'; import { message, newKey, publicKey, object, text, type Message } from './protocol.ts'; import { readPrivate, writePrivate } from './storage.ts'; import { prepareConversation } from './conversation.ts'; @@ -87,23 +87,24 @@ async function main() { } else if (command === 'tui') { const secret = text(object(readPrivate(resolve(text(args[0])))).secret); const inventory = new Map(); - const client = connect(text(args[1]),secret,m => { + const client = managementClient(join(dirname(resolve(text(args[0]))), 'management-intents'),text(args[1]),secret,m => { if (m.type === 'inventory') { const prior = inventory.get(m.host); if (!prior || Number(m.body.observedAt) > Number(prior.body.observedAt)) inventory.set(m.host,m); } - if (m.type === 'receipt') console.log(`\nHost receipt (may be replayed) ${m.body.operation}: ${JSON.stringify(m.body)}`); + }, () => { + console.log(client.connected ? '\nManagement relay connected.' : '\nRelay disconnected: pending results UNKNOWN; reconnect attempts are bounded.'); + for (const operation of client.status()) console.log(`${operation.request.host} ${operation.request.type}: ${operation.state} | ${operation.result ?? operation.publication}`); }); - await client.ready; - let intentionalClose = false; - client.socket.on('close',() => { if (!intentionalClose) console.log('\nRelay disconnected: live status and unresolved requests UNKNOWN. Received receipts remain valid. Reopen TUI.'); }); + 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: hosts, 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, hosts, select , show, save, start, stop, 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') { console.log(JSON.stringify(client.status(),null,2)); continue; } if (line === 'hosts') { for (const [name,m] of inventory) console.log(`${name}: ${m.body.phase} | ${m.body.readiness} | ${Date.now()-Number(m.body.observedAt) > 6000 ? 'STALE/UNKNOWN' : 'recent host report'}`); continue; } if (line.startsWith('select ')) { selected = line.slice(7); continue; } const current = inventory.get(selected); @@ -117,10 +118,10 @@ async function main() { 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',selected,current.agent,current.revision,body); - client.send(request); - console.log(`Requested ${line} (${request.id}); publication is NOT host acceptance.`); + 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.`); } - } finally { intentionalClose = true; ui.close(); client.close(); console.log('UI closed; host lifetime is independent.'); } + } finally { ui.close(); client.close(); console.log('UI closed; host lifetime is independent.'); } } else console.log(help); } main().catch(error => { console.error(error instanceof Error ? error.message : 'Beehive failed'); process.exitCode = 1; }); diff --git a/beehive/src/client.ts b/beehive/src/client.ts index 65ce8479e4..617deb9696 100644 --- a/beehive/src/client.ts +++ b/beehive/src/client.ts @@ -1,5 +1,5 @@ import { WebSocket } from 'ws'; -import { open, seal, type Message } from './protocol.ts'; +import { open, seal, type Message, type Envelope } from './protocol.ts'; /** Validate transport before acquiring installation resources. */ export function validateRelayURL(url: string) { const endpoint = new URL(url); @@ -10,7 +10,7 @@ export function validateRelayURL(url: string) { * Initial connection failure still rejects ready, allowing installation lock unwind. * The caller owns durable intent/receipts; an open socket is not an operation ACK. */ -export function connect(url: string, secret: string, receive: (m: Message) => void, recovered?: () => void) { +export function connect(url: string, secret: string, receive: (m: Message) => void, recovered?: () => void, disconnected?: (code: number) => void) { validateRelayURL(url); let closed = false; let established = false; let attempts = 0; let retry: ReturnType | undefined; @@ -25,12 +25,15 @@ export function connect(url: string, secret: string, receive: (m: Message) => vo }); current.on('message', data => { if (closed || current !== socket) return; - try { receive(open(JSON.parse(data.toString()),secret)); } catch { /* Reject malformed/untrusted data, never print payloads. */ } + let message: Message; + try { message = open(JSON.parse(data.toString()),secret); } catch { return; } + receive(message); // Application persistence failures must not masquerade as malformed wire data. }); current.on('error', error => { if (!established) rejectReady(error); }); - current.on('close', () => { + current.on('close', code => { if (!established) rejectReady(Error('Relay closed before ready')); - if (closed || current !== socket || !established || !recovered || attempts >= 8) return; + if (!closed && current === socket) disconnected?.(code); + if (code === 1008 || closed || current !== socket || !established || !recovered || attempts >= 8) return; retry = setTimeout(() => { retry = undefined; if (!closed) socket = dial(); }, Math.min(2000, 100 * 2 ** attempts++)); }); return current; @@ -39,6 +42,7 @@ export function connect(url: string, secret: string, receive: (m: Message) => vo return { get socket() { return socket; }, ready, + sendEnvelope(e: Envelope) { if (closed || socket.readyState !== WebSocket.OPEN) throw Error('Relay disconnected; result unknown'); socket.send(JSON.stringify(e)); }, send(m: Message) { if (closed || socket.readyState !== WebSocket.OPEN) throw Error('Relay disconnected; result unknown'); socket.send(JSON.stringify(seal(m,secret))); }, close() { closed = true; clearTimeout(retry); socket.close(); }, }; diff --git a/beehive/src/host.ts b/beehive/src/host.ts index bee34427cc..b3013bc42a 100644 --- a/beehive/src/host.ts +++ b/beehive/src/host.ts @@ -74,7 +74,7 @@ export async function host(directory: string, url: string) { const fingerprint = digest(JSON.stringify(m)).toString('hex'); const previous = Object.hasOwn(state.operations,m.id) ? state.operations[m.id] : undefined; if (previous) { - publish(previous.fingerprint === fingerprint ? previous.reply : message('receipt',setup.host,m.agent,state.revision,{ operation: m.id, result: 'operation-id-conflict' })); 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) result = 'not-authority'; @@ -82,7 +82,7 @@ export async function host(directory: string, url: string) { else if ((state.phase === 'quarantined' || state.phase === 'transitioning') && !(m.type === 'stop' && (owned instanceof ConversationSession ? owned.connected : owned?.child.connected))) result = 'quarantined'; else { // Reserve operation before any spawn/kill; interrupted admission never retries effects. - const pending = message('receipt',setup.host,agent,state.revision,{ operation: m.id, result: 'interrupted-reconcile-locally' }); + 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') { @@ -143,7 +143,7 @@ export async function host(directory: string, url: string) { result = error instanceof Error ? error.message : 'Operation failed'; } } - const reply = message('receipt',setup.host,m.agent,state.revision,{ operation: m.id, result }); + const reply = message('receipt',setup.host,m.agent,state.revision,{ operation: m.id, fingerprint, result }); Object.defineProperty(state.operations,m.id,{ value: { fingerprint, reply }, enumerable: true, configurable: true, writable: true }); state.outbox.push(reply); save(); publish(reply); publish(inventory()); } diff --git a/beehive/src/intents.ts b/beehive/src/intents.ts new file mode 100644 index 0000000000..0212614fbc --- /dev/null +++ b/beehive/src/intents.ts @@ -0,0 +1,112 @@ +import { existsSync, mkdirSync, readdirSync, statSync, lstatSync } from 'node:fs'; +import { join } from 'node:path'; +import { connect, validateRelayURL } from './client.ts'; +import { digest, fields, object, open, publicKey, seal, type Envelope, type Message } from './protocol.ts'; +import { readPrivate, writePrivate } from './storage.ts'; + +type Intent = { envelope: Envelope; request: Message; receipt?: Message; published: boolean; blocked?: boolean }; +const fingerprint = (m: Message) => digest(JSON.stringify(m)).toString('hex'); +function matches(request: Message, receipt: Message) { + return receipt.type === 'receipt' && receipt.host === request.host && receipt.agent === request.agent + && receipt.body.operation === request.id && receipt.body.fingerprint === fingerprint(request) + && typeof receipt.body.result === 'string' && receipt.body.result.length > 0; +} +/** Local operator journal, scoped to exact relay URL and signer; no agent keys. + * Immutable intent files precede all network effects. Receipt files are separate + * so another reader's old publication observation cannot erase a terminal result. + */ +export function managementClient(root: string, url: string, secret: string, receive: (m: Message) => void, changed: () => void = () => {}) { + validateRelayURL(url); + const scope = digest(JSON.stringify([url, publicKey(secret)])).toString('hex'); + const dir = join(root, scope); + mkdirSync(dir, { recursive: true, mode: 0o700 }); + for (const path of [root, dir]) { + const stat = lstatSync(path); + if (!stat.isDirectory() || (stat.mode & 0o077)) throw Error('Management journal directory must be owner-only'); + } + const intents = new Map(); + const names = readdirSync(dir); + if (names.length > 4000 || names.filter(n => n.endsWith('.intent')).length > 1000) throw Error('Management journal full'); + for (const name of names.filter(n => n.endsWith('.intent'))) { + const path = join(dir, name); + if (statSync(path).size > 70000) throw Error('Oversized intent'); + 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'); + const intent: Intent = { envelope: value.envelope as Envelope, request, published: false }; + const receiptPath = join(dir, `${request.id}.receipt`); + if (existsSync(receiptPath)) { + if (statSync(receiptPath).size > 70000) throw Error('Oversized receipt'); + const receipt = open(readPrivate(receiptPath), secret); + if (!matches(request, receipt)) throw Error('Unmatched journal receipt'); + intent.receipt = receipt; + } + const blockedPath = join(dir, `${request.id}.blocked`); + if (existsSync(blockedPath)) { + if (statSync(blockedPath).size > 70000) throw Error('Oversized blocked intent'); + const blocked = open(readPrivate(blockedPath), secret); + if (fingerprint(blocked) !== fingerprint(request)) throw Error('Invalid blocked intent'); + intent.blocked = true; + } + intents.set(request.id, intent); + } + let closed = false; + function replay() { + for (const intent of intents.values()) if (!intent.receipt && !intent.blocked) { + try { transport.sendEnvelope(intent.envelope); } catch { break; } // Intent remains durable. + } + changed(); + } + const transport = connect(url, secret, m => { + if (m.type === 'receipt') { + const intent = intents.get(String(m.body.operation)); + if (intent && !intent.receipt && matches(intent.request, m)) { + // Persist before exposing completion. Failure propagates, never reports success. + const path = join(dir, `${intent.request.id}.receipt`); + try { writePrivate(path, seal(m, secret), true); } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'EEXIST') throw error; + } + const persisted = open(readPrivate(path), secret); + if (!matches(intent.request, persisted)) throw Error('Unmatched durable receipt'); + intent.receipt = persisted; changed(); + } + } else { + const intent = intents.get(m.id); + if (intent && fingerprint(intent.request) === fingerprint(m)) { intent.published = true; changed(); } + } + receive(m); + }, replay, code => { + if (code === 1008) for (const intent of intents.values()) if (!intent.receipt) { + // Relay policy errors do not identify a committed host result. Preserve + // UNKNOWN, but never blindly republish rejected traffic across reopen. + writePrivate(join(dir, `${intent.request.id}.blocked`), intent.envelope); + intent.blocked = true; + } + changed(); + }); + const ready = transport.ready.then(replay); + return { + ready, + get connected() { return transport.socket.readyState === 1; }, + 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 (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); + const immutable = open(envelope, secret); + writePrivate(join(dir, `${request.id}.intent`), { scope, envelope }, true); + intents.set(request.id, { request: immutable, envelope, published: false }); + replay(); + }, + status() { + return [...intents.values()].map(i => ({ request: structuredClone(i.request), + state: i.receipt ? (['accepted','saved; running configuration unchanged'].includes(String(i.receipt.body.result)) ? 'completed' : i.receipt.body.result === 'interrupted-reconcile-locally' ? 'unknown' : 'failed') : i.blocked ? 'unknown' : 'pending', + publication: i.blocked ? 'relay policy failure; automatic retry disabled' : i.receipt ? 'host terminal receipt' : i.published ? 'relay observed; not host admission' : 'unconfirmed', + result: i.receipt?.body.result })); + }, + close() { closed = true; transport.close(); }, + }; +} diff --git a/beehive/src/storage.ts b/beehive/src/storage.ts index 476f90fb30..229465d557 100644 --- a/beehive/src/storage.ts +++ b/beehive/src/storage.ts @@ -1,13 +1,15 @@ -import { closeSync, fsyncSync, mkdirSync, openSync, readFileSync, renameSync, writeFileSync, lstatSync } from 'node:fs'; +import { closeSync, fsyncSync, mkdirSync, openSync, readFileSync, renameSync, writeFileSync, lstatSync, linkSync, unlinkSync } from 'node:fs'; import { dirname } from 'node:path'; import { randomUUID } from 'node:crypto'; /** Private atomic replacement, including file/directory fsync. Parent must be operator-owned. */ -export function writePrivate(path: string, value: unknown): void { +export function writePrivate(path: string, value: unknown, exclusive = false): void { mkdirSync(dirname(path), { recursive: true, mode: 0o700 }); const temp = `${path}.${randomUUID()}.tmp`; const fd = openSync(temp, 'wx', 0o600); try { writeFileSync(fd, JSON.stringify(value)); fsyncSync(fd); } finally { closeSync(fd); } - renameSync(temp,path); + if (exclusive) { + try { linkSync(temp,path); } finally { unlinkSync(temp); } + } else renameSync(temp,path); const dir = openSync(dirname(path),'r'); try { fsyncSync(dir); } finally { closeSync(dir); } } diff --git a/beehive/test/intents.test.ts b/beehive/test/intents.test.ts new file mode 100644 index 0000000000..f880b5250e --- /dev/null +++ b/beehive/test/intents.test.ts @@ -0,0 +1,132 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, realpathSync, readFileSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join, resolve } from 'node:path'; +import { setTimeout as delay } from 'node:timers/promises'; +import { host } from '../src/host.ts'; +import { managementClient } from '../src/intents.ts'; +import { connect } from '../src/client.ts'; +import { relay } from '../src/relay.ts'; +import { message, newKey, publicKey, open, type Message } from '../src/protocol.ts'; +import { writePrivate } from '../src/storage.ts'; + +async function until(predicate: () => boolean) { + for (let i = 0; i < 200; i++) { if (predicate()) return; await delay(25); } + throw Error('Missing reconnect convergence evidence'); +} + +test('durable client reopens exact unpublished intent, recovers lost post-commit receipt, and fences unrelated results', async () => { + const dir = realpathSync(mkdtempSync(join(tmpdir(), 'beehive-intents-'))); + const secret = newKey(); const agentSecret = newKey(); const agent = publicKey(agentSecret); + writePrivate(join(dir, 'setup.json'), { host: 'reconnect-host', ownerSecret: secret, agentSecret, + runner: process.execPath, args: [resolve('test/runner.ts')], workspace: dir, mode: 'fixture' }); + const server = await relay(0, publicKey(secret), join(dir, 'relay.json')); + const address = server.address(); assert.ok(address && typeof address !== 'string'); + const url = `ws://127.0.0.1:${address.port}`; + const h = await host(dir, url); + let ui = managementClient(join(dir, 'client'), url, secret, () => {}); await ui.ready; + const journal = () => JSON.parse(readFileSync(join(dir, 'journal.json'), 'utf8')); + try { + // Actual dropped WS before publication; synchronous submission persists offline. + ui.socket.terminate(); + const start = message('start', 'reconnect-host', agent, 0); + ui.submit(start); + assert.equal(ui.status()[0]?.state, 'pending'); + ui.close(); // Quit is not Stop and does not delete the unsent operation. + assert.equal(journal().revision, 0); + ui = managementClient(join(dir, 'client'), url, secret, () => {}); await ui.ready; + await until(() => ui.status()[0]?.state === 'completed'); + assert.equal(journal().revision, 1); + const actual = journal().actual; + const receipt = journal().operations[start.id].reply; + const history = () => JSON.parse(readFileSync(join(dir, 'relay.json'), 'utf8')); + const before = history().length; + ui.socket.terminate(); ui.close(); + ui = managementClient(join(dir, 'client'), url, secret, () => {}); await ui.ready; + await delay(150); + assert.equal(ui.status()[0]?.state, 'completed'); + assert.deepEqual(journal().actual, actual); + assert.deepEqual(journal().operations[start.id].reply, receipt); + assert.equal(journal().revision, 1); + // Permanent host rejection stays terminal, never regenerated on reconnect. + const rejected = message('start', 'reconnect-host', agent, 0); ui.submit(rejected); + await until(() => ui.status().find(s => s.request.id === rejected.id)?.state === 'failed'); + const unrelated = connect(url, secret, () => {}); await unrelated.ready; + unrelated.send(message('receipt', 'other-host', agent, 1, { ...receipt.body, operation: rejected.id })); + unrelated.send(receipt); unrelated.close(); + await delay(100); + assert.equal(ui.status().find(s => s.request.id === rejected.id)?.state, 'failed'); + ui.close(); ui = managementClient(join(dir, 'client'), url, secret, () => {}); await ui.ready; + assert.equal(ui.status().find(s => s.request.id === rejected.id)?.state, 'failed'); + const stop = message('stop', 'reconnect-host', agent, 1); + const target = [...server.clients].at(-1); + assert.ok(target); + const send = target.send.bind(target); + target.send = ((data: Parameters[0], ...args: unknown[]) => { + const m = open(JSON.parse(String(data)), secret); + if (m.type === 'receipt' && m.body.operation === stop.id) { + assert.equal(journal().phase, 'stopped', 'loss really occurs after host commit'); + target.terminate(); return; + } + Reflect.apply(send, target, [data, ...args]); + }) as typeof target.send; + ui.submit(stop); + await until(() => journal().phase === 'stopped'); + await until(() => ui.status().find(s => s.request.id === stop.id)?.state === 'completed'); + assert.equal(journal().phase, 'stopped'); assert.equal(journal().revision, 2); + assert.ok(history().length >= before); + } finally { + ui.close(); await h.close(); for (const socket of server.clients) socket.terminate(); + await new Promise(resolve => server.close(() => resolve())); + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('process exit before network effects leaves an exact prepared envelope; corrupt file fails closed', async () => { + const { spawnSync } = await import('node:child_process'); + const { readdirSync } = await import('node:fs'); + const dir = realpathSync(mkdtempSync(join(tmpdir(), 'beehive-crash-'))); + const secret = newKey(); + const server = await relay(0, publicKey(secret), join(dir, 'relay.json')); + const address = server.address(); assert.ok(address && typeof address !== 'string'); + const url = `ws://127.0.0.1:${address.port}`; + const root = join(dir, 'client'); const request = message('start', 'offline-host', publicKey(newKey())); + const child = spawnSync(process.execPath, ['--input-type=module', '-e', ` + import { managementClient } from ${JSON.stringify(new URL('../src/intents.ts', import.meta.url).href)}; + const client = managementClient(${JSON.stringify(root)}, ${JSON.stringify(url)}, ${JSON.stringify(secret)}, () => {}); + client.submit(${JSON.stringify(request)}); + process.exit(23); // No close, no event-loop turn, no websocket publication. + `], { encoding: 'utf8' }); + assert.equal(child.status, 23, child.stderr); + const scope = readdirSync(root)[0]!; + const intentPath = join(root, scope, `${request.id}.intent`); + const prepared = JSON.parse(readFileSync(intentPath, 'utf8')).envelope; + let ui = managementClient(root, url, secret, () => {}); + try { + await ui.ready; + await until(() => ui.status()[0]?.publication === 'relay observed; not host admission'); + const stored = JSON.parse(readFileSync(join(dir, 'relay.json'), 'utf8')); + assert.deepEqual(stored, [prepared], 'exact signed/encrypted event reused'); + const sender = connect(url, secret, () => {}); await sender.ready; + sender.send(message('receipt', 'unrelated-host', request.agent, 1, { operation: request.id, fingerprint: 'f'.repeat(64), result: 'accepted' })); + await delay(80); sender.close(); + assert.equal(ui.status()[0]?.state, 'pending'); + for (const socket of server.clients) socket.close(1008, 'Policy refusal'); + await until(() => ui.status()[0]?.state === 'unknown'); + ui.close(); + let republished = 0; + server.on('connection', socket => socket.on('message', () => { republished++; })); + ui = managementClient(root, url, secret, () => {}); await ui.ready; + await delay(150); + assert.equal(ui.status()[0]?.state, 'unknown'); + assert.equal(republished, 0, 'policy refusal is not blindly retried on reopen'); + ui.close(); + writePrivate(intentPath, { scope, envelope: { ...prepared, signature: '0'.repeat(128) } }); + assert.throws(() => managementClient(root, url, secret, () => {}), /signature/); + } finally { + ui.close(); for (const socket of server.clients) socket.terminate(); + await new Promise(resolve => server.close(() => resolve())); + rmSync(dir, { recursive: true, force: true }); + } +});