feat(beehive): persist client intents and replay across TUI reconnect

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

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/<relay+owner digest>/`
+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,
+**Next:** independently review and exercise the local client-journal recovery
-then wire TUI reconnect/replay so a command saved locally before socket loss
+implemented above, including policy-failure UNKNOWN handling, then proceed to
-converges using its original operation ID even if it never reached the relay.
+two-host identity assignment/Move. The bounded journal is not full recovery/parity
-Use a dropped-before-publication and dropped-after-commit test; retain UNKNOWN
+completion. Independent multi-tool and reconnect review remains pending.
-until a matching host receipt. Host-side recovery is now implemented above.
-Independent multi-tool and reconnect delta review can proceed against this artifact.
 
 README retains the full parity matrix: reconnect/ACKs/crash recovery; multiple
 hosts/agents/setups and S1 two-host identity; reusable profiles/metadata/history;

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

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<string,Message>();
-    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;
+    try { await client.ready; } catch (error) { client.close(); throw error; }
-    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.'); });
     const ui = createInterface({ input: stdin, output: stdout });
-    console.log('Beehive | Hosts → assigned agent → selected-next / actual run\nCommands: hosts, select <host>, 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 <host>, 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);
+        try { client.submit(request); } catch (error) { console.log(error instanceof Error ? error.message : 'Operation not submitted'); continue; }
-        console.log(`Requested ${line} (${request.id}); publication is NOT host acceptance.`);
+        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; });

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<typeof setTimeout> | 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(); },
   };

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());
   }

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<string, Intent>();
+  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(); },
+  };
+}

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); }
 }

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<typeof target.send>[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<void>(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<void>(resolve => server.close(() => resolve()));
+    rmSync(dir, { recursive: true, force: true });
+  }
+});