Drop the LNG_BOUNDARY_LEAF_REUSE escape hatch

PIPELINE_SETTINGS.md

@@ -99,3 +99,16 @@ If you drive DeltaSignal over its HTTP API, two things matter:
   is an in-memory, per-process cache, so re-parse if the server restarts.
 
 Full request/response schemas: `deltasignal/docs/API.md`.
+
+## Removed flags
+
+These names are still **read**, and setting one is a hard error rather than a
+no-op. A flag is removed once the question it existed to ask has been answered:
+the spec keeps the evidence, the code does not keep a second path nobody
+measures. Rejecting the name rather than ignoring it means a stale value in a
+shell, a compose file or an unversioned benchmark script cannot quietly make a
+run measure the default while the operator believes otherwise.
+
+| flag | removed | why |
+|---|---|---|
+| `LNG_BOUNDARY_LEAF_REUSE` | 2026-09-21 | Boundary leaves never reuse a node the root complex can reach at the time the leaf is chosen. Held-out +173, p<1e-4 (deltasignal `specs/018-derived-edge-loops`). |

bin/create-pathways.py

@@ -22,6 +22,7 @@
 sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
 
 from src.argument_parser import configure_logging, logger, parse_args
+from src.logic_network_generator import _reject_removed_env
 from src.pathway_generator import generate_pathway_file
 from src.neo4j_connector import get_top_level_pathways, get_pathway_name
 
@@ -40,6 +41,14 @@ def canonical_pathway_id(raw: str) -> str:
 
 
 def main() -> None:
+    # Fail before ANY work if a removed flag is set. The per-pathway
+    # guards inside the generator would each raise, but the loop below catches
+    # every exception and continues, so without this check a stale flag would
+    # produce "93 failed" buried in the log, leave the PREVIOUS run's
+    # logic_network.csv files untouched on disk, and still exit 0 -- and the
+    # benchmark that reads that directory would silently score the old catalog.
+    _reject_removed_env()
+
     dotenv_path = os.path.join(os.path.dirname(__file__), "..", ".env")
     # load_dotenv populates os.environ (without clobbering already-set vars) so
     # NEO4J_URL/USER/PASSWORD in .env actually reach neo4j_connector.get_graph(),
@@ -63,15 +72,15 @@ def main() -> None:
     if pathway_list_file:
         if not os.path.exists(pathway_list_file):
             logger.error(f"Pathway list file '{pathway_list_file}' does not exist.")
-            return
+            sys.exit(1)
         elif not os.access(pathway_list_file, os.R_OK):
             logger.error(f"Pathway list file '{pathway_list_file}' is not readable.")
-            return
+            sys.exit(1)
     elif not args.pathway_list and not args.pathway_id and not args.top_level_pathways:
         logger.error(
             "One of the following is required: '--pathway-id', '--pathway-list', '--top-level-pathways', or 'PATHWAY_LIST_FILE' environment variable."
         )
-        return
+        sys.exit(1)
 
     pathway_list: List[Tuple[str, str]] = []
 
@@ -84,7 +93,7 @@ def main() -> None:
             logger.info(f"Found {len(pathway_list)} top-level pathways")
         except Exception as e:
             logger.error(f"Error fetching top-level pathways: {e}")
-            return
+            sys.exit(1)
     elif args.pathway_id:
         # Single pathway by ID - fetch name from database
         pathway_id = args.pathway_id
@@ -93,10 +102,10 @@ def main() -> None:
             logger.info(f"Found pathway: {pathway_name} (stId: {pathway_id})")
         except ValueError:
             logger.error(f"Pathway with ID {pathway_id} not found in database")
-            return
+            sys.exit(1)
         except Exception as e:
             logger.error(f"Error fetching pathway name: {e}")
-            return
+            sys.exit(1)
         pathway_list = [(pathway_id, pathway_name)]
     elif pathway_list_file:
         try:
@@ -106,7 +115,7 @@ def main() -> None:
                                             pathways_df["pathway_name"])]
         except Exception as e:
             logger.error(f"Error reading pathway list file: {e}")
-            return
+            sys.exit(1)
 
     logger.info(f"Processing {len(pathway_list)} pathway(s)")
     logger.info(f"Output directory: {output_dir}")
@@ -124,6 +133,13 @@ def main() -> None:
             continue
 
     logger.info(f"Completed: {successful} successful, {failed} failed")
+    if failed:
+        # A partial catalog is not a catalog: downstream benchmarks glob the
+        # output directory and would score whatever mix of new and stale
+        # pathways happens to be there. Exit non-zero so a shell `&&` chain,
+        # a Makefile or CI stops instead of proceeding.
+        logger.error(f"{failed} pathway(s) failed; exiting non-zero")
+        sys.exit(1)
 
 
 if __name__ == "__main__":

src/logic_network_generator.py

@@ -481,6 +481,26 @@ def _register_entity_uuid(
     return entity_uuid_registry[key]
 
 
+# Flags that were removed once the question they existed to ask was answered
+# (see the flag-expiry policy in deltasignal specs/020). The names are still
+# READ so that a stale value in a shell, a compose file or an unversioned
+# benchmark script fails loudly instead of being silently ignored while the run
+# measures the default. That silent-input failure mode has bitten this project
+# four times (deltasignal DS #40-43).
+_REMOVED_ENV = {
+    "LNG_BOUNDARY_LEAF_REUSE":
+        "boundary leaves never reuse a node the root complex can reach at the time "
+        "the leaf is chosen. See deltasignal specs/018-derived-edge-loops.",
+}
+
+
+def _reject_removed_env() -> None:
+    """Raise if any removed LNG_* flag is set. Safe to call repeatedly."""
+    for name, why in sorted(_REMOVED_ENV.items()):
+        if name in os.environ:
+            raise ValueError(f"{name} was removed: {why}")
+
+
 def _register_phase1(
     vr_entities: Dict[str, tuple],
     entity_uuid_registry: Dict[tuple, str],
@@ -1712,11 +1732,22 @@ def _emit_boundary_decomposition_edges(
     # takes TP53's strongly connected component from 836 nodes to 56 and DSB's
     # from 1,127 to ~290 (deltasignal specs/018). Reachability is computed over
     # every edge emitted so far (bridges and depletion edges included).
-    # LNG_BOUNDARY_LEAF_REUSE=any restores the old behaviour.
+    #
+    # This is unconditional. The LNG_BOUNDARY_LEAF_REUSE escape hatch that
+    # restored the welding behaviour was removed once the question it existed
+    # to ask was answered (held-out +173, p<1e-4).
+    #
+    # SCOPE OF THE RULE, stated precisely: a leaf is refused if the root
+    # complex can reach it *at the moment that leaf is chosen*. `_succ` is a
+    # snapshot taken before the loop and is NOT updated as this function
+    # appends its own assembly edges, so a later root complex can still be
+    # handed a leaf that became reachable through an edge emitted for an
+    # earlier one. That residual weld is a real pre-existing bug -- it exists
+    # identically on both sides of the flag removal -- and it is tracked
+    # separately rather than fixed here, because fixing it changes emitted
+    # networks and would invalidate the measurement this rule rests on.
     from collections import defaultdict
-    reuse_mode = os.environ.get("LNG_BOUNDARY_LEAF_REUSE", "downstream_free")
-    if reuse_mode not in ("downstream_free", "any"):
-        raise ValueError(f"LNG_BOUNDARY_LEAF_REUSE={reuse_mode!r}: expected 'downstream_free' or 'any'")
+    _reject_removed_env()
     _succ: Dict[str, List[str]] = defaultdict(list)
     for e in pathway_logic_network_data:
         _succ[str(e.get("source_id"))].append(str(e.get("target_id")))
@@ -1739,14 +1770,10 @@ def _downstream_of(root_uuid: str) -> Set[str]:
 
     def _leaf_uuid(leaf_stid: str, root_uuid: str) -> str:
         candidates = stid_to_existing_uuids.get(leaf_stid, [])
-        if reuse_mode == "any":
-            if candidates:
-                return candidates[0]
-        else:
-            downstream = _downstream_of(root_uuid)
-            for cand in candidates:
-                if cand not in downstream:
-                    return cand
+        downstream = _downstream_of(root_uuid)
+        for cand in candidates:
+            if cand not in downstream:
+                return cand
         if leaf_stid not in leaf_uuid_registry:
             leaf_uuid_registry[leaf_stid] = str(uuid.uuid4())
             reactome_id_to_uuid[leaf_uuid_registry[leaf_stid]] = leaf_stid

tests/conftest.py

@@ -0,0 +1,19 @@
+"""Shared fixtures.
+
+Removed LNG_* flags are rejected at runtime (see `_REMOVED_ENV`), so a stale
+value exported in a developer's shell or a CI runner would turn every test that
+builds a network into a spurious failure. Tests should exercise the code, not
+the ambient environment, so clear those names for the whole suite. The guard
+itself is tested explicitly in tests/test_removed_env_flags.py.
+"""
+
+import pytest
+
+from src.logic_network_generator import _REMOVED_ENV
+
+
+@pytest.fixture(autouse=True)
+def _clear_removed_env_flags(monkeypatch):
+    for name in _REMOVED_ENV:
+        monkeypatch.delenv(name, raising=False)
+    yield

tests/test_boundary_leaf_reuse.py

@@ -5,7 +5,6 @@
 v97 catalog; deltasignal specs/018). A produced node that is NOT downstream is a real
 feed-forward link (hasComponent with no reaction) and must still be reused -- severing
 those cost Mitotic G1 28 cases in the first version of this fix."""
-import os
 import pytest
 import src.neo4j_connector as nc
 import src.logic_network_generator as lng
@@ -39,8 +38,7 @@ def assembly_sources(data):
     return {e["source_id"]: e for e in data if e["edge_type"] == "assembly"}
 
 
-def test_produced_node_is_not_reused_as_a_boundary_leaf(stub, monkeypatch):
-    monkeypatch.delenv("LNG_BOUNDARY_LEAF_REUSE", raising=False)
+def test_produced_node_is_not_reused_as_a_boundary_leaf(stub):
     data, r2u = network()
     _emit_boundary_decomposition_edges(data, r2u)
     asm = assembly_sources(data)
@@ -57,10 +55,9 @@ def test_produced_node_is_not_reused_as_a_boundary_leaf(stub, monkeypatch):
     assert not (sources_into_C & produced)
 
 
-def test_a_copy_reached_only_through_a_bridge_is_downstream_too(stub, monkeypatch):
+def test_a_copy_reached_only_through_a_bridge_is_downstream_too(stub):
     # P has two copies: u_Pout (reaction output, downstream of C) and u_Pin (an input copy whose
     # only incoming edge is a diagram bridge from u_Pout -- also downstream). Neither may be reused.
-    monkeypatch.delenv("LNG_BOUNDARY_LEAF_REUSE", raising=False)
     data, r2u = network()
     data.append({"source_id": "u_Pout", "target_id": "u_Pin", "pos_neg": "pos", "and_or": "and", "edge_type": "diagram_bridge", "stoichiometry": 1})
     data.append({"source_id": "u_Pin", "target_id": "r3", "pos_neg": "pos", "and_or": "and", "edge_type": "input", "stoichiometry": 1})
@@ -71,11 +68,10 @@ def test_a_copy_reached_only_through_a_bridge_is_downstream_too(stub, monkeypatc
     assert "u_Pout" not in asm and "u_Pin" not in asm
 
 
-def test_a_produced_copy_that_is_not_downstream_is_reused(stub, monkeypatch):
+def test_a_produced_copy_that_is_not_downstream_is_reused(stub):
     # The Mitotic G1 shape: P's only copy is produced by a reaction the root complex C does NOT
     # reach (r2, fed by Q). Reusing it adds no cycle and keeps the feed-forward link
     # P -> (assembly) -> C that Reactome joins only by hasComponent.
-    monkeypatch.delenv("LNG_BOUNDARY_LEAF_REUSE", raising=False)
     data = [
         {"source_id": "u_C", "target_id": "r1", "pos_neg": "pos", "and_or": "and", "edge_type": "input", "stoichiometry": 1},
         {"source_id": "r1", "target_id": "u_W", "pos_neg": "pos", "and_or": "or", "edge_type": "output", "stoichiometry": 1},
@@ -93,7 +89,6 @@ def test_a_produced_copy_that_is_not_downstream_is_reused(stub, monkeypatch):
 def test_two_root_complexes_sharing_a_subunit_each_get_an_acyclic_leaf(stub, monkeypatch):
     # C reaches P's produced copy (u_Pout) -> C gets a fresh leaf. C2 does NOT reach it ->
     # C2 reuses u_Pout (a real feed-forward link). Neither assembly edge closes a cycle.
-    monkeypatch.delenv("LNG_BOUNDARY_LEAF_REUSE", raising=False)
     data, r2u = network()
     data.append({"source_id": "u_C2", "target_id": "r4", "pos_neg": "pos", "and_or": "and", "edge_type": "input", "stoichiometry": 1})
     data.append({"source_id": "r4", "target_id": "u_Z", "pos_neg": "pos", "and_or": "or", "edge_type": "output", "stoichiometry": 1})
@@ -118,16 +113,11 @@ def reach(u):
         assert not (srcs & reach(cpx))
 
 
-def test_legacy_reuse_is_available_and_welds_the_cycle(stub, monkeypatch):
-    monkeypatch.setenv("LNG_BOUNDARY_LEAF_REUSE", "any")
-    data, r2u = network()
-    _emit_boundary_decomposition_edges(data, r2u)
-    asm = assembly_sources(data)
-    assert "u_Pout" in asm          # the old behaviour: the produced copy is reused -> C -> r1 -> u_Pout -> C
-
-
-def test_bad_mode_is_an_error(stub, monkeypatch):
-    monkeypatch.setenv("LNG_BOUNDARY_LEAF_REUSE", "unproduced")
-    data, r2u = network()
-    with pytest.raises(ValueError):
-        _emit_boundary_decomposition_edges(data, r2u)
+def test_removed_flag_is_an_error_not_a_noop(stub, monkeypatch):
+    # The escape hatch was removed; a stale value must fail loudly rather than
+    # letting a run silently measure the default while claiming the old mode.
+    for val in ("any", "downstream_free", "unproduced"):
+        monkeypatch.setenv("LNG_BOUNDARY_LEAF_REUSE", val)
+        data, r2u = network()
+        with pytest.raises(ValueError, match="was removed"):
+            _emit_boundary_decomposition_edges(data, r2u)

tests/test_removed_env_flags.py

@@ -0,0 +1,80 @@
+"""The removed-flag guard, tested for real.
+
+This file exists because the first version of the guard was tested only by
+asserting that a string appeared in `inspect.getsource(...)`. That passes
+whether or not the guard runs: mutating it to `if False and ...` left the whole
+suite green. These tests call the guard and the entry point that uses it.
+"""
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+from src.logic_network_generator import _REMOVED_ENV, _reject_removed_env
+
+REPO = Path(__file__).resolve().parents[1]
+
+
+def test_clean_environment_is_fine(monkeypatch):
+    for name in _REMOVED_ENV:
+        monkeypatch.delenv(name, raising=False)
+    _reject_removed_env()          # must not raise
+
+
+@pytest.mark.parametrize("name", sorted(_REMOVED_ENV))
+@pytest.mark.parametrize("value", ["1", "0", "any", "", "downstream_free"])
+def test_any_value_of_a_removed_flag_raises(monkeypatch, name, value):
+    # Including "0" and "": the point is that the name is gone, so there is no
+    # value that means "use the old behaviour" and none that means "off".
+    monkeypatch.setenv(name, value)
+    with pytest.raises(ValueError, match=f"{name} was removed"):
+        _reject_removed_env()
+
+
+@pytest.mark.parametrize("name", sorted(_REMOVED_ENV))
+def test_the_message_names_the_spec_that_holds_the_evidence(monkeypatch, name):
+    monkeypatch.setenv(name, "1")
+    with pytest.raises(ValueError) as exc:
+        _reject_removed_env()
+    assert "specs/" in str(exc.value)
+
+
+@pytest.mark.parametrize("name", sorted(_REMOVED_ENV))
+def test_entry_point_exits_nonzero_before_doing_any_work(monkeypatch, name, tmp_path):
+    # The generator's per-pathway guard is not enough on its own: the pathway
+    # loop catches every exception and continues, so a stale flag would leave
+    # the previous run's files on disk and still exit 0.
+    env = {**dict(__import__("os").environ), name: "1"}
+    out = tmp_path / "out"
+    proc = subprocess.run(
+        [sys.executable, "bin/create-pathways.py", "--pathway-id", "R-HSA-73894",
+         "--output-dir", str(out)],
+        cwd=REPO, env=env, capture_output=True, text=True, timeout=180,
+    )
+    assert proc.returncode != 0, "a removed flag must stop the run"
+    assert f"{name} was removed" in (proc.stderr + proc.stdout)
+    assert not out.exists() or not any(out.iterdir()), "must fail before writing output"
+
+
+def test_a_failing_pathway_makes_the_run_exit_nonzero(tmp_path):
+    """A partial catalog must not look like a successful run.
+
+    The pathway loop catches every exception and continues, so before this the
+    command logged "N failed" and still exited 0. A benchmark that globs the
+    output directory would then score whatever mix of new and stale pathways
+    happened to be there. Any failure reason serves here -- a bogus stable id
+    or an unreachable database both take the same path.
+    """
+    listing = tmp_path / "list.tsv"
+    listing.write_text("id\tpathway_name\nR-HSA-NOT-A-REAL-PATHWAY\tbogus\n")
+    out = tmp_path / "out"
+    proc = subprocess.run(
+        [sys.executable, "bin/create-pathways.py",
+         "--pathway-list", str(listing), "--output-dir", str(out)],
+        cwd=REPO, capture_output=True, text=True, timeout=300,
+    )
+    assert proc.returncode != 0, (
+        "a run in which every pathway failed must not exit 0\n"
+        f"stdout:\n{proc.stdout[-2000:]}\nstderr:\n{proc.stderr[-2000:]}"
+    )