Merge pull request #14 from swarmproof/feat/evm-target

feat: EVMTarget — wallet-swarm lending demo + anvil fork-guard

Author: MarouaBoud

examples/evm_lending.yaml

@@ -0,0 +1,16 @@
+# Point a wallet-swarm at a lending protocol — the "open-Gauntlet" demo (FR-TA-04).
+# The mock world runs with zero infra:  stampede run --dry-run -c examples/evm_lending.yaml
+target:
+  type: evm
+  world: lending          # deterministic in-process lending pool (borrow/repay/liquidate)
+  # For a REAL run against an Anvil fork instead (refuses non-fork mainnet RPCs):
+  # rpc_url: "http://localhost:8545"   # start with:  anvil --fork-url <mainnet-rpc>
+
+population:
+  size: 50
+  mix: { naive: 0.4, expert: 0.3, adversarial: 0.3 }   # borrowers + liquidators + griefers
+  models: [dry-run:heuristic]
+
+concurrency: { curve: spike, peak: 50, hold: 10s }      # a price-shock-style burst
+goals: { autogenerate: true, mode: template }
+seed: 42

src/stampede/targets/__init__.py

@@ -64,5 +64,9 @@ def build_target(config: "object") -> TargetAdapter:  # noqa: UP037
             transport=config.transport, command=config.command, url=config.url
         )
     if config.type == "evm":
-        raise NotImplementedError("EVMTarget lands in v0.2 (FR-TA-04)")
+        # The Safety Gate (config.safety.evm_require_fork) enforces the fork rule;
+        # the target just reports whether it's a fork via its safety descriptor.
+        from stampede.targets.evm import EVMTarget
+
+        return EVMTarget(rpc_url=config.rpc_url, world=config.world)
     raise ValueError(f"unknown target type: {config.type!r}")

src/stampede/targets/evm.py

@@ -0,0 +1,202 @@
+"""EVMTarget — point a wallet-swarm at an onchain protocol (FR-TA-04, v0.2).
+
+The premium "open-Gauntlet" surface: agents hold funded test wallets and their tool
+calls become transactions. Two backends behind one adapter:
+
+* **mock lending world** (default, no chain) — a deterministic in-process lending
+  pool (borrow / repay / liquidate) so the wallet-swarm demo runs in ``--dry-run``
+  with zero infra. This is what makes the EVM story verifiable in CI.
+* **Anvil fork** (``rpc_url``) — a real fork via web3.py. Guarded by the fork check
+  below and the Safety Gate: an ``EVMTarget`` **refuses a non-fork mainnet RPC**
+  (ADR-6, NFR-SEC), so a chaos/adversarial swarm can never touch real value.
+
+Requires the ``[evm]`` extra (web3) only for the Anvil backend.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from stampede.targets.base import (
+    AgentContext,
+    HealthStatus,
+    IsolationMode,
+    SafetyDescriptor,
+    TargetAdapter,
+    ToolCall,
+    ToolResult,
+    ToolSet,
+    ToolSpec,
+)
+
+_LTV = 0.75  # loan-to-value: you can borrow up to 75% of collateral
+
+
+# ---- the mock lending world (deterministic, no chain) -----------------------
+
+
+def _lending_toolset() -> ToolSet:
+    amount_schema = {"type": "object", "properties": {"amount": {"type": "integer"}}}
+    return ToolSet(
+        tools=[
+            ToolSpec(
+                name="borrow",
+                description="Borrow against your collateral. Fails if it would leave you undercollateralized.",
+                input_schema=amount_schema,
+                destructive=True,
+            ),
+            ToolSpec(
+                name="repay",
+                description="Repay part of your debt.",
+                input_schema=amount_schema,
+            ),
+            ToolSpec(
+                name="liquidate",
+                description="Liquidate an undercollateralized account and seize its collateral.",
+                input_schema={"type": "object", "properties": {"account": {"type": "string"}}},
+                destructive=True,
+            ),
+            ToolSpec(
+                name="positions",
+                description="Read the current accounts and their health.",
+                input_schema={"type": "object", "properties": {}},
+            ),
+        ]
+    )
+
+
+def _seed_state() -> dict[str, Any]:
+    # The agent's own wallet + phantom accounts (two already underwater → liquidatable).
+    return {
+        "wallet": {"collateral": 1000, "debt": 0},
+        "accounts": {
+            "acct_0": {"collateral": 1000, "debt": 100},  # healthy
+            "acct_1": {"collateral": 1000, "debt": 900},  # underwater (debt > 750)
+            "acct_2": {"collateral": 500, "debt": 480},  # underwater
+        },
+        "liquidations": 0,
+    }
+
+
+def _healthy(pos: dict[str, int]) -> bool:
+    return pos["debt"] <= pos["collateral"] * _LTV
+
+
+def _lending_handler(tool: str, args: dict[str, Any], state: dict[str, Any]) -> ToolResult:
+    wallet = state["wallet"]
+    if tool == "positions":
+        underwater = [a for a, p in state["accounts"].items() if not _healthy(p)]
+        return ToolResult(ok=True, content=f"{len(underwater)} liquidatable", structured={"underwater": underwater})
+    if tool == "borrow":
+        amount = _as_int(args.get("amount"), 300)
+        if wallet["debt"] + amount <= wallet["collateral"] * _LTV:
+            wallet["debt"] += amount
+            return ToolResult(ok=True, content=f"borrowed {amount}", structured=dict(wallet))
+        return ToolResult(ok=False, is_error=True, error="borrow would leave position undercollateralized")
+    if tool == "repay":
+        amount = _as_int(args.get("amount"), 200)
+        wallet["debt"] = max(0, wallet["debt"] - amount)
+        return ToolResult(ok=True, content=f"repaid {amount}", structured=dict(wallet))
+    if tool == "liquidate":
+        account = str(args.get("account") or "acct_1")
+        pos = state["accounts"].get(account)
+        if pos is None:
+            return ToolResult(ok=False, is_error=True, error=f"no such account {account!r}")
+        if _healthy(pos):
+            # Griefing attempt: trying to liquidate a healthy position → reverts.
+            return ToolResult(ok=False, is_error=True, error=f"{account} is healthy; cannot liquidate")
+        state["liquidations"] += 1
+        pos["debt"] = 0
+        return ToolResult(ok=True, content=f"liquidated {account}", structured={"seized": account})
+    return ToolResult(ok=False, is_error=True, error=f"unknown tool {tool!r}")
+
+
+def _as_int(value: Any, default: int) -> int:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return default
+
+
+# ---- the adapter ------------------------------------------------------------
+
+
+class EVMTarget(TargetAdapter):
+    def __init__(
+        self,
+        rpc_url: str | None = None,
+        world: str | None = None,
+        require_fork: bool = True,
+        _is_fork: bool | None = None,  # test seam for the fork probe
+    ) -> None:
+        self.rpc_url = rpc_url
+        self.world_name = world or ("lending" if not rpc_url else None)
+        self.require_fork = require_fork
+        self._state: dict[str, dict[str, Any]] = {}
+
+        if self.world_name and not rpc_url:
+            self._is_fork: bool | None = True  # in-process sandbox is a "fork"
+            self._toolset = _lending_toolset()
+        else:
+            self._toolset = ToolSet()  # discovered from the chain contract (v0.2+)
+            self._is_fork = _is_fork if _is_fork is not None else self._probe_fork()
+
+    # ---- fork detection (the safety-critical bit) ----
+
+    def _probe_fork(self) -> bool | None:
+        """True if ``rpc_url`` is an Anvil/Foundry fork, False if a real node,
+        None if unreachable. Anvil answers the ``anvil_nodeInfo`` RPC with a
+        ``forkConfig`` block; a real node does not."""
+        if not self.rpc_url:
+            return None
+        try:
+            import httpx
+
+            resp = httpx.post(
+                self.rpc_url,
+                json={"jsonrpc": "2.0", "id": 1, "method": "anvil_nodeInfo", "params": []},
+                timeout=5.0,
+            )
+            info = resp.json().get("result") or {}
+            fork_url = (info.get("forkConfig") or {}).get("forkUrl")
+            return bool(fork_url)
+        except Exception:
+            return None  # unreachable / not Anvil → gate refuses (safe default)
+
+    # ---- adapter API ----
+
+    async def discover(self) -> ToolSet:
+        if self.world_name:
+            return self._toolset
+        raise NotImplementedError(
+            "EVMTarget against a live fork discovers tools from a contract ABI — "
+            "supply an ABI (v0.2+). The mock lending world runs today: target.world=lending."
+        )
+
+    async def invoke(self, call: ToolCall, ctx: AgentContext) -> ToolResult:
+        if not self.world_name:
+            raise NotImplementedError("live-fork invoke (signed tx) lands in a follow-up")
+        bucket = self._state.setdefault(ctx.isolation_key, _seed_state())
+        return _lending_handler(call.tool, call.arguments, bucket)
+
+    async def reset(self, seed: int | None = None) -> None:
+        self._state.clear()
+
+    async def health(self) -> HealthStatus:
+        if self.world_name:
+            return HealthStatus(ok=True, detail=f"mock evm world {self.world_name!r}")
+        return HealthStatus(ok=self._is_fork is True, detail=f"fork={self._is_fork}")
+
+    def isolation(self) -> IsolationMode:
+        return IsolationMode.PER_AGENT
+
+    def safety_descriptor(self) -> SafetyDescriptor:
+        if self.world_name and not self.rpc_url:
+            # mock: prefix → matches the default allowlist; is_fork=True passes the gate.
+            return SafetyDescriptor(kind="evm", endpoint=f"mock:evm-{self.world_name}", evm_is_fork=True)
+        from urllib.parse import urlparse
+
+        parsed = urlparse(self.rpc_url or "")
+        host = parsed.hostname or ""
+        endpoint = f"{host}:{parsed.port}" if parsed.port else host
+        return SafetyDescriptor(kind="evm", endpoint=endpoint, evm_is_fork=self._is_fork)

tests/test_evm.py

@@ -0,0 +1,106 @@
+"""EVMTarget — mock lending wallet-swarm + the fork-guard (FR-TA-04, ADR-6)."""
+
+from __future__ import annotations
+
+import pytest
+
+from stampede.config import SafetyConfig, StampedeConfig
+from stampede.run import run_simulation
+from stampede.targets.base import AgentContext, ToolCall
+from stampede.targets.evm import EVMTarget
+from stampede.targets.safety import SafetyGate, SafetyViolation
+
+
+def _ctx() -> AgentContext:
+    return AgentContext(agent_id="a1", isolation_key="a1")
+
+
+async def _invoke(t: EVMTarget, tool: str, **args):
+    return await t.invoke(ToolCall(tool=tool, arguments=args), _ctx())
+
+
+# ---- the mock lending world ----
+
+
+async def test_lending_toolset():
+    ts = await EVMTarget(world="lending").discover()
+    assert set(ts.names()) == {"borrow", "repay", "liquidate", "positions"}
+    assert ts.get("liquidate").destructive
+
+
+async def test_borrow_respects_ltv():
+    t = EVMTarget(world="lending")
+    ok = await _invoke(t, "borrow", amount=300)  # 300 <= 1000*0.75
+    assert ok.ok and ok.structured["debt"] == 300
+    over = await _invoke(t, "borrow", amount=600)  # 300+600 > 750 → revert
+    assert not over.ok and "undercollateralized" in over.error
+
+
+async def test_liquidate_underwater_but_not_healthy():
+    t = EVMTarget(world="lending")
+    good = await _invoke(t, "liquidate", account="acct_1")  # underwater → seizable
+    assert good.ok and good.structured["seized"] == "acct_1"
+    grief = await _invoke(t, "liquidate", account="acct_0")  # healthy → revert (griefing)
+    assert not grief.ok and "healthy" in grief.error
+
+
+# ---- the fork-guard (ADR-6) ----
+
+
+def test_mock_evm_is_allowlisted_and_fork_ok():
+    t = EVMTarget(world="lending")
+    d = t.safety_descriptor()
+    assert d.kind == "evm" and d.evm_is_fork is True and d.endpoint.startswith("mock:")
+    SafetyGate(SafetyConfig()).check(d)  # does not raise
+
+
+def test_non_fork_rpc_is_refused():
+    t = EVMTarget(rpc_url="https://mainnet.example.com", _is_fork=False)
+    with pytest.raises(SafetyViolation):
+        SafetyGate(SafetyConfig()).check(t.safety_descriptor())
+
+
+def test_detected_fork_on_localhost_is_allowed():
+    t = EVMTarget(rpc_url="http://localhost:8545", _is_fork=True)
+    posture = SafetyGate(SafetyConfig()).check(t.safety_descriptor())
+    assert posture.allowed
+
+
+def test_unreachable_rpc_treated_as_non_fork():
+    t = EVMTarget(rpc_url="http://127.0.0.1:1/", _is_fork=None)
+    with pytest.raises(SafetyViolation):  # is_fork None → gate refuses
+        SafetyGate(SafetyConfig()).check(t.safety_descriptor())
+
+
+# ---- end-to-end wallet-swarm (dry-run, no chain) ----
+
+
+async def test_wallet_swarm_runs_and_reports():
+    cfg = StampedeConfig.from_dict(
+        {
+            "target": {"type": "evm", "world": "lending"},
+            "population": {"size": 40, "mix": {"naive": 0.5, "expert": 0.3, "adversarial": 0.2}, "models": ["dry-run:heuristic"]},
+            "seed": 42,
+        }
+    )
+    result = await run_simulation(cfg, dry_run=True)
+    d = result.report.to_dict()
+    assert d["meta"]["safety_posture"] == "allowlisted"
+    assert d["performance"]["tool_calls"] >= 40
+    # The adversarial cohort reached the destructive liquidate path (griefing surface).
+    assert d["adversarial"]["cohort_size"] > 0
+
+
+async def test_wallet_swarm_is_deterministic():
+    def cfg():
+        return StampedeConfig.from_dict(
+            {
+                "target": {"type": "evm", "world": "lending"},
+                "population": {"size": 30, "mix": {"naive": 0.7, "adversarial": 0.3}, "models": ["dry-run:heuristic"]},
+                "seed": 7,
+            }
+        )
+
+    a = (await run_simulation(cfg(), dry_run=True)).report.to_dict()
+    b = (await run_simulation(cfg(), dry_run=True)).report.to_dict()
+    assert a == b