From d31f95eae84aeecfafeeb4366c659b5bd7250183 Mon Sep 17 00:00:00 2001 From: Adam Wright <668955+adamjohnwright@users.noreply.github.com> Date: Thu, 22 Oct 2026 15:41:09 -0400 Subject: [PATCH 1/2] LNG_UB_LIGASE_UNIT (deltasignal specs/089): a ubiquitination does not deplete its own ligase - a leaf of the curated activeUnit, or of a complex released unchanged, that leaves the reaction unmodified is not the substrate; default off Co-Authored-By: Claude Opus 5.5 --- PIPELINE_SETTINGS.md | 1 + src/env_flags.py | 1 + src/logic_network_generator.py | 81 +++++++++++++++++++++---- tests/test_ub_ligase_unit.py | 107 +++++++++++++++++++++++++++++++++ 4 files changed, 177 insertions(+), 13 deletions(-) create mode 100644 tests/test_ub_ligase_unit.py diff --git a/PIPELINE_SETTINGS.md b/PIPELINE_SETTINGS.md index 1236cc1..2bb595a 100644 --- a/PIPELINE_SETTINGS.md +++ b/PIPELINE_SETTINGS.md @@ -74,6 +74,7 @@ what NOT to touch. Overriding is for A/B benchmarking only. | `LNG_CAP_POOLS` | `1` | past the variant cap, one OR pool per set instead of an all-required bundle (deltasignal specs/045) | | `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` | `0` | 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_CATALYST_BUNDLE` | `0` | legacy; subsumed by `LNG_COMPLEX_AS_NODE` | | `LNG_HANDOFF_EDGES` | `0` | precedingEvent hand-off bridges (net-negative, off) | diff --git a/src/env_flags.py b/src/env_flags.py index cf0ea57..e72037c 100644 --- a/src/env_flags.py +++ b/src/env_flags.py @@ -34,6 +34,7 @@ "LNG_SET_EXPAND": True, "LNG_SET_MEMBERS_OR": False, "LNG_SET_POOL": True, + "LNG_UB_LIGASE_UNIT": False, # deltasignal specs/089 "LNG_VARIANT_NODES": True, # default since deltasignal specs/046 (vn7: held-out +130, experimental -10 ns) } diff --git a/src/logic_network_generator.py b/src/logic_network_generator.py index cc512f9..f955599 100755 --- a/src/logic_network_generator.py +++ b/src/logic_network_generator.py @@ -1444,6 +1444,63 @@ def base_of(e: Dict[str, Any]) -> str: f"{sum(len(set(v)) for v in pooled.values())} catalyst variants") +def _ub_substrates(in_leaves: List[Tuple[Optional[str], Optional[List[str]]]], + output_genes: Set[str], output_leaves: Set[str], + ligase_leaves: Set[str]) -> Tuple[Set[str], Set[str]]: + """A ubiquitination's substrate: the input leaves whose gene also appears + in the output (the modified form), as (genes, leaf stIds). + + deltasignal specs/089: a leaf of the ligase (``ligase_leaves``) is not the + substrate when its own stId is also an output leaf, i.e. it leaves the + reaction unmodified. Without this, SCF(Skp2):Cks1 drained its own free + CKS1B. A ligase that is itself modified (autoubiquitination: Ub-BRAP has + another stId) stays a substrate, and so do the substrate's partners + (EGFR and GRB2 under CBL, specs/078).""" + subst_stids: Set[str] = set() + subst_genes: Set[str] = set() + if not output_genes: + return subst_genes, subst_stids + for leaf_stid, leaf_genes in in_leaves: + if not leaf_stid or not leaf_genes: + continue + if leaf_stid in ligase_leaves and leaf_stid in output_leaves: + continue + common = output_genes & set(leaf_genes) + if common: + subst_stids.add(leaf_stid) + subst_genes.update(common) + return subst_genes, subst_stids + + +def _ub_ligase_leaves(rstids: List[str]) -> Dict[str, Set[str]]: + """The ligase's leaf stIds in each ubiquitination (deltasignal specs/089): + the leaves of its catalyst activity's curated ``activeUnit``; or, where no + activeUnit is curated, the leaves of an output entity that is also a direct + component of an input complex (the ligase released unchanged, as + CUL1:SKP1:SKP2:CKS1B from R-HSA-1363331).""" + from src.neo4j_connector import get_graph + leaf = ("OPTIONAL MATCH (x)-[:hasComponent|hasMember|hasCandidate*0..3]->(l:PhysicalEntity) " + "RETURN id AS rxn, l.stId AS leaf_stid") + au_rows = get_graph().run( + "UNWIND $ids AS id " + "MATCH (:ReactionLikeEvent {stId: id})-[:catalystActivity]->(:CatalystActivity)-[:activeUnit]->(x) " + + leaf, ids=rstids).data() + out: Dict[str, Set[str]] = {} + for r in au_rows: + if r["leaf_stid"]: + out.setdefault(r["rxn"], set()).add(r["leaf_stid"]) + rest = [r for r in rstids if r not in out] + if rest: + rel_rows = get_graph().run( + "UNWIND $ids AS id " + "MATCH (rx:ReactionLikeEvent {stId: id})-[:output]->(x)<-[:hasComponent]-(:Complex)<-[:input]-(rx) " + + leaf, ids=rest).data() + for r in rel_rows: + if r["leaf_stid"]: + out.setdefault(r["rxn"], set()).add(r["leaf_stid"]) + return out + + def _emit_substrate_depletion_edges( pathway_logic_network_data: List[Dict[str, Any]], reactome_id_to_uuid: Dict[str, str], @@ -1556,32 +1613,30 @@ def _emit_substrate_depletion_edges( "WHERE NOT o.stId IN $ub_stids " "OPTIONAL MATCH (o)-[:hasComponent|hasMember|hasCandidate*0..3]->(leaf:PhysicalEntity) " "OPTIONAL MATCH (leaf)-[:referenceEntity]->(re:ReferenceEntity) " - "RETURN id AS rxn, re.geneName AS genes", + "RETURN id AS rxn, leaf.stId AS leaf_stid, re.geneName AS genes", ids=ubiq_rstids, ub_stids=list(_UBIQUITIN_STIDS), ).data() if ubiq_rstids else [] + # deltasignal specs/089: the ligase's own leaves, which a + # ubiquitination carries through rather than modifies. + ligase_by_rxn = _ub_ligase_leaves(ubiq_rstids) \ + if ubiq_rstids and env_flag("LNG_UB_LIGASE_UNIT") else {} in_by_rxn: Dict[str, List[Tuple[Optional[str], Optional[List[str]]]]] = {} for r in in_rows: in_by_rxn.setdefault(r["rxn"], []).append((r["leaf_stid"], r["genes"])) out_by_rxn: Dict[str, Set[str]] = {} + out_leaves_by_rxn: Dict[str, Set[str]] = {} for r in out_rows: + if r.get("leaf_stid"): + out_leaves_by_rxn.setdefault(r["rxn"], set()).add(r["leaf_stid"]) g = r["genes"] if g: for gn in g: out_by_rxn.setdefault(r["rxn"], set()).add(gn) for rxn in ubiq_rstids: - output_genes = out_by_rxn.get(rxn, set()) - if not output_genes: continue - # The substrate: input-leaf proteins whose gene also appears - # in the output (modified form). Collect their leaf stids. - subst_stids: Set[str] = set() - subst_genes: Set[str] = set() - for leaf_stid, leaf_genes in in_by_rxn.get(rxn, []): - if not leaf_stid or not leaf_genes: continue - common = output_genes & set(leaf_genes) - if common: - subst_stids.add(leaf_stid) - subst_genes.update(common) + subst_genes, subst_stids = _ub_substrates( + in_by_rxn.get(rxn, []), out_by_rxn.get(rxn, set()), + out_leaves_by_rxn.get(rxn, set()), ligase_by_rxn.get(rxn, set())) if subst_stids: ubiquitin_subst_by_rstid[rxn] = (subst_genes, subst_stids) except Exception as exc: diff --git a/tests/test_ub_ligase_unit.py b/tests/test_ub_ligase_unit.py new file mode 100644 index 0000000..72cce77 --- /dev/null +++ b/tests/test_ub_ligase_unit.py @@ -0,0 +1,107 @@ +"""deltasignal specs/089: a ubiquitination does not deplete its own ligase +(LNG_UB_LIGASE_UNIT).""" +from typing import Any, Dict + +import pandas as pd +import pytest + +import src.logic_network_generator as lng +import src.neo4j_connector as nc + +UB = "R-HSA-113595" +RX = "R-HSA-1363331" # SCF(Skp2) ubiquitinates p130 and is released +CKS1B, RBL2, UB_RBL2 = "R-HSA-CKS1B", "R-HSA-RBL2", "R-HSA-UB-RBL2" + + +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} + + +# --- the substrate rule on its own ------------------------------------------ + +def test_a_ligase_leaf_carried_unchanged_is_not_the_substrate(): + genes, stids = lng._ub_substrates([(CKS1B, ["CKS1B"]), (RBL2, ["RBL2"])], + {"CKS1B", "RBL2"}, {CKS1B, UB_RBL2}, {CKS1B}) + assert stids == {RBL2} and genes == {"RBL2"} + + +def test_without_ligase_leaves_every_shared_gene_is_the_substrate(): + _, stids = lng._ub_substrates([(CKS1B, ["CKS1B"]), (RBL2, ["RBL2"])], + {"CKS1B", "RBL2"}, {CKS1B, UB_RBL2}, set()) + assert stids == {CKS1B, RBL2} + + +def test_an_autoubiquitinated_ligase_stays_the_substrate(): + # BRAP autoubiquitinates: BRAP is the active unit, but its output form is + # Ub-BRAP, another stId, so it is modified and still depleted. + _, stids = lng._ub_substrates([("R-HSA-BRAP", ["BRAP"])], {"BRAP"}, + {"R-HSA-UB-BRAP"}, {"R-HSA-BRAP"}) + assert stids == {"R-HSA-BRAP"} + + +def test_the_substrates_partners_are_kept(): + # CBL ubiquitinates EGFR: CBL (active unit) is dropped, the receptor complex + # partners are still depleted (specs/078 lost the CBL class by dropping them). + ins = [("R-HSA-CBL", ["CBL"]), ("R-HSA-EGFR", ["EGFR"]), ("R-HSA-GRB2", ["GRB2"])] + _, stids = lng._ub_substrates(ins, {"CBL", "EGFR", "GRB2"}, + {"R-HSA-CBL", "R-HSA-EGFR", "R-HSA-GRB2"}, {"R-HSA-CBL"}) + assert stids == {"R-HSA-EGFR", "R-HSA-GRB2"} + + +# --- the whole pass, against a stub graph ------------------------------------ + +class _Result: + def __init__(self, rows): self.rows = rows + def data(self): return self.rows + + +class _Graph: + """Answers the ubiquitin pass's queries for one reaction: catalyst C + (p-RBL2:SCF(Skp2):Cks1) + Ub -> Ub-p-RBL2 + the released ligase, no activeUnit.""" + def run(self, q: str, **kw: Any) -> _Result: + if "activeUnit" in q: + return _Result([]) + if "<-[:hasComponent]-(:Complex)<-[:input]-" in q: + return _Result([{"rxn": RX, "leaf_stid": CKS1B}]) + if "-[:output]->" in q: + return _Result([{"rxn": RX, "leaf_stid": CKS1B, "genes": ["CKS1B"]}, + {"rxn": RX, "leaf_stid": UB_RBL2, "genes": ["RBL2"]}]) + if "-[:input]->(inp" in q: + return _Result([{"rxn": RX, "leaf_stid": CKS1B, "genes": ["CKS1B"]}, + {"rxn": RX, "leaf_stid": RBL2, "genes": ["RBL2"]}]) + if "ub.stId IN $ub_stids" in q: + return _Result([{"rxn": RX}]) + raise AssertionError(q) + + +@pytest.fixture +def pass_(monkeypatch): + monkeypatch.setattr(nc, "get_graph", lambda: _Graph()) + monkeypatch.setattr(lng, "_cofactor_stids", lambda: frozenset()) + monkeypatch.setenv("LNG_VARIANT_NODES", "0") + + def run() -> set: + data = [_e("u-c", "rx", "catalyst"), _e("u-c", "rx", "input"), _e("rx", "u-out", "output"), + _e("u-cks", "x1", "input"), _e("u-rbl2", "x2", "input")] + r2u: Dict[str, str] = {"u-c": "R-HSA-1363326", "u-out": UB_RBL2, + "u-cks": CKS1B, "u-rbl2": RBL2} + cmap = pd.DataFrame([{"reaction_uuid": "rx", "reaction_id": RX}]) + lng._emit_substrate_depletion_edges(data, r2u, cmap) + return {(e["source_id"], e["target_id"]) for e in data if e["edge_type"] == "depletion"} + return run + + +def test_off_the_ligase_drains_its_own_subunit(pass_, monkeypatch): + monkeypatch.delenv("LNG_UB_LIGASE_UNIT", raising=False) + assert pass_() == {("u-c", "u-cks"), ("u-c", "u-rbl2")} + + +def test_on_only_the_substrate_is_drained(pass_, monkeypatch): + monkeypatch.setenv("LNG_UB_LIGASE_UNIT", "1") + assert pass_() == {("u-c", "u-rbl2")} + + +def test_the_switch_is_off_by_default(monkeypatch): + monkeypatch.delenv("LNG_UB_LIGASE_UNIT", raising=False) + assert lng.env_flag("LNG_UB_LIGASE_UNIT") is False From 90836b02bb6e67cf1038cb84445bfae10fa8d2ae Mon Sep 17 00:00:00 2001 From: Adam Wright <668955+adamjohnwright@users.noreply.github.com> Date: Thu, 22 Oct 2026 19:58:57 -0400 Subject: [PATCH 2/2] LNG_UB_LIGASE_UNIT on by default (deltasignal specs/089, adopted by Adam 2026-10-22: tuning +22, experimental +3, held-out -2 traced) Co-Authored-By: Claude Opus 5.5 --- PIPELINE_SETTINGS.md | 2 +- src/env_flags.py | 2 +- tests/test_ub_ligase_unit.py | 7 ++++--- 3 files changed, 6 insertions(+), 5 deletions(-) diff --git a/PIPELINE_SETTINGS.md b/PIPELINE_SETTINGS.md index 2bb595a..2f51fbd 100644 --- a/PIPELINE_SETTINGS.md +++ b/PIPELINE_SETTINGS.md @@ -74,7 +74,7 @@ what NOT to touch. Overriding is for A/B benchmarking only. | `LNG_CAP_POOLS` | `1` | past the variant cap, one OR pool per set instead of an all-required bundle (deltasignal specs/045) | | `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` | `0` | 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_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_CATALYST_BUNDLE` | `0` | legacy; subsumed by `LNG_COMPLEX_AS_NODE` | | `LNG_HANDOFF_EDGES` | `0` | precedingEvent hand-off bridges (net-negative, off) | diff --git a/src/env_flags.py b/src/env_flags.py index e72037c..14ff1bd 100644 --- a/src/env_flags.py +++ b/src/env_flags.py @@ -34,7 +34,7 @@ "LNG_SET_EXPAND": True, "LNG_SET_MEMBERS_OR": False, "LNG_SET_POOL": True, - "LNG_UB_LIGASE_UNIT": False, # deltasignal specs/089 + "LNG_UB_LIGASE_UNIT": True, # default since deltasignal specs/089 (tuning +22, experimental +3, held-out -2 traced) "LNG_VARIANT_NODES": True, # default since deltasignal specs/046 (vn7: held-out +130, experimental -10 ns) } diff --git a/tests/test_ub_ligase_unit.py b/tests/test_ub_ligase_unit.py index 72cce77..0d9cf22 100644 --- a/tests/test_ub_ligase_unit.py +++ b/tests/test_ub_ligase_unit.py @@ -93,7 +93,7 @@ def run() -> set: def test_off_the_ligase_drains_its_own_subunit(pass_, monkeypatch): - monkeypatch.delenv("LNG_UB_LIGASE_UNIT", raising=False) + monkeypatch.setenv("LNG_UB_LIGASE_UNIT", "0") assert pass_() == {("u-c", "u-cks"), ("u-c", "u-rbl2")} @@ -102,6 +102,7 @@ def test_on_only_the_substrate_is_drained(pass_, monkeypatch): assert pass_() == {("u-c", "u-rbl2")} -def test_the_switch_is_off_by_default(monkeypatch): +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 False + assert lng.env_flag("LNG_UB_LIGASE_UNIT") is True + assert pass_() == {("u-c", "u-rbl2")}