Problem
When you need this. A remembered preference changes, a booking code expires, and a user asks the assistant to forget a passport number. The store needs different operations for correction, expiry and deletion.
The first version of the memory service stores memories without expiry or deletion. Real memories change. Alice tells the assistant she is vegetarian; two days later she says she eats fish. A hotel booking code matters for a week. A passport number should never have been stored, and Alice asks the assistant to forget it.
Each of these breaks the first store in a different way. Storing the correction as one more memory leaves both statements in recall, and the assistant has no way to tell which is current. Overwriting the old text in place removes the record of what the assistant used to believe and where it came from. Deleting a row with a plain DELETE removes it from queries, but in our test the text was still readable in the database file afterwards.
This build adds four operations to the store: correct, expiry, forget and history. It also migrates an existing version 1 file to the new schema without losing data.
Success check: after the scenario, recall returns the corrected diet and never the old one, the booking code is gone after its seven days, the passport number cannot be recalled and is not in the database file, and a deletion record names the deleted IDs without their text.
You need: Python 3.11 or later and the files from this build. Nothing is downloaded.
Concepts
Provenance: where a memory came from
Provenance is the record of where a piece of information came from. Each memory here keeps a source, such as trip-planning/turn-3, and a creation time. When the assistant says something wrong, provenance lets you find the turn that produced the memory and decide whether the extraction or the user’s statement was the problem.
A correction is a new version
A correction does not edit the old row. correct inserts a new memory whose supersedes column points at the old one. The old row stays as history; recall skips any memory that a newer one supersedes. A memory can be superseded once, enforced by a UNIQUE constraint, so versions form a single chain and there is always one current version.
Only the current version can be corrected. If a caller holds an old ID, correct refuses it rather than forking the chain.
Expiry: a time limit set when the memory is stored
Some memories are only true for a while. remember(..., ttl_days=7) sets expires_at seven days after creation. From that moment recall skips the memory, even if no cleanup job has run yet. purge_expired() deletes expired memories from the file. If the newest version of a chain has expired, it deletes the whole chain: deleting only the newest version would make the older version current again.
Forgetting: delete every version, then check the file
forget deletes the memory and every version in its chain, then writes a row to deletions with the user, the deleted IDs, the reason and the time. The record has no memory text. It is there so you can show that a deletion happened without keeping what was deleted. The reason is free text, so the caller must not copy the forgotten content into it: the store cannot tell, and the record is kept.
A deleted row can stay in the database file, as the vector store lifecycle build showed for deleted passages. This store turns on the same fix, SQLite’s secure_delete, which overwrites deleted content with zeros.
Architecture
The store is still one Python object over one SQLite file, now at schema version 2. The memories table gains expires_at and supersedes; a new deletions table holds the text-free records. Every read still goes through one query filtered by user and task, and that query now also returns which memory, if any, supersedes each row.
The saved scenario result retains each step and its outcome. On day 2, memory #4 supersedes #1, and recall returns only the new version. On day 8, the booking code’s seven days have passed and recall returns nothing, before any purge runs. The purge then removes 1 memory, and forget deletes #3. With secure_delete on, the passport number is not in the file afterwards; with it off, the same scenario leaves it there.
Working code
Get the files
Download the complete lab, extract it and open a terminal in build-memory-provenance-and-deletion/. It includes the source, fixtures, recorded results and tests listed in the README.
python3.11 -m venv .venv
source .venv/bin/activateThe lab uses the Python standard library; there are no packages to install.
The version 2 store
"""Memory service, version 2: provenance, corrections, expiry and deletion.
Standard library only. Extends the version 1 store from the memory service
build. A correction adds a new memory that supersedes the old one; recall
returns only current, unexpired memories; forget deletes every version of a
memory and records the deletion without its text.
"""
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
import json
import re
import sqlite3
SCHEMA_VERSION = 2
V1_SCHEMA = """
CREATE TABLE memories (
id INTEGER PRIMARY KEY,
user_id TEXT NOT NULL,
task_id TEXT,
text TEXT NOT NULL CHECK (length(trim(text)) > 0),
source TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE INDEX memories_by_scope ON memories (user_id, task_id);
"""
V2_TABLES = [
"""CREATE TABLE memories_v2 (
id INTEGER PRIMARY KEY,
user_id TEXT NOT NULL,
task_id TEXT,
text TEXT NOT NULL CHECK (length(trim(text)) > 0),
source TEXT NOT NULL,
created_at TEXT NOT NULL,
expires_at TEXT,
supersedes INTEGER UNIQUE REFERENCES memories_v2 (id)
)""",
"""CREATE TABLE deletions (
id INTEGER PRIMARY KEY,
user_id TEXT NOT NULL,
memory_ids TEXT NOT NULL,
reason TEXT NOT NULL,
deleted_at TEXT NOT NULL
)""",
]
V2_FINISH = [
'ALTER TABLE memories_v2 RENAME TO memories',
'CREATE INDEX memories_by_scope ON memories (user_id, task_id)',
f'PRAGMA user_version = {SCHEMA_VERSION}',
]
IDENTIFIER = re.compile(r'[a-z0-9][a-z0-9-]{0,63}')
MAX_TEXT = 2_000
WORD = re.compile(r'[a-z0-9]+')
STOPWORDS = frozenset('a an and are as at be by did do does for from how i in is it my of on or that the this '
'to was we what when where which who will with you your'.split())
class MemoryInputError(ValueError):
"""Invalid scope, text, source or setting; nothing was written."""
class MemoryNotFound(LookupError):
"""No such memory in this scope. The same error covers another user's memory."""
class SchemaError(RuntimeError):
"""The database file has a schema this code does not understand."""
def _identifier(value, where):
if not isinstance(value, str) or not IDENTIFIER.fullmatch(value):
raise MemoryInputError(f'{where} must match {IDENTIFIER.pattern}')
return value
@dataclass(frozen=True)
class Scope:
user: str
task: str
def __post_init__(self):
_identifier(self.user, 'scope.user')
_identifier(self.task, 'scope.task')
class Clock:
"""A deterministic clock: each reading is one minute later; advance() jumps ahead."""
def __init__(self, start: str = '2026-10-20T09:00:00+00:00'):
self.moment = datetime.fromisoformat(start)
def now(self) -> str:
value = self.moment
self.moment += timedelta(minutes=1)
return value.isoformat()
def peek(self) -> str:
return self.moment.isoformat()
def advance(self, **delta):
self.moment += timedelta(**delta)
def words(text: str) -> set[str]:
return set(WORD.findall(text.lower())) - STOPWORDS
def _text(value, where):
if not isinstance(value, str) or not value.strip() or len(value) > MAX_TEXT:
raise MemoryInputError(f'{where} must be nonblank and at most {MAX_TEXT} characters')
return value
class MemoryStore:
def __init__(self, path: str, clock: Clock, secure_delete: bool = True):
self.clock = clock
self.db = sqlite3.connect(path)
self.db.execute('PRAGMA foreign_keys = ON')
# Overwrite deleted rows with zeros so forgotten text does not stay in the file.
self.db.execute(f'PRAGMA secure_delete = {"ON" if secure_delete else "OFF"}')
version = self.db.execute('PRAGMA user_version').fetchone()[0]
tables = {row[0] for row in self.db.execute("SELECT name FROM sqlite_master WHERE type = 'table'")}
if version == 0 and not tables:
self._migrate(create_v1=True)
elif version == 1 and tables == {'memories'}:
self._migrate(create_v1=False)
elif version != SCHEMA_VERSION:
self.db.close()
raise SchemaError(f'database schema version {version}; this code reads versions 1 and {SCHEMA_VERSION}')
def _migrate(self, create_v1: bool):
"""Move a new or version 1 file to version 2 in one transaction."""
try:
self.db.execute('BEGIN')
if create_v1:
for statement in filter(str.strip, V1_SCHEMA.split(';')):
self.db.execute(statement)
for statement in V2_TABLES:
self.db.execute(statement)
self.db.execute('INSERT INTO memories_v2 (id, user_id, task_id, text, source, created_at) '
'SELECT id, user_id, task_id, text, source, created_at FROM memories')
self.db.execute('DROP TABLE memories')
for statement in V2_FINISH:
self.db.execute(statement)
self.db.commit()
except BaseException:
self.db.rollback()
self.db.close()
raise
def close(self):
self.db.close()
# Every read of memories goes through this scoped query.
def _rows(self, scope: Scope) -> list[dict]:
cursor = self.db.execute(
'SELECT id, task_id, text, source, created_at, expires_at, supersedes, '
'(SELECT newer.id FROM memories AS newer WHERE newer.supersedes = memories.id) AS superseded_by '
'FROM memories WHERE user_id = ? AND (task_id IS NULL OR task_id = ?) ORDER BY id',
(scope.user, scope.task))
names = [column[0] for column in cursor.description]
return [dict(zip(names, row)) for row in cursor.fetchall()]
def _find(self, scope: Scope, memory_id: int) -> tuple[dict, dict]:
rows = {row['id']: row for row in self._rows(scope)}
if memory_id not in rows:
raise MemoryNotFound(f'no memory {memory_id} in this scope')
return rows[memory_id], rows
def _expiry(self, created: str, ttl_days: int | None) -> str | None:
if ttl_days is None:
return None
if type(ttl_days) is not int or not 1 <= ttl_days <= 3650:
raise MemoryInputError('ttl_days must be an integer from 1 to 3650')
return (datetime.fromisoformat(created) + timedelta(days=ttl_days)).isoformat()
def remember(self, scope: Scope, level: str, text: str, source: str, ttl_days: int | None = None) -> int:
if level not in ('user', 'task'):
raise MemoryInputError('level must be user or task')
_text(text, 'memory text')
_text(source, 'source')
self._expiry(self.clock.peek(), ttl_days) # validate before reading the clock
created = self.clock.now()
with self.db:
cursor = self.db.execute(
'INSERT INTO memories (user_id, task_id, text, source, created_at, expires_at) VALUES (?, ?, ?, ?, ?, ?)',
(scope.user, scope.task if level == 'task' else None, text, source, created,
self._expiry(created, ttl_days)))
return cursor.lastrowid
def correct(self, scope: Scope, memory_id: int, text: str, source: str, ttl_days: int | None = None) -> int:
"""Store a corrected memory that supersedes the current one. The old version stays as history."""
old, _ = self._find(scope, memory_id)
if old['superseded_by'] is not None:
raise MemoryInputError(f'memory {memory_id} was already corrected by {old["superseded_by"]}; correct the newest')
_text(text, 'memory text')
_text(source, 'source')
self._expiry(self.clock.peek(), ttl_days)
created = self.clock.now()
with self.db:
cursor = self.db.execute(
'INSERT INTO memories (user_id, task_id, text, source, created_at, expires_at, supersedes) '
'VALUES (?, ?, ?, ?, ?, ?, ?)',
(scope.user, old['task_id'], text, source, created, self._expiry(created, ttl_days), memory_id))
return cursor.lastrowid
def history(self, scope: Scope, memory_id: int) -> list[dict]:
"""Every version of a memory, oldest first, with its source."""
_, rows = self._find(scope, memory_id)
chain = [rows[memory_id]]
while chain[0]['supersedes'] is not None:
chain.insert(0, rows[chain[0]['supersedes']])
while chain[-1]['superseded_by'] is not None:
chain.append(rows[chain[-1]['superseded_by']])
return [{key: row[key] for key in ('id', 'text', 'source', 'created_at', 'supersedes')} for row in chain]
def forget(self, scope: Scope, memory_id: int, reason: str) -> list[int]:
"""Delete every version of a memory and record which IDs were deleted, without text."""
_text(reason, 'reason')
ids = [row['id'] for row in self.history(scope, memory_id)]
marks = ', '.join('?' * len(ids))
with self.db:
# Newest first, so no remaining row refers to a deleted one.
for deleted in reversed(ids):
self.db.execute('DELETE FROM memories WHERE id = ? AND user_id = ?', (deleted, scope.user))
self.db.execute('INSERT INTO deletions (user_id, memory_ids, reason, deleted_at) VALUES (?, ?, ?, ?)',
(scope.user, json.dumps(ids), reason, self.clock.now()))
if self.db.execute(f'SELECT count(*) FROM memories WHERE id IN ({marks})', ids).fetchone()[0]:
raise RuntimeError('a deleted memory is still stored')
return ids
def purge_expired(self) -> int:
"""Delete every version of each memory whose newest version has expired, for every user.
Deleting only the expired newest version would make the older version
it corrected current again. Run this as a scheduled job.
"""
now = self.clock.peek()
expired = self.db.execute(
'SELECT user_id, id FROM memories WHERE expires_at IS NOT NULL AND expires_at <= ? '
'AND NOT EXISTS (SELECT 1 FROM memories AS newer WHERE newer.supersedes = memories.id) ORDER BY id',
(now,)).fetchall()
removed = 0
with self.db:
for user, newest in expired:
chain = [newest]
while (older := self.db.execute('SELECT supersedes FROM memories WHERE id = ? AND user_id = ?',
(chain[-1], user)).fetchone()[0]) is not None:
chain.append(older)
for memory_id in chain: # newest first
self.db.execute('DELETE FROM memories WHERE id = ? AND user_id = ?', (memory_id, user))
self.db.execute('INSERT INTO deletions (user_id, memory_ids, reason, deleted_at) VALUES (?, ?, ?, ?)',
(user, json.dumps(sorted(chain)), 'expired', now))
removed += len(chain)
return removed
def recall(self, scope: Scope, query: str, limit: int = 3) -> list[dict]:
"""Current memories only: not superseded and not expired at the current time."""
if type(limit) is not int or not 1 <= limit <= 20:
raise MemoryInputError('limit must be an integer from 1 to 20')
now = self.clock.peek()
wanted = words(query)
scored = []
for row in self._rows(scope):
if row['superseded_by'] is not None or (row['expires_at'] and row['expires_at'] <= now):
continue
score = len(wanted & words(row['text']))
if score:
scored.append({'id': row['id'], 'level': 'task' if row['task_id'] else 'user', 'text': row['text'],
'source': row['source'], 'supersedes': row['supersedes'],
'expires_at': row['expires_at'], 'score': score})
scored.sort(key=lambda m: (-m['score'], -m['id']))
return scored[:limit]
def deletions(self, scope: Scope) -> list[dict]:
cursor = self.db.execute('SELECT memory_ids, reason, deleted_at FROM deletions WHERE user_id = ? ORDER BY id',
(scope.user,))
return [{'memory_ids': json.loads(ids), 'reason': reason, 'deleted_at': at} for ids, reason, at in cursor]Opening and migrating. A new file is created as version 1 and migrated, so new and old files take the same path. A version 1 file is migrated in one transaction: create memories_v2 and deletions, copy every row, drop the old table, rename, rebuild the index, set user_version = 2. The migration starts with an explicit BEGIN, because in its default mode the sqlite3 module opens transactions only before INSERT, UPDATE, DELETE and REPLACE. If any statement fails, the rollback leaves the file at version 1 with its data.
Reading. _rows is still the only query that reads memories. A subquery adds superseded_by to each row. _find looks a memory up within the scope’s rows, so a memory that belongs to another user raises the same MemoryNotFound message as an ID that does not exist. A different message would tell Bob that memory 1 exists.
Writing. correct checks that the memory is visible and current, then inserts the new version with supersedes set. forget uses history to collect every version, deletes them newest first, and writes the deletion record in the same transaction. Foreign keys are on, so a supersedes value must point at an existing memory.
Time. The clock is injected. recall and purge_expired compare expires_at with the current time without advancing it. A memory with expires_at equal to the current time is expired.
Run the scenario
{
"user": ["alice", "trip-planning"],
"steps": [
{"day": 0, "op": "remember", "key": "diet", "level": "user", "text": "Alice is vegetarian.", "source": "trip-planning/turn-3"},
{"day": 0, "op": "remember", "key": "booking", "level": "task", "text": "The Lisbon hotel booking code is 4F7Q.", "source": "trip-planning/turn-5", "ttl_days": 7},
{"day": 0, "op": "remember", "key": "passport", "level": "user", "text": "Alice's passport number is P1234567.", "source": "trip-planning/turn-6"},
{"day": 0, "op": "recall", "query": "What food does Alice eat? Is she vegetarian?"},
{"day": 2, "op": "correct", "key": "diet", "new_key": "diet-2", "text": "Alice eats fish but no meat, so she is not vegetarian.", "source": "trip-planning/turn-12"},
{"day": 2, "op": "recall", "query": "What food does Alice eat? Is she vegetarian?"},
{"day": 2, "op": "history", "key": "diet-2"},
{"day": 2, "op": "recall", "query": "What is the hotel booking code?"},
{"day": 8, "op": "recall", "query": "What is the hotel booking code?"},
{"day": 8, "op": "purge_expired"},
{"day": 8, "op": "forget", "key": "passport", "reason": "user request at trip-planning/turn-20"},
{"day": 8, "op": "recall", "query": "What is Alice's passport number?"},
{"day": 8, "op": "deletions"}
],
"probe_text": "P1234567"
}"""Replay scenario.json: correct a memory, let one expire, forget one.
The scenario runs twice: with secure_delete on (the store's default) and off.
After each run the demo checks whether the forgotten text is still in the file.
"""
import json
from pathlib import Path
import tempfile
from memory import Clock, MemoryStore, Scope
def replay(scenario: dict, secure_delete: bool) -> dict:
scope = Scope(*scenario['user'])
with tempfile.TemporaryDirectory() as folder:
path = Path(folder) / 'memory.sqlite3'
clock = Clock()
store = MemoryStore(str(path), clock, secure_delete=secure_delete)
ids, day, log = {}, 0, []
for step in scenario['steps']:
clock.advance(days=step['day'] - day)
day = step['day']
op = step['op']
entry = {'day': day, 'op': op}
if op == 'remember':
ids[step['key']] = store.remember(scope, step['level'], step['text'], step['source'],
step.get('ttl_days'))
entry['id'] = ids[step['key']]
elif op == 'correct':
ids[step['new_key']] = store.correct(scope, ids[step['key']], step['text'], step['source'])
entry.update(corrected=ids[step['key']], id=ids[step['new_key']])
elif op == 'recall':
entry.update(query=step['query'], recalled=store.recall(scope, step['query']))
elif op == 'history':
entry['versions'] = store.history(scope, ids[step['key']])
elif op == 'purge_expired':
entry['removed'] = store.purge_expired()
elif op == 'forget':
entry['deleted_ids'] = store.forget(scope, ids[step['key']], step['reason'])
elif op == 'deletions':
entry['records'] = store.deletions(scope)
log.append(entry)
store.close()
raw = path.read_bytes()
return {'secure_delete': secure_delete, 'steps': log,
'forgotten_text_in_file': scenario['probe_text'].encode() in raw}
if __name__ == '__main__':
scenario = json.loads(Path(__file__).with_name('scenario.json').read_text())
on, off = replay(scenario, True), replay(scenario, False)
assert on['steps'] == off['steps']
print(json.dumps({'steps': on['steps'],
'forgotten_text_in_file': {'secure_delete_on': on['forgotten_text_in_file'],
'secure_delete_off': off['forgotten_text_in_file']}}, indent=2))From the directory with the files, run:
python3 demo.pyThe demo replays the scenario twice in temporary files, once with secure_delete on and once off, checks that both runs return the same results, and reads the raw bytes of each file after the store is closed. The history of the corrected diet memory:
[
{
"id": 1,
"text": "Alice is vegetarian.",
"source": "trip-planning/turn-3",
"created_at": "2026-10-20T09:00:00+00:00",
"supersedes": null
},
{
"id": 4,
"text": "Alice eats fish but no meat, so she is not vegetarian.",
"source": "trip-planning/turn-12",
"created_at": "2026-10-22T09:03:00+00:00",
"supersedes": 1
}
]The recorded file check: {"secure_delete_on":false,"secure_delete_off":true}.
Notice the first recall on day 0: the food question also returns the passport memory, because both contain the word alice. Word-overlap recall can bring sensitive memories into unrelated requests. That is a retrieval problem, and it is one more reason to store as little sensitive data as possible.
Inspect the complete recorded output
{
"steps": [
{
"day": 0,
"op": "remember",
"id": 1
},
{
"day": 0,
"op": "remember",
"id": 2
},
{
"day": 0,
"op": "remember",
"id": 3
},
{
"day": 0,
"op": "recall",
"query": "What food does Alice eat? Is she vegetarian?",
"recalled": [
{
"id": 1,
"level": "user",
"text": "Alice is vegetarian.",
"source": "trip-planning/turn-3",
"supersedes": null,
"expires_at": null,
"score": 2
},
{
"id": 3,
"level": "user",
"text": "Alice's passport number is P1234567.",
"source": "trip-planning/turn-6",
"supersedes": null,
"expires_at": null,
"score": 1
}
]
},
{
"day": 2,
"op": "correct",
"corrected": 1,
"id": 4
},
{
"day": 2,
"op": "recall",
"query": "What food does Alice eat? Is she vegetarian?",
"recalled": [
{
"id": 4,
"level": "user",
"text": "Alice eats fish but no meat, so she is not vegetarian.",
"source": "trip-planning/turn-12",
"supersedes": 1,
"expires_at": null,
"score": 3
},
{
"id": 3,
"level": "user",
"text": "Alice's passport number is P1234567.",
"source": "trip-planning/turn-6",
"supersedes": null,
"expires_at": null,
"score": 1
}
]
},
{
"day": 2,
"op": "history",
"versions": [
{
"id": 1,
"text": "Alice is vegetarian.",
"source": "trip-planning/turn-3",
"created_at": "2026-10-20T09:00:00+00:00",
"supersedes": null
},
{
"id": 4,
"text": "Alice eats fish but no meat, so she is not vegetarian.",
"source": "trip-planning/turn-12",
"created_at": "2026-10-22T09:03:00+00:00",
"supersedes": 1
}
]
},
{
"day": 2,
"op": "recall",
"query": "What is the hotel booking code?",
"recalled": [
{
"id": 2,
"level": "task",
"text": "The Lisbon hotel booking code is 4F7Q.",
"source": "trip-planning/turn-5",
"supersedes": null,
"expires_at": "2026-10-27T09:01:00+00:00",
"score": 3
}
]
},
{
"day": 8,
"op": "recall",
"query": "What is the hotel booking code?",
"recalled": []
},
{
"day": 8,
"op": "purge_expired",
"removed": 1
},
{
"day": 8,
"op": "forget",
"deleted_ids": [
3
]
},
{
"day": 8,
"op": "recall",
"query": "What is Alice's passport number?",
"recalled": [
{
"id": 4,
"level": "user",
"text": "Alice eats fish but no meat, so she is not vegetarian.",
"source": "trip-planning/turn-12",
"supersedes": 1,
"expires_at": null,
"score": 1
}
]
},
{
"day": 8,
"op": "deletions",
"records": [
{
"memory_ids": [
2
],
"reason": "expired",
"deleted_at": "2026-10-28T09:04:00+00:00"
},
{
"memory_ids": [
3
],
"reason": "user request at trip-planning/turn-20",
"deleted_at": "2026-10-28T09:04:00+00:00"
}
]
}
],
"forgotten_text_in_file": {
"secure_delete_on": false,
"secure_delete_off": true
}
}Check your understanding
Suppose purge_expired deleted only expired rows that no newer row supersedes, and nothing else. A booking code was stored with a seven-day expiry and corrected on day 3, also with a seven-day expiry. What does recall return on day 11, after the purge? Work it out, then read test_purge_removes_a_whole_expired_chain. The day 3 correction expires on day 10 and is deleted. The original then has nothing superseding it, so it would count as current again, even though it expired on day 7. Recall still skips it by its own expiry, but history and any code that reads current memories without checking expiry would see a fact that was replaced. Deleting the whole chain avoids the problem.
Read the explanation
Expiry is enforced at read time. Purging is cleanup; it is not what keeps an expired memory out of a request.
Failure cases
Run the tests:
python3 -m unittest -v test_memory.pyThe suite has 13 tests. The first runs demo.py and compares its output with the recording. The others cover these cases:
| Case | Expected behavior |
|---|---|
| A fact is corrected | Recall returns only the new version; history returns both with their sources. |
| An old version is corrected again | Reject; only the newest version can be corrected. |
| Another user’s memory is corrected, forgotten or inspected | Raise the same MemoryNotFound message as for a missing ID. |
| A memory reaches its expiry time | Recalled one second before; not recalled at the exact time. |
| The purge runs at a memory’s exact expiry time | Deleted at that time; kept one second before. |
| The newest version of a corrected memory expires | Purge deletes every version and records it. |
| A memory is forgotten | Every version is deleted; the record lists IDs and reason, not text. |
| Forgotten text in the database file | Gone with secure_delete on; still present with it off. |
| A version 1 file is opened | Migrated to version 2 with data, sources and foreign keys. |
| The migration fails partway | The file stays at version 1 with its data. |
| A file with another version or other tables | Refuse to open it. |
| Invalid expiry, level, text or reason | Reject before writing. |
To check that the tests catch real defects, we changed the store one rule at a time. Turning secure_delete off fails the file test and the fixture test. Keeping superseded memories in recall fails the correction test. Using < instead of <= for expiry fails the boundary test. Purging only the newest version fails the chain test. Forgetting one version fails the forget test. Removing the explicit BEGIN fails the migration test. Allowing corrections of old versions fails the newest-version test.
Inspect the failure tests
from datetime import datetime, timedelta
import json
from pathlib import Path
import sqlite3
import subprocess
import sys
import tempfile
import unittest
import memory
from memory import Clock, MemoryInputError, MemoryNotFound, MemoryStore, SchemaError, Scope, V1_SCHEMA
HERE = Path(__file__).parent
ALICE = Scope('alice', 'trip-planning')
BOB = Scope('bob', 'onboarding')
class ProvenanceAndDeletionTest(unittest.TestCase):
def setUp(self):
folder = tempfile.TemporaryDirectory()
self.addCleanup(folder.cleanup)
self.path = str(Path(folder.name) / 'memory.sqlite3')
self.clock = Clock()
def open(self, **options):
store = MemoryStore(self.path, self.clock, **options)
self.addCleanup(store.close)
return store
def test_fixture_matches_recorded_result(self):
output = subprocess.run([sys.executable, 'demo.py'], cwd=HERE, check=True,
capture_output=True, text=True).stdout
self.assertEqual(json.loads(output), json.loads((HERE / 'memory-result.json').read_text()))
def test_correction_replaces_the_fact_and_keeps_history(self):
store = self.open()
old = store.remember(ALICE, 'user', 'Alice is vegetarian.', 'turn-3')
new = store.correct(ALICE, old, 'Alice eats fish but no meat.', 'turn-12')
self.assertEqual([m['id'] for m in store.recall(ALICE, 'Alice vegetarian fish meat', limit=5)], [new])
versions = store.history(ALICE, old)
self.assertEqual([(v['id'], v['source'], v['supersedes']) for v in versions],
[(old, 'turn-3', None), (new, 'turn-12', old)])
self.assertEqual(store.history(ALICE, new), versions)
def test_only_the_newest_version_can_be_corrected(self):
store = self.open()
old = store.remember(ALICE, 'user', 'Alice is vegetarian.', 'turn-3')
store.correct(ALICE, old, 'Alice eats fish.', 'turn-12')
with self.assertRaisesRegex(MemoryInputError, 'already corrected'):
store.correct(ALICE, old, 'Alice eats meat.', 'turn-13')
def test_other_users_memories_look_missing(self):
store = self.open()
alice_memory = store.remember(ALICE, 'user', 'Alice is vegetarian.', 'turn-3')
for action in (lambda: store.correct(BOB, alice_memory, 'Changed.', 'turn-1'),
lambda: store.forget(BOB, alice_memory, 'Bob asked.'),
lambda: store.history(BOB, alice_memory)):
with self.assertRaises(MemoryNotFound) as caught:
action()
self.assertEqual(str(caught.exception), f'no memory {alice_memory} in this scope')
with self.assertRaises(MemoryNotFound) as missing:
store.history(BOB, 999)
self.assertEqual(str(missing.exception), 'no memory 999 in this scope')
self.assertEqual(len(store.recall(ALICE, 'vegetarian')), 1)
def test_memory_expires_at_its_exact_time(self):
store = self.open()
store.remember(ALICE, 'task', 'The hotel booking code is 4F7Q.', 'turn-5', ttl_days=7)
expires = datetime.fromisoformat(store.recall(ALICE, 'hotel booking code')[0]['expires_at'])
self.clock.moment = expires - timedelta(seconds=1)
self.assertEqual(len(store.recall(ALICE, 'hotel booking code')), 1)
self.clock.moment = expires
self.assertEqual(store.recall(ALICE, 'hotel booking code'), [])
def test_purge_removes_a_memory_at_its_exact_expiry(self):
store = self.open()
store.remember(ALICE, 'task', 'The hotel booking code is 4F7Q.', 'turn-5', ttl_days=7)
expires = datetime.fromisoformat(store.recall(ALICE, 'hotel booking code')[0]['expires_at'])
self.clock.moment = expires - timedelta(seconds=1)
self.assertEqual(store.purge_expired(), 0)
self.clock.moment = expires
self.assertEqual(store.purge_expired(), 1)
def test_purge_removes_a_whole_expired_chain(self):
store = self.open()
first = store.remember(ALICE, 'task', 'The booking code is 4F7Q.', 'turn-5', ttl_days=7)
second = store.correct(ALICE, first, 'The booking code is 9K2M.', 'turn-9', ttl_days=7)
self.clock.advance(days=8)
self.assertEqual(store.purge_expired(), 2)
self.assertEqual(store.recall(ALICE, 'booking code'), [])
with self.assertRaises(MemoryNotFound):
store.history(ALICE, first)
self.assertEqual(store.deletions(ALICE), [{'memory_ids': [first, second], 'reason': 'expired',
'deleted_at': self.clock.peek()}])
def test_forget_deletes_every_version_and_records_no_text(self):
store = self.open()
old = store.remember(ALICE, 'user', "Alice's passport number is P1234567.", 'turn-6')
new = store.correct(ALICE, old, "Alice's passport number is P7654321.", 'turn-7')
self.assertEqual(store.forget(ALICE, new, 'user request'), [old, new])
self.assertEqual(store.recall(ALICE, 'passport number'), [])
records = store.deletions(ALICE)
self.assertEqual(records[0]['memory_ids'], [old, new])
self.assertNotIn('P1234567', json.dumps(records))
self.assertEqual(store.deletions(BOB), [])
def test_forgotten_text_is_overwritten_in_the_file(self):
# None opens the store with its default, which must be secure_delete on.
for secure, expected in ((None, False), (False, True)):
with self.subTest(secure_delete=secure):
path = Path(self.path).with_name(f'secure-{secure}.sqlite3')
options = {} if secure is None else {'secure_delete': secure}
store = MemoryStore(str(path), self.clock, **options)
memory_id = store.remember(ALICE, 'user', "Alice's passport number is P1234567.", 'turn-6')
store.remember(ALICE, 'user', 'Alice is vegetarian.', 'turn-3')
store.forget(ALICE, memory_id, 'user request')
store.close()
self.assertEqual(b'P1234567' in path.read_bytes(), expected)
def test_version_1_file_is_migrated(self):
db = sqlite3.connect(self.path)
db.executescript(V1_SCHEMA)
db.execute("INSERT INTO memories (user_id, task_id, text, source, created_at) "
"VALUES ('alice', NULL, 'Alice is vegetarian.', 'turn-3', '2026-10-01T09:00:00+00:00')")
db.execute('PRAGMA user_version = 1')
db.commit()
db.close()
store = self.open()
self.assertEqual(store.db.execute('PRAGMA user_version').fetchone()[0], 2)
self.assertEqual(store.recall(ALICE, 'vegetarian')[0]['source'], 'turn-3')
new = store.correct(ALICE, 1, 'Alice eats fish.', 'turn-12')
self.assertEqual(store.history(ALICE, new)[0]['id'], 1)
with self.assertRaises(sqlite3.IntegrityError):
with store.db:
store.db.execute("INSERT INTO memories (user_id, text, source, created_at, supersedes) "
"VALUES ('alice', 'x', 's', 't', 999)")
def test_failed_migration_leaves_version_1_intact(self):
db = sqlite3.connect(self.path)
db.executescript(V1_SCHEMA)
db.execute("INSERT INTO memories (user_id, task_id, text, source, created_at) "
"VALUES ('alice', NULL, 'Alice is vegetarian.', 'turn-3', '2026-10-01T09:00:00+00:00')")
db.execute('PRAGMA user_version = 1')
db.commit()
db.close()
original = memory.V2_FINISH
memory.V2_FINISH = [*original[:1], 'THIS IS NOT SQL', *original[1:]]
try:
with self.assertRaises(sqlite3.OperationalError):
MemoryStore(self.path, self.clock)
finally:
memory.V2_FINISH = original
db = sqlite3.connect(self.path)
self.assertEqual(db.execute('PRAGMA user_version').fetchone()[0], 1)
self.assertEqual([row[0] for row in db.execute("SELECT name FROM sqlite_master WHERE type = 'table'")],
['memories'])
self.assertEqual(db.execute('SELECT text FROM memories').fetchall(), [('Alice is vegetarian.',)])
db.close()
def test_unknown_or_foreign_files_are_refused(self):
db = sqlite3.connect(self.path)
db.execute('CREATE TABLE notes (body TEXT)')
db.execute('PRAGMA user_version = 1')
db.commit()
db.close()
with self.assertRaises(SchemaError):
MemoryStore(self.path, self.clock)
db = sqlite3.connect(self.path)
db.execute('PRAGMA user_version = 3')
db.close()
with self.assertRaisesRegex(SchemaError, 'version 3'):
MemoryStore(self.path, self.clock)
def test_invalid_input_is_rejected_before_writing(self):
store = self.open()
memory_id = store.remember(ALICE, 'user', 'Alice is vegetarian.', 'turn-3')
for action in (lambda: store.remember(ALICE, 'user', 'x', 'turn-1', ttl_days=0),
lambda: store.remember(ALICE, 'user', 'x', 'turn-1', ttl_days=True),
lambda: store.remember(ALICE, 'global', 'x', 'turn-1'),
lambda: store.correct(ALICE, memory_id, ' ', 'turn-2'),
lambda: store.forget(ALICE, memory_id, ' ')):
with self.assertRaises(MemoryInputError):
action()
self.assertEqual(len(store.history(ALICE, memory_id)), 1)
self.assertEqual(store.deletions(ALICE), [])
if __name__ == '__main__':
unittest.main()Deployment
Migrating real files
Before you deploy version 2, copy a production database file, open the copy with the new code, and compare row counts and a sample of memories with the original. The migration runs on first open; for a large file, run it as a separate step during a maintenance window rather than inside a user request. Keep the version 1 code and a backup together: version 1 code refuses a version 2 file, so rolling back the code means restoring the backup too.
Running expiry and deletion
Run purge_expired on a schedule. Recall already skips expired memories, so the schedule decides how long expired text stays in the file, not whether the assistant uses it. Run forget when a user asks, and send the same request to every other place the text may be: backups on their own retention schedule, logs, caches and any export.
Keep the deletions table. It shows that a request was handled without keeping the data. Log memory IDs and counts, never memory text.
No server deployment was made for this build. The tests pass on Python 3.11 and 3.15.
Scaling
Where the work grows
History rows stay in the table, so a user with many corrections has more rows to scan on every recall. _rows also runs a subquery per row to find superseded_by. For large histories, an index on supersedes keeps that lookup fast; this build records no timing, so measure with your own data.
secure_delete costs extra writes on every delete. SQLite’s documentation says the default is off to reduce CPU cycles and disk I/O. The alternative it gives is running VACUUM after deletes, which rewrites the whole file. Measure both on your file sizes before choosing.
Key takeaways
- Store a source with every memory, and correct by superseding, so you can see what the assistant believed and why.
- Recall only current, unexpired memories; check expiry at read time, not only in a cleanup job.
- Forget means every version, an overwritten file, and a record without the text.
- Migrate schemas in one explicit transaction and test that a failed migration changes nothing.
Best practices
- Return the same error for another user’s memory and a missing one.
- Delete a whole correction chain when its newest version expires.
- Check deletion against the raw file, not only against queries.
- Keep the old code and a backup together when a schema migration ships.
References
- SQLite PRAGMA secure_delete: overwriting deleted content with zeros, the default, the performance trade-off,
VACUUMas the alternative, and traces in FTS shadow tables. Checked October 20, 2026. - Python
sqlite3: transaction control: which statements open a transaction implicitly in the default mode. Checked October 20, 2026 (Python 3.15 documentation). - Build a memory service with explicit user and task scope: the version 1 store and file format this build migrates.
The implementation, scenario, recorded output and tests belong to this build. The documentation describes SQLite and Python; it does not validate this store or any legal deletion requirement.