diff --git a/tui_gateway/session_auto_continue.py b/tui_gateway/session_auto_continue.py index d4313c21a3..494fbf5e28 100644 --- a/tui_gateway/session_auto_continue.py +++ b/tui_gateway/session_auto_continue.py @@ -1,23 +1,16 @@ -"""Auto-continue: resume a turn killed by a process/machine death, plus queued-prompt drain and busy-submit handling. - -Bodies are rebound onto server.py's globals at install time (see -method_ctx.bind_module), so they reference server.py globals bare. -""" +"""Auto-continue: resume a turn killed by a process/machine death, plus queued-prompt drain and +busy-submit handling. 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 -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. +from .method_ctx import bind_module +# 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 _AUTO_CONTINUE_MAX_ATTEMPTS_DEFAULT = 2 @@ -27,8 +20,7 @@ def _auto_continue_config() -> tuple[bool, float, int]: """(enabled, freshness window in seconds, max attempts) from config.yaml.""" desktop = _load_cfg().get("desktop") cfg = desktop.get("auto_continue") if isinstance(desktop, dict) else None - if not isinstance(cfg, dict): - cfg = {} + cfg = cfg if isinstance(cfg, dict) else {} try: minutes = float(cfg.get("freshness_minutes", _AUTO_CONTINUE_FRESHNESS_MINUTES_DEFAULT)) except (TypeError, ValueError): @@ -42,18 +34,14 @@ def _auto_continue_config() -> tuple[bool, float, int]: def _session_home(session: dict) -> Path: """The HERMES_HOME the session's durable state lives in (profile-aware).""" - profile_home = session.get("profile_home") - return Path(profile_home) if profile_home else Path(_hermes_home) + return Path(session.get("profile_home") or _hermes_home) 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. - """ + """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.""" home = _session_home(session) for key in dict.fromkeys((*keys, str(session.get("session_key") or ""))): if key: @@ -61,8 +49,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 " @@ -84,15 +72,12 @@ def _ac_dispatch_failed(what: str, exc: BaseException) -> None: 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. - """ - # Hosted room turns are recovered by their durable task/lease state machine; generic - # auto-continue would bypass its execution generation and duplicate work. + """Kick off a continuation turn for a crash-interrupted session (session.resume 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. if session.get("source") == "bot_room": return None home = _session_home(session) @@ -130,17 +115,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. + # 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: @@ -152,28 +137,28 @@ def _maybe_schedule_auto_continue(sid: str, session: dict, session_key: str) -> _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 - ) + logger.info("auto-continue scheduled for session %s (attempt %d, interrupted %.0fs ago)", session_key, attempt, age) return {"attempt": attempt, "interrupted_at": marker["started_at"]} -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. +def _ac_inflight_original(session: dict) -> str: + turn = session.get("inflight_turn") + return str(turn.get("user") or "").strip() if isinstance(turn, dict) else "" - 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. - """ + +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.""" image_paths = list(image_paths or []) - # Scrub live-turn self-duplicates first so the text merge below can't glue - # "{original}\n\n{later}" and re-fire the original after a correction settles. + # Scrub live-turn self-duplicates first so the text merge below can't glue "{original}\n\n{later}" + # and re-fire the original after a correction settles. _drop_queued_duplicates_of_inflight_user(session) # 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 = _ac_inflight_original(session) if original and text.strip() == original: return queued = {"text": text, "transport": transport} @@ -193,12 +178,10 @@ def _enqueue_prompt(session: dict, text: Any, transport: Any, image_paths: list[ 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). - """ + """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).""" if not isinstance(entry, dict): return None text = entry.get("text") @@ -217,14 +200,11 @@ def _sanitize_queued_entry_vs_inflight_user(entry: Any, original: str) -> dict | 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. - """ - turn = session.get("inflight_turn") - original = str(turn.get("user") or "").strip() if isinstance(turn, dict) else "" + """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.""" + original = _ac_inflight_original(session) if not original: return head = session.get("queued_prompt") @@ -243,12 +223,10 @@ 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. - """ + """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, and 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") if not use_agent and not _session_uses_compute_host(session): return @@ -274,8 +252,8 @@ def _interrupt_busy_session(sid: str, session: dict, agent: Any) -> None: def _ac_try_correction(rid, session: dict, agent: Any, method: str, plain_text: str, status: str) -> dict | None: """Apply ``agent.(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.""" + 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 @@ -290,24 +268,19 @@ def _ac_try_correction(rid, session: dict, agent: Any, method: str, plain_text: 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. - """ + ``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.""" mode = "queue" if queued else _load_busy_input_mode() agent = session.get("agent") with session["history_lock"]: if not session.get("running"): - # Turn ended since prompt.submit's busy check; caller retries on the idle session. - return None + return None # turn ended since prompt.submit's busy check; caller retries on the idle session image_paths = list(session.get("attached_images", [])) if image_paths: - # Claim now so a later paste isn't consumed by this prompt when the turn yields. - session["attached_images"] = [] + session["attached_images"] = [] # claim now so a later paste isn't consumed when the turn yields text_only = not image_paths and _is_text_only_busy_payload(text) plain_text = _coerce_message_text(text).strip() if text_only else "" if plain_text and agent is not None: @@ -315,13 +288,9 @@ def _handle_busy_submit(rid, sid: str, session: dict, text: Any, transport: Any, 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") - ): + # 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 @@ -334,21 +303,18 @@ def _handle_busy_submit(rid, sid: str, session: dict, text: Any, transport: Any, 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"}) 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. - """ + """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.""" with session["history_lock"]: if session.get("_closing"): return False @@ -405,10 +371,8 @@ def _inflight_snapshot(session: dict) -> dict | None: turn = session.get("inflight_turn") if not isinstance(turn, dict): return None - user = str(turn.get("user") or "").strip() - assistant = str(turn.get("assistant") or "") - streaming = bool(turn.get("streaming")) - error = str(turn.get("error") or "").strip() + user, assistant = str(turn.get("user") or "").strip(), str(turn.get("assistant") or "") + streaming, error = bool(turn.get("streaming")), str(turn.get("error") or "").strip() if not user and not assistant and not streaming and not error: return None snapshot = {"assistant": assistant, "streaming": streaming, "user": user} @@ -418,17 +382,15 @@ def _inflight_snapshot(session: dict) -> dict | None: 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. - snapshot["error"] = error - snapshot["status"] = str(turn.get("status") or "error") - snapshot["recoverable"] = bool(turn.get("recoverable")) + # Retained failed turn (_fail_inflight_turn): a resuming client must rebuild the failed bubble, + # not render the partial text as a healthy reply. + snapshot.update(error=error, status=str(turn.get("status") or "error"), recoverable=bool(turn.get("recoverable"))) surface = turn.get("error_surface") if isinstance(surface, dict) and surface: snapshot["error_surface"] = surface @@ -440,13 +402,11 @@ def _emit_terminal_turn_error( ) -> 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. - """ + 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 + ({layer, code, retryable} so the desktop can say "Provider error" / "Gateway error"; advisory, + never raises).""" agent = session.get("agent") - # {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 @@ -463,31 +423,24 @@ def _emit_terminal_turn_error( partial = str(turn.get("assistant") or "") cols = int(session.get("cols", 80)) text = partial or f"Error: {message}" - payload = { - "text": text, - "usage": _get_usage(agent) if agent is not None else {}, - "status": "error", - "error": message, - "recoverable": True, - } - if error_surface: - payload["error_surface"] = error_surface - if partial: - payload["partial"] = True try: rendered = render_message(text, cols) except Exception: rendered = "" - if rendered: - payload["rendered"] = rendered + payload = { + "text": text, "usage": _get_usage(agent) if agent is not None else {}, "status": "error", + "error": message, "recoverable": True, + **({"error_surface": error_surface} if error_surface else {}), + **({"partial": True} if partial else {}), **({"rendered": rendered} if rendered else {}), + } if retire_marker: _retire_turn_marker(session) _emit("message.complete", sid, payload) 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 @@ -501,9 +454,7 @@ 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).""" queued = session.get("queued_prompt") - if not isinstance(queued, dict): - return None - user = _inflight_text(queued.get("text")) + user = _inflight_text(queued.get("text")) if isinstance(queued, dict) else "" return {"user": user} if user else None diff --git a/tui_gateway/session_compression.py b/tui_gateway/session_compression.py index f0fb53a2f8..668232f950 100644 --- a/tui_gateway/session_compression.py +++ b/tui_gateway/session_compression.py @@ -9,9 +9,7 @@ from __future__ import annotations import contextlib -from .method_ctx import HandlerRegistry, bind_module - -_registry = HandlerRegistry() +from .method_ctx import bind_module def _tui_compression_config_signature(cfg: dict | None) -> tuple: @@ -78,36 +76,31 @@ _COMPRESSION_INT_KEYS = ( def _apply_live_compression_config(agent: Any, cfg: dict | None) -> None: - """Update a live session's compressor from current config.yaml, preserving the agent object, - session identity, history and callbacks. Recomputes the trigger from the ratio threshold, then - applies ``compression.threshold_tokens`` so raising/lowering/clearing the cap lands next preflight. + """Update a live session's compressor from current config.yaml, preserving the agent object, session + identity, history and callbacks. Recomputes the trigger from the ratio threshold, then applies + ``compression.threshold_tokens`` so raising/lowering/clearing the cap lands next preflight. Every adopted key has UNSET semantics: a removed key restores the normalized default (or the - model-derived value) through the construction path's own derivation (ctor signature defaults, - Codex autoraise, deferred context-length re-inference). Acting only on PRESENT keys would leave - stale values active forever. + model-derived value) through the construction path's own derivation (ctor signature defaults, Codex + autoraise, deferred context-length re-inference). Acting only on PRESENT keys would leave stale + values active forever. """ cfg = cfg if isinstance(cfg, dict) else {} - compression_raw = cfg.get("compression") - compression = compression_raw if isinstance(compression_raw, dict) else {} + compression = cfg.get("compression") if isinstance(cfg.get("compression"), dict) else {} model_cfg = cfg.get("model") if isinstance(cfg.get("model"), dict) else {} enabled_raw = compression.get("enabled", True) 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) try: - if isinstance(native_threshold_raw, bool): - raise ValueError native_threshold = int(native_threshold_raw) - if native_threshold <= 0: + if isinstance(native_threshold_raw, bool) or native_threshold <= 0: raise ValueError except (TypeError, ValueError): logger.warning("Invalid compression.codex_responses_compact_threshold=%r; using 200000.", native_threshold_raw) native_threshold = 200_000 agent.codex_responses_compact_threshold = native_threshold - # Absence restores the agent_init/config default (0 = disabled). with contextlib.suppress(TypeError, ValueError): agent.compression_idle_compact_after_seconds = max(0, int(compression.get("idle_compact_after_seconds", 0) or 0)) @@ -115,30 +108,22 @@ def _apply_live_compression_config(agent: Any, cfg: dict | None) -> None: cc = getattr(agent, "context_compressor", None) if cc is None: return - # tail_mode: unknown/absent values land on the ctor default ("lean"), matching agent_init. default_tail = str(_compressor_ctor_default("tail_mode", "lean")) mode = str(compression.get("tail_mode", default_tail) or default_tail).strip().lower() cc.tail_mode = mode if mode in ("legacy", "lean") else default_tail - for key, fallback, min_value in _COMPRESSION_INT_KEYS: default = int(_compressor_ctor_default(key, fallback)) raw = compression.get(key, default) - try: - value = default if raw is None else int(raw) - except (TypeError, ValueError): - continue - setattr(cc, key, max(min_value, value)) - + with contextlib.suppress(TypeError, ValueError): + setattr(cc, key, max(min_value, default if raw is None else int(raw))) 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)) - # Absent or invalid shape (agent_init treats both as empty): stale overrides must stop steering. raw_thresholds = compression.get("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) + 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). @@ -148,23 +133,17 @@ def _apply_live_compression_config(agent: Any, cfg: dict | None) -> None: pct = float(compression["threshold"]) if pct is None: pct = _derived_default_threshold_percent(agent, compression) - with contextlib.suppress(TypeError, ValueError): - cc._config_threshold_percent = pct - cc._configured_threshold_percent = pct - base = pct - model_thresholds = getattr(cc, "model_thresholds", None) or {} - if model_thresholds: - from agent.context_compressor import resolve_model_threshold + cc._config_threshold_percent = cc._configured_threshold_percent = pct + base = pct + if cc.model_thresholds: + from agent.context_compressor import resolve_model_threshold - base = resolve_model_threshold(getattr(agent, "model", "") or "", model_thresholds, pct) - cc._base_threshold_percent = base - if hasattr(cc, "_effective_threshold_percent"): - try: - cc.threshold_percent = cc._effective_threshold_percent(cc.context_length, base) - except Exception: - cc.threshold_percent = pct - else: - cc.threshold_percent = pct + base = resolve_model_threshold(getattr(agent, "model", "") or "", cc.model_thresholds, pct) + cc._base_threshold_percent = base + try: + cc.threshold_percent = cc._effective_threshold_percent(cc.context_length, base) + except Exception: + cc.threshold_percent = pct raw_ctx = model_cfg.get("context_length") if raw_ctx is not None: @@ -177,40 +156,13 @@ def _apply_live_compression_config(agent: Any, cfg: dict | None) -> None: with contextlib.suppress(Exception): cc.context_length = new_ctx 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. - cc._config_context_length = None - cc._resolved_context_length = None - - coerce_cap = getattr(cc, "_coerce_threshold_tokens_cap", None) - if callable(coerce_cap): - cc.threshold_tokens_cap = coerce_cap(compression.get("threshold_tokens")) - elif "threshold_tokens" in compression: - try: - cap = int(compression.get("threshold_tokens")) - cc.threshold_tokens_cap = cap if cap > 0 else None - except (TypeError, ValueError): - cc.threshold_tokens_cap = None - else: - cc.threshold_tokens_cap = 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. + cc._config_context_length = cc._resolved_context_length = None + cc.threshold_tokens_cap = cc._coerce_threshold_tokens_cap(compression.get("threshold_tokens")) # Invalidate the cached trigger so the next preflight re-derives from percent/window, then the cap. - if hasattr(cc, "_threshold_tokens"): - cc._threshold_tokens = None - if hasattr(cc, "_tail_token_budget"): - cc._tail_token_budget = None - elif hasattr(cc, "_apply_threshold_tokens_cap"): - 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) - ) - cc._apply_threshold_tokens_cap() - else: - cap = getattr(cc, "threshold_tokens_cap", None) - current = getattr(cc, "threshold_tokens", None) - if cap and current: - cc.threshold_tokens = min(int(current), int(cap)) + cc._threshold_tokens = cc._tail_token_budget = None def _sync_agent_compression_with_config(sid: str, session: dict) -> None: diff --git a/tui_gateway/session_notifications.py b/tui_gateway/session_notifications.py index b782921735..5652220ae5 100644 --- a/tui_gateway/session_notifications.py +++ b/tui_gateway/session_notifications.py @@ -7,9 +7,7 @@ from __future__ import annotations import contextlib -from .method_ctx import HandlerRegistry, bind_module - -_registry = HandlerRegistry() +from .method_ctx import bind_module def _notif_locked_sessions(fn, default): @@ -58,16 +56,16 @@ def _notification_event_belongs_elsewhere(sid: str, session: dict, evt: dict) -> return False if _notif_locked_sessions(lambda ss: evt_ui_sid in ss and not ss[evt_ui_sid].get("_finalized"), False): 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. + # 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 - # parent, else a stale parent tab could consume the event before the current conversation. + # 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 parent, else a + # stale parent tab could consume the event before the current conversation. resolved_key = _notif_resolve_event_key(evt_key) if resolved_key != evt_key: if resolved_key in current_keys: @@ -85,9 +83,9 @@ def _notification_event_belongs_elsewhere(sid: str, session: dict, evt: dict) -> def _session_owns_notification_event(sid: str, session: dict, evt: dict) -> bool: """True iff *this* session PROVABLY owns ``evt``: positive mirror of - ``_notification_event_belongs_elsewhere`` minus its orphan-adoption fallback (UI origin is - this live session, or ``session_key`` raw/compression-resolved matches this session). Fail- - closed gate for every addressed notification — "not provably elsewhere" is NOT enough.""" + ``_notification_event_belongs_elsewhere`` minus its orphan-adoption fallback (UI origin is this + live session, or ``session_key`` raw/compression-resolved matches). Fail-closed gate for every + addressed notification — "not provably elsewhere" is NOT enough.""" if session.get("_finalized"): return False if str(evt.get("origin_ui_session_id") or "") == str(sid or ""): @@ -106,27 +104,28 @@ def _notification_event_requires_owner(evt: dict) -> bool: ) +# Extra dedup fields per event type. Completions are terminal (one-shot per process session); watch +# events are not — one process can match patterns many times, so their content is part of the key. +_DEDUP_EXTRA_FIELDS = { + "watch_match": ("command", "pattern", "output", "suppressed", "message_id"), + "watch_disabled": ("command", "message", "suppressed"), + "watch_overflow_": ("command", "message", "suppressed"), # prefix match +} + + def _notification_event_dedup_key(evt: dict) -> tuple: - """UI-emission identity for a process notification event. Completions are terminal (one-shot - per process session); watch events are not — one process can match patterns many times, so - include event content to avoid suppressing later distinct matches.""" + """UI-emission identity for a process notification event.""" evt_type = evt.get("type", "completion") - evt_sid = evt.get("session_id", "") - if evt_type == "watch_match": - return (evt_sid, evt_type, evt.get("command", ""), evt.get("pattern", ""), evt.get("output", ""), - evt.get("suppressed", 0), evt.get("message_id", "")) - if evt_type.startswith("watch_overflow_") or evt_type == "watch_disabled": - return (evt_sid, evt_type, evt.get("command", ""), evt.get("message", ""), evt.get("suppressed", 0)) if evt_type == "async_delegation": - # No process session_id: without this every completion keys as - # ("", "async_delegation") and the second one is suppressed forever. + # No process session_id: without this every completion keys as ("", "async_delegation") + # and the second one is suppressed forever. return (evt.get("delegation_id", ""), evt_type) - return (evt_sid, evt_type) + extra = _DEDUP_EXTRA_FIELDS.get("watch_overflow_" if evt_type.startswith("watch_overflow_") else evt_type, ()) + return (evt.get("session_id", ""), evt_type, *(evt.get(f, 0 if f == "suppressed" else "") for f in extra)) -# 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. +# 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_POLL_SECONDS = 5.0 _LOOP_POLL_SECONDS = 5.0 @@ -161,10 +160,14 @@ def _notif_submit(rid: str, sid: str, session: dict, text: str, what: str, **kwa raise +def _notif_loop_status(sid: str, text: str) -> None: + _emit("status.update", sid, {"kind": "loop", "text": text}) + + 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 - a racing user prompt wins cleanly; the turn dispatcher's post-turn hook completes the tick.""" + """Fire a due /loop wakeup for an idle TUI/Desktop/dashboard session (per-session poller, coarse + cadence). Claims the session (running=True) before dispatching so a racing user prompt wins + cleanly; the turn dispatcher's post-turn hook completes the tick.""" try: from hermes_cli.loops import LoopManager, goal_blocks_loop_tick except Exception: @@ -179,16 +182,15 @@ def _maybe_fire_tui_loop_tick(sid: str, session: dict) -> None: 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…"}) + _notif_loop_status(sid, f"↻ /loop wakeup #{mgr.state.ticks_fired if mgr.state else '?'} firing…") if not wakeup.lstrip().startswith("/"): _emit("message.start", sid) _run_prompt_submit(rid, sid, session, wakeup) return - # Slash-command loop: route through the slash pipeline, not the model. No model reply - # to evaluate — complete immediately. + # 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) @@ -198,10 +200,10 @@ def _maybe_fire_tui_loop_tick(sid: str, session: dict) -> None: payload = (resp or {}).get("result") or {} out = str(payload.get("output") or "").strip() if out: - _emit("status.update", sid, {"kind": "loop", "text": out}) + _notif_loop_status(sid, 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. + # 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 @@ -212,7 +214,7 @@ def _maybe_fire_tui_loop_tick(sid: str, session: dict) -> None: pass decision = mgr.complete_tick("") if decision.get("message"): - _emit("status.update", sid, {"kind": "loop", "text": decision["message"]}) + _notif_loop_status(sid, decision["message"]) except Exception as exc: _notif_log_failure("loop wakeup dispatch failed", exc) _notif_release_turn(session) @@ -268,14 +270,49 @@ def _format_kanban_event_text(sub: dict, task, ev, board_slug: str) -> Optional[ return f"{glyph} {board_tag}{tag}Kanban {task_id}{fmt(task, getattr(ev, 'payload', None) or {}, title)}" +def _kb_board_key(_kb, board_meta) -> tuple[str, str]: + """(slug, resolved DB identity) — multiple slugs can point at one DB when HERMES_KANBAN_DB pins it.""" + slug = (board_meta or {}).get("slug") or _kb.DEFAULT_BOARD + db_path = (board_meta or {}).get("db_path") + try: + return slug, str(Path(db_path).expanduser().resolve() if db_path else _kb.kanban_db_path(slug).resolve()) + except Exception: + return slug, f"slug:{slug}" + + +def _kb_poll_board(_kb, conn, slug: str, session_key: str) -> list: + """Claim + format this session's unseen events on one open board connection.""" + try: + subs = _kb.list_notify_subs(conn) + except Exception: + return [] + texts: list = [] + for sub in subs: + if (sub.get("platform") or "").lower() != "tui" or sub.get("chat_id") != session_key: + continue + sub_ident = dict( + task_id=sub["task_id"], platform=sub["platform"], chat_id=sub["chat_id"], thread_id=sub.get("thread_id") or "" + ) + _old, _new, events = _kb.claim_unseen_events_for_sub(conn, kinds=_KANBAN_NOTIFY_KINDS, **sub_ident) + if not events: + continue + task = _kb.get_task(conn, sub["task_id"]) + 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) + return texts + + def _collect_kanban_notifications(session: dict) -> list: """Claim unseen terminal kanban events for this TUI session's subscriptions. ``kanban_create`` auto-subscribes TUI/desktop sessions with ``platform="tui"`` and - ``chat_id=HERMES_SESSION_KEY``; there is no "tui" messaging adapter, so this poller is the - delivery path. Same atomic cursor-claim (``claim_unseen_events_for_sub``) as the gateway - notifier, so a sub is delivered exactly once even if a gateway and a TUI poll the same board - DB. Returns formatted notification texts (may be empty).""" + ``chat_id=HERMES_SESSION_KEY``; there is no "tui" messaging adapter, so this poller is the delivery + path. Same atomic cursor-claim (``claim_unseen_events_for_sub``) as the gateway notifier, so a sub + is delivered exactly once even if a gateway and a TUI poll the same board DB.""" session_key = str(session.get("session_key") or "") if not session_key or session.get("_finalized"): return [] @@ -283,7 +320,6 @@ def _collect_kanban_notifications(session: dict) -> list: from hermes_cli import kanban_db as _kb except Exception: return [] - texts: list = [] try: boards = _kb.list_boards(include_archived=False) except Exception: @@ -291,22 +327,15 @@ 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). + texts: list = [] seen_db_paths: set = set() for board_meta in boards: - slug = (board_meta or {}).get("slug") or _kb.DEFAULT_BOARD - db_path = (board_meta or {}).get("db_path") - try: - resolved = str(Path(db_path).expanduser().resolve() if db_path else _kb.kanban_db_path(slug).resolve()) - except Exception: - resolved = f"slug:{slug}" + slug, resolved = _kb_board_key(_kb, board_meta) 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. try: if _kb.count_notify_subs(board=slug, platform="tui", chat_id=session_key) == 0: continue @@ -317,37 +346,15 @@ def _collect_kanban_notifications(session: dict) -> list: except Exception: continue try: - try: - subs = _kb.list_notify_subs(conn) - except Exception: - continue - for sub in subs: - if (sub.get("platform") or "").lower() != "tui" or sub.get("chat_id") != session_key: - continue - sub_ident = dict( - task_id=sub["task_id"], platform=sub["platform"], chat_id=sub["chat_id"], - thread_id=sub.get("thread_id") or "", - ) - _old, _new, events = _kb.claim_unseen_events_for_sub(conn, kinds=_KANBAN_NOTIFY_KINDS, **sub_ident) - if not events: - continue - task = _kb.get_task(conn, sub["task_id"]) - 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) + texts.extend(_kb_poll_board(_kb, conn, slug, session_key)) finally: conn.close() return texts def _notif_poll_kanban(sid: str, session: dict) -> None: - """One kanban poll: emit new texts, buffer them, and run the buffered batch as a turn if idle. - Events are cursor-claimed (never re-queued), so they are buffered until the session is idle - instead of dropping the agent turn.""" + """One kanban poll: emit new texts, buffer them, and run the buffered batch as a turn if idle. Events + are cursor-claimed (never re-queued), so they wait in the buffer instead of dropping the agent turn.""" try: _kanban_texts = _collect_kanban_notifications(session) except Exception as _kb_exc: @@ -357,27 +364,20 @@ def _notif_poll_kanban(sid: str, session: dict) -> None: _emit("status.update", sid, {"kind": "process", "text": _kb_text}) if _kanban_texts: session.setdefault("_kanban_pending", []).extend(_kanban_texts) - _pending = session.get("_kanban_pending") or [] - if not _pending: + if not session.get("_kanban_pending") or not _notif_claim_turn(session): return - _batch: list = [] with session["history_lock"]: - if not session.get("running"): - session["running"] = True - _batch = list(_pending) - session["_kanban_pending"] = [] - if _batch: - with contextlib.suppress(Exception): - _notif_submit( - f"__notif__{int(time.time() * 1000)}", sid, session, "\n".join(_batch), - "kanban notification dispatch failed", - ) + _batch, session["_kanban_pending"] = list(session.get("_kanban_pending") or []), [] + 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]: - """Format ``evt`` and emit its status.update once; None means skip the event (consumed - completion or unformattable). Re-queued completions would otherwise re-emit every 0.5s while - the session is busy, while distinct watch_match events from one process must stay visible.""" + """Format ``evt`` and emit its status.update once; None means skip the event (consumed completion + or unformattable). Re-queued completions would otherwise re-emit every 0.5s while the session is + busy, while distinct watch_match events from one process must stay visible.""" if evt.get("type") == "completion" and process_registry.is_completion_consumed(evt.get("session_id", "")): return None text = format_process_notification(evt) @@ -412,8 +412,8 @@ def _notif_dispatch_event(sid: str, session: dict, evt: dict, text: str) -> None 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).""" + """'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): @@ -422,15 +422,15 @@ def _notif_classify_event(sid: str, session: dict, evt: dict) -> str: 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).""" + """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. + # 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: @@ -459,8 +459,8 @@ def _notif_handle_event(sid, session, evt, emitted, registry, fmt, deferred) -> 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. + # 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) @@ -470,10 +470,10 @@ def _notif_handle_event(sid, session, evt, emitted, registry, fmt, deferred) -> def _notification_poller_loop(stop_event: threading.Event, sid: str, session: dict) -> None: """Poll completion_queue and dispatch notifications autonomously (daemon thread started by _init_session()): emit a status.update (kind=process), then chain an agent turn via - _run_prompt_submit if the session is idle. The queue is process-global: each poller requeues - events owned by another live session and drops addressed events whose owner is gone; - ownerless legacy notifications remain global. Also polls ``kanban_notify_subs`` every - ``_KANBAN_POLL_SECONDS`` — the delivery path for platform="tui" rows.""" + _run_prompt_submit if the session is idle. The queue is process-global: each poller requeues events + owned by another live session and drops addressed events whose owner is gone; ownerless legacy + notifications remain global. Also polls ``kanban_notify_subs`` every ``_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 @@ -481,8 +481,8 @@ def _notification_poller_loop(stop_event: threading.Event, sid: str, session: di _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 - # as kanban dispatch). An active non-parked /goal owns the idle boundary and defers it. + # /loop wakeup driver: fire a due tick for THIS session while idle (same claim-under-lock as + # kanban dispatch). An active non-parked /goal owns the idle boundary and defers it. if _now - _last_loop_poll >= _LOOP_POLL_SECONDS: _last_loop_poll = _now try: @@ -497,7 +497,6 @@ def _notification_poller_loop(stop_event: threading.Event, sid: str, session: di except Exception: continue _notif_handle_event(sid, session, evt, _emitted, process_registry, format_process_notification, None) - # 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 = [] @@ -529,11 +528,17 @@ def _async_delegation_display_metadata(evt: dict) -> dict: return metadata -def _wire_agent_terminal_output() -> None: - """Idempotently route background-process output (`agent.terminal.output` chunks) and - `process_registry.request_close_terminal` (`terminal.close`, drops a tab without killing the - process) to the window owning the process (its gateway session), keyed by process id. - `_emit` is `_stdout_lock`-guarded, so the registry's reader threads may call it.""" +_desktop_ui_wired = False + + +def _wire_desktop_sinks() -> None: + """Idempotently wire process-registry and desktop-tool sinks to renderer events. Background-process + output (`agent.terminal.output` chunks) and `process_registry.request_close_terminal` + (`terminal.close`, drops a tab without killing the process) route to the window owning the process + (its gateway session), keyed by process id; desktop-only tools (open_preview, close_preview, + focus_pane) hand back the turn's ``HERMES_UI_SESSION_ID`` as ``sid`` so the event routes to the + window that asked. `_emit` is `_stdout_lock`-guarded, so registry reader / tool threads may call it.""" + global _desktop_ui_wired from tools.process_registry import process_registry def _owner_sid_for_process(session) -> str: @@ -541,52 +546,35 @@ def _wire_agent_terminal_output() -> None: if not session_key: return "" with _sessions_lock: - for sid, tui_session in _sessions.items(): - if str(tui_session.get("session_key") or "") == session_key: - return sid - return "" - - 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. - _emit("terminal.close", _owner_sid_for_process(session) if session is not None else "", {"process_id": process_id}) + return next((sid for sid, s in _sessions.items() if str(s.get("session_key") or "") == session_key), "") 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 + # session may be None (process already finished/pruned) — the tab can still linger and be closed. + process_registry.on_close = lambda session, process_id: _emit( + "terminal.close", _owner_sid_for_process(session) if session is not None else "", {"process_id": process_id} + ) + if not _desktop_ui_wired: + try: + from tools import desktop_ui + except Exception: + return + desktop_ui.set_emitter(lambda sid, event, payload: _emit(event, sid, payload)) + _desktop_ui_wired = True -_desktop_ui_wired = False - - -def _wire_desktop_ui() -> None: - """Bridge desktop-only tools (open_preview, close_preview, focus_pane) to renderer events. - Idempotent. The tool hands back the turn's ``HERMES_UI_SESSION_ID`` as ``sid`` so the event - routes to the window that asked (``_emit`` is ``_stdout_lock``-guarded; tool thread may call it).""" - global _desktop_ui_wired - if _desktop_ui_wired: - return - try: - from tools import desktop_ui - except Exception: - return - desktop_ui.set_emitter(lambda sid, event, payload: _emit(event, sid, payload)) - _desktop_ui_wired = True - - -# (stop_event, thread) for every poller started in this process, pruned of dead threads on each -# spawn. Test teardowns reap leaked pollers through it: an unjoined poller steals events off the -# process-global completion_queue mid-assertion in a LATER test. +# (stop_event, thread) for every poller started in this process, pruned of dead threads on each spawn. +# Test teardowns reap leaked pollers through it: an unjoined poller steals events off the process-global +# completion_queue mid-assertion in a LATER test. _notification_pollers: list = [] def _start_notification_poller(sid: str, session: dict) -> threading.Event: """Start the background notification poller for a TUI session (thread name is greppable).""" - _wire_agent_terminal_output() - _wire_desktop_ui() + _wire_desktop_sinks() stop = threading.Event() t = threading.Thread( target=_notification_poller_loop, args=(stop, sid, session), daemon=True, name=f"tui-notif-poller-{sid}" @@ -607,10 +595,10 @@ def _hud_surface_note(session: dict) -> str: def _prepend_note(run_message: Any, note: str) -> Any: - """Prefix a per-turn note onto the MODEL INPUT, leaving the prompt alone. Everything the model - must know about the turn that the user did not type (interrupted reply, reactions, surface) - arrives this way; persist_user_message keeps the clean prompt, so no scaffolding reaches the - transcript, and annotating the NEW turn never rewrites a sent message — cached prefix survives.""" + """Prefix a per-turn note onto the MODEL INPUT, leaving the prompt alone. Everything the model must + know about the turn that the user did not type (interrupted reply, reactions, surface) arrives this + way; persist_user_message keeps the clean prompt, so no scaffolding reaches the transcript, and + annotating the NEW turn never rewrites a sent message — cached prefix survives.""" if not note: return run_message if isinstance(run_message, str): diff --git a/tui_gateway/synthetic_turn.py b/tui_gateway/synthetic_turn.py index 6c1fd45c59..93231557e8 100644 --- a/tui_gateway/synthetic_turn.py +++ b/tui_gateway/synthetic_turn.py @@ -14,6 +14,7 @@ rides in the prompt text as a JSON object; any other prompt falls back to env / from __future__ import annotations +import contextlib import json import os import threading @@ -23,6 +24,21 @@ from typing import Any, Callable, Optional from tui_gateway._env import env_float as _env_float, env_int as _env_int +# Per-turn intensity spec: (key, caster, default source). ``chunk`` is pure-Python ops per +# interrupt-check chunk (small enough that an interrupt lands within ms, large enough to stay hot on +# the GIL); ``delta_interval_s`` is the streamed-delta cadence (each delta is a loop wakeup +# marshalling a frame); ``tokens_per_delta`` drives the 100K+-token heavy-turn proxy; ``sleep_s`` is +# an 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). +_SPEC_FIELDS = ( + ("duration_s", float, lambda: _env_float("HERMES_ISO_CERTIFY_DURATION_S", 8.0)), + ("chunk", int, lambda: _env_int("HERMES_ISO_CERTIFY_CHUNK", 20_000)), + ("delta_interval_s", float, lambda: _env_float("HERMES_ISO_CERTIFY_DELTA_S", 0.05)), + ("tokens_per_delta", int, lambda: _env_int("HERMES_ISO_CERTIFY_TPD", 512)), + ("sleep_s", float, lambda: 0.0), +) + + def synth_turn_armed() -> bool: """True when the synthetic-turn test seam is armed via env.""" return os.environ.get("HERMES_ISO_CERTIFY_SYNTH_TURN") == "1" @@ -74,26 +90,10 @@ class SyntheticHeavyAgent: def _parse_spec(message: Any) -> dict[str, Any]: spec: dict[str, Any] = {} if isinstance(message, str) and message.strip().startswith("{"): - try: + with contextlib.suppress(ValueError, TypeError): parsed = json.loads(message.strip()) - except (ValueError, TypeError): - parsed = None - if isinstance(parsed, dict): - spec = parsed - 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. - "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. - "sleep_s": float(spec.get("sleep_s", 0.0)), - } + spec = parsed if isinstance(parsed, dict) else {} + return {key: cast(spec.get(key, default())) for key, cast, default in _SPEC_FIELDS} def run_conversation( self, diff --git a/tui_gateway/turn_marker.py b/tui_gateway/turn_marker.py index ae938120a9..7a65b23baa 100644 --- a/tui_gateway/turn_marker.py +++ b/tui_gateway/turn_marker.py @@ -120,14 +120,9 @@ def read_turn_marker(home: Path | str, session_key: str) -> dict[str, Any] | Non try: with _lock: entry = _load(_marker_path(home)).get(session_key) + prompt = str(entry.get("prompt") or "") if isinstance(entry, dict) else "" + if not prompt.strip(): + return None + return {"attempts": max(0, int(entry.get("attempts") or 0)), "prompt": prompt, "started_at": _started_at(entry)} except Exception: return None - if not isinstance(entry, dict): - return None - prompt = str(entry.get("prompt") or "") - if not prompt.strip(): - return None - try: - return {"attempts": max(0, int(entry.get("attempts") or 0)), "prompt": prompt, "started_at": _started_at(entry)} - except (TypeError, ValueError): - return None