AID-D-005.004: Specialized Agent & Session Logging

Version 1.20261015 · Actionable sub-technique

Record privacy-preserving forensic telemetry for autonomous and agentic AI systems beyond standard API request/response logs. Capture approved goal identifiers, observable plan or rationale summaries, proposed and executed actions, tool argument and result hashes, policy and verifier decisions, authenticated principals, and memory or knowledge-base interactions.

Pillars: app

Lifecycle: operation

Scope boundary

Owns privacy-bounded agent and session forensic telemetry beyond ordinary API logs, including approved goals, observable rationale summaries, proposed and executed actions, tool bindings, policy decisions, authenticated principals, and memory interactions. It excludes hidden chain-of-thought and raw scratchpads from ordinary logging.

AID-D-005.001: AID-D-005.004 defines specialized agent-session event semantics; AID-D-005.001 supplies the foundational instrumentation, collection, forwarding, and retention pipeline. Agent forensics needs the shared log pipeline but has a narrower, richer event contract.

AID-M-001.003: AID-D-005.004 records what an agent and its skills did during a session; AID-M-001.003 inventories which skill and instruction artifacts are installed, enabled, and reconciled. Session events require stable inventory keys, but activity evidence does not prove the complete deployed skill population.

AID-H-037.001: AID-D-005.004 intentionally logs observable summaries and action evidence without raw hidden reasoning; AID-H-037.001 governs exceptional confidential reasoning-trace access, storage, and break-glass controls. Ordinary forensic logging must not use the exceptional raw-reasoning path as its default data source.

Implementation guidance

AID-D-005.004-G001

Log approved goals, observable rationale summaries, actions, results, and authenticated policy-decision records without persisting hidden chain-of-thought.
Concept and privacy boundary

Forensic telemetry must reconstruct the observable execution path without persisting hidden chain-of-thought, internal scratchpads, or raw model reasoning. Record the approved goal identifier and digest, a short user-visible or policy-approved rationale summary, proposed and executed actions, keyed hashes of arguments and results, independent policy/verifier evidence, and the authenticated principal. Raw reasoning capture is prohibited in ordinary logs; exceptional capture must use the explicit break-glass authorization, encryption, access, retention, and review controls in AID-H-037.001.

Step 1: Emit recursively redacted, evidence-bound events
from __future__ import annotations

# File: agent/forensic_logging.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


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


agent_logger = logging.getLogger("agent_forensic")
agent_logger.setLevel(logging.INFO)

SENSITIVE_KEY = re.compile(
    r"(authorization|cookie|password|passwd|secret|token|api[_-]?key|private[_-]?key)",
    re.IGNORECASE,
)
BEARER_VALUE = re.compile(r"(?i)bearer\s+[a-z0-9._~+/=-]+")
API_KEY_VALUE = re.compile(r"(?i)\b(?:sk|api)[-_][a-z0-9_-]{12,}\b")
AGENT_FORENSIC_POLICY = _aidefend_load_signed_policy(
    path_env="D005004_FORENSIC_POLICY_PATH",
    signature_env="D005004_FORENSIC_POLICY_SIGNATURE",
    key_env="D005004_FORENSIC_POLICY_VERIFY_KEY",
    schema_version="aidefend.agent-forensic-logging-policy.v1",
)
MAX_AGENT_STEPS = AGENT_FORENSIC_POLICY.get("max_steps")
MAX_SUMMARY_CHARS = AGENT_FORENSIC_POLICY.get("max_summary_chars")
if any(
    isinstance(value, bool) or not isinstance(value, int) or value < 1
    for value in (MAX_AGENT_STEPS, MAX_SUMMARY_CHARS)
):
    raise RuntimeError("signed agent-forensic logging bounds are invalid")


@dataclass(frozen=True)
class AuthenticatedPrincipal:
    # Construct this only from verified mTLS/OIDC middleware, never request JSON.
    principal_id: str
    authentication_event_id: str
    workload_id: str


@dataclass(frozen=True)
class AgentDecision:
    observable_rationale_summary: str
    tool_name: str
    tool_params: dict[str, Any]


@dataclass(frozen=True)
class PolicyEvidence:
    allowed: bool
    policy_id: str
    policy_version: str
    verifier_id: str
    evidence_ref: str
    issued_at: str


@dataclass(frozen=True)
class Planner:
    """Concrete deployment adapter bundle; composition root supplies the callable."""
    next_step: Callable[[str, list[dict]], AgentDecision]


@dataclass(frozen=True)
class PolicyVerifier:
    """Authenticated policy adapter with a closed PolicyEvidence return contract."""
    evaluate: Callable[..., PolicyEvidence]


@dataclass(frozen=True)
class ToolExecutor:
    """Bound tool adapter; no default, no-op, or fail-open implementation exists."""
    execute: Callable[[str, dict[str, Any]], dict[str, Any]]


def redact_text(value: str) -> str:
    value = BEARER_VALUE.sub("Bearer <redacted>", value)
    return API_KEY_VALUE.sub("<redacted-key>", value)


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 redact_text(value)
    if value is None or isinstance(value, (bool, int, float)):
        return value
    return redact_text(str(value))


def evidence_digest(value: Any) -> str:
    key = os.environ["AIDEFEND_EVIDENCE_HMAC_KEY"].encode("utf-8")
    if len(key) < 32:
        raise RuntimeError("AIDEFEND_EVIDENCE_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 emit_event(*, session_id: str, event_type: str, principal: AuthenticatedPrincipal, fields: dict) -> None:
    record = {
        "schema_version": "aidefend.agent_forensic.v2",
        "event_id": str(uuid.uuid4()),
        "observed_at_unix": time.time(),
        "event_type": event_type,
        "session_id": session_id,
        "principal_id": principal.principal_id,
        "authentication_event_id": principal.authentication_event_id,
        "workload_id": principal.workload_id,
        "fields": redact_nested(fields),
    }
    agent_logger.info(json.dumps(record, sort_keys=True, separators=(",", ":")))


class ForensicAgentRunner:
    def __init__(
        self,
        planner: Planner,
        policy_verifier: PolicyVerifier,
        tool_executor: ToolExecutor,
        principal: AuthenticatedPrincipal,
        max_steps: int = MAX_AGENT_STEPS,
    ):
        self.planner = planner
        self.policy_verifier = policy_verifier
        self.tool_executor = tool_executor
        self.principal = principal
        self.max_steps = max_steps

    def run(self, *, approved_goal_id: str, approved_goal_text: str) -> str:
        session_id = str(uuid.uuid4())
        history: list[dict] = []
        emit_event(
            session_id=session_id,
            event_type="agent_goal_accepted",
            principal=self.principal,
            fields={
                "approved_goal_id": approved_goal_id,
                "approved_goal_hmac_sha256": evidence_digest(approved_goal_text),
            },
        )

        for step_index in range(self.max_steps):
            decision = self.planner.next_step(approved_goal_id, history)
            args_digest = evidence_digest(decision.tool_params)
            emit_event(
                session_id=session_id,
                event_type="agent_action_proposed",
                principal=self.principal,
                fields={
                    "step": step_index,
                    "approved_goal_id": approved_goal_id,
                    "observable_rationale_summary": redact_text(
                        decision.observable_rationale_summary[:MAX_SUMMARY_CHARS]
                    ),
                    "tool_name": decision.tool_name,
                    "tool_args_hmac_sha256": args_digest,
                },
            )

            policy = self.policy_verifier.evaluate(
                principal_id=self.principal.principal_id,
                approved_goal_id=approved_goal_id,
                tool_name=decision.tool_name,
                args_digest=args_digest,
            )
            emit_event(
                session_id=session_id,
                event_type="agent_policy_verdict",
                principal=self.principal,
                fields={
                    "step": step_index,
                    "allowed": policy.allowed,
                    "policy_id": policy.policy_id,
                    "policy_version": policy.policy_version,
                    "verifier_id": policy.verifier_id,
                    "verifier_evidence_ref": policy.evidence_ref,
                    "verifier_issued_at": policy.issued_at,
                },
            )
            if not policy.allowed:
                break

            tool_started = time.monotonic()
            try:
                result = self.tool_executor.execute(decision.tool_name, decision.tool_params)
            except Exception as exc:
                emit_event(
                    session_id=session_id,
                    event_type="agent_tool_result",
                    principal=self.principal,
                    fields={
                        "step": step_index,
                        "tool_name": decision.tool_name,
                        "tool_args_hmac_sha256": args_digest,
                        "outcome": "error",
                        "error_type": exc.__class__.__name__,
                        "latency_ms": round((time.monotonic() - tool_started) * 1000),
                    },
                )
                raise
            emit_event(
                session_id=session_id,
                event_type="agent_tool_result",
                principal=self.principal,
                fields={
                    "step": step_index,
                    "tool_name": decision.tool_name,
                    "tool_args_hmac_sha256": args_digest,
                    "tool_result_hmac_sha256": evidence_digest(result),
                    "outcome": "success",
                    "latency_ms": round((time.monotonic() - tool_started) * 1000),
                    "goal_complete": bool(result.get("goal_complete", False)),
                },
            )
            history.append({"step": step_index, "result_digest": evidence_digest(result)})
            if result.get("goal_complete") is True:
                emit_event(
                    session_id=session_id,
                    event_type="agent_goal_complete",
                    principal=self.principal,
                    fields={"approved_goal_id": approved_goal_id, "steps": step_index + 1},
                )
                break

        return session_id
Step 2: Verify completeness independently

Ship these events to the append-only store defined by AID-D-005.001. For every executed tool call, a separate verifier should confirm that the same session_id contains an authenticated principal, proposed-action event, policy verdict with a resolvable evidence reference, and result digest. Alert on missing stages, duplicate event IDs, unresolved verifier evidence, or a tool execution with no preceding allow verdict. Retain the signed completeness report for at least the incident-retention period.

AID-D-005.004-G002

Emit OTel GenAI-compatible agent and tool spans in parallel with forensic session logs.
Concept and stability warning

Forensic logs preserve evidence; traces preserve execution shape, timing, and cross-service correlation. Emit both. The OpenTelemetry GenAI semantic conventions are currently marked Development, so pin the convention version and test attribute/span-name compatibility on upgrades. Use current invoke_agent {agent name} and execute_tool {tool name} naming, and keep raw prompts, hidden reasoning, arguments, and outputs out of span attributes.

Instrument agent and tool spans
from __future__ import annotations

# File: agent/otel_agent_trace.py

import hashlib
import json
from collections.abc import Callable

from opentelemetry import trace


tracer = trace.get_tracer("aidefend.agent.session")


def stable_hash(value: object) -> str:
    payload = json.dumps(value, sort_keys=True, default=str).encode("utf-8")
    return hashlib.sha256(payload).hexdigest()


def set_attrs(span, attrs: dict[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)
        else:
            span.set_attribute(key, str(value))


def run_agent_with_trace(
    *,
    session_id: str,
    agent_id: str,
    provider_name: str,
    model_id: str,
    goal_id: str,
    planner_callable: Callable[[], object],
) -> object:
    with tracer.start_as_current_span(f"invoke_agent {agent_id}") as span:
        set_attrs(
            span,
            {
                "gen_ai.operation.name": "invoke_agent",
                "gen_ai.provider.name": provider_name,
                "gen_ai.agent.name": agent_id,
                "gen_ai.request.model": model_id,
                "gen_ai.response.model": model_id,
                "aidefend.session_id": session_id,
                "aidefend.agent_id": agent_id,
                "aidefend.goal_id": goal_id,
            },
        )
        return planner_callable()


def run_tool_with_trace(
    *,
    session_id: str,
    agent_id: str,
    tool_name: str,
    tool_call_id: str,
    tool_params: dict,
    policy_id: str,
    evidence_ref: str,
    tool_callable: Callable[[], dict],
) -> dict:
    with tracer.start_as_current_span(f"execute_tool {tool_name}") as span:
        set_attrs(
            span,
            {
                "gen_ai.operation.name": "execute_tool",
                "gen_ai.tool.name": tool_name,
                "gen_ai.tool.call.id": tool_call_id,
                "aidefend.session_id": session_id,
                "aidefend.agent_id": agent_id,
                "aidefend.policy_id": policy_id,
                "aidefend.tool.params_sha256": stable_hash(tool_params),
                "aidefend.evidence.ref": evidence_ref,
                "aidefend.redaction.applied": True,
            },
        )
        try:
            result = tool_callable()
        except Exception as exc:
            span.record_exception(exc)
            span.set_attribute("aidefend.tool.outcome", "error")
            span.set_attribute("aidefend.error.type", exc.__class__.__name__)
            raise
        span.set_attribute("aidefend.tool.outcome", "success")
        span.set_attribute("aidefend.tool.result_sha256", stable_hash(result))
        return result
Verification

Open the trace in your collector or observability platform and verify that one agent session has a parent invoke_agent span, child execute_tool spans, the same aidefend.session_id as the forensic JSON logs, and no raw secrets in attributes.

Action: Add tracing decorators or callbacks around the planner, retriever, memory writer, HITL checkpoint, and tool executor. Keep forensic logs as the source of replay detail and use OTel spans for correlation, timing, and cross-service investigation.

AID-D-005.004-G003

Log all interactions with external knowledge bases (RAG) in a minimally sensitive form.
Concept

RAG telemetry should prove which retriever and document set participated without copying the raw query or confidential chunks into logs. Record an authenticated principal, session and trace IDs, a keyed query digest, retriever/index version, keyed document-ID digests, scores, authorization policy, and verifier evidence reference.

Instrument the retriever boundary
from __future__ import annotations

# File: agent/rag_audit.py

import json
import logging
import time
import uuid

from agent.forensic_logging import evidence_digest


rag_logger = logging.getLogger("agent_rag")


class AuditedRAGRetriever:
    def __init__(self, vector_db_client, index_id: str, index_version: str):
        self.db_client = vector_db_client
        self.index_id = index_id
        self.index_version = index_version

    def retrieve_documents(
        self,
        *,
        session_id: str,
        trace_id: str,
        authenticated_principal_id: str,
        query_text: str,
        policy_id: str,
        policy_version: str,
        verifier_evidence_ref: str,
        top_k: int = 3,
    ):
        if not authenticated_principal_id or not verifier_evidence_ref:
            raise PermissionError("verified identity and retrieval-policy evidence are required")

        documents = self.db_client.search(query_text=query_text, top_k=top_k)
        event = {
            "schema_version": "aidefend.rag_retrieval.v2",
            "event_id": str(uuid.uuid4()),
            "observed_at_unix": time.time(),
            "event_type": "rag_retrieval",
            "session_id": session_id,
            "trace_id": trace_id,
            "principal_id": authenticated_principal_id,
            "query_hmac_sha256": evidence_digest(query_text),
            "index_id": self.index_id,
            "index_version": self.index_version,
            "retrieved_documents": [
                {
                    "doc_id_hmac_sha256": evidence_digest(getattr(doc, "id", None)),
                    "score": float(getattr(doc, "score", 0.0)),
                }
                for doc in documents
            ],
            "policy_id": policy_id,
            "policy_version": policy_version,
            "verifier_evidence_ref": verifier_evidence_ref,
        }
        rag_logger.info(json.dumps(event, sort_keys=True, separators=(",", ":")))
        return documents
Verify the detector and retain evidence

For each retrieval, independently resolve the policy evidence, recompute document-ID digests from the retriever audit export, and compare the logged index version with the deployed index manifest. Alert on missing retrieval events, unknown index versions, or unresolvable evidence. The event is fresh per retrieval and contains no raw query or chunk text.

Action: Instrument the retriever boundary with keyed digests and evidence references. Keep raw queries and documents in their authorized data stores, not the general telemetry pipeline.

AID-D-005.004-G004

Log secure session initialization to bind identity, integrity, and trust state.
Resolve identity and trust roots outside the request

The session initializer consumes an authenticated workload context from mTLS/OIDC middleware, an attestation policy and key registry loaded from the deployment trust bundle, and nonce/JTI stores owned by the attestation verifier. A request may carry the token and launch ID, but it cannot choose its workload identity, verification key, issuer, audience, or expected nonce.

Verify one-time attestation and measured bytes
from __future__ import annotations

# File: agent/session_start_logger.py

import hashlib
import hmac
import os
import stat
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Mapping, Protocol

import jwt


@dataclass(frozen=True)
class AuthenticatedWorkload:
    principal_id: str
    authentication_event_id: str


@dataclass(frozen=True)
class AttestationPolicy:
    policy_version: str
    issuer: str
    audience: str
    maximum_attestation_age_seconds: int
    clock_leeway_seconds: int


@dataclass(frozen=True)
class SessionCommitReceipt:
    session_id: str
    event_id: str
    acknowledgement_id: str


class SessionInitializationStore(Protocol):
    def commit_once(
        self,
        *,
        launch_id: str,
        principal_id: str,
        issuer: str,
        jti: str,
        presented_nonce: str,
        expires_at: int,
        event: dict,
    ) -> SessionCommitReceipt: ...


def _hash_regular_file(path: Path) -> str:
    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
    digest = hashlib.sha256()
    try:
        info = os.fstat(descriptor)
        if not stat.S_ISREG(info.st_mode):
            raise ValueError("agent bundle is not a regular file")
        while True:
            block = os.read(descriptor, 1024 * 1024)
            if not block:
                break
            digest.update(block)
    finally:
        os.close(descriptor)
    return digest.hexdigest()


def initialize_agent_session(
    *,
    auth: AuthenticatedWorkload,
    launch_id: str,
    agent_bundle_path: Path,
    attestation_token: str,
    policy: AttestationPolicy,
    trusted_keys_by_kid: Mapping[str, object],
    initialization_store: SessionInitializationStore,
) -> str:
    if not auth.principal_id or not auth.authentication_event_id or not launch_id:
        raise PermissionError("authenticated launch binding is incomplete")
    header = jwt.get_unverified_header(attestation_token)
    key_id = header.get("kid")
    if header.get("alg") != "EdDSA" or not isinstance(key_id, str):
        raise PermissionError("attestation header algorithm or key ID is invalid")
    verification_key = trusted_keys_by_kid.get(key_id)
    if verification_key is None:
        raise PermissionError("attestation key is absent from the pinned registry")

    code_digest = _hash_regular_file(agent_bundle_path)
    claims = jwt.decode(
        attestation_token,
        verification_key,
        algorithms=["EdDSA"],
        audience=policy.audience,
        issuer=policy.issuer,
        options={
            "require": [
                "iss", "aud", "sub", "iat", "exp", "jti", "nonce",
                "code_sha256", "evidence_ref",
            ]
        },
        leeway=policy.clock_leeway_seconds,
    )
    now = int(time.time())
    issued_at, expires_at = int(claims["iat"]), int(claims["exp"])
    if (
        issued_at > now + policy.clock_leeway_seconds
        or now - issued_at > policy.maximum_attestation_age_seconds
        or expires_at <= now
    ):
        raise PermissionError("attestation is stale, future-dated, or expired")
    if not hmac.compare_digest(str(claims["sub"]), auth.principal_id):
        raise PermissionError("attestation subject differs from authenticated workload")
    presented_nonce = str(claims["nonce"])
    if not presented_nonce:
        raise PermissionError("attestation nonce is absent")
    if not hmac.compare_digest(str(claims["code_sha256"]), code_digest):
        raise PermissionError("attested code digest differs from measured bundle")
    event = {
        "schema_version": "aidefend.agent_session_start.v3",
        "event_type": "agent_session_start",
        "launch_id": launch_id,
        "principal_id": auth.principal_id,
        "authentication_event_id": auth.authentication_event_id,
        "code_sha256": code_digest,
        "attestation_issuer": policy.issuer,
        "attestation_key_id": key_id,
        "attestation_jti": str(claims["jti"]),
        "attestation_evidence_ref": claims["evidence_ref"],
        "attestation_issued_at": issued_at,
        "attestation_expires_at": expires_at,
        "policy_version": policy.policy_version,
        "observed_at_unix": now,
    }
    commit = initialization_store.commit_once(
        launch_id=launch_id,
        principal_id=auth.principal_id,
        issuer=policy.issuer,
        jti=str(claims["jti"]),
        presented_nonce=presented_nonce,
        expires_at=expires_at,
        event=event,
    )
    if not all((commit.session_id, commit.event_id, commit.acknowledgement_id)):
        raise RuntimeError("session initialization did not return a durable receipt")
    return commit.session_id
Launch and replay tests

Issue the nonce before requesting attestation. After JWT, subject, time, and measured-code verification, commit_once atomically compares and consumes that nonce, claims the issuer/JTI, and appends the session-start event through a transactional outbox. An exact retry returns the existing session and acknowledgement; a changed token or binding is a conflict. Keep the key registry and initialization store outside the agent's identity. Test attacker-generated keys, unknown kid, wrong subject/audience/nonce/code digest, stale/future tokens, reused JTI, symlinked bundle, and sink outage. An auditor reads the event and referenced evidence back and re-verifies the JWT and measured bundle digest against the historical trust registry.

AID-D-005.004-G005

Log every Human-in-the-Loop (HITL) intervention with operator identity and justification.
Concept

HITL telemetry must identify the human through verified authentication context rather than a caller-supplied operator_id. Record the checkpoint, action binding, decision, policy version, and signed approval evidence reference. Store only a recursively redacted justification and a keyed digest of the triggering event.

Emit an authenticated HITL event
from __future__ import annotations

# File: agent/hitl_logger.py

import json
import logging
import time
import uuid
from typing import Literal

from fastapi import HTTPException, Request

from agent.forensic_logging import evidence_digest, redact_nested


hitl_logger = logging.getLogger("agent_hitl")
Decision = Literal["APPROVED", "REJECTED"]


def log_hitl_event(
    *,
    request: Request,
    checkpoint_id: str,
    action_digest: str,
    trigger_event: dict,
    decision: Decision,
    justification: str,
    policy_id: str,
    policy_version: str,
    approval_evidence_ref: str,
) -> None:
    # Verified OIDC/WebAuthn middleware owns these values.
    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 operator identity required")
    if not approval_evidence_ref.startswith(("s3://", "https://")):
        raise ValueError("approval_evidence_ref must reference retained verifier evidence")

    record = {
        "schema_version": "aidefend.hitl_intervention.v2",
        "event_id": str(uuid.uuid4()),
        "observed_at_unix": time.time(),
        "event_type": "hitl_intervention",
        "checkpoint_id": checkpoint_id,
        "action_hmac_sha256": action_digest,
        "trigger_event_hmac_sha256": evidence_digest(trigger_event),
        "operator_principal_id": str(principal_id),
        "authentication_event_id": str(authentication_event_id),
        "decision": decision,
        "justification": redact_nested(justification)[:500],
        "policy_id": policy_id,
        "policy_version": policy_version,
        "approval_evidence_ref": approval_evidence_ref,
    }
    hitl_logger.info(json.dumps(record, sort_keys=True, separators=(",", ":")))
Verify the detector and retain evidence

For every high-risk execution, independently join the HITL event to the signed approval and authentication event, re-verify the signature and action digest, and alert when any link is missing. The HITL JSON event is fresh per intervention; the signed approval and identity-provider event are the independent evidence artifacts.

Action: Resolve operator identity only from verified authentication middleware and retain the signed approval reference with every intervention. Confirmation-token replay and one-time consumption are monitored by AID-D-015.

Open-source tools

Source-available / open-weight tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-D-005.004: Specialized Agent & Session Logging | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service