fix(semantics): report the formal detail ratios from what the checks walked

examples/semantic-vocabulary-drift-smoke.py

@@ -771,15 +771,28 @@ def summarise_formal_domains(registry: dict[str, Any], sources: list[SourceFile]
     covered = [name for name in kernel if 'producers' in vocabularies[name]]
     unverified = [name for name in cross_runtime if 'producers' not in vocabularies[name]]
     scanned, tracked = producer_scan_reach(sources)
-    projections = len(registry['projections'])
-    contexts = sum(len(entry['contexts']) for entry in registry['scope_declarations'].values())
+    # Both sides of these ratios came from one expression, so they printed 100%
+    # by construction: ``projections={len(registry['projections'])}`` over
+    # itself reported full coverage however many registered projections the
+    # check never executed. The pair now comes from the same code-owned
+    # selector ``check_invariant_domain`` validates the declared domain
+    # against, so the detail line cannot disagree with the invariant above it.
+    projections_walked, projections_registered = FORMAL_DOMAIN_SELECTORS["projections[*]"](registry)
+    # F4's selector still derives both sides from the declaration list. That is
+    # not a second self-satisfying ratio but a fail-closed count:
+    # ``check_scope_declarations`` validates every declared context against the
+    # modules the inventory really found, or raises. The ratio therefore reads
+    # 100% whenever the smoke gets far enough to print it, and what it reports
+    # is how many contexts that check had to clear.
+    contexts_walked, contexts_registered = FORMAL_DOMAIN_SELECTORS["scope_declarations[*].contexts"](registry)
     return (
         f"formal_domain={sizes} (verified/registered)\n"
         f"  formal_domain_bounds: kernel_with_producers={len(covered)}/{len(kernel)}"
         f" cross_runtime_unverified={len(unverified)}/{len(cross_runtime)}"
         f" producer_scan_reach={scanned}/{tracked}_files"
-        f" projections={projections}/{projections}"
-        f" scope_declarations={len(registry['scope_declarations'])} declared_contexts={contexts}/{contexts}"
+        f" projections={projections_walked}/{projections_registered}"
+        f" scope_declarations={len(registry['scope_declarations'])}"
+        f" declared_contexts={contexts_walked}/{contexts_registered}"
     )
 
 

loopx/chat_server.py

@@ -1055,18 +1055,21 @@ def _action_preview(self) -> None:
             status=201,
         )
 
+    def _action_not_found(self) -> None:
+        self._send_error(
+            "typed Chat action proposal was not found",
+            status=404,
+            error_code="action_not_found",
+        )
+
     def _action_snapshot(self, proposal_id: str) -> None:
         try:
             proposal = self.server.action_service.load(proposal_id)
         except ValueError as exc:
             self._send_error(str(exc), status=400, error_code="invalid_proposal_id")
             return
         if proposal is None:
-            self._send_error(
-                "typed Chat action proposal was not found",
-                status=404,
-                error_code="action_not_found",
-            )
+            self._action_not_found()
             return
         self._send_json(
             {
@@ -1107,11 +1110,7 @@ def _action_cancel(self, proposal_id: str) -> None:
                 raise ValueError("action cancel request must be empty")
             proposal = self.server.action_service.cancel(proposal_id)
         except KeyError:
-            self._send_error(
-                "typed Chat action proposal was not found",
-                status=404,
-                error_code="action_not_found",
-            )
+            self._action_not_found()
             return
         except ActionConflictError as exc:
             self._send_error(str(exc), status=409, error_code="action_conflict")
@@ -1144,11 +1143,7 @@ def _action_transition(self, proposal_id: str, transition: str) -> None:
             else:
                 raise ValueError("unsupported action transition")
         except KeyError:
-            self._send_error(
-                "typed Chat action proposal was not found",
-                status=404,
-                error_code="action_not_found",
-            )
+            self._action_not_found()
             return
         except ActionConflictError as exc:
             self._send_error(str(exc), status=409, error_code="action_conflict")
@@ -1190,11 +1185,7 @@ def _action_apply(self, proposal_id: str) -> None:
             )
             return
         except KeyError:
-            self._send_error(
-                "typed Chat action proposal was not found",
-                status=404,
-                error_code="action_not_found",
-            )
+            self._action_not_found()
             return
         except ActionConflictError as exc:
             self._send_error(str(exc), status=409, error_code="action_conflict")

loopx/extensions/lark/goal_topic_connections.py

@@ -92,6 +92,7 @@
     IngressMode,
     ReplyMode,
     _routing_value,
+    _connection_routing_modes,
     decide_lark_topic_route_event,
 )
 from .presentation.kanban import (
@@ -1035,29 +1036,7 @@ def list_lark_connections(
             )
             connector_status: dict[str, Any] | None = None
             try:
-                capture_scope = _routing_value(
-                    CaptureScope,
-                    routing.get("capture_scope")
-                    or (
-                        "configured_chat_all"
-                        if routing.get("incoming_mode") == "all"
-                        else "addressed_only"
-                    ),
-                    default=CaptureScope.ADDRESSED_ONLY.value,
-                    field="capture_scope",
-                )
-                ingress_mode = _routing_value(
-                    IngressMode,
-                    routing.get("ingress_mode"),
-                    default=IngressMode.DIRECT_SESSION.value,
-                    field="ingress_mode",
-                )
-                reply_mode = _routing_value(
-                    ReplyMode,
-                    routing.get("reply_mode"),
-                    default=ReplyMode.TOPIC_REPLY.value,
-                    field="reply_mode",
-                )
+                capture_scope, ingress_mode, reply_mode = _connection_routing_modes(routing)
                 raw_connector = binding.get("connector")
                 if raw_connector is not None:
                     if not isinstance(raw_connector, Mapping):
@@ -1216,29 +1195,7 @@ def decide_lark_topic_event(
                 else {}
             )
             try:
-                capture_scope = _routing_value(
-                    CaptureScope,
-                    routing.get("capture_scope")
-                    or (
-                        "configured_chat_all"
-                        if routing.get("incoming_mode") == "all"
-                        else "addressed_only"
-                    ),
-                    default=CaptureScope.ADDRESSED_ONLY.value,
-                    field="capture_scope",
-                )
-                ingress_mode = _routing_value(
-                    IngressMode,
-                    routing.get("ingress_mode"),
-                    default=IngressMode.DIRECT_SESSION.value,
-                    field="ingress_mode",
-                )
-                reply_mode = _routing_value(
-                    ReplyMode,
-                    routing.get("reply_mode"),
-                    default=ReplyMode.TOPIC_REPLY.value,
-                    field="reply_mode",
-                )
+                capture_scope, ingress_mode, reply_mode = _connection_routing_modes(routing)
                 connector = binding.get("connector")
                 if connector is not None:
                     if not isinstance(connector, Mapping):

loopx/extensions/lark/goal_topic_routing.py

@@ -47,6 +47,37 @@ def _routing_value(
         raise ValueError(f"{field} must be one of: {allowed}") from exc
 
 
+def _connection_routing_modes(
+    routing: Mapping[str, Any],
+) -> tuple[str, str, str]:
+    """Normalize persisted modes for both connection readback and event routing."""
+
+    capture_scope = _routing_value(
+        CaptureScope,
+        routing.get("capture_scope")
+        or (
+            "configured_chat_all"
+            if routing.get("incoming_mode") == "all"
+            else "addressed_only"
+        ),
+        default=CaptureScope.ADDRESSED_ONLY.value,
+        field="capture_scope",
+    )
+    ingress_mode = _routing_value(
+        IngressMode,
+        routing.get("ingress_mode"),
+        default=IngressMode.DIRECT_SESSION.value,
+        field="ingress_mode",
+    )
+    reply_mode = _routing_value(
+        ReplyMode,
+        routing.get("reply_mode"),
+        default=ReplyMode.TOPIC_REPLY.value,
+        field="reply_mode",
+    )
+    return capture_scope, ingress_mode, reply_mode
+
+
 def _normalize_mention_name(name: str) -> str:
     cleaned = str(name or "").strip()
     if cleaned.startswith("@"):

tests/architecture/test_semantic_vocabulary_drift.py

@@ -520,6 +520,41 @@ def test_f1_f2_domain_names_exactly_the_vocabularies_the_producer_check_walks()
         assert domain["evidence_bound"] == "producer_scan_reach"
 
 
+def test_the_formal_detail_line_cannot_report_a_projection_it_never_executed() -> None:
+    """A ratio whose two sides come from one expression cannot show a gap.
+
+    ``projections`` was printed as ``len(registry['projections'])`` on both
+    sides of the slash, so the detail line read 100% however many registered
+    projections ``check_projections`` never imports. The invariant above it had
+    already been fixed to count only executed projections, so the report and
+    the invariant could disagree while both stayed green.
+    """
+    smoke = runpy.run_path(str(SMOKE))
+    registry = copy.deepcopy(smoke["load_registry"]())
+    assert "projections=1/1" in smoke["summarise_formal_domains"](registry, [])
+
+    registry["projections"]["never_executed"] = {
+        "owner": "loopx/nowhere.py::absent",
+        "mapping": {},
+    }
+    detail = smoke["summarise_formal_domains"](registry, [])
+    assert "projections=1/2" in detail, detail
+
+
+def test_the_formal_detail_line_agrees_with_the_selectors_it_reports() -> None:
+    """The detail line reports the pair the invariant's domain is checked on."""
+    smoke = runpy.run_path(str(SMOKE))
+    registry = smoke["load_registry"]()
+    selectors = smoke["FORMAL_DOMAIN_SELECTORS"]
+    detail = smoke["summarise_formal_domains"](registry, [])
+    for selector, label in (
+        ("projections[*]", "projections"),
+        ("scope_declarations[*].contexts", "declared_contexts"),
+    ):
+        walked, registered = selectors[selector](registry)
+        assert f"{label}={walked}/{registered}" in detail, (label, detail)
+
+
 @pytest.mark.parametrize("denial", [
     " The cross_runtime tier declares no producers.",
     " No vocabulary in the cross_runtime tier has any producer.",

tests/extensions/test_lark_goal_topic_connections.py

@@ -2122,6 +2122,75 @@ def _prep_goal_channel_target(root: Path) -> Path:
     return target_path
 
 
+@pytest.mark.parametrize(
+    ("routing", "expected"),
+    [
+        ({}, ("addressed_only", "direct_session", "topic_reply")),
+        (
+            {"incoming_mode": "all"},
+            ("configured_chat_all", "direct_session", "topic_reply"),
+        ),
+        (
+            {
+                "incoming_mode": "all",
+                "capture_scope": " ADDRESSED_ONLY ",
+                "ingress_mode": " SESSION_QUEUE ",
+                "reply_mode": " TOPIC_REPLY ",
+            },
+            ("addressed_only", "session_queue", "topic_reply"),
+        ),
+        ({"capture_scope": "invalid"}, None),
+        ({"ingress_mode": "async-inbox"}, None),
+        ({"reply_mode": "invalid"}, None),
+    ],
+)
+def test_connection_readback_and_event_route_share_persisted_mode_rules(
+    tmp_path: Path,
+    routing: dict[str, str],
+    expected: tuple[str, str, str] | None,
+) -> None:
+    target_path = _prep_goal_channel_target(tmp_path)
+    binding_path = tmp_path / "binding.json"
+    payload = _legacy_v0_binding_payload("om_topic_alpha", "agent-alpha")
+    payload["bindings"]["goal-alpha"]["routing"] = routing
+    write_goal_channel_binding(binding_path, payload)
+    before = binding_path.read_bytes()
+    rows = list_lark_connections(
+        registry=_registry(tmp_path),
+        target_path=target_path,
+        binding_paths={"goal-alpha": binding_path},
+        runner=_runner({}),
+    )
+    decision = decide_lark_topic_event(
+        target_payload=read_goal_channel_targets(target_path),
+        binding_payloads={"goal-alpha": read_goal_channel_binding(binding_path)},
+        event={
+            "chat_id": CHAT_ID,
+            "root_id": "om_topic_alpha",
+            "message_id": "om_incoming",
+            "content": "@mew bot hello",
+        },
+    )
+    assert len(rows) == 1
+    if expected is None:
+        assert rows[0]["reply_ready"] is False
+        assert rows[0]["health_error_code"] == "invalid_routing_state"
+        assert decision == {
+            "matched": False,
+            "reason": "invalid_routing_state",
+            "route": None,
+        }
+    else:
+        assert rows[0]["reply_ready"] is True
+        assert decision["matched"] is True
+        for key, value in zip(
+            ("capture_scope", "ingress_mode", "reply_mode"), expected
+        ):
+            assert rows[0][key] == value
+            assert decision["route"][key] == value
+    assert binding_path.read_bytes() == before
+
+
 def test_reconnect_after_upgrade_reuses_legacy_topic_root_without_resend(
     tmp_path: Path,
 ) -> None:

tests/extensions/test_lark_goal_topic_runtime.py

@@ -5,6 +5,7 @@
 import subprocess
 import threading
 from collections.abc import Mapping
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
 
@@ -283,6 +284,21 @@ def test_mention_uses_existing_inbox_reply_and_ack_path(tmp_path: Path) -> None:
     assert projection["processed_count"] == 1
 
 
+@pytest.fixture
+def manager_context_clock(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Keep the dated fixture within retention without disabling compaction."""
+    from loopx.extensions.lark import manager_context
+
+    class FixtureDatetime(datetime):
+        @classmethod
+        def now(cls, tz=None):
+            instant = datetime(2026, 9, 13, 6, 1, tzinfo=UTC)
+            return instant.astimezone(tz) if tz is not None else instant.replace(tzinfo=None)
+
+    monkeypatch.setattr(manager_context, "datetime", FixtureDatetime)
+
+
+@pytest.mark.usefixtures("manager_context_clock")
 def test_manager_captures_unaddressed_context_without_granting_turn_authority(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
@@ -467,6 +483,7 @@ def decision(**options: Any) -> dict[str, Any]:
     assert answer_calls == []
 
 
+@pytest.mark.usefixtures("manager_context_clock")
 def test_manager_authorized_turn_quietly_recovers_history_as_context(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None: