From 187e35950db5ea17ffa7d02b4f51c7b9cf76f967 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Thu, 3 Sep 2026 02:36:02 -0700 Subject: [PATCH] refactor(tui_gateway): predicate-collapse hosted proof, table-driven receipt status, compact attachment staging --- tui_gateway/methods_prompt.py | 57 +++++++++++++------------------ tui_gateway/prompt_attachments.py | 22 ++++-------- tui_gateway/prompt_turn.py | 43 +++++++++-------------- 3 files changed, 48 insertions(+), 74 deletions(-) diff --git a/tui_gateway/methods_prompt.py b/tui_gateway/methods_prompt.py index f052258f3c..99208d1f3c 100644 --- a/tui_gateway/methods_prompt.py +++ b/tui_gateway/methods_prompt.py @@ -161,7 +161,6 @@ def _pending_reaction_notes(session: dict) -> str: # ── prompt.submit pieces ──────────────────────────────────────────────────── - def _typed_stop_phrase_response(rid, text): """RPC reply ending the voice chat when a bare stop phrase is TYPED while backend voice mode is on (typed twin of the spoken stop phrase), or None for a normal message.""" @@ -186,15 +185,13 @@ 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) - 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 + valid = ( + isinstance(hosted_task, dict) and callable(hosted_terminal_callback) + and set(hosted_task) == _HOSTED_TASK_FIELDS + and all(isinstance(hosted_task.get(f), str) and hosted_task[f] + for f in _HOSTED_TASK_FIELDS - {"execution_generation"}) + and isinstance(hosted_task.get("execution_generation"), int)) + return None if valid else _err(rid, 4120, "invalid hosted room turn proof") def _legacy_group_fence_error(rid, session, params): @@ -214,12 +211,11 @@ def _legacy_group_fence_error(rid, session, params): 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 "") - or str(params.get("profile") or "").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=( + (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())) except RoomProbeUnavailableError: return _err(rid, 5122, _GROUP_PROBE_FAILED_MSG) except HostedRoomError: @@ -468,10 +464,10 @@ def _run_after_agent_ready(rid, sid, session, text, display_kind, hosted_termina session["running"] = False _clear_inflight_turn(session) # Without this emit the turn vanishes silently after {"status": "streaming"}. - _emit("error", sid, { - "message": "Turn cancelled 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"}) + else "Session no longer running before the agent was ready")}) return _run_prompt_submit( rid, sid, session, text, display_kind=display_kind, @@ -495,7 +491,8 @@ def _lock_in_submit_turn( 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"), fields + "confirm_truncate requires truncate_before_user_ordinal, truncate_before_message_id, or truncate_before_row_id", + ), fields if has_truncation: err, fields = _truncate_history_for_submit( rid, sid, session, params, requested_rebind_ids) @@ -606,7 +603,6 @@ def _(rid, params: dict) -> dict: # ── attachments ───────────────────────────────────────────────────────────── - def _attached_image_result(session, image_path, **extra) -> dict: """Common ``{attached, path, count, ...meta}`` reply after queuing an image.""" return { @@ -789,7 +785,10 @@ def _(rid, params: dict) -> dict: attached_pages = [] for src in rendered: page_num = src.stem.split("-", 1)[-1] - page_int = int(page_num) if page_num.isdigit() else first_page + len(attached_pages) + try: + 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}") attached_pages.append({"path": str(dst), "page": page_int, **_image_meta(dst)}) @@ -806,9 +805,8 @@ def _(rid, params: dict) -> dict: session, err = _sess_building(params, rid) if err: return err - raw = str(params.get("path", "") or "").strip() - data_url = str(params.get("data_url", "") or "").strip() - name = str(params.get("name", "") or "").strip() + raw, data_url, name = ( + str(params.get(k, "") or "").strip() for k in ("path", "data_url", "name")) if not raw and not data_url: return _err(rid, 4015, "path or data_url required") try: @@ -866,7 +864,6 @@ def _(rid, params: dict) -> dict: # ── side agents (background / btw / preview.restart) ──────────────────────── - def _final_response_text(result) -> str: return (result.get("final_response", str(result)) if isinstance(result, dict) else str(result)) @@ -975,9 +972,7 @@ 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() + url, cwd, context = (str(params.get(k) or "").strip() for k in ("url", "cwd", "context")) if not url: return _err(rid, 4012, "url required") task_id = f"preview_{uuid.uuid4().hex[:6]}" @@ -1048,8 +1043,6 @@ _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"} - - for _name, _key in _LATE_RESPOND_KEYS.items(): method(_name)(lambda rid, params, _k=_key: _respond(rid, params, _k, allow_expired=True)) del _name, _key @@ -1057,7 +1050,6 @@ del _name, _key # ── approvals ─────────────────────────────────────────────────────────────── - def _approval_reply(rid, result_key, call): """``_ok({result_key: call(tools.approval)})``, 5004 on any failure.""" try: @@ -1081,8 +1073,7 @@ def _(rid, params: dict) -> dict: session, err = _sess(params, rid) if err: return err - request_id = params.get("request_id") - if not isinstance(request_id, str) or not request_id: + if not isinstance(request_id := params.get("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)) diff --git a/tui_gateway/prompt_attachments.py b/tui_gateway/prompt_attachments.py index 36290d8147..bdb749e4ef 100644 --- a/tui_gateway/prompt_attachments.py +++ b/tui_gateway/prompt_attachments.py @@ -28,13 +28,11 @@ 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).""" + """Strip an optional ``data:...;base64,`` wrapper and all whitespace, then strictly decode.""" import base64 as _base64 import re as _re cleaned = (raw or "").strip() - m = _re.match(data_url_re, cleaned, flags) - if m: + if m := _re.match(data_url_re, cleaned, flags): cleaned = m.group(1) return _base64.b64decode(_re.sub(r"\s+", "", cleaned), validate=True) @@ -64,8 +62,7 @@ def _decode_attach_payload( 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``.""" + """Extension from the filename hint, else magic bytes (WebP: RIFF container), else ``.png``.""" if filename and (suffix := Path(filename).suffix.lower()): return suffix head = img_bytes[:16] @@ -112,8 +109,7 @@ def _queue_attached_image(session: dict, img_bytes: bytes, ext: str, *, prefix: def _format_ref_value(value: str) -> str: - """Quote a context-ref value containing whitespace/brackets/quotes so the staged - ``@file:`` ref round-trips through ``agent.context_references``.""" + """Quote a value with whitespace/brackets/quotes so the ``@file:`` ref round-trips.""" if not value or not _ATTACHMENT_REF_NEEDS_QUOTING_RE.search(value): return value for q in ("`", '"', "'"): @@ -140,13 +136,9 @@ def _sanitize_attachment_name(name: str) -> str: def _stage_session_file_attachment( session: dict, *, raw_path: str, data_url: str, name: str) -> tuple[Path, bool]: """Make a desktop file attachment available to the gateway agent: ``(stored_path, uploaded)``. - - 1. Path resolves INSIDE the session workspace -> use as-is (``uploaded=False``). - 2. Gateway-visible file OUTSIDE the workspace -> copy into ``attachments/`` (registered - in ``tools.credential_files._CACHE_DIRS`` and bind-mounted into container backends) - so ``@file:`` resolves in the sandbox. - 3. Not on the gateway (remote client disk) -> decode ``data_url`` bytes into ``attachments/``. - """ + Inside the workspace -> as-is; gateway-visible but outside -> copied into ``attachments/`` + (bind-mounted into container backends so ``@file:`` resolves in the sandbox); not on the + gateway -> ``data_url`` bytes decoded into ``attachments/``.""" workspace = Path(_session_cwd(session)).resolve() resolved = None if raw_path: diff --git a/tui_gateway/prompt_turn.py b/tui_gateway/prompt_turn.py index 68460ac420..fa1235cbf9 100644 --- a/tui_gateway/prompt_turn.py +++ b/tui_gateway/prompt_turn.py @@ -74,8 +74,7 @@ def _plan_goal_compression_recovery( goal_mgr.pause(reason="context compression exhausted twice consecutively") # A later explicit /goal resume gets a fresh bounded recovery cycle. session.pop(_GOAL_COMPRESSION_RECOVERY_ATTEMPTS, None) - return ( - None, + return None, ( "Goal paused after context compression was exhausted twice. " "Run /compress, then /goal resume to continue.") @@ -505,18 +504,14 @@ def _invoke_agent( st.tts_queue.put(delta) _emit("message.delta", sid, payload) - # Interim assistant text (commentary beside tool calls, pre-nudge final answer) is - # sealed by the desktop as its own segment instead of being lost when - # message.complete replaces the streaming buffer. - 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 - # Feature-detect optional run_conversation parameters. A synthesized turn is typed at - # turn START so a crash persist writes a timeline event, not a raw user bubble; the - # post-turn stamp is the fallback for an older agent. + # Interim assistant text (commentary beside tool calls, pre-nudge final answer) is sealed + # by the desktop as its own segment instead of being lost to message.complete. + 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 if _load_interim_assistant_messages() else None) + # A synthesized turn is typed at turn START so a crash persist writes a timeline event, + # not a raw user bubble; the post-turn stamp is the fallback for an older agent. st.run_kwargs = run_kwargs = { "conversation_history": list(st.history), "stream_callback": _stream, @@ -539,9 +534,8 @@ def _invoke_agent( try: st.result = agent.run_conversation(run_message, **st.run_kwargs) finally: - # Stop AND join before anything emits: a tick surviving past message.complete - # would roll the client's usage back to a stale snapshot. The unbounded join only - # waits out one in-flight _get_usage/_emit (same worst case as the emit itself). + # Stop AND join before anything emits: a tick surviving past message.complete would + # roll the client's usage back to a stale snapshot (unbounded join: same worst case). _usage_stop.set() _usage_thread.join() @@ -625,8 +619,8 @@ def _complete_turn_payload(session: dict, st: _TurnRun, status_note: str | None, payload["failure_reason"] = result.get("failure_reason") if rendered := render_message(raw, cols): payload["rendered"] = rendered - # {layer, code, retryable} descriptor (advisory), computed before the retain below so - # resume replay carries the same one. + # Advisory {layer, code, retryable} descriptor; computed before the retain so resume + # replay carries the same one. _error_surface = None if status == "error": try: @@ -639,8 +633,8 @@ def _complete_turn_payload(session: dict, st: _TurnRun, status_note: str | None, error_value = result.get("error") with session["history_lock"]: if status == "error": - # Retain the failed turn: if this frame is lost to a disconnect, resume's - # inflight payload is the only carrier of the failure. + # Retain the failed turn: resume's inflight payload is the only carrier of the + # failure if this frame is lost to a disconnect. _fail_inflight_turn(session, error_value, error_surface=_error_surface) st.error_retained = True st.error_detail = _turn_failure_detail( @@ -655,9 +649,7 @@ def _complete_turn_payload(session: dict, st: _TurnRun, status_note: str | None, 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"), + "status": {"interrupted": "cancelled", "error": "failed"}.get(status, "settled"), "text": raw if isinstance(raw, str) else str(raw), **({"error": str(error_value or raw)} if status == "error" else {})}) st.receipt_committed = True @@ -677,8 +669,7 @@ def _recover_turn_exception(sid: str, session: dict, st: _TurnRun, e: BaseExcept f"{time.strftime('%Y-%m-%d %H:%M:%S')} · sid={sid} ===\n") f.write(traceback.format_exc()) print(f"[gateway-turn] {type(e).__name__}: {e}", file=sys.stderr, flush=True) - # A finalizer exception can leave in-memory history at the turn-start snapshot; keep - # the partial turn available to the next prompt. + # A finalizer exception can leave in-memory history at the turn-start snapshot. _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