fix(backup): never treat a file as covered without a surviving archive object (S)

src/brainlayer/backup_daily.py

@@ -695,7 +695,10 @@ def upload_file_to_drive_raw(
     total = file_path.stat().st_size
     metadata = {"name": file_path.name, "parents": [folder_id]}
     init = requests.post(
-        "https://www.googleapis.com/upload/drive/v3/files?uploadType=resumable&supportsAllDrives=true&fields=id,name,size",
+        "https://www.googleapis.com/upload/drive/v3/files?uploadType=resumable&supportsAllDrives=true"
+        # md5Checksum is REQUIRED: retention coverage compares it against the surviving
+        # object. Without it the integrity branch silently becomes dead code (PR #815 review).
+        "&fields=id,name,size,md5Checksum",
         headers={
             "Authorization": f"Bearer {credentials.token}",
             "Content-Type": "application/json; charset=UTF-8",

src/brainlayer/jsonl_backup.py

@@ -184,10 +184,47 @@ def _discover_jsonl_candidates(source_roots: list[Path | BackupSourceRoot]) -> l
     return candidates
 
 
-def _state_matches(entry: Any, candidate: JsonlCandidate) -> bool:
+def _state_matches(
+    entry: Any,
+    candidate: JsonlCandidate,
+    surviving_archives: dict[str, str | None] | None = None,
+) -> bool:
+    """A file is covered only while a SURVIVING archive object holds its exact bytes.
+
+    ``surviving_archives`` is the set of archive object names still present in the
+    backup folder. When it is supplied (the upload path) an entry must name one of
+    them and its recorded content hash must still match the file on disk. Entries
+    written before archive provenance was recorded name no archive, so they cannot
+    prove survival and are deliberately treated as uncovered -- re-bundling costs a
+    night of upload, trusting them can cost the only remaining copy.
+    """
     if not isinstance(entry, dict):
         return False
-    return entry.get("mtime") == candidate.mtime and entry.get("size") == candidate.size
+    if entry.get("mtime") != candidate.mtime or entry.get("size") != candidate.size:
+        return False
+    if surviving_archives is None:
+        return True
+    archive_id = entry.get("archive_id")
+    if not isinstance(archive_id, str) or archive_id not in surviving_archives:
+        return False
+    recorded_md5 = entry.get("archive_md5")
+    if isinstance(recorded_md5, str) and recorded_md5:
+        live_md5 = surviving_archives[archive_id]
+        # An object we cannot re-verify cannot prove coverage. Fail closed: the cost is
+        # re-bundling, the cost of the other direction is the only remaining copy.
+        if not isinstance(live_md5, str) or live_md5 != recorded_md5:
+            return False
+    recorded_hash = entry.get("sha256")
+    if not isinstance(recorded_hash, str) or not recorded_hash:
+        return False
+    try:
+        return recorded_hash == _sha256_file(candidate.path)
+    except OSError:
+        # The source vanished between discovery and hashing. It cannot prove coverage, and it
+        # must not abort the run: this job is what PREVENTS data loss, so a single unreadable
+        # file taking the whole nightly backup down is the wrong failure. _drop_vanished keeps
+        # it out of the bundle too, so returning False here cannot strand it in `changed`.
+        return False
 
 
 def _backup_unit_key(candidate: JsonlCandidate) -> tuple[int, str]:
@@ -204,24 +241,66 @@ def _select_backup_candidates(
     state: dict[str, Any],
     now: float,
     active_skip_seconds: int,
-) -> tuple[list[JsonlCandidate], list[JsonlCandidate], int]:
+    surviving_archives: dict[str, str | None] | None = None,
+) -> tuple[list[JsonlCandidate], list[JsonlCandidate], int, int]:
     grouped: dict[tuple[int, str], list[JsonlCandidate]] = {}
     for candidate in candidates:
         grouped.setdefault(_backup_unit_key(candidate), []).append(candidate)
 
     changed: list[JsonlCandidate] = []
     active: list[JsonlCandidate] = []
     covered = 0
+    vanished = 0
     state_files = state.get("files", {})
     for backup_unit in grouped.values():
         if any(now - candidate.mtime < active_skip_seconds for candidate in backup_unit):
             active.extend(backup_unit)
             continue
-        if all(_state_matches(state_files.get(candidate.path.as_posix()), candidate) for candidate in backup_unit):
+        if all(
+            _state_matches(state_files.get(candidate.path.as_posix()), candidate, surviving_archives)
+            for candidate in backup_unit
+        ):
             covered += len(backup_unit)
             continue
-        changed.extend(backup_unit)
-    return changed, active, covered
+        readable = [c for c in backup_unit if c.path.exists()]
+        vanished += len(backup_unit) - len(readable)
+        changed.extend(readable)
+    return changed, active, covered, vanished
+
+
+def _list_surviving_archives(service: Any, folder_parts: list[str]) -> dict[str, str | None]:
+    """Surviving archive objects in the backup folder, keyed by Drive object ID.
+
+    Keyed by ID, never by name: Drive permits duplicate names in one folder, so a
+    same-named replacement would otherwise masquerade as the original object and
+    prove a survival that never happened. The value is the object's md5Checksum
+    when Drive reports one, so out-of-band modification of a surviving object is
+    detectable too.
+    """
+    folder_id = backup_daily.ensure_drive_folder_chain(service, folder_parts)
+    surviving: dict[str, str | None] = {}
+    page_token = None
+    while True:
+        response = (
+            service.files()
+            .list(
+                q=f"'{folder_id}' in parents and trashed = false",
+                spaces="drive",
+                fields="nextPageToken,files(id,name,md5Checksum)",
+                pageSize=1000,
+                pageToken=page_token,
+                supportsAllDrives=True,
+            )
+            .execute()
+        )
+        for item in response.get("files", []):
+            file_id = item.get("id")
+            if isinstance(file_id, str) and file_id:
+                surviving[file_id] = item.get("md5Checksum")
+        page_token = response.get("nextPageToken")
+        if not page_token:
+            break
+    return surviving
 
 
 def _archive_name(candidate: JsonlCandidate) -> str:
@@ -292,7 +371,30 @@ def _upload_forever_files(
     return uploaded
 
 
-def create_jsonl_bundle(candidates: list[JsonlCandidate], staging_dir: Path, *, date_stamp: str) -> Path:
+class _HashingReader:
+    """File wrapper that digests exactly the bytes handed to tarfile.
+
+    tarfile pulls through ``read``, so the digest is of the archived content itself --
+    not of a separate read that could see different bytes -- at O(buffer) memory rather
+    than O(file). The largest real source JSONL is ~375MB, so that distinction matters.
+    """
+
+    def __init__(self, handle: Any) -> None:
+        self._handle = handle
+        self._digest = hashlib.sha256()
+
+    def read(self, size: int = -1) -> bytes:
+        chunk = self._handle.read(size)
+        self._digest.update(chunk)
+        return chunk
+
+    def hexdigest(self) -> str:
+        return self._digest.hexdigest()
+
+
+def create_jsonl_bundle_with_digests(
+    candidates: list[JsonlCandidate], staging_dir: Path, *, date_stamp: str
+) -> tuple[Path, dict[str, str]]:
     if not candidates:
         raise ValueError("create_jsonl_bundle requires at least one candidate")
     staging_dir = Path(staging_dir).expanduser()
@@ -302,13 +404,29 @@ def create_jsonl_bundle(candidates: list[JsonlCandidate], staging_dir: Path, *,
         prefix=f".{archive_path.name}.", suffix=".tmp", dir=staging_dir, delete=False
     ) as tmp:
         temp_path = Path(tmp.name)
+    digests: dict[str, str] = {}
     try:
         with tarfile.open(temp_path, "w:gz") as tar:
             for candidate in candidates:
-                tar.add(candidate.path, arcname=_archive_name(candidate), recursive=False)
+                # Hash and archive the SAME bytes, without holding the file in memory.
+                # Re-reading the source afterwards could record a digest for content the
+                # archive does not contain -- a file that changed while keeping its mtime
+                # and size would then read as covered. Sources reach ~375MB, so the digest
+                # is taken from the very stream tarfile consumes rather than from a copy.
+                info = tar.gettarinfo(str(candidate.path), arcname=_archive_name(candidate))
+                with candidate.path.open("rb") as handle:
+                    reader = _HashingReader(handle)
+                    tar.addfile(info, reader)
+                digests[candidate.path.as_posix()] = reader.hexdigest()
         os.replace(temp_path, archive_path)
     finally:
         temp_path.unlink(missing_ok=True)
+    return archive_path, digests
+
+
+def create_jsonl_bundle(candidates: list[JsonlCandidate], staging_dir: Path, *, date_stamp: str) -> Path:
+    """Backwards-compatible wrapper returning only the archive path."""
+    archive_path, _ = create_jsonl_bundle_with_digests(candidates, staging_dir, date_stamp=date_stamp)
     return archive_path
 
 
@@ -336,10 +454,32 @@ def verify_jsonl_bundle(archive_path: Path, *, expected_file_count: int) -> dict
     return result
 
 
-def _update_state_for_uploaded(state: dict[str, Any], candidates: list[JsonlCandidate]) -> dict[str, Any]:
+def _update_state_for_uploaded(
+    state: dict[str, Any],
+    candidates: list[JsonlCandidate],
+    archive_name: str | None = None,
+    *,
+    archive_id: str | None = None,
+    archive_md5: str | None = None,
+    digests: dict[str, str] | None = None,
+) -> dict[str, Any]:
+    """Record which archive object carries each file, and the bytes it carried.
+
+    Without this provenance a later prune cannot know it is removing the last
+    surviving copy of a file, which is exactly how a covered file becomes
+    unrecoverable while state still reports it as backed up.
+    """
     files = dict(state.get("files") or {})
     for candidate in candidates:
-        files[candidate.path.as_posix()] = {"mtime": candidate.mtime, "size": candidate.size}
+        entry: dict[str, Any] = {"mtime": candidate.mtime, "size": candidate.size}
+        if archive_name and archive_id:
+            entry["archive"] = archive_name
+            entry["archive_id"] = archive_id
+            if archive_md5:
+                entry["archive_md5"] = archive_md5
+            digest = (digests or {}).get(candidate.path.as_posix())
+            entry["sha256"] = digest if digest else _sha256_file(candidate.path)
+        files[candidate.path.as_posix()] = entry
     return {"files": files, "updated_at": dt.datetime.now(dt.UTC).isoformat()}
 
 
@@ -390,11 +530,26 @@ def run_backup(
     state_path = Path(state_path).expanduser()
     state = _load_state(state_path)
     candidates = _discover_jsonl_candidates(roots)
-    changed, active, covered = _select_backup_candidates(
+    credentials = None
+    service = None
+    surviving_archives: dict[str, str | None] | None = None
+    if upload:
+        # Authenticate only when a listing is actually needed. When no entry claims
+        # archive-backed coverage there is nothing to verify, so a run that would be a
+        # clean no-op does not touch Drive. Once entries DO claim coverage the listing is
+        # unavoidable -- survival cannot be proven without asking.
+        if any(isinstance(e, dict) and e.get("archive_id") for e in (state.get("files") or {}).values()):
+            credentials = backup_daily.get_drive_credentials()
+            service = backup_daily.build_drive_service()
+            surviving_archives = _list_surviving_archives(service, folder_parts)
+        else:
+            surviving_archives = {}
+    changed, active, covered, vanished = _select_backup_candidates(
         candidates,
         state=state,
         now=now,
         active_skip_seconds=active_skip_seconds,
+        surviving_archives=surviving_archives,
     )
 
     if not changed:
@@ -405,13 +560,14 @@ def run_backup(
             "already_covered_files": covered,
             "discovered_file_count": len(candidates),
             "skipped_active_count": len(active),
+            "vanished_source_count": vanished,
             "message": f"no-op, {covered} files already covered",
         }
         _append_json_log(log_path, result)
         _enqueue_run_summary(result, queue_dir=queue_dir)
         return result
 
-    archive_path = create_jsonl_bundle(changed, staging_dir, date_stamp=date_stamp)
+    archive_path, bundle_digests = create_jsonl_bundle_with_digests(changed, staging_dir, date_stamp=date_stamp)
     archive_size = archive_path.stat().st_size
     result = {
         "status": "uploaded" if upload else "created",
@@ -422,15 +578,17 @@ def run_backup(
         "bundled_file_count": len(changed),
         "skipped_active_count": len(active),
         "already_covered_files": covered,
+        "vanished_source_count": vanished,
         "source_file_count": len(candidates),
         "retention_deleted": [],
         "forever_uploaded_file_count": 0,
         "forever_files": [],
     }
 
     if upload:
-        credentials = backup_daily.get_drive_credentials()
-        service = backup_daily.build_drive_service()
+        if service is None:
+            credentials = backup_daily.get_drive_credentials()
+            service = backup_daily.build_drive_service()
         folder_id = backup_daily.ensure_drive_folder_chain(service, folder_parts)
         uploaded = backup_daily.upload_file_to_drive_raw(archive_path, folder_id, credentials)
         file_id = uploaded.get("id")
@@ -446,7 +604,17 @@ def run_backup(
 
     result.update(verify_jsonl_bundle(archive_path, expected_file_count=len(changed)))
     if result["verified"] and upload:
-        _atomic_write_json(state_path, _update_state_for_uploaded(state, changed))
+        _atomic_write_json(
+            state_path,
+            _update_state_for_uploaded(
+                state,
+                changed,
+                archive_path.name,
+                archive_id=file_id,
+                archive_md5=uploaded.get("md5Checksum"),
+                digests=bundle_digests,
+            ),
+        )
         try:
             deleted = backup_daily.prune_drive_backups(
                 service,