refactor(hermes_cli): compact session_recovery docstrings/comments and result literals
This commit is contained in:
@@ -1,10 +1,8 @@
|
||||
"""Offline, non-destructive recovery for a damaged Hermes session database.
|
||||
|
||||
* the supplied source database is never opened by SQLite; * the source file and any
|
||||
WAL/SHM/rollback-journal sidecars are copied into a disposable working directory first; * canonical
|
||||
rows are copied into a newly initialized current-schema database; * derived FTS tables and migration
|
||||
bookkeeping are rebuilt, not copied; and * the recovered database is never installed over the active
|
||||
database.
|
||||
The source is never opened by SQLite: it and its WAL/SHM/journal sidecars are copied to a disposable
|
||||
work dir first. Canonical rows are copied into a fresh current-schema database; derived FTS tables and
|
||||
migration bookkeeping are rebuilt, not copied; the result is never installed over the active database.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -24,12 +22,7 @@ from hermes_state import (FTS_STORAGE_VERSION, SCHEMA_VERSION, SessionDB, _db_op
|
||||
ProgressCallback = Callable[[dict[str, Any]], None]
|
||||
|
||||
_CANONICAL_TABLES = (
|
||||
"system_prompts",
|
||||
"sessions",
|
||||
"messages",
|
||||
"session_model_usage",
|
||||
"compression_locks",
|
||||
"gateway_routing",
|
||||
"system_prompts", "sessions", "messages", "session_model_usage", "compression_locks", "gateway_routing",
|
||||
"async_delegations",
|
||||
)
|
||||
|
||||
@@ -42,13 +35,9 @@ def _init_delivery_ledger_schema(conn: sqlite3.Connection) -> None:
|
||||
_initialize_schema(conn)
|
||||
|
||||
|
||||
# Tables that live in state.db but are created lazily by a gateway module on
|
||||
# first use, so base ``SessionDB`` never creates them on a fresh destination.
|
||||
# Every entry maps the table to the initializer that owns its DDL; recovery
|
||||
# creates the table on the destination before copying, so owed rows survive
|
||||
# instead of silently vanishing from a "complete" salvage (#100313, #86236).
|
||||
# Add new lazily-created state.db tables HERE, never as one-off ``if table ==``
|
||||
# branches.
|
||||
# state.db tables created lazily by a gateway module (base ``SessionDB`` never creates them on a fresh
|
||||
# destination) -> the initializer owning their DDL. Recovery creates them before copying so owed rows
|
||||
# don't silently vanish from a "complete" salvage. Register new lazy tables HERE, not as ``if table ==``.
|
||||
_AUXILIARY_TABLE_SCHEMAS: dict[str, Callable[[sqlite3.Connection], None]] = {
|
||||
"delivery_obligations": _init_delivery_ledger_schema,
|
||||
}
|
||||
@@ -57,17 +46,10 @@ _AUXILIARY_TABLES = tuple(_AUXILIARY_TABLE_SCHEMAS)
|
||||
|
||||
_INVENTORY_TABLES = (*_CANONICAL_TABLES, "state_meta", *_TOPIC_TABLES, *_AUXILIARY_TABLES)
|
||||
|
||||
# These values describe derived indexes or the schema that owns an optional
|
||||
# table. A fresh destination must generate them from its own current schema.
|
||||
# Derived-index / optional-schema markers: a fresh destination regenerates these, never copies them.
|
||||
_GENERATED_META_KEYS = frozenset({
|
||||
"fts_storage_version",
|
||||
"fts_optimize_available",
|
||||
"fts_rebuild_high_water",
|
||||
"fts_rebuild_progress",
|
||||
"fts_cjk_stale",
|
||||
"fts_cjk_rebuild_high_water",
|
||||
"fts_cjk_rebuild_progress",
|
||||
"telegram_dm_topic_schema_version",
|
||||
"fts_storage_version", "fts_optimize_available", "fts_rebuild_high_water", "fts_rebuild_progress",
|
||||
"fts_cjk_stale", "fts_cjk_rebuild_high_water", "fts_cjk_rebuild_progress", "telegram_dm_topic_schema_version",
|
||||
})
|
||||
|
||||
_SIDECAR_SUFFIXES = ("", "-wal", "-shm", "-journal")
|
||||
@@ -95,9 +77,7 @@ def _sidecar_path(db_path: Path, suffix: str) -> Path:
|
||||
|
||||
def _resolved_output_path(path: Path) -> Path:
|
||||
"""Resolve a not-yet-created output path without requiring it to exist."""
|
||||
|
||||
parent = path.expanduser().parent.resolve(strict=True)
|
||||
return parent / path.name
|
||||
return path.expanduser().parent.resolve(strict=True) / path.name
|
||||
|
||||
|
||||
def _validate_paths(
|
||||
@@ -157,33 +137,26 @@ def _format_bytes(value: int) -> str:
|
||||
def _same_filesystem(left: Path, right: Path) -> bool:
|
||||
try:
|
||||
return os.stat(left).st_dev == os.stat(right).st_dev
|
||||
except OSError:
|
||||
# Existing directories were already required by _validate_paths. This
|
||||
# fallback is defensive for platforms with incomplete st_dev support.
|
||||
except OSError: # both dirs exist (_validate_paths); fallback for incomplete st_dev support
|
||||
return left.anchor.casefold() == right.anchor.casefold()
|
||||
|
||||
|
||||
def _disk_space_preflight(source: Path, work_root: Path, output_parent: Optional[Path]) -> dict[str, Any]:
|
||||
"""Require space for the disposable bundle, output, and safety headroom."""
|
||||
|
||||
bundle_bytes = sum(
|
||||
_sidecar_path(source, suffix).stat().st_size
|
||||
for suffix in _SIDECAR_SUFFIXES
|
||||
if _sidecar_path(source, suffix).exists()
|
||||
)
|
||||
# The v23 external-content rebuild is normally substantially smaller than
|
||||
# a legacy database, but using the complete source bundle as the estimate
|
||||
# avoids betting the user's disk on that expectation.
|
||||
# The v23 external-content rebuild is usually much smaller than a legacy database, but estimating
|
||||
# with the complete bundle avoids betting the user's disk on that.
|
||||
output_allowance = bundle_bytes if output_parent is not None else 0
|
||||
headroom = max(_MINIMUM_SPACE_HEADROOM, int((bundle_bytes + output_allowance) * 0.05))
|
||||
|
||||
work_free = int(shutil.disk_usage(work_root).free)
|
||||
report: dict[str, Any] = {
|
||||
"source_bundle_bytes": bundle_bytes,
|
||||
"estimated_output_bytes": output_allowance,
|
||||
"headroom_bytes": headroom,
|
||||
"work_dir": str(work_root),
|
||||
"work_dir_free_bytes": work_free,
|
||||
"source_bundle_bytes": bundle_bytes, "estimated_output_bytes": output_allowance, "headroom_bytes": headroom,
|
||||
"work_dir": str(work_root), "work_dir_free_bytes": work_free,
|
||||
}
|
||||
|
||||
if output_parent is None or _same_filesystem(work_root, output_parent):
|
||||
@@ -206,23 +179,16 @@ def _disk_space_preflight(source: Path, work_root: Path, output_parent: Optional
|
||||
work_required = bundle_bytes + headroom
|
||||
output_required = output_allowance + headroom
|
||||
report.update({
|
||||
"shared_filesystem": False,
|
||||
"work_dir_required_bytes": work_required,
|
||||
"output_dir": str(output_parent),
|
||||
"output_dir_free_bytes": output_free,
|
||||
"output_dir_required_bytes": output_required,
|
||||
"shared_filesystem": False, "work_dir_required_bytes": work_required, "output_dir": str(output_parent),
|
||||
"output_dir_free_bytes": output_free, "output_dir_required_bytes": output_required,
|
||||
})
|
||||
shortages: list[str] = []
|
||||
if work_free < work_required:
|
||||
shortages.append(
|
||||
f"{work_root}: {_format_bytes(work_free)} available, "
|
||||
f"{_format_bytes(work_required)} required"
|
||||
)
|
||||
if output_free < output_required:
|
||||
shortages.append(
|
||||
f"{output_parent}: {_format_bytes(output_free)} available, "
|
||||
f"{_format_bytes(output_required)} required"
|
||||
shortages = [
|
||||
f"{where}: {_format_bytes(free)} available, {_format_bytes(required)} required"
|
||||
for where, free, required in (
|
||||
(work_root, work_free, work_required), (output_parent, output_free, output_required),
|
||||
)
|
||||
if free < required
|
||||
]
|
||||
if shortages:
|
||||
raise SessionRecoverySafetyError(
|
||||
"Not enough free disk space for safe recovery: "
|
||||
@@ -235,12 +201,9 @@ def _disk_space_preflight(source: Path, work_root: Path, output_parent: Optional
|
||||
def _copy_source_bundle(source: Path, snapshot_dir: Path) -> tuple[Path, list[str]]:
|
||||
"""Copy the source DB bundle aside so SQLite never opens the original.
|
||||
|
||||
The whole copy runs inside ``offline_file_access``, which holds the connection-lifecycle lock
|
||||
for its duration.
|
||||
|
||||
Recovery normally runs as its own short-lived CLI process against an offline/quarantined file,
|
||||
so the refusal should never fire; the guard keeps this path consistent with
|
||||
``hermes_state._backup_db_file``.
|
||||
The whole copy runs inside ``offline_file_access`` (holds the connection-lifecycle lock). Recovery
|
||||
normally runs as its own CLI process against an offline file, so the refusal should never fire; the
|
||||
guard keeps this path consistent with ``hermes_state._backup_db_file``.
|
||||
"""
|
||||
from hermes_cli.sqlite_safe_read import LiveConnectionError, offline_file_access
|
||||
|
||||
@@ -250,11 +213,10 @@ def _copy_source_bundle(source: Path, snapshot_dir: Path) -> tuple[Path, list[st
|
||||
with offline_file_access(source, what="snapshot"):
|
||||
for suffix in _SIDECAR_SUFFIXES:
|
||||
source_part = _sidecar_path(source, suffix)
|
||||
if not source_part.exists():
|
||||
continue
|
||||
destination_part = _sidecar_path(snapshot_source, suffix)
|
||||
shutil.copy2(source_part, destination_part)
|
||||
copied.append(destination_part.name)
|
||||
if source_part.exists():
|
||||
destination_part = _sidecar_path(snapshot_source, suffix)
|
||||
shutil.copy2(source_part, destination_part)
|
||||
copied.append(destination_part.name)
|
||||
except LiveConnectionError as exc:
|
||||
raise SessionRecoverySafetyError(str(exc)) from exc
|
||||
return snapshot_source, copied
|
||||
@@ -340,9 +302,10 @@ def _copy_rows(
|
||||
result["error"] = str(exc)
|
||||
return result
|
||||
|
||||
complete = expected_rows is None or result["copied_rows"] == expected_rows
|
||||
result["status"] = "complete" if complete else "partial"
|
||||
if result["status"] == "partial":
|
||||
if expected_rows is None or result["copied_rows"] == expected_rows:
|
||||
result["status"] = "complete"
|
||||
else:
|
||||
result["status"] = "partial"
|
||||
result["error"] = f"copied {result['copied_rows']} of {expected_rows} readable rows"
|
||||
return result
|
||||
|
||||
@@ -373,8 +336,7 @@ def _inspect_connection(conn: sqlite3.Connection) -> dict[str, Any]:
|
||||
report["journal_mode"] = _journal_mode(conn)
|
||||
except sqlite3.DatabaseError as exc:
|
||||
report["journal_mode"] = None
|
||||
# Journal metadata is useful context but not canonical session data.
|
||||
# A damaged journal pragma must not block rows that are still readable.
|
||||
# Journal metadata is context, not canonical data: a damaged pragma must not block readable rows.
|
||||
report["warnings"].append(f"journal mode: {exc}")
|
||||
|
||||
for table in _INVENTORY_TABLES:
|
||||
@@ -394,11 +356,9 @@ def _snapshot_and_inspect(
|
||||
) -> tuple[tempfile.TemporaryDirectory[str], Path, dict[str, Any]]:
|
||||
before = _source_fingerprint(source)
|
||||
temp_dir = tempfile.TemporaryDirectory(prefix="hermes-session-recovery-", dir=str(work_root))
|
||||
snapshot_dir = Path(temp_dir.name)
|
||||
try:
|
||||
snapshot_source, copied = _copy_source_bundle(source, snapshot_dir)
|
||||
after = _source_fingerprint(source)
|
||||
if before != after:
|
||||
snapshot_source, copied = _copy_source_bundle(source, Path(temp_dir.name))
|
||||
if _source_fingerprint(source) != before:
|
||||
raise SessionRecoverySafetyError(
|
||||
"The source database bundle changed while it was being copied. "
|
||||
"Stop every Hermes process using this profile and retry. "
|
||||
@@ -427,7 +387,6 @@ def _snapshot_and_inspect(
|
||||
|
||||
def inspect_session_database(source_path: Path, *, work_dir: Optional[Path] = None) -> dict[str, Any]:
|
||||
"""Inspect canonical table readability without opening the source itself."""
|
||||
|
||||
source, _, work_root = _validate_paths(source_path, work_dir=work_dir)
|
||||
disk_space = _disk_space_preflight(source, work_root, None)
|
||||
temp_dir, _, inspection = _snapshot_and_inspect(source, work_root)
|
||||
@@ -444,13 +403,8 @@ def inspect_session_database(source_path: Path, *, work_dir: Optional[Path] = No
|
||||
|
||||
|
||||
def _ensure_auxiliary_destination_schema(destination: sqlite3.Connection, table: str) -> None:
|
||||
"""Create a lazy auxiliary table on the recovered destination.
|
||||
|
||||
Recovery initializes the destination via base ``SessionDB``, which does not create gateway-owned
|
||||
tables; copying into a missing table would report ``missing``/``no compatible columns`` and drop
|
||||
the rows.
|
||||
"""
|
||||
|
||||
"""Create a lazy gateway-owned table on the destination; without it the copy would report
|
||||
``missing``/``no compatible columns`` and drop the rows."""
|
||||
initialize = _AUXILIARY_TABLE_SCHEMAS.get(table)
|
||||
if initialize is None:
|
||||
raise SessionRecoverySafetyError(f"no destination schema initializer registered for table {table!r}")
|
||||
@@ -506,8 +460,7 @@ def _copy_table(
|
||||
|
||||
|
||||
def _append_skipped_range(ranges: list[dict[str, Any]], low: int, high: int, error: str) -> None:
|
||||
"""Record skipped rowid ranges without producing one entry per row."""
|
||||
|
||||
"""Record skipped rowid ranges, merging adjacent same-error ranges (not one entry per row)."""
|
||||
if ranges and ranges[-1]["high"] + 1 == low and ranges[-1]["error"] == error:
|
||||
ranges[-1]["high"] = high
|
||||
return
|
||||
@@ -516,11 +469,9 @@ def _append_skipped_range(ranges: list[dict[str, Any]], low: int, high: int, err
|
||||
|
||||
def _salvage_rowid_bounds(source: sqlite3.Connection, table: str) -> dict[str, Any]:
|
||||
"""Find the readable rowid edges without scanning the complete table."""
|
||||
|
||||
result: dict[str, Any] = {"errors": [], "fallback_edges": []}
|
||||
rows: dict[str, Optional[int]] = {"low": None, "high": None}
|
||||
directions = (("low", "ASC"), ("high", "DESC"))
|
||||
for edge, direction in directions:
|
||||
for edge, direction in (("low", "ASC"), ("high", "DESC")):
|
||||
try:
|
||||
row = source.execute(f'SELECT rowid FROM "{table}" ORDER BY rowid {direction} LIMIT 1').fetchone()
|
||||
if row is not None:
|
||||
@@ -532,10 +483,8 @@ def _salvage_rowid_bounds(source: sqlite3.Connection, table: str) -> dict[str, A
|
||||
result["empty" if not result["errors"] else "unavailable"] = True
|
||||
return result
|
||||
|
||||
# A damaged edge can prevent one of the ordered probes from completing.
|
||||
# Keep the other readable edge and bound the missing side by SQLite's
|
||||
# rowid domain. Range bisection can then approach the surviving data
|
||||
# without assuming that user-created databases contain only positive IDs.
|
||||
# A damaged edge can stop one ordered probe. Keep the readable edge and bound the other side by the
|
||||
# SQLite rowid domain, so bisection never assumes user databases hold only positive ids.
|
||||
if rows["low"] is None:
|
||||
rows["low"] = _MIN_SQLITE_ROWID
|
||||
result["fallback_edges"].append("low")
|
||||
@@ -544,10 +493,8 @@ def _salvage_rowid_bounds(source: sqlite3.Connection, table: str) -> dict[str, A
|
||||
result["fallback_edges"].append("high")
|
||||
|
||||
result.update(rows)
|
||||
# Issue #80205: a damaged ordered edge probe used to substitute the whole
|
||||
# SQLite rowid domain, and bisecting that synthetic tail exhausted the
|
||||
# range-query budget while readable tail rows were still waiting to be
|
||||
# copied. Gallop outward from the surviving edge for a finite bound first.
|
||||
# Bisecting the whole synthetic domain tail used to exhaust the range-query budget before readable
|
||||
# tail rows were copied (#80205); gallop outward from the surviving edge for a finite bound first.
|
||||
if result["fallback_edges"]:
|
||||
result["edge_probes"] = []
|
||||
for edge, anchor_edge in (("high", "low"), ("low", "high")):
|
||||
@@ -560,14 +507,9 @@ def _salvage_rowid_bounds(source: sqlite3.Connection, table: str) -> dict[str, A
|
||||
|
||||
|
||||
def _probe_populated_edge(source: sqlite3.Connection, table: str, *, edge: str, anchor: int) -> dict[str, Any]:
|
||||
"""Find a finite bound for a damaged rowid edge (issue #80205).
|
||||
|
||||
Substituting the whole rowid domain for a failed edge probe made range bisection burn the entire
|
||||
salvage budget on an empty synthetic tail, after which readable rows were silently skipped.
|
||||
Instead gallop outward from the readable ``anchor`` with exponential offsets: a clean "no rows
|
||||
beyond X" caps the domain, an error or hit keeps growing. At most ~64 probes per edge.
|
||||
"""Finite bound for a damaged rowid edge: gallop outward from the readable ``anchor`` with doubling
|
||||
offsets. A clean "no rows beyond X" caps the domain; an error or a hit keeps growing (~64 probes max).
|
||||
"""
|
||||
|
||||
ascending = edge == "high"
|
||||
comparison = ">" if ascending else "<"
|
||||
probe_sql = (
|
||||
@@ -582,27 +524,19 @@ def _probe_populated_edge(source: sqlite3.Connection, table: str, *, edge: str,
|
||||
while True:
|
||||
candidate = position + span if ascending else position - span
|
||||
if (ascending and candidate >= domain_limit) or (not ascending and candidate <= domain_limit):
|
||||
# No clean empty-tail answer before the domain edge; keep the
|
||||
# domain fallback rather than inventing a bound.
|
||||
result["bound"] = domain_limit
|
||||
result["bound"] = domain_limit # no clean empty-tail answer: keep the domain fallback
|
||||
return result
|
||||
result["probes"] += 1
|
||||
try:
|
||||
row = source.execute(probe_sql, (candidate,)).fetchone()
|
||||
except sqlite3.DatabaseError:
|
||||
# Damage on the probe path — inconclusive, widen further.
|
||||
except sqlite3.DatabaseError: # damage on the probe path: inconclusive, widen further
|
||||
span *= 2
|
||||
continue
|
||||
if row is None:
|
||||
# Clean answer: nothing beyond ``candidate``. The salvageable
|
||||
# data ends at or before it, so the synthetic domain tail is
|
||||
# provably empty and need not be bisected at all.
|
||||
if row is None: # nothing beyond candidate: the synthetic domain tail is provably empty
|
||||
result["bound"] = candidate
|
||||
result["capped"] = True
|
||||
return result
|
||||
# Rows exist beyond the candidate; advance the anchor. The span keeps
|
||||
# doubling (never resets) so the whole gallop stays O(log range).
|
||||
position = int(row[0])
|
||||
position = int(row[0]) # rows exist beyond; advance. Span never resets -> O(log range)
|
||||
span *= 2
|
||||
|
||||
|
||||
@@ -724,16 +658,9 @@ def _copy_table_salvage(
|
||||
row_filter: Optional[Callable[[tuple[Any, ...], tuple[str, ...]], bool]] = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Best-effort rowid-range copy that continues past damaged source pages."""
|
||||
|
||||
result: dict[str, Any] = {
|
||||
"mode": "rowid_range_salvage",
|
||||
"source_rows": source_rows,
|
||||
"copied_rows": 0,
|
||||
"excluded_rows": 0,
|
||||
"columns": [],
|
||||
"range_queries": 0,
|
||||
"exact_lookup_recovered": 0,
|
||||
"skipped_rowid_ranges": [],
|
||||
"mode": "rowid_range_salvage", "source_rows": source_rows, "copied_rows": 0, "excluded_rows": 0,
|
||||
"columns": [], "range_queries": 0, "exact_lookup_recovered": 0, "skipped_rowid_ranges": [],
|
||||
}
|
||||
columns = _compatible_columns(source, destination, table, result)
|
||||
if columns is None:
|
||||
@@ -802,16 +729,14 @@ def _state_meta_precheck(
|
||||
) -> Optional[dict[str, Any]]:
|
||||
"""Terminal ``state_meta`` result when the key/value schema is unusable, else ``None``.
|
||||
|
||||
Status matters here. In salvage mode an unusable-but-PRESENT table reports ``failed``, not
|
||||
``missing``: verification only escalates ``failed``/``partial`` into a warning +
|
||||
``loss_detected``, so reporting ``missing`` would silently drop real metadata and still claim
|
||||
``complete=True``.
|
||||
In salvage mode an unusable-but-PRESENT table reports ``failed``, not ``missing``: verification
|
||||
only escalates ``failed``/``partial`` into a warning + ``loss_detected``, so ``missing`` would
|
||||
silently drop real metadata and still claim ``complete=True``.
|
||||
"""
|
||||
extra = {"mode": "rowid_range_salvage"} if salvage else {}
|
||||
source_columns = _table_columns(source, "state_meta")
|
||||
if not {"key", "value"}.issubset(source_columns):
|
||||
if salvage and source_columns:
|
||||
# Present but unusable: this IS data loss and must be reported.
|
||||
if salvage and source_columns: # present but unusable: real data loss
|
||||
return _state_meta_result(
|
||||
source_rows,
|
||||
**extra,
|
||||
@@ -821,8 +746,7 @@ def _state_meta_precheck(
|
||||
f"columns (found: {', '.join(source_columns) or 'none'})"
|
||||
),
|
||||
)
|
||||
# Genuinely absent from the source — nothing was lost.
|
||||
return _state_meta_result(source_rows, **extra, status="missing")
|
||||
return _state_meta_result(source_rows, **extra, status="missing") # genuinely absent: nothing lost
|
||||
if not {"key", "value"}.issubset(_table_columns(destination, "state_meta")):
|
||||
return _state_meta_result(
|
||||
source_rows, **extra, status="failed", error="destination state_meta schema is incomplete"
|
||||
@@ -889,14 +813,9 @@ def _copy_state_meta(
|
||||
def _reconstruct_missing_sessions(destination: sqlite3.Connection) -> dict[str, Any]:
|
||||
"""Recreate placeholder session rows for salvaged orphaned messages.
|
||||
|
||||
When the ``sessions`` b-tree is damaged worse than ``messages``, salvage can recover the
|
||||
conversation text while recovering few or none of the session rows that own it. Deleting those
|
||||
messages as "orphans" throws away the only readable copy of the user's data — the exact opposite
|
||||
of what ``--allow-partial`` is for.
|
||||
|
||||
Instead, synthesize a minimal session row per orphaned ``session_id`` (only
|
||||
``id``/``source``/``started_at`` are NOT NULL) so the messages stay reachable and foreign keys
|
||||
hold. ``started_at`` is taken from the earliest surviving message so ordering stays sane.
|
||||
When ``sessions`` is damaged worse than ``messages``, deleting the recovered messages as orphans
|
||||
would discard the only readable copy of the user's data. Instead synthesize a minimal session per
|
||||
orphaned ``session_id`` (``started_at`` = earliest surviving message) so FKs hold.
|
||||
"""
|
||||
result: dict[str, Any] = {"sessions_reconstructed": 0, "messages_retained": 0}
|
||||
if not (_table_columns(destination, "sessions") and _table_columns(destination, "messages")):
|
||||
@@ -948,24 +867,15 @@ _RELINK_COUNTERS = ("session_prompt_refs_cleared", "sessions_parent_cleared")
|
||||
|
||||
|
||||
def _cleanup_partial_orphans(destination: sqlite3.Connection) -> dict[str, Any]:
|
||||
"""Reconcile references to sessions that could not be salvaged.
|
||||
|
||||
Messages are never discarded for lack of a session row: their owning session is reconstructed as
|
||||
a placeholder first (see :func:`_reconstruct_missing_sessions`). Only rows that remain orphaned
|
||||
after that — and rows in tables carrying no recoverable user content — are removed.
|
||||
"""
|
||||
|
||||
"""Reconcile references to unsalvageable sessions. Messages are never dropped for lack of a
|
||||
session row (placeholders are rebuilt first); only rows still orphaned after that are removed."""
|
||||
result: dict[str, Any] = {
|
||||
"session_prompt_refs_cleared": 0,
|
||||
"system_prompts_removed": 0,
|
||||
"sessions_parent_cleared": 0,
|
||||
"sessions_reconstructed": 0,
|
||||
"messages_retained": 0,
|
||||
"session_prompt_refs_cleared": 0, "system_prompts_removed": 0, "sessions_parent_cleared": 0,
|
||||
"sessions_reconstructed": 0, "messages_retained": 0,
|
||||
**{f"{table}_removed": 0 for table in _DEPENDENT_TABLES},
|
||||
}
|
||||
with _immediate_transaction(destination):
|
||||
# Rebuild owners BEFORE any orphan deletion so salvaged conversation
|
||||
# text is never dropped for want of a session row.
|
||||
# Rebuild owners BEFORE any orphan deletion.
|
||||
result.update(_reconstruct_missing_sessions(destination))
|
||||
|
||||
result["sessions_parent_cleared"] = _reconcile(
|
||||
@@ -997,9 +907,7 @@ def _cleanup_partial_orphans(destination: sqlite3.Connection) -> dict[str, Any]:
|
||||
f'NOT EXISTS (SELECT 1 FROM sessions WHERE sessions.id = "{table}".session_id)',
|
||||
f'DELETE FROM "{table}"',
|
||||
)
|
||||
# Only destructive/relinking actions belong in this total. The
|
||||
# reconstruction counters describe data RETAINED, so summing them here
|
||||
# would report saving the user's messages as if it were losing them.
|
||||
# Only destructive/relinking actions count: reconstruction counters describe RETAINED data.
|
||||
result["total_removed_or_relinked"] = sum(
|
||||
int(result[key]) for key in (*_RELINK_COUNTERS, *(f"{table}_removed" for table in _DEPENDENT_TABLES))
|
||||
)
|
||||
@@ -1169,7 +1077,6 @@ def _verify_recovered_database(
|
||||
|
||||
def _finalize_derived_metadata(destination: sqlite3.Connection) -> dict[str, Any]:
|
||||
"""Stamp only metadata that the newly created destination actually owns."""
|
||||
|
||||
fts_tables = {
|
||||
str(row[0])
|
||||
for row in destination.execute(
|
||||
@@ -1205,21 +1112,12 @@ def _recover_via_lost_and_found(
|
||||
disk_space: dict[str, Any],
|
||||
missing_required: list[str],
|
||||
) -> dict[str, Any]:
|
||||
"""Best-effort page-level salvage when table schemas are unreadable.
|
||||
|
||||
Shells out to the sqlite3 CLI's ``.recover`` (a shell-only feature, not part of Python's
|
||||
``sqlite3`` module) to rebuild rows into a scratch lost_and_found database, then heuristically
|
||||
maps them into a fresh current-schema database. The result is explicitly labeled best-effort.
|
||||
"""
|
||||
|
||||
"""Best-effort page-level salvage when table schemas are unreadable: the sqlite3 CLI's ``.recover``
|
||||
(shell-only, not in Python's ``sqlite3``) rebuilds rows into a scratch lost_and_found database which
|
||||
is then heuristically mapped into a fresh current-schema database."""
|
||||
from hermes_cli.session_lost_and_found import (
|
||||
SQLITE3_CLI_GUIDANCE,
|
||||
LostAndFoundError,
|
||||
find_sqlite3_cli,
|
||||
map_lost_and_found_rows,
|
||||
rebuild_fts_indexes,
|
||||
run_cli_lost_and_found_recover,
|
||||
stub_missing_parent_sessions,
|
||||
SQLITE3_CLI_GUIDANCE, LostAndFoundError, find_sqlite3_cli, map_lost_and_found_rows, rebuild_fts_indexes,
|
||||
run_cli_lost_and_found_recover, stub_missing_parent_sessions,
|
||||
)
|
||||
|
||||
sqlite3_bin = find_sqlite3_cli()
|
||||
@@ -1255,21 +1153,15 @@ def _recover_via_lost_and_found(
|
||||
table: {
|
||||
"mode": "lost_and_found_salvage",
|
||||
"status": "partial",
|
||||
"copied_rows": (
|
||||
int(mapping["direct_table_rows"].get(table) or 0)
|
||||
+ int(mapping["mapped"].get(table) or 0)
|
||||
),
|
||||
"copied_rows": int(mapping["direct_table_rows"].get(table) or 0) + int(mapping["mapped"].get(table) or 0),
|
||||
"error": "recovered via page-level lost_and_found salvage; "
|
||||
"row completeness cannot be verified against the source",
|
||||
}
|
||||
for table in ("sessions", "messages", "session_model_usage")
|
||||
}
|
||||
|
||||
orphan_cleanup = {
|
||||
"sessions_reconstructed": stubbing["sessions_stubbed"],
|
||||
"messages_retained": stubbing["messages_retained"],
|
||||
"messages_removed": 0,
|
||||
"total_removed_or_relinked": 0,
|
||||
"sessions_reconstructed": stubbing["sessions_stubbed"], "messages_retained": stubbing["messages_retained"],
|
||||
"messages_removed": 0, "total_removed_or_relinked": 0,
|
||||
}
|
||||
verification = _verify_recovered_database(
|
||||
output,
|
||||
@@ -1287,23 +1179,10 @@ def _recover_via_lost_and_found(
|
||||
verification["complete"] = False
|
||||
|
||||
return _recovery_report(
|
||||
source,
|
||||
output,
|
||||
inspection,
|
||||
disk_space,
|
||||
verification,
|
||||
on_source_change="healthy",
|
||||
allow_partial=True,
|
||||
mode="lost_and_found_salvage",
|
||||
best_effort=True,
|
||||
unreadable_schemas=missing_required,
|
||||
sqlite3_cli=cli_report,
|
||||
lost_and_found=mapping,
|
||||
session_stubs=stubbing,
|
||||
fts_rebuild=fts,
|
||||
copy=copy_report,
|
||||
orphan_cleanup=orphan_cleanup,
|
||||
derived_metadata=derived_metadata,
|
||||
source, output, inspection, disk_space, verification, on_source_change="healthy",
|
||||
allow_partial=True, mode="lost_and_found_salvage", best_effort=True, unreadable_schemas=missing_required,
|
||||
sqlite3_cli=cli_report, lost_and_found=mapping, session_stubs=stubbing, fts_rebuild=fts, copy=copy_report,
|
||||
orphan_cleanup=orphan_cleanup, derived_metadata=derived_metadata,
|
||||
)
|
||||
|
||||
|
||||
@@ -1317,10 +1196,8 @@ def _recovery_report(
|
||||
on_source_change: str,
|
||||
**fields: Any,
|
||||
) -> dict[str, Any]:
|
||||
"""The ``recover`` report: shared header, mode-specific ``fields``, then the verdict.
|
||||
|
||||
A source bundle that changed during recovery is a verification error and additionally clears
|
||||
``verification[on_source_change]``.
|
||||
"""The ``recover`` report: shared header, mode-specific ``fields``, then the verdict. A source bundle
|
||||
that changed during recovery is a verification error and also clears ``verification[on_source_change]``.
|
||||
"""
|
||||
source_unchanged = _source_fingerprint(source) == inspection["source_fingerprint"]
|
||||
if not source_unchanged:
|
||||
@@ -1335,10 +1212,8 @@ def _recovery_report(
|
||||
"source_unchanged": source_unchanged,
|
||||
"disk_space": disk_space,
|
||||
"inspection": {
|
||||
"journal_mode": inspection.get("journal_mode"),
|
||||
"tables": inspection["tables"],
|
||||
"errors": inspection["errors"],
|
||||
"warnings": inspection["warnings"],
|
||||
"journal_mode": inspection.get("journal_mode"), "tables": inspection["tables"],
|
||||
"errors": inspection["errors"], "warnings": inspection["warnings"],
|
||||
},
|
||||
**fields,
|
||||
"verification": verification,
|
||||
@@ -1358,12 +1233,8 @@ def recover_session_database(
|
||||
progress_cb: Optional[ProgressCallback] = None,
|
||||
allow_partial: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
"""Recover canonical rows into a separate current-schema database.
|
||||
|
||||
The source path and its sidecars are copied before SQLite opens anything. ``output_path`` must
|
||||
not exist and is never swapped into place.
|
||||
"""
|
||||
|
||||
"""Recover canonical rows into a separate current-schema database. The source and its sidecars are
|
||||
copied before SQLite opens anything; ``output_path`` must not exist and is never swapped into place."""
|
||||
if chunk_size <= 0:
|
||||
raise SessionRecoverySafetyError("chunk_size must be greater than zero")
|
||||
|
||||
@@ -1382,22 +1253,12 @@ def recover_session_database(
|
||||
)
|
||||
if allow_partial:
|
||||
missing_required = [
|
||||
table
|
||||
for table in ("sessions", "messages")
|
||||
if not inspection["tables"][table].get("available")
|
||||
table for table in ("sessions", "messages") if not inspection["tables"][table].get("available")
|
||||
]
|
||||
if missing_required:
|
||||
# SQL-level salvage is impossible without readable table
|
||||
# schemas. Fall back to page-level lost_and_found salvage via
|
||||
# the sqlite3 CLI's .recover (a shell-only feature).
|
||||
if missing_required: # no readable schema -> page-level lost_and_found salvage
|
||||
return _recover_via_lost_and_found(
|
||||
source=source,
|
||||
snapshot_source=snapshot_source,
|
||||
snapshot_dir=Path(temp_dir.name),
|
||||
output=output,
|
||||
inspection=inspection,
|
||||
disk_space=disk_space,
|
||||
missing_required=missing_required,
|
||||
source=source, snapshot_source=snapshot_source, snapshot_dir=Path(temp_dir.name), output=output,
|
||||
inspection=inspection, disk_space=disk_space, missing_required=missing_required,
|
||||
)
|
||||
|
||||
source_conn = _connect(snapshot_source)
|
||||
@@ -1417,13 +1278,8 @@ def recover_session_database(
|
||||
if table in _AUXILIARY_TABLES:
|
||||
_ensure_auxiliary_destination_schema(destination_conn, table)
|
||||
copy_report[table] = _copy_table(
|
||||
source_conn,
|
||||
destination_conn,
|
||||
table,
|
||||
salvage=allow_partial,
|
||||
chunk_size=chunk_size,
|
||||
progress_cb=progress_cb,
|
||||
source_rows=inspection["tables"][table].get("rows"),
|
||||
source_conn, destination_conn, table, salvage=allow_partial, chunk_size=chunk_size,
|
||||
progress_cb=progress_cb, source_rows=inspection["tables"][table].get("rows"),
|
||||
)
|
||||
orphan_cleanup = _cleanup_partial_orphans(destination_conn) if allow_partial else None
|
||||
derived_metadata = _finalize_derived_metadata(destination_conn)
|
||||
@@ -1445,15 +1301,8 @@ def recover_session_database(
|
||||
orphan_cleanup=orphan_cleanup,
|
||||
)
|
||||
return _recovery_report(
|
||||
source,
|
||||
output,
|
||||
inspection,
|
||||
disk_space,
|
||||
verification,
|
||||
on_source_change="complete",
|
||||
allow_partial=allow_partial,
|
||||
copy=copy_report,
|
||||
orphan_cleanup=orphan_cleanup,
|
||||
source, output, inspection, disk_space, verification, on_source_change="complete",
|
||||
allow_partial=allow_partial, copy=copy_report, orphan_cleanup=orphan_cleanup,
|
||||
derived_metadata=derived_metadata,
|
||||
)
|
||||
finally:
|
||||
@@ -1462,7 +1311,6 @@ def recover_session_database(
|
||||
|
||||
def write_recovery_report(path: Path, report: dict[str, Any]) -> Path:
|
||||
"""Write a JSON report without overwriting an existing file."""
|
||||
|
||||
destination = _resolved_output_path(path)
|
||||
with destination.open("x", encoding="utf-8") as handle:
|
||||
json.dump(report, handle, indent=2, sort_keys=True)
|
||||
|
||||
Reference in New Issue
Block a user