fix(#5570): skip clear sentinel backups during recovery

Closes the crash-window resurrection corner in the #5532 clear-conversation
data-loss fix. /api/session/clear writes a .json.bak before truncating, and
because the cleared sidecar's truncation_watermark is 0.0, startup recovery saw
bak_count > live_count and could restore the backup — resurrecting the cleared
transcript after a crash/restart in the clear window.

Fix: stamp a unique per-clear `clear_generation` UUID on the sidecar when
clearing a session that had messages (persisted via Session model). Recovery's
_session_records_clear_sentinel() returns no_action when the live sidecar carries
a clear_generation the backup doesn't share AND the live sidecar matches the
exact cleared-empty shape (messages/context empty, watermark/boundary 0.0,
pending fields cleared). Same-generation backups stay recoverable; unreadable or
partial matches FAIL OPEN (normal recovery), so a real crash-loss is never
suppressed.

Contributor stage; gate-pass. Co-authored-by: rodboev <rodboev@users.noreply.github.com>

api/models.py

@@ -1053,6 +1053,7 @@ class Session:
                  last_prompt_tokens=None,
                  truncation_watermark=None,
                  truncation_boundary=None,
+                 clear_generation=None,
                  gateway_routing=None, gateway_routing_history=None,
                  llm_title_generated: bool=False,
                  manual_title: bool=False,
@@ -1104,6 +1105,7 @@ class Session:
         self.last_prompt_tokens = last_prompt_tokens
         self.truncation_watermark = truncation_watermark
         self.truncation_boundary = truncation_boundary
+        self.clear_generation = clear_generation
         self.gateway_routing = gateway_routing if isinstance(gateway_routing, dict) else None
         self.gateway_routing_history = gateway_routing_history if isinstance(gateway_routing_history, list) else []
         self.llm_title_generated = bool(llm_title_generated)
@@ -1174,6 +1176,7 @@ class Session:
             'context_length', 'threshold_tokens', 'last_prompt_tokens',
             'truncation_watermark',
             'truncation_boundary',
+            'clear_generation',
             'gateway_routing', 'gateway_routing_history', 'llm_title_generated', 'manual_title',
             'parent_session_id',
             'worktree_path', 'worktree_branch', 'worktree_repo_root', 'worktree_created_at',

api/routes.py

@@ -13795,6 +13795,7 @@ def handle_post(handler, parsed) -> bool:
             s.pending_attachments = []
             s.pending_started_at = None
             s.pending_user_source = None
+            s.clear_generation = uuid.uuid4().hex if had_sidecar_messages else 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
@@ -13815,6 +13816,7 @@ def handle_post(handler, parsed) -> bool:
                     and persisted.get("pending_attachments") == []
                     and persisted.get("pending_started_at") is None
                     and persisted.get("pending_user_source") is None
+                    and persisted.get("clear_generation") == s.clear_generation
                 )
             except (OSError, json.JSONDecodeError, ValueError):
                 logger.warning("session clear could not verify persisted empty state for %s", sid, exc_info=True)

api/session_recovery.py

@@ -211,6 +211,42 @@ def _backup_predates_intentional_shrink(session_path: Path, bak_path: Path) -> b
     return bak_ctx_len > live_ctx_len
 
 
+def _session_records_clear_sentinel(session_path: Path, bak_path: Path) -> bool:
+    """Return True when the live sidecar records a provenanced clear sentinel.
+
+    The live sidecar must carry the explicit /api/session/clear marker, and
+    the backup must not carry the same marker. Same-generation backups stay
+    recoverable; unreadable or partial matches fail open.
+    """
+    try:
+        data = json.loads(session_path.read_text(encoding='utf-8'))
+        bak = json.loads(bak_path.read_text(encoding='utf-8'))
+    except (OSError, json.JSONDecodeError, ValueError):
+        return False
+    if not isinstance(data, dict) or not isinstance(bak, dict):
+        return False
+    clear_generation = data.get('clear_generation')
+    if not isinstance(clear_generation, str) or not clear_generation:
+        return False
+    if bak.get('clear_generation') == clear_generation:
+        return False
+    expected = {
+        'messages': [],
+        'context_messages': [],
+        'truncation_watermark': 0.0,
+        'truncation_boundary': 0.0,
+        'active_stream_id': None,
+        'pending_user_message': None,
+        'pending_attachments': [],
+        'pending_started_at': None,
+        'pending_user_source': None,
+    }
+    for key, value in expected.items():
+        if key not in data or data.get(key) != value:
+            return False
+    return True
+
+
 def inspect_session_recovery_status(session_path: Path) -> dict:
     """Return a status dict describing whether recovery is recommended.
 
@@ -232,6 +268,14 @@ def inspect_session_recovery_status(session_path: Path) -> dict:
         }
     bak_count = _msg_count(bak_path)
     if bak_count > live_count:
+        if _session_records_clear_sentinel(session_path, bak_path):
+            return {
+                "session_id": session_path.stem,
+                "live_messages": live_count,
+                "bak_messages": bak_count,
+                "recommend": "no_action",
+                "intentional_clear_truncate": True,
+            }
         if (
             _session_records_intentional_compress_shrink(session_path)
             and _backup_predates_intentional_shrink(session_path, bak_path)

tests/test_issue5532_session_clear_state_db_replay.py

@@ -123,6 +123,7 @@ def test_session_clear_persists_empty_context_and_blocks_state_db_replay(monkeyp
     assert loaded.pending_attachments == []
     assert loaded.pending_started_at is None
     assert loaded.pending_user_source is None
+    assert loaded.clear_generation
     assert loaded.title == "Untitled"
 
     persisted = json.loads(loaded.path.read_text(encoding="utf-8"))
@@ -135,6 +136,7 @@ def test_session_clear_persists_empty_context_and_blocks_state_db_replay(monkeyp
     assert persisted["pending_attachments"] == []
     assert persisted["pending_started_at"] is None
     assert persisted["pending_user_source"] is None
+    assert persisted["clear_generation"] == loaded.clear_generation
 
     state_db_messages = [
         _msg("user", "state prompt", 100.0, "s-u1"),
@@ -192,6 +194,12 @@ def test_clear_sentinel_does_not_suppress_later_backup_recovery(tmp_path):
         "context_messages": [],
         "truncation_watermark": 0.0,
         "truncation_boundary": 0.0,
+        "active_stream_id": None,
+        "pending_user_message": None,
+        "pending_attachments": [],
+        "pending_started_at": None,
+        "pending_user_source": None,
+        "clear_generation": "clear-post",
     }
     backup = {
         **live,
@@ -241,6 +249,7 @@ def test_clearing_empty_live_session_preserves_existing_recoverable_backup(monke
 
     assert captured["status"] == 200
     assert session.path.with_suffix(".json.bak").exists()
+    assert Session.load(sid).clear_generation is None
     assert inspect_session_recovery_status(session.path)["recommend"] == "restore"
     recovered = recover_session(session.path)
     assert recovered["restored"] is True

tests/test_issue5570_clear_backup_recovery.py

@@ -0,0 +1,365 @@
+from __future__ import annotations
+
+import json
+
+from api.session_recovery import (
+    inspect_session_recovery_status,
+    recover_all_sessions_on_startup,
+    recover_session,
+)
+
+
+def _msg(role: str, content: str, ts: float, mid: str) -> dict:
+    return {"id": mid, "role": role, "content": content, "timestamp": ts}
+
+
+def _write_json(path, payload):
+    path.write_text(json.dumps(payload), encoding="utf-8")
+
+
+def _clear_sentinel(sid: str) -> dict:
+    return {
+        "session_id": sid,
+        "messages": [],
+        "context_messages": [],
+        "truncation_watermark": 0.0,
+        "truncation_boundary": 0.0,
+        "active_stream_id": None,
+        "pending_user_message": None,
+        "pending_attachments": [],
+        "pending_started_at": None,
+        "pending_user_source": None,
+        "clear_generation": f"clear-{sid}",
+    }
+
+
+def _stale_pre_clear_backup(sid: str) -> dict:
+    return {
+        "session_id": sid,
+        "messages": [
+            _msg("user", "pre-clear prompt", 1.0, "u1"),
+            _msg("assistant", "pre-clear reply", 2.0, "a1"),
+        ],
+        "context_messages": [
+            _msg("user", "pre-clear prompt", 1.0, "cu1"),
+            _msg("assistant", "pre-clear reply", 2.0, "ca1"),
+        ],
+        "truncation_watermark": None,
+        "truncation_boundary": None,
+    }
+
+
+def _recoverable_post_clear_backup(sid: str) -> dict:
+    backup = _clear_sentinel(sid)
+    backup["messages"] = [
+        _msg("user", "post-clear prompt", 10.0, "u10"),
+        _msg("assistant", "post-clear reply", 11.0, "a11"),
+    ]
+    backup["context_messages"] = list(backup["messages"])
+    return backup
+
+
+def _normal_loss_fixture(sid: str) -> tuple[dict, dict]:
+    live = {
+        "session_id": sid,
+        "messages": [_msg("user", "live prompt", 1.0, "u1")],
+        "context_messages": [_msg("user", "live prompt", 1.0, "cu1")],
+    }
+    backup = {
+        "session_id": sid,
+        "messages": [
+            _msg("user", "live prompt", 1.0, "u1"),
+            _msg("assistant", "live reply", 2.0, "a1"),
+            _msg("user", "extra prompt", 3.0, "u2"),
+        ],
+        "context_messages": [
+            _msg("user", "live prompt", 1.0, "cu1"),
+            _msg("assistant", "live reply", 2.0, "ca1"),
+            _msg("user", "extra prompt", 3.0, "cu2"),
+        ],
+    }
+    return live, backup
+
+
+def test_stale_pre_clear_backup_inspect_decision(tmp_path):
+    sid = "issue5570_clear"
+    live_path = tmp_path / f"{sid}.json"
+    bak_path = live_path.with_suffix(".json.bak")
+
+    _write_json(live_path, _clear_sentinel(sid))
+    _write_json(bak_path, _stale_pre_clear_backup(sid))
+
+    status = inspect_session_recovery_status(live_path)
+
+    assert status["recommend"] == "no_action"
+    assert status["intentional_clear_truncate"] is True
+
+
+def test_clear_sentinel_backup_is_not_restored(tmp_path):
+    sid = "issue5570_clear_prompt_contract"
+    live_path = tmp_path / f"{sid}.json"
+    bak_path = live_path.with_suffix(".json.bak")
+
+    _write_json(live_path, _clear_sentinel(sid))
+    _write_json(bak_path, _stale_pre_clear_backup(sid))
+
+    status = inspect_session_recovery_status(live_path)
+    result = recover_session(live_path)
+    restored_live = json.loads(live_path.read_text(encoding="utf-8"))
+
+    assert status["recommend"] == "no_action"
+    assert status["intentional_clear_truncate"] is True
+    assert result["restored"] is False
+    assert restored_live["messages"] == []
+    assert restored_live["truncation_watermark"] == 0.0
+
+
+def test_stale_pre_clear_backup_manual_recovery_leaves_clear_sidecar(tmp_path):
+    sid = "issue5570_clear_manual"
+    live_path = tmp_path / f"{sid}.json"
+    bak_path = live_path.with_suffix(".json.bak")
+
+    _write_json(live_path, _clear_sentinel(sid))
+    _write_json(bak_path, _stale_pre_clear_backup(sid))
+
+    result = recover_session(live_path)
+
+    assert result["restored"] is False
+    assert json.loads(live_path.read_text(encoding="utf-8"))["messages"] == []
+    assert json.loads(live_path.read_text(encoding="utf-8"))["truncation_watermark"] == 0.0
+
+
+def test_stale_pre_clear_backup_startup_recovery_skips_restore(tmp_path):
+    sid = "issue5570_clear_startup"
+    live_path = tmp_path / f"{sid}.json"
+    bak_path = live_path.with_suffix(".json.bak")
+
+    _write_json(live_path, _clear_sentinel(sid))
+    _write_json(bak_path, _stale_pre_clear_backup(sid))
+
+    result = recover_all_sessions_on_startup(tmp_path)
+
+    assert result["restored"] == 0
+    assert json.loads(live_path.read_text(encoding="utf-8"))["messages"] == []
+
+
+def test_same_generation_clear_backup_still_restores(tmp_path):
+    sid = "issue5570_clear_marker_pair"
+    live_path = tmp_path / f"{sid}.json"
+    bak_path = live_path.with_suffix(".json.bak")
+
+    _write_json(live_path, _clear_sentinel(sid))
+    _write_json(bak_path, _recoverable_post_clear_backup(sid))
+
+    status = inspect_session_recovery_status(live_path)
+    result = recover_session(live_path)
+
+    assert status["recommend"] == "restore"
+    assert result["restored"] is True
+    assert json.loads(live_path.read_text(encoding="utf-8"))["messages"][0]["content"] == "post-clear prompt"
+
+
+def test_normal_larger_backup_recovery_still_restores(tmp_path):
+    sid = "issue5570_normal_loss"
+    live_path = tmp_path / f"{sid}.json"
+    bak_path = live_path.with_suffix(".json.bak")
+
+    live, backup = _normal_loss_fixture(sid)
+    _write_json(live_path, live)
+    _write_json(bak_path, backup)
+
+    status = inspect_session_recovery_status(live_path)
+    result = recover_session(live_path)
+
+    assert status["recommend"] == "restore"
+    assert result["restored"] is True
+    assert len(json.loads(live_path.read_text(encoding="utf-8"))["messages"]) == 3
+
+
+def test_empty_non_sentinel_live_session_still_recovers(tmp_path):
+    sid = "issue5570_empty_non_sentinel"
+    live_path = tmp_path / f"{sid}.json"
+    bak_path = live_path.with_suffix(".json.bak")
+
+    live = {
+        "session_id": sid,
+        "messages": [],
+        "context_messages": [],
+        "truncation_watermark": None,
+        "truncation_boundary": None,
+        "active_stream_id": None,
+        "pending_user_message": None,
+        "pending_attachments": [],
+        "pending_started_at": None,
+        "pending_user_source": None,
+    }
+    backup = {
+        "session_id": sid,
+        "messages": [
+            _msg("user", "replay prompt", 4.0, "u4"),
+            _msg("assistant", "replay reply", 5.0, "a5"),
+        ],
+        "context_messages": [
+            _msg("user", "replay prompt", 4.0, "cu4"),
+            _msg("assistant", "replay reply", 5.0, "ca5"),
+        ],
+    }
+    _write_json(live_path, live)
+    _write_json(bak_path, backup)
+
+    status = inspect_session_recovery_status(live_path)
+    result = recover_session(live_path)
+
+    assert status["recommend"] == "restore"
+    assert result["restored"] is True
+    assert json.loads(live_path.read_text(encoding="utf-8"))["messages"][0]["content"] == "replay prompt"
+
+
+def test_clear_shaped_live_without_clear_generation_still_restores(tmp_path):
+    sid = "issue5570_shape_without_marker"
+    live_path = tmp_path / f"{sid}.json"
+    bak_path = live_path.with_suffix(".json.bak")
+    live = _clear_sentinel(sid)
+    live.pop("clear_generation")
+    _write_json(live_path, live)
+    _write_json(bak_path, _stale_pre_clear_backup(sid))
+
+    status = inspect_session_recovery_status(live_path)
+    result = recover_session(live_path)
+
+    assert status["recommend"] == "restore"
+    assert result["restored"] is True
+    assert json.loads(live_path.read_text(encoding="utf-8"))["messages"][0]["content"] == "pre-clear prompt"
+
+
+def test_empty_live_without_clear_watermark_still_restores(tmp_path):
+    sid = "issue5570_empty_without_watermark"
+    live_path = tmp_path / f"{sid}.json"
+    bak_path = live_path.with_suffix(".json.bak")
+
+    _write_json(live_path, {
+        **_clear_sentinel(sid),
+        "truncation_watermark": None,
+        "truncation_boundary": None,
+    })
+    _write_json(bak_path, _stale_pre_clear_backup(sid))
+
+    result = recover_session(live_path)
+
+    assert result["restored"] is True
+    assert json.loads(live_path.read_text(encoding="utf-8"))["messages"][0]["content"] == "pre-clear prompt"
+
+
+def test_active_or_pending_clear_like_sidecar_still_restores(tmp_path):
+    for field, value in (
+        ("active_stream_id", "stream-1"),
+        ("pending_user_message", "pending prompt"),
+    ):
+        sid = f"issue5570_{field}"
+        live_path = tmp_path / f"{sid}.json"
+        bak_path = live_path.with_suffix(".json.bak")
+        live = _clear_sentinel(sid)
+        live[field] = value
+        _write_json(live_path, live)
+        _write_json(bak_path, _stale_pre_clear_backup(sid))
+
+        result = recover_session(live_path)
+
+        assert result["restored"] is True
+        restored = json.loads(live_path.read_text(encoding="utf-8"))
+        assert restored["messages"][0]["content"] == "pre-clear prompt"
+
+
+def test_incomplete_clear_like_sidecar_missing_pending_fields_still_restores(tmp_path):
+    sid = "issue5570_missing_pending_fields"
+    live_path = tmp_path / f"{sid}.json"
+    bak_path = live_path.with_suffix(".json.bak")
+    live = _clear_sentinel(sid)
+    live.pop("active_stream_id")
+    live.pop("pending_user_message")
+    _write_json(live_path, live)
+    _write_json(bak_path, _stale_pre_clear_backup(sid))
+
+    status = inspect_session_recovery_status(live_path)
+    result = recover_session(live_path)
+
+    assert status["recommend"] == "restore"
+    assert result["restored"] is True
+    assert json.loads(live_path.read_text(encoding="utf-8"))["messages"][0]["content"] == "pre-clear prompt"
+
+
+def test_malformed_live_sidecar_still_recovers_larger_backup(tmp_path):
+    sid = "issue5570_malformed_live"
+    live_path = tmp_path / f"{sid}.json"
+    bak_path = live_path.with_suffix(".json.bak")
+    live_path.write_text("{not json", encoding="utf-8")
+    _write_json(bak_path, _stale_pre_clear_backup(sid))
+
+    status = inspect_session_recovery_status(live_path)
+    result = recover_session(live_path)
+
+    assert status["recommend"] == "restore"
+    assert result["restored"] is True
+    assert json.loads(live_path.read_text(encoding="utf-8"))["messages"][0]["content"] == "pre-clear prompt"
+
+
+def test_manual_compression_recovery_behavior_is_preserved(tmp_path):
+    sid = "issue5570_compression"
+    live_path = tmp_path / f"{sid}.json"
+    bak_path = live_path.with_suffix(".json.bak")
+    marker = _msg("user", "[context compaction] summary of earlier turns", 1.0, "marker")
+
+    pre_compress_live = {
+        "session_id": sid,
+        "messages": [
+            _msg("user", "one", 1.0, "u1"),
+            _msg("assistant", "two", 2.0, "a2"),
+            _msg("user", "three", 3.0, "u3"),
+            _msg("assistant", "four", 4.0, "a4"),
+        ],
+        "context_messages": [
+            _msg("user", "one", 1.0, "cu1"),
+            _msg("assistant", "two", 2.0, "ca2"),
+        ],
+        "compression_anchor_summary": "Compressed: 4 -> 2 messages",
+        "compression_anchor_message_key": {"role": "assistant", "ts": 2.0, "text": "two", "attachments": 0},
+        "compression_anchor_mode": "manual",
+        "truncation_watermark": 2.0,
+        "truncation_boundary": 2.0,
+    }
+    pre_compress_backup = dict(pre_compress_live)
+    pre_compress_backup["messages"] = pre_compress_live["messages"] + [
+        _msg("user", "five", 5.0, "u5"),
+        _msg("assistant", "six", 6.0, "a6"),
+    ]
+    pre_compress_backup["context_messages"] = pre_compress_backup["messages"]
+    _write_json(live_path, pre_compress_live)
+    _write_json(bak_path, pre_compress_backup)
+
+    pre_status = inspect_session_recovery_status(live_path)
+    pre_result = recover_session(live_path)
+
+    assert pre_status["recommend"] == "no_action"
+    assert pre_status["intentional_compress_shrink"] is True
+    assert pre_result["restored"] is False
+
+    post_backup = dict(pre_compress_live)
+    post_backup["messages"] = [
+        _msg("user", "one", 1.0, "u1"),
+        _msg("assistant", "two", 2.0, "a2"),
+        _msg("user", "three", 3.0, "u3"),
+        _msg("assistant", "four", 4.0, "a4"),
+    ]
+    post_backup["context_messages"] = [marker]
+    _write_json(live_path, {
+        **pre_compress_live,
+        "messages": [_msg("user", "one", 1.0, "u1")],
+        "context_messages": [_msg("user", "one", 1.0, "cu1")],
+    })
+    _write_json(bak_path, post_backup)
+
+    post_status = inspect_session_recovery_status(live_path)
+    post_result = recover_session(live_path)
+
+    assert post_status["recommend"] == "restore"
+    assert post_result["restored"] is True