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

Bind visualization status to snapshot freshness

Browse source
This commit is contained in:
Andraxion 2026-07-29 05:23:52 -04:00
parent 24bd13f9d9
commit 176b2d2784
9 changed files with 627 additions and 43 deletions
Download patch file Download diff file Expand all files Collapse all files

23
tests/test_adapter_contract.py
View file

@ -533,6 +533,20 @@ class AdapterContractTests(unittest.TestCase):
self.assertEqual(0, read_counters["adapter_source_extractions"])
self.assertEqual(0, read_counters["index_builds"])
pinned_state = project.incremental_state()
assert pinned_state is not None
loader.sources["guide.foundation"] = "Changed after viewer pin."
client = ViewerManagerClient(index)
self.assertEqual(
"stale",
client._source_state(
{
"revision": pinned_state.revision,
"source_hash": pinned_state.source_hash,
}
),
)
legacy = AdapterProject(
Loader(self.projection(root)),
cache_root=root / ".cache" / "legacy",
@ -544,6 +558,15 @@ class AdapterContractTests(unittest.TestCase):
self.assertEqual(1, legacy_counters["project_loads"])
self.assertEqual(1, legacy_counters["adapter_projection_loads"])
self.assertEqual(0, legacy_counters["adapter_source_extractions"])
self.assertEqual(
"unknown",
ViewerManagerClient(ProjectIndex(legacy))._source_state(
{
"revision": "legacy",
"source_hash": "0" * 64,
}
),
)
def test_incremental_adapter_reuses_sources_and_invalidates_reverse_dependencies(self) -> None:
with tempfile.TemporaryDirectory() as directory:

14
tests/test_mcp_server.py
View file

@ -193,7 +193,7 @@ class DocForgeMcpTests(unittest.IsolatedAsyncioTestCase):
finally:
service.visualization.stop()
for result in results:
for position, result in enumerate(results):
self.assertFalse(result.isError)
self.assertIsNotNone(result.structuredContent)
payload = result.structuredContent
@ -201,7 +201,10 @@ class DocForgeMcpTests(unittest.IsolatedAsyncioTestCase):
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"])
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"])
@ -222,6 +225,13 @@ class DocForgeMcpTests(unittest.IsolatedAsyncioTestCase):
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"])

255
tests/test_visualization.py
View file

@ -12,11 +12,14 @@ import urllib.parse
import urllib.request
from contextlib import contextmanager
from pathlib import Path
from unittest import mock
from docforge.errors import DocForgeError
from docforge.index import ProjectIndex
from docforge.mcp_server import DocForgeService
from docforge.models import ProjectState
from docforge.project import Project
from docforge.viewer_manager import ViewerManager, ViewerManagerClient
from docforge.viewer_manager import ViewerManager, ViewerManagerClient, _ManagedWorker
from docforge.visualization import (
_GRAPH_BROWSER_CSS,
_GRAPH_BROWSER_HTML,
@ -61,6 +64,256 @@ class VisualizationTests(unittest.TestCase):
manager.shutdown()
thread.join(timeout=2)
def test_snapshot_spec_binds_the_exact_validated_index_publication(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = self.copy_fixture("alpha", Path(directory))
index = ProjectIndex(Project.open(root))
index.build()
snapshot = VisualizationIndexSnapshot(index, index.check())
spec = snapshot.spec()
self.assertEqual(1, spec["schema_version"])
self.assertEqual(1, spec["index_signature"]["schema_version"])
self.assertEqual("current", VisualizationIndexSnapshot.from_spec(spec).index_state())
malformed = {**spec, "index_signature": {"schema_version": 1}}
with self.assertRaises(DocForgeError) as invalid:
VisualizationIndexSnapshot.from_spec(malformed)
self.assertEqual("invalid_index", invalid.exception.code)
wrong_fingerprint = {
**spec,
"identity": {
**spec["identity"],
"project_root_fingerprint": "0" * 16,
},
}
with self.assertRaises(DocForgeError) as invalid_fingerprint:
VisualizationIndexSnapshot.from_spec(wrong_fingerprint)
self.assertEqual("invalid_index", invalid_fingerprint.exception.code)
string_count = {
**spec,
"identity": {
**spec["identity"],
"node_count": str(spec["identity"]["node_count"]),
},
}
with self.assertRaises(DocForgeError) as invalid_count:
VisualizationIndexSnapshot.from_spec(string_count)
self.assertEqual("invalid_index", invalid_count.exception.code)
with index.path.open("ab") as stream:
stream.write(b"\n")
self.assertEqual("stale", snapshot.index_state())
with self.assertRaises(DocForgeError) as stale:
VisualizationIndexSnapshot.from_spec(spec)
self.assertEqual("visualization_stale", stale.exception.code)
def test_snapshot_index_state_rejects_missing_and_symlinked_publications(self) -> None:
for mutation in ("delete", "replace", "symlink"):
with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as directory:
root = self.copy_fixture("alpha", Path(directory))
index = ProjectIndex(Project.open(root))
index.build()
snapshot = VisualizationIndexSnapshot(index, index.check())
if mutation == "delete":
index.path.unlink()
elif mutation == "replace":
replacement = index.path.with_suffix(".replacement")
shutil.copy2(index.path, replacement)
replacement.replace(index.path)
else:
backup = index.path.with_suffix(".backup")
index.path.rename(backup)
index.path.symlink_to(backup)
self.assertEqual("stale", snapshot.index_state())
def test_health_is_stat_only_and_reports_stale_without_renewing_activity(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = self.copy_fixture("alpha", Path(directory))
index = ProjectIndex(Project.open(root))
index.build()
runner = VisualizationRunner(index)
try:
result = runner.start()
url = str(result["url"]).split("?", 1)[0] + "api/health"
with (
mock.patch(
"docforge.visualization.sqlite3.connect",
side_effect=AssertionError("health opened SQLite"),
),
urllib.request.urlopen(url, timeout=2) as response,
):
current = json.load(response)
self.assertEqual("current", current["index_state"])
with index.path.open("ab") as stream:
stream.write(b"\n")
with (
mock.patch(
"docforge.visualization.sqlite3.connect",
side_effect=AssertionError("health opened SQLite"),
),
urllib.request.urlopen(url, timeout=2) as response,
):
stale = json.load(response)
self.assertEqual("stale", stale["index_state"])
self.assertEqual(current["last_activity_at"], stale["last_activity_at"])
finally:
runner.stop()
def test_manager_health_rejects_malformed_and_identity_mismatched_payloads(self) -> None:
snapshot = {
"project_id": "alpha-docs",
"project_root_fingerprint": "0" * 16,
"revision": "revision",
"source_hash": "a" * 64,
"adapter": "generic",
"node_count": 3,
"edge_count": 2,
}
worker = _ManagedWorker(
process=mock.Mock(),
port=12345,
token="token",
snapshot=snapshot,
last_activity_at=1.0,
)
valid = {
"status": "ok",
"viewer": "alive",
"last_activity_at": 1.0,
"index_state": "current",
**snapshot,
}
invalid_payloads = (
{key: value for key, value in valid.items() if key != "status"},
{**valid, "project_id": "other"},
{**valid, "last_activity_at": True},
{**valid, "last_activity_at": float("nan")},
{key: value for key, value in valid.items() if key != "index_state"},
)
for payload in invalid_payloads:
with self.subTest(payload=payload):
response = mock.MagicMock()
response.__enter__.return_value.read.return_value = json.dumps(payload).encode()
with mock.patch(
"docforge.viewer_manager.urllib.request.urlopen",
return_value=response,
):
self.assertIsNone(ViewerManager._health(worker))
def test_manager_status_separates_lifecycle_index_and_source_freshness(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = self.copy_fixture("alpha", Path(directory))
project = Project.open(root)
index = ProjectIndex(project)
index.build()
state_path = Path(directory) / "viewer-manager.json"
with self.running_manager(state_path) as manager:
client = ViewerManagerClient(index, state_path=state_path)
first = client.start()
with (
mock.patch.object(
project,
"load",
side_effect=AssertionError("status loaded the project"),
),
mock.patch.object(
index,
"check",
side_effect=AssertionError("status checked the index"),
),
mock.patch.object(
index,
"build",
side_effect=AssertionError("status built the index"),
),
mock.patch.object(
index,
"synchronize",
side_effect=AssertionError("status synchronized the index"),
),
):
current = client.status()
self.assertEqual("running", current["state"])
self.assertEqual("current", current["snapshot_state"])
self.assertEqual(
{"index": "current", "source": "current"},
current["freshness"],
)
self.assertEqual(first["snapshot"]["source_hash"], current["source_hash"])
service = DocForgeService(project, diagnostics=True)
service.visualization = ViewerManagerClient(
service.index,
state_path=state_path,
)
mcp_current = service.visualization_status()
counters = mcp_current["diagnostics"]["counters"]
self.assertEqual(0, counters["project_loads"])
self.assertEqual(0, counters["source_files_parsed"])
self.assertEqual(0, counters["index_checks"])
self.assertEqual(0, counters["index_synchronizations"])
self.assertEqual(0, counters["index_builds"])
self.assertEqual(1, counters["viewer_manager_requests"])
project.generation_path.unlink()
unknown = client.status()
self.assertEqual("running", unknown["state"])
self.assertEqual("unknown", unknown["snapshot_state"])
self.assertEqual("unknown", unknown["freshness"]["source"])
index.build()
stale = client.status()
self.assertEqual("running", stale["state"])
self.assertEqual("stale", stale["snapshot_state"])
self.assertEqual("stale", stale["freshness"]["index"])
time.sleep(0.06)
self.assertEqual(1, len(manager._workers))
restarted = client.start()
self.assertFalse(restarted["reused"])
self.assertNotEqual(
str(first["url"]).split("?", 1)[0],
str(restarted["url"]).split("?", 1)[0],
)
self.assertEqual("current", client.status()["snapshot_state"])
def test_client_source_freshness_distinguishes_mismatch_and_unknown(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = self.copy_fixture("alpha", Path(directory))
project = Project.open(root)
index = ProjectIndex(project)
checked = index.build()
client = ViewerManagerClient(index)
response = {
"status": "ok",
"state": "running",
"index_state": "current",
"snapshot": {
"revision": checked["revision"],
"source_hash": checked["source_hash"],
},
"project_id": project.descriptor.project_id,
"project_root_fingerprint": "test",
"adapter": project.descriptor.adapter,
}
with mock.patch.object(client, "_lifecycle_request", return_value=response):
with mock.patch.object(
project,
"incremental_state",
return_value=ProjectState(source_hash="f" * 64, revision="changed"),
):
stale = client.status()
self.assertEqual("stale", stale["freshness"]["source"])
self.assertEqual("stale", stale["snapshot_state"])
with mock.patch.object(project, "incremental_state", return_value=None):
unknown = client.status()
self.assertEqual("unknown", unknown["freshness"]["source"])
self.assertEqual("unknown", unknown["snapshot_state"])
def test_overview_and_neighborhood_are_deterministic_and_bounded(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = self.copy_fixture("alpha", Path(directory))