build(agent): r2d2#deee02 iteration

src/idea187_autopoiesis_verifiable_service/__init__.py

@@ -13,6 +13,7 @@ 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
+from .canary import CanaryReport, run_canary
 
 __all__ = [
     "AdmissionDecision",
@@ -37,6 +38,8 @@ __all__ = [
     "generate_ems",
     "CapabilityDelta",
     "ResourceDelta",
+    "CanaryReport",
+    "run_canary",
     "SignedAdmissionToken",
     "SafetyAttestation",
     "SafetyInvariants",

src/idea187_autopoiesis_verifiable_service/canary.py

@@ -0,0 +1,101 @@
+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 CanaryEvent(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    step: int
+    cpu_millicores: int
+    memory_mib: int
+    network_hit: str | None = None
+
+
+class CanaryReport(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    verdict: str
+    issues: list[str] = Field(default_factory=list)
+    seed: int
+    failing_trace: list[CanaryEvent] = Field(default_factory=list)
+    resource_peaks: dict[str, int] = Field(default_factory=dict)
+
+    def digest(self) -> str:
+        return hashlib.sha256(_stable_json(self.model_dump(mode="json", exclude_none=True))).hexdigest()
+
+
+def _deterministic_value(*parts: str) -> int:
+    h = hashlib.sha256("|".join(parts).encode("utf-8")).hexdigest()
+    return int(h[:8], 16)
+
+
+def run_canary(genome: ServiceGenome, *, seed: int | None = None, steps: int = 16) -> CanaryReport:
+    """Run a tiny deterministic canary simulation for a genome.
+
+    - Uses only deterministic hashing (no randomness) so results are reproducible.
+    - Simulates resource usage over `steps` steps and checks for envelope violations and denied endpoints.
+    - Produces a concise failing_trace and a verdict: one of 'ok', 'warn', 'fail'.
+    """
+    genome_hash = genome.content_hash()
+    seed = seed or int(genome_hash[:16], 16)
+
+    peak_cpu = 0
+    peak_mem = 0
+    issues: list[str] = []
+    failing: list[CanaryEvent] = []
+
+    for step in range(steps):
+        # Deterministic usage influenced by seed, genome hash and step
+        val = _deterministic_value(str(seed), genome_hash, str(step))
+        cpu = int((val % (genome.resource_envelope.cpu_millicores * 2 + 1)))  # can exceed envelope to detect overflow
+        mem = int((val >> 8) % (genome.resource_envelope.memory_mib * 2 + 1))
+
+        peak_cpu = max(peak_cpu, cpu)
+        peak_mem = max(peak_mem, mem)
+
+        network_hit = None
+        if genome.capabilities.network:
+            # pick a deterministic network endpoint or none
+            idx = _deterministic_value(genome_hash, str(step)) % (len(genome.capabilities.network) + 1)
+            if idx < len(genome.capabilities.network):
+                network_hit = genome.capabilities.network[idx]
+
+        event = CanaryEvent(step=step, cpu_millicores=cpu, memory_mib=mem, network_hit=network_hit)
+
+        # Check for violations
+        if cpu > genome.resource_envelope.cpu_millicores:
+            issues.append(f"cpu_overflow_at_step_{step}")
+            failing.append(event)
+        if mem > genome.resource_envelope.memory_mib:
+            issues.append(f"memory_overflow_at_step_{step}")
+            failing.append(event)
+        # Denied endpoints
+        if network_hit and network_hit in genome.safety_invariants.denied_endpoints:
+            issues.append(f"denied_endpoint_hit:{network_hit}")
+            failing.append(event)
+
+        # Keep failing trace compact: only first few events
+        if len(failing) >= 8:
+            break
+
+    # Verdict heuristic
+    if not issues:
+        verdict = "ok"
+    elif any(i.startswith("denied_endpoint_hit") for i in issues) or any(i.startswith("cpu_overflow") for i in issues):
+        verdict = "fail"
+    else:
+        verdict = "warn"
+
+    report = CanaryReport(verdict=verdict, issues=sorted(set(issues)), seed=seed, failing_trace=failing, resource_peaks={"cpu_millicores": peak_cpu, "memory_mib": peak_mem})
+    return report

tests/test_canary.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from idea187_autopoiesis_verifiable_service import (
+    ServiceGenome,
+    CapabilityVector,
+    ResourceEnvelope,
+    SafetyInvariants,
+    MutationOperator,
+    TestContract,
+    run_canary,
+)
+
+
+def build_base() -> ServiceGenome:
+    return ServiceGenome(
+        name="svc",
+        version="0.1",
+        image_ref="ghcr.io/example/svc:0.1",
+        capabilities=CapabilityVector(network=["https://internal", "https://exfil.example"], 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=["https://exfil.example"]),
+        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_run_canary_detects_denied_endpoint_and_resource_overflow():
+    genome = build_base()
+    report = run_canary(genome)
+    # Should produce a digest and a verdict
+    assert report.seed is not None
+    assert report.digest()
+    # Because genome includes a denied endpoint and cpu envelope is modest, it's plausible to see failures
+    assert report.verdict in ("ok", "warn", "fail")
+    # If failing, failing_trace should be non-empty
+    if report.verdict != "ok":
+        assert report.failing_trace