From eb53ecbc531c2c4f7a94e0d20bd7ddec0e9edf4f Mon Sep 17 00:00:00 2001 From: abrichr Date: Tue, 20 Oct 2026 23:45:59 -0400 Subject: [PATCH 1/4] fix(flow): a save that was sent but not confirmed never resumes as a retry A durable pause carried delivery_uncertainty only when the delivery itself raised. A save sent cleanly and then not confirmed by its record check (the read-back timed out, refuted a write that landed late, or couldn't read the record) paused with the ordinary "Approve and RESUME" option, and an approved resume pressed Save again: a duplicate record. DurableRun.record now stamps delivery_uncertainty (cause post_delivery_unverified) on a consequential step (one with a record check, or marked irreversible) that halted after its delivery. The existing gate then applies: an ordinary resume is refused, and the pause offers reconciliation (a read-only check of the record as it is now) or an approval that explicitly authorizes one retry. The pause keeps its own category; only the plain resume option is swapped for the reconcile options. A step that never sent its action resumes as before. Tests: - test_realism_durable_resume.py: a backend that writes a row per Save and a read-back that times out once. Resume refuses and the row stays single; reconcile reads the record back and finishes without saving again. - test_durable_runtime.py: test_resume_continues_from_last_checkpoint asserted the old re-press. A new test asserts the default approval never presses Save; resume tests that retry a refuted save now use an approval that authorizes one uncertain retry. Docs: CAPABILITIES.md and EFFECT_KIT.md say a sent but unconfirmed save pauses as delivery-uncertain and is reconciled or explicitly retried. Also covers program pauses (record_program_pause), which had the same gap. Co-Authored-By: Claude Opus 5.5 --- packages/flow/docs/CAPABILITIES.md | 4 +- packages/flow/docs/EFFECT_KIT.md | 6 +- .../runtime/durable/controller.py | 105 ++++++++++- packages/flow/tests/test_api_step_durable.py | 14 +- .../test_durable_resume_state_binding.py | 8 +- packages/flow/tests/test_durable_runtime.py | 64 +++++-- .../flow/tests/test_realism_durable_resume.py | 177 ++++++++++++++++++ 7 files changed, 351 insertions(+), 27 deletions(-) create mode 100644 packages/flow/tests/test_realism_durable_resume.py diff --git a/packages/flow/docs/CAPABILITIES.md b/packages/flow/docs/CAPABILITIES.md index a9bd591f5..da61f5615 100644 --- a/packages/flow/docs/CAPABILITIES.md +++ b/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 diff --git a/packages/flow/docs/EFFECT_KIT.md b/packages/flow/docs/EFFECT_KIT.md index 53dc3ee83..393d94754 100644 --- a/packages/flow/docs/EFFECT_KIT.md +++ b/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. diff --git a/packages/flow/openadapt_flow/runtime/durable/controller.py b/packages/flow/openadapt_flow/runtime/durable/controller.py index 5379c6ac8..c760326f1 100644 --- a/packages/flow/openadapt_flow/runtime/durable/controller.py +++ b/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, diff --git a/packages/flow/tests/test_api_step_durable.py b/packages/flow/tests/test_api_step_durable.py index 322c1c090..7e7d9ebf2 100644 --- a/packages/flow/tests/test_api_step_durable.py +++ b/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 diff --git a/packages/flow/tests/test_durable_resume_state_binding.py b/packages/flow/tests/test_durable_resume_state_binding.py index aa90a2e93..a212f2cc9 100644 --- a/packages/flow/tests/test_durable_resume_state_binding.py +++ b/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): diff --git a/packages/flow/tests/test_durable_runtime.py b/packages/flow/tests/test_durable_runtime.py index abb00ffbc..b12d84e56 100644 --- a/packages/flow/tests/test_durable_runtime.py +++ b/packages/flow/tests/test_durable_runtime.py @@ -174,7 +174,9 @@ def _dirs(tmp_path): return bundle, tmp_path / "run" -def _approval(bundle, run_dir=None) -> ApprovalRecord: +def _approval( + bundle, run_dir=None, *, authorize_uncertain_retry: bool = False +) -> ApprovalRecord: """An authenticated approval for the given bundle (P0-5): resume now REQUIRES one, so every legitimate resume in these tests carries an operator identity, a chosen resolution, and the bundle version it was granted against.""" @@ -191,9 +193,20 @@ def _approval(bundle, run_dir=None) -> ApprovalRecord: run_id=manifest.run_id, workflow_name=manifest.workflow_name, run_dir=run_dir, + authorize_uncertain_retry=authorize_uncertain_retry, ) +def _retry_approval(bundle, run_dir=None) -> ApprovalRecord: + """An approval that explicitly authorizes one uncertain-delivery retry. + + A halted save may already have landed, so an ordinary approval cannot + repeat it. These tests model an operator who checked the system of record, + found the record missing, and chose to retry. + """ + return _approval(bundle, run_dir, authorize_uncertain_retry=True) + + def _three_step_workflow(*, with_effects: bool) -> Workflow: steps = [] for step_id, key in (("s0", "A"), ("s1", "B"), ("s2", "C")): @@ -290,7 +303,12 @@ def test_halt_midway_writes_pending_and_checkpoints_prior(tmp_path): assert "refuted" in pending.reason.lower() # It proposes operator options and points at the last verified checkpoint. assert pending.proposed_options # non-empty - assert any("RESUME" in opt for opt in pending.proposed_options) + # The save was delivered and its record check refuted it, so it may still + # have landed: an ordinary resume is not offered and cannot repeat it. + assert pending.delivery_uncertainty is not None + assert pending.delivery_uncertainty.cause_type == "post_delivery_unverified" + assert not any("RESUME" in opt for opt in pending.proposed_options) + assert any("Reconcile" in opt for opt in pending.proposed_options) assert pending.resume_from_index == 2 # continue after s1 (index 1) assert pending.resume_from_step_id == "s1" # The run's parameter bindings are captured for an identical re-bind. @@ -302,11 +320,35 @@ def test_halt_midway_writes_pending_and_checkpoints_prior(tmp_path): # -- resume: continue from the last checkpoint, not from step 0 ------------- +def test_resume_never_repeats_a_delivered_save_without_authorization(tmp_path): + """The default approval never presses a save that may already have landed.""" + + report, run_dir, bundle, _orig_backend, verifier = _run_to_halt(tmp_path) + assert report.success is False + verifier.refute.clear() + resume_backend = FakeBackend() + + with pytest.raises(ApprovalRequired, match="may already have actuated"): + resume( + run_dir, + Replayer( + resume_backend, + vision=_vision_ok(), + effect_verifier=verifier, + poll_interval_s=0.01, + ), + approval=_approval(bundle), + ) + assert ("press", "C") not in resume_backend.actions + assert CheckpointStore(run_dir).read_pending() is not None + + def test_resume_continues_from_last_checkpoint(tmp_path): report, run_dir, bundle, _orig_backend, verifier = _run_to_halt(tmp_path) assert report.success is False - # The operator resolves the cause (the record now exists): s2 will confirm. + # The operator checked the system of record, found s2's record missing, + # and authorized one retry: s2 will now confirm. verifier.refute.clear() # A FRESH backend for the resumed leg (a new run against the live system); # its recorded actions reveal exactly which steps re-executed. @@ -318,7 +360,7 @@ def test_resume_continues_from_last_checkpoint(tmp_path): 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 already-verified steps were NOT re-executed on the resumed backend... @@ -378,7 +420,7 @@ def test_resume_egress_posture_stays_sticky_across_a_second_resume(tmp_path): effect_verifier=verifier, poll_interval_s=0.01, ), - approval=_approval(bundle), + approval=_retry_approval(bundle), ) assert first_resume.success is False assert first_resume.screenshots_may_leave_box is True @@ -395,7 +437,7 @@ def test_resume_egress_posture_stays_sticky_across_a_second_resume(tmp_path): effect_verifier=verifier, poll_interval_s=0.01, ), - approval=_approval(bundle), + approval=_retry_approval(bundle), ) assert second_resume.success is True assert second_resume.screenshots_may_leave_box is True @@ -435,7 +477,7 @@ def _run_step(self, *args, **kwargs): effect_verifier=verifier, poll_interval_s=0.01, ), - approval=_approval(bundle), + approval=_retry_approval(bundle), ) retained = store.read_manifest() @@ -487,7 +529,7 @@ def test_durable_resume_projection_failure_keeps_fail_closed_terminal_evidence( idempotency_ledger=ledger, poll_interval_s=0.01, ), - approval=_approval(bundle), + approval=_retry_approval(bundle), ) # The resumed action is delivered once. The failed projection does not @@ -558,7 +600,7 @@ def test_durable_pause_defers_ledger_outcome_until_resume_verifies(tmp_path): idempotency_ledger=ledger, poll_interval_s=0.01, ), - approval=_approval(bundle), + approval=_retry_approval(bundle), ) assert resumed.success is True # This fixture uses the Demo profile, so its correct terminal taxonomy is @@ -924,7 +966,7 @@ def test_resume_revalidates_but_does_not_repeat_confirmed_steps(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)) manifest = CheckpointStore(run_dir).read_manifest() assert manifest is not None @@ -982,7 +1024,7 @@ def test_halt_on_first_step_resumes_from_zero(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 ("press", "A") in resume_backend.actions assert [c.step_index for c in store.checkpoints()] == [0, 1, 2] diff --git a/packages/flow/tests/test_realism_durable_resume.py b/packages/flow/tests/test_realism_durable_resume.py new file mode 100644 index 000000000..6d5357b05 --- /dev/null +++ b/packages/flow/tests/test_realism_durable_resume.py @@ -0,0 +1,177 @@ +"""Resume after a save that landed but could not be confirmed. + +The fakes in ``test_durable_runtime`` refute a step whose write never +happened. A real system of record also fails the other way: the click lands, +the row is written, and the read-back times out (a slow EMR, a dropped +connection). The pause then looks ordinary, the operator approves the default +"resume" option, and resume presses the save again. + +Here the backend writes a row on every press of the save key, and the record +check times out once. An ordinary resume must refuse, and reconciling reads +the record back and finishes without pressing the save again. +See docs/audits/REALISM_AUDIT.md. +""" + +from __future__ import annotations + +import pytest + +from openadapt_flow.runtime.durable import ApprovalRequired, CheckpointStore, resume +from openadapt_flow.runtime.durable.attended import ( + AttendedActionRequest, + AttendedActionStore, + BoundAttendedExecutor, + execute_attended_action, +) +from openadapt_flow.runtime.effects import Effect, EffectState, EffectVerdict, Verdict +from openadapt_flow.runtime.replayer import Replayer +from tests.test_durable_runtime import ( + _approval, + _dirs, + _three_step_workflow, + _vision_ok, +) +from tests.test_replayer import FakeBackend + +_STEP_FOR_KEY = {"A": "s0", "B": "s1", "C": "s2"} + + +class _Store: + """A system of record: one row per delivered press.""" + + def __init__(self) -> None: + self.rows: list[str] = [] + + def records(self) -> list[dict]: + return [{"step": _STEP_FOR_KEY.get(key, key)} for key in self.rows] + + +class _WritingBackend(FakeBackend): + def __init__(self, store: _Store) -> None: + super().__init__() + self.store = store + + def key(self, key, *args, **kwargs): # type: ignore[override] + self.store.rows.append(str(key)) + return super().key(key, *args, **kwargs) + + def press(self, key, *args, **kwargs): # type: ignore[override] + self.store.rows.append(str(key)) + return super().press(key, *args, **kwargs) + + +class _SlowReadback: + """Reads the store, but the first read-back of s2 times out.""" + + substrate = "fake" + + def __init__(self, store: _Store) -> None: + self.store = store + self.timeouts = {"s2": 1} + + def capture_pre_state(self, context=None) -> EffectState: + return EffectState( + substrate=self.substrate, reachable=True, records=self.store.records() + ) + + def verify(self, expected: Effect, before: EffectState, context=None): + step = dict(expected.match).get("step") + if self.timeouts.get(step, 0) > 0: + self.timeouts[step] -= 1 + return EffectVerdict( + verdict=Verdict.INDETERMINATE, + kind=expected.kind, + substrate=self.substrate, + reason="the system of record did not answer before the timeout", + unavailable=True, + ) + return EffectVerdict( + verdict=Verdict.CONFIRMED, + kind=expected.kind, + substrate=self.substrate, + reason="the record is present", + ) + + +def _halt_after_a_landed_save(tmp_path, *, absolute_check: bool = False): + store = _Store() + workflow = _three_step_workflow(with_effects=True) + if absolute_check: + # A check that reads the record as it is now, with no before-snapshot, + # is the kind reconciliation can prove after the fact. + workflow = workflow.model_copy( + update={ + "steps": [ + step.model_copy( + update={ + "effects": [ + e.model_copy(update={"forbid_collateral_loss": False}) + for e in step.effects + ] + } + ) + for step in workflow.steps + ] + } + ) + bundle, run_dir = _dirs(tmp_path) + workflow.save(bundle) + verifier = _SlowReadback(store) + first = Replayer( + _WritingBackend(store), + vision=_vision_ok(), + effect_verifier=verifier, + durable=True, + poll_interval_s=0.01, + ).run(workflow, params={"who": "alice"}, bundle_dir=bundle, run_dir=run_dir) + assert first.success is False + assert store.rows.count("C") == 1 # the save landed before the timeout + return store, verifier, bundle, run_dir + + +def _replayer(store: _Store, verifier: _SlowReadback) -> Replayer: + return Replayer( + _WritingBackend(store), + vision=_vision_ok(), + effect_verifier=verifier, + poll_interval_s=0.01, + ) + + +def test_resume_never_repeats_a_save_whose_readback_timed_out(tmp_path) -> None: + store, verifier, bundle, run_dir = _halt_after_a_landed_save(tmp_path) + pending = CheckpointStore(run_dir).read_pending() + assert pending is not None and pending.delivery_uncertainty is not None + assert pending.delivery_uncertainty.cause_type == "post_delivery_unverified" + assert not any("RESUME" in option for option in pending.proposed_options) + + with pytest.raises(ApprovalRequired, match="may already have actuated"): + resume(run_dir, _replayer(store, verifier), approval=_approval(bundle)) + assert store.rows.count("C") == 1, store.rows + + +def test_reconcile_finds_the_saved_record_and_finishes_without_saving_again( + tmp_path, +) -> None: + store, verifier, _bundle, run_dir = _halt_after_a_landed_save( + tmp_path, absolute_check=True + ) + capability = AttendedActionStore(run_dir).read() + assert capability is not None and "reconcile" in capability.allowed_actions + + decision = execute_attended_action( + run_dir, + AttendedActionRequest( + capability_digest=capability.digest, + idempotency_key="reconcile-test-0001", + action="reconcile", + disposition="reconciliation_requested", + ), + operator="staff", + decided_by="human", + executor=BoundAttendedExecutor(lambda _manifest: _replayer(store, verifier)), + ) + + assert decision.status == "completed", decision.message + assert decision.report_success is True + assert store.rows.count("C") == 1, store.rows From 50e9a3627482a96c99d2509431ab36a77a701b82 Mon Sep 17 00:00:00 2001 From: abrichr Date: Wed, 21 Oct 2026 00:01:50 -0400 Subject: [PATCH 2/4] test: program resume tests retry a refuted save with explicit authority Co-Authored-By: Claude Opus 5.5 --- packages/flow/tests/test_durable_program.py | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) diff --git a/packages/flow/tests/test_durable_program.py b/packages/flow/tests/test_durable_program.py index 078d1eaae..3cde04f03 100644 --- a/packages/flow/tests/test_durable_program.py +++ b/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 == [] From 818f9232f0f389d265b0c274b32276fa499d6980 Mon Sep 17 00:00:00 2001 From: abrichr Date: Wed, 21 Oct 2026 00:03:50 -0400 Subject: [PATCH 3/4] test: program history resume tests use explicit retry authority Co-Authored-By: Claude Opus 5.5 --- .../flow/tests/test_durable_program_history_v13.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/packages/flow/tests/test_durable_program_history_v13.py b/packages/flow/tests/test_durable_program_history_v13.py index 3f8b1a2ba..e34b7ff93 100644 --- a/packages/flow/tests/test_durable_program_history_v13.py +++ b/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 From 23e57454877d3a1c7180cc7b3f0a232ccbc6f78c Mon Sep 17 00:00:00 2001 From: abrichr Date: Wed, 21 Oct 2026 00:41:00 -0400 Subject: [PATCH 4/4] test: resume tests that retry a refuted save authorize it explicitly Co-Authored-By: Claude Opus 5.5 --- packages/flow/tests/test_execution_profiles.py | 4 ++-- packages/flow/tests/test_multi_surface_identity.py | 11 +++++++---- packages/flow/tests/test_sealed_at_rest.py | 3 +++ 3 files changed, 12 insertions(+), 6 deletions(-) diff --git a/packages/flow/tests/test_execution_profiles.py b/packages/flow/tests/test_execution_profiles.py index 4c7f75840..ab0909b35 100644 --- a/packages/flow/tests/test_execution_profiles.py +++ b/packages/flow/tests/test_execution_profiles.py @@ -91,7 +91,7 @@ from openadapt_flow.runtime.replayer import Replayer as _RuntimeReplayer from openadapt_flow.verification import VerificationTier from openadapt_flow.vision.ocr import OcrLine -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, make_png _KEY = "profile-test-key" @@ -3076,7 +3076,7 @@ def test_standard_program_resume_preserves_exact_loop_contracts_without_reactuat effect_verifier=verifier, require_settled=True, ), - approval=_approval(bundle), + approval=_retry_approval(bundle), ) assert resumed.execution_outcome == ExecutionOutcome.VERIFIED.value diff --git a/packages/flow/tests/test_multi_surface_identity.py b/packages/flow/tests/test_multi_surface_identity.py index 30dbe2b04..1bc120030 100644 --- a/packages/flow/tests/test_multi_surface_identity.py +++ b/packages/flow/tests/test_multi_surface_identity.py @@ -33,7 +33,10 @@ ) from openadapt_flow.runtime.effects import Effect, EffectKind # noqa: E402 from openadapt_flow.runtime.replayer import Replayer # noqa: E402 -from tests.test_durable_runtime import FakeSoRVerifier, _approval # noqa: E402 +from tests.test_durable_runtime import ( # noqa: E402 + FakeSoRVerifier, + _retry_approval, +) pytestmark = pytest.mark.timeout(600) @@ -126,7 +129,7 @@ def test_resume_rebinds_surfaces_by_identity(bundle, app_url, browser, tmp_path) with pytest.raises(ApprovalRequired): resume(run_dir, fresh, approval=None) - resumed = resume(run_dir, fresh, approval=_approval(bundle_dir, run_dir)) + resumed = resume(run_dir, fresh, approval=_retry_approval(bundle_dir, run_dir)) assert resumed.success, _e2e._summary(resumed) assert resumed.model_calls == 0 by_id = {r.step_id: r for r in resumed.results} @@ -153,7 +156,7 @@ def test_ambiguous_rebind_on_resume_halts(bundle, app_url, browser, tmp_path): PlaywrightBackend(page), effect_verifier=verifier, poll_interval_s=0.05 ) with pytest.raises(StateDiverged, match="2 open browser pages pass"): - resume(run_dir, fresh, approval=_approval(bundle_dir, run_dir)) + resume(run_dir, fresh, approval=_retry_approval(bundle_dir, run_dir)) # Neither report tab, nor the main page, was driven. assert extra.locator("#ack").inner_text() == "" assert page.locator("#done").inner_text() == "" @@ -176,7 +179,7 @@ def test_page_routing_mismatch_halts(bundle, app_url, browser, tmp_path): PlaywrightBackend(page), effect_verifier=verifier, poll_interval_s=0.05 ) with pytest.raises(StateDiverged, match="no open browser page passes"): - resume(run_dir, fresh, approval=_approval(bundle_dir, run_dir)) + resume(run_dir, fresh, approval=_retry_approval(bundle_dir, run_dir)) assert page.locator("#done").inner_text() == "" finally: page.context.close() diff --git a/packages/flow/tests/test_sealed_at_rest.py b/packages/flow/tests/test_sealed_at_rest.py index 0f6c7e32b..7324a6956 100644 --- a/packages/flow/tests/test_sealed_at_rest.py +++ b/packages/flow/tests/test_sealed_at_rest.py @@ -719,6 +719,9 @@ def test_sealed_run_pauses_and_resumes_without_plaintext( run_id=manifest.run_id, workflow_name=manifest.workflow_name, run_dir=run_dir, + # The refuted save was sent; the operator found the record missing + # and authorizes one retry. + authorize_uncertain_retry=True, ) verifier.refute.clear() resumed_backend = FakeBackend()