File size: 4,974 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 | """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)
# Check all held bytes against the same inventory, not just a caller's tree label.
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"}
|