feat(totlib): activate eq->tr BPSD verify rule on mono (#208 Phase 2c PR-B)

.github/workflows/python-tests.yml

@@ -586,3 +586,26 @@ jobs:
         run: |
           python -m pytest python/totlib/tests/test_mono_routing.py \
                  --forked --timeout=120 --timeout-method=signal -v
+
+      - name: Mono pipeline coupling (Phase 2c PR-B)
+        # #208 PR-B — drives eq -> tr through TotPipeline.run_pipeline
+        # on the mono image and asserts the ("eq","tr") verify rule
+        # fires (TestEqToTrPipelineCoupling). Also exercises the
+        # negative case via TestEqToTrRuleDormantOnDefault — that
+        # test skip-gates on the per-module .so files at the repo
+        # default paths (eq/libeqapi.so, tr/libtrapi.so,
+        # tot/libtotapi.so), which the per-module build steps
+        # earlier in this job have already produced.
+        #
+        # TOTLIB_PATH is set so the totlib wrapper picks the
+        # default libtotapi.so from the workspace consistently with
+        # other tests in this job. eqlib / trlib do NOT honor
+        # TOTLIB_PATH; they resolve via the repo default at
+        # workspace/eq/libeqapi.so etc.
+        env:
+          MONO_LIB_PATH: ${{ github.workspace }}/tot/libtotapi_mono.so
+          TOTLIB_PATH: ${{ github.workspace }}/tot/libtotapi.so
+          PYTHONPATH: python
+        run: |
+          python -m pytest python/totlib/tests/test_mono_pipeline_coupling.py \
+                 --forked --timeout=120 --timeout-method=signal -v

python/totlib/README.md

@@ -433,3 +433,12 @@ Explicit `lib_path=...` to a wrapper constructor still wins
 
 The behavioral eq→tr BPSD coupling rule that this routing enables
 is activated in PR-B (issue #208 follow-up).
+
+**Phase 2c PR-B (#208 follow-up)**: the `("eq","tr")` BPSD coupling
+rule is now active in `_MONO_ONLY_RULES`. When `MONO_LIB_PATH` is
+set, `TotPipeline.run_pipeline([("eq", ...), ("tr", ...)])` runs
+the rule via `Trlib.check_bpsd_pull()` and reports it in
+`PipelineResult.last("tr").coupling_applied`. On the default
+per-module image the rule stays dormant — eq and tr each have
+their own private bpsd storage so cross-module verification is
+meaningless there.

python/totlib/pipeline.py

@@ -292,42 +292,54 @@ def compute_rjt_volint(state, *, R0: float, a: float) -> float:
 
 
 # ------------------------------------------------------------------
-# Mono-only coupling rules (L-7b-ii Phase 2b infrastructure)
+# Mono-only coupling rules
 # ------------------------------------------------------------------
-# This dict is the home for coupling rules that are activated ONLY
-# when the loaded libtotapi*.so is the monolithic build
-# (tot_is_mono() == 1). On the default per-module .so path, BPSD
-# broker storage is private to each per-module .so, so cross-module
-# BPSD coupling cannot be verified meaningfully and these rules stay
-# dormant.
+# Rules in this dict are activated ONLY when the loaded
+# libtotapi*.so is the monolithic build (tot_is_mono() == 1). On
+# the default per-module .so path each per-module library has
+# private bpsd storage, so cross-module BPSD verification is
+# meaningless there and these rules stay dormant via the
+# _detect_mono() / _build_active_rules() overlay mechanism that
+# Phase 2b put in place.
 #
-# Activation mechanism: TotPipeline._build_active_rules() overlays
-# these onto the instance's _active_rules dict if _detect_mono()
-# returns True. The legacy module-level COUPLING_RULES dict above is
-# NOT mutated.
+# Phase 2c PR-A (#212) plumbed the wrappers to honor MONO_LIB_PATH
+# so all 6 wrappers can route to the same mono image. PR-B (this
+# commit) activates the ("eq","tr") rule that PR-A's plumbing
+# enables.
 #
-# Phase 2b status (2026-05-18): the OVERLAY MECHANISM is in place
-# (verified by Layer-C tests in test_mono_bpsd_smoke.py), but no
-# actual rule is populated here yet — populating ("eq","tr") today
-# would cause run_pipeline([("eq",...),("tr",...)]) to fire the
-# verify rule against TotPipeline's Eqlib/Trlib wrappers, which
-# still load their OWN per-module lib<mod>api.so files (NOT the
-# mono image). The verify would fail because eq pushed to one
-# private bpsd and tr reads from a different private bpsd.
-#
-# Phase 2c will plumb the wrappers to honor a unified mono path
-# (e.g. MONO_LIB_PATH env var honored by Eqlib/Trlib/Fplib/Tilib/
-# Wrxlib's _ffi.py), at which point activating ("eq","tr") here
-# becomes meaningful. Until then this dict stays empty so the
-# detection infrastructure can land safely.
-#
-# Codex retrospective 2026-05-18 (review of Phase 2b PR) caught
-# that prematurely populating this dict would create a user-facing
-# regression for mono users running multi-module pipelines.
+# Adding a new rule here?
+#   1. Define a named module-level callable (not a lambda) so
+#      pipeline.run_pipeline error messages show a meaningful
+#      __qualname__ when the verify fails.
+#   2. Verify the timing is safe at the rule-firing phase
+#      (run_pipeline fires verify rules AFTER current-module init
+#      but BEFORE current-module run — see lines 610-631).
+#   3. Add a Layer-D integration test covering both the happy
+#      mono path AND the dormant non-mono path (see
+#      python/totlib/tests/test_mono_pipeline_coupling.py for
+#      the ("eq","tr") example).
+
+def _eq_to_tr_bpsd_check(trlib_inst) -> bool:
+    """Verify eq's BPSD push is visible to tr (mono image only).
+
+    Returns True iff Trlib.check_bpsd_pull() finds the device /
+    equ1D / metric1D slots that eq pushed during eq_run(1). On the
+    default per-module image this would always return False because
+    each per-module .so has private bpsd storage — but the rule
+    only activates when _detect_mono() == True so the path is never
+    exercised against the default image.
+    """
+    return trlib_inst.check_bpsd_pull()
+
 
 _MONO_ONLY_RULES: Dict[Tuple[str, str], List[CouplingRule]] = {
-    # ("eq", "tr"): populated in #201 Phase 2c after Eqlib/Trlib mono
-    # routing is implemented.
+    ("eq", "tr"): [
+        CouplingRule(
+            kind="verify",
+            verify=_eq_to_tr_bpsd_check,
+            doc="eq -> tr BPSD broker round-trip (mono image only)",
+        ),
+    ],
 }
 
 

python/totlib/tests/test_mono_pipeline_coupling.py

@@ -0,0 +1,187 @@
+"""Layer-D — TotPipeline rule firing on the mono image.
+
+Phase 2c PR-B (#208). Drives eq -> tr through
+TotPipeline.run_pipeline on the mono image and asserts the
+("eq","tr") verify rule fires (mono) / stays dormant (default).
+
+Process isolation: pytest.mark.forked module-level. The mono .so
+carries module-level state across calls; without forking,
+tot/eq/tr init state from a prior test could leak into this
+test's pipeline.
+
+Spec: docs/superpowers/specs/2026-05-26-l7b-ii-phase-2c-prb-rule-activation-design.md
+"""
+from __future__ import annotations
+
+import os
+import shutil
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+import pytest
+
+pytestmark = [pytest.mark.forked]
+
+HERE = Path(__file__).resolve()
+REPO = HERE.parents[3]
+PYTHON_ROOT = REPO / "python"
+if str(PYTHON_ROOT) not in sys.path:
+    sys.path.insert(0, str(PYTHON_ROOT))
+
+_EQDATA_FIXTURE = (
+    REPO / "python" / "totlib" / "tests" / "fixtures" / "eqdata-HT6M"
+)
+
+
+def _mono_path() -> str:
+    p = os.environ.get("MONO_LIB_PATH", "")
+    return p if p and os.path.exists(p) else ""
+
+
+@unittest.skipUnless(
+    _mono_path() and _EQDATA_FIXTURE.exists(),
+    "MONO_LIB_PATH and/or eqdata-HT6M fixture missing; "
+    "build with `make -C tot libtotapi_mono.so` and ensure PR #199 "
+    "fixture is in place",
+)
+class TestEqToTrPipelineCoupling(unittest.TestCase):
+    """Layer-D happy path: pipeline runs eq -> tr on mono, the
+    ("eq","tr") verify rule fires, broker round-trip succeeds.
+    """
+
+    def test_eq_to_tr_bpsd_pipeline_succeeds_on_mono(self):
+        import _runtime_mode
+        from totlib import TotPipeline
+
+        _runtime_mode.mono_lib_path.cache_clear()
+        result = None
+        with tempfile.TemporaryDirectory(prefix="prb_") as td:
+            shutil.copy2(_EQDATA_FIXTURE, Path(td) / "eqdata-HT6M")
+            prev = os.getcwd()
+            os.chdir(td)
+            try:
+                with TotPipeline() as p:
+                    p.set_param("eq:MODELG", 3.0)
+                    p.set_param("eq:KNAMEQ", "eqdata-HT6M")
+                    result = p.run_pipeline([
+                        ("eq", {"mode": 1}),
+                        ("tr", {"ntmax": 1}),
+                    ])
+            finally:
+                os.chdir(prev)
+
+        # Assertions OUTSIDE the `with TotPipeline()` block (Codex
+        # 2026-05-26 LOW-5: a __exit__ failure should not be able
+        # to mask the assertion).
+        self.assertIsNotNone(result, "run_pipeline returned None")
+        tr_step = result.last("tr")
+        self.assertIn(
+            "eq -> tr BPSD broker round-trip",
+            " ".join(tr_step.coupling_applied),
+            f"verify rule did not fire; coupling_applied="
+            f"{tr_step.coupling_applied}",
+        )
+
+
+def _default_per_module_so_present() -> bool:
+    """Return True iff every per-module .so the negative test
+    actually loads via the resolution chain exists at its default
+    repo path. eqlib/trlib/totlib wrappers each look in
+    <repo>/<mod>/lib<mod>api.so (priority 2 in PR-A's spec §D-5),
+    so we check those directly. Per Codex 2026-05-26 spec round-3
+    review, TOTLIB_PATH is NOT included here: eqlib and trlib do
+    not honor it (they use EQLIB_PATH / TRLIB_PATH respectively),
+    so gating on it would either over-skip (false-negative on
+    TOTLIB_PATH-only setups) or under-cover (false-positive when
+    eq/tr .so are missing).
+    """
+    return all(
+        (REPO / mod / f"lib{name}.so").exists()
+        for mod, name in (
+            ("eq", "eqapi"),
+            ("tr", "trapi"),
+            ("tot", "totapi"),
+        )
+    )
+
+
+@unittest.skipUnless(
+    _default_per_module_so_present() and _EQDATA_FIXTURE.exists(),
+    "per-module .so (eq/libeqapi.so, tr/libtrapi.so, "
+    "tot/libtotapi.so) and/or eqdata-HT6M fixture missing; "
+    "build via setup.sh or `make -C <mod> lib<mod>api.so`",
+)
+class TestEqToTrRuleDormantOnDefault(unittest.TestCase):
+    """Layer-D negative path: rule must NOT fire on the default
+    per-module image even though _MONO_ONLY_RULES is now populated.
+
+    Pinned per Codex 2026-05-26 design review MED-6: if a future
+    refactor of _detect_mono() or _build_active_rules() silently
+    activates the overlay on non-mono, this test catches it.
+    """
+
+    def test_eq_to_tr_rule_dormant_on_default(self):
+        import _runtime_mode
+        from totlib import TotPipeline
+
+        # Unset MONO_LIB_PATH so the wrappers route to per-module
+        # .so files via their priority-1 env vars (EQLIB_PATH /
+        # TRLIB_PATH / TOTLIB_PATH) or priority-2 repo defaults.
+        # _detect_mono() reads tot_is_mono() from the totlib
+        # wrapper's loaded image, which is the default
+        # libtotapi.so, returning 0 — so the overlay stays inactive.
+        # Defensive: also pop the per-module env vars (Codex 2026-05-26
+        # cumulative review LOW-1). EQLIB_PATH / TRLIB_PATH would route
+        # eqlib / trlib away from the repo-default per-module .so files
+        # we just asserted exist, defeating the test's intent.
+        original_mono = os.environ.pop("MONO_LIB_PATH", None)
+        original_eqlib = os.environ.pop("EQLIB_PATH", None)
+        original_trlib = os.environ.pop("TRLIB_PATH", None)
+        _runtime_mode.mono_lib_path.cache_clear()
+
+        result = None
+        try:
+            with tempfile.TemporaryDirectory(prefix="prb_neg_") as td:
+                shutil.copy2(_EQDATA_FIXTURE, Path(td) / "eqdata-HT6M")
+                prev = os.getcwd()
+                os.chdir(td)
+                try:
+                    with TotPipeline() as p:
+                        p.set_param("eq:MODELG", 3.0)
+                        p.set_param("eq:KNAMEQ", "eqdata-HT6M")
+                        # IMPORTANT: this is expected to SUCCEED.
+                        # On default per-module .so, eq's BPSD push
+                        # lands in libeqapi's private storage; tr
+                        # reads from libtrapi's private storage. The
+                        # rule MUST stay dormant or the pipeline
+                        # would fail for the wrong reason (the
+                        # verify would return False).
+                        result = p.run_pipeline([
+                            ("eq", {"mode": 1}),
+                            ("tr", {"ntmax": 1}),
+                        ])
+                finally:
+                    os.chdir(prev)
+        finally:
+            if original_mono is not None:
+                os.environ["MONO_LIB_PATH"] = original_mono
+            if original_eqlib is not None:
+                os.environ["EQLIB_PATH"] = original_eqlib
+            if original_trlib is not None:
+                os.environ["TRLIB_PATH"] = original_trlib
+            _runtime_mode.mono_lib_path.cache_clear()
+
+        self.assertIsNotNone(result, "run_pipeline returned None")
+        tr_step = result.last("tr")
+        self.assertNotIn(
+            "eq -> tr BPSD broker round-trip",
+            " ".join(tr_step.coupling_applied),
+            f"rule should NOT fire on default per-module image; "
+            f"coupling_applied={tr_step.coupling_applied}",
+        )
+
+
+if __name__ == "__main__":  # pragma: no cover
+    unittest.main()

python/totlib/tests/test_pipeline_verify.py

@@ -83,6 +83,10 @@ def test_verify_true_records_rule_in_applied(patch_wrappers, monkeypatch):
         ],
     }
     monkeypatch.setattr("totlib.pipeline.COUPLING_RULES", rules)
+    # Suppress mono-only rules so this unit test is independent of
+    # MONO_LIB_PATH in the environment. The mono overlay is covered
+    # by test_mono_pipeline_coupling.py (Phase 2c PR-B).
+    monkeypatch.setattr("totlib.pipeline._MONO_ONLY_RULES", {})
     pipe = TotPipeline()
     result = pipe.run_pipeline([("eq", {}), ("tr", {})])
     tr_step = result.last("tr")
@@ -168,6 +172,15 @@ def test_mixed_transfer_then_verify_fires_in_order(
         ],
     }
     monkeypatch.setattr("totlib.pipeline.COUPLING_RULES", rules)
+    # Patch _MONO_ONLY_RULES to empty dict (Codex 2026-05-26 cumulative
+    # review LOW-2). Under MONO_LIB_PATH, the real ("eq","tr") mono rule
+    # would silently fire alongside our mock rules and break the exact-order
+    # assertion. Same isolation that test_verify_true_records_rule_in_applied
+    # uses in Task 1.
+    monkeypatch.setattr(
+        "totlib.pipeline._MONO_ONLY_RULES",
+        {},
+    )
     pipe = TotPipeline()
     pipe.run_pipeline([("eq", {}), ("tr", {})])
     assert fired == ["transfer", "verify"]