diff --git a/docs/data-locations.md b/docs/data-locations.md index 1cd5c6a3..4482ffaf 100644 --- a/docs/data-locations.md +++ b/docs/data-locations.md @@ -72,6 +72,14 @@ BrainLayer resolves the database path in this order (see `src/brainlayer/paths.p } ``` +## Natural-state transcript backup contract + +Each `claude-jsonl-YYYY-MM-DD.tar.gz` bundle stores every selected transcript as a regular raw file +under `source-{root-index}/` outside the BrainLayer database, and verification +must stream-compare every tar member byte-for-byte with the source before the bundle counts as a +natural-state copy; an exact match or a strict prefix of a source that only grew after bundling is +valid, while any divergence within the archived bytes fails the entire bundle before upload. + ## Backups (Manual) Before any bulk operation, back up the database: diff --git a/src/brainlayer/jsonl_backup.py b/src/brainlayer/jsonl_backup.py index 54c792ac..077d7abb 100644 --- a/src/brainlayer/jsonl_backup.py +++ b/src/brainlayer/jsonl_backup.py @@ -14,6 +14,8 @@ from __future__ import annotations import datetime as dt +import fcntl +import functools import hashlib import json import os @@ -394,7 +396,7 @@ def hexdigest(self) -> str: def create_jsonl_bundle_with_digests( candidates: list[JsonlCandidate], staging_dir: Path, *, date_stamp: str -) -> tuple[Path, dict[str, str]]: +) -> tuple[Path, dict[str, str], dict[str, int]]: if not candidates: raise ValueError("create_jsonl_bundle requires at least one candidate") staging_dir = Path(staging_dir).expanduser() @@ -405,6 +407,7 @@ def create_jsonl_bundle_with_digests( ) as tmp: temp_path = Path(tmp.name) digests: dict[str, str] = {} + sizes: dict[str, int] = {} try: with tarfile.open(temp_path, "w:gz") as tar: for candidate in candidates: @@ -413,28 +416,70 @@ def create_jsonl_bundle_with_digests( # 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: + # Discovery follows file symlinks, so archive their resolved target as a + # regular member too; preserving the link would make verification reject + # the whole bundle as non-regular. + source_path = candidate.path.resolve() + info = tar.gettarinfo(str(source_path), arcname=_archive_name(candidate)) + with source_path.open("rb") as handle: reader = _HashingReader(handle) tar.addfile(info, reader) digests[candidate.path.as_posix()] = reader.hexdigest() + sizes[candidate.path.as_posix()] = info.size os.replace(temp_path, archive_path) finally: temp_path.unlink(missing_ok=True) - return archive_path, digests + return archive_path, digests, sizes 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) + archive_path, _, _ = create_jsonl_bundle_with_digests(candidates, staging_dir, date_stamp=date_stamp) return archive_path -def verify_jsonl_bundle(archive_path: Path, *, expected_file_count: int) -> dict[str, Any]: +def _compare_member_to_source(extracted: Any, source_path: Path, *, bundle_digest: str | None = None) -> str: + """Compare archived bytes with the live source, or their bundle-time digest if it vanished.""" + try: + source = source_path.open("rb") + except FileNotFoundError: + if bundle_digest is None: + raise + digest = hashlib.sha256() + while member_chunk := extracted.read(1024 * 1024): + digest.update(member_chunk) + return "vanished" if digest.hexdigest() == bundle_digest else "diverged" + with source: + while member_chunk := extracted.read(1024 * 1024): + if source.read(len(member_chunk)) != member_chunk: + return "diverged" + return "append_snapshot" if source.read(1) else "exact" + + +def verify_jsonl_bundle( + archive_path: Path, + *, + expected_file_count: int | None = None, + expected_candidates: list[JsonlCandidate] | None = None, + expected_digests: dict[str, str] | None = None, + expected_sizes: dict[str, int] | None = None, +) -> dict[str, Any]: + """Verify a bundle; production callers should supply expected_candidates for content proof.""" + if expected_candidates is not None: + candidate_count = len(expected_candidates) + if expected_file_count is not None and expected_file_count != candidate_count: + raise ValueError("expected_file_count does not match expected_candidates") + expected_file_count = candidate_count + if expected_file_count is None: + raise ValueError("expected_file_count or expected_candidates is required") + result: dict[str, Any] = { "verified": False, "bundled_file_count": expected_file_count, "archive_listing_count": 0, + "content_verified_file_count": 0, + "append_snapshot_file_count": 0, + "vanished_after_bundle_file_count": 0, } try: subprocess.run(["gunzip", "-t", str(archive_path)], check=True, capture_output=True, text=True) @@ -447,7 +492,49 @@ def verify_jsonl_bundle(archive_path: Path, *, expected_file_count: int) -> dict f"tar listing count mismatch: expected={expected_file_count} actual={len(entries)}" ) return result + if expected_candidates is not None: + expected_by_name = {_archive_name(candidate): candidate for candidate in expected_candidates} + if len(expected_by_name) != len(expected_candidates): + result["verification_error"] = "duplicate archive member name generated for source files" + return result + with tarfile.open(archive_path, "r:gz") as archive: + members = [member for member in archive.getmembers() if not member.isdir()] + actual_names = [member.name for member in members] + if len(actual_names) != len(set(actual_names)): + result["verification_error"] = "archive contains duplicate member names" + return result + if set(actual_names) != set(expected_by_name): + result["verification_error"] = "archive member names do not match source files" + return result + for member in members: + if not member.isfile(): + result["verification_error"] = f"archive member is not a regular file: {member.name}" + return result + candidate = expected_by_name[member.name] + expected_size = (expected_sizes or {}).get(candidate.path.as_posix(), candidate.size) + if member.size != expected_size: + result["verification_error"] = f"archive member size differs from bundle: {member.name}" + return result + extracted = archive.extractfile(member) + if extracted is None: + result["verification_error"] = f"archive member cannot be read: {member.name}" + return result + comparison = _compare_member_to_source( + extracted, + candidate.path, + bundle_digest=(expected_digests or {}).get(candidate.path.as_posix()), + ) + if comparison == "diverged": + result["verification_error"] = f"archive member differs from source bytes: {member.name}" + return result + if comparison == "append_snapshot": + result["append_snapshot_file_count"] += 1 + elif comparison == "vanished": + result["vanished_after_bundle_file_count"] += 1 + result["content_verified_file_count"] += 1 result["verified"] = True + except backup_daily.BackupTimeoutError: + raise except Exception as exc: result.setdefault("gzip_test", False) result["verification_error"] = str(exc) @@ -510,6 +597,22 @@ def _enqueue_run_summary(result: dict[str, Any], *, queue_dir: Path | None) -> N ) +def _serialized_by_staging_dir(function): + """Serialize discovery through state persistence for one staging directory.""" + + @functools.wraps(function) + def wrapper(*args, **kwargs): + staging_dir = Path(kwargs.get("staging_dir", DEFAULT_STAGING_DIR)).expanduser() + staging_dir.mkdir(parents=True, exist_ok=True) + lock_path = staging_dir / ".jsonl-backup.lock" + with lock_path.open("a") as lock_file: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + return function(*args, **kwargs) + + return wrapper + + +@_serialized_by_staging_dir def run_backup( *, source_roots: list[Path | BackupSourceRoot] | None = None, @@ -574,11 +677,13 @@ def run_backup( _enqueue_run_summary(result, queue_dir=queue_dir) return result - archive_path, bundle_digests = create_jsonl_bundle_with_digests(changed, staging_dir, date_stamp=date_stamp) + archive_path, bundle_digests, bundle_sizes = create_jsonl_bundle_with_digests( + changed, staging_dir, date_stamp=date_stamp + ) archive_size = archive_path.stat().st_size result = { "attempted_at": attempted_at, - "status": "uploaded" if upload else "created", + "status": "created", "archive": str(archive_path), "bytes": archive_size, "uploaded": False, @@ -593,7 +698,15 @@ def run_backup( "forever_files": [], } - if upload: + result.update( + verify_jsonl_bundle( + archive_path, + expected_candidates=changed, + expected_digests=bundle_digests, + expected_sizes=bundle_sizes, + ) + ) + if result["verified"] and upload: if service is None: credentials = backup_daily.get_drive_credentials() service = backup_daily.build_drive_service() @@ -608,14 +721,7 @@ def run_backup( expected_name=archive_path.name, expected_size=archive_size, ) - result.update({"uploaded": True, "drive_file": uploaded}) - - result.update(verify_jsonl_bundle(archive_path, expected_file_count=len(changed))) - if result["verified"] and upload: - # The same incident was two individually reasonable deletions composed together: - # successful upload removed local staging, then Drive retention removed the remote - # bundle. Persist the exact Drive object and archived-source digests before either - # deletion path runs so the next selection cannot silently trust the dead copy. + result.update({"status": "uploaded", "uploaded": True, "drive_file": uploaded}) _atomic_write_json( state_path, _update_state_for_uploaded( @@ -652,6 +758,12 @@ def run_backup( # invariant rather than an optional integrity check (2026-09-09 / PR #815). archive_path.unlink(missing_ok=True) result["local_archive_removed"] = True + elif not result["verified"]: + result["status"] = "failed" + result["message"] = f"local bundle verification failed: {result.get('verification_error', 'unknown error')}" + _append_json_log(log_path, result) + _enqueue_run_summary(result, queue_dir=queue_dir) + return result _append_json_log(log_path, result) _enqueue_run_summary(result, queue_dir=queue_dir) diff --git a/tests/test_jsonl_backup.py b/tests/test_jsonl_backup.py index 2b87fab4..27206b8a 100644 --- a/tests/test_jsonl_backup.py +++ b/tests/test_jsonl_backup.py @@ -1,7 +1,9 @@ import hashlib +import io import json import os import tarfile +import threading import time from pathlib import Path @@ -61,6 +63,353 @@ def test_jsonl_retention_invariant_is_a_ci_guard_not_only_a_behavior_fixture(): assert expected_error in inspect_jsonl_retention_invariant(unsafe) +def test_jsonl_bundle_round_trips_fixture_byte_identical(tmp_path): + from brainlayer import jsonl_backup + + source_root = tmp_path / "sessions" + source = source_root / "nested" / "session.jsonl" + original = '{"type":"user","message":"raw\\r\\ntext שלום"}\r\n'.encode() + source.parent.mkdir(parents=True) + source.write_bytes(original) + candidates = jsonl_backup._discover_jsonl_candidates([source_root]) + + archive = jsonl_backup.create_jsonl_bundle(candidates, tmp_path / "staging", date_stamp="2026-09-09") + + with tarfile.open(archive, "r:gz") as bundle: + extracted = bundle.extractfile("source-0/nested/session.jsonl") + assert extracted is not None + assert extracted.read() == original + + verification = jsonl_backup.verify_jsonl_bundle(archive, expected_candidates=candidates) + assert verification["verified"] is True + assert verification["content_verified_file_count"] == 1 + + +def test_jsonl_bundle_accepts_source_growth_after_discovery_before_bundling(tmp_path): + from brainlayer import jsonl_backup + + source_root = tmp_path / "sessions" + source = source_root / "session.jsonl" + source.parent.mkdir(parents=True) + source.write_bytes(b'{"first":true}\n') + candidates = jsonl_backup._discover_jsonl_candidates([source_root]) + + with source.open("ab") as handle: + handle.write(b'{"second":true}\n') + archive, digests, sizes = jsonl_backup.create_jsonl_bundle_with_digests( + candidates, tmp_path / "staging", date_stamp="2026-09-09" + ) + + verification = jsonl_backup.verify_jsonl_bundle( + archive, + expected_candidates=candidates, + expected_digests=digests, + expected_sizes=sizes, + ) + + assert verification["verified"] is True + assert verification["content_verified_file_count"] == 1 + assert verification["append_snapshot_file_count"] == 0 + + +def test_jsonl_bundle_uses_bundle_digest_when_source_vanishes_before_verification(tmp_path): + from brainlayer import jsonl_backup + + source_root = tmp_path / "sessions" + source = source_root / "session.jsonl" + source.parent.mkdir(parents=True) + source.write_bytes(b'{"archived":true}\n') + candidates = jsonl_backup._discover_jsonl_candidates([source_root]) + archive, digests, sizes = jsonl_backup.create_jsonl_bundle_with_digests( + candidates, tmp_path / "staging", date_stamp="2026-09-09" + ) + source.unlink() + + verification = jsonl_backup.verify_jsonl_bundle( + archive, + expected_candidates=candidates, + expected_digests=digests, + expected_sizes=sizes, + ) + + assert verification["verified"] is True + assert verification["content_verified_file_count"] == 1 + assert verification["vanished_after_bundle_file_count"] == 1 + + +def test_jsonl_bundle_rejects_digest_mismatch_when_source_vanishes(tmp_path): + from brainlayer import jsonl_backup + + source_root = tmp_path / "sessions" + source = source_root / "session.jsonl" + source.parent.mkdir(parents=True) + original = b'{"archived":true}\n' + source.write_bytes(original) + candidates = jsonl_backup._discover_jsonl_candidates([source_root]) + _, digests, sizes = jsonl_backup.create_jsonl_bundle_with_digests( + candidates, tmp_path / "staging", date_stamp="2026-09-09" + ) + source.unlink() + archive = tmp_path / "changed.tar.gz" + changed = b'{"archived":null}\n' + member = tarfile.TarInfo("source-0/session.jsonl") + member.size = len(changed) + with tarfile.open(archive, "w:gz") as bundle: + bundle.addfile(member, io.BytesIO(changed)) + + verification = jsonl_backup.verify_jsonl_bundle( + archive, + expected_candidates=candidates, + expected_digests=digests, + expected_sizes=sizes, + ) + + assert verification["verified"] is False + assert verification["verification_error"] == "archive member differs from source bytes: source-0/session.jsonl" + + +def test_jsonl_bundle_dereferences_discovered_symlink_as_regular_file(tmp_path): + from brainlayer import jsonl_backup + + source_root = tmp_path / "sessions" + target = tmp_path / "source.jsonl" + target.write_bytes(b'{"through":"symlink"}\n') + link = source_root / "session.jsonl" + link.parent.mkdir(parents=True) + link.symlink_to(target) + candidates = jsonl_backup._discover_jsonl_candidates([source_root]) + + archive = jsonl_backup.create_jsonl_bundle(candidates, tmp_path / "staging", date_stamp="2026-09-09") + verification = jsonl_backup.verify_jsonl_bundle(archive, expected_candidates=candidates) + + with tarfile.open(archive, "r:gz") as bundle: + member = bundle.getmember("source-0/session.jsonl") + extracted = bundle.extractfile(member) + assert member.isfile() + assert extracted is not None + assert extracted.read() == target.read_bytes() + assert verification["verified"] is True + + +def test_jsonl_bundle_verification_rejects_same_count_with_changed_bytes(tmp_path): + from brainlayer import jsonl_backup + + source_root = tmp_path / "sessions" + source = source_root / "session.jsonl" + source.parent.mkdir(parents=True) + source.write_bytes(b'{"original":true}\n') + candidates = jsonl_backup._discover_jsonl_candidates([source_root]) + archive = tmp_path / "changed.tar.gz" + changed = b'{"original":null}\n' + member = tarfile.TarInfo("source-0/session.jsonl") + member.size = len(changed) + with tarfile.open(archive, "w:gz") as bundle: + bundle.addfile(member, io.BytesIO(changed)) + + verification = jsonl_backup.verify_jsonl_bundle(archive, expected_candidates=candidates) + + assert verification["verified"] is False + assert verification["content_verified_file_count"] == 0 + assert verification["verification_error"] == "archive member differs from source bytes: source-0/session.jsonl" + + +def test_jsonl_bundle_verification_rejects_member_shorter_than_discovered_candidate(tmp_path): + from brainlayer import jsonl_backup + + source_root = tmp_path / "sessions" + source = source_root / "session.jsonl" + source.parent.mkdir(parents=True) + source.write_bytes(b"abcdef") + candidates = jsonl_backup._discover_jsonl_candidates([source_root]) + archive = tmp_path / "truncated.tar.gz" + member = tarfile.TarInfo("source-0/session.jsonl") + member.size = 3 + with tarfile.open(archive, "w:gz") as bundle: + bundle.addfile(member, io.BytesIO(b"abc")) + + verification = jsonl_backup.verify_jsonl_bundle(archive, expected_candidates=candidates) + + assert verification["verified"] is False + assert verification["verification_error"] == "archive member size differs from bundle: source-0/session.jsonl" + + +def test_jsonl_bundle_rejects_member_shorter_than_bundle_time_size(tmp_path): + from brainlayer import jsonl_backup + + source_root = tmp_path / "sessions" + source = source_root / "session.jsonl" + source.parent.mkdir(parents=True) + source.write_bytes(b"abc") + candidates = jsonl_backup._discover_jsonl_candidates([source_root]) + source.write_bytes(b"abcdef") + archive = tmp_path / "truncated-after-growth.tar.gz" + member = tarfile.TarInfo("source-0/session.jsonl") + member.size = 4 + with tarfile.open(archive, "w:gz") as bundle: + bundle.addfile(member, io.BytesIO(b"abcd")) + + verification = jsonl_backup.verify_jsonl_bundle( + archive, + expected_candidates=candidates, + expected_sizes={source.as_posix(): 6}, + ) + + assert verification["verified"] is False + assert verification["verification_error"] == "archive member size differs from bundle: source-0/session.jsonl" + + +def test_jsonl_bundle_verification_does_not_swallow_backup_timeout(tmp_path, monkeypatch): + from brainlayer import backup_daily, jsonl_backup + + source_root = tmp_path / "sessions" + source = source_root / "session.jsonl" + source.parent.mkdir(parents=True) + source.write_bytes(b"content") + candidates = jsonl_backup._discover_jsonl_candidates([source_root]) + archive = jsonl_backup.create_jsonl_bundle(candidates, tmp_path / "staging", date_stamp="2026-09-09") + monkeypatch.setattr( + jsonl_backup, + "_compare_member_to_source", + lambda *args, **kwargs: (_ for _ in ()).throw(backup_daily.BackupTimeoutError("timed out")), + ) + + with pytest.raises(backup_daily.BackupTimeoutError, match="timed out"): + jsonl_backup.verify_jsonl_bundle(archive, expected_candidates=candidates) + + +def test_jsonl_bundle_verification_accepts_and_counts_append_only_snapshot(tmp_path): + from brainlayer import jsonl_backup + + source_root = tmp_path / "sessions" + source = source_root / "session.jsonl" + source.parent.mkdir(parents=True) + source.write_bytes(b'{"first":true}\n') + candidates = jsonl_backup._discover_jsonl_candidates([source_root]) + archive = jsonl_backup.create_jsonl_bundle(candidates, tmp_path / "staging", date_stamp="2026-09-09") + with source.open("ab") as handle: + handle.write(b'{"appended":true}\n') + + verification = jsonl_backup.verify_jsonl_bundle(archive, expected_candidates=candidates) + + assert verification["verified"] is True + assert verification["content_verified_file_count"] == 1 + assert verification["append_snapshot_file_count"] == 1 + + +def test_jsonl_backup_does_not_upload_or_advance_state_when_content_verification_fails(tmp_path, monkeypatch): + from brainlayer import jsonl_backup + + now = time.time() + source_root = tmp_path / "sessions" + _write_jsonl(source_root / "session.jsonl", mtime=now - 3600) + monkeypatch.setattr( + jsonl_backup, + "verify_jsonl_bundle", + lambda *args, **kwargs: { + "verified": False, + "bundled_file_count": 1, + "archive_listing_count": 1, + "content_verified_file_count": 0, + "verification_error": "archive member differs from source bytes: source-0/session.jsonl", + }, + ) + monkeypatch.setattr( + jsonl_backup.backup_daily, + "get_drive_credentials", + lambda: pytest.fail("upload must not start before content verification"), + ) + + result = jsonl_backup.run_backup( + 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", + date_stamp="2026-09-09", + now=now, + upload=True, + ) + + assert result["status"] == "failed" + assert result["uploaded"] is False + assert result["verified"] is False + assert not (tmp_path / "state.json").exists() + + +def test_concurrent_jsonl_backups_serialize_creation_through_state_persistence(tmp_path, monkeypatch): + from brainlayer import jsonl_backup + + now = time.time() + source_root = tmp_path / "sessions" + _write_jsonl(source_root / "session.jsonl", mtime=now - 3600) + upload_started = threading.Event() + release_upload = threading.Event() + uploads: list[bytes] = [] + surviving: list[dict] = [] + results: list[dict] = [] + errors: list[BaseException] = [] + + monkeypatch.setattr(jsonl_backup.backup_daily, "get_drive_credentials", lambda: object()) + monkeypatch.setattr( + jsonl_backup.backup_daily, + "build_drive_service", + lambda: _drive_service_with_surviving(surviving), + ) + monkeypatch.setattr(jsonl_backup.backup_daily, "ensure_drive_folder_chain", lambda *args: "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: []) + + def fake_upload(file_path, folder_id, credentials): # noqa: ARG001 + uploads.append(Path(file_path).read_bytes()) + if len(uploads) == 1: + upload_started.set() + assert release_upload.wait(timeout=2) + uploaded = { + "id": f"drive-{len(uploads)}", + "name": Path(file_path).name, + "size": str(Path(file_path).stat().st_size), + "md5Checksum": f"md5-{len(uploads)}", + } + surviving.append(uploaded) + return uploaded + + monkeypatch.setattr(jsonl_backup.backup_daily, "upload_file_to_drive_raw", fake_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", + "date_stamp": "2026-09-09", + "now": now, + "upload": True, + } + + def run() -> None: + try: + results.append(jsonl_backup.run_backup(**kwargs)) + except BaseException as exc: + errors.append(exc) + + first = threading.Thread(target=run) + second = threading.Thread(target=run) + first.start() + assert upload_started.wait(timeout=2) + second.start() + time.sleep(0.1) + assert len(uploads) == 1 + assert second.is_alive() + release_upload.set() + first.join(timeout=2) + second.join(timeout=2) + + assert errors == [] + assert not first.is_alive() + assert not second.is_alive() + assert len(uploads) == 1 + assert sorted(result["status"] for result in results) == ["no-op", "uploaded"] + + def test_run_jsonl_backup_uploads_incremental_bundle_verifies_and_enqueues_summary(tmp_path, monkeypatch): from brainlayer import jsonl_backup @@ -113,6 +462,7 @@ def fake_prune(service, *, folder_parts, retention_policy): # noqa: ARG001 assert result["verified"] is True assert result["bundled_file_count"] == 3 assert result["archive_listing_count"] == 3 + assert result["content_verified_file_count"] == 3 assert result["skipped_active_count"] == 1 assert active.as_posix() not in (tmp_path / "state.json").read_text() state = json.loads((tmp_path / "state.json").read_text())