diff --git a/api/routes.py b/api/routes.py index 35393dc4..0b5de609 100644 --- a/api/routes.py +++ b/api/routes.py @@ -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. diff --git a/docs/rfcs/session-sse-contract-v1.md b/docs/rfcs/session-sse-contract-v1.md index 0c4aa5f1..c0e0dd16 100644 --- a/docs/rfcs/session-sse-contract-v1.md +++ b/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 diff --git a/tests/test_issue5532_session_clear_state_db_replay.py b/tests/test_issue5532_session_clear_state_db_replay.py new file mode 100644 index 00000000..ed90316e --- /dev/null +++ b/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"