Memory with provenance

MemoryStore is an async, SQLite-backed key-value store for agent memory where every entry carries a trust tier and a source. Values from delegated, external, or untrusted tiers are passed through the prompt-injection sanitiser before storage — the primary defence against MINJA-style memory-poisoning attacks.

The five trust tiers

Defined as TrustTier in kitelogik.memory.models:

Tier	Source examples	Sanitised on write?
TRUSTED	Internal verified systems (authoritative)	No
INTERNAL	Internal but not cryptographically verified	No
DELEGATED	Written by a delegated worker agent (depth > 0)	Yes
EXTERNAL	External tool outputs / MCP responses	Yes
UNTRUSTED	Unknown origin — treat as adversarial	Yes

The sanitised tiers — DELEGATED, EXTERNAL, UNTRUSTED — are scrubbed through kitelogik.tether.sanitizer.sanitize_tool_output before persistence. The MemoryEntry records sanitized=True if the value was modified.

Set up the store

from kitelogik import MemoryStore, TrustTier

mem = MemoryStore(db_path="memory.db")
await mem.setup()                      # one-time table creation

In-memory SQLite (":memory:") is explicitly rejected with a clear error — it would silently lose state across thread hops. Use a real file path; for tests, point at tempfile.mkdtemp() + "/memory.db".

Write with provenance

# Authoritative internal write — no sanitisation
await mem.write(
    key="customer:c1:plan",
    value="enterprise",
    trust_tier=TrustTier.TRUSTED,
    source="internal:crm",
    session_id="sess_001",
)

# External MCP server response — sanitised on write
await mem.write(
    key="recent_search",
    value=mcp_response.content,
    trust_tier=TrustTier.EXTERNAL,
    source="mcp:search-server",
    session_id="sess_001",
)

write() returns the persisted MemoryEntry with sanitized=True/False so you know whether the sanitiser modified the value:

entry = await mem.write(
    key="user_note",
    value=untrusted_text,
    trust_tier=TrustTier.UNTRUSTED,
    source="user:upload",
    session_id="sess_001",
)

if entry.sanitized:
    print("Injection patterns were redacted before storage")

Read with provenance

read() returns the full MemoryEntry (or None):

entry = await mem.read("customer:c1:plan")

if entry is None:
    ...
else:
    # Always inspect the trust tier before using the value
    if entry.trust_tier == TrustTier.UNTRUSTED:
        # Don't echo it directly into a system prompt
        ...
    else:
        prompt += f"\nKnown plan: {entry.value}"

Reads always return the trust tier so the agent (or the session layer) can decide how much weight to give the recalled value.

List keys

keys = await mem.list_keys(session_id="sess_001")    # session-scoped
keys = await mem.list_keys()                          # all sessions

Returned in updated_at DESC order so the most recent writes come first.

Why provenance matters

MINJA (Memory INjection Attack) and similar attacks work by getting a tool result, web page, or document into the agent's memory store without a trust signal. Later, the agent treats the poisoned memory as authoritative — the attacker has effectively written the system prompt.

MemoryStore blocks this in two ways:

1. Sanitise on write — DELEGATED, EXTERNAL, and UNTRUSTED values pass through the prompt-injection scanner before they hit storage. Known injection patterns are redacted; the MemoryEntry.sanitized=True flag preserves the audit signal.
2. Tier-aware read — every read returns trust_tier. Your agent loop can choose to weight tiers differently, refuse to splice untrusted content into prompts, or pass the value through the sanitiser again before use.

The trust tier doesn't change automatically based on the writing session — a delegated agent (depth > 0) can write TRUSTED if you let it, but the convention is that writes from delegated sessions are tagged DELEGATED so the recipient knows they didn't come from an authoritative source.

Pattern: tier the writes by who wrote them

def tier_for_session(context: SessionContext) -> TrustTier:
    if context.delegation_depth > 0:
        return TrustTier.DELEGATED
    if context.user_role in {"system", "internal_service"}:
        return TrustTier.TRUSTED
    return TrustTier.INTERNAL

await mem.write(
    key=f"task:{task_id}:result",
    value=result_text,
    trust_tier=tier_for_session(context),
    source=f"agent:{context.session_id}",
    session_id=context.session_id,
)

Pattern: tier MCP server responses

If you're using MCP tools, every server response is EXTERNAL by default — the call originates outside your trust boundary:

async def remember_mcp_result(mcp_call_name: str, response: str) -> None:
    await mem.write(
        key=f"mcp:{mcp_call_name}:last",
        value=response,
        trust_tier=TrustTier.EXTERNAL,        # ← always
        source=f"mcp:{server_name}",
        session_id=context.session_id,
    )

Sanitisation happens automatically on write.

What the audit log captures

Every MemoryEntry row carries:

• key, value, trust_tier, source, session_id
• created_at, updated_at (UTC ISO strings)
• sanitized (was the value modified by the sanitiser)
• tenant_id (for multi-tenant isolation)

So you can answer "which session wrote this value, from what tier, when, and was the sanitiser triggered?" directly from the SQLite table — no separate audit log lookup required.

Storage

MemoryStore uses SQLite by default — single-file, fine for a single-process agent or a small team. The store interface is small (write, read, query), so a Postgres or other backend swap is a subclass exercise. Trust tiers, the sanitiser, and the per-write provenance metadata are all defined above the storage layer and stay unchanged.

Related

• Governance events — data_classification field for in-flight data, complementary to memory tiers
• data_classification.rego — flow control on classification labels at evaluation time
• Architecture — where MemoryStore sits in the Anchor layer