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

refactor(tui_gateway): predicate-collapse hosted proof, table-driven receipt status, compact attachment staging

Browse Source
This commit is contained in:
Teknium
2026-09-03 02:36:02 -07:00
parent 1fe670403e
commit 187e35950d
3 changed files with 48 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
tui_gateway/methods_prompt.py
View File

@@ -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))

22
tui_gateway/prompt_attachments.py
View File

@@ -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:

43
tui_gateway/prompt_turn.py
View File

@@ -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