fix(routing): gate the codex canary check on a real turn, clear it per launch

`subagent_routing_unenforced` was posted on codex-native sessions whose
routing hooks were in fact trusted and running. Codex dispatches
`SessionStart` (the canary) when a thread's *first turn* begins, but the
enforcement watcher's first-turn gate was released by any
`thread/status/changed → active` or `item/*` event — and the MCP startup
round activates the thread and emits items without running a turn. So a
session that had not been asked anything yet (or whose first turn was
interrupted before it started) failed the canary check 30s later. Live
evidence (session e6074fb1...): thread activated by the MCP startup round
at 13:58:06, warning posted at 13:58:36, and the canary file for that same
session/app-server finally appeared at 14:01:36 when a real turn ran —
proving the hooks were trusted and effective. The stale warning stuck only
because the runner was stopped before the repair tick.

Direct probes against `codex app-server` (isolated CODEX_HOME) also
disprove the "codex captures hook trust at process start" theory: trust
written after the spawn (the shipped ordering) takes effect, even for a
turn already in flight when `config/batchWrite` lands. The real invariant
is that trust must land before the first *turn*, which `start()` already
guarantees — now written down where it can be broken.

Second fix: the canary is the proof that *this* launch's hooks ran, so
`clear_bridge_state` now drops it. The per-workspace bridge dir is reused
across launches, and a canary left by an earlier launch masked a genuine
fail-open for the rest of the session. Transition-only posting still
clears a previous launch's warning on the new forwarder's first check.

Co-authored-by: Isaac

omnigent/codex_native_app_server.py

@@ -723,6 +723,13 @@ class CodexNativeAppServer:
             self._stderr_loop(),
             name="codex-native-app-server-stderr",
         )
+        # Ordering invariant: hooks.json is written before the spawn above,
+        # and the trust handshake must complete before the first turn — codex
+        # resolves trust when it dispatches a hook, so trust landing after the
+        # spawn is fine, but a turn started before it runs unhooked. The
+        # handshake cannot precede the spawn (``hooks/list`` is an app-server
+        # RPC), so callers must not launch the TUI or dispatch a turn until
+        # ``start()`` returns.
         # Readiness failure (the app-server never came up) is fatal and
         # tears down the subprocess so it is not orphaned. Policy-hook
         # trust, by contrast, is best-effort: a trust failure degrades the

omnigent/codex_native_bridge.py

@@ -430,10 +430,17 @@ def clear_bridge_state(bridge_dir: Path) -> None:
     its current transport and thread instead of injecting into stale
     state.
 
+    The subagent-routing canary is cleared for the same reason: it is the
+    proof that *this* launch's hooks ran, so a canary left by an earlier
+    launch would mask a genuine fail-open (codex skipping untrusted
+    hooks) for the rest of the session.
+
     :param bridge_dir: Native Codex bridge directory.
     :returns: None.
     """
-    for name in (_STATE_FILE, _STARTUP_ERROR_FILE, _MCP_STARTUP_FILE):
+    from omnigent.inner.hook_scripts.codex_router_hook import CANARY_FILENAME
+
+    for name in (_STATE_FILE, _STARTUP_ERROR_FILE, _MCP_STARTUP_FILE, CANARY_FILENAME):
         try:
             (bridge_dir / name).unlink()
         except FileNotFoundError:

omnigent/codex_native_forwarder.py

@@ -1843,8 +1843,12 @@ async def supervise_forwarder(
                     # waiting forever on an idle fresh thread.
                     if not thread_active.is_set() and _event_indicates_thread_active(event):
                         thread_active.set()
-                        # Same signal releases the routing-enforcement watcher:
-                        # the canary can only exist once a turn has begun.
+                    # The routing-enforcement watcher needs a stricter signal:
+                    # codex dispatches SessionStart when a turn begins, so only
+                    # a turn-start event proves the canary should exist by now.
+                    # Thread-active alone also covers the MCP startup round,
+                    # which activates the thread without running a turn.
+                    if not turn_observed.is_set() and _event_indicates_turn_started(event):
                         turn_observed.set()
                     await _handle_event(
                         ap_client,
@@ -2200,6 +2204,23 @@ def _event_indicates_thread_active(event: CodexMessage) -> bool:
     return False
 
 
+def _event_indicates_turn_started(event: CodexMessage) -> bool:
+    """
+    Return whether a notification proves a turn has actually begun.
+
+    Codex dispatches ``SessionStart`` (the routing canary) when a thread's
+    first *turn* starts. ``thread/status/changed → active`` and ``item/*``
+    are weaker: the MCP startup round activates a thread and emits items
+    without any turn, so using them to release the routing-enforcement
+    watcher flags a session that has simply not been asked anything yet.
+
+    :param event: A Codex JSON-RPC notification envelope.
+    :returns: ``True`` for a ``turn/*`` notification.
+    """
+    method = event.get("method")
+    return isinstance(method, str) and method.startswith("turn/")
+
+
 def _is_thread_not_ready_error(exc: Exception) -> bool:
     """
     Return whether a subscription failure is Codex's fresh-thread not-ready gap.
@@ -5829,14 +5850,16 @@ async def _watch_subagent_routing_enforcement(
     Codex dispatches ``SessionStart`` when a thread's *first turn* begins,
     not at ``thread/start``, so checking before then would flag every idle
     session. *turn_observed* holds the first check until the forwarder has
-    seen the thread go active.
+    seen a ``turn/*`` event — thread activity alone also covers the MCP
+    startup round, which runs no turn and so dispatches no ``SessionStart``.
 
     :param client: HTTP client for Omnigent event posts.
     :param session_id: Omnigent conversation id.
     :param bridge_dir: Native Codex bridge directory.
     :param interval_s: Seconds between checks.
     :param turn_observed: Set by the forwarder on the thread's first
-        activity. ``None`` checks immediately (tests / resumed sessions).
+        turn-start event. ``None`` checks immediately (tests / resumed
+        sessions).
     :returns: None. Runs until cancelled.
     """
     armed = subagent_routing_armed(bridge_dir)

tests/test_codex_native_bridge.py

@@ -350,6 +350,20 @@ def test_read_mcp_startup_ignores_malformed_entries(bridge_dir: Path) -> None:
     assert read_mcp_startup(bridge_dir) == {"ok": {"status": "ready", "error": None}}
 
 
+def test_clear_bridge_state_removes_the_routing_canary(bridge_dir: Path) -> None:
+    # A canary from a previous launch would otherwise vouch for this launch's
+    # hooks, masking a fail-open for the whole session.
+    from omnigent.inner.codex_executor import codex_router_canary_fired
+    from omnigent.inner.hook_scripts.codex_router_hook import CANARY_FILENAME
+
+    (bridge_dir / CANARY_FILENAME).write_text('{"session_id": "conv_old"}\n')
+    assert codex_router_canary_fired(bridge_dir) is True
+
+    clear_bridge_state(bridge_dir)
+
+    assert codex_router_canary_fired(bridge_dir) is False
+
+
 def test_clear_bridge_state_removes_mcp_startup(bridge_dir: Path) -> None:
     """
     ``clear_bridge_state`` drops the MCP startup map with the other

tests/test_codex_native_forwarder.py

@@ -2401,6 +2401,106 @@ class _EmptyStreamCodexClient:
         self.closed = True
 
 
+class _ScriptedCodexClient(_EmptyStreamCodexClient):
+    """App-server client replaying a fixed notification script."""
+
+    def __init__(self, events: list[dict[str, Any]]) -> None:
+        super().__init__()
+        self._events = events
+
+    async def iter_events(self):  # type: ignore[no-untyped-def]
+        for event in self._events:
+            yield event
+
+
+async def _turn_gate_after(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, events: list[dict[str, Any]]
+) -> bool:
+    """Replay *events* through the forwarder and report the turn gate state."""
+    captured: dict[str, asyncio.Event] = {}
+
+    async def _fake_watch(
+        client: Any,  # type: ignore[explicit-any]
+        session_id: str,
+        bridge_dir: Path,
+        *,
+        interval_s: float = 30.0,
+        turn_observed: asyncio.Event | None = None,
+    ) -> None:
+        del client, session_id, bridge_dir, interval_s
+        assert turn_observed is not None
+        captured["gate"] = turn_observed
+        await turn_observed.wait()
+
+    monkeypatch.setattr(fwd, "_watch_subagent_routing_enforcement", _fake_watch)
+    await fwd.supervise_forwarder(
+        base_url="http://127.0.0.1:1",
+        headers={},
+        session_id="conv_gate",
+        bridge_dir=tmp_path,
+        app_server_url=str(tmp_path / "app-server.sock"),
+        thread_id="thread_gate",
+        client=_ScriptedCodexClient(events),  # type: ignore[arg-type]
+    )
+    return captured["gate"].is_set()
+
+
+async def test_mcp_startup_activity_does_not_release_the_enforcement_watcher(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """
+    Thread activation without a turn must keep the canary check parked.
+
+    Codex activates the thread (and emits items) while booting its MCP
+    servers, but dispatches ``SessionStart`` only when a turn begins, so
+    releasing the watcher here flagged ``subagent_routing_unenforced`` on
+    sessions that had simply not been asked anything yet.
+    """
+    released = await _turn_gate_after(
+        monkeypatch,
+        tmp_path,
+        [
+            {
+                "method": "thread/status/changed",
+                "params": {"threadId": "thread_gate", "status": {"type": "active"}},
+            },
+            {"method": "item/started", "params": {"threadId": "thread_gate"}},
+            {
+                "method": "thread/status/changed",
+                "params": {"threadId": "thread_gate", "status": {"type": "idle"}},
+            },
+        ],
+    )
+
+    assert released is False
+
+
+async def test_turn_start_releases_the_enforcement_watcher(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """A real turn releases the canary check — that is when SessionStart runs."""
+    released = await _turn_gate_after(
+        monkeypatch,
+        tmp_path,
+        [{"method": "turn/started", "params": {"threadId": "thread_gate"}}],
+    )
+
+    assert released is True
+
+
+def test_event_indicates_turn_started_only_for_turn_events() -> None:
+    assert fwd._event_indicates_turn_started({"method": "turn/started"}) is True
+    assert fwd._event_indicates_turn_started({"method": "turn/completed"}) is True
+    assert fwd._event_indicates_turn_started({"method": "item/started"}) is False
+    assert (
+        fwd._event_indicates_turn_started(
+            {"method": "thread/status/changed", "params": {"status": {"type": "active"}}}
+        )
+        is False
+    )
+    assert fwd._event_indicates_turn_started({}) is False
+
+
 async def test_enforcement_watcher_does_not_leak_on_a_session_without_turns(
     tmp_path: Path,
 ) -> None: