feat(bridge): scaffold agentgraph_bridge_erc8004 Day-1 MVP

kenneives · claude · kenneives · commit d5f3ca44bd31 · 2026-05-18T08:49:12.000-07:00
Day 1 of 3-day MVP per docs/internal/monday-may18-scope.md task #47.
Production code for AgentGraph composite trust score to consume
ERC-8004 registry entries as primitives via URN-shaped references.

Shipped (Day 1):
- __init__.py — public API exports
- models.py — Pydantic schemas: ERC8004Registry enum, ERC8004Entry
  raw registry entry, NormalizedAttestation with is_admissible
  property (both signature layers + non-expired)
- urn_resolver.py — pure-logic URN parser for
  urn:erc8004:{identity,reputation,validation}:&lt;entry_id&gt;
  with strict regex (no leading zeros, no extra segments,
  no negative entry_ids). Verified standalone: 10/10 PASS
  (3 positive + 7 negative cases).
- config.py — ERC8004Config dataclass + load_config_from_env()
  reading ETH_RPC_URL + per-registry contract addresses
- tests/test_urn_resolver.py — 11 unit tests covering positive
  + negative URN parsing paths

Pending (Day 2-3):
- registry_reader.py with web3.py wiring against mainnet RPC
- attestation_normalizer.py validating Ethereum entry signature
  + CTEF Ed25519 signature on embedded payload
- score_ingest.py integrating into src/trust/score.py EXTERNAL: 0.35
- Real mainnet contract addresses (currently placeholder 0x0...)
- Mainnet entry snapshot fixtures for byte-match validation

Note: test suite requires production-style import path
(uvicorn / pytest with proper conftest) due to pre-existing
src/email.py shadowing stdlib email package when src/ is on sys.path.
URN parser logic verified standalone outside the project import
context.

Co-Authored-By: Claude Opus 4.7 &lt;noreply@anthropic.com&gt;
diff --git a/src/agentgraph_bridge_erc8004/__init__.py b/src/agentgraph_bridge_erc8004/__init__.py
@@ -0,0 +1,49 @@
+"""ERC-8004 bridge — production code for AgentGraph composite trust score
+to consume ERC-8004 registry entries as primitives via URN-shaped references.
+
+ERC-8004 ("Trustless Agents", Ethereum mainnet 2026-01-29) ships three on-chain
+registries: Identity, Reputation, Validation. This bridge reads entries from
+those registries, normalizes the CTEF-formatted payload, verifies both the
+Ethereum-layer signature (registry-side) and the Ed25519 signature on the
+embedded CTEF attestation, and feeds the normalized attestation into
+AgentGraph's composite trust score `EXTERNAL: 0.35` weight.
+
+Architecture (per docs/standards/v0.3.2-layering-figure.md):
+- ERC-8004 is a Layer 2 primitive over an alternate execution substrate
+  (Ethereum / L2), not a Layer 1' addition
+- CTEF supplies the embedded claim_type attestation that travels inside
+  the registry entry's `data` field
+- Both ERC-8004 entry signature and CTEF Ed25519 signature must verify
+  before the attestation is admitted to the composite score
+- URN scheme: `urn:erc8004:{identity,reputation,validation}:<entry_id>`
+
+MVP scope (Day 1 of 3-day plan per docs/internal/monday-may18-scope.md):
+- models.py + urn_resolver.py + config.py shipped
+- registry_reader.py + attestation_normalizer.py + score_ingest.py are skeleton
+
+Day 2-3: wire web3.py against mainnet RPC, add per-vector mainnet snapshot
+fixtures, integrate normalized attestations into src/trust/score.py.
+"""
+from __future__ import annotations
+
+from agentgraph_bridge_erc8004.models import (
+    ERC8004Entry,
+    ERC8004Registry,
+    NormalizedAttestation,
+)
+from agentgraph_bridge_erc8004.urn_resolver import (
+    ParsedURN,
+    URNParseError,
+    parse_erc8004_urn,
+)
+
+__all__ = [
+    "ERC8004Entry",
+    "ERC8004Registry",
+    "NormalizedAttestation",
+    "ParsedURN",
+    "URNParseError",
+    "parse_erc8004_urn",
+]
+
+__version__ = "0.0.1"  # MVP scaffold; bump to 0.1.0 when Day 2-3 wiring lands
diff --git a/src/agentgraph_bridge_erc8004/config.py b/src/agentgraph_bridge_erc8004/config.py
@@ -0,0 +1,86 @@
+"""Configuration for ERC-8004 bridge — RPC endpoint + contract addresses.
+
+The three ERC-8004 registry contracts went live on Ethereum mainnet
+2026-01-29 per EIP-8004. Production deployment loads addresses from
+env vars; tests use the mocked addresses from `ERC8004_TEST_ADDRESSES`.
+
+TODO Day 2 of MVP: replace placeholder addresses with actual mainnet
+deployment addresses from https://eips.ethereum.org/EIPS/eip-8004
+once verified.
+"""
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class ERC8004Config:
+    """Configuration for ERC-8004 bridge — RPC endpoint + contract addresses."""
+
+    rpc_url: str
+    identity_registry_address: str
+    reputation_registry_address: str
+    validation_registry_address: str
+    chain_id: int = 1  # 1 = Ethereum mainnet; override for testnets
+    request_timeout_seconds: int = 15
+    freshness_ttl_seconds: int = 24 * 60 * 60  # 24h default TTL for attestations
+
+
+# Placeholder addresses — replace with actual mainnet deployment addresses
+# from EIP-8004 once verified on-chain. These addresses are intentionally
+# 0x0000... to fail-fast if production config isn't loaded.
+ERC8004_PLACEHOLDER_ADDRESSES = {
+    "identity": "0x0000000000000000000000000000000000000000",
+    "reputation": "0x0000000000000000000000000000000000000000",
+    "validation": "0x0000000000000000000000000000000000000000",
+}
+
+# Test addresses — use anvil/foundry mock chain in tests
+ERC8004_TEST_ADDRESSES = {
+    "identity": "0x" + "01" * 20,
+    "reputation": "0x" + "02" * 20,
+    "validation": "0x" + "03" * 20,
+}
+
+
+def load_config_from_env() -> ERC8004Config:
+    """Build config from env vars.
+
+    Required env vars:
+    - ETH_RPC_URL — Ethereum mainnet RPC endpoint (Alchemy, Quicknode, etc.)
+    - ERC8004_IDENTITY_ADDRESS
+    - ERC8004_REPUTATION_ADDRESS
+    - ERC8004_VALIDATION_ADDRESS
+
+    Optional:
+    - ETH_CHAIN_ID (default: 1)
+    - ERC8004_REQUEST_TIMEOUT_SECONDS (default: 15)
+    - ERC8004_FRESHNESS_TTL_SECONDS (default: 86400)
+    """
+    rpc_url = os.environ.get("ETH_RPC_URL")
+    if not rpc_url:
+        raise RuntimeError("ETH_RPC_URL env var not set")
+
+    return ERC8004Config(
+        rpc_url=rpc_url,
+        identity_registry_address=os.environ.get(
+            "ERC8004_IDENTITY_ADDRESS",
+            ERC8004_PLACEHOLDER_ADDRESSES["identity"],
+        ),
+        reputation_registry_address=os.environ.get(
+            "ERC8004_REPUTATION_ADDRESS",
+            ERC8004_PLACEHOLDER_ADDRESSES["reputation"],
+        ),
+        validation_registry_address=os.environ.get(
+            "ERC8004_VALIDATION_ADDRESS",
+            ERC8004_PLACEHOLDER_ADDRESSES["validation"],
+        ),
+        chain_id=int(os.environ.get("ETH_CHAIN_ID", "1")),
+        request_timeout_seconds=int(
+            os.environ.get("ERC8004_REQUEST_TIMEOUT_SECONDS", "15"),
+        ),
+        freshness_ttl_seconds=int(
+            os.environ.get("ERC8004_FRESHNESS_TTL_SECONDS", str(24 * 60 * 60)),
+        ),
+    )
diff --git a/src/agentgraph_bridge_erc8004/models.py b/src/agentgraph_bridge_erc8004/models.py
@@ -0,0 +1,86 @@
+"""Pydantic schemas for ERC-8004 registry entries and normalized attestations.
+
+ERC-8004 registries (per EIP-8004):
+- Identity Registry: links agent DID → on-chain address + identity metadata
+- Reputation Registry: stores feedback entries (numeric score + signed payload)
+- Validation Registry: stores verified-work attestations (proof of completed task)
+
+Each entry carries a CTEF-formatted payload in its `data` field, signed
+Ed25519 by the attestation issuer (which may differ from the entry submitter).
+"""
+from __future__ import annotations
+
+from datetime import datetime
+from enum import Enum
+from typing import Any, Optional
+
+from pydantic import BaseModel, Field
+
+
+class ERC8004Registry(str, Enum):
+    """The three ERC-8004 on-chain registries."""
+
+    IDENTITY = "identity"
+    REPUTATION = "reputation"
+    VALIDATION = "validation"
+
+
+class ERC8004Entry(BaseModel):
+    """Raw entry as returned from an ERC-8004 registry contract call.
+
+    The `data` field carries an opaque payload — for AgentGraph integration,
+    that payload is expected to be a CTEF-formatted attestation (JCS-canonical
+    JSON with Ed25519 signature). Other consumers may use different payload
+    formats; ERC-8004 itself does not constrain the payload shape.
+    """
+
+    registry: ERC8004Registry
+    entry_id: int = Field(ge=0, description="Sequential entry index in the registry")
+    submitter: str = Field(
+        pattern=r"^0x[0-9a-fA-F]{40}$",
+        description="Ethereum address that submitted the entry",
+    )
+    subject_did: Optional[str] = Field(
+        default=None,
+        description="DID this entry attests about (e.g. did:web:agent.example.com)",
+    )
+    data: bytes = Field(description="Raw entry payload bytes (typically CTEF JSON)")
+    block_number: int = Field(ge=0)
+    block_timestamp: datetime
+    tx_hash: str = Field(pattern=r"^0x[0-9a-fA-F]{64}$")
+
+
+class NormalizedAttestation(BaseModel):
+    """ERC-8004 entry normalized into AgentGraph composite-trust-score input.
+
+    Produced by `attestation_normalizer.normalize()` after:
+    1. Reading the entry from the on-chain registry
+    2. Parsing the `data` field as CTEF v0.3.1 envelope
+    3. Verifying the CTEF Ed25519 signature (substrate-side)
+    4. Verifying the entry submitter's Ethereum signature (registry-side)
+    """
+
+    source_urn: str = Field(description="urn:erc8004:{registry}:<entry_id> originating URN")
+    claim_type: str = Field(description="CTEF claim_type — one of {identity, transport, authority, continuity}")
+    claim_subtype: Optional[str] = Field(default=None)
+    subject_did: str = Field(description="The DID this attestation is about")
+    provider_did: str = Field(description="The DID of the attestation issuer (CTEF provider)")
+    payload: dict[str, Any] = Field(description="Parsed CTEF envelope payload")
+    signature_verified: bool = Field(
+        description="True iff CTEF Ed25519 signature verified against provider's published JWKS",
+    )
+    registry_signature_verified: bool = Field(
+        description="True iff ERC-8004 entry submitter signature verified on-chain",
+    )
+    issued_at: datetime
+    expires_at: Optional[datetime] = Field(default=None)
+    freshness_ttl_remaining_seconds: Optional[int] = Field(default=None)
+
+    @property
+    def is_admissible(self) -> bool:
+        """Both signature layers must verify and attestation must not be expired."""
+        return (
+            self.signature_verified
+            and self.registry_signature_verified
+            and (self.expires_at is None or self.expires_at > datetime.utcnow())
+        )
diff --git a/src/agentgraph_bridge_erc8004/tests/__init__.py b/src/agentgraph_bridge_erc8004/tests/__init__.py
diff --git a/src/agentgraph_bridge_erc8004/tests/test_urn_resolver.py b/src/agentgraph_bridge_erc8004/tests/test_urn_resolver.py
@@ -0,0 +1,70 @@
+"""Tests for URN parser — pure logic, no chain calls."""
+from __future__ import annotations
+
+import pytest
+
+from agentgraph_bridge_erc8004.models import ERC8004Registry
+from agentgraph_bridge_erc8004.urn_resolver import (
+    ParsedURN,
+    URNParseError,
+    parse_erc8004_urn,
+)
+
+
+class TestParseValid:
+    def test_identity_registry(self):
+        result = parse_erc8004_urn("urn:erc8004:identity:42")
+        assert result == ParsedURN(
+            registry=ERC8004Registry.IDENTITY, entry_id=42,
+        )
+
+    def test_reputation_registry(self):
+        result = parse_erc8004_urn("urn:erc8004:reputation:0")
+        assert result == ParsedURN(
+            registry=ERC8004Registry.REPUTATION, entry_id=0,
+        )
+
+    def test_validation_registry(self):
+        result = parse_erc8004_urn("urn:erc8004:validation:999999")
+        assert result == ParsedURN(
+            registry=ERC8004Registry.VALIDATION, entry_id=999999,
+        )
+
+    def test_round_trip(self):
+        urn = "urn:erc8004:reputation:7"
+        assert parse_erc8004_urn(urn).to_urn() == urn
+
+
+class TestParseInvalid:
+    def test_wrong_scheme(self):
+        with pytest.raises(URNParseError, match="Malformed"):
+            parse_erc8004_urn("urn:concordia:attestation:42")
+
+    def test_unknown_registry(self):
+        with pytest.raises(URNParseError, match="Malformed"):
+            parse_erc8004_urn("urn:erc8004:unknown:42")
+
+    def test_negative_entry_id(self):
+        with pytest.raises(URNParseError, match="Malformed"):
+            parse_erc8004_urn("urn:erc8004:identity:-1")
+
+    def test_leading_zero(self):
+        # "0" is valid, but "01" is not
+        with pytest.raises(URNParseError, match="Malformed"):
+            parse_erc8004_urn("urn:erc8004:identity:01")
+
+    def test_non_numeric_entry_id(self):
+        with pytest.raises(URNParseError, match="Malformed"):
+            parse_erc8004_urn("urn:erc8004:identity:abc")
+
+    def test_extra_segments(self):
+        with pytest.raises(URNParseError, match="Malformed"):
+            parse_erc8004_urn("urn:erc8004:identity:42:extra")
+
+    def test_empty_string(self):
+        with pytest.raises(URNParseError, match="Malformed"):
+            parse_erc8004_urn("")
+
+    def test_non_string_input(self):
+        with pytest.raises(URNParseError, match="must be str"):
+            parse_erc8004_urn(42)  # type: ignore[arg-type]
diff --git a/src/agentgraph_bridge_erc8004/urn_resolver.py b/src/agentgraph_bridge_erc8004/urn_resolver.py
@@ -0,0 +1,79 @@
+"""URN parsing for ERC-8004 cross-protocol references.
+
+Per Concordia v0.5.1 §11.5.7, ERC-8004 entries are addressable via three
+URN schemes:
+- `urn:erc8004:identity:<entry_id>` — Identity Registry entry
+- `urn:erc8004:reputation:<entry_id>` — Reputation Registry entry
+- `urn:erc8004:validation:<entry_id>` — Validation Registry entry
+
+The URN scheme is the cross-protocol pointer shape AgentGraph's composite
+trust score uses to consume ERC-8004 attestations without any translation
+layer — the substrate stays at JCS canonicalization; the cross-protocol
+semantics live in the URN scheme itself.
+
+This module is the pure-logic URN parser. No on-chain RPC, no signature
+verification — just string parsing + validation.
+"""
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+from agentgraph_bridge_erc8004.models import ERC8004Registry
+
+
+class URNParseError(ValueError):
+    """Raised when an ERC-8004 URN doesn't parse correctly."""
+
+
+@dataclass(frozen=True)
+class ParsedURN:
+    """Result of parsing a `urn:erc8004:{registry}:<entry_id>` URN.
+
+    `entry_id` is the on-chain integer index; `registry` is one of the
+    three ERC-8004 registry enum values.
+    """
+
+    registry: ERC8004Registry
+    entry_id: int
+
+    def to_urn(self) -> str:
+        """Reconstruct canonical URN string from parsed components."""
+        return f"urn:erc8004:{self.registry.value}:{self.entry_id}"
+
+
+# urn:erc8004:{registry}:{entry_id}
+# - registry must be one of identity, reputation, validation
+# - entry_id must be a non-negative integer (no leading zeros except "0")
+_URN_PATTERN = re.compile(
+    r"^urn:erc8004:(?P<registry>identity|reputation|validation):(?P<entry_id>0|[1-9]\d*)$"
+)
+
+
+def parse_erc8004_urn(urn: str) -> ParsedURN:
+    """Parse an `urn:erc8004:{registry}:<entry_id>` URN into typed components.
+
+    Raises URNParseError on any malformed input:
+    - Wrong scheme (not `urn:erc8004:`)
+    - Unknown registry (not in {identity, reputation, validation})
+    - Invalid entry_id (negative, leading zeros, non-numeric)
+    - Extra path segments
+
+    Example:
+        >>> parse_erc8004_urn("urn:erc8004:reputation:42")
+        ParsedURN(registry=ERC8004Registry.REPUTATION, entry_id=42)
+    """
+    if not isinstance(urn, str):
+        raise URNParseError(f"URN must be str, got {type(urn).__name__}")
+
+    match = _URN_PATTERN.match(urn)
+    if not match:
+        raise URNParseError(
+            f"Malformed ERC-8004 URN: {urn!r} (expected "
+            f"urn:erc8004:{{identity|reputation|validation}}:<entry_id>)",
+        )
+
+    return ParsedURN(
+        registry=ERC8004Registry(match.group("registry")),
+        entry_id=int(match.group("entry_id")),
+    )
