administrator/DocForge2
1
0
Fork 0
Code Issues Pull requests Projects Releases 2 Packages Wiki Activity Actions Pages

feat: add isolated proposal changesets

Browse source
This commit is contained in:
Andraxion 2026-07-22 02:58:51 -04:00
parent 9702ed1265
commit 8c75f4f44d
22 changed files with 2314 additions and 64 deletions
Download patch file Download diff file Expand all files Collapse all files

11
tests/fixtures/alpha/.docforge/project.toml vendored
View file

@ -11,6 +11,14 @@ authority_files = ["POLICY.md"]
cache_root = ".docforge/cache"
index = ".docforge/cache/index.sqlite3"
[changesets]
root = ".docforge/changesets"
[[changesets.writers]]
id = "alpha-editor"
families = ["guide", "proof"]
operations = ["create", "update", "move", "delete"]
[graph]
allowed_relations = ["depends_on", "proves", "supersedes", "relates_to", "returns_to"]
@ -21,6 +29,9 @@ max_query_chars = 200
max_results = 20
max_traversal_depth = 4
max_context_tokens = 2000
max_changesets = 20
max_changeset_operations = 20
max_changeset_bytes = 100000
[[profiles]]
id = "active"

11
tests/fixtures/beta/.docforge/project.toml vendored
View file

@ -11,6 +11,14 @@ authority_files = []
cache_root = ".docforge/cache"
index = ".docforge/cache/index.sqlite3"
[changesets]
root = ".docforge/changesets"
[[changesets.writers]]
id = "beta-editor"
families = ["research"]
operations = ["create", "update", "move", "delete"]
[graph]
allowed_relations = ["depends_on", "relates_to"]
@ -21,6 +29,9 @@ max_query_chars = 200
max_results = 20
max_traversal_depth = 3
max_context_tokens = 1000
max_changesets = 10
max_changeset_operations = 10
max_changeset_bytes = 50000
[[profiles]]
id = "overview"

558
tests/test_changesets.py Normal file
View file

@ -0,0 +1,558 @@
from __future__ import annotations
import hashlib
import json
import multiprocessing
import shutil
import tempfile
import unittest
from pathlib import Path
from typing import Any
from unittest import mock
from docforge.changesets import ChangesetStore
from docforge.errors import DocForgeError
from docforge.project import Project
ROOT = Path(__file__).resolve().parents[1]
FIXTURES = ROOT / "tests" / "fixtures"
def _append_in_process(
root: str,
node_id: str,
expected_changeset_hash: str,
expected_content_hash: str,
start: Any,
result: Any,
) -> None:
"""Attempt one append in an independent process and report its stable outcome."""
start.wait()
try:
ChangesetStore(Project.open(root), "alpha-editor").propose_update(
changeset_id="simultaneous",
expected_changeset_hash=expected_changeset_hash,
node_id=node_id,
expected_content_hash=expected_content_hash,
metadata={"summary": f"Concurrent proposal for {node_id}."},
content=None,
relationship_changes=[],
rationale="Prove process-safe optimistic serialization.",
)
except DocForgeError as error:
result.send(error.code)
else:
result.send("ok")
finally:
result.close()
class DocForgeChangesetTests(unittest.TestCase):
def copy_fixture(self, destination: Path) -> Path:
root = destination / "alpha"
shutil.copytree(FIXTURES / "alpha", root)
return root
def canonical_digest(self, root: Path) -> str:
paths = [
root / ".docforge" / "project.toml",
root / "POLICY.md",
*(root / "docs" / "content").glob("*"),
]
digest = hashlib.sha256()
for path in sorted((path for path in paths if path.is_file()), key=lambda item: str(item)):
digest.update(path.relative_to(root).as_posix().encode())
digest.update(path.read_bytes())
return digest.hexdigest()
@staticmethod
def node_hash(project: Project, node_id: str) -> str:
return next(node.content_hash for node in project.load().nodes if node.node_id == node_id)
@staticmethod
def new_metadata(*, family: str = "guide") -> dict[str, object]:
return {
"title": "New proposal node",
"family": family,
"authority": "proposal",
"status": "active",
"tags": ["proposal", "test"],
"summary": "Exercises an isolated structured creation proposal.",
}
def test_all_four_operations_validate_and_never_change_canonical_sources(self) -> None:
with tempfile.TemporaryDirectory() as directory:
parent = Path(directory)
create_root = self.copy_fixture(parent / "create")
create_before = self.canonical_digest(create_root)
create_store = ChangesetStore(Project.open(create_root), "alpha-editor")
created = create_store.create("create-node")
proposed = create_store.propose_create(
changeset_id="create-node",
expected_changeset_hash=created["changeset_hash"],
node_id="guide.new-node",
target_source="docs/content/new-node.md",
metadata=self.new_metadata(),
content="This content exists only inside the proposal.",
relationship_changes=[
{
"action": "add",
"source_id": "guide.new-node",
"relation": "depends_on",
"target_id": "guide.foundation",
}
],
rationale="Add one independently reviewable guide node.",
)
self.assertEqual(1, proposed["operation_count"])
self.assertEqual(create_before, self.canonical_digest(create_root))
self.assertFalse((create_root / "docs/content/new-node.md").exists())
create_diff = create_store.diff("create-node")
self.assertEqual("create", create_diff["changes"][0]["operation"])
self.assertIn("proposal", create_diff["changes"][0]["content_diff"])
self.assertEqual(create_diff, create_store.diff("create-node"))
update_root = self.copy_fixture(parent / "update")
update_before = self.canonical_digest(update_root)
update_project = Project.open(update_root)
update_store = ChangesetStore(update_project, "alpha-editor")
update_changeset = update_store.create("update-node")
updated = update_store.propose_update(
changeset_id="update-node",
expected_changeset_hash=update_changeset["changeset_hash"],
node_id="guide.workflow",
expected_content_hash=self.node_hash(update_project, "guide.workflow"),
metadata={"summary": "A proposed replacement workflow summary."},
content="Proposed workflow content that is not canonical yet.",
relationship_changes=[],
rationale="Replace the workflow explanation without applying it.",
)
self.assertEqual(3, updated["projected_node_count"])
update_diff = update_store.diff("update-node")["changes"][0]
self.assertEqual("update", update_diff["operation"])
self.assertIn("summary", update_diff["metadata"])
self.assertEqual(update_before, self.canonical_digest(update_root))
move_root = self.copy_fixture(parent / "move")
move_before = self.canonical_digest(move_root)
move_project = Project.open(move_root)
move_store = ChangesetStore(move_project, "alpha-editor")
move_changeset = move_store.create("move-node")
move_store.propose_move(
changeset_id="move-node",
expected_changeset_hash=move_changeset["changeset_hash"],
node_id="guide.foundation",
expected_content_hash=self.node_hash(move_project, "guide.foundation"),
target_source="docs/content/foundation-moved.md",
rationale="Move the node while preserving its stable identity.",
)
move_diff = move_store.diff("move-node")["changes"][0]
self.assertEqual(
{
"before": "docs/content/foundation.md",
"after": "docs/content/foundation-moved.md",
},
move_diff["source"],
)
self.assertEqual(move_before, self.canonical_digest(move_root))
delete_root = self.copy_fixture(parent / "delete")
delete_before = self.canonical_digest(delete_root)
delete_project = Project.open(delete_root)
delete_store = ChangesetStore(delete_project, "alpha-editor")
delete_changeset = delete_store.create("delete-node")
deleted = delete_store.propose_delete(
changeset_id="delete-node",
expected_changeset_hash=delete_changeset["changeset_hash"],
node_id="proof.validation",
expected_content_hash=self.node_hash(delete_project, "proof.validation"),
relationship_changes=[
{
"action": "remove",
"source_id": "proof.validation",
"relation": "proves",
"target_id": "guide.workflow",
}
],
rationale="Remove the proof node and its owned relationship in one proposal.",
)
self.assertEqual(2, deleted["projected_node_count"])
self.assertEqual("delete", delete_store.diff("delete-node")["changes"][0]["operation"])
self.assertEqual(delete_before, self.canonical_digest(delete_root))
def test_optimistic_and_cross_changeset_conflicts_preserve_both_proposals(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = self.copy_fixture(Path(directory))
project = Project.open(root)
store = ChangesetStore(project, "alpha-editor")
first = store.create("first")
second = store.create("second")
foundation_hash = self.node_hash(project, "guide.foundation")
first_result = store.propose_update(
changeset_id="first",
expected_changeset_hash=first["changeset_hash"],
node_id="guide.foundation",
expected_content_hash=foundation_hash,
metadata={"summary": "First proposal owns this node."},
content=None,
relationship_changes=[],
rationale="Claim the foundation node in the first proposal.",
)
with self.assertRaisesRegex(DocForgeError, "overlaps another proposal") as conflict:
store.propose_update(
changeset_id="second",
expected_changeset_hash=second["changeset_hash"],
node_id="guide.foundation",
expected_content_hash=foundation_hash,
metadata={"summary": "Second proposal conflicts with the first."},
content=None,
relationship_changes=[],
rationale="Prove exact overlap detection.",
)
self.assertEqual("proposal_conflict", conflict.exception.code)
self.assertEqual("first", conflict.exception.details["conflicts"][0]["changeset_id"])
self.assertEqual(0, store.validate("second")["operation_count"])
with self.assertRaisesRegex(DocForgeError, "changed after") as stale_hash:
store.propose_create(
changeset_id="first",
expected_changeset_hash=first["changeset_hash"],
node_id="guide.another",
target_source="docs/content/another.md",
metadata=self.new_metadata(),
content="Another proposal operation.",
relationship_changes=[],
rationale="Use an obsolete changeset hash.",
)
self.assertEqual("changeset_conflict", stale_hash.exception.code)
self.assertEqual(
first_result["changeset_hash"], store.validate("first")["changeset_hash"]
)
def test_simultaneous_process_appends_serialize_without_losing_an_operation(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = self.copy_fixture(Path(directory))
project = Project.open(root)
initial = ChangesetStore(project, "alpha-editor").create("simultaneous")
hashes = {
node.node_id: node.content_hash
for node in project.load().nodes
if node.node_id in {"guide.foundation", "guide.workflow"}
}
context = multiprocessing.get_context("spawn")
start = context.Event()
receivers: list[Any] = []
processes: list[multiprocessing.Process] = []
for node_id in ("guide.foundation", "guide.workflow"):
receiver, sender = context.Pipe(duplex=False)
process = context.Process(
target=_append_in_process,
args=(
str(root),
node_id,
initial["changeset_hash"],
hashes[node_id],
start,
sender,
),
)
process.start()
sender.close()
receivers.append(receiver)
processes.append(process)
start.set()
results = sorted(receiver.recv() for receiver in receivers)
for receiver in receivers:
receiver.close()
for process in processes:
process.join(timeout=10)
self.assertEqual(0, process.exitcode)
self.assertEqual(["changeset_conflict", "ok"], results)
validated = ChangesetStore(project).validate("simultaneous")
self.assertEqual(1, validated["operation_count"])
def test_configured_changeset_count_operation_and_size_limits_fail_atomically(self) -> None:
with tempfile.TemporaryDirectory() as directory:
parent = Path(directory)
count_root = self.copy_fixture(parent / "count")
descriptor = count_root / ".docforge/project.toml"
descriptor.write_text(
descriptor.read_text(encoding="utf-8")
.replace("max_changesets = 20", "max_changesets = 1")
.replace("max_changeset_operations = 20", "max_changeset_operations = 1"),
encoding="utf-8",
)
count_project = Project.open(count_root)
count_store = ChangesetStore(count_project, "alpha-editor")
limited = count_store.create("limited")
with self.assertRaisesRegex(DocForgeError, "changeset limit") as count_error:
count_store.create("excess")
self.assertEqual("changeset_limit", count_error.exception.code)
self.assertFalse((count_root / ".docforge/changesets/excess.json").exists())
first = count_store.propose_update(
changeset_id="limited",
expected_changeset_hash=limited["changeset_hash"],
node_id="guide.foundation",
expected_content_hash=self.node_hash(count_project, "guide.foundation"),
metadata={"summary": "The only permitted operation."},
content=None,
relationship_changes=[],
rationale="Reach the configured operation limit.",
)
limited_path = count_root / ".docforge/changesets/limited.json"
limited_bytes = limited_path.read_bytes()
with self.assertRaisesRegex(DocForgeError, "operation limit") as operation_error:
count_store.propose_update(
changeset_id="limited",
expected_changeset_hash=first["changeset_hash"],
node_id="guide.workflow",
expected_content_hash=self.node_hash(count_project, "guide.workflow"),
metadata={"summary": "This operation exceeds the limit."},
content=None,
relationship_changes=[],
rationale="Prove the configured operation cap.",
)
self.assertEqual("changeset_operation_limit", operation_error.exception.code)
self.assertEqual(limited_bytes, limited_path.read_bytes())
size_root = self.copy_fixture(parent / "size")
size_descriptor = size_root / ".docforge/project.toml"
size_descriptor.write_text(
size_descriptor.read_text(encoding="utf-8").replace(
"max_changeset_bytes = 100000", "max_changeset_bytes = 64"
),
encoding="utf-8",
)
size_store = ChangesetStore(Project.open(size_root), "alpha-editor")
with self.assertRaisesRegex(DocForgeError, "size limit") as size_error:
size_store.create("too-large")
self.assertEqual("changeset_too_large", size_error.exception.code)
self.assertFalse((size_root / ".docforge/changesets/too-large.json").exists())
def test_stale_base_bad_content_hash_and_failed_delete_are_atomic(self) -> None:
with tempfile.TemporaryDirectory() as directory:
parent = Path(directory)
stale_root = self.copy_fixture(parent / "stale")
stale_project = Project.open(stale_root)
stale_store = ChangesetStore(stale_project, "alpha-editor")
stale = stale_store.create("stale")
stale_path = stale_root / ".docforge/changesets/stale.json"
stale_bytes = stale_path.read_bytes()
workflow = stale_root / "docs/content/workflow.md"
workflow.write_text(workflow.read_text() + "\nCanonical change.\n", encoding="utf-8")
with self.assertRaisesRegex(DocForgeError, "Canonical project changed") as base:
stale_store.propose_move(
changeset_id="stale",
expected_changeset_hash=stale["changeset_hash"],
node_id="guide.foundation",
expected_content_hash=self.node_hash(
Project.open(stale_root), "guide.foundation"
),
target_source="docs/content/foundation-moved.md",
rationale="This proposal has an obsolete base.",
)
self.assertEqual("base_conflict", base.exception.code)
self.assertEqual(stale_bytes, stale_path.read_bytes())
self.assertEqual("stale", stale_store.inspect("stale")["base_state"])
self.assertEqual("stale", stale_store.list_changesets()["changesets"][0]["base_state"])
delete_root = self.copy_fixture(parent / "delete")
delete_project = Project.open(delete_root)
delete_store = ChangesetStore(delete_project, "alpha-editor")
delete = delete_store.create("delete")
delete_path = delete_root / ".docforge/changesets/delete.json"
delete_bytes = delete_path.read_bytes()
with self.assertRaisesRegex(DocForgeError, "every incident relationship") as edges:
delete_store.propose_delete(
changeset_id="delete",
expected_changeset_hash=delete["changeset_hash"],
node_id="proof.validation",
expected_content_hash=self.node_hash(delete_project, "proof.validation"),
relationship_changes=[],
rationale="An incomplete delete must not be stored.",
)
self.assertEqual("unresolved_relationships", edges.exception.code)
self.assertEqual(delete_bytes, delete_path.read_bytes())
with self.assertRaisesRegex(DocForgeError, "changed after"):
delete_store.propose_update(
changeset_id="delete",
expected_changeset_hash=delete["changeset_hash"],
node_id="guide.foundation",
expected_content_hash="0" * 64,
metadata={"summary": "Wrong base hash."},
content=None,
relationship_changes=[],
rationale="Reject a mismatched node hash.",
)
self.assertEqual(delete_bytes, delete_path.read_bytes())
def test_writer_family_permissions_ownership_and_paths_fail_closed(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = self.copy_fixture(Path(directory))
descriptor = root / ".docforge/project.toml"
descriptor.write_text(
descriptor.read_text().replace(
'families = ["guide", "proof"]', 'families = ["guide"]'
)
+ """
[[changesets.writers]]
id = "other-editor"
families = ["guide"]
operations = ["create", "update", "move", "delete"]
""",
encoding="utf-8",
)
project = Project.open(root)
store = ChangesetStore(project, "alpha-editor")
changeset = store.create("permissions")
path = root / ".docforge/changesets/permissions.json"
before = path.read_bytes()
with self.assertRaisesRegex(DocForgeError, "families") as family:
store.propose_create(
changeset_id="permissions",
expected_changeset_hash=changeset["changeset_hash"],
node_id="proof.new",
target_source="docs/content/proof-new.md",
metadata=self.new_metadata(family="proof"),
content="A forbidden family proposal.",
relationship_changes=[],
rationale="Prove family permission enforcement.",
)
self.assertEqual("family_forbidden", family.exception.code)
self.assertEqual(before, path.read_bytes())
other = ChangesetStore(project, "other-editor")
with self.assertRaisesRegex(DocForgeError, "does not own") as owner:
other.propose_create(
changeset_id="permissions",
expected_changeset_hash=changeset["changeset_hash"],
node_id="guide.other",
target_source="docs/content/other.md",
metadata=self.new_metadata(),
content="Another writer cannot append here.",
relationship_changes=[],
rationale="Prove immutable changeset ownership.",
)
self.assertEqual("changeset_owner_conflict", owner.exception.code)
with self.assertRaisesRegex(DocForgeError, "canonical content root"):
store.propose_create(
changeset_id="permissions",
expected_changeset_hash=changeset["changeset_hash"],
node_id="guide.escape",
target_source="outside.md",
metadata=self.new_metadata(),
content="This path is outside the content root.",
relationship_changes=[],
rationale="Prove target confinement.",
)
outside = Path(directory) / "outside.json"
outside.write_text(json.dumps({"secret": True}), encoding="utf-8")
symlink = root / ".docforge/changesets/symlink.json"
symlink.symlink_to(outside)
with self.assertRaisesRegex(DocForgeError, "symbolic links"):
store.validate("symlink")
self.assertEqual({"secret": True}, json.loads(outside.read_text()))
def test_changeset_configuration_overlap_and_invalid_permissions_are_rejected(self) -> None:
with tempfile.TemporaryDirectory() as directory:
parent = Path(directory)
overlap_root = self.copy_fixture(parent / "overlap")
overlap = overlap_root / ".docforge/project.toml"
overlap.write_text(
overlap.read_text().replace(
'root = ".docforge/changesets"', 'root = "docs/content/changesets"'
),
encoding="utf-8",
)
with self.assertRaisesRegex(DocForgeError, "must not overlap"):
Project.open(overlap_root)
invalid_root = self.copy_fixture(parent / "invalid")
invalid = invalid_root / ".docforge/project.toml"
invalid.write_text(
invalid.read_text().replace(
'operations = ["create", "update", "move", "delete"]',
'operations = ["create", "execute"]',
),
encoding="utf-8",
)
with self.assertRaisesRegex(DocForgeError, "operations are empty or invalid"):
Project.open(invalid_root)
symlink_root = self.copy_fixture(parent / "symlink")
outside = parent / "outside-changesets"
outside.mkdir()
(symlink_root / ".docforge/changesets").symlink_to(outside, target_is_directory=True)
with self.assertRaisesRegex(DocForgeError, "outside the project root"):
Project.open(symlink_root)
def test_toml_anchors_and_mid_write_source_set_changes_fail_atomically(self) -> None:
with tempfile.TemporaryDirectory() as directory:
parent = Path(directory)
toml_root = self.copy_fixture(parent / "toml")
toml_store = ChangesetStore(Project.open(toml_root), "alpha-editor")
changeset = toml_store.create("toml-create")
path = toml_root / ".docforge/changesets/toml-create.json"
before = path.read_bytes()
with self.assertRaisesRegex(DocForgeError, "explicit source_anchor"):
toml_store.propose_create(
changeset_id="toml-create",
expected_changeset_hash=changeset["changeset_hash"],
node_id="proof.second",
target_source="docs/content/proof.toml",
metadata=self.new_metadata(family="proof"),
content="A TOML proposal needs an explicit anchor.",
relationship_changes=[],
rationale="Reject ambiguous TOML placement.",
)
self.assertEqual(before, path.read_bytes())
duplicate_anchor = {
**self.new_metadata(family="proof"),
"source_anchor": "node-1",
}
with self.assertRaisesRegex(DocForgeError, "anchors must be unique"):
toml_store.propose_create(
changeset_id="toml-create",
expected_changeset_hash=changeset["changeset_hash"],
node_id="proof.second",
target_source="docs/content/proof.toml",
metadata=duplicate_anchor,
content="A duplicate anchor is still ambiguous.",
relationship_changes=[],
rationale="Reject duplicate TOML anchors.",
)
self.assertEqual(before, path.read_bytes())
race_root = self.copy_fixture(parent / "race")
race_project = Project.open(race_root)
race_store = ChangesetStore(race_project, "alpha-editor")
sources = race_project.canonical_source_paths()
invented = race_root / "docs/content/invented.md"
with (
mock.patch.object(
race_project,
"canonical_source_paths",
side_effect=[sources, sources, sources, (*sources, invented)],
),
self.assertRaisesRegex(DocForgeError, "changed during changeset storage"),
):
race_store.create("source-race")
self.assertFalse((race_root / ".docforge/changesets/source-race.json").exists())
if __name__ == "__main__":
unittest.main()

4
tests/test_core.py
View file

@ -247,11 +247,11 @@ class DocForgeCoreTests(unittest.TestCase):
def test_source_set_change_during_load_fails_closed(self) -> None:
project = Project.open(FIXTURES / "alpha")
sources = project._canonical_source_paths()
sources = project.canonical_source_paths()
invented = project.descriptor.root / "docs" / "content" / "invented.md"
with (
mock.patch.object(
project, "_canonical_source_paths", side_effect=[sources, (*sources, invented)]
project, "canonical_source_paths", side_effect=[sources, (*sources, invented)]
),
self.assertRaisesRegex(DocForgeError, "source set changed"),
):

147
tests/test_mcp_server.py
View file

@ -11,7 +11,7 @@ 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 CONTENT_WARNING, READ_TOOLS, create_server
from docforge.mcp_server import ALL_TOOLS, CONTENT_WARNING, PROPOSAL_TOOLS, create_server
from docforge.project import Project
ROOT = Path(__file__).resolve().parents[1]
@ -24,7 +24,7 @@ class DocForgeMcpTests(unittest.IsolatedAsyncioTestCase):
shutil.copytree(FIXTURES / name, root)
return root
async def test_protocol_lists_only_the_fixed_read_surface(self) -> None:
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()
@ -34,12 +34,13 @@ class DocForgeMcpTests(unittest.IsolatedAsyncioTestCase):
response = await session.list_tools()
names = tuple(tool.name for tool in response.tools)
self.assertEqual(READ_TOOLS, names)
self.assertEqual(ALL_TOOLS, names)
self.assertEqual(9, len(PROPOSAL_TOOLS))
self.assertFalse(
any(
token in name
for name in names
for token in ("write", "apply", "commit", "push", "deploy", "propose")
for token in ("apply", "commit", "push", "deploy", "publish", "shell")
)
)
@ -78,6 +79,8 @@ class DocForgeMcpTests(unittest.IsolatedAsyncioTestCase):
self.assertFalse(contract["canonical_writes_allowed"])
self.assertFalse(contract["project_switching_allowed"])
self.assertIn("canonical_writes", contract["excluded_operations"])
self.assertFalse(contract["isolated_changeset_writes_allowed"])
self.assertFalse(contract["proposal_access"]["enabled"])
context = results[8].structuredContent
self.assertLessEqual(context["estimated_tokens"], 180)
self.assertTrue(context["omissions"])
@ -131,6 +134,142 @@ class DocForgeMcpTests(unittest.IsolatedAsyncioTestCase):
self.assertEqual("result_too_large", payload["error"]["code"])
self.assertNotIn("canonical_paths", payload)
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"}
)
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"]],
)
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())
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_stdio_transport_serves_the_same_project_bound_contract(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = self.copy_fixture("beta", Path(directory))