369 lines
14 KiB
Python
369 lines
14 KiB
Python
|
"""Rehearse an actual v1.0.0 project, index, and proposal under DocForge 1.4."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import hashlib
|
||
|
|
import io
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import sqlite3
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
import tarfile
|
||
|
|
import tempfile
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import cast
|
||
|
|
|
||
|
|
from docforge import __version__
|
||
|
|
from docforge.changesets import ChangesetStore
|
||
|
|
from docforge.command_reference import cli_command_references
|
||
|
|
from docforge.index import INDEX_SCHEMA_VERSION, ProjectIndex
|
||
|
|
from docforge.mcp_server import ALL_TOOLS, APPLICATION_TOOLS, SERVER_VERSION
|
||
|
|
from docforge.models import ProjectSnapshot
|
||
|
|
from docforge.project import Project
|
||
|
|
|
||
|
|
ROOT = Path(__file__).resolve().parents[1]
|
||
|
|
V1_TAG = "v1.0.0"
|
||
|
|
V1_RUNTIME_SCRIPT = r"""
|
||
|
|
import argparse
|
||
|
|
import hashlib
|
||
|
|
import json
|
||
|
|
import sqlite3
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import docforge
|
||
|
|
from docforge.changesets import ChangesetStore
|
||
|
|
from docforge.cli import _parser
|
||
|
|
from docforge.index import ProjectIndex
|
||
|
|
from docforge.mcp_server import ALL_TOOLS, APPLICATION_TOOLS, SERVER_VERSION
|
||
|
|
from docforge.project import Project
|
||
|
|
|
||
|
|
|
||
|
|
def digest_snapshot(snapshot):
|
||
|
|
document = {
|
||
|
|
"source_hash": snapshot.source_hash,
|
||
|
|
"revision": snapshot.revision,
|
||
|
|
"nodes": [
|
||
|
|
{
|
||
|
|
"id": node.node_id,
|
||
|
|
"title": node.title,
|
||
|
|
"family": node.family,
|
||
|
|
"authority": node.authority,
|
||
|
|
"status": node.status,
|
||
|
|
"tags": list(node.tags),
|
||
|
|
"summary": node.summary,
|
||
|
|
"content": node.content,
|
||
|
|
"source_path": node.source_path,
|
||
|
|
"source_anchor": node.source_anchor,
|
||
|
|
"content_hash": node.content_hash,
|
||
|
|
}
|
||
|
|
for node in snapshot.nodes
|
||
|
|
],
|
||
|
|
"edges": [
|
||
|
|
{
|
||
|
|
"source_id": edge.source_id,
|
||
|
|
"relation": edge.relation,
|
||
|
|
"target_id": edge.target_id,
|
||
|
|
}
|
||
|
|
for edge in snapshot.edges
|
||
|
|
],
|
||
|
|
}
|
||
|
|
encoded = json.dumps(document, sort_keys=True, separators=(",", ":")).encode()
|
||
|
|
return hashlib.sha256(encoded).hexdigest()
|
||
|
|
|
||
|
|
|
||
|
|
def digest_canonical(project):
|
||
|
|
digest = hashlib.sha256()
|
||
|
|
paths = (
|
||
|
|
project.descriptor.descriptor_path,
|
||
|
|
*project.descriptor.authority_files,
|
||
|
|
*project.canonical_source_paths(),
|
||
|
|
)
|
||
|
|
for path in sorted(
|
||
|
|
set(paths),
|
||
|
|
key=lambda value: value.relative_to(project.descriptor.root).as_posix(),
|
||
|
|
):
|
||
|
|
relative = path.relative_to(project.descriptor.root).as_posix()
|
||
|
|
digest.update(relative.encode())
|
||
|
|
digest.update(b"\0")
|
||
|
|
digest.update(path.read_bytes())
|
||
|
|
return digest.hexdigest()
|
||
|
|
|
||
|
|
|
||
|
|
def command_names():
|
||
|
|
parser = _parser()
|
||
|
|
for action in parser._actions:
|
||
|
|
if getattr(action, "dest", None) == "command":
|
||
|
|
return sorted(action.choices)
|
||
|
|
raise RuntimeError("v1 CLI command parser is missing")
|
||
|
|
|
||
|
|
|
||
|
|
root = Path(sys.argv[1])
|
||
|
|
project = Project.open(root)
|
||
|
|
snapshot = project.load()
|
||
|
|
canonical_before = digest_canonical(project)
|
||
|
|
index = ProjectIndex(project)
|
||
|
|
index_result = index.build()
|
||
|
|
with sqlite3.connect(index.path) as connection:
|
||
|
|
index_schema = connection.execute("PRAGMA user_version").fetchone()[0]
|
||
|
|
store = ChangesetStore(project, "alpha-editor")
|
||
|
|
created = store.create("m5-migration")
|
||
|
|
node = next(item for item in snapshot.nodes if item.node_id == "guide.workflow")
|
||
|
|
proposed = store.propose_update(
|
||
|
|
changeset_id="m5-migration",
|
||
|
|
expected_changeset_hash=created["changeset_hash"],
|
||
|
|
node_id=node.node_id,
|
||
|
|
expected_content_hash=node.content_hash,
|
||
|
|
metadata={"summary": "A migration-preserved v1 proposal."},
|
||
|
|
content=None,
|
||
|
|
relationship_changes=[],
|
||
|
|
rationale="Prove active proposal compatibility across the successor release.",
|
||
|
|
)
|
||
|
|
changeset_path = root / ".docforge/changesets/m5-migration.json"
|
||
|
|
print(json.dumps({
|
||
|
|
"package_metadata_version": "1.0.0",
|
||
|
|
"module_version": docforge.__version__,
|
||
|
|
"server_version": SERVER_VERSION,
|
||
|
|
"snapshot_hash": digest_snapshot(snapshot),
|
||
|
|
"source_hash": snapshot.source_hash,
|
||
|
|
"revision": snapshot.revision,
|
||
|
|
"canonical_hash": canonical_before,
|
||
|
|
"canonical_hash_after_proposal": digest_canonical(project),
|
||
|
|
"changeset_hash": proposed["changeset_hash"],
|
||
|
|
"changeset_file_hash": hashlib.sha256(changeset_path.read_bytes()).hexdigest(),
|
||
|
|
"index_schema": index_schema,
|
||
|
|
"index_action": index_result.get("action", "built"),
|
||
|
|
"cli_commands": command_names(),
|
||
|
|
"mcp_tools": sorted((*ALL_TOOLS, *APPLICATION_TOOLS)),
|
||
|
|
}, sort_keys=True, separators=(",", ":")))
|
||
|
|
"""
|
||
|
|
|
||
|
|
|
||
|
|
class MigrationProofError(RuntimeError):
|
||
|
|
"""The actual v1 migration rehearsal changed authoritative evidence."""
|
||
|
|
|
||
|
|
|
||
|
|
def _snapshot_hash(snapshot: ProjectSnapshot) -> str:
|
||
|
|
document = {
|
||
|
|
"source_hash": snapshot.source_hash,
|
||
|
|
"revision": snapshot.revision,
|
||
|
|
"nodes": [
|
||
|
|
{
|
||
|
|
"id": node.node_id,
|
||
|
|
"title": node.title,
|
||
|
|
"family": node.family,
|
||
|
|
"authority": node.authority,
|
||
|
|
"status": node.status,
|
||
|
|
"tags": list(node.tags),
|
||
|
|
"summary": node.summary,
|
||
|
|
"content": node.content,
|
||
|
|
"source_path": node.source_path,
|
||
|
|
"source_anchor": node.source_anchor,
|
||
|
|
"content_hash": node.content_hash,
|
||
|
|
}
|
||
|
|
for node in snapshot.nodes
|
||
|
|
],
|
||
|
|
"edges": [
|
||
|
|
{
|
||
|
|
"source_id": edge.source_id,
|
||
|
|
"relation": edge.relation,
|
||
|
|
"target_id": edge.target_id,
|
||
|
|
}
|
||
|
|
for edge in snapshot.edges
|
||
|
|
],
|
||
|
|
}
|
||
|
|
encoded = json.dumps(document, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
||
|
|
return hashlib.sha256(encoded).hexdigest()
|
||
|
|
|
||
|
|
|
||
|
|
def _canonical_hash(project: Project) -> str:
|
||
|
|
digest = hashlib.sha256()
|
||
|
|
paths = (
|
||
|
|
project.descriptor.descriptor_path,
|
||
|
|
*project.descriptor.authority_files,
|
||
|
|
*project.canonical_source_paths(),
|
||
|
|
)
|
||
|
|
for path in sorted(
|
||
|
|
set(paths),
|
||
|
|
key=lambda value: value.relative_to(project.descriptor.root).as_posix(),
|
||
|
|
):
|
||
|
|
relative = path.relative_to(project.descriptor.root).as_posix()
|
||
|
|
digest.update(relative.encode("utf-8"))
|
||
|
|
digest.update(b"\0")
|
||
|
|
digest.update(path.read_bytes())
|
||
|
|
return digest.hexdigest()
|
||
|
|
|
||
|
|
|
||
|
|
def _command_names() -> set[str]:
|
||
|
|
return {reference.name for reference in cli_command_references()}
|
||
|
|
|
||
|
|
|
||
|
|
def _extract_v1(destination: Path) -> Path:
|
||
|
|
completed = subprocess.run(
|
||
|
|
["git", "archive", "--format=tar", V1_TAG],
|
||
|
|
cwd=ROOT,
|
||
|
|
check=False,
|
||
|
|
capture_output=True,
|
||
|
|
)
|
||
|
|
if completed.returncode != 0:
|
||
|
|
raise MigrationProofError(completed.stderr.decode("utf-8", errors="replace").strip())
|
||
|
|
with tarfile.open(fileobj=io.BytesIO(completed.stdout), mode="r:") as archive:
|
||
|
|
archive.extractall(destination, filter="data")
|
||
|
|
return destination
|
||
|
|
|
||
|
|
|
||
|
|
def _run_v1(root: Path) -> dict[str, object]:
|
||
|
|
environment = dict(os.environ)
|
||
|
|
environment["PYTHONPATH"] = str(root / "src")
|
||
|
|
completed = subprocess.run(
|
||
|
|
[sys.executable, "-c", V1_RUNTIME_SCRIPT, str(root / "tests/fixtures/alpha")],
|
||
|
|
cwd=root,
|
||
|
|
env=environment,
|
||
|
|
check=False,
|
||
|
|
capture_output=True,
|
||
|
|
text=True,
|
||
|
|
)
|
||
|
|
if completed.returncode != 0:
|
||
|
|
raise MigrationProofError(f"v1 runtime rehearsal failed: {completed.stderr.strip()}")
|
||
|
|
try:
|
||
|
|
value: object = json.loads(completed.stdout)
|
||
|
|
except json.JSONDecodeError as error:
|
||
|
|
raise MigrationProofError("v1 runtime returned invalid evidence") from error
|
||
|
|
if not isinstance(value, dict):
|
||
|
|
raise MigrationProofError("v1 runtime returned an invalid evidence shape")
|
||
|
|
return cast(dict[str, object], value)
|
||
|
|
|
||
|
|
|
||
|
|
def _index_schema(index: ProjectIndex) -> int:
|
||
|
|
connection = sqlite3.connect(index.path)
|
||
|
|
try:
|
||
|
|
row = connection.execute("PRAGMA user_version").fetchone()
|
||
|
|
finally:
|
||
|
|
connection.close()
|
||
|
|
if row is None or type(row[0]) is not int:
|
||
|
|
raise MigrationProofError("Derived index schema is unavailable")
|
||
|
|
return cast(int, row[0])
|
||
|
|
|
||
|
|
|
||
|
|
def build_migration_evidence() -> dict[str, object]:
|
||
|
|
"""Run the tagged v1 implementation, then load its state through the current release."""
|
||
|
|
|
||
|
|
with tempfile.TemporaryDirectory(prefix="docforge-m5-migration-") as directory_name:
|
||
|
|
v1_root = _extract_v1(Path(directory_name) / "v1")
|
||
|
|
v1 = _run_v1(v1_root)
|
||
|
|
project_root = v1_root / "tests/fixtures/alpha"
|
||
|
|
project = Project.open(project_root)
|
||
|
|
snapshot = project.load()
|
||
|
|
index = ProjectIndex(project)
|
||
|
|
schema_before = _index_schema(index)
|
||
|
|
changeset_path = project_root / ".docforge/changesets/m5-migration.json"
|
||
|
|
changeset_file_before = changeset_path.read_bytes()
|
||
|
|
canonical_before = _canonical_hash(project)
|
||
|
|
current_snapshot_hash = _snapshot_hash(snapshot)
|
||
|
|
if current_snapshot_hash != v1.get("snapshot_hash"):
|
||
|
|
raise MigrationProofError("Current loading changed the tagged v1 graph")
|
||
|
|
if canonical_before != v1.get("canonical_hash"):
|
||
|
|
raise MigrationProofError("Current loading changed tagged v1 canonical sources")
|
||
|
|
if v1.get("canonical_hash") != v1.get("canonical_hash_after_proposal"):
|
||
|
|
raise MigrationProofError("Tagged v1 proposal mutated canonical sources")
|
||
|
|
store = ChangesetStore(project, "alpha-editor")
|
||
|
|
inspected = store.inspect("m5-migration")
|
||
|
|
if inspected.get("changeset_hash") != v1.get("changeset_hash"):
|
||
|
|
raise MigrationProofError("Current loading changed the tagged v1 proposal hash")
|
||
|
|
build = index.build()
|
||
|
|
schema_after = _index_schema(index)
|
||
|
|
if schema_after != INDEX_SCHEMA_VERSION:
|
||
|
|
raise MigrationProofError("Current release did not rebuild the legacy index schema")
|
||
|
|
if _snapshot_hash(project.load()) != current_snapshot_hash:
|
||
|
|
raise MigrationProofError("Index migration changed the canonical graph")
|
||
|
|
if _canonical_hash(project) != canonical_before:
|
||
|
|
raise MigrationProofError("Index migration changed canonical sources")
|
||
|
|
if changeset_path.read_bytes() != changeset_file_before:
|
||
|
|
raise MigrationProofError("Index migration changed the active v1 proposal")
|
||
|
|
v1_commands = set(cast(list[str], v1.get("cli_commands")))
|
||
|
|
v1_tools = set(cast(list[str], v1.get("mcp_tools")))
|
||
|
|
if not v1_commands <= _command_names():
|
||
|
|
raise MigrationProofError("Current CLI is missing a tagged v1 command")
|
||
|
|
if not v1_tools <= set((*ALL_TOOLS, *APPLICATION_TOOLS)):
|
||
|
|
raise MigrationProofError("Current MCP surface is missing a tagged v1 tool")
|
||
|
|
if __version__ != SERVER_VERSION:
|
||
|
|
raise MigrationProofError("Current package and MCP versions disagree")
|
||
|
|
evidence: dict[str, object] = {
|
||
|
|
"schema_version": 1,
|
||
|
|
"tag": V1_TAG,
|
||
|
|
"v1": v1,
|
||
|
|
"current": {
|
||
|
|
"version": __version__,
|
||
|
|
"server_version": SERVER_VERSION,
|
||
|
|
"snapshot_hash": current_snapshot_hash,
|
||
|
|
"source_hash": snapshot.source_hash,
|
||
|
|
"revision": snapshot.revision,
|
||
|
|
"canonical_hash": canonical_before,
|
||
|
|
"changeset_hash": inspected["changeset_hash"],
|
||
|
|
"changeset_file_hash": hashlib.sha256(changeset_file_before).hexdigest(),
|
||
|
|
"index_schema_before": schema_before,
|
||
|
|
"index_schema_after": schema_after,
|
||
|
|
"index_action": build.get("action", build.get("status")),
|
||
|
|
"cli_command_count": len(_command_names()),
|
||
|
|
"mcp_tool_count": len(set((*ALL_TOOLS, *APPLICATION_TOOLS))),
|
||
|
|
},
|
||
|
|
"proofs": {
|
||
|
|
"canonical_bytes_preserved": True,
|
||
|
|
"graph_preserved": True,
|
||
|
|
"proposal_preserved": True,
|
||
|
|
"legacy_index_rebuilt": True,
|
||
|
|
"cli_superset": True,
|
||
|
|
"mcp_superset": True,
|
||
|
|
"v1_version_mismatch_recorded": (
|
||
|
|
v1.get("package_metadata_version") == "1.0.0"
|
||
|
|
and v1.get("module_version") == "0.15.0"
|
||
|
|
and v1.get("server_version") == "0.15.0"
|
||
|
|
),
|
||
|
|
},
|
||
|
|
}
|
||
|
|
if not cast(dict[str, object], evidence["proofs"])["v1_version_mismatch_recorded"]:
|
||
|
|
raise MigrationProofError("The inherited v1 version mismatch was not reproduced")
|
||
|
|
return evidence
|
||
|
|
|
||
|
|
|
||
|
|
def _write_output(path: Path, payload: bytes) -> None:
|
||
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
||
|
|
descriptor, temporary_name = tempfile.mkstemp(prefix=".m5-migration-", dir=path.parent)
|
||
|
|
temporary = Path(temporary_name)
|
||
|
|
try:
|
||
|
|
with os.fdopen(descriptor, "wb") as handle:
|
||
|
|
handle.write(payload)
|
||
|
|
handle.flush()
|
||
|
|
os.fsync(handle.fileno())
|
||
|
|
os.replace(temporary, path)
|
||
|
|
directory_fd = os.open(path.parent, os.O_RDONLY)
|
||
|
|
try:
|
||
|
|
os.fsync(directory_fd)
|
||
|
|
finally:
|
||
|
|
os.close(directory_fd)
|
||
|
|
except Exception:
|
||
|
|
temporary.unlink(missing_ok=True)
|
||
|
|
raise
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> int:
|
||
|
|
parser = argparse.ArgumentParser()
|
||
|
|
parser.add_argument("--output", type=Path)
|
||
|
|
arguments = parser.parse_args()
|
||
|
|
try:
|
||
|
|
evidence = build_migration_evidence()
|
||
|
|
except (MigrationProofError, OSError, subprocess.SubprocessError) as error:
|
||
|
|
print(str(error), file=sys.stderr)
|
||
|
|
return 2
|
||
|
|
payload = json.dumps(evidence, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n"
|
||
|
|
if arguments.output is not None:
|
||
|
|
_write_output(arguments.output, payload)
|
||
|
|
print(payload.decode("utf-8"), end="")
|
||
|
|
return 0
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
raise SystemExit(main())
|