Memory & Persistence¶

The SynthOrg framework separates two distinct storage concerns:

• Agent memory -- what agents know, remember, and learn (working, episodic, semantic, procedural, social)
• Operational data -- tasks, cost records, messages, and audit logs generated during execution

Both are implemented behind pluggable protocol interfaces, making storage backends swappable via configuration without modifying application code.

Memory Architecture¶

+-------------------------------------------------+
|              Agent Memory System                |
+----------+----------+-----------+---------------+
| Working  | Episodic | Semantic  | Procedural    |
| Memory   | Memory   | Memory    | Memory        |
|          |          |           |               |
| Current  | Past     | Knowledge | Skills &      |
| task     | events & | & facts   | how-to        |
| context  | decisions| learned   |               |
+----------+----------+-----------+---------------+
|            Storage Backend                      |
|   Mem0 (initial, implemented) / Custom (future) |
|   Qdrant (embedded) + SQLite history             |
|     See Decision Log                             |
+-------------------------------------------------+

Each agent maintains its own memory store. The storage backend is selected via configuration and all access flows through the MemoryBackend protocol.

Memory Types¶

Memory Levels¶

Memory persistence is configurable per agent, from no persistence to fully persistent storage.

memory:
  level: "persistent"            # none | session | project | persistent (default: session)
  backend: "mem0"               # mem0 | custom | cognee | graphiti (future)
  storage:
    data_dir: "/data/memory"    # mounted Docker volume path
    vector_store: "qdrant"      # hardcoded to embedded qdrant in Mem0 backend
    history_store: "sqlite"     # hardcoded to sqlite in Mem0 backend
  options:
    retention_days: null         # null = forever
    max_memories_per_agent: 10000
    consolidation_interval: "daily"  # compress old memories
    shared_knowledge_base: true      # agents can access shared facts

Shared Organizational Memory¶

Beyond individual agent memory, the framework provides organizational memory -- company-wide knowledge that all agents can access: policies, conventions, architecture decision records (ADRs), coding standards, and operational procedures. This is not personal episodic memory ("what I did last Tuesday") but institutional knowledge ("the team always uses Litestar, not Flask").

Shared organizational memory is implemented behind an OrgMemoryBackend protocol, making the system highly modular and extensible. New backends can be added without modifying existing ones.

Backend 1: Hybrid Prompt + Retrieval (Default)¶

Critical rules (5--10 items, e.g., "no commits to main," "all PRs need 2 approvals") are injected into every agent's system prompt. Extended knowledge (ADRs, detailed procedures, style guides) is stored in a queryable store and retrieved on demand at task start.

org_memory:
  backend: "hybrid_prompt_retrieval"    # hybrid_prompt_retrieval, graph_rag, temporal_kg
  core_policies:                        # always in system prompt
    - "All code must have 80%+ test coverage"
    - "Use Litestar, not Flask"
    - "PRs require 2 approvals"
  extended_store:
    backend: "sqlite"                   # sqlite, postgresql
    max_retrieved_per_query: 5
  write_access:
    policies: ["human"]                 # only humans write core policies
    adrs: ["human", "senior", "lead", "c_suite"]
    procedures: ["human", "senior", "lead", "c_suite"]

Strengths: Simple to implement. Core rules are always present. Extended knowledge scales with the organization.

Limitations: Basic retrieval may miss relational connections between policies.

Research Directions¶

The following backends illustrate why OrgMemoryBackend is a protocol -- the architecture supports future upgrades without modifying existing code. These are research directions that may inform future work if organizational memory needs outgrow the Hybrid Prompt + Retrieval approach.

org_memory:
  backend: "graph_rag"
  graph:
    store: "sqlite"                     # graph stored in relational DB, or dedicated graph DB
    entity_extraction: "auto"           # auto-extract entities from ADRs and policies

org_memory:
  backend: "temporal_kg"
  temporal:
    track_changes: true
    history_retention_days: null        # null = forever

OrgMemoryBackend Protocol¶

All backends implement the OrgMemoryBackend protocol:

• query(OrgMemoryQuery) -> tuple[OrgFact, ...]
• write(OrgFactWriteRequest, *, author: OrgFactAuthor) -> NotBlankStr
• list_policies() -> tuple[OrgFact, ...]
• Lifecycle methods: connect, disconnect, health_check, is_connected, backend_name

The MVP ships with Backend 1 (Hybrid Prompt + Retrieval). The selected memory layer backend Mem0 (Decision Log) provides optional graph memory via Neo4j/FalkorDB, which could reduce implementation effort for the research direction backends.

Memory Backend Protocol¶

Agent memory is implemented behind a pluggable MemoryBackend protocol (Mem0 initial, custom stack future -- see Decision Log). Application code depends only on the protocol; the storage engine is an implementation detail swappable via config.

Enums¶

MemoryBackend Protocol¶

@runtime_checkable
class MemoryBackend(Protocol):
    """Lifecycle + CRUD for agent memory storage."""

    async def connect(self) -> None: ...
    async def disconnect(self) -> None: ...
    async def health_check(self) -> bool: ...

    @property
    def is_connected(self) -> bool: ...
    @property
    def backend_name(self) -> NotBlankStr: ...

    async def store(self, agent_id: NotBlankStr, request: MemoryStoreRequest) -> NotBlankStr:
        """Raises: MemoryConnectionError, MemoryStoreError."""
        ...
    async def retrieve(self, agent_id: NotBlankStr, query: MemoryQuery) -> tuple[MemoryEntry, ...]:
        """Raises: MemoryConnectionError, MemoryRetrievalError."""
        ...
    async def get(self, agent_id: NotBlankStr, memory_id: NotBlankStr) -> MemoryEntry | None:
        """Raises: MemoryConnectionError, MemoryRetrievalError."""
        ...
    async def delete(self, agent_id: NotBlankStr, memory_id: NotBlankStr) -> bool:
        """Raises: MemoryConnectionError, MemoryStoreError."""
        ...
    async def count(self, agent_id: NotBlankStr, *, category: MemoryCategory | None = None) -> int:
        """Raises: MemoryConnectionError, MemoryRetrievalError."""
        ...

MemoryCapabilities Protocol¶

Backends that implement MemoryCapabilities expose what features they support, enabling runtime capability checks before attempting operations.

@runtime_checkable
class MemoryCapabilities(Protocol):
    """Capability discovery for memory backends."""

    @property
    def supported_categories(self) -> frozenset[MemoryCategory]: ...
    @property
    def supports_graph(self) -> bool: ...
    @property
    def supports_temporal(self) -> bool: ...
    @property
    def supports_vector_search(self) -> bool: ...
    @property
    def supports_shared_access(self) -> bool: ...
    @property
    def max_memories_per_agent(self) -> int | None: ...

SharedKnowledgeStore Protocol¶

Backends that support cross-agent shared knowledge implement this protocol alongside MemoryBackend. Not all backends require cross-agent queries -- this keeps the base protocol clean.

@runtime_checkable
class SharedKnowledgeStore(Protocol):
    """Cross-agent shared knowledge operations."""

    async def publish(self, agent_id: NotBlankStr, request: MemoryStoreRequest) -> NotBlankStr:
        """Raises: MemoryConnectionError, MemoryStoreError."""
        ...
    async def search_shared(self, query: MemoryQuery, *, exclude_agent: NotBlankStr | None = None) -> tuple[MemoryEntry, ...]:
        """Raises: MemoryConnectionError, MemoryRetrievalError."""
        ...
    async def retract(self, agent_id: NotBlankStr, memory_id: NotBlankStr) -> bool:
        """Raises: MemoryConnectionError, MemoryStoreError."""
        ...

Error Hierarchy¶

All memory errors inherit from MemoryError so callers can catch the entire family with a single except clause.

Configuration¶

memory:
  backend: "mem0"
  level: "persistent"              # none, session, project, persistent (default: session)
  storage:
    data_dir: "/data/memory"
    vector_store: "qdrant"          # hardcoded to embedded qdrant in Mem0 backend
    history_store: "sqlite"         # hardcoded to sqlite in Mem0 backend
  options:
    retention_days: null            # null = forever
    max_memories_per_agent: 10000
    consolidation_interval: "daily"
    shared_knowledge_base: true

# Embedder config is passed programmatically via the factory:
#   create_memory_backend(config, embedder=Mem0EmbedderConfig(
#       provider="<embedding-provider>",
#       model="<embedding-model-id>",
#       dims=1536,
#   ))

Configuration is modeled by CompanyMemoryConfig (top-level), MemoryStorageConfig (storage paths/backends), and MemoryOptionsConfig (behaviour tuning). All are frozen Pydantic models. The create_memory_backend(config, *, embedder=...) factory returns an isolated MemoryBackend instance per company. The embedder kwarg is required for the Mem0 backend (must be a Mem0EmbedderConfig).

Embedding Model Selection¶

Embedding model quality directly determines memory retrieval accuracy. The LMEB benchmark (Zhao et al., March 2026) evaluates embedding models on long-horizon memory retrieval across four types that map directly to SynthOrg's MemoryCategory enum:

MTEB scores do not predict memory retrieval quality (Pearson: -0.115, Spearman: -0.130). Embedding model selection must be evaluated on LMEB, not MTEB. See Decision Log and the Embedding Evaluation reference page for the full analysis, model rankings, and deployment tier recommendations.

Key findings:

• Larger models do not always outperform smaller ones on memory retrieval
• Dialogue/social memory is the hardest retrieval category for all models
• Instruction sensitivity varies per model -- must be validated per deployment
• Three deployment tiers are recommended: full-resource (7-12B), mid-resource (1-4B), and CPU-only (< 1B)

class EmbeddingFineTuneConfig(BaseModel):
    model_config = ConfigDict(frozen=True, allow_inf_nan=False)

    enabled: bool = False
    checkpoint_path: NotBlankStr | None = None
    base_model: NotBlankStr | None = None
    training_data_dir: NotBlankStr | None = None

class FineTuningPipeline(Protocol):
    async def generate_training_data(self, source_dir: str) -> Path: ...
    async def mine_hard_negatives(self, training_data: Path) -> Path: ...
    async def fine_tune(self, training_data: Path, base_model: str) -> Path: ...

Consolidation and Retention¶

Memory consolidation, retention enforcement, and archival are configured via frozen Pydantic models in memory/consolidation/config.py:

Dual-Mode Archival¶

When ArchivalConfig.dual_mode.enabled is True, consolidation classifies content density before choosing an archival mode. This prevents catastrophic information loss from naively summarizing dense content (code, structured data, identifiers). Based on research: Memex (arXiv:2603.04257) and KV Cache Attention Matching (arXiv:2602.16284).

Classification is heuristic-based (DensityClassifier), using five weighted signals: code patterns, structured data markers, identifier density, numeric density, and line structure. No LLM is needed for classification -- only for abstractive summarization. Groups are classified by majority vote: if most entries in a category group are dense, the group uses extractive mode.

Deterministic restore: When entries are archived, the service builds an archival_index (mapping original_id → archival_id) on ConsolidationResult. Agents can use this index to call ArchivalStore.restore(agent_id, entry_id) directly by ID, bypassing semantic search.

Per-Agent Retention Overrides¶

Individual agents can override company-level retention rules via MemoryConfig.retention_overrides (per-category) and MemoryConfig.retention_days (agent-level default).

Resolution order per category:

1. Agent per-category rule
2. Company per-category rule
3. Agent global default
4. Company global default
5. Keep forever (no expiry)

Operational Data Persistence¶

Agent memory is handled by the MemoryBackend protocol (Mem0 initial, custom stack future -- see Decision Log). Operational data -- tasks, cost records, messages, audit logs -- is a separate concern managed by a pluggable PersistenceBackend protocol. Application code depends only on repository protocols; the storage engine is an implementation detail swappable via config.

Architecture¶

+------------------------------------------------------------------+
|                     Application Code                             |
|  engine/  budget/  communication/  security/                     |
|     |        |           |             |                         |
|     v        v           v             v                         |
|  +------+ +------+ +----------+ +----------+                    |
|  | Task | | Cost | | Message  | |  Audit   |  <-- Repository    |
|  | Repo | | Repo | |  Repo    | |  Repo    |      Protocols     |
|  +--+---+ +--+---+ +----+-----+ +----+-----+                    |
|     +--------+----------+------------+                           |
|                      |                                           |
|  +-------------------+-------------------------------------------+
|  |              PersistenceBackend (protocol)                    |
|  |  connect() . disconnect() . health_check() . migrate()       |
|  +-------------------+-------------------------------------------+
|                      |                                           |
|  +-------------------+-------------------------------------------+
|  |  SQLitePersistenceBackend (initial)                           |
|  |  PostgresPersistenceBackend (future)                          |
|  |  MariaDBPersistenceBackend (future)                           |
|  +---------------------------------------------------------------+
+------------------------------------------------------------------+

Protocol Design¶

@runtime_checkable
class PersistenceBackend(Protocol):
    """Lifecycle management for operational data storage."""

    async def connect(self) -> None: ...
    async def disconnect(self) -> None: ...
    async def health_check(self) -> bool: ...
    async def migrate(self) -> None: ...

    @property
    def is_connected(self) -> bool: ...
    @property
    def backend_name(self) -> NotBlankStr: ...

    @property
    def tasks(self) -> TaskRepository: ...
    @property
    def cost_records(self) -> CostRecordRepository: ...
    @property
    def messages(self) -> MessageRepository: ...
    # ... plus lifecycle_events, task_metrics, collaboration_metrics,
    #     parked_contexts, audit_entries, users, api_keys, checkpoints,
    #     heartbeats, agent_states, settings, artifacts, projects,
    #     custom_presets

Each entity type has its own repository protocol:

@runtime_checkable
class TaskRepository(Protocol):
    """CRUD + query interface for Task persistence."""

    async def save(self, task: Task) -> None: ...
    async def get(self, task_id: str) -> Task | None: ...
    async def list_tasks(self, *, status: TaskStatus | None = None, assigned_to: str | None = None, project: str | None = None) -> tuple[Task, ...]: ...
    async def delete(self, task_id: str) -> bool: ...

@runtime_checkable
class CostRecordRepository(Protocol):
    """CRUD + aggregation interface for CostRecord persistence."""

    async def save(self, record: CostRecord) -> None: ...
    async def query(self, *, agent_id: str | None = None, task_id: str | None = None) -> tuple[CostRecord, ...]: ...
    async def aggregate(self, *, agent_id: str | None = None) -> float: ...

@runtime_checkable
class MessageRepository(Protocol):
    """CRUD + query interface for Message persistence."""

    async def save(self, message: Message) -> None: ...
    async def get_history(self, channel: str, *, limit: int | None = None) -> tuple[Message, ...]: ...

Configuration¶

persistence:
  backend: "sqlite"                   # sqlite, postgresql, mariadb (future)
  sqlite:
    path: "/data/synthorg.db"       # database file path (mounted volume in Docker)
    wal_mode: true                    # WAL for concurrent read performance
    journal_size_limit: 67108864      # 64 MB WAL journal limit
  # postgresql:                       # future
  #   url: "postgresql://user:pass@host:5432/synthorg"
  #   pool_size: 10
  # mariadb:                          # future
  #   url: "mariadb://user:pass@host:3306/synthorg"
  #   pool_size: 10

Entities Persisted¶

Schema Strategy¶

• Schema is applied at startup via PersistenceBackend.migrate() which calls apply_schema()
• The canonical schema lives in src/synthorg/persistence/sqlite/schema.sql (single source of truth)
• All DDL uses IF NOT EXISTS guards, making application idempotent
• No sequential migrations exist yet -- when data stability is declared, adopt Atlas for declarative migrations (diff schema.sql against the live DB)

Key Principles¶

Application code never imports a concrete backend

Only repository protocols are used. This ensures complete decoupling from the storage engine.

Adding a new backend requires no changes to consumers

Implement PersistenceBackend + all repository protocols. Existing application code works unchanged.

Same entity models everywhere

Repositories accept and return the existing frozen Pydantic models (Task, CostRecord, Message). No ORM models or data transfer objects.

Async throughout

All repository methods are async, matching the framework's concurrency model.

Multi-Tenancy¶

Each company gets its own database. The PersistenceConfig embedded in a company's RootConfig specifies the backend type and connection details (e.g., a unique SQLite file path or PostgreSQL database URL). The create_backend(config) factory returns an isolated PersistenceBackend instance per company -- no shared state, no cross-company data leakage.

# One database per company -- configured in each company's YAML
company_a_backend = create_backend(company_a_config.persistence)
company_b_backend = create_backend(company_b_config.persistence)
# Each backend has independent lifecycle: connect -> migrate -> use -> disconnect

Procedural Memory Auto-Generation¶

When an agent fails a task, the engine's post-execution pipeline can automatically generate a procedural memory entry -- a structured "next time, do X when encountering Y" lesson learned. This follows the EvoSkill three-agent separation principle: the failed agent does not write its own lesson; a separate proposer LLM call analyses the failure.

Pipeline¶

1. Failure analysis payload (FailureAnalysisPayload): Built from RecoveryResult + ExecutionResult. Includes task metadata, sanitized error message, tool calls made, retry count, and turn count. Deliberately excludes raw conversation messages (privacy boundary).

2. Proposer LLM call (ProceduralMemoryProposer): A separate completion call with its own system prompt analyses the payload and returns a structured ProceduralMemoryProposal.

3. Three-tier progressive disclosure:

   • Discovery (~100 tokens): concise summary for retrieval ranking.
   • Activation (condition + action + rationale): when/what/why.
   • Execution (ordered steps): concrete steps for applying the knowledge.

4. Storage: The proposal is stored via MemoryBackend.store() as a MemoryCategory.PROCEDURAL entry with "non-inferable" tag for retrieval filtering.

5. SKILL.md materialization (optional): When ProceduralMemoryConfig.skill_md_directory is set, the proposal is also written as a portable SKILL.md file following the Agent Skills format for git-native versioning.

Configuration¶

ProceduralMemoryConfig (nested in CompanyMemoryConfig.procedural) controls:

• enabled: Toggle auto-generation on/off (default: True).
• model: Model identifier for the proposer LLM call (default: "example-small-001").
• temperature: Sampling temperature (default: 0.3).
• max_tokens: Token budget for the proposer response (default: 1500).
• min_confidence: Discard proposals below this threshold (default: 0.5).
• skill_md_directory: Optional path for SKILL.md file materialization.

Integration Point¶

AgentEngine._try_procedural_memory() runs after error recovery in _post_execution_pipeline. It is non-critical: failures are logged at WARNING and never block the execution result.

Memory Injection Strategies¶

Agent memory reaches agents through pluggable injection strategies behind the MemoryInjectionStrategy protocol. The strategy determines how memories are surfaced to the agent during execution.

MemoryInjectionStrategy Protocol¶

All strategies implement MemoryInjectionStrategy:

class MemoryInjectionStrategy(Protocol):

    async def prepare_messages(
        self, agent_id: NotBlankStr, query_text: NotBlankStr, token_budget: int
    ) -> tuple[ChatMessage, ...]: ...

    def get_tool_definitions(self) -> tuple[ToolDefinition, ...]: ...

    @property
    def strategy_name(self) -> str: ...

Strategy selection via config: memory.retrieval.strategy: context | tool_based | self_editing