LNG_RELEASE_POOLS (default off): release pools for deltasignal specs/086

PIPELINE_SETTINGS.md

@@ -73,6 +73,7 @@ what NOT to touch. Overriding is for A/B benchmarking only.
 | `LNG_POOL_ACTIVE_VIA` | `made_from` | how a pool's active form is found (deltasignal specs/039) |
 | `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_CATALYST_BUNDLE` | `0` | legacy; subsumed by `LNG_COMPLEX_AS_NODE` |
 | `LNG_HANDOFF_EDGES` | `0` | precedingEvent hand-off bridges (net-negative, off) |
 

src/env_flags.py

@@ -29,6 +29,7 @@
     "LNG_PATHOGEN_PROTEIN": True,   # default since deltasignal specs/048 amendment 2 (arm pathp: exp +7, held-out +1, 0 broken)
     "LNG_POOL_JOIN_MEMBERS": False,   # deltasignal specs/048 lever B1
     "LNG_POOL_SHARED": False,         # deltasignal specs/049
+    "LNG_RELEASE_POOLS": True,        # default since deltasignal specs/086 (WNT tuning +73, experimental +8, held-out 0)
     "LNG_RECYCLE_POOLS": True,        # default since deltasignal specs/081 (tuning +70/0, experimental +8, held-out 0)
     "LNG_SET_EXPAND": True,
     "LNG_SET_MEMBERS_OR": False,

src/logic_network_generator.py

@@ -4736,6 +4736,174 @@ def copies(a: str, b: str, rx: str) -> List[str]:
     return forms, transitions, edges
 
 
+RELEASE_MAX_REACTIONS = 10   # deltasignal specs/086: beta-catenin's route is 8 reactions
+
+
+def find_release_pools(pathway_logic_network: pd.DataFrame, reaction_id_map: pd.DataFrame,
+                       uuid_mapping: Dict[str, str], profiles: Dict[str, Dict[str, Any]],
+                       claimed: Optional[Set[str]] = None, stats: Optional[Dict[str, int]] = None):
+    """Release pools (deltasignal specs/086): a protein bound, modified and released by a
+    complex, whose released copy carries on downstream.
+
+    beta-catenin in Signaling by WNT: its starting copy S binds the destruction complex;
+    CK1 and GSK3 phosphorylate it there, PP2A dephosphorylates it, and the released copy P
+    goes on to the nucleus. P's only producer is the release, so the nuclear signal was
+    the traffic through the destruction complex, and an APC knockout read DOWN. A candidate:
+
+    - S, a node of a single protein R with no in-edge that is an ``input`` or ``catalyst``
+      of a reaction;
+    - P, another node of the same entity that a reaction produces and that has an out-edge;
+    - a route S -> ... -> P of at most ``RELEASE_MAX_REACTIONS`` reactions (input or
+      catalyst to output) through forms that carry R;
+    - on that route a form whose modification signature for R differs from S's (a partner
+      released unchanged is a carrier, deltasignal specs/077, and is excluded).
+
+    The pool's states are every form M_v a release copy that makes P consumes (one per
+    variant of the complex; all bases: at rest the protein is mostly bound), and S; the
+    forms between are intermediates (amendment 1). Paths: S -> ... -> M_v along each
+    M_v's own route, and M_v -> S through its release. P is the pool's carrier, its
+    release steps every release copy, so it reads its other producers; the
+    edge S -> P (``released_twin``, OR) makes that S. A structural ``recycle`` edge
+    P -> S (specs/081) makes the cycle one component. A candidate any of whose nodes is in
+    ``claimed`` is skipped and counted.
+
+    Returns (forms, transitions, carriers, edges) in :func:`find_pools`'s row formats, the
+    edges as ``(source_uuid, target_uuid, edge_type)``."""
+    from src.variant_keys import variant_parts
+    stats = stats if stats is not None else {}
+    for k in ("release_pools", "release_paths", "release_claimed"):
+        stats[k] = 0
+    claimed = claimed or set()
+    if pathway_logic_network.empty:
+        return [], [], [], []
+    ent = _uuid_to_stable_id_map(pathway_logic_network, uuid_mapping)
+    vr = {str(u): str(r) for u, r in zip(reaction_id_map["uid"], reaction_id_map["reactome_id"])}
+    feeds: Dict[str, Set[str]] = {}      # entity -> reaction nodes it is an input or catalyst of
+    rin: Dict[str, Set[str]] = {}        # reaction node -> its input and catalyst entities
+    rout: Dict[str, Set[str]] = {}
+    catalysed: Set[str] = set()
+    produced: Set[str] = set(); has_out: Set[str] = set(); has_in: Set[str] = set()
+    for _, e in pathway_logic_network.iterrows():
+        s, t, et = str(e["source_id"]), str(e["target_id"]), e.get("edge_type")
+        has_out.add(s); has_in.add(t)
+        if et in ("input", "catalyst") and t in vr:
+            feeds.setdefault(s, set()).add(t); rin.setdefault(t, set()).add(s)
+            if et == "catalyst":
+                catalysed.add(t)
+        elif et == "output" and s in vr:
+            rout.setdefault(s, set()).add(t); produced.add(t)
+
+    def pe_of(u: str) -> str:
+        return variant_parts(ent.get(u, ""))[0]
+
+    def proteins(u: str) -> frozenset:
+        parent, members = variant_parts(ent.get(u, ""))
+        if not members:
+            return frozenset(profiles.get(parent, {}).get("proteins", ()))
+        out = set(profiles.get(parent, {}).get("fixed", ()))
+        for mm in members:
+            out |= set(profiles.get(mm, {}).get("proteins", ()))
+        return frozenset(out)
+
+    def sig(u: str, r):
+        return profiles.get(pe_of(u), {}).get("sig", {}).get(r, frozenset())
+
+    by_pe: Dict[str, List[str]] = {}
+    for u in sorted(ent):
+        by_pe.setdefault(pe_of(u), []).append(u)
+
+    def route(src: str, targets: Set[str], r) -> Optional[List[Tuple[str, str, str]]]:
+        prev: Dict[str, Optional[Tuple[str, str]]] = {src: None}
+        frontier = [src]
+        for _ in range(RELEASE_MAX_REACTIONS):
+            nxt = []
+            for x in frontier:
+                for rx in sorted(feeds.get(x, ())):
+                    for y in sorted(rout.get(rx, ())):
+                        if y in prev or r not in proteins(y):
+                            continue
+                        prev[y] = (x, rx); nxt.append(y)
+                        if y in targets:
+                            steps: List[Tuple[str, str, str]] = []
+                            link = prev[y]
+                            while link is not None:
+                                a, rr = link; steps.append((a, rr, y)); y = a
+                                link = prev[y]
+                            return steps[::-1]
+            frontier = nxt
+        return None
+
+    def copies(a: str, b: str, rx: str) -> List[str]:
+        return sorted(u for u in vr if vr[u] == vr[rx] and a in rin.get(u, ()) and b in rout.get(u, ()))
+
+    forms: List[Tuple] = []; transitions: List[Tuple] = []; carriers: List[Tuple] = []
+    edges: List[Tuple[str, str, str]] = []
+    n = 0
+    for s_node in sorted(ent):
+        if s_node in has_in or s_node in vr or not feeds.get(s_node):
+            continue
+        rs = proteins(s_node)
+        if len(rs) != 1:
+            continue
+        r = next(iter(rs))
+        ps = {u for u in by_pe.get(pe_of(s_node), []) if u != s_node and u in produced and u in has_out}
+        if not ps:
+            continue
+        st = route(s_node, ps, r)
+        if not st or len(st) < 2:
+            continue
+        if not any(sig(y, r) != sig(s_node, r) for _, _, y in st[:-1]):
+            continue
+        p_node = st[-1][2]
+        rel_stid = vr[st[-1][1]]
+        # amendment 1: every release copy that makes P, each with its own consumed form
+        # M_v and its own route from S (one chain per variant of the complex)
+        chains = []
+        for rc in sorted(u for u in vr if vr[u] == rel_stid and p_node in rout.get(u, ())):
+            for m_node in sorted(x for x in rin.get(rc, ()) if r in proteins(x) and x != s_node):
+                cst = route(s_node, {m_node}, r)
+                if cst:
+                    chains.append((m_node, rc, cst))
+        if not chains:
+            continue
+        nodes = {s_node}
+        fwd_by_m: Dict[str, List[Tuple[str, str, List[str]]]] = {}
+        rel_by_m: Dict[str, List[str]] = {}
+        inter: Set[str] = set()
+        for m_node, rc, cst in chains:
+            fwd_by_m.setdefault(m_node, [(a, y, copies(a, y, rx)) for a, rx, y in cst])
+            rel_by_m.setdefault(m_node, []).append(rc)
+            inter |= {y for _, _, y in cst[:-1]}
+            nodes |= {m_node}
+        inter -= set(fwd_by_m) | {s_node}
+        nodes |= inter | {u for f in fwd_by_m.values() for _, _, cs in f for u in cs} | \
+                 {u for rs in rel_by_m.values() for u in rs}
+        if nodes & claimed or any(not cs for f in fwd_by_m.values() for _, _, cs in f):
+            stats["release_claimed"] += 1
+            continue
+        n += 1
+        pid = f"release{n}"
+        for m_node in sorted(fwd_by_m):
+            forms.append((pid, m_node, ent.get(m_node, ""), "state", True))
+        forms.append((pid, s_node, ent.get(s_node, ""), "state", False))
+        for u in sorted(inter):
+            forms.append((pid, u, ent.get(u, ""), "intermediate", False))
+        for i, m_node in enumerate(sorted(fwd_by_m), 1):
+            fwd = fwd_by_m[m_node]
+            enz = any(u in catalysed for _, _, cs in fwd for u in cs)
+            for si, (a, b, cs) in enumerate(fwd, 1):
+                for u in cs:
+                    transitions.append((pid, f"{pid}_f{i}", si, a, b, u, vr[u], enz))
+            for u in sorted(set(rel_by_m[m_node])):
+                transitions.append((pid, f"{pid}_r{i}", 1, m_node, s_node, u, vr[u], u in catalysed))
+                carriers.append((pid, p_node, u))
+            stats["release_paths"] += 2
+        edges += [(s_node, p_node, "released_twin"), (p_node, s_node, "recycle")]
+        claimed |= nodes
+        stats["release_pools"] += 1
+    return forms, transitions, carriers, edges
+
+
 def export_pools(pathway_id: str, pathway_logic_network: pd.DataFrame, reaction_id_map: pd.DataFrame,
                  uuid_mapping: Dict[str, str], pools_file: str, transitions_file: str,
                  carriers_file: str, carrier_loops_file: Optional[str] = None,
@@ -4784,6 +4952,24 @@ def export_pools(pathway_id: str, pathway_logic_network: pd.DataFrame, reaction_
         logger.info(f"Recycle pools (specs/081): {rstats['recycle_pools']} pools, {rstats['recycle_states']} states, "
                     f"{rstats['recycle_paths']} paths, {rstats['recycle_edges']} recycle edges, "
                     f"{rstats['recycle_claimed']} skipped (nodes claimed by a pool)")
+    if env_flag("LNG_RELEASE_POOLS"):
+        # deltasignal specs/086: after the ordinary and recycle pools, which keep their nodes
+        claimed = {f[1] for f in forms} | {t[5] for t in transitions}
+        lstats: Dict[str, int] = {}
+        lf, lt, lc, ledges = find_release_pools(pathway_logic_network, reaction_id_map, uuid_mapping, profiles,
+                                                claimed, lstats)
+        forms = list(forms) + lf
+        transitions = list(transitions) + lt
+        carriers = list(carriers) + lc
+        if ledges:
+            if boundary_file is None:
+                raise ValueError("LNG_RELEASE_POOLS found release pools but no boundary file to write their edges to")
+            cols = list(pd.read_csv(boundary_file, nrows=0).columns)
+            add = pd.DataFrame([{"source_id": a, "target_id": b, "pos_neg": "pos", "and_or": "or",
+                                 "edge_type": t, "stoichiometry": 1} for a, b, t in ledges])
+            add.reindex(columns=cols).to_csv(boundary_file, mode="a", header=False, index=False)
+        logger.info(f"Release pools (specs/086): {lstats['release_pools']} pools, {lstats['release_paths']} paths, "
+                    f"{lstats['release_claimed']} skipped (nodes claimed by a pool or no reaction copies)")
     pd.DataFrame(forms, columns=["pool_id", "node_uuid", "stable_id", "role", "is_base"]).to_csv(pools_file, index=False)
     pd.DataFrame(transitions, columns=["pool_id", "path_id", "step", "source_uuid", "target_uuid", "reaction_uuid",
                                        "reaction_stid", "enzyme_driven"]).to_csv(transitions_file, index=False)

tests/test_release_pools.py

@@ -0,0 +1,99 @@
+"""Release pools (deltasignal specs/086): a protein bound, modified and released by a
+complex, whose released copy carries on. Shape: beta-catenin (s) binds the destruction
+complex (dc) into c1; a kinase phosphorylates it (c2); a phosphatase dephosphorylates it
+(c3); the release (v4) makes p, which goes on to the nucleus (vt -> n). The pool: c3 (base)
+and s, intermediates c1 and c2; paths s -> c1 -> c2 -> c3 and c3 -> s; p is the carrier
+with release v4; edges s -> p (released_twin) and p -> s (recycle)."""
+import pandas as pd
+
+import src.logic_network_generator as m
+
+NAMES = ["s", "dc", "v1", "c1", "kin", "v2", "c2", "pp", "v3", "c3", "v4", "p", "vt", "n"]
+U = {k: f"00000000-0000-0000-0000-{i:012d}" for i, k in enumerate(NAMES, 1)}
+CTNNB1, DC, KIN, PP = 1, 2, 3, 4
+
+
+def e(s, t, et):
+    return {"source_id": U[s], "target_id": U[t], "edge_type": et, "stoichiometry": 1}
+
+
+def prof(proteins, sig=()):
+    return {"proteins": set(proteins), "fixed": set(proteins), "sig": {r: frozenset(sig) for r in proteins},
+            "mods": 0, "comps": 0, "slots": 1}
+
+
+def bcat(modified=True, carries_on=True):
+    edges = [e("s", "v1", "input"), e("dc", "v1", "input"), e("v1", "c1", "output"),
+             e("c1", "v2", "input"), e("kin", "v2", "catalyst"), e("v2", "c2", "output"),
+             e("c2", "v3", "input"), e("pp", "v3", "catalyst"), e("v3", "c3", "output"),
+             e("c3", "v4", "input"), e("v4", "p", "output")]
+    if carries_on:
+        edges += [e("p", "vt", "input"), e("vt", "n", "output")]
+    net = pd.DataFrame(edges)
+    rx = ["v1", "v2", "v3", "v4", "vt"]
+    rmap = pd.DataFrame({"uid": [U[k] for k in rx], "reactome_id": [f"R-HSA-{k.upper()}" for k in rx]})
+    umap = {U["s"]: "R-HSA-CTNNB1", U["p"]: "R-HSA-CTNNB1", U["dc"]: "R-HSA-DC", U["c1"]: "R-HSA-C1",
+            U["c2"]: "R-HSA-C2", U["c3"]: "R-HSA-C3", U["kin"]: "R-HSA-KIN", U["pp"]: "R-HSA-PP", U["n"]: "R-HSA-N"}
+    phos = (("res", "p-S45"),) if modified else ()
+    profiles = {"R-HSA-CTNNB1": prof([CTNNB1]), "R-HSA-DC": prof([DC]),
+                "R-HSA-C1": prof([CTNNB1, DC]), "R-HSA-C2": prof([CTNNB1, DC], phos),
+                "R-HSA-C3": prof([CTNNB1, DC]), "R-HSA-KIN": prof([KIN]), "R-HSA-PP": prof([PP]),
+                "R-HSA-N": prof([CTNNB1])}
+    return net, rmap, umap, profiles
+
+
+def test_beta_catenin_is_a_free_bound_pool_whose_released_copy_reads_the_free_state():
+    st = {}
+    forms, trans, carriers, edges = m.find_release_pools(*bcat(), stats=st)
+    assert [(u, role, base) for _, u, _, role, base in forms] == [
+        (U["c3"], "state", True), (U["s"], "state", False),
+        (U["c1"], "intermediate", False), (U["c2"], "intermediate", False)]
+    rows = sorted((t[1], t[2], t[3], t[4], t[5]) for t in trans)
+    assert rows == sorted([("release1_f1", 1, U["s"], U["c1"], U["v1"]), ("release1_f1", 2, U["c1"], U["c2"], U["v2"]),
+                           ("release1_f1", 3, U["c2"], U["c3"], U["v3"]), ("release1_r1", 1, U["c3"], U["s"], U["v4"])])
+    assert [(c[1], c[2]) for c in carriers] == [(U["p"], U["v4"])]
+    assert edges == [(U["s"], U["p"], "released_twin"), (U["p"], U["s"], "recycle")]
+    assert st["release_pools"] == 1 and st["release_paths"] == 2 and st["release_claimed"] == 0
+
+
+def test_a_partner_released_unchanged_is_not_a_release_pool():
+    assert m.find_release_pools(*bcat(modified=False)) == ([], [], [], [])
+
+
+def test_a_released_copy_that_leads_nowhere_is_left_to_recycle_pools():
+    assert m.find_release_pools(*bcat(carries_on=False)) == ([], [], [], [])
+
+
+def test_a_candidate_on_a_claimed_node_is_skipped_and_counted():
+    st = {}
+    assert m.find_release_pools(*bcat(), claimed={U["c2"]}, stats=st) == ([], [], [], [])
+    assert st["release_claimed"] == 1
+
+
+def test_the_flag_is_on_by_default():
+    # default since deltasignal specs/086 was adopted (Adam, 2026-10-21)
+    from src.env_flags import BOOL_FLAGS
+    assert BOOL_FLAGS["LNG_RELEASE_POOLS"] is True
+
+
+def test_every_variant_chain_is_in_the_pool():
+    # amendment 1: a second variant of the complex (dc2 -> c1b -> c2b -> c3b) also releases p
+    net, rmap, umap, profiles = bcat()
+    extra = ["dc2", "w1", "c1b", "w2", "c2b", "w3", "c3b", "w4"]
+    for i, k in enumerate(extra, 100):
+        U[k] = f"00000000-0000-0000-0000-{i:012d}"
+    rows = [e("s", "w1", "input"), e("dc2", "w1", "input"), e("w1", "c1b", "output"),
+            e("c1b", "w2", "input"), e("kin", "w2", "catalyst"), e("w2", "c2b", "output"),
+            e("c2b", "w3", "input"), e("pp", "w3", "catalyst"), e("w3", "c3b", "output"),
+            e("c3b", "w4", "input"), e("w4", "p", "output")]
+    net = pd.concat([net, pd.DataFrame(rows)], ignore_index=True)
+    rmap = pd.concat([rmap, pd.DataFrame({"uid": [U[k] for k in ("w1", "w2", "w3", "w4")],
+                                          "reactome_id": ["R-HSA-V1", "R-HSA-V2", "R-HSA-V3", "R-HSA-V4"]})],
+                     ignore_index=True)
+    umap = dict(umap); umap.update({U["dc2"]: "R-HSA-DC", U["c1b"]: "R-HSA-C1", U["c2b"]: "R-HSA-C2", U["c3b"]: "R-HSA-C3"})
+    st = {}
+    forms, trans, carriers, edges = m.find_release_pools(net, rmap, umap, profiles, stats=st)
+    assert sorted(u for _, u, _, role, base in forms if base) == sorted([U["c3"], U["c3b"]])
+    assert sorted(c[2] for c in carriers) == sorted([U["v4"], U["w4"]])
+    assert {t[1] for t in trans} == {"release1_f1", "release1_r1", "release1_f2", "release1_r2"}
+    assert st["release_pools"] == 1 and st["release_paths"] == 4