v0.11 item 1: a reader that names a bad row and blinds nothing else

CHANGELOG.md

@@ -17,6 +17,37 @@ any change to one appears here.
   verify as intact after two SQL statements, because the head that would catch them is a row in the
   same database. Transcribed from a real store rather than argued.
 
+- **A reader that names a bad row and blinds nothing else** (`SPEC-v0.11.md` §5, rule 3). A single
+  malformed *value* of a declared key raised out of `Receipt.from_dict`, and because both stores
+  build every row before any caller sees one, that one `UPDATE` stopped `ctrlrun receipts`,
+  `receipts --verify-chain`, `ctrlrun inspect`, `ctrlrun stats` and the operator MCP server's
+  `receipts` and `stats` tools together. `inspect` on an action the tamper never touched is what
+  the blast radius really was: not "this receipt is unreadable" but "this store is unreadable".
+  `SPEC-v0.7.md` §12.5 recorded it and deferred it twice.
+
+  **One tampered row now costs one row.** `ctrlrun.receipt.UnreadableReceipt` is what a store hands
+  back for a row it cannot construct, carrying the row's `seq`, its `receipt_id` where that field
+  alone is readable, and the **type** of what refused it, never the message. `CHAIN_BREAKS` did not
+  change: §12.5 offered a new break name as one of two candidates and `SPEC-v0.11.md` §5.1 declines
+  it, because `content_altered` already names a document that cannot be canonicalized and a second
+  name for one fact would be two names for one break.
+
+### Changed
+
+- **`StateStore.receipts()` returns `tuple[Receipt | UnreadableReceipt, ...]`**, amending
+  `SPEC-v0.6.md` §9.2's frozen protocol. Before, it raised. Both backends change, and a second
+  backend could not implement §5 without it. A store with no bad row is unaffected: every row still
+  reads back as a `Receipt`.
+- **A receipt's position now comes from the `seq` column**, which is what `verify_chain`'s docstring
+  has claimed since v0.6 and what was not true as shipped. Both stores selected `json, hash` and
+  ordered by a column they never read, so every `Receipt.seq` came out of `document.get("seq")`, the
+  one field a tamperer controls. Rewriting one document's `seq` from 2 to 99 reported `missing 2`,
+  `content_altered 99`, `missing 100` and `link_broken 3`: four breaks at three positions, two of
+  them rows that do not exist. The same tamper now reports `content_altered` once, at 2.
+- `ctrlrun stats` reports `unreadable receipts` and the `ctrlrun.stats/v1` document carries
+  `unreadable_receipts`, **omitted entirely where there is none**, on `ledger_rows`' precedent. A
+  total that silently dropped a row nobody could read would be `SPEC-v0.4.md` §3.8's false green.
+
 ## [0.10.0] — Multi-agent
 
 One question: when one agent hands work to another, what does the second one hold?

src/ctrlrun/cli/main.py

@@ -40,6 +40,8 @@
     EventType,
     JSONLEventSink,
     Receipt,
+    UnreadableReceipt,
+    _readable,
     iso_timestamp,
     verify_chain,
 )
@@ -483,13 +485,29 @@ def receipts(
         # document defines is not an error here -- it matches nothing, which is the right answer
         # for a reader running against a store whose policy has since changed. A dangling
         # citation is a *load* error, in the place that can see the registry.
-        found = tuple(receipt for receipt in found if control_id in receipt.controls)
+        # An unreadable row is **kept**, whatever the filter says (SPEC-v0.11 §5.2). Its
+        # `controls` could not be read, so it cannot be shown not to cite this id, and dropping
+        # it would let one `UPDATE` hide a row from exactly the query an operator runs to find
+        # a control's evidence.
+        found = tuple(
+            receipt
+            for receipt in found
+            if isinstance(receipt, UnreadableReceipt) or control_id in receipt.controls
+        )
     if last is not None:
         found = found[-last:]
     if not found:
         click.echo("no receipts yet" if control_id is None else f"no receipts cite {control_id!r}")
         return
     for receipt in found:
+        if isinstance(receipt, UnreadableReceipt):
+            # SPEC-v0.11 §5.2: named in place, at its `seq`, and the rows around it still print.
+            click.echo(
+                json.dumps(receipt.to_dict(), ensure_ascii=False)
+                if as_json
+                else _unreadable_line(receipt)
+            )
+            continue
         click.echo(receipt.to_json() if as_json else _receipt_line(receipt))
 
 
@@ -674,7 +692,12 @@ def inspect(
         # today are two that disagree later (T193).
         document = inspection_for(store, action_id)
         events = tuple(event for event in store.events() if event.action_id == action_id)
-        receipt = next((found for found in store.receipts() if found.action_id == action_id), None)
+        # `_readable`: a row that cannot be read carries no `action_id`, so it can never be
+        # the receipt for *this* action. SPEC-v0.11 §2.3's sharp case is exactly this line:
+        # `inspect` on an action the tamper never touched used to raise here.
+        receipt = next(
+            (found for found in _readable(store.receipts()) if found.action_id == action_id), None
+        )
     except CTRLRunError as exc:
         raise _fail(exc) from exc
 
@@ -934,18 +957,24 @@ def stats(since: str | None, as_json: bool, store_url: str | None) -> None:
         # kernel's own refusal; the exit code an operator scripts against stays 2.
         raise click.UsageError(str(exc)) from exc
     try:
+        rows = _store(store_url).receipts()
+        # A refused row has no `finished_at` to compare and no result to count, so it cannot
+        # enter a total. It is reported separately below rather than dropped in silence: a
+        # count that quietly omitted it would be `SPEC-v0.4 §3.8`'s false green (§5.2).
         counted = [
             receipt
-            for receipt in _store(store_url).receipts()
+            for receipt in _readable(rows)
             if boundary is None or receipt.finished_at >= boundary
         ]
+        unreadable = tuple(row for row in rows if isinstance(row, UnreadableReceipt))
     except CTRLRunError as exc:
         raise _fail(exc) from exc
     document = stats_document(
         counted,
         mode=policy.mode,
         boundary=boundary,
         ledger_rows=ledger_rows(_store(store_url)),
+        unreadable=len(unreadable),
     )
     if as_json:
         click.echo(json.dumps(document, ensure_ascii=False, indent=2))
@@ -972,6 +1001,12 @@ def _stats_lines(document: Mapping[str, Any]) -> list[str]:
     if "ledger_rows" in document:
         # §7.3: growth is observable before it is a problem.
         lines.append(_stat("budget ledger rows", document["ledger_rows"]))
+    if "unreadable_receipts" in document:
+        # SPEC-v0.11 §5.2. Present only where there is one, so a clean store prints what it
+        # printed at 0.10.0 (§5.3). Without this line `actions` silently under-counts a store
+        # with a tampered row in it and the operator reading the terminal sees nothing at all,
+        # which is the number reading as a verdict about a store nobody could fully read.
+        lines.append(_stat("unreadable receipts", document["unreadable_receipts"]))
     lines.append("")
     if document["mode"] != OBSERVE:
         # §6.4 — say what is missing rather than print a line the receipts cannot substantiate.
@@ -980,6 +1015,11 @@ def _stats_lines(document: Mapping[str, Any]) -> list[str]:
             "and ambiguous breakdown is not reported."
         )
     lines.append("Actions still awaiting a human have no receipt yet and are not counted.")
+    if "unreadable_receipts" in document:
+        lines.append(
+            "Some rows could not be read back as receipts and are not counted above. Run "
+            "`ctrlrun receipts --verify-chain` to see where."
+        )
     return lines
 
 
@@ -1778,6 +1818,17 @@ def gateway(
         click.echo("")
 
 
+def _unreadable_line(row: UnreadableReceipt) -> str:
+    """One row this binary could not read back, named where the receipt would have printed.
+
+    SPEC-v0.11 §5.2, and rule 3: one tampered row costs one row. The refusal is printed **by
+    type** and never by message, because the canonicalizer quotes what it refused and a lone
+    surrogate echoed here is a line that cannot be printed.
+    """
+    at = "no seq" if row.seq is None else f"seq {row.seq}"
+    return f"{at}  {row.receipt_id or '-'}  UNREADABLE  this row could not be read ({row.refusal})"
+
+
 def _receipt_line(receipt: Receipt) -> str:
     return (
         f"{iso_timestamp(receipt.finished_at)}  {receipt.receipt_id}  {receipt.action}  "

src/ctrlrun/conformance/store/suites.py

@@ -47,7 +47,7 @@
     NotExecuted,
 )
 from ...policy import Decision
-from ...receipt import Event, EventType, Receipt, ReceiptResult
+from ...receipt import Event, EventType, Receipt, ReceiptResult, UnreadableReceipt
 from ...state import (
     ClockSkew,
     DelegationRecord,
@@ -1766,6 +1766,18 @@ def evidence_receipt(backend: StoreBackend, processes: int = CONTENDERS) -> Case
     back = [held for held in store.receipts() if held.receipt_id == receipt.receipt_id]
     if not back:
         return failed("receipt-round-trip", title, "the receipt did not come back")
+    if isinstance(back[0], UnreadableReceipt):
+        # SPEC-v0.11 §5.2 lets a store hand back a row it cannot construct instead of raising,
+        # so that one tampered row costs one row. **A row this store just wrote is not that
+        # case.** A candidate backend that cannot read back its own write fails here, by name,
+        # rather than falling into the field-by-field diff below and reporting a missing
+        # attribute.
+        return failed(
+            "receipt-round-trip",
+            title,
+            f"the receipt came back as unreadable ({back[0].refusal}); a store that cannot read "
+            "back the receipt it just wrote has not stored it",
+        )
     receipt = written
     # Every field, not two of seventeen. A store that mangled `decision`, `approver`,
     # `arguments`, `attempt` or the timestamps passed this case while the two it compared

src/ctrlrun/control.py

@@ -125,6 +125,7 @@
     JSONLEventSink,
     Receipt,
     ReceiptResult,
+    UnreadableReceipt,
     _WouldHave,
     iso_timestamp,
     new_receipt_id,
@@ -4268,6 +4269,22 @@ def _replay_policy(self, candidate: Policy, *, limit: int) -> list[dict[str, Any
         rows: list[dict[str, Any]] = []
         receipts = list(self._store.receipts())[-limit:]
         for receipt in receipts:
+            if isinstance(receipt, UnreadableReceipt):
+                # SPEC-v0.11 §5.2: a row this binary cannot read back is **skipped and named**,
+                # in the shape this loop already uses for a receipt it cannot rebuild an action
+                # from. It is not dropped: a replay that silently left out the one row somebody
+                # tampered with would report "no decision changes" about a store it could not
+                # read, which is `SPEC-v0.4 §3.8`'s false green.
+                rows.append(
+                    {
+                        "receipt_id": receipt.receipt_id,
+                        "action": None,
+                        "skipped": (
+                            f"this row could not be read back as a receipt ({receipt.refusal})"
+                        ),
+                    }
+                )
+                continue
             rebuilt = _action_from_receipt(receipt, self._environment)
             if rebuilt is None:
                 rows.append(

src/ctrlrun/gateway/operator.py

@@ -52,7 +52,7 @@
     IdentityContext,
     IdentityProvider,
 )
-from ..receipt import Event, EventType, iso_timestamp
+from ..receipt import Event, EventType, UnreadableReceipt, _readable, iso_timestamp
 from ..reporting import (
     effect_document,
     inspection_for,
@@ -790,7 +790,17 @@ def _receipts(self, limit: int, control_id: object) -> dict[str, Any]:
         if control_id is not None:
             # `v0.6 §7.3` — a filter and not a lookup, exactly as `ctrlrun receipts --control`
             # is: an id no document defines matches nothing rather than erroring.
-            found = tuple(receipt for receipt in found if str(control_id) in receipt.controls)
+            #
+            # An unreadable row is kept whatever the filter says (SPEC-v0.11 §5.2): its
+            # `controls` could not be read, so it cannot be shown not to cite this id.
+            found = tuple(
+                receipt
+                for receipt in found
+                if isinstance(receipt, UnreadableReceipt) or str(control_id) in receipt.controls
+            )
+        # SPEC-v0.11 §2.3: this tool is a **network** surface, and one `UPDATE` used to take out
+        # the remote console as well as the terminal. A row that cannot be read back is rendered
+        # as its refusal, in place and at its `seq`, and the rows around it are returned.
         return {"receipts": [receipt.to_dict() for receipt in found[-limit:]]}
 
     def _effects(self, state: object) -> dict[str, Any]:
@@ -812,9 +822,12 @@ def _stats(self, since: object) -> dict[str, Any]:
             boundary = since_boundary(None if since is None else str(since))
         except InvalidArgument as exc:
             raise _Refused(_INVALID_PARAMS, "ctrlrun.invalid_argument", 200, str(exc)) from exc
+        rows = self.store.receipts()
+        # A refused row has no `finished_at` to compare, so it cannot enter a total; it is
+        # counted separately below rather than dropped in silence (SPEC-v0.11 §5.2).
         counted = [
             receipt
-            for receipt in self.store.receipts()
+            for receipt in _readable(rows)
             if boundary is None or receipt.finished_at >= boundary
         ]
         # §9.1 — one producer for `ctrlrun.stats/v1`. T193 asserts equality with the CLI's, and
@@ -825,6 +838,7 @@ def _stats(self, since: object) -> dict[str, Any]:
             mode=self._control.policy.mode,
             boundary=boundary,
             ledger_rows=_ledger_rows(self.store),
+            unreadable=sum(1 for row in rows if isinstance(row, UnreadableReceipt)),
         )
 
     # --- the write tools (§4.5) -----------------------------------------------------------

src/ctrlrun/postgres.py

@@ -74,7 +74,16 @@
     MissingDependency,
 )
 from .migrations import migrate
-from .receipt import RECEIPT_SCHEMA, Event, EventType, Receipt, _document_hash, _stored_receipt
+from .receipt import (
+    RECEIPT_SCHEMA,
+    Event,
+    EventType,
+    Receipt,
+    UnreadableReceipt,
+    _document_hash,
+    _read_receipt,
+    _readable,
+)
 from .state import (
     Charge,
     ClockSkew,
@@ -2148,8 +2157,17 @@ def put_receipt(self, receipt: Receipt) -> Receipt:
                     # branch exists: an advanced head with no row behind it is a permanent gap.
                     # The caller gets the row as it stands, not the one it tried to write.
                     self._rollback(connection)
+                    # `_readable`: this is the *writer* looking for the row it just tried to
+                    # write, and a row this binary cannot read back is not that row. It falls
+                    # through to returning `receipt`, which is what the caller already gets when
+                    # the row is not found (SPEC-v0.11 §5.2).
                     existing = next(
-                        (r for r in self.receipts() if r.receipt_id == receipt.receipt_id), None
+                        (
+                            r
+                            for r in _readable(self.receipts())
+                            if r.receipt_id == receipt.receipt_id
+                        ),
+                        None,
                     )
                     return existing if existing is not None else receipt
                 cursor.execute(
@@ -2161,18 +2179,21 @@ def put_receipt(self, receipt: Receipt) -> Receipt:
         self._commit(connection)
         return replace(chained, hash=digest)
 
-    def receipts(self) -> tuple[Receipt, ...]:
+    def receipts(self) -> tuple[Receipt | UnreadableReceipt, ...]:
+        # SPEC-v0.11 §5.2: `seq` is **selected** and not only ordered by, so a receipt's position
+        # comes from the column rather than from the document a tamperer controls, and a row this
+        # binary cannot construct still has a position to be named at.
         with self._connection().cursor() as cursor:
             cursor.execute(
-                f"SELECT json, hash FROM {self._q}.receipts "
+                f"SELECT seq, json, hash FROM {self._q}.receipts "
                 "ORDER BY seq NULLS FIRST, ts, receipt_id"
             )
             rows = cursor.fetchall()
         # `hash` comes off the column: a document cannot contain its own hash (§6.2). The stored
         # `json` here is `json.dumps(..., sort_keys=True)` and SQLite's is `to_json()`, which are
         # different byte strings -- and the chain does not care, because `chain_hash` recomputes
         # the canonical form from the parsed document rather than hashing whatever was stored.
-        return tuple(_stored_receipt(json.loads(str(row[0])), row[1]) for row in rows)
+        return tuple(_read_receipt(json.loads(str(row[1])), row[2], row[0]) for row in rows)
 
     def chain_head(self) -> tuple[int, str] | None:
         with self._connection().cursor() as cursor: