build(agent): c3po#b883b4 iteration

AGENTS.md

@@ -8,6 +8,7 @@ This repository contains a Python reference implementation for the Autopoiesis s
 
 - `genome.py`: genome schema, canonical hashing, Ed25519 signing, signed artifact verification.
 - `policy.py`: policy templates, deterministic smoke-test evaluation, safety attestation.
+- `admission.py`: signed capability tokens and deterministic admission decisions.
 - `ledger.py`: append-only provenance ledger, Merkle root computation, replay trace models.
 - `replicator.py`: deterministic replication planner and lightweight ReplicaWeave intent model.
 

README.md

@@ -7,6 +7,7 @@ It focuses on the trust substrate needed for safe self-replication across edge a
 - `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.
+- `AdmissionTokenIssuer` and `AdmissionGate` sign and verify lightweight capability tokens for constrained edges.
 - `ProvenanceLedger` records append-only replication events with a Merkle root and chain verification for tamper evidence.
 - `ReplicatorPlanner` turns an approved genome into a deterministic rollout trace.
 - `ReplayHarness` replays and verifies a rollout trace against the genome and attestation that produced it.
@@ -20,6 +21,7 @@ This repository ships a production-shaped core for the autonomy fabric:
 - content addressing via canonical JSON hashing
 - signed artifact creation and verification
 - deterministic smoke-test evaluation
+- signed admission tokens for budgeted edge installs
 - provenance logging with Merkle aggregation
 - replayable replication traces
 - a small test suite and buildable Python package

src/idea187_autopoiesis_verifiable_service/__init__.py

@@ -1,3 +1,4 @@
+from .admission import AdmissionDecision, AdmissionGate, AdmissionTokenClaims, AdmissionTokenIssuer, SignedAdmissionToken
 from .genome import (
     CapabilityVector,
     GenomeSigner,
@@ -13,6 +14,10 @@ from .policy import PolicySandbox, PolicyTemplate, SafetyAttestation, SmokeTestR
 from .replicator import ReplicaWeaveIntent, ReplayHarness, ReplicatorPlanner, ReplicationPlan, RolloutStage
 
 __all__ = [
+    "AdmissionDecision",
+    "AdmissionGate",
+    "AdmissionTokenClaims",
+    "AdmissionTokenIssuer",
     "CapabilityVector",
     "GenomeSigner",
     "LedgerEntry",
@@ -27,6 +32,7 @@ __all__ = [
     "ResourceEnvelope",
     "ReplayTrace",
     "RolloutStage",
+    "SignedAdmissionToken",
     "SafetyAttestation",
     "SafetyInvariants",
     "ServiceGenome",

src/idea187_autopoiesis_verifiable_service/admission.py

@@ -0,0 +1,176 @@
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+from dataclasses import dataclass
+from datetime import datetime, timezone
+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
+
+from .genome import ResourceEnvelope, ServiceGenome
+
+
+def _canonical_json(value: Any) -> bytes:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8")
+
+
+class AdmissionTokenClaims(BaseModel):
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    issuer_did: str
+    subject_did: str
+    audience: str
+    genome_hash: str
+    allowed_intents: list[str] = Field(default_factory=list)
+    safety_labels: list[str] = Field(default_factory=list)
+    resource_budget: ResourceEnvelope
+    issued_at: datetime
+    not_before: datetime
+    expires_at: datetime
+    nonce: str
+
+    @model_validator(mode="after")
+    def _validate_timestamps(self) -> "AdmissionTokenClaims":
+        if not self.issuer_did.strip():
+            raise ValueError("issuer_did cannot be empty")
+        if not self.subject_did.strip():
+            raise ValueError("subject_did cannot be empty")
+        if not self.audience.strip():
+            raise ValueError("audience cannot be empty")
+        if self.not_before > self.expires_at:
+            raise ValueError("not_before must be before expires_at")
+        if self.issued_at > self.expires_at:
+            raise ValueError("issued_at must be before expires_at")
+        return self
+
+
+@dataclass(frozen=True)
+class SignedAdmissionToken:
+    claims: AdmissionTokenClaims
+    signature_b64: str
+    public_key_b64: str
+
+    def canonical_payload(self) -> bytes:
+        return _canonical_json(self.claims.model_dump(mode="json", exclude_none=True))
+
+    def digest(self) -> str:
+        payload = {
+            "claims": self.claims.model_dump(mode="json", exclude_none=True),
+            "public_key_b64": self.public_key_b64,
+            "signature_b64": self.signature_b64,
+        }
+        return hashlib.sha256(_canonical_json(payload)).hexdigest()
+
+
+@dataclass(frozen=True)
+class AdmissionDecision:
+    approved: bool
+    reasons: tuple[str, ...]
+    token_digest: str
+
+    def digest(self) -> str:
+        return hashlib.sha256(
+            _canonical_json({"approved": self.approved, "reasons": list(self.reasons), "token_digest": self.token_digest})
+        ).hexdigest()
+
+
+class AdmissionTokenIssuer:
+    def __init__(self, private_key: Ed25519PrivateKey) -> None:
+        self._private_key = private_key
+
+    @classmethod
+    def generate(cls) -> "AdmissionTokenIssuer":
+        return cls(Ed25519PrivateKey.generate())
+
+    @property
+    def public_key(self) -> Ed25519PublicKey:
+        return self._private_key.public_key()
+
+    def issue(self, claims: AdmissionTokenClaims) -> SignedAdmissionToken:
+        signature = self._private_key.sign(_canonical_json(claims.model_dump(mode="json", exclude_none=True)))
+        return SignedAdmissionToken(
+            claims=claims,
+            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"),
+        )
+
+
+class AdmissionGate:
+    def authorize(
+        self,
+        genome: ServiceGenome,
+        token: SignedAdmissionToken,
+        *,
+        intent: Any | None = None,
+        at: datetime | None = None,
+    ) -> AdmissionDecision:
+        reasons: list[str] = []
+        now = at or datetime.now(timezone.utc)
+
+        try:
+            self.verify_signature(token)
+        except ValueError as exc:
+            reasons.append(str(exc))
+            return AdmissionDecision(approved=False, reasons=tuple(reasons), token_digest=token.digest())
+
+        claims = token.claims
+        token_digest = token.digest()
+
+        if claims.genome_hash != genome.content_hash():
+            reasons.append("genome hash does not match admission token")
+        if now < claims.not_before:
+            reasons.append("admission token is not yet valid")
+        if now > claims.expires_at:
+            reasons.append("admission token has expired")
+
+        envelope = genome.resource_envelope
+        budget = claims.resource_budget
+        if envelope.cpu_millicores > budget.cpu_millicores:
+            reasons.append("cpu budget exceeds admission token")
+        if envelope.memory_mib > budget.memory_mib:
+            reasons.append("memory budget exceeds admission token")
+        if envelope.bandwidth_mbps > budget.bandwidth_mbps:
+            reasons.append("bandwidth budget exceeds admission token")
+        if envelope.cost_usd_per_hour > budget.cost_usd_per_hour:
+            reasons.append("cost budget exceeds admission token")
+        if envelope.max_replicas > budget.max_replicas:
+            reasons.append("replica budget exceeds admission token")
+
+        if intent is not None:
+            intent_kind = getattr(intent, "intent_kind", None)
+            safety_labels = list(getattr(intent, "safety_labels", []) or [])
+            budget_hint = getattr(intent, "budget_hint", {}) or {}
+
+            if claims.allowed_intents and intent_kind not in claims.allowed_intents:
+                reasons.append(f"intent kind not allowed: {intent_kind}")
+
+            missing_labels = sorted(set(safety_labels) - set(claims.safety_labels))
+            if missing_labels:
+                reasons.append(f"missing safety label(s): {', '.join(missing_labels)}")
+
+            budget_checks = (
+                ("cpu_millicores", "cpu budget hint exceeds admission token"),
+                ("memory_mib", "memory budget hint exceeds admission token"),
+                ("bandwidth_mbps", "bandwidth budget hint exceeds admission token"),
+                ("cost_usd_per_hour", "cost budget hint exceeds admission token"),
+                ("max_replicas", "replica budget hint exceeds admission token"),
+            )
+            for field_name, reason in budget_checks:
+                hinted = budget_hint.get(field_name)
+                if hinted is not None and hinted > getattr(budget, field_name):
+                    reasons.append(reason)
+
+        return AdmissionDecision(approved=not reasons, reasons=tuple(reasons), token_digest=token_digest)
+
+    @staticmethod
+    def verify_signature(token: SignedAdmissionToken) -> bool:
+        public_key = Ed25519PublicKey.from_public_bytes(base64.b64decode(token.public_key_b64))
+        signature = base64.b64decode(token.signature_b64)
+        public_key.verify(signature, token.canonical_payload())
+        return True

src/idea187_autopoiesis_verifiable_service/replicator.py

@@ -7,6 +7,7 @@ from typing import Any
 
 from pydantic import BaseModel, ConfigDict, Field, model_validator
 
+from .admission import AdmissionDecision, AdmissionGate, SignedAdmissionToken
 from .genome import ServiceGenome
 from .ledger import ReplayTrace, TraceDelta
 from .policy import SafetyAttestation
@@ -44,6 +45,7 @@ class ReplicationPlan(BaseModel):
     stages: list[RolloutStage] = Field(default_factory=list)
     trace: ReplayTrace
     rationale: list[str] = Field(default_factory=list)
+    admission_decision: AdmissionDecision | None = None
 
     def digest(self) -> str:
         payload = self.model_dump(mode="json", exclude_none=True)
@@ -62,25 +64,89 @@ def _rollout_stages(max_replicas: int, score: float) -> list[RolloutStage]:
 
 
 class ReplicatorPlanner:
-    def plan(self, genome: ServiceGenome, attestation: SafetyAttestation) -> ReplicationPlan:
+    def plan(
+        self,
+        genome: ServiceGenome,
+        attestation: SafetyAttestation,
+        *,
+        intent: ReplicaWeaveIntent | None = None,
+        admission_token: SignedAdmissionToken | None = None,
+    ) -> ReplicationPlan:
         genome_hash = genome.content_hash()
         trace_seed = int(genome_hash[:16], 16)
         rationale = list(attestation.reasons)
+        admission_decision = None
+
+        if admission_token is not None:
+            admission_decision = AdmissionGate().authorize(genome, admission_token, intent=intent)
+            if not admission_decision.approved:
+                rationale.extend(admission_decision.reasons)
+                trace = ReplayTrace(
+                    seed=trace_seed,
+                    deltas=[
+                        TraceDelta(
+                            stage="blocked",
+                            operation="deny",
+                            payload={
+                                "policy": attestation.policy_name,
+                                "admission_token": admission_decision.token_digest,
+                                "reasons": list(admission_decision.reasons),
+                            },
+                        )
+                    ],
+                )
+                return ReplicationPlan(
+                    genome_hash=genome_hash,
+                    approved=False,
+                    stages=[],
+                    trace=trace,
+                    rationale=rationale,
+                    admission_decision=admission_decision,
+                )
+
         if not attestation.approved:
-            trace = ReplayTrace(seed=trace_seed, deltas=[TraceDelta(stage="blocked", operation="deny", payload={"policy": attestation.policy_name})])
+            trace = ReplayTrace(
-            return ReplicationPlan(genome_hash=genome_hash, approved=False, stages=[], trace=trace, rationale=rationale)
+                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,
+                admission_decision=admission_decision,
+            )
 
         stages = _rollout_stages(genome.resource_envelope.max_replicas, attestation.score)
         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()])
+        signatures = [genome_hash, attestation.digest()]
-        return ReplicationPlan(genome_hash=genome_hash, approved=True, stages=stages, trace=trace, rationale=rationale)
+        if admission_decision is not None:
+            signatures.append(admission_decision.digest())
+        trace = ReplayTrace(seed=trace_seed, deltas=deltas, signatures=signatures)
+        return ReplicationPlan(
+            genome_hash=genome_hash,
+            approved=True,
+            stages=stages,
+            trace=trace,
+            rationale=rationale,
+            admission_decision=admission_decision,
+        )
 
 
 class ReplayHarness:
-    def verify(self, genome: ServiceGenome, attestation: SafetyAttestation, plan: ReplicationPlan) -> ReplayTrace:
+    def verify(
+        self,
+        genome: ServiceGenome,
+        attestation: SafetyAttestation,
+        plan: ReplicationPlan,
+        *,
+        intent: ReplicaWeaveIntent | None = None,
+        admission_token: SignedAdmissionToken | None = None,
+    ) -> ReplayTrace:
         genome_hash = genome.content_hash()
         if plan.genome_hash != genome_hash:
             raise ValueError("plan genome hash does not match genome content")
@@ -91,8 +157,20 @@ class ReplayHarness:
         if plan.approved != attestation.approved:
             raise ValueError("plan approval does not match attestation")
 
+        expected_admission = None
+        if admission_token is not None:
+            expected_admission = AdmissionGate().authorize(genome, admission_token, intent=intent)
+            if plan.admission_decision != expected_admission:
+                raise ValueError("plan admission decision is not reproducible from the token and intent")
+            if expected_admission.approved and expected_admission.digest() not in plan.trace.signatures:
+                raise ValueError("trace signatures do not include the admission decision")
+
         if not plan.approved:
-            if plan.trace.deltas != [TraceDelta(stage="blocked", operation="deny", payload={"policy": attestation.policy_name})]:
+            expected_payload: dict[str, object] = {"policy": attestation.policy_name}
+            if expected_admission is not None and not expected_admission.approved:
+                expected_payload["admission_token"] = expected_admission.token_digest
+                expected_payload["reasons"] = list(expected_admission.reasons)
+            if plan.trace.deltas != [TraceDelta(stage="blocked", operation="deny", payload=expected_payload)]:
                 raise ValueError("blocked trace does not match attestation")
             return plan.trace
 
@@ -106,6 +184,9 @@ class ReplayHarness:
             raise ValueError("plan stages are not reproducible from the genome and attestation")
         if plan.trace.deltas != expected_deltas:
             raise ValueError("trace deltas are not reproducible from the rollout stages")
-        if plan.trace.signatures != [genome_hash, attestation.digest()]:
+        expected_signatures = [genome_hash, attestation.digest()]
+        if expected_admission is not None:
+            expected_signatures.append(expected_admission.digest())
+        if plan.trace.signatures != expected_signatures:
             raise ValueError("trace signatures do not match the expected audit chain")
         return plan.trace

tests/test_core.py

@@ -3,6 +3,9 @@ from __future__ import annotations
 from datetime import datetime, timezone
 
 from idea187_autopoiesis_verifiable_service import (
+    AdmissionGate,
+    AdmissionTokenClaims,
+    AdmissionTokenIssuer,
     GenomeSigner,
     CapabilityVector,
     PolicySandbox,
@@ -12,6 +15,7 @@ from idea187_autopoiesis_verifiable_service import (
     ReplicatorPlanner,
     MutationOperator,
     ResourceEnvelope,
+    ReplicaWeaveIntent,
     SafetyInvariants,
     ServiceGenome,
     TestContract,
@@ -154,3 +158,94 @@ def test_audit_verification_rejects_tampering() -> None:
         assert "seed" in str(exc)
     else:
         raise AssertionError("expected trace verification to fail")
+
+
+def test_admission_token_binds_genome_budget_and_intent() -> None:
+    genome = build_genome()
+    attestation = PolicySandbox().evaluate(
+        genome,
+        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"],
+        ),
+    )
+    issuer = AdmissionTokenIssuer.generate()
+    claims = AdmissionTokenClaims(
+        issuer_did="did:key:zissuer",
+        subject_did="did:key:zsubject",
+        audience="edge-fleet-a",
+        genome_hash=genome.content_hash(),
+        allowed_intents=["fork"],
+        safety_labels=["read-only", "deterministic"],
+        resource_budget=ResourceEnvelope(
+            cpu_millicores=500,
+            memory_mib=512,
+            bandwidth_mbps=100,
+            cost_usd_per_hour=1.0,
+            max_replicas=3,
+        ),
+        issued_at=datetime(2024, 1, 1, tzinfo=timezone.utc),
+        not_before=datetime(2024, 1, 1, tzinfo=timezone.utc),
+        expires_at=datetime(2030, 1, 2, tzinfo=timezone.utc),
+        nonce="nonce-1",
+    )
+    token = issuer.issue(claims)
+    intent = ReplicaWeaveIntent(
+        source="catopt",
+        target="edge-01",
+        intent_kind="fork",
+        budget_hint={"cpu_millicores": 250, "max_replicas": 2},
+        safety_labels=["read-only"],
+    )
+
+    decision = AdmissionGate().authorize(genome, token, intent=intent, at=datetime(2024, 1, 1, 12, tzinfo=timezone.utc))
+    assert decision.approved is True
+
+    plan = ReplicatorPlanner().plan(genome, attestation, intent=intent, admission_token=token)
+    assert plan.approved is True
+    assert plan.admission_decision == decision
+    assert decision.digest() in plan.trace.signatures
+    assert ReplayHarness().verify(genome, attestation, plan, intent=intent, admission_token=token) == plan.trace
+
+
+def test_admission_token_rejects_label_mismatch() -> None:
+    genome = build_genome()
+    issuer = AdmissionTokenIssuer.generate()
+    token = issuer.issue(
+        AdmissionTokenClaims(
+            issuer_did="did:key:zissuer",
+            subject_did="did:key:zsubject",
+            audience="edge-fleet-a",
+            genome_hash=genome.content_hash(),
+            allowed_intents=["fork"],
+            safety_labels=["read-only"],
+            resource_budget=ResourceEnvelope(
+                cpu_millicores=500,
+                memory_mib=512,
+                bandwidth_mbps=100,
+                cost_usd_per_hour=1.0,
+                max_replicas=3,
+            ),
+            issued_at=datetime(2024, 1, 1, tzinfo=timezone.utc),
+            not_before=datetime(2024, 1, 1, tzinfo=timezone.utc),
+            expires_at=datetime(2030, 1, 2, tzinfo=timezone.utc),
+            nonce="nonce-2",
+        )
+    )
+    intent = ReplicaWeaveIntent(
+        source="catopt",
+        target="edge-01",
+        intent_kind="fork",
+        budget_hint={"cpu_millicores": 250},
+        safety_labels=["read-only", "requires-attestation"],
+    )
+
+    decision = AdmissionGate().authorize(genome, token, intent=intent, at=datetime(2024, 1, 1, 12, tzinfo=timezone.utc))
+    assert decision.approved is False
+    assert any("missing safety label" in reason for reason in decision.reasons)