diff --git a/docs/contracts/bifrost_memory_provider.contract.md b/docs/contracts/bifrost_memory_provider.contract.md index 277110e..6eb2dd9 100644 --- a/docs/contracts/bifrost_memory_provider.contract.md +++ b/docs/contracts/bifrost_memory_provider.contract.md @@ -116,9 +116,13 @@ interpreted. 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`, + `describe_store` advertises ONLY what is implemented: `relational_edges_supported=False`, `atomic_supersede_supported=False`, `transaction_supported=False`, - `optimistic_locking_supported=True`, `filterable_metadata_fields=[]`. + `optimistic_locking_supported=True`, `filterable_metadata_fields=[]`, + **`sortable_chunk_fields=[{"name": "updated_at"}]`** (the ONLY globally-sortable field; + gates `scan`'s sort at the bifrost dispatch `_validate_scan_sort` AND Worldtree's #349 + person-prime Branch-A `"updated_at" in caps.sort_fields_supported` — advertising it is + what lights up turn-1 durable-fact injection). (`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. @@ -127,6 +131,20 @@ interpreted. `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. +- **INV-009** [hard]: **`scan` is LIVE-only.** `scan` returns ONLY live chunks — + superseded / tombstoned / any non-live governance state is EXCLUDED server-side. This + is load-bearing because Worldtree's person-prime requests `lifecycle_state="live"` but + that filter does NOT ride the scan wire today and the client does not re-check it + (worldtree-dev flagged the adapter gap); server-side live-only is authoritative, so a + dead fact can never inject. The additive `lifecycle_state` scan arg, when present, is + honored but never relied upon. +- **INV-010** [hard]: **`scan` is globally ordered before pagination.** The FULL + scope-filtered live set is ordered by `(sort.field, direction)` GLOBALLY before the + `limit` page is taken — never page-local. Missing sort value sorts LAST; ties broken by + `chunk_id` (stable). A single `limit`-page returns the N globally-newest (for + `updated_at desc`), matching bifrost's cross-pagination conformance negative. The sort + field is indexed (`json_extract(record_json, '$.updated_at')`) so the read stays within + person-prime's 500 ms fail-open budget. ## Concurrency @@ -163,7 +181,7 @@ negotiation, routes). **This contract** owns the store (the basic verbs + SQLite ## 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. +- **Gated/maintenance verbs:** `upsert_edges`/`get_edges_for`, `mark_invalid`/`mark_superseded`, `patch_many`, `atomic_supersede`, lease/checkpoint. Absent + advertised-unsupported. (`scan` is NO LONGER deferred — it is implemented + advertised via `sortable_chunk_fields` to light up Worldtree's #349 person-prime turn-1 durable-fact injection; see the `scan` FN spec + INV-009/INV-010.) - **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. @@ -270,6 +288,29 @@ TESTS: delete_absent [boundary]: unknown id → {"deleted":0} ``` +```contract +FN scan(self, *, scope_all: dict | None = None, scope_any: list | None = None, cursor: str | None = None, limit: int, sort: dict | None = None, lifecycle_state=None) -> dict +BRIEF: Query-LESS paginated LIVE-chunk scan, globally ordered by an advertised sort field (updated_at) — the #349 person-prime turn-1 durable-fact injection primitive (no query vector, unlike search). Returns {records, cursor}. +PRE: [PRE-001 hard] limit is a positive int -- else InvalidArguments +PRE: [PRE-002 hard] scope_all/scope_any shape + lattice-validated via _validate_scope (identical to search PRE-003) -- else InvalidArguments / InvalidFilter +PRE: [PRE-003 hard] sort, when present, is {field, direction}: field ∈ the advertised sortable_chunk_fields names ("updated_at"), direction ∈ {asc,desc}. The bifrost dispatch layer (_validate_scan_sort) is the enforcement gate; an unadvertised/malformed sort → InvalidArguments — NEVER a silent unsorted fallback +POST: [POST-001 return_value] {records: [], cursor: }; ≤ limit records; each record carries updated_at + agent_id + subject{type,id} + worldtree_scope (the fields person-prime's client _scan_filter_matches keys on — a record missing any is silently dropped client-side) -- assert +POST: [POST-002 return_value] LIVE-only — returns ONLY live chunks; superseded/tombstoned excluded server-side (INV-009) +POST: [POST-003 return_value] GLOBAL-order — the FULL scope-filtered live set is ordered by (sort.field, direction) GLOBALLY before the limit page; missing value LAST; chunk_id tiebreak (INV-010) +STEPS: + 1. [setup] validate limit (>0) + scope (as search); sort ← the dispatch-validated {field,direction} + 2. [sequential, flexibility=indicative] SELECT scope-filtered LIVE chunks ordered by the indexed sort field (json_extract(record_json,'$.updated_at')) in `direction`, missing-last, chunk_id tiebreak, GLOBALLY; apply cursor offset; take limit + 3. [cleanup] RETURN {records: verbatim chunks, cursor: next-page-or-None} +TESTS: + scan_recency [happy,tracer]: upsert 4 live chunks w/ distinct updated_at; scan(scope_all={end_user}, limit=3, sort={field:updated_at,direction:desc}) → the 3 newest, newest-first + global_before_paginate [scenario]: 5 chunks, limit=2 → page-1 = the 2 globally-newest; the cursor page continues the GLOBAL order, not a page-local re-sort (INV-010; bifrost cross-pagination conformance) + live_only [adversarial]: a superseded/tombstoned chunk is NEVER returned even if it is the newest (INV-009) + scope_isolation [adversarial]: scope_all one end_user → never returns another partition's chunk (INV-005 applies to scan) + unadvertised_sort [adversarial]: sort.field ∉ sortable_chunk_fields → InvalidArguments at dispatch (never silent unsorted) + person_prime_record_shape [scenario]: each record carries agent_id + subject{type,id} + worldtree_scope + updated_at + verbatim/distillate — the _scan_filter_matches keys (else the client silently drops it) + parity_vs_reference [scenario]: identical scan envelopes vs InMemoryMemoryStore → same ordered chunk_ids/shape (#195) +``` + ```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. diff --git a/pyproject.toml b/pyproject.toml index b538500..a6f5f89 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "ratatoskr" -version = "0.20.10" +version = "0.20.11" description = "Worldtree Conversation API debug TUI — multi-pane observability dashboard" readme = "README.md" requires-python = ">=3.12" diff --git a/src/ratatoskr/provider/memory_store.py b/src/ratatoskr/provider/memory_store.py index 0807566..937f91e 100644 --- a/src/ratatoskr/provider/memory_store.py +++ b/src/ratatoskr/provider/memory_store.py @@ -132,6 +132,22 @@ def _validate_injection(record: dict) -> None: raise InvalidArguments("injection_source only valid for injected_context origin") +_SORTABLE_CHUNK_FIELDS: list[dict] = [{"name": "updated_at"}] +_SORTABLE_FIELD_NAMES = frozenset(f["name"] for f in _SORTABLE_CHUNK_FIELDS) + + +def _is_live(record: dict) -> bool: + """INV-009: a chunk is live unless a lifecycle/governance marker says otherwise. + scan returns live-only server-side (person-prime's `lifecycle_state=live` does not + ride the scan wire, so this is authoritative — a dead fact can never inject).""" + state = record.get("lifecycle_state") + if isinstance(state, str) and state and state != "live": + return False + verbatim = record.get("verbatim") + gov = verbatim.get("governance_state") if isinstance(verbatim, dict) else None + return gov not in ("superseded", "tombstoned") + + def _chunk_content_preview(record: dict) -> str: """Best-effort human-readable content for the DEBUG memory viewer only. Prefers an explicit text field, then the distillate summary, and last-resorts to a compact JSON @@ -170,6 +186,7 @@ class RatatoskrMemoryStore: atomic_supersede_supported=False, transaction_supported=False, filterable_metadata_fields=[], + sortable_chunk_fields=list(_SORTABLE_CHUNK_FIELDS), # INV-006: gates scan sort + #349 person-prime ).to_dict() async def upsert_many( @@ -350,6 +367,68 @@ class RatatoskrMemoryStore: self._conn.execute("DELETE FROM memory_vec WHERE chunk_id = ?", (chunk_id,)) return {"deleted": deleted} + async def scan( + self, + *, + scope_all: dict | None = None, + scope_any: list | None = None, + cursor: str | None = None, + limit: int, + sort: dict | None = None, + lifecycle_state: Any = None, + ) -> dict: + # #349 person-prime: query-LESS, LIVE-only (INV-009), globally-ordered (INV-010) scan. + if isinstance(limit, bool) or not isinstance(limit, int) or limit <= 0: # PRE-001 + raise InvalidArguments("limit must be a positive int") + scope_all = scope_all or {} + scope_any = scope_any or [] + _validate_scope(scope_all, scope_any) # PRE-002 (same lattice as search) + field = (sort or {}).get("field", "updated_at") + direction = (sort or {}).get("direction", "desc") + if field not in _SORTABLE_FIELD_NAMES or direction not in ("asc", "desc"): # PRE-003 + raise InvalidArguments(f"sort.field {field!r} is not globally sortable") + _log.info( + "memory-call scan REQUEST: scope_all=%r scope_any=%r limit=%s sort=%s", + scope_all, scope_any, limit, sort, + ) + # INV-010: global order by the INDEXED sort field, missing-last, chunk_id tiebreak + # (field is whitelisted above, so the interpolation is injection-safe). + order = "DESC" if direction == "desc" else "ASC" + rows = self._conn.execute( + "SELECT record_json FROM memory_chunks " + f"ORDER BY (json_extract(record_json, '$.{field}') IS NULL), " + f"json_extract(record_json, '$.{field}') {order}, chunk_id ASC" + ).fetchall() + skip = 0 + if cursor is not None: + try: + skip = int(cursor) + except (TypeError, ValueError): + raise InvalidArguments("invalid scan cursor") + if skip < 0: + raise InvalidArguments("invalid scan cursor") + records: list[dict] = [] + matched = 0 + has_more = False + for (record_json,) in rows: + record = json.loads(record_json) + if not _matches_scope(record.get("scope"), scope_all, scope_any): # INV-005 + continue + if not _is_live(record): # INV-009 + continue + matched += 1 + if matched <= skip: # cursor is an offset into the GLOBAL order (INV-010) + continue + if len(records) >= limit: # POST-001: single limit page; one more match => next page exists + has_more = True + break + records.append(record) + # Emit a cursor ONLY when a further match exists — so a page that exactly exhausts + # the matched set returns cursor=None (no empty trailing page), matching the reference. + next_cursor = str(skip + len(records)) if has_more else None + _log.info("memory-call scan RESPONSE: %d record(s) next_cursor=%s", len(records), next_cursor) + return {"records": records, "cursor": next_cursor} + def count_chunks(self) -> int: """DEBUG read seam: total stored chunk rows (unfiltered). Lets the memory viewer distinguish 'store is empty' (total 0 — no upsert ever landed) from @@ -420,6 +499,12 @@ def open_memory_store(db_path: str, *, embedding_dim: int) -> RatatoskrMemorySto "chunk_id TEXT PRIMARY KEY, record_json TEXT NOT NULL, " "revision INTEGER NOT NULL, scope_json TEXT, origin TEXT)" ) + # INV-010: expression index on the scan sort field (updated_at) so the globally-ordered + # person-prime scan stays within its 500ms fail-open budget. + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_chunks_updated_at " + "ON memory_chunks (json_extract(record_json, '$.updated_at'))" + ) conn.execute( "CREATE TABLE IF NOT EXISTS memory_idempotency (" "idempotency_id TEXT PRIMARY KEY, digest TEXT NOT NULL, expires_at REAL)" diff --git a/tests/test_provider_memory.py b/tests/test_provider_memory.py index 33dd7c5..ed42955 100644 --- a/tests/test_provider_memory.py +++ b/tests/test_provider_memory.py @@ -376,6 +376,120 @@ async def test_delete_absent_counts_zero(): assert await store.delete_many(["nope"]) == {"deleted": 0} +# --- scan (#349 person-prime: sorted, live-only, paginated) --- + +async def test_scan_recency_returns_newest_live_chunks_desc(): + # tracer: upsert 4 live chunks with distinct updated_at; scan limit=3 desc -> 3 newest + store = open_memory_store(":memory:", embedding_dim=EMBEDDING_DIM) + recs = [ + _chunk(f"c{i}", scope={"end_user": "u1"}, updated_at=f"2026-07-15T00:0{i}:00+00:00") + for i in range(4) + ] + await store.upsert_many(recs, idempotency_key="k1", ctx=_ctx()) + out = await store.scan( + scope_all={"end_user": "u1"}, + limit=3, + sort={"field": "updated_at", "direction": "desc"}, + ) + assert [r["id"] for r in out["records"]] == ["c3", "c2", "c1"] # 3 globally-newest, newest-first + assert "cursor" in out + + +async def test_scan_excludes_superseded_and_tombstoned(): + # INV-009: dead chunks never returned, even if they're the newest. + store = open_memory_store(":memory:", embedding_dim=EMBEDDING_DIM) + recs = [ + _chunk("live1", scope={"end_user": "u1"}, updated_at="2026-07-15T00:01:00+00:00"), + _chunk("dead1", scope={"end_user": "u1"}, updated_at="2026-07-15T00:09:00+00:00", lifecycle_state="superseded"), + _chunk("dead2", scope={"end_user": "u1"}, updated_at="2026-07-15T00:08:00+00:00", verbatim={"text": "x", "governance_state": "tombstoned"}), + ] + await store.upsert_many(recs, idempotency_key="k", ctx=_ctx()) + out = await store.scan(scope_all={"end_user": "u1"}, limit=10, sort={"field": "updated_at", "direction": "desc"}) + assert [r["id"] for r in out["records"]] == ["live1"] + + +async def test_scan_scope_isolation_excludes_other_partition(): + # INV-005 applies to scan. + store = open_memory_store(":memory:", embedding_dim=EMBEDDING_DIM) + recs = [ + _chunk("a", scope={"end_user": "u1"}, updated_at="2026-07-15T00:01:00+00:00"), + _chunk("b", scope={"end_user": "u2"}, updated_at="2026-07-15T00:09:00+00:00"), + ] + await store.upsert_many(recs, idempotency_key="k", ctx=_ctx()) + out = await store.scan(scope_all={"end_user": "u1"}, limit=10, sort={"field": "updated_at", "direction": "desc"}) + assert [r["id"] for r in out["records"]] == ["a"] # u2's newer chunk never surfaces + + +async def test_scan_unadvertised_sort_field_rejected(): + # PRE-003: a sort field not in sortable_chunk_fields -> InvalidArguments (never silent unsorted). + store = open_memory_store(":memory:", embedding_dim=EMBEDDING_DIM) + with pytest.raises(InvalidArguments): + await store.scan(scope_all={"end_user": "u1"}, limit=3, sort={"field": "salience", "direction": "desc"}) + + +async def test_scan_records_carry_person_prime_filter_fields(): + # The client _scan_filter_matches keys on agent_id + subject + worldtree_scope; a record + # missing any is silently dropped -> the scan record must carry them verbatim. + store = open_memory_store(":memory:", embedding_dim=EMBEDDING_DIM) + rec = _chunk( + "c1", scope={"end_user": "u1"}, updated_at="2026-07-15T00:01:00+00:00", + agent_id="ratatoskr:sindra", subject={"type": "end_user", "id": "u1"}, worldtree_scope="end_user", + ) + await store.upsert_many([rec], idempotency_key="k", ctx=_ctx()) + out = await store.scan(scope_all={"end_user": "u1"}, limit=3, sort={"field": "updated_at", "direction": "desc"}) + r = out["records"][0] + assert r["agent_id"] == "ratatoskr:sindra" + assert r["subject"] == {"type": "end_user", "id": "u1"} + assert r["worldtree_scope"] == "end_user" + assert r["updated_at"] == "2026-07-15T00:01:00+00:00" + + +async def test_scan_global_order_across_pages_via_cursor(): + # INV-010: the cursor page continues the GLOBAL order, never a page-local re-sort. + store = open_memory_store(":memory:", embedding_dim=EMBEDDING_DIM) + recs = [_chunk(f"c{i}", scope={"end_user": "u1"}, updated_at=f"2026-07-15T00:0{i}:00+00:00") for i in range(5)] + await store.upsert_many(recs, idempotency_key="k", ctx=_ctx()) + p1 = await store.scan(scope_all={"end_user": "u1"}, limit=2, sort={"field": "updated_at", "direction": "desc"}) + assert [r["id"] for r in p1["records"]] == ["c4", "c3"] # 2 globally-newest + assert p1["cursor"] is not None + p2 = await store.scan(scope_all={"end_user": "u1"}, limit=2, cursor=p1["cursor"], sort={"field": "updated_at", "direction": "desc"}) + assert [r["id"] for r in p2["records"]] == ["c2", "c1"] # continues the global order + + +async def test_scan_parity_vs_reference_inmemory_store(): + # #195: identical scan envelopes vs the bifrost reference InMemoryMemoryStore produce + # the SAME ordered chunk_ids + verbatim record shape. All chunks LIVE — our scan is + # live-only (INV-009) while the reference does NOT lifecycle-filter, so parity is only + # defined over the live set (the person-prime case). Both READ updated_at from the + # record (neither stamps it), so ordering is a pure function of the shared input. + from bifrost.consumer.testing import InMemoryMemoryStore + + records = [ + _chunk("z1", scope={"end_user": "u1"}, updated_at="2026-07-15T00:03:00+00:00"), + _chunk("a2", scope={"end_user": "u1"}, updated_at="2026-07-15T00:01:00+00:00"), + _chunk("a3", scope={"end_user": "u1"}, updated_at="2026-07-15T00:01:00+00:00"), + _chunk("m4", scope={"end_user": "u1"}), # no updated_at -> sorts LAST, both directions + ] + scope_all = {"end_user": "u1"} + sort = {"field": "updated_at", "direction": "desc"} + + ours = open_memory_store(":memory:", embedding_dim=EMBEDDING_DIM) + await ours.upsert_many(records, idempotency_key="k", ctx=_ctx()) + ref = InMemoryMemoryStore() + await ref.upsert_many(records, idempotency_key="k", ctx=_ctx()) + + # identical scan envelope on both stores + out_ours = await ours.scan(scope_all=scope_all, limit=10, sort=sort) + out_ref = await ref.scan(scope_all=scope_all, limit=10, sort=sort) + + # recency beats id (z1 first despite 'z' > 'a'); tie broken by id asc (a2 < a3); + # missing updated_at sorts last (m4). + expected = ["z1", "a2", "a3", "m4"] + assert [r["id"] for r in out_ref["records"]] == expected + assert [r["id"] for r in out_ours["records"]] == expected + assert out_ours["records"] == out_ref["records"] # verbatim record shape parity + + # --- build_memory_provider_app --- def test_build_app_exposes_handshake_and_memory_routes(): diff --git a/uv.lock b/uv.lock index 5e972dd..62e0c99 100644 --- a/uv.lock +++ b/uv.lock @@ -1052,7 +1052,7 @@ wheels = [ [[package]] name = "ratatoskr" -version = "0.20.10" +version = "0.20.11" source = { editable = "." } dependencies = [ { name = "httpx" },