AID-D-002.004: Reasoning Trace & Thinking-Token Anomaly Telemetry

Version 1.20261015 · Actionable sub-technique

Collect and analyze output-side reasoning metadata for systems that expose extended-thinking summaries, provider thinking blocks, redaction flags, or thinking-token accounting. Monitor token counts, hidden-to-visible ratios, omission or redaction state, summary length, display mode, request intent, latency, and budget decisions to detect forced over-reasoning, abnormal trace disclosure, and reasoning-specific availability abuse. The meaning of an emitted reasoning trace is assessed separately by AID-D-002.005.

Pillars: app, model, infra

Lifecycle: operation

Scope boundary

Owns anomalies in provider or API-visible reasoning metadata and accounting, including thinking-token counts, reasoning-to-visible-output ratios, redaction flags, trace-summary shape, display mode, latency, and reasoning-budget decisions. It does not semantically judge the content of an emitted reasoning trace.

AID-D-002.003: AID-D-002.004 monitors output-side reasoning metadata and accounting; AID-D-002.003 monitors white-box attention, logits, and activation-like internal signals. Reasoning blocks exposed by an API are not equivalent to direct instrumentation of model internals.

AID-D-005.007: AID-D-002.004 identifies reasoning-specific trace and thinking-token anomalies; AID-D-005.007 detects broader request-volume, token, tool-use, parameter, latency, and cost spikes. Reasoning telemetry feeds the broader consumption detector but retains its own semantics and applicability.

Implementation guidance

AID-D-002.004-G001

Collect reasoning-token and trace-shape telemetry at the gateway and alert on reasoning-specific anomalies such as budget overruns, high hidden-output ratios, repeated redaction, or unexpected trace disclosure.
Observation boundary and responsibilities

Run this evaluator at the inference gateway over the complete immutable response-event population for a signed measurement window. Select JSON paths by exact provider, model, and API version from a signed capability profile; never infer that an absent value is zero or false. This detector emits reasoning-specific findings only. Blocking and budget enforcement remain with AID-H-037.002, reasoning confidentiality with AID-H-037.001, and general cost/latency detection with AID-D-005.007. Latency is retained below only as a correlation join field and is never scored here.

Signed inputs

Sign reasoning-capabilities.json, reasoning-detection-policy.json, the sanitized JSONL export, and a separately produced completeness receipt. Each capability entry has exact provider, model, and api_version keys, an applicable Boolean, JSON Pointer paths, and required fields. The receipt binds the source checkpoint, window, digest, record count, and eligible count. The policy owns every numeric threshold.

Verify, normalize, evaluate, and sign
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")
REASONING_TELEMETRY_COMMAND_TIMEOUT_SECONDS = _aidefend_positive_float("D002004_REASONING_TELEMETRY_COMMAND_TIMEOUT_SECONDS")

# file: detection/evaluate_reasoning_telemetry.py

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

RECORDS = Path("artifacts/reasoning/provider-responses.jsonl")
RECEIPT = Path("artifacts/reasoning/collection-receipt.json")
CAPS = Path("policy/reasoning-capabilities.json")
POLICY = Path("policy/reasoning-detection-policy.json")
EVIDENCE = Path("artifacts/reasoning/AID-D-002.004-evidence.json")


def sha256_file(path: Path) -> str:
    captured = _VERIFIED_INPUTS.get(Path(path))
    if captured is not None:
        return hashlib.sha256(captured).hexdigest()
    digest = hashlib.sha256()
    with Path(path).open("rb") as handle:
        for chunk in iter(lambda: handle.read(1048576), b""):
            digest.update(chunk)
    return digest.hexdigest()


_VERIFIED_INPUTS: dict[Path, bytes] = {}


def _write_private_snapshot(path: Path, payload: bytes) -> None:
    descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(descriptor, "wb") as stream:
        stream.write(payload)
        stream.flush()
        os.fsync(stream.fileno())
    os.chmod(path, 0o400)


def _strict_json(raw: bytes | str) -> object:
    text = raw.decode("utf-8", errors="strict") if isinstance(raw, bytes) else raw

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

    return json.loads(
        text,
        object_pairs_hook=unique_object,
        parse_constant=lambda value: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + value)
        ),
    )


def verify(path: Path, key_env: str) -> None:
    payload_path = Path(path)
    signature_path = Path(str(payload_path) + ".sig")
    payload_raw = payload_path.read_bytes()
    signature_raw = signature_path.read_bytes()
    key_reference = os.environ[key_env]
    remote_key = bool(re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*://", key_reference
    )) or key_reference.startswith("pkcs11:")
    key_raw = None if remote_key else (
        key_reference.encode("utf-8")
        if key_reference.startswith("-----BEGIN ")
        else Path(key_reference).read_bytes()
    )
    with tempfile.TemporaryDirectory(prefix="aidefend-d002004-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        signature_snapshot = root / "signature"
        _write_private_snapshot(payload_snapshot, payload_raw)
        _write_private_snapshot(signature_snapshot, signature_raw)
        captured = [(payload_snapshot, payload_raw), (signature_snapshot, signature_raw)]
        key_argument = key_reference
        if key_raw is not None:
            key_snapshot = root / "verify-key.pub"
            _write_private_snapshot(key_snapshot, key_raw)
            captured.append((key_snapshot, key_raw))
            key_argument = str(key_snapshot)
        try:
            subprocess.run(
                [
                    "cosign", "verify-blob", "--key", key_argument,
                    "--bundle", str(signature_snapshot), str(payload_snapshot),
                ],
                check=True, capture_output=True, text=True,
                timeout=REASONING_TELEMETRY_COMMAND_TIMEOUT_SECONDS,
            )
            if any(
                not hmac.compare_digest(candidate.read_bytes(), expected)
                for candidate, expected in captured
            ):
                raise RuntimeError("private signature-verification snapshot changed")
        finally:
            for candidate, _ in captured:
                os.chmod(candidate, 0o600)
    _VERIFIED_INPUTS[payload_path] = payload_raw


def _verified_bytes(path: Path) -> bytes:
    try:
        return _VERIFIED_INPUTS[Path(path)]
    except KeyError as error:
        raise RuntimeError("signed input was not captured and verified") from error


def _verified_text(path: Path) -> str:
    return _verified_bytes(path).decode("utf-8", errors="strict")


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


def pointer(document: Any, path: str) -> tuple[bool, Any]:
    if not isinstance(path, str) or not path.startswith("/"):
        raise ValueError("invalid JSON Pointer")
    current = document
    for raw in path[1:].split("/"):
        token = raw.replace("~1", "/").replace("~0", "~")
        if isinstance(current, dict) and token in current:
            current = current[token]
        elif isinstance(current, list) and token.isdigit() and int(token) < len(current):
            current = current[int(token)]
        else:
            return False, None
    return True, current


def number(value: Any, field: str) -> float:
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError(f"{field} is not numeric")
    result = float(value)
    if not math.isfinite(result) or result < 0:
        raise ValueError(f"{field} is outside its domain")
    return result


def load() -> tuple[list[dict], dict, dict, dict]:
    verify(RECORDS, "D002004_COLLECTOR_VERIFY_KEY")
    verify(RECEIPT, "D002004_COMPLETENESS_VERIFY_KEY")
    verify(CAPS, "D002004_CAPABILITY_VERIFY_KEY")
    verify(POLICY, "D002004_POLICY_VERIFY_KEY")
    receipt = _strict_json(_verified_bytes(RECEIPT))
    caps = _strict_json(_verified_bytes(CAPS))
    policy = _strict_json(_verified_bytes(POLICY))
    receipt_fields = {"schema_version", "window_start", "window_end", "source_id",
                      "source_checkpoint", "record_count", "eligible_record_count",
                      "records_sha256"}
    if (not isinstance(receipt, dict) or set(receipt) != receipt_fields
            or receipt["schema_version"] != "aidefend.d002004-receipt.v1"
            or receipt["records_sha256"] != sha256_file(RECORDS)
            or isinstance(receipt["record_count"], bool)
            or not isinstance(receipt["record_count"], int)
            or isinstance(receipt["eligible_record_count"], bool)
            or not isinstance(receipt["eligible_record_count"], int)
            or min(receipt["record_count"], receipt["eligible_record_count"]) < 0):
        raise ValueError("collection receipt is invalid or misbound")
    policy_fields = {"schema_version", "policy_version", "max_hidden_visible_ratio",
                     "min_hidden_tokens_without_output", "max_summary_chars",
                     "max_redacted_blocks", "evidence_max_age_seconds"}
    if not isinstance(policy, dict) or set(policy) != policy_fields or policy["schema_version"] != "aidefend.d002004-policy.v1":
        raise ValueError("policy schema differs")
    for field in policy_fields - {"schema_version", "policy_version"}:
        number(policy[field], field)
    if (not isinstance(caps, dict)
            or set(caps) != {"schema_version", "profile_version", "profiles"}
            or caps["schema_version"] != "aidefend.reasoning-capabilities.v1"
            or not isinstance(caps["profiles"], list)):
        raise ValueError("capability profile schema differs")
    lines = _verified_text(RECORDS).splitlines()
    if any(not line.strip() for line in lines):
        raise ValueError("JSONL contains a blank record")
    records = [_strict_json(line) for line in lines]
    if len(records) != receipt["record_count"]:
        raise ValueError("record population differs")
    ids = [record.get("record_id") for record in records if isinstance(record, dict)]
    if len(ids) != len(records) or any(not isinstance(item, str) or not item for item in ids) or len(set(ids)) != len(ids):
        raise ValueError("record IDs are absent or duplicated")
    start, end = timestamp(receipt["window_start"]), timestamp(receipt["window_end"])
    if start >= end:
        raise ValueError("invalid measurement window")
    fields = {"record_id", "provider", "model", "api_version", "request_id",
              "observed_at", "latency_ms", "response"}
    for record in records:
        if set(record) != fields or not start <= timestamp(record["observed_at"]) < end:
            raise ValueError("record schema or window binding differs")
        number(record["latency_ms"], "latency_ms")
    return records, receipt, caps, policy


def select_profile(record: dict, profiles: list[dict]) -> dict | None:
    matches = [item for item in profiles if isinstance(item, dict)
               and item.get("provider") == record["provider"]
               and item.get("model") == record["model"]
               and item.get("api_version") == record["api_version"]]
    if len(matches) > 1:
        raise ValueError("ambiguous capability profile")
    return matches[0] if matches else None


def evaluate(record: dict, profiles: list[dict], policy: dict) -> dict:
    profile = select_profile(record, profiles)
    base = {"record_id": record["record_id"], "request_id": record["request_id"],
            "provider": record["provider"], "model": record["model"],
            "api_version": record["api_version"], "observed_at": record["observed_at"],
            "latency_ms_join_only": record["latency_ms"]}
    if profile is None or profile.get("applicable") is False:
        return {**base, "control_status": "NOT_APPLICABLE",
                "finding_status": "NOT_EVALUATED", "presence": {},
                "reason": "no signed applicable capability profile"}
    expected = {"provider", "model", "api_version", "applicable", "paths", "required_fields"}
    if set(profile) != expected or profile["applicable"] is not True:
        raise ValueError("applicable capability profile is malformed")
    paths, required = profile["paths"], profile["required_fields"]
    allowed = {"thinking_tokens", "visible_output_tokens", "budget_tokens", "display_mode",
               "redacted_block_count", "omitted_reasoning", "trace_summary"}
    if (not isinstance(paths, dict) or not set(paths).issubset(allowed)
            or not isinstance(required, list) or not set(required).issubset(set(paths))):
        raise ValueError("capability paths are invalid")
    extracted = {name: pointer(record["response"], path) for name, path in paths.items()}
    presence = {name: present for name, (present, _value) in extracted.items()}
    missing = sorted(name for name in required if not presence[name])
    if missing:
        return {**base, "control_status": "INSUFFICIENT_DATA",
                "finding_status": "UNKNOWN", "presence": presence,
                "missing_fields": missing, "reason": "advertised fields are absent"}
    try:
        values = {name: value for name, (present, value) in extracted.items() if present}
        for field in ("thinking_tokens", "visible_output_tokens", "budget_tokens", "redacted_block_count"):
            if field in values:
                values[field] = number(values[field], field)
        if "omitted_reasoning" in values and not isinstance(values["omitted_reasoning"], bool):
            raise ValueError("omitted_reasoning is not Boolean")
        if "display_mode" in values and not isinstance(values["display_mode"], str):
            raise ValueError("display_mode is not a string")
        if "trace_summary" in values and not isinstance(values["trace_summary"], str):
            raise ValueError("trace_summary is not a string")
    except ValueError as error:
        return {**base, "control_status": "ERROR",
                "finding_status": "UNKNOWN", "presence": presence,
                "reason": str(error)}
    findings: list[str] = []
    hidden, visible, budget = values.get("thinking_tokens"), values.get("visible_output_tokens"), values.get("budget_tokens")
    if hidden is not None and budget is not None and hidden > budget:
        findings.append("thinking_budget_overrun")
    if hidden is not None and visible is not None:
        if visible == 0 and hidden >= policy["min_hidden_tokens_without_output"]:
            findings.append("hidden_reasoning_without_visible_output")
        elif visible > 0 and hidden / visible > policy["max_hidden_visible_ratio"]:
            findings.append("high_hidden_visible_ratio")
    if values.get("display_mode") == "raw":
        findings.append("raw_reasoning_trace_displayed")
    if "redacted_block_count" in values and values["redacted_block_count"] >= policy["max_redacted_blocks"]:
        findings.append("repeated_reasoning_redaction")
    summary_chars = len(values["trace_summary"]) if "trace_summary" in values else None
    if summary_chars is not None and summary_chars > policy["max_summary_chars"]:
        findings.append("reasoning_summary_length_anomaly")
    return {**base, "control_status": "PASS",
            "finding_status": "FINDING" if findings else "NO_FINDING",
            "presence": presence,
            "findings": sorted(findings), "summary_chars": summary_chars,
            "thinking_tokens": hidden, "visible_output_tokens": visible,
            "budget_tokens": budget, "redacted_block_count": values.get("redacted_block_count"),
            "omitted_reasoning": values.get("omitted_reasoning"), "display_mode": values.get("display_mode")}


def main() -> None:
    try:
        records, receipt, caps, policy = load()
        results = [evaluate(record, caps["profiles"], policy) for record in records]
        eligible = sum(
            result["control_status"] != "NOT_APPLICABLE"
            for result in results
        )
        if eligible != receipt["eligible_record_count"]:
            raise ValueError("eligible population differs from completeness receipt")
        order = {
            "NOT_APPLICABLE": 0, "PASS": 1,
            "INSUFFICIENT_DATA": 2, "ERROR": 3,
        }
        control_status = max(
            (item["control_status"] for item in results),
            key=order.get,
            default="INSUFFICIENT_DATA",
        )
        finding_status = (
            "UNKNOWN"
            if control_status not in {"PASS", "NOT_APPLICABLE"}
            else "FINDING"
            if any(item["finding_status"] == "FINDING" for item in results)
            else "NO_FINDING"
            if eligible
            else "NOT_EVALUATED"
        )
        evidence = {"schema_version": "aidefend.d002004-evidence.v1", "control": "AID-D-002.004",
                    "control_status": control_status,
                    "finding_status": finding_status,
                    "receipt_sha256": sha256_file(RECEIPT),
                    "records_sha256": sha256_file(RECORDS), "capability_sha256": sha256_file(CAPS),
                    "capability_version": caps["profile_version"], "policy_sha256": sha256_file(POLICY),
                    "policy_version": policy["policy_version"], "evaluated_population": len(results),
                    "results": results, "evaluated_at": datetime.now(timezone.utc).isoformat()}
    except Exception as error:
        evidence = {"schema_version": "aidefend.d002004-evidence.v1", "control": "AID-D-002.004",
                    "control_status": "ERROR", "finding_status": "UNKNOWN",
                    "reason": f"{type(error).__name__}: {error}",
                    "evaluated_at": datetime.now(timezone.utc).isoformat()}
    EVIDENCE.parent.mkdir(parents=True, exist_ok=True)
    EVIDENCE.write_text(json.dumps(evidence, sort_keys=True, separators=(",", ":")) + "\n", encoding="utf-8")
    subprocess.run(["cosign", "sign-blob", "--yes", "--key", "env://D002004_EVIDENCE_SIGNING_KEY",
                    "--bundle", str(EVIDENCE) + ".sig", str(EVIDENCE)], check=True, timeout=REASONING_TELEMETRY_COMMAND_TIMEOUT_SECONDS)
    verify(EVIDENCE, "D002004_EVIDENCE_VERIFY_KEY")


if __name__ == "__main__":
    main()
Interpret results and retain evidence

Action: deploy the adapter at the provider-response observation point, require exact capability and completeness receipts, and forward signed findings to SIEM. Do not block traffic or turn absent provider fields into a clean result.

Verify safely

An independent verifier must re-extract fields from immutable sanitized response envelopes and recompute reasoning ratios, budget findings, disclosure findings, and population states, read evidence back, verify signatures and digests, and compare the canonical result digest.

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-002.004: Reasoning Trace & Thinking-Token Anomaly Telemetry | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service