build(agent): r2d2#deee02 iteration

README.md

@@ -1,35 +1,10 @@
-ReplicaSurety — idea199-replicasurety-economic-surety
+ReplicaSurety — minimal reference implementation
-===============================================
 
-This repository implements a focused, production-minded chunk of the ReplicaSurety vision:
+This repository provides a focused implementation of core ReplicaSurety primitives:
 
 - PoeticCert and ReplayBundle JSON schemas (Pydantic models)
-- A tiny off-chain Surety Ledger (in-memory with JSON persistence)
+- A tiny off-chain Surety Ledger FastAPI service with basic escrow lifecycle endpoints
-- A deterministic ReplayBundle verifier that emits a signed verdict (demo)
+- A deterministic ReplayBundle verifier (reference behavior)
-- FastAPI endpoints to interact with the ledger and verifier
+- Tests and packaging metadata so `bash test.sh` runs pytest and builds the package
-- Tests, README, AGENTS.md, and a test.sh that runs tests and builds the package
 
-This is an intentionally small, well-tested foundation intended for further extension.
+This is intentionally minimal and meant as a starting point for the community PoC.
-
-Quickstart
-----------
-
-Install (editable):
-
-python3 -m pip install -e .
-
-Run tests and build (this repository includes test.sh):
-
-bash test.sh
-
-Run the demo API locally:
-
-uvicorn src.idea199_replicasurety_economic_surety.api:app --reload
-
-Project layout
---------------
-- src/idea199_replicasurety_economic_surety: implementation
-- tests/: pytest test suite
-- AGENTS.md: architecture and rules for contributors
-
-License: MIT

pyproject.toml

@@ -1,18 +1,20 @@
 [build-system]
-requires = ["setuptools", "wheel"]
+requires = ["setuptools>=42", "wheel"]
 build-backend = "setuptools.build_meta"
 
 [project]
 name = "idea199-replicasurety-economic-surety"
 version = "0.1.0"
-description = "ReplicaSurety: PoeticCert, ReplayBundle, and a tiny off-chain surety ledger and verifier"
+description = "ReplicaSurety: PoeticCert schemas, ReplayBundle format and a tiny off-chain surety ledger"
 readme = "README.md"
 requires-python = ">=3.9"
+license = {text = "MIT"}
 dependencies = [
     "fastapi>=0.95",
     "pydantic>=1.10",
-    "uvicorn>=0.18",
+    "cryptography>=40.0",
+    "httpx>=0.24",
 ]
 
-[project.urls]
+[tool.setuptools.packages.find]
-"Homepage" = "https://example.com/replicasurety"
+where = ["src"]

src/idea199_replicasurety_economic_surety/__init__.py

@@ -1,2 +1 @@
-"""idea199_replicasurety_economic_surety package"""
+__all__ = ["models", "ledger"]
-__version__ = "0.1.0"

src/idea199_replicasurety_economic_surety/crypto.py

@@ -0,0 +1,38 @@
+from cryptography.hazmat.primitives.asymmetric import rsa, padding
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat, PrivateFormat, NoEncryption
+
+
+def generate_rsa_keypair(key_size: int = 2048):
+    private_key = rsa.generate_private_key(public_exponent=65537, key_size=key_size)
+    public_key = private_key.public_key()
+    return private_key, public_key
+
+
+def sign_bytes(private_key, data: bytes) -> bytes:
+    return private_key.sign(
+        data,
+        padding.PSS(mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH),
+        hashes.SHA256(),
+    )
+
+
+def verify_bytes(public_key, signature: bytes, data: bytes) -> bool:
+    try:
+        public_key.verify(
+            signature,
+            data,
+            padding.PSS(mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.MAX_LENGTH),
+            hashes.SHA256(),
+        )
+        return True
+    except Exception:
+        return False
+
+
+def serialize_public_key(public_key) -> bytes:
+    return public_key.public_bytes(Encoding.PEM, PublicFormat.SubjectPublicKeyInfo)
+
+
+def serialize_private_key(private_key) -> bytes:
+    return private_key.private_bytes(Encoding.PEM, PrivateFormat.TraditionalOpenSSL, NoEncryption())

src/idea199_replicasurety_economic_surety/ledger.py

@@ -1,57 +1,51 @@
-import json
+from fastapi import FastAPI, HTTPException
-from dataclasses import dataclass, asdict
+from typing import Dict
-from typing import Dict, Optional
+from .models import BondRecord, ClaimSubmission, ReplayBundle
-from pathlib import Path
+from uuid import uuid4
+from datetime import datetime
+
+app = FastAPI(title="ReplicaSurety - Surety Ledger")
+
+# Simple in-memory store. This is intentionally minimal and pluggable.
+_BONDS: Dict[str, BondRecord] = {}
+_CLAIMS: Dict[str, ClaimSubmission] = {}
 
 
-@dataclass
+@app.post("/bonds", response_model=BondRecord)
-class Escrow:
+def create_bond(record: BondRecord):
-    escrow_id: str
+    if record.escrow_id in _BONDS:
-    issuer: str
+        raise HTTPException(status_code=400, detail="escrow_id already exists")
-    stake_amount: float
+    _BONDS[record.escrow_id] = record
-    policy_hash: str
+    return record
-    ttl: int
-    status: str = "open"
 
 
-class SuretyLedger:
+@app.get("/bonds/{escrow_id}", response_model=BondRecord)
-    """Tiny JSON-backed off-chain ledger for escrows.
+def get_bond(escrow_id: str):
+    r = _BONDS.get(escrow_id)
+    if not r:
+        raise HTTPException(status_code=404, detail="not found")
+    return r
 
-    This is intentionally minimal: an in-memory dict persisted to a JSON file for
-    demo and tests.
-    """
 
-    def __init__(self, path: Optional[str] = None):
+@app.post("/claims", response_model=ClaimSubmission)
-        self._path = Path(path) if path else None
+def submit_claim(claim: ClaimSubmission):
-        self._escrows: Dict[str, Escrow] = {}
+    if claim.claim_id in _CLAIMS:
-        if self._path and self._path.exists():
+        raise HTTPException(status_code=400, detail="claim_id already submitted")
-            self._load()
+    # Minimal validation: evidence_root looks present
+    if not claim.evidence_root:
+        raise HTTPException(status_code=400, detail="evidence_root required")
+    _CLAIMS[claim.claim_id] = claim
+    # In a real system: dispatch to verifier/adjudicator plugins
+    return claim
 
-    def _load(self):
-        with self._path.open("r", encoding="utf-8") as f:
-            raw = json.load(f)
-        for k, v in raw.items():
-            self._escrows[k] = Escrow(**v)
 
-    def _save(self):
+@app.post("/verify_replay")
-        if not self._path:
+def verify_replay(bundle: ReplayBundle):
-            return
+    # Minimal deterministic verifier: ensure trace seq monotonic and evidence root present
-        serial = {k: asdict(v) for k, v in self._escrows.items()}
+    last_seq = -1
-        with self._path.open("w", encoding="utf-8") as f:
+    for ev in bundle.trace:
-            json.dump(serial, f, indent=2)
+        if ev.seq <= last_seq:
-
+            raise HTTPException(status_code=400, detail="non-monotonic trace sequence")
-    def create_escrow(self, escrow: Escrow):
+        last_seq = ev.seq
-        if escrow.escrow_id in self._escrows:
+    # In a real verifier we'd replay deterministically and compute a DamageScore
-            raise ValueError("escrow exists")
+    return {"verdict": "ok", "evidence_root": bundle.evidence_root}
-        self._escrows[escrow.escrow_id] = escrow
-        self._save()
-
-    def get_escrow(self, escrow_id: str) -> Optional[Escrow]:
-        return self._escrows.get(escrow_id)
-
-    def settle_escrow(self, escrow_id: str):
-        e = self._escrows.get(escrow_id)
-        if not e:
-            raise KeyError("not found")
-        e.status = "settled"
-        self._save()

src/idea199_replicasurety_economic_surety/main.py

@@ -0,0 +1,3 @@
+from .ledger import app
+
+__all__ = ["app"]

src/idea199_replicasurety_economic_surety/models.py

@@ -0,0 +1,64 @@
+from typing import Dict, List, Optional, Any
+from pydantic import BaseModel, Field, constr
+from datetime import datetime
+
+
+class PoeticCert(BaseModel):
+    """Compact PoeticCert schema.
+
+    Fields:
+    - invariants: free-form list of invariant descriptors (strings or structured)
+    - resource_limits: optional resource constraints
+    - replica_intent_hash: hex string hash of the intent being certified
+    - expected_receipt: optional acceptance witness
+    - attester_signatures: mapping signer_id -> signature (base64 or hex)
+    - issued_at: timestamp
+    """
+
+    invariants: List[Any]
+    resource_limits: Optional[Dict[str, Any]] = None
+    replica_intent_hash: constr(min_length=1)
+    expected_receipt: Optional[Dict[str, Any]] = None
+    attester_signatures: Dict[str, str] = Field(default_factory=dict)
+    issued_at: datetime = Field(default_factory=datetime.utcnow)
+
+
+class TraceEvent(BaseModel):
+    seq: int
+    timestamp: datetime
+    message: str
+    meta: Optional[Dict[str, Any]] = None
+
+
+class ReplayBundle(BaseModel):
+    """Deterministic ReplayBundle reference format.
+
+    Fields:
+    - trace: sequence of TraceEvent
+    - monitors: signed attestations produced by monitors
+    - evidence_root: a top-level digest (hex) that represents merkle/root of evidence
+    - artifacts: optional dict of artifact_name -> digest
+    """
+
+    trace: List[TraceEvent]
+    monitors: Dict[str, str] = Field(default_factory=dict)
+    evidence_root: constr(min_length=1)
+    artifacts: Optional[Dict[str, str]] = None
+
+
+class BondRecord(BaseModel):
+    escrow_id: str
+    issuer: str
+    stake_amount: float
+    policy_hash: Optional[str] = None
+    ttl: Optional[datetime] = None
+    created_at: datetime = Field(default_factory=datetime.utcnow)
+    status: str = "active"  # active, settled, cancelled
+
+
+class ClaimSubmission(BaseModel):
+    claim_id: str
+    evidence_root: str
+    claimant: str
+    timestamp: datetime = Field(default_factory=datetime.utcnow)
+    details: Optional[Dict[str, Any]] = None

test.sh

@@ -1,10 +1,10 @@
 #!/usr/bin/env bash
 set -euo pipefail
-echo "Installing test dependencies..."
+
-python3 -m pip install -q --upgrade pip
+echo "Running tests..."
-python3 -m pip install -q pydantic fastapi uvicorn pytest
+pytest -q
-echo "Running pytest..."
+
-PYTHONPATH=src pytest -q
 echo "Building package..."
 python3 -m build
+
 echo "All done."

tests/test_ledger.py

@@ -1,15 +1,33 @@
-import tempfile
+from idea199_replicasurety_economic_surety import ledger
-import os
+from idea199_replicasurety_economic_surety.models import BondRecord, ClaimSubmission, ReplayBundle, TraceEvent
-from idea199_replicasurety_economic_surety.ledger import SuretyLedger, Escrow
+from datetime import datetime
 
 
-def test_ledger_create_and_get(tmp_path):
+def test_create_and_get_bond():
-    path = tmp_path / "ledger.json"
+    br = BondRecord(
-    ledger = SuretyLedger(path=str(path))
+        escrow_id="escrow-1",
-    e = Escrow(escrow_id="e1", issuer="me", stake_amount=10.0, policy_hash="ph", ttl=3600)
+        issuer="alice",
-    ledger.create_escrow(e)
+        stake_amount=100.0,
-    got = ledger.get_escrow("e1")
+        policy_hash="phash",
-    assert got is not None
+        ttl=None,
-    assert got.issuer == "me"
+    )
-    ledger.settle_escrow("e1")
+    r = ledger.create_bond(br)
-    assert ledger.get_escrow("e1").status == "settled"
+    assert r.escrow_id == "escrow-1"
+    got = ledger.get_bond("escrow-1")
+    assert got.issuer == "alice"
+
+
+def test_submit_claim():
+    claim = ClaimSubmission(claim_id="claim-1", evidence_root="evidence123", claimant="bob")
+    r = ledger.submit_claim(claim)
+    assert r.claim_id == "claim-1"
+
+
+def test_verify_replay():
+    trace = [
+        TraceEvent(seq=0, timestamp=datetime.utcnow(), message="a"),
+        TraceEvent(seq=1, timestamp=datetime.utcnow(), message="b"),
+    ]
+    rb = ReplayBundle(trace=trace, monitors={}, evidence_root="r")
+    r = ledger.verify_replay(rb)
+    assert r["verdict"] == "ok"

tests/test_models.py

@@ -0,0 +1,31 @@
+import json
+from idea199_replicasurety_economic_surety.models import PoeticCert, ReplayBundle, TraceEvent
+from idea199_replicasurety_economic_surety import crypto
+from datetime import datetime
+
+
+def test_poeticcert_roundtrip_and_signature():
+    priv, pub = crypto.generate_rsa_keypair()
+    cert = PoeticCert(
+        invariants=["no-privilege-escalation"],
+        replica_intent_hash="deadbeef",
+    )
+    # sign canonical bytes (here simple JSON)
+    b = cert.json().encode()
+    sig = crypto.sign_bytes(priv, b)
+    cert.attester_signatures["test-signer"] = sig.hex()
+
+    # verify
+    sig2 = bytes.fromhex(cert.attester_signatures["test-signer"])
+    assert crypto.verify_bytes(pub, sig2, b)
+
+
+def test_replaybundle_structure_and_verifier_input():
+    trace = [
+        TraceEvent(seq=0, timestamp=datetime.utcnow(), message="start"),
+        TraceEvent(seq=1, timestamp=datetime.utcnow(), message="do work"),
+    ]
+    rb = ReplayBundle(trace=trace, monitors={"m1": "sig"}, evidence_root="roothash")
+    s = rb.json()
+    parsed = ReplayBundle.parse_raw(s)
+    assert parsed.evidence_root == "roothash"