feat(list): cursor pagination for all kb.list_* methods

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

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]:

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