AID-D-004.007: AI Lifecycle-Path Drift Detection

Version 1.20261015 · Actionable sub-technique

Detect when an AI asset, dataset, prompt, index, evaluator, or deployment follows a lifecycle transition or processing path that differs from its approved architecture and stage-authorization record.

Pillars: data, model, infra, app

Lifecycle: operation

Scope boundary

Owns findings for observed AI lifecycle transitions or processing paths that differ from the approved architecture and stage-authorization record. It does not authorize stage ingress or block the transition.

AID-H-029.002: AID-D-004.007 detects an unauthorized or unexplained lifecycle path after observing lineage and transitions; AID-H-029.002 authorizes and enforces whether data or artifacts may enter a lifecycle stage. Drift monitoring identifies bypass or divergence, while the Harden gate owns the synchronous stage decision.

Implementation guidance

AID-D-004.007-G001

Run periodic cross-stage path audits that compare actual lineage against the expected stage-authorized paths and alert on indirect unauthorized transitions.
Scope and signed policy

Run this audit only for lifecycle assets that the versioned environment policy places in scope; record NOT_APPLICABLE at control selection when no governed stage-transition path exists. Materialize the allowed directed acyclic graph, complete paths, authoritative sources, asset population, closed window, and evaluator revision from the exact verified policy bytes.

set -euo pipefail
umask 077
snapshot="$(mktemp -d)"
trap 'rm -rf -- "$snapshot"' EXIT

capture_signed_blob() {
  local label="$1" payload="$2" bundle="$3" key="$4" source
  for source in "$payload" "$bundle" "$key"; do
    test -f "$source" && test ! -L "$source" || return 1
  done
  install -m 0400 -- "$payload" "$snapshot/$label.payload"
  install -m 0400 -- "$bundle" "$snapshot/$label.bundle"
  install -m 0400 -- "$key" "$snapshot/$label.pub"
  cosign verify-blob --key "$snapshot/$label.pub" \
    --bundle "$snapshot/$label.bundle" "$snapshot/$label.payload"
}

capture_signed_blob environment-policy \
  policy/environment-assurance-policy.json \
  policy/environment-assurance-policy.bundle \
  keys/environment-policy.pub
test -f keys/lifecycle-graph.pub && test ! -L keys/lifecycle-graph.pub
install -m 0400 -- keys/lifecycle-graph.pub "$snapshot/lifecycle-graph.pub"

python - "$snapshot/environment-policy.payload" \
  "$snapshot/lifecycle-path-policy.json" <<'PY'
import hashlib
import json
import math
import re
import sys
from datetime import datetime
from pathlib import Path

NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/@+-]*$")

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

def strict_json(raw):
    return json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=lambda token: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + token)
        ),
    )

def bounded(value, label):
    if not isinstance(value, str) or not NAME.fullmatch(value):
        raise ValueError(label + " is not a bounded identifier")
    return value

def unique_names(value, label):
    if not isinstance(value, list) or not value or len(value) != len(set(value)):
        raise ValueError(label + " must be a non-empty unique array")
    return [bounded(item, label) for item in value]

def timestamp(value, label):
    if not isinstance(value, str) or not value:
        raise ValueError(label + " is absent")
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        raise ValueError(label + " lacks a timezone")
    return parsed

raw = Path(sys.argv[1]).read_bytes()
profile = strict_json(raw)
if (
    not isinstance(profile, dict)
    or profile.get("schema_version") != "aidefend.environment-assurance-policy.v1"
    or not isinstance(profile.get("policy_version"), str)
    or not profile["policy_version"]
    or not isinstance(profile.get("controls"), dict)
    or not isinstance(profile["controls"].get("AID-D-004.007"), dict)
):
    raise ValueError("environment policy identity or control is invalid")
control = profile["controls"]["AID-D-004.007"]
required_fields = {
    "window_id", "window_start", "window_end", "evaluator_revision",
    "authorized_stages", "allowed_edges", "allowed_paths",
    "required_sources", "asset_scope", "max_staleness_seconds",
    "max_future_skew_seconds",
}
if not required_fields.issubset(control):
    raise ValueError("lifecycle-path control policy is incomplete")
for field in ("window_id", "evaluator_revision"):
    bounded(control[field], field)
if timestamp(control["window_start"], "window_start") >= timestamp(
    control["window_end"], "window_end"
):
    raise ValueError("measurement window is not ordered")
for field, positive in (
    ("max_staleness_seconds", True),
    ("max_future_skew_seconds", False),
):
    value = control[field]
    if (
        isinstance(value, bool) or not isinstance(value, (int, float))
        or not math.isfinite(float(value))
        or (float(value) <= 0 if positive else float(value) < 0)
    ):
        raise ValueError(field + " is invalid")

stages = unique_names(control["authorized_stages"], "authorized_stages")
stage_set = set(stages)
edges = []
edge_set = set()
for number, row in enumerate(control["allowed_edges"], 1):
    if not isinstance(row, dict) or set(row) != {"from", "to"}:
        raise ValueError(f"allowed_edges:{number}: schema differs")
    edge = (
        bounded(row["from"], f"allowed_edges:{number}:from"),
        bounded(row["to"], f"allowed_edges:{number}:to"),
    )
    if edge[0] == edge[1] or not set(edge) <= stage_set or edge in edge_set:
        raise ValueError(f"allowed_edges:{number}: invalid or duplicate edge")
    edge_set.add(edge)
    edges.append({"from": edge[0], "to": edge[1]})
if not edges:
    raise ValueError("allowed edge set is empty")

paths = []
path_set = set()
for number, raw_path in enumerate(control["allowed_paths"], 1):
    path = unique_names(raw_path, f"allowed_paths:{number}")
    signature = tuple(path)
    if (
        len(path) < 2 or not set(path) <= stage_set or signature in path_set
        or any((a, b) not in edge_set for a, b in zip(path, path[1:]))
    ):
        raise ValueError(f"allowed_paths:{number}: invalid or duplicate path")
    path_set.add(signature)
    paths.append(path)

adjacency = {stage: set() for stage in stages}
for left, right in edge_set:
    adjacency[left].add(right)
visiting = set()
visited = set()
def visit(stage):
    if stage in visiting:
        raise ValueError("authorized lifecycle graph contains a cycle")
    if stage in visited:
        return
    visiting.add(stage)
    for target in adjacency[stage]:
        visit(target)
    visiting.remove(stage)
    visited.add(stage)
for stage in stages:
    visit(stage)

required_sources = unique_names(control["required_sources"], "required_sources")
required_set = set(required_sources)
scope = []
seen_assets = set()
if not isinstance(control["asset_scope"], list) or not control["asset_scope"]:
    raise ValueError("asset_scope is empty; record NOT_APPLICABLE before this job")
for number, row in enumerate(control["asset_scope"], 1):
    if not isinstance(row, dict) or set(row) != {"asset_id", "required_sources"}:
        raise ValueError(f"asset_scope:{number}: schema differs")
    asset = bounded(row["asset_id"], f"asset_scope:{number}:asset_id")
    sources = unique_names(
        row["required_sources"], f"asset_scope:{number}:required_sources"
    )
    if asset in seen_assets or not set(sources) <= required_set:
        raise ValueError(f"asset_scope:{number}: duplicate asset or unknown source")
    seen_assets.add(asset)
    scope.append({"asset_id": asset, "required_sources": sources})

output = {
    "schema_version": "aidefend.lifecycle-path-policy.v2",
    "environment_policy_version": profile["policy_version"],
    "environment_policy_sha256": hashlib.sha256(raw).hexdigest(),
    "window_id": control["window_id"],
    "window_start": control["window_start"],
    "window_end": control["window_end"],
    "evaluator_revision": control["evaluator_revision"],
    "authorized_stages": stages,
    "allowed_edges": edges,
    "allowed_paths": paths,
    "required_sources": required_sources,
    "asset_scope": scope,
    "transition_population_source": "signed_lifecycle_transition_ledger",
    "max_staleness_seconds": control["max_staleness_seconds"],
    "max_future_skew_seconds": control["max_future_skew_seconds"],
}
Path(sys.argv[2]).write_bytes(
    (json.dumps(output, sort_keys=True, separators=(",", ":")) + "\n").encode()
)
PY

cosign sign-blob --yes --key env://LIFECYCLE_GRAPH_SIGNING_KEY \
  --bundle "$snapshot/lifecycle-path-policy.bundle" \
  "$snapshot/lifecycle-path-policy.json"
cosign verify-blob --key "$snapshot/lifecycle-graph.pub" \
  --bundle "$snapshot/lifecycle-path-policy.bundle" \
  "$snapshot/lifecycle-path-policy.json"
install -d -m 0750 policy
install -m 0440 -- "$snapshot/lifecycle-path-policy.bundle" \
  policy/lifecycle-path-policy.bundle.next
install -m 0440 -- "$snapshot/lifecycle-path-policy.json" \
  policy/lifecycle-path-policy.json.next
mv -f -- policy/lifecycle-path-policy.bundle.next policy/lifecycle-path-policy.bundle
mv -f -- policy/lifecycle-path-policy.json.next policy/lifecycle-path-policy.json
cmp --silent "$snapshot/lifecycle-path-policy.bundle" policy/lifecycle-path-policy.bundle
cmp --silent "$snapshot/lifecycle-path-policy.json" policy/lifecycle-path-policy.json
Capture first, then reconcile the complete window

Each payload, Sigstore bundle, and verification key is copied into a private snapshot before verification. The evaluator reads only those verified copies; a producer cannot swap a path, ledger, or proof between verification and use. A signed empty authoritative ledger is the only valid zero-event population.

set -euo pipefail
umask 077
snapshot="$(mktemp -d)"
trap 'rm -rf -- "$snapshot"' EXIT
result="$snapshot/lifecycle-path-findings.json"
result_bundle="$snapshot/lifecycle-path-findings.bundle"

capture_signed_blob() {
  local label="$1" payload="$2" bundle="$3" key="$4" source
  for source in "$payload" "$bundle" "$key"; do
    test -f "$source" && test ! -L "$source" || return 1
  done
  install -m 0400 -- "$payload" "$snapshot/$label.payload" || return 1
  install -m 0400 -- "$bundle" "$snapshot/$label.bundle" || return 1
  install -m 0400 -- "$key" "$snapshot/$label.pub" || return 1
  cosign verify-blob --key "$snapshot/$label.pub" \
    --bundle "$snapshot/$label.bundle" "$snapshot/$label.payload"
}

test -f keys/lifecycle-detector.pub && test ! -L keys/lifecycle-detector.pub
install -m 0400 -- keys/lifecycle-detector.pub "$snapshot/lifecycle-detector.pub"

sign_and_publish() {
  cosign sign-blob --yes --key env://LIFECYCLE_DETECTOR_KEY \
    --bundle "$result_bundle" "$result"
  cosign verify-blob --key "$snapshot/lifecycle-detector.pub" \
    --bundle "$result_bundle" "$result"
  install -d -m 0750 evidence
  install -m 0440 -- "$result_bundle" evidence/lifecycle-path-findings.bundle.next
  install -m 0440 -- "$result" evidence/lifecycle-path-findings.json.next
  mv -f -- evidence/lifecycle-path-findings.bundle.next \
    evidence/lifecycle-path-findings.bundle
  mv -f -- evidence/lifecycle-path-findings.json.next \
    evidence/lifecycle-path-findings.json
  cmp --silent "$result_bundle" evidence/lifecycle-path-findings.bundle
  cmp --silent "$result" evidence/lifecycle-path-findings.json
}

write_verification_error() {
  python - "$result" <<'PY'
import json
import sys
from datetime import datetime, timezone
from pathlib import Path

value = {
    "schema_version": "aidefend.lifecycle-path-findings.v2",
    "control": "AID-D-004.007",
    "status": "ERROR",
    "status_reason": "signed input capture or verification failed",
    "evaluated_at": datetime.now(timezone.utc).isoformat(),
}
Path(sys.argv[1]).write_text(
    json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n",
    encoding="utf-8",
)
PY
}

if ! capture_signed_blob path-policy \
     policy/lifecycle-path-policy.json policy/lifecycle-path-policy.bundle \
     keys/lifecycle-graph.pub \
  || ! capture_signed_blob source-populations \
     evidence/lifecycle-source-populations.json \
     evidence/lifecycle-source-populations.bundle \
     keys/lifecycle-source-reader.pub \
  || ! capture_signed_blob transition-population \
     evidence/lifecycle-transition-population.json \
     evidence/lifecycle-transition-population.bundle \
     keys/lifecycle-transition-ledger.pub \
  || ! capture_signed_blob observed-transitions \
     evidence/lifecycle-transitions.jsonl \
     evidence/lifecycle-transitions.bundle \
     keys/lifecycle-observer.pub; then
  write_verification_error
  sign_and_publish
  exit 2
fi

evaluated_at="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
set +e
python - \
  "$snapshot/path-policy.payload" "$snapshot/path-policy.bundle" \
  "$snapshot/source-populations.payload" "$snapshot/source-populations.bundle" \
  "$snapshot/transition-population.payload" "$snapshot/transition-population.bundle" \
  "$snapshot/observed-transitions.payload" "$snapshot/observed-transitions.bundle" \
  "$evaluated_at" "$result" <<'PY'
import hashlib
import json
import math
import re
import sys
from collections import defaultdict
from datetime import datetime
from pathlib import Path

NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/@+-]*$")
SHA256 = re.compile(r"^[a-f0-9]{64}$")
POLICY_FIELDS = {
    "schema_version", "environment_policy_version",
    "environment_policy_sha256", "window_id", "window_start", "window_end",
    "evaluator_revision", "authorized_stages", "allowed_edges",
    "allowed_paths", "required_sources", "asset_scope",
    "transition_population_source", "max_staleness_seconds",
    "max_future_skew_seconds",
}
TRANSITION_FIELDS = {
    "event_id", "asset_id", "source", "from_stage", "to_stage",
    "path", "observed_at",
}

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

def strict_json(raw):
    return json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=lambda token: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + token)
        ),
    )

def canonical(value):
    return json.dumps(
        value, sort_keys=True, separators=(",", ":"), ensure_ascii=False
    ).encode("utf-8")

def digest(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()

def name(value, label):
    if not isinstance(value, str) or not NAME.fullmatch(value):
        raise ValueError(label + " is not a bounded identifier")
    return value

def timestamp(value, label):
    if not isinstance(value, str) or not value:
        raise ValueError(label + " is absent")
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        raise ValueError(label + " lacks a timezone")
    return parsed

def numeric(value, label, positive):
    if (
        isinstance(value, bool) or not isinstance(value, (int, float))
        or not math.isfinite(float(value))
        or (float(value) <= 0 if positive else float(value) < 0)
    ):
        raise ValueError(label + " is invalid")
    return float(value)

files = [Path(value) for value in sys.argv[1:9]]
evaluated_at = sys.argv[9]
output = Path(sys.argv[10])
input_digests = {
    label: {
        "payload_sha256": digest(files[index]),
        "bundle_sha256": digest(files[index + 1]),
    }
    for label, index in (
        ("path_policy", 0), ("source_populations", 2),
        ("transition_population", 4), ("observed_transitions", 6),
    )
}

try:
    now = timestamp(evaluated_at, "evaluated_at").timestamp()
    policy = strict_json(files[0].read_bytes())
    if (
        not isinstance(policy, dict) or set(policy) != POLICY_FIELDS
        or policy.get("schema_version") != "aidefend.lifecycle-path-policy.v2"
        or policy.get("transition_population_source")
        != "signed_lifecycle_transition_ledger"
        or not SHA256.fullmatch(policy.get("environment_policy_sha256", ""))
    ):
        raise ValueError("signed lifecycle-path policy schema differs")
    for field in ("environment_policy_version", "window_id", "evaluator_revision"):
        name(policy[field], field)
    stale = numeric(policy["max_staleness_seconds"], "max_staleness_seconds", True)
    future = numeric(
        policy["max_future_skew_seconds"], "max_future_skew_seconds", False
    )
    window_start = timestamp(policy["window_start"], "window_start")
    window_end = timestamp(policy["window_end"], "window_end")
    if window_start >= window_end:
        raise ValueError("signed measurement window is not ordered")

    stages = policy["authorized_stages"]
    if not isinstance(stages, list) or not stages or len(stages) != len(set(stages)):
        raise ValueError("authorized stage set is invalid")
    for stage in stages:
        name(stage, "authorized_stage")
    stage_set = set(stages)
    edges = set()
    for number, row in enumerate(policy["allowed_edges"], 1):
        if not isinstance(row, dict) or set(row) != {"from", "to"}:
            raise ValueError(f"allowed_edge:{number}: schema differs")
        edge = (
            name(row["from"], f"allowed_edge:{number}:from"),
            name(row["to"], f"allowed_edge:{number}:to"),
        )
        if edge[0] == edge[1] or not set(edge) <= stage_set or edge in edges:
            raise ValueError(f"allowed_edge:{number}: invalid or duplicate")
        edges.add(edge)
    paths_allowed = set()
    for number, value in enumerate(policy["allowed_paths"], 1):
        if (
            not isinstance(value, list) or len(value) < 2
            or len(value) != len(set(value))
            or not all(isinstance(stage, str) and stage in stage_set for stage in value)
        ):
            raise ValueError(f"allowed_path:{number}: invalid")
        signature = tuple(value)
        if signature in paths_allowed or any(
            (left, right) not in edges for left, right in zip(value, value[1:])
        ):
            raise ValueError(f"allowed_path:{number}: undeclared edge or duplicate")
        paths_allowed.add(signature)

    required_sources = policy["required_sources"]
    if (
        not isinstance(required_sources, list) or not required_sources
        or len(required_sources) != len(set(required_sources))
    ):
        raise ValueError("required source set is invalid")
    for source in required_sources:
        name(source, "required_source")
    required_sources = set(required_sources)
    scope = {}
    for number, row in enumerate(policy["asset_scope"], 1):
        if not isinstance(row, dict) or set(row) != {"asset_id", "required_sources"}:
            raise ValueError(f"asset_scope:{number}: schema differs")
        asset = name(row["asset_id"], f"asset_scope:{number}:asset_id")
        sources = row["required_sources"]
        if (
            asset in scope or not isinstance(sources, list) or not sources
            or len(sources) != len(set(sources))
            or not set(sources) <= required_sources
        ):
            raise ValueError(f"asset_scope:{number}: invalid")
        scope[asset] = set(sources)
    if not scope:
        raise ValueError("asset scope is empty")

    evidence_gaps = []
    def note_freshness(value, label):
        age = now - timestamp(value, label).timestamp()
        if age > stale:
            evidence_gaps.append({"kind": "stale_evidence", "source": label})
        elif age < -future:
            evidence_gaps.append({"kind": "future_evidence", "source": label})

    note_freshness(policy["window_end"], "measurement_window")
    exports = strict_json(files[2].read_bytes())
    if not isinstance(exports, list):
        raise ValueError("source populations must be an array")
    by_source = {}
    membership = defaultdict(set)
    live_stage = {}
    for number, row in enumerate(exports, 1):
        if (
            not isinstance(row, dict)
            or set(row) != {"source", "observed_at", "population_sha256", "assets"}
        ):
            raise ValueError(f"source_population:{number}: schema differs")
        source = name(row["source"], f"source_population:{number}:source")
        if source in by_source or source not in required_sources:
            raise ValueError(f"source_population:{number}: duplicate or unexpected")
        assets = row["assets"]
        if (
            not isinstance(assets, list)
            or not SHA256.fullmatch(row.get("population_sha256", ""))
            or hashlib.sha256(canonical(assets)).hexdigest()
            != row["population_sha256"]
        ):
            raise ValueError(f"source_population:{source}: digest differs")
        note_freshness(row["observed_at"], "source_population:" + source)
        seen = set()
        for index, item in enumerate(assets, 1):
            if not isinstance(item, dict) or set(item) != {"asset_id", "stage"}:
                raise ValueError(f"source_population:{source}:{index}: schema differs")
            asset = name(item["asset_id"], f"source_population:{source}:{index}:asset")
            stage = name(item["stage"], f"source_population:{source}:{index}:stage")
            if asset in seen or asset not in scope or source not in scope[asset]:
                raise ValueError(f"source_population:{source}:{index}: out of scope")
            if stage not in stage_set:
                raise ValueError(f"source_population:{source}:{index}: unknown stage")
            seen.add(asset)
            membership[asset].add(source)
            live_stage[(asset, source)] = stage
        by_source[source] = row
    for source in sorted(required_sources - set(by_source)):
        evidence_gaps.append({"kind": "missing_source_population", "source": source})
    for asset, expected_sources in sorted(scope.items()):
        missing_sources = sorted(expected_sources - membership[asset])
        if missing_sources:
            evidence_gaps.append({
                "kind": "missing_asset_source_membership",
                "asset_id": asset,
                "sources": missing_sources,
            })

    population = strict_json(files[4].read_bytes())
    if (
        not isinstance(population, dict)
        or set(population) != {
            "schema_version", "window_id", "observed_at",
            "population_sha256", "transitions",
        }
        or population.get("schema_version")
        != "aidefend.lifecycle-transition-population.v2"
        or population.get("window_id") != policy["window_id"]
    ):
        raise ValueError("authoritative transition population schema differs")
    authoritative_rows = population["transitions"]
    if (
        not isinstance(authoritative_rows, list)
        or not SHA256.fullmatch(population.get("population_sha256", ""))
        or hashlib.sha256(canonical(authoritative_rows)).hexdigest()
        != population["population_sha256"]
    ):
        raise ValueError("authoritative transition population digest differs")
    note_freshness(population["observed_at"], "transition_population")

    def parse_transition(row, label):
        if not isinstance(row, dict) or set(row) != TRANSITION_FIELDS:
            raise ValueError(label + ": schema differs")
        event = name(row["event_id"], label + ":event_id")
        asset = name(row["asset_id"], label + ":asset_id")
        source = name(row["source"], label + ":source")
        before = name(row["from_stage"], label + ":from_stage")
        after = name(row["to_stage"], label + ":to_stage")
        path = row["path"]
        if (
            not isinstance(path, list) or len(path) < 2
            or not all(isinstance(stage, str) and NAME.fullmatch(stage) for stage in path)
        ):
            raise ValueError(label + ": path is malformed")
        observed = timestamp(row["observed_at"], label + ":observed_at")
        if not window_start <= observed < window_end:
            raise ValueError(label + ": event lies outside the signed window")
        return event, asset, source, before, after, tuple(path)

    authoritative = {}
    for number, row in enumerate(authoritative_rows, 1):
        parsed = parse_transition(row, f"authoritative_transition:{number}")
        if parsed[0] in authoritative:
            raise ValueError(f"authoritative_transition:{number}: duplicate event_id")
        authoritative[parsed[0]] = (row, parsed)

    lines = files[6].read_bytes().decode("utf-8", errors="strict").splitlines()
    if any(not line.strip() for line in lines):
        raise ValueError("observed transition JSONL contains a blank record")
    observed = {}
    for number, line in enumerate(lines, 1):
        row = strict_json(line.encode("utf-8"))
        parsed = parse_transition(row, f"observed_transition:{number}")
        if parsed[0] in observed:
            raise ValueError(f"observed_transition:{number}: duplicate event_id")
        observed[parsed[0]] = (row, parsed)

    missing = sorted(authoritative.keys() - observed.keys())
    extra = sorted(observed.keys() - authoritative.keys())
    mismatched = sorted(
        event for event in authoritative.keys() & observed.keys()
        if canonical(authoritative[event][0]) != canonical(observed[event][0])
    )
    if missing or extra or mismatched:
        evidence_gaps.append({
            "kind": "observer_population_mismatch", "missing": missing,
            "extra": extra, "mismatched": mismatched,
        })

    findings = []
    for event in sorted(authoritative):
        _, parsed = authoritative[event]
        _, asset, source, before, after, path = parsed
        reasons = []
        if asset not in scope:
            reasons.append("orphan_asset")
        elif source not in scope[asset]:
            reasons.append("unexpected_source")
        if before not in stage_set or after not in stage_set or not set(path) <= stage_set:
            reasons.append("unknown_stage")
        if path[0] != before or path[-1] != after:
            reasons.append("endpoint_binding")
        if len(path) != len(set(path)):
            reasons.append("cycle")
        if path not in paths_allowed or any(
            (left, right) not in edges for left, right in zip(path, path[1:])
        ):
            reasons.append("unauthorized_path")
        if asset in scope and source in scope[asset] and (asset, source) in live_stage:
            if live_stage[(asset, source)] != after:
                reasons.append("source_readback_mismatch")
        if reasons:
            findings.append({
                "event_id": event, "asset_id": asset, "source": source,
                "from_stage": before, "to_stage": after, "path": list(path),
                "reasons": sorted(set(reasons)),
            })

    status = (
        "FAIL" if findings else
        "INSUFFICIENT_DATA" if evidence_gaps else
        "PASS"
    )
    result_value = {
        "schema_version": "aidefend.lifecycle-path-findings.v2",
        "control": "AID-D-004.007",
        "window_id": policy["window_id"],
        "window_start": policy["window_start"],
        "window_end": policy["window_end"],
        "evaluated_at": evaluated_at,
        "evaluator_revision": policy["evaluator_revision"],
        "policy_sha256": digest(files[0]),
        "input_digests": input_digests,
        "expected_asset_population": len(scope),
        "authoritative_transition_population": len(authoritative),
        "observed_transition_population": len(observed),
        "finding_count": len(findings),
        "coverage_gap_count": len(evidence_gaps),
        "findings": findings,
        "coverage_gaps": evidence_gaps,
        "status": status,
    }
except Exception as error:
    result_value = {
        "schema_version": "aidefend.lifecycle-path-findings.v2",
        "control": "AID-D-004.007",
        "status": "ERROR",
        "status_reason": "verified-input parsing, reconciliation, or evaluation failed",
        "error_class": type(error).__name__,
        "evaluated_at": evaluated_at,
        "input_digests": input_digests,
    }

output.write_text(
    json.dumps(result_value, sort_keys=True, separators=(",", ":")) + "\n",
    encoding="utf-8",
)
raise SystemExit(2 if result_value["status"] == "ERROR" else 0)
PY
evaluation_rc=$?
set -e
test -s "$result"
sign_and_publish
status="$(python - "$result" <<'PY'
import json
import sys
from pathlib import Path

value = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8"))
status = value.get("status")
if status not in {"PASS", "FAIL", "INSUFFICIENT_DATA", "NOT_APPLICABLE", "ERROR"}:
    raise SystemExit("invalid result status")
print(status)
PY
)"
case "$status" in
  PASS|NOT_APPLICABLE) exit 0 ;;
  FAIL) exit 1 ;;
  ERROR) exit 2 ;;
  INSUFFICIENT_DATA) exit 3 ;;
esac
exit "$evaluation_rc"
Independent replay

A separate read-only identity captures and verifies the result bytes as below, then checks out the recorded evaluator revision, re-runs it over the same signed input digests and recorded evaluation time, and compares the canonical result. Exercise an allowed path, a signed empty window, an omitted observer event, an unlisted direct edge, a hidden intermediate, a cycle, an orphan asset, stale and missing source exports, duplicate IDs, changed payload/bundle bytes, and invalid JSON. Confirmed path violations are FAIL; incomplete or stale populations are INSUFFICIENT_DATA; verification, schema, or evaluator failures are ERROR.

set -euo pipefail
umask 077
snapshot="$(mktemp -d)"
trap 'rm -rf -- "$snapshot"' EXIT
for source in \
  evidence/lifecycle-path-findings.json \
  evidence/lifecycle-path-findings.bundle \
  keys/lifecycle-detector.pub; do
  test -f "$source" && test ! -L "$source"
done
install -m 0400 -- evidence/lifecycle-path-findings.json "$snapshot/result.json"
install -m 0400 -- evidence/lifecycle-path-findings.bundle "$snapshot/result.bundle"
install -m 0400 -- keys/lifecycle-detector.pub "$snapshot/lifecycle-detector.pub"
cosign verify-blob --key "$snapshot/lifecycle-detector.pub" \
  --bundle "$snapshot/result.bundle" "$snapshot/result.json"
python - "$snapshot/result.json" <<'PY'
import json
import sys
from pathlib import Path

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

result = json.loads(
    Path(sys.argv[1]).read_text(encoding="utf-8", errors="strict"),
    object_pairs_hook=unique_object,
    parse_constant=lambda token: (_ for _ in ()).throw(
        ValueError("non-finite JSON constant: " + token)
    ),
)
if (
    not isinstance(result, dict)
    or result.get("schema_version") != "aidefend.lifecycle-path-findings.v2"
    or result.get("control") != "AID-D-004.007"
    or result.get("status")
    not in {"PASS", "FAIL", "INSUFFICIENT_DATA", "NOT_APPLICABLE", "ERROR"}
):
    raise SystemExit("signed lifecycle-path result is invalid")
print(result["status"])
PY

Open-source tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-D-004.007: AI Lifecycle-Path Drift Detection | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service