File size: 4,520 Bytes
09aeaeb | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 | """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()
|