specs/049: shared-intermediate pools and carrier_loops.csv (draft; arms pending)

src/env_flags.py

@@ -26,6 +26,8 @@
     "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_POOL_JOIN_MEMBERS": False,   # deltasignal specs/048 lever B1
+    "LNG_POOL_SHARED": False,         # deltasignal specs/049
     "LNG_SET_EXPAND": True,
     "LNG_SET_MEMBERS_OR": False,
     "LNG_SET_POOL": True,
@@ -39,6 +41,7 @@
     "LNG_HANDOFF_HUB_MAX",        # _int_env
     "LNG_MAX_VARIANTS",           # _int_env
     "LNG_POOL_ACTIVE_VIA",        # pool_active_via()
+    "LNG_POOL_REGEN",             # pool_regen()
     "LNG_PYTHON",                 # catalog.sh: the interpreter, not a switch
 })
 

src/logic_network_generator.py

@@ -512,6 +512,7 @@ def _reject_removed_env() -> None:
             raise ValueError(f"{name} was removed: {why}")
     validate_env(removed=_REMOVED_ENV)
     pool_active_via()
+    pool_regen()
 
 
 _POOL_ACTIVE_VIA = ("direct", "made_from")
@@ -533,6 +534,24 @@ def pool_active_via() -> str:
     return value
 
 
+_POOL_REGEN = ("entity", "protein")
+
+
+def pool_regen() -> str:
+    """LNG_POOL_REGEN (deltasignal specs/048 lever B2): how a multi-step pool
+    path proves it regenerates what it consumes (specs/039 amendment 5).
+    ``entity`` (default): every consumed non-R, non-small input is output again
+    by a step of the path, compared by stable id. ``protein``: every protein a
+    consumed input carries is output again, in any form, so a partner released
+    under another stable id (RAS:GTP:activated RAF dimers, R-HSA-5672718 in and
+    R-HSA-5672712 out) or modified (MAP2K in, p-2S MAP2K out) still counts.
+    Any other value is an error at startup."""
+    value = os.environ.get("LNG_POOL_REGEN", "entity")
+    if value not in _POOL_REGEN:
+        raise ValueError(f"LNG_POOL_REGEN must be one of {_POOL_REGEN}, got {value!r}")
+    return value
+
+
 def _register_phase1(
     vr_entities: Dict[str, tuple],
     entity_uuid_registry: Dict[tuple, str],
@@ -3972,14 +3991,16 @@ def parts(node_id: str):
                    "shared_nodes_dropped", "ties", "carriers", "autocat_source", "autocat_product",
                    "autocat_other", "carrier_loops", "no_path_pools", "off_path_intermediates",
                    "off_path_states", "ambiguous_copies", "budget_dropped", "merged_proteins", "copy_rows",
-                   "carrier_conflicts", "nonregenerating_paths", "oriented_by_activity", "activity_fallbacks")
+                   "carrier_conflicts", "nonregenerating_paths", "oriented_by_activity", "activity_fallbacks",
+                   "carrier_loop_sets", "carrier_loops_closed", "carrier_loop_forms", "carrier_loop_conflicts")
 
 
 def find_pools(pathway_logic_network: pd.DataFrame, reaction_id_map: pd.DataFrame,
                uuid_mapping: Dict[str, str], steps, profiles: Dict[str, Dict[str, Any]],
                donor_reactions: Optional[Set[str]] = None,
                set_members: Optional[Dict[str, Set[str]]] = None,
-               stats: Optional[Dict[str, int]] = None):
+               stats: Optional[Dict[str, int]] = None,
+               carrier_loops: Optional[List[Tuple]] = None):
     """Modification-cycle pools at NODE level (deltasignal specs/039 amendment 2).
 
     ``steps`` are the pathway's R-steps at stId level, ``(reaction, R, input
@@ -4043,6 +4064,18 @@ def find_pools(pathway_logic_network: pd.DataFrame, reaction_id_map: pd.DataFram
                    source, target and reaction_stid and differ in reaction_uuid. ``enzyme_driven`` is the
                    STEP's flag; a path is enzyme-driven if any step is.
       carriers:    [(pool_id, carrier_uuid, release_reaction_uuid)]
+    ``carrier_loops``, when given a list, receives the carrier loops (deltasignal
+    specs/049): each R-form set rejected above as a carrier loop on the first
+    pass, de-duplicated across R. One row per (loop, free form, loop source):
+    ``(loop_id, r, closed, free_uuid, free_stid, loop_source_uuid)``. A free form
+    is a loop node whose proteins are exactly the loop's core; a loop source is
+    a node inside the loop (a step reaction node, a form of the loop, or a plain
+    node a step writes that variant_split hands to a form) with a positive
+    in-edge into that free form (any type but input, catalyst and regulator).
+    The loop is closed when no form of it has such an in-edge from outside the
+    loop. Free forms and conflicts are counted once per node, and a (free form,
+    source) pair in several loops is closed if any of them is. A free form that a pool manages (state, intermediate,
+    step node or carrier) is not listed (``carrier_loop_conflicts``).
     ``stats`` receives the counts in ``POOL_STATS_KEYS``; the autocatalysis
     counts are distinct (reaction node, catalyst) pairs where a form of the pool
     catalyses a step of its own pool, split by whether the catalyst is the
@@ -4065,6 +4098,11 @@ def find_pools(pathway_logic_network: pd.DataFrame, reaction_id_map: pd.DataFram
     acts: Dict[str, Set[str]] = {}     # node -> reaction nodes it catalyses or positively regulates
     set_pools_of: Dict[str, Set[str]] = {}   # member node -> the set-pool nodes it feeds (specs/033)
     feeds: Dict[str, Set[str]] = {}          # node -> reaction nodes it is an input of
+    into: Dict[str, Set[str]] = {}           # entity node -> sources of its entity-level in-edges (specs/049)
+    pool_members: Dict[str, Set[str]] = {}   # pool node -> the member nodes feeding it (specs/048 B1)
+    split_to: Dict[str, Set[str]] = {}       # plain output node -> the variants it is split to (specs/048 B1)
+    join_members = env_flag("LNG_POOL_JOIN_MEMBERS")
+    regen_by_protein = pool_regen() == "protein"
     active_via = pool_active_via()
     has_st = "stoichiometry" in pathway_logic_network.columns
     for _, e in pathway_logic_network.iterrows():
@@ -4088,8 +4126,17 @@ def find_pools(pathway_logic_network: pd.DataFrame, reaction_id_map: pd.DataFram
             acts.setdefault(s, set()).add(t)
         elif et == "set_member":
             set_pools_of.setdefault(s, set()).add(t)
+        if et in ("set_member", "variant_pool"):
+            pool_members.setdefault(t, set()).add(s)
+        elif et == "variant_split":
+            split_to.setdefault(s, set()).add(t)
         if et == "input" and t in vr:
             feeds.setdefault(s, set()).add(t)
+        if et not in ("input", "catalyst", "regulator", "depletion") and t not in vr:
+            # every positive in-edge into an entity counts as supply: output,
+            # variant_split/variant_pool/set_member, and the flagged derived
+            # classes (diagram_bridge, composition, handoff) when present
+            into.setdefault(t, set()).add(s)
     copies: Dict[str, List[str]] = {}
     for u, r in vr.items():
         copies.setdefault(r, []).append(u)
@@ -4130,7 +4177,24 @@ def maps(u: str, pe: str) -> bool:
     for rx, r, a, b, enz in steps:
         for rx_u in sorted(copies.get(rx, [])):
             cin = [u for u, st in ins.get(rx_u, {}).items() if st == 1 and maps(u, a) and r in proteins(u)]
+            if join_members:
+                # specs/048 B1: a step reading a pool node (variant_pool or
+                # set_member target) starts from the pool's members that carry
+                # R, so a cycle closed through the pool (MAPK3 released by
+                # DUSPs re-entering R-HSA-169291::pool) is visible here as it
+                # is to the solver.
+                cin = sorted({m for u in cin for m in (
+                    [m for m in sorted(pool_members[u]) if r in proteins(m)]
+                    if u in pool_members else [u])})
             cout = [u for u, st in outs.get(rx_u, {}).items() if st == 1 and maps(u, b) and r in proteins(u)]
+            if join_members:
+                # and a step writing a plain node that variant_split edges hand
+                # to its variants (a capped reaction's output) ends at those
+                # variants that carry R: R-HSA-5672980 releases R-HSA-169289,
+                # split to p-MAPK1 and p-MAPK3.
+                cout = sorted({m for u in cout for m in (
+                    [m for m in sorted(split_to[u]) if r in proteins(m)]
+                    if u in split_to else [u])})
             if len(cin) != 1 or len(cout) != 1:
                 if len(cin) > 1 or len(cout) > 1:
                     stats["ambiguous_copies"] += 1
@@ -4139,6 +4203,7 @@ def maps(u: str, pe: str) -> bool:
                 ustep[(r, cin[0], cout[0], rx_u)] = (rx, bool(enz))
     if not ustep:
         return [], [], []
+    first_loops: List[Tuple[Any, frozenset, frozenset, Set[str]]] = []   # (R, forms, core, step reaction nodes)
 
     def build(removed: Set[str]):
         local = {k: 0 for k in POOL_STATS_KEYS}
@@ -4154,9 +4219,16 @@ def regenerates(path) -> bool:
             produced: Set[str] = set()
             for a_u, _, _, _, cp in path:
                 for rx_u in cp:
-                    consumed |= {ent.get(j, "") for j in ins.get(rx_u, {})
-                                 if j != a_u and not prof(j, "small", False) and pe_of(j) not in _UBIQUITIN_STIDS}
-                    produced |= {ent.get(o, "") for o in outs.get(rx_u, {})}
+                    eaten = [j for j in ins.get(rx_u, {})
+                             if j != a_u and not prof(j, "small", False) and pe_of(j) not in _UBIQUITIN_STIDS
+                             and not (join_members and a_u in pool_members.get(j, ()))]
+                    if regen_by_protein:
+                        # specs/048 B2: by the proteins carried, in any form
+                        consumed |= {p for j in eaten for p in proteins(j)}
+                        produced |= {p for o in outs.get(rx_u, {}) for p in proteins(o)}
+                    else:
+                        consumed |= {ent.get(j, "") for j in eaten}
+                        produced |= {ent.get(o, "") for o in outs.get(rx_u, {})}
             if consumed <= produced:
                 return True
             local["nonregenerating_paths"] += 1
@@ -4184,6 +4256,9 @@ def regenerates(path) -> bool:
                         by_sig.setdefault(sig(u, r), []).append(u)
                 if len(by_sig) < 2:
                     local["carrier_loops"] += 1
+                    if not removed:
+                        first_loops.append((r, frozenset(scc), core,
+                                            {x for a_u, b_u, x, _, _ in by_r[r] if a_u in scc and b_u in scc}))
                     continue
                 states: Set[str] = set()
                 for members in by_sig.values():
@@ -4259,13 +4334,20 @@ def walk(u, so_far, seen):
 
     removed: Set[str] = set()
     dropped_shared = 0
+    keep_shared = env_flag("LNG_POOL_SHARED")
     while True:
         pools, local = build(removed)
         claimed: Dict[str, int] = {}
         for p in pools:
             for n in p["nodes"]:
                 claimed[n] = claimed.get(n, 0) + 1
         shared = {n for n, c in claimed.items() if c > 1}
+        if keep_shared:
+            # deltasignal specs/049: an intermediate or step reaction node that
+            # two pools pass through (MAPK1 and MAPK3 through one capped
+            # RAF:MAP2K:MAPKs complex) stays in both; the consumer writes it from
+            # the flux of every pool through it. A shared STATE still goes.
+            shared = {n for n in shared if any(n in p["states"] for p in pools)}
         if not shared:
             break
         removed |= shared
@@ -4397,6 +4479,45 @@ def acts_on(u: str) -> Set[str]:
         stats["copy_rows"] += sum(len(cp) for path in paths for *_, cp in path)
         stats["multi_step_pools"] += int(any(len(path) > 1 for path in paths))
         stats["carriers"] += len({j for j, _ in found})
+    if carrier_loops is not None:
+        # deltasignal specs/049: a carrier loop's free form reads its supply, not
+        # its own release; the consumer drops these in-edges (all of them for a
+        # closed loop, whose free form then reads baseline)
+        pool_carriers = {c for _, c, _ in carriers}
+        merged: Dict[frozenset, Tuple[Any, frozenset, Set[str]]] = {}
+        for r, scc, core, rx in first_loops:
+            if scc in merged:
+                merged[scc][2].update(rx)
+            else:
+                merged[scc] = (r, core, set(rx))
+        # one row per (free form, loop source); a pair in several loops (one per
+        # R) keeps the first loop's id and is closed if ANY of its loops is
+        rows: Dict[Tuple[str, str], List[Any]] = {}
+        loop_conflicts: Set[str] = set()
+        order = sorted(merged, key=lambda f: min(ent.get(u, "") + u for u in f))
+        for i, scc in enumerate(order, start=1):
+            r, core, rx = merged[scc]
+            # a step that writes a plain node handing on to the loop's forms by
+            # variant_split (a capped release, under LNG_POOL_JOIN_MEMBERS) is
+            # inside the loop, and so is that plain node
+            inside = set(scc) | rx | {pn for x in rx for pn in outs.get(x, {}) if split_to.get(pn, set()) & scc}
+            closed = not any(into.get(u, set()) - inside for u in scc)
+            stats["carrier_loop_sets"] += 1
+            stats["carrier_loops_closed"] += int(closed)
+            for u in sorted(scc, key=lambda u: (ent.get(u, ""), u)):
+                if proteins(u) != core:
+                    continue
+                if u in managed or u in pool_carriers:
+                    loop_conflicts.add(u)
+                    continue
+                for src in sorted(into.get(u, set()) & inside):
+                    if (u, src) in rows:
+                        rows[(u, src)][2] = rows[(u, src)][2] or closed
+                    else:
+                        rows[(u, src)] = [f"loop{i}", r, closed, u, ent.get(u, ""), src]
+        stats["carrier_loop_conflicts"] = len(loop_conflicts)
+        stats["carrier_loop_forms"] = len({u for u, _ in rows})
+        carrier_loops.extend(tuple(v) for v in rows.values())
     return forms, transitions, carriers
 
 
@@ -4433,7 +4554,7 @@ def r_steps(participants, profiles: Dict[str, Dict[str, Any]], ubiquitin_stids=(
 
 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) -> None:
+                 carriers_file: str, carrier_loops_file: Optional[str] = None) -> None:
     """Write pools.csv, pool_transitions.csv and pool_carriers.csv (deltasignal
     specs/039 amendment 2): each protein's modification-cycle pool, for a
     consumer that solves it at steady state (pi = pi P) instead of multiplying
@@ -4457,12 +4578,16 @@ def export_pools(pathway_id: str, pathway_logic_network: pd.DataFrame, reaction_
     donors = get_donor_reactions({s[0] for s in steps}, _UBIQUITIN_STIDS)
     set_members = get_set_descendants({x for s in steps for x in (s[2], s[3])})
     stats: Dict[str, int] = {}
+    loops: List[Tuple] = []
     forms, transitions, carriers = find_pools(pathway_logic_network, reaction_id_map, uuid_mapping, steps,
-                                              profiles, donors, set_members, stats)
+                                              profiles, donors, set_members, stats, carrier_loops=loops)
     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)
     pd.DataFrame(carriers, columns=["pool_id", "carrier_uuid", "release_reaction_uuid"]).to_csv(carriers_file, index=False)
+    if carrier_loops_file is not None:
+        pd.DataFrame(loops, columns=["loop_id", "r", "closed", "free_uuid", "free_stid", "loop_source_uuid"]).to_csv(
+            carrier_loops_file, index=False)
     logger.info(
         f"Exported {stats['pools']} pools ({stats['states']} states, {stats['intermediates']} intermediates, "
         f"{stats['paths']} paths in {stats['copy_rows']} step-copy rows, {stats['multi_step_pools']} multi-step; "
@@ -4472,7 +4597,9 @@ def export_pools(pathway_id: str, pathway_logic_network: pd.DataFrame, reaction_
         f"{stats['shared_nodes_dropped']} shared nodes dropped, {stats['oriented_by_activity']} oriented by the active "
         f"state, {stats['activity_fallbacks']} by residues/donor/components of which {stats['ties']} by tie-break, "
         f"{stats['carriers']} carriers ({stats['carrier_conflicts']} excluded as pool nodes), "
-        f"{stats['carrier_loops']} carrier loops, "
+        f"{stats['carrier_loops']} carrier loops ({stats['carrier_loop_sets']} distinct, "
+        f"{stats['carrier_loops_closed']} closed, {stats['carrier_loop_forms']} free forms listed, "
+        f"{stats['carrier_loop_conflicts']} excluded as pool nodes), "
         f"{stats['merged_proteins']} co-travelling proteins merged, {stats['no_path_pools']} candidates with no path, "
         f"{stats['off_path_states']}/{stats['off_path_intermediates']} states/intermediates off every path, "
         f"{stats['ambiguous_copies']} ambiguous reaction copies, {stats['budget_dropped']} over the path budget; "

src/pathway_generator.py

@@ -43,6 +43,9 @@
     "LNG_SET_POOL",
     "LNG_CAP_POOLS",
     "LNG_POOL_ACTIVE_VIA",
+    "LNG_POOL_JOIN_MEMBERS",
+    "LNG_POOL_REGEN",
+    "LNG_POOL_SHARED",
     "LNG_BIND_STOICH",
     "LNG_PATHOGEN_PROTEIN",
     "LNG_VARIANT_NODES",
@@ -223,7 +226,7 @@ def _write_cache_fingerprint(
     "nodes.csv", "node_reaction_context.csv", "node_resolution.csv", "node_exclusions.csv",
     "boundary_edges.csv", "cofactors.csv", "containment.csv", "containment_structure.csv",
     "drugs.csv", "pathogens.csv",
-    "pools.csv", "pool_transitions.csv", "pool_carriers.csv",
+    "pools.csv", "pool_transitions.csv", "pool_carriers.csv", "carrier_loops.csv",
 )
 
 
@@ -590,6 +593,7 @@ def generate_pathway_file(
                 str(pathway_output_dir / "pools.csv"),
                 str(pathway_output_dir / "pool_transitions.csv"),
                 str(pathway_output_dir / "pool_carriers.csv"),
+                str(pathway_output_dir / "carrier_loops.csv"),
             )
         except Exception as e:
             logger.error(f"Failed to write pools.csv: {e}", exc_info=True)