Review 2026-10-22: pool claims, forward intermediates, mutual specialization pairs, provenance, tests

PIPELINE_SETTINGS.md

@@ -75,12 +75,17 @@ what NOT to touch. Overriding is for A/B benchmarking only.
 | `LNG_RECYCLE_POOLS` | `1` | write a modified protein's return to the complex it starts in as a pool with several resting states, plus a structural `recycle` edge from the free copy to the starting copy in `boundary_edges.csv` (p130/p107 in Mitotic G1; deltasignal specs/081) |
 | `LNG_RELEASE_POOLS` | `1` | a protein bound, modified and released by a complex, whose released copy carries on (beta-catenin and the destruction complex), as a free/bound pool whose released copy reads the free state; writes `released_twin` and `recycle` edges (deltasignal specs/086) |
 | `LNG_UB_LIGASE_UNIT` | `1` | a ubiquitination does not deplete its own ligase: a leaf of the catalyst's curated `activeUnit`, or of a complex released unchanged, that leaves the reaction unmodified is not the substrate (CKS1B under SCF(Skp2); deltasignal specs/089) |
+| `LNG_SPECIALIZATION_BRIDGE` | `add` | join a generic complex of sets and the specific member combination Reactome curates under another id, ordered by `precedingEvent` (`specialization` OR edges; `replace` \| `off`; deltasignal specs/067) |
+| `LNG_SPECIALIZATION_CLOSED` | `twin` | a specialization in-edge does not open a carrier loop (`off`; deltasignal specs/068) |
 | `LNG_CATALYST_BUNDLE` | `0` | legacy; subsumed by `LNG_COMPLEX_AS_NODE` |
 | `LNG_HANDOFF_EDGES` | `0` | precedingEvent hand-off bridges (net-negative, off) |
 
 A switch takes exactly `0` or `1`; any other value, or an `LNG_*` name the
-generator does not read (a typo), stops the run at startup. The full list and
-defaults are `BOOL_FLAGS` in `src/env_flags.py`.
+generator does not read (a typo), stops the run at startup. The boolean
+switches and their defaults are `BOOL_FLAGS` in `src/env_flags.py`; the
+switches with other values are listed in `OTHER_FLAGS` there, and their
+defaults are in the modules that read them (`src/specialization.py` for the two
+above).
 
 **Solver (DeltaSignal):** the defaults live in the deltasignal repo's
 CLAUDE.md ("Solver Configuration"), which CI keeps in step with the code; they

src/logic_network_generator.py

@@ -4660,7 +4660,8 @@ def find_recycle_pools(pathway_logic_network: pd.DataFrame, reaction_id_map: pd.
     the edges as ``(source_uuid, target_uuid)``."""
     from src.variant_keys import variant_parts
     stats = stats if stats is not None else {}
-    for k in ("recycle_pools", "recycle_states", "recycle_paths", "recycle_edges", "recycle_claimed"):
+    for k in ("recycle_pools", "recycle_states", "recycle_paths", "recycle_edges", "recycle_claimed",
+              "recycle_no_copies"):
         stats[k] = 0
     claimed = claimed or set()
     if pathway_logic_network.empty:
@@ -4758,16 +4759,24 @@ def copies(a: str, b: str, rx: str) -> List[str]:
         for c, (st, k) in sorted(g["cs"].items()):
             fwd = st[:k + 1]
             back = st[k + 1:]
+            # forms between C and M (km >= 1) are pool-written too (review 2026-10-22)
+            nodes |= {y for _, _, y in fwd[:-1]}
             nodes |= {y for _, _, y in back[:-1]}
             fwd_steps = [(a, y, copies(a, y, rx)) for a, rx, y in fwd]
             back_steps = [(a, y, copies(a, y, rx)) for a, rx, y in back]
             a_last, _, _ = back[-1]
             back_steps[-1] = (a_last, c, back_steps[-1][2])        # the return lands in C
             rows.append((c, fwd_steps, back_steps, [rx for _, rx, _ in st]))
         step_nodes = {u for _, fs, bs, _ in rows for _, _, cs in fs + bs for u in cs}
-        if (nodes | step_nodes) & claimed or any(not cs for _, fs, bs, _ in rows for _, _, cs in fs + bs):
+        if (nodes | step_nodes) & claimed:
             stats["recycle_claimed"] += 1
             continue
+        if any(not cs for _, fs, bs, _ in rows for _, _, cs in fs + bs):
+            stats["recycle_no_copies"] += 1
+            continue
+        # an accepted pool claims its nodes, so a later candidate cannot share them
+        # (review 2026-10-22; find_release_pools already did this)
+        claimed |= nodes | step_nodes
         n += 1
         pid = f"recycle{n}"
         for c in sorted(g["cs"]):
@@ -5006,7 +5015,8 @@ def export_pools(pathway_id: str, pathway_logic_network: pd.DataFrame, reaction_
             add.reindex(columns=cols).to_csv(boundary_file, mode="a", header=False, index=False)
         logger.info(f"Recycle pools (specs/081): {rstats['recycle_pools']} pools, {rstats['recycle_states']} states, "
                     f"{rstats['recycle_paths']} paths, {rstats['recycle_edges']} recycle edges, "
-                    f"{rstats['recycle_claimed']} skipped (nodes claimed by a pool)")
+                    f"{rstats['recycle_claimed']} skipped (nodes claimed by a pool), "
+                    f"{rstats['recycle_no_copies']} skipped (no reaction copies)")
     if env_flag("LNG_RELEASE_POOLS"):
         # deltasignal specs/086: after the ordinary and recycle pools, which keep their nodes
         claimed = {f[1] for f in forms} | {t[5] for t in transitions}

src/pathway_generator.py

@@ -56,6 +56,9 @@
     "LNG_DIAGRAM_SUBPATHWAYS",
     "LNG_SPECIALIZATION_BRIDGE",
     "LNG_SPECIALIZATION_CLOSED",
+    "LNG_RECYCLE_POOLS",
+    "LNG_RELEASE_POOLS",
+    "LNG_UB_LIGASE_UNIT",
     # Of these three, only LNG_EMIT_ONE_SIDED is a genuine cache gap: it is
     # read in reaction_generator.decompose_by_reactions, so it changes
     # best_matches.csv / decomposed_uid_mapping.csv, which ARE cached.

src/specialization.py

@@ -130,7 +130,9 @@ def linked_pairs(reactions: Sequence[Tuple[str, Set[str], Set[str], Set[str], Se
     preceding reactions). A pair qualifies when one complex specializes the
     other and some reaction consuming the second lists some reaction producing
     the first as ``precedingEvent``: the curators' own statement of order, so
-    a bridge never runs against it (and a mutual pair is not welded)."""
+    a bridge never runs against it. A mutual pair (two complexes with identical
+    components, each preceding the other) would be a positive 2-cycle, so it is
+    not bridged at all (review 2026-10-22; none occurs in Release 97)."""
     produced: Dict[str, Set[str]] = defaultdict(set)
     consumed: Dict[str, Set[str]] = defaultdict(set)
     preceding: Dict[str, Set[str]] = {}
@@ -149,7 +151,8 @@ def linked_pairs(reactions: Sequence[Tuple[str, Set[str], Set[str], Set[str], Se
                 continue
             if any(p in preceding[c] for c in consumed[b] for p in produced[a]):
                 out.append((a, b))
-    return out
+    pairs = set(out)
+    return [(a, b) for a, b in out if (b, a) not in pairs]
 
 
 def copy_bridges(pairs: Sequence[Tuple[str, str]],

tests/test_recycle_pools.py

@@ -8,7 +8,8 @@
 
 import src.logic_network_generator as m
 
-NAMES = ["s", "e2f", "cx1", "cx2", "kin", "vk1", "vk2", "pm", "pp", "vp", "f", "w", "vw", "y", "vy", "q"]
+NAMES = ["s", "e2f", "cx1", "cx2", "kin", "vk1", "vk2", "pm", "pp", "vp", "f", "w", "vw", "y", "vy", "q",
+         "vp2", "f2", "x", "vk3"]
 U = {k: f"00000000-0000-0000-0000-{i:012d}" for i, k in enumerate(NAMES, 1)}
 RBL2, E2F, KIN, PP, W = 1, 2, 3, 4, 5
 
@@ -28,15 +29,18 @@ def p130(extra=(), m_sig=(("res", "p-S672"),)):
              e("cx2", "vk2", "input"), e("kin", "vk2", "catalyst"), e("vk2", "pm", "output"),
              e("pm", "vp", "input"), e("pp", "vp", "catalyst"), e("vp", "f", "output")] + list(extra)
     net = pd.DataFrame(edges)
-    rmap = pd.DataFrame({"uid": [U[k] for k in ("vk1", "vk2", "vp", "vw", "vy")],
-                         "reactome_id": ["R-HSA-K", "R-HSA-K", "R-HSA-P", "R-HSA-W", "R-HSA-Y"]})
+    rmap = pd.DataFrame({"uid": [U[k] for k in ("vk1", "vk2", "vp", "vw", "vy", "vp2", "vk3")],
+                         "reactome_id": ["R-HSA-K", "R-HSA-K", "R-HSA-P", "R-HSA-W", "R-HSA-Y", "R-HSA-P2",
+                                         "R-HSA-K3"]})
     umap = {U["s"]: "R-HSA-RBL2", U["f"]: "R-HSA-RBL2", U["e2f"]: "R-HSA-E2F",
             U["cx1"]: "R-HSA-CX1", U["cx2"]: "R-HSA-CX2", U["pm"]: "R-HSA-PM",
-            U["kin"]: "R-HSA-KIN", U["pp"]: "R-HSA-PP", U["w"]: "R-HSA-W", U["y"]: "R-HSA-Y", U["q"]: "R-HSA-Q"}
+            U["kin"]: "R-HSA-KIN", U["pp"]: "R-HSA-PP", U["w"]: "R-HSA-W", U["y"]: "R-HSA-Y", U["q"]: "R-HSA-Q",
+            U["f2"]: "R-HSA-RBL2", U["x"]: "R-HSA-X"}
     profiles = {"R-HSA-RBL2": prof([RBL2]), "R-HSA-E2F": prof([E2F]),
                 "R-HSA-CX1": prof([RBL2, E2F]), "R-HSA-CX2": prof([RBL2, E2F]),
                 "R-HSA-PM": prof([RBL2], m_sig), "R-HSA-KIN": prof([KIN]), "R-HSA-PP": prof([PP]),
-                "R-HSA-W": prof([W]), "R-HSA-Y": prof([RBL2, W]), "R-HSA-Q": prof([RBL2])}
+                "R-HSA-W": prof([W]), "R-HSA-Y": prof([RBL2, W]), "R-HSA-Q": prof([RBL2]),
+                "R-HSA-X": prof([RBL2])}
     return net, rmap, umap, profiles
 
 
@@ -83,3 +87,35 @@ def test_the_flag_is_on_by_default():
     # default since deltasignal specs/081 was adopted (Adam, 2026-10-20)
     from src.env_flags import BOOL_FLAGS
     assert BOOL_FLAGS["LNG_RECYCLE_POOLS"] is True
+
+
+def test_two_free_copies_through_one_modified_form_give_one_pool():
+    # review 2026-10-22: a second dephosphorylation copy pm -> vp2 -> f2 made a
+    # second pool over the same cx1, cx2 and pm (accepted pools did not claim)
+    st = {}
+    net, rmap, umap, profiles = p130(extra=[e("pm", "vp2", "input"), e("pp", "vp2", "catalyst"),
+                                            e("vp2", "f2", "output")])
+    forms, trans, edges = m.find_recycle_pools(net, rmap, umap, profiles, stats=st)
+    assert {f[0] for f in forms} == {"recycle1"}
+    assert st["recycle_pools"] == 1 and st["recycle_claimed"] == 1
+    assert len({f[1] for f in forms}) == len(forms)
+
+
+def test_a_form_between_the_complex_and_the_modified_form_is_listed():
+    # review 2026-10-22: cx1 -> x (unmodified) -> pm; x is pool-written, so it is
+    # listed as an intermediate and claimed
+    net = pd.DataFrame([r for r in p130()[0].to_dict("records")
+                        if not (r["source_id"] == U["vk1"] and r["target_id"] == U["pm"])]
+                       + [e("vk1", "x", "output"), e("x", "vk3", "input"), e("kin", "vk3", "catalyst"),
+                          e("vk3", "pm", "output")])
+    _, rmap, umap, profiles = p130()
+    forms, trans, _ = m.find_recycle_pools(net, rmap, umap, profiles)
+    assert (U["x"], "intermediate") in {(u, role) for _, u, _, role, _ in forms}
+    assert any(t[3] == U["x"] for t in trans)
+
+
+def test_no_copies_is_counted_apart_from_claimed():
+    # the log names each cause (review 2026-10-22)
+    st = {}
+    m.find_recycle_pools(*p130(), stats=st)
+    assert st["recycle_no_copies"] == 0 and st["recycle_claimed"] == 0

tests/test_specialization.py

@@ -196,3 +196,11 @@ def test_pipeline_step_applies_the_bridges(monkeypatch):
     out = pg.apply_specialization_step(_result(), "R-HSA-1")
     assert len(out.logic_network) == 6 and out.boundary_mask == MASK + [False]
     assert out.logic_network.iloc[-1].edge_type == "specialization"
+
+
+def test_a_mutual_pair_is_not_bridged(structure):
+    # review 2026-10-22: each complex preceding the other would weld a positive
+    # 2-cycle, which the docstring already ruled out
+    make_g = ("makeG", {"setP", "setR"}, {"G"}, set(), set())
+    uses_s = ("useS", {"S", "ATP"}, {"pS"}, set(), {"makeG"})
+    assert sp.linked_pairs([AUTOPHOS, GRB2_BINDS, make_g, uses_s], structure) == []

tests/test_ub_ligase_unit.py

@@ -106,3 +106,20 @@ def test_the_rule_is_the_default(pass_, monkeypatch):
     monkeypatch.delenv("LNG_UB_LIGASE_UNIT", raising=False)
     assert lng.env_flag("LNG_UB_LIGASE_UNIT") is True
     assert pass_() == {("u-c", "u-rbl2")}
+
+
+def test_rule_b_applies_only_where_no_active_unit_is_curated(monkeypatch):
+    # review 2026-10-22: the stub above answers activeUnit with nothing, so rule A
+    # and its precedence over rule B were untested. R1 has an active unit (the
+    # ligase L); its release query would also name a partner P, which must stay.
+    class G:
+        def run(self, q, **kw):
+            if "activeUnit" in q:
+                return _Result([{"rxn": "R1", "leaf_stid": "L"}])
+            if "<-[:hasComponent]-(:Complex)<-[:input]-" in q:
+                assert kw["ids"] == ["R2"]          # R1 is not asked again
+                rows = [{"rxn": "R1", "leaf_stid": "P"}, {"rxn": "R2", "leaf_stid": "Q"}]
+                return _Result([r for r in rows if r["rxn"] in kw["ids"]])
+            raise AssertionError(q)
+    monkeypatch.setattr(nc, "get_graph", lambda: G())
+    assert lng._ub_ligase_leaves(["R1", "R2"]) == {"R1": {"L"}, "R2": {"Q"}}