Fix Adams failure semantics: restore state after a failed native source, record the rejection before its floor

changelog.d/adams-rejection-floor-row.fixed.md

@@ -0,0 +1,4 @@
+Fixed the Adams march dropping the attempt row of a rejected trial whose reduced
+proposal fell below `dz_min`: the completed trial is now recorded, on both the NumPy
+reference and the native backend, before the floor's `ConvergenceError` is raised; the
+proposal, accepted state and call ledger are unchanged (#651).

changelog.d/native-adams-source-failure.fixed.md

@@ -0,0 +1,3 @@
+Fixed a native Adams trial whose source evaluation failed leaving its prediction in the
+state, so that the failed run's terminal checkpoint and march snapshot held the
+prediction instead of the last accepted step (#129).

docs/dev/kernel-contract.md

@@ -1290,7 +1290,10 @@ ledger. A resume carries all of it exactly (`MarchSnapshot`).
 
 An acceptance commits the engine's state, the history, the ledger, the stored order, the
 position and a kink restart together. Only then is the next proposal tested against the
-floor. A floor so found is raised after the attempt is recorded.
+floor. A floor so found is raised after the attempt is recorded. A rejection's shrunk
+proposal is tested in the same way, and a floor it falls below leaves the proposal as
+it was and is raised after the rejected attempt is recorded (issue #651); a replay
+records no rejected attempt and raises it at once.
 
 **What a native call takes**: the policy, once per run and again when the monitor's
 interval changes (`MarchPolicy`, every field of the record), with a replay's

native/include/solprop_native/core.hpp

@@ -15,6 +15,7 @@
 #include <complex>
 #include <cstddef>
 #include <cstdint>
+#include <functional>
 #include <map>
 #include <memory>
 #include <span>
@@ -1174,6 +1175,9 @@ class GeneralState {
   [[nodiscard]] HostSource* host_source() const noexcept;
   /** The next host evaluation's position, m, and purpose, which the engine sets. */
   void host_evaluation(double z, HostPurpose purpose) noexcept;
+  /** A test hook (solprop issue #129): `fault` runs, once, as the next Adams source
+      evaluation starts, a trial's with its prediction in the state; empty disarms. */
+  void arm_source_fault(std::function<void()> fault) noexcept;
 
 private:
   /** `function(workspace, arguments...)` inside the workspace's CPU team scope

native/include/solprop_native/engines.hpp

@@ -34,12 +34,15 @@ namespace solprop_native::engines {
 /** A failure injected at one of the resident engine's stages, after that stage has run
     `after` more times: solprop's lifecycle tests fail the native march with it. With
     `signal`, that signal is raised in the native call instead and the stage goes on,
-    as a signal arriving while native code holds no interpreter lock does. */
+    as a signal arriving while native code holds no interpreter lock does. With
+    `in_source`, the failure is raised inside the stage's source evaluation instead,
+    a trial's with its prediction in the state (solprop issue #129). */
 struct MarchFault {
   march::Stage stage = march::Stage::none;
   std::size_t after = 0;
   march::Category category = march::Category::backend;
   int signal = 0;
+  bool in_source = false;
 };
 
 /** A scalar carrier frame's shift and rotations (solprop SEP 35), packed by solprop's

native/src/adams_march.cpp

@@ -372,6 +372,13 @@ bool reached(const double position, const double target, const int ulps) {
 
 namespace {
 
+/** End the march at the step floor that the proposal `next` fell below. */
+Result& floored(Result& result, const double next) {
+  result.reason = Reason::floor;
+  result.floor_proposal = next;
+  return result;
+}
+
 /** Record a native failure's stage, its solprop error class and its message. */
 void fail(Result& result, const Stage stage, const std::exception& error) {
   result.reason = Reason::error;
@@ -573,23 +580,31 @@ Result run(const Policy& policy, State& state, Engine& engine, const Bounds& bou
         ++state.ledger.failure_restarts;
       }
       const double next = proposal(policy, step.dz, outcome.factor);
-      if (next < policy.dz_min) {
-        // `limit_step` raises inside the rejection, before the attempt is recorded.
-        result.reason = Reason::floor;
-        result.floor_proposal = next;
-        return result;
-      }
-      state.proposal = next;
-      if (bounds.replay) {
-        // A replay cannot reject: the field has diverged (`replay_step`), and the
-        // attempt is not recorded.
+      const bool floor = next < policy.dz_min;
+      if (bounds.replay && !floor) {
+        // A replay cannot reject: its field has diverged (`replay_step`).
+        state.proposal = next;
         result.reason = Reason::diverged;
         result.replay_dz = step.dz;
         result.replay_end = step.end;
         result.replay_norm = norm;
         return result;
       }
+      if (floor) {
+        // A floor leaves the proposal; a replay's is raised unrecorded. Any other
+        // attempt is recorded first (`HostAdamsMarch._reject`, solprop issue #651);
+        // a failure of its release is the host's to keep beside the floor.
+        if (!bounds.replay) {
+          record(step.end, step.dz, norm, false, trial.order);
+          try {
+            attempt.end(HostOutcome::rejected);
+          } catch (...) { // NOLINT(bugprone-empty-catch) - kept by the host, see above
+          }
+        }
+        return floored(result, next);
+      }
       record(step.end, step.dz, norm, false, trial.order);
+      state.proposal = next;
       try {
         attempt.end(HostOutcome::rejected);
       } catch (const std::exception& error) {
@@ -652,11 +667,8 @@ Result run(const Policy& policy, State& state, Engine& engine, const Bounds& bou
         fail(result, Stage::accept, error);
         return result;
       }
-      if (next < policy.dz_min) {
-        result.reason = Reason::floor;
-        result.floor_proposal = next;
-        return result;
-      }
+      if (next < policy.dz_min)
+        return floored(result, next);
       state.proposal = next;
       ++accepted;
       if (bounds.replay)

native/src/bindings.cpp

@@ -2067,14 +2067,18 @@ class BoundState {
                    const std::string& category) {
     require_open();
     native::engines::MarchFault fault;
+    // `trial_source` fails inside the trial's source evaluation, its prediction in
+    // the state.
+    fault.in_source = stage == "trial_source";
     if (stage == "startup")
       fault.stage = march::Stage::startup;
-    else if (stage == "trial")
+    else if (stage == "trial" || fault.in_source)
       fault.stage = march::Stage::trial;
     else if (stage == "accept")
       fault.stage = march::Stage::accept;
     else
-      throw std::invalid_argument("a march fault's stage is startup, trial or accept");
+      throw std::invalid_argument(
+          "a march fault's stage is startup, trial, trial_source or accept");
     if (category == "convergence")
       fault.category = march::Category::convergence;
     else if (category == "value")
@@ -3850,7 +3854,8 @@ NB_MODULE(_core, module_) {
           "partials"_a.noconvert(), "rtol"_a, "atol"_a)
       .def("flow_guard_configure", &BoundState::flow_guard_configure, "row_sum"_a,
            "condition_bound"_a, "taylor_scale"_a, "max_squarings"_a, "allowance"_a)
-      // A test hook: fail the resident engine's `stage` after `after` more calls.
+      // A test hook: fail the resident engine's `stage` after `after` more calls;
+      // `trial_source` fails inside the trial's source evaluation.
       .def("adams_march_fault", &BoundState::march_fault, "stage"_a, "after"_a,
            "category"_a)
       .def(

native/src/engines.cpp

@@ -160,17 +160,26 @@ void ResidentEngine::inject(const march::Stage stage) {
   }
   const auto category = fault_->category;
   const int signal = fault_->signal;
+  const bool in_source = fault_->in_source;
   fault_.reset();
   if (signal != 0) {
     // Python's handler only notes it: it runs once the call returns to Python.
     static_cast<void>(std::raise(signal));
     return;
   }
-  if (category == march::Category::convergence)
-    throw EnvelopeConvergenceError("an injected native failure");
-  if (category == march::Category::value)
-    throw std::invalid_argument("an injected native failure");
-  throw std::runtime_error("an injected native failure");
+  const auto raise = [category] {
+    if (category == march::Category::convergence)
+      throw EnvelopeConvergenceError("an injected native failure");
+    if (category == march::Category::value)
+      throw std::invalid_argument("an injected native failure");
+    throw std::runtime_error("an injected native failure");
+  };
+  if (in_source) {
+    // Raised by the stage's own source evaluation, inside its swap.
+    state_.arm_source_fault(raise);
+    return;
+  }
+  raise();
 }
 
 ResidentRungeKutta::ResidentRungeKutta(GeneralState& state, Trial trial,

native/src/general/kernels.hpp

@@ -132,6 +132,25 @@ class FlowAside {
   std::unique_ptr<CoupledFlowAction> held_;
 };
 
+// Swaps two views for a scope and back however it is left: an Adams trial's
+// prediction stands in the state for its source evaluation alone, so that a failure
+// there leaves the accepted state in place (solprop issue #129).
+template <class View> class SwappedAside {
+public:
+  SwappedAside(View& first, View& second) : first_(first), second_(second) {
+    std::swap(first_, second_);
+  }
+  ~SwappedAside() { std::swap(first_, second_); }
+  SwappedAside(const SwappedAside&) = delete;
+  SwappedAside& operator=(const SwappedAside&) = delete;
+  SwappedAside(SwappedAside&&) = delete;
+  SwappedAside& operator=(SwappedAside&&) = delete;
+
+private:
+  View& first_;
+  View& second_;
+};
+
 // One frozen-residual weight row from the host into a View's row.
 template <class Row>
 void upload_weights(const ExecutionSpace& space, const Row& out,

native/src/general/state.cpp

@@ -464,6 +464,10 @@ void GeneralState::bind_host_source(std::shared_ptr<HostSource> source) {
 HostSource* GeneralState::host_source() const noexcept {
   return workspace_ ? workspace_->host_source() : nullptr;
 }
+void GeneralState::arm_source_fault(std::function<void()> fault) noexcept {
+  if (workspace_)
+    workspace_->arm_source_fault(std::move(fault));
+}
 void GeneralState::host_evaluation(const double z, const HostPurpose purpose) noexcept {
   if (workspace_)
     workspace_->host_evaluation(z, purpose);

native/src/general/trial.cpp

@@ -1685,6 +1685,8 @@ void GeneralWorkspace::adams_phase() {
 // The stage-zero source at an anchor-frame field held in current_, as adams_source
 // evaluates it, into stages_(0).
 void GeneralWorkspace::adams_evaluate_source(const TrialScalars& scalars) {
+  if (source_fault_)
+    std::exchange(source_fault_, {})();
   if (general_)
     general_->set_weights(scalars.transform_weights);
   if (problem_.raman.enabled)
@@ -1697,9 +1699,15 @@ void GeneralWorkspace::adams_evaluate_source(const TrialScalars& scalars) {
   has_recycled_ = false;
 }
 
-// The same, with current_ restored from adams_saved_ afterwards.
+// The same, with current_ restored from adams_saved_ afterwards, a failure's too: the
+// accepted state stands where the source failed (solprop issue #129).
 void GeneralWorkspace::adams_evaluate(const TrialScalars& scalars) {
-  adams_evaluate_source(scalars);
+  try {
+    adams_evaluate_source(scalars);
+  } catch (...) {
+    graph::deep_copy(execution_space_, current_, adams_saved_);
+    throw;
+  }
   graph::deep_copy(execution_space_, current_, adams_saved_);
 }
 
@@ -1891,9 +1899,10 @@ GeneralWorkspace::adams_fused_trial(const double dz, const AdamsWeights& weights
           total += weights.predictor[t] * store(weights.slot[t], column, index);
         predicted(column, index) = current(column, index) + dz * total;
       });
-  std::swap(current_, adams_saved_);
-  adams_evaluate_source(scalars);
-  std::swap(current_, adams_saved_);
+  {
+    const SwappedAside aside(current_, adams_saved_);
+    adams_evaluate_source(scalars);
+  }
   const auto stages = stages_;
   const auto source = adams_source_;
   const auto propagator = adams_propagator_;
@@ -3715,11 +3724,12 @@ std::vector<double> GeneralWorkspace::adams_elementwise_trial(
     launch::parallel_reduce("solprop_native.adams.exponential_predict", tasks, predict,
                             unused);
   }
-  std::swap(current_, adams_saved_);
-  use_uploaded_sample(sample);
-  // On a device the source's launches replay as one graph while they repeat.
-  adams_source_graph_.run(execution_space_, [&] { adams_evaluate_source(sample); });
-  std::swap(current_, adams_saved_);
+  {
+    const SwappedAside aside(current_, adams_saved_);
+    use_uploaded_sample(sample);
+    // On a device the source's launches replay as one graph while they repeat.
+    adams_source_graph_.run(execution_space_, [&] { adams_evaluate_source(sample); });
+  }
   const ExponentialCorrection reducer{.value_count = weights.n_rows,
                                       .store = store,
                                       .moments = moments,

native/src/general/workspace.hpp

@@ -190,6 +190,9 @@ class GeneralWorkspace {
     host_z_ = z;
     host_purpose_ = purpose;
   }
+  void arm_source_fault(std::function<void()> fault) noexcept {
+    source_fault_ = std::move(fault);
+  }
   /** The team a compact problem's FFTW loops take (`CpuTeamScope`): its pools'
       widest instance, which every threaded kernel and transform of it uses. Zero for
       any other problem: the runtime's team. */
@@ -750,6 +753,9 @@ class GeneralWorkspace {
   ComplexView2 host_input_, host_spectrum_;
   double host_z_ = 0.0;
   HostPurpose host_purpose_ = HostPurpose::startup;
+  // A test hook (solprop issue #129): run, once, as the next Adams source evaluation
+  // starts.
+  std::function<void()> source_fault_;
   ComplexView2 current_, advanced_, solution_, error_, residual_, recycled_, noise_;
   // solprop SEP 24 phase 3: the low-storage march's registers B and C (the state is
   // A) and its flows.

src/solprop/backends/host/adams.py

@@ -141,8 +141,8 @@ def __init__(
         # Whether the stages left a quiet collar out at the last attempt: the step
         # after it returns is held to the collar's stable step (`_collar_reentry`).
         self._collar_was_quiet = False
-        # A floor an accepted step's next proposal fell below, raised once the step is
-        # recorded (`_decide`, `_raise_floor`).
+        # A floor an attempt's next proposal fell below, raised once the attempt is
+        # recorded (`_propose`, `_reject`, `_raise_floor`).
         self._floor: ConvergenceError | None = None
 
     def reset(self) -> None:
@@ -619,6 +619,10 @@ def _decide(
         if not norm <= 1.0 and not (fixed and math.isfinite(norm)):
             self.ledger.rejected += 1
             self._reject(dz, norm, order)
+            if fixed:
+                # A replay records no rejected attempt (`replay_step`): its floor is
+                # raised at once.
+                self._raise_floor()
             return False, norm
         z = backend._z
         keep = self._history[-policy.order :] if policy.order else []
@@ -669,9 +673,9 @@ def _propose(self, dz: float, factor: float) -> None:
             self._floor = error
 
     def _raise_floor(self) -> None:
-        """Raise the floor an accepted step's proposal fell below, once it is recorded.
+        """Raise the floor an attempt's proposal fell below, once it is recorded.
 
-        Every caller that records an accepted step calls this straight after, and every
+        Every caller that records an attempt calls this straight after, and every
         attempt calls it before it starts (:meth:`_afford`), so a floor is never
         passed over.
         """
@@ -769,6 +773,11 @@ def _reject(self, dz: float, norm: float, order: int) -> None:
         :data:`~solprop.stepper.multistep.RESTART_AFTER` in a row the method restarts
         at order one from the accepted state, paying one startup call.
 
+        A proposal below the floor is left as it was, and the floor is raised once the
+        march has recorded the rejected attempt (:meth:`_raise_floor`), so that the
+        completed trial has its row (issue #651), as an accepted step's floor is
+        (:meth:`_propose`).
+
         - ``dz``: the rejected step, m.
         - ``norm``: its scaled defect, possibly not finite.
         """
@@ -782,9 +791,12 @@ def _reject(self, dz: float, norm: float, order: int) -> None:
         if outcome.restart:
             self._history = []
             self.ledger.failure_restarts += 1
-        self._backend._proposal = limit_step(
-            dz, outcome.factor, policy.dz_min, policy.dz_max
-        )
+        try:
+            self._backend._proposal = limit_step(
+                dz, outcome.factor, policy.dz_min, policy.dz_max
+            )
+        except ConvergenceError as error:
+            self._floor = error
 
 
 _ENGINE_MEMBERS = (