fix(flow): a save that was sent but not confirmed never resumes as a retry

packages/flow/docs/CAPABILITIES.md

@@ -118,7 +118,9 @@ at replaying one. These capabilities layer onto the same $0, model-free runtime.
   (`openadapt_flow/runtime/durable/`) so a halt becomes a durable pause the
   operator can approve and resume from the last verified state (`resume` /
   `approve`), not a restart, and explicitly not "hand the rest to a free-form
-  agent."
+  agent." A save that was sent but not confirmed pauses as delivery-uncertain,
+  so an ordinary resume can't send it again; the operator reconciles it or
+  explicitly approves one retry.
 - **After a stop: change ledger, write-ordering lint, declared undo.** A stop
   isn't a rollback, so every run report lists, in order, each step that could
   have changed the app and whether each declared effect was confirmed, refuted

packages/flow/docs/EFFECT_KIT.md

@@ -1359,7 +1359,11 @@ one-way contract hash (never the resolved values), the verdict evidence
 (observed/expected counts and values, matched records), and a
 `suggested_action` for the operator. The pattern is **halt + evidence**: the
 run stops, the task tells a human exactly what could not be certified, the
-human repairs the system of record, re-verifies, and resumes. There is
+human repairs the system of record, re-verifies, and resumes. The save was
+already sent, so its pause is delivery-uncertain and an ordinary resume can't
+send it again. The person reconciles it (a read-only check of the record as it
+is now) or, after confirming the record is missing, approves one retry with
+`--authorize-uncertain-retry`. There is
 intentionally **no compensation engine** beyond the single proven safe undo
 (duplicate-record deletion via a configured `Compensator`) — automatic repair
 of missing/partial/collateral state would be another wrong write.

packages/flow/openadapt_flow/runtime/durable/controller.py

@@ -199,6 +199,83 @@ def verified_checkpoint(
     )
 
 
+_RECONCILE_OPTIONS = (
+    "Reconcile the live application and independently verify whether "
+    "the effect landed; the attended Continue path can checkpoint the "
+    "verified outcome without repeating the action",
+    "If reconciliation proves a retry is necessary, record a fresh "
+    "approval explicitly authorizing one uncertain-delivery retry",
+)
+
+
+def _without_plain_resume(options: list[str]) -> list[str]:
+    """Swap a plain "approve and resume" option for the reconcile options.
+
+    The step may already have written its record, so an ordinary resume is
+    refused; offering it would point the operator at a dead end.
+    """
+
+    kept = [option for option in options if not option.startswith("Approve and RESUME")]
+    at = len(kept) - 1 if kept and kept[-1].startswith("Abort") else len(kept)
+    return kept[:at] + list(_RECONCILE_OPTIONS) + kept[at:]
+
+
+def post_delivery_uncertainty(
+    step: Optional[Step], result: StepResult
+) -> Optional[ActionDeliveryUncertainty]:
+    """Uncertainty for a consequential step that halted after its delivery.
+
+    ``delivery_uncertainty`` is otherwise set only when the delivery itself
+    raised. A save that was sent cleanly and then failed its record check
+    (timed out, refuted, unreadable) can still have landed, so retrying it on
+    an ordinary resume can write the record twice. A step counts as
+    consequential when it carries a record check or is marked irreversible.
+    A step that positively never sent its action (``delivery_attempted`` is
+    False) or that ended verified is not affected. An unknown step (``None``)
+    counts as consequential.
+    """
+
+    if result.ok or result.skipped or result.delivery_attempted is False:
+        return None
+    if step is not None and not step.effects and step.risk != "irreversible":
+        return None
+    from openadapt_flow.transaction import _reached_delivery
+
+    if result.delivery_attempted is not True and not _reached_delivery(result):
+        return None
+    return ActionDeliveryUncertainty(
+        operation=step.action.value if step is not None else "unknown",
+        native=False,
+        observed_at=_utc_now(),
+        cause_type="post_delivery_unverified",
+        verification_attempted=result.effect_verified is not None
+        or result.postconditions_ok is not None,
+        postconditions_confirmed=result.postconditions_ok,
+        effects_confirmed=result.effect_verified,
+    )
+
+
+def _program_pause_step(
+    workflow: Optional[Workflow],
+    state_id: str,
+    frames: Optional[list[GraphFrame]],
+) -> Optional[Step]:
+    """The action step of the program state that paused, if it can be found."""
+
+    if workflow is None or workflow.program is None or not frames:
+        return None
+    leaf = frames[-1]
+    graph = (
+        workflow.program
+        if leaf.graph_id == "__program__"
+        else workflow.subflows.get(leaf.graph_id)
+    )
+    if graph is None or leaf.state_id != state_id:
+        return None
+    state = graph.states.get(state_id)
+    return state.step if state is not None else None
+
+
 def _utc_now() -> str:
     from datetime import datetime, timezone
 
@@ -259,14 +336,7 @@ def classify_halt(step: Optional[Step], result: StepResult) -> tuple[str, list[s
         ]
 
     if result.delivery_uncertainty is not None:
-        return "delivery_uncertain", [
-            "Reconcile the live application and independently verify whether "
-            "the effect landed; the attended Continue path can checkpoint the "
-            "verified outcome without repeating the action",
-            "If reconciliation proves a retry is necessary, record a fresh "
-            "approval explicitly authorizing one uncertain-delivery retry",
-            abort,
-        ]
+        return "delivery_uncertain", [*_RECONCILE_OPTIONS, abort]
 
     # CAPTCHA, MFA, and re-authentication are human-presence requirements.
     # OpenAdapt only halts and leaves the live application to the operator.
@@ -619,11 +689,20 @@ def record(
             # pause so the winning Reject request can make it terminal.
             return
 
+        # A consequential step that was delivered but did not end verified
+        # may have written its record (the read-back timed out, or refuted a
+        # write that landed late). Mark the pause delivery-uncertain so an
+        # ordinary resume reconciles instead of sending the save again.
+        category, options = classify_halt(step, result)
+        if result.delivery_uncertainty is None:
+            result.delivery_uncertainty = post_delivery_uncertainty(step, result)
+            if result.delivery_uncertainty is not None:
+                options = _without_plain_resume(options)
+
         # HALT: durably pause instead of just dying. Resume from the last
         # verified checkpoint (0 when nothing verified yet).
         last = self.store.last_checkpoint()
         resume_from = last.next_step_index if last is not None else 0
-        category, options = classify_halt(step, result)
         pending = PendingEscalation(
             run_id=self.run_id,
             workflow_name=self.workflow_name,
@@ -846,6 +925,14 @@ def record_program_halt(
             return
         last = self.store.last_program_checkpoint()
         category, options = classify_halt(None, result)
+        if result.delivery_uncertainty is None:
+            # As for a linear step: a sent save that didn't end verified may
+            # have landed, so the pause reconciles instead of repeating it.
+            result.delivery_uncertainty = post_delivery_uncertainty(
+                _program_pause_step(workflow, state_id, program_frames), result
+            )
+            if result.delivery_uncertainty is not None:
+                options = _without_plain_resume(options)
         pending = PendingEscalation(
             run_id=self.run_id,
             workflow_name=self.workflow_name,

packages/flow/tests/test_api_step_durable.py

@@ -275,8 +275,20 @@ def test_a_save_the_record_check_refutes_is_not_redone_by_resume(fx):
     assert report.success is False and fx.records() == []
     assert fx.store().read_pending().resume_from_index == 1
 
+    # The save was sent, so its pause is delivery-uncertain: an ordinary
+    # resume is refused before anything runs.
+    assert fx.store().read_pending().delivery_uncertainty is not None
     second = ScriptedActuator(fx.url, ActuationStatus.ACTUATED)
-    resumed = resume(fx.run_dir, fx.replayer(second), approval=fx.approval())
+    with pytest.raises(ApprovalRequired, match="may already have actuated"):
+        resume(fx.run_dir, fx.replayer(second), approval=fx.approval())
+    assert second.calls == 0 and fx.backend.actions == []
+
+    # Even a resume that authorizes one uncertain retry never resends it.
+    resumed = resume(
+        fx.run_dir,
+        fx.replayer(second),
+        approval=fx.approval(authorize_uncertain_retry=True),
+    )
     assert second.calls == 0 and fx.backend.actions == []
     assert resumed.success is False
     assert "never sends it again" in resumed.results[-1].error

packages/flow/tests/test_durable_program.py

@@ -67,7 +67,12 @@
 from openadapt_flow.runtime.replayer import Replayer, _ProgramHalt
 
 # Reuse the scripted fakes + the scripted system-of-record verifier.
-from tests.test_durable_runtime import FakeSoRVerifier, _approval, _vision_ok
+from tests.test_durable_runtime import (
+    FakeSoRVerifier,
+    _approval,
+    _retry_approval,
+    _vision_ok,
+)
 from tests.test_replayer import FakeBackend, FakeVision, make_png
 
 # -- builders ----------------------------------------------------------------
@@ -836,7 +841,7 @@ def test_program_resume_restores_interpreter_and_completes(tmp_path):
         effect_verifier=verifier,
         poll_interval_s=0.01,
     )
-    resumed = resume(run_dir, resume_replayer, approval=_approval(bundle))
+    resumed = resume(run_dir, resume_replayer, approval=_retry_approval(bundle))
 
     assert resumed.success is True
     assert resumed.terminal_outcome == "success"
@@ -872,7 +877,7 @@ def test_program_resume_report_lists_every_loop_row_binding(tmp_path):
         effect_verifier=verifier,
         poll_interval_s=0.01,
     )
-    resumed = resume(run_dir, resume_replayer, approval=_approval(bundle))
+    resumed = resume(run_dir, resume_replayer, approval=_retry_approval(bundle))
     assert resumed.success is True
     assert [
         (b.loop_state_id, b.row_index, b.params["patient"])
@@ -895,7 +900,7 @@ def test_program_resume_does_not_reperform_confirmed_write(tmp_path):
         effect_verifier=verifier,
         poll_interval_s=0.01,
     )
-    resumed = resume(run_dir, resume_replayer, approval=_approval(bundle))
+    resumed = resume(run_dir, resume_replayer, approval=_retry_approval(bundle))
 
     assert resumed.success is True
     # The confirmed row's consequential write is NEVER re-performed: Alice's TYPE
@@ -1083,7 +1088,7 @@ def test_program_resume_refused_when_app_state_diverged(tmp_path):
         poll_interval_s=0.01,
     )
     with pytest.raises(StateDiverged):
-        resume(run_dir, resume_replayer, approval=_approval(bundle))
+        resume(run_dir, resume_replayer, approval=_retry_approval(bundle))
 
 
 def test_program_resume_refused_when_confirmed_effect_no_longer_holds(tmp_path):
@@ -1101,7 +1106,7 @@ def test_program_resume_refused_when_confirmed_effect_no_longer_holds(tmp_path):
         poll_interval_s=0.01,
     )
     with pytest.raises(StateDiverged):
-        resume(run_dir, resume_replayer, approval=_approval(bundle))
+        resume(run_dir, resume_replayer, approval=_retry_approval(bundle))
 
 
 def test_program_resume_revalidates_every_retained_effect_before_input(tmp_path):
@@ -1144,7 +1149,7 @@ def test_program_resume_revalidates_every_retained_effect_before_input(tmp_path)
                 effect_verifier=verifier,
                 poll_interval_s=0.01,
             ),
-            approval=_approval(bundle),
+            approval=_retry_approval(bundle),
         )
 
     assert backend.actions == []

packages/flow/tests/test_durable_program_history_v13.py

@@ -25,7 +25,11 @@
     _request,
     _run_attended_program_to_pause,
 )
-from tests.test_durable_program import _approval, _run_branch_loop_to_pause
+from tests.test_durable_program import (
+    _approval,
+    _retry_approval,
+    _run_branch_loop_to_pause,
+)
 from tests.test_replayer import FakeBackend, FakeVision, Match
 
 
@@ -99,7 +103,7 @@ def test_direct_resume_extends_the_pause_history_without_repeating_it(tmp_path):
             effect_verifier=verifier,
             poll_interval_s=0.0,
         ),
-        approval=_approval(bundle),
+        approval=_retry_approval(bundle),
     )
 
     assert resumed.success is True
@@ -131,7 +135,7 @@ def test_restart_after_post_pause_checkpoint_uses_latest_history_boundary(
 ):
     _report, run_dir, bundle, verifier = _run_branch_loop_to_pause(tmp_path)
     verifier.refute.clear()
-    approval = _approval(bundle)
+    approval = _retry_approval(bundle)
 
     class CrashBeforeSecondResumedAction(Replayer):
         resumed_actions = 0

packages/flow/tests/test_durable_resume_state_binding.py

@@ -26,7 +26,7 @@
 )
 from openadapt_flow.runtime.effects import Effect, EffectKind
 from openadapt_flow.runtime.replayer import Replayer
-from tests.test_durable_runtime import FakeSoRVerifier, _approval
+from tests.test_durable_runtime import FakeSoRVerifier, _approval, _retry_approval
 from tests.test_replayer import FakeBackend, FakeVision, Match
 
 
@@ -135,7 +135,7 @@ def test_program_resume_revalidates_retained_text_absent_before_input(tmp_path):
                 effect_verifier=verifier,
                 poll_interval_s=0.0,
             ),
-            approval=_approval(bundle),
+            approval=_retry_approval(bundle),
         )
 
     assert backend.actions == []
@@ -157,7 +157,7 @@ def test_linear_resume_revalidates_retained_text_absent_before_input(tmp_path):
                 effect_verifier=verifier,
                 poll_interval_s=0.0,
             ),
-            approval=_approval(bundle),
+            approval=_retry_approval(bundle),
         )
 
     assert backend.actions == []
@@ -175,7 +175,7 @@ def test_resume_rechecks_bundle_inside_admission_before_input(tmp_path):
         effect_verifier=verifier,
         poll_interval_s=0.0,
     )
-    approved = _approval(bundle)
+    approved = _retry_approval(bundle)
     admit = replayer._admit_durable_resume
 
     def mutate_bundle_then_admit(*args, **kwargs):