#!/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()