F7: a set inside a root complex is one node fed by its members (OR), not every member a required input (deltasignal specs/044)

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.

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:])