drugs.csv: drug-derived entities per pathway (deltasignal specs/032)

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,62 @@ 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
+
+    node_ids: set[str] = set()
+    if not pathway_logic_network.empty:
+        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 = []
+    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}")
+
+
 def export_node_resolution(pathway_id: str,
                            pathway_logic_network: pd.DataFrame,
                            reaction_id_map: pd.DataFrame,

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,77 @@ 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]]:
+    """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, 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:
+            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]])
+                _drug_meta_cache[r["x"]] = (r.get("c") or "", r.get("d") or "")
+        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
+
+    # 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]:
     """The Reactome release number of the connected graph, or None.
 

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,
@@ -494,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:

tests/test_provenance_exports.py

@@ -285,3 +285,94 @@ 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}
+    # 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):
+    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")]),
+        # 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 [{"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({"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