Scope boundary

Owns early-warning findings for abnormal token consumption, tool volume, retry bursts, latency, provider cost, request fan-out, and model-specific request-control parameters. It estimates and correlates abuse or runaway automation but does not throttle, stop loops, or enforce parameters.

AID-D-002.004: AID-D-005.007 detects broad request, token, tool, latency, parameter, and cost spikes; AID-D-002.004 detects anomalies specific to reasoning traces and thinking-token accounting. Reasoning telemetry is one specialized input to the broader resource and cost-abuse detector.

AID-I-003: AID-D-005.007 emits budget, fan-out, token, or runaway-loop risk findings; AID-I-003 applies throttling, quarantine, or safe-mode containment. An alert does not reduce resource consumption until an Isolate control enforces the response.

Warning

Low on Precision of Cost Estimation

Estimated spend signals may lag actual provider billing or differ by routing, cached responses, or vendor-specific pricing behavior. Treat this sub-technique as an early-warning and correlation layer, not as the sole source of financial truth. Recalibrate alert thresholds against actual billing data on the cadence required by the signed environment policy and record the calibration evidence and policy version.

Implementation guidance

• AID-D-005.007-G001: Emit low-cardinality aggregate cost metrics and exact identity-correlatable request events without placing user, session, agent, request, or raw tenant IDs in metric labels.
• AID-D-005.007-G002: Trigger alerts when cost-linked metrics deviate sharply from historical baselines or violate rate-of-change thresholds, especially when correlated with repeated tool chains or non-progressing loops.
• AID-D-005.007-G003: Profile generative inference control parameters at the API layer and alert on out-of-range, high-risk, or baseline-breaking requests.
• AID-D-005.007-G004: Correlate abnormal spend signals with downstream response events such as throttle, safe-mode downgrade, session quarantine, or human review so cost-abuse investigations have a single auditable incident timeline.

AID-D-005.007-G001

Policy-bound normalized telemetry

from __future__ import annotations

# File: cost_monitor/telemetry.py

import hashlib
import hmac
import json
import math
import os
import re
import subprocess
import tempfile
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path

from prometheus_client import Counter

POLICY_PATH = Path(os.environ["COST_TELEMETRY_POLICY_PATH"])
POLICY_SIGNATURE = Path(os.environ["COST_TELEMETRY_POLICY_SIGNATURE_PATH"])
POLICY_PUBLIC_KEY = Path(os.environ["COST_TELEMETRY_POLICY_PUBLIC_KEY_PATH"])
CORRELATION_KEY = bytes.fromhex(os.environ["COST_EVENT_HMAC_HEX"])
SIGNATURE_VERIFY_TIMEOUT_SECONDS = float(
    os.environ["D005007_SIGNATURE_VERIFY_TIMEOUT_SECONDS"]
)
if not math.isfinite(SIGNATURE_VERIFY_TIMEOUT_SECONDS) or SIGNATURE_VERIFY_TIMEOUT_SECONDS <= 0:
    raise RuntimeError("signature verification timeout must be finite and positive")

REQUESTS = Counter(
    "ai_gateway_requests_total",
    "Authenticated AI gateway outcomes",
    ["model_class", "workload_class", "outcome"],
)
COST_UNITS = Counter(
    "ai_gateway_cost_units_total",
    "Trusted provider/tokenizer-derived cost units",
    ["model_class", "workload_class", "outcome"],
)
TOOL_CALLS = Counter(
    "ai_gateway_tool_calls_total",
    "Brokered tool-call outcomes",
    ["tool_class", "workload_class", "outcome"],
)


@dataclass(frozen=True)
class VerifiedIdentity:
    issuer: str
    tenant_id: str
    subject: str
    session_id: str


@dataclass(frozen=True)
class TrustedRequestResult:
    request_id: str
    model_id: str
    model_class: str
    workload_class: str
    outcome: str
    cost_units: int
    input_tokens: int
    output_tokens: int
    tool_class_counts: dict[str, int]
    parameter_digest: str
    occurred_at: datetime


_AIDEFEND_CAPTURED: dict[Path, bytes] = {}


def _aidefend_bytes(path: Path) -> bytes:
    normalized = Path(path)
    if normalized not in _AIDEFEND_CAPTURED:
        _AIDEFEND_CAPTURED[normalized] = normalized.read_bytes()
    return _AIDEFEND_CAPTURED[normalized]


def _aidefend_write_private(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 _aidefend_strict_json(raw: bytes) -> object:
    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(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=lambda token: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + token)
        ),
    )


def _aidefend_strict_jsonl(raw: bytes) -> list[object]:
    lines = raw.decode("utf-8", errors="strict").splitlines()
    if not lines or any(not line.strip() for line in lines):
        raise ValueError("JSONL input is empty or contains a blank record")
    return [_aidefend_strict_json(line.encode("utf-8")) for line in lines]


def _aidefend_capture_signed_bytes(
    payload_path: Path,
    proof_path: Path,
    key_reference: str | Path,
    timeout_seconds: float,
) -> bytes:
    payload_path = Path(payload_path)
    proof_path = Path(proof_path)
    if payload_path == proof_path:
        raise RuntimeError("signed payload and signature paths collide")
    payload_raw = payload_path.read_bytes()
    proof_raw = proof_path.read_bytes()
    key_text = str(key_reference)
    remote_key = bool(re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*://", key_text
    )) or key_text.startswith("pkcs11:")
    key_raw = None if remote_key else (
        key_text.encode("utf-8")
        if key_text.startswith("-----BEGIN ")
        else Path(key_text).read_bytes()
    )
    with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        proof_snapshot = root / "payload.sig"
        _aidefend_write_private(payload_snapshot, payload_raw)
        _aidefend_write_private(proof_snapshot, proof_raw)
        captured = [(payload_snapshot, payload_raw), (proof_snapshot, proof_raw)]
        key_argument = key_text
        if key_raw is not None:
            key_snapshot = root / "verify-key.pub"
            _aidefend_write_private(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(proof_snapshot), str(payload_snapshot)],
                check=True, capture_output=True, text=True,
                timeout=timeout_seconds,
            )
            if any(
                not hmac.compare_digest(snapshot.read_bytes(), expected)
                for snapshot, expected in captured
            ):
                raise RuntimeError("private signature-verification snapshot changed")
        finally:
            for snapshot, _ in captured:
                os.chmod(snapshot, 0o600)
    _AIDEFEND_CAPTURED[payload_path] = payload_raw
    return payload_raw


def load_policy():
    raw = _aidefend_capture_signed_bytes(
        POLICY_PATH, POLICY_SIGNATURE, POLICY_PUBLIC_KEY,
        SIGNATURE_VERIFY_TIMEOUT_SECONDS,
    )
    policy = _aidefend_strict_json(raw)
    required = {
        "schema_version", "policy_version", "model_classes",
        "workload_classes", "tool_classes", "outcomes",
        "cost_method_version",
    }
    if not isinstance(policy, dict) or set(policy) != required:
        raise ValueError("telemetry policy schema mismatch")
    if policy["schema_version"] != "aidefend.cost_telemetry_policy.v1":
        raise ValueError("unsupported telemetry policy")
    for field in (
        "model_classes", "workload_classes", "tool_classes", "outcomes"
    ):
        values = policy[field]
        if (
            not isinstance(values, list) or not values
            or any(not isinstance(value, str) or not value for value in values)
            or len(values) != len(set(values))
        ):
            raise ValueError(f"{field} is not a bounded enum")
    return policy, hashlib.sha256(raw).hexdigest()


def ref(*parts):
    material = "\0".join(parts).encode()
    return hmac.new(CORRELATION_KEY, material, hashlib.sha256).hexdigest()


def non_negative_int(value, field):
    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
        raise ValueError(f"{field} must be a non-negative integer")
    return value


def record(identity, result):
    policy, policy_sha256 = load_policy()
    if not all((
        identity.issuer, identity.tenant_id, identity.subject,
        identity.session_id, result.request_id, result.model_id,
    )):
        raise ValueError("authenticated identity/result context incomplete")
    if result.model_class not in policy["model_classes"]:
        raise ValueError("unknown model class")
    if result.workload_class not in policy["workload_classes"]:
        raise ValueError("unknown workload class")
    if result.outcome not in policy["outcomes"]:
        raise ValueError("unknown outcome")
    for field in ("cost_units", "input_tokens", "output_tokens"):
        non_negative_int(getattr(result, field), field)
    if (
        not isinstance(result.tool_class_counts, dict)
        or any(key not in policy["tool_classes"] for key in result.tool_class_counts)
    ):
        raise ValueError("unknown tool class")
    for key, value in result.tool_class_counts.items():
        non_negative_int(value, f"tool_class_counts.{key}")
    if (
        not isinstance(result.parameter_digest, str)
        or len(result.parameter_digest) != 64
    ):
        raise ValueError("parameter digest invalid")
    if result.occurred_at.utcoffset() is None:
        raise ValueError("occurred_at lacks timezone")

    REQUESTS.labels(
        result.model_class, result.workload_class, result.outcome
    ).inc()
    COST_UNITS.labels(
        result.model_class, result.workload_class, result.outcome
    ).inc(result.cost_units)
    for tool_class, count in result.tool_class_counts.items():
        TOOL_CALLS.labels(
            tool_class, result.workload_class, result.outcome
        ).inc(count)

    event = {
        "schema_version": "aidefend.cost_event.v1",
        "event_id": ref(result.request_id, result.occurred_at.isoformat()),
        "occurred_at": result.occurred_at.astimezone(timezone.utc).isoformat(),
        "request_id": result.request_id,
        "principal_ref": ref(
            identity.issuer, identity.tenant_id, identity.subject
        ),
        "session_ref": ref(identity.issuer, identity.session_id),
        "tenant_ref": ref(identity.issuer, identity.tenant_id),
        "model_id": result.model_id,
        "model_class": result.model_class,
        "workload_class": result.workload_class,
        "outcome": result.outcome,
        "cost_units": result.cost_units,
        "input_tokens": result.input_tokens,
        "output_tokens": result.output_tokens,
        "tool_class_counts": result.tool_class_counts,
        "parameter_digest": result.parameter_digest,
        "authn_context_sha256": ref(
            identity.issuer, identity.subject, identity.tenant_id
        ),
        "telemetry_policy_version": policy["policy_version"],
        "telemetry_policy_sha256": policy_sha256,
        "cost_method_version": policy["cost_method_version"],
    }
    print(json.dumps(event, sort_keys=True, allow_nan=False))
    return event

Evidence: preserve the signed telemetry policy and signature, policy digest/version, normalized events, append-only ingestion receipt, and metric scrape coverage. Exact IDs remain in the protected event stream, never Prometheus labels. Missing identity, unknown enum, invalid numeric value, policy verification outage, or append-only delivery failure is ERROR, not a quiet zero.

Verify safely

An independent verifier must recompute per-request units and cost from immutable usage and pricing artifacts, compare provider invoices or usage exports, and reconcile exact events with low-cardinality aggregate series without identity labels, read evidence back, verify signatures, digests, and immutability, and compare the canonical result digest.

AID-D-005.007-G002

Deterministic signed-policy analyzer

from __future__ import annotations

# File: cost_monitor/window_analyzer.py

import hashlib
import hmac
import json
import math
import os
import re
import subprocess
import tempfile
from collections import Counter
from datetime import datetime
from pathlib import Path

POLICY_FIELDS = {
    "schema_version", "policy_version", "window_seconds",
    "min_events_per_window", "min_previous_cost_units",
    "max_cost_units_by_workload_class",
    "max_requests_by_workload_class",
    "max_tool_calls_by_class", "max_cost_rate_ratio",
}
RECEIPT_FIELDS = {
    "schema_version", "events_sha256", "window_start", "window_end",
    "expected_sources", "observed_sources", "collection_complete",
}


SIGNATURE_VERIFY_TIMEOUT_SECONDS = float(
    os.environ["D005007_SIGNATURE_VERIFY_TIMEOUT_SECONDS"]
)
if not math.isfinite(SIGNATURE_VERIFY_TIMEOUT_SECONDS) or SIGNATURE_VERIFY_TIMEOUT_SECONDS <= 0:
    raise RuntimeError("signature verification timeout must be finite and positive")


_AIDEFEND_CAPTURED: dict[Path, bytes] = {}


def _aidefend_bytes(path: Path) -> bytes:
    normalized = Path(path)
    if normalized not in _AIDEFEND_CAPTURED:
        _AIDEFEND_CAPTURED[normalized] = normalized.read_bytes()
    return _AIDEFEND_CAPTURED[normalized]


def _aidefend_write_private(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 _aidefend_strict_json(raw: bytes) -> object:
    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(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=lambda token: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + token)
        ),
    )


def _aidefend_strict_jsonl(raw: bytes) -> list[object]:
    lines = raw.decode("utf-8", errors="strict").splitlines()
    if not lines or any(not line.strip() for line in lines):
        raise ValueError("JSONL input is empty or contains a blank record")
    return [_aidefend_strict_json(line.encode("utf-8")) for line in lines]


def _aidefend_capture_signed_bytes(
    payload_path: Path,
    proof_path: Path,
    key_reference: str | Path,
    timeout_seconds: float,
) -> bytes:
    payload_path = Path(payload_path)
    proof_path = Path(proof_path)
    if payload_path == proof_path:
        raise RuntimeError("signed payload and signature paths collide")
    payload_raw = payload_path.read_bytes()
    proof_raw = proof_path.read_bytes()
    key_text = str(key_reference)
    remote_key = bool(re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*://", key_text
    )) or key_text.startswith("pkcs11:")
    key_raw = None if remote_key else (
        key_text.encode("utf-8")
        if key_text.startswith("-----BEGIN ")
        else Path(key_text).read_bytes()
    )
    with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        proof_snapshot = root / "payload.sig"
        _aidefend_write_private(payload_snapshot, payload_raw)
        _aidefend_write_private(proof_snapshot, proof_raw)
        captured = [(payload_snapshot, payload_raw), (proof_snapshot, proof_raw)]
        key_argument = key_text
        if key_raw is not None:
            key_snapshot = root / "verify-key.pub"
            _aidefend_write_private(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(proof_snapshot), str(payload_snapshot)],
                check=True, capture_output=True, text=True,
                timeout=timeout_seconds,
            )
            if any(
                not hmac.compare_digest(snapshot.read_bytes(), expected)
                for snapshot, expected in captured
            ):
                raise RuntimeError("private signature-verification snapshot changed")
        finally:
            for snapshot, _ in captured:
                os.chmod(snapshot, 0o600)
    _AIDEFEND_CAPTURED[payload_path] = payload_raw
    return payload_raw



def sha256_file(path):
    return hashlib.sha256(_aidefend_bytes(Path(path))).hexdigest()


def verify_blob(path, signature, public_key):
    return _aidefend_capture_signed_bytes(
        Path(path), Path(signature), public_key,
        SIGNATURE_VERIFY_TIMEOUT_SECONDS,
    )


def parse_time(value):
    if not isinstance(value, str):
        raise ValueError("timestamp is not text")
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.utcoffset() is None:
        raise ValueError("timestamp lacks timezone")
    return parsed


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


def finite_positive(value, field):
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError(f"{field} must be numeric")
    number = float(value)
    if not math.isfinite(number) or number <= 0:
        raise ValueError(f"{field} must be finite and positive")
    return number


def load_policy(path, signature, public_key):
    raw = verify_blob(path, signature, public_key)
    policy = _aidefend_strict_json(raw)
    if not isinstance(policy, dict) or set(policy) != POLICY_FIELDS:
        raise ValueError("cost analysis policy schema mismatch")
    if policy["schema_version"] != "aidefend.cost_analysis_policy.v1":
        raise ValueError("unsupported cost analysis policy")
    if not isinstance(policy["policy_version"], str) or not policy["policy_version"]:
        raise ValueError("missing policy version")
    positive_int(policy["window_seconds"], "window_seconds")
    positive_int(policy["min_events_per_window"], "min_events_per_window")
    positive_int(policy["min_previous_cost_units"], "min_previous_cost_units")
    finite_positive(policy["max_cost_rate_ratio"], "max_cost_rate_ratio")
    for field in (
        "max_cost_units_by_workload_class",
        "max_requests_by_workload_class",
        "max_tool_calls_by_class",
    ):
        mapping = policy[field]
        if not isinstance(mapping, dict) or not mapping:
            raise ValueError(f"{field} must be a non-empty mapping")
        for key, value in mapping.items():
            if not isinstance(key, str) or not key:
                raise ValueError(f"{field} contains invalid key")
            positive_int(value, f"{field}.{key}")
    return policy


def load_window(events_path, receipt_path, signature, public_key, policy):
    receipt_raw = verify_blob(receipt_path, signature, public_key)
    receipt = _aidefend_strict_json(receipt_raw)
    events_raw = _aidefend_bytes(Path(events_path))
    if not isinstance(receipt, dict) or set(receipt) != RECEIPT_FIELDS:
        raise ValueError("collection receipt schema mismatch")
    if receipt["schema_version"] != "aidefend.cost_collection.v1":
        raise ValueError("unsupported collection receipt")
    if receipt["events_sha256"] != sha256_file(events_path):
        raise ValueError("receipt/events digest mismatch")
    start, end = parse_time(receipt["window_start"]), parse_time(receipt["window_end"])
    if int((end - start).total_seconds()) != policy["window_seconds"]:
        raise ValueError("window duration differs from signed policy")
    complete = (
        receipt["collection_complete"] is True
        and isinstance(receipt["expected_sources"], list)
        and receipt["expected_sources"] == receipt["observed_sources"]
    )
    events, seen = [], set()
    for event in _aidefend_strict_jsonl(events_raw):
            required = {
                "schema_version", "event_id", "occurred_at", "request_id",
                "principal_ref", "session_ref", "tenant_ref", "model_id",
                "model_class", "workload_class", "outcome", "cost_units",
                "input_tokens", "output_tokens", "tool_class_counts",
                "parameter_digest", "authn_context_sha256",
                "telemetry_policy_version", "telemetry_policy_sha256",
                "cost_method_version",
            }
            if not isinstance(event, dict) or set(event) != required:
                raise ValueError("cost event schema mismatch")
            if event["event_id"] in seen:
                raise ValueError("duplicate cost event ID")
            seen.add(event["event_id"])
            occurred = parse_time(event["occurred_at"])
            if not start <= occurred < end:
                raise ValueError("cost event outside declared window")
            for field in ("cost_units", "input_tokens", "output_tokens"):
                value = event[field]
                if isinstance(value, bool) or not isinstance(value, int) or value < 0:
                    raise ValueError(f"{field} invalid")
            if event["telemetry_policy_version"] != policy["policy_version"]:
                raise ValueError("event/policy version mismatch")
            events.append(event)
    return events, receipt, complete


def analyze(
    policy_path,
    policy_signature,
    policy_public_key,
    current_events,
    current_receipt,
    current_receipt_signature,
    previous_events,
    previous_receipt,
    previous_receipt_signature,
    receipt_public_key,
):
    policy = load_policy(policy_path, policy_signature, policy_public_key)
    current, current_meta, current_complete = load_window(
        current_events,
        current_receipt,
        current_receipt_signature,
        receipt_public_key,
        policy,
    )
    previous, previous_meta, previous_complete = load_window(
        previous_events,
        previous_receipt,
        previous_receipt_signature,
        receipt_public_key,
        policy,
    )
    minimum = policy["min_events_per_window"]
    previous_cost = sum(event["cost_units"] for event in previous)
    if (
        not current_complete or not previous_complete
        or len(current) < minimum or len(previous) < minimum
        or previous_cost < policy["min_previous_cost_units"]
    ):
        return {
            "control_status": "INSUFFICIENT_DATA",
            "finding_status": "UNKNOWN",
            "reason": "current/history coverage below signed policy",
            "current_event_count": len(current),
            "previous_event_count": len(previous),
            "previous_cost_units": previous_cost,
        }

    cost_by_workload = Counter()
    requests_by_workload = Counter()
    tool_calls_by_class = Counter()
    for event in current:
        workload = event["workload_class"]
        if workload not in policy["max_cost_units_by_workload_class"]:
            raise ValueError("workload class missing signed cost policy")
        if workload not in policy["max_requests_by_workload_class"]:
            raise ValueError("workload class missing signed request policy")
        cost_by_workload[workload] += event["cost_units"]
        requests_by_workload[workload] += 1
        for tool_class, count in event["tool_class_counts"].items():
            if (
                tool_class not in policy["max_tool_calls_by_class"]
                or isinstance(count, bool) or not isinstance(count, int)
                or count < 0
            ):
                raise ValueError("tool-class evidence invalid or uncovered")
            tool_calls_by_class[tool_class] += count

    current_cost = sum(cost_by_workload.values())
    rate_ratio = current_cost / previous_cost
    findings = []
    for key, value in cost_by_workload.items():
        if value > policy["max_cost_units_by_workload_class"][key]:
            findings.append({"type": "cost_limit", "class": key, "value": value})
    for key, value in requests_by_workload.items():
        if value > policy["max_requests_by_workload_class"][key]:
            findings.append({"type": "request_limit", "class": key, "value": value})
    for key, value in tool_calls_by_class.items():
        if value > policy["max_tool_calls_by_class"][key]:
            findings.append({"type": "tool_call_limit", "class": key, "value": value})
    if rate_ratio > policy["max_cost_rate_ratio"]:
        findings.append({"type": "cost_rate_change", "value": rate_ratio})

    return {
        "schema_version": "aidefend.cost_window_result.v1",
        "control_status": "PASS",
        "finding_status": "FINDING" if findings else "NO_FINDING",
        "policy_version": policy["policy_version"],
        "policy_sha256": sha256_file(policy_path),
        "current_events_sha256": sha256_file(current_events),
        "previous_events_sha256": sha256_file(previous_events),
        "current_event_count": len(current),
        "previous_event_count": len(previous),
        "current_cost_units": current_cost,
        "previous_cost_units": previous_cost,
        "cost_rate_ratio": rate_ratio,
        "cost_by_workload": dict(cost_by_workload),
        "requests_by_workload": dict(requests_by_workload),
        "tool_calls_by_class": dict(tool_calls_by_class),
        "findings": findings,
    }


def safe_analyze(*args, **kwargs):
    try:
        return analyze(*args, **kwargs)
    except Exception as exc:
        return {
            "control_status": "ERROR",
            "finding_status": "UNKNOWN",
            "reason": "policy, event, history, or coverage evidence invalid",
            "error_class": type(exc).__name__,
        }

Evidence: retain the signed analysis policy, both exact event windows, signed completeness receipts, source coverage, aggregates, result, and exception class. Independently replay from raw events. NO_THRESHOLD_BREACH requires complete current and historical coverage; it is never synthesized from an empty or undersized baseline.

AID-D-005.007-G003

Strict per-model parameter detector

from __future__ import annotations

# File: cost_monitor/parameter_detector.py

import hashlib
import hmac
import json
import math
import os
import re
import subprocess
import tempfile
from pathlib import Path


POLICY_FIELDS = {"schema_version", "policy_version", "models"}
MODEL_FIELDS = {"applicability", "reason", "parameter_rules"}
BASE_RULE_FIELDS = {"type", "required"}
NUMERIC_RULE_FIELDS = BASE_RULE_FIELDS | {"minimum", "maximum"}
ENUM_RULE_FIELDS = BASE_RULE_FIELDS | {"allowed_values"}
BOOLEAN_RULE_FIELDS = BASE_RULE_FIELDS


SIGNATURE_VERIFY_TIMEOUT_SECONDS = float(
    os.environ["D005007_SIGNATURE_VERIFY_TIMEOUT_SECONDS"]
)
if not math.isfinite(SIGNATURE_VERIFY_TIMEOUT_SECONDS) or SIGNATURE_VERIFY_TIMEOUT_SECONDS <= 0:
    raise RuntimeError("signature verification timeout must be finite and positive")


_AIDEFEND_CAPTURED: dict[Path, bytes] = {}


def _aidefend_bytes(path: Path) -> bytes:
    normalized = Path(path)
    if normalized not in _AIDEFEND_CAPTURED:
        _AIDEFEND_CAPTURED[normalized] = normalized.read_bytes()
    return _AIDEFEND_CAPTURED[normalized]


def _aidefend_write_private(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 _aidefend_strict_json(raw: bytes) -> object:
    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(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=lambda token: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + token)
        ),
    )


def _aidefend_strict_jsonl(raw: bytes) -> list[object]:
    lines = raw.decode("utf-8", errors="strict").splitlines()
    if not lines or any(not line.strip() for line in lines):
        raise ValueError("JSONL input is empty or contains a blank record")
    return [_aidefend_strict_json(line.encode("utf-8")) for line in lines]


def _aidefend_capture_signed_bytes(
    payload_path: Path,
    proof_path: Path,
    key_reference: str | Path,
    timeout_seconds: float,
) -> bytes:
    payload_path = Path(payload_path)
    proof_path = Path(proof_path)
    if payload_path == proof_path:
        raise RuntimeError("signed payload and signature paths collide")
    payload_raw = payload_path.read_bytes()
    proof_raw = proof_path.read_bytes()
    key_text = str(key_reference)
    remote_key = bool(re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*://", key_text
    )) or key_text.startswith("pkcs11:")
    key_raw = None if remote_key else (
        key_text.encode("utf-8")
        if key_text.startswith("-----BEGIN ")
        else Path(key_text).read_bytes()
    )
    with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        proof_snapshot = root / "payload.sig"
        _aidefend_write_private(payload_snapshot, payload_raw)
        _aidefend_write_private(proof_snapshot, proof_raw)
        captured = [(payload_snapshot, payload_raw), (proof_snapshot, proof_raw)]
        key_argument = key_text
        if key_raw is not None:
            key_snapshot = root / "verify-key.pub"
            _aidefend_write_private(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(proof_snapshot), str(payload_snapshot)],
                check=True, capture_output=True, text=True,
                timeout=timeout_seconds,
            )
            if any(
                not hmac.compare_digest(snapshot.read_bytes(), expected)
                for snapshot, expected in captured
            ):
                raise RuntimeError("private signature-verification snapshot changed")
        finally:
            for snapshot, _ in captured:
                os.chmod(snapshot, 0o600)
    _AIDEFEND_CAPTURED[payload_path] = payload_raw
    return payload_raw



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


def verify_blob(path: Path, signature: Path, public_key: Path) -> bytes:
    return _aidefend_capture_signed_bytes(
        path, signature, public_key, SIGNATURE_VERIFY_TIMEOUT_SECONDS,
    )


def finite(value, field):
    if isinstance(value, bool) or not isinstance(value, (int, float)):
        raise ValueError(f"{field} must be numeric")
    number = float(value)
    if not math.isfinite(number):
        raise ValueError(f"{field} must be finite")
    return number


def validate_rule(model_id: str, name: str, rule: dict) -> None:
    if not isinstance(rule, dict) or "type" not in rule or "required" not in rule:
        raise ValueError("parameter rule schema invalid")
    if rule["type"] == "number":
        expected = NUMERIC_RULE_FIELDS
    elif rule["type"] == "integer":
        expected = NUMERIC_RULE_FIELDS
    elif rule["type"] == "enum":
        expected = ENUM_RULE_FIELDS
    elif rule["type"] == "boolean":
        expected = BOOLEAN_RULE_FIELDS
    else:
        raise ValueError("unsupported parameter rule type")
    if set(rule) != expected or not isinstance(rule["required"], bool):
        raise ValueError("parameter rule fields invalid")
    if rule["type"] in {"number", "integer"}:
        minimum = finite(rule["minimum"], f"{model_id}.{name}.minimum")
        maximum = finite(rule["maximum"], f"{model_id}.{name}.maximum")
        if maximum < minimum:
            raise ValueError("parameter rule bounds reversed")
        if rule["type"] == "integer" and (not float(minimum).is_integer() or not float(maximum).is_integer()):
            raise ValueError("integer rule bounds must be integers")
    if rule["type"] == "enum":
        allowed = rule["allowed_values"]
        if not isinstance(allowed, list) or not allowed or len(allowed) != len(set(map(str, allowed))):
            raise ValueError("enum allowed_values invalid")


def load_policy(path: Path, signature: Path, public_key: Path) -> tuple[dict, str]:
    raw = verify_blob(path, signature, public_key)
    policy = _aidefend_strict_json(raw)
    if not isinstance(policy, dict) or set(policy) != POLICY_FIELDS:
        raise ValueError("parameter policy schema mismatch")
    if policy["schema_version"] != "aidefend.parameter_policy.v2":
        raise ValueError("unsupported parameter policy")
    if not isinstance(policy["policy_version"], str) or not policy["policy_version"]:
        raise ValueError("missing policy version")
    if not isinstance(policy["models"], dict) or not policy["models"]:
        raise ValueError("model policy map is empty")
    for model_id, model in policy["models"].items():
        if not isinstance(model_id, str) or not model_id:
            raise ValueError("model ID invalid")
        if not isinstance(model, dict) or set(model) != MODEL_FIELDS:
            raise ValueError("model policy schema mismatch")
        if model["applicability"] not in {"IN_SCOPE", "OUT_OF_SCOPE"}:
            raise ValueError("invalid model applicability")
        if not isinstance(model["reason"], str) or not model["reason"]:
            raise ValueError("model applicability reason missing")
        rules = model["parameter_rules"]
        if model["applicability"] == "OUT_OF_SCOPE":
            if rules != {}:
                raise ValueError("out-of-scope model cannot define rules")
            continue
        if not isinstance(rules, dict) or not rules:
            raise ValueError("in-scope model has no parameter rules")
        for name, rule in rules.items():
            if not isinstance(name, str) or not name:
                raise ValueError("parameter name invalid")
            validate_rule(model_id, name, rule)
    return policy, canonical_sha256(policy)


def check_value(value, rule: dict) -> str:
    if rule["type"] == "number":
        number = finite(value, "parameter")
        return "OK" if rule["minimum"] <= number <= rule["maximum"] else "FINDING"
    if rule["type"] == "integer":
        if isinstance(value, bool) or not isinstance(value, int):
            return "FINDING"
        return "OK" if rule["minimum"] <= value <= rule["maximum"] else "FINDING"
    if rule["type"] == "enum":
        return "OK" if value in rule["allowed_values"] else "FINDING"
    if rule["type"] == "boolean":
        return "OK" if isinstance(value, bool) else "FINDING"
    raise ValueError("unsupported rule type")


def evaluate_request(policy: dict, policy_sha256: str, model_id: str, parameters: dict) -> dict:
    if model_id not in policy["models"]:
        return {"control_status": "ERROR", "finding_status": "UNKNOWN",
                "reason": "model absent from signed parameter policy"}
    model = policy["models"][model_id]
    if model["applicability"] == "OUT_OF_SCOPE":
        return {"control_status": "NOT_APPLICABLE",
                "finding_status": "NOT_EVALUATED", "reason": model["reason"]}
    rules = model["parameter_rules"]
    if not isinstance(parameters, dict):
        return {"control_status": "ERROR", "finding_status": "UNKNOWN",
                "reason": "parameters must be an object"}
    unknown = sorted(set(parameters) - set(rules))
    if unknown:
        return {"control_status": "ERROR", "finding_status": "UNKNOWN",
                "reason": "unknown request-control parameter",
                "parameters": unknown}
    missing = sorted(name for name, rule in rules.items() if rule["required"] and name not in parameters)
    if missing:
        return {"control_status": "INSUFFICIENT_DATA",
                "finding_status": "UNKNOWN", "missing_required": missing}
    anomalies = []
    for name, value in parameters.items():
        outcome = check_value(value, rules[name])
        if outcome != "OK":
            anomalies.append({"parameter": name, "value": value, "rule": rules[name]})
    return {
        "control_status": "PASS",
        "finding_status": "FINDING" if anomalies else "NO_FINDING",
        "policy_version": policy["policy_version"],
        "policy_sha256": policy_sha256,
        "model_id": model_id,
        "anomalies": anomalies,
    }

Example policy fragment

{
  "schema_version": "aidefend.parameter_policy.v2",
  "policy_version": "2026-07-11",
  "models": {
    "gpt-image-1": {
      "applicability": "IN_SCOPE",
      "reason": "provider image generation request controls",
      "parameter_rules": {
        "quality": {"type": "enum", "required": false, "allowed_values": ["auto", "low", "medium", "high"]},
        "size": {"type": "enum", "required": false, "allowed_values": ["auto", "1024x1024", "1024x1536", "1536x1024"]},
        "output_format": {"type": "enum", "required": false, "allowed_values": ["png", "jpeg", "webp"]},
        "partial_images": {"type": "integer", "required": false, "minimum": 0, "maximum": 3},
        "stream": {"type": "boolean", "required": false}
      }
    },
    "self-hosted-stable-diffusion-xl": {
      "applicability": "IN_SCOPE",
      "reason": "self-hosted diffusion runtime request controls",
      "parameter_rules": {
        "guidance_scale": {"type": "number", "required": false, "minimum": 0.0, "maximum": 12.0},
        "num_inference_steps": {"type": "integer", "required": false, "minimum": 1, "maximum": 80}
      }
    }
  }
}

Evidence: retain the signed policy, signature verification result, model ID, normalized parameter object, policy digest, and detector result. Regression-test unknown model, GPT image enum violations, diffusion numeric violations, NaN/infinity, missing required fields, and explicit out-of-scope models.

Verify safely

An independent verifier must parse immutable request bytes independently, resolve effective defaults from the pinned provider profile, and replay in-range, out-of-range, omitted, duplicate, malformed, unsupported and baseline-breaking fixtures, read evidence back, verify signatures, digests, and immutability, and compare the canonical result digest.

AID-D-005.007-G004

Detect-side correlation boundary

This step observes whether an independently owned response system produced the expected authenticated receipt; it never throttles, quarantines, downgrades, or approves an action. A signed environment policy maps each finding type to the expected response action and SLA. Correlate signed finding and response events by exact incident ID, finding digest, and scope. Caller-supplied IDs/timestamps and ?any response exists? checks are not evidence.

Authenticated scope/action/SLA correlation

from __future__ import annotations

# File: cost_monitor/response_correlation.py

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

import jwt

POLICY_FIELDS = {
    "schema_version", "policy_version", "finding_issuer",
    "response_issuer", "finding_audience", "response_audience",
    "expectations",
}
EXPECTATION_FIELDS = {"expected_action", "sla_seconds"}
SIGNATURE_VERIFY_TIMEOUT_SECONDS = float(os.environ["D005007_SIGNATURE_VERIFY_TIMEOUT_SECONDS"])
if not math.isfinite(SIGNATURE_VERIFY_TIMEOUT_SECONDS) or SIGNATURE_VERIFY_TIMEOUT_SECONDS <= 0:
    raise RuntimeError("signed runtime-profile verification timeout is invalid")


_AIDEFEND_CAPTURED: dict[Path, bytes] = {}


def _aidefend_bytes(path: Path) -> bytes:
    normalized = Path(path)
    if normalized not in _AIDEFEND_CAPTURED:
        _AIDEFEND_CAPTURED[normalized] = normalized.read_bytes()
    return _AIDEFEND_CAPTURED[normalized]


def _aidefend_write_private(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 _aidefend_strict_json(raw: bytes) -> object:
    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(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=unique_object,
        parse_constant=lambda token: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + token)
        ),
    )


def _aidefend_strict_jsonl(raw: bytes) -> list[object]:
    lines = raw.decode("utf-8", errors="strict").splitlines()
    if not lines or any(not line.strip() for line in lines):
        raise ValueError("JSONL input is empty or contains a blank record")
    return [_aidefend_strict_json(line.encode("utf-8")) for line in lines]


def _aidefend_capture_signed_bytes(
    payload_path: Path,
    proof_path: Path,
    key_reference: str | Path,
    timeout_seconds: float,
) -> bytes:
    payload_path = Path(payload_path)
    proof_path = Path(proof_path)
    if payload_path == proof_path:
        raise RuntimeError("signed payload and signature paths collide")
    payload_raw = payload_path.read_bytes()
    proof_raw = proof_path.read_bytes()
    key_text = str(key_reference)
    remote_key = bool(re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*://", key_text
    )) or key_text.startswith("pkcs11:")
    key_raw = None if remote_key else (
        key_text.encode("utf-8")
        if key_text.startswith("-----BEGIN ")
        else Path(key_text).read_bytes()
    )
    with tempfile.TemporaryDirectory(prefix="aidefend-signed-input-") as directory:
        root = Path(directory)
        os.chmod(root, 0o700)
        payload_snapshot = root / "payload"
        proof_snapshot = root / "payload.sig"
        _aidefend_write_private(payload_snapshot, payload_raw)
        _aidefend_write_private(proof_snapshot, proof_raw)
        captured = [(payload_snapshot, payload_raw), (proof_snapshot, proof_raw)]
        key_argument = key_text
        if key_raw is not None:
            key_snapshot = root / "verify-key.pub"
            _aidefend_write_private(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(proof_snapshot), str(payload_snapshot)],
                check=True, capture_output=True, text=True,
                timeout=timeout_seconds,
            )
            if any(
                not hmac.compare_digest(snapshot.read_bytes(), expected)
                for snapshot, expected in captured
            ):
                raise RuntimeError("private signature-verification snapshot changed")
        finally:
            for snapshot, _ in captured:
                os.chmod(snapshot, 0o600)
    _AIDEFEND_CAPTURED[payload_path] = payload_raw
    return payload_raw


def parse_time(value):
    if not isinstance(value, str):
        raise ValueError("timestamp is not text")
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.utcoffset() is None:
        raise ValueError("timestamp lacks timezone")
    return parsed.astimezone(timezone.utc)


def verify_blob(path, signature, public_key):
    return _aidefend_capture_signed_bytes(
        Path(path), Path(signature), public_key,
        SIGNATURE_VERIFY_TIMEOUT_SECONDS,
    )


def load_policy(path, signature, public_key):
    raw = verify_blob(path, signature, public_key)
    policy = _aidefend_strict_json(raw)
    if not isinstance(policy, dict) or set(policy) != POLICY_FIELDS:
        raise ValueError("response-correlation policy schema mismatch")
    if policy["schema_version"] != "aidefend.response_correlation_policy.v1":
        raise ValueError("unsupported response-correlation policy")
    if not isinstance(policy["expectations"], dict) or not policy["expectations"]:
        raise ValueError("response expectation map is empty")
    for finding_type, expectation in policy["expectations"].items():
        if (
            not isinstance(finding_type, str) or not finding_type
            or not isinstance(expectation, dict)
            or set(expectation) != EXPECTATION_FIELDS
            or not isinstance(expectation["expected_action"], str)
            or not expectation["expected_action"]
            or isinstance(expectation["sla_seconds"], bool)
            or not isinstance(expectation["sla_seconds"], int)
            or expectation["sla_seconds"] < 1
        ):
            raise ValueError("response expectation invalid")
    return policy


def verify_event(token, key, issuer, audience, typ, required):
    header = jwt.get_unverified_header(token)
    if header.get("alg") != "EdDSA" or header.get("typ") != typ:
        raise ValueError("unexpected event envelope")
    return jwt.decode(
        token,
        key,
        algorithms=["EdDSA"],
        issuer=issuer,
        audience=audience,
        leeway=5,
        options={"require": [
            "iss", "aud", "iat", "nbf", "exp", "jti", *required
        ]},
    )


def correlate(
    finding_jws,
    response_jws,
    *,
    policy_path,
    policy_signature,
    policy_public_key,
    finding_public_key,
    response_public_key,
    collection_complete,
    now=None,
):
    policy = load_policy(
        policy_path, policy_signature, policy_public_key
    )
    if not isinstance(collection_complete, bool):
        raise ValueError("collection completeness is not authenticated boolean")
    finding = verify_event(
        finding_jws,
        finding_public_key,
        policy["finding_issuer"],
        policy["finding_audience"],
        "cost-finding+jwt",
        [
            "incident_id", "finding_type", "finding_sha256",
            "scope_ref", "occurred_at",
        ],
    )
    expectation = policy["expectations"].get(finding["finding_type"])
    if expectation is None:
        raise ValueError("finding type lacks signed response expectation")
    finding_time = parse_time(finding["occurred_at"])
    current = now or datetime.now(timezone.utc)
    deadline = finding_time.timestamp() + expectation["sla_seconds"]

    if response_jws is None:
        if not collection_complete or current.timestamp() <= deadline:
            return {
                "status": "INSUFFICIENT_DATA",
                "reason": "response window not complete",
                "incident_id": finding["incident_id"],
            }
        findings = [{"type": "missing_response_after_sla"}]
        response = None
    else:
        response = verify_event(
            response_jws,
            response_public_key,
            policy["response_issuer"],
            policy["response_audience"],
            "response-receipt+jwt",
            [
                "incident_id", "finding_sha256", "scope_ref", "action",
                "status", "occurred_at", "receipt_sha256",
            ],
        )
        response_time = parse_time(response["occurred_at"])
        findings = []
        expected_bindings = {
            "incident_id": finding["incident_id"],
            "finding_sha256": finding["finding_sha256"],
            "scope_ref": finding["scope_ref"],
            "action": expectation["expected_action"],
        }
        for field, expected in expected_bindings.items():
            if response.get(field) != expected:
                findings.append({
                    "type": f"{field}_mismatch",
                    "expected": expected,
                    "observed": response.get(field),
                })
        if response["status"] != "APPLIED":
            findings.append({
                "type": "response_not_applied",
                "observed": response["status"],
            })
        if response_time < finding_time:
            findings.append({"type": "response_before_finding"})
        elif response_time.timestamp() > deadline:
            findings.append({"type": "response_after_sla"})

    return {
        "schema_version": "aidefend.response_correlation_result.v1",
        "control_status": "PASS",
        "finding_status": "FINDING" if findings else "NO_FINDING",
        "policy_version": policy["policy_version"],
        "policy_sha256": hashlib.sha256(_AIDEFEND_CAPTURED[Path(policy_path)]).hexdigest(),
        "incident_id": finding["incident_id"],
        "finding_jti": finding["jti"],
        "finding_sha256": finding["finding_sha256"],
        "scope_ref": finding["scope_ref"],
        "expected_action": expectation["expected_action"],
        "sla_seconds": expectation["sla_seconds"],
        "response_jti": response["jti"] if response else None,
        "response_receipt_sha256": (
            response["receipt_sha256"] if response else None
        ),
        "findings": findings,
    }


def safe_correlate(*args, **kwargs):
    try:
        return correlate(*args, **kwargs)
    except Exception as exc:
        return {
            "control_status": "ERROR",
            "finding_status": "UNKNOWN",
            "reason": "policy, event authentication, or correlation invalid",
            "error_class": type(exc).__name__,
        }

Evidence: obtain collection completeness from a signed append-only ingestion receipt and retain both compact JWS events, policy/signature, exact bindings, SLA computation, result, and exception class. Independently replay signatures and correlation. Findings feed investigation/SIEM; response execution and its success criteria remain owned by Isolate/Evict controls.

Open-source tools

• OpenTelemetry
• Prometheus
• Grafana
• Fluent Bit
• Fluentd
• ClickHouse
• Langfuse (cost, token-volume, user, session, model, and trace telemetry for downstream spike rules)
• Helicone (LLM request, token, latency, and cost telemetry for downstream spike rules)

Commercial / hosted tools

• Datadog
• Splunk Enterprise Security
• New Relic
• Elastic Security

Threat mappings

MITRE ATLAS

• AML.T0034 Cost Harvesting
• AML.T0034.000 Cost Harvesting: Excessive Queries
• AML.T0034.001 Cost Harvesting: Resource-Intensive Queries
• AML.T0034.002 Cost Harvesting: Agentic Resource Consumption
• AML.T0029 Denial of AI Service
• AML.T0053 AI Agent Tool Invocation
• AML.T0040 AI Model Inference API Access
• AML.T0126 Automated Collection (abnormal tool-invocation volume and request fan-out reveal scripted collection)

MAESTRO

• Resource Hijacking (L4)
• Denial of Service on Framework APIs (L3)
• Agent Tool Misuse (L7)

OWASP LLM Top 10 2026

• LLM06:2026 Unbounded Consumption

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI02:2026 Tool Misuse and Exploitation
• ASI08:2026 Cascading Failures
• ASI10:2026 Rogue Agents

NIST Adversarial Machine Learning 2025

• NISTAML.014 Energy-latency

Cisco Integrated AI Security and Safety Framework

• AISubtech-12.1.1 Parameter Manipulation (model or tool parameters outside signed ranges)
• AISubtech-13.1.1 Compute Exhaustion
• AISubtech-13.1.2 Memory Flooding
• AISubtech-13.1.3 Model Denial of Service
• AISubtech-13.1.4 Application Denial of Service
• AISubtech-13.2.1 Service Misuse for Cost Inflation
• AITech-12.1 Tool Exploitation (model or tool parameters outside signed ranges)
• AITech-13.1 Disruption of Availability
• AITech-13.2 Cost Harvesting / Repurposing

Google Secure AI Framework 2.0 - Risks

• DMS: Denial of ML Service
• RA: Rogue Actions (cost spikes from rogue agent tool abuse)

Databricks AI Security Framework 3.0

• Model Serving - Inference requests 9.7: Denial of Service (DoS)
• Agents - Core 13.4: Resource Overload
• Agents - Core 13.2: Tool Misuse (recursive tool loops cause cost spikes)