Masterarbeit/.claude/skills/run-experiment/prepare-codex-workspace.py

#!/usr/bin/env python3
"""Create and verify an isolated Codex analysis workspace."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import shutil
import subprocess
import tempfile
from pathlib import Path


def fail(message: str) -> None:
    raise SystemExit(message)


def resolved(path: str) -> Path:
    return Path(path).resolve()


def native_path(path: Path) -> str:
    value = str(path)
    if os.name == "nt" and not value.startswith("\\\\?\\"):
        return "\\\\?\\" + value
    return value


def allowed_workspace(workspace: Path, meta: Path) -> bool:
    temporary_root = Path(tempfile.gettempdir()).resolve()
    return (
        (workspace.parent == meta and workspace.name == "workspace")
        or (workspace.parent == temporary_root and workspace.name.startswith("codex-experiment-"))
    )


def git_output(cwd: Path, *args: str) -> bytes:
    completed = subprocess.run(
        ["git", "-C", str(cwd), *args],
        check=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    return completed.stdout


def source_files(source: Path) -> tuple[list[Path], str | None]:
    try:
        top = resolved(git_output(source, "rev-parse", "--show-toplevel").decode().strip())
        relative_source = source.relative_to(top)
        raw = git_output(
            top,
            "ls-files",
            "-z",
            "--cached",
            "--others",
            "--exclude-standard",
            "--",
            relative_source.as_posix(),
        )
        paths = []
        for entry in raw.split(b"\0"):
            if not entry:
                continue
            candidate = resolved(top / os.fsdecode(entry))
            if candidate.is_file() and candidate.is_relative_to(source):
                paths.append(candidate)
        return sorted(set(paths), key=lambda item: item.as_posix().casefold()), str(top)
    except (subprocess.CalledProcessError, ValueError):
        paths = [item for item in source.rglob("*") if item.is_file() and ".git" not in item.parts]
        return sorted(paths, key=lambda item: item.as_posix().casefold()), None


def sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with open(native_path(path), "rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest().upper()


def manifest(workspace: Path) -> dict[str, dict[str, int | str]]:
    result: dict[str, dict[str, int | str]] = {}
    workspace_native = native_path(workspace)
    files: list[tuple[str, Path]] = []
    for directory, _, names in os.walk(workspace_native):
        for name in names:
            full_path = Path(directory) / name
            relative = os.path.relpath(str(full_path), workspace_native).replace("\\", "/")
            files.append((relative, full_path))
    for relative, path in sorted(files, key=lambda item: item[0].casefold()):
        result[relative] = {"size": os.stat(str(path)).st_size, "sha256": sha256(path)}
    return result


def write_json(path: Path, value: object) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")


def create(source_arg: str, workspace_arg: str, meta_arg: str) -> None:
    source = resolved(source_arg)
    workspace = resolved(workspace_arg)
    meta = resolved(meta_arg)
    if not source.is_dir():
        fail(f"Source directory does not exist: {source}")
    if not allowed_workspace(workspace, meta):
        fail("Workspace must be _meta/workspace or a codex-experiment-* directory in system temp")
    if workspace.exists():
        before_manifest = meta / "workspace-manifest-before.json"
        if workspace.parent == meta and workspace.name == "workspace" and not before_manifest.exists():
            shutil.rmtree(native_path(workspace))
        else:
            fail(f"Workspace already exists: {workspace}")

    files, git_root = source_files(source)
    workspace.mkdir(parents=True)
    for path in files:
        relative = path.relative_to(source)
        target = workspace / relative
        os.makedirs(native_path(target.parent), exist_ok=True)
        try:
            shutil.copy2(native_path(path), native_path(target), follow_symlinks=False)
        except OSError as error:
            fail(f"Failed to copy {path} -> {target}: {error}")

    before = manifest(workspace)
    write_json(meta / "workspace-manifest-before.json", before)
    write_json(
        meta / "workspace-source.json",
        {
            "source": str(source),
            "workspace": str(workspace),
            "git_root": git_root,
            "file_count": len(before),
            "total_bytes": sum(int(item["size"]) for item in before.values()),
        },
    )
    print(f"Created isolated workspace with {len(before)} files: {workspace}")


def verify(workspace_arg: str, meta_arg: str) -> None:
    workspace = resolved(workspace_arg)
    meta = resolved(meta_arg)
    before_path = meta / "workspace-manifest-before.json"
    if not workspace.is_dir() or not before_path.is_file():
        fail("Workspace or before-manifest is missing")

    before = json.loads(before_path.read_text(encoding="utf-8"))
    after = manifest(workspace)
    added = sorted(set(after) - set(before), key=str.casefold)
    removed = sorted(set(before) - set(after), key=str.casefold)
    modified = sorted(
        (path for path in set(before) & set(after) if before[path] != after[path]),
        key=str.casefold,
    )
    write_json(meta / "workspace-manifest-after.json", after)
    result = {
        "unchanged": not (added or removed or modified),
        "added": added,
        "removed": removed,
        "modified": modified,
        "file_count_before": len(before),
        "file_count_after": len(after),
    }
    write_json(meta / "workspace-integrity.json", result)
    print(json.dumps(result, ensure_ascii=False))
    if not result["unchanged"]:
        raise SystemExit(3)


def cleanup(workspace_arg: str, meta_arg: str) -> None:
    workspace = resolved(workspace_arg)
    meta = resolved(meta_arg)
    if not allowed_workspace(workspace, meta):
        fail("Refusing to remove a directory outside the approved experiment workspace locations")
    if workspace.exists():
        shutil.rmtree(native_path(workspace))
    print(f"Removed temporary workspace: {workspace}")


def main() -> None:
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest="command", required=True)
    create_parser = subparsers.add_parser("create")
    create_parser.add_argument("source")
    create_parser.add_argument("workspace")
    create_parser.add_argument("meta")
    verify_parser = subparsers.add_parser("verify")
    verify_parser.add_argument("workspace")
    verify_parser.add_argument("meta")
    cleanup_parser = subparsers.add_parser("cleanup")
    cleanup_parser.add_argument("workspace")
    cleanup_parser.add_argument("meta")
    args = parser.parse_args()
    if args.command == "create":
        create(args.source, args.workspace, args.meta)
    elif args.command == "verify":
        verify(args.workspace, args.meta)
    else:
        cleanup(args.workspace, args.meta)


if __name__ == "__main__":
    main()