diff --git a/src/idea187_autopoiesis_verifiable_service/__init__.py b/src/idea187_autopoiesis_verifiable_service/__init__.py index 2a13acf..ad53ac9 100644 --- a/src/idea187_autopoiesis_verifiable_service/__init__.py +++ b/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", diff --git a/src/idea187_autopoiesis_verifiable_service/ems.py b/src/idea187_autopoiesis_verifiable_service/ems.py new file mode 100644 index 0000000..ce97f9b --- /dev/null +++ b/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 + ) diff --git a/tests/test_ems.py b/tests/test_ems.py new file mode 100644 index 0000000..0c12d49 --- /dev/null +++ b/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)