AID-I-003.003: Resource, Token & Spend Budget Throttling

Version 1.20261015 · Actionable sub-technique

Throttle cumulative resource consumption when suspicious tenants, sessions, users, agents, or API keys exceed GPU, CPU, token, tool-call, retry, concurrency, or spend budgets. This sub-technique owns multi-dimensional quota policy, server-side usage accounting, Kubernetes resource clamps, budget ledgers, and quota evidence.

Pillars: infra, app

Lifecycle: response, operation

Scope boundary

Owns multi-dimensional cumulative resource and spend budgets, server-side accounting, quota ledgers, platform resource clamps, and quota evidence. It controls expensive low-frequency or recursive consumption rather than only request arrival rate.

AID-I-003.002: AID-I-003.003 throttles cumulative compute, token, tool-call, retry, concurrency, and spend consumption; AID-I-003.002 throttles request admission by identity, tenant, session, API key, or trust tier. Both may apply to one caller, but they measure and enforce different populations and cannot share a PASS result.

Implementation guidance

AID-I-003.003-G001

Throttle GPU/CPU for a suspicious tenant namespace in Kubernetes.
Kubernetes runtime prerequisites

ResourceQuota constrains aggregate admission; LimitRange defaults or rejects new container resources. Neither retroactively throttles running Pods. Extended resources use requests.nvidia.com/gpu; do not publish a separate limits.nvidia.com/gpu quota. A production clamp therefore has two explicit phases: first apply the signed admission quota to stop new expensive work, then reconcile every existing controller and unmanaged Pod to the signed quarantine profile, preserving required forensic evidence before any restart or eviction.

Transactional clamp controller
# File: containment/kubernetes_compute_clamp.py
from __future__ import annotations

from dataclasses import dataclass
from typing import Callable


@dataclass(frozen=True)
class VerifiedComputeClampPolicy:
    policy_sha256: str
    incident_id: str
    namespace: str
    namespace_uid: str
    selector: str
    quota_hard: dict[str, str]
    max_cpu_per_container: str
    max_memory_per_container: str
    target_replicas: int
    allow_running_gpu: bool


@dataclass(frozen=True)
class KubernetesClampAdapter:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    apply_admission_quota: Callable[[VerifiedComputeClampPolicy], dict[str, object]]
    preserve_forensics: Callable[[VerifiedComputeClampPolicy], dict[str, object]]
    reconcile_existing_workloads: Callable[[VerifiedComputeClampPolicy], dict[str, object]]
    read_complete_population: Callable[[VerifiedComputeClampPolicy], dict[str, object]]


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


def apply_compute_clamp(
    *,
    policy: VerifiedComputeClampPolicy,
    kubernetes: KubernetesClampAdapter,
    evidence: EvidenceWriter,
) -> dict[str, object]:
    if policy.target_replicas < 0:
        raise ValueError("target replicas cannot be negative")
    if "limits.nvidia.com/gpu" in policy.quota_hard:
        raise ValueError("GPU extended-resource quota uses requests only")
    if not policy.allow_running_gpu:
        gpu_limit = policy.quota_hard.get("requests.nvidia.com/gpu")
        if gpu_limit not in {"0", None}:
            raise ValueError("GPU quarantine policy is internally inconsistent")

    quota = kubernetes.apply_admission_quota(policy)
    forensics = kubernetes.preserve_forensics(policy)
    reconcile = kubernetes.reconcile_existing_workloads(policy)
    state = kubernetes.read_complete_population(policy)
    if (
        state.get("namespace_uid") != policy.namespace_uid
        or state.get("quota_hard") != policy.quota_hard
        or state.get("population_complete") is not True
    ):
        raise PermissionError("effective namespace/quota population mismatch")
    if state.get("nonconforming_containers"):
        raise PermissionError("running container exceeds quarantine resources")
    if not policy.allow_running_gpu and state.get("running_gpu_requests", 0) != 0:
        raise PermissionError("GPU workload remains after quarantine reconciliation")
    if state.get("controllers_above_target_replicas"):
        raise PermissionError("controller replica clamp is not effective")

    final = evidence.append_and_verify({
        "schema_version": "aidefend.kubernetes-compute-clamp.v1",
        "control": "AID-I-003.003",
        "incident_id": policy.incident_id,
        "namespace": policy.namespace,
        "namespace_uid": policy.namespace_uid,
        "selector": policy.selector,
        "policy_sha256": policy.policy_sha256,
        "quota_receipt_sha256": quota["receipt_sha256"],
        "forensic_receipt_sha256": forensics["receipt_sha256"],
        "reconcile_receipt_sha256": reconcile["receipt_sha256"],
        "population_receipt_sha256": state["receipt_sha256"],
        "pod_population_sha256": state["pod_population_sha256"],
        "controller_population_sha256": state["controller_population_sha256"],
        "outcome": "PASS",
    })
    return {"outcome": "PASS", "evidence": final}
Adapter and evidence requirements

The signed policy supplies actual CPU/memory/GPU quantities, selector, controller types, disruption/forensic rules, deadlines, and restoration approval; no sample values are universal. The adapter verifies namespace UID before every write, uses server-side apply with a dedicated field manager, applies only to the exact selector, records prior specs, waits for controller rollout/scale convergence, and explicitly handles Jobs, bare Pods, init/ephemeral containers, device-plugin resources, and terminating Pods. Restoration is a separate compare-and-swap action against the recorded prior state.

Action: Throttle GPU/CPU for a suspicious tenant namespace in Kubernetes. Enforce the decision at the named isolation boundary and fail closed: deny, quarantine, downgrade, throttle or halt as this step specifies whenever policy, adapter execution, evidence commit or independent readback is incomplete.

AID-I-003.003-G002

Enforce budget-triggered throttling and safe-mode downgrade when per-session, per-user, per-agent, or per-API-key token and spend thresholds are exceeded or when abnormal cost-abuse scores cross policy thresholds.
Reservation-before-dispatch model

Count cumulative budgets with integer units: tokens, tool calls, retries, concurrency slots, GPU milliseconds, and microcurrency—not binary floating-point dollars. Resolve session, user, agent, API-key, and tenant scopes from authenticated server state. Before dispatch, atomically reserve the request's worst-case usage against every applicable scope/dimension; configure provider/model/tool hard caps so actual use cannot exceed the reservation; after completion, settle server-verified actual usage and release the remainder. A post-call increment is too late to prevent overspend.

Multi-scope budget gate
# File: gateway/budget_gate.py
from __future__ import annotations

from dataclasses import dataclass
from typing import Callable


DIMENSIONS = {
    "input_tokens", "output_tokens", "tool_calls", "retries",
    "concurrency_slots", "gpu_milliseconds", "spend_microunits",
}


@dataclass(frozen=True)
class VerifiedBudgetPolicy:
    policy_sha256: str
    policy_version: int
    window_id: str
    scope_ids: tuple[str, ...]
    limits_by_scope: dict[str, dict[str, int]]
    on_breach_action: str


@dataclass(frozen=True)
class BudgetRequest:
    request_id: str
    tenant_id: str
    reservation_units: dict[str, int]


@dataclass(frozen=True)
class BudgetLedger:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    reserve_once: Callable[[BudgetRequest, VerifiedBudgetPolicy], dict[str, object]]
    settle_once: Callable[[str, dict[str, int], str], dict[str, object]]


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


@dataclass(frozen=True)
class CappedDispatcher:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    execute: Callable[[BudgetRequest, dict[str, int], str], dict[str, object]]


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


def validate(policy: VerifiedBudgetPolicy, request: BudgetRequest) -> None:
    if not policy.scope_ids or set(policy.scope_ids) != set(policy.limits_by_scope):
        raise ValueError("signed budget policy has an invalid scope population")
    if not request.reservation_units or set(request.reservation_units) - DIMENSIONS:
        raise ValueError("request uses an unknown or empty budget dimension")
    for scope_id, limits in policy.limits_by_scope.items():
        if set(request.reservation_units) - set(limits):
            raise ValueError("scope is missing a required budget dimension: " + scope_id)
        if any(isinstance(value, bool) or not isinstance(value, int) or value < 0 for value in limits.values()):
            raise ValueError("budget limits require nonnegative integer units")
    if any(isinstance(value, bool) or not isinstance(value, int) or value < 0 for value in request.reservation_units.values()):
        raise ValueError("reservation requires nonnegative integer units")


def execute_with_budget(
    *,
    policy: VerifiedBudgetPolicy,
    request: BudgetRequest,
    ledger: BudgetLedger,
    dispatcher: CappedDispatcher,
    usage_verifier: UsageVerifier,
    evidence: EvidenceWriter,
) -> dict[str, object]:
    validate(policy, request)
    reservation = ledger.reserve_once(request=request, policy=policy)
    if reservation.get("outcome") == "FAIL":
        receipt = evidence.append_and_verify({
            "schema_version": "aidefend.budget-decision.v1",
            "control": "AID-I-003.003",
            "request_id": request.request_id,
            "tenant_id": request.tenant_id,
            "policy_sha256": policy.policy_sha256,
            "window_id": policy.window_id,
            "scope_ids": list(policy.scope_ids),
            "reservation_receipt_sha256": reservation["receipt_sha256"],
            "breached_dimensions": reservation["breached_dimensions"],
            "action": policy.on_breach_action,
            "outcome": "FAIL",
        })
        return {"allowed": False, "action": policy.on_breach_action, "evidence": receipt}
    if reservation.get("outcome") != "PASS":
        raise RuntimeError("budget ledger returned no closed reservation outcome")

    execution = dispatcher.execute(
        request=request,
        hard_caps=request.reservation_units,
        reservation_id=str(reservation["reservation_id"]),
    )
    usage = usage_verifier.verify(execution, request)
    actual = usage.get("actual_units")
    if not isinstance(actual, dict) or set(actual) != set(request.reservation_units):
        raise PermissionError("usage receipt has an invalid dimension population")
    for dimension, value in actual.items():
        if (
            isinstance(value, bool)
            or not isinstance(value, int)
            or value < 0
            or value > request.reservation_units[dimension]
        ):
            raise PermissionError("actual usage exceeds or violates reservation")
    settlement = ledger.settle_once(
        reservation_id=str(reservation["reservation_id"]),
        actual_units=actual,
        usage_receipt_sha256=str(usage["receipt_sha256"]),
    )
    if settlement.get("outcome") != "PASS":
        raise RuntimeError("budget settlement failed")
    final = evidence.append_and_verify({
        "schema_version": "aidefend.budget-decision.v1",
        "control": "AID-I-003.003",
        "request_id": request.request_id,
        "tenant_id": request.tenant_id,
        "policy_sha256": policy.policy_sha256,
        "policy_version": policy.policy_version,
        "window_id": policy.window_id,
        "scope_ids": list(policy.scope_ids),
        "reserved_units": request.reservation_units,
        "actual_units": actual,
        "reservation_receipt_sha256": reservation["receipt_sha256"],
        "usage_receipt_sha256": usage["receipt_sha256"],
        "settlement_receipt_sha256": settlement["receipt_sha256"],
        "outcome": "PASS",
    })
    return {"allowed": True, "result_ref": execution["result_ref"], "evidence": final}
Ledger and recovery requirements

The ledger uses one serializable transaction or atomic script over all sorted scope/dimension keys, a unique request ID, server time, signed window boundaries, and durable reservation states. It records DENIED reservations for idempotent replay. Settlement is compare-and-swap from RESERVED to SETTLED exactly once. If execution, usage verification, or evidence fails, do not release blindly: a reconciler queries the provider/gateway usage receipt, settles observed use, and only then releases the remainder. Concurrency slots are leases with fencing tokens; expired leases require independent no-running-work readback.

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-003.003: Resource, Token & Spend Budget Throttling | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service