Memory Protection

Defend agent memory banks (episodic / semantic memory) against query-only memory injection attacks — where an attacker never touches the store directly, but writes malicious records through normal-looking interaction.

Overview

Memory protection wraps the write and retrieval boundary of whatever memory backend your agent uses (vector store, LangGraph checkpointer, custom store). It does not replace the backend — the same pattern as Tool Authorization wrapping tool execution without owning the tools.

Use this guardrail when agents persist conversation turns, reasoning traces, or retrieved facts into long-term memory that later becomes few-shot context for other sessions or users.

No single layer is a complete guarantee. The goal is to raise attack cost and create a forensic trail, not to claim absolute protection.

Defense Layers

Layer	Component	When it runs	What it does
0	ProvenanceChain	Every write	Append-only, hash-chained audit log (who / what / when)
1	MemoryTrustScorer	Write	Composite writer trust + progressive-shortening query signals
2	MemorySanitizer	Write	Query ↔ reasoning coherence and topic-drift checks
3	TrustAwareRetriever	Retrieval	Trust weighting + temporal decay before few-shot use
4	MemoryBehavioralMonitor	Write	Burst / anomaly alerts (non-blocking by default)
5	MemoryIsolationGuard	Retrieval	Cross-user / cross-session access boundary
6	TTL expiration	Scheduled cleanup	Tombstone + optional content purge after expiry
7	MemoryIntegrityValidator	Audit	Hash-chain, duplicate id, and backdating checks
8	MemoryPersistenceGuard	Write	Role/category authorization + sensitive-content blocking

How It Works

1. Enable memory_guardrails in your policy YAML.
2. At persistence time, call before_write() / write() (or the LangGraph memory_write_node) before consolidating short-term memory into episodic memory.
3. Blocked writes return a clear error and never enter the provenance chain as accepted content.
4. After vector search, call filter_retrieval() (or memory_retrieval_node) before results become prompt context.
5. Run run_expiration_cleanup() on a schedule for TTL hygiene — not on the per-request path.
6. Periodically call verify_integrity() / validate_integrity() to detect tampering outside the enforcer.

Configuration

Enable Memory Protection

guardrails:
  memory_guardrails:
    enabled: true

    # Write-time trust and sanitization
    min_writer_trust: 0.3
    min_coherence: 0.15
    topic_drift_ratio: 0.85
    topic_drift_min_terms: 6
    block_on_low_trust: true
    block_on_sanitization_failure: true

    # Behavioral monitoring (alerts; feed back into trust optionally)
    write_burst_window_seconds: 300
    write_burst_threshold: 8

    # Retrieval ranking
    retrieval_half_life_days: 7
    retrieval_min_trust: 0.3
    retrieval_top_k: 5

    # Cross-user / cross-session isolation
    enforce_isolation: true
    isolation_scope: session   # session | user | global

    # TTL / automatic expiration
    enforce_ttl: true
    ttl_days: 90
    auto_purge_content_on_expiry: true

    # Persistence control (what content may be stored)
    enforce_persistence_control: true
    denied_categories_global:
      - credentials
      - secrets
    allowed_categories_by_role:
      guest:
        - general
      analyst:
        - general
        - research
      admin:
        - general
        - research
        - financial
    block_on_sensitive_content: true

Parameters

Option	Type	Default	Description
enabled	bool	false	Enable memory protection enforcement
min_writer_trust	float	0.3	Minimum writer trust score to allow a write
min_coherence	float	0.15	Minimum query/reasoning coherence score
topic_drift_ratio	float	0.85	Max share of reasoning terms absent from the query
topic_drift_min_terms	int	6	Minimum reasoning terms before topic-drift applies
block_on_low_trust	bool	true	Block writes below min_writer_trust
block_on_sanitization_failure	bool	true	Block writes that fail coherence/drift checks
write_burst_window_seconds	float	300	Window for burst anomaly detection
write_burst_threshold	int	8	Writes in the window that trigger a burst alert
retrieval_half_life_days	float	7	Temporal decay half-life for retrieval ranking
retrieval_min_trust	float	0.3	Drop candidates whose writer trust is below this
retrieval_top_k	int	5	Max candidates returned after filtering
enforce_isolation	bool	true	Enforce cross-boundary filtering at retrieval
isolation_scope	str	"session"	session (strictest), user, or global
enforce_ttl	bool	true	Attach TTL and enable expiration cleanup
ttl_days	float	90	Record lifetime before tombstoning
auto_purge_content_on_expiry	bool	true	Erase query/reasoning content after tombstone
enforce_persistence_control	bool	true	Enforce role/category and sensitive-content rules
allowed_categories_by_role	dict	{}	Per-role allowlists of content_category values
denied_categories_global	list	[]	Categories blocked for every role
block_on_sensitive_content	bool	true	Block when the PII/PHI checker flags content

Isolation Scopes

Scope	Visibility
session	Only the exact session that wrote the record (default)
user	Any session belonging to the same writer_id
global	No isolation — any requester can see any record (opt-in only)

Usage with GuardrailSystem

Memory protection is enforced through GuardrailSystem hooks, not automatically inside LLMRails.generate(). Integrate at your memory write and retrieval call sites.

Initialize Guardrails

from elsai_guardrails.guardrails import GuardrailSystem
from elsai_guardrails.guardrails.guardrail_policy import GuardrailPolicy

guardrails = GuardrailSystem(
    guardrail_policy=GuardrailPolicy.from_file("config.yaml"),
)

enforcer = guardrails.memory_guardrail
hook = guardrails.memory_guardrail_hook

When memory_guardrails.enabled is true and PII detection is also enabled, the enforcer automatically reuses the same PII/PHI middleware for Layer 8 sensitive-content checks.

Check Before Memory Write

result = enforcer.before_write(
    session_id="sess-123",
    writer_id="user-42",
    query="what is the weather in Paris tomorrow morning",
    reasoning_trace="looking up weather forecast for Paris tomorrow morning",
    writer_role="analyst",
    content_category="general",
)

if result.passed:
    # Persist to your memory backend using result.record
    print("Allowed:", result.record.record_id)
else:
    print("Blocked:", result.error)

Or raise on block:

from elsai_guardrails.guardrails import MemoryWriteBlockedError

try:
    record = enforcer.write(
        session_id="sess-123",
        writer_id="user-42",
        query=query,
        reasoning_trace=reasoning_trace,
        writer_role="guest",
        content_category="credentials",  # blocked if denied globally
    )
except MemoryWriteBlockedError as exc:
    print("Write denied:", exc)

Filter Retrieval Results

from elsai_guardrails.guardrails import RetrievalCandidate

# After your vector search returns candidates:
candidates = [
    RetrievalCandidate(record=rec_a, similarity=0.91),
    RetrievalCandidate(record=rec_b, similarity=0.88),
]

context = enforcer.filter_retrieval(
    candidates,
    requesting_user_id="user-42",
    requesting_session_id="sess-123",
    top_k=5,
)
# Use `context` as few-shot memory — isolation + trust decay already applied

Direct Record Access

from elsai_guardrails.guardrails import MemoryIsolationError

try:
    enforcer.require_record_access(
        record,
        requesting_user_id="user-99",
        requesting_session_id="sess-other",
    )
except MemoryIsolationError as exc:
    print("Cross-boundary access denied:", exc)

TTL Cleanup and Integrity

# Run on a schedule (cron / periodic task), not per request
report = enforcer.run_expiration_cleanup()
print(f"Tombstoned={report.tombstoned}, redacted={report.redacted}")

if not enforcer.verify_integrity():
    print("Hash chain broken — investigate direct writes bypassing the guardrail")

integrity = enforcer.validate_integrity()
if not integrity.valid:
    print("Integrity report:", integrity)

Feed Alerts Into Trust

result = enforcer.before_write(...)
enforcer.apply_alerts_to_trust(result.alerts)  # degrades standing reputation

LangGraph Integration Pattern

MemoryGuardrailHook mirrors the rate-limit / authorization node pattern:

hook = guardrails.memory_guardrail_hook

def memory_write_node(state: dict) -> dict:
    return hook.memory_write_node(state)

def memory_retrieval_node(state: dict) -> dict:
    return hook.memory_retrieval_node(state)

def memory_cleanup_node(state: dict) -> dict:
    return hook.memory_cleanup_node(state)

Expected state keys for writes:

Key	Required	Description
session_id	Yes	Session that owns the write
query	Yes	User / agent query being stored
reasoning_trace	No	Associated reasoning (defaults to "")
user_id	No	Writer id (defaults to session_id)
writer_role	No	Role for persistence control
content_category	No	Category for persistence control
memory_metadata	No	Extra metadata stored on the provenance record

Expected state keys for retrieval:

Key	Description
memory_candidates	List of RetrievalCandidate from vector search
user_id / session_id	Requester identity for isolation

Outputs written back to state:

• Write pass: memory_write_record_id
• Write block: memory_write_blocked, memory_write_block_reason
• Alerts: memory_write_alerts
• Retrieval: memory_context
• Cleanup: memory_expiration_report

Recommended graph flow:

retrieve → memory_retrieval → agent → (optional) memory_write → agent

Run memory_cleanup on a maintenance graph or cron, not on every turn.

Example Scenarios

Scenario	Result
Coherent query/reasoning, trusted writer	✅ Write allowed; chained in provenance log
Writer trust below min_writer_trust	❌ Blocked (writer trust … below threshold)
Reasoning pivots to unrelated topic	❌ Blocked (sanitization / topic drift)
content_category: credentials in deny list	❌ Blocked (persistence control)
Guest writing a non-allowed category	❌ Blocked (role/category authorization)
Retrieve another user's session memory (isolation_scope: session)	Dropped from candidates / MemoryIsolationError on direct access
Expired record past ttl_days	Tombstoned; invisible to retrieval; content optionally purged

Best Practices

1. Hook at the persistence boundary — Call write / before_write where short-term memory consolidates into long-term storage, not on every chat turn unless you intend to store every turn.
2. Keep isolation strict by default — Prefer isolation_scope: session; use user only when memory must follow a person across devices; avoid global in multi-tenant systems.
3. Pair with PII policies — Enable pii so Layer 8 can block sensitive content without a second Presidio setup.
4. Schedule TTL cleanup — Expiration is lifecycle hygiene; tombstoned records are already invisible to retrieval.
5. Monitor integrity — Periodically validate_integrity() to catch writes that bypassed the enforcer.
6. Treat alerts as signals — Call apply_alerts_to_trust (or use the hook's auto_degrade_trust=True) so burst campaigns degrade future write privilege.

Next Steps

• Tool Authorization — Restrict which tools agents may call
• Rate Limiting — Limit requests, tool calls, and execution time
• PII/PHI Detection — Sensitive-content checks reused by persistence control
• GuardrailSystem — Core API reference
• Guardrails Configuration — Full configuration reference