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

refactor(tui_gateway): fold thin prompt-turn wrappers, unify survivor-field plumbing, shared side-agent admission

Browse Source
This commit is contained in:
Teknium
2026-09-03 02:06:29 -07:00
parent e1dc1ff22d
commit 70eac79526
2 changed files with 231 additions and 306 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tui_gateway/methods_prompt.py
+178 -213
View File
@@ -24,7 +24,7 @@ def _history_user_indices(history: list) -> list:
def _message_row_id(msg: dict):
"""Parse durable SQLite row id from a history entry, or None."""
"""Durable SQLite row id from a history entry (``_row_id`` else ``row_id``), or None."""
raw = msg.get("_row_id")
if raw is None:
raw = msg.get("row_id")
@@ -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 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.
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
@@ -64,11 +64,10 @@ def _mem_db_pair_agrees(mem, db_msg) -> bool:
def _find_user_turn_by_row_id(history: list, target_row_id: int):
"""Return ``(user_ordinal, history_index)`` for ``target_row_id``, or None."""
for u_ord, h_idx in enumerate(_history_user_indices(history)):
if _message_row_id(history[h_idx]) == target_row_id:
return u_ord, h_idx
return None
"""``(user_ordinal, history_index)`` for ``target_row_id``, or None."""
return next(
((u_ord, h_idx) for u_ord, h_idx in enumerate(_history_user_indices(history))
if _message_row_id(history[h_idx]) == target_row_id), None)
def _load_durable_truncation_history(
@@ -96,48 +95,42 @@ 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 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.
``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:
if (hit := _find_user_turn_by_row_id(history, target_row_id)) is not None:
return hit
db_history = _load_durable_truncation_history(session)
if db_history is None:
return None
# Heal missing stamps only when EVERY pair agrees (all-or-nothing): 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.
# 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:
if (db_rid := _message_row_id(db_msg)) is not None and _message_row_id(mem) is None:
mem["_row_id"] = db_rid
hit = _find_user_turn_by_row_id(history, target_row_id)
if hit is not None:
if (hit := _find_user_turn_by_row_id(history, target_row_id)) is not None:
return hit
db_hit = _find_user_turn_by_row_id(db_history, target_row_id)
if db_hit is None:
if (db_hit := _find_user_turn_by_row_id(db_history, target_row_id)) is None:
return None
db_ord, db_idx = db_hit
mem_user_indices = _history_user_indices(history)
if db_ord < 0 or db_ord >= len(mem_user_indices):
# Same-ordinal mapping across lists that can diverge (repair may have merged a
# user;user pair): trust it only when the mapped live turn shows the same content as
# the durable target — else refuse (caller fails closed, 4018).
if db_ord >= len(mem_user_indices) or not _mem_db_pair_agrees(
history[mem_user_indices[db_ord]], db_history[db_idx]):
return None
mem_idx = mem_user_indices[db_ord]
# Same-ordinal mapping across lists that can diverge (repair may have merged
# a user;user pair): trust it only when the mapped live turn shows the same
# content as the durable target — else refuse (caller fails closed, 4018).
if not _mem_db_pair_agrees(history[mem_idx], db_history[db_idx]):
return None
return db_ord, mem_idx
return db_ord, mem_user_indices[db_ord]
def _coerce_truncate_int(rid, value, param_name="truncate_before_user_ordinal"):
"""``(int_value, error_response)`` for a client integer param. bool is refused
like any non-integer: JSON ``true`` would int() to 1 and aim at the wrong turn."""
"""``(int_value, error_response)`` for a client integer param. bool is refused like
any non-integer: JSON ``true`` would int() to 1 and aim at the wrong turn."""
if not isinstance(value, bool):
with contextlib.suppress(TypeError, ValueError):
return int(value), None
@@ -148,12 +141,12 @@ def _reconcile_client_ordinal(
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. The cut is always aimed
by the durable target, so this can never re-aim a truncation.
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. 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
@@ -175,9 +168,9 @@ def _reconcile_client_ordinal(
def _pending_reaction_notes(session: dict) -> str:
"""Note block for reactions added since the last turn, or "". Applied to the
MODEL INPUT only, never the persisted prompt; each reaction is announced once
(rows are stamped ``seen`` on read). Feature-gated (display.message_reactions)."""
"""Note block for reactions added since the last turn, or "". Applied to the MODEL
INPUT only, never the persisted prompt; each reaction is announced once (rows are
stamped ``seen`` on read). Feature-gated (display.message_reactions)."""
session_key = str(session.get("session_key") or "")
if not session_key:
return ""
@@ -214,10 +207,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). 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."""
"""End the voice chat when a bare stop phrase is TYPED while backend voice mode 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 handles its own typed stop."""
if not (isinstance(text, str) and _voice_mode_enabled()):
return None
try:
@@ -253,8 +245,8 @@ def _hosted_submit_error(rid, session, hosted_task, hosted_terminal_callback):
def _legacy_group_fence_error(rid, session, params):
"""Older Desktop builds know the ``Group: <room-id>`` title but not the hosted
authority marker; once a gateway owns that room a direct prompt would start a
second renderer driver. Fence server-side instead of trusting the client."""
authority marker; once a gateway owns that room a direct prompt would start a second
renderer driver. Fence server-side instead of trusting the client."""
title = str(session.get("title") or "")
if not title.startswith("Group: "):
return None
@@ -279,8 +271,7 @@ def _legacy_group_fence_error(rid, session, params):
except RoomProbeUnavailableError:
return _err(rid, 5122, _GROUP_PROBE_FAILED_MSG)
except HostedRoomError:
# Legacy Desktop sessions used the display name after "Group: "; those
# names are not hosted room ids.
# Legacy Desktop sessions used the display name after "Group: " — not a room id.
return None
except Exception:
return _err(rid, 5122, _GROUP_PROBE_FAILED_MSG)
@@ -294,11 +285,10 @@ def _legacy_group_fence_error(rid, session, params):
def _parse_truncation_params(rid, sid, session, params, history):
"""Coerce + admit the truncation params; ``(target_row_id, client_ordinal, err)``.
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().
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")
@@ -330,9 +320,9 @@ def _parse_truncation_params(rid, sid, session, params, history):
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 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.
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)
@@ -351,18 +341,18 @@ def _ordinal_only_truncation_error(rid, sid, session, history, user_indices, cli
def _resolve_truncation_ordinal(rid, sid, session, params, history):
"""Resolve the truncation target to ``(ordinal, cut_index, err)``.
After ``_parse_truncation_params``: unresolvable target (4018, fail closed —
never degrade a missing row_id/message_id into an ordinal cut) -> ordinal
drift (4030) -> ordinal-only on a durable session (4004).
After ``_parse_truncation_params``: unresolvable target (4018, fail closed — never
degrade a missing row_id/message_id into an ordinal cut) -> ordinal drift (4030) ->
ordinal-only on a durable session (4004).
"""
target_row_id, client_ordinal, err = _parse_truncation_params(
rid, sid, session, params, history)
if err is not None:
return None, None, err
truncate_message_id = params.get("truncate_before_message_id")
# 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.
# 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)
@@ -384,11 +374,9 @@ def _resolve_truncation_ordinal(rid, sid, session, params, history):
param_name = "truncate_before_message_id"
target_repr = msg_id_str = str(truncate_message_id)
found_match = next(
(
(u_ord, h_idx) for u_ord, h_idx in enumerate(user_indices)
if history[h_idx].get("id") == msg_id_str
or history[h_idx].get("message_id") == msg_id_str),
None)
((u_ord, h_idx) for u_ord, h_idx in enumerate(user_indices)
if history[h_idx].get("id") == msg_id_str
or history[h_idx].get("message_id") == msg_id_str), None)
not_found = "target message_id %s not found in history for session %s"
if found_match is None:
logger.warning(
@@ -408,8 +396,8 @@ def _resolve_truncation_ordinal(rid, sid, session, params, history):
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
# negative indexing (user_indices[-1] -> the LAST user turn) and persist the loss.
# Reject out-of-range on BOTH ends: a negative ordinal would hit Python's negative
# indexing (user_indices[-1] -> the LAST user turn) and persist the loss.
if ordinal < 0 or ordinal >= len(user_indices):
return _stale(resolved_ordinal=ordinal)
return ordinal, user_indices[ordinal], None
@@ -422,14 +410,14 @@ def _row_ids_of(messages) -> set:
def _persist_truncation(rid, sid, session, history, truncated, ordinal, requested_rebind_ids):
"""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, 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)``.
If replace_messages failed after session["history"] was rewritten, the turn 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_fields)``; the fields are the client rowId-rebind payload.
"""
survivor_user_row_ids = survivor_row_id_map = None
fields = {}
with _session_db(session) as db:
if db is not None:
try:
@@ -438,21 +426,20 @@ def _persist_truncation(rid, sid, session, history, truncated, ordinal, requeste
truncation_key = session.get("session_key") or sid
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:
# read the un-repaired pre-write active-id set so a rewritten
# row is never mistaken for an untouched archived/ancestor row.
# Row-id fallback can resolve a target the live list is too 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)
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: 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.
# 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)
@@ -462,39 +449,39 @@ def _persist_truncation(rid, sid, session, history, truncated, ordinal, requeste
"(ordinal=%d); refusing turn so memory and DB stay "
"aligned: %s",
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 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)]
if requested_rebind_ids is not None:
survivor_row_id_map = {
return _err(rid, 5008, f"failed to persist history truncation: {exc}"), fields
# 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.
if requested_rebind_ids is None:
fields["survivor_user_row_ids"] = [
_message_row_id(truncated[i]) for i in _history_user_indices(truncated)]
else:
fields["survivor_row_id_map"] = 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))
if isinstance(old_row_id, int) and isinstance(new_row_id, int)
and old_row_id in requested_rebind_ids}
for dropped_row_id in requested_rebind_ids.intersection(old_active_row_ids):
survivor_row_id_map.setdefault(str(dropped_row_id), None)
return None, survivor_user_row_ids, survivor_row_id_map
row_id_map.setdefault(str(dropped_row_id), None)
return None, fields
def _truncate_history_for_submit(rid, sid, session, params, requested_rebind_ids):
"""Rewind/regenerate cut, under ``history_lock``. Returns
``(err, survivor_user_row_ids, survivor_row_id_map)``; on success
``session["history"]`` is replaced and ``history_version`` bumped."""
"""Rewind/regenerate cut, under ``history_lock``. Returns ``(err, survivor_fields)``;
on success ``session["history"]`` is replaced and ``history_version`` bumped."""
history = _history_without_ephemeral_scaffolding(session.get("history", []))
ordinal, cut_index, err = _resolve_truncation_ordinal(rid, sid, session, params, history)
if err is not None:
return err, None, None
return err, {}
from agent.context_compressor import history_before_user_originated_turn
truncated, _live_view = history_before_user_originated_turn(history, cut_index)
# Second gate on top of confirm_truncate: ordinal 0 -> history[:0] == [] and
# replace_messages() DELETEs every durable row. Wiping the whole transcript
# needs its own opt-in (legitimate restore/regenerate of the first turn).
# replace_messages() DELETEs every durable row. Wiping the whole transcript needs
# its own opt-in (legitimate restore/regenerate of the first turn).
if not truncated and history and not is_truthy_value(params.get("confirm_empty_truncate")):
logger.warning(
"prompt.submit: REFUSED empty truncation of session %s "
@@ -503,36 +490,25 @@ def _truncate_history_for_submit(rid, sid, session, params, requested_rebind_ids
return _err(
rid, 4028,
"truncation would erase the entire session transcript; "
"resubmit with confirm_empty_truncate=true if this is intended",
), None, None
"resubmit with confirm_empty_truncate=true if this is intended"), {}
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)
err, survivor_user_row_ids, survivor_row_id_map = _persist_truncation(
err, fields = _persist_truncation(
rid, sid, session, history, truncated, ordinal, requested_rebind_ids)
if err is not None:
return err, None, None
return err, {}
session["history"] = truncated
session["history_version"] = int(session.get("history_version", 0)) + 1
return None, survivor_user_row_ids, survivor_row_id_map
def _survivor_fields(survivor_user_row_ids, survivor_row_id_map, requested_rebind_ids) -> dict:
"""Client rowId-rebind payload for a submit that truncated a durable session."""
fields = {}
if survivor_user_row_ids is not None and requested_rebind_ids is None:
fields["survivor_user_row_ids"] = survivor_user_row_ids
if survivor_row_id_map is not None:
fields["survivor_row_id_map"] = survivor_row_id_map
return fields
return None, fields
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 it to a toast) or
None. On failure the in-flight turn is released."""
"""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 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(
@@ -558,15 +534,15 @@ def _persist_session_row_for_submit(rid, session):
def _run_after_agent_ready(rid, sid, session, text, display_kind, hosted_terminal_callback):
"""Turn thread body: patient wait for a deferred build (the message is already
the accepted in-flight turn, so a slow build must not eat it), then run."""
"""Turn thread body: patient wait for a deferred build (the message is already the
accepted in-flight turn, so a slow build must not eat it), then run."""
err = _wait_agent_for_prompt(session, rid, sid)
if err:
# Terminal frame + retained snapshot (not a bare "error" event): if the
# client is disconnected, the snapshot is the only way resume shows this.
# Terminal frame + retained snapshot (not a bare "error" event): if the client is
# disconnected, the snapshot is the only way resume shows this. Construction
# never reached the provider: local-runtime failure.
_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})
with session["history_lock"]:
session["running"] = False
@@ -594,14 +570,14 @@ _TRUNCATION_PARAMS = (
def _claim_submit_slot(rid, sid, session, text, params, transport, internal_hosted_submit):
"""Claim the turn against a possibly-running session; returns an early RPC
reply (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 (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.
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"]:
@@ -618,33 +594,32 @@ def _claim_submit_slot(rid, sid, session, text, params, transport, internal_host
def _lock_in_submit_turn(
rid, sid, session, text, params, has_truncation, requested_rebind_ids, hosted_task):
"""Under ``history_lock``: refuse watch-child races and malformed truncation,
apply the cut, then mark the turn running + in flight.
Returns ``(err, survivor_user_row_ids, survivor_row_id_map)``."""
survivor_user_row_ids = survivor_row_id_map = None
"""Under ``history_lock``: refuse watch-child races and malformed truncation, apply
the cut, then mark the turn running + in flight. Returns ``(err, survivor_fields)``."""
fields = {}
with session["history_lock"]:
# A watch session's run lives in the PARENT turn, so its own running flag
# is False; typing mid-run would build a second agent racing the child
# on the same stored session. After the run completes, submitting is fine.
# A watch session's run lives in the PARENT turn, so its own running flag is
# False; typing mid-run would build a second agent racing the child on the same
# stored session. After the run completes, submitting is fine.
if session.get("lazy") and _child_run_active(str(session.get("session_key") or "")):
return _err(rid, 4009, "subagent still running — wait for it to finish"), None, None
return _err(rid, 4009, "subagent still running — wait for it to finish"), fields
if is_truthy_value(params.get("confirm_truncate")) and not has_truncation:
return _err(
rid, 4004,
"confirm_truncate requires truncate_before_user_ordinal, truncate_before_message_id, or truncate_before_row_id",
), None, None
), fields
if has_truncation:
err, survivor_user_row_ids, survivor_row_id_map = _truncate_history_for_submit(
err, fields = _truncate_history_for_submit(
rid, sid, session, params, requested_rebind_ids)
if err is not None:
return err, None, None
return err, {}
session["running"] = True
session["_turn_cancel_requested"] = False
session["last_active"] = time.time()
if hosted_task is not None:
session["_hosted_room_task"] = dict(hosted_task)
_start_inflight_turn(session, text)
return None, survivor_user_row_ids, survivor_row_id_map
return None, fields
@method("prompt.submit")
@@ -706,12 +681,10 @@ def _(rid, params: dict) -> dict:
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)
err, survivor_user_row_ids, survivor_row_id_map = _lock_in_submit_turn(
err, survivor_fields = _lock_in_submit_turn(
rid, sid, session, text, params, has_truncation, requested_rebind_ids, hosted_task)
if err is not None:
return err
survivor_fields = _survivor_fields(
survivor_user_row_ids, survivor_row_id_map, requested_rebind_ids)
if turn_isolation:
isolated_response = _submit_prompt_to_compute_host(
rid, sid, session, text, display_kind=display_kind)
@@ -806,7 +779,7 @@ def _(rid, params: dict) -> dict:
@method("image.attach_bytes")
def _(rid, params: dict) -> dict:
"""Attach an image from base64 bytes (remote client: its file isn't on our disk).
Reply shape mirrors ``image.attach``. ``content_base64``/``data`` accept a
Reply shape mirrors ``image.attach``. ``content_base64``/``data`` accept a
``data:image/...;base64,`` prefix; ``filename``/``ext`` hint the extension, else
magic bytes decide (PNG/JPEG/GIF/WebP/BMP, fallback ``.png``)."""
session, err = _sess_building(params, rid)
@@ -887,8 +860,8 @@ def _pdf_page_range(rid, params):
@method("pdf.attach")
def _(rid, params: dict) -> dict:
"""Attach a PDF by rendering each page to PNG (``pdftoppm`` @150 DPI, poppler-utils;
5028 if missing) and queuing the pages as images. Accepts a host ``path`` or
base64 ``content_base64``. Caps: 50 MB / 25 pages per call."""
5028 if missing) and queuing the pages as images. Accepts a host ``path`` or base64
``content_base64``. Caps: 50 MB / 25 pages per call."""
import shutil
import subprocess
import tempfile
@@ -914,12 +887,11 @@ def _(rid, params: dict) -> dict:
str(pdf_path), str(td_path / "page")]
from hermes_cli._subprocess_compat import windows_hide_flags
try:
# UTF-8 + lossy decode: non-UTF-8 child output must not crash the gateway
# thread on locale-mismatched Windows.
res = subprocess.run(
argv, capture_output=True, text=True, timeout=120, stdin=subprocess.DEVNULL,
# 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())
encoding="utf-8", errors="replace", creationflags=windows_hide_flags())
except subprocess.TimeoutExpired:
return _err(rid, 5028, "pdftoppm timed out (>120s)")
if res.returncode != 0:
@@ -946,10 +918,10 @@ def _(rid, params: dict) -> dict:
@method("file.attach")
def _(rid, params: dict) -> dict:
"""Stage a non-image file into the session workspace and return a
workspace-relative ``@file:`` ref the agent's file tools can read. ``path`` is
the client/host path (naming + local resolution); ``data_url`` carries the bytes
when the path isn't visible to the gateway; ``name`` overrides the filename."""
"""Stage a non-image file into the session workspace and return a workspace-relative
``@file:`` ref the agent's file tools can read. ``path`` is the client/host path
(naming + local resolution); ``data_url`` carries the bytes when the path isn't
visible to the gateway; ``name`` overrides the filename."""
session, err = _sess_building(params, rid)
if err:
return err
@@ -1017,37 +989,30 @@ def _(rid, params: dict) -> dict:
# ── side agents (background / btw / preview.restart) ────────────────────────
@contextlib.contextmanager
def _session_profile_home_scope(session):
"""Bind the session's HERMES_HOME override for an ephemeral agent thread: the
ContextVar set on the session-create thread doesn't propagate, so a turn under
a non-default profile would otherwise run against the wrong home."""
profile_home = session.get("profile_home")
home_token = set_hermes_home_override(profile_home) if profile_home else None
try:
yield
finally:
if home_token is not None:
reset_hermes_home_override(home_token)
def _final_response_text(result) -> str:
return (result.get("final_response", str(result)) if isinstance(result, dict) else str(result))
def _spawn_side_agent(
rid, session, task_id, parent, event, body, *, cwd="", extra=None, cleanup=None):
"""Run ``body()`` (an ephemeral agent call) on a daemon thread under the
session's profile home and cwd; its text — or ``error: <exc>`` — lands on
``parent`` as ``event`` with ``task_id`` (+ ``extra``). ``cleanup`` runs in
the finally before the session context is cleared. Replies ``{task_id}``."""
"""Run ``body()`` (an ephemeral agent call) on a daemon thread under the session's
profile home and cwd; its text — or ``error: <exc>`` — lands on ``parent`` as
``event`` with ``task_id`` (+ ``extra``). ``cleanup`` runs in the finally before the
session context is cleared. Replies ``{task_id}``. The HERMES_HOME ContextVar set
on the session-create thread doesn't propagate, so the profile home is re-bound here
or a non-default profile would run against the wrong home."""
extra = extra or {}
def run():
session_tokens = _set_session_context(task_id, cwd=(cwd or _session_cwd(session)))
profile_home = session.get("profile_home")
home_token = set_hermes_home_override(profile_home) if profile_home else None
try:
with _session_profile_home_scope(session):
try:
text = body()
finally:
if home_token is not None:
reset_hermes_home_override(home_token)
_emit(event, parent, {"task_id": task_id, **extra, "text": text})
except Exception as e:
_emit(event, parent, {"task_id": task_id, **extra, "text": f"error: {e}"})
@@ -1060,15 +1025,22 @@ def _spawn_side_agent(
return _ok(rid, {"task_id": task_id})
@method("prompt.background")
def _(rid, params: dict) -> dict:
def _side_agent_args(rid, params, prefix):
"""Shared admission for the side-agent RPCs: ``(session, text, parent, task_id, err)``."""
session, err = _sess(params, rid)
if err:
return err
return None, None, None, None, err
text, parent = params.get("text", ""), params.get("session_id", "")
if not text:
return _err(rid, 4012, "text required")
task_id = f"bg_{uuid.uuid4().hex[:6]}"
return None, None, None, None, _err(rid, 4012, "text required")
return session, text, parent, f"{prefix}_{uuid.uuid4().hex[:6]}", None
@method("prompt.background")
def _(rid, params: dict) -> dict:
session, text, parent, task_id, err = _side_agent_args(rid, params, "bg")
if err:
return err
def body():
from run_agent import AIAgent
@@ -1082,16 +1054,12 @@ def _(rid, params: dict) -> dict:
@method("prompt.btw")
def _(rid, params: dict) -> dict:
"""Answer a side question without touching session history: snapshot the live
conversation (in-flight ``_session_messages`` else ``session["history"]``) and
run a one-shot auxiliary call (``agent/side_question.py``). History, role
alternation and prompt cache stay untouched; answer arrives as ``btw.complete``."""
session, err = _sess(params, rid)
conversation (in-flight ``_session_messages`` else ``session["history"]``) and run a
one-shot auxiliary call (``agent/side_question.py``). History, role alternation and
prompt cache stay untouched; answer arrives as ``btw.complete``."""
session, text, parent, task_id, err = _side_agent_args(rid, params, "btw")
if err:
return err
text, parent = params.get("text", ""), params.get("session_id", "")
if not text:
return _err(rid, 4012, "text required")
task_id = f"btw_{uuid.uuid4().hex[:6]}"
agent = session.get("agent")
snapshot = list(getattr(agent, "_session_messages", None) or session.get("history") or [])
main_runtime = {
@@ -1101,8 +1069,7 @@ def _(rid, params: dict) -> dict:
def body():
from agent.side_question import answer_side_question
return answer_side_question(
text, snapshot, parent_agent=agent, main_runtime=main_runtime,
) or ""
text, snapshot, parent_agent=agent, main_runtime=main_runtime) or ""
return _spawn_side_agent(
rid, session, task_id, parent, "btw.complete", body, extra={"question": text})
@@ -1173,9 +1140,9 @@ def _(rid, params: dict) -> dict:
_emit(
"preview.restart.progress", parent,
{"task_id": task_id, "text": f"Starting hidden restart agent{history_note}"})
# Deliberately NOT closed through task-wide process cleanup: the whole
# point is to leave a background server running under this task_id,
# and AIAgent.close() would kill every process for it.
# Deliberately NOT closed through task-wide process cleanup: the whole point is
# to leave a background server running under this task_id, and AIAgent.close()
# would kill every process for it.
result = AIAgent(
**_ephemeral_preview_agent_kwargs(session["agent"], task_id),
**_preview_restart_callbacks(parent, task_id),
@@ -1188,20 +1155,18 @@ def _(rid, params: dict) -> dict:
from tools.terminal_tool import clear_task_env_overrides
clear_task_env_overrides(task_id)
# Pin the validated preview cwd, else the parent workspace — never an
# invalid client path (which would silently fall back to the launch dir).
# Pin the validated preview cwd, else the parent workspace — never an invalid
# client path (which would silently fall back to the launch dir).
return _spawn_side_agent(
rid, session, task_id, parent, "preview.restart.complete", body,
cwd=preview_cwd, cleanup=cleanup)
# ── late-answer RPCs for tool-driven UI cards ───────────────────────────────
# All use allow_expired=True: each tool's bounded wait (read_terminal 30s,
# setup_mcp 10min, clarify ...) can expire — its _pending entry popped — while the
# card is still visible (e.g. a WS reconnect dropped tool.complete). A late answer
# must resolve gracefully instead of the raw 4009 "no pending answer request".
# All use allow_expired=True: each tool's bounded wait (read_terminal 30s, setup_mcp
# 10min, clarify ...) can expire — its _pending entry popped — while the card is still
# visible (e.g. a WS reconnect dropped tool.complete). A late answer must resolve
# gracefully instead of the raw 4009 "no pending answer request".
@method("clarify.respond")
@@ -1262,11 +1227,11 @@ def _(rid, params: dict) -> dict:
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
STORED id mapped to its live record. Returns the live session or None."""
"""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 STORED id mapped to
its live record. Returns the live session or None."""
request_id = str(params.get("request_id") or "")
if request_id:
try:
tui_gateway/prompt_turn.py
+53 -93
View File
@@ -21,7 +21,7 @@ def _hook_failure(what: str, exc: BaseException) -> None:
def _is_successful_goal_turn(result: Any, status: str, raw: Any) -> bool:
"""Return whether a turn produced a real response the goal judge can use."""
"""Whether a turn produced a real response the goal judge can use."""
return bool(
status == "complete" and isinstance(raw, str) and raw.strip()
and not (isinstance(result, dict) and result.get("failed"))
@@ -45,8 +45,7 @@ def _plan_goal_compression_recovery(
spinning until a random user message wakes it. Returns
``(continuation_prompt, status_notice)``; no active goal -> ``(None, None)``.
"""
compression_exhausted = bool(isinstance(result, dict) and result.get("compression_exhausted"))
if not compression_exhausted:
if not (isinstance(result, dict) and result.get("compression_exhausted")):
if _is_successful_goal_turn(result, status, raw):
session.pop(_GOAL_COMPRESSION_RECOVERY_ATTEMPTS, None)
return None, None
@@ -66,17 +65,15 @@ def _plan_goal_compression_recovery(
isinstance(recovery_state, dict)
and recovery_state.get("goal_created_at") == goal_created_at
and recovery_state.get("goal") == goal_text):
try:
with contextlib.suppress(TypeError, ValueError):
attempts = int(recovery_state.get("attempts", 0) or 0)
except (TypeError, ValueError):
attempts = 0
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}
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.
session.pop(_GOAL_COMPRESSION_RECOVERY_ATTEMPTS, None)
@@ -86,9 +83,6 @@ def _plan_goal_compression_recovery(
"Run /compress, then /goal resume to continue.")
# ── turn admission ───────────────────────────────────────────────────
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:
@@ -156,9 +150,6 @@ def _record_turn_marker(session: dict, text: Any) -> str:
return marker_key
# ── per-turn scopes ──────────────────────────────────────────────────
@dataclasses.dataclass(slots=True)
class _TurnScopes:
"""Reset tokens for the thread/context scopes a turn binds (filled incrementally)."""
@@ -187,10 +178,9 @@ 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, 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).
# The sudo password callback is thread-local: the build thread's wiring doesn't
# reach this turn thread and sudo prompts would fall through to /dev/tty and
# hang the headless gateway (secret capture is a module global; re-run is a no-op).
_wire_callbacks(sid)
@@ -209,9 +199,6 @@ def _release_turn_scopes(scopes: _TurnScopes) -> None:
_clear_session_context(scopes.session_tokens)
# ── message resolution ───────────────────────────────────────────────
def _expand_context_references(agent, prompt: str, cwd: str):
"""Expand ``@file`` references; returns the preprocess result (``.blocked``/``.message``)."""
from agent.context_references import preprocess_context_references
@@ -241,9 +228,8 @@ def _route_turn_images(agent, prompt: Any, images: list[str]) -> Any:
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)
print(f"[tui_gateway] image_routing decision failed, defaulting to text: {_img_exc}",
file=sys.stderr)
mode = "text"
if mode != "native":
return _build_image_ref_message(prompt, images)
@@ -256,10 +242,8 @@ def _route_turn_images(agent, prompt: Any, images: list[str]) -> Any:
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,
)
print(f"[tui_gateway] native attach failed, falling back to text: {_img_exc}",
file=sys.stderr)
return _build_image_ref_message(prompt, images)
@@ -293,17 +277,10 @@ def _start_turn_voice() -> tuple[Any, bool]:
return tts_queue, False
def _stop_thinking_sound() -> None:
with contextlib.suppress(Exception):
from tools.voice_mode import stop_thinking_sound
stop_thinking_sound()
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, 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)."""
was typed into (HUD mode is per-turn state; not for 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)
@@ -337,9 +314,6 @@ def _build_run_kwargs(
return run_kwargs
# ── post-run bookkeeping ─────────────────────────────────────────────
def _stamp_synthetic_display_kind(
agent, session: dict, result: Any, text: str, display_kind: str, display_metadata: dict | None
) -> None:
@@ -381,8 +355,7 @@ def _restore_moa_one_shot(sid: str, session: dict) -> None:
_apply_model_switch(
sid, session, _raw, confirm_expensive_model=False,
pin_session_override=bool(_prev_override),
persist_override=False, # session-internal restore, never config.yaml
)
persist_override=False) # session-internal restore, never config.yaml
except Exception as _moa_restore_exc:
logger.warning("MoA one-shot model restore failed: %s", _moa_restore_exc)
elif _restore is None:
@@ -411,9 +384,8 @@ def _commit_turn_history(
current_history = list(session["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)
if pivot_only:
if current_no_markers == history_no_markers and any(
_is_pivot_marker(e) for e in current_history):
# Auto-compression can make result["messages"] shorter than the
# turn-start history; then the full result is the base.
if len(result["messages"]) > len(history):
@@ -433,14 +405,18 @@ def _commit_turn_history(
"but was not saved to session history.")
def _result_status(result: dict) -> str:
return (
"interrupted" if result.get("interrupted")
else "error" if result.get("error") else "complete")
def _turn_outcome(result: Any) -> tuple[Any, str, str | None]:
"""Reduce a run_conversation result to ``(raw_text, status, last_reasoning)``."""
if not isinstance(result, dict):
return str(result), "complete", None
raw = result.get("final_response", "")
status = (
"interrupted" if result.get("interrupted")
else "error" if result.get("error") else "complete")
status = _result_status(result)
# 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.
@@ -449,7 +425,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
@@ -467,9 +443,6 @@ def _turn_error_surface(agent, result: Any) -> Any:
return None
# ── post-turn hooks ──────────────────────────────────────────────────
def _goal_followup_after_turn(
sid: str, session: dict, result: Any, status: str, raw: Any) -> str | None:
"""/goal continuation (mirrors gateway/run._post_turn_goal_continuation).
@@ -549,27 +522,6 @@ def _apply_pending_title(sid: str, session: dict) -> None:
pass # transient DB failure — keep pending_title for retry
def _speak_turn_fallback(raw: str) -> None:
"""Voice TTS fallback when the streaming pipeline couldn't start: speak the final text whole."""
try:
# Barge-aware: spoken interruptions must cut this playback too.
threading.Thread(target=_speak_text_with_barge, args=(raw,), daemon=True).start()
except ImportError:
logger.warning("voice TTS skipped: hermes_cli.voice unavailable")
except Exception as e:
logger.warning("voice TTS dispatch failed: %s", e)
def _append_turn_crash_log(sid: str, trace: str) -> None:
with contextlib.suppress(Exception):
os.makedirs(os.path.dirname(_CRASH_LOG), exist_ok=True)
with open(_CRASH_LOG, "a", encoding="utf-8") as f:
f.write(
f"\n=== turn-dispatcher exception · "
f"{time.strftime('%Y-%m-%d %H:%M:%S')} · sid={sid} ===\n")
f.write(trace)
def _dispatch_followup_turn(rid, sid: str, session: dict, prompt: Any, what: str, *,
on_done=None, on_error=None) -> None:
"""Chain one follow-up turn (caller already set ``running``); a dispatch failure
@@ -610,7 +562,6 @@ def _run_post_turn_followups(
return # user already sent something — their turn wins
session["running"] = True
_dispatch_followup_turn(rid, sid, session, goal_followup, "goal continuation dispatch")
# 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
@@ -642,20 +593,17 @@ def _run_post_turn_followups(
_hook_failure("completion queue drain", _drain_exc)
# ── the turn ─────────────────────────────────────────────────────────
@dataclasses.dataclass(slots=True)
class _TurnRun:
"""Mutable state the phase helpers of one turn thread share.
``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.
throws (re-read after ``_sync_bot_capabilities`` may swap in a rebuilt agent).
``error_retained``: the finally skips the inflight clear (failed snapshot stays
for resume replay). ``error_detail``: cause for the "tui turn finished" bookend,
stashed by both failure paths (the finally sees neither ``result`` nor the
exception reliably); ``prompt_text`` is the post-@-expansion prompt the cause
is checked against for quoting it back.
"""
agent: Any
@@ -849,7 +797,13 @@ def _complete_turn_payload(session: dict, st: _TurnRun, status_note: str | None,
def _recover_turn_exception(sid: str, session: dict, st: _TurnRun, e: BaseException) -> None:
"""Except-path of the turn: crash log, history restore, terminal error frame."""
import traceback
_append_turn_crash_log(sid, traceback.format_exc())
with contextlib.suppress(Exception):
os.makedirs(os.path.dirname(_CRASH_LOG), exist_ok=True)
with open(_CRASH_LOG, "a", encoding="utf-8") as f:
f.write(
f"\n=== turn-dispatcher exception · "
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)
# An exception in the agent's finalizer can leave the gateway's in-memory
# history at the turn-start snapshot; keep the partial turn available to
@@ -892,7 +846,9 @@ def _finish_turn(sid: str, session: dict, st: _TurnRun) -> None:
except Exception:
logger.debug("post-turn memory trim failed", exc_info=True)
if st.thinking_started:
_stop_thinking_sound()
with contextlib.suppress(Exception):
from tools.voice_mode import stop_thinking_sound
stop_thinking_sound()
if st.tts_queue is not None:
st.tts_queue.put(None) # end-of-text sentinel — flush + finish speaking
if st.one_turn_restore:
@@ -914,10 +870,7 @@ def _log_turn_finished(sid: str, session: dict, st: _TurnRun, started_monotonic:
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"
)
status = _result_status(result)
else:
status = "error" if st.error_retained else "complete"
logger.info(
@@ -966,8 +919,7 @@ def _run_prompt_submit(
prompt, run_message, cols, streamer = prepared
_invoke_agent(
sid, session, st, prompt, run_message, streamer, images, display_kind,
display_metadata,
)
display_metadata)
status_note = _absorb_turn_result(
sid, session, st, text, display_kind, display_metadata)
payload, raw, status = _complete_turn_payload(session, st, status_note, cols)
@@ -976,11 +928,19 @@ def _run_prompt_submit(
if status == "complete":
_complete_loop_tick(sid, session, raw)
_apply_pending_title(sid, session)
# The streaming path already spoke everything via tts_queue.
# Voice fallback when the streaming pipeline couldn't start (the
# streaming path already spoke everything via tts_queue); barge-aware
# so spoken interruptions cut this playback too.
if (
st.tts_queue is None and isinstance(raw, str) and raw.strip()
and _voice_tts_enabled()):
_speak_turn_fallback(raw)
try:
threading.Thread(
target=_speak_text_with_barge, args=(raw,), daemon=True).start()
except ImportError:
logger.warning("voice TTS skipped: hermes_cli.voice unavailable")
except Exception as e:
logger.warning("voice TTS dispatch failed: %s", e)
except Exception as e:
_recover_turn_exception(sid, session, st, e)
finally:
@@ -1008,7 +968,7 @@ def _run_prompt_submit(
run_thread = threading.Thread(target=run, daemon=True)
with _sessions_lock:
registered = _sessions.get(sid)
can_start = (not session.get("_closing") and (registered is None or registered is session))
can_start = not session.get("_closing") and (registered is None or registered is session)
if can_start:
session["_run_thread"] = run_thread
run_thread.start()