From f2ba9cc16c7448eb92581b26d03b69764f09a870 Mon Sep 17 00:00:00 2001 From: Adam Wright <668955+adamjohnwright@users.noreply.github.com> Date: Sat, 26 Sep 2026 13:51:15 -0400 Subject: [PATCH 1/8] drugs.csv: ship the pathway's drug-derived entities (Drug; complex with any drug component; set whose every member is one) beside cofactors.csv, for a consumer modelling a cell without the drug (deltasignal specs/032). Network output unchanged. Also: stale +228 comment -> +197. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/logic_network_generator.py | 38 ++++++++++++++- src/neo4j_connector.py | 80 +++++++++++++++++++++++++++++++- src/pathway_generator.py | 9 ++++ tests/test_provenance_exports.py | 50 ++++++++++++++++++++ 4 files changed, 175 insertions(+), 2 deletions(-) diff --git a/src/logic_network_generator.py b/src/logic_network_generator.py index 1851928..e95def8 100755 --- a/src/logic_network_generator.py +++ b/src/logic_network_generator.py @@ -1846,7 +1846,7 @@ def _is_complex(entity_id: str) -> bool: # root ISGF3:KPNA1:KPNB1 is translocated to the nucleus, and every IFN # alpha/beta perturbation upstream of ISGF3 was severed there (200 held-out # cases). The downstream-reuse rule (specs/018) applies unchanged. - hierarchy = os.environ.get("LNG_BOUNDARY_HIERARCHY", "1") == "1" # default since deltasignal specs/030 (+228 held-out) + hierarchy = os.environ.get("LNG_BOUNDARY_HIERARCHY", "1") == "1" # default since deltasignal specs/030 (held-out +197, re-measured) from src.neo4j_connector import get_complex_components nested_registry: Dict[tuple, str] = {} seen_edges: Set[tuple] = set() @@ -3180,6 +3180,42 @@ def export_cofactors(pathway_logic_network: pd.DataFrame, f"({present_count} present in this network) to {output_file}") +def export_drugs(pathway_logic_network: pd.DataFrame, + reactome_id_to_uuid: Dict[str, str], + output_file: str) -> None: + """Write drugs.csv: which entities in THIS network are drug-derived. + + Reactome curates drug actions inside signalling pathways (MAP2K inhibitors + in the RAF/MAP kinase cascade, PARP inhibitors in DNA repair). The network + keeps them, since they are curated. A consumer modelling a cell WITHOUT the + drug needs to know which nodes they are, and, as with cofactors.csv, that + knowledge travels with the artifacts (deltasignal specs/032). + + Only entities present in the network are listed (a release has thousands). + The file is written even when there are none, so a consumer can tell "no + drugs here" from "a bundle that predates this file". + + Output CSV columns: stable_id, schema_class, name, reactome_release. + """ + from src.neo4j_connector import get_drug_entities, get_reactome_release + + present: set[str] = set() + if not pathway_logic_network.empty: + for node_id in _uuid_to_stable_id_map( + pathway_logic_network, reactome_id_to_uuid).values(): + # set_variant ids are "{parent}::variant::{members}": the drug rule + # is decided on the entity itself. + present.add(str(node_id).split("::")[0]) + drugs = get_drug_entities(present) + release = get_reactome_release() + rows = [{"stable_id": s, "schema_class": d["schema_class"], "name": d["name"], + "reactome_release": release if release is not None else ""} + for s, d in sorted(drugs.items())] + pd.DataFrame(rows, columns=["stable_id", "schema_class", "name", "reactome_release"]).to_csv( + output_file, index=False) + logger.info(f"Exported {len(rows)} drug-derived entities to {output_file}") + + def export_node_resolution(pathway_id: str, pathway_logic_network: pd.DataFrame, reaction_id_map: pd.DataFrame, diff --git a/src/neo4j_connector.py b/src/neo4j_connector.py index 3c08384..d98ac44 100755 --- a/src/neo4j_connector.py +++ b/src/neo4j_connector.py @@ -1,7 +1,7 @@ import os from src.credential_redaction import scrub -from typing import Any, Dict, List, Optional, Set, Union +from typing import Any, Dict, List, Optional, Set, Tuple, Union import pandas as pd from py2neo import Graph # type: ignore @@ -861,6 +861,84 @@ def get_reference_entity_id(entity_id: str) -> Union[str, None]: +_drug_structure_cache: Dict[str, Tuple[bool, List[Tuple[str, str]]]] = {} + + +def get_drug_entities(stable_ids) -> Dict[str, Dict[str, str]]: + """The entities among ``stable_ids`` that are DRUG-DERIVED, with their + schema class and display name (deltasignal specs/032). + + Drug-derived is decided structurally, from the release itself: + - a Reactome ``Drug`` (ChemicalDrug, ProteinDrug, RNADrug) is; + - a complex is if ANY component is (a MAP2K bound to trametinib); + - a set is only if EVERY member is, so a ligand set that merely includes + a therapeutic antibody does not flag its physiological members. + + A benchmark describes a cell without the drug, and a consumer that holds + these at baseline needs the list to travel with the network rather than + keep its own copy, which is how the cofactor lists once diverged. + """ + ids = sorted({s for s in stable_ids if s}) + missing = [s for s in ids if s not in _drug_structure_cache] + if missing: + query = """ + UNWIND $ids AS s + MATCH (n:PhysicalEntity {stId: s})-[:hasComponent|hasMember|hasCandidate*0..10]->(x) + WITH DISTINCT x + OPTIONAL MATCH (x)-[r:hasComponent|hasMember|hasCandidate]->(y) + RETURN x.stId AS x, x:Drug AS is_drug, + collect(CASE WHEN y IS NULL THEN NULL ELSE [type(r), y.stId] END) AS kids + """ + try: + rows = get_graph().run(query, ids=missing).data() + except Exception: + logger.error("Error in get_drug_entities", **_traceback_kwargs()) + raise + for r in rows: + if r.get("x"): + _drug_structure_cache[r["x"]] = ( + bool(r["is_drug"]), [tuple(k) for k in r["kids"] if k and k[1]]) + for s in missing: + _drug_structure_cache.setdefault(s, (False, [])) + + memo: Dict[str, bool] = {} + + def derived(s: str, depth: int = 0) -> bool: + if s in memo: + return memo[s] + memo[s] = False # a malformed self-containing entity cannot recurse + is_drug, kids = _drug_structure_cache.get(s, (False, [])) + comps = [y for t, y in kids if t == "hasComponent"] + members = [y for t, y in kids if t != "hasComponent"] + if is_drug: + out = True + elif depth >= 10: + out = False + elif comps: + out = any(derived(y, depth + 1) for y in comps) + elif members: + out = all(derived(y, depth + 1) for y in members) + else: + out = False + memo[s] = out + return out + + hits = [s for s in ids if derived(s)] + if not hits: + return {} + try: + meta = get_graph().run( + "UNWIND $ids AS s MATCH (n {stId: s}) " + "RETURN n.stId AS s, n.schemaClass AS c, n.displayName AS d", + ids=hits).data() + except Exception: + logger.error("Error in get_drug_entities", **_traceback_kwargs()) + raise + by = {m["s"]: m for m in meta} + return {s: {"schema_class": (by.get(s) or {}).get("c") or "", + "name": (by.get(s) or {}).get("d") or ""} for s in hits} + + def get_reactome_release() -> Optional[int]: """The Reactome release number of the connected graph, or None. diff --git a/src/pathway_generator.py b/src/pathway_generator.py index cfe7d4f..cceff34 100755 --- a/src/pathway_generator.py +++ b/src/pathway_generator.py @@ -11,6 +11,7 @@ from src.logic_network_generator import ( create_pathway_logic_network, export_cofactors, + export_drugs, export_containment, export_entity_reaction_proxy_mapping, export_node_reaction_context, @@ -484,6 +485,14 @@ def generate_pathway_file( str(pathway_output_dir / "cofactors.csv"), ) + # Drug-derived entities (deltasignal specs/032), for a consumer that + # models a cell without the drug. + export_drugs( + result.logic_network, + result.uuid_mapping, + str(pathway_output_dir / "drugs.csv"), + ) + # What each node CONTAINS, so a consumer can select nodes by # containment instead of us inventing assembly/dissociation edges. export_containment( diff --git a/tests/test_provenance_exports.py b/tests/test_provenance_exports.py index 7d7e149..0f58e68 100644 --- a/tests/test_provenance_exports.py +++ b/tests/test_provenance_exports.py @@ -285,3 +285,53 @@ def test_export_cofactors_finds_an_entity_split_across_uuids(tmp_path, monkeypat df = pd.read_csv(out) assert int(df.in_network.sum()) == 1, "split entity missed when only one uuid is used" + + +# --- drugs.csv (deltasignal specs/032) --------------------------------------- + +def test_export_drugs_lists_present_drug_entities_only(tmp_path, monkeypatch): + seen = {} + def fake(stids): + seen["asked"] = set(stids) + return {"R-HSA-D": {"schema_class": "ChemicalDrug", "name": "trametinib [cytosol]"}} + monkeypatch.setattr(neo4j_connector, "get_drug_entities", fake) + monkeypatch.setattr(neo4j_connector, "get_reactome_release", lambda: 97) + edges = pd.DataFrame([{"source_id": "u-d", "target_id": "u-x"}]) + out = tmp_path / "drugs.csv" + m.export_drugs(edges, {"R-HSA-D": "u-d", "R-HSA-X::variant::R-HSA-A": "u-x"}, str(out)) + df = pd.read_csv(out) + assert list(df.columns) == ["stable_id", "schema_class", "name", "reactome_release"] + assert list(df.stable_id) == ["R-HSA-D"] and set(df.reactome_release) == {97} + assert seen["asked"] == {"R-HSA-D", "R-HSA-X"} # a variant is judged by its entity + + +def test_export_drugs_writes_a_header_only_file_when_none(tmp_path, monkeypatch): + monkeypatch.setattr(neo4j_connector, "get_drug_entities", lambda s: {}) + monkeypatch.setattr(neo4j_connector, "get_reactome_release", lambda: 97) + out = tmp_path / "drugs.csv" + m.export_drugs(pd.DataFrame([{"source_id": "u-x", "target_id": "u-y"}]), {"R-HSA-X": "u-x"}, str(out)) + assert out.exists() and len(pd.read_csv(out)) == 0 + + +def test_drug_rule_complex_any_component_set_every_member(monkeypatch): + # D is a drug. C1 = complex(P, D) is drug-derived; S_all = set{D, D2} is; + # S_mixed = set{D, P} is NOT (its physiological member must not be held); + # C2 = complex(P, S_mixed) is NOT either; P is not. + struct = { + "D": (True, []), "D2": (True, []), "P": (False, []), + "C1": (False, [("hasComponent", "P"), ("hasComponent", "D")]), + "S_all": (False, [("hasMember", "D"), ("hasCandidate", "D2")]), + "S_mixed": (False, [("hasMember", "D"), ("hasMember", "P")]), + "C2": (False, [("hasComponent", "P"), ("hasComponent", "S_mixed")]), + } + monkeypatch.setattr(neo4j_connector, "_drug_structure_cache", dict(struct)) + + class G: + def run(self, q, **kw): + class R: + def data(_): + return [{"s": s, "c": "X", "d": s} for s in kw["ids"]] + return R() + monkeypatch.setattr(neo4j_connector, "get_graph", lambda: G()) + got = neo4j_connector.get_drug_entities(struct) + assert set(got) == {"D", "D2", "C1", "S_all"} From 46aed80bd20b0a77c1731683623b9be3f3d93d48 Mon Sep 17 00:00:00 2001 From: Adam Wright <668955+adamjohnwright@users.noreply.github.com> Date: Sat, 26 Sep 2026 14:21:25 -0400 Subject: [PATCH 2/8] Review of #98: drugs.csv in its own try block after containment (a failed drug query no longer costs containment.csv); class+name from the structure query (the unlabelled MATCH scanned every node, ~30 s a pathway; whole catalog now 0.8 s); list each node's exact id, a set_variant judged by its parent or chosen members; tests: candidate-only set (dropping hasCandidate now caught), the Cypher+cache path, variant ids. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/logic_network_generator.py | 40 ++++++++++++++++++------- src/neo4j_connector.py | 23 +++++--------- src/pathway_generator.py | 21 ++++++++----- tests/test_provenance_exports.py | 51 ++++++++++++++++++++++++++++---- 4 files changed, 97 insertions(+), 38 deletions(-) diff --git a/src/logic_network_generator.py b/src/logic_network_generator.py index e95def8..4b49966 100755 --- a/src/logic_network_generator.py +++ b/src/logic_network_generator.py @@ -3199,18 +3199,38 @@ def export_drugs(pathway_logic_network: pd.DataFrame, """ from src.neo4j_connector import get_drug_entities, get_reactome_release - present: set[str] = set() + node_ids: set[str] = set() if not pathway_logic_network.empty: - for node_id in _uuid_to_stable_id_map( - pathway_logic_network, reactome_id_to_uuid).values(): - # set_variant ids are "{parent}::variant::{members}": the drug rule - # is decided on the entity itself. - present.add(str(node_id).split("::")[0]) - drugs = get_drug_entities(present) + node_ids = {str(n) for n in _uuid_to_stable_id_map( + pathway_logic_network, reactome_id_to_uuid).values()} + + # A set_variant node id is "{parent}::variant::{m1}_{m2}...", and that exact + # string is what a consumer reads as the node's stable id + # (stid_to_uuid_mapping.csv), so it is what must be listed (review of PR + # #98). A variant is drug-derived if its parent is, or if a member it + # CHOSE is: a variant that picks the drug out of a mixed set is the drug. + def parts(node_id: str): + if "::variant::" not in node_id: + return node_id, [] + parent, members = node_id.split("::variant::", 1) + return parent, [m for m in members.split("_") if m] + + wanted = set() + for n in node_ids: + parent, members = parts(n) + wanted.add(parent) + wanted.update(members) + drugs = get_drug_entities(wanted) release = get_reactome_release() - rows = [{"stable_id": s, "schema_class": d["schema_class"], "name": d["name"], - "reactome_release": release if release is not None else ""} - for s, d in sorted(drugs.items())] + rows = [] + for n in sorted(node_ids): + parent, members = parts(n) + hit = parent if parent in drugs else next((m for m in members if m in drugs), None) + if hit is None: + continue + rows.append({"stable_id": n, "schema_class": drugs[hit]["schema_class"], + "name": drugs[hit]["name"], + "reactome_release": release if release is not None else ""}) pd.DataFrame(rows, columns=["stable_id", "schema_class", "name", "reactome_release"]).to_csv( output_file, index=False) logger.info(f"Exported {len(rows)} drug-derived entities to {output_file}") diff --git a/src/neo4j_connector.py b/src/neo4j_connector.py index d98ac44..cdc3b45 100755 --- a/src/neo4j_connector.py +++ b/src/neo4j_connector.py @@ -862,6 +862,7 @@ def get_reference_entity_id(entity_id: str) -> Union[str, None]: _drug_structure_cache: Dict[str, Tuple[bool, List[Tuple[str, str]]]] = {} +_drug_meta_cache: Dict[str, Tuple[str, str]] = {} def get_drug_entities(stable_ids) -> Dict[str, Dict[str, str]]: @@ -886,7 +887,7 @@ def get_drug_entities(stable_ids) -> Dict[str, Dict[str, str]]: MATCH (n:PhysicalEntity {stId: s})-[:hasComponent|hasMember|hasCandidate*0..10]->(x) WITH DISTINCT x OPTIONAL MATCH (x)-[r:hasComponent|hasMember|hasCandidate]->(y) - RETURN x.stId AS x, x:Drug AS is_drug, + RETURN x.stId AS x, x:Drug AS is_drug, x.schemaClass AS c, x.displayName AS d, collect(CASE WHEN y IS NULL THEN NULL ELSE [type(r), y.stId] END) AS kids """ try: @@ -898,6 +899,7 @@ def get_drug_entities(stable_ids) -> Dict[str, Dict[str, str]]: if r.get("x"): _drug_structure_cache[r["x"]] = ( bool(r["is_drug"]), [tuple(k) for k in r["kids"] if k and k[1]]) + _drug_meta_cache[r["x"]] = (r.get("c") or "", r.get("d") or "") for s in missing: _drug_structure_cache.setdefault(s, (False, [])) @@ -923,20 +925,11 @@ def derived(s: str, depth: int = 0) -> bool: memo[s] = out return out - hits = [s for s in ids if derived(s)] - if not hits: - return {} - try: - meta = get_graph().run( - "UNWIND $ids AS s MATCH (n {stId: s}) " - "RETURN n.stId AS s, n.schemaClass AS c, n.displayName AS d", - ids=hits).data() - except Exception: - logger.error("Error in get_drug_entities", **_traceback_kwargs()) - raise - by = {m["s"]: m for m in meta} - return {s: {"schema_class": (by.get(s) or {}).get("c") or "", - "name": (by.get(s) or {}).get("d") or ""} for s in hits} + # Class and name come back with the structure query: a second, unlabelled + # MATCH (n {stId: s}) scanned every node (review of PR #98: ~30 s a pathway). + return {s: {"schema_class": _drug_meta_cache.get(s, ("", ""))[0], + "name": _drug_meta_cache.get(s, ("", ""))[1]} + for s in ids if derived(s)} def get_reactome_release() -> Optional[int]: diff --git a/src/pathway_generator.py b/src/pathway_generator.py index cceff34..56f7ff7 100755 --- a/src/pathway_generator.py +++ b/src/pathway_generator.py @@ -485,14 +485,6 @@ def generate_pathway_file( str(pathway_output_dir / "cofactors.csv"), ) - # Drug-derived entities (deltasignal specs/032), for a consumer that - # models a cell without the drug. - export_drugs( - result.logic_network, - result.uuid_mapping, - str(pathway_output_dir / "drugs.csv"), - ) - # What each node CONTAINS, so a consumer can select nodes by # containment instead of us inventing assembly/dissociation edges. export_containment( @@ -503,6 +495,19 @@ def generate_pathway_file( logger.error(f"Failed to write node provenance files: {e}", exc_info=True) # Don't raise - supplementary + # Drug-derived entities (deltasignal specs/032), in their OWN block: a + # failed drug query must not cost containment.csv, which the default + # solver needs (review of PR #98). A missing drugs.csv is reported by + # the consumer as "no drug table", not mistaken for "no drugs". + try: + export_drugs( + result.logic_network, + result.uuid_mapping, + str(pathway_output_dir / "drugs.csv"), + ) + except Exception as e: + logger.error(f"Failed to write drugs.csv: {e}", exc_info=True) + logger.info(f"Output directory: {pathway_output_dir}") except (ConnectionError, ValueError) as e: diff --git a/tests/test_provenance_exports.py b/tests/test_provenance_exports.py index 0f58e68..82cc15d 100644 --- a/tests/test_provenance_exports.py +++ b/tests/test_provenance_exports.py @@ -302,7 +302,22 @@ def fake(stids): df = pd.read_csv(out) assert list(df.columns) == ["stable_id", "schema_class", "name", "reactome_release"] assert list(df.stable_id) == ["R-HSA-D"] and set(df.reactome_release) == {97} - assert seen["asked"] == {"R-HSA-D", "R-HSA-X"} # a variant is judged by its entity + # A variant is judged by its parent AND the members it chose. + assert seen["asked"] == {"R-HSA-D", "R-HSA-X", "R-HSA-A"} + + +def test_export_drugs_lists_a_variant_by_its_exact_node_id(tmp_path, monkeypatch): + # Set S mixes a drug D and a protein P, so S is not drug-derived; the variant + # that CHOSE D is, and it must be listed under the exact id a consumer reads + # (stid_to_uuid_mapping.csv), not the parent's. + monkeypatch.setattr(neo4j_connector, "get_drug_entities", + lambda stids: {"R-HSA-D": {"schema_class": "ChemicalDrug", "name": "d"}}) + monkeypatch.setattr(neo4j_connector, "get_reactome_release", lambda: 97) + vd, vp = "R-HSA-S::variant::R-HSA-D", "R-HSA-S::variant::R-HSA-P" + edges = pd.DataFrame([{"source_id": "u-1", "target_id": "u-2"}]) + out = tmp_path / "drugs.csv" + m.export_drugs(edges, {vd: "u-1", vp: "u-2"}, str(out)) + assert list(pd.read_csv(out).stable_id) == [vd] def test_export_drugs_writes_a_header_only_file_when_none(tmp_path, monkeypatch): @@ -320,18 +335,44 @@ def test_drug_rule_complex_any_component_set_every_member(monkeypatch): struct = { "D": (True, []), "D2": (True, []), "P": (False, []), "C1": (False, [("hasComponent", "P"), ("hasComponent", "D")]), - "S_all": (False, [("hasMember", "D"), ("hasCandidate", "D2")]), + # candidate-only, so dropping hasCandidate from the rule is caught + "S_all": (False, [("hasCandidate", "D"), ("hasCandidate", "D2")]), "S_mixed": (False, [("hasMember", "D"), ("hasMember", "P")]), "C2": (False, [("hasComponent", "P"), ("hasComponent", "S_mixed")]), } monkeypatch.setattr(neo4j_connector, "_drug_structure_cache", dict(struct)) + monkeypatch.setattr(neo4j_connector, "_drug_meta_cache", {}) + + def boom(): + raise AssertionError("fully cached: no query expected") + monkeypatch.setattr(neo4j_connector, "get_graph", boom) + got = neo4j_connector.get_drug_entities(struct) + assert set(got) == {"D", "D2", "C1", "S_all"} + + +def test_get_drug_entities_queries_uncached_ids_and_keeps_misses(monkeypatch): + # The Cypher path: the structure query returns one row per reachable entity + # (with class and name), and an id it does not return (a reaction stid) is + # cached as not-a-drug instead of being re-queried. + monkeypatch.setattr(neo4j_connector, "_drug_structure_cache", {}) + monkeypatch.setattr(neo4j_connector, "_drug_meta_cache", {}) + calls = [] + class G: def run(self, q, **kw): + calls.append(sorted(kw["ids"])) + class R: def data(_): - return [{"s": s, "c": "X", "d": s} for s in kw["ids"]] + return [{"x": "C", "is_drug": False, "c": "Complex", "d": "P:D", + "kids": [["hasComponent", "P"], ["hasComponent", "D"]]}, + {"x": "P", "is_drug": False, "c": "EWAS", "d": "P", "kids": [None]}, + {"x": "D", "is_drug": True, "c": "ChemicalDrug", "d": "d", "kids": []}] return R() monkeypatch.setattr(neo4j_connector, "get_graph", lambda: G()) - got = neo4j_connector.get_drug_entities(struct) - assert set(got) == {"D", "D2", "C1", "S_all"} + got = neo4j_connector.get_drug_entities({"C", "R-RXN"}) + assert got == {"C": {"schema_class": "Complex", "name": "P:D"}} + assert calls == [["C", "R-RXN"]] + assert neo4j_connector.get_drug_entities({"C", "R-RXN"}) == got + assert len(calls) == 1 # second call served from the cache From 95d577648dd849b75ca14041f266733ea2e7b9ea Mon Sep 17 00:00:00 2001 From: Adam Wright <668955+adamjohnwright@users.noreply.github.com> Date: Sat, 26 Sep 2026 22:14:40 -0400 Subject: [PATCH 3/8] LNG_SET_POOL (deltasignal specs/033, default off): a bare EntitySet catalyst/regulator becomes one pool node (the set) fed by set_member edges, feeding each reaction copy once, instead of every member wired onto every copy as a separate AND (catalyst/pos) or OR (neg) term. Reachability unchanged. nodes.csv kind set_pool. Co-Authored-By: Claude Opus 5.5 (1M context) --- schema/logic_network.linkml.yaml | 6 ++ src/logic_network_generator.py | 52 +++++++++++++ src/pathway_generator.py | 1 + tests/test_set_pool.py | 122 +++++++++++++++++++++++++++++++ 4 files changed, 181 insertions(+) create mode 100644 tests/test_set_pool.py diff --git a/schema/logic_network.linkml.yaml b/schema/logic_network.linkml.yaml index 6fa5649..32e3778 100644 --- a/schema/logic_network.linkml.yaml +++ b/schema/logic_network.linkml.yaml @@ -58,6 +58,12 @@ enums: an assembly edge. regulator_variant: description: A set-variant produced by catalyst / negative-regulator decomposition. + set_pool: + description: >- + An EntitySet catalyst or regulator kept as ONE node (LNG_SET_POOL, + deltasignal specs/033): its members feed it by set_member edges and it + feeds each reaction it serves once. diagram_entity_id is the set; + member_leaves are its members. reaction: description: A virtual-reaction node (edges route input->reaction->output). other: diff --git a/src/logic_network_generator.py b/src/logic_network_generator.py index 1851928..53a88c8 100755 --- a/src/logic_network_generator.py +++ b/src/logic_network_generator.py @@ -2014,6 +2014,8 @@ def append_regulators( if entity_dbId not in stid_to_existing_uuid: stid_to_existing_uuid[entity_dbId] = entity_uuid + set_pool_uuids: Dict[str, str] = {} # set stId -> its pool node (LNG_SET_POOL) + regulator_configs = [ (catalyst_map, "pos", "catalyst"), (negative_regulator_map, "neg", "regulator"), @@ -2050,6 +2052,10 @@ def append_regulators( # Emit set-derived positive regulator members as OR alternatives # (see the and_or comment below). Opt-in while it is being A/B'd. set_members_or = os.environ.get("LNG_SET_MEMBERS_OR", "0") == "1" + # deltasignal specs/033: a bare EntitySet regulator becomes ONE pool node + # (the set itself) fed by its members, instead of every member wired + # onto every reaction copy as a separate required (or blocking) term. + set_pool = os.environ.get("LNG_SET_POOL", "0") == "1" from src.neo4j_connector import get_labels, get_set_members for _, row in map_df.iterrows(): @@ -2100,6 +2106,45 @@ def append_regulators( if len(members) > 1 and len(terminal_members) == len(members): and_or = "or" + if (set_pool and len(terminal_members) > 1 + and entity_id not in modifier_isoform_set_ids() + and any(x in (get_labels(entity_id) or []) + for x in ("EntitySet", "DefinedSet", "CandidateSet"))): + # One pool node per set per pathway, keyed apart from any node + # the same stId has as a produced/consumed entity, so a pool + # never mixes its members with a producing reaction. Members + # reach every reaction the set serves exactly as before, so + # reachability (and every cycle) is unchanged. + pool_uuid = set_pool_uuids.get(entity_id) + if pool_uuid is None: + pool_uuid = str(uuid.uuid4()) + set_pool_uuids[entity_id] = pool_uuid + reactome_id_to_uuid[pool_uuid] = entity_id + for member_id, member_stoich in terminal_members: + if member_id in stid_to_existing_uuid: + member_uuid = stid_to_existing_uuid[member_id] + else: + member_uuid = str(uuid.uuid4()) + stid_to_existing_uuid[member_id] = member_uuid + pathway_logic_network_data.append({ + "source_id": member_uuid, + "target_id": pool_uuid, + "pos_neg": "pos", + "and_or": "or", + "edge_type": "set_member", + "stoichiometry": member_stoich, + }) + reactome_id_to_uuid[member_uuid] = member_id + pathway_logic_network_data.append({ + "source_id": pool_uuid, + "target_id": row["reaction_uuid"], + "pos_neg": pos_neg, + "and_or": and_or, + "edge_type": edge_type, + "stoichiometry": 1, + }) + continue + for member_id, member_stoich in terminal_members: if member_id in stid_to_existing_uuid: member_uuid = stid_to_existing_uuid[member_id] @@ -2929,6 +2974,13 @@ def export_nodes(pathway_logic_network: pd.DataFrame, inc = incoming_types.get(u, set()) if "dissociation" in inc and u not in has_outgoing: kind, members = "dissociation_sink", [s] + elif "set_member" in inc: + # LNG_SET_POOL (specs/033): the set itself, fed by its members. + kind = "set_pool" + try: + members = sorted(get_terminal_components(s)) + except Exception: + members = [s] else: try: labels = get_labels(s) diff --git a/src/pathway_generator.py b/src/pathway_generator.py index cfe7d4f..a0aa03b 100755 --- a/src/pathway_generator.py +++ b/src/pathway_generator.py @@ -35,6 +35,7 @@ "LNG_HANDOFF_EDGES", "LNG_HANDOFF_HUB_MAX", "LNG_SET_MEMBERS_OR", + "LNG_SET_POOL", "LNG_DIAGRAM_SET_MEMBER", # 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 diff --git a/tests/test_set_pool.py b/tests/test_set_pool.py new file mode 100644 index 0000000..a4a9622 --- /dev/null +++ b/tests/test_set_pool.py @@ -0,0 +1,122 @@ +"""LNG_SET_POOL (deltasignal specs/033): a bare EntitySet catalyst or regulator +becomes ONE pool node fed by its members, instead of every member wired onto +every reaction copy as a separate required (catalyst, positive regulator) or +blocking (negative regulator) term. + +Shape: RAF/MAP kinase. "RAF activating kinases" (S) catalyses "Phosphorylation +of RAF", which exists as several reaction copies; each copy used to receive all +members as separate AND catalysts.""" +from typing import Any, Dict, List +from unittest.mock import patch + +import pandas as pd +import pytest + +with patch("py2neo.Graph"): + import src.logic_network_generator as lng + from src.logic_network_generator import append_regulators + +S, K = "R-HSA-S", "R-HSA-K" # S = a set; K = a complex (not a set) +MEMBERS = {S: [("R-HSA-M1", 1), ("R-HSA-M2", 1), ("R-HSA-M3", 1)], + K: [("R-HSA-A", 1), ("R-HSA-B", 1)]} +LABELS = {S: ["DefinedSet"], K: ["Complex"]} + + +@pytest.fixture +def stub(monkeypatch): + monkeypatch.setattr(lng, "_decompose_regulator_entity", + lambda e, variant_decomposition=False, bundle_complex=False: + MEMBERS.get(e, [(e, 1)])) + import src.neo4j_connector as nc + monkeypatch.setattr(nc, "get_labels", lambda e: LABELS.get(e, [])) + monkeypatch.setattr(nc, "get_set_members", lambda e: [m for m, _ in MEMBERS.get(e, [])]) + monkeypatch.setattr(lng, "modifier_isoform_set_ids", lambda: set()) + monkeypatch.delenv("LNG_SET_MEMBERS_OR", raising=False) + + +def row(entity, rxn): + return {"reaction_id": "R-HSA-RX", "entity_id": entity, "edge_type": "x", + "uuid": "u", "reaction_uuid": rxn} + + +def run(cat=(), neg=(), pos=()): + data: List[Dict[str, Any]] = [] + r2u: Dict[str, str] = {} + append_regulators(pd.DataFrame([row(*x) for x in cat]), pd.DataFrame([row(*x) for x in neg]), + pd.DataFrame([row(*x) for x in pos]), data, r2u) + return data, r2u + + +def reach(data, src): + fwd: Dict[str, set] = {} + for e in data: + fwd.setdefault(e["source_id"], set()).add(e["target_id"]) + seen, stack = {src}, [src] + while stack: + for t in fwd.get(stack.pop(), ()): + if t not in seen: + seen.add(t) + stack.append(t) + return seen + + +def test_off_is_the_legacy_member_fan_out(stub, monkeypatch): + monkeypatch.delenv("LNG_SET_POOL", raising=False) + data, r2u = run(cat=[(S, "rx1"), (S, "rx2")]) + assert len(data) == 6 # 3 members x 2 copies + assert {e["and_or"] for e in data} == {"and"} and {e["edge_type"] for e in data} == {"catalyst"} + assert not any(e["edge_type"] == "set_member" for e in data) + + +def test_on_one_pool_node_feeds_every_copy_once(stub, monkeypatch): + monkeypatch.setenv("LNG_SET_POOL", "1") + data, r2u = run(cat=[(S, "rx1"), (S, "rx2")]) + pools = [u for u, s in r2u.items() if s == S] + assert len(pools) == 1 + p = pools[0] + members = [e for e in data if e["edge_type"] == "set_member"] + assert len(members) == 3 and {e["target_id"] for e in members} == {p} + assert {r2u[e["source_id"]] for e in members} == {"R-HSA-M1", "R-HSA-M2", "R-HSA-M3"} + assert all(e["and_or"] == "or" and e["pos_neg"] == "pos" for e in members) + into = [e for e in data if e["source_id"] == p] + assert sorted(e["target_id"] for e in into) == ["rx1", "rx2"] + assert all(e["edge_type"] == "catalyst" and e["and_or"] == "and" and e["pos_neg"] == "pos" + for e in into) + # reachability is what it was: every member still reaches every copy + for e in members: + assert {"rx1", "rx2"} <= reach(data, e["source_id"]) + + +def test_negative_regulator_set_is_one_inhibitor_term(stub, monkeypatch): + monkeypatch.setenv("LNG_SET_POOL", "1") + data, r2u = run(neg=[(S, "rx1")]) + p = next(u for u, s in r2u.items() if s == S) + into = [e for e in data if e["target_id"] == "rx1"] + assert len(into) == 1 and into[0]["source_id"] == p + assert into[0]["pos_neg"] == "neg" and into[0]["and_or"] == "or" and into[0]["edge_type"] == "regulator" + + +def test_one_pool_per_set_across_roles_and_reactions(stub, monkeypatch): + monkeypatch.setenv("LNG_SET_POOL", "1") + data, r2u = run(cat=[(S, "rx1")], neg=[(S, "rx2")], pos=[(S, "rx3")]) + assert len([u for u, s in r2u.items() if s == S]) == 1 + assert len([e for e in data if e["edge_type"] == "set_member"]) == 3 # emitted once + + +def test_a_complex_and_a_modifier_set_are_not_pooled(stub, monkeypatch): + monkeypatch.setenv("LNG_SET_POOL", "1") + data, r2u = run(cat=[(K, "rx1")]) + assert not any(e["edge_type"] == "set_member" for e in data) + assert len(data) == 2 # complex members as before + monkeypatch.setattr(lng, "modifier_isoform_set_ids", lambda: {S}) + data, r2u = run(cat=[(S, "rx1")]) + assert not any(e["edge_type"] == "set_member" for e in data) + + +def test_a_member_already_in_the_network_is_reused(stub, monkeypatch): + monkeypatch.setenv("LNG_SET_POOL", "1") + data: List[Dict[str, Any]] = [] + r2u: Dict[str, str] = {} + registry = {("R-HSA-M2", "some-rxn", "input"): "u-m2-existing"} + append_regulators(pd.DataFrame([row(S, "rx1")]), pd.DataFrame(), pd.DataFrame(), data, r2u, registry) + assert any(e["source_id"] == "u-m2-existing" and e["edge_type"] == "set_member" for e in data) From 81499ef14972316aad4012ef7f87f488e4dfb4dc Mon Sep 17 00:00:00 2001 From: Adam Wright <668955+adamjohnwright@users.noreply.github.com> Date: Sat, 26 Sep 2026 22:28:59 -0400 Subject: [PATCH 4/8] Review of set-pool: a pooled catalyst no longer depletes its own member (the stId guards compared the SET's id; 4 E3-ligase sets would have closed member->pool-|member cycles); pooling only under complex-as-node; pool->reaction keeps the role flag under LNG_SET_MEMBERS_OR. Tests: depletion self-skip (mutation-checked), complex-as-node guard, SET_MEMBERS_OR, export_nodes set_pool kind. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/logic_network_generator.py | 28 +++++++++++++++++-- tests/test_set_pool.py | 50 ++++++++++++++++++++++++++++++++++ 2 files changed, 76 insertions(+), 2 deletions(-) diff --git a/src/logic_network_generator.py b/src/logic_network_generator.py index 53a88c8..db298ef 100755 --- a/src/logic_network_generator.py +++ b/src/logic_network_generator.py @@ -1304,6 +1304,22 @@ def _emit_substrate_depletion_edges( if sid: stid_to_uuids_in_net.setdefault(sid, []).append(uid) + # LNG_SET_POOL (specs/033): a pooled set catalyst depletes through its pool + # node, but the "a catalyst does not deplete itself" guards below compare + # stIds, and the pool's stId is the SET's. Without this a pool would deplete + # its own member (an E3-ligase set containing the substrate gene), closing a + # member -> pool -| member cycle the unpooled network did not have (review). + pool_members: Dict[str, Tuple[Set[str], Set[str]]] = {} + for edge in pathway_logic_network_data: + if edge.get("edge_type") == "set_member": + uu, ss = pool_members.setdefault(edge["target_id"], (set(), set())) + uu.add(edge["source_id"]) + ss.add(reactome_id_to_uuid.get(edge["source_id"], "")) + + def _pool_contains(cat_uuid: str, uid: str, sid: str) -> bool: + uu, ss = pool_members.get(cat_uuid, (set(), set())) + return uid in uu or (bool(sid) and sid in ss) + # Emit depletion edges. For phosphatase reactions: catalyst → input # (free substrate IS the input). For ubiquitin reactions: catalyst → # all network UUIDs of the substrate protein stId (free form, not the @@ -1325,6 +1341,8 @@ def _emit_substrate_depletion_edges( inp_stid = reactome_id_to_uuid.get(inp_uuid, "") if inp_stid == cat_stid and inp_stid: continue # same biological entity at different positions + if _pool_contains(cat_uuid, inp_uuid, inp_stid): + continue # a pooled set catalyst does not deplete its own member if inp_stid in _cofactor_stids(): continue key = (cat_uuid, inp_uuid) @@ -1357,10 +1375,12 @@ def _emit_substrate_depletion_edges( cat_stid = reactome_id_to_uuid.get(cat_uuid, "") for subst_stid in subst_stids: if subst_stid == cat_stid: continue + if _pool_contains(cat_uuid, "", subst_stid): continue if subst_stid in _cofactor_stids(): continue target_uuids = stid_to_uuids_in_net.get(subst_stid, []) for tgt_uuid in target_uuids: if tgt_uuid == cat_uuid: continue + if _pool_contains(cat_uuid, tgt_uuid, ""): continue key = (cat_uuid, tgt_uuid) if key in seen_edges: continue seen_edges.add(key) @@ -2106,7 +2126,7 @@ def append_regulators( if len(members) > 1 and len(terminal_members) == len(members): and_or = "or" - if (set_pool and len(terminal_members) > 1 + if (set_pool and complex_as_node and len(terminal_members) > 1 and entity_id not in modifier_isoform_set_ids() and any(x in (get_labels(entity_id) or []) for x in ("EntitySet", "DefinedSet", "CandidateSet"))): @@ -2135,11 +2155,15 @@ def append_regulators( "stoichiometry": member_stoich, }) reactome_id_to_uuid[member_uuid] = member_id + # The pool IS the "any one member" semantics, so its edge keeps + # the role's reaction-level flag (pos AND, neg OR) even under + # LNG_SET_MEMBERS_OR, which would otherwise make the pool an OR + # alternative that bypasses the reaction's AND inputs (review). pathway_logic_network_data.append({ "source_id": pool_uuid, "target_id": row["reaction_uuid"], "pos_neg": pos_neg, - "and_or": and_or, + "and_or": "and" if pos_neg == "pos" else "or", "edge_type": edge_type, "stoichiometry": 1, }) diff --git a/tests/test_set_pool.py b/tests/test_set_pool.py index a4a9622..c49c316 100644 --- a/tests/test_set_pool.py +++ b/tests/test_set_pool.py @@ -120,3 +120,53 @@ def test_a_member_already_in_the_network_is_reused(stub, monkeypatch): registry = {("R-HSA-M2", "some-rxn", "input"): "u-m2-existing"} append_regulators(pd.DataFrame([row(S, "rx1")]), pd.DataFrame(), pd.DataFrame(), data, r2u, registry) assert any(e["source_id"] == "u-m2-existing" and e["edge_type"] == "set_member" for e in data) + + +# --- review of the set-pool change ------------------------------------------ + +def _e(a, b, et, pn="pos", ao="and"): + return {"source_id": a, "target_id": b, "pos_neg": pn, "and_or": ao, "edge_type": et, "stoichiometry": 1} + + +def test_a_pooled_phosphatase_does_not_deplete_its_own_member(monkeypatch): + # Pool P = {M1, M2} catalyses rx, a phosphatase (outputs Pi) whose inputs are + # M2 (a member: skipped, or member -> P -| member would be a new cycle) and + # S (a real substrate: depleted by the pool). + monkeypatch.setattr(lng, "_cofactor_stids", lambda: frozenset()) + data = [_e("u-m1", "pool", "set_member", ao="or"), _e("u-m2", "pool", "set_member", ao="or"), + _e("pool", "rx", "catalyst"), _e("u-m2", "rx", "input"), _e("u-s", "rx", "input"), + _e("rx", "u-pi", "output")] + r2u = {"u-m1": "R-HSA-M1", "u-m2": "R-HSA-M2", "pool": S, "u-s": "R-HSA-SUB", "u-pi": "R-ALL-29372"} + lng._emit_substrate_depletion_edges(data, r2u, pd.DataFrame()) + dep = {(e["source_id"], e["target_id"]) for e in data if e["edge_type"] == "depletion"} + assert dep == {("pool", "u-s")} + + +def test_pooling_needs_complex_as_node(stub, monkeypatch): + monkeypatch.setenv("LNG_SET_POOL", "1") + monkeypatch.setenv("LNG_COMPLEX_AS_NODE", "0") + data, _ = run(cat=[(S, "rx1")]) + assert not any(e["edge_type"] == "set_member" for e in data) + + +def test_set_members_or_does_not_turn_the_pool_into_an_or_bypass(stub, monkeypatch): + monkeypatch.setenv("LNG_SET_POOL", "1") + monkeypatch.setenv("LNG_SET_MEMBERS_OR", "1") + data, r2u = run(cat=[(S, "rx1")]) + p = next(u for u, s in r2u.items() if s == S) + assert [e["and_or"] for e in data if e["source_id"] == p] == ["and"] + + +def test_export_nodes_marks_a_pool(tmp_path, monkeypatch): + import src.neo4j_connector as nc + monkeypatch.setattr(nc, "get_labels", lambda e: ["DefinedSet"] if e == S else ["EntityWithAccessionedSequence"]) + monkeypatch.setattr(lng, "get_terminal_components", lambda s: {"R-HSA-M1", "R-HSA-M2"} if s == S else {s}) + a, b, p = ("aaaaaaaa-0000-0000-0000-00000000000%d" % i for i in (1, 2, 3)) + edges = pd.DataFrame([{"source_id": a, "target_id": p, "edge_type": "set_member"}, + {"source_id": b, "target_id": p, "edge_type": "set_member"}]) + out = tmp_path / "nodes.csv" + lng.export_nodes(edges, pd.DataFrame(columns=["uid", "reactome_id"]), + {a: "R-HSA-M1", b: "R-HSA-M2", p: S}, str(out)) + rows = {r["uuid"]: r for r in pd.read_csv(out).to_dict("records")} + assert rows[p]["node_kind"] == "set_pool" + assert set(str(rows[p]["member_leaves"]).split("|")) == {"R-HSA-M1", "R-HSA-M2"} From 9e8a504fb18cf36a0093611529529dda45413742 Mon Sep 17 00:00:00 2001 From: Adam Wright <668955+adamjohnwright@users.noreply.github.com> Date: Sat, 26 Sep 2026 23:18:21 -0400 Subject: [PATCH 5/8] LNG_SET_POOL default on (trial, specs/033: held-out +6, tuning +10, exp +3); legacy regulator tests pinned to LNG_SET_POOL=0; test that pooling is the default Co-Authored-By: Claude Opus 5.5 (1M context) --- src/logic_network_generator.py | 4 +++- tests/test_regulators_and_catalysts.py | 2 ++ tests/test_set_pool.py | 8 +++++++- 3 files changed, 12 insertions(+), 2 deletions(-) diff --git a/src/logic_network_generator.py b/src/logic_network_generator.py index db298ef..ca263db 100755 --- a/src/logic_network_generator.py +++ b/src/logic_network_generator.py @@ -2075,7 +2075,9 @@ def append_regulators( # deltasignal specs/033: a bare EntitySet regulator becomes ONE pool node # (the set itself) fed by its members, instead of every member wired # onto every reaction copy as a separate required (or blocking) term. - set_pool = os.environ.get("LNG_SET_POOL", "0") == "1" + # Default ON since specs/033's arms (held-out +6, tuning +10, exp +3 with + # DS_SET_POOL_MODE=product); "0" restores the member fan-out. + set_pool = os.environ.get("LNG_SET_POOL", "1") == "1" from src.neo4j_connector import get_labels, get_set_members for _, row in map_df.iterrows(): diff --git a/tests/test_regulators_and_catalysts.py b/tests/test_regulators_and_catalysts.py index 19970c1..d201616 100644 --- a/tests/test_regulators_and_catalysts.py +++ b/tests/test_regulators_and_catalysts.py @@ -8,6 +8,7 @@ 5. Regulatory relationships are properly created """ +import os import pytest import pandas as pd from typing import Dict, List, Any @@ -29,6 +30,7 @@ def _mock_decompose(entity_id, variant_decomposition=False, bundle_complex=False return [(entity_id, 1)] +@patch.dict(os.environ, {"LNG_SET_POOL": "0"}) # these pin the member fan-out; pooling: tests/test_set_pool.py class TestRegulatorsAndCatalysts: """Test regulatory and catalytic relationships in logic networks.""" diff --git a/tests/test_set_pool.py b/tests/test_set_pool.py index c49c316..225e184 100644 --- a/tests/test_set_pool.py +++ b/tests/test_set_pool.py @@ -61,7 +61,7 @@ def reach(data, src): def test_off_is_the_legacy_member_fan_out(stub, monkeypatch): - monkeypatch.delenv("LNG_SET_POOL", raising=False) + monkeypatch.setenv("LNG_SET_POOL", "0") data, r2u = run(cat=[(S, "rx1"), (S, "rx2")]) assert len(data) == 6 # 3 members x 2 copies assert {e["and_or"] for e in data} == {"and"} and {e["edge_type"] for e in data} == {"catalyst"} @@ -170,3 +170,9 @@ def test_export_nodes_marks_a_pool(tmp_path, monkeypatch): rows = {r["uuid"]: r for r in pd.read_csv(out).to_dict("records")} assert rows[p]["node_kind"] == "set_pool" assert set(str(rows[p]["member_leaves"]).split("|")) == {"R-HSA-M1", "R-HSA-M2"} + + +def test_pooling_is_the_default(stub, monkeypatch): + monkeypatch.delenv("LNG_SET_POOL", raising=False) + data, r2u = run(cat=[(S, "rx1")]) + assert sum(e["edge_type"] == "set_member" for e in data) == 3 From 7a544be87e0420d33d5037222a12225a7bfc0047 Mon Sep 17 00:00:00 2001 From: Adam Wright <668955+adamjohnwright@users.noreply.github.com> Date: Sun, 27 Sep 2026 01:11:51 -0400 Subject: [PATCH 6/8] WIP LNG_CAP_POOLS (deltasignal specs/036, default off): a set input that resolves to several nodes in one virtual reaction (alternatives bundled by the variant cap) is emitted as a set pool feeding the reaction once. RAF's scaffold binding reaction: 28 inputs -> its 4 curated components. Tests to follow. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/logic_network_generator.py | 57 +++++++++++++++++++++++++++++++--- src/pathway_generator.py | 1 + 2 files changed, 54 insertions(+), 4 deletions(-) diff --git a/src/logic_network_generator.py b/src/logic_network_generator.py index ca263db..1323590 100755 --- a/src/logic_network_generator.py +++ b/src/logic_network_generator.py @@ -823,6 +823,11 @@ def _map_annotated_entity_to_nodes(entity_id: str, member_set: Set[str]) -> Set[ return {str(entity_id)} # simple entity (protein / small molecule / …) +# vr_uid -> {input set stId -> member node ids}: set inputs whose alternatives +# were bundled (specs/036, LNG_CAP_POOLS). Filled by _resolve_vr_entities. +_vr_input_pools: Dict[str, Dict[str, Set[str]]] = {} + + def _resolve_vr_entities( reaction_id_map: pd.DataFrame, uid_index: Dict[str, tuple] @@ -863,6 +868,8 @@ def _annotated_stoich(reaction_id: str, io: str) -> Dict[str, int]: stoich_cache[key] = get_reaction_io_stoichiometry(reaction_id, io) return stoich_cache[key] + cap_pools = os.environ.get("LNG_CAP_POOLS", "0") == "1" + def _resolve_io(reaction_id: str, io: str, members: Set[str]) -> tuple: # Node identity comes from the reaction's annotated entities; the # curated stoichiometry is carried per annotated entity and attached to @@ -872,13 +879,30 @@ def _resolve_io(reaction_id: str, io: str, members: Set[str]) -> tuple: by_entity = _annotated_stoich(reaction_id, io) node_ids: Set[str] = set() node_stoich: Dict[str, int] = {} + pools: Dict[str, Set[str]] = {} for e in _annotated(reaction_id, io): s = by_entity.get(str(e), 1) - for n in _map_annotated_entity_to_nodes(str(e), members): + mapped = _map_annotated_entity_to_nodes(str(e), members) + # deltasignal specs/036: in a normal virtual reaction a set input + # resolves to the ONE member this copy chose. It resolves to several + # only when the alternatives were bundled -- by the variant cap + # (issue #40), which merges every alternative into one variant -- and + # then every member became a separate REQUIRED input (RAF's scaffold + # binding: 8 alternative scaffolds, all required). Record the set as + # a pool of those members instead; Phase 3 wires members -> pool -> + # reaction, so the reaction has one input per curated component. + if (cap_pools and io == "input" and len(mapped) > 1 + and str(e) not in modifier_isoform_set_ids()): + from src.neo4j_connector import get_labels + if any(x in (get_labels(str(e)) or []) + for x in ("EntitySet", "DefinedSet", "CandidateSet")): + pools[str(e)] = set(mapped) + for n in mapped: node_ids.add(n) node_stoich[n] = node_stoich.get(n, 0) + s - return list(node_ids), node_stoich + return list(node_ids), node_stoich, pools + _vr_input_pools.clear() # one pathway per call; the dict is module state vr_entities: Dict[str, tuple] = {} for _, row in reaction_id_map.iterrows(): vr_uid = row["uid"] @@ -886,8 +910,10 @@ def _resolve_io(reaction_id: str, io: str, members: Set[str]) -> tuple: input_members = set(_resolve_to_terminal_reactome_ids(uid_index, row["input_hash"])) output_members = set(_resolve_to_terminal_reactome_ids(uid_index, row["output_hash"])) - input_ids, input_stoich = _resolve_io(reaction_id, "input", input_members) - output_ids, output_stoich = _resolve_io(reaction_id, "output", output_members) + input_ids, input_stoich, input_pools = _resolve_io(reaction_id, "input", input_members) + output_ids, output_stoich, _ = _resolve_io(reaction_id, "output", output_members) + if input_pools: + _vr_input_pools[vr_uid] = input_pools vr_entities[vr_uid] = ( input_ids, output_ids, @@ -2456,8 +2482,31 @@ def create_pathway_logic_network( continue reaction_stid = vr_to_reaction.get(str(vr_uid)) + # specs/036: bundled set alternatives feed ONE pool node per set, which + # feeds the reaction once (members -> pool by set_member, OR). + pools = _vr_input_pools.get(vr_uid, {}) + pooled_member: Dict[str, str] = {} + for set_stid, mem in sorted(pools.items()): + pool_uuid = str(uuid.uuid4()) + reactome_id_to_uuid[pool_uuid] = set_stid + for m in mem: + pooled_member.setdefault(m, pool_uuid) + pathway_logic_network_data.append({ + "source_id": pool_uuid, "target_id": vr_uid, "pos_neg": "pos", + "and_or": "and", "edge_type": "input", + "stoichiometry": input_stoich.get(set_stid, 1), + "edge_reaction_id": reaction_stid, + }) + for eid in input_ids: input_uuid = entity_uuid_registry[(eid, vr_uid, "input")] + if eid in pooled_member: + pathway_logic_network_data.append({ + "source_id": input_uuid, "target_id": pooled_member[eid], + "pos_neg": "pos", "and_or": "or", "edge_type": "set_member", + "stoichiometry": 1, "edge_reaction_id": reaction_stid, + }) + continue pathway_logic_network_data.append({ "source_id": input_uuid, "target_id": vr_uid, diff --git a/src/pathway_generator.py b/src/pathway_generator.py index a0aa03b..2d25364 100755 --- a/src/pathway_generator.py +++ b/src/pathway_generator.py @@ -36,6 +36,7 @@ "LNG_HANDOFF_HUB_MAX", "LNG_SET_MEMBERS_OR", "LNG_SET_POOL", + "LNG_CAP_POOLS", "LNG_DIAGRAM_SET_MEMBER", # 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 From cf6fc0bade2ef0f63206c65ce6dd458e9f72b2f1 Mon Sep 17 00:00:00 2001 From: Adam Wright <668955+adamjohnwright@users.noreply.github.com> Date: Sun, 27 Sep 2026 08:16:47 -0400 Subject: [PATCH 7/8] tests for LNG_CAP_POOLS: bundled alternatives pooled, single chosen member not, flag off, no leak between pathways, modifier sets excluded Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/test_cap_pools.py | 68 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 68 insertions(+) create mode 100644 tests/test_cap_pools.py diff --git a/tests/test_cap_pools.py b/tests/test_cap_pools.py new file mode 100644 index 0000000..6b29743 --- /dev/null +++ b/tests/test_cap_pools.py @@ -0,0 +1,68 @@ +"""LNG_CAP_POOLS (deltasignal specs/036): a set input that resolves to SEVERAL +nodes in one virtual reaction -- its alternatives were bundled by the variant +cap -- is recorded as a set pool, so Phase 3 wires members -> pool -> reaction +and the reaction has one input per curated component. + +Shape: RAF "MAP2Ks and MAPKs bind to the activated RAF complex". Its input set S +("RAF/MAPK scaffolds") is a CandidateSet; over the cap every candidate landed in +one virtual reaction as a separate required input.""" +import pandas as pd +import pytest + +import src.logic_network_generator as m +from src import neo4j_connector + +S, C, M1, M2 = "R-HSA-S", "R-HSA-C", "R-HSA-M1", "R-HSA-M2" + + +@pytest.fixture +def stub(monkeypatch): + monkeypatch.setattr(neo4j_connector, "get_reaction_input_output_ids", + lambda rid, io: {S, C} if io == "input" else {"R-HSA-OUT"}) + monkeypatch.setattr(neo4j_connector, "get_reaction_io_stoichiometry", lambda rid, io: {}) + monkeypatch.setattr(neo4j_connector, "get_labels", + lambda e: ["CandidateSet"] if e == S else ["EntityWithAccessionedSequence"]) + monkeypatch.setattr(m, "_matching_leaves", lambda e: {M1, M2} if e == S else {e}) + monkeypatch.setattr(m, "modifier_isoform_set_ids", lambda: set()) + + +def resolve(present): + uid_index = {"in": ([], set(present) | {C}, {}), "out": ([], {"R-HSA-OUT"}, {})} + rmap = pd.DataFrame({"uid": ["vr1"], "input_hash": ["in"], "output_hash": ["out"], "reactome_id": [1]}) + return m._resolve_vr_entities(rmap, uid_index) + + +def test_bundled_alternatives_become_a_pool(stub, monkeypatch): + monkeypatch.setenv("LNG_CAP_POOLS", "1") + ve = resolve({M1, M2}) + assert m._vr_input_pools == {"vr1": {S: {M1, M2}}} + # members stay in the VR's inputs, so Phase 2 still joins them to producers + assert set(ve["vr1"][0]) == {M1, M2, C} + + +def test_a_single_chosen_member_is_not_a_pool(stub, monkeypatch): + monkeypatch.setenv("LNG_CAP_POOLS", "1") + resolve({M1}) + assert m._vr_input_pools == {} + + +def test_off_records_no_pools(stub, monkeypatch): + monkeypatch.delenv("LNG_CAP_POOLS", raising=False) + resolve({M1, M2}) + assert m._vr_input_pools == {} + + +def test_pools_do_not_leak_between_pathways(stub, monkeypatch): + monkeypatch.setenv("LNG_CAP_POOLS", "1") + resolve({M1, M2}) + assert m._vr_input_pools + resolve({M1}) # the next pathway's resolution starts clean + assert m._vr_input_pools == {} + + +def test_a_modifier_isoform_set_is_not_pooled(stub, monkeypatch): + monkeypatch.setenv("LNG_CAP_POOLS", "1") + monkeypatch.setattr(m, "modifier_isoform_set_ids", lambda: {S}) + monkeypatch.setattr(m, "_matching_leaves", lambda e: {e}) + resolve({M1, M2}) + assert m._vr_input_pools == {} From bddfe7848099de9041fdb4c6ac757176ef87c868 Mon Sep 17 00:00:00 2001 From: Adam Wright <668955+adamjohnwright@users.noreply.github.com> Date: Sun, 27 Sep 2026 08:33:19 -0400 Subject: [PATCH 8/8] Review of combined-037: cap pools only in reactions the variant cap bundled (CAPPED_IDS; an uncapped copy's chosen complex alternative resolves to co-required subunits -- miR-93 RISC in PIP3 was averaged); leaves grouped by alternative, a complex alternative is an AND unit (leaves -assembly-> alternative -set_member-> pool); a leaf another input also maps to stays direct; curated set stoichiometry on the pool edge; depletion onto a pool goes to its members. Phase 3 input emission factored into _emit_vr_inputs and tested; gate mutation-checked. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/logic_network_generator.py | 150 +++++++++++++++++++++------------ src/reaction_generator.py | 5 ++ tests/test_cap_pools.py | 107 ++++++++++++++++------- 3 files changed, 181 insertions(+), 81 deletions(-) diff --git a/src/logic_network_generator.py b/src/logic_network_generator.py index 068d869..25ea14b 100755 --- a/src/logic_network_generator.py +++ b/src/logic_network_generator.py @@ -13,6 +13,7 @@ modifier_isoform_set_ids, get_terminal_components, MAX_VARIANTS, + CAPPED_IDS, _int_env, ) @@ -823,9 +824,10 @@ def _map_annotated_entity_to_nodes(entity_id: str, member_set: Set[str]) -> Set[ return {str(entity_id)} # simple entity (protein / small molecule / …) -# vr_uid -> {input set stId -> member node ids}: set inputs whose alternatives -# were bundled (specs/036, LNG_CAP_POOLS). Filled by _resolve_vr_entities. -_vr_input_pools: Dict[str, Dict[str, Set[str]]] = {} +# vr_uid -> {input set stId -> {"alts": {member stId -> leaf node ids}, +# "stoich": curated coefficient}}: set inputs whose alternatives the variant cap +# bundled (specs/036, LNG_CAP_POOLS). Filled by _resolve_vr_entities. +_vr_input_pools: Dict[str, Dict[str, dict]] = {} def _resolve_vr_entities( @@ -879,27 +881,41 @@ def _resolve_io(reaction_id: str, io: str, members: Set[str]) -> tuple: by_entity = _annotated_stoich(reaction_id, io) node_ids: Set[str] = set() node_stoich: Dict[str, int] = {} - pools: Dict[str, Set[str]] = {} + mapped_by: Dict[str, Set[str]] = {} for e in _annotated(reaction_id, io): s = by_entity.get(str(e), 1) mapped = _map_annotated_entity_to_nodes(str(e), members) - # deltasignal specs/036: in a normal virtual reaction a set input - # resolves to the ONE member this copy chose. It resolves to several - # only when the alternatives were bundled -- by the variant cap - # (issue #40), which merges every alternative into one variant -- and - # then every member became a separate REQUIRED input (RAF's scaffold - # binding: 8 alternative scaffolds, all required). Record the set as - # a pool of those members instead; Phase 3 wires members -> pool -> - # reaction, so the reaction has one input per curated component. - if (cap_pools and io == "input" and len(mapped) > 1 - and str(e) not in modifier_isoform_set_ids()): - from src.neo4j_connector import get_labels - if any(x in (get_labels(str(e)) or []) - for x in ("EntitySet", "DefinedSet", "CandidateSet")): - pools[str(e)] = set(mapped) + mapped_by[str(e)] = set(mapped) for n in mapped: node_ids.add(n) node_stoich[n] = node_stoich.get(n, 0) + s + # deltasignal specs/036: the variant cap (issue #40) merges every + # alternative of every set into ONE variant, so each alternative's + # leaves became separate REQUIRED inputs (RAF's scaffold binding: 8 + # alternative scaffolds, all required at once). Only for a reaction the + # cap actually bundled (CAPPED_IDS; an uncapped copy that chose one + # complex alternative also resolves to several leaves -- its subunits -- + # and those ARE co-required), group each set input's leaves by the + # member (alternative) they belong to. With >= 2 alternatives present + # the set is a pool: each alternative is an AND unit of its leaves, the + # pool ORs the alternatives, and the reaction reads the pool once. + # A leaf another annotated input also maps to stays a direct input. + pools: Dict[str, dict] = {} + if cap_pools and io == "input" and str(reaction_id) in CAPPED_IDS: + from src.neo4j_connector import get_labels, get_set_members + for e, mapped in mapped_by.items(): + if len(mapped) < 2 or e in modifier_isoform_set_ids(): + continue + if not any(x in (get_labels(e) or []) for x in ("EntitySet", "DefinedSet", "CandidateSet")): + continue + elsewhere = set().union(*(v for k, v in mapped_by.items() if k != e)) + alts: Dict[str, Set[str]] = {} + for mem in sorted(get_set_members(e) or []): + lv = (mapped & _matching_leaves(str(mem))) - elsewhere + if lv: + alts[str(mem)] = lv + if len(alts) >= 2: + pools[e] = {"alts": alts, "stoich": by_entity.get(e, 1)} return list(node_ids), node_stoich, pools _vr_input_pools.clear() # one pathway per call; the dict is module state @@ -1359,6 +1375,12 @@ def _pool_contains(cat_uuid: str, uid: str, sid: str) -> bool: inputs = by_target_inputs.get(rxn_uuid, []) if not catalyst_uuids or not inputs: continue + # specs/036: a capped input set arrives as a pool; deplete its members, + # which is where the substrate's own nodes are (as before pooling). + expanded: List[str] = [] + for inp in inputs: + expanded.extend(sorted(pool_members[inp][0]) if inp in pool_members else [inp]) + inputs = expanded for cat_uuid in catalyst_uuids: cat_stid = reactome_id_to_uuid.get(cat_uuid, "") for inp_uuid in inputs: @@ -2246,6 +2268,61 @@ def _print_regulator_statistics( ) +def _emit_vr_inputs(vr_uid: str, input_ids, input_stoich: Dict[str, int], pools: Dict[str, dict], + entity_uuid_registry: Dict[tuple, str], reactome_id_to_uuid: Dict[str, str], + pathway_logic_network_data: List[Dict[str, Any]], reaction_stid) -> None: + """Phase 3 input edges of one virtual reaction, with specs/036 cap pools.""" + # specs/036: a capped set input becomes ONE pool node feeding the + # reaction once. Each alternative with one leaf feeds the pool directly + # (set_member, OR); an alternative with several leaves (a complex member) + # is an AND unit: leaves -> alternative node (assembly) -> pool. + pooled: Set[str] = set() + for set_stid, spec in sorted(pools.items()): + pool_uuid = str(uuid.uuid4()) + reactome_id_to_uuid[pool_uuid] = set_stid + pathway_logic_network_data.append({ + "source_id": pool_uuid, "target_id": vr_uid, "pos_neg": "pos", + "and_or": "and", "edge_type": "input", + "stoichiometry": spec["stoich"], "edge_reaction_id": reaction_stid, + }) + for mem, leaves in sorted(spec["alts"].items()): + if len(leaves) == 1: + feed = pool_uuid + alt_uuid = None + else: + alt_uuid = str(uuid.uuid4()) + reactome_id_to_uuid[alt_uuid] = mem + pathway_logic_network_data.append({ + "source_id": alt_uuid, "target_id": pool_uuid, "pos_neg": "pos", + "and_or": "or", "edge_type": "set_member", "stoichiometry": 1, + "edge_reaction_id": reaction_stid, + }) + feed = alt_uuid + for leaf in sorted(leaves): + pathway_logic_network_data.append({ + "source_id": entity_uuid_registry[(leaf, vr_uid, "input")], + "target_id": feed, "pos_neg": "pos", + "and_or": "or" if alt_uuid is None else "and", + "edge_type": "set_member" if alt_uuid is None else "assembly", + "stoichiometry": 1, "edge_reaction_id": reaction_stid, + }) + pooled.add(leaf) + + for eid in input_ids: + input_uuid = entity_uuid_registry[(eid, vr_uid, "input")] + if eid in pooled: + continue # it reaches the reaction through its pool + pathway_logic_network_data.append({ + "source_id": input_uuid, + "target_id": vr_uid, + "pos_neg": "pos", + "and_or": "and", + "edge_type": "input", + "stoichiometry": input_stoich.get(eid, 1), + "edge_reaction_id": reaction_stid, + }) + + def create_pathway_logic_network( decomposed_uid_mapping: pd.DataFrame, reaction_connections: pd.DataFrame, @@ -2482,40 +2559,9 @@ def create_pathway_logic_network( continue reaction_stid = vr_to_reaction.get(str(vr_uid)) - # specs/036: bundled set alternatives feed ONE pool node per set, which - # feeds the reaction once (members -> pool by set_member, OR). - pools = _vr_input_pools.get(vr_uid, {}) - pooled_member: Dict[str, str] = {} - for set_stid, mem in sorted(pools.items()): - pool_uuid = str(uuid.uuid4()) - reactome_id_to_uuid[pool_uuid] = set_stid - for m in mem: - pooled_member.setdefault(m, pool_uuid) - pathway_logic_network_data.append({ - "source_id": pool_uuid, "target_id": vr_uid, "pos_neg": "pos", - "and_or": "and", "edge_type": "input", - "stoichiometry": input_stoich.get(set_stid, 1), - "edge_reaction_id": reaction_stid, - }) - - for eid in input_ids: - input_uuid = entity_uuid_registry[(eid, vr_uid, "input")] - if eid in pooled_member: - pathway_logic_network_data.append({ - "source_id": input_uuid, "target_id": pooled_member[eid], - "pos_neg": "pos", "and_or": "or", "edge_type": "set_member", - "stoichiometry": 1, "edge_reaction_id": reaction_stid, - }) - continue - pathway_logic_network_data.append({ - "source_id": input_uuid, - "target_id": vr_uid, - "pos_neg": "pos", - "and_or": "and", - "edge_type": "input", - "stoichiometry": input_stoich.get(eid, 1), - "edge_reaction_id": reaction_stid, - }) + _emit_vr_inputs(vr_uid, input_ids, input_stoich, _vr_input_pools.get(vr_uid, {}), + entity_uuid_registry, reactome_id_to_uuid, + pathway_logic_network_data, reaction_stid) for eid in output_ids: output_uuid = entity_uuid_registry[(eid, vr_uid, "output")] diff --git a/src/reaction_generator.py b/src/reaction_generator.py index 98404dc..801dce4 100755 --- a/src/reaction_generator.py +++ b/src/reaction_generator.py @@ -95,6 +95,10 @@ def _int_env(name: str, default: str) -> int: MAX_VARIANTS = _int_env("LNG_MAX_VARIANTS", "512") +# Ids whose set combinations exceeded MAX_VARIANTS and were bundled into one +# variant (deltasignal specs/036): LNG_CAP_POOLS pools only these. +CAPPED_IDS: Set[str] = set() + class _DecompositionStore: """Append-mostly buffer for decomposition rows with O(1) lookups. @@ -288,6 +292,7 @@ def get_broken_apart_ids( merged: Set[str] = set() for member in new_broken_apart_members: merged |= member + CAPPED_IDS.add(str(reactome_id)) logger.warning( f"variant cap hit for {reactome_id}: {cartesian_size} " f"combinations > {MAX_VARIANTS}; bundling " diff --git a/tests/test_cap_pools.py b/tests/test_cap_pools.py index 6b29743..c34f72e 100644 --- a/tests/test_cap_pools.py +++ b/tests/test_cap_pools.py @@ -1,68 +1,117 @@ -"""LNG_CAP_POOLS (deltasignal specs/036): a set input that resolves to SEVERAL -nodes in one virtual reaction -- its alternatives were bundled by the variant -cap -- is recorded as a set pool, so Phase 3 wires members -> pool -> reaction -and the reaction has one input per curated component. +"""LNG_CAP_POOLS (deltasignal specs/036): in a reaction the variant cap BUNDLED +(every alternative of every set merged into one variant), a set input becomes +one pool: each alternative is an AND unit of its leaves, the pool ORs the +alternatives, and the reaction reads the pool once. Shape: RAF "MAP2Ks and MAPKs bind to the activated RAF complex". Its input set S -("RAF/MAPK scaffolds") is a CandidateSet; over the cap every candidate landed in -one virtual reaction as a separate required input.""" +("RAF/MAPK scaffolds") has alternatives A (a single protein) and K (a complex, +leaves K1, K2); over the cap all three leaves were separate required inputs. + +Review of #combined-037: the first version pooled ANY set resolving to several +nodes, which also fires in UNCAPPED reactions where the one chosen alternative +is a complex (miR-93 RISC in PIP3: its four subunits were averaged, so a miR-93 +knockout read 0.75x instead of 0).""" import pandas as pd import pytest import src.logic_network_generator as m from src import neo4j_connector -S, C, M1, M2 = "R-HSA-S", "R-HSA-C", "R-HSA-M1", "R-HSA-M2" +S, C, A, K, K1, K2 = "R-HSA-S", "R-HSA-C", "R-HSA-A", "R-HSA-K", "R-HSA-K1", "R-HSA-K2" +LEAVES = {S: {A, K1, K2}, A: {A}, K: {K1, K2}, C: {C}} @pytest.fixture def stub(monkeypatch): monkeypatch.setattr(neo4j_connector, "get_reaction_input_output_ids", lambda rid, io: {S, C} if io == "input" else {"R-HSA-OUT"}) - monkeypatch.setattr(neo4j_connector, "get_reaction_io_stoichiometry", lambda rid, io: {}) + monkeypatch.setattr(neo4j_connector, "get_reaction_io_stoichiometry", + lambda rid, io: {S: 2} if io == "input" else {}) monkeypatch.setattr(neo4j_connector, "get_labels", lambda e: ["CandidateSet"] if e == S else ["EntityWithAccessionedSequence"]) - monkeypatch.setattr(m, "_matching_leaves", lambda e: {M1, M2} if e == S else {e}) + monkeypatch.setattr(neo4j_connector, "get_set_members", lambda e: [A, K] if e == S else []) + monkeypatch.setattr(m, "_matching_leaves", lambda e: LEAVES.get(e, {e})) monkeypatch.setattr(m, "modifier_isoform_set_ids", lambda: set()) + monkeypatch.setattr(m, "CAPPED_IDS", {"1"}) -def resolve(present): +def resolve(present, rid=1): uid_index = {"in": ([], set(present) | {C}, {}), "out": ([], {"R-HSA-OUT"}, {})} - rmap = pd.DataFrame({"uid": ["vr1"], "input_hash": ["in"], "output_hash": ["out"], "reactome_id": [1]}) + rmap = pd.DataFrame({"uid": ["vr1"], "input_hash": ["in"], "output_hash": ["out"], "reactome_id": [rid]}) return m._resolve_vr_entities(rmap, uid_index) -def test_bundled_alternatives_become_a_pool(stub, monkeypatch): +def test_a_capped_bundle_becomes_a_pool_of_alternatives(stub, monkeypatch): monkeypatch.setenv("LNG_CAP_POOLS", "1") - ve = resolve({M1, M2}) - assert m._vr_input_pools == {"vr1": {S: {M1, M2}}} - # members stay in the VR's inputs, so Phase 2 still joins them to producers - assert set(ve["vr1"][0]) == {M1, M2, C} + ve = resolve({A, K1, K2}) + assert m._vr_input_pools == {"vr1": {S: {"alts": {A: {A}, K: {K1, K2}}, "stoich": 2}}} + assert set(ve["vr1"][0]) == {A, K1, K2, C} # members stay VR inputs (Phase 2 joins) -def test_a_single_chosen_member_is_not_a_pool(stub, monkeypatch): +def test_an_uncapped_reaction_is_never_pooled(stub, monkeypatch): + # the miR-93 case: one chosen alternative that is a complex resolves to its + # subunits, which ARE co-required monkeypatch.setenv("LNG_CAP_POOLS", "1") - resolve({M1}) + resolve({K1, K2}, rid=2) assert m._vr_input_pools == {} -def test_off_records_no_pools(stub, monkeypatch): - monkeypatch.delenv("LNG_CAP_POOLS", raising=False) - resolve({M1, M2}) +def test_the_cap_gate_itself(stub, monkeypatch): + # two alternatives present in a reaction the cap did NOT bundle: not pooled + # (only a capped reaction merged its alternatives) + monkeypatch.setenv("LNG_CAP_POOLS", "1") + resolve({A, K1, K2}, rid=2) assert m._vr_input_pools == {} -def test_pools_do_not_leak_between_pathways(stub, monkeypatch): +def test_one_alternative_present_is_not_a_pool(stub, monkeypatch): monkeypatch.setenv("LNG_CAP_POOLS", "1") - resolve({M1, M2}) - assert m._vr_input_pools - resolve({M1}) # the next pathway's resolution starts clean + resolve({K1, K2}) assert m._vr_input_pools == {} -def test_a_modifier_isoform_set_is_not_pooled(stub, monkeypatch): +def test_a_leaf_another_input_also_maps_to_stays_direct(stub, monkeypatch): monkeypatch.setenv("LNG_CAP_POOLS", "1") - monkeypatch.setattr(m, "modifier_isoform_set_ids", lambda: {S}) - monkeypatch.setattr(m, "_matching_leaves", lambda e: {e}) - resolve({M1, M2}) + monkeypatch.setattr(m, "_matching_leaves", lambda e: {A, K1, K2} if e == S else ({K1, K2, C} if e == C else LEAVES.get(e, {e}))) + monkeypatch.setattr(neo4j_connector, "get_labels", + lambda e: ["CandidateSet"] if e in (S,) else (["Complex"] if e == C else ["EntityWithAccessionedSequence"])) + monkeypatch.setattr(m, "_map_annotated_entity_to_nodes", + lambda e, mem: {A, K1, K2} & mem if e == S else ({K1} if e == C else {e})) + resolve({A, K1, K2}) + alts = m._vr_input_pools["vr1"][S]["alts"] + assert alts == {A: {A}, K: {K2}} # K1 is also C's node: not pooled + + +def test_off_and_no_leak_between_pathways(stub, monkeypatch): + monkeypatch.delenv("LNG_CAP_POOLS", raising=False) + resolve({A, K1, K2}) + assert m._vr_input_pools == {} + monkeypatch.setenv("LNG_CAP_POOLS", "1") + resolve({A, K1, K2}) + assert m._vr_input_pools + resolve({A}, rid=2) assert m._vr_input_pools == {} + + +def test_emission_wires_alternatives_as_and_units_into_one_pool(): + reg = {(x, "vr1", "input"): f"u-{x}" for x in (A, K1, K2, C)} + r2u, data = {}, [] + pools = {S: {"alts": {A: {A}, K: {K1, K2}}, "stoich": 2}} + m._emit_vr_inputs("vr1", [A, K1, K2, C], {C: 1}, pools, reg, r2u, data, "R-HSA-RX") + into_vr = [(e["source_id"], e["edge_type"], e["stoichiometry"]) for e in data if e["target_id"] == "vr1"] + pool = next(u for u, s in r2u.items() if s == S) + alt = next(u for u, s in r2u.items() if s == K) + assert sorted(into_vr) == sorted([(pool, "input", 2), ("u-" + C, "input", 1)]) # one input per component + assert {(e["source_id"], e["and_or"], e["edge_type"]) for e in data if e["target_id"] == pool} == \ + {("u-" + A, "or", "set_member"), (alt, "or", "set_member")} + assert {(e["source_id"], e["and_or"], e["edge_type"]) for e in data if e["target_id"] == alt} == \ + {("u-" + K1, "and", "assembly"), ("u-" + K2, "and", "assembly")} + assert not any(e["target_id"] == "vr1" and e["source_id"] in ("u-" + A, "u-" + K1, "u-" + K2) for e in data) + + +def test_without_pools_emission_is_the_plain_input_fan_in(): + reg = {(x, "vr1", "input"): f"u-{x}" for x in (A, C)} + data = [] + m._emit_vr_inputs("vr1", [A, C], {}, {}, reg, {}, data, "R-HSA-RX") + assert sorted(e["source_id"] for e in data) == ["u-" + A, "u-" + C] + assert {e["edge_type"] for e in data} == {"input"}