339 lines
14 KiB
Python
339 lines
14 KiB
Python
|
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()
|