AID-D-001.006: Prompt Gate Decision Telemetry Correlation

Version 1.20261015 · Actionable sub-technique

Consumes verdict telemetry emitted by synchronous prompt gates and correlates it across requests, sessions, and identities to detect evasive probing, repeated near-miss attempts, and cumulative policy risk. This sub-technique is detect-side only: it does not block traffic and should not be confused with the inline hardening gate itself.

Pillars: app

Lifecycle: operation

Scope boundary

Owns cross-request, cross-session, and cross-identity correlation of synchronous prompt-gate verdict telemetry to detect evasive probing, repeated near misses, and cumulative policy risk. It does not block traffic or replace the inline gate.

AID-D-001.001: AID-D-001.006 correlates the history of prompt-analysis and gate outcomes; AID-D-001.001 analyzes the content, intent, and obfuscation of an individual prompt. Correlation detects campaign behavior that no single request establishes.

AID-D-001.004: AID-D-001.006 detects repeated or cumulative patterns across gate events; AID-D-001.004 emits a request-scoped canary integrity finding. A canary anomaly is one input to correlation, not the correlation control itself.

Implementation guidance

AID-D-001.006-G001

Ingest and analyze synchronous gate decisions to detect evasions, correlate incidents, and escalate.
Verify the signed policy and authenticated gate event
from __future__ import annotations

import math as _aidefend_math
import os as _aidefend_os

# Admission injects these required values from one signature-verified, versioned runtime profile.
def _aidefend_required(name):
    value = _aidefend_os.environ.get(name)
    if value is None or not value.strip():
        raise RuntimeError(f"required runtime-profile field is absent: {name}")
    return value.strip()


def _aidefend_positive_float(name):
    try:
        value = float(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not numeric: {name}") from error
    if not _aidefend_math.isfinite(value) or value <= 0:
        raise RuntimeError(f"runtime-profile field must be finite and positive: {name}")
    return value


def _aidefend_positive_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 1:
        raise RuntimeError(f"runtime-profile field must be positive: {name}")
    return value


def _aidefend_nonnegative_int(name):
    try:
        value = int(_aidefend_required(name))
    except ValueError as error:
        raise RuntimeError(f"runtime-profile field is not an integer: {name}") from error
    if value < 0:
        raise RuntimeError(f"runtime-profile field cannot be negative: {name}")
    return value


def _aidefend_fraction(name):
    value = _aidefend_positive_float(name)
    if value > 1:
        raise RuntimeError(f"runtime-profile field must be in (0, 1]: {name}")
    return value


RUNTIME_PROFILE_VERSION = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_VERSION")
RUNTIME_PROFILE_SHA256 = _aidefend_required("AIDEFEND_RUNTIME_PROFILE_SHA256").lower()
if (len(RUNTIME_PROFILE_SHA256) != 64
        or set(RUNTIME_PROFILE_SHA256) - set("0123456789abcdef")):
    raise RuntimeError("runtime-profile digest must be lowercase SHA-256")
DECISION_ANALYSIS_COMMAND_TIMEOUT_SECONDS = _aidefend_positive_float("D001006_DECISION_ANALYSIS_COMMAND_TIMEOUT_SECONDS")

# File: detect/gate_decision_consumer.py

import hashlib
import json
import math
import os
import subprocess
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
from uuid import UUID

import jwt
import redis


POLICY_PATH = Path(os.environ["D001006_POLICY_PATH"])
POLICY_SIGNATURE = Path(os.environ["D001006_POLICY_SIGNATURE"])
POLICY_VERIFY_KEY = os.environ["D001006_POLICY_VERIFY_KEY"]
EVENT_VERIFY_KEY = Path(os.environ["D001006_EVENT_VERIFY_KEY"]).read_text(encoding="utf-8")
EVENT_ISSUER = os.environ["D001006_EVENT_ISSUER"]
EVENT_AUDIENCE = "aidefend:d001006"
REDIS = redis.Redis.from_url(os.environ["D001006_REDIS_URL"], decode_responses=True)
POLICY_FIELDS = {
    "schema_version", "policy_version", "applicable_routes",
    "applicable_model_prefixes", "window_seconds", "minimum_events",
    "max_event_age_seconds", "max_future_skew_seconds", "replay_ttl_seconds",
    "deny_burst_threshold", "soft_flag_variant_threshold",
    "cumulative_risk_threshold", "policy_code_weights",
}
EVENT_FIELDS = {
    "schema_version", "event_id", "request_id", "session_id", "user_id",
    "observed_at", "verdict", "policy_code", "normalized_hash", "route",
    "model", "signals", "iss", "aud", "iat", "nbf", "exp", "jti",
}


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


def canonical(value: dict) -> str:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False)


def verify_blob(path: Path, signature: Path, key: str) -> None:
    subprocess.run([
        "cosign", "verify-blob", "--key", key,
        "--bundle", str(signature), str(path),
    ], check=True, capture_output=True, text=True, timeout=DECISION_ANALYSIS_COMMAND_TIMEOUT_SECONDS)


def positive_int(value: object, field: str) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or value < 1:
        raise ValueError(f"{field} must be a positive integer")
    return value


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


def load_policy() -> tuple[dict, str]:
    # Authenticate and parse the same immutable byte snapshot.
    raw = POLICY_PATH.read_bytes()
    signature_raw = POLICY_SIGNATURE.read_bytes()
    with tempfile.TemporaryDirectory(prefix="aidefend-d001006-") as directory:
        snapshot = Path(directory) / "policy.json"
        signature = Path(directory) / "policy.sig"
        snapshot.write_bytes(raw)
        signature.write_bytes(signature_raw)
        verify_blob(snapshot, signature, POLICY_VERIFY_KEY)
    policy = json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=reject_duplicate_keys, parse_constant=lambda value: (_ for _ in ()).throw(ValueError('non-finite JSON constant: ' + value)),
    )
    if not isinstance(policy, dict) or set(policy) != POLICY_FIELDS:
        raise ValueError("correlation policy schema mismatch")
    if policy["schema_version"] != "aidefend.d001006-policy.v1":
        raise ValueError("unsupported correlation policy")
    for field in (
        "window_seconds", "minimum_events", "max_event_age_seconds",
        "max_future_skew_seconds", "replay_ttl_seconds",
        "deny_burst_threshold", "soft_flag_variant_threshold",
    ):
        positive_int(policy[field], field)
    threshold = policy["cumulative_risk_threshold"]
    if isinstance(threshold, bool) or not isinstance(threshold, (int, float)) or not math.isfinite(float(threshold)) or threshold <= 0:
        raise ValueError("cumulative risk threshold invalid")
    for field in ("applicable_routes", "applicable_model_prefixes"):
        values = policy[field]
        if not isinstance(values, list) or not values or len(values) != len(set(values)) or any(not isinstance(item, str) or not item for item in values):
            raise ValueError(f"{field} invalid")
    weights = policy["policy_code_weights"]
    if not isinstance(weights, dict) or not weights:
        raise ValueError("policy-code weights missing")
    for code, weight in weights.items():
        if not isinstance(code, str) or not code or isinstance(weight, bool) or not isinstance(weight, (int, float)) or not math.isfinite(float(weight)) or weight < 0:
            raise ValueError("policy-code weight invalid")
    return policy, sha256_bytes(raw)


def parse_time(value: object) -> datetime:
    if not isinstance(value, str):
        raise ValueError("observed_at missing")
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        raise ValueError("observed_at lacks timezone")
    return parsed.astimezone(timezone.utc)


def decode_event(compact_jws: bytes) -> dict[str, Any]:
    token = compact_jws.decode("ascii", errors="strict")
    header = jwt.get_unverified_header(token)
    if header.get("alg") != "EdDSA" or header.get("typ") != "gate-event+jwt":
        raise ValueError("unexpected gate-event envelope")
    event = jwt.decode(
        token, EVENT_VERIFY_KEY, algorithms=["EdDSA"],
        issuer=EVENT_ISSUER, audience=EVENT_AUDIENCE,
        options={"require": ["iss", "aud", "iat", "nbf", "exp", "jti"]},
    )
    if not isinstance(event, dict) or set(event) != EVENT_FIELDS:
        raise ValueError("gate-event schema mismatch")
    if event["schema_version"] != "aidefend.gate-event.v1":
        raise ValueError("unsupported gate-event schema")
    for field in ("event_id", "request_id", "session_id", "normalized_hash", "route", "model"):
        if not isinstance(event[field], str) or not event[field]:
            raise ValueError(f"{field} missing")
    UUID(event["event_id"])
    UUID(event["request_id"])
    if event["verdict"] not in {"allow", "deny", "soft_flag"}:
        raise ValueError("invalid gate verdict")
    if event["policy_code"] is not None and (not isinstance(event["policy_code"], str) or not event["policy_code"]):
        raise ValueError("invalid policy code")
    if not isinstance(event["signals"], dict):
        raise ValueError("signals must be an object")
    if len(event["normalized_hash"]) != 64:
        raise ValueError("normalized_hash must be SHA-256")
    int(event["normalized_hash"], 16)
    return event
Correlate idempotently across replicas
# File: detect/gate_decision_runner.py
import json
from datetime import datetime, timezone

from detect.gate_decision_consumer import (
    REDIS,
    canonical,
    decode_event,
    load_policy,
    parse_time,
    sha256_bytes,
)

def outcome(
    control_status: str,
    finding_status: str,
    reason: str,
    **evidence: object,
) -> dict:
    return {
        "schema_version": "aidefend.d001006-result.v1",
        "control": "AID-D-001.006",
        "control_status": control_status,
        "finding_status": finding_status,
        "reason": reason,
        **evidence,
    }


def correlate(compact_jws: bytes, *, now: datetime | None = None) -> dict:
    current = (now or datetime.now(timezone.utc)).astimezone(timezone.utc)
    policy, policy_sha256 = load_policy()
    event = decode_event(compact_jws)
    observed = parse_time(event["observed_at"])
    age = (current - observed).total_seconds()
    if age > policy["max_event_age_seconds"] or age < -policy["max_future_skew_seconds"]:
        raise ValueError("gate event outside signed freshness bounds")
    if event["route"] not in policy["applicable_routes"] or not any(event["model"].startswith(prefix) for prefix in policy["applicable_model_prefixes"]):
        return outcome("NOT_APPLICABLE", "NOT_EVALUATED", "route or model outside signed applicability", event_id=event["event_id"], policy_sha256=policy_sha256)

    event_digest = sha256_bytes(compact_jws)
    replay_key = f"d001006:event:{event['event_id']}"
    claimed = REDIS.set(replay_key, event_digest, nx=True, ex=policy["replay_ttl_seconds"])
    if not claimed:
        if REDIS.get(replay_key) != event_digest:
            raise ValueError("event_id replayed with different authenticated bytes")
        cached = REDIS.get(f"d001006:result:{event['event_id']}")
        if cached is None:
            raise RuntimeError("idempotent event is still being evaluated")
        return json.loads(cached)

    window_key = f"d001006:session:{event['session_id']}"
    member = canonical({
        "event_id": event["event_id"], "verdict": event["verdict"],
        "policy_code": event["policy_code"], "normalized_hash": event["normalized_hash"],
    })
    score = observed.timestamp()
    with REDIS.pipeline(transaction=True) as pipe:
        pipe.zadd(window_key, {member: score})
        pipe.zremrangebyscore(window_key, "-inf", score - policy["window_seconds"])
        pipe.expire(window_key, policy["replay_ttl_seconds"])
        pipe.zrange(window_key, 0, -1)
        window = [json.loads(item) for item in pipe.execute()[-1]]

    findings: list[str] = []
    deny_count = sum(item["verdict"] == "deny" for item in window)
    soft_hashes = {item["normalized_hash"] for item in window if item["verdict"] == "soft_flag"}
    cumulative = 0.0
    unmapped_codes = set()
    for item in window:
        if item["verdict"] not in {"deny", "soft_flag"} or item["policy_code"] is None:
            continue
        if item["policy_code"] not in policy["policy_code_weights"]:
            unmapped_codes.add(item["policy_code"])
            continue
        cumulative += float(policy["policy_code_weights"][item["policy_code"]])
    if unmapped_codes:
        result = outcome("ERROR", "UNKNOWN", "window contains unmapped policy codes", event_id=event["event_id"], unmapped_policy_codes=sorted(unmapped_codes), policy_sha256=policy_sha256)
    elif len(window) < policy["minimum_events"]:
        result = outcome("INSUFFICIENT_DATA", "UNKNOWN", "valid window below signed minimum", event_id=event["event_id"], observation_count=len(window), policy_sha256=policy_sha256)
    else:
        if deny_count >= policy["deny_burst_threshold"]:
            findings.append("deny_burst")
        if len(soft_hashes) >= policy["soft_flag_variant_threshold"]:
            findings.append("soft_flag_variant_evasion")
        if cumulative >= float(policy["cumulative_risk_threshold"]):
            findings.append("cumulative_policy_risk")
        result = outcome("PASS", "FINDING" if findings else "NO_FINDING", "correlation finding" if findings else "no signed threshold breached", event_id=event["event_id"], session_id=event["session_id"], observation_count=len(window), findings=findings, policy_sha256=policy_sha256)

    REDIS.set(f"d001006:result:{event['event_id']}", canonical(result), ex=policy["replay_ttl_seconds"])
    return result


def safe_correlate(compact_jws: bytes) -> dict:
    try:
        return correlate(compact_jws)
    except Exception as exc:
        return outcome("ERROR", "UNKNOWN", "gate event, policy, replay, storage, or evaluation invalid", error_class=type(exc).__name__)
Verify safely

Retain the compact JWS digest, signed policy/signature, policy digest, exact event and session identifiers, window population, Redis replay result, findings, and evaluator version. An independent identity re-verifies the event and policy signatures, replays the same bounded window from immutable events, and compares the result digest.

Verify complete coverage

For this detector, evaluate every applicable gate-decision event between signed source checkpoints for each route, model, session, and correlation window, including allow, deny, soft-flag, duplicate, late, and processing-error events. Require a source-authenticated completeness manifest binding the asset and version, request or window, start and end checkpoints, expected IDs, evaluated IDs, exclusions, and canonical digests.

Open-source tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-D-001.006: Prompt Gate Decision Telemetry Correlation | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service