From 83a71a231f4ffebc167655fa1d9ec72965458c6f Mon Sep 17 00:00:00 2001 From: Adam Wright <668955+adamjohnwright@users.noreply.github.com> Date: Sat, 3 Oct 2026 11:12:21 -0400 Subject: [PATCH] F7: a set inside a root complex is one node fed by its members (OR), not every member a required input _decompose_hier flattened a set component into its members, each emitted as an AND assembly input of the complex, so a knockout of one alternative removed the complex (VAV1 effectors:GDP needed all five GTPases). A set component is now one node: members feed it over set_member/OR edges (built like components: an existing node, a nested complex, a nested set, or a leaf) and it is the complex's AND input. All in boundary_edges.csv. How members combine is the consumer's set-pool rule (deltasignal specs/033, default product), so this is a structural change; deltasignal specs/044 measures whether predictions move. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/logic_network_generator.py | 56 ++++++++++++++++++++++++-- tests/test_boundary_set_nodes.py | 68 ++++++++++++++++++++++++++++++++ 2 files changed, 120 insertions(+), 4 deletions(-) create mode 100644 tests/test_boundary_set_nodes.py diff --git a/src/logic_network_generator.py b/src/logic_network_generator.py index ecfdd50..826d338 100755 --- a/src/logic_network_generator.py +++ b/src/logic_network_generator.py @@ -1960,7 +1960,7 @@ def _is_complex(entity_id: str) -> bool: # alpha/beta perturbation upstream of ISGF3 was severed there (200 held-out # cases). The downstream-reuse rule (specs/018) applies unchanged. hierarchy = env_flag("LNG_BOUNDARY_HIERARCHY") # default since deltasignal specs/030 (held-out +197, re-measured) - from src.neo4j_connector import get_complex_components + from src.neo4j_connector import get_complex_components, get_set_members nested_registry: Dict[tuple, str] = {} seen_edges: Set[tuple] = set() assembly_count = 0 @@ -1985,14 +1985,15 @@ def _existing_upstream(comp_stid: str, root_uuid: str): roots = [c for c in ok if c not in targets] return (roots or ok)[0] - def _emit(src: str, dst: str) -> None: + def _emit(src: str, dst: str, kind: str = "assembly") -> None: nonlocal assembly_count if (src, dst) in seen_edges: return seen_edges.add((src, dst)) pathway_logic_network_data.append({ - "source_id": src, "target_id": dst, "pos_neg": "pos", "and_or": "and", - "edge_type": "assembly", "stoichiometry": 1, + "source_id": src, "target_id": dst, "pos_neg": "pos", + "and_or": "or" if kind == "set_member" else "and", + "edge_type": kind, "stoichiometry": 1, }) assembly_count += 1 if hierarchy: @@ -2019,10 +2020,57 @@ def _decompose_hier(container_uuid: str, container_stid: str, root_uuid: str, de nested_built += 1 _decompose_hier(nested_registry[key], comp, root_uuid, depth + 1) _emit(nested_registry[key], container_uuid) + elif _is_set(comp) and depth < 6: + _emit(_set_node(comp, root_uuid, depth), container_uuid) else: for leaf in sorted(get_terminal_components(comp)): _emit(_existing_upstream(leaf, root_uuid) or _leaf_uuid(leaf, root_uuid), container_uuid) + # A SET component is "any one of its members" (LNG code review F7, + # deltasignal specs/044). It used to be flattened into its members, each a + # required AND input of the complex, so losing one alternative removed the + # complex. It is now one node, fed by its members over `set_member` (OR) + # edges, and that node is the complex's input. The consumer's set-pool rule + # (deltasignal specs/033) decides how members combine. Members are built + # the way components are: a node the network already has, a nested complex, + # a nested set, or a leaf. Modifier-isoform sets stay atomic, as before. + def _set_node(set_stid: str, root_uuid: str, depth: int) -> str: + nonlocal nested_built + key = (root_uuid, set_stid) + if key in nested_registry: + return nested_registry[key] + set_uuid = nested_registry[key] = str(uuid.uuid4()) + reactome_id_to_uuid[set_uuid] = set_stid + nested_built += 1 + for m in sorted(get_set_members(set_stid) or {}): + existing = _existing_upstream(m, root_uuid) + if existing is not None: + _emit(existing, set_uuid, "set_member") + elif _is_complex(m) and depth < 5: + mkey = (root_uuid, m) + if mkey not in nested_registry: + nested_registry[mkey] = str(uuid.uuid4()) + reactome_id_to_uuid[nested_registry[mkey]] = m + nested_built += 1 + _decompose_hier(nested_registry[mkey], m, root_uuid, depth + 2) + _emit(nested_registry[mkey], set_uuid, "set_member") + elif _is_set(m) and depth < 5: + _emit(_set_node(m, root_uuid, depth + 1), set_uuid, "set_member") + else: + for leaf in sorted(get_terminal_components(m)): + _emit(_existing_upstream(leaf, root_uuid) or _leaf_uuid(leaf, root_uuid), + set_uuid, "set_member") + return set_uuid + + def _is_set(entity_id: str) -> bool: + if "::variant::" in entity_id or entity_id in modifier_isoform_set_ids(): + return False + try: + labels = get_labels(entity_id) + except IndexError: + return False + return any(t in labels for t in ("EntitySet", "DefinedSet", "CandidateSet", "OpenSet")) + # Deterministic order under the hierarchy: with the downstream test now # updated as edges are emitted, which of two jointly cycle-closing joins # survives depends on order, and set order of uuid4 strings is not stable. diff --git a/tests/test_boundary_set_nodes.py b/tests/test_boundary_set_nodes.py new file mode 100644 index 0000000..7bf3195 --- /dev/null +++ b/tests/test_boundary_set_nodes.py @@ -0,0 +1,68 @@ +"""A SET component of a root complex is one node, fed by its members over OR +`set_member` edges (LNG code review F7, deltasignal specs/044). It used to be +flattened into its members, each a required AND input of the complex, so a +knockout of one alternative removed the complex. + +Shape: GPVI, "VAV1 Rho/Rac effectors:GDP" = GDP + CandidateSet{RHOA, RAC1, CDC42}.""" +import pytest +import src.neo4j_connector as nc +import src.logic_network_generator as lng +import src.reaction_generator as rg +from src.logic_network_generator import _emit_boundary_decomposition_edges + +CX, GDP, SET, RHOA, RAC1, CDC42 = "R-HSA-CX", "R-ALL-GDP", "R-HSA-SET", "R-HSA-RHOA", "R-HSA-RAC1", "R-HSA-CDC42" +INNER, SUB1, SUB2 = "R-HSA-INNER", "R-HSA-SUB1", "R-HSA-SUB2" # a member that is itself a complex +LABELS = {CX: ["Complex"], SET: ["EntitySet", "CandidateSet"], INNER: ["Complex"], + GDP: ["SimpleEntity"], **{x: ["EntityWithAccessionedSequence"] for x in (RHOA, RAC1, CDC42, SUB1, SUB2)}} +COMPONENTS = {CX: {GDP: 1, SET: 1}, INNER: {SUB1: 1, SUB2: 1}} +MEMBERS = {SET: {RHOA, RAC1, CDC42, INNER}} + + +@pytest.fixture +def stub(monkeypatch): + monkeypatch.setattr(nc, "get_labels", lambda s: LABELS.get(s, [])) + monkeypatch.setattr(nc, "get_complex_components", lambda s: COMPONENTS.get(s, {})) + monkeypatch.setattr(nc, "get_set_members", lambda s: MEMBERS.get(s, set())) + # as the real walk does: a set or complex flattens to all its leaves + leaves = {SET: {RHOA, RAC1, CDC42, SUB1, SUB2}, INNER: {SUB1, SUB2}, CX: {GDP, RHOA, RAC1, CDC42, SUB1, SUB2}} + monkeypatch.setattr(lng, "get_terminal_components", lambda s: leaves.get(s, {s}), raising=False) + monkeypatch.setattr(rg, "_modifier_set_cache", set()) + monkeypatch.setenv("LNG_BOUNDARY_HIERARCHY", "1") + monkeypatch.delenv("LNG_COMPOSITION_EDGES", raising=False) + + +def run(): + data = [{"source_id": "u_CX", "target_id": "r1", "pos_neg": "pos", "and_or": "and", + "edge_type": "input", "stoichiometry": 1}, + {"source_id": "r1", "target_id": "u_Z", "pos_neg": "pos", "and_or": "or", + "edge_type": "output", "stoichiometry": 1}] + r2u = {"u_CX": CX, "u_Z": "R-HSA-Z"} + _emit_boundary_decomposition_edges(data, r2u) + return data, r2u + + +def test_the_set_is_one_node_feeding_the_complex(stub): + data, r2u = run() + into_cx = [(r2u[e["source_id"]], e["edge_type"], e["and_or"]) for e in data if e["target_id"] == "u_CX"] + assert sorted(into_cx) == [(GDP, "assembly", "and"), (SET, "assembly", "and")] + + +def test_members_feed_the_set_node_as_alternatives(stub): + data, r2u = run() + set_uuid = next(u for u, s in r2u.items() if s == SET) + into_set = {(r2u[e["source_id"]], e["edge_type"], e["and_or"]) for e in data if e["target_id"] == set_uuid} + assert into_set == {(m, "set_member", "or") for m in (RHOA, RAC1, CDC42, INNER)} + # no member is a required input of the complex any more + assert not any(e["target_id"] == "u_CX" and r2u[e["source_id"]] in (RHOA, RAC1, CDC42) for e in data) + + +def test_a_complex_member_is_built_from_its_components(stub): + data, r2u = run() + inner = next(u for u, s in r2u.items() if s == INNER) + assert {(r2u[e["source_id"]], e["and_or"]) for e in data if e["target_id"] == inner} == \ + {(SUB1, "and"), (SUB2, "and")} + + +def test_every_new_edge_is_in_the_boundary_layer(stub): + data, _ = run() + assert all(e.get("_boundary") for e in data[2:])