FreeCADWasmDev/tools/read_implementation_source.py

#!/usr/bin/env python3
"""Read a bounded missing patch anchor from the current implementation packet."""
import argparse
import fcntl
import hashlib
import itertools
import json
import os
import re
import shlex
from pathlib import Path
import sys

ROOT = Path(__file__).resolve().parents[1]
PACKET = ROOT / "docs/current/module-implementation-packet.json"
MAX_WINDOWS = 4
CONTINUOUS_MAX_WINDOWS = 16
MAX_TOTAL_WINDOW_BYTES = 48000
MAX_WINDOW_BYTES = 12000


def window_limit():
    try:
        policy = json.loads((ROOT / "docs/current/execution-policy.json").read_text())
    except (OSError, ValueError):
        return MAX_WINDOWS
    return CONTINUOUS_MAX_WINDOWS if policy.get("continuousDevelopment") is True else MAX_WINDOWS


def source_limits(packet):
    limit = window_limit()
    count = len({row.get("apiId") for row in packet.get("records", [])
                 if isinstance(row, dict) and isinstance(row.get("apiId"), str) and row["apiId"]})
    scale = min(3, max(1, count)) if limit == CONTINUOUS_MAX_WINDOWS else 1
    return limit * scale, MAX_TOTAL_WINDOW_BYTES * scale


def packet_paths(packet):
    names = {row.get("path") for key in ("supportFiles", "editableExcerpts") for row in packet.get(key, [])}
    gap = packet.get("confirmedGap") or {}
    names.add(gap.get("source"))
    names.update(name for name in gap.get("sourceAnchors", []) if isinstance(name, str))
    for member in packet.get("confirmedGaps", []):
        names.add(member.get("source"))
        names.update(name for name in member.get("sourceAnchors", []) if isinstance(name, str))
    contract_name = packet.get("acceptanceContract")
    if isinstance(contract_name, str):
        contract_path = (ROOT / contract_name).resolve()
        if contract_path.is_relative_to(ROOT) and contract_path.is_relative_to((ROOT / "docs/business-acceptance").resolve()) and contract_path.is_file():
            try:
                contract = json.loads(contract_path.read_text(encoding="utf-8"))
                if isinstance(contract, dict):
                    names.add(contract_name)
                    command = contract.get("command", [])
                    if isinstance(command, list) and len(command) > 1 and isinstance(command[1], str):
                        names.add(command[1])
                    for key in ("dependencies", "caseDependencies"):
                        paths = contract.get(key, [])
                        if isinstance(paths, list):
                            names.update(name for name in paths if isinstance(name, str))
            except (OSError, ValueError):
                pass
    for row in packet.get("records", []):
        source = row.get("source")
        if source:
            names.add(source if (ROOT / source).is_file() else str(Path("freecad-source") / source))
        names.update(match.get("path") for match in row.get("nativeMatches", []))
        for gap in row.get("implementationGaps", []):
            names.update(match.get("path") for match in gap.get("sourceCandidates", []))
    return {(ROOT / name).resolve() for name in names if isinstance(name, str) and name}


def module_source_path(path, packet):
    """Permit named in-module implementation anchors before case declaration."""
    module = packet.get("module")
    if window_limit() != CONTINUOUS_MAX_WINDOWS or not isinstance(module, str) or not module.isalnum():
        return False
    if not path.is_relative_to(ROOT):
        return False
    relative = path.relative_to(ROOT)
    if packet.get("confirmedGaps") and relative.parent == Path("freecad-source/src/WebAssembly/FreeCADCoreProbe"):
        # Functional packages legitimately cross module boundaries through
        # existing document owners. Keep exact-file/window budgets, but don't
        # turn missing packet metadata into a user-authorization blocker.
        return path.suffix.lower() in {".cpp", ".h", ".hpp", ".py"} or path.name == "CMakeLists.txt"
    if relative.is_relative_to(Path("freecad-source/src/Mod") / module):
        return path.suffix.lower() in {".cpp", ".cxx", ".h", ".hpp", ".py", ".ui", ".xml", ".txt"}
    slug = module.lower()
    if relative.parent == Path("tools"):
        return path.name.startswith(slug + "-") and path.name.endswith("-business-cases.mjs")
    return (relative.parent == Path("web/src") and path.name.startswith(slug + "-")
            and path.suffix in {".ts", ".tsx", ".json"})


def frontend_dependency_path(path, packet):
    """Allow one exact relative import from a confirmed package's named sources.

    Workbench UI packages cross module boundaries through document facades and
    shared task components. Resolve only direct imports; this is not a recursive
    module crawl and it does not change the existing excerpt/turn budgets.
    """
    if window_limit() != CONTINUOUS_MAX_WINDOWS or not (packet.get("confirmedGaps") or packet.get("confirmedGap")):
        return False
    frontend = (ROOT / "web/src").resolve()
    if not path.is_relative_to(frontend) or path.suffix not in {".ts", ".tsx", ".json"}:
        return False
    for anchor in sorted(packet_paths(packet)):
        if (not anchor.is_relative_to(frontend) or anchor.suffix not in {".ts", ".tsx"}
                or not anchor.is_file() or anchor.stat().st_size > 512 * 1024):
            continue
        source = anchor.read_text(encoding="utf-8")
        for specifier in re.findall(r'''(?:\bfrom\s*|\bimport\s*(?:\(\s*)?)["'](\.[^"'\n]+)["']''', source):
            dependency = (anchor.parent / specifier).resolve()
            candidates = {dependency}
            if dependency.suffix == ".js":
                candidates.update(dependency.with_suffix(suffix) for suffix in (".ts", ".tsx"))
            elif not dependency.suffix:
                candidates.update(Path(str(dependency) + suffix) for suffix in (".ts", ".tsx", ".json"))
            if path in candidates:
                return True
    return False


def packet_summary(packet):
    # Task metadata is structured JSON, not a source file to slice by lines.
    result = {"batchId": packet.get("batchId"), "records": [], "supportPaths": [],
              "instruction": "This is task metadata, not source lines. Use sourceReadCommand below; changing --start on the packet does not read C++ or Python."}
    retained = packet.get("retainedBusinessResults", [])
    result["retainedBusinessResults"] = []
    for row in retained:
        result["retainedBusinessResults"].append(row)
        if len(json.dumps(result).encode("utf-8")) > MAX_WINDOW_BYTES // 2:
            result["retainedBusinessResults"].pop()
            break
    for row in packet.get("records", []):
        entry = {key: row[key] for key in ("apiId", "name", "source", "excerptLine") if key in row}
        if isinstance(row.get("source"), str) and row.get("excerptLine"):
            source = row["source"]
            if not (ROOT / source).is_file(): source = str(Path("freecad-source") / source)
            entry["sourceReadCommand"] = shlex.join([
                "python3", "tools/read_implementation_source.py", source,
                "--start", str(max(1, int(row["excerptLine"]))), "--lines", "60",
                "--reason", "Read selected command native source anchor",
            ])
        entry["gaps"] = [{key: gap[key] for key in ("stage", "classification", "reason") if key in gap}
                         for gap in row.get("implementationGaps", [])]
        result["records"].append(entry)
        if len(json.dumps(result).encode("utf-8")) > MAX_WINDOW_BYTES - 1000:
            result["records"].pop()
            break
    result["omittedRecords"] = len(packet.get("records", [])) - len(result["records"])
    for path in sorted(packet_paths(packet)):
        if not path.is_relative_to(ROOT) or not path.is_file():
            continue
        result["supportPaths"].append(str(path.relative_to(ROOT)))
        if len(json.dumps(result).encode("utf-8")) > MAX_WINDOW_BYTES - 200:
            result["supportPaths"].pop()
            break
    return json.dumps(result)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("path")
    parser.add_argument("--start", required=True, type=int)
    parser.add_argument("--lines", type=int, default=60)
    parser.add_argument("--reason", required=True)
    args = parser.parse_args()
    try:
        marker_value = os.environ.get("FREECAD_SOURCE_BUDGET_MARKER")
        if not marker_value:
            raise ValueError("This reader requires a scheduler-owned turn budget")
        if not 1 <= args.lines <= 100 or args.start < 1:
            raise ValueError("Use a positive --start and 1..100 --lines")
        if not args.reason.strip() or len(args.reason) > 240:
            raise ValueError("Explain the missing anchor in --reason (1..240 characters)")
        packet = json.loads(PACKET.read_text(encoding="utf-8"))
        path = (ROOT / args.path).resolve()
        metadata = path == PACKET.resolve()
        if path.is_relative_to(ROOT) and not path.is_file():
            raise ValueError("Requested source file does not exist or is not a file; locate the exact filename with a scoped file search")
        if not path.is_relative_to(ROOT) or (not metadata and path not in packet_paths(packet)
                                            and not module_source_path(path, packet)
                                            and not frontend_dependency_path(path, packet)) or not path.is_file():
            raise ValueError("Use an exact packet dependency, its direct frontend import, or current-module source/case path; no directory scans")
        if path.stat().st_size > 16 * 1024 * 1024:
            raise ValueError("This file exceeds the bounded source reader size limit")
        if metadata:
            excerpt = packet_summary(packet)
            args.start = 1
        else:
            with path.open(encoding="utf-8") as source:
                excerpt = "".join(itertools.islice(source, args.start - 1, args.start - 1 + args.lines))
        if not excerpt:
            raise ValueError("The requested anchor is beyond EOF; use the packet's named line")
        size = len(excerpt.encode("utf-8"))
        if size > MAX_WINDOW_BYTES:
            raise ValueError("Excerpt exceeds 12000 bytes; reduce --lines, do not dump a minified file")
        relative = str(path.relative_to(ROOT))
        fingerprint = hashlib.sha256(f"{relative}:{args.start}:{excerpt}".encode("utf-8")).hexdigest()
        # Per-turn file locking prevents parallel requests from exceeding the
        # bounded allowance. Repeated unchanged windows add no context.
        budget_path = Path(marker_value + ".windows.json")
        with budget_path.open("a+", encoding="utf-8") as budget:
            fcntl.flock(budget, fcntl.LOCK_EX)
            budget.seek(0)
            data = budget.read()
            state = json.loads(data) if data else {"batchId": packet["batchId"], "windows": [], "bytes": 0}
            if state["batchId"] != packet["batchId"]:
                raise ValueError("Batch changed during this turn; retain the current implementation checkpoint")
            if any(row["fingerprint"] == fingerprint for row in state["windows"]):
                print(json.dumps({"status": "already_in_context", "path": relative, "start": args.start,
                                  "instruction": "Reuse the unchanged excerpt already returned in this turn"}))
                return 0
            limit, byte_limit = source_limits(packet)
            if len(state["windows"]) >= limit or state["bytes"] + size > byte_limit:
                raise ValueError(f"Supplemental source budget reached ({limit} windows / {byte_limit} bytes); reuse retained excerpts and continue the identified repair")
            state["windows"].append({"path": relative, "start": args.start, "lines": args.lines,
                                     "fingerprint": fingerprint, "bytes": size, "reason": args.reason})
            state["bytes"] += size
            budget.seek(0)
            budget.truncate()
            budget.write(json.dumps(state, separators=(",", ":")))
            budget.flush()
        print(json.dumps({"status": "packet_summary" if metadata else "source_window", "batchId": packet["batchId"], "path": relative,
                          "start": args.start, "bytes": size, "remainingWindows": limit - len(state["windows"]),
                          "excerpt": excerpt}, ensure_ascii=False))
        return 0
    except (OSError, ValueError, KeyError) as error:
        print(f"IMPLEMENTATION_SOURCE_WINDOW_REJECTED: {error}", file=sys.stderr)
        return 64


if __name__ == "__main__":
    raise SystemExit(main())