fix(smokes): restore three public smoke contracts broken on main

examples/loopx-managed-turn-operator-flow-smoke.py

@@ -96,6 +96,7 @@ def main() -> None:
             machine_defaults=MANAGED_DEFAULTS,
             credential_source="machine_store",
             session=None,
+            module_probe=_runtime_installed,
         )
         if channel.get("executor_endpoint") != "dsh":
             fail(f"the channel must resolve the machine's executor: {channel}")

examples/operator-provider-credential-smoke.py

@@ -59,6 +59,19 @@ def fail(message: str) -> None:
     raise SystemExit(f"operator provider credential smoke failed: {message}")
 
 
+def _runtime_installed(*_args: object, **_kwargs: object) -> bool:
+    """Report the optional managed runtime as installed.
+
+    This smoke is about which credential authenticates the managed host, not
+    about whether this host happens to have that runtime on disk, so it must not
+    read the machine it runs on. The typed ``dsh_runtime_unavailable`` refusal is
+    covered by ``examples/loopx-turn-managed-executor-binding-smoke.py`` and
+    ``tests/test_turn_managed_executor_binding.py``.
+    """
+
+    return True
+
+
 class _Handler(OperatorProviderRequestMixin):
     """Drive the real request mixin without opening a socket."""
 
@@ -133,7 +146,9 @@ def main() -> None:
             fail("an unconfigured machine must keep the individual default host")
 
         refused = managed_executor_binding(
-            "dsh", environ=operator_provider_environ(runtime_root)
+            "dsh",
+            environ=operator_provider_environ(runtime_root),
+            module_probe=_runtime_installed,
         )
         if refused.get("unavailable_reason") != OPERATOR_CREDENTIAL_UNCONFIGURED:
             fail(f"the managed host must name the missing credential: {refused}")

loopx/chat_event_cache.py

@@ -0,0 +1,126 @@
+"""Resident rows of a chat Turn's event log, with bounded finished retention.
+
+A chat Turn's event log is written once and then replayed: the SSE endpoint
+re-reads the same rows on every reconnect, and the Turn stops being written the
+moment its terminal event lands. Two obligations meet here.
+
+- Replaying a finished Turn must not re-read its log per request, so the rows a
+  reader just read stay resident.
+- Retention must stay bounded, so only the most recent finished Turns stay
+  resident and anything older falls out of the budget and is read once more.
+
+The store keeps file I/O, sequence ordering and the file lock; this module owns
+which rows are resident and which finished Turns fall out of the budget, so both
+obligations are stated once instead of being inferred from a cache miss.
+"""
+
+from __future__ import annotations
+
+import threading
+from pathlib import Path
+from typing import Any, Callable, Sequence
+
+
+TERMINAL_EVENT_CACHE_TURNS = 8
+
+EventKey = tuple[str, str]
+EventRevision = tuple[int, int, int] | None
+
+
+def event_revision(path: Path) -> EventRevision:
+    """Identify one event log's on-disk revision, or ``None`` when it is absent."""
+
+    try:
+        stat = path.stat()
+    except OSError:
+        return None
+    return stat.st_ino, stat.st_size, stat.st_mtime_ns
+
+
+class ChatEventCache:
+    """Rows of chat Turns that stay resident, bounding finished Turns.
+
+    ``rows`` and ``revisions`` stay public mappings: a caller legitimately reads
+    and seeds them, and an operator surface can count resident Turns without a
+    second accessor.
+    """
+
+    def __init__(
+        self,
+        *,
+        lock: threading.RLock,
+        budget: int = TERMINAL_EVENT_CACHE_TURNS,
+    ) -> None:
+        self.rows: dict[EventKey, list[dict[str, Any]]] = {}
+        self.revisions: dict[EventKey, EventRevision] = {}
+        self._lock = lock
+        self._budget = max(1, int(budget))
+        self._finished: dict[EventKey, None] = {}
+
+    def get(self, key: EventKey, path: Path) -> list[dict[str, Any]] | None:
+        """Return resident rows only while they still match the log on disk."""
+
+        revision = event_revision(path)
+        with self._lock:
+            rows = self.rows.get(key)
+            if rows is not None and self.revisions.get(key) == revision:
+                return rows
+        return None
+
+    def put(self, key: EventKey, rows: list[dict[str, Any]], path: Path) -> None:
+        with self._lock:
+            self.rows[key] = rows
+            self.revisions[key] = event_revision(path)
+
+    def load(
+        self,
+        key: EventKey,
+        path: Path,
+        read: Callable[[Path], list[dict[str, Any]]],
+    ) -> list[dict[str, Any]]:
+        """Return one Turn's rows, reading the log only when it cannot be served."""
+
+        cached = self.get(key, path)
+        if cached is not None:
+            return cached
+        rows = read(path)
+        self.put(key, rows, path)
+        return rows
+
+    def drop(self, key: EventKey) -> None:
+        with self._lock:
+            self.rows.pop(key, None)
+            self.revisions.pop(key, None)
+            self._finished.pop(key, None)
+
+    def retain_finished(
+        self,
+        key: EventKey,
+        rows: Sequence[dict[str, Any]],
+        *,
+        terminal_kinds: frozenset[str],
+    ) -> tuple[EventKey, ...]:
+        """Keep a finished Turn resident and evict past the budget.
+
+        Returns the keys that left the cache, so a caller can report or audit the
+        eviction instead of guessing which Turn a later read paid for. Only the
+        last row is read: a replay must not walk the rows it is about to skip.
+        """
+
+        if not rows or rows[-1].get("kind") not in terminal_kinds:
+            return ()
+        with self._lock:
+            self._finished.pop(key, None)
+            self._finished[key] = None
+            evicted: list[EventKey] = []
+            while len(self._finished) > self._budget:
+                oldest = next(iter(self._finished))
+                self._finished.pop(oldest, None)
+                self.rows.pop(oldest, None)
+                self.revisions.pop(oldest, None)
+                evicted.append(oldest)
+        return tuple(evicted)
+
+    def resident_finished_keys(self) -> tuple[EventKey, ...]:
+        with self._lock:
+            return tuple(self._finished)

loopx/chat_manager.py

@@ -4,7 +4,7 @@
 
 import hashlib
 from pathlib import Path
-from typing import Any, Mapping
+from typing import Any, Callable, Mapping
 
 from .control_plane.operator_credential import (
     env_text,
@@ -451,6 +451,7 @@ def manager_channel_binding(
     session: Mapping[str, Any] | None = None,
     machine_defaults: Mapping[str, Any] | None = None,
     credential_source: str | None = None,
+    module_probe: Callable[[str], bool] | None = None,
 ) -> dict[str, Any]:
     """Project the steward channel's resolved executor, model, and their source.
 
@@ -477,6 +478,12 @@ def manager_channel_binding(
     one. Its mode and status are quoted so a frontend can show which execution
     mode is serving the channel, and an absent Session reads as unbound rather
     than as a mode this projection guessed.
+
+    ``module_probe`` is read by the governed Turn surface exactly as it is when
+    that surface is read on its own. A caller that resolves both readbacks on a
+    host whose optional runtime is not installed has to be able to hand the same
+    probe to both, or the channel and the planned Turn can disagree about
+    whether the resolved executor can run here.
     """
 
     endpoint, endpoint_source, default_reason = _resolve_manager_endpoint(
@@ -486,7 +493,9 @@ def manager_channel_binding(
     credential_env = ""
     execution_profile: str | None = None
     if executor_kind == MANAGER_EXECUTOR_KIND_MANAGED:
-        managed = managed_executor_binding(endpoint, environ=environ)
+        managed = managed_executor_binding(
+            endpoint, environ=environ, module_probe=module_probe
+        )
         credential_env = str(managed.get("credential_env") or "")
         execution_profile = managed.get("execution_profile")
         available: bool | None = managed.get("available")

loopx/chat_store.py

@@ -14,6 +14,7 @@
 from weakref import WeakValueDictionary
 
 from .chat import require_matching_replay, resolve_attached_completion_replay
+from .chat_event_cache import ChatEventCache, event_revision
 from .file_lock import exclusive_file_lock
 
 
@@ -143,8 +144,9 @@ def __init__(self, runtime_root: Path) -> None:
         self._session_lock_guard = threading.Lock()
         self._session_locks: WeakValueDictionary[str, threading.Lock] = WeakValueDictionary()
         self._event_lock = threading.RLock()
-        self._event_cache: dict[tuple[str, str], list[dict[str, Any]]] = {}
-        self._event_cache_revision: dict[tuple[str, str], tuple[int, int, int] | None] = {}
+        self._event_log = ChatEventCache(lock=self._event_lock)
+        self._event_cache: dict[tuple[str, str], list[dict[str, Any]]] = self._event_log.rows
+        self._event_cache_revision = self._event_log.revisions
         self._event_pending: dict[tuple[str, str], list[dict[str, Any]]] = {}
         self._event_flush_locks: WeakValueDictionary[tuple[str, str], threading.Lock] = WeakValueDictionary()
         self.sessions_root.mkdir(parents=True, exist_ok=True, mode=0o700)
@@ -1291,35 +1293,23 @@ def _close_session_if_idle(
                 return True
 
     def _event_rows_locked(self, session_id: str, turn_id: str) -> list[dict[str, Any]]:
-        key = (session_id, turn_id)
-        path = self._event_path(session_id, turn_id)
-        revision = self._event_revision(path)
-        with self._event_lock:
-            if self._event_cache_revision.get(key) == revision and key in self._event_cache:
-                return self._event_cache[key]
-        rows = _read_jsonl(path)
-        with self._event_lock:
-            self._event_cache[key] = rows
-            self._event_cache_revision[key] = revision
-        return rows
+        return self._event_log.load(
+            (session_id, turn_id), self._event_path(session_id, turn_id), _read_jsonl
+        )
 
     @staticmethod
     def _event_revision(path: Path) -> tuple[int, int, int] | None:
-        try:
-            stat = path.stat()
-        except OSError:
-            return None
-        return stat.st_ino, stat.st_size, stat.st_mtime_ns
+        return event_revision(path)
 
     def _event_flush_lock(self, key: tuple[str, str]) -> threading.Lock:
         with self._event_lock:
             return self._event_flush_locks.setdefault(key, threading.Lock())
 
-    def _drop_event_cache(self, key: tuple[str, str], expected: list[dict[str, Any]] | None = None) -> None:
-        with self._event_lock:
-            if expected is None or self._event_cache.get(key) is expected:
-                self._event_cache.pop(key, None)
-                self._event_cache_revision.pop(key, None)
+    def _drop_event_cache(self, key: tuple[str, str]) -> None:
+        self._event_log.drop(key)
+
+    def _retain_terminal_event_cache(self, key: tuple[str, str], rows: list[dict[str, Any]]) -> None:
+        self._event_log.retain_finished(key, rows, terminal_kinds=TERMINAL_EVENT_KINDS)
 
     def append_event(
         self,
@@ -1370,9 +1360,7 @@ def flush_events(self, session_id: str, turn_id: str) -> int:
                         if any(row["kind"] in TERMINAL_EVENT_KINDS for row in pending):
                             self._drop_event_cache(key)
                         else:
-                            with self._event_lock:
-                                self._event_cache[key] = [*rows, *pending]
-                                self._event_cache_revision[key] = self._event_revision(path)
+                            self._event_log.put(key, [*rows, *pending], path)
                 except Exception:
                     with self._event_lock:
                         later = self._event_pending.get(key, [])
@@ -1388,19 +1376,15 @@ def events_after(self, session_id: str, turn_id: str, event_id: str | None) -> l
         self.flush_events(session_id, turn_id)
         key = (session_id, turn_id)
         path = self._event_path(session_id, turn_id)
-        revision = self._event_revision(path)
-        with self._event_lock:
-            cached = self._event_cache.get(key)
-            rows = cached if cached is not None and self._event_cache_revision.get(key) == revision else None
+        rows = self._event_log.get(key, path)
         if rows is None:
             with exclusive_file_lock(path, agent_id="loopx-chat", operation="read_chat_events"):
                 rows = self._event_rows_locked(session_id, turn_id)
         # Relies on the sequence ordering maintained by flush_events and compaction;
         # gaps are valid, but inserting or rewriting rows must preserve that order.
         start = bisect_right(rows, after, key=lambda row: int(row.get("sequence") or 0))
         result = rows[start:]
-        if rows and rows[-1].get("kind") in TERMINAL_EVENT_KINDS:
-            self._drop_event_cache(key, rows)
+        self._retain_terminal_event_cache(key, rows)
         return result
 
     def compact_completed_events(self, *, older_than_hours: float = 24.0) -> int:

tests/test_chat_event_retention.py

@@ -6,6 +6,7 @@
 from pathlib import Path
 
 import loopx.chat_store as chat_store
+from loopx.chat_event_cache import TERMINAL_EVENT_CACHE_TURNS
 from loopx.chat_store import CHAT_TURN_SCHEMA_VERSION, ChatSessionStore
 
 
@@ -39,8 +40,9 @@ def _write_completed_turn(root: Path, *, with_events: bool = True) -> tuple[Path
     return turn_path, event_path
 
 
-def test_terminal_event_history_does_not_remain_in_the_hot_cache(
+def test_completed_turn_replay_reads_its_log_once_within_a_bounded_budget(
     tmp_path: Path,
+    monkeypatch,
 ) -> None:
     store = ChatSessionStore(tmp_path)
     key = ("session", "turn")
@@ -52,14 +54,50 @@ def test_terminal_event_history_does_not_remain_in_the_hot_cache(
     )
     store.append_event(*key, kind="turn.completed", payload={})
 
+    # The writer still refuses to retain terminal history, so a completed Turn
+    # only becomes resident when a reader asks to replay it.
     assert key not in store._event_cache
     assert key not in store._event_cache_revision
-    assert [row["kind"] for row in store.events_after(*key, None)] == [
-        "assistant.delta",
-        "turn.completed",
+
+    reads: list[Path] = []
+    original_read_jsonl = chat_store._read_jsonl
+
+    def counted_read_jsonl(path: Path):
+        reads.append(path)
+        return original_read_jsonl(path)
+
+    monkeypatch.setattr(chat_store, "_read_jsonl", counted_read_jsonl)
+    for _ in range(5):
+        assert [row["kind"] for row in store.events_after(*key, None)] == [
+            "assistant.delta",
+            "turn.completed",
+        ]
+
+    # A finished Turn stops being written, so one read has to serve every later
+    # replay of it; re-reading per replay is what the throughput contract bans.
+    assert len(reads) == 1, reads
+
+
+def test_completed_turn_retention_is_bounded_to_the_replay_budget(
+    tmp_path: Path,
+) -> None:
+    store = ChatSessionStore(tmp_path)
+    keys = [
+        (f"session-{index}", "turn")
+        for index in range(TERMINAL_EVENT_CACHE_TURNS + 2)
     ]
-    assert key not in store._event_cache
-    assert key not in store._event_cache_revision
+    for key in keys:
+        store.append_event(*key, kind="assistant.delta", payload={"text": "d"}, buffered=True)
+        store.append_event(*key, kind="turn.completed", payload={})
+        store.events_after(*key, None)
+
+    # Retention is bounded by the budget, not by nothing: the oldest finished
+    # Turns are dropped and the newest stay replayable.
+    resident = store._event_log.resident_finished_keys()
+    assert len(resident) == TERMINAL_EVENT_CACHE_TURNS
+    assert len(store._event_cache) == TERMINAL_EVENT_CACHE_TURNS
+    assert keys[0] not in resident
+    assert keys[-1] in resident
 
 
 def test_keyed_locks_are_released_after_callers_drop_them(tmp_path: Path) -> None: