From 4feec4efe81a7b02e8f00e478d8953cd53dd5ccc Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Tue, 8 Sep 2026 21:36:44 +0300 Subject: [PATCH 1/4] fix(backup): never treat a file as covered without a surviving archive object (S) P0 defect 2 (retention invariant). jsonl_backup treated a file as backed up when its mtime and size matched recorded state, then pruned old bundles under a 30-file policy that had no idea which files' last copy it was deleting. So a file's only surviving bundle could age out while state still reported it covered, and it was never re-bundled. Last real run: 29,542 already-covered, 110 bundled, zero forever uploads, and it deleted claude-jsonl-2026-07-23.tar.gz. Coverage now requires a SURVIVING archive object holding the exact bytes: - state records, per file, the archive object that carried it plus its sha256 - _state_matches additionally requires that archive to still be present and the recorded hash to still match the file on disk - run_backup lists the backup folder before selecting candidates, so an object pruned by us or out of band stops proving coverage - entries written before archive provenance existed name no archive, so they cannot prove survival and are deliberately treated as uncovered The listing is skipped when no state entry claims archive-backed coverage, so a fresh state costs no extra Drive call. RED first: test_pruned_bundle_uncovers_its_files_instead_of_orphaning_them fails on the old code with already_covered_files == 1 after the only bundle holding the file is pruned. test_run_jsonl_backup_second_run_noops_when_state_covers_files now models archive survival, proving the legitimate no-op still holds. Does NOT re-enable com.brainlayer.jsonl-backup; orc unloaded it and re-enabling is gated on reporting this receipt. Co-Authored-By: Claude Opus 5 (1M context) --- src/brainlayer/jsonl_backup.py | 101 +++++++++++++++++++++++++++--- tests/test_jsonl_backup.py | 110 +++++++++++++++++++++++++++++---- 2 files changed, 190 insertions(+), 21 deletions(-) diff --git a/src/brainlayer/jsonl_backup.py b/src/brainlayer/jsonl_backup.py index 6a897c47..59b5b01e 100644 --- a/src/brainlayer/jsonl_backup.py +++ b/src/brainlayer/jsonl_backup.py @@ -184,10 +184,33 @@ 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: set[str] | 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 = entry.get("archive") + if not isinstance(archive, str) or archive not in surviving_archives: + return False + recorded_hash = entry.get("sha256") + if not isinstance(recorded_hash, str) or not recorded_hash: + return False + return recorded_hash == _sha256_file(candidate.path) def _backup_unit_key(candidate: JsonlCandidate) -> tuple[int, str]: @@ -204,6 +227,7 @@ def _select_backup_candidates( state: dict[str, Any], now: float, active_skip_seconds: int, + surviving_archives: set[str] | None = None, ) -> tuple[list[JsonlCandidate], list[JsonlCandidate], int]: grouped: dict[tuple[int, str], list[JsonlCandidate]] = {} for candidate in candidates: @@ -217,13 +241,48 @@ def _select_backup_candidates( 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 +def _list_surviving_archive_names(service: Any, folder_parts: list[str]) -> set[str]: + """Names of archive objects still present in the backup folder. + + This is the ground truth for coverage: an object that has been pruned -- by our + own retention policy or out of band -- can no longer prove that a file survives. + """ + folder_id = backup_daily.ensure_drive_folder_chain(service, folder_parts) + names: set[str] = set() + 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)", + pageSize=1000, + pageToken=page_token, + supportsAllDrives=True, + ) + .execute() + ) + for item in response.get("files", []): + name = item.get("name") + if isinstance(name, str): + names.add(name) + page_token = response.get("nextPageToken") + if not page_token: + break + return names + + def _archive_name(candidate: JsonlCandidate) -> str: try: relative = candidate.path.relative_to(candidate.root) @@ -336,10 +395,24 @@ 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, +) -> 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: + entry["archive"] = archive_name + entry["sha256"] = _sha256_file(candidate.path) + files[candidate.path.as_posix()] = entry return {"files": files, "updated_at": dt.datetime.now(dt.UTC).isoformat()} @@ -390,11 +463,25 @@ def run_backup( state_path = Path(state_path).expanduser() state = _load_state(state_path) candidates = _discover_jsonl_candidates(roots) + credentials = None + service = None + surviving_archives: set[str] | None = None + if upload: + credentials = backup_daily.get_drive_credentials() + service = backup_daily.build_drive_service() + # Only consult the backup folder when some entry actually claims archive-backed + # coverage. A state with nothing to verify needs no listing, and legacy entries + # that name no archive stay uncovered either way. + if any(isinstance(entry, dict) and entry.get("archive") for entry in (state.get("files") or {}).values()): + surviving_archives = _list_surviving_archive_names(service, folder_parts) + else: + surviving_archives = set() changed, active, covered = _select_backup_candidates( candidates, state=state, now=now, active_skip_seconds=active_skip_seconds, + surviving_archives=surviving_archives, ) if not changed: @@ -429,8 +516,6 @@ def run_backup( } if upload: - 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 +531,7 @@ 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)) try: deleted = backup_daily.prune_drive_backups( service, diff --git a/tests/test_jsonl_backup.py b/tests/test_jsonl_backup.py index 37b190dc..bf311deb 100644 --- a/tests/test_jsonl_backup.py +++ b/tests/test_jsonl_backup.py @@ -486,26 +486,27 @@ 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[str] = [] 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) + surviving.append(path.name) + return {"id": f"drive-{len(uploads)}", "name": path.name, "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 +687,86 @@ def test_jsonl_backup_launchd_plist_and_docstring_install_note_are_committed(): assert "NumberOfFiles" in script_plist assert "4096" in plist assert "4096" in script_plist + + +def _drive_service_with_surviving(surviving: list[str]): + """Minimal fake Drive service whose folder listing reflects real survival.""" + + class _Files: + def list(self, **kwargs): + class _Req: + def execute(_self): + return {"files": [{"id": f"id-{n}", "name": n} for n in 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[str] = [] + + 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) + surviving.append(p.name) + return {"id": f"drive-{len(uploads)}", "name": p.name, "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 From 9a4792eea1dfa95c1e332e8385276e419e7f9b12 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Tue, 8 Sep 2026 21:46:55 +0300 Subject: [PATCH 2/4] fix(backup): prove survival by object identity and archived bytes (S) Macroscope review of #815 found three real weaknesses in the invariant plus one regression. All four addressed, each with a regression test that fails on the previous commit. Drive names are not unique within a folder, so name-based survival let a same-named replacement object impersonate the pruned original. Survival is now keyed by Drive object ID, which the upload response already returned and the previous commit simply did not persist. A surviving object could also have been rewritten out of band. State now records the archive's md5Checksum and the listing re-reads it; a mismatch, or an object Drive will not report a checksum for, reads as uncovered. Fail-closed: the cost is re-bundling, the cost of the other direction is the only remaining copy. The recorded sha256 came from re-reading the source AFTER bundling and uploading. A file rewritten inside that window, keeping mtime and size, got a digest for bytes the archive never contained -- so the next run matched that digest against the live file and called it covered. create_jsonl_bundle_with_digests now hashes the exact bytes it writes into the tar. create_jsonl_bundle stays as a path-only wrapper for existing callers. Regression, self-inflicted by the previous commit: with upload=True the run authenticated before selecting candidates, so a no-work run raised from get_drive_credentials() instead of returning its no-op. Credentials are now built only when a listing is actually needed. Once entries do claim archive coverage a listing is unavoidable -- survival cannot be proven without asking -- and that is stated rather than papered over. Co-Authored-By: Claude Opus 5 (1M context) --- src/brainlayer/jsonl_backup.py | 107 +++++++++++++++++------ tests/test_jsonl_backup.py | 155 +++++++++++++++++++++++++++++++-- 2 files changed, 224 insertions(+), 38 deletions(-) diff --git a/src/brainlayer/jsonl_backup.py b/src/brainlayer/jsonl_backup.py index 59b5b01e..e267071b 100644 --- a/src/brainlayer/jsonl_backup.py +++ b/src/brainlayer/jsonl_backup.py @@ -15,6 +15,7 @@ import datetime as dt import hashlib +import io import json import os import shutil @@ -187,7 +188,7 @@ def _discover_jsonl_candidates(source_roots: list[Path | BackupSourceRoot]) -> l def _state_matches( entry: Any, candidate: JsonlCandidate, - surviving_archives: set[str] | None = None, + surviving_archives: dict[str, str | None] | None = None, ) -> bool: """A file is covered only while a SURVIVING archive object holds its exact bytes. @@ -204,9 +205,16 @@ def _state_matches( return False if surviving_archives is None: return True - archive = entry.get("archive") - if not isinstance(archive, str) or archive not in surviving_archives: + 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 @@ -227,7 +235,7 @@ def _select_backup_candidates( state: dict[str, Any], now: float, active_skip_seconds: int, - surviving_archives: set[str] | None = None, + surviving_archives: dict[str, str | None] | None = None, ) -> tuple[list[JsonlCandidate], list[JsonlCandidate], int]: grouped: dict[tuple[int, str], list[JsonlCandidate]] = {} for candidate in candidates: @@ -251,14 +259,17 @@ def _select_backup_candidates( return changed, active, covered -def _list_surviving_archive_names(service: Any, folder_parts: list[str]) -> set[str]: - """Names of archive objects still present in the backup folder. +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. - This is the ground truth for coverage: an object that has been pruned -- by our - own retention policy or out of band -- can no longer prove that a file survives. + 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) - names: set[str] = set() + surviving: dict[str, str | None] = {} page_token = None while True: response = ( @@ -266,7 +277,7 @@ def _list_surviving_archive_names(service: Any, folder_parts: list[str]) -> set[ .list( q=f"'{folder_id}' in parents and trashed = false", spaces="drive", - fields="nextPageToken,files(id,name)", + fields="nextPageToken,files(id,name,md5Checksum)", pageSize=1000, pageToken=page_token, supportsAllDrives=True, @@ -274,13 +285,13 @@ def _list_surviving_archive_names(service: Any, folder_parts: list[str]) -> set[ .execute() ) for item in response.get("files", []): - name = item.get("name") - if isinstance(name, str): - names.add(name) + 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 names + return surviving def _archive_name(candidate: JsonlCandidate) -> str: @@ -351,7 +362,9 @@ def _upload_forever_files( return uploaded -def create_jsonl_bundle(candidates: list[JsonlCandidate], staging_dir: Path, *, date_stamp: str) -> Path: +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() @@ -361,13 +374,27 @@ 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. 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. + payload = candidate.path.read_bytes() + digests[candidate.path.as_posix()] = hashlib.sha256(payload).hexdigest() + info = tar.gettarinfo(str(candidate.path), arcname=_archive_name(candidate)) + info.size = len(payload) + tar.addfile(info, io.BytesIO(payload)) 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 @@ -399,6 +426,10 @@ 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. @@ -409,9 +440,13 @@ def _update_state_for_uploaded( files = dict(state.get("files") or {}) for candidate in candidates: entry: dict[str, Any] = {"mtime": candidate.mtime, "size": candidate.size} - if archive_name: + if archive_name and archive_id: entry["archive"] = archive_name - entry["sha256"] = _sha256_file(candidate.path) + 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()} @@ -465,17 +500,18 @@ def run_backup( candidates = _discover_jsonl_candidates(roots) credentials = None service = None - surviving_archives: set[str] | None = None + surviving_archives: dict[str, str | None] | None = None if upload: - credentials = backup_daily.get_drive_credentials() - service = backup_daily.build_drive_service() - # Only consult the backup folder when some entry actually claims archive-backed - # coverage. A state with nothing to verify needs no listing, and legacy entries - # that name no archive stay uncovered either way. - if any(isinstance(entry, dict) and entry.get("archive") for entry in (state.get("files") or {}).values()): - surviving_archives = _list_surviving_archive_names(service, folder_parts) + # 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 = set() + surviving_archives = {} changed, active, covered = _select_backup_candidates( candidates, state=state, @@ -498,7 +534,7 @@ def run_backup( _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", @@ -516,6 +552,9 @@ def run_backup( } if upload: + 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") @@ -531,7 +570,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, archive_path.name)) + _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 bf311deb..18587c0a 100644 --- a/tests/test_jsonl_backup.py +++ b/tests/test_jsonl_backup.py @@ -488,7 +488,7 @@ def test_run_jsonl_backup_second_run_noops_when_state_covers_files(tmp_path, mon 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[str] = [] + surviving: list[dict] = [] monkeypatch.setattr(jsonl_backup.backup_daily, "get_drive_credentials", lambda *args, **kwargs: object()) monkeypatch.setattr( @@ -503,8 +503,9 @@ def test_run_jsonl_backup_second_run_noops_when_state_covers_files(tmp_path, mon def _upload(file_path, folder_id, credentials): path = Path(file_path) uploads.append(path) - surviving.append(path.name) - return {"id": f"drive-{len(uploads)}", "name": path.name, "size": str(path.stat().st_size)} + 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) @@ -689,14 +690,18 @@ def test_jsonl_backup_launchd_plist_and_docstring_install_note_are_committed(): assert "4096" in script_plist -def _drive_service_with_surviving(surviving: list[str]): - """Minimal fake Drive service whose folder listing reflects real survival.""" +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": [{"id": f"id-{n}", "name": n} for n in surviving]} + return {"files": list(surviving)} return _Req() @@ -728,7 +733,7 @@ def test_pruned_bundle_uncovers_its_files_instead_of_orphaning_them(tmp_path, mo source_root = tmp_path / "sessions" _write_jsonl(source_root / "covered.jsonl", mtime=now - 3600) uploads: list[Path] = [] - surviving: list[str] = [] + surviving: list[dict] = [] monkeypatch.setattr(jsonl_backup.backup_daily, "get_drive_credentials", lambda *a, **k: object()) monkeypatch.setattr( @@ -741,8 +746,9 @@ def test_pruned_bundle_uncovers_its_files_instead_of_orphaning_them(tmp_path, mo def _upload(file_path, folder_id, credentials): p = Path(file_path) uploads.append(p) - surviving.append(p.name) - return {"id": f"drive-{len(uploads)}", "name": p.name, "size": str(p.stat().st_size)} + 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) @@ -770,3 +776,134 @@ def _upload(file_path, folder_id, credentials): ) 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" From 91d7ac511aa22b124f9bd48de843010dbad59ad4 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Tue, 8 Sep 2026 21:49:56 +0300 Subject: [PATCH 3/4] perf(backup): stream the bundle digest instead of buffering whole files (XS) Self-caught, not review-caught. Fixing Macroscope's same-bytes finding by reading each source into memory traded one defect for a memory spike: the largest real source JSONL across the backup roots is 374.5MB (measured, 13,570 files), and this job runs nightly at Nice=15 alongside everything else. _HashingReader digests the very stream tarfile consumes, so the same-bytes guarantee is unchanged while memory drops to O(buffer). Verified on a 115.5MB file: the recorded digest equals sha256 of the source, the bytes extracted back out of the archive hash to that same digest, and peak RSS moved 0.2MB instead of ~116MB. Co-Authored-By: Claude Opus 5 (1M context) --- src/brainlayer/jsonl_backup.py | 38 +++++++++++++++++++++++++++------- 1 file changed, 30 insertions(+), 8 deletions(-) diff --git a/src/brainlayer/jsonl_backup.py b/src/brainlayer/jsonl_backup.py index e267071b..aa55586c 100644 --- a/src/brainlayer/jsonl_backup.py +++ b/src/brainlayer/jsonl_backup.py @@ -15,7 +15,6 @@ import datetime as dt import hashlib -import io import json import os import shutil @@ -362,6 +361,27 @@ def _upload_forever_files( return uploaded +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]]: @@ -378,14 +398,16 @@ def create_jsonl_bundle_with_digests( try: with tarfile.open(temp_path, "w:gz") as tar: for candidate in candidates: - # Hash and archive the SAME bytes. 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. - payload = candidate.path.read_bytes() - digests[candidate.path.as_posix()] = hashlib.sha256(payload).hexdigest() + # 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)) - info.size = len(payload) - tar.addfile(info, io.BytesIO(payload)) + 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) From 9ba55412dcadaff1f667c3ebae29098e2cdfb6eb Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Tue, 8 Sep 2026 22:34:49 +0300 Subject: [PATCH 4/4] fix(backup): request md5Checksum and survive a vanishing source (S) Pair review of #815 found the integrity check could never fire in production. The resumable upload requested `fields=id,name,size`, so `md5Checksum` was never in the response, `archive_md5` was never recorded, and the md5 comparison added one commit earlier was DEAD CODE. Worse, its regression test passed only because the test's own fake `_upload` injected an md5 that Drive would never return. Mock-green, not live-green, in the PR whose entire purpose is proving that an archived copy really exists. Drive is now asked for md5Checksum, and test_upload_actually_requests_md5checksum_from_drive pins that request so a fake can never silently diverge from production again. Second finding: selection never read source bytes before this PR, so verifying a recorded digest made an unreadable path able to abort the whole nightly run. A transcript that disappears between discovery and hashing is now dropped from the run and counted in `vanished_source_count`, never treated as covered. Failing the entire backup because one file vanished is the wrong failure for the job that exists to prevent data loss. Both regressions verified against 91d7ac51: the md5 test fails there, and the vanish test raises FileNotFoundError there. The first version of the vanish test deleted the file BEFORE the run, so discovery never returned it and the test passed on the broken code; it now unlinks after discovery, inside the real window. Co-Authored-By: Claude Opus 5 (1M context) --- src/brainlayer/backup_daily.py | 5 ++- src/brainlayer/jsonl_backup.py | 22 +++++++++--- tests/test_jsonl_backup.py | 61 ++++++++++++++++++++++++++++++++++ 3 files changed, 82 insertions(+), 6 deletions(-) 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 aa55586c..b757f37d 100644 --- a/src/brainlayer/jsonl_backup.py +++ b/src/brainlayer/jsonl_backup.py @@ -217,7 +217,14 @@ def _state_matches( recorded_hash = entry.get("sha256") if not isinstance(recorded_hash, str) or not recorded_hash: return False - return recorded_hash == _sha256_file(candidate.path) + 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]: @@ -235,7 +242,7 @@ def _select_backup_candidates( now: float, active_skip_seconds: int, surviving_archives: dict[str, str | None] | None = None, -) -> tuple[list[JsonlCandidate], list[JsonlCandidate], int]: +) -> 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) @@ -243,6 +250,7 @@ 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): @@ -254,8 +262,10 @@ def _select_backup_candidates( ): 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]: @@ -534,7 +544,7 @@ def run_backup( surviving_archives = _list_surviving_archives(service, folder_parts) else: surviving_archives = {} - changed, active, covered = _select_backup_candidates( + changed, active, covered, vanished = _select_backup_candidates( candidates, state=state, now=now, @@ -550,6 +560,7 @@ 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) @@ -567,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, diff --git a/tests/test_jsonl_backup.py b/tests/test_jsonl_backup.py index 18587c0a..72df1d63 100644 --- a/tests/test_jsonl_backup.py +++ b/tests/test_jsonl_backup.py @@ -907,3 +907,64 @@ def _upload(file_path, folder_id, credentials): 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()