from __future__ import annotations import hashlib import importlib import sqlite3 import sys import tempfile import unittest from collections.abc import Mapping from contextlib import closing from dataclasses import replace from pathlib import Path from mcp.shared.memory import create_connected_server_and_client_session from docforge.adapter_contract import ( AdapterAssembly, AdapterEdge, AdapterImplementation, AdapterManifest, AdapterNode, AdapterProject, AdapterProjection, AdapterProjectSettings, AdapterSource, AdapterSourceProjection, ShadowArtifact, compare_artifacts, validate_projection, ) from docforge.errors import DocForgeError from docforge.index import ProjectIndex from docforge.mcp_server import ( ALL_TOOLS, READ_TOOLS, create_project_server, create_read_only_server, ) from docforge.models import ( Edge, LogicEdge, LogicNode, LogicProjection, Node, ProjectSnapshot, ProposalWriter, RenderConfig, RenderView, ) from docforge.viewer_manager import ViewerManagerClient from docforge.visualization import VisualizationIndexSnapshot class Loader: def __init__(self, projection: AdapterProjection) -> None: self.projection = projection def load_projection(self) -> AdapterProjection: return self.projection class IncrementalLoader: def __init__(self, root: Path) -> None: self.root = root self.sources = { "guide.foundation": "Foundation content.", "guide.workflow": "Workflow content.", } self.source_paths = { "guide.foundation": "docs/foundation.md", "guide.workflow": "docs/workflow.md", } self.workflow_dependencies = ("guide.foundation",) self.extract_calls: list[str] = [] self.fail_source: str | None = None @staticmethod def _hash(value: str) -> str: return hashlib.sha256(value.encode()).hexdigest() def load_manifest(self) -> AdapterManifest: source_items = tuple( AdapterSource( source_id=source_id, source_path=self.source_paths[source_id], fingerprint=self._hash(content), extractor_version="python-ast@1", dependencies=(self.workflow_dependencies if source_id == "guide.workflow" else ()), ) for source_id, content in sorted(self.sources.items()) ) digest = hashlib.sha256() for source in source_items: digest.update(source.source_id.encode()) digest.update(source.source_path.encode()) digest.update(source.fingerprint.encode()) digest.update(source.extractor_version.encode()) for dependency in source.dependencies: digest.update(dependency.encode()) return AdapterManifest( project_id="incremental-fixture", title="Incremental fixture", adapter_id="fixture-incremental", adapter_version="1", root=self.root, revision=digest.hexdigest()[:12], source_hash=digest.hexdigest(), families=("guide",), allowed_relations=("depends_on",), sources=source_items, estimated_nodes=10, ) def extract_source(self, source: AdapterSource) -> AdapterSourceProjection: self.extract_calls.append(source.source_id) if self.fail_source == source.source_id: raise RuntimeError("intentional extraction failure") content = self.sources[source.source_id] node = Node( node_id=source.source_id, title=source.source_id.rsplit(".", 1)[-1].title(), family="guide", authority="authoritative", status="active", tags=("guide",), summary=f"{source.source_id} summary.", content=content, source_path=source.source_path, source_anchor=None, content_hash=self._hash(content), ) edges = ( (AdapterEdge(Edge("guide.workflow", "depends_on", "guide.foundation")),) if source.source_id == "guide.workflow" else () ) logic = ( ( LogicProjection( owner_node_id="guide.workflow", source_id="guide.workflow", nodes=( LogicNode("entry", "entry", "Entry", None), LogicNode("return", "return", "Return", None), ), edges=(LogicEdge("entry", "return", "return", "RETURN", 0),), ), ) if source.source_id == "guide.workflow" else () ) return AdapterSourceProjection( source_id=source.source_id, fingerprint=source.fingerprint, nodes=(AdapterNode(node),), edges=edges, logic=logic, ) def load_projection(self) -> AdapterProjection: manifest = self.load_manifest() contributions = tuple(self.extract_source(source) for source in manifest.sources) return AdapterProjection( project_id=manifest.project_id, title=manifest.title, adapter_id=manifest.adapter_id, adapter_version=manifest.adapter_version, root=manifest.root, revision=manifest.revision, source_hash=manifest.source_hash, nodes=tuple( sorted( (node for contribution in contributions for node in contribution.nodes), key=lambda item: item.node.node_id, ) ), edges=tuple( sorted( (edge for contribution in contributions for edge in contribution.edges), key=lambda item: ( item.edge.source_id, item.edge.relation, item.edge.target_id, ), ) ), ) class OverlappingIncrementalLoader(IncrementalLoader): def extract_source(self, source: AdapterSource) -> AdapterSourceProjection: self.extract_calls.append(source.source_id) content = self.sources[source.source_id] shared = Node( node_id="guide.shared", title="Shared", family="guide", authority="derived", status="active", tags=("guide",), summary=f"Evidence selected from {source.source_id}.", content=content, source_path=source.source_path, source_anchor=None, content_hash=self._hash(content), ) return AdapterSourceProjection( source_id=source.source_id, fingerprint=source.fingerprint, nodes=(AdapterNode(shared),), edges=(), ) def assemble_projection( self, manifest: AdapterManifest, contributions: tuple[AdapterSourceProjection, ...], ) -> AdapterAssembly: selected = min(contributions, key=lambda item: item.source_id) return AdapterAssembly( AdapterProjection( project_id=manifest.project_id, title=manifest.title, adapter_id=manifest.adapter_id, adapter_version=manifest.adapter_version, root=manifest.root, revision=manifest.revision, source_hash=manifest.source_hash, nodes=selected.nodes, edges=(), ) ) def load_projection(self) -> AdapterProjection: manifest = self.load_manifest() contributions = tuple(self.extract_source(source) for source in manifest.sources) return self.assemble_projection(manifest, contributions).projection class NonLogicIncrementalLoader(IncrementalLoader): """Exercise non-AST incremental caching without publishing function Logic.""" def extract_source(self, source: AdapterSource) -> AdapterSourceProjection: return replace(super().extract_source(source), logic=()) class AdapterContractTests(unittest.TestCase): def projection(self, root: Path) -> AdapterProjection: foundation = Node( node_id="guide.foundation", title="Foundation", family="guide", authority="authoritative", status="active", tags=("guide",), summary="The base contract.", content="Foundation content.", source_path="docs/foundation.md", source_anchor=None, content_hash="1" * 64, ) workflow = Node( node_id="guide.workflow", title="Workflow", family="guide", authority="approved_plan", status="planned", tags=("guide", "workflow"), summary="The editing workflow.", content="Workflow content.", source_path="docs/workflow.md", source_anchor=None, content_hash="2" * 64, ) return AdapterProjection( project_id="adapter-fixture", title="Adapter fixture", adapter_id="fixture-shadow", adapter_version="1", root=root, revision="fixture-revision", source_hash="3" * 64, nodes=( AdapterNode(foundation, (("acceptance", "proven"),)), AdapterNode(workflow, (("acceptance", "pending"),)), ), edges=( AdapterEdge( Edge("guide.workflow", "depends_on", "guide.foundation"), (("source", "docs/workflow.md"),), ), ), ) def test_adapter_projection_builds_and_checks_through_standard_index(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() projection = self.projection(root) project = AdapterProject(Loader(projection), cache_root=root / ".cache" / "shadow") index = ProjectIndex(project) built = index.build() checked = index.check() self.assertEqual("fixture-shadow@1", built["adapter"]) self.assertEqual(2, checked["node_count"]) self.assertEqual("Workflow", index.get_node("guide.workflow")["node"]["title"]) self.assertEqual(projection.identity(), projection.identity()) def test_projection_rejects_unsorted_metadata_graph_and_identity_changes(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() projection = self.projection(root) invalid_metadata = replace( projection, nodes=( replace( projection.nodes[0], metadata=(("z", "last"), ("a", "first")), ), projection.nodes[1], ), ) with self.assertRaisesRegex(DocForgeError, "metadata keys"): validate_projection(invalid_metadata) invalid_source = replace( projection, nodes=( replace( projection.nodes[0], node=replace(projection.nodes[0].node, source_path="../outside.md"), ), projection.nodes[1], ), ) with self.assertRaisesRegex(DocForgeError, "source path"): validate_projection(invalid_source) broken = replace( projection, edges=(AdapterEdge(Edge("guide.workflow", "depends_on", "missing.node")),), ) with self.assertRaisesRegex(DocForgeError, "missing nodes"): validate_projection(broken) loader = Loader(projection) project = AdapterProject(loader, cache_root=root / ".cache" / "shadow") loader.projection = replace(projection, adapter_version="2") with self.assertRaisesRegex(DocForgeError, "identity changed"): project.load() def test_adapter_cache_must_remain_inside_project(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() outside = root.parent / "outside-adapter-cache" with self.assertRaisesRegex(DocForgeError, "confined"): AdapterProject(Loader(self.projection(root)), cache_root=outside) def test_adapter_implementation_changes_require_a_process_restart(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() implementation_root = root / "adapter" implementation_root.mkdir() implementation = implementation_root / "loader.py" implementation.write_text("VERSION = 1\n", encoding="utf-8") ignored = implementation_root / "loader.pyc" ignored.write_bytes(b"derived") project = AdapterProject( Loader(self.projection(root)), cache_root=root / ".cache" / "shadow", settings=AdapterProjectSettings( implementation=AdapterImplementation( roots=(implementation_root,), suffixes=(".py",), ) ), ) ProjectIndex(project).build() ignored.write_bytes(b"changed derived state") project.validate_runtime() implementation.write_text("VERSION = 2\n", encoding="utf-8") with self.assertRaises(DocForgeError) as captured: project.validate_runtime() self.assertEqual("adapter_restart_required", captured.exception.code) self.assertEqual(["adapter/loader.py"], captured.exception.details["changed"]) self.assertEqual([], captured.exception.details["added"]) self.assertEqual([], captured.exception.details["deleted"]) self.assertEqual(1, captured.exception.details["changed_count"]) self.assertFalse(captured.exception.details["paths_truncated"]) def test_adapter_implementation_additions_and_deletions_require_a_restart(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() implementation_root = root / "adapter" implementation_root.mkdir() original = implementation_root / "loader.py" original.write_text("VERSION = 1\n", encoding="utf-8") project = AdapterProject( Loader(self.projection(root)), cache_root=root / ".cache" / "shadow", settings=AdapterProjectSettings( implementation=AdapterImplementation( roots=(implementation_root,), suffixes=(".py",), ) ), ) added = implementation_root / "helpers.py" added.write_text("VALUE = 1\n", encoding="utf-8") with self.assertRaises(DocForgeError) as addition: project.load() self.assertEqual("adapter_restart_required", addition.exception.code) self.assertEqual(["adapter/helpers.py"], addition.exception.details["added"]) with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() implementation_root = root / "adapter" implementation_root.mkdir() original = implementation_root / "loader.py" original.write_text("VERSION = 1\n", encoding="utf-8") project = AdapterProject( Loader(self.projection(root)), cache_root=root / ".cache" / "shadow", settings=AdapterProjectSettings( implementation=AdapterImplementation( roots=(implementation_root,), suffixes=(".py",), ) ), ) original.unlink() with self.assertRaises(DocForgeError) as deletion: project.incremental_state() self.assertEqual("adapter_restart_required", deletion.exception.code) self.assertEqual(["adapter/loader.py"], deletion.exception.details["deleted"]) def test_adapter_implementation_is_inferred_from_a_project_local_package(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() package = root / "adapter_fixture_dynamic" package.mkdir() (package / "__init__.py").write_text("", encoding="utf-8") (package / "loader.py").write_text( "class Loader:\n" " def __init__(self, projection):\n" " self.projection = projection\n" " def load_projection(self):\n" " return self.projection\n", encoding="utf-8", ) sys.path.insert(0, str(root)) try: module = importlib.import_module("adapter_fixture_dynamic.loader") project = AdapterProject( module.Loader(self.projection(root)), cache_root=root / ".cache" / "shadow", ) (package / "helper.py").write_text("VALUE = 1\n", encoding="utf-8") with self.assertRaises(DocForgeError) as captured: project.validate_runtime() finally: sys.path.remove(str(root)) sys.modules.pop("adapter_fixture_dynamic.loader", None) sys.modules.pop("adapter_fixture_dynamic", None) self.assertEqual("adapter_restart_required", captured.exception.code) self.assertEqual( ["adapter_fixture_dynamic/helper.py"], captured.exception.details["added"], ) def test_adapter_descriptor_changes_require_a_restart(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() descriptor_root = root / ".docforge" descriptor_root.mkdir() descriptor = descriptor_root / "project.toml" descriptor.write_text("adapter_version = 1\n", encoding="utf-8") project = AdapterProject( Loader(self.projection(root)), cache_root=root / ".cache" / "shadow", settings=AdapterProjectSettings(descriptor_path=descriptor), ) descriptor.write_text("adapter_version = 2\n", encoding="utf-8") with self.assertRaises(DocForgeError) as captured: project.validate_runtime() self.assertEqual("adapter_restart_required", captured.exception.code) self.assertEqual( [".docforge/project.toml"], captured.exception.details["changed"], ) def test_incremental_adapter_reuses_sources_and_invalidates_reverse_dependencies(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() loader = IncrementalLoader(root) project = AdapterProject(loader, cache_root=root / ".cache" / "incremental") index = ProjectIndex(project) first = index.build() self.assertEqual(2, first["build"]["reparsed_sources"]) self.assertEqual(0, first["build"]["cache_hits"]) self.assertEqual(1, first["logic_projection_count"]) logic = index.get_logic("guide.workflow") self.assertTrue(logic["available"]) self.assertEqual("guide.workflow", logic["projection"]["owner_node_id"]) self.assertEqual(2, len(logic["projection"]["nodes"])) self.assertFalse(index.get_logic("guide.foundation")["available"]) visual_logic = VisualizationIndexSnapshot(index, index.check()).logic("guide.workflow") self.assertTrue(visual_logic["available"]) self.assertEqual("entry", visual_logic["root"]) self.assertEqual("return", visual_logic["edges"][0]["relation"]) loader.extract_calls.clear() cache_path = root / ".cache" / "incremental" / "extractions.json" cache_modified = cache_path.stat().st_mtime_ns second = index.build() self.assertEqual([], loader.extract_calls) self.assertEqual(0, second["build"]["reparsed_sources"]) self.assertEqual(2, second["build"]["cache_hits"]) self.assertEqual(cache_modified, cache_path.stat().st_mtime_ns) loader.sources["guide.foundation"] = "Changed foundation." loader.extract_calls.clear() with self.assertRaisesRegex(DocForgeError, "does not match"): index.check() self.assertEqual([], loader.extract_calls) changed = index.build() self.assertEqual( ["guide.foundation", "guide.workflow"], loader.extract_calls, ) self.assertEqual(2, changed["build"]["invalidated_sources"]) self.assertEqual( "Changed foundation.", index.get_node("guide.foundation")["node"]["content"], ) loader.source_paths["guide.workflow"] = "docs/workflow-renamed.md" loader.workflow_dependencies = () loader.extract_calls.clear() renamed = index.build() self.assertEqual(["guide.workflow"], loader.extract_calls) self.assertEqual(1, renamed["build"]["invalidated_sources"]) self.assertEqual( "docs/workflow-renamed.md", index.get_node("guide.workflow")["node"]["source_path"], ) def test_no_ast_index_policy_rejects_logic_publication(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() project = AdapterProject( IncrementalLoader(root), cache_root=root / ".cache" / "no-ast", ) with self.assertRaises(DocForgeError) as captured: ProjectIndex(project, allow_logic=False).build() self.assertEqual("adapter_policy_forbids_logic", captured.exception.code) self.assertFalse(project.descriptor.index_path.exists()) def test_no_ast_index_accepts_legacy_and_non_logic_incremental_adapters(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() legacy = AdapterProject( Loader(self.projection(root)), cache_root=root / ".cache" / "legacy-no-ast", ) legacy_index = ProjectIndex(legacy, allow_logic=False) self.assertEqual(2, legacy_index.build()["node_count"]) self.assertEqual( "guide.workflow", legacy_index.get_node("guide.workflow")["node"]["node_id"], ) with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() loader = NonLogicIncrementalLoader(root) project = AdapterProject( loader, cache_root=root / ".cache" / "incremental-no-ast", ) index = ProjectIndex(project, allow_logic=False) first = index.build() self.assertEqual(2, first["build"]["reparsed_sources"]) loader.extract_calls.clear() second = index.build() self.assertEqual([], loader.extract_calls) self.assertEqual(2, second["build"]["cache_hits"]) self.assertEqual(0, second["build"]["reparsed_sources"]) self.assertEqual(0, second["logic_projection_count"]) def test_no_ast_rejects_preexisting_logic_index_and_viewer_snapshot(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() project = AdapterProject( IncrementalLoader(root), cache_root=root / ".cache" / "preexisting-logic", ) ProjectIndex(project).build() preserved = ProjectIndex(project, allow_logic=False) with self.assertRaises(DocForgeError) as checked: preserved.check(verify_rows=False) self.assertEqual("adapter_policy_forbids_logic", checked.exception.code) with self.assertRaises(DocForgeError) as viewed: ViewerManagerClient( preserved, state_path=root / ".cache" / "viewer-state.json", ).start() self.assertEqual("adapter_policy_forbids_logic", viewed.exception.code) def test_fast_incremental_reads_reverify_a_changed_index_file(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() loader = IncrementalLoader(root) project = AdapterProject(loader, cache_root=root / ".cache" / "incremental") index = ProjectIndex(project) index.build() index.synchronize() self.assertTrue(index.attestation_path.is_file()) fresh = ProjectIndex(project) self.assertEqual( "current", fresh.synchronize()["synchronization"]["action"], ) with closing(sqlite3.connect(index.path)) as connection: connection.execute( "UPDATE nodes SET content = ? WHERE node_id = ?", ("tampered", "guide.foundation"), ) connection.commit() with self.assertRaisesRegex(DocForgeError, "rows do not match metadata"): index.get_node("guide.foundation") repaired = index.synchronize() self.assertEqual("rebuilt", repaired["synchronization"]["action"]) self.assertEqual( "Foundation content.", index.get_node("guide.foundation")["node"]["content"], ) def test_incremental_delete_failure_and_equivalence_are_safe(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() loader = IncrementalLoader(root) project = AdapterProject(loader, cache_root=root / ".cache" / "incremental") index = ProjectIndex(project) index.build() (root / ".cache" / "incremental" / "extractions.json").write_text( "{broken", encoding="utf-8" ) loader.extract_calls.clear() cache_miss = index.build() self.assertEqual( ["guide.foundation", "guide.workflow"], loader.extract_calls, ) self.assertEqual(2, cache_miss["build"]["reparsed_sources"]) original_index = index.path.read_bytes() cache_path = root / ".cache" / "incremental" / "extractions.json" original_cache = cache_path.read_bytes() loader.sources["guide.foundation"] = "Broken extraction." loader.fail_source = "guide.foundation" with self.assertRaisesRegex(RuntimeError, "intentional"): index.build() self.assertEqual(original_index, index.path.read_bytes()) self.assertEqual(original_cache, cache_path.read_bytes()) loader.fail_source = None index.build() loader.sources.pop("guide.workflow") deleted = index.build() self.assertEqual(1, deleted["build"]["deleted_sources"]) self.assertEqual(1, deleted["node_count"]) self.assertEqual(0, deleted["edge_count"]) self.assertIsNone(project.logic_projection("guide.workflow")) loader.extract_calls.clear() equivalent = project.verify_incremental_equivalence() self.assertEqual("ok", equivalent["status"]) self.assertEqual(1, equivalent["node_count"]) def test_incremental_adapter_can_assemble_overlapping_source_evidence(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() loader = OverlappingIncrementalLoader(root) project = AdapterProject(loader, cache_root=root / ".cache" / "overlap") index = ProjectIndex(project) first = index.build() self.assertEqual(2, first["build"]["reparsed_sources"]) self.assertEqual(1, first["node_count"]) self.assertEqual( "Foundation content.", index.get_node("guide.shared")["node"]["content"], ) loader.extract_calls.clear() warm = index.build() self.assertEqual(2, warm["build"]["cache_hits"]) self.assertEqual([], loader.extract_calls) loader.sources["guide.workflow"] = "Changed overlapping evidence." loader.extract_calls.clear() changed = index.build() self.assertEqual(["guide.workflow"], loader.extract_calls) self.assertEqual(1, changed["build"]["reparsed_sources"]) self.assertEqual(1, changed["node_count"]) self.assertEqual( "Foundation content.", index.get_node("guide.shared")["node"]["content"], ) self.assertEqual("ok", project.verify_incremental_equivalence()["status"]) def test_artifact_comparison_is_complete_and_byte_exact(self) -> None: reference = ( ShadowArtifact("manual", b"same"), ShadowArtifact("timeline", b"old"), ) exact = compare_artifacts(reference, reference) self.assertEqual("ok", exact["status"]) self.assertEqual(2, exact["count"]) mismatch = compare_artifacts( reference, ( ShadowArtifact("timeline", b"new"), ShadowArtifact("extra", b"extra"), ), ) self.assertEqual("mismatch", mismatch["status"]) self.assertEqual(["manual"], mismatch["missing"]) self.assertEqual(["extra"], mismatch["unexpected"]) self.assertEqual(["timeline"], mismatch["changed"]) class AdapterReadOnlyMcpTests(unittest.IsolatedAsyncioTestCase): async def test_mcp_reports_adapter_restart_remediation_without_synchronizing(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() implementation_root = root / "adapter" implementation_root.mkdir() implementation = implementation_root / "loader.py" implementation.write_text("VERSION = 1\n", encoding="utf-8") fixture = AdapterContractTests() project = AdapterProject( Loader(fixture.projection(root)), cache_root=root / ".cache" / "adapter-read-only", settings=AdapterProjectSettings( implementation=AdapterImplementation( roots=(implementation_root,), suffixes=(".py",), ) ), ) ProjectIndex(project).build() server = create_read_only_server(project) implementation.write_text("VERSION = 2\n", encoding="utf-8") async with create_connected_server_and_client_session( server, raise_exceptions=True ) as session: result = await session.call_tool("docforge_project_info", {}) self.assertEqual("error", result.structuredContent["status"]) self.assertEqual( "adapter_restart_required", result.structuredContent["error"]["code"], ) self.assertEqual("stale", result.structuredContent["staleness"]) self.assertEqual( {"retryable": False, "action": "restart_project_server"}, result.structuredContent["error"]["remediation"], ) self.assertNotIn("synchronization", result.structuredContent) async def test_adapter_project_exposes_only_read_tools_and_custom_context(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() fixture = AdapterContractTests() project = AdapterProject( Loader(fixture.projection(root)), cache_root=root / ".cache" / "adapter-read-only", ) index = ProjectIndex(project) index.build() calls: list[tuple[str, int | None]] = [] def context_provider( current: ProjectIndex, profile: str, budget: int | None ) -> dict[str, object]: checked = current.check() calls.append((profile, budget)) return { "status": "ok", "project_id": checked["project_id"], "project_root_fingerprint": checked["project_root_fingerprint"], "revision": checked["revision"], "source_hash": checked["source_hash"], "adapter": checked["adapter"], "profile": profile, "budget": budget, "estimated_tokens": 1, "entries": [], "omissions": [], } server = create_read_only_server(project, context_provider=context_provider) async with create_connected_server_and_client_session( server, raise_exceptions=True ) as session: tools = await session.list_tools() info = await session.call_tool("docforge_project_info", {}) contract = await session.call_tool("docforge_get_contract", {}) context = await session.call_tool( "docforge_get_context", {"profile": "fixture", "budget": 321} ) self.assertEqual(READ_TOOLS, tuple(tool.name for tool in tools.tools)) self.assertEqual("adapter-fixture", info.structuredContent["project_id"]) self.assertEqual(list(READ_TOOLS), contract.structuredContent["allowed_tools"]) self.assertIn( "isolated_changeset_writes", contract.structuredContent["excluded_operations"], ) self.assertFalse(contract.structuredContent["proposal_access"]["enabled"]) self.assertFalse(contract.structuredContent["isolated_changeset_writes_allowed"]) self.assertEqual("fixture", context.structuredContent["profile"]) self.assertEqual([("fixture", 321)], calls) self.assertFalse(project.descriptor.changeset_root.exists()) async def test_adapter_project_proposals_require_explicit_policy_and_stay_isolated( self, ) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() (root / ".docforge").mkdir() descriptor = root / ".docforge" / "project.toml" descriptor.write_text("adapter fixture\n", encoding="utf-8") content_root = root / "docs" content_root.mkdir() sources = (content_root / "foundation.md", content_root / "workflow.md") for source in sources: source.write_text(f"canonical {source.stem}\n", encoding="utf-8") template_root = root / "templates" template_root.mkdir() template = template_root / "review.html" template.write_text( "{{ docforge_title }}" "{{ docforge_content }}\n", encoding="utf-8", ) validation_calls: list[tuple[Mapping[str, object], ...]] = [] def validate_proposal( base: ProjectSnapshot, projected: ProjectSnapshot, operations: tuple[Mapping[str, object], ...], ) -> None: self.assertEqual(base.edges, projected.edges) validation_calls.append(operations) settings = AdapterProjectSettings( descriptor_path=descriptor, content_roots=(content_root,), canonical_sources=sources, changeset_root=root / ".cache" / "changesets", proposal_writers=(ProposalWriter("fixture-writer", ("guide",), ("update",)),), render=RenderConfig( template_root=template_root, preview_root=root / ".cache" / "previews", views=( RenderView( "review", "generic_html", template, root / ".cache" / "rendered" / "review.html", "Adapter review", ("guide",), ), ), ), proposal_validator=validate_proposal, ) project = AdapterProject( Loader(AdapterContractTests().projection(root)), cache_root=root / ".cache" / "adapter", settings=settings, ) ProjectIndex(project).build() before = {source: source.read_bytes() for source in sources} server = create_project_server(project, proposal_writer="fixture-writer") async with create_connected_server_and_client_session( server, raise_exceptions=True ) as session: tools = await session.list_tools() created = await session.call_tool( "docforge_create_changeset", {"changeset_id": "adapter-update"} ) node = await session.call_tool("docforge_get_node", {"node_id": "guide.workflow"}) proposed = await session.call_tool( "docforge_propose_node_update", { "changeset_id": "adapter-update", "expected_changeset_hash": created.structuredContent["changeset_hash"], "node_id": "guide.workflow", "expected_content_hash": node.structuredContent["node"]["content_hash"], "metadata": None, "content": "Updated workflow content.", "relationship_changes": [], "rationale": "Prove adapter proposal policy.", }, ) validated = await session.call_tool( "docforge_validate_changeset", {"changeset_id": "adapter-update"}, ) diff = await session.call_tool( "docforge_get_changeset_diff", {"changeset_id": "adapter-update"}, ) preview = await session.call_tool( "docforge_preview_changeset", {"changeset_id": "adapter-update", "view_id": "review"}, ) self.assertEqual(ALL_TOOLS, tuple(tool.name for tool in tools.tools)) self.assertEqual("ok", proposed.structuredContent["status"]) self.assertTrue(validated.structuredContent["valid"]) self.assertEqual("update", diff.structuredContent["changes"][0]["operation"]) preview_path = root / preview.structuredContent["preview"]["path"] self.assertTrue(preview_path.is_file()) self.assertTrue(validation_calls) self.assertEqual(before, {source: source.read_bytes() for source in sources}) if __name__ == "__main__": unittest.main()