feat(provider): person-prime scan verb + sortable_chunk_fields cap (WT #349)

Implement the memory-store `scan` verb — a query-LESS, LIVE-only, globally
ordered top-N-by-recency read — and advertise `sortable_chunk_fields=
[{updated_at}]` at the Bifrost handshake. Advertising the cap is what lights
up Worldtree's #349 person-prime turn-1 durable-fact injection (Branch-A
`"updated_at" in caps.sort_fields_supported`); the fix is ZERO Worldtree
change — the running provider announcing the cap is the trigger.

scan is:
- LIVE-only server-side (INV-009): superseded/tombstoned excluded — a dead
  fact can never inject; person-prime's `lifecycle_state=live` does not ride
  the scan wire, so server-side is authoritative.
- Globally ordered before pagination (INV-010): the full scope-filtered live
  set is ordered by (sort.field, direction) globally; missing value LAST,
  chunk_id tiebreak. Backed by an expression index on
  json_extract(record_json,'$.updated_at') to stay in the 500ms budget.
- Cursor = offset into the global order; emits a next cursor only when a
  further match exists (no empty trailing page — matches the reference).

Sort is dispatch-gated: an unadvertised sort.field raises InvalidArguments,
never a silent unsorted fallback.

Contract amended: un-defers scan, adds the FN spec + INV-009/INV-010 +
sortable_chunk_fields to INV-006. TDD 7/7 green (scan_recency tracer,
live_only, scope_isolation, unadvertised_sort, person_prime_record_shape,
cursor pagination, parity_vs_reference vs InMemoryMemoryStore #195). Full
suite 638 green.

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: [<verbatim chunk wire records, same shape as a search hit's chunk>], cursor: <opaque next-page str | None>}; ≤ 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.

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"

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)"

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():

uv.lock

@@ -1052,7 +1052,7 @@ wheels = [
 
 [[package]]
 name = "ratatoskr"
-version = "0.20.10"
+version = "0.20.11"
 source = { editable = "." }
 dependencies = [
     { name = "httpx" },