audita/python/audita

#!/usr/bin/env python3
import json
import os
from datetime import datetime
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Optional
from uuid import uuid4


DEFAULT_WORK_DIR = "/tmp/audita"
_SECRET_FLAGS = {"--llm-api-key", "--validation-llm-api-key"}


def _redact_argv(argv: list[str]) -> list[str]:
    redacted: list[str] = []
    index = 0
    while index < len(argv):
        arg = argv[index]
        matched_flag = next((flag for flag in _SECRET_FLAGS if arg == flag or arg.startswith(flag + "=")), None)
        if matched_flag is None:
            redacted.append(arg)
            index += 1
            continue
        if arg == matched_flag:
            redacted.append(arg)
            if index + 1 < len(argv):
                redacted.append("[REDACTED]")
                index += 2
            else:
                index += 1
            continue
        redacted.append(f"{matched_flag}=[REDACTED]")
        index += 1
    return redacted


def _resolve_work_root(argv: list[str]) -> Path:
    for index, arg in enumerate(argv):
        if arg == "--work-dir" and index + 1 < len(argv):
            return Path(argv[index + 1])
        if arg.startswith("--work-dir="):
            return Path(arg.split("=", 1)[1])
    return Path(os.environ.get("AUDITA_WORK_DIR") or DEFAULT_WORK_DIR)


def _create_run_dir(root: Path) -> Path:
    root.mkdir(parents=True, exist_ok=True)
    timestamp = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
    run_dir = root / f"run-{timestamp}-{uuid4().hex[:8]}"
    run_dir.mkdir(parents=False, exist_ok=False)
    return run_dir


def _capture_run_dirs(root: Path) -> set[str]:
    if not root.exists():
        return set()
    return {path.name for path in root.iterdir() if path.is_dir() and path.name.startswith("run-")}


def _find_new_run_dir(root: Path, before: set[str]) -> Optional[Path]:
    if not root.exists():
        return None
    candidates = [
        path for path in root.iterdir() if path.is_dir() and path.name.startswith("run-") and path.name not in before
    ]
    if not candidates:
        return None
    return max(candidates, key=lambda path: path.name)


def _write_launcher_error_log(
    path: Path,
    *,
    message: str,
    exit_code: int,
    argv: list[str],
    command: Optional[list[str]],
) -> None:
    payload = {
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "message": message,
        "exit_code": exit_code,
        "argv": argv,
        "cwd": os.getcwd(),
        "command": command,
    }
    path.write_text(
        "Audita Launcher Diagnostics\n" + json.dumps(payload, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )


def _emit_console_line(message: str) -> None:
    for stream in (sys.stderr, sys.stdout):
        if stream is None:
            continue
        try:
            stream.write(f"{message}\n")
            stream.flush()
            return
        except (OSError, ValueError):
            continue


def main() -> int:
    argv = list(sys.argv[1:])
    work_root = _resolve_work_root(argv)
    redacted_argv = _redact_argv(argv)
    uv = shutil.which("uv")
    if uv is None:
        run_dir = _create_run_dir(work_root)
        error_log = run_dir / "error.log"
        message = "uv is required to run this launcher. Install uv and run `uv sync` in the Audita project."
        _write_launcher_error_log(error_log, message=message, exit_code=1, argv=redacted_argv, command=None)
        _emit_console_line(f"audita: error: {message}")
        _emit_console_line("audita: exit code: 1")
        _emit_console_line(f"audita: run directory: {run_dir}")
        _emit_console_line(f"audita: error log: {error_log}")
        return 1

    project_root = Path(__file__).resolve().parent
    command = [uv, "run", "--project", str(project_root), "python", "-m", "audita", *sys.argv[1:]]
    before = _capture_run_dirs(work_root)
    result = subprocess.run(command, cwd=project_root, check=False)
    if result.returncode == 0:
        return 0
    if _find_new_run_dir(work_root, before) is None:
        run_dir = _create_run_dir(work_root)
        error_log = run_dir / "error.log"
        _write_launcher_error_log(
            error_log,
            message=f"Audita subprocess exited with status {result.returncode}.",
            exit_code=result.returncode,
            argv=redacted_argv,
            command=_redact_argv(command),
        )
        _emit_console_line(f"audita: subprocess exited with status {result.returncode}")
        _emit_console_line(f"audita: run directory: {run_dir}")
        _emit_console_line(f"audita: error log: {error_log}")
    return result.returncode


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