from __future__ import annotations import json import shutil import tempfile import unittest from pathlib import Path from jsonschema import Draft202012Validator from docforge.changesets import ChangesetStore from docforge.errors import DocForgeError from docforge.mcp_server import DocForgeService from docforge.pagination import decode_cursor, encode_cursor from docforge.project import Project, project_root_fingerprint ROOT = Path(__file__).resolve().parents[1] FIXTURES = ROOT / "tests" / "fixtures" class PaginationTests(unittest.TestCase): def copy_fixture(self, destination: Path, *, max_tool_chars: int | None = None) -> Path: root = destination / "alpha" shutil.copytree(FIXTURES / "alpha", root) if max_tool_chars is not None: descriptor = root / ".docforge" / "project.toml" text = descriptor.read_text(encoding="utf-8") text = text.replace( "max_changeset_bytes = 100000", (f"max_changeset_bytes = 100000\nmax_tool_output_chars = {max_tool_chars}"), ) descriptor.write_text(text, encoding="utf-8") return root def test_cursor_is_canonical_corruption_detecting_and_binding_bound(self) -> None: binding = {"project_id": "alpha", "source_hash": "a" * 64} cursor = encode_cursor(kind="context.items", binding=binding, position=2) self.assertEqual( 2, decode_cursor( cursor, kind="context.items", binding=binding, total_count=4, ), ) with self.assertRaises(DocForgeError) as corrupt: decode_cursor( f"{cursor[:-1]}{'A' if cursor[-1] != 'A' else 'B'}", kind="context.items", binding=binding, total_count=4, ) self.assertEqual("invalid_cursor", corrupt.exception.code) with self.assertRaises(DocForgeError) as foreign: decode_cursor( cursor, kind="context.items", binding={**binding, "source_hash": "b" * 64}, total_count=4, ) self.assertEqual("stale_cursor", foreign.exception.code) with self.assertRaises(DocForgeError) as wrong_operation: decode_cursor( cursor, kind="changeset.list", binding=binding, total_count=4, ) self.assertEqual("invalid_cursor", wrong_operation.exception.code) def test_context_pages_entries_then_omissions_without_loss(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture(Path(directory)) project = Project.open(root) source_hash = "a" * 64 def context_provider( _index: object, profile: str, budget: int | None ) -> dict[str, object]: return { "status": "ok", "project_id": project.descriptor.project_id, "project_root_fingerprint": project_root_fingerprint(root), "adapter": project.descriptor.adapter, "revision": "test-revision", "source_hash": source_hash, "profile": profile, "budget": budget, "estimated_tokens": 3, "entries": [ { "node_id": "guide.one", "reason": "required", "estimated_tokens": 1, "text": "one", }, { "node_id": "guide.two", "reason": "eligible", "estimated_tokens": 2, "text": "two", }, ], "omissions": [ {"node_id": "guide.three", "reason": "token budget"}, {"node_id": "guide.four", "reason": "token budget"}, ], } service = DocForgeService(project, context_provider=context_provider) result_validator = Draft202012Validator( json.loads((ROOT / "schemas" / "result.schema.json").read_text(encoding="utf-8")) ) cursor: str | None = None evidence: list[tuple[str, str]] = [] limits = [1, 2, 1] page_index = 0 while True: page = service.context( "active", 600, limit=limits[min(page_index, len(limits) - 1)], cursor=cursor, ) result_validator.validate(page) evidence.extend( ("entry", str(item["node_id"])) for item in page["entries"] if isinstance(item, dict) ) evidence.extend( ("omission", str(item["node_id"])) for item in page["omissions"] if isinstance(item, dict) ) pagination = page["pagination"] self.assertIsInstance(pagination, dict) cursor = pagination["next_cursor"] page_index += 1 if cursor is None: break self.assertEqual( [ ("entry", "guide.one"), ("entry", "guide.two"), ("omission", "guide.three"), ("omission", "guide.four"), ], evidence, ) def test_context_oversized_item_advances_as_a_bounded_omission(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture(Path(directory), max_tool_chars=4_000) project = Project.open(root) def context_provider( _index: object, profile: str, budget: int | None ) -> dict[str, object]: return { "status": "ok", "project_id": project.descriptor.project_id, "project_root_fingerprint": project_root_fingerprint(root), "adapter": project.descriptor.adapter, "revision": "test-revision", "source_hash": "a" * 64, "profile": profile, "budget": budget, "estimated_tokens": 10_000, "entries": [ { "node_id": "guide.oversized", "reason": "required", "estimated_tokens": 10_000, "text": "x" * 20_000, } ], "omissions": [], } result = DocForgeService(project, context_provider=context_provider).context( "active", 600, limit=1, ) self.assertEqual("ok", result["status"]) self.assertEqual([], result["entries"]) self.assertEqual("response size limit", result["omissions"][0]["reason"]) self.assertLess(len(json.dumps(result, separators=(",", ":"))), 4_000) self.assertFalse(result["pagination"]["has_more"]) def test_page_limits_reject_boolean_zero_and_policy_overflow(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture(Path(directory)) project = Project.open(root) store = ChangesetStore(project, "alpha-editor") store.create("bounded-page") for limit in (True, 0, project.descriptor.limits.max_results + 1): with self.subTest(limit=limit), self.assertRaises(DocForgeError) as invalid: store.list_changesets(limit=limit) self.assertEqual("invalid_limit", invalid.exception.code) def test_changeset_pages_preserve_full_direct_defaults_and_exact_order(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture(Path(directory)) project = Project.open(root) store = ChangesetStore(project, "alpha-editor") created = store.create("paged-operations") first = store.propose_update( changeset_id="paged-operations", expected_changeset_hash=str(created["changeset_hash"]), node_id="guide.foundation", expected_content_hash=next( node.content_hash for node in project.load().nodes if node.node_id == "guide.foundation" ), metadata={"summary": "First paged update."}, content=None, relationship_changes=[], rationale="First page.", ) store.propose_update( changeset_id="paged-operations", expected_changeset_hash=str(first["changeset_hash"]), node_id="proof.validation", expected_content_hash=next( node.content_hash for node in project.load().nodes if node.node_id == "proof.validation" ), metadata={"summary": "Second paged update."}, content=None, relationship_changes=[], rationale="Second page.", ) full = store.inspect("paged-operations") self.assertNotIn("pagination", full) first_page = store.inspect("paged-operations", limit=1) second_page = store.inspect( "paged-operations", limit=2, cursor=str(first_page["pagination"]["next_cursor"]), ) combined = [*first_page["operations"], *second_page["operations"]] self.assertEqual(full["operations"], combined) self.assertEqual(full["changeset_hash"], second_page["changeset_hash"]) full_diff = store.diff("paged-operations") diff_page = store.diff("paged-operations", limit=1) diff_next = store.diff( "paged-operations", limit=1, cursor=str(diff_page["pagination"]["next_cursor"]), ) self.assertEqual( full_diff["changes"], [*diff_page["changes"], *diff_next["changes"]], ) def test_changeset_list_cursor_stales_when_collection_changes(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture(Path(directory)) store = ChangesetStore(Project.open(root), "alpha-editor") store.create("page-a") store.create("page-b") first = store.list_changesets(limit=1) cursor = str(first["pagination"]["next_cursor"]) second = store.list_changesets(limit=2, cursor=cursor) self.assertEqual( ["page-a", "page-b"], [ *( item["changeset_id"] for item in first["changesets"] if isinstance(item, dict) ), *( item["changeset_id"] for item in second["changesets"] if isinstance(item, dict) ), ], ) store.create("page-c") with self.assertRaises(DocForgeError) as stale: store.list_changesets(limit=1, cursor=cursor) self.assertEqual("stale_cursor", stale.exception.code) def test_oversized_diff_is_reconstructable_from_hash_bound_chunks(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture(Path(directory), max_tool_chars=20_000) project = Project.open(root) store = ChangesetStore(project, "alpha-editor") created = store.create("chunked-diff") store.propose_update( changeset_id="chunked-diff", expected_changeset_hash=str(created["changeset_hash"]), node_id="guide.foundation", expected_content_hash=next( node.content_hash for node in project.load().nodes if node.node_id == "guide.foundation" ), metadata=None, content="replacement " * 4_000, relationship_changes=[], rationale="Exercise deterministic chunk transport.", ) full = store.diff("chunked-diff") cursor: str | None = None chunks: list[str] = [] while True: page = store.diff("chunked-diff", limit=20, cursor=cursor) self.assertEqual("canonical_json_chunk", page["result_mode"]) chunks.append(page["chunk"]["content"]) cursor = page["pagination"]["next_cursor"] if cursor is None: break reconstructed = json.loads("".join(chunks)) self.assertEqual( { "changes": full["changes"], "operations": full["operations"], }, reconstructed, ) if __name__ == "__main__": unittest.main()