Engrava

API Reference

Complete reference for engrava's public API.

Core Store

SqliteEngravaCore

The main persistence engine. All operations are async.

SqliteEngravaCore wraps an already-open aiosqlite.Connection. For a config-driven, one-call setup that opens and owns the connection, use the from_config factory instead.

import aiosqlite
from engrava import SqliteEngravaCore

# Abbreviated for brevity: a bare async-with here can let a close failure
# mask a real error. See "Graceful shutdown" in deployment.md for the safe
# explicit-close shape.
async with aiosqlite.connect(":memory:") as conn:
    conn.row_factory = aiosqlite.Row
    store = SqliteEngravaCore(conn)
    await store.ensure_schema()

# Or, config-driven (opens and owns the connection):
async with await SqliteEngravaCore.from_config("engrava.yaml") as store:
    ...  # schema already applied by from_config

Constructor

The SqliteEngravaCore(db_path=...) form does not exist — pass a connection, or use await SqliteEngravaCore.from_config(path).

Factory

Schema

Thought CRUD

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 text content.",
    priority=Priority.P2,
    lifecycle_status=LifecycleStatus.ACTIVE,
    created_cycle=0,
    updated_cycle=0,
    source="human",
)
stored = await store.create_thought(record)

from engrava import LifecycleStatus

# Archive, then restore — a lossless round-trip.
await store.update_thought(stored.thought_id, lifecycle_status=LifecycleStatus.ARCHIVED)
restored = await store.restore_thought(stored.thought_id, current_cycle=42)
assert restored.lifecycle_status is LifecycleStatus.ACTIVE

create_thought keyword-only options

list_thoughts / count_thoughts filters (keyword-only)

list_thoughts also supports visibility, exclude_visibility, and provenance_filter (see Provenance capture).

Provenance capture

Attach optional, typed write-time provenance to a thought — the signals that become irrecoverable once a synthesised thought exists (which session/actor produced it, what query and instruction shaped it, which thoughts it was built from). Set ThoughtRecord.provenance to a ProvenanceContext before create_thought; every field is optional and bounded, and the default None is byte-identical to not using provenance at all.

from engrava.domain.models.provenance import ProvenanceContext

record = ThoughtRecord(
    # ... the usual fields ...
    provenance=ProvenanceContext(
        session_id="sess-42",                              # indexed identity hint
        actor_id="agent-a",                                # indexed identity hint
        retrieval_query="remote work trade-offs",          # ≤4096 chars
        instruction_context="summarise for a busy exec",   # ≤4096 chars
        retrieval_context_ids=["t-1", "t-2"],              # ≤128 ids, ≤256 chars each
    ),
)
await store.create_thought(record)

Query by provenance. list_thoughts(provenance_filter=...) takes a MetadataFilter — an AND of FieldPredicates over the provenance JSON column. Predicates on $.session_id / $.actor_id use the provenance identity index; rows whose provenance is NULL or malformed JSON never match a non-empty filter.

from engrava.domain.models.filters import FieldOp, FieldPredicate, MetadataFilter

mine = await store.list_thoughts(
    provenance_filter=MetadataFilter(
        [FieldPredicate("$.session_id", FieldOp.EQ, "sess-42")]
    ),
)

Untrusted hint — never identity, authentication, or authorization. Provenance is descriptive-only: the engine grants it zero authority and consults it for no access, ranking, or consolidation decision. actor_id is not a tenant boundary — tenant isolation is the store's file boundary (one store per tenant). The engine never infers provenance; the caller passes it explicitly.

Edge CRUD

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, typed relationship already exists.

The database permits at most one edge for each (from_thought_id, to_thought_id, edge_type) tuple, independently of edge_id; attempting to insert the same directed, typed relationship twice raises DuplicateEdgeError. Reversing the endpoints or using a different edge type produces a distinct edge. A duplicate caller-supplied edge_id for a different relationship remains a separate integrity failure. A hard thought delete cascades to edges where the thought is either endpoint; invalidate_thought() does not remove or invalidate any edge.

Below core schema 12 there is no such cascade. The ON DELETE CASCADE on edge, embedding and action arrives with the core-12 migration. delete_thought does not rely on it: it issues its own deletes for the thought's rows in those three tables, in the same savepoint as the parent delete, which runs first. See Deletion on a database that has not been migrated for what engrava migrate cleans up on a database that already holds dangling embedding rows.

import uuid
from engrava import EdgeRecord, EdgeType

await store.create_edge(
    EdgeRecord(
        edge_id=str(uuid.uuid4()),
        from_thought_id=src_id,
        to_thought_id=dst_id,
        edge_type=EdgeType.ASSOCIATED,
        weight=0.8,
        created_cycle=0,
    )
)

REFLECTION lineage

Helpers for navigating the CONSOLIDATED_FROM graph that dreaming builds between a REFLECTION and the source thoughts it summarises.

# Walk a REFLECTION down to its sources, and back from a source to its REFLECTIONs.
member_ids = await store.consolidated_member_ids(reflection_id)
for thought_id in member_ids:
    source = await store.get_thought(thought_id)
    if source is not None:
        print(source.essence)

# Detect an orphaned cluster — every source archived/gone:
statuses = await store.consolidated_source_statuses(reflection_id)
is_orphaned = bool(statuses) and all(s != "ACTIVE" for s in statuses)

# Reverse direction: which REFLECTIONs summarise this source?
parents = await store.reflections_consolidated_from(member_ids[0])

# Exact-source existence check (e.g. dreaming's idempotency guard — a REFLECTION's
# source is "dreaming:<cluster_hash>", so match the full value, not a prefix):
exists = await store.thought_exists_by_source(
    source="dreaming:abc123def4567890", thought_type_value="REFLECTION"
)

Embedding Operations

Full-Text & Hybrid Search

search_fts and search_similar return (thought_id, score) tuples — fetch the record with get_thought when you need its fields. search_hybrid returns a single HybridSearchResult container.

The complete search_hybrid signature is:

async def search_hybrid(
    query_text: str,
    query_vector: list[float] | None = None,
    *,
    top_k: int = 10,
    fts_weight: float | None = None,
    vector_weight: float | None = None,
    recency_weight: float | None = None,
    recency_half_life: int | None = None,
    current_cycle: int | None = None,
    recency_now: str | None = None,
    recency_now_half_life: int | None = None,
    fts_top_k: int = 50,
    vector_top_k: int = 50,
    priority_weight: float | None = None,
    graph_weight: float | None = None,
    graph_edge_decay: float | None = None,
    include_reflections: bool = True,
    reflection_boost: float | None = None,
    filters: MetadataFilter | None = None,
    visibility: VisibilityQueryFilter | None = None,
    collapse_key: str | Sequence[str] | None = None,
    collapse_max_per_unit: int | None = None,
    include_archived: bool = False,
) -> HybridSearchResult: ...

current_cycle and recency_now are mutually exclusive when both are supplied explicitly. recency_now_half_life is measured in seconds and is only valid with recency_now; recency_half_life is measured in cognitive cycles.

Archived thoughts are excluded by default. recall, search_fts, search_hybrid, and search_similar drop ARCHIVED thoughts from the default candidate set (the mechanism behind Forgetting). Pass include_archived=True to re-admit them for one call, or restore_thought to re-activate a row. search_reflections_only remains restricted to active reflections and has no archive opt-in. See Search.

Scoped & de-fragmented retrieval (keyword-only)

Both recall and search_hybrid accept the same four opt-in keywords for narrowing and de-duplicating ranked results. Every one defaults to None and the None path is byte-identical to an unscoped query, so existing callers see no change.

from engrava import MetadataFilter, VisibilityQueryFilter
from engrava import FieldOp, FieldPredicate

# Scope recall to one session's memories, keep only "public" or my own rows,
# and collapse per-turn fragments to one best row each.
result = await store.recall(
    "remote work trade-offs",
    top_k=10,
    filters=MetadataFilter([FieldPredicate("$.session_id", FieldOp.EQ, "sess-42")]),
    visibility=VisibilityQueryFilter(frozenset({"public"}), owner="alice"),
    collapse_key="$.session_turn",
    collapse_max_per_unit=2,   # keep up to 2 rows per turn instead of 1
)

Metrics & maintenance

Read-only health counters (plain properties, not await, not in the metrics snapshot): store.fts_match_failure_count and store.vector_arm_degradation_count are monotonic int counters that surface silent, self-healing search-arm degradations. See Observability signals.

# Bulk ingest: one transaction instead of one commit per write.
async with store.suspend_auto_commit():
    for record in many_records:
        await store.create_thought(record)
# commit happens once on clean exit; any exception rolls the whole block back

ReadOnlyEngrava

A composition wrapper that delegates reads to the wrapped store and raises ReadOnlyViolationError on any write. Use it to hand a retrieval-only view of shared memory to a component that should never mutate it — e.g. a sub-agent or worker whose job is only to look things up. Delegated reads suppress access tracking, so they do not stage deferred access-count mutations.

The guarantee is behavioural, not structural. ReadOnlyEngrava implements the narrower EngravaReadProtocol, but also exposes signature-compatible write methods whose only behaviour is to raise. Because EngravaCoreProtocol is a runtime-checkable structural protocol, those blockers mean isinstance(ro, EngravaCoreProtocol) is also True. Use EngravaReadProtocol to type a read capability; do not use the structural core check as evidence that a value is writable.

from engrava import ReadOnlyEngrava

ro = ReadOnlyEngrava(store)
thought = await ro.get_thought("abc")  # OK
await ro.create_thought(...)           # Raises ReadOnlyViolationError

EngravaManager

Multi-service database isolation. Each service owns a separate <data_dir>/<service_name>.db file, with its own schema, connection, embedding configuration, and journal. Service names accepted by get_store(), service_exists(), and delete_service() must match ^[a-z][a-z0-9_-]{0,62}$: start with a lowercase letter, then use lowercase letters, digits, _, or -, up to 63 characters.

from pathlib import Path

from engrava import EngravaManager

async with EngravaManager(data_dir=Path("./data")) as mgr:
    store = await mgr.get_store("my-service")
    assert mgr.service_exists("my-service")
    print(await mgr.list_services())  # ["my-service"]

# The context manager has closed its cached connections, but the database
# remains. Delete is a separate, destructive operation:
async with EngravaManager(data_dir=Path("./data")) as mgr:
    await mgr.delete_service("my-service")

Domain Models

All models are frozen Pydantic objects. For ThoughtRecord, use thought.evolve(...) when changing lifecycle or temporal fields: it validates lifecycle transitions, preserves created_at immutability, refreshes updated_at unless supplied explicitly, and revalidates the complete record. Likewise, ActionRecord.evolve(...) validates action-status transitions.

Pydantic's model_copy(update={...}) does not validate its update payload. Reserve it for trusted, already-validated changes to fields on which no domain invariant depends (or use model_copy() without update for an unchanged copy). Do not use it for ThoughtRecord lifecycle or timestamp changes.

ThoughtRecord

metadata field

MetadataValue = str | int | float | bool | None | dict[str, MetadataValue] — leaf values must be JSON-serializable scalars; nested dict values are accepted for structured namespaces (e.g. metadata["source"] = {"is_self": True, "confidence": "high"}). Lists, tuples, and other rich containers are rejected at the create_thought / update_thought boundaries with a ValueError.

Conventional keys (recommended, not enforced — callers decide what to populate):

await store.create_thought(
    ThoughtRecord(
        thought_id="...",
        thought_type=ThoughtType.OBSERVATION,
        essence="...",
        content="...",
        priority=Priority.P3,
        lifecycle_status=LifecycleStatus.CREATED,
        created_cycle=0,
        updated_cycle=0,
        source="ingest",
        metadata={"role": "user", "lang": "en", "turn_index": 5},
    ),
)

Size limits. Serialized payloads above ~4 KiB emit a WARNING log; payloads above ~64 KiB are rejected with ValueError. If the data is genuinely large (e.g. transcripts, structured documents), store it in content or via an external reference rather than in metadata.

Persistence. Stored as a metadata_json TEXT NOT NULL DEFAULT '{}' column on the thought table since user_version = 11. JSON1 extension is recommended for filtering queries (json_extract(metadata_json, '$.role')).

EdgeRecord

metadata field (edges)

Edges carry the same generic metadata field as thoughts — dict[str, MetadataValue], default {} — with the identical value domain (str | int | float | bool | None | dict[str, MetadataValue]; lists and non-finite floats such as NaN / Infinity are rejected) and the same ~4 KiB warn / ~64 KiB hard-reject size limits. It is validated at the create_edge / update_edge boundaries with a ValueError.

Unlike thought metadata, edge metadata keys carry no reserved meaning and no conventional-key table — the field is a purely generic carrier. Engrava assigns no domain semantics to any key, applies no metadata-driven ranking, enforces no vocabulary, and makes no compatibility guarantee about key names, so do not treat any key as "well-known". Engrava's own edge creators (dreaming, derived records) write an empty {}.

Filtering. list_edges(filters=...) accepts a typed MetadataFilter — an AND of FieldPredicates over the edge metadata — reusing the exact machinery that backs thought-metadata filtering: operators EQ and IN only, the restricted $ / $.key / $[0] JSONPath grammar, and a 250-predicate cap. An edge whose stored metadata is malformed JSON never matches a non-empty filter. This is a query refinement, not a security boundary.

from engrava import EdgeType
from engrava.domain.models.filters import FieldOp, FieldPredicate, MetadataFilter

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

Persistence. Stored as a metadata_json TEXT NOT NULL DEFAULT '{}' column on the edge table since user_version = 19.

EmbeddingRecord

ActionRecord

Records an action the agent took (a tool call, a message, …), linked to the thought that prompted it, with execution and verification state.

Store methods (on SqliteEngravaCore):

ActionStatus is a state machine: PLANNED → EXECUTING → CONFIRMED / FAILED, and PLANNED → BLOCKED → PLANNED. can_transition_to(...) / evolve(...) enforce valid transitions on the in-memory record, and update_action(...) enforces the same transitions when advancing the stored action (an illegal change raises InvalidTransitionError).

When a linked action reaches a terminal state (CONFIRMED / FAILED), the source thought's action_outcome_score — the mean outcome value over its terminal actions, None when it has none — is recomputed. That score is also read by the optional action_outcome dreaming signal and by Memory Hygiene's usage-history gate and scoring path.

The per-action value is exact and depends on both terminal execution status and verification:

action_outcome_score is the arithmetic mean of those values across terminal actions only. It remains None when a thought has no terminal action.

import uuid
from engrava import ActionRecord, ActionType, ActionStatus, VerificationStatus

action = ActionRecord(
    action_id=str(uuid.uuid4()),
    source_thought_id=prompting_thought_id,
    action_type=ActionType.TOOL_CALL,
    intent="search the web for flight prices",
    status=ActionStatus.PLANNED,
    verification_status=VerificationStatus.PENDING,
)
await store.create_action(action)

# advance the STORED action through its lifecycle (journaled; validates each
# transition). A terminal status recomputes the source thought's outcome score:
await store.update_action(action.action_id, status=ActionStatus.EXECUTING)
await store.update_action(action.action_id, status=ActionStatus.CONFIRMED)
# verification can still advance while the status stays terminal:
await store.update_action(
    action.action_id, verification_status=VerificationStatus.CONFIRMED
)

actions = await store.get_actions(prompting_thought_id)

HybridSearchResult

A frozen container of ranked results plus backend diagnostics. It is not a per-result object with score breakdowns.

result = await store.search_hybrid("query text", top_k=5)
for thought_id, score in result.results:
    record = await store.get_thought(thought_id)
    ...

Forgetting (Memory Hygiene) types

Configured under hygiene_policy and returned by run_hygiene — see Memory Hygiene and Configuration → hygiene_policy.

Metadata Helpers

Three exported helpers build the structured metadata dict that pins a thought's origin (self vs external, source id, language). They are pure functions — same arguments always return an equal dict — and you are free to pass a literal dict instead; the helpers exist to remove typo-driven shape mismatches at the call site.

from engrava import percept

metadata = percept(source_id="user-1", label="user")
# -> {'perspective': 'percept',
#     'source': {'is_self': False, 'confidence': 'high', 'id': 'user-1', 'label': 'user'},
#     'lang': 'en', 'content_type': 'natural_language'}

Pass the returned dict as ThoughtRecord(..., metadata=...).

Enums

All enums are StrEnum — JSON-serializable and stored as strings.

Exceptions

The table below defines the public types. For operational retry, repair, and store-replacement guidance, see Error handling and recovery.

create_edge raises ReferentialIntegrityError when an endpoint thought does not exist and DuplicateEdgeError when the relationship already exists. All three of ReferentialIntegrityError, DuplicateEdgeError, and CoreMigrationError are importable directly from engrava, the same as every other exception in the table above.

Protocols

EngravaReadProtocol

The non-mutating capability shared by the writable store and ReadOnlyEngrava: thought/edge/action/embedding reads, FTS/vector/hybrid retrieval, recall, metrics, and max_cycle. Read-only retrieval suppresses the concrete SQLite store's optional access-frequency instrumentation. ReadOnlyEngrava declares and type-checks against this read-side protocol. Its compatibility write methods are blocked and raise ReadOnlyViolationError.

EngravaCoreProtocol

The writable abstract interface for an engrava implementation. It extends EngravaReadProtocol, the separately exported read capability implemented by both SqliteEngravaCore and ReadOnlyEngrava. Because runtime structural checks inspect member names and ReadOnlyEngrava exposes rejecting write methods, the wrapper also satisfies isinstance(ro, EngravaCoreProtocol). That result describes structural compatibility, not permission to write; use the narrower read protocol for capability boundaries.

CycleProvider

A runtime-checkable, synchronous pull protocol with one method: current_cycle() -> int. It supplies a cognitive cycle only when a supported call omits an explicit one and no explicit recency_now selects transaction-time recency. It never advances a cycle, never stamps writes, and is never loaded from YAML. The store validates each pulled value as a real, non-negative int.

EngravaHooksProtocol

Extension hook interface — see Extensions.

DerivedRecordProducerProtocol

Optional capability: return N derived records from one stored thought. Core persists each as an ordinary thought after the source is durable, through a core-owned, guarded, per-child lifecycle. See Extension hooks §1A.

class DerivedRecordProducerProtocol(Protocol):
    async def derive_records(
        self, thought: ThoughtRecord, ctx: DeriveContext
    ) -> Sequence[DerivedRecord]: ...

Companion generic types (all public, X.Y.x-stable): DerivedRecord (producer-owned non-empty content / thought_type / priority / metadata / attach_provenance_edge; core derives the essence and owns identity), DeriveContext (source id, content hash, cycle, informational origin; no store handle), DeriveGates (enabled, on_error, max_derived_per_source), and DeriveResult (the created / reused / skipped tally returned by derive_existing). Configure via the derive: YAML section or the derive_gates= constructor argument. The shipped reference producer is StructuralSplitProducer (see Structural split modes).

EmbeddingProviderProtocol

class EmbeddingProviderProtocol(Protocol):
    @property
    def dimension(self) -> int: ...
    @property
    def model_name(self) -> str: ...
    async def embed(self, text: str) -> list[float]: ...

MindQL

See MindQL for the query language reference.

execute_mindql (on the store)

SqliteEngravaCore exposes a convenience that runs a MindQL query directly against the store's own connection, so you don't have to construct a MindQLExecutor yourself. Like the executor, it takes a parsed MindQLQuery — parse() the string first.

from engrava import parse

result = await store.execute_mindql(
    parse("FIND thoughts WHERE lifecycle_status = 'ACTIVE' LIMIT 10")
)
print(result.rows)

MindQLExtension, ReadOnlyAccessor, and ReadOnlyCursor

MindQLExtension registers a custom MindQL command: command_name, handler, description, and category (defaults to "extension"). The executor invokes handler with two positional arguments — a ReadOnlyAccessor and the parsed argument list — and awaits a list[dict[str, object]] back. See Custom MindQL Commands for a worked example.

A handler receives a ReadOnlyAccessor, never the store's live connection — its only capability is execute(), restricted to a single SELECT:

See Extension Commands for the read-only guarantee this closes off.

MindQLExecutor

MindQLExecutor runs against an open aiosqlite.Connection (the same connection the store wraps), and execute() takes a parsed MindQLQuery — parse the string first.

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)

MindQLResult fields: columns: list[str], rows: list[dict], count: int | None, command: str.

parse

parse() returns a MindQLQuery plan (not a MindQLCommand). The grammar requires a table name and an optional WHERE clause: FIND <table> [WHERE <field> <op> '<value>' [AND ...]] [LIMIT n].

from engrava import parse, MindQLQuery

query: MindQLQuery = parse("FIND thoughts WHERE thought_type = 'OBSERVATION' LIMIT 5")
print(query.command)     # MindQLCommand.FIND
print(query.table)       # "thought"
print(query.conditions)  # list of parsed conditions (field / operator / value)
print(query.limit)       # 5

See MindQL for the full grammar, operators, and per-table filterable columns.

Additional top-level exports

The sections above cover the primary store, records, protocols, and query surface in depth. The remaining names exported directly from engrava are listed here so the top-level package has a complete index; follow the linked specialist guide for operational detail.

Configuration objects and resolvers

See Configuration for the fields, defaults, validation, and YAML examples.

Embedding and vector implementations

See Embeddings and Performance before selecting a provider or vector backend.

Dreaming and scoring types

See Dreaming for activation, gates, signal semantics, and the no-retrieval-accuracy-claim boundary.

Journal, cleanup, and metrics value objects

Discovery and parsing helpers

Deprecated compatibility aliases

CoreThoughtRecord is a compatibility alias of ThoughtRecord. The legacy MindStore-era names below resolve lazily to their Engrava equivalents and emit a DeprecationWarning; new code should not use them.