DocForge2/tests/test_mcp_server.py

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 jsonschema import Draft202012Validator
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.application import GenericCanonicalApplier
from docforge.changesets import ChangesetStore
from docforge.errors import DocForgeError
from docforge.index import ProjectIndex
from docforge.mcp_server import (
    ALL_TOOLS,
    APPLICATION_TOOLS,
    CONTENT_WARNING,
    PROPOSAL_TOOLS,
    READ_TOOLS,
    SERVER_VERSION,
    DocForgeService,
    _create_bound_server,
    create_project_server,
    create_server,
)
from docforge.project import Project, project_root_fingerprint
from docforge.viewer_manager import ViewerManager

ROOT = Path(__file__).resolve().parents[1]
FIXTURES = ROOT / "tests" / "fixtures"
CAPSULE_SCHEMA = json.loads(
    (ROOT / "schemas" / "context-capsule.schema.json").read_text(encoding="utf-8")
)


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_get_task_context",
                "docforge_get_generation_diff",
                "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.assertEqual(
                {"task_kind", "task"},
                set(tools["docforge_get_task_context"].inputSchema["required"]),
            )
            self.assertEqual(
                [
                    "change",
                    "implementation",
                    "failure",
                    "ownership",
                    "test",
                    "operation",
                    "release",
                ],
                tools["docforge_get_task_context"].inputSchema["properties"]["task_kind"]["enum"],
            )
            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_explicit_capability_modes_preserve_surfaces_and_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, capability_mode="read"),
                raise_exceptions=True,
            ) as session:
                read_names = tuple(tool.name for tool in (await session.list_tools()).tools)
                read_bootstrap = await session.call_tool("docforge_bootstrap", {})

            self.assertEqual(READ_TOOLS, read_names)
            self.assertEqual(
                "read",
                read_bootstrap.structuredContent["effective_policy"]["capability_mode"],
            )
            self.assertNotIn(
                "docforge_register_changes",
                read_bootstrap.structuredContent["recommended_workflow"],
            )

            async with create_connected_server_and_client_session(
                create_server(
                    root,
                    "alpha-editor",
                    canonical_applier_id="alpha-editor",
                    capability_mode="application",
                ),
                raise_exceptions=True,
            ) as session:
                application_names = tuple(tool.name for tool in (await session.list_tools()).tools)
            self.assertEqual((*ALL_TOOLS, *APPLICATION_TOOLS), application_names)

            with self.assertRaises(DocForgeError) as unavailable:
                create_server(root, capability_mode="application")
            self.assertEqual("capability_unavailable", unavailable.exception.code)

    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_task_context_is_hash_stable_paged_and_generation_bound(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, capability_mode="read", diagnostics=True),
                raise_exceptions=True,
            ) as session:
                first = await session.call_tool(
                    "docforge_get_task_context",
                    {
                        "task_kind": "change",
                        "task": "Change the editing workflow",
                        "focus_node_id": "guide.workflow",
                        "limit": 1,
                    },
                )
                first_capsule = first.structuredContent["capsule"]
                Draft202012Validator(CAPSULE_SCHEMA).validate(first_capsule)
                cursor = first_capsule["pagination"]["next_cursor"]
                hashes = {
                    first_capsule["capsule_hash"],
                    first_capsule["collection_hash"],
                    first_capsule["plan"]["plan_hash"],
                }
                evidence_ids = [item["node_id"] for item in first_capsule["evidence"]]
                while cursor is not None:
                    page = await session.call_tool(
                        "docforge_get_task_context",
                        {
                            "task_kind": "change",
                            "task": "Change the editing workflow",
                            "focus_node_id": "guide.workflow",
                            "limit": 2,
                            "cursor": cursor,
                        },
                    )
                    capsule = page.structuredContent["capsule"]
                    self.assertEqual(first_capsule["capsule_hash"], capsule["capsule_hash"])
                    self.assertEqual(
                        first_capsule["collection_hash"],
                        capsule["collection_hash"],
                    )
                    self.assertEqual(
                        first_capsule["plan"]["plan_hash"],
                        capsule["plan"]["plan_hash"],
                    )
                    evidence_ids.extend(item["node_id"] for item in capsule["evidence"])
                    cursor = capsule["pagination"]["next_cursor"]

                self.assertEqual(
                    ["guide.workflow", "guide.foundation", "proof.validation"],
                    evidence_ids,
                )
                self.assertEqual(3, len(hashes))
                self.assertEqual(
                    "mcp.task_context",
                    first.structuredContent["diagnostics"]["operation"],
                )
                counters = first.structuredContent["diagnostics"]["counters"]
                self.assertEqual(0, counters["project_loads"])
                self.assertEqual(0, counters["source_files_parsed"])
                self.assertEqual(0, counters["adapter_projection_loads"])
                self.assertEqual(0, counters["adapter_source_extractions"])
                self.assertEqual(0, counters["index_builds"])
                self.assertLessEqual(
                    len(json.dumps(first.structuredContent, separators=(",", ":"))),
                    Project.open(root).descriptor.limits.max_tool_output_chars,
                )

                for changed_arguments in (
                    {"task": "A different task"},
                    {"task_kind": "failure"},
                    {"focus_node_id": "guide.foundation"},
                    {"budget": 100},
                ):
                    arguments = {
                        "task_kind": "change",
                        "task": "Change the editing workflow",
                        "focus_node_id": "guide.workflow",
                        "limit": 1,
                        "cursor": first_capsule["pagination"]["next_cursor"],
                        **changed_arguments,
                    }
                    changed_cursor = await session.call_tool(
                        "docforge_get_task_context",
                        arguments,
                    )
                    self.assertEqual(
                        "stale_cursor",
                        changed_cursor.structuredContent["error"]["code"],
                    )

                different_policy = DocForgeService(
                    Project.open(root),
                    capability_mode_name="proposal",
                ).task_context(
                    "change",
                    "Change the editing workflow",
                    focus_node_id="guide.workflow",
                    limit=1,
                    cursor=first_capsule["pagination"]["next_cursor"],
                )
                self.assertEqual("stale_cursor", different_policy["error"]["code"])

                changed = root / "docs/content/foundation.md"
                changed.write_text(
                    changed.read_text(encoding="utf-8") + "\nNew generation.\n",
                    encoding="utf-8",
                )
                stale = await session.call_tool(
                    "docforge_get_task_context",
                    {
                        "task_kind": "change",
                        "task": "Change the editing workflow",
                        "focus_node_id": "guide.workflow",
                        "limit": 1,
                        "cursor": first_capsule["pagination"]["next_cursor"],
                    },
                )
            self.assertEqual("error", stale.structuredContent["status"])
            self.assertEqual("stale_cursor", stale.structuredContent["error"]["code"])

    async def test_custom_context_policy_does_not_silently_gain_task_planning(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = self.copy_fixture("alpha", Path(directory))
            project = Project.open(root)
            ProjectIndex(project).build()
            service = DocForgeService(
                project,
                context_provider=lambda index, profile, budget: {
                    "profile": profile,
                    "budget": budget,
                    "entries": [],
                    "omissions": [],
                },
                capability_mode_name="read",
            )
            async with create_connected_server_and_client_session(
                _create_bound_server(service, read_only=True),
                raise_exceptions=True,
            ) as session:
                bootstrap = await session.call_tool("docforge_bootstrap", {})
                with (
                    mock.patch.object(
                        service.project,
                        "load",
                        side_effect=AssertionError("capability errors must not load"),
                    ),
                    mock.patch.object(
                        service.index,
                        "check",
                        side_effect=AssertionError("capability errors must not check"),
                    ),
                    mock.patch.object(
                        service.index,
                        "build",
                        side_effect=AssertionError("capability errors must not build"),
                    ),
                    mock.patch.object(
                        service.index,
                        "synchronize",
                        side_effect=AssertionError("capability errors must not synchronize"),
                    ),
                ):
                    result = await session.call_tool(
                        "docforge_get_task_context",
                        {
                            "task_kind": "change",
                            "task": "Do not widen the adapter context policy",
                        },
                    )
            self.assertEqual(
                "docforge_get_context",
                bootstrap.structuredContent["session_contract"]["recommended_first_operation"][
                    "tool"
                ],
            )
            self.assertFalse(bootstrap.structuredContent["capabilities"]["task_context"]["enabled"])
            self.assertEqual("error", result.structuredContent["status"])
            self.assertEqual(
                "task_context_unavailable",
                result.structuredContent["error"]["code"],
            )

    def test_task_context_page_hash_binds_final_page_envelope(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 = 8000",
                ),
                encoding="utf-8",
            )
            project = Project.open(root)
            ProjectIndex(project).build()
            service = DocForgeService(project, capability_mode_name="read")
            one = service.task_context(
                "change",
                "Change the editing workflow",
                focus_node_id="guide.workflow",
                limit=1,
            )
            two = service.task_context(
                "change",
                "Change the editing workflow",
                focus_node_id="guide.workflow",
                limit=2,
            )
            one_capsule = one["capsule"]
            two_capsule = two["capsule"]
            self.assertEqual(
                ["guide.workflow"],
                [item["node_id"] for item in one_capsule["evidence"]],
            )
            self.assertEqual(
                ["guide.workflow"],
                [item["node_id"] for item in two_capsule["evidence"]],
            )
            self.assertEqual("complete", one_capsule["page_state"])
            self.assertEqual("incomplete", two_capsule["page_state"])
            self.assertNotEqual(one_capsule["page_hash"], two_capsule["page_hash"])
            self.assertNotEqual(one_capsule["pagination"], two_capsule["pagination"])
            self.assertLessEqual(
                len(json.dumps(one, sort_keys=True, separators=(",", ":"))),
                8_000,
            )
            self.assertLessEqual(
                len(json.dumps(two, sort_keys=True, separators=(",", ":"))),
                8_000,
            )

    def test_dense_task_context_page_packing_is_logarithmic(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_results = 20", "max_results = 1000")
                .replace(
                    "max_context_tokens = 2000",
                    "max_context_tokens = 2000\nmax_tool_output_chars = 20000",
                ),
                encoding="utf-8",
            )
            project = Project.open(root)
            service = DocForgeService(project, capability_mode_name="read")
            capsule = {
                "schema_version": 1,
                "plan": {
                    "effective_policy_hash": "1" * 64,
                    "request_hash": "2" * 64,
                    "plan_hash": "3" * 64,
                },
                "generation": {"index_schema_version": 3},
                "evidence": [
                    {
                        "node_id": f"node.{index:04d}",
                        "content": "bounded evidence " * 40,
                    }
                    for index in range(1_000)
                ],
                "gaps": [],
                "omissions": [],
                "collection_hash": "4" * 64,
                "capsule_hash": "5" * 64,
                "state": "complete",
                "summary": {"evidence_count": 1_000},
            }
            result = {
                "status": "ok",
                "project_id": project.descriptor.project_id,
                "project_root_fingerprint": project_root_fingerprint(project.descriptor.root),
                "adapter": project.descriptor.adapter,
                "revision": "test-revision",
                "source_hash": "6" * 64,
                "capsule": capsule,
            }

            with mock.patch.object(
                service,
                "_encoded_length",
                wraps=service._encoded_length,
            ) as encoded_length:
                page = service._page_task_context_result(
                    result,
                    selected_limit=1_000,
                    cursor=None,
                )

            pagination = page["pagination"]
            self.assertGreater(pagination["returned_count"], 0)
            self.assertLess(pagination["returned_count"], 1_000)
            returned_count = pagination["returned_count"]
            self.assertEqual(
                capsule["evidence"][:returned_count],
                page["capsule"]["evidence"],
            )
            self.assertEqual([], page["capsule"]["omissions"])
            self.assertEqual(
                pagination["next_cursor"],
                page["capsule"]["pagination"]["next_cursor"],
            )
            self.assertEqual(
                returned_count,
                page["capsule"]["summary"]["page_item_count"],
            )
            self.assertLessEqual(encoded_length.call_count, 11)
            decorated = {
                **page,
                "server_version": SERVER_VERSION,
                "content_warning": CONTENT_WARNING,
                "staleness": "current",
            }
            self.assertLessEqual(
                len(json.dumps(decorated, sort_keys=True, separators=(",", ":"))),
                20_000,
            )

    def test_task_context_default_page_clamps_to_small_project_limit(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_results = 20",
                    "max_results = 2",
                ),
                encoding="utf-8",
            )
            project = Project.open(root)
            ProjectIndex(project).build()
            result = DocForgeService(
                project,
                capability_mode_name="read",
            ).task_context(
                "change",
                "Change the editing workflow",
                focus_node_id="guide.workflow",
            )
            self.assertEqual(2, result["pagination"]["limit"])
            self.assertLessEqual(result["pagination"]["returned_count"], 2)

    def test_oversized_task_evidence_advances_once_as_an_omission(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 = 50000\nmax_tool_output_chars = 8000",
                ),
                encoding="utf-8",
            )
            workflow = root / "docs" / "content" / "workflow.md"
            workflow.write_text(
                workflow.read_text(encoding="utf-8") + "\n" + ("large evidence " * 2_000),
                encoding="utf-8",
            )
            project = Project.open(root)
            ProjectIndex(project).build()
            service = DocForgeService(project, capability_mode_name="read")
            first = service.task_context(
                "change",
                "Change the editing workflow",
                focus_node_id="guide.workflow",
                budget=50_000,
                limit=1,
            )
            first_capsule = first["capsule"]
            self.assertEqual([], first_capsule["evidence"])
            self.assertEqual("response_limit", first_capsule["omissions"][0]["code"])
            self.assertEqual("guide.workflow", first_capsule["omissions"][0]["subject"])
            self.assertEqual(1, first_capsule["pagination"]["returned_count"])
            self.assertTrue(first_capsule["pagination"]["has_more"])
            second = service.task_context(
                "change",
                "Change the editing workflow",
                focus_node_id="guide.workflow",
                budget=50_000,
                limit=1,
                cursor=first_capsule["pagination"]["next_cursor"],
            )
            self.assertNotEqual(first_capsule["page_hash"], second["capsule"]["page_hash"])
            self.assertNotIn(
                "guide.workflow",
                [
                    item.get("node_id", item.get("subject"))
                    for item in (
                        *second["capsule"]["evidence"],
                        *second["capsule"]["omissions"],
                    )
                ],
            )

    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"}
                )
                task_context = await session.call_tool(
                    "docforge_get_task_context",
                    {
                        "task_kind": "change",
                        "task": "Change the editing workflow without AST analysis",
                        "focus_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"],
            )
            self.assertEqual("ok", task_context.structuredContent["status"])
            self.assertNotIn(
                "logic",
                {
                    step["operation"]
                    for step in task_context.structuredContent["capsule"]["plan"]["steps"]
                },
            )

    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", {}),
                (
                    "docforge_get_task_context",
                    {
                        "task_kind": "change",
                        "task": "Change the editing workflow",
                        "focus_node_id": "guide.workflow",
                    },
                ),
            )
            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"])
            self.assertEqual("complete", results[17].structuredContent["capsule"]["state"])

    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_project_server_accepts_an_explicit_contributor_without_granting_apply(
        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")
                + """

[[changesets.writers]]
id = "contributor"
families = ["guide"]
operations = ["update"]
""",
                encoding="utf-8",
            )
            project = Project.open(root)
            ProjectIndex(project).build()
            async with create_connected_server_and_client_session(
                create_server(
                    root,
                    "contributor",
                    capability_mode="proposal",
                ),
                raise_exceptions=True,
            ) as contributor:
                contributor_tools = tuple(
                    tool.name for tool in (await contributor.list_tools()).tools
                )
                registered = await contributor.call_tool(
                    "docforge_register_changes",
                    {
                        "changeset_id": "accepted-contribution",
                        "operations": [
                            {
                                "operation": "update",
                                "node_id": "guide.workflow",
                                "metadata": {"summary": "Accepted through a separate applier."},
                                "rationale": "Prove explicit contributor acceptance over MCP.",
                            }
                        ],
                    },
                )

            async with create_connected_server_and_client_session(
                create_project_server(
                    project,
                    proposal_writer="alpha-editor",
                    canonical_applier_id="alpha-editor",
                    canonical_applier=GenericCanonicalApplier(project),
                    accepted_proposal_writers=("contributor",),
                    capability_mode="application",
                ),
                raise_exceptions=True,
            ) as developer:
                contract = await developer.call_tool("docforge_get_contract", {})
                applied = await developer.call_tool(
                    "docforge_apply_changeset",
                    {
                        "changeset_id": "accepted-contribution",
                        "expected_changeset_hash": registered.structuredContent["changeset_hash"],
                    },
                )

            self.assertEqual(ALL_TOOLS, contributor_tools)
            self.assertNotIn("docforge_apply_changeset", contributor_tools)
            self.assertEqual(
                ["contributor"],
                contract.structuredContent["canonical_application_access"][
                    "accepted_proposal_writers"
                ],
            )
            self.assertTrue(applied.structuredContent["applied"])
            self.assertEqual("contributor", applied.structuredContent["proposal_creator"])
            self.assertEqual("alpha-editor", applied.structuredContent["applied_by"])

    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()