Bound cache-info snapshot IDs; add pagination

docs/cache-integration.md

@@ -61,6 +61,13 @@ match the indexed selection, even with an explicit `--db`. Without a scope,
 metadata only: it does not verify source checkpoints or payloads, and remains
 usable for inspecting a stale index before rebuilding. Its JSON explicitly
 marks both source-verification flags false; it is not a freshness claim.
+The `reposition.cache-info.v2` response gives `indexed_snapshot_count` and
+omits `indexed_snapshots` by default. Consumers of the v1 full list must opt
+into pagination. Use `--snapshot-ids-limit N` (maximum 100) to request IDs in
+sorted order, and pass the returned
+`indexed_snapshots_next_offset` as `--snapshot-ids-offset` for the next page.
+A null next offset means the list is complete. The manifest still retains the
+complete checksummed set of IDs.
 
 ## Scope and authority
 

src/reposition/cache_index.py

@@ -1062,19 +1062,41 @@ def retrieve(
         self._scope(source, snapshot, corpus)
         return output
 
-    def info(self, source: TriageCache, *, snapshot: str | None = None, corpus: str | None = None):
+    def info(
+        self,
+        source: TriageCache,
+        *,
+        snapshot: str | None = None,
+        corpus: str | None = None,
+        snapshot_ids_limit: int = 0,
+        snapshot_ids_offset: int = 0,
+    ):
         """Inspect identity-bound index metadata without auditing source payloads or freshness."""
+        _limit(snapshot_ids_limit, "snapshot_ids_limit", 100, 0)
+        _limit(snapshot_ids_offset, "snapshot_ids_offset", 2**63 - 1, 0)
+        if snapshot_ids_offset and not snapshot_ids_limit:
+            raise ValueError("snapshot_ids_offset requires snapshot_ids_limit")
         self._identity(source)
         if snapshot is not None or corpus is not None:
             self._selection(snapshot, corpus)
-        return {
-            "schema": "reposition.cache-info.v1",
-            **self.manifest,
+        snapshots = self.manifest["indexed_snapshots"]
+        page_end = min(len(snapshots), snapshot_ids_offset + snapshot_ids_limit)
+        result = {
+            "schema": "reposition.cache-info.v2",
+            **{key: value for key, value in self.manifest.items() if key != "indexed_snapshots"},
+            "indexed_snapshot_count": len(snapshots),
             "index_bytes": self.path.stat().st_size,
             "source_checkpoint_verified": False,
             "source_payloads_verified": False,
             "requests": 0,
         }
+        if snapshot_ids_limit:
+            result.update(
+                indexed_snapshots=snapshots[snapshot_ids_offset:page_end],
+                indexed_snapshots_offset=snapshot_ids_offset,
+                indexed_snapshots_next_offset=page_end if page_end < len(snapshots) else None,
+            )
+        return result
 
     def close(self):
         self.connection.close()

src/reposition/cli.py

@@ -121,6 +121,15 @@ def parser() -> argparse.ArgumentParser:
     cache_retrieve.add_argument(
         "--checkpoint", required=True, help="index checkpoint from discovery"
     )
+    cache_info.add_argument(
+        "--snapshot-ids-limit",
+        type=int,
+        default=0,
+        help="include up to 100 indexed snapshot IDs; default returns only their count",
+    )
+    cache_info.add_argument(
+        "--snapshot-ids-offset", type=int, default=0, help="starting offset for snapshot IDs"
+    )
     cache_retrieve.add_argument(
         "--byte-offset",
         type=int,
@@ -226,7 +235,13 @@ def main(argv: list[str] | None = None) -> int:
                             byte_offset=args.byte_offset,
                         )
                     else:
-                        output = index.info(source, snapshot=args.snapshot, corpus=args.corpus)
+                        output = index.info(
+                            source,
+                            snapshot=args.snapshot,
+                            corpus=args.corpus,
+                            snapshot_ids_limit=args.snapshot_ids_limit,
+                            snapshot_ids_offset=args.snapshot_ids_offset,
+                        )
             sys.stdout.write(encode_response(output))
             return 0
         if args.command == "index":

src/reposition/projections.py

@@ -219,7 +219,7 @@ def project(
             row.get("data") if component == "checks" and isinstance(row.get("data"), dict) else row
         )
         record_id = source_row.get("node_id", source_row.get("id"))
-        if not isinstance(record_id, (str, int)) or isinstance(record_id, bool):
+        if not isinstance(record_id, str | int) or isinstance(record_id, bool):
             record_id = None
         filename = row.get("filename", row.get("path"))
         if not isinstance(filename, str):

tests/test_cache.py

@@ -290,7 +290,11 @@ def test_metadata_inspection_checks_identity_and_scope_without_a_source_audit(se
         with contextlib.redirect_stdout(io.StringIO()) as output:
             self.assertEqual(main(arguments), 0)
         info = json.loads(output.getvalue())
+        self.assertEqual(info["schema"], "reposition.cache-info.v2")
         self.assertEqual(info["repository"], self.manifest["repository"])
+        self.assertEqual(info["indexed_snapshot_count"], 1)
+        self.assertNotIn("indexed_snapshots", info)
+        self.assertNotIn("indexed_snapshots_next_offset", info)
         self.assertFalse(info["source_checkpoint_verified"])
         self.assertFalse(info["source_payloads_verified"])
         with contextlib.redirect_stderr(io.StringIO()):
@@ -315,6 +319,52 @@ def test_metadata_inspection_checks_identity_and_scope_without_a_source_audit(se
                 (self.root / "cache.json").write_text(canonical(foreign))
                 self.assertEqual(main(arguments), 2)
 
+    def test_metadata_snapshot_ids_are_bounded_and_paginated(self):
+        snapshots = [f"{number:064x}" for number in range(4932)]
+        with sqlite3.connect(self.database) as connection:
+            manifest = json.loads(connection.execute("SELECT data FROM manifest").fetchone()[0])
+            manifest.pop("checksum")
+            manifest["indexed_snapshots"] = snapshots
+            manifest["checksum"] = sha256(canonical(manifest))
+            connection.execute("UPDATE manifest SET data=?", (canonical(manifest),))
+
+        arguments = ["cache-info", "--cache", str(self.root), "--db", str(self.database)]
+
+        def inspect(*flags):
+            with contextlib.redirect_stdout(io.StringIO()) as output:
+                self.assertEqual(main([*arguments, *flags]), 0)
+            return json.loads(output.getvalue()), len(output.getvalue().encode("utf-8"))
+
+        default, size = inspect()
+        self.assertEqual(default["indexed_snapshot_count"], len(snapshots))
+        self.assertNotIn("indexed_snapshots", default)
+        self.assertNotIn("indexed_snapshots_next_offset", default)
+        self.assertLess(size, 10000)
+
+        first, size = inspect("--snapshot-ids-limit", "100")
+        self.assertEqual(first["indexed_snapshots"], snapshots[:100])
+        self.assertEqual(first["indexed_snapshots_next_offset"], 100)
+        self.assertLess(size, 20000)
+
+        second, _ = inspect(
+            "--snapshot-ids-limit", "100", "--snapshot-ids-offset", str(first["indexed_snapshots_next_offset"])
+        )
+        self.assertEqual(second["indexed_snapshots"], snapshots[100:200])
+        self.assertEqual(second["indexed_snapshots_next_offset"], 200)
+
+        last, _ = inspect("--snapshot-ids-limit", "100", "--snapshot-ids-offset", "4900")
+        self.assertEqual(last["indexed_snapshots"], snapshots[4900:])
+        self.assertIsNone(last["indexed_snapshots_next_offset"])
+
+        for flags in (
+            ("--snapshot-ids-limit", "101"),
+            ("--snapshot-ids-limit", "-1"),
+            ("--snapshot-ids-offset", "1"),
+            ("--snapshot-ids-limit", "10", "--snapshot-ids-offset", "-1"),
+        ):
+            with self.subTest(flags=flags), contextlib.redirect_stderr(io.StringIO()):
+                self.assertEqual(main([*arguments, *flags]), 2)
+
     def test_failed_build_keeps_previous_index_and_cleans_temporaries(self):
         before = self.database.read_bytes()
         for limits in ({"max_units": 1}, {"max_index_bytes": 65536}):