| """Trusted local plan for exact additional files from one pinned complete tree.""" |
| import hashlib |
| import json |
| import sys |
| from pathlib import Path |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| sys.path[:0] = [str(ROOT / "packages/core")] |
| from build_module_overlay import build_module_overlay, parse_module_source |
| from chameleon_core.adaptation_contracts import Intake |
| from chameleon_core.contracts import digest |
| from chameleon_core.entry_capture import capture_entry |
| from chameleon_core.module_resolution import POLICY, resolve_relative_import |
| from chameleon_core.repository_capture import fetch_bytes, target, tree_files, verify_package |
|
|
|
|
| def acquire_module_candidate(package, importer, *, policy, fetch=fetch_bytes): |
| """Bounded local acquire/capture/construct chain; does not verify behavior.""" |
| evidence = acquire_module_capture_plan(package, importer, policy=policy, fetch=fetch) |
| plan = evidence["plan"] |
| request = plan["intake"] |
| captured = package |
| if plan["status"] == "CAPTURE_REQUIRED": |
| captured = capture_entry(request["source"], request["repository_entry"], fetch, |
| additional_paths=request["repository_assets"]) |
| before, after = package["repository"], captured["repository"] |
| if any(before[key] != after[key] for key in ("source", "commit_sha", "tree_sha", "entry")): |
| raise ValueError("MODULE_RECAPTURE_IDENTITY_MISMATCH") |
| if any(after["files"].get(name) != record for name, record in before["files"].items()): |
| raise ValueError("MODULE_RECAPTURE_HELD_FILE_CHANGED") |
| candidate = build_module_overlay(captured, importer, policy=policy) |
| return {"package": captured, "candidate": candidate, "acquisition_evidence": evidence, |
| "status": "UNVERIFIED_CANDIDATE", "source_modified": False} |
|
|
|
|
| def acquire_module_capture_plan(package, importer, *, policy, fetch=fetch_bytes): |
| """One guarded metadata GET. Errors propagate; no retries or redirect bypass.""" |
| if policy != POLICY: |
| raise ValueError("EXPLICIT_MODULE_POLICY_REQUIRED") |
| verify_package(package) |
| repository = package["repository"] |
| owner, repo, _ = target(repository["source"]) |
| endpoint = f"https://api.github.com/repos/{owner}/{repo}/git/trees/{repository['tree_sha']}?recursive=1" |
| body = fetch(endpoint, 524288) |
| if not isinstance(body, bytes) or len(body) > 524288: |
| raise ValueError("MODULE_TREE_WIRE_BUDGET") |
| tree = json.loads(body) |
| plan = build_module_capture(package, importer, tree, policy=policy) |
| return {"plan": plan, "tree": tree, "acquisition": { |
| "url": endpoint, "external_reads": 1, "wire_bytes": len(body), |
| "response_sha256": hashlib.sha256(body).hexdigest(), "source_writes": False}} |
|
|
|
|
| def build_module_capture(package, importer, tree, *, policy): |
| if policy != POLICY: |
| raise ValueError("EXPLICIT_MODULE_POLICY_REQUIRED") |
| files = verify_package(package) |
| repository = package["repository"] |
| if not isinstance(tree, dict) or tree.get("sha") != repository["tree_sha"]: |
| raise ValueError("MODULE_TREE_BINDING_MISMATCH") |
| inventory = tree_files(tree, metadata_only=True) |
| |
| for name, record in repository["files"].items(): |
| if (name not in inventory or inventory[name]["sha"] != record["git_blob_sha"] |
| or inventory[name]["size"] != record["bytes"]): |
| raise ValueError("MODULE_HELD_TREE_MISMATCH") |
| if importer not in files or not importer.endswith((".js", ".mjs")): |
| raise ValueError("CAPTURED_IMPORTER_REQUIRED") |
| syntax = parse_module_source(files[importer]) |
| plans, missing = [], set() |
| for reference in syntax["references"]: |
| plan = resolve_relative_import(importer, reference["specifier"], inventory) |
| if plan["status"] != "RESOLVED": |
| raise ValueError("MODULE_" + plan["status"]) |
| plans.append(plan) |
| destination = plan["matches"][0]["path"] |
| if destination not in files: |
| missing.add(destination) |
| prior = repository.get("projection", {}).get("additional_local_paths", []) |
| paths = sorted(set(prior) | missing) |
| if len(paths) > 16: |
| raise ValueError("MODULE_CAPTURE_PATH_BUDGET") |
| intake = Intake(source=repository["source"] + "/tree/" + repository["commit_sha"], |
| source_type="repository", repository_scope="entry", repository_entry=repository["entry"], |
| repository_assets=paths, build_counterpart=True) |
| return {"policy": POLICY, "source_digest": package["digest"], "tree_sha": tree["sha"], |
| "inventory_digest": digest(tree), "syntax": syntax, "resolutions": plans, |
| "missing_paths": sorted(missing), "status": "CAPTURE_REQUIRED" if missing else "READY", |
| "intake": intake.model_dump(mode="json"), |
| "scope": "Direct imports only; no acquisition, recursive closure or behavior claim"} |
|
|
