[ADC-691] Prove complete IMEX-AMR lowering and rollback

docs/design/final-advection-imex-amr.md

@@ -28,8 +28,12 @@ property of this graph and tableau, not a repeated `order=2` option.
 
 Every fallible public solve returns an unreadable `SolveOutcome`. The example consumes every field solve with
 `RejectAttempt()`. A failed solve therefore raises the typed native rejection signal before a field,
-state, diagnostic or output can read a partial result. Local affine elimination remains a value
-operation because it has no iterative outcome to classify.
+state, diagnostic or output can read a partial result. The executable acceptance also compiles a
+separate negative case whose explicitly widened parameter domain makes the second IMEX diagonal
+system exactly singular. It compares state, solved fields, hierarchy topology, the canonical opaque
+Program accepted-state image, Program cache/history/clock/ledger registries and consumer cursors
+before and after the rejected attempt, then requires that its output directory contain no file. The
+normal physical case retains the strictly positive relaxation-rate domain.
 
 `Model.field_operator(...)` declares the physical equation and its RHS providers. The sole callable
 time-Program authority is the `FieldHandle` returned by `Case.field(operator, discretization)`: both
@@ -43,6 +47,7 @@ The AMR Program driver owns the accepted-state boundary. A hierarchy attempt sta
 - level state and clocks;
 - coarse/fine flux ledgers and reflux contributions;
 - history rings and their flux publications;
+- persistent AMR tagging hysteresis state;
 - regrid-dependent synchronization state;
 - field materializations and consumer schedule cursors.
 
@@ -60,15 +65,21 @@ The adaptive layout owns:
   subcycled execution; this is the installed provider's executable composite-field envelope;
 - strict above/below refinement and coarsening predicates;
 - a discrete gradient predicate resolved against the selected FV stencil;
-- explicit hysteresis/equality/conflict semantics;
+- two-cycle persistent hysteresis plus explicit equality/conflict semantics;
 - conservative state prolongation, restriction, coarse/fine fill and time interpolation;
 - elliptic recomputation after regrid instead of interpolating a stale solved field.
 
+Resolution adds each hierarchy, regrid, tagging predicate, hysteresis/conflict policy, transfer
+entry, bootstrap authority and subcycling relation to the global `LoweringCoverageReport`. The
+non-zero hysteresis row names its Program accepted-state persistence route as well as the native
+tagger. Every row therefore names a concrete runtime target; the report is a machine-readable
+lowering gate rather than an `inspect()` narrative inferred after compilation.
+
 The acceptance target intentionally requests a regrid on every accepted macro-step. The first
 snapshot may still expose zero completed regrids: cadence is a due condition, not proof that a
 non-empty tag set rebuilt the hierarchy. `simulation.amr.explain_regrid()` publishes the native
 `regrid_count` and `topology_epoch`; after the continuation step the example requires both values to
-have increased, and requires the uninterrupted and restarted instances to report identical values.
+remain monotone, and requires the uninterrupted and restarted instances to report identical values.
 `regrid_count` advances only after the native regrid completes, while `topology_epoch` identifies
 the installed hierarchy topology. A scheduled or no-op regrid is therefore never accepted as
 completed runtime evidence.
@@ -105,10 +116,13 @@ python examples/final/EXEMPLE_SPEC_FINALE_ADVECTION_IMEX_AMR.py --output-dir /tm
 The command reopens the emitted HDF5 and ParaView files, retains a real accepted-state checkpoint and
 restarts a fresh bound simulation from it. It compares time, macro-step, every AMR level of every
 qualified conservative state and solved-field route, patch topology, Program/consumer identities and
-consumer cursors bit-for-bit. The snapshot also carries the live completed-regrid count and topology
-epoch, so checkpoint restore, uninterrupted/restarted continuation and manual/factory parity must
-preserve exactly the same AMR generation evidence. It then advances the uninterrupted and restarted
-instances once more, requires a real counter/epoch increase, verifies the accepted multi-level flux
+consumer cursors bit-for-bit. It also compares the complete opaque Program accepted-state bytes,
+which include the persistent tagging history without duplicating its native codec in Python. The
+snapshot carries the live completed-regrid count and topology epoch, so checkpoint restore,
+uninterrupted/restarted continuation and manual/factory parity must preserve exactly the same AMR
+generation evidence. It then advances the uninterrupted and restarted
+instances once more, requires monotone counter/epoch evidence, verifies the accepted multi-level flux
 ledger plus reflux-then-average-down trace, and repeats the complete comparison before exercising the
-preset parity run. A printed success therefore follows real I/O, a completed regrid, restart,
-continuation and manual/factory checks; it is not a demonstration placeholder.
+preset parity run. A printed success therefore follows a real rejected-attempt rollback, real I/O,
+an executed regrid cadence window, restart, continuation and manual/factory checks; it is not a
+demonstration placeholder.

examples/final/EXEMPLE_SPEC_FINALE_ADVECTION_IMEX_AMR.py

@@ -98,6 +98,7 @@ def _bind_artifact(artifact: Any, **inputs: Any) -> Any:
     implicit_c=(Fraction(0), Fraction(1)),
     name="cn-heun-imex",
 )
+HYSTERESIS_MIN_CYCLES = 2
 
 
 @dataclass(frozen=True, slots=True)
@@ -148,6 +149,8 @@ class IMEXRuntimeSnapshot:
     regrid_count: int
     topology_epoch: int
     program_hash: str
+    program_accepted_state: bytes
+    program_transaction_state: str
     consumer_graph_identity: str
     consumer_cursors: dict[str, Any]
 
@@ -160,6 +163,15 @@ class IMEXAMRProgramEvidence:
     synchronization_phases: tuple[str, ...]
 
 
+@dataclass(frozen=True, slots=True)
+class IMEXRejectedAttemptEvidence:
+    """Exact proof that one consumed SolveOutcome rolled back without publication."""
+
+    error: str
+    before: IMEXRuntimeSnapshot
+    after: IMEXRuntimeSnapshot
+
+
 @dataclass(frozen=True, slots=True)
 class IMEXExecutionEvidence:
     """Artifacts plus exact pre/post-restart and continuation snapshots."""
@@ -287,7 +299,10 @@ def _preset_imex_program(core: IMEXAMRAuthoring, *, solve_action: Any) -> Progra
 
 
 def build_authoring(
-    *, use_preset: bool = False, field_solver: Any | None = None,
+    *,
+    use_preset: bool = False,
+    field_solver: Any | None = None,
+    relaxation_domain: Any | None = None,
 ) -> IMEXAMRAuthoring:
     domain = Rectangle(
         "unit_square",
@@ -310,7 +325,11 @@ def build_authoring(
     # the incoming subspace; a static boundary table must not silently pretend to support them.
     velocity_x = model.param(RuntimeParam("a_x", default=1.0, domain=Positive()))
     velocity_y = model.param(RuntimeParam("a_y", default=0.25, domain=Positive()))
-    relaxation_rate = model.param(RuntimeParam("lambda", default=50.0, domain=Positive()))
+    relaxation_rate = model.param(RuntimeParam(
+        "lambda",
+        default=50.0,
+        domain=Positive() if relaxation_domain is None else relaxation_domain,
+    ))
     inlet_value = model.param(RuntimeParam("u_in", default=0.0, domain=Interval(-10.0, 10.0)))
     a_x = model.value(velocity_x)
     a_y = model.value(velocity_y)
@@ -483,9 +502,13 @@ def build_layout(core: IMEXAMRAuthoring) -> Any:
             Coarsen(value < core.case.value(core.coarsen_value)),
             Buffer(cells=2),
         ),
-        # Equality is explicit. A non-zero temporal dwell requires a checkpointed per-cell tagging
-        # state provider; this example does not pretend that an in-memory counter is restart-safe.
-        hysteresis=Hysteresis(min_cycles=0, equality=EqualityPolicy.HOLD),
+        # Keep one full tagging cycle between opposite decisions. The native Program accepted-state
+        # image owns this sparse, topology-independent history, so rejection and strict restart
+        # restore the same hysteresis authority instead of resetting an in-memory Python counter.
+        hysteresis=Hysteresis(
+            min_cycles=HYSTERESIS_MIN_CYCLES,
+            equality=EqualityPolicy.HOLD,
+        ),
         conflict_policy=ConflictPolicy.REFINE_WINS,
     )
     transfer = AMRTransfer()
@@ -538,10 +561,15 @@ def build_consumers(core: IMEXAMRAuthoring, *, output_mode: Any = None) -> Any:
 
 def build_final_case(
     *, use_preset: bool = False, field_solver: Any | None = None,
+    relaxation_domain: Any | None = None,
     initial_background: float = 0.05, initial_amplitude: float = 0.95,
     output_mode: Any = None,
 ) -> FinalIMEXAMRCase:
-    core = build_authoring(use_preset=use_preset, field_solver=field_solver)
+    core = build_authoring(
+        use_preset=use_preset,
+        field_solver=field_solver,
+        relaxation_domain=relaxation_domain,
+    )
     core.numerics.boundaries.add(build_boundaries(core))
     core.case.numerics(core.numerics, block=core.tracer)
     core.case.initials.add(build_initial(
@@ -551,12 +579,17 @@ def build_final_case(
     return FinalIMEXAMRCase(core, build_layout(core))
 
 
-def build_bind_params(core: IMEXAMRAuthoring, *, inlet_value: float = 0.0) -> dict[Any, float]:
+def build_bind_params(
+    core: IMEXAMRAuthoring,
+    *,
+    inlet_value: float = 0.0,
+    relaxation_rate: float = 50.0,
+) -> dict[Any, float]:
     resolve = core.case.resolve
     return {
         resolve(core.velocity_x): 1.0,
         resolve(core.velocity_y): 0.25,
-        resolve(core.relaxation_rate): 50.0,
+        resolve(core.relaxation_rate): relaxation_rate,
         resolve(core.inlet_value): inlet_value,
         resolve(core.refine_value): 0.70,
         resolve(core.coarsen_value): 0.25,
@@ -566,16 +599,35 @@ def build_bind_params(core: IMEXAMRAuthoring, *, inlet_value: float = 0.0) -> di
 
 def compile_final_case(
     *, use_preset: bool = False,
+    relaxation_domain: Any | None = None,
 ) -> tuple[FinalIMEXAMRCase, Any, Any]:
     """Compile one exact manual or preset-authored target through the public lifecycle."""
 
     target = build_final_case(
-        use_preset=use_preset, output_mode=_native_output_mode()
+        use_preset=use_preset,
+        relaxation_domain=relaxation_domain,
+        output_mode=_native_output_mode(),
     )
     resolved = pops.resolve(pops.validate(target.authoring.case), layout=target.layout)
     return target, resolved, pops.compile(resolved)
 
 
+def _program_transaction_state(simulation: Any) -> str:
+    """Canonicalize every rollback-sensitive Program registry without field arrays."""
+
+    report = simulation.program_report().to_dict()
+    return json.dumps({
+        "cache": report["cache"],
+        "clocks": report["clocks"],
+        "diagnostics": report["diagnostics"],
+        "flux_ledger": report["flux_ledger"],
+        "histories": report["histories"],
+        "level_relations": report["level_relations"],
+        "synchronization": report["synchronization"],
+        "temporal": report["temporal"],
+    }, sort_keys=True, separators=(",", ":"))
+
+
 def _snapshot(simulation: Any) -> IMEXRuntimeSnapshot:
     """Capture state, solved fields, hierarchy, clocks, identities and consumer cursors."""
 
@@ -594,6 +646,9 @@ def _snapshot(simulation: Any) -> IMEXRuntimeSnapshot:
     }
     if any(count <= 0 for count in field_level_counts.values()):
         raise RuntimeError("IMEX acceptance installed an empty diagnostic-field hierarchy")
+    program_accepted_state = bytes(simulation.program_accepted_state())
+    if not program_accepted_state:
+        raise RuntimeError("IMEX acceptance installed no canonical Program accepted-state image")
     regrid = simulation.amr.explain_regrid()
     return IMEXRuntimeSnapshot(
         time=float(simulation.time()),
@@ -623,6 +678,8 @@ def _snapshot(simulation: Any) -> IMEXRuntimeSnapshot:
         regrid_count=int(regrid.regrid_count),
         topology_epoch=int(regrid.topology_epoch),
         program_hash=str(simulation.installed_program_hash()),
+        program_accepted_state=program_accepted_state,
+        program_transaction_state=_program_transaction_state(simulation),
         consumer_graph_identity=simulation.consumer_graph.identity.token,
         consumer_cursors=simulation.consumer_cursors.to_data(),
     )
@@ -710,6 +767,14 @@ def _require_same_snapshot(
         "regrid_count": (left.regrid_count, right.regrid_count),
         "topology_epoch": (left.topology_epoch, right.topology_epoch),
         "program_hash": (left.program_hash, right.program_hash),
+        "program_accepted_state": (
+            left.program_accepted_state,
+            right.program_accepted_state,
+        ),
+        "program_transaction_state": (
+            left.program_transaction_state,
+            right.program_transaction_state,
+        ),
         "consumer_graph_identity": (
             left.consumer_graph_identity,
             right.consumer_graph_identity,
@@ -792,6 +857,54 @@ def _reopen_scientific_outputs(root: Path) -> tuple[Path, Path, str, str]:
     )
 
 
+def run_rejected_attempt_rollback(output_dir: Any) -> IMEXRejectedAttemptEvidence:
+    """Force one singular IMEX solve and prove complete rollback before publication."""
+
+    root = Path(output_dir)
+    if root.exists() and any(root.iterdir()):
+        raise ValueError("rejected-attempt proof requires an empty output directory")
+    target, _resolved, artifact = compile_final_case(
+        use_preset=False,
+        relaxation_domain=Interval(-1.0e6, 1.0e6),
+    )
+    first_dt = float(target.authoring.run_controls["t_end"])
+    diagonal = float(IMEX_CN_HEUN.implicit_A[1][1])
+    singular_rate = -1.0 / (first_dt * diagonal)
+    simulation = _bind_artifact(
+        artifact,
+        params=build_bind_params(
+            target.authoring,
+            relaxation_rate=singular_rate,
+        ),
+    )
+    before = _snapshot(simulation)
+    try:
+        pops.run(
+            simulation,
+            t_end=first_dt,
+            max_steps=1,
+            output_dir=root,
+        )
+    except RuntimeError as error:
+        message = str(error)
+        if not message.startswith("step attempt rejected during "):
+            raise RuntimeError(
+                "negative IMEX proof failed for an unexpected reason: %s" % message
+            ) from error
+    else:
+        raise RuntimeError("singular IMEX solve unexpectedly accepted its macro-step")
+
+    after = _snapshot(simulation)
+    _require_same_snapshot(before, after, where="rejected IMEX attempt")
+    leaked = tuple(path for path in root.rglob("*") if path.is_file())
+    if leaked:
+        raise RuntimeError(
+            "rejected IMEX attempt published files: %s"
+            % ", ".join(str(path) for path in leaked)
+        )
+    return IMEXRejectedAttemptEvidence(message, before, after)
+
+
 def run_manual_and_restart(output_dir: Any) -> IMEXExecutionEvidence:
     """Run the manual Program, reopen output, restart fresh, then continue bit-identically."""
 
@@ -916,6 +1029,7 @@ def main(argv: list[str] | None = None) -> None:
     args = parser.parse_args(argv)
     output_dir = args.output_dir.resolve()
 
+    rejected = run_rejected_attempt_rollback(output_dir / "rejected")
     evidence = run_manual_and_restart(output_dir / "manual")
     preset = run_preset_parity(output_dir / "preset", evidence.accepted)
     restart_equal = _snapshots_bit_identical(evidence.accepted, evidence.restored)
@@ -940,6 +1054,9 @@ def main(argv: list[str] | None = None) -> None:
     print("bit-identical restart: %s" % restart_equal)
     print("bit-identical continuation: %s" % continuation_equal)
     print("manual/pops.lib.time.IMEX parity: %s" % preset_equal)
+    print("rejected-attempt rollback: %s" % _snapshots_bit_identical(
+        rejected.before, rejected.after,
+    ))
     print(
         "regrid count: %d -> %d (topology epoch %d -> %d)"
         % (
@@ -957,7 +1074,13 @@ def main(argv: list[str] | None = None) -> None:
         "flux_ledger_levels": list(evidence.program_evidence.flux_ledger_levels),
         "levels": evidence.level_count,
         "manual_preset_bit_identical": preset_equal,
+        "rejected_attempt_error": rejected.error,
+        "rejected_attempt_rollback": _snapshots_bit_identical(
+            rejected.before, rejected.after,
+        ),
         "program_hash": preset.program_hash,
+        "program_accepted_state_bytes": len(preset.program_accepted_state),
+        "tagging_hysteresis_min_cycles": HYSTERESIS_MIN_CYCLES,
         "regrid_count": evidence.accepted.regrid_count,
         "regrid_count_after_continuation": evidence.restarted.regrid_count,
         "runtime_steps": evidence.accepted.macro_step,

examples/final/README.md

@@ -6,8 +6,9 @@ concern and no fallback to an older or lower-level API.
 
 [`EXEMPLE_SPEC_FINALE_ADVECTION_IMEX_AMR.py`](EXEMPLE_SPEC_FINALE_ADVECTION_IMEX_AMR.py)
 extends the same public lifecycle with an explicit additive IMEX tableau, typed field solves,
-two-level subcycled AMR, conservative transfers and accepted-state consumers. Its matching
-contract note is
+two-level subcycled AMR, conservative transfers, globally reported AMR lowering coverage, an
+executed rejected-attempt rollback proof, persistent tagging hysteresis and accepted-state consumers.
+Its matching contract note is
 [`docs/design/final-advection-imex-amr.md`](../../docs/design/final-advection-imex-amr.md).
 
 [`EXEMPLE_SPEC_FINALE_15_MOMENTS_HYQMOM.py`](EXEMPLE_SPEC_FINALE_15_MOMENTS_HYQMOM.py)