Scope boundary

Owns independent validation and, when required, fresh step-up or human approval for a high-impact immutable action plan, including local verification of identity, action digest, signatures, evidence, expiry, policy, one-time use, and the canonical signed gate receipt.

AID-M-006.001: AID-H-018.003 executes the runtime validation and approval gate; AID-M-006.001 defines which checkpoint exists, what evidence the operator sees, and which procedure applies. Checkpoint design cannot issue the runtime receipt.

AID-M-006.003: AID-H-018.003 issues the high-impact gate decision and receipt; AID-M-006.003 defines escalation routing, deadlines, event semantics, producer obligations, and lifecycle reconciliation. The approval decision and the audit or escalation contract have separate owners.

AID-H-018.002: AID-H-018.003 provides independent high-impact validation and approval; AID-H-018.002 provides general policy-based authorization for tool and resource access. A policy allow may still require a separate high-impact approval receipt.

AID-H-018.006: AID-H-018.003 approves the exact immutable action plan; AID-H-018.006 freshly re-verifies authorization, approval nonce, resource preconditions, and context immediately before dispatch. Approval can become stale or misbound before use.

AID-D-015: AID-H-018.003 enforces and records approval; AID-D-015 independently detects missing, expired, replayed, mismatched, unattended, or anomalous approval activity. The gate cannot be the sole detector of its own bypass.

AID-I-008.002: AID-H-018.003 issues the canonical approval receipt; AID-I-008.002 suspends, re-observes, binds, and resumes the exact browser action while consuming that receipt. The browser adapter adds observation and suspension but does not create a second approval authority.

AID-H-021.005: AID-H-018.003 authorizes one exact high-impact action plan after identity, policy, freshness, and approval checks; AID-H-021.005 decides whether an AI development workspace may activate its declared execution-capable surfaces. Workspace trust is not transaction approval, and a trusted workspace cannot bypass the per-action gate.

Implementation guidance

• AID-H-018.003-G001: Verify the exact canonical action, current signed policy and identity registry, typed signed evidence, and policy-selected multi-party approvals; atomically consume approval nonces and issue a signed, replayable outcome for every gate evaluation.
• AID-H-018.003-G002: Optionally use a separately invoked semantic reviewer as a typed negative signal before a high-impact action, while the deterministic gate retains final authority and all required human approval remains mandatory.

AID-H-018.003-G001

Complete deployment unit: canonical high-impact gate

Run this service under a control-plane identity that the agent and caller cannot assume. The caller supplies only a request ID. Trust roots, verification keys, the signed policy and identity registry, the Redis socket, signing keys, inbox, and evidence outbox are fixed on root-owned mounts; no request field or command-line option can select a verification key. The action plan itself is accepted only when its bytes are the exact canonical representation hashed by the gate.

# File: action_gate/issue_h018_gate_decision.py
from __future__ import annotations

import hashlib
import json
import math
import os
import re
import secrets
import subprocess
import tempfile
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any

import redis


def required_positive_seconds(name: str) -> float:
    value = float(os.environ[name])
    if not math.isfinite(value) or value <= 0:
        raise RuntimeError(f"{name} must be a finite positive duration")
    return value


COSIGN = "/usr/local/bin/cosign"
COSIGN_TIMEOUT_SECONDS = required_positive_seconds("VERIFIED_H018_COSIGN_TIMEOUT_SECONDS")
DECISION_FALLBACK_TTL_SECONDS = required_positive_seconds("VERIFIED_H018_DECISION_TTL_SECONDS")
CONTROL_DIR = Path("/etc/aidefend/h018")
ROOT_VERIFY_KEY = CONTROL_DIR / "trust-root.pub"
POLICY_PATH = CONTROL_DIR / "gate-policy.json"
REGISTRY_PATH = CONTROL_DIR / "identity-key-registry.json"
KEY_DIR = (CONTROL_DIR / "keys").resolve()
GATE_SIGNING_KEY = Path("/run/secrets/aidefend/h018-gate-signing.key")
GATE_VERIFY_KEY = CONTROL_DIR / "keys/h018-gate.pub"
INBOX = Path("/var/lib/aidefend/h018/inbox")
OUTBOX = Path("/var/lib/aidefend/h018/evidence/gate-decisions")
REDIS_SOCKET = "/run/aidefend-h018/redis.sock"
REQUEST_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
SHA256 = re.compile(r"^[a-f0-9]{64}$")
TOKEN = re.compile(r"^[A-Za-z0-9._:-]{1,256}$")

ACTION_FIELDS = {
    "schema_version", "action_id", "action_profile", "evidence_profile",
    "subject_id", "workload_id", "session_id", "task_id", "tool",
    "target", "parameters", "state_binding", "requested_at",
}
ENVELOPE_FIELDS = {
    "schema_version", "artifact_type", "issuer_id", "key_id", "issued_at",
    "expires_at", "nonce", "claims",
}
COMMON_CLAIMS = {"action_id", "action_sha256", "policy_version", "evidence_profile"}
CLAIM_FIELDS = {
    "identity_context": COMMON_CLAIMS | {
        "subject_id", "workload_id", "delegation_chain",
    },
    "validator_decision": COMMON_CLAIMS | {
        "decision", "observed_population_sha256", "observed_population_count",
    },
    "approval": COMMON_CLAIMS | {
        "approver_id", "role", "assurance", "decision",
        "effect_preview_sha256", "summary_sha256",
    },
    "control_evidence": COMMON_CLAIMS | {
        "evidence_kind", "result", "subject", "observed_population_sha256",
        "observed_population_count",
    },
}
EFFECT_PREVIEW_KIND = "effect_preview"
EFFECT_PREVIEW_FIELDS = {
    "preview_mode", "resolved_target_count", "resolved_targets_sha256",
    "reversibility", "state_generation_sha256", "summary_sha256",
}


class GateStop(Exception):
    def __init__(self, outcome: str, reason_code: str):
        super().__init__(reason_code)
        self.outcome = outcome
        self.reason_code = reason_code


def stop(outcome: str, reason_code: str) -> None:
    raise GateStop(outcome, reason_code)


def reject_floats(value: Any) -> None:
    if isinstance(value, float):
        stop("FAIL", "floating_point_not_allowed")
    if isinstance(value, list):
        for item in value:
            reject_floats(item)
    if isinstance(value, dict):
        for key, item in value.items():
            if not isinstance(key, str):
                stop("FAIL", "non_string_object_key")
            reject_floats(item)


def canonical(value: dict[str, Any] | list[Any]) -> bytes:
    reject_floats(value)
    return (json.dumps(
        value, sort_keys=True, separators=(",", ":"), ensure_ascii=False,
        allow_nan=False,
    ) + "\n").encode("utf-8")


def sha256(raw: bytes) -> str:
    return hashlib.sha256(raw).hexdigest()


def parse_time(value: Any) -> datetime:
    if not isinstance(value, str):
        stop("FAIL", "timestamp_type_invalid")
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        stop("FAIL", "timestamp_invalid")
    if parsed.utcoffset() is None:
        stop("FAIL", "timestamp_timezone_missing")
    return parsed.astimezone(timezone.utc)


def verify_blob_bytes(raw: bytes, signature: Path, key: Path) -> None:
    if not signature.is_file() or not key.is_file():
        stop("INSUFFICIENT_DATA", "signed_input_missing")
    snapshot: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(
            prefix="h018-signed-input-", suffix=".json", delete=False
        ) as handle:
            handle.write(raw)
            handle.flush()
            os.fsync(handle.fileno())
            snapshot = Path(handle.name)
        subprocess.run(
            [COSIGN, "verify-blob", "--key", str(key),
             "--bundle", str(signature), str(snapshot)],
            check=True, capture_output=True, text=True, timeout=COSIGN_TIMEOUT_SECONDS,
        )
    except subprocess.CalledProcessError:
        stop("FAIL", "signature_invalid")
    finally:
        if snapshot is not None:
            snapshot.unlink(missing_ok=True)


def verify_blob(blob: Path, signature: Path, key: Path) -> bytes:
    if not blob.is_file():
        stop("INSUFFICIENT_DATA", "signed_input_missing")
    raw = blob.read_bytes()
    verify_blob_bytes(raw, signature, key)
    return raw

def signature_path(path: Path) -> Path:
    return Path(str(path) + ".sig")


def load_canonical_bytes(
    raw: bytes, required_fields: set[str],
) -> tuple[dict[str, Any], bytes, str]:
    try:
        def unique_pairs(pairs):
            value = {}
            for key, item in pairs:
                if key in value:
                    raise ValueError(f"duplicate artifact key: {key}")
                value[key] = item
            return value

        def reject_constant(value):
            raise ValueError(f"non-finite artifact number: {value}")

        document = json.loads(
            raw.decode("utf-8", errors="strict"),
            object_pairs_hook=unique_pairs, parse_constant=reject_constant,
        )
    except (UnicodeDecodeError, json.JSONDecodeError, ValueError):
        stop("FAIL", "artifact_json_invalid")
    if not isinstance(document, dict) or set(document) != required_fields:
        stop("FAIL", "artifact_schema_mismatch")
    if canonical(document) != raw:
        stop("FAIL", "artifact_not_canonical")
    return document, raw, sha256(raw)


def load_canonical(
    path: Path, required_fields: set[str],
) -> tuple[dict[str, Any], bytes, str]:
    if not path.is_file():
        stop("INSUFFICIENT_DATA", "required_artifact_missing")
    return load_canonical_bytes(path.read_bytes(), required_fields)

def load_root_signed(path: Path, required_fields: set[str]) -> tuple[dict[str, Any], str]:
    raw = verify_blob(path, signature_path(path), ROOT_VERIFY_KEY)
    document, _, digest = load_canonical_bytes(raw, required_fields)
    return document, digest


def active(record: dict[str, Any], now: datetime) -> bool:
    if record.get("status") != "active" or parse_time(record.get("valid_until")) <= now:
        return False
    revoked_at = record.get("revoked_at")
    return revoked_at is None or now < parse_time(revoked_at)


def require_identity(
    registry: dict[str, Any], identity_id: str, now: datetime, role: str | None = None,
) -> dict[str, Any]:
    record = registry["identities"].get(identity_id)
    required = {"status", "roles", "can_issue", "valid_until", "revoked_at"}
    if not isinstance(record, dict) or set(record) != required or not active(record, now):
        stop("FAIL", "identity_inactive_or_revoked")
    if role is not None and role not in record["roles"]:
        stop("FAIL", "approver_role_not_granted")
    return record


def trusted_key(registry: dict[str, Any], key_id: str, issuer_id: str, artifact_type: str, now: datetime) -> Path:
    record = registry["keys"].get(key_id)
    required = {"owner_id", "status", "file_name", "artifact_types", "valid_until", "revoked_at"}
    if not isinstance(record, dict) or set(record) != required or not active(record, now):
        stop("FAIL", "signing_key_inactive_or_revoked")
    if record["owner_id"] != issuer_id or artifact_type not in record["artifact_types"]:
        stop("FAIL", "signing_key_not_authorized")
    if not isinstance(record["file_name"], str) or Path(record["file_name"]).name != record["file_name"]:
        stop("FAIL", "registry_key_path_invalid")
    key = (KEY_DIR / record["file_name"]).resolve()
    if key.parent != KEY_DIR or not key.is_file():
        stop("INSUFFICIENT_DATA", "trusted_key_unavailable")
    return key


def load_envelope(path: Path, registry: dict[str, Any], now: datetime) -> dict[str, Any]:
    if path.is_symlink() or path.resolve().parent != path.parent.resolve():
        stop("FAIL", "artifact_path_invalid")
    envelope, raw, digest = load_canonical(path, ENVELOPE_FIELDS)
    if envelope["schema_version"] != "aidefend.h018.signed_artifact.v2":
        stop("FAIL", "artifact_schema_version_unsupported")
    artifact_type = envelope["artifact_type"]
    if artifact_type not in CLAIM_FIELDS:
        stop("FAIL", "artifact_type_unsupported")
    issuer = require_identity(registry, envelope["issuer_id"], now)
    if artifact_type not in issuer["can_issue"]:
        stop("FAIL", "issuer_not_authorized")
    key = trusted_key(registry, envelope["key_id"], envelope["issuer_id"], artifact_type, now)
    verify_blob_bytes(raw, signature_path(path), key)
    issued_at, expires_at = parse_time(envelope["issued_at"]), parse_time(envelope["expires_at"])
    if not issued_at <= now < expires_at:
        stop("FAIL", "artifact_not_current")
    if not isinstance(envelope["nonce"], str) or not TOKEN.fullmatch(envelope["nonce"]):
        stop("FAIL", "artifact_nonce_invalid")
    claims = envelope["claims"]
    if not isinstance(claims, dict) or set(claims) != CLAIM_FIELDS[artifact_type]:
        stop("FAIL", "typed_claim_schema_mismatch")
    return {"document": envelope, "raw": raw, "sha256": digest, "expires_at": expires_at}


def logical_type(envelope: dict[str, Any]) -> str:
    if envelope["artifact_type"] != "control_evidence":
        return envelope["artifact_type"]
    kind = envelope["claims"]["evidence_kind"]
    if not isinstance(kind, str) or not TOKEN.fullmatch(kind):
        stop("FAIL", "evidence_kind_invalid")
    return "control_evidence:" + kind


def load_action(path: Path) -> tuple[dict[str, Any], str]:
    action, raw, digest = load_canonical(path, ACTION_FIELDS)
    if action["schema_version"] != "aidefend.h018.action_plan.v2":
        stop("FAIL", "action_schema_version_unsupported")
    for field in (
        "action_id", "action_profile", "evidence_profile", "subject_id",
        "workload_id", "session_id", "task_id", "tool", "target",
    ):
        if not isinstance(action[field], str) or not TOKEN.fullmatch(action[field]):
            stop("FAIL", "action_identifier_invalid")
    if not isinstance(action["parameters"], dict) or not isinstance(action["state_binding"], dict):
        stop("FAIL", "action_payload_invalid")
    if parse_time(action["requested_at"]) > datetime.now(timezone.utc):
        stop("FAIL", "action_requested_in_future")
    return action, digest


def validate_policy(policy: dict[str, Any], now: datetime) -> None:
    if policy["schema_version"] != "aidefend.h018.gate_policy.v2":
        stop("FAIL", "policy_schema_version_unsupported")
    if not parse_time(policy["valid_from"]) <= now < parse_time(policy["expires_at"]):
        stop("FAIL", "policy_not_current")
    if not isinstance(policy["profiles"], dict) or not policy["profiles"]:
        stop("FAIL", "policy_profiles_invalid")


def profile_for(policy: dict[str, Any], action: dict[str, Any]) -> dict[str, Any]:
    profile = policy["profiles"].get(action["action_profile"])
    fields = {
        "applicability", "required_approval_count", "required_distinct_roles",
        "required_assurance", "maximum_receipt_seconds",
        "allowed_evidence_profiles", "evidence_requirements",
    }
    if not isinstance(profile, dict) or set(profile) != fields:
        stop("FAIL", "action_profile_not_registered")
    if profile["applicability"] == "not_applicable":
        stop("NOT_APPLICABLE", "policy_marks_control_not_applicable")
    if profile["applicability"] != "required":
        stop("FAIL", "profile_applicability_invalid")
    count = profile["required_approval_count"]
    roles = profile["required_distinct_roles"]
    ttl = profile["maximum_receipt_seconds"]
    if (
        isinstance(count, bool) or not isinstance(count, int) or count < 1
        or not isinstance(roles, list) or len(roles) != count
        or len(set(roles)) != len(roles) or not all(isinstance(x, str) and TOKEN.fullmatch(x) for x in roles)
        or isinstance(ttl, bool) or not isinstance(ttl, int) or ttl < 1
        or not isinstance(profile["required_assurance"], str)
        or not profile["required_assurance"]
        or not isinstance(profile["allowed_evidence_profiles"], list)
        or not profile["allowed_evidence_profiles"]
        or len(profile["allowed_evidence_profiles"]) != len(set(profile["allowed_evidence_profiles"]))
        or not all(isinstance(x, str) and TOKEN.fullmatch(x) for x in profile["allowed_evidence_profiles"])
        or not isinstance(profile["evidence_requirements"], dict)
    ):
        stop("FAIL", "profile_approval_policy_invalid")
    evidence_profile = action["evidence_profile"]
    if evidence_profile not in profile["allowed_evidence_profiles"]:
        stop("FAIL", "evidence_profile_not_allowed")
    requirements = profile["evidence_requirements"].get(evidence_profile)
    if (
        not isinstance(requirements, list) or len(requirements) != len(set(requirements))
        or not all(isinstance(x, str) and x for x in requirements)
    ):
        stop("FAIL", "evidence_profile_policy_invalid")
    if "identity_context" not in requirements or "validator_decision" not in requirements:
        stop("FAIL", "mandatory_evidence_type_missing_from_policy")
    return profile


def validate_effect_preview(claims: dict[str, Any], action: dict[str, Any]) -> str:
    # A preview is typed control evidence: side-effect free, bound to the exact
    # pre-state the action binds, and counting the targets it actually resolved.
    subject = claims["subject"]
    if not isinstance(subject, dict) or set(subject) != EFFECT_PREVIEW_FIELDS:
        stop("FAIL", "effect_preview_schema_mismatch")
    if subject["preview_mode"] not in {"dry_run", "plan_diff"}:
        stop("FAIL", "effect_preview_mode_unsupported")
    count = subject["resolved_target_count"]
    if isinstance(count, bool) or not isinstance(count, int) or count < 0:
        stop("FAIL", "effect_preview_target_count_invalid")
    if count != claims["observed_population_count"]:
        stop("FAIL", "effect_preview_population_mismatch")
    if subject["reversibility"] not in {"reversible", "compensable", "irreversible"}:
        stop("FAIL", "effect_preview_reversibility_invalid")
    for key in ("resolved_targets_sha256", "state_generation_sha256", "summary_sha256"):
        if not isinstance(subject[key], str) or not SHA256.fullmatch(subject[key]):
            stop("FAIL", "effect_preview_digest_invalid")
    if subject["state_generation_sha256"] != sha256(canonical(action["state_binding"])):
        stop("FAIL", "effect_preview_state_binding_stale")
    return subject["summary_sha256"]


def bind_approvals_to_preview(
    previews: list[dict[str, Any]], approvals: list[dict[str, Any]], summary_sha256: str | None,
) -> str | None:
    # An approval is valid only for the exact preview the approver saw: its
    # signed claims must carry the preview artifact digest and the digest of
    # the displayed summary, and it must postdate that preview. A timestamp
    # alone proves nothing about what was shown.
    if summary_sha256 is None:
        for approval in approvals:
            claims = approval["document"]["claims"]
            if claims["effect_preview_sha256"] is not None or claims["summary_sha256"] is not None:
                stop("FAIL", "approval_bound_to_unselected_effect_preview")
        return None
    if len(previews) != 1:
        stop("INSUFFICIENT_DATA", "effect_preview_evidence_not_exact")
    preview = previews[0]
    preview_issued_at = parse_time(preview["document"]["issued_at"])
    for approval in approvals:
        claims = approval["document"]["claims"]
        if (
            claims["effect_preview_sha256"] != preview["sha256"]
            or claims["summary_sha256"] != summary_sha256
        ):
            stop("FAIL", "approval_not_bound_to_effect_preview")
        if parse_time(approval["document"]["issued_at"]) <= preview_issued_at:
            stop("FAIL", "approval_issued_before_effect_preview")
    return preview["sha256"]


def consume_approvals_once(action_sha: str, approvals: list[dict[str, Any]], ttl: int) -> None:
    nonce_hashes = sorted(sha256(x["document"]["nonce"].encode("utf-8")) for x in approvals)
    if len(nonce_hashes) != len(set(nonce_hashes)):
        stop("FAIL", "approval_nonce_not_distinct")
    keys = ["h018:approval:" + item for item in nonce_hashes]
    keys.append("h018:gate-issued:" + action_sha)
    script = """
for i = 1, #KEYS do
  if redis.call('EXISTS', KEYS[i]) == 1 then return 0 end
end
for i = 1, #KEYS do
  redis.call('SET', KEYS[i], ARGV[1], 'EX', ARGV[2])
end
return 1
"""
    client = redis.Redis(unix_socket_path=REDIS_SOCKET, decode_responses=True)
    if client.eval(script, len(keys), *keys, action_sha, ttl) != 1:
        stop("FAIL", "approval_nonce_or_action_replayed")


def evaluate(request_dir: Path, context: dict[str, Any]) -> datetime:
    now = datetime.now(timezone.utc)
    action, action_sha = load_action(request_dir / "action-plan.json")
    context.update(action_id=action["action_id"], action_sha256=action_sha)
    policy, policy_sha = load_root_signed(
        POLICY_PATH, {"schema_version", "policy_version", "valid_from", "expires_at", "profiles"},
    )
    registry, registry_sha = load_root_signed(
        REGISTRY_PATH,
        {"schema_version", "registry_version", "issued_at", "expires_at", "keys", "identities"},
    )
    if registry["schema_version"] != "aidefend.h018.identity_key_registry.v2":
        stop("FAIL", "registry_schema_version_unsupported")
    if not parse_time(registry["issued_at"]) <= now < parse_time(registry["expires_at"]):
        stop("FAIL", "registry_not_current")
    validate_policy(policy, now)
    context.update(
        policy_version=policy["policy_version"], policy_sha256=policy_sha,
        registry_sha256=registry_sha,
    )
    profile = profile_for(policy, action)

    artifact_dir = request_dir / "artifacts"
    paths = sorted(artifact_dir.glob("*.json")) if artifact_dir.is_dir() else []
    if not paths:
        stop("INSUFFICIENT_DATA", "evidence_population_missing")
    if len(paths) > 256:
        stop("FAIL", "evidence_population_too_large")
    artifacts = [load_envelope(path, registry, now) for path in paths]
    digests = sorted(item["sha256"] for item in artifacts)
    if len(digests) != len(set(digests)):
        stop("FAIL", "duplicate_evidence_artifact")
    context["evidence_sha256s"] = digests

    binding = (action["action_id"], action_sha, policy["policy_version"], action["evidence_profile"])
    for item in artifacts:
        claims = item["document"]["claims"]
        observed = (
            claims["action_id"], claims["action_sha256"],
            claims["policy_version"], claims["evidence_profile"],
        )
        if observed != binding:
            stop("FAIL", "signed_artifact_binding_mismatch")

    approvals = [x for x in artifacts if x["document"]["artifact_type"] == "approval"]
    evidence = [x for x in artifacts if x["document"]["artifact_type"] != "approval"]
    evidence_types = [logical_type(x["document"]) for x in evidence]
    required_types = profile["evidence_requirements"][action["evidence_profile"]]
    if sorted(evidence_types) != sorted(required_types):
        stop("INSUFFICIENT_DATA", "evidence_population_not_exact")

    identities = [x for x in evidence if x["document"]["artifact_type"] == "identity_context"]
    validators = [x for x in evidence if x["document"]["artifact_type"] == "validator_decision"]
    if len(identities) != 1 or len(validators) != 1:
        stop("INSUFFICIENT_DATA", "identity_or_validator_evidence_not_exact")
    identity_claims = identities[0]["document"]["claims"]
    if (
        identity_claims["subject_id"] != action["subject_id"]
        or identity_claims["workload_id"] != action["workload_id"]
        or not isinstance(identity_claims["delegation_chain"], list)
        or len(identity_claims["delegation_chain"]) != len(set(identity_claims["delegation_chain"]))
        or not all(isinstance(x, str) and TOKEN.fullmatch(x) for x in identity_claims["delegation_chain"])
    ):
        stop("FAIL", "identity_context_binding_mismatch")
    require_identity(registry, action["subject_id"], now)
    require_identity(registry, action["workload_id"], now)
    for delegated_identity in identity_claims["delegation_chain"]:
        require_identity(registry, delegated_identity, now)
    validator_claims = validators[0]["document"]["claims"]
    if (
        validator_claims["decision"] != "allow"
        or not SHA256.fullmatch(validator_claims["observed_population_sha256"])
        or isinstance(validator_claims["observed_population_count"], bool)
        or not isinstance(validator_claims["observed_population_count"], int)
        or validator_claims["observed_population_count"] < 1
    ):
        stop("FAIL", "validator_decision_or_population_invalid")
    for item in evidence:
        if item["document"]["artifact_type"] == "control_evidence":
            claims = item["document"]["claims"]
            if (
                claims["result"] != "PASS"
                or not SHA256.fullmatch(claims["observed_population_sha256"])
                or isinstance(claims["observed_population_count"], bool)
                or not isinstance(claims["observed_population_count"], int)
                or claims["observed_population_count"] < 0
            ):
                stop("FAIL", "control_evidence_result_or_population_invalid")
    previews = [
        x for x in evidence
        if x["document"]["artifact_type"] == "control_evidence"
        and x["document"]["claims"]["evidence_kind"] == EFFECT_PREVIEW_KIND
    ]
    if "control_evidence:" + EFFECT_PREVIEW_KIND in required_types:
        if len(previews) != 1:
            stop("INSUFFICIENT_DATA", "effect_preview_evidence_not_exact")
        summary_sha256 = validate_effect_preview(previews[0]["document"]["claims"], action)
        context["effect_preview_sha256"] = bind_approvals_to_preview(previews, approvals, summary_sha256)
        context["effect_summary_sha256"] = summary_sha256
    elif previews:
        stop("FAIL", "effect_preview_not_selected_by_policy")
    else:
        bind_approvals_to_preview([], approvals, None)

    if len(approvals) != profile["required_approval_count"]:
        stop("INSUFFICIENT_DATA", "approval_count_not_exact")
    approver_ids: list[str] = []
    approval_roles: list[str] = []
    for item in approvals:
        claims = item["document"]["claims"]
        if claims["decision"] != "approved" or claims["assurance"] != profile["required_assurance"]:
            stop("FAIL", "approval_decision_or_assurance_invalid")
        require_identity(registry, claims["approver_id"], now, claims["role"])
        approver_ids.append(claims["approver_id"])
        approval_roles.append(claims["role"])
    if len(set(approver_ids)) != len(approver_ids) or len(set(approval_roles)) != len(approval_roles):
        stop("FAIL", "approver_identity_or_role_not_distinct")
    if sorted(approval_roles) != sorted(profile["required_distinct_roles"]):
        stop("FAIL", "required_approval_roles_not_satisfied")

    expires_at = min(
        [now + timedelta(seconds=profile["maximum_receipt_seconds"])]
        + [item["expires_at"] for item in artifacts]
    )
    ttl = max(1, int((expires_at - now).total_seconds()))
    consume_approvals_once(action_sha, approvals, ttl)
    context["approval_nonce_hashes"] = sorted(
        sha256(item["document"]["nonce"].encode("utf-8")) for item in approvals
    )
    return expires_at


def publish_signed(record: dict[str, Any]) -> tuple[Path, str]:
    OUTBOX.mkdir(parents=True, exist_ok=True, mode=0o750)
    name = record["request_id"] + "." + record["decision_id"]
    temporary = OUTBOX / ("." + name + ".tmp")
    temporary_signature = signature_path(temporary)
    final = OUTBOX / (name + ".json")
    final_signature = signature_path(final)
    payload = canonical(record)
    with temporary.open("xb") as handle:
        os.fchmod(handle.fileno(), 0o640)
        handle.write(payload)
        handle.flush()
        os.fsync(handle.fileno())
    subprocess.run(
        [
            COSIGN, "sign-blob", "--yes", "--key", str(GATE_SIGNING_KEY),
            "--bundle", str(temporary_signature), str(temporary),
        ],
        check=True, capture_output=True, text=True, timeout=COSIGN_TIMEOUT_SECONDS,
    )
    verify_blob(temporary, temporary_signature, GATE_VERIFY_KEY)
    os.replace(temporary, final)
    os.replace(temporary_signature, final_signature)
    if final.read_bytes() != payload:
        raise OSError("gate decision readback mismatch")
    verify_blob(final, final_signature, GATE_VERIFY_KEY)
    return final, sha256(payload)


def main() -> int:
    if len(sys.argv) != 2 or not REQUEST_ID.fullmatch(sys.argv[1]):
        raise SystemExit("usage: issue_h018_gate_decision.py REQUEST_ID")
    request_id = sys.argv[1]
    request_dir = INBOX / request_id
    context: dict[str, Any] = {
        "action_id": None, "action_sha256": None, "policy_version": None,
        "policy_sha256": None, "registry_sha256": None,
        "evidence_sha256s": [], "approval_nonce_hashes": [],
        "effect_preview_sha256": None, "effect_summary_sha256": None,
    }
    outcome, reason = "PASS", "all_gate_requirements_satisfied"
    expires_at = datetime.now(timezone.utc) + timedelta(seconds=DECISION_FALLBACK_TTL_SECONDS)
    try:
        expires_at = evaluate(request_dir, context)
    except GateStop as exc:
        outcome, reason = exc.outcome, exc.reason_code
    except (redis.RedisError, subprocess.TimeoutExpired, OSError):
        outcome, reason = "ERROR", "control_plane_dependency_error"
    except Exception:
        outcome, reason = "ERROR", "unhandled_gate_error"

    now = datetime.now(timezone.utc)
    evidence = context["evidence_sha256s"]
    record = {
        "schema_version": "aidefend.h018.gate_decision.v3",
        "control": "AID-H-018.003",
        "decision_id": secrets.token_urlsafe(24),
        "request_id": request_id,
        "outcome": outcome,
        "reason_code": reason,
        "action_id": context["action_id"],
        "action_sha256": context["action_sha256"],
        "policy_version": context["policy_version"],
        "policy_sha256": context["policy_sha256"],
        "registry_sha256": context["registry_sha256"],
        "evidence_sha256s": evidence,
        "evidence_population_sha256": sha256(canonical(evidence)),
        "evidence_population_count": len(evidence),
        "approval_nonce_hashes": context["approval_nonce_hashes"],
        "effect_preview_sha256": context["effect_preview_sha256"],
        "effect_summary_sha256": context["effect_summary_sha256"],
        "issued_at": now.isoformat(),
        "expires_at": expires_at.isoformat(),
        "gate_nonce": secrets.token_urlsafe(32) if outcome == "PASS" else None,
    }
    path, digest = publish_signed(record)
    print(json.dumps({"outcome": outcome, "decision": str(path), "sha256": digest}))
    return 0 if outcome in {"PASS", "NOT_APPLICABLE"} else 2


if __name__ == "__main__":
    raise SystemExit(main())

Independent replay/readback under a separate identity

Run the following verifier from a different workload identity with read-only access to the inbox and gate outbox and write-only access to its own evidence directory. It verifies the gate signature and canonical bytes, recomputes the action and complete artifact-population digests, and signs a replay receipt. A PASS decision is not consumable until this replay receipt is present and itself verifies.

# File: action_gate/replay_h018_gate_decision.py
from __future__ import annotations

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


def required_positive_seconds(name: str) -> float:
    value = float(os.environ[name])
    if not math.isfinite(value) or value <= 0:
        raise RuntimeError(f"{name} must be a finite positive duration")
    return value


COSIGN = "/usr/local/bin/cosign"
COSIGN_TIMEOUT_SECONDS = required_positive_seconds("VERIFIED_H018_REPLAY_COSIGN_TIMEOUT_SECONDS")
INBOX = Path("/var/lib/aidefend/h018/inbox")
DECISIONS = Path("/var/lib/aidefend/h018/evidence/gate-decisions")
REPLAYS = Path("/var/lib/aidefend/h018/evidence/gate-replays")
GATE_KEY = Path("/etc/aidefend/h018/keys/h018-gate.pub")
REPLAY_SIGNING_KEY = Path("/run/secrets/aidefend/h018-replay-signing.key")
REPLAY_VERIFY_KEY = Path("/etc/aidefend/h018/keys/h018-replay.pub")
SAFE_NAME = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,255}\.json$")


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


def digest(raw: bytes) -> str:
    return hashlib.sha256(raw).hexdigest()


def verify(blob: Path, signature: Path, key: Path) -> bytes:
    raw = blob.read_bytes()
    snapshot: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(
            prefix="h018-gate-replay-input-", suffix=".json", delete=False
        ) as handle:
            handle.write(raw)
            handle.flush()
            os.fsync(handle.fileno())
            snapshot = Path(handle.name)
        subprocess.run(
            [COSIGN, "verify-blob", "--key", str(key),
             "--bundle", str(signature), str(snapshot)],
            check=True, capture_output=True, text=True, timeout=COSIGN_TIMEOUT_SECONDS,
        )
        return raw
    finally:
        if snapshot is not None:
            snapshot.unlink(missing_ok=True)

def main() -> int:
    if len(sys.argv) != 2 or not SAFE_NAME.fullmatch(sys.argv[1]):
        raise SystemExit("usage: replay_h018_gate_decision.py DECISION_FILE.json")
    decision_path = DECISIONS / sys.argv[1]
    raw = verify(decision_path, Path(str(decision_path) + ".sig"), GATE_KEY)
    def unique_pairs(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError(f"duplicate gate-decision key: {key}")
            value[key] = item
        return value
    def reject_constant(value):
        raise ValueError(f"non-finite gate-decision number: {value}")
    decision = json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_pairs, parse_constant=reject_constant,
    )
    if canonical(decision) != raw or decision.get("schema_version") != "aidefend.h018.gate_decision.v3":
        raise ValueError("gate decision is not canonical v3 evidence")

    request_dir = INBOX / decision["request_id"]
    action_path = request_dir / "action-plan.json"
    action_digest = digest(action_path.read_bytes()) if action_path.is_file() else None
    artifact_digests = sorted(
        digest(path.read_bytes()) for path in (request_dir / "artifacts").glob("*.json")
    )
    valid = action_digest == decision["action_sha256"]
    if decision["outcome"] == "PASS":
        valid = valid and artifact_digests == decision["evidence_sha256s"]
    else:
        valid = valid and set(decision["evidence_sha256s"]).issubset(artifact_digests)
    valid = valid and digest(canonical(decision["evidence_sha256s"])) == decision["evidence_population_sha256"]

    replay = {
        "schema_version": "aidefend.h018.gate_replay.v2",
        "control": "AID-H-018.003",
        "decision_sha256": digest(raw),
        "request_id": decision["request_id"],
        "action_sha256": action_digest,
        "artifact_population_sha256": digest(canonical(artifact_digests)),
        "artifact_population_count": len(artifact_digests),
        "replay_outcome": "PASS" if valid else "FAIL",
        "observed_at": datetime.now(timezone.utc).isoformat(),
    }
    REPLAYS.mkdir(parents=True, exist_ok=True, mode=0o750)
    output = REPLAYS / (decision_path.stem + ".replay.json")
    temporary = output.with_name("." + output.name + ".tmp")
    temporary.write_bytes(canonical(replay))
    os.chmod(temporary, 0o640)
    temporary_signature = Path(str(temporary) + ".sig")
    subprocess.run(
        [
            COSIGN, "sign-blob", "--yes", "--key", str(REPLAY_SIGNING_KEY),
            "--bundle", str(temporary_signature), str(temporary),
        ],
        check=True, capture_output=True, text=True, timeout=COSIGN_TIMEOUT_SECONDS,
    )
    verify(temporary, temporary_signature, REPLAY_VERIFY_KEY)
    os.replace(temporary, output)
    final_signature = Path(str(output) + ".sig")
    os.replace(temporary_signature, final_signature)
    if output.read_bytes() != canonical(replay):
        raise OSError("replay receipt readback mismatch")
    verify(output, final_signature, REPLAY_VERIFY_KEY)
    return 0 if valid else 2


if __name__ == "__main__":
    raise SystemExit(main())

Show the approver the effect, not the parameters

An approval that shows only the action parameters lets a human authorize a blast radius nobody resolved: a recipient list that expands to a whole directory, a filter that matches every row, a payment whose currency or account was substituted. For action profiles that policy marks as preview-required, list control_evidence:effect_preview in the profile's evidence_requirements. The preview is produced by a separate, side-effect-free path (a dry run or a plan diff against the exact pre-state), is signed like any other typed evidence, and its subject carries the resolved target count, the digest of the resolved targets, the reversibility class, the digest of the pre-state the action itself binds, and the digest of the human-readable summary that the approver saw. The gate checks that the preview resolved the same population it reports, that it was computed against the same state_binding as the action so a target moved after preview fails as stale rather than executing, and that every approval's signed claims carry the digest of that exact preview artifact and of the summary the approver saw, issued after the preview existed: an approval signed against a different summary, or with no preview binding when policy requires one, fails as approval_not_bound_to_effect_preview, because a timestamp alone proves nothing about what was shown. The aidefend.h018.gate_decision.v3 record carries both digests. Its field set is a shared contract with the AID-H-018.006 executor and the replay verifier below: a field added on one side is added to the others in the same change, the schema version moves with it, and a test feeds a freshly issued decision through the executor loader so a mismatch fails in CI rather than as artifact_schema_mismatch in production. A producer that cannot compute a truthful preview must say so as a non-PASS result; the gate never lets an unsupported preview pass as a precise dry run. AID-M-006.001 defines what the approver must be shown, and AID-H-018.006 re-verifies the pre-state generation at execution time.

Required policy, evidence, and outcome semantics

Action: Make the H-018.003 decision and independent replay receipt mandatory inputs to H-018.006. Never execute directly from an approval UI response, validator boolean, model-generated role, or unsigned audit event.

Keep the action safely bounded

Reconcile the complete population of all required validator and approver slots plus every gate attempt for the action; reject expired, revoked, or stale policy and asset bindings under the policy-defined freshness window.

AID-H-018.003-G002

Use the reviewer only as defense in depth

Enable this method only for action classes where measured false-negative reduction justifies its latency, cost, and false-positive denial-of-service risk. The reviewer is not an authorization service. It may return only NO_FINDING, HOLD, ESCALATE, DENY_RECOMMENDATION, or ERROR. A deterministic gate evaluates that signal together with the ordinary H-018.002 authorization and any mandatory human approval. NO_FINDING merely means that this reviewer supplied no blocking signal; it never creates an approval or overrides a deterministic or human denial.

Validate a narrowly typed, fully bound result

# File: approval/semantic_reviewer.py
from __future__ import annotations

import hashlib
import json
import math
import re
from dataclasses import dataclass
from datetime import datetime, timezone
from enum import Enum
from typing import Callable


SHA256 = re.compile(r"^[a-f0-9]{64}$")
REASON = re.compile(r"^[A-Z][A-Z0-9_]{0,63}$")


class ReviewMode(str, Enum):
    DISABLED = "DISABLED"
    SHADOW = "SHADOW"
    REQUIRED = "REQUIRED"


class ReviewStatus(str, Enum):
    NO_FINDING = "NO_FINDING"
    HOLD = "HOLD"
    ESCALATE = "ESCALATE"
    DENY_RECOMMENDATION = "DENY_RECOMMENDATION"
    ERROR = "ERROR"


@dataclass(frozen=True)
class ReviewPolicy:
    mode: ReviewMode
    policy_version: str
    model_id: str
    model_version: str
    prompt_sha256: str
    output_schema_version: str
    timeout_seconds: float
    maximum_response_bytes: int
    maximum_result_age_seconds: int
    maximum_result_ttl_seconds: int
    permitted_review_fields: frozenset[str]


@dataclass(frozen=True)
class ReviewResult:
    status: ReviewStatus
    reason_code: str
    action_sha256: str
    goal_sha256: str
    policy_version: str
    model_id: str
    model_version: str
    prompt_sha256: str
    issued_at: datetime
    expires_at: datetime


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


def strict_object(raw: bytes) -> dict:
    def unique(pairs):
        value = {}
        for key, item in pairs:
            if key in value:
                raise ValueError("duplicate reviewer result key: " + key)
            value[key] = item
        return value

    def reject_constant(value):
        raise ValueError("non-finite reviewer value: " + value)

    value = json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique,
        parse_constant=reject_constant,
    )
    if not isinstance(value, dict):
        raise ValueError("reviewer result must be an object")
    return value


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


def validate_policy(policy: ReviewPolicy) -> None:
    if (
        not isinstance(policy.mode, ReviewMode)
        or not policy.policy_version
        or not policy.model_id
        or not policy.model_version
        or not SHA256.fullmatch(policy.prompt_sha256)
        or not policy.output_schema_version
        or isinstance(policy.timeout_seconds, bool)
        or not isinstance(policy.timeout_seconds, (int, float))
        or not math.isfinite(float(policy.timeout_seconds))
        or policy.timeout_seconds <= 0
        or type(policy.maximum_response_bytes) is not int
        or policy.maximum_response_bytes <= 0
        or type(policy.maximum_result_age_seconds) is not int
        or policy.maximum_result_age_seconds <= 0
        or type(policy.maximum_result_ttl_seconds) is not int
        or policy.maximum_result_ttl_seconds <= 0
        or not isinstance(policy.permitted_review_fields, frozenset)
        or not policy.permitted_review_fields
        or any(
            not isinstance(field, str) or not field
            for field in policy.permitted_review_fields
        )
    ):
        raise ValueError("semantic-review policy is incomplete")


def minimized_view(action: dict, permitted: frozenset[str]) -> dict:
    view = action.get("semantic_review_view")
    if not isinstance(view, dict) or not view or not set(view).issubset(permitted):
        raise ValueError("semantic review view violates field policy")
    for key, value in view.items():
        if not isinstance(key, str) or not key:
            raise ValueError("semantic review field name is invalid")
        if not isinstance(value, (str, int, bool)) or isinstance(value, float):
            raise ValueError("semantic review fields must be bounded scalars")
        if isinstance(value, str) and len(value.encode("utf-8")) > 1024:
            raise ValueError("semantic review field is too large")
    return view


def local_error(action_sha256: str, goal_sha256: str, policy: ReviewPolicy) -> ReviewResult:
    now = datetime.now(timezone.utc)
    return ReviewResult(
        ReviewStatus.ERROR, "REVIEWER_UNAVAILABLE", action_sha256,
        goal_sha256, policy.policy_version, policy.model_id,
        policy.model_version, policy.prompt_sha256, now, now,
    )


def review_action(
    *,
    immutable_action: dict,
    goal_sha256: str,
    policy: ReviewPolicy,
    call_reviewer: Callable[[bytes, float], bytes],
) -> ReviewResult:
    validate_policy(policy)
    if not SHA256.fullmatch(goal_sha256):
        raise ValueError("goal digest is invalid")
    action_sha256 = hashlib.sha256(canonical(immutable_action)).hexdigest()
    request = {
        "schema_version": "aidefend.semantic-review-request.v1",
        "output_schema_version": policy.output_schema_version,
        "action_sha256": action_sha256,
        "goal_sha256": goal_sha256,
        "policy_version": policy.policy_version,
        "model_id": policy.model_id,
        "model_version": policy.model_version,
        "prompt_sha256": policy.prompt_sha256,
        "action_view": minimized_view(
            immutable_action, policy.permitted_review_fields
        ),
    }
    try:
        raw = call_reviewer(canonical(request), policy.timeout_seconds)
        if not isinstance(raw, bytes) or len(raw) > policy.maximum_response_bytes:
            raise ValueError("reviewer response violates byte policy")
        result = strict_object(raw)
    except (OSError, TimeoutError, TypeError, ValueError):
        return local_error(action_sha256, goal_sha256, policy)

    required = {
        "schema_version", "status", "reason_code", "action_sha256",
        "goal_sha256", "policy_version", "model_id", "model_version",
        "prompt_sha256", "issued_at", "expires_at",
    }
    if set(result) != required:
        return local_error(action_sha256, goal_sha256, policy)
    try:
        status = ReviewStatus(result["status"])
        issued_at = parse_time(result["issued_at"])
        expires_at = parse_time(result["expires_at"])
    except (TypeError, ValueError):
        return local_error(action_sha256, goal_sha256, policy)
    if (
        status is ReviewStatus.ERROR
        or not isinstance(result["reason_code"], str)
        or not REASON.fullmatch(result["reason_code"])
        or result["schema_version"] != policy.output_schema_version
        or result["action_sha256"] != action_sha256
        or result["goal_sha256"] != goal_sha256
        or result["policy_version"] != policy.policy_version
        or result["model_id"] != policy.model_id
        or result["model_version"] != policy.model_version
        or result["prompt_sha256"] != policy.prompt_sha256
    ):
        return local_error(action_sha256, goal_sha256, policy)
    now = datetime.now(timezone.utc)
    age = (now - issued_at).total_seconds()
    ttl = (expires_at - issued_at).total_seconds()
    if (
        age < 0 or age > policy.maximum_result_age_seconds
        or ttl <= 0 or ttl > policy.maximum_result_ttl_seconds
        or expires_at <= now
    ):
        return local_error(action_sha256, goal_sha256, policy)
    return ReviewResult(
        status, result["reason_code"], action_sha256, goal_sha256,
        policy.policy_version, policy.model_id, policy.model_version,
        policy.prompt_sha256, issued_at, expires_at,
    )


def deterministic_gate(
    *,
    policy_authorized: bool,
    human_required: bool,
    human_approved: bool,
    mode: ReviewMode,
    review: ReviewResult | None,
) -> str:
    if not policy_authorized:
        return "DENY"
    if human_required and not human_approved:
        return "HOLD"
    if mode is ReviewMode.DISABLED:
        return "ALLOW"
    if mode is ReviewMode.SHADOW:
        return "ALLOW"  # Record the signal, but never change authorization.
    if review is None or review.status in {
        ReviewStatus.ERROR, ReviewStatus.HOLD, ReviewStatus.ESCALATE,
    }:
        return "HOLD"
    if review.status is ReviewStatus.DENY_RECOMMENDATION:
        return "DENY"  # Deterministic policy turns this typed signal into denial.
    if review.status is ReviewStatus.NO_FINDING:
        return "ALLOW"  # All non-reviewer authorization prerequisites already passed.
    return "HOLD"

Deployment and privacy boundary

Build semantic_review_view in a trusted adapter from policy-selected, normalized fields such as tool, operation class, target class, destination class, and bounded business values. Do not send credentials, raw files, hidden prompts, unrestricted user content, or arguments unnecessary to the review; represent secrets and large objects by classification and digest. Invoke the reviewer through an authenticated, bounded channel and retain only request/result digests plus the bound model, prompt, schema, policy, action, goal, issue, and expiry metadata. A same-model or same-provider reviewer may provide useful diversity but is not an independent trust domain.

Roll out and verify safely

Start in SHADOW mode and compare reviewer signals with deterministic policy, human decisions, and final outcomes. Before selecting REQUIRED, replay benign and adversarial high-impact actions and measure false negatives, false-positive denial-of-service, malformed/oversized output, prompt and model drift, timeout/outage behavior, latency, cost, and sensitive-data leakage through both request and response channels. In required mode every timeout, malformed result, binding mismatch, expiry, or outage becomes ERROR and the deterministic gate holds or escalates; in shadow mode the same failure is observable but cannot authorize, deny, or delay an action.

Open-source tools

• Cosign
• Valkey
• Redis 8+ (AGPL-3.0)
• OpenBao (approval-receipt signing-key custody only; separate approval and validation channel required)

Commercial / hosted tools

• Microsoft Entra ID Conditional Access (step-up identity signal only; separate approval ledger required)
• Okta Identity Engine (step-up identity signal only; separate approval ledger required)
• Cisco Duo (step-up identity signal only; separate approval ledger required)
• AWS Key Management Service (AWS KMS)
• HashiCorp Vault Enterprise
• HCP Vault Dedicated

Threat mappings

MITRE ATLAS

• AML.T0053 AI Agent Tool Invocation
• AML.T0051 LLM Prompt Injection (second channel independently validates injection attempts)
• AML.T0052.001 Phishing: Deepfake-Assisted Phishing (independent validation blocks spoofed high-impact requests)

MAESTRO

• Agent Tool Misuse (L7)
• Agent Goal Manipulation (L7)

OWASP LLM Top 10 2026

• LLM01:2026 Prompt Injection (independent validation and approval block high-impact consequences of injected instructions but do not detect the injection itself)
• LLM03:2026 Excessive Agency
• LLM07:2026 Misinformation (evidence-bound independent validation implements a claim-check-act barrier for high-impact actions but does not fact-check ordinary responses)
• LLM10:2026 Improper Output Handling

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI02:2026 Tool Misuse and Exploitation
• ASI01:2026 Agent Goal Hijack (independent validator catches goal hijack attempts)
• ASI03:2026 Identity and Privilege Abuse
• ASI05:2026 Unexpected Code Execution (RCE) (two-channel validation catches code execution attempts in high-impact actions)
• ASI09:2026 Human-Agent Trust Exploitation (exact-plan step-up approval and independent validation prevent blind approval of persuasive agent recommendations)

NIST Adversarial Machine Learning 2025

• NISTAML.039 Compromising connected resources (validator blocks unauthorized resource access)
• NISTAML.018 Prompt Injection (independent validator catches prompt injection-driven high-impact actions)

Cisco Integrated AI Security and Safety Framework

• AISubtech-12.1.1 Parameter Manipulation
• AISubtech-2.1.1 Execution Approval Bypass
• AISubtech-2.2.3 Authorization and Constraint Erosion
• AISubtech-4.3.4 Replay Exploitation
• AITech-12.1 Tool Exploitation
• AITech-14.2 Abuse of Delegated Authority
• AITech-2.1 Excessive Agency
• AITech-2.2 Goal Drift

Google Secure AI Framework 2.0 - Risks

• RA: Rogue Actions (independent validator catches rogue high-impact actions)
• PIJ: Prompt Injection (second channel independently validates injection attempts)
• IMO: Insecure Model Output (validator catches insecure outputs before high-impact execution)

Databricks AI Security Framework 3.0

• Agents - Core 13.2: Tool Misuse (two-channel validation prevents high-impact tool misuse)
• Agents - Core 13.3: Privilege Compromise
• Agents - Core 13.5: Cascading Hallucination Attacks (independent validator catches hallucination-driven high-impact actions)
• Agents - Core 13.7: Misaligned & Deceptive Behaviors (second channel detects misaligned high-impact actions)