Skip to content

perf(webui): skip replay repair for proven-clean sidecars (3.8x warm concurrent loads) #7282

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ruizanthony wants to merge 5 commits into
base: master
Choose a base branch
from
Open

perf(webui): skip replay repair for proven-clean sidecars (3.8x warm concurrent loads) #7282

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
a8d4cc9
fix(webui): make replay reconciliation payload-strict
ruizanthony Aug 26, 2026
da7fc6e
test(webui): cover sanitized boundary replay
ruizanthony Sep 27, 2026
fcbe0c2
fix(webui): reconcile sanitized replay boundaries
ruizanthony Sep 27, 2026
36dbd0b
perf(webui): skip replay repair for sidecars already proven clean
ruizanthony Aug 24, 2026
8e68051
test(webui): characterize overlapping cold replay-repair loads
ruizanthony Sep 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions ARCHITECTURE.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -260,6 +260,22 @@ usable timestamp and rows at or after the sidecar tail also append normally.
The fallback therefore preserves an accepted state-only row when exact ordering
is ambiguous, while safely placeable recovery rows remain chronological.

#### Replay reconciliation authority

The visible `messages` projection, provider-facing `context_messages`, and
persisted session repair all use the same assistant replay pipeline. Adjacent
non-empty assistants collapse only when their complete strict-JSON payload
digests match; ids, timestamps, reasoning, annotations, attachments, and other
provider metadata therefore remain authoritative. Empty, partial, and
incomplete assistants use the narrower typed replay identities implemented by
that pipeline. Incomparable payloads fail closed and remain in order.

The active-turn boundary is an atomic `(current_turn_user_idx, turn_id)` pair
owned by one completed agent attempt. A credential retry clears any pair from
the failed attempt, then accepts either a complete pair from the new result or
a complete pair from the new agent. Fields from separate attempts or sources
must never be combined into deletion authority.

#### Imported `state.db` sidebar projection

`api.models.get_cli_sessions()` projects conversations from the active Hermes
Expand Down
809 changes: 707 additions & 102 deletions api/models.py
View file
Open in desktop

Large diffs are not rendered by default.

131 changes: 50 additions & 81 deletions api/routes.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25418,18 +25418,12 @@ def _handle_chat_sync(handler, body):
)
from api.streaming import (
_WEBUI_PROGRESS_PROMPT,
_active_turn_boundary,
_assign_stable_message_ids,
_dedupe_replayed_context_messages,
_find_active_turn_checkpoint_index,
_merge_display_messages_after_agent_result,
_record_agent_history_replay_authority,
_resolve_active_turn_authority,
_restore_display_reasoning_metadata,
_restore_reasoning_metadata_before_boundary,
_settle_current_turn_boundary,
_sanitize_messages_for_agent,
_compact_session_image_parts_for_persistence,
_context_messages_for_new_turn,
_settle_result_messages,
_workspace_context_prefix,
)
workspace_ctx = _workspace_context_prefix(str(s.workspace))
Expand All @@ -25454,17 +25448,38 @@ def _handle_chat_sync(handler, body):

_previous_messages = list(s.messages or [])
_previous_context_messages = list(_context_messages_for_new_turn(s, msg))
_sync_turn_source = getattr(s, "pending_user_source", None) or "webui"
# Synchronous requests have no SSE stream token, but they still
# need an explicit request-local turn identity. The Agent's
# persisted user index + turn id complete this provenance after
# run_conversation returns; strict settlement then uses the same
# authority as the asynchronous path instead of visible-text
# prefix inference.
_sync_active_turn_identity = {
"token": f"sync:{uuid.uuid4().hex}",
"text": msg,
"timestamp": time.time(),
"source": _sync_turn_source,
"attachments": [],
"current_turn_user_idx": None,
"turn_id": "",
}
_sync_agent_bound_history = _sanitize_messages_for_agent(
_previous_context_messages,
cfg=get_config(),
effective_model=_model,
effective_provider=_provider,
effective_base_url=_base_url,
)
_record_agent_history_replay_authority(
_sync_active_turn_identity,
_sync_agent_bound_history,
)

result = agent.run_conversation(
user_message=workspace_ctx + msg,
system_message=workspace_system_msg,
conversation_history=_sanitize_messages_for_agent(
_previous_context_messages,
cfg=get_config(),
effective_model=_model,
effective_provider=_provider,
effective_base_url=_base_url,
),
conversation_history=_sync_agent_bound_history,
task_id=s.session_id,
persist_user_message=msg,
)
Expand All @@ -25484,79 +25499,33 @@ def _handle_chat_sync(handler, body):
os.environ["HERMES_SESSION_KEY"] = old_session_key
with _get_session_agent_lock(s.session_id):
_result_messages = result.get("messages") or _previous_context_messages
# Active-turn boundary is fixed BEFORE any restoration (same as streaming),
# using whatever exact turn authority the result/Agent pair exported.
_active_turn_identity = _resolve_active_turn_authority(
{"token": None, "text": msg, "current_turn_user_idx": None, "turn_id": ""},
_sync_active_turn_identity = _resolve_active_turn_authority(
_sync_active_turn_identity,
result=result,
agent=agent,
)
if (
isinstance(_active_turn_identity, dict)
and _active_turn_identity.get("agent_turn_boundary_resolved") is True
and not _active_turn_identity.get("token")
):
_active_image_index = _find_active_turn_checkpoint_index(
_result_messages,
_previous_context_messages,
_active_turn_identity,
msg,
)
_active_image_content = (
_result_messages[_active_image_index].get("content")
if _active_image_index is not None
else None
)
if isinstance(_active_image_content, list) and any(
isinstance(part, dict)
and part.get("type") in {"image", "image_url", "input_image"}
for part in _active_image_content
):
from api.process_event_utils import build_active_turn_token

_active_turn_identity["token"] = build_active_turn_token(
f"sync:{s.session_id}:{_active_turn_identity['turn_id']}",
time.time(),
)
_turn_boundary = _active_turn_boundary(
_result_messages, _previous_context_messages, _active_turn_identity, msg,
)
_next_context_messages = _restore_reasoning_metadata_before_boundary(
_previous_context_messages,
_result_messages,
_turn_boundary,
)
# Mint ids on the shared result rows BEFORE dedupe deep-copies any
# stale-user boundary row, so both arrays share the id (#5564).
_assign_stable_message_ids(
_result_messages, _previous_messages, _previous_context_messages
)
_next_context_messages = _dedupe_replayed_context_messages(
_previous_context_messages,
_next_context_messages,
msg,
)
if _active_turn_identity.get("token"):
_next_context_messages = _settle_current_turn_boundary(
_previous_context_messages,
_next_context_messages,
_active_turn_identity,
msg,
getattr(s, "pending_user_source", None) or "webui",
)
s.context_messages = _next_context_messages
s.messages = _merge_display_messages_after_agent_result(
_settle_result_messages(
s,
_previous_messages,
_previous_context_messages,
_restore_display_reasoning_metadata(
_previous_messages, _result_messages, current_turn_boundary=_turn_boundary,
),
_result_messages,
msg,
source=getattr(s, "pending_user_source", None) or "webui",
verification_nudge_provenance={
"active_turn_identity": _active_turn_identity,
},
_sync_turn_source,
_sync_active_turn_identity,
)
# The synchronous endpoint has no reconnectable stream. Its request-
# local token is useful only while the shared settlement pipeline aligns
# display/context ownership; do not persist it as durable transcript
# metadata after the request has reached a terminal result.
_sync_turn_token = _sync_active_turn_identity.get("token")
if _sync_turn_token:
for _projection in (s.messages, s.context_messages):
for _message in _projection or []:
if (
isinstance(_message, dict)
and _message.get("_active_turn_token") == _sync_turn_token
):
_message.pop("_active_turn_token", None)
_compact_session_image_parts_for_persistence(s)
# Only auto-generate title when still default; preserves user renames
if s.title == "Untitled":
Expand Down
55 changes: 47 additions & 8 deletions api/session_recovery.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,6 @@
import logging
import os
import re
import shutil
import sqlite3
import threading
from contextlib import closing
Expand Down Expand Up @@ -65,7 +64,7 @@ def _is_valid_intentional_shrink_generation(value) -> bool:


def _msg_count(p: Path) -> int:
"""Return the number of messages in a session JSON file, or -1 on read/parse error.
"""Return the effective message count, or -1 on read/parse error.

Returns -1 for any non-session-shape file:
- File can't be read (OSError)
Expand All @@ -82,7 +81,21 @@ def _msg_count(p: Path) -> int:
if not isinstance(data, dict):
return -1
msgs = data.get('messages')
return len(msgs) if isinstance(msgs, list) else -1
if not isinstance(msgs, list):
return -1
# A shrink caused only by collapsing replayed empty ``incomplete`` rows is
# an intentional repair, not data loss. Compare live and backup using the
# same narrow identity rule as Session.save() so startup recovery does not
# resurrect the amplification. Unique backup messages still increase the
# effective count and remain recoverable.
try:
from api.models import _collapse_replayed_assistant_rows

msgs, _ = _collapse_replayed_assistant_rows(msgs)
except Exception:
logger.debug("Failed to compute effective recovery message count for %s", p, exc_info=True)
return -1
return len(msgs)


def _rebuild_recovery_session_index(session_dir: Path) -> None:
Expand Down Expand Up @@ -408,13 +421,28 @@ def recover_session(session_path: Path) -> dict:
if status["recommend"] != "restore":
return {**status, "restored": False}
bak_path = session_path.with_suffix('.json.bak')
# Stage the recovery via a tmp copy + atomic replace so a crash mid-restore
# cannot leave a half-written session.json.
# Stage the recovery via a tmp write + atomic replace so a crash
# mid-restore cannot leave a half-written session.json.
tmp_path = session_path.with_suffix('.json.recover.tmp')
try:
shutil.copyfile(bak_path, tmp_path)
# #6600: restore the SAME effective payload that _msg_count()
# evaluated — collapse replayed empty ``incomplete`` rows and
# recompute message_count from the collapsed list — so recovery never
# resurrects the duplicate amplification it just decided to repair.
bak_data = json.loads(bak_path.read_text(encoding='utf-8'))
if not isinstance(bak_data, dict):
raise ValueError("backup payload is not a session object")
from api.models import _repair_session_message_projections

bak_data, _, _ = _repair_session_message_projections(bak_data)
bak_messages = bak_data.get('messages')
if isinstance(bak_messages, list):
bak_data['message_count'] = len(bak_messages)
tmp_path.write_text(
json.dumps(bak_data, ensure_ascii=False, indent=2), encoding='utf-8'
)
tmp_path.replace(session_path)
except OSError as exc:
except (OSError, json.JSONDecodeError, ValueError) as exc:
logger.warning("recover_session: copy failed for %s: %s", session_path, exc)
try:
tmp_path.unlink(missing_ok=True)
Expand Down Expand Up @@ -551,13 +579,24 @@ def _read_state_db_missing_sidecar_rows(
if {'session_id', 'role', 'content'}.issubset(message_cols):
order = "timestamp, id" if 'timestamp' in message_cols and 'id' in message_cols else "rowid"
ts_expr = 'timestamp' if 'timestamp' in message_cols else 'NULL AS timestamp'
# A recovered sidecar needs durable per-row provenance before
# the normal Session.load/save replay reducers run. Without
# it, two legitimate state.db rows with identical role,
# content, and timestamp collapse irreversibly on first load.
row_id_expr = (
'id AS _state_db_row_id'
if 'id' in message_cols
else 'rowid AS _state_db_row_id'
)
for msg in conn.execute(
f"SELECT role, content, {ts_expr} FROM messages WHERE session_id = ? ORDER BY {order}",
f"SELECT role, content, {ts_expr}, {row_id_expr} "
f"FROM messages WHERE session_id = ? ORDER BY {order}",
(sid,),
).fetchall():
message = {
'role': msg['role'],
'content': msg['content'] or '',
'_state_db_row_id': msg['_state_db_row_id'],
}
if msg['timestamp'] is not None:
message['timestamp'] = msg['timestamp']
Expand Down
Loading
Loading