AID-H-030.005: Admission Decision Orchestration & Triggered Revalidation
Fuse exact stage outputs into one policy-bound block, review, or allow decision tied to the immutable skill digest. Invalidate it when publisher identity, artifact digest, scanner logic, policy, dependency intelligence, or freshness changes, and hold the skill before further install or execution.
Pillars: infra, app
Lifecycle: validation, operation, improvement
Scope boundary
Owns technical orchestration of exact skill-admission stage outputs, immutable digest binding, block or review or allow decision issuance, freshness invalidation, and triggered revalidation. It does not own generic governance workflows or ticketing.
AID-H-030.003: AID-H-030.005 makes the final skill installation or continued-use decision; AID-H-030.003 supplies the skill-specific behavioral-consistency result. No individual analysis stage may authorize installation.
AID-H-003.001: AID-H-030.005 admits the composite behavior and semantics of an agentic skill; AID-H-003.001 admits the exact third-party package population and bytes consumed by build or deployment. A skill can require both decisions when it includes dependencies or executable packages.
AID-H-021.005: AID-H-030.005 admits an exact skill artifact for installation or continued use; AID-H-021.005 decides whether the complete repository-controlled activation manifest, including hooks, tasks, servers, settings, and already admitted local skills, may activate in this workspace. Skill admission does not trust a workspace, and workspace trust does not admit an unreviewed skill or authorize later tool actions.
Implementation guidance
- AID-H-030.005-G001: Fuse stage results into a policy-driven block/review/allow decision and persist the decision evidence with the skill record.
- AID-H-030.005-G002: Enforce the digest-bound admission decision through an enterprise install broker that accepts only approved skill artifacts from approved sources.
- AID-H-030.005-G003: Invalidate and re-run admission when publisher intelligence, artifact identity, scanner logic, policy, dependency findings, or freshness changes.
AID-H-030.005-G001
Fuse stage results into a policy-driven block/review/allow decision and persist the decision evidence with the skill record.
Concept:
The final admission policy decision point (PDP) consumes signed stage receipts, not caller-supplied booleans or scores. A signature-verified stage matrix defines the exact required stages and the policy/scanner version and freshness required for each one. Every receipt must bind the same skill identity and artifact digest. Missing, stale, malformed, mixed-artifact, FAIL, INSUFFICIENT_DATA, and ERROR evidence can never fall through to ALLOW.
Deterministic receipt fusion
# File: skill_admission/final_pdp.py
from __future__ import annotations
import hashlib
import json
import time
from dataclasses import asdict, dataclass
from typing import Protocol
@dataclass(frozen=True)
class Candidate:
skill_id: str
skill_version: str
artifact_sha256: str
@dataclass(frozen=True)
class StageRequirement:
stage: str
policy_revision: str
scanner_id: str
scanner_version: str
maximum_age_seconds: int
@dataclass(frozen=True)
class AdmissionPolicy:
revision: str
maximum_future_skew_seconds: int
required_stages: tuple[StageRequirement, ...]
@dataclass(frozen=True)
class VerifiedStageReceipt:
receipt_id: str
stage: str
status: str
skill_id: str
skill_version: str
artifact_sha256: str
policy_revision: str
scanner_id: str
scanner_version: str
evidence_sha256: str
issued_at_epoch: int
class VerifiedAdmissionPolicyProvider(Protocol):
def load_verified(self) -> AdmissionPolicy:
"""Verify signature, schema, validity window, and policy revision."""
raise RuntimeError("protocol-only method")
class StageReceiptVerifier(Protocol):
def verify(self, serialized_receipt: bytes) -> VerifiedStageReceipt:
"""Verify the receipt signature and closed schema before returning it."""
raise RuntimeError("protocol-only method")
class DecisionStore(Protocol):
def persist_and_readback(
self, decision_id: str, canonical_record: bytes
) -> dict:
"""Atomically put-if-absent, sign, and read back the durable decision."""
raise RuntimeError("protocol-only method")
class SkillRegistry(Protocol):
def apply_decision_and_readback(
self, candidate: Candidate, decision_receipt: dict
) -> str:
"""Apply the exact durable decision and return a registry receipt."""
raise RuntimeError("protocol-only method")
def _canonical_json(value: object) -> bytes:
return json.dumps(
value, sort_keys=True, separators=(",", ":"), allow_nan=False
).encode("utf-8")
def _validate_policy(policy: AdmissionPolicy) -> dict[str, StageRequirement]:
if not policy.revision or policy.maximum_future_skew_seconds < 0:
raise RuntimeError("invalid_verified_admission_policy")
requirements: dict[str, StageRequirement] = {}
for requirement in policy.required_stages:
if (
not requirement.stage or requirement.stage in requirements
or not requirement.policy_revision
or not requirement.scanner_id or not requirement.scanner_version
or requirement.maximum_age_seconds <= 0
):
raise RuntimeError("invalid_or_duplicate_stage_requirement")
requirements[requirement.stage] = requirement
if not requirements:
raise RuntimeError("empty_stage_matrix")
return requirements
def decide(
candidate: Candidate,
serialized_receipts: tuple[bytes, ...],
policy: AdmissionPolicy,
verifier: StageReceiptVerifier,
now_epoch: int | None = None,
) -> dict:
requirements = _validate_policy(policy)
now = int(time.time()) if now_epoch is None else now_epoch
if (
type(now) is not int or now <= 0
or not candidate.skill_id or not candidate.skill_version
or len(candidate.artifact_sha256) != 64
or set(candidate.artifact_sha256) - set("0123456789abcdef")
):
raise ValueError("invalid candidate binding or decision clock")
receipts: dict[str, VerifiedStageReceipt] = {}
blockers: list[dict[str, str]] = []
reviews: list[dict[str, str]] = []
allowed_statuses = {
"PASS", "REVIEW", "FAIL", "INSUFFICIENT_DATA", "ERROR"
}
for serialized in serialized_receipts:
receipt = verifier.verify(serialized)
if (
not receipt.receipt_id or not receipt.stage
or receipt.stage in receipts
or receipt.status not in allowed_statuses
or type(receipt.issued_at_epoch) is not int
or len(receipt.evidence_sha256) != 64
or set(receipt.evidence_sha256) - set("0123456789abcdef")
):
raise RuntimeError("invalid_or_duplicate_stage_receipt")
if receipt.stage not in requirements:
raise RuntimeError("receipt_for_unrecognized_stage")
receipts[receipt.stage] = receipt
for stage, requirement in requirements.items():
receipt = receipts.get(stage)
if receipt is None:
blockers.append({"stage": stage, "reason": "missing_receipt"})
continue
expected_bindings = {
"skill_id": candidate.skill_id,
"skill_version": candidate.skill_version,
"artifact_sha256": candidate.artifact_sha256,
"policy_revision": requirement.policy_revision,
"scanner_id": requirement.scanner_id,
"scanner_version": requirement.scanner_version,
}
if any(
getattr(receipt, key) != value
for key, value in expected_bindings.items()
):
blockers.append({"stage": stage, "reason": "binding_mismatch"})
continue
if (
receipt.issued_at_epoch > now + policy.maximum_future_skew_seconds
or now - receipt.issued_at_epoch > requirement.maximum_age_seconds
):
blockers.append({"stage": stage, "reason": "stale_or_future_receipt"})
elif receipt.status in {"FAIL", "INSUFFICIENT_DATA", "ERROR"}:
blockers.append({
"stage": stage,
"reason": "stage_" + receipt.status.lower(),
})
elif receipt.status == "REVIEW":
reviews.append({"stage": stage, "reason": "stage_review"})
if blockers:
final_decision = "BLOCK"
elif reviews:
final_decision = "REVIEW"
else:
final_decision = "ALLOW"
ordered_receipts = [
asdict(receipts[stage]) for stage in sorted(receipts)
]
return {
"schema_version": 1,
"candidate": asdict(candidate),
"admission_policy_revision": policy.revision,
"required_stages": sorted(requirements),
"stage_receipts": ordered_receipts,
"blockers": sorted(
blockers, key=lambda item: (item["stage"], item["reason"])
),
"reviews": sorted(
reviews, key=lambda item: (item["stage"], item["reason"])
),
"decision": final_decision,
}
def persist_then_apply(
candidate: Candidate,
serialized_receipts: tuple[bytes, ...],
policy_provider: VerifiedAdmissionPolicyProvider,
receipt_verifier: StageReceiptVerifier,
decision_store: DecisionStore,
registry: SkillRegistry,
) -> dict:
policy = policy_provider.load_verified()
record = decide(candidate, serialized_receipts, policy, receipt_verifier)
canonical_record = _canonical_json(record)
decision_id = hashlib.sha256(canonical_record).hexdigest()
durable = decision_store.persist_and_readback(
decision_id, canonical_record
)
if (
durable.get("decision_id") != decision_id
or durable.get("record_sha256")
!= hashlib.sha256(canonical_record).hexdigest()
or durable.get("durable") is not True
or not isinstance(durable.get("signature"), str)
or not durable["signature"]
):
raise RuntimeError("decision_persistence_readback_failed")
registry_receipt = registry.apply_decision_and_readback(
candidate, durable
)
if not registry_receipt:
raise RuntimeError("registry_decision_readback_failed")
record["decision_id"] = decision_id
record["registry_receipt"] = registry_receipt
if record["decision"] != "ALLOW":
raise RuntimeError("skill_admission_" + record["decision"].lower())
return record
Production implementation
The policy provider and receipt verifier are adapters to the organization's signing and key-management system; they must reject revoked keys, expired profiles, unknown schemas, and ambiguous encodings. The decision store must make the canonical record immutable and idempotent under decision_id, then sign its readback receipt. Only after that durable acknowledgement may the registry transition. The install broker in AID-H-030.005-G002 consumes that exact decision ID and artifact digest; a registry label alone is not admission evidence. Preserve raw stage evidence under each receipt's digest without copying untrusted stage verdicts into policy logic.
Independent verification
Replay the PDP with each required stage missing, duplicated, forged, expired, future-dated, wrong-policy, wrong-scanner, wrong-skill, and mixed-artifact; exercise every status and simultaneous BLOCK/REVIEW combinations to verify deterministic precedence. Inject decision-store timeout, conflicting idempotent writes, bad readback digest/signature, and registry failure. No ALLOW or installable registry state may exist unless all exact required receipts are fresh PASS and the durable decision readback succeeds first.
AID-H-030.005-G002
Enforce the digest-bound admission decision through an enterprise install broker that accepts only approved skill artifacts from approved sources.
Concept:
The enterprise install broker—not a mutable inventory flag—is the installation enforcement point. The client verifies a signed, digest-bound ALLOW token, reads the exact reviewed package from a confined staging root, and uploads those bytes to the broker. It never sends a local path for the broker to dereference. The broker independently re-verifies the token and artifact, enforces one-time decision use and registry policy, installs the exact bytes, and returns a signed receipt whose installed digest is read back before success is reported.
Digest-bound broker client
# File: skill_governance/install_via_broker.py
from __future__ import annotations
import hashlib
import json
import math
import os
import stat
from dataclasses import dataclass
from pathlib import Path
from typing import Protocol
from urllib.parse import urlsplit
import jwt
import requests
@dataclass(frozen=True)
class InstallPolicy:
revision: str
allowed_registries: tuple[str, ...]
staging_root: str
broker_url: str
request_timeout_seconds: float
maximum_artifact_bytes: int
maximum_receipt_bytes: int
ca_bundle: str
client_cert: tuple[str, str]
approval_issuer: str
approval_audience: str
approval_algorithm: str
approval_public_key: str
receipt_issuer: str
receipt_audience: str
receipt_algorithm: str
receipt_public_key: str
clock_skew_seconds: int
@dataclass(frozen=True)
class InstallManifest:
skill_id: str
skill_version: str
source_registry: str
package_path: str
artifact_sha256: str
decision_id: str
class VerifiedInstallPolicyProvider(Protocol):
def load_verified(self) -> InstallPolicy:
"""Verify the install profile signature, schema, expiry, and revision."""
raise RuntimeError("protocol-only method")
class ApprovalTokenProvider(Protocol):
def load_for_install(self, decision_id: str) -> str:
"""Return the protected ALLOW token; the broker enforces idempotent one-time use."""
raise RuntimeError("protocol-only method")
class InstallReceiptStore(Protocol):
def persist_and_readback(self, receipt_jwt: str) -> str:
"""Persist the broker receipt and return its durable record ID."""
raise RuntimeError("protocol-only method")
def _strict_json_object(raw: bytes) -> dict:
def unique(pairs: list[tuple[str, object]]) -> dict:
result = {}
for key, value in pairs:
if key in result:
raise ValueError(f"duplicate JSON key: {key}")
result[key] = value
return result
value = json.loads(raw.decode("utf-8"), object_pairs_hook=unique)
if not isinstance(value, dict):
raise ValueError("install manifest must be an object")
return value
def load_manifest(path_text: str) -> InstallManifest:
value = _strict_json_object(Path(path_text).read_bytes())
expected = {
"skill_id", "skill_version", "source_registry", "package_path",
"artifact_sha256", "decision_id",
}
if set(value) != expected or any(
not isinstance(value[key], str) or not value[key]
for key in expected
):
raise ValueError("install manifest has an invalid closed schema")
return InstallManifest(**value)
def _https_url(value: str) -> str:
parts = urlsplit(value)
if (
parts.scheme != "https" or not parts.hostname
or parts.username is not None or parts.password is not None
or parts.fragment
):
raise ValueError("expected an HTTPS URL without credentials or fragment")
return value
def _validate_policy(policy: InstallPolicy) -> None:
algorithms = {"RS256", "ES256", "EdDSA"}
if (
not policy.revision or not policy.allowed_registries
or len(policy.allowed_registries) != len(set(policy.allowed_registries))
or policy.approval_algorithm not in algorithms
or policy.receipt_algorithm not in algorithms
or not math.isfinite(policy.request_timeout_seconds)
or policy.request_timeout_seconds <= 0
or policy.maximum_artifact_bytes <= 0
or policy.maximum_receipt_bytes <= 0
or policy.clock_skew_seconds < 0
):
raise RuntimeError("invalid verified install policy")
_https_url(policy.broker_url)
for registry in policy.allowed_registries:
_https_url(registry)
def _read_staged_artifact(
path_text: str, staging_root_text: str, maximum_bytes: int
) -> bytes:
root = Path(staging_root_text).resolve(strict=True)
path = Path(path_text)
resolved = path.resolve(strict=True)
try:
resolved.relative_to(root)
except ValueError as error:
raise RuntimeError("artifact_outside_staging_root") from error
before = path.lstat()
if (
stat.S_ISLNK(before.st_mode) or not stat.S_ISREG(before.st_mode)
or before.st_size > maximum_bytes
):
raise RuntimeError("artifact_is_not_a_bounded_regular_file")
flags = os.O_RDONLY | getattr(os, "O_BINARY", 0)
if hasattr(os, "O_NOFOLLOW"):
flags |= os.O_NOFOLLOW
descriptor = os.open(path, flags)
try:
opened = os.fstat(descriptor)
if (
opened.st_dev != before.st_dev or opened.st_ino != before.st_ino
or opened.st_size != before.st_size
or opened.st_mtime_ns != before.st_mtime_ns
):
raise RuntimeError("artifact_changed_before_read")
chunks = []
total = 0
while True:
block = os.read(
descriptor, min(1024 * 1024, maximum_bytes + 1 - total)
)
if not block:
break
chunks.append(block)
total += len(block)
if total > maximum_bytes:
raise RuntimeError("artifact_exceeds_policy_cap")
after = os.fstat(descriptor)
if (
after.st_size != opened.st_size
or after.st_mtime_ns != opened.st_mtime_ns
):
raise RuntimeError("artifact_changed_during_read")
finally:
os.close(descriptor)
return b"".join(chunks)
def _decode_approval(
token: str, policy: InstallPolicy, manifest: InstallManifest
) -> dict:
claims = jwt.decode(
token,
policy.approval_public_key,
algorithms=[policy.approval_algorithm],
audience=policy.approval_audience,
issuer=policy.approval_issuer,
leeway=policy.clock_skew_seconds,
options={"require": [
"iss", "aud", "sub", "exp", "iat", "jti", "decision",
"decision_id", "skill_id", "skill_version", "artifact_sha256",
"source_registry", "install_policy_revision",
]},
)
expected = {
"sub": manifest.skill_id,
"skill_id": manifest.skill_id,
"skill_version": manifest.skill_version,
"artifact_sha256": manifest.artifact_sha256,
"source_registry": manifest.source_registry,
"decision_id": manifest.decision_id,
"decision": "ALLOW",
"install_policy_revision": policy.revision,
}
if any(claims.get(key) != value for key, value in expected.items()):
raise RuntimeError("approval_token_binding_mismatch")
if not isinstance(claims.get("jti"), str) or not claims["jti"]:
raise RuntimeError("approval_token_missing_jti")
return claims
def _read_bounded_text(
response: requests.Response, maximum_bytes: int
) -> str:
response.raise_for_status()
raw = bytearray()
for chunk in response.iter_content(chunk_size=min(65536, maximum_bytes + 1)):
if chunk:
raw.extend(chunk)
if len(raw) > maximum_bytes:
raise RuntimeError("broker receipt exceeds policy cap")
return bytes(raw).decode("ascii")
def install(
manifest_path: str,
policy_provider: VerifiedInstallPolicyProvider,
token_provider: ApprovalTokenProvider,
receipt_store: InstallReceiptStore,
) -> dict:
policy = policy_provider.load_verified()
_validate_policy(policy)
manifest = load_manifest(manifest_path)
if manifest.source_registry not in set(policy.allowed_registries):
raise RuntimeError("source_registry_not_allowed")
if (
len(manifest.artifact_sha256) != 64
or set(manifest.artifact_sha256) - set("0123456789abcdef")
or len(manifest.decision_id) != 64
or set(manifest.decision_id) - set("0123456789abcdef")
):
raise RuntimeError("invalid_manifest_digest_or_decision_id")
artifact = _read_staged_artifact(
manifest.package_path,
policy.staging_root,
policy.maximum_artifact_bytes,
)
actual_sha256 = hashlib.sha256(artifact).hexdigest()
if actual_sha256 != manifest.artifact_sha256:
raise RuntimeError("staged_artifact_digest_mismatch")
approval_token = token_provider.load_for_install(manifest.decision_id)
approval = _decode_approval(approval_token, policy, manifest)
request_binding = {
"skill_id": manifest.skill_id,
"skill_version": manifest.skill_version,
"source_registry": manifest.source_registry,
"artifact_sha256": actual_sha256,
"decision_id": manifest.decision_id,
"approval_jti": approval["jti"],
"install_policy_revision": policy.revision,
}
binding_bytes = json.dumps(
request_binding, sort_keys=True, separators=(",", ":")
).encode("utf-8")
request_id = hashlib.sha256(
b"aidefend.install.v1|" + binding_bytes
).hexdigest()
broker_manifest = {
"schema_version": 1,
"request_id": request_id,
**request_binding,
}
canonical_manifest = json.dumps(
broker_manifest, sort_keys=True, separators=(",", ":")
).encode("utf-8")
manifest_sha256 = hashlib.sha256(canonical_manifest).hexdigest()
with requests.post(
policy.broker_url,
data={"manifest": canonical_manifest.decode("utf-8")},
files={
"artifact": (
manifest.skill_id + ".zip",
artifact,
"application/zip",
),
"approval_token": (
"approval.jwt",
approval_token.encode("ascii"),
"application/jwt",
),
},
timeout=policy.request_timeout_seconds,
allow_redirects=False,
verify=policy.ca_bundle,
cert=policy.client_cert,
stream=True,
) as response:
receipt_jwt = _read_bounded_text(
response, policy.maximum_receipt_bytes
)
receipt = jwt.decode(
receipt_jwt,
policy.receipt_public_key,
algorithms=[policy.receipt_algorithm],
audience=policy.receipt_audience,
issuer=policy.receipt_issuer,
leeway=policy.clock_skew_seconds,
options={"require": [
"iss", "aud", "sub", "exp", "iat", "jti", "status",
"request_id", "decision_id", "approval_jti", "skill_id",
"skill_version", "artifact_sha256", "installed_artifact_sha256",
"manifest_sha256", "install_policy_revision",
]},
)
if not isinstance(receipt.get("jti"), str) or not receipt["jti"]:
raise RuntimeError("broker_receipt_missing_jti")
expected_receipt = {
"sub": manifest.skill_id,
"status": "INSTALLED",
"request_id": request_id,
"decision_id": manifest.decision_id,
"approval_jti": approval["jti"],
"skill_id": manifest.skill_id,
"skill_version": manifest.skill_version,
"artifact_sha256": actual_sha256,
"installed_artifact_sha256": actual_sha256,
"manifest_sha256": manifest_sha256,
"install_policy_revision": policy.revision,
}
if any(
receipt.get(key) != value for key, value in expected_receipt.items()
):
raise RuntimeError("broker_receipt_binding_mismatch")
durable_record_id = receipt_store.persist_and_readback(receipt_jwt)
if not durable_record_id:
raise RuntimeError("install_receipt_persistence_failed")
return {
"status": "INSTALLED",
"request_id": request_id,
"decision_id": manifest.decision_id,
"artifact_sha256": actual_sha256,
"broker_receipt_jti": receipt["jti"],
"durable_record_id": durable_record_id,
}
Production implementation
The broker must independently stream-hash the uploaded bytes, verify the ALLOW token against the authoritative decision store, enforce token jti and decision ID as single-use while idempotently returning the same signed receipt for the same request ID, verify the approved source registry, install only from its quarantined copy, read back the installed package digest, and sign the receipt. Keep IdP, key rotation, mTLS issuance, broker storage, and registry access as existing service adapters. Separately enforce egress and endpoint policy so developer workstations and runtimes cannot reach public registries or invoke a local installer around the broker; denied attempts are detection evidence, not proof that this broker request succeeded.
Independent verification
Exercise expired, replayed, wrong-issuer/audience/algorithm, REVIEW/BLOCK, wrong-policy, wrong-registry, mixed-skill/version, swapped-path, symlinked, concurrently changed, oversized, and digest-mismatched packages; broker redirects and TLS failures; response substitution; installed-byte mismatch; and receipt-store failure. Confirm the broker never dereferences the client path, the decision is consumed once, and installation is successful only when the signed receipt binds the exact uploaded and installed digest, request, decision, token, policy revision, and skill identity.
AID-H-030.005-G003
Invalidate and re-run admission when publisher intelligence, artifact identity, scanner logic, policy, dependency findings, or freshness changes.
Concept:
Admission is valid only for the exact artifact and trust inputs recorded at the last decision. Compare every installed skill in the authoritative registry with a signature-verified current trigger snapshot and versioned re-scan policy. Missing timestamps or trigger subjects, changed installed bytes, publisher compromise/status changes, dependency-finding changes, scanner/profile changes, policy changes, and freshness expiry all invalidate the prior admission; missing data never defaults to fresh.
Idempotent full-population scheduler
# File: skill_admission/rescan_scheduler.py
from __future__ import annotations
import hashlib
import json
import time
from dataclasses import dataclass
from typing import Protocol
@dataclass(frozen=True)
class RescanPolicy:
revision: str
admission_policy_revision: str
scanner_id: str
scanner_version: str
scanner_profile_sha256: str
maximum_age_seconds: int
maximum_future_skew_seconds: int
additional_hold_reasons: tuple[str, ...]
@dataclass(frozen=True)
class PublisherStatus:
publisher: str
status_sha256: str
compromised: bool
@dataclass(frozen=True)
class DependencyStatus:
artifact_sha256: str
findings_sha256: str
@dataclass(frozen=True)
class TriggerSnapshot:
snapshot_id: str
snapshot_sha256: str
publisher_intel_revision: str
dependency_intel_revision: str
publishers: tuple[PublisherStatus, ...]
dependencies: tuple[DependencyStatus, ...]
@dataclass(frozen=True)
class InstalledSkill:
skill_id: str
skill_version: str
publisher: str
installed_artifact_sha256: str
admitted_artifact_sha256: str
admission_policy_revision: str
scanner_id: str
scanner_version: str
scanner_profile_sha256: str
publisher_status_sha256: str
dependency_findings_sha256: str
last_scan_at_epoch: int
prior_decision_id: str
@dataclass(frozen=True)
class RegistrySnapshot:
snapshot_id: str
population_complete: bool
records: tuple[InstalledSkill, ...]
class VerifiedRescanPolicyProvider(Protocol):
def load_verified(self) -> RescanPolicy:
raise RuntimeError("protocol-only method")
class VerifiedTriggerProvider(Protocol):
def load_verified(self) -> TriggerSnapshot:
"""Verify snapshot signature, schema, revisions, and digest."""
raise RuntimeError("protocol-only method")
class SkillRegistry(Protocol):
def snapshot_installed(self) -> RegistrySnapshot:
"""Return the complete authoritative installed-skill population."""
raise RuntimeError("protocol-only method")
def mark_pending_and_readback(
self, skill: InstalledSkill, job_key: str,
reasons: tuple[str, ...], hold: bool,
) -> str:
"""Preserve prior evidence, mark pending/hold, and return a receipt."""
raise RuntimeError("protocol-only method")
class RescanQueue(Protocol):
def enqueue_once(self, job_key: str, canonical_job: bytes) -> dict:
"""Idempotently enqueue by job key and return a durable acknowledgement."""
raise RuntimeError("protocol-only method")
def _is_sha256(value: str) -> bool:
return (
isinstance(value, str) and len(value) == 64
and not (set(value) - set("0123456789abcdef"))
)
def _canonical_json(value: object) -> bytes:
return json.dumps(
value, sort_keys=True, separators=(",", ":"), allow_nan=False
).encode("utf-8")
def _index_triggers(
snapshot: TriggerSnapshot,
) -> tuple[dict[str, PublisherStatus], dict[str, DependencyStatus]]:
if (
not snapshot.snapshot_id or not _is_sha256(snapshot.snapshot_sha256)
or not snapshot.publisher_intel_revision
or not snapshot.dependency_intel_revision
):
raise RuntimeError("invalid_verified_trigger_snapshot")
publishers: dict[str, PublisherStatus] = {}
for item in snapshot.publishers:
if (
not item.publisher or item.publisher in publishers
or not _is_sha256(item.status_sha256)
):
raise RuntimeError("invalid_or_duplicate_publisher_trigger")
publishers[item.publisher] = item
dependencies: dict[str, DependencyStatus] = {}
for item in snapshot.dependencies:
if (
not _is_sha256(item.artifact_sha256)
or item.artifact_sha256 in dependencies
or not _is_sha256(item.findings_sha256)
):
raise RuntimeError("invalid_or_duplicate_dependency_trigger")
dependencies[item.artifact_sha256] = item
return publishers, dependencies
def rescan_reasons(
skill: InstalledSkill,
policy: RescanPolicy,
publisher: PublisherStatus | None,
dependency: DependencyStatus | None,
now_epoch: int,
) -> tuple[str, ...]:
reasons: set[str] = set()
digests = (
skill.installed_artifact_sha256,
skill.admitted_artifact_sha256,
skill.scanner_profile_sha256,
skill.publisher_status_sha256,
skill.dependency_findings_sha256,
skill.prior_decision_id,
)
if (
not skill.skill_id or not skill.skill_version or not skill.publisher
or any(not _is_sha256(item) for item in digests)
):
reasons.add("incomplete_or_invalid_registry_record")
if skill.installed_artifact_sha256 != skill.admitted_artifact_sha256:
reasons.add("artifact_identity_changed")
if skill.admission_policy_revision != policy.admission_policy_revision:
reasons.add("admission_policy_changed")
if (
skill.scanner_id != policy.scanner_id
or skill.scanner_version != policy.scanner_version
or skill.scanner_profile_sha256 != policy.scanner_profile_sha256
):
reasons.add("scanner_or_profile_changed")
if publisher is None or dependency is None:
reasons.add("trigger_snapshot_missing_subject")
else:
if publisher.compromised:
reasons.add("publisher_compromised")
if publisher.status_sha256 != skill.publisher_status_sha256:
reasons.add("publisher_status_changed")
if dependency.findings_sha256 != skill.dependency_findings_sha256:
reasons.add("dependency_findings_changed")
if (
not isinstance(skill.last_scan_at_epoch, int)
or skill.last_scan_at_epoch <= 0
or skill.last_scan_at_epoch
> now_epoch + policy.maximum_future_skew_seconds
):
reasons.add("invalid_or_future_last_scan_time")
elif now_epoch - skill.last_scan_at_epoch > policy.maximum_age_seconds:
reasons.add("freshness_expired")
return tuple(sorted(reasons))
def schedule_rescans(
policy_provider: VerifiedRescanPolicyProvider,
trigger_provider: VerifiedTriggerProvider,
registry: SkillRegistry,
queue: RescanQueue,
now_epoch: int | None = None,
) -> dict:
policy = policy_provider.load_verified()
if (
not policy.revision or not policy.admission_policy_revision
or not policy.scanner_id or not policy.scanner_version
or not _is_sha256(policy.scanner_profile_sha256)
or policy.maximum_age_seconds <= 0
or policy.maximum_future_skew_seconds < 0
or len(policy.additional_hold_reasons)
!= len(set(policy.additional_hold_reasons))
):
raise RuntimeError("invalid_verified_rescan_policy")
triggers = trigger_provider.load_verified()
publishers, dependencies = _index_triggers(triggers)
inventory = registry.snapshot_installed()
if not inventory.snapshot_id or not inventory.population_complete:
raise RuntimeError("installed_skill_population_incomplete")
identities = [
(item.skill_id, item.skill_version) for item in inventory.records
]
if len(identities) != len(set(identities)):
raise RuntimeError("duplicate_installed_skill_identity")
now = int(time.time()) if now_epoch is None else now_epoch
if type(now) is not int or now <= 0:
raise ValueError("invalid scheduler clock")
mandatory_hold = {
"artifact_identity_changed", "publisher_compromised",
"trigger_snapshot_missing_subject",
"incomplete_or_invalid_registry_record",
"invalid_or_future_last_scan_time",
}
policy_hold = mandatory_hold.union(policy.additional_hold_reasons)
queued: list[dict] = []
unchanged = 0
for skill in inventory.records:
publisher = publishers.get(skill.publisher)
dependency = dependencies.get(skill.installed_artifact_sha256)
reasons = rescan_reasons(
skill, policy, publisher, dependency, now
)
if not reasons:
unchanged += 1
continue
job_core = {
"schema_version": 1,
"skill_id": skill.skill_id,
"skill_version": skill.skill_version,
"installed_artifact_sha256": skill.installed_artifact_sha256,
"prior_decision_id": skill.prior_decision_id,
"rescan_policy_revision": policy.revision,
"publisher_status_sha256": (
publisher.status_sha256 if publisher else "MISSING"
),
"publisher_compromised": (
publisher.compromised if publisher else None
),
"dependency_findings_sha256": (
dependency.findings_sha256 if dependency else "MISSING"
),
"reasons": list(reasons),
}
canonical_job = _canonical_json(job_core)
job_key = hashlib.sha256(canonical_job).hexdigest()
hold = bool(set(reasons).intersection(policy_hold))
registry_receipt = registry.mark_pending_and_readback(
skill, job_key, reasons, hold
)
if not registry_receipt:
raise RuntimeError("registry_pending_readback_failed")
acknowledgement = queue.enqueue_once(job_key, canonical_job)
if (
acknowledgement.get("job_key") != job_key
or acknowledgement.get("durable") is not True
):
raise RuntimeError("rescan_queue_acknowledgement_failed")
queued.append({
"skill_id": skill.skill_id,
"skill_version": skill.skill_version,
"job_key": job_key,
"hold": hold,
"reasons": list(reasons),
"registry_receipt": registry_receipt,
})
return {
"schema_version": 1,
"registry_snapshot_id": inventory.snapshot_id,
"trigger_snapshot_id": triggers.snapshot_id,
"rescan_policy_revision": policy.revision,
"installed_population_count": len(inventory.records),
"queued_count": len(queued),
"unchanged_count": unchanged,
"queued": queued,
}
Production implementation
The trigger provider must publish a signed status digest for every installed publisher and artifact, including a signed no-findings value; absence is therefore detectable. Registry mark_pending_and_readback happens before queue acknowledgement so a queue outage cannot leave stale evidence presented as current. Derive the job key from the skill's exact subject-specific trigger values, so unrelated global-intelligence revisions do not duplicate work; retry that same key until enqueue_once acknowledges it. The worker reruns the complete signed stage matrix, and AID-H-030.005-G001 persists a new final decision before the registry atomically replaces current evidence; retain the prior decision for audit and keep mandatory-hold skills unavailable meanwhile.
Independent verification
Reconcile the scheduler report with the complete registry snapshot and replay unchanged, compromised publisher, publisher-status change, installed/admitted digest mismatch, new dependency findings, scanner/profile and policy changes, exact freshness boundary, missing/future timestamp, missing trigger subject, duplicate registry row, duplicate trigger, queue retry, queue outage, and registry readback failure. Confirm unchanged skills are not queued, each changed tuple has one stable job key, and no missing or failed dependency becomes fresh or silently usable.
Open-source tools
- Open Policy Agent (OPA)
- OpenTelemetry
- Apache Airflow
- Temporal
- Argo Workflows (scheduled re-scan orchestration)
Commercial / hosted tools
- Splunk
- Microsoft Sentinel
- Datadog
Threat mappings
MITRE ATLAS
- AML.T0011 User Execution
- AML.T0110 AI Agent Tool Poisoning
- AML.T0115 Publish Poisoned AI Artifacts
- AML.T0110.001 AI Agent Tool Poisoning: Implementation
- AML.T0110.000 AI Agent Tool Poisoning: Definition and Instructions
- AML.T0010.005 AI Supply Chain Compromise: AI Agent Tool
- AML.T0074 Masquerading
- AML.T0011.002 User Execution: Poisoned AI Agent Tool
- AML.T0010 AI Supply Chain Compromise (continuous re-scan detects newly risky skill artifacts already present in the environment)
- AML.T0115.002 Publish Poisoned AI Artifacts: AI Agent Tools (policy orchestration keeps previously admitted poisoned skills from remaining trusted indefinitely)
- AML.T0109 AI Supply Chain Rug Pull (continuous re-scan and policy reevaluation catch trusted artifacts that become malicious later)
MAESTRO
- Supply Chain Attacks (Cross-Layer) (continuous re-scan governance responds to newly discovered supply-chain compromise after initial admission)
- Compromised Agent Registry (L7) (registry trust changes can trigger policy reevaluation and re-scan)
OWASP LLM Top 10 2026
- N/A
OWASP ML Top 10 2023
- ML06:2023 AI Supply Chain Attacks (ongoing rescanning revisits skill artifacts when intelligence or policy changes)
OWASP Top 10 for Agentic Applications 2026
- ASI04:2026 Agentic Supply Chain Vulnerabilities (continuous re-scan ensures admission is not a one-time trust decision)
NIST Adversarial Machine Learning 2025
- N/A (policy orchestration and rescanning are governance controls rather than direct AML threat mappings)
Cisco Integrated AI Security and Safety Framework
- AISubtech-9.3.1 Malicious Package / Tool Injection
- AISubtech-9.3.3 Dependency Replacement / Rug Pull
- AITech-9.3 Dependency / Plugin Compromise
Google Secure AI Framework 2.0 - Risks
- IIC: Insecure Integrated Component (continuous reevaluation maintains trust in integrated skill components)
Databricks AI Security Framework 3.0
- Agents - Tools MCP Server 13.21: Supply Chain Attacks (re-scan governance revisits installed skills when supply-chain intelligence changes)