AI Agent Memory: Short-Term Context, Long-Term Memory, and State Management

AI Agent Memory: Short-Term Context, Long-Term Memory, and State Management

Agent memory is not one storage system. It is a set of differently scoped, differently governed layers: short-term context for the current task, long-term memory for durable facts across sessions, and state management for resumable workflow progress. In healthcare the separation matters because a scheduling preference, an appointment and a temporary triage conversation need different lifecycles, access controls and retention rules.

Scope. The three Python snippets are in-memory teaching demos and must never hold real patient data. Section 4 is the production-oriented pattern, tested against PostgreSQL 16 with row-level security; it still needs the organisation's identity, consent, EHR and scheduling integrations, privacy review and security testing before use.

1. Short-term context

A patient writes: "I need to reschedule my cardiology follow-up. I can't do mornings, and I'm currently taking blood thinners." A moment later: "What about Thursday?" The agent must resolve "Thursday" against the cardiology follow-up and still honour no mornings, a constraint stated only in the first turn, without permanently storing transient details such as an unfinished search or an unverified clinical remark.

Short-term memory is the working context: recent messages, collected slots, tool results and intermediate decisions, bounded by the context window and discarded or summarised when the task ends.

The demo below parses slots on every turn. A version that returns early on the first message ("reschedule") drops the "can't do mornings" constraint, and a naive keyword match would even read "mornings" as a preference rather than an exclusion. The specialty comes from the message rather than a hard-coded default, so "reschedule my dermatology visit" is not booked as cardiology. Context is trimmed by size (a stand-in for a token budget) rather than by message count, and the patient ID never enters model context.

# short_term_memory.py -- in-memory teaching demo; production keeps sessions in a TTL store
from __future__ import annotations
import re
from dataclasses import dataclass, field
DAYS = ("monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday")
PERIODS = {"morning": range(0, 12), "afternoon": range(12, 17), "evening": range(17, 24)}
SPECIALTIES = {"cardiology": "cardiology_followup", "endocrinology": "endocrinology_followup"}
NEGATION = r"(?:can't|cannot|can not|not|no|don't|unable to)\s+(?:do\s+|make\s+)?"
APPOINTMENTS = {"cardiology_followup": [  # stand-in for the scheduling system of record
    {"slot_id": "A1", "day": "Thursday", "time": "14:00"},
    {"slot_id": "A2", "day": "Thursday", "time": "09:00"},
    {"slot_id": "A3", "day": "Monday", "time": "15:30"}]}
@dataclass
class ShortTermMemory:
    session_id: str
    messages: list[dict] = field(default_factory=list)
    slots: dict = field(default_factory=lambda: {"excluded_periods": set()})
    max_context_chars: int = 2_000          # proxy for a token budget
    def add(self, role: str, text: str) -> None:
        self.messages.append({"role": role, "text": text})
    def model_context(self) -> dict:
        """Newest turns that fit the budget, plus structured slots (no patient ID)."""
        kept, used = [], 0
        for m in reversed(self.messages):
            if used + len(m["text"]) > self.max_context_chars:
                break                          # older turns would be summarised instead
            kept.append(m)
            used += len(m["text"])
        return {"recent_messages": kept[::-1],
                "slots": {k: sorted(v) if isinstance(v, set) else v for k, v in self.slots.items()}}
def update_slots(mem: ShortTermMemory, text: str) -> None:
    t = text.casefold()
    for word, specialty in SPECIALTIES.items():
        if word in t:
            mem.slots["specialty"] = specialty   # from the message, never a hard-coded default
    if day := next((d for d in DAYS if re.search(rf"\b{d}s?\b", t)), None):
        mem.slots["preferred_day"] = day.title()
    for period in PERIODS:
        if re.search(rf"{NEGATION}{period}s?\b", t):
            mem.slots["excluded_periods"].add(period)      # "can't do mornings"
        elif re.search(rf"\b{period}s?\b", t):
            mem.slots["preferred_period"] = period
def booking_turn(mem: ShortTermMemory, text: str) -> dict:
    mem.add("user", text)
    update_slots(mem, text)                    # every turn, including the first one
    if "specialty" not in mem.slots:
        return {"next_action": "clarify_appointment"}
    s = mem.slots
    options = [a for a in APPOINTMENTS.get(s["specialty"], [])
               if (not s.get("preferred_day") or a["day"] == s["preferred_day"])
               and (not s.get("preferred_period")
                    or int(a["time"][:2]) in PERIODS[s["preferred_period"]])
               and not any(int(a["time"][:2]) in PERIODS[p] for p in s["excluded_periods"])]
    return {"next_action": "select_slot" if options else "offer_alternatives",
            "slot_ids": [a["slot_id"] for a in options]}

In production, keep session state in an external store with a TTL so a restart does not lose a booking in progress; clear slots and tool results on completion, timeout or abandonment; summarise rather than drop older turns; keep raw PHI out of prompts, logs, caches and vector stores unless policy permits; log context size and truncation without content; and test follow-up references, interruptions, timeouts and concurrent sessions for one patient.

2. Long-term memory

Weeks ago the patient said, "Please don't book me before 10 a.m. because I need time for morning medication and transport." Now they ask to book an endocrinology visit. The agent should retrieve that durable preference, but never treat an old, unverified symptom report or a cancelled appointment's constraint as current truth. The hard questions are what deserves to persist, how it is updated, who may read it, and when it expires.

Long-term memory holds validated, reusable facts (communication preferences, confirmed care-team instructions, recurring scheduling constraints, consent choices) in an external, access-controlled store, retrieved selectively. Each record carries patient and organisation scope, type, attribute, the fact, source, validity window, consent purpose and provenance.

Two corrections shape the demo. First, the parser keeps the direction of a constraint. A pattern that matches after|not before|before and always emits "not before" turns "I prefer appointments before 3 pm" into the opposite rule. It also claimed 24-hour support while requiring an AM/PM suffix. Here "before 3 pm" becomes a latest time, "don't book me before 10" an earliest time, "after 14:30" parses, and an ambiguous "after 3" returns nothing so the agent asks. Second, supersession is per attribute. Superseding every record of the same type would let a new "no later than 4 pm" erase "not before 10 am".

# long_term_memory.py -- in-memory teaching demo of curated, superseding memory
from __future__ import annotations
import re
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
TIME_PREF = re.compile(
    r"\b(?P<rel>not before|no earlier than|after|before|no later than|not after)\s+"
    r"(?P<h>\d{1,2})(?::(?P<m>\d{2}))?\s*(?P<ampm>a\.?m\.?|p\.?m\.?)?(?![\w:])", re.IGNORECASE)
EARLIEST = {"not before", "no earlier than", "after"}
@dataclass
class MemoryRecord:
    patient_id: str
    memory_type: str            # e.g. scheduling_preference
    attribute: str              # e.g. earliest_time: supersession is per attribute
    value: str
    source: str
    valid_until: datetime | None
    memory_id: str = field(default_factory=lambda: uuid.uuid4().hex)
    superseded_by: str | None = None
    def active(self, now: datetime) -> bool:
        return self.superseded_by is None and (self.valid_until is None or now < self.valid_until)
class LongTermMemoryStore:
    def __init__(self) -> None:
        self.records: dict[str, MemoryRecord] = {}
    def upsert(self, new: MemoryRecord) -> MemoryRecord:
        # Supersede only the same attribute: a new "latest time" must not erase "earliest time".
        for old in self.records.values():
            if ((old.patient_id, old.memory_type, old.attribute)
                    == (new.patient_id, new.memory_type, new.attribute) and old.superseded_by is None):
                old.superseded_by = new.memory_id
        self.records[new.memory_id] = new
        return new
    def retrieve(self, patient_id: str, memory_type: str) -> list[MemoryRecord]:
        now = datetime.now(timezone.utc)
        return [r for r in self.records.values()
                if r.patient_id == patient_id and r.memory_type == memory_type and r.active(now)]
def parse_time_preference(text: str) -> tuple[str, str] | None:
    """Return (attribute, HH:MM). Keeps the direction: 'before 3 pm' is a latest time.
    A narrow demo parser, not clinical NLP; ambiguous input returns None."""
    m = TIME_PREF.search(text)
    if not m:
        return None
    hour, minute = int(m["h"]), int(m["m"] or 0)
    ampm = (m["ampm"] or "").replace(".", "").lower()
    if ampm:
        if not 1 <= hour <= 12:
            return None
        hour = hour % 12 + (12 if ampm == "pm" else 0)
    elif m["m"] is None:                        # bare "after 3" is ambiguous: ask, don't guess
        return None
    if hour > 23 or minute > 59:
        return None
    attribute = "earliest_time" if m["rel"].lower() in EARLIEST else "latest_time"
    if m["rel"].lower() == "before" and re.search(r"\b(don't|do not|never)\b[^.]*$",
                                                  text[:m.start()], re.IGNORECASE):
        attribute = "earliest_time"             # "don't book me before 10" = not before 10
    return attribute, f"{hour:02d}:{minute:02d}"
def remember_preference(store: LongTermMemoryStore, patient_id: str, text: str,
                        consent_granted: bool) -> MemoryRecord | None:
    if not consent_granted:
        raise PermissionError("consent is required before persisting a preference")
    if (parsed := parse_time_preference(text)) is None:
        return None                             # ask a clarifying question instead
    attribute, value = parsed
    return store.upsert(MemoryRecord(
        patient_id=patient_id, memory_type="scheduling_preference", attribute=attribute,
        value=value, source="patient_stated",
        valid_until=datetime.now(timezone.utc) + timedelta(days=365)))

Production rules. Store extracted facts and decisions, not transcripts. Filter deterministically by patient, organisation, type, consent and validity before any semantic ranking. Require explicit promotion rules for conversation data, keep provenance on every record, support supersede, expire and delete without silently overwriting clinical facts, and never let the model overwrite authoritative clinical data. Measure retrieval precision, stale-memory rate and cross-patient leakage, and give patients a way to view, correct and delete non-clinical preferences.

3. State management

A patient starts a prior-authorisation workflow: uploads a referral, confirms diagnosis and medication, requests authorisation, and the insurer API times out. On return the assistant must resume at the right step, never ask for the referral again, never submit twice, and never forget that the outcome is unknown. Memory answers "what does the agent know?"; state answers "where is the workflow, and what is safe to do next?"

A timeout proves neither success nor failure, and the dangerous case is the request that was accepted but whose response was lost. The demo models exactly that. The workflow moves to needs_reconciliation; on resume it first asks the insurer whether the idempotency key is known, adopts the existing authorisation if so, and resubmits with the same key only if not. The key is bound to a payload hash, so changed data cannot be sent under an old key. Transitions follow an explicit table with optimistic versioning, so a failed step can never silently advance and a stale writer is rejected.

# workflow_state.py -- in-memory teaching demo of resumable, idempotent workflow state
from __future__ import annotations
import hashlib
import json
import uuid
from dataclasses import dataclass, field
TRANSITIONS = {                      # the only legal moves; anything else is a bug
    "verify_referral": {"submit_prior_authorization"},
    "submit_prior_authorization": {"await_insurer_decision", "needs_reconciliation"},
    "needs_reconciliation": {"await_insurer_decision", "submit_prior_authorization"},
    "await_insurer_decision": {"completed"},
}
class WorkflowError(Exception):
    pass
class InsurerTimeout(Exception):
    pass
@dataclass
class Workflow:
    patient_id: str
    step: str = "verify_referral"
    data: dict = field(default_factory=dict)
    operations: dict = field(default_factory=dict)
    audit: list = field(default_factory=list)
    version: int = 1
    workflow_id: str = field(default_factory=lambda: uuid.uuid4().hex)
    def move(self, to: str, expected_version: int) -> None:
        if expected_version != self.version:
            raise WorkflowError("stale version; reload state")       # optimistic concurrency
        if to not in TRANSITIONS.get(self.step, set()):
            raise WorkflowError(f"illegal transition {self.step} -> {to}")
        self.audit.append({"from": self.step, "to": to, "version": self.version + 1})
        self.step, self.version = to, self.version + 1
class FakeInsurer:
    """Honours idempotency keys, and can time out *after* accepting a request."""
    def __init__(self) -> None:
        self.requests: dict[str, dict] = {}
        self.timeout_after_accept = False
    def submit(self, key: str, payload_hash: str) -> dict:
        if key in self.requests and self.requests[key]["payload_hash"] != payload_hash:
            raise WorkflowError("idempotency key reused with a different payload")
        self.requests.setdefault(key, {"authorization_id": f"EXT-{uuid.uuid4().hex[:8]}",
                                       "payload_hash": payload_hash})
        if self.timeout_after_accept:
            raise InsurerTimeout()             # the request landed; the response did not
        return self.requests[key]
    def lookup(self, key: str) -> dict | None:
        return self.requests.get(key)
def submit_prior_authorization(wf: Workflow, insurer: FakeInsurer) -> dict:
    if wf.step not in ("submit_prior_authorization", "needs_reconciliation"):
        raise WorkflowError("workflow is not ready for submission")
    payload_hash = hashlib.sha256(json.dumps(wf.data, sort_keys=True).encode()).hexdigest()
    op = wf.operations.setdefault("submit", {"key": f"AUTH-{wf.workflow_id}-1",
                                             "payload_hash": payload_hash})
    if op["payload_hash"] != payload_hash:   # data changed: this is a new submission
        raise WorkflowError("payload changed since first attempt; start a new operation")
    if wf.step == "needs_reconciliation" and (found := insurer.lookup(op["key"])):
        wf.data["authorization_id"] = found["authorization_id"]       # it did go through
        wf.move("await_insurer_decision", wf.version)
        return {"status": "reconciled", "authorization_id": found["authorization_id"]}
    try:
        result = insurer.submit(op["key"], payload_hash)
    except InsurerTimeout:
        if wf.step != "needs_reconciliation":
            wf.move("needs_reconciliation", wf.version)   # outcome unknown, not failed
        return {"status": "outcome_unknown"}
    wf.data["authorization_id"] = result["authorization_id"]
    wf.move("await_insurer_decision", wf.version)
    return {"status": "submitted", "authorization_id": result["authorization_id"]}

Production rules. Separate run, session, user and workflow state from external state, and treat external systems as the source of truth for clinical, authorisation and appointment status. Persist every checkpoint durably with a version and audit event, use stable idempotency keys for every non-read action, bound retries with backoff and circuit breakers, isolate state by patient, encounter, organisation and workflow, and define degraded behaviour when the memory or workflow store is down. Test crash recovery, duplicate submission, concurrent updates and expired sessions.

4. Production-oriented implementation

The reference service persists curated preferences and workflow checkpoints with FastAPI and PostgreSQL. It covers non-clinical preferences and workflow metadata only, and does not replace an EHR, scheduling platform or payer system. Trust boundaries: OIDC tokens verified for issuer, audience and expiry; organisation, patient relationship, purpose and active consent checked on every read and write through row-level security, never inferred from a URL; transactions, parameterised SQL, constraints and optimistic concurrency; typed facts with provenance, consent and validity; and redacted logs, encryption, managed secrets and audited access.

Schema and row-level security

Testing the original schema on a real PostgreSQL 16 server exposed two defects. Superseding wrote the new memory's ID into the old row before the new row existed, so the immediate foreign key made every second preference for a patient fail. And after a transaction-local set_config, a custom setting reads back as an empty string, not NULL, so current_setting(...)::uuid throws on every later query from a pooled connection. The schema below defers the self-referencing foreign key, adds a partial unique index so only one record per attribute can be current (concurrent writers get a conflict instead of two current rows), wraps settings in NULLIF so an unset context denies quietly, and puts the access rule in one function shared by all policies. The app connects as a non-owner role, and FORCE ROW LEVEL SECURITY covers the table owner too.

-- schema.sql: PostgreSQL 15+. Apply as the owner; the app connects as a non-owner role.
CREATE TABLE patient_access_grants (
  organization_id uuid NOT NULL, patient_id uuid NOT NULL, actor_sub text NOT NULL,
  purpose text NOT NULL, expires_at timestamptz, revoked_at timestamptz,
  PRIMARY KEY (organization_id, patient_id, actor_sub, purpose));
CREATE TABLE patient_consent (
  consent_record_id uuid PRIMARY KEY, organization_id uuid NOT NULL, patient_id uuid NOT NULL,
  purpose text NOT NULL, status text NOT NULL CHECK (status IN ('granted', 'revoked', 'expired')),
  granted_at timestamptz NOT NULL, revoked_at timestamptz, expires_at timestamptz);
CREATE TABLE patient_memory (
  memory_id uuid PRIMARY KEY, organization_id uuid NOT NULL, patient_id uuid NOT NULL,
  memory_type text NOT NULL CHECK (memory_type IN ('scheduling_preference', 'communication_preference')),
  attribute text NOT NULL, fact jsonb NOT NULL,
  source text NOT NULL CHECK (source IN ('patient_stated', 'staff_entered')),
  consent_purpose text NOT NULL,
  consent_record_id uuid NOT NULL REFERENCES patient_consent (consent_record_id),
  valid_from timestamptz NOT NULL, valid_until timestamptz CHECK (valid_until > valid_from),
  superseded_by uuid REFERENCES patient_memory (memory_id) DEFERRABLE INITIALLY DEFERRED,
  created_by text NOT NULL, created_at timestamptz NOT NULL DEFAULT now());
-- Deferred FK: the old row can point at the new one before the new one is inserted.
-- At most one current record per attribute: concurrent writers get a unique violation
-- (reported as 409) instead of silently leaving two "current" preferences.
CREATE UNIQUE INDEX patient_memory_one_current ON patient_memory
  (organization_id, patient_id, memory_type, attribute, consent_purpose)
  WHERE superseded_by IS NULL;
CREATE TABLE agent_workflow (
  workflow_id uuid PRIMARY KEY, organization_id uuid NOT NULL, patient_id uuid NOT NULL,
  workflow_type text NOT NULL, current_step text NOT NULL, state jsonb NOT NULL DEFAULT '{}',
  version bigint NOT NULL DEFAULT 1, updated_at timestamptz NOT NULL DEFAULT now());
-- Custom settings revert to '' (not NULL) after a transaction-local set_config, so a
-- bare ::uuid cast makes every query on a reused pooled connection fail. NULLIF keeps
-- an unset context denying access instead.
CREATE FUNCTION app_setting(name text) RETURNS text LANGUAGE sql STABLE
  AS $$ SELECT NULLIF(current_setting(name, true), '') $$;
-- Parameters are prefixed: in SQL functions a bare name that matches a column resolves to
-- the column, so "g.purpose = purpose" would compare the column with itself (always true).
CREATE FUNCTION app_may_access(p_org uuid, p_patient uuid, p_purpose text) RETURNS boolean
  LANGUAGE sql STABLE AS $$
  SELECT p_org = app_setting('app.org_id')::uuid AND (
    p_patient = app_setting('app.patient_self_id')::uuid
    OR EXISTS (SELECT 1 FROM patient_access_grants g
               WHERE g.organization_id = p_org AND g.patient_id = p_patient
                 AND g.actor_sub = app_setting('app.actor_sub') AND g.purpose = p_purpose
                 AND g.revoked_at IS NULL AND (g.expires_at IS NULL OR g.expires_at > now())))
  $$;
ALTER TABLE patient_memory ENABLE ROW LEVEL SECURITY;
ALTER TABLE patient_memory FORCE ROW LEVEL SECURITY;
CREATE POLICY scope ON patient_memory   -- FOR ALL: USING doubles as WITH CHECK
  USING (app_may_access(organization_id, patient_id, consent_purpose));
ALTER TABLE patient_consent ENABLE ROW LEVEL SECURITY;
ALTER TABLE patient_consent FORCE ROW LEVEL SECURITY;
CREATE POLICY scope ON patient_consent USING (app_may_access(organization_id, patient_id, purpose));
ALTER TABLE agent_workflow ENABLE ROW LEVEL SECURITY;
ALTER TABLE agent_workflow FORCE ROW LEVEL SECURITY;
CREATE POLICY scope ON agent_workflow USING (app_may_access(organization_id, patient_id, workflow_type));
CREATE ROLE memory_app LOGIN;          -- no ownership, no BYPASSRLS, no grant management
GRANT SELECT ON patient_access_grants TO memory_app;
GRANT SELECT ON patient_consent TO memory_app;
GRANT SELECT, INSERT, UPDATE ON patient_memory, agent_workflow TO memory_app;

FastAPI service

Facts are typed per memory type: a dict field accepts arbitrary nested text, which is how transcripts leak into "memory". Provenance records whether the patient or a staff member wrote the fact, rather than always claiming "patient_stated". The pool and JWKS client are created in the lifespan, not at import. An identity-provider outage returns 503, not 401, and unique and RLS violations map to 409 and 403. The checkpoint endpoint enforces the same transition table as the demo, locks the row and merges a bounded state patch, instead of letting the client set any step, status and whole state blob.

"""Curated patient preferences and workflow checkpoints (FastAPI + PostgreSQL + RLS).
Dependencies: fastapi, uvicorn, "psycopg[binary,pool]", "pyjwt[crypto]".
"""
from __future__ import annotations
import os
import uuid
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from typing import Annotated, Literal
import jwt
from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from psycopg import errors
from psycopg.rows import dict_row
from psycopg.types.json import Jsonb
from psycopg_pool import ConnectionPool
from pydantic import BaseModel, ConfigDict, Field
from workflow_state import TRANSITIONS
class TimePreference(BaseModel):          # typed facts: no free-text transcript fragments
    model_config = ConfigDict(extra="forbid")
    kind: Literal["earliest_time", "latest_time"]
    time: str = Field(pattern=r"^([01]\d|2[0-3]):[0-5]\d$")
class PreferenceIn(BaseModel):
    model_config = ConfigDict(extra="forbid")
    memory_type: Literal["scheduling_preference"]
    fact: TimePreference
    consent_purpose: Literal["appointment_scheduling"]
    consent_record_id: uuid.UUID
    valid_until: datetime | None = None
class CheckpointIn(BaseModel):
    model_config = ConfigDict(extra="forbid")
    expected_version: int = Field(ge=1)
    next_step: str = Field(max_length=80)
    state_patch: dict[str, str | int | bool | None] = Field(default_factory=dict, max_length=20)
@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.jwks = jwt.PyJWKClient(os.environ["OIDC_JWKS_URL"], lifespan=300)
    app.state.pool = ConnectionPool(os.environ["DATABASE_URL"], min_size=1, max_size=10,
                                    kwargs={"row_factory": dict_row}, open=False)
    app.state.pool.open(wait=True, timeout=10)   # opened at startup, not at import
    yield
    app.state.pool.close()
app = FastAPI(title="Healthcare Agent Memory API", lifespan=lifespan)
bearer = HTTPBearer()
def claims(request: Request, cred: Annotated[HTTPAuthorizationCredentials, Depends(bearer)]) -> dict:
    try:
        key = request.app.state.jwks.get_signing_key_from_jwt(cred.credentials).key
    except jwt.PyJWKClientConnectionError:
        raise HTTPException(503, "identity_provider_unavailable") from None  # not a 401
    except jwt.PyJWTError:
        raise HTTPException(401, "invalid_token") from None
    try:
        c = jwt.decode(cred.credentials, key, algorithms=["RS256"],
                       issuer=os.environ["OIDC_ISSUER"], audience=os.environ["OIDC_AUDIENCE"],
                       options={"require": ["exp", "iat", "iss", "aud", "sub"]})
        uuid.UUID(str(c["org_id"]))
        if c.get("patient_id"):
            uuid.UUID(str(c["patient_id"]))
    except (jwt.PyJWTError, KeyError, ValueError):
        raise HTTPException(401, "invalid_token") from None
    return c
def need(c: dict, scope: str) -> None:
    if scope not in c.get("scope", "").split():
        raise HTTPException(403, "insufficient_scope")
def tx(request: Request, c: dict):
    """Connection whose transaction carries the verified identity into RLS."""
    conn = request.app.state.pool.getconn()
    conn.execute("BEGIN")
    for name, value in (("app.org_id", c["org_id"]), ("app.actor_sub", c["sub"]),
                        ("app.patient_self_id", c.get("patient_id") or "")):
        conn.execute("SELECT set_config(%s, %s, true)", (name, str(value)))
    return conn
def finish(request: Request, conn, ok: bool) -> None:
    conn.execute("COMMIT" if ok else "ROLLBACK")
    request.app.state.pool.putconn(conn)
@app.post("/v1/patients/{patient_id}/memories", status_code=201)
def create_memory(patient_id: uuid.UUID, body: PreferenceIn, request: Request,
                  c: Annotated[dict, Depends(claims)]):
    need(c, "memory:write")
    now, memory_id = datetime.now(timezone.utc), uuid.uuid4()
    if body.valid_until and body.valid_until <= now:
        raise HTTPException(422, "valid_until must be in the future")
    conn, ok = tx(request, c), False
    try:
        # RLS hides consent rows the caller may not see, so "not found" also covers
        # "not authorised"; only an active, matching consent lets the write proceed.
        if not conn.execute(
                """SELECT 1 FROM patient_consent WHERE consent_record_id = %s AND patient_id = %s
                   AND purpose = %s AND status = 'granted' AND revoked_at IS NULL
                   AND (expires_at IS NULL OR expires_at > now())""",
                (body.consent_record_id, patient_id, body.consent_purpose)).fetchone():
            raise HTTPException(403, "no_active_consent_or_access")
        # Supersede first (the FK is deferred), then insert the new current record.
        conn.execute(
            """UPDATE patient_memory SET superseded_by = %s
               WHERE organization_id = %s AND patient_id = %s AND memory_type = %s
                 AND attribute = %s AND consent_purpose = %s AND superseded_by IS NULL""",
            (memory_id, c["org_id"], patient_id, body.memory_type, body.fact.kind,
             body.consent_purpose))
        source = "patient_stated" if str(c.get("patient_id")) == str(patient_id) else "staff_entered"
        conn.execute(
            """INSERT INTO patient_memory (memory_id, organization_id, patient_id, memory_type,
                 attribute, fact, source, consent_purpose, consent_record_id, valid_from,
                 valid_until, created_by) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
            (memory_id, c["org_id"], patient_id, body.memory_type, body.fact.kind,
             Jsonb(body.fact.model_dump()), source, body.consent_purpose,
             body.consent_record_id, now, body.valid_until, c["sub"]))
        ok = True
    except errors.UniqueViolation:
        raise HTTPException(409, "concurrent_update_retry") from None
    except errors.InsufficientPrivilege:          # RLS WITH CHECK rejected the row
        raise HTTPException(403, "no_access") from None
    finally:
        finish(request, conn, ok)
    return {"memory_id": str(memory_id), "status": "stored"}
@app.get("/v1/patients/{patient_id}/memories")
def list_memories(patient_id: uuid.UUID, request: Request, c: Annotated[dict, Depends(claims)]):
    need(c, "memory:read")
    conn = tx(request, c)
    try:
        rows = conn.execute(
            """SELECT memory_id, attribute, fact, source, valid_from, valid_until
               FROM patient_memory WHERE patient_id = %s AND superseded_by IS NULL
                 AND valid_from <= now() AND (valid_until IS NULL OR valid_until > now())
               ORDER BY attribute LIMIT 50""", (patient_id,)).fetchall()
    finally:
        finish(request, conn, True)
    return {"items": rows}
@app.post("/v1/workflows/{workflow_id}/checkpoint")
def checkpoint(workflow_id: uuid.UUID, body: CheckpointIn, request: Request,
               c: Annotated[dict, Depends(claims)]):
    need(c, "workflow:write")
    conn, ok = tx(request, c), False
    try:
        row = conn.execute("SELECT current_step, version FROM agent_workflow "
                           "WHERE workflow_id = %s FOR UPDATE", (workflow_id,)).fetchone()
        if row is None or row["version"] != body.expected_version:
            raise HTTPException(409, "workflow_changed_or_not_found; reload state")
        if body.next_step not in TRANSITIONS.get(row["current_step"], set()):
            raise HTTPException(409, "illegal_transition")   # the server owns the state machine
        updated = conn.execute(
            """UPDATE agent_workflow SET current_step = %s, state = state || %s,
                 version = version + 1, updated_at = now() WHERE workflow_id = %s
               RETURNING workflow_id, current_step, version""",
            (body.next_step, Jsonb(body.state_patch), workflow_id)).fetchone()
        ok = True
    finally:
        finish(request, conn, ok)
    return updated
@app.get("/health/ready")
def ready(request: Request):
    try:
        with request.app.state.pool.connection(timeout=2) as conn:
            conn.execute("SELECT 1")
    except Exception:
        raise HTTPException(503, "database_unavailable") from None
    return {"status": "ready"}

Verified against PostgreSQL 16: repeated writes supersede correctly per attribute; a staff grant for one purpose cannot read or write another purpose; other patients and organisations see nothing; revoked consent and untyped facts are refused; six concurrent writers leave exactly one current record; illegal, stale and cross-purpose workflow updates are rejected; and an unset context returns no rows instead of an error. Before production, add an audit-event table or service (actor, reference, purpose, outcome, correlation ID; no prompts or tokens), a transactional outbox and worker for external actions that persist an operation key before dispatch and reconcile after timeouts, managed ingress with TLS and rate limits, and threat modelling, penetration, disaster-recovery and load testing plus a compliance review.

Practical architecture rule

Treat memory as scoped, governed infrastructure. Short-term context is minimal, ephemeral and task-specific. Long-term memory is curated, validated, patient-scoped and provenance-backed. State is durable, resumable, idempotent and auditable. Retrieve the smallest relevant context, enforce policy in code and the database rather than in prompts, and never let a transient conversation detail become an unreviewed permanent clinical fact.

Back to all blogs