Integration Guide — How to put Cairntir behind your own tool

This guide is for developers embedding Cairntir as a library inside another application. If you are using Cairntir as a Claude Code plugin, you want how-to-use.md instead.

The 30-second version

from cairntir import Drawer, Layer, Store
from cairntir.impl import DrawerStore, HashEmbeddingProvider

# Open a store. Any path; any embedder that satisfies EmbeddingProvider.
store: Store = DrawerStore(
    "/path/to/cairntir.db",
    HashEmbeddingProvider(dimension=384),
)

# Write a verbatim drawer.
saved = store.add(Drawer(
    wing="my-app",
    room="decisions",
    content="we picked postgres over sqlite for the live tier",
    layer=Layer.ESSENTIAL,
    claim="postgres scales past 10M rows for our workload",
    predicted_outcome="p99 read latency stays under 50ms at target volume",
))

# Search.
hits = store.search("which database did we pick", wing="my-app", limit=5)
for drawer, distance in hits:
    print(f"{distance:.3f}  {drawer.content}")

store.close()

That’s it. Everything else is variations on this theme.

The three surfaces you’ll touch

1. The protocol surface (cairntir.*)

The stable seam. Import from cairntir itself — never from submodules — if you want your code to survive future releases without refactors. Every name in cairntir.__all__ is covered by the deprecation policy.

The core protocols you’ll reference:

2. The concrete namespace (cairntir.impl.*)

The default implementations. These are supported but not covered by the deprecation policy — Cairntir reserves the right to rename DrawerStore to SQLiteVecDrawerStore in a minor release if that makes the library better. If you import from cairntir.impl.*, you are opting into upgrade friction in exchange for convenience.

Rule of thumb: use cairntir.impl.* to construct things, use cairntir.* to type-annotate things.

from cairntir import Store
from cairntir.impl import DrawerStore, HashEmbeddingProvider

def build_store(path: str) -> Store:  # Store is the protocol
    return DrawerStore(path, HashEmbeddingProvider())  # impl is concrete

3. The portable format (cairntir.portable)

For moving drawers between stores (or between machines, or offline, or through gossip). Covered by the policy at the envelope-format level: the JSON shape of an envelope is stable within a major version. See how-to-use.md for CLI use.

Implementing a custom Store

Store is a runtime_checkable Protocol. No inheritance required — any object that provides the methods will pass isinstance(x, Store).

from datetime import datetime
from cairntir import Drawer, Layer, Store, MemoryStoreError

class MyRedisStore:
    """Toy example. Not production-ready."""

    def __init__(self, redis_client: object) -> None:
        self._r = redis_client
        self._next_id = 1

    def add(self, drawer: Drawer) -> Drawer:
        drawer_id = self._next_id
        self._next_id += 1
        # ... persist drawer.model_dump() under drawer_id ...
        return drawer.model_copy(update={"id": drawer_id})

    def get(self, drawer_id: int) -> Drawer | None:
        # ... fetch by id, return None if missing ...
        ...

    def list_by(self, *, wing=None, room=None, layer=None, limit=100):
        ...

    def search(self, query, *, wing=None, room=None, limit=10, rerank_by_belief=True):
        ...

    def update_layer(self, drawer_id: int, layer: Layer) -> None:
        ...

    def reinforce(self, drawer_id: int, *, amount: float = 1.0) -> float:
        ...

    def weaken(self, drawer_id: int, *, amount: float = 1.0) -> float:
        ...

    def stale_ids(self, *, older_than: datetime, layer: Layer, wing=None):
        ...

    def close(self) -> None:
        ...

Run the contract test suite against your impl

The v1.0 promise is that every Store impl passes tests/contract/test_store_contract.py. To run the suite against your backend, add a factory to the parametrized list:

# In your own test file:
from cairntir import Store
from my_package import MyRedisStore

def _my_store_factory(tmp_path):
    return MyRedisStore(redis_client=start_ephemeral_redis(tmp_path))

# Then parametrize the same fixture shape the suite uses.

Every failing invariant is a bug in your backend, not a bug in the suite. The invariants are the minimum the library assumes; breaking them will cause higher-level code (consolidate_room, ReasonLoop, the reranker) to misbehave in subtle ways.

Implementing a custom EmbeddingProvider

from collections.abc import Sequence
from cairntir import EmbeddingProvider

class MyOpenAIEmbeddings:
    @property
    def dimension(self) -> int:
        return 1536  # text-embedding-3-small

    def embed(self, texts: Sequence[str]) -> list[list[float]]:
        # ... call OpenAI, return one vector per input ...
        ...

Pass an instance to DrawerStore:

store = DrawerStore("/path/to/db", MyOpenAIEmbeddings())

All the existing contract and property tests will run against your embedder through the store.

Wiring the Reason loop

ReasonLoop is under cairntir.impl because it’s a concrete orchestration. The four ports it takes (HypothesisProposer, ExperimentRunner, BeliefStore, MemoryGateway) are protocols in cairntir.*.

The shipped wiring (stdlib-only, zero network)

Since v1.1 Cairntir ships four concrete adapters in cairntir.production that cover every port. All stdlib-only — no API keys, no paid tokens, no tracked telemetry:

from cairntir.impl import ReasonLoop
from cairntir.production import (
    ManualProposer,
    NullRunner,
    StoreBackedBeliefs,
    StoreBackedMemory,
)

# ``store`` is any cairntir.Store (DrawerStore is the default impl).
loop = ReasonLoop(
    proposer=ManualProposer(
        claim="rate-limiting reduces the p99 by more than 200ms",
        predicted_outcome="p99 drops under 400ms after deploy",
    ),
    runner=NullRunner(observed="p99 dropped to 380ms", success=True),
    beliefs=StoreBackedBeliefs(store=store),
    memory=StoreBackedMemory(store=store),
)
update = loop.step(
    question="should we rate-limit this endpoint?",
    wing="my-app",
    room="decisions",
)
print(update.mass_change, update.delta)

Rolling your own proposer — e.g. a local Gemma

Cairntir does not do inference. If you want a model to propose the hypothesis, implement HypothesisProposer in your own code. A local Gemma 4 (via llama.cpp or Ollama) is the planned path for Cairntir itself; here’s the shape:

from cairntir import Hypothesis, HypothesisProposer


class LocalGemmaProposer:
    """Example: talk to a local Gemma 4 instance via Ollama.

    The loop never knows this exists — it just sees the protocol.
    Zero API spend, zero telemetry, fully local.
    """

    def __init__(self, *, model: str = "gemma3:4b", host: str = "http://localhost:11434") -> None:
        self._model = model
        self._host = host

    def propose(self, *, question: str, wing: str, room: str) -> Hypothesis:
        # ... POST to {host}/api/chat, parse the response into a claim
        # and predicted_outcome, return a Hypothesis ...
        ...

Drop it in as the proposer= argument to ReasonLoop — nothing else changes.

The loop writes two drawers per step(): a prediction drawer up front and an observation drawer that supersedes_ids it after the runner reports back. Everything is verbatim, nothing is summarized.

Migrations

When a new Cairntir minor release bumps the on-disk schema, the migration runs automatically the first time you open the store:

store = DrawerStore("/path/to/existing.db", HashEmbeddingProvider())
# Migration ran here. PRAGMA user_version is now the current SCHEMA_VERSION.

To apply migrations deliberately (e.g. in a deploy script) without going through DrawerStore, use the CLI:

cairntir migrate /path/to/existing.db
cairntir migrate --check /path/to/existing.db  # dry run, reports version only

Migrations are always forward-only. Downgrading to an older library version against an already-migrated database is not supported — the contract is that every minor release reads every prior schema, not that every schema reads every library version.

What you should not do

Where to go from here