Scope boundary

Owns longitudinal anomaly scoring for one registered agent against its signed behavioral fingerprint and normal operating patterns. It emits behavioral deviation evidence for investigation, without classifying identity, authorization or goal violations, and does not quarantine or evict the agent.

AID-D-011.003: AID-D-011.001 detects behavioral deviation by an individual registered agent; AID-D-011.003 detects replica, namespace, workload, and image-lineage drift across the approved registered population. One agent can behave abnormally without population drift, and population drift can occur before behavioral evidence is available.

AID-D-011.004: AID-D-011.001 scores broad operational behavior over time; AID-D-011.004 focuses on abuse of valid non-human identities and delegated tokens at the issuance, audience, scope, client, workload, and resource planes. Credential abuse may look behaviorally normal at the tool layer, while behavioral compromise may occur without token misuse.

Implementation guidance

• AID-D-011.001-G001: Create behavioral fingerprints for each agent based on normal operational patterns.
• AID-D-011.001-G002: Implement continuous behavioral scoring that tracks agent trustworthiness and alerts on persistent trust degradation.
• AID-D-011.001-G003: Deploy behavioral drift detection to identify gradual changes in agent behavior.

AID-D-011.001-G001

Concept

Score the Euclidean norm of per-feature z-scores against a signed, cohort-specific baseline. The baseline is security evidence, so verify its Ed25519 signature before reading any threshold or scaler value. Bind it to the exact feature schema, feature order, detector model, protected model, and agent cohort. Reject missing, malformed, stale, future-dated, undersized, or non-finite evidence instead of returning a healthy result.

Step 1: Define and featurize the exact schema

from __future__ import annotations

# File: agent_monitoring/fingerprinting.py

import base64
import hashlib
import json
import math
import os
import subprocess
import time
from pathlib import Path
import hmac
import re
import tempfile
from typing import Any




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 value: (_ for _ in ()).throw(
            ValueError("non-finite JSON constant: " + value)
        ),
    )


def _aidefend_capture_signed_bytes(
    payload_path: Path,
    proof_path: Path,
    key_reference: str,
    timeout_seconds: float,
    proof_flag: str,
) -> bytes:
    if payload_path == proof_path or proof_flag != "--bundle":
        raise RuntimeError("signed payload/proof inputs are invalid")
    payload_raw = payload_path.read_bytes()
    proof_raw = proof_path.read_bytes()
    remote_key = bool(re.match(
        r"^[A-Za-z][A-Za-z0-9+.-]*://", key_reference
    )) or key_reference.startswith("pkcs11:")
    key_raw = None if remote_key else (
        key_reference.encode("utf-8")
        if key_reference.startswith("-----BEGIN ")
        else Path(key_reference).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 / "proof"
        _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_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:
            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(path.read_bytes(), expected)
                for path, expected in captured
            ):
                raise RuntimeError("private signature-verification snapshot changed")
        finally:
            for path, _ in captured:
                os.chmod(path, 0o600)
    return payload_raw


import numpy as np
import pandas as pd
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
    Ed25519PrivateKey,
    Ed25519PublicKey,
)
from scipy.stats import entropy


FEATURE_SCHEMA_VERSION = "aidefend.agent_behavior_features.v2"
BASELINE_SCHEMA_VERSION = "aidefend.agent_behavior_baseline.v3"
DETECTOR_MODEL_ID = "per_feature_zscore_l2.v1"
FEATURE_NAMES = [
    "action_count",
    "error_rate",
    "avg_latency_ms",
    "tool_entropy",
    "distinct_tools",
]


def strict_json_loads(raw: bytes | str) -> object:
    def reject_constant(value: str) -> None:
        raise ValueError("non-standard JSON constant: " + value)

    def unique_object(pairs: list[tuple[str, object]]) -> dict:
        result = {}
        for key, value in pairs:
            if key in result:
                raise ValueError("duplicate JSON key: " + key)
            result[key] = value
        return result

    return json.loads(
        raw,
        parse_constant=reject_constant,
        object_pairs_hook=unique_object,
    )


def load_signed_detector_policy() -> tuple[dict, str]:
    policy_path = Path(os.environ["D011001_DETECTOR_POLICY_PATH"])
    bundle_path = Path(os.environ["D011001_DETECTOR_POLICY_SIGSTORE_BUNDLE"])
    verify_timeout = float(os.environ["AIDEFEND_POLICY_VERIFY_TIMEOUT_SECONDS"])
    if not math.isfinite(verify_timeout) or verify_timeout <= 0:
        raise RuntimeError("signed bootstrap-profile verification timeout is invalid")
    raw = _aidefend_capture_signed_bytes(
        policy_path, bundle_path, os.environ["D011001_DETECTOR_POLICY_VERIFY_KEY"],
        verify_timeout, "--bundle",
    )
    policy = strict_json_loads(raw)
    required = {
        "schema_version", "policy_version", "minimum_baseline_sessions",
        "minimum_calibration_sessions", "calibration_quantile",
        "maximum_baseline_age_seconds", "maximum_future_skew_seconds",
        "training_fraction", "scale_floor",
    }
    if not isinstance(policy, dict) or set(policy) != required:
        raise RuntimeError("signed behavior-detector policy fields differ")
    if (
        policy["schema_version"] != "aidefend.agent-behavior-detector-policy.v1"
        or not isinstance(policy["policy_version"], str)
        or not policy["policy_version"]
    ):
        raise RuntimeError("signed behavior-detector policy identity is invalid")
    for field in (
        "minimum_baseline_sessions", "minimum_calibration_sessions",
        "maximum_baseline_age_seconds",
    ):
        value = policy[field]
        if isinstance(value, bool) or not isinstance(value, int) or value < 1:
            raise RuntimeError("signed behavior-detector integer is invalid: " + field)
    future_skew = policy["maximum_future_skew_seconds"]
    if isinstance(future_skew, bool) or not isinstance(future_skew, int) or future_skew < 0:
        raise RuntimeError("signed behavior-detector future skew is invalid")
    for field in ("calibration_quantile", "training_fraction"):
        value = policy[field]
        if (
            isinstance(value, bool)
            or not isinstance(value, (int, float))
            or not math.isfinite(float(value))
            or not 0 < float(value) < 1
        ):
            raise RuntimeError("signed behavior-detector fraction is invalid: " + field)
    scale_floor = policy["scale_floor"]
    if (
        isinstance(scale_floor, bool)
        or not isinstance(scale_floor, (int, float))
        or not math.isfinite(float(scale_floor))
        or float(scale_floor) <= 0
    ):
        raise RuntimeError("signed behavior-detector scale floor is invalid")
    if policy["minimum_calibration_sessions"] >= policy["minimum_baseline_sessions"]:
        raise RuntimeError("signed calibration population cannot consume the baseline")
    return policy, "sha256:" + hashlib.sha256(raw).hexdigest()


DETECTOR_POLICY, DETECTOR_POLICY_SHA256 = load_signed_detector_policy()


def canonical_json(value: Any) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")


def featurize_agent_session(session_logs: list[dict]) -> np.ndarray:
    if not isinstance(session_logs, list) or not session_logs:
        raise ValueError("live session must be a non-empty event list")
    if any(not isinstance(event, dict) for event in session_logs):
        raise ValueError("every live event must be an object")

    action_events = [
        event for event in session_logs
        if event.get("event_type") == "agent_action_proposed"
    ]
    result_events = [
        event for event in session_logs
        if event.get("event_type") == "agent_tool_result"
    ]
    event_ids = [event.get("event_id") for event in session_logs]
    if (
        any(not isinstance(event_id, str) or not event_id for event_id in event_ids)
        or len(event_ids) != len(set(event_ids))
    ):
        raise ValueError("session event IDs are absent or duplicated")
    if len(result_events) != len(action_events):
        raise ValueError("action/result population is incomplete")
    tools = []
    latencies = []
    errors = 0
    for event in action_events:
        fields = event.get("fields")
        if (
            not isinstance(fields, dict)
            or not isinstance(fields.get("tool_name"), str)
            or not fields["tool_name"].strip()
        ):
            raise ValueError("malformed agent_action_proposed event")
        tools.append(fields["tool_name"])
    for event in result_events:
        fields = event.get("fields")
        if not isinstance(fields, dict) or fields.get("outcome") not in {"success", "error"}:
            raise ValueError("malformed agent_tool_result event")
        errors += int(fields["outcome"] == "error")
        latency = fields.get("latency_ms")
        if (
            isinstance(latency, bool)
            or not isinstance(latency, (int, float))
            or not math.isfinite(float(latency))
            or float(latency) < 0
        ):
            raise ValueError("tool result lacks finite non-negative latency_ms")
        latencies.append(float(latency))

    tool_counts = pd.Series(tools, dtype="object").value_counts()
    vector = np.asarray(
        [
            len(action_events),
            errors / len(result_events) if result_events else 0.0,
            float(np.mean(latencies)) if latencies else 0.0,
            float(entropy(tool_counts.values)) if not tool_counts.empty else 0.0,
            len(tool_counts),
        ],
        dtype=float,
    )
    if vector.shape != (len(FEATURE_NAMES),) or not np.isfinite(vector).all():
        raise ValueError("live behavior vector has wrong shape or non-finite values")
    return vector

Step 2: Build and sign the baseline

# Continuing in agent_monitoring/fingerprinting.py
def build_signed_behavior_baseline(
    *,
    normal_sessions: list[list[dict]],
    protected_model_id: str,
    cohort_id: str,
    output_path: str,
    signing_key_path: str,
    key_id: str,
) -> dict:
    if not protected_model_id or not cohort_id or not key_id:
        raise ValueError("model, cohort, and key bindings are required")
    minimum_sessions = DETECTOR_POLICY["minimum_baseline_sessions"]
    if len(normal_sessions) < minimum_sessions:
        raise RuntimeError(
            f"insufficient baseline: need {minimum_sessions}, got {len(normal_sessions)}"
        )

    matrix = np.vstack([featurize_agent_session(logs) for logs in normal_sessions])
    if matrix.shape[1] != len(FEATURE_NAMES) or not np.isfinite(matrix).all():
        raise RuntimeError("baseline feature matrix is malformed")

    split = int(matrix.shape[0] * float(DETECTOR_POLICY["training_fraction"]))
    training = matrix[:split]
    calibration = matrix[split:]
    if len(training) == 0 or len(calibration) < DETECTOR_POLICY["minimum_calibration_sessions"]:
        raise RuntimeError("insufficient held-out calibration population")

    mean = training.mean(axis=0)
    raw_scale = training.std(axis=0, ddof=0)
    scale = np.where(
        raw_scale < float(DETECTOR_POLICY["scale_floor"]),
        float(DETECTOR_POLICY["scale_floor"]),
        raw_scale,
    )
    calibration_distances = np.linalg.norm((calibration - mean) / scale, ord=2, axis=1)
    threshold = float(
        np.quantile(calibration_distances, DETECTOR_POLICY["calibration_quantile"], method="higher")
    )
    if (
        not np.isfinite(mean).all()
        or not np.isfinite(scale).all()
        or np.any(scale <= 0)
        or not math.isfinite(threshold)
        or threshold <= 0
    ):
        raise RuntimeError("calibrated baseline contains invalid numeric values")

    now = int(time.time())
    baseline = {
        "schema_version": BASELINE_SCHEMA_VERSION,
        "policy_version": DETECTOR_POLICY["policy_version"],
        "policy_sha256": DETECTOR_POLICY_SHA256,
        "feature_schema_version": FEATURE_SCHEMA_VERSION,
        "detector_model_id": DETECTOR_MODEL_ID,
        "protected_model_id": protected_model_id,
        "cohort_id": cohort_id,
        "feature_names": FEATURE_NAMES,
        "generated_at_unix": now,
        "expires_at_unix": now + DETECTOR_POLICY["maximum_baseline_age_seconds"],
        "mean": mean.tolist(),
        "scale": scale.tolist(),
        "distance_metric": "l2_norm_of_per_feature_z_scores",
        "calibration_quantile": DETECTOR_POLICY["calibration_quantile"],
        "distance_threshold": threshold,
        "training_sample_count": int(len(training)),
        "calibration_sample_count": int(len(calibration)),
    }

    loaded = serialization.load_pem_private_key(
        Path(signing_key_path).read_bytes(),
        password=None,
    )
    if not isinstance(loaded, Ed25519PrivateKey):
        raise RuntimeError("baseline signing key must be Ed25519")
    signature = loaded.sign(canonical_json(baseline))
    envelope = {
        "baseline": baseline,
        "signature": {
            "algorithm": "Ed25519",
            "key_id": key_id,
            "value_base64": base64.b64encode(signature).decode("ascii"),
        },
    }
    Path(output_path).write_text(
        json.dumps(envelope, indent=2, sort_keys=True),
        encoding="utf-8",
    )
    return envelope

Step 3: Verify every binding and numeric invariant before scoring

# Continuing in agent_monitoring/fingerprinting.py
BASELINE_KEYS = {
    "schema_version",
    "policy_version",
    "policy_sha256",
    "feature_schema_version",
    "detector_model_id",
    "protected_model_id",
    "cohort_id",
    "feature_names",
    "generated_at_unix",
    "expires_at_unix",
    "mean",
    "scale",
    "distance_metric",
    "calibration_quantile",
    "distance_threshold",
    "training_sample_count",
    "calibration_sample_count",
}


def load_verified_baseline(
    *,
    envelope_path: str,
    public_key_path: str,
    expected_key_id: str,
    expected_protected_model_id: str,
    expected_cohort_id: str,
) -> dict:
    envelope = strict_json_loads(Path(envelope_path).read_bytes())
    if not isinstance(envelope, dict) or set(envelope) != {"baseline", "signature"}:
        raise RuntimeError("malformed baseline envelope")
    baseline = envelope["baseline"]
    signature = envelope["signature"]
    if not isinstance(baseline, dict) or set(baseline) != BASELINE_KEYS:
        raise RuntimeError("baseline fields do not match the exact schema")
    if not isinstance(signature, dict) or set(signature) != {
        "algorithm", "key_id", "value_base64"
    }:
        raise RuntimeError("malformed baseline signature")
    if signature["algorithm"] != "Ed25519" or signature["key_id"] != expected_key_id:
        raise RuntimeError("unexpected baseline signing identity")

    loaded = serialization.load_pem_public_key(Path(public_key_path).read_bytes())
    if not isinstance(loaded, Ed25519PublicKey):
        raise RuntimeError("baseline verification key must be Ed25519")
    try:
        decoded_signature = base64.b64decode(signature["value_base64"], validate=True)
        loaded.verify(decoded_signature, canonical_json(baseline))
    except (ValueError, InvalidSignature) as exc:
        raise RuntimeError("baseline signature verification failed") from exc

    exact_bindings = {
        "schema_version": BASELINE_SCHEMA_VERSION,
        "policy_version": DETECTOR_POLICY["policy_version"],
        "policy_sha256": DETECTOR_POLICY_SHA256,
        "feature_schema_version": FEATURE_SCHEMA_VERSION,
        "detector_model_id": DETECTOR_MODEL_ID,
        "protected_model_id": expected_protected_model_id,
        "cohort_id": expected_cohort_id,
        "feature_names": FEATURE_NAMES,
        "distance_metric": "l2_norm_of_per_feature_z_scores",
        "calibration_quantile": DETECTOR_POLICY["calibration_quantile"],
    }
    for field, expected in exact_bindings.items():
        if baseline.get(field) != expected:
            raise RuntimeError(f"baseline binding mismatch: {field}")

    now = int(time.time())
    generated = baseline.get("generated_at_unix")
    expires = baseline.get("expires_at_unix")
    if isinstance(generated, bool) or not isinstance(generated, int):
        raise RuntimeError("baseline generated_at_unix must be an integer")
    if isinstance(expires, bool) or not isinstance(expires, int):
        raise RuntimeError("baseline expires_at_unix must be an integer")
    if generated > now + DETECTOR_POLICY["maximum_future_skew_seconds"]:
        raise RuntimeError("baseline is future-dated")
    if expires <= generated or expires - generated > DETECTOR_POLICY["maximum_baseline_age_seconds"]:
        raise RuntimeError("baseline lifetime is invalid")
    if now > expires or now - generated > DETECTOR_POLICY["maximum_baseline_age_seconds"]:
        raise RuntimeError("behavior baseline is stale")

    training_count = baseline.get("training_sample_count")
    calibration_count = baseline.get("calibration_sample_count")
    if (
        isinstance(training_count, bool)
        or isinstance(calibration_count, bool)
        or not isinstance(training_count, int)
        or not isinstance(calibration_count, int)
        or training_count <= 0
        or calibration_count <= 0
        or training_count + calibration_count < DETECTOR_POLICY["minimum_baseline_sessions"]
        or calibration_count < DETECTOR_POLICY["minimum_calibration_sessions"]
    ):
        raise RuntimeError("baseline sample population is insufficient")

    mean = np.asarray(baseline.get("mean"), dtype=float)
    scale = np.asarray(baseline.get("scale"), dtype=float)
    threshold = baseline.get("distance_threshold")
    if (
        mean.shape != (len(FEATURE_NAMES),)
        or scale.shape != (len(FEATURE_NAMES),)
        or not np.isfinite(mean).all()
        or not np.isfinite(scale).all()
        or np.any(scale <= 0)
        or isinstance(threshold, bool)
        or not isinstance(threshold, (int, float))
        or not math.isfinite(float(threshold))
        or float(threshold) <= 0
    ):
        raise RuntimeError("baseline numeric invariants failed")
    return baseline


def score_behavior(
    *,
    live_session_logs: list[dict],
    live_protected_model_id: str,
    live_cohort_id: str,
    envelope_path: str,
    public_key_path: str,
    expected_key_id: str,
) -> dict:
    baseline = load_verified_baseline(
        envelope_path=envelope_path,
        public_key_path=public_key_path,
        expected_key_id=expected_key_id,
        expected_protected_model_id=live_protected_model_id,
        expected_cohort_id=live_cohort_id,
    )
    live = featurize_agent_session(live_session_logs)
    mean = np.asarray(baseline["mean"], dtype=float)
    scale = np.asarray(baseline["scale"], dtype=float)
    if live.shape != mean.shape or not np.isfinite(live).all():
        raise RuntimeError("live feature vector is malformed")

    z_scores = (live - mean) / scale
    if not np.isfinite(z_scores).all():
        raise RuntimeError("live z-scores are non-finite")
    distance = float(np.linalg.norm(z_scores, ord=2))
    threshold = float(baseline["distance_threshold"])
    if not math.isfinite(distance):
        raise RuntimeError("live distance is non-finite")

    return {
        "schema_version": "aidefend.agent_behavior_score.v2",
        "policy_version": DETECTOR_POLICY["policy_version"],
        "policy_sha256": DETECTOR_POLICY_SHA256,
        "status": "fail" if distance > threshold else "pass",
        "feature_schema_version": FEATURE_SCHEMA_VERSION,
        "detector_model_id": DETECTOR_MODEL_ID,
        "protected_model_id": live_protected_model_id,
        "cohort_id": live_cohort_id,
        "feature_names": FEATURE_NAMES,
        "distance": distance,
        "threshold": threshold,
        "feature_z_scores": dict(zip(FEATURE_NAMES, z_scores.tolist())),
        "baseline_generated_at_unix": baseline["generated_at_unix"],
        "baseline_expires_at_unix": baseline["expires_at_unix"],
    }

Verify the detector and retain evidence

Build the baseline from chronologically ordered known-good sessions so the signed policy-selected fraction remains held out for calibration. In CI, mutate the signed envelope, reorder a feature, change model/cohort IDs, inject NaN/Infinity, zero or negative scales, future/stale timestamps, low sample counts, and malformed live events; every case must fail. Retain the signed baseline envelope, raw calibration matrix, and per-session score. An independent auditor verifies the signature and recomputes the exact feature vector, scaler, percentile, and distance.

Action: Never score from an unsigned or weakly bound baseline. Treat any verification or validation failure as insufficient evidence and alert on impaired behavioral detection.

Before you begin

Verify the signed baseline envelope before use, extract the exact feature vector from every eligible finalized session, reject malformed/non-finite or unsupported features, score only against the matching cohort/model release, and emit a fingerprint anomaly without enforcing or changing agent behavior.

AID-D-011.001-G002

Production implementation

Authenticate and validate every event and its evidence before scoring, persist append-only events plus durable high-water/checkpoint state, compute the exact versioned score for finalized windows, preserve late/revised/duplicate disposition and per-event contributions, and emit sustained-degradation findings. Unknown events never default to neutral; attacker-controllable healthy events cannot automatically restore trust; resets require signed recovery authorization and preserve history.

Independent verification

An independent verifier reconstructs state from the immutable event ledger, recomputes every score/window/threshold and reset, tests adverse, healthy, unknown, forged, duplicate, late, restart, state-loss and unauthorized-recovery fixtures, and reconciles all eligible producer events and finalized windows.

AID-D-011.001-G003

Production implementation

Build comparable reference/current feature matrices from authenticated immutable events, enforce schema/stratum/sample/missingness requirements, normalize using only the reference population, run every pinned per-feature test plus declared correction, preserve raw statistics and uncertainty, and emit distribution-drift findings. Baselines never self-update from monitored traffic.

Independent verification

An independent runner reconstructs matrices from event manifests, recomputes normalization/statistics/aggregate result, tests no-drift, gradual/abrupt drift, zero-variance, sparse/missing, seasonal-stratum, model-release and poisoned-reference fixtures, and reconciles every eligible finalized window.

Open-source tools

• scikit-learn
• PyOD
• pandas
• NumPy
• SciPy
• Evidently
• NannyML
• MLflow
• TensorBoard
• JupyterLab

Source-available / open-weight tools

• Arize Phoenix (Elastic License 2.0; source-available)

Commercial / hosted tools

• Arize AX
• Fiddler AI Observability
• NannyML Cloud
• Splunk Enterprise Security
• Exabeam New-Scale SIEM
• Securonix Unified Defense SIEM
• Datadog
• Splunk Machine Learning Toolkit

Threat mappings

MITRE ATLAS

• N/A

MAESTRO

• Compromised Agents (L7) (longitudinal deviations in tool activity support investigation of compromised-agent behavior)

OWASP LLM Top 10 2026

• N/A

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI10:2026 Rogue Agents (behavioral deviation findings support rogue-agent investigation without identifying hijacked goals or authority violations)

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• N/A

Google Secure AI Framework 2.0 - Risks

• RA: Rogue Actions (agent activity deviations provide a behavioral warning layer; individual action authorization remains separate)

Databricks AI Security Framework 3.0

• Agents - Core 13.13: Rogue Agents in Multi-Agent Systems (longitudinal tool-activity deviations support registered-agent investigation)