fix(#5532): prevent clear from replaying state.db history

This commit is contained in:
Rod Boev committed 2026-07-04 15:08:26 -04:00
1 parent 980e98dc7e
commit ea2dc31c09
3 files changed
+288 -11

No files matched your search

+30
View File
@@ -13694,8 +13694,17 @@ def handle_post(handler, parsed) -> bool:
return bad(handler, "Session not found", 404)
sid = body["session_id"]
with _get_session_agent_lock(sid):
had_sidecar_messages = bool(s.messages or [])
s.messages = []
s.tool_calls = []
s.context_messages = []
s.truncation_watermark = 0.0
s.truncation_boundary = 0.0
s.active_stream_id = None
s.pending_user_message = None
s.pending_attachments = []
s.pending_started_at = None
s.pending_user_source = None
# Reset the title via the rename helper so clearing a manually-named
# session also clears manual_title/llm_title_generated — otherwise the
# reused session keeps its manual-title protection and never auto-names
@@ -13703,6 +13712,27 @@ def handle_post(handler, parsed) -> bool:
from api.session_ops import apply_session_title_rename
apply_session_title_rename(s, "Untitled")
s.save()
persisted_clear = False
try:
persisted = json.loads(s.path.read_text(encoding="utf-8"))
persisted_clear = (
persisted.get("messages") == []
and persisted.get("context_messages") == []
and persisted.get("truncation_watermark") == 0.0
and persisted.get("truncation_boundary") == 0.0
and persisted.get("active_stream_id") is None
and persisted.get("pending_user_message") is None
and persisted.get("pending_attachments") == []
and persisted.get("pending_started_at") is None
and persisted.get("pending_user_source") is None
)
except (OSError, json.JSONDecodeError, ValueError):
logger.warning("session clear could not verify persisted empty state for %s", sid, exc_info=True)
if had_sidecar_messages and persisted_clear:
try:
s.path.with_suffix('.json.bak').unlink(missing_ok=True)
except OSError:
logger.warning("session clear could not remove stale backup for %s", sid, exc_info=True)
# Evict cached agent outside the per-session lock. Eviction may run a
# boundary memory commit for batch-extraction providers, and provider
# I/O must not hold the session mutation lock.
+11 -11
View File
@@ -39,7 +39,7 @@ against current source before any route is added.
route, handler, or related code is added in this PR.
- This RFC does **not** modify `GET /api/sessions/events` (the existing global
session-list invalidation stream routed at `api/routes.py:12345-12346` and
implemented by `_handle_session_events_stream()` at `api/routes.py:16177`).
implemented by `_handle_session_events_stream()` at `api/routes.py:16207`).
- This RFC does **not** replace or modify existing streams: `/api/chat/stream`,
`/api/approval/stream`, or `/api/clarify/stream`.
- This RFC does **not** introduce Android, iOS, or PWA client code.
@@ -54,7 +54,7 @@ against current source before any route is added.
`GET /api/sessions/events` is a **different endpoint** from the one this RFC
proposes. It is routed at `api/routes.py:12345-12346` and implemented by
`_handle_session_events_stream()` at `api/routes.py:16177`. It emits bare
`_handle_session_events_stream()` at `api/routes.py:16207`. It emits bare
`sessions_changed` events and keepalives for any change to the session list. It
is a global invalidation signal, not a per-session lifecycle stream. The proposed
`GET /api/sessions/{session_id}/events` is per-session and path-distinct.
@@ -73,19 +73,19 @@ Line ranges in this inventory were verified against WebUI `master` when this
RFC was written. Function, constant, and endpoint names are the stable anchors
if source layout moves later.
- `_parse_run_journal_event_id()` (`api/routes.py:15673-15686`) and
`_parse_run_journal_after_seq()` (`api/routes.py:15688-15701`) parse the replay
- `_parse_run_journal_event_id()` (`api/routes.py:15703-15715`) and
`_parse_run_journal_after_seq()` (`api/routes.py:15718-15730`) parse the replay
cursor from the `after_event_id` / `after_seq` **query params** (not the
`Last-Event-ID` header — that header is the *proposed* new-endpoint contract
below, §Reconnect).
- `_runner_event_id()` at `api/routes.py:15765-15772` constructs the event `id`
- `_runner_event_id()` at `api/routes.py:15795-15802` constructs the event `id`
field as `stream_id:seq`.
- SSE frames carry their `id:` via the `_sse_with_id()` helper, emitted on the
live `/api/chat/stream` path at `api/routes.py:15918`, on the runner-observe
path at `api/routes.py:15811`, and during journal replay at
`api/routes.py:15721` / `15734`.
live `/api/chat/stream` path at `api/routes.py:15948`, on the runner-observe
path at `api/routes.py:15841`, and during journal replay at
`api/routes.py:15751` / `15764`.
- `_replay_run_journal()` reads events by `(session_id, stream_id)` at
`api/routes.py:15703-15735`.
`api/routes.py:15733-15765`.
- `api/streaming.py:6265-6285` writes current live agent streams to
`STREAMS[stream_id]`.
- `api/streaming.py:6620-6634` appends SSE events to the run journal and carries
@@ -165,7 +165,7 @@ position.
**`event_id` is opaque to clients.** Its current source-compatible form is
`stream_id:seq`, as constructed by `_runner_event_id()` at
`api/routes.py:15765-15772`. Clients must treat it as an opaque string and must
`api/routes.py:15795-15802`. Clients must treat it as an opaque string and must
not parse or construct cursor values.
**`seq` is monotonic within a stream/run.** It is not a session-global counter
@@ -183,7 +183,7 @@ events. The live `STREAMS[stream_id]` queue (`api/streaming.py:6265-6285`) is
not a reliable replay source because it holds only recent in-memory state.
A future implementation must replay from the run journal via the existing
`_replay_run_journal()` path (`api/routes.py:15703-15735`) and fall back to the
`_replay_run_journal()` path (`api/routes.py:15733-15765`) and fall back to the
snapshot mechanism when journal entries are unavailable for a given cursor.
## Snapshot fallback
@@ -0,0 +1,247 @@
"""Regression coverage for #5532 clear-route state.db replay suppression."""
from __future__ import annotations
import json
from collections import OrderedDict
from io import BytesIO
from types import SimpleNamespace
import pytest
pytestmark = pytest.mark.requires_agent_modules
def _msg(role: str, content: str, ts: float, mid: str) -> dict:
return {"id": mid, "role": role, "content": content, "timestamp": ts}
def _install_isolated_session_env(monkeypatch, tmp_path):
import api.config as config
import api.models as models
import api.profiles as profiles
import api.routes as routes
monkeypatch.setattr(config, "STATE_DIR", tmp_path, raising=False)
session_dir = tmp_path / "sessions"
monkeypatch.setattr(config, "SESSION_DIR", session_dir, raising=False)
monkeypatch.setattr(config, "SESSION_INDEX_FILE", session_dir / "_index.json", raising=False)
monkeypatch.setattr(models, "SESSION_DIR", session_dir, raising=False)
monkeypatch.setattr(models, "SESSION_INDEX_FILE", session_dir / "_index.json", raising=False)
monkeypatch.setattr(models, "SESSIONS", OrderedDict(), raising=False)
monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: tmp_path, raising=False)
monkeypatch.setattr(models, "_active_state_db_path", lambda: tmp_path / "state.db", raising=False)
monkeypatch.setattr(routes, "_active_state_db_path", lambda: tmp_path / "state.db", raising=False)
monkeypatch.setattr(config, "_evict_session_agent", lambda _sid: None, raising=False)
session_dir.mkdir(parents=True, exist_ok=True)
return session_dir
def _post_clear(monkeypatch, sid: str):
import api.routes as routes
body = b'{"session_id":"%s"}' % sid.encode("utf-8")
monkeypatch.setattr(routes, "_check_csrf", lambda handler: True)
captured = {}
def fake_j(_handler, payload, status=200, extra_headers=None):
captured["payload"] = payload
captured["status"] = status
captured["extra_headers"] = extra_headers
monkeypatch.setattr(routes, "j", fake_j)
handler = SimpleNamespace(
headers={"Content-Length": str(len(body))},
rfile=BytesIO(body),
)
routes.handle_post(handler, SimpleNamespace(path="/api/session/clear"))
return captured
def test_session_clear_persists_empty_context_and_blocks_state_db_replay(monkeypatch, tmp_path):
import api.models as models
from api.models import Session, merge_session_messages_append_only
from api.session_recovery import inspect_session_recovery_status, recover_session
_install_isolated_session_env(monkeypatch, tmp_path)
sid = "issue5532_clear_replay"
session = Session(
session_id=sid,
title="Reconcile",
workspace=str(tmp_path),
model="test-model",
messages=[
_msg("user", "live prompt", 1.0, "u1"),
_msg("assistant", "live reply", 2.0, "a1"),
],
context_messages=[
_msg("user", "live prompt", 1.0, "cu1"),
_msg("assistant", "live reply", 2.0, "ca1"),
],
created_at=1000.0,
updated_at=1001.0,
)
session.tool_calls = [{"id": "call-1", "function": {"name": "terminal"}}]
session.truncation_watermark = 9.0
session.truncation_boundary = 8.0
session.active_stream_id = "stale-stream"
session.pending_user_message = "pending prompt"
session.pending_attachments = [{"name": "pending.txt"}]
session.pending_started_at = 1002.0
session.pending_user_source = "webui"
session.save(touch_updated_at=False)
session.path.with_suffix(".json.bak").write_text(
json.dumps(
{
"session_id": sid,
"messages": [_msg("user", "older backup prompt", 0.5, "bu1")],
"context_messages": [_msg("user", "older backup prompt", 0.5, "bcu1")],
}
),
encoding="utf-8",
)
captured = _post_clear(monkeypatch, sid)
assert captured["status"] == 200
assert captured["payload"]["ok"] is True
assert captured["payload"]["session"]["active_stream_id"] is None
assert captured["payload"]["session"]["pending_user_message"] is None
loaded = Session.load(sid)
assert loaded is not None
assert loaded.messages == []
assert loaded.context_messages == []
assert loaded.tool_calls == []
assert loaded.truncation_watermark == 0.0
assert loaded.truncation_boundary == 0.0
assert loaded.active_stream_id is None
assert loaded.pending_user_message is None
assert loaded.pending_attachments == []
assert loaded.pending_started_at is None
assert loaded.pending_user_source is None
assert loaded.title == "Untitled"
persisted = json.loads(loaded.path.read_text(encoding="utf-8"))
assert persisted["messages"] == []
assert persisted["context_messages"] == []
assert persisted["truncation_watermark"] == 0.0
assert persisted["truncation_boundary"] == 0.0
assert persisted["active_stream_id"] is None
assert persisted["pending_user_message"] is None
assert persisted["pending_attachments"] == []
assert persisted["pending_started_at"] is None
assert persisted["pending_user_source"] is None
state_db_messages = [
_msg("user", "state prompt", 100.0, "s-u1"),
_msg("assistant", "state reply", 101.0, "s-a1"),
]
merged = merge_session_messages_append_only(
loaded.messages,
state_db_messages,
truncation_watermark=loaded.truncation_watermark,
truncation_boundary=loaded.truncation_boundary,
)
assert merged == []
assert models.Session.load(sid).context_messages == []
assert not loaded.path.with_suffix(".json.bak").exists()
status = inspect_session_recovery_status(loaded.path)
assert status["recommend"] == "no_backup"
recovered = recover_session(loaded.path)
assert recovered["restored"] is False
assert Session.load(sid).messages == []
def test_empty_sidecar_without_watermark_still_recovers_state_db_rows():
from api.models import Session, merge_session_messages_append_only
session = Session(
session_id="issue5532_negative_space",
messages=[],
context_messages=[],
truncation_watermark=None,
truncation_boundary=None,
)
state_db_messages = [
_msg("user", "state prompt", 100.0, "s-u1"),
_msg("assistant", "state reply", 101.0, "s-a1"),
]
merged = merge_session_messages_append_only(
session.messages,
state_db_messages,
truncation_watermark=session.truncation_watermark,
truncation_boundary=session.truncation_boundary,
)
assert [m["content"] for m in merged] == ["state prompt", "state reply"]
def test_clear_sentinel_does_not_suppress_later_backup_recovery(tmp_path):
from api.session_recovery import inspect_session_recovery_status, recover_session
live_path = tmp_path / "post_clear_loss.json"
live = {
"session_id": "post_clear_loss",
"messages": [],
"context_messages": [],
"truncation_watermark": 0.0,
"truncation_boundary": 0.0,
}
backup = {
**live,
"messages": [_msg("user", "post-clear prompt", 10.0, "u10")],
"context_messages": [_msg("user", "post-clear prompt", 10.0, "cu10")],
}
live_path.write_text(json.dumps(live), encoding="utf-8")
live_path.with_suffix(".json.bak").write_text(json.dumps(backup), encoding="utf-8")
status = inspect_session_recovery_status(live_path)
assert status["recommend"] == "restore"
recovered = recover_session(live_path)
assert recovered["restored"] is True
restored = json.loads(live_path.read_text(encoding="utf-8"))
assert restored["messages"][0]["content"] == "post-clear prompt"
def test_clearing_empty_live_session_preserves_existing_recoverable_backup(monkeypatch, tmp_path):
from api.models import Session
from api.session_recovery import inspect_session_recovery_status, recover_session
_install_isolated_session_env(monkeypatch, tmp_path)
sid = "issue5532_empty_live_with_backup"
session = Session(
session_id=sid,
title="Already empty",
workspace=str(tmp_path),
model="test-model",
messages=[],
context_messages=[],
truncation_watermark=0.0,
truncation_boundary=0.0,
created_at=1000.0,
updated_at=1001.0,
)
session.save(touch_updated_at=False)
backup = {
"session_id": sid,
"messages": [_msg("user", "recoverable prompt", 10.0, "u10")],
"context_messages": [_msg("user", "recoverable prompt", 10.0, "cu10")],
"truncation_watermark": 0.0,
"truncation_boundary": 0.0,
}
session.path.with_suffix(".json.bak").write_text(json.dumps(backup), encoding="utf-8")
captured = _post_clear(monkeypatch, sid)
assert captured["status"] == 200
assert session.path.with_suffix(".json.bak").exists()
assert inspect_session_recovery_status(session.path)["recommend"] == "restore"
recovered = recover_session(session.path)
assert recovered["restored"] is True
assert Session.load(sid).messages[0]["content"] == "recoverable prompt"