From 663817570cecfe5fdefa422cd047caf00cae22d9 Mon Sep 17 00:00:00 2001 From: ai-ag2026 <261867348+ai-ag2026@users.noreply.github.com> Date: Mon, 11 May 2026 02:03:37 +0200 Subject: [PATCH 1/4] fix: recover orphaned session backups on startup --- api/routes.py | 1 + api/session_recovery.py | 92 ++++++++++++++++++++++++--- server.py | 7 +- tests/test_metadata_save_wipe_1558.py | 65 +++++++++++++++++++ tests/test_regressions.py | 13 ++++ 5 files changed, 167 insertions(+), 11 deletions(-) diff --git a/api/routes.py b/api/routes.py index 33567d0dbb3..015b37b6b1d 100644 --- a/api/routes.py +++ b/api/routes.py @@ -4147,6 +4147,7 @@ def handle_post(handler, parsed) -> bool: return bad(handler, "Invalid session_id", 400) 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) # Prune the per-session agent lock so deleted sessions don't leak diff --git a/api/session_recovery.py b/api/session_recovery.py index 9ae6d2543d6..86fc16ac1da 100644 --- a/api/session_recovery.py +++ b/api/session_recovery.py @@ -5,13 +5,16 @@ ``Session.save()`` writes a ``.json.bak`` snapshot of the previous state whenever an incoming save would shrink the messages array. This module reads those snapshots back and restores any session whose live -file has fewer messages than its backup. +file has fewer messages than its backup, or whose live file is missing +while a valid backup remains. Three integration points: 1. ``recover_all_sessions_on_startup()`` — called from server.py at boot, scans the session dir, restores any session whose JSON has fewer - messages than its .bak. Idempotent: a clean run is a no-op. + messages than its .bak, and recreates a missing ``.json`` from an + orphaned ``.json.bak`` when the canonical state DB still has that + session. Idempotent: a clean run is a no-op. 2. ``recover_session(sid)`` — single-session helper backing the ``POST /api/session/recover`` endpoint, so users can re-run recovery @@ -25,6 +28,7 @@ import json import logging import shutil +import sqlite3 from pathlib import Path logger = logging.getLogger(__name__) @@ -117,24 +121,81 @@ def recover_session(session_path: Path) -> dict: return {**status, "restored": True} -def recover_all_sessions_on_startup(session_dir: Path) -> dict: - """Scan session_dir for shrunken sessions, restore each from its .bak. +def _state_db_has_session(session_id: str, state_db_path: Path | None) -> bool: + """Return whether state.db still knows this session. - Returns {"scanned": N, "restored": M, "details": [...]}. + The check is deliberately fail-open: recovery must not be prevented by a + locked, absent, or older-schema state DB. When a DB is readable and has no + row, treat the orphan backup as a tombstoned/deleted session and skip it. + """ + if state_db_path is None or not state_db_path.exists(): + return True + try: + with sqlite3.connect(f"file:{state_db_path}?mode=ro", uri=True) as conn: + cur = conn.execute( + "select 1 from sqlite_master where type='table' and name='sessions'" + ) + if cur.fetchone() is None: + return True + cur = conn.execute("select 1 from sessions where id = ? limit 1", (session_id,)) + return cur.fetchone() is not None + except Exception as exc: + logger.debug("state_db session tombstone check failed for %s: %s", session_id, exc) + return True + + +def _orphaned_backup_live_paths( + session_dir: Path, + state_db_path: Path | None = None, +) -> list[Path]: + """Return live ``.json`` paths whose ``.json.bak`` exists. + + ``Path.glob('*.json')`` does not see orphan backups because their suffix is + ``.bak``. Existing startup recovery only handled shrunken live files; this + helper covers the crash shape where the live sidecar is gone but the rescue + copy remains. + """ + paths: list[Path] = [] + for bak_path in sorted(session_dir.glob('*.json.bak')): + live_path = bak_path.with_suffix('') + if live_path.name.startswith('_') or live_path.exists(): + continue + if _msg_count(bak_path) < 0: + continue + session_id = live_path.stem + if not _state_db_has_session(session_id, state_db_path): + logger.info( + "recover_all_sessions_on_startup: skipped orphan backup %s; " + "state.db has no live session row", + bak_path.name, + ) + continue + paths.append(live_path) + return paths + + +def recover_all_sessions_on_startup( + session_dir: Path, + rebuild_index: bool = False, + state_db_path: Path | None = None, +) -> dict: + """Scan session_dir for shrunken/orphaned sessions and restore from .bak. + + Returns {"scanned": N, "restored": M, "orphaned_backups": K, "details": [...]}. """ if not session_dir.exists(): - return {"scanned": 0, "restored": 0, "details": []} + return {"scanned": 0, "restored": 0, "orphaned_backups": 0, "details": []} scanned = 0 restored = 0 details: list[dict] = [] - for path in session_dir.glob('*.json'): + live_paths = [path for path in sorted(session_dir.glob('*.json')) if not path.name.startswith('_')] + orphan_paths = _orphaned_backup_live_paths(session_dir, state_db_path=state_db_path) + for path in [*live_paths, *orphan_paths]: # Skip non-session JSON files in the same dir: # - ``_index.json`` is a top-level list of session metadata # - any future non-session JSON marked with the ``_`` convention is # skipped automatically (project convention for system files in # directories that otherwise hold user data) - if path.name.startswith('_'): - continue scanned += 1 try: result = recover_session(path) @@ -155,4 +216,15 @@ def recover_all_sessions_on_startup(session_dir: Path) -> dict: "If you weren't expecting this, check the session list for missing " "messages — see #1558.", restored, scanned, ) - return {"scanned": scanned, "restored": restored, "details": details} + if rebuild_index: + try: + from api.models import _write_session_index + _write_session_index(updates=None) + except Exception as exc: + logger.warning("recover_all_sessions_on_startup: index rebuild failed: %s", exc) + return { + "scanned": scanned, + "restored": restored, + "orphaned_backups": len(orphan_paths), + "details": details, + } diff --git a/server.py b/server.py index 7e1563ac087..bbaf1cb88f6 100644 --- a/server.py +++ b/server.py @@ -220,8 +220,13 @@ def main() -> None: # its .bak (the data-loss shape #1558 produced), restore from the .bak. # Safe to run unconditionally — a clean install is a no-op. try: + from api.models import _active_state_db_path from api.session_recovery import recover_all_sessions_on_startup - result = recover_all_sessions_on_startup(SESSION_DIR) + result = recover_all_sessions_on_startup( + SESSION_DIR, + rebuild_index=True, + state_db_path=_active_state_db_path(), + ) if result.get("restored"): print(f"[recovery] Restored {result['restored']}/{result['scanned']} sessions from .bak (see #1558).", flush=True) except Exception as exc: diff --git a/tests/test_metadata_save_wipe_1558.py b/tests/test_metadata_save_wipe_1558.py index 3cb5153d21d..ce1b76cce2f 100644 --- a/tests/test_metadata_save_wipe_1558.py +++ b/tests/test_metadata_save_wipe_1558.py @@ -204,6 +204,71 @@ def test_recover_all_sessions_on_startup_restores_shrunken_session(temp_session_ assert len(restored["messages"]) == 1000 +def test_recover_all_sessions_on_startup_restores_orphan_bak(temp_session_dir): + """Startup self-heal: if only .json.bak survived, recreate .json.""" + sid = _make_session_on_disk(temp_session_dir, n_msgs=293) + live_path = temp_session_dir / f"{sid}.json" + bak_path = temp_session_dir / f"{sid}.json.bak" + bak_path.write_text(live_path.read_text(encoding="utf-8"), encoding="utf-8") + live_path.unlink() + + from api.session_recovery import recover_all_sessions_on_startup + result = recover_all_sessions_on_startup(temp_session_dir) + + assert result["restored"] == 1 + assert result["scanned"] == 1 + assert result.get("orphaned_backups") == 1 + restored = json.loads(live_path.read_text(encoding="utf-8")) + assert len(restored["messages"]) == 293 + + +def test_recover_all_sessions_on_startup_rebuilds_index_after_orphan_restore(temp_session_dir, monkeypatch): + """A restored orphan must be visible through the WebUI session index immediately.""" + import api.models as _m + + sid = _make_session_on_disk(temp_session_dir, n_msgs=42) + live_path = temp_session_dir / f"{sid}.json" + bak_path = temp_session_dir / f"{sid}.json.bak" + bak_path.write_text(live_path.read_text(encoding="utf-8"), encoding="utf-8") + live_path.unlink() + + stale_index = temp_session_dir / "_index.json" + stale_index.write_text(json.dumps([]), encoding="utf-8") + monkeypatch.setattr(_m, "SESSION_INDEX_FILE", stale_index) + + from api.session_recovery import recover_all_sessions_on_startup + result = recover_all_sessions_on_startup(temp_session_dir, rebuild_index=True) + + assert result["restored"] == 1 + index = json.loads(stale_index.read_text(encoding="utf-8")) + assert [entry["session_id"] for entry in index] == [sid] + assert index[0]["message_count"] == 42 + + +def test_orphan_bak_recovery_skips_sessions_absent_from_state_db(temp_session_dir): + """Do not resurrect an explicitly deleted session when state.db lacks the row.""" + import sqlite3 + + sid = _make_session_on_disk(temp_session_dir, n_msgs=12) + live_path = temp_session_dir / f"{sid}.json" + bak_path = temp_session_dir / f"{sid}.json.bak" + bak_path.write_text(live_path.read_text(encoding="utf-8"), encoding="utf-8") + live_path.unlink() + + state_db = temp_session_dir / "state.db" + with sqlite3.connect(state_db) as conn: + conn.execute("create table sessions (id text primary key)") + conn.execute("insert into sessions (id) values (?)", ("different_session",)) + + from api.session_recovery import recover_all_sessions_on_startup + result = recover_all_sessions_on_startup(temp_session_dir, state_db_path=state_db) + + assert result["restored"] == 0 + assert result["scanned"] == 0 + assert result["orphaned_backups"] == 0 + assert not live_path.exists() + + def test_recover_all_sessions_on_startup_is_idempotent_no_op_on_clean_state(temp_session_dir): """A clean install (no .bak files) must not modify anything.""" sid = _make_session_on_disk(temp_session_dir, n_msgs=1000) diff --git a/tests/test_regressions.py b/tests/test_regressions.py index 0022bcb947b..068afdcf2a6 100644 --- a/tests/test_regressions.py +++ b/tests/test_regressions.py @@ -335,6 +335,19 @@ def test_server_delete_invalidates_index(cleanup_test_sessions): return assert False, "session/delete handler not found in server.py or api/routes.py" + +def test_server_delete_removes_session_bak_snapshot(cleanup_test_sessions): + """session/delete must remove sidecar backups so deleted sessions stay deleted.""" + routes_src = (REPO_ROOT / "api" / "routes.py").read_text() + delete_idx = max( + routes_src.find("if parsed.path == '/api/session/delete':"), + routes_src.find('if parsed.path == "/api/session/delete":'), + ) + assert delete_idx >= 0, "session/delete handler not found in api/routes.py" + delete_block = routes_src[delete_idx:delete_idx+1400] + assert "with_suffix('.json.bak').unlink" in delete_block or 'with_suffix(".json.bak").unlink' in delete_block, \ + "session/delete must unlink .json.bak to avoid later orphan-backup recovery" + # ── R9: Token/tool SSE events write to wrong session after switch ───────────── def test_token_handler_guards_session_id(cleanup_test_sessions): From 7b6d91d490d71eea851f01fec297981805d9ffa7 Mon Sep 17 00:00:00 2001 From: ai-ag2026 <261867348+ai-ag2026@users.noreply.github.com> Date: Mon, 11 May 2026 02:06:43 +0200 Subject: [PATCH 2/4] feat: add read-only session recovery audit --- api/session_recovery.py | 132 +++++++++++++++++++++++++++ tests/test_session_recovery_audit.py | 100 ++++++++++++++++++++ 2 files changed, 232 insertions(+) create mode 100644 tests/test_session_recovery_audit.py diff --git a/api/session_recovery.py b/api/session_recovery.py index 86fc16ac1da..65e772a74ab 100644 --- a/api/session_recovery.py +++ b/api/session_recovery.py @@ -25,6 +25,7 @@ """ from __future__ import annotations +import argparse import json import logging import shutil @@ -174,6 +175,120 @@ def _orphaned_backup_live_paths( return paths +def _new_audit_item( + session_id: str, + kind: str, + category: str, + recommendation: str, + live_messages: int = -1, + bak_messages: int = -1, +) -> dict: + return { + "session_id": session_id, + "kind": kind, + "category": category, + "recommendation": recommendation, + "live_messages": live_messages, + "bak_messages": bak_messages, + } + + +def _read_index_session_ids(index_path: Path) -> set[str]: + try: + data = json.loads(index_path.read_text(encoding='utf-8')) + except (OSError, json.JSONDecodeError, ValueError): + return set() + if not isinstance(data, list): + return set() + ids: set[str] = set() + for entry in data: + if isinstance(entry, dict) and isinstance(entry.get('session_id'), str): + ids.add(entry['session_id']) + return ids + + +def audit_session_recovery(session_dir: Path, state_db_path: Path | None = None) -> dict: + """Read-only audit of session recovery state. + + The audit intentionally does not mutate files. It classifies only the safe + recovery primitives this module knows how to perform: backup restores and + derived index rebuilds. Call ``recover_all_sessions_on_startup`` separately + for safe repairs. + """ + if not session_dir.exists(): + return { + "status": "ok", + "summary": {"ok": 0, "repairable": 0, "unsafe_to_repair": 0}, + "items": [], + } + + 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} + + for live_path in live_paths: + status = inspect_session_recovery_status(live_path) + if status.get('recommend') == 'restore': + items.append(_new_audit_item( + status['session_id'], + "shrunken_live", + "repairable", + "restore_from_bak", + status.get('live_messages', -1), + status.get('bak_messages', -1), + )) + + for bak_path in sorted(session_dir.glob('*.json.bak')): + live_path = bak_path.with_suffix('') + if live_path.exists() or live_path.name.startswith('_'): + continue + bak_messages = _msg_count(bak_path) + session_id = live_path.stem + if bak_messages < 0: + items.append(_new_audit_item( + session_id, "malformed_orphan_backup", "unsafe_to_repair", "manual_review", -1, bak_messages + )) + elif _state_db_has_session(session_id, state_db_path): + items.append(_new_audit_item( + session_id, "orphan_backup", "repairable", "restore_from_bak", -1, bak_messages + )) + else: + items.append(_new_audit_item( + session_id, + "orphan_backup_without_state_row", + "unsafe_to_repair", + "manual_review", + -1, + bak_messages, + )) + + index_path = session_dir / '_index.json' + if index_path.exists(): + index_ids = _read_index_session_ids(index_path) + for session_id in sorted(index_ids - live_ids): + items.append(_new_audit_item( + session_id, "index_missing_file", "repairable", "rebuild_index" + )) + for session_id in sorted(live_ids - index_ids): + items.append(_new_audit_item( + session_id, "index_missing_entry", "repairable", "rebuild_index", + _msg_count(session_dir / f"{session_id}.json"), -1, + )) + + summary = {"ok": len(live_paths), "repairable": 0, "unsafe_to_repair": 0} + for item in items: + category = item.get('category') + if category in summary: + summary[category] += 1 + if summary["unsafe_to_repair"]: + overall = "needs_manual_review" + elif summary["repairable"]: + overall = "warn" + else: + overall = "ok" + return {"status": overall, "summary": summary, "items": items} + + def recover_all_sessions_on_startup( session_dir: Path, rebuild_index: bool = False, @@ -228,3 +343,20 @@ def recover_all_sessions_on_startup( "orphaned_backups": len(orphan_paths), "details": details, } + + +def _main() -> int: + parser = argparse.ArgumentParser(description="Audit Hermes WebUI session recovery state") + parser.add_argument("--audit", action="store_true", help="run a read-only recovery audit") + parser.add_argument("--session-dir", type=Path, required=True, help="path to WebUI sessions directory") + parser.add_argument("--state-db", type=Path, default=None, help="optional Hermes state.db path") + args = parser.parse_args() + if not args.audit: + parser.error("currently only --audit is supported") + report = audit_session_recovery(args.session_dir, state_db_path=args.state_db) + print(json.dumps(report, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(_main()) diff --git a/tests/test_session_recovery_audit.py b/tests/test_session_recovery_audit.py new file mode 100644 index 00000000000..dc9ad49b2a6 --- /dev/null +++ b/tests/test_session_recovery_audit.py @@ -0,0 +1,100 @@ +import json +import sqlite3 +import subprocess +import sys +from pathlib import Path + +from api.session_recovery import audit_session_recovery + +REPO_ROOT = Path(__file__).resolve().parents[1] + + +def _write_session(session_dir, sid, messages=1): + path = session_dir / f"{sid}.json" + path.write_text( + json.dumps({"id": sid, "session_id": sid, "title": sid, "messages": [{"role": "user", "content": str(i)} for i in range(messages)]}), + encoding="utf-8", + ) + return path + + +def _state_db(session_dir, *session_ids): + db = session_dir / "state.db" + with sqlite3.connect(db) as conn: + conn.execute("create table sessions (id text primary key)") + conn.executemany("insert into sessions (id) values (?)", [(sid,) for sid in session_ids]) + return db + + +def test_audit_reports_repairable_orphan_backup_when_state_db_has_session(tmp_path): + sid = "abc123" + live = _write_session(tmp_path, sid, messages=3) + bak = tmp_path / f"{sid}.json.bak" + bak.write_text(live.read_text(encoding="utf-8"), encoding="utf-8") + live.unlink() + db = _state_db(tmp_path, sid) + + report = audit_session_recovery(tmp_path, state_db_path=db) + + assert report["status"] == "warn" + assert report["summary"]["repairable"] == 1 + assert report["items"] == [ + { + "session_id": sid, + "kind": "orphan_backup", + "category": "repairable", + "recommendation": "restore_from_bak", + "live_messages": -1, + "bak_messages": 3, + } + ] + + +def test_audit_marks_orphan_backup_without_state_row_unsafe(tmp_path): + sid = "abc123" + live = _write_session(tmp_path, sid, messages=2) + bak = tmp_path / f"{sid}.json.bak" + bak.write_text(live.read_text(encoding="utf-8"), encoding="utf-8") + live.unlink() + db = _state_db(tmp_path, "different") + + report = audit_session_recovery(tmp_path, state_db_path=db) + + assert report["status"] == "needs_manual_review" + assert report["summary"]["unsafe_to_repair"] == 1 + assert report["items"][0]["kind"] == "orphan_backup_without_state_row" + assert report["items"][0]["recommendation"] == "manual_review" + + +def test_audit_reports_index_drift(tmp_path): + sid = "abc123" + _write_session(tmp_path, sid, messages=1) + (tmp_path / "_index.json").write_text( + json.dumps([{"session_id": "missing", "message_count": 1}]), + encoding="utf-8", + ) + + report = audit_session_recovery(tmp_path) + kinds = {item["kind"] for item in report["items"]} + + assert "index_missing_file" in kinds + assert "index_missing_entry" in kinds + assert report["summary"]["repairable"] == 2 + + +def test_session_recovery_module_audit_cli_outputs_json(tmp_path): + sid = "abc123" + _write_session(tmp_path, sid, messages=1) + + result = subprocess.run( + [sys.executable, "-m", "api.session_recovery", "--audit", "--session-dir", str(tmp_path)], + cwd=str(REPO_ROOT), + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=True, + ) + + payload = json.loads(result.stdout) + assert payload["status"] == "ok" + assert payload["summary"]["ok"] == 1 From 017e5af9a2f6a269319562811f65419b3993e4ae Mon Sep 17 00:00:00 2001 From: ai-ag2026 <261867348+ai-ag2026@users.noreply.github.com> Date: Mon, 11 May 2026 02:26:08 +0200 Subject: [PATCH 3/4] feat: expose session recovery audit and safe repair endpoints --- api/routes.py | 9 ++++ api/session_recovery.py | 35 ++++++++++++++-- tests/test_session_recovery_api.py | 67 ++++++++++++++++++++++++++++++ 3 files changed, 108 insertions(+), 3 deletions(-) create mode 100644 tests/test_session_recovery_api.py diff --git a/api/routes.py b/api/routes.py index 015b37b6b1d..52a0f3127cb 100644 --- a/api/routes.py +++ b/api/routes.py @@ -3217,6 +3217,10 @@ def handle_get(handler, parsed) -> bool: return bad(handler, "Session not found", 404) return j(handler, report) + if parsed.path == "/api/session/recovery/audit": + from api.session_recovery import audit_session_recovery + return j(handler, audit_session_recovery(SESSION_DIR, state_db_path=_active_state_db_path())) + if parsed.path == "/api/session/status": sid = parse_qs(parsed.query).get("session_id", [""])[0] if not sid: @@ -3770,6 +3774,11 @@ def handle_post(handler, parsed) -> bool: diag.finish() raise + if parsed.path == "/api/session/recovery/repair-safe": + from api.session_recovery import repair_safe_session_recovery + result = repair_safe_session_recovery(SESSION_DIR, state_db_path=_active_state_db_path()) + return j(handler, result, status=200 if result.get("ok") else 409) + if parsed.path.startswith("/api/kanban/"): from api.kanban_bridge import handle_kanban_post diff --git a/api/session_recovery.py b/api/session_recovery.py index 65e772a74ab..b42ab7f4be2 100644 --- a/api/session_recovery.py +++ b/api/session_recovery.py @@ -289,6 +289,31 @@ def audit_session_recovery(session_dir: Path, state_db_path: Path | None = None) return {"status": overall, "summary": summary, "items": items} +def repair_safe_session_recovery(session_dir: Path, state_db_path: Path | None = None) -> dict: + """Run safe, deterministic session recovery repairs. + + This mutates only repairable classes already handled by startup recovery: + shrunken live sidecars and orphan backups that are not tombstoned by a + readable state.db. Unsafe audit findings remain for manual review. + """ + before = audit_session_recovery(session_dir, state_db_path=state_db_path) + repair = recover_all_sessions_on_startup( + session_dir, + rebuild_index=True, + state_db_path=state_db_path, + ) + after = audit_session_recovery(session_dir, state_db_path=state_db_path) + unsafe_remaining = int((after.get("summary") or {}).get("unsafe_to_repair") or 0) + repairable_remaining = int((after.get("summary") or {}).get("repairable") or 0) + return { + "ok": unsafe_remaining == 0 and repairable_remaining == 0, + "repaired": int(repair.get("restored") or 0), + "before": before, + "repair": repair, + "after": after, + } + + def recover_all_sessions_on_startup( session_dir: Path, rebuild_index: bool = False, @@ -350,10 +375,14 @@ def _main() -> int: parser.add_argument("--audit", action="store_true", help="run a read-only recovery audit") parser.add_argument("--session-dir", type=Path, required=True, help="path to WebUI sessions directory") parser.add_argument("--state-db", type=Path, default=None, help="optional Hermes state.db path") + parser.add_argument("--repair-safe", action="store_true", help="run safe deterministic repairs after auditing") args = parser.parse_args() - if not args.audit: - parser.error("currently only --audit is supported") - report = audit_session_recovery(args.session_dir, state_db_path=args.state_db) + if args.repair_safe: + report = repair_safe_session_recovery(args.session_dir, state_db_path=args.state_db) + elif args.audit: + report = audit_session_recovery(args.session_dir, state_db_path=args.state_db) + else: + parser.error("choose --audit or --repair-safe") print(json.dumps(report, sort_keys=True)) return 0 diff --git a/tests/test_session_recovery_api.py b/tests/test_session_recovery_api.py new file mode 100644 index 00000000000..2638219f290 --- /dev/null +++ b/tests/test_session_recovery_api.py @@ -0,0 +1,67 @@ +import json + +from api.session_recovery import audit_session_recovery, repair_safe_session_recovery + + +def _write_session(session_dir, sid, messages=1): + path = session_dir / f"{sid}.json" + path.write_text( + json.dumps({"id": sid, "session_id": sid, "title": sid, "messages": [{"role": "user", "content": str(i)} for i in range(messages)]}), + encoding="utf-8", + ) + return path + + +def test_repair_safe_session_recovery_restores_backup_and_rebuilds_index(tmp_path, monkeypatch): + import api.models as _m + + sid = "abc123" + live = _write_session(tmp_path, sid, messages=4) + bak = tmp_path / f"{sid}.json.bak" + bak.write_text(live.read_text(encoding="utf-8"), encoding="utf-8") + live.unlink() + index = tmp_path / "_index.json" + index.write_text(json.dumps([]), encoding="utf-8") + monkeypatch.setattr(_m, "SESSION_DIR", tmp_path) + monkeypatch.setattr(_m, "SESSION_INDEX_FILE", index) + + result = repair_safe_session_recovery(tmp_path) + + assert result["ok"] is True + assert result["repaired"] == 1 + assert live.exists() + assert audit_session_recovery(tmp_path)["status"] == "ok" + idx = json.loads(index.read_text(encoding="utf-8")) + assert [entry["session_id"] for entry in idx] == [sid] + + +def test_repair_safe_session_recovery_leaves_unsafe_orphan_for_manual_review(tmp_path): + import sqlite3 + + sid = "abc123" + live = _write_session(tmp_path, sid, messages=1) + bak = tmp_path / f"{sid}.json.bak" + bak.write_text(live.read_text(encoding="utf-8"), encoding="utf-8") + live.unlink() + db = tmp_path / "state.db" + with sqlite3.connect(db) as conn: + conn.execute("create table sessions (id text primary key)") + conn.execute("insert into sessions (id) values (?)", ("other",)) + + result = repair_safe_session_recovery(tmp_path, state_db_path=db) + + assert result["ok"] is False + assert result["repaired"] == 0 + assert not live.exists() + assert result["after"]["status"] == "needs_manual_review" + + +def test_recovery_audit_routes_are_registered(): + from pathlib import Path + + src = Path("api/routes.py").read_text(encoding="utf-8") + + assert 'parsed.path == "/api/session/recovery/audit"' in src + assert 'parsed.path == "/api/session/recovery/repair-safe"' in src + assert "audit_session_recovery" in src + assert "repair_safe_session_recovery" in src From 1a3ad18b56ba3ccf3eb17afbf6d5394aefeca512 Mon Sep 17 00:00:00 2001 From: ai-ag2026 <261867348+ai-ag2026@users.noreply.github.com> Date: Mon, 11 May 2026 02:30:00 +0200 Subject: [PATCH 4/4] feat: reconcile missing WebUI sidecars from state db --- api/session_recovery.py | 169 +++++++++++++++++- .../test_session_db_sidecar_reconciliation.py | 69 +++++++ 2 files changed, 235 insertions(+), 3 deletions(-) create mode 100644 tests/test_session_db_sidecar_reconciliation.py diff --git a/api/session_recovery.py b/api/session_recovery.py index b42ab7f4be2..6347bdb7a30 100644 --- a/api/session_recovery.py +++ b/api/session_recovery.py @@ -175,6 +175,150 @@ def _orphaned_backup_live_paths( return paths +def _read_state_db_missing_sidecar_rows(session_dir: Path, state_db_path: Path | None) -> list[dict]: + """Return WebUI-origin state.db rows whose JSON sidecar is missing.""" + if state_db_path is None or not state_db_path.exists(): + return [] + try: + with sqlite3.connect(f"file:{state_db_path}?mode=ro", uri=True) as conn: + conn.row_factory = sqlite3.Row + session_cols = {row[1] for row in conn.execute("PRAGMA table_info(sessions)").fetchall()} + message_cols = {row[1] for row in conn.execute("PRAGMA table_info(messages)").fetchall()} + if not {'id', 'source'}.issubset(session_cols): + return [] + title_expr = _sql_optional_col('title', session_cols) + model_expr = _sql_optional_col('model', session_cols) + started_expr = _sql_optional_col('started_at', session_cols, '0') + parent_expr = _sql_optional_col('parent_session_id', session_cols) + msg_count_expr = _sql_optional_col('message_count', session_cols, '0') + rows = [] + for row in conn.execute( + f""" + SELECT id, source, {title_expr}, {model_expr}, {started_expr}, + {parent_expr}, {msg_count_expr} + FROM sessions + WHERE source = 'webui' + ORDER BY COALESCE(started_at, 0) DESC + """ + ).fetchall(): + data = dict(row) + sid = str(data.get('id') or '').strip() + if not sid or (session_dir / f"{sid}.json").exists(): + 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" + ts_expr = 'timestamp' if 'timestamp' in message_cols else 'NULL AS timestamp' + for msg in conn.execute( + f"SELECT role, content, {ts_expr} FROM messages WHERE session_id = ? ORDER BY {order}", + (sid,), + ).fetchall(): + message = { + 'role': msg['role'], + 'content': msg['content'] or '', + } + if msg['timestamp'] is not None: + message['timestamp'] = msg['timestamp'] + message_rows.append(message) + if not message_rows: + continue + data['messages'] = message_rows + rows.append(data) + return rows + except Exception as exc: + logger.debug("state_db sidecar reconciliation scan failed for %s: %s", state_db_path, exc) + return [] + + +def _sql_optional_col(name: str, columns: set[str], fallback: str = "NULL") -> str: + return name if name in columns else f"{fallback} AS {name}" + + +def _state_db_row_to_sidecar(row: dict) -> dict: + try: + from api.agent_sessions import normalize_agent_session_source + except Exception: + normalize_agent_session_source = None + source = str(row.get('source') or '').strip().lower() + source_meta = normalize_agent_session_source(source) if normalize_agent_session_source else { + 'raw_source': source or None, + 'session_source': source or None, + 'source_label': source.title() if source else None, + } + started_at = row.get('started_at') or 0 + messages = row.get('messages') if isinstance(row.get('messages'), list) else [] + last_ts = messages[-1].get('timestamp') if messages and isinstance(messages[-1], dict) else started_at + return { + 'session_id': row.get('id'), + 'title': row.get('title') or 'Recovered WebUI Session', + 'workspace': '', + 'model': row.get('model') or 'unknown', + 'model_provider': None, + 'created_at': started_at, + 'updated_at': last_ts or started_at, + 'pinned': False, + 'archived': False, + 'project_id': None, + 'profile': None, + 'input_tokens': 0, + 'output_tokens': 0, + 'estimated_cost': None, + 'personality': None, + 'active_stream_id': None, + 'pending_user_message': None, + 'pending_attachments': [], + 'pending_started_at': None, + 'compression_anchor_visible_idx': None, + 'compression_anchor_message_key': None, + 'compression_anchor_summary': None, + 'context_length': None, + 'threshold_tokens': None, + 'last_prompt_tokens': None, + 'gateway_routing': None, + 'gateway_routing_history': [], + 'llm_title_generated': False, + 'parent_session_id': row.get('parent_session_id'), + 'is_cli_session': False, + 'source_tag': source or None, + **source_meta, + 'enabled_toolsets': None, + 'composer_draft': {}, + 'messages': messages, + 'tool_calls': [], + '_recovered_from_state_db': True, + } + + +def recover_missing_sidecars_from_state_db(session_dir: Path, state_db_path: Path | None) -> dict: + """Materialize missing WebUI JSON sidecars from canonical state.db rows.""" + rows = _read_state_db_missing_sidecar_rows(session_dir, state_db_path) + materialized = 0 + details: list[dict] = [] + session_dir.mkdir(parents=True, exist_ok=True) + for row in rows: + sid = str(row.get('id') or '').strip() + if not sid: + continue + target = session_dir / f"{sid}.json" + if target.exists(): + continue + payload = _state_db_row_to_sidecar(row) + tmp = target.with_suffix('.json.reconcile.tmp') + try: + tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding='utf-8') + tmp.replace(target) + except OSError as exc: + try: + tmp.unlink(missing_ok=True) + except OSError: + pass + details.append({'session_id': sid, 'materialized': False, 'error': str(exc)}) + continue + materialized += 1 + details.append({'session_id': sid, 'materialized': True, 'messages': len(payload.get('messages') or [])}) + return {'scanned': len(rows), 'materialized': materialized, 'details': details} + + def _new_audit_item( session_id: str, kind: str, @@ -275,6 +419,17 @@ 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): + sid = str(row.get('id') or '') + items.append(_new_audit_item( + sid, + "state_db_missing_sidecar", + "repairable", + "materialize_from_state_db", + -1, + -1, + )) + summary = {"ok": len(live_paths), "repairable": 0, "unsafe_to_repair": 0} for item in items: category = item.get('category') @@ -297,19 +452,27 @@ def repair_safe_session_recovery(session_dir: Path, state_db_path: Path | None = readable state.db. Unsafe audit findings remain for manual review. """ before = audit_session_recovery(session_dir, state_db_path=state_db_path) - repair = recover_all_sessions_on_startup( + backup_repair = recover_all_sessions_on_startup( session_dir, rebuild_index=True, state_db_path=state_db_path, ) + sidecar_repair = recover_missing_sidecars_from_state_db(session_dir, state_db_path) + if sidecar_repair.get('materialized'): + try: + from api.models import _write_session_index + _write_session_index(updates=None) + except Exception as exc: + logger.warning("repair_safe_session_recovery: index rebuild after state.db reconciliation failed: %s", exc) after = audit_session_recovery(session_dir, state_db_path=state_db_path) unsafe_remaining = int((after.get("summary") or {}).get("unsafe_to_repair") or 0) repairable_remaining = int((after.get("summary") or {}).get("repairable") or 0) return { "ok": unsafe_remaining == 0 and repairable_remaining == 0, - "repaired": int(repair.get("restored") or 0), + "repaired": int(backup_repair.get("restored") or 0) + int(sidecar_repair.get("materialized") or 0), "before": before, - "repair": repair, + "backup_repair": backup_repair, + "sidecar_repair": sidecar_repair, "after": after, } diff --git a/tests/test_session_db_sidecar_reconciliation.py b/tests/test_session_db_sidecar_reconciliation.py new file mode 100644 index 00000000000..631bf227fde --- /dev/null +++ b/tests/test_session_db_sidecar_reconciliation.py @@ -0,0 +1,69 @@ +import json +import sqlite3 + +from api.session_recovery import recover_missing_sidecars_from_state_db, audit_session_recovery + + +def _make_state_db(path, *, sid="state_only_001", source="webui", messages=2): + conn = sqlite3.connect(path) + conn.execute( + "CREATE TABLE sessions (id TEXT PRIMARY KEY, source TEXT, title TEXT, model TEXT, started_at REAL, message_count INTEGER, parent_session_id TEXT)" + ) + conn.execute( + "CREATE TABLE messages (id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT, role TEXT, content TEXT, timestamp REAL)" + ) + conn.execute( + "INSERT INTO sessions (id, source, title, model, started_at, message_count, parent_session_id) VALUES (?, ?, ?, ?, ?, ?, ?)", + (sid, source, "Recovered from DB", "openai/gpt-5", 1234.0, messages, "parent-1"), + ) + for i in range(messages): + conn.execute( + "INSERT INTO messages (session_id, role, content, timestamp) VALUES (?, ?, ?, ?)", + (sid, "user" if i % 2 == 0 else "assistant", f"message {i + 1}", 1234.0 + i), + ) + conn.commit() + conn.close() + return sid + + +def test_recover_missing_sidecars_from_state_db_materializes_webui_row(tmp_path): + sid = _make_state_db(tmp_path / "state.db") + + 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["title"] == "Recovered from DB" + assert data["model"] == "openai/gpt-5" + assert data["parent_session_id"] == "parent-1" + assert data["source_tag"] == "webui" + assert data["session_source"] == "webui" + assert [m["content"] for m in data["messages"]] == ["message 1", "message 2"] + + +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" + existing.write_text(json.dumps({"session_id": sid, "messages": [{"role": "user", "content": "keep"}]}), encoding="utf-8") + + result = recover_missing_sidecars_from_state_db(tmp_path, tmp_path / "state.db") + + assert result["materialized"] == 0 + assert json.loads(existing.read_text(encoding="utf-8"))["messages"][0]["content"] == "keep" + + +def test_audit_reports_state_db_row_missing_sidecar(tmp_path): + sid = _make_state_db(tmp_path / "state.db") + + report = audit_session_recovery(tmp_path, state_db_path=tmp_path / "state.db") + + assert any( + item["session_id"] == sid + and item["kind"] == "state_db_missing_sidecar" + and item["category"] == "repairable" + and item["recommendation"] == "materialize_from_state_db" + for item in report["items"] + )