refactor(tui): render/slash_fuzzy/turn_marker/synthetic_turn/slash_worker — shared import bridge, tier table, marker update helper
This commit is contained in:
Teknium
@@ -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.<name>(*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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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}
|
||||
|
||||
Reference in New Issue
Block a user