diff --git a/src/brainlayer/backup_daily.py b/src/brainlayer/backup_daily.py
index ec7b35e3..f6a35c3f 100644
--- a/src/brainlayer/backup_daily.py
+++ b/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",
diff --git a/src/brainlayer/jsonl_backup.py b/src/brainlayer/jsonl_backup.py
index 6a897c47..b757f37d 100644
--- a/src/brainlayer/jsonl_backup.py
+++ b/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,7 +241,8 @@ 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)
@@ -212,16 +250,57 @@ def _select_backup_candidates(
     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,6 +578,7 @@ 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,
@@ -429,8 +586,9 @@ def run_backup(
     }
 
     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,
diff --git a/tests/test_jsonl_backup.py b/tests/test_jsonl_backup.py
index 37b190dc..72df1d63 100644
--- a/tests/test_jsonl_backup.py
+++ b/tests/test_jsonl_backup.py
@@ -486,26 +486,28 @@ def test_run_jsonl_backup_second_run_noops_when_state_covers_files(tmp_path, mon
     source_root = tmp_path / "sessions"
     _write_jsonl(source_root / "covered.jsonl", mtime=now - 3600)
     uploads: list[Path] = []
+    # The bundle uploaded by the first run is still present for the second run, so the
+    # no-op is legitimate: a surviving archive object really does hold these bytes.
+    surviving: list[dict] = []
 
     monkeypatch.setattr(jsonl_backup.backup_daily, "get_drive_credentials", lambda *args, **kwargs: object())
-    monkeypatch.setattr(jsonl_backup.backup_daily, "build_drive_service", lambda *args, **kwargs: object())
+    monkeypatch.setattr(
+        jsonl_backup.backup_daily, "build_drive_service", lambda *a, **k: _drive_service_with_surviving(surviving)
+    )
     monkeypatch.setattr(
         jsonl_backup.backup_daily, "ensure_drive_folder_chain", lambda service, folder_parts: "folder-id"
     )
     monkeypatch.setattr(jsonl_backup.backup_daily, "verify_drive_upload", lambda *args, **kwargs: None)
     monkeypatch.setattr(jsonl_backup.backup_daily, "prune_drive_backups", lambda *args, **kwargs: [])
-    monkeypatch.setattr(
-        jsonl_backup.backup_daily,
-        "upload_file_to_drive_raw",
-        lambda file_path, folder_id, credentials: (
-            uploads.append(Path(file_path))  # noqa: ARG005
-            or {
-                "id": f"drive-{len(uploads)}",
-                "name": Path(file_path).name,
-                "size": str(Path(file_path).stat().st_size),
-            }
-        ),
-    )
+
+    def _upload(file_path, folder_id, credentials):
+        path = Path(file_path)
+        uploads.append(path)
+        obj = {"id": f"drive-{len(uploads)}", "name": path.name, "md5Checksum": f"md5-{len(uploads)}"}
+        surviving.append(obj)
+        return {**obj, "size": str(path.stat().st_size)}
+
+    monkeypatch.setattr(jsonl_backup.backup_daily, "upload_file_to_drive_raw", _upload)
 
     kwargs = {
         "source_roots": [source_root],
@@ -686,3 +688,283 @@ def test_jsonl_backup_launchd_plist_and_docstring_install_note_are_committed():
     assert "<key>NumberOfFiles</key>" in script_plist
     assert "<integer>4096</integer>" in plist
     assert "<integer>4096</integer>" in script_plist
+
+
+def _drive_service_with_surviving(surviving: list[dict]):
+    """Minimal fake Drive service whose folder listing reflects real survival.
+
+    Entries are the objects themselves ({"id", "name", "md5Checksum"}), because Drive
+    identity is the object ID -- names are not unique within a folder.
+    """
+
+    class _Files:
+        def list(self, **kwargs):
+            class _Req:
+                def execute(_self):
+                    return {"files": list(surviving)}
+
+            return _Req()
+
+        def delete(self, **kwargs):
+            class _Req:
+                def execute(_self):
+                    return {}
+
+            return _Req()
+
+    class _Service:
+        def files(self):
+            return _Files()
+
+    return _Service()
+
+
+def test_pruned_bundle_uncovers_its_files_instead_of_orphaning_them(tmp_path, monkeypatch):
+    """P0 retention invariant (defect 2).
+
+    A file is 'covered' only while a SURVIVING archive object holds its exact bytes.
+    Once the bundle that carried it is pruned, the file must be re-bundled -- otherwise
+    its last copy ages out under the 30-file policy while state still claims coverage,
+    and the raw transcript is gone for good.
+    """
+    from brainlayer import jsonl_backup
+
+    now = time.time()
+    source_root = tmp_path / "sessions"
+    _write_jsonl(source_root / "covered.jsonl", mtime=now - 3600)
+    uploads: list[Path] = []
+    surviving: list[dict] = []
+
+    monkeypatch.setattr(jsonl_backup.backup_daily, "get_drive_credentials", lambda *a, **k: object())
+    monkeypatch.setattr(
+        jsonl_backup.backup_daily, "build_drive_service", lambda *a, **k: _drive_service_with_surviving(surviving)
+    )
+    monkeypatch.setattr(jsonl_backup.backup_daily, "ensure_drive_folder_chain", lambda service, folder_parts: "fid")
+    monkeypatch.setattr(jsonl_backup.backup_daily, "verify_drive_upload", lambda *a, **k: None)
+    monkeypatch.setattr(jsonl_backup.backup_daily, "prune_drive_backups", lambda *a, **k: [])
+
+    def _upload(file_path, folder_id, credentials):
+        p = Path(file_path)
+        uploads.append(p)
+        obj = {"id": f"drive-{len(uploads)}", "name": p.name, "md5Checksum": f"md5-{len(uploads)}"}
+        surviving.append(obj)
+        return {**obj, "size": str(p.stat().st_size)}
+
+    monkeypatch.setattr(jsonl_backup.backup_daily, "upload_file_to_drive_raw", _upload)
+
+    kwargs = {
+        "source_roots": [source_root],
+        "state_path": tmp_path / "state.json",
+        "staging_dir": tmp_path / "staging",
+        "log_path": tmp_path / "jsonl-backup.log",
+        "queue_dir": tmp_path / "queue",
+        "now": now,
+        "upload": True,
+    }
+    first = jsonl_backup.run_backup(date_stamp="2026-06-05", **kwargs)
+    assert first["status"] == "uploaded"
+    assert len(uploads) == 1
+
+    # Retention prunes the only bundle holding covered.jsonl. The source file is untouched,
+    # so mtime/size still "match" -- but no surviving object holds its bytes any more.
+    surviving.clear()
+
+    second = jsonl_backup.run_backup(date_stamp="2026-06-06", **kwargs)
+
+    assert second["already_covered_files"] == 0, (
+        "file whose only archive object was pruned must NOT be reported as covered"
+    )
+    assert second["status"] == "uploaded", "an orphaned file must be re-bundled, not skipped as a no-op"
+    assert second["bundled_file_count"] == 1
+
+
+def _covered_state_kwargs(tmp_path, source_root, now):
+    return {
+        "source_roots": [source_root],
+        "state_path": tmp_path / "state.json",
+        "staging_dir": tmp_path / "staging",
+        "log_path": tmp_path / "jsonl-backup.log",
+        "queue_dir": tmp_path / "queue",
+        "now": now,
+        "upload": True,
+    }
+
+
+def _install_drive(monkeypatch, jsonl_backup, uploads, surviving):
+    monkeypatch.setattr(jsonl_backup.backup_daily, "get_drive_credentials", lambda *a, **k: object())
+    monkeypatch.setattr(
+        jsonl_backup.backup_daily, "build_drive_service", lambda *a, **k: _drive_service_with_surviving(surviving)
+    )
+    monkeypatch.setattr(jsonl_backup.backup_daily, "ensure_drive_folder_chain", lambda service, folder_parts: "fid")
+    monkeypatch.setattr(jsonl_backup.backup_daily, "verify_drive_upload", lambda *a, **k: None)
+    monkeypatch.setattr(jsonl_backup.backup_daily, "prune_drive_backups", lambda *a, **k: [])
+
+    def _upload(file_path, folder_id, credentials):
+        path = Path(file_path)
+        uploads.append(path)
+        obj = {"id": f"drive-{len(uploads)}", "name": path.name, "md5Checksum": f"md5-{len(uploads)}"}
+        surviving.append(obj)
+        return {**obj, "size": str(path.stat().st_size)}
+
+    monkeypatch.setattr(jsonl_backup.backup_daily, "upload_file_to_drive_raw", _upload)
+
+
+def test_same_named_replacement_object_does_not_prove_survival(tmp_path, monkeypatch):
+    """Drive names are not unique in a folder, so identity must be the object ID.
+
+    A same-named object standing where the original was pruned must not be able to
+    impersonate it and vouch for files it never contained.
+    """
+    from brainlayer import jsonl_backup
+
+    now = time.time()
+    source_root = tmp_path / "sessions"
+    _write_jsonl(source_root / "covered.jsonl", mtime=now - 3600)
+    uploads: list[Path] = []
+    surviving: list[dict] = []
+    _install_drive(monkeypatch, jsonl_backup, uploads, surviving)
+    kwargs = _covered_state_kwargs(tmp_path, source_root, now)
+
+    first = jsonl_backup.run_backup(date_stamp="2026-06-05", **kwargs)
+    assert first["status"] == "uploaded"
+    original_name = surviving[0]["name"]
+
+    # Original object pruned; a DIFFERENT object with the same name remains.
+    surviving.clear()
+    surviving.append({"id": "some-other-object", "name": original_name, "md5Checksum": "md5-1"})
+
+    second = jsonl_backup.run_backup(date_stamp="2026-06-06", **kwargs)
+    assert second["already_covered_files"] == 0
+    assert second["status"] == "uploaded"
+
+
+def test_modified_surviving_object_does_not_prove_survival(tmp_path, monkeypatch):
+    """A surviving object whose bytes changed out of band cannot vouch for coverage."""
+    from brainlayer import jsonl_backup
+
+    now = time.time()
+    source_root = tmp_path / "sessions"
+    _write_jsonl(source_root / "covered.jsonl", mtime=now - 3600)
+    uploads: list[Path] = []
+    surviving: list[dict] = []
+    _install_drive(monkeypatch, jsonl_backup, uploads, surviving)
+    kwargs = _covered_state_kwargs(tmp_path, source_root, now)
+
+    jsonl_backup.run_backup(date_stamp="2026-06-05", **kwargs)
+    surviving[0]["md5Checksum"] = "tampered-or-rewritten"
+
+    second = jsonl_backup.run_backup(date_stamp="2026-06-06", **kwargs)
+    assert second["already_covered_files"] == 0
+    assert second["status"] == "uploaded"
+
+
+def test_recorded_digest_describes_the_bundled_bytes_not_a_later_read(tmp_path, monkeypatch):
+    """The digest must describe what went INTO the archive, not a later re-read.
+
+    The window is real: create_jsonl_bundle reads the file, then the state write used to
+    re-read it. A source rewritten inside that window -- same mtime, same size -- gets a
+    recorded digest for bytes the archive does not contain, so the NEXT run matches that
+    digest against the live file, calls it covered, and the archived version is the one
+    nobody can recover.
+    """
+    from brainlayer import jsonl_backup
+
+    now = time.time()
+    source_root = tmp_path / "sessions"
+    target = _write_jsonl(source_root / "covered.jsonl", line='{"v":"aaa"}\n', mtime=now - 3600)
+    uploads: list[Path] = []
+    surviving: list[dict] = []
+
+    monkeypatch.setattr(jsonl_backup.backup_daily, "get_drive_credentials", lambda *a, **k: object())
+    monkeypatch.setattr(
+        jsonl_backup.backup_daily, "build_drive_service", lambda *a, **k: _drive_service_with_surviving(surviving)
+    )
+    monkeypatch.setattr(jsonl_backup.backup_daily, "ensure_drive_folder_chain", lambda service, folder_parts: "fid")
+    monkeypatch.setattr(jsonl_backup.backup_daily, "verify_drive_upload", lambda *a, **k: None)
+    monkeypatch.setattr(jsonl_backup.backup_daily, "prune_drive_backups", lambda *a, **k: [])
+
+    def _upload(file_path, folder_id, credentials):
+        path = Path(file_path)
+        uploads.append(path)
+        # Runs AFTER the bundle was built and BEFORE state is written: rewrite the source
+        # with identical length and restore its mtime, exactly the window the old code lost.
+        target.write_text('{"v":"bbb"}\n', encoding="utf-8")
+        os.utime(target, (now - 3600, now - 3600))
+        obj = {"id": f"drive-{len(uploads)}", "name": path.name, "md5Checksum": f"md5-{len(uploads)}"}
+        surviving.append(obj)
+        return {**obj, "size": str(path.stat().st_size)}
+
+    monkeypatch.setattr(jsonl_backup.backup_daily, "upload_file_to_drive_raw", _upload)
+    kwargs = _covered_state_kwargs(tmp_path, source_root, now)
+
+    jsonl_backup.run_backup(date_stamp="2026-06-05", **kwargs)
+    state = json.loads((tmp_path / "state.json").read_text())
+    recorded = state["files"][target.as_posix()]["sha256"]
+    assert recorded == hashlib.sha256(b'{"v":"aaa"}\n').hexdigest(), (
+        "digest must be of the bytes placed in the archive, not of a later re-read"
+    )
+
+    second = jsonl_backup.run_backup(date_stamp="2026-06-06", **kwargs)
+    assert second["already_covered_files"] == 0, "the unarchived rewrite must not read as covered"
+    assert second["status"] == "uploaded"
+
+
+def test_upload_actually_requests_md5checksum_from_drive():
+    """The integrity branch is only real if Drive is ASKED for md5Checksum.
+
+    PR #815's first attempt shipped an md5 comparison that could never fire: the resumable
+    upload requested `fields=id,name,size`, so `md5Checksum` was always absent, `archive_md5`
+    was never recorded, and the branch was dead in production. The regression test for it
+    passed only because the fake `_upload` injected an md5 Drive would never return —
+    mock-green, not live-green. This pins the real request so a fake can never diverge
+    from production again.
+    """
+    import inspect
+
+    from brainlayer import backup_daily
+
+    source = inspect.getsource(backup_daily.upload_file_to_drive_raw)
+    assert "md5Checksum" in source, (
+        "the resumable upload must request md5Checksum, or retention's integrity check is dead code"
+    )
+
+
+def test_vanished_source_does_not_abort_the_nightly_run(tmp_path, monkeypatch):
+    """A file deleted under us mid-run must not take the whole backup down.
+
+    Selection never read source bytes before this PR; it does now, so a path that disappears
+    BETWEEN discovery and hashing became able to kill the run. Deleting it before the run
+    proves nothing — discovery simply would not find it. The window is the race, so the test
+    unlinks the file after discovery has already returned it as a candidate.
+
+    This job is what PREVENTS data loss; failing the entire nightly backup because one
+    transcript vanished is the wrong failure.
+    """
+    from brainlayer import jsonl_backup
+
+    now = time.time()
+    source_root = tmp_path / "sessions"
+    keeper = _write_jsonl(source_root / "keeper.jsonl", mtime=now - 3600)
+    doomed = _write_jsonl(source_root / "doomed.jsonl", mtime=now - 3600)
+    uploads: list[Path] = []
+    surviving: list[dict] = []
+    _install_drive(monkeypatch, jsonl_backup, uploads, surviving)
+    kwargs = _covered_state_kwargs(tmp_path, source_root, now)
+
+    first = jsonl_backup.run_backup(date_stamp="2026-06-05", **kwargs)
+    assert first["bundled_file_count"] == 2
+
+    real_discover = jsonl_backup._discover_jsonl_candidates
+
+    def _discover_then_vanish(roots):
+        found = real_discover(roots)
+        doomed.unlink(missing_ok=True)  # gone after discovery, before coverage hashing
+        return found
+
+    monkeypatch.setattr(jsonl_backup, "_discover_jsonl_candidates", _discover_then_vanish)
+
+    second = jsonl_backup.run_backup(date_stamp="2026-06-06", **kwargs)
+    assert second["status"] == "no-op", "the surviving file was still covered; the run must not die"
+    assert second["already_covered_files"] == 1
+    assert second["vanished_source_count"] == 1
+    assert keeper.exists()