from __future__ import annotations import json import os import shutil import sys import tempfile import threading import time import unittest from contextlib import contextmanager from pathlib import Path from unittest import mock from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from mcp.shared.memory import create_connected_server_and_client_session from docforge.changesets import ChangesetStore from docforge.index import ProjectIndex from docforge.mcp_server import ( ALL_TOOLS, APPLICATION_TOOLS, CONTENT_WARNING, PROPOSAL_TOOLS, DocForgeService, _create_bound_server, create_server, ) from docforge.project import Project from docforge.viewer_manager import ViewerManager ROOT = Path(__file__).resolve().parents[1] FIXTURES = ROOT / "tests" / "fixtures" class DocForgeMcpTests(unittest.IsolatedAsyncioTestCase): def copy_fixture(self, name: str, destination: Path) -> Path: root = destination / name shutil.copytree(FIXTURES / name, root) return root @contextmanager def running_manager(self, state_path: Path): manager = ViewerManager(state_path, check_interval_seconds=0.02) thread = threading.Thread(target=manager.serve_forever, daemon=True) previous = os.environ.get("DOCFORGE_VIEWER_MANAGER_STATE") os.environ["DOCFORGE_VIEWER_MANAGER_STATE"] = str(state_path) thread.start() deadline = time.monotonic() + 2 while not state_path.exists() and time.monotonic() < deadline: time.sleep(0.01) self.assertTrue(state_path.exists()) try: yield finally: manager.shutdown() thread.join(timeout=2) if previous is None: os.environ.pop("DOCFORGE_VIEWER_MANAGER_STATE", None) else: os.environ["DOCFORGE_VIEWER_MANAGER_STATE"] = previous async def test_protocol_lists_only_the_fixed_safe_surface(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) ProjectIndex(Project.open(root)).build() async with create_connected_server_and_client_session( create_server(root), raise_exceptions=True ) as session: response = await session.list_tools() names = tuple(tool.name for tool in response.tools) self.assertEqual(ALL_TOOLS, names) self.assertEqual(14, len(PROPOSAL_TOOLS)) tools = {tool.name: tool for tool in response.tools} for name in ( "docforge_backlinks", "docforge_dependencies", "docforge_impact", ): self.assertIn("limit", tools[name].inputSchema["properties"]) self.assertNotIn("limit", tools[name].inputSchema.get("required", [])) for name in ( "docforge_get_context", "docforge_list_changesets", "docforge_get_changeset", "docforge_validate_changeset", "docforge_get_changeset_diff", ): for field in ("limit", "cursor"): self.assertIn(field, tools[name].inputSchema["properties"]) self.assertNotIn(field, tools[name].inputSchema.get("required", [])) self.assertIn( "deep", tools["docforge_render_status"].inputSchema["properties"], ) self.assertFalse( any( token in name for name in names for token in ("apply", "commit", "push", "deploy", "publish", "shell") ) ) async def test_factory_diagnostics_are_additive_through_real_mcp(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) ProjectIndex(Project.open(root)).build() async with create_connected_server_and_client_session( create_server(root, diagnostics=True), raise_exceptions=True, ) as session: result = await session.call_tool( "docforge_get_node", {"node_id": "guide.workflow"}, ) diagnostics = result.structuredContent["diagnostics"] self.assertEqual("mcp.get_node", diagnostics["operation"]) self.assertEqual(0, diagnostics["counters"]["project_loads"]) self.assertEqual(0, diagnostics["counters"]["source_files_parsed"]) async def test_context_pagination_is_complete_and_stale_cursors_fail_closed(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) ProjectIndex(Project.open(root)).build() async with create_connected_server_and_client_session( create_server(root), raise_exceptions=True ) as session: first = await session.call_tool( "docforge_get_context", {"profile": "active", "limit": 1}, ) cursor = first.structuredContent["pagination"]["next_cursor"] pages = [first.structuredContent] while cursor is not None: page = await session.call_tool( "docforge_get_context", { "profile": "active", "limit": 2, "cursor": cursor, }, ) pages.append(page.structuredContent) cursor = page.structuredContent["pagination"]["next_cursor"] stale_cursor = first.structuredContent["pagination"]["next_cursor"] workflow = root / "docs" / "content" / "workflow.md" workflow.write_text( workflow.read_text(encoding="utf-8") + "\nChanged between pages.\n", encoding="utf-8", ) stale = await session.call_tool( "docforge_get_context", { "profile": "active", "limit": 1, "cursor": stale_cursor, }, ) evidence = [ *(("entry", item["node_id"]) for page in pages for item in page["entries"]), *(("omission", item["node_id"]) for page in pages for item in page["omissions"]), ] self.assertEqual(len(evidence), pages[0]["summary"]["evidence_count"]) self.assertEqual(len(evidence), len(set(evidence))) self.assertEqual("stale_cursor", stale.structuredContent["error"]["code"]) self.assertEqual("stale", stale.structuredContent["staleness"]) self.assertEqual( "restart_pagination", stale.structuredContent["error"]["remediation"]["action"], ) async def test_oversized_changeset_reads_return_exact_pages_and_chunks(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) descriptor = root / ".docforge" / "project.toml" descriptor.write_text( descriptor.read_text(encoding="utf-8").replace( "max_changeset_bytes = 100000", "max_changeset_bytes = 100000\nmax_tool_output_chars = 20000", ), encoding="utf-8", ) project = Project.open(root) ProjectIndex(project).build() store = ChangesetStore(project, "alpha-editor") created = store.create("mcp-chunked-diff") foundation = next( node for node in project.load().nodes if node.node_id == "guide.foundation" ) proposed = store.propose_update( changeset_id="mcp-chunked-diff", expected_changeset_hash=str(created["changeset_hash"]), node_id=foundation.node_id, expected_content_hash=foundation.content_hash, metadata=None, content="replacement " * 4_000, relationship_changes=[], rationale="Exercise bounded MCP diff reconstruction.", ) direct = store.diff("mcp-chunked-diff") async with create_connected_server_and_client_session( create_server(root), raise_exceptions=True ) as session: inspected = await session.call_tool( "docforge_get_changeset", {"changeset_id": "mcp-chunked-diff"}, ) validated = await session.call_tool( "docforge_validate_changeset", {"changeset_id": "mcp-chunked-diff"}, ) cursor: str | None = None chunks: list[str] = [] while True: page = await session.call_tool( "docforge_get_changeset_diff", { "changeset_id": "mcp-chunked-diff", "cursor": cursor, }, ) self.assertLessEqual( len(json.dumps(page.structuredContent, separators=(",", ":"))), 20_000, ) chunks.append(page.structuredContent["chunk"]["content"]) cursor = page.structuredContent["pagination"]["next_cursor"] if cursor is None: break self.assertEqual("operation_summaries", inspected.structuredContent["result_mode"]) self.assertEqual("operation_summaries", validated.structuredContent["result_mode"]) self.assertTrue(validated.structuredContent["valid"]) self.assertEqual( proposed["changeset_hash"], validated.structuredContent["changeset_hash"], ) self.assertEqual( { "changes": direct["changes"], "operations": direct["operations"], }, json.loads("".join(chunks)), ) async def test_no_ast_binding_preserves_adapter_and_blocks_logic(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) project = Project.open(root) ProjectIndex(project).build() service = DocForgeService(project, no_ast=True) self.assertIs(service.index, service.application.index) self.assertFalse(service.application.index.allow_logic) async with create_connected_server_and_client_session( create_server(root, no_ast=True), raise_exceptions=True ) as session: bootstrap = await session.call_tool("docforge_bootstrap", {}) contract = await session.call_tool("docforge_get_contract", {}) logic = await session.call_tool( "docforge_get_logic", {"owner_node_id": "guide.workflow"} ) policy = bootstrap.structuredContent["adapter_policy"] self.assertEqual("preserve-no-ast", policy["mode"]) self.assertEqual("forbidden", policy["ast_analysis"]) self.assertEqual("forbidden", policy["logic_projection"]) self.assertEqual("allowed", policy["incremental_extraction"]) self.assertEqual(["docforge_get_logic"], policy["blocked_tools"]) self.assertEqual( policy, bootstrap.structuredContent["binding"]["adapter_policy"], ) self.assertIn( "preserve the current adapter", bootstrap.structuredContent["recommended_workflow"][1], ) self.assertEqual(policy, contract.structuredContent["adapter_policy"]) self.assertIn( "adapter_ast_upgrade", contract.structuredContent["excluded_operations"], ) self.assertEqual("error", logic.structuredContent["status"]) self.assertEqual( "adapter_policy_forbids_logic", logic.structuredContent["error"]["code"], ) async def test_every_read_tool_returns_scoped_structured_results(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) ProjectIndex(Project.open(root)).build() calls = ( ("docforge_project_info", {}), ("docforge_get_contract", {}), ("docforge_get_node", {"node_id": "guide.workflow"}), ("docforge_get_logic", {"owner_node_id": "guide.workflow"}), ("docforge_search", {"query": "canonical nodes", "limit": 5}), ("docforge_filter_nodes", {"family": "proof", "tag": "validation"}), ("docforge_backlinks", {"node_id": "guide.workflow", "limit": 5}), ( "docforge_dependencies", {"node_id": "guide.workflow", "depth": 2, "limit": 5}, ), ( "docforge_impact", {"node_id": "guide.foundation", "depth": 2, "limit": 5}, ), ("docforge_get_context", {"profile": "active", "budget": 180}), ("docforge_validate_project", {}), ("docforge_render_status", {}), ("docforge_visualize", {"node_id": "guide.workflow", "depth": 1}), ("docforge_stop_visualization", {}), ("docforge_visualization_status", {}), ("docforge_bootstrap", {}), ("docforge_sync", {}), ) with self.running_manager(Path(directory) / "viewer-manager.json"): service = DocForgeService(Project.open(root)) try: async with create_connected_server_and_client_session( _create_bound_server(service, read_only=True), raise_exceptions=True ) as session: results = [ await session.call_tool(name, arguments) for name, arguments in calls ] finally: service.visualization.stop() for position, result in enumerate(results): self.assertFalse(result.isError) self.assertIsNotNone(result.structuredContent) payload = result.structuredContent self.assertEqual("ok", payload["status"]) self.assertEqual("alpha-docs", payload["project_id"]) self.assertEqual(CONTENT_WARNING, payload["content_warning"]) self.assertTrue(payload["project_root_fingerprint"]) self.assertEqual( "unknown" if position == 14 else "current", payload["staleness"], ) contract = results[1].structuredContent self.assertFalse(contract["canonical_writes_allowed"]) self.assertFalse(contract["project_switching_allowed"]) self.assertIn("canonical_writes", contract["excluded_operations"]) self.assertIn("arbitrary_renderer_execution", contract["excluded_operations"]) self.assertFalse(contract["isolated_changeset_writes_allowed"]) self.assertFalse(contract["proposal_access"]["enabled"]) self.assertFalse(results[3].structuredContent["available"]) self.assertTrue(results[11].structuredContent["configured"]) self.assertEqual("receipt", results[11].structuredContent["verification"]) self.assertEqual("stale", results[11].structuredContent["state"]) visualization = results[12].structuredContent["visualization"] self.assertTrue(visualization["read_only"]) self.assertTrue(visualization["project_bound"]) self.assertEqual("graph-browser@17", visualization["template"]) self.assertEqual("managed_idle", visualization["lifetime"]["policy"]) self.assertEqual("docforge_stop_visualization", visualization["lifetime"]["stop_tool"]) self.assertTrue(visualization["url"].startswith("http://127.0.0.1:")) self.assertEqual("stopped", results[13].structuredContent["state"]) self.assertEqual("not_running", results[14].structuredContent["state"]) self.assertEqual("unknown", results[14].structuredContent["revision"]) self.assertIsNone(results[14].structuredContent["source_hash"]) self.assertEqual("unknown", results[14].structuredContent["snapshot_state"]) self.assertEqual( {"index": "unknown", "source": "unknown"}, results[14].structuredContent["freshness"], ) context = results[9].structuredContent self.assertLessEqual(context["estimated_tokens"], 180) self.assertTrue(context["omissions"]) async def test_invalid_traversal_limit_is_a_structured_domain_error(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) ProjectIndex(Project.open(root)).build() async with create_connected_server_and_client_session( create_server(root), raise_exceptions=True ) as session: result = await session.call_tool( "docforge_impact", {"node_id": "guide.foundation", "limit": 0}, ) self.assertEqual("error", result.structuredContent["status"]) self.assertEqual( "invalid_limit", result.structuredContent["error"]["code"], ) async def test_render_status_error_never_loads_or_synchronizes_project(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) project = Project.open(root) service = DocForgeService(project) with ( mock.patch.object( project, "load", side_effect=AssertionError("status error decoration must remain cheap"), ), mock.patch.object( service.index, "synchronize", side_effect=AssertionError("status must not synchronize"), ), ): result = service.render_status("not-a-view") self.assertEqual("error", result["status"]) self.assertEqual("unknown_render_view", result["error"]["code"]) self.assertEqual("unknown", result["revision"]) self.assertIsNone(result["source_hash"]) self.assertEqual("unknown", result["staleness"]) async def test_sync_register_rebase_apply_and_lifecycle_are_one_bound_workflow( self, ) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) project = Project.open(root) ProjectIndex(project).build() async with create_connected_server_and_client_session( create_server( root, "alpha-editor", canonical_applier_id="alpha-editor", diagnostics=True, ), raise_exceptions=True, ) as session: bootstrap = await session.call_tool("docforge_bootstrap", {}) self.assertEqual("current", bootstrap.structuredContent["staleness"]) self.assertEqual( str(root), bootstrap.structuredContent["binding"]["project_root"], ) proof = root / "docs/content/proof.toml" proof.write_text( proof.read_text(encoding="utf-8") + "\n# Current validation evidence.\n", encoding="utf-8", ) synchronized = await session.call_tool("docforge_sync", {}) self.assertEqual( "rebuilt", synchronized.structuredContent["synchronization"]["action"], ) registered = await session.call_tool( "docforge_register_changes", { "changeset_id": "bound-workflow", "operations": [ { "operation": "update", "node_id": "guide.workflow", "metadata": { "summary": "Registered and applied in one bound workflow." }, "rationale": "Verify atomic registration without caller hashes.", } ], }, ) self.assertTrue(registered.structuredContent["ready_for_review"]) self.assertEqual("ready", registered.structuredContent["lifecycle"]) foundation = root / "docs/content/foundation.md" foundation.write_text( foundation.read_text(encoding="utf-8") + "\nUnrelated current fact.\n", encoding="utf-8", ) rebased = await session.call_tool( "docforge_rebase_changeset", { "changeset_id": "bound-workflow", "expected_changeset_hash": registered.structuredContent["changeset_hash"], }, ) self.assertTrue(rebased.structuredContent["rebased"]) difference = await session.call_tool( "docforge_get_changeset_diff", {"changeset_id": "bound-workflow"}, ) self.assertEqual("ok", difference.structuredContent["status"]) applied = await session.call_tool( "docforge_apply_changeset", { "changeset_id": "bound-workflow", "expected_changeset_hash": rebased.structuredContent["changeset_hash"], }, ) self.assertEqual( "applied", applied.structuredContent["lifecycle"]["status"], ) closed = await session.call_tool( "docforge_rebase_changeset", { "changeset_id": "bound-workflow", "expected_changeset_hash": rebased.structuredContent["changeset_hash"], }, ) active = await session.call_tool("docforge_list_changesets", {}) history = await session.call_tool( "docforge_list_changesets", {"include_history": True, "status": "applied"}, ) self.assertEqual( "changeset_closed", closed.structuredContent["error"]["code"], ) self.assertEqual(0, active.structuredContent["count"]) self.assertEqual(1, history.structuredContent["count"]) self.assertEqual( "applied", history.structuredContent["changesets"][0]["lifecycle"]["status"], ) async def test_missing_node_fails_and_stale_index_self_heals(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) ProjectIndex(Project.open(root)).build() server = create_server(root) async with create_connected_server_and_client_session( server, raise_exceptions=True ) as session: missing = await session.call_tool( "docforge_get_node", {"node_id": "research.question"} ) workflow = root / "docs" / "content" / "workflow.md" workflow.write_text( workflow.read_text(encoding="utf-8") + "\nChanged after startup.\n", encoding="utf-8", ) repaired = await session.call_tool( "docforge_get_node", {"node_id": "guide.workflow"} ) self.assertEqual("missing_node", missing.structuredContent["error"]["code"]) self.assertEqual("ok", repaired.structuredContent["status"]) self.assertEqual("current", missing.structuredContent["staleness"]) self.assertEqual("current", repaired.structuredContent["staleness"]) self.assertTrue(missing.structuredContent["source_hash"]) self.assertTrue(repaired.structuredContent["source_hash"]) self.assertEqual( "rebuilt", repaired.structuredContent["synchronization"]["action"], ) self.assertFalse(missing.isError) self.assertFalse(repaired.isError) async def test_output_limit_fails_without_returning_partial_content(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) descriptor = root / ".docforge" / "project.toml" descriptor.write_text( descriptor.read_text(encoding="utf-8").replace( "max_context_tokens = 2000", "max_context_tokens = 2000\nmax_tool_output_chars = 700", ), encoding="utf-8", ) ProjectIndex(Project.open(root)).build() async with create_connected_server_and_client_session( create_server(root), raise_exceptions=True ) as session: result = await session.call_tool("docforge_get_contract", {}) payload = result.structuredContent self.assertEqual("error", payload["status"]) self.assertEqual("result_too_large", payload["error"]["code"]) self.assertNotIn("canonical_paths", payload) async def test_mutation_overflow_returns_exact_compact_success_receipts(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) descriptor = root / ".docforge" / "project.toml" descriptor.write_text( descriptor.read_text(encoding="utf-8").replace( "max_context_tokens = 2000", "max_context_tokens = 2000\nmax_tool_output_chars = 1600", ), encoding="utf-8", ) project = Project.open(root) ProjectIndex(project).build() node_hashes = {node.node_id: node.content_hash for node in project.load().nodes} async with create_connected_server_and_client_session( create_server( root, "alpha-editor", canonical_applier_id="alpha-editor", ), raise_exceptions=True, ) as session: created = await session.call_tool( "docforge_create_changeset", {"changeset_id": "compact-mutation"}, ) first = await session.call_tool( "docforge_propose_node_update", { "changeset_id": "compact-mutation", "expected_changeset_hash": created.structuredContent["changeset_hash"], "node_id": "guide.workflow", "expected_content_hash": node_hashes["guide.workflow"], "metadata": None, "content": "Updated workflow.\n\n" + ("bounded receipt evidence " * 200), "relationship_changes": [], "rationale": "Exercise exact compact append receipts.", }, ) second = await session.call_tool( "docforge_propose_node_update", { "changeset_id": "compact-mutation", "expected_changeset_hash": first.structuredContent["changeset_hash"], "node_id": "guide.foundation", "expected_content_hash": node_hashes["guide.foundation"], "metadata": None, "content": "Updated foundation.\n\n" + ("second exact receipt " * 200), "relationship_changes": [], "rationale": "Prove the returned hash supports the next append.", }, ) applied = await session.call_tool( "docforge_apply_changeset", { "changeset_id": "compact-mutation", "expected_changeset_hash": second.structuredContent["changeset_hash"], }, ) for result in (first, second, applied): payload = result.structuredContent self.assertEqual("ok", payload["status"]) self.assertTrue(payload["mutation_committed"]) self.assertEqual("receipt", payload["result_mode"]) self.assertNotIn("diagnostics", payload) self.assertLessEqual( len(json.dumps(payload, sort_keys=True, separators=(",", ":"))), 1600, ) self.assertNotEqual( first.structuredContent["changeset_hash"], second.structuredContent["changeset_hash"], ) self.assertTrue(applied.structuredContent["applied"]) self.assertEqual( "applied", applied.structuredContent["lifecycle"]["status"], ) self.assertEqual( "ok", applied.structuredContent["derived_refresh"]["status"], ) self.assertIn( "Updated workflow.", (root / "docs/content/workflow.md").read_text(encoding="utf-8"), ) async def test_mutation_preflight_rejects_before_writing(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) descriptor = root / ".docforge" / "project.toml" descriptor.write_text( descriptor.read_text(encoding="utf-8").replace( "max_context_tokens = 2000", "max_context_tokens = 2000\nmax_tool_output_chars = 700", ), encoding="utf-8", ) ProjectIndex(Project.open(root)).build() changeset_id = "must-not-exist-" + ("x" * 100) async with create_connected_server_and_client_session( create_server(root, "alpha-editor", diagnostics=True), raise_exceptions=True, ) as session: result = await session.call_tool( "docforge_create_changeset", {"changeset_id": changeset_id}, ) payload = result.structuredContent self.assertEqual("error", payload["status"]) self.assertEqual("result_too_large", payload["error"]["code"]) self.assertEqual("preflight", payload["error"]["details"]["stage"]) self.assertFalse(payload["error"]["details"]["mutation_committed"]) self.assertNotIn("diagnostics", payload) self.assertFalse((root / f".docforge/changesets/{changeset_id}.json").exists()) async def test_proposal_tools_use_fixed_writer_and_never_change_canonical_content(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) project = Project.open(root) ProjectIndex(project).build() canonical_before = { path.relative_to(root).as_posix(): path.read_bytes() for path in (root / "docs/content").glob("*") if path.is_file() } node_hashes = {node.node_id: node.content_hash for node in project.load().nodes} async with create_connected_server_and_client_session( create_server(root, "alpha-editor"), raise_exceptions=True ) as session: contract = await session.call_tool("docforge_get_contract", {}) created = await session.call_tool( "docforge_create_changeset", {"changeset_id": "mcp-update"} ) updated = await session.call_tool( "docforge_propose_node_update", { "changeset_id": "mcp-update", "expected_changeset_hash": created.structuredContent["changeset_hash"], "node_id": "guide.workflow", "expected_content_hash": node_hashes["guide.workflow"], "metadata": {"summary": "A proposal written through MCP."}, "content": None, "relationship_changes": [], "rationale": "Prove fixed-writer isolated proposal access.", }, ) created_node = await session.call_tool( "docforge_propose_node_create", { "changeset_id": "mcp-update", "expected_changeset_hash": updated.structuredContent["changeset_hash"], "node_id": "guide.mcp-node", "target_source": "docs/content/mcp-node.md", "metadata": { "title": "MCP proposal node", "family": "guide", "authority": "proposal", "status": "active", "tags": ["mcp", "proposal"], "summary": "A node creation proposed through MCP.", }, "content": "This node is not canonical until external integration.", "relationship_changes": [], "rationale": "Prove isolated MCP creation.", }, ) moved = await session.call_tool( "docforge_propose_node_move", { "changeset_id": "mcp-update", "expected_changeset_hash": created_node.structuredContent["changeset_hash"], "node_id": "guide.foundation", "expected_content_hash": node_hashes["guide.foundation"], "target_source": "docs/content/foundation-moved.md", "rationale": "Prove isolated MCP movement.", }, ) deleted = await session.call_tool( "docforge_propose_node_delete", { "changeset_id": "mcp-update", "expected_changeset_hash": moved.structuredContent["changeset_hash"], "node_id": "proof.validation", "expected_content_hash": node_hashes["proof.validation"], "relationship_changes": [ { "action": "remove", "source_id": "proof.validation", "relation": "proves", "target_id": "guide.workflow", } ], "rationale": "Prove isolated MCP deletion.", }, ) validated = await session.call_tool( "docforge_validate_changeset", {"changeset_id": "mcp-update"} ) listed = await session.call_tool("docforge_list_changesets", {}) inspected = await session.call_tool( "docforge_get_changeset", {"changeset_id": "mcp-update"} ) diff = await session.call_tool( "docforge_get_changeset_diff", {"changeset_id": "mcp-update"} ) preview = await session.call_tool( "docforge_preview_changeset", {"changeset_id": "mcp-update", "view_id": "manual"}, ) undeclared = await session.call_tool( "docforge_preview_changeset", {"changeset_id": "mcp-update", "view_id": "not-declared"}, ) self.assertTrue(contract.structuredContent["proposal_access"]["enabled"]) self.assertEqual( "alpha-editor", contract.structuredContent["proposal_access"]["writer"] ) self.assertTrue(contract.structuredContent["isolated_changeset_writes_allowed"]) self.assertFalse(contract.structuredContent["canonical_writes_allowed"]) self.assertEqual("alpha-editor", created.structuredContent["creator"]) self.assertEqual(1, updated.structuredContent["operation_count"]) self.assertEqual(4, deleted.structuredContent["operation_count"]) self.assertTrue(validated.structuredContent["valid"]) self.assertEqual(1, listed.structuredContent["count"]) self.assertEqual( "mcp-update", listed.structuredContent["changesets"][0]["changeset_id"] ) self.assertEqual("current", inspected.structuredContent["base_state"]) self.assertEqual(4, inspected.structuredContent["operation_count"]) self.assertEqual( ["update", "create", "move", "delete"], [change["operation"] for change in diff.structuredContent["changes"]], ) self.assertEqual("current", preview.structuredContent["state"]) self.assertEqual( ".docforge/previews/mcp-update/manual.html", preview.structuredContent["preview"]["path"], ) self.assertTrue(preview.structuredContent["preview_identity"]) self.assertEqual("error", undeclared.structuredContent["status"]) self.assertEqual("unknown_render_view", undeclared.structuredContent["error"]["code"]) canonical_after = { path.relative_to(root).as_posix(): path.read_bytes() for path in (root / "docs/content").glob("*") if path.is_file() } self.assertEqual(canonical_before, canonical_after) self.assertFalse((root / "docs/content/mcp-node.md").exists()) self.assertTrue((root / ".docforge/changesets/mcp-update.json").is_file()) self.assertTrue((root / ".docforge/previews/mcp-update/manual.html").is_file()) self.assertFalse((root / ".docforge/rendered/manual.html").exists()) async def test_relationship_only_mcp_tool_queues_no_content_rewrite(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) project = Project.open(root) ProjectIndex(project).build() workflow = next( node for node in project.load().nodes if node.node_id == "guide.workflow" ) async with create_connected_server_and_client_session( create_server(root, "alpha-editor"), raise_exceptions=True ) as session: created = await session.call_tool( "docforge_create_changeset", {"changeset_id": "mcp-relationship"} ) proposed = await session.call_tool( "docforge_propose_relationship_update", { "changeset_id": "mcp-relationship", "expected_changeset_hash": created.structuredContent["changeset_hash"], "node_id": "guide.workflow", "expected_content_hash": workflow.content_hash, "relationship_changes": [ { "action": "remove", "source_id": "guide.workflow", "relation": "depends_on", "target_id": "guide.foundation", } ], "rationale": "Exercise the relationship-only MCP boundary.", }, ) diff = await session.call_tool( "docforge_get_changeset_diff", {"changeset_id": "mcp-relationship"}, ) self.assertEqual("ok", proposed.structuredContent["status"]) self.assertEqual("", diff.structuredContent["changes"][0]["content_diff"]) self.assertEqual({}, diff.structuredContent["changes"][0]["metadata"]) self.assertTrue((root / ".docforge/changesets/mcp-relationship.json").is_file()) self.assertFalse((root / ".docforge/rendered/manual.html").exists()) async def test_server_without_writer_rejects_proposal_mutation_structurally(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) ProjectIndex(Project.open(root)).build() async with create_connected_server_and_client_session( create_server(root), raise_exceptions=True ) as session: result = await session.call_tool( "docforge_create_changeset", {"changeset_id": "disabled"} ) self.assertFalse(result.isError) self.assertEqual("error", result.structuredContent["status"]) self.assertEqual("proposal_access_disabled", result.structuredContent["error"]["code"]) self.assertFalse((root / ".docforge/changesets/disabled.json").exists()) async def test_canonical_apply_tool_is_opt_in_hash_bound_and_refreshes_outputs(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("alpha", Path(directory)) project = Project.open(root) ProjectIndex(project).build() node_hash = next( node.content_hash for node in project.load().nodes if node.node_id == "guide.workflow" ) async with create_connected_server_and_client_session( create_server( root, "alpha-editor", canonical_applier_id="alpha-editor", ), raise_exceptions=True, ) as session: names = tuple(tool.name for tool in (await session.list_tools()).tools) contract = await session.call_tool("docforge_get_contract", {}) created = await session.call_tool( "docforge_create_changeset", {"changeset_id": "mcp-apply"}, ) updated = await session.call_tool( "docforge_propose_node_update", { "changeset_id": "mcp-apply", "expected_changeset_hash": created.structuredContent["changeset_hash"], "node_id": "guide.workflow", "expected_content_hash": node_hash, "metadata": {"summary": "Applied through the gated MCP tool."}, "content": None, "relationship_changes": [], "rationale": "Verify canonical MCP application.", }, ) wrong = await session.call_tool( "docforge_apply_changeset", { "changeset_id": "mcp-apply", "expected_changeset_hash": "0" * 64, }, ) applied = await session.call_tool( "docforge_apply_changeset", { "changeset_id": "mcp-apply", "expected_changeset_hash": updated.structuredContent["changeset_hash"], }, ) self.assertEqual((*ALL_TOOLS, *APPLICATION_TOOLS), names) self.assertTrue(contract.structuredContent["canonical_writes_allowed"]) self.assertTrue(contract.structuredContent["canonical_application_access"]["enabled"]) self.assertNotIn( "canonical_changeset_application", contract.structuredContent["excluded_operations"], ) self.assertEqual("changeset_conflict", wrong.structuredContent["error"]["code"]) self.assertTrue(applied.structuredContent["applied"]) workflow = next( node for node in project.load().nodes if node.node_id == "guide.workflow" ) self.assertEqual("Applied through the gated MCP tool.", workflow.summary) self.assertTrue((root / ".docforge/rendered/manual.html").is_file()) async def test_stdio_transport_serves_the_same_project_bound_contract(self) -> None: with tempfile.TemporaryDirectory() as directory: root = self.copy_fixture("beta", Path(directory)) ProjectIndex(Project.open(root)).build() parameters = StdioServerParameters( command=sys.executable, args=["-m", "docforge.mcp_server", "--project-root", str(root)], ) async with ( stdio_client(parameters) as (read, write), ClientSession(read, write) as session, ): await session.initialize() result = await session.call_tool("docforge_project_info", {}) self.assertFalse(result.isError) self.assertEqual("beta-notes", result.structuredContent["project_id"]) self.assertEqual(1, result.structuredContent["node_count"]) if __name__ == "__main__": unittest.main()