From 2bcf4eff35c08edb373da41082a9b47277f72f5f Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Wed, 9 Sep 2026 21:30:25 +0300 Subject: [PATCH 1/5] fix(test): close retention guard semantic bypasses (S) Co-Authored-By: brainlayerCodex-4c4386f6 --- src/brainlayer/backup_retention_invariant.py | 202 ++++++++++++++++--- tests/test_jsonl_backup.py | 97 ++++++++- 2 files changed, 270 insertions(+), 29 deletions(-) diff --git a/src/brainlayer/backup_retention_invariant.py b/src/brainlayer/backup_retention_invariant.py index e8a92c32..b7310f15 100644 --- a/src/brainlayer/backup_retention_invariant.py +++ b/src/brainlayer/backup_retention_invariant.py @@ -2,6 +2,8 @@ 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. +It deliberately proves call-graph shape and the specific md5 producer/consumer seam from #815; +it is not general data-flow analysis. Behavioral tests own proof that runtime values are populated. """ from __future__ import annotations @@ -39,10 +41,34 @@ 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: +def _parent_map(tree: ast.AST) -> dict[ast.AST, ast.AST]: + return {child: parent for parent in ast.walk(tree) for child in ast.iter_child_nodes(parent)} + + +def _inside_statically_dead_branch(node: ast.AST, parents: dict[ast.AST, ast.AST]) -> bool: + child = node + while parent := parents.get(child): + if isinstance(parent, ast.If) and isinstance(parent.test, ast.Constant): + if parent.test.value is False and child in parent.body: + return True + if parent.test.value is True and child in parent.orelse: + return True + child = parent + return False + + +def _has_reachable_compare( + function: ast.FunctionDef, + *, + operator: type[ast.cmpop], + terms: tuple[str, ...], + parents: dict[ast.AST, ast.AST], +) -> 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 + if _inside_statically_dead_branch(node, parents): + continue rendered = ast.unparse(node) if all(term in rendered for term in terms): return True @@ -60,37 +86,94 @@ def _passes_live_inventory(call: ast.Call) -> bool: 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 _is_exact_verified_upload_gate(node: ast.If) -> bool: + expected = ast.parse('result["verified"] and upload', mode="eval").body + return ast.dump(node.test, include_attributes=False) == ast.dump(expected, include_attributes=False) + + +def _calls_in_statements(statements: list[ast.stmt]) -> list[ast.Call]: + return [child for statement in statements for child in ast.walk(statement) if isinstance(child, ast.Call)] + + +def _direct_expression_call(statement: ast.stmt, name: str) -> ast.Call | None: + if not isinstance(statement, ast.Expr) or not isinstance(statement.value, ast.Call): + return None + return statement.value if _call_name(statement.value) == name else None + + +def _allowed_coverage_return(node: ast.Return, parents: dict[ast.AST, ast.AST]) -> bool: + if isinstance(node.value, ast.Constant) and node.value.value is False: + return True + if isinstance(node.value, ast.Compare): + rendered = ast.unparse(node.value) + return ( + any(isinstance(operator, ast.Eq) for operator in node.value.ops) + and "recorded_hash" in rendered + and "_sha256_file(candidate.path)" in rendered + ) + if not (isinstance(node.value, ast.Constant) and node.value.value is True): + return False + parent = parents.get(node) + if not isinstance(parent, ast.If) or node not in parent.body: + return False + expected = ast.parse("surviving_archives is None", mode="eval").body + return ast.dump(parent.test, include_attributes=False) == ast.dump(expected, include_attributes=False) + + +def _has_exact_single_assignment( + function: ast.FunctionDef, + *, + target: str, + expression: str, + parents: dict[ast.AST, ast.AST], +) -> bool: + stores = [ + node + for node in ast.walk(function) + if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Store) and node.id == target + ] + if len(stores) != 1: + return False + assignment = parents.get(stores[0]) + if not isinstance(assignment, ast.Assign) or assignment.targets != [stores[0]]: + return False + expected = ast.parse(expression, mode="eval").body + return ast.dump(assignment.value, include_attributes=False) == ast.dump(expected, include_attributes=False) + + +def _keyword_matches(call: ast.Call, *, name: str, expression: str) -> bool: + expected = ast.parse(expression, mode="eval").body + return any( + keyword.arg == name + and ast.dump(keyword.value, include_attributes=False) == ast.dump(expected, include_attributes=False) + for keyword in call.keywords + ) -def inspect_jsonl_retention_invariant(source: str) -> list[str]: +def inspect_jsonl_retention_invariant(source: str, *, backup_daily_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}"] + try: + backup_daily_tree = ast.parse(backup_daily_source) + except SyntaxError as exc: + return [f"backup_daily.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") + upload_file = _function(backup_daily_tree, "upload_file_to_drive_raw") required = { "_state_matches": state_matches, "_select_backup_candidates": select_candidates, "_update_state_for_uploaded": update_state, "run_backup": run_backup, + "upload_file_to_drive_raw": upload_file, } for name, function in required.items(): if function is None: @@ -102,25 +185,44 @@ def inspect_jsonl_retention_invariant(source: str) -> list[str]: assert select_candidates is not None assert update_state is not None assert run_backup is not None + assert upload_file is not None + parents = _parent_map(tree) - if not _has_compare( + if not _has_exact_single_assignment( + state_matches, + target="recorded_md5", + expression='entry.get("archive_md5")', + parents=parents, + ): + errors.append("coverage must read the recorded archive md5 from persisted state") + + if not _has_reachable_compare( state_matches, operator=ast.NotIn, terms=("archive_id", "surviving_archives"), + parents=parents, ): errors.append("coverage must reject archive IDs absent from the live Drive inventory") - if not _has_compare( + if not _has_reachable_compare( state_matches, operator=ast.NotEq, terms=("live_md5", "recorded_md5"), + parents=parents, ): errors.append("coverage must reject a surviving Drive object whose archived bytes changed") - if not _has_compare( + if not _has_reachable_compare( state_matches, operator=ast.Eq, terms=("recorded_hash", "_sha256_file(candidate.path)"), + parents=parents, ): errors.append("coverage must compare the live source bytes with the archived source digest") + if any( + not _allowed_coverage_return(node, parents) + for node in ast.walk(state_matches) + if isinstance(node, ast.Return) and not _inside_statically_dead_branch(node, parents) + ): + errors.append("every successful coverage path must require surviving-copy evidence") select_calls = _calls(select_candidates, "_state_matches") if not select_calls or not any(_passes_live_inventory(call) for call in select_calls): @@ -137,13 +239,43 @@ def inspect_jsonl_retention_invariant(source: str) -> list[str]: 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") + verified_gates = [ + node for node in ast.walk(run_backup) if isinstance(node, ast.If) and _is_exact_verified_upload_gate(node) + ] + verified_gate = verified_gates[0] if len(verified_gates) == 1 else None + safe_calls = _calls_in_statements(verified_gate.body) if verified_gate is not None else [] + + persistence: list[tuple[int, ast.Call, ast.Call]] = [] + if verified_gate is not None: + for index, statement in enumerate(verified_gate.body): + atomic_write = _direct_expression_call(statement, "_atomic_write_json") + if atomic_write is None: + continue + updates = [ + child + for child in ast.walk(atomic_write) + if isinstance(child, ast.Call) and _call_name(child) == "_update_state_for_uploaded" + ] + if len(updates) == 1: + persistence.append((index, atomic_write, updates[0])) + required_state_keywords = {"archive_id", "archive_md5", "digests"} - if not state_write_calls or not any( + if not persistence or not any( required_state_keywords <= {keyword.arg for keyword in call.keywords if keyword.arg} - for call in state_write_calls + for _, _, call in persistence ): errors.append("uploaded state must persist archive identity, archive bytes, and source-byte digests") + if not any( + _keyword_matches(call, name="archive_md5", expression='uploaded.get("md5Checksum")') + for _, _, call in persistence + ): + errors.append("uploaded state must persist md5Checksum from the upload response") + + if not any( + isinstance(node, ast.Constant) and isinstance(node.value, str) and "md5Checksum" in node.value + for node in ast.walk(upload_file) + ): + errors.append("Drive upload must request md5Checksum from the API") prune_calls = _calls(run_backup, "prune_drive_backups") archive_unlinks = [ @@ -154,19 +286,27 @@ def inspect_jsonl_retention_invariant(source: str) -> list[str]: 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): + if verified_gate is None or any(call not in safe_calls for call in delete_calls): errors.append("backup deletion calls must remain inside verified-upload control flow") + deletion_statement_indexes = ( + [ + index + for index, statement in enumerate(verified_gate.body) + if any(call in delete_calls for call in ast.walk(statement) if isinstance(call, ast.Call)) + ] + if verified_gate is not None + else [] + ) 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) + not persistence + or not deletion_statement_indexes + or min(index for index, _, _ in persistence) >= min(deletion_statement_indexes) ): - errors.append("surviving-copy provenance must be persisted before any backup deletion call") + errors.append("surviving-copy provenance must be durably persisted before any backup deletion call") return _with_refactor_guidance(errors) @@ -174,7 +314,15 @@ def inspect_jsonl_retention_invariant(source: str) -> list[str]: 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")) + backup_daily_path = path.with_name("backup_daily.py") + if not backup_daily_path.exists(): + print(f"FAIL: required sibling source is missing: {backup_daily_path}") + print(f"FAIL: {REFACTOR_GUIDANCE}") + return 1 + errors = inspect_jsonl_retention_invariant( + path.read_text(encoding="utf-8"), + backup_daily_source=backup_daily_path.read_text(encoding="utf-8"), + ) if errors: for error in errors: print(f"FAIL: {error}") diff --git a/tests/test_jsonl_backup.py b/tests/test_jsonl_backup.py index b205c208..3632fa90 100644 --- a/tests/test_jsonl_backup.py +++ b/tests/test_jsonl_backup.py @@ -20,8 +20,9 @@ def test_jsonl_retention_invariant_is_a_ci_guard_not_only_a_behavior_fixture(): from brainlayer.backup_retention_invariant import inspect_jsonl_retention_invariant source = Path("src/brainlayer/jsonl_backup.py").read_text(encoding="utf-8") + backup_daily_source = Path("src/brainlayer/backup_daily.py").read_text(encoding="utf-8") - assert inspect_jsonl_retention_invariant(source) == [] + assert inspect_jsonl_retention_invariant(source, backup_daily_source=backup_daily_source) == [] mutations = ( ( @@ -34,6 +35,11 @@ def test_jsonl_retention_invariant_is_a_ci_guard_not_only_a_behavior_fixture(): "live_md5 == recorded_md5", "coverage must reject a surviving Drive object whose archived bytes changed", ), + ( + ' recorded_md5 = entry.get("archive_md5")', + ' if False:\n recorded_md5 = entry.get("archive_md5")\n recorded_md5 = None', + "coverage must read the recorded archive md5 from persisted state", + ), ( "recorded_hash == _sha256_file(candidate.path)", "recorded_hash != _sha256_file(candidate.path)", @@ -54,11 +60,98 @@ def test_jsonl_retention_invariant_is_a_ci_guard_not_only_a_behavior_fixture(): "if upload:", "backup deletion calls must remain inside verified-upload control flow", ), + ( + 'if result["verified"] and upload:', + 'if result["verified"] or upload:', + "backup deletion calls must remain inside verified-upload control flow", + ), + ( + 'if result["verified"] and upload:', + 'if not result["verified"] and upload:', + "backup deletion calls must remain inside verified-upload control flow", + ), + ( + 'if result["verified"] and upload:', + 'if result["verified"] and upload:\n pass\n else:', + "backup deletion calls must remain inside verified-upload control flow", + ), + ( + " _atomic_write_json(\n state_path,", + " if False:\n _atomic_write_json(\n state_path,", + "surviving-copy provenance must be durably persisted before any backup deletion call", + ), + ( + " if not isinstance(entry, dict):", + ( + " if surviving_archives is not None and isinstance(entry, dict):\n" + " return True\n" + " if not isinstance(entry, dict):" + ), + "every successful coverage path must require surviving-copy evidence", + ), + ( + (" if not isinstance(archive_id, str) or archive_id not in surviving_archives:\n return False"), + (" if False:\n if archive_id not in surviving_archives:\n return False"), + "coverage must reject archive IDs absent from the live Drive inventory", + ), ) 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) + assert expected_error in inspect_jsonl_retention_invariant(unsafe, backup_daily_source=backup_daily_source) + + unsafe = source.replace( + 'archive_md5=uploaded.get("md5Checksum")', + "archive_md5=None", + 1, + ) + assert "uploaded state must persist md5Checksum from the upload response" in ( + inspect_jsonl_retention_invariant(unsafe, backup_daily_source=backup_daily_source) + ) + + unsafe_backup_daily = backup_daily_source.replace( + '"&fields=id,name,size,md5Checksum"', + '"&fields=id,name,size"', + 1, + ) + assert "Drive upload must request md5Checksum from the API" in ( + inspect_jsonl_retention_invariant(source, backup_daily_source=unsafe_backup_daily) + ) + + persisted_state_block = """ _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, + ), + ) +""" + constructed_state_block = """ uploaded_state = _update_state_for_uploaded( + state, + changed, + archive_path.name, + archive_id=file_id, + archive_md5=uploaded.get("md5Checksum"), + digests=bundle_digests, + ) +""" + assert persisted_state_block in source + unsafe = source.replace(persisted_state_block, constructed_state_block, 1).replace( + ' result["local_archive_removed"] = True\n', + ( + ' result["local_archive_removed"] = True\n' + " _atomic_write_json(state_path, uploaded_state)\n" + ), + 1, + ) + assert ( + "surviving-copy provenance must be durably persisted before any backup deletion call" + in inspect_jsonl_retention_invariant(unsafe, backup_daily_source=backup_daily_source) + ) def test_run_jsonl_backup_uploads_incremental_bundle_verifies_and_enqueues_summary(tmp_path, monkeypatch): From a936aef8830ba1d7dd4a4bba1602d003183cd056 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Wed, 9 Sep 2026 21:36:40 +0300 Subject: [PATCH 2/5] fix(test): bind guard to effective md5 request (XS) Co-Authored-By: brainlayerCodex-4c4386f6 --- src/brainlayer/backup_retention_invariant.py | 24 +++++++++++++------- tests/test_jsonl_backup.py | 18 +++++++++++++++ 2 files changed, 34 insertions(+), 8 deletions(-) diff --git a/src/brainlayer/backup_retention_invariant.py b/src/brainlayer/backup_retention_invariant.py index b7310f15..391c67fc 100644 --- a/src/brainlayer/backup_retention_invariant.py +++ b/src/brainlayer/backup_retention_invariant.py @@ -11,6 +11,7 @@ import ast import sys from pathlib import Path +from urllib.parse import parse_qs, urlsplit REFACTOR_GUIDANCE = ( "PR #815 shipped an integrity check that could never execute while its tests passed. " @@ -23,10 +24,10 @@ def _with_refactor_guidance(errors: list[str]) -> list[str]: 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, - ) + definitions = [ + node for node in getattr(tree, "body", []) if isinstance(node, ast.FunctionDef) and node.name == name + ] + return definitions[0] if len(definitions) == 1 else None def _call_name(call: ast.Call) -> str | None: @@ -150,6 +151,16 @@ def _keyword_matches(call: ast.Call, *, name: str, expression: str) -> bool: ) +def _upload_requests_md5(function: ast.FunctionDef) -> bool: + for call in _calls(function, "post"): + if not call.args or not isinstance(call.args[0], ast.Constant) or not isinstance(call.args[0].value, str): + continue + fields = parse_qs(urlsplit(call.args[0].value).query).get("fields", []) + if any("md5Checksum" in value.split(",") for value in fields): + return True + return False + + def inspect_jsonl_retention_invariant(source: str, *, backup_daily_source: str) -> list[str]: """Return deterministic violations of the PR #815 surviving-copy contract.""" try: @@ -271,10 +282,7 @@ def inspect_jsonl_retention_invariant(source: str, *, backup_daily_source: str) ): errors.append("uploaded state must persist md5Checksum from the upload response") - if not any( - isinstance(node, ast.Constant) and isinstance(node.value, str) and "md5Checksum" in node.value - for node in ast.walk(upload_file) - ): + if not _upload_requests_md5(upload_file): errors.append("Drive upload must request md5Checksum from the API") prune_calls = _calls(run_backup, "prune_drive_backups") diff --git a/tests/test_jsonl_backup.py b/tests/test_jsonl_backup.py index 3632fa90..3d0f2d7f 100644 --- a/tests/test_jsonl_backup.py +++ b/tests/test_jsonl_backup.py @@ -113,11 +113,29 @@ def test_jsonl_retention_invariant_is_a_ci_guard_not_only_a_behavior_fixture(): '"&fields=id,name,size,md5Checksum"', '"&fields=id,name,size"', 1, + ).replace( + '"""Upload large backups with Drive\'s raw resumable protocol."""', + '"""Upload large backups; mention md5Checksum without requesting it."""', + 1, ) assert "Drive upload must request md5Checksum from the API" in ( inspect_jsonl_retention_invariant(source, backup_daily_source=unsafe_backup_daily) ) + duplicate_definition = ( + source + + """ +def _state_matches(entry, candidate, surviving_archives=None): + return True +""" + ) + assert "required retention function is missing: _state_matches" in ( + inspect_jsonl_retention_invariant( + duplicate_definition, + backup_daily_source=backup_daily_source, + ) + ) + persisted_state_block = """ _atomic_write_json( state_path, _update_state_for_uploaded( From da6ad9b0a06599f2534a7d38cba529d1fbc5731a Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Wed, 9 Sep 2026 21:41:28 +0300 Subject: [PATCH 3/5] fix(test): make survivor evidence dominate deletion (S) Co-Authored-By: brainlayerCodex-4c4386f6 --- src/brainlayer/backup_retention_invariant.py | 52 +++++++++++++++++--- tests/test_jsonl_backup.py | 17 +++++++ 2 files changed, 62 insertions(+), 7 deletions(-) diff --git a/src/brainlayer/backup_retention_invariant.py b/src/brainlayer/backup_retention_invariant.py index 391c67fc..f43e4ca3 100644 --- a/src/brainlayer/backup_retention_invariant.py +++ b/src/brainlayer/backup_retention_invariant.py @@ -76,6 +76,25 @@ def _has_reachable_compare( return False +def _has_exact_false_rejection( + function: ast.FunctionDef, + *, + condition: str, + parents: dict[ast.AST, ast.AST], +) -> bool: + expected = ast.parse(condition, mode="eval").body + for node in ast.walk(function): + if not isinstance(node, ast.If) or _inside_statically_dead_branch(node, parents): + continue + if ast.dump(node.test, include_attributes=False) != ast.dump(expected, include_attributes=False): + continue + if len(node.body) != 1 or not isinstance(node.body[0], ast.Return): + continue + if isinstance(node.body[0].value, ast.Constant) and node.body[0].value.value is False: + return True + return False + + def _passes_live_inventory(call: ast.Call) -> bool: if any( keyword.arg == "surviving_archives" @@ -92,6 +111,21 @@ def _is_exact_verified_upload_gate(node: ast.If) -> bool: return ast.dump(node.test, include_attributes=False) == ast.dump(expected, include_attributes=False) +def _is_verification_update(statement: ast.stmt) -> bool: + if not isinstance(statement, ast.Expr) or not isinstance(statement.value, ast.Call): + return False + call = statement.value + if not ( + isinstance(call.func, ast.Attribute) + and isinstance(call.func.value, ast.Name) + and call.func.value.id == "result" + and call.func.attr == "update" + and len(call.args) == 1 + ): + return False + return isinstance(call.args[0], ast.Call) and _call_name(call.args[0]) == "verify_jsonl_bundle" + + def _calls_in_statements(statements: list[ast.stmt]) -> list[ast.Call]: return [child for statement in statements for child in ast.walk(statement) if isinstance(child, ast.Call)] @@ -207,17 +241,15 @@ def inspect_jsonl_retention_invariant(source: str, *, backup_daily_source: str) ): errors.append("coverage must read the recorded archive md5 from persisted state") - if not _has_reachable_compare( + if not _has_exact_false_rejection( state_matches, - operator=ast.NotIn, - terms=("archive_id", "surviving_archives"), + condition="not isinstance(archive_id, str) or archive_id not in surviving_archives", parents=parents, ): errors.append("coverage must reject archive IDs absent from the live Drive inventory") - if not _has_reachable_compare( + if not _has_exact_false_rejection( state_matches, - operator=ast.NotEq, - terms=("live_md5", "recorded_md5"), + condition="not isinstance(live_md5, str) or live_md5 != recorded_md5", parents=parents, ): errors.append("coverage must reject a surviving Drive object whose archived bytes changed") @@ -253,7 +285,13 @@ def inspect_jsonl_retention_invariant(source: str, *, backup_daily_source: str) verified_gates = [ node for node in ast.walk(run_backup) if isinstance(node, ast.If) and _is_exact_verified_upload_gate(node) ] - verified_gate = verified_gates[0] if len(verified_gates) == 1 else None + verified_gate = verified_gates[0] if len(verified_gates) == 1 and verified_gates[0] in run_backup.body else None + verification_indexes = [ + index for index, statement in enumerate(run_backup.body) if _is_verification_update(statement) + ] + gate_index = run_backup.body.index(verified_gate) if verified_gate is not None else None + if len(verification_indexes) != 1 or gate_index != verification_indexes[0] + 1: + errors.append("verified-upload deletion gate must consume the bundle verification result without override") safe_calls = _calls_in_statements(verified_gate.body) if verified_gate is not None else [] persistence: list[tuple[int, ast.Call, ast.Call]] = [] diff --git a/tests/test_jsonl_backup.py b/tests/test_jsonl_backup.py index 3d0f2d7f..4e576c63 100644 --- a/tests/test_jsonl_backup.py +++ b/tests/test_jsonl_backup.py @@ -30,6 +30,11 @@ def test_jsonl_retention_invariant_is_a_ci_guard_not_only_a_behavior_fixture(): "archive_id in surviving_archives", "coverage must reject archive IDs absent from the live Drive inventory", ), + ( + "archive_id not in surviving_archives", + "archive_id not in surviving_archives and False", + "coverage must reject archive IDs absent from the live Drive inventory", + ), ( "live_md5 != recorded_md5", "live_md5 == recorded_md5", @@ -100,6 +105,18 @@ def test_jsonl_retention_invariant_is_a_ci_guard_not_only_a_behavior_fixture(): unsafe = source.replace(original, weakened, 1) assert expected_error in inspect_jsonl_retention_invariant(unsafe, backup_daily_source=backup_daily_source) + unsafe = source.replace( + "result.update(verify_jsonl_bundle(archive_path, expected_file_count=len(changed)))", + ( + "result.update(verify_jsonl_bundle(archive_path, expected_file_count=len(changed)))\n" + ' result["verified"] = True' + ), + 1, + ) + assert "verified-upload deletion gate must consume the bundle verification result without override" in ( + inspect_jsonl_retention_invariant(unsafe, backup_daily_source=backup_daily_source) + ) + unsafe = source.replace( 'archive_md5=uploaded.get("md5Checksum")', "archive_md5=None", From 9f61371b4337d01cfa38d967ee4be2a3a784565e Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Wed, 9 Sep 2026 21:49:17 +0300 Subject: [PATCH 4/5] test(prepush): execute guard mutations from owner (XS) Co-Authored-By: brainlayerCodex-4c4386f6 --- tests/test_backup_retention_invariant.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/test_backup_retention_invariant.py b/tests/test_backup_retention_invariant.py index 332d0164..e394c750 100644 --- a/tests/test_backup_retention_invariant.py +++ b/tests/test_backup_retention_invariant.py @@ -3,6 +3,9 @@ def test_jsonl_retention_guard_failure_tells_refactors_to_update_not_delete(tmp_path: Path, capsys) -> None: from brainlayer.backup_retention_invariant import main + from tests import test_jsonl_backup + + test_jsonl_backup.test_jsonl_retention_invariant_is_a_ci_guard_not_only_a_behavior_fixture() unsafe_path = tmp_path / "jsonl_backup.py" unsafe_path.write_text("def run_backup():\n pass\n", encoding="utf-8") From 96e9afecf6a0a4cc5df5f7e7494565fa337ef6ee Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Wed, 9 Sep 2026 23:13:03 +0300 Subject: [PATCH 5/5] test(backup): keep guard mutations valid after #819 (XS) Co-Authored-By: brainlayerCodex-4c4386f6 --- tests/test_jsonl_backup.py | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/tests/test_jsonl_backup.py b/tests/test_jsonl_backup.py index 731258b9..64b5347e 100644 --- a/tests/test_jsonl_backup.py +++ b/tests/test_jsonl_backup.py @@ -79,7 +79,7 @@ def test_jsonl_retention_invariant_is_a_ci_guard_not_only_a_behavior_fixture(): ), ( 'if result["verified"] and upload:', - 'if result["verified"] and upload:\n pass\n else:', + ('if result["verified"] and upload:\n return result\n if result["verified"] and upload:'), "backup deletion calls must remain inside verified-upload control flow", ), ( @@ -107,12 +107,11 @@ def test_jsonl_retention_invariant_is_a_ci_guard_not_only_a_behavior_fixture(): unsafe = source.replace(original, weakened, 1) assert expected_error in inspect_jsonl_retention_invariant(unsafe, backup_daily_source=backup_daily_source) + verified_upload_gate = ' if result["verified"] and upload:' + assert verified_upload_gate in source, "mutation fixture drifted: verified-upload gate" unsafe = source.replace( - "result.update(verify_jsonl_bundle(archive_path, expected_file_count=len(changed)))", - ( - "result.update(verify_jsonl_bundle(archive_path, expected_file_count=len(changed)))\n" - ' result["verified"] = True' - ), + verified_upload_gate, + f' result["verified"] = True\n{verified_upload_gate}', 1, ) assert "verified-upload deletion gate must consume the bundle verification result without override" in (