785 lines
37 KiB
Python
785 lines
37 KiB
Python
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 mcp import ClientSession, StdioServerParameters
|
|
from mcp.client.stdio import stdio_client
|
|
from mcp.shared.memory import create_connected_server_and_client_session
|
|
|
|
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", []))
|
|
self.assertFalse(
|
|
any(
|
|
token in name
|
|
for name in names
|
|
for token in ("apply", "commit", "push", "deploy", "publish", "shell")
|
|
)
|
|
)
|
|
|
|
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 result in 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("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("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"])
|
|
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_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",
|
|
),
|
|
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.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"),
|
|
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.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()
|