Publish reviewed strict execution consumption confirmation client

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade

approval_engine/execution_confirmation.py

@@ -0,0 +1,96 @@
+"""Exact consumption evidence for an independently authenticated mediator.
+
+This client supplies PIP facts, never an execution admission or PDP decision.
+The mediator still owns claim/PDP validation, deadlines and a one-shot journal.
+"""
+
+from datetime import datetime
+from typing import Any
+from urllib.parse import quote
+
+from .binding import DIGEST_RE, canonical_binding
+from .pep import ApprovalHTTPClient, ApprovalProtocolError, _ID_RE
+
+
+def _inputs(approval_id: str, request_digest: str, decision_id: str) -> None:
+    if not isinstance(approval_id, str) or not _ID_RE.fullmatch(approval_id):
+        raise ApprovalProtocolError("approval id is invalid")
+    if not isinstance(request_digest, str) or not DIGEST_RE.fullmatch(request_digest):
+        raise ApprovalProtocolError("canonical request digest is required")
+    if not isinstance(decision_id, str) or not decision_id.strip():
+        raise ApprovalProtocolError("exact decision id is required")
+
+
+def _evidence(payload: dict[str, Any], request_digest: str, decision_id: str) -> str:
+    if payload.get("request_digest") != request_digest:
+        raise ApprovalProtocolError("consumed request digest does not match")
+    if payload.get("decision_id") != decision_id:
+        raise ApprovalProtocolError("consumed decision id does not match")
+    value = payload.get("consumed_at")
+    try:
+        stamp = datetime.fromisoformat(value) if isinstance(value, str) else None
+    except ValueError:
+        stamp = None
+    if stamp is None or stamp.tzinfo is None or stamp.utcoffset() is None:
+        raise ApprovalProtocolError("consumption time is missing or invalid")
+    if any(key in payload for key in ("allow", "deny", "effect")):
+        raise ApprovalProtocolError("consumption evidence is not a decision")
+    return value
+
+
+class ExecutionConfirmationClient(ApprovalHTTPClient):
+    def consume_exact(
+        self, approval_id: str, request_digest: str, decision_id: str
+    ) -> dict[str, Any]:
+        """One CAS call, with exact response validation and no automatic retry."""
+        _inputs(approval_id, request_digest, decision_id)
+        result = self._request(
+            "POST", f"/v1/approvals/{quote(approval_id, safe='')}/consume",
+            {"request_digest": request_digest, "decision_id": decision_id},
+        )
+        if result.get("approval_id") != approval_id or result.get("status") != "consumed":
+            raise ApprovalProtocolError("exact consumption was not confirmed")
+        _evidence(result, request_digest, decision_id)
+        if type(result.get("idempotent")) is not bool:
+            raise ApprovalProtocolError("idempotent must be an explicit boolean")
+        return {"approval_id": approval_id, "status": "consumed",
+                "request_digest": request_digest, "decision_id": decision_id,
+                "consumed_at": result["consumed_at"], "idempotent": result["idempotent"]}
+
+    def confirm_consumption(
+        self, approval_id: str, request_digest: str, decision_id: str, *,
+        expected_binding: dict[str, Any], expected_pdp_digest: str,
+    ) -> dict[str, str]:
+        """Read the authoritative object; return historical facts only.
+
+        Approval entries may be present on the wire but are not republished.
+        This method is safe for reconciliation after an unknown CAS outcome;
+        reconciliation cannot authorize retrying protected backend work.
+        """
+        _inputs(approval_id, request_digest, decision_id)
+        if not isinstance(expected_pdp_digest, str) or not DIGEST_RE.fullmatch(expected_pdp_digest):
+            raise ApprovalProtocolError("exact approval-free PDP digest is required")
+        try:
+            expected = canonical_binding(expected_binding)
+        except (TypeError, ValueError, KeyError):
+            raise ApprovalProtocolError("expected act binding is invalid") from None
+        obj = self._request("GET", f"/v1/approvals/{quote(approval_id, safe='')}")
+        if obj.get("id") != approval_id or obj.get("status") != "consumed":
+            raise ApprovalProtocolError("durable consumption was not confirmed")
+        if any(key in obj for key in ("allow", "deny", "effect")):
+            raise ApprovalProtocolError("consumption object is not a decision")
+        served = obj.get("binding")
+        try:
+            same_act = isinstance(served, dict) and canonical_binding(served) == expected
+        except (TypeError, ValueError, KeyError):
+            same_act = False
+        if not same_act or served.get("pdp_digest") != expected_pdp_digest:
+            raise ApprovalProtocolError("consumed approval binding does not match")
+        if served.get("pdp_path") is not True or served.get("human_control") is not True:
+            raise ApprovalProtocolError("production execution requires declared PDP and human control")
+        consumption = obj.get("consumption")
+        if not isinstance(consumption, dict):
+            raise ApprovalProtocolError("durable consumption evidence is missing")
+        consumed_at = _evidence(consumption, request_digest, decision_id)
+        return {"approval_id": approval_id, "request_digest": request_digest,
+                "decision_id": decision_id, "consumed_at": consumed_at}