From 2318b89b6f62cc43df794bc30c9ebfd95d8a672b Mon Sep 17 00:00:00 2001 From: abrichr Date: Wed, 4 Nov 2026 06:00:59 -0500 Subject: [PATCH 1/3] verify: keep owed rechecks in the ledger and tell spent, lost and miscalled rechecks apart A provisional CONFIRMED stays listed as recheck-owed by openadapt verify ledger list until a recheck settles it, so a restart no longer loses it. A confirming recheck records a rechecked event; a retry is then handed provisional: false. verify_recheck says whether a recheck was already run, is owed but not held by this server, or never existed. A recheck called with other params or another actor says the call was wrong. A recheck that can't run keeps the recheck owed. A later UNCERTAIN no longer replaces a REFUTED the ledger holds. Closes #1670 --- openadapt/cli.py | 17 +++- packages/flow/docs/VERIFY_WRITES.md | 8 +- .../flow/openadapt_flow/agent/verify_tools.py | 88 +++++++++++++++---- packages/flow/openadapt_flow/verifier.py | 88 ++++++++++++++++++- packages/flow/openadapt_flow/write_ledger.py | 65 ++++++++++++++ packages/flow/tests/test_verifier_recheck.py | 83 +++++++++++++++++ packages/flow/tests/test_verify_tools.py | 75 +++++++++++++++- tests/test_verify_command.py | 27 ++++++ 8 files changed, 427 insertions(+), 24 deletions(-) diff --git a/openadapt/cli.py b/openadapt/cli.py index cb1306fc9e..c1ed2775e2 100644 --- a/openadapt/cli.py +++ b/openadapt/cli.py @@ -1350,12 +1350,23 @@ def verify_ledger() -> None: @verify_ledger.command("list") @_LEDGER_OPTION def verify_ledger_list(ledger_dir: Optional[str]) -> None: - """Show held and open writes: key, state, when it began, outcome.""" + """Show held and open writes: key, state, when it began, outcome. + + Also each provisional CONFIRMED whose recheck is still owed, with its due + time, so a recheck isn't lost when the server that holds it restarts. + """ ledger = _write_ledger(ledger_dir) rows = ledger.unresolved() - if not rows: - click.echo("No held or open writes.") + owed = ledger.rechecks_owed() + if not rows and not owed: + click.echo("No held or open writes. No rechecks owed.") return + for key, attempt in owed: + click.echo( + f"{key} recheck-owed began {attempt.begun.get('at')} " + f"due {attempt.recheck_owed} by {attempt.begun.get('actor_kind')}:" + f"{attempt.begun.get('actor_id')}" + ) for key, attempt in rows: outcome = (attempt.resolved or {}).get("outcome", "-") state = "review" if attempt.review else ledger.state(attempt) diff --git a/packages/flow/docs/VERIFY_WRITES.md b/packages/flow/docs/VERIFY_WRITES.md index 2888b1e5f7..7ec0877cfe 100644 --- a/packages/flow/docs/VERIFY_WRITES.md +++ b/packages/flow/docs/VERIFY_WRITES.md @@ -82,15 +82,17 @@ Every result also says what it establishes (`scope`): the read path (database, A A check sees what is in the store when it reads. A write that lands after that read is not part of the result: a retry that fires two minutes later, a queued job, a trigger. To catch those, the contract can ask for a recheck. Set `frame.database.recheck_after_s` (a contract that reads the whole database), and every `CONFIRMED` result says `provisional: true` and when the recheck is due (`recheck_due_at`). The recheck reads the whole database again and judges every change since the first check: - In Python: `recheck(contract, result, params=values, actor=AGENT)` at or after `recheck_due_at`, or with `wait=True` to sleep until then. -- Over MCP: call `verify_recheck` with the `write_id`. Before the due time it returns `due: false` and how long to wait. Your agent's later writes under the same contract, once checked, account for their own rows. A write under another contract can't: an agent that creates a contact and then an opportunity inside the window gets the first write `REFUTED`. +- Over MCP: call `verify_recheck` with the `write_id`. Before the due time it returns `due: false` and how long to wait. Your agent's later writes under the same contract and `agent_id`, once checked, account for their own rows. A write under another contract or another `agent_id` can't: an agent that creates a contact and then an opportunity inside the window gets the first write `REFUTED`. | Recheck outcome | What it means | What to do | | --- | --- | --- | | `CONFIRMED` | Nothing moved since the first check that the contract doesn't explain. | The write is final as of the recheck. | | `REFUTED` | Something changed after the first check that nothing explains: the earlier `CONFIRMED` no longer stands. | Don't retry. A person checks the record. | -| `UNCERTAIN` | The database couldn't be read again, or was still changing. | Don't retry. A person checks the record. | +| `UNCERTAIN` | The database couldn't be read again or was still changing; the recheck state wasn't issued under this state directory or doesn't match its receipt; the call gave other params, another actor or another contract than the write (the reason then says the call was wrong, not the write); or a change may belong to a later write that was itself refuted. | Don't retry. A person checks the record. | -A recheck that doesn't confirm also marks the write in the ledger, so a retry of it is held for a person (`openadapt verify ledger list`) instead of being handed the earlier `CONFIRMED`. That holds for a recheck run from a saved `recheck.json` too. If the recheck can't find the write in the ledger, its reasons say so: hold the write by hand. +A recheck that doesn't confirm also marks the write in the ledger, so a retry of it is held for a person (`openadapt verify ledger list`) instead of being handed the earlier `CONFIRMED`. That holds for a recheck run from a saved `recheck.json` too. If the recheck can't find the write in the ledger, its reasons say so: hold the write by hand. A later `UNCERTAIN` never replaces a `REFUTED` the ledger already holds. + +An owed recheck is in the ledger too. Until a recheck settles it, `openadapt verify ledger list` shows the write as `recheck-owed` with its due time, also after the server restarts; a retry of the write is handed its result with `provisional: true` and a `next` that says how to recheck it. The MCP server can run only the rechecks it holds itself: after a restart, `verify_recheck` says the recheck is owed but not held here, and a person rechecks it in Python from the saved `recheck.json` (`RecheckState.load`). A recheck already run says so, with its outcome. Once a recheck confirms the write, a retry is handed `provisional: false`. There is no default window: pick one longer than the slowest write that can follow a save in your system. A write after the recheck's read is not seen by anything. See [VERIFY_RUN.md](VERIFY_RUN.md#writes-that-land-later) for how the recheck judges changes and what was measured on real apps. diff --git a/packages/flow/openadapt_flow/agent/verify_tools.py b/packages/flow/openadapt_flow/agent/verify_tools.py index 1ed0587d8f..1c4106b4d8 100644 --- a/packages/flow/openadapt_flow/agent/verify_tools.py +++ b/packages/flow/openadapt_flow/agent/verify_tools.py @@ -27,6 +27,8 @@ again and returns ``CONFIRMED`` (the earlier verdict stands), ``REFUTED`` (something moved since that nothing explains: the earlier ``CONFIRMED`` no longer stands) or ``UNCERTAIN``. A write after this read is not seen. + An owed recheck is also in the uncertain-write ledger, so ``openadapt + verify ledger list`` shows it after a restart, until a recheck settles it. ``verify_begin`` consults the uncertain-write ledger, so a write whose outcome is unknown is refused, also after a restart or in another server. @@ -55,6 +57,7 @@ recheck, verify_write, ) +from openadapt_flow.write_ledger import default_ledger __all__ = ["VERIFY_TOOLS", "VerifyBridge", "contract_params"] @@ -113,9 +116,13 @@ def __init__(self, contracts_dir: Path, state_dir: Path) -> None: raise ValueError(f"no write contracts (*.yaml) in {self.contracts_dir}") self._pending: dict[str, PendingWrite] = {} #: Provisional results whose recheck is still to run, by write_id. - #: Held in this server's memory only: after a restart, recheck from - #: Python with the saved ``recheck.json`` (``RecheckState.load``). + #: This server can run them only while it holds them; the ledger + #: records each one owed (``openadapt verify ledger list``), and + #: after a restart it is rechecked from Python with the saved + #: ``recheck.json`` (``RecheckState.load``). self._rechecks: dict[str, tuple[PendingWrite, WriteResult]] = {} + #: Rechecks this server already ran, by write_id: their outcome. + self._spent: dict[str, str] = {} #: Every checked write with a recheck state, of any outcome. self._checked: list[tuple[PendingWrite, WriteResult]] = [] self._lock = threading.Lock() @@ -204,10 +211,11 @@ def list_tool_specs(self) -> list[ToolSpec]: "Reads the whole database again: CONFIRMED (the verdict " "stands), REFUTED (something moved after it that nothing " "explains: the write is no longer verified) or UNCERTAIN. " - "Your later writes under the same contract, checked with " - "verify_check, account for their own rows; a write under " - "another contract does not, and refutes it. Before the due " - "time it returns due: false and the time to call again." + "Your later writes under the same contract and agent_id, " + "checked with verify_check, account for their own rows; a " + "write under another contract or another agent_id does " + "not, and refutes it. Before the due time it returns due: " + "false and the time to call again." ), input_schema=_obj( {"write_id": {"type": "string", "maxLength": 64}}, ["write_id"] @@ -261,11 +269,14 @@ def _begin(self, args: Mapping[str, Any]) -> dict[str, Any]: write_key=None if write_key is None else str(write_key), ) except AlreadyWritten as exc: - return { + out: dict[str, Any] = { "ok_to_write": False, "reason": f"{exc}; don't make the write", "earlier_result": exc.result, } + if exc.result.get("provisional"): + out["next"] = self._owed_next(str(exc.result.get("write_id") or "")) + return out except WriteBlocked as exc: return {"ok_to_write": False, "reason": f"{exc}; don't make the write"} with self._lock: @@ -303,6 +314,53 @@ def _check(self, args: Mapping[str, Any]) -> dict[str, Any]: ) return out + def _owed_next(self, write_id: str) -> str: + """What a retry handed a provisional result should do next.""" + + with self._lock: + held = write_id in self._rechecks + if held: + return ( + "The earlier result is provisional: call verify_recheck with " + f"write_id {write_id} at or after recheck_due_at." + ) + return ( + "The earlier result is provisional and its recheck is owed, but " + "this server doesn't hold it (it restarted, or another server " + "checked the write): a person rechecks it from the saved " + "recheck.json. openadapt verify ledger list shows it until then." + ) + + def _unheld(self, write_id: str) -> str: + """Why this server has no recheck to run for ``write_id``.""" + + with self._lock: + spent = self._spent.get(write_id) + if spent is not None: + return ( + f"that write was already rechecked ({spent}): each write is " + "rechecked once" + ) + owed = None + try: + found = default_ledger().find_attempt(write_id) + if found is not None: + owed = found[1].recheck_owed + except Exception: # noqa: BLE001 - only the wording depends on it + owed = None + if owed: + return ( + "no provisional result for that write_id in this server, but " + f"the ledger shows its recheck owed (due {owed}): the server " + "restarted, or another server checked the write. A person " + "rechecks it from the saved recheck.json; openadapt verify " + "ledger list shows it until then" + ) + return ( + "no provisional result for that write_id: only a CONFIRMED write " + "whose contract sets frame.database.recheck_after_s has a recheck" + ) + def _later_writes( self, entry: PendingWrite, result: WriteResult ) -> list[WriteResult]: @@ -341,12 +399,7 @@ def _recheck(self, args: Mapping[str, Any]) -> dict[str, Any]: with self._lock: held = self._rechecks.get(write_id) if held is None: - raise BridgeError( - "no provisional result for that write_id: only a CONFIRMED write " - "whose contract sets frame.database.recheck_after_s has a " - "recheck, each is rechecked once, and this server holds them in " - "memory only" - ) + raise BridgeError(self._unheld(write_id)) entry, result = held with self._lock: unchecked = [ @@ -379,11 +432,14 @@ def _recheck(self, args: Mapping[str, Any]) -> dict[str, Any]: "recheck_due_at.", } except VerifierError as exc: - with self._lock: - self._rechecks.pop(write_id, None) - raise BridgeError(f"the write can't be rechecked: {exc}") from exc + # Nothing was judged: the recheck is still owed and held here. + raise BridgeError( + f"the recheck couldn't run: {exc}; it is still owed, so call " + "verify_recheck again once that is fixed" + ) from exc with self._lock: self._rechecks.pop(write_id, None) + self._spent[write_id] = late.outcome if late.verified: nxt = "The verdict stands. A write after read_at is not seen." elif late.late: diff --git a/packages/flow/openadapt_flow/verifier.py b/packages/flow/openadapt_flow/verifier.py index 51e4672c3b..cdef7a0c34 100644 --- a/packages/flow/openadapt_flow/verifier.py +++ b/packages/flow/openadapt_flow/verifier.py @@ -1053,6 +1053,12 @@ def _settle( assert attempt is not None and attempt.resolved is not None result = dict(attempt.resolved.get("result") or {}) result["not_sent_again"] = True + if attempt.rechecked is not None: + # A recheck confirmed it since: no longer provisional, and no + # recheck is due. + result["provisional"] = False + result["recheck_due_at"] = None + result["rechecked_at"] = attempt.rechecked.get("at") if not attempt.delivered: led.delivered(key, attempt.attempt_id) raise AlreadyWritten( @@ -2234,9 +2240,53 @@ def recheck( ) if late.outcome != "CONFIRMED": late = _hold_after_recheck(ledger, write, late, trusted=trusted) + else: + _mark_rechecked(ledger, write, late) return late +def _mark_rechecked( + ledger: Any, write: WriteResult | RecheckState, late: RecheckResult +) -> None: + """Record on the write's ledger attempt that a recheck confirmed it, so + ``openadapt verify ledger list`` no longer shows a recheck owed and a + retry is handed a result that is no longer provisional. + + Best effort: the verdict stands either way. If nothing is recorded, the + list keeps showing the recheck as owed, which is the safe side. + """ + + try: + if isinstance(write, WriteResult): + write_id, key = write.write_id, write.write_key + if ledger is _NO_LEDGER and write.ledger is not None: + ledger = write.ledger + else: + receipt = write.payload.get("receipt") or {} + write_id, key = str(receipt.get("correlation_id") or ""), "" + led = _resolve_ledger(ledger) + if led is None or not write_id: + return + if key: + attempt = led.latest(key) + else: + found = led.find_attempt(write_id) + key, attempt = found if found is not None else ("", None) + if attempt is None or attempt.attempt_id != write_id: + return + if (attempt.resolved or {}).get("outcome") != "CONFIRMED": + return + led.rechecked( + key, + write_id, + outcome=late.outcome, + read_at=late.read_at, + receipt_sha256=late.receipt_sha256 or None, + ) + except Exception: # noqa: BLE001 - the list keeps showing it owed + return + + def _hold_after_recheck( ledger: Any, write: WriteResult | RecheckState, @@ -2304,6 +2354,16 @@ def unheld(why: str) -> RecheckResult: if attempt.attempt_id != write_id: return unheld("a newer attempt of this write is in the ledger") prior = attempt.resolved or {} + if prior.get("outcome") == "REFUTED" and late.outcome != "REFUTED": + # An earlier recheck refuted it: a later UNCERTAIN doesn't soften + # that. The attempt stays held as it is. + return _replace( + late, + reasons=[ + *late.reasons, + "the ledger already holds this write as REFUTED; left as is", + ], + ) reason = "; ".join(late.reasons) or "the recheck did not confirm the write" led.resolved( key, @@ -2337,6 +2397,32 @@ def unheld(why: str) -> RecheckResult: return late +#: ``recheck_problem`` sentences that fault the recheck call (the params or +#: actor it was given), not the write. +_CALL_PROBLEMS = ( + "the params don't bind the contract", + "the params are not the ones the write was verified with", + "the recheck state's write was made by a different actor", + "the contract can't be bound for the recheck", + "the recheck state was taken under a different contract", +) + + +def _recheck_refusal(problem: str) -> str: + """The UNCERTAIN reason for a recheck that couldn't run. When the call + was given other params, another actor or another contract than the + write was made with, it says the call was wrong: the write itself may + be fine, and stays unrechecked until a call with its own does.""" + + if problem.startswith(_CALL_PROBLEMS): + return ( + f"{problem}: the recheck call doesn't match the write (its params, " + "actor or contract), so nothing was rechecked; this says nothing " + "about the write itself" + ) + return f"{problem}, so the write can't be rechecked" + + def _judge_recheck( contract: Contract, write: WriteResult | RecheckState, @@ -2381,7 +2467,7 @@ def _judge_recheck( ) or recheck_problem(cfg, state.payload, params=params, run_record=run_record) if problem is not None: return ( - _replace(base, reasons=[f"{problem}, so the write can't be rechecked"]), + _replace(base, reasons=[_recheck_refusal(problem)]), False, ) wait_s = seconds_until(state.due_at) diff --git a/packages/flow/openadapt_flow/write_ledger.py b/packages/flow/openadapt_flow/write_ledger.py index 4b313a158a..c1fa59e10f 100644 --- a/packages/flow/openadapt_flow/write_ledger.py +++ b/packages/flow/openadapt_flow/write_ledger.py @@ -290,6 +290,8 @@ class Attempt: resolved: Optional[dict[str, Any]] = None delivered: bool = False released: bool = False + #: The ``rechecked`` event: a recheck confirmed a provisional result. + rechecked: Optional[dict[str, Any]] = None events: list[dict[str, Any]] = field(default_factory=list) @property @@ -333,6 +335,23 @@ def review(self) -> bool: return bool(self.resolved and self.resolved.get("review")) and not self.released + @property + def recheck_owed(self) -> Optional[str]: + """The ``recheck_due_at`` of a provisional ``CONFIRMED`` that no + recheck has settled yet, else None. + + A recheck that confirms the write records ``rechecked``; one that + doesn't resolves the attempt again (no longer provisional). A + person's release settles it too. + """ + + if self.released or self.rechecked is not None or self.resolved is None: + return None + result = self.resolved.get("result") + if not isinstance(result, Mapping) or not result.get("provisional"): + return None + return str(result.get("recheck_due_at") or "") or None + @property def folder(self) -> Path: return Path(str(self.begun.get("folder", ""))) @@ -374,10 +393,13 @@ def fold(events: Sequence[Mapping[str, Any]]) -> Optional[Attempt]: if kind == "resolved": latest.resolved = dict(event) latest.delivered = bool(event.get("delivered")) + latest.rechecked = None # a new outcome; any recheck was of the old elif kind == "delivered": latest.delivered = True elif kind == "released": latest.released = True + elif kind == "rechecked": + latest.rechecked = dict(event) if latest is not None: latest.events = [ dict(e) for e in events if e.get("attempt") == latest.attempt_id @@ -512,6 +534,21 @@ def resolved(self, key: str, attempt_id: str, **fields: Any) -> None: }, ) + def rechecked(self, key: str, attempt_id: str, **fields: Any) -> None: + """A recheck confirmed the attempt's provisional result: no + recheck is owed any more (:attr:`Attempt.recheck_owed`).""" + + self.backend.append( + key, + { + "event": "rechecked", + "attempt": attempt_id, + "at": _now(), + "t": time.time(), + **fields, + }, + ) + def delivered(self, key: str, attempt_id: str) -> None: self.backend.append( key, {"event": "delivered", "attempt": attempt_id, "at": _now()} @@ -598,6 +635,34 @@ def unresolved(self) -> list[tuple[str, Attempt]]: out.append((key, attempt)) return out + def rechecks_owed(self) -> list[tuple[str, Attempt]]: + """Every key whose latest attempt is a provisional ``CONFIRMED`` + still owed its recheck. Kept here, not in a server's memory, so a + restart doesn't lose it.""" + + out = [] + for key in self.backend.keys(): + attempt = self.latest(key) + if attempt is not None and attempt.recheck_owed: + out.append((key, attempt)) + return out + + def find_attempt(self, attempt_id: str) -> Optional[tuple[str, Attempt]]: + """The key whose latest attempt has this id (a write id), or None + when none does or more than one key claims it.""" + + found = [ + key + for key, begun, _ in self.attempts() + if str(begun.get("attempt")) == attempt_id + ] + if len(found) != 1: + return None + attempt = self.latest(found[0]) + if attempt is None or attempt.attempt_id != attempt_id: + return None + return found[0], attempt + def _now() -> str: return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) diff --git a/packages/flow/tests/test_verifier_recheck.py b/packages/flow/tests/test_verifier_recheck.py index 7f93b40b15..fc8a1f803c 100644 --- a/packages/flow/tests/test_verifier_recheck.py +++ b/packages/flow/tests/test_verifier_recheck.py @@ -1004,3 +1004,86 @@ def test_a_write_result_with_wrong_params_is_still_held_in_the_ledger( with pytest.raises(verifier.WriteBlocked) as raised: begin_write(contract, _params(2), actor=BOT, write_key="k-own") assert not isinstance(raised.value, verifier.AlreadyWritten) + + +# -- review #1670: owed rechecks in the ledger, call faults, a held REFUTED ---- + + +def test_a_recheck_owed_is_in_the_ledger_until_a_recheck_confirms_it( + tmp_path: Path, due_now: None +) -> None: + from openadapt_flow.write_ledger import default_ledger + + contract, _, result = _keyed(tmp_path, "k-owed") + owed = dict(default_ledger().rechecks_owed()) + assert owed[result.write_key].recheck_owed == result.recheck_due_at + with pytest.raises(verifier.AlreadyWritten) as raised: + begin_write(contract, _params(2), actor=BOT, write_key="k-owed") + assert raised.value.result["provisional"] is True + + assert _recheck(contract, result).outcome == "CONFIRMED" + assert default_ledger().rechecks_owed() == [] + with pytest.raises(verifier.AlreadyWritten) as raised: + begin_write(contract, _params(2), actor=BOT, write_key="k-owed") + earlier = raised.value.result + assert earlier["outcome"] == "CONFIRMED" + assert earlier["provisional"] is False and earlier["recheck_due_at"] is None + assert earlier["rechecked_at"] + + +def test_a_confirming_recheck_from_a_saved_state_settles_the_owed_recheck( + tmp_path: Path, due_now: None +) -> None: + from openadapt_flow.write_ledger import default_ledger + + contract, _, result = _keyed(tmp_path, "k-owed-saved") + assert result.recheck is not None and result.recheck.path is not None + saved = RecheckState.load(result.recheck.path, params=_params(2)) + assert _recheck(contract, saved).outcome == "CONFIRMED" + assert default_ledger().rechecks_owed() == [] + + +def test_a_refuting_recheck_settles_the_owed_recheck_and_holds_the_write( + tmp_path: Path, due_now: None +) -> None: + from openadapt_flow.write_ledger import default_ledger + + contract, db, result = _keyed(tmp_path, "k-owed-ref") + _sql(db, "INSERT INTO calendar (pid, title) VALUES (2, 'New Patient')") + assert _recheck(contract, result).outcome == "REFUTED" + led = default_ledger() + assert led.rechecks_owed() == [] + assert result.write_key in dict(led.unresolved()) + + +def test_a_recheck_call_with_other_params_or_actor_says_the_call_was_wrong( + tmp_path: Path, due_now: None +) -> None: + contract, _, first = _first(tmp_path) + for late in ( + _recheck(contract, first, n=7), + recheck( + contract, first, params=_params(2), actor=Actor("agent", "someone-else") + ), + ): + assert late.outcome == "UNCERTAIN" + assert "the recheck call doesn't match the write" in late.reasons[0] + assert "says nothing about the write itself" in late.reasons[0] + + +def test_a_later_uncertain_recheck_does_not_replace_a_held_refuted( + tmp_path: Path, due_now: None +) -> None: + from openadapt_flow.write_ledger import default_ledger + + contract, db, result = _keyed(tmp_path, "k-keep") + _sql(db, "INSERT INTO calendar (pid, title) VALUES (2, 'New Patient')") + assert _recheck(contract, result).outcome == "REFUTED" + db.unlink() # a second recheck can't read the database: UNCERTAIN + late = _recheck(contract, result) + assert late.outcome == "UNCERTAIN" + assert "already holds this write as REFUTED" in late.reasons[-1] + attempt = default_ledger().latest(result.write_key) + assert attempt is not None and attempt.resolved["outcome"] == "REFUTED" + with pytest.raises(verifier.WriteHeld): + begin_write(contract, _params(2), actor=BOT, write_key="k-keep") diff --git a/packages/flow/tests/test_verify_tools.py b/packages/flow/tests/test_verify_tools.py index 5d81e31009..ec81508184 100644 --- a/packages/flow/tests/test_verify_tools.py +++ b/packages/flow/tests/test_verify_tools.py @@ -241,7 +241,7 @@ def test_verify_recheck_confirms_then_is_spent(sqlite_bridge, monkeypatch) -> No assert late["due"] is True and late["outcome"] == "CONFIRMED" assert late["verified"] is True and late["late"] is False assert late["receipt_sha256"] == checked["receipt_sha256"] - with pytest.raises(BridgeError, match="no provisional result"): + with pytest.raises(BridgeError, match=r"already rechecked \(CONFIRMED\)"): bridge.dispatch("verify_recheck", {"write_id": checked["write_id"]}) @@ -338,3 +338,76 @@ def test_a_write_without_a_recheck_has_none_to_run(sqlite_bridge) -> None: bridge, _ = sqlite_bridge with pytest.raises(BridgeError, match="no provisional result"): bridge.dispatch("verify_recheck", {"write_id": "nope"}) + + +# -- review #1670: owed rechecks survive a restart; a fault doesn't spend one -- + + +def test_an_owed_recheck_is_named_after_a_restart(sqlite_bridge, tmp_path) -> None: + from openadapt_flow.write_ledger import default_ledger + + bridge, db = sqlite_bridge + checked = _provisional(bridge, db) + assert len(default_ledger().rechecks_owed()) == 1 + # A new server on the same folders: the recheck isn't held here, and the + # error says it is owed, not that there never was one. + restarted = VerifyBridge(tmp_path / "contracts", tmp_path / "state") + with pytest.raises(BridgeError, match="ledger shows its recheck owed"): + restarted.dispatch("verify_recheck", {"write_id": checked["write_id"]}) + retry = restarted.dispatch( + "verify_begin", {"contract": "patient", "values": VALUES} + ) + assert retry["ok_to_write"] is False + assert retry["earlier_result"]["provisional"] is True + assert "doesn't hold it" in retry["next"] and "ledger list" in retry["next"] + + +def test_a_retry_before_the_recheck_is_told_to_recheck(sqlite_bridge) -> None: + bridge, db = sqlite_bridge + checked = _provisional(bridge, db) + retry = bridge.dispatch("verify_begin", {"contract": "patient", "values": VALUES}) + assert retry["earlier_result"]["provisional"] is True + assert checked["write_id"] in retry["next"] + assert "verify_recheck" in retry["next"] + + +def test_a_retry_after_a_confirming_recheck_is_not_provisional( + sqlite_bridge, monkeypatch +) -> None: + from openadapt_flow.write_ledger import default_ledger + + bridge, db = sqlite_bridge + checked = _provisional(bridge, db) + monkeypatch.setattr(verifier, "seconds_until", lambda due: 0.0) + late = bridge.dispatch("verify_recheck", {"write_id": checked["write_id"]}) + assert late["outcome"] == "CONFIRMED" + assert default_ledger().rechecks_owed() == [] + retry = bridge.dispatch("verify_begin", {"contract": "patient", "values": VALUES}) + earlier = retry["earlier_result"] + assert earlier["provisional"] is False and earlier["recheck_due_at"] is None + assert "next" not in retry + + +def test_a_recheck_that_could_not_run_is_still_owed(sqlite_bridge, monkeypatch) -> None: + from openadapt_flow.agent import verify_tools + + bridge, db = sqlite_bridge + checked = _provisional(bridge, db) + monkeypatch.setattr(verifier, "seconds_until", lambda due: 0.0) + + def broken(*args, **kwargs): + raise verifier.VerifierError("excused_by[0] has no params") + + real = verify_tools.recheck + monkeypatch.setattr(verify_tools, "recheck", broken) + with pytest.raises(BridgeError, match="still owed"): + bridge.dispatch("verify_recheck", {"write_id": checked["write_id"]}) + monkeypatch.setattr(verify_tools, "recheck", real) + late = bridge.dispatch("verify_recheck", {"write_id": checked["write_id"]}) + assert late["outcome"] == "CONFIRMED", late["reasons"] + + +def test_the_recheck_tool_says_another_agent_id_is_not_excused(sqlite_bridge) -> None: + bridge, _ = sqlite_bridge + spec = {t.name: t for t in bridge.list_tool_specs()}["verify_recheck"] + assert "another agent_id" in spec.description diff --git a/tests/test_verify_command.py b/tests/test_verify_command.py index 088b6e00c1..9ce361a041 100644 --- a/tests/test_verify_command.py +++ b/tests/test_verify_command.py @@ -103,3 +103,30 @@ def test_verify_ledger_lists_and_releases_a_held_write(tmp_path) -> None: assert released.exit_code == 0, released.output listed = runner.invoke(main, ["verify", "ledger", "list", "--ledger", str(folder)]) assert "No held or open writes." in listed.output + + +def test_verify_ledger_lists_a_recheck_owed_until_it_runs(tmp_path) -> None: + ledger_mod = pytest.importorskip("openadapt_flow.write_ledger") + folder = tmp_path / "ledger" + ledger = ledger_mod.Ledger(ledger_mod.FileLedger(folder)) + key = "d" * 64 + due = "2026-01-01T00:02:00Z" + ledger.begun(key, "write-2", contract="c" * 64, actor_kind="agent", actor_id="bot") + ledger.resolved( + key, + "write-2", + outcome="CONFIRMED", + retry_safe=False, + delivered=True, + result={"outcome": "CONFIRMED", "provisional": True, "recheck_due_at": due}, + ) + runner = CliRunner() + listed = runner.invoke(main, ["verify", "ledger", "list", "--ledger", str(folder)]) + assert listed.exit_code == 0, listed.output + assert key in listed.output and "recheck-owed" in listed.output + assert f"due {due}" in listed.output + + ledger.rechecked(key, "write-2", outcome="CONFIRMED") + listed = runner.invoke(main, ["verify", "ledger", "list", "--ledger", str(folder)]) + assert "recheck-owed" not in listed.output + assert "No rechecks owed." in listed.output From 8559a6b240aa4ec475e9ff4b1e96c640146117ca Mon Sep 17 00:00:00 2001 From: abrichr Date: Wed, 4 Nov 2026 07:29:32 -0500 Subject: [PATCH 2/3] ledger: hand back a write while its recheck is owed, so a new attempt never hides it --- packages/flow/docs/VERIFY_WRITES.md | 2 +- packages/flow/openadapt_flow/write_ledger.py | 12 ++++++-- packages/flow/tests/test_verifier_recheck.py | 31 ++++++++++++++++++++ 3 files changed, 41 insertions(+), 4 deletions(-) diff --git a/packages/flow/docs/VERIFY_WRITES.md b/packages/flow/docs/VERIFY_WRITES.md index 7ec0877cfe..96d6c82a7a 100644 --- a/packages/flow/docs/VERIFY_WRITES.md +++ b/packages/flow/docs/VERIFY_WRITES.md @@ -92,7 +92,7 @@ A check sees what is in the store when it reads. A write that lands after that r A recheck that doesn't confirm also marks the write in the ledger, so a retry of it is held for a person (`openadapt verify ledger list`) instead of being handed the earlier `CONFIRMED`. That holds for a recheck run from a saved `recheck.json` too. If the recheck can't find the write in the ledger, its reasons say so: hold the write by hand. A later `UNCERTAIN` never replaces a `REFUTED` the ledger already holds. -An owed recheck is in the ledger too. Until a recheck settles it, `openadapt verify ledger list` shows the write as `recheck-owed` with its due time, also after the server restarts; a retry of the write is handed its result with `provisional: true` and a `next` that says how to recheck it. The MCP server can run only the rechecks it holds itself: after a restart, `verify_recheck` says the recheck is owed but not held here, and a person rechecks it in Python from the saved `recheck.json` (`RecheckState.load`). A recheck already run says so, with its outcome. Once a recheck confirms the write, a retry is handed `provisional: false`. +An owed recheck is in the ledger too. Until a recheck settles it, `openadapt verify ledger list` shows the write as `recheck-owed` with its due time, also after the server restarts; a retry of the write is handed its result with `provisional: true` and a `next` that says how to recheck it. This holds past the repeat window too: while a recheck is owed, a retry of the same values never starts a new attempt that would hide it. The MCP server can run only the rechecks it holds itself: after a restart, `verify_recheck` says the recheck is owed but not held here, and a person rechecks it in Python from the saved `recheck.json` (`RecheckState.load`). A recheck already run says so, with its outcome. Once a recheck confirms the write, a retry is handed `provisional: false`. There is no default window: pick one longer than the slowest write that can follow a save in your system. A write after the recheck's read is not seen by anything. See [VERIFY_RUN.md](VERIFY_RUN.md#writes-that-land-later) for how the recheck judges changes and what was measured on real apps. diff --git a/packages/flow/openadapt_flow/write_ledger.py b/packages/flow/openadapt_flow/write_ledger.py index c1fa59e10f..ae36427c89 100644 --- a/packages/flow/openadapt_flow/write_ledger.py +++ b/packages/flow/openadapt_flow/write_ledger.py @@ -320,9 +320,15 @@ def state_with(self, repeat_s: float) -> str: if r.get("retry_safe"): return "free" if r.get("outcome") == "CONFIRMED" or r.get("landed"): - if r.get("write_key_bound") or r.get("review") or not self.delivered: - # Under review: handed back until a person releases it, so - # no later attempt hides it from the list. + if ( + r.get("write_key_bound") + or r.get("review") + or not self.delivered + or self.recheck_owed + ): + # Under review, or a recheck still owed: handed back until a + # person releases it (or the recheck settles it), so no later + # attempt hides it from the list. return "undelivered" return "free" if _age(r.get("at")) >= repeat_s else "undelivered" if r.get("outcome") == "REFUTED": diff --git a/packages/flow/tests/test_verifier_recheck.py b/packages/flow/tests/test_verifier_recheck.py index fc8a1f803c..21de74371c 100644 --- a/packages/flow/tests/test_verifier_recheck.py +++ b/packages/flow/tests/test_verifier_recheck.py @@ -1031,6 +1031,37 @@ def test_a_recheck_owed_is_in_the_ledger_until_a_recheck_confirms_it( assert earlier["rechecked_at"] +def test_an_owed_recheck_outlives_the_repeat_window_and_a_restart( + tmp_path: Path, due_now: None +) -> None: + # No write key: past the repeat window a retry of the same values would + # start a new attempt and hide the owed recheck. It is handed the + # provisional result instead until the recheck settles it. + from openadapt_flow.write_ledger import FileLedger, Ledger + + def ledger() -> Ledger: # a fresh one each time, as after a restart + return Ledger(FileLedger(tmp_path / "ledger"), repeat_s=0) + + db = _db(tmp_path) + contract = Contract.from_dict(_doc(db, recheck_after_s=3600)) + before = begin_write(contract, _params(2), actor=BOT, ledger=ledger()) + _sql(db, *_save_sql(2)) + result = verify_write(before) + assert result.verified and result.provisional, result.reasons + for _ in range(2): + with pytest.raises(verifier.AlreadyWritten) as raised: + begin_write(contract, _params(2), actor=BOT, ledger=ledger()) + assert raised.value.result["provisional"] is True + [(key, owed)] = ledger().rechecks_owed() + assert owed.recheck_owed == result.recheck_due_at + assert ledger().find_attempt(owed.attempt_id) == (key, owed) + + assert _recheck(contract, result).outcome == "CONFIRMED" + assert ledger().rechecks_owed() == [] + # Settled: the same values may be written again. + begin_write(contract, _params(2), actor=BOT, ledger=ledger()) + + def test_a_confirming_recheck_from_a_saved_state_settles_the_owed_recheck( tmp_path: Path, due_now: None ) -> None: From a4163b5289f897dd31a8dc1a9885abe460e88d2c Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Wed, 4 Nov 2026 08:06:45 -0500 Subject: [PATCH 3/3] launcher: move the openadapt/cli.py recheck change out of this PR The launcher installs the released openadapt-flow from the lock, which has no Ledger.rechecks_owed yet. The launcher change and its tests move to a separate draft PR that waits for the next flow release. Flow changes are unchanged. --- openadapt/cli.py | 17 +++-------------- tests/test_verify_command.py | 27 --------------------------- 2 files changed, 3 insertions(+), 41 deletions(-) diff --git a/openadapt/cli.py b/openadapt/cli.py index c1ed2775e2..cb1306fc9e 100644 --- a/openadapt/cli.py +++ b/openadapt/cli.py @@ -1350,23 +1350,12 @@ def verify_ledger() -> None: @verify_ledger.command("list") @_LEDGER_OPTION def verify_ledger_list(ledger_dir: Optional[str]) -> None: - """Show held and open writes: key, state, when it began, outcome. - - Also each provisional CONFIRMED whose recheck is still owed, with its due - time, so a recheck isn't lost when the server that holds it restarts. - """ + """Show held and open writes: key, state, when it began, outcome.""" ledger = _write_ledger(ledger_dir) rows = ledger.unresolved() - owed = ledger.rechecks_owed() - if not rows and not owed: - click.echo("No held or open writes. No rechecks owed.") + if not rows: + click.echo("No held or open writes.") return - for key, attempt in owed: - click.echo( - f"{key} recheck-owed began {attempt.begun.get('at')} " - f"due {attempt.recheck_owed} by {attempt.begun.get('actor_kind')}:" - f"{attempt.begun.get('actor_id')}" - ) for key, attempt in rows: outcome = (attempt.resolved or {}).get("outcome", "-") state = "review" if attempt.review else ledger.state(attempt) diff --git a/tests/test_verify_command.py b/tests/test_verify_command.py index 9ce361a041..088b6e00c1 100644 --- a/tests/test_verify_command.py +++ b/tests/test_verify_command.py @@ -103,30 +103,3 @@ def test_verify_ledger_lists_and_releases_a_held_write(tmp_path) -> None: assert released.exit_code == 0, released.output listed = runner.invoke(main, ["verify", "ledger", "list", "--ledger", str(folder)]) assert "No held or open writes." in listed.output - - -def test_verify_ledger_lists_a_recheck_owed_until_it_runs(tmp_path) -> None: - ledger_mod = pytest.importorskip("openadapt_flow.write_ledger") - folder = tmp_path / "ledger" - ledger = ledger_mod.Ledger(ledger_mod.FileLedger(folder)) - key = "d" * 64 - due = "2026-01-01T00:02:00Z" - ledger.begun(key, "write-2", contract="c" * 64, actor_kind="agent", actor_id="bot") - ledger.resolved( - key, - "write-2", - outcome="CONFIRMED", - retry_safe=False, - delivered=True, - result={"outcome": "CONFIRMED", "provisional": True, "recheck_due_at": due}, - ) - runner = CliRunner() - listed = runner.invoke(main, ["verify", "ledger", "list", "--ledger", str(folder)]) - assert listed.exit_code == 0, listed.output - assert key in listed.output and "recheck-owed" in listed.output - assert f"due {due}" in listed.output - - ledger.rechecked(key, "write-2", outcome="CONFIRMED") - listed = runner.invoke(main, ["verify", "ledger", "list", "--ledger", str(folder)]) - assert "recheck-owed" not in listed.output - assert "No rechecks owed." in listed.output