jinghui/hermes-webui
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge master (exp-v0.52.361) into stage/7578-save-count

Browse Source 
# Conflicts:
#	CHANGELOG.md
This commit is contained in:
nesquena-hermes committed 2026-09-23 21:13:41 +00:00
2 parents 19693918fa b978bedfbc
commit ac0a6d08a7
5 files changed
+1769
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

api/models.py
+256
View File
@@ -2498,6 +2498,191 @@ def _latest_user_matches_pending_text(messages, pending_text):
return False
def _pending_active_turn_token(session):
"""Return the exact WebUI active-turn token for the pending turn.
The token is ``"<stream_id>:<pending_started_at>"`` — the same value the
eager user-message checkpoint and the agent-result merge stamp onto the
materialized user row (``stamp_message_source``). It is unforgeable
per-turn identity: two distinct streams that submit the same prompt
inside the same second produce different tokens.
"""
from api.process_event_utils import build_active_turn_token
return build_active_turn_token(
getattr(session, 'active_stream_id', None),
getattr(session, 'pending_started_at', None),
)
def _transcript_user_row_is_pending_turn(message, session, pending_token) -> bool:
"""Return True only when a transcript user row provably IS the pending turn.
Identity is bound to the active-turn token, never to integer-second
timestamp equality: two different streams that send the same prompt
within one second truncate to the same ``int(timestamp)``, so the
checkpoint matcher alone would accept the *other* stream's row as the
pending turn. A token mismatch therefore always loses, and a row that
carries no token at all cannot be proven to be this turn — which means
the caller must recover, not suppress.
"""
if not isinstance(message, dict) or message.get('role') != 'user':
return False
row_token = message.get('_active_turn_token')
if pending_token:
if not row_token:
# Ambiguous legacy identity: the pending turn has a resolvable
# token but this row predates token stamping. We cannot prove it
# is the same turn, so recovery must run.
return False
return row_token == pending_token
if row_token:
# The row belongs to a different, token-bearing turn. Never let a
# same-second timestamp collision override that mismatch.
return False
# Neither side carries a token (legacy/dead-stream pending state): fall
# back to the strict content + timestamp + source + attachments
# checkpoint, which is the only identity signal available.
return _message_matches_pending_checkpoint(
message,
getattr(session, 'pending_user_message', None),
getattr(session, 'pending_started_at', None),
getattr(session, 'pending_user_source', None),
getattr(session, 'pending_attachments', None),
)
def _pending_turn_has_final_assistant_answer(messages, start_idx: int) -> bool:
"""Return True when the matched user turn ends in a genuine final answer.
Reuses the established final-answer semantics
(``_assistant_message_has_final_visible_text``) instead of "an assistant
row exists". An assistant row only settles the turn when it carries real
visible answer text: empty rows, tool-call-only rows, interim
``_partial`` progress rows, rows recovered from the run journal and
compaction reference cards are all rejected, so a turn that was
interrupted mid-tool-execution is never mistaken for a completed one.
The scan stops at the next real user row — a final answer that belongs
to a *later* turn must not settle this pending turn.
"""
from api.streaming import (
_assistant_message_has_final_visible_text,
_is_synthetic_control_message,
)
for later in messages[start_idx + 1:]:
if not isinstance(later, dict):
continue
if _is_synthetic_control_message(later):
continue
role = later.get('role')
if role == 'user':
if is_context_compression_marker(later):
# Synthetic compaction cards are not a user-turn boundary.
continue
return False
if role != 'assistant':
# Tool rows never prove a final answer.
continue
if (
later.get('_partial')
or later.get('_error')
or later.get('_recovered')
or later.get('_recovered_from_run_journal')
):
continue
if is_context_compression_marker(later):
continue
if _assistant_message_has_final_visible_text(later):
return True
return False
def _transcript_already_advanced_past_pending(session) -> bool:
"""#6366: a stale pending user message whose own user row already appears
in the durable transcript, followed by a genuine final assistant answer
inside that same user-turn boundary, means the transcript has already
advanced past this pending turn. Returning True here lets the recovery
path skip the recovered user / ``_partial`` clone / journal replay /
generic no-response error that would otherwise append after the valid
final answer. The tail-only check at line ~3485 (``session.messages[-1]``)
misses this case: when the pending turn's user row is older than
the transcript tail, the tail is a *newer* assistant row that
can never match the pending **user** checkpoint, and the recovery
path falls through into the append branch.
Identity is bound to the pending stream's exact active-turn token
(``_transcript_user_row_is_pending_turn``), so a repeated prompt from a
different stream — even one submitted in the same integer second —
can never be matched against this pending turn and silently dropped.
Completion requires a genuine final visible assistant answer
(``_pending_turn_has_final_assistant_answer``): empty, tool-call-only,
interim ``_partial``, journal-recovered and compaction tails all keep
recovery active, because that is precisely the interrupted-turn shape
the recovery path exists for.
The transcript heuristic alone is not durable: an unflagged interim
prose row ("Let me check the logs first.") followed by tool rows that
were never followed by a final answer in the transcript passes the
predicate as a "completed" turn, but the stream may have died mid-
tool and the run journal was never marked done. 9/23 re-gate: also
require durable same-stream terminal evidence. The run journal
cannot supply that evidence on this path — the caller in
``_apply_core_sync_or_error_marker`` already returns early whenever
the run journal reports ``completed`` for the stream, so the branch
that reaches this predicate can never observe that state (which is
why the previous run-journal gate was unreachable dead code). The
evidence therefore has to come from the turn journal, which records
the exact-stream ``completed`` event for the pending turn
(``_turn_journal_records_completion``). Without it, fall through to
the normal journal-recovery path so the partial output is replayed
and the interruption is marked.
The suppression is therefore deliberately conservative: whenever the
turn's identity, its completion, or its durable terminal evidence
cannot be positively proven, this helper returns False and the
prompt is recovered instead of discarded. A false negative leaves a
visible cosmetic duplicate; a false positive silently destroys a
prompt or a response.
Idempotent: this is a pure read that does not mutate the session.
Running recovery twice therefore produces the same outcome.
"""
pending_text = getattr(session, 'pending_user_message', None)
if not pending_text:
return False
messages = getattr(session, 'messages', None)
if not isinstance(messages, list):
return False
pending_token = _pending_active_turn_token(session)
for idx, message in enumerate(messages):
if not _transcript_user_row_is_pending_turn(message, session, pending_token):
continue
# Found the pending turn's own user row at ``idx``. A genuine
# final answer inside that turn's boundary is necessary but not
# sufficient: durable same-stream terminal evidence must also
# exist, otherwise unflagged interim prose
# ("Let me check the logs first.") followed by tool rows can be
# mistaken for a finished turn.
if not _pending_turn_has_final_assistant_answer(messages, idx):
continue
stream_id = getattr(session, 'active_stream_id', None)
# The turn journal — not the run journal — carries the
# completion evidence that is still observable here: the caller
# (``_apply_core_sync_or_error_marker``) already consumed the
# run-journal ``completed`` state on its own early return, so a
# run-journal gate at this depth can never fire.
if not _turn_journal_records_completion(session, stream_id):
# No durable same-stream terminal evidence — the turn may
# have died mid-tool. Fall through to the journal-recovery
# path instead of suppressing it.
return False
return True
return False
def _partial_message_signature(message: dict) -> tuple:
"""Return a stable identity for partial assistant markers recovered on load."""
if not isinstance(message, dict):
@@ -2699,6 +2884,54 @@ def _run_journal_terminal_state(session, stream_id: str | None) -> str | None:
return str(terminal.get('terminal_state') or '') or None
def _turn_journal_records_completion(session, stream_id: str | None) -> bool:
"""Return True when the crash-safe turn journal proves the turn for
``stream_id`` reached its terminal ``completed`` state.
The turn journal is written per stream/turn id by the exact-stream
workers and by :mod:`api.routes` on submission, so a ``completed``
event for the *pending stream id* is completion evidence that
survives a run-journal write failure: the run journal only reaches
``done`` / ``stream_end`` on its own write path, and a process that
dies between the turn journal write and the run-journal terminal
append leaves the run journal without terminal state while the turn
journal still records that the turn actually finished.
Bound strictly to ``stream_id`` — a ``completed`` turn belonging to
another stream of the same session never settles this one — and to
the latest turn recorded for that stream, so a turn that was
cancelled or crashed (``interrupted``) keeps recovery active even
when an earlier turn of the same stream completed normally.
"""
if not stream_id:
return False
try:
from api.turn_journal import (
derive_turn_journal_states,
read_turn_journal,
)
journal = read_turn_journal(session.session_id)
states, _ = derive_turn_journal_states(journal.get('events') or [])
except Exception:
return False
latest: tuple[float, str, dict] | None = None
for turn_id, event in states.items():
if str(event.get('stream_id') or '') != str(stream_id):
continue
try:
created_at = float(event.get('created_at') or 0)
except (TypeError, ValueError):
created_at = 0.0
# ``>=`` on both elements keeps the winner deterministic when two
# turns of the same stream share a ``created_at``.
candidate = (created_at, turn_id, event)
if latest is None or candidate[:2] >= latest[:2]:
latest = candidate
if latest is None:
return False
return latest[2].get('event') == 'completed'
def _recoverable_unsaved_gateway_terminal_error(
session,
stream_id: str | None,
@@ -3575,6 +3808,29 @@ def _apply_core_sync_or_error_marker(
)
return True
if not _tail_user_already_checkpointed:
# #6366 re-gate: when the durable transcript has already
# advanced past this pending turn into a newer settled
# user/assistant boundary, the recovery path must NOT
# append a recovered user row + ``_partial`` clone +
# journal replay + generic no-response error after the
# valid final answer. The tail-only check above misses
# that case (the tail is a newer assistant row that can
# never match the pending user checkpoint). Clear only
# the stale pending fields and return; the transcript is
# already correct and durable.
if _transcript_already_advanced_past_pending(session):
session.active_stream_id = None
session.pending_user_message = None
session.pending_attachments = []
session.pending_started_at = None
session.pending_user_source = None
session.save(touch_updated_at=touch_updated_at)
logger.info(
"Session %s: cleared stale pending state for stream %s — transcript already advanced past this turn",
sid,
_stream_id,
)
return True
_append_recovered_pending_turn(session, timestamp=_recovered_ts)
else:
recovered = {