build(agent): r2d2#deee02 iteration

src/idea187_autopoiesis_verifiable_service/__init__.py

@@ -12,6 +12,7 @@ from .genome import (
 from .ledger import LedgerEntry, ProvenanceLedger, ReplayTrace, TraceDelta
 from .policy import PolicySandbox, PolicyTemplate, SafetyAttestation, SmokeTestResult
 from .replicator import ReplicaWeaveIntent, ReplayHarness, ReplicatorPlanner, ReplicationPlan, RolloutStage
+from .ems import ExplainableMutationSummary, generate_ems, CapabilityDelta, ResourceDelta
 
 __all__ = [
     "AdmissionDecision",
@@ -32,6 +33,10 @@ __all__ = [
     "ResourceEnvelope",
     "ReplayTrace",
     "RolloutStage",
+    "ExplainableMutationSummary",
+    "generate_ems",
+    "CapabilityDelta",
+    "ResourceDelta",
     "SignedAdmissionToken",
     "SafetyAttestation",
     "SafetyInvariants",

src/idea187_autopoiesis_verifiable_service/ems.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from .genome import ServiceGenome
+
+
+def _stable_json(value: Any) -> bytes:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8")
+
+
+class CapabilityDelta(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    added_network: list[str] = Field(default_factory=list)
+    removed_network: list[str] = Field(default_factory=list)
+    added_db: list[str] = Field(default_factory=list)
+    removed_db: list[str] = Field(default_factory=list)
+    added_actuators: list[str] = Field(default_factory=list)
+    removed_actuators: list[str] = Field(default_factory=list)
+    exposure_score_delta: int = 0
+
+
+class ResourceDelta(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    cpu_millicores: int
+    memory_mib: int
+    bandwidth_mbps: int
+    cost_usd_per_hour: float
+    max_replicas: int
+
+
+class ExplainableMutationSummary(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    capability_delta: CapabilityDelta
+    resource_delta: ResourceDelta
+    safety_risk_delta: float
+    counterfactual_sample: dict[str, Any] = Field(default_factory=dict)
+
+
+def _list_diff(before: list[str], after: list[str]) -> tuple[list[str], list[str]]:
+    bset = set(before)
+    aset = set(after)
+    added = sorted(list(aset - bset))
+    removed = sorted(list(bset - aset))
+    return added, removed
+
+
+def generate_ems(before: ServiceGenome, after: ServiceGenome) -> ExplainableMutationSummary:
+    # Capability diffs
+    added_net, removed_net = _list_diff(before.capabilities.network, after.capabilities.network)
+    added_db, removed_db = _list_diff(before.capabilities.db_access, after.capabilities.db_access)
+    added_act, removed_act = _list_diff(before.capabilities.actuators, after.capabilities.actuators)
+
+    # Simple exposure score: each network endpoint adds 3 points, each DB 5 points, actuators 7 points
+    before_exposure = 3 * len(before.capabilities.network) + 5 * len(before.capabilities.db_access) + 7 * len(
+        before.capabilities.actuators
+    )
+    after_exposure = 3 * len(after.capabilities.network) + 5 * len(after.capabilities.db_access) + 7 * len(
+        after.capabilities.actuators
+    )
+    exposure_delta = after_exposure - before_exposure
+
+    capability_delta = CapabilityDelta(
+        added_network=added_net,
+        removed_network=removed_net,
+        added_db=added_db,
+        removed_db=removed_db,
+        added_actuators=added_act,
+        removed_actuators=removed_act,
+        exposure_score_delta=exposure_delta,
+    )
+
+    # Resource deltas (after - before)
+    rd = ResourceDelta(
+        cpu_millicores=after.resource_envelope.cpu_millicores - before.resource_envelope.cpu_millicores,
+        memory_mib=after.resource_envelope.memory_mib - before.resource_envelope.memory_mib,
+        bandwidth_mbps=after.resource_envelope.bandwidth_mbps - before.resource_envelope.bandwidth_mbps,
+        cost_usd_per_hour=after.resource_envelope.cost_usd_per_hour - before.resource_envelope.cost_usd_per_hour,
+        max_replicas=after.resource_envelope.max_replicas - before.resource_envelope.max_replicas,
+    )
+
+    # Safety risk heuristics: penalize introduction of denied endpoints, increased max_exposure, and removed deterministic tests
+    risk = 0.0
+    added_denied = sorted(set(after.safety_invariants.denied_endpoints) - set(before.safety_invariants.denied_endpoints))
+    if added_denied:
+        risk += 25.0
+
+    risk += max(0, after.safety_invariants.max_exposure - before.safety_invariants.max_exposure) * 0.5
+
+    # If deterministic tests were required before but removed in the after genome, increase risk
+    if before.safety_invariants.require_deterministic_tests and not after.safety_invariants.require_deterministic_tests:
+        risk += 20.0
+
+    # Mutation operator changes
+    before_ops = {op.name for op in before.mutation_operators}
+    after_ops = {op.name for op in after.mutation_operators}
+    new_ops = sorted(after_ops - before_ops)
+    if new_ops:
+        risk += 5.0 * len(new_ops)
+
+    # Small normalization to keep risk in [0,100]
+    safety_risk_delta = min(100.0, max(0.0, risk))
+
+    # Counterfactual sample: deterministic tiny trace demonstrating primary delta
+    # Construct a small, reproducible seed from content hashes
+    before_hash = before.content_hash()
+    after_hash = after.content_hash()
+    seed = int(hashlib.sha256((before_hash + after_hash).encode("utf-8")).hexdigest()[:16], 16)
+    sample = {
+        "seed": seed,
+        "summary": {
+            "capability_changes": {
+                "added_network": capability_delta.added_network,
+                "removed_network": capability_delta.removed_network,
+            },
+            "resource_hint": {
+                "cpu_delta": rd.cpu_millicores,
+                "memory_delta": rd.memory_mib,
+            },
+        },
+    }
+
+    # Add a tiny deterministic counterexample string (<=512 bytes)
+    counterfactual_blob = hashlib.sha256(_stable_json(sample)).hexdigest()
+    sample["counterfactual_hash"] = counterfactual_blob
+
+    return ExplainableMutationSummary(
+        capability_delta=capability_delta, resource_delta=rd, safety_risk_delta=safety_risk_delta, counterfactual_sample=sample
+    )

tests/test_ems.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from idea187_autopoiesis_verifiable_service import (
+    ServiceGenome,
+    CapabilityVector,
+    ResourceEnvelope,
+    SafetyInvariants,
+    MutationOperator,
+    TestContract,
+    generate_ems,
+)
+
+
+def build_base() -> ServiceGenome:
+    return ServiceGenome(
+        name="svc",
+        version="0.1",
+        image_ref="ghcr.io/example/svc:0.1",
+        capabilities=CapabilityVector(network=["https://internal"], db_access=["sqlite"], actuators=[]),
+        resource_envelope=ResourceEnvelope(cpu_millicores=100, memory_mib=128, bandwidth_mbps=5, cost_usd_per_hour=0.01, max_replicas=1),
+        safety_invariants=SafetyInvariants(max_exposure=5, read_only_scopes=[], denied_endpoints=[]),
+        mutation_operators=[MutationOperator(name="safe-fork", kind="fork")],
+        test_contracts=[TestContract(name="t1", seed=1, input_fingerprint="a", expected_output_hash=__import__("hashlib").sha256("1:a".encode()).hexdigest())],
+    )
+
+
+def test_generate_ems_detects_added_network_and_resource_increase():
+    before = build_base()
+    after = before.model_copy(update={
+        "capabilities": before.capabilities.model_copy(update={"network": before.capabilities.network + ["https://exfil.example"]}),
+        "resource_envelope": before.resource_envelope.model_copy(update={"cpu_millicores": 200, "max_replicas": 2}),
+    })
+
+    ems = generate_ems(before, after)
+    assert "https://exfil.example" in ems.capability_delta.added_network
+    assert ems.resource_delta.cpu_millicores == 100
+    assert ems.resource_delta.max_replicas == 1
+    # safety risk should reflect the new network exposure
+    assert ems.safety_risk_delta >= 0
+    assert isinstance(ems.counterfactual_sample.get("seed"), int)