fix(#5498): prevent deleted WebUI sessions from resurrecting from state.db

api/models.py

@@ -160,6 +160,7 @@ _SESSION_INDEX_REBUILD_THREAD_TARGET: tuple[Path, Path] | None = None
 # WebUI sidebar polling path is single-process) but must wrap the WHOLE
 # load-modify-write/unlink sequence in both helpers.
 _WEBUI_ZERO_MESSAGE_ORPHAN_TOMBSTONE_LOCK = threading.Lock()
+_WEBUI_DELETED_SESSION_TOMBSTONE_LOCK = threading.Lock()
 
 # Path-safety contract for session IDs.  Accept alphanumerics, underscore, and
 # hyphen so API/gateway-issued ids (``api-*``, ``reachy-voice-*``) round-trip
@@ -513,6 +514,8 @@ def prune_session_from_index(session_id: str) -> None:
 
 WEBUI_ZERO_MESSAGE_ORPHAN_TOMBSTONE_CAP = 500
 WEBUI_ZERO_MESSAGE_ORPHAN_TOMBSTONE_VERSION = 1
+WEBUI_DELETED_SESSION_TOMBSTONE_CAP = 1000
+WEBUI_DELETED_SESSION_TOMBSTONE_VERSION = 1
 
 
 def _webui_zero_message_orphan_tombstone_file() -> "Path":
@@ -676,6 +679,98 @@ def _clear_webui_zero_message_orphan_tombstone(sid: str) -> None:
             )
 
 
+def _webui_deleted_session_tombstone_file() -> "Path":
+    return SESSION_DIR / "_deleted_webui_sessions.json"
+
+
+def _load_webui_deleted_session_tombstone() -> frozenset[str]:
+    p = _webui_deleted_session_tombstone_file()
+    if not p.exists():
+        return frozenset()
+    try:
+        raw = json.loads(p.read_text(encoding='utf-8'))
+    except Exception:
+        logger.debug("Failed to load webui deleted-session tombstone", exc_info=True)
+        return frozenset()
+    if not isinstance(raw, dict):
+        return frozenset()
+    try:
+        if int(raw.get("version", 0)) != WEBUI_DELETED_SESSION_TOMBSTONE_VERSION:
+            return frozenset()
+    except (TypeError, ValueError):
+        return frozenset()
+    ids = raw.get("ids", [])
+    if not isinstance(ids, list):
+        return frozenset()
+    return frozenset(
+        str(sid).strip() for sid in ids if str(sid or "").strip()
+    )
+
+
+def _save_webui_deleted_session_tombstone(ids) -> None:
+    try:
+        sorted_ids = sorted(set(
+            str(sid).strip() for sid in (ids or []) if str(sid or "").strip()
+        ))
+    except TypeError:
+        return
+    if len(sorted_ids) > WEBUI_DELETED_SESSION_TOMBSTONE_CAP:
+        sorted_ids = sorted_ids[-WEBUI_DELETED_SESSION_TOMBSTONE_CAP:]
+    payload = {
+        "version": WEBUI_DELETED_SESSION_TOMBSTONE_VERSION,
+        "ids": sorted_ids,
+    }
+    p = _webui_deleted_session_tombstone_file()
+    _tmp = None
+    try:
+        SESSION_DIR.mkdir(parents=True, exist_ok=True)
+        _tmp = p.with_suffix(
+            f'.tmp.{os.getpid()}.{threading.current_thread().ident}'
+        )
+        with open(_tmp, 'w', encoding='utf-8') as f:
+            json.dump(payload, f, ensure_ascii=False, indent=2)
+            f.flush()
+            os.fsync(f.fileno())
+        os.replace(_tmp, p)
+    except Exception:
+        logger.debug("Failed to save webui deleted-session tombstone", exc_info=True)
+        if _tmp is not None:
+            try:
+                _tmp.unlink(missing_ok=True)
+            except Exception:
+                pass
+
+
+def _record_webui_deleted_session_tombstone(sid: str) -> None:
+    sid = str(sid or "").strip()
+    if not sid:
+        return
+    with _WEBUI_DELETED_SESSION_TOMBSTONE_LOCK:
+        current = set(_load_webui_deleted_session_tombstone())
+        if sid in current:
+            return
+        current.add(sid)
+        _save_webui_deleted_session_tombstone(current)
+
+
+def _clear_webui_deleted_session_tombstone(sid: str) -> None:
+    sid = str(sid or "").strip()
+    if not sid:
+        return
+    with _WEBUI_DELETED_SESSION_TOMBSTONE_LOCK:
+        current = set(_load_webui_deleted_session_tombstone())
+        if sid not in current:
+            return
+        current.discard(sid)
+        if current:
+            _save_webui_deleted_session_tombstone(current)
+            return
+        try:
+            _webui_deleted_session_tombstone_file().unlink(missing_ok=True)
+        except Exception:
+            logger.debug("Failed to remove empty webui deleted-session tombstone", exc_info=True)
+
+
 def _active_stream_ids():
     with STREAMS_LOCK:
         active_ids = set(STREAMS.keys())
@@ -1190,9 +1285,10 @@ class Session:
         if self.messages:
             try:
                 _clear_webui_zero_message_orphan_tombstone(self.session_id)
+                _clear_webui_deleted_session_tombstone(self.session_id)
             except Exception:
                 logger.debug(
-                    "Failed to clear webui zero-message orphan tombstone for %s",
+                    "Failed to clear webui tombstone for %s",
                     self.session_id,
                     exc_info=True,
                 )
@@ -3253,9 +3349,10 @@ def new_session(workspace=None, model=None, profile=None, model_provider=None, p
     # must never block new-session creation.
     try:
         _clear_webui_zero_message_orphan_tombstone(s.session_id)
+        _clear_webui_deleted_session_tombstone(s.session_id)
     except Exception:
         logger.debug(
-            "Failed to clear webui zero-message orphan tombstone for %s",
+            "Failed to clear webui tombstone for %s",
             s.session_id,
             exc_info=True,
         )
@@ -4741,9 +4838,10 @@ def import_cli_session(
     # an import.
     try:
         _clear_webui_zero_message_orphan_tombstone(s.session_id)
+        _clear_webui_deleted_session_tombstone(s.session_id)
     except Exception:
         logger.debug(
-            "Failed to clear webui zero-message orphan tombstone for %s",
+            "Failed to clear webui tombstone for %s",
             s.session_id,
             exc_info=True,
         )

api/routes.py

@@ -7146,6 +7146,13 @@ def _session_index_marks_was_webui(sid: str) -> bool:
     return False
 
 
+def _session_deleted_tombstone_marks_was_webui(sid: str) -> bool:
+    try:
+        return sid in _load_webui_deleted_session_tombstone()
+    except Exception:
+        return False
+
+
 def _state_db_session_source(sid: str) -> str:
     """Return the lowercased ``sessions.source`` for ``sid`` from state.db.
 
@@ -7380,7 +7387,13 @@ def _claim_or_synthesize_cli_session(sid: str, cli_meta: dict = None):
 
     if not is_safe_session_id(sid):
         return None, "invalid_sid"
-    if _session_index_marks_was_webui(sid) and not _is_subagent_child_session_id(sid):
+    if (
+        (
+            _session_index_marks_was_webui(sid)
+            or _session_deleted_tombstone_marks_was_webui(sid)
+        )
+        and not _is_subagent_child_session_id(sid)
+    ):
         # A delegated subagent child (source='subagent' in state.db) can be
         # registered in the WebUI index as a webui/fork/blank-source row (it
         # shares the parent's lineage) yet have no WebUI sidecar of its own.
@@ -8584,6 +8597,8 @@ from api.models import (
     _load_webui_zero_message_orphan_tombstone,
     _record_webui_zero_message_orphan_tombstone,
     _clear_webui_zero_message_orphan_tombstone,
+    _load_webui_deleted_session_tombstone,
+    _record_webui_deleted_session_tombstone,
     ensure_cron_project,
     _profile_has_user_projects,
     is_cron_session,
@@ -13593,15 +13608,25 @@ def handle_post(handler, parsed) -> bool:
             p.relative_to(SESSION_DIR.resolve())
         except Exception:
             return bad(handler, "Invalid session_id", 400)
+        sidecar_deleted = False
         try:
             p.unlink(missing_ok=True)
-            p.with_suffix('.json.bak').unlink(missing_ok=True)
         except Exception:
             logger.debug("Failed to unlink session file %s", p)
+        sidecar_deleted = not p.exists()
+        try:
+            p.with_suffix('.json.bak').unlink(missing_ok=True)
+        except Exception:
+            logger.debug("Failed to unlink session backup file %s", p.with_suffix('.json.bak'))
         try:
             prune_session_from_index(sid)
         except Exception:
             logger.debug("Failed to prune deleted session from index: %s", sid, exc_info=True)
+        if sidecar_deleted:
+            try:
+                _record_webui_deleted_session_tombstone(sid)
+            except Exception:
+                logger.debug("Failed to tombstone deleted WebUI session %s", sid, exc_info=True)
         try:
             from api.upload import _session_attachment_dir
 

api/session_recovery.py

@@ -383,6 +383,9 @@ def _read_state_db_missing_sidecar_rows(
                 sid = str(data.get('id') or '').strip()
                 if not sid or (session_dir / f"{sid}.json").exists():
                     continue
+                tombstoned = _marks_deleted_webui_session(session_dir, sid)
+                if tombstoned and not include_empty:
+                    continue
                 message_rows: list[dict] = []
                 if {'session_id', 'role', 'content'}.issubset(message_cols):
                     order = "timestamp, id" if 'timestamp' in message_cols and 'id' in message_cols else "rowid"
@@ -402,6 +405,8 @@ def _read_state_db_missing_sidecar_rows(
                     continue
                 data['messages'] = message_rows
                 data['_state_db_empty_messages'] = not message_rows
+                if tombstoned:
+                    data['_state_db_deleted_webui_tombstone'] = True
                 rows.append(data)
             return rows
     except Exception as exc:
@@ -484,6 +489,9 @@ def recover_missing_sidecars_from_state_db(session_dir: Path, state_db_path: Pat
         sid = str(row.get('id') or '').strip()
         if not sid:
             continue
+        if row.get('_state_db_deleted_webui_tombstone'):
+            details.append({'session_id': sid, 'materialized': False, 'skipped': 'deleted_webui_tombstone'})
+            continue
         target = session_dir / f"{sid}.json"
         if target.exists():
             continue
@@ -565,6 +573,75 @@ def _read_index_session_ids(index_path: Path) -> set[str]:
     return ids
 
 
+def _index_marks_deleted_webui_session(session_dir: Path, sid: str) -> bool:
+    """Return True when _index.json marks sid as a deleted WebUI session."""
+    if not sid or (session_dir / f"{sid}.json").exists():
+        return False
+    index_path = session_dir / '_index.json'
+    if not index_path.exists():
+        return False
+    try:
+        data = json.loads(index_path.read_text(encoding='utf-8'))
+    except (OSError, json.JSONDecodeError, ValueError):
+        return False
+    if not isinstance(data, list):
+        return False
+    for entry in data:
+        if not isinstance(entry, dict) or entry.get('session_id') != sid:
+            continue
+        srcs = [
+            str(entry.get('source_tag') or '').strip().lower(),
+            str(entry.get('raw_source') or '').strip().lower(),
+            str(entry.get('session_source') or '').strip().lower(),
+        ]
+        explicit = [src for src in srcs if src]
+        if any(src in ('webui', 'fork') for src in explicit):
+            return True
+        if explicit:
+            return False
+        is_cli = entry.get('is_cli_session') is True
+        is_read_only = bool(entry.get('read_only') or entry.get('is_read_only'))
+        return not (is_cli or is_read_only)
+    return False
+
+
+def _durable_tombstone_marks_deleted_webui_session(session_dir: Path, sid: str) -> bool:
+    """Return True when the durable WebUI delete tombstone contains sid."""
+    if not sid or (session_dir / f"{sid}.json").exists():
+        return False
+    try:
+        from api import models as _models
+
+        if Path(_models.SESSION_DIR).resolve() == session_dir.resolve():
+            return sid in _models._load_webui_deleted_session_tombstone()
+    except Exception:
+        pass
+    tombstone_path = session_dir / '_deleted_webui_sessions.json'
+    try:
+        raw = json.loads(tombstone_path.read_text(encoding='utf-8'))
+    except (OSError, json.JSONDecodeError, ValueError):
+        return False
+    if not isinstance(raw, dict):
+        return False
+    try:
+        version = int(raw.get('version', 0))
+    except (TypeError, ValueError):
+        return False
+    if version != 1:
+        return False
+    ids = raw.get('ids')
+    if not isinstance(ids, list):
+        return False
+    return sid in {str(value).strip() for value in ids if str(value or '').strip()}
+
+
+def _marks_deleted_webui_session(session_dir: Path, sid: str) -> bool:
+    return (
+        _index_marks_deleted_webui_session(session_dir, sid)
+        or _durable_tombstone_marks_deleted_webui_session(session_dir, sid)
+    )
+
+
 def audit_session_recovery(session_dir: Path, state_db_path: Path | None = None) -> dict:
     """Read-only audit of session recovery state.
 
@@ -583,6 +660,16 @@ def audit_session_recovery(session_dir: Path, state_db_path: Path | None = None)
     items: list[dict] = []
     live_paths = sorted(p for p in session_dir.glob('*.json') if not p.name.startswith('_'))
     live_ids = {p.stem for p in live_paths}
+    state_db_missing_rows = _read_state_db_missing_sidecar_rows(
+        session_dir,
+        state_db_path,
+        include_empty=True,
+    )
+    state_db_deleted_webui_ids = {
+        str(row.get('id') or '')
+        for row in state_db_missing_rows
+        if row.get('_state_db_deleted_webui_tombstone')
+    }
 
     for live_path in live_paths:
         status = inspect_session_recovery_status(live_path)
@@ -624,6 +711,8 @@ def audit_session_recovery(session_dir: Path, state_db_path: Path | None = None)
     if index_path.exists():
         index_ids = _read_index_session_ids(index_path)
         for session_id in sorted(index_ids - live_ids):
+            if session_id in state_db_deleted_webui_ids:
+                continue
             items.append(_new_audit_item(
                 session_id, "index_missing_file", "repairable", "rebuild_index"
             ))
@@ -633,8 +722,18 @@ def audit_session_recovery(session_dir: Path, state_db_path: Path | None = None)
                 _msg_count(session_dir / f"{session_id}.json"), -1,
             ))
 
-    for row in _read_state_db_missing_sidecar_rows(session_dir, state_db_path, include_empty=True):
+    for row in state_db_missing_rows:
         sid = str(row.get('id') or '')
+        if row.get('_state_db_deleted_webui_tombstone'):
+            items.append(_new_audit_item(
+                sid,
+                "state_db_deleted_webui_tombstone",
+                "unsafe_to_repair",
+                "deleted_session_skipped",
+                -1,
+                -1,
+            ))
+            continue
         if row.get('_state_db_empty_messages'):
             items.append(_new_audit_item(
                 sid,

tests/test_5307_subagent_child_transcript.py

@@ -162,10 +162,12 @@ def test_was_webui_gate_excludes_subagent_children():
     m = re.search(r"\n(?:def |class )", src[start + 1:])
     block = src[start:(start + 1 + m.start()) if m else len(src)]
     assert "_session_index_marks_was_webui(sid)" in block
-    assert re.search(
-        r"_session_index_marks_was_webui\(sid\)\s+and\s+not\s+_is_subagent_child_session_id\(sid\)",
-        block,
-    ), "was_webui and not subagent-child must gate the same 404 return (#5307)"
+    assert "_session_deleted_tombstone_marks_was_webui(sid)" in block
+    guard_start = block.index("if (")
+    guard = block[guard_start:block.index("return None, \"was_webui\"", guard_start)]
+    assert "_is_subagent_child_session_id(sid)" in guard, (
+        "all was_webui predicates must share the subagent-child exclusion (#5307)"
+    )
 
 
 def test_sessions_js_open_handlers_keep_isexternalsession_contract():

tests/test_chat_start_claim_cli_session.py

@@ -374,6 +374,30 @@ def test_helper_returns_was_webui_for_deleted_webui_session(
     assert reason == "was_webui"
 
 
+def test_helper_returns_was_webui_for_durable_deleted_tombstone_without_index(
+    routes_module, isolated_state_db
+):
+    """A full WebUI delete tombstone must keep the 404 self-heal contract even
+    after /api/session/delete prunes _index.json before state.db cleanup fails.
+    """
+    import api.models as _models
+
+    sid = "webui-deleted-db-survives"
+    _make_state_db(
+        isolated_state_db["db"],
+        sid,
+        message_count=2,
+        title="Deleted WebUI",
+        source="webui",
+    )
+    _models._record_webui_deleted_session_tombstone(sid)
+
+    sess, reason = routes_module._claim_or_synthesize_cli_session(sid)
+
+    assert sess is None
+    assert reason == "was_webui"
+
+
 def test_helper_keeps_cli_orphan_with_blank_source(
     routes_module, tmp_path, monkeypatch, isolated_state_db
 ):

tests/test_issue2057_worktree_lifecycle.py

@@ -1,4 +1,5 @@
 from types import SimpleNamespace
+from pathlib import Path
 
 import api.models as models
 import api.routes as routes
@@ -67,6 +68,41 @@ def test_delete_worktree_session_reports_retained_worktree_without_cleanup(tmp_p
     assert worktree.exists(), "session delete must not remove the git worktree directory"
 
 
+def test_delete_session_records_tombstone_when_state_db_delete_fails(tmp_path, monkeypatch):
+    session_dir = _isolate_session_store(tmp_path, monkeypatch)
+    sid = "dbfaildelete1"
+    session = Session(
+        session_id=sid,
+        title="Delete failure",
+        messages=[{"role": "user", "content": "keep deleted"}],
+    )
+    session.save()
+    (session_dir / f"{sid}.json.bak").write_text("backup", encoding="utf-8")
+    captured = _capture_post(monkeypatch, {"session_id": sid})
+    monkeypatch.setattr(routes, "_lookup_cli_session_metadata", lambda value: {})
+    monkeypatch.setattr(routes, "_is_messaging_session_id", lambda value: False)
+
+    def fail_delete(value):
+        raise RuntimeError("state.db locked")
+
+    real_unlink = Path.unlink
+
+    def fail_backup_unlink(path, *args, **kwargs):
+        if path.name == f"{sid}.json.bak":
+            raise PermissionError("backup locked")
+        return real_unlink(path, *args, **kwargs)
+
+    monkeypatch.setattr(models, "delete_cli_session", fail_delete)
+    monkeypatch.setattr(Path, "unlink", fail_backup_unlink)
+
+    assert routes.handle_post(object(), SimpleNamespace(path="/api/session/delete")) is True
+
+    assert captured["status"] == 200
+    assert captured["payload"]["ok"] is True
+    assert not (session_dir / f"{sid}.json").exists()
+    assert sid in models._load_webui_deleted_session_tombstone()
+
+
 def test_archive_worktree_session_reports_retained_worktree_without_cleanup(tmp_path, monkeypatch):
     _isolate_session_store(tmp_path, monkeypatch)
     session, worktree = _worktree_session(tmp_path, "wtarchive1")

tests/test_session_db_sidecar_reconciliation.py

@@ -26,6 +26,10 @@ def _make_state_db(path, *, sid="state_only_001", source="webui", messages=2):
     return sid
 
 
+def _write_index(path, entries):
+    (path / "_index.json").write_text(json.dumps(entries), encoding="utf-8")
+
+
 def test_recover_missing_sidecars_from_state_db_materializes_webui_row(tmp_path):
     sid = _make_state_db(tmp_path / "state.db")
 
@@ -44,6 +48,78 @@ def test_recover_missing_sidecars_from_state_db_materializes_webui_row(tmp_path)
     assert [m["content"] for m in data["messages"]] == ["message 1", "message 2"]
 
 
+def test_recover_missing_sidecars_from_state_db_skips_deleted_webui_tombstone(tmp_path):
+    sid = _make_state_db(tmp_path / "state.db", sid="deleted_webui_001")
+    _write_index(tmp_path, [
+        {
+            "session_id": sid,
+            "source_tag": "webui",
+            "raw_source": "webui",
+            "session_source": "webui",
+        }
+    ])
+
+    result = recover_missing_sidecars_from_state_db(tmp_path, tmp_path / "state.db")
+
+    assert result["materialized"] == 0
+    assert not (tmp_path / f"{sid}.json").exists()
+
+
+def test_audit_reports_deleted_webui_tombstone_is_unsafe(tmp_path):
+    sid = _make_state_db(tmp_path / "state.db", sid="deleted_webui_001")
+    _write_index(tmp_path, [
+        {
+            "session_id": sid,
+            "source_tag": "webui",
+            "raw_source": "webui",
+            "session_source": "webui",
+        }
+    ])
+
+    report = audit_session_recovery(tmp_path, state_db_path=tmp_path / "state.db")
+
+    assert any(
+        item["session_id"] == sid
+        and item["kind"] == "state_db_deleted_webui_tombstone"
+        and item["category"] == "unsafe_to_repair"
+        and item["recommendation"] == "deleted_session_skipped"
+        for item in report["items"]
+    )
+    assert not any(
+        item["session_id"] == sid and item["kind"] == "state_db_missing_sidecar"
+        for item in report["items"]
+    )
+    assert not any(
+        item["session_id"] == sid and item["category"] == "repairable"
+        for item in report["items"]
+    )
+
+
+def test_recover_missing_sidecars_skips_durable_delete_tombstone_without_index(tmp_path, monkeypatch):
+    import api.models as _m
+
+    sid = _make_state_db(tmp_path / "state.db", sid="durable_deleted_001")
+    monkeypatch.setattr(_m, "SESSION_DIR", tmp_path)
+    _m._record_webui_deleted_session_tombstone(sid)
+
+    result = recover_missing_sidecars_from_state_db(tmp_path, tmp_path / "state.db")
+    report = audit_session_recovery(tmp_path, state_db_path=tmp_path / "state.db")
+
+    assert result["materialized"] == 0
+    assert not (tmp_path / f"{sid}.json").exists()
+    assert any(
+        item["session_id"] == sid
+        and item["kind"] == "state_db_deleted_webui_tombstone"
+        and item["category"] == "unsafe_to_repair"
+        and item["recommendation"] == "deleted_session_skipped"
+        for item in report["items"]
+    )
+    assert not any(
+        item["session_id"] == sid and item["category"] == "repairable"
+        for item in report["items"]
+    )
+
+
 def test_recover_missing_sidecars_from_state_db_skips_existing_sidecar(tmp_path):
     sid = _make_state_db(tmp_path / "state.db")
     existing = tmp_path / f"{sid}.json"
@@ -92,6 +168,41 @@ def test_empty_state_db_webui_row_is_unsafe_not_materialized(tmp_path):
     assert not (tmp_path / f"{sid}.json").exists()
 
 
+def test_recover_missing_sidecars_from_state_db_ignores_subagent_row(tmp_path):
+    sid = _make_state_db(tmp_path / "state.db", sid="subagent_001", source="subagent")
+
+    result = recover_missing_sidecars_from_state_db(tmp_path, tmp_path / "state.db")
+
+    assert result["materialized"] == 0
+    assert not (tmp_path / f"{sid}.json").exists()
+
+    report = audit_session_recovery(tmp_path, state_db_path=tmp_path / "state.db")
+    assert not any(item["session_id"] == sid for item in report["items"])
+
+
+def test_recover_missing_sidecars_from_state_db_ignores_read_only_index_row(tmp_path):
+    sid = _make_state_db(tmp_path / "state.db", sid="read_only_index_001")
+    _write_index(tmp_path, [
+        {
+            "session_id": sid,
+            "source_tag": "",
+            "raw_source": "",
+            "session_source": "",
+            "read_only": True,
+        }
+    ])
+
+    result = recover_missing_sidecars_from_state_db(tmp_path, tmp_path / "state.db")
+
+    assert result["materialized"] == 1
+    sidecar = tmp_path / f"{sid}.json"
+    assert sidecar.exists()
+    data = json.loads(sidecar.read_text(encoding="utf-8"))
+    assert data["session_id"] == sid
+    assert data["source_tag"] == "webui"
+    assert data["session_source"] == "webui"
+
+
 def test_materialized_sidecar_round_trips_through_session_load(tmp_path, monkeypatch):
     """Schema parity guard: a materialized sidecar must be readable by Session.load
     and the resulting Session must have the same messages we put in state.db.