jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-11-02 07:22:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(beehive): fence queued Start cancellation by admitted Stop identity

Browse Source 
Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
Logan Johnson
2026-09-09 23:49:20 -04:00
parent d12ed7b4e0
commit 64305a7a00
4 changed files with 214 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
beehive/CHECKPOINT.md
View File

@@ -5,6 +5,35 @@ Base `051c3a270be9c73da9ab06700bcab7d5552fceaa`; continuation starts at
`023c9274767ef50fa0f5b37ef1883336f8be59fd`. Candidate is the commit containing
this checkpoint (`git rev-parse HEAD`). No merge/release.
## Continuation 4341 — R1 reconciliation
Recovered d12ed7b4e08501e2e5829da80689c9759b32bd49 plus the prior worker's
uncommitted host/TESTING/admission-cancel changes. Fresh substantive review found
an additional pending-ID gap: a same-batch Stop reusing the queued Start's ID
could retract admission although its own eventual receipt was operation-id-conflict.
Receive-order reservations now fence that early authority, alongside durable IDs,
agent/host/revision/body checks. Added conflicting-ID and invalid-body batch controls.
Executing ACP cancellation may surface `ACP session unavailable`; after verified
teardown the receipt now consistently names concurrent Stop cancellation. No sleeps,
replay exception, stale-revision bypass or changed command preconditions.
FIFO Stop handling clears its retraction before future Starts; terminal historical
IDs cannot retract again. Close drops not-yet-handled messages (UNKNOWN, replayable)
and awaits executing admission/owned teardown; no new close semantics claimed.
Fresh strict TypeScript and full installed-enabled package suite **18/18** passed
on this candidate (Node24.15.0). Exact log:
`WORK_LOGS/BEEHIVE_DESIGN_183FFAF0/CONTINUATION_4341_EVIDENCE/r1-full.log`.
One initial targeted run exposed the cancellation wording variation above, corrected
before this full run. Source R1 checkpoint is the commit containing this section;
publication/head will be recorded in the next UX checkpoint.
Completed independent `INTENT_TUI_REVIEW_D12ED7B4.md` is consumed: strict16/16,
36-file byte match, seven receipt negatives, real PTY lost-Start recovery and safe
Quit; no additional blocking journal correctness defect in trusted non-cloned /
non-rollback scope. U1 is a concrete policy-correction recovery UX gate, NOT pending
review: implement supported reconcile and explicit unchanged retry next. Earlier
pending-review/future-recovery notes below are historical, superseded here.
## Local management-client journal continuation 90950bdd
Started verified published HEAD `6f893f51d20d93ff25770b66f3b52b50bcff151e`.

22
beehive/TESTING.md
View File

@@ -52,6 +52,17 @@ local storage/power loss or a polished manual recovery UX. See current CHECKPOIN
rather than wait for the response; interrupted Start cannot commit running.
Wrong-agent/revision/body Stop cannot interrupt. D3 malformed completion tail
fails at the host seam as well as AgentSession.
- `admission-cancel.test.ts`: a Stop published beside a still-pending Start (live
same-batch and through disconnected relay-history replay, in ACP and fixture
host modes) cancels that admission before any run commits: the Start receipt is
non-accepted/cancelled, the Stop is accepted at the same revision, the journal
ends stopped with no actual run and no live-run inventory, and the verified stop
releases the ownership lock. Controls keep the fence exact: a wrong-authority
Stop in the same batch cannot cancel, a Stop published before its Start cannot
cancel that later Start, replaying the already-terminal batch after another
disconnect neither reruns nor recancels, and a fresh stale-revision Stop stays
a conflict. This is the R1 replay-batch cancellation fix from the independent
`TOOLS_RECONNECT_REVIEW_6F893F51.md` review.
- D1 `slice.test.ts` variants: resistant orphan descendants observe TERM but stay
alive, then are removed by bounded live-anchor KILL escalation through both
remote Stop and host close. `disconnect.test.ts` verifies parent IPC loss during
@@ -87,7 +98,7 @@ local storage/power loss or a polished manual recovery UX. See current CHECKPOIN
CLI error (returned without killing the session) all clean up the shim and any TERM-resistant in-group descendants.
Tests never open existing owner keys/profiles/cache, production services or native
client resources. With installed opt-in: 14/14 pass; without: 13 pass/1 explicit
client resources. With installed opt-in: 18/18 pass; without: 17 pass/1 explicit
skip. The installed fixture relay does not establish actual community admission,
current relay compatibility, provider/model attestation. Multi-thread delivery is isolated legacy-binary evidence only.
@@ -99,8 +110,13 @@ Prior c4f30176 broker independently passed strict/11 tests and installed replay
`WORK_LOGS/BEEHIVE_DESIGN_183FFAF0/BROKER_REVIEW_C4F30176.md`. Its B1 optional-config
finding is fixed with production regressions here. Independent `TOOL_REVIEW_023C9274.md` found no new blocking defect in the
predecessor MCP delta and independently confirmed B1 resolved (strict/full12/12,
original/extended B1 probes and signed installed replay). The new ordinary-CLI
multi-conversation delta has not yet received independent review. The generic ACP fixture and signed/encrypted WS
original/extended B1 probes and signed installed replay). Independent
`TOOLS_RECONNECT_REVIEW_6F893F51.md` then covered the ordinary-CLI and reconnect
deltas (strict/full 14/14 installed plus its own probes; no new tool-authority
expansion) and is now consumed: its single R1 medium finding, a replayed
same-batch Start+Stop leaving the agent running, is fixed here with
`admission-cancel.test.ts` production regressions; its receipt-loss, backoff,
duplicate/fingerprint and CLI-option probes remain reusable evidence. The generic ACP fixture and signed/encrypted WS
management protocol are different seams; passing them does not prove current
Buzz relay auth/member/owner semantics.

35
beehive/src/host.ts
View File

@@ -48,12 +48,32 @@ export async function host(directory: string, url: string) {
let closing = false;
let queue = Promise.resolve();
let initialized = false;
// Exact-fenced Stop retractions received ahead of their serialized handling, keyed
// by that Stop's operation ID. One invariant covers every not-yet-committed Start
// admission at the retracted revision, independent of callback/microtask timing:
// a still-queued admission observes the retraction before any effect, and an
// executing admission is interrupted below and re-checked before the running
// commit. FIFO queue order confines each retraction to Starts received before its
// Stop; that Stop's own serialized processing resolves it, so unseen future
// Starts and already-committed runs are never cancelled by it.
const queuedOperations = new Set<string>();
const retracted = new Map<string, number>();
const retracting = (revision: number) => { for (const target of retracted.values()) if (target === revision) return true; return false; };
let client: ReturnType<typeof connect>;
try { client = connect(url,setup.ownerSecret,m => {
if (!initialized) return;
// Cancellation is signalled outside the serialized mutation queue. Admission is
// still exact-authority/revision and never bypasses durable receipt processing.
if (!closing && m.type === 'stop' && m.host === setup.host && m.agent === agent && m.revision === state.revision && state.phase === 'transitioning' && state.actual && !Object.hasOwn(state.operations, m.id) && Object.keys(m.body).length === 0) acp?.cancel();
const operation = m.host === setup.host && ['save','start','stop'].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.
if (first && !closing && m.type === 'stop' && m.host === setup.host && m.agent === agent && m.revision === state.revision && !Object.hasOwn(state.operations, m.id) && Object.keys(m.body).length === 0) {
retracted.set(m.id,m.revision);
// Only an executing admission has a session to interrupt; queued admissions
// and the pre-commit recheck consume the retraction inside handle().
if (state.phase === 'transitioning' && state.actual) acp?.cancel();
}
queue = queue.then(() => handle(m)).catch(() => { state.phase = 'quarantined'; save(); });
}, () => {
if (closing || !initialized) return;
@@ -69,7 +89,10 @@ export async function host(directory: string, url: string) {
await owned.stop(); owned = undefined; acp = undefined;
}
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;
// Serialized Stop processing durably resolves that Stop's own retraction.
if (m.type === 'stop') retracted.delete(m.id);
if (m.type === 'inspect') { publish(inventory()); return; }
const fingerprint = digest(JSON.stringify(m)).toString('hex');
const previous = Object.hasOwn(state.operations,m.id) ? state.operations[m.id] : undefined;
@@ -93,6 +116,8 @@ export async function host(directory: string, url: string) {
} 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');
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 };
@@ -115,6 +140,7 @@ export async function host(directory: string, url: string) {
if (!session.healthy) throw Error('Conversation failed before running commit');
} catch (e) {
await stopOwned(); state.phase = 'stopped'; state.actual = null;
if (retracting(m.revision)) throw Error('Start cancelled by concurrent Stop');
throw e;
}
} else {
@@ -127,9 +153,16 @@ export async function host(directory: string, url: string) {
} catch (e) {
catalog = { state: 'failed', authentication: 'unverified' };
await stopOwned(); state.phase = 'stopped'; state.actual = null;
if (retracting(m.revision)) throw Error('Start cancelled by concurrent Stop');
throw e;
}
}
// The retraction may also arrive while this admission executes: verified
// teardown must precede the running commit, mirroring the queued path.
if (retracting(m.revision)) {
await stopOwned(); state.phase = 'stopped'; state.actual = null;
throw Error('Start cancelled by concurrent Stop');
}
state.phase = 'running';
} else {
fields(m.body,[]);

132
beehive/test/admission-cancel.test.ts Normal file
View File

@@ -0,0 +1,132 @@
import { test } from 'node:test';
import assert from 'node:assert/strict';
import { mkdtempSync, realpathSync, rmSync, readFileSync, existsSync } 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 { relay } from '../src/relay.ts';
import { connect } from '../src/client.ts';
import { message, newKey, publicKey, type Message } from '../src/protocol.ts';
import { writePrivate } from '../src/storage.ts';
async function until(predicate: () => boolean) {
for (let i = 0; i < 300; i++) { if (predicate()) return; await delay(25); }
throw Error('Missing Start/Stop cancellation evidence');
}
/** A Stop published beside a still-pending Start must cancel that admission
* before any run commits, whether the pair arrives live in one batch or through
* disconnected relay-history replay. No accepted live run may remain. */
test('a Stop received beside a pending Start cancels it before running, live and across replay', async () => {
for (const scenario of ['replay', 'live', 'fixture-live'] as const) {
const dir = realpathSync(mkdtempSync(join(tmpdir(), 'beehive-admission-')));
const secret = newKey(); const agentSecret = newKey(); const agent = publicKey(agentSecret);
const acp = scenario !== 'fixture-live';
writePrivate(join(dir, 'setup.json'), { host: 'admission-host', ownerSecret: secret, agentSecret,
runner: process.execPath, args: acp ? [resolve('test/acp-fixture.ts'), 'delayed'] : [resolve('test/runner.ts')],
workspace: dir, mode: acp ? 'buzz-agent-databricks-v2' : 'fixture',
...(acp ? { serviceHome: dir, configDirectory: dir, databricksHost: 'https://fixture.invalid' } : {}) });
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);
const hostSocket = [...server.clients][0]; assert.ok(hostSocket);
const seen: Message[] = [];
const ui = connect(url, secret, m => seen.push(m)); await ui.ready;
const journal = () => JSON.parse(readFileSync(join(dir, 'journal.json'), 'utf8'));
try {
if (scenario === 'replay') hostSocket.terminate(); // Publish while disconnected; recovery replays relay history.
const began = Date.now();
const start = message('start', 'admission-host', agent, 0);
const stop = message('stop', 'admission-host', agent, 0);
ui.send(start); ui.send(stop); // Same batch as the pending Start: admission must not escape its preceding Stop.
await until(() => seen.some(m => m.type === 'receipt' && m.body.operation === stop.id));
assert.equal(seen.find(m => m.type === 'receipt' && m.body.operation === stop.id)?.body.result, 'accepted');
assert.notEqual(journal().operations[start.id].reply.body.result, 'accepted');
assert.match(String(journal().operations[start.id].reply.body.result), /cancelled/);
assert.equal(journal().phase, 'stopped'); assert.equal(journal().actual, null); assert.equal(journal().revision, 1);
assert.ok(Date.now() - began < 2500, 'Cancellation must not wait for the four-second fixture prompt');
assert.ok(!seen.some(m => m.type === 'inventory' && m.body.phase === 'running'), 'No live run may be left behind');
await h.close(); await delay(200);
assert.ok(!existsSync(join(dir, 'host.lock')), 'Verified stopped state must release the ownership lock');
assert.equal(server.clients.size, 1, 'Closed host must not reconnect');
} finally {
await h.close(); ui.close(); for (const socket of server.clients) socket.terminate();
await new Promise<void>(resolve => server.close(() => resolve())); rmSync(dir, { recursive: true, force: true });
}
}
});
/** The same-batch cancellation fence stays exact: a wrong-authority Stop cannot
* cancel, a Stop published before its Start cannot cancel that later Start, and
* replayed already-terminal operations or a stale revision cannot cancel or rerun. */
test('same-batch Stop cancellation keeps authority, ordering, duplicate and revision fences', async () => {
for (const control of ['wrong-authority', 'conflicting-id', 'invalid-body', 'stop-first', 'duplicate-replay'] as const) {
const dir = realpathSync(mkdtempSync(join(tmpdir(), `beehive-admission-${control}-`)));
const secret = newKey(); const agentSecret = newKey(); const agent = publicKey(agentSecret);
writePrivate(join(dir, 'setup.json'), { host: 'admission-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);
const hostSocket = [...server.clients][0]; assert.ok(hostSocket);
const seen: Message[] = [];
const ui = connect(url, secret, m => seen.push(m)); await ui.ready;
const uiSocket = [...server.clients].find(s => s !== hostSocket); assert.ok(uiSocket);
const journal = () => JSON.parse(readFileSync(join(dir, 'journal.json'), 'utf8'));
const receipt = (id: string) => seen.find(m => m.type === 'receipt' && m.body.operation === id)?.body.result;
try {
hostSocket.terminate(); // All controls replay through disconnected relay history, like the defect.
const began = Date.now();
if (['wrong-authority','conflicting-id','invalid-body'].includes(control)) {
const start = message('start', 'admission-host', agent, 0);
const wrongStop = message('stop', 'admission-host', publicKey(newKey()), 0);
if (control === 'conflicting-id') { wrongStop.agent = agent; wrongStop.id = start.id; }
if (control === 'invalid-body') { wrongStop.agent = agent; wrongStop.body = { invalid: true }; }
ui.send(start); ui.send(wrongStop);
await until(() => receipt(start.id) !== undefined && receipt(wrongStop.id) !== undefined);
assert.equal(receipt(start.id), 'accepted');
if (control === 'wrong-authority') assert.equal(receipt(wrongStop.id), 'not-authority');
if (control === 'conflicting-id') await until(() => seen.some(m => m.body.operation === start.id && m.body.result === 'operation-id-conflict'));
assert.equal(journal().phase, 'running'); assert.equal(journal().revision, 1);
const stop = message('stop', 'admission-host', agent, 1); ui.send(stop);
await until(() => receipt(stop.id) !== undefined);
assert.equal(receipt(stop.id), 'accepted');
assert.equal(journal().phase, 'stopped'); assert.equal(journal().actual, null); assert.equal(journal().revision, 2);
} else if (control === 'stop-first') {
const stop = message('stop', 'admission-host', agent, 0);
const start = message('start', 'admission-host', agent, 0);
ui.send(stop); ui.send(start); // A Stop cannot cancel a Start it never saw.
await until(() => receipt(start.id) !== undefined && receipt(stop.id) !== undefined);
assert.equal(receipt(stop.id), 'accepted'); assert.equal(receipt(start.id), 'revision-conflict');
assert.equal(journal().phase, 'stopped'); assert.equal(journal().actual, null); assert.equal(journal().revision, 1);
} else {
const start = message('start', 'admission-host', agent, 0);
const stop = message('stop', 'admission-host', agent, 0);
ui.send(start); ui.send(stop);
await until(() => receipt(stop.id) !== undefined);
assert.equal(receipt(stop.id), 'accepted'); assert.equal(journal().phase, 'stopped');
const receipts = journal().operations;
// Replay the same committed batch after another disconnect: both operations
// are already terminal, so history must not rerun the Start or recancel anything.
const replayed = [...server.clients].find(s => s !== uiSocket); assert.ok(replayed);
replayed.terminate();
await until(() => [...server.clients].some(s => s !== uiSocket && s !== replayed));
await delay(300);
assert.deepEqual(journal().operations, receipts);
assert.equal(journal().phase, 'stopped'); assert.equal(journal().actual, null); assert.equal(journal().revision, 1);
const stale = message('stop', 'admission-host', agent, 0); ui.send(stale);
await until(() => receipt(stale.id) !== undefined);
assert.equal(receipt(stale.id), 'revision-conflict');
assert.equal(journal().phase, 'stopped'); assert.equal(journal().revision, 1);
}
assert.ok(Date.now() - began < 2500, 'Controls must resolve without waiting on a run');
if (!['wrong-authority','conflicting-id','invalid-body'].includes(control)) assert.ok(!seen.some(m => m.type === 'inventory' && m.body.phase === 'running'), 'No live run may be left behind');
} finally {
await h.close(); ui.close(); for (const socket of server.clients) socket.terminate();
await new Promise<void>(resolve => server.close(() => resolve())); rmSync(dir, { recursive: true, force: true });
}
}
});