diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..bd5590b --- /dev/null +++ b/.gitignore @@ -0,0 +1,21 @@ +node_modules/ +.npmrc +.env +.env.* +__tests__/ +coverage/ +.nyc_output/ +dist/ +build/ +.cache/ +*.log +.DS_Store +tmp/ +.tmp/ +__pycache__/ +*.pyc +.venv/ +venv/ +*.egg-info/ +.pytest_cache/ +READY_TO_PUBLISH diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..1d5862e --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,34 @@ +# AGENTS.md + +## Repository Purpose + +This repository contains a Python reference implementation for the Autopoiesis service-genome fabric. + +## Architecture + +- `genome.py`: genome schema, canonical hashing, Ed25519 signing, signed artifact verification. +- `policy.py`: policy templates, deterministic smoke-test evaluation, safety attestation. +- `ledger.py`: append-only provenance ledger, Merkle root computation, replay trace models. +- `replicator.py`: deterministic replication planner and lightweight ReplicaWeave intent model. + +## Tech Stack + +- Python 3.11+ +- `pydantic` for strict data models +- `cryptography` for Ed25519 signing and verification +- `pytest` for tests +- `build` for packaging verification + +## Rules + +- Keep changes deterministic and canonical where hashes or signatures are involved. +- Preserve frozen model semantics unless there is a concrete reason to relax them. +- Prefer the smallest correct change that improves safety, verification, or reproducibility. +- Update tests whenever schema, policy, or hashing behavior changes. +- Keep public APIs importable from `idea187_autopoiesis_verifiable_service.__init__`. + +## Test Commands + +- `bash test.sh` +- `python3 -m build` +- `python3 -m pytest` diff --git a/README.md b/README.md index 60d8c21..263ebe4 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,85 @@ -# idea187-autopoiesis-verifiable-service +# Autopoiesis: Verifiable Service-Genome Fabric -Source logic for Idea #187 \ No newline at end of file +Autopoiesis is a Python reference implementation for a compact, verifiable service-genome layer. + +It focuses on the trust substrate needed for safe self-replication across edge and cloud environments: + +- `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. +- `ReplicatorPlanner` turns an approved genome into a deterministic rollout trace. +- `ReplicaWeaveIntent` provides a lightweight admission contract for solver/adapter integrations. + +## What is implemented + +This repository ships a production-shaped core for the autonomy fabric: + +- schema validation for genomes and policies +- content addressing via canonical JSON hashing +- signed artifact creation and verification +- deterministic smoke-test evaluation +- provenance logging with Merkle aggregation +- replayable replication traces +- a small test suite and buildable Python package + +## Install + +```bash +python3 -m pip install -e .[dev] +``` + +## Test + +```bash +bash test.sh +``` + +## Example + +```python +from idea187_autopoiesis_verifiable_service import ( + GenomeSigner, + PolicySandbox, + PolicyTemplate, + ProvenanceLedger, + ReplicatorPlanner, + ResourceEnvelope, + SafetyInvariants, + ServiceGenome, + TestContract, +) + +genome = ServiceGenome( + name="http-service", + version="1.0.0", + image_ref="ghcr.io/acme/http-service:1.0.0", + capabilities={"network": ["https://internal.example"]}, + resource_envelope=ResourceEnvelope(cpu_millicores=250, memory_mib=256, bandwidth_mbps=20, cost_usd_per_hour=0.04, 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"}], + test_contracts=[TestContract(name="startup", seed=7, input_fingerprint="boot", expected_output_hash="...")], +) + +signer = GenomeSigner.generate() +artifact = signer.sign(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) +attestation = PolicySandbox().evaluate(genome, policy) + +ledger = ProvenanceLedger() +entry = ledger.append( + actor_did="did:key:z6Mkh...", + action="replicate", + genome_hash=artifact.genome_hash, + attestation_hash=attestation.digest(), + trace_hash="trace-hash", + resource_snapshot=genome.resource_envelope.model_dump(mode="json"), +) + +plan = ReplicatorPlanner().plan(genome, attestation) +``` + +## Package + +The package name is `idea187-autopoiesis-verifiable-service` and the import name is `idea187_autopoiesis_verifiable_service`. diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..3d346c9 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,26 @@ +[build-system] +requires = ["setuptools>=69", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "idea187-autopoiesis-verifiable-service" +version = "0.1.0" +description = "Verifiable service-genome fabric for safe self-replication and auditable provenance" +readme = "README.md" +requires-python = ">=3.11" +dependencies = [ + "cryptography>=42.0.0", + "pydantic>=2.7.0", +] + +[project.optional-dependencies] +dev = [ + "build>=1.2.0", + "pytest>=8.0.0", +] + +[tool.setuptools] +package-dir = {"" = "src"} + +[tool.setuptools.packages.find] +where = ["src"] diff --git a/src/idea187_autopoiesis_verifiable_service/__init__.py b/src/idea187_autopoiesis_verifiable_service/__init__.py new file mode 100644 index 0000000..0333725 --- /dev/null +++ b/src/idea187_autopoiesis_verifiable_service/__init__.py @@ -0,0 +1,36 @@ +from .genome import ( + CapabilityVector, + GenomeSigner, + MutationOperator, + ResourceEnvelope, + SafetyInvariants, + ServiceGenome, + SignedGenomeArtifact, + TestContract, +) +from .ledger import LedgerEntry, ProvenanceLedger, ReplayTrace, TraceDelta +from .policy import PolicySandbox, PolicyTemplate, SafetyAttestation, SmokeTestResult +from .replicator import ReplicaWeaveIntent, ReplicatorPlanner, ReplicationPlan, RolloutStage + +__all__ = [ + "CapabilityVector", + "GenomeSigner", + "LedgerEntry", + "MutationOperator", + "PolicySandbox", + "PolicyTemplate", + "ProvenanceLedger", + "ReplicaWeaveIntent", + "ReplicatorPlanner", + "ReplicationPlan", + "ResourceEnvelope", + "ReplayTrace", + "RolloutStage", + "SafetyAttestation", + "SafetyInvariants", + "ServiceGenome", + "SignedGenomeArtifact", + "SmokeTestResult", + "TestContract", + "TraceDelta", +] diff --git a/src/idea187_autopoiesis_verifiable_service/genome.py b/src/idea187_autopoiesis_verifiable_service/genome.py new file mode 100644 index 0000000..40d4a63 --- /dev/null +++ b/src/idea187_autopoiesis_verifiable_service/genome.py @@ -0,0 +1,166 @@ +from __future__ import annotations + +import base64 +import hashlib +import json +from dataclasses import dataclass +from typing import Any + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey +from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat +from pydantic import BaseModel, ConfigDict, Field, model_validator + + +def _canonical_json(value: Any) -> bytes: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8") + + +class CapabilityVector(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + network: list[str] = Field(default_factory=list) + sensors: list[str] = Field(default_factory=list) + actuators: list[str] = Field(default_factory=list) + db_access: list[str] = Field(default_factory=list) + + +class ResourceEnvelope(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 = 1 + + @model_validator(mode="after") + def _validate_positive(self) -> "ResourceEnvelope": + fields = { + "cpu_millicores": self.cpu_millicores, + "memory_mib": self.memory_mib, + "bandwidth_mbps": self.bandwidth_mbps, + "cost_usd_per_hour": self.cost_usd_per_hour, + "max_replicas": self.max_replicas, + } + for name, value in fields.items(): + if value <= 0: + raise ValueError(f"{name} must be positive") + return self + + +class SafetyInvariants(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + max_exposure: int + read_only_scopes: list[str] = Field(default_factory=list) + denied_endpoints: list[str] = Field(default_factory=list) + require_deterministic_tests: bool = True + + @model_validator(mode="after") + def _validate_exposure(self) -> "SafetyInvariants": + if not 0 <= self.max_exposure <= 100: + raise ValueError("max_exposure must be between 0 and 100") + return self + + +class MutationOperator(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + name: str + kind: str + parameters: dict[str, Any] = Field(default_factory=dict) + + +class TestContract(BaseModel): + __test__ = False + + model_config = ConfigDict(frozen=True, extra="forbid") + + name: str + seed: int + input_fingerprint: str + expected_output_hash: str + + @model_validator(mode="after") + def _validate_hash(self) -> "TestContract": + if len(self.expected_output_hash) != 64: + raise ValueError("expected_output_hash must be a 64-character sha256 hex digest") + return self + + +class ServiceGenome(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + name: str + version: str + image_ref: str + capabilities: CapabilityVector = Field(default_factory=CapabilityVector) + resource_envelope: ResourceEnvelope + safety_invariants: SafetyInvariants + mutation_operators: list[MutationOperator] = Field(default_factory=list) + test_contracts: list[TestContract] = Field(default_factory=list) + metadata: dict[str, Any] = Field(default_factory=dict) + + @model_validator(mode="after") + def _validate_identity(self) -> "ServiceGenome": + if not self.name.strip(): + raise ValueError("name cannot be empty") + if not self.version.strip(): + raise ValueError("version cannot be empty") + if not self.image_ref.strip(): + raise ValueError("image_ref cannot be empty") + return self + + def canonical_dict(self) -> dict[str, Any]: + return self.model_dump(mode="json", exclude_none=True) + + def canonical_bytes(self) -> bytes: + return _canonical_json(self.canonical_dict()) + + def content_hash(self) -> str: + return hashlib.sha256(self.canonical_bytes()).hexdigest() + + +@dataclass(frozen=True) +class SignedGenomeArtifact: + genome: ServiceGenome + genome_hash: str + signature_b64: str + public_key_b64: str + + def verify_integrity(self) -> None: + if self.genome_hash != self.genome.content_hash(): + raise ValueError("genome hash does not match genome content") + + +class GenomeSigner: + def __init__(self, private_key: Ed25519PrivateKey) -> None: + self._private_key = private_key + + @classmethod + def generate(cls) -> "GenomeSigner": + return cls(Ed25519PrivateKey.generate()) + + @property + def public_key(self) -> Ed25519PublicKey: + return self._private_key.public_key() + + def sign(self, genome: ServiceGenome) -> SignedGenomeArtifact: + payload = genome.canonical_bytes() + signature = self._private_key.sign(payload) + return SignedGenomeArtifact( + genome=genome, + genome_hash=genome.content_hash(), + signature_b64=base64.b64encode(signature).decode("ascii"), + public_key_b64=base64.b64encode( + self.public_key.public_bytes(encoding=Encoding.Raw, format=PublicFormat.Raw) + ).decode("ascii"), + ) + + @staticmethod + def verify(artifact: SignedGenomeArtifact) -> bool: + artifact.verify_integrity() + public_key = Ed25519PublicKey.from_public_bytes(base64.b64decode(artifact.public_key_b64)) + signature = base64.b64decode(artifact.signature_b64) + public_key.verify(signature, artifact.genome.canonical_bytes()) + return True diff --git a/src/idea187_autopoiesis_verifiable_service/ledger.py b/src/idea187_autopoiesis_verifiable_service/ledger.py new file mode 100644 index 0000000..ae91a92 --- /dev/null +++ b/src/idea187_autopoiesis_verifiable_service/ledger.py @@ -0,0 +1,102 @@ +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timezone +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field + + +def _stable_json(value: Any) -> bytes: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8") + + +class LedgerEntry(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + index: int + timestamp: datetime + actor_did: str + action: str + genome_hash: str + attestation_hash: str + trace_hash: str + resource_snapshot: dict[str, Any] + previous_hash: str + entry_hash: str + + +class TraceDelta(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + stage: str + operation: str + payload: dict[str, Any] = Field(default_factory=dict) + + +class ReplayTrace(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + seed: int + deltas: list[TraceDelta] = Field(default_factory=list) + signatures: list[str] = Field(default_factory=list) + + def digest(self) -> str: + return hashlib.sha256(_stable_json(self.model_dump(mode="json", exclude_none=True))).hexdigest() + + +class ProvenanceLedger: + def __init__(self) -> None: + self._entries: list[LedgerEntry] = [] + + @property + def entries(self) -> tuple[LedgerEntry, ...]: + return tuple(self._entries) + + def append( + self, + *, + actor_did: str, + action: str, + genome_hash: str, + attestation_hash: str, + trace_hash: str, + resource_snapshot: dict[str, Any], + timestamp: datetime | None = None, + ) -> LedgerEntry: + timestamp = timestamp or datetime.now(timezone.utc) + 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(), + "actor_did": actor_did, + "action": action, + "genome_hash": genome_hash, + "attestation_hash": attestation_hash, + "trace_hash": trace_hash, + "resource_snapshot": resource_snapshot, + "previous_hash": previous_hash, + } + entry_hash = hashlib.sha256(_stable_json(payload)).hexdigest() + entry = LedgerEntry(entry_hash=entry_hash, **payload) + self._entries.append(entry) + return entry + + def merkle_root(self) -> str: + if not self._entries: + return hashlib.sha256(b"").hexdigest() + + layer = [bytes.fromhex(entry.entry_hash) for entry in self._entries] + while len(layer) > 1: + if len(layer) % 2 == 1: + layer.append(layer[-1]) + next_layer = [] + for left, right in zip(layer[0::2], layer[1::2], strict=False): + next_layer.append(hashlib.sha256(left + right).digest()) + layer = next_layer + return layer[0].hex() + + def export(self) -> dict[str, Any]: + return {"entries": [entry.model_dump(mode="json") for entry in self._entries], "merkle_root": self.merkle_root()} diff --git a/src/idea187_autopoiesis_verifiable_service/policy.py b/src/idea187_autopoiesis_verifiable_service/policy.py new file mode 100644 index 0000000..7afb1d4 --- /dev/null +++ b/src/idea187_autopoiesis_verifiable_service/policy.py @@ -0,0 +1,136 @@ +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timezone +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from .genome import ServiceGenome + + +def _stable_digest(value: Any) -> str: + payload = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8") + return hashlib.sha256(payload).hexdigest() + + +class PolicyTemplate(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + name: str + max_cpu_millicores: int + max_memory_mib: int + max_bandwidth_mbps: int + max_cost_usd_per_hour: float + allowed_egress_endpoints: list[str] = Field(default_factory=list) + allowed_mutation_operators: list[str] = Field(default_factory=list) + allow_network_egress: bool = False + require_deterministic_tests: bool = True + + @model_validator(mode="after") + def _validate_positive(self) -> "PolicyTemplate": + for field_name in ("max_cpu_millicores", "max_memory_mib", "max_bandwidth_mbps", "max_cost_usd_per_hour"): + if getattr(self, field_name) <= 0: + raise ValueError(f"{field_name} must be positive") + return self + + +class SmokeTestResult(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + name: str + passed: bool + seed: int + input_fingerprint: str + expected_output_hash: str + observed_output_hash: str + + +class SafetyAttestation(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + policy_name: str + approved: bool + score: float + reasons: list[str] = Field(default_factory=list) + smoke_tests: list[SmokeTestResult] = Field(default_factory=list) + created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) + + def digest(self) -> str: + return _stable_digest(self.model_dump(mode="json", exclude_none=True)) + + +class PolicySandbox: + def evaluate(self, genome: ServiceGenome, policy: PolicyTemplate) -> SafetyAttestation: + reasons: list[str] = [] + penalties = 0.0 + + envelope = genome.resource_envelope + if envelope.cpu_millicores > policy.max_cpu_millicores: + reasons.append("cpu budget exceeded") + penalties += 20 + if envelope.memory_mib > policy.max_memory_mib: + reasons.append("memory budget exceeded") + penalties += 20 + if envelope.bandwidth_mbps > policy.max_bandwidth_mbps: + reasons.append("bandwidth budget exceeded") + penalties += 15 + if envelope.cost_usd_per_hour > policy.max_cost_usd_per_hour: + reasons.append("cost budget exceeded") + penalties += 20 + + if not policy.allow_network_egress and genome.capabilities.network: + reasons.append("network egress disabled by policy") + penalties += 20 + + denied = set(genome.safety_invariants.denied_endpoints) + egress = set(genome.capabilities.network) + violations = sorted(denied.intersection(egress)) + if violations: + reasons.append(f"denied endpoint(s) present: {', '.join(violations)}") + penalties += 25 + + if policy.allowed_egress_endpoints: + unauthorized = sorted(endpoint for endpoint in genome.capabilities.network if endpoint not in policy.allowed_egress_endpoints) + if unauthorized: + reasons.append(f"unauthorized egress endpoint(s): {', '.join(unauthorized)}") + penalties += 15 + + if policy.allowed_mutation_operators: + unauthorized_ops = sorted( + op.name for op in genome.mutation_operators if op.name not in policy.allowed_mutation_operators + ) + if unauthorized_ops: + reasons.append(f"mutation operator(s) not allowed: {', '.join(unauthorized_ops)}") + penalties += 10 + + smoke_tests = self.run_smoke_tests(genome) + failed_tests = [result.name for result in smoke_tests if not result.passed] + if genome.safety_invariants.require_deterministic_tests and failed_tests: + reasons.append(f"smoke test failure(s): {', '.join(failed_tests)}") + penalties += 30 + + exposure_headroom = max(0, 100 - genome.safety_invariants.max_exposure) + score = max(0.0, min(100.0, exposure_headroom - penalties + 50.0)) + approved = not reasons + return SafetyAttestation(policy_name=policy.name, approved=approved, score=score, reasons=reasons, smoke_tests=smoke_tests) + + def run_smoke_tests(self, genome: ServiceGenome) -> list[SmokeTestResult]: + genome_hash = genome.content_hash() + results: list[SmokeTestResult] = [] + for contract in genome.test_contracts: + observed = hashlib.sha256( + f"{contract.seed}:{contract.input_fingerprint}".encode("utf-8") + ).hexdigest() + results.append( + SmokeTestResult( + name=contract.name, + passed=observed == contract.expected_output_hash, + seed=contract.seed, + input_fingerprint=contract.input_fingerprint, + expected_output_hash=contract.expected_output_hash, + observed_output_hash=observed, + ) + ) + return results diff --git a/src/idea187_autopoiesis_verifiable_service/replicator.py b/src/idea187_autopoiesis_verifiable_service/replicator.py new file mode 100644 index 0000000..8b69879 --- /dev/null +++ b/src/idea187_autopoiesis_verifiable_service/replicator.py @@ -0,0 +1,76 @@ +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from .genome import ServiceGenome +from .ledger import ReplayTrace, TraceDelta +from .policy import SafetyAttestation + + +class ReplicaWeaveIntent(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + source: str + target: str + intent_kind: str + budget_hint: dict[str, Any] = Field(default_factory=dict) + safety_labels: list[str] = Field(default_factory=list) + + @model_validator(mode="after") + def _validate_identity(self) -> "ReplicaWeaveIntent": + if not self.source.strip() or not self.target.strip(): + raise ValueError("source and target are required") + return self + + +class RolloutStage(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + name: str + cohort_size: int + percentage: int + + +class ReplicationPlan(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + genome_hash: str + approved: bool + stages: list[RolloutStage] = Field(default_factory=list) + trace: ReplayTrace + rationale: list[str] = Field(default_factory=list) + + def digest(self) -> str: + payload = self.model_dump(mode="json", exclude_none=True) + return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8")).hexdigest() + + +class ReplicatorPlanner: + def plan(self, genome: ServiceGenome, attestation: SafetyAttestation) -> ReplicationPlan: + genome_hash = genome.content_hash() + trace_seed = int(genome_hash[:16], 16) + rationale = list(attestation.reasons) + if not attestation.approved: + 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 + 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) diff --git a/test.sh b/test.sh new file mode 100755 index 0000000..a842389 --- /dev/null +++ b/test.sh @@ -0,0 +1,6 @@ +#!/usr/bin/env bash +set -euo pipefail + +python3 -m pip install -e ".[dev]" +python3 -m pytest +python3 -m build diff --git a/tests/test_core.py b/tests/test_core.py new file mode 100644 index 0000000..8917b44 --- /dev/null +++ b/tests/test_core.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +from datetime import datetime, timezone + +from idea187_autopoiesis_verifiable_service import ( + GenomeSigner, + PolicySandbox, + PolicyTemplate, + ProvenanceLedger, + ReplicatorPlanner, + ResourceEnvelope, + SafetyInvariants, + ServiceGenome, + TestContract, +) + + +def build_genome() -> ServiceGenome: + expected = __import__("hashlib").sha256("7:boot".encode("utf-8")).hexdigest() + return ServiceGenome( + name="http-service", + version="1.0.0", + image_ref="ghcr.io/example/http-service:1.0.0", + capabilities={"network": ["https://internal.example"]}, + resource_envelope=ResourceEnvelope( + cpu_millicores=250, + memory_mib=256, + bandwidth_mbps=20, + cost_usd_per_hour=0.05, + 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"}], + test_contracts=[TestContract(name="startup", seed=7, input_fingerprint="boot", expected_output_hash=expected)], + ) + + +def test_genome_sign_and_verify() -> None: + genome = build_genome() + signer = GenomeSigner.generate() + artifact = signer.sign(genome) + + assert artifact.genome_hash == genome.content_hash() + assert GenomeSigner.verify(artifact) is True + + +def test_policy_sandbox_approves_safe_genome() -> 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) + + assert attestation.approved is True + assert attestation.reasons == [] + assert all(result.passed for result in attestation.smoke_tests) + + +def test_ledger_merkle_root_and_replay_trace_are_deterministic() -> None: + genome = build_genome() + signer = GenomeSigner.generate() + artifact = signer.sign(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) + + ledger = ProvenanceLedger() + first = ledger.append( + actor_did="did:key:z6Mkh1", + action="replicate", + genome_hash=artifact.genome_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), + ) + second = ledger.append( + actor_did="did:key:z6Mkh2", + action="promote", + genome_hash=artifact.genome_hash, + attestation_hash=attestation.digest(), + trace_hash=plan.trace.digest(), + resource_snapshot=genome.resource_envelope.model_dump(mode="json"), + timestamp=datetime(2024, 1, 1, 0, 1, tzinfo=timezone.utc), + ) + + assert first.entry_hash != second.entry_hash + assert ledger.merkle_root() + assert plan.approved is True + assert plan.trace.digest() == plan.trace.digest()