feat(middleware): Vibes-Coded trust guard — pre-settlement screening + signed action-receipts

core/ag402_core/middleware/vibes_coded_guard.py

@@ -0,0 +1,152 @@
+"""Vibes-Coded trust guard — pre-settlement screening + signed action-receipts.
+
+Drop-in ag402 middleware (lives alongside ``budget_guard``). Before an irreversible
+action is settled, it screens the action through Vibes-Coded's hosted pay-per-call
+guards (idempotency / spend / contract-pin / agent-state) and, on allow, mints a
+signed, offline-verifiable action-receipt.
+
+The guard calls are x402 402s — ag402's own payment middleware pays them transparently,
+so this integrates as a pure trust layer: the agent only pays a tiny per-call USDC fee
+when an action is actually screened. No seats, no minimums.
+
+Commercials: pay-per-call x402 (Agent Economy > Vibes-Coded on vibes-coded.com).
+"""
+
+from __future__ import annotations
+
+import logging
+import uuid
+from dataclasses import dataclass
+from typing import Any
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+# Hosted guard endpoints (x402 pay-per-call). The preimage the action-receipt binds
+# is the frozen action tuple, so any downstream agent can recompute it offline.
+VIBES_ORIGIN = "https://vibes-coded-production.up.railway.app"
+ACTION_RECEIPT_URL = f"{VIBES_ORIGIN}/api/v1/outcomes/action-receipt"
+
+# Map an action intent -> the Vibes-Coded guard that should gate it.
+GUARD_ROUTES: dict[str, str] = {
+    "payment_or_transfer": "idempotency-guard",   # replay / double-spend
+    "stateful_action": "agent-state-guard",        # pre-action reliability gate
+    "idempotent_call": "idempotency-guard",
+    "external_write": "memory-exfil-guard",        # data-leak / exfil screen
+    "value_transfer": "rate-limit-guard",          # amount/velocity anomaly
+    "contract_call": "contract-pin-guard",         # runtime contract-drift gate
+}
+
+
+@dataclass
+class GuardResult:
+    allowed: bool
+    reason: str = ""
+    guard: str | None = None
+    receipt_id: str | None = None
+    detail: dict[str, Any] = None  # type: ignore[assignment]
+
+
+class VibesCodedGuard:
+    """Screen irreversible actions through Vibes-Coded's hosted trust guards.
+
+    Usage::
+
+        guard = VibesCodedGuard(sandbox_key=AG402_SANDBOX_KEY)
+        result = await guard.screen(intent="payment_or_transfer",
+                                    payload={"amount_usdc": 0.50, "to": "..."})
+        if not result.allowed:
+            raise PermissionError(result.reason)   # fail closed
+        # settle via ag402, then result.receipt_id proves what happened
+    """
+
+    def __init__(
+        self,
+        origin: str = VIBES_ORIGIN,
+        timeout: float = 20.0,
+        sandbox_key: str = "",
+        emit_receipt: bool = True,
+    ) -> None:
+        self._origin = origin.rstrip("/")
+        self._timeout = timeout
+        self._sandbox_key = sandbox_key
+        self._emit_receipt = emit_receipt
+        self._client = httpx.Client(timeout=timeout)
+
+    # -- helpers ---------------------------------------------------------------
+
+    def _headers(self) -> dict[str, str]:
+        if self._sandbox_key:
+            return {"X-Ag402-Sandbox-Key": self._sandbox_key}
+        return {}
+
+    def _guard_slug(self, intent: str) -> str | None:
+        return GUARD_ROUTES.get(intent)
+
+    def _emit_receipt(self, intent: str, payload: dict[str, Any], guard: str) -> str | None:
+        import hashlib
+
+        digest = payload.get("payload_digest") or hashlib.sha256(
+            repr(sorted(payload.items())).encode()
+        ).hexdigest()
+        try:
+            resp = self._client.post(
+                f"{self._origin}/api/v1/outcomes/action-receipt",
+                json={
+                    "agent_id": payload.get("agent_id", "observer-agent"),
+                    "action": intent,
+                    "payload_digest": digest,
+                    "nonce": uuid.uuid4().hex,
+                    "quote": guard,
+                },
+                headers=self._headers(),
+            )
+            data = resp.json()
+            return data.get("receipt_id")
+        except Exception as exc:  # receipt is best-effort; never block on it
+            logger.warning("[VIBES] receipt mint failed: %s", exc)
+            return None
+
+    # -- main entry ------------------------------------------------------------
+
+    def screen(self, intent: str, payload: dict[str, Any]) -> GuardResult:
+        """Run the mapped guard; block if it rejects. Fail closed on any error."""
+        guard = self._guard_slug(intent)
+        if not guard:
+            # No guard maps to this intent — advisory allow (observer is opt-in here).
+            return GuardResult(allowed=True, reason="no guard mapped", guard=None)
+
+        try:
+            resp = self._client.post(
+                f"{self._origin}/api/v1/outcomes/{guard}",
+                json={"action": intent, "context": payload},
+                headers=self._headers(),
+            )
+            resp.raise_for_status()
+            data = resp.json()
+        except Exception as exc:  # network/parse failure -> fail closed
+            return GuardResult(allowed=False, reason=f"guard call failed: {exc}", guard=guard)
+
+        allowed = bool(data.get("allowed", False))
+        if not allowed:
+            return GuardResult(
+                allowed=False,
+                reason=data.get("reason", "rejected"),
+                guard=guard,
+                detail=data,
+            )
+
+        receipt_id = None
+        if self._emit_receipt:
+            receipt_id = self._emit_receipt(intent, payload, guard)
+        return GuardResult(
+            allowed=True,
+            reason=data.get("reason", "ok"),
+            guard=guard,
+            receipt_id=receipt_id,
+            detail=data,
+        )
+
+    def close(self) -> None:
+        self._client.close()

examples/vibes_coded_observer.py

@@ -0,0 +1,198 @@
+"""
+Vibes-Coded Observer Middleware — ag402 integration example
+===========================================================
+
+Wraps an agent's *irreversible* actions (payments, transfers, stateful writes)
+so a trust check runs against Vibes-Coded's pay-per-call x402 guard APIs
+*before* ag402 settles the action.
+
+ag402 already auto-pays any HTTP 402 (see `ag402_core.enable()`), so the guard
+calls below cost a tiny per-call USDC fee and require zero payment code here.
+
+## Setup (3 steps)
+1. Install:        pip install ag402-core httpx
+2. Export keys:    export AG402_SANDBOX_KEY="op_sk_sandbox_ag402_..."   # from ag402 team
+                    export VIBES_ORIGIN="https://vibes-coded-production.up.railway.app"
+3. Run:            python examples/vibes_coded_observer.py
+
+## What it does
+For every outgoing action the agent intends to take, `observe()` classifies the
+intent and, if it maps to a guard, calls that guard endpoint. A guard returning
+`allowed: false` blocks the action (the agent must justify / back off). A guard
+returning `allowed: true` lets ag402 proceed to settlement.
+
+Guards (all hosted, x402 pay-per-call):
+  - idempotency-guard     -> replay / double-spend protection
+  - agent-state-guard     -> pre-action reliability gate
+  - memory-exfil-guard    -> data-leak / exfil screen
+  - rate-limit-guard      -> amount/velocity anomaly check
+  - action-receipt        -> signed, replayable post-settlement receipt (offline-verifiable)
+"""
+
+from __future__ import annotations
+
+import hashlib
+import os
+import uuid
+from dataclasses import dataclass, field
+from typing import Any, Callable
+
+import ag402_core
+import httpx
+
+VIBES_ORIGIN = os.getenv("VIBES_ORIGIN", "https://vibes-coded-production.up.railway.app").rstrip("/")
+AG402_SANDBOX_KEY = os.getenv("AG402_SANDBOX_KEY", "")
+
+# Map an action intent -> the Vibes-Coded guard endpoint that should gate it.
+# Inbound actions an Observer should screen before ag402 settles them.
+# All slugs below are live endpoints (see https://vibes-coded.com/for-agents).
+GUARD_ROUTES: dict[str, str] = {
+    "payment_or_transfer": "idempotency-guard",          # replay / double-spend protection
+    "stateful_action": "agent-state-guard",              # pre-action reliability gate
+    "idempotent_call": "idempotency-guard",              # same-spend guard
+    "external_write": "memory-exfil-guard",              # data-leak / exfil screen
+    "value_transfer": "rate-limit-guard",                # amount/velocity anomaly check
+}
+
+
+def emit_receipt(action: Action, observation: Observation) -> dict | None:
+    """After a gated action is allowed + settled, mint a signed action-receipt.
+
+    This is the post-settlement accountability layer impeachmentright / giskard09
+    called for: a machine-checkable, replayable record of who did what, bound to the
+    action tuple + observer verdict. Verifiable offline against our published key,
+    so a downstream agent or auditor can trust the handoff without our service.
+    """
+    if not observation.allowed:
+        return None
+    payload_digest = action.payload.get("payload_digest") or hashlib.sha256(
+        repr(sorted(action.payload.items())).encode()
+    ).hexdigest()
+    url = f"{VIBES_ORIGIN}/api/v1/outcomes/action-receipt"
+    try:
+        resp = httpx.post(
+            url,
+            json={
+                "agent_id": action.payload.get("agent_id", "observer-agent"),
+                "action": action.intent,
+                "payload_digest": payload_digest,
+                "nonce": uuid.uuid4().hex,          # unique per action -> replay-safe
+                "quote": observation.guard,          # the guard that approved it
+            },
+            headers={"X-Ag402-Sandbox-Key": AG402_SANDBOX_KEY} if AG402_SANDBOX_KEY else {},
+            timeout=20.0,
+        )
+        return resp.json()
+    except Exception:
+        return None
+
+
+@dataclass
+class Action:
+    """An agent action about to be settled by ag402."""
+    intent: str
+    payload: dict[str, Any] = field(default_factory=dict)
+    # Optional: a callable that actually executes the action (after guards pass).
+    execute: Callable[[], Any] | None = None
+
+
+@dataclass
+class Observation:
+    allowed: bool
+    reason: str
+    guard: str | None = None
+    detail: dict[str, Any] = field(default_factory=dict)
+
+
+class VibesObserver:
+    """Lightweight Observer that gates irreversible actions via Vibes-Coded guards."""
+
+    def __init__(self, origin: str = VIBES_ORIGIN, timeout: float = 20.0) -> None:
+        self.origin = origin.rstrip("/")
+        self.timeout = timeout
+        self._client = httpx.Client(timeout=timeout)
+
+    def _guard_slug(self, intent: str) -> str | None:
+        return GUARD_ROUTES.get(intent)
+
+    def observe(self, action: Action) -> Observation:
+        """Run the guard for `action.intent`. Block if the guard rejects."""
+        guard = self._guard_slug(action.intent)
+        if not guard:
+            # No guard maps to this intent — allow (observer is advisory here).
+            return Observation(allowed=True, reason="no guard mapped", guard=None)
+
+        url = f"{self.origin}/api/v1/outcomes/{guard}"
+        # ag402_core.enable() auto-pays the 402 on this POST; we just send the
+        # action context. The guard returns {"allowed": bool, "reason": str, ...}.
+        try:
+            resp = self._client.post(
+                url,
+                json={"action": action.intent, "context": action.payload},
+                headers={"X-Ag402-Sandbox-Key": AG402_SANDBOX_KEY} if AG402_SANDBOX_KEY else {},
+            )
+            resp.raise_for_status()
+            data = resp.json()
+        except Exception as exc:  # network/parse failure -> fail closed
+            return Observation(allowed=False, reason=f"guard call failed: {exc}", guard=guard)
+
+        allowed = bool(data.get("allowed", False))
+        return Observation(
+            allowed=allowed,
+            reason=data.get("reason", "ok" if allowed else "rejected"),
+            guard=guard,
+            detail=data,
+        )
+
+    def run(self, action: Action) -> Any:
+        """Observe, and only execute the action if guards allow it.
+        On success, mints a signed action-receipt for post-settlement accountability.
+        """
+        obs = self.observe(action)
+        if not obs.allowed:
+            raise PermissionError(f"[VibesObserver] blocked '{action.intent}': {obs.reason}")
+        result = action.execute() if action.execute else obs
+        receipt = emit_receipt(action, obs)
+        if receipt:
+            print(f"[RECEIPT] {receipt.get('receipt_id')} (verify: {receipt.get('verify_endpoint')})")
+        return result
+
+    def close(self) -> None:
+        self._client.close()
+
+
+# ---------------------------------------------------------------------------
+# Demo: a payment action gated by the orchestration guard, then settled by ag402
+# ---------------------------------------------------------------------------
+
+def _demo_payment() -> dict:
+    """Simulated irreversible action ag402 would settle."""
+    return {"settled": True, "amount_usdc": 0.50, "to": "9xQeWvG816bUx9EPjHmaT23yFs2nA8Lg4bQ9Z"}
+
+
+def main() -> None:
+    print("\n=== Vibes-Coded Observer + ag402 ===\n")
+    print("ag402 auto-payment enabled — guard 402s are paid transparently.\n")
+    ag402_core.enable()
+
+    obs = VibesObserver()
+
+    # 1) A payment the Observer should screen first.
+    payment = Action(intent="payment_or_transfer", payload={"amount_usdc": 0.50, "to": "9xQeWvG816bUx9EPjHmaT23yFs2nA8Lg4bQ9Z"}, execute=_demo_payment)
+    try:
+        result = obs.run(payment)
+        print(f"[ALLOWED] payment settled: {result}")
+    except PermissionError as e:
+        print(f"[BLOCKED] {e}")
+
+    # 2) A benign read that maps to no guard -> advisory allow.
+    read = Action(intent="read_only", payload={"resource": "market_data"})
+    o = obs.observe(read)
+    print(f"[ADVISORY] {read.intent}: allowed={o.allowed} ({o.reason})")
+
+    obs.close()
+    ag402_core.disable()
+
+
+if __name__ == "__main__":
+    main()