API Reference

Key entry points for the engrava Python library. For the full reference, see docs/api-reference.md in the package repository.

EngravaManager

Multi-service store manager. Use when your agent needs isolated memory spaces per service (e.g. per-user, per-conversation). Each service gets its own SQLite database under data_dir.

from pathlib import Path

from engrava import EngravaManager

async with EngravaManager(data_dir=Path("./data")) as mgr:
    store = await mgr.get_store("main")
    # store is a fully initialised SqliteEngravaCore
    await mgr.list_services()        # -> ["main"]
    await mgr.delete_service("old")

Methods

SqliteEngravaCore

The primary store. Wraps an already-open aiosqlite.Connection and provides the full thought + edge + search API. All operations are async.

import aiosqlite
from engrava import SqliteEngravaCore

async with aiosqlite.connect("engrava.db") as conn:
    conn.row_factory = aiosqlite.Row
    store = SqliteEngravaCore(conn)
    await store.ensure_schema()

For a config-driven, one-call setup, use the from_config factory — it opens and owns the connection and applies the schema for you. Use it as an async context manager:

async with await SqliteEngravaCore.from_config("engrava.yaml") as store:
    ...  # schema already applied by from_config

SqliteEngravaCore(db_path=...) does not exist — pass an open connection, or use await SqliteEngravaCore.from_config(path).

Thought Methods

create_thought

create_thought takes a single frozen ThoughtRecord object (build it, then pass it) and returns the persisted record. It does not take field keyword arguments and does not return a UUID string.

import uuid
from engrava import ThoughtRecord, ThoughtType, Priority, LifecycleStatus

record = ThoughtRecord(
    thought_id=str(uuid.uuid4()),
    thought_type=ThoughtType.OBSERVATION,
    essence="Short summary (1-200 chars)",
    content="Full content text.",
    priority=Priority.P2,
    lifecycle_status=LifecycleStatus.ACTIVE,
    created_cycle=0,
    updated_cycle=0,
    source="human",
    metadata={"role": "user", "lang": "en"},  # optional structured attributes
)
stored = await store.create_thought(record)
# stored is the persisted ThoughtRecord

thought_type values: TASK, OBSERVATION, BELIEF, REFLECTION, OUTPUT_DRAFT, NOTE.

Keyword-only options: expires_after_seconds (relative TTL) and deduplicate (collapse identical content into the existing thought, bumping its confirmation_count).

remember

remember is the one-call shorthand over create_thought for the common case of storing a bare string — it generates the UUID and fills the fields for you.

stored = await store.remember("User prefers concise answers")
# stored is the persisted ThoughtRecord (thought_type=NOTE, priority=P3,
# essence=text[:200], content=text, created_cycle=updated_cycle=0)

Signature: remember(text, *, metadata=None, deduplicate=False) -> ThoughtRecord. For a write that needs an explicit cognitive cycle, build a ThoughtRecord and call create_thought instead — remember always stamps cycle 0.

recall

recall is the one-call shorthand over search_hybrid for retrieval.

result = await store.recall("what does the user prefer?")
for thought_id, score in result.results:
    ...

Signature: recall(query, *, top_k=10, current_cycle=None, recency_now=None, recency_now_half_life=None, filters=None, visibility=None, collapse_key=None, collapse_max_per_unit=None, include_archived=False) -> HybridSearchResult. With no explicit current_cycle, no configured cycle provider and no recency_now, the recency signal is inactive; pass one of the two references to blend recency into ranking. Every keyword argument other than top_k is delegated to search_hybrid (see Search Methods).

get_thought

thought = await store.get_thought(thought_id)
# Returns ThoughtRecord, or None if not found

update_thought

await store.update_thought(
    thought_id,
    essence="Updated summary",
    priority="P1",
)
# Optimistic-concurrency update; raises ThoughtNotFoundError / StaleDataError

restore_thought

restored = await store.restore_thought(thought_id, current_cycle=42)
# Returns the restored ThoughtRecord (lifecycle_status is now ACTIVE)

Restore an ARCHIVED thought back to ACTIVE, clearing its archive stamp. current_cycle is keyword-only and optional (stamps the new updated_cycle). Raises ThoughtNotFoundError if the thought does not exist, or InvalidTransitionError if it is not currently ARCHIVED.

list_thoughts

thoughts = await store.list_thoughts(
    limit=50,
    lifecycle_status="ACTIVE",     # optional filter
    thought_type="OBSERVATION",    # optional filter
    visibility="public",           # optional; include only this visibility
)

Filters are keyword-only: priority, lifecycle_status, thought_type, min_cycle, max_cycle, visibility, exclude_visibility, include_expired, provenance_filter (a MetadataFilter over the provenance column), limit, offset. Use count_thoughts(...) for a lightweight count under the common filters (lifecycle_status, thought_type, priority, include_expired).

Edge Methods

create_edge

create_edge takes a single EdgeRecord object and returns the persisted record. It raises ReferentialIntegrityError when an endpoint thought does not exist, and DuplicateEdgeError when the same directed endpoints and edge type already identify a persisted edge.

import uuid
from engrava import EdgeRecord, EdgeType

edge = await store.create_edge(
    EdgeRecord(
        edge_id=str(uuid.uuid4()),
        from_thought_id=thought_id,
        to_thought_id=other_id,
        edge_type=EdgeType.ASSOCIATED,
        weight=0.8,               # required; float in [0.0, 1.0]
        created_cycle=0,
        metadata={"subtype": "supports"},  # optional structured attributes
    )
)

edge_type values: ASSOCIATED, DEPENDS_ON, DERIVED_FROM, MESSAGE_OF, BRIDGE, CONSOLIDATED_FROM, CONTESTED_BY.

metadata on an edge

Edges carry the same generic metadata field as thoughts — a dict[str, MetadataValue] defaulting to {}, under the same contract: leaf values must be scalars (str, int, float, bool, None), nested dict[str, MetadataValue] values are accepted for structured namespaces, lists and other rich containers are rejected at write time, floats must be finite, and the serialised payload has a 64 KiB hard limit. It is persisted as an edge.metadata_json column from schema user_version = 19; edges written before that read back as {}.

The keys carry no reserved meaning. Engrava assigns no domain semantics to any edge-metadata key, applies no metadata-driven ranking, enforces no vocabulary, and makes no compatibility promise about key names — so do not treat any key as “well-known”. Engrava’s own edge creators write an empty {}.

get_edges

edges = await store.get_edges(
    thought_id,
    direction="BOTH",  # "IN", "OUT", or "BOTH" (keyword-only)
)

list_edges

from engrava import EdgeType, FieldOp, FieldPredicate, MetadataFilter

supports = await store.list_edges(
    edge_type=EdgeType.ASSOCIATED,
    filters=MetadataFilter([FieldPredicate("$.subtype", FieldOp.EQ, "supports")]),
)

Signature: list_edges(*, edge_type=None, source=None, filters=None, limit=5000) -> list[EdgeRecord], ordered by created_cycle descending. filters is a MetadataFilter over the edge metadata, reusing exactly the machinery that backs thought-metadata filtering: EQ and IN only, the restricted $ / $.key / $[0] path grammar, AND-conjunction, and a 250-predicate cap. An edge whose stored metadata is malformed JSON never matches a non-empty filter.

Like filters on search_hybrid, this is a query refinement, not a security boundary — it enforces nothing and the caller can bypass it.

Valid-time Methods

invalidate_thought / invalidate_edge

Close a record’s valid-time interval at a given instant — mark a fact as no longer true without deleting it. Deterministic, idempotent, non-cascading (invalidating a thought leaves its edges untouched). The row and its history remain on file; a point-in-time query for an instant before the cut-off still returns it.

# Reality changed on 2026-06-01 — close the window, keep the history.
updated = await store.invalidate_thought(
    thought_id,
    valid_until="2026-06-01T00:00:00+00:00",
)
# Returns the updated ThoughtRecord (valid_until now set).

await store.invalidate_edge(
    edge_id,
    valid_until="2026-06-01T00:00:00+00:00",
)
# Returns the updated EdgeRecord.

Use these when a fact stopped being true; use delete_thought only for a fact that should never have existed. See Bi-temporal Model.

Search Methods

search_hybrid

The primary search method — fuses FTS5, vector, recency, priority, and optionally graph signals.

result = await store.search_hybrid(
    "python async agents",        # query_text (positional)
    query_vector=embedding,       # optional; enables vector signal
    top_k=10,
    current_cycle=42,             # optional; enables recency signal
    fts_weight=0.3,               # override per-call
    vector_weight=0.55,
    recency_weight=0.1,
    priority_weight=0.05,
    graph_weight=0.0,             # opt-in
    include_reflections=True,
    reflection_boost=1.0,
)

for thought_id, score in result.results:
    thought = await store.get_thought(thought_id)
    print(f"{score:.3f}  {thought.essence}")

search_hybrid returns a single HybridSearchResult with three fields: result.results (a list[tuple[str, float]] of (thought_id, combined_score), highest first), result.backends_used (a frozenset[str] of the signals that were available for this query, e.g. {"fts5", "vector", "graph_expansion"} — a backend appears even if it returned zero results, and its absence means it was unavailable or not applicable), and result.reflections_evicted (an int). The last one is the count of REFLECTION results that reflection_topk_cap removed from the final top-K window and backfilled with off-list non-REFLECTION candidates: 0 (the default) means the cap changed nothing, so the ranked results are exactly what fusion and boosting produced, while a positive value is the programmatic signal that the cap reshaped the window.

Scoped retrieval — filters and visibility

search_hybrid (and its recall shorthand) accept two optional keyword arguments that narrow the ranked path by metadata:

from engrava import FieldOp, FieldPredicate, MetadataFilter, VisibilityQueryFilter

result = await store.search_hybrid(
    "python async agents",
    filters=MetadataFilter([
        FieldPredicate("$.session_id", FieldOp.EQ, "s-1"),
        FieldPredicate("$.role", FieldOp.IN, ("user", "system")),
    ]),
    visibility=VisibilityQueryFilter(frozenset({"public"}), owner="alice"),
)

• filters — a MetadataFilter, an AND of typed FieldPredicates over a thought’s metadata (EQ for a single scalar, IN for a set). Paths use the restricted JSONPath shape $, $.key, or $[0]. Applied before each arm’s candidate limit, so a filtered query never starves top_k. An empty filter (or None) is a match-all no-op.
• visibility — a VisibilityQueryFilter for the “public-or-mine” pattern: admit rows whose $.visibility is in allowed, plus (optionally) rows whose $.owner equals a given value.

visibility is a query refinement, not access control. It performs no authentication, authorization, or ownership enforcement, the caller can forge owner, and it is bypassable. Do not use it to protect tenant data — for cross-tenant isolation, use one store per tenant via EngravaManager.

Archived rows — include_archived

include_archived (default False) admits thoughts whose lifecycle_status is ARCHIVED back into the candidate set for one call, without restoring them. With the default, archived thoughts are excluded from every ranked path — search_hybrid, recall, search_fts, and search_similar. list_thoughts / count_thoughts are unaffected and stay lifecycle-neutral. Use restore_thought to return a thought to ACTIVE permanently. See Hybrid Search.

De-duplication — collapse_key

collapse_key (a metadata path, or an ordered sequence of paths for a composite key) keeps only the single best-ranked row per caller-defined unit and backfills the freed slots with deeper distinct units — a presentation/de-dup convenience, not a filter. collapse_max_per_unit (an integer >= 1) relaxes the retention to keep up to that many highest-ranked members of each unit. Both default to None; only collapse_key=None leaves the candidate/score/order path byte-identical to the unfiltered query.

Full signature: search_hybrid(query_text, query_vector=None, *, top_k=10, fts_weight=None, vector_weight=None, recency_weight=None, recency_half_life=None, current_cycle=None, recency_now=None, recency_now_half_life=None, fts_top_k=50, vector_top_k=50, priority_weight=None, graph_weight=None, graph_edge_decay=None, include_reflections=True, reflection_boost=None, filters=None, visibility=None, collapse_key=None, collapse_max_per_unit=None, include_archived=False).

search_fts

results = await store.search_fts("keyword query", top_k=10)
# Returns list[tuple[str, float]] — (thought_id, bm25_score)

FTS5-only search, ordered by BM25 score. Fetch each record with get_thought when you need its fields. Takes the same keyword-only include_archived=False.

search_similar

results = await store.search_similar(query_vector, top_k=10)
# Returns list[tuple[str, float]] — (thought_id, score)

Vector-only cosine similarity search. Signature: search_similar(query_vector, top_k=10, threshold=0.0, *, include_archived=False).

The query vector is contract-checked before either backend runs, and the two bad shapes are treated differently:

• A wrong-dimension vector — its length differs from the dimension the store declares — raises VectorDimensionMismatchError (an EngravaError subclass). This is a caller-contract violation, not a degraded query. Code written against 0.5 that caught the incidental numpy ValueError, or that relied on a wrong-length all-zero vector quietly returning [], must catch the typed error instead.
• A degenerate vector — empty, all-zero, or carrying a non-finite component — still returns [], because cosine similarity has no direction to work with. It does not raise; instead it increments the read-only vector_arm_degradation_count counter, so a bad query embedding is distinguishable from an empty neighbourhood.

search_reflections_only

result = await store.search_reflections_only(
    "recurring themes",
    query_vector=embedding,
    top_k=5,
    current_cycle=42,
)

Hybrid search restricted to REFLECTION thoughts. Returns a HybridSearchResult (same shape as search_hybrid), so reflections do not compete against regular thoughts for result slots.

Cognitive-cycle Methods

max_cycle

resume_from = await store.max_cycle()   # 0 on an empty store

A read-only accessor returning the store’s cognitive-cycle high-water mark: the maximum across every cycle-bearing record, MAX(thought.updated_cycle) unioned with MAX(edge.created_cycle). The union is deliberate — an edge created at a higher cycle than any thought would otherwise under-report the mark and let a resumed counter go backwards. Use it to resume your counter after a restart.

Chicken-and-egg, disclosed. On a store where every write is stamped cycle 0 — a consumer that never advances the cycle — max_cycle() returns 0. It helps a consumer that does advance cycles resume; it cannot recover a cadence that was never expressed in the data.

The cycle_provider seam

Threading current_cycle through every search_hybrid / recall / consolidate / run_hygiene call is repetitive when your application already owns a cadence. As an opt-in convenience, pass a cycle_provider once — it is a keyword-only argument on both the constructor and from_config, and a live runtime object, never serialised into engrava.yaml:

from engrava import MaxCycleProvider, SqliteEngravaCore, StaticCycleProvider

store = SqliteEngravaCore(conn, cycle_provider=StaticCycleProvider(0))

# Or resume from the store's own high-water mark:
store = SqliteEngravaCore(conn, cycle_provider=await MaxCycleProvider.create(store))

Resolution is deliberately simple, and an explicit argument always wins:

1. You passed current_cycle (including 0) — that value is used. The check is “was an argument given”, never truthiness, so an explicit 0 never silently falls through to the provider.
2. Otherwise a configured provider is pulled and validated: it must return a real, non-negative int (a bool is rejected). An invalid value raises CycleProviderError.
3. Otherwise the cycle stays None. On the retrieval paths — search_hybrid, recall — that means recency is simply off, exactly as a store built without a provider has always behaved. consolidate() and run_hygiene() genuinely need a cycle for their age arithmetic, so they raise a ValueError naming the operation rather than inventing a default.

Read-time only. The provider feeds ranking and eligibility; it never stamps created_cycle / updated_cycle on writes. Write-side cycles stay your explicit choice on each ThoughtRecord.

Three reference providers ship:

The seam standardises injection; it does not manufacture a cadence. It gives you one plug point instead of threading current_cycle everywhere. A consumer that already advances a cycle can plug it in once; a consumer with no natural cycle still has nothing to supply.

Schema and Lifecycle

await store.ensure_schema()   # idempotent — call once on startup
await store.close()           # closes the connection only if the store owns it
                              # (from_config); for a manually-supplied connection
                              # this is a no-op — close the connection yourself.
                              # Does not checkpoint the WAL.

ThoughtRecord

The core domain model. All fields are immutable (frozen=True).

from engrava import ThoughtRecord

thought: ThoughtRecord = await store.get_thought(thought_id)

thought.thought_id          # UUID str
thought.thought_type        # "TASK" | "OBSERVATION" | "BELIEF" | "REFLECTION" | "OUTPUT_DRAFT" | "NOTE"
thought.essence             # short summary (1-200 chars, indexed for FTS)
thought.content             # full content text
thought.priority            # "P1" | "P2" | "P3" | "P4"
thought.lifecycle_status    # "CREATED" | "ACTIVE" | "DONE" | "ARCHIVED"
thought.confidence          # float | None
thought.source              # origin identifier
thought.metadata            # dict (structured attributes; defaults to {})
thought.created_cycle       # int (creation cycle number)
thought.updated_cycle       # int (last update cycle)
thought.confirmation_count  # int
thought.access_count        # int
thought.created_at          # ISO-8601 str | None (when persisted)
thought.updated_at          # ISO-8601 str | None (last mutation)
thought.valid_from          # ISO-8601 str | None (valid-time lower bound; None = −∞)
thought.valid_until         # ISO-8601 str | None (valid-time upper bound, exclusive; None = +∞)

valid_from / valid_until are the optional valid-time bounds — the real-world window during which the fact is true, independent of when it was stored. Both default to None (valid for all time). EdgeRecord carries the same two fields. See Bi-temporal Model for the semantics and query predicates.

To produce a modified copy:

updated = thought.model_copy(update={"priority": "P1", "confidence": 0.9})

Exceptions

Two names in that list do not descend from EngravaError: ConfigError is a ValueError (an invalid configuration value is a value-domain error, so the historical except ValueError still catches it) and MindQLParseError is a plain Exception. except EngravaError does not cover either.

engrava.MindStoreError is also importable, but it is a deprecated backward-compatibility alias for EngravaError rather than a distinct type; importing it emits a DeprecationWarning. Use EngravaError.

create_edge raises ReferentialIntegrityError when an endpoint thought does not exist, and DuplicateEdgeError when an edge with the same (from_thought_id, to_thought_id, edge_type) identity is already persisted — a collision on the caller-supplied edge_id alone is a different constraint and propagates unchanged. Neither exception is re-exported from the top-level engrava package; catch them via from engrava.domain.exceptions import DuplicateEdgeError, ReferentialIntegrityError. Both subclass EngravaError, not aiosqlite.IntegrityError — an except aiosqlite.IntegrityError does not catch them.

MindQL

MindQL is engrava’s read-only query language. MindQLExecutor runs against an open aiosqlite.Connection (the same connection the store wraps), and execute() takes a parsed MindQLQuery — parse the string first with parse().

from engrava import MindQLExecutor, MindQLResult, parse

executor = MindQLExecutor(conn)  # an aiosqlite.Connection, not a store
result: MindQLResult = await executor.execute(
    parse("FIND thoughts WHERE thought_type = 'OBSERVATION' LIMIT 10")
)
print(result.rows)    # list[dict]
print(result.count)   # int | None (set for COUNT queries)

If your connection is owned by the store, you can run a parsed query directly via the store instead of constructing an executor:

from engrava import parse

result = await store.execute_mindql(
    parse("FIND thoughts WHERE valid_now")
)

execute_mindql is the store-level entry point for the same execution contract — it accepts an already-parsed MindQLQuery and returns a MindQLResult.

Grammar

FIND and COUNT require a table name; the rest is optional:

[EXPLAIN] FIND  <table> [WHERE <bool-expr>] [ORDER BY <sort> ...] [LIMIT <n>] [OFFSET <n>]
[EXPLAIN] COUNT <table> [WHERE <bool-expr>]
[EXPLAIN] SELECT <raw read-only SQL>

Tables: thoughts, edges, embeddings, actions (singular forms accepted too). The command verb is case-insensitive.

• Operators: =, !=, >, <, >=, <=, and IN (...).
• Values: string values are single-quoted and kept verbatim as strings, so source = '007' matches the stored '007'; a bare number is coerced to int / float, so created_cycle = 7 compares as an integer.
• Boolean WHERE: conditions combine with AND, OR, and parentheses, with SQLite precedence (AND binds tighter than OR) — e.g. WHERE (priority = 'P1' OR priority = 'P2') AND source = 'x'. There is no NOT operator.
• ORDER BY: FIND accepts one or more sort keys, each with an optional ASC / DESC, e.g. ORDER BY priority ASC, created_cycle DESC.
• OFFSET: FIND accepts OFFSET <n> (n >= 0) for pagination. SQLite requires a LIMIT for OFFSET, so an OFFSET with no explicit LIMIT applies the default row cap as the limit. Pair it with an ORDER BY so pages are deterministic.
• EXPLAIN: any read query may be prefixed with EXPLAIN to return the compiled SQL without running it.

ORDER BY and OFFSET are rejected on COUNT (which returns a single aggregate row) and on the SELECT passthrough (where you write the SQL yourself). A column outside the per-table allowlist raises MindQLParseError.

FIND

Retrieve rows matching a filter.

await executor.execute(parse("FIND thoughts WHERE lifecycle_status = 'ACTIVE' LIMIT 10"))
await executor.execute(parse("FIND thoughts WHERE priority = 'P1' AND thought_type = 'OBSERVATION'"))
await executor.execute(parse("FIND thoughts WHERE priority IN ('P1', 'P2') ORDER BY created_cycle DESC LIMIT 20"))
await executor.execute(parse("FIND thoughts ORDER BY created_cycle DESC LIMIT 10 OFFSET 20"))
await executor.execute(parse("FIND edges WHERE from_thought_id = 'abc'"))

A FIND with no LIMIT is capped at 100 rows at execution time, so an unqualified FIND thoughts can never trigger an unbounded scan. The cap is applied by the executor, not the parser — parse("FIND thoughts") leaves query.limit as None.

COUNT

Count rows matching a filter.

result = await executor.execute(parse("COUNT thoughts WHERE thought_type = 'OBSERVATION'"))
print(result.count)

SELECT

Read-only SQL passthrough.

result = await executor.execute(
    parse("SELECT thought_id, essence FROM thought WHERE lifecycle_status = 'ACTIVE'")
)
for row in result.rows:
    print(row["thought_id"], row["essence"])

parse

parse() returns a MindQLQuery plan:

from engrava import parse, MindQLQuery

query: MindQLQuery = parse("FIND thoughts WHERE thought_type = 'OBSERVATION' LIMIT 5")
query.command     # MindQLCommand.FIND
query.table       # "thought"
query.conditions  # list of parsed conditions (field / operator / value)
query.where       # boolean-expression tree, or None
query.order_by    # tuple of (column, direction) sort keys; empty when absent
query.limit       # 5
query.offset      # int | None
query.explain     # bool — whether the query was EXPLAIN-prefixed

A WHERE made only of simple comparisons joined by AND populates the flat .conditions list and leaves .where as None. Any use of OR, parentheses, or IN produces a boolean-expression tree on .where instead.

Result Structure

result.columns   # list[str] — column names in result order
result.rows      # list[dict]
result.count     # int | None — set for COUNT queries
result.command   # str — the command that was executed

Embedding Providers

Install one of the extras and pass a provider to the store:

pip install 'engrava[embeddings-local]'     # sentence-transformers
pip install 'engrava[embeddings-openai]'    # OpenAI-compatible text-embedding-*
pip install 'engrava[embeddings-ollama]'    # Ollama local models
pip install 'engrava[embeddings-hf]'        # Hugging Face Inference API

from engrava import SentenceTransformerProvider

provider = SentenceTransformerProvider(model_name="all-MiniLM-L12-v2")
vector = await provider.embed("text to embed")
await store.store_embedding(thought_id, vector, model_name=provider.model_name)

store_embedding is keyword-only for model_name (and optional embedding_id); the vector dimension is derived from len(vector).

The CallbackProvider accepts any Python callable as an embedding function — useful for testing or for providers not yet wrapped:

from engrava import CallbackProvider

provider = CallbackProvider(
    callback=lambda text: [0.0] * 384,
    dimension=384,
    model_name="stub",
)