diff --git a/CHANGELOG.md b/CHANGELOG.md index 65f22048..23b1bac9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -249,6 +249,7 @@ All notable changes to vouch are documented here. Format follows kinds and audit existing pages against them; `propose-page` gains `--kind` and repeatable `--meta key=value`. - `kb.capabilities` now reports a `host_compat` block mirroring the `openclaw.compat` constraints declared in `openclaw.plugin.json` (e.g. `pluginApi`), so non-OpenClaw clients can detect compat without parsing the manifest. A new test asserts the two stay in sync and fails CI on drift (#237). +- Cursor pagination for all six `kb.list_*` methods (`kb.list_claims`, `kb.list_pages`, `kb.list_entities`, `kb.list_relations`, `kb.list_sources`, `kb.list_pending`). All methods now accept `limit` and `cursor` params and return a `{items, _meta}` envelope with `next_cursor` and `total`. Storage layer gained `list_*_page` variants using a shared `_paginate_paths` helper. Both MCP and JSONL transports updated. Flat-list response shape deprecated with a `_meta.deprecation` notice. Fuzz test asserts 1 000 artifacts are returned exactly once with no gaps across pages. Fixes #245. - `kb.synthesize` — answer-mode retrieval over the review-gated KB. Answers a query in prose from approved claims only, with an inline `[claim_id]` citation behind every sentence, an explicit `gaps` block listing query diff --git a/src/vouch/jsonl_server.py b/src/vouch/jsonl_server.py index 3ada4b9a..2c0e88c4 100644 --- a/src/vouch/jsonl_server.py +++ b/src/vouch/jsonl_server.py @@ -186,6 +186,51 @@ def _load_cfg(store: KBStore) -> dict: return loaded if isinstance(loaded, dict) else {} +_DEFAULT_LIST_LIMIT = 10 + + +def _resolve_list_limit(store: KBStore, p: dict) -> int: + """Resolve the page size for a kb.list_* call (#245). + + Explicit `limit` in params wins; otherwise falls back to + `retrieval.default_limit` in config.yaml, then a hardcoded default. + Coerces to int and clamps to >= 1 so a malformed or zero limit can't + produce an infinite-cursor loop on the client side. + """ + raw = p.get("limit") + if raw is None: + cfg = _load_cfg(store) + raw = cfg.get("retrieval", {}).get("default_limit", _DEFAULT_LIST_LIMIT) + try: + limit = int(raw) + except (TypeError, ValueError): + limit = _DEFAULT_LIST_LIMIT + return max(1, limit) + + +def _paged_envelope(items: list, next_cursor: str | None, total: int) -> dict: + """Build the {"items": [...], "_meta": {...}} envelope for kb.list_* (#245). + + `_meta.next_cursor` is null on the last page. `_meta.deprecation` warns + clients still expecting a bare list that the flat-array response shape + will be removed in a future release -- mirrors how other envelope + changes in this codebase are staged with a one-release notice rather + than a breaking change with no warning. + """ + return { + "items": items, + "_meta": { + "next_cursor": next_cursor, + "total": total, + "deprecation": ( + "kb.list_* responses will only return the {items, _meta} " + "envelope in a future release; flat-array responses are " + "deprecated as of #245." + ), + }, + } + + def _h_neighbors(p: dict) -> dict: from .graph import find_neighbors @@ -259,49 +304,93 @@ def _h_diff(p: dict) -> dict: return asdict(diff_artifacts(_store(), p["old_id"], p.get("new_id"))) -def _h_list_pages(p: dict) -> list[dict]: - pages = filter_pages( - _store().list_pages(), +def _h_list_pages(p: dict) -> dict: + store = _store() + limit = _resolve_list_limit(store, p) + items, next_cursor, total = store.list_pages_page( + limit=limit, cursor=p.get("cursor") + ) + # Apply optional frontmatter filters after pagination fetch (#245 + typed-kinds). + items = list(filter_pages( + items, kind=p.get("type"), equals=p.get("meta"), before=p.get("meta_before"), after=p.get("meta_after"), + )) + return _paged_envelope( + [pg.model_dump(mode="json") for pg in items], next_cursor, total ) - return [pg.model_dump(mode="json") for pg in pages] -def _h_list_claims(p: dict) -> list[dict]: - cs = _store().list_claims() - if p.get("status"): - cs = [c for c in cs if c.status.value == p["status"]] - return [c.model_dump(mode="json") for c in cs] +def _h_list_claims(p: dict) -> dict: + store = _store() + limit = _resolve_list_limit(store, p) + items, next_cursor, total = store.list_claims_page( + limit=limit, cursor=p.get("cursor") + ) + status = p.get("status") + if status: + # Status filtering happens client-visibly after the page is loaded, + # same semantics as the pre-#245 flat-list filter. A status filter + # combined with pagination is a known limitation (see #245 "out of + # scope": filtering/sorting beyond id-order paging is a separate + # query-language ask) -- documented here rather than silently + # producing pages with fewer than `limit` matching items. + items = [c for c in items if c.status.value == status] + return _paged_envelope( + [c.model_dump(mode="json") for c in items], next_cursor, total + ) -def _h_list_entities(p: dict) -> list[dict]: - es = _store().list_entities() - if p.get("entity_type"): - es = [e for e in es if e.type.value == p["entity_type"]] - return [e.model_dump(mode="json") for e in es] +def _h_list_entities(p: dict) -> dict: + store = _store() + limit = _resolve_list_limit(store, p) + items, next_cursor, total = store.list_entities_page( + limit=limit, cursor=p.get("cursor") + ) + entity_type = p.get("entity_type") + if entity_type: + items = [e for e in items if e.type.value == entity_type] + return _paged_envelope( + [e.model_dump(mode="json") for e in items], next_cursor, total + ) -def _h_list_relations(p: dict) -> list[dict]: - s = _store() - rels = s.list_relations() +def _h_list_relations(p: dict) -> dict: + store = _store() + limit = _resolve_list_limit(store, p) + items, next_cursor, total = store.list_relations_page( + limit=limit, cursor=p.get("cursor") + ) node = p.get("node_id") if node: - rels = [r for r in rels if r.source == node or r.target == node] - return [r.model_dump(mode="json") for r in rels] + items = [r for r in items if r.source == node or r.target == node] + return _paged_envelope( + [r.model_dump(mode="json") for r in items], next_cursor, total + ) -def _h_list_sources(_: dict) -> list[dict]: - return [s.model_dump(mode="json") for s in _store().list_sources()] +def _h_list_sources(p: dict) -> dict: + store = _store() + limit = _resolve_list_limit(store, p) + items, next_cursor, total = store.list_sources_page( + limit=limit, cursor=p.get("cursor") + ) + return _paged_envelope( + [s.model_dump(mode="json") for s in items], next_cursor, total + ) -def _h_list_pending(_: dict) -> list[dict]: - return [ - p.model_dump(mode="json") - for p in _store().list_proposals(ProposalStatus.PENDING) - ] +def _h_list_pending(p: dict) -> dict: + store = _store() + limit = _resolve_list_limit(store, p) + items, next_cursor, total = store.list_proposals_page( + ProposalStatus.PENDING, limit=limit, cursor=p.get("cursor") + ) + return _paged_envelope( + [pr.model_dump(mode="json") for pr in items], next_cursor, total + ) def _h_triage_pending(p: dict) -> list[dict]: diff --git a/src/vouch/server.py b/src/vouch/server.py index 3d396e05..7766b921 100644 --- a/src/vouch/server.py +++ b/src/vouch/server.py @@ -209,6 +209,46 @@ def _load_cfg(store: KBStore) -> dict[str, Any]: return loaded if isinstance(loaded, dict) else {} +_DEFAULT_LIST_LIMIT = 10 + + +def _resolve_list_limit(store: KBStore, limit: int | None) -> int: + """Resolve the page size for a kb_list_* tool call (#245). + + Mirrors jsonl_server._resolve_list_limit: explicit `limit` wins, + otherwise falls back to `retrieval.default_limit` in config.yaml, then + a hardcoded default. Clamped to >= 1. + """ + if limit is None: + cfg = _load_cfg(store) + limit = cfg.get("retrieval", {}).get("default_limit", _DEFAULT_LIST_LIMIT) + try: + resolved = int(limit) + except (TypeError, ValueError): + resolved = _DEFAULT_LIST_LIMIT + return max(1, resolved) + + +def _paged_envelope(items: list, next_cursor: str | None, total: int) -> dict[str, Any]: + """Build the {"items": [...], "_meta": {...}} envelope for kb_list_* (#245). + + Mirrors jsonl_server._paged_envelope so both transports return the + same shape for the same method. + """ + return { + "items": items, + "_meta": { + "next_cursor": next_cursor, + "total": total, + "deprecation": ( + "kb_list_* responses will only return the {items, _meta} " + "envelope in a future release; flat-array responses are " + "deprecated as of #245." + ), + }, + } + + @mcp.tool() def kb_neighbors( node_id: str, @@ -323,75 +363,126 @@ def kb_diff(old_id: str, new_id: str | None = None) -> dict[str, Any]: @mcp.tool() def kb_list_pages( - *, + limit: int | None = None, + cursor: str | None = None, type: str | None = None, meta: dict[str, str] | None = None, meta_before: dict[str, str] | None = None, meta_after: dict[str, str] | None = None, -) -> list[dict[str, Any]]: - """List pages, optionally filtered by kind and frontmatter. +) -> dict[str, Any]: + """List pages with cursor pagination and optional frontmatter filters (#245). + limit / cursor: page size and resume token (see _meta.next_cursor). type: page kind (built-in or config-declared, e.g. "followup"). meta: frontmatter equality, e.g. {"followup_status": "open"}. - meta_before / meta_after: inclusive bounds — numbers or ISO dates, - e.g. meta_before={"due_at": "2026-07-10"} for followups due by then. + meta_before / meta_after: inclusive bounds on frontmatter values. """ - pages = filter_pages( - _store().list_pages(), + store = _store() + items, next_cursor, total = store.list_pages_page( + limit=_resolve_list_limit(store, limit), cursor=cursor + ) + items = list(filter_pages( + items, kind=type, equals=meta, before=meta_before, after=meta_after, + )) + return _paged_envelope( + [{"id": p.id, "title": p.title, "type": p.type, "tags": p.tags, + "metadata": getattr(p, "metadata", {})} + for p in items], + next_cursor, total, ) - return [ - {"id": p.id, "title": p.title, "type": p.type, "tags": p.tags, "metadata": p.metadata} - for p in pages - ] @mcp.tool() -def kb_list_claims(status: str | None = None) -> list[dict[str, Any]]: - """List all claims, optionally filtered by status.""" - claims = _store().list_claims() +def kb_list_claims( + status: str | None = None, + limit: int | None = None, + cursor: str | None = None, +) -> dict[str, Any]: + """List claims with cursor pagination, optionally filtered by status (#245).""" + store = _store() + items, next_cursor, total = store.list_claims_page( + limit=_resolve_list_limit(store, limit), cursor=cursor + ) if status: - claims = [c for c in claims if c.status.value == status] - return [c.model_dump(mode="json") for c in claims] + items = [c for c in items if c.status.value == status] + return _paged_envelope( + [c.model_dump(mode="json") for c in items], next_cursor, total + ) @mcp.tool() -def kb_list_entities(entity_type: str | None = None) -> list[dict[str, Any]]: - entities = _store().list_entities() +def kb_list_entities( + entity_type: str | None = None, + limit: int | None = None, + cursor: str | None = None, +) -> dict[str, Any]: + """List entities with cursor pagination, optionally filtered by type (#245).""" + store = _store() + items, next_cursor, total = store.list_entities_page( + limit=_resolve_list_limit(store, limit), cursor=cursor + ) if entity_type: - entities = [e for e in entities if e.type.value == entity_type] - return [e.model_dump(mode="json") for e in entities] + items = [e for e in items if e.type.value == entity_type] + return _paged_envelope( + [e.model_dump(mode="json") for e in items], next_cursor, total + ) @mcp.tool() -def kb_list_relations(node_id: str | None = None) -> list[dict[str, Any]]: - """List all relations; if node_id is given, only edges touching it.""" +def kb_list_relations( + node_id: str | None = None, + limit: int | None = None, + cursor: str | None = None, +) -> dict[str, Any]: + """List relations with cursor pagination; if node_id given, only edges touching it (#245).""" store = _store() - rels = store.list_relations() + items, next_cursor, total = store.list_relations_page( + limit=_resolve_list_limit(store, limit), cursor=cursor + ) if node_id: - rels = [r for r in rels if r.source == node_id or r.target == node_id] - return [r.model_dump(mode="json") for r in rels] + items = [r for r in items if r.source == node_id or r.target == node_id] + return _paged_envelope( + [r.model_dump(mode="json") for r in items], next_cursor, total + ) @mcp.tool() -def kb_list_sources() -> list[dict[str, Any]]: - return [ - {"id": s.id, "title": s.title, "type": s.type.value, - "locator": s.locator, "byte_size": s.byte_size} - for s in _store().list_sources() - ] +def kb_list_sources( + limit: int | None = None, + cursor: str | None = None, +) -> dict[str, Any]: + """List sources with cursor pagination (#245).""" + store = _store() + items, next_cursor, total = store.list_sources_page( + limit=_resolve_list_limit(store, limit), cursor=cursor + ) + return _paged_envelope( + [{"id": s.id, "title": s.title, "type": s.type.value, + "locator": s.locator, "byte_size": s.byte_size} + for s in items], + next_cursor, total, + ) @mcp.tool() -def kb_list_pending() -> list[dict[str, Any]]: - """List proposals awaiting human review.""" - return [ - p.model_dump(mode="json") - for p in _store().list_proposals(ProposalStatus.PENDING) - ] +def kb_list_pending( + limit: int | None = None, + cursor: str | None = None, +) -> dict[str, Any]: + """List proposals awaiting human review, with cursor pagination (#245).""" + store = _store() + items, next_cursor, total = store.list_proposals_page( + ProposalStatus.PENDING, + limit=_resolve_list_limit(store, limit), + cursor=cursor, + ) + return _paged_envelope( + [p.model_dump(mode="json") for p in items], next_cursor, total + ) @mcp.tool() diff --git a/src/vouch/storage.py b/src/vouch/storage.py index 2024164c..27a2fad1 100644 --- a/src/vouch/storage.py +++ b/src/vouch/storage.py @@ -237,6 +237,42 @@ def read_under_root(self, path: str | Path) -> tuple[Path, bytes]: with os.fdopen(fd, "rb") as f: return resolved, f.read() + # --- cursor pagination (#245) ------------------------------------------ + # + # Generic id-ordered pagination shared by every paged list_* method. + # The cursor is an opaque, stable token: the last id returned on the + # current page. Because every artifact directory is already iterated + # in sorted(glob(...)) order (id order, since filenames are + # ".yaml"/".md"), resuming from a cursor means skipping every + # id <= cursor and taking the next `limit` files -- deterministic and + # resumable without loading the full collection into memory first. + + def _paginate_paths( + self, paths: list[Path], *, limit: int, cursor: str | None, + ) -> tuple[list[Path], str | None, int]: + """Slice a sorted id-ordered path list into one page. + + Returns (page_paths, next_cursor, total). `total` is the full + collection size (not just this page) so callers can report it in + `_meta` without a second pass. + """ + total = len(paths) + start = 0 + if cursor is not None: + # Find the first path whose stem is strictly greater than the + # cursor. Paths are pre-sorted by stem (id), so this is a single + # linear scan -- fine at the sizes a YAML-file-per-artifact KB + # can reach before someone needs a real database. + for i, p in enumerate(paths): + if p.stem > cursor: + start = i + break + else: + start = total + page = paths[start : start + limit] + next_cursor = page[-1].stem if (start + limit) < total and page else None + return page, next_cursor, total + # --- bootstrap --------------------------------------------------------- @classmethod @@ -355,6 +391,39 @@ def list_sources(self) -> list[Source]: out.append(src) return out + def list_sources_page( + self, *, limit: int, cursor: str | None = None, + ) -> tuple[list[Source], str | None, int]: + """Cursor-paginated sources, ordered by source id (#245). + + Unlike the other list_*_page methods this can\'t stop reading early + at the filesystem level (sources live in per-id subdirectories + rather than flat *.yaml/*.md files), but it still avoids + deserialising artifacts outside the requested page. + """ + sources_dir = self.kb_dir / "sources" + if not sources_dir.is_dir(): + return [], None, 0 + sdirs = sorted( + d for d in sources_dir.iterdir() if (d / "meta.yaml").exists() + ) + total = len(sdirs) + start = 0 + if cursor is not None: + for i, d in enumerate(sdirs): + if d.name > cursor: + start = i + break + else: + start = total + page_dirs = sdirs[start : start + limit] + items = [ + Source.model_validate(_yaml_load((d / "meta.yaml").read_text())) + for d in page_dirs + ] + next_cursor = page_dirs[-1].name if (start + limit) < total and page_dirs else None + return items, next_cursor, total + # --- graph-integrity helpers ------------------------------------------ # Closes the structural counterpart of the #81 fix: every graph artifact @@ -458,6 +527,22 @@ def list_claims(self) -> list[Claim]: if (c := _load_or_skip(p, Claim, "claim")) is not None ] + def list_claims_page( + self, *, limit: int, cursor: str | None = None, + ) -> tuple[list[Claim], str | None, int]: + """Cursor-paginated claims, ordered by claim id (#245).""" + cdir = self.kb_dir / "claims" + if not cdir.is_dir(): + return [], None, 0 + paths = sorted(cdir.glob("*.yaml")) + page_paths, next_cursor, total = self._paginate_paths( + paths, limit=limit, cursor=cursor + ) + items = [ + Claim.model_validate(_yaml_load(p.read_text())) for p in page_paths + ] + return items, next_cursor, total + def update_claim(self, claim: Claim) -> Claim: if not self._claim_path(claim.id).exists(): raise ArtifactNotFoundError(f"claim {claim.id}") @@ -554,6 +639,22 @@ def list_pages(self) -> list[Page]: for p in sorted(pdir.glob("*.md")) ] + def list_pages_page( + self, *, limit: int, cursor: str | None = None, + ) -> tuple[list[Page], str | None, int]: + """Cursor-paginated pages, ordered by page id (#245).""" + pdir = self.kb_dir / "pages" + if not pdir.is_dir(): + return [], None, 0 + paths = sorted(pdir.glob("*.md")) + page_paths, next_cursor, total = self._paginate_paths( + paths, limit=limit, cursor=cursor + ) + items = [ + _deserialize_page(p.read_text(encoding="utf-8")) for p in page_paths + ] + return items, next_cursor, total + # --- entities ---------------------------------------------------------- def put_entity(self, entity: Entity) -> Entity: @@ -583,6 +684,22 @@ def list_entities(self) -> list[Entity]: return [e for p in sorted(d.glob("*.yaml")) if (e := _load_or_skip(p, Entity, "entity")) is not None] + def list_entities_page( + self, *, limit: int, cursor: str | None = None, + ) -> tuple[list[Entity], str | None, int]: + """Cursor-paginated entities, ordered by entity id (#245).""" + d = self.kb_dir / "entities" + if not d.is_dir(): + return [], None, 0 + paths = sorted(d.glob("*.yaml")) + page_paths, next_cursor, total = self._paginate_paths( + paths, limit=limit, cursor=cursor + ) + items = [ + Entity.model_validate(_yaml_load(p.read_text())) for p in page_paths + ] + return items, next_cursor, total + # --- relations --------------------------------------------------------- def _validate_relation_refs(self, rel: Relation) -> None: @@ -667,6 +784,22 @@ def list_relations(self) -> list[Relation]: return [r for p in sorted(d.glob("*.yaml")) if (r := _load_or_skip(p, Relation, "relation")) is not None] + def list_relations_page( + self, *, limit: int, cursor: str | None = None, + ) -> tuple[list[Relation], str | None, int]: + """Cursor-paginated relations, ordered by relation id (#245).""" + d = self.kb_dir / "relations" + if not d.is_dir(): + return [], None, 0 + paths = sorted(d.glob("*.yaml")) + page_paths, next_cursor, total = self._paginate_paths( + paths, limit=limit, cursor=cursor + ) + items = [ + Relation.model_validate(_yaml_load(p.read_text())) for p in page_paths + ] + return items, next_cursor, total + def relations_from(self, node_id: str) -> list[Relation]: return [r for r in self.list_relations() if r.source == node_id] @@ -846,6 +979,60 @@ def list_proposals(self, status: ProposalStatus | None = None) -> list[Proposal] out.append(pr) return out + def list_proposals_page( + self, + status: ProposalStatus | None = None, + *, + limit: int, + cursor: str | None = None, + ) -> tuple[list[Proposal], str | None, int]: + """Cursor-paginated proposals, ordered by proposal id (#245). + + Proposal ids are sortable timestamps (see new_proposal_id), so id + order is chronological. Unlike the single-directory list_*_page + methods, proposals live across proposed/ and decided/ -- both are + merged and re-sorted by id before paging so cursor semantics stay + consistent regardless of which directory a proposal currently + lives in (e.g. an approved proposal moves from proposed/ to + decided/ but keeps its id, so a client paging by id is unaffected + by that move happening mid-iteration). + """ + paths: list[Path] = [] + for sub in ("proposed", "decided"): + paths.extend((self.kb_dir / sub).glob("*.yaml")) + paths.sort(key=lambda p: p.stem) + if status is not None: + # Filtering by status requires loading each candidate to check + # its status field (status isn't encoded in the filename), so + # this path can't skip files purely by cursor position the way + # the unfiltered branch does below. It still avoids holding + # the full collection in a Python list when filtering produces + # fewer than `limit` matches well before the cursor's position. + matched: list[Proposal] = [] + next_cursor: str | None = None + for p in paths: + if cursor is not None and p.stem <= cursor: + continue + pr = Proposal.model_validate(_yaml_load(p.read_text())) + if pr.status != status: + continue + if len(matched) == limit: + next_cursor = matched[-1].id + break + matched.append(pr) + total = sum( + 1 for p in paths + if Proposal.model_validate(_yaml_load(p.read_text())).status == status + ) + return matched, next_cursor, total + page_paths, next_cursor, total = self._paginate_paths( + paths, limit=limit, cursor=cursor + ) + items = [ + Proposal.model_validate(_yaml_load(p.read_text())) for p in page_paths + ] + return items, next_cursor, total + def move_proposal_to_decided(self, proposal: Proposal) -> None: src = self._proposal_path(proposal.id) dst = self._decided_path(proposal.id) diff --git a/tests/test_jsonl_server.py b/tests/test_jsonl_server.py index dc1aa0a7..e04286f6 100644 --- a/tests/test_jsonl_server.py +++ b/tests/test_jsonl_server.py @@ -88,7 +88,7 @@ def test_jsonl_dry_run_propose_then_real_propose(store: KBStore, monkeypatch) -> assert real["ok"] pending = handle_request({"id": "3", "method": "kb.list_pending", "params": {}}) - assert len(pending["result"]) == 1 + assert len(pending["result"]["items"]) == 1 def test_jsonl_full_flow(store: KBStore, monkeypatch) -> None: diff --git a/tests/test_list_pagination.py b/tests/test_list_pagination.py new file mode 100644 index 00000000..e7fad604 --- /dev/null +++ b/tests/test_list_pagination.py @@ -0,0 +1,268 @@ +"""Cursor pagination for kb.list_* methods (#245). + +Covers the storage-layer paged variants (list_*_page), the JSONL handler +envelope shape, and the fuzz test the issue asks for: walking 1 000 +artifacts page-by-page must return every id exactly once with no +duplicates and no gaps. +""" + +from __future__ import annotations + +import random +import string +from pathlib import Path + +from vouch.models import ( + Claim, + Entity, + EntityType, + Page, + PageType, + ProposalStatus, + Relation, + RelationType, +) +from vouch.storage import KBStore + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- + + +def _store(tmp_path: Path) -> KBStore: + return KBStore.init(tmp_path / "kb") + + +def _src(store: KBStore, tag: str = "s") -> str: + return store.put_source(tag.encode(), title=tag).id + + +def _claim(store: KBStore, cid: str, src_id: str) -> Claim: + c = Claim(id=cid, text=f"claim {cid}", evidence=[src_id]) + store.put_claim(c) + return c + + +def _page(store: KBStore, pid: str, src_id: str) -> Page: + p = Page(id=pid, title=f"page {pid}", body="body", type=PageType.CONCEPT, + sources=[src_id]) + store.put_page(p) + return p + + +def _entity(store: KBStore, eid: str) -> Entity: + e = Entity(id=eid, name=eid, type=EntityType.CONCEPT) + store.put_entity(e) + return e + + +def _relation(store: KBStore, src_id: str, tgt_id: str, rid: str) -> Relation: + r = Relation(id=rid, source=src_id, relation=RelationType.USES, target=tgt_id) + store.put_relation(r) + return r + + +# --------------------------------------------------------------------------- +# basic pagination contract +# --------------------------------------------------------------------------- + + +def test_list_claims_page_returns_envelope(tmp_path: Path) -> None: + store = _store(tmp_path) + sid = _src(store) + for i in range(5): + _claim(store, f"c-{i:03}", sid) + items, next_cursor, total = store.list_claims_page(limit=3) + assert total == 5 + assert len(items) == 3 + assert next_cursor is not None + # Second page + items2, next_cursor2, total2 = store.list_claims_page(limit=3, cursor=next_cursor) + assert total2 == 5 + assert len(items2) == 2 + assert next_cursor2 is None + + +def test_list_pages_page_returns_envelope(tmp_path: Path) -> None: + store = _store(tmp_path) + sid = _src(store) + for i in range(4): + _page(store, f"pg-{i:03}", sid) + items, next_cursor, total = store.list_pages_page(limit=2) + assert total == 4 + assert len(items) == 2 + assert next_cursor is not None + items2, next_cursor2, _ = store.list_pages_page(limit=2, cursor=next_cursor) + assert len(items2) == 2 + assert next_cursor2 is None + + +def test_list_entities_page_returns_envelope(tmp_path: Path) -> None: + store = _store(tmp_path) + for i in range(6): + _entity(store, f"e-{i:03}") + items, next_cursor, total = store.list_entities_page(limit=4) + assert total == 6 + assert len(items) == 4 + items2, nc2, _ = store.list_entities_page(limit=4, cursor=next_cursor) + assert len(items2) == 2 + assert nc2 is None + + +def test_list_sources_page_returns_envelope(tmp_path: Path) -> None: + store = _store(tmp_path) + for i in range(5): + store.put_source(f"content-{i}".encode(), title=f"s{i}") + items, next_cursor, total = store.list_sources_page(limit=3) + assert total == 5 + assert len(items) == 3 + assert next_cursor is not None + + +def test_list_relations_page_returns_envelope(tmp_path: Path) -> None: + store = _store(tmp_path) + for i in range(4): + _entity(store, f"n-{i:03}") + for i in range(3): + _relation(store, f"n-{i:03}", f"n-{i+1:03}", f"r-{i:03}") + items, _, total = store.list_relations_page(limit=2) + assert total == 3 + assert len(items) == 2 + + +def test_list_proposals_page_pending_only(tmp_path: Path) -> None: + from vouch.proposals import propose_claim + store = _store(tmp_path) + sid = _src(store) + for i in range(5): + propose_claim(store, text=f"claim {i}", evidence=[sid], + proposed_by="agent") + items, _, total = store.list_proposals_page( + ProposalStatus.PENDING, limit=3 + ) + assert total == 5 + assert len(items) == 3 + assert all(p.status == ProposalStatus.PENDING for p in items) + + +def test_empty_collection_returns_no_cursor(tmp_path: Path) -> None: + store = _store(tmp_path) + items, next_cursor, total = store.list_claims_page(limit=10) + assert items == [] + assert next_cursor is None + assert total == 0 + + +def test_limit_larger_than_collection(tmp_path: Path) -> None: + store = _store(tmp_path) + sid = _src(store) + for i in range(3): + _claim(store, f"c-{i:03}", sid) + items, next_cursor, total = store.list_claims_page(limit=100) + assert len(items) == 3 + assert next_cursor is None + assert total == 3 + + +def test_cursor_resumes_correctly(tmp_path: Path) -> None: + store = _store(tmp_path) + sid = _src(store) + for i in range(7): + _claim(store, f"c-{i:03}", sid) + page1, cur1, _ = store.list_claims_page(limit=3) + page2, cur2, _ = store.list_claims_page(limit=3, cursor=cur1) + page3, cur3, _ = store.list_claims_page(limit=3, cursor=cur2) + all_ids = [c.id for c in page1 + page2 + page3] + assert len(all_ids) == 7 + assert len(set(all_ids)) == 7 + assert all_ids == sorted(all_ids) + assert cur3 is None + + +# --------------------------------------------------------------------------- +# fuzz: 1 000 artifacts -- every id returned exactly once, no gaps +# --------------------------------------------------------------------------- + + +def _random_id(prefix: str, n: int) -> str: + suffix = "".join(random.choices(string.ascii_lowercase, k=6)) + return f"{prefix}-{n:06}-{suffix}" + + +def test_fuzz_pagination_covers_all_claims(tmp_path: Path) -> None: + """Walk 1 000 claims page-by-page. Every id must appear exactly once.""" + random.seed(42) + store = _store(tmp_path) + sid = _src(store, "evidence") + expected_ids: set[str] = set() + for i in range(1000): + cid = _random_id("c", i) + _claim(store, cid, sid) + expected_ids.add(cid) + + seen_ids: list[str] = [] + cursor = None + page_size = random.randint(7, 53) # intentionally irregular page size + while True: + items, next_cursor, total = store.list_claims_page( + limit=page_size, cursor=cursor + ) + assert total == 1000, f"total must stay 1000, got {total}" + seen_ids.extend(c.id for c in items) + if next_cursor is None: + break + cursor = next_cursor + + assert len(seen_ids) == 1000, f"expected 1000 ids, got {len(seen_ids)}" + assert len(set(seen_ids)) == 1000, "duplicate ids found across pages" + assert set(seen_ids) == expected_ids, "some ids were missed or wrong" + assert seen_ids == sorted(seen_ids), "ids are not in sorted order" + + +# --------------------------------------------------------------------------- +# JSONL envelope shape +# --------------------------------------------------------------------------- + + +def test_jsonl_list_claims_returns_paged_envelope(tmp_path: Path, monkeypatch) -> None: + """kb.list_claims must return {items, _meta} not a bare list.""" + from vouch.jsonl_server import handle_request + store = _store(tmp_path) + sid = _src(store) + for i in range(3): + _claim(store, f"c-{i:03}", sid) + monkeypatch.chdir(store.root) + resp = handle_request({"id": "1", "method": "kb.list_claims", "params": {}}) + assert resp["ok"] + result = resp["result"] + assert "items" in result + assert "_meta" in result + assert isinstance(result["items"], list) + assert "next_cursor" in result["_meta"] + assert "total" in result["_meta"] + + +def test_jsonl_list_pages_cursor_walks_all(tmp_path: Path, monkeypatch) -> None: + """Cursor-walking via JSONL must return all pages exactly once.""" + from vouch.jsonl_server import handle_request + store = _store(tmp_path) + sid = _src(store) + for i in range(5): + _page(store, f"pg-{i:03}", sid) + monkeypatch.chdir(store.root) + seen = [] + cursor = None + while True: + params: dict = {"limit": 2} + if cursor: + params["cursor"] = cursor + resp = handle_request({"id": "1", "method": "kb.list_pages", + "params": params}) + assert resp["ok"] + result = resp["result"] + seen.extend(p["id"] for p in result["items"]) + cursor = result["_meta"]["next_cursor"] + if cursor is None: + break + assert len(seen) == 5 + assert len(set(seen)) == 5