---
contract_version: "2.1"
module: "ratatoskr.provider.memory_store"
purpose: "Memory-plane Bifrost consumer (v1 basic plane): a SQLite+sqlite-vec-backed durable memory store Worldtree persists Tier-3 agent memory chunks into and recalls via vector search."
touches:
  - src/ratatoskr/provider/memory_store.py
  - tests/test_provider_memory.py
language: "python"
complexity: "high"
estimated_loc: 320
confidence: 0.82
assumptions:
  - "bifrost>=0.6.1 exposes build_memory_app, dispatch_memory_call, JwtVerifier, ConsumerRegistration, StoreCapabilities, MemoryDataStore, InvalidArguments, IdempotencyConflict, RevisionMismatch per bifrost/reference_server/memory.py + bifrost.memory."
  - "v1 = worldtree-dev's BASIC PLANE only (search / get / get_many / upsert_many / delete_many + describe_store + health), the ONLY surface Tier-3's live path touches (#294); Worldtree v0.35.3 already negotiates it."
  - "Chunk record field names are taken from the reference store (named inline below) but the AUTHORITATIVE pin is TDD against InMemoryMemoryStore, as it was for affect."
  - "Embedding dimension matches Worldtree's PINNED_EMBEDDER_DIM, supplied as config (env RATATOSKR_MEMORY_EMBEDDING_DIM); the sqlite-vec virtual table is created at that fixed dim."
open_questions:
  - "Whether the memory DB shares one SQLite file with affect or its own — default SEPARATE per plane; the dev-shell entrypoint reads RATATOSKR_MEMORY_DB (analogous to RATATOSKR_AFFECT_DB). Revisit at the combined two-plane server (guide §7)."
external_invariants:
  - source: ~/development/bifrost/bifrost/reference_server/memory.py
    invariant_id: "InMemoryMemoryStore"
  - source: ~/development/bifrost/docs/implementing-a-consumer.md
    invariant_id: "§5 memory plane"
revisions:
  - version: "1.2"
    at: 2026-06-16
    summary: "Repin bifrost 0.7.0→0.8.0 (wire v0.5→v0.6, #11): search `scope_filter` split into `scope_all` (AND/intersection) + `scope_any` (OR/union over a list of conjunctive scopes). No-compat: `scope_filter` removed. Adds union-visibility recall in one call — the fix for the #295/#297 AND silent-zero foot-gun. Store at parity with the v0.6 reference `_matches_scope` / `_validate_scope`."
    delta:
      MODIFIED:
        - "search signature: scope_filter -> scope_all + scope_any"
        - "INV-005 scope isolation -> composed v0.6 (scope_all AND ∧ scope_any OR-union)"
        - "PRE-003 validates axes in BOTH fields; STEP 1 = _validate_scope (shape + lattice)"
      ADDED:
        - "scope_any_union test (#295/#297 union capability); both-fields-empty match-all"
  - version: "1.1"
    at: 2026-06-15
    summary: "Heid-contract-review fixup: semantic-not-byte-equal round-trip; reconcile idempotency 4-tuple; search returns top_k IN-SCOPE; define recalled_view + scope_filter + named field keys inline; clarify metadata_filter-v1 + transaction-term + delete atomicity + get_many + revision-on-replay; drop scan from INV-005."
    delta:
      MODIFIED:
        - "INV-001 byte-equal -> semantic round-trip; named structural field keys inline"
        - "INV-002 idempotency_id = (\"default\", verb, actor, key) — reconciled with STEPS"
        - "INV-005 search only (scan was deferred)"
        - "search: top_k in-scope, scope_filter shape, recalled_view, metadata_filter-v1 reject"
        - "delete_many atomicity; get_many clarified; transaction-term clarified"
---

## Context

The second plane of ratatoskr's Tier-3 Bifrost consumer (after the shipped affect
plane). A **SQLite + sqlite-vec** durable store Worldtree writes agent **memory
chunks** into (`upsert_many`) and recalls from by **vector similarity**
(`search`), plus point reads (`get`/`get_many`) and deletes (`delete_many`).
v1 is worldtree-dev's **basic plane** — the only surface Tier-3's live path uses;
the gated verbs (edges, scan, atomic_supersede, mark_*, patch, maintenance) are
deferred. We implement **bifrost's own** `MemoryDataStore` Protocol and hand it to
`build_memory_app`. Conformance is #195 parity vs `InMemoryMemoryStore`.

The boundary (ADR-0001/0002/0009): **Worldtree owns intelligence — appraisal,
consolidation, trust; we own permanence.** But unlike affect (blind conduit),
memory is a **structural index**: we read a few fields of each chunk —
`record["embedding"]` (rank), `record["scope"]` (isolation), `record["id"]` +
revision (optimistic locking), and `origin`/`injection_source` (the consistency
rule). The semantic content, `record["distillate"]`, and inert fields
(`trust_tier`/`provenance`/`source_role`) are persisted verbatim and never
interpreted.

## Data flow

- **In:** Worldtree → `POST /bifrost/memory-call` → library validates envelope +
  per-dispatch JWT → the verb on our store.
- **Chunk record (key fields we read; rest is opaque payload):** `id` (the chunk
  id — reference falls back to `chunk_id`/`memory_id`), `embedding` (the vector —
  fallback `vector`), `scope` (a `{axis: value}` dict — the isolation key),
  `origin` + `injection_source` (consistency rule), `distillate` (the recall
  view). Everything else (content, `metadata`, `trust_tier`, …) is stored verbatim.
- **At rest:** SQLite —
  - `memory_chunks(chunk_id PK, record_json, revision, scope_json, origin, ...)` —
    the verbatim chunk + extracted columns (chunk_id, scope) for isolation.
  - sqlite-vec virtual table `memory_vec(chunk_id, embedding[<dim>])` — the index.
  - `memory_idempotency(idempotency_id PK, digest, expires_at)` — replay/conflict
    cache (affect-parallel shape).
- **Out:** `upsert_many` → `{"upserted": N, "replayed": bool}`; `search` → list of
  `{chunk, chunk_id, score, recalled_view, revision}` where **`recalled_view`** =
  the chunk's `distillate` field, or the whole chunk if absent (per the reference);
  `delete_many` → `{"deleted": N}`; `get` → the verbatim record + a `revision` key,
  or `None`; `get_many(ids)` → the list form of `get` (found records only).

## Invariants

- **INV-001** [hard]: **Persist verbatim (semantic round-trip); read only the
  structural surface.** The whole chunk is stored and a read deserializes to a
  Python object EQUAL to the input (`json.loads(record_json) == input`) — **not**
  byte-equal (key order / formatting may differ); `get` additionally attaches a
  `revision` key to the returned object. The store reads ONLY `record["embedding"]`,
  `record["scope"]`, `record["id"]` + revision, and `origin`/`injection_source`.
  Content / `distillate` / inert fields (`trust_tier`/`provenance`/`source_role`)
  are NOT interpreted.
- **INV-002** [hard]: **Idempotency = replay-or-conflict, actor-scoped** (affect-
  parallel). `idempotency_id = ("default", <verb>, _ctx_actor(ctx), idempotency_key)`
  — the literal `"default"` class slot + the verb name, matching the reference
  4-tuple (`idempotency_class` tunes only the cache TTL, not the id). Same digest →
  replay (`replayed: True`, no re-write); different digest → raise
  `IdempotencyConflict`. Actor from `ctx`, never from the record.
- **INV-003** [hard]: **Optimistic locking.** When `upsert_many` carries
  `expected_revisions`, each record's stored revision must equal the expected; any
  mismatch → raise `RevisionMismatch` and the whole batch rolls back. Each
  successful upsert increments the chunk's revision (a first insert → revision 1).
- **INV-004** [hard]: **Atomic batch.** `upsert_many` applies all records + their
  vec rows + the idempotency record in one transaction; on any error nothing is
  persisted (no partial batch, no orphaned vec rows).
- **INV-005** [hard]: **Scope isolation (wire v0.6, #11).** `search` filters by two
  explicit fields: `scope_all` (AND/intersection — record ⊇ every named axis) and
  `scope_any` (OR/union over a LIST of conjunctive scope dicts — record ⊇ ≥1 element,
  each element AND-matched as a whole). They compose by AND; both empty → no scope
  constraint. A search never returns a chunk outside the composed filter. Byte-faithful
  to the reference `_matches_scope`. (`scope_any` is the union-visibility primitive that
  resolves the #295/#297 silent-zero — a subset-scoped chunk now recalls via an OR member.)
- **INV-006** [hard]: **Capabilities match implementation** (advertise-⇒-implement).
  `describe_store` advertises ONLY what v1 implements: `relational_edges_supported=False`,
  `atomic_supersede_supported=False`, `transaction_supported=False`,
  `optimistic_locking_supported=True`, `filterable_metadata_fields=[]`.
  (`transaction_supported` is the bifrost **wire-level** multi-op transaction
  capability — NOT our internal SQLite transactions, which we use for atomic
  batches.) The client gates the gated verbs off these.
- **INV-007** [hard]: `origin == "injected_context"` requires `injection_source`; a
  non-injected record carrying `injection_source` is rejected — both raise
  `InvalidArguments` (mirrors the reference).
- **INV-008** [hard]: The store is REQUIRED (`build_memory_app(store=None)` raises);
  identity/scope/actor come from `ctx`, never call args.

## Concurrency

SQLite WAL (concurrent readers, single writer). `upsert_many`/`delete_many`
serialize on the writer; `search`/`get` are concurrent reads. sqlite-vec index
writes ride inside the upsert/delete transaction. The connection is opened
`check_same_thread=False` with `PRAGMA busy_timeout=5000` (mirrors the affect store):
the provider is an ASGI app, so uvicorn/Starlette (and TestClient always) may run a
handler off the connection's creating thread — the event loop serializes the sync
sqlite calls, so this is safe; busy_timeout preps the composite/standalone two-process
topology over the same db. (Surfaced by a TestClient-driven memory `search` through the
#18 D1 combined provider — the direct-store tests structurally could not.)

## Division of labor (library vs store)

The **bifrost library** owns the wire (envelope validation, per-dispatch JWT,
scope authorization, error mapping of our typed exceptions, capability
negotiation, routes). **This contract** owns the store (the basic verbs + SQLite
+ sqlite-vec persistence/index) + the thin `build_memory_provider_app` wiring.

## Integration points

- `bifrost.consumer.build_memory_app(store, verifier, registration, maintenance_store=None, hooks=None)` → Starlette app.
- `bifrost.reference_server.JwtVerifier` + `bifrost.consumer.ConsumerRegistration`.
- `bifrost.memory.{StoreCapabilities, InvalidArguments, IdempotencyConflict, RevisionMismatch}` — typed surface.
- **Conformance (tests):** `bifrost.consumer.testing.InMemoryMemoryStore` + `bifrost.memory.dispatch_memory_call` (#195). The reference is the authoritative pin for exact field names + wire shapes.
- `sqlite-vec` — the vector index extension loaded into the connection.

## Constraints

- **[security]** Never log chunk content / `distillate`. Index the vector + scope; don't interpret semantics.
- **[compatibility]** Implement bifrost's MemoryDataStore shape exactly; raise its typed exceptions; never fork the wire. Gated verbs are absent + advertised unsupported.
- **[correctness]** `search` ranks by cosine over `record["embedding"]`; scope isolation (INV-005) is non-negotiable; `top_k` counts IN-SCOPE results (see search STEPS).

## Out of scope (deferred — do NOT flag as drift)

- **Gated/maintenance verbs:** `upsert_edges`/`get_edges_for`, `scan`, `mark_invalid`/`mark_superseded`, `patch_many`, `atomic_supersede`, lease/checkpoint. Absent + advertised-unsupported.
- **metadata_filter beyond scope:** advertise `filterable_metadata_fields=[]`; a non-empty `metadata_filter` is unsupported in v1 (rejected — see search PRE).
- **The combined two-plane server** (guide §7) — separate memory + affect apps in v1.
- **Deployment** — dev-box background shell (`ratatoskr-memory-provider`), no systemd/infra.
- **idempotency-cache TTL pruning** — `expires_at` recorded, eviction deferred (affect-parallel).

```contract
FN open_memory_store(db_path: str, *, embedding_dim: int) -> RatatoskrMemoryStore
BRIEF: Open the SQLite+sqlite-vec memory store, creating schema + the vec index on first use.
PRE: [PRE-001 hard] db_path writable or ":memory:" -- guard
PRE: [PRE-002 hard] embedding_dim is a positive int (matches Worldtree PINNED_EMBEDDER_DIM) -- assert
POST: [POST-001 return_value] store.describe_store() advertises the v1 capability set (INV-006) -- assert
POST: [POST-002 state_change] memory_chunks + memory_vec(dim) + memory_idempotency exist -- schema present
STEPS:
  1. [setup] CONNECT sqlite3; enable_load_extension; LOAD sqlite-vec; WAL (skip for ":memory:")
  2. [sequential, flexibility=prescriptive] CREATE memory_chunks + memory_idempotency tables IF NOT EXISTS
  3. [sequential, flexibility=prescriptive] CREATE VIRTUAL TABLE memory_vec USING vec0(chunk_id TEXT PRIMARY KEY, embedding float[embedding_dim]) IF NOT EXISTS
  4. [cleanup] RETURN RatatoskrMemoryStore(conn, embedding_dim)
TESTS:
  fresh_db [happy,tracer]: open ":memory:" dim=8 → describe_store() has the v1 caps; tables queryable
  reopen [happy]: open existing file twice → idempotent schema
```

```contract
FN describe_store(self) -> dict
BRIEF: Static capability descriptor (sync, no I/O).
POST: [POST-001 return_value] returns the bifrost StoreCapabilities dict with v1 values (INV-006) -- assert relational_edges/atomic_supersede/transaction False, optimistic_locking True, filterable_metadata_fields []
STEPS:
  1. [cleanup] RETURN StoreCapabilities(relational_edges_supported=False, optimistic_locking_supported=True, atomic_supersede_supported=False, transaction_supported=False, filterable_metadata_fields=[]).to_dict()
TESTS:
  caps [happy]: returns exactly the v1 capability dict; advertise-⇒-implement holds
```

```contract
FN upsert_many(self, records: list[dict], *, idempotency_key: str, ctx, expected_revisions: dict | None = None, idempotency_class: str | None = None) -> dict
BRIEF: Persist chunks verbatim + index their vectors, atomically, replay-or-conflict idempotent, optimistic-locked.
PRE: [PRE-001 hard] idempotency_key non-empty str -- else InvalidArguments
PRE: [PRE-002 hard] each injected_context record has injection_source; non-injected has none -- else InvalidArguments (INV-007)
POST: [POST-001 return_value] {"upserted": len(records), "replayed": False} on persist; {"...","replayed": True} on replay (INV-002) -- assert
POST: [POST-002 state_change] each chunk stored verbatim + vector indexed + revision incremented (first insert → 1); expected_revisions enforced (INV-003) -- assert
POST: [POST-003 side_effect] on ANY error, nothing persisted (INV-004) -- rollback
ERROR_ROUTING:
  InvalidArguments: { local_handling: raise on bad key / injection_source rule, flow_control: abort, state_recovery: none }
  IdempotencyConflict: { local_handling: raise on key-reuse-different-digest, flow_control: abort, state_recovery: none }
  RevisionMismatch: { local_handling: raise on stale expected_revision, flow_control: abort, state_recovery: full batch rollback }
STEPS:
  1. [setup] validate idempotency_key; digest over {records, expected_revisions}; idempotency_id = ("default", "upsert_many", _ctx_actor(ctx), idempotency_key)  -- matches INV-002
  2. [branch] idempotency lookup: same digest → RETURN replayed; different → RAISE IdempotencyConflict
  3. [sequential, flexibility=prescriptive] BEGIN; IF expected_revisions: assert each stored revision matches else RAISE RevisionMismatch
  4. [loop] FOR each record: validate origin/injection_source; UPSERT memory_chunks (record_json + scope_json, revision+1); UPSERT memory_vec(record["id"], record["embedding"])
  5. [sequential] record idempotency (digest, expires_at = now + ttl(idempotency_class)); COMMIT
  6. [cleanup] RETURN {"upserted": len(records), "replayed": False}
TESTS:
  basic_upsert [happy,tracer]: 2 records → {"upserted":2,"replayed":False}; get() round-trips each verbatim + revision=1
  replay [happy]: same key+payload twice → first writes (revision 1), second {"replayed":True} with NO further write (revision stays 1)
  conflict [adversarial]: same key, different records → IdempotencyConflict; first batch intact
  optimistic_lock [adversarial]: expected_revisions stale → RevisionMismatch; nothing written
  injection_rule [adversarial]: injected_context w/o injection_source → InvalidArguments; no write
  parity_vs_reference [scenario]: same upsert_many envelopes through dispatch_memory_call vs InMemoryMemoryStore → wire bodies agree (#195)
```

```contract
FN search(self, vector: list[float], *, top_k: int, scope_all: dict | None = None, scope_any: list | None = None, metadata_filter: dict | None = None, include: dict | None = None, fidelity_target=None) -> list[dict]
BRIEF: Vector (cosine) recall over sqlite-vec, scoped by the v0.6 scope_all/scope_any filter, returning the top_k IN-SCOPE chunks.
PRE: [PRE-001 hard] len(vector) == embedding_dim -- else InvalidArguments
PRE: [PRE-002 hard] metadata_filter is empty/None -- v1 advertises no filterable fields; a non-empty filter → InvalidArguments
PRE: [PRE-003 hard] scope_all is a flat dict and scope_any a list of flat dicts (else InvalidArguments); every axis in BOTH ∈ {end_user, group, tenant, agent_self} -- else InvalidFilter (memory.invalid_filter 400); the bifrost wire-v0.6 lattice, matching the reference _validate_scope (#10 agent_self canonical, #11 scope split)
POST: [POST-001 return_value] returns the top_k highest-cosine records passing the composed v0.6 filter — `(scope_all empty OR record ⊇ scope_all) AND (scope_any empty OR record ⊇ ≥1 element)`; at most top_k, never fewer than min(top_k, in-scope count) (INV-005). Each: {chunk (verbatim), chunk_id, score, recalled_view (= chunk["distillate"] or chunk), revision} -- assert
STEPS:
  1. [setup] scope_all ← scope_all or {}; scope_any ← scope_any or []; validate via _validate_scope (flat-dict / list-of-dicts shape + every axis ∈ the v0.6 lattice, else InvalidArguments / InvalidFilter)
  2. [sequential, flexibility=indicative] rank candidates by cosine over record["embedding"]; KEEP only records passing _matches_scope(scope_all, scope_any) (INV-005); THEN take top_k — so top_k counts IN-SCOPE hits, not pre-filter hits (over-fetch from the vec index or post-filter rank as needed)
  3. [cleanup] RETURN result rows (chunk verbatim + score + recalled_view + revision)
TESTS:
  basic_search [happy,tracer]: upsert 3 scoped chunks, search → ranked by cosine, ≤ top_k, recalled_view present
  scope_isolation [adversarial]: two scopes, scope_all one → never returns the other's chunk, and returns top_k of the IN-SCOPE set even if out-of-scope chunks score higher (INV-005)
  scope_any_union [scenario]: scope_any=[{end_user:u},{agent_self:a}] recalls BOTH a subject-scoped and a self-scoped chunk in one call (#295/#297 union capability); scope_all+scope_any compose by AND
  empty [boundary]: search empty store → []; both fields empty → match all
  metadata_filter_rejected [adversarial]: non-empty metadata_filter → InvalidArguments
  lattice_axes [adversarial]: out-of-lattice axis in scope_all OR scope_any → InvalidFilter; non-list scope_any → InvalidArguments; agent_self admitted (wire v0.5, #10)
  parity_vs_reference [scenario]: identical search envelopes vs InMemoryMemoryStore → same ranked chunk_ids/shape (#195)
```

```contract
FN get(self, chunk_id: str) -> dict | None
BRIEF: Point read; returns the verbatim chunk + current revision, or None. get_many(ids) is the list form (found records only).
POST: [POST-001 return_value] stored record (verbatim, json.loads) + "revision" key, or None if absent (INV-001) -- assert
STEPS:
  1. [sequential] SELECT record_json, revision WHERE chunk_id; RETURN json.loads + revision, or None
TESTS:
  get_hit [happy]: after upsert → record equal + revision present
  get_absent [boundary]: unknown id → None
  get_many [happy]: get_many([present, absent]) → [present record] only
```

```contract
FN delete_many(self, ids: list[str]) -> dict
BRIEF: Delete chunks (+ their vec rows) by id, transactionally.
POST: [POST-001 return_value] {"deleted": N} where N = ids that existed -- assert
POST: [POST-002 state_change] in ONE transaction, deleted chunks gone from memory_chunks AND memory_vec; partial failure rolls back the whole batch (no orphan vec rows) -- assert
STEPS:
  1. [sequential, flexibility=prescriptive] BEGIN; FOR each id present: DELETE from memory_chunks + memory_vec; count; COMMIT
  2. [cleanup] RETURN {"deleted": count}
TESTS:
  delete_hit [happy]: delete 1 of 2 → {"deleted":1}; gone from chunks + vec; search won't surface it
  delete_absent [boundary]: unknown id → {"deleted":0}
```

```contract
FN build_memory_provider_app(store: RatatoskrMemoryStore, heimdall_key: bytes, consumer_id: str = "ratatoskr") -> Starlette
BRIEF: Wire JwtVerifier + registration; hand the store to bifrost's build_memory_app.
PRE: [PRE-001 hard] store.describe_store() returns a dict (advertises caps) -- assert (INV-008)
PRE: [PRE-002 hard] heimdall_key non-empty bytes -- assert
POST: [POST-001 return_value] Starlette app exposing POST /bifrost/handshake + POST /bifrost/memory-call -- assert routes
STEPS:
  1. [setup] verifier = JwtVerifier(HS256, heimdall_key); registration = ConsumerRegistration(consumer_id)
  2. [sequential, flexibility=prescriptive] app = build_memory_app(store=store, verifier=verifier, registration=registration)
  3. [cleanup] RETURN app
TESTS:
  builds_app [happy,tracer]: valid store + key → app with the two routes (incl. POST)
  bad_key [error]: empty heimdall_key → raises at construction
```