From f80e2c77c1190aafb575a9061bea6921b85a2566 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 23:14:19 -0700 Subject: [PATCH] refactor(tui_gateway): split prompt turn/submit god functions into phase helpers, unify side-agent + followup dispatch --- tui_gateway/methods_prompt.py | 917 +++++++++++++----------------- tui_gateway/prompt_attachments.py | 111 ++-- tui_gateway/prompt_turn.py | 837 ++++++++++++++------------- 3 files changed, 863 insertions(+), 1002 deletions(-) diff --git a/tui_gateway/methods_prompt.py b/tui_gateway/methods_prompt.py index 0419bcaa08..b07b8f9f6a 100644 --- a/tui_gateway/methods_prompt.py +++ b/tui_gateway/methods_prompt.py @@ -14,12 +14,12 @@ _profile_scoped = _registry.profile_scoped _STALE_TARGET_MSG = "target user message is no longer in session history" +_GROUP_PROBE_FAILED_MSG = "Could not verify this group. Try again after the gateway recovers." def _history_user_indices(history: list) -> list: """Indices of canonical live-user turns, including composite carriers.""" from agent.context_compressor import user_originated_turn_view - return [i for i, m in enumerate(history) if user_originated_turn_view(m) is not None] @@ -28,10 +28,8 @@ def _message_row_id(msg: dict): raw = msg.get("_row_id") if raw is None: raw = msg.get("row_id") - if raw is None: - return None try: - return int(raw) + return None if raw is None else int(raw) except (TypeError, ValueError): return None @@ -51,7 +49,6 @@ def _mem_db_pair_agrees(mem, db_msg) -> bool: if mem.get("role") == "user": from agent.context_compressor import user_originated_turn_view from agent.memory_manager import sanitize_context - mem_view = user_originated_turn_view(mem) db_view = user_originated_turn_view(db_msg) if (mem_view is None) != (db_view is None): @@ -61,14 +58,9 @@ def _mem_db_pair_agrees(mem, db_msg) -> bool: mem_content = mem_view.get("content") db_content = db_view.get("content") if isinstance(mem_content, str) and isinstance(db_content, str): - if sanitize_context(mem_content).strip() != sanitize_context( - db_content - ).strip(): - return False + return sanitize_context(mem_content).strip() == sanitize_context(db_content).strip() return True - if bool(mem.get("display_kind")) != bool(db_msg.get("display_kind")): - return False - return True + return bool(mem.get("display_kind")) == bool(db_msg.get("display_kind")) def _find_user_turn_by_row_id(history: list, target_row_id: int): @@ -80,9 +72,7 @@ def _find_user_turn_by_row_id(history: list, target_row_id: int): def _load_durable_truncation_history( - session: dict, - fallback_sid: str = "", - repair_alternation: bool = True, + session: dict, fallback_sid: str = "", repair_alternation: bool = True ): """Load the durable live-replay transcript, or None when it cannot be proven safe.""" session_key = str(session.get("session_key") or fallback_sid or "") @@ -116,19 +106,16 @@ def _resolve_truncate_row_id(session: dict, history: list, target_row_id: int): hit = _find_user_turn_by_row_id(history, target_row_id) if hit is not None: return hit - db_history = _load_durable_truncation_history(session) if db_history is None: return None - # Heal missing stamps only when EVERY pair agrees (all-or-nothing). Equal # length alone is not alignment: the durable copy is alternation-repaired # (may merge/drop rows) while the live list is not and can carry # optimistic/marker rows; a stamp on a misaligned pair is sticky and # re-aims every later rewind at the wrong durable row. if len(db_history) == len(history) and all( - _mem_db_pair_agrees(mem, db_msg) - for mem, db_msg in zip(history, db_history) + _mem_db_pair_agrees(mem, db_msg) for mem, db_msg in zip(history, db_history) ): for mem, db_msg in zip(history, db_history): db_rid = _message_row_id(db_msg) if isinstance(db_msg, dict) else None @@ -137,7 +124,6 @@ def _resolve_truncate_row_id(session: dict, history: list, target_row_id: int): hit = _find_user_turn_by_row_id(history, target_row_id) if hit is not None: return hit - db_hit = _find_user_turn_by_row_id(db_history, target_row_id) if db_hit is None: return None @@ -157,17 +143,14 @@ def _resolve_truncate_row_id(session: dict, history: list, target_row_id: int): def _coerce_truncate_int(rid, value, param_name="truncate_before_user_ordinal"): """``(int_value, error_response)`` for a client integer param. bool is refused like any non-integer: JSON ``true`` would int() to 1 and aim at the wrong turn.""" - if isinstance(value, bool): - return None, _err(rid, 4004, f"{param_name} must be an integer") - try: - return int(value), None - except (TypeError, ValueError): - return None, _err(rid, 4004, f"{param_name} must be an integer") + if not isinstance(value, bool): + with contextlib.suppress(TypeError, ValueError): + return int(value), None + return None, _err(rid, 4004, f"{param_name} must be an integer") def _reconcile_client_ordinal( - rid, sid, client_ordinal, msg_ordinal, param_name, target_repr, - prefix_user_count=0, + rid, sid, client_ordinal, msg_ordinal, param_name, target_repr, prefix_user_count=0, ): """Cross-check a client ordinal against a resolved durable target. @@ -183,25 +166,18 @@ def _reconcile_client_ordinal( ordinal, err = _coerce_truncate_int(rid, client_ordinal) if err is not None: return None, err - if ordinal == msg_ordinal: - return msg_ordinal, None - if prefix_user_count > 0 and ordinal == msg_ordinal + prefix_user_count: + if ordinal == msg_ordinal or ( + prefix_user_count > 0 and ordinal == msg_ordinal + prefix_user_count + ): return msg_ordinal, None logger.warning( "prompt.submit: REFUSED truncation due to ordinal mismatch for session %s " "(ordinal=%d, %s_ordinal=%d, %s=%s, prefix_user_count=%d). " "Stale truncate_before_user_ordinal detected.", - sid, - ordinal, - param_name, - msg_ordinal, - param_name, - target_repr, - prefix_user_count, + sid, ordinal, param_name, msg_ordinal, param_name, target_repr, prefix_user_count, ) return None, _err( - rid, - 4030, + rid, 4030, f"truncate_before_user_ordinal ({ordinal}) does not match " f"{param_name} target turn ({msg_ordinal})", ) @@ -214,14 +190,12 @@ def _pending_reaction_notes(session: dict) -> str: session_key = str(session.get("session_key") or "") if not session_key: return "" - try: display = _load_cfg().get("display") if not (isinstance(display, dict) and bool(display.get("message_reactions", False))): return "" except Exception: return "" - try: with _session_db(session) as db: if db is None: @@ -230,12 +204,8 @@ def _pending_reaction_notes(session: dict) -> str: except Exception: logger.debug("Failed to read pending reactions", exc_info=True) return "" - - if not pending: - return "" - notes = [] - for entry in pending: + for entry in pending or (): snippet = (entry.get("text") or "").strip().replace("\n", " ") if len(snippet) > 120: snippet = snippet[:120] + "…" @@ -246,7 +216,6 @@ def _pending_reaction_notes(session: dict) -> str: else: # Attachment-only / tool-call-only rows: no quote beats an empty quote. notes.append(f"[The user reacted {emoji} to {whose} earlier message]") - return "\n".join(notes) @@ -263,7 +232,6 @@ def _typed_stop_phrase_response(rid, text): return None try: from tools.voice_mode import is_voice_stop_phrase - typed_stop = is_voice_stop_phrase(text) except Exception: typed_stop = False @@ -282,12 +250,15 @@ def _hosted_submit_error(rid, session, hosted_task, hosted_terminal_callback): """Validate the hosted-room turn proof carried by an internal submit.""" if session.get("source") != "bot_room": return _err(rid, 4120, "hosted room turns require a bot_room session") - if not isinstance(hosted_task, dict) or not callable(hosted_terminal_callback): - return _err(rid, 4120, "invalid hosted room turn proof") - if set(hosted_task) != _HOSTED_TASK_FIELDS or not all( - isinstance(hosted_task.get(field), str) and hosted_task[field] - for field in _HOSTED_TASK_FIELDS - {"execution_generation"} - ) or not isinstance(hosted_task.get("execution_generation"), int): + if ( + not isinstance(hosted_task, dict) or not callable(hosted_terminal_callback) + or set(hosted_task) != _HOSTED_TASK_FIELDS + or not all( + isinstance(hosted_task.get(field), str) and hosted_task[field] + for field in _HOSTED_TASK_FIELDS - {"execution_generation"} + ) + or not isinstance(hosted_task.get("execution_generation"), int) + ): return _err(rid, 4120, "invalid hosted room turn proof") return None @@ -304,25 +275,16 @@ def _legacy_group_fence_error(rid, session, params): return None try: from gateway.hosted_rooms import ( - HostedRoomError, - RoomProbeUnavailableError, - default_db_path, - probe_hosted_room, + HostedRoomError, RoomProbeUnavailableError, default_db_path, probe_hosted_room, probe_peer_room_reservation, ) - hosted = probe_hosted_room(default_db_path(), room_id=room_id) peer = False if not hosted: from hermes_constants import named_profile_home - session_profile_home = named_profile_home(str(session.get("profile_home") or "")) requested_profile = ( - ( - session_profile_home.name - if session_profile_home is not None - else "" - ) + (session_profile_home.name if session_profile_home is not None else "") or str(params.get("profile") or "").strip() or str(_current_profile_name() or "default").strip() ) @@ -330,74 +292,96 @@ def _legacy_group_fence_error(rid, session, params): default_db_path(), room_id=room_id, target_profile=requested_profile, ) except RoomProbeUnavailableError: - return _err(rid, 5122, "Could not verify this group. Try again after the gateway recovers.") + return _err(rid, 5122, _GROUP_PROBE_FAILED_MSG) except HostedRoomError: # Legacy Desktop sessions used the display name after "Group: "; those # names are not hosted room ids. return None except Exception: - return _err(rid, 5122, "Could not verify this group. Try again after the gateway recovers.") + return _err(rid, 5122, _GROUP_PROBE_FAILED_MSG) if hosted or peer: + owner = "its gateway" if hosted else "its home host" return _err( - rid, - 4122, - ( - "This room is managed by its gateway. " - if hosted - else "This room is managed by its home host. " - ) - + "Update Hermes Desktop to continue it.", + rid, 4122, f"This room is managed by {owner}. Update Hermes Desktop to continue it.", ) return None +def _parse_truncation_params(rid, sid, session, params, history): + """Coerce + admit the truncation params; ``(target_row_id, client_ordinal, err)``. + + Refusal precedence: malformed params (4004) -> unconfirmed (4029, checked + BEFORE target resolution so a leaked-state request never pays the durable + read or heal-stamps live dicts). An ordinal/id alone is not consent: a + leftover ordinal on an ORDINARY submit is field-for-field indistinguishable + from a real rewind, and the cut is a destructive replace_messages(). + """ + truncate_user_ordinal = params.get("truncate_before_user_ordinal") + truncate_row_id = params.get("truncate_before_row_id") + target_row_id = client_ordinal = None + if truncate_row_id is not None: + target_row_id, err = _coerce_truncate_int(rid, truncate_row_id, "truncate_before_row_id") + if err is not None: + return None, None, err + if truncate_user_ordinal is not None: + client_ordinal, err = _coerce_truncate_int(rid, truncate_user_ordinal) + if err is not None: + return None, None, err + if is_truthy_value(params.get("confirm_truncate")): + return target_row_id, client_ordinal, None + logger.warning( + "prompt.submit: REFUSED unconfirmed truncation of session %s " + "(%d messages held; ordinal=%s, row_id=%s, message_id=%s). " + "The client attached truncation parameters without " + "confirm_truncate — likely stale truncation parameters on " + "an ordinary submit.", + sid, len(history), client_ordinal, target_row_id, params.get("truncate_before_message_id"), + ) + return None, None, _err( + rid, 4029, + "truncation parameters require confirm_truncate=true; " + "an ordinary prompt.submit must not drop session history " + "(update your Hermes client if a rewind was intended)", + ) + + +def _ordinal_only_truncation_error(rid, sid, session, history, user_indices, client_ordinal): + """4004 refusal when an ordinal-only cut targets a durable session, else None. + + Durability is a state.db property, not an optional annotation on the live + copy (resume paths historically omitted _row_id stamps). If the durable + state cannot be read, fail closed too: absence of proof is not proof of an + ephemeral conversation. + """ + has_stamped_user = any(_message_row_id(history[h_idx]) is not None for h_idx in user_indices) + durable_history = [] if has_stamped_user else _load_durable_truncation_history(session, sid) + if not (has_stamped_user or durable_history is None or durable_history): + return None + logger.warning( + "prompt.submit: REFUSED ordinal-only truncation of durable " + "session %s (ordinal=%d); truncate_before_row_id required", + sid, client_ordinal, + ) + return _err( + rid, 4004, + "ordinal-only truncation is unsafe for durable session history; " + "include truncate_before_row_id", + ) + + def _resolve_truncation_ordinal(rid, sid, session, params, history): """Resolve the truncation target to ``(ordinal, cut_index, err)``. - Refusal precedence: malformed params (4004) → unconfirmed (4029, checked - BEFORE target resolution so a leaked-state request never pays the durable - read or heal-stamps live dicts) → unresolvable target (4018, fail closed — - never degrade a missing row_id/message_id into an ordinal cut) → ordinal - drift (4030) → ordinal-only on a durable session (4004). + After ``_parse_truncation_params``: unresolvable target (4018, fail closed — + never degrade a missing row_id/message_id into an ordinal cut) -> ordinal + drift (4030) -> ordinal-only on a durable session (4004). """ - truncate_user_ordinal = params.get("truncate_before_user_ordinal") + target_row_id, client_ordinal, err = _parse_truncation_params( + rid, sid, session, params, history + ) + if err is not None: + return None, None, err truncate_message_id = params.get("truncate_before_message_id") - truncate_row_id = params.get("truncate_before_row_id") - - target_row_id = None - if truncate_row_id is not None: - target_row_id, err = _coerce_truncate_int(rid, truncate_row_id, "truncate_before_row_id") - if err is not None: - return None, None, err - client_ordinal = None - if truncate_user_ordinal is not None: - client_ordinal, err = _coerce_truncate_int(rid, truncate_user_ordinal) - if err is not None: - return None, None, err - - # An ordinal/id alone is not consent: a leftover ordinal on an ORDINARY - # submit is field-for-field indistinguishable from a real rewind, and the - # cut is a destructive replace_messages(). Only the client knows. - if not is_truthy_value(params.get("confirm_truncate")): - logger.warning( - "prompt.submit: REFUSED unconfirmed truncation of session %s " - "(%d messages held; ordinal=%s, row_id=%s, message_id=%s). " - "The client attached truncation parameters without " - "confirm_truncate — likely stale truncation parameters on " - "an ordinary submit.", - sid, - len(history), - client_ordinal, - target_row_id, - truncate_message_id, - ) - return None, None, _err( - rid, - 4029, - "truncation parameters require confirm_truncate=true; " - "an ordinary prompt.submit must not drop session history " - "(update your Hermes client if a rewind was intended)", - ) # Client ordinals count the full displayed lineage; after compression the # tip segment is session["history"] and the ancestors live in # display_history_prefix. Count the ancestor user turns once so client and @@ -416,78 +400,46 @@ def _resolve_truncation_ordinal(rid, sid, session, params, history): "segment_ordinal": segment, "prefix_user_count": prefix_user_count, }) - if target_row_id is not None: - found_match = _resolve_truncate_row_id(session, history, target_row_id) + if target_row_id is not None or truncate_message_id is not None: + if target_row_id is not None: + param_name, target_repr = "truncate_before_row_id", target_row_id + found_match = _resolve_truncate_row_id(session, history, target_row_id) + not_found = "target row_id %d not found for session %s (in-memory + durable)" + else: + param_name = "truncate_before_message_id" + target_repr = msg_id_str = str(truncate_message_id) + found_match = next( + ( + (u_ord, h_idx) for u_ord, h_idx in enumerate(user_indices) + if history[h_idx].get("id") == msg_id_str + or history[h_idx].get("message_id") == msg_id_str + ), + None, + ) + not_found = "target message_id %s not found in history for session %s" if found_match is None: logger.warning( - "prompt.submit: target row_id %d not found for session %s " - "(in-memory + durable); refusing truncation without fallback", - target_row_id, - sid, + "prompt.submit: " + not_found + "; refusing truncation without fallback", + target_repr, sid, ) return _stale() ordinal, err = _reconcile_client_ordinal( - rid, sid, client_ordinal, found_match[0], "truncate_before_row_id", target_row_id, - prefix_user_count=prefix_user_count, - ) - if err is not None: - return None, None, err - elif truncate_message_id is not None: - msg_id_str = str(truncate_message_id) - found_match = next( - ( - (u_ord, h_idx) - for u_ord, h_idx in enumerate(user_indices) - if history[h_idx].get("id") == msg_id_str - or history[h_idx].get("message_id") == msg_id_str - ), - None, - ) - if found_match is None: - logger.warning( - "prompt.submit: target message_id %s not found in history " - "for session %s; refusing truncation without fallback", - msg_id_str, - sid, - ) - return _stale() - ordinal, err = _reconcile_client_ordinal( - rid, sid, client_ordinal, found_match[0], "truncate_before_message_id", msg_id_str, + rid, sid, client_ordinal, found_match[0], param_name, target_repr, prefix_user_count=prefix_user_count, ) if err is not None: return None, None, err else: - segment_ordinal = client_ordinal - prefix_user_count - if segment_ordinal < 0 or segment_ordinal >= len(user_indices): + ordinal = client_ordinal - prefix_user_count + if ordinal < 0 or ordinal >= len(user_indices): return _stale() - # Durability is a state.db property, not an optional annotation on the - # live copy (resume paths historically omitted _row_id stamps). If the - # durable state cannot be read, fail closed too: absence of proof is - # not proof of an ephemeral conversation. - has_stamped_user = any( - _message_row_id(history[h_idx]) is not None for h_idx in user_indices + err = _ordinal_only_truncation_error( + rid, sid, session, history, user_indices, client_ordinal ) - durable_history = ( - [] if has_stamped_user else _load_durable_truncation_history(session, sid) - ) - if has_stamped_user or durable_history is None or durable_history: - logger.warning( - "prompt.submit: REFUSED ordinal-only truncation of durable " - "session %s (ordinal=%d); truncate_before_row_id required", - sid, - client_ordinal, - ) - return None, None, _err( - rid, - 4004, - "ordinal-only truncation is unsafe for durable session history; " - "include truncate_before_row_id", - ) - ordinal = segment_ordinal - + if err is not None: + return None, None, err # Reject out-of-range on BOTH ends: a negative ordinal would hit Python's - # negative indexing (user_indices[-1] → the LAST user turn) and persist the loss. + # negative indexing (user_indices[-1] -> the LAST user turn) and persist the loss. if ordinal < 0 or ordinal >= len(user_indices): return _stale(resolved_ordinal=ordinal) return ordinal, user_indices[ordinal], None @@ -509,8 +461,7 @@ def _persist_truncation(rid, sid, session, history, truncated, ordinal, requeste Returns ``(err, survivor_user_row_ids, survivor_row_id_map)``. """ - survivor_user_row_ids = None - survivor_row_id_map = None + survivor_user_row_ids = survivor_row_id_map = None with _session_db(session) as db: if db is not None: try: @@ -546,10 +497,7 @@ def _persist_truncation(rid, sid, session, history, truncated, ordinal, requeste "prompt.submit: replace_messages failed for session %s " "(ordinal=%d); refusing turn so memory and DB stay " "aligned: %s", - sid, - ordinal, - exc, - exc_info=True, + sid, ordinal, exc, exc_info=True, ) return _err(rid, 5008, f"failed to persist history truncation: {exc}"), None, None # replace_messages re-inserted the survivors as NEW rows and stamped @@ -565,16 +513,12 @@ def _persist_truncation(rid, sid, session, history, truncated, ordinal, requeste survivor_row_id_map = { str(old_row_id): new_row_id for old_row_id, new_row_id in zip( - old_survivor_row_ids, - (_message_row_id(message) for message in truncated), + old_survivor_row_ids, (_message_row_id(message) for message in truncated), ) - if isinstance(old_row_id, int) - and isinstance(new_row_id, int) + if isinstance(old_row_id, int) and isinstance(new_row_id, int) and old_row_id in requested_rebind_ids } - for dropped_row_id in requested_rebind_ids.intersection( - old_active_row_ids - ): + for dropped_row_id in requested_rebind_ids.intersection(old_active_row_ids): survivor_row_id_map.setdefault(str(dropped_row_id), None) return None, survivor_user_row_ids, survivor_row_id_map @@ -588,26 +532,18 @@ def _truncate_history_for_submit(rid, sid, session, params, requested_rebind_ids if err is not None: return err, None, None from agent.context_compressor import history_before_user_originated_turn - truncated, _live_view = history_before_user_originated_turn(history, cut_index) - # Second gate on top of confirm_truncate: ordinal 0 → history[:0] == [] and + # Second gate on top of confirm_truncate: ordinal 0 -> history[:0] == [] and # replace_messages() DELETEs every durable row. Wiping the whole transcript # needs its own opt-in (legitimate restore/regenerate of the first turn). - if ( - not truncated - and history - and not is_truthy_value(params.get("confirm_empty_truncate")) - ): + if not truncated and history and not is_truthy_value(params.get("confirm_empty_truncate")): logger.warning( "prompt.submit: REFUSED empty truncation of session %s " "(%d messages would be wiped; ordinal=%d).", - sid, - len(history), - ordinal, + sid, len(history), ordinal, ) return _err( - rid, - 4028, + rid, 4028, "truncation would erase the entire session transcript; " "resubmit with confirm_empty_truncate=true if this is intended", ), None, None @@ -644,8 +580,7 @@ def _persist_session_row_for_submit(rid, session): try: if _ensure_session_db_row(session) is False: return _err( - rid, - 5072, + rid, 5072, "session storage unavailable: " f"{_db_error or 'state.db could not be opened'} — the message " "was not saved; repair state.db and try again", @@ -653,15 +588,13 @@ def _persist_session_row_for_submit(rid, session): _persist_branch_seed(session) except Exception as exc: from hermes_state import is_disk_full_error - with session["history_lock"]: session["running"] = False session["last_active"] = time.time() _clear_inflight_turn(session) if is_disk_full_error(exc): return _err( - rid, - 5070, + rid, 5070, "disk full: session storage could not be written — free some disk space and try again", ) logger.warning("prompt.submit: session persist failed: %s", exc, exc_info=True) @@ -677,9 +610,7 @@ def _run_after_agent_ready(rid, sid, session, text, display_kind, hosted_termina # Terminal frame + retained snapshot (not a bare "error" event): if the # client is disconnected, the snapshot is the only way resume shows this. _emit_terminal_turn_error( - sid, - session, - (err.get("error") or {}).get("message", "agent initialization failed"), + sid, session, (err.get("error") or {}).get("message", "agent initialization failed"), # Construction never reached the provider: local-runtime failure. error_surface={"layer": "runtime", "code": "agent_init_failed", "retryable": True}, ) @@ -694,15 +625,11 @@ def _run_after_agent_ready(rid, sid, session, text, display_kind, hosted_termina _clear_inflight_turn(session) # Without this emit the turn vanishes silently: the client saw # {"status": "streaming"} but never gets message.start or error. - _emit( - "error", - sid, - { - "message": "Turn cancelled before the agent was ready" - if session.get("_turn_cancel_requested") - else "Session no longer running before the agent was ready" - }, - ) + _emit("error", sid, { + "message": "Turn cancelled before the agent was ready" + if session.get("_turn_cancel_requested") + else "Session no longer running before the agent was ready" + }) return _run_prompt_submit( rid, sid, session, text, display_kind=display_kind, @@ -710,10 +637,71 @@ def _run_after_agent_ready(rid, sid, session, text, display_kind, hosted_termina ) +_TRUNCATION_PARAMS = ( + "truncate_before_user_ordinal", "truncate_before_row_id", "truncate_before_message_id", +) + + +def _claim_submit_slot(rid, sid, session, text, params, transport, internal_hosted_submit): + """Claim the turn against a possibly-running session; returns an RPC reply to + return early (busy/queued) or None once ``running`` is observed False. + + A mid-turn prompt is queued (and by default interrupts the live turn) instead + of rejected. The provider interrupt must happen after ``history_lock`` is + released: a non-interruptible tool may hold it. If the old turn finished + between the two lock acquisitions, retry the claim rather than strand this + prompt in a queue whose drain already ran. + """ + while True: + with session["history_lock"]: + if not session.get("running"): + return None + if internal_hosted_submit: + return _err(rid, 4091, "hosted room member session is busy") + busy_transport = transport or session.get("transport") + busy_response = _handle_busy_submit( + rid, sid, session, text, busy_transport, queued=bool(params.get("queued")), + ) + if busy_response is not None: + return busy_response + + +def _lock_in_submit_turn( + rid, sid, session, text, params, has_truncation, requested_rebind_ids, hosted_task +): + """Under ``history_lock``: refuse watch-child races and malformed truncation, + apply the cut, then mark the turn running + in flight. + Returns ``(err, survivor_user_row_ids, survivor_row_id_map)``.""" + survivor_user_row_ids = survivor_row_id_map = None + with session["history_lock"]: + # A watch session's run lives in the PARENT turn, so its own running flag + # is False; typing mid-run would build a second agent racing the child + # on the same stored session. After the run completes, submitting is fine. + if session.get("lazy") and _child_run_active(str(session.get("session_key") or "")): + return _err(rid, 4009, "subagent still running — wait for it to finish"), None, None + if is_truthy_value(params.get("confirm_truncate")) and not has_truncation: + return _err( + rid, 4004, + "confirm_truncate requires truncate_before_user_ordinal, truncate_before_message_id, or truncate_before_row_id", + ), None, None + if has_truncation: + err, survivor_user_row_ids, survivor_row_id_map = _truncate_history_for_submit( + rid, sid, session, params, requested_rebind_ids + ) + if err is not None: + return err, None, None + session["running"] = True + session["_turn_cancel_requested"] = False + session["last_active"] = time.time() + if hosted_task is not None: + session["_hosted_room_task"] = dict(hosted_task) + _start_inflight_turn(session, text) + return None, survivor_user_row_ids, survivor_row_id_map + + @method("prompt.submit") def _(rid, params: dict) -> dict: from hermes_cli.input_sanitize import sanitize_user_prompt_text - sid = params.get("session_id", "") raw_text = params.get("text", "") text = sanitize_user_prompt_text(raw_text) if isinstance(raw_text, str) else raw_text @@ -726,7 +714,6 @@ def _(rid, params: dict) -> dict: # Client-side barge-in (desktop VAD / typing over playback): latch it so # this turn's model message carries the interruption note. from tools.tts_streaming import mark_speech_interrupted - mark_speech_interrupted() session, err = _sess_nowait(params, rid) if err: @@ -750,79 +737,34 @@ def _(rid, params: dict) -> dict: # Rewritten on every submit: one session can be driven from the app window # and the HUD in turn, and a stale "hud" misinforms the model. session["client_surface"] = "hud" if params.get("surface") == "hud" else "" - has_truncation = any( - params.get(k) is not None - for k in ("truncate_before_user_ordinal", "truncate_before_row_id", "truncate_before_message_id") - ) + has_truncation = any(params.get(k) is not None for k in _TRUNCATION_PARAMS) if has_truncation and isinstance(text, str): # A rewind replays what the transcript shows; a skill turn shows its # invocation, so re-expand it or `/work fix it` sends nine literal chars. text = _expand_skill_invocation_for_replay(text, str(session.get("session_key") or "")) - isolation_cfg = _load_dashboard_process_isolation_config() - turn_isolation = _session_uses_compute_host(session, isolation_cfg) + turn_isolation = _session_uses_compute_host(session, _load_dashboard_process_isolation_config()) if internal_hosted_submit and turn_isolation: return _err(rid, 4121, "hosted room turns do not support isolated compute workers yet") # Re-bind to the current client transport so streaming stays on the active # websocket even if a disconnect/fallback moved the session to stdio. if (t := current_transport()) is not None: session["transport"] = t - while True: - busy_transport = None - with session["history_lock"]: - if session.get("running"): - if internal_hosted_submit: - return _err(rid, 4091, "hosted room member session is busy") - # Queue a mid-turn prompt (and by default interrupt the live turn) - # instead of rejecting. The provider interrupt must happen after - # this lock is released: a non-interruptible tool may hold it. - busy_transport = t or session.get("transport") - else: - break - busy_response = _handle_busy_submit( - rid, sid, session, text, busy_transport, queued=bool(params.get("queued")), - ) - if busy_response is not None: - return busy_response - # The old turn finished between the two lock acquisitions: retry the - # claim rather than strand this prompt in a queue whose drain already ran. - - survivor_user_row_ids = None - survivor_row_id_map = None + busy = _claim_submit_slot(rid, sid, session, text, params, t, internal_hosted_submit) + if busy is not None: + return busy raw_rebind_ids = params.get("rebind_survivor_row_ids") requested_rebind_ids = ( { - row_id - for row_id in raw_rebind_ids + row_id for row_id in raw_rebind_ids if isinstance(row_id, int) and not isinstance(row_id, bool) } - if isinstance(raw_rebind_ids, list) - else None + if isinstance(raw_rebind_ids, list) else None ) - with session["history_lock"]: - # A watch session's run lives in the PARENT turn, so its own running flag - # is False; typing mid-run would build a second agent racing the child - # on the same stored session. After the run completes, submitting is fine. - if session.get("lazy") and _child_run_active(str(session.get("session_key") or "")): - return _err(rid, 4009, "subagent still running — wait for it to finish") - if is_truthy_value(params.get("confirm_truncate")) and not has_truncation: - return _err( - rid, - 4004, - "confirm_truncate requires truncate_before_user_ordinal, truncate_before_message_id, or truncate_before_row_id", - ) - if has_truncation: - err, survivor_user_row_ids, survivor_row_id_map = _truncate_history_for_submit( - rid, sid, session, params, requested_rebind_ids - ) - if err is not None: - return err - session["running"] = True - session["_turn_cancel_requested"] = False - session["last_active"] = time.time() - if internal_hosted_submit: - session["_hosted_room_task"] = dict(hosted_task) - _start_inflight_turn(session, text) - + err, survivor_user_row_ids, survivor_row_id_map = _lock_in_submit_turn( + rid, sid, session, text, params, has_truncation, requested_rebind_ids, hosted_task, + ) + if err is not None: + return err survivor_fields = _survivor_fields( survivor_user_row_ids, survivor_row_id_map, requested_rebind_ids ) @@ -838,16 +780,12 @@ def _(rid, params: dict) -> dict: "compute-host dispatch failed for session %s; falling back inline: %s", sid, isolated_response["error"].get("message", "unknown error"), ) - if (err := _persist_session_row_for_submit(rid, session)) is not None: return err # A completed FAILED build must not wedge the session: rebuild with fresh # provider resolution instead of replaying the cached failure forever. - if not _restart_completed_failed_agent_build( - sid, session, session.get("agent_ready") - ): + if not _restart_completed_failed_agent_build(sid, session, session.get("agent_ready")): _start_agent_build(sid, session) - run_thread = threading.Thread( target=lambda: _run_after_agent_ready( rid, sid, session, text, display_kind, hosted_terminal_callback @@ -880,24 +818,20 @@ def _(rid, params: dict) -> dict: from hermes_cli.clipboard import has_clipboard_image, save_clipboard_image except Exception as e: return _err(rid, 5027, f"clipboard unavailable: {e}") - session["image_counter"] = session.get("image_counter", 0) + 1 img_dir = _session_images_dir(session) img_dir.mkdir(parents=True, exist_ok=True) img_path = ( img_dir / f"clip_{datetime.now().strftime('%Y%m%d_%H%M%S')}_{session['image_counter']}.png" ) - # Save-first (CLI keybinding parity): more robust than a has_image() precheck. if not save_clipboard_image(img_path): session["image_counter"] = max(0, session["image_counter"] - 1) msg = ( - "Clipboard has image but extraction failed" - if has_clipboard_image() + "Clipboard has image but extraction failed" if has_clipboard_image() else "No image found in clipboard" ) return _ok(rid, {"attached": False, "message": msg}) - session.setdefault("attached_images", []).append(str(img_path)) return _ok(rid, _attached_image_result(session, img_path)) @@ -912,12 +846,8 @@ def _(rid, params: dict) -> dict: return _err(rid, 4015, "path required") try: from cli import ( - _IMAGE_EXTENSIONS, - _detect_file_drop, - _resolve_attachment_path, - _split_path_input, + _IMAGE_EXTENSIONS, _detect_file_drop, _resolve_attachment_path, _split_path_input, ) - dropped = _detect_file_drop(raw) if dropped: image_path = dropped["path"] @@ -932,8 +862,7 @@ def _(rid, params: dict) -> dict: session.setdefault("attached_images", []).append(str(image_path)) return _ok(rid, _attached_image_result( session, image_path, - remainder=remainder, - text=remainder or f"[User attached image: {image_path.name}]", + remainder=remainder, text=remainder or f"[User attached image: {image_path.name}]", )) except Exception as e: return _err(rid, 5027, str(e)) @@ -948,18 +877,15 @@ def _(rid, params: dict) -> dict: session, err = _sess_building(params, rid) if err: return err - raw_b64 = str(params.get("content_base64") or params.get("data") or "").strip() if not raw_b64: return _err(rid, 4015, "content_base64 required") - img_bytes, err = _decode_attach_payload( rid, raw_b64, mime_prefix="image/", max_bytes=_ATTACH_BYTES_MAX_BYTES, label="image", empty_msg="image is empty", ) if err is not None: return err - filename = str(params.get("filename", "") or "") ext_hint = str(params.get("ext", "") or "").strip().lower() if ext_hint and not ext_hint.startswith("."): @@ -967,20 +893,66 @@ def _(rid, params: dict) -> dict: ext = _sniff_image_ext(img_bytes, filename or (f"x{ext_hint}" if ext_hint else "")) if ext not in _allowed_image_extensions(): return _err(rid, 4016, f"unsupported image extension: {ext}") - try: img_path = _queue_attached_image(session, img_bytes, ext, prefix="upload") except Exception as e: return _err(rid, 5027, f"write failed: {e}") - return _ok(rid, _attached_image_result( session, img_path, - remainder="", - text=f"[User attached image: {img_path.name}]", - bytes=len(img_bytes), + remainder="", text=f"[User attached image: {img_path.name}]", bytes=len(img_bytes), )) +def _pdf_attach_source(rid, params, td_path, raw_path, raw_b64): + """Materialize the PDF to render: ``(pdf_path, display_name, err)``.""" + if raw_b64: + pdf_bytes, err = _decode_attach_payload( + rid, raw_b64, mime_prefix="application/pdf", max_bytes=_PDF_ATTACH_MAX_BYTES, + label="PDF", empty_msg="decoded PDF is empty", + ) + if err is not None: + return None, None, err + if pdf_bytes[:5] != b"%PDF-": + return None, None, _err(rid, 4017, "payload is not a PDF (missing %PDF- magic bytes)") + pdf_path = td_path / "input.pdf" + pdf_path.write_bytes(pdf_bytes) + return pdf_path, str(params.get("filename", "") or "uploaded.pdf"), None + try: + from cli import _resolve_attachment_path + resolved = _resolve_attachment_path(raw_path) + except Exception: + resolved = None + if resolved is None or not Path(resolved).is_file(): + return None, None, _err(rid, 4016, f"PDF not found: {raw_path}") + if Path(resolved).suffix.lower() != ".pdf": + return None, None, _err(rid, 4016, f"not a PDF: {Path(resolved).name}") + if Path(resolved).stat().st_size > _PDF_ATTACH_MAX_BYTES: + mb = _PDF_ATTACH_MAX_BYTES // (1024 * 1024) + return None, None, _err(rid, 4018, f"PDF too large; cap is {mb} MB") + return Path(resolved), Path(resolved).name, None + + +def _pdf_page_range(rid, params): + """Validate first/last page against the per-call cap: ``(first, last, err)``.""" + try: + first_page = int(params.get("first_page") or 1) + last_page_param = params.get("last_page") + last_page = int(last_page_param) if last_page_param is not None else None + except (TypeError, ValueError): + return None, None, _err(rid, 4015, "first_page/last_page must be integers") + if first_page < 1: + return None, None, _err(rid, 4015, "first_page must be >= 1") + if last_page is None: + last_page = first_page + _PDF_ATTACH_MAX_PAGES - 1 + if last_page < first_page: + return None, None, _err(rid, 4015, "last_page must be >= first_page") + if last_page - first_page + 1 > _PDF_ATTACH_MAX_PAGES: + return None, None, _err( + rid, 4019, f"page range exceeds cap of {_PDF_ATTACH_MAX_PAGES} pages per attach call" + ) + return first_page, last_page, None + + @method("pdf.attach") def _(rid, params: dict) -> dict: """Attach a PDF by rendering each page to PNG (``pdftoppm`` @150 DPI, poppler-utils; @@ -989,73 +961,28 @@ def _(rid, params: dict) -> dict: import shutil import subprocess import tempfile - session, err = _sess_building(params, rid) if err: return err - if shutil.which("pdftoppm") is None: return _err(rid, 5028, "pdftoppm not installed (poppler-utils package required)") - raw_path = str(params.get("path", "") or "").strip() raw_b64 = str(params.get("content_base64") or params.get("data") or "").strip() if not raw_path and not raw_b64: return _err(rid, 4015, "path or content_base64 required") - with tempfile.TemporaryDirectory(prefix="pdf_attach_") as td: td_path = Path(td) - if raw_b64: - pdf_bytes, err = _decode_attach_payload( - rid, raw_b64, mime_prefix="application/pdf", max_bytes=_PDF_ATTACH_MAX_BYTES, - label="PDF", empty_msg="decoded PDF is empty", - ) - if err is not None: - return err - if pdf_bytes[:5] != b"%PDF-": - return _err(rid, 4017, "payload is not a PDF (missing %PDF- magic bytes)") - pdf_path = td_path / "input.pdf" - pdf_path.write_bytes(pdf_bytes) - display_name = str(params.get("filename", "") or "uploaded.pdf") - else: - try: - from cli import _resolve_attachment_path - - resolved = _resolve_attachment_path(raw_path) - except Exception: - resolved = None - if resolved is None or not Path(resolved).is_file(): - return _err(rid, 4016, f"PDF not found: {raw_path}") - if Path(resolved).suffix.lower() != ".pdf": - return _err(rid, 4016, f"not a PDF: {Path(resolved).name}") - if Path(resolved).stat().st_size > _PDF_ATTACH_MAX_BYTES: - mb = _PDF_ATTACH_MAX_BYTES // (1024 * 1024) - return _err(rid, 4018, f"PDF too large; cap is {mb} MB") - pdf_path = Path(resolved) - display_name = pdf_path.name - - try: - first_page = int(params.get("first_page") or 1) - last_page_param = params.get("last_page") - last_page = int(last_page_param) if last_page_param is not None else None - except (TypeError, ValueError): - return _err(rid, 4015, "first_page/last_page must be integers") - - if first_page < 1: - return _err(rid, 4015, "first_page must be >= 1") - if last_page is None: - last_page = first_page + _PDF_ATTACH_MAX_PAGES - 1 - if last_page < first_page: - return _err(rid, 4015, "last_page must be >= first_page") - if last_page - first_page + 1 > _PDF_ATTACH_MAX_PAGES: - return _err(rid, 4019, f"page range exceeds cap of {_PDF_ATTACH_MAX_PAGES} pages per attach call") - - out_prefix = td_path / "page" + pdf_path, display_name, err = _pdf_attach_source(rid, params, td_path, raw_path, raw_b64) + if err is not None: + return err + first_page, last_page, err = _pdf_page_range(rid, params) + if err is not None: + return err argv = [ "pdftoppm", "-png", "-r", "150", "-f", str(first_page), "-l", str(last_page), - str(pdf_path), str(out_prefix), + str(pdf_path), str(td_path / "page"), ] from hermes_cli._subprocess_compat import windows_hide_flags - try: res = subprocess.run( argv, capture_output=True, text=True, timeout=120, stdin=subprocess.DEVNULL, @@ -1069,11 +996,9 @@ def _(rid, params: dict) -> dict: if res.returncode != 0: tail = (res.stderr or res.stdout or "").strip().splitlines()[-3:] return _err(rid, 5028, "pdftoppm failed: " + " | ".join(tail)) - rendered = sorted(td_path.glob("page-*.png")) if not rendered: return _err(rid, 5028, "pdftoppm produced no pages (corrupt PDF?)") - attached_pages = [] for src in rendered: page_num = src.stem.split("-", 1)[-1] @@ -1081,20 +1006,15 @@ def _(rid, params: dict) -> dict: page_int = int(page_num) except ValueError: page_int = first_page + len(attached_pages) - dst = _queue_attached_image(session, src.read_bytes(), ".png", prefix=f"pdf_p{page_num}") + dst = _queue_attached_image( + session, src.read_bytes(), ".png", prefix=f"pdf_p{page_num}" + ) attached_pages.append({"path": str(dst), "page": page_int, **_image_meta(dst)}) - - return _ok( - rid, - { - "attached": True, - "filename": display_name, - "pages_attached": len(attached_pages), - "pages": attached_pages, - "count": len(session["attached_images"]), - "text": f"[User attached PDF: {display_name} ({len(attached_pages)} page(s))]", - }, - ) + return _ok(rid, { + "attached": True, "filename": display_name, "pages_attached": len(attached_pages), + "pages": attached_pages, "count": len(session["attached_images"]), + "text": f"[User attached PDF: {display_name} ({len(attached_pages)} page(s))]", + }) @method("file.attach") @@ -1116,17 +1036,11 @@ def _(rid, params: dict) -> dict: session, raw_path=raw, data_url=data_url, name=name ) ref_path = _attachment_ref_path(session, stored_path) - return _ok( - rid, - { - "attached": True, - "name": stored_path.name, - "path": str(stored_path), - "ref_path": ref_path, - "ref_text": f"@file:{_format_ref_value(ref_path)}", - "uploaded": uploaded, - }, - ) + return _ok(rid, { + "attached": True, "name": stored_path.name, "path": str(stored_path), + "ref_path": ref_path, "ref_text": f"@file:{_format_ref_value(ref_path)}", + "uploaded": uploaded, + }) except Exception as e: return _err(rid, 5028, str(e)) @@ -1142,13 +1056,10 @@ def _(rid, params: dict) -> dict: images = session.setdefault("attached_images", []) before = len(images) session["attached_images"] = [path for path in images if path != raw] - return _ok( - rid, - { - "detached": len(session["attached_images"]) != before, - "count": len(session["attached_images"]), - }, - ) + return _ok(rid, { + "detached": len(session["attached_images"]) != before, + "count": len(session["attached_images"]), + }) @method("input.detect_drop") @@ -1158,40 +1069,25 @@ def _(rid, params: dict) -> dict: return err try: from cli import _detect_file_drop - raw = str(params.get("text", "") or "") dropped = _detect_file_drop(raw) if not dropped: return _ok(rid, {"matched": False}) - drop_path = dropped["path"] remainder = dropped["remainder"] if dropped["is_image"]: session.setdefault("attached_images", []).append(str(drop_path)) - text = remainder or f"[User attached image: {drop_path.name}]" - return _ok( - rid, - { - "matched": True, - "is_image": True, - "path": str(drop_path), - "count": len(session["attached_images"]), - "text": text, - **_image_meta(drop_path), - }, - ) - + return _ok(rid, { + "matched": True, "is_image": True, "path": str(drop_path), + "count": len(session["attached_images"]), + "text": remainder or f"[User attached image: {drop_path.name}]", + **_image_meta(drop_path), + }) text = f"[User attached file: {drop_path}]" + (f"\n{remainder}" if remainder else "") - return _ok( - rid, - { - "matched": True, - "is_image": False, - "path": str(drop_path), - "name": drop_path.name, - "text": text, - }, - ) + return _ok(rid, { + "matched": True, "is_image": False, "path": str(drop_path), "name": drop_path.name, + "text": text, + }) except Exception as e: return _err(rid, 5027, str(e)) @@ -1217,6 +1113,32 @@ def _final_response_text(result) -> str: return (result.get("final_response", str(result)) if isinstance(result, dict) else str(result)) +def _spawn_side_agent( + rid, session, task_id, parent, event, body, *, cwd="", extra=None, cleanup=None +): + """Run ``body()`` (an ephemeral agent call) on a daemon thread under the + session's profile home and cwd; its text — or ``error: `` — lands on + ``parent`` as ``event`` with ``task_id`` (+ ``extra`` fields). ``cleanup`` + runs in the finally before the session context is cleared. Replies ``{task_id}``.""" + extra = extra or {} + + def run(): + session_tokens = _set_session_context(task_id, cwd=(cwd or _session_cwd(session))) + try: + with _session_profile_home_scope(session): + text = body() + _emit(event, parent, {"task_id": task_id, **extra, "text": text}) + except Exception as e: + _emit(event, parent, {"task_id": task_id, **extra, "text": f"error: {e}"}) + finally: + if cleanup is not None: + cleanup() + _clear_session_context(session_tokens) + + threading.Thread(target=run, daemon=True).start() + return _ok(rid, {"task_id": task_id}) + + @method("prompt.background") def _(rid, params: dict) -> dict: session, err = _sess(params, rid) @@ -1227,29 +1149,14 @@ def _(rid, params: dict) -> dict: return _err(rid, 4012, "text required") task_id = f"bg_{uuid.uuid4().hex[:6]}" - def run(): - session_tokens = _set_session_context(task_id, cwd=_session_cwd(session)) - try: - from run_agent import AIAgent + def body(): + from run_agent import AIAgent + result = AIAgent(**_background_agent_kwargs(session["agent"], task_id)).run_conversation( + user_message=text, task_id=task_id, + ) + return _final_response_text(result) - with _session_profile_home_scope(session): - result = AIAgent( - **_background_agent_kwargs(session["agent"], task_id) - ).run_conversation( - user_message=text, - task_id=task_id, - ) - _emit( - "background.complete", parent, - {"task_id": task_id, "text": _final_response_text(result)}, - ) - except Exception as e: - _emit("background.complete", parent, {"task_id": task_id, "text": f"error: {e}"}) - finally: - _clear_session_context(session_tokens) - - threading.Thread(target=run, daemon=True).start() - return _ok(rid, {"task_id": task_id}) + return _spawn_side_agent(rid, session, task_id, parent, "background.complete", body) @method("prompt.btw") @@ -1265,38 +1172,21 @@ def _(rid, params: dict) -> dict: if not text: return _err(rid, 4012, "text required") task_id = f"btw_{uuid.uuid4().hex[:6]}" - agent = session.get("agent") snapshot = list(getattr(agent, "_session_messages", None) or session.get("history") or []) main_runtime = { - "model": getattr(agent, "model", None), "provider": getattr(agent, "provider", None), - "base_url": getattr(agent, "base_url", None), "api_key": getattr(agent, "api_key", None), - "api_mode": getattr(agent, "api_mode", None), + k: getattr(agent, k, None) for k in ("model", "provider", "base_url", "api_key", "api_mode") } - def run(): - session_tokens = _set_session_context(task_id, cwd=_session_cwd(session)) - try: - from agent.side_question import answer_side_question + def body(): + from agent.side_question import answer_side_question + return answer_side_question( + text, snapshot, parent_agent=agent, main_runtime=main_runtime, + ) or "" - with _session_profile_home_scope(session): - answer = answer_side_question( - text, snapshot, parent_agent=agent, main_runtime=main_runtime, - ) - _emit( - "btw.complete", parent, - {"task_id": task_id, "question": text, "text": answer or ""}, - ) - except Exception as e: - _emit( - "btw.complete", parent, - {"task_id": task_id, "question": text, "text": f"error: {e}"}, - ) - finally: - _clear_session_context(session_tokens) - - threading.Thread(target=run, daemon=True).start() - return _ok(rid, {"task_id": task_id}) + return _spawn_side_agent( + rid, session, task_id, parent, "btw.complete", body, extra={"question": text}, + ) _PREVIEW_RESTART_RULES = ( @@ -1326,14 +1216,11 @@ def _(rid, params: dict) -> dict: session, err = _sess(params, rid) if err: return err - url = str(params.get("url") or "").strip() cwd = str(params.get("cwd") or "").strip() context = str(params.get("context") or "").strip() - if not url: return _err(rid, 4012, "url required") - task_id = f"preview_{uuid.uuid4().hex[:6]}" parent = params.get("session_id", "") parent_history = _preview_restart_history(session) @@ -1349,7 +1236,6 @@ def _(rid, params: dict) -> dict: ] if line ) - # A malformed client path (embedded NUL, etc.) must not blow up the restart: # treat it as "no validated cwd". try: @@ -1359,54 +1245,41 @@ def _(rid, params: dict) -> dict: except Exception: preview_cwd = "" - def run(): - # Pin the validated preview cwd, else the parent workspace — never an - # invalid client path (which would silently fall back to the launch dir). - session_tokens = _set_session_context(task_id, cwd=(preview_cwd or _session_cwd(session))) - try: - from run_agent import AIAgent - from tools.terminal_tool import register_task_env_overrides + def body(): + from run_agent import AIAgent + from tools.terminal_tool import register_task_env_overrides + if preview_cwd: + register_task_env_overrides(task_id, {"cwd": preview_cwd}) + history_note = ( + f" (with {len(parent_history)} parent-session messages of context)" + if parent_history else "" + ) + _emit( + "preview.restart.progress", parent, + {"task_id": task_id, "text": f"Starting hidden restart agent{history_note}"}, + ) + # Deliberately NOT closed through task-wide process cleanup: the whole + # point is to leave a background server running under this task_id, + # and AIAgent.close() would kill every process for it. + result = AIAgent( + **_ephemeral_preview_agent_kwargs(session["agent"], task_id), + **_preview_restart_callbacks(parent, task_id), + ).run_conversation( + user_message=prompt, task_id=task_id, conversation_history=parent_history or None, + ) + return _final_response_text(result) - if preview_cwd: - register_task_env_overrides(task_id, {"cwd": preview_cwd}) + def cleanup(): + with contextlib.suppress(Exception): + from tools.terminal_tool import clear_task_env_overrides + clear_task_env_overrides(task_id) - history_note = ( - f" (with {len(parent_history)} parent-session messages of context)" - if parent_history - else "" - ) - _emit( - "preview.restart.progress", parent, - {"task_id": task_id, "text": f"Starting hidden restart agent{history_note}"}, - ) - # Deliberately NOT closed through task-wide process cleanup: the whole - # point is to leave a background server running under this task_id, - # and AIAgent.close() would kill every process for it. - with _session_profile_home_scope(session): - result = AIAgent( - **_ephemeral_preview_agent_kwargs(session["agent"], task_id), - **_preview_restart_callbacks(parent, task_id), - ).run_conversation( - user_message=prompt, task_id=task_id, - conversation_history=parent_history or None, - ) - _emit( - "preview.restart.complete", parent, - {"task_id": task_id, "text": _final_response_text(result)}, - ) - except Exception as e: - _emit("preview.restart.complete", parent, {"task_id": task_id, "text": f"error: {e}"}) - finally: - try: - from tools.terminal_tool import clear_task_env_overrides - - clear_task_env_overrides(task_id) - except Exception: - pass - _clear_session_context(session_tokens) - - threading.Thread(target=run, daemon=True).start() - return _ok(rid, {"task_id": task_id}) + # Pin the validated preview cwd, else the parent workspace — never an + # invalid client path (which would silently fall back to the launch dir). + return _spawn_side_agent( + rid, session, task_id, parent, "preview.restart.complete", body, + cwd=preview_cwd, cleanup=cleanup, + ) # ── late-answer RPCs for tool-driven UI cards ─────────────────────────────── @@ -1433,7 +1306,6 @@ _LATE_RESPOND_KEYS = { def _late_respond(key: str): def handler(rid, params: dict) -> dict: return _respond(rid, params, key, allow_expired=True) - return handler @@ -1449,7 +1321,6 @@ def _approval_reply(rid, result_key, call): """``_ok({result_key: call(tools.approval)})``, 5004 on any failure.""" try: import tools.approval as approval - return _ok(rid, {result_key: call(approval)}) except Exception as e: return _err(rid, 5004, str(e)) @@ -1488,16 +1359,15 @@ def _approval_respond_session_fallback(params: dict): if request_id: try: from tools.approval import list_gateway_approvals - with _sessions_lock: live = list(_sessions.items()) for sid, session in live: key = str(session.get("session_key") or "") - if not key: - continue - for pending in list_gateway_approvals(key): - if str(pending.get("request_id") or "") == request_id: - return session + if key and any( + str(pending.get("request_id") or "") == request_id + for pending in list_gateway_approvals(key) + ): + return session except Exception: logger.debug("approval.respond request_id fallback failed", exc_info=True) target = str(params.get("session_id") or "") @@ -1516,8 +1386,7 @@ def _(rid, params: dict) -> dict: session, err = _sess(params, rid) if err: # Session-not-found (4001) only: resolve by durable identity before failing. - code = (err.get("error") or {}).get("code") - if code != 4001: + if (err.get("error") or {}).get("code") != 4001: return err session = _approval_respond_session_fallback(params) if session is None: @@ -1525,10 +1394,8 @@ def _(rid, params: dict) -> dict: return _approval_reply( rid, "resolved", lambda a: a.resolve_gateway_approval( - session["session_key"], - params.get("choice", "deny"), - resolve_all=params.get("all", False), - request_id=params.get("request_id"), + session["session_key"], params.get("choice", "deny"), + resolve_all=params.get("all", False), request_id=params.get("request_id"), ), ) diff --git a/tui_gateway/prompt_attachments.py b/tui_gateway/prompt_attachments.py index da3821d0d0..acb9c5a54c 100644 --- a/tui_gateway/prompt_attachments.py +++ b/tui_gateway/prompt_attachments.py @@ -1,4 +1,4 @@ -"""Attachment staging for prompt.submit / prompt.attach: image sniffing, size caps, per-session attachment dirs, gateway attachment path resolution. +"""Attachment staging: image sniffing, size caps, per-session attachment dirs, path resolution. Bodies are rebound onto server.py's globals at install time (see method_ctx.bind_module), so they reference server.py globals bare. @@ -17,13 +17,10 @@ _ATTACH_BYTES_MAX_BYTES = 25 * 1024 * 1024 _PDF_ATTACH_MAX_BYTES = 50 * 1024 * 1024 _PDF_ATTACH_MAX_PAGES = 25 -# Leading magic bytes → file extension, for filename-less uploads. +# Leading magic bytes -> file extension, for filename-less uploads. _IMAGE_MAGIC: tuple[tuple[bytes, str], ...] = ( - (b"\x89PNG\r\n\x1a\n", ".png"), - (b"\xff\xd8\xff", ".jpg"), - (b"GIF87a", ".gif"), - (b"GIF89a", ".gif"), - (b"BM", ".bmp"), + (b"\x89PNG\r\n\x1a\n", ".png"), (b"\xff\xd8\xff", ".jpg"), (b"GIF87a", ".gif"), + (b"GIF89a", ".gif"), (b"BM", ".bmp"), ) # Context-ref values containing any of these must be quoted (desktop formatRefValue parity). @@ -31,27 +28,33 @@ _ATTACHMENT_REF_NEEDS_QUOTING_RE = _re.compile(r"""[\s()\[\]{}<>"'`]""") del _re # bodies are rebound onto server globals: import inside functions only +def _b64_payload(raw: str, data_url_re: str, flags: int) -> bytes: + """Strip an optional ``data:...;base64,`` wrapper and all whitespace, then + strictly decode (raises ``binascii.Error``/``ValueError`` on bad base64).""" + import base64 as _base64 + import re as _re + cleaned = (raw or "").strip() + m = _re.match(data_url_re, cleaned, flags) + if m: + cleaned = m.group(1) + return _base64.b64decode(_re.sub(r"\s+", "", cleaned), validate=True) + + def _decode_attach_base64(raw: str, *, mime_prefix: str) -> bytes | None: """Decode a base64 payload, optionally ``data:...;base64,``-wrapped, tolerating embedded whitespace. ``None`` when not valid base64.""" - import base64 as _base64 import re as _re - - cleaned = raw.strip() - m = _re.match( - rf"^data:{_re.escape(mime_prefix)}[a-zA-Z0-9.+-]*;base64,(.*)$", cleaned, _re.DOTALL, - ) - if m: - cleaned = m.group(1) - cleaned = _re.sub(r"\s+", "", cleaned) try: - return _base64.b64decode(cleaned, validate=True) + return _b64_payload( + raw, rf"^data:{_re.escape(mime_prefix)}[a-zA-Z0-9.+-]*;base64,(.*)$", _re.DOTALL, + ) except Exception: return None -def _decode_attach_payload(rid, raw_b64: str, *, mime_prefix: str, max_bytes: int, - label: str, empty_msg: str): +def _decode_attach_payload( + rid, raw_b64: str, *, mime_prefix: str, max_bytes: int, label: str, empty_msg: str +): """``(bytes, None)`` or ``(None, error)`` for an upload: 4017 on bad/empty base64, 4018 over *max_bytes*.""" data = _decode_attach_base64(raw_b64, mime_prefix=mime_prefix) @@ -68,23 +71,17 @@ def _decode_attach_payload(rid, raw_b64: str, *, mime_prefix: str, max_bytes: in def _sniff_image_ext(img_bytes: bytes, filename: str = "") -> str: """Extension from the filename hint, else magic bytes (WebP needs the RIFF/WEBP container check), else ``.png``.""" - if filename: - suffix = Path(filename).suffix.lower() - if suffix: - return suffix + if filename and (suffix := Path(filename).suffix.lower()): + return suffix head = img_bytes[:16] if head.startswith(b"RIFF") and head[8:12] == b"WEBP": return ".webp" - for sig, ext in _IMAGE_MAGIC: - if head.startswith(sig): - return ext - return ".png" + return next((ext for sig, ext in _IMAGE_MAGIC if head.startswith(sig)), ".png") def _allowed_image_extensions() -> frozenset[str]: try: from cli import _IMAGE_EXTENSIONS - return frozenset(_IMAGE_EXTENSIONS) except Exception: return frozenset({".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp"}) @@ -100,8 +97,7 @@ def _session_home_dir(session: dict, name: str) -> Path: can never see the file. """ profile_home = session.get("profile_home") - base = Path(profile_home) if profile_home else _hermes_home - return base / name + return (Path(profile_home) if profile_home else _hermes_home) / name def _session_images_dir(session: dict) -> Path: @@ -131,12 +127,9 @@ def _format_ref_value(value: str) -> str: ``@file:`` ref round-trips through ``agent.context_references``.""" if not value or not _ATTACHMENT_REF_NEEDS_QUOTING_RE.search(value): return value - if "`" not in value: - return f"`{value}`" - if '"' not in value: - return f'"{value}"' - if "'" not in value: - return f"'{value}'" + for q in ("`", '"', "'"): + if q not in value: + return f"{q}{value}{q}" return value @@ -144,8 +137,7 @@ def _attachment_ref_path(session: dict, target: Path) -> str: """Workspace-relative path for an attachment, or the absolute path if outside.""" workspace = Path(_session_cwd(session)).resolve() try: - rel = target.resolve().relative_to(workspace) - return str(rel).replace(os.sep, "/") + return str(target.resolve().relative_to(workspace)).replace(os.sep, "/") except ValueError: return str(target.resolve()) @@ -161,11 +153,8 @@ def _desktop_attachment_dir(session: dict) -> Path: def _sanitize_attachment_name(name: str) -> str: import re as _re - - candidate = Path(str(name or "").strip()).name - candidate = _re.sub(r"[\x00-\x1f]+", "_", candidate) - candidate = candidate.strip().strip(".") - return candidate or "attachment" + candidate = _re.sub(r"[\x00-\x1f]+", "_", Path(str(name or "").strip()).name) + return candidate.strip().strip(".") or "attachment" def _unique_attachment_path(root: Path, filename: str) -> Path: @@ -175,11 +164,9 @@ def _unique_attachment_path(root: Path, filename: str) -> Path: stem = Path(filename).stem or "attachment" suffix = Path(filename).suffix counter = 2 - while True: - next_candidate = root / f"{stem}-{counter}{suffix}" - if not next_candidate.exists(): - return next_candidate + while (next_candidate := root / f"{stem}-{counter}{suffix}").exists(): counter += 1 + return next_candidate def _resolve_gateway_attachment_path(raw: str) -> Path | None: @@ -190,7 +177,6 @@ def _resolve_gateway_attachment_path(raw: str) -> Path | None: from cli import _detect_file_drop, _resolve_attachment_path, _split_path_input except Exception: return None - dropped = _detect_file_drop(raw) if dropped: return Path(dropped["path"]).resolve() @@ -202,34 +188,25 @@ def _resolve_gateway_attachment_path(raw: str) -> Path | None: def _decode_attachment_data_url(data_url: str) -> bytes: """Decode a ``data:;base64,`` payload (any media type, unlike the image-specific ``_decode_attach_base64``); bare base64 also accepted.""" - import base64 as _base64 import binascii as _binascii import re as _re - - cleaned = (data_url or "").strip() - m = _re.match(r"^data:[^;,]*(?:;[^;,=]+=[^;,]+)*;base64,(.*)$", cleaned, _re.DOTALL | _re.I) - if m: - cleaned = m.group(1) - cleaned = _re.sub(r"\s+", "", cleaned) try: - return _base64.b64decode(cleaned, validate=True) + return _b64_payload( + data_url, r"^data:[^;,]*(?:;[^;,=]+=[^;,]+)*;base64,(.*)$", _re.DOTALL | _re.I, + ) except (ValueError, _binascii.Error) as exc: raise ValueError("invalid data_url payload") from exc def _stage_session_file_attachment( - session: dict, - *, - raw_path: str, - data_url: str, - name: str, + session: dict, *, raw_path: str, data_url: str, name: str ) -> tuple[Path, bool]: """Make a desktop file attachment available to the gateway agent. - 1. Path resolves INSIDE the session workspace → use as-is (``uploaded=False``). - 2. Gateway-visible file OUTSIDE the workspace → copy into ``attachments/`` + 1. Path resolves INSIDE the session workspace -> use as-is (``uploaded=False``). + 2. Gateway-visible file OUTSIDE the workspace -> copy into ``attachments/`` (bind-mounted into container backends) so ``@file:`` resolves in the sandbox. - 3. Not on the gateway (remote client disk) → decode ``data_url`` bytes into + 3. Not on the gateway (remote client disk) -> decode ``data_url`` bytes into ``attachments/``. Returns ``(stored_path, uploaded)``. """ @@ -247,9 +224,9 @@ def _stage_session_file_attachment( raise ValueError("file not found on gateway and no data_url provided") payload = _decode_attachment_data_url(data_url) filename = _sanitize_attachment_name(name or Path(str(raw_path or "")).name) - - upload_dir = _desktop_attachment_dir(session) - target = _unique_attachment_path(upload_dir, _sanitize_attachment_name(filename)) + target = _unique_attachment_path( + _desktop_attachment_dir(session), _sanitize_attachment_name(filename) + ) target.write_bytes(payload) return target.resolve(), True diff --git a/tui_gateway/prompt_turn.py b/tui_gateway/prompt_turn.py index 009276dbff..280a1f0fa7 100644 --- a/tui_gateway/prompt_turn.py +++ b/tui_gateway/prompt_turn.py @@ -16,12 +16,14 @@ from .method_ctx import HandlerRegistry, bind_module _registry = HandlerRegistry() +def _hook_failure(what: str, exc: BaseException) -> None: + print(f"[tui_gateway] {what} failed: {type(exc).__name__}: {exc}", file=sys.stderr) + + def _is_successful_goal_turn(result: Any, status: str, raw: Any) -> bool: """Return whether a turn produced a real response the goal judge can use.""" return bool( - status == "complete" - and isinstance(raw, str) - and raw.strip() + status == "complete" and isinstance(raw, str) and raw.strip() and not (isinstance(result, dict) and result.get("failed")) and not (isinstance(result, dict) and result.get("completed") is False) ) @@ -40,13 +42,10 @@ def _plan_goal_compression_recovery( ) -> tuple[str | None, str | None]: """Plan a bounded active-goal retry after compression exhaustion. - Compression exhaustion is a failed turn, so it must not be sent to the - goal judge or consume the goal's turn budget. One fresh continuation turn - is allowed. If that turn also exhausts compression, pause the goal rather - than spinning until an arbitrary user message happens to wake it up. - - Returns ``(continuation_prompt, status_notice)``. Sessions without an - active goal retain the existing error-only behavior. + Compression exhaustion is a failed turn: never judge input, never a spent goal + turn. One fresh continuation is allowed; if that also exhausts, pause the goal + instead of spinning until a random user message wakes it. Returns + ``(continuation_prompt, status_notice)``; no active goal -> ``(None, None)``. """ compression_exhausted = bool(isinstance(result, dict) and result.get("compression_exhausted")) if not compression_exhausted: @@ -62,12 +61,13 @@ def _plan_goal_compression_recovery( session.pop(_GOAL_COMPRESSION_RECOVERY_ATTEMPTS, None) return None, None goal_created_at = float(getattr(goal_mgr.state, "created_at", 0.0) or 0.0) + goal_text = getattr(goal_mgr.state, "goal", "") recovery_state = session.get(_GOAL_COMPRESSION_RECOVERY_ATTEMPTS) attempts = 0 if ( isinstance(recovery_state, dict) and recovery_state.get("goal_created_at") == goal_created_at - and recovery_state.get("goal") == getattr(goal_mgr.state, "goal", "") + and recovery_state.get("goal") == goal_text ): try: attempts = int(recovery_state.get("attempts", 0) or 0) @@ -76,8 +76,7 @@ def _plan_goal_compression_recovery( continuation_prompt = goal_mgr.next_continuation_prompt() if attempts < _GOAL_COMPRESSION_RECOVERY_LIMIT and continuation_prompt: session[_GOAL_COMPRESSION_RECOVERY_ATTEMPTS] = { - "goal_created_at": goal_created_at, "goal": getattr(goal_mgr.state, "goal", ""), - "attempts": attempts + 1, + "goal_created_at": goal_created_at, "goal": goal_text, "attempts": attempts + 1, } return ( continuation_prompt, @@ -102,11 +101,10 @@ def _admit_prompt_turn( ) -> tuple[list[str], Any] | None: """Ownership + liveness gate every fresh turn source must cross. - prompt.submit already claims the slot in its RPC handler, but crash - auto-continue, wake-ups and other synthesized turns call - ``_run_prompt_submit`` directly — the bypass that once let a second backend - run a duplicate turn. Returns ``(images, agent)`` or None when refused - (``running`` already reset). + prompt.submit claims the slot in its RPC handler, but crash auto-continue, + wake-ups and other synthesized turns call ``_run_prompt_submit`` directly — + the bypass that once let a second backend run a duplicate turn. Returns + ``(images, agent)`` or None when refused (``running`` already reset). """ if (ownership_refusal := _ensure_active_session_slot(sid, session)) is not None: logger.info( @@ -137,22 +135,20 @@ def _admit_prompt_turn( _start_inflight_turn(session, text) agent = session["agent"] if hasattr(agent, "clear_interrupt"): - try: + with contextlib.suppress(Exception): agent.clear_interrupt() - except Exception: - pass return images, agent def _record_turn_marker(session: dict, text: Any) -> str: """Write the durable crash marker; returns the session key it was written under. - Retired the moment the outcome reaches the client (_retire_turn_marker), so - a surviving marker means the process died mid-turn and session.resume - auto-continues from it. Compression can rotate session_key mid-turn, so - the caller keeps this key. The key is published before the disk write so - an interrupt racing startup can retire it; the post-write cancel check - closes the inverse race where Stop lands first and clears no file yet. + Retired when the outcome reaches the client (_retire_turn_marker), so a + surviving marker means the process died mid-turn and session.resume + auto-continues from it. Compression can rotate session_key mid-turn, so the + caller keeps this key. The key is published before the disk write so an + interrupt racing startup can retire it; the post-write cancel check closes + the inverse race where Stop lands first and clears no file yet. """ marker_home = _session_home(session) marker_key = str(session.get("session_key") or "") @@ -213,12 +209,10 @@ def _bind_turn_scopes(sid: str, session: dict, scopes: _TurnScopes) -> None: def _release_turn_scopes(scopes: _TurnScopes) -> None: - try: + with contextlib.suppress(Exception): if scopes.approval is not None: from tools.approval import reset_current_session_key reset_current_session_key(scopes.approval) - except Exception: - pass if scopes.home is not None: reset_hermes_home_override(scopes.home) if scopes.secret is not None: @@ -282,8 +276,7 @@ def _route_turn_images(agent, prompt: Any, images: list[str]) -> Any: return parts except Exception as _img_exc: print( - f"[tui_gateway] native attach failed, falling back to text: {_img_exc}", - file=sys.stderr, + f"[tui_gateway] native attach failed, falling back to text: {_img_exc}", file=sys.stderr, ) return _build_image_ref_message(prompt, images) @@ -291,14 +284,13 @@ def _route_turn_images(agent, prompt: Any, images: list[str]) -> Any: def _start_turn_voice() -> tuple[Any, bool]: """Arm voice-mode turn audio; returns ``(tts_queue, thinking_started)``. - Streaming TTS speaks replies sentence-by-sentence as tokens arrive (CLI - parity). ``_tts_stream_begin`` goes first: it cuts any still-speaking - previous turn, and that cut IS this turn's barge-in, so it must latch - before the caller consumes the latch. The full-duplex listener lets the - user interject DURING generation (covers voice mode without working TTS). - The ambient "thinking" sound keeps long silent stretches from reading as - a dead session; per-blip gate skips while TTS audio flows or the mic is - capturing; stopped in the turn's finally. + ``_tts_stream_begin`` goes first: it cuts any still-speaking previous turn, + and that cut IS this turn's barge-in, so it must latch before the caller + consumes the latch. The full-duplex listener lets the user interject DURING + generation (covers voice mode without working TTS). The ambient "thinking" + sound keeps long silent stretches from reading as a dead session; the + per-blip gate skips while TTS audio flows or the mic is capturing; stopped in + the turn's finally. """ tts_queue = _tts_stream_begin() if not _voice_mode_enabled(): @@ -322,20 +314,18 @@ def _start_turn_voice() -> tuple[Any, bool]: def _stop_thinking_sound() -> None: - try: + with contextlib.suppress(Exception): from tools.voice_mode import stop_thinking_sound stop_thinking_sound() - except Exception: - pass def _apply_turn_notes(run_message: Any, session: dict) -> Any: """Prepend the per-turn API-message notes (same enrichment channel as images). - Barge mid-speech → tell the model so it can react instead of being - oblivious to its own interruption; then reactions added since the last - turn; then which window the message was typed into (HUD mode is per-turn - state, so it cannot live in the byte-stable system prompt). + Barge mid-speech -> tell the model so it can react to its own interruption; + then reactions added since the last turn; then which window the message was + typed into (HUD mode is per-turn state, so it cannot live in the byte-stable + system prompt). """ from tools.tts_streaming import SPEECH_INTERRUPTED_NOTE, take_speech_interrupted if take_speech_interrupted(): @@ -420,10 +410,7 @@ def _restore_moa_one_shot(sid: str, session: dict) -> None: _raw = f"{_prev_model} --provider {_prev_provider}" if _prev_provider else _prev_model try: _apply_model_switch( - sid, - session, - _raw, - confirm_expensive_model=False, + sid, session, _raw, confirm_expensive_model=False, pin_session_override=bool(_prev_override), persist_override=False, # session-internal restore, never config.yaml ) @@ -489,9 +476,10 @@ def _turn_outcome(result: Any) -> tuple[Any, str, str | None]: return str(result), "complete", None raw = result.get("final_response", "") status = ( - "interrupted" if result.get("interrupted") else "error" if result.get("error") else "complete" + "interrupted" if result.get("interrupted") + else "error" if result.get("error") else "complete" ) - # No visible response AND a real error (e.g. invalid model slug → provider + # No visible response AND a real error (e.g. invalid model slug -> provider # 4xx): surface the error as the text (classic CLI parity) instead of an # empty turn. An empty successful turn still renders as empty. if (not raw) and result.get("error") and (result.get("failed") or result.get("partial")): @@ -508,7 +496,7 @@ def _turn_outcome(result: Any) -> tuple[Any, str, str | None]: def _turn_error_surface(agent, result: Any) -> Any: - """Structured {layer, code, retryable} descriptor for an error result (advisory; never raises).""" + """Structured {layer, code, retryable} descriptor for an error result (advisory, never raises).""" try: from agent.error_surface import build_error_surface_from_result return build_error_surface_from_result( @@ -522,7 +510,9 @@ def _turn_error_surface(agent, result: Any) -> Any: # ── post-turn hooks ────────────────────────────────────────────────── -def _goal_followup_after_turn(sid: str, session: dict, result: Any, status: str, raw: Any) -> str | None: +def _goal_followup_after_turn( + sid: str, session: dict, result: Any, status: str, raw: Any +) -> str | None: """/goal continuation (Ralph-style loop; mirrors gateway/run._post_turn_goal_continuation). Asks the judge whether the goal is done and, if not and still under budget, @@ -542,45 +532,36 @@ def _goal_followup_after_turn(sid: str, session: dict, result: Any, status: str, if recovery_prompt: goal_followup = recovery_prompt except Exception as _goal_recovery_exc: - print( - f"[tui_gateway] goal compression recovery failed: " - f"{type(_goal_recovery_exc).__name__}: {_goal_recovery_exc}", - file=sys.stderr, - ) + _hook_failure("goal compression recovery", _goal_recovery_exc) if compression_exhausted or not _is_successful_goal_turn(result, status, raw): return goal_followup try: from hermes_cli.goals import GoalManager sid_key = session.get("session_key") or "" - if sid_key: - goal_mgr = GoalManager(session_id=sid_key, default_max_turns=_goal_max_turns()) - if goal_mgr.is_active(): - try: - from hermes_cli.goals import gather_background_processes as _gather_bg - _bg_procs = _gather_bg() - except Exception: - _bg_procs = None - decision = goal_mgr.evaluate_after_turn( - raw, user_initiated=True, background_processes=_bg_procs - ) - verdict_msg = decision.get("message") or "" - if verdict_msg: - _emit("status.update", sid, {"kind": "goal", "text": verdict_msg}) - if decision.get("should_continue"): - cont_prompt = decision.get("continuation_prompt") or "" - if cont_prompt: - goal_followup = cont_prompt + if sid_key and ( + goal_mgr := GoalManager(session_id=sid_key, default_max_turns=_goal_max_turns()) + ).is_active(): + try: + from hermes_cli.goals import gather_background_processes as _gather_bg + _bg_procs = _gather_bg() + except Exception: + _bg_procs = None + decision = goal_mgr.evaluate_after_turn( + raw, user_initiated=True, background_processes=_bg_procs + ) + if verdict_msg := decision.get("message") or "": + _emit("status.update", sid, {"kind": "goal", "text": verdict_msg}) + if decision.get("should_continue") and ( + cont_prompt := decision.get("continuation_prompt") or "" + ): + goal_followup = cont_prompt except Exception as _goal_exc: - print( - f"[tui_gateway] goal continuation hook failed: " - f"{type(_goal_exc).__name__}: {_goal_exc}", - file=sys.stderr, - ) + _hook_failure("goal continuation hook", _goal_exc) return goal_followup def _complete_loop_tick(sid: str, session: dict, raw: Any) -> None: - """If this turn was a /loop wakeup, evaluate it (LOOP_COMPLETE, --until judge, caps, next tick).""" + """If this turn was a /loop wakeup, evaluate it (LOOP_COMPLETE, --until judge, caps, next).""" try: from hermes_cli.loops import LoopManager loop_sid_key = session.get("session_key") or "" @@ -589,15 +570,10 @@ def _complete_loop_tick(sid: str, session: dict, raw: Any) -> None: loop_state = loop_mgr.state if loop_state is not None and loop_state.awaiting_response: loop_decision = loop_mgr.complete_tick(raw if isinstance(raw, str) else "") - loop_msg = loop_decision.get("message") or "" - if loop_msg: + if loop_msg := loop_decision.get("message") or "": _emit("status.update", sid, {"kind": "loop", "text": loop_msg}) except Exception as _loop_exc: - print( - f"[tui_gateway] loop completion hook failed: " - f"{type(_loop_exc).__name__}: {_loop_exc}", - file=sys.stderr, - ) + _hook_failure("loop completion hook", _loop_exc) def _apply_pending_title(sid: str, session: dict) -> None: @@ -630,7 +606,7 @@ def _speak_turn_fallback(raw: str) -> None: def _append_turn_crash_log(sid: str, trace: str) -> None: - try: + with contextlib.suppress(Exception): os.makedirs(os.path.dirname(_CRASH_LOG), exist_ok=True) with open(_CRASH_LOG, "a", encoding="utf-8") as f: f.write( @@ -638,11 +614,28 @@ def _append_turn_crash_log(sid: str, trace: str) -> None: f"{time.strftime('%Y-%m-%d %H:%M:%S')} · sid={sid} ===\n" ) f.write(trace) - except Exception: - pass -def _run_post_turn_followups(rid, sid: str, session: dict, result: Any, goal_followup: str | None) -> None: +def _dispatch_followup_turn(rid, sid: str, session: dict, prompt: Any, what: str, *, + on_done=None, on_error=None) -> None: + """Chain one follow-up turn (caller already set ``running``); a dispatch failure + runs ``on_error``, logs, and releases ``running``.""" + try: + _emit("message.start", sid) + _run_prompt_submit(rid, sid, session, prompt) + if on_done is not None: + on_done() + except Exception as exc: + if on_error is not None: + on_error() + _hook_failure(what, exc) + with session["history_lock"]: + session["running"] = False + + +def _run_post_turn_followups( + rid, sid: str, session: dict, result: Any, goal_followup: str | None +) -> None: """Chain whatever should run after ``running`` was released. Order: a user prompt that arrived mid-turn (interrupt + queue) wins over @@ -666,17 +659,7 @@ def _run_post_turn_followups(rid, sid: str, session: dict, result: Any, goal_fol if session.get("running"): return # user already sent something — their turn wins session["running"] = True - try: - _emit("message.start", sid) - _run_prompt_submit(rid, sid, session, goal_followup) - except Exception as _cont_exc: - print( - f"[tui_gateway] goal continuation dispatch failed: " - f"{type(_cont_exc).__name__}: {_cont_exc}", - file=sys.stderr, - ) - with session["history_lock"]: - session["running"] = False + _dispatch_followup_turn(rid, sid, session, goal_followup, "goal continuation dispatch") # The background poller handles between-turn delivery; this is the safety # net for events that arrived mid-turn. Ownership is positive-proof and @@ -704,30 +687,322 @@ def _run_post_turn_followups(rid, sid: str, session: dict, result: Any, goal_fol _claim = claim_event_delivery(_evt, "tui-post-turn") if _claim is None: continue - try: - _emit("message.start", sid) - _run_prompt_submit(rid, sid, session, synth) - complete_event_delivery(_evt, _claim) - except Exception as _n_exc: - release_event_delivery(_evt, _claim) - print( - f"[tui_gateway] completion notification dispatch failed: " - f"{type(_n_exc).__name__}: {_n_exc}", - file=sys.stderr, - ) - with session["history_lock"]: - session["running"] = False + _dispatch_followup_turn( + rid, sid, session, synth, "completion notification dispatch", + on_done=lambda: complete_event_delivery(_evt, _claim), + on_error=lambda: release_event_delivery(_evt, _claim), + ) except Exception as _drain_exc: - print( - f"[tui_gateway] completion queue drain failed: " - f"{type(_drain_exc).__name__}: {_drain_exc}", - file=sys.stderr, - ) + _hook_failure("completion queue drain", _drain_exc) # ── the turn ───────────────────────────────────────────────────────── +class _TurnRun: + """Mutable state the phase helpers of one turn thread share. + + ``agent`` is bound eagerly so the except/finally paths always have one even + if setup throws; re-read after ``_sync_bot_capabilities``, which may swap in + a rebuilt agent for Bot Chat sessions. ``error_retained`` tells the finally + to skip the normal inflight clear (a failed turn's snapshot stays for resume + replay). ``error_detail`` is the cause for the "tui turn finished" bookend, + stashed by both failure paths because the finally sees neither ``result`` + nor the exception reliably; ``prompt_text`` is what was actually submitted + (post @-expansion) so the cause can be checked for quoting it back. + """ + + __slots__ = ( + "agent", "scopes", "goal_followup", "result", "tts_queue", "thinking_started", "history", + "history_version", "run_kwargs", "one_turn_restore", "error_retained", "error_detail", + "prompt_text", "marker_key", "terminal_callback", "receipt_attempted", "receipt_committed", + ) + + def __init__(self, session: dict, terminal_callback) -> None: + self.agent = session["agent"] + self.scopes = _TurnScopes() + self.goal_followup = None + self.result = None # read after the finally for leftover /steer + self.tts_queue = None + self.thinking_started = False + self.history: list = [] + self.history_version = 0 + self.run_kwargs = None + self.one_turn_restore = session.pop("one_turn_model_restore", None) + self.error_retained = False + self.error_detail = "" + self.prompt_text = "" + self.marker_key = "" + self.terminal_callback = terminal_callback + self.receipt_attempted = False + self.receipt_committed = terminal_callback is None + + +def _prepare_turn_input(sid: str, session: dict, st: _TurnRun, text: Any, images: list[str]): + """Bind scopes, sync the agent, snapshot history and build the run message. + + Returns ``(prompt, run_message, cols, streamer)`` or None when @-expansion was refused + (error already emitted). The config-model sync is skipped while a + /model --once override is active: the once-model is deliberately not pinned + as a session model_override, so the sync would clobber it back to the config + model (a config.yaml change is adopted on the NEXT turn). A model picked + mid-turn was queued, not applied — apply it here, before the first model call + and before the config sync so the explicit pick wins over a config change. + """ + _bind_turn_scopes(sid, session, st.scopes) + if not st.one_turn_restore: + _apply_pending_model_switch(sid, session) + _sync_agent_model_with_config(sid, session) + _sync_agent_compression_with_config(sid, session) + # Bot Chat: adopt Settings->Capabilities edits into the eternal bot session first. + _sync_bot_capabilities(sid, session) + st.agent = agent = session["agent"] + # Snapshot after turn-start model sync: a deferred switch mutates history + # and its version, and that mutation belongs to this turn. + with session["history_lock"]: + st.history = list(session["history"]) + st.history_version = int(session.get("history_version", 0)) + cwd = _session_cwd(session) + _register_session_cwd(session) + cols = session.get("cols", 80) + streamer = make_stream_renderer(cols) + prompt = text + if isinstance(prompt, str) and "@" in prompt: + ctx = _expand_context_references(agent, prompt, cwd) + if ctx.blocked: + _emit( + "error", sid, {"message": "\n".join(ctx.warnings) or "Context injection refused."}, + ) + return None + prompt = ctx.message + st.prompt_text = prompt if isinstance(prompt, str) else "" + run_message: Any = _route_turn_images(agent, prompt, images) if images else prompt + st.tts_queue, st.thinking_started = _start_turn_voice() + return prompt, _apply_turn_notes(run_message, session), cols, streamer + + +def _invoke_agent( + sid: str, session: dict, st: _TurnRun, prompt: Any, run_message: Any, streamer, + images: list[str], display_kind: str | None, display_metadata: dict | None, +) -> None: + """Wire the streaming callbacks and run the conversation into ``st.result``.""" + agent = st.agent + + def _stream(delta): + with session["history_lock"]: + _append_inflight_delta(session, delta) + payload = {"text": delta} + if streamer and (r := streamer.feed(delta)) is not None: + payload["rendered"] = r + if st.tts_queue is not None and isinstance(delta, str): + st.tts_queue.put(delta) + _emit("message.delta", sid, payload) + + # Interim assistant text (commentary beside tool calls, or the attempted + # final answer before a verify-on-stop nudge) is sealed by the desktop as + # its own segment instead of being lost when message.complete replaces the + # streaming buffer. Gated on display.interim_assistant_messages (default true). + if _load_interim_assistant_messages(): + def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None: + _emit("message.interim", sid, {"text": text, "already_streamed": already_streamed}) + agent.interim_assistant_callback = _interim_assistant_cb + else: + agent.interim_assistant_callback = None + st.run_kwargs = _build_run_kwargs( + agent, session, st.history, prompt, images, run_message, _stream, display_kind, + display_metadata, + ) + # Auto-titling fires inside the turn prologue; this live-rename hook + # repaints the sidebar the moment a title lands. + _title_key = session.get("session_key") or sid + agent._on_session_title = lambda t, _src, _k=_title_key: _emit( + "session.title", sid, {"session_id": _k, "title": t} + ) + _usage_stop, _usage_thread = _start_usage_ticker(sid, agent) + try: + st.result = agent.run_conversation(run_message, **st.run_kwargs) + finally: + # Stop AND join before anything below emits: a tick surviving past + # message.complete would roll the client's final usage back to a stale + # snapshot. The join is deliberately unbounded — once stop is set it + # only waits out one in-flight _get_usage/_emit, whose worst case (a + # stalled transport write) would stall the message.complete emit just + # the same. + _usage_stop.set() + _usage_thread.join() + + +def _absorb_turn_result( + sid: str, session: dict, st: _TurnRun, text: Any, display_kind: str | None, display_metadata +) -> str | None: + """Stamp, restore /moa, commit history, re-sync the session key; returns the history warning.""" + result = st.result + if display_kind and isinstance(text, str): + _stamp_synthetic_display_kind( + st.agent, session, result, text, display_kind, display_metadata + ) + if "moa_one_shot_restore" in session: + _restore_moa_one_shot(sid, session) + status_note = None + if isinstance(result, dict): + if isinstance(result.get("messages"), list): + status_note = _commit_turn_history(session, result, st.history, st.history_version) + # Auto-compression inside run_conversation() may have rotated + # agent.session_id: sync session_key before title/goal/finalize + # handling uses it, keep pending_title (user intent) for the + # continuation, and restart the slash worker so worker-backed + # commands (/title etc.) target the live session. + _sync_session_key_after_compress( + sid, session, clear_pending_title=False, restart_slash_worker=True, + ) + return status_note + + +def _complete_turn_payload(session: dict, st: _TurnRun, status_note: str | None, cols: int): + """Build the ``message.complete`` payload, retain/clear the inflight turn and + settle the hosted-room terminal receipt. Returns ``(payload, raw, status)``.""" + result, agent = st.result, st.agent + raw, status, last_reasoning = _turn_outcome(result) + payload = {"text": raw, "usage": _get_usage(agent), "status": status} + if last_reasoning: + payload["reasoning"] = last_reasoning + if status_note: + payload["warning"] = status_note + if result.get("response_previewed"): + payload["response_previewed"] = True + # Structured billing-wall descriptor so the client renders a + # billing-specific recovery surface instead of re-parsing text. + _billing_block = result.get("billing_block") if isinstance(result, dict) else None + if _billing_block: + payload["billing"] = _billing_block + payload["failure_reason"] = result.get("failure_reason") + if rendered := render_message(raw, cols): + payload["rendered"] = rendered + # Layer descriptor computed before the retain below so resume replay + # carries the same one (advisory; older clients ignore it). + _error_surface = _turn_error_surface(agent, result) if status == "error" else None + _result_error = result.get("error") if isinstance(result, dict) else None + error_value = _result_error if isinstance(result, dict) else raw + with session["history_lock"]: + if status == "error": + # Retain the failed turn for resume replay: if this terminal frame + # is lost to a disconnect, resume's inflight payload is the only + # carrier of the failure. + _fail_inflight_turn(session, error_value, error_surface=_error_surface) + st.error_retained = True + st.error_detail = _turn_failure_detail( + error_value, result.get("failure_reason") if isinstance(result, dict) else None, + st.prompt_text, + ) + else: + _clear_inflight_turn(session) + if status == "error": + payload["error"] = str(error_value or raw) + payload["recoverable"] = True + if _error_surface: + payload["error_surface"] = _error_surface + if st.terminal_callback is not None: + st.receipt_attempted = True + st.terminal_callback({ + "status": ( + "cancelled" if status == "interrupted" + else "failed" if status == "error" else "settled" + ), + "text": raw if isinstance(raw, str) else str(raw), + **( + {"error": str(_result_error or raw)} + if status == "error" and isinstance(result, dict) else {} + ), + }) + st.receipt_committed = True + if st.receipt_committed: + _retire_turn_marker(session, st.marker_key) + return payload, raw, status + + +def _recover_turn_exception(sid: str, session: dict, st: _TurnRun, e: BaseException) -> None: + """Except-path of the turn: crash log, history restore, terminal error frame.""" + import traceback + _append_turn_crash_log(sid, traceback.format_exc()) + print(f"[gateway-turn] {type(e).__name__}: {e}", file=sys.stderr, flush=True) + # An exception in the agent's finalizer can leave the gateway's in-memory + # history at the turn-start snapshot; keep the partial turn available to + # the next prompt (the durable inflight record still carries the + # recoverable error state). + _restore_agent_history_after_turn_error(session, st.agent) + if st.terminal_callback is not None and not st.receipt_attempted: + st.receipt_attempted = True + try: + st.terminal_callback({"status": "failed", "text": "", "error": str(e)}) + st.receipt_committed = True + except Exception: + logger.exception("hosted room terminal receipt commit failed") + try: + # Same terminal error frame shape as the returned-error path (uniform + # client handling), retaining the turn for replay. + _emit_terminal_turn_error(sid, session, e, retire_marker=st.receipt_committed) + st.error_retained = True + st.error_detail = _turn_failure_detail(e, type(e).__name__, st.prompt_text) + except Exception as emit_exc: + print( + f"[gateway-turn] terminal error emit failed: {type(emit_exc).__name__}: {emit_exc}", + file=sys.stderr, flush=True, + ) + _emit("error", sid, {"message": str(e)}) + + +def _finish_turn(sid: str, session: dict, st: _TurnRun) -> None: + """Finally-path of the turn: release everything, then the "tui turn finished" bookend.""" + agent, history, run_kwargs = st.agent, st.history, st.run_kwargs + # Drop both snapshots of the pre-turn history before asking glibc to return + # pages; session["history"] already points at the new/pruned result. + history.clear() + if isinstance(run_kwargs, dict): + run_kwargs.clear() + # While any profile-specific HERMES_HOME override is still active, so + # context.memory_trim resolves from the session's own config. + try: + from hermes_cli.mem_trim import trim_memory + trim_memory(reason="tui turn completion") + except Exception: + logger.debug("post-turn memory trim failed", exc_info=True) + if st.thinking_started: + _stop_thinking_sound() + if st.tts_queue is not None: + st.tts_queue.put(None) # end-of-text sentinel — flush + finish speaking + if st.one_turn_restore: + try: + _restore_agent_model_runtime(agent, st.one_turn_restore) + _restart_slash_worker(sid, session) + _persist_live_session_runtime(session) + _persist_live_session_system_prompt(session) + except Exception: + logger.debug("TUI one-turn model restore failed", exc_info=True) + _release_turn_scopes(st.scopes) + + +def _log_turn_finished(sid: str, session: dict, st: _TurnRun, started_monotonic: float) -> None: + """Closing bookend of "tui prompt accepted" — fires on every path, so one + accepted prompt always produces exactly one finished record. agent.session_id + is re-read because compression may have rotated it mid-turn (an + accepted/finished pair whose id changed IS a rotation trace). A missing + finished record means the thread died before the finally.""" + result = st.result + if isinstance(result, dict): + status = ( + result.get("interrupted") and "interrupted" or result.get("error") and "error" or "complete" + ) + else: + status = "error" if st.error_retained else "complete" + logger.info( + "tui turn finished: ui_session=%s session_key=%s " + "agent_session_id=%s status=%s error_retained=%s duration=%.1fs" + "%s", + sid, session.get("session_key") or "", getattr(st.agent, "session_id", "") or "", status, + st.error_retained, time.monotonic() - started_monotonic, st.error_detail, + ) + + def _run_prompt_submit( rid, sid: str, session: dict, text: Any, *, display_kind: str | None = None, display_metadata: dict | None = None, image_paths: list[str] | None = None, @@ -746,325 +1021,67 @@ def _run_prompt_submit( logger.info( "tui prompt accepted: ui_session=%s session_key=%s agent_session_id=%s " "kind=%s chars=%s images=%d", - sid, - session.get("session_key") or "", - getattr(agent, "session_id", "") or "", - display_kind or "user", - len(text) if isinstance(text, str) else "-", - len(images), + sid, session.get("session_key") or "", getattr(agent, "session_id", "") or "", + display_kind or "user", len(text) if isinstance(text, str) else "-", len(images), ) _emit("message.start", sid) def run(): - terminal_receipt_attempted = False - terminal_receipt_committed = terminal_callback is None # ContextVars from the RPC dispatcher do not follow onto this thread. # Rebind the exact transport stored on this session generation before # any tool can commission a child; delegate_task captures it as # non-serializable authority. transport_token = bind_transport(session.get("transport")) runtime_session_token = _current_runtime_session_record.set(session) - # Bound eagerly so the except/finally paths always have an agent even - # if turn setup throws; re-read after _sync_bot_capabilities, which may - # swap in a rebuilt agent for Bot Chat sessions. - agent = session["agent"] - scopes = _TurnScopes() - goal_followup = None - result = None # turn outcome; read after the finally for leftover /steer - tts_queue = None - thinking_started = False - history: list = [] - run_kwargs = None - one_turn_restore = session.pop("one_turn_model_restore", None) - # True once a failed turn's snapshot was retained for resume replay — - # tells the finally to skip the normal inflight clear. - turn_error_retained = False - # Cause for the "tui turn finished" bookend, stashed by both failure - # paths because the finally sees neither `result` nor the exception - # reliably. turn_prompt_text is what was actually submitted (post - # @-expansion, so injected file contents count), kept only so the cause - # can be checked for quoting it back (_strip_prompt_echo). - turn_error_detail = "" - turn_prompt_text = "" - marker_key = _record_turn_marker(session, text) + st = _TurnRun(session, terminal_callback) + st.marker_key = _record_turn_marker(session, text) try: - _bind_turn_scopes(sid, session, scopes) - # Skip the config-model sync while a /model --once override is - # active: the once-model is intentionally not pinned as a session - # model_override, so the sync would clobber it back to the config - # model. A config.yaml change is adopted on the NEXT turn. A model - # picked mid-turn was queued, not applied in place — apply it on - # this thread before the first model call and before the config - # sync so the explicit pick wins over a config.yaml change. - if not one_turn_restore: - _apply_pending_model_switch(sid, session) - _sync_agent_model_with_config(sid, session) - _sync_agent_compression_with_config(sid, session) - # Bot Chat: adopt Settings→Capabilities edits into the eternal bot - # session before the turn runs. No-op for other session shapes. - _sync_bot_capabilities(sid, session) - agent = session["agent"] - # Snapshot after turn-start model sync: a deferred switch mutates - # history and its version, and that mutation belongs to this turn. - with session["history_lock"]: - history = list(session["history"]) - history_version = int(session.get("history_version", 0)) - cwd = _session_cwd(session) - _register_session_cwd(session) - cols = session.get("cols", 80) - streamer = make_stream_renderer(cols) - prompt = text - if isinstance(prompt, str) and "@" in prompt: - ctx = _expand_context_references(agent, prompt, cwd) - if ctx.blocked: - _emit( - "error", sid, - {"message": "\n".join(ctx.warnings) or "Context injection refused."}, - ) - return - prompt = ctx.message - turn_prompt_text = prompt if isinstance(prompt, str) else "" - run_message: Any = _route_turn_images(agent, prompt, images) if images else prompt - tts_queue, thinking_started = _start_turn_voice() - run_message = _apply_turn_notes(run_message, session) - - def _stream(delta): - with session["history_lock"]: - _append_inflight_delta(session, delta) - payload = {"text": delta} - if streamer and (r := streamer.feed(delta)) is not None: - payload["rendered"] = r - if tts_queue is not None and isinstance(delta, str): - tts_queue.put(delta) - _emit("message.delta", sid, payload) - - # Interim assistant text (commentary beside tool calls, or the - # attempted final answer before a verify-on-stop nudge) is sealed - # by the desktop as its own segment instead of being lost when - # message.complete replaces the streaming buffer. Gated on - # display.interim_assistant_messages (default true). - if _load_interim_assistant_messages(): - def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None: - _emit("message.interim", sid, { - "text": text, - "already_streamed": already_streamed, - }) - agent.interim_assistant_callback = _interim_assistant_cb - else: - agent.interim_assistant_callback = None - run_kwargs = _build_run_kwargs( - agent, session, history, prompt, images, run_message, _stream, - display_kind, display_metadata, + prepared = _prepare_turn_input(sid, session, st, text, images) + if prepared is None: + return + prompt, run_message, cols, streamer = prepared + _invoke_agent( + sid, session, st, prompt, run_message, streamer, images, display_kind, display_metadata, ) - # Auto-titling fires inside the turn prologue; this live-rename - # hook repaints the sidebar the moment a title lands. - _title_key = session.get("session_key") or sid - agent._on_session_title = lambda t, _src, _k=_title_key: _emit( - "session.title", sid, {"session_id": _k, "title": t} + status_note = _absorb_turn_result( + sid, session, st, text, display_kind, display_metadata ) - _usage_stop, _usage_thread = _start_usage_ticker(sid, agent) - try: - result = agent.run_conversation(run_message, **run_kwargs) - finally: - # Stop AND join before anything below emits: a tick surviving - # past message.complete would roll the client's final usage - # back to a stale snapshot. The join is deliberately unbounded - # — once stop is set it only waits out one in-flight - # _get_usage/_emit, whose worst case (a stalled transport - # write) would stall the message.complete emit just the same. - _usage_stop.set() - _usage_thread.join() - if display_kind and isinstance(text, str): - _stamp_synthetic_display_kind( - agent, session, result, text, display_kind, display_metadata - ) - if "moa_one_shot_restore" in session: - _restore_moa_one_shot(sid, session) - status_note = None - if isinstance(result, dict): - if isinstance(result.get("messages"), list): - status_note = _commit_turn_history(session, result, history, history_version) - # Auto-compression inside run_conversation() may have rotated - # agent.session_id: sync session_key before title/goal/finalize - # handling uses it, keep pending_title (user intent) for the - # continuation, and restart the slash worker so worker-backed - # commands (/title etc.) target the live session. - _sync_session_key_after_compress( - sid, session, clear_pending_title=False, restart_slash_worker=True, - ) - raw, status, last_reasoning = _turn_outcome(result) - payload = {"text": raw, "usage": _get_usage(agent), "status": status} - if last_reasoning: - payload["reasoning"] = last_reasoning - if status_note: - payload["warning"] = status_note - if result.get("response_previewed"): - payload["response_previewed"] = True - # Structured billing-wall descriptor so the client renders a - # billing-specific recovery surface instead of re-parsing text. - _billing_block = result.get("billing_block") if isinstance(result, dict) else None - if _billing_block: - payload["billing"] = _billing_block - payload["failure_reason"] = result.get("failure_reason") - rendered = render_message(raw, cols) - if rendered: - payload["rendered"] = rendered - # Layer descriptor computed before the retain below so resume - # replay carries the same one (advisory; older clients ignore it). - _error_surface = _turn_error_surface(agent, result) if status == "error" else None - _result_error = result.get("error") if isinstance(result, dict) else None - with session["history_lock"]: - if status == "error": - # Retain the failed turn for resume replay: if this - # terminal frame is lost to a disconnect, resume's inflight - # payload is the only carrier of the failure. - _fail_inflight_turn( - session, _result_error if isinstance(result, dict) else raw, - error_surface=_error_surface, - ) - turn_error_retained = True - turn_error_detail = _turn_failure_detail( - (_result_error if isinstance(result, dict) else raw), - (result.get("failure_reason") if isinstance(result, dict) else None), - turn_prompt_text, - ) - else: - _clear_inflight_turn(session) - if status == "error": - payload["error"] = str((_result_error if isinstance(result, dict) else "") or raw) - payload["recoverable"] = True - if _error_surface: - payload["error_surface"] = _error_surface - if terminal_callback is not None: - terminal_receipt_attempted = True - terminal_callback( - { - "status": ( - "cancelled" - if status == "interrupted" - else "failed" if status == "error" else "settled" - ), - "text": raw if isinstance(raw, str) else str(raw), - **( - {"error": str(_result_error or raw)} - if status == "error" and isinstance(result, dict) - else {} - ), - } - ) - terminal_receipt_committed = True - if terminal_receipt_committed: - _retire_turn_marker(session, marker_key) + payload, raw, status = _complete_turn_payload(session, st, status_note, cols) _emit("message.complete", sid, payload) - goal_followup = _goal_followup_after_turn(sid, session, result, status, raw) + st.goal_followup = _goal_followup_after_turn(sid, session, st.result, status, raw) if status == "complete": _complete_loop_tick(sid, session, raw) _apply_pending_title(sid, session) # The streaming path already spoke everything via tts_queue. - if tts_queue is None and isinstance(raw, str) and raw.strip() and _voice_tts_enabled(): + if ( + st.tts_queue is None and isinstance(raw, str) and raw.strip() + and _voice_tts_enabled() + ): _speak_turn_fallback(raw) except Exception as e: - import traceback - _append_turn_crash_log(sid, traceback.format_exc()) - print(f"[gateway-turn] {type(e).__name__}: {e}", file=sys.stderr, flush=True) - # An exception in the agent's finalizer can leave the gateway's - # in-memory history at the turn-start snapshot; keep the partial - # turn available to the next prompt (the durable inflight record - # still carries the recoverable error state). - _restore_agent_history_after_turn_error(session, agent) - if terminal_callback is not None and not terminal_receipt_attempted: - terminal_receipt_attempted = True - try: - terminal_callback({"status": "failed", "text": "", "error": str(e)}) - terminal_receipt_committed = True - except Exception: - logger.exception("hosted room terminal receipt commit failed") - try: - # Same terminal error frame shape as the returned-error path - # (uniform client handling), retaining the turn for replay. - _emit_terminal_turn_error(sid, session, e, retire_marker=terminal_receipt_committed) - turn_error_retained = True - turn_error_detail = _turn_failure_detail(e, type(e).__name__, turn_prompt_text) - except Exception as emit_exc: - print( - f"[gateway-turn] terminal error emit failed: " - f"{type(emit_exc).__name__}: {emit_exc}", - file=sys.stderr, - flush=True, - ) - _emit("error", sid, {"message": str(e)}) + _recover_turn_exception(sid, session, st, e) finally: - # Drop both local snapshots of the pre-turn history before asking - # glibc to return pages; session["history"] already points at the - # new/pruned result. - history.clear() - if isinstance(run_kwargs, dict): - run_kwargs.clear() - # While any profile-specific HERMES_HOME override is still active, - # so context.memory_trim resolves from the session's own config. - try: - from hermes_cli.mem_trim import trim_memory - trim_memory(reason="tui turn completion") - except Exception: - logger.debug("post-turn memory trim failed", exc_info=True) - if thinking_started: - _stop_thinking_sound() - if tts_queue is not None: - tts_queue.put(None) # end-of-text sentinel — flush + finish speaking - if one_turn_restore: - try: - _restore_agent_model_runtime(agent, one_turn_restore) - _restart_slash_worker(sid, session) - _persist_live_session_runtime(session) - _persist_live_session_system_prompt(session) - except Exception: - logger.debug("TUI one-turn model restore failed", exc_info=True) - _release_turn_scopes(scopes) + _finish_turn(sid, session, st) _current_runtime_session_record.reset(runtime_session_token) reset_transport(transport_token) # A stale interim closure must not fire during a later turn. - agent.interim_assistant_callback = None + st.agent.interim_assistant_callback = None with session["history_lock"]: session["running"] = False session["last_active"] = time.time() - if not turn_error_retained: + if not st.error_retained: _clear_inflight_turn(session) - # Closing bookend of "tui prompt accepted" — fires on every path, - # so one accepted prompt always produces exactly one finished - # record. agent.session_id is re-read because compression may have - # rotated it mid-turn (an accepted/finished pair whose id changed IS - # a rotation trace). A missing finished record means the thread - # died before this finally. - logger.info( - "tui turn finished: ui_session=%s session_key=%s " - "agent_session_id=%s status=%s error_retained=%s duration=%.1fs" - "%s", - sid, - session.get("session_key") or "", - getattr(agent, "session_id", "") or "", - ( - result.get("interrupted") - and "interrupted" - or result.get("error") - and "error" - or "complete" - ) - if isinstance(result, dict) - else ("error" if turn_error_retained else "complete"), - turn_error_retained, - time.monotonic() - _turn_started_monotonic, - turn_error_detail, - ) + _log_turn_finished(sid, session, st, _turn_started_monotonic) # Backstop for turns that never reached a terminal frame. - if terminal_receipt_committed: - _retire_turn_marker(session, marker_key) + if st.receipt_committed: + _retire_turn_marker(session, st.marker_key) with session["history_lock"]: - if session.get("_active_turn_marker_key") == marker_key: + if session.get("_active_turn_marker_key") == st.marker_key: session.pop("_active_turn_marker_key", None) session.pop("_hosted_room_task", None) session.pop("_auto_continue_scheduled", None) - _emit_settled_session_info(sid, session, agent) - _run_post_turn_followups(rid, sid, session, result, goal_followup) + _emit_settled_session_info(sid, session, st.agent) + _run_post_turn_followups(rid, sid, session, st.result, st.goal_followup) run_thread = threading.Thread(target=run, daemon=True) with _sessions_lock: registered = _sessions.get(sid)