1801 lines
75 KiB
Python
1801 lines
75 KiB
Python
"""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
|
|
from api.run_journal import append_run_event
|
|
|
|
|
|
# ── 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()
|
|
config.ACTIVE_RUNS.clear()
|
|
yield
|
|
config.STREAMS.clear()
|
|
config.CANCEL_FLAGS.clear()
|
|
config.AGENT_INSTANCES.clear()
|
|
config.STREAM_PARTIAL_TEXT.clear()
|
|
config.ACTIVE_RUNS.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()
|
|
|
|
|
|
def _register_active_run(stream_id):
|
|
"""Register stream_id as an active worker without an attached SSE stream."""
|
|
config.register_active_run(stream_id, session_id="stale_sid", phase="running")
|
|
|
|
|
|
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)"
|
|
)
|
|
assert sid not in models.SESSIONS, (
|
|
"Still-stale session should not stay pinned in cache after "
|
|
"lock-contended repair skip"
|
|
)
|
|
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'}"
|
|
)
|
|
|
|
def test_lock_contended_skip_retries_on_next_cache_miss(self, hermes_home, monkeypatch):
|
|
"""A lock-contended repair skip should not become stuck forever.
|
|
|
|
The first get_session() call happens while the per-session lock is held,
|
|
so repair must bail to avoid deadlock. The still-stale object is evicted
|
|
from SESSIONS, allowing a later get_session() after lock release to reload
|
|
from disk and repair normally.
|
|
"""
|
|
sid = "stale_retry_sid"
|
|
s = _make_stale_session(session_id=sid, pending_msg="Recover me")
|
|
s.save()
|
|
_write_core_transcript(
|
|
hermes_home,
|
|
sid,
|
|
[
|
|
{"role": "user", "content": "Recover me"},
|
|
{"role": "assistant", "content": "Recovered answer"},
|
|
],
|
|
)
|
|
models.SESSIONS.pop(sid, None)
|
|
|
|
lock = config._get_session_agent_lock(sid)
|
|
assert lock.acquire(blocking=False)
|
|
try:
|
|
skipped = models.get_session(sid)
|
|
assert skipped.pending_user_message == "Recover me"
|
|
assert sid not in models.SESSIONS
|
|
finally:
|
|
lock.release()
|
|
|
|
repaired = models.get_session(sid)
|
|
assert repaired.pending_user_message is None
|
|
assert repaired.active_stream_id is None
|
|
assert [m["content"] for m in repaired.messages] == ["Recover me", "Recovered answer"]
|
|
assert models.SESSIONS.get(sid) is repaired
|
|
|
|
|
|
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
|
|
a clear restart interruption marker — 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_pending_message_recovered_into_context_messages(self, hermes_home, monkeypatch):
|
|
"""A recovered pending prompt must remain visible to the next agent turn.
|
|
|
|
Sessions that have been auto-compressed feed context_messages to the
|
|
model, not the full display transcript. If stale-stream repair appends
|
|
the recovered user prompt only to messages, the user can see the prompt
|
|
in WebUI but the next agent turn cannot.
|
|
"""
|
|
s = _make_session(
|
|
messages=[{"role": "user", "content": "older visible turn"}],
|
|
context_messages=[
|
|
{"role": "user", "content": "older context turn"},
|
|
{"role": "assistant", "content": "older context answer"},
|
|
],
|
|
)
|
|
s.pending_user_message = "Clip this article https://example.com/post"
|
|
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
|
|
assert any(
|
|
m.get("role") == "user"
|
|
and m.get("content") == "Clip this article https://example.com/post"
|
|
and m.get("_recovered") is True
|
|
for m in s.messages
|
|
)
|
|
assert any(
|
|
m.get("role") == "user"
|
|
and m.get("content") == "Clip this article https://example.com/post"
|
|
for m in s.context_messages
|
|
), "Recovered pending user turn must be included in model context."
|
|
|
|
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 "Response interrupted" in content
|
|
assert "live response stream stopped" in content
|
|
assert "WebUI process restarted" not in content
|
|
# The marker now arms the lazy-retry hook when a stream id is known
|
|
# ("Recovering the partial output… reload to retry."). The legacy
|
|
# "user message above was preserved" wording is reserved for the
|
|
# no-stream-id repair case; the post-retry-give-up case demotes to
|
|
# the neutral "Partial output may have been lost." wording instead.
|
|
assert (
|
|
"user message above was preserved" in content
|
|
or "Recovering the partial output" in content
|
|
)
|
|
assert error_msgs[0].get("type") == "interrupted"
|
|
|
|
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 result == Path(os.environ["HERMES_HOME"]).expanduser()
|
|
|
|
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 preserves existing messages when
|
|
session.messages is non-empty, while still recovering the pending user turn
|
|
before clearing stale stream runtime fields."""
|
|
|
|
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 recovers the pending user turn,
|
|
clears the pending fields, and appends an error marker."""
|
|
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, pending turn recovered, then marker appended
|
|
assert len(s.messages) == 3 # original + recovered user turn + error marker
|
|
assert s.messages[0]["content"] == "hello"
|
|
assert s.messages[1]["role"] == "user"
|
|
assert s.messages[1]["content"] == "Another question"
|
|
assert s.messages[1].get("_recovered") is True
|
|
# Error marker appended
|
|
assert s.messages[2].get("_error") is True
|
|
# Pending fields cleared
|
|
assert s.pending_user_message is None
|
|
assert s.active_stream_id is None
|
|
|
|
def test_repeated_pending_prompt_uses_current_checkpoint_identity(self, hermes_home, monkeypatch):
|
|
"""Historical identical text must not suppress the current pending turn."""
|
|
s = _make_session(
|
|
messages=[
|
|
{"role": "user", "content": "repeat this", "timestamp": 111, "attachments": [{"name": "old.png"}]},
|
|
{"role": "assistant", "content": "Earlier answer"},
|
|
],
|
|
context_messages=[
|
|
{"role": "user", "content": "repeat this", "timestamp": 111, "attachments": [{"name": "old.png"}]},
|
|
{"role": "assistant", "content": "Earlier answer"},
|
|
],
|
|
)
|
|
s.pending_user_message = "repeat this"
|
|
s.pending_started_at = 222.0
|
|
s.pending_attachments = [{"name": "current.png"}]
|
|
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
|
|
users = [m for m in s.messages if m.get("role") == "user"]
|
|
assert len(users) == 2
|
|
assert users[-1]["timestamp"] == 222
|
|
assert users[-1]["attachments"] == [{"name": "current.png"}]
|
|
context_users = [m for m in s.context_messages if m.get("role") == "user"]
|
|
assert len(context_users) == 2
|
|
assert context_users[-1]["timestamp"] == 222
|
|
assert context_users[-1]["attachments"] == [{"name": "current.png"}]
|
|
|
|
def test_recovered_assistant_context_row_still_deduplicates(self):
|
|
s = _make_session(
|
|
context_messages=[{"role": "assistant", "content": "Recovered answer"}],
|
|
)
|
|
recovered = {
|
|
"role": "assistant",
|
|
"content": "Recovered answer",
|
|
"timestamp": 222,
|
|
}
|
|
|
|
models._append_recovered_turn_to_context(s, recovered)
|
|
|
|
assert s.context_messages == [{"role": "assistant", "content": "Recovered answer"}]
|
|
|
|
def test_recovered_user_context_row_requires_tail_checkpoint(self):
|
|
s = _make_session(
|
|
context_messages=[
|
|
{"role": "user", "content": "repeat this", "timestamp": 222, "attachments": [{"name": "same.png"}]},
|
|
{"role": "assistant", "content": "Earlier answer"},
|
|
],
|
|
)
|
|
recovered = {
|
|
"role": "user",
|
|
"content": "repeat this",
|
|
"timestamp": 222,
|
|
"attachments": [{"name": "same.png"}],
|
|
"_recovered": True,
|
|
}
|
|
|
|
models._append_recovered_turn_to_context(s, recovered)
|
|
|
|
context_users = [m for m in s.context_messages if m.get("role") == "user"]
|
|
assert len(context_users) == 2
|
|
assert context_users[-1]["timestamp"] == 222
|
|
|
|
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
|
|
preserves existing messages, recovers the pending user turn, and appends
|
|
exactly one error marker without 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 recovers that pending turn before clearing runtime
|
|
fields and appending 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, pending turn recovered, error marker appended
|
|
assert len(s.messages) == 3, (
|
|
f"Expected 3 messages (original + recovered turn + 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 recovered user turn
|
|
recovered_msgs = [m for m in s.messages if m.get("_recovered")]
|
|
assert len(recovered_msgs) == 1
|
|
assert recovered_msgs[0]["role"] == "user"
|
|
assert recovered_msgs[0]["content"] == "Stuck draft"
|
|
assert recovered_msgs[0]["attachments"] == [{"type": "image", "name": "screenshot.png"}]
|
|
|
|
# Exactly one error marker
|
|
error_msgs = [m for m in s.messages if m.get("_error")]
|
|
assert len(error_msgs) == 1
|
|
assert "Response interrupted" in error_msgs[0]["content"]
|
|
assert "live response stream stopped" in error_msgs[0]["content"]
|
|
assert "WebUI process restarted" not in error_msgs[0]["content"]
|
|
assert error_msgs[0].get("type") == "interrupted"
|
|
|
|
# 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
|
|
|
|
def test_journaled_partial_output_is_recovered_before_interrupted_marker(self, hermes_home, monkeypatch):
|
|
"""When a WebUI restart leaves a dead stream with journaled partial
|
|
output, repair should not collapse the user-visible transcript to only
|
|
a generic interrupted marker."""
|
|
s = _make_session(messages=[{"role": "user", "content": "existing turn"}])
|
|
s.pending_user_message = "Check maintainer activity"
|
|
s.pending_started_at = time.time() - 120
|
|
s.active_stream_id = "journaled_stream"
|
|
s.save()
|
|
|
|
append_run_event(
|
|
s.session_id,
|
|
"journaled_stream",
|
|
"token",
|
|
{"text": "I will check GitHub first."},
|
|
)
|
|
append_run_event(
|
|
s.session_id,
|
|
"journaled_stream",
|
|
"tool",
|
|
{
|
|
"name": "terminal",
|
|
"preview": "gh pr list --repo nesquena/hermes-webui",
|
|
"args": {"command": "gh pr list --repo nesquena/hermes-webui"},
|
|
},
|
|
)
|
|
append_run_event(
|
|
s.session_id,
|
|
"journaled_stream",
|
|
"tool_complete",
|
|
{"name": "terminal", "duration": 1.2, "is_error": False},
|
|
)
|
|
append_run_event(
|
|
s.session_id,
|
|
"journaled_stream",
|
|
"token",
|
|
{"text": "The first check finished before the restart."},
|
|
)
|
|
|
|
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="journaled_stream",
|
|
)
|
|
|
|
assert result is True
|
|
contents = [m.get("content", "") for m in s.messages]
|
|
assert any("I will check GitHub first." in c for c in contents)
|
|
assert any("The first check finished before the restart." in c for c in contents)
|
|
assert s.tool_calls, "journaled tool starts should become visible settled tool cards"
|
|
assert s.tool_calls[0]["name"] == "terminal"
|
|
assert s.tool_calls[0]["done"] is True
|
|
assert s.tool_calls[0]["assistant_msg_idx"] < len(s.messages)
|
|
error_msgs = [m for m in s.messages if m.get("_error")]
|
|
assert len(error_msgs) == 1
|
|
assert "partial output above was recovered" in error_msgs[0]["content"]
|
|
assert "no agent output was recovered" not in error_msgs[0]["content"]
|
|
|
|
def test_journal_recovery_restores_reasoning_only_as_display_metadata(
|
|
self, hermes_home, monkeypatch,
|
|
):
|
|
"""Run-journal repair keeps live Thinking without making it chat prose
|
|
or provider-facing history."""
|
|
s = _make_session(messages=[{"role": "user", "content": "existing turn"}])
|
|
s.pending_user_message = "Keep going"
|
|
s.pending_started_at = time.time() - 120
|
|
s.active_stream_id = "reasoning_only_stream"
|
|
s.save()
|
|
|
|
append_run_event(
|
|
s.session_id,
|
|
"reasoning_only_stream",
|
|
"reasoning",
|
|
{"text": "private scratchpad text"},
|
|
)
|
|
|
|
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="reasoning_only_stream",
|
|
)
|
|
|
|
assert result is True
|
|
contents = [m.get("content", "") for m in s.messages]
|
|
assert not any("private scratchpad text" in c for c in contents)
|
|
reasoning_msgs = [m for m in s.messages if m.get("reasoning")]
|
|
assert [m.get("reasoning") for m in reasoning_msgs] == ["private scratchpad text"]
|
|
error_msgs = [m for m in s.messages if m.get("_error")]
|
|
assert len(error_msgs) == 1
|
|
assert error_msgs[0].get("_pending_journal_recovery") is not True
|
|
assert "partial output above was recovered" in error_msgs[0]["content"]
|
|
provider_history = streaming._sanitize_messages_for_api(s.messages)
|
|
assert "private scratchpad text" not in json.dumps(provider_history)
|
|
|
|
def test_journal_recovery_keeps_consecutive_tools_on_one_anchor(self, hermes_home, monkeypatch):
|
|
"""Consecutive journaled tools without an intervening visible update
|
|
should recover as one activity group instead of repeated empty anchors."""
|
|
s = _make_session(messages=[{"role": "user", "content": "existing turn"}])
|
|
s.pending_user_message = "Inspect files"
|
|
s.pending_started_at = time.time() - 120
|
|
s.active_stream_id = "tool_burst_stream"
|
|
s.save()
|
|
|
|
append_run_event(
|
|
s.session_id,
|
|
"tool_burst_stream",
|
|
"token",
|
|
{"text": "I will inspect the relevant files first."},
|
|
)
|
|
for name in ("search_files", "read_file"):
|
|
append_run_event(
|
|
s.session_id,
|
|
"tool_burst_stream",
|
|
"tool",
|
|
{"name": name, "preview": name, "args": {"query": "stream recovery"}},
|
|
)
|
|
|
|
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="tool_burst_stream",
|
|
)
|
|
|
|
assert result is True
|
|
assert len(s.tool_calls) == 2
|
|
assert s.tool_calls[0]["assistant_msg_idx"] == s.tool_calls[1]["assistant_msg_idx"]
|
|
|
|
def test_core_sync_branch_recovers_visible_journal_output(self, hermes_home, monkeypatch):
|
|
"""The empty-sidecar + populated-core repair branch should still restore
|
|
already-journaled visible output from the interrupted stream."""
|
|
s = _make_session(messages=[])
|
|
s.pending_user_message = "Check maintainer activity"
|
|
s.pending_started_at = time.time() - 120
|
|
s.active_stream_id = "core_journal_stream"
|
|
s.save()
|
|
|
|
core_messages = [
|
|
{"role": "user", "content": "Earlier question"},
|
|
{"role": "assistant", "content": "Earlier answer"},
|
|
]
|
|
core_path = _write_core_transcript(hermes_home, s.session_id, core_messages)
|
|
|
|
append_run_event(
|
|
s.session_id,
|
|
"core_journal_stream",
|
|
"token",
|
|
{"text": "I will check GitHub first."},
|
|
)
|
|
append_run_event(
|
|
s.session_id,
|
|
"core_journal_stream",
|
|
"tool",
|
|
{
|
|
"name": "terminal",
|
|
"preview": "gh pr list --repo nesquena/hermes-webui",
|
|
"args": {"command": "gh pr list --repo nesquena/hermes-webui"},
|
|
},
|
|
)
|
|
append_run_event(
|
|
s.session_id,
|
|
"core_journal_stream",
|
|
"tool_complete",
|
|
{"name": "terminal", "duration": 1.2, "is_error": False},
|
|
)
|
|
append_run_event(
|
|
s.session_id,
|
|
"core_journal_stream",
|
|
"token",
|
|
{"text": "The first check finished before the restart."},
|
|
)
|
|
|
|
result = _apply_core_sync_or_error_marker(
|
|
s,
|
|
core_path,
|
|
stream_id_for_recheck="core_journal_stream",
|
|
)
|
|
|
|
assert result is True
|
|
contents = [m.get("content", "") for m in s.messages]
|
|
assert contents[:2] == [m["content"] for m in core_messages]
|
|
recovered_users = [m for m in s.messages if m.get("_recovered")]
|
|
assert len(recovered_users) == 1
|
|
assert recovered_users[0]["role"] == "user"
|
|
assert recovered_users[0]["content"] == "Check maintainer activity"
|
|
assert any("I will check GitHub first." in c for c in contents)
|
|
assert any("The first check finished before the restart." in c for c in contents)
|
|
assert s.tool_calls, "journaled tool starts should become visible settled tool cards"
|
|
assert s.tool_calls[0]["name"] == "terminal"
|
|
error_msgs = [m for m in s.messages if m.get("_error")]
|
|
assert len(error_msgs) == 1
|
|
assert "partial output above was recovered" in error_msgs[0]["content"]
|
|
assert s.pending_user_message is None
|
|
assert s.active_stream_id is None
|
|
|
|
def test_finished_worker_can_supersede_its_own_interrupted_marker(self):
|
|
"""A live worker that finishes after stale repair should be allowed to
|
|
replace the recovery marker for the same user turn."""
|
|
s = _make_session(
|
|
messages=[
|
|
{"role": "user", "content": "deploy"},
|
|
models._interrupted_recovery_marker(),
|
|
]
|
|
)
|
|
s.active_stream_id = None
|
|
s.pending_user_message = None
|
|
s.pending_attachments = []
|
|
|
|
assert streaming._stream_writeback_can_supersede_recovery_marker(s, "deploy")
|
|
|
|
def test_finished_worker_does_not_supersede_after_newer_turn_appended(self):
|
|
"""Once a follow-up turn changes the visible tail, stale writeback stays
|
|
blocked so old workers cannot overwrite newer transcript state."""
|
|
s = _make_session(
|
|
messages=[
|
|
{"role": "user", "content": "deploy"},
|
|
models._interrupted_recovery_marker(),
|
|
{"role": "user", "content": "what happened?"},
|
|
{"role": "assistant", "content": "I checked the deployment status."},
|
|
]
|
|
)
|
|
s.active_stream_id = None
|
|
s.pending_user_message = None
|
|
s.pending_attachments = []
|
|
|
|
assert not streaming._stream_writeback_can_supersede_recovery_marker(s, "deploy")
|
|
|
|
def test_finished_worker_does_not_supersede_different_user_turn(self):
|
|
"""The supersede path is tied to the pending prompt that was repaired."""
|
|
s = _make_session(
|
|
messages=[
|
|
{"role": "user", "content": "deploy"},
|
|
models._interrupted_recovery_marker(),
|
|
]
|
|
)
|
|
s.active_stream_id = None
|
|
s.pending_user_message = None
|
|
s.pending_attachments = []
|
|
|
|
assert not streaming._stream_writeback_can_supersede_recovery_marker(s, "ship it")
|
|
|
|
def test_core_sync_branch_does_not_duplicate_journal_output_already_in_core(
|
|
self, hermes_home, monkeypatch
|
|
):
|
|
"""If the core transcript already contains the same visible output, the
|
|
journal repair must not append a second copy."""
|
|
s = _make_session(messages=[])
|
|
s.pending_user_message = "Check maintainer activity"
|
|
s.pending_started_at = time.time() - 120
|
|
s.active_stream_id = "duplicate_core_journal_stream"
|
|
s.save()
|
|
|
|
core_messages = [
|
|
{"role": "user", "content": "Check maintainer activity"},
|
|
{"role": "assistant", "content": "I will check GitHub first."},
|
|
]
|
|
core_tool_calls = [
|
|
{
|
|
"name": "terminal",
|
|
"preview": "gh pr list --repo nesquena/hermes-webui",
|
|
"snippet": "gh pr list --repo nesquena/hermes-webui",
|
|
"assistant_msg_idx": 1,
|
|
"done": True,
|
|
},
|
|
]
|
|
core_path = _write_core_transcript(
|
|
hermes_home,
|
|
s.session_id,
|
|
core_messages,
|
|
tool_calls=core_tool_calls,
|
|
)
|
|
|
|
append_run_event(
|
|
s.session_id,
|
|
"duplicate_core_journal_stream",
|
|
"token",
|
|
{"text": "I will check GitHub first."},
|
|
)
|
|
append_run_event(
|
|
s.session_id,
|
|
"duplicate_core_journal_stream",
|
|
"tool",
|
|
{
|
|
"name": "terminal",
|
|
"preview": "gh pr list --repo nesquena/hermes-webui",
|
|
"args": {"command": "gh pr list --repo nesquena/hermes-webui"},
|
|
},
|
|
)
|
|
|
|
result = _apply_core_sync_or_error_marker(
|
|
s,
|
|
core_path,
|
|
stream_id_for_recheck="duplicate_core_journal_stream",
|
|
)
|
|
|
|
assert result is True
|
|
contents = [m.get("content", "") for m in s.messages]
|
|
assert contents.count("I will check GitHub first.") == 1
|
|
assert len(s.tool_calls) == 1
|
|
assert s.tool_calls[0]["name"] == "terminal"
|
|
assert not [m for m in s.messages if m.get("_error")]
|
|
|
|
|
|
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_recovers_when_messages_nonempty(self, hermes_home, monkeypatch):
|
|
"""Pre-check: if messages is non-empty, repair still preserves the
|
|
pending user turn instead of silently discarding it."""
|
|
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 True
|
|
assert [m["content"] for m in s.messages if m["role"] == "user"] == ["hi", "more"]
|
|
assert s.messages[1].get("_recovered") is True
|
|
assert any(m.get("_error") for m in s.messages)
|
|
|
|
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_worker_alive_without_sse_stream(self, hermes_home, monkeypatch):
|
|
"""Pre-check: if ACTIVE_RUNS still owns the worker, repair is skipped.
|
|
|
|
STREAMS is the browser/SSE observation layer and may disappear while the
|
|
backend worker is still running. A detached-but-active worker must not be
|
|
mistaken for a WebUI restart or crashed turn.
|
|
"""
|
|
s = _make_stale_session(stream_id="detached_stream")
|
|
s.save()
|
|
|
|
_register_active_run("detached_stream")
|
|
|
|
assert "detached_stream" in _active_stream_ids()
|
|
result = _repair_stale_pending(s)
|
|
assert result is False
|
|
assert s.pending_user_message == "Hello hermes"
|
|
assert s.active_stream_id == "detached_stream"
|
|
assert s.messages == []
|
|
|
|
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
|
|
|
|
|
|
# ── Lazy run-journal recovery (read-side self-heal) ─────────────────────────
|
|
|
|
class TestInterruptedRecoveryMarker:
|
|
"""Pure-function tests for _interrupted_recovery_marker(pending_retry=…)."""
|
|
|
|
def test_marker_recovered_output_excludes_pending_retry_flag(self):
|
|
marker = models._interrupted_recovery_marker(recovered_output=True)
|
|
assert marker["_error"] is True
|
|
assert marker["type"] == "interrupted"
|
|
assert "_pending_journal_recovery" not in marker
|
|
assert "recovered from the run journal" in marker["content"]
|
|
|
|
def test_marker_pending_retry_sets_flag_and_wording(self):
|
|
marker = models._interrupted_recovery_marker(pending_retry=True)
|
|
assert marker.get("_pending_journal_recovery") is True
|
|
assert "Recovering the partial output" in marker["content"]
|
|
assert "no agent output was recovered" not in marker["content"]
|
|
|
|
def test_marker_recovered_output_beats_pending_retry(self):
|
|
marker = models._interrupted_recovery_marker(
|
|
recovered_output=True, pending_retry=True,
|
|
)
|
|
assert "_pending_journal_recovery" not in marker
|
|
assert "recovered from the run journal" in marker["content"]
|
|
|
|
def test_marker_default_wording_unchanged_for_no_output_no_retry(self):
|
|
marker = models._interrupted_recovery_marker()
|
|
assert "_pending_journal_recovery" not in marker
|
|
assert "no agent output was recovered" in marker["content"]
|
|
|
|
|
|
class TestRetryJournalRecoveryInPlace:
|
|
"""In-place retry helper: promote / increment / give up."""
|
|
|
|
def _make_pending_session(self, hermes_home, stream_id="lazy_stream",
|
|
attempts=0, first_seen_ts=None):
|
|
s = _make_session(messages=[
|
|
{"role": "user", "content": "earlier turn"},
|
|
{"role": "assistant", "content": "earlier reply"},
|
|
{"role": "user", "content": "Stuck draft", "_recovered": True},
|
|
])
|
|
marker = models._interrupted_recovery_marker(pending_retry=True)
|
|
marker["_journal_retry_stream_id"] = stream_id
|
|
marker["_journal_retry_attempts"] = attempts
|
|
marker["_journal_retry_first_seen_ts"] = (
|
|
first_seen_ts if first_seen_ts is not None else int(time.time())
|
|
)
|
|
s.messages.append(marker)
|
|
s.save()
|
|
return s
|
|
|
|
def test_promotes_marker_when_journal_now_available(self, hermes_home, monkeypatch):
|
|
stream_id = "lazy_stream_promote"
|
|
s = self._make_pending_session(hermes_home, stream_id=stream_id)
|
|
marker_before = s.messages[-1]
|
|
|
|
append_run_event(s.session_id, stream_id, "token", {"text": "Late tokens."})
|
|
append_run_event(
|
|
s.session_id, stream_id, "tool",
|
|
{"name": "terminal", "preview": "echo", "args": {"cmd": "echo"}},
|
|
)
|
|
|
|
ok = models._retry_journal_recovery_in_place(s)
|
|
assert ok is True
|
|
|
|
# Marker promoted in place and meta stripped
|
|
marker_idx = next(
|
|
i for i, m in enumerate(s.messages)
|
|
if m.get("type") == "interrupted" and m.get("_error")
|
|
)
|
|
promoted = s.messages[marker_idx]
|
|
assert promoted is marker_before
|
|
assert "recovered from the run journal" in promoted["content"]
|
|
assert "_pending_journal_recovery" not in promoted
|
|
assert "_journal_retry_stream_id" not in promoted
|
|
assert "_journal_retry_attempts" not in promoted
|
|
assert "_journal_retry_first_seen_ts" not in promoted
|
|
|
|
# Journaled rows reordered ABOVE the marker, preserving order
|
|
before_marker = s.messages[:marker_idx]
|
|
recovered = [m for m in before_marker if m.get("_recovered_from_run_journal")]
|
|
assert recovered, "journaled assistant rows must sit above the marker"
|
|
assert any("Late tokens." in m.get("content", "") for m in recovered)
|
|
# tool_call.assistant_msg_idx still points into the valid range
|
|
for tc in s.tool_calls or []:
|
|
idx = tc.get("assistant_msg_idx")
|
|
if isinstance(idx, int):
|
|
assert 0 <= idx < len(s.messages)
|
|
|
|
def test_increments_attempts_when_journal_still_empty(self, hermes_home, monkeypatch):
|
|
stream_id = "lazy_stream_increment"
|
|
s = self._make_pending_session(hermes_home, stream_id=stream_id)
|
|
# No append_run_event — journal stays empty.
|
|
ok = models._retry_journal_recovery_in_place(s)
|
|
assert ok is False
|
|
marker = s.messages[-1]
|
|
assert marker.get("_pending_journal_recovery") is True
|
|
assert marker.get("_journal_retry_attempts") == 1
|
|
assert "Recovering the partial output" in marker["content"]
|
|
|
|
def test_demotes_to_neutral_after_max_attempts(self, hermes_home, monkeypatch):
|
|
stream_id = "lazy_stream_giveup_attempts"
|
|
s = self._make_pending_session(
|
|
hermes_home, stream_id=stream_id,
|
|
attempts=models._JOURNAL_RETRY_MAX_ATTEMPTS,
|
|
)
|
|
ok = models._retry_journal_recovery_in_place(s)
|
|
assert ok is False
|
|
marker = s.messages[-1]
|
|
assert "_pending_journal_recovery" not in marker
|
|
assert "_journal_retry_stream_id" not in marker
|
|
assert "_journal_retry_attempts" not in marker
|
|
assert "_journal_retry_first_seen_ts" not in marker
|
|
assert "Partial output may have been lost" in marker["content"]
|
|
assert "Recovering the partial output" not in marker["content"]
|
|
|
|
def test_demotes_to_neutral_after_giveup_seconds(self, hermes_home, monkeypatch):
|
|
stream_id = "lazy_stream_giveup_age"
|
|
first_seen = int(time.time()) - (models._JOURNAL_RETRY_GIVEUP_SECONDS + 60)
|
|
s = self._make_pending_session(
|
|
hermes_home, stream_id=stream_id, first_seen_ts=first_seen,
|
|
)
|
|
ok = models._retry_journal_recovery_in_place(s)
|
|
assert ok is False
|
|
marker = s.messages[-1]
|
|
assert "_pending_journal_recovery" not in marker
|
|
assert "Partial output may have been lost" in marker["content"]
|
|
|
|
def test_noop_when_no_pending_marker(self, hermes_home, monkeypatch):
|
|
s = _make_session(messages=[
|
|
{"role": "user", "content": "x"},
|
|
{"role": "assistant", "content": "y"},
|
|
])
|
|
s.save()
|
|
spy = []
|
|
original = models._append_journaled_partial_output
|
|
|
|
def _spy(*a, **kw):
|
|
spy.append(1)
|
|
return original(*a, **kw)
|
|
|
|
monkeypatch.setattr(models, "_append_journaled_partial_output", _spy)
|
|
ok = models._retry_journal_recovery_in_place(s)
|
|
assert ok is False
|
|
assert spy == [], "no pending marker → must not call recovery"
|
|
|
|
def test_short_circuit_helper_detects_pending_marker(self, hermes_home, monkeypatch):
|
|
s = _make_session(messages=[
|
|
{"role": "user", "content": "x"},
|
|
{"role": "assistant", "content": "y"},
|
|
])
|
|
assert models._session_has_pending_journal_retry(s) is False
|
|
marker = models._interrupted_recovery_marker(pending_retry=True)
|
|
marker["_journal_retry_stream_id"] = "abc"
|
|
marker["_journal_retry_attempts"] = 0
|
|
marker["_journal_retry_first_seen_ts"] = int(time.time())
|
|
s.messages.append(marker)
|
|
assert models._session_has_pending_journal_retry(s) is True
|
|
|
|
def test_short_circuit_helper_stops_at_normal_assistant(self, hermes_home, monkeypatch):
|
|
s = _make_session(messages=[
|
|
{"role": "user", "content": "x"},
|
|
])
|
|
# An old, already-promoted marker followed by a normal assistant turn —
|
|
# the pending flag belongs to a prior turn that was healed; helper must
|
|
# not loop back into it.
|
|
marker = models._interrupted_recovery_marker(pending_retry=True)
|
|
marker["_journal_retry_stream_id"] = "abc"
|
|
marker["_journal_retry_attempts"] = 0
|
|
marker["_journal_retry_first_seen_ts"] = int(time.time())
|
|
s.messages.append(marker)
|
|
s.messages.append({"role": "user", "content": "later"})
|
|
s.messages.append({"role": "assistant", "content": "later reply"})
|
|
assert models._session_has_pending_journal_retry(s) is False
|
|
|
|
|
|
class TestGetSessionLazyRetryHook:
|
|
"""get_session() must trigger _retry_journal_recovery_in_place on both
|
|
cache-hit and cold-load paths when a pending marker exists, and skip
|
|
quickly when nothing is pending."""
|
|
|
|
def _make_session_with_pending_marker(self, sid="lazy_get", stream_id="st"):
|
|
s = _make_session(session_id=sid, messages=[
|
|
{"role": "user", "content": "u"},
|
|
{"role": "assistant", "content": "a"},
|
|
{"role": "user", "content": "u2", "_recovered": True},
|
|
])
|
|
marker = models._interrupted_recovery_marker(pending_retry=True)
|
|
marker["_journal_retry_stream_id"] = stream_id
|
|
marker["_journal_retry_attempts"] = 0
|
|
marker["_journal_retry_first_seen_ts"] = int(time.time())
|
|
s.messages.append(marker)
|
|
return s
|
|
|
|
def test_triggers_retry_on_cache_hit(self, hermes_home, monkeypatch):
|
|
sid = "lazy_get_cache"
|
|
stream_id = "stream_cache"
|
|
s = self._make_session_with_pending_marker(sid=sid, stream_id=stream_id)
|
|
s.save()
|
|
models.SESSIONS[sid] = s
|
|
append_run_event(sid, stream_id, "token", {"text": "Late."})
|
|
|
|
reloaded = models.get_session(sid)
|
|
assert reloaded is s
|
|
marker = s.messages[-1] if "Recovering" in s.messages[-1]["content"] else next(
|
|
m for m in s.messages
|
|
if m.get("type") == "interrupted" and m.get("_error")
|
|
)
|
|
assert "recovered from the run journal" in marker["content"]
|
|
assert "_pending_journal_recovery" not in marker
|
|
|
|
def test_triggers_retry_on_cold_load(self, hermes_home, monkeypatch):
|
|
sid = "lazy_get_cold"
|
|
stream_id = "stream_cold"
|
|
s = self._make_session_with_pending_marker(sid=sid, stream_id=stream_id)
|
|
s.save()
|
|
models.SESSIONS.pop(sid, None)
|
|
append_run_event(sid, stream_id, "token", {"text": "Late."})
|
|
|
|
reloaded = models.get_session(sid)
|
|
marker = next(
|
|
m for m in reloaded.messages
|
|
if m.get("type") == "interrupted" and m.get("_error")
|
|
)
|
|
assert "recovered from the run journal" in marker["content"]
|
|
assert "_pending_journal_recovery" not in marker
|
|
|
|
def test_short_circuit_when_no_pending_marker(self, hermes_home, monkeypatch):
|
|
sid = "lazy_get_no_pending"
|
|
s = _make_session(session_id=sid, messages=[
|
|
{"role": "user", "content": "x"},
|
|
{"role": "assistant", "content": "y"},
|
|
])
|
|
s.save()
|
|
models.SESSIONS[sid] = s
|
|
spy = []
|
|
monkeypatch.setattr(
|
|
models, "_retry_journal_recovery_in_place",
|
|
lambda session: spy.append(1) or False,
|
|
)
|
|
models.get_session(sid)
|
|
assert spy == []
|
|
|
|
def test_metadata_only_skips_retry(self, hermes_home, monkeypatch):
|
|
sid = "lazy_get_meta"
|
|
stream_id = "stream_meta"
|
|
s = self._make_session_with_pending_marker(sid=sid, stream_id=stream_id)
|
|
s.save()
|
|
models.SESSIONS[sid] = s
|
|
spy = []
|
|
monkeypatch.setattr(
|
|
models, "_retry_journal_recovery_in_place",
|
|
lambda session: spy.append(1) or False,
|
|
)
|
|
models.get_session(sid, metadata_only=True)
|
|
assert spy == [], "metadata_only must skip the lazy-retry helper"
|
|
|
|
|
|
class TestLazyRetryBackwardsCompat:
|
|
"""Pre-fix session shapes must continue to work."""
|
|
|
|
def test_legacy_marker_without_flag_unchanged(self, hermes_home, monkeypatch):
|
|
"""An old session whose marker carries the legacy 'no agent output'
|
|
wording (no flag) must not be touched by get_session()."""
|
|
sid = "legacy_marker_sid"
|
|
s = _make_session(session_id=sid, messages=[
|
|
{"role": "user", "content": "x"},
|
|
{"role": "assistant", "content": "y"},
|
|
{"role": "user", "content": "later", "_recovered": True},
|
|
])
|
|
legacy = models._interrupted_recovery_marker(recovered_output=False)
|
|
s.messages.append(legacy)
|
|
s.save()
|
|
models.SESSIONS.pop(sid, None)
|
|
spy = []
|
|
original = models._append_journaled_partial_output
|
|
monkeypatch.setattr(
|
|
models, "_append_journaled_partial_output",
|
|
lambda *a, **kw: spy.append(1) or original(*a, **kw),
|
|
)
|
|
models.get_session(sid)
|
|
assert spy == [], "legacy marker (no flag) must not re-trigger recovery"
|
|
|
|
def test_pending_retry_marker_round_trips_through_session_save_and_load(
|
|
self, hermes_home, monkeypatch):
|
|
"""All four retry meta keys must survive Session.save() / Session.load()."""
|
|
sid = "round_trip_sid"
|
|
s = _make_session(session_id=sid, messages=[
|
|
{"role": "user", "content": "x"},
|
|
])
|
|
marker = models._interrupted_recovery_marker(pending_retry=True)
|
|
marker["_journal_retry_stream_id"] = "abc"
|
|
marker["_journal_retry_attempts"] = 3
|
|
marker["_journal_retry_first_seen_ts"] = 1779200000
|
|
s.messages.append(marker)
|
|
s.save()
|
|
models.SESSIONS.pop(sid, None)
|
|
|
|
reloaded = Session.load(sid)
|
|
last = reloaded.messages[-1]
|
|
assert last.get("_pending_journal_recovery") is True
|
|
assert last.get("_journal_retry_stream_id") == "abc"
|
|
assert last.get("_journal_retry_attempts") == 3
|
|
assert last.get("_journal_retry_first_seen_ts") == 1779200000
|
|
|
|
|
|
class TestJournalToolDedupeScoping:
|
|
"""`_journal_tool_already_present` must only collapse against tool cards
|
|
recovered from the same stream — a repeated tool (e.g. ``terminal: ls``)
|
|
in a previous turn must NOT pre-empt this turn's recovery."""
|
|
|
|
def test_repeated_tool_in_earlier_turn_does_not_block_recovery(self, hermes_home, monkeypatch):
|
|
sid = "dedupe_scope_sid"
|
|
stream_id = "dedupe_scope_stream"
|
|
s = _make_session(session_id=sid, messages=[
|
|
{"role": "user", "content": "earlier turn"},
|
|
{"role": "assistant", "content": "earlier reply"},
|
|
{"role": "user", "content": "later turn", "_recovered": True},
|
|
])
|
|
# Pre-existing tool card from an earlier turn with same (name, preview)
|
|
# but a different recovered-stream-id. The retry must NOT see this as a
|
|
# hit when dedupe_existing is asked to scope to ``stream_id``.
|
|
s.tool_calls = [
|
|
{
|
|
"name": "terminal",
|
|
"preview": "ls",
|
|
"snippet": "ls",
|
|
"tid": "old-1",
|
|
"_recovered_from_run_journal": True,
|
|
"_recovered_stream_id": "earlier_stream",
|
|
"done": True,
|
|
}
|
|
]
|
|
marker = models._interrupted_recovery_marker(pending_retry=True)
|
|
marker["_journal_retry_stream_id"] = stream_id
|
|
marker["_journal_retry_attempts"] = 0
|
|
marker["_journal_retry_first_seen_ts"] = int(time.time())
|
|
s.messages.append(marker)
|
|
s.save()
|
|
append_run_event(sid, stream_id, "token", {"text": "Listing."})
|
|
append_run_event(
|
|
sid, stream_id, "tool",
|
|
{"name": "terminal", "preview": "ls", "args": {"cmd": "ls"}},
|
|
)
|
|
|
|
ok = models._retry_journal_recovery_in_place(s)
|
|
assert ok is True
|
|
|
|
# Two tool cards now: the old one untouched, plus a new one for this
|
|
# stream. If dedupe were session-wide the new one would be dropped.
|
|
scoped = [
|
|
tc for tc in s.tool_calls
|
|
if tc.get("_recovered_stream_id") == stream_id
|
|
]
|
|
assert len(scoped) == 1
|
|
assert scoped[0]["name"] == "terminal"
|
|
assert scoped[0]["preview"] == "ls"
|
|
# Old tool card is preserved.
|
|
assert any(
|
|
tc.get("_recovered_stream_id") == "earlier_stream"
|
|
for tc in s.tool_calls
|
|
)
|
|
|
|
def test_untagged_tool_still_matches_for_core_transcript_invariant(self, hermes_home, monkeypatch):
|
|
"""Tool cards without ``_recovered_stream_id`` (live tools, or tools
|
|
carried over from the core transcript) match regardless of the
|
|
``stream_id`` argument. This preserves the "core transcript already
|
|
contains this tool, don't duplicate it" invariant the original repair
|
|
path relies on."""
|
|
s = _make_session(messages=[
|
|
{"role": "user", "content": "x"},
|
|
{"role": "assistant", "content": "y"},
|
|
])
|
|
s.tool_calls = [
|
|
{"name": "terminal", "preview": "ls", "snippet": "ls"},
|
|
]
|
|
# No stream_id → legacy session-wide check returns True.
|
|
assert models._journal_tool_already_present(s, "terminal", "ls") is True
|
|
# With a stream_id, untagged tool cards still match (different stream
|
|
# ids only override the match decision when the existing card itself
|
|
# is tagged with a stream id that disagrees).
|
|
assert models._journal_tool_already_present(
|
|
s, "terminal", "ls", stream_id="some_stream",
|
|
) is True
|
|
|
|
def test_tagged_tool_with_different_stream_does_not_match(self, hermes_home, monkeypatch):
|
|
"""A tool card tagged with a different recovered_stream_id must NOT
|
|
be considered a duplicate when the retry is scoped to a different
|
|
stream."""
|
|
s = _make_session(messages=[
|
|
{"role": "user", "content": "x"},
|
|
])
|
|
s.tool_calls = [
|
|
{
|
|
"name": "terminal",
|
|
"preview": "ls",
|
|
"snippet": "ls",
|
|
"_recovered_stream_id": "other_stream",
|
|
},
|
|
]
|
|
assert models._journal_tool_already_present(
|
|
s, "terminal", "ls", stream_id="this_stream",
|
|
) is False
|
|
# But scoping to the same stream id matches.
|
|
assert models._journal_tool_already_present(
|
|
s, "terminal", "ls", stream_id="other_stream",
|
|
) is True
|
|
|
|
|
|
class TestWslPageCacheRace:
|
|
"""Cover the WSL2 / network-FS shape: read_run_events returns empty / errors
|
|
first, recovers on a later call."""
|
|
|
|
def test_first_read_raises_oserror_second_read_succeeds(self, hermes_home, monkeypatch):
|
|
sid = "wsl_race_sid"
|
|
stream_id = "wsl_race_stream"
|
|
s = _make_session(session_id=sid, messages=[
|
|
{"role": "user", "content": "x"},
|
|
])
|
|
s.pending_user_message = "Keep going"
|
|
s.pending_started_at = time.time() - 120
|
|
s.active_stream_id = stream_id
|
|
|
|
# Simulate first read raising IOError, then succeeding.
|
|
import api.run_journal as run_journal
|
|
real = run_journal.read_run_events
|
|
attempts = {"n": 0}
|
|
|
|
def flaky_read(sid_, run_id, **kw):
|
|
attempts["n"] += 1
|
|
if attempts["n"] == 1:
|
|
raise OSError("EIO simulated by test")
|
|
return real(sid_, run_id, **kw)
|
|
|
|
monkeypatch.setattr(run_journal, "read_run_events", flaky_read)
|
|
|
|
core_path = hermes_home / "sessions" / f"session_{sid}.json"
|
|
result = _apply_core_sync_or_error_marker(
|
|
s, core_path, stream_id_for_recheck=stream_id,
|
|
)
|
|
assert result is True
|
|
marker = next(m for m in s.messages if m.get("type") == "interrupted")
|
|
assert marker.get("_pending_journal_recovery") is True
|
|
|
|
# Now write journal events; the next retry call will read them.
|
|
append_run_event(sid, stream_id, "token", {"text": "Came back."})
|
|
ok = models._retry_journal_recovery_in_place(s)
|
|
assert ok is True
|
|
marker_after = next(m for m in s.messages if m.get("type") == "interrupted")
|
|
assert "recovered from the run journal" in marker_after["content"]
|
|
assert "_pending_journal_recovery" not in marker_after
|
|
|
|
def test_journal_grows_between_reads(self, hermes_home, monkeypatch):
|
|
sid = "wsl_grow_sid"
|
|
stream_id = "wsl_grow_stream"
|
|
s = _make_session(session_id=sid, messages=[
|
|
{"role": "user", "content": "x"},
|
|
])
|
|
s.pending_user_message = "Keep going"
|
|
s.pending_started_at = time.time() - 120
|
|
s.active_stream_id = stream_id
|
|
|
|
# First repair pass: nothing visible yet.
|
|
core_path = hermes_home / "sessions" / f"session_{sid}.json"
|
|
result = _apply_core_sync_or_error_marker(
|
|
s, core_path, stream_id_for_recheck=stream_id,
|
|
)
|
|
assert result is True
|
|
marker = next(m for m in s.messages if m.get("type") == "interrupted")
|
|
assert marker.get("_pending_journal_recovery") is True
|
|
|
|
# Journal grows.
|
|
append_run_event(sid, stream_id, "token", {"text": "Partial 1."})
|
|
append_run_event(sid, stream_id, "token", {"text": " Partial 2."})
|
|
append_run_event(
|
|
sid, stream_id, "tool",
|
|
{"name": "terminal", "preview": "ls", "args": {"cmd": "ls"}},
|
|
)
|
|
|
|
ok = models._retry_journal_recovery_in_place(s)
|
|
assert ok is True
|
|
marker_after = next(m for m in s.messages if m.get("type") == "interrupted")
|
|
assert "recovered from the run journal" in marker_after["content"]
|
|
# Both tokens recovered, in order, before the marker.
|
|
marker_idx = s.messages.index(marker_after)
|
|
recovered_text = " ".join(
|
|
m.get("content", "") for m in s.messages[:marker_idx]
|
|
if m.get("_recovered_from_run_journal")
|
|
)
|
|
assert "Partial 1." in recovered_text and "Partial 2." in recovered_text
|
|
|
|
def test_concurrent_get_session_calls_idempotent(self, hermes_home, monkeypatch):
|
|
sid = "wsl_concurrent_sid"
|
|
stream_id = "wsl_concurrent_stream"
|
|
s = _make_session(session_id=sid, messages=[
|
|
{"role": "user", "content": "x"},
|
|
{"role": "assistant", "content": "y"},
|
|
{"role": "user", "content": "later", "_recovered": True},
|
|
])
|
|
marker = models._interrupted_recovery_marker(pending_retry=True)
|
|
marker["_journal_retry_stream_id"] = stream_id
|
|
marker["_journal_retry_attempts"] = 0
|
|
marker["_journal_retry_first_seen_ts"] = int(time.time())
|
|
s.messages.append(marker)
|
|
s.save()
|
|
append_run_event(sid, stream_id, "token", {"text": "ConcurrentTokens"})
|
|
models.SESSIONS[sid] = s
|
|
|
|
results = []
|
|
|
|
def _worker():
|
|
try:
|
|
models.get_session(sid)
|
|
results.append("ok")
|
|
except Exception as exc: # noqa: BLE001
|
|
results.append(exc)
|
|
|
|
threads = [threading.Thread(target=_worker, daemon=True) for _ in range(2)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join(timeout=5)
|
|
|
|
assert results.count("ok") == 2, f"both workers must succeed: {results}"
|
|
|
|
# Exactly one interrupted marker remains, exactly one journal-recovered
|
|
# body (deduped by dedupe_existing=True).
|
|
interrupted_markers = [m for m in s.messages if m.get("type") == "interrupted"]
|
|
assert len(interrupted_markers) == 1
|
|
recovered = [m for m in s.messages if m.get("_recovered_from_run_journal")]
|
|
assert sum(1 for m in recovered if "ConcurrentTokens" in m.get("content", "")) == 1
|