Add checkpoint recovery, CI evidence checks, and proof artifacts

.github/workflows/verify.yml

@@ -46,6 +46,10 @@ jobs:
         run: python3 proofs/agda/verify.py --ci
       - name: Exercise water-warning continuity and acknowledgement scenarios
         run: mix run -e 'Firmboot.WaterLeak.Experiment.run()'
+      - name: Exercise journal recovery, gaps and measured resource use
+        run: mix run -e 'Firmboot.WaterLeak.RecoveryExperiment.run(".ci-results/recovery")'
+      - name: Check the assurance register and its negative controls
+        run: python3 scripts/assurance.py check --ci --output .ci-results/assurance-check.json
       - name: Ensure observed witnesses are reproducible
         run: |
           git -c safe.directory="$GITHUB_WORKSPACE" -C "$GITHUB_WORKSPACE" rev-parse --show-toplevel
@@ -55,7 +59,7 @@ jobs:
         uses: actions/upload-artifact@v7.0.1
         with:
           name: agda-elixir-verification
-          path: .ci-results/agda
+          path: .ci-results
           include-hidden-files: true
           if-no-files-found: error
           retention-days: 30
@@ -84,7 +88,20 @@ jobs:
           tar --zstd -xf "$RUNNER_TEMP/firmboot-lean.tar.zst" --strip-components=1 -C "$FIRMBOOT_LEAN_DIR"
           echo "$FIRMBOOT_LEAN_DIR/bin" >> "$GITHUB_PATH"
       - name: Build all Lean proofs, audit axioms and reject false claims
-        run: bash proofs/lean/verify.sh
+        shell: bash
+        run: |
+          set -o pipefail
+          mkdir -p .ci-results/lean
+          bash proofs/lean/verify.sh | tee .ci-results/lean/verify.log
+      - name: Retain the complete Lean proof audit log
+        if: always()
+        uses: actions/upload-artifact@v7.0.1
+        with:
+          name: lean-verification
+          path: .ci-results/lean
+          include-hidden-files: true
+          if-no-files-found: error
+          retention-days: 30
 
   actions-lock:
     name: Actions dependency lock

lib/firmboot/water_leak/recovery_experiment.ex

@@ -0,0 +1,142 @@
+defmodule Firmboot.WaterLeak.RecoveryExperiment do
+  @moduledoc """
+  Replays a finite water-warning journal from a versioned checkpoint.
+
+  The checkpoint is a local experiment artefact, not a durable storage format.
+  Resource evidence is limited to its encoded byte size and the BEAM reductions
+  used by one replay; it is not a wall-clock or deployment bound.
+  """
+  alias Firmboot.{Update, WaterLeak}
+  alias Firmboot.WaterLeak.{Checker, Experiment}
+
+  @checkpoint_version 1
+  @first_warning {"sim-loop", 3}
+
+  def journal do
+    Enum.flat_map(Experiment.readings(), fn reading ->
+      before_reading =
+        if reading.seq == 5 do
+          [
+            {:request,
+             %Update{
+               id: :water_window_upgrade,
+               target: :v2,
+               prepare_ticks: 1
+             }}
+          ]
+        else
+          []
+        end
+
+      after_reading =
+        if reading.seq == 7,
+          do: [{:acknowledge, @first_warning, "operator-A"}],
+          else: []
+
+      before_reading ++ [{:reading, reading}] ++ after_reading
+    end)
+  end
+
+  def recover(entries) when is_list(entries) do
+    entries
+    |> Enum.with_index(1)
+    |> Enum.reduce_while({:ok, WaterLeak.new(Experiment.contract())}, fn {entry, index},
+                                                                         {:ok, state} ->
+      case replay(state, entry) do
+        {:ok, next} -> {:cont, {:ok, next}}
+        {:error, reason} -> {:halt, {:error, %{entry: index, reason: reason}}}
+      end
+    end)
+  end
+
+  def recover(_entries), do: {:error, %{entry: 0, reason: :invalid_journal}}
+
+  def checkpoint(entries) when is_list(entries) do
+    :erlang.term_to_binary({:firmboot_water_leak_journal, @checkpoint_version, entries}, [
+      :deterministic
+    ])
+  end
+
+  def recover_checkpoint(binary) when is_binary(binary) do
+    case :erlang.binary_to_term(binary, [:safe]) do
+      {:firmboot_water_leak_journal, @checkpoint_version, entries} -> recover(entries)
+      _other -> {:error, %{entry: 0, reason: :invalid_checkpoint}}
+    end
+  rescue
+    ArgumentError -> {:error, %{entry: 0, reason: :invalid_checkpoint}}
+  end
+
+  def recover_checkpoint(_binary),
+    do: {:error, %{entry: 0, reason: :invalid_checkpoint}}
+
+  def run(output_dir) when is_binary(output_dir) do
+    File.mkdir_p!(output_dir)
+    entries = journal()
+    encoded = checkpoint(entries)
+    checkpoint_path = Path.join(output_dir, "journal.checkpoint")
+    File.write!(checkpoint_path, encoded)
+
+    {before_reductions, _} = process_info(:reductions)
+    {:ok, recovered} = checkpoint_path |> File.read!() |> recover_checkpoint()
+    {after_reductions, _} = process_info(:reductions)
+    replay_reductions = after_reductions - before_reductions
+
+    {expected, receipts, _snapshots} = Experiment.scenario()
+    :ok = Checker.verify(Experiment.readings(), receipts, recovered, Experiment.contract())
+    true = recovered == expected
+
+    {:error, %{reason: {:expected_sequence, 6}}} = recover(with_gap(entries))
+
+    truncated = binary_part(encoded, 0, byte_size(encoded) - 1)
+    {:error, %{reason: :invalid_checkpoint}} = recover_checkpoint(truncated)
+
+    report = """
+    {
+      "success": true,
+      "journal_entries": #{length(entries)},
+      "checkpoint_bytes": #{byte_size(encoded)},
+      "replay_reductions": #{replay_reductions},
+      "checks": {
+        "checkpoint_round_trip": true,
+        "exact_replay": true,
+        "gap_rejected": true,
+        "truncation_rejected": true
+      }
+    }
+    """
+
+    File.write!(Path.join(output_dir, "verification.json"), report)
+
+    IO.puts(
+      "PASS: recovered #{length(entries)} journal entries from #{byte_size(encoded)} bytes " <>
+        "using #{replay_reductions} BEAM reductions"
+    )
+
+    IO.puts("PASS: rejected a sequence gap and a truncated checkpoint")
+    :ok
+  end
+
+  defp replay(state, {:request, %Update{} = update}),
+    do: {:ok, WaterLeak.request(state, update)}
+
+  defp replay(state, {:reading, reading}), do: WaterLeak.sample(state, reading)
+
+  defp replay(state, {:acknowledge, id, operator}),
+    do: WaterLeak.acknowledge(state, id, operator)
+
+  defp replay(_state, _entry), do: {:error, :invalid_journal_entry}
+
+  defp with_gap(entries) do
+    Enum.map(entries, fn
+      {:reading, %{seq: 6} = reading} -> {:reading, %{reading | seq: 7}}
+      entry -> entry
+    end)
+  end
+
+  defp process_info(item) do
+    case Process.info(self(), item) do
+      {^item, value} -> {value, item}
+      nil -> raise "current BEAM process is unavailable"
+    end
+  end
+end

scripts/assurance.py

@@ -0,0 +1,132 @@
+#!/usr/bin/env python3
+# SPDX-FileCopyrightText: 2026 Jonathan D.A. Jewell <j.d.a.jewell@open.ac.uk>
+# SPDX-License-Identifier: MPL-2.0
+"""Check the CI evidence register and prove that planted bad evidence is rejected."""
+
+import argparse
+import copy
+import json
+from pathlib import Path
+
+
+PROJECT = Path(__file__).resolve().parents[1]
+
+
+def evaluate(documents):
+    agda = documents.get("agda", {})
+    recovery = documents.get("recovery", {})
+    agda = agda if isinstance(agda, dict) else {}
+    recovery = recovery if isinstance(recovery, dict) else {}
+    checks = recovery.get("checks", {})
+    checks = checks if isinstance(checks, dict) else {}
+
+    return {
+        "agda_contract": agda.get("success") is True,
+        "nonempty_runtime_suite": (
+            isinstance(agda.get("runtime_tests_passed"), int)
+            and not isinstance(agda.get("runtime_tests_passed"), bool)
+            and agda["runtime_tests_passed"] > 0
+        ),
+        "all_elixir_witnesses": agda.get("elixir_witnesses_checked") == 64,
+        "recovery_experiment": recovery.get("success") is True,
+        "checkpoint_round_trip": checks.get("checkpoint_round_trip") is True,
+        "exact_replay": checks.get("exact_replay") is True,
+        "gap_control": checks.get("gap_rejected") is True,
+        "truncation_control": checks.get("truncation_rejected") is True,
+        "checkpoint_measured": (
+            isinstance(recovery.get("checkpoint_bytes"), int)
+            and not isinstance(recovery.get("checkpoint_bytes"), bool)
+            and recovery["checkpoint_bytes"] > 0
+        ),
+        "replay_measured": (
+            isinstance(recovery.get("replay_reductions"), int)
+            and not isinstance(recovery.get("replay_reductions"), bool)
+            and recovery["replay_reductions"] > 0
+        ),
+    }
+
+
+def negative_controls(documents):
+    controls = []
+
+    for name, mutate in (
+        ("failed-agda", lambda data: data["agda"].update(success=False)),
+        (
+            "empty-runtime-suite",
+            lambda data: data["agda"].update(runtime_tests_passed=0),
+        ),
+        (
+            "missing-elixir-witness",
+            lambda data: data["agda"].update(elixir_witnesses_checked=63),
+        ),
+        (
+            "failed-recovery",
+            lambda data: data["recovery"].update(success=False),
+        ),
+        (
+            "unchecked-gap",
+            lambda data: data["recovery"]["checks"].update(gap_rejected=False),
+        ),
+        (
+            "unmeasured-checkpoint",
+            lambda data: data["recovery"].update(checkpoint_bytes=0),
+        ),
+    ):
+        planted = copy.deepcopy(documents)
+        mutate(planted)
+        controls.append({"name": name, "rejected": not all(evaluate(planted).values())})
+
+    return controls
+
+
+def load_evidence(results_dir):
+    paths = {
+        "agda": results_dir / "agda" / "verification.json",
+        "recovery": results_dir / "recovery" / "verification.json",
+    }
+    return {name: json.loads(path.read_text()) for name, path in paths.items()}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("command", choices=("check",))
+    parser.add_argument("--ci", action="store_true")
+    parser.add_argument("--output", type=Path, required=True)
+    parser.add_argument("--results-dir", type=Path, default=PROJECT / ".ci-results")
+    args = parser.parse_args()
+
+    report = {"ci": args.ci, "requirements": {}, "negative_controls": [], "success": False}
+
+    try:
+        documents = load_evidence(args.results_dir.resolve())
+        report["requirements"] = evaluate(documents)
+
+        if not all(report["requirements"].values()):
+            raise RuntimeError("one or more registered assurance requirements failed")
+
+        report["negative_controls"] = negative_controls(documents)
+
+        if not all(control["rejected"] for control in report["negative_controls"]):
+            raise RuntimeError("an assurance negative control was not rejected")
+
+        report["success"] = True
+    except (OSError, ValueError, KeyError, TypeError, RuntimeError) as error:
+        report["error"] = str(error)
+
+    output = args.output.resolve()
+    output.parent.mkdir(parents=True, exist_ok=True)
+    output.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n")
+
+    if report["success"]:
+        print(
+            f"PASS: {len(report['requirements'])} assurance requirements and "
+            f"{len(report['negative_controls'])} negative controls"
+        )
+        return 0
+
+    print(f"FAIL: {report.get('error', 'assurance register did not pass')}")
+    return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

test/firmboot/water_leak_test.exs

@@ -1,7 +1,7 @@
 defmodule Firmboot.WaterLeakTest do
   use ExUnit.Case, async: true
   alias Firmboot.{Update, WaterLeak}
-  alias Firmboot.WaterLeak.{Checker, Experiment}
+  alias Firmboot.WaterLeak.{Checker, Experiment, RecoveryExperiment}
 
   @contract [source: "sim-loop", threshold_ml: 100]
   @first {"sim-loop", 3}
@@ -148,6 +148,46 @@ defmodule Firmboot.WaterLeakTest do
     end
   end
 
+  test "a checkpoint replays exactly and rejects gaps or truncation" do
+    entries = RecoveryExperiment.journal()
+    {:ok, recovered} = RecoveryExperiment.recover(entries)
+    {expected, receipts, _snapshots} = Experiment.scenario()
+
+    assert recovered == expected
+    assert :ok == Checker.verify(Experiment.readings(), receipts, recovered, @contract)
+
+    checkpoint = RecoveryExperiment.checkpoint(entries)
+    assert {:ok, ^recovered} = RecoveryExperiment.recover_checkpoint(checkpoint)
+
+    gap =
+      Enum.map(entries, fn
+        {:reading, %{seq: 6} = reading} -> {:reading, %{reading | seq: 7}}
+        entry -> entry
+      end)
+
+    assert {:error, %{reason: {:expected_sequence, 6}}} = RecoveryExperiment.recover(gap)
+
+    truncated = binary_part(checkpoint, 0, byte_size(checkpoint) - 1)
+
+    assert {:error, %{reason: :invalid_checkpoint}} =
+             RecoveryExperiment.recover_checkpoint(truncated)
+  end
+
+  test "the recovery CLI records its checks and measured replay resources" do
+    tmp_dir =
+      Path.join(System.tmp_dir!(), "firmboot-recovery-#{System.unique_integer([:positive])}")
+
+    on_exit(fn -> File.rm_rf!(tmp_dir) end)
+
+    assert :ok == RecoveryExperiment.run(tmp_dir)
+    report = tmp_dir |> Path.join("verification.json") |> File.read!()
+
+    assert report =~ ~s("success": true)
+    assert report =~ ~s("gap_rejected": true)
+    assert report =~ ~s("truncation_rejected": true)
+    assert File.stat!(Path.join(tmp_dir, "journal.checkpoint")).size > 0
+  end
+
   defp assert_error(state, receipts, expected) do
     assert {:error, errors} = Checker.verify(Experiment.readings(), receipts, state, @contract)
     assert expected in errors