diff --git a/docs/PWA.md b/docs/PWA.md new file mode 100644 index 0000000..0a7e63a --- /dev/null +++ b/docs/PWA.md @@ -0,0 +1,189 @@ +# Switchboard Lab PWA + the "Switchboard Plugin" Proposal + +**Status:** Draft v1 +**App shell:** [`web/manifest.json`](../web/manifest.json) · [`web/sw.js`](../web/sw.js) +**Registration:** [`web/lab/shared.js`](../web/lab/shared.js) (lab pages) · [`web/index.html`](../web/index.html) (root) +**Verified by:** [`tests/test_pwa.py`](../tests/test_pwa.py) + +--- + +## 1. Why a PWA + +The Switchboard Lab is a static, no-build set of pages. Making it a **Progressive +Web App** gets us three things at zero infra cost: + +1. **Installable** — agents-payments demos sit one tap away on a phone home screen + or a desktop dock, in a standalone window with no browser chrome. +2. **Offline-capable** — the whole lab (every scene, the canvas, the docs) loads + with no network. Good for conference Wi-Fi, planes, and demo booths. +3. **A distribution surface** — the same app shell is the reference for embedding + Switchboard payments into a *third-party* PWA (see §4, the "switchboard plugin"). + +This is intentionally additive: the lab still works as plain static HTML if the +service worker never registers (e.g. served from `file://`). + +## 2. What shipped + +| File | Role | +|------|------| +| `web/manifest.json` | Web App Manifest — name, icons, `start_url`, `display: standalone`, shortcuts | +| `web/icon.svg`, `web/icon-maskable.svg` | Vector icons (`any` + `maskable` purposes), obsidian + gold brand mark | +| `web/sw.js` | Service worker — precache app shell, offline navigation fallback, runtime caching | +| `web/index.html` | Root registers the SW at scope `./` + links the manifest | +| `web/lab/shared.js` | Every lab page registers the SW (`../sw.js`) and injects the manifest link | + +### Manifest highlights + +- `scope: "./"` and `start_url: "./lab/index.html"` — the installed app opens on + the lab dashboard but controls the entire `web/` tree. +- `display: "standalone"` with `display_override` falling back to `minimal-ui`. +- `shortcuts` — long-press / right-click the installed icon jumps straight to + **Agentic Pay + Swap**, the **Canvas Lab**, or the **x402 Paywall** scene. +- `theme_color` / `background_color` `#06060b` match the lab's obsidian splash so + the launch transition is seamless. + +## 3. Install flow + +``` +┌──────────────┐ browser detects ┌──────────────────┐ user installs ┌────────────────┐ +│ open lab in │ manifest + SW + │ install prompt │ ───────────────▶ │ standalone app │ +│ Chrome/Edge │ ─ HTTPS + icons ────▶ │ (omnibox / menu) │ │ on home / dock │ +└──────────────┘ └──────────────────┘ └────────────────┘ + │ │ + │ first load: SW `install` event precaches the app shell ──────────────────────┘ + │ subsequent loads: served from cache, revalidated in the background + ▼ + works fully offline (every scene + docs) +``` + +**Manual install:** +- **Desktop Chrome/Edge:** open `web/lab/index.html` over HTTPS (or `localhost`), + click the install icon in the address bar (or *⋮ → Install Switchboard*). +- **Android Chrome:** *⋮ → Add to Home screen / Install app*. +- **iOS Safari:** *Share → Add to Home Screen* (Safari reads the manifest name + + `apple-touch-icon`; service-worker offline support applies in standalone mode). + +**Local dev / verifying offline:** + +```bash +cd web && python3 -m http.server 8731 +# open http://localhost:8731/lab/index.html, let it load once, +# then DevTools → Network → Offline, and reload — the lab still renders. +``` + +> Service workers require a secure context: `https://` **or** `http://localhost`. +> Served from `file://`, the lab degrades gracefully to plain static pages +> (registration is guarded with `location.protocol !== 'file:'`). + +## 4. Offline architecture + +The service worker (`web/sw.js`) uses three strategies keyed off the request: + +1. **App-shell precache** (`install`): the lab pages, `shared.css`/`shared.js`, the + icons, and the root pages are fetched with `cache: "reload"` and stored under a + versioned cache (`switchboard-lab-v1`). Adds are resilient — a single 404 does + not abort the whole install. +2. **Navigations → network-first** with a cache fallback, then the offline shell + (`./lab/index.html`). So a brand-new route works online and a previously-visited + route works offline. +3. **Same-origin assets → stale-while-revalidate**; **cross-origin (Google Fonts) + → cache-first** so type renders offline. + +`activate` deletes stale caches; bump `CACHE_VERSION` on deploy to invalidate. A +page can post `SKIP_WAITING` to adopt a new worker immediately. + +## 5. The "switchboard plugin" PWA proposal + +The lab PWA doubles as the **reference embedding** for shipping Switchboard +payments inside *someone else's* PWA — a wallet, a marketplace, an agent console. +The idea: a drop-in **"switchboard plugin"** an app installs once and then calls to +gate features behind agent payments. + +### 5.1 Shape + +``` +host PWA (installed) + │ + ├─ + │ registers a SECOND service worker scoped to /pay/* + │ (or a module imported by the host SW) that: + │ • intercepts fetches that come back 402 + │ • parses the x402 PaymentRequirements (switchboard/x402) + │ • drives the on-chain pay/escrow flow + │ • retries with the X-Payment proof header + │ + └─ UI: an install-time permission ("allow agentic payments up to N USDC/day") + backed by the gas-budget primitive (switchboard.gas_tracker) +``` + +The plugin reuses the exact wire types the Python library defines so host and +agent speak the same protocol: + +- **402 challenge / proof** — `switchboard/x402/server.py` + (`PaymentRequirements`, `X-Payment` / `X-Payment-Proof`, `WWW-Authenticate: x402`). +- **Escrow settlement** — `src/payment_protocol.py` + `contracts/AgentEscrow.sol`. +- **Spend caps** — `switchboard.gas_tracker.GasTracker` enforces the per-hour / + per-day budget the user grants at install. +- **Agentic swap** — after receiving funds, route through SafeSwap exactly as in + [`examples/agentic_demo`](../examples/agentic_demo/) (`SafeSwapClient`). + +### 5.2 Install + consent flow + +``` +1. user installs the host PWA (manifest + SW) +2. host PWA imports the switchboard plugin +3. plugin shows a one-time consent sheet: + "Switchboard may pay agents on your behalf, up to 20 USDC/day, + only to recipients you approve. Funds settle on-chain via escrow." +4. consent persists the budget + allowlist (IndexedDB) +5. from then on, any fetch the host makes that returns 402 is auto-paid + within budget — fully offline-first for the UI, on-chain for settlement +``` + +This mirrors the policy gate already implemented server-agnostically in +`X402Middleware._validate_offer()` (cap, recipient allowlist, gas budget) — the +plugin is that check, moved into the browser. + +### 5.3 How the plugin embeds switchboard payments + +A host page gates a paid feature with a single call: + +```js +import { switchboardPay } from './switchboard-plugin.js'; + +// fetch a paid agent endpoint; the plugin handles the 402 → pay → retry loop +const res = await switchboardPay('https://agent-b.example/v1/inference', { + method: 'POST', + body: JSON.stringify(job), + // policy comes from install-time consent; can be tightened per call + maxUsd: 5, + allow: ['0xB0b0…'], +}); +``` + +Under the hood that is the browser twin of the Python demo: parse the offer, +validate against the budget, settle into escrow, retry with proof — and +optionally route the proceeds through SafeSwap. The lab's +[Agentic Pay + Swap scene](../web/lab/swap.html) is the visual spec for exactly +this loop. + +### 5.4 Roadmap + +| Step | Deliverable | +|------|-------------| +| 1 | `switchboard-plugin.js` — `switchboardPay()` + 402 interception (ships the §5.3 API) | +| 2 | Consent sheet + IndexedDB-backed budget/allowlist (the in-browser `GasTracker`) | +| 3 | Wallet binding (EIP-1193 / EIP-7702 smart-account) for real on-chain settlement | +| 4 | SafeSwap routing of received funds, surfaced as an optional auto-rebalance | +| 5 | Publish alongside [`@kcolbchain/eliza-switchboard`](../packages/plugin-switchboard) as a browser counterpart | + +## 6. Verification + +```bash +PYTHONPATH=. python -m pytest tests/test_pwa.py -q +``` + +Asserts the manifest is valid + complete, its icons and `start_url`/shortcut +targets exist, the service worker is valid JS that precaches a real on-disk app +shell with an offline fallback, and that registration is wired into both the root +page and every lab page. diff --git a/examples/agentic_demo/README.md b/examples/agentic_demo/README.md new file mode 100644 index 0000000..a30a025 --- /dev/null +++ b/examples/agentic_demo/README.md @@ -0,0 +1,75 @@ +# Agentic Payments Demo — A2A pay + escrow settle + SafeSwap route + +A runnable scenario where **Agent A pays Agent B for work** via the Switchboard +x402 middleware + on-chain escrow, settles on delivery, then **Agent B routes the +received token through [SafeSwap](#safeswap)** to rebalance into a target asset. + +Everything runs **offline** — no RPC node, no live SafeSwap — by driving the real +`switchboard` package surface against an in-memory chain and an in-process SafeSwap +mock. Swapping in a live RPC `PaymentClient` and `SafeSwapClient(base_url=...)` +needs no changes to the scenario code. + +## Run it + +```bash +PYTHONPATH=. python examples/agentic_demo/run.py +PYTHONPATH=. python examples/agentic_demo/run.py --swap-to LUX --price 8 --json +``` + +Exit code is `0` only when **both** `402 offer -> pay -> settle` and the +**agentic swap** succeed. + +## The flow + +``` +Agent A Agent B (paid endpoint) SafeSwap + │ GET /inference ───────────▶ │ + │ ◀─────────── 402 + x402 PaymentOffer (escrow, 5 USDC) │ + │ validate offer (cap / allowlist / gas budget) │ + │ lock 5 USDC in AgentEscrow ──▶ [Locked] │ + │ ◀─────────── 200 OK + deliverable │ + │ confirmPayment() ──────────▶ escrow [Released] → B paid │ + │ route 5 USDC ─────────────────▶│ quote + │ ◀────────────── best route + amountOut + │ execute ─────────────────────▶ │ settle + │ ◀────────────── SwapReceipt │ +``` + +1. **402 offer** — `AgentBEndpoint.offer()` returns a real + `switchboard.x402_middleware.PaymentOffer` with the **escrow** scheme. +2. **validate + pay** — Agent A's `X402Middleware._validate_offer()` enforces the + payment cap, recipient allowlist, and gas budget, then `_pay_onchain()` locks + funds via the escrow `create_payment()` path. +3. **deliver** — Agent B serves the work against the payment proof. +4. **settle** — Agent A `confirm_payment()` → escrow transitions + `Locked → Released`, crediting Agent B. +5. **agentic swap** — Agent B routes the received USDC through `SafeSwapClient` + (`quote` → `execute`) into ETH/LUX, getting a best-execution route + receipt. + +## SafeSwap + +`safeswap.py` is a tiny client against SafeSwap's orchestrator HTTP API +(`/v1/quote`, `/v1/execute`). It ships with `MockSafeSwapOrchestrator`, an +in-process transport with deterministic pricing so the demo and tests run with no +network. Point `SafeSwapClient(base_url=...)` at the live orchestrator for real +routing. + +## Files + +| File | Role | +|------|------| +| `run.py` | CLI entrypoint (`--swap-to`, `--price`, `--json`) | +| `scenario.py` | the orchestration + `BudgetGuard` + `AgentBEndpoint` | +| `onchain.py` | `MockChain` ledger + escrow + `MockPaymentClient` (PaymentClient surface) | +| `safeswap.py` | `SafeSwapClient` + `MockSafeSwapOrchestrator` | + +## Test + +```bash +PYTHONPATH=. python -m pytest tests/test_agentic_demo.py -q +``` + +Asserts: the 402 offer carries the escrow scheme + price, the escrow ends +`Released` (not just `Locked`), funds move payer → escrow → payee, the SafeSwap +orchestrator is genuinely called (`quote` then `execute`), and the swap routes +with a non-empty venue path and positive output. diff --git a/examples/agentic_demo/__init__.py b/examples/agentic_demo/__init__.py new file mode 100644 index 0000000..9c47198 --- /dev/null +++ b/examples/agentic_demo/__init__.py @@ -0,0 +1,34 @@ +"""Agentic payments demo: Agent A pays Agent B via x402 + escrow, then B routes +the received token through SafeSwap. + +Run with:: + + PYTHONPATH=. python examples/agentic_demo/run.py +""" + +from .safeswap import ( + MockSafeSwapOrchestrator, + SafeSwapClient, + SafeSwapError, + SwapQuote, + SwapReceipt, + SwapRequest, +) +from .onchain import EscrowState, MockChain, MockPaymentClient +from .scenario import AgentBEndpoint, ScenarioResult, StepLog, run_scenario + +__all__ = [ + "run_scenario", + "ScenarioResult", + "StepLog", + "AgentBEndpoint", + "MockChain", + "MockPaymentClient", + "EscrowState", + "SafeSwapClient", + "MockSafeSwapOrchestrator", + "SafeSwapError", + "SwapRequest", + "SwapQuote", + "SwapReceipt", +] diff --git a/examples/agentic_demo/onchain.py b/examples/agentic_demo/onchain.py new file mode 100644 index 0000000..e59c708 --- /dev/null +++ b/examples/agentic_demo/onchain.py @@ -0,0 +1,183 @@ +"""In-memory on-chain substrate for the agentic demo. + +The real :class:`src.payment_protocol.PaymentClient` and +:class:`switchboard.x402_middleware.X402Middleware` talk to a node via web3. +For a runnable, node-free demo we provide a ``MockChain`` ledger plus a +``MockPaymentClient`` that implements exactly the surface those components call: + + - ``wallet_address`` + - ``sign_and_send(tx)`` (direct value transfer — x402 EXACT scheme) + - ``wait_for_confirmations(tx)`` + - ``create_payment(payee, amount_wei, ...)`` (escrow lock — ESCROW scheme) + - ``confirm_payment(request_id)`` (escrow release) + - ``get_payment_state(request_id)`` + +It models a minimal **AgentEscrow** (lock -> confirm -> release / refund) on top +of a balance ledger, so the demo exercises the genuine +``402 offer -> pay -> settle`` state machine, just against memory instead of a +testnet. Swapping in a real ``PaymentClient`` against an RPC needs no code change +upstream — the scenario only depends on this surface. +""" + +from __future__ import annotations + +import hashlib +import time +import uuid +from dataclasses import dataclass, field +from enum import Enum + + +class EscrowState(Enum): + LOCKED = "Locked" + CONFIRMED = "Confirmed" + RELEASED = "Released" + REFUNDED = "Refunded" + CANCELLED = "Cancelled" + + +@dataclass +class _Escrow: + request_id: str + payer: str + payee: str + amount_wei: int + created_block: int + timeout_blocks: int + challenge_period: int + state: EscrowState = EscrowState.LOCKED + + +@dataclass +class _PaymentRequest: + """Mirror of src.payment_protocol.PaymentRequest's load-bearing fields.""" + + request_id: str + payer: str + payee: str + amount_wei: int + status: str = "locked" + + +class MockChain: + """A toy ledger: ETH/native balances per address + an escrow vault.""" + + def __init__(self) -> None: + self.balances: dict[str, int] = {} + self.escrows: dict[str, _Escrow] = {} + self.block_number: int = 0 + self.tx_log: list[dict] = [] + + def fund(self, address: str, amount_wei: int) -> None: + self.balances[address] = self.balances.get(address, 0) + amount_wei + + def balance_of(self, address: str) -> int: + return self.balances.get(address, 0) + + def mine(self, blocks: int = 1) -> None: + self.block_number += blocks + + # — direct transfer (x402 EXACT) — + def transfer(self, frm: str, to: str, amount_wei: int) -> str: + if self.balances.get(frm, 0) < amount_wei: + raise RuntimeError(f"insufficient balance: {frm} has {self.balances.get(frm,0)} < {amount_wei}") + self.balances[frm] -= amount_wei + self.balances[to] = self.balances.get(to, 0) + amount_wei + self.mine() + tx_hash = "0x" + hashlib.sha256(f"{frm}{to}{amount_wei}{uuid.uuid4()}".encode()).hexdigest() + self.tx_log.append({"type": "transfer", "from": frm, "to": to, "amount": amount_wei, "tx": tx_hash}) + return tx_hash + + # — escrow lifecycle (AgentEscrow) — + def escrow_lock(self, payer: str, payee: str, amount_wei: int, request_id: str, + timeout_blocks: int, challenge_period: int) -> str: + if self.balances.get(payer, 0) < amount_wei: + raise RuntimeError(f"insufficient balance to lock: {payer}") + self.balances[payer] -= amount_wei + esc = _Escrow( + request_id=request_id, payer=payer, payee=payee, amount_wei=amount_wei, + created_block=self.block_number, timeout_blocks=timeout_blocks, + challenge_period=challenge_period, + ) + self.escrows[request_id] = esc + self.mine() + tx_hash = "0x" + hashlib.sha256(f"lock{request_id}".encode()).hexdigest() + self.tx_log.append({"type": "escrow_lock", "request_id": request_id, "amount": amount_wei, "tx": tx_hash}) + return tx_hash + + def escrow_confirm(self, request_id: str) -> str: + esc = self.escrows[request_id] + if esc.state is not EscrowState.LOCKED: + raise RuntimeError(f"escrow {request_id} not in Locked state: {esc.state}") + esc.state = EscrowState.RELEASED + self.balances[esc.payee] = self.balances.get(esc.payee, 0) + esc.amount_wei + self.mine() + tx_hash = "0x" + hashlib.sha256(f"release{request_id}".encode()).hexdigest() + self.tx_log.append({"type": "escrow_release", "request_id": request_id, + "payee": esc.payee, "amount": esc.amount_wei, "tx": tx_hash}) + return tx_hash + + def escrow_refund(self, request_id: str) -> str: + esc = self.escrows[request_id] + unlock_block = esc.created_block + esc.timeout_blocks + esc.challenge_period + if self.block_number < unlock_block: + raise RuntimeError( + f"challenge period not over: available at block {unlock_block}, current {self.block_number}" + ) + esc.state = EscrowState.REFUNDED + self.balances[esc.payer] = self.balances.get(esc.payer, 0) + esc.amount_wei + self.mine() + return "0x" + hashlib.sha256(f"refund{request_id}".encode()).hexdigest() + + def escrow_state(self, request_id: str) -> EscrowState: + return self.escrows[request_id].state + + +class MockPaymentClient: + """Implements the PaymentClient surface used by X402Middleware + the scenario, + backed by a :class:`MockChain`. No node required.""" + + def __init__(self, chain: MockChain, wallet_address: str): + self.chain = chain + self.wallet_address = wallet_address + self.pending_payments: dict[str, _PaymentRequest] = {} + + # — used by X402Middleware EXACT path — + def sign_and_send(self, tx: dict) -> str: + return self.chain.transfer(self.wallet_address, tx["to"], int(tx["value"])) + + def wait_for_confirmations(self, tx_hash: str, confirmations: int | None = None) -> dict: + return {"status": 1, "transactionHash": tx_hash} + + # — used by X402Middleware ESCROW path + scenario — + def create_payment(self, payee: str, amount_wei: int, timeout_blocks: int = 50, + challenge_period_blocks: int = 10, request_id: str | None = None, + description: str = "", metadata: dict | None = None) -> _PaymentRequest: + request_id = request_id or str(uuid.uuid4()) + self.chain.escrow_lock( + payer=self.wallet_address, payee=payee, amount_wei=amount_wei, + request_id=request_id, timeout_blocks=timeout_blocks, + challenge_period=challenge_period_blocks, + ) + req = _PaymentRequest(request_id=request_id, payer=self.wallet_address, + payee=payee, amount_wei=amount_wei, status="locked") + self.pending_payments[request_id] = req + return req + + def confirm_payment(self, request_id: str) -> bool: + self.chain.escrow_confirm(request_id) + if request_id in self.pending_payments: + self.pending_payments[request_id].status = "confirmed" + return True + + def request_refund(self, request_id: str) -> bool: + self.chain.escrow_refund(request_id) + if request_id in self.pending_payments: + self.pending_payments[request_id].status = "refunded" + return True + + def get_payment_state(self, request_id: str) -> str: + return self.chain.escrow_state(request_id).value + + def get_balance(self) -> int: + return self.chain.balance_of(self.wallet_address) diff --git a/examples/agentic_demo/run.py b/examples/agentic_demo/run.py new file mode 100644 index 0000000..752a25d --- /dev/null +++ b/examples/agentic_demo/run.py @@ -0,0 +1,90 @@ +#!/usr/bin/env python3 +"""Runnable agentic-payments demo. + + PYTHONPATH=. python examples/agentic_demo/run.py + PYTHONPATH=. python examples/agentic_demo/run.py --swap-to LUX --json + +Agent A pays Agent B for an inference job through the x402 middleware + on-chain +escrow, settles on delivery, then Agent B routes the received USDC through the +SafeSwap orchestrator into a target asset. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +# Allow ``python examples/agentic_demo/run.py`` from the repo root without +# requiring PYTHONPATH=. to be set, while still preferring an installed package. +_REPO_ROOT = Path(__file__).resolve().parents[2] +if str(_REPO_ROOT) not in sys.path: + sys.path.insert(0, str(_REPO_ROOT)) + +try: + from examples.agentic_demo.scenario import USDC, run_scenario +except ModuleNotFoundError: # pragma: no cover - fallback when run as a script + from scenario import USDC, run_scenario # type: ignore + + +BAR = "─" * 64 + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Agentic A2A payment + SafeSwap demo") + parser.add_argument("--swap-to", default="ETH", choices=["ETH", "LUX", "USDC"], + help="asset Agent B rebalances into via SafeSwap") + parser.add_argument("--price", type=float, default=5.0, help="job price in USDC") + parser.add_argument("--json", action="store_true", help="emit machine-readable JSON") + args = parser.parse_args(argv) + + result = run_scenario( + price_units=int(args.price * USDC), + swap_to=args.swap_to, + verbose=not args.json, + ) + + if args.json: + out = { + "settled": result.settled, + "swap_routed": result.swap_routed, + "escrow": { + "request_id": result.escrow_request_id, + "state": result.escrow_state_after_settle, + }, + "offer": { + "amount_units": result.offer.amount_wei, + "currency": result.offer.currency, + "scheme": result.offer.scheme.value, + "recipient": result.offer.recipient, + }, + "swap": result.swap_receipt.to_dict(), + "spend_summary": result.spend_summary, + } + print(json.dumps(out, indent=2)) + return 0 if (result.settled and result.swap_routed) else 1 + + print() + print(BAR) + print(" AGENTIC PAYMENTS DEMO — A2A pay + escrow settle + SafeSwap route") + print(BAR) + for s in result.steps: + print(f" {s.step:<13} │ {s.detail}") + print(BAR) + r = result.swap_receipt + print(f" RESULT") + print(f" 402 offer -> pay -> settle : {'OK' if result.settled else 'FAILED'} " + f"(escrow {result.escrow_state_after_settle})") + print(f" agentic swap routed : {'OK' if result.swap_routed else 'FAILED'} " + f"({r.amount_in} USDC units -> {r.amount_out} {r.token_out} units via {' -> '.join(r.route)})") + print(f" total spent : {result.spend_summary['total_spent_wei'] / USDC} USDC") + print(BAR) + print() + + ok = result.settled and result.swap_routed + return 0 if ok else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/examples/agentic_demo/safeswap.py b/examples/agentic_demo/safeswap.py new file mode 100644 index 0000000..2dc3afe --- /dev/null +++ b/examples/agentic_demo/safeswap.py @@ -0,0 +1,253 @@ +"""SafeSwap orchestrator client (mockable). + +In the demo, after Agent B is paid for its work, it routes the received token +through **SafeSwap** — an external best-execution swap orchestrator — to rebalance +into a target asset (e.g. swap the inbound USDC into ETH for gas, or into a yield +asset). + +This module exposes a tiny client against SafeSwap's orchestrator HTTP API plus an +in-process ``MockSafeSwapOrchestrator`` so the whole flow is runnable and testable +with **no network**. The contract between the two is the ``SafeSwapClient`` surface: + + quote = client.quote(SwapRequest(...)) -> SwapQuote + receipt = client.execute(quote) -> SwapReceipt + +A real deployment would point ``SafeSwapClient(base_url=...)`` at the live +orchestrator; the demo and tests inject ``transport=MockSafeSwapOrchestrator()``. +""" + +from __future__ import annotations + +import hashlib +import json +import time +import uuid +from dataclasses import dataclass, field +from decimal import Decimal +from typing import Protocol + + +# ─── Wire types ────────────────────────────────────────────────────────────── + + +@dataclass +class SwapRequest: + """A request to route ``amount_in`` of ``token_in`` into ``token_out``.""" + + token_in: str + token_out: str + amount_in: int # base units of token_in (e.g. wei / 6-dp USDC units) + chain_id: int = 8453 + recipient: str = "" + slippage_bps: int = 50 # 0.50% default max slippage + deadline_s: int = 120 + + def to_dict(self) -> dict: + return { + "tokenIn": self.token_in, + "tokenOut": self.token_out, + "amountIn": str(self.amount_in), + "chainId": self.chain_id, + "recipient": self.recipient, + "slippageBps": self.slippage_bps, + "deadlineS": self.deadline_s, + } + + +@dataclass +class SwapQuote: + """A priced route returned by the orchestrator. Must be ``execute``-d to settle.""" + + quote_id: str + token_in: str + token_out: str + amount_in: int + amount_out: int # quoted output in base units of token_out + route: list[str] # venue path, e.g. ["UniswapV3", "Curve"] + price: str # human-readable token_out per token_in + fee_bps: int + expires_at: int + + def is_expired(self, now: float | None = None) -> bool: + return (now or time.time()) > self.expires_at + + +@dataclass +class SwapReceipt: + """Proof a routed swap settled.""" + + quote_id: str + tx_hash: str + token_in: str + token_out: str + amount_in: int + amount_out: int + route: list[str] + settled_at: float = field(default_factory=time.time) + + def to_dict(self) -> dict: + return { + "quoteId": self.quote_id, + "txHash": self.tx_hash, + "tokenIn": self.token_in, + "tokenOut": self.token_out, + "amountIn": str(self.amount_in), + "amountOut": str(self.amount_out), + "route": self.route, + "settledAt": int(self.settled_at), + } + + +class SafeSwapError(RuntimeError): + """Raised when a quote or execution fails.""" + + +# ─── Transport protocol ────────────────────────────────────────────────────── + + +class SafeSwapTransport(Protocol): + """Minimal transport SafeSwapClient drives. The real impl is HTTP; the test + impl is :class:`MockSafeSwapOrchestrator`.""" + + def post(self, path: str, body: dict) -> dict: ... + + +# ─── Mock orchestrator (in-process, no network) ────────────────────────────── + + +class MockSafeSwapOrchestrator: + """In-process stand-in for SafeSwap's orchestrator API. + + Deterministic pricing so tests can assert exact outputs: + ``amount_out = amount_in * rate * (1 - fee)`` with a fixed per-pair rate table. + Tracks calls so tests can assert the swap was actually routed. + """ + + # token_out units per 1 unit token_in (toy but deterministic) + RATES: dict[tuple[str, str], Decimal] = { + ("USDC", "ETH"): Decimal("0.0004"), # 1 USDC -> 0.0004 ETH (ETH ~ $2500) + ("USDC", "LUX"): Decimal("2.0"), # 1 USDC -> 2 LUX + ("ETH", "USDC"): Decimal("2500"), + ("USDC", "USDC"): Decimal("1"), + } + FEE_BPS = 30 # 0.30% orchestrator fee + + def __init__(self) -> None: + self.quotes: dict[str, SwapQuote] = {} + self.calls: list[tuple[str, dict]] = [] + + def post(self, path: str, body: dict) -> dict: + self.calls.append((path, body)) + if path == "/v1/quote": + return self._quote(body) + if path == "/v1/execute": + return self._execute(body) + raise SafeSwapError(f"unknown SafeSwap path: {path}") + + def _quote(self, body: dict) -> dict: + token_in = body["tokenIn"] + token_out = body["tokenOut"] + amount_in = int(body["amountIn"]) + rate = self.RATES.get((token_in, token_out)) + if rate is None: + raise SafeSwapError(f"no SafeSwap route for {token_in}->{token_out}") + + gross = Decimal(amount_in) * rate + fee = gross * Decimal(self.FEE_BPS) / Decimal(10_000) + amount_out = int(gross - fee) + quote_id = "ssq_" + hashlib.sha256( + json.dumps(body, sort_keys=True).encode() + uuid.uuid4().bytes + ).hexdigest()[:16] + + route = ["SafeSwap.Router", "UniswapV3"] if token_out != "LUX" else ["SafeSwap.Router", "LuxDEX"] + quote = SwapQuote( + quote_id=quote_id, + token_in=token_in, + token_out=token_out, + amount_in=amount_in, + amount_out=amount_out, + route=route, + price=str(rate), + fee_bps=self.FEE_BPS, + expires_at=int(time.time()) + 60, + ) + self.quotes[quote_id] = quote + return { + "quoteId": quote.quote_id, + "tokenIn": quote.token_in, + "tokenOut": quote.token_out, + "amountIn": str(quote.amount_in), + "amountOut": str(quote.amount_out), + "route": quote.route, + "price": quote.price, + "feeBps": quote.fee_bps, + "expiresAt": quote.expires_at, + } + + def _execute(self, body: dict) -> dict: + quote_id = body.get("quoteId", "") + quote = self.quotes.get(quote_id) + if quote is None: + raise SafeSwapError(f"unknown or expired quoteId: {quote_id}") + if quote.is_expired(): + raise SafeSwapError(f"quote {quote_id} expired") + tx_hash = "0x" + hashlib.sha256(("exec" + quote_id).encode()).hexdigest() + return { + "quoteId": quote_id, + "txHash": tx_hash, + "tokenIn": quote.token_in, + "tokenOut": quote.token_out, + "amountIn": str(quote.amount_in), + "amountOut": str(quote.amount_out), + "route": quote.route, + "settledAt": int(time.time()), + } + + +# ─── Client ────────────────────────────────────────────────────────────────── + + +class SafeSwapClient: + """Calls SafeSwap's orchestrator. ``transport`` defaults to a mock so the demo + runs offline; pass ``base_url`` + a real HTTP transport for live routing.""" + + def __init__( + self, + transport: SafeSwapTransport | None = None, + base_url: str = "https://orchestrator.safeswap.example", + ) -> None: + self.transport: SafeSwapTransport = transport or MockSafeSwapOrchestrator() + self.base_url = base_url + + def quote(self, req: SwapRequest) -> SwapQuote: + data = self.transport.post("/v1/quote", req.to_dict()) + return SwapQuote( + quote_id=data["quoteId"], + token_in=data["tokenIn"], + token_out=data["tokenOut"], + amount_in=int(data["amountIn"]), + amount_out=int(data["amountOut"]), + route=list(data["route"]), + price=data["price"], + fee_bps=int(data["feeBps"]), + expires_at=int(data["expiresAt"]), + ) + + def execute(self, quote: SwapQuote) -> SwapReceipt: + if quote.is_expired(): + raise SafeSwapError(f"quote {quote.quote_id} expired before execute") + data = self.transport.post("/v1/execute", {"quoteId": quote.quote_id}) + return SwapReceipt( + quote_id=data["quoteId"], + tx_hash=data["txHash"], + token_in=data["tokenIn"], + token_out=data["tokenOut"], + amount_in=int(data["amountIn"]), + amount_out=int(data["amountOut"]), + route=list(data["route"]), + settled_at=float(data["settledAt"]), + ) + + def route(self, req: SwapRequest) -> SwapReceipt: + """Convenience: quote then execute in one call (best-execution route).""" + return self.execute(self.quote(req)) diff --git a/examples/agentic_demo/scenario.py b/examples/agentic_demo/scenario.py new file mode 100644 index 0000000..8d3d656 --- /dev/null +++ b/examples/agentic_demo/scenario.py @@ -0,0 +1,234 @@ +"""Agentic payments scenario: Agent A pays Agent B, then B swaps via SafeSwap. + +Flow exercised end-to-end (offline, node-free): + + 1. Agent A asks Agent B's paid endpoint for work (an inference job). + 2. Agent B replies ``402 Payment Required`` with an x402 ``PaymentOffer`` + (escrow scheme — funds locked, released on delivery). + 3. Agent A's :class:`X402Middleware` validates the offer against policy + (cap, allowlist, gas budget), pays into escrow, and retries with proof. + 4. Agent B delivers the work; Agent A *settles* the escrow (confirm -> release). + 5. **Agentic swap**: now-funded Agent B routes the received USDC through the + **SafeSwap** orchestrator into a target asset (ETH for gas), getting a + best-execution route + receipt. + +The scenario depends only on the real ``switchboard`` package surface plus the +mock substrate in this package, so the same code runs against a live RPC + +SafeSwap by swapping in a real ``PaymentClient`` and ``SafeSwapClient(base_url=...)``. +""" + +from __future__ import annotations + +import time +import uuid +from dataclasses import dataclass, field + +from switchboard.gas_tracker import GasTracker +from switchboard.x402_middleware import ( + PaymentOffer, + PaymentProof, + PaymentScheme, + X402Middleware, +) + +from .onchain import EscrowState, MockChain, MockPaymentClient +from .safeswap import ( + MockSafeSwapOrchestrator, + SafeSwapClient, + SwapReceipt, + SwapRequest, +) + + +# token decimals (USDC 6dp on the wire; we keep ETH in wei = 18dp) +USDC = 10**6 +ETH = 10**18 + + +class BudgetGuard: + """Adapts the real :class:`switchboard.gas_tracker.GasTracker` to the + duck-typed contract :class:`X402Middleware` expects from a ``gas_tracker``: + ``can_send_transaction(wallet, amount)`` and ``record_gas_usage(wallet, amount)``. + + The stdlib ``GasTracker`` is a process singleton with single-arg methods; this + wrapper drops the per-wallet arg the middleware passes (the tracker enforces a + global budget) and resets state on construction so repeated demo/test runs are + independent. + """ + + def __init__(self, hourly_limit: int, daily_limit: int): + self._tracker = GasTracker(hourly_limit=hourly_limit, daily_limit=daily_limit) + # GasTracker is a singleton; clear any leaked state then apply our limits. + self._tracker.reset_all() + self._tracker.set_limits(hourly_limit=hourly_limit, daily_limit=daily_limit) + + def can_send_transaction(self, wallet: str, amount: int) -> bool: + return self._tracker.can_send_transaction(amount) + + def record_gas_usage(self, wallet: str, amount: int) -> None: + self._tracker.record_gas_usage(amount) + + +@dataclass +class StepLog: + """One ledger-visible event in the flow, for printing + assertions.""" + + step: str + detail: str + data: dict = field(default_factory=dict) + + +@dataclass +class ScenarioResult: + offer: PaymentOffer + proof: PaymentProof + escrow_request_id: str + escrow_state_after_settle: str + swap_receipt: SwapReceipt + agent_b_balance_token_out: int + steps: list[StepLog] + spend_summary: dict + + @property + def settled(self) -> bool: + return self.escrow_state_after_settle == EscrowState.RELEASED.value + + @property + def swap_routed(self) -> bool: + return bool(self.swap_receipt and self.swap_receipt.route and self.swap_receipt.amount_out > 0) + + +class AgentBEndpoint: + """Agent B's paid 'inference' endpoint, x402-gated with the ESCROW scheme. + + Returns a 402 ``PaymentOffer`` on a cold call, then serves the work once a + valid ``X-Payment-Proof`` for the escrow request is presented. + """ + + def __init__(self, recipient: str, price_units: int = 5 * USDC, chain_id: int = 8453): + self.recipient = recipient + self.price_units = price_units + self.chain_id = chain_id + self.served: list[str] = [] + + def offer(self, endpoint: str) -> PaymentOffer: + return PaymentOffer( + amount_wei=self.price_units, + currency="USDC", + recipient=self.recipient, + chain_id=self.chain_id, + scheme=PaymentScheme.ESCROW, + description="agent-B inference job", + endpoint=endpoint, + nonce=uuid.uuid4().hex[:16], + expires_at=int(time.time()) + 300, + ) + + def deliver(self, proof: PaymentProof) -> dict: + """Verify the proof carries an escrow ref + serve the deliverable.""" + if not proof.tx_hash: + raise ValueError("missing payment proof") + self.served.append(proof.tx_hash) + return { + "status": 200, + "result": {"embedding_dim": 1536, "tokens": 4096, "job": proof.tx_hash}, + } + + +def run_scenario( + *, + chain: MockChain | None = None, + safeswap: SafeSwapClient | None = None, + agent_a_addr: str = "0xA0A0a0a0a0A0a0A0a0A0a0a0a0A0a0a0A0A0A0a0", + agent_b_addr: str = "0xB0b0B0b0b0b0b0B0b0b0b0b0B0b0b0B0b0B0B0b0", + price_units: int = 5 * USDC, + swap_to: str = "ETH", + verbose: bool = False, +) -> ScenarioResult: + """Run the full A2A pay -> settle -> swap flow and return a structured result.""" + + steps: list[StepLog] = [] + + def record(step: str, detail: str, **data) -> None: + steps.append(StepLog(step, detail, data)) + if verbose: + print(f"[{step}] {detail}") + + chain = chain or MockChain() + safeswap = safeswap or SafeSwapClient(transport=MockSafeSwapOrchestrator()) + + # Fund Agent A with USDC to pay for work. + chain.fund(agent_a_addr, 100 * USDC) + record("setup", f"Agent A funded with {100} USDC; Agent B starts empty", + agent_a_balance=chain.balance_of(agent_a_addr)) + + # — Agent A's payment stack — + payment_client = MockPaymentClient(chain, wallet_address=agent_a_addr) + gas_tracker = BudgetGuard(hourly_limit=50 * USDC, daily_limit=200 * USDC) + middleware = X402Middleware( + payment_client=payment_client, + gas_tracker=gas_tracker, + max_payment_wei=20 * USDC, + allowed_recipients={agent_b_addr}, + ) + + # — Agent B's paid endpoint — + agent_b = AgentBEndpoint(recipient=agent_b_addr, price_units=price_units, chain_id=8453) + endpoint = "https://agent-b.example/v1/inference" + + # 1. Cold call -> 402 offer + offer = agent_b.offer(endpoint) + record("402", f"Agent B -> 402 Payment Required: {price_units / USDC} USDC (escrow)", + recipient=offer.recipient, amount=offer.amount_wei, scheme=offer.scheme.value) + + # 2. Agent A validates + pays into escrow (reuses real middleware logic) + middleware._validate_offer(offer) + record("validate", "Agent A: offer passes policy (cap / allowlist / gas budget)") + + proof = middleware._pay_onchain(offer) # ESCROW path -> create_payment -> lock + escrow_request_id = proof.tx_hash + gas_tracker.record_gas_usage(agent_a_addr, offer.amount_wei) + middleware.total_spent_wei += offer.amount_wei + record("pay", f"Agent A locked {price_units / USDC} USDC in escrow", + request_id=escrow_request_id, + escrow_state=payment_client.get_payment_state(escrow_request_id)) + + # 3. Agent B delivers the work against the proof + delivery = agent_b.deliver(proof) + record("deliver", f"Agent B delivered work (HTTP {delivery['status']})", result=delivery["result"]) + + # 4. Agent A settles: confirm -> release escrow to Agent B + payment_client.confirm_payment(escrow_request_id) + state_after = payment_client.get_payment_state(escrow_request_id) + record("settle", f"Agent A confirmed -> escrow {state_after}; Agent B paid", + escrow_state=state_after, agent_b_balance=chain.balance_of(agent_b_addr)) + + # record the completed payment for the spend summary + from switchboard.x402_middleware import PaymentRecord + middleware.payment_history.append( + PaymentRecord(endpoint=endpoint, offer=offer, proof=proof, response_status=delivery["status"]) + ) + + # 5. AGENTIC SWAP — Agent B routes received USDC through SafeSwap into swap_to + received = chain.balance_of(agent_b_addr) + swap_req = SwapRequest( + token_in="USDC", token_out=swap_to, amount_in=received, + chain_id=offer.chain_id, recipient=agent_b_addr, + ) + quote = safeswap.quote(swap_req) + record("swap.quote", f"SafeSwap quote: {received / USDC} USDC -> {quote.amount_out} {swap_to} units " + f"via {' -> '.join(quote.route)}", route=quote.route, amount_out=quote.amount_out) + receipt = safeswap.execute(quote) + record("swap.execute", f"SafeSwap routed swap settled (tx {receipt.tx_hash[:12]}...)", + tx=receipt.tx_hash, amount_out=receipt.amount_out, route=receipt.route) + + return ScenarioResult( + offer=offer, + proof=proof, + escrow_request_id=escrow_request_id, + escrow_state_after_settle=state_after, + swap_receipt=receipt, + agent_b_balance_token_out=receipt.amount_out, + steps=steps, + spend_summary=middleware.get_spend_summary(), + ) diff --git a/tests/test_agentic_demo.py b/tests/test_agentic_demo.py new file mode 100644 index 0000000..d7064bc --- /dev/null +++ b/tests/test_agentic_demo.py @@ -0,0 +1,169 @@ +"""Tests for the agentic-payments demo (examples/agentic_demo). + +Asserts the full flow end-to-end: + 402 offer -> pay -> settle (escrow Released) and the SafeSwap swap routes. + +The demo is node/RPC-free: it drives the real ``switchboard`` x402 middleware + +gas budget against an in-memory chain, and the real ``SafeSwapClient`` against an +in-process mock orchestrator. +""" + +from __future__ import annotations + +import pytest + +from examples.agentic_demo import ( + EscrowState, + MockChain, + MockSafeSwapOrchestrator, + SafeSwapClient, + SafeSwapError, + SwapRequest, + run_scenario, +) +from examples.agentic_demo.safeswap import SwapQuote +from examples.agentic_demo.scenario import USDC, AgentBEndpoint + + +# ─── full flow ─────────────────────────────────────────────────────────────── + + +def test_full_flow_offer_pay_settle_and_swap_routes(): + chain = MockChain() + transport = MockSafeSwapOrchestrator() + result = run_scenario(chain=chain, safeswap=SafeSwapClient(transport=transport)) + + # 402 offer was made with the escrow scheme + correct price + assert result.offer.scheme.value == "escrow" + assert result.offer.amount_wei == 5 * USDC + assert result.offer.currency == "USDC" + + # pay -> settle: escrow ended Released, not just Locked + assert result.settled is True + assert result.escrow_state_after_settle == EscrowState.RELEASED.value + assert chain.escrow_state(result.escrow_request_id) is EscrowState.RELEASED + + # the swap actually routed through SafeSwap + assert result.swap_routed is True + assert result.swap_receipt.token_in == "USDC" + assert result.swap_receipt.token_out == "ETH" + assert result.swap_receipt.amount_out > 0 + assert result.swap_receipt.route # non-empty venue path + assert result.swap_receipt.tx_hash.startswith("0x") + + # SafeSwap orchestrator was genuinely called: quote then execute + paths = [p for p, _ in transport.calls] + assert paths == ["/v1/quote", "/v1/execute"] + + +def test_step_order_is_offer_pay_deliver_settle_swap(): + result = run_scenario() + steps = [s.step for s in result.steps] + # the load-bearing ordering: 402 before pay, settle before swap + assert steps.index("402") < steps.index("pay") < steps.index("settle") + assert steps.index("settle") < steps.index("swap.quote") < steps.index("swap.execute") + + +def test_funds_move_payer_to_payee_then_swap_out(): + chain = MockChain() + a = "0xAAaA" + b = "0xBBbB" + result = run_scenario(chain=chain, agent_a_addr=a, agent_b_addr=b, price_units=5 * USDC) + + # Agent A spent 5 USDC (100 funded - 5), escrow released the 5 to Agent B + assert chain.balance_of(a) == 95 * USDC + assert chain.balance_of(b) == 5 * USDC + # spend summary reflects exactly one settled payment of 5 USDC + assert result.spend_summary["total_payments"] == 1 + assert result.spend_summary["total_spent_wei"] == 5 * USDC + + +def test_swap_to_lux_uses_lux_route(): + result = run_scenario(swap_to="LUX") + assert result.swap_receipt.token_out == "LUX" + assert "LuxDEX" in result.swap_receipt.route + + +# ─── escrow state machine ──────────────────────────────────────────────────── + + +def test_escrow_starts_locked_before_settle(): + chain = MockChain() + # run only up to the lock by inspecting an isolated escrow via the client + from examples.agentic_demo.onchain import MockPaymentClient + + chain.fund("0xA", 10 * USDC) + client = MockPaymentClient(chain, "0xA") + req = client.create_payment("0xB", 5 * USDC) + assert client.get_payment_state(req.request_id) == EscrowState.LOCKED.value + assert chain.balance_of("0xB") == 0 # not yet released + client.confirm_payment(req.request_id) + assert client.get_payment_state(req.request_id) == EscrowState.RELEASED.value + assert chain.balance_of("0xB") == 5 * USDC + + +def test_escrow_refund_blocked_until_challenge_window(): + from examples.agentic_demo.onchain import MockPaymentClient + + chain = MockChain() + chain.fund("0xA", 10 * USDC) + client = MockPaymentClient(chain, "0xA") + req = client.create_payment("0xB", 5 * USDC, timeout_blocks=5, challenge_period_blocks=3) + with pytest.raises(RuntimeError, match="challenge period not over"): + client.request_refund(req.request_id) + chain.mine(20) + assert client.request_refund(req.request_id) is True + assert client.get_payment_state(req.request_id) == EscrowState.REFUNDED.value + assert chain.balance_of("0xA") == 10 * USDC # fully refunded + + +# ─── SafeSwap orchestrator ─────────────────────────────────────────────────── + + +def test_safeswap_quote_then_execute_roundtrip(): + client = SafeSwapClient(transport=MockSafeSwapOrchestrator()) + quote = client.quote(SwapRequest(token_in="USDC", token_out="ETH", amount_in=5 * USDC)) + assert isinstance(quote, SwapQuote) + assert quote.amount_out > 0 + receipt = client.execute(quote) + assert receipt.amount_out == quote.amount_out + assert receipt.tx_hash.startswith("0x") + + +def test_safeswap_fee_is_applied(): + # 5 USDC -> ETH at 0.0004 with 0.30% fee: 5e6 * 0.0004 = 2000, minus 0.3% = 1994 + client = SafeSwapClient(transport=MockSafeSwapOrchestrator()) + quote = client.quote(SwapRequest(token_in="USDC", token_out="ETH", amount_in=5 * USDC)) + assert quote.amount_out == 1994 + assert quote.fee_bps == 30 + + +def test_safeswap_unknown_pair_raises(): + client = SafeSwapClient(transport=MockSafeSwapOrchestrator()) + with pytest.raises(SafeSwapError, match="no SafeSwap route"): + client.quote(SwapRequest(token_in="DOGE", token_out="ETH", amount_in=100)) + + +def test_safeswap_route_convenience_quotes_and_executes(): + client = SafeSwapClient(transport=MockSafeSwapOrchestrator()) + receipt = client.route(SwapRequest(token_in="USDC", token_out="LUX", amount_in=3 * USDC)) + assert receipt.token_out == "LUX" + assert receipt.amount_out > 0 + + +# ─── x402 endpoint shape ───────────────────────────────────────────────────── + + +def test_agent_b_endpoint_offer_and_delivery(): + ep = AgentBEndpoint(recipient="0xB", price_units=2 * USDC) + offer = ep.offer("https://x/y") + assert offer.recipient == "0xB" + assert offer.amount_wei == 2 * USDC + assert offer.endpoint == "https://x/y" + + from switchboard.x402_middleware import PaymentProof + + proof = PaymentProof(tx_hash="req-1", chain_id=8453, payer="0xA", amount_wei=2 * USDC) + out = ep.deliver(proof) + assert out["status"] == 200 + assert "req-1" in ep.served diff --git a/tests/test_pwa.py b/tests/test_pwa.py new file mode 100644 index 0000000..16a18bf --- /dev/null +++ b/tests/test_pwa.py @@ -0,0 +1,156 @@ +"""Tests for the Switchboard Lab PWA (web/manifest.json + web/sw.js). + +Validates that the lab is installable + offline-capable: +- the manifest is valid JSON with the fields browsers require to offer install +- the icons it references exist on disk +- the service worker exists, is valid JS (node --check), precaches an app shell, + and the shell entries it lists actually exist +- service-worker registration is wired into the pages +""" + +from __future__ import annotations + +import json +import shutil +import subprocess +import tempfile +from pathlib import Path + +import pytest + +HERE = Path(__file__).resolve().parent +WEB = HERE.parent / "web" +MANIFEST = WEB / "manifest.json" +SW = WEB / "sw.js" + + +def runnable_node() -> str | None: + node = shutil.which("node") + if node is None: + return None + try: + subprocess.run([node, "--version"], capture_output=True, check=False, timeout=5) + except OSError: + return None + return node + + +# ─── manifest ──────────────────────────────────────────────────────────────── + + +@pytest.fixture(scope="module") +def manifest() -> dict: + assert MANIFEST.is_file(), f"missing {MANIFEST}" + return json.loads(MANIFEST.read_text()) + + +def test_manifest_has_install_fields(manifest: dict) -> None: + for field in ("name", "short_name", "start_url", "display", "icons", + "background_color", "theme_color"): + assert field in manifest, f"manifest missing required field: {field}" + assert manifest["display"] in ("standalone", "fullscreen", "minimal-ui") + + +def test_manifest_icons_exist_and_cover_purposes(manifest: dict) -> None: + icons = manifest["icons"] + assert icons, "manifest declares no icons" + purposes = set() + for icon in icons: + src = icon["src"] + # resolve relative to the manifest location (scope "./" == web/) + path = (WEB / src.lstrip("./")).resolve() + assert path.is_file(), f"icon file missing: {src} -> {path}" + purposes.update(icon.get("purpose", "any").split()) + assert "any" in purposes, "need at least one 'any' purpose icon" + assert "maskable" in purposes, "need a maskable icon for a polished install" + + +def test_manifest_start_url_exists(manifest: dict) -> None: + start = manifest["start_url"].lstrip("./") + assert (WEB / start).is_file(), f"start_url target missing: {manifest['start_url']}" + + +def test_manifest_shortcuts_resolve(manifest: dict) -> None: + for sc in manifest.get("shortcuts", []): + target = sc["url"].lstrip("./") + assert (WEB / target).is_file(), f"shortcut url target missing: {sc['url']}" + + +def test_theme_color_matches_manifest(manifest: dict) -> None: + # The root page advertises the same theme-color so install chrome matches. + home = (WEB / "index.html").read_text() + assert manifest["theme_color"] in home + + +# ─── service worker ────────────────────────────────────────────────────────── + + +def test_sw_exists_and_caches_shell() -> None: + assert SW.is_file(), f"missing {SW}" + body = SW.read_text() + # core lifecycle + strategy hooks must be present + for hook in ("install", "activate", "fetch", "caches.open", "skipWaiting", "clients.claim"): + assert hook in body, f"service worker missing: {hook}" + # navigation offline fallback is the load-bearing offline behavior + assert "navigate" in body + assert "OFFLINE_FALLBACK" in body + + +def test_sw_shell_entries_exist_on_disk() -> None: + """Every same-origin SHELL entry the SW precaches should exist (so install + doesn't silently drop the offline app shell).""" + body = SW.read_text() + import re + + m = re.search(r"const SHELL\s*=\s*\[(.*?)\];", body, re.DOTALL) + assert m, "SHELL array not found in sw.js" + entries = re.findall(r'"([^"]+)"', m.group(1)) + assert entries, "SHELL is empty" + for entry in entries: + if entry in ("./",): + continue # directory index, served by start_url + path = (WEB / entry.lstrip("./")).resolve() + assert path.is_file(), f"SHELL precache target missing: {entry} -> {path}" + + +def test_sw_registered_from_root_and_lab() -> None: + home = (WEB / "index.html").read_text() + assert "serviceWorker" in home and "register('./sw.js'" in home, "root page does not register the SW" + shared = (WEB / "lab" / "shared.js").read_text() + assert "serviceWorker" in shared and "../sw.js" in shared, "lab pages do not register the SW" + + +def test_swap_page_links_manifest() -> None: + swap = (WEB / "lab" / "swap.html").read_text() + assert 'rel="manifest"' in swap, "swap page should link the PWA manifest" + + +def _node_check(source: str, suffix: str) -> tuple[int, str]: + node = runnable_node() + if node is None: + pytest.skip("node not runnable on PATH") + with tempfile.NamedTemporaryFile("w", suffix=suffix, delete=False) as f: + f.write(source) + path = f.name + try: + proc = subprocess.run([node, "--check", path], capture_output=True, text=True, + check=False, timeout=15) + finally: + Path(path).unlink(missing_ok=True) + return proc.returncode, (proc.stderr or proc.stdout) + + +def test_sw_is_valid_js() -> None: + code, out = _node_check(SW.read_text(), ".js") + assert code == 0, out + + +def test_swap_page_script_is_valid_js() -> None: + import re + + html = (WEB / "lab" / "swap.html").read_text() + scripts = re.findall(r"", html, re.DOTALL) + assert scripts, "expected an inline + + + diff --git a/web/lab/_build.js b/web/lab/_build.js index b06f8aa..b9b68a1 100644 --- a/web/lab/_build.js +++ b/web/lab/_build.js @@ -45,6 +45,7 @@ const SIDEBAR = ` 02Escrow & Refundlive 03Streaming MPPlive 04Compute Auctionlive + 05Agentic Pay + Swapnew
Real-World Scenes
diff --git a/web/lab/index.html b/web/lab/index.html index 76cef9a..0aec7b7 100644 --- a/web/lab/index.html +++ b/web/lab/index.html @@ -427,6 +427,7 @@ 02Escrow & Refundlive 03Streaming MPPlive 04Compute Auctionlive + 05Agentic Pay + Swapnew
Real-World Scenes
@@ -514,6 +515,12 @@

AI Compute Auction

Three providers bid. Lowest valid bid wins. Escrow settles the winner. The market sets the price.

shipped
+ +
05 · Pay → Settle → Swap
+

Agentic Pay + Swap

+

Agent A pays Agent B via x402 + escrow; B routes the proceeds through SafeSwap to rebalance. The full agentic_demo, visualized.

+
new SafeSwap
+
diff --git a/web/lab/shared.js b/web/lab/shared.js index 073ad2f..6c92701 100644 --- a/web/lab/shared.js +++ b/web/lab/shared.js @@ -83,4 +83,32 @@ }, 2200); }); + /* ── PWA: manifest link + service worker registration ──────── */ + /* The lab is installable + offline-capable. Pages under web/lab/ reach the + manifest + service worker at the web/ root (one level up). Both are injected + here so every lab page participates without per-page edits. */ + (function pwa() { + // Ensure a manifest exists (some pages declare it statically). + if (!document.querySelector('link[rel="manifest"]')) { + const link = document.createElement('link'); + link.rel = 'manifest'; + link.href = '../manifest.json'; + document.head.appendChild(link); + } + if (!document.querySelector('meta[name="theme-color"]')) { + const meta = document.createElement('meta'); + meta.name = 'theme-color'; + meta.content = '#06060b'; + document.head.appendChild(meta); + } + // Register the service worker with scope at the web/ root. + if ('serviceWorker' in navigator && location.protocol !== 'file:') { + window.addEventListener('load', () => { + navigator.serviceWorker + .register('../sw.js', { scope: '../' }) + .catch(() => { /* offline support is progressive; ignore failures */ }); + }); + } + })(); + })(); diff --git a/web/lab/swap.html b/web/lab/swap.html new file mode 100644 index 0000000..2b07c43 --- /dev/null +++ b/web/lab/swap.html @@ -0,0 +1,533 @@ + + + + + +Agentic Pay + Swap — switchboard lab + + + + + + + + + + + +
+ +
+ +
+ ← Back to dashboard +
05 Protocol Pattern
+

Agentic Pay + Swap

+

+ Agent A hires Agent B for an inference job. B replies 402 Payment Required + with an escrow-scheme offer. A’s x402 middleware checks it against a cap, an allowlist and a + gas budget, then locks 5 USDC in AgentEscrow. B delivers; A confirms; + the escrow releases. Then B does the agentic part — it routes its freshly-earned USDC + through SafeSwap to rebalance into ETH for gas. One unattended loop, settled on-chain. +

+
+ ▶ Run the flow + Open Canvas Lab + View source & tests +
+
+ +
+
Live Visualization
+

The loop, step by step

+

+ This is the exact flow of examples/agentic_demo — the same seven steps the + pytest suite asserts. Watch funds move payer → escrow → payee, then route out through SafeSwap. +

+ +
+ + + + + idle +
+ +
+
+ +
+
+
+
+
Agent A · payer
+
100.00USDC
+
+
+
+
Agent B · payee
+
0.00USDC
+
+
+
+
+
+ flow ledger + block #0 +
+
+
+
+
+
+ +
+
How it maps to code
+

From scene to scenario

+

Every step above corresponds to a real call in the Python demo.

+
+
+
x402 + escrow
+

Pay & settle

+

+ X402Middleware._validate_offer() enforces cap / allowlist / gas budget, then + _pay_onchain() takes the escrow path: create_payment() locks + funds, confirm_payment() releases them on delivery. +

+
+
+
SafeSwap orchestrator
+

Route the proceeds

+

+ SafeSwapClient.quote() then .execute() hit the orchestrator API + (/v1/quote, /v1/execute). Mockable in-process, so the whole flow runs + with no node and no network. +

+
+
+ 
# PYTHONPATH=. python examples/agentic_demo/run.py --swap-to ETH
+from examples.agentic_demo import run_scenario
+result = run_scenario(swap_to="ETH")
+assert result.settled           # escrow Released
+assert result.swap_routed       # SafeSwap quote -> execute
+
+ +
+ ← PreviousCompute Auction + +
+ + +
+
+ + + + + + diff --git a/web/manifest.json b/web/manifest.json new file mode 100644 index 0000000..4e67d94 --- /dev/null +++ b/web/manifest.json @@ -0,0 +1,50 @@ +{ + "name": "Switchboard Lab — Agentic Payments", + "short_name": "Switchboard", + "description": "Interactive lab of agent-to-agent payment patterns: x402 paywalls, on-chain escrow, streaming micropayments, and agentic pay + SafeSwap routing. Installable and offline-capable.", + "id": "/lab/", + "scope": "./", + "start_url": "./lab/index.html", + "display": "standalone", + "display_override": ["standalone", "minimal-ui", "browser"], + "orientation": "any", + "background_color": "#06060b", + "theme_color": "#06060b", + "categories": ["finance", "developer", "productivity"], + "lang": "en", + "dir": "ltr", + "icons": [ + { + "src": "./icon.svg", + "sizes": "any", + "type": "image/svg+xml", + "purpose": "any" + }, + { + "src": "./icon-maskable.svg", + "sizes": "any", + "type": "image/svg+xml", + "purpose": "maskable" + } + ], + "shortcuts": [ + { + "name": "Agentic Pay + Swap", + "short_name": "Pay + Swap", + "description": "A2A pay + escrow settle + SafeSwap route, visualized", + "url": "./lab/swap.html" + }, + { + "name": "Canvas Lab", + "short_name": "Canvas", + "description": "Animated agent-payment scenes", + "url": "./agents-demo.html" + }, + { + "name": "x402 Paywall", + "short_name": "x402", + "description": "HTTP-402 payment-required flow", + "url": "./lab/x402.html" + } + ] +} diff --git a/web/sw.js b/web/sw.js new file mode 100644 index 0000000..dcf00cf --- /dev/null +++ b/web/sw.js @@ -0,0 +1,148 @@ +/* ============================================================ + SWITCHBOARD LAB — SERVICE WORKER + Makes the lab installable + offline-capable. + + Strategy: + - precache the app shell on install (lab pages, shared css/js, icons, root pages) + - navigations: network-first, fall back to cache, then to the offline shell + - same-origin static assets: stale-while-revalidate + - cross-origin (fonts etc.): pass through, opportunistically cache + Bump CACHE_VERSION to invalidate old caches on deploy. + ============================================================ */ + +const CACHE_VERSION = "switchboard-lab-v1"; +const RUNTIME = "switchboard-runtime-v1"; + +// Resolve against the SW scope so it works under any base path +// (e.g. project Pages at /switchboard/web/). +const SHELL = [ + "./", + "./manifest.json", + "./icon.svg", + "./icon-maskable.svg", + // root pages + "./index.html", + "./agents-demo.html", + "./simulator.html", + "./docs.html", + // multi-page lab + "./lab/index.html", + "./lab/shared.css", + "./lab/shared.js", + "./lab/x402.html", + "./lab/escrow.html", + "./lab/streaming.html", + "./lab/auction.html", + "./lab/swap.html", + "./lab/taxi.html", + "./lab/cafe.html", + "./lab/delivery.html", + "./lab/trip.html", + "./lab/multichain.html", + "./lab/pq.html", + "./lab/rails.html", + "./lab/tools.html", +]; + +const OFFLINE_FALLBACK = "./lab/index.html"; + +self.addEventListener("install", (event) => { + event.waitUntil( + (async () => { + const cache = await caches.open(CACHE_VERSION); + // addAll is atomic; if any single asset 404s the whole install fails, + // so add resiliently (the shell list may drift from disk over time). + await Promise.all( + SHELL.map((url) => + cache.add(new Request(url, { cache: "reload" })).catch(() => {}) + ) + ); + await self.skipWaiting(); + })() + ); +}); + +self.addEventListener("activate", (event) => { + event.waitUntil( + (async () => { + const keys = await caches.keys(); + await Promise.all( + keys + .filter((k) => k !== CACHE_VERSION && k !== RUNTIME) + .map((k) => caches.delete(k)) + ); + await self.clients.claim(); + })() + ); +}); + +// Allow the page to trigger an immediate update. +self.addEventListener("message", (event) => { + if (event.data === "SKIP_WAITING") self.skipWaiting(); +}); + +self.addEventListener("fetch", (event) => { + const req = event.request; + if (req.method !== "GET") return; + + const url = new URL(req.url); + const sameOrigin = url.origin === self.location.origin; + + // 1) Navigations -> network-first, fall back to cache, then offline shell. + if (req.mode === "navigate") { + event.respondWith( + (async () => { + try { + const fresh = await fetch(req); + const cache = await caches.open(RUNTIME); + cache.put(req, fresh.clone()); + return fresh; + } catch (e) { + const cached = await caches.match(req); + if (cached) return cached; + return ( + (await caches.match(OFFLINE_FALLBACK)) || + new Response("Offline", { status: 503, statusText: "Offline" }) + ); + } + })() + ); + return; + } + + // 2) Same-origin static assets -> stale-while-revalidate. + if (sameOrigin) { + event.respondWith( + (async () => { + const cached = await caches.match(req); + const network = fetch(req) + .then((resp) => { + if (resp && resp.ok) { + caches.open(CACHE_VERSION).then((c) => c.put(req, resp.clone())); + } + return resp; + }) + .catch(() => null); + return cached || (await network) || new Response("", { status: 504 }); + })() + ); + return; + } + + // 3) Cross-origin (Google Fonts, etc.) -> cache-first, opportunistic fill. + event.respondWith( + (async () => { + const cached = await caches.match(req); + if (cached) return cached; + try { + const resp = await fetch(req); + // Opaque responses are fine to cache for offline font rendering. + const cache = await caches.open(RUNTIME); + cache.put(req, resp.clone()); + return resp; + } catch (e) { + return cached || new Response("", { status: 504 }); + } + })() + ); +});