build(agent): semicolon#54de0b iteration

README.md

@@ -7,8 +7,9 @@ It focuses on the trust substrate needed for safe self-replication across edge a
 - `ServiceGenome` models immutable service blueprints with resources, capabilities, invariants, mutation operators, and test contracts.
 - `GenomeSigner` produces and verifies Ed25519 signatures for genome artifacts.
 - `PolicySandbox` checks a genome against operational budgets and deterministic smoke-test contracts.
-- `ProvenanceLedger` records append-only replication events with a Merkle root for tamper evidence.
+- `ProvenanceLedger` records append-only replication events with a Merkle root and chain verification for tamper evidence.
 - `ReplicatorPlanner` turns an approved genome into a deterministic rollout trace.
+- `ReplayHarness` replays and verifies a rollout trace against the genome and attestation that produced it.
 - `ReplicaWeaveIntent` provides a lightweight admission contract for solver/adapter integrations.
 
 ## What is implemented

src/idea187_autopoiesis_verifiable_service/__init__.py

@@ -10,7 +10,7 @@ from .genome import (
 )
 from .ledger import LedgerEntry, ProvenanceLedger, ReplayTrace, TraceDelta
 from .policy import PolicySandbox, PolicyTemplate, SafetyAttestation, SmokeTestResult
-from .replicator import ReplicaWeaveIntent, ReplicatorPlanner, ReplicationPlan, RolloutStage
+from .replicator import ReplicaWeaveIntent, ReplayHarness, ReplicatorPlanner, ReplicationPlan, RolloutStage
 
 __all__ = [
     "CapabilityVector",
@@ -21,6 +21,7 @@ __all__ = [
     "PolicyTemplate",
     "ProvenanceLedger",
     "ReplicaWeaveIntent",
+    "ReplayHarness",
     "ReplicatorPlanner",
     "ReplicationPlan",
     "ResourceEnvelope",

src/idea187_autopoiesis_verifiable_service/ledger.py

@@ -26,6 +26,12 @@ class LedgerEntry(BaseModel):
     previous_hash: str
     entry_hash: str
 
+    def payload(self) -> dict[str, Any]:
+        return self.model_dump(mode="json", exclude={"entry_hash"})
+
+    def recompute_hash(self) -> str:
+        return hashlib.sha256(_stable_json(self.payload())).hexdigest()
+
 
 class TraceDelta(BaseModel):
     model_config = ConfigDict(frozen=True, extra="forbid")
@@ -45,6 +51,12 @@ class ReplayTrace(BaseModel):
     def digest(self) -> str:
         return hashlib.sha256(_stable_json(self.model_dump(mode="json", exclude_none=True))).hexdigest()
 
+    def assert_valid(self, *, seed: int | None = None) -> None:
+        if seed is not None and self.seed != seed:
+            raise ValueError("trace seed does not match expected seed")
+        if len(self.signatures) != len(set(self.signatures)):
+            raise ValueError("trace signatures must be unique")
+
 
 class ProvenanceLedger:
     def __init__(self) -> None:
@@ -66,11 +78,12 @@ class ProvenanceLedger:
         timestamp: datetime | None = None,
     ) -> LedgerEntry:
         timestamp = timestamp or datetime.now(timezone.utc)
+        timestamp_json = timestamp.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
         index = len(self._entries)
         previous_hash = self._entries[-1].entry_hash if self._entries else hashlib.sha256(b"genesis").hexdigest()
         payload = {
             "index": index,
-            "timestamp": timestamp.isoformat(),
+            "timestamp": timestamp_json,
             "actor_did": actor_did,
             "action": action,
             "genome_hash": genome_hash,
@@ -84,6 +97,17 @@ class ProvenanceLedger:
         self._entries.append(entry)
         return entry
 
+    def verify(self) -> None:
+        expected_previous_hash = hashlib.sha256(b"genesis").hexdigest()
+        for index, entry in enumerate(self._entries):
+            if entry.index != index:
+                raise ValueError("ledger index mismatch")
+            if entry.previous_hash != expected_previous_hash:
+                raise ValueError("ledger previous hash mismatch")
+            if entry.recompute_hash() != entry.entry_hash:
+                raise ValueError("ledger entry hash mismatch")
+            expected_previous_hash = entry.entry_hash
+
     def merkle_root(self) -> str:
         if not self._entries:
             return hashlib.sha256(b"").hexdigest()

src/idea187_autopoiesis_verifiable_service/replicator.py

@@ -50,6 +50,17 @@ class ReplicationPlan(BaseModel):
         return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8")).hexdigest()
 
 
+def _rollout_stages(max_replicas: int, score: float) -> list[RolloutStage]:
+    canary = min(1, max_replicas)
+    cohort = max(1, min(max_replicas, max(2, max_replicas // 2)))
+    stages = [
+        RolloutStage(name="canary", cohort_size=canary, percentage=5),
+        RolloutStage(name="cohort", cohort_size=cohort, percentage=25),
+        RolloutStage(name="regional", cohort_size=max_replicas, percentage=100),
+    ]
+    return stages[: 2 if score < 85 else 3]
+
+
 class ReplicatorPlanner:
     def plan(self, genome: ServiceGenome, attestation: SafetyAttestation) -> ReplicationPlan:
         genome_hash = genome.content_hash()
@@ -59,18 +70,42 @@ class ReplicatorPlanner:
             trace = ReplayTrace(seed=trace_seed, deltas=[TraceDelta(stage="blocked", operation="deny", payload={"policy": attestation.policy_name})])
             return ReplicationPlan(genome_hash=genome_hash, approved=False, stages=[], trace=trace, rationale=rationale)
 
-        max_replicas = genome.resource_envelope.max_replicas
+        stages = _rollout_stages(genome.resource_envelope.max_replicas, attestation.score)
-        canary = min(1, max_replicas)
-        cohort = max(1, min(max_replicas, max(2, max_replicas // 2)))
-        stages = [
-            RolloutStage(name="canary", cohort_size=canary, percentage=5),
-            RolloutStage(name="cohort", cohort_size=cohort, percentage=25),
-            RolloutStage(name="regional", cohort_size=max_replicas, percentage=100),
-        ]
-        stages = stages[: 2 if attestation.score < 85 else 3]
         deltas = [
             TraceDelta(stage=stage.name, operation="rollout", payload={"cohort_size": stage.cohort_size, "percentage": stage.percentage})
             for stage in stages
         ]
         trace = ReplayTrace(seed=trace_seed, deltas=deltas, signatures=[genome_hash, attestation.digest()])
         return ReplicationPlan(genome_hash=genome_hash, approved=True, stages=stages, trace=trace, rationale=rationale)
+
+
+class ReplayHarness:
+    def verify(self, genome: ServiceGenome, attestation: SafetyAttestation, plan: ReplicationPlan) -> ReplayTrace:
+        genome_hash = genome.content_hash()
+        if plan.genome_hash != genome_hash:
+            raise ValueError("plan genome hash does not match genome content")
+
+        expected_seed = int(genome_hash[:16], 16)
+        plan.trace.assert_valid(seed=expected_seed)
+
+        if plan.approved != attestation.approved:
+            raise ValueError("plan approval does not match attestation")
+
+        if not plan.approved:
+            if plan.trace.deltas != [TraceDelta(stage="blocked", operation="deny", payload={"policy": attestation.policy_name})]:
+                raise ValueError("blocked trace does not match attestation")
+            return plan.trace
+
+        expected_stages = _rollout_stages(genome.resource_envelope.max_replicas, attestation.score)
+        expected_deltas = [
+            TraceDelta(stage=stage.name, operation="rollout", payload={"cohort_size": stage.cohort_size, "percentage": stage.percentage})
+            for stage in expected_stages
+        ]
+
+        if plan.stages != expected_stages:
+            raise ValueError("plan stages are not reproducible from the genome and attestation")
+        if plan.trace.deltas != expected_deltas:
+            raise ValueError("trace deltas are not reproducible from the rollout stages")
+        if plan.trace.signatures != [genome_hash, attestation.digest()]:
+            raise ValueError("trace signatures do not match the expected audit chain")
+        return plan.trace

tests/test_core.py

@@ -4,10 +4,13 @@ from datetime import datetime, timezone
 
 from idea187_autopoiesis_verifiable_service import (
     GenomeSigner,
+    CapabilityVector,
     PolicySandbox,
     PolicyTemplate,
     ProvenanceLedger,
+    ReplayHarness,
     ReplicatorPlanner,
+    MutationOperator,
     ResourceEnvelope,
     SafetyInvariants,
     ServiceGenome,
@@ -21,7 +24,7 @@ def build_genome() -> ServiceGenome:
         name="http-service",
         version="1.0.0",
         image_ref="ghcr.io/example/http-service:1.0.0",
-        capabilities={"network": ["https://internal.example"]},
+        capabilities=CapabilityVector(network=["https://internal.example"]),
         resource_envelope=ResourceEnvelope(
             cpu_millicores=250,
             memory_mib=256,
@@ -30,7 +33,7 @@ def build_genome() -> ServiceGenome:
             max_replicas=3,
         ),
         safety_invariants=SafetyInvariants(max_exposure=10, read_only_scopes=["/var/lib/cache"], denied_endpoints=["https://example.com"]),
-        mutation_operators=[{"name": "safe-fork", "kind": "fork"}],
+        mutation_operators=[MutationOperator(name="safe-fork", kind="fork")],
         test_contracts=[TestContract(name="startup", seed=7, input_fingerprint="boot", expected_output_hash=expected)],
     )
 
@@ -104,3 +107,50 @@ def test_ledger_merkle_root_and_replay_trace_are_deterministic() -> None:
     assert ledger.merkle_root()
     assert plan.approved is True
     assert plan.trace.digest() == plan.trace.digest()
+    ledger.verify()
+    assert ReplayHarness().verify(genome, attestation, plan) == plan.trace
+
+
+def test_audit_verification_rejects_tampering() -> None:
+    genome = build_genome()
+    policy = PolicyTemplate(
+        name="default-edge",
+        max_cpu_millicores=500,
+        max_memory_mib=512,
+        max_bandwidth_mbps=100,
+        max_cost_usd_per_hour=1.0,
+        allow_network_egress=True,
+        allowed_egress_endpoints=["https://internal.example"],
+        allowed_mutation_operators=["safe-fork"],
+    )
+    attestation = PolicySandbox().evaluate(genome, policy)
+    plan = ReplicatorPlanner().plan(genome, attestation)
+
+    tampered_plan = plan.model_copy(update={"trace": plan.trace.model_copy(update={"seed": plan.trace.seed + 1})})
+
+    ledger = ProvenanceLedger()
+    ledger.append(
+        actor_did="did:key:z6Mkh1",
+        action="replicate",
+        genome_hash=genome.content_hash(),
+        attestation_hash=attestation.digest(),
+        trace_hash=plan.trace.digest(),
+        resource_snapshot=genome.resource_envelope.model_dump(mode="json"),
+        timestamp=datetime(2024, 1, 1, tzinfo=timezone.utc),
+    )
+
+    ledger._entries[0] = ledger._entries[0].model_copy(update={"previous_hash": "broken"})
+
+    try:
+        ledger.verify()
+    except ValueError as exc:
+        assert "previous hash" in str(exc)
+    else:
+        raise AssertionError("expected ledger verification to fail")
+
+    try:
+        ReplayHarness().verify(genome, attestation, tampered_plan)
+    except ValueError as exc:
+        assert "seed" in str(exc)
+    else:
+        raise AssertionError("expected trace verification to fail")