diff --git a/examples/loopx-managed-turn-operator-flow-smoke.py b/examples/loopx-managed-turn-operator-flow-smoke.py index f11ed160b6..b3badaad28 100644 --- a/examples/loopx-managed-turn-operator-flow-smoke.py +++ b/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}") diff --git a/examples/operator-provider-credential-smoke.py b/examples/operator-provider-credential-smoke.py index 44940527d2..f6ff9082df 100644 --- a/examples/operator-provider-credential-smoke.py +++ b/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}") diff --git a/loopx/chat_event_cache.py b/loopx/chat_event_cache.py new file mode 100644 index 0000000000..b741de2af6 --- /dev/null +++ b/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) diff --git a/loopx/chat_manager.py b/loopx/chat_manager.py index a70af92a87..f5c4c97e8a 100644 --- a/loopx/chat_manager.py +++ b/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") diff --git a/loopx/chat_store.py b/loopx/chat_store.py index 7ead472c8b..f2a347b383 100644 --- a/loopx/chat_store.py +++ b/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,10 +1376,7 @@ 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) @@ -1399,8 +1384,7 @@ def events_after(self, session_id: str, turn_id: str, event_id: str | None) -> l # 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: diff --git a/tests/test_chat_event_retention.py b/tests/test_chat_event_retention.py index 811415ac0a..af4093d9b4 100644 --- a/tests/test_chat_event_retention.py +++ b/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: