diff --git a/beehive/CHECKPOINT.md b/beehive/CHECKPOINT.md index 60d95844b5..3e586c87fa 100644 --- a/beehive/CHECKPOINT.md +++ b/beehive/CHECKPOINT.md @@ -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`. diff --git a/beehive/TESTING.md b/beehive/TESTING.md index b28ed8a534..bb0bb46e4c 100644 --- a/beehive/TESTING.md +++ b/beehive/TESTING.md @@ -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. diff --git a/beehive/src/host.ts b/beehive/src/host.ts index b3013bc42a..e1712af992 100644 --- a/beehive/src/host.ts +++ b/beehive/src/host.ts @@ -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(); + const retracted = new Map(); + const retracting = (revision: number) => { for (const target of retracted.values()) if (target === revision) return true; return false; }; let client: ReturnType; 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,[]); diff --git a/beehive/test/admission-cancel.test.ts b/beehive/test/admission-cancel.test.ts new file mode 100644 index 0000000000..192e332301 --- /dev/null +++ b/beehive/test/admission-cancel.test.ts @@ -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(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(resolve => server.close(() => resolve())); rmSync(dir, { recursive: true, force: true }); + } + } +});