AID-D-010: AI Goal Integrity Monitoring & Deviation Detection

Version 1.20261015 · Actionable technique

Continuously monitor whether an agent's proposed actions, intermediate plans, and long-running behavior remain aligned with the currently approved goal state. Emit goal-consistency findings, rolling adherence scores, and drift alerts when an agent appears to diverge from approved objectives.

Pillars: app

Lifecycle: operation

Scope boundary

Owns detection, scoring, and alerting for deviation from the currently approved agent goal state. It does not sign, approve, promote, persist, or roll back the goal manifest.

AID-M-009.005: AID-D-010 detects whether live plans and actions diverge from the approved goal; AID-M-009.005 governs changes to the approved goal manifest, approval quorum, promotion, provenance, and rollback. Goal drift is a runtime finding, while changing the authoritative goal is a governed lifecycle action.

AID-D-011.002: AID-D-010 evaluates semantic alignment with the approved goal; AID-D-011.002 verifies the authenticity and contract integrity of inter-agent events that may carry goals, plans, or actions. Authentic interaction evidence can still express goal deviation, and aligned content can still arrive through an invalid event chain.

Implementation guidance

AID-D-010-G001

Continuously score semantic alignment between the approved goal state and the agent's proposed actions or plans, and emit a structured finding when the score falls below the goal-specific threshold.
Production implementation

At the trusted pre-execution observation point, canonicalize the immutable approved goal and proposed plan/action through pinned adapters, compute the declared entailment/alignment/constraint features, preserve per-constraint results and uncertainty, validate signed exceptions, and emit a finding before execution when latency permits. Agent-generated explanations are untrusted features, and the detector neither authorizes nor blocks the action.

Independent verification

An independent isolated runner replays exact goal and proposal bytes with pinned artifacts, recomputes features/thresholds, tests aligned, deviating, ambiguous, stale-goal, exception and unsupported-action fixtures, and reconciles every eligible proposal with the authoritative action ledger.

Executable goal-to-action alignment score
# File: goal_monitor/alignment.py
from __future__ import annotations

import numpy as np
from sentence_transformers import SentenceTransformer


MODEL = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")


def alignment(goal: str, proposed_action: str, minimum_score: float) -> dict:
    if not goal.strip() or not proposed_action.strip() or not -1 <= minimum_score <= 1:
        raise ValueError("invalid goal, action, or signed threshold")
    vectors = MODEL.encode(
        [goal, proposed_action], normalize_embeddings=True,
        convert_to_numpy=True,
    )
    score = float(np.dot(vectors[0], vectors[1]))
    if not np.isfinite(score):
        raise RuntimeError("alignment score is non-finite")
    return {
        "score": score,
        "finding": score < minimum_score,
        "model": "sentence-transformers/all-MiniLM-L6-v2",
    }

Pin the embedding artifact and tokenizer by digest, calibrate the threshold per goal class on signed benign/adversarial cases, and emit a finding only; authorization or blocking remains with the owning Harden control.

AID-D-010-G002

Maintain a rolling goal-adherence score per agent so low-and-slow deviation trends trigger investigation before they become a full goal hijack incident.
Production implementation

Consume authenticated goal-bound action findings in event-time order, persist append-only events plus durable high-water/checkpoint state, compute the versioned rolling score for every finalized window, preserve late/revised event disposition, and emit persistent-degradation findings. Never restore trust from attacker-controllable healthy events alone; reset/recovery requires the signed policy event and does not erase prior evidence.

Independent verification

An independent verifier reconstructs every finalized window from the immutable event ledger, recomputes score/decay and threshold crossings, tests gradual drift, burst, late/duplicate event, restart, state-loss, goal-change and unauthorized-recovery fixtures, and reconciles eligible events/windows.

Executable low-and-slow deviation detector
# File: goal_monitor/rolling_deviation.py
from __future__ import annotations


def update_cusum(
    previous: float, alignment_score: float, *,
    expected_floor: float, slack: float, alert_threshold: float,
) -> dict:
    if not all(0 <= value <= 1 for value in (
        alignment_score, expected_floor, slack, alert_threshold
    )):
        raise ValueError("signed rolling policy is invalid")
    deviation = max(0.0, expected_floor - alignment_score - slack)
    current = max(0.0, previous + deviation)
    return {
        "cusum": current,
        "finding": current >= alert_threshold,
        "instant_deviation": deviation,
    }

Persist state by authenticated agent and approved-goal revision, reset only on a signed goal transition, and replay the ordered complete event series. Missing intervals or state rollback produce a coverage finding, not a clean trend.

AID-D-010-G003

Profile expected tool and workflow patterns for each goal class and flag statistically significant drift that suggests indirect goal manipulation.
Production implementation

Derive declared features from authenticated immutable action events, compare each finalized stratum/window with its exact baseline using the pinned calibrated tests, control multiple comparisons as policy specifies, preserve sample-size and uncertainty, and emit tool/resource/sequence drift findings. Baseline updates require a separately approved version and never learn automatically from the monitored window.

Independent verification

An independent runner rebuilds feature windows and statistics from the event ledger, checks baseline and approved-change signatures, replays expected, abrupt, gradual, sparse-sample, seasonal, newly approved and poisoning fixtures, and reconciles every eligible event/window.

Executable workflow-distribution drift check
# File: goal_monitor/workflow_drift.py
from __future__ import annotations

import numpy as np
from scipy.spatial.distance import jensenshannon


def workflow_drift(
    observed_counts: dict[str, int],
    baseline_probabilities: dict[str, float],
    maximum_distance: float,
) -> dict:
    keys = sorted(set(observed_counts) | set(baseline_probabilities))
    if not keys or not 0 < maximum_distance <= 1:
        raise ValueError("invalid workflow population or threshold")
    if any(observed_counts.get(key, 0) < 0 for key in keys):
        raise ValueError("negative workflow count")
    observed = np.array([observed_counts.get(key, 0) for key in keys], dtype=float)
    baseline = np.array([baseline_probabilities.get(key, 0.0) for key in keys])
    if observed.sum() == 0 or baseline.sum() <= 0:
        raise ValueError("empty observed or baseline distribution")
    distance = float(jensenshannon(observed / observed.sum(), baseline / baseline.sum()))
    return {"distance": distance, "finding": distance > maximum_distance, "keys": keys}

Version the tool/workflow vocabulary and baseline by goal class. Unknown tools remain explicit categories; do not discard them before computing drift.

No-progress probing is a goal-deviation case only when the goal boundary is crossed

Add policy-labelled features for repeated denied destinations, unavailable resources, exhausted tool or reasoning budgets, peer-suggested out-of-scope actions, and repeated attempts that make no measurable progress. Emit a goal-deviation finding only when the authenticated sequence also departs from the approved goal or its allowed recovery path; ordinary retries, a blocked task, or slow progress alone are not goal drift. Reconcile the finding against external network, IAM, and tool observations so an agent cannot self-report that it stopped while continuing to probe.

AID-D-010-G004

Implement Rationale-vs-Action Consistency Monitoring by comparing observable plan, rationale summary, proposed tool call, and approved goal state, then emit anomaly telemetry when the action does not match the stated intent.
Production implementation

Capture only the approved observable summary, bind it and the approved goal to the immutable proposed tool call, compute declared summary-to-action and goal-to-action consistency features, then bind the eventual execution receipt to detect last-mile substitution. Resolve tool approval from a signature-verified registry fact bound to the tool, action, goal revision, registry revision, policy digest, threshold, and expiry; never accept a caller-supplied approval boolean. Emit mismatch or missing-binding evidence without requesting private reasoning, exposing secrets, or granting or denying execution.

Executable observable-intent consistency check
# File: goal_monitor/intent_consistency.py
from __future__ import annotations

import base64
import json
import math
import re
from datetime import datetime, timezone
from typing import Any

import numpy as np
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey

FACT_FIELDS = {
    "schema_version", "tool_id", "action_id", "goal_revision", "registry_revision",
    "policy_sha256", "decision", "minimum_similarity", "expires_at",
}
SHA256 = re.compile(r"^[a-f0-9]{64}$")


def _reject_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ValueError(f"duplicate approval-fact key: {key}")
        result[key] = value
    return result


def _approval_fact(
    document: bytes,
    signature_b64: str,
    public_key: Ed25519PublicKey,
    *,
    tool_id: str,
    action_id: str,
    goal_revision: str,
    policy_sha256: str,
) -> dict[str, Any]:
    public_key.verify(base64.b64decode(signature_b64, validate=True), document)
    fact = json.loads(
        document.decode("utf-8"), object_pairs_hook=_reject_duplicates,
        parse_constant=lambda value: (_ for _ in ()).throw(ValueError(value)),
    )
    if not isinstance(fact, dict) or set(fact) != FACT_FIELDS:
        raise ValueError("tool approval fact schema differs")
    if fact["schema_version"] != "aidefend.tool-approval-fact.v1":
        raise ValueError("unsupported tool approval fact")
    expected = {
        "tool_id": tool_id, "action_id": action_id,
        "goal_revision": goal_revision, "policy_sha256": policy_sha256,
    }
    if any(fact[field] != value for field, value in expected.items()):
        raise ValueError("tool approval fact is misbound")
    if not isinstance(fact["registry_revision"], str) or not fact["registry_revision"].strip():
        raise ValueError("missing registry revision")
    if not isinstance(fact["policy_sha256"], str) or not SHA256.fullmatch(fact["policy_sha256"]):
        raise ValueError("invalid policy digest")
    minimum = fact["minimum_similarity"]
    if isinstance(minimum, bool) or not isinstance(minimum, (int, float)) or not math.isfinite(minimum) or not -1 <= minimum <= 1:
        raise ValueError("invalid signed similarity threshold")
    expiry = datetime.fromisoformat(str(fact["expires_at"]).replace("Z", "+00:00"))
    if expiry.utcoffset() is None or expiry.astimezone(timezone.utc) <= datetime.now(timezone.utc):
        raise ValueError("tool approval fact is expired")
    if fact["decision"] not in {"APPROVED", "DENIED"}:
        raise ValueError("unknown tool approval decision")
    return fact


def consistency(
    plan_vector: np.ndarray,
    action_vector: np.ndarray,
    *,
    approval_document: bytes,
    approval_signature_b64: str,
    approval_public_key: Ed25519PublicKey,
    tool_id: str,
    action_id: str,
    goal_revision: str,
    policy_sha256: str,
) -> dict[str, object]:
    fact = _approval_fact(
        approval_document, approval_signature_b64, approval_public_key,
        tool_id=tool_id, action_id=action_id,
        goal_revision=goal_revision, policy_sha256=policy_sha256,
    )
    plan = np.asarray(plan_vector, dtype=float)
    action = np.asarray(action_vector, dtype=float)
    if plan.shape != action.shape or plan.ndim != 1 or plan.size == 0:
        raise ValueError("observable plan/action vectors differ")
    if not np.isfinite(plan).all() or not np.isfinite(action).all():
        raise ValueError("non-finite vector")
    denominator = float(np.linalg.norm(plan) * np.linalg.norm(action))
    if denominator == 0:
        raise ValueError("zero-norm observable intent")
    score = float(np.dot(plan, action) / denominator)
    approved = fact["decision"] == "APPROVED"
    return {
        "similarity": score,
        "tool_approved": approved,
        "registry_revision": fact["registry_revision"],
        "policy_sha256": fact["policy_sha256"],
        "finding": (not approved) or score < float(fact["minimum_similarity"]),
    }
Runnable negative fixtures
# File: tests/test_intent_consistency.py
import base64
import json

import numpy as np
import pytest
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from goal_monitor.intent_consistency import consistency

KEY = Ed25519PrivateKey.generate()
BASE = {
    "schema_version": "aidefend.tool-approval-fact.v1",
    "tool_id": "tool:payments.read", "action_id": "act-2041",
    "goal_revision": "goal-7", "registry_revision": "registry-31",
    "policy_sha256": "a" * 64, "decision": "APPROVED",
    "minimum_similarity": 0.75, "expires_at": "2099-01-01T00:00:00Z",
}


def signed_fact(**changes):
    fact = {**BASE, **changes}
    document = json.dumps(fact, sort_keys=True, separators=(",", ":")).encode()
    signature = base64.b64encode(KEY.sign(document)).decode()
    return document, signature


def run(document, signature):
    return consistency(
        np.array([1.0, 0.0]), np.array([1.0, 0.0]),
        approval_document=document, approval_signature_b64=signature,
        approval_public_key=KEY.public_key(), tool_id="tool:payments.read",
        action_id="act-2041", goal_revision="goal-7", policy_sha256="a" * 64,
    )


def test_signed_denial_is_a_finding():
    document, signature = signed_fact(decision="DENIED")
    assert run(document, signature)["finding"] is True


def test_tool_binding_mismatch_is_error():
    document, signature = signed_fact(tool_id="tool:payments.write")
    with pytest.raises(ValueError, match="misbound"):
        run(document, signature)


def test_tampered_fact_is_error():
    document, signature = signed_fact()
    with pytest.raises(InvalidSignature):
        run(document.replace(b"APPROVED", b"DENIED__"), signature)
Independent verification

Replay exact redacted summary, goal, proposal, signed registry fact, and execution bytes with pinned artifacts. Include matching, signed denial, invalid signature, wrong tool/action/goal revision or policy digest, stale registry fact, expired approval, misleading summary, hidden argument, resource substitution, no-summary, and scorer-failure fixtures; reconcile every eligible proposal and execution pair. Use only observable plan or rationale summaries and proposed action records, never raw chain-of-thought. This detector reports findings; the owning Harden control decides whether execution is allowed.

Open-source tools

Commercial / hosted tools

Threat mappings

MITRE ATLAS

MAESTRO

OWASP LLM Top 10 2026

OWASP ML Top 10 2023

OWASP Top 10 for Agentic Applications 2026

NIST Adversarial Machine Learning 2025

Cisco Integrated AI Security and Safety Framework

Google Secure AI Framework 2.0 - Risks

Databricks AI Security Framework 3.0

AID-D-010: AI Goal Integrity Monitoring & Deviation Detection | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service