fix(bus): keep the request route across setup waits and recheck association after admission (#1839)

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 planArgs = { project: p.project, name: p.name, source_project: p.source.project, tracker_project: p.tracker.project };
-        const mappings = await services.listProjects?.(structuredClone(state)) ?? [];
+      const planSource = { text: p.intent.text, source_message: state.source_message };
-        if (!Array.isArray(mappings) || !mappings.some(m => m.business === business && m.project === p.project && m.name === p.name &&
+      const inventory = async () => p.existing ? await services.listProjects?.(structuredClone(state)) ?? [] : null;
-            m.tracker?.root === p.tracker.root && m.tracker?.project === p.tracker.project)) return wait(state, 'request-existing-project-changed');
+      // Check before asking admission, so a plan that is already stale never reaches it.
-      }
+      const early = staleCode(p, state, planArgs, planSource, await inventory());
-      try {
+      if (early) return wait(state, early);
-        localPlan(p, state, { project: p.project, name: p.name, source_project: p.source.project, tracker_project: p.tracker.project },
+      const missing = await admission(p, state);
-          { 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);
       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,

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

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