From 014f16c359c61a05979a97553936111407318228 Mon Sep 17 00:00:00 2001 From: starship-s <45587122+starship-s@users.noreply.github.com> Date: Tue, 28 Apr 2026 04:09:26 -0600 Subject: [PATCH] fix: harden session sidecar repair --- api/models.py | 206 ++++++-- api/streaming.py | 69 +-- tests/test_session_sidecar_repair.py | 764 +++++++++++++++++++++++++++ tests/test_title_aux_routing.py | 8 +- 4 files changed, 940 insertions(+), 107 deletions(-) create mode 100644 tests/test_session_sidecar_repair.py diff --git a/api/models.py b/api/models.py index f2706c58..30bf68d0 100644 --- a/api/models.py +++ b/api/models.py @@ -456,18 +456,153 @@ class Session: ) if include_runtime else False, } +def _get_profile_home(profile) -> Path: + """Resolve the hermes agent home directory for the given profile. + + Prefers the profile-specific helper from api.profiles; falls back to the + HERMES_HOME environment variable or ~/.hermes, expanding ~ correctly. + """ + try: + from api.profiles import get_hermes_home_for_profile + return Path(get_hermes_home_for_profile(profile)) + except ImportError: + return Path(os.environ.get('HERMES_HOME') or '~/.hermes').expanduser() + + +def _apply_core_sync_or_error_marker( + session, + core_path, + stream_id_for_recheck=None, + *, + require_stream_dead=True, +) -> bool: + """Inner repair logic. Must be called with the per-session lock already held. + + Re-checks session state under the lock, then either syncs messages from the + core transcript (if present and non-empty) or restores the pending user + message as a recovered user turn and appends an error marker. + + stream_id_for_recheck: when provided, repair bails if session.active_stream_id + changed (e.g. context compression rotated it). The cache-miss repair path + also requires the stream to be absent from active streams; the streaming + thread's final fallback passes require_stream_dead=False because it runs + before its own stream is removed from STREAMS. + + Returns True if repair was applied, False if the re-check bailed out. + Must never raise — caller is responsible for exception handling. + """ + sid = session.session_id + # Bail if pending is unset — nothing to repair. + if not session.pending_user_message: + return False + if stream_id_for_recheck is not None: + # Bail if active_stream_id rotated between the pre-lock check and now. + # Cache-miss repair must also skip if the stream is alive again, but the + # streaming thread's final fallback runs before removing its own stream + # from STREAMS and must be allowed to repair that same active stream. + if session.active_stream_id != stream_id_for_recheck: + return False + if require_stream_dead and session.active_stream_id in _active_stream_ids(): + return False + + # When messages is already non-empty the core-sync overwrite and recovered + # user turn are skipped (we cannot clobber in-memory mutations), but the + # stuck pending fields MUST still be cleared and an error marker appended + # so the session isn't permanently left in stale-pending state. + if len(session.messages) != 0: + session.active_stream_id = None + session.pending_user_message = None + session.pending_attachments = [] + session.pending_started_at = None + session.messages.append({ + 'role': 'assistant', + 'content': '**Previous turn did not complete.**', + 'timestamp': int(time.time()), + '_error': True, + }) + session.save() + logger.info( + "Session %s: pending cleared (messages non-empty), added error marker", + sid, + ) + return True + + # ── messages *is* empty ─ full repair ───────────────────────────────── + + if core_path.exists(): + with open(core_path, encoding='utf-8') as f: + core = json.load(f) + core_messages = core.get('messages', []) + if core_messages: + session.messages = core_messages + session.tool_calls = core.get('tool_calls', []) + for field in ('input_tokens', 'output_tokens', 'estimated_cost'): + if core.get(field) is not None: + setattr(session, field, core[field]) + session.active_stream_id = None + session.pending_user_message = None + session.pending_attachments = [] + session.pending_started_at = None + session.save() + logger.info( + "Session %s: synced %d messages from core transcript", + sid, len(core_messages), + ) + return True + + # Core missing or empty — restore the pending user message as a recovered + # user turn (preserving the draft), then append an error marker. + if session.pending_user_message: + # Use the original send time if available so the recovered turn + # appears in the correct chronological position. + _recovered_ts = int(time.time()) + if isinstance(session.pending_started_at, (int, float)) and session.pending_started_at > 0: + _recovered_ts = int(session.pending_started_at) + recovered: dict = { + 'role': 'user', + 'content': session.pending_user_message, + 'timestamp': _recovered_ts, + '_recovered': True, + } + if session.pending_attachments: + recovered['attachments'] = list(session.pending_attachments) + session.messages.append(recovered) + session.active_stream_id = None + session.pending_user_message = None + session.pending_attachments = [] + session.pending_started_at = None + session.messages.append({ + 'role': 'assistant', + 'content': '**Previous turn did not complete.**', + 'timestamp': int(time.time()), + '_error': True, + }) + session.save() + logger.info("Session %s: no core transcript found, added error marker", sid) + return True + + def _repair_stale_pending(session) -> bool: """Recover a sidecar stuck with messages=[] and stale pending state. Fires only when messages is empty, pending_user_message is set, active_stream_id is set, and the stream is no longer alive. + + Uses a non-blocking lock acquire so a caller that already holds the + per-session lock (e.g. retry_last, undo_last, cancel_stream) cannot + deadlock when get_session() triggers this on a cache miss. + Returns True if repair was applied, False otherwise. Must never raise — all errors are caught and logged. """ + # Capture the stream id seen at pre-check time; the under-lock re-check in + # _apply_core_sync_or_error_marker uses this to detect a rotated active_stream_id + # (e.g. context compression) or a stream that came back alive. + _seen_stream_id = session.active_stream_id if (len(session.messages) != 0 or not session.pending_user_message - or not session.active_stream_id - or session.active_stream_id in _active_stream_ids()): + or not _seen_stream_id + or _seen_stream_id in _active_stream_ids()): return False sid = session.session_id @@ -475,59 +610,24 @@ def _repair_stale_pending(session) -> bool: return False try: - try: - from api.profiles import get_hermes_home_for_profile - profile_home = get_hermes_home_for_profile(session.profile) - except ImportError: - profile_home = os.environ.get('HERMES_HOME', '~/.hermes') + profile_home = _get_profile_home(session.profile) + core_path = profile_home / 'sessions' / f'session_{sid}.json' - core_dir = Path(profile_home) / 'sessions' - core_path = core_dir / f'session_{sid}.json' - - with _get_session_agent_lock(sid): - # Re-check under lock — stream may have completed since the pre-lock check - if (len(session.messages) != 0 - or not session.pending_user_message - or not session.active_stream_id): - return False - - if core_path.exists(): - with open(core_path, encoding='utf-8') as f: - core = json.load(f) - core_messages = core.get('messages', []) - if core_messages: - session.messages = core_messages - session.tool_calls = core.get('tool_calls', []) - for field in ('input_tokens', 'output_tokens', 'estimated_cost'): - if core.get(field) is not None: - setattr(session, field, core[field]) - session.active_stream_id = None - session.pending_user_message = None - session.pending_attachments = [] - session.pending_started_at = None - session.save() - logger.info( - "Repaired stale sidecar for session %s: synced %d messages from core", - sid, len(core_messages), - ) - return True - - # Core missing or has no messages — clear pending and add error marker - session.active_stream_id = None - session.pending_user_message = None - session.pending_attachments = [] - session.pending_started_at = None - session.messages.append({ - 'role': 'assistant', - 'content': '**Previous turn did not complete:** the original message has been preserved as a draft.', - 'timestamp': int(time.time()), - '_error': True, - }) - session.save() - logger.info( - "Repaired stale sidecar for session %s: no core transcript, added error marker", sid, + lock = _get_session_agent_lock(sid) + # Non-blocking acquire: bail immediately if the caller already holds this + # lock (e.g. retry_last, undo_last, cancel_stream). Blocking would deadlock + # because _get_session_agent_lock returns a non-reentrant threading.Lock. + if not lock.acquire(blocking=False): + logger.debug( + "_repair_stale_pending: lock contended, skipping repair for session %s", sid, ) - return True + return False + try: + return _apply_core_sync_or_error_marker( + session, core_path, stream_id_for_recheck=_seen_stream_id, + ) + finally: + lock.release() except Exception: logger.exception("_repair_stale_pending failed for session %s", sid) return False diff --git a/api/streaming.py b/api/streaming.py index 1b46c913..044a7548 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -1115,56 +1115,23 @@ def _last_resort_sync_from_core(session, stream_id, agent_lock): Called from the outer finally block of _run_agent_streaming. Must never raise. """ + from api.models import _get_profile_home, _apply_core_sync_or_error_marker try: - try: - from api.profiles import get_hermes_home_for_profile - profile_home = get_hermes_home_for_profile(session.profile) - except ImportError: - profile_home = os.environ.get('HERMES_HOME', '~/.hermes') + # Guard: if a cancel was already requested, bail out — cancel_stream() has + # already saved partial content and we must not double-append error markers. + if stream_id in CANCEL_FLAGS and CANCEL_FLAGS[stream_id].is_set(): + return - core_dir = Path(profile_home) / 'sessions' - core_path = core_dir / f'session_{session.session_id}.json' + profile_home = _get_profile_home(session.profile) + core_path = profile_home / 'sessions' / f'session_{session.session_id}.json' _lock_ctx = agent_lock if agent_lock is not None else contextlib.nullcontext() with _lock_ctx: - # Re-check under lock — success/except paths may have already cleared these - if (getattr(session, 'active_stream_id', None) != stream_id - or not getattr(session, 'pending_user_message', None)): - return - - if core_path.exists(): - with open(core_path, encoding='utf-8') as f: - core = json.load(f) - core_messages = core.get('messages', []) - if core_messages: - session.messages = core_messages - session.tool_calls = core.get('tool_calls', []) - session.active_stream_id = None - session.pending_user_message = None - session.pending_attachments = [] - session.pending_started_at = None - session.save() - logger.info( - "_last_resort_sync_from_core: synced %d messages for session %s", - len(core_messages), session.session_id, - ) - return - - # Core missing or empty — clear pending and add error marker - session.active_stream_id = None - session.pending_user_message = None - session.pending_attachments = [] - session.pending_started_at = None - session.messages.append({ - 'role': 'assistant', - 'content': '**Previous turn did not complete:** the original message has been preserved as a draft.', - 'timestamp': int(time.time()), - '_error': True, - }) - session.save() - logger.info( - "_last_resort_sync_from_core: no core transcript for session %s, added error marker", - session.session_id, + _apply_core_sync_or_error_marker( + session, + core_path, + stream_id_for_recheck=stream_id, + require_stream_dead=False, ) except Exception: logger.exception( @@ -2167,15 +2134,17 @@ def _run_agent_streaming(session_id, msg_text, model, workspace, stream_id, atta _apperror_payload['hint'] = _exc_hint put('apperror', _apperror_payload) finally: - if (s is not None - and getattr(s, 'active_stream_id', None) == stream_id - and getattr(s, 'pending_user_message', None)): - _last_resort_sync_from_core(s, stream_id, _agent_lock) - # Stop periodic checkpoint thread if it was started (Issue #765) + # Stop the periodic checkpoint thread before the final recovery path. + # The checkpoint thread also uses the per-session lock; joining it first + # avoids contending with checkpoint writes during stale-pending repair. if _checkpoint_stop is not None: _checkpoint_stop.set() if _ckpt_thread is not None: _ckpt_thread.join(timeout=15) + if (s is not None + and getattr(s, 'active_stream_id', None) == stream_id + and getattr(s, 'pending_user_message', None)): + _last_resort_sync_from_core(s, stream_id, _agent_lock) _clear_thread_env() # TD1: always clear thread-local context with STREAMS_LOCK: STREAMS.pop(stream_id, None) diff --git a/tests/test_session_sidecar_repair.py b/tests/test_session_sidecar_repair.py new file mode 100644 index 00000000..3b81efd4 --- /dev/null +++ b/tests/test_session_sidecar_repair.py @@ -0,0 +1,764 @@ +"""Regression tests for session sidecar repair logic.""" +import json +import queue +import os +import sys +import threading +import time +from pathlib import Path +from unittest.mock import patch + +import pytest + +import api.models as models +from api.models import ( + Session, + _get_profile_home, + _apply_core_sync_or_error_marker, + _repair_stale_pending, + _active_stream_ids, +) +import api.config as config +import api.streaming as streaming +import api.profiles as profiles + + +# ── Fixtures ──────────────────────────────────────────────────────────────── + +@pytest.fixture(autouse=True) +def _isolate_session_dir(tmp_path, monkeypatch): + """Redirect SESSION_DIR and SESSION_INDEX_FILE to a temp directory.""" + session_dir = tmp_path / "sessions" + session_dir.mkdir() + index_file = session_dir / "_index.json" + + monkeypatch.setattr(models, "SESSION_DIR", session_dir) + monkeypatch.setattr(models, "SESSION_INDEX_FILE", index_file) + + models.SESSIONS.clear() + yield session_dir, index_file + models.SESSIONS.clear() + + +@pytest.fixture(autouse=True) +def _isolate_stream_state(): + """Isolate shared stream state between tests.""" + config.STREAMS.clear() + config.CANCEL_FLAGS.clear() + config.AGENT_INSTANCES.clear() + config.STREAM_PARTIAL_TEXT.clear() + yield + config.STREAMS.clear() + config.CANCEL_FLAGS.clear() + config.AGENT_INSTANCES.clear() + config.STREAM_PARTIAL_TEXT.clear() + + +@pytest.fixture(autouse=True) +def _isolate_agent_locks(): + """Clear per-session agent locks between tests.""" + config.SESSION_AGENT_LOCKS.clear() + yield + config.SESSION_AGENT_LOCKS.clear() + + +@pytest.fixture() +def hermes_home(tmp_path, monkeypatch): + """Set up a HERMES_HOME directory with a sessions subdirectory.""" + home = tmp_path / "hermes_home" + home.mkdir() + sessions_dir = home / "sessions" + sessions_dir.mkdir() + monkeypatch.setenv("HERMES_HOME", str(home)) + monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", home) + return home + + +def _make_session(session_id="test_sid", messages=None, **kwargs): + """Helper to create a Session with sensible defaults for repair tests.""" + defaults = { + "session_id": session_id, + "title": "Test Session", + "messages": messages or [], + } + defaults.update(kwargs) + return Session(**defaults) + + +def _make_stale_session(session_id="stale_sid", pending_msg="Hello hermes", stream_id="stream_1"): + """Helper to create a session in stale-pending state (messages empty, pending set).""" + s = _make_session(session_id=session_id, messages=[]) + s.pending_user_message = pending_msg + s.active_stream_id = stream_id + s.pending_attachments = [] + s.pending_started_at = None + return s + + +def _write_core_transcript(hermes_home, session_id, messages, **extra): + """Write a core transcript JSON file for a session.""" + core_path = hermes_home / "sessions" / f"session_{session_id}.json" + data = {"messages": messages, **extra} + core_path.parent.mkdir(parents=True, exist_ok=True) + core_path.write_text(json.dumps(data), encoding="utf-8") + return core_path + + +def _register_active_stream(stream_id): + """Register stream_id as live in the same state _run_agent_streaming uses.""" + with config.STREAMS_LOCK: + config.STREAMS[stream_id] = queue.Queue() + + +class TestRepairStalePendingNoDeadlock: + """_repair_stale_pending uses non-blocking lock acquire so callers that + already hold the per-session lock (retry_last, undo_last, cancel_stream) + cannot deadlock when get_session() triggers repair on a cache miss.""" + + def test_returns_false_when_lock_already_held(self, hermes_home, monkeypatch): + """If the per-session lock is already held, _repair_stale_pending returns + False instead of blocking forever (deadlock prevention).""" + s = _make_stale_session() + s.save() + + lock = config._get_session_agent_lock(s.session_id) + # Acquire the lock ourselves — simulating retry_last/undo_last holding it + assert lock.acquire(blocking=False) + + try: + result = _repair_stale_pending(s) + assert result is False, "Should bail out when lock is contended" + finally: + lock.release() + + def test_no_deadlock_when_get_session_triggers_repair(self, hermes_home, monkeypatch): + """Simulate the real deadlock scenario: a caller holds the per-session + lock and then calls get_session(), which evicts the session from cache + and re-loads it, triggering _repair_stale_pending. + + Spawns a worker thread that acquires the per-session lock and then calls + get_session(). The test asserts the worker completes within 5 seconds + and raises no exception — this reproduces the exact production deadlock + the prior fix was for. + + When the lock is already held, _repair_stale_pending's non-blocking + acquire fails, so pending fields are deliberately NOT cleared — this + preserves safety over repair; the deadlock is avoided.""" + s = _make_stale_session() + s.save() + models.SESSIONS[s.session_id] = s + + sid = s.session_id + completed = threading.Event() + worker_exc = [] + + def _worker(): + lock = config._get_session_agent_lock(sid) + try: + with lock: + # Evict from cache so get_session re-loads from disk + models.SESSIONS.pop(sid, None) + # This would deadlock if _repair_stale_pending blocked on the + # per-session lock that the caller already holds. + result = models.get_session(sid) + assert result is not None, "get_session should return a session" + # When the lock is held, repair bails (non-blocking acquire + # fails) — pending fields are intentionally preserved rather + # than risking a deadlock. + assert result.pending_user_message is not None, ( + "Pending fields preserved when lock is held (deadlock prevention)" + ) + except Exception as exc: + worker_exc.append(exc) + finally: + completed.set() + + worker = threading.Thread(target=_worker, daemon=True) + worker.start() + + # Worker must finish within 5 seconds — if it doesn't, we deadlocked. + assert completed.wait(timeout=5), ( + "Worker thread did not complete within 5 seconds — likely deadlock " + "in get_session() repair path" + ) + worker.join(timeout=1) + + assert len(worker_exc) == 0, ( + f"Worker raised exception: {worker_exc[0] if worker_exc else 'none'}" + ) + + +class TestDraftRecovery: + """When no core transcript exists, the pending user message is restored as + a recovered user turn (_recovered=True) and the error marker says + 'Previous turn did not complete.' — NOT 'preserved as a draft'.""" + + def test_pending_message_recovered_as_user_turn(self, hermes_home, monkeypatch): + """When core transcript is missing, the pending_user_message is appended + as a user turn with _recovered=True, and its timestamp matches + pending_started_at when available.""" + _ts = time.time() - 60 # 60 seconds ago + s = _make_stale_session(pending_msg="My important question") + s.pending_started_at = _ts + lock = config._get_session_agent_lock(s.session_id) + + with lock: + core_path = hermes_home / "sessions" / f"session_{s.session_id}.json" + result = _apply_core_sync_or_error_marker(s, core_path, stream_id_for_recheck="stream_1") + + assert result is True + # Find the recovered user turn + user_msgs = [m for m in s.messages if m.get("role") == "user"] + assert len(user_msgs) == 1 + assert user_msgs[0]["content"] == "My important question" + assert user_msgs[0].get("_recovered") is True + assert user_msgs[0]["timestamp"] == int(_ts), ( + f"Recovered turn timestamp should match pending_started_at ({_ts}), " + f"got {user_msgs[0]['timestamp']}" + ) + + def test_error_marker_no_preserved_as_draft(self, hermes_home, monkeypatch): + """Error marker text must NOT say 'preserved as a draft'.""" + s = _make_stale_session() + lock = config._get_session_agent_lock(s.session_id) + + with lock: + core_path = hermes_home / "sessions" / f"session_{s.session_id}.json" + _apply_core_sync_or_error_marker(s, core_path, stream_id_for_recheck="stream_1") + + error_msgs = [m for m in s.messages if m.get("_error")] + assert len(error_msgs) == 1 + content = error_msgs[0]["content"] + assert "preserved as a draft" not in content, ( + f"Error marker should not say 'preserved as a draft', got: {content}" + ) + assert "Previous turn did not complete" in content + + def test_pending_attachments_recovered(self, hermes_home, monkeypatch): + """Attachments on the pending message are carried over to the recovered turn.""" + s = _make_stale_session() + s.pending_attachments = [{"type": "image", "name": "photo.png"}] + lock = config._get_session_agent_lock(s.session_id) + + with lock: + core_path = hermes_home / "sessions" / f"session_{s.session_id}.json" + _apply_core_sync_or_error_marker(s, core_path, stream_id_for_recheck="stream_1") + + user_msgs = [m for m in s.messages if m.get("role") == "user"] + assert len(user_msgs) == 1 + assert user_msgs[0].get("attachments") == [{"type": "image", "name": "photo.png"}] + + def test_pending_fields_cleared_after_recovery(self, hermes_home, monkeypatch): + """After recovery, all pending fields are cleared.""" + s = _make_stale_session() + s.pending_attachments = [{"type": "image", "name": "photo.png"}] + s.pending_started_at = time.time() + lock = config._get_session_agent_lock(s.session_id) + + with lock: + core_path = hermes_home / "sessions" / f"session_{s.session_id}.json" + _apply_core_sync_or_error_marker(s, core_path, stream_id_for_recheck="stream_1") + + assert s.pending_user_message is None + assert s.pending_attachments == [] + assert s.pending_started_at is None + assert s.active_stream_id is None + + +class TestStreamIdRecheck: + """Under-lock re-check in _apply_core_sync_or_error_marker bails out when + active_stream_id has rotated or the stream has come back alive.""" + + def test_bails_when_stream_id_rotated(self, hermes_home, monkeypatch): + """If active_stream_id changed between pre-lock and under-lock check, + repair bails out (prevents clobbering a new stream's state).""" + s = _make_stale_session(stream_id="stream_old") + lock = config._get_session_agent_lock(s.session_id) + + # Simulate the stream ID rotating (e.g. context compression) + s.active_stream_id = "stream_new" + + with lock: + core_path = hermes_home / "sessions" / f"session_{s.session_id}.json" + result = _apply_core_sync_or_error_marker( + s, core_path, stream_id_for_recheck="stream_old", + ) + + assert result is False, "Should bail when stream_id rotated" + + def test_bails_when_stream_came_alive(self, hermes_home, monkeypatch): + """If the stream is alive in STREAMS (cancel not yet processed), + repair bails out — the streaming thread is still managing the session.""" + s = _make_stale_session(stream_id="stream_alive") + lock = config._get_session_agent_lock(s.session_id) + + # Register the stream as alive + _register_active_stream("stream_alive") + + try: + with lock: + core_path = hermes_home / "sessions" / f"session_{s.session_id}.json" + result = _apply_core_sync_or_error_marker( + s, core_path, stream_id_for_recheck="stream_alive", + ) + + assert result is False, "Should bail when stream is still alive" + finally: + with config.STREAMS_LOCK: + config.STREAMS.pop("stream_alive", None) + + def test_proceeds_when_stream_is_dead(self, hermes_home, monkeypatch): + """When the stream is not alive (not in STREAMS), repair proceeds.""" + s = _make_stale_session(stream_id="stream_dead") + lock = config._get_session_agent_lock(s.session_id) + + # Stream is NOT in STREAMS — repair should proceed + with lock: + core_path = hermes_home / "sessions" / f"session_{s.session_id}.json" + result = _apply_core_sync_or_error_marker( + s, core_path, stream_id_for_recheck="stream_dead", + ) + + assert result is True + + +class TestGetProfileHome: + """_get_profile_home expands ~ correctly in the ImportError fallback path.""" + + def test_expands_tilde_when_profiles_unavailable(self, monkeypatch): + """When api.profiles import fails, fallback uses HERMES_HOME or ~/.hermes + with proper tilde expansion.""" + # Make api.profiles import fail + monkeypatch.setitem(sys.modules, "api.profiles", None) + + # Default fallback without HERMES_HOME env var + monkeypatch.delenv("HERMES_HOME", raising=False) + result = _get_profile_home(None) + assert "~" not in str(result), f"Path should have ~ expanded, got: {result}" + assert str(result) == str(Path.home() / ".hermes") + + def test_uses_hermes_home_env_var(self, monkeypatch): + """When HERMES_HOME is set, fallback uses it with expansion.""" + monkeypatch.setitem(sys.modules, "api.profiles", None) + monkeypatch.setenv("HERMES_HOME", "/custom/hermes") + result = _get_profile_home(None) + assert str(result) == "/custom/hermes" + + def test_expands_tilde_in_hermes_home(self, monkeypatch): + """If HERMES_HOME contains ~, it gets expanded.""" + monkeypatch.setitem(sys.modules, "api.profiles", None) + monkeypatch.setenv("HERMES_HOME", "~/my-hermes") + result = _get_profile_home(None) + assert "~" not in str(result) + assert str(result) == str(Path.home() / "my-hermes") + + +class TestCancelInProgressGuard: + """_last_resort_sync_from_core bails out when a cancel is in progress, + preventing duplicate markers (cancel_stream already saves partial + cancel marker).""" + + def test_bails_when_cancel_flag_set(self, hermes_home, monkeypatch): + """If CANCEL_FLAGS[stream_id].is_set(), _last_resort_sync_from_core + returns immediately without appending any messages.""" + s = _make_stale_session(stream_id="cancel_stream") + s.save() + + # Set up cancel flag + cancel_event = threading.Event() + cancel_event.set() + config.CANCEL_FLAGS["cancel_stream"] = cancel_event + + # Create an agent lock + agent_lock = config._get_session_agent_lock(s.session_id) + + # Record message count before + msg_count_before = len(s.messages) + + streaming._last_resort_sync_from_core(s, "cancel_stream", agent_lock) + + # Should NOT have appended any messages + assert len(s.messages) == msg_count_before, ( + "Should not append messages when cancel is in progress" + ) + # Pending fields should NOT have been cleared by _last_resort_sync_from_core + # (cancel_stream handles that separately) + assert s.pending_user_message is not None + + def test_proceeds_when_cancel_flag_not_set(self, hermes_home, monkeypatch): + """When cancel flag is not set, _last_resort_sync_from_core proceeds + with repair normally.""" + s = _make_stale_session(stream_id="normal_stream") + s.save() + + # Cancel flag exists but is NOT set + cancel_event = threading.Event() + config.CANCEL_FLAGS["normal_stream"] = cancel_event + + agent_lock = config._get_session_agent_lock(s.session_id) + _register_active_stream("normal_stream") + + streaming._last_resort_sync_from_core(s, "normal_stream", agent_lock) + + # Should have performed repair (appended messages) + assert len(s.messages) > 0, "Should have appended messages" + + def test_proceeds_when_cancel_flag_absent(self, hermes_home, monkeypatch): + """When no cancel flag exists for the stream, repair proceeds normally.""" + s = _make_stale_session(stream_id="no_flag_stream") + s.save() + + # No CANCEL_FLAGS entry at all + agent_lock = config._get_session_agent_lock(s.session_id) + _register_active_stream("no_flag_stream") + + streaming._last_resort_sync_from_core(s, "no_flag_stream", agent_lock) + + assert len(s.messages) > 0 + + +class TestEmptyMessagesGuard: + """_apply_core_sync_or_error_marker bails out when session.messages is + non-empty, preventing it from clobbering in-memory mutations made by the + streaming thread or cancel path.""" + + def test_pending_cleared_when_messages_nonempty_direct(self, hermes_home, monkeypatch): + """When _apply_core_sync_or_error_marker is called on a session with + non-empty messages and pending set, it clears the pending fields and + appends an error marker, returning True.""" + s = _make_session(messages=[{"role": "user", "content": "hello"}]) + s.pending_user_message = "Another question" + s.active_stream_id = "stream_1" + lock = config._get_session_agent_lock(s.session_id) + + with lock: + core_path = hermes_home / "sessions" / f"session_{s.session_id}.json" + result = _apply_core_sync_or_error_marker( + s, core_path, stream_id_for_recheck="stream_1", + ) + + assert result is True + # Original message should be untouched + assert len(s.messages) == 2 # original + error marker + assert s.messages[0]["content"] == "hello" + # Error marker appended + assert s.messages[1].get("_error") is True + # Pending fields cleared + assert s.pending_user_message is None + assert s.active_stream_id is None + + def test_bails_when_pending_user_message_none(self, hermes_home, monkeypatch): + """If pending_user_message is None, repair bails out.""" + s = _make_session(messages=[]) + s.pending_user_message = None + s.active_stream_id = "stream_1" + lock = config._get_session_agent_lock(s.session_id) + + with lock: + core_path = hermes_home / "sessions" / f"session_{s.session_id}.json" + result = _apply_core_sync_or_error_marker( + s, core_path, stream_id_for_recheck="stream_1", + ) + + assert result is False + + def test_proceeds_when_messages_empty(self, hermes_home, monkeypatch): + """When messages is empty and pending_user_message is set, repair proceeds.""" + s = _make_stale_session() + lock = config._get_session_agent_lock(s.session_id) + + with lock: + core_path = hermes_home / "sessions" / f"session_{s.session_id}.json" + result = _apply_core_sync_or_error_marker( + s, core_path, stream_id_for_recheck="stream_1", + ) + + assert result is True + + +class TestNonEmptyMessagesPendingCleared: + """When messages is non-empty and pending is stuck, _last_resort_sync_from_core + clears the pending fields and appends exactly one error marker without + clobbering existing messages or syncing from core.""" + + def test_pending_cleared_when_messages_nonempty(self, hermes_home, monkeypatch): + """_last_resort_sync_from_core on a session with both messages and + pending_user_message clears pending fields and appends exactly one + error marker.""" + s = _make_session(messages=[{"role": "user", "content": "existing turn"}]) + s.pending_user_message = "Stuck draft" + s.pending_attachments = [{"type": "image", "name": "screenshot.png"}] + s.pending_started_at = time.time() - 120 + s.active_stream_id = "stale_stream" + s.save() + + # Write a core transcript — must NOT be synced because messages is non-empty + core_messages = [ + {"role": "user", "content": "Core user msg"}, + {"role": "assistant", "content": "Core assistant msg"}, + ] + _write_core_transcript(hermes_home, s.session_id, core_messages) + + agent_lock = config._get_session_agent_lock(s.session_id) + _register_active_stream("stale_stream") + + streaming._last_resort_sync_from_core(s, "stale_stream", agent_lock) + + # Existing messages preserved untouched + assert len(s.messages) == 2, ( + f"Expected 2 messages (original + error marker), got {len(s.messages)}" + ) + assert s.messages[0]["role"] == "user" + assert s.messages[0]["content"] == "existing turn" + assert "Core user msg" not in [m["content"] for m in s.messages], ( + "Core transcript must NOT be synced when messages is non-empty" + ) + + # Exactly one error marker + error_msgs = [m for m in s.messages if m.get("_error")] + assert len(error_msgs) == 1 + assert "Previous turn did not complete" in error_msgs[0]["content"] + + # No recovered user turn (messages is non-empty, so skip that) + recovered_msgs = [m for m in s.messages if m.get("_recovered")] + assert len(recovered_msgs) == 0 + + # Pending fields fully cleared + assert s.pending_user_message is None + assert s.pending_attachments == [] + assert s.pending_started_at is None + assert s.active_stream_id is None + + +class TestLastResortSyncDelegation: + """_last_resort_sync_from_core delegates to the shared helpers + _get_profile_home and _apply_core_sync_or_error_marker, ensuring + consistent behavior between the streaming exit path and the cache-miss + repair path.""" + + def test_uses_shared_get_profile_home(self, hermes_home, monkeypatch): + """_last_resort_sync_from_core uses _get_profile_home for path + resolution, not a local ImportError fallback.""" + s = _make_stale_session() + s.save() + + agent_lock = config._get_session_agent_lock(s.session_id) + + # Patch _get_profile_home to verify it's called + called = [] + original_get_profile_home = models._get_profile_home + + def tracking_get_profile_home(profile): + called.append(profile) + return original_get_profile_home(profile) + + with patch.object(models, "_get_profile_home", tracking_get_profile_home): + _register_active_stream("stream_1") + streaming._last_resort_sync_from_core(s, "stream_1", agent_lock) + + assert len(called) == 1, "_get_profile_home should have been called once" + assert called[0] == s.profile + + def test_uses_shared_apply_core_sync_or_error_marker(self, hermes_home, monkeypatch): + """_last_resort_sync_from_core delegates to _apply_core_sync_or_error_marker + instead of duplicating the logic.""" + s = _make_stale_session() + s.save() + + agent_lock = config._get_session_agent_lock(s.session_id) + + # Patch _apply_core_sync_or_error_marker to verify it's called + called = [] + original_fn = models._apply_core_sync_or_error_marker + + def tracking_fn(session, core_path, stream_id_for_recheck=None, **kwargs): + called.append((session.session_id, stream_id_for_recheck, kwargs)) + return original_fn(session, core_path, stream_id_for_recheck, **kwargs) + + with patch.object(models, "_apply_core_sync_or_error_marker", tracking_fn): + _register_active_stream("stream_1") + streaming._last_resort_sync_from_core(s, "stream_1", agent_lock) + + assert len(called) == 1, "_apply_core_sync_or_error_marker should have been called" + assert called[0][0] == s.session_id + assert called[0][1] == "stream_1" + assert called[0][2] == {"require_stream_dead": False} + + def test_core_sync_from_last_resort(self, hermes_home, monkeypatch): + """When a core transcript exists, _last_resort_sync_from_core syncs + messages from it (end-to-end test via shared helper).""" + s = _make_stale_session(pending_msg="My question") + s.save() + + # Write core transcript with messages + core_messages = [ + {"role": "user", "content": "My question"}, + {"role": "assistant", "content": "Here is the answer"}, + ] + _write_core_transcript(hermes_home, s.session_id, core_messages) + + agent_lock = config._get_session_agent_lock(s.session_id) + _register_active_stream("stream_1") + + streaming._last_resort_sync_from_core(s, "stream_1", agent_lock) + + assert len(s.messages) == 2 + assert s.messages[0]["content"] == "My question" + assert s.messages[1]["content"] == "Here is the answer" + assert s.pending_user_message is None + assert s.active_stream_id is None + + +class TestCheckpointOrdering: + """In _run_agent_streaming's outer finally block, checkpoint stop/join + happens BEFORE _last_resort_sync_from_core. This prevents deadlock because + the checkpoint thread holds the per-session lock.""" + + def test_checkpoint_stops_before_recovery_code_structure(self): + """Verify the code ordering in the outer finally block of + _run_agent_streaming: checkpoint stop appears before + _last_resort_sync_from_core.""" + import inspect + source = inspect.getsource(streaming._run_agent_streaming) + + # Find the finally block + finally_idx = source.rfind("finally:") + assert finally_idx != -1, "Could not find 'finally:' in _run_agent_streaming" + + finally_block = source[finally_idx:] + + # _checkpoint_stop should appear before _last_resort_sync_from_core + ckpt_pos = finally_block.find("_checkpoint_stop") + recovery_pos = finally_block.find("_last_resort_sync_from_core") + + assert ckpt_pos != -1, "Could not find _checkpoint_stop in finally block" + assert recovery_pos != -1, "Could not find _last_resort_sync_from_core in finally block" + assert ckpt_pos < recovery_pos, ( + f"_checkpoint_stop (pos {ckpt_pos}) must appear BEFORE " + f"_last_resort_sync_from_core (pos {recovery_pos}) in finally block" + ) + + +# ── Integration: _repair_stale_pending end-to-end ──────────────────────────── + +class TestRepairStalePendingIntegration: + """End-to-end tests for _repair_stale_pending (cache-miss repair path).""" + + def test_repairs_when_core_exists(self, hermes_home, monkeypatch): + """Full repair path: stale session with core transcript gets synced.""" + s = _make_stale_session() + s.save() + + core_messages = [ + {"role": "user", "content": "Hello"}, + {"role": "assistant", "content": "World"}, + ] + _write_core_transcript(hermes_home, s.session_id, core_messages) + + result = _repair_stale_pending(s) + assert result is True + assert len(s.messages) == 2 + assert s.pending_user_message is None + + def test_repairs_when_core_missing(self, hermes_home, monkeypatch): + """Full repair path: stale session without core gets error marker + and recovered user turn.""" + s = _make_stale_session(pending_msg="Lost message") + s.save() + + # No core transcript written + result = _repair_stale_pending(s) + assert result is True + + # Should have recovered user turn + error marker + assert len(s.messages) == 2 + user_msgs = [m for m in s.messages if m["role"] == "user"] + assert len(user_msgs) == 1 + assert user_msgs[0]["content"] == "Lost message" + assert user_msgs[0].get("_recovered") is True + + error_msgs = [m for m in s.messages if m.get("_error")] + assert len(error_msgs) == 1 + + def test_skips_when_messages_nonempty(self, hermes_home, monkeypatch): + """Pre-check: if messages is non-empty, repair is skipped entirely.""" + s = _make_session(messages=[{"role": "user", "content": "hi"}]) + s.pending_user_message = "more" + s.active_stream_id = "stream_1" + + result = _repair_stale_pending(s) + assert result is False + + def test_skips_when_stream_alive(self, hermes_home, monkeypatch): + """Pre-check: if the stream is still alive in STREAMS, repair is skipped.""" + s = _make_stale_session(stream_id="live_stream") + s.save() + + _register_active_stream("live_stream") + + try: + result = _repair_stale_pending(s) + assert result is False + finally: + with config.STREAMS_LOCK: + config.STREAMS.pop("live_stream", None) + + def test_skips_when_no_pending(self, hermes_home, monkeypatch): + """Pre-check: if pending_user_message is None, repair is skipped.""" + s = _make_session(messages=[]) + s.pending_user_message = None + s.active_stream_id = "stream_1" + + result = _repair_stale_pending(s) + assert result is False + + +# ── Core sync with metadata fields ─────────────────────────────────────────── + +class TestCoreSyncMetadata: + """When syncing from core transcript, token/cost metadata is carried over.""" + + def test_syncs_token_and_cost_fields(self, hermes_home, monkeypatch): + """Core transcript with input_tokens/output_tokens/estimated_cost + has those fields copied to the session.""" + s = _make_stale_session() + lock = config._get_session_agent_lock(s.session_id) + + core_messages = [ + {"role": "user", "content": "Hello"}, + {"role": "assistant", "content": "World"}, + ] + core_path = _write_core_transcript( + hermes_home, s.session_id, core_messages, + input_tokens=100, output_tokens=50, estimated_cost=0.05, + ) + + with lock: + result = _apply_core_sync_or_error_marker( + s, core_path, stream_id_for_recheck="stream_1", + ) + + assert result is True + assert s.input_tokens == 100 + assert s.output_tokens == 50 + assert s.estimated_cost == 0.05 + + def test_core_empty_messages_falls_through_to_recovery(self, hermes_home, monkeypatch): + """If core transcript exists but messages is empty, the recovery path + (restoring pending user message + error marker) is taken instead.""" + s = _make_stale_session(pending_msg="My question") + lock = config._get_session_agent_lock(s.session_id) + + # Core exists but has empty messages + core_path = _write_core_transcript(hermes_home, s.session_id, []) + + with lock: + result = _apply_core_sync_or_error_marker( + s, core_path, stream_id_for_recheck="stream_1", + ) + + assert result is True + # Should have recovered user turn + error marker + user_msgs = [m for m in s.messages if m["role"] == "user"] + assert len(user_msgs) == 1 + assert user_msgs[0]["content"] == "My question" + assert user_msgs[0].get("_recovered") is True diff --git a/tests/test_title_aux_routing.py b/tests/test_title_aux_routing.py index f18647e0..3027aef7 100644 --- a/tests/test_title_aux_routing.py +++ b/tests/test_title_aux_routing.py @@ -3,8 +3,8 @@ Covers: - _aux_title_configured() broad detection (provider, model, base_url) - generate_title_raw_via_aux() reads timeout from config instead of hardcoding 15.0 - - aux→agent fallback triggers on 'llm_invalid_aux' status (Comment 1) - - _aux_title_timeout rejects zero, negative, and non-numeric values (Comment 4) + - aux→agent fallback triggers on 'llm_invalid_aux' status + - _aux_title_timeout rejects zero, negative, and non-numeric values """ import sys import types @@ -313,7 +313,7 @@ class TestReasoningModelTitleGeneration(unittest.TestCase): class TestAuxTitleTimeoutEdgeCases(unittest.TestCase): - """Comment 4: _aux_title_timeout must reject zero, negative, and non-numeric values.""" + """_aux_title_timeout must reject zero, negative, and non-numeric values.""" def _call(self, tg_config, default=15.0): from api.streaming import _aux_title_timeout @@ -352,7 +352,7 @@ class TestAuxTitleTimeoutEdgeCases(unittest.TestCase): class TestAuxInvalidAuxTriggersAgentFallback(unittest.TestCase): - """Comment 1: when aux returns llm_invalid_aux, the agent route must be tried as fallback. + """When aux returns llm_invalid_aux, the agent route must be tried as fallback. Pins the behaviour so the fallback tuple in _run_background_title_update stays synchronised with the statuses that _generate_llm_session_title_via_aux