Raise the invariants that were only asserted

backend/asset_store.py

@@ -134,7 +134,10 @@ def put(self, data: bytes) -> str:
         bytes twice writes once and returns the same ref both times."""
         ref = content_ref(data)
         target = self._path_for(ref)
-        assert target is not None  # content_ref always yields a valid ref
+        if target is None:  # content_ref always yields a valid ref
+            # Raised, not asserted - `python -O` strips an assert, and this
+            # one guards the line below from writing to None.
+            raise ValueError(f"asset ref {ref!r} does not map to a storage path")
         if target.is_file():
             return ref
 

backend/events.py

@@ -459,13 +459,22 @@ def register_topic(
         `baseline_builder` must accompany `patch_builder` - it supplies the
         last-published state send_snapshot serves, without which a
         subscriber can be handed state newer than the revision stamped on it
-        and diverge permanently. Asserted rather than merely documented,
-        because that failure is completely silent."""
-        assert name not in self._topics, f"topic {name!r} registered twice"
-        assert patch_builder is None or baseline_builder is not None, (
-            f"topic {name!r}: a patch_builder needs a baseline_builder, or "
-            f"send_snapshot serves live state stamped with a stale revision"
-        )
+        and diverge permanently. Enforced rather than merely documented,
+        because that failure is completely silent.
+
+        RAISED, not asserted. These were `assert` statements, which `python
+        -O` strips - and with them gone a duplicate registration silently
+        REPLACES the previous handler. backend/app.py's _configure_session
+        registers 12 topics and ~90 intents in an order its own comments
+        call load-bearing, so a silent overwrite there is exactly the class
+        of failure these checks exist to make loud."""
+        if name in self._topics:
+            raise ValueError(f"topic {name!r} registered twice")
+        if patch_builder is not None and baseline_builder is None:
+            raise ValueError(
+                f"topic {name!r}: a patch_builder needs a baseline_builder, or "
+                f"send_snapshot serves live state stamped with a stale revision"
+            )
         self._topics[name] = _Topic(
             name, builder, schema_version, min_compatible, patch_builder, baseline_builder
         )
@@ -503,10 +512,13 @@ def register_intent(
              become a real schema in a later increment.
         """
         key = (topic, intent)
-        assert key not in self._intents, f"intent {topic}/{intent} registered twice"
-        assert args_schema is None or dataclasses.is_dataclass(args_schema), (
-            f"args_schema for {topic}/{intent} must be a dataclass type, got {args_schema!r}"
-        )
+        # Raised, not asserted - see register_topic's own note.
+        if key in self._intents:
+            raise ValueError(f"intent {topic}/{intent} registered twice")
+        if args_schema is not None and not dataclasses.is_dataclass(args_schema):
+            raise TypeError(
+                f"args_schema for {topic}/{intent} must be a dataclass type, got {args_schema!r}"
+            )
         self._intents[key] = _IntentRegistration(handler, args_schema)
 
     def has_topic(self, name: str) -> bool:

backend/mcp_client.py

@@ -426,7 +426,12 @@ def _drain_oversized_line(stdout) -> bool:
 
     def _read_loop(self) -> None:
         process = self._process
-        assert process is not None and process.stdout is not None
+        if process is None or process.stdout is None:
+            # Raised, not asserted: `python -O` strips an assert, and the
+            # AttributeError that follows on a None pipe is a far worse
+            # message than this - it surfaces deep inside a reader thread
+            # with no hint that the process simply was not connected.
+            raise RuntimeError("read loop started before the process was connected")
         try:
             while True:
                 # SECURITY-FIX: readline(size) bounds a single read to

backend/plugin_sdk.py

@@ -1099,7 +1099,12 @@ def call(self, method: str, params: dict) -> dict:
 
     def _read_loop(self) -> None:
         process = self._process
-        assert process is not None and process.stdout is not None
+        if process is None or process.stdout is None:
+            # Raised, not asserted: `python -O` strips an assert, and the
+            # AttributeError that follows on a None pipe is a far worse
+            # message than this - it surfaces deep inside a reader thread
+            # with no hint that the process simply was not connected.
+            raise RuntimeError("read loop started before the process was connected")
         try:
             for line in process.stdout:
                 line = line.strip()

backend/tests/test_event_bus.py

@@ -164,10 +164,14 @@ def configure(session_bus):
 
 
 def test_duplicate_registration_is_a_programming_error():
+    """ValueError, not AssertionError: these were asserts, and `python -O`
+    strips those - a duplicate registration would then silently REPLACE the
+    previous handler rather than failing. See register_topic's own note and
+    tests/test_no_production_asserts.py."""
     bus, _ = make_session()
-    with pytest.raises(AssertionError):
+    with pytest.raises(ValueError, match="registered twice"):
         bus.register_topic("counter", dict)
-    with pytest.raises(AssertionError):
+    with pytest.raises(ValueError, match="registered twice"):
         bus.register_intent("counter", "bump", lambda: None)
 
 

backend/tests/test_scene_patch_protocol.py

@@ -413,11 +413,13 @@ async def run():
 
 
 def test_registering_a_patch_builder_without_a_baseline_builder_is_a_programming_error():
-    # The pairing is load-bearing and its failure is completely silent, so
-    # it is asserted rather than merely documented.
+    # The pairing is load-bearing and its failure is completely silent, so it
+    # is enforced rather than merely documented - and RAISED rather than
+    # asserted, because `python -O` strips an assert and would take the
+    # enforcement with it.
     document = SceneDocument()
     bus = SessionBus("no-baseline")
-    with pytest.raises(AssertionError):
+    with pytest.raises(ValueError, match="baseline_builder"):
         bus.register_topic("scene", document.scene_payload, patch_builder=document.take_dirty_patch_ops)
 
 

tests/test_no_production_asserts.py

@@ -0,0 +1,83 @@
+"""Shipped code must not enforce an invariant with `assert`.
+
+`python -O` and `PYTHONOPTIMIZE=1` strip assert statements entirely. Anything
+an assert was guarding is then simply not guarded - silently, with no error
+and no log. That is fine for a test, where the whole process runs under
+pytest, and wrong for shipped code.
+
+The seven that existed when this gate was written were not stylistic. Two of
+them were the event bus's duplicate-registration guards:
+
+    assert name not in self._topics,  "topic registered twice"
+    assert key  not in self._intents, "intent registered twice"
+
+backend/app.py's _configure_session registers 12 topics and roughly 90
+intents in an order its own comments describe as load-bearing. With those
+asserts stripped, a duplicate registration silently REPLACES the previous
+handler - the exact failure the checks exist to make loud, made silent by an
+interpreter flag. Two more guarded a subprocess pipe against None, where the
+alternative to the assert is an AttributeError raised deep inside a reader
+thread with no hint that the process was never connected.
+
+Nothing in the repo runs under -O today. That is the point: this is cheap to
+keep true and expensive to discover is not.
+
+AST-based rather than grep, same as tests/test_domain_purity.py and
+test_node_state_migration.py - a grep for "assert " also matches the word
+inside docstrings and comments, and backend/events.py has one of those.
+"""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+
+# The four shipped packages (pyproject's [tool.setuptools.packages.find])
+# plus the loose root modules that ship with them.
+SHIPPED_TREES = ("backend", "graphlink_plugins", "provider_runtime", "settings_store")
+
+
+def _shipped_modules() -> list[Path]:
+    modules: list[Path] = []
+    for tree in SHIPPED_TREES:
+        for path in (REPO_ROOT / tree).rglob("*.py"):
+            if "tests" in path.parts or "__pycache__" in path.parts:
+                continue
+            modules.append(path)
+    modules.extend(
+        path for path in REPO_ROOT.glob("*.py")
+        if not path.name.startswith("test_")
+    )
+    return modules
+
+
+def _asserts_in(path: Path) -> list[int]:
+    try:
+        tree = ast.parse(path.read_text(encoding="utf-8"))
+    except SyntaxError:  # pragma: no cover - a broken module fails elsewhere
+        return []
+    return [node.lineno for node in ast.walk(tree) if isinstance(node, ast.Assert)]
+
+
+def test_no_shipped_module_enforces_an_invariant_with_assert():
+    offenders = [
+        f"{path.relative_to(REPO_ROOT).as_posix()}:{line}"
+        for path in _shipped_modules()
+        for line in _asserts_in(path)
+    ]
+    assert not offenders, (
+        "assert statements in shipped code - `python -O` strips these, taking "
+        "the invariant with them. Raise instead:\n  " + "\n  ".join(sorted(offenders))
+    )
+
+
+def test_the_scan_actually_reaches_the_shipped_code():
+    """Guards the guard: a wrong root or a changed layout would make the check
+    above pass over an empty file list."""
+    modules = _shipped_modules()
+    assert len(modules) > 100, len(modules)
+    names = {path.name for path in modules}
+    for expected in ("events.py", "mcp_client.py", "plugin_sdk.py", "asset_store.py"):
+        assert expected in names, expected