Memory — API reference

CLI

main

main()

Storage

connection

connection(db_path: Path | None = None)

Context manager that opens and closes a memory database connection.

ensure_schema

ensure_schema(conn: Connection) -> None

Apply any pending migrations to the database.

Safe to call on every connection open — checks a single integer and returns immediately if the schema is current. Gracefully skips vec0 migrations if the sqlite-vec extension is not loaded.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection

get_connection

get_connection(db_path: Path | None = None) -> Connection

Open a memory database connection with sqlite-vec loaded.

The connection has the sqlite-vec extension loaded (for vec0 support), WAL journal mode enabled (for concurrent reads), and the current schema applied via migrations.

PARAMETER	DESCRIPTION
db_path	Override the database path (defaults to {data_dir}/memory.db). TYPE: Path | None DEFAULT: None

RETURNS	DESCRIPTION
Connection	A sqlite3.Connection ready for use. The caller must close it.

get_data_dir

get_data_dir() -> Path

Return the mait-code data directory, creating it if needed.

get_db_path

get_db_path() -> Path

Return the memory database path.

Embeddings

BedrockProvider

BedrockProvider()

Bases: EmbeddingProvider

AWS Bedrock embedding provider.

EMBEDDING_DIM module-attribute

EMBEDDING_DIM: int = _get_embedding_dim()

EMBEDDING_MODEL module-attribute

EMBEDDING_MODEL: str = _get_embedding_model()

EmbeddingProvider

Bases: ABC

Internal interface for embedding providers.

dimension abstractmethod property

dimension: int

Return the embedding dimension for this provider/model.

model_name abstractmethod property

model_name: str

Return the human-readable model identifier.

embed abstractmethod

embed(texts: list[str]) -> list[list[float]]

Embed a batch of texts and return one float vector per input.

LocalProvider

LocalProvider()

Bases: EmbeddingProvider

fastembed/HuggingFace local embeddings.

check_dimension_match

check_dimension_match(
    conn: Connection,
) -> tuple[bool, int | None, int]

Check whether the vec table dimension matches the configured provider.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection

RETURNS	DESCRIPTION
bool	A tuple (matches, table_dim, expected_dim). table_dim is
int | None	None if the vec table doesn't exist.

embed_text

embed_text(
    text: str, *, prefix: str = "search_document"
) -> list[float] | None

Embed a single text string.

PARAMETER	DESCRIPTION
text	The text to embed. TYPE: str
prefix	Task prefix for nomic-embed. Use "search_document" when storing/indexing, "search_query" when searching. Ignored for Bedrock providers. TYPE: str DEFAULT: 'search_document'

RETURNS	DESCRIPTION
list[float] | None	The embedding vector, or None if embeddings are unavailable.

embed_texts

embed_texts(
    texts: list[str], *, prefix: str = "search_document"
) -> list[list[float]] | None

Embed a batch of texts.

PARAMETER	DESCRIPTION
texts	The texts to embed. TYPE: list[str]
prefix	Task prefix for nomic-embed. Ignored for Bedrock providers. TYPE: str DEFAULT: 'search_document'

RETURNS	DESCRIPTION
list[list[float]] | None	The list of embedding vectors, or None if unavailable.

get_provider

get_provider() -> EmbeddingProvider | None

Return the lazily-initialised embedding provider, or None.

On first call, instantiates the provider configured by MAIT_CODE_EMBEDDING_PROVIDER. If instantiation fails, returns None on this and all subsequent calls.

RETURNS	DESCRIPTION
EmbeddingProvider | None	The provider instance, or None if initialisation failed.

is_available

is_available() -> bool

Return True if the embedding provider can be loaded.

serialize_f32

serialize_f32(vec: list[float]) -> bytes

Serialise a float vector to raw bytes for sqlite-vec.

Search

delete_entry

delete_entry(conn: Connection, entry_id: int) -> bool

Delete a memory entry by ID.

The memory_vec cleanup is handled by the database trigger (memory_entries_vec_ad), so no explicit vec deletion is needed.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
entry_id	Primary-key id of the entry to delete. TYPE: int

RETURNS	DESCRIPTION
bool	True if a row was deleted, False if no entry had that id.

hybrid_search

hybrid_search(
    conn: Connection,
    query: str,
    limit: int = 20,
    entry_type: str | None = None,
    *,
    project: str | None = None,
    branch: str | None = None,
    include_superseded: bool = False,
) -> list[dict]

Run a combined FTS5 + vector search and return merged results.

Runs both search methods, merges by entry ID, and assigns a relevance score suitable for composite_score():

• Entries found by both: use vector similarity as relevance.
• Vector-only entries: keep their cosine similarity as relevance.
• FTS-only entries: default relevance 0.3 (no vector to score with).

Falls back to FTS-only if no embeddings are available.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
query	Query text. TYPE: str
limit	Maximum number of results from each underlying search. TYPE: int DEFAULT: 20
entry_type	Optional entry-type filter. TYPE: str | None DEFAULT: None
project	Project context for scope filtering. TYPE: str | None DEFAULT: None
branch	Branch context for scope filtering. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
list[dict]	A list of dicts with the standard fields plus a "relevance" key.

list_entries

list_entries(
    conn: Connection,
    limit: int = 20,
    entry_type: str | None = None,
    since: str | None = None,
    *,
    project: str | None = None,
    branch: str | None = None,
    scope: str | None = None,
    include_superseded: bool = False,
) -> list[dict]

List recent memory entries, optionally filtered by type, time, and scope.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
limit	Maximum number of results. TYPE: int DEFAULT: 20
entry_type	Optional entry-type filter. TYPE: str | None DEFAULT: None
since	Human-readable period like "24h", "7d", "1w". TYPE: str | None DEFAULT: None
project	Filter to this project context (includes global). TYPE: str | None DEFAULT: None
branch	Filter to this branch context. TYPE: str | None DEFAULT: None
scope	Explicit scope filter ("global", "project", "branch"); overrides project-based filtering when set. TYPE: str | None DEFAULT: None
include_superseded	Include entries that have been superseded (hidden by default). TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
list[dict]	A list of dicts with the standard memory entry fields, ordered by
list[dict]	created_at descending.

list_projects

list_projects(conn: Connection) -> list[str]

List the distinct projects current memory entries span, alphabetically.

Feeds filter UIs (the memory browser's project Select); global entries carry no project and so don't contribute a name, and superseded entries don't keep a project alive on their own.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection

RETURNS	DESCRIPTION
list[str]	Sorted distinct project names (possibly empty).

search_entries

search_entries(
    conn: Connection,
    query: str,
    limit: int = 20,
    entry_type: str | None = None,
    *,
    project: str | None = None,
    branch: str | None = None,
    include_superseded: bool = False,
) -> list[dict]

Search memory entries using FTS5 BM25 ranking.

Falls back to LIKE if FTS5 is not available. When project is provided, filters to global plus matching project/branch entries.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
query	FTS5 query string (or substring for the fallback path). TYPE: str
limit	Maximum number of results. TYPE: int DEFAULT: 20
entry_type	Optional entry-type filter. TYPE: str | None DEFAULT: None
project	Project context for scope filtering. TYPE: str | None DEFAULT: None
branch	Branch context for scope filtering. TYPE: str | None DEFAULT: None
include_superseded	Include entries that have been superseded (hidden by default). TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
list[dict]	A list of dicts with the standard memory entry fields.

vector_search_entries

vector_search_entries(
    conn: Connection,
    query: str,
    limit: int = 20,
    entry_type: str | None = None,
    *,
    project: str | None = None,
    branch: str | None = None,
    include_superseded: bool = False,
) -> list[dict]

Search memory entries using vector similarity via sqlite-vec.

Over-fetches and post-filters by scope, since sqlite-vec doesn't support arbitrary WHERE clauses in k-NN queries.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
query	Query text to embed and search. TYPE: str
limit	Maximum number of results. TYPE: int DEFAULT: 20
entry_type	Optional entry-type filter. TYPE: str | None DEFAULT: None
project	Project context for scope filtering. TYPE: str | None DEFAULT: None
branch	Branch context for scope filtering. TYPE: str | None DEFAULT: None
include_superseded	Include entries that have been superseded (hidden by default). TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
list[dict]	A list of dicts with the standard fields plus a "similarity"
list[dict]	key in [0.0, 1.0]. Empty list if embeddings are unavailable.

Scoring

composite_score

composite_score(
    created_at: str | datetime,
    importance: int,
    relevance: float = 0.5,
    *,
    memory_class: str | None = None,
    entry_scope: str | None = None,
    entry_project: str | None = None,
    entry_branch: str | None = None,
    query_project: str | None = None,
    query_branch: str | None = None,
    w_recency: float = W_RECENCY,
    w_importance: float = W_IMPORTANCE,
    w_relevance: float = W_RELEVANCE,
    now: datetime | None = None,
) -> float

Compute the composite score for a memory entry.

PARAMETER	DESCRIPTION
created_at	When the entry was created (ISO string or datetime). TYPE: str | datetime
importance	Importance level 1-10. TYPE: int
relevance	Relevance score 0.0-1.0 (from search or default 0.5). TYPE: float DEFAULT: 0.5
memory_class	"episodic", "semantic", or "procedural" (controls decay rate). TYPE: str | None DEFAULT: None
entry_scope	Scope of the entry ("global", "project", "branch"). TYPE: str | None DEFAULT: None
entry_project	Project of the entry. TYPE: str | None DEFAULT: None
entry_branch	Branch of the entry. TYPE: str | None DEFAULT: None
query_project	Project context for the query. TYPE: str | None DEFAULT: None
query_branch	Branch context for the query. TYPE: str | None DEFAULT: None
w_recency	Weight for the recency component. TYPE: float DEFAULT: W_RECENCY
w_importance	Weight for the importance component. TYPE: float DEFAULT: W_IMPORTANCE
w_relevance	Weight for the relevance component. TYPE: float DEFAULT: W_RELEVANCE
now	Override current time (for testing). TYPE: datetime | None DEFAULT: None

RETURNS	DESCRIPTION
float	Composite score, roughly in [0.0, 1.0].

importance_score

importance_score(importance: int) -> float

Normalise an importance value (1-10) to [0.0, 1.0].

rank_results

rank_results(
    results: list[dict],
    *,
    limit: int,
    query_project: str | None = None,
    query_branch: str | None = None,
) -> list[tuple[float, dict]]

Score search results with :func:composite_score and keep the best.

The shared ranking step behind mc-tool-memory search and :func:mait_code.remote.search_memories, so both order results the same way. A result without a relevance key scores as 0.5.

PARAMETER	DESCRIPTION
results	Entries from one of the search functions. TYPE: list[dict]
limit	Maximum number of results to keep. TYPE: int
query_project	Project context the query was made in. TYPE: str | None DEFAULT: None
query_branch	Branch context the query was made in. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
list[tuple[float, dict]]	(score, entry) pairs, best first, at most limit long.

recency_score

recency_score(
    created_at: str | datetime,
    now: datetime | None = None,
    *,
    memory_class: str | None = None,
) -> float

Compute the recency score using exponential decay.

Half-life depends on memory_class:

• episodic: 3 days (events decay fast).
• semantic: 90 days (facts persist).
• procedural: 180 days (workflows go stale when superseded, not with time).
• None/unknown: 7 days (fallback).

PARAMETER	DESCRIPTION
created_at	When the entry was created (ISO string or datetime). TYPE: str | datetime
now	Override current time (for testing); defaults to UTC now. TYPE: datetime | None DEFAULT: None
memory_class	Memory class controlling the decay rate. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
float	A score in [0.0, 1.0]; returns 0.0 if created_at cannot
float	be parsed.

scope_boost

scope_boost(
    entry_scope: str,
    entry_project: str | None,
    entry_branch: str | None,
    *,
    query_project: str | None = None,
    query_branch: str | None = None,
) -> float

Compute a multiplicative boost based on scope match.

PARAMETER	DESCRIPTION
entry_scope	The entry's scope ("global", "project", "branch"). TYPE: str
entry_project	The entry's project, or None. TYPE: str | None
entry_branch	The entry's branch, or None. TYPE: str | None
query_project	Project context for the query, or None. TYPE: str | None DEFAULT: None
query_branch	Branch context for the query, or None. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
float	1.0 for a branch match, 0.85 for a project match,
float	SCOPE_BOOST_GLOBAL (default 0.7) for global entries,
float	SCOPE_BOOST_CROSS_PROJECT (default 0.3) across projects, and
float	1.0 when no query context is provided (backward compat).

Entities

find_entity_by_name

find_entity_by_name(
    conn: Connection, name: str
) -> dict | None

Look up an entity by name (case-insensitive).

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
name	Entity name to find. TYPE: str

RETURNS	DESCRIPTION
dict | None	A dict with entity fields, or None if no match.

get_ego_graph

get_ego_graph(conn: Connection, name: str) -> dict | None

The 1-hop neighbourhood of an entity: its node, neighbours, and edges.

The graph explorer's centre query. Both orderings are deterministic (neighbours by mention count then name; relationships with the centre's own edges first, then by source/type/target) so renders and snapshot tests are stable for a given database state.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
name	Centre entity name (case-insensitive). TYPE: str

RETURNS	DESCRIPTION
dict | None	A dict with centre (the entity's fields), entities (centre
dict | None	first, then neighbours), and relationships (every edge incident
dict | None	to the centre, with source/target names) — or None if no entity
dict | None	matches name.

get_entity_relationships

get_entity_relationships(
    conn: Connection, entity_id: int
) -> list[dict]

Return all relationships involving an entity (as source or target).

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
entity_id	Entity id to look up. TYPE: int

RETURNS	DESCRIPTION
list[dict]	A list of relationship dicts, each including the source and target
list[dict]	entity names.

list_graph_entities

list_graph_entities(
    conn: Connection,
    query: str = "",
    *,
    min_mentions: int = 1,
    require_relationship: bool = False,
    limit: int | None = None,
) -> list[dict]

List entities with their relationship degree, for graph surfaces.

The graph explorer's entity list: every entity matching query, each carrying a degree (the number of relationships it participates in) so callers can filter or weight by connectedness. The default filters are permissive; pass min_mentions=2 and require_relationship=True for the noise-hiding defaults the explorer uses (84% of entities are single-mention tail).

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
query	Substring to match against entity names ("" matches all). TYPE: str DEFAULT: ''
min_mentions	Keep entities mentioned at least this many times. TYPE: int DEFAULT: 1
require_relationship	Drop entities with no relationships (degree 0). TYPE: bool DEFAULT: False
limit	Maximum number of results, or None for all. TYPE: int | None DEFAULT: None

RETURNS	DESCRIPTION
list[dict]	A list of entity dicts (including degree), ordered by
list[dict]	mention_count descending, then last_seen descending, then id
list[dict]	— a deterministic order for rendering and tests.

merge_entities

merge_entities(
    conn: Connection, source_name: str, target_name: str
) -> dict

Merge one entity into another, repointing its relationships.

Aliases accumulate in the graph (e.g. "User" alongside the user's real name) and split what should be one node. Merging folds the source entity into the target: relationships are repointed (deduplicating against the target's existing edges on the (source, target, type) unique index), mention counts are summed, the seen window widens to span both, the target's type is upgraded from "unknown" if the source's is more specific, and the source entity is deleted. Edges directly between the two (which would become self-loops) are dropped.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
source_name	Name of the entity to fold in (case-insensitive). TYPE: str
target_name	Name of the surviving entity (case-insensitive). TYPE: str

RETURNS	DESCRIPTION
dict	A summary dict: target (the surviving entity's fields, post-merge),
dict	relationships_repointed, relationships_deduplicated, and
dict	self_loops_dropped.

RAISES	DESCRIPTION
ValueError	If either entity does not exist, or both names resolve to the same entity.

search_entities

search_entities(
    conn: Connection, query: str, limit: int = 20
) -> list[dict]

Search entities by name using a LIKE substring match.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
query	Substring to match against entity names. TYPE: str
limit	Maximum number of results to return. TYPE: int DEFAULT: 20

RETURNS	DESCRIPTION
list[dict]	A list of entity dicts ordered by mention_count descending,
list[dict]	then last_seen descending.

upsert_entity

upsert_entity(
    conn: Connection, name: str, entity_type: str
) -> int

Insert or update an entity by name and return its id.

On conflict, increments mention_count, refreshes last_seen, and upgrades entity_type from "unknown" if a more specific type is provided.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
name	Entity name (case-insensitive unique key). TYPE: str
entity_type	Entity type (e.g. "person", "project"). TYPE: str

RETURNS	DESCRIPTION
int	The entity's primary-key id.

upsert_relationship

upsert_relationship(
    conn: Connection,
    source_entity_id: int,
    target_entity_id: int,
    relationship_type: str,
    context: str,
) -> int

Insert or update a relationship and return its id.

On conflict (same source/target/type), refreshes last_seen and updates context when the new value differs from the stored one.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
source_entity_id	Source entity id. TYPE: int
target_entity_id	Target entity id. TYPE: int
relationship_type	Relationship label (e.g. "uses"). TYPE: str
context	Free-text context describing the relationship. TYPE: str

RETURNS	DESCRIPTION
int	The relationship's primary-key id.

Writer

find_duplicate

find_duplicate(
    conn: Connection,
    content: str,
    entry_type: str,
    *,
    project: str | None = None,
) -> int | None

Check for near-duplicate content in the database.

Thin wrapper over :func:_assess_candidates that discards the contradiction-band candidates and returns only the merge target.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
content	The candidate memory content. TYPE: str
entry_type	Entry type used to scope the candidate search. TYPE: str
project	Project identifier; None matches global-only entries. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
int | None	The id of the duplicate if found, otherwise None.

merge_memories

merge_memories(
    conn: Connection,
    old_ids: list[int],
    content: str,
    *,
    importance: int | None = None,
) -> dict

Fold several memory entries into one consolidated entry.

Inserts content as a single fresh entry, then supersedes every existing row in old_ids by pointing each superseded_by at the new id. This is the N→1 counterpart to :func:supersede_memory: use it when reflection finds two or more overlapping facts that should read as one.

The new entry inherits type, class, and scope (project/branch) from the first existing row in old_ids. Its importance defaults to the highest importance among the merged rows — a merge promotes, it does not dilute — unless overridden. Ids that don't exist are skipped and reported back in not_found; the merge proceeds on whatever remains.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
old_ids	Ids of the entries being merged. Order matters only in that the first existing id donates type/class/scope. TYPE: list[int]
content	The consolidated content for the new entry. TYPE: str
importance	Optional importance for the new entry (1-10, clamped). Defaults to the maximum importance among the merged rows. TYPE: int | None DEFAULT: None

RETURNS	DESCRIPTION
dict	On success, a dict with action="merged", merged_ids (the ids
dict	actually superseded), not_found (ids that didn't exist), id
dict	(the new entry), content, scope, project, branch. If
dict	none of old_ids exist, {"action": "not_found", "old_ids": ...}.

retire_memory

retire_memory(conn: Connection, entry_id: int) -> dict

Retire a memory entry — drop it with no replacement.

Stamps superseded_at while leaving superseded_by NULL. That pair is the retired marker: the row is hidden from default surfacing (see :data:~mait_code.tools.memory.db.LIVE_ENTRY_SQL) but kept for auditability. Use this when a fact is stale or contradicted and has no successor; reach for :func:supersede_memory when there is a replacement.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
entry_id	Id of the entry to retire. TYPE: int

RETURNS	DESCRIPTION
dict	{"action": "retired", "id": entry_id} on success. If the row does
dict	not exist, {"action": "not_found", "id": entry_id}. If it is already
dict	superseded or retired, ``{"action": "already_superseded" |
dict	"already_retired", "id": entry_id}`` and nothing is written.

store_memory

store_memory(
    conn: Connection,
    content: str,
    entry_type: str = "fact",
    importance: int = 5,
    *,
    scope: str = "global",
    project: str | None = None,
    branch: str | None = None,
) -> dict

Store a memory entry, deduplicating near-identical content.

On duplicate, refreshes the timestamp and keeps the maximum importance. On a new entry, inserts with memory_class derived from entry_type.

PARAMETER	DESCRIPTION
conn	Database connection (with schema applied). TYPE: Connection
content	The memory content to store. TYPE: str
entry_type	One of fact, preference, decision, event, insight, task, relationship. decision is an extracted architectural decision; insight is reserved for reflection output. Invalid values fall back to fact. TYPE: str DEFAULT: 'fact'
importance	Importance level 1-10 (clamped). TYPE: int DEFAULT: 5
scope	Memory scope — "global", "project", or "branch". TYPE: str DEFAULT: 'global'
project	Project identifier (e.g. repo basename). TYPE: str | None DEFAULT: None
branch	Branch name (for branch-scoped memories). TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
dict	A dict with keys action ("created" or "deduplicated"),
dict	id, content, scope, project, branch, and
dict	potential_conflicts — a list of {"id", "content", "similarity"}
dict	for existing entries in the contradiction band that this write may
dict	contradict. The write is never blocked; the conflicts are surfaced so
dict	the companion can suggest superseding one of them. Empty on a
dict	deduplicated write.

supersede_memory

supersede_memory(
    conn: Connection,
    old_id: int,
    content: str,
    *,
    importance: int | None = None,
) -> dict

Supersede an existing memory entry with an evolved version.

Inserts content as a fresh entry that inherits the old entry's type, class, and scope (project/branch), then marks the old entry superseded — pointing superseded_by at the new id and stamping superseded_at. The old row is kept for auditability but hidden from default surfacing. This is the explicit, manually-driven counterpart to dedup: it is called when a fact has genuinely changed rather than merely been repeated.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
old_id	Id of the entry being superseded. TYPE: int
content	The new, current content. TYPE: str
importance	Optional importance for the new entry (1-10, clamped). Defaults to the superseded entry's importance. TYPE: int | None DEFAULT: None

RETURNS	DESCRIPTION
dict	On success, a dict with action="superseded", old_id, id
dict	(the new entry), content, scope, project, branch. If
dict	old_id does not exist, {"action": "not_found", "old_id": old_id}.

Stats

MemoryStats dataclass

MemoryStats(
    total: int,
    by_type: list[tuple[str, int]],
    by_class: list[tuple[str, int]],
    by_scope: list[tuple[str, int]],
    by_project: list[tuple[str, int]],
    superseded: int,
    retired: int,
    embedded: int,
    provider: str,
    model: str,
    dim: int,
    unreflected: int,
    last_reflected_at: datetime | None,
)

A snapshot of the memory store's shape and embedding coverage.

unembedded property

unembedded: int

Entries with no vector in memory_vec.

embedded_pct property

embedded_pct: int

Embedding coverage as a whole-number percentage (0 when empty).

collect_stats

collect_stats(conn: Connection) -> MemoryStats

Collect store statistics from an open memory connection.

All counts come from SQL; provider/model/dimension come from configuration. unreflected and last_reflected_at describe the global reflection watermark (observation backlog and freshness).

Reflection

count_unreflected

count_unreflected(
    conn: Connection, *, project: str | None = None
) -> int

Return the number of entries above the reflection watermark.

Counts the non-insight entries that a future :func:reflect run would consider — the observation backlog. With no watermark (no reflection has ever run for the scope), every non-insight entry counts.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
project	Project identifier to scope the count; None for global. TYPE: str | None DEFAULT: None

get_last_reflected_at

get_last_reflected_at(
    conn: Connection, *, project: str | None = None
) -> datetime | None

Return when reflection last ran for a project (or global scope).

Reads last_reflected_at from the watermark table — the authoritative record of the most recent :func:reflect run, regardless of whether it produced insights.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
project	Project identifier, or None for the global watermark. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
datetime | None	The parsed timestamp, or None if reflection has never run.

reflect

Reflection system — synthesise recent observations into durable insights.

Reads recent memory entries and observation logs, calls Claude to identify patterns and themes, stores insights back to memory.db, and proposes updates to MEMORY.md.

get_watermark

get_watermark(
    conn: Connection, *, project: str | None = None
) -> int | None

Return the last reflected entry ID for a project (or global scope).

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
project	Project identifier, or None for the global watermark. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
int | None	The watermark, or None if no reflection has ever been done for
int | None	this scope.

update_watermark

update_watermark(
    conn: Connection,
    last_id: int,
    *,
    project: str | None = None,
) -> None

Set the watermark for a project (or global) to last_id.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
last_id	The new high-water-mark entry ID. TYPE: int
project	Project identifier, or None for the global watermark. TYPE: str | None DEFAULT: None

count_unreflected

count_unreflected(
    conn: Connection, *, project: str | None = None
) -> int

Return the number of entries above the reflection watermark.

Counts the non-insight entries that a future :func:reflect run would consider — the observation backlog. With no watermark (no reflection has ever run for the scope), every non-insight entry counts.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
project	Project identifier to scope the count; None for global. TYPE: str | None DEFAULT: None

get_last_reflected_at

get_last_reflected_at(
    conn: Connection, *, project: str | None = None
) -> datetime | None

Return when reflection last ran for a project (or global scope).

Reads last_reflected_at from the watermark table — the authoritative record of the most recent :func:reflect run, regardless of whether it produced insights.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
project	Project identifier, or None for the global watermark. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
datetime | None	The parsed timestamp, or None if reflection has never run.

check_novelty_gate_v2

check_novelty_gate_v2(
    conn: Connection,
    min_new: int = 3,
    *,
    project: str | None = None,
) -> bool

Return True if there are enough unreflected entries to justify reflection.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
min_new	Minimum number of new non-insight entries required. TYPE: int DEFAULT: 3
project	Project identifier to scope the check; None for global. TYPE: str | None DEFAULT: None

get_unreflected_entries

get_unreflected_entries(
    conn: Connection,
    batch_size: int = 50,
    days: int | None = None,
    exclude_types: tuple[str, ...] = ("insight",),
    *,
    project: str | None = None,
    watermark: int | None = None,
) -> list[tuple]

Return entries that haven't been reflected on yet.

Uses watermark (last reflected ID) for idempotency. If no watermark exists and days is provided, uses days as a bootstrap window.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
batch_size	Maximum number of entries to return. TYPE: int DEFAULT: 50
days	Bootstrap window in days when no watermark exists. TYPE: int | None DEFAULT: None
exclude_types	Entry types to exclude (e.g. "insight"). TYPE: tuple[str, ...] DEFAULT: ('insight',)
project	Project identifier; when set, includes global plus project-scoped entries. TYPE: str | None DEFAULT: None
watermark	High-water-mark entry ID; only entries above this are returned. TYPE: int | None DEFAULT: None

RETURNS	DESCRIPTION
list[tuple]	Tuples of (id, content, entry_type, importance, created_at).

get_last_reflection_date

get_last_reflection_date(
    conn: Connection, *, project: str | None = None
) -> datetime | None

Return the timestamp of the most recent insight entry.

Deprecated: replaced by get_watermark(); kept for backward compat.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
project	Project identifier, or None for any scope. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
datetime | None	The parsed timestamp, or None if there is no insight entry.

count_entries_since

count_entries_since(
    conn: Connection,
    since: datetime,
    exclude_types: tuple[str, ...] = ("insight",),
    *,
    project: str | None = None,
) -> int

Count non-insight memory entries added since a given timestamp.

Deprecated: replaced by check_novelty_gate_v2(); kept for backward compat.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
since	Lower bound timestamp. TYPE: datetime
exclude_types	Entry types to exclude from the count. TYPE: tuple[str, ...] DEFAULT: ('insight',)
project	Project identifier, or None to count all scopes. TYPE: str | None DEFAULT: None

check_novelty_gate

check_novelty_gate(
    conn: Connection,
    min_new: int = 3,
    *,
    project: str | None = None,
) -> bool

Return True if there are enough new observations to justify reflection.

Deprecated: replaced by check_novelty_gate_v2(); kept for backward compat.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
min_new	Minimum number of new non-insight entries required. TYPE: int DEFAULT: 3
project	Project identifier, or None for any scope. TYPE: str | None DEFAULT: None

get_recent_entries

get_recent_entries(
    conn: Connection,
    days: int = 7,
    limit: int = 200,
    exclude_types: tuple[str, ...] = ("insight",),
    *,
    project: str | None = None,
) -> list[tuple]

Return memory entries from the last days days.

Deprecated: replaced by get_unreflected_entries(); kept for backward compat.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
days	Lookback window in days. TYPE: int DEFAULT: 7
limit	Maximum number of entries to return. TYPE: int DEFAULT: 200
exclude_types	Entry types to exclude. TYPE: tuple[str, ...] DEFAULT: ('insight',)
project	Project identifier; when set, includes global plus project-scoped entries. TYPE: str | None DEFAULT: None

read_observation_logs

read_observation_logs(days: int = 7) -> str

Read JSONL observation files from the last days days.

Parses extraction dicts and formats them into readable text for the reflection prompt.

PARAMETER	DESCRIPTION
days	Lookback window in days. TYPE: int DEFAULT: 7

RETURNS	DESCRIPTION
str	A newline-joined string of formatted extraction items.

format_entries_text

format_entries_text(entries: list[tuple]) -> str

Format DB entries as text for the reflection prompt.

Accepts both 4-tuples (legacy: (content, type, importance, created_at)) and 5-tuples ((id, content, type, importance, created_at)).

PARAMETER	DESCRIPTION
entries	Rows fetched from memory_entries. TYPE: list[tuple]

RETURNS	DESCRIPTION
str	A newline-joined string of formatted entries.

parse_reflection_response

parse_reflection_response(response: str) -> dict

Parse the LLM response into insights and structured memory operations.

Recognises INSIGHT: lines plus the consolidation op markers in :data:_OP_MARKERS. Each op is a dict with keys op ("add" / "rewrite" / "merge" / "retire"), old (existing MEMORY.md text, or None), new (new text, or None for retire), and entry_ids (backing db entries to consolidate, possibly empty). Unrecognised or malformed lines are ignored, not fatal.

PARAMETER	DESCRIPTION
response	The raw LLM response text. TYPE: str

RETURNS	DESCRIPTION
dict	A dict {"insights": [...], "ops": [...]}.

store_insights

store_insights(
    conn: Connection,
    insights: list[str],
    *,
    project: str | None = None,
) -> int

Store insights in the memory database with fixed importance=6.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
insights	Insight strings to store. TYPE: list[str]
project	Project identifier; when set, insights are project-scoped, otherwise they are stored globally. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
int	The number of insights stored.

read_memory_md

read_memory_md() -> str | None

Return the current MEMORY.md content, or None if missing.

generate_memory_diff

generate_memory_diff(
    ops: list[dict], previews: dict[int, str] | None = None
) -> str

Render proposed MEMORY.md operations as a readable before/after diff.

Additions show as +, retirements as -, rewrites as the old line (~) followed by its replacement (→), and merges as one - per folded db entry followed by the single consolidated + line. Each op that names backing db entries is annotated with the store-level verb applied to them.

PARAMETER	DESCRIPTION
ops	Structured ops from :func:parse_reflection_response. TYPE: list[dict]
previews	Optional {entry_id: content snippet} used to show the source rows of a merge; falls back to an inline id tag when absent. TYPE: dict[int, str] | None DEFAULT: None

reflect

reflect(
    conn: Connection,
    days: int = 7,
    min_new: int = 3,
    batch_size: int = 50,
    *,
    project: str | None = None,
    branch: str | None = None,
) -> dict

Run the main reflection orchestrator.

Uses a watermark to track which entries have been reflected on, ensuring idempotent operation. Processes entries in batches. When project is provided, reflects only on global plus project-scoped entries.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
days	Bootstrap window in days when no watermark exists. TYPE: int DEFAULT: 7
min_new	Minimum new non-insight entries required to proceed. TYPE: int DEFAULT: 3
batch_size	Maximum entries processed per reflection. TYPE: int DEFAULT: 50
project	Project identifier scoping the reflection. TYPE: str | None DEFAULT: None
branch	Branch name included in the prompt context only. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
dict	A dict with keys skipped, reason, insights, ops (the
dict	structured MEMORY.md operations proposed — see
dict	func:parse_reflection_response), stored, memory_diff, and
dict	batch_info (the last containing processed and watermark
dict	counts, or None when skipped).

Observations

daily_batches

daily_batches(day: str) -> list[dict]

Summarise a day's capture batches from its JSONL observation log.

Each observe-hook run appends one record to memory/observations/<day>.jsonl; this reads them back as light summaries — when and why a capture happened and how much it extracted — without the extraction bodies (those live in memory.db).

Best-effort by design: a missing file, an unreadable file, or a malformed line yields no batch rather than raising — the JSONL is supplementary metadata, never load-bearing.

PARAMETER	DESCRIPTION
day	The log's date stamp, YYYY-MM-DD. TYPE: str

RETURNS	DESCRIPTION
list[dict]	Dicts with timestamp, trigger, project, branch and
list[dict]	counts (items per :data:BATCH_CATEGORIES category; zero-count
list[dict]	categories omitted), in file (chronological) order.

list_observations

list_observations(
    conn: Connection,
    *,
    project: str | None = None,
    limit: int = _FETCH_LIMIT,
) -> list[dict]

List the raw observations, newest first, flagged against the watermark.

Returns every non-insight entry (insights are reflection output, not observations), each carrying the standard entry fields plus reflected: whether the entry sits at or below the relevant reflection watermark — the project's own when project is given, the global one otherwise, matching :func:~mait_code.tools.memory.reflect.count_unreflected. With no watermark for the scope, nothing has been reflected yet.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection
project	Filter to this project context (includes global entries) and judge against its watermark; None for everything. TYPE: str | None DEFAULT: None
limit	Maximum number of results. TYPE: int DEFAULT: _FETCH_LIMIT

RETURNS	DESCRIPTION
list[dict]	A list of dicts ordered by entry ID descending (newest first).

observation_projects

observation_projects(conn: Connection) -> list[str]

List the distinct projects observations span, alphabetically.

Feeds filter UIs (the observations browser's project Select); global entries carry no project and so don't contribute a name.

PARAMETER	DESCRIPTION
conn	Open memory database connection. TYPE: Connection

Native auto memory

list_native_memories

list_native_memories(
    projects_dir: Path | None = None,
    *,
    root: Path = Path("/"),
) -> list[dict]

Enumerate every project's native memory files, across all projects.

Scans projects_dir (default :func:native_projects_dir) for <slug>/memory/**/*.md. Projects without a memory directory, or whose memory directory holds no markdown, are skipped — only projects with something to read appear. A missing or unreadable projects dir yields an empty list rather than an error.

PARAMETER	DESCRIPTION
projects_dir	The Claude Code projects directory to scan. TYPE: Path | None DEFAULT: None
root	Filesystem root for :func:resolve_slug label resolution. TYPE: Path DEFAULT: Path('/')

RETURNS	DESCRIPTION
list[dict]	One dict per project, sorted by label: slug (the munged dir
list[dict]	name), label (the resolved project name, or the slug when
list[dict]	unresolvable), path (the resolved original project path as a
list[dict]	string, or None), memory_dir, and files — MEMORY.md
list[dict]	first, the rest alphabetical, each {name, path, modified}.

native_projects_dir

native_projects_dir() -> Path

Claude Code's per-project state root, usually ~/.claude/projects.

Honours CLAUDE_CONFIG_DIR when set (Claude Code's own override for relocating ~/.claude).

RETURNS	DESCRIPTION
Path	The projects directory path; it may not exist.

resolve_slug

resolve_slug(
    slug: str, root: Path = Path("/")
) -> Path | None

Best-effort reverse of Claude Code's path munging, filesystem-guided.

The munge is lossy — any non-alphanumeric character became - — so the reverse can't be computed from the string alone. Instead this walks the real filesystem from root: at each level it re-munges every existing subdirectory's name (with :func:munge_path) and follows the one whose munged form the remaining slug starts with, backtracking when a branch dead-ends. So -home-w-mait-code resolves to /home/w/mait.code, /home/w/mait_code or /home/w/mait-code — whichever exists.

PARAMETER	DESCRIPTION
slug	A munged directory name from the projects dir. TYPE: str
root	Filesystem root the munged path is relative to (tests inject a temporary tree here). TYPE: Path DEFAULT: Path('/')

RETURNS	DESCRIPTION
Path | None	The resolved original path, or None when no existing directory
Path | None	chain re-munges to the slug (e.g. the project has since been deleted,
Path | None	or its slug was hash-truncated for length).