AID-I-004.001: Runtime Context Isolation & Hygiene

Version 1.20261015 · Actionable sub-technique

Enforces strict boundaries and hygiene for volatile working memory (RAM/Redis). Prevents cross-session/cross-tenant context bleed, limits the temporal blast radius of prompt injections (via windowing/resets), and prevents resource exhaustion (DoS/cost burn) via size/token ceilings and TTL.

Pillars: app

Lifecycle: operation

Scope boundary

Owns isolation and hygiene of volatile active context and working memory by session and tenant, including bounded size, token ceilings, TTL, resets, and fail-closed serialization. It does not partition persistent memory, govern phase-specific secret demotion, or validate summary transformations.

AID-I-004.002: AID-I-004.001 isolates volatile session working context; AID-I-004.002 partitions persistent vector, RAG, and long-term memory by tenant and trust tier. A Redis or in-memory session boundary does not establish durable namespace isolation, and persistent partitioning does not enforce active-context size, TTL, or reset hygiene.

AID-I-004.007: AID-I-004.001 separates and limits active context by session or tenant; AID-I-004.007 removes or demotes secrets and privileged intermediate state at verified task-phase transitions within a live session. General context isolation does not prove that no-longer-needed sensitive values were replaced when the task moved to a new phase.

AID-I-004.008: AID-I-004.001 owns active-context boundaries and hygiene; AID-I-004.008 validates the security integrity of a summarization or compaction transformation. A bounded context window does not prove that a generated summary preserved constraints, provenance, or secret-removal requirements.

AID-H-017.004: AID-I-004.001 owns steady-state isolation, size, TTL, reset, and serialization enforcement for volatile runtime context stores; AID-H-017.004 owns the application architecture that avoids retained cross-task state and reloads trusted mission objectives. Store isolation and TTL receive their outcome here even when a stateless design references the same runtime evidence.

Implementation guidance

AID-I-004.001-G001

Enforce per-session isolation with mandatory TTL, deterministic size/token limits, and fail-closed serialization to prevent cross-tenant bleed and DoS.
Session-store boundary

Derive tenant, task, and session identity from authenticated orchestration state, never from model text. Load TTL, byte/message/token limits, role schema, refresh behavior, and retention from a signature-verified environment policy; there are no framework defaults. A missing, expired, corrupt, wrong-scope, or unreceipted context blocks the operation with its own outcome—it is not silently converted to an empty safe context.

Redis/Valkey CAS module
# File: memory/runtime_context_store.py
# Dependencies: redis>=5.1,<6, jcs>=0.2,<1
from __future__ import annotations

import hashlib
import json
import secrets
from dataclasses import dataclass
from typing import Mapping, Sequence, Callable

import jcs


CAS_WRITE = """
local current = redis.call('HGET', KEYS[1], 'version')
if current then
  if tonumber(current) ~= tonumber(ARGV[1]) then return -1 end
else
  if tonumber(ARGV[1]) ~= 0 then return -2 end
end
redis.call('HSET', KEYS[1],
  'scope_sha256', ARGV[2],
  'version', ARGV[3],
  'blob', ARGV[4],
  'blob_sha256', ARGV[5],
  'policy_sha256', ARGV[6],
  'operation_id', ARGV[7])
redis.call('EXPIRE', KEYS[1], tonumber(ARGV[8]))
redis.call('XADD', KEYS[2], '*',
  'operation_id', ARGV[7],
  'operation', 'WRITE',
  'version', ARGV[3],
  'blob_sha256', ARGV[5],
  'policy_sha256', ARGV[6])
return 1
"""

CLOSE_SESSION = """
local existed = redis.call('EXISTS', KEYS[1])
local version = redis.call('HGET', KEYS[1], 'version')
local digest = redis.call('HGET', KEYS[1], 'blob_sha256')
redis.call('DEL', KEYS[1])
redis.call('XADD', KEYS[2], '*',
  'operation_id', ARGV[1],
  'operation', 'CLOSE',
  'version', version or 'ABSENT',
  'blob_sha256', digest or 'ABSENT',
  'policy_sha256', ARGV[2])
return {existed, redis.call('EXISTS', KEYS[1])}
"""


@dataclass(frozen=True)
class VerifiedScope:
    tenant_id: str
    task_id: str
    session_id: str


@dataclass(frozen=True)
class PolicyVerifier:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    current_session_policy: Callable[[VerifiedScope], Mapping[str, object]]


@dataclass(frozen=True)
class EvidenceClient:
    """Required authenticated runtime adapter callables; no fallback implementation."""
    append_and_verify: Callable[[Mapping[str, object]], Mapping[str, object]]
    require_write: Callable[[str, str, str], Mapping[str, object]]
    require_close: Callable[[str, str], Mapping[str, object]]


def canonical(value: object) -> bytes:
    return bytes(jcs.canonicalize(value))


def sha256(data: bytes) -> str:
    return "sha256:" + hashlib.sha256(data).hexdigest()


def redis_text(value: object) -> str:
    if isinstance(value, bytes):
        return value.decode("utf-8")
    if isinstance(value, str):
        return value
    raise RuntimeError("Redis field has unexpected type")


def scope_digest(scope: VerifiedScope) -> str:
    return sha256(canonical({
        "tenant_id": scope.tenant_id,
        "task_id": scope.task_id,
        "session_id": scope.session_id,
    }))


def validate_messages(
    messages: Sequence[Mapping[str, object]],
    policy: Mapping[str, object],
) -> list[dict[str, str]]:
    if (
        not isinstance(messages, (list, tuple))
        or len(messages) > int(policy["max_messages"])
    ):
        raise ValueError("message population exceeds signed policy")
    allowed_roles = set(policy["allowed_roles"])
    normalized: list[dict[str, str]] = []
    total_tokens = 0
    for message in messages:
        if set(message) != {"role", "content"}:
            raise ValueError("message schema invalid")
        role = message["role"]
        content = message["content"]
        if role not in allowed_roles or not isinstance(content, str):
            raise ValueError("message role/content invalid")
        # The tokenizer adapter used here must match policy.tokenizer_sha256.
        total_tokens += int(policy["token_counter"](content))
        normalized.append({"role": str(role), "content": content})
    if total_tokens > int(policy["max_tokens"]):
        raise ValueError("context token population exceeds signed policy")
    return normalized


class RuntimeContextStore:
    def __init__(
        self,
        redis_client,
        policies: PolicyVerifier,
        evidence: EvidenceClient,
    ):
        self.redis = redis_client
        self.policies = policies
        self.evidence = evidence

    @staticmethod
    def keys(scope: VerifiedScope) -> tuple[str, str]:
        digest = scope_digest(scope).removeprefix("sha256:")
        slot = "{" + digest + "}"
        return (
            "aidefend:" + slot + ":runtime-context",
            "aidefend:" + slot + ":context-outbox",
        )

    def store(
        self,
        *,
        scope: VerifiedScope,
        expected_version: int,
        messages: Sequence[Mapping[str, object]],
    ) -> Mapping[str, object]:
        policy = self.policies.current_session_policy(scope)
        normalized = validate_messages(messages, policy)
        new_version = expected_version + 1
        operation_id = secrets.token_hex(16)
        scope_sha256 = scope_digest(scope)
        envelope = {
            "schema_version": 1,
            "scope_sha256": scope_sha256,
            "version": new_version,
            "messages": normalized,
            "policy_sha256": policy["policy_sha256"],
            "tokenizer_sha256": policy["tokenizer_sha256"],
        }
        blob = canonical(envelope)
        if len(blob) > int(policy["max_context_bytes"]):
            raise ValueError("serialized context exceeds signed byte policy")
        blob_sha256 = sha256(blob)
        context_key, outbox_key = self.keys(scope)
        result = int(self.redis.eval(
            CAS_WRITE,
            2,
            context_key,
            outbox_key,
            str(expected_version),
            scope_sha256,
            str(new_version),
            blob,
            blob_sha256,
            policy["policy_sha256"],
            operation_id,
            str(int(policy["ttl_seconds"])),
        ))
        if result != 1:
            raise RuntimeError("context version conflict or missing expected state")
        receipt = self.evidence.append_and_verify({
            "control": "AID-I-004.001",
            "guidance_ref": "AID-I-004.001-G001",
            "operation": "WRITE",
            "operation_id": operation_id,
            "scope_sha256": scope_sha256,
            "version": new_version,
            "blob_sha256": blob_sha256,
            "byte_count": len(blob),
            "message_count": len(normalized),
            "policy_sha256": policy["policy_sha256"],
            "outcome": "PASS",
        })
        return {
            "version": new_version,
            "blob_sha256": blob_sha256,
            "operation_id": operation_id,
            "evidence_receipt_sha256": receipt["receipt_sha256"],
        }

    def load(self, *, scope: VerifiedScope) -> list[dict[str, str]]:
        policy = self.policies.current_session_policy(scope)
        context_key, _ = self.keys(scope)
        values = self.redis.hmget(
            context_key,
            "scope_sha256",
            "version",
            "blob",
            "blob_sha256",
            "policy_sha256",
            "operation_id",
        )
        ttl = int(self.redis.ttl(context_key))
        if any(value is None for value in values) or ttl <= 0:
            raise RuntimeError("context missing, expired, or incomplete")
        (
            stored_scope, stored_version, blob, stored_digest,
            stored_policy, operation_id,
        ) = values
        stored_scope = redis_text(stored_scope)
        stored_version = redis_text(stored_version)
        stored_digest = redis_text(stored_digest)
        stored_policy = redis_text(stored_policy)
        operation_id = redis_text(operation_id)
        if not isinstance(blob, bytes):
            blob = str(blob).encode("utf-8")
        if (
            stored_scope != scope_digest(scope)
            or stored_policy != policy["policy_sha256"]
            or sha256(blob) != stored_digest
        ):
            raise RuntimeError("context scope, policy, or digest mismatch")
        envelope = json.loads(blob)
        if (
            envelope["scope_sha256"] != stored_scope
            or int(envelope["version"]) != int(stored_version)
            or envelope["policy_sha256"] != stored_policy
        ):
            raise RuntimeError("context envelope mismatch")
        messages = validate_messages(envelope["messages"], policy)
        write_receipt = self.evidence.require_write(
            operation_id=operation_id,
            blob_sha256=stored_digest,
            policy_sha256=stored_policy,
        )
        if write_receipt["outcome"] != "PASS":
            raise RuntimeError("context write evidence is not PASS")
        return messages

    def close(self, *, scope: VerifiedScope) -> Mapping[str, object]:
        policy = self.policies.current_session_policy(scope)
        context_key, outbox_key = self.keys(scope)
        operation_id = secrets.token_hex(16)
        existed, residual = [
            int(value) for value in self.redis.eval(
            CLOSE_SESSION,
            2,
            context_key,
            outbox_key,
            operation_id,
            policy["policy_sha256"],
        )]
        if residual != 0 or int(self.redis.exists(context_key)) != 0:
            raise RuntimeError("session close left context residue")
        if existed == 0:
            prior = self.evidence.require_close(
                scope_sha256=scope_digest(scope),
                policy_sha256=policy["policy_sha256"],
            )
            if prior["outcome"] != "PASS":
                raise RuntimeError("absent context lacks prior close proof")
            return prior
        return self.evidence.append_and_verify({
            "control": "AID-I-004.001",
            "guidance_ref": "AID-I-004.001-G001",
            "operation": "CLOSE",
            "operation_id": operation_id,
            "scope_sha256": scope_digest(scope),
            "policy_sha256": policy["policy_sha256"],
            "residual_count": residual,
            "outcome": "PASS",
        })
Deployment and verification

Use TLS-authenticated Redis/Valkey with ACLs that grant only this service access to the scope keys and outbox stream. The policy verifier must validate signature, issuer/audience/type/algorithm, key state, applicability, monotonic version, positive limits, and the exact tokenizer implementation; inject a callable tokenizer only after that verification. Drain the per-scope outbox to independent append-only evidence and reconcile every operation ID. Independently recompute scope/blob digests, TTL and message/token/byte limits, and prove closure by both key absence and a complete write/read/close population. Test delimiter/case/Unicode identities, cross-tenant/session access, stale version, concurrent write, expiry boundary, corruption, policy rollback, tokenizer mismatch, oversized data, Redis/evidence outage, crash after Lua before receipt, and close retry. Never return [] for corrupt or unverified state.

AID-I-004.001-G002

Enforce Context Pinning and Instruction Re-injection to prevent eviction and 'Lost in the Middle' degradation
Concept

Reset and pinning are privileged state transitions. Derive the checkpoint key from authenticated tenant/agent/session identity, load the baseline and window bound from a signature-verifying policy adapter, discard prior system/control messages, and serialize reset with a coordinator that issues a durable authorization receipt. Do not expose thread_id, baseline text, or a caller-supplied ALLOW object to the model/request body.

Use LangGraph with explicit trusted adapters

This reusable module depends on three platform adapters: ContextPolicyProvider verifies the signed policy and baseline bytes; ResetCoordinator acquires a per-thread lease and consumes a reset authorization token; HistoryNormalizer returns a provider-valid tail of only human, AI, and tool data messages without synthesizing system messages. Their implementations belong to the organization's policy service, database, and model-provider integration.

# File: memory/langgraph_resettable_context.py
from __future__ import annotations

import hashlib
import json
import os
from contextlib import contextmanager
from dataclasses import dataclass
from typing import Annotated, ContextManager, Iterator, Protocol

from langchain_core.messages import (
    AIMessage,
    BaseMessage,
    HumanMessage,
    RemoveMessage,
    SystemMessage,
    ToolMessage,
)
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import REMOVE_ALL_MESSAGES, add_messages
from typing_extensions import TypedDict


@dataclass(frozen=True)
class AuthenticatedSession:
    tenant_id: str
    agent_id: str
    session_id: str
    principal_id: str


@dataclass(frozen=True)
class VerifiedContextPolicy:
    policy_version: str
    policy_sha256: str
    baseline_text: str
    baseline_sha256: str
    max_context_messages: int


@dataclass(frozen=True)
class ResetReceipt:
    receipt_id: str
    thread_key: str
    principal_id: str
    policy_sha256: str
    reason: str


class ContextPolicyProvider(Protocol):
    def load_verified(self, session: AuthenticatedSession) -> VerifiedContextPolicy: ...


class HistoryNormalizer(Protocol):
    def trusted_tail(
        self,
        messages: list[BaseMessage],
        *,
        maximum: int,
    ) -> list[HumanMessage | AIMessage | ToolMessage]: ...


class ResetCoordinator(Protocol):
    def lock_thread(
        self,
        *,
        session: AuthenticatedSession,
        thread_key: str,
    ) -> ContextManager[object]: ...

    def authorize_and_lock(
        self,
        *,
        session: AuthenticatedSession,
        thread_key: str,
        reset_token: str,
        reason: str,
        policy_sha256: str,
    ) -> ContextManager[ResetReceipt]: ...

    def read_receipt(self, receipt_id: str) -> ResetReceipt: ...


class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    reset_receipt_id: str | None


def thread_key(session: AuthenticatedSession) -> str:
    parts = (session.tenant_id, session.agent_id, session.session_id)
    if any(not isinstance(value, str) or not value for value in parts):
        raise PermissionError("authenticated session identity is incomplete")
    canonical = json.dumps(parts, ensure_ascii=False, separators=(",", ":"), allow_nan=False).encode("utf-8")
    return "ctx-" + hashlib.sha256(canonical).hexdigest()


def checked_policy(
    provider: ContextPolicyProvider,
    session: AuthenticatedSession,
) -> VerifiedContextPolicy:
    policy = provider.load_verified(session)
    if (
        not policy.policy_version
        or len(policy.policy_sha256) != 64
        or hashlib.sha256(policy.baseline_text.encode("utf-8")).hexdigest()
        != policy.baseline_sha256
        or isinstance(policy.max_context_messages, bool)
        or policy.max_context_messages < 1
    ):
        raise RuntimeError("verified context policy is invalid")
    return policy


def build_workflow(
    policy: VerifiedContextPolicy,
    normalizer: HistoryNormalizer,
):
    def normalize_context(state: AgentState) -> AgentState:
        prior = state.get("messages", [])
        if any(not isinstance(message, BaseMessage) for message in prior):
            raise RuntimeError("checkpoint contains a non-message object")
        tail = normalizer.trusted_tail(prior, maximum=policy.max_context_messages)
        if (
            len(tail) > policy.max_context_messages
            or any(not isinstance(message, (HumanMessage, AIMessage, ToolMessage)) for message in tail)
        ):
            raise RuntimeError("history normalizer violated its trust contract")
        return {
            "messages": [
                RemoveMessage(id=REMOVE_ALL_MESSAGES),
                SystemMessage(content=policy.baseline_text),
                *tail,
            ],
            "reset_receipt_id": state.get("reset_receipt_id"),
        }

    workflow = StateGraph(AgentState)
    workflow.add_node("normalize_context", normalize_context)
    workflow.add_edge(START, "normalize_context")
    workflow.add_edge("normalize_context", END)
    return workflow


@contextmanager
def build_context_app(
    *,
    session: AuthenticatedSession,
    policy_provider: ContextPolicyProvider,
    normalizer: HistoryNormalizer,
) -> Iterator[tuple[object, VerifiedContextPolicy]]:
    policy = checked_policy(policy_provider, session)
    database_uri = os.environ["LANGGRAPH_CHECKPOINT_DB_URI"]
    serializer = JsonPlusSerializer(pickle_fallback=False)
    with PostgresSaver.from_conn_string(
        database_uri,
        serde=serializer,
    ) as checkpointer:
        checkpointer.setup()
        yield build_workflow(policy, normalizer).compile(
            checkpointer=checkpointer
        ), policy


def append_user_turn(
    app,
    *,
    session: AuthenticatedSession,
    coordinator: ResetCoordinator,
    user_text: str,
) -> dict:
    if not isinstance(user_text, str) or not user_text:
        raise ValueError("user message is empty")
    key = thread_key(session)
    with coordinator.lock_thread(session=session, thread_key=key):
        return app.invoke(
            {"messages": [HumanMessage(content=user_text)], "reset_receipt_id": None},
            config={"configurable": {"thread_id": key}},
        )


def reset_thread(
    app,
    *,
    session: AuthenticatedSession,
    policy: VerifiedContextPolicy,
    coordinator: ResetCoordinator,
    reset_token: str,
    reason: str,
) -> None:
    key = thread_key(session)
    if not reset_token or not reason:
        raise PermissionError("reset authorization token and reason are required")
    with coordinator.authorize_and_lock(
        session=session,
        thread_key=key,
        reset_token=reset_token,
        reason=reason,
        policy_sha256=policy.policy_sha256,
    ) as receipt:
        if (
            receipt.thread_key != key
            or receipt.principal_id != session.principal_id
            or receipt.policy_sha256 != policy.policy_sha256
            or receipt.reason != reason
            or not receipt.receipt_id
        ):
            raise PermissionError("reset authorization receipt binding differs")
        config = {"configurable": {"thread_id": key}}
        app.update_state(
            config=config,
            values={
                "messages": [
                    RemoveMessage(id=REMOVE_ALL_MESSAGES),
                    SystemMessage(content=policy.baseline_text),
                ],
                "reset_receipt_id": receipt.receipt_id,
            },
        )
        observed = app.get_state(config).values
        messages = observed.get("messages", [])
        durable = coordinator.read_receipt(receipt.receipt_id)
        if (
            len(messages) != 1
            or not isinstance(messages[0], SystemMessage)
            or messages[0].content != policy.baseline_text
            or observed.get("reset_receipt_id") != receipt.receipt_id
            or durable != receipt
        ):
            raise RuntimeError("reset state or durable receipt readback differs")

Action: Trigger resets through authenticated orchestration only. Configure JsonPlusSerializer(pickle_fallback=False); an environment flag alone does not disable pickle fallback. The coordinator must reject replayed/expired tokens and serialize both ordinary appends and resets through the same per-thread lease namespace. A missing policy, default window, caller-selected thread, retained prior system message, incomplete receipt, or readback mismatch fails closed. Validate the normalizer against the target provider's tool-call ordering rules before rollout.

Checkpoint keys now use an unambiguous JSON identity tuple. At deployment, retire old delimiter-derived checkpoints; do not fall back to an old key when the new key is absent. Migrate only records whose authenticated tenant, agent and session binding can be independently recovered.

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-I-004.001: Runtime Context Isolation & Hygiene | AIDEFEND

Loading interactive view…

AIDEFEND™

A Practical Knowledge Base for AI Security Defenses

Version

GitHub MCP Service