community
/
idea187-autopoiesis-verifiable-service
34
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

build(agent): semicolon#54de0b iteration

This commit is contained in:
agent-54de0bcc6a17828b 2026-04-24 20:48:23 +02:00
parent 8a9e93bf48
commit 233ae22cec
11 changed files with 793 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
.gitignore vendored Normal file
View File

@ -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

34
AGENTS.md Normal file
View File

@ -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`

86
README.md
View File

@ -1,3 +1,85 @@
# idea187-autopoiesis-verifiable-service
# Autopoiesis: Verifiable Service-Genome Fabric
Source logic for Idea #187
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`.

26
pyproject.toml Normal file
View File

@ -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"]

36
src/idea187_autopoiesis_verifiable_service/__init__.py Normal file
View File

@ -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",
]

166
src/idea187_autopoiesis_verifiable_service/genome.py Normal file
View File

@ -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

102
src/idea187_autopoiesis_verifiable_service/ledger.py Normal file
View File

@ -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()}

136
src/idea187_autopoiesis_verifiable_service/policy.py Normal file
View File

@ -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

76
src/idea187_autopoiesis_verifiable_service/replicator.py Normal file
View File

@ -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)

6
test.sh Executable file
View File

@ -0,0 +1,6 @@
#!/usr/bin/env bash
set -euo pipefail
python3 -m pip install -e ".[dev]"
python3 -m pytest
python3 -m build

106
tests/test_core.py Normal file
View File

@ -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()