feat(c5): bind source policy to epoch provenance

active-runtime/app/aios_task_admission_bootstrap.py

@@ -144,6 +144,8 @@ class OwnerDelegationEpoch:
     revocation_epoch: int
     policy_digest: str
     schema: str = SCHEMA_EPOCH
+    production_scope: str | None = None
+    destructive_policy: str | None = None
     # Provenance retained by the consumer.  These fields are populated by the
     # signed-directory provider for v2 (and left at their v1-compatible
     # defaults when reading legacy epochs); they are never used to translate or

active-runtime/app/aios_task_admission_live_adapters.py

@@ -478,6 +478,7 @@ class TaskAdmissionPrincipalProvider:
 class TaskAdmissionRuntimeBundle:
     handlers: Mapping[str, Any]
     principal_verifier: TaskAdmissionPrincipalProvider
+    service: Any | None = None
 
 
 class TaskAdmissionHostError(RuntimeError):
@@ -767,6 +768,14 @@ class SignedEpochDirectoryProvider:
             revocation_epoch=int(value["revocation_epoch"]),
             policy_digest=str(value["policy_digest"]),
             schema=str(value["schema"]),
+            production_scope=(
+                str(value["production_scope"])
+                if schema == SCHEMA_EPOCH_V2 else None
+            ),
+            destructive_policy=(
+                str(value["destructive_policy"])
+                if schema == SCHEMA_EPOCH_V2 else None
+            ),
             status=str(value.get("status", "")),
             signature_namespace=str(value.get("signature_namespace", "")),
             signature_key_id=str(value.get("signature_key_id", "")),
@@ -986,6 +995,7 @@ def _build_task_admission_runtime(
             task_port=task_port,
             epoch_id=epoch.epoch_id,
         ),
+        service=service,
     )
 
 

active-runtime/app/governed_source_publication_policy.py

@@ -46,6 +46,13 @@ _POLICY_FIELDS = frozenset(
 )
 _REQUEST_FIELDS = frozenset(SourcePublicationPolicyRequest.__dataclass_fields__)
 
+# The typed Owner epoch producer uses the configured policy's production_scope.
+# The installed C5 source-only policy contract uses ``none`` for production
+# impact; keep this exact value bound to the C5 policy rather than accepting an
+# arbitrary non-empty scope from a raw signer.
+SOURCE_PUBLICATION_EPOCH_PRODUCTION_SCOPE = "none"
+SOURCE_PUBLICATION_EPOCH_DESTRUCTIVE_POLICY = "forbidden"
+
 
 @dataclass(frozen=True, slots=True)
 class TaskAdmissionPolicyProjection:
@@ -207,6 +214,29 @@ class SourcePublicationPolicyVerifier:
         return dict(artifact.payload)
 
 
+def require_source_publication_epoch_scope(epoch: Any, policy_payload: Mapping[str, Any]) -> None:
+    """Require v2 epoch provenance to match the signed C5 policy exactly."""
+    schema = getattr(epoch, "schema", None)
+    if schema is None and isinstance(epoch, Mapping):
+        schema = epoch.get("schema")
+    if schema != "aios.owner-delegation-epoch.v2":
+        raise SingleHostAuthorityError("SOURCE_PUBLICATION_EPOCH_SCHEMA_REQUIRED")
+
+    def field(name: str) -> Any:
+        if isinstance(epoch, Mapping):
+            return epoch.get(name)
+        return getattr(epoch, name, None)
+
+    if (
+        field("production_scope") != SOURCE_PUBLICATION_EPOCH_PRODUCTION_SCOPE
+        or field("destructive_policy") != SOURCE_PUBLICATION_EPOCH_DESTRUCTIVE_POLICY
+        or field("policy_digest") != policy_payload.get("policy_digest")
+        or field("repository") != policy_payload.get("repository")
+        or field("trust_domain") != policy_payload.get("trust_domain")
+    ):
+        raise SingleHostAuthorityError("SOURCE_PUBLICATION_EPOCH_SCOPE_MISMATCH")
+
+
 class ScopedSourcePublicationPolicyAdmission:
     """Rechecks a policy for every exact request before scoped admission."""
 
@@ -305,6 +335,14 @@ class ScopedSourcePublicationPolicyAdmission:
         request = self._request_for_raw(envelope_raw)
         if not self.verifier.verify(self.artifact, request):
             raise SingleHostAuthorityError("SOURCE_PUBLICATION_POLICY_DENIED")
+        epoch = self.bootstrap._verify_epoch(
+            epoch_id=str(raw.get("delegation_epoch_id") or ""),
+            envelope=envelope,
+            now=int(self.verifier.clock()),
+        )
+        require_source_publication_epoch_scope(
+            epoch, self.verifier._validate_payload(self.artifact)
+        )
         return self.bootstrap.authorize_and_admit(raw)
 
     def status(self, raw: Mapping[str, Any] | str) -> Any:
@@ -323,11 +361,14 @@ class ScopedSourcePublicationPolicyAdmission:
             epoch_id = grant.parent_epoch_id if grant is not None else ""
         if not epoch_id or not hasattr(self.bootstrap, "_verify_epoch"):
             raise SingleHostAuthorityError("SOURCE_PUBLICATION_STATUS_EPOCH_REQUIRED")
-        self.bootstrap._verify_epoch(
+        epoch = self.bootstrap._verify_epoch(
             epoch_id=epoch_id,
             envelope=envelope,
             now=int(self.verifier.clock()),
         )
+        require_source_publication_epoch_scope(
+            epoch, self.verifier._validate_payload(self.artifact)
+        )
         if hasattr(self.bootstrap, "status"):
             return self.bootstrap.status(envelope.envelope_id)
         return envelope_raw

tests/test_governed_source_publication_policy.py

@@ -22,6 +22,7 @@ from aios_single_host_authority import (
 from governed_source_publication_policy import (
     ScopedSourcePublicationPolicyAdmission,
     SourcePublicationPolicyVerifier,
+    require_source_publication_epoch_scope,
 )
 from aios_task_admission_bootstrap import TaskAdmissionPolicy
 
@@ -145,6 +146,77 @@ class SourcePublicationPolicyTests(unittest.TestCase):
         )
         self.assertFalse(verifier.verify(self.artifact, self.request))
 
+    def test_epoch_scope_requires_v2_and_exact_c5_provenance(self) -> None:
+        matching = SimpleNamespace(
+            schema="aios.owner-delegation-epoch.v2",
+            production_scope="none", destructive_policy="forbidden",
+            policy_digest=self.artifact.payload["policy_digest"],
+            repository=self.artifact.payload["repository"],
+            trust_domain=self.artifact.payload["trust_domain"],
+        )
+        require_source_publication_epoch_scope(matching, self.artifact.payload)
+        with self.assertRaisesRegex(SingleHostAuthorityError, "SCHEMA_REQUIRED"):
+            require_source_publication_epoch_scope(
+                SimpleNamespace(schema="aios.owner-delegation-epoch.v1"),
+                self.artifact.payload,
+            )
+        for field, value in (("production_scope", "staging"), ("destructive_policy", "rollback-only")):
+            changed = dict(matching.__dict__, **{field: value})
+            with self.assertRaisesRegex(SingleHostAuthorityError, "SCOPE_MISMATCH"):
+                require_source_publication_epoch_scope(
+                    SimpleNamespace(**changed), self.artifact.payload,
+                )
+
+    def test_admission_and_status_reject_v1_epoch_provenance(self) -> None:
+        artifact = self.artifact
+        request = self.request
+
+        class Bootstrap:
+            def __init__(self):
+                self.policy = TaskAdmissionPolicy(
+                    policy_digest=artifact.payload["policy_digest"],
+                    trust_domain=artifact.payload["trust_domain"],
+                )
+                self.envelope = SimpleNamespace(as_dict=lambda: {
+                    "repository": request.repository,
+                    "authority_base_oid": request.authority_base_oid,
+                    "target_branch": request.target_branch,
+                    "generation": request.generation,
+                    "execution_digest": request.execution_digest,
+                    "risk": request.risk,
+                    "requested_capabilities": list(artifact.payload["capabilities"]),
+                    "file_allowlist": list(request.exact_files),
+                    "policy_digest": artifact.payload["policy_digest"],
+                    "trust_domain": artifact.payload["trust_domain"],
+                    "task_id": "task-c5",
+                })
+
+            def _load_envelope(self, _envelope_id):
+                return self.envelope
+
+            def _verify_epoch(self, **_kwargs):
+                return SimpleNamespace(
+                    schema="aios.owner-delegation-epoch.v1",
+                )
+
+            def authorize_and_admit(self, _raw):
+                return {"admitted": True}
+
+            def status(self, _envelope_id):
+                return {"status": "QUEUED"}
+
+        bootstrap = Bootstrap()
+        scoped = ScopedSourcePublicationPolicyAdmission(
+            self.verifier, bootstrap=bootstrap, artifact=artifact,
+            request=request, expected_agent_id=request.target_agent,
+        )
+        raw = {"envelope_id": "env-c5", "authorization_kind": "delegation_epoch",
+               "delegation_epoch_id": "epoch-c5"}
+        with self.assertRaisesRegex(SingleHostAuthorityError, "SCHEMA_REQUIRED"):
+            scoped.authorize_and_admit(raw)
+        with self.assertRaisesRegex(SingleHostAuthorityError, "SCHEMA_REQUIRED"):
+            scoped.status({"envelope_id": "env-c5", "delegation_epoch_id": "epoch-c5"})
+
     def test_bootstrap_wrapper_rechecks_each_call_and_blocks_owner_bypass(self) -> None:
         class Bootstrap:
             def __init__(self) -> None: