diff --git a/core/ag402_core/middleware/vibes_coded_guard.py b/core/ag402_core/middleware/vibes_coded_guard.py new file mode 100644 index 0000000..ff59857 --- /dev/null +++ b/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() diff --git a/examples/vibes_coded_observer.py b/examples/vibes_coded_observer.py new file mode 100644 index 0000000..429952f --- /dev/null +++ b/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()