ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

Merge branch 'simp/r3-37-C' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:41:48 -07:00
parent cc9eaea52b 4af78ae474
commit e408219d13
8 changed files with 498 additions and 671 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tui_gateway/render.py
+15 -28
View File
@@ -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)
tui_gateway/session_auto_continue.py
+121 -155
View File
@@ -12,11 +12,11 @@ from .method_ctx import HandlerRegistry, bind_module
_registry = HandlerRegistry()
# A concluded turn (success, handled error, interrupt) clears its durable marker
# (turn_marker.py) in _run_prompt_submit's finally; only a process death leaves it
# behind, so a marker at session.resume proves the turn never finished AND the
# client never saw a terminal frame. Fresh: re-submit automatically (as the
# messaging gateway does). Stale: clear it and let the partial transcript speak.
# A concluded turn (success, handled error, interrupt) clears its durable marker (turn_marker.py)
# in _run_prompt_submit's finally; only a process death leaves it behind, so a marker at
# session.resume proves the turn never finished AND the client never saw a terminal frame. Fresh:
# re-submit automatically (as the messaging gateway does). Stale: clear it and let the partial
# transcript speak.
_AUTO_CONTINUE_ENABLED_DEFAULT = True
_AUTO_CONTINUE_FRESHNESS_MINUTES_DEFAULT = 15
@@ -36,9 +36,7 @@ def _auto_continue_config() -> tuple[bool, float, int]:
return (
is_truthy_value(cfg.get("enabled"), default=_AUTO_CONTINUE_ENABLED_DEFAULT),
max(0.0, minutes) * 60.0,
_coerce_int_config_value(
cfg.get("max_attempts"), _AUTO_CONTINUE_MAX_ATTEMPTS_DEFAULT, min_value=0
),
_coerce_int_config_value(cfg.get("max_attempts"), _AUTO_CONTINUE_MAX_ATTEMPTS_DEFAULT, min_value=0),
)
@@ -51,10 +49,10 @@ def _session_home(session: dict) -> Path:
def _retire_turn_marker(session: dict, *keys: str) -> None:
"""Drop the crash marker for a turn whose outcome is about to reach the client.
Called right before the terminal frame, not at turn-thread end: post-turn work
(titles, memory sync, goal hooks) outlives the client's answer, and quitting in
that window would leave a marker that re-runs a finished turn on next launch.
Extra ``keys`` cover a session_key that compression rotated mid-turn.
Called right before the terminal frame, not at turn-thread end: post-turn work (titles, memory
sync, goal hooks) outlives the client's answer, and quitting in that window would leave a marker
that re-runs a finished turn on next launch. Extra ``keys`` cover a session_key that compression
rotated mid-turn.
"""
home = _session_home(session)
for key in dict.fromkeys((*keys, str(session.get("session_key") or ""))):
@@ -63,8 +61,8 @@ def _retire_turn_marker(session: dict, *keys: str) -> None:
def _auto_continue_note(prompt: str) -> str:
# Same opening as the gateway's recovery notes (transcript tooling recognizes
# both). The prompt is embedded: a hard crash persists nothing else of the turn.
# Same opening as the gateway's recovery notes (transcript tooling recognizes both). The prompt
# is embedded: a hard crash persists nothing else of the turn.
return (
f"{_AUTO_CONTINUE_NOTE_PREFIX} — the app or its backend process "
"stopped before the turn could finish. Some of the work may already "
@@ -81,19 +79,22 @@ def _ac_release_turn(session: dict, *, unschedule: bool = False) -> None:
session["_auto_continue_scheduled"] = False
def _ac_dispatch_failed(what: str, exc: BaseException) -> None:
print(f"[tui_gateway] {what} dispatch failed: {type(exc).__name__}: {exc}", file=sys.stderr)
def _maybe_schedule_auto_continue(sid: str, session: dict, session_key: str) -> dict | None:
"""Kick off a continuation turn for a crash-interrupted session.
Called from session.resume's cold paths once the live record is registered.
Returns a descriptor for the resume payload when scheduled, else None. The turn
runs on a background thread after the deferred agent build via the normal
_run_prompt_submit path, so the client that just resumed streams it live.
Called from session.resume's cold paths once the live record is registered. Returns a descriptor
for the resume payload when scheduled, else None. The turn runs on a background thread after the
deferred agent build via the normal _run_prompt_submit path, so the client that just resumed
streams it live.
"""
# Hosted room turns are recovered by their durable task/lease state machine;
# generic auto-continue would bypass its execution generation and duplicate work.
# Hosted room turns are recovered by their durable task/lease state machine; generic
# auto-continue would bypass its execution generation and duplicate work.
if session.get("source") == "bot_room":
return None
home = _session_home(session)
marker = read_turn_marker(home, session_key)
if marker is None:
@@ -129,17 +130,17 @@ def _maybe_schedule_auto_continue(sid: str, session: dict, session_key: str) ->
return
session["running"] = True
session["last_active"] = time.time()
# Ownership admission BEFORE message.start: a sibling backend sharing this
# HERMES_HOME may have written the marker and still be mid-turn. Leave the
# marker so a later resume retries once the owner finishes or dies.
# Ownership admission BEFORE message.start: a sibling backend sharing this HERMES_HOME may
# have written the marker and still be mid-turn. Leave the marker so a later resume retries
# once the owner finishes or dies.
if _ensure_active_session_slot(sid, session) is not None:
logger.info("auto-continue for %s refused: session has another live owner", session_key)
_ac_release_turn(session, unschedule=True)
return
with session["history_lock"]:
# Marker inputs read back by _run_prompt_submit: count the attempt (crash
# breaker) and re-record the ORIGINAL prompt (no nested notes). Set here,
# not at schedule time, so a bail above leaves nothing for a racing user turn.
# Marker inputs read back by _run_prompt_submit: count the attempt (crash breaker) and
# re-record the ORIGINAL prompt (no nested notes). Set here, not at schedule time, so a
# bail above leaves nothing for a racing user turn.
session["_auto_continue_attempt"] = attempt
session["_auto_continue_prompt"] = marker["prompt"]
try:
@@ -147,13 +148,12 @@ def _maybe_schedule_auto_continue(sid: str, session: dict, session_key: str) ->
_emit("message.start", sid)
_run_prompt_submit(rid, sid, session, text, display_kind="auto_continue")
except Exception as exc:
print(f"[tui_gateway] auto-continue dispatch failed: {type(exc).__name__}: {exc}", file=sys.stderr)
_ac_dispatch_failed("auto-continue", exc)
_ac_release_turn(session)
threading.Thread(target=kickoff, daemon=True).start()
logger.info(
"auto-continue scheduled for session %s (attempt %d, interrupted %.0fs ago)",
session_key, attempt, age,
"auto-continue scheduled for session %s (attempt %d, interrupted %.0fs ago)", session_key, attempt, age
)
return {"attempt": attempt, "interrupted_at": marker["started_at"]}
@@ -161,10 +161,10 @@ def _maybe_schedule_auto_continue(sid: str, session: dict, session_key: str) ->
def _enqueue_prompt(session: dict, text: Any, transport: Any, image_paths: list[str] | None = None) -> None:
"""Stash a message to run as the very next turn once the live one ends.
Text-only arrivals share a slot and merge losslessly (like the consecutive-user
merge in ``repair_message_sequence``); image-bearing ones stay separate envelopes
so attachment ownership/chronology survive. ``transport`` is pinned so the drained
turn streams back to its sender even if the session transport is rebound.
Text-only arrivals share a slot and merge losslessly (like the consecutive-user merge in
``repair_message_sequence``); image-bearing ones stay separate envelopes so attachment
ownership/chronology survive. ``transport`` is pinned so the drained turn streams back to its
sender even if the session transport is rebound.
"""
image_paths = list(image_paths or [])
# Scrub live-turn self-duplicates first so the text merge below can't glue
@@ -173,7 +173,7 @@ def _enqueue_prompt(session: dict, text: Any, transport: Any, image_paths: list[
# Never queue a text-only self-copy of the live prompt: draining it would restart it.
if not image_paths and isinstance(text, str):
turn = session.get("inflight_turn")
original = (str(turn.get("user") or "").strip() if isinstance(turn, dict) else "")
original = str(turn.get("user") or "").strip() if isinstance(turn, dict) else ""
if original and text.strip() == original:
return
queued = {"text": text, "transport": transport}
@@ -181,35 +181,28 @@ def _enqueue_prompt(session: dict, text: Any, transport: Any, image_paths: list[
queued["image_paths"] = image_paths
existing = session.get("queued_prompt")
if (
existing
and isinstance(existing.get("text"), str)
and isinstance(text, str)
and not existing.get("image_paths")
and not image_paths
and not session.get("queued_prompts")
existing and isinstance(existing.get("text"), str) and isinstance(text, str)
and not existing.get("image_paths") and not image_paths and not session.get("queued_prompts")
):
prev = existing["text"]
existing["text"] = f"{prev}\n\n{text}" if prev and text else (prev or text)
return
if existing:
elif existing:
session.setdefault("queued_prompts", []).append(queued)
return
session["queued_prompt"] = queued
else:
session["queued_prompt"] = queued
def _sanitize_queued_entry_vs_inflight_user(entry: Any, original: str) -> dict | None:
"""Drop (``None``) or rewrite a queue envelope that re-carries the live user text.
Text-only self-duplicates are dropped; a merged slot ``"{original}\\n\\n{later}"``
is rewritten to ``later`` so the correction survives without re-firing the
original. Image-bearing envelopes are left alone (chronology is load-bearing).
Text-only self-duplicates are dropped; a merged slot ``"{original}\\n\\n{later}"`` is rewritten
to ``later`` so the correction survives without re-firing the original. Image-bearing envelopes
are left alone (chronology is load-bearing).
"""
if not original or not isinstance(entry, dict):
return entry if isinstance(entry, dict) else None
if entry.get("image_paths"):
return entry
if not isinstance(entry, dict):
return None
text = entry.get("text")
if not isinstance(text, str):
if not original or entry.get("image_paths") or not isinstance(text, str):
return entry
stripped = text.strip()
if not stripped or stripped == original:
@@ -219,35 +212,25 @@ def _sanitize_queued_entry_vs_inflight_user(entry: Any, original: str) -> dict |
prefix = original + sep
if text.startswith(prefix):
rest = text[len(prefix) :].strip()
if not rest or rest == original:
return None
return {**entry, "text": rest}
return None if not rest or rest == original else {**entry, "text": rest}
return entry
def _drop_queued_duplicates_of_inflight_user(session: dict) -> None:
"""Remove server-queue copies of the live turn's original user text.
A mid-turn ``prompt.submit`` of the same text can be queued while redirect is
unavailable (build window, tool boundary); after a later redirect it must not
drain and restart the original as a fresh turn. Unrelated follow-ups stay.
A mid-turn ``prompt.submit`` of the same text can be queued while redirect is unavailable
(build window, tool boundary); after a later redirect it must not drain and restart the original
as a fresh turn. Unrelated follow-ups stay.
"""
turn = session.get("inflight_turn")
if not isinstance(turn, dict):
return
original = str(turn.get("user") or "").strip()
original = str(turn.get("user") or "").strip() if isinstance(turn, dict) else ""
if not original:
return
head = session.get("queued_prompt")
rest = list(session.get("queued_prompts") or [])
kept: list[dict] = []
for entry in ([head] if head else []) + rest:
cleaned = _sanitize_queued_entry_vs_inflight_user(entry, original)
if cleaned is not None:
kept.append(cleaned)
_ac_set_queue(session, kept)
entries = ([head] if head else []) + list(session.get("queued_prompts") or [])
cleaned = (_sanitize_queued_entry_vs_inflight_user(entry, original) for entry in entries)
_ac_set_queue(session, [c for c in cleaned if c is not None])
def _ac_set_queue(session: dict, entries: list) -> None:
@@ -262,15 +245,13 @@ def _ac_set_queue(session: dict, entries: list) -> None:
def _interrupt_busy_session(sid: str, session: dict, agent: Any) -> None:
"""Interrupt a busy turn on a worker thread, never under ``history_lock``.
Some providers can't apply ``interrupt()`` until a blocking tool/network call
returns; doing it inline stalled ``session.resume`` and the queued prompt.
At most one interrupt worker per session so repeated steering can't leak threads.
Some providers can't apply ``interrupt()`` until a blocking tool/network call returns; doing it
inline stalled ``session.resume`` and the queued prompt. At most one interrupt worker per session
so repeated steering can't leak threads.
"""
use_agent = agent is not None and hasattr(agent, "interrupt")
use_compute_host = not use_agent and _session_uses_compute_host(session)
if not use_agent and not use_compute_host:
if not use_agent and not _session_uses_compute_host(session):
return
with session["history_lock"]:
if session.get("_busy_interrupt_pending"):
return
@@ -291,27 +272,31 @@ def _interrupt_busy_session(sid: str, session: dict, agent: Any) -> None:
threading.Thread(target=interrupt, daemon=True, name=f"busy-interrupt-{sid}").start()
def _ac_record_inflight_correction(session: dict, plain_text: str) -> None:
"""Record an accepted steer/redirect; scrub stale self-duplicates so the
live turn's original text is not re-fired from the queue after settle."""
def _ac_try_correction(rid, session: dict, agent: Any, method: str, plain_text: str, status: str) -> dict | None:
"""Apply ``agent.<method>(plain_text)`` (steer/redirect); on acceptance record the correction, scrub
stale self-duplicates so the live turn's original text is not re-fired after settle, and return
the ``status`` reply. None → caller falls through to the queue path."""
try:
if not getattr(agent, method)(plain_text):
return None
except Exception:
return None
with session["history_lock"]:
_record_inflight_correction(session, plain_text)
_drop_queued_duplicates_of_inflight_user(session)
session["last_active"] = time.time()
return _ok(rid, {"status": status})
def _handle_busy_submit(
rid, sid: str, session: dict, text: Any, transport: Any, queued: bool = False
) -> dict | None:
"""Apply ``display.busy_input_mode`` to a prompt that lands mid-turn instead of
rejecting it with ``session busy`` (rejection made clients busy-retry and
silently drop sends when teardown outlived their deadline).
def _handle_busy_submit(rid, sid: str, session: dict, text: Any, transport: Any, queued: bool = False) -> dict | None:
"""Apply ``display.busy_input_mode`` to a prompt that lands mid-turn instead of rejecting it with
``session busy`` (rejection made clients busy-retry and silently drop sends when teardown
outlived their deadline).
Modes: ``interrupt`` (default) → redirect the live turn, falling back to hard
interrupt + queue for older agents; ``queue`` → queue only; ``steer`` → inject
after the current atomic action. ``queued=True`` (client queue drain) forces
queue mode: a "run after" message must NEVER become a live-turn correction,
even when the drain loses the settle race against a still-unwinding turn.
Modes: ``interrupt`` (default) → redirect the live turn, falling back to hard interrupt + queue
for older agents; ``queue`` → queue only; ``steer`` → inject after the current atomic action.
``queued=True`` (client queue drain) forces queue mode: a "run after" message must NEVER become a
live-turn correction, even when the drain loses the settle race against a still-unwinding turn.
"""
mode = "queue" if queued else _load_busy_input_mode()
agent = session.get("agent")
@@ -325,31 +310,23 @@ def _handle_busy_submit(
session["attached_images"] = []
text_only = not image_paths and _is_text_only_busy_payload(text)
plain_text = _coerce_message_text(text).strip() if text_only else ""
if mode == "steer" and text_only and plain_text and agent is not None and hasattr(agent, "steer"):
try:
if agent.steer(plain_text):
_ac_record_inflight_correction(session, plain_text)
return _ok(rid, {"status": "steered"})
except Exception:
pass # fall through to queue
# Text-only corrections redirect in place when supported; media payloads and
# older agents fall through to the proven interrupt + queue path.
if (
mode == "interrupt"
and text_only
and plain_text
and agent is not None
and getattr(agent, "_supports_active_turn_redirect", False) is True
and hasattr(agent, "redirect")
):
try:
if agent.redirect(plain_text):
_ac_record_inflight_correction(session, plain_text)
return _ok(rid, {"status": "redirected"})
except Exception:
pass # preserve the proven interrupt + queue fallback below
# Queue before asking the live turn to stop. Never call a provider/compute-host
# method under history_lock: an interrupt can wait behind the op it cancels.
if plain_text and agent is not None:
if mode == "steer" and hasattr(agent, "steer"):
resp = _ac_try_correction(rid, session, agent, "steer", plain_text, "steered")
if resp is not None:
return resp
# Text-only corrections redirect in place when supported; media payloads and older agents
# fall through to the proven interrupt + queue path.
if (
mode == "interrupt"
and getattr(agent, "_supports_active_turn_redirect", False) is True
and hasattr(agent, "redirect")
):
resp = _ac_try_correction(rid, session, agent, "redirect", plain_text, "redirected")
if resp is not None:
return resp
# Queue before asking the live turn to stop. Never call a provider/compute-host method under
# history_lock: an interrupt can wait behind the op it cancels.
with session["history_lock"]:
if not session.get("running"):
if image_paths:
@@ -357,11 +334,10 @@ def _handle_busy_submit(
return None
_enqueue_prompt(session, text, transport, image_paths=image_paths)
session["last_active"] = time.time()
# Attachments need their own model invocation: queue without cancelling so the
# user gets both results in order. ``steer`` must NEVER escalate to a hard
# interrupt: it would kill the live turn AND drop ``AIAgent._pending_steer``,
# destroying earlier accepted steers; steer fall-throughs stay FIFO-queued.
# Attachments need their own model invocation: queue without cancelling so the user gets both
# results in order. ``steer`` must NEVER escalate to a hard interrupt: it would kill the live
# turn AND drop ``AIAgent._pending_steer``, destroying earlier accepted steers; steer
# fall-throughs stay FIFO-queued.
if mode == "interrupt" and not image_paths:
_interrupt_busy_session(sid, session, agent)
return _ok(rid, {"status": "queued"})
@@ -370,8 +346,8 @@ def _handle_busy_submit(
def _drain_queued_prompt(rid, sid: str, session: dict) -> bool:
"""Fire a queued next-turn prompt if one is waiting and the session is idle.
True when dispatched: the caller skips lower-priority follow-ups this cycle
(the user's message wins). Claim-under-lock like the goal-continuation re-fire.
True when dispatched: the caller skips lower-priority follow-ups this cycle (the user's message
wins). Claim-under-lock like the goal-continuation re-fire.
"""
with session["history_lock"]:
if session.get("_closing"):
@@ -390,15 +366,11 @@ def _drain_queued_prompt(rid, sid: str, session: dict) -> bool:
use_compute_host = _session_uses_compute_host(session)
with session["history_lock"]:
if int(session.get("_queued_prompt_generation", 0)) != queue_generation:
# Generation bump cancelled the claim (Stop, compress re-anchor, …): don't
# dispatch, but restore the envelope (claimed head first, then whatever
# advanced into the slot) so a legitimate follow-up isn't dropped.
rest: list = []
# Generation bump cancelled the claim (Stop, compress re-anchor, …): don't dispatch, but
# restore the envelope (claimed head first, then whatever advanced into the slot) so a
# legitimate follow-up isn't dropped.
advanced = session.get("queued_prompt")
if advanced:
rest.append(advanced)
rest.extend(session.get("queued_prompts") or [])
_ac_set_queue(session, [queued, *rest])
_ac_set_queue(session, [queued, *([advanced] if advanced else []), *(session.get("queued_prompts") or [])])
session["running"] = False
return True
kwargs: dict = {"queued_prompt_generation": queue_generation}
@@ -418,7 +390,7 @@ def _drain_queued_prompt(rid, sid: str, session: dict) -> bool:
else:
_run_prompt_submit(rid, sid, session, queued["text"], **kwargs)
except Exception as exc:
print(f"[tui_gateway] queued prompt dispatch failed: {type(exc).__name__}: {exc}", file=sys.stderr)
_ac_dispatch_failed("queued prompt", exc)
_ac_release_turn(session)
dispatch_failed = True
if dispatch_failed:
@@ -440,23 +412,20 @@ def _inflight_snapshot(session: dict) -> dict | None:
if not user and not assistant and not streaming and not error:
return None
snapshot = {"assistant": assistant, "streaming": streaming, "user": user}
raw_corrections = turn.get("corrections") or []
raw_offsets = turn.get("correction_offsets") or []
correction_pairs = [
(str(c), raw_offsets[i] if i < len(raw_offsets) else None)
for i, c in enumerate(raw_corrections)
if str(c).strip()
for i, c in enumerate(turn.get("corrections") or []) if str(c).strip()
]
if correction_pairs:
# Mid-turn redirects alongside (not over) the original prompt so resume can
# rebuild every user bubble; offsets only when every correction has one so
# clients can trust the pairing.
# Mid-turn redirects alongside (not over) the original prompt so resume can rebuild every
# user bubble; offsets only when every correction has one so clients can trust the pairing.
snapshot["corrections"] = [c for c, _ in correction_pairs]
if all(isinstance(offset, int) and offset >= 0 for _, offset in correction_pairs):
snapshot["correction_offsets"] = [int(offset) for _, offset in correction_pairs] # type: ignore[arg-type]
if error:
# Retained failed turn (_fail_inflight_turn): a resuming client must rebuild
# the failed bubble, not render the partial text as a healthy reply.
# Retained failed turn (_fail_inflight_turn): a resuming client must rebuild the failed
# bubble, not render the partial text as a healthy reply.
snapshot["error"] = error
snapshot["status"] = str(turn.get("status") or "error")
snapshot["recoverable"] = bool(turn.get("recoverable"))
@@ -469,23 +438,21 @@ def _inflight_snapshot(session: dict) -> dict | None:
def _emit_terminal_turn_error(
sid: str, session: dict, error: Any, error_surface: Optional[dict] = None, *, retire_marker: bool = True
) -> None:
"""Close a failed turn with the same ``status: "error"`` ``message.complete``
frame as ``_run_prompt_submit``'s returned-error path, retaining the turn via
``_fail_inflight_turn`` so a client that missed the frame recovers it from
``session.resume``'s ``inflight``. Callers that know the failing layer pass
``error_surface``; exception callers leave it None and it is classified here.
"""Close a failed turn with the same ``status: "error"`` ``message.complete`` frame as
``_run_prompt_submit``'s returned-error path, retaining the turn via ``_fail_inflight_turn`` so a
client that missed the frame recovers it from ``session.resume``'s ``inflight``. Callers that
know the failing layer pass ``error_surface``; exception callers leave it None and it is
classified here.
"""
agent = session.get("agent")
# {layer, code, retryable} descriptor so the desktop can say "Provider error" /
# "Gateway error" with matching recovery actions. Advisory: never raises.
# {layer, code, retryable} descriptor so the desktop can say "Provider error" / "Gateway error"
# with matching recovery actions. Advisory: never raises.
if error_surface is None and isinstance(error, BaseException):
try:
from agent.error_surface import build_error_surface_from_exception
error_surface = build_error_surface_from_exception(
error,
provider=str(getattr(agent, "provider", "") or ""),
model=str(getattr(agent, "model", "") or ""),
error, provider=str(getattr(agent, "provider", "") or ""), model=str(getattr(agent, "model", "") or "")
)
except Exception:
error_surface = None
@@ -519,9 +486,8 @@ def _emit_terminal_turn_error(
def _restore_agent_history_after_turn_error(session: dict, agent) -> bool:
"""Keep a failed turn's working transcript: ``AIAgent`` persists its messages
independently, so after a raise the next prompt must see them, not the
pre-turn snapshot."""
"""Keep a failed turn's working transcript: ``AIAgent`` persists its messages independently, so
after a raise the next prompt must see them, not the pre-turn snapshot."""
agent_messages = getattr(agent, "_session_messages", None)
if not isinstance(agent_messages, list):
return False
@@ -532,8 +498,8 @@ def _restore_agent_history_after_turn_error(session: dict, agent) -> bool:
def _queued_prompt_snapshot(session: dict) -> dict | None:
"""The accepted next-turn prompt without its transport handle, for the
live-session projection (Desktop may reconnect while it is still queued)."""
"""The accepted next-turn prompt without its transport handle, for the live-session projection
(Desktop may reconnect while it is still queued)."""
queued = session.get("queued_prompt")
if not isinstance(queued, dict):
return None
tui_gateway/session_compression.py
+25 -53
View File
@@ -57,9 +57,7 @@ def _derived_default_threshold_percent(agent: Any, compression: dict) -> float:
model = getattr(agent, "model", "") or ""
provider = getattr(agent, "provider", "") or ""
autoraise_enabled = str(compression.get("codex_gpt55_autoraise", True)).lower() in {"true", "1", "yes"}
model_cthresh = _compression_threshold_for_model(
model, provider, allow_codex_gpt55_autoraise=autoraise_enabled
)
model_cthresh = _compression_threshold_for_model(model, provider, allow_codex_gpt55_autoraise=autoraise_enabled)
pct, _notice = _resolve_compression_threshold(
pct, model_cthresh, model=model,
is_codex_autoraise=_is_codex_gpt54_or_gpt55(model, provider) or _is_codex_spark(model, provider),
@@ -95,10 +93,7 @@ def _apply_live_compression_config(agent: Any, cfg: dict | None) -> None:
model_cfg = cfg.get("model") if isinstance(cfg.get("model"), dict) else {}
enabled_raw = compression.get("enabled", True)
if isinstance(enabled_raw, bool):
agent.compression_enabled = enabled_raw
else:
agent.compression_enabled = str(enabled_raw).lower() in {"true", "1", "yes"}
agent.compression_enabled = enabled_raw if isinstance(enabled_raw, bool) else str(enabled_raw).lower() in {"true", "1", "yes"}
agent.codex_responses_native_compaction = is_truthy_value(compression.get("codex_responses_native", False))
native_threshold_raw = compression.get("codex_responses_compact_threshold", 200_000)
@@ -114,11 +109,8 @@ def _apply_live_compression_config(agent: Any, cfg: dict | None) -> None:
agent.codex_responses_compact_threshold = native_threshold
# Absence restores the agent_init/config default (0 = disabled).
idle_raw = compression.get("idle_compact_after_seconds", 0)
try:
agent.compression_idle_compact_after_seconds = max(0, int(idle_raw or 0))
except (TypeError, ValueError):
pass
with contextlib.suppress(TypeError, ValueError):
agent.compression_idle_compact_after_seconds = max(0, int(compression.get("idle_compact_after_seconds", 0) or 0))
cc = getattr(agent, "context_compressor", None)
if cc is None:
@@ -138,32 +130,25 @@ def _apply_live_compression_config(agent: Any, cfg: dict | None) -> None:
continue
setattr(cc, key, max(min_value, value))
try:
with contextlib.suppress(TypeError, ValueError):
ratio_raw = compression.get("target_ratio", _compressor_ctor_default("summary_target_ratio", 0.20))
cc.summary_target_ratio = max(0.10, min(float(ratio_raw), 0.80))
except (TypeError, ValueError):
pass
# Absent or invalid shape (agent_init treats both as empty): stale overrides must stop steering.
raw_thresholds = compression.get("model_thresholds")
if isinstance(raw_thresholds, dict):
cc.model_thresholds = {
str(k): float(v)
for k, v in raw_thresholds.items() if isinstance(v, (int, float)) and not isinstance(v, bool)
}
else:
# Absent or invalid shape (agent_init treats both as empty): stale overrides must stop steering.
cc.model_thresholds = {}
cc.model_thresholds = {
str(k): float(v)
for k, v in raw_thresholds.items() if isinstance(v, (int, float)) and not isinstance(v, bool)
} if isinstance(raw_thresholds, dict) else {}
# threshold: present value wins; absence derives via the agent_init resolution (default + autoraise).
pct: float | None = None
if "threshold" in compression:
try:
with contextlib.suppress(TypeError, ValueError):
pct = float(compression["threshold"])
except (TypeError, ValueError):
pct = None
if pct is None:
pct = _derived_default_threshold_percent(agent, compression)
try:
with contextlib.suppress(TypeError, ValueError):
cc._config_threshold_percent = pct
cc._configured_threshold_percent = pct
base = pct
@@ -180,8 +165,6 @@ def _apply_live_compression_config(agent: Any, cfg: dict | None) -> None:
cc.threshold_percent = pct
else:
cc.threshold_percent = pct
except (TypeError, ValueError):
pass
raw_ctx = model_cfg.get("context_length")
if raw_ctx is not None:
@@ -191,10 +174,8 @@ def _apply_live_compression_config(agent: Any, cfg: dict | None) -> None:
new_ctx = 0
if new_ctx > 0:
cc._config_context_length = new_ctx
try:
with contextlib.suppress(Exception):
cc.context_length = new_ctx
except Exception:
pass
elif getattr(cc, "_config_context_length", None) is not None:
# model.context_length removed: drop the override and force re-inference from model metadata
# on next access (construction's deferred resolution); re-applies the small-context floor too.
@@ -222,9 +203,7 @@ def _apply_live_compression_config(agent: Any, cfg: dict | None) -> None:
compute = getattr(cc, "_compute_threshold_tokens", None)
if callable(compute):
cc.threshold_tokens = compute(
getattr(cc, "context_length", 0) or 0,
getattr(cc, "threshold_percent", 0.5),
getattr(cc, "max_tokens", None),
getattr(cc, "context_length", 0) or 0, getattr(cc, "threshold_percent", 0.5), getattr(cc, "max_tokens", None)
)
cc._apply_threshold_tokens_cap()
else:
@@ -313,18 +292,14 @@ def _compress_session_history(
partial, keep_last, focus_topic = parse_partial_compress_args(focus_topic or "")
# Only the head is summarized; the last `keep_last` exchanges ride along verbatim. A degenerate
# split (empty tail) falls back to full compression so the user still gets an action.
tail: list = []
head = history
if partial:
head, tail = split_history_for_partial_compress(history, keep_last)
if not tail:
partial = False
head = history
head, tail = (split_history_for_partial_compress(history, keep_last) if partial else (history, []))
if not tail:
partial, head = False, history
if approx_tokens is None:
# Include system prompt + tool schemas so the figure reflects real request pressure.
_sys_prompt = getattr(agent, "_cached_system_prompt", "") or ""
_tools = getattr(agent, "tools", None) or None
approx_tokens = estimate_request_tokens_rough(history, system_prompt=_sys_prompt, tools=_tools)
approx_tokens = estimate_request_tokens_rough(
history, system_prompt=getattr(agent, "_cached_system_prompt", "") or "", tools=getattr(agent, "tools", None) or None
)
# system_message=None: _compress_context rebuilds the system prompt via _build_system_prompt(None);
# passing the cached prompt (already holding the identity block) appends the identity twice.
# force=True: every caller is a manual /compress path, which bypasses the summary-failure
@@ -346,7 +321,7 @@ def _compress_session_history(
finalize_context_engine_compression_notification(agent, committed=False)
raise CompressionLockHeld(_lock_skipped if isinstance(_lock_skipped, str) else None)
if partial and tail:
if partial:
compressed = rejoin_compressed_head_and_tail(compressed, tail)
with session["history_lock"]:
if int(session.get("history_version", 0)) != history_version:
@@ -375,14 +350,15 @@ def _sync_session_key_after_compress(
if not new_session_id or new_session_id == old_key:
return
lease_reanchored = _transfer_active_session_slot(sid, session, new_session_id=new_session_id)
if not lease_reanchored:
if not _transfer_active_session_slot(sid, session, new_session_id=new_session_id):
logger.warning(
"Compression session lease did not re-anchor: sid=%s old_session_id=%s new_session_id=%s",
sid, old_key, new_session_id,
)
try:
# Even if the approval module fails to import, anchor session_key on the continuation id.
session["session_key"] = new_session_id
with contextlib.suppress(Exception):
from tools.approval import (
disable_session_yolo, enable_session_yolo, is_session_yolo_enabled, register_gateway_notify,
unregister_gateway_notify,
@@ -390,7 +366,6 @@ def _sync_session_key_after_compress(
with contextlib.suppress(Exception):
unregister_gateway_notify(old_key)
session["session_key"] = new_session_id
try:
yolo_was_on = is_session_yolo_enabled(old_key)
except Exception:
@@ -401,9 +376,6 @@ def _sync_session_key_after_compress(
disable_session_yolo(old_key)
with contextlib.suppress(Exception):
register_gateway_notify(new_session_id, lambda data: _emit_approval_request(sid, data))
except Exception:
# Even if the approval module fails to import, anchor session_key on the continuation id.
session["session_key"] = new_session_id
# Invalidate any in-flight ``_drain_queued_prompt`` claim taken under the pre-rotation key: a raced
# drain must not dispatch on the continuation (its envelope is restored to the queue).
tui_gateway/session_notifications.py
+208 -223
View File
@@ -1,8 +1,6 @@
"""Per-session notification poller: kanban/loop/delegation events routed to the owning session, desktop UI wiring, HUD surface note.
Bodies are rebound onto server.py's globals at install time (see
method_ctx.bind_module), so they reference server.py globals bare.
"""
"""Per-session notification poller: kanban/loop/delegation events routed to the owning session,
desktop UI wiring, HUD surface note. Bodies are rebound onto server.py's globals at install time
(method_ctx.bind_module), so they reference server.py globals bare."""
from __future__ import annotations
@@ -15,8 +13,7 @@ _registry = HandlerRegistry()
def _notif_locked_sessions(fn, default):
"""Run ``fn(_sessions)`` under ``_sessions_lock``; ``default`` if that fails (the poller
thread must never crash on a lock/enumeration failure)."""
"""Run ``fn(_sessions)`` under ``_sessions_lock``; ``default`` on failure (poller must never crash)."""
try:
with _sessions_lock:
return fn(_sessions)
@@ -32,6 +29,13 @@ def _notif_session_matches(s: dict, keys) -> bool:
return str(s.get("session_key") or "") in keys or _session_lookup_key(s, fallback="") in keys
def _notif_live_session_matches(keys) -> bool:
"""Any non-finalized live session matches ``keys`` (False if the registry can't be read)."""
return _notif_locked_sessions(
lambda ss: any(not s.get("_finalized") and _notif_session_matches(s, keys) for s in ss.values()), False
)
def _notif_resolve_event_key(evt_key: str) -> str:
"""Resolve a compression-rotated session key to its continuation tip (or itself)."""
try:
@@ -56,13 +60,10 @@ def _notification_event_belongs_elsewhere(sid: str, session: dict, evt: dict) ->
return True
# Exact UI tab gone: fall through to durable session_key routing so a
# resumed continuation with the same key/lineage can still claim it.
evt_key = str(evt.get("session_key") or "")
if not evt_key:
return False
current_keys = _notif_current_keys(sid, session)
# Compression can rotate AIAgent.session_id while the detached child is still running: map
# the event's original key to its continuation tip so it reaches the live session instead of
# becoming an orphan any poller may consume. A live continuation wins over the compressed
@@ -71,19 +72,13 @@ def _notification_event_belongs_elsewhere(sid: str, session: dict, evt: dict) ->
if resolved_key != evt_key:
if resolved_key in current_keys:
return False
if _notif_locked_sessions(
lambda ss: any(not s.get("_finalized") and _notif_session_matches(s, {resolved_key}) for s in ss.values()),
False,
):
if _notif_live_session_matches({resolved_key}):
return True
if evt_key in current_keys:
return False
snapshot = _notif_locked_sessions(lambda ss: list(ss.values()), None)
if snapshot is None:
return False # can't enumerate: fail open rather than drop the event
keys = {evt_key, resolved_key}
return any(s is not session and not s.get("_finalized") and _notif_session_matches(s, keys) for s in snapshot)
@@ -101,9 +96,7 @@ def _session_owns_notification_event(sid: str, session: dict, evt: dict) -> bool
if not evt_key:
return False
current_keys = _notif_current_keys(sid, session)
if evt_key in current_keys:
return True
return _notif_resolve_event_key(evt_key) in current_keys
return evt_key in current_keys or _notif_resolve_event_key(evt_key) in current_keys
def _notification_event_requires_owner(evt: dict) -> bool:
@@ -131,11 +124,10 @@ def _notification_event_dedup_key(evt: dict) -> tuple:
return (evt_sid, evt_type)
# Mirror gateway/kanban_watchers.py TERMINAL_KINDS: claim silent kinds too so
# the cursor advances past them and they can't wedge a later completed/blocked
# event behind an unclaimed row.
# Mirror gateway/kanban_watchers.py TERMINAL_KINDS: claim silent kinds (archived/unblocked) too so
# the cursor advances past them and they can't wedge a later completed/blocked event behind an
# unclaimed row.
_KANBAN_NOTIFY_KINDS = ("completed", "blocked", "gave_up", "crashed", "timed_out", "status", "archived", "unblocked")
_KANBAN_SILENT_KINDS = frozenset({"archived", "unblocked"})
_KANBAN_POLL_SECONDS = 5.0
_LOOP_POLL_SECONDS = 5.0
@@ -145,10 +137,30 @@ def _notif_release_turn(session: dict) -> None:
session["running"] = False
def _notif_claim_turn(session: dict) -> bool:
"""Claim the idle session (running=True) under history_lock; False if a turn is live."""
with session["history_lock"]:
if session.get("running"):
return False
session["running"] = True
return True
def _notif_log_failure(what: str, exc: BaseException) -> None:
print(f"[tui_gateway] {what}: {type(exc).__name__}: {exc}", file=sys.stderr)
def _notif_submit(rid: str, sid: str, session: dict, text: str, what: str, **kwargs) -> None:
"""message.start + _run_prompt_submit for a claimed (running=True) turn; releases on failure."""
try:
_emit("message.start", sid)
_run_prompt_submit(rid, sid, session, text, **kwargs)
except Exception as exc:
_notif_log_failure(what, exc)
_notif_release_turn(session)
raise
def _maybe_fire_tui_loop_tick(sid: str, session: dict) -> None:
"""Fire a due /loop wakeup for an idle TUI/Desktop/dashboard session (per-session poller,
coarse cadence). Claims the session under history_lock (running=True) before dispatching so
@@ -157,61 +169,50 @@ def _maybe_fire_tui_loop_tick(sid: str, session: dict) -> None:
from hermes_cli.loops import LoopManager, goal_blocks_loop_tick
except Exception:
return
sid_key = session.get("session_key") or ""
if not sid_key:
return
mgr = LoopManager(session_id=sid_key)
if not mgr.is_due() or goal_blocks_loop_tick(sid_key):
return
with session["history_lock"]:
if session.get("running"):
return # busy — stays due, next poll retries
session["running"] = True
if not mgr.is_due() or goal_blocks_loop_tick(sid_key) or not _notif_claim_turn(session):
return # busy — stays due, next poll retries
wakeup = mgr.fire_tick()
if not wakeup:
_notif_release_turn(session)
return
tick_no = mgr.state.ticks_fired if mgr.state else "?"
rid = f"__loop__{int(time.time() * 1000)}"
try:
_emit("status.update", sid, {"kind": "loop", "text": f"↻ /loop wakeup #{tick_no} firing…"})
if wakeup.lstrip().startswith("/"):
# Slash-command loop: route through the slash pipeline, not the
# model. No model reply to evaluate — complete immediately.
_notif_release_turn(session)
try:
parts = wakeup.lstrip()[1:].split(None, 1)
resp = _methods["command.dispatch"](
rid,
{"name": parts[0] if parts else "", "arg": parts[1] if len(parts) > 1 else "", "session_id": sid},
)
payload = (resp or {}).get("result") or {}
out = str(payload.get("output") or "").strip()
if out:
_emit("status.update", sid, {"kind": "loop", "text": out})
if payload.get("type") == "send" and payload.get("message"):
# Command resolves to a prompt (skill command etc.) — run it
# as a normal turn; the post-turn hook completes the tick.
with session["history_lock"]:
if session.get("running"):
mgr.abandon_tick()
return
session["running"] = True
_emit("message.start", sid)
_run_prompt_submit(rid, sid, session, payload["message"])
return
except Exception:
pass
decision = mgr.complete_tick("")
if decision.get("message"):
_emit("status.update", sid, {"kind": "loop", "text": decision["message"]})
if not wakeup.lstrip().startswith("/"):
_emit("message.start", sid)
_run_prompt_submit(rid, sid, session, wakeup)
return
_emit("message.start", sid)
_run_prompt_submit(rid, sid, session, wakeup)
# Slash-command loop: route through the slash pipeline, not the model. No model reply
# to evaluate — complete immediately.
_notif_release_turn(session)
try:
parts = wakeup.lstrip()[1:].split(None, 1)
resp = _methods["command.dispatch"](
rid, {"name": parts[0] if parts else "", "arg": parts[1] if len(parts) > 1 else "", "session_id": sid}
)
payload = (resp or {}).get("result") or {}
out = str(payload.get("output") or "").strip()
if out:
_emit("status.update", sid, {"kind": "loop", "text": out})
if payload.get("type") == "send" and payload.get("message"):
# Command resolves to a prompt (skill command etc.) — run it as a normal turn;
# the post-turn hook completes the tick.
if not _notif_claim_turn(session):
mgr.abandon_tick()
return
_emit("message.start", sid)
_run_prompt_submit(rid, sid, session, payload["message"])
return
except Exception:
pass
decision = mgr.complete_tick("")
if decision.get("message"):
_emit("status.update", sid, {"kind": "loop", "text": decision["message"]})
except Exception as exc:
_notif_log_failure("loop wakeup dispatch failed", exc)
_notif_release_turn(session)
@@ -219,45 +220,52 @@ def _maybe_fire_tui_loop_tick(sid: str, session: dict) -> None:
mgr.abandon_tick()
def _kb_first_line(value: Any, limit: int) -> str:
lines = str(value).strip().splitlines()
return f"\n{lines[0][:limit]}" if lines else ""
def _kb_completed(task, payload: dict, title: str) -> str:
if payload.get("summary"):
handoff = _kb_first_line(payload["summary"], 200)
else:
handoff = _kb_first_line(task.result, 160) if getattr(task, "result", None) else ""
return f" done — {title}{handoff}"
def _kb_timed_out(task, payload: dict, title: str) -> str:
try:
limit = int(payload.get("limit_seconds") or 0)
except (TypeError, ValueError):
limit = 0
return f" timed out (max_runtime={limit}s); will retry"
# kind -> (glyph, suffix after "Kanban <id>"); silent kinds (archived/unblocked) are absent → None.
_KANBAN_EVENT_FORMATTERS = {
"completed": ("✔", _kb_completed),
"blocked": ("⏸", lambda t, p, title: " blocked" + (f": {str(p.get('reason'))[:160]}" if p.get("reason") else "")),
"gave_up": ("✖", lambda t, p, title: " gave up after repeated spawn failures"
+ (f"\n{str(p.get('error'))[:200]}" if p.get("error") else "")),
"crashed": ("✖", lambda t, p, title: " worker crashed (pid gone); dispatcher will retry"),
"timed_out": ("⏱", _kb_timed_out),
"status": ("🔄", lambda t, p, title: f" → {p.get('status') or ''}"),
}
def _format_kanban_event_text(sub: dict, task, ev, board_slug: str) -> Optional[str]:
"""Single-line notification text for one kanban event; wording mirrors
gateway/kanban_watchers.py so it reads the same as on Telegram. None for silent kinds."""
kind = getattr(ev, "kind", "")
if not kind or kind in _KANBAN_SILENT_KINDS:
"""Single-line notification text for one kanban event; wording mirrors gateway/kanban_watchers.py
so it reads the same as on Telegram. None for silent kinds."""
entry = _KANBAN_EVENT_FORMATTERS.get(getattr(ev, "kind", ""))
if entry is None:
return None
glyph, fmt = entry
task_id = sub.get("task_id", "")
title = (getattr(task, "title", None) or task_id)[:120]
board_tag = f"[{board_slug}] " if board_slug else ""
who = getattr(task, "assignee", None) or ""
tag = f"@{who} " if who else ""
payload = getattr(ev, "payload", None) or {}
if kind == "completed":
handoff = ""
summary = payload.get("summary")
if summary:
lines = str(summary).strip().splitlines()
handoff = f"\n{lines[0][:200]}" if lines else ""
elif getattr(task, "result", None):
lines = str(task.result).strip().splitlines()
handoff = f"\n{lines[0][:160]}" if lines else ""
return f"✔ {board_tag}{tag}Kanban {task_id} done — {title}{handoff}"
if kind == "blocked":
reason = f": {str(payload.get('reason'))[:160]}" if payload.get("reason") else ""
return f"⏸ {board_tag}{tag}Kanban {task_id} blocked{reason}"
if kind == "gave_up":
err = f"\n{str(payload.get('error'))[:200]}" if payload.get("error") else ""
return f"✖ {board_tag}{tag}Kanban {task_id} gave up after repeated spawn failures{err}"
if kind == "crashed":
return f"✖ {board_tag}{tag}Kanban {task_id} worker crashed (pid gone); dispatcher will retry"
if kind == "timed_out":
try:
limit = int(payload.get("limit_seconds") or 0)
except (TypeError, ValueError):
limit = 0
return f"⏱ {board_tag}{tag}Kanban {task_id} timed out (max_runtime={limit}s); will retry"
if kind == "status":
return f"🔄 {board_tag}{tag}Kanban {task_id} → {payload.get('status') or ''}"
return None
return f"{glyph} {board_tag}{tag}Kanban {task_id}{fmt(task, getattr(ev, 'payload', None) or {}, title)}"
def _collect_kanban_notifications(session: dict) -> list:
@@ -283,8 +291,8 @@ def _collect_kanban_notifications(session: dict) -> list:
boards = [_kb.read_board_metadata(_kb.DEFAULT_BOARD)]
except Exception:
return []
# Poll each resolved DB path once — multiple slugs can point at the same
# DB when HERMES_KANBAN_DB pins the board path (same guard as the gateway).
# Poll each resolved DB path once — multiple slugs can point at the same DB when
# HERMES_KANBAN_DB pins the board path (same guard as the gateway).
seen_db_paths: set = set()
for board_meta in boards:
slug = (board_meta or {}).get("slug") or _kb.DEFAULT_BOARD
@@ -296,9 +304,9 @@ def _collect_kanban_notifications(session: dict) -> list:
if resolved in seen_db_paths:
continue
seen_db_paths.add(resolved)
# One poller per live session: don't open this board writable unless it
# has a subscription owned by this exact session. If the read-only probe
# fails (locked/corrupt DB), preserve delivery and fall through.
# One poller per live session: don't open this board writable unless it has a subscription
# owned by this exact session. If the read-only probe fails (locked/corrupt DB), preserve
# delivery and fall through.
try:
if _kb.count_notify_subs(board=slug, platform="tui", chat_id=session_key) == 0:
continue
@@ -324,13 +332,10 @@ def _collect_kanban_notifications(session: dict) -> list:
if not events:
continue
task = _kb.get_task(conn, sub["task_id"])
for ev in events:
text = _format_kanban_event_text(sub, task, ev, slug)
if text:
texts.append(text)
# Unsubscribe only on archive: ``done`` is reversible in review/controller flows,
# so keeping the sub lets a later reopen notify the same session. The claimed
# cursor prevents replay.
texts.extend(t for t in (_format_kanban_event_text(sub, task, ev, slug) for ev in events) if t)
# Unsubscribe only on archive: ``done`` is reversible in review/controller flows, so
# keeping the sub lets a later reopen notify the same session. The claimed cursor
# prevents replay.
if task and getattr(task, "status", "") == "archived":
with contextlib.suppress(Exception):
_kb.remove_notify_sub(conn, **sub_ident)
@@ -348,9 +353,9 @@ def _notif_poll_kanban(sid: str, session: dict) -> None:
except Exception as _kb_exc:
_notif_log_failure("kanban notification poll failed", _kb_exc)
_kanban_texts = []
for _kb_text in _kanban_texts:
_emit("status.update", sid, {"kind": "process", "text": _kb_text})
if _kanban_texts:
for _kb_text in _kanban_texts:
_emit("status.update", sid, {"kind": "process", "text": _kb_text})
session.setdefault("_kanban_pending", []).extend(_kanban_texts)
_pending = session.get("_kanban_pending") or []
if not _pending:
@@ -362,13 +367,11 @@ def _notif_poll_kanban(sid: str, session: dict) -> None:
_batch = list(_pending)
session["_kanban_pending"] = []
if _batch:
rid = f"__notif__{int(time.time() * 1000)}"
try:
_emit("message.start", sid)
_run_prompt_submit(rid, sid, session, "\n".join(_batch))
except Exception as exc:
_notif_log_failure("kanban notification dispatch failed", exc)
_notif_release_turn(session)
with contextlib.suppress(Exception):
_notif_submit(
f"__notif__{int(time.time() * 1000)}", sid, session, "\n".join(_batch),
"kanban notification dispatch failed",
)
def _notif_render_event(sid: str, evt: dict, emitted: set, process_registry, format_process_notification) -> Optional[str]:
@@ -391,24 +394,77 @@ def _notif_dispatch_event(sid: str, session: dict, evt: dict, text: str) -> None
"""Run the claimed (running=True) agent turn for one notification event."""
from tools.async_delegation import claim_event_delivery, complete_event_delivery, release_event_delivery
rid = f"__notif__{int(time.time() * 1000)}"
claim = claim_event_delivery(evt, "tui-poller")
if claim is None:
return
kwargs = (
{"display_kind": "async_delegation_complete", "display_metadata": _async_delegation_display_metadata(evt)}
if evt.get("type") == "async_delegation" else {}
)
try:
_emit("message.start", sid)
if evt.get("type") == "async_delegation":
_run_prompt_submit(
rid, sid, session, text, display_kind="async_delegation_complete",
display_metadata=_async_delegation_display_metadata(evt),
)
else:
_run_prompt_submit(rid, sid, session, text)
complete_event_delivery(evt, claim)
except Exception as exc:
_notif_submit(
f"__notif__{int(time.time() * 1000)}", sid, session, text, "notification poller dispatch failed", **kwargs
)
except Exception:
release_event_delivery(evt, claim)
_notif_log_failure("notification poller dispatch failed", exc)
_notif_release_turn(session)
return
complete_event_delivery(evt, claim)
def _notif_classify_event(sid: str, session: dict, evt: dict) -> str:
"""'foreign' (another live session owns it), 'unowned' (addressed but unprovable — never adopt
an orphan), or 'mine' (ours, or an ownerless legacy notification kept process-global)."""
if _notification_event_belongs_elsewhere(sid, session, evt):
return "foreign"
if _notification_event_requires_owner(evt) and not _session_owns_notification_event(sid, session, evt):
return "unowned"
return "mine"
def _notif_handle_event(sid, session, evt, emitted, registry, fmt, deferred) -> bool:
"""Route one dequeued event. ``deferred`` is None in the live loop (foreign events go straight
back on the queue) and the shutdown-drain list otherwise (foreign + orphaned delegation
payloads are set aside for a resume). Returns False when the drain must stop (session busy)."""
queue = registry.completion_queue
owner = _notif_classify_event(sid, session, evt)
if owner == "foreign":
if deferred is None:
# Leave foreign events for their owner — otherwise a process started in session A
# surfaces its completion in whichever poller wakes first.
queue.put(evt)
time.sleep(0.1)
else:
deferred.append(evt)
return True
if owner == "unowned":
is_delegation = evt.get("type") == "async_delegation"
origin, key = str(evt.get("origin_ui_session_id") or ""), str(evt.get("session_key") or "")
if deferred is None:
(logger.warning if is_delegation else logger.debug)(
"Dropping unowned %s notification (origin=%r key=%r) instead of delivering to session %s",
evt.get("type", "completion"), origin, key, sid,
)
elif is_delegation:
deferred.append(evt)
else:
logger.debug(
"Dropping unowned %s notification during shutdown drain (origin=%r key=%r)",
evt.get("type", "completion"), origin, key,
)
return True
text = _notif_render_event(sid, evt, emitted, registry, fmt)
if not text:
return True
if not _notif_claim_turn(session):
queue.put(evt)
if deferred is not None:
return False
# Back off: the re-queued event keeps the queue non-empty, so without a sleep this loop
# spins at 100% CPU while busy.
time.sleep(0.25)
return True
_notif_dispatch_event(sid, session, evt, text)
return True
def _notification_poller_loop(stop_event: threading.Event, sid: str, session: dict) -> None:
@@ -420,9 +476,9 @@ def _notification_poller_loop(stop_event: threading.Event, sid: str, session: di
``_KANBAN_POLL_SECONDS`` — the delivery path for platform="tui" rows."""
from tools.process_registry import process_registry, format_process_notification
queue = process_registry.completion_queue
_emitted = set() # dedup re-queued events so one completion isn't emitted 50 times while busy
_last_kanban_poll = 0.0
_last_loop_poll = 0.0
_last_kanban_poll = _last_loop_poll = 0.0
while not stop_event.is_set() and not session.get("_finalized"):
_now = time.monotonic()
# /loop wakeup driver: fire a due tick for THIS session while idle (same claim-under-lock
@@ -437,85 +493,23 @@ def _notification_poller_loop(stop_event: threading.Event, sid: str, session: di
_last_kanban_poll = _now
_notif_poll_kanban(sid, session)
try:
evt = process_registry.completion_queue.get(timeout=0.5)
evt = queue.get(timeout=0.5)
except Exception:
continue
_notif_handle_event(sid, session, evt, _emitted, process_registry, format_process_notification, None)
# Leave foreign events for their owner — otherwise a process started in
# session A surfaces its completion in whichever poller wakes first.
if _notification_event_belongs_elsewhere(sid, session, evt):
process_registry.completion_queue.put(evt)
time.sleep(0.1)
continue
# Not owned by another LIVE session, but addressed events still need positive proof
# (exact UI origin, direct durable key, or compression lineage) — an orphan must not be
# adopted by this chat. Truly ownerless ordinary notifications keep legacy global delivery.
if _notification_event_requires_owner(evt) and not _session_owns_notification_event(sid, session, evt):
log = logger.warning if evt.get("type") == "async_delegation" else logger.debug
log(
"Dropping unowned %s notification (origin=%r key=%r) instead of delivering to session %s",
evt.get("type", "completion"), str(evt.get("origin_ui_session_id") or ""),
str(evt.get("session_key") or ""), sid,
)
continue
text = _notif_render_event(sid, evt, _emitted, process_registry, format_process_notification)
if not text:
continue
_requeued = False
with session["history_lock"]:
if session.get("running"):
process_registry.completion_queue.put(evt)
_requeued = True
else:
session["running"] = True
if _requeued:
# Back off: the re-queued event keeps the queue non-empty, so
# without a sleep this loop spins at 100% CPU while busy.
time.sleep(0.25)
continue
_notif_dispatch_event(sid, session, evt, text)
# Drain remaining events after the stop signal so nothing is lost on shutdown. Other live
# sessions' events are set aside and re-queued; orphaned events (owner gone) are dropped —
# same guard as the main loop, except orphaned delegation payloads are deferred for a resume.
# Drain remaining events after the stop signal so nothing is lost on shutdown; foreign and
# orphaned-delegation events are handed back to the shared queue afterwards.
deferred: list = []
while not process_registry.completion_queue.empty():
while not queue.empty():
try:
evt = process_registry.completion_queue.get_nowait()
evt = queue.get_nowait()
except Exception:
break
if _notification_event_belongs_elsewhere(sid, session, evt):
deferred.append(evt)
continue
if _notification_event_requires_owner(evt) and not _session_owns_notification_event(sid, session, evt):
if evt.get("type") == "async_delegation":
deferred.append(evt)
else:
logger.debug(
"Dropping unowned %s notification during shutdown drain (origin=%r key=%r)",
evt.get("type", "completion"), str(evt.get("origin_ui_session_id") or ""),
str(evt.get("session_key") or ""),
)
continue
text = _notif_render_event(sid, evt, _emitted, process_registry, format_process_notification)
if not text:
continue
with session["history_lock"]:
if session.get("running"):
process_registry.completion_queue.put(evt)
break
session["running"] = True
_notif_dispatch_event(sid, session, evt, text)
# Hand any other sessions' events back to the shared queue.
if not _notif_handle_event(sid, session, evt, _emitted, process_registry, format_process_notification, deferred):
break
for evt in deferred:
process_registry.completion_queue.put(evt)
queue.put(evt)
def _async_delegation_display_metadata(evt: dict) -> dict:
@@ -542,11 +536,6 @@ def _wire_agent_terminal_output() -> None:
`_emit` is `_stdout_lock`-guarded, so the registry's reader threads may call it."""
from tools.process_registry import process_registry
has_output_sink = getattr(process_registry, "on_output", None) is not None
has_close_sink = getattr(process_registry, "on_close", None) is not None
if has_output_sink and has_close_sink:
return
def _owner_sid_for_process(session) -> str:
session_key = str(getattr(session, "session_key", "") or "")
if not session_key:
@@ -557,18 +546,16 @@ def _wire_agent_terminal_output() -> None:
return sid
return ""
def _emit_agent_terminal_output(session, chunk):
_emit("agent.terminal.output", _owner_sid_for_process(session), {"process_id": session.id, "chunk": chunk})
def _emit_agent_terminal_close(session, process_id):
# session may be None (process already finished/pruned) — the tab can
# still linger and be closed; route to the owning window when we can.
sid = _owner_sid_for_process(session) if session is not None else ""
_emit("terminal.close", sid, {"process_id": process_id})
# session may be None (process already finished/pruned) — the tab can still linger and
# be closed; route to the owning window when we can.
_emit("terminal.close", _owner_sid_for_process(session) if session is not None else "", {"process_id": process_id})
if not has_output_sink:
process_registry.on_output = _emit_agent_terminal_output
if not has_close_sink:
if getattr(process_registry, "on_output", None) is None:
process_registry.on_output = lambda session, chunk: _emit(
"agent.terminal.output", _owner_sid_for_process(session), {"process_id": session.id, "chunk": chunk}
)
if getattr(process_registry, "on_close", None) is None:
process_registry.on_close = _emit_agent_terminal_close
@@ -586,7 +573,6 @@ def _wire_desktop_ui() -> None:
from tools import desktop_ui
except Exception:
return
desktop_ui.set_emitter(lambda sid, event, payload: _emit(event, sid, payload))
_desktop_ui_wired = True
@@ -598,11 +584,10 @@ _notification_pollers: list = []
def _start_notification_poller(sid: str, session: dict) -> threading.Event:
"""Start the background notification poller for a TUI session."""
"""Start the background notification poller for a TUI session (thread name is greppable)."""
_wire_agent_terminal_output()
_wire_desktop_ui()
stop = threading.Event()
# Thread name is greppable for debuggers/test teardowns.
t = threading.Thread(
target=_notification_poller_loop, args=(stop, sid, session), daemon=True, name=f"tui-notif-poller-{sid}"
)
tui_gateway/slash_fuzzy.py
+28 -50
View File
@@ -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)
tui_gateway/slash_worker.py
+25 -35
View File
@@ -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
tui_gateway/synthetic_turn.py
+30 -61
View File
@@ -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"]
tui_gateway/turn_marker.py
+46 -66
View File
@@ -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}