From 4af78ae474fae786e823e56f81a6cc2cefc2278b Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:39:38 -0700 Subject: [PATCH] =?UTF-8?q?refactor(tui):=20render/slash=5Ffuzzy/turn=5Fma?= =?UTF-8?q?rker/synthetic=5Fturn/slash=5Fworker=20=E2=80=94=20shared=20imp?= =?UTF-8?q?ort=20bridge,=20tier=20table,=20marker=20update=20helper?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tui_gateway/render.py | 43 +++++-------- tui_gateway/slash_fuzzy.py | 78 +++++++++-------------- tui_gateway/slash_worker.py | 60 ++++++++---------- tui_gateway/synthetic_turn.py | 91 +++++++++------------------ tui_gateway/turn_marker.py | 112 ++++++++++++++-------------------- 5 files changed, 144 insertions(+), 240 deletions(-) diff --git a/tui_gateway/render.py b/tui_gateway/render.py index c15ddef7c0..5a7adeda49 100644 --- a/tui_gateway/render.py +++ b/tui_gateway/render.py @@ -6,44 +6,31 @@ everything returns None and the TUI falls back to its own markdown.tsx. from __future__ import annotations +import importlib -def render_message(text: str, cols: int = 80) -> str | None: + +def _rich(name: str, *args, cols: int): + """Call ``agent.rich_output.(*args, cols=cols)``; retry without ``cols`` for older + signatures; None when the module is missing or the renderer fails.""" try: - from agent.rich_output import format_response - except ImportError: + fn = getattr(importlib.import_module("agent.rich_output"), name) + except (ImportError, AttributeError): return None - try: - return format_response(text, cols=cols) + return fn(*args, cols=cols) except TypeError: - return format_response(text) + return fn(*args) except Exception: return None +def render_message(text: str, cols: int = 80) -> str | None: + return _rich("format_response", text, cols=cols) + + def render_diff(text: str, cols: int = 80) -> str | None: - try: - from agent.rich_output import render_diff as _rd - except ImportError: - return None - - try: - return _rd(text, cols=cols) - except TypeError: - return _rd(text) - except Exception: - return None + return _rich("render_diff", text, cols=cols) def make_stream_renderer(cols: int = 80): - try: - from agent.rich_output import StreamingRenderer - except ImportError: - return None - - try: - return StreamingRenderer(cols=cols) - except TypeError: - return StreamingRenderer() - except Exception: - return None + return _rich("StreamingRenderer", cols=cols) diff --git a/tui_gateway/slash_fuzzy.py b/tui_gateway/slash_fuzzy.py index 254b811e3a..01031ef631 100644 --- a/tui_gateway/slash_fuzzy.py +++ b/tui_gateway/slash_fuzzy.py @@ -1,12 +1,11 @@ """Description-aware fuzzy scoring for slash-menu completions. -Ported from superagent-ai/grok-cli ``src/ui/slash-menu.ts`` (mirrored on the -TUI client in ``ui-tui/src/app/slash/fuzzyScore.ts``): candidates are scored -in tiers — exact match on the command token (0), prefix (1), substring (2) — -and the DESCRIPTION text is tokenized and matched at a +3 offset (exact word -3, word prefix 4, word substring 5). Typing ``/summary`` thus surfaces a -command whose description mentions summaries even though no command name -starts with it. Lower score wins; ``math.inf`` means no match. +Ported from superagent-ai/grok-cli ``src/ui/slash-menu.ts`` (mirrored on the TUI client in +``ui-tui/src/app/slash/fuzzyScore.ts``): candidates are scored in tiers — exact match on the +command token (0), prefix (1), substring (2) — and the DESCRIPTION text is tokenized and matched +at a +3 offset (exact word 3, word prefix 4, word substring 5). Typing ``/summary`` thus surfaces +a command whose description mentions summaries even though no command name starts with it. +Lower score wins; ``math.inf`` means no match. """ from __future__ import annotations @@ -17,6 +16,13 @@ from typing import Callable _TOKEN_SPLIT = re.compile(r"[^a-z0-9]+") +# (tier bump, field predicate) in priority order: exact, prefix, substring. +_TIERS = ( + (0, lambda field, q: field == q or f"/{field}" == q), + (1, lambda field, q: field.startswith(q) or f"/{field}".startswith(q)), + (2, lambda field, q: q in field), +) + def tokenize_search_text(value: str) -> list[str]: """Lowercase ``value`` and return it alongside its alphanumeric words.""" @@ -30,31 +36,16 @@ def normalize_slash_search_query(query: str) -> str: def _score_fields(fields: list[str], query: str, offset: int) -> float: - for field in fields: - if field == query or f"/{field}" == query: - return offset - for field in fields: - if field.startswith(query) or f"/{field}".startswith(query): - return offset + 1 - for field in fields: - if query in field: - return offset + 2 - return math.inf + return next((offset + bump for bump, pred in _TIERS if any(pred(f, query) for f in fields)), math.inf) def score_slash_completion_item(item: dict, query: str) -> float: - """Score one completion item dict (``text`` + ``meta``) against ``query``. - - ``text`` is the replacement token (may carry a leading slash or trailing - space); ``meta`` is the human description. Lower is better; ``math.inf`` - means no match at all. - """ + """Score one completion item dict against ``query``: ``text`` is the replacement token (may carry + a leading slash or trailing space); ``meta`` is the human description. Lower is better.""" name = str(item.get("text", "")).strip().lstrip("/") - command_fields = tokenize_search_text(name) - description_fields = tokenize_search_text(str(item.get("meta", ""))) return min( - _score_fields(command_fields, query, 0), - _score_fields(description_fields, query, 3), + _score_fields(tokenize_search_text(name), query, 0), + _score_fields(tokenize_search_text(str(item.get("meta", ""))), query, 3), ) @@ -63,29 +54,16 @@ def fuzzy_rank_slash_items( ) -> tuple[list[dict], Callable[[dict], float]]: """Merge description/substring matches into ``items`` and sort by score. - ``items`` are the completer's own (prefix-filtered) rows and keep their - identity; ``catalog`` is the full command/skill universe, from which any - entry the prefix filter missed but the fuzzy scorer matches is appended. - Returns the score-sorted rows (stable within a tier) plus a ``score_of`` - lookup for downstream rankers to use as a leading sort key. + ``items`` are the completer's own (prefix-filtered) rows and keep their identity; ``catalog`` is + the full command/skill universe, from which any entry the prefix filter missed but the fuzzy + scorer matches is appended. Returns the score-sorted rows (stable within a tier) plus a + ``score_of`` lookup for downstream rankers to use as a leading sort key. """ seen = {str(item.get("text", "")).strip() for item in items} - merged = list(items) - for item in catalog: - if str(item.get("text", "")).strip() in seen: - continue - if not math.isinf(score_slash_completion_item(item, query)): - merged.append(item) - - scores: dict[int, float] = {} - scored: list[tuple[float, int, dict]] = [] - for index, item in enumerate(merged): - score = score_slash_completion_item(item, query) - if math.isinf(score): - continue - scores[id(item)] = score - scored.append((score, index, item)) - scored.sort(key=lambda entry: (entry[0], entry[1])) - - ranked = [item for _, _, item in scored] + merged = list(items) + [ + item for item in catalog + if str(item.get("text", "")).strip() not in seen and not math.isinf(score_slash_completion_item(item, query)) + ] + scores = {id(item): score_slash_completion_item(item, query) for item in merged} + ranked = sorted((item for item in merged if not math.isinf(scores[id(item)])), key=lambda item: scores[id(item)]) return ranked, lambda item: scores.get(id(item), math.inf) diff --git a/tui_gateway/slash_worker.py b/tui_gateway/slash_worker.py index 7e55e22949..d7b4b33162 100644 --- a/tui_gateway/slash_worker.py +++ b/tui_gateway/slash_worker.py @@ -3,12 +3,11 @@ Protocol: reads JSON lines from stdin {id, command}, writes {id, ok, output|error} to stdout. """ -# Stop a ``utils/`` (or ``proxy/``, ``ui/``) package in the launch directory -# from shadowing Hermes's own top-level modules. This worker is spawned as -# ``-m tui_gateway.slash_worker`` and inherits the user's CWD, so the ``import -# cli`` below would otherwise resolve ``utils`` to a colliding local package and -# crash the child in a retry loop. ``hermes_bootstrap`` lives at the repo root -# (no collision risk), so importing it before the guard runs is safe. +# Stop a ``utils/`` (or ``proxy/``, ``ui/``) package in the launch directory from shadowing Hermes's +# own top-level modules. This worker is spawned as ``-m tui_gateway.slash_worker`` and inherits the +# user's CWD, so the ``import cli`` below would otherwise resolve ``utils`` to a colliding local +# package and crash the child in a retry loop. ``hermes_bootstrap`` lives at the repo root (no +# collision risk), so importing it before the guard runs is safe. import hermes_bootstrap hermes_bootstrap.harden_import_path() @@ -42,11 +41,8 @@ def _is_orphaned(original_ppid, getppid=os.getppid) -> bool: def _prepare_slash_worker_runtime() -> None: - """Start bounded MCP discovery before HermesCLI snapshots tools. - - Each slash_worker child is its own process — the parent ``hermes serve`` - discovery thread does not populate this registry. - """ + """Start bounded MCP discovery before HermesCLI snapshots tools: each slash_worker child is its + own process — the parent ``hermes serve`` discovery thread does not populate this registry.""" from hermes_cli.mcp_startup import start_background_mcp_discovery, wait_for_mcp_discovery start_background_mcp_discovery(logger=logger, thread_name="slash-worker-mcp-discovery") @@ -71,27 +67,22 @@ def _run(cli: HermesCLI, command: str) -> str: return "" if not cmd.startswith("/"): cmd = f"/{cmd}" - buf = io.StringIO() - - # Rich Console captures its file handle at construction, so redirect_stdout - # won't affect it; swap the console's file so self.console.print() is captured. + # Rich Console captures its file handle at construction, so redirect_stdout won't affect it; + # swap the console's file so self.console.print() is captured. cli.console = Console(file=buf, force_terminal=True, width=120) - old = getattr(cli_mod, "_cprint", None) if old is not None: cli_mod._cprint = lambda text: print(text) - try: with contextlib.redirect_stdout(buf), contextlib.redirect_stderr(buf): cli.process_command(cmd) finally: if old is not None: cli_mod._cprint = old - - # Desktop chat bubbles render plain text, not ANSI. A command that emits - # Rich color (e.g. /journey building its own Console under the gateway's - # inherited COLORTERM) would leak raw escapes; strip at this single choke point. + # Desktop chat bubbles render plain text, not ANSI. A command that emits Rich color (e.g. + # /journey building its own Console under the gateway's inherited COLORTERM) would leak raw + # escapes; strip at this single choke point. from tools.ansi_strip import strip_ansi return strip_ansi(buf.getvalue().rstrip()) @@ -106,48 +97,47 @@ def main(): os.environ["HERMES_SESSION_KEY"] = args.session_key os.environ["HERMES_INTERACTIVE"] = "1" - # Start before the (hundreds-of-ms) HermesCLI build — that window is itself - # an orphan risk if the gateway dies mid-spawn. + # Start before the (hundreds-of-ms) HermesCLI build — that window is itself an orphan risk if + # the gateway dies mid-spawn. _start_parent_death_watchdog(os.getppid()) _prepare_slash_worker_runtime() with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()): cli = HermesCLI(model=args.model or None, compact=True, resume=args.session_key, verbose=False) - # Spurious stdin-EOF recovery (same shared-file-description O_NONBLOCK issue - # as the gateway entry point — any child inheriting fd 0 can flip the flag). + # Spurious stdin-EOF recovery (same shared-file-description O_NONBLOCK issue as the gateway + # entry point — any child inheriting fd 0 can flip the flag). _sw_recovery_times: list[float] = [] def _sw_log(reason: str) -> None: print(f"[slash-worker] {reason}", file=sys.stderr, flush=True) + def _reply(**fields) -> None: + sys.stdout.write(json.dumps(fields) + "\n") + sys.stdout.flush() + while True: raw = sys.stdin.readline() if not raw: if not handle_spurious_eof(_sw_recovery_times, _sw_log): break continue - line = raw.strip() if not line: continue - _in_flight.set() rid = None try: req = json.loads(line) rid = req.get("id") - out = _run(cli, req.get("command", "")) - sys.stdout.write(json.dumps({"id": rid, "ok": True, "output": out}) + "\n") - sys.stdout.flush() + _reply(id=rid, ok=True, output=_run(cli, req.get("command", ""))) except Exception as e: - sys.stdout.write(json.dumps({"id": rid, "ok": False, "error": str(e)}) + "\n") - sys.stdout.flush() + _reply(id=rid, ok=False, error=str(e)) finally: _in_flight.clear() - # Workers persist for the TUI session: release allocator pages at the - # command boundary like other long-lived gateway processes - # (trim_memory's shared cooldown coalesces nearby activity). + # Workers persist for the TUI session: release allocator pages at the command boundary + # like other long-lived gateway processes (trim_memory's shared cooldown coalesces + # nearby activity). try: from hermes_cli.mem_trim import trim_memory diff --git a/tui_gateway/synthetic_turn.py b/tui_gateway/synthetic_turn.py index a1df1965a5..6c1fd45c59 100644 --- a/tui_gateway/synthetic_turn.py +++ b/tui_gateway/synthetic_turn.py @@ -1,22 +1,15 @@ """Synthetic GIL-heavy turn driver for the AC-4 isolation certify harness. -The regime under test is interpreter-wide GIL starvation: concurrent heavy agent -turns run compute in threads of the SERVING process and starve the event loop -that flushes WebSocket frames (loop thread parked in ``take_gil`` — NOT blocked -on I/O). To certify the isolation fix without spending real tokens, the turn -driver must reproduce THAT: sustained pure-Python CPU holding the GIL for the -turn's duration. A network/sleep stub is WRONG — it releases the GIL during I/O -and never reproduces ``take_gil`` contention, so a green off it is fake. +The regime under test is interpreter-wide GIL starvation: concurrent heavy agent turns run compute in +threads of the SERVING process and starve the event loop that flushes WebSocket frames (loop thread +parked in ``take_gil`` — NOT blocked on I/O). To certify the isolation fix without spending real +tokens the turn driver must reproduce THAT: sustained pure-Python CPU holding the GIL. A network/ +sleep stub is WRONG — it releases the GIL during I/O, so a green off it is fake. -This module is a **test seam**: dead unless ``HERMES_ISO_CERTIFY_SYNTH_TURN=1``. -When armed, ``tui_gateway.server._make_agent`` returns a -:class:`SyntheticHeavyAgent` instead of a real ``AIAgent``. Both the in-process -``_pool`` path (isolation OFF) and the compute-host child path (isolation ON) -build through ``_make_agent``, so the isolation boundary is the only variable. - -Per-turn intensity (wall duration, CPU chunk size, delta cadence, token -accounting) rides in the prompt text as a JSON object; any other prompt falls -back to env / built-in defaults. +Test seam: dead unless ``HERMES_ISO_CERTIFY_SYNTH_TURN=1``. When armed, ``server._make_agent`` returns +a :class:`SyntheticHeavyAgent` on both the in-process ``_pool`` path (isolation OFF) and the compute- +host child path (isolation ON), so the isolation boundary is the only variable. Per-turn intensity +rides in the prompt text as a JSON object; any other prompt falls back to env / built-in defaults. """ from __future__ import annotations @@ -38,10 +31,10 @@ def synth_turn_armed() -> bool: class SyntheticHeavyAgent: """An AIAgent-shaped object whose turn is a GIL-holding CPU burn. - Presents only the surface ``tui_gateway.server``'s turn path and status - helpers read (``run_conversation``/``interrupt``/``clear_interrupt`` plus the - ``model``/``provider``/``session_*`` attributes consumed by ``_get_usage`` and - ``_session_info``). Never opens a socket or spawns a subprocess. + Presents only the surface ``tui_gateway.server``'s turn path and status helpers read + (``run_conversation``/``interrupt``/``clear_interrupt`` plus the ``model``/``provider``/ + ``session_*`` attributes consumed by ``_get_usage`` and ``_session_info``). Never opens a socket + or spawns a subprocess. """ def __init__(self, session_id: str, *, model: str = "synthetic-heavy") -> None: @@ -49,9 +42,7 @@ class SyntheticHeavyAgent: self.model = model self.provider = "synthetic" self.api_mode = "chat_completions" - self.base_url = "" - self.api_key = "" - self.platform = "" + self.base_url = self.api_key = self.platform = "" self.tools: list[Any] = [] self.reasoning_config: dict | None = None self.service_tier: str | None = None @@ -59,17 +50,13 @@ class SyntheticHeavyAgent: self._config_context_length = 200_000 self._cached_system_prompt = "" # Cumulative session counters (read by _get_usage → status bar). - self.session_input_tokens = 0 - self.session_output_tokens = 0 - self.session_prompt_tokens = 0 - self.session_completion_tokens = 0 - self.session_reasoning_tokens = 0 - self.session_total_tokens = 0 + self.session_input_tokens = self.session_output_tokens = self.session_prompt_tokens = 0 + self.session_completion_tokens = self.session_reasoning_tokens = self.session_total_tokens = 0 self.session_api_calls = 0 self.history: list[dict[str, str]] = [] self._interrupt = threading.Event() - # ── interrupt contract (mirrors AIAgent) ─────────────────────────── + # interrupt contract (mirrors AIAgent) def clear_interrupt(self) -> None: self._interrupt.clear() @@ -96,15 +83,15 @@ class SyntheticHeavyAgent: return { # Wall-clock seconds of GIL-holding compute. "duration_s": float(spec.get("duration_s", _env_float("HERMES_ISO_CERTIFY_DURATION_S", 8.0))), - # Pure-Python ops per interrupt-check chunk: small enough that an - # interrupt lands within ms, large enough to stay hot on the GIL. + # Pure-Python ops per interrupt-check chunk: small enough that an interrupt lands within + # ms, large enough to stay hot on the GIL. "chunk": int(spec.get("chunk", _env_int("HERMES_ISO_CERTIFY_CHUNK", 20_000))), # Streamed-delta cadence: each delta is a loop wakeup marshalling a frame. "delta_interval_s": float(spec.get("delta_interval_s", _env_float("HERMES_ISO_CERTIFY_DELTA_S", 0.05))), # Notional output tokens per delta (drives the 100K+-token heavy-turn proxy). "tokens_per_delta": int(spec.get("tokens_per_delta", _env_int("HERMES_ISO_CERTIFY_TPD", 512))), - # Optional per-chunk sleep for a mixed regime (0 = pure burn). --dry-run - # uses a short duration, NOT a sleep, so it still exercises the real seam. + # Optional per-chunk sleep for a mixed regime (0 = pure burn). --dry-run uses a short + # duration, NOT a sleep, so it still exercises the real seam. "sleep_s": float(spec.get("sleep_s", 0.0)), } @@ -136,8 +123,8 @@ class SyntheticHeavyAgent: now = time.monotonic() if now - start >= duration: break - # A tight integer loop never releases the GIL — the exact contention - # that starves the serving loop. + # A tight integer loop never releases the GIL — the exact contention that starves the + # serving loop. for _ in range(chunk): acc = (acc * 1_000_003 + 12_345) & 0xFFFFFFFFFFFFFFFF if sleep_s: @@ -152,34 +139,20 @@ class SyntheticHeavyAgent: last_delta = now self.session_api_calls += 1 - # Fold the checksum into the reply so the loop can't be eliminated and - # the turn is deterministic and inspectable. + # Fold the checksum into the reply so the loop can't be eliminated and the turn is + # deterministic and inspectable. final = ( f"[synthetic heavy turn] deltas={deltas} " f"out_tokens={self.session_output_tokens} " f"interrupted={interrupted} checksum={acc & 0xFFFF:04x}" ) - messages = [ - *base_history, - {"role": "user", "content": str(message)[:200]}, - {"role": "assistant", "content": final}, - ] - self.history = messages - return { - "final_response": final, - "messages": messages, - "interrupted": interrupted, - "error": None, - "last_reasoning": None, - } + self.history = [*base_history, {"role": "user", "content": str(message)[:200]}, {"role": "assistant", "content": final}] + return {"final_response": final, "messages": self.history, "interrupted": interrupted, "error": None, "last_reasoning": None} def maybe_build_synthetic_agent(session_id: str, model_override: Any = None) -> SyntheticHeavyAgent | None: - """Return a :class:`SyntheticHeavyAgent` when the seam is armed, else ``None``. - - ``model_override`` (dict or str) only influences the reported ``model`` label; - it never changes the compute. - """ + """Return a :class:`SyntheticHeavyAgent` when the seam is armed, else ``None``. ``model_override`` + (dict or str) only influences the reported ``model`` label; it never changes the compute.""" if not synth_turn_armed(): return None model = "synthetic-heavy" @@ -190,8 +163,4 @@ def maybe_build_synthetic_agent(session_id: str, model_override: Any = None) -> return SyntheticHeavyAgent(session_id, model=model) -__all__ = [ - "SyntheticHeavyAgent", - "maybe_build_synthetic_agent", - "synth_turn_armed", -] +__all__ = ["SyntheticHeavyAgent", "maybe_build_synthetic_agent", "synth_turn_armed"] diff --git a/tui_gateway/turn_marker.py b/tui_gateway/turn_marker.py index dfb58e4f8c..ae938120a9 100644 --- a/tui_gateway/turn_marker.py +++ b/tui_gateway/turn_marker.py @@ -1,23 +1,18 @@ """Durable interrupted-turn markers for the desktop/TUI auto-continue path. -A running turn's progress lives only in process memory (the agent flushes to -SQLite at turn end), so an app/backend/machine death mid-turn leaves no durable -trace of the interrupted prompt. A marker is written when a turn starts and -cleared when it concludes — success, handled error, or interrupt all clear it — -so only a process death leaves one behind. ``session.resume`` reads it to -decide whether to auto-continue (``_maybe_schedule_auto_continue`` in -``tui_gateway/server.py``). - -Markers are stored per ``HERMES_HOME`` (profile sessions keep state in their own -profile dir) and the file is bounded: writes prune entries older than -``_MAX_AGE_SECS`` and cap the count, so a crash streak can't grow it unboundedly. - -Every function is best-effort — marker bookkeeping must never break a turn — so -I/O errors degrade to "no marker" instead of raising. +A running turn's progress lives only in process memory (the agent flushes to SQLite at turn end), +so an app/backend/machine death mid-turn leaves no durable trace of the interrupted prompt. A marker +is written when a turn starts and cleared when it concludes (success, handled error, or interrupt), +so only a process death leaves one behind; ``session.resume`` reads it to decide whether to +auto-continue (``_maybe_schedule_auto_continue``). Markers are stored per ``HERMES_HOME`` (profile +sessions keep state in their own profile dir); writes prune entries older than ``_MAX_AGE_SECS`` and +cap the count so a crash streak can't grow the file unboundedly. Every function is best-effort — +marker bookkeeping must never break a turn — so I/O errors degrade to "no marker" instead of raising. """ from __future__ import annotations +import contextlib import json import logging import os @@ -29,19 +24,16 @@ from typing import Any logger = logging.getLogger(__name__) -_MARKER_DIR = "desktop" -_MARKER_FILE = "interrupted_turns.json" _MAX_AGE_SECS = 24 * 3600 _MAX_ENTRIES = 32 -# Enough to re-submit any realistic prompt; guards against a multi-megabyte -# paste being journaled on every turn. +# Enough to re-submit any realistic prompt; guards against a multi-megabyte paste being journaled. _MAX_PROMPT_CHARS = 64_000 _lock = threading.Lock() def _marker_path(home: Path | str) -> Path: - return Path(home) / _MARKER_DIR / _MARKER_FILE + return Path(home) / "desktop" / "interrupted_turns.json" def _started_at(entry: dict) -> float: @@ -57,17 +49,14 @@ def _load(path: Path) -> dict[str, dict]: except Exception: logger.debug("unreadable turn-marker file %s; starting fresh", path, exc_info=True) return {} - if not isinstance(data, dict): - return {} - return {k: v for k, v in data.items() if isinstance(v, dict)} + return {k: v for k, v in data.items() if isinstance(v, dict)} if isinstance(data, dict) else {} def _prune(entries: dict[str, dict], now: float) -> dict[str, dict]: fresh = {k: e for k, e in entries.items() if now - _started_at(e) <= _MAX_AGE_SECS} if len(fresh) <= _MAX_ENTRIES: return fresh - newest = sorted(fresh.items(), key=lambda item: _started_at(item[1]), reverse=True) - return dict(newest[:_MAX_ENTRIES]) + return dict(sorted(fresh.items(), key=lambda item: _started_at(item[1]), reverse=True)[:_MAX_ENTRIES]) def _store(path: Path, entries: dict[str, dict]) -> None: @@ -81,54 +70,47 @@ def _store(path: Path, entries: dict[str, dict]) -> None: json.dump(entries, f) os.replace(tmp, path) except Exception: - try: + with contextlib.suppress(OSError): os.unlink(tmp) - except OSError: - pass raise -def record_turn_start( - home: Path | str, session_key: str, prompt: str, *, attempts: int = 0 -) -> None: - """Persist the marker for a turn that is about to run. - - ``attempts`` counts how many auto-continues led to this run: 0 for a - user-initiated turn, N for the Nth automatic re-run — the crash-loop - breaker reads it back on the next resume. - """ - if not session_key or not prompt: - return - now = time.time() - entry = { - "attempts": max(0, int(attempts)), - "prompt": prompt[:_MAX_PROMPT_CHARS], - "started_at": now, - } - try: - with _lock: - path = _marker_path(home) - entries = _prune(_load(path), now) - entries[session_key] = entry - _store(path, entries) - except Exception: - logger.debug("failed to record turn marker for %s", session_key, exc_info=True) - - -def clear_turn_marker(home: Path | str, session_key: str) -> None: - """Remove the marker once its turn concluded (any outcome the client saw).""" - if not session_key: - return +def _update(home: Path | str, session_key: str, mutate, what: str) -> None: + """Load → ``mutate(entries)`` → store under the lock; ``mutate`` returns False to skip the write.""" try: with _lock: path = _marker_path(home) entries = _load(path) - if session_key not in entries: - return - del entries[session_key] - _store(path, entries) + if mutate(entries) is not False: + _store(path, entries) except Exception: - logger.debug("failed to clear turn marker for %s", session_key, exc_info=True) + logger.debug("failed to %s turn marker for %s", what, session_key, exc_info=True) + + +def record_turn_start(home: Path | str, session_key: str, prompt: str, *, attempts: int = 0) -> None: + """Persist the marker for a turn that is about to run. + + ``attempts`` counts how many auto-continues led to this run: 0 for a user-initiated turn, N for + the Nth automatic re-run — the crash-loop breaker reads it back on the next resume. + """ + if not session_key or not prompt: + return + now = time.time() + entry = {"attempts": max(0, int(attempts)), "prompt": prompt[:_MAX_PROMPT_CHARS], "started_at": now} + + def mutate(entries: dict) -> None: + pruned = _prune(entries, now) + entries.clear() + entries.update(pruned) + entries[session_key] = entry + + _update(home, session_key, mutate, "record") + + +def clear_turn_marker(home: Path | str, session_key: str) -> None: + """Remove the marker once its turn concluded (any outcome the client saw).""" + if session_key: + _update(home, session_key, lambda entries: entries.pop(session_key, None) is not None, "clear") def read_turn_marker(home: Path | str, session_key: str) -> dict[str, Any] | None: @@ -146,8 +128,6 @@ def read_turn_marker(home: Path | str, session_key: str) -> dict[str, Any] | Non if not prompt.strip(): return None try: - started_at = _started_at(entry) - attempts = max(0, int(entry.get("attempts") or 0)) + return {"attempts": max(0, int(entry.get("attempts") or 0)), "prompt": prompt, "started_at": _started_at(entry)} except (TypeError, ValueError): return None - return {"attempts": attempts, "prompt": prompt, "started_at": started_at}