test(backup): make survivor-copy retention invariant structural

src/brainlayer/backup_retention_invariant.py

@@ -0,0 +1,187 @@
+"""Static CI guard for the JSONL backup retention invariant introduced in PR #815.
+
+Behavior tests prove today's examples. This guard also pins the production call graph so a
+future refactor cannot keep the fixtures green while bypassing the surviving-copy evidence.
+"""
+
+from __future__ import annotations
+
+import ast
+import sys
+from pathlib import Path
+
+REFACTOR_GUIDANCE = (
+    "PR #815 shipped an integrity check that could never execute while its tests passed. "
+    "If you refactored deliberately, UPDATE this guard; do not delete it."
+)
+
+
+def _with_refactor_guidance(errors: list[str]) -> list[str]:
+    return [*errors, REFACTOR_GUIDANCE] if errors else errors
+
+
+def _function(tree: ast.AST, name: str) -> ast.FunctionDef | None:
+    return next(
+        (node for node in ast.walk(tree) if isinstance(node, ast.FunctionDef) and node.name == name),
+        None,
+    )
+
+
+def _call_name(call: ast.Call) -> str | None:
+    if isinstance(call.func, ast.Name):
+        return call.func.id
+    if isinstance(call.func, ast.Attribute):
+        return call.func.attr
+    return None
+
+
+def _calls(function: ast.FunctionDef, name: str) -> list[ast.Call]:
+    return [node for node in ast.walk(function) if isinstance(node, ast.Call) and _call_name(node) == name]
+
+
+def _has_compare(function: ast.FunctionDef, *, operator: type[ast.cmpop], terms: tuple[str, ...]) -> bool:
+    for node in ast.walk(function):
+        if not isinstance(node, ast.Compare) or not any(isinstance(op, operator) for op in node.ops):
+            continue
+        rendered = ast.unparse(node)
+        if all(term in rendered for term in terms):
+            return True
+    return False
+
+
+def _passes_live_inventory(call: ast.Call) -> bool:
+    if any(
+        keyword.arg == "surviving_archives"
+        and isinstance(keyword.value, ast.Name)
+        and keyword.value.id == "surviving_archives"
+        for keyword in call.keywords
+    ):
+        return True
+    return len(call.args) >= 3 and isinstance(call.args[2], ast.Name) and call.args[2].id == "surviving_archives"
+
+
+def _verified_upload_delete_lines(function: ast.FunctionDef) -> set[int]:
+    lines: set[int] = set()
+    for node in ast.walk(function):
+        if not isinstance(node, ast.If):
+            continue
+        condition = ast.unparse(node.test)
+        if 'result["verified"]' not in condition and "result['verified']" not in condition:
+            continue
+        if "upload" not in condition:
+            continue
+        lines.update(child.lineno for child in ast.walk(node) if isinstance(child, ast.Call))
+    return lines
+
+
+def inspect_jsonl_retention_invariant(source: str) -> list[str]:
+    """Return deterministic violations of the PR #815 surviving-copy contract."""
+    try:
+        tree = ast.parse(source)
+    except SyntaxError as exc:
+        return [f"jsonl_backup.py is not valid Python: {exc}"]
+
+    errors: list[str] = []
+    state_matches = _function(tree, "_state_matches")
+    select_candidates = _function(tree, "_select_backup_candidates")
+    update_state = _function(tree, "_update_state_for_uploaded")
+    run_backup = _function(tree, "run_backup")
+    required = {
+        "_state_matches": state_matches,
+        "_select_backup_candidates": select_candidates,
+        "_update_state_for_uploaded": update_state,
+        "run_backup": run_backup,
+    }
+    for name, function in required.items():
+        if function is None:
+            errors.append(f"required retention function is missing: {name}")
+    if errors:
+        return _with_refactor_guidance(errors)
+
+    assert state_matches is not None
+    assert select_candidates is not None
+    assert update_state is not None
+    assert run_backup is not None
+
+    if not _has_compare(
+        state_matches,
+        operator=ast.NotIn,
+        terms=("archive_id", "surviving_archives"),
+    ):
+        errors.append("coverage must reject archive IDs absent from the live Drive inventory")
+    if not _has_compare(
+        state_matches,
+        operator=ast.NotEq,
+        terms=("live_md5", "recorded_md5"),
+    ):
+        errors.append("coverage must reject a surviving Drive object whose archived bytes changed")
+    if not _has_compare(
+        state_matches,
+        operator=ast.Eq,
+        terms=("recorded_hash", "_sha256_file(candidate.path)"),
+    ):
+        errors.append("coverage must compare the live source bytes with the archived source digest")
+
+    select_calls = _calls(select_candidates, "_state_matches")
+    if not select_calls or not any(_passes_live_inventory(call) for call in select_calls):
+        errors.append("candidate selection must pass the live Drive inventory into the coverage predicate")
+
+    list_calls = _calls(run_backup, "_list_surviving_archives")
+    selection_calls = _calls(run_backup, "_select_backup_candidates")
+    if (
+        not list_calls
+        or not selection_calls
+        or min(call.lineno for call in list_calls) >= min(call.lineno for call in selection_calls)
+    ):
+        errors.append("run_backup must list surviving Drive objects before selecting covered files")
+    elif not any(_passes_live_inventory(call) for call in selection_calls):
+        errors.append("run_backup must hand its live Drive inventory to candidate selection")
+
+    state_write_calls = _calls(run_backup, "_update_state_for_uploaded")
+    required_state_keywords = {"archive_id", "archive_md5", "digests"}
+    if not state_write_calls or not any(
+        required_state_keywords <= {keyword.arg for keyword in call.keywords if keyword.arg}
+        for call in state_write_calls
+    ):
+        errors.append("uploaded state must persist archive identity, archive bytes, and source-byte digests")
+
+    prune_calls = _calls(run_backup, "prune_drive_backups")
+    archive_unlinks = [
+        call
+        for call in _calls(run_backup, "unlink")
+        if isinstance(call.func, ast.Attribute)
+        and isinstance(call.func.value, ast.Name)
+        and call.func.value.id == "archive_path"
+    ]
+    delete_calls = [*prune_calls, *archive_unlinks]
+    verified_lines = _verified_upload_delete_lines(run_backup)
+    if not prune_calls:
+        errors.append("the Drive retention deletion call disappeared instead of retaining its safety contract")
+    if not archive_unlinks:
+        errors.append("the local staging deletion call disappeared instead of retaining its safety contract")
+    if any(call.lineno not in verified_lines for call in delete_calls):
+        errors.append("backup deletion calls must remain inside verified-upload control flow")
+    if (
+        state_write_calls
+        and delete_calls
+        and max(call.lineno for call in state_write_calls) >= min(call.lineno for call in delete_calls)
+    ):
+        errors.append("surviving-copy provenance must be persisted before any backup deletion call")
+
+    return _with_refactor_guidance(errors)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = sys.argv[1:] if argv is None else argv
+    path = Path(args[0]) if args else Path("src/brainlayer/jsonl_backup.py")
+    errors = inspect_jsonl_retention_invariant(path.read_text(encoding="utf-8"))
+    if errors:
+        for error in errors:
+            print(f"FAIL: {error}")
+        return 1
+    print(f"PASS: PR #815 JSONL retention invariant is structurally intact in {path}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/brainlayer/jsonl_backup.py

@@ -535,6 +535,11 @@ def run_backup(
     service = None
     surviving_archives: dict[str, str | None] | None = None
     if upload:
+        # 2026-09-09 retention incident / PR #815: 22 uploaded+verified daily bundles
+        # later disappeared under retention while unchanged files still read as covered.
+        # The live Drive inventory must be fetched BEFORE coverage is decided; otherwise
+        # state can vouch for an object that no longer exists. CI pins this ordering in
+        # backup_retention_invariant.py.
         # 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
@@ -607,6 +612,10 @@ def run_backup(
 
     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.
         _atomic_write_json(
             state_path,
             _update_state_for_uploaded(
@@ -619,6 +628,8 @@ def run_backup(
             ),
         )
         try:
+            # Do not move this ahead of the provenance write or loosen `_state_matches` to
+            # mtime/size. That exact shape left 22 successful nights with no surviving bundle.
             deleted = backup_daily.prune_drive_backups(
                 service,
                 folder_parts=folder_parts,
@@ -636,6 +647,9 @@ def run_backup(
                 result["forever_files"] = forever_files
                 result["forever_uploaded_file_count"] = len(forever_files)
         finally:
+            # Local staging may disappear only inside verified-upload control flow. Together
+            # with the retention call above, this unlink is why survivor identity is a deletion
+            # invariant rather than an optional integrity check (2026-09-09 / PR #815).
             archive_path.unlink(missing_ok=True)
             result["local_archive_removed"] = True
 

tests/test_backup_retention_invariant.py

@@ -0,0 +1,13 @@
+from pathlib import Path
+
+
+def test_jsonl_retention_guard_failure_tells_refactors_to_update_not_delete(tmp_path: Path, capsys) -> None:
+    from brainlayer.backup_retention_invariant import main
+
+    unsafe_path = tmp_path / "jsonl_backup.py"
+    unsafe_path.write_text("def run_backup():\n    pass\n", encoding="utf-8")
+
+    assert main([str(unsafe_path)]) == 1
+    output = capsys.readouterr().out
+    assert "UPDATE this guard; do not delete it" in output
+    assert "PR #815" in output

tests/test_jsonl_backup.py

@@ -15,6 +15,52 @@ def _write_jsonl(path: Path, line: str = '{"type":"message"}\n', *, mtime: float
     return path
 
 
+def test_jsonl_retention_invariant_is_a_ci_guard_not_only_a_behavior_fixture():
+    """Fail CI when #815's surviving-copy predicate or call-site ordering is loosened."""
+    from brainlayer.backup_retention_invariant import inspect_jsonl_retention_invariant
+
+    source = Path("src/brainlayer/jsonl_backup.py").read_text(encoding="utf-8")
+
+    assert inspect_jsonl_retention_invariant(source) == []
+
+    mutations = (
+        (
+            "archive_id not in surviving_archives",
+            "archive_id in surviving_archives",
+            "coverage must reject archive IDs absent from the live Drive inventory",
+        ),
+        (
+            "live_md5 != recorded_md5",
+            "live_md5 == recorded_md5",
+            "coverage must reject a surviving Drive object whose archived bytes changed",
+        ),
+        (
+            "recorded_hash == _sha256_file(candidate.path)",
+            "recorded_hash != _sha256_file(candidate.path)",
+            "coverage must compare the live source bytes with the archived source digest",
+        ),
+        (
+            "surviving_archives=surviving_archives",
+            "surviving_archives=None",
+            "run_backup must hand its live Drive inventory to candidate selection",
+        ),
+        (
+            "archive_id=file_id",
+            "missing_archive_id=file_id",
+            "uploaded state must persist archive identity, archive bytes, and source-byte digests",
+        ),
+        (
+            'if result["verified"] and upload:',
+            "if upload:",
+            "backup deletion calls must remain inside verified-upload control flow",
+        ),
+    )
+    for original, weakened, expected_error in mutations:
+        assert original in source, f"mutation fixture drifted: {original}"
+        unsafe = source.replace(original, weakened, 1)
+        assert expected_error in inspect_jsonl_retention_invariant(unsafe)
+
+
 def test_run_jsonl_backup_uploads_incremental_bundle_verifies_and_enqueues_summary(tmp_path, monkeypatch):
     from brainlayer import jsonl_backup