specs/048: pathogens.csv; homodimer-aware output binding (LNG_BIND_STOICH)

src/env_flags.py

@@ -14,6 +14,7 @@
 BOOL_FLAGS = {
     "LNG_BOUNDARY_EXPANSION": True,
     "LNG_BOUNDARY_HIERARCHY": True,
+    "LNG_BIND_STOICH": True,     # default since deltasignal specs/048 (homodimer binding; 0 predictions moved)
     "LNG_CAP_POOLS": True,      # default since deltasignal specs/045 (RAF experimental +14, held-out 0)
     "LNG_CATALYST_BUNDLE": False,
     "LNG_COMPLEX_AS_NODE": True,
@@ -24,6 +25,7 @@
     "LNG_DIAGRAM_SET_MEMBER": False,
     "LNG_EMIT_ONE_SIDED": True,
     "LNG_HANDOFF_EDGES": False,
+    "LNG_PATHOGEN_PROTEIN": True,   # default since deltasignal specs/048 amendment 2 (arm pathp: exp +7, held-out +1, 0 broken)
     "LNG_SET_EXPAND": True,
     "LNG_SET_MEMBERS_OR": False,
     "LNG_SET_POOL": True,

src/logic_network_generator.py

@@ -3901,7 +3901,34 @@ def export_drugs(pathway_logic_network: pd.DataFrame,
 
     Output CSV columns: stable_id, schema_class, name, reactome_release.
     """
-    from src.neo4j_connector import get_drug_entities, get_reactome_release
+    from src.neo4j_connector import get_drug_entities
+    _export_derived(pathway_logic_network, reactome_id_to_uuid, output_file,
+                    get_drug_entities, "drug-derived")
+
+
+def export_pathogens(pathway_logic_network: pd.DataFrame,
+                     reactome_id_to_uuid: Dict[str, str],
+                     output_file: str) -> None:
+    """Write pathogens.csv: which entities in THIS network are pathogen-derived
+    (deltasignal specs/048), by the rule of
+    :func:`src.neo4j_connector.get_pathogen_entities`. Reactome curates
+    host-pathogen interactions inside human pathways (the SARS-CoV-2
+    N:M:PDPK1 complex inhibits AKT T308 phosphorylation in PIP3 signalling); a
+    consumer modelling an uninfected cell holds them at baseline, as it holds
+    drugs. Same columns, variant resolution and always-written rule as
+    drugs.csv.
+    """
+    from src.neo4j_connector import get_pathogen_entities
+    _export_derived(pathway_logic_network, reactome_id_to_uuid, output_file,
+                    get_pathogen_entities, "pathogen-derived")
+
+
+def _export_derived(pathway_logic_network: pd.DataFrame,
+                    reactome_id_to_uuid: Dict[str, str],
+                    output_file: str, finder, label: str) -> None:
+    """The node listing shared by drugs.csv and pathogens.csv. ``finder`` maps
+    stable ids to {stId: {"schema_class", "name"}} for the derived ones."""
+    from src.neo4j_connector import get_reactome_release
 
     node_ids: set[str] = set()
     if not pathway_logic_network.empty:
@@ -3923,20 +3950,20 @@ def parts(node_id: str):
         parent, members = parts(n)
         wanted.add(parent)
         wanted.update(members)
-    drugs = get_drug_entities(wanted)
+    derived = finder(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)
+        hit = parent if parent in derived else next((m for m in members if m in derived), None)
         if hit is None:
             continue
-        rows.append({"stable_id": n, "schema_class": drugs[hit]["schema_class"],
-                     "name": drugs[hit]["name"],
+        rows.append({"stable_id": n, "schema_class": derived[hit]["schema_class"],
+                     "name": derived[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}")
+    logger.info(f"Exported {len(rows)} {label} entities to {output_file}")
 
 
 POOL_MAX_STEPS = 6          # a state -> intermediates -> state path longer than this is dropped

src/neo4j_connector.py

@@ -933,6 +933,97 @@ def derived(s: str, depth: int = 0) -> bool:
             for s in ids if derived(s)}
 
 
+HUMAN_TAXON = "9606"
+# stId -> (species taxIds, [(relationship type, child stId)], schema class, display name)
+_pathogen_structure_cache: Dict[str, Tuple[List[str], List[Tuple[str, str]], str, str]] = {}
+# stId -> schema class of its referenceEntity ("" when it has none), for LNG_PATHOGEN_PROTEIN
+_pathogen_ref_cache: Dict[str, str] = {}
+# A protein leaf: an EWAS on a ReferenceGeneProduct (ReferenceIsoform is its subclass).
+PROTEIN_REFERENCE_CLASSES = frozenset({"ReferenceGeneProduct", "ReferenceIsoform"})
+
+
+def get_pathogen_entities(stable_ids) -> Dict[str, Dict[str, str]]:
+    """The entities among ``stable_ids`` that are PATHOGEN-DERIVED, with their
+    schema class and display name (deltasignal specs/048).
+
+    Reactome curates host-pathogen interactions inside human pathways: the
+    SARS-CoV-2 N:M:PDPK1 complex negatively regulates "PDPK1 phosphorylates
+    AKT at T308". A benchmark describes a cell without the pathogen. Decided
+    structurally, as drugs are (:func:`get_drug_entities`):
+      - a leaf (no components, members or candidates) is, when it has a species
+        and none is Homo sapiens;
+      - a complex is, when ANY component is;
+      - a set is, only when EVERY member is.
+    A leaf with no species (most small molecules) is not.
+
+    Under ``LNG_PATHOGEN_PROTEIN=1`` (specs/048 amendment 2) a leaf counts only
+    when it is also a PROTEIN: an EWAS on a ReferenceGeneProduct or
+    ReferenceIsoform. Viral nucleic acids are then not pathogen-derived, whether
+    curated as a GenomeEncodedEntity with no reference entity (the DDX58/IFIH1
+    ligands in our catalog) or as an EWAS on a ReferenceRNASequence /
+    ReferenceDNASequence: in DDX58/IFIH1 the RNA is the ligand the pathway
+    senses, and every signalling complex there carries one.
+
+    A Polymer's ``repeatedUnit`` is read as a component, so a viral protein
+    multimer (HIV-1 Rev-multimer) counts through its unit; a Polymer has no
+    reference entity of its own.
+    """
+    protein_only = env_flag("LNG_PATHOGEN_PROTEIN")
+    ids = sorted({s for s in stable_ids if s})
+    missing = [s for s in ids if s not in _pathogen_structure_cache]
+    if missing:
+        query = """
+            UNWIND $ids AS s
+            MATCH (n:PhysicalEntity {stId: s})-[:hasComponent|hasMember|hasCandidate|repeatedUnit*0..10]->(x)
+            WITH DISTINCT x
+            OPTIONAL MATCH (x)-[:species]->(sp)
+            OPTIONAL MATCH (x)-[:referenceEntity]->(re)
+            OPTIONAL MATCH (x)-[r:hasComponent|hasMember|hasCandidate|repeatedUnit]->(y)
+            RETURN x.stId AS x, x.schemaClass AS c, x.displayName AS d,
+                   re.schemaClass AS ref,
+                   collect(DISTINCT toString(sp.taxId)) AS tax,
+                   collect(DISTINCT 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_pathogen_entities", **_traceback_kwargs())
+            raise
+        for r in rows:
+            if r.get("x"):
+                _pathogen_structure_cache[r["x"]] = (
+                    [t for t in r["tax"] if t], [tuple(k) for k in r["kids"] if k and k[1]],
+                    r.get("c") or "", r.get("d") or "")
+                _pathogen_ref_cache[r["x"]] = r.get("ref") or ""
+        for s in missing:
+            _pathogen_structure_cache.setdefault(s, ([], [], "", ""))
+
+    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
+        tax, kids, _, _ = _pathogen_structure_cache.get(s, ([], [], "", ""))
+        comps = [y for t, y in kids if t in ("hasComponent", "repeatedUnit")]
+        members = [y for t, y in kids if t in ("hasMember", "hasCandidate")]
+        if 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 = bool(tax) and HUMAN_TAXON not in tax and (
+                not protein_only or _pathogen_ref_cache.get(s, "") in PROTEIN_REFERENCE_CLASSES)
+        memo[s] = out
+        return out
+
+    return {s: {"schema_class": _pathogen_structure_cache[s][2],
+                "name": _pathogen_structure_cache[s][3]}
+            for s in ids if derived(s)}
+
+
 # Group donors: a reaction consuming one ADDS a group to its substrate, so its
 # product is the modified form of an interconversion pool (specs/039). ChEBI
 # ids: ATP, GTP, S-adenosyl-L-methionine, acetyl-CoA, NAD+.

src/pathway_generator.py

@@ -13,6 +13,7 @@
     create_pathway_logic_network,
     export_cofactors,
     export_drugs,
+    export_pathogens,
     export_pools,
     export_containment,
     export_containment_structure,
@@ -42,6 +43,8 @@
     "LNG_SET_POOL",
     "LNG_CAP_POOLS",
     "LNG_POOL_ACTIVE_VIA",
+    "LNG_BIND_STOICH",
+    "LNG_PATHOGEN_PROTEIN",
     "LNG_VARIANT_NODES",
     "LNG_DIAGRAM_SET_MEMBER",
     # Of these three, only LNG_EMIT_ONE_SIDED is a genuine cache gap: it is
@@ -219,7 +222,7 @@ def _write_cache_fingerprint(
     "logic_network.csv", "stid_to_uuid_mapping.csv", "entity_reaction_proxy_mapping.csv",
     "nodes.csv", "node_reaction_context.csv", "node_resolution.csv", "node_exclusions.csv",
     "boundary_edges.csv", "cofactors.csv", "containment.csv", "containment_structure.csv",
-    "drugs.csv",
+    "drugs.csv", "pathogens.csv",
     "pools.csv", "pool_transitions.csv", "pool_carriers.csv",
 )
 
@@ -566,6 +569,17 @@ def generate_pathway_file(
             logger.error(f"Failed to write drugs.csv: {e}", exc_info=True)
             raise
 
+        # Pathogen-derived entities (deltasignal specs/048), likewise on their own.
+        try:
+            export_pathogens(
+                result.logic_network,
+                result.uuid_mapping,
+                str(pathway_output_dir / "pathogens.csv"),
+            )
+        except Exception as e:
+            logger.error(f"Failed to write pathogens.csv: {e}", exc_info=True)
+            raise
+
         # Interconversion pools (deltasignal specs/039), in their own block.
         try:
             export_pools(

src/variant_emission.py

@@ -38,6 +38,7 @@
 import pandas as pd
 
 from src import variant_keys as vk
+from src.env_flags import env_flag
 
 # `participant` is the curated catalyst/regulator the row came from (a bare
 # set's copies name its MEMBERS in entity_id), read by depleter pooling.
@@ -93,22 +94,44 @@ def _slots_of(participants: Sequence[str]) -> Set[str]:
     return out
 
 
-_sig_cache: Dict[str, Tuple[str, ...]] = {}
+_sig_cache: Dict[Tuple[str, bool], Tuple[str, ...]] = {}   # keyed by (member, LNG_BIND_STOICH)
 
 
 def _reference_signature(member: str) -> Tuple[str, ...]:
     """Isoform-level identity of a set member: the sorted multiset of its
     leaves' reference entities (a member complex keeps its leaf multiset)."""
-    if member in _sig_cache:
-        return _sig_cache[member]
+    stoich = env_flag("LNG_BIND_STOICH")
+    if (member, stoich) in _sig_cache:
+        return _sig_cache[(member, stoich)]
     from src.neo4j_connector import get_reference_entity_id
     from src.reaction_generator import get_terminal_components
-    leaves = sorted(get_terminal_components(member)) if vk.is_complex(member) else [member]
+    if vk.is_complex(member) and stoich:
+        leaves = _leaf_multiset(member)
+    else:
+        leaves = sorted(get_terminal_components(member)) if vk.is_complex(member) else [member]
     sig = tuple(sorted(str(get_reference_entity_id(x) or x) for x in leaves))
-    _sig_cache[member] = sig
+    _sig_cache[(member, stoich)] = sig
     return sig
 
 
+def _leaf_multiset(entity: str, depth: int = 0) -> List[str]:
+    """The leaves of a complex with their stoichiometry, as a multiset
+    (deltasignal specs/048 B3). ``get_terminal_components`` returns a set, so a
+    homodimer (MAPK1 dimer, R-HSA-5675354) had the signature of its monomer
+    (MAPK1, R-HSA-59282), and "Cytosolic DUSPs dephosphorylate MAPKs"
+    (R-HSA-5675376) bound p-MAPK1 to the MAPK1 DIMER and the p-MAPK1 dimer to
+    the monomer, by sorted rank. A set inside a complex contributes its
+    leaves once each, as before."""
+    from src.neo4j_connector import get_complex_components
+    from src.reaction_generator import get_terminal_components
+    if depth > 10 or not vk.is_complex(entity):
+        return sorted(get_terminal_components(entity)) or [entity]
+    out: List[str] = []
+    for comp, st in sorted(get_complex_components(entity).items()):
+        out.extend(_leaf_multiset(comp, depth + 1) * int(st or 1))
+    return sorted(out) or [entity]
+
+
 def bind_output_slots(input_side: Set[str], output_slots: Set[str]) -> Dict[str, Tuple[str, Dict[str, str]]]:
     """Output-only slot -> (input-side slot, {input member: output member}).
 

tests/test_export_failfast.py

@@ -24,7 +24,7 @@
 EXPORTS = ["export_uuid_to_reactome_mapping", "export_entity_reaction_proxy_mapping",
            "export_nodes", "export_node_reaction_context", "export_node_resolution",
            "export_cofactors", "export_containment", "export_containment_structure",
-           "export_drugs", "export_pools"]
+           "export_drugs", "export_pathogens", "export_pools"]
 
 
 @pytest.fixture