memsmith 0.1.0

Enterprise memory lifecycle management for AI agents

Memsmith

Enterprise memory lifecycle management for AI agents. Agent proposes, library decides.

Memsmith is a write-layer for agent memory. It owns the lifecycle — propose, validate, store, monitor, expire, retract — and surfaces structured reads. Framework-agnostic, async-first, Pydantic-native.

---

Why Memsmith exists

Agents write to memory like a global mutable variable. No types, no rules, no audit. Current solutions only handle retrieval (vector search, RAG). The write path is an unguarded append() — and memory rots silently:

• Wrong inferences become permanent facts.
• Stale data accumulates with no expiration mechanism.
• Contradictions go undetected.
• No one knows who wrote what, when, or why.

Memsmith adds a governance layer to the write path. Every memory proposal runs through a validation pipeline, transitions through a documented state machine, and leaves an audit trail. Downstream consumers subscribe via hooks.

---

The lifecycle model

propose --> validate --> store --> monitor --> expire --> retract

When an agent proposes a memory:

1. Propose — agent submits content with metadata (session, confidence, source).
2. Validate — the validator pipeline runs. Each validator can pass, fail (non-critical = PendingConfirmation), or hard-fail (critical = Rejected).
3. Store — the memory transitions from PROPOSED to ACTIVE, PENDING_CONFIRMATION, or REJECTED. The state change is audit-logged and hooks fire.
4. Monitor — active memories can be queried, confirmed, or revised.
5. Expire — expire_batch() transitions past-due ACTIVE memories to EXPIRED.
6. Retract — a memory can be retracted at any point from ACTIVE or PENDING_CONFIRMATION.

Every state change is audit-logged. Every state change fires hooks for downstream sync.

---

Quick start

import asyncio
from pydantic import BaseModel
from memsmith import MemoryStore, SQLiteBackend, SchemaCheck, ConfidenceThreshold

# 1. Define a memory schema
class UserFact(BaseModel):
    attribute: str
    value: str

async def main():
    # 2. Create store with SQLite backend + validators
    store = await MemoryStore.init(
        storage=SQLiteBackent("demo.db"),
        validators=[
            SchemaCheck(UserFact, strict=False),
            ConfidenceThreshold(0.5),
        ],
    )

    # 3. Propose a memory
    result = await store.propose(
        content={"attribute": "language", "value": "Python"},
        session_id="session-1",
        confidence=0.9,
    )
    # result is Accepted(memory_id="...", memory=Memory(...))
    print(type(result).__name__, result.memory_id)

    # 4. Low confidence -> PendingConfirmation
    result = await store.propose(
        content={"attribute": "os", "value": "Linux"},
        session_id="session-1",
        confidence=0.3,
    )
    print(type(result).__name__, result.memory_id)
    # PendingConfirmation — confidence < 0.5 threshold

    # 5. Bad schema -> Rejected
    result = await store.propose(
        content={"bad": "data"},
        session_id="session-1",
        confidence=0.9,
    )
    print(type(result).__name__, result.reason)
    # Rejected — data doesn't match UserFact schema

    # 6. Confirm a pending memory
    if isinstance(result := ..., PendingConfirmation):
        memory = await store.confirm(result.memory_id)

    # 7. Retract
    await store.retract(memory.id, reason="Outdated")

    # 8. Read back
    memory = await store.get(memory.id)
    memories = await store.list(session_id="session-1")
    audit = await store.audit(memory_id=memory.id)

    # 9. Register a hook (e.g., sync to vector DB)
    async def sync_to_vector_db(
        memory_id: str, from_status, to_status,
    ):
        print(f"Syncing {memory_id}: {from_status} -> {to_status}")

    store.on_transition(sync_to_vector_db)

    await store.close()

asyncio.run(main())

---

API reference

MemoryStore

@classmethod
async def init(
    storage: StorageBackend,
    validators: list[ValidatorSpec] | None = None,
    hooks: list[TransitionHandler] | None = None,
) -> MemoryStore

Create a store. Accepts any StorageBackend implementation (SQLiteBackend ships, PostgresBackend protocol-ready).

async def close(self) -> None

Close the storage backend.

Write

async def propose(
    content: Any,
    session_id: str,
    confidence: float,
    thread_id: str | None = None,
    source: str | None = None,
    dedup: bool = True,
) -> Accepted | Rejected | PendingConfirmation

Propose a memory. Runs validators, transitions state, logs audit, fires hooks. Returns a discriminated result type.

async def propose_many(
    memories: list[ProposalInput],
    session_id: str,
    thread_id: str | None = None,
) -> list[Accepted | Rejected | PendingConfirmation]

Propose multiple memories in a single transaction.

async def confirm(self, memory_id: str) -> Memory

Transition PENDING_CONFIRMATION -> ACTIVE.

async def reject(self, memory_id: str, reason: str | None = None) -> Memory

Transition PENDING_CONFIRMATION or ACTIVE -> RETRACTED.

async def retract(self, memory_id: str, reason: str | None = None) -> Memory

Retract an ACTIVE or PENDING_CONFIRMATION memory.

async def revise(self, memory_id: str, new_content: Any, reason: str | None = None) -> Memory

Update content in-place. Status unchanged. Audit-logged.

async def update_confidence(self, memory_id: str, new_value: float) -> Memory

Increase confidence monotonically. Raises ConfidenceMonotonicityError if new value <= current.

async def expire_batch(self) -> int

Find all ACTIVE memories with expires_at <= now and transition them to EXPIRED. Returns count.

Read

async def get(self, memory_id: str) -> Memory | None

Fetch a single memory by ID.

async def list(
    status: MemoryStatus | None = None,
    session_id: str | None = None,
    field_filters: dict[str, Any] | None = None,
    limit: int = 50,
    offset: int = 0,
    order_by: str = "created_at",
) -> list[Memory]

List memories with optional filters. Allowed order_by values: created_at, updated_at, confidence, status, session_id.

async def audit(
    memory_id: str | None = None,
    limit: int = 50,
) -> list[AuditEntry]

Fetch audit log entries, optionally filtered by memory ID.

Hooks

def on_transition(self, callback: TransitionHandler) -> None

Register a post-commit callback. Signature: (memory_id: str, from_status: MemoryStatus, to_status: MemoryStatus) -> Awaitable[None]. Errors are logged, not propagated.

Return types

All result types are frozen dataclasses:

Type	Fields
Accepted(memory_id, memory)	Accepted proposal
Rejected(memory_id, memory, reason)	Rejected with reason
PendingConfirmation(memory_id, memory)	Needs human confirm
ProposalResult	Union of the above three

---

State machine

                    +--> ACTIVE --+--> RETRACTED
                    |             |
                    |             +--> EXPIRED
                    |
  PROPOSED ---------+--> PENDING_CONFIRMATION --+--> ACTIVE
                    |                           |
                    |                           +--> RETRACTED
                    |
                    +--> REJECTED (terminal)

  RETRACTED  (terminal)
  EXPIRED    (terminal)
  REJECTED   (terminal)

Transitions are enforced by _VALID_TRANSITIONS in the store. Invalid transitions raise InvalidTransitionError.

---

Storage model

memories table

Column	Type	Notes
id	TEXT	UUID hex
content	TEXT	JSON-serialized payload
session_id	TEXT	Agent session identifier
thread_id	TEXT	Optional thread grouping
source	TEXT	Optional provenance label
confidence	REAL	0.0 to 1.0
content_hash	TEXT	SHA-256 of serialized content
status	TEXT	Enum: proposed, pending_confirmation, active, rejected, retracted, expired
depends_on	TEXT	JSON list of memory IDs
created_at	TEXT	ISO 8601
updated_at	TEXT	ISO 8601
expires_at	TEXT	ISO 8601 or null
retracted_at	TEXT	ISO 8601 or null
retraction_reason	TEXT	Optional explanation

Indexes: (session_id), (status), (session_id, status), (created_at), (expires_at), (content_hash, session_id).

audit_log table

Column	Type	Notes
id	TEXT	UUID hex
memory_id	TEXT	FK to memories
action	TEXT	proposed, confirmed, retracted, expired, revised, confidence_updated, etc.
from_status	TEXT	Previous status (nullable)
to_status	TEXT	New status (nullable)
timestamp	TEXT	ISO 8601
metadata	TEXT	JSON blob

---

Built-in validators

SchemaCheck(model, strict=False)

Validates that memory.content conforms to a Pydantic model. Uses model.model_validate(). Set strict=True for strict type coercion.

from memsmith import SchemaCheck
from pydantic import BaseModel

class UserFact(BaseModel):
    attribute: str
    value: str

validator = SchemaCheck(UserFact)

ConfidenceThreshold(threshold)

Rejects memories below a minimum confidence level.

from memsmith import ConfidenceThreshold

validator = ConfidenceThreshold(0.7)  # Rejects confidence < 0.7

Custom validators

Any async callable matching the Validator protocol:

from memsmith import Validator, ValidationResult, Memory

class MyValidator:
    async def __call__(self, memory: Memory) -> ValidationResult:
        if "sensitive" in str(memory.content):
            return ValidationResult(passed=False, errors=["Sensitive content blocked"])
        return ValidationResult(passed=True)

Pass it as a ValidatorSpec:

from memsmith import ValidatorSpec

store = await MemoryStore.init(
    storage=...,
    validators=[
        ValidatorSpec(
            fn=MyValidator(),
            timeout=5.0,       # per-validator timeout (default 10s)
            critical=True,     # hard-fail on rejection (default False)
        ),
    ],
)

Non-critical validators that fail push the memory to PENDING_CONFIRMATION instead of REJECTED. Critical validators that fail immediately reject.

---

Architecture decisions

SQLite default, any DB via protocol. SQLite via aiosqlite ships as default — works for single-process agents, local dev, and embedded use. Need a different database? Implement the StorageBackend protocol (12 async methods: initialize, close, transaction, CRUD, audit, dedup, expire). No inheritance, no registration — structural typing. Any class with the right methods is a valid backend. Postgres via asyncpg, MySQL, Redis, whatever your stack uses. The protocol IS the contract.

Async-first. Memory operations are I/O-bound (SQLite WAL writes, network calls in validators, hook callbacks). The entire API is async. Synchronous convenience wrappers are deliberately absent — agents are already async.

No SQLAlchemy ORM. The SQLite backend uses raw SQL with aiosqlite. The schema is two tables, the queries are straightforward, and an ORM would add a dependency without reducing meaningful code. The migration system is a simple SQL file list.

Write-layer only. Memsmith does not do semantic search, vector embedding, or retrieval. That's the vector store's job. Memsmith owns the write path and structured reads. The on_transition hook system lets you sync memory state to any external index.

SAVEPOINT transactions. Nesting safe. Unlike BEGIN/COMMIT, SAVEPOINTs allow propose_many to run inside the store's own transaction scope without conflict when the caller is already in a transaction.

Frozen dataclasses for return types. Accepted, Rejected, PendingConfirmation are frozen dataclasses — lightweight, hashable, pattern-matchable. No Pydantic overhead on hot-path return objects.

Confidence is monotonically increasing. update_confidence only accepts values greater than the current. Confidence should never go down — if new evidence weakens a memory, retract and re-propose.

Dedup by content hash. propose hashes content and skips re-insertion if an identical memory exists in a non-terminal state for the same session. propose_many does per-item dedup, but cross-item dedup within the batch is not implemented (items earlier in the batch won't suppress duplicates later in the same batch).

---

What's in v0

• Memory lifecycle: propose, confirm, reject, retract, revise, expire
• State machine with 6 states and enforced transitions
• Validation pipeline with per-validator timeout, critical/non-critical modes
• Built-in validators: SchemaCheck (Pydantic), ConfidenceThreshold
• SQLite backend via aiosqlite with WAL mode
• Storage backend protocol for pluggable implementations
• Content dedup by hash per session
• Batch propose in a single transaction
• Audit log for every state change
• Hook system for downstream sync
• Confidence monotonicity enforcement
• Per-memory TTL via expires_at + expire_batch()
• Metrics facade (counter + histogram)
• 129 tests

---

What's deferred (post-v0)

Postgres backend. Not shipped, but the StorageBackend protocol supports any database. Implement the 12 methods and pass it to MemoryStore.init(). SQLite covers single-process agents; Postgres (or MySQL, or Redis) is added when you need concurrent writes or your stack requires it.

Semantic search. A vector store's job, not the write layer's. Memsmith emits content changes via hooks — pipe those into your vector DB of choice.

Source credibility engine. The source field exists on Memory. Automatic trust baselines per source (e.g., "user input > LLM inference > web scrape") are a policy layer on top. v0 keeps it simple: source is a string label.

TTL policy engine. expires_at works per-memory. Declarative policies ("all chat memories expire in 24h") are a configuration layer deferred to v1.

Dependency cascade rules. depends_on exists on Memory. Automatic cascading (retracting a parent retracts all children) is deferred. Hooks fire on every transition — you can implement cascade at the app layer today.

Confidence reinforcement. update_confidence works. Automatic boost when corroborating memories arrive is app-layer logic.

File-per-memory backend. The StorageBackend protocol is backend-agnostic. A filesystem backend (one JSON file per memory) isn't needed yet.

Batch dedup. Individual propose dedup works. Cross-item dedup within a propose_many batch is not implemented — items earlier in the batch won't suppress later duplicates.

No caching, no CLI, no web server, no retry logic, no distributed locking, no full-text search, no plugin system beyond protocols.

---

Installation

pip install memsmith

With SQLite support (default):

pip install memsmith[sqlite]

With Postgres support:

pip install memsmith[postgres]

All extras:

pip install memsmith[all]

Requires Python 3.11+.

---

Testing

pip install -e ".[sqlite,postgres]"
pytest

Tests use pytest-asyncio (asyncio_mode = auto). The SQLite backend runs in :memory: mode for tests.

---

License

MIT