workinf_Blender_Wasm/my-tools/wbv2_preparation.py

"""Prepare bounded source/test pointers for the next local business adapter."""
import hashlib
import json
from pathlib import Path
import subprocess


def candidate_test_paths(root, candidates):
    """Keep preparation watches bounded and inside the local test tree."""
    if not isinstance(candidates, list):
        return ()
    root = Path(root).resolve()
    accepted = set()
    for candidate in candidates[:12]:
        if not isinstance(candidate, str) or len(candidate) > 512 or "\x00" in candidate:
            continue
        relative = Path(candidate)
        if relative.is_absolute() or ".." in relative.parts:
            continue
        if relative.parts[:3] not in (("web", "tests", "e2e"), ("web", "tests", "unit")):
            continue
        if len(relative.parts) < 4:
            continue
        try:
            path = root
            for component in relative.parts:
                path /= component
                if path.is_symlink():
                    break
            else:
                if not path.exists() or path.is_file():
                    accepted.add(relative.as_posix())
        except (OSError, ValueError):
            continue
    return tuple(sorted(accepted))


def preparation_test_paths(root, job):
    """Read only the small existing discovery report during idle polling."""
    import re

    identifier = job.get("id")
    if not isinstance(identifier, str) or not re.fullmatch(r"[A-Za-z0-9_-]+", identifier):
        return ()
    root = Path(root).resolve()
    relative = Path("tests/golden/WBV2/autopilot/preparation") / (identifier + ".json")
    try:
        path = root
        for component in relative.parts:
            path /= component
            if path.is_symlink():
                return ()
        if path.stat().st_size > 131072:
            return ()
        document = json.loads(path.read_text())
        if not isinstance(document, dict) or document.get("job") != identifier:
            return ()
    except (OSError, ValueError):
        return ()
    return candidate_test_paths(root, document.get("candidateTests"))


def _candidate_fingerprint(root, inputs, candidates):
    from wbv2_autopilot import sha

    values = {path: sha(Path(root) / path) for path in candidate_test_paths(root, candidates)}
    payload = {**inputs, "version": 3, "candidateTests": values}
    return hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()


def _blocker_input_paths(root, candidates):
    root = Path(root).resolve()
    accepted = set()
    for candidate in candidates[:64]:
        if not isinstance(candidate, str) or len(candidate) > 512 or "\x00" in candidate:
            continue
        relative = Path(candidate)
        if relative.is_absolute() or ".." in relative.parts or not relative.parts:
            continue
        if relative.parts[0] not in {"web", "blender-5.2.0", "my-tools", "tests"}:
            continue
        try:
            path = root
            for component in relative.parts:
                path /= component
                if path.is_symlink():
                    break
            else:
                if not path.exists() or path.is_file():
                    accepted.add(relative.as_posix())
        except (OSError, ValueError):
            continue
    return sorted(accepted)


def _business_blocker(root, report, previous):
    import re
    from wbv2_autopilot import now

    identifier = report["job"]
    if not isinstance(identifier, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,128}", identifier):
        raise ValueError("invalid preparation job identifier")
    commands = [value for value in report["commands"][:32]
                if isinstance(value, str) and re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]{0,127}", value)]
    units = [value for value in report["sourceUnits"][:16] if isinstance(value, dict)
             and isinstance(value.get("id"), str)
             and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9:_.-]{0,159}", value["id"])]
    tests = list(candidate_test_paths(root, report["candidateTests"]))
    if not units or report["missingSourceBindings"]:
        category = "source_binding_missing"
    elif not tests:
        category = "behavioral_test_missing"
    else:
        category = "workflow_contract_missing"
    search = None
    if units and commands:
        pattern = units[0]["id"].split(":")[-1]
        if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]{0,127}", pattern):
            pattern = commands[0]
        search = {"argv": ["node", "tools/web/search-wbv2-batch-source.mjs", "--unit", units[0]["id"],
                           "--pattern", pattern, "--max-hits", "8"], "cwd": "."}
    contract = "my-tools/local-workflows/" + identifier + ".json"
    proposed_test = "web/tests/e2e/" + identifier.lower() + "-ui-production.spec.ts"
    locations = [location.get("path") for unit in units if isinstance(unit.get("locations"), list)
                 for location in unit["locations"][:4] if isinstance(location, dict)]
    owned = _blocker_input_paths(root, list(report["ownedInputs"]))
    retry_on = _blocker_input_paths(root, [contract, "tests/golden/WBV2/dynamic-inventory.json",
                                          *tests, *owned[:16], *locations[:24], proposed_test])
    actions = []
    if search:
        actions.append({"kind": "inspect_source", "sourceUnit": units[0]["id"], **search})
    if category == "source_binding_missing":
        actions.append({"kind": "resolve_source_binding", "commands": commands,
                        "inventoryPath": "tests/golden/WBV2/dynamic-inventory.json"})
    actions.append({"kind": "implement_business_adapter", "commands": commands, "ownedPaths": owned})
    actions.append({"kind": "complete_behavioral_test", "paths": tests or [proposed_test],
                    "commands": commands, "requiredCases": ["positive", "negative", "undo_redo",
                                                            "save_reopen", "determinism"]})
    actions.append({"kind": "write_workflow_contract", "path": contract, "candidateTests": tests})
    timestamp = now()
    existing = previous.get("businessBlocker", {}) if previous.get("job") == identifier else {}
    first_seen = existing.get("firstSeenAt", timestamp) if isinstance(existing, dict) else timestamp
    report["state"] = "needs_implementation"
    report["nextActions"] = actions
    report["businessBlocker"] = {"state": "needs_implementation", "category": category,
                                  "firstSeenAt": first_seen, "inputChangedAt": timestamp,
                                  "inputFingerprint": report["fingerprint"], "sourceSearch": search,
                                  "candidateTests": tests, "contractPath": contract, "retryOn": retry_on}


def prepare_module(root, directory, job):
    from wbv2_autopilot import contained, read_json, sha, write_json

    root, directory = Path(root), Path(directory)
    import re
    if not isinstance(job.get("id"), str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,128}", job["id"]):
        raise ValueError("invalid preparation job identifier")
    commands = [value for value in job.get("commands", []) if isinstance(value, str)][:32]
    inputs = {path: sha(contained(root, path)) for path in job.get("ownedPaths", [])}
    test_directories = [root / "web/tests/e2e", root / "web/tests/unit"]
    test_listing = {str(path.relative_to(root)): path.stat().st_mtime_ns if path.is_dir() else None for path in test_directories}
    destination = directory / "preparation" / (job["id"] + ".json")
    previous = read_json(destination, {})
    fingerprint_inputs = {"commands": commands, "inputs": inputs, "inventory": sha(root / "tests/golden/WBV2/dynamic-inventory.json"), "testDirectories": test_listing, "version": 1}
    fingerprint = _candidate_fingerprint(root, fingerprint_inputs, previous.get("candidateTests", []))
    if previous.get("fingerprint") == fingerprint:
        return str(destination.relative_to(root))
    inventory = read_json(root / "tests/golden/WBV2/dynamic-inventory.json", {})
    wanted = {"web-command:" + command for command in commands}
    units = [{"id": unit["id"], "locations": unit.get("locations", [])} for unit in inventory.get("units", []) if unit["id"] in wanted][:16]
    tests = []
    directories = [str(path.relative_to(root)) for path in test_directories if path.is_dir()]
    if commands and directories:
        arguments = ["rg", "--files-with-matches", "--fixed-strings", "--glob", "*.{ts,mjs}"]
        for command in commands:
            arguments.extend(["--regexp", command])
        result = subprocess.run([*arguments, "--", *directories], cwd=root, capture_output=True, text=True, timeout=10)
        if result.returncode not in {0, 1}:
            raise RuntimeError("bounded local test discovery failed")
        tests = sorted(set(result.stdout.splitlines()))[:12]
    fingerprint = _candidate_fingerprint(root, fingerprint_inputs, tests)
    report = {
        "schemaVersion": 1, "job": job["id"], "module": job.get("module"),
        "state": "needs_implementation", "fingerprint": fingerprint,
        "workflow": job.get("title"), "commands": commands, "ownedInputs": inputs,
        "sourceUnits": units, "candidateTests": tests,
        "missingSourceBindings": [command for command in commands if "web-command:" + command not in {unit["id"] for unit in units}],
        "nextActions": [],
        "acceptance": "Source/test pointers only; candidates have not been executed or accepted by preparation",
    }
    _business_blocker(root, report, previous)
    write_json(destination, report)
    return str(destination.relative_to(root))