Add bounded source reconciliation counts and clarify receiver dependencies

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a1046e-fdbf-7b93-a267-5f72b9b24526

approval_engine/store.py

@@ -1055,6 +1055,51 @@ class Engine:
             counts[row["class"]] = row["n"]
         return counts
 
+    def reconciliation_counts(self, since: str, until: str) -> dict[str, Any]:
+        """One source snapshot; occurrence and local acknowledgment are distinct.
+
+        Audit Core counts receiver acceptance time. Neither local timestamp
+        proves that time, especially after delayed delivery or a duplicate ack.
+        """
+        bounds = []
+        for value in (since, until):
+            try:
+                bound = parse_iso(value)
+            except (TypeError, ValueError) as exc:
+                raise Unprocessable("since/until must be timezone-aware timestamps") from exc
+            if bound.tzinfo is None:
+                raise Unprocessable("since/until must be timezone-aware timestamps")
+            bounds.append(bound.astimezone(timezone.utc))
+        start, end = bounds
+        if start >= end:
+            raise Unprocessable("since must precede until")
+        try:
+            # A single statement holds a consistent snapshot while the worker
+            # acknowledges rows. No payload or identity data is selected.
+            rows = self._conn().execute(
+                "SELECT class, created_at, drained_at FROM outbox"
+            ).fetchall()
+        except sqlite3.Error as exc:
+            raise StoreUnavailable(str(exc)) from exc
+        counts = {c: {"committed": 0, "drained": 0, "pending": 0,
+                      "acknowledged_in_window": 0} for c in EVENT_CLASSES}
+        for row in rows:
+            entry = counts[row["class"]]
+            if start <= parse_iso(row["created_at"]) < end:
+                entry["committed"] += 1
+                entry["pending" if row["drained_at"] is None else "drained"] += 1
+            if row["drained_at"] is not None and start <= parse_iso(row["drained_at"]) < end:
+                entry["acknowledged_in_window"] += 1
+        return {
+            "source": "approval-engine", "tenant": self.tenant,
+            "since": start.isoformat(), "until": end.isoformat(),
+            "bounds": "since-inclusive-until-exclusive",
+            "window_basis": "created_at", "counts": counts,
+            "means": "Source occurrence counts and local acknowledgments; receiver "
+                     "acceptance-time comparison requires drain-lag and boundary "
+                     "accounting. Agreement does not prove completeness.",
+        }
+
     def outbox_stats(self) -> dict[str, Any]:
         conn = self._conn()
         now = self.now()