verify: keep owed rechecks in the ledger; tell spent, lost and miscalled rechecks apart

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. 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.
 

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:

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)