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

Add versioned effective policy

Browse source
This commit is contained in:
Andraxion 2026-07-29 06:26:40 -04:00
parent a9a75c5c27
commit 34cd5f74c1
11 changed files with 708 additions and 70 deletions
Download patch file Download diff file Expand all files Collapse all files

39
tests/test_mcp_server.py
View file

@ -17,12 +17,14 @@ from mcp.client.stdio import stdio_client
from mcp.shared.memory import create_connected_server_and_client_session
from docforge.changesets import ChangesetStore
from docforge.errors import DocForgeError
from docforge.index import ProjectIndex
from docforge.mcp_server import (
ALL_TOOLS,
APPLICATION_TOOLS,
CONTENT_WARNING,
PROPOSAL_TOOLS,
READ_TOOLS,
DocForgeService,
_create_bound_server,
create_server,
@ -103,6 +105,43 @@ class DocForgeMcpTests(unittest.IsolatedAsyncioTestCase):
)
)
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))

161
tests/test_policy.py Normal file
View file

@ -0,0 +1,161 @@
from __future__ import annotations
import json
import shutil
import tempfile
import unittest
from pathlib import Path
from unittest import mock
from jsonschema import Draft202012Validator
from docforge.application import GenericCanonicalApplier
from docforge.errors import DocForgeError
from docforge.index import ProjectIndex
from docforge.mcp_server import (
ALL_TOOLS,
APPLICATION_TOOLS,
READ_TOOLS,
DocForgeService,
)
from docforge.policy import (
POLICY_PRECEDENCE,
capability_mode,
compose_effective_policy,
)
from docforge.project import Project
ROOT = Path(__file__).resolve().parents[1]
FIXTURES = ROOT / "tests" / "fixtures"
POLICY_SCHEMA = json.loads((ROOT / "schemas" / "policy.schema.json").read_text(encoding="utf-8"))
class EffectivePolicyTests(unittest.TestCase):
def copy_fixture(self, destination: Path) -> Path:
root = destination / "alpha"
shutil.copytree(FIXTURES / "alpha", root)
return root
def test_policy_schema_and_legacy_adapter_projection_are_exact(self) -> None:
standard = compose_effective_policy(
selected_mode="proposal",
capability_source="factory_default",
no_ast=False,
diagnostics=False,
render_configured=True,
application_enabled=False,
)
preserve = compose_effective_policy(
selected_mode="read",
capability_source="explicit",
no_ast=True,
diagnostics=True,
render_configured=False,
application_enabled=False,
)
validator = Draft202012Validator(POLICY_SCHEMA)
validator.validate(standard.as_dict())
validator.validate(preserve.as_dict())
self.assertEqual(list(POLICY_PRECEDENCE), preserve.as_dict()["precedence"])
self.assertEqual(
{
"mode": "standard",
"ast_analysis": "allowed",
"logic_projection": "allowed",
"incremental_extraction": "allowed",
"adapter_rewrite": "not_requested",
},
standard.adapter_policy(),
)
self.assertEqual("preserve-no-ast", preserve.adapter_policy()["mode"])
self.assertEqual(["docforge_get_logic"], preserve.adapter_policy()["blocked_tools"])
self.assertEqual("off", preserve.as_dict()["logic_indexing"])
self.assertEqual("enabled", preserve.as_dict()["profiling"])
def test_invalid_or_unavailable_capability_fails_closed(self) -> None:
with self.assertRaises(DocForgeError) as invalid:
capability_mode("admin", default="read")
self.assertEqual("invalid_capability_mode", invalid.exception.code)
with self.assertRaises(DocForgeError) as unavailable:
compose_effective_policy(
selected_mode="application",
capability_source="explicit",
no_ast=False,
diagnostics=False,
render_configured=True,
application_enabled=False,
)
self.assertEqual("capability_unavailable", unavailable.exception.code)
def test_modes_preserve_default_surfaces_and_narrow_authority(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = self.copy_fixture(Path(directory))
project = Project.open(root)
read = DocForgeService(
project,
proposal_writer="alpha-editor",
capability_mode_name="read",
)
proposal = DocForgeService(
project,
proposal_writer="alpha-editor",
capability_mode_name="proposal",
)
application = DocForgeService(
project,
proposal_writer="alpha-editor",
canonical_applier_id="alpha-editor",
canonical_applier=GenericCanonicalApplier(project),
capability_mode_name="application",
)
operator = DocForgeService(
project,
proposal_writer="alpha-editor",
capability_mode_name="operator",
)
self.assertEqual(READ_TOOLS, read.tool_surface)
self.assertFalse(read.changesets.access()["enabled"])
self.assertEqual(ALL_TOOLS, proposal.tool_surface)
self.assertTrue(proposal.changesets.access()["enabled"])
self.assertFalse(proposal.application.enabled)
self.assertEqual((*ALL_TOOLS, *APPLICATION_TOOLS), application.tool_surface)
self.assertTrue(application.application.enabled)
self.assertEqual(ALL_TOOLS, operator.tool_surface)
self.assertTrue(operator.capabilities()["operator"]["enabled"])
def test_bootstrap_reuses_synchronized_identity_and_filters_workflow(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = self.copy_fixture(Path(directory))
project = Project.open(root)
ProjectIndex(project).build()
read = DocForgeService(project, capability_mode_name="read")
with mock.patch.object(
project,
"load",
side_effect=AssertionError("bootstrap must not reload after synchronization"),
):
result = read.bootstrap()
self.assertEqual("ok", result["status"])
self.assertEqual(1, result["session_contract"]["schema_version"])
self.assertEqual("read", result["effective_policy"]["capability_mode"])
self.assertEqual(
result["effective_policy"], result["session_contract"]["effective_policy"]
)
self.assertNotIn("docforge_register_changes", result["recommended_workflow"])
self.assertNotIn("docforge_apply_changeset", result["recommended_workflow"])
self.assertEqual(
"docforge_get_context",
result["session_contract"]["recommended_first_operation"]["tool"],
)
proposal = DocForgeService(
project,
proposal_writer="alpha-editor",
capability_mode_name="proposal",
).bootstrap()
self.assertIn("docforge_register_changes", proposal["recommended_workflow"])
self.assertIn("docforge_get_changeset_diff", proposal["recommended_workflow"])
self.assertNotIn("docforge_apply_changeset", proposal["recommended_workflow"])

5
tests/test_public_contract.py
View file

@ -77,6 +77,11 @@ PUBLIC_IMPORTS = {
"ProjectService",
"ProjectSnapshot",
),
"docforge.policy": (
"EffectivePolicyV1",
"capability_mode",
"compose_effective_policy",
),
"docforge.render_contract": (
"GenericHtmlRenderer",
"PreparedRender",