Add fresh-wheel adoption proof
This commit is contained in:
parent
f0dea0cedc
commit
983a55f156
3 changed files with 571 additions and 1 deletions
430
tools/milestone4_adoption.py
Normal file
430
tools/milestone4_adoption.py
Normal file
|
|
@ -0,0 +1,430 @@
|
|||
#!/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()
|
||||
Loading…
Add table
Add a link
Reference in a new issue