fix(bus): defer halted startup recovery; latest launch only gates notices (#1848)

Review round 1 on 4048e0f60 found two defects.

Startup recovery ran unconditionally. A continuation retry now refuses
while processing is halted, so a broker restart with a waiting
continuation failed with processing-halted. Startup recovery now waits
while processing is halted or invalid, as the periodic pass does;
processing.resume schedules it again.

#continuationStarting matched any launch without an end, so an older
launch whose end was lost (#1851) hid every later run's end. Only the
target's latest lifecycle event counts now.

Both regression tests fail on 4048e0f60 and pass here.

packages/bus/src/broker.mjs

@@ -720,8 +720,9 @@ export class Broker {
       prerequisite: digestPrd({ business: b, revision, claims, revoked, credentials: this.#credentialStatus() }) };
   }
   #continuationStarting(business, target) {
-    // A launch without an end is still starting when the target has no holder.
-    return !!this.#store.get("SELECT 1 FROM events l WHERE l.business=? AND l.actor_role=? AND l.kind='session.launched' AND NOT EXISTS (SELECT 1 FROM events e WHERE e.business=l.business AND e.kind='session.ended' AND e.subject=l.subject) LIMIT 1", business, target);
+    // The target's latest lifecycle event is a launch without an end: it is still starting. An older
+    // launch whose end was lost (#1851) never hides a later run's end.
+    return !!this.#store.get("SELECT 1 FROM events l WHERE l.business=? AND l.actor_role=? AND l.kind='session.launched' AND l.seq=(SELECT max(seq) FROM events WHERE business=l.business AND actor_role=l.actor_role AND kind IN ('session.launched','session.ended')) AND NOT EXISTS (SELECT 1 FROM events e WHERE e.business=l.business AND e.kind='session.ended' AND e.subject=l.subject)", business, target);
   }
   #continuationMessage(state, target, body, context) {
     this.#processingWork();

packages/bus/src/runtime.mjs

@@ -1025,7 +1025,10 @@ export async function startBroker({
     usageTimer = setInterval(usageTick, 1000);
     usageTimer.unref();
     usageTick(); // Optional telemetry never blocks the native ready answer.
-    await requestLifecycle.recover();
+    // Startup recovery waits while processing is halted or invalid, like the periodic pass below;
+    // processing.resume schedules it again.
+    const startState = processing.read();
+    if (!startState.halted && !startState.invalid) await requestLifecycle.recover();
     requestTimer = setInterval(() => {
       const state = processing.read();
       if (state.halted || state.invalid) return;

packages/bus/tests/request-continuation.test.mjs

@@ -1094,3 +1094,42 @@ test('delivery continuation reports advance refusal and requires delivery phase'
   assert.match(body, /refus.*message.*owner thread/i);
   assert.match(body, /do not tell the owner.*delivered/i);
 });
+
+// #1848 review: a continuation retry refuses while processing is halted. Startup recovery must wait
+// for processing.resume instead of failing the broker start, and resume must still send the notice.
+test('a halted bus restarts with a waiting continuation and notices the chief after resume', async t => {
+  const f = await requestFixture(t);
+  await f.bus.requestLifecycle.recover();
+  const attempts = () => f.journal.effects('demo', f.request).filter(e => e.result?.type === 'attempt').length;
+  const notices = () => f.store.get("SELECT count(*) n FROM messages WHERE from_role='_bus' AND to_role='chief' AND body LIKE 'Message %'").n;
+  const halt = await f.call('human', 'processing.halt');
+  f.bus.broker.endLaunch({ business: 'demo', run: 'pm-run', reason: 'exit' });
+  await assert.rejects(f.bus.requestLifecycle.recover(), { code: 'processing-halted' });
+  const before = attempts(), told = notices();
+  await f.bus.close();
+  const next = await f.boot();
+  const human = next.broker.bindHuman({ business: 'demo', human: 'jason', via: 'cli', outsideAgent: true });
+  assert.equal(next.broker.request(human, { verb: 'processing.status', args: {} }).halted, true);
+  assert.equal(attempts(), before, 'no attempt while halted');
+  assert.equal(notices(), told, 'no notice while halted');
+  await next.broker.request(human, { verb: 'processing.resume', args: { generation: halt.generation } });
+  const deadline = Date.now() + 5000;
+  while (attempts() === before && Date.now() < deadline) await new Promise(r => setTimeout(r, 20));
+  assert.equal(attempts(), before + 1);
+  assert.equal(notices(), told + 1);
+});
+
+// #1848 review: an older launch whose end was lost (#1851) must not hide a later run's real end.
+test('a stale unmatched launch does not suppress the notice when a later target run ends', async t => {
+  const f = await requestFixture(t);
+  await f.bus.requestLifecycle.recover();
+  const notices = () => f.store.get("SELECT count(*) n FROM messages WHERE from_role='_bus' AND to_role='chief' AND body LIKE 'Message %'").n;
+  f.call('pm', 'role.release'); // pm-run keeps its session.launched with no session.ended.
+  const successor = f.bus.broker.bindLaunch({ business: 'demo', role: 'pm', run: 'pm-successor', harness: 'pi' });
+  f.bus.broker.request(successor, { verb: 'role.claim', args: {} });
+  await f.bus.requestLifecycle.recover();
+  const before = notices();
+  f.bus.broker.endLaunch({ business: 'demo', run: 'pm-successor', reason: 'exit' });
+  await f.bus.requestLifecycle.recover();
+  assert.equal(notices(), before + 1);
+});