test: always-run canary smoke set spanning physics modules

toolchain/mfc/test/case.py

@@ -157,14 +157,18 @@ class TestCase(case.Case):
     restart_check: bool = False
     kind: str = "golden"
     convergence_spec: Optional[dict] = None
+    canary: bool = False
 
-    def __init__(self, trace: str, mods: dict, ppn: int = None, override_tol: float = None, restart_check: bool = False, kind: str = "golden", convergence_spec: Optional[dict] = None) -> None:
+    def __init__(
+        self, trace: str, mods: dict, ppn: int = None, override_tol: float = None, restart_check: bool = False, kind: str = "golden", convergence_spec: Optional[dict] = None, canary: bool = False
+    ) -> None:
         self.trace = trace
         self.ppn = ppn or 1
         self.override_tol = override_tol
         self.restart_check = restart_check
         self.kind = kind
         self.convergence_spec = convergence_spec
+        self.canary = canary
         merge = {**BASE_CFG.copy(), **mods}
         merge = {key: val for key, val in merge.items() if val is not None}
         super().__init__(merge)
@@ -372,6 +376,7 @@ class TestCaseBuilder:
     restart_check: bool = False
     kind: str = "golden"
     convergence_spec: Optional[dict] = None
+    canary: bool = False
 
     def get_uuid(self) -> str:
         return trace_to_uuid(self.trace)
@@ -383,7 +388,7 @@ def to_case(self) -> TestCase:
         if self.kind == "convergence":
             # Convergence cases drive their own runs — the BASE_CFG mods/path
             # machinery is unused. Trace + spec are the only inputs.
-            return TestCase(self.trace, {}, self.ppn, self.override_tol, self.restart_check, kind=self.kind, convergence_spec=self.convergence_spec)
+            return TestCase(self.trace, {}, self.ppn, self.override_tol, self.restart_check, kind=self.kind, convergence_spec=self.convergence_spec, canary=self.canary)
 
         dictionary = {}
         if self.path:
@@ -404,7 +409,7 @@ def to_case(self) -> TestCase:
         if self.functor:
             self.functor(dictionary)
 
-        return TestCase(self.trace, dictionary, self.ppn, self.override_tol, self.restart_check)
+        return TestCase(self.trace, dictionary, self.ppn, self.override_tol, self.restart_check, canary=self.canary)
 
 
 @dataclasses.dataclass

toolchain/mfc/test/cases.py

@@ -259,6 +259,29 @@ def get_dimensions():
     return r
 
 
+# Always-run "canary" smoke set: one cheap, feature-dominant regression case per major
+# physics module. Tagged canary=True in list_cases() so coverage-based selection
+# (toolchain/mfc/test/coverage.py:select_tests) can never skip them on any lane -- a silent
+# regression that disables a feature then trips on every PR. Validated in list_cases(), so a
+# renamed/removed trace fails loudly instead of silently un-tagging the canary.
+_CANARY_TRACES = frozenset(
+    {
+        "1D -> 1 Fluid(s) -> Viscous",  # m_viscous (Newtonian, Re=1e-4)
+        "1D -> 1 Fluid(s) -> Non-Newtonian -> nn=0.5",  # m_hb_function (Herschel-Bulkley)
+        "2D -> 2 Fluid(s) -> capillary=T -> model_eqns=3",  # m_surface_tension
+        "1D -> Bubbles -> QBMM",  # m_qbmm / m_bubbles_EE
+        "2D -> Lagrange Bubbles -> One-way Coupling",  # m_bubbles_EL
+        "1D -> MHD -> HLLD",  # m_mhd / m_riemann_solver_hlld
+        "1D -> Hypoelasticity -> 1 Fluid(s)",  # m_hypoelastic
+        "1D -> Chemistry -> Perfect Reactor",  # chemistry
+        "2D -> 1 Fluid(s) -> IBM -> Circle -> slip",  # m_ibm
+        "1D -> Phase Change model 5 -> 2 Fluid(s) -> model equation -> 3",  # m_pressure_relaxation (6-eq)
+        "1D -> Acoustic Source -> Sine -> Frequency",  # m_acoustic_src
+        "1D -> Bodyforces",  # m_body_forces
+    }
+)
+
+
 def list_cases() -> typing.List[TestCaseBuilder]:
     stack, cases = CaseGeneratorStack(), []
 
@@ -2455,4 +2478,13 @@ def kernel_golden_tests():
     if l1 != l2:
         raise common.MFCException(f"list_cases: uuids aren't unique ({l1} cases but {l2} unique uuids)")
 
+    # Tag the always-run canary smoke set (see _CANARY_TRACES). Validate first so a renamed
+    # or removed trace is a loud error, not a silently dropped canary.
+    missing = _CANARY_TRACES - {case.trace for case in cases}
+    if missing:
+        raise common.MFCException(f"list_cases: canary trace(s) not found (renamed/removed?): {sorted(missing)}")
+    for case in cases:
+        if case.trace in _CANARY_TRACES:
+            case.canary = True
+
     return cases

toolchain/mfc/test/coverage.py

@@ -156,6 +156,9 @@ def select_tests(cases, coverage_map, changed_files):
     # Rungs 5-7: per-test.
     to_run, skipped = [], []
     for case in cases:
+        if getattr(case, "canary", False):  # canary smoke set: always run, never skipped
+            to_run.append(case)
+            continue
         key = case.coverage_key()
         cov = coverage_map.get(key)
         if not cov:  # rung 5: unmapped/new test, or empty (uncertain) coverage -> run

toolchain/mfc/test/test_coverage_unit.py

@@ -72,9 +72,10 @@ def test_ordinary_sim_module_does_not_force_all():
 
 
 class _Case:
-    def __init__(self, ph, params=None):
+    def __init__(self, ph, params=None, canary=False):
         self._ph = ph
         self.params = params or {}
+        self.canary = canary
 
     def coverage_key(self):
         return self._ph
@@ -123,6 +124,22 @@ def test_rung6_and_7_overlap_selects_subset():
     assert [c.coverage_key() for c in skip] == ["miss"]
 
 
+def test_canary_always_runs_even_when_its_coverage_misses():
+    # 'canary' and 'plain' have identical, non-overlapping coverage; only the canary must
+    # run. 'other' covers the changed file so rung 4 (run-all) does not fire and we reach
+    # the per-test rungs where the canary bypass is exercised.
+    cases = [_Case("canary", canary=True), _Case("plain"), _Case("other")]
+    cov = {
+        "canary": ["src/simulation/m_viscous.fpp"],
+        "plain": ["src/simulation/m_viscous.fpp"],
+        "other": ["src/simulation/m_rhs.fpp"],
+    }
+    run, skip, _ = select_tests(cases, cov, {"src/simulation/m_rhs.fpp"})
+    run_keys = {c.coverage_key() for c in run}
+    assert "canary" in run_keys and "other" in run_keys
+    assert [c.coverage_key() for c in skip] == ["plain"]
+
+
 def test_case_coverage_key_uses_full_params():
     from mfc.test.case import TestCase