Scope boundary

Owns SIEM-grade findings for abnormal memory-write behavior, repeated payloads, namespace violations, verified-read failures, and quarantine-pipeline health. It does not contain the writer or cryptographically accept or reject recalled memory.

AID-D-001.005: AID-D-003.006 monitors write-pattern abuse and integrity or quarantine failures across the memory system; AID-D-001.005 scans the content of each recalled record before prompt rehydration. Systemic write abuse and semantic danger in one recalled item require different observation points.

AID-I-004.003: AID-D-003.006 alerts on read-verification failures and suspicious memory activity; AID-I-004.003 performs cryptographic signed-write and verified-read enforcement. Monitoring a verification failure does not reject the record, while enforcement does not replace trend and campaign detection.

Implementation guidance

• AID-D-003.006-G001: Emit SIEM-grade telemetry for abnormal memory write behavior.
• AID-D-003.006-G002: Monitor read-path integrity failures and quarantine-pipeline health signals.

AID-D-003.006-G001

Observe committed writes, not caller claims

Instrument the durable memory-store adapter after it has a committed write result. Its receipt verifier supplies authenticated tenant, actor, namespace, write ID, record version, commit time, and value digest. The detector does not accept these fields from an agent request. Persist every receipt before evaluating bursts so restarts, duplicates, and concurrent workers cannot reset or inflate state.

Policy-driven event-time detector

from __future__ import annotations

# File: detection/memory_write_detector.py

import hashlib
import hmac
import math
import re
from collections.abc import Callable
from dataclasses import dataclass
from typing import Protocol


@dataclass(frozen=True)
class VerifiedMemoryWrite:
    write_id: str
    tenant_id: str
    authenticated_actor_id: str
    session_id: str
    namespace: str
    record_id: str
    record_version: str
    committed_at_unix: float
    value_sha256: str
    receipt_sha256: str


@dataclass(frozen=True)
class MemoryWritePolicy:
    policy_version: str
    window_seconds: int
    maximum_writes_per_window: int
    repeated_value_count: int
    shared_namespace_prefixes: tuple[str, ...]


@dataclass(frozen=True)
class StateObservation:
    inserted: bool
    actor_window_count: int
    repeated_value_count: int


class StateStore(Protocol):
    def record_and_count(
        self,
        *,
        write: VerifiedMemoryWrite,
        value_ref: str,
        window_seconds: int,
    ) -> StateObservation: ...


VerifyReceipt = Callable[[bytes], VerifiedMemoryWrite]
AppendEvent = Callable[[dict], str]
HEX64 = re.compile(r"^[0-9a-f]{64}$")


def _is_sha256(value: object) -> bool:
    return isinstance(value, str) and HEX64.fullmatch(value) is not None


def _positive_integer(value: object, name: str, *, minimum: int = 1) -> int:
    if isinstance(value, bool) or not isinstance(value, int) or value < minimum:
        raise ValueError(name + " is invalid")
    return value


def evaluate_committed_write(
    *,
    signed_receipt: bytes,
    verify_receipt: VerifyReceipt,
    state: StateStore,
    policy: MemoryWritePolicy,
    evidence_hmac_key: bytes,
    append_event: AppendEvent,
) -> dict:
    write = verify_receipt(signed_receipt)
    strings = (
        write.write_id, write.tenant_id, write.authenticated_actor_id,
        write.session_id, write.namespace, write.record_id,
        write.record_version, policy.policy_version,
    )
    if any(not isinstance(value, str) or not value for value in strings):
        raise ValueError("verified memory-write binding is incomplete")
    window_seconds = _positive_integer(policy.window_seconds, "window_seconds")
    maximum_writes = _positive_integer(
        policy.maximum_writes_per_window, "maximum_writes_per_window"
    )
    repeated_threshold = _positive_integer(
        policy.repeated_value_count, "repeated_value_count", minimum=2
    )
    if (
        isinstance(write.committed_at_unix, bool)
        or not isinstance(write.committed_at_unix, (int, float))
        or not math.isfinite(float(write.committed_at_unix))
        or not _is_sha256(write.value_sha256)
        or not _is_sha256(write.receipt_sha256)
    ):
        raise ValueError("write receipt time or digest is malformed")
    prefixes = tuple(prefix.rstrip("/") for prefix in policy.shared_namespace_prefixes)
    if (
        len(prefixes) != len(set(prefixes))
        or any(not isinstance(prefix, str) or not prefix for prefix in prefixes)
    ):
        raise ValueError("shared namespace policy is malformed")
    if not isinstance(evidence_hmac_key, bytes) or len(evidence_hmac_key) < 32:
        raise ValueError("evidence HMAC key must contain at least 256 bits")

    value_ref = hmac.new(
        evidence_hmac_key,
        (write.tenant_id + ":" + write.value_sha256).encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()
    observation = state.record_and_count(
        write=write, value_ref=value_ref, window_seconds=window_seconds
    )
    if not isinstance(observation.inserted, bool):
        raise RuntimeError("state store returned an invalid insertion state")
    actor_count = _positive_integer(
        observation.actor_window_count, "actor_window_count"
    )
    repeated_count = _positive_integer(
        observation.repeated_value_count, "repeated_value_count"
    )

    shared = any(
        write.namespace == prefix
        or write.namespace.startswith(prefix + "/")
        for prefix in prefixes
    )
    reasons = []
    if actor_count > maximum_writes:
        reasons.append("actor_write_burst")
    if repeated_count >= repeated_threshold:
        reasons.append("repeated_value")
    if shared:
        reasons.append("shared_namespace_write")
    event = {
        "schema_version": "aidefend.memory_write_signal.v2",
        "control_status": "PASS",
        "finding_status": "FINDING" if reasons else "NO_FINDING",
        "write_id": write.write_id,
        "tenant_id": write.tenant_id,
        "authenticated_actor_id": write.authenticated_actor_id,
        "session_id": write.session_id,
        "namespace": write.namespace,
        "record_id": write.record_id,
        "record_version": write.record_version,
        "committed_at_unix": float(write.committed_at_unix),
        "value_hmac_sha256": value_ref,
        "receipt_sha256": write.receipt_sha256,
        "policy_version": policy.policy_version,
        "window_seconds": window_seconds,
        "actor_window_count": actor_count,
        "repeated_value_count": repeated_count,
        "reasons": reasons,
    }
    acknowledgement_id = append_event(event)
    if not acknowledgement_id:
        raise RuntimeError("memory-write evidence sink did not acknowledge the event")
    return {
        **event,
        "acknowledgement_id": acknowledgement_id,
        "duplicate_replay": not observation.inserted,
    }

State-store and verification contract

record_and_count must be an atomic, durable, idempotent event-time operation keyed by tenant and write ID. On a duplicate it returns the original persisted counts, not a fabricated clean result. append_event is idempotent by write ID and returns the existing acknowledgement for an exact retry; a conflicting retry is an error. Redis, SQL, or a stream processor may implement those adapters without changing this detector. Replay restart, concurrent writers, duplicate IDs, late events, cross-tenant identities, shared namespaces, repeated values, receipt failure, and sink outage. Rebuild counts from the durable write ledger and compare every expected write ID; an absent terminal event is a coverage defect.

AID-D-003.006-G002

Concept

Read-path monitoring should report the signed verifier and quarantine outcomes that actually occurred, rather than accepting caller-supplied booleans. Count each authenticated read candidate exactly once before deriving a ratio; keep quarantine lifecycle events in a separate metric so promotion events cannot distort the read denominator.

Instrument the closed outcome contract

# File: detection/read_path_metrics.py
from __future__ import annotations

import base64
import hashlib
import json
import os
import re
from pathlib import Path

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from cryptography.hazmat.primitives.serialization import load_pem_public_key
from prometheus_client import Counter

FIELDS = {
    "schema_version", "receipt_id", "event_kind", "candidate_id",
    "memory_revision", "state", "policy_version", "policy_sha256", "observed_at",
}
READ_STATES = {"VERIFIED_ACTIVE", "REJECTED_INTEGRITY", "QUARANTINED", "ERROR"}
QUARANTINE_STATES = {"SUBMITTED", "PROMOTED"}
SHA256 = re.compile(r"^[a-f0-9]{64}$")

read_path_outcome_total = Counter(
    "aidefend_memory_read_path_outcome_total",
    "Authenticated verifier outcomes for retrieval candidates",
    ("state", "policy_version"),
)
quarantine_transition_total = Counter(
    "aidefend_memory_quarantine_transition_total",
    "Authenticated quarantine lifecycle transitions",
    ("state", "policy_version"),
)


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


def _load_pinned_key() -> Ed25519PublicKey:
    raw = Path(os.environ["D003006_VERIFIER_PUBLIC_KEY"]).read_bytes()
    expected = os.environ["D003006_VERIFIER_PUBLIC_KEY_SHA256"]
    if not SHA256.fullmatch(expected) or hashlib.sha256(raw).hexdigest() != expected:
        raise RuntimeError("verifier public-key digest differs from signed deployment policy")
    key = load_pem_public_key(raw)
    if not isinstance(key, Ed25519PublicKey):
        raise TypeError("read-path receipt key must be Ed25519")
    return key


VERIFY_KEY = _load_pinned_key()


def observe_signed_outcome(document: bytes, signature_b64: str) -> None:
    signature = base64.b64decode(signature_b64, validate=True)
    VERIFY_KEY.verify(signature, document)
    receipt = json.loads(
        document.decode("utf-8"), object_pairs_hook=_no_duplicate_object,
        parse_constant=lambda value: (_ for _ in ()).throw(ValueError(value)),
    )
    if not isinstance(receipt, dict) or set(receipt) != FIELDS:
        raise ValueError("read-path outcome receipt schema differs")
    if receipt["schema_version"] != "aidefend.memory-read-path-outcome.v1":
        raise ValueError("unsupported read-path outcome schema")
    for field in ("receipt_id", "candidate_id", "memory_revision", "policy_version", "observed_at"):
        if not isinstance(receipt[field], str) or not receipt[field].strip():
            raise ValueError(f"missing outcome binding: {field}")
    if not isinstance(receipt["policy_sha256"], str) or not SHA256.fullmatch(receipt["policy_sha256"]):
        raise ValueError("invalid outcome policy digest")
    kind, state = receipt["event_kind"], receipt["state"]
    if kind == "READ_RESULT" and state in READ_STATES:
        read_path_outcome_total.labels(state=state, policy_version=receipt["policy_version"]).inc()
    elif kind == "QUARANTINE_TRANSITION" and state in QUARANTINE_STATES:
        quarantine_transition_total.labels(state=state, policy_version=receipt["policy_version"]).inc()
    else:
        raise ValueError("invalid event-kind/state combination")

Consume these receipts from the authenticated verifier/quarantine event stream after receipt-ID deduplication. The verifier key path and SHA-256 come from the versioned signed deployment policy mounted read-only; an invalid signature, duplicate receipt, unknown state, or binding error is telemetry-pipeline ERROR and is never counted as a verified read.

Compute and alert on the ratio

Store the following recording rule in the versioned, signed monitoring policy. Its five-minute range is an example environment-policy value, not a framework-wide threshold; materialize a different range by versioning and re-signing that rule.

sum by (policy_version) (
  rate(aidefend_memory_read_path_outcome_total{state="VERIFIED_ACTIVE"}[5m])
)
/
clamp_min(
  sum by (policy_version) (rate(aidefend_memory_read_path_outcome_total[5m])),
  1e-12
)

Record this expression as read_path_integrity_ratio. Alert only when the signed rule's minimum ratio is breached and the same policy-version denominator is non-zero; alert separately on absent expected traffic, verifier receipt rejection, quarantine submissions without terminal disposition, and exporter failure. These are detection-health findings and do not make the enforcement decision.

Verify safely

Replay signed valid, tampered-signature, unknown-state, duplicate-receipt, quarantine-hit, promotion, bypass, exporter-outage, and backlog fixtures. Recompute the numerator and complete read-candidate denominator from the immutable receipt ledger and compare them with the Prometheus counters and recorded ratio; quarantine transitions must never change the read denominator.

Open-source tools

• OpenTelemetry
• Prometheus
• Grafana
• Valkey
• Redis 8+ AGPLv3 source build
• Apache Kafka
• OpenSearch
• Open Policy Agent (OPA)

Commercial / hosted tools

• Datadog
• Splunk Enterprise Security
• Microsoft Sentinel
• Elastic Observability

Threat mappings

MITRE ATLAS

• AML.T0080 AI Agent Context Poisoning
• AML.T0080.000 AI Agent Context Poisoning: Memory
• AML.T0080.001 AI Agent Context Poisoning: Thread (cross-namespace writes indicate thread poisoning)
• AML.T0051 LLM Prompt Injection (findings identify injection payloads persisted to memory)
• AML.T0051.001 LLM Prompt Injection: Indirect (findings identify indirect injections written to memory)
• AML.T0061 LLM Prompt Self-Replication (repeated write fingerprints reveal self-replicating prompts)
• AML.T0092 Manipulate User LLM Chat History (memory-write telemetry covers chat-history manipulation)
• AML.T0118 Autonomous AI Agent Communication
• AML.T0118.000 Autonomous AI Agent Communication: Communication via Shared Artifacts (cross-namespace and repeated-fingerprint writes reveal shared memory used as a coordination channel)

MAESTRO

• Data Poisoning (L2)
• Agent Goal Manipulation (L7)
• Agent Tool Misuse (L7)
• Data Tampering (L2) (memory-write telemetry reveals tampering in memory stores)

OWASP LLM Top 10 2026

• LLM05:2026 Data and Model Poisoning

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI06:2026 Memory & Context Poisoning
• ASI01:2026 Agent Goal Hijack (memory poisoning enables persistent goal hijacking)
• ASI10:2026 Rogue Agents (rogue agents exhibit abnormal memory write patterns)

NIST Adversarial Machine Learning 2025

• NISTAML.018 Prompt Injection (findings identify injection payloads persisted to memory)
• NISTAML.015 Indirect Prompt Injection

Cisco Integrated AI Security and Safety Framework

• AISubtech-5.1.1 Long-term / Short-term Memory Injection
• AITech-5.1 Memory System Persistence
• AITech-7.2 Memory System Corruption

Google Secure AI Framework 2.0 - Risks

• PIJ: Prompt Injection (findings identify injection payloads persisted to memory)

Databricks AI Security Framework 3.0

• Agents - Core 13.1: Memory Poisoning
• Agents - Core 13.6: Intent Breaking & Goal Manipulation (memory poisoning can redirect agent goals)