Scope boundary

Owns reasoning-effort and hidden thinking-token budget tiers, clamping or denial, and enforcement evidence for model-specific reasoning compute. It does not own general request quotas, infrastructure throttling, anomaly detection, or ordinary invocation settings.

AID-I-003: AID-H-037.002 bounds model-specific reasoning or thinking compute; AID-I-003 owns general request rate limits, concurrency, service quotas, resource controls, and denial-of-wallet throttling. Reasoning budgets should feed but not duplicate the authoritative general quota system.

AID-D-005.007: AID-H-037.002 prevents excessive configured reasoning effort before or during invocation; AID-D-005.007 detects observed token, tool-use, parameter, and cost spikes. Detection reports abuse but does not set the reasoning budget.

AID-H-033.006: AID-H-037.002 owns reasoning-effort and thinking-token fields; AID-H-033.006 owns general sampling, output-token, log-probability, and provider safety-profile resolution. Both are enforced at invocation but have distinct policy dimensions and evidence.

Implementation guidance

• AID-H-037.002-G001: Enforce reasoning-specific budget tiers at the AI gateway and emit budget evidence for downstream cost-spike and availability monitoring.

AID-H-037.002-G001

Concept:

Reasoning-capable models may expose controls such as reasoning effort or thinking-budget tokens. Treat those settings as security-relevant. The gateway should choose a budget from a reviewed policy, clamp or reject caller-supplied overrides, and emit an event that downstream cost-spike detection can correlate with latency and billing.

Step 1: Define and bind reasoning budget tiers

Load this policy only after verifying its signed immutable bytes. Pass the SHA-256 digest of those exact verified bytes into the gate; the resulting decision and evidence event carry both the policy version and digest. Unknown intent classes fail closed instead of falling through to a default tier.

# File: reasoning_security/thinking_budget_policy.yaml
policy_version: "2026.07"
intent_tiers:
  low_risk_lookup: "low_risk_lookup"
  standard_request: "standard"
  high_impact_review: "high_impact_review"
  public_request: "untrusted_public"
tiers:
  low_risk_lookup:
    max_thinking_tokens: 256
    max_reasoning_effort: "low"
    allowed_user_trust: ["employee", "service"]
  standard:
    max_thinking_tokens: 1024
    max_reasoning_effort: "medium"
    allowed_user_trust: ["employee", "partner", "service"]
  high_impact_review:
    max_thinking_tokens: 4096
    max_reasoning_effort: "high"
    allowed_user_trust: ["employee"]
  untrusted_public:
    max_thinking_tokens: 512
    max_reasoning_effort: "low"
    allowed_user_trust: ["anonymous", "untrusted"]
deny:
  client_unbounded_budget: true
  unknown_reasoning_effort: true

Step 2: Validate policy, then clamp or reject requests

# File: reasoning_security/thinking_budget_gate.py
from __future__ import annotations

from dataclasses import dataclass
from typing import Callable, Literal


ReasoningEffort = Literal["low", "medium", "high"]
EFFORT_ORDER = {"low": 1, "medium": 2, "high": 3}
POLICY_FIELDS = {"policy_version", "intent_tiers", "tiers", "deny"}
TIER_FIELDS = {
    "max_thinking_tokens", "max_reasoning_effort", "allowed_user_trust",
}
REQUIRED_DENIALS = {"client_unbounded_budget", "unknown_reasoning_effort"}
HEX = set("0123456789abcdef")
ReserveQuota = Callable[[str, int], str]


@dataclass(frozen=True)
class BudgetDecision:
    decision: str
    request_id: str
    intent_class: str
    tier: str
    thinking_tokens: int
    reasoning_effort: ReasoningEffort | None
    policy_version: str
    policy_sha256: str
    quota_receipt_id: str | None
    reasons: list[str]


def require_sha256(value: object, field: str) -> str:
    if (
        not isinstance(value, str)
        or len(value) != 64
        or set(value) - HEX
        or value == "0" * 64
    ):
        raise ValueError(f"{field} must be a nonzero lowercase SHA-256 digest")
    return value


def validate_policy(
    policy: dict, policy_sha256: str, provider_capabilities: dict,
) -> tuple[str, dict[str, str], dict[str, dict]]:
    if not isinstance(policy, dict) or set(policy) != POLICY_FIELDS:
        raise ValueError("verified thinking-budget policy schema mismatch")
    version = policy["policy_version"]
    if not isinstance(version, str) or not version.strip() or len(version) > 64:
        raise ValueError("policy_version must be a bounded nonempty string")
    require_sha256(policy_sha256, "policy_sha256")

    intent_tiers = policy["intent_tiers"]
    tiers = policy["tiers"]
    deny = policy["deny"]
    if (
        not isinstance(intent_tiers, dict)
        or not intent_tiers
        or any(
            not isinstance(intent, str) or not intent
            or not isinstance(tier_name, str) or not tier_name
            for intent, tier_name in intent_tiers.items()
        )
        or not isinstance(tiers, dict)
        or not tiers
        or set(intent_tiers.values()) != set(tiers)
    ):
        raise ValueError("intent_tiers must map exactly onto the tier population")
    if (
        not isinstance(deny, dict)
        or set(deny) != REQUIRED_DENIALS
        or any(deny[name] is not True for name in REQUIRED_DENIALS)
    ):
        raise ValueError("policy must explicitly deny unbounded budgets and unknown effort")
    if (
        not isinstance(provider_capabilities, dict)
        or set(provider_capabilities)
        != {"max_thinking_tokens", "reasoning_efforts"}
    ):
        raise ValueError("verified provider-capability schema mismatch")
    provider_maximum = provider_capabilities["max_thinking_tokens"]
    provider_efforts = provider_capabilities["reasoning_efforts"]
    if (
        type(provider_maximum) is not int
        or provider_maximum <= 0
        or not isinstance(provider_efforts, list)
        or not provider_efforts
        or len(provider_efforts) != len(set(provider_efforts))
        or any(effort not in EFFORT_ORDER for effort in provider_efforts)
    ):
        raise ValueError("verified provider capabilities are invalid")

    for tier_name, tier in tiers.items():
        if not isinstance(tier_name, str) or not tier_name:
            raise ValueError("tier name must be nonempty")
        if not isinstance(tier, dict) or set(tier) != TIER_FIELDS:
            raise ValueError(f"thinking-budget tier schema differs: {tier_name}")
        maximum_tokens = tier["max_thinking_tokens"]
        maximum_effort = tier["max_reasoning_effort"]
        allowed_trust = tier["allowed_user_trust"]
        if (
            type(maximum_tokens) is not int
            or maximum_tokens <= 0
            or maximum_tokens > provider_maximum
            or maximum_effort not in EFFORT_ORDER
            or maximum_effort not in provider_efforts
            or not isinstance(allowed_trust, list)
            or not allowed_trust
            or len(allowed_trust) != len(set(allowed_trust))
            or any(not isinstance(item, str) or not item for item in allowed_trust)
        ):
            raise ValueError(f"tier is invalid or provider-incompatible: {tier_name}")
    return version, intent_tiers, tiers


def enforce_budget(
    *, request_id: str, authenticated_context: dict, policy: dict,
    policy_sha256: str, provider_capabilities: dict,
    requested_tokens: int | None, requested_effort: str | None,
    reserve_quota: ReserveQuota,
) -> BudgetDecision:
    policy_version, intent_tiers, tiers = validate_policy(
        policy, policy_sha256, provider_capabilities
    )
    if not isinstance(request_id, str) or not request_id:
        raise ValueError("request_id is required")
    context = authenticated_context if isinstance(authenticated_context, dict) else {}
    intent_class = context.get("intent_class")
    user_trust = context.get("user_trust")

    def block(tier_name: str, reason: str) -> BudgetDecision:
        return BudgetDecision(
            "block", request_id, str(intent_class or ""), tier_name,
            0, None, policy_version, policy_sha256, None, [reason],
        )

    if (
        not isinstance(intent_class, str) or not intent_class
        or not isinstance(user_trust, str) or not user_trust
    ):
        return block("", "authoritative_context_missing")
    tier_name = intent_tiers.get(intent_class)
    if tier_name is None:
        return block("", "intent_not_mapped")
    tier = tiers[tier_name]
    if user_trust not in tier["allowed_user_trust"]:
        return block(tier_name, "user_trust_not_allowed")
    if requested_tokens is not None and (
        type(requested_tokens) is not int or requested_tokens <= 0
    ):
        return block(tier_name, "invalid_token_budget")
    if requested_effort is not None and (
        not isinstance(requested_effort, str)
        or requested_effort not in EFFORT_ORDER
    ):
        return block(tier_name, "unknown_reasoning_effort")

    maximum_tokens = tier["max_thinking_tokens"]
    maximum_effort = tier["max_reasoning_effort"]
    requested = maximum_tokens if requested_tokens is None else requested_tokens
    effective_tokens = min(requested, maximum_tokens)
    effective_effort = requested_effort or maximum_effort
    reasons: list[str] = []
    if requested > maximum_tokens:
        reasons.append("thinking_budget_clamped")
    if EFFORT_ORDER[effective_effort] > EFFORT_ORDER[maximum_effort]:
        effective_effort = maximum_effort
        reasons.append("reasoning_effort_clamped")
    quota_receipt_id = reserve_quota(request_id, effective_tokens)
    if not isinstance(quota_receipt_id, str) or not quota_receipt_id:
        raise RuntimeError("quota reservation returned no receipt")
    return BudgetDecision(
        "allow", request_id, intent_class, tier_name,
        effective_tokens, effective_effort, policy_version, policy_sha256,
        quota_receipt_id, reasons,
    )

Step 3: Emit policy-bound budget evidence

# File: reasoning_security/budget_event.py
from __future__ import annotations

import hashlib
import json
from dataclasses import asdict
from typing import Callable

from reasoning_security.thinking_budget_gate import BudgetDecision


WriteEvidence = Callable[[str, bytes], str]
ReadEvidence = Callable[[str], bytes]


def emit_reasoning_budget_event(
    *, event_id: str, subject_pseudonym: str, provider_model: str,
    decision: BudgetDecision, write: WriteEvidence, read: ReadEvidence,
) -> dict:
    if any(
        not isinstance(value, str) or not value
        for value in (event_id, subject_pseudonym, provider_model)
    ):
        raise ValueError("event identity and provider model are required")
    event = {
        "schema_version": "aidefend.reasoning-budget-event.v1",
        "event_id": event_id,
        "subject_pseudonym": subject_pseudonym,
        "provider_model": provider_model,
        **asdict(decision),
    }
    payload = (
        json.dumps(event, sort_keys=True, separators=(",", ":"), allow_nan=False)
        + "\n"
    ).encode("utf-8")
    receipt_id = write(event_id, payload)
    if not isinstance(receipt_id, str) or not receipt_id or read(receipt_id) != payload:
        raise RuntimeError("reasoning-budget evidence readback mismatch")
    return {
        "receipt_id": receipt_id,
        "event_sha256": hashlib.sha256(payload).hexdigest(),
        "outcome": "PASS",
    }

Step 4: Run the policy fixture

# File: tests/test_thinking_budget_gate.py
from copy import deepcopy

from reasoning_security.thinking_budget_gate import enforce_budget


POLICY_SHA256 = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
POLICY = {
    "policy_version": "2026.07",
    "intent_tiers": {
        "low_risk_lookup": "low_risk_lookup",
        "standard_request": "standard",
        "high_impact_review": "high_impact_review",
        "public_request": "untrusted_public",
    },
    "tiers": {
        "low_risk_lookup": {
            "max_thinking_tokens": 256,
            "max_reasoning_effort": "low",
            "allowed_user_trust": ["employee", "service"],
        },
        "standard": {
            "max_thinking_tokens": 1024,
            "max_reasoning_effort": "medium",
            "allowed_user_trust": ["employee", "partner", "service"],
        },
        "high_impact_review": {
            "max_thinking_tokens": 4096,
            "max_reasoning_effort": "high",
            "allowed_user_trust": ["employee"],
        },
        "untrusted_public": {
            "max_thinking_tokens": 512,
            "max_reasoning_effort": "low",
            "allowed_user_trust": ["anonymous", "untrusted"],
        },
    },
    "deny": {
        "client_unbounded_budget": True,
        "unknown_reasoning_effort": True,
    },
}
CAPABILITIES = {
    "max_thinking_tokens": 8192,
    "reasoning_efforts": ["low", "medium", "high"],
}


def reserve(request_id: str, tokens: int) -> str:
    assert request_id == "req-2026-0719-0042"
    assert 0 < tokens <= CAPABILITIES["max_thinking_tokens"]
    return "quota-receipt-0042"


def decide(**overrides):
    values = {
        "request_id": "req-2026-0719-0042",
        "authenticated_context": {
            "intent_class": "high_impact_review",
            "user_trust": "employee",
        },
        "policy": POLICY,
        "policy_sha256": POLICY_SHA256,
        "provider_capabilities": CAPABILITIES,
        "requested_tokens": 8192,
        "requested_effort": "high",
        "reserve_quota": reserve,
    }
    values.update(overrides)
    return enforce_budget(**values)


overflow = decide()
assert overflow.decision == "allow"
assert overflow.thinking_tokens == 4096
assert overflow.reasons == ["thinking_budget_clamped"]
assert overflow.policy_version == "2026.07"
assert overflow.policy_sha256 == POLICY_SHA256
assert decide(requested_tokens=None).thinking_tokens == 4096
assert decide(requested_tokens=-1).reasons == ["invalid_token_budget"]
assert decide(requested_tokens="unbounded").reasons == ["invalid_token_budget"]
assert decide(requested_effort="extreme").reasons == ["unknown_reasoning_effort"]
assert decide(authenticated_context={
    "intent_class": "high_impact_review", "user_trust": "anonymous",
}).reasons == ["user_trust_not_allowed"]

for denial in ("client_unbounded_budget", "unknown_reasoning_effort"):
    invalid_policy = deepcopy(POLICY)
    invalid_policy["deny"][denial] = False
    try:
        decide(policy=invalid_policy)
    except ValueError as exc:
        assert "explicitly deny" in str(exc)
    else:
        raise AssertionError(f"disabled deny flag was accepted: {denial}")

print("PASS")

Run python tests/test_thinking_budget_gate.py. The fixture proves exact YAML/Python schema parity, policy digest/version propagation, bounded defaulting, overflow clamping, invalid-input and trust denial, and fail-closed rejection when a required deny flag is disabled.

Operational notes

• Do not expose an unlimited or user-controlled thinking budget for public or unauthenticated traffic.
• Review budget tiers when model pricing, provider semantics, or workflow criticality changes.
• Feed budget decisions, actual thinking-token usage, latency, and provider cost into AID-D-002.004 and AID-D-005.007 so abuse is visible after enforcement.

Action: make reasoning effort a gateway-controlled field, not a free-form client parameter. Clamp excessive requests, block invalid effort levels, and log every budget decision with enough context for cost and availability monitoring.

Keep the action safely bounded

Reconcile the complete population of every eligible reasoning request, requested/effective budget decision, provider attempt, and actual usage record; reject expired, revoked, or stale policy and asset bindings under the policy-defined freshness window.

Verify safely

A separately credentialed verifier independently replays policy/tier selection, provider translation, quota and actual usage with forged intent/trust, absent/negative/overflow budget, unknown effort, route failover, concurrency, provider drift, and missing-usage fixtures, reads back effective state from the enforcement system, exercises positive, negative, boundary, stale-evidence, and dependency-failure fixtures, and reconciles every eligible item.

Open-source tools

• LiteLLM
• OpenTelemetry
• Prometheus
• Grafana
• Envoy (custom reasoning-budget policy required)

Commercial / hosted tools

• Datadog (budget evidence only; not reasoning-budget enforcement)
• New Relic (budget evidence only; not reasoning-budget enforcement)
• Splunk Enterprise Security (budget evidence only; not reasoning-budget enforcement)
• Amazon Bedrock Converse API (Claude extended/adaptive thinking)

Threat mappings

MITRE ATLAS

• AML.T0034 Cost Harvesting
• AML.T0034.001 Cost Harvesting: Resource-Intensive Queries
• AML.T0029 Denial of AI Service

MAESTRO

• Resource Hijacking (L4)

OWASP LLM Top 10 2026

• LLM06:2026 Unbounded Consumption

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• N/A

NIST Adversarial Machine Learning 2025

• NISTAML.014 Energy-latency

Cisco Integrated AI Security and Safety Framework

• AISubtech-13.1.1 Compute Exhaustion
• AISubtech-13.1.3 Model Denial of Service
• AISubtech-13.1.5 Decision Paralysis Attacks (bounds unproductive hidden reasoning; it does not guarantee task completion)
• AISubtech-13.2.1 Service Misuse for Cost Inflation
• AITech-13.1 Disruption of Availability
• AITech-13.2 Cost Harvesting / Repurposing

Google Secure AI Framework 2.0 - Risks

• DMS: Denial of ML Service

Databricks AI Security Framework 3.0

• Model Serving - Inference requests 9.7: Denial of Service (DoS)
• Agents - Core 13.4: Resource Overload