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.
from cairntir import Drawer, Layer, Store
from cairntir.impl import DrawerStore, HashEmbeddingProvider
# HashEmbeddingProvider is a test embedder — any dimension, no model
# download. Production is `production_embedding_provider()` (jina,
# 512-dim, 8192-token window). Do not copy 384 from an old MiniLM note.
store: Store = DrawerStore(
"/path/to/cairntir.db",
HashEmbeddingProvider(dimension=32),
)
# 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.
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:
Store — the memory backend. Implement this to plug a custom
backend (Redis, Postgres, a cloud-hosted vector DB) into the rest
of Cairntir.EmbeddingProvider — vector embeddings. Implement this to swap
the default sentence-transformers model for whatever you already
have running.HypothesisProposer, ExperimentRunner,
BeliefStore, MemoryGateway) — the four seams of the library
Reason loop. Implement these for tool-driven reasoning without
touching the default memory layer.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
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.
StoreStore 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:
...
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.
EmbeddingProviderfrom 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.
ReasonLoop is under cairntir.impl because it’s a concrete
orchestration. The four ports it takes (HypothesisProposer,
ExperimentRunner, BeliefStore, MemoryGateway) are protocols in
cairntir.*.
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,
delta="the fallback limiter, not the primary limiter, handled the load",
),
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)
Verdict and surprise are separate. A prediction can hold while the route differs;
pass that path-level evidence as delta instead of discarding it. The loop rejects
blank commitments, scope changes, outcomes bound to another hypothesis, incomplete
experiment records, and idempotency claims made through a non-durable gateway.
Automatic reflection counts only uniquely bound Reason prediction/observation pairs. It keeps repeated claims separated by room and may propose a reviewable candidate, but it never corroborates or promotes one.
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.
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.
cairntir.memory.* directly. It is still the
canonical source for the concrete impls, but the stable import
paths are cairntir.* (protocols) and cairntir.impl.* (concrete).Exception around Cairntir calls. Every error
Cairntir raises is a subclass of CairntirError. Catch that (or a
more specific subclass) and let everything else propagate.Drawer instances. They are frozen pydantic
models. Use drawer.model_copy(update={...}) if you need a
modified copy.id values across stores. A drawer’s id is
local to the store that assigned it. If you export and re-import,
the new store will assign fresh ids. Use content_hash from the
portable format for cross-store identity.tests/contract/test_store_contract.py — the invariants your
custom Store must satisfy