AID-D-003.005: Stateful Session Monitoring: Intent Drift + Invariant-Breach Signals

Version 1.20261015 · Actionable sub-technique

Attackers can split one malicious goal into many small, individually normal requests. Stateful monitoring reconnects those steps into one picture by maintaining a tenant-scoped session ledger and emitting risk signals across turns. Named patterns include Crescendo-style gradual escalation, many-shot pressure, policy-puppetry role or policy shifts, delayed payload activation, and context-saturation campaigns that only become visible when turns are evaluated together.

The monitor tracks:

Pillars: app, infra

Lifecycle: operation

Scope boundary

Owns tenant-scoped cross-turn intent-drift and invariant-breach signals for multi-step session attacks. It is detective-only and does not implement fail-closed session authorization.

AID-D-003.004: AID-D-003.005 detects broader intent and invariant changes across the session; AID-D-003.004 detects anomalous ordering and composition of tool calls. The session ledger covers prompts, reads, destinations, and accumulated context beyond the tool sequence alone.

AID-D-003.007: AID-D-003.005 tracks cumulative cross-turn context and intent risk; AID-D-003.007 decides whether accumulated signals produced a protected sensitive-attribute inference. Session accumulation is an enabling signal, while the sensitive-inference detector owns the category-specific privacy finding.

AID-H-018.006: AID-D-003.005 emits cross-turn intent and invariant risk; AID-H-018.006 re-verifies authorization and binding immediately before execution and fails closed on stale or changed facts. A session finding informs the gate but does not itself prevent the action.

Implementation guidance

AID-D-003.005-G001

Maintain a session security ledger and compute intent-drift and risk signals across turns.
Bind monitoring to an approved session baseline

Run this detective service behind authenticated gateway, retrieval and execution telemetry adapters. Resolve tenant/user/session and task authority from server state. Fetch sensitivity from the current resource catalog, and destinations/tools from resolved execution descriptors; do not accept client flags. Verify policy, task and producer receipts through AID-H-004.005 or the protocol-native equivalent before calling the module. The embedding adapter runs a locally admitted, digest-pinned model bundle under AID-H-003.009; its signed observation binds that manifest, the exact text digest and returned vector. A registry model name alone is insufficient. Thresholds and event/retention bounds come from a versioned calibrated policy.

# File: detection/session_monitor.py
from __future__ import annotations
import hashlib
import json
import math
import sqlite3

def canonical(value):
    return json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False)

def digest(value):
    return hashlib.sha256(canonical(value).encode()).hexdigest()

def cosine(a, b):
    if not a or len(a) != len(b) or any(isinstance(x, bool) or not math.isfinite(x) for x in [*a, *b]):
        raise ValueError("invalid embedding")
    aa, bb = math.sqrt(sum(x*x for x in a)), math.sqrt(sum(x*x for x in b))
    if aa == 0 or bb == 0:
        raise ValueError("zero embedding")
    return max(-1.0, min(1.0, sum(x*y for x, y in zip(a, b))/(aa*bb)))

class SessionMonitor:
    """Trusted telemetry consumer; this class is never exposed as a user API.

    The gateway adapter authenticates each source and verifies event/policy
    bytes before calling it. document_catalog is the current data catalog;
    callers cannot supply a sensitive flag or replace the approved task.
    Use a protected local SQLite volume for a single detector installation.
    A multi-site deployment uses the same transaction in a central database,
    never independent SQLite copies with conflicting session ownership.
    """
    def __init__(self, connection, policy, document_catalog):
        self.db, self.policy, self.catalog = connection, dict(policy), document_catalog
        required = {"version", "sha256", "embedding_manifest_sha256", "ttl_seconds",
                    "maximum_events", "drift_threshold", "sensitive_read_threshold",
                    "context_token_threshold", "repeat_threshold"}
        if set(policy) != required or not policy["version"]:
            raise ValueError("session policy schema")
        for key in ("sha256", "embedding_manifest_sha256"):
            if len(policy[key]) != 64 or set(policy[key]) - set("0123456789abcdef"):
                raise ValueError("session policy digest")
        for key in ("ttl_seconds", "maximum_events", "sensitive_read_threshold",
                    "context_token_threshold", "repeat_threshold"):
            if type(policy[key]) is not int or policy[key] < 1:
                raise ValueError("positive policy bound required")
        if type(policy["drift_threshold"]) not in (int, float) or not math.isfinite(policy["drift_threshold"]) or not 0 < policy["drift_threshold"] <= 2:
            raise ValueError("invalid calibrated drift threshold")
        self.db.executescript("""
            CREATE TABLE IF NOT EXISTS sessions (key TEXT PRIMARY KEY, state TEXT NOT NULL);
            CREATE TABLE IF NOT EXISTS events (key TEXT NOT NULL, id TEXT NOT NULL,
                digest TEXT NOT NULL, result TEXT NOT NULL, PRIMARY KEY(key,id));
        """)

    def register(self, *, tenant, user, session, task, baseline_embedding, now):
        # task is obtained from the signed approved task authority, not turn 1.
        if not all(isinstance(x, str) and x for x in (tenant, user, session)):
            raise ValueError("authenticated session identity required")
        cosine(baseline_embedding, baseline_embedding)
        if set(task) != {"task_sha256", "allowed_tools", "allowed_destinations"} or not task["task_sha256"]:
            raise ValueError("approved task binding required")
        key = digest([tenant, user, session])
        state = {"tenant": tenant, "task": task, "baseline": baseline_embedding,
                 "policy_sha256": self.policy["sha256"],
                 "embedding_manifest_sha256": self.policy["embedding_manifest_sha256"],
                 "created": now, "sequence": 0, "events": 0, "sensitive_reads": 0,
                 "last_text_sha256": None, "repeats": 0, "incomplete": False}
        # An existing session cannot be reset by a new benign-looking turn.
        with self.db:
            self.db.execute("INSERT INTO sessions VALUES (?,?)", (key, canonical(state)))
        return key

    def record(self, key, event, *, now):
        """event contains server-observed IDs and gateway-measured usage.

        turn: embedding + embedding_manifest_sha256 + content_sha256;
        sensitive_read: resource_id resolved in the data catalog;
        action: resolved tool_id + destination_id;
        context: gateway-measured context_tokens.
        All kinds carry event_id, sequence and task_sha256.
        """
        event_hash = digest(event)
        self.db.execute("BEGIN IMMEDIATE")
        try:
            prior = self.db.execute("SELECT digest,result FROM events WHERE key=? AND id=?",
                                    (key, event["event_id"])).fetchone()
            if prior:
                if prior[0] != event_hash:
                    raise ValueError("event identity reused with different bytes")
                self.db.rollback()
                return json.loads(prior[1])
            row = self.db.execute("SELECT state FROM sessions WHERE key=?", (key,)).fetchone()
            if not row:
                raise ValueError("no approved session baseline")
            state = json.loads(row[0])
            if (state["policy_sha256"] != self.policy["sha256"]
                    or state["embedding_manifest_sha256"] != self.policy["embedding_manifest_sha256"]
                    or event["task_sha256"] != state["task"]["task_sha256"]
                    or not state["created"] <= now < state["created"] + self.policy["ttl_seconds"]):
                raise ValueError("stale or mismatched task/profile")
            if type(event["sequence"]) is not int or event["sequence"] <= state["sequence"]:
                raise ValueError("late or reordered event requires ledger reconciliation")
            if event["sequence"] != state["sequence"] + 1:
                state["incomplete"] = True
            if state["events"] >= self.policy["maximum_events"]:
                raise ValueError("session population bound reached; finalize and reconcile")
            findings, signals = [], {}
            kind = event["kind"]
            if kind == "turn":
                if event["embedding_manifest_sha256"] != self.policy["embedding_manifest_sha256"]:
                    raise ValueError("embedding release mismatch")
                drift = 1 - cosine(state["baseline"], event["embedding"])
                signals["intent_drift"] = drift
                if drift >= self.policy["drift_threshold"]:
                    findings.append("intent_drift")
                state["repeats"] = state["repeats"] + 1 if state["last_text_sha256"] == event["content_sha256"] else 1
                state["last_text_sha256"] = event["content_sha256"]
                if state["repeats"] >= self.policy["repeat_threshold"]:
                    findings.append("repeated_input_sequence")
            elif kind == "sensitive_read":
                classification = self.catalog.get((state["tenant"], event["resource_id"]))
                if classification is None:
                    state["incomplete"] = True
                elif classification == "sensitive":
                    state["sensitive_reads"] += 1
                    if state["sensitive_reads"] >= self.policy["sensitive_read_threshold"]:
                        findings.append("sensitive_read_accumulation")
            elif kind == "action":
                if event["tool_id"] not in state["task"]["allowed_tools"]:
                    findings.append("tool_outside_approved_task")
                destination = event["destination_id"]
                if destination and destination not in state["task"]["allowed_destinations"]:
                    findings.append("destination_outside_approved_task")
                    if state["sensitive_reads"]:
                        findings.append("sensitive_read_to_external_destination")
            elif kind == "context":
                if type(event["context_tokens"]) is not int or event["context_tokens"] < 0:
                    raise ValueError("context usage is invalid")
                if event["context_tokens"] >= self.policy["context_token_threshold"]:
                    findings.append("context_saturation")
            else:
                raise ValueError("unsupported verified telemetry kind")
            state["sequence"], state["events"] = event["sequence"], state["events"] + 1
            result = {"schema_version": "aidefend.session-risk.v2", "control": "AID-D-003.005",
                      "session_key": key, "event_id": event["event_id"],
                      "policy_version": self.policy["version"], "policy_sha256": self.policy["sha256"],
                      "embedding_manifest_sha256": self.policy["embedding_manifest_sha256"],
                      "task_sha256": event["task_sha256"], "source_event_sha256": event_hash,
                      "sequence": state["sequence"], "event_count": state["events"],
                      "coverage": "INSUFFICIENT_DATA" if state["incomplete"] else "PASS",
                      "finding_status": "FINDING" if findings else "UNKNOWN" if state["incomplete"] else "NO_FINDING",
                      "signals": signals, "findings": findings, "ledger_sha256": digest(state)}
            self.db.execute("UPDATE sessions SET state=? WHERE key=?", (canonical(state), key))
            self.db.execute("INSERT INTO events VALUES (?,?,?,?)", (key, event["event_id"], event_hash, canonical(result)))
            self.db.commit()
            return result
        except Exception:
            self.db.rollback()
            raise

# Bootstrap with sqlite3.connect(path, isolation_level=None, timeout=<policy>).
# Place the DB outside agent/user writable storage. The service wrapper maps
# processing failures to ERROR and emits them to SIEM; it never returns a clean
# score on invalid input. Retention deletes a closed session and its event rows
# together only after signed telemetry export and independent reconciliation.

def observe_session(monitor, key, event, *, now):
    """The telemetry service signs/exports either measured output or ERROR."""
    try:
        return monitor.record(key, event, now=now)
    except Exception as error:
        return {"schema_version": "aidefend.session-risk.v2", "control": "AID-D-003.005",
                "session_key": key, "policy_version": monitor.policy["version"],
                "policy_sha256": monitor.policy["sha256"], "coverage": "ERROR",
                "finding_status": "UNKNOWN", "findings": [],
                "reason_code": type(error).__name__}
Interpret and verify the result

Intent drift is a probabilistic warning, not proof of injection. Task-scope violations and sensitive-read-to-unapproved-destination sequences are independently reported invariants. PASS describes measured coverage, while FINDING/NO_FINDING describes those measured signals; missing telemetry is INSUFFICIENT_DATA and a processing failure is ERROR. Sign and export every result, retaining its task, model, policy and source-event binding. Reconstruct the complete gateway sequence independently; replay benign topic changes, an approved external destination, drift, sensitive-read accumulation followed by exfiltration, duplicate/conflicting events, sequence gaps, late events, zero vectors, profile changes, expiry and storage failures. These findings do not authorize or block tools; H-018 and Isolate own enforcement.

Verify the cross-turn boundary

Replay an approved mailbox task that reads a synthetic sensitive record, processes intervening ordinary turns, then attempts a deferred action toward an unapproved destination. The later event must retain the earlier read and emit sensitive_read_to_external_destination. Replay an out-of-task tool action with otherwise normal text and require tool_outside_approved_task. These signals expose a bounded behavioral deviation, not the hidden instruction, its trigger, an attacker motive, a changed credential, or every form of rogue behavior. Missing telemetry remains incomplete coverage; this monitor does not prove privilege escalation.

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-003.005: Stateful Session Monitoring: Intent Drift + Invariant-Breach Signals | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service