Extensions

engrava provides a hook-based extension system that lets you plug into the thought lifecycle without modifying core code.

EngravaHooksProtocol

All extensions implement the EngravaHooksProtocol:

from engrava import (
    EngravaHooksProtocol,
    ThoughtRecord,
    ScoringContext,
    MindQLExtension,
)

class MyHooks(EngravaHooksProtocol):
    async def on_store(self, thought: ThoughtRecord) -> ThoughtRecord:
        """Called after a thought is persisted. Return the (enriched) thought."""
        return thought

    async def on_retrieve(self, thought: ThoughtRecord) -> ThoughtRecord:
        """Called after a thought is loaded from DB. Return the (enriched) thought."""
        return thought

    async def score_function(
        self, thought: ThoughtRecord, context: ScoringContext
    ) -> float:
        """Custom relevance score (reserved — not currently called by core)."""
        return thought.confidence or 0.5

    async def decay_function(
        self, thought: ThoughtRecord, elapsed_cycles: int
    ) -> float:
        """Decay multiplier used by an enabled Memory Hygiene pass."""
        return 1.0

    def mindql_extension_registry(self) -> dict[str, MindQLExtension]:
        """Reserved — core wires MindQL verbs via ExtensionManifest, not this hook."""
        return {}

Core invokes on_store, on_retrieve and decay_function. The last is consulted only when an enabled Memory Hygiene pass reaches archive scoring, and hygiene_policy.enabled is false by default — so a stub written on the assumption that nothing calls it can end up deciding which thoughts get archived. score_function and mindql_extension_registry() are reserved and not called by core. Subclass DefaultEngravaHooks if you only want to override one or two methods.

Using Hooks

Pass hooks when creating a store (the store wraps an open connection):

import aiosqlite
from engrava import SqliteEngravaCore

hooks = MyHooks()
async with aiosqlite.connect("my.db") as conn:
    conn.row_factory = aiosqlite.Row
    store = SqliteEngravaCore(conn, hooks=hooks)
    await store.ensure_schema()
    # on_store now runs on the inserts create_thought performs; on_retrieve runs
    # on get_thought and list_thoughts (not on search results); decay_function
    # runs only when this store executes an enabled Memory Hygiene pass

Default Hooks

If no hooks are provided, DefaultEngravaHooks is used — every method is a pass-through or returns a neutral value. on_store and on_retrieve return the thought unchanged and mindql_extension_registry() returns an empty dict; decay_function returns 1.0 (no decay) and score_function returns a score derived from the thought’s priority (P1 = 4.0 down to P4 = 1.0), which core does not currently call.

Hook Interface Reference

Hook contracts:

• on_store and on_retrieve are the data-flow methods: both async, both return a value.
• Hooks must not raise — unexpected exceptions will propagate to the caller.
• Hooks must not have side effects that modify shared state; return an enriched copy instead.
• Engrava is frozen=True-first — if you need to mutate a ThoughtRecord, return thought.model_copy(update={...}).

Write Your Own Hook in 20 Lines

on_store is the one to start with: core calls it for you, with nothing to enable first.

from __future__ import annotations

from engrava.domain.models.thought import ThoughtRecord
from engrava.domain.protocols.hooks import DefaultEngravaHooks


class WordCountHooks(DefaultEngravaHooks):
    """Returns the thought with a word_count entry added to its metadata."""

    async def on_store(self, thought: ThoughtRecord) -> ThoughtRecord:
        counted = {**thought.metadata, "word_count": len(thought.content.split())}
        return thought.model_copy(update={"metadata": counted})


# Registration:
import aiosqlite
from engrava import SqliteEngravaCore

async def build_store(db_path: str) -> SqliteEngravaCore:
    conn = await aiosqlite.connect(db_path)
    conn.row_factory = aiosqlite.Row
    store = SqliteEngravaCore(conn, hooks=WordCountHooks())
    await store.ensure_schema()
    return store

Core calls on_store after the row is inserted, and hands its return value back to the caller — so create_thought gives you the annotated record while the stored row keeps the metadata you submitted. Enrichment that has to survive a reload goes on the record before you store it.

DefaultEngravaHooks supplies the other hooks, so override only the ones you care about.

Derived Records

on_store is one-in / one-out, so it cannot express “turn one stored thought into several records” — splitting a document into sections, distilling an observation into atomic facts, extracting structured items. The derived-records seam fills that gap as a separate, optional capability protocol, so EngravaHooksProtocol is unchanged and an existing hooks class keeps working byte-identically.

The contract

Implement derive_records on your hooks object. Core detects the capability with isinstance(hooks, DerivedRecordProducerProtocol) — it is @runtime_checkable; if the method is absent the seam is simply absent.

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

• It is called only after the source thought is durable, and only when the seam is enabled. If on_store raises, derivation never runs.
• The producer describes what to derive; core owns how it is persisted. A DerivedRecord carries only producer-owned fields — a non-empty content, thought_type, priority, a metadata payload, and an attach_provenance_edge flag. Identity, essence, timestamps, cycle, and lifecycle status are assigned by core and are not representable on the type, so there is nothing for a producer to forge.
• DeriveContext exposes only stable facts about the source (source_thought_id, source_content_hash, cycle_at_derivation) plus an informational origin. It carries no store handle: a producer must not persist, query, or mutate anything itself, and must not spawn background tasks.
• Make the output a deterministic function of the source — that is what buys exact idempotency.

Persistence is source-first, per-child, deferred, and non-atomic. For each returned record, in producer order, core runs the lifecycle an ordinary thought gets: insert → commit → auto-embed → (if requested) attach the single DERIVED_FROM provenance edge from child back to source. A child’s row commits as its own durable unit and its enrichment completes afterwards, so a crash mid-family leaves a partial but regenerable result — re-running derivation fills the gaps, because content-addressed identity makes a re-run idempotent.

That per-child durability describes a create outside a caller-held suspend_auto_commit window. A create issued inside one does not auto-derive at all: the caller owns the open transaction and the source is not yet durable, so trigger derivation with an explicit backfill once your transaction has committed.

Gates

Configure the seam with DeriveGates (or the derive: YAML section):

Durability is decoupled from derivation: the source is always durable even when a producer or a child fails, so with on_error="raise" a caller can see an error while the source persists.

The shipped producer — StructuralSplitProducer

StructuralSplitProducer is a complete reference consumer that runs purely on the stored text: no model, no network, no external service.

from engrava import DeriveGates, SqliteEngravaCore, SplitMode, StructuralSplitProducer

store = SqliteEngravaCore(
    conn,
    hooks=StructuralSplitProducer(),
    derive_gates=DeriveGates(enabled=True),
)
await store.ensure_schema()
# Storing a multi-paragraph thought now also persists one derived child per
# paragraph, each carrying a DERIVED_FROM edge back to the source.

Two deterministic split modes are selected with split_mode:

In fixed-window mode, window_size (default 1000) and window_overlap (default 0) are counted in window_unit — "char" (default) or "word" — windows advance by window_size - window_overlap and fully cover the content. Every child records its split_mode, segment_index, and source char_start / char_end in its metadata. In either mode, fewer than two resulting segments produces no children: a single segment is not a structural split.

Backfilling an existing store — derive_existing

derive_records fires automatically on a durable create. To run a producer over thoughts that are already stored, call derive_existing:

result = await store.derive_existing(thought_id)
print(result.created, result.reused, result.skipped)

Two properties make this usable on a live base:

• It converges with the automatic path. Backfilled children go through the same core-owned per-child lifecycle, content-addressed identity, and DERIVED_FROM edge as an on-store write, so the output is byte-identical to what an on-store write would have produced for the same content — backfilled and auto-derived records dedup against one another. Re-running is idempotent: already-present children come back as reused, not duplicates.
• Its gating is independent of the master switch. It runs whenever a producer capability is present, honouring on_error and max_derived_per_source but not DeriveGates.enabled — so you can backfill an existing base once without committing to automatic derivation on every future write. With no producer registered it is a clean no-op.

Unlike the automatic trigger, derive_existing does not early-return inside a caller-held suspend_auto_commit window — the source is already durable, so there is nothing to defer. In that case the children join the caller’s open transaction like any other write instead of committing per-child, and under on_error="raise" a child failure rolls that whole transaction back. The already-committed source thought is unaffected.

derive_existing raises SourceThoughtNotFoundError when the id does not exist (a precondition failure, distinct from the clean empty result returned for an already-derived source), and DerivedRecordError under on_error="raise" when a producer’s return violates the deterministic contract. A source that is itself a derived record is never re-derived.

Custom MindQL Commands

A custom command is an MindQLExtension. Its handler is an async callable that the executor invokes with two positional arguments — the open aiosqlite.Connection and the parsed argument list — and returns a list[dict[str, object]]. The MindQLExtension fields are command_name, handler, description, and category (there is no help_text field):

import aiosqlite
from engrava import MindQLExtension


async def _handle_stats(
    db: aiosqlite.Connection,
    args: list[str],  # noqa: ARG001 — STATS takes no args
) -> list[dict[str, object]]:
    cursor = await db.execute(
        "SELECT thought_type, COUNT(*) AS n FROM thought GROUP BY thought_type"
    )
    rows = await cursor.fetchall()
    return [{row["thought_type"]: row["n"]} for row in rows]


STATS_COMMAND = MindQLExtension(
    command_name="STATS",
    handler=_handle_stats,
    description="Show thought statistics",
)

Then run it through the executor, passing the command in extensions= and telling parse() which verbs are registered:

from engrava import MindQLExecutor, parse

executor = MindQLExecutor(conn, extensions={"STATS": STATS_COMMAND})
result = await executor.execute(parse("STATS", known_extensions={"STATS"}))

Dreaming Extension

The built-in DreamingExtension performs periodic memory consolidation:

from engrava import DreamingExtension, DreamingConfig, DreamingGates

config = DreamingConfig(
    enabled=True,
    candidates_limit=100,
    promote_threshold=0.6,
    gates=DreamingGates(
        min_confirmations=2,
        min_age_cycles=1,
        max_promoted_per_run=20,
    ),
)

dreaming = DreamingExtension(config=config)
result = await dreaming.run_consolidation(store, current_cycle=42)
print(f"Promoted {result.promoted_count} thoughts")

The weighted-score cutoff is DreamingConfig.promote_threshold; DreamingGates controls eligibility (confirmations, age, per-run cap, and the clustering/quality thresholds). See Dreaming for the full consolidation reference.

Custom Signals

DreamingSignalProtocol is a callable protocol — implement __call__(thought, ctx) returning a score in [0.0, 1.0]. There is no name/weight attribute or score() method; a signal’s weight is set separately in DreamingConfig.signals, and the instance is wired in via DreamingExtension(config, custom_signals={...}).

from engrava import DreamingContext, DreamingExtension, DreamingConfig, ThoughtRecord
from engrava import Priority


class ImportanceSignal:
    """Custom scoring signal — must be callable as (thought, ctx) -> float."""

    def __call__(self, thought: ThoughtRecord, ctx: DreamingContext) -> float:
        if thought.priority == Priority.P1:
            return 1.0
        if thought.priority == Priority.P2:
            return 0.7
        return 0.3


# Register the signal AND give it a weight in the signals map, or it never runs.
dreaming = DreamingExtension(
    config=DreamingConfig(enabled=True, signals={"importance": 0.3}),
    custom_signals={"importance": ImportanceSignal()},
)

Extension Manifest

For distributing extensions as packages, use ExtensionManifest:

from pathlib import Path
from engrava import ExtensionManifest

manifest = ExtensionManifest(
    name="my-engrava-plugin",
    version="1.0.0",
    hooks_class=MyHooks,
    mindql_extensions=[],
    schema_migrations=[
        Path("migrations/001_initial.sql"),
        Path("migrations/002_add_tags.sql"),
    ],
)

Migration Files

Place SQL migration scripts alongside your extension package using the convention NNN_slug.sql (e.g. 001_initial.sql, 002_add_tags.sql). The runner sorts files lexicographically and applies them in order.

Each .sql file should contain SQLite DDL or DML statements, each terminated by a semicolon. Use CREATE TABLE IF NOT EXISTS to keep migrations idempotent.

Transaction control is rejected. The runner applies every migration inside a SAVEPOINT it owns, so a statement beginning BEGIN, COMMIT, END, ROLLBACK, SAVEPOINT or RELEASE raises ExtensionMigrationError during preparation — before any statement of that file runs. A file whose final statement is missing its ;, or that leaves a string literal or block comment unterminated, is rejected the same way.

Loading Extensions with Migrations

Pass manifests explicitly to SqliteEngravaCore. Schema migrations are applied automatically during ensure_schema():

import aiosqlite
from engrava import SqliteEngravaCore

async with aiosqlite.connect("my.db") as db:
    store = SqliteEngravaCore(db, manifests=[manifest])
    await store.ensure_schema()
    # migrations are now applied

Or use the opt-in discovery helper to load all installed extensions:

from engrava import SqliteEngravaCore
from engrava.extensions.discovery import discover_manifests

store = SqliteEngravaCore(db, manifests=discover_manifests())
await store.ensure_schema()

Note: Discovery is never automatic — always opt in explicitly. Schema migrations have side-effects (ALTER TABLE, CREATE TABLE) and should only run when the caller is aware of them.

YAML Configuration

Manifests can also be declared in engrava.yaml:

# Explicit dotted paths
manifests:
  - "my_plugin.manifest:MANIFEST"

# Auto-discover via entry points
manifests:
  discover: true

# Both
manifests:
  discover: true
  paths:
    - "my_plugin.manifest:MANIFEST"

Version Tracking

The runner keeps two tables. extension_schema_migrations is the append-only history: one row per applied migration, recording its 1-based index, filename and a content checksum. extension_schema_versions is a single-row-per-extension summary — extension name, count of applied migrations, timestamp, last applied filename and the extension version at apply time.

History is the authority. Pending work is derived from the history table, and before anything is applied the runner re-verifies every recorded migration against the manifest. A renamed, reordered or inserted earlier file, an edited one (checksum mismatch), a duplicate basename, or a summary that disagrees with the history all raise ExtensionMigrationError rather than being applied over. Applied migrations are immutable: add a new file instead of editing an old one.

Runner behavior at startup:

On SQL failure the version counter is not advanced. ExtensionMigrationError is raised with the extension name and failing filename so the caller can surface a clear error message.

Migration Path Resolution

Relative paths in schema_migrations are resolved in this order:

1. Absolute path — used as-is (CI / developer override).
2. manifest.package_root is set — joined with package_root (useful for test fixtures or non-installable manifests).
3. Default — resolved via importlib.resources.files against the top-level package that contains hooks_class. Works correctly for installed wheels, editable installs, and zipapps.

from pathlib import Path
from engrava import ExtensionManifest

# Default (importlib.resources -- recommended for distributed packages)
manifest = ExtensionManifest(
    name="my-plugin",
    version="1.0.0",
    hooks_class=MyHooks,
    schema_migrations=[Path("migrations/001_initial.sql")],
)

# Absolute path (CI / local dev)
manifest = ExtensionManifest(
    name="my-plugin",
    version="1.0.0",
    hooks_class=MyHooks,
    schema_migrations=[Path("/abs/path/to/001_initial.sql")],
)

# package_root override (test fixtures)
manifest = ExtensionManifest(
    name="my-plugin",
    version="1.0.0",
    hooks_class=MyHooks,
    schema_migrations=[Path("migrations/001_initial.sql")],
    package_root=Path(__file__).parent,
)

Subclassing SqliteEngravaCore

For deeper customization, subclass SqliteEngravaCore and override the template methods:

from engrava import SqliteEngravaCore, ThoughtRecord

class ExtendedStore(SqliteEngravaCore):
    def _row_to_thought(self, row: dict) -> ThoughtRecord:
        """Override to produce a richer model type."""
        return super()._row_to_thought(row)

This is the recommended pattern for adding domain-specific fields to the thought model without forking the core.

Contract Testing

Add a contract test to verify your implementation satisfies the protocol:

from engrava.domain.protocols.hooks import EngravaHooksProtocol

def test_my_hooks_satisfy_protocol() -> None:
    assert isinstance(RecencyBoostHooks(), EngravaHooksProtocol)

EngravaHooksProtocol is @runtime_checkable, so isinstance works without metaclass magic.