AID-D-007.002: Realtime Multimodal Stream & Session Anomaly Detection
Pillars: data, app, model
Lifecycle: operation
Scope boundary
Owns ephemeral live-session findings from voice activity, ASR confidence, transcript churn, OCR and frame deltas, source switches, interruption patterns, synthetic-signal indicators, and proposed-action timing before those signals drive privileged behavior.
AID-D-007.001: AID-D-007.002 analyzes realtime camera, microphone, call, and screen-share sessions; AID-D-007.001 analyzes complete non-realtime files and artifact batches. Live timing and incremental changes may disappear after capture, while batch analysis can perform deeper replay over stable bytes.
AID-D-001.002: AID-D-007.002 owns live multimodal session-risk signals and source-change timing; AID-D-001.002 owns forensic synthetic-media and manipulation analysis. A suspected synthetic signal may feed live risk scoring, but forensic authenticity remains a separate claim.
AID-H-002.008: AID-D-007.002 emits realtime multimodal risk findings; AID-H-002.008 owns preventive stream admission, gating, and action-time enforcement. The detector must not silently become the stream gate.
Implementation guidance
- AID-D-007.002-G001: Emit realtime audio-turn findings from ASR confidence, voice-activity boundaries, transcript deltas, spoken-instruction markers, and high-risk action context.
- AID-D-007.002-G002: Correlate realtime transcript, OCR, and visual-context deltas with proposed tool actions so short-lived stream injections can be detected before high-impact execution.
- AID-D-007.002-G003: Maintain session-level baselines for realtime stream sources, speaker/source changes, barge-in frequency, screen-share context switches, and modality dropout patterns.
AID-D-007.002-G001
Production boundary
Finalize turns from authenticated ASR/VAD events before scoring them. The signed session ledger is authoritative for the eligible turn population; an adapter result cannot nominate its own scope. Preserve transcript revisions and uncertainty, and bind every result to the exact participant, source, sequence interval, policy digest, and adapter digest. A spoken-instruction feature is never a verdict by itself, and this detector does not authorize or block the proposed action.
Turn-population evaluator
# File: multimodal/score_realtime_turns.py
from __future__ import annotations
import argparse
import hashlib
import json
import math
from pathlib import Path
TURN_FIELDS = {"turn_id", "participant_id", "source_id", "first_sequence", "last_sequence",
"transcript_revision_count", "asr_confidence", "transcript_churn",
"spoken_instruction_score", "high_risk_action", "policy_sha256", "adapter_sha256", "observation_sha256"}
def strict_json(raw: bytes) -> object:
def unique(pairs):
value = {}
for key, item in pairs:
if key in value:
raise ValueError("duplicate_json_key")
value[key] = item
return value
return json.loads(raw.decode("utf-8", errors="strict"), object_pairs_hook=unique,
parse_constant=lambda value: (_ for _ in ()).throw(ValueError("nonfinite_json")))
def unit(value: object, field: str) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError(field + "_type")
number = float(value)
if not math.isfinite(number) or not 0.0 <= number <= 1.0:
raise ValueError(field + "_range")
return number
def evaluate(policy: dict, ledger: dict, rows: object, policy_sha256: str) -> dict:
if set(policy) != {"schema_version", "policy_version", "adapter_sha256", "thresholds"}:
raise ValueError("policy_schema")
if set(ledger) != {"schema_version", "session_id", "agent_id", "eligible_turns"}:
raise ValueError("ledger_schema")
if policy["schema_version"] != "aidefend.realtime-turn-policy.v1" or ledger["schema_version"] != "aidefend.realtime-turn-ledger.v2":
raise ValueError("schema_version")
thresholds = policy["thresholds"]
if not isinstance(thresholds, dict) or set(thresholds) != {"min_asr_confidence", "max_transcript_churn", "spoken_instruction_min", "max_revisions"}:
raise ValueError("threshold_schema")
min_confidence = unit(thresholds["min_asr_confidence"], "min_asr_confidence")
max_churn = unit(thresholds["max_transcript_churn"], "max_transcript_churn")
instruction_min = unit(thresholds["spoken_instruction_min"], "spoken_instruction_min")
max_revisions = thresholds["max_revisions"]
if isinstance(max_revisions, bool) or not isinstance(max_revisions, int) or max_revisions < 0:
raise ValueError("max_revisions")
expected = {}
if not isinstance(ledger["eligible_turns"], list):
raise ValueError("turn_population")
for turn in ledger["eligible_turns"]:
if not isinstance(turn, dict) or set(turn) != {"turn_id", "participant_id", "source_id", "observation_sha256"}:
raise ValueError("eligible_turn_schema")
turn_id = turn["turn_id"]
if not isinstance(turn_id, str) or not turn_id or turn_id in expected:
raise ValueError("eligible_turn_identity")
expected[turn_id] = turn
if not isinstance(rows, list):
raise ValueError("observation_population")
measured = {}
for row in rows:
if not isinstance(row, dict) or set(row) != TURN_FIELDS:
raise ValueError("turn_observation_schema")
turn_id = row["turn_id"]
if turn_id in measured or turn_id not in expected:
raise ValueError("duplicate_or_extra_turn")
binding = expected[turn_id]
if (row["participant_id"] != binding["participant_id"] or row["source_id"] != binding["source_id"]
or row["policy_sha256"] != policy_sha256 or row["adapter_sha256"] != policy["adapter_sha256"]
or row["observation_sha256"] != binding["observation_sha256"]):
raise ValueError("turn_binding")
first_sequence, last_sequence = row["first_sequence"], row["last_sequence"]
revisions = row["transcript_revision_count"]
if (isinstance(first_sequence, bool) or not isinstance(first_sequence, int) or first_sequence < 0
or isinstance(last_sequence, bool) or not isinstance(last_sequence, int) or last_sequence < first_sequence
or isinstance(revisions, bool) or not isinstance(revisions, int) or revisions < 0):
raise ValueError("turn_sequence")
confidence = unit(row["asr_confidence"], "asr_confidence")
churn = unit(row["transcript_churn"], "transcript_churn")
instruction = unit(row["spoken_instruction_score"], "spoken_instruction_score")
if not isinstance(row["high_risk_action"], bool):
raise ValueError("high_risk_action")
insufficient = confidence < min_confidence or revisions > max_revisions
finding = not insufficient and row["high_risk_action"] and (churn >= max_churn or instruction >= instruction_min)
measured[turn_id] = {"observation_sha256": row["observation_sha256"], "first_sequence": first_sequence, "last_sequence": last_sequence,
"transcript_revision_count": revisions, "asr_confidence": confidence,
"transcript_churn": churn, "spoken_instruction_score": instruction,
"high_risk_action": row["high_risk_action"], "insufficient": insufficient,
"finding": finding}
missing = sorted(set(expected) - set(measured))
insufficient = sorted(turn_id for turn_id, row in measured.items() if row["insufficient"])
findings = sorted(turn_id for turn_id, row in measured.items() if row["finding"])
outcome = "FAIL" if findings else "INSUFFICIENT_DATA" if missing or insufficient else "PASS"
return {"schema_version": "aidefend.realtime-turn-finding.v2", "control_id": "AID-D-007.002",
"session_id": ledger["session_id"], "agent_id": ledger["agent_id"],
"policy_version": policy["policy_version"], "policy_sha256": policy_sha256,
"required_turn_ids": sorted(expected), "measured_turn_ids": sorted(measured),
"missing_turn_ids": missing, "insufficient_turn_ids": insufficient,
"finding_turn_ids": findings, "turn_features": measured, "outcome": outcome}
def main() -> None:
parser = argparse.ArgumentParser()
for name in ("policy", "ledger", "turns", "out"):
parser.add_argument("--" + name, type=Path, required=True)
args = parser.parse_args()
raw = {name: getattr(args, name).read_bytes() for name in ("policy", "ledger", "turns")}
try:
result = evaluate(strict_json(raw["policy"]), strict_json(raw["ledger"]), strict_json(raw["turns"]), hashlib.sha256(raw["policy"]).hexdigest())
except Exception as error:
result = {"schema_version": "aidefend.realtime-turn-finding.v2", "control_id": "AID-D-007.002",
"input_sha256": {name: hashlib.sha256(value).hexdigest() for name, value in sorted(raw.items())},
"outcome": "ERROR", "reason_code": type(error).__name__}
temporary = args.out.with_suffix(args.out.suffix + ".tmp")
temporary.write_text(json.dumps(result, sort_keys=True, separators=(",", ":"), allow_nan=False) + chr(10), encoding="utf-8")
temporary.replace(args.out)
if result["outcome"] == "ERROR":
raise SystemExit(2)
if __name__ == "__main__":
main()Verification and failure handling
Verify and privately snapshot the signed policy, signed session ledger, and signed adapter observations before running this evaluator; sign its output even when it exits nonzero. An independent runner reconstructs turns from encrypted samples or sufficient signed derived evidence and replays normal, high-risk instruction, low-confidence, transcript-churn, revision-overflow, source-mismatch, duplicate, missing, extra, and adapter-drift fixtures. Reconcile every finalized eligible turn against the authoritative session ledger and preserve event-time ordering evidence. Source continuity and session-baseline deviations are owned by G003; transient OCR/visual-to-action correlation is owned by G002.
The v2 eligible-turn ledger captures observation_sha256 before feature extraction, using the canonical input object defined by the H002 gate. Every signed feature row and resulting turn feature binds that same digest. For revised input, create a new observation identity or finalize the revised observation before scoring; never attach an older score to new bytes.
AID-D-007.002-G002
Production implementation
Consume immutable, signature-verified modality events and action proposals. Validate producer, scope, freshness, and sequence; correlate only events within the signed causal window; and emit exact transient-instruction and context-to-action edges. Publication occurs before execution when feasible, but this detector never grants, denies, or rewrites an action.
# File: multimodal/correlate_stream_actions.py
from __future__ import annotations
import hashlib
import json
import re
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
SHA256 = re.compile(r"^[a-f0-9]{64}$")
EVENT_FIELDS = {
"schema_version", "event_id", "session_id_sha256", "producer_identity",
"modality", "sequence", "event_time", "content_sha256",
"instruction_markers", "proposed_action",
}
@dataclass(frozen=True)
class VerifiedCorrelationPolicy:
policy_sha256: str
session_id_sha256: str
expected_producers: dict[str, str]
expected_action_event_ids: tuple[str, ...]
causal_window_ms: int
allowed_clock_skew_ms: int
high_impact_actions: frozenset[str]
def when(value: str) -> datetime:
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.tzinfo is None:
raise ValueError("event timestamp lacks timezone")
return parsed.astimezone(timezone.utc)
def canonical_sha256(value: object) -> str:
return hashlib.sha256(
json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()
).hexdigest()
def correlate(
policy: VerifiedCorrelationPolicy,
verified_events: list[dict[str, object]],
) -> dict[str, object]:
if (
not SHA256.fullmatch(policy.policy_sha256)
or not SHA256.fullmatch(policy.session_id_sha256)
or not policy.expected_producers
or "action_proposal" not in policy.expected_producers
or not policy.expected_action_event_ids
or len(set(policy.expected_action_event_ids)) != len(policy.expected_action_event_ids)
or isinstance(policy.causal_window_ms, bool)
or not isinstance(policy.causal_window_ms, int) or policy.causal_window_ms < 1
or isinstance(policy.allowed_clock_skew_ms, bool)
or not isinstance(policy.allowed_clock_skew_ms, int)
or policy.allowed_clock_skew_ms < 0
or not policy.high_impact_actions
):
raise ValueError("verified correlation policy is invalid")
events = []
event_ids = set()
sequence_keys = set()
for event in verified_events:
if not isinstance(event, dict) or set(event) != EVENT_FIELDS:
raise ValueError("stream event schema differs")
modality, sequence = event["modality"], event["sequence"]
if (
event["schema_version"] != "aidefend.multimodal-stream-event.v1"
or event["session_id_sha256"] != policy.session_id_sha256
or modality not in policy.expected_producers
or event["producer_identity"] != policy.expected_producers[modality]
or not event["event_id"] or event["event_id"] in event_ids
or isinstance(sequence, bool) or not isinstance(sequence, int) or sequence < 0
or (modality, sequence) in sequence_keys
or not SHA256.fullmatch(str(event["content_sha256"]))
or not isinstance(event["instruction_markers"], list)
or any(not isinstance(value, str) or not value for value in event["instruction_markers"])
):
raise ValueError("stream event binding or value is invalid")
if modality == "action_proposal":
if (
event["event_id"] not in policy.expected_action_event_ids
or event["proposed_action"] not in policy.high_impact_actions
or event["instruction_markers"]
):
raise ValueError("action proposal is not policy-bound")
elif event["proposed_action"] is not None:
raise ValueError("non-action event contains proposed_action")
event_ids.add(event["event_id"])
sequence_keys.add((modality, sequence))
events.append({**event, "_time": when(str(event["event_time"]))})
proposals = {
event["event_id"]: event
for event in events
if event["modality"] == "action_proposal"
}
missing_proposals = sorted(set(policy.expected_action_event_ids) - set(proposals))
required_modalities = set(policy.expected_producers) - {"action_proposal"}
correlations = []
for proposal_id in sorted(proposals):
proposal = proposals[proposal_id]
lower = proposal["_time"] - timedelta(milliseconds=policy.causal_window_ms)
upper = proposal["_time"] + timedelta(milliseconds=policy.allowed_clock_skew_ms)
window_events = {}
for modality in sorted(required_modalities):
candidates = [
event for event in events
if event["modality"] == modality and lower <= event["_time"] <= upper
]
if candidates:
window_events[modality] = sorted(
candidates, key=lambda item: (item["_time"], item["sequence"], item["event_id"])
)
missing_modalities = sorted(required_modalities - set(window_events))
edges = [
{
"source_event_id": event["event_id"],
"source_modality": modality,
"instruction_markers": sorted(set(event["instruction_markers"])),
"action_event_id": proposal_id,
"proposed_action": proposal["proposed_action"],
"delta_ms": int((proposal["_time"] - event["_time"]).total_seconds() * 1000),
}
for modality in sorted(window_events)
for event in window_events[modality]
if event["instruction_markers"]
]
outcome = "FAIL" if edges else ("INSUFFICIENT_DATA" if missing_modalities else "PASS")
correlations.append({
"action_event_id": proposal_id,
"proposed_action": proposal["proposed_action"],
"source_event_ids": sorted(
event["event_id"]
for modality in sorted(window_events)
for event in window_events[modality]
),
"missing_modalities": missing_modalities,
"edges": edges,
"outcome": outcome,
})
aggregate_outcome = (
"FAIL" if any(item["outcome"] == "FAIL" for item in correlations)
else "INSUFFICIENT_DATA"
if missing_proposals or any(item["outcome"] == "INSUFFICIENT_DATA" for item in correlations)
else "PASS"
)
result = {
"schema_version": "aidefend.multimodal-action-correlation.v1",
"control": "AID-D-007.002",
"session_id_sha256": policy.session_id_sha256,
"policy_sha256": policy.policy_sha256,
"expected_action_event_ids": sorted(policy.expected_action_event_ids),
"missing_action_event_ids": missing_proposals,
"correlations": correlations,
"outcome": aggregate_outcome,
}
return {**result, "correlation_sha256": canonical_sha256(result)}Independent verification
An independent verifier reorders and replays the signed event ledger with valid, transient-injection, stale, missing-modality, clock-skew, duplicate-sequence, and action-mismatch fixtures; recomputes every correlation edge and digest; and reconciles the exact policy-selected high-impact proposal population. The result schema contains detector outcomes and evidence only—never an execution authorization.
AID-D-007.002-G003
Production implementation
Maintain durable event-time state and high-water marks; measure declared continuity, source/role changes, turn cadence, barge-in, context switches and modality dropout for every finalized turn/window; preserve late/revised event handling; and emit exact deviation evidence. The detector does not authenticate a person, restore a baseline after attacker-controlled healthy events, or terminate a session.
Independent verification
An independent verifier reconstructs sessions from the append-only event ledger, replays normal, source-change, dropout, barge-in, late-event, restart and state-corruption fixtures, recomputes windows and thresholds, and reconciles every eligible session and finalized turn.
Publish a live observation finding for H-002.008
Save the module below as multimodal/stream_findings.py. For audio, construct a single-entry G001 eligible-turn ledger with observation_id audio:<turn_index>. For a finalized visual/OCR observation, use visual:<frame_index> and the separately calibrated visual adapter/profile: confidence, instruction-bearing content, source continuity and revisions must describe that exact visual observation. Capture the input before scoring; do not let a request supply features, policy or the ledger. An absent visual adapter is INSUFFICIENT_DATA, not a clean frame. G002 remains an action-time correlation method and is not substituted for a per-frame finding. Sign the canonical adapter result under the registered detector identity; the H002 receiver verifies that signature, audience, freshness and input binding before passing the parsed object to the gate.
# File: multimodal/stream_findings.py
from __future__ import annotations
import hashlib
import json
import re
HEX64 = re.compile(r"^[0-9a-f]{64}$")
FIELDS = {"schema_version", "control", "status", "severity", "session_id",
"observation_id", "modality", "policy_version", "policy_sha256",
"observation_sha256", "evidence_sha256", "findings"}
def digest(value):
return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":"),
ensure_ascii=False, allow_nan=False).encode()).hexdigest()
def stream_finding(batch, *, observation, observation_id, modality,
expected_session_id, policy_version, policy_sha256):
"""Called inside the trusted detector after G001 evaluates this observation.
The adapter constructs the single-entry eligible ledger from server state,
captures observation bytes BEFORE scoring, and signs this resulting record.
A batch for other turns or a different policy cannot stand in for this input.
"""
if modality not in {"audio", "visual"} or not observation_id:
raise ValueError("unsupported observation binding")
if not isinstance(policy_version, str) or not policy_version or not HEX64.fullmatch(policy_sha256):
raise ValueError("invalid verified policy")
if (batch.get("schema_version") != "aidefend.realtime-turn-finding.v2"
or batch.get("control_id") != "AID-D-007.002"
or batch.get("session_id") != expected_session_id
or batch.get("policy_version") != policy_version
or batch.get("policy_sha256") != policy_sha256):
raise ValueError("detector batch identity/policy mismatch")
features = batch.get("turn_features", {})
if batch.get("outcome") != "ERROR" and observation_id in features:
if features[observation_id].get("observation_sha256") != digest(observation):
raise ValueError("detector measured different observation bytes")
required = batch.get("required_turn_ids", [])
measured = batch.get("measured_turn_ids", [])
found = batch.get("finding_turn_ids", [])
missing = batch.get("missing_turn_ids", [])
insufficient = batch.get("insufficient_turn_ids", [])
outcome = batch.get("outcome")
if outcome == "ERROR":
status = "ERROR"
elif required != [observation_id]:
raise ValueError("live admission requires the exact observation population")
elif observation_id in found and observation_id in measured:
status = "FINDING"
elif (outcome == "INSUFFICIENT_DATA" or missing or insufficient
or measured != [observation_id]):
status = "INSUFFICIENT_DATA"
elif outcome == "PASS" and not found:
status = "NO_FINDING"
else:
raise ValueError("detector finding state is inconsistent")
return {
"schema_version": "aidefend.stream-admission-finding.v1",
"control": "AID-D-007.002", "status": status,
"severity": "high" if status == "FINDING" else "none",
"session_id": expected_session_id, "observation_id": observation_id,
"modality": modality, "policy_version": policy_version,
"policy_sha256": policy_sha256, "observation_sha256": digest(observation),
"evidence_sha256": digest(batch),
"findings": [batch["turn_features"][observation_id]] if status == "FINDING" else [],
}
def admission_reasons(finding, *, observation, observation_id, modality,
session_id, policy_version, policy_sha256):
"""Signature/producer verification is performed by the receiving adapter.
Receive this object only from the authenticated detector output, never from
the upload body. This function checks binding and decision consistency.
"""
if not isinstance(finding, dict) or set(finding) != FIELDS:
return ["d007_finding_schema_invalid"]
if (finding["schema_version"] != "aidefend.stream-admission-finding.v1"
or finding["control"] != "AID-D-007.002"
or finding["session_id"] != session_id
or finding["observation_id"] != observation_id
or finding["modality"] != modality
or finding["policy_version"] != policy_version
or finding["policy_sha256"] != policy_sha256
or finding["observation_sha256"] != digest(observation)
or not HEX64.fullmatch(str(finding["evidence_sha256"]))):
return ["d007_finding_binding_invalid"]
status = finding["status"]
if status in {"ERROR", "INSUFFICIENT_DATA"}:
return ["d007_finding_unavailable"]
if (status not in {"NO_FINDING", "FINDING"}
or not isinstance(finding["findings"], list)
or (status == "NO_FINDING" and (finding["findings"] or finding["severity"] != "none"))
or (status == "FINDING" and (not finding["findings"] or finding["severity"] != "high"))):
return ["d007_finding_state_invalid"]
return ["d007_high_risk_stream_finding"] if status == "FINDING" else []Replay a measured benign observation, a high-risk finding, missing features, a different session/index, altered OCR/transcript and a stale policy. The producer and consumer must agree on every state. A full batch PASS covering different observations cannot authorize the current input.
Open-source tools
- FFmpeg
- OpenCV
- Tesseract OCR
- WebRTC VAD
- Silero VAD (speech-activity segmentation for downstream stream-risk correlation)
- librosa
- OpenTelemetry
- Prometheus
Source-available / open-weight tools
- OpenAI Whisper (MIT; open-weight)
Commercial / hosted tools
- Azure AI Speech
- Amazon Transcribe
- Google Cloud Speech-to-Text
- Deepgram
- AssemblyAI
- Pindrop Pulse for Meetings
- Azure AI Content Safety
- Amazon Rekognition
- Google Cloud Vision API
Threat mappings
MITRE ATLAS
- AML.T0051 LLM Prompt Injection (realtime audio, video, or screen-share prompt injection)
- AML.T0051.001 LLM Prompt Injection: Indirect (live external streams can inject instructions through transcript or OCR paths)
- AML.T0052.001 Phishing: Deepfake-Assisted Phishing
- AML.T0043 Craft Adversarial Data (adversarial audio or visual stream content crafted to manipulate recognition or intent)
- AML.T0015 Evade AI Model (audio/video perturbations can evade transcription, OCR, or moderation)
- AML.T0129 Triggers in Multimodal Inputs (correlates transient audio and visual instructions with proposed agent actions)
MAESTRO
- Input Validation Attacks (L3)
- Adversarial Examples (L1)
- Evasion of Detection (L5)
OWASP LLM Top 10 2026
- LLM01:2026 Prompt Injection
OWASP ML Top 10 2023
- ML01:2023 Input Manipulation Attack
OWASP Top 10 for Agentic Applications 2026
- ASI01:2026 Agent Goal Hijack
NIST Adversarial Machine Learning 2025
- NISTAML.018 Prompt Injection
- NISTAML.022 Evasion
- NISTAML.015 Indirect Prompt Injection
Cisco Integrated AI Security and Safety Framework
- AISubtech-1.3.1 Image-Text Injection (transient spoken or displayed instructions and source-continuity signals)
- AISubtech-1.3.3 Audio Command Injection (transient spoken or displayed instructions and source-continuity signals)
- AISubtech-1.3.4 Video Overlay Manipulation (transient spoken or displayed instructions and source-continuity signals)
- AISubtech-17.1.1 Sensor Spoofing: Action Signals (audio, visual) (transient spoken or displayed instructions and source-continuity signals)
- AITech-1.3 Multimodal Injection and Manipulation (transient spoken or displayed instructions and source-continuity signals)
- AITech-17.1 Sensor Spoofing (transient spoken or displayed instructions and source-continuity signals)
Google Secure AI Framework 2.0 - Risks
- PIJ: Prompt Injection
- MEV: Model Evasion
Databricks AI Security Framework 3.0
- Model Serving - Inference requests 9.1: Prompt inject
- Model Serving - Inference requests 9.9: Input Resource Control
- Agents - Core 13.6: Intent Breaking & Goal Manipulation