Scope boundary

Owns verification of referenced prerequisite-result signatures and digests, dependency ordering, trusted recovery-workflow execution, assembled live-state readback, and cutover authorization for a multi-component recovery. Each component control retains ownership of its bounded technical result. This orchestrator does not reinterpret or replace another control's result. Missing components, unknown dependencies, cycles, mutable artifacts or workflows, unverifiable prerequisites, incomplete live populations, failed external probes, or unmet signed RTO/RPO objectives block cutover.

AID-R-004: AID-R-006 assembles and cuts over one dependency-ordered recovered system; AID-R-004 propagates an incident-proven remediation across the affected fleet and future baselines. R-006 may invoke component and remediation workflows, but it does not prove that every affected deployment or future build inherited the fix.

Implementation guidance

• AID-R-006-G001: Create an authority-signed recovery dependency manifest that binds each restored component, immutable artifact, trusted workflow, prerequisite control result, and dependency edge.
• AID-R-006-G002: Verify prerequisite control-result digests and execute trusted recovery workflow templates in validated dependency order, stopping before dependents on any mismatch.
• AID-R-006-G003: Read back the exact live component and external-probe populations through separate read-only adapters, recompute RTO/RPO, and sign the cutover authorization for the bound release.

AID-R-006-G001

When the dependency manifest applies

Use this path when the clean-room release contains multiple components or prerequisite control results whose order matters. The recovery authority derives one exact component population from the incident-bound recovery scope and signs the manifest. Each component binds one immutable artifact digest, one workflow-template digest from the trusted registry, exact predecessor component IDs, and the signed result digests required before execution.

Bind cutover prerequisites in the same manifest

{
  "schema_version": "aidefend.recovery-dependency-manifest/v1",
  "system_or_tenant": "example-system",
  "incident_id": "INC-EXAMPLE",
  "release_id": "recovery-example",
  "outage_started_at": "2026-07-10T17:30:00Z",
  "trusted_cutoff_at": "2026-07-10T17:25:00Z",
  "recovery_policy_sha256": "POLICY_SHA256",
  "component_population_sha256": "SORTED_COMPONENT_ID_POPULATION_SHA256",
  "components": [
    {
      "component_id": "model-primary",
      "kind": "model",
      "target_id": "inference-prod",
      "artifact_sha256": "MODEL_ARTIFACT_SHA256",
      "workflow_template_sha256": "RESTORE_WORKFLOW_SHA256",
      "depends_on": [],
      "prerequisites": [
        {"control_id": "AID-R-001.001", "result_sha256": "MODEL_RESULT_SHA256"}
      ]
    }
  ],
  "cutover_prerequisites": {
    "readiness_profile_sha256": "READINESS_PROFILE_SHA256",
    "required_probe_ids": ["external-inference-readiness"],
    "max_readback_staleness_seconds": 300,
    "max_future_skew_seconds": 30,
    "rto_target_seconds": 7200,
    "rpo_target_seconds": null,
    "rpo_relevant_component_ids": []
  }
}

RTO/RPO numbers are examples of schema placement only; production values come from the signed recovery policy. Artifact and workflow digests resolve only through preconfigured trusted registries. The manifest contains no commands, mutable tags, arbitrary URLs, credentials, public keys, or generic pass fields.

Validate exact population, DAG, and cutover inputs

# File: recovery/validate_dependency_manifest.py
from __future__ import annotations
import hashlib, json, math, os, re, subprocess, sys, tempfile
from datetime import datetime
from pathlib import Path
from typing import Any

SHA256 = re.compile(r"^[0-9a-f]{64}$")
CONTROL = re.compile(r"^AID-[A-Z]+-[0-9]{3}(?:\.[0-9]{3})?$")
MAXIMUM_MANIFEST_BYTES = int(os.environ["R006_MAXIMUM_SIGNED_DOCUMENT_BYTES"])
SIGNATURE_TIMEOUT_SECONDS = float(os.environ["R006_SIGNATURE_VERIFY_TIMEOUT_SECONDS"])
if (MAXIMUM_MANIFEST_BYTES <= 0 or not math.isfinite(SIGNATURE_TIMEOUT_SECONDS)
        or SIGNATURE_TIMEOUT_SECONDS <= 0):
    raise RuntimeError("versioned R006 runtime profile is invalid")


def reject_constant(value: str) -> None:
    raise ValueError(f"non-finite JSON number is forbidden: {value}")


def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    value: dict[str, Any] = {}
    for name, item in pairs:
        if name in value:
            raise ValueError(f"duplicate JSON key: {name}")
        value[name] = item
    return value


def strict_object(raw: bytes) -> dict[str, Any]:
    value = json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=reject_constant,
    )
    if not isinstance(value, dict):
        raise ValueError("manifest must contain one JSON object")
    return value


def read_limited(path: Path) -> bytes:
    with path.open("rb") as source:
        raw = source.read(MAXIMUM_MANIFEST_BYTES + 1)
    if len(raw) > MAXIMUM_MANIFEST_BYTES:
        raise ValueError(f"{path} exceeds the versioned runtime bound")
    return raw


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


def verified_manifest(path: Path, bundle: Path, public_key: str) -> tuple[dict[str, Any], bytes]:
    raw = read_limited(path)
    bundle_raw = read_limited(bundle)
    if not raw or not bundle_raw:
        raise ValueError("dependency manifest and Sigstore bundle must both be non-empty")
    with tempfile.TemporaryDirectory(prefix="r006-plan-") as directory:
        snapshot_root = Path(directory)
        os.chmod(snapshot_root, 0o700)
        payload_snapshot = snapshot_root / "dependency-manifest.json"
        bundle_snapshot = snapshot_root / "dependency-manifest.json.sigstore.json"
        write_private_snapshot(payload_snapshot, raw)
        write_private_snapshot(bundle_snapshot, bundle_raw)
        subprocess.run(
            ["cosign", "verify-blob", "--key", public_key, "--bundle", str(bundle_snapshot), str(payload_snapshot)],
            check=True, capture_output=True, text=True, timeout=SIGNATURE_TIMEOUT_SECONDS,
        )
        if read_limited(payload_snapshot) != raw:
            raise RuntimeError("verified dependency-manifest payload snapshot changed")
        if read_limited(bundle_snapshot) != bundle_raw:
            raise RuntimeError("verified dependency-manifest Sigstore bundle snapshot changed")
    return strict_object(raw), raw


if len(sys.argv) != 4:
    raise SystemExit("usage: validate_dependency_manifest.py MANIFEST BUNDLE PUBLIC_KEY")
manifest_path = Path(sys.argv[1])
manifest, manifest_bytes = verified_manifest(manifest_path, Path(sys.argv[2]), sys.argv[3])
root_fields = {
    "schema_version", "system_or_tenant", "incident_id", "release_id",
    "outage_started_at", "trusted_cutoff_at", "recovery_policy_sha256",
    "component_population_sha256", "components", "cutover_prerequisites",
}
if not isinstance(manifest, dict) or set(manifest) != root_fields:
    raise ValueError("recovery dependency manifest schema differs")
if manifest["schema_version"] != "aidefend.recovery-dependency-manifest/v1":
    raise ValueError("unsupported recovery dependency schema")
for field in ("recovery_policy_sha256", "component_population_sha256"):
    if not isinstance(manifest[field], str) or SHA256.fullmatch(manifest[field]) is None:
        raise ValueError(f"{field} must be SHA-256")
for field in ("outage_started_at", "trusted_cutoff_at"):
    value = datetime.fromisoformat(manifest[field].replace("Z", "+00:00"))
    if value.tzinfo is None:
        raise ValueError(f"{field} must include a timezone")

components = manifest["components"]
if not isinstance(components, list) or not components:
    raise ValueError("nonempty recovery component population required")
component_fields = {
    "component_id", "kind", "target_id", "artifact_sha256",
    "workflow_template_sha256", "depends_on", "prerequisites",
}
prerequisite_fields = {"control_id", "result_sha256"}
by_id = {}
for component in components:
    if not isinstance(component, dict) or set(component) != component_fields:
        raise ValueError("component schema differs")
    component_id = component["component_id"]
    if not isinstance(component_id, str) or not component_id or component_id in by_id:
        raise ValueError("component IDs must be nonempty and unique")
    if not isinstance(component["kind"], str) or not component["kind"]:
        raise ValueError(f"{component_id}: kind is required")
    if not isinstance(component["target_id"], str) or not component["target_id"]:
        raise ValueError(f"{component_id}: target ID is required")
    for field in ("artifact_sha256", "workflow_template_sha256"):
        if not isinstance(component[field], str) or SHA256.fullmatch(component[field]) is None:
            raise ValueError(f"{component_id}: invalid {field}")
    if not isinstance(component["depends_on"], list):
        raise ValueError(f"{component_id}: dependency list required")
    prerequisites = component["prerequisites"]
    if not isinstance(prerequisites, list) or not prerequisites:
        raise ValueError(f"{component_id}: prerequisite control result required")
    seen_controls = set()
    for prerequisite in prerequisites:
        if not isinstance(prerequisite, dict) or set(prerequisite) != prerequisite_fields:
            raise ValueError(f"{component_id}: prerequisite schema differs")
        control = prerequisite["control_id"]
        if not isinstance(control, str) or CONTROL.fullmatch(control) is None:
            raise ValueError(f"{component_id}: invalid prerequisite control")
        if control in seen_controls:
            raise ValueError(f"{component_id}: duplicate prerequisite control")
        seen_controls.add(control)
        digest = prerequisite["result_sha256"]
        if not isinstance(digest, str) or SHA256.fullmatch(digest) is None:
            raise ValueError(f"{component_id}: invalid prerequisite result digest")
    by_id[component_id] = component

population = json.dumps(sorted(by_id), separators=(",", ":")).encode("utf-8")
if hashlib.sha256(population).hexdigest() != manifest["component_population_sha256"]:
    raise ValueError("component population digest differs")
for component_id, component in by_id.items():
    dependencies = component["depends_on"]
    if (
        len(dependencies) != len(set(dependencies))
        or component_id in dependencies
        or any(dep not in by_id for dep in dependencies)
    ):
        raise ValueError(f"{component_id}: duplicate, self, or unknown dependency")

order, visiting, visited = [], set(), set()
def visit(component_id: str) -> None:
    if component_id in visiting:
        raise ValueError("recovery dependency cycle")
    if component_id in visited:
        return
    visiting.add(component_id)
    for dependency in sorted(by_id[component_id]["depends_on"]):
        visit(dependency)
    visiting.remove(component_id)
    visited.add(component_id)
    order.append(component_id)
for component_id in sorted(by_id):
    visit(component_id)

cutover = manifest["cutover_prerequisites"]
cutover_fields = {
    "readiness_profile_sha256", "required_probe_ids",
    "max_readback_staleness_seconds", "max_future_skew_seconds",
    "rto_target_seconds", "rpo_target_seconds", "rpo_relevant_component_ids",
}
if not isinstance(cutover, dict) or set(cutover) != cutover_fields:
    raise ValueError("cutover prerequisite schema differs")
if (
    not isinstance(cutover["readiness_profile_sha256"], str)
    or SHA256.fullmatch(cutover["readiness_profile_sha256"]) is None
):
    raise ValueError("readiness profile digest is invalid")
probe_ids = cutover["required_probe_ids"]
if (
    not isinstance(probe_ids, list)
    or not probe_ids
    or len(probe_ids) != len(set(probe_ids))
    or any(not isinstance(value, str) or not value for value in probe_ids)
):
    raise ValueError("required probe IDs must be nonempty and unique")
maximum_age = cutover["max_readback_staleness_seconds"]
future_skew = cutover["max_future_skew_seconds"]
if (isinstance(maximum_age, bool) or not isinstance(maximum_age, (int, float))
        or not math.isfinite(maximum_age) or maximum_age <= 0
        or isinstance(future_skew, bool) or not isinstance(future_skew, (int, float))
        or not math.isfinite(future_skew) or future_skew < 0):
    raise ValueError("readback freshness bounds are invalid")
rto = cutover["rto_target_seconds"]
if isinstance(rto, bool) or not isinstance(rto, (int, float)) or not math.isfinite(rto) or rto <= 0:
    raise ValueError("RTO target must be finite and positive")
rpo_ids = cutover["rpo_relevant_component_ids"]
if (
    not isinstance(rpo_ids, list)
    or len(rpo_ids) != len(set(rpo_ids))
    or any(value not in by_id for value in rpo_ids)
):
    raise ValueError("RPO-relevant component population differs")
rpo = cutover["rpo_target_seconds"]
if rpo_ids:
    if isinstance(rpo, bool) or not isinstance(rpo, (int, float)) or not math.isfinite(rpo) or rpo < 0:
        raise ValueError("RPO target must be finite and nonnegative")
elif rpo is not None:
    raise ValueError("RPO target must be null when no component is RPO-relevant")

result = {
    "schema_version": "aidefend.validated-recovery-order/v1",
    "manifest_sha256": hashlib.sha256(manifest_bytes).hexdigest(),
    "release_id": manifest["release_id"],
    "component_population_sha256": manifest["component_population_sha256"],
    "dependency_order": order,
    "cutover_prerequisites_sha256": hashlib.sha256(
        json.dumps(cutover, sort_keys=True, separators=(",", ":")).encode("utf-8")
    ).hexdigest(),
}
print(json.dumps(result, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False))

set -euo pipefail
set -o noclobber
umask 077
python recovery/validate_dependency_manifest.py \
  recovery/recovery-dependency-manifest.json \
  recovery/recovery-dependency-manifest.json.sigstore.json \
  keys/recovery-authority.pub \
  > recovery/validated-dependency-order.json
cosign sign-blob --yes --key env://RECOVERY_PLAN_VALIDATOR_KEY \
  --bundle recovery/validated-dependency-order.json.sigstore.json \
  recovery/validated-dependency-order.json
chmod 0400 recovery/validated-dependency-order.json recovery/validated-dependency-order.json.sigstore.json
order_sha256="$(sha256sum -- recovery/validated-dependency-order.json | awk '{print $1}')"
bundle_sha256="$(sha256sum -- recovery/validated-dependency-order.json.sigstore.json | awk '{print $1}')"
cosign verify-blob --key keys/recovery-plan-validator.pub \
  --bundle recovery/validated-dependency-order.json.sigstore.json \
  recovery/validated-dependency-order.json
test "$(sha256sum -- recovery/validated-dependency-order.json | awk '{print $1}')" = "$order_sha256"
test "$(sha256sum -- recovery/validated-dependency-order.json.sigstore.json | awk '{print $1}')" = "$bundle_sha256"

G002 consumes this exact sibling payload/bundle pair. Before execution it re-verifies each prerequisite result with the pinned key for its owning control and requires the canonical digest to equal the manifest. The manifest orders recovery work only; it does not replace the technical verification performed by any prerequisite control.

AID-R-006-G002

Execute only validated, trusted recovery work

G002 consumes the signed dependency manifest and signed validated order from G001. It resolves each workflow-template digest through a root-controlled registry; the manifest never supplies commands. Before a component runs, verify every referenced prerequisite result's exact bytes, digest, and native control signature from a fixed control-specific trust-root path and sibling .sigstore.json bundle. This verifies the referenced technical result without defining a universal cross-control receipt schema.

Stop before dependents on any mismatch

Run each digest-pinned template only after all predecessor components completed. The template receives the fixed manifest path and component ID through controlled environment variables and writes one strict component-local action result. An action result proves only that the named artifact and workflow were applied to the named target; G003 independently verifies the assembled live release.

# File: recovery/execute_dependency_order.py
from __future__ import annotations
import hashlib, json, math, os, re, stat, subprocess, tempfile
from pathlib import Path
from typing import Any

MANIFEST = Path("recovery/recovery-dependency-manifest.json")
ORDER = Path("recovery/validated-dependency-order.json")
WORKFLOWS = Path("/opt/aidefend/recovery/workflow-registry.json")
PREREQUISITES = Path("recovery/prerequisites")
RUN_RESULTS = Path("recovery/workflow-results")
OUTPUT = Path("recovery/orchestration-result.json")
WORKFLOW_ROOT = Path("/opt/aidefend/recovery/bin")
CONTROL = re.compile(r"^AID-[A-Z]+-[0-9]{3}(?:\.[0-9]{3})?$")
SHA256 = re.compile(r"^[0-9a-f]{64}$")
SIGNATURE_VERIFY_TIMEOUT_SECONDS = float(os.environ["R006_SIGNATURE_VERIFY_TIMEOUT_SECONDS"])
MAXIMUM_SIGNED_DOCUMENT_BYTES = int(os.environ["R006_MAXIMUM_SIGNED_DOCUMENT_BYTES"])
MAXIMUM_WORKFLOW_RESULT_BYTES = int(os.environ["R006_MAXIMUM_WORKFLOW_RESULT_BYTES"])
MAXIMUM_WORKFLOW_TEMPLATE_BYTES = int(os.environ["R006_MAXIMUM_WORKFLOW_TEMPLATE_BYTES"])
if (not math.isfinite(SIGNATURE_VERIFY_TIMEOUT_SECONDS) or SIGNATURE_VERIFY_TIMEOUT_SECONDS <= 0
        or MAXIMUM_SIGNED_DOCUMENT_BYTES <= 0
        or MAXIMUM_WORKFLOW_RESULT_BYTES <= 0
        or MAXIMUM_WORKFLOW_TEMPLATE_BYTES <= 0):
    raise RuntimeError("versioned R006 runtime profile is invalid")
SNAPSHOT_DIRECTORY_HANDLE = tempfile.TemporaryDirectory(prefix="r006-execute-")
SNAPSHOT_DIRECTORY = Path(SNAPSHOT_DIRECTORY_HANDLE.name)
os.chmod(SNAPSHOT_DIRECTORY, 0o700)

def reject_constant(value: str) -> None:
    raise ValueError(f"non-finite JSON number is forbidden: {value}")


def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    value: dict[str, Any] = {}
    for name, item in pairs:
        if name in value:
            raise ValueError(f"duplicate JSON key: {name}")
        value[name] = item
    return value


def strict_object(raw: bytes, label: str) -> dict[str, Any]:
    value = json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=reject_constant,
    )
    if not isinstance(value, dict):
        raise ValueError(f"{label} must contain one JSON object")
    return value


def read_limited(path: Path, maximum: int) -> bytes:
    with path.open("rb") as source:
        raw = source.read(maximum + 1)
    if len(raw) > maximum:
        raise ValueError(f"{path} exceeds the versioned runtime bound")
    return raw


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


def read_trusted_workflow(path: Path, maximum: int) -> bytes:
    if path.parent != WORKFLOW_ROOT or re.fullmatch(
        r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}", path.name
    ) is None:
        raise ValueError("workflow executable must be a direct safe-name member of the trusted root")
    nofollow = getattr(os, "O_NOFOLLOW", None)
    if nofollow is None:
        raise RuntimeError("workflow verification requires O_NOFOLLOW")
    descriptor = os.open(
        path, os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | nofollow
    )
    try:
        before = os.fstat(descriptor)
        if (
            not stat.S_ISREG(before.st_mode)
            or before.st_size < 1
            or before.st_size > maximum
            or before.st_uid != 0
            or stat.S_IMODE(before.st_mode) & 0o022
        ):
            raise PermissionError("workflow executable type, size, owner, or mode is unsafe")
        chunks: list[bytes] = []
        remaining = before.st_size
        while remaining:
            chunk = os.read(descriptor, min(1_048_576, remaining))
            if not chunk:
                raise OSError("short workflow executable read")
            chunks.append(chunk)
            remaining -= len(chunk)
        if os.read(descriptor, 1):
            raise OSError("workflow executable grew during read")
        after = os.fstat(descriptor)
        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):
            raise OSError("workflow executable changed during capture")
        return b"".join(chunks)
    finally:
        os.close(descriptor)


def snapshot_workflow(path: Path, expected_sha256: str) -> Path:
    raw = read_trusted_workflow(path, MAXIMUM_WORKFLOW_TEMPLATE_BYTES)
    if hashlib.sha256(raw).hexdigest() != expected_sha256:
        raise RuntimeError("workflow executable digest differs from the registry")
    snapshot = SNAPSHOT_DIRECTORY / ("workflow-" + expected_sha256)
    write_private_snapshot(snapshot, raw)
    snapshot.chmod(0o500)
    if read_limited(snapshot, MAXIMUM_WORKFLOW_TEMPLATE_BYTES) != raw:
        raise RuntimeError("workflow executable snapshot readback differs")
    return snapshot


def verified_bytes(
    payload: Path,
    bundle: Path,
    key: Path,
    snapshot_name: str,
) -> tuple[bytes, Path]:
    raw = read_limited(payload, MAXIMUM_SIGNED_DOCUMENT_BYTES)
    bundle_raw = read_limited(bundle, MAXIMUM_SIGNED_DOCUMENT_BYTES)
    if not raw or not bundle_raw:
        raise ValueError("signed payload and Sigstore bundle must both be non-empty")
    snapshot_root = Path(tempfile.mkdtemp(prefix="pair-", dir=SNAPSHOT_DIRECTORY))
    os.chmod(snapshot_root, 0o700)
    payload_snapshot = snapshot_root / snapshot_name
    bundle_snapshot = snapshot_root / (snapshot_name + ".sigstore.json")
    write_private_snapshot(payload_snapshot, raw)
    write_private_snapshot(bundle_snapshot, bundle_raw)
    subprocess.run(
        ["cosign", "verify-blob", "--key", str(key), "--bundle", str(bundle_snapshot), str(payload_snapshot)],
        check=True, capture_output=True, text=True, timeout=SIGNATURE_VERIFY_TIMEOUT_SECONDS,
    )
    if read_limited(payload_snapshot, MAXIMUM_SIGNED_DOCUMENT_BYTES) != raw:
        raise RuntimeError("verified evidence payload snapshot changed")
    if read_limited(bundle_snapshot, MAXIMUM_SIGNED_DOCUMENT_BYTES) != bundle_raw:
        raise RuntimeError("verified evidence Sigstore bundle snapshot changed")
    return raw, payload_snapshot


def verified_object(
    payload: Path,
    bundle: Path,
    key: Path,
    snapshot_name: str,
) -> tuple[dict[str, Any], bytes, Path]:
    raw, snapshot = verified_bytes(payload, bundle, key, snapshot_name)
    return strict_object(raw, str(payload)), raw, snapshot

manifest, manifest_bytes, manifest_snapshot = verified_object(
    MANIFEST, Path(str(MANIFEST) + ".sigstore.json"), Path("keys/recovery-authority.pub"), "dependency-manifest.json"
)
order, order_bytes, _ = verified_object(
    ORDER, Path(str(ORDER) + ".sigstore.json"), Path("keys/recovery-plan-validator.pub"), "validated-order.json"
)
registry, registry_bytes, _ = verified_object(
    WORKFLOWS, Path(str(WORKFLOWS) + ".sigstore.json"), Path("keys/recovery-platform.pub"), "workflow-registry.json"
)
if (
    order.get("schema_version") != "aidefend.validated-recovery-order/v1"
    or order.get("manifest_sha256") != hashlib.sha256(manifest_bytes).hexdigest()
    or order.get("release_id") != manifest.get("release_id")
    or order.get("component_population_sha256")
       != manifest.get("component_population_sha256")
):
    raise ValueError("validated dependency order is misbound")
components = {row["component_id"]: row for row in manifest["components"]}
dependency_order = order.get("dependency_order")
if (
    not isinstance(dependency_order, list)
    or len(dependency_order) != len(set(dependency_order))
    or set(dependency_order) != set(components)
):
    raise ValueError("validated dependency order population differs")
if (
    not isinstance(registry, dict)
    or registry.get("schema_version") != "aidefend.recovery-workflows/v1"
    or not isinstance(registry.get("workflows"), list)
):
    raise ValueError("trusted workflow registry schema differs")
workflow_by_digest = {}
for workflow in registry["workflows"]:
    if set(workflow) != {"workflow_template_sha256", "kind", "argv", "timeout_seconds"}:
        raise ValueError("workflow registry entry schema differs")
    digest = workflow["workflow_template_sha256"]
    argv = workflow["argv"]
    timeout_seconds = workflow["timeout_seconds"]
    if (
        not isinstance(digest, str)
        or SHA256.fullmatch(digest) is None
        or digest in workflow_by_digest
        or not isinstance(argv, list)
        or not argv
        or any(not isinstance(value, str) or not value for value in argv)
        or isinstance(timeout_seconds, bool)
        or not isinstance(timeout_seconds, (int, float))
        or not math.isfinite(timeout_seconds)
        or timeout_seconds <= 0
    ):
        raise ValueError("workflow registry entry is unsafe or duplicate")
    executable_snapshot = snapshot_workflow(Path(argv[0]), digest)
    workflow_by_digest[digest] = {
        **workflow,
        "argv": [str(executable_snapshot), *argv[1:]],
    }

RUN_RESULTS.mkdir(mode=0o700)
completed = set()
results = []
for component_id in dependency_order:
    component = components[component_id]
    if any(dependency not in completed for dependency in component["depends_on"]):
        raise RuntimeError(f"{component_id}: dependency was not completed")
    for prerequisite in component["prerequisites"]:
        control = prerequisite["control_id"]
        digest = prerequisite["result_sha256"]
        if CONTROL.fullmatch(control) is None or SHA256.fullmatch(digest) is None:
            raise ValueError(f"{component_id}: invalid prerequisite identity")
        payload = PREREQUISITES / control / f"{digest}.json"
        bundle = PREREQUISITES / control / f"{digest}.json.sigstore.json"
        key = Path("keys/prerequisites") / f"{control}.pub"
        prerequisite_bytes, _ = verified_bytes(
            payload, bundle, key, f"prerequisite-{digest}.json"
        )
        if hashlib.sha256(prerequisite_bytes).hexdigest() != digest:
            raise RuntimeError(f"{component_id}: prerequisite digest differs")
    workflow = workflow_by_digest.get(component["workflow_template_sha256"])
    if workflow is None or workflow["kind"] != component["kind"]:
        raise RuntimeError(f"{component_id}: trusted workflow is unavailable")
    result_path = RUN_RESULTS / f"{component_id}.json"
    executable_snapshot = Path(workflow["argv"][0])
    measured_workflow_sha256 = hashlib.sha256(
        read_limited(executable_snapshot, MAXIMUM_WORKFLOW_TEMPLATE_BYTES)
    ).hexdigest()
    if measured_workflow_sha256 != component["workflow_template_sha256"]:
        raise RuntimeError(f"{component_id}: measured workflow executable digest differs")
    environment = {
        "PATH": "/usr/sbin:/usr/bin:/sbin:/bin",
        "LANG": "C",
        "LC_ALL": "C",
        "TZ": "UTC",
        "AIDEFEND_RECOVERY_MANIFEST": str(manifest_snapshot),
        "AIDEFEND_RECOVERY_COMPONENT_ID": component_id,
        "AIDEFEND_RECOVERY_RESULT_PATH": str(result_path),
        "AIDEFEND_RECOVERY_WORKFLOW_TEMPLATE_SHA256": measured_workflow_sha256,
    }
    subprocess.run(
        workflow["argv"], check=True, env=environment,
        timeout=float(workflow["timeout_seconds"]),
    )
    if hashlib.sha256(
        read_limited(executable_snapshot, MAXIMUM_WORKFLOW_TEMPLATE_BYTES)
    ).hexdigest() != measured_workflow_sha256:
        raise RuntimeError(f"{component_id}: workflow executable snapshot changed during execution")
    result_bytes = read_limited(result_path, MAXIMUM_WORKFLOW_RESULT_BYTES)
    result = strict_object(result_bytes, f"workflow result {component_id}")
    result_fields = {
        "component_id", "target_id", "artifact_sha256",
        "workflow_template_sha256", "run_id", "completed_at", "result",
    }
    if (
        not isinstance(result, dict)
        or set(result) != result_fields
        or result["component_id"] != component_id
        or result["target_id"] != component["target_id"]
        or result["artifact_sha256"] != component["artifact_sha256"]
        or result["workflow_template_sha256"] != measured_workflow_sha256
        or result["result"] != "APPLIED"
    ):
        raise RuntimeError(f"{component_id}: workflow result differs")
    results.append({
        "component_id": component_id,
        "workflow_template_sha256": measured_workflow_sha256,
        "workflow_result_sha256": hashlib.sha256(result_bytes).hexdigest(),
        "run_id": result["run_id"],
        "completed_at": result["completed_at"],
    })
    completed.add(component_id)

output = {
    "schema_version": "aidefend.recovery-orchestration-result/v1",
    "manifest_sha256": hashlib.sha256(manifest_bytes).hexdigest(),
    "release_id": manifest["release_id"],
    "component_population_sha256": manifest["component_population_sha256"],
    "dependency_order": dependency_order,
    "component_results": results,
}
with OUTPUT.open("x", encoding="utf-8") as handle:
    json.dump(output, handle, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False)
    handle.write("\n")

set -euo pipefail
umask 077
python recovery/execute_dependency_order.py
cosign sign-blob --yes --key env://RECOVERY_ORCHESTRATOR_KEY \
  --bundle recovery/orchestration-result.json.sigstore.json \
  recovery/orchestration-result.json
cosign verify-blob --key keys/recovery-orchestrator.pub \
  --bundle recovery/orchestration-result.json.sigstore.json \
  recovery/orchestration-result.json
chmod 0400 recovery/orchestration-result.json recovery/orchestration-result.json.sigstore.json
result_sha256="$(sha256sum -- recovery/orchestration-result.json | awk '{print $1}')"
bundle_sha256="$(sha256sum -- recovery/orchestration-result.json.sigstore.json | awk '{print $1}')"
cosign verify-blob --key keys/recovery-orchestrator.pub \
  --bundle recovery/orchestration-result.json.sigstore.json \
  recovery/orchestration-result.json
test "$(sha256sum -- recovery/orchestration-result.json | awk '{print $1}')" = "$result_sha256"
test "$(sha256sum -- recovery/orchestration-result.json.sigstore.json | awk '{print $1}')" = "$bundle_sha256"

Action: execute the exact validated DAG through trusted workflow templates, verify every prerequisite result locally, stop before dependents on any bundle verification, digest, workflow, or component-result error, and sign the exact orchestration result consumed by G003.

AID-R-006-G003

Independently read back the assembled live release

A readiness observer that cannot execute recovery workflows queries every live component through its authoritative runtime or control-plane adapter and signs one exact population. Each row binds the component and target IDs, live artifact digest, observation time, readiness state, and recovery point where RPO applies. External probes run from outside the recovery environment and bind their observations to the exact release ID.

Recompute cutover prerequisites and recovery objectives

G003 consumes the same signed dependency manifest as G001/G002 plus the signed orchestration result from G002. It requires exact component and probe populations, compares every live target and artifact digest, recomputes RTO from the signed outage start, and computes worst-component RPO only for the manifest's rpo_relevant_component_ids. Missing live identity, unavailable probe, stale or unverifiable input, or an unmet signed objective prevents a cutover receipt.

# File: recovery/verify_cutover.py
from __future__ import annotations
import hashlib, json, math, os, subprocess, tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

MANIFEST = Path("recovery/recovery-dependency-manifest.json")
ORCHESTRATION = Path("recovery/orchestration-result.json")
READBACK = Path("recovery/live-readback.json")
OUTPUT = Path("recovery/cutover-receipt.json")
SIGNATURE_VERIFY_TIMEOUT_SECONDS = float(os.environ["R006_SIGNATURE_VERIFY_TIMEOUT_SECONDS"])
MAXIMUM_SIGNED_DOCUMENT_BYTES = int(os.environ["R006_MAXIMUM_SIGNED_DOCUMENT_BYTES"])
if (not math.isfinite(SIGNATURE_VERIFY_TIMEOUT_SECONDS) or SIGNATURE_VERIFY_TIMEOUT_SECONDS <= 0
        or MAXIMUM_SIGNED_DOCUMENT_BYTES <= 0):
    raise RuntimeError("versioned R006 runtime profile is invalid")

def reject_constant(value: str) -> None:
    raise ValueError(f"non-finite JSON number is forbidden: {value}")


def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    value: dict[str, Any] = {}
    for name, item in pairs:
        if name in value:
            raise ValueError(f"duplicate JSON key: {name}")
        value[name] = item
    return value


def strict_object(raw: bytes, label: str) -> dict[str, Any]:
    value = json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=reject_constant,
    )
    if not isinstance(value, dict):
        raise ValueError(f"{label} must contain one JSON object")
    return value


def read_limited(path: Path) -> bytes:
    with path.open("rb") as source:
        raw = source.read(MAXIMUM_SIGNED_DOCUMENT_BYTES + 1)
    if len(raw) > MAXIMUM_SIGNED_DOCUMENT_BYTES:
        raise ValueError(f"{path} exceeds the versioned runtime bound")
    return raw


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


def verified_object(payload: Path, bundle: Path, key: Path) -> tuple[dict[str, Any], bytes]:
    raw = read_limited(payload)
    bundle_raw = read_limited(bundle)
    if not raw or not bundle_raw:
        raise ValueError("signed payload and Sigstore bundle must both be non-empty")
    with tempfile.TemporaryDirectory(prefix="r006-cutover-") as directory:
        snapshot_root = Path(directory)
        os.chmod(snapshot_root, 0o700)
        payload_snapshot = snapshot_root / "signed-input.json"
        bundle_snapshot = snapshot_root / "signed-input.json.sigstore.json"
        write_private_snapshot(payload_snapshot, raw)
        write_private_snapshot(bundle_snapshot, bundle_raw)
        subprocess.run(
            ["cosign", "verify-blob", "--key", str(key), "--bundle", str(bundle_snapshot), str(payload_snapshot)],
            check=True, capture_output=True, text=True, timeout=SIGNATURE_VERIFY_TIMEOUT_SECONDS,
        )
        if read_limited(payload_snapshot) != raw:
            raise RuntimeError("verified cutover-input payload snapshot changed")
        if read_limited(bundle_snapshot) != bundle_raw:
            raise RuntimeError("verified cutover-input Sigstore bundle snapshot changed")
    return strict_object(raw, str(payload)), raw


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


def fresh(value: object, field: str, now: datetime, maximum_age: float, future_skew: float) -> datetime:
    parsed = time_value(value, field)
    age = (now - parsed).total_seconds()
    if age > maximum_age or age < -future_skew:
        raise ValueError(f"{field} is stale or future-dated")
    return parsed

manifest, manifest_bytes = verified_object(
    MANIFEST, Path(str(MANIFEST) + ".sigstore.json"), Path("keys/recovery-authority.pub")
)
orchestration, orchestration_bytes = verified_object(
    ORCHESTRATION, Path(str(ORCHESTRATION) + ".sigstore.json"), Path("keys/recovery-orchestrator.pub")
)
readback, readback_bytes = verified_object(
    READBACK, Path(str(READBACK) + ".sigstore.json"), Path("keys/recovery-readiness-observer.pub")
)
manifest_digest = hashlib.sha256(manifest_bytes).hexdigest()
if (
    orchestration.get("schema_version")
       != "aidefend.recovery-orchestration-result/v1"
    or orchestration.get("manifest_sha256") != manifest_digest
    or orchestration.get("release_id") != manifest["release_id"]
    or orchestration.get("component_population_sha256")
       != manifest["component_population_sha256"]
):
    raise ValueError("orchestration result is misbound")
readback_fields = {
    "schema_version", "manifest_sha256", "orchestration_result_sha256",
    "release_id", "readiness_profile_sha256", "ready_at",
    "components", "probes",
}
if (
    not isinstance(readback, dict)
    or set(readback) != readback_fields
    or readback["schema_version"] != "aidefend.recovery-live-readback/v1"
    or readback["manifest_sha256"] != manifest_digest
    or readback["orchestration_result_sha256"] != hashlib.sha256(orchestration_bytes).hexdigest()
    or readback["release_id"] != manifest["release_id"]
):
    raise ValueError("live readback schema or release binding differs")
cutover = manifest["cutover_prerequisites"]
if readback["readiness_profile_sha256"] != cutover["readiness_profile_sha256"]:
    raise ValueError("readiness profile digest differs")
maximum_age = cutover.get("max_readback_staleness_seconds")
future_skew = cutover.get("max_future_skew_seconds")
if (isinstance(maximum_age, bool) or not isinstance(maximum_age, (int, float))
        or not math.isfinite(maximum_age) or maximum_age <= 0
        or isinstance(future_skew, bool) or not isinstance(future_skew, (int, float))
        or not math.isfinite(future_skew) or future_skew < 0):
    raise ValueError("signed readback freshness bounds are invalid")
now = datetime.now(timezone.utc)

expected = {row["component_id"]: row for row in manifest["components"]}
orchestration_rows = orchestration.get("component_results")
if not isinstance(orchestration_rows, list) or not orchestration_rows:
    raise ValueError("orchestration component-result population is empty")
orchestrated = {row.get("component_id"): row for row in orchestration_rows}
if (
    len(orchestrated) != len(orchestration_rows)
    or set(orchestrated) != set(expected)
    or orchestration.get("dependency_order") != [
        row["component_id"] for row in orchestration_rows
    ]
):
    raise ValueError("orchestration component population or order differs")
observed_rows = readback["components"]
if not isinstance(observed_rows, list) or not observed_rows:
    raise ValueError("live component population is empty")
observed = {row.get("component_id"): row for row in observed_rows}
if len(observed) != len(observed_rows) or set(observed) != set(expected):
    raise ValueError("live component population differs")
component_fields = {
    "component_id", "target_id", "artifact_sha256",
    "observed_at", "ready", "recovery_point_at",
}
for component_id, wanted in expected.items():
    actual = observed[component_id]
    if (
        not isinstance(actual, dict)
        or set(actual) != component_fields
        or actual["target_id"] != wanted["target_id"]
        or actual["artifact_sha256"] != wanted["artifact_sha256"]
        or actual["ready"] is not True
    ):
        raise RuntimeError(f"{component_id}: live identity or readiness differs")
    fresh(actual["observed_at"], f"{component_id}.observed_at", now, maximum_age, future_skew)

expected_probes = set(cutover["required_probe_ids"])
probe_rows = readback["probes"]
if not isinstance(probe_rows, list) or not probe_rows:
    raise ValueError("external probe population is empty")
probe_fields = {"probe_id", "release_id", "success", "observed_at"}
probes = {row.get("probe_id"): row for row in probe_rows}
if len(probes) != len(probe_rows) or set(probes) != expected_probes:
    raise ValueError("external probe population differs")
for probe_id, probe in probes.items():
    if (
        not isinstance(probe, dict)
        or set(probe) != probe_fields
        or probe["release_id"] != manifest["release_id"]
        or probe["success"] is not True
    ):
        raise RuntimeError(f"{probe_id}: external readiness failed")
    fresh(probe["observed_at"], f"{probe_id}.observed_at", now, maximum_age, future_skew)

outage_started = time_value(manifest["outage_started_at"], "outage_started_at")
ready_at = fresh(readback["ready_at"], "ready_at", now, maximum_age, future_skew)
if ready_at < outage_started:
    raise ValueError("ready_at precedes outage start")
rto_seconds = (ready_at - outage_started).total_seconds()
if rto_seconds > cutover["rto_target_seconds"]:
    raise RuntimeError("signed RTO target was not met")

rpo_ids = cutover["rpo_relevant_component_ids"]
rpo_seconds = None
if rpo_ids:
    losses = []
    for component_id in rpo_ids:
        recovery_point = time_value(
            observed[component_id]["recovery_point_at"],
            f"{component_id}.recovery_point_at",
        )
        if recovery_point > outage_started:
            raise ValueError("recovery point occurs after outage start")
        losses.append((outage_started - recovery_point).total_seconds())
    rpo_seconds = max(losses)
    if rpo_seconds > cutover["rpo_target_seconds"]:
        raise RuntimeError("signed RPO target was not met")

receipt = {
    "schema_version": "aidefend.recovery-cutover-receipt/v1",
    "manifest_sha256": manifest_digest,
    "orchestration_result_sha256": hashlib.sha256(orchestration_bytes).hexdigest(),
    "live_readback_sha256": hashlib.sha256(readback_bytes).hexdigest(),
    "release_id": manifest["release_id"],
    "component_population_sha256": manifest["component_population_sha256"],
    "component_ids": sorted(expected),
    "probe_ids": sorted(expected_probes),
    "ready_at": readback["ready_at"],
    "rto_seconds": rto_seconds,
    "rto_target_seconds": cutover["rto_target_seconds"],
    "rpo_seconds": rpo_seconds,
    "rpo_target_seconds": cutover["rpo_target_seconds"],
    "cutover_authorized": True,
}
with OUTPUT.open("x", encoding="utf-8") as handle:
    json.dump(receipt, handle, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False)
    handle.write("\n")

set -euo pipefail
umask 077
python recovery/verify_cutover.py
cosign sign-blob --yes --key env://RECOVERY_CUTOVER_SIGNING_KEY \
  --bundle recovery/cutover-receipt.json.sigstore.json \
  recovery/cutover-receipt.json
cosign verify-blob --key keys/recovery-cutover.pub \
  --bundle recovery/cutover-receipt.json.sigstore.json \
  recovery/cutover-receipt.json
chmod 0400 recovery/cutover-receipt.json recovery/cutover-receipt.json.sigstore.json
receipt_sha256="$(sha256sum -- recovery/cutover-receipt.json | awk '{print $1}')"
bundle_sha256="$(sha256sum -- recovery/cutover-receipt.json.sigstore.json | awk '{print $1}')"
cosign verify-blob --key keys/recovery-cutover.pub \
  --bundle recovery/cutover-receipt.json.sigstore.json \
  recovery/cutover-receipt.json
test "$(sha256sum -- recovery/cutover-receipt.json | awk '{print $1}')" = "$receipt_sha256"
test "$(sha256sum -- recovery/cutover-receipt.json.sigstore.json | awk '{print $1}')" = "$bundle_sha256"

Action: authorize traffic only after an independently signed live readback proves the exact component population and external probes for the release, the verifier recomputes RTO/RPO against the manifest's signed targets, and the traffic controller independently verifies the cutover payload/bundle pair, release binding, freshness, and target environment.

Open-source tools

• Argo Workflows
• Argo CD (Kubernetes GitOps deployment and desired-versus-live resource readback; external service readiness remains separate)
• Temporal
• Tekton
• OpenTofu
• Cosign
• Open Policy Agent (OPA)

Source-available / open-weight tools

• HashiCorp Terraform Community (BUSL-1.1; source-available)

Commercial / hosted tools

• AWS Step Functions
• Azure Durable Functions
• Google Cloud Workflows
• HCP Terraform
• Terraform Enterprise

Threat mappings

MITRE ATLAS

• AML.T0010 AI Supply Chain Compromise (signed whole-system redeployment restores verified components after supply-chain compromise)
• AML.T0076 Corrupt AI Model (orchestration requires verified model restoration before service readiness)
• AML.T0101 Data Destruction via AI Agent Tool Invocation (dependency-ordered data and service recovery restores operation after destructive agent actions)

MAESTRO

• Supply Chain Attacks (Cross-Layer) (clean-room redeployment restores a verified cross-layer release)
• Compromised Container Images (L4) (runtime redeployment is pinned to an approved image digest)
• Compromised Agents (L7) (clean identity, state, prompt, tool, and routing bundles prevent redeploying the compromised agent state)

OWASP LLM Top 10 2026

• LLM04:2026 Supply Chain (signed dependency manifests prevent recovery from unverified components or workflows)

OWASP ML Top 10 2023

• ML06:2023 AI Supply Chain Attacks (clean-room deployment restores only digest-pinned approved components)
• ML10:2023 Model Poisoning (model restoration must pass before the recovered service is promoted)

OWASP Top 10 for Agentic Applications 2026

• ASI04:2026 Agentic Supply Chain Vulnerabilities (redeployment verifies prompt, tool, MCP, routing, runtime, and model release assets)
• ASI06:2026 Memory & Context Poisoning (redeployment requires purged state and verified context stores)
• ASI10:2026 Rogue Agents (clean identity and configuration promotion prevents the rogue agent instance from being restored)

NIST Adversarial Machine Learning 2025

• NISTAML.026 Model Poisoning (Integrity) (verified model recovery is a mandatory redeployment dependency)
• NISTAML.039 Compromising connected resources (clean credentials and integrated-component bundles are required before reconnection)
• NISTAML.051 Model Poisoning (Supply Chain) (signed manifest and receipts prevent restoring a compromised model supply chain)

Cisco Integrated AI Security and Safety Framework

• N/A

Google Secure AI Framework 2.0 - Risks

• MST: Model Source Tampering (redeployment accepts only the verified model restoration receipt and digest)
• MDT: Model Deployment Tampering (signed release orchestration protects the deployment cutover)
• IIC: Insecure Integrated Component (tool, MCP, routing, and runtime bundles are restored by verified digest)
• RA: Rogue Actions (old identities and state remain denied while a clean replacement is progressively re-enabled)

Databricks AI Security Framework 3.0

• Model 7.3: ML Supply chain vulnerabilities (clean-room release orchestration restores verified supply-chain artifacts)
• Model 7.4: Source code control attack (signed manifest pins the approved code and runtime release)
• Agents - Core 13.1: Memory Poisoning (purged and restored state is a required readiness dependency)
• Agents - Core 13.13: Rogue Agents in Multi-Agent Systems (clean identities and signed agent assets prevent restoring rogue instances)
• Agents - Tools MCP Server 13.20: Insecure Server Configuration (MCP server configuration is restored from a verified release bundle)
• Platform 12.3: Lack of incident response (signed end-to-end recovery orchestration and readiness evidence close the response lifecycle)