11. Promotion gate — write path to durable memory

This is the highest-risk operation in the system. Promote everything and the store poisons itself. Promote nothing and the agent is amnesiac.

The promotion gate is a single Aurora transaction that decides: should this candidate enter durable memory? If yes, with what scope, type, provenance, and computed status?

Diagram 4: Promotion gate flow. Every turn produces candidate facts and episodes. The gate filters, deduplicates, and writes to Aurora + OpenSearch in a single transaction. Rejected candidates are logged for evaluation.

Promotion criteria

SQL — Promotion gate decision logic:

-- 1. Deduplicate by content hash
WITH existing AS (
    SELECT fact_id, reconfirmed_count
      FROM fact_memory
     WHERE tenant_id    = :tenant_id
       AND content_hash = :content_hash
       AND status       = 'active'
)
-- 2. If exists, increment reconfirmed_count
UPDATE fact_memory
   SET reconfirmed_count = reconfirmed_count + 1,
       updated_at        = NOW()
 WHERE fact_id = (SELECT fact_id FROM existing)
RETURNING fact_id, 'reconfirmed' AS action;

-- 3. If not exists, insert new fact
INSERT INTO fact_memory (
    fact_id, tenant_id, user_id, agent_id,
    subject, predicate, content, content_hash,
    embedding, metadata,
    status, source_run_id, confidence, created_at
)
SELECT
    gen_random_uuid(),
    :tenant_id, :user_id, :agent_id,
    :subject, :predicate, :content, :content_hash,
    :embedding, :metadata,
    -- Computed status based on confidence + source
    CASE
        WHEN :confidence >= 0.9 AND :source = 'user_confirmed' THEN 'active'
        WHEN :confidence >= 0.7 THEN 'provisional'
        ELSE 'pending_review'
    END AS status,
    :source_run_id, :confidence, NOW()
WHERE NOT EXISTS (SELECT 1 FROM existing)
RETURNING fact_id, 'promoted' AS action;

Key decision points:

• Confidence threshold. Facts with confidence < 0.5 are rejected at the gate. Those between 0.5–0.7 are marked status = 'pending_review' and never surfaced to agents until a human approves them.
• Source provenance. Facts marked source = 'user_confirmed' (the user explicitly said "yes, remember this") bypass the confidence check and go straight to status = 'active'.
• SHA-256 dedup on (tenant, user, content). If the same fact is extracted twice, the gate increments reconfirmed_count instead of duplicating. A high reconfirm count is a strong signal the fact is stable.
• Embedding at write time. The gate calls Bedrock Embed (Titan Embeddings v2) once per promoted fact and stores the vector in the same Aurora transaction. This keeps read-path latency bounded — you never embed at query time.

Rejection logging

Python — Log rejected candidates for evaluation:

async def promotion_gate(candidate: FactCandidate, tenant_id: str,
                        user_id: str, run_id: str):
    if candidate.confidence < 0.5:
        await log_rejection(
            reason="confidence_below_threshold",
            candidate=candidate,
            tenant_id=tenant_id,
            run_id=run_id
        )
        cloudwatch.put_metric(
            "GateRejectionRate",
            value=1.0,
            dimensions={"Tenant": tenant_id, "Reason": "low_confidence"}
        )
        return None

    # ... continue with promotion logic

Every rejected candidate is written to a CloudWatch Log Group /aws/memory/gate-rejections. This log feeds your evaluation loop — when users complain "the agent forgot X", check this log first. Most memory bugs are gate rejections, not retrieval failures.

12. Read and prompt assembly

The prompt is reassembled from durable memory on every turn — it never accumulates. This is the key architectural decision that keeps context bounded and makes Bedrock prompt caching effective. The prompt has a fixed structure: a static prefix (system prompt + policies + preferences) that hits the cache, and a dynamic suffix (facts + episodes + recent context) that changes per turn.

Figure 11: Token-budgeted prompt assembly. The static prefix (system + policy + preferences) stays stable across turns — this is what hits the Bedrock prompt cache. The dynamic suffix (facts + episodes + recent) is rebuilt every turn.

The structure separates cacheable content from volatile content:

Prompt structure

Python — Token-budgeted prompt assembly:

class PromptAssembler:
    def __init__(self, token_budget: int = 16_000):
        self.budget = token_budget
        self.reserved = {
            "system":      2_000,  # static system prompt + tool schemas
            "policy":        500,  # all active policies (exhaustive)
            "preference":    300,  # all active preferences (exhaustive)
            "facts":       1_500,  # top-5 facts (ranked, score >= 0.4)
            "episodes":    1_200,  # top-3 episodes (ranked, score >= 0.5)
            "recent":      2_000,  # compressed tail of last N turns
            "user_msg":      500   # current user message
        }
        self.remaining = self.budget - sum(self.reserved.values())

    async def assemble(self, query: str, memory_bundle: MemoryBundle,
                      recent_turns: List[Turn]) -> str:
        # STATIC PREFIX — hits Bedrock prompt cache
        static_blocks = [
            self._render_system_prompt(),
            self._render_policies(memory_bundle.policies),
            self._render_preferences(memory_bundle.preferences)
        ]
        static_prefix = "\n\n".join(static_blocks)

        # DYNAMIC SUFFIX — changes every turn
        dynamic_blocks = [
            self._render_facts(memory_bundle.facts, limit=5),
            self._render_episodes(memory_bundle.episodes, limit=3),
            self._render_recent_context(recent_turns, token_limit=2000),
            f"User: {query}"
        ]
        dynamic_suffix = "\n\n".join(dynamic_blocks)

        return static_prefix + "\n\n---\n\n" + dynamic_suffix

    def _render_facts(self, facts: List[Fact], limit: int) -> str:
        if not facts: return ""

        lines = ["# Relevant Facts"]
        for i, fact in enumerate(facts[:limit], 1):
            lines.append(
                f"{i}. [{fact.subject}] {fact.content} "
                f"(confidence: {fact.confidence:.2f}, "
                f"source: run {fact.source_run_id[:8]})"
            )
        return "\n".join(lines)

    def _render_recent_context(self, turns: List[Turn],
                               token_limit: int) -> str:
        # Compress last N turns into token budget
        compressed = []
        token_count = 0

        for turn in reversed(turns):
            turn_summary = f"User: {turn.user_msg}\nAgent: {turn.agent_response}"
            turn_tokens  = estimate_tokens(turn_summary)

            if token_count + turn_tokens > token_limit:
                break

            compressed.insert(0, turn_summary)
            token_count += turn_tokens

        if not compressed:
            return ""

        return "# Recent conversation\n" + "\n\n".join(compressed)

Static vs dynamic prompt structure for caching

Bedrock prompt caching works when the prefix of the prompt stays stable across turns. The key is to structure the prompt so that everything that changes infrequently (system instructions, policies, preferences) sits at the top, and everything volatile (facts, recent context, user message) sits at the bottom.

System prompt
User message         ← changes every turn
Facts                ← changes every turn
Policies             ← stable
Preferences          ← stable

System prompt        ← stable
Policies             ← stable
Preferences          ← stable
---
Facts                ← changes every turn
Episodes             ← changes every turn
Recent context       ← changes every turn
User message         ← changes every turn

Python — Bedrock prompt caching integration:

async def invoke_bedrock_with_cache(prompt: str):
    # Split prompt into static prefix + dynamic suffix
    static_prefix, dynamic_suffix = prompt.split("\n\n---\n\n", 1)

    response = await bedrock.converse(
        modelId="anthropic.claude-3-5-sonnet-20241022-v2:0",
        messages=[
            {
                "role": "user",
                "content": [
                    {
                        "text": static_prefix,
                        "cache_control": {"type": "ephemeral"}  # ← cache this
                    },
                    {
                        "text": dynamic_suffix
                    }
                ]
            }
        ]
    )

    # Log cache performance
    cache_metrics = response["usage"]
    cloudwatch.put_metric("CacheHitRate", value=(
        cache_metrics["inputTokens"] /
        (cache_metrics["inputTokens"] + cache_metrics["cacheReadInputTokens"])
    ))

13. Update, supersession, and invalidation

Facts change. A user's timezone changes. A customer's primary contact changes. The system has to handle updates without deleting history — because deleted facts break the audit trail and make replay impossible. The solution is supersession: a contradicted fact is marked superseded_by = new_fact_id and excluded from reads, but remains in the database for audit.

Figure 12: Supersession as an Aurora transaction. The old fact isn't deleted — it's marked superseded_by = new_fact_id and filtered out by WHERE superseded_by IS NULL.

Three operations handle fact lifecycle:

• Update — the fact is still fundamentally true, but a field changed (e.g., reconfirmed_count incremented).
• Supersession — a new fact replaces an old fact. The old fact is marked superseded_by = new_fact_id and excluded from reads.
• Invalidation — the fact is no longer true and has no replacement. Mark status = 'revoked'.

Supersession

SQL — Supersession transaction (Aurora):

-- Step 1: Insert new fact
INSERT INTO fact_memory (
    fact_id, tenant_id, user_id, subject, predicate, content,
    content_hash, embedding, status, source_run_id, created_at
) VALUES (
    gen_random_uuid(), :tenant_id, :user_id, :subject, :predicate, :new_content,
    :new_hash, :new_embedding, 'active', :run_id, NOW()
)
RETURNING fact_id AS new_fact_id;

-- Step 2: Mark old fact as superseded
UPDATE fact_memory
   SET superseded_by = :new_fact_id,
       status        = 'superseded',
       updated_at    = NOW()
 WHERE tenant_id = :tenant_id
   AND subject   = :subject
   AND predicate = :predicate
   AND status    = 'active'
   AND fact_id  != :new_fact_id;

Retrieval queries always filter WHERE superseded_by IS NULL, so superseded facts never appear in results — but they remain in the database for audit and replay.

GDPR deletion

When a user exercises the right to be forgotten, the system must delete or anonymise all memory records tied to that user.

Python — GDPR cascade deletion:

async def gdpr_delete_user(tenant_id: str, user_id: str):
    # 1. Aurora — tombstone pattern
    await aurora.execute("""
        UPDATE fact_memory
           SET status       = 'deleted',
               content      = '[DELETED]',
               embedding    = NULL,
               metadata     = NULL,
               deleted_at   = NOW()
         WHERE tenant_id = :tenant_id
           AND user_id   = :user_id
    """, {'tenant_id': tenant_id, 'user_id': user_id})

    # 2. DynamoDB — hard delete
    preference_keys = dynamo.query(
        TableName="preference_memory",
        KeyConditionExpression="user_id = :uid",
        ExpressionAttributeValues={":uid": user_id}
    )
    for item in preference_keys["Items"]:
        dynamo.delete_item(
            TableName="preference_memory",
            Key={"user_id": item["user_id"], "pref_key": item["pref_key"]}
        )

    # 3. S3 trace logs — EventBridge → Batch Delete
    s3_prefix = f"traces/tenant={tenant_id}/user={user_id}/"
    await s3_batch_delete(bucket="memory-trace-logs", prefix=s3_prefix)

    # 4. OpenSearch — delete by query
    await opensearch.delete_by_query(
        index="fact_memory",
        body={"query": {"term": {"user_id": user_id}}}
    )

14. Summarisation and distillation from trace

Long conversations produce too much trace to keep in working memory. Summarisation is the process of distilling a sequence of turns into a reusable artifact.

Two distillation modes:

• Rolling summarisation — every N turns, summarise the last M turns and replace them with the summary. Used for very long conversations (100+ turns).
• Episodic extraction — when a task is completed, extract a structured summary of what was accomplished and store it as an episode. Future sessions can retrieve this episode instead of re-deriving the solution.

Episodic extraction with Step Functions

Diagram 5: Episodic extraction workflow. When a task completes, Step Functions orchestrates: retrieve trace → extract summary with Bedrock → embed → promote to Aurora. The user never waits for this — it runs asynchronously.

Python — Episodic extraction Lambda function:

import boto3
import json

bedrock = boto3.client("bedrock-runtime")
aurora   = boto3.client("rds-data")

def lambda_handler(event, context):
    run_id    = event["run_id"]
    tenant_id = event["tenant_id"]
    user_id   = event["user_id"]

    # 1. Retrieve trace from S3
    trace_events = retrieve_trace_from_s3(run_id)
    transcript   = "\n".join([
        f"{e['role']}: {e['content']}" for e in trace_events
    ])

    # 2. Extract structured summary with Bedrock
    response = bedrock.converse(
        modelId="amazon.nova-micro-v1:0",  # cheap model for extraction
        messages=[{
            "role": "user",
            "content": f"""Extract a structured summary of this completed task.

Transcript:
{transcript}

Return JSON:
{{
  "task": "one-line description",
  "outcome": "what was accomplished",
  "key_decisions": ["decision 1", "decision 2"],
  "reusable_pattern": "future applicability"
}}"""
        }]
    )

    summary = json.loads(response["content"][0]["text"])

    # 3. Embed summary
    embed_response = bedrock.invoke_model(
        modelId="amazon.titan-embed-text-v2:0",
        body=json.dumps({"inputText": summary["task"] + " " + summary["outcome"]})
    )
    embedding = json.loads(embed_response["body"])["embedding"]

    # 4. Promote to Aurora episodic_memory table
    aurora.execute_statement(
        resourceArn=os.environ["AURORA_ARN"],
        secretArn=os.environ["SECRET_ARN"],
        database="memory",
        sql="""
            INSERT INTO episodic_memory (
                episode_id, tenant_id, user_id, task, outcome,
                key_decisions, reusable_pattern, embedding,
                source_run_id, created_at
            ) VALUES (
                gen_random_uuid(), :tenant_id, :user_id, :task, :outcome,
                :key_decisions, :reusable_pattern, :embedding,
                :run_id, NOW()
            )
        """,
        parameters=[
            {"name": "tenant_id", "value": {"stringValue": tenant_id}},
            {"name": "user_id", "value": {"stringValue": user_id}},
            {"name": "task", "value": {"stringValue": summary["task"]}},
            {"name": "outcome", "value": {"stringValue": summary["outcome"]}},
            {"name": "key_decisions", "value": {"stringValue": json.dumps(summary["key_decisions"])}},
            {"name": "reusable_pattern", "value": {"stringValue": summary["reusable_pattern"]}},
            {"name": "embedding", "value": {"stringValue": json.dumps(embedding)}},
            {"name": "run_id", "value": {"stringValue": run_id}}
        ]
    )

    return {"status": "extracted", "episode_id": "..."}

15. Scope, privacy, and GDPR

Every memory record carries a scope: tenant_id, user_id, agent_id. The scope determines who can read the record. This isn't just business logic — it's enforced at the database layer via PostgreSQL Row-Level Security (RLS). Scope violations are impossible, not just discouraged.

GDPR adds another requirement: the right to be forgotten. A deletion request must cascade across all five memory stores — Aurora, DynamoDB, OpenSearch, S3, and any derived indexes — in a single tracked operation.

Figure 13: Scope enforcement via PostgreSQL RLS (every query is filtered by tenant_id + user_id at the database layer) and GDPR cascade deletion across all five memory stores.

Scope is enforced as a hard predicate before any ranking or scoring:

Row-Level Security (RLS) on Aurora

SQL — PostgreSQL RLS policy:

-- Enable RLS on fact_memory table
ALTER TABLE fact_memory ENABLE ROW LEVEL SECURITY;

-- Policy: users can only see facts scoped to their tenant + user
CREATE POLICY fact_memory_tenant_isolation
    ON fact_memory
    FOR SELECT
    USING (
        tenant_id = current_setting('app.tenant_id')::text
        AND (
            user_id IS NULL  -- tenant-scoped fact, visible to all users in tenant
            OR user_id = current_setting('app.user_id')::text
        )
    );

-- Before every query, set session variables:
SET app.tenant_id = 'tenant-abc-123';
SET app.user_id   = 'user-xyz-789';

-- Now all queries are automatically scoped:
SELECT * FROM fact_memory WHERE status = 'active';
-- PostgreSQL RLS adds: AND tenant_id = 'tenant-abc-123' AND ...

This is the strongest form of tenant isolation — even if application code forgets to add WHERE tenant_id = ?, the database enforces it.

Scope visibility rules