LNG_CAP_POOLS on by default (deltasignal specs/045); fix CAPPED_IDS cache dependence (F6)

PIPELINE_SETTINGS.md

@@ -71,6 +71,7 @@ what NOT to touch. Overriding is for A/B benchmarking only.
 | `LNG_BOUNDARY_HIERARCHY` | `1` | decompose a root complex one `hasComponent` level at a time (deltasignal specs/030) |
 | `LNG_SET_POOL` | `1` | one OR node per set-valued catalyst or regulator (deltasignal specs/033) |
 | `LNG_POOL_ACTIVE_VIA` | `made_from` | how a pool's active form is found (deltasignal specs/039) |
+| `LNG_CAP_POOLS` | `1` | past the variant cap, one OR pool per set instead of an all-required bundle (deltasignal specs/045) |
 | `LNG_CATALYST_BUNDLE` | `0` | legacy; subsumed by `LNG_COMPLEX_AS_NODE` |
 | `LNG_HANDOFF_EDGES` | `0` | precedingEvent hand-off bridges (net-negative, off) |
 

src/env_flags.py

@@ -14,7 +14,7 @@
 BOOL_FLAGS = {
     "LNG_BOUNDARY_EXPANSION": True,
     "LNG_BOUNDARY_HIERARCHY": True,
-    "LNG_CAP_POOLS": False,
+    "LNG_CAP_POOLS": True,      # default since deltasignal specs/045 (RAF experimental +14, held-out 0)
     "LNG_CATALYST_BUNDLE": False,
     "LNG_COMPLEX_AS_NODE": True,
     "LNG_COMPOSITION_EDGES": False,

src/pathway_generator.py

@@ -23,7 +23,7 @@
     export_uuid_to_reactome_mapping,
 )
 from src.neo4j_connector import get_reaction_connections, get_reactome_release
-from src.reaction_generator import get_decomposed_uid_mapping, prime_entity_caches
+from src.reaction_generator import CAPPED_IDS, get_decomposed_uid_mapping, prime_entity_caches
 
 # Env vars that change what a generated network contains. Anything listed here
 # is part of the cache fingerprint, so flipping one invalidates stale caches
@@ -294,6 +294,12 @@ def generate_pathway_file(
     reaction_connections_file = cache_dir / "reaction_connections.csv"
     decomposed_uid_mapping_file = cache_dir / "decomposed_uid_mapping.csv"
     best_matches_file = cache_dir / "best_matches.csv"
+    # Reactions the variant cap bundled, for LNG_CAP_POOLS. Filled only while
+    # decomposing, so it is cached beside the decomposition and restored on
+    # reuse (code review F6: a cached rebuild silently lost every cap pool, and
+    # the set was never cleared, so ids leaked from one pathway to the next).
+    capped_ids_file = cache_dir / "capped_ids.txt"
+    CAPPED_IDS.clear()
 
     try:
         # Decide ONCE per pathway whether this cache dir may be reused. Without
@@ -318,13 +324,15 @@ def generate_pathway_file(
                 # Continue without caching
 
         # Load or generate decomposition and best matches
-        if cache_usable and os.path.exists(decomposed_uid_mapping_file) and os.path.exists(best_matches_file):
+        if (cache_usable and os.path.exists(decomposed_uid_mapping_file)
+                and os.path.exists(best_matches_file) and capped_ids_file.exists()):
             logger.info(f"Loading cached decomposition from {decomposed_uid_mapping_file}")
             decomposed_uid_mapping = pd.read_csv(
                 decomposed_uid_mapping_file,
                 dtype=decomposed_uid_mapping_column_types,  # type: ignore[arg-type]
             )
             best_matches = pd.read_csv(best_matches_file)
+            CAPPED_IDS.update(x for x in capped_ids_file.read_text().split() if x)
             # Reusing the cached decomposition skips get_decomposed_uid_mapping,
             # which is the only place the entity caches are cleared and primed.
             # Prime them here or this pathway inherits the previous pathway's
@@ -343,6 +351,7 @@ def generate_pathway_file(
             try:
                 decomposed_uid_mapping.to_csv(decomposed_uid_mapping_file, index=False)
                 best_matches.to_csv(best_matches_file, index=False)
+                capped_ids_file.write_text("".join(f"{x}\n" for x in sorted(CAPPED_IDS)))
                 logger.info(f"Cached decomposition to {decomposed_uid_mapping_file}")
                 # Stamp the cache we just PRODUCED with real provenance. Without
                 # a write here the mechanism only ever adopts: generate, change

tests/test_cap_pools.py

@@ -83,9 +83,12 @@ def test_a_leaf_another_input_also_maps_to_stays_direct(stub, monkeypatch):
 
 
 def test_off_and_no_leak_between_pathways(stub, monkeypatch):
-    monkeypatch.delenv("LNG_CAP_POOLS", raising=False)
+    monkeypatch.setenv("LNG_CAP_POOLS", "0")      # on by default since deltasignal specs/045
     resolve({A, K1, K2})
     assert m._vr_input_pools == {}
+    monkeypatch.delenv("LNG_CAP_POOLS", raising=False)
+    resolve({A, K1, K2})
+    assert m._vr_input_pools                       # the default pools
     monkeypatch.setenv("LNG_CAP_POOLS", "1")
     resolve({A, K1, K2})
     assert m._vr_input_pools

tests/test_cap_pools_cache.py

@@ -0,0 +1,76 @@
+"""CAPPED_IDS survives a cached rebuild and does not leak between pathways
+(code review F6; LNG_CAP_POOLS default since deltasignal specs/045).
+
+CAPPED_IDS is filled only while decomposing. A rebuild that reused the cached
+decomposition skipped that, so the network builder saw no capped reaction and
+emitted no cap pools; and the set was never cleared, so ids leaked from one
+pathway to the next. Driven through generate_pathway_file."""
+import pandas as pd
+import pytest
+
+import src.pathway_generator as pg
+import src.reaction_generator as rg
+from tests.test_export_failfast import mocked  # noqa: F401  (fixture)
+
+
+@pytest.fixture
+def seen(mocked):  # noqa: F811
+    record = []
+    mocked.setattr(pg, "_cache_fingerprint", lambda: {"fixed": 1})
+    mocked.setattr(pg, "_write_cache_fingerprint", lambda d, fp, provenance=None:
+                   (d / "fingerprint.json").write_text("{}"))
+    mocked.setattr(pg, "_cache_is_reusable", lambda d, fp: (d / "fingerprint.json").exists())
+    mocked.setattr(pg, "prime_entity_caches", lambda rc: None)
+    real = pg.create_pathway_logic_network
+
+    def spy(*a, **k):
+        record.append(set(rg.CAPPED_IDS))
+        return real(*a, **k)
+    mocked.setattr(pg, "create_pathway_logic_network", spy)
+    return mocked, record
+
+
+def decompose_capping(ids):
+    def f(pid, rc):
+        rg.CAPPED_IDS.update(ids)
+        return pd.DataFrame({"uid": ["u"]}), [("a", "b", "R-HSA-1")]
+    return f
+
+
+def test_a_cached_rebuild_restores_the_capped_ids(tmp_path, seen):
+    mp, record = seen
+    mp.setattr(pg, "get_decomposed_uid_mapping", decompose_capping({"R-HSA-CAP"}))
+    pg.generate_pathway_file("R-HSA-1", "x", str(tmp_path))
+    assert (tmp_path / "R-HSA-1" / "cache" / "capped_ids.txt").read_text() == "R-HSA-CAP\n"
+
+    def must_not_decompose(pid, rc):
+        raise AssertionError("the cache should have been reused")
+    mp.setattr(pg, "get_decomposed_uid_mapping", must_not_decompose)
+    rg.CAPPED_IDS.clear()
+    pg.generate_pathway_file("R-HSA-1", "x", str(tmp_path))
+    assert record == [{"R-HSA-CAP"}, {"R-HSA-CAP"}]
+
+
+def test_capped_ids_do_not_leak_between_pathways(tmp_path, seen):
+    mp, record = seen
+    mp.setattr(pg, "get_decomposed_uid_mapping", decompose_capping({"R-HSA-CAP"}))
+    pg.generate_pathway_file("R-HSA-1", "x", str(tmp_path))
+    mp.setattr(pg, "get_decomposed_uid_mapping", decompose_capping(set()))
+    pg.generate_pathway_file("R-HSA-2", "y", str(tmp_path))
+    assert record == [{"R-HSA-CAP"}, set()]
+
+
+def test_a_cache_without_the_capped_ids_file_is_not_reused(tmp_path, seen):
+    mp, record = seen
+    mp.setattr(pg, "get_decomposed_uid_mapping", decompose_capping({"R-HSA-CAP"}))
+    pg.generate_pathway_file("R-HSA-1", "x", str(tmp_path))
+    (tmp_path / "R-HSA-1" / "cache" / "capped_ids.txt").unlink()
+    calls = []
+
+    def again(pid, rc):
+        calls.append(1)
+        rg.CAPPED_IDS.update({"R-HSA-CAP"})
+        return pd.DataFrame({"uid": ["u"]}), [("a", "b", "R-HSA-1")]
+    mp.setattr(pg, "get_decomposed_uid_mapping", again)
+    pg.generate_pathway_file("R-HSA-1", "x", str(tmp_path))
+    assert calls == [1] and record[-1] == {"R-HSA-CAP"}