From bec3583286921967debc8e8fda800fc4eadfca45 Mon Sep 17 00:00:00 2001 From: Jason Woltje Date: Tue, 3 Nov 2026 18:15:48 -0600 Subject: [PATCH] fix(bus): keep the request route across setup waits and recheck association after admission (#1839) --- packages/bus/src/request-lifecycle.mjs | 68 +++++++--- packages/bus/src/request-store.mjs | 6 +- packages/bus/tests/request-intake.test.mjs | 137 +++++++++++++++++++++ 3 files changed, 189 insertions(+), 22 deletions(-) diff --git a/packages/bus/src/request-lifecycle.mjs b/packages/bus/src/request-lifecycle.mjs index 3ac0cfb83..ab04b732d 100644 --- a/packages/bus/src/request-lifecycle.mjs +++ b/packages/bus/src/request-lifecycle.mjs @@ -16,11 +16,12 @@ export function createRequestLifecycle({ journal, services }) { const stateOf = (business, request) => journal.readRequest(business, request) ?? refuse('request-not-found'); const actor = identity => ({ role: identity?.role ?? null, run: identity?.run ?? null }); const authorize = (identity, state, operation) => services.authorize(identity, state, operation); - const append = (state, kind, body, identity = null) => journal.appendRequest({ business: state.business, request: state.request, + // `write` is journal.appendRequest outside a transaction and tx.appendRequest inside one. + const append = (state, kind, body, identity = null, write = journal.appendRequest) => write({ business: state.business, request: state.request, expectedRevision: state.revision, kind, body, actor: actor(identity) }); - const wait = (state, code, identity = null, proposal = {}) => { + const wait = (state, code, identity = null, proposal = {}, write = journal.appendRequest) => { if (state.wait?.kind === 'setup' && state.wait.code === code && !proposal.op) return state; - return append(state, 'setup.wait', { code, subject: state.request, ...proposal }, identity); + return append(state, 'setup.wait', { code, subject: state.request, ...proposal }, identity, write); }; async function open(identity, args) { exact(args, ['request']); if (!id(args.request)) refuse('invalid-request'); @@ -79,6 +80,20 @@ export function createRequestLifecycle({ journal, services }) { } return p; } + // The host inventory and admission are asynchronous and neither bumps the request revision, so a + // plan can go stale while a call is in flight. `staleCode` is the guard for that: callers read the + // inventory as their last await and call it inside the synchronous journal transaction, so no + // mapping or board change can land between this check and the append. It never confirms a stale plan. + const associated = (rows, p) => Array.isArray(rows) && rows.some(m => object(m) && m.business === p.business && m.project === p.project && + m.name === p.name && m.tracker?.root === p.tracker.root && m.tracker?.project === p.tracker.project); + function staleCode(plan, state, args, source, rows) { + if (plan.existing && !associated(rows, plan)) return 'request-existing-project-changed'; + try { localPlan(plan, state, args, source); } catch (e) { + if (!e.code?.startsWith('request-')) throw e; + return e.code; + } + return null; + } async function admission(plan, state) { if (!services.admitPlan) return 'request-admission-unavailable'; const result = await services.admitPlan(structuredClone(plan), structuredClone(state)); @@ -98,7 +113,8 @@ export function createRequestLifecycle({ journal, services }) { if (state.phase !== 'selection' || state.wait?.kind === 'decision') refuse('request-phase-refused'); const source = services.origin(identity, args.request); if (!services.project(state.business, args.source_project)) refuse('request-project-refused'); - const existing = state.wait?.code === 'request-existing-project'; + // The route is journaled when the owner stages it; no setup wait code can erase or imply it. + const existing = state.route === 'existing'; if (!existing && services.project(state.business, args.project)) refuse('request-project-collision'); let candidate = null; if (existing) { @@ -111,7 +127,7 @@ export function createRequestLifecycle({ journal, services }) { if (current.revision !== args.revision) refuse('request-revision-conflict'); return current; }; - const pendingWait = code => wait(recheck(), code, identity, { op: args.op, proposal_digest }); + const pendingWait = (code, write) => wait(recheck(), code, identity, { op: args.op, proposal_digest }, write); if (!prepare) return pendingWait('request-preparation-unavailable'); const prepared = await prepare(structuredClone(state), structuredClone(args), structuredClone(source)); recheck(); @@ -124,9 +140,13 @@ export function createRequestLifecycle({ journal, services }) { } const plans = existing ? [prepared] : prepared; if (!Array.isArray(plans) || plans.length !== (existing ? 1 : 2)) refuse('invalid-request-plan'); + // Validate every plan before the first admission await. A change after this point is not a + // malformed plan; the guard in the final transaction turns it into a stale wait. for (let i = 0; i < plans.length; i++) { const p = localPlan(plans[i], state, args, source); if (p.existing !== existing || (!existing && p.mode !== ['autonomous','guided'][i]) || (existing && p.tracker.root !== candidate.tracker.root)) refuse('request-plan-binding-refused'); + } + for (const p of plans) { const missing = await admission(p, state); recheck(); if (missing) return pendingWait(missing); } // Both modes must display the same concrete resources and original intent. @@ -136,8 +156,15 @@ export function createRequestLifecycle({ journal, services }) { const question = `Confirm the resource plan for ${args.name}. ${existing ? `Existing project mode: ${mode}.` : 'Both new project choices use these resources with the selected mode.'} Original user intent and exact resources:\n${JSON.stringify(resources)}`; // Preserve the ordinary decision channel's existing display limit and all source bytes. if (question.length > 4096) return pendingWait('request-plan-display-too-large'); + // Last await. Preparation and admission may have outlived the association they bound, so read + // it again and let the synchronous guard in the transaction decide before any decision exists. + const rows = existing ? await projects(identity, { request: args.request }) : null; return journal.transaction(tx => { - authorize(identity, state, 'propose'); + state = recheck(); + for (const plan of plans) { + const late = staleCode(plan, state, args, source, rows); + if (late) return pendingWait(late, tx.appendRequest); + } const options = existing ? [{ key: 'confirm', text: 'Confirm this existing project association' }, { key: 'back', text: 'Choose another existing project' }] : [{ key: 'autonomous', text: 'New project — autonomous' }, { key: 'guided', text: 'New project — guided' }, { key: 'existing', text: 'Add to existing project' }]; const decision = services.raiseDecision(identity, { action: 'project.bootstrap', question, options, recommendation: options[0].key, @@ -161,21 +188,22 @@ export function createRequestLifecycle({ journal, services }) { const binding = state.selection.bindings.find(b => b.choice === close.choice); if (!binding) refuse('request-selection-refused'); const p = binding.plan; - if (p.existing) { - const mappings = await services.listProjects?.(structuredClone(state)) ?? []; - if (!Array.isArray(mappings) || !mappings.some(m => m.business === business && m.project === p.project && m.name === p.name && - m.tracker?.root === p.tracker.root && m.tracker?.project === p.tracker.project)) return wait(state, 'request-existing-project-changed'); - } - try { - localPlan(p, state, { project: p.project, name: p.name, source_project: p.source.project, tracker_project: p.tracker.project }, - { text: p.intent.text, source_message: state.source_message }); - } catch (e) { - if (!e.code?.startsWith('request-')) throw e; - return wait(state, e.code); - } - const missing = await admission(binding.plan, state); + const planArgs = { project: p.project, name: p.name, source_project: p.source.project, tracker_project: p.tracker.project }; + const planSource = { text: p.intent.text, source_message: state.source_message }; + const inventory = async () => p.existing ? await services.listProjects?.(structuredClone(state)) ?? [] : null; + // Check before asking admission, so a plan that is already stale never reaches it. + const early = staleCode(p, state, planArgs, planSource, await inventory()); + if (early) return wait(state, early); + const missing = await admission(p, state); if (missing) return wait(state, missing); - return append(state, 'selection.confirmed', { decision: binding.decision, choice: binding.choice, digest: binding.digest }); + // Admission is the last slow call. Read the association once more and confirm only if it still + // matches, in the same synchronous step as the append. + const rows = await inventory(); + return journal.transaction(tx => { + const late = staleCode(p, state, planArgs, planSource, rows); + if (late) return wait(state, late, null, {}, tx.appendRequest); + return append(state, 'selection.confirmed', { decision: binding.decision, choice: binding.choice, digest: binding.digest }, null, tx.appendRequest); + }); }); } return Object.freeze({ open, get, projects, propose, resume, diff --git a/packages/bus/src/request-store.mjs b/packages/bus/src/request-store.mjs index d4bfbd47e..2e1219686 100644 --- a/packages/bus/src/request-store.mjs +++ b/packages/bus/src/request-store.mjs @@ -55,7 +55,9 @@ function initial(business, request, b) { if (!text(b.intent.text) || digest(b.intent.text) !== b.intent.sha256 || b.intent.source_message !== b.source_message || b.intent.generated !== false) refuse('invalid-request-event'); return { version: 1, request, business, revision: 1, source_message: b.source_message, thread: b.thread, session: b.session, owner: b.owner, pm: b.pm, phase: 'selection', project: null, chat_project: null, mode: null, mode_revision: 0, - plan_digest: null, spec_commit: null, snapshot: null, tasks: [], delivery: null, question: null, candidate: null, selection: null, wait: null }; + plan_digest: null, spec_commit: null, snapshot: null, tasks: [], delivery: null, question: null, candidate: null, selection: null, + // The owner's chosen route survives every setup wait; a wait code never says which route was chosen. + route: null, wait: null }; } function reduce(state, e) { const code = 'invalid-request-event', b = e.body; @@ -84,7 +86,7 @@ function reduce(state, e) { exact(b, ['decision','choice'], [], code); if (s.wait?.kind !== 'decision' || s.selection?.decision !== b.decision || !['existing','back'].includes(b.choice) || (b.choice === 'existing') !== (s.wait.code === 'request-mode')) refuse(code); - s.selection = null; + s.selection = null; s.route = 'existing'; s.wait = { kind: 'setup', code: 'request-existing-project', subject: s.request, resume_phase: 'selection', observed_revision: s.revision }; } else if (e.kind === 'selection.confirmed') { exact(b, ['decision','choice','digest'], [], code); diff --git a/packages/bus/tests/request-intake.test.mjs b/packages/bus/tests/request-intake.test.mjs index 20af68dba..7342b6d34 100644 --- a/packages/bus/tests/request-intake.test.mjs +++ b/packages/bus/tests/request-intake.test.mjs @@ -406,3 +406,140 @@ test('runtime_restart_recovers_a_recorded_autonomous_selection_once_without_boot assert.equal(existsSync(join(x.root, 'products')), false); assert.equal(existsSync(join(x.root, 'projects')), false); }); + +// Review r1 I2/I3: a host whose inventory, faults and in-flight calls the test controls. +function controlled(t) { + const control = { board: 2, fault: {}, pause: null }; + const hold = async at => { + const pause = control.pause; + if (pause?.at !== at) return; + control.pause = null; pause.start(); await pause.gate; + }; + // Arm one pause; the next preparation/admission call of that kind parks until `release()`. + control.pauseAt = at => { + let start, release; + const started = new Promise(r => { start = r; }), gate = new Promise(r => { release = r; }); + control.pause = { at, start, gate }; return { started, release }; + }; + const x = setup(t, { prepared: true, configureHost: host => { + host.listProjects = () => [{ business: 'demo', project: 'old', name: 'Old', tracker: { root: 1, project: control.board } }]; + const { prepareExisting, preparePlans, admitPlan } = host; + host.prepareExisting = async (...args) => { + await hold('prepare'); + if (control.fault.prepare) return { ready: false, code: control.fault.prepare }; + const plan = prepareExisting(...args); + if (control.fault.big) plan.project_vars = { suites: Array.from({ length: 20 }, (_, i) => String(i).padStart(3, '0') + 'a'.repeat(280)) }; + return plan; + }; + host.preparePlans = async (...args) => { await hold('prepare'); return preparePlans(...args); }; + host.admitPlan = async plan => { await hold('admit'); return control.fault.admit ? { ready: false, code: control.fault.admit } : admitPlan(plan); }; + } }); + return { x, control }; +} +async function reachPicker(x) { + const opened = await x.call('pm', 'request.open', { request: x.source.request }); + const mode = await x.call('pm', 'request.propose', x.propose(opened)); + x.call('human', 'decision.resolve', { id: mode.selection.decision, choice: 'existing' }); + return x.lifecycle.resume('demo', opened.request); +} +const existingArgs = (state, op, extra = {}) => ({ request: state.request, revision: state.revision, op, source_project: 'old', project: 'old', name: 'Old', tracker_project: 2, ...extra }); +const count = (x, sql) => x.store.get(sql).n; +const moveOldMapping = x => x.b.reloadBusiness('demo', { ...x.business, projects: { ...x.business.projects, old: { ...x.business.projects.old, repo: join(x.root, 'old-code-moved') } } }); +// A fresh-operation retry after the setup correction must land in the existing-project confirmation and bind it. +async function confirmRetry(x, retried) { + assert.equal(retried.route, 'existing'); assert.equal(retried.project, null); + assert.equal(retried.wait.kind, 'decision'); assert.equal(retried.wait.code, 'request-existing-confirmation'); + assert.equal(retried.selection.bindings.length, 1); assert.equal(retried.selection.bindings[0].plan.existing, true); + x.call('human', 'decision.resolve', { id: retried.selection.decision, choice: 'confirm' }); + const bound = await x.lifecycle.resume('demo', retried.request); + assert.equal(bound.project, 'old'); assert.equal(bound.phase, 'discovery'); assert.equal(bound.wait.code, 'request-attachment-unavailable'); + assert.deepEqual(bound.tasks, []); assert.equal(bound.snapshot, null); +} + +for (const w of [ + { name: 'preparation', code: 'tracker-temporarily-unavailable', arm: c => { c.fault.prepare = 'tracker-temporarily-unavailable'; } }, + { name: 'admission', code: 'organization-stale', arm: c => { c.fault.admit = 'organization-stale'; } }, + { name: 'oversized_display', code: 'request-plan-display-too-large', arm: c => { c.fault.big = true; } }, +]) test(`existing_route_survives_a_${w.name}_wait_and_a_fresh_operation_retry_resumes_the_picker`, async t => { + const { x, control } = controlled(t), picker = await reachPicker(x); + assert.equal(picker.route, 'existing'); assert.equal(picker.wait.code, 'request-existing-project'); + w.arm(control); + const waiting = await x.call('pm', 'request.propose', existingArgs(picker, 'existing-1')); + assert.equal(waiting.wait.kind, 'setup'); assert.equal(waiting.wait.code, w.code); + assert.equal(waiting.route, 'existing'); assert.equal(waiting.selection, null); assert.equal(waiting.project, null); + assert.equal(count(x, 'SELECT count(*) n FROM decisions'), 1, 'only the mode decision exists'); + control.fault = {}; // the capability is restored + const retried = await x.call('pm', 'request.propose', existingArgs(waiting, 'existing-2')); + assert.equal(count(x, 'SELECT count(*) n FROM decisions'), 2); + // The earlier operation stays one immutable wait event; replaying it records nothing new. + assert.equal(count(x, "SELECT count(*) n FROM request_events WHERE json_extract(body,'$.op')='existing-1'"), 1); + const events = count(x, 'SELECT count(*) n FROM request_events'); + assert.deepEqual(await x.call('pm', 'request.propose', existingArgs(picker, 'existing-1')), x.journal.readRequest('demo', picker.request)); + assert.equal(count(x, 'SELECT count(*) n FROM request_events'), events); + await confirmRetry(x, retried); +}); + +test('existing_route_survives_a_changed_association_wait_and_a_fresh_operation_retry_resumes_the_picker', async t => { + const { x, control } = controlled(t), picker = await reachPicker(x); + const candidate = await x.call('pm', 'request.propose', existingArgs(picker, 'existing-1')); + control.board = 3; + x.call('human', 'decision.resolve', { id: candidate.selection.decision, choice: 'confirm' }); + const stopped = await x.lifecycle.resume('demo', picker.request); + assert.equal(stopped.wait.kind, 'setup'); assert.equal(stopped.wait.code, 'request-existing-project-changed'); + assert.equal(stopped.route, 'existing'); assert.equal(stopped.project, null); + const retried = await x.call('pm', 'request.propose', existingArgs(stopped, 'existing-2', { tracker_project: 3 })); + assert.equal(retried.selection.bindings[0].plan.tracker.project, 3); + await confirmRetry(x, retried); +}); + +test('new_project_route_still_refuses_an_existing_project_name', async t => { + const { x } = controlled(t), opened = await x.call('pm', 'request.open', { request: x.source.request }); + assert.equal(opened.route, null); + await assert.rejects(async () => x.call('pm', 'request.propose', { ...x.propose(opened), project: 'old' }), /request-project-collision/); +}); + +// Review r1 I3: the association changes while preparation or admission is in flight. +for (const c of [ + { name: 'existing_board', route: 'existing', code: 'request-existing-project-changed', change: (x, control) => { control.board = 3; } }, + { name: 'existing_mapping', route: 'existing', code: 'request-plan-binding-refused', change: moveOldMapping }, + { name: 'new_project_collision', route: 'mode', code: 'request-project-collision', + change: x => x.b.reloadBusiness('demo', { ...x.business, projects: { ...x.business.projects, timer: { root: join(x.root, 'timer-records') } } }) }, +]) test(`selection_is_not_confirmed_when_${c.name}_changes_during_admission`, async t => { + const { x, control } = controlled(t); + let state, choice; + if (c.route === 'existing') { + state = await x.call('pm', 'request.propose', existingArgs(await reachPicker(x), 'existing-1')); choice = 'confirm'; + } else { + state = await x.call('pm', 'request.propose', x.propose(await x.call('pm', 'request.open', { request: x.source.request }))); choice = 'autonomous'; + } + x.call('human', 'decision.resolve', { id: state.selection.decision, choice }); + const gate = control.pauseAt('admit'), resuming = x.lifecycle.resume('demo', state.request); + await gate.started; await c.change(x, control); gate.release(); + const stopped = await resuming; + assert.equal(count(x, "SELECT count(*) n FROM request_events WHERE kind='selection.confirmed'"), 0); + assert.equal(stopped.wait.kind, 'setup'); assert.equal(stopped.wait.code, c.code); + assert.equal(stopped.project, null); assert.equal(stopped.mode, null); assert.equal(stopped.plan_digest, null); + assert.equal(stopped.phase, 'selection'); assert.deepEqual(stopped.tasks, []); + assert.deepEqual(await x.lifecycle.resume('demo', state.request), stopped); +}); + +for (const c of [ + { name: 'existing_board_during_preparation', route: 'existing', at: 'prepare', code: 'request-existing-project-changed', change: (x, control) => { control.board = 3; } }, + { name: 'existing_board_during_admission', route: 'existing', at: 'admit', code: 'request-existing-project-changed', change: (x, control) => { control.board = 3; } }, + { name: 'existing_mapping_during_admission', route: 'existing', at: 'admit', code: 'request-plan-binding-refused', change: moveOldMapping }, + { name: 'new_project_collision_during_admission', route: 'mode', at: 'admit', code: 'request-project-collision', + change: x => x.b.reloadBusiness('demo', { ...x.business, projects: { ...x.business.projects, timer: { root: join(x.root, 'timer-records') } } }) }, +]) test(`proposal_records_a_stale_wait_not_a_decision_when_${c.name}`, async t => { + const { x, control } = controlled(t); + const base = c.route === 'existing' ? await reachPicker(x) : await x.call('pm', 'request.open', { request: x.source.request }); + const args = c.route === 'existing' ? existingArgs(base, 'late-1') : x.propose(base); + const decisions = count(x, 'SELECT count(*) n FROM decisions'), proposed = count(x, "SELECT count(*) n FROM request_events WHERE kind='selection.proposed'"); + const gate = control.pauseAt(c.at), proposing = x.call('pm', 'request.propose', args); + await gate.started; await c.change(x, control); gate.release(); + const stopped = await proposing; + assert.equal(count(x, 'SELECT count(*) n FROM decisions'), decisions, 'no decision for a stale plan'); + assert.equal(count(x, "SELECT count(*) n FROM request_events WHERE kind='selection.proposed'"), proposed); + assert.equal(stopped.wait.kind, 'setup'); assert.equal(stopped.wait.code, c.code); + assert.equal(stopped.selection, null); assert.equal(stopped.project, null); assert.equal(stopped.phase, 'selection'); + assert.equal(stopped.route, c.route === 'existing' ? 'existing' : null); +});