431 lines
14 KiB
Python
431 lines
14 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Run the offline Milestone 4 fresh-wheel adoption proof."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import hashlib
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import shutil
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
import tempfile
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import cast
|
||
|
|
|
||
|
|
PROOF_SCHEMA_VERSION = 1
|
||
|
|
COMMAND_TIMEOUT_SECONDS = 120
|
||
|
|
VALIDATION_TIMEOUT_SECONDS = 45
|
||
|
|
OPTIONAL_FRONTEND_MODULES = (
|
||
|
|
"tree_sitter",
|
||
|
|
"tree_sitter_cpp",
|
||
|
|
"tree_sitter_javascript",
|
||
|
|
"tree_sitter_typescript",
|
||
|
|
)
|
||
|
|
|
||
|
|
_VALIDATION_PROGRAM = r"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
import importlib.metadata
|
||
|
|
import importlib.util
|
||
|
|
import json
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from docforge.errors import DocForgeError
|
||
|
|
from docforge.index import ProjectIndex
|
||
|
|
from docforge.reference_mcp import create_reference_project
|
||
|
|
from mcp import ClientSession, StdioServerParameters
|
||
|
|
from mcp.client.stdio import stdio_client
|
||
|
|
|
||
|
|
|
||
|
|
def write_config(
|
||
|
|
root: Path,
|
||
|
|
*,
|
||
|
|
language: str,
|
||
|
|
source_roots: tuple[str, ...],
|
||
|
|
compilation_database: str | None = None,
|
||
|
|
) -> None:
|
||
|
|
config = root / ".docforge" / "reference-adapter.toml"
|
||
|
|
config.parent.mkdir(parents=True)
|
||
|
|
roots = ", ".join(json.dumps(item) for item in source_roots)
|
||
|
|
lines = [
|
||
|
|
"schema_version = 1",
|
||
|
|
'project_id = "fresh-wheel-adoption"',
|
||
|
|
'title = "Fresh wheel adoption proof"',
|
||
|
|
f"language = {json.dumps(language)}",
|
||
|
|
f"source_roots = [{roots}]",
|
||
|
|
]
|
||
|
|
if compilation_database is not None:
|
||
|
|
lines.append(f"compilation_database = {json.dumps(compilation_database)}")
|
||
|
|
config.write_text("\n".join((*lines, "")), encoding="utf-8")
|
||
|
|
|
||
|
|
|
||
|
|
async def retrieve(python: Path, root: Path) -> dict[str, object]:
|
||
|
|
parameters = StdioServerParameters(
|
||
|
|
command=str(python),
|
||
|
|
args=[
|
||
|
|
"-I",
|
||
|
|
"-m",
|
||
|
|
"docforge.reference_mcp",
|
||
|
|
"--project-root",
|
||
|
|
str(root),
|
||
|
|
"--capability-mode",
|
||
|
|
"read",
|
||
|
|
],
|
||
|
|
env={},
|
||
|
|
)
|
||
|
|
async with (
|
||
|
|
stdio_client(parameters) as streams,
|
||
|
|
ClientSession(*streams) as session,
|
||
|
|
):
|
||
|
|
await session.initialize()
|
||
|
|
tools = tuple(tool.name for tool in (await session.list_tools()).tools)
|
||
|
|
bootstrap = await session.call_tool("docforge_bootstrap", {})
|
||
|
|
search = await session.call_tool(
|
||
|
|
"docforge_search",
|
||
|
|
{"query": "Service", "limit": 5},
|
||
|
|
)
|
||
|
|
if search.isError or not search.structuredContent:
|
||
|
|
raise RuntimeError("fresh-wheel reference MCP search failed")
|
||
|
|
results = search.structuredContent.get("results")
|
||
|
|
if not isinstance(results, list) or not results:
|
||
|
|
raise RuntimeError("fresh-wheel reference MCP search returned no evidence")
|
||
|
|
first = results[0]
|
||
|
|
if not isinstance(first, dict) or not isinstance(first.get("node_id"), str):
|
||
|
|
raise RuntimeError("fresh-wheel reference MCP search returned invalid evidence")
|
||
|
|
node_id = first["node_id"]
|
||
|
|
node = await session.call_tool("docforge_get_node", {"node_id": node_id})
|
||
|
|
if bootstrap.isError or not bootstrap.structuredContent:
|
||
|
|
raise RuntimeError("fresh-wheel reference MCP bootstrap failed")
|
||
|
|
if node.isError or not node.structuredContent:
|
||
|
|
raise RuntimeError("fresh-wheel reference MCP node retrieval failed")
|
||
|
|
retrieved = node.structuredContent.get("node")
|
||
|
|
if not isinstance(retrieved, dict) or retrieved.get("node_id") != node_id:
|
||
|
|
raise RuntimeError("fresh-wheel reference MCP returned the wrong node")
|
||
|
|
return {
|
||
|
|
"bootstrap_status": bootstrap.structuredContent.get("status"),
|
||
|
|
"node_id": node_id,
|
||
|
|
"read_tool_count": len(tools),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
workspace = Path(sys.argv[1]).resolve(strict=True)
|
||
|
|
python = Path(sys.executable).absolute()
|
||
|
|
|
||
|
|
present_modules = [
|
||
|
|
name
|
||
|
|
for name in (
|
||
|
|
"tree_sitter",
|
||
|
|
"tree_sitter_cpp",
|
||
|
|
"tree_sitter_javascript",
|
||
|
|
"tree_sitter_typescript",
|
||
|
|
)
|
||
|
|
if importlib.util.find_spec(name) is not None
|
||
|
|
]
|
||
|
|
installed_frontends = sorted(
|
||
|
|
distribution.metadata["Name"]
|
||
|
|
for distribution in importlib.metadata.distributions()
|
||
|
|
if distribution.metadata["Name"].casefold().startswith("tree-sitter")
|
||
|
|
)
|
||
|
|
if present_modules or installed_frontends:
|
||
|
|
raise RuntimeError(
|
||
|
|
f"base wheel unexpectedly installed language frontends: "
|
||
|
|
f"{present_modules or installed_frontends}"
|
||
|
|
)
|
||
|
|
|
||
|
|
python_root = workspace / "python-project"
|
||
|
|
(python_root / "src" / "sample").mkdir(parents=True)
|
||
|
|
(python_root / "src" / "sample" / "__init__.py").write_text("", encoding="utf-8")
|
||
|
|
(python_root / "src" / "sample" / "service.py").write_text(
|
||
|
|
"class Service:\n"
|
||
|
|
" def execute(self, ready: bool) -> str:\n"
|
||
|
|
" if ready:\n"
|
||
|
|
' return "ready"\n'
|
||
|
|
' return "waiting"\n',
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
write_config(python_root, language="python", source_roots=("src",))
|
||
|
|
|
||
|
|
project = create_reference_project(python_root)
|
||
|
|
index = ProjectIndex(project)
|
||
|
|
built = index.build()
|
||
|
|
checked = index.check()
|
||
|
|
if built.get("status") != "ok" or checked.get("status") != "ok":
|
||
|
|
raise RuntimeError("fresh-wheel Python reference build/check failed")
|
||
|
|
mcp = asyncio.run(retrieve(python, python_root))
|
||
|
|
if mcp["bootstrap_status"] != "ok":
|
||
|
|
raise RuntimeError("fresh-wheel reference MCP bootstrap did not report ok")
|
||
|
|
|
||
|
|
cpp_root = workspace / "cpp-project"
|
||
|
|
(cpp_root / "src").mkdir(parents=True)
|
||
|
|
(cpp_root / "src" / "main.cpp").write_text("int main() { return 0; }\n", encoding="utf-8")
|
||
|
|
(cpp_root / "compile_commands.json").write_text(
|
||
|
|
json.dumps(
|
||
|
|
[
|
||
|
|
{
|
||
|
|
"directory": str(cpp_root),
|
||
|
|
"file": "src/main.cpp",
|
||
|
|
"arguments": ["c++", "-c", "src/main.cpp"],
|
||
|
|
}
|
||
|
|
],
|
||
|
|
sort_keys=True,
|
||
|
|
)
|
||
|
|
+ "\n",
|
||
|
|
encoding="utf-8",
|
||
|
|
)
|
||
|
|
write_config(
|
||
|
|
cpp_root,
|
||
|
|
language="cpp",
|
||
|
|
source_roots=("src",),
|
||
|
|
compilation_database="compile_commands.json",
|
||
|
|
)
|
||
|
|
|
||
|
|
try:
|
||
|
|
ProjectIndex(create_reference_project(cpp_root)).build()
|
||
|
|
except DocForgeError as error:
|
||
|
|
if error.code != "optional_dependency_missing":
|
||
|
|
raise
|
||
|
|
if error.details.get("install") != "docforge[cpp]":
|
||
|
|
raise RuntimeError("C++ missing-extra failure was not actionable") from error
|
||
|
|
cpp_failure = {
|
||
|
|
"code": error.code,
|
||
|
|
"install": error.details["install"],
|
||
|
|
"missing_module": error.details.get("missing_module"),
|
||
|
|
}
|
||
|
|
else:
|
||
|
|
raise RuntimeError("C++ reference build unexpectedly worked without docforge[cpp]")
|
||
|
|
|
||
|
|
print(
|
||
|
|
json.dumps(
|
||
|
|
{
|
||
|
|
"schema_version": 1,
|
||
|
|
"base_frontend_modules": present_modules,
|
||
|
|
"base_frontend_distributions": installed_frontends,
|
||
|
|
"python": {
|
||
|
|
"build_status": built["status"],
|
||
|
|
"check_status": checked["status"],
|
||
|
|
"node_count": checked["node_count"],
|
||
|
|
},
|
||
|
|
"mcp": mcp,
|
||
|
|
"cpp_without_extra": cpp_failure,
|
||
|
|
},
|
||
|
|
sort_keys=True,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
"""
|
||
|
|
|
||
|
|
|
||
|
|
class AdoptionProofError(RuntimeError):
|
||
|
|
"""The disposable adoption proof could not be completed."""
|
||
|
|
|
||
|
|
|
||
|
|
def _run_checked(
|
||
|
|
arguments: list[str],
|
||
|
|
*,
|
||
|
|
cwd: Path,
|
||
|
|
timeout: int,
|
||
|
|
environment: dict[str, str] | None = None,
|
||
|
|
) -> subprocess.CompletedProcess[str]:
|
||
|
|
result = subprocess.run(
|
||
|
|
arguments,
|
||
|
|
cwd=cwd,
|
||
|
|
env=environment,
|
||
|
|
text=True,
|
||
|
|
capture_output=True,
|
||
|
|
check=False,
|
||
|
|
timeout=timeout,
|
||
|
|
)
|
||
|
|
if result.returncode != 0:
|
||
|
|
detail = result.stderr.strip() or result.stdout.strip() or "no command output"
|
||
|
|
raise AdoptionProofError(
|
||
|
|
f"command failed with exit {result.returncode}: {arguments[0]}\n{detail}"
|
||
|
|
)
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
def find_wheel(directory: Path) -> Path:
|
||
|
|
wheels = tuple(sorted(directory.glob("docforge-*.whl")))
|
||
|
|
if len(wheels) != 1:
|
||
|
|
raise AdoptionProofError(f"expected exactly one DocForge wheel, found {len(wheels)}")
|
||
|
|
return wheels[0]
|
||
|
|
|
||
|
|
|
||
|
|
def parse_evidence(raw: str) -> dict[str, object]:
|
||
|
|
try:
|
||
|
|
loaded: object = json.loads(raw)
|
||
|
|
except json.JSONDecodeError as error:
|
||
|
|
raise AdoptionProofError("fresh-wheel validation returned invalid JSON") from error
|
||
|
|
if not isinstance(loaded, dict):
|
||
|
|
raise AdoptionProofError("fresh-wheel validation returned the wrong schema")
|
||
|
|
document = cast(dict[str, object], loaded)
|
||
|
|
if document.get("schema_version") != PROOF_SCHEMA_VERSION:
|
||
|
|
raise AdoptionProofError("fresh-wheel validation returned the wrong schema")
|
||
|
|
if document.get("base_frontend_modules") != []:
|
||
|
|
raise AdoptionProofError("fresh-wheel validation found optional frontend modules")
|
||
|
|
if document.get("base_frontend_distributions") != []:
|
||
|
|
raise AdoptionProofError("fresh-wheel validation found optional frontend distributions")
|
||
|
|
raw_python = document.get("python")
|
||
|
|
raw_mcp = document.get("mcp")
|
||
|
|
raw_cpp = document.get("cpp_without_extra")
|
||
|
|
python = cast(dict[str, object], raw_python) if isinstance(raw_python, dict) else None
|
||
|
|
mcp = cast(dict[str, object], raw_mcp) if isinstance(raw_mcp, dict) else None
|
||
|
|
cpp = cast(dict[str, object], raw_cpp) if isinstance(raw_cpp, dict) else None
|
||
|
|
if (
|
||
|
|
python is None
|
||
|
|
or python.get("build_status") != "ok"
|
||
|
|
or python.get("check_status") != "ok"
|
||
|
|
or mcp is None
|
||
|
|
or mcp.get("bootstrap_status") != "ok"
|
||
|
|
or cpp is None
|
||
|
|
or cpp.get("code") != "optional_dependency_missing"
|
||
|
|
or cpp.get("install") != "docforge[cpp]"
|
||
|
|
):
|
||
|
|
raise AdoptionProofError("fresh-wheel validation evidence is incomplete")
|
||
|
|
return document
|
||
|
|
|
||
|
|
|
||
|
|
def run_adoption_proof(repository_root: Path, *, uv: str = "uv") -> dict[str, object]:
|
||
|
|
"""Build and exercise a base wheel without using the network."""
|
||
|
|
|
||
|
|
root = repository_root.resolve(strict=True)
|
||
|
|
if not root.joinpath("pyproject.toml").is_file() or not root.joinpath("uv.lock").is_file():
|
||
|
|
raise AdoptionProofError("repository root lacks pyproject.toml or uv.lock")
|
||
|
|
|
||
|
|
offline_environment = {
|
||
|
|
**os.environ,
|
||
|
|
"UV_OFFLINE": "1",
|
||
|
|
"UV_PYTHON_DOWNLOADS": "never",
|
||
|
|
}
|
||
|
|
offline_environment.pop("PYTHONPATH", None)
|
||
|
|
with tempfile.TemporaryDirectory(prefix="docforge-m4-adoption-") as directory:
|
||
|
|
workspace = Path(directory).resolve()
|
||
|
|
distributions = workspace / "dist"
|
||
|
|
requirements = workspace / "base-requirements.txt"
|
||
|
|
virtual_environment = workspace / "venv"
|
||
|
|
python = virtual_environment / "bin" / "python"
|
||
|
|
|
||
|
|
_run_checked(
|
||
|
|
[
|
||
|
|
uv,
|
||
|
|
"build",
|
||
|
|
"--wheel",
|
||
|
|
"--offline",
|
||
|
|
"--no-python-downloads",
|
||
|
|
"--out-dir",
|
||
|
|
str(distributions),
|
||
|
|
str(root),
|
||
|
|
],
|
||
|
|
cwd=root,
|
||
|
|
timeout=COMMAND_TIMEOUT_SECONDS,
|
||
|
|
environment=offline_environment,
|
||
|
|
)
|
||
|
|
wheel = find_wheel(distributions)
|
||
|
|
_run_checked(
|
||
|
|
[
|
||
|
|
uv,
|
||
|
|
"export",
|
||
|
|
"--frozen",
|
||
|
|
"--offline",
|
||
|
|
"--no-dev",
|
||
|
|
"--no-emit-project",
|
||
|
|
"--output-file",
|
||
|
|
str(requirements),
|
||
|
|
],
|
||
|
|
cwd=root,
|
||
|
|
timeout=COMMAND_TIMEOUT_SECONDS,
|
||
|
|
environment=offline_environment,
|
||
|
|
)
|
||
|
|
exported = requirements.read_text(encoding="utf-8").casefold()
|
||
|
|
if any(module.replace("_", "-") in exported for module in OPTIONAL_FRONTEND_MODULES):
|
||
|
|
raise AdoptionProofError("locked base dependency export contains a language frontend")
|
||
|
|
_run_checked(
|
||
|
|
[
|
||
|
|
uv,
|
||
|
|
"venv",
|
||
|
|
"--python",
|
||
|
|
sys.executable,
|
||
|
|
"--no-project",
|
||
|
|
"--no-python-downloads",
|
||
|
|
str(virtual_environment),
|
||
|
|
],
|
||
|
|
cwd=workspace,
|
||
|
|
timeout=COMMAND_TIMEOUT_SECONDS,
|
||
|
|
environment=offline_environment,
|
||
|
|
)
|
||
|
|
_run_checked(
|
||
|
|
[
|
||
|
|
uv,
|
||
|
|
"pip",
|
||
|
|
"sync",
|
||
|
|
"--python",
|
||
|
|
str(python),
|
||
|
|
"--offline",
|
||
|
|
"--strict",
|
||
|
|
str(requirements),
|
||
|
|
],
|
||
|
|
cwd=workspace,
|
||
|
|
timeout=COMMAND_TIMEOUT_SECONDS,
|
||
|
|
environment=offline_environment,
|
||
|
|
)
|
||
|
|
_run_checked(
|
||
|
|
[
|
||
|
|
uv,
|
||
|
|
"pip",
|
||
|
|
"install",
|
||
|
|
"--python",
|
||
|
|
str(python),
|
||
|
|
"--offline",
|
||
|
|
"--strict",
|
||
|
|
"--no-deps",
|
||
|
|
str(wheel),
|
||
|
|
],
|
||
|
|
cwd=workspace,
|
||
|
|
timeout=COMMAND_TIMEOUT_SECONDS,
|
||
|
|
environment=offline_environment,
|
||
|
|
)
|
||
|
|
validation = _run_checked(
|
||
|
|
[str(python), "-I", "-c", _VALIDATION_PROGRAM, str(workspace)],
|
||
|
|
cwd=workspace,
|
||
|
|
timeout=VALIDATION_TIMEOUT_SECONDS,
|
||
|
|
environment=offline_environment,
|
||
|
|
)
|
||
|
|
evidence = parse_evidence(validation.stdout.strip())
|
||
|
|
evidence["wheel"] = {
|
||
|
|
"filename": wheel.name,
|
||
|
|
"sha256": hashlib.sha256(wheel.read_bytes()).hexdigest(),
|
||
|
|
}
|
||
|
|
evidence["network"] = "offline"
|
||
|
|
return evidence
|
||
|
|
|
||
|
|
|
||
|
|
def main() -> None:
|
||
|
|
parser = argparse.ArgumentParser(
|
||
|
|
description="Run the offline Milestone 4 fresh-wheel adoption proof"
|
||
|
|
)
|
||
|
|
parser.add_argument(
|
||
|
|
"--repository-root",
|
||
|
|
type=Path,
|
||
|
|
default=Path(__file__).resolve().parents[1],
|
||
|
|
)
|
||
|
|
parser.add_argument("--uv", default=shutil.which("uv") or "uv")
|
||
|
|
parser.add_argument("--output", type=Path)
|
||
|
|
arguments = parser.parse_args()
|
||
|
|
|
||
|
|
try:
|
||
|
|
evidence = run_adoption_proof(arguments.repository_root, uv=arguments.uv)
|
||
|
|
except (AdoptionProofError, OSError, subprocess.SubprocessError) as error:
|
||
|
|
print(json.dumps({"status": "error", "error": str(error)}, sort_keys=True))
|
||
|
|
raise SystemExit(1) from error
|
||
|
|
|
||
|
|
payload = json.dumps({"status": "ok", **evidence}, indent=2, sort_keys=True) + "\n"
|
||
|
|
if arguments.output is not None:
|
||
|
|
arguments.output.write_text(payload, encoding="utf-8")
|
||
|
|
print(payload, end="")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|