Scope boundary

Owns application-level statelessness, task-local state construction, cross-task persistence avoidance, and trusted mission reload. It does not own runtime context-store tenant or session isolation, size limits, TTL, reset, serialization, process teardown, or memory zeroization.

AID-I-004.001: AID-H-017.004 owns the application architecture that avoids or minimizes retained cross-task state and reloads the trusted mission for each task; AID-I-004.001 owns steady-state runtime enforcement for any volatile context store, including tenant and session isolation, bounded size, TTL, resets, and fail-closed serialization. A stateless design can consume AID-I-004.001 evidence when a store exists, but retained-state TTL and isolation receive only the AID-I-004.001 outcome.

AID-I-001: AID-H-017.004 controls application-level state ownership and lifetime; AID-I-001 controls process, container, sandbox, filesystem, and runtime teardown or zeroization. Application expiry does not prove low-level execution residue was removed.

Implementation guidance

• AID-H-017.004-G001: Implement stateless agents with ephemeral memory for each task.

AID-H-017.004-G001

Concept:

An agent's memory can be a vector for persistent attacks. A secure architecture treats an agent's short-term memory as ephemeral and untrusted by default. The agent should be designed to be as stateless as possible, reloading its core, signed objective from a trusted source at the beginning of each new task or session.

Implement a Stateless Agent with Ephemeral Memory

This design pattern ensures that each new task starts with a clean slate, purging any potential manipulation from previous conversations.

# File: agent_arch/stateless_handler.py
from __future__ import annotations
import uuid
import json
import math
import ssl
from urllib.parse import urlsplit, urlunsplit

import base64
import os

import httpx
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

def _required_positive_int(name: str) -> int:
    value = int(os.environ[name])
    if value <= 0:
        raise RuntimeError(f"{name} must be a positive integer")
    return value

TASK_INPUT_MAX_UTF8_BYTES = _required_positive_int("VERIFIED_H017_TASK_INPUT_MAX_UTF8_BYTES")


# aidefend.stateless-task-request.v1 requires a UUID request ID and non-empty input.
NON_EMPTY_TASK_INPUT_LENGTH = 1

def _required_positive_seconds(name: str) -> float:
    value = float(os.environ[name])
    if not math.isfinite(value) or value <= 0:
        raise RuntimeError(f"{name} must be a finite positive number of seconds")
    return value

STATELESS_WORKER_TOTAL_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H017_STATELESS_WORKER_TOTAL_TIMEOUT_SECONDS")
STATELESS_WORKER_CONNECT_TIMEOUT_SECONDS = _required_positive_seconds("VERIFIED_H017_STATELESS_WORKER_CONNECT_TIMEOUT_SECONDS")
STATELESS_WORKER_RESPONSE_MAX_BYTES = _required_positive_int("VERIFIED_H017_STATELESS_WORKER_RESPONSE_MAX_BYTES")


app = FastAPI()
MODEL_GATEWAY_URL = os.environ["MODEL_GATEWAY_URL"]
_gateway = urlsplit(MODEL_GATEWAY_URL)
if (
    _gateway.scheme != "https" or not _gateway.hostname
    or _gateway.username is not None or _gateway.password is not None
    or _gateway.query or _gateway.fragment or not _gateway.path.startswith("/")
    or MODEL_GATEWAY_URL != urlunsplit(("https", _gateway.netloc.lower(), _gateway.path, "", ""))
):
    raise RuntimeError("model gateway URL must be the exact canonical HTTPS endpoint")
MODEL_ID = os.environ["MODEL_ID"]
SYSTEM_PROMPT = os.environ["TRUSTED_SYSTEM_PROMPT"]
GOAL_BYTES = base64.b64decode(os.environ["SIGNED_GOAL_B64"], validate=True)
GOAL_SIGNATURE = base64.b64decode(os.environ["SIGNED_GOAL_SIGNATURE_B64"], validate=True)
GOAL_PUBLIC_KEY = Ed25519PublicKey.from_public_bytes(
    base64.b64decode(os.environ["GOAL_VERIFY_KEY_B64"], validate=True)
)


class TaskRequest(BaseModel):
    request_id: uuid.UUID
    input: str = Field(min_length=NON_EMPTY_TASK_INPUT_LENGTH)


def verified_goal() -> str:
    try:
        GOAL_PUBLIC_KEY.verify(GOAL_SIGNATURE, GOAL_BYTES)
    except InvalidSignature as exc:
        raise RuntimeError("signed goal verification failed") from exc
    return GOAL_BYTES.decode("utf-8")


@app.post("/new_task")
def handle_new_task(request: TaskRequest) -> dict:
    if len(request.input.encode("utf-8")) > TASK_INPUT_MAX_UTF8_BYTES:
        raise HTTPException(status_code=413, detail="task input exceeds signed-policy byte limit")
    try:
        # The task-local list is the complete memory for this request. Nothing is
        # loaded from, or written to, a cross-session conversation store.
        messages = [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "developer", "content": verified_goal()},
            {"role": "user", "content": request.input},
        ]
        context = ssl.create_default_context(cafile=os.environ["VERIFIED_H017_MODEL_GATEWAY_CA_BUNDLE"])
        context.load_cert_chain(
            os.environ["VERIFIED_H017_MODEL_GATEWAY_CLIENT_CERT"],
            os.environ["VERIFIED_H017_MODEL_GATEWAY_CLIENT_KEY"],
        )
        with httpx.Client(
            verify=context, follow_redirects=False,
            timeout=httpx.Timeout(STATELESS_WORKER_TOTAL_TIMEOUT_SECONDS, connect=STATELESS_WORKER_CONNECT_TIMEOUT_SECONDS),
        ) as client:
            with client.stream(
                "POST", MODEL_GATEWAY_URL,
                json={"model": MODEL_ID, "messages": messages},
                headers={"X-Request-ID": str(request.request_id), "Accept": "application/json"},
            ) as response:
                if response.status_code != 200:
                    raise RuntimeError(f"model gateway returned HTTP {response.status_code}")
                raw = bytearray()
                for chunk in response.iter_bytes():
                    raw.extend(chunk)
                    if len(raw) > STATELESS_WORKER_RESPONSE_MAX_BYTES:
                        raise RuntimeError("model gateway response exceeds signed decoded-byte cap")
            payload = json.loads(bytes(raw))
            if not isinstance(payload, dict) or set(payload) != {"output"} or not isinstance(payload["output"], str):
                raise ValueError("model gateway response schema differs")
        return {"request_id": str(request.request_id), "result": payload["output"]}
    except (RuntimeError, httpx.HTTPError, KeyError, ValueError) as exc:
        raise HTTPException(status_code=503, detail="task execution unavailable") from exc
    finally:
        if "messages" in locals():
            messages.clear()

Verification and evidence

Run two sessions with distinct canary strings and prove neither canary appears in the other request, logs, cache, trace attributes, or subsequent model input. Tamper with the goal bytes and require fail-closed behavior before the gateway call. Preserve the model and prompt versions, signed-goal digest and signer identity, request/trace IDs, outbound message-role readback with content redacted, storage/cache query results, and explicit PASS/FAIL/ERROR verdict. Clearing a Python list is lifecycle hygiene, not guaranteed memory zeroization; highly sensitive tasks still require the process or sandbox teardown owned by AID-I-001.003.

Action: Keep this handler stateless: construct request-local state from the authenticated task, reload the trusted signed prompt and objective, and discard that state when the task ends. Do not create or score a shared context store here. If the architecture retains context in RAM, Redis, or another shared store, consume the tenant/session isolation, size, TTL, reset, and serialization evidence owned by AID-I-004.001.

Keep the action safely bounded

Reconcile the complete population of every eligible task, its request-local state objects, temporary credentials, and teardown events; reject expired, revoked, or stale policy and asset bindings under the policy-defined freshness window.

Open-source tools

• LangGraph
• Valkey
• Redis 8+ (AGPL-3.0)
• Memcached

Commercial / hosted tools

• Amazon ElastiCache
• Google Cloud Memorystore
• Azure Managed Redis

Threat mappings

MITRE ATLAS

• AML.T0080 AI Agent Context Poisoning
• AML.T0080.000 AI Agent Context Poisoning: Memory
• AML.T0092 Manipulate User LLM Chat History (ephemeral state prevents persistent chat history manipulation across sessions)
• AML.T0118 Autonomous AI Agent Communication
• AML.T0118.000 Autonomous AI Agent Communication: Communication via Shared Artifacts (request-local state leaves no shared persistent artifacts for agents to coordinate through)
• AML.T0121 AI Agent Environment Reconstruction (state is rebuilt only from the signed mission source, not from recovered working artifacts)
• AML.T0130 AI Agent Response Biasing (task-local state and disabled persistent memory limit cross-session recommendation poisoning)

MAESTRO

• N/A

OWASP LLM Top 10 2026

• LLM01:2026 Prompt Injection (per-task isolation and bounded retention limit persistence of injected instructions but do not detect their semantics)
• LLM05:2026 Data and Model Poisoning (state isolation and expiry limit durable memory contamination but do not validate candidate memory content)

OWASP ML Top 10 2023

• N/A

OWASP Top 10 for Agentic Applications 2026

• ASI06:2026 Memory & Context Poisoning

NIST Adversarial Machine Learning 2025

• N/A

Cisco Integrated AI Security and Safety Framework

• AISubtech-4.2.2 Session Boundary Violation
• AISubtech-5.1.1 Long-term / Short-term Memory Injection
• AITech-4.2 Context Boundary Attacks
• AITech-5.1 Memory System Persistence

Google Secure AI Framework 2.0 - Risks

• N/A

Databricks AI Security Framework 3.0

• Agents - Core 13.1: Memory Poisoning