diff --git a/tui_gateway/methods_prompt.py b/tui_gateway/methods_prompt.py index b07b8f9f6a..50470cf147 100644 --- a/tui_gateway/methods_prompt.py +++ b/tui_gateway/methods_prompt.py @@ -37,10 +37,10 @@ def _message_row_id(msg: dict): def _mem_db_pair_agrees(mem, db_msg) -> bool: """True when a live-memory entry plausibly corresponds to a durable row. - Positional trust needs evidence beyond equal lengths: roles must match, - display-marker status must match (a marker on one side only shifts every - later position), and an addressable user turn must show the same text. - Multimodal content can't be compared cheaply — role/marker agreement suffices. + Positional trust needs more than equal lengths: roles and display-marker + status must match (a marker on one side shifts every later position), and an + addressable user turn must show the same text. Multimodal content can't be + compared cheaply — role/marker agreement suffices. """ if not isinstance(mem, dict) or not isinstance(db_msg, dict): return False @@ -72,8 +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 "") if not session_key: @@ -84,13 +83,11 @@ def _load_durable_truncation_history( if not callable(get_conv): return None history = get_conv( - session_key, repair_alternation=repair_alternation, include_row_ids=True, - ) + session_key, repair_alternation=repair_alternation, include_row_ids=True) except Exception: logger.debug( "prompt.submit: failed loading durable history for session %s", session_key, - exc_info=True, - ) + exc_info=True) return None return history if isinstance(history, list) else None @@ -98,10 +95,10 @@ def _load_durable_truncation_history( def _resolve_truncate_row_id(session: dict, history: list, target_row_id: int): """Resolve ``truncate_before_row_id`` to ``(user_ordinal, history_index)``. - Prefer in-memory ``_row_id``/``row_id`` stamps. When a live turn rewrote - ``session["history"]`` without stamps, load the durable transcript with - ``include_row_ids=True`` and map the matched user-turn ordinal onto the live - list. Never falls back to a client-supplied ordinal — unknown row ids refuse. + Prefer in-memory ``_row_id``/``row_id`` stamps; when a live turn rewrote + ``session["history"]`` without them, load the durable transcript and map the + matched user-turn ordinal onto the live list. Never falls back to a + client-supplied ordinal — unknown row ids refuse. """ hit = _find_user_turn_by_row_id(history, target_row_id) if hit is not None: @@ -109,14 +106,12 @@ def _resolve_truncate_row_id(session: dict, history: list, target_row_id: int): 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. + # Heal missing stamps only when EVERY pair agrees (all-or-nothing): the + # durable copy is alternation-repaired (may merge/drop rows) while the live + # list 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 if db_rid is not None and _message_row_id(mem) is None: @@ -150,16 +145,15 @@ def _coerce_truncate_int(rid, value, param_name="truncate_before_user_ordinal"): 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. Returns ``(ordinal, error_response)``: the target's tip-relative ordinal when the client sent none or agreed, else the 4004/4030 refusal — a stale ordinal beside a *resolved* durable id is drift; never guess which the user meant. Client ordinals count the full displayed lineage, so after compression - ``msg_ordinal + prefix_user_count`` is the SAME turn, not drift. The cut is - always aimed by the durable target, so this can never re-aim a truncation. + ``msg_ordinal + prefix_user_count`` is the SAME turn. The cut is always aimed + by the durable target, so this can never re-aim a truncation. """ if client_ordinal is None: return msg_ordinal, None @@ -167,20 +161,17 @@ def _reconcile_client_ordinal( if err is not None: return None, err if ordinal == msg_ordinal or ( - prefix_user_count > 0 and ordinal == msg_ordinal + prefix_user_count - ): + 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, f"truncate_before_user_ordinal ({ordinal}) does not match " - f"{param_name} target turn ({msg_ordinal})", - ) + f"{param_name} target turn ({msg_ordinal})") def _pending_reaction_notes(session: dict) -> str: @@ -224,10 +215,9 @@ def _pending_reaction_notes(session: dict) -> str: def _typed_stop_phrase_response(rid, text): """End the voice chat when a bare stop phrase is TYPED while backend voice mode - is on (typed twin of the spoken stop phrase, at the one server-side choke - point). Returns the RPC reply, or None when this is a normal message. The - desktop's renderer-owned voice chat never flips the backend flag and handles - its own typed stop.""" + is on (typed twin of the spoken stop phrase). Returns the RPC reply, or None + for a normal message. The desktop's renderer-owned voice chat never flips the + backend flag and handles its own typed stop.""" if not (isinstance(text, str) and _voice_mode_enabled()): return None try: @@ -255,10 +245,8 @@ def _hosted_submit_error(rid, session, hosted_task, 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) - ): + 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 @@ -276,8 +264,7 @@ def _legacy_group_fence_error(rid, session, params): try: from gateway.hosted_rooms import ( HostedRoomError, RoomProbeUnavailableError, default_db_path, probe_hosted_room, - probe_peer_room_reservation, - ) + probe_peer_room_reservation) hosted = probe_hosted_room(default_db_path(), room_id=room_id) peer = False if not hosted: @@ -286,11 +273,9 @@ def _legacy_group_fence_error(rid, session, params): requested_profile = ( (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() - ) + or str(_current_profile_name() or "default").strip()) peer = probe_peer_room_reservation( - default_db_path(), room_id=room_id, target_profile=requested_profile, - ) + default_db_path(), room_id=room_id, target_profile=requested_profile) except RoomProbeUnavailableError: return _err(rid, 5122, _GROUP_PROBE_FAILED_MSG) except HostedRoomError: @@ -302,19 +287,18 @@ def _legacy_group_fence_error(rid, session, params): if hosted or peer: owner = "its gateway" if hosted else "its home host" return _err( - rid, 4122, f"This room is managed by {owner}. 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(). + 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 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") @@ -335,23 +319,20 @@ def _parse_truncation_params(rid, sid, session, params, history): "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"), - ) + 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)", - ) + "(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. + Durability is a state.db property, not an annotation on the live copy (resume + paths historically omitted _row_id stamps). An unreadable durable state + fails 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) @@ -360,13 +341,11 @@ def _ordinal_only_truncation_error(rid, sid, session, history, user_indices, cli logger.warning( "prompt.submit: REFUSED ordinal-only truncation of durable " "session %s (ordinal=%d); truncate_before_row_id required", - sid, client_ordinal, - ) + sid, client_ordinal) return _err( rid, 4004, "ordinal-only truncation is unsafe for durable session history; " - "include truncate_before_row_id", - ) + "include truncate_before_row_id") def _resolve_truncation_ordinal(rid, sid, session, params, history): @@ -377,15 +356,13 @@ def _resolve_truncation_ordinal(rid, sid, session, params, history): drift (4030) -> ordinal-only on a durable session (4004). """ target_row_id, client_ordinal, err = _parse_truncation_params( - rid, sid, session, params, history - ) + rid, sid, session, params, history) if err is not None: return None, None, err truncate_message_id = params.get("truncate_before_message_id") - # 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 - # tip-relative ordinals can translate without loading ancestors into the tip. + # Client ordinals count the full displayed lineage; after compression the tip + # is session["history"] and ancestors live in display_history_prefix. Count + # ancestor user turns once so client and tip-relative ordinals translate. prefix_user_count = len(_history_user_indices(session.get("display_history_prefix") or [])) user_indices = _history_user_indices(history) @@ -393,12 +370,10 @@ def _resolve_truncation_ordinal(rid, sid, session, params, history): # Structured recovery fields: Desktop resyncs + retries on a stale target # and shows "compressed away" when segment_ordinal < 0 (ancestor-only). segment = ( - client_ordinal - prefix_user_count if client_ordinal is not None else resolved_ordinal - ) + client_ordinal - prefix_user_count if client_ordinal is not None else resolved_ordinal) return None, None, _err(rid, 4018, _STALE_TARGET_MSG, data={ "user_turn_count": len(user_indices), "ordinal": client_ordinal, - "segment_ordinal": segment, "prefix_user_count": prefix_user_count, - }) + "segment_ordinal": segment, "prefix_user_count": prefix_user_count}) if target_row_id is not None or truncate_message_id is not None: if target_row_id is not None: @@ -412,21 +387,17 @@ def _resolve_truncation_ordinal(rid, sid, session, params, history): ( (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, - ) + 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: " + not_found + "; refusing truncation without fallback", - target_repr, sid, - ) + target_repr, sid) return _stale() ordinal, err = _reconcile_client_ordinal( rid, sid, client_ordinal, found_match[0], param_name, target_repr, - prefix_user_count=prefix_user_count, - ) + prefix_user_count=prefix_user_count) if err is not None: return None, None, err else: @@ -434,8 +405,7 @@ def _resolve_truncation_ordinal(rid, sid, session, params, history): if ordinal < 0 or ordinal >= len(user_indices): return _stale() err = _ordinal_only_truncation_error( - rid, sid, session, history, user_indices, client_ordinal - ) + rid, sid, session, history, user_indices, client_ordinal) if err is not None: return None, None, err # Reject out-of-range on BOTH ends: a negative ordinal would hit Python's @@ -453,13 +423,11 @@ def _persist_truncation(rid, sid, session, history, truncated, ordinal, requeste """Write the truncated transcript BEFORE touching memory (fail closed). If replace_messages failed after session["history"] was rewritten, the turn - would run against the short list while state.db kept the old tail; the - append-only agent flush would then stack the new exchange on the "undone" - turns — zombie history on resume. Writes through ``_session_db`` (the db that - owns this session's row), never ``_get_db()``: a profile session's transcript - lives in its own profile's state.db. - - Returns ``(err, survivor_user_row_ids, survivor_row_id_map)``. + would run against the short list while state.db kept the old tail, and the + append-only flush would stack the new exchange on the "undone" turns — zombie + history on resume. Writes through ``_session_db`` (owner of this session's + row), never ``_get_db()``: a profile session's transcript lives in its own + profile's state.db. Returns ``(err, survivor_user_row_ids, survivor_row_id_map)``. """ survivor_user_row_ids = survivor_row_id_map = None with _session_db(session) as db: @@ -471,53 +439,44 @@ def _persist_truncation(rid, sid, session, history, truncated, ordinal, requeste old_active_row_ids = _row_ids_of(history) if requested_rebind_ids is not None: # Row-id fallback can resolve a target the live list is too - # misaligned to stamp, and repair can merge a user;user pair - # keeping only the first id: read the authoritative - # un-repaired pre-write active-id set so a rewritten row is - # never mistaken for an untouched archived/ancestor row. + # misaligned to stamp, and repair can merge a user;user pair: + # read the un-repaired pre-write active-id set so a rewritten + # row is never mistaken for an untouched archived/ancestor row. durable_rebind_history = _load_durable_truncation_history( - session, truncation_key, repair_alternation=False, - ) + session, truncation_key, repair_alternation=False) if durable_rebind_history is None: raise RuntimeError("could not load durable row identities for truncation") old_active_row_ids.update(_row_ids_of(durable_rebind_history)) old_survivor_row_ids = [_message_row_id(message) for message in truncated] - # active_only=True: in-place compaction keeps the pre-compaction - # transcript as active=0 rows under this key; a bare replace would - # DELETE that archive on every edit. archive_dropped=True: this - # write is the last step before the dropped turns are gone — - # soft-archive (active=0, still in FTS) so a mis-aimed cut is + # active_only=True: compaction keeps the pre-compaction transcript + # as active=0 rows under this key; a bare replace would DELETE that + # archive on every edit. archive_dropped=True: soft-archive the + # dropped turns (active=0, still in FTS) so a mis-aimed cut is # recoverable. db.replace_messages( truncation_key, truncated, active_only=True, archive_dropped=True, - reject_active_turn_lease=True, - ) + reject_active_turn_lease=True) except Exception as exc: logger.error( "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 - # fresh _row_id values onto these dicts. Surface the surviving - # user-turn ids (visible-user-ordinal order) so the client rebinds - # its cached rowIds — else a second rewind sends the pre-rewind id - # and the fail-closed resolver refuses with 4018. None entries mean - # the client must drop its cached id for that turn. + # replace_messages re-inserted the survivors as NEW rows with fresh + # _row_id stamps. Surface the surviving user-turn ids (visible-user- + # ordinal order) so the client rebinds its cached rowIds — else a + # second rewind sends the pre-rewind id and the resolver refuses with + # 4018. None entries: the client must drop its cached id for that turn. survivor_user_row_ids = [ - _message_row_id(truncated[i]) for i in _history_user_indices(truncated) - ] + _message_row_id(truncated[i]) for i in _history_user_indices(truncated)] if requested_rebind_ids is not None: 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) - and old_row_id in requested_rebind_ids - } + and old_row_id in requested_rebind_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 @@ -540,8 +499,7 @@ def _truncate_history_for_submit(rid, sid, session, params, requested_rebind_ids 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, "truncation would erase the entire session transcript; " @@ -550,11 +508,9 @@ def _truncate_history_for_submit(rid, sid, session, params, requested_rebind_ids log_fn = logger.warning if not truncated else logger.info log_fn( "prompt.submit: truncating session %s history %d -> %d messages (ordinal=%d)", - sid, len(history), len(truncated), ordinal, - ) + sid, len(history), len(truncated), ordinal) err, survivor_user_row_ids, survivor_row_id_map = _persist_truncation( - rid, sid, session, history, truncated, ordinal, requested_rebind_ids - ) + rid, sid, session, history, truncated, ordinal, requested_rebind_ids) if err is not None: return err, None, None session["history"] = truncated @@ -575,16 +531,15 @@ def _survivor_fields(survivor_user_row_ids, survivor_row_id_map, requested_rebin def _persist_session_row_for_submit(rid, session): """Lazily persist the DB row now that the user actually sent a message; a branch becomes real here (parent transcript copied as its seed). Returns an - error reply — the only user-visible signal; desktop maps the string to a - toast — or None. On failure the in-flight turn is released.""" + error reply (the only user-visible signal; desktop maps it to a toast) or + None. On failure the in-flight turn is released.""" try: if _ensure_session_db_row(session) is False: return _err( 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", - ) + "was not saved; repair state.db and try again") _persist_branch_seed(session) except Exception as exc: from hermes_state import is_disk_full_error @@ -612,8 +567,7 @@ def _run_after_agent_ready(rid, sid, session, text, display_kind, hosted_termina _emit_terminal_turn_error( 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}, - ) + error_surface={"layer": "runtime", "code": "agent_init_failed", "retryable": True}) with session["history_lock"]: session["running"] = False session["last_active"] = time.time() @@ -628,29 +582,26 @@ def _run_after_agent_ready(rid, sid, session, text, display_kind, hosted_termina _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" - }) + else "Session no longer running before the agent was ready"}) return _run_prompt_submit( rid, sid, session, text, display_kind=display_kind, - terminal_callback=hosted_terminal_callback, - ) + terminal_callback=hosted_terminal_callback) _TRUNCATION_PARAMS = ( - "truncate_before_user_ordinal", "truncate_before_row_id", "truncate_before_message_id", -) + "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. + """Claim the turn against a possibly-running session; returns an early RPC + reply (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. + A mid-turn prompt is queued (by default interrupting the live turn) instead of + rejected. The provider interrupt happens 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"]: @@ -660,15 +611,13 @@ def _claim_submit_slot(rid, sid, session, text, params, transport, internal_host 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")), - ) + 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 -): + 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)``.""" @@ -686,8 +635,7 @@ def _lock_in_submit_turn( ), 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 - ) + rid, sid, session, params, requested_rebind_ids) if err is not None: return err, None, None session["running"] = True @@ -756,30 +704,24 @@ def _(rid, params: dict) -> dict: requested_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(row_id, int) and not isinstance(row_id, bool)} + if isinstance(raw_rebind_ids, list) else None) 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, - ) + 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 - ) + survivor_user_row_ids, survivor_row_id_map, requested_rebind_ids) if turn_isolation: isolated_response = _submit_prompt_to_compute_host( - rid, sid, session, text, display_kind=display_kind - ) + rid, sid, session, text, display_kind=display_kind) if not isolated_response.get("error"): # The truncation already happened inline above (memory + DB). isolated_response["result"].update(survivor_fields) return isolated_response logger.warning( "compute-host dispatch failed for session %s; falling back inline: %s", sid, - isolated_response["error"].get("message", "unknown error"), - ) + 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 @@ -788,10 +730,8 @@ def _(rid, params: dict) -> dict: _start_agent_build(sid, session) run_thread = threading.Thread( target=lambda: _run_after_agent_ready( - rid, sid, session, text, display_kind, hosted_terminal_callback - ), - daemon=True, - ) + rid, sid, session, text, display_kind, hosted_terminal_callback), + daemon=True) # Handle lets session.interrupt tell a live turn from a stuck `running` flag. session["_run_thread"] = run_thread run_thread.start() @@ -805,8 +745,7 @@ def _attached_image_result(session, image_path, **extra) -> dict: """Common ``{attached, path, count, ...meta}`` reply after queuing an image.""" return { "attached": True, "path": str(image_path), "count": len(session["attached_images"]), - **extra, **_image_meta(image_path), - } + **extra, **_image_meta(image_path)} @method("clipboard.paste") @@ -822,15 +761,13 @@ def _(rid, params: dict) -> dict: 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" - ) + 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() - else "No image found in clipboard" - ) + 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)) @@ -846,8 +783,7 @@ 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"] @@ -862,8 +798,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)) @@ -882,8 +817,7 @@ def _(rid, params: dict) -> dict: 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", - ) + label="image", empty_msg="image is empty") if err is not None: return err filename = str(params.get("filename", "") or "") @@ -899,8 +833,7 @@ def _(rid, params: dict) -> dict: 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): @@ -908,8 +841,7 @@ def _pdf_attach_source(rid, params, td_path, raw_path, raw_b64): 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", - ) + label="PDF", empty_msg="decoded PDF is empty") if err is not None: return None, None, err if pdf_bytes[:5] != b"%PDF-": @@ -948,8 +880,7 @@ def _pdf_page_range(rid, params): 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" - ) + rid, 4019, f"page range exceeds cap of {_PDF_ATTACH_MAX_PAGES} pages per attach call") return first_page, last_page, None @@ -980,8 +911,7 @@ def _(rid, params: dict) -> dict: return err argv = [ "pdftoppm", "-png", "-r", "150", "-f", str(first_page), "-l", str(last_page), - str(pdf_path), str(td_path / "page"), - ] + str(pdf_path), str(td_path / "page")] from hermes_cli._subprocess_compat import windows_hide_flags try: res = subprocess.run( @@ -989,8 +919,7 @@ def _(rid, params: dict) -> dict: # UTF-8 + lossy decode: non-UTF-8 child output must not crash the # gateway thread on locale-mismatched Windows. encoding="utf-8", errors="replace", - creationflags=windows_hide_flags(), - ) + creationflags=windows_hide_flags()) except subprocess.TimeoutExpired: return _err(rid, 5028, "pdftoppm timed out (>120s)") if res.returncode != 0: @@ -1007,14 +936,12 @@ def _(rid, params: dict) -> dict: except ValueError: page_int = first_page + len(attached_pages) dst = _queue_attached_image( - session, src.read_bytes(), ".png", prefix=f"pdf_p{page_num}" - ) + 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))]", - }) + "text": f"[User attached PDF: {display_name} ({len(attached_pages)} page(s))]"}) @method("file.attach") @@ -1033,14 +960,12 @@ def _(rid, params: dict) -> dict: return _err(rid, 4015, "path or data_url required") try: stored_path, uploaded = _stage_session_file_attachment( - session, raw_path=raw, data_url=data_url, name=name - ) + 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, - }) + "uploaded": uploaded}) except Exception as e: return _err(rid, 5028, str(e)) @@ -1058,8 +983,7 @@ def _(rid, params: dict) -> dict: 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"]), - }) + "count": len(session["attached_images"])}) @method("input.detect_drop") @@ -1081,13 +1005,11 @@ def _(rid, params: dict) -> dict: "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), - }) + **_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, - }) + "text": text}) except Exception as e: return _err(rid, 5027, str(e)) @@ -1114,12 +1036,11 @@ def _final_response_text(result) -> str: def _spawn_side_agent( - rid, session, task_id, parent, event, body, *, cwd="", extra=None, cleanup=None -): + 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}``.""" + ``parent`` as ``event`` with ``task_id`` (+ ``extra``). ``cleanup`` runs in + the finally before the session context is cleared. Replies ``{task_id}``.""" extra = extra or {} def run(): @@ -1152,8 +1073,7 @@ def _(rid, params: dict) -> dict: 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, - ) + user_message=text, task_id=task_id) return _final_response_text(result) return _spawn_side_agent(rid, session, task_id, parent, "background.complete", body) @@ -1185,8 +1105,7 @@ def _(rid, params: dict) -> dict: ) or "" return _spawn_side_agent( - rid, session, task_id, parent, "btw.complete", body, extra={"question": text}, - ) + rid, session, task_id, parent, "btw.complete", body, extra={"question": text}) _PREVIEW_RESTART_RULES = ( @@ -1232,10 +1151,8 @@ def _(rid, params: dict) -> dict: f"Current working directory: {cwd or '(unknown)'}", f"Preview console:\n{context}" if context else "", _PREVIEW_RESTART_HISTORY_NOTE if parent_history else None, - *_PREVIEW_RESTART_RULES, - ] - if line - ) + *_PREVIEW_RESTART_RULES] + if line) # A malformed client path (embedded NUL, etc.) must not blow up the restart: # treat it as "no validated cwd". try: @@ -1252,12 +1169,10 @@ def _(rid, params: dict) -> dict: 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 "" - ) + if parent_history else "") _emit( "preview.restart.progress", parent, - {"task_id": task_id, "text": f"Starting hidden restart agent{history_note}"}, - ) + {"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. @@ -1265,8 +1180,7 @@ def _(rid, params: dict) -> dict: **_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, - ) + user_message=prompt, task_id=task_id, conversation_history=parent_history or None) return _final_response_text(result) def cleanup(): @@ -1278,8 +1192,7 @@ def _(rid, params: dict) -> dict: # 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, - ) + cwd=preview_cwd, cleanup=cleanup) # ── late-answer RPCs for tool-driven UI cards ─────────────────────────────── @@ -1299,8 +1212,7 @@ def _(rid, params: dict) -> dict: _LATE_RESPOND_KEYS = { "terminal.read.respond": "text", "preview.read.respond": "text", "preview.act.respond": "text", "window.read.respond": "text", "tour.respond": "text", "mcp.setup.respond": "result", - "sudo.respond": "password", "secret.respond": "value", -} + "sudo.respond": "password", "secret.respond": "value"} def _late_respond(key: str): @@ -1332,8 +1244,7 @@ def _(rid, params: dict) -> dict: if err: return err return _approval_reply( - rid, "approvals", lambda a: a.list_gateway_approvals(session["session_key"]) - ) + rid, "approvals", lambda a: a.list_gateway_approvals(session["session_key"])) @method("approval.received") @@ -1345,15 +1256,14 @@ def _(rid, params: dict) -> dict: if not isinstance(request_id, str) or not request_id: return _err(rid, 4006, "request_id required") return _approval_reply( - rid, "acknowledged", lambda a: a.ack_gateway_approval(session["session_key"], request_id), - ) + rid, "acknowledged", lambda a: a.ack_gateway_approval(session["session_key"], request_id)) def _approval_respond_session_fallback(params: dict): """Durable-identity fallback for ``approval.respond``: the desktop can answer with a stale live sid (runtime re-minted after a reconnect while the prompt - stayed on screen). Try (1) the approval ``request_id`` — unique across sessions - — against every live session's pending approvals, then (2) ``session_id`` as a + stayed on screen). Try (1) the approval ``request_id`` (unique across sessions) + against every live session's pending approvals, then (2) ``session_id`` as a STORED id mapped to its live record. Returns the live session or None.""" request_id = str(params.get("request_id") or "") if request_id: @@ -1365,8 +1275,7 @@ def _approval_respond_session_fallback(params: dict): key = str(session.get("session_key") or "") if key and any( str(pending.get("request_id") or "") == request_id - for pending in list_gateway_approvals(key) - ): + for pending in list_gateway_approvals(key)): return session except Exception: logger.debug("approval.respond request_id fallback failed", exc_info=True) @@ -1395,9 +1304,7 @@ def _(rid, params: dict) -> dict: 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"), - ), - ) + resolve_all=params.get("all", False), request_id=params.get("request_id"))) def register(server) -> None: diff --git a/tui_gateway/prompt_attachments.py b/tui_gateway/prompt_attachments.py index acb9c5a54c..60d1687ce3 100644 --- a/tui_gateway/prompt_attachments.py +++ b/tui_gateway/prompt_attachments.py @@ -20,8 +20,7 @@ _PDF_ATTACH_MAX_PAGES = 25 # 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"GIF89a", ".gif"), (b"BM", ".bmp")) # Context-ref values containing any of these must be quoted (desktop formatRefValue parity). _ATTACHMENT_REF_NEEDS_QUOTING_RE = _re.compile(r"""[\s()\[\]{}<>"'`]""") @@ -46,15 +45,13 @@ def _decode_attach_base64(raw: str, *, mime_prefix: str) -> bytes | None: import re as _re try: return _b64_payload( - raw, rf"^data:{_re.escape(mime_prefix)}[a-zA-Z0-9.+-]*;base64,(.*)$", _re.DOTALL, - ) + 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 -): + 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) @@ -192,15 +189,13 @@ def _decode_attachment_data_url(data_url: str) -> bytes: import re as _re try: return _b64_payload( - data_url, r"^data:[^;,]*(?:;[^;,=]+=[^;,]+)*;base64,(.*)$", _re.DOTALL | _re.I, - ) + 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 -) -> tuple[Path, bool]: + 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``). @@ -225,8 +220,7 @@ def _stage_session_file_attachment( payload = _decode_attachment_data_url(data_url) filename = _sanitize_attachment_name(name or Path(str(raw_path or "")).name) target = _unique_attachment_path( - _desktop_attachment_dir(session), _sanitize_attachment_name(filename) - ) + _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 280a1f0fa7..872544d7d7 100644 --- a/tui_gateway/prompt_turn.py +++ b/tui_gateway/prompt_turn.py @@ -25,8 +25,7 @@ def _is_successful_goal_turn(result: Any, status: str, raw: Any) -> bool: return bool( 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) - ) + and not (isinstance(result, dict) and result.get("completed") is False)) def _goal_max_turns() -> int: @@ -38,13 +37,12 @@ def _goal_max_turns() -> int: def _plan_goal_compression_recovery( - session: dict, result: Any, *, status: str, raw: Any, -) -> tuple[str | None, str | None]: + session: dict, result: Any, *, status: str, raw: Any) -> tuple[str | None, str | None]: """Plan a bounded active-goal retry after compression exhaustion. - 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 + 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")) @@ -67,8 +65,7 @@ def _plan_goal_compression_recovery( if ( isinstance(recovery_state, dict) and recovery_state.get("goal_created_at") == goal_created_at - and recovery_state.get("goal") == goal_text - ): + and recovery_state.get("goal") == goal_text): try: attempts = int(recovery_state.get("attempts", 0) or 0) except (TypeError, ValueError): @@ -76,11 +73,9 @@ 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": goal_text, "attempts": attempts + 1, - } + "goal_created_at": goal_created_at, "goal": goal_text, "attempts": attempts + 1} return ( - continuation_prompt, - "Context compression was exhausted. Retrying the active goal once.", + continuation_prompt, "Context compression was exhausted. Retrying the active goal once." ) goal_mgr.pause(reason="context compression exhausted twice consecutively") # A later explicit /goal resume gets a fresh bounded recovery cycle. @@ -88,8 +83,7 @@ def _plan_goal_compression_recovery( return ( None, "Goal paused after context compression was exhausted twice. " - "Run /compress, then /goal resume to continue.", - ) + "Run /compress, then /goal resume to continue.") # ── turn admission ─────────────────────────────────────────────────── @@ -97,21 +91,19 @@ def _plan_goal_compression_recovery( def _admit_prompt_turn( sid: str, session: dict, text: Any, image_paths: list[str] | None, - queued_prompt_generation: int | None, -) -> tuple[list[str], Any] | None: + queued_prompt_generation: int | None) -> tuple[list[str], Any] | None: """Ownership + liveness gate every fresh turn source must cross. - 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 + prompt.submit claims the slot in its RPC handler, but 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( "Refusing turn for session %s at _run_prompt_submit: %s", session.get("session_key") or sid, - getattr(ownership_refusal, "reason", None) or "refused", - ) + getattr(ownership_refusal, "reason", None) or "refused") with session["history_lock"]: session["running"] = False _emit("error", sid, {"message": str(ownership_refusal)}) @@ -119,8 +111,7 @@ def _admit_prompt_turn( with session["history_lock"]: if session.get("_closing") or ( queued_prompt_generation is not None - and int(session.get("_queued_prompt_generation", 0)) != queued_prompt_generation - ): + and int(session.get("_queued_prompt_generation", 0)) != queued_prompt_generation): session["running"] = False return None if image_paths is None: @@ -143,12 +134,12 @@ def _admit_prompt_turn( def _record_turn_marker(session: dict, text: Any) -> str: """Write the durable crash marker; returns the session key it was written under. - 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. + Retired when the outcome reaches the client; 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 (Stop landed first, no + file to clear yet). """ marker_home = _session_home(session) marker_key = str(session.get("session_key") or "") @@ -184,12 +175,10 @@ class _TurnScopes: def _bind_turn_scopes(sid: str, session: dict, scopes: _TurnScopes) -> None: """Bind approval/session/profile/terminal scopes for this turn thread. - Fills ``scopes`` field by field so a failure midway still leaves every - already-bound token for ``_release_turn_scopes``. The profile's COMPLETE - terminal policy is bound too (dashboard/TUI analogue of the gateway's - per-turn scope): terminal_tool otherwise reads the launch process's pinned - env, and a failed install leaves a refusal scope so terminal tools fail - closed instead of inheriting ambient policy. + Fills ``scopes`` field by field so a failure midway still leaves every bound + token for ``_release_turn_scopes``. The profile's COMPLETE terminal policy is + bound too: terminal_tool otherwise reads the launch process's pinned env, and + a failed install leaves a refusal scope so terminal tools fail closed. """ from tools.approval import set_current_session_key scopes.approval = set_current_session_key(session["session_key"]) @@ -200,11 +189,10 @@ def _bind_turn_scopes(sid: str, session: dict, scopes: _TurnScopes) -> None: scopes.secret = set_secret_scope(build_profile_secret_scope(Path(profile_home))) from tools.terminal_scope import install_profile_terminal_scope scopes.terminal = install_profile_terminal_scope(Path(profile_home)) - # The sudo password callback is thread-local (tools.terminal_tool - # _callback_tls), so the build thread's wiring doesn't reach this turn - # thread — sudo prompts would fall through to /dev/tty and hang the - # headless gateway. Re-wire so they route to the sudo.request overlay - # (secret capture is a module global, so re-running is a no-op). + # The sudo password callback is thread-local, so the build thread's wiring + # doesn't reach this turn thread — sudo prompts would fall through to + # /dev/tty and hang the headless gateway. Re-wire to the sudo.request + # overlay (secret capture is a module global; re-running is a no-op). _wire_callbacks(sid) @@ -234,18 +222,16 @@ def _expand_context_references(agent, prompt: str, cwd: str): getattr(agent, "model", "") or _resolve_model(), base_url=getattr(agent, "base_url", "") or "", api_key=getattr(agent, "api_key", "") or "", provider=getattr(agent, "provider", "") or "", - config_context_length=getattr(agent, "_config_context_length", None), - ) + config_context_length=getattr(agent, "_config_context_length", None)) return preprocess_context_references(prompt, cwd=cwd, allowed_root=cwd, context_length=ctx_len) def _route_turn_images(agent, prompt: Any, images: list[str]) -> Any: """Build the run message for a turn with attached images. - "native" passes pixels to the main model as OpenAI-style content parts - (adapters translate per provider); "text" references the paths so the - agent analyzes them in-loop with vision_analyze, never blocking the submit - path on vision calls. Decision table: agent/image_routing.py. + "native" passes pixels as OpenAI-style content parts; "text" references the + paths so the agent analyzes them in-loop with vision_analyze, never blocking + the submit path on vision calls. Decision table: agent/image_routing.py. """ try: from agent.image_routing import build_native_content_parts, decide_image_input_mode @@ -253,15 +239,13 @@ def _route_turn_images(agent, prompt: Any, images: list[str]) -> Any: _provider, _model = _active_image_routing_identity(agent) mode = decide_image_input_mode( _provider, _model, _tui_load_config(), - requested_provider=getattr(agent, "requested_provider", ""), - ) + requested_provider=getattr(agent, "requested_provider", "")) if getattr(agent, "api_mode", "") == "codex_app_server": mode = "text" except Exception as _img_exc: print( f"[tui_gateway] image_routing decision failed, defaulting to text: {_img_exc}", - file=sys.stderr, - ) + file=sys.stderr) mode = "text" if mode != "native": return _build_image_ref_message(prompt, images) @@ -270,13 +254,13 @@ def _route_turn_images(agent, prompt: Any, images: list[str]) -> Any: if skipped: print( f"[tui_gateway] native image attachment skipped {len(skipped)} unreadable path(s)", - file=sys.stderr, - ) + file=sys.stderr) if any(p.get("type") == "image_url" for p in parts): 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) @@ -284,13 +268,11 @@ 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)``. - ``_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_stream_begin`` goes first: cutting a still-speaking previous turn 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. The + "thinking" sound keeps long silences from reading as a dead session; its + gate skips while TTS plays or the mic captures; stopped in the turn's finally. """ tts_queue = _tts_stream_begin() if not _voice_mode_enabled(): @@ -320,13 +302,10 @@ def _stop_thinking_sound() -> None: 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 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). - """ + """Prepend the per-turn API-message notes (same enrichment channel as images): + barge mid-speech, reactions since the last turn, then which window the message + was typed into (HUD mode is per-turn state; it cannot live in the byte-stable + system prompt).""" from tools.tts_streaming import SPEECH_INTERRUPTED_NOTE, take_speech_interrupted if take_speech_interrupted(): run_message = _prepend_note(run_message, SPEECH_INTERRUPTED_NOTE) @@ -336,22 +315,18 @@ def _apply_turn_notes(run_message: Any, session: dict) -> Any: def _build_run_kwargs( agent, session: dict, history: list, prompt: Any, images: list[str], run_message: Any, - stream_cb, display_kind: str | None, display_metadata: dict | None, -) -> dict: + stream_cb, display_kind: str | None, display_metadata: dict | None) -> dict: """Assemble ``run_conversation`` kwargs, feature-detecting optional parameters. - A synthesized turn is typed at turn START so the crash persist writes its - row as a timeline event instead of a raw user bubble (forever, if the turn - never ends — the auto-continue case). The post-turn stamp is the fallback - for an older agent without the parameter; re-stamping is a no-op. + A synthesized turn is typed at turn START so the crash persist writes its row + as a timeline event, not a raw user bubble (forever, if the turn never ends). + The post-turn stamp is the fallback for an older agent; re-stamping is a no-op. """ run_kwargs = { "conversation_history": list(history), "stream_callback": stream_cb, "persist_user_message": ( - _build_persist_user_message(prompt, images, run_message) if images else prompt - ), - } + _build_persist_user_message(prompt, images, run_message) if images else prompt)} try: run_params = inspect.signature(agent.run_conversation).parameters except (TypeError, ValueError): @@ -377,8 +352,7 @@ def _stamp_synthetic_display_kind( try: db.set_latest_matching_message_display_kind( current_session_id, role="user", content=text, display_kind=display_kind, - display_metadata=display_metadata, - ) + display_metadata=display_metadata) except Exception: logger.debug("failed to stamp synthetic display kind", exc_info=True) if isinstance(result, dict) and isinstance(result.get("messages"), list): @@ -391,12 +365,9 @@ def _stamp_synthetic_display_kind( def _restore_moa_one_shot(sid: str, session: dict) -> None: - """Undo a /moa one-shot after its turn. - - The one-shot did a real in-place ``agent.switch_model()`` to MoA, so undoing - it must go back through the switch path — resetting ``model_override`` - alone would leave the live client pinned to MoA for the next turn. - """ + """Undo a /moa one-shot after its turn — through the switch path, because the + one-shot did a real in-place ``agent.switch_model()``; resetting + ``model_override`` alone would leave the live client pinned to MoA.""" _restore = session.pop("moa_one_shot_restore", None) if isinstance(_restore, dict): _prev_override = _restore.get("override") @@ -423,18 +394,15 @@ def _restore_moa_one_shot(sid: str, session: dict) -> None: def _commit_turn_history( - session: dict, result: dict, history: list, history_version: int -) -> str | None: + session: dict, result: dict, history: list, history_version: int) -> str | None: """Write the agent's messages back to session history; returns a client warning or None. Caller holds no lock. If history_version moved during the turn, the only tolerated mutation is a pivot marker the gateway itself inserted mid-turn - (model switch, or /personality which lands immediately with no pending - queue); then the agent output is merged after the current history. - ``_append_model_switch_marker`` strips prior markers in place then appends, - so the delta is NOT a tail slice — compare content, not indices. Any other - desync (undo/compress/retry/rollback) is surfaced instead of silently - dropping the output. + (model switch, /personality); then the output is merged after the current + history. ``_append_model_switch_marker`` strips prior markers in place then + appends, so the delta is NOT a tail slice — compare content, not indices. + Any other desync (undo/compress/retry/rollback) is surfaced, never dropped. """ with session["history_lock"]: current_version = int(session.get("history_version", 0)) @@ -446,8 +414,7 @@ def _commit_turn_history( history_no_markers = [e for e in history if not _is_pivot_marker(e)] current_no_markers = [e for e in current_history if not _is_pivot_marker(e)] pivot_only = current_no_markers == history_no_markers and any( - _is_pivot_marker(e) for e in current_history - ) + _is_pivot_marker(e) for e in current_history) if pivot_only: # Auto-compression can make result["messages"] shorter than the # turn-start history; then the full result is the base. @@ -462,12 +429,10 @@ def _commit_turn_history( f"[tui_gateway] prompt.submit: history_version mismatch " f"(expected={history_version} current={current_version}) — " f"agent output NOT written to session history", - file=sys.stderr, - ) + file=sys.stderr) return ( "History changed during this turn — the response above is visible " - "but was not saved to session history." - ) + "but was not saved to session history.") def _turn_outcome(result: Any) -> tuple[Any, str, str | None]: @@ -477,8 +442,7 @@ def _turn_outcome(result: Any) -> tuple[Any, str, str | None]: raw = result.get("final_response", "") status = ( "interrupted" if result.get("interrupted") - else "error" if result.get("error") else "complete" - ) + else "error" if result.get("error") else "complete") # 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. @@ -487,8 +451,7 @@ def _turn_outcome(result: Any) -> tuple[Any, str, str | None]: # "Operation interrupted: waiting for model response (…)" is cancellation # metadata, not assistant prose (gateway/run.py and ACP suppress it too). if status == "interrupted" and isinstance(raw, str) and raw.strip().startswith( - INTERRUPT_WAITING_FOR_MODEL_PREFIX - ): + INTERRUPT_WAITING_FOR_MODEL_PREFIX): raw = "" lr = result.get("last_reasoning") last_reasoning = lr.strip() if isinstance(lr, str) and lr.strip() else None @@ -496,13 +459,12 @@ 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).""" + """{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( result, provider=str(getattr(agent, "provider", "") or ""), - model=str(getattr(agent, "model", "") or ""), - ) + model=str(getattr(agent, "model", "") or "")) except Exception: return None @@ -511,22 +473,19 @@ def _turn_error_surface(agent, result: Any) -> Any: 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). + sid: str, session: dict, result: Any, status: str, raw: Any) -> str | None: + """/goal continuation (mirrors gateway/run._post_turn_goal_continuation). - Asks the judge whether the goal is done and, if not and still under budget, - returns the continuation prompt to chain once this thread releases - ``running``. The verdict is surfaced as a status line either way. - Compression failures are never judge input: the error text is not work - toward the goal, and evaluating it would spend a turn. + Asks the judge whether the goal is done and, if not and under budget, returns + the continuation prompt to chain once ``running`` is released. The verdict is + surfaced as a status line either way. Compression failures are never judge + input: the error text is not work toward the goal, and judging it spends a turn. """ goal_followup = None compression_exhausted = bool(isinstance(result, dict) and result.get("compression_exhausted")) try: recovery_prompt, recovery_notice = _plan_goal_compression_recovery( - session, result, status=status, raw=raw - ) + session, result, status=status, raw=raw) if recovery_notice: _emit("status.update", sid, {"kind": "goal", "text": recovery_notice}) if recovery_prompt: @@ -547,13 +506,11 @@ def _goal_followup_after_turn( except Exception: _bg_procs = None decision = goal_mgr.evaluate_after_turn( - raw, user_initiated=True, background_processes=_bg_procs - ) + 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 "" - ): + cont_prompt := decision.get("continuation_prompt") or ""): goal_followup = cont_prompt except Exception as _goal_exc: _hook_failure("goal continuation hook", _goal_exc) @@ -611,8 +568,7 @@ def _append_turn_crash_log(sid: str, trace: str) -> None: with open(_CRASH_LOG, "a", encoding="utf-8") as f: f.write( f"\n=== turn-dispatcher exception · " - f"{time.strftime('%Y-%m-%d %H:%M:%S')} · sid={sid} ===\n" - ) + f"{time.strftime('%Y-%m-%d %H:%M:%S')} · sid={sid} ===\n") f.write(trace) @@ -634,19 +590,15 @@ def _dispatch_followup_turn(rid, sid: str, session: dict, prompt: Any, what: str def _run_post_turn_followups( - rid, sid: str, session: dict, result: Any, goal_followup: str | None -) -> None: + 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 - every auto follow-up — drain it and skip the rest this cycle (the goal - judge / notifications re-evaluate after that turn). A leftover /steer the - agent couldn't inject (arrived during the final API call) is requeued - first so it isn't dropped; a real queued prompt still wins because - ``_enqueue_prompt`` merges both texts. Then the goal continuation, then - completion notifications that arrived mid-turn. Each nested - ``_run_prompt_submit`` checks ``running`` under the lock first, so a - racing user prompt (which sets running=True) wins. + Order: a user prompt that arrived mid-turn wins over every auto follow-up — + drain it and skip the rest this cycle. A leftover /steer the agent couldn't + inject is requeued first so it isn't dropped (a real queued prompt still wins: + ``_enqueue_prompt`` merges both). Then the goal continuation, then completion + notifications that arrived mid-turn. Each nested ``_run_prompt_submit`` checks + ``running`` under the lock first, so a racing user prompt wins. """ _leftover_steer = result.get("pending_steer") if isinstance(result, dict) else None if isinstance(_leftover_steer, str) and _leftover_steer.strip(): @@ -661,19 +613,17 @@ def _run_post_turn_followups( session["running"] = True _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 - # compression-chain aware (same fail-closed gate as the poller): a turn - # finishing in session B must not consume session A's event, and a - # post-compression session still claims its pre-compression dispatches. - # Everything this session cannot claim is requeued for the poller. + # Safety net for completion events that arrived mid-turn (the poller handles + # between-turn delivery). Ownership is positive-proof and compression-chain + # aware (same fail-closed gate as the poller): session B must not consume + # session A's event; a post-compression session still claims its + # pre-compression dispatches. Unclaimable events are requeued for the poller. try: from tools.process_registry import process_registry drained = process_registry.drain_notifications( session_key=session.get("session_key", ""), owns_event=lambda e: _session_owns_notification_event(sid, session, e), - skip_poll_observed=False, - ) + skip_poll_observed=False) for index, (_evt, synth) in enumerate(drained): with session["history_lock"]: if session.get("running"): @@ -682,16 +632,14 @@ def _run_post_turn_followups( break session["running"] = True from tools.async_delegation import ( - claim_event_delivery, complete_event_delivery, release_event_delivery, - ) + claim_event_delivery, complete_event_delivery, release_event_delivery) _claim = claim_event_delivery(_evt, "tui-post-turn") if _claim is None: continue _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), - ) + on_error=lambda: release_event_delivery(_evt, _claim)) except Exception as _drain_exc: _hook_failure("completion queue drain", _drain_exc) @@ -702,21 +650,19 @@ def _run_post_turn_followups( 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. + ``agent`` is bound eagerly so except/finally always have one even if setup + throws; re-read after ``_sync_bot_capabilities`` (may swap in a rebuilt Bot + Chat agent). ``error_retained``: the finally skips the inflight clear (the + failed snapshot stays for resume replay). ``error_detail``: 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 + 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", - ) + "prompt_text", "marker_key", "terminal_callback", "receipt_attempted", "receipt_committed") def __init__(self, session: dict, terminal_callback) -> None: self.agent = session["agent"] @@ -741,13 +687,12 @@ class _TurnRun: 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. + 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 model_override, so the sync would clobber it (a config.yaml change is + adopted NEXT turn). A model picked mid-turn was queued, not applied — apply + it 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: @@ -771,8 +716,7 @@ def _prepare_turn_input(sid: str, session: dict, st: _TurnRun, text: Any, images ctx = _expand_context_references(agent, prompt, cwd) if ctx.blocked: _emit( - "error", sid, {"message": "\n".join(ctx.warnings) or "Context injection refused."}, - ) + "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 "" @@ -783,8 +727,7 @@ def _prepare_turn_input(sid: str, session: dict, st: _TurnRun, text: Any, images 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: + 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 @@ -798,10 +741,10 @@ def _invoke_agent( 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). + # Interim assistant text (commentary beside tool calls, or a pre-nudge final + # answer) 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}) @@ -810,24 +753,21 @@ def _invoke_agent( agent.interim_assistant_callback = None st.run_kwargs = _build_run_kwargs( agent, session, st.history, prompt, images, run_message, _stream, display_kind, - display_metadata, - ) + 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} - ) + "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. + # 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() @@ -839,8 +779,7 @@ def _absorb_turn_result( result = st.result if display_kind and isinstance(text, str): _stamp_synthetic_display_kind( - st.agent, session, result, text, display_kind, display_metadata - ) + 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 @@ -848,13 +787,11 @@ def _absorb_turn_result( 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. + # agent.session_id: sync session_key before title/goal/finalize use it, + # keep pending_title (user intent), 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, - ) + sid, session, clear_pending_title=False, restart_slash_worker=True) return status_note @@ -892,8 +829,7 @@ def _complete_turn_payload(session: dict, st: _TurnRun, status_note: str | None, 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, - ) + st.prompt_text) else: _clear_inflight_turn(session) if status == "error": @@ -906,14 +842,11 @@ def _complete_turn_payload(session: dict, st: _TurnRun, status_note: str | None, st.terminal_callback({ "status": ( "cancelled" if status == "interrupted" - else "failed" if status == "error" else "settled" - ), + 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 {} - ), - }) + if status == "error" and isinstance(result, dict) else {})}) st.receipt_committed = True if st.receipt_committed: _retire_turn_marker(session, st.marker_key) @@ -946,8 +879,7 @@ def _recover_turn_exception(sid: str, session: dict, st: _TurnRun, e: BaseExcept except Exception as emit_exc: print( f"[gateway-turn] terminal error emit failed: {type(emit_exc).__name__}: {emit_exc}", - file=sys.stderr, flush=True, - ) + file=sys.stderr, flush=True) _emit("error", sid, {"message": str(e)}) @@ -983,14 +915,15 @@ def _finish_turn(sid: str, session: dict, st: _TurnRun) -> None: 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.""" + accepted prompt 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" + result.get("interrupted") and "interrupted" + or result.get("error") and "error" or "complete" ) else: status = "error" if st.error_retained else "complete" @@ -999,38 +932,34 @@ def _log_turn_finished(sid: str, session: dict, st: _TurnRun, started_monotonic: "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, - ) + 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, queued_prompt_generation: int | None = None, - terminal_callback: Callable[[dict[str, Any]], None] | None = None, -) -> bool: + terminal_callback: Callable[[dict[str, Any]], None] | None = None) -> bool: admitted = _admit_prompt_turn(sid, session, text, image_paths, queued_prompt_generation) if admitted is None: return False images, agent = admitted # The ONE INFO record proving a Desktop/TUI prompt was accepted by THIS - # process; ties together the UI session id, the gateway session_key and - # the agent's live session_id (compression rotates the last independently), - # which is what a rotation-mute trace needs. No prompt content is logged. + # process; ties the UI session id, gateway session_key and the agent's live + # session_id (compression rotates the last independently) together for a + # rotation-mute trace. No prompt content is logged. _turn_started_monotonic = time.monotonic() 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), - ) + display_kind or "user", len(text) if isinstance(text, str) else "-", len(images)) _emit("message.start", sid) def run(): - # 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. + # 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 authority). transport_token = bind_transport(session.get("transport")) runtime_session_token = _current_runtime_session_record.set(session) st = _TurnRun(session, terminal_callback) @@ -1041,11 +970,11 @@ def _run_prompt_submit( return prompt, run_message, cols, streamer = prepared _invoke_agent( - sid, session, st, prompt, run_message, streamer, images, display_kind, display_metadata, + sid, session, st, prompt, run_message, streamer, images, display_kind, + display_metadata, ) status_note = _absorb_turn_result( - sid, session, st, text, display_kind, display_metadata - ) + sid, session, st, text, display_kind, display_metadata) payload, raw, status = _complete_turn_payload(session, st, status_note, cols) _emit("message.complete", sid, payload) st.goal_followup = _goal_followup_after_turn(sid, session, st.result, status, raw) @@ -1055,8 +984,7 @@ def _run_prompt_submit( # The streaming path already spoke everything via tts_queue. if ( st.tts_queue is None and isinstance(raw, str) and raw.strip() - and _voice_tts_enabled() - ): + and _voice_tts_enabled()): _speak_turn_fallback(raw) except Exception as e: _recover_turn_exception(sid, session, st, e)