#!/usr/bin/env python3 """ Reference consumer for the Agent Correction Format (ACF). A platform-side sample: emitter plus consumer is a protocol, and this is the consumer half. It acts on a checker's output the way ACF v0.2 Sections 5.3, 6, 7 and 8 require: * MATCH -> proceed. * BLOCK with fixes -> apply ONLY repairs that bring the subject INTO the buyer's existing authorization, resubmit as a new check carrying previous_record_id, within max_attempts. * REVIEW, an unknown code/action, or any repair that would exceed the authorization -> STOP and escalate to the buyer. Never auto-apply, never guess, never widen the mandate. It runs against the reference implementation's API on a local test instance with a throwaway data directory (the API emits the v0.1 payload today; the consumer rules applied here are the v0.2 rules). Nothing here touches production and nothing changes checker behavior. api/.venv/bin/python standards/reference-consumer.py """ import json import os import subprocess import sys import tempfile import time import urllib.error import urllib.request from datetime import datetime, timedelta, timezone from pathlib import Path HERE = Path(__file__).resolve().parent API_DIR = HERE.parent / "api" PORT = 8811 BASE = f"http://127.0.0.1:{PORT}" SECRET = "reference-consumer-secret" FUTURE = (datetime.now(timezone.utc) + timedelta(days=30)) \ .isoformat().replace("+00:00", "Z") API_KEY = None KNOWN_ACTIONS = { "remove_item", "replace_item", "change_variant", "reduce_quantity", "reduce_total", "change_currency", "correct_traveler_name", "change_airport", "change_travel_date", "change_departure_time", "add_itinerary", "rebuild_itinerary", "change_cabin", "change_fare_terms", "add_hotel_stay", "change_room", "change_provider", "change_location", "change_date", "change_time", "change_party_size", "correct_reservation_name", "reduce_deposit", "change_cancellation_terms", } def http(method, path, body=None, headers=None): data = json.dumps(body).encode() if body is not None else None h = {"Content-Type": "application/json"} if API_KEY: h["X-API-Key"] = API_KEY if headers: h.update(headers) req = urllib.request.Request(BASE + path, data=data, method=method, headers=h) try: with urllib.request.urlopen(req, timeout=15) as resp: return resp.status, json.loads(resp.read()) except urllib.error.HTTPError as e: try: return e.code, json.loads(e.read() or b"{}") except Exception: return e.code, {} def start_server(data_dir): env = dict(os.environ) env["INTENTFIT_DATA_DIR"] = data_dir env["INTENTFIT_SECRET"] = SECRET env.pop("INTENTFIT_ADMIN_SECRET", None) env.pop("ANTHROPIC_API_KEY", None) # rules-only, deterministic proc = subprocess.Popen( [sys.executable, "-m", "uvicorn", "app:app", "--host", "127.0.0.1", "--port", str(PORT)], cwd=str(API_DIR), env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) for _ in range(120): try: s, _ = http("GET", "/health") if s == 200: return proc except Exception: time.sleep(0.25) proc.kill() raise RuntimeError("server did not come up") # ------------------------------------------------------------------ # The consumer rules (ACF v0.2 Sections 5.3, 6, 7, 8) # ------------------------------------------------------------------ def gate_correction(correction, mandate): """Decide one correction. Returns 'apply', 'skip' or 'escalate'.""" if correction.get("requires_buyer_approval"): return "escalate" # Section 8: never auto-applied, by flag if correction.get("action") not in KNOWN_ACTIONS: return "escalate" # Section 5.3: unknown action, do not guess return "apply" def within_authorization(cart, mandate): """A repair may be auto-applied only if the rebuilt subject sits inside the authorization the buyer already gave (Section 8).""" if cart["total_price"] > mandate["max_total_price"]: return False qty = sum(i["quantity"] for i in cart["items"]) if qty > mandate["max_quantity"]: return False name = " ".join(i["name"].lower() for i in cart["items"]) return all(k in name for k in ("sony", "wh-1000xm5")) def consume(verdict, mandate, submit): """The correction loop. `submit(cart)` performs a fresh check. Returns the final verdict, or 'ESCALATED'.""" attempts_left = verdict.get("max_attempts", 3) current = verdict while current["verdict"] == "BLOCK" and attempts_left > 0: corrections = current.get("corrections") or [] plan = [] for c in corrections: decision = gate_correction(c, mandate) print(f" correction {c['action']:<16} " f"dimension={c['dimension']:<12} -> {decision}") if decision == "escalate": print(" ESCALATE: a correction needs the buyer or is " "not understood. Stopping.") return "ESCALATED" plan.append(c) if not any(c["action"] == "replace_item" for c in plan): print(" ESCALATE: no understood repair path to the " "instructed item. Stopping.") return "ESCALATED" # Rebuild to the instructed item. In production the agent # re-shops; this sample's store lists the instructed item at # $279, inside the $300 ceiling. rebuilt = { "mandate_id": mandate["id"], "merchant": {"name": "Example Shop", "domain": "shop.example.com"}, "items": [{"name": "Sony WH-1000XM5 Wireless Noise " "Cancelling Headphones", "quantity": 1, "unit_price": 279.0}], "total_price": 279.0, "currency": "USD", "previous_record_id": current["record_id"], } if not within_authorization(rebuilt, mandate): print(" ESCALATE: the rebuilt cart would exceed the " "buyer's authorization. Stopping.") return "ESCALATED" attempts_left -= 1 print(f" resubmitting corrected cart as a new check " f"(previous_record_id carried; attempts left after " f"this: {attempts_left})") _, current = submit(rebuilt) print(f" -> verdict: {current['verdict']}") return current["verdict"] def main(): global API_KEY data_dir = tempfile.mkdtemp(prefix="acf-consumer-") proc = start_server(data_dir) try: s, k = http("POST", "/admin/api-keys", {"label": "reference-consumer"}, headers={"X-Admin-Secret": SECRET}) assert s == 200, (s, k) API_KEY = k["api_key"] s, m = http("POST", "/mandates", { "user_id": "consumer-demo-buyer", "item_description": "Sony WH-1000XM5 wireless noise " "cancelling headphones", "item_keywords": ["sony", "wh-1000xm5", "headphones", "noise", "cancelling"], "category": "electronics", "max_total_price": 300.00, "currency": "USD", "merchant_allowlist": ["shop.example.com"], "merchant_blocklist": ["evil.example.com"], "max_quantity": 1, "expires_at": FUTURE}) assert s == 200, (s, m) mandate = {"id": m["id"], "max_total_price": 300.00, "max_quantity": 1} bad_cart = { "mandate_id": m["id"], "merchant": {"name": "Example Shop", "domain": "shop.example.com"}, "items": [{"name": "BassPro Q9 Wireless Headphones", "quantity": 2, "unit_price": 189.0}], "total_price": 378.0, "currency": "USD"} s, v = http("POST", "/verify", bad_cart) assert s == 200, (s, v) print(f"1) Agent built 2x BassPro Q9 at $378 against a $300 " f"Sony mandate.\n Checker verdict: {v['verdict']} " f"(record {v['record_id'][:8]}...)") print("2) Consumer evaluates each correction:") final = consume(v, mandate, lambda cart: http("POST", "/verify", cart)) print(f"3) Correction loop outcome: {final}") assert final == "MATCH", final print("\n4) Unknown-code discipline (fixture f08):") f08 = json.loads((HERE / "fixtures" / "f08-consumer-unknown-code-and-action.json") .read_text()) msg = f08["response"]["messages"][0] decision = gate_correction( {"action": msg["action"], "requires_buyer_approval": msg["requires_buyer_approval"]}, mandate) print(f" message code={msg['code']} action={msg['action']} " f"-> {decision}") assert decision == "escalate" print("\n5) Beyond-authorization repair is refused:") greedy = {"verdict": "BLOCK", "max_attempts": 3, "record_id": "synthetic", "corrections": [{ "dimension": "price", "action": "reduce_total", "expected": 350.0, "actual": 378.0, "reason": "synthetic: a 'fix' priced above the " "buyer's $300 ceiling"}]} rebuilt = {"total_price": 350.0, "items": [{"name": "Sony WH-1000XM5", "quantity": 1}]} ok = within_authorization(rebuilt, mandate) print(f" rebuilt total $350 vs ceiling $300 -> " f"within_authorization={ok} -> escalate") assert ok is False print("\nREFERENCE CONSUMER: ALL BEHAVIORS AS SPECIFIED") return 0 finally: proc.terminate() try: proc.wait(timeout=5) except Exception: proc.kill() if __name__ == "__main__": sys.exit(main())