Scope boundary

Owns incident-time detachment, quarantine, tombstone propagation, and derivative-impact tracing for compromised LoRA, QLoRA, PEFT, prompt-tuning, and other adapter artifacts across serving, registries, caches, and lineage. It routes irreducible derivative impact to rollback or retraining.

AID-H-003.007: AID-E-003.005 removes a compromised adapter after an incident; AID-H-003.007 prevents unsafe adapters from loading through provenance, binding, admission, and behavioral validation. Preventive admission does not evict an adapter already present in serving or derivatives, and incident-time removal does not replace future admission enforcement.

AID-R-001.001: AID-E-003.005 detaches and tombstones compromised adapters and traces derivatives; AID-R-001.001 restores an exact known-good model or adapter attachment state. Rollback consumes eviction evidence but does not prove that the compromised adapter or derivative is unavailable across every registry, cache, and serving route.

AID-R-001.002: AID-E-003.005 removes detachable adapter influence and identifies irreducible derivatives; AID-R-001.002 builds and validates a newly remediated model when clean detachment or rollback is insufficient. Retraining does not itself prove adapter eviction, while adapter removal does not establish the replacement model's clean utility or incident-attack resistance.

Implementation guidance

• AID-E-003.005-G001: Detach and quarantine compromised adapters from serving routes, model runtimes, and active deployment manifests.
• AID-E-003.005-G002: Propagate adapter tombstones from the trusted adapter registry into serving runtimes, cache layers, and promotion gates.
• AID-E-003.005-G003: Trace merged and derivative model artifacts influenced by the compromised adapter, then route irreducible impact to rollback or retraining.

AID-E-003.005-G001

Runtime policy: Load adapter-inventory freshness, verifier timeout, text/list limits, and observation deadline from the signed, versioned eviction policy bound to this model service, adapter digest, and route/cache population, bind its version/digest to the independent export, and return ERROR if the policy or any value is absent, invalid, or unverifiable.

Concept:

The eviction executor may report what it attempted, but that action receipt is not proof that every serving route and cache stopped using the compromised adapter. Use a signed incident request with an exact target population, a signed executor receipt, and a fresh read-only export signed by a collector identity that cannot perform eviction. A third identity verifies the export and signs the verdict.

Required serving receipts

• adapter-eviction-request.json: incident ID, compromised digest, expected deployment IDs, expected cache IDs, and request timestamp; signed by incident command authority.
• adapter-detachment-action-receipt.json: request digest, executor identity, attempted deployments, control-plane operation IDs, and completion timestamp; signed by the eviction executor.
• adapter-live-state-snapshot.json: request digest, complete deployment/cache populations, active/resident adapter digests, collector timestamp, and population_complete; signed by a separately authorized read-only collector after querying the real serving control planes and caches.

Independent verifier

# File: incident_response/verify_adapter_detachment.py
from __future__ import annotations

import hashlib
import json
import math
import os
import subprocess
import tempfile
from pathlib import Path


REQUEST = Path("artifacts/adapter-eviction-request.json")
ACTION = Path("artifacts/adapter-detachment-action-receipt.json")
SNAPSHOT = Path("artifacts/adapter-live-state-snapshot.json")
VERDICT = Path("artifacts/adapter-detachment-verdict.json")
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["E003_ADAPTER_COMMAND_TIMEOUT_SECONDS"])
if (not RUNTIME_PROFILE_VERSION or len(RUNTIME_PROFILE_SHA256) != 64
        or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")
        or RUNTIME_PROFILE_SHA256 == "0" * 64
        or not math.isfinite(COMMAND_TIMEOUT_SECONDS) or COMMAND_TIMEOUT_SECONDS <= 0):
    raise RuntimeError("versioned adapter-detachment runtime profile is invalid")


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


def reject_nonfinite(value):
    raise ValueError(f"non-finite signed JSON value: {value}")


def strict_json(raw: bytes, label: str):
    try:
        return json.loads(
            raw.decode("utf-8", errors="strict"),
            object_pairs_hook=reject_duplicate_keys,
            parse_constant=reject_nonfinite,
        )
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError(f"{label} is not strict UTF-8 JSON") from exc


def verified_signed_bytes(
    payload_path: Path,
    bundle_path: Path,
    trust_key: str | Path,
    timeout_seconds: float,
    label: str,
) -> bytes:
    payload = payload_path.read_bytes()
    bundle = bundle_path.read_bytes()
    if not payload or not bundle:
        raise ValueError(f"{label} payload or bundle is empty")
    with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        bundle_snapshot = root / "payload.sigstore.json"
        payload_snapshot.write_bytes(payload)
        bundle_snapshot.write_bytes(bundle)
        os.chmod(payload_snapshot, 0o400)
        os.chmod(bundle_snapshot, 0o400)
        subprocess.run(
            ["cosign", "verify-blob", "--key", str(trust_key),
             "--bundle", str(bundle_snapshot), str(payload_snapshot)],
            check=True, capture_output=True, text=True, timeout=timeout_seconds,
        )
        verified = payload_snapshot.read_bytes()
        verified_bundle = bundle_snapshot.read_bytes()
        if verified != payload or verified_bundle != bundle:
            raise OSError(f"{label} verified payload or bundle snapshot changed")
    return verified


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def verify_blob(path: Path, key: str) -> bytes:
    return verified_signed_bytes(
        path, Path(str(path) + ".sigstore.json"), key, COMMAND_TIMEOUT_SECONDS, str(path)
    )


request_raw = verify_blob(REQUEST, os.environ["INCIDENT_COMMAND_VERIFY_KEY"])
action_raw = verify_blob(ACTION, os.environ["ADAPTER_EVICTOR_VERIFY_KEY"])
snapshot_raw = verify_blob(SNAPSHOT, os.environ["ADAPTER_STATE_COLLECTOR_VERIFY_KEY"])
request = strict_json(request_raw, "adapter eviction request")
action = strict_json(action_raw, "adapter detachment action")
snapshot = strict_json(snapshot_raw, "adapter live-state snapshot")

if set(request) != {
    "schema_version", "incident_id", "compromised_digest",
    "deployment_ids", "cache_ids", "requested_at",
} or request["schema_version"] != "aidefend.adapter-eviction-request.v1":
    raise SystemExit("adapter detachment blocked: request schema differs")
request_sha256 = hashlib.sha256(request_raw).hexdigest()
if (
    set(action) != {
        "schema_version", "request_sha256", "executor_identity",
        "attempted_deployment_ids", "operation_ids", "completed_at",
    }
    or action["schema_version"] != "aidefend.adapter-detachment-action.v1"
    or action["request_sha256"] != request_sha256
):
    raise SystemExit("adapter detachment blocked: action receipt is misbound")

snapshot_fields = {
    "schema_version", "request_sha256", "collected_at",
    "population_complete", "deployments", "caches",
}
if (
    set(snapshot) != snapshot_fields
    or snapshot["schema_version"] != "aidefend.adapter-live-state.v1"
    or snapshot["request_sha256"] != request_sha256
    or snapshot["population_complete"] is not True
):
    raise SystemExit("adapter detachment blocked: live-state snapshot is incomplete")
deployment_ids = {
    item["deployment_id"] for item in snapshot["deployments"]
    if set(item) == {"deployment_id", "active_adapter_digests"}
}
cache_ids = {
    item["cache_id"] for item in snapshot["caches"]
    if set(item) == {"cache_id", "resident_adapter_digests"}
}
if (
    deployment_ids != set(request["deployment_ids"])
    or cache_ids != set(request["cache_ids"])
    or len(deployment_ids) != len(snapshot["deployments"])
    or len(cache_ids) != len(snapshot["caches"])
):
    raise SystemExit("adapter detachment blocked: target population differs")

compromised = request["compromised_digest"]
still_live = [
    item["deployment_id"] for item in snapshot["deployments"]
    if compromised in item["active_adapter_digests"]
]
still_cached = [
    item["cache_id"] for item in snapshot["caches"]
    if compromised in item["resident_adapter_digests"]
]
if still_live or still_cached:
    raise SystemExit(
        f"adapter remains live: deployments={still_live}, caches={still_cached}"
    )

verdict = {
    "schema_version": "aidefend.adapter-detachment-verdict.v1",
    "incident_id": request["incident_id"],
    "compromised_digest": compromised,
    "request_sha256": request_sha256,
    "action_receipt_sha256": hashlib.sha256(action_raw).hexdigest(),
    "live_state_snapshot_sha256": hashlib.sha256(snapshot_raw).hexdigest(),
    "verified_deployment_count": len(deployment_ids),
    "verified_cache_count": len(cache_ids),
    "status": "PASS",
}
VERDICT.write_text(
    json.dumps(verdict, sort_keys=True, separators=(",", ":")) + "\n",
    encoding="utf-8",
)
subprocess.run(
    [
        "cosign", "sign-blob", "--yes",
        "--key", "env://ADAPTER_VERIFIER_SIGNING_KEY",
        "--bundle", str(Path(str(VERDICT) + ".sigstore.json")), str(VERDICT),
    ],
    check=True, timeout=COMMAND_TIMEOUT_SECONDS,
)
verify_blob(VERDICT, os.environ["ADAPTER_VERIFIER_VERIFY_KEY"])

Action: Detach or hot-unload through the real serving control plane, invalidate all enumerated caches, and keep the executor receipt as an action claim. Promotion or incident closure accepts only the independently signed PASS verdict whose complete read-only snapshot proves the digest absent from every requested deployment and cache.

AID-E-003.005-G002

Runtime policy: Load tombstone-registry byte/entry/identifier/text ceilings, signature deadline, freshness, and retention from the signed, versioned loader policy bound to this registry authority and loader population, bind its version/digest to load/readback evidence, and return ERROR if the policy or any value is absent, invalid, or unverifiable.

Concept:

Adapter removal must survive retries, rollback, and cache repopulation. Make the tombstone registry a signed security artifact and verify it before every loader, cache warmer, and promotion decision. A local JSON file without signature verification can be edited to reactivate the compromised digest.

Signed tombstone registry

{
  "schema_version": "aidefend.adapter-tombstones.v1",
  "registry_version": "2026.07.13.1",
  "generation": 1,
  "tombstones": [
    {
      "adapter_id": "support-lora-v2",
      "adapter_digest": "sha256:9e7d3a4c1b28f506d87211c685f59ac73851ee3c309f82de4c76f0ab1d29e546",
      "reason": "backdoor trigger confirmed",
      "incident_id": "INC-2026-0704-002",
      "tombstoned_at": "2026-07-13T10:00:00Z"
    }
  ]
}

cosign sign-blob --yes --key env://ADAPTER_TOMBSTONE_SIGNING_KEY \
  --bundle registry/adapter_tombstones.json.sigstore.json \
  registry/adapter_tombstones.json

Authoritative rollback fence

Add an integer generation to the signed registry. A separate publisher verifies the new signed registry and the complete retained tombstone population, then advances its generation and exact-byte digest using publish_head below. Loader credentials have GET only; publisher credentials alone have the transaction/write permissions on the profile-bound head key. Keep the head in a durable primary outside artifact rollback and cache restoration, with no eviction or TTL. Initial enrollment is an explicit publisher operation; loss or rollback of the authority requires loader admission to remain disabled until an independent latest-generation recovery has completed. Every loader consults this authority and rejects a validly signed older registry or conflicting bytes at the same generation. Pin TLS endpoint, key namespace, credentials and timeouts through the verified loader profile. Do not permit deployment rollback to select an older authority.

Fail-closed loader enforcement

# File: runtime/enforce_adapter_tombstones.py
from __future__ import annotations

import hashlib
import json
import math
import os
import re
import subprocess
import tempfile
from pathlib import Path

import redis


REGISTRY = Path("registry/adapter_tombstones.json")
COSIGN_BUNDLE = Path("registry/adapter_tombstones.json.sigstore.json")
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["E003_ADAPTER_COMMAND_TIMEOUT_SECONDS"])
DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
if (not RUNTIME_PROFILE_VERSION or len(RUNTIME_PROFILE_SHA256) != 64
        or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")
        or RUNTIME_PROFILE_SHA256 == "0" * 64
        or not math.isfinite(COMMAND_TIMEOUT_SECONDS) or COMMAND_TIMEOUT_SECONDS <= 0):
    raise RuntimeError("versioned adapter-tombstone runtime profile is invalid")


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


def reject_nonfinite(value):
    raise ValueError(f"non-finite signed JSON value: {value}")


def strict_json(raw: bytes, label: str):
    try:
        return json.loads(
            raw.decode("utf-8", errors="strict"),
            object_pairs_hook=reject_duplicate_keys,
            parse_constant=reject_nonfinite,
        )
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError(f"{label} is not strict UTF-8 JSON") from exc


def verified_signed_bytes(
    payload_path: Path,
    bundle_path: Path,
    trust_key: str | Path,
    timeout_seconds: float,
    label: str,
) -> bytes:
    payload = payload_path.read_bytes()
    bundle = bundle_path.read_bytes()
    if not payload or not bundle:
        raise ValueError(f"{label} payload or bundle is empty")
    with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        bundle_snapshot = root / "payload.sigstore.json"
        payload_snapshot.write_bytes(payload)
        bundle_snapshot.write_bytes(bundle)
        os.chmod(payload_snapshot, 0o400)
        os.chmod(bundle_snapshot, 0o400)
        subprocess.run(
            ["cosign", "verify-blob", "--key", str(trust_key),
             "--bundle", str(bundle_snapshot), str(payload_snapshot)],
            check=True, capture_output=True, text=True, timeout=timeout_seconds,
        )
        verified = payload_snapshot.read_bytes()
        verified_bundle = bundle_snapshot.read_bytes()
        if verified != payload or verified_bundle != bundle:
            raise OSError(f"{label} verified payload or bundle snapshot changed")
    return verified


class TombstonedAdapterError(RuntimeError):
    pass


def head_connection():
    # Values are injected only from the verified loader profile. This is the
    # authoritative primary, not a read replica or a restored local snapshot.
    url = os.environ["ADAPTER_HEAD_REDIS_URL"]
    if not url.startswith("rediss://"):
        raise TombstonedAdapterError("head authority requires authenticated TLS")
    return redis.Redis.from_url(
        url, decode_responses=True, ssl_cert_reqs="required",
        socket_connect_timeout=COMMAND_TIMEOUT_SECONDS,
        socket_timeout=COMMAND_TIMEOUT_SECONDS,
    )


def require_current_head(generation: int, digest: str) -> None:
    key = os.environ["ADAPTER_HEAD_REDIS_KEY"]
    if not key:
        raise TombstonedAdapterError("head authority key is missing")
    # No TTL/cache: each admission consults the live monotonic authority.
    with head_connection() as connection:
        raw = connection.get(key)
    if raw is None:
        raise TombstonedAdapterError("head is unavailable; enrollment is required")
    head = strict_json(raw.encode("utf-8"), "registry head")
    if (not isinstance(head, dict) or set(head) != {"generation", "sha256"}
            or type(head["generation"]) is not int
            or head["generation"] != generation or head["sha256"] != digest):
        raise TombstonedAdapterError("stale or conflicting tombstone registry")


def publish_head(connection, key: str, previous: str | None, generation: int, digest: str) -> None:
    # Called by the separate authorized publisher after registry signature and
    # full tombstone-population verification, never by a loader credential.
    if type(generation) is not int or generation < 1 or re.fullmatch(r"[0-9a-f]{64}", digest) is None:
        raise ValueError("invalid registry head")
    if previous is not None:
        prior = strict_json(previous.encode(), "previous registry head")
        if type(prior.get("generation")) is not int or generation <= prior["generation"]:
            raise ValueError("registry generation must increase")
    elif generation != 1:
        raise ValueError("initial enrollment must use generation one")
    value = json.dumps({"generation": generation, "sha256": digest}, sort_keys=True, separators=(",", ":"))
    # WATCH guards the exact previous bytes, including concurrent enrollment.
    with connection.pipeline() as transaction:
        transaction.watch(key)
        if transaction.get(key) != previous:
            raise RuntimeError("head changed; reread and verify before retry")
        transaction.multi()
        transaction.set(key, value)
        transaction.execute()


def load_tombstones() -> set[str]:
    raw = verified_signed_bytes(
        REGISTRY, COSIGN_BUNDLE, os.environ["ADAPTER_TOMBSTONE_VERIFY_KEY"],
        COMMAND_TIMEOUT_SECONDS, "adapter tombstone registry",
    )
    data = strict_json(raw, "adapter tombstone registry")
    if (
        not isinstance(data, dict)
        or set(data) != {"schema_version", "registry_version", "generation", "tombstones"}
        or data["schema_version"] != "aidefend.adapter-tombstones.v1"
        or not isinstance(data["tombstones"], list)
    ):
        raise TombstonedAdapterError("tombstone registry schema differs")
    if (type(data["generation"]) is not int or data["generation"] < 1
            or not isinstance(data["registry_version"], str) or not data["registry_version"]):
        raise TombstonedAdapterError("registry generation/version is invalid")
    require_current_head(data["generation"], hashlib.sha256(raw).hexdigest())
    digests = set()
    for entry in data["tombstones"]:
        if not isinstance(entry, dict) or set(entry) != {
            "adapter_id", "adapter_digest", "reason",
            "incident_id", "tombstoned_at",
        }:
            raise TombstonedAdapterError("tombstone entry schema differs")
        for field in ("adapter_id", "reason", "incident_id", "tombstoned_at"):
            if not isinstance(entry[field], str) or not entry[field]:
                raise TombstonedAdapterError(f"tombstone {field} is required")
        if DIGEST_RE.fullmatch(entry["adapter_digest"]) is None:
            raise TombstonedAdapterError("tombstone adapter digest is invalid")
        if entry["adapter_digest"] in digests:
            raise TombstonedAdapterError("duplicate tombstone digest")
        digests.add(entry["adapter_digest"])
    return digests


def enforce_not_tombstoned(adapter_digest: str) -> None:
    if DIGEST_RE.fullmatch(adapter_digest) is None:
        raise TombstonedAdapterError("adapter digest is invalid")
    if adapter_digest in load_tombstones():
        raise TombstonedAdapterError(
            f"adapter_digest_tombstoned:{adapter_digest}"
        )

Action: Call enforce_not_tombstoned before registry promotion, download, cache fill, hot-load, and process startup. An independent drill identity must attempt to load every newly tombstoned digest through each applicable path and require denial; the read-only collector from the detachment guidance must also show the digest absent from the complete live/cache population. Sign the drill result with the verification identity, not the loader or evictor.

AID-E-003.005-G003

Runtime policy: Load lineage-export size/depth, object/list, freshness, verifier-deadline, and decision-lifetime bounds from the signed, versioned routing policy bound to this registry snapshot and adapter digest, bind its version/digest to dispositions and registry replay, and return ERROR if the policy or any value is absent, invalid, or unverifiable.

Concept:

Merged, distilled, quantized, or otherwise derived models cannot be cleaned by detaching the source adapter. Query a complete lineage export from the authoritative registry, require that export to be signed by a read-only lineage authority, and let a separate incident router sign the remediation decision.

Export signed adapter lineage

{
  "schema_version": "aidefend.adapter-derivative-lineage.v1",
  "registry_snapshot_id": "lineage-2026-07-13T10:05:00Z",
  "population_complete": true,
  "derivatives": [
    {
      "adapter_digest": "sha256:9e7d3a4c1b28f506d87211c685f59ac73851ee3c309f82de4c76f0ab1d29e546",
      "artifact_id": "support-llm-merged-v4",
      "artifact_digest": "sha256:4fb7f4d9f2ce7cb28bb7c5311a5f3f0b3ea99c1f8a95638997276e48e97c5fa1",
      "artifact_type": "merged_model",
      "reversible": false,
      "serving_deployment": "llm-support-prod"
    }
  ]
}

Verify lineage and sign routing

# File: incident_response/trace_adapter_derivatives.py
from __future__ import annotations

import json
import math
import os
import re
import subprocess
import tempfile
from pathlib import Path


LINEAGE = Path("artifacts/adapter-derivative-lineage.json")
COSIGN_BUNDLE = Path(str(LINEAGE) + ".sigstore.json")
OUTPUT = Path("artifacts/adapter-derivative-routing.json")
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["E003_ADAPTER_COMMAND_TIMEOUT_SECONDS"])
DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
if (not RUNTIME_PROFILE_VERSION or len(RUNTIME_PROFILE_SHA256) != 64
        or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")
        or RUNTIME_PROFILE_SHA256 == "0" * 64
        or not math.isfinite(COMMAND_TIMEOUT_SECONDS) or COMMAND_TIMEOUT_SECONDS <= 0):
    raise RuntimeError("versioned adapter-derivative runtime profile is invalid")


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


def reject_nonfinite(value):
    raise ValueError(f"non-finite signed JSON value: {value}")


def strict_json(raw: bytes, label: str):
    try:
        return json.loads(
            raw.decode("utf-8", errors="strict"),
            object_pairs_hook=reject_duplicate_keys,
            parse_constant=reject_nonfinite,
        )
    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ValueError(f"{label} is not strict UTF-8 JSON") from exc


def verified_signed_bytes(
    payload_path: Path,
    bundle_path: Path,
    trust_key: str | Path,
    timeout_seconds: float,
    label: str,
) -> bytes:
    payload = payload_path.read_bytes()
    bundle = bundle_path.read_bytes()
    if not payload or not bundle:
        raise ValueError(f"{label} payload or bundle is empty")
    with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        bundle_snapshot = root / "payload.sigstore.json"
        payload_snapshot.write_bytes(payload)
        bundle_snapshot.write_bytes(bundle)
        os.chmod(payload_snapshot, 0o400)
        os.chmod(bundle_snapshot, 0o400)
        subprocess.run(
            ["cosign", "verify-blob", "--key", str(trust_key),
             "--bundle", str(bundle_snapshot), str(payload_snapshot)],
            check=True, capture_output=True, text=True, timeout=timeout_seconds,
        )
        verified = payload_snapshot.read_bytes()
        verified_bundle = bundle_snapshot.read_bytes()
        if verified != payload or verified_bundle != bundle:
            raise OSError(f"{label} verified payload or bundle snapshot changed")
    return verified


def verify_lineage() -> dict:
    raw = verified_signed_bytes(
        LINEAGE, COSIGN_BUNDLE, os.environ["LINEAGE_AUTHORITY_VERIFY_KEY"],
        COMMAND_TIMEOUT_SECONDS, "adapter derivative lineage",
    )
    lineage = strict_json(raw, "adapter derivative lineage")
    if (
        not isinstance(lineage, dict)
        or set(lineage) != {
            "schema_version", "registry_snapshot_id",
            "population_complete", "derivatives",
        }
        or lineage["schema_version"]
        != "aidefend.adapter-derivative-lineage.v1"
        or lineage["population_complete"] is not True
        or not isinstance(lineage["derivatives"], list)
    ):
        raise SystemExit("signed derivative lineage is incomplete")
    return lineage


def route_derivatives(
    compromised_digest: str, incident_id: str
) -> dict:
    if DIGEST_RE.fullmatch(compromised_digest) is None:
        raise ValueError("compromised adapter digest is invalid")
    if (
        not isinstance(incident_id, str)
        or not incident_id
        or len(incident_id) > 128
    ):
        raise ValueError("incident ID is invalid")
    lineage = verify_lineage()
    affected = []
    seen = set()
    for item in lineage["derivatives"]:
        if set(item) != {
            "adapter_digest", "artifact_id", "artifact_digest",
            "artifact_type", "reversible", "serving_deployment",
        }:
            raise SystemExit("derivative lineage entry schema differs")
        if (
            DIGEST_RE.fullmatch(item["adapter_digest"]) is None
            or DIGEST_RE.fullmatch(item["artifact_digest"]) is None
        ):
            raise SystemExit("derivative lineage digest is invalid")
        for field in ("artifact_id", "artifact_type", "serving_deployment"):
            if not isinstance(item[field], str) or not item[field]:
                raise SystemExit(f"derivative lineage {field} is required")
        if type(item["reversible"]) is not bool:
            raise SystemExit("derivative lineage reversible flag is invalid")
        identity = (item["artifact_id"], item["artifact_digest"])
        if identity in seen:
            raise SystemExit("duplicate derivative in lineage export")
        seen.add(identity)
        if item["adapter_digest"] != compromised_digest:
            continue
        route = (
            "AID-E-003.005_detach_adapter"
            if item["reversible"] is True
            else "AID-R-001.001_or_AID-R-001.002"
        )
        affected.append({
            **item,
            "incident_id": incident_id,
            "remediation_route": route,
        })

    result = {
        "schema_version": "aidefend.adapter-derivative-routing.v1",
        "incident_id": incident_id,
        "compromised_digest": compromised_digest,
        "registry_snapshot_id": lineage["registry_snapshot_id"],
        "affected_artifact_count": len(affected),
        "affected_artifacts": affected,
        "requires_restore_workflow": any(
            item["reversible"] is False for item in affected
        ),
    }
    OUTPUT.write_text(
        json.dumps(result, sort_keys=True, separators=(",", ":")) + "\n",
        encoding="utf-8",
    )
    subprocess.run(
        [
            "cosign", "sign-blob", "--yes",
            "--key", "env://DERIVATIVE_ROUTER_SIGNING_KEY",
            "--bundle", str(Path(str(OUTPUT) + ".sigstore.json")),
            str(OUTPUT),
        ],
        check=True, timeout=COMMAND_TIMEOUT_SECONDS,
    )
    return result


if __name__ == "__main__":
    route_derivatives(
        os.environ["COMPROMISED_ADAPTER_DIGEST"],
        os.environ["INCIDENT_ID"],
    )

Action: Preserve the signed complete lineage snapshot and signed routing decision. Detach only truly reversible runtime bindings here; quarantine and route every irreducible derivative to AID-R-001.001 rollback or AID-R-001.002 clean retraining. A mutable or incomplete lineage file is INSUFFICIENT_DATA and cannot support incident closure.

Open-source tools

• Hugging Face PEFT
• safetensors
• MLflow Model Registry
• Kubernetes
• DVC

Commercial / hosted tools

• Amazon SageMaker AI Model Registry
• Google Vertex AI Model Registry
• Databricks Models in Unity Catalog
• W&B Registry

Threat mappings

MITRE ATLAS

• AML.T0011 User Execution
• AML.T0115 Publish Poisoned AI Artifacts
• AML.T0011.000 User Execution: Unsafe AI Artifacts
• AML.T0018.003 Manipulate AI Model: Modify Prompt Construction Logic
• AML.T0018 Manipulate AI Model
• AML.T0018.002 Manipulate AI Model: Embed Malware
• AML.T0115.001 Publish Poisoned AI Artifacts: Models
• AML.T0010 AI Supply Chain Compromise
• AML.T0010.003 AI Supply Chain Compromise: Model (adapter quarantine removes the compromised model component from registries and serving)

MAESTRO

• Backdoor Attacks (L1)
• Supply Chain Attacks (Cross-Layer)

OWASP LLM Top 10 2026

• LLM04:2026 Supply Chain
• LLM05:2026 Data and Model Poisoning

OWASP ML Top 10 2023

• ML06:2023 AI Supply Chain Attacks
• ML10:2023 Model Poisoning

OWASP Top 10 for Agentic Applications 2026

• ASI04:2026 Agentic Supply Chain Vulnerabilities

NIST Adversarial Machine Learning 2025

• NISTAML.051 Model Poisoning (Supply Chain)
• NISTAML.023 Backdoor Poisoning

Cisco Integrated AI Security and Safety Framework

• AISubtech-9.2.2 Backdoors and Trojans

Google Secure AI Framework 2.0 - Risks

• MST: Model Source Tampering (removes tampered adapter artifacts)
• MDT: Model Deployment Tampering (removes compromised adapters from serving deployments)

Databricks AI Security Framework 3.0

• Model 7.1: Backdoor machine learning / Trojaned model
• Model 7.3: ML Supply chain vulnerabilities