fix(reactor-cli): keep serving the other reactors when one reactor's poll rejects

CHANGELOG.md

@@ -13,6 +13,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- `reactor serve` no longer takes every hosted reactor down when one reactor's
+  continuity poll, gateway poll or shutdown rejects: the host settles every
+  reactor before handling failures, the failing reactor is reported (`reactor
+  <name>: poll failed: <message>`, with the categorized error event sampled once
+  per reactor), and the other reactors keep being served. `bootHost` gains an
+  `onReactorError` handler; without one the first failure is still rethrown, but
+  only once every reactor has settled.
+
 ## [reactor 0.3.3 / reactor-cli 0.2.4 / reactor-devtools 0.3.1] - 2026-08-24
 
 Metadata release from the standalone repository. No runtime changes.

packages/reactor-cli/src/__tests__/serve-host.test.ts

@@ -298,6 +298,137 @@ describe('reactor serve host — multi-reactor isolation (offline gate)', () =>
     }
   });
 
+  it('a reactor whose poll rejects is reported while the other reactors keep being served', async () => {
+    const stateRoot = freshState();
+    const projectDir = writeTwoReactorProject(stateRoot);
+    // alpha's freshness reader arms the monitor at boot (a lapsed `valid_until`),
+    // then rejects on the next read: the one the continuity poll performs.
+    let alphaReads = 0;
+    const alphaFreshness = (node: string): unknown => {
+      if (node !== MONITOR) {
+        return null;
+      }
+      alphaReads += 1;
+      if (alphaReads > 1) {
+        throw new Error('alpha freshness store unavailable');
+      }
+      return {
+        node: MONITOR,
+        contract_fingerprint: 'c:monitor@1',
+        input_fingerprints: {},
+        facets: [{ facet: 'freshness', fingerprint: 'f:stale', valid_until: '2020-01-01T00:00:00.000Z' }],
+        prev: null,
+      };
+    };
+    const failures: { name: string; phase: string; message: string }[] = [];
+    try {
+      const host = await bootHost({
+        projectDir,
+        stateDir: stateRoot,
+        offline: true,
+        concurrency: 2,
+        testSeams: {
+          alpha: { ...seamFor(stateRoot, 'alpha'), testReadFreshness: alphaFreshness },
+          beta: seamFor(stateRoot, 'beta'),
+        },
+        onReactorError: (failure) => {
+          failures.push({
+            name: failure.name,
+            phase: failure.phase,
+            message: failure.error instanceof Error ? failure.error.message : String(failure.error),
+          });
+        },
+      });
+      try {
+        const beta = host.byName('beta');
+        assert.ok(beta, 'beta resolvable');
+        const betaReceiptsBefore = beta.reactor.ledger.all().length;
+
+        // The host resolves: alpha's failure is handed over, beta is untouched.
+        await host.pollAll(host.reactors[0]!.reactor.clock.now());
+        assert.deepEqual(failures, [
+          { name: 'alpha', phase: 'poll', message: 'alpha freshness store unavailable' },
+        ]);
+        assert.equal(alphaReads, 2, 'alpha was armed at boot and read once by the poll');
+        assert.equal(beta.reactor.ledger.all().length, betaReceiptsBefore, 'beta ledger untouched');
+
+        // beta keeps being served on the next cycle; alpha keeps being reported.
+        await host.pollAll(host.reactors[0]!.reactor.clock.now());
+        assert.equal(failures.length, 2);
+        assert.ok(failures.every((f) => f.name === 'alpha'));
+        await beta.trigger(MONITOR);
+        assert.ok(beta.reactor.ledger.all().length > betaReceiptsBefore, 'beta still triggers');
+      } finally {
+        await host.shutdown();
+      }
+    } finally {
+      rmSync(stateRoot, { recursive: true, force: true });
+    }
+  });
+
+  it('without an error handler a rejected poll surfaces only after every reactor settled', async () => {
+    const stateRoot = freshState();
+    const projectDir = writeTwoReactorProject(stateRoot);
+    // alpha rejects on the poll's read; beta's reader is armed the same way but
+    // keeps answering, and records the read the poll performs. Serial pool:
+    // alpha polls first, so beta's poll can only run after alpha rejected.
+    const stale = {
+      node: MONITOR,
+      contract_fingerprint: 'c:monitor@1',
+      input_fingerprints: {},
+      facets: [{ facet: 'freshness', fingerprint: 'f:stale', valid_until: '2020-01-01T00:00:00.000Z' }],
+      prev: null,
+    };
+    let alphaReads = 0;
+    const alphaFreshness = (node: string): unknown => {
+      if (node !== MONITOR) {
+        return null;
+      }
+      alphaReads += 1;
+      if (alphaReads > 1) {
+        throw new Error('alpha freshness store unavailable');
+      }
+      return stale;
+    };
+    let betaPollReads = 0;
+    let betaArmed = false;
+    const betaFreshness = (node: string): unknown => {
+      if (node !== MONITOR) {
+        return null;
+      }
+      if (betaArmed) {
+        betaPollReads += 1;
+        return null; // freshness moved: re-arm, fire nothing
+      }
+      betaArmed = true;
+      return stale;
+    };
+    try {
+      const host = await bootHost({
+        projectDir,
+        stateDir: stateRoot,
+        offline: true,
+        concurrency: 1,
+        testSeams: {
+          alpha: { ...seamFor(stateRoot, 'alpha'), testReadFreshness: alphaFreshness },
+          beta: { ...seamFor(stateRoot, 'beta'), testReadFreshness: betaFreshness },
+        },
+      });
+      try {
+        await assert.rejects(
+          host.pollAll(host.reactors[0]!.reactor.clock.now()),
+          /alpha freshness store unavailable/,
+        );
+        // By the time the rejection is observed, beta's poll has already run.
+        assert.equal(betaPollReads, 1, 'beta polled despite alpha rejecting first');
+      } finally {
+        await host.shutdown();
+      }
+    } finally {
+      rmSync(stateRoot, { recursive: true, force: true });
+    }
+  });
+
   it('committed fingerprints match between a concurrency=2 host and a serial host', async () => {
     const fingerprintsFor = async (concurrency: number) => {
       const stateRoot = freshState();

packages/reactor-cli/src/commands/serve.ts

@@ -581,6 +581,7 @@ export async function runServeCommand(
   };
 
   let host: Awaited<ReturnType<typeof bootHost>>;
+  const reactorErrorsReported = new Set<string>();
   try {
     host = await bootHost({
       ...(options.stateDir !== undefined ? { stateDir: options.stateDir } : {}),
@@ -589,6 +590,35 @@ export async function runServeCommand(
       ...(options.offline !== undefined ? { offline: options.offline } : {}),
       ...(options.concurrency !== undefined ? { concurrency: options.concurrency } : {}),
       ...(options.testSeams !== undefined ? { testSeams: options.testSeams } : {}),
+      onReactorError: (failure) => {
+        // One reactor's failed poll is an operator-visible line every cycle, but
+        // the categorized error event is sampled once per reactor (a persistent
+        // fault polled on a 60s cadence would otherwise flood the backend).
+        const msg =
+          failure.error instanceof Error ? failure.error.message : String(failure.error);
+        if (options.json === true) {
+          write(
+            JSON.stringify({
+              status: 'reactor-error',
+              reactor: failure.name,
+              phase: failure.phase,
+              message: msg,
+            }),
+          );
+        } else {
+          write(`reactor ${failure.name}: ${failure.phase} failed: ${msg}`);
+        }
+        if (!reactorErrorsReported.has(failure.name)) {
+          reactorErrorsReported.add(failure.name);
+          telemetry.event(
+            TelemetryEvent.ERROR,
+            buildEventProperties(
+              { command: 'serve', outcome: 'failure', durationMs: Date.now() - startedAt },
+              { errorCategory: errorCategory(failure.error) },
+            ),
+          );
+        }
+      },
     });
   } catch (err) {
     // Boot failure (stale IR / compile failed / no contracts) gets a clean

packages/reactor-cli/src/run/host.ts

@@ -68,6 +68,19 @@ export interface BootHostOptions extends ConfigOverrides {
    * A single-reactor host uses the name `"default"`.
    */
   readonly testSeams?: Readonly<Record<string, PerReactorTestSeam>>;
+  /**
+   * Receives each reactor whose poll, gateway poll or shutdown rejected. With a
+   * handler the host keeps serving the other reactors and resolves; without one
+   * the first failure is rethrown, but only once every reactor has settled.
+   */
+  readonly onReactorError?: (failure: ReactorFailure) => void;
+}
+
+/** One hosted reactor's rejected poll, gateway poll or shutdown. */
+export interface ReactorFailure {
+  readonly name: string;
+  readonly phase: 'poll' | 'gateways' | 'shutdown';
+  readonly error: unknown;
 }
 
 /** A running multi-reactor host: the isolated handles + the across-reactor pool. */
@@ -181,20 +194,51 @@ export async function bootHost(options: BootHostOptions = {}): Promise<HostHandl
   const pool = createWorkerPool(options.concurrency ?? 1);
   const byNameMap = new Map(handles.map((h) => [h.name, h] as const));
 
+  // Reactors are isolated: one reactor's rejection must not cut the others'
+  // work short, so every per-reactor task settles before failures are handled.
+  // With `onReactorError` each failure is handed over and the host carries on;
+  // without it the first failure is rethrown once everything has settled.
+  const settleAll = async <T>(
+    phase: ReactorFailure['phase'],
+    run: (handle: ServeHandle) => Promise<T>,
+  ): Promise<readonly T[]> => {
+    const settled = await Promise.allSettled(handles.map((h) => run(h)));
+    const fulfilled: T[] = [];
+    let firstError: { error: unknown } | undefined;
+    settled.forEach((result, i) => {
+      if (result.status === 'fulfilled') {
+        fulfilled.push(result.value);
+        return;
+      }
+      const failure: ReactorFailure = {
+        name: handles[i]!.name,
+        phase,
+        error: result.reason,
+      };
+      if (options.onReactorError !== undefined) {
+        options.onReactorError(failure);
+      } else {
+        firstError ??= { error: result.reason };
+      }
+    });
+    if (firstError !== undefined) {
+      throw firstError.error;
+    }
+    return fulfilled;
+  };
+
   const pollAll = async (now: string): Promise<void> => {
     // Submit each reactor's poll to the across-reactor pool (bounded parallel);
     // each poll is itself serialized behind that reactor's own queue.
-    await Promise.all(
-      handles.map((h) => pool.submit(() => h.pollOnce(now))),
-    );
+    await settleAll('poll', (h) => pool.submit(() => h.pollOnce(now)));
   };
 
   const pollGatewaysAll = async (now: string): Promise<readonly GatewayPollOutcome[]> => {
     // Each reactor's gateway poll is submitted to the across-reactor pool; the
     // poll itself enqueues onto that reactor's serialization queue (correction #4),
     // so within a reactor a gateway poll never overlaps a continuity poll/trigger.
-    const perReactor = await Promise.all(
-      handles.map((h) => pool.submit(() => h.pollGatewaysOnce(now))),
+    const perReactor = await settleAll('gateways', (h) =>
+      pool.submit(() => h.pollGatewaysOnce(now)),
     );
     return perReactor.flat();
   };
@@ -209,8 +253,18 @@ export async function bootHost(options: BootHostOptions = {}): Promise<HostHandl
   };
 
   const shutdown = async (): Promise<void> => {
-    await Promise.all(handles.map((h) => h.shutdown()));
+    // Drain every reactor before surfacing a failed shutdown, so one reactor's
+    // rejection never leaves another's in-flight work undrained.
+    let failure: unknown;
+    try {
+      await settleAll('shutdown', (h) => h.shutdown());
+    } catch (err) {
+      failure = err;
+    }
     await pool.onIdle();
+    if (failure !== undefined) {
+      throw failure;
+    }
   };
 
   return {