Engram reference.

# Engram ships inside the base cosmonapse package  -  InMemoryEngram,
# SqliteEngram and PostgresEngram all included.
$ pip install cosmonapse

# Everything public is re-exported from cosmonapse.engram.
from cosmonapse.engram import (
    # Core types
    Engram,            # ABC every backend implements
    EngramBinding,     # declarative wiring stored on an Axon
    EngramClient,      # caller-side correlation table (one per Dendrite)
    Hit,               # one search result
    RecallResult,      # what recall() returns
    ImprintReceipt,    # what imprint() returns

    # Backends
    InMemoryEngram,    # dict-backed; default for dev/tests
    SqliteEngram,      # stdlib sqlite3, single-file DB
    PostgresEngram,    # asyncpg pool; for real deployments

    # Errors
    EngramError,
    EngramTimeout,
    EngramCancelled,
    EngramNotBound,
    EngramOverloaded,
)

# eng_… ULID helper lives on the package root
from cosmonapse import new_engram_id

from cosmonapse import Axon
from cosmonapse.engram import EngramBinding

axon = Axon(
    neuron_id="summariser",
    neuron_fn=summariser,
    engrams=[
        # Addressed routing  -  Neuron says recall("notes", ...)
        EngramBinding(name="notes", directed_id="notes-default"),
        # Slot routing  -  deployment owns the concrete vector store
        EngramBinding(
            name="memory",
            directed_type="semantic",
            default_deadline_ms=250,
            default_recall_mode="merge",
        ),
    ],
)

# Injected into the Neuron when neuron_fn declares the kwargs.
async def summariser(input, context, *, recall, imprint):
    # READ  -  returns a RecallResult (iterable of Hit)
    prior = await recall(
        "notes",                       # binding name (must be wired on the Axon)
        query={"text": input["topic"]},
        filters={"tags": ["kept"]},
        deadline_ms=200,
        recall_mode="first",
        min_confidence=0.5,
    )

    note = f"summary of {input['topic']} ({len(prior)} priors)"

    # WRITE  -  fire-and-forget by default (await_ack=False)
    await imprint(
        "notes",
        op="append",                    # add | append | merge | upsert | delete
        entry={"content": note, "tags": ["kept"]},
        merge_key=input["topic"],
    )

    # WRITE + receipt  -  await the IMPRINTED ack
    receipt = await imprint(
        "notes", op="upsert", entry={"content": note},
        merge_key=input["topic"], await_ack=True, deadline_ms=500,
    )
    assert receipt.ok

    return {"summary": note, "prior": [h.entry for h in prior]}

from cosmonapse.engram import Engram, Hit, ImprintReceipt

class RedisEngram(Engram):
    def __init__(self, url: str):
        self.engram_id   = "redis-default"
        self.engram_kind = "keyvalue"
        self.capabilities = ["substring", "tags"]
        self._url = url

    async def connect(self): self._r = await redis.from_url(self._url)
    async def close(self):   await self._r.aclose()

    async def recall(self, query, **kw) -> list[Hit]:
        raw = await self._r.get(query["key"])
        return [Hit(id=query["key"], entry={"value": raw})] if raw else []

    async def imprint(self, op, entry, **kw) -> ImprintReceipt:
        await self._r.set(entry["key"], entry["value"])
        return ImprintReceipt(engram_id=self.engram_id, op=op, id=entry["key"])

@dataclass(frozen=True)
class Hit:
    id:    str
    entry: dict
    score: float = 1.0      # cosine in [0,1] for semantic; 1.0 for relational

@dataclass(frozen=True)
class RecallResult:
    hits:       list[Hit] = []
    engram_ids: tuple[str, ...] = ()
    truncated:  bool = False
    took_ms:    int | None = None
    # Iterable + truthy: 'for h in result', 'len(result)', 'if result:'

@dataclass(frozen=True)
class ImprintReceipt:
    engram_id: str
    op:        str
    id:        str | None = None
    version:   int | None = None
    took_ms:   int | None = None
    error:     str | None = None

    @property
    def ok(self) -> bool:
        return self.error is None

from cosmonapse.engram import InMemoryEngram, SqliteEngram, PostgresEngram

# Dict-backed. Zero deps. Resets on process exit. Use in tests.
mem = InMemoryEngram(
    engram_id="engram-memory",
    engram_kind="keyvalue",
    capabilities=["substring", "tags", "merge_key"],
)

# Single-file sqlite3 via threadpool. Survives restarts.
sql = SqliteEngram(
    path="./memory.db",            # ":memory:" by default
    engram_id="engram-sqlite",
    engram_kind="relational",
)

# asyncpg pool. Production. Driver is lazy-imported.
pg = PostgresEngram(
    dsn="postgresql://user:pw@localhost/cosmo",
    engram_id="engram-postgres",
    engram_kind="relational",
    min_size=1, max_size=5,
)

await sql.connect()   # open resources before attaching

from cosmonapse import Dendrite
from cosmonapse.engram import SqliteEngram
from cosmonapse.synapse.memory import InMemorySynapse

synapse = InMemorySynapse()

# A Dendrite that *hosts* the Engram and services RECALL / IMPRINT.
host = Dendrite(synapse=synapse)

notes = SqliteEngram(path="notes.db", engram_id="notes-default")
await notes.connect()

host.attach_engram(notes)        # index by engram_id + engram_kind
await host.start()

# Inspect / tear down
host.engrams                     # {"notes-default": notes}
await host.detach_engram("notes-default")   # closes backend, unsubscribes

// RECALL  -  emitted by a hosting Dendrite on the Neuron's behalf
{
  "type": "RECALL",
  "trace_id": "trc_01JV…",   // inherited from the containing TASK
  "parent_id": "evt_01JV…",
  "payload": {
    "engram_id":   "pgvector-default",  // OR engram_kind  -  id wins
    "engram_kind": "semantic",
    "query":       { "text": "eviction cause" },
    "filters":     { "tags": ["k8s"] },
    "deadline_ms": 250,
    "recall_mode": "first"      // "first" | "merge" | "all"
  }
}

// RECALLED  -  one per responding Engram; parent_id → the RECALL
{
  "type": "RECALLED",
  "payload": {
    "engram_id": "pgvector-default",
    "hits": [
      { "id": "eng_01JV…", "score": 0.91, "entry": { } },
      { "id": "eng_01JV…", "score": 0.74, "entry": { } }
    ],
    "truncated": false,
    "took_ms":   38
  }
}

// IMPRINT  -  addressed write (broadcast is opt-in via meta.broadcast)
{
  "type": "IMPRINT",
  "payload": {
    "engram_id":   "ctx-default",
    "op":          "append",   // add|append|merge|upsert|delete
    "entry": {
      "id":      "eng_01JV…",
      "content": "Eviction triggered by memory pressure.",
      "tags":    ["k8s", "eviction"]
    },
    "merge_key": "incident:42"   // required for merge / upsert
  }
}

// IMPRINTED  -  ack; parent_id → the IMPRINT
{
  "type": "IMPRINTED",
  "payload": {
    "engram_id": "ctx-default", "op": "append",
    "id": "eng_01JV…", "version": 3, "took_ms": 12
  }
}

from cosmonapse import new_engram_id

new_engram_id()   # 'eng_01JVZ8K3M2…'  -  sortable ULID, eng_ prefix