AID-I-005: Emergency "Kill-Switch" / AI System Halt

Version 1.20261015 · Actionable technique

Establish and maintain a reliable, rapidly invokable mechanism to immediately halt, disable, or severely restrict the operation of an AI model or autonomous agent if it exhibits confirmed critical malicious behavior, behaves as a rogue agent (acts far outside its intended parameters in a harmful way), or if a severe, ongoing attack is detected and other containment measures are insufficient. This is a last-resort containment measure designed to prevent catastrophic harm or further compromise.

Pillars: infra, app

Lifecycle: response

Scope boundary

Owns the authoritative emergency RUN, HALT, or severely restricted operating decision and its verified enforcement at every required inference and agent-loop boundary. It is a last-resort containment mechanism; it does not itself revoke credentials, terminate every runtime object, or delete durable application state.

AID-I-003: AID-I-005 halts or severely restricts a complete system or tenant under an emergency decision; AID-I-003 applies bounded quarantine, degradation, throttling, or action containment to scoped targets. Use I-003 when limited reversible containment is sufficient and I-005 when policy requires a last-resort halt with complete propagation and restart-gate evidence.

AID-E-002: AID-I-005 publishes and enforces the emergency operating state; AID-E-002 terminates exact active processes, agent runs, execution leases, jobs, pods, and containers. A HALT decision can block execution gates without proving that every live runtime object was terminated, so E-002 supplies the separate eviction receipt when required.

Implementation guidance

AID-I-005-G001

Automated safety monitors and triggers for critical deviations.
Monitor scope and responsibilities

Run the monitor outside the agent and model-controlled configuration. Rules, thresholds, measurement windows, minimum coverage, scope expansion, and fail-safe behavior come from a signature-verified, versioned environment policy—never from framework-wide sample numbers. Trusted telemetry must identify the exact scope and complete observed population. A breached rule is FAIL even when containment succeeds; record containment as a separate result so a successful HALT cannot hide the finding. This step requests the canonical I-005 HALT decision and independently reads propagation; the sibling receipt remains a separately verified input.

Policy-driven monitor
# File: safety_monitor/policy_monitor.py
from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass
from decimal import Decimal
from typing import Mapping, Sequence, Callable


@dataclass(frozen=True)
class PolicyVerifier:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    verify_monitor_policy: Callable[[str], Mapping[str, object]]


@dataclass(frozen=True)
class TelemetryReader:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    measure: Callable[[str, str, Mapping[str, object]], Mapping[str, object]]


@dataclass(frozen=True)
class MeasurementVerifier:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    verify: Callable[[Mapping[str, object], Mapping[str, object], str], None]


@dataclass(frozen=True)
class HaltAuthority:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    request_halt: Callable[[str, str, str, str, str], Mapping[str, object]]


@dataclass(frozen=True)
class BoundaryReader:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    reconcile: Callable[[str, Mapping[str, object]], Mapping[str, object]]


@dataclass(frozen=True)
class EvidenceClient:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    append: Callable[[Mapping[str, object]], Mapping[str, object]]
    read: Callable[[str], Mapping[str, object]]


@dataclass(frozen=True)
class Dependencies:
    policies: PolicyVerifier
    telemetry: TelemetryReader
    measurements: MeasurementVerifier
    halts: HaltAuthority
    boundaries: BoundaryReader
    evidence: EvidenceClient


def canonical_sha256(value: object) -> str:
    encoded = json.dumps(
        value, sort_keys=True, separators=(",", ":"), ensure_ascii=True
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def compare(value: Decimal, operator: str, threshold: Decimal) -> bool:
    operations = {
        "GT": lambda: value > threshold,
        "GTE": lambda: value >= threshold,
        "LT": lambda: value < threshold,
        "LTE": lambda: value <= threshold,
    }
    try:
        return bool(operations[operator]())
    except KeyError as exc:
        raise ValueError("unsupported signed-policy comparator") from exc


def evaluate_once(
    deps: Dependencies,
    *,
    policy_jws: str,
    evaluation_id: str,
) -> Mapping[str, object]:
    policy = deps.policies.verify_monitor_policy(policy_jws)
    evaluations: list[dict[str, object]] = []
    breached_scopes: dict[str, list[str]] = {}
    unverifiable_scopes: dict[str, list[str]] = {}

    for rule in policy["rules"]:
        for scope in rule["scopes"]:
            measurement = deps.telemetry.measure(
                metric_id=str(rule["metric_id"]),
                scope=str(scope),
                window=rule["window"],
            )
            try:
                deps.measurements.verify(
                    measurement, rule=rule, scope=str(scope)
                )
                expected = int(measurement["expected_count"])
                observed = int(measurement["observed_count"])
                coverage = (
                    Decimal(observed) / Decimal(expected)
                    if expected > 0 else Decimal(0)
                )
                if coverage < Decimal(str(rule["minimum_coverage"])):
                    raise ValueError("telemetry population coverage is incomplete")
                breached = compare(
                    Decimal(str(measurement["value"])),
                    str(rule["operator"]),
                    Decimal(str(rule["threshold"])),
                )
                state = "BREACH" if breached else "WITHIN_POLICY"
                if breached:
                    breached_scopes.setdefault(str(scope), []).append(
                        str(rule["rule_id"])
                    )
            except Exception as exc:
                state = "UNVERIFIABLE"
                unverifiable_scopes.setdefault(str(scope), []).append(
                    str(rule["rule_id"])
                )
                measurement = {
                    "error_class": type(exc).__name__,
                    "measurement_sha256": canonical_sha256(measurement),
                }
            evaluations.append({
                "rule_id": rule["rule_id"],
                "scope": scope,
                "state": state,
                "measurement": measurement,
            })

    halt_targets = set(breached_scopes)
    for scope in unverifiable_scopes:
        if policy["unverifiable_action"] == "HALT":
            halt_targets.add(scope)

    containment: list[dict[str, object]] = []
    evaluation_sha256 = canonical_sha256(evaluations)
    for scope in sorted(halt_targets):
        reason = {
            "evaluation_id": evaluation_id,
            "scope": scope,
            "breached_rule_ids": sorted(breached_scopes.get(scope, [])),
            "unverifiable_rule_ids": sorted(
                unverifiable_scopes.get(scope, [])
            ),
            "evaluation_sha256": evaluation_sha256,
        }
        halt = deps.halts.request_halt(
            idempotency_key=canonical_sha256(reason),
            scope=scope,
            reason_sha256=canonical_sha256(reason),
            policy_sha256=str(policy["policy_sha256"]),
            evaluation_id=evaluation_id,
        )
        propagation = deps.boundaries.reconcile(
            scope=scope, halt_receipt=halt
        )
        containment.append({
            "scope": scope,
            "halt_receipt": halt,
            "propagation": propagation,
        })

    if breached_scopes:
        outcome = "FAIL"
    elif unverifiable_scopes:
        outcome = "INSUFFICIENT_DATA"
    else:
        outcome = "PASS"
    event = {
        "schema_version": 1,
        "control_id": "AID-I-005",
        "guidance_ref": "AID-I-005-G001",
        "evaluation_id": evaluation_id,
        "policy_sha256": policy["policy_sha256"],
        "evaluation_sha256": evaluation_sha256,
        "evaluation_count": len(evaluations),
        "breached_scopes": breached_scopes,
        "unverifiable_scopes": unverifiable_scopes,
        "containment": containment,
        "outcome": outcome,
    }
    receipt = deps.evidence.append(event)
    readback = deps.evidence.read(str(receipt["receipt_id"]))
    if readback["event_sha256"] != receipt["event_sha256"]:
        raise RuntimeError("evidence readback mismatch")
    return {**event, "evidence_receipt_id": receipt["receipt_id"]}
Deployment requirements

The adapters must pin policy issuer/audience/type/algorithm, key state, monotonic version and applicability; reject NaN/infinity, mixed units, mismatched scope, stale windows, partial populations, collector rollback, duplicate/same-version mutation, and unsigned metrics. The halt adapter calls the canonical authority used by every I-005 enforcement boundary and supplies an idempotency key; the boundary reader independently enumerates the signed expected boundary set and reads each current decision/version. Persist the evaluation and evidence before reporting. If halt/evidence/reconciliation fails, the outcome is ERROR and the monitor keeps requesting fail-safe containment according to policy. Schedule cadence, thresholds, windows, and evidence retention all come from the signed policy.

AID-I-005-G002

Provide secure, MFA-protected manual override for human operators.
Manual HALT-only boundary

Expose a private, rate-limited HALT endpoint that cannot issue RUN or bypass the post-halt restart gate. Authenticate the operator with a pinned issuer/audience/algorithm and policy-required phishing-resistant step-up, derive tenant authority from verified entitlements rather than request fields, and consume a separate co-approval when signed policy requires it. MFA freshness, accepted assurance methods, co-approval count, scopes, and propagation timeout are policy values—not framework constants.

FastAPI control adapter
# File: admin/manual_halt.py
# Dependencies: fastapi>=0.115,<1, pydantic>=2.9,<3
from __future__ import annotations

import hashlib as _aidefend_policy_hashlib
import json as _aidefend_policy_json
import math as _aidefend_policy_math
import os as _aidefend_policy_os
import subprocess as _aidefend_policy_subprocess
import tempfile as _aidefend_policy_tempfile
from pathlib import Path as _AidefendPolicyPath


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


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


def _aidefend_load_signed_policy(*, path_env, signature_env, key_env, schema_version):
    policy_path = _AidefendPolicyPath(_aidefend_policy_os.environ[path_env])
    signature_path = _AidefendPolicyPath(_aidefend_policy_os.environ[signature_env])
    verify_timeout = float(_aidefend_policy_os.environ["AIDEFEND_POLICY_VERIFY_TIMEOUT_SECONDS"])
    if not _aidefend_policy_math.isfinite(verify_timeout) or verify_timeout <= 0:
        raise RuntimeError("policy-verification timeout is invalid")
    payload = policy_path.read_bytes()
    signature = signature_path.read_bytes()
    if not payload or not signature:
        raise RuntimeError("signed policy payload or signature is empty")
    with _aidefend_policy_tempfile.TemporaryDirectory(prefix="aidefend-policy-") as directory:
        root = _AidefendPolicyPath(directory)
        _aidefend_policy_os.chmod(root, 0o700)
        payload_path = root / "policy.json"
        signature_path = root / "policy.sig"
        payload_path.write_bytes(payload)
        signature_path.write_bytes(signature)
        _aidefend_policy_os.chmod(payload_path, 0o400)
        _aidefend_policy_os.chmod(signature_path, 0o400)
        _aidefend_policy_subprocess.run(
            ["cosign", "verify-blob", "--key", _aidefend_policy_os.environ[key_env],
             "--bundle", str(signature_path), str(payload_path)],
            check=True, capture_output=True, text=True, timeout=verify_timeout,
        )
        verified = payload_path.read_bytes()
        if verified != payload:
            raise RuntimeError("verified policy snapshot changed")
    policy = _aidefend_policy_json.loads(
        verified.decode("utf-8", errors="strict"),
        object_pairs_hook=_aidefend_reject_duplicate_keys,
        parse_constant=_aidefend_reject_nonfinite,
    )
    if (not isinstance(policy, dict) or policy.get("schema_version") != schema_version
            or not isinstance(policy.get("policy_version"), str) or not policy["policy_version"]):
        raise RuntimeError("signature-verified policy schema or version is invalid")
    if "_aidefend_verified_sha256" in policy:
        raise RuntimeError("signed policy uses a reserved evidence field")
    policy["_aidefend_verified_sha256"] = _aidefend_policy_hashlib.sha256(verified).hexdigest()
    return policy


import hashlib
import json
from dataclasses import asdict, dataclass
from typing import Literal, Mapping, Sequence, Callable

from fastapi import Depends, FastAPI, Header, HTTPException
from pydantic import BaseModel, Field, model_validator


@dataclass(frozen=True)
class VerifiedOperator:
    subject: str
    issuer: str
    authentication_event_id: str
    assurance_method: str
    authenticated_at_epoch: int
    authorized_tenant_ids: tuple[str, ...]
    scopes: tuple[str, ...]
    token_jti: str


@dataclass(frozen=True)
class IdentityVerifier:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    verify_operator: Callable[[str, Mapping[str, object]], VerifiedOperator]


@dataclass(frozen=True)
class PolicyVerifier:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    current_manual_halt_policy: Callable[[], Mapping[str, object]]


@dataclass(frozen=True)
class CoApprovalClient:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    consume: Callable[[str, VerifiedOperator, Mapping[str, object]], Mapping[str, object]]


@dataclass(frozen=True)
class RequestLedger:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    reserve: Callable[[str, str, str], Mapping[str, object]]
    finish: Callable[[str, Mapping[str, object]], None]


@dataclass(frozen=True)
class HaltAuthority:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    issue_halt: Callable[[str, Mapping[str, object], VerifiedOperator, Mapping[str, object], Mapping[str, object]], Mapping[str, object]]


@dataclass(frozen=True)
class BoundaryReader:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    reconcile: Callable[[str, Mapping[str, object], Mapping[str, object]], Mapping[str, object]]


@dataclass(frozen=True)
class EvidenceClient:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    append: Callable[[Mapping[str, object]], Mapping[str, object]]
    read: Callable[[str], Mapping[str, object]]


class Dependencies:
    policies: PolicyVerifier
    identities: IdentityVerifier
    approvals: CoApprovalClient
    ledger: RequestLedger
    halts: HaltAuthority
    boundaries: BoundaryReader
    evidence: EvidenceClient


_deps: Dependencies | None = None
app = FastAPI()


def configure_dependencies(value: Dependencies) -> None:
    """Called exactly once by the deployment composition root."""
    global _deps
    if _deps is not None:
        raise RuntimeError("manual HALT dependencies already configured")
    _deps = value


def runtime_dependencies() -> Dependencies:
    if _deps is None:
        raise RuntimeError("manual HALT dependencies are not configured")
    return _deps


MANUAL_HALT_SCHEMA_POLICY = _aidefend_load_signed_policy(
    path_env="I005_HALT_SCHEMA_POLICY_PATH",
    signature_env="I005_HALT_SCHEMA_POLICY_SIGNATURE",
    key_env="I005_HALT_SCHEMA_POLICY_VERIFY_KEY",
    schema_version="aidefend.manual-halt-request-schema-policy.v1",
)
_HALT_BOUND_NAMES = (
    "tenant_id_min_length", "tenant_id_max_length",
    "incident_ref_min_length", "incident_ref_max_length",
    "justification_min_length", "justification_max_length",
)
HALT_BOUNDS = {name: MANUAL_HALT_SCHEMA_POLICY.get(name) for name in _HALT_BOUND_NAMES}
if any(
    isinstance(value, bool) or not isinstance(value, int) or value < 1
    for value in HALT_BOUNDS.values()
):
    raise RuntimeError("signed manual-HALT schema bounds are invalid")
for minimum_name, maximum_name in (
    ("tenant_id_min_length", "tenant_id_max_length"),
    ("incident_ref_min_length", "incident_ref_max_length"),
    ("justification_min_length", "justification_max_length"),
):
    if HALT_BOUNDS[minimum_name] > HALT_BOUNDS[maximum_name]:
        raise RuntimeError("signed manual-HALT schema minimum exceeds maximum")


class HaltRequest(BaseModel):
    request_id: str = Field(pattern=r"^[0-9a-f]{32}$")
    scope_type: Literal["GLOBAL", "TENANT"]
    tenant_id: str | None = Field(default=None, min_length=HALT_BOUNDS["tenant_id_min_length"], max_length=HALT_BOUNDS["tenant_id_max_length"])
    incident_ref: str = Field(min_length=HALT_BOUNDS["incident_ref_min_length"], max_length=HALT_BOUNDS["incident_ref_max_length"])
    justification: str = Field(min_length=HALT_BOUNDS["justification_min_length"], max_length=HALT_BOUNDS["justification_max_length"])

    @model_validator(mode="after")
    def validate_scope(self):
        if (self.scope_type == "TENANT") != (self.tenant_id is not None):
            raise ValueError("tenant_id must appear exactly for TENANT scope")
        return self


def canonical_sha256(value: object) -> str:
    encoded = json.dumps(
        value, sort_keys=True, separators=(",", ":"), ensure_ascii=True
    ).encode("utf-8")
    return hashlib.sha256(encoded).hexdigest()


def bearer_value(authorization: str = Header(alias="Authorization")) -> str:
    scheme, separator, token = authorization.partition(" ")
    if separator != " " or scheme.lower() != "bearer" or not token:
        raise HTTPException(status_code=401, detail="bearer token required")
    return token


def verified_context(
    token: str = Depends(bearer_value),
) -> tuple[VerifiedOperator, Mapping[str, object]]:
    runtime = runtime_dependencies()
    policy = runtime.policies.current_manual_halt_policy()
    try:
        operator = runtime.identities.verify_operator(token, policy)
    except Exception as exc:
        raise HTTPException(status_code=401, detail="operator verification failed") from exc
    required_scope = str(policy["required_scope"])
    if required_scope not in operator.scopes:
        raise HTTPException(status_code=403, detail="manual HALT scope required")
    return operator, policy


@app.post("/v1/admin/manual-halt", status_code=201)
def manual_halt(
    body: HaltRequest,
    context: tuple[VerifiedOperator, Mapping[str, object]] = Depends(
        verified_context
    ),
):
    runtime = runtime_dependencies()
    operator, policy = context
    if (
        body.scope_type == "TENANT"
        and body.tenant_id not in operator.authorized_tenant_ids
    ):
        raise HTTPException(status_code=403, detail="tenant scope not authorized")

    scope = "global" if body.scope_type == "GLOBAL" else (
        "tenant:" + str(body.tenant_id)
    )
    action = {
        "schema_version": 1,
        "action": "HALT",
        "scope": scope,
        "request_id": body.request_id,
        "incident_ref": body.incident_ref,
        "justification_sha256": hashlib.sha256(
            body.justification.encode("utf-8")
        ).hexdigest(),
        "operator_subject": operator.subject,
        "operator_issuer": operator.issuer,
        "authentication_event_id": operator.authentication_event_id,
        "policy_sha256": policy["policy_sha256"],
    }
    action_sha256 = canonical_sha256(action)
    reservation = runtime.ledger.reserve(
        request_id=body.request_id,
        token_jti=operator.token_jti,
        action_sha256=action_sha256,
    )
    try:
        coapproval = runtime.approvals.consume(
            action_sha256=action_sha256,
            operator=operator,
            policy=policy,
        )
        halt = runtime.halts.issue_halt(
            idempotency_key=str(reservation["reservation_id"]),
            action=action,
            operator=operator,
            coapproval_receipt=coapproval,
            policy=policy,
        )
        propagation = runtime.boundaries.reconcile(
            scope=scope,
            halt_receipt=halt,
            policy=policy,
        )
        event = {
            "schema_version": 1,
            "control_id": "AID-I-005",
            "guidance_ref": "AID-I-005-G002",
            "action": action,
            "action_sha256": action_sha256,
            "operator": {
                "subject": operator.subject,
                "issuer": operator.issuer,
                "authentication_event_id": operator.authentication_event_id,
                "assurance_method": operator.assurance_method,
            },
            "coapproval_receipt": coapproval,
            "halt_receipt": halt,
            "propagation": propagation,
            "outcome": "PASS",
        }
        evidence = runtime.evidence.append(event)
        readback = runtime.evidence.read(str(evidence["receipt_id"]))
        if readback["event_sha256"] != evidence["event_sha256"]:
            raise RuntimeError("evidence readback mismatch")
        result = {
            "scope": scope,
            "halt_decision_id": halt["decision_id"],
            "halt_version": halt["version"],
            "evidence_receipt_id": evidence["receipt_id"],
            "outcome": "PASS",
        }
        runtime.ledger.finish(str(reservation["reservation_id"]), result)
        return result
    except Exception as exc:
        error = {
            "outcome": "ERROR",
            "error_class": type(exc).__name__,
            "action_sha256": action_sha256,
        }
        runtime.ledger.finish(str(reservation["reservation_id"]), error)
        raise HTTPException(
            status_code=503,
            detail="HALT was requested but complete verified propagation is unavailable",
        ) from exc
Required adapters and tests

Back the request ledger with a durable conditional insert so one request/token action is consumed once and retries return the stored result. The identity adapter uses a locally pinned/refreshed key set or trusted gateway assertion with mTLS provenance; it must not perform arbitrary jku/x5u fetches. The co-approval receipt binds different operator identity, action digest, scope, policy, expiry, and one-time ID; when not required it returns an explicit signed N/A policy receipt. The halt authority emits the canonical signed monotonic decision consumed by I-005 boundaries. The boundary reader independently compares expected and observed boundary populations. Preserve local ledger state even when the external evidence sink fails. Test stale/forged MFA, wrong audience/tenant, replayed token/request/approval, same-operator co-approval, policy rollback, partial propagation, identity/approval/evidence outage, and retry after each crash point. Never add a RUN/unhalt path to this endpoint.

AID-I-005-G003

Run an external watchdog that issues a signed scope-bound HALT decision when trusted liveness or lease evidence breaches signed policy, and request separately owned live-execution termination.
Watchdog scope and deployment

The watchdog runs outside the agent pod/process and does not trust an agent-owned thread or self-reported health flag. It evaluates a signed liveness receipt from the trusted dispatcher or orchestrator against a signed policy. On breach or unverifiable liveness it asks the canonical halt authority to publish a scope-bound HALT and verifies every work-admission boundary. Process and worker termination remain owned by AID-E-002; the watchdog requests that action and retains its receipt without claiming the termination itself.

External watchdog reconcile loop
# File: isolate/external_agent_watchdog.py
from __future__ import annotations

import hashlib
import json
from datetime import datetime, timezone
from dataclasses import dataclass
from typing import Callable


@dataclass(frozen=True)
class SignedPolicyClient:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    current_and_verify: Callable[[str], dict]


@dataclass(frozen=True)
class LivenessClient:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    read_and_verify: Callable[[str, str], dict]


@dataclass(frozen=True)
class HaltAuthority:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    publish_and_verify: Callable[[dict], dict]
    verify_all_boundaries: Callable[[dict], dict]


@dataclass(frozen=True)
class EvidenceSpool:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    append_fsync_and_readback: Callable[[dict], dict]


@dataclass(frozen=True)
class EvidenceSink:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    append_and_verify: Callable[[dict], dict]


@dataclass(frozen=True)
class EvictionOwner:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    request_and_verify: Callable[[dict], dict]


def canonical_sha256(value: object) -> str:
    return "sha256:" + hashlib.sha256(
        json.dumps(
            value, sort_keys=True, separators=(",", ":"), allow_nan=False
        ).encode("utf-8")
    ).hexdigest()


def parse_utc(value: str) -> datetime:
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        raise ValueError("liveness timestamp must be timezone-aware")
    return parsed


def reconcile_workload(
    *,
    workload_id: str,
    scope: str,
    policy_client: SignedPolicyClient,
    liveness_client: LivenessClient,
    halt_authority: HaltAuthority,
    evidence_spool: EvidenceSpool,
    evidence_sink: EvidenceSink,
    eviction_owner: EvictionOwner,
) -> dict:
    now = datetime.now(timezone.utc)
    policy = policy_client.current_and_verify("agent-external-watchdog")
    if (
        policy.get("schema_version") != "aidefend.watchdog-policy.v1"
        or not policy.get("policy_sha256")
        or not isinstance(policy.get("heartbeat_max_age_seconds"), int)
        or isinstance(policy["heartbeat_max_age_seconds"], bool)
        or policy["heartbeat_max_age_seconds"] < 1
    ):
        raise RuntimeError("signed watchdog policy invalid")

    outcome = "ERROR"
    reason = "liveness_unverifiable"
    liveness = None
    try:
        liveness = liveness_client.read_and_verify(workload_id, scope)
        required = {
            "schema_version", "workload_id", "scope", "lease_id",
            "lease_version", "last_dispatch_at", "observed_at",
            "dispatcher_identity", "population_sha256", "population_count",
            "receipt_sha256",
        }
        if (
            not isinstance(liveness, dict)
            or set(liveness) != required
            or liveness["schema_version"] != "aidefend.dispatch-liveness.v1"
            or liveness["workload_id"] != workload_id
            or liveness["scope"] != scope
            or liveness["population_count"] < 1
        ):
            raise ValueError("signed liveness receipt schema or binding invalid")
        last_dispatch_at = parse_utc(liveness["last_dispatch_at"])
        observed_at = parse_utc(liveness["observed_at"])
        if observed_at > now or last_dispatch_at > observed_at:
            raise ValueError("liveness receipt time ordering invalid")
        age_seconds = (now - last_dispatch_at).total_seconds()
        if age_seconds > policy["heartbeat_max_age_seconds"]:
            outcome = "FAIL"
            reason = "trusted_dispatch_liveness_expired"
        else:
            outcome = "PASS"
            reason = "trusted_dispatch_liveness_within_policy"
    except Exception as exc:
        outcome = "ERROR"
        reason = "liveness_" + type(exc).__name__

    local = evidence_spool.append_fsync_and_readback({
        "schema_version": "aidefend.external-watchdog-event.v1",
        "workload_id": workload_id,
        "scope": scope,
        "policy_sha256": policy["policy_sha256"],
        "liveness_receipt_sha256": (
            liveness.get("receipt_sha256") if isinstance(liveness, dict)
            else None
        ),
        "liveness_population_sha256": (
            liveness.get("population_sha256") if isinstance(liveness, dict)
            else None
        ),
        "outcome": outcome,
        "reason": reason,
        "evaluated_at": now.isoformat(),
    })
    external = None
    try:
        external = evidence_sink.append_and_verify(local)
    except Exception:
        external = None
        outcome = "ERROR"
        reason = "external_evidence_unavailable"
        local = evidence_spool.append_fsync_and_readback({
            "schema_version": "aidefend.external-watchdog-event.v1",
            "workload_id": workload_id,
            "scope": scope,
            "policy_sha256": policy["policy_sha256"],
            "prior_local_receipt_sha256": local["receipt_sha256"],
            "outcome": outcome,
            "reason": reason,
            "evaluated_at": datetime.now(timezone.utc).isoformat(),
        })

    if outcome == "PASS":
        return {
            "outcome": "PASS",
            "local_evidence_receipt_sha256": local["receipt_sha256"],
            "external_evidence_receipt_sha256": (
                external["receipt_sha256"] if external else None
            ),
        }

    halt = halt_authority.publish_and_verify({
        "schema_version": "aidefend.watchdog-halt-request.v1",
        "scope": scope,
        "workload_id": workload_id,
        "reason": reason,
        "policy_sha256": policy["policy_sha256"],
        "local_evidence_receipt_sha256": local["receipt_sha256"],
        "idempotency_key": canonical_sha256({
            "scope": scope,
            "workload_id": workload_id,
            "lease_version": (
                liveness.get("lease_version")
                if isinstance(liveness, dict) else None
            ),
            "policy_sha256": policy["policy_sha256"],
        }),
    })
    propagation = halt_authority.verify_all_boundaries({
        "scope": scope,
        "halt_decision_sha256": halt["decision_sha256"],
        "halt_version": halt["version"],
        "expected_boundary_manifest_sha256": (
            policy["boundary_manifest_sha256"]
        ),
    })
    if (
        propagation.get("outcome") != "PASS"
        or propagation.get("residual_boundary_ids")
    ):
        raise RuntimeError("HALT did not reach every admission boundary")
    eviction = eviction_owner.request_and_verify({
        "owner": "AID-E-002",
        "scope": scope,
        "workload_ids": [workload_id],
        "halt_decision_sha256": halt["decision_sha256"],
        "idempotency_key": halt["decision_sha256"],
    })
    return {
        "outcome": outcome,
        "halt_decision_sha256": halt["decision_sha256"],
        "halt_version": halt["version"],
        "propagation_receipt_sha256": propagation["receipt_sha256"],
        "eviction_request_receipt_sha256": eviction["receipt_sha256"],
        "local_evidence_receipt_sha256": local["receipt_sha256"],
        "external_evidence_receipt_sha256": (
            external["receipt_sha256"] if external else None
        ),
    }

Trusted heartbeat rule: the dispatcher updates liveness only after it admits and observes work under verified workload identity; the model process cannot write, extend, or choose the lease. The liveness service signs the complete dispatcher-event population and a monotonic lease version. The local spool runs under the watchdog identity on a separate durable volume, signs records, uses append/fsync/readback, and replays to external append-only evidence. Policy supplies timeout, scope inventory, freshness, and retry values; the framework does not publish universal values.

AID-I-005-G004

Enforce a machine-verifiable post-halt restart gate that admits safe mode first and issues a new signed RUN decision only after exact remediation, regression, approval, and readback evidence passes.
Restart-gate responsibilities and sequence

A post-halt restart is a machine-enforced transition, not a document checklist. I-005 verifies that the exact halted scope has current signed owner receipts, admits only a signed safe-mode capability set, executes the signed regression population, consumes a one-time restart approval, and asks the halt authority to publish the next RUN version. It does not claim artifact restoration, credential cleanup, process eviction, or memory cleanup; their canonical owners supply independently verified receipts.

Signed restart manifest
{
  "schema_version": "aidefend.restart-manifest.v1",
  "restart_id": "incident-2026-0716-restart-01",
  "incident_id": "incident-2026-0716",
  "scope": {"type": "tenant", "tenant_ref": "hmac-sha256:tenant-ref"},
  "halt_decision_sha256": "sha256:current-halt-decision",
  "halt_version": 42,
  "restart_policy_sha256": "sha256:signed-restart-policy",
  "safe_mode_capability_sha256": "sha256:signed-safe-mode-capabilities",
  "owner_receipt_manifest_sha256": "sha256:exact-owner-receipt-manifest",
  "regression_population_sha256": "sha256:exact-regression-case-manifest",
  "approval_jti": "one-time-restart-approval-jti"
}
Machine-verifiable restart gate
# File: isolate/post_halt_restart_gate.py
from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass
from typing import Callable


@dataclass(frozen=True)
class ReceiptVerifier:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    verify_owner_set: Callable[[dict, str], dict]


@dataclass(frozen=True)
class HaltAuthority:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    current_and_verify: Callable[[dict], dict]
    publish_run_and_verify: Callable[[dict], dict]


@dataclass(frozen=True)
class SafeModeGate:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    deploy_and_readback: Callable[[dict], dict]


@dataclass(frozen=True)
class RegressionRunner:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    run_and_verify: Callable[[dict], dict]


@dataclass(frozen=True)
class ApprovalLedger:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    verify_and_consume: Callable[[str, dict], dict]


@dataclass(frozen=True)
class Evidence:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    append_and_verify: Callable[[dict], dict]


def canonical_sha256(value: object) -> str:
    return "sha256:" + hashlib.sha256(
        json.dumps(
            value, sort_keys=True, separators=(",", ":"), allow_nan=False
        ).encode("utf-8")
    ).hexdigest()


def run_restart_gate(
    *,
    signed_manifest: dict,
    owner_receipt_manifest: dict,
    regression_manifest: dict,
    approval_jws: str,
    receipt_verifier: ReceiptVerifier,
    halt_authority: HaltAuthority,
    safe_mode_gate: SafeModeGate,
    regression_runner: RegressionRunner,
    approval_ledger: ApprovalLedger,
    evidence: Evidence,
) -> dict:
    required = {
        "schema_version", "restart_id", "incident_id", "scope",
        "halt_decision_sha256", "halt_version",
        "restart_policy_sha256", "safe_mode_capability_sha256",
        "owner_receipt_manifest_sha256",
        "regression_population_sha256", "approval_jti",
    }
    if (
        not isinstance(signed_manifest, dict)
        or set(signed_manifest) != required
        or signed_manifest["schema_version"]
            != "aidefend.restart-manifest.v1"
    ):
        raise ValueError("verified restart manifest schema invalid")
    if (
        canonical_sha256(owner_receipt_manifest)
        != signed_manifest["owner_receipt_manifest_sha256"]
        or canonical_sha256(regression_manifest)
        != signed_manifest["regression_population_sha256"]
    ):
        raise ValueError("restart population manifest binding invalid")

    current_halt = halt_authority.current_and_verify(
        signed_manifest["scope"]
    )
    if (
        current_halt.get("state") != "HALT"
        or current_halt.get("decision_sha256")
            != signed_manifest["halt_decision_sha256"]
        or current_halt.get("version") != signed_manifest["halt_version"]
    ):
        raise PermissionError("restart does not bind current HALT")

    owners = receipt_verifier.verify_owner_set(
        owner_receipt_manifest,
        signed_manifest["owner_receipt_manifest_sha256"],
    )
    if (
        owners.get("outcome") != "PASS"
        or owners.get("unverified_owner_ids")
        or owners.get("residual_target_refs")
        or owners.get("verified_target_population_sha256")
            != owner_receipt_manifest["target_population_sha256"]
    ):
        raise PermissionError("owner remediation population not verified")

    safe_mode = safe_mode_gate.deploy_and_readback({
        "restart_id": signed_manifest["restart_id"],
        "scope": signed_manifest["scope"],
        "halt_decision_sha256": current_halt["decision_sha256"],
        "capability_manifest_sha256": (
            signed_manifest["safe_mode_capability_sha256"]
        ),
        "restart_policy_sha256": signed_manifest["restart_policy_sha256"],
    })
    if (
        safe_mode.get("outcome") != "PASS"
        or safe_mode.get("effective_capability_sha256")
            != signed_manifest["safe_mode_capability_sha256"]
        or not safe_mode.get("high_risk_actions_blocked")
    ):
        raise PermissionError("effective safe-mode capability readback failed")

    regression = regression_runner.run_and_verify({
        "restart_id": signed_manifest["restart_id"],
        "scope": signed_manifest["scope"],
        "regression_manifest": regression_manifest,
        "regression_population_sha256": (
            signed_manifest["regression_population_sha256"]
        ),
        "safe_mode_receipt_sha256": safe_mode["receipt_sha256"],
    })
    if (
        regression.get("outcome") != "PASS"
        or regression.get("executed_population_sha256")
            != signed_manifest["regression_population_sha256"]
        or regression.get("unexecuted_case_ids")
        or regression.get("failed_case_ids")
    ):
        raise PermissionError("regression population did not pass completely")

    approval = approval_ledger.verify_and_consume(
        approval_jws,
        {
            "action": "ISSUE_POST_HALT_RUN",
            "restart_id": signed_manifest["restart_id"],
            "incident_id": signed_manifest["incident_id"],
            "scope": signed_manifest["scope"],
            "halt_decision_sha256": current_halt["decision_sha256"],
            "halt_version": current_halt["version"],
            "owner_receipt_manifest_sha256": (
                signed_manifest["owner_receipt_manifest_sha256"]
            ),
            "regression_population_sha256": (
                signed_manifest["regression_population_sha256"]
            ),
            "safe_mode_capability_sha256": (
                signed_manifest["safe_mode_capability_sha256"]
            ),
            "jti": signed_manifest["approval_jti"],
        },
    )
    pre_run = evidence.append_and_verify({
        "schema_version": "aidefend.restart-gate-event.v1",
        "restart_id": signed_manifest["restart_id"],
        "scope": signed_manifest["scope"],
        "halt_decision_sha256": current_halt["decision_sha256"],
        "owner_receipt_set_sha256": owners["receipt_set_sha256"],
        "safe_mode_receipt_sha256": safe_mode["receipt_sha256"],
        "regression_receipt_sha256": regression["receipt_sha256"],
        "approval_receipt_sha256": approval["receipt_sha256"],
        "outcome": "INSUFFICIENT_DATA",
        "stage": "all_gates_passed_run_not_issued",
    })
    run_decision = halt_authority.publish_run_and_verify({
        "scope": signed_manifest["scope"],
        "previous_halt_version": current_halt["version"],
        "previous_halt_decision_sha256": current_halt["decision_sha256"],
        "restart_id": signed_manifest["restart_id"],
        "restart_policy_sha256": signed_manifest["restart_policy_sha256"],
        "gate_evidence_receipt_sha256": pre_run["receipt_sha256"],
        "idempotency_key": signed_manifest["restart_id"],
    })
    if (
        run_decision.get("state") != "RUN"
        or run_decision.get("version") <= current_halt["version"]
        or run_decision.get("restart_id") != signed_manifest["restart_id"]
    ):
        raise RuntimeError("new signed RUN decision invalid")
    final = evidence.append_and_verify({
        "schema_version": "aidefend.restart-gate-event.v1",
        "restart_id": signed_manifest["restart_id"],
        "scope": signed_manifest["scope"],
        "gate_evidence_receipt_sha256": pre_run["receipt_sha256"],
        "run_decision_sha256": run_decision["decision_sha256"],
        "run_version": run_decision["version"],
        "outcome": "PASS",
        "stage": "run_issued",
    })
    return {
        "outcome": "PASS",
        "run_decision_sha256": run_decision["decision_sha256"],
        "run_version": run_decision["version"],
        "evidence_receipt_sha256": final["receipt_sha256"],
    }

AID-I-005-G005

Run recursive halt-readiness drills that verify signed halt propagation and aggregate separately signed AID-E-001 credential, AID-E-002 live-execution, and AID-E-005 durable-state receipts without claiming their actions.
Drill scope and responsibilities

AID-I-005 owns issuing the signed halt decision, verifying that every work-admission boundary observes it, and aggregating independent owner receipts. It does not claim credential revocation, process termination, or durable-state deletion as kill-switch actions. Route credential/JWT/token/grant revocation to AID-E-001; active agents, subagents, processes, workers, and run leases to AID-E-002; and durable server-side session, memory, webhook, tool-registration, queued-job, and schedule records to AID-E-005.

Signed drill manifest
{
  "schema_version": "aidefend.recursive_halt_drill.v1",
  "drill_id": "shutdown-drill-change-ticket-id",
  "scope": {"type": "tenant", "tenant_ref": "hmac-reference"},
  "halt_policy_sha256": "signed-halt-policy-digest",
  "shutdown_slo_policy_sha256": "signed-slo-policy-digest",
  "targets": {
    "AID-E-001": {
      "credential_ids": ["credential-ref"],
      "jwt_jtis": ["jwt-jti-ref"],
      "token_ids": ["token-ref"],
      "grant_ids": ["grant-ref"]
    },
    "AID-E-002": {
      "agent_ids": ["agent-ref"],
      "subagent_ids": ["subagent-ref"],
      "process_ids": ["process-ref"],
      "worker_ids": ["worker-ref"],
      "run_lease_ids": ["lease-ref"]
    },
    "AID-E-005": {
      "session_record_ids": ["session-ref"],
      "memory_record_ids": ["memory-ref"],
      "webhook_registration_ids": ["webhook-ref"],
      "tool_registration_ids": ["tool-ref"],
      "queued_job_ids": ["job-ref"],
      "schedule_record_ids": ["schedule-ref"]
    }
  }
}
Halt propagation and owner-receipt aggregation
# File: drills/recursive_halt_drill.py
from __future__ import annotations

import hashlib
import json
from datetime import datetime, timezone


def canonical_sha256(value):
    return hashlib.sha256(
        json.dumps(
            value, sort_keys=True, separators=(",", ":"), allow_nan=False
        ).encode()
    ).hexdigest()


def run_drill(manifest, clients, signed_slo_policy):
    # Both inputs are signature/schema/digest verified before this function.
    drill_id = manifest["drill_id"]
    target_digest = canonical_sha256(manifest["targets"])

    # I-005 action: publish the signed halt decision and verify propagation.
    halt_receipt = clients["halt_authority"].publish_halt(
        scope=manifest["scope"],
        drill_id=drill_id,
        halt_policy_sha256=manifest["halt_policy_sha256"],
    )
    propagation_receipt = clients["halt_verifier"].verify_all_boundaries(
        scope=manifest["scope"],
        drill_id=drill_id,
        deadline_seconds=signed_slo_policy["deadline_seconds"],
    )

    # Request separate owner drills. Their services perform and attest their
    # own actions; I-005 does not call revoke/terminate/delete APIs directly.
    owner_receipts = {}
    for owner in ("AID-E-001", "AID-E-002", "AID-E-005"):
        owner_receipts[owner] = clients["owner_drills"].request_and_wait(
            owner=owner,
            drill_id=drill_id,
            scope=manifest["scope"],
            targets=manifest["targets"][owner],
            target_set_sha256=canonical_sha256(manifest["targets"][owner]),
            deadline_seconds=signed_slo_policy["deadline_seconds"],
        )

    # Verify each compact signed receipt with the registered owner key and
    # bind it to this drill, scope, exact target set, and action owner.
    verified = {}
    for owner, receipt_jws in owner_receipts.items():
        verified[owner] = clients["receipt_verifier"].verify(
            receipt_jws,
            expected_owner=owner,
            expected_drill_id=drill_id,
            expected_scope=manifest["scope"],
            expected_target_set_sha256=canonical_sha256(
                manifest["targets"][owner]
            ),
        )

    receipts = {
        "halt": halt_receipt,
        "propagation": propagation_receipt,
        "owners": verified,
    }
    residual = {
        owner: receipt["residual_target_refs"]
        for owner, receipt in verified.items()
        if receipt["residual_target_refs"]
    }
    status = (
        "PASS"
        if propagation_receipt["status"] == "VERIFIED"
        and all(receipt["status"] == "VERIFIED" for receipt in verified.values())
        and not residual
        else "FAIL"
    )
    return {
        "schema_version": "aidefend.recursive_halt_drill_result.v1",
        "drill_id": drill_id,
        "status": status,
        "target_manifest_sha256": target_digest,
        "halt_policy_sha256": manifest["halt_policy_sha256"],
        "shutdown_slo_policy_sha256": manifest[
            "shutdown_slo_policy_sha256"
        ],
        "completed_at": datetime.now(timezone.utc).isoformat(),
        "residual_targets_by_owner": residual,
        "receipt_set_sha256": canonical_sha256(receipts),
        "receipts": receipts,
    }


def safe_run(*args, **kwargs):
    try:
        return run_drill(*args, **kwargs)
    except Exception as exc:
        return {
            "status": "ERROR",
            "reason": "manifest, halt propagation, or owner receipt invalid",
            "error_class": type(exc).__name__,
        }
Evidence and pass criteria

Retain the signed manifest and SLO policy, exact target-set digests, signed halt/propagation receipt, each owner-specific signed receipt, residual targets, aggregate digest, and exception class. PASS means halt propagation was verified and every owner separately attested complete coverage with no residual targets. Missing, invalid, misbound, or late receipts are ERROR or FAIL according to the signed drill policy; they are never counted as I-005 action evidence.

AID-I-005-G006

Enforce signed, versioned global and tenant RUN/HALT decisions at every inference and agent-loop boundary using verified identities, rollback-resistant version floors, and fail-closed reads.
Keep the action safely bounded

Every work path must receive its tenant from verified OIDC or workload-identity middleware, never from X-Tenant-ID, request JSON, model output, or a constructor argument copied from untrusted input. Missing, expired, corrupt, stale, rolled-back, or unreadable decisions and decision-store/version-floor outages are ERROR and deny work.

Signed, rollback-resistant halt gate
# File: runtime/halt_gate.py
# Dependencies: redis>=5.1,<6, PyJWT>=2.9,<3,
# cryptography>=43,<45, boto3>=1.35,<2
from __future__ import annotations

import hashlib
import json
import math
import os
from dataclasses import dataclass

import boto3
import jwt
import redis
import re
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

# Admission projects these mandatory values from one signature-verified,
# versioned runtime profile; absence or invalidity aborts startup.
RUNTIME_PROFILE_VERSION = os.environ["AIDEFEND_RUNTIME_PROFILE_VERSION"]
RUNTIME_PROFILE_SHA256 = os.environ["AIDEFEND_RUNTIME_PROFILE_SHA256"]
if (
    not RUNTIME_PROFILE_VERSION
    or not re.fullmatch(r"[0-9a-f]{64}", RUNTIME_PROFILE_SHA256)
):
    raise RuntimeError("verified runtime-profile identity is invalid")


def finite_profile_number(name: str, *, allow_zero: bool = False) -> float:
    value = float(os.environ[name])
    if (
        not math.isfinite(value)
        or value < 0
        or (not allow_zero and value == 0)
    ):
        raise RuntimeError("verified runtime-profile numeric value is invalid: " + name)
    return value


REDIS_CONNECT_TIMEOUT_SECONDS = finite_profile_number(
    "HALT_REDIS_CONNECT_TIMEOUT_SECONDS"
)
REDIS_SOCKET_TIMEOUT_SECONDS = finite_profile_number(
    "HALT_REDIS_SOCKET_TIMEOUT_SECONDS"
)
DECISION_CLOCK_SKEW_SECONDS = finite_profile_number(
    "HALT_DECISION_CLOCK_SKEW_SECONDS", allow_zero=True
)

DECISION_KEY = open(os.environ["HALT_DECISION_PUBLIC_KEY_PATH"], "rb").read()
DECISION_ISSUER = os.environ["HALT_DECISION_ISSUER"]
VERSION_TABLE = os.environ["HALT_VERSION_FLOOR_TABLE"]
r = redis.Redis.from_url(
    os.environ["HALT_REDIS_URL"],
    ssl=True,
    ssl_cert_reqs="required",
    socket_connect_timeout=REDIS_CONNECT_TIMEOUT_SECONDS,
    socket_timeout=REDIS_SOCKET_TIMEOUT_SECONDS,
    decode_responses=True,
)
ddb = boto3.client("dynamodb")
app = FastAPI()


@dataclass(frozen=True)
class VerifiedIdentity:
    issuer: str
    tenant_id: str
    subject: str
    workload_id: str | None = None


class HaltGateError(Exception):
    pass


def load_decision(scope: str) -> dict:
    # An explicit signed RUN record is required; key absence is not permission.
    token = r.get(f"halt:decision:{scope}")
    if token is None:
        raise HaltGateError("missing halt decision")
    token_sha256 = hashlib.sha256(token.encode()).hexdigest()
    header = jwt.get_unverified_header(token)
    if header.get("alg") != "EdDSA" or header.get("typ") != "halt-decision+jwt":
        raise HaltGateError("unexpected halt-decision envelope")
    claims = jwt.decode(
        token,
        DECISION_KEY,
        algorithms=["EdDSA"],
        issuer=DECISION_ISSUER,
        audience="ai-runtime-halt-gate",
        leeway=DECISION_CLOCK_SKEW_SECONDS,
        options={"require": [
            "iss", "aud", "iat", "nbf", "exp", "jti", "scope",
            "state", "version", "policy_sha256", "incident_id",
        ]},
    )
    if claims["scope"] != scope or claims["state"] not in {"RUN", "HALT"}:
        raise HaltGateError("halt decision is misbound")
    version = claims["version"]
    if isinstance(version, bool) or not isinstance(version, int) or version < 1:
        raise HaltGateError("invalid halt-decision version")

    floor_item = ddb.get_item(
        TableName=VERSION_TABLE,
        Key={"scope": {"S": scope}},
        ConsistentRead=True,
    ).get("Item")
    if floor_item is None:
        raise HaltGateError("missing authoritative version floor")
    required_floor_fields = {"scope", "version", "token_sha256", "state", "policy_sha256"}
    if not required_floor_fields.issubset(floor_item):
        raise HaltGateError("authoritative floor binding is incomplete")
    floor = int(floor_item["version"]["N"])
    if (
        version != floor
        or floor_item["token_sha256"]["S"] != token_sha256
        or floor_item["state"]["S"] != claims["state"]
        or floor_item["policy_sha256"]["S"] != claims["policy_sha256"]
    ):
        raise HaltGateError(
            "decision differs from digest/state/policy-bound authoritative floor"
        )
    return {
        "scope": scope,
        "state": claims["state"],
        "version": version,
        "decision_jti": claims["jti"],
        "policy_sha256": claims["policy_sha256"],
        "incident_id": claims["incident_id"],
        "token_sha256": token_sha256,
    }


def evaluate(identity: VerifiedIdentity) -> dict:
    if not identity.issuer or not identity.subject or not identity.tenant_id:
        raise HaltGateError("unverified or incomplete identity")
    decisions = [
        load_decision("global"),
        load_decision(f"tenant:{identity.tenant_id}"),
    ]
    halted = next((d for d in decisions if d["state"] == "HALT"), None)
    return {
        "runtime_profile_version": RUNTIME_PROFILE_VERSION,
        "runtime_profile_sha256": RUNTIME_PROFILE_SHA256,
        "decision": "HALT_ENFORCED" if halted else "RUN_ALLOWED",
        "allow": halted is None,
        "matched_scope": halted["scope"] if halted else None,
        "decisions": decisions,
        "identity": {
            "issuer": identity.issuer,
            "tenant_id": identity.tenant_id,
            "subject": identity.subject,
            "workload_id": identity.workload_id,
        },
    }


@app.middleware("http")
async def halt_gate(request: Request, call_next):
    try:
        # Set only by upstream authentication middleware after token and
        # tenant-membership verification. Headers and body are ignored here.
        identity = request.state.verified_identity
        if not isinstance(identity, VerifiedIdentity):
            raise HaltGateError("verified identity middleware did not run")
        evidence = evaluate(identity)
    except Exception as exc:
        print(json.dumps({
            "event": "HALT_GATE",
            "decision": "ERROR",
            "allow": False,
            "runtime_profile_version": RUNTIME_PROFILE_VERSION,
            "runtime_profile_sha256": RUNTIME_PROFILE_SHA256,
            "error_class": type(exc).__name__,
        }))
        return JSONResponse(
            status_code=503,
            content={"error": "HALT_STATE_UNVERIFIABLE"},
            headers={"X-System-Halt": "error"},
        )

    print(json.dumps({"event": "HALT_GATE", **evidence}))
    if not evidence["allow"]:
        return JSONResponse(
            status_code=503,
            content={"error": "EMERGENCY_HALT_ACTIVE"},
            headers={"X-System-Halt": "true"},
        )
    return await call_next(request)


class HaltAwareAgent:
    def __init__(self, verified_workload_identity: VerifiedIdentity):
        # Construct this object from verified SPIFFE/mTLS middleware mapping,
        # not from model output, a job payload, or an arbitrary tenant string.
        if not verified_workload_identity.workload_id:
            raise HaltGateError("workload identity is required")
        self.identity = verified_workload_identity

    def run_one_step(self, step):
        evidence = evaluate(self.identity)
        print(json.dumps({"event": "AGENT_HALT_GATE", **evidence}))
        if not evidence["allow"]:
            raise HaltGateError("emergency halt active")
        return step()
Issuance, rollout, and evidence

The halt authority must use a fail-closed two-phase rollout across stores: first publish the new signed decision, then create/advance the strongly consistent floor that binds its exact token digest, state, policy digest, and version. Until both reads agree exactly, the gate deliberately denies work; do not describe Redis and the version store as one atomic transaction. Populate explicit signed RUN records before enabling the gate. Refresh them before expiry and use deployment-specific SLOs from signed policy rather than embedding a universal duration. Send each gate result, decision-token digest, version, policy digest, scope, and authenticated identity reference to append-only evidence storage.

Independent checks

Test global and tenant halt, tenant-header/body spoofing, wrong issuer/audience/scope, bad signature, expiry, missing record, missing version floor, old-token replay, future/uncommitted version, Redis and version-store outages, and both HTTP and non-HTTP paths. Replay the compact JWS and strong-read version floor independently. Only RUN_ALLOWED permits work; HALT_ENFORCED and ERROR deny it.

Open-source tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-I-005: Emergency "Kill-Switch" / AI System Halt | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service