feat(standards): per-measure executor router, wired and dark (PR-7b)

backend-ts/src/admin/scheduler.ts

@@ -17,7 +17,7 @@ import type { Stores, StoresEnv } from "../stores/factory.ts";
 import { getStores } from "../stores/factory.ts";
 import type { HydratedSegment } from "../stores/segment-store.ts";
 import type { EvaluateMeasureBinding } from "../engine/evaluate-measure.ts";
-import { engineForEnv } from "../wiring/engine-factory.ts";
+import { routedEngineForEnv } from "../wiring/executor-router.ts";
 import type { EmployeeProfile } from "../engine/synthetic/employee-catalog.ts";
 import { ensureSegmentSeed } from "../segment/segment-seed.ts";
 import {
@@ -330,7 +330,7 @@ export async function schedulerTick(
   try {
     await ensureSegmentSeed(env);
     const stores = await getStores(env);
-    const engine = await engineForEnv(env);
+    const engine = await routedEngineForEnv(env);
     const allSegments = await stores.segments.listSegments();
     const enabledSegments = allSegments.filter((s) => s.enabled);
     await runTick({ stores, engine, segments: enabledSegments, alertChannels, webChartEnv: env, incremental: isIncrementalEnabled(env), expansionActive: isVsacConfigured(env) });

backend-ts/src/config/seam-inventory.test.ts

@@ -20,15 +20,15 @@ test("describeSeams: everything off with no env vars set", () => {
   const seams = describeSeams({});
   assert.deepEqual(
     seams.map((s) => s.name),
-    ["sendgrid", "datachaser", "ice", "eh-fhir", "webchart", "sql-executor", "vsac", "alert-webhook", "bucket-s3", "incremental-eval"],
+    ["sendgrid", "datachaser", "ice", "eh-fhir", "webchart", "sql-executor", "vsac", "alert-webhook", "bucket-s3", "incremental-eval", "official-measures"],
   );
   for (const s of seams) assert.equal(s.active, false, `${s.name} should default off`);
 });
 
 test("formatSeamLogLine: all-off shape matches the documented boot log line", () => {
   assert.equal(
     formatSeamLogLine({}),
-    "seams: sendgrid=off datachaser=off ice=off eh-fhir=off webchart=off sql-executor=off vsac=off alert-webhook=off bucket-s3=off incremental-eval=off",
+    "seams: sendgrid=off datachaser=off ice=off eh-fhir=off webchart=off sql-executor=off vsac=off alert-webhook=off bucket-s3=off incremental-eval=off official-measures=off",
   );
 });
 
@@ -242,9 +242,10 @@ test("formatSeamLogLine: all-on shape when every seam is configured", () => {
     WORKWELL_BUCKET_S3_ACCESS_KEY_ID: "AKIA",
     WORKWELL_BUCKET_S3_SECRET_ACCESS_KEY: "s",
     WORKWELL_INCREMENTAL_EVAL: "true",
+    WORKWELL_OFFICIAL_MEASURES: "cms122,cms125",
   };
   assert.equal(
     formatSeamLogLine(allOn),
-    "seams: sendgrid=on datachaser=on ice=on eh-fhir=on webchart=on sql-executor=on vsac=on alert-webhook=on bucket-s3=on incremental-eval=on",
+    "seams: sendgrid=on datachaser=on ice=on eh-fhir=on webchart=on sql-executor=on vsac=on alert-webhook=on bucket-s3=on incremental-eval=on official-measures=on",
   );
 });

backend-ts/src/config/seam-inventory.ts

@@ -27,6 +27,7 @@ import { isSqlPushdownSelected, type MeasureExecutorEnv } from "../engine/measur
 import { isVsacConfigured, type VsacEnv } from "../engine/cql/resolve-value-set-resolver.ts";
 import { isAlertWebhookConfigured, type AlertEnv } from "../run/alert-channel.ts";
 import { isIncrementalEnabled, type IncrementalEnv } from "../run/incremental/incremental-eval.ts";
+import { isOfficialRoutingConfigured, type OfficialMeasuresEnv } from "../wiring/official-routing.ts";
 
 /** The union of every seam's env-var shape (all optional — assignable from the worker's `Env`). */
 export type SeamEnv = EmailEnv &
@@ -38,7 +39,8 @@ export type SeamEnv = EmailEnv &
   VsacEnv &
   AlertEnv &
   BucketSeamEnv &
-  IncrementalEnv;
+  IncrementalEnv &
+  OfficialMeasuresEnv;
 
 export interface SeamStatus {
   /** Short, stable, log-line-friendly seam name. */
@@ -62,6 +64,11 @@ export function describeSeams(env: SeamEnv): SeamStatus[] {
     { name: "alert-webhook", active: isAlertWebhookConfigured(env) },
     { name: "bucket-s3", active: isS3BucketConfigured(env) },
     { name: "incremental-eval", active: isIncrementalEnabled(env) },
+    // The 11th seam, and the only one that changes what a measure COMPUTES rather than how or where.
+    // The line reports only on/off, like every other seam; WHICH measures are routed comes from the
+    // OFFICIAL_ROUTING_MISCONFIGURED alert and the run's own evidence. On/off is still the thing worth
+    // seeing at a glance, because it is the difference between two engines.
+    { name: "official-measures", active: isOfficialRoutingConfigured(env) },
   ];
 }
 

backend-ts/src/routes/cases.ts

@@ -20,7 +20,7 @@ import type { CloudDatabase, CloudBucket } from "@mieweb/cloud";
 import { getStores } from "../stores/factory.ts";
 import type { CaseStore } from "../stores/case-store.ts";
 import type { OutcomeStore } from "../stores/outcome-store.ts";
-import { engineForEnv } from "../wiring/engine-factory.ts";
+import { routedEngineForEnv } from "../wiring/executor-router.ts";
 import { toCaseSummary, type CaseSummary } from "../case/case-read-models.ts";
 import { bucketPeriodForMeasure } from "../run/compliance-period.ts";
 import { toCaseDetail } from "../case/case-detail-read-model.ts";
@@ -86,7 +86,7 @@ async function actionDeps(env: CasesEnv): Promise<CaseActionDeps & { channels: (
 }
 async function rerunDeps(env: CasesEnv): Promise<RerunDeps> {
   const s = await getStores(env);
-  const engine = await engineForEnv(env);
+  const engine = await routedEngineForEnv(env);
   return { cases: s.cases, events: s.events, outcomes: s.outcomes, runStore: s.runs, engine };
 }
 async function evidenceDeps(env: CasesEnv): Promise<EvidenceDeps> {

backend-ts/src/routes/compliance-simulation.ts

@@ -5,7 +5,7 @@
  * (all roles, like the immunization forecast). Writes nothing; no schema. handleEmployees only matches
  * `/profile` + `/search`, so this `/simulate` path is not intercepted.
  */
-import { engineForEnv } from "../wiring/engine-factory.ts";
+import { routedEngineForEnv } from "../wiring/executor-router.ts";
 import type { StoresEnv } from "../stores/factory.ts";
 import { parseQueryDate, QueryDateError } from "./query-dates.ts";
 import { simulateComplianceAsOf } from "../run/employee-compliance-snapshot.ts";
@@ -35,7 +35,7 @@ export async function handleComplianceSimulation(req: Request, env: StoresEnv):
   }
 
   const today = new Date().toISOString().slice(0, 10);
-  const engine = await engineForEnv(env);
+  const engine = await routedEngineForEnv(env);
   const snapshot = await simulateComplianceAsOf(externalId, asOf ?? today, { engine, today });
   if (!snapshot) return json({ error: "not_found", externalId }, 404);
   return json(snapshot);

backend-ts/src/routes/measures.ts

@@ -21,7 +21,7 @@ import type { CloudDatabase } from "@mieweb/cloud";
 import { getStores } from "../stores/factory.ts";
 import type { MeasureStore } from "../stores/measure-store.ts";
 import type { ValueSetStore } from "../stores/value-set-store.ts";
-import { engineForEnv } from "../wiring/engine-factory.ts";
+import { routedEngineForEnv } from "../wiring/executor-router.ts";
 import { MEASURES } from "../engine/cql/measure-registry.ts";
 import { ELM_LIBRARIES } from "../engine/cql/elm/index.ts";
 import { compileCql, reconstructCql } from "../engine/cql/cql-translator.ts";
@@ -314,7 +314,7 @@ export async function handleMeasures(req: Request, env: MeasuresEnv, actor = "sy
       const body = (await req.json().catch(() => ({}))) as ImpactPreviewRequest;
       try {
         const s = await getStores(env);
-        const engine = await engineForEnv(env);
+        const engine = await routedEngineForEnv(env);
         const deps = { cases: s.cases, events: s.events, engine };
         return json(await previewImpact(deps, measure, body, actor));
       } catch (err) {
@@ -358,7 +358,7 @@ export async function handleMeasures(req: Request, env: MeasuresEnv, actor = "sy
       const bundle = (await req.json().catch(() => null)) as unknown;
       if (!bundle || typeof bundle !== "object") return json({ error: "invalid_bundle" }, 400);
       try {
-        const engine = await engineForEnv(env);
+        const engine = await routedEngineForEnv(env);
         const outcome = await engine.evaluate({ measureId: evalId, patientBundle: bundle, evaluationDate: url.searchParams.get("date") ?? undefined });
         return json(outcome);
       } catch (err) {

backend-ts/src/routes/runs.ts

@@ -25,7 +25,7 @@ import type { OutcomeStore } from "../stores/outcome-store.ts";
 import type { CaseStore } from "../stores/case-store.ts";
 import type { HydratedSegment } from "../stores/segment-store.ts";
 import { ensureSegmentSeed } from "../segment/segment-seed.ts";
-import { engineForEnv } from "../wiring/engine-factory.ts";
+import { routedEngineForEnv } from "../wiring/executor-router.ts";
 import { toRunListItemFromCounts, toRunSummaryFromCounts, toRunLogEntries, toRunOutcomeRows, matchesRunFilters, type RunFilters } from "../run/read-models.ts";
 import { recoverStuckRuns } from "../run/recover-stuck-runs.ts";
 import { resolveAlertChannels } from "../run/alert-channel.ts";
@@ -48,6 +48,7 @@ import { isVsacConfigured } from "../engine/cql/resolve-value-set-resolver.ts";
 import { rerunToVerify, UnsupportedCaseRerunError } from "../case/case-rerun.ts";
 import type { PopulationCounts } from "../fhir/measure-report.ts";
 import {
+  officialMembership,
   buildMeasureReportBundle,
   buildSummaryMeasureReportFromCounts,
   countPopulations,
@@ -153,13 +154,38 @@ const MAX_INDIVIDUAL_REPORT_SUBJECTS = 5000;
  * So official runs read rows, bounded by the same subject cap the individual/bundle reports use;
  * over the cap we refuse rather than emit a status-derived (wrong) regulatory artifact.
  */
+/**
+ * Did THIS run's outcomes come from the official executor? One row settles it: `evidence.official` is
+ * written only by that executor, and a run evaluates one measure with one engine. Bounded on purpose —
+ * this sits in front of a path that must stay O(1) for a 120k `seed:scale` run.
+ */
+async function runProducedOfficialEvidence(
+  os: Awaited<ReturnType<typeof outcomes>>,
+  runId: string,
+): Promise<boolean> {
+  const [first] = await os.listOutcomes(runId, { limit: 1 });
+  return officialMembership(first?.evidence) !== null;
+}
+
 async function aggregateCountsForRun(
   os: Awaited<ReturnType<typeof outcomes>>,
   runId: string,
   measureId: string,
   env: RunsEnv,
 ): Promise<{ counts: PopulationCounts } | { error: Response }> {
-  if (!isOfficialRouted(measureId, env as unknown as Record<string, unknown>)) {
+  // Provenance comes from the RUN, not from the current deployment flag (Codex P1). A run's outcomes
+  // were produced by whichever engine was configured *then*; consulting `WORKWELL_OFFICIAL_MEASURES`
+  // now means that turning the flag off — the documented rollback — silently reinterprets every
+  // historical official run through the status histogram, reversing cms122's numerator (its official
+  // numerator is poor control, and the workflow status inverts it). An export of a past run must not
+  // change meaning because of a config change made after it.
+  //
+  // The env flag is still consulted first, as a cheap way to skip a read for the overwhelmingly common
+  // case; when it is off, one bounded row settles it. `evidence.official` is written only by the
+  // official executor, and a run evaluates one measure with one engine, so a single row is decisive.
+  const routedNow = isOfficialRouted(measureId, env as unknown as Record<string, unknown>);
+  const official = routedNow || (await runProducedOfficialEvidence(os, runId));
+  if (!official) {
     return { counts: populationCountsFromStatus(await os.countOutcomesByStatus(runId), measureId) };
   }
   const total = (await os.countOutcomesByStatus(runId)).reduce((sum, c) => sum + c.count, 0);
@@ -287,7 +313,7 @@ export async function handleRuns(
   if (pathname === "/api/runs/manual" && req.method === "POST") {
     const body = (await req.json().catch(() => ({}))) as ManualRunRequest;
     body.triggeredBy = externalTriggeredBy(body.triggeredBy); // Fable M1: no forged seed:*/scheduler labels
-    const engine = await engineForEnv(env);
+    const engine = await routedEngineForEnv(env);
     const deps = {
       runStore: await store(env),
       outcomeStore: await outcomes(env),
@@ -320,7 +346,7 @@ export async function handleRuns(
   const rerunId = pathname.match(/^\/api\/runs\/([^/]+)\/rerun$/)?.[1];
   if (rerunId && req.method === "POST") {
     const runStore = await store(env);
-    const engine = await engineForEnv(env);
+    const engine = await routedEngineForEnv(env);
     // A CASE run reruns through rerun-to-verify (the case scope), reading the caseId
     // persisted in requested_scope — matches Java's rerunSameScope CASE branch. Other
     // scopes go through executeRerun.
@@ -415,10 +441,13 @@ export async function handleRuns(
     // period, then today, mirroring that default.
     const evaluationPeriod =
       body.evaluationDate ?? (run.requestedScope.evaluationDate as string | undefined) ?? new Date().toISOString().slice(0, 10);
+    // Build the engine BEFORE claiming the run. `routedEngineForEnv` validates official-measure
+    // configuration and can throw; doing that after markRunning would leave an orphaned RUNNING run for
+    // `recoverStuckRuns` to sweep, for a failure that has nothing to do with the run.
+    const engine = await routedEngineForEnv(env);
     // A run being processed must leave the QUEUED claim path so it isn't re-handed
     // to a worker (QUEUED → RUNNING; idempotent for already-running runs).
     await runStore.markRunning(evalId);
-    const engine = await engineForEnv(env);
     try {
       const result = await engine.evaluate({
         measureId: body.measureId,

backend-ts/src/server.ts

@@ -79,6 +79,12 @@ async function main(): Promise<void> {
     // #263 incremental evaluation opt-in — the nightly run must see the flag too, or a scheduled run
     // would always do a full evaluation while a manual run reuses.
     WORKWELL_INCREMENTAL_EVAL: process.env.WORKWELL_INCREMENTAL_EVAL,
+    // PR-7b, and the THIRD instance of the bug the two comments above describe. Without this the
+    // nightly ALL_PROGRAMS run — the one that actually populates /compliance, /programs and
+    // quality_snapshots — evaluates cms122 with the AUTHORED CQL while a manual run evaluates it
+    // officially: two engines, two answers for the same measure, latest-run-wins, with
+    // `official-measures=on` on the boot line the whole time.
+    WORKWELL_OFFICIAL_MEASURES: process.env.WORKWELL_OFFICIAL_MEASURES,
   };
   const schedulerInterval = setInterval(() => {
     void schedulerTick(schedulerEnv).catch((e: unknown) =>

backend-ts/src/standards/fqm-isolation.test.ts

@@ -191,8 +191,9 @@ const EXECUTOR_IMPORTERS_ALLOWLIST = [
   // until a measure is actually flipped.
   "wiring/official-executor-adapter.ts",
   // NOT run/cli/official-cases.ts — that shell reaches fqm only through standards/official-cases.ts,
-  // which is exactly the layering this list is meant to preserve.
-  // PR-7b adds "wiring/executor-router.ts" here — deliberately a conscious edit, not an open door.
+  // which is exactly the layering this list is meant to preserve. Same for PR-7b's
+  // `wiring/executor-router.ts`: it consumes the ADAPTER, so the package stays one hop away from the
+  // thing production routes through. Adding it here pre-emptively was wrong, and this guard said so.
 ];
 
 test("5/5 consumers: only the approved app layers may import @workwell/official-executor", () => {