test: cover baseline sidecar crash recovery

active-runtime/app/release_metadata_adapter.py

@@ -247,17 +247,91 @@ def secure_baseline_authority_file(path: Path) -> None:
         raise RuntimeError("BASELINE_AUTHORITY_PERMISSIONS_INVALID")
 
 
+def baseline_sidecar_path(path: Path) -> Path:
+    return path.with_name("current.sha256")
+
+
+def secure_baseline_sidecar_file(path: Path) -> None:
+    if not path.is_file() or path.is_symlink():
+        raise RuntimeError("BASELINE_SIDECAR_FILE_INVALID")
+    sidecar_stat = path.lstat()
+    if sidecar_stat.st_uid != 0 or sidecar_stat.st_mode & 0o022:
+        raise RuntimeError("BASELINE_SIDECAR_PERMISSIONS_INVALID")
+
+
+def read_baseline_sidecar(path: Path) -> str | None:
+    if not path.exists() and not path.is_symlink():
+        return None
+    secure_baseline_sidecar_file(path)
+    fields = path.read_text(encoding="utf-8").strip().split()
+    if not fields or not SHA_RE.fullmatch(fields[0]):
+        raise RuntimeError("BASELINE_SIDECAR_DIGEST_INVALID")
+    return fields[0]
+
+
+def reconcile_baseline_sidecar(path: Path, baseline_digest: str) -> str:
+    if not SHA_RE.fullmatch(baseline_digest):
+        raise RuntimeError("BASELINE_DIGEST_INVALID")
+    sidecar = baseline_sidecar_path(path)
+    if sidecar.exists() or sidecar.is_symlink():
+        secure_baseline_sidecar_file(sidecar)
+        metadata = sidecar.lstat()
+        target_mode = stat.S_IMODE(metadata.st_mode)
+        target_uid = metadata.st_uid
+        target_gid = metadata.st_gid
+    else:
+        metadata = path.lstat()
+        target_mode = 0o440
+        target_uid = 0
+        target_gid = metadata.st_gid
+
+    fd, name = tempfile.mkstemp(prefix=".current.sha256.", dir=str(sidecar.parent))
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as output:
+            output.write(f"{baseline_digest}  current.json" + chr(10))
+            output.flush()
+            os.fsync(output.fileno())
+        os.chmod(name, target_mode)
+        if os.geteuid() == 0:
+            os.chown(name, target_uid, target_gid)
+        os.replace(name, sidecar)
+        dfd = os.open(sidecar.parent, os.O_RDONLY | os.O_DIRECTORY)
+        try:
+            os.fsync(dfd)
+        finally:
+            os.close(dfd)
+    finally:
+        try:
+            os.unlink(name)
+        except FileNotFoundError:
+            pass
+
+    if sha256_file(path) != baseline_digest:
+        raise RuntimeError("RECONCILIATION_REQUIRED_CONCURRENT_BASELINE_CHANGE")
+    if read_baseline_sidecar(sidecar) != baseline_digest:
+        raise RuntimeError("BASELINE_SIDECAR_READBACK_MISMATCH")
+    return baseline_digest
+
+
 def baseline_probe(args: argparse.Namespace, value: Mapping[str, Any]) -> dict[str, Any]:
     tag_oid, _ = verify_local_tag(args.repository, value)
     secure_baseline_authority_file(args.baseline_path)
     record = json.loads(args.baseline_path.read_text(encoding="utf-8"))
     if baseline_matches(record, value, tag_oid):
+        baseline_digest = sha256_file(args.baseline_path)
+        sidecar_digest = read_baseline_sidecar(baseline_sidecar_path(args.baseline_path))
+        if sidecar_digest != baseline_digest:
+            sidecar_digest = reconcile_baseline_sidecar(
+                args.baseline_path,
+                baseline_digest,
+            )
         return exact_response(
             value,
             "VERIFIED",
             tag_oid=tag_oid,
             baseline_path=str(args.baseline_path),
-            baseline_sha256=sha256_file(args.baseline_path),
+            baseline_sha256=baseline_digest,
+            baseline_sidecar_sha256=sidecar_digest,
         )
     current = record.get("current_baseline") or record.get("version")
     current_sha = sha256_file(args.baseline_path)
@@ -337,6 +411,8 @@ def baseline_apply(args: argparse.Namespace, value: Mapping[str, Any]) -> dict[s
         "promoted_by": "aios-approved-production-release-executor",
     }
     atomic_json(args.baseline_path, record)
+    baseline_digest = sha256_file(args.baseline_path)
+    reconcile_baseline_sidecar(args.baseline_path, baseline_digest)
     return baseline_probe(args, value)
 
 

tests/test_release_metadata_adapter.py

@@ -114,3 +114,81 @@ def test_atomic_json_preserves_existing_mode(tmp_path):
     assert after.st_mode & 0o777 == 0o440
     assert after.st_uid == before.st_uid
     assert after.st_gid == before.st_gid
+
+
+def baseline_args(tmp_path, baseline, previous_sha):
+    return type("A", (), {
+        "repository": tmp_path / "repo.git",
+        "registry_root": tmp_path / "versions",
+        "baseline_path": baseline,
+        "expected_previous_baseline": "V2R1C5",
+        "expected_previous_baseline_sha256": previous_sha,
+        "health_url": "http://127.0.0.1:18927/health",
+    })()
+
+
+def prepare_exact_release(tmp_path, monkeypatch):
+    baseline = tmp_path / "current.json"
+    previous_sha = old_baseline(baseline)
+    registry_root = tmp_path / "versions"
+    registry_root.mkdir()
+    args = baseline_args(tmp_path, baseline, previous_sha)
+    monkeypatch.setattr(adapter, "verify_local_tag", lambda *a: (TAG, TARGET))
+    monkeypatch.setattr(adapter, "secure_baseline_authority_file", lambda path: None)
+    monkeypatch.setattr(adapter, "health", lambda *a: {
+        "deployment_state": "ACTIVE",
+        "production_deployed": True,
+    })
+    registry = adapter.registry_record(value(), TAG, {"deployment_state": "ACTIVE"})
+    adapter.atomic_json(registry_root / "V2R1C7.json", registry)
+    assert adapter.baseline_apply(args, value())["status"] == "VERIFIED"
+    return args, baseline
+
+
+def test_body_write_updates_sidecar(tmp_path, monkeypatch):
+    args, baseline = prepare_exact_release(tmp_path, monkeypatch)
+    assert adapter.read_baseline_sidecar(baseline.with_name("current.sha256")) == adapter.sha256_file(baseline)
+
+
+def test_stale_sidecar_reconciles_idempotently(tmp_path, monkeypatch):
+    args, baseline = prepare_exact_release(tmp_path, monkeypatch)
+    sidecar = baseline.with_name("current.sha256")
+    sidecar.write_text("0" * 64 + "  current.json" + chr(10), encoding="utf-8")
+    before = baseline.read_bytes()
+    result = adapter.baseline_probe(args, value())
+    assert result["status"] == "VERIFIED"
+    assert adapter.read_baseline_sidecar(sidecar) == adapter.sha256_file(baseline)
+    assert baseline.read_bytes() == before
+
+
+def test_wrong_binding_never_repairs_sidecar(tmp_path, monkeypatch):
+    baseline = tmp_path / "current.json"
+    old_sha = old_baseline(baseline)
+    sidecar = baseline.with_name("current.sha256")
+    sidecar.write_text("1" * 64 + "  current.json" + chr(10), encoding="utf-8")
+    args = baseline_args(tmp_path, baseline, old_sha)
+    monkeypatch.setattr(adapter, "verify_local_tag", lambda *a: (TAG, TARGET))
+    monkeypatch.setattr(adapter, "secure_baseline_authority_file", lambda path: None)
+    before = sidecar.read_bytes()
+    result = adapter.baseline_probe(args, value())
+    assert result["status"] == "NOT_APPLIED"
+    assert sidecar.read_bytes() == before
+
+
+def test_body_unchanged_during_sidecar_repair(tmp_path, monkeypatch):
+    args, baseline = prepare_exact_release(tmp_path, monkeypatch)
+    sidecar = baseline.with_name("current.sha256")
+    sidecar.write_text("2" * 64 + "  current.json" + chr(10), encoding="utf-8")
+    before = hashlib.sha256(baseline.read_bytes()).hexdigest()
+    adapter.baseline_probe(args, value())
+    after = hashlib.sha256(baseline.read_bytes()).hexdigest()
+    assert after == before
+
+
+def test_crash_between_body_and_sidecar_recovers(tmp_path, monkeypatch):
+    args, baseline = prepare_exact_release(tmp_path, monkeypatch)
+    sidecar = baseline.with_name("current.sha256")
+    sidecar.unlink()
+    result = adapter.baseline_probe(args, value())
+    assert result["status"] == "VERIFIED"
+    assert adapter.read_baseline_sidecar(sidecar) == adapter.sha256_file(baseline)