AID-M-006.003: HITL Escalation Topology & Audit Event Contract

Version 1.20261015 · Actionable sub-technique

Define and test a timeout-bounded escalation topology and a canonical, tamper-evident audit-event contract for every Human-in-the-Loop checkpoint. The contract links activation, escalation, human decision or timeout, the independently produced runtime-gate receipt, and the observed final action so a separate verifier can prove whether the lifecycle is complete and internally consistent.

Pillars: app

Lifecycle: operation

Scope boundary

Owns timeout-bounded HITL escalation routing, canonical tamper-evident audit events, producer obligations, external retention, delivery verification, and after-the-fact reconciliation from checkpoint activation through final action. It never authorizes or unlocks the action.

AID-M-006.001: AID-M-006.003 owns runtime escalation and the lifecycle event contract; AID-M-006.001 owns checkpoint requirements, reviewer evidence, pause and resume interfaces, and operator procedures. The escalation graph implements the defined checkpoint but does not decide which actions require one.

AID-M-006.002: AID-M-006.003 defines the topology and evidence contract; AID-M-006.002 qualifies operators and exercises every required branch end to end. A complete event schema is not proof that operators and delivery paths are ready under realistic conditions.

AID-H-018.003: AID-M-006.003 routes escalation and reconciles the lifecycle record; AID-H-018.003 validates the exact high-impact action and issues the canonical runtime gate receipt. Audit-chain ownership must remain separate from the component that returns the allow or deny decision.

AID-D-015: AID-M-006.003 defines and reconciles the expected HITL event chain; AID-D-015 independently detects missing, mismatched, replayed, bypassed, or anomalous approval activity. Contract reconciliation supplies evidence, while the Detect control owns production detection and alerting on violations.

Implementation guidance

AID-M-006.003-G001: Define, codify, and test clear escalation paths for human intervention.
AID-M-006.003-G002: Publish canonical HITL audit events to an external retention-locked store and independently reconcile lifecycle completeness without authorizing execution.

AID-M-006.003-G001

Define, codify, and test clear escalation paths for human intervention.

Concept:

Ensure checkpoint requests cannot be dropped: first-line -> L2 analyst -> system owner, with policy-versioned deadlines and a fail-closed timeout action.

Terraform (PagerDuty) Example

# File: infrastructure/pagerduty_escalations.tf
# Render these values only after verifying the signed HITL policy and current directory export.
variable "route_hops" {
  type = list(object({
    subject_id    = string
    schedule_id   = string
    delay_minutes = number
  }))
  validation {
    condition     = length(var.route_hops) >= 2 && alltrue([for hop in var.route_hops : length(trimspace(hop.subject_id)) > 0 && length(trimspace(hop.schedule_id)) > 0 && hop.delay_minutes >= 0 && floor(hop.delay_minutes) == hop.delay_minutes])
    error_message = "route_hops must contain the ordered first-line and escalation schedules with non-negative integer delays"
  }
}
variable "route_loops" {
  type = number
  validation {
    condition     = var.route_loops >= 1 && floor(var.route_loops) == var.route_loops
    error_message = "route_loops must be a positive policy-derived integer"
  }
}
variable "decision_deadline_minutes" {
  type = number
  validation {
    condition     = var.decision_deadline_minutes > 0
    error_message = "decision_deadline_minutes must be positive"
  }
}

resource "pagerduty_escalation_policy" "ai_hitl_escalation" {
  name      = "AI HITL Escalation Policy"
  num_loops = var.route_loops
  dynamic "rule" {
    for_each = var.route_hops
    content {
      escalation_delay_in_minutes = rule.value.delay_minutes
      target {
        type = "schedule_reference"
        id   = rule.value.schedule_id
      }
    }
  }
  lifecycle {
    precondition {
      condition     = sum([for hop in var.route_hops : hop.delay_minutes]) < var.decision_deadline_minutes
      error_message = "the complete escalation route must occur before the terminal decision deadline"
    }
  }
}

Generate route_hops, route_loops, and the decision deadline from the same verified policy. Resolve every subject to a current schedule through identity-directory readback; the ordered list must begin with the policy's first-line route. Missing, duplicated, revoked, stale, or out-of-order subjects block deployment.

Direct Events API response verifier

# File: tools/verify_pagerduty_events_response.py
from __future__ import annotations

import argparse
import json
import sys


def positive_int(raw: str) -> int:
    value = int(raw)
    if value < 2:
        raise argparse.ArgumentTypeError("maximum response bytes must be at least two")
    return value


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


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


def read_bounded(limit: int) -> bytes:
    body = bytearray()
    while True:
        chunk = sys.stdin.buffer.read(min(64 * 1024, limit - len(body) + 1))
        if not chunk:
            return bytes(body)
        body.extend(chunk)
        if len(body) > limit:
            raise ValueError("PagerDuty response exceeds the signed byte bound")


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--expected-dedup-key", required=True)
    parser.add_argument("--max-response-bytes", required=True, type=positive_int)
    args = parser.parse_args()
    body = json.loads(
        read_bounded(args.max_response_bytes).decode("utf-8"),
        object_pairs_hook=reject_duplicate_keys,
        parse_constant=reject_nonfinite,
    )
    if (
        not isinstance(body, dict)
        or body.get("status") != "success"
        or body.get("dedup_key") != args.expected_dedup_key
    ):
        raise ValueError("PagerDuty receipt is invalid or bound to a different checkpoint")
    print(json.dumps({"status": "PASS", "dedup_key": args.expected_dedup_key}, sort_keys=True))


if __name__ == "__main__":
    main()

Direct Events API sender

set -euo pipefail
: "${PD_ROUTING_KEY:?injected by the secret manager}"
: "${M006003_CHECKPOINT_ID:?injected from verified signed HITL policy}"
: "${M006003_HITL_POLICY_VERSION:?injected from verified signed HITL policy}"
: "${M006003_ROUTE_ID:?injected from verified signed HITL policy}"
: "${M006003_EVENTS_CONNECT_TIMEOUT_SECONDS:?injected from verified signed HITL policy}"
: "${M006003_EVENTS_TOTAL_TIMEOUT_SECONDS:?injected from verified signed HITL policy}"
: "${M006003_EVENTS_MAX_RESPONSE_BYTES:?injected from verified signed HITL policy}"
export PD_ROUTING_KEY

checkpoint_instance_id="${1:?usage: send_hitl_event.sh CHECKPOINT_INSTANCE_ID}"
[[ "$checkpoint_instance_id" =~ ^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$ ]] || { printf '%s\n' "invalid checkpoint instance ID" >&2; exit 64; }
for value in "$M006003_EVENTS_CONNECT_TIMEOUT_SECONDS" "$M006003_EVENTS_TOTAL_TIMEOUT_SECONDS"; do
  jq -en --arg value "$value" '($value | test("^[0-9]+([.][0-9]+)?$")) and (($value | tonumber) > 0)' >/dev/null
done
[[ "$M006003_EVENTS_MAX_RESPONSE_BYTES" =~ ^[0-9]+$ ]] && (( M006003_EVENTS_MAX_RESPONSE_BYTES >= 2 )) || { printf '%s\n' "invalid response byte bound" >&2; exit 64; }
jq -en --arg connect "$M006003_EVENTS_CONNECT_TIMEOUT_SECONDS" --arg total "$M006003_EVENTS_TOTAL_TIMEOUT_SECONDS" '($total | tonumber) >= ($connect | tonumber)' >/dev/null

jq -n --arg dedup_key "$checkpoint_instance_id" --arg checkpoint_id "$M006003_CHECKPOINT_ID" --arg policy_version "$M006003_HITL_POLICY_VERSION" --arg route_id "$M006003_ROUTE_ID" '{routing_key: env.PD_ROUTING_KEY, event_action: "trigger", dedup_key: $dedup_key, payload: {summary: "HITL approval required", source: "hitl-service", severity: "critical", custom_details: {checkpoint_id: $checkpoint_id, checkpoint_instance_id: $dedup_key, hitl_policy_version: $policy_version, route_id: $route_id}}}' |
curl --fail-with-body --silent --show-error --proto '=https' --proto-redir '=https' --tlsv1.2 --location --max-redirs 0 --connect-timeout "$M006003_EVENTS_CONNECT_TIMEOUT_SECONDS" --max-time "$M006003_EVENTS_TOTAL_TIMEOUT_SECONDS" --max-filesize "$M006003_EVENTS_MAX_RESPONSE_BYTES" --header 'Content-Type: application/json' --header 'Accept: application/json' --header 'Accept-Encoding: identity' --data-binary @- 'https://events.pagerduty.com/v2/enqueue' |
python3 tools/verify_pagerduty_events_response.py --expected-dedup-key "$checkpoint_instance_id" --max-response-bytes "$M006003_EVENTS_MAX_RESPONSE_BYTES"

Action: Treat provider names as illustrative. Render the complete first-line-through-fallback route, hop deadlines, loop bound, and timeout default only from the verified signed policy and current directory; validate the Events API status and exact deduplication key before accepting notification delivery. Bind the policy version and route to every activation event, test each hop and timeout default with synthetic checkpoint instances, and retain the resulting linked event chain. A notification delivery receipt alone does not prove that the checkpoint blocked or that the final action followed the decision.

Verify safely

A read-only identity that cannot alter the controlled asset must obtain signed HITL and routing policies, identity-directory readback, immutable event stream, provider delivery receipts, and externally retained audit records; validate the effective asset, policy, input, and scope versions or digests; then resolve every route target, synthesize every branch and timeout, correlate the exact event chain by checkpoint instance, and compare provider delivery with append-only event readback.

AID-M-006.003-G002

Publish canonical HITL audit events to an external retention-locked store and independently reconcile lifecycle completeness without authorizing execution.

Canonical audit schema

Publish one schema for the HITL lifecycle so AID-H-018.003 can attach its independently signed gate receipt and AID-D-015 can consume the same fields without translation. The audit producer records observations only; an event or receipt in this chain is never itself an instruction to execute.

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "type": "object",
  "required": [
    "schema_version", "event_type", "event_id", "checkpoint_instance_id",
    "checkpoint_id", "action_id", "action_digest", "policy_version", "sequence",
    "occurred_at", "actor_id", "outcome", "causation_event_id",
    "gate_receipt_sha256"
  ],
  "properties": {
    "schema_version": {"const": "aidefend.hitl_audit_event.v3"},
    "event_type": {"enum": [
      "hitl_activation", "hitl_escalation", "hitl_decision",
      "hitl_timeout", "gate_receipt_observed", "final_action_observed"
    ]},
    "event_id": {"type": "string", "minLength": 16},
    "checkpoint_instance_id": {"type": "string", "minLength": 8},
    "checkpoint_id": {"type": "string", "minLength": 3},
    "action_id": {"type": "string", "minLength": 8},
    "action_digest": {"type": "string", "pattern": "^[a-f0-9]{64}$"},
    "policy_version": {"type": "string", "minLength": 1},
    "sequence": {"type": "integer", "minimum": 0},
    "occurred_at": {"type": "string", "format": "date-time"},
    "actor_id": {"type": "string", "minLength": 3},
    "outcome": {"enum": [
      "pending", "escalated", "approved", "rejected", "timeout",
      "allowed", "denied", "executed", "failed", "cancelled"
    ]},
    "causation_event_id": {"type": ["string", "null"]},
    "gate_receipt_sha256": {
      "type": ["string", "null"],
      "pattern": "^[a-f0-9]{64}$"
    }
  },
  "additionalProperties": false
}

Reconcile completeness without authorizing execution

from __future__ import annotations

# File: audit_verifier/reconcile_hitl_audit.py

from datetime import datetime
import re
from typing import Literal


Status = Literal["PASS", "FAIL", "INSUFFICIENT_DATA", "ERROR"]
EVENT_TYPES = {
    "hitl_activation", "hitl_escalation", "hitl_decision",
    "hitl_timeout", "gate_receipt_observed", "final_action_observed",
}


def parse_time(value: str) -> datetime:
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.utcoffset() is None:
        raise ValueError("event time lacks timezone")
    return parsed


def reconcile(events: list[dict], expected_policy_version: str) -> dict:
    required_fields = {
        "schema_version", "event_type", "event_id",
        "checkpoint_instance_id", "checkpoint_id", "action_id",
        "action_digest", "policy_version", "sequence", "occurred_at",
        "actor_id", "outcome", "causation_event_id",
        "gate_receipt_sha256",
    }
    hex_digest = re.compile(r"^[a-f0-9]{64}$")
    try:
        if not events:
            return {"status": "INSUFFICIENT_DATA", "reason": "no authenticated events"}
        seen_ids: set[str] = set()
        parsed_times: dict[int, datetime] = {}
        for event in events:
            if not isinstance(event, dict) or set(event) != required_fields:
                return {"status": "ERROR", "reason": "event schema differs"}
            if event["schema_version"] != "aidefend.hitl_audit_event.v3":
                return {"status": "ERROR", "reason": "event schema version differs"}
            if event["event_type"] not in EVENT_TYPES:
                return {"status": "ERROR", "reason": "unknown event type"}
            if (
                not isinstance(event["sequence"], int)
                or isinstance(event["sequence"], bool)
                or event["sequence"] < 0
            ):
                return {"status": "ERROR", "reason": "invalid event sequence"}
            event_id = event["event_id"]
            if not isinstance(event_id, str) or event_id in seen_ids:
                return {"status": "FAIL", "reason": "missing or duplicate event id"}
            if (
                not isinstance(event["action_digest"], str)
                or not hex_digest.fullmatch(event["action_digest"])
            ):
                return {"status": "ERROR", "reason": "invalid action digest"}
            parsed_times[event["sequence"]] = parse_time(event["occurred_at"])
            seen_ids.add(event_id)

        sequences = [event["sequence"] for event in events]
        if (
            len(sequences) != len(set(sequences))
            or sorted(sequences) != list(range(len(events)))
        ):
            return {"status": "FAIL", "reason": "event sequence is duplicated or incomplete"}
        ordered = sorted(events, key=lambda event: event["sequence"])
        if any(
            parsed_times[index] > parsed_times[index + 1]
            for index in range(len(ordered) - 1)
        ):
            return {"status": "FAIL", "reason": "event time contradicts canonical sequence"}

        bindings = {
            (
                event["checkpoint_instance_id"], event["checkpoint_id"],
                event["action_id"], event["action_digest"],
                event["policy_version"],
            )
            for event in ordered
        }
        if (
            len(bindings) != 1
            or next(iter(bindings))[-1] != expected_policy_version
        ):
            return {"status": "FAIL", "reason": "checkpoint, action, or policy binding mismatch"}

        types = [event["event_type"] for event in ordered]
        terminal_positions = [
            index for index, kind in enumerate(types)
            if kind in {"hitl_decision", "hitl_timeout"}
        ]
        if len(terminal_positions) != 1:
            return {"status": "FAIL", "reason": "missing or duplicate terminal human result"}
        terminal_index = terminal_positions[0]
        expected_types = (
            ["hitl_activation"]
            + ["hitl_escalation"] * (terminal_index - 1)
            + [types[terminal_index], "gate_receipt_observed", "final_action_observed"]
        )
        if types != expected_types:
            return {"status": "FAIL", "reason": "lifecycle event order or extras differ"}

        previous: str | None = None
        for event in ordered:
            if previous is None:
                if event["causation_event_id"] is not None:
                    return {"status": "FAIL", "reason": "activation is not the causal root"}
            elif event["causation_event_id"] != previous:
                return {"status": "FAIL", "reason": "broken causation chain"}
            previous = event["event_id"]

        if ordered[0]["outcome"] != "pending" or any(
            event["outcome"] != "escalated"
            for event in ordered[1:terminal_index]
        ):
            return {"status": "FAIL", "reason": "activation or escalation outcome differs"}
        terminal_event = ordered[terminal_index]
        gate_event = ordered[-2]
        final_event = ordered[-1]
        if (
            terminal_event["event_type"] == "hitl_decision"
            and terminal_event["outcome"] not in {"approved", "rejected"}
        ):
            return {"status": "FAIL", "reason": "decision outcome differs"}
        if (
            terminal_event["event_type"] == "hitl_timeout"
            and terminal_event["outcome"] != "timeout"
        ):
            return {"status": "FAIL", "reason": "timeout outcome differs"}
        receipt = gate_event["gate_receipt_sha256"]
        if not isinstance(receipt, str) or not hex_digest.fullmatch(receipt):
            return {"status": "INSUFFICIENT_DATA", "reason": "gate receipt is unavailable or malformed"}
        if final_event["gate_receipt_sha256"] != receipt:
            return {"status": "FAIL", "reason": "final observation is not bound to the gate receipt"}
        if any(event["gate_receipt_sha256"] is not None for event in ordered[:-2]):
            return {"status": "FAIL", "reason": "pre-gate event contains a gate receipt"}

        approved = terminal_event["outcome"] == "approved"
        gate = gate_event["outcome"]
        final = final_event["outcome"]
        if gate not in {"allowed", "denied"}:
            return {"status": "FAIL", "reason": "gate outcome differs"}
        if final not in {"executed", "failed", "cancelled"}:
            return {"status": "FAIL", "reason": "final outcome differs"}
        if gate == "allowed" and not approved:
            return {"status": "FAIL", "reason": "gate allowed without human approval"}
        if final == "executed" and (not approved or gate != "allowed"):
            return {"status": "FAIL", "reason": "execution lacks approval and allowing gate"}
        if gate == "denied" and final != "cancelled":
            return {"status": "FAIL", "reason": "denied action was not stopped before dispatch"}
        return {
            "status": "PASS",
            "checkpoint_instance_id": ordered[0]["checkpoint_instance_id"],
            "gate_receipt_sha256": receipt,
            "event_count": len(ordered),
        }
    except (KeyError, TypeError, ValueError) as exc:
        return {"status": "ERROR", "reason": str(exc)}

Interpret decisions separately

Human approval, the runtime gate decision, and the observed final action are distinct. Approval may still lead to denial after revocation, expiry, or another safety check, or to cancellation or failure after an allowing gate. The gate must validate approval for the exact action at dispatch; this audit verifier does not issue or renew that authority. In this example, cancelled means no action was dispatched (including a denied checkpoint); failed is an admitted attempt that failed and may have partial effects. Never relabel partial execution as cancellation. Recovery for partial effects is separate from lifecycle completeness. A completed chain can therefore PASS with a failed operation; that does not mean the operation succeeded.

Evidence flow and responsibilities

An external writer validates the exact schema, signs canonical event-batch bytes, and stores them under retention selected by organizational policy in append-only storage that the checkpoint, agent, and executor identities cannot modify. A read-only verifier retrieves the batch and its population manifest, verifies the signature over stable bytes before parsing, rejects missing or extra manifest members, reruns reconciliation, and signs its result. PASS proves audit-chain completeness only. It does not authorize execution and must never be accepted by an executor in place of the AID-H-018.003 gate receipt.

Action: Emit and retain the v3 audit events for every checkpoint lifecycle, bind the observed gate event to the exact H-018.003 receipt digest, and provide the independently replayed chain to AID-D-015.

Open-source tools

OpenTelemetry Collector
Vector
Sigstore Rekor v2 (signed audit-event digest transparency anchoring)
python-jsonschema

Commercial / hosted tools

PagerDuty Operations Cloud
Amazon S3 Object Lock
Azure Immutable Blob Storage
Google Cloud Bucket Lock
AWS Key Management Service
Azure Key Vault
Google Cloud Key Management Service

Threat mappings

MITRE ATLAS

AML.T0048 External Harms (timeout-bounded escalation enables timely intervention before harmful actions proceed)
AML.T0053 AI Agent Tool Invocation (durable lifecycle evidence makes HITL bypass independently detectable)
AML.T0101 Data Destruction via AI Agent Tool Invocation (escalation and decision evidence expose destructive-action bypass)

MAESTRO

Repudiation (L7)
Poisoning Observability Data (L5) (externally signed, retention-locked HITL events resist silent audit-history alteration)

OWASP LLM Top 10 2026

LLM03:2026 Excessive Agency (tested escalation and audit-lifecycle evidence supports the independent approval gate but does not authorize or deny the action itself)

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

ASI02:2026 Tool Misuse and Exploitation (high-impact tool actions are bound to an escalation and decision record)
ASI08:2026 Cascading Failures (escalation prevents unhandled HITL events from compounding)
ASI03:2026 Identity and Privilege Abuse (operator identity is bound to the immutable approval record)

NIST Adversarial Machine Learning 2025

NISTAML.039 Compromising connected resources (monitoring and escalation evidence exposes risky connected-action bypass)
NISTAML.027 Misaligned Outputs (timeout-bounded escalation supports intervention before a high-impact action proceeds)

Cisco Integrated AI Security and Safety Framework

AISubtech-2.1.1 Execution Approval Bypass (Reconciles human decisions and signed gate receipts with the observed final action)
AITech-2.1 Excessive Agency (Reconciles human decisions and signed gate receipts with the observed final action)

Google Secure AI Framework 2.0 - Risks

RA: Rogue Actions (timeout-bounded escalation evidence reveals unattended high-impact actions)

Databricks AI Security Framework 3.0

Model Serving - Inference requests 9.13: Excessive agency
Agents - Core 13.2: Tool Misuse
Agents - Core 13.8: Repudiation & Untraceability
Agents - Core 13.10: Overwhelming Human in the Loop