| """Bounded syntax-only module discovery and optional pinned subset expansion.""" |
|
|
| import argparse |
| import json |
| import shutil |
| import sys |
| from pathlib import Path |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| sys.path.insert(0, str(ROOT / "packages/core")) |
| from chameleon_core.entry_capture import capture_entry |
| from chameleon_core.module_completion import complete_modules |
| from chameleon_core.repository_capture import safe_path, verify_package |
| from chameleon_core.windows_job import run_bounded |
|
|
|
|
| def plan_package(package, output): |
| output.mkdir(exist_ok=False) |
| node = shutil.which("node") |
| if not node: |
| raise ValueError("NODE_UNAVAILABLE") |
| runtime = run_bounded( |
| [node, str(ROOT / "scripts/module_dependency_driver.mjs")], json.dumps(package), ROOT |
| ) |
| stdout = runtime.pop("stdout") |
| (output / "runtime.json").write_text(json.dumps(runtime, indent=2), encoding="utf-8") |
| if ( |
| runtime.get("returncode") != 0 |
| or runtime.get("remaining_job_processes") != 0 |
| or any( |
| runtime.get(k) is not False |
| for k in ("timed_out", "cpu_budget_exhausted", "output_budget_exhausted") |
| ) |
| ): |
| raise ValueError("MODULE_PARSER_EXECUTION_FAILED") |
| plan = json.loads(stdout) |
| (output / "plan.json").write_text(json.dumps(plan, indent=2), encoding="utf-8") |
| return plan |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser() |
| parser.add_argument("package") |
| parser.add_argument("output") |
| parser.add_argument("--capture", action="store_true") |
| parser.add_argument("--transitive", action="store_true") |
| args = parser.parse_args() |
| source, output = Path(args.package).resolve(), Path(args.output).resolve() |
| pilot = ROOT / ".local/repository-pilot" |
| if ( |
| not source.is_relative_to(pilot) |
| or not output.is_relative_to(pilot) |
| or output == pilot |
| or source.stat().st_size > 8 * 1024 * 1024 |
| ): |
| raise ValueError("MODULE_PILOT_BOUNDARY") |
| package = json.loads(source.read_text(encoding="utf-8")) |
| verify_package(package) |
| output.mkdir(exist_ok=False) |
| if args.transitive: |
| if not args.capture: |
| raise ValueError("TRANSITIVE_REQUIRES_CAPTURE") |
|
|
| def save(candidate, index): |
| (output / f"package-{index}.json").write_text(json.dumps(candidate), encoding="utf-8") |
|
|
| save(package, 0) |
| try: |
| result = complete_modules( |
| package, lambda p, i: plan_package(p, output / f"round-{i}"), capture_entry, save |
| ) |
| except (ValueError, RuntimeError) as exc: |
| result = {"status": "FAILED", "reason": str(exc)[:500]} |
| (output / "completion.json").write_text(json.dumps(result, indent=2), encoding="utf-8") |
| print(json.dumps(result)) |
| if result["status"] != "LOCAL_DEPENDENCIES_CAPTURED": |
| raise SystemExit(1) |
| return |
| node = shutil.which("node") |
| if not node: |
| raise ValueError("NODE_UNAVAILABLE") |
| runtime = run_bounded( |
| [node, str(ROOT / "scripts/module_dependency_driver.mjs")], json.dumps(package), ROOT |
| ) |
| stdout = runtime.pop("stdout") |
| (output / "runtime.json").write_text(json.dumps(runtime, indent=2), encoding="utf-8") |
| if ( |
| runtime.get("returncode") != 0 |
| or runtime.get("remaining_job_processes") != 0 |
| or any( |
| runtime.get(k) is not False |
| for k in ("timed_out", "cpu_budget_exhausted", "output_budget_exhausted") |
| ) |
| ): |
| raise ValueError("MODULE_PARSER_EXECUTION_FAILED") |
| plan = json.loads(stdout) |
| if plan["package_digest"] != package["digest"] or len(plan["paths"]) > 16: |
| raise ValueError("MODULE_PLAN_BINDING_FAILED") |
| for name in plan["paths"]: |
| safe_path(name) |
| (output / "plan.json").write_text(json.dumps(plan, indent=2), encoding="utf-8") |
| if args.capture and plan["paths"]: |
| repo = package["repository"] |
| paths = sorted(set(plan["paths"]) | set(repo.get("projection", {}).get("additional_local_paths", []))) |
| expanded = capture_entry( |
| repo["source"] + "/tree/" + repo["commit_sha"], repo["entry"], additional_paths=paths |
| ) |
| if any(expanded["repository"]["files"].get(n) != v for n, v in repo["files"].items()): |
| raise ValueError("MODULE_CAPTURE_EXISTING_BYTES_CHANGED") |
| (output / "expanded.json").write_text(json.dumps(expanded), encoding="utf-8") |
| print(json.dumps(plan)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
