Scope boundary

Owns foundational structured event instrumentation, secure collection, normalization, timestamping, forwarding, and tamper-evident retention for AI system activity. It creates the event record but does not define every specialized detector or prove an alert fired.

AID-D-005.002: AID-D-005.001 generates and transports normalized telemetry; AID-D-005.002 packages, tests, deploys, routes, and monitors the health of detection rules that consume that telemetry. A complete event pipeline is a prerequisite, not proof that the correct rule is deployed or evaluating successfully.

AID-D-005.004: AID-D-005.001 owns general AI event logging and collection; AID-D-005.004 owns agent-specific forensic session records such as goals, actions, tool bindings, policy decisions, and memory interactions. Agent session semantics extend rather than replace the foundational event pipeline.

AID-D-005.011: AID-D-005.001 continuously collects routine telemetry; AID-D-005.011 performs incident-triggered, deadline-bounded capture of otherwise volatile runtime state before containment. Ordinary logs may reference but cannot reconstruct every in-flight queue, pending action, or ephemeral process fact.

Implementation guidance

• AID-D-005.001-G001: Implement structured, context-rich logging for all AI interactions.
• AID-D-005.001-G002: Ensure the logging pipeline supports structured agentic event schemas and correlation IDs.
• AID-D-005.001-G003: Map AIDEFEND AI security events to OpenTelemetry GenAI-compatible spans while preserving security evidence fields as custom attributes.
• AID-D-005.001-G004: Use a dedicated log shipper for secure and reliable collection.
• AID-D-005.001-G005: Ensure logs are timestamped, immutable, and stored in a tamper-evident archive.
• AID-D-005.001-G006: Emit evidence-ready AI security events with stable identifiers, evidence references, retention labels, legal-hold flags, and chain-of-custody linkage.
• AID-D-005.001-G007: Instrument MCP request, task, subscription, budget, disable-check, and backend-effect lifecycles with privacy-bounded correlated events and reconcile terminal coverage.

AID-D-005.001-G001

Concept and content boundary

Generate one structured event per security-relevant interaction, but do not copy raw prompts, responses, tool arguments, or hidden reasoning into ordinary logs. Recursively redact nested objects, record authenticated identity from trusted middleware, and retain keyed digests plus short policy-approved previews where investigation requires content context. Keyed HMAC digests reduce offline guessing of low-entropy values compared with plain hashes.

Step 1: Build a recursive redactor and evidence-safe event emitter

from __future__ import annotations

# File: api/ai_event_logging.py

import hashlib
import hmac
import json
import logging
import os
import re
import time
import uuid
from collections.abc import Mapping
from typing import Any

from fastapi import HTTPException, Request


logger = logging.getLogger("ai_events")
SENSITIVE_KEY = re.compile(
    r"(authorization|cookie|password|passwd|secret|token|api[_-]?key|private[_-]?key)",
    re.IGNORECASE,
)
SECRET_VALUE = re.compile(r"(?i)(bearer\s+)[a-z0-9._~+/=-]+|\bsk-[a-z0-9_-]{12,}\b")
MAX_PREVIEW_CHARS = 512


def redact_nested(value: Any) -> Any:
    if isinstance(value, Mapping):
        return {
            str(key): (
                "<redacted>"
                if SENSITIVE_KEY.search(str(key))
                else redact_nested(child)
            )
            for key, child in value.items()
        }
    if isinstance(value, (list, tuple, set)):
        return [redact_nested(child) for child in value]
    if isinstance(value, str):
        return SECRET_VALUE.sub("<redacted>", value)
    if value is None or isinstance(value, (bool, int, float)):
        return value
    return SECRET_VALUE.sub("<redacted>", str(value))


def payload_digest(value: Any) -> str:
    key = os.environ["AIDEFEND_LOG_HMAC_KEY"].encode("utf-8")
    if len(key) < 32:
        raise RuntimeError("AIDEFEND_LOG_HMAC_KEY must contain at least 32 bytes")
    canonical = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str)
    return hmac.new(key, canonical.encode("utf-8"), hashlib.sha256).hexdigest()


def approved_preview(value: Any) -> str:
    redacted = redact_nested(value)
    return json.dumps(redacted, sort_keys=True, default=str)[:MAX_PREVIEW_CHARS]


def authenticated_principal(request: Request) -> tuple[str, str, str]:
    # OIDC/mTLS middleware must set both values after verification.
    principal_id = getattr(request.state, "principal_id", None)
    authentication_event_id = getattr(request.state, "authentication_event_id", None)
    if not principal_id or not authentication_event_id:
        raise HTTPException(status_code=401, detail="verified principal required")
    return str(principal_id), str(authentication_event_id)


def emit_ai_interaction(
    *,
    request: Request,
    trace_id: str,
    model_id: str,
    request_payload: Any,
    response_payload: Any,
    policy_id: str,
    policy_version: str,
    verifier_evidence_ref: str,
    latency_ms: int,
) -> None:
    principal_id, authentication_event_id = authenticated_principal(request)
    event = {
        "schema_version": "aidefend.ai_interaction.v2",
        "event_id": str(uuid.uuid4()),
        "observed_at_unix": time.time(),
        "event_type": "ai_interaction",
        "trace_id": trace_id,
        "principal_id": principal_id,
        "authentication_event_id": authentication_event_id,
        "model_id": model_id,
        "request_hmac_sha256": payload_digest(request_payload),
        "response_hmac_sha256": payload_digest(response_payload),
        "request_preview": approved_preview(request_payload),
        "response_preview": approved_preview(response_payload),
        "policy_id": policy_id,
        "policy_version": policy_version,
        "verifier_evidence_ref": verifier_evidence_ref,
        "latency_ms": latency_ms,
        "redaction_profile": "recursive-v2",
    }
    logger.info(json.dumps(event, sort_keys=True, separators=(",", ":")))

Step 2: Call the emitter after authentication and policy evaluation

Invoke emit_ai_interaction from the inference route or a streaming-aware ASGI hook after verified OIDC or mTLS middleware has populated the request state. Never derive principal_id, policy verdicts, or verifier evidence from request JSON or caller-controlled headers. Send the JSON event to an append-only store through the shipper below.

Verify the detector and retain evidence

On every release, send a nested fixture containing secrets in objects and arrays, then have an independent test re-read the stored event and fail if any fixture secret appears. The fresh JSON event is the runtime artifact; the signed redaction-test report and recomputation of HMACs in a separate verifier are the independent checks.

Action: Instrument each inference boundary with recursively redacted, authenticated, evidence-linked event emission. Keep raw content and hidden reasoning outside the ordinary telemetry path.

AID-D-005.001-G002

Concept

The general collection pipeline must preserve specialized agent events without inventing a raw-reasoning event type. Detailed agent/session instrumentation belongs to AID-D-005.004; this detector owns schema validation, routing, correlation, and lossless transport.

Required agent event contract

Accept event types such as agent_goal_accepted, agent_action_proposed, agent_policy_verdict, agent_tool_result, rag_retrieval, and hitl_intervention. Preserve event_id, session_id, trace_id, authenticated principal_id, authentication event, approved goal ID/digest, observable rationale summary, action name, keyed argument/result digests, policy/version, verifier ID/evidence reference, and timestamp. Reject ordinary events containing fields named thought, chain_of_thought, scratchpad, or raw_reasoning.

{
  "schema_version": "aidefend.agent_forensic.v2",
  "event_type": "agent_policy_verdict",
  "event_id": "0190f31a-1f3e-7c42-b5b2-5fdd3da76cd9",
  "session_id": "0190f31a-26a8-7a54-8c07-5f85ef4c7d43",
  "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736",
  "principal_id": "spiffe://aidefend.net/ns/prod/sa/payment-agent",
  "authentication_event_id": "0190f31a-2a6f-70f1-8f28-1d892c4b14b7",
  "approved_goal_id": "goal-42",
  "tool_name": "payments.create",
  "tool_args_hmac_sha256": "7cf7b7e310f1e5e221d4473e3ec68bd803f4ba402b7cc694da89b91cd4679f3f",
  "allowed": false,
  "policy_id": "agent-tools",
  "policy_version": "2026.07.1",
  "verifier_id": "policy-verifier-prod",
  "verifier_evidence_ref": "s3://security-evidence/verdicts/0190f31a-1f3e-7c42-b5b2-5fdd3da76cd9.json",
  "observed_at": "2026-07-10T12:00:00Z"
}

Verify the event boundary

Run a release test that submits each allowed event plus a prohibited raw-reasoning fixture. Confirm allowed events retain correlation and evidence fields end-to-end and the prohibited fixture is rejected and alerted. If raw reasoning is exceptionally required, route it only through the separately authorized break-glass path in AID-H-037.001, never this general pipeline.

Action: Version the agent-event schema, validate it at ingress, and monitor schema rejection counts. Store the signed schema-conformance report as the independent release artifact.

AID-D-005.001-G003

Concept and stability warning

The OpenTelemetry GenAI semantic conventions are marked Development and can evolve between releases. Pin the semantic-conventions version used by your instrumentation, test exported attribute names during upgrades, and keep AIDEFEND evidence fields such as detector verdict, policy ID, verifier evidence reference, redaction status, and HITL checkpoint under an aidefend.* namespace. Use current standard attributes for interoperability without treating the evolving convention as a stable security evidence contract.

Step 1: Configure OTLP tracing once per service

from __future__ import annotations

# File: telemetry/genai_tracing.py

import os
from collections.abc import Mapping

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor


def configure_tracing(service_name: str = "ai-inference-gateway") -> None:
    provider = TracerProvider(
        resource=Resource.create(
            {
                "service.name": service_name,
                "deployment.environment": os.getenv("ENVIRONMENT", "production"),
            }
        )
    )
    exporter = OTLPSpanExporter(
        endpoint=os.environ["OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"],
        headers={"authorization": "Bearer " + os.environ["OTEL_EXPORTER_OTLP_TOKEN"]},
    )
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)


tracer = trace.get_tracer("aidefend.genai")


def set_flat_attributes(span, attrs: Mapping[str, object]) -> None:
    for key, value in attrs.items():
        if value is None:
            continue
        if isinstance(value, (str, bool, int, float)):
            span.set_attribute(key, value)
        elif isinstance(value, list) and all(isinstance(item, str) for item in value):
            span.set_attribute(key, value)
        else:
            span.set_attribute(key, str(value))

Step 2: Emit an agent invocation span with gen_ai.* and aidefend.* attributes

from __future__ import annotations

# File: telemetry/instrumented_gateway.py

from telemetry.genai_tracing import set_flat_attributes, tracer


def record_agent_invocation(
    *,
    tenant_id: str,
    user_id: str,
    session_id: str,
    agent_id: str,
    provider: str,
    model_id: str,
    input_tokens: int,
    output_tokens: int,
    detector_verdict: str,
    evidence_ref: str,
    redaction_applied: bool,
    run_callable,
):
    with tracer.start_as_current_span(f"invoke_agent {agent_id}") as span:
        set_flat_attributes(
            span,
            {
                "gen_ai.operation.name": "invoke_agent",
                "gen_ai.provider.name": provider,
                "gen_ai.agent.name": agent_id,
                "gen_ai.request.model": model_id,
                "gen_ai.response.model": model_id,
                "gen_ai.usage.input_tokens": input_tokens,
                "gen_ai.usage.output_tokens": output_tokens,
                "user.id": user_id,
                "aidefend.tenant_id": tenant_id,
                "aidefend.session_id": session_id,
                "aidefend.agent_id": agent_id,
                "aidefend.detector.verdict": detector_verdict,
                "aidefend.evidence.ref": evidence_ref,
                "aidefend.redaction.applied": redaction_applied,
            },
        )
        try:
            result = run_callable()
        except Exception as exc:
            span.record_exception(exc)
            span.set_attribute("aidefend.error.type", exc.__class__.__name__)
            span.set_attribute("aidefend.security_outcome", "failed")
            raise
        span.set_attribute("aidefend.security_outcome", "completed")
        return result

Step 3: Keep raw content out of traces by default

Trace attributes are widely replicated across collectors, vendors, and debugging views. Store prompts, responses, retrieved chunks, and tool outputs only after redaction and only when policy allows it. Prefer hashes, byte counts, labels, evidence object references, and short sanitized previews.

Action: Add an OTel GenAI mapping layer to the general AI logging pipeline. Validate that traces show invoke_agent and execute_tool spans in your collector while the AIDEFEND-specific investigation fields remain searchable.

Verify safely

An independent verifier must rebuild spans from immutable source events with an independently pinned mapper, then compare trace and span IDs, links, canonical GenAI attributes, custom security fields, redaction, and counts, read evidence back, verify signatures, digests, and immutability, and compare the canonical result digest.

AID-D-005.001-G004

Configure durable, acknowledged transport

Vector's file source checkpoints offsets in data_dir; for newly discovered files, read_from = "beginning" prevents a first-deployment backlog from being silently skipped. The Firehose sink supports end-to-end acknowledgements and at-least-once delivery, while its default sink buffer is memory. Use a disk buffer with blocking backpressure for security logs, route parse failures to a separate durable path, and include a stable event ID so destination reconciliation can tolerate duplicates. Disk-buffer capacity is an environment policy value and has no framework default.

# File: /etc/vector/vector.toml.in

data_dir = "/var/lib/vector"

[sources.ai_app_logs]
type = "file"
include = ["/var/log/my_ai_app/events.log"]
read_from = "beginning"
data_dir = "/var/lib/vector/checkpoints"
offset_key = "source_offset"
max_line_bytes = 1048576

[transforms.parse_logs]
type = "remap"
inputs = ["ai_app_logs"]
drop_on_error = true
reroute_dropped = true
source = '''
. = parse_json!(string!(.message))
assert!(exists(.event_id), message: "event_id is required")
assert!(exists(.schema_version), message: "schema_version is required")
.ingest_source = "ai_app_logs"
'''

[sinks.kinesis_firehose]
type = "aws_kinesis_firehose"
inputs = ["parse_logs"]
stream_name = "ai-security-events-prod"
region = "us-east-1"
encoding.codec = "json"
compression = "gzip"
acknowledgements.enabled = true

[sinks.kinesis_firehose.buffer]
type = "disk"
max_size = __AIDEFEND_DISK_BUFFER_MAX_SIZE_BYTES__
when_full = "block"

[sinks.rejected_events]
type = "file"
inputs = ["parse_logs.dropped"]
path = "/var/lib/vector/rejected/ai-events-%Y-%m-%d.log"
encoding.codec = "json"

Verify the deployment profile and render the effective config

The deployment adapter must deliver the signed transport profile, its Sigstore bundle, and the trusted verification key. The renderer snapshots and verifies those bytes before reading them, accepts no capacity default, requires a positive integer, substitutes exactly one token, and validates the completed Vector configuration before installation.

# File: /usr/local/libexec/render-vector-config
#!/usr/bin/env bash
set -euo pipefail
umask 077

: "${VECTOR_TRANSPORT_PROFILE:?signed transport profile is required}"
: "${VECTOR_TRANSPORT_PROFILE_BUNDLE:?transport-profile bundle is required}"
: "${VECTOR_TRANSPORT_PROFILE_VERIFY_KEY:?transport-profile verification key is required}"

root="$(mktemp -d)"
trap 'rm -rf -- "$root"' EXIT
install -m 0400 -- "$VECTOR_TRANSPORT_PROFILE" "$root/profile.json"
install -m 0400 -- "$VECTOR_TRANSPORT_PROFILE_BUNDLE" "$root/profile.sigstore.json"
cosign verify-blob --key "$VECTOR_TRANSPORT_PROFILE_VERIFY_KEY" \
  --bundle "$root/profile.sigstore.json" "$root/profile.json" >/dev/null

capacity="$(jq -er '
  if type == "object"
     and (keys | sort) == (["deployment_id","disk_buffer_max_size_bytes","profile_version","schema_version"] | sort)
     and .schema_version == "aidefend.vector-transport-profile.v1"
     and (.profile_version | type == "string" and length > 0)
     and (.deployment_id | type == "string" and length > 0)
     and (.disk_buffer_max_size_bytes | type == "number" and floor == . and . > 0)
  then .disk_buffer_max_size_bytes | tostring
  else error("transport profile or positive disk-buffer capacity is invalid")
  end
' "$root/profile.json")"
[[ "$capacity" =~ ^[1-9][0-9]*$ ]] || {
  echo "disk-buffer capacity must render as a positive decimal integer" >&2
  exit 1
}

template="/etc/vector/vector.toml.in"
[[ -f "$template" && ! -L "$template" ]] || {
  echo "Vector configuration template is missing or not a regular file" >&2
  exit 1
}
placeholder="__AIDEFEND_DISK_BUFFER_MAX_SIZE_BYTES__"
placeholder_count="$(grep -oF -- "$placeholder" "$template" | wc -l | tr -d '[:space:]')"
[[ "$placeholder_count" == "1" ]] || {
  echo "Vector capacity token must occur exactly once" >&2
  exit 1
}
sed "s/$placeholder/$capacity/" "$template" > "$root/vector.toml"
if grep -F -- "$placeholder" "$root/vector.toml" >/dev/null; then
  echo "Vector configuration contains an unresolved capacity token" >&2
  exit 1
fi
vector validate "$root/vector.toml"
install -m 0640 -- "$root/vector.toml" /etc/vector/vector.toml
cmp -s -- "$root/vector.toml" /etc/vector/vector.toml
vector validate /etc/vector/vector.toml

Validate and monitor the environment profile

Treat the stream name, region, paths, maximum line size, disk capacity, and backpressure choice as fields in the same signature-verified deployment profile; the external configuration-distribution adapter may expand the renderer for those fields without changing this fail-closed capacity contract. Use the workload's AWS credential provider rather than static keys, run vector test /etc/vector/vector.toml in CI, and alert on source checkpoint errors, transform errors, dropped/rejected events, disk-buffer utilization, Firehose request failures, retries, and acknowledgement lag.

Prove end-to-end coverage

Inject unique signed canary event IDs before and after startup, rotation, Vector restart, downstream outage, poison JSON, and disk-pressure tests. Query Firehose's destination under an independent identity and reconcile every expected event ID, allowing documented at-least-once duplicates but no missing IDs. Read the checkpoint, rejected-event path, buffer metrics, destination object or record, and acknowledgement evidence back; do not claim lossless transport from a syntactically valid config alone.

AID-D-005.001-G005

Render retention from signed policy

S3 Object Lock applies to object versions and requires bucket versioning. The deployment pipeline verifies the organization's retention/legal-policy artifact, then renders its approved bucket name, KMS key, tags, and positive retention duration into Terraform variables. The framework does not publish a universal number of days.

Create a versioned Compliance-mode archive

# File: infrastructure/secure_log_storage.tf

variable "archive_bucket_name" {
  type = string
  validation {
    condition     = length(var.archive_bucket_name) >= 3 && length(var.archive_bucket_name) <= 63
    error_message = "archive_bucket_name must be a valid resolved S3 bucket name"
  }
}

variable "retention_days" {
  type = number
  validation {
    condition     = var.retention_days >= 1 && floor(var.retention_days) == var.retention_days
    error_message = "retention_days must be a positive integer from signed policy"
  }
}

variable "kms_key_arn" {
  type = string
  validation {
    condition     = startswith(var.kms_key_arn, "arn:aws:kms:")
    error_message = "kms_key_arn must identify the approved KMS key"
  }
}

variable "policy_version" {
  type = string
}

resource "aws_s3_bucket" "secure_log_archive" {
  bucket              = var.archive_bucket_name
  object_lock_enabled = true
  force_destroy       = false

  tags = {
    DataClass      = "ai-security-evidence"
    RetentionPolicy = var.policy_version
  }
}

resource "aws_s3_bucket_versioning" "secure_log_archive" {
  bucket = aws_s3_bucket.secure_log_archive.id
  versioning_configuration {
    status = "Enabled"
  }
}

resource "aws_s3_bucket_server_side_encryption_configuration" "secure_log_archive" {
  bucket = aws_s3_bucket.secure_log_archive.id
  rule {
    apply_server_side_encryption_by_default {
      sse_algorithm     = "aws:kms"
      kms_master_key_id = var.kms_key_arn
    }
    bucket_key_enabled = true
  }
}

resource "aws_s3_bucket_public_access_block" "secure_log_archive" {
  bucket                  = aws_s3_bucket.secure_log_archive.id
  block_public_acls       = true
  block_public_policy     = true
  ignore_public_acls      = true
  restrict_public_buckets = true
}

resource "aws_s3_bucket_object_lock_configuration" "secure_log_archive" {
  depends_on = [aws_s3_bucket_versioning.secure_log_archive]
  bucket     = aws_s3_bucket.secure_log_archive.id

  rule {
    default_retention {
      mode = "COMPLIANCE"
      days = var.retention_days
    }
  }
}

output "archive_binding" {
  value = {
    bucket         = aws_s3_bucket.secure_log_archive.id
    bucket_arn     = aws_s3_bucket.secure_log_archive.arn
    policy_version = var.policy_version
    retention_days = var.retention_days
  }
}

Plan and verify before accepting evidence

After verifying the signed policy and generating a concrete retention.auto.tfvars.json, run terraform fmt -check, terraform validate, and review a saved terraform plan -out=secure-log-storage.tfplan before applying it through the infrastructure identity. Under a separate read-only verifier identity, call aws s3api get-bucket-versioning, aws s3api get-object-lock-configuration, and aws s3api get-bucket-encryption. Upload a signed canary, record its version ID, read its retention metadata and bytes back, and verify that permanent deletion of that protected version is denied. A simple delete without a version ID may create a delete marker, so it is not the immutability test.

AID-D-005.001-G006

Concept:

Basic logs are not always investigation-ready. An evidence-ready AI security event must carry stable identifiers that join model calls, agent sessions, tool calls, retrieved chunks, prompts, responses, and evidence objects without relying on fragile text search. This step is technical evidence schema design; legal reporting decisions remain outside this technique.

Step 1: Define a strict event contract

from __future__ import annotations

# File: ai_logging/ai_security_event_schema.py

import hmac as _aidefend_policy_hmac
import json as _aidefend_policy_json
import math as _aidefend_policy_math
import os as _aidefend_policy_os
import re as _aidefend_policy_re
import subprocess as _aidefend_policy_subprocess
import tempfile as _aidefend_policy_tempfile
from pathlib import Path as _AidefendPolicyPath


def _aidefend_write_private(path, payload):
    descriptor = _aidefend_policy_os.open(
        path,
        _aidefend_policy_os.O_WRONLY
        | _aidefend_policy_os.O_CREAT
        | _aidefend_policy_os.O_EXCL,
        0o600,
    )
    with _aidefend_policy_os.fdopen(descriptor, "wb") as stream:
        stream.write(payload)
        stream.flush()
        _aidefend_policy_os.fsync(stream.fileno())
    _aidefend_policy_os.chmod(path, 0o400)


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

    return _aidefend_policy_json.loads(
        raw.decode("utf-8", errors="strict"),
        object_pairs_hook=reject_duplicate_keys,
        parse_constant=lambda value: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + value)
        ),
    )


def _aidefend_load_signed_policy(*, path_env, signature_env, key_env, schema_version):
    policy_path = _AidefendPolicyPath(_aidefend_policy_os.environ[path_env])
    signature_path = _AidefendPolicyPath(_aidefend_policy_os.environ[signature_env])
    if policy_path == signature_path:
        raise RuntimeError("policy and signature paths must differ")
    policy_raw = policy_path.read_bytes()
    signature_raw = signature_path.read_bytes()
    key_reference = _aidefend_policy_os.environ[key_env]
    remote_key = bool(_aidefend_policy_re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*:(?://)?", key_reference
    )) and not key_reference.startswith("-----BEGIN ")
    key_raw = None if remote_key else (
        key_reference.encode("utf-8")
        if key_reference.startswith("-----BEGIN ")
        else _AidefendPolicyPath(key_reference).read_bytes()
    )
    verify_timeout = float(_aidefend_policy_os.environ["AIDEFEND_POLICY_VERIFY_TIMEOUT_SECONDS"])
    if not _aidefend_policy_math.isfinite(verify_timeout) or verify_timeout <= 0:
        raise RuntimeError("policy-verification timeout from the versioned runtime profile is invalid")
    with _aidefend_policy_tempfile.TemporaryDirectory(
        prefix="aidefend-signed-policy-"
    ) as directory:
        root = _AidefendPolicyPath(directory)
        _aidefend_policy_os.chmod(root, 0o700)
        policy_snapshot = root / "policy.json"
        signature_snapshot = root / "policy.sig"
        _aidefend_write_private(policy_snapshot, policy_raw)
        _aidefend_write_private(signature_snapshot, signature_raw)
        captured = [(policy_snapshot, policy_raw), (signature_snapshot, signature_raw)]
        key_argument = key_reference
        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:
            _aidefend_policy_subprocess.run(
                [
                    "cosign", "verify-blob", "--key", key_argument,
                    "--bundle", str(signature_snapshot), str(policy_snapshot),
                ],
                check=True, capture_output=True, text=True, timeout=verify_timeout,
            )
            if any(
                not _aidefend_policy_hmac.compare_digest(path.read_bytes(), expected)
                for path, expected in captured
            ):
                raise RuntimeError("private signature-verification snapshot changed")
            policy = _aidefend_strict_json(policy_raw)
        finally:
            for path, _ in captured:
                _aidefend_policy_os.chmod(path, 0o600)
    if (
        not isinstance(policy, dict)
        or policy.get("schema_version") != schema_version
        or not isinstance(policy.get("policy_version"), str)
        or not policy["policy_version"]
    ):
        raise RuntimeError("signature-verified policy schema or version is invalid")
    return policy


from datetime import datetime, timezone
from typing import Literal

from pydantic import BaseModel, Field, field_validator


AI_EVENT_SCHEMA_POLICY = _aidefend_load_signed_policy(
    path_env="D005001_EVENT_SCHEMA_POLICY_PATH",
    signature_env="D005001_EVENT_SCHEMA_POLICY_SIGNATURE",
    key_env="D005001_EVENT_SCHEMA_POLICY_VERIFY_KEY",
    schema_version="aidefend.ai-security-event-schema-policy.v1",
)
EVENT_ID_MIN_LENGTH = AI_EVENT_SCHEMA_POLICY.get("event_id_min_length")
EVENT_TYPE_MIN_LENGTH = AI_EVENT_SCHEMA_POLICY.get("event_type_min_length")
EVENT_TYPE_MAX_LENGTH = AI_EVENT_SCHEMA_POLICY.get("event_type_max_length")
EVIDENCE_REFS_MAX_COUNT = AI_EVENT_SCHEMA_POLICY.get("evidence_refs_max_count")
VALIDATION_FINDINGS_MAX_COUNT = AI_EVENT_SCHEMA_POLICY.get("validation_findings_max_count")
if (
    any(
        isinstance(value, bool) or not isinstance(value, int) or value < 1
        for value in (
            EVENT_ID_MIN_LENGTH,
            EVENT_TYPE_MIN_LENGTH,
            EVENT_TYPE_MAX_LENGTH,
            EVIDENCE_REFS_MAX_COUNT,
            VALIDATION_FINDINGS_MAX_COUNT,
        )
    )
    or EVENT_TYPE_MIN_LENGTH > EVENT_TYPE_MAX_LENGTH
):
    raise RuntimeError("signed AI-event schema bounds are invalid")


Severity = Literal["low", "medium", "high", "critical"]


class EvidenceReadyAiEvent(BaseModel):
    schema_version: Literal["aidefend.ai_security_event.v1"] = "aidefend.ai_security_event.v1"
    event_id: str = Field(min_length=EVENT_ID_MIN_LENGTH)
    observed_at: datetime
    event_type: str = Field(min_length=EVENT_TYPE_MIN_LENGTH, max_length=EVENT_TYPE_MAX_LENGTH)
    severity: Severity
    tenant_id: str
    principal_id: str | None = None
    session_id: str | None = None
    trace_id: str
    model_id: str | None = None
    model_provider: str | None = None
    tool_call_id: str | None = None
    source_technique_id: str | None = None
    evidence_refs: list[str] = Field(default_factory=list, max_length=EVIDENCE_REFS_MAX_COUNT)
    retention_class: Literal["standard", "security_1y", "security_7y", "legal_hold"]
    legal_hold: bool = False
    chain_of_custody_id: str | None = None
    redaction_profile: str
    policy_version: str

    @field_validator("evidence_refs")
    @classmethod
    def require_uri_refs(cls, refs: list[str]) -> list[str]:
        for ref in refs:
            if "://" not in ref:
                raise ValueError("evidence_refs must be URI-like references")
        return refs


def build_event(**kwargs) -> EvidenceReadyAiEvent:
    kwargs.setdefault("observed_at", datetime.now(timezone.utc))
    return EvidenceReadyAiEvent(**kwargs)

Step 2: Emit a validated event from the application

from __future__ import annotations

# File: ai_logging/emit_ai_security_event.py

import json
import sys

from ai_logging.ai_security_event_schema import build_event


def emit_event(event) -> None:
    sys.stdout.write(event.model_dump_json() + "\n")


event = build_event(
    event_id="0190f31a-1f3e-7c42-b5b2-5fdd3da76cd9",
    event_type="prompt_sink_policy_denied",
    severity="high",
    tenant_id="tenant-7f1b3a91",
    principal_id="spiffe://aidefend.net/ns/prod/sa/prompt-gateway",
    session_id="0190f31a-26a8-7a54-8c07-5f85ef4c7d43",
    trace_id="4bf92f3577b34da6a3ce929d0e0e4736",
    model_id="fraud-risk-classifier-2026-07-04",
    model_provider="on-premises",
    tool_call_id=None,
    source_technique_id="AID-H-018.005",
    evidence_refs=["s3://security-evidence/prompt-sink/0190f31a-1f3e-7c42-b5b2-5fdd3da76cd9.json"],
    retention_class="security_7y",
    legal_hold=False,
    chain_of_custody_id="0190f31a-321d-7520-b709-67e99e11d7b3",
    redaction_profile="prompt-body-hashed",
    policy_version="2026.07.1",
)
emit_event(event)

Verify the detector and retain evidence

from __future__ import annotations

# File: ai_logging/verify_ai_security_events.py

import json
from pathlib import Path

from ai_logging.ai_security_event_schema import (
    EvidenceReadyAiEvent, VALIDATION_FINDINGS_MAX_COUNT,
)


events = []
failures = []
for line_no, line in enumerate(Path("out/ai_security_events.jsonl").read_text(encoding="utf-8").splitlines(), start=1):
    try:
        events.append(EvidenceReadyAiEvent.model_validate_json(line))
    except Exception as exc:
        failures.append({"line": line_no, "error": str(exc)})

missing_coc = [event.event_id for event in events if event.retention_class in {"security_7y", "legal_hold"} and not event.chain_of_custody_id]
artifact = {
    "schema_version": "aidefend.ai_security_event.validation.v1",
    "validated_event_count": len(events),
    "schema_failure_count": len(failures),
    "schema_failures": failures[:VALIDATION_FINDINGS_MAX_COUNT],
    "missing_chain_of_custody": missing_coc[:VALIDATION_FINDINGS_MAX_COUNT],
    "control_status": (
        "PASS" if events and not failures and not missing_coc else "FAIL"
    ),
    "finding_status": "NOT_EVALUATED",
}
Path("artifacts").mkdir(exist_ok=True)
Path("artifacts/ai-security-event-schema-validation.json").write_text(
    json.dumps(artifact, indent=2, sort_keys=True),
    encoding="utf-8",
)
if artifact["control_status"] != "PASS":
    raise SystemExit("AI security event schema validation failed")

Action: Require security-relevant AI events to pass this schema before ingestion. The evidence artifact for this step is artifacts/ai-security-event-schema-validation.json, proving event count, schema validity, evidence references, retention class, and chain-of-custody completeness.

Verify complete coverage

The population is every evidence-bearing AI event and its stable event ID, evidence references, retention class, legal-hold state, chain-of-custody transition, archive object, and readback receipt.

Verify safely

An independent verifier must resolve each reference with a separate read-only identity, verify evidence digests and custody transitions, read archived objects back, and detect missing, mutable, orphaned, cross-case, or conflicting records, read evidence back, verify signatures, digests, and immutability, and compare the canonical result digest.

AID-D-005.001-G007

Applicability and boundary

Use this implementation path for an MCP server endpoint, whether stdio or Streamable HTTP, whose requests, Tasks, subscriptions, tools, resources, prompts, budget decisions, disable checks, or backend effects must be reconstructed during detection or incident investigation. It is protocol-specific instrumentation under AID-D-005.001, not a separate detector: anomaly rules remain in the applicable Detect control, budget enforcement remains in AID-H-034.008, and disable enforcement remains in AID-H-034.009.

Emit one correlated lifecycle

Derive identity and policy fields from authenticated server context. Treat the caller's JSON-RPC ID only as untrusted protocol correlation input and record a keyed digest. Never write raw tool arguments, elicitation values, resource contents, credentials, authorization headers, session identifiers, or opaque handles to the general log stream.

// File: src/telemetry/mcpSecurityEvents.js
import crypto from "node:crypto";

const allowedPhases = new Set([
  "request_received", "admission_decided", "handler_started",
  "backend_effect_started", "backend_effect_finished",
  "task_state_changed", "subscription_state_changed",
  "response_finished", "request_failed"
]);
const blockablePhases = new Set([
  "request_received", "admission_decided",
  "handler_started", "backend_effect_started"
]);
const policyDecisions = new Set(["ALLOW", "DENY", "NOT_EVALUATED"]);
const budgetStates = new Set([
  "NOT_SELECTED", "RESERVED", "PENDING_RECONCILIATION",
  "SETTLED_COMMITTED", "SETTLED_NO_EFFECT", "DENIED"
]);
const outcomes = new Set([
  "completed", "failed", "denied", "cancelled", "unknown"
]);
const fieldKeys = new Set([
  "backendEffectId", "budgetReservationId", "budgetState",
  "disableGenerations", "durationMs", "errorClass", "outcome",
  "policyDecision", "policyVersion", "subscriptionId", "taskId"
]);

const canonical = value => {
  if (value === null || typeof value === "string" ||
      typeof value === "boolean") return JSON.stringify(value);
  if (Number.isSafeInteger(value)) return String(value);
  if (Array.isArray(value)) return "[" + value.map(canonical).join(",") + "]";
  if (value && Object.getPrototypeOf(value) === Object.prototype) {
    return "{" + Object.keys(value).sort().map(
      key => JSON.stringify(key) + ":" + canonical(value[key])
    ).join(",") + "}";
  }
  throw new Error("non_canonical_mcp_telemetry_value");
};

const digest = (key, label, value) =>
  value === null || value === undefined ? null :
  crypto.createHmac("sha256", key)
    .update(label).update(Buffer.from([0])).update(canonical(value)).digest("hex");

const exactKeys = (value, allowed) =>
  value && Object.getPrototypeOf(value) === Object.prototype
  && Object.keys(value).every(key => allowed.has(key));

function validatePolicy(policy) {
  if (
    !policy || Object.getPrototypeOf(policy) !== Object.prototype
    || !/^[a-f0-9]{64}$/.test(policy.policyDigest)
    || typeof policy.policyVersion !== "string" || !policy.policyVersion
    || !Number.isSafeInteger(policy.maxTextLength)
    || policy.maxTextLength <= 0
    || !Number.isSafeInteger(policy.maxDisableTargets)
    || policy.maxDisableTargets <= 0
    || !Array.isArray(policy.auditRequiredPhases)
    || policy.auditRequiredPhases.length
      !== new Set(policy.auditRequiredPhases).size
    || policy.auditRequiredPhases.some(
      phase => !blockablePhases.has(phase)
    )
  ) throw new Error("signed_mcp_telemetry_policy_invalid");
  return Object.freeze({...policy});
}

function bounded(value, field, maximum) {
  if (value === null || value === undefined) return null;
  if (
    typeof value !== "string" || !value || value.length > maximum
    || /[-]/.test(value)
  ) throw new Error("invalid_" + field);
  return value;
}

function validateFields(fields, policy) {
  if (!exactKeys(fields, fieldKeys)) {
    throw new Error("mcp_telemetry_fields_schema_differs");
  }
  const generations = fields.disableGenerations || [];
  if (
    !Array.isArray(generations)
    || generations.length > policy.maxDisableTargets
    || generations.some(item =>
      !item || Object.getPrototypeOf(item) !== Object.prototype
      || Object.keys(item).sort().join(",") !== "generation,kind"
      || !/^[a-z][a-z0-9_]{0,63}$/.test(item.kind)
      || !Number.isSafeInteger(item.generation)
      || item.generation <= 0
    )
  ) throw new Error("invalid_disable_generation_summary");
  if (
    fields.policyDecision !== undefined
    && !policyDecisions.has(fields.policyDecision)
  ) throw new Error("invalid_policy_decision");
  if (
    fields.budgetState !== undefined
    && !budgetStates.has(fields.budgetState)
  ) throw new Error("invalid_budget_state");
  if (fields.outcome !== undefined && !outcomes.has(fields.outcome)) {
    throw new Error("invalid_outcome");
  }
  if (
    fields.durationMs !== undefined
    && (!Number.isSafeInteger(fields.durationMs) || fields.durationMs < 0)
  ) throw new Error("invalid_duration");
  return {
    ...fields,
    taskId: bounded(fields.taskId, "task_id", policy.maxTextLength),
    subscriptionId: bounded(
      fields.subscriptionId, "subscription_id", policy.maxTextLength
    ),
    policyVersion: bounded(
      fields.policyVersion, "decision_policy_version", policy.maxTextLength
    ),
    budgetReservationId: bounded(
      fields.budgetReservationId,
      "budget_reservation_id",
      policy.maxTextLength
    ),
    backendEffectId: bounded(
      fields.backendEffectId, "backend_effect_id", policy.maxTextLength
    ),
    errorClass: bounded(
      fields.errorClass, "error_class", policy.maxTextLength
    ),
    disableGenerations: generations
  };
}

export function createMcpTelemetry(emitter, input) {
  const policy = validatePolicy(input.auditPolicy);
  if (!Buffer.isBuffer(input.digestKey) || input.digestKey.length < 32 ||
      typeof input.digestKeyVersion !== "string" ||
      !input.digestKeyVersion ||
      !input.auth?.subject || !input.auth?.activeTenantId ||
      !input.clientId || !input.serverVersion ||
      !["stdio", "streamable_http"].includes(input.transport) ||
      !input.method || !/^[0-9a-f-]{36}$/.test(input.operationId) ||
      !/^[0-9a-f-]{36}$/.test(input.traceId) ||
      typeof emitter?.enqueueDurably !== "function") {
    throw new Error("trusted_mcp_telemetry_context_missing");
  }
  let sequence = 0;
  let tail = Promise.resolve();
  const prepared = new Map();
  async function emitPhase(phase, transitionKey, rawFields = {}) {
    if (!allowedPhases.has(phase)) {
      throw new Error("unknown_mcp_event_phase");
    }
    const stableKey = bounded(
      transitionKey, "transition_key", policy.maxTextLength
    );
    if (!stableKey) throw new Error("transition_key_required");
    const fields = validateFields(rawFields, policy);
    const contentDigest = crypto.createHash("sha256").update(canonical({
      phase, transitionKey: stableKey, fields
    })).digest("hex");
    let existing = prepared.get(stableKey);
    if (existing && existing.contentDigest !== contentDigest) {
      throw new Error("telemetry_transition_content_conflict");
    }
    if (!existing) {
      const event = Object.freeze({
      schema_version: "aidefend.mcp_security_event.v2",
      event_id: digest(input.digestKey, "event_id", {
        operationId: input.operationId,
        phase,
        transitionKey: stableKey
      }),
      observed_at: new Date().toISOString(),
      operation_id: input.operationId,
      sequence: ++sequence,
      phase,
      transition_key: stableKey,
      trace_id: input.traceId,
      server_version: input.serverVersion,
      transport: input.transport,
      method: input.method,
      name: input.name || null,
      digest_key_version: input.digestKeyVersion,
      principal_digest: digest(input.digestKey, "principal", input.auth.subject),
      tenant_digest: digest(input.digestKey, "tenant", input.auth.activeTenantId),
      client_digest: digest(input.digestKey, "client", input.clientId),
      protocol_request_id_digest: digest(
        input.digestKey, "jsonrpc_id", input.protocolRequestId
      ),
      task_id_digest: digest(input.digestKey, "task", fields.taskId),
      subscription_id_digest: digest(
        input.digestKey, "subscription", fields.subscriptionId
      ),
      argument_digest: digest(
        input.digestKey, "arguments", input.arguments ?? null
      ),
      telemetry_policy_version: policy.policyVersion,
      telemetry_policy_digest: policy.policyDigest,
      policy_version: fields.policyVersion || null,
      policy_decision: fields.policyDecision || "NOT_EVALUATED",
      budget_reservation_digest: digest(
        input.digestKey, "budget_reservation", fields.budgetReservationId
      ),
      budget_state: fields.budgetState || "NOT_SELECTED",
      disable_generations: fields.disableGenerations,
      backend_effect_digest: digest(
        input.digestKey, "backend_effect", fields.backendEffectId
      ),
      outcome: fields.outcome || "unknown",
      error_class: fields.errorClass || null,
      duration_ms: fields.durationMs ?? null
      });
      existing = {contentDigest, event};
      prepared.set(stableKey, existing);
    }
    const enqueue = () => emitter.enqueueDurably(existing.event, {
      idempotencyKey: existing.event.event_id,
      rejectConflictingPayload: true
    });
    const pending = tail.then(enqueue, enqueue);
    tail = pending.then(() => undefined, () => undefined);
    await pending;
    return existing.event.event_id;
  }
  return Object.freeze({policy, emit: emitPhase});
}

Wrap every request path, including failures

// File: src/telemetry/instrumentMcpRequest.js
import crypto from "node:crypto";
import { createMcpTelemetry } from "./mcpSecurityEvents.js";

async function emitWithPolicy(input, telemetry, phase, key, fields) {
  try {
    await telemetry.emit(phase, key, fields);
    return true;
  } catch (error) {
    await input.emitter.recordHealthFailure({
      component: "mcp_security_telemetry",
      phase,
      operationId: input.operationId,
      errorClass: error?.constructor?.name || "Error"
    });
    await input.onTelemetryFailure({phase, error});
    if (telemetry.policy.auditRequiredPhases.includes(phase)) {
      throw new Error("required_mcp_audit_event_unavailable", {cause: error});
    }
    return false;
  }
}

export async function instrumentMcpRequest(input, handler) {
  const started = performance.now();
  const context = {
    ...input,
    operationId: crypto.randomUUID(),
    traceId: crypto.randomUUID()
  };
  const telemetry = createMcpTelemetry(input.emitter, context);
  const scoped = {...context, telemetry};
  await emitWithPolicy(
    scoped, telemetry, "request_received", "request_received", {}
  );
  let result;
  let handlerError;
  try {
    result = await handler({
      operationId: context.operationId,
      traceId: context.traceId,
      emitSecurityPhase: (phase, key, fields = {}) =>
        emitWithPolicy(scoped, telemetry, phase, key, fields)
    });
  } catch (error) {
    handlerError = error;
  }
  const durationMs = Math.round(performance.now() - started);
  const terminalPhase = handlerError ? "request_failed" : "response_finished";
  const terminalFields = handlerError ? {
    outcome: "failed",
    errorClass: handlerError?.constructor?.name || "Error",
    durationMs
  } : {
    outcome: "completed",
    durationMs,
    taskId: result?.taskId,
    subscriptionId: result?.subscriptionId,
    budgetReservationId: result?.budgetReservationId,
    budgetState: result?.budgetState,
    disableGenerations: result?.disableGenerations || []
  };
  try {
    await emitWithPolicy(
      scoped, telemetry, terminalPhase, "request_terminal", terminalFields
    );
  } catch (telemetryError) {
    await input.onTelemetryFailure({
      phase: terminalPhase,
      error: telemetryError,
      terminalEventUnavailable: true
    });
  }
  if (handlerError) throw handlerError;
  return result;
}

Integration requirements: Call telemetry at the real admission decision, immediately before and after each backend effect, and on every Task or subscription transition. The emitter must durably spool before acknowledging security-critical audit events, retry with bounded storage, expose dropped/rejected/backpressure counters, and prevent log-recursion failures. If a signed technical policy requires audit-before-effect for a covered high-impact operation, failure to enqueue that event blocks the effect; ordinary telemetry failure must still raise an operational alert rather than disappear.

Verify complete lifecycle coverage

From a server request counter or ingress access log independent of the application emitter, enumerate the measured MCP operation population. Join by operation and trace identifiers and require exactly one start plus one terminal event for every operation, ordered phase sequences, matched pre/post backend-effect events, and complete Task/subscription transitions. Test malformed JSON-RPC, authentication rejection, policy denial, handler exception, timeout, client disconnect, process crash, queue saturation, shipper outage, duplicate delivery, retry, and restart. Compare keyed digests and policy versions without reintroducing raw sensitive content.

Open-source tools

• Loguru
• structlog
• Fluentd
• Vector
• Logstash
• Apache Kafka
• OpenTelemetry
• Langfuse
• OpenInference

Source-available / open-weight tools

• Arize Phoenix (Elastic License 2.0; source-available)
• Sentry self-hosted (FSL-1.1-Apache-2.0; source-available)

Commercial / hosted tools

• Datadog
• Splunk Enterprise
• New Relic
• Logz.io
• Amazon CloudWatch Logs
• Amazon Kinesis Data Streams
• Google Cloud Logging
• Azure Monitor Logs
• LangSmith
• Braintrust
• Datadog LLM Observability

Threat mappings

MITRE ATLAS

• N/A

MAESTRO

• Repudiation (L7) (tamper-evident, identity-bound logs preserve accountability for agent actions)

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• N/A

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-16.1.1 Logging Sensitive Conversations (sensitive data retained in this control's AI interaction logs)
• AITech-16.1 Eavesdropping (sensitive data retained in this control's AI interaction logs)
• AITech-8.2 Data Exfiltration / Exposure (recursive secret redaction and raw-reasoning denial at AI log emission)

Google Secure AI Framework 2.0 - Risks

• N/A

Databricks AI Security Framework 3.0

• Raw Data 1.10: Lack of data access logs
• Platform 12.3: Lack of incident response
• Agents - Core 13.8: Repudiation & Untraceability