Merge pull request #5610 from nesquena/release/stage-5584

release #5584: skip clear-sentinel backups during recovery

CHANGELOG.md

@@ -14,6 +14,7 @@
 
 ### Fixed
 
+- **Clearing a conversation can no longer be undone by crash recovery — even after you send a new message.** `Clear conversation` writes a one-time `.json.bak` before wiping the transcript; if the app then crashed or restarted, startup recovery saw the backup as "larger" than the cleared session and could restore the cleared history — including on top of a message you'd sent after clearing. Clear now stamps a unique per-clear marker on the session, and recovery treats a backup that predates that marker as stale, so a cleared conversation stays cleared. Genuine crash-loss recovery (unmarked or same-generation backups, unreadable/partial files) still restores as before. Hardens the clear-conversation data-loss fix from the prior releases. Thanks @rodboev. (#5584, #5570)
 - **A message containing a broken emoji (lone Unicode surrogate) no longer breaks scroll position tracking.** If a message's text held a lone UTF-16 surrogate — e.g. a half-copied emoji from a truncated paste or interrupted stream — computing that message's scroll-anchor key threw an error, which broke scroll-position restore for the whole transcript. The anchor-key builder now handles lone surrogates safely (valid emoji are preserved; only the broken half is dropped), so scroll anchoring keeps working. Thanks @rumotoshino. (#5573, #5552)
 - **An in-progress CLI session now shows in the sidebar before its final output lands.** A CLI/agent session that had real activity (messages and user turns) but hadn't produced its final output yet was hidden from the sidebar until it completed, so an actively-running CLI session could be invisible while you waited on it. Such a session (real message count + user turns, not yet ended, no end reason) is now surfaced while it's still running. Empty/never-really-started rows stay hidden as before. Thanks @rodboev. (#5593, #5587)
 - **A completed answer is no longer followed by a misleading "No response from provider" card.** On the streaming settlement path, a stale terminal/partial-result flag could force the generic `no_response` error even when the current turn had already produced a full assistant answer — so you'd see the reply and then an error card under it. The generic silent-error path is now suppressed only when the merged current turn actually has a completed assistant answer; genuine failures (real silent failures, partial failures, auth/quota errors, cancellation, replayed rows, terminal tool/compression cases) all keep their existing error behavior. Thanks @rodboev. (#5592, #5575)

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,78 @@ 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 _live_supersedes_backup_by_clear_generation(session_path: Path, bak_path: Path) -> bool:
+    """Return True when the live sidecar provably post-dates the backup via a
+    clear sentinel even though the user has since sent NEW messages.
+
+    Scope: this handles ONLY the post-clear-message case that the exact-empty
+    sentinel (_session_records_clear_sentinel) can't. After /api/session/clear
+    stamps a unique ``clear_generation`` and resets the truncation boundary to
+    0.0, a pre-clear ``.json.bak`` is stale — restoring it would resurrect
+    cleared history on top of the post-clear message. We require: live carries a
+    ``clear_generation`` the backup lacks, live has a NON-EMPTY transcript, and
+    the live boundary still shows the clear reset (watermark == boundary == 0.0).
+    Empty clear-shaped sidecars stay governed by the exact-empty sentinel and its
+    existing recovery semantics. Same-generation backups stay recoverable;
+    unreadable/partial reads fail open (return False -> normal recovery), so a
+    genuine crash-loss is never suppressed.
+    """
+    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
+    live_messages = data.get('messages')
+    if not isinstance(live_messages, list) or len(live_messages) == 0:
+        return False
+    if data.get('truncation_watermark') != 0.0 or data.get('truncation_boundary') != 0.0:
+        return False
+    return True
+
+
 def inspect_session_recovery_status(session_path: Path) -> dict:
     """Return a status dict describing whether recovery is recommended.
 
@@ -232,6 +304,17 @@ 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)
+            or _live_supersedes_backup_by_clear_generation(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,418 @@
+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
+
+
+def test_post_clear_message_with_stale_pre_clear_backup_is_not_restored(tmp_path):
+    """#5584 gate (Codex CORE): after a clear, the user sends one post-clear
+    message and the stale pre-clear .json.bak still exists (unlink failed). The
+    backup has MORE messages than the live sidecar, but the live sidecar carries
+    a clear_generation the backup lacks + the clear boundary reset (watermarks
+    0.0) — so recovery must NOT resurrect the pre-clear transcript on top of the
+    new message."""
+    sid = "s-postclear"
+    live_path = tmp_path / f"{sid}.json"
+    bak_path = tmp_path / f"{sid}.json.bak"
+    # Live: cleared sentinel shape + one post-clear message (boundary still 0.0,
+    # clear_generation intact — the clear handler's reset).
+    live = _clear_sentinel(sid)
+    live["messages"] = [_msg("user", "brand new post-clear question", 20.0, "u20")]
+    live["context_messages"] = [_msg("user", "brand new post-clear question", 20.0, "cu20")]
+    _write_json(live_path, live)
+    # Stale pre-clear backup: 2 messages, no clear_generation (predates the clear).
+    _write_json(bak_path, _stale_pre_clear_backup(sid))
+
+    status = inspect_session_recovery_status(live_path)
+    result = recover_session(live_path)
+
+    assert status["bak_messages"] > status["live_messages"]  # backup is "larger"
+    assert status["recommend"] == "no_action"
+    assert status.get("intentional_clear_truncate") is True
+    assert result["restored"] is False
+    # Live sidecar must still be the post-clear content, not the resurrected pre-clear.
+    reread = json.loads(live_path.read_text(encoding="utf-8"))
+    assert [m["content"] for m in reread["messages"]] == ["brand new post-clear question"]
+
+
+def test_post_clear_message_after_real_compaction_still_recovers(tmp_path):
+    """Guard the guard: if the live sidecar's boundary has moved OFF the clear
+    reset (a real compaction happened after the clear), the clear-generation
+    supersede check must decline so a genuine larger backup can still restore."""
+    sid = "s-postclear-compact"
+    live_path = tmp_path / f"{sid}.json"
+    bak_path = tmp_path / f"{sid}.json.bak"
+    live = _clear_sentinel(sid)
+    live["messages"] = [_msg("user", "q", 20.0, "u20")]
+    live["context_messages"] = [_msg("user", "q", 20.0, "cu20")]
+    # A later compaction moved the boundary off 0.0 — no longer the clear reset.
+    live["truncation_watermark"] = 30.0
+    live["truncation_boundary"] = 30.0
+    _write_json(live_path, live)
+    _write_json(bak_path, _stale_pre_clear_backup(sid))
+
+    status = inspect_session_recovery_status(live_path)
+    # Boundary moved off the clear reset -> supersede check declines -> normal
+    # recovery decides (larger backup, no compress-shrink provenance -> restore).
+    assert status["recommend"] == "restore"