Scope boundary

Owns discovery, identity, version, provenance, owner, installation state, permissions, dependencies, and declared-versus-observed reconciliation for agent skills, instruction assets, rules, plugins, and workflow artifacts. It emits lifecycle findings but does not admit, install, enforce, retire, contain, or evict them.

AID-D-005.004: AID-M-001.003 inventories which skill or instruction artifact is deployed and supplies its stable identity and version; AID-D-005.004 records what an agent and those artifacts did during a session. Inventory reconciliation does not replace event capture, and session logs do not prove the complete installed-skill population.

Implementation guidance

• AID-M-001.003-G001: Establish a central skill inventory with mandatory registration for all installed skills.
• AID-M-001.003-G002: Import an authoritative approval or exception decision into the inventory and bind it to the exact skill digest without making the inventory the approval or admission authority.
• AID-M-001.003-G003: Discover installed instruction packages and reconcile their complete digests, ownership, usage, and scan coverage against central inventory.
• AID-M-001.003-G004: Ingest trusted ownership, vulnerability, publisher, and incident events; update inventory state; and emit signed, control-routed lifecycle triggers without claiming the downstream action completed.

AID-M-001.003-G001

Concept:

Every installed skill must be registered in a central inventory, regardless of agent platform. The inventory records: skill name, version, content hash (SHA-256), source registry (e.g., ClawHub, skills.sh, internal mirror), publisher/signer identity, install date, installer's enterprise identity (employee ID or service account), current owner / business owner, permission manifest summary (if present), last security scan status and date, and approval/exception status.

Automated Registration via CLI Hook

from __future__ import annotations

import math as _aidefend_math
import os as _aidefend_os

# Admission injects these required values from one signature-verified, versioned runtime profile.
def _aidefend_required(name):
    value = _aidefend_os.environ.get(name)
    if value is None or not value.strip():
        raise RuntimeError(f"required runtime-profile field is absent: {name}")
    return value.strip()


def _aidefend_positive_float(name):
    try:
        value = float(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not numeric: {name}") from error
    if not _aidefend_math.isfinite(value) or value <= 0:
        raise RuntimeError(f"runtime-profile field must be finite and positive: {name}")
    return value


def _aidefend_positive_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 1:
        raise RuntimeError(f"runtime-profile field must be positive: {name}")
    return value


def _aidefend_nonnegative_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 0:
        raise RuntimeError(f"runtime-profile field cannot be negative: {name}")
    return value


def _aidefend_fraction(name):
    value = _aidefend_positive_float(name)
    if value > 1:
        raise RuntimeError(f"runtime-profile field must be in (0, 1]: {name}")
    return value


RUNTIME_PROFILE_VERSION = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_VERSION")
RUNTIME_PROFILE_SHA256 = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_SHA256").lower()
if (len(RUNTIME_PROFILE_SHA256) != 64
        or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")):
    raise RuntimeError("runtime-profile digest must be lowercase SHA-256")
SKILL_INVENTORY_REQUEST_TIMEOUT_SECONDS = _aidefend_positive_float("M001003_SKILL_INVENTORY_REQUEST_TIMEOUT_SECONDS")
SKILL_INVENTORY_MAX_RESPONSE_BYTES = _aidefend_positive_int("M001003_SKILL_INVENTORY_MAX_RESPONSE_BYTES")

# File: skill_governance/install_hook.py

import datetime
import hashlib
import json
import os
import stat
import unicodedata
from pathlib import Path
from urllib.parse import urlsplit

import requests


def _reject_duplicate_keys(pairs):
    result = {}
    for key, item in pairs:
        if key in result:
            raise ValueError(f"duplicate JSON key: {key}")
        result[key] = item
    return result


def _reject_nonfinite(token):
    raise ValueError(f"non-finite JSON number: {token}")


def _https_origin(name: str) -> str:
    raw = _aidefend_required(name)
    try:
        parsed = urlsplit(raw)
        parsed.port
    except ValueError as error:
        raise RuntimeError("skill-inventory origin is invalid") from error
    if (
        parsed.scheme.lower() != "https"
        or not parsed.hostname
        or parsed.username is not None
        or parsed.password is not None
        or parsed.path not in {"", "/"}
        or parsed.query
        or parsed.fragment
    ):
        raise RuntimeError("skill-inventory URL must be an exact HTTPS origin")
    return raw.rstrip("/")


def _stable_regular_bytes(path: Path) -> tuple[bytes, os.stat_result]:
    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(path, flags)
    with os.fdopen(descriptor, "rb") as handle:
        before = os.fstat(handle.fileno())
        if not stat.S_ISREG(before.st_mode) or before.st_nlink != 1:
            raise ValueError(f"skill member is not one unambiguous regular file: {path}")
        value = handle.read()
        after = os.fstat(handle.fileno())
    identity = lambda item: (
        item.st_dev, item.st_ino, item.st_size,
        item.st_mtime_ns, item.st_ctime_ns,
    )
    if identity(before) != identity(after) or len(value) != before.st_size:
        raise ValueError(f"skill member changed during capture: {path}")
    return value, after


def _canonical_skill_package(
    package_root: Path,
    entry_relative_path: str,
) -> tuple[str, list[dict]]:
    if package_root.is_symlink() or not package_root.is_dir():
        raise ValueError("skill package root must be a non-symlink directory")
    root = package_root.resolve(strict=True)
    normalized_entry = unicodedata.normalize(
        "NFC", Path(entry_relative_path).as_posix()
    )
    if (
        Path(entry_relative_path).is_absolute()
        or normalized_entry in {"", "."}
        or normalized_entry.startswith("../")
        or "/../" in normalized_entry
    ):
        raise ValueError("skill entry path escapes the package root")

    members: list[dict] = []
    normalized_paths: set[str] = set()
    directory_state: dict[str, tuple[int, int, int, int]] = {}
    for current, directory_names, file_names in os.walk(root, followlinks=False):
        current_path = Path(current)
        current_status = os.stat(current_path, follow_symlinks=False)
        directory_state[str(current_path)] = (
            current_status.st_dev, current_status.st_ino,
            current_status.st_mtime_ns, current_status.st_ctime_ns,
        )
        for name in directory_names:
            child = current_path / name
            status = os.stat(child, follow_symlinks=False)
            if stat.S_ISLNK(status.st_mode) or not stat.S_ISDIR(status.st_mode):
                raise ValueError(f"unsafe directory member: {child}")
        for name in file_names:
            child = current_path / name
            status = os.stat(child, follow_symlinks=False)
            if stat.S_ISLNK(status.st_mode) or not stat.S_ISREG(status.st_mode):
                raise ValueError(f"unsafe skill member: {child}")
            resolved = child.resolve(strict=True)
            if not resolved.is_relative_to(root):
                raise ValueError(f"skill member escapes package root: {child}")
            relative = unicodedata.normalize("NFC", resolved.relative_to(root).as_posix())
            collision_key = relative.casefold()
            if collision_key in normalized_paths:
                raise ValueError(f"duplicate normalized skill path: {relative}")
            normalized_paths.add(collision_key)
            value, stable_status = _stable_regular_bytes(resolved)
            members.append({
                "path": relative,
                "size": stable_status.st_size,
                "sha256": hashlib.sha256(value).hexdigest(),
            })

    for directory, expected in directory_state.items():
        current = os.stat(directory, follow_symlinks=False)
        observed = (
            current.st_dev, current.st_ino,
            current.st_mtime_ns, current.st_ctime_ns,
        )
        if observed != expected:
            raise ValueError(f"skill package changed during enumeration: {directory}")
    members.sort(key=lambda item: item["path"].encode("utf-8"))
    if not members or normalized_entry not in {item["path"] for item in members}:
        raise ValueError("skill entry file is absent from the captured package")
    manifest_bytes = json.dumps(
        {
            "schema_version": "aidefend.skill-package-manifest.v1",
            "entry_path": normalized_entry,
            "members": members,
        },
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
    ).encode("utf-8")
    return "sha256:" + hashlib.sha256(manifest_bytes).hexdigest(), members


def _read_bounded_json(response) -> object:
    try:
        status = response.status_code
        if isinstance(status, bool) or not isinstance(status, int) or not 200 <= status < 300:
            raise ValueError("skill-inventory HTTP status is not successful")
        content_type = response.headers.get("Content-Type")
        if (
            not isinstance(content_type, str)
            or content_type.split(";", 1)[0].strip().lower() != "application/json"
        ):
            raise ValueError("skill-inventory response media type is not application/json")
        encoding = response.headers.get("Content-Encoding")
        if encoding is not None and encoding.strip().lower() not in {"", "identity"}:
            raise ValueError("skill-inventory response encoding is unsupported")
        declared = response.headers.get("Content-Length")
        expected = None
        if declared is not None:
            text = declared.strip()
            if not text or any(character not in "0123456789" for character in text):
                raise ValueError("skill-inventory Content-Length is invalid")
            expected = int(text)
            if expected > SKILL_INVENTORY_MAX_RESPONSE_BYTES:
                raise ValueError("skill-inventory response exceeds the signed byte bound")
        body = bytearray()
        for chunk in response.iter_content(chunk_size=min(64 * 1024, SKILL_INVENTORY_MAX_RESPONSE_BYTES)):
            if not isinstance(chunk, bytes):
                raise ValueError("skill-inventory response yielded non-byte content")
            body.extend(chunk)
            if len(body) > SKILL_INVENTORY_MAX_RESPONSE_BYTES:
                raise ValueError("skill-inventory response exceeds the signed byte bound")
        if expected is not None and len(body) != expected:
            raise ValueError("skill-inventory response length differs from Content-Length")
        return json.loads(
            body.decode("utf-8"),
            object_pairs_hook=_reject_duplicate_keys,
            parse_constant=_reject_nonfinite,
        )
    finally:
        response.close()


INVENTORY_ORIGIN = _https_origin("M001003_SKILL_INVENTORY_ORIGIN")
INVENTORY_CA = _aidefend_required("M001003_SKILL_INVENTORY_CA_BUNDLE")
INVENTORY_CERT = _aidefend_required("M001003_SKILL_INVENTORY_CLIENT_CERT")
INVENTORY_KEY = _aidefend_required("M001003_SKILL_INVENTORY_CLIENT_KEY")


def register_skill(
    skill_name: str,
    version: str,
    package_root: str,
    entry_relative_path: str,
    installer: str,
    source: str,
) -> dict:
    content_hash, package_members = _canonical_skill_package(
        Path(package_root), entry_relative_path
    )
    payload = {
        "name": skill_name,
        "version": version,
        "content_hash": content_hash,
        "artifact_manifest_schema": "aidefend.skill-package-manifest.v1",
        "artifact_member_count": len(package_members),
        "entry_path": unicodedata.normalize("NFC", Path(entry_relative_path).as_posix()),
        "source_registry": source,
        "install_date": datetime.datetime.now(datetime.timezone.utc).isoformat(),
        "installer_identity": installer,
        "current_owner": installer,
        "business_owner": None,
        "scan_status": "pending",
        "approval_status": "pending_review",
        "runtime_profile_version": RUNTIME_PROFILE_VERSION,
        "runtime_profile_sha256": RUNTIME_PROFILE_SHA256,
    }
    try:
        with requests.Session() as session:
            with session.post(
                INVENTORY_ORIGIN + "/api/v1/skills",
                json=payload,
                timeout=SKILL_INVENTORY_REQUEST_TIMEOUT_SECONDS,
                verify=INVENTORY_CA,
                cert=(INVENTORY_CERT, INVENTORY_KEY),
                headers={"Accept": "application/json", "Accept-Encoding": "identity"},
                allow_redirects=False,
                stream=True,
            ) as response:
                receipt = _read_bounded_json(response)
    except (requests.RequestException, TypeError, ValueError) as error:
        raise RuntimeError("skill-inventory registration failed closed") from error
    required = {
        "schema_version", "skill_id", "version", "content_hash",
        "artifact_member_count", "entry_path", "inventory_version",
    }
    if (
        not isinstance(receipt, dict)
        or set(receipt) != required
        or receipt["schema_version"] != "aidefend.skill-inventory-receipt.v1"
        or not isinstance(receipt["skill_id"], str) or not receipt["skill_id"]
        or receipt["version"] != version
        or receipt["content_hash"] != content_hash
        or receipt["artifact_member_count"] != len(package_members)
        or receipt["entry_path"] != payload["entry_path"]
        or not isinstance(receipt["inventory_version"], str) or not receipt["inventory_version"]
    ):
        raise RuntimeError("skill-inventory receipt is invalid or bound to different bytes")
    return receipt

Action: Pass the package root and its entry path from each install/update hook. The digest is over the canonical complete package tree—scripts, assets, dependencies, and metadata—not merely the entry file. Reject path escape, symbolic or hard links, special files, normalized-path collisions, and mutation during capture. For platforms that install one immutable bundle, treat that bundle as the package root's sole captured member. Brownfield reconciliation must compare this complete digest and member count against the central inventory.

AID-M-001.003-G002

Purpose and ownership

Import a decision issued by the skill-approval authority; do not let the inventory create approval. The decision must bind one complete package digest from AID-M-001.003-G001, its allowed scope, issuer, policy and authority generations, status, validity window, and the inventory revision being changed. Runtime admission remains owned by AID-H-030.005.

The launcher must verify a versioned runtime profile against its admission-pinned organizational trust root before injecting the command deadline, profile version, and canonical profile digest used below. Environment variables carry that already-verified profile into the process; they are not an independent policy authority.

Decision record

{
  "schema_version": "aidefend.skill-decision.v1",
  "decision_id": "SKILL-2026-0412",
  "issuer": "spiffe://company.example/security/skill-approval",
  "asset_digest": "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
  "scope_digest": "sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
  "status": "APPROVED",
  "policy_version": "skill-policy-18",
  "authority_generation": "skill-approvers-2026-08-28T00:00:00Z",
  "inventory_revision": "skills-004291",
  "issued_at": "2026-08-28T01:00:00Z",
  "expires_at": "2026-09-28T01:00:00Z",
  "revoked": false
}

Verify stable bytes and the inventory readback

from __future__ import annotations

import argparse
import hashlib
import json
import math
import os
import re
import stat
import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path

DIGEST = re.compile(r"^sha256:[a-f0-9]{64}$")
DECISION_FIELDS = {
    "schema_version", "decision_id", "issuer", "asset_digest",
    "scope_digest", "status", "policy_version", "authority_generation",
    "inventory_revision", "issued_at", "expires_at", "revoked",
}
RECEIPT_FIELDS = {
    "schema_version", "decision_id", "asset_digest", "recorded_status",
    "authority_generation", "inventory_revision_before",
    "inventory_revision_after", "record_sha256",
}
RUNTIME_PROFILE_VERSION = os.environ["AIDEFEND_RUNTIME_PROFILE_VERSION"].strip()
RUNTIME_PROFILE_SHA256 = os.environ["AIDEFEND_RUNTIME_PROFILE_SHA256"].strip()
COMMAND_TIMEOUT_SECONDS = float(
    os.environ["SKILL_DECISION_VERIFY_TIMEOUT_SECONDS"]
)
if (
    not RUNTIME_PROFILE_VERSION
    or re.fullmatch(r"[0-9a-f]{64}", RUNTIME_PROFILE_SHA256) is None
    or not math.isfinite(COMMAND_TIMEOUT_SECONDS)
    or COMMAND_TIMEOUT_SECONDS <= 0
):
    raise RuntimeError("verified skill-decision runtime profile is invalid")


def regular_bytes(path: Path) -> bytes:
    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(path, flags)
    with os.fdopen(descriptor, "rb") as handle:
        before = os.fstat(handle.fileno())
        if not stat.S_ISREG(before.st_mode) or before.st_size == 0:
            raise ValueError(f"missing, empty, or unsafe input: {path}")
        raw = handle.read()
        after = os.fstat(handle.fileno())
    identity = lambda value: (
        value.st_dev, value.st_ino, value.st_size,
        value.st_mtime_ns, value.st_ctime_ns,
    )
    if identity(before) != identity(after):
        raise ValueError(f"input changed during capture: {path}")
    return raw


def strict_json(raw: bytes, label: str) -> dict:
    def unique(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError(f"duplicate JSON key in {label}: {key}")
            value[key] = item
        return value
    parsed = json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique,
        parse_constant=lambda token: (_ for _ in ()).throw(ValueError(token)),
    )
    if not isinstance(parsed, dict):
        raise ValueError(f"{label} must be a JSON object")
    return parsed


def when(value: object) -> datetime:
    if not isinstance(value, str):
        raise ValueError("timestamp must be a string")
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        raise ValueError("timestamp lacks timezone")
    return parsed.astimezone(timezone.utc)


def verified_decision(path: Path, bundle: Path, key: Path) -> tuple[dict, bytes]:
    payload, signature, public_key = (
        regular_bytes(path), regular_bytes(bundle), regular_bytes(key)
    )
    with tempfile.TemporaryDirectory(prefix="skill-decision-") as directory:
        root = Path(directory)
        snapshots = [root / "decision.json", root / "decision.sig", root / "authority.pub"]
        for target, raw in zip(snapshots, (payload, signature, public_key), strict=True):
            target.write_bytes(raw)
            target.chmod(0o400)
        subprocess.run(
            ["cosign", "verify-blob", "--key", str(snapshots[2]),
             "--bundle", str(snapshots[1]), str(snapshots[0])],
            check=True, capture_output=True, text=True, timeout=COMMAND_TIMEOUT_SECONDS,
        )
        if snapshots[0].read_bytes() != payload:
            raise ValueError("verified decision snapshot changed")
    return strict_json(payload, "skill decision"), payload


parser = argparse.ArgumentParser()
parser.add_argument("--decision", type=Path, required=True)
parser.add_argument("--bundle", type=Path, required=True)
parser.add_argument("--authority-key", type=Path, required=True)
parser.add_argument("--asset-digest", required=True)
parser.add_argument("--scope-digest", required=True)
parser.add_argument("--policy-version", required=True)
parser.add_argument("--authority-generation", required=True)
parser.add_argument("--inventory-revision", required=True)
parser.add_argument("--inventory-receipt", type=Path, required=True)
args = parser.parse_args()

decision, decision_raw = verified_decision(
    args.decision, args.bundle, args.authority_key
)
if set(decision) != DECISION_FIELDS:
    raise SystemExit("decision schema differs")
if (
    decision["schema_version"] != "aidefend.skill-decision.v1"
    or decision["status"] not in {"APPROVED", "DENIED", "EXCEPTION"}
    or decision["revoked"] is not False
    or decision["asset_digest"] != args.asset_digest
    or decision["scope_digest"] != args.scope_digest
    or decision["policy_version"] != args.policy_version
    or decision["authority_generation"] != args.authority_generation
    or decision["inventory_revision"] != args.inventory_revision
    or not all(DIGEST.fullmatch(str(decision[field]))
               for field in ("asset_digest", "scope_digest"))
    or any(not isinstance(decision[field], str) or not decision[field]
           for field in ("decision_id", "issuer", "policy_version",
                         "authority_generation", "inventory_revision"))
):
    raise SystemExit("decision identity, authority, scope, or status differs")
now = datetime.now(timezone.utc)
if not when(decision["issued_at"]) <= now < when(decision["expires_at"]):
    raise SystemExit("decision is not current")

receipt = strict_json(regular_bytes(args.inventory_receipt), "inventory receipt")
if set(receipt) != RECEIPT_FIELDS or (
    receipt["schema_version"] != "aidefend.skill-inventory-update.v1"
    or receipt["decision_id"] != decision["decision_id"]
    or receipt["asset_digest"] != decision["asset_digest"]
    or receipt["recorded_status"] != decision["status"]
    or receipt["authority_generation"] != decision["authority_generation"]
    or receipt["inventory_revision_before"] != decision["inventory_revision"]
    or not isinstance(receipt["inventory_revision_after"], str)
    or not receipt["inventory_revision_after"]
    or receipt["inventory_revision_after"] == receipt["inventory_revision_before"]
    or receipt["record_sha256"]
       != "sha256:" + hashlib.sha256(decision_raw).hexdigest()
):
    raise SystemExit("inventory compare-and-swap readback differs")
print(json.dumps({
    "decision_id": decision["decision_id"],
    "asset_digest": decision["asset_digest"],
    "recorded_status": decision["status"],
    "inventory_revision": receipt["inventory_revision_after"],
    "runtime_profile_version": RUNTIME_PROFILE_VERSION,
    "runtime_profile_sha256": RUNTIME_PROFILE_SHA256,
}, sort_keys=True))

Action: The inventory adapter must compare-and-swap from the bound prior revision and return the readback receipt verified above. Missing, unsigned, expired, revoked, wrong-digest, wrong-scope, wrong-generation, ambiguous-authority, or conflicting-revision input is a finding and cannot be represented as approved. Recording this decision is not an admission receipt.

AID-M-001.003-G003

Concept:

Periodic reconciliation is an inventory hygiene control. Its job is to identify skills that no longer have a valid owner, are no longer used, or have drifted out of scanning coverage. That should be measured on a schedule even when no incident is in progress.

Step 1: Join inventory, HR/IAM, usage, and scan feeds

from __future__ import annotations

# File: skill_governance/reconcile_skill_inventory.py
import hashlib
import io
import json
import math
import os
import stat
import subprocess
import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pandas as pd


def strict_json_bytes(raw: bytes, label: str) -> object:
    def unique_object(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError(f"duplicate JSON key in {label}: {key}")
            value[key] = item
        return value

    def reject_constant(value):
        raise ValueError(f"non-finite JSON constant in {label}: {value}")

    try:
        return json.loads(
            raw.decode("utf-8", errors="strict"),
            object_pairs_hook=unique_object,
            parse_constant=reject_constant,
        )
    except (UnicodeDecodeError, json.JSONDecodeError, ValueError) as error:
        raise RuntimeError(f"{label} is not strict UTF-8 JSON") from error


def read_regular_bytes(path: Path) -> bytes:
    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(path, flags)
    with os.fdopen(descriptor, "rb") as handle:
        status = os.fstat(handle.fileno())
        if not stat.S_ISREG(status.st_mode) or status.st_size == 0:
            raise RuntimeError(f"missing, empty, or unsafe signed input: {path}")
        return handle.read()


def write_private_snapshot(path: Path, raw: bytes) -> None:
    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(path, flags, 0o400)
    with os.fdopen(descriptor, "wb") as handle:
        handle.write(raw)
        handle.flush()
        os.fsync(handle.fileno())


def capture_verified_bundle(
    payload_path: Path,
    bundle_path: Path,
    verification_key: str,
    timeout_seconds: float,
) -> bytes:
    payload = read_regular_bytes(payload_path)
    bundle = read_regular_bytes(bundle_path)
    with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
        snapshot_root = Path(directory)
        snapshot_root.chmod(0o700)
        payload_snapshot = snapshot_root / "payload"
        bundle_snapshot = snapshot_root / "bundle.json"
        write_private_snapshot(payload_snapshot, payload)
        write_private_snapshot(bundle_snapshot, bundle)
        subprocess.run(
            [
                "cosign", "verify-blob", "--key", verification_key,
                "--bundle", str(bundle_snapshot), str(payload_snapshot),
            ],
            check=True, capture_output=True, text=True, timeout=timeout_seconds,
        )
        if read_regular_bytes(payload_snapshot) != payload or read_regular_bytes(bundle_snapshot) != bundle:
            raise RuntimeError("private signed-input snapshot changed during verification")
    return payload

def load_signed_reconciliation_policy() -> tuple[dict, str]:
    policy_path = Path(os.environ["M001003_SKILL_POLICY_PATH"])
    bundle_path = Path(os.environ["M001003_SKILL_POLICY_SIGSTORE_BUNDLE"])
    verify_timeout = float(os.environ["AIDEFEND_POLICY_VERIFY_TIMEOUT_SECONDS"])
    if not math.isfinite(verify_timeout) or verify_timeout <= 0:
        raise RuntimeError("signed bootstrap-profile verification timeout is invalid")
    raw = capture_verified_bundle(
        policy_path,
        bundle_path,
        os.environ["M001003_SKILL_POLICY_VERIFY_KEY"],
        verify_timeout,
    )
    policy = strict_json_bytes(raw, "signed skill-reconciliation policy")
    required = {
        "schema_version", "policy_version", "inventory_path",
        "active_workers_path", "usage_path", "scan_status_path", "input_sha256",
        "report_path", "evidence_path", "dormant_after_seconds",
        "scan_max_age_seconds", "reconciliation_interval_seconds",
    }
    if not isinstance(policy, dict) or set(policy) != required:
        raise RuntimeError("signed skill-reconciliation policy fields differ")
    if (
        policy["schema_version"] != "aidefend.skill-reconciliation-policy.v1"
        or not isinstance(policy["policy_version"], str)
        or not policy["policy_version"]
    ):
        raise RuntimeError("signed skill-reconciliation policy identity is invalid")
    for field in (
        "inventory_path", "active_workers_path", "usage_path",
        "scan_status_path", "report_path", "evidence_path",
    ):
        if not isinstance(policy[field], str) or not policy[field]:
            raise RuntimeError("signed reconciliation path is invalid: " + field)
    for field in (
        "dormant_after_seconds", "scan_max_age_seconds",
        "reconciliation_interval_seconds",
    ):
        value = policy[field]
        if isinstance(value, bool) or not isinstance(value, int) or value < 1:
            raise RuntimeError("signed reconciliation interval is invalid: " + field)
    expected_input_keys = {"inventory", "workers", "usage", "scans"}
    if (
        not isinstance(policy["input_sha256"], dict)
        or set(policy["input_sha256"]) != expected_input_keys
        or any(
            not isinstance(value, str)
            or len(value) != 71
            or not value.startswith("sha256:")
            or set(value[7:]) - set("0123456789abcdef")
            for value in policy["input_sha256"].values()
        )
    ):
        raise RuntimeError("signed reconciliation input digests are invalid")
    return policy, "sha256:" + hashlib.sha256(raw).hexdigest()


POLICY, POLICY_SHA256 = load_signed_reconciliation_policy()
paths = {
    "inventory": Path(POLICY["inventory_path"]),
    "workers": Path(POLICY["active_workers_path"]),
    "usage": Path(POLICY["usage_path"]),
    "scans": Path(POLICY["scan_status_path"]),
}
input_bytes = {name: read_regular_bytes(path) for name, path in paths.items()}
input_sha256 = {
    name: "sha256:" + hashlib.sha256(raw).hexdigest()
    for name, raw in input_bytes.items()
}
if input_sha256 != POLICY["input_sha256"]:
    raise RuntimeError("reconciliation inputs differ from signed policy")

inventory = pd.read_csv(io.BytesIO(input_bytes["inventory"]))
active_workers = pd.read_csv(io.BytesIO(input_bytes["workers"]))
usage = pd.read_csv(io.BytesIO(input_bytes["usage"]))
scans = pd.read_csv(io.BytesIO(input_bytes["scans"]))
required_columns = {
    "inventory": {"skill_id", "owner_id"},
    "workers": {"worker_id"},
    "usage": {"skill_id", "last_invoked_at"},
    "scans": {"skill_id", "last_scan_at"},
}
for name, frame in {
    "inventory": inventory, "workers": active_workers,
    "usage": usage, "scans": scans,
}.items():
    if not required_columns[name].issubset(frame.columns):
        raise RuntimeError("reconciliation input schema differs: " + name)
    key = "worker_id" if name == "workers" else "skill_id"
    if frame[key].isna().any() or frame[key].duplicated().any():
        raise RuntimeError("reconciliation population key is absent or duplicated: " + name)
inventory_ids = set(inventory["skill_id"].astype(str))
if (
    not set(usage["skill_id"].astype(str)).issubset(inventory_ids)
    or not set(scans["skill_id"].astype(str)).issubset(inventory_ids)
):
    raise RuntimeError("usage or scan population contains an unregistered skill")

report = (
    inventory.merge(
        active_workers[["worker_id"]],
        left_on="owner_id",
        right_on="worker_id",
        how="left",
        indicator="owner_match",
    )
    .merge(usage, on="skill_id", how="left", validate="one_to_one")
    .merge(scans, on="skill_id", how="left", validate="one_to_one")
)
now_utc = datetime.now(timezone.utc)
last_invoked = pd.to_datetime(report["last_invoked_at"], utc=True, errors="coerce")
last_scan = pd.to_datetime(report["last_scan_at"], utc=True, errors="coerce")
report["is_orphaned"] = report["owner_match"].eq("left_only")
report["is_dormant"] = last_invoked.isna() | last_invoked.lt(
    now_utc - timedelta(seconds=POLICY["dormant_after_seconds"])
)
report["is_scan_gap"] = last_scan.isna() | last_scan.lt(
    now_utc - timedelta(seconds=POLICY["scan_max_age_seconds"])
)
findings = report[
    report["is_orphaned"] | report["is_dormant"] | report["is_scan_gap"]
].copy()

report_path = Path(POLICY["report_path"])
evidence_path = Path(POLICY["evidence_path"])
report_path.parent.mkdir(parents=True, exist_ok=True)
evidence_path.parent.mkdir(parents=True, exist_ok=True)
findings.to_csv(report_path, index=False)
evidence = {
    "schema_version": "aidefend.skill-reconciliation-evidence.v1",
    "control": "AID-M-001.003",
    "policy_version": POLICY["policy_version"],
    "policy_sha256": POLICY_SHA256,
    "reconciliation_interval_seconds": POLICY["reconciliation_interval_seconds"],
    "input_sha256": input_sha256,
    "expected_skill_count": len(inventory),
    "measured_skill_count": len(report),
    "finding_count": len(findings),
    "finding_skill_ids_sha256": "sha256:" + hashlib.sha256(
        json.dumps(
            sorted(findings["skill_id"].astype(str)),
            separators=(",", ":"),
        ).encode("utf-8")
    ).hexdigest(),
    "generated_at": now_utc.isoformat(),
    "finding_status": "FINDING" if len(findings) else "NO_FINDING",
}
evidence_path.write_text(
    json.dumps(evidence, sort_keys=True, separators=(",", ":")) + "\n",
    encoding="utf-8",
)
if json.loads(evidence_path.read_text(encoding="utf-8")) != evidence:
    raise RuntimeError("reconciliation evidence readback differs")

Action: Run this reconciliation at the cadence required by the signed skill-governance policy; do not use a framework-wide schedule. Emit a signed finding and downstream lifecycle trigger for every flagged skill, using the cadence and response windows in the signed organizational policy. Keep the finding open until the owning control returns a separately verified action receipt; this inventory control does not itself disable, block, or remediate the skill.

Discover existing SKILL.md packages before reconciliation

The collector below is a concrete filesystem path for deployed SKILL.md package trees. Configure roots from the managed agent installation inventory, not arbitrary user paths. Other platforms, including .mdc rule files and vendor-hosted skills, need an explicit platform collector and must remain unmeasured until it supplies complete coverage. Reuse the complete package-tree capture from G001; discovering a package does not approve it. A failed root or package capture prevents a clean inventory result.

# File: skill_governance/discover_installed.py
import os
from pathlib import Path
from skill_governance.install_hook import _canonical_skill_package

def discover_installed(roots, maximum_directories):
    if type(maximum_directories) is not int or maximum_directories < 1 or not roots:
        raise ValueError("a bounded managed root population is required")
    observed, failures, visited = [], [], 0
    for root_id, configured_path in sorted(roots.items()):
        root = Path(configured_path)
        if root.is_symlink() or not root.is_dir():
            failures.append({"root_id": root_id, "reason": "root_unavailable"})
            continue
        root = root.resolve(strict=True)
        def walk_error(error):
            failures.append({"root_id": root_id, "reason": type(error).__name__})
        for directory, dirs, files in os.walk(root, followlinks=False, onerror=walk_error):
            visited += 1
            if visited > maximum_directories:
                raise RuntimeError("managed scan exceeds approved population bound")
            directory = Path(directory)
            # Skipped links are missing coverage, never evidence of absence.
            for name in list(dirs):
                if (directory / name).is_symlink():
                    failures.append({"root_id": root_id, "reason": "linked_directory"})
                    dirs.remove(name)
            if "SKILL.md" in files:
                try:
                    digest, members = _canonical_skill_package(directory, "SKILL.md")
                    observed.append({"root_id": root_id,
                                     "package_path": directory.relative_to(root).as_posix(),
                                     "content_sha256": digest, "member_count": len(members)})
                except Exception as error:
                    failures.append({"root_id": root_id, "reason": type(error).__name__})
    return {"status": "INSUFFICIENT_DATA" if failures else "PASS",
            "observed_packages": observed, "coverage_failures": failures}

Join each observed managed root/package path and complete digest to the central registry; emit undeclared-package or digest-drift findings instead of automatically registering approval. Independently enumerate the same declared roots and compare package and member counts. Test unregistered packages, changed scripts with an unchanged SKILL.md, unreadable roots, links, capture mutation and a fully measured empty root.

AID-M-001.003-G004

Concept:

Trusted ownership, vulnerability, publisher, and incident events must immediately update inventory state and create a durable trigger for the control that owns the required action. Inventory state records the finding and routing decision; it never substitutes for a technical admission, retirement, containment, or eviction receipt.

Step 1: Define event-to-state transitions

# File: skill_governance/lifecycle_transitions.yaml
owner_departed:
  new_state: action_required
  downstream_control: AID-M-010.002
  requested_action: transfer_or_retire
publisher_compromised:
  new_state: action_required
  downstream_control: AID-H-030.005
  requested_action: hold_and_revalidate
cve_match:
  new_state: action_required
  downstream_control: AID-H-030.005
  requested_action: hold_and_revalidate
incident_containment:
  new_state: action_required
  downstream_control: AID-E-001
  requested_action: evict_applicable_authority

Step 2: Consume events and apply transitions through the inventory API

import math as _aidefend_math
import os as _aidefend_os
from urllib.parse import quote, urlsplit

# Admission injects these required values from one signature-verified, versioned runtime profile.
def _aidefend_required(name):
    value = _aidefend_os.environ.get(name)
    if value is None or not value.strip():
        raise RuntimeError(f"required runtime-profile field is absent: {name}")
    return value.strip()


def _aidefend_positive_float(name):
    try:
        value = float(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not numeric: {name}") from error
    if not _aidefend_math.isfinite(value) or value <= 0:
        raise RuntimeError(f"runtime-profile field must be finite and positive: {name}")
    return value


def _aidefend_positive_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 1:
        raise RuntimeError(f"runtime-profile field must be positive: {name}")
    return value


def _aidefend_nonnegative_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 0:
        raise RuntimeError(f"runtime-profile field cannot be negative: {name}")
    return value


def _aidefend_fraction(name):
    value = _aidefend_positive_float(name)
    if value > 1:
        raise RuntimeError(f"runtime-profile field must be in (0, 1]: {name}")
    return value


RUNTIME_PROFILE_VERSION = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_VERSION")
RUNTIME_PROFILE_SHA256 = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_SHA256").lower()
if (len(RUNTIME_PROFILE_SHA256) != 64
        or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")):
    raise RuntimeError("runtime-profile digest must be lowercase SHA-256")
SKILL_TRANSITION_REQUEST_TIMEOUT_SECONDS = _aidefend_positive_float("M001003_SKILL_TRANSITION_REQUEST_TIMEOUT_SECONDS")


def _skill_inventory_origin() -> str:
    raw = _aidefend_required("M001003_SKILL_INVENTORY_ORIGIN")
    try:
        parsed = urlsplit(raw)
        parsed.port
    except ValueError as error:
        raise RuntimeError("skill-inventory origin is invalid") from error
    if (
        parsed.scheme.lower() != "https"
        or not parsed.hostname
        or parsed.username is not None
        or parsed.password is not None
        or parsed.path not in {"", "/"}
        or parsed.query
        or parsed.fragment
    ):
        raise RuntimeError("skill-inventory URL must be an exact HTTPS origin")
    return raw.rstrip("/")


SKILL_INVENTORY_ORIGIN = _skill_inventory_origin()
SKILL_INVENTORY_CA = _aidefend_required("M001003_SKILL_INVENTORY_CA_BUNDLE")
SKILL_INVENTORY_CERT = _aidefend_required("M001003_SKILL_INVENTORY_CLIENT_CERT")
SKILL_INVENTORY_KEY = _aidefend_required("M001003_SKILL_INVENTORY_CLIENT_KEY")

# File: skill_governance/lifecycle_event_consumer.py
from pathlib import Path

import requests
import yaml
from fastapi import FastAPI
from pydantic import BaseModel

policy = yaml.safe_load(
    Path("skill_governance/lifecycle_transitions.yaml").read_text(encoding="utf-8")
)
app = FastAPI()


class LifecycleEvent(BaseModel):
    event_type: str
    skill_ids: list[str]
    source_event_id: str
    detected_at: str


@app.post("/v1/skill-lifecycle-events")
def handle_event(event: LifecycleEvent):
    transition = policy[event.event_type]
    for skill_id in event.skill_ids:
        payload = {
            "state": transition["new_state"],
            "source_event_id": event.source_event_id,
            "detected_at": event.detected_at,
            "downstream_control": transition["downstream_control"],
            "requested_action": transition["requested_action"],
        }
        try:
            with requests.post(
                SKILL_INVENTORY_ORIGIN + "/api/v1/skills/" + quote(skill_id, safe="") + "/transition",
                json=payload,
                timeout=SKILL_TRANSITION_REQUEST_TIMEOUT_SECONDS,
                verify=SKILL_INVENTORY_CA,
                cert=(SKILL_INVENTORY_CERT, SKILL_INVENTORY_KEY),
                headers={"Accept-Encoding": "identity"},
                allow_redirects=False,
                stream=True,
            ) as response:
                response.raise_for_status()
        except requests.RequestException as error:
            raise RuntimeError(f"inventory transition failed for {skill_id}") from error
    return {"transitioned": len(event.skill_ids)}

Action: Accept only authenticated, schema-validated events from authoritative HR/IAM, publisher-intelligence, vulnerability-correlation, and incident-response sources. Persist the source event ID, exact affected skill population, inventory transition, downstream control, and trigger receipt. Close the finding only after the owning control returns a separately verified receipt; never report a disablement, revalidation, retirement, or revocation merely because the inventory trigger was accepted.

Open-source tools

• MLflow
• Snyk Agent Scan (deployed agent, MCP server, and skill discovery)
• Snipe-IT
• SPIRE
• OpenTelemetry
• OpenBao

Commercial / hosted tools

• Wiz AI-SPM
• Palo Alto Networks Prisma Cloud
• ServiceNow IT Service Management
• Atlassian Jira Service Management
• HashiCorp Vault Enterprise
• HCP Vault Dedicated
• Splunk Enterprise Security
• Microsoft Sentinel
• Datadog Cloud SIEM

Threat mappings

MITRE ATLAS

• AML.T0010 AI Supply Chain Compromise (skill inventory enables tracking of supply-chain origin and publisher for every installed skill)
• AML.T0010.005 AI Supply Chain Compromise: AI Agent Tool

MAESTRO

• Compromised Agent Registry (L7) (inventory tracks source registry and publisher for every skill)
• Supply Chain Attacks (Cross-Layer) (skill lifecycle governance enables rapid identification and revocation of compromised skills)

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI04:2026 Agentic Supply Chain Vulnerabilities (skill inventory enables enterprise-wide supply-chain tracking for agentic skills)

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-5.2.1 Agent Profile Tampering (Detects installed instruction-package digest drift against recorded versions)
• AITech-5.2 Configuration Persistence (Detects installed instruction-package digest drift against recorded versions)
• AITech-9.3 Dependency / Plugin Compromise (Detects digest changes in installed third-party skill or plugin packages)

Google Secure AI Framework 2.0 - Risks

• IIC: Insecure Integrated Component (skill inventory identifies all integrated skill components for security assessment)

Databricks AI Security Framework 3.0

• Agents - Tools MCP Server 13.18: Tool Poisoning
• Agents - Tools MCP Server 13.21: Supply Chain Attacks