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

api/routes.py

@@ -13694,15 +13694,45 @@ 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
             # again (#3542 lifecycle gap).
             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.

docs/rfcs/session-sse-contract-v1.md

@@ -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

tests/test_issue5532_session_clear_state_db_replay.py

@@ -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"