# SPDX-License-Identifier: AGPL-3.0-only
# Copyright (c) 2026 sol pbc

"""Shared isolation harness for API baseline tests and baseline regeneration.

Used by:
- tests/test_api_baselines.py - module-scoped fixtures
- tests/verify_api.py - verify/update CLI mode

Keeps both paths on identical isolation so generated baselines match the test
oracle.
"""

from __future__ import annotations

import atexit
import json
import os
import shutil
import subprocess
import tempfile
import threading
from contextlib import contextmanager
from pathlib import Path
from typing import Iterator
from unittest.mock import patch

FROZEN_DATE = "2026-04-15"
FROZEN_TZ_OFFSET = -7

# The fixture journal is built once per process from an immutable snapshot of
# HEAD rather than copied file-by-file from the live working tree. make dev /
# make sandbox write runtime artifacts into tests/fixtures/journal (AGENTS.md
# §6), so a concurrent writer can delete or replace a tracked file between
# `git ls-files` and the copy — raising FileNotFoundError at fixture *setup*
# (a spurious pytest ERROR, not a real failure). `git archive HEAD` reads the
# committed tree from the object store, never the working tree, so it is immune
# to concurrent working-tree mutation.
_REPO_ROOT = Path(__file__).resolve().parent.parent
_FIXTURE_JOURNAL_REL = "tests/fixtures/journal"
_LIVE_FIXTURE_JOURNAL = (_REPO_ROOT / _FIXTURE_JOURNAL_REL).resolve()

_snapshot_lock = threading.Lock()
_snapshot_root: Path | None = None


def _fixture_journal_snapshot() -> Path:
    """Return an immutable, process-scoped snapshot of the tracked fixture journal.

    Built once per process via `git archive HEAD` → tar, which reads HEAD from
    the git object store and is therefore unaffected by concurrent writes to the
    live tests/fixtures/journal tree. The same set of git-tracked files that
    `git ls-files` selects (working tree clean ⇒ HEAD ≡ index), with symlinks
    preserved as symlinks. The temp dir is removed at interpreter exit.
    """
    global _snapshot_root
    with _snapshot_lock:
        if _snapshot_root is not None and _snapshot_root.exists():
            return _snapshot_root
        tmp = Path(tempfile.mkdtemp(prefix="solstone-journal-snapshot-"))
        archive = subprocess.run(
            ["git", "archive", "HEAD", "--", _FIXTURE_JOURNAL_REL],
            cwd=str(_REPO_ROOT),
            capture_output=True,
            check=True,
        )
        subprocess.run(
            ["tar", "-x", "-C", str(tmp)],
            input=archive.stdout,
            check=True,
        )
        # `git archive` preserves the tests/fixtures/journal/ prefix in the tar.
        _snapshot_root = (tmp / _FIXTURE_JOURNAL_REL).resolve()
        atexit.register(shutil.rmtree, tmp, ignore_errors=True)
        return _snapshot_root


def copytree_tracked(src: Path, dst: Path) -> None:
    """Copy only git-tracked files from src to dst.

    For paths within tests/fixtures/journal (the journal fixture and any subtree
    of it), copy from an immutable, process-scoped snapshot of HEAD instead of
    enumerating and copying from the live working tree — see
    `_fixture_journal_snapshot` for why this defeats the concurrent-write race.
    For any other src, fall back to enumerating tracked files from the live tree.
    """
    src = Path(src).resolve()
    dst = Path(dst)
    try:
        rel = src.relative_to(_LIVE_FIXTURE_JOURNAL)
    except ValueError:
        rel = None
    if rel is not None:
        snap_src = _fixture_journal_snapshot() / rel
        shutil.copytree(snap_src, dst, symlinks=True, dirs_exist_ok=True)
        return
    result = subprocess.run(
        ["git", "ls-files", "."],
        cwd=str(src),
        capture_output=True,
        text=True,
        check=True,
    )
    for rel_path in result.stdout.splitlines():
        if not rel_path:
            continue
        src_file = src / rel_path
        dst_file = dst / rel_path
        dst_file.parent.mkdir(parents=True, exist_ok=True)
        if src_file.is_symlink():
            os.symlink(os.readlink(src_file), dst_file)
        else:
            shutil.copy2(src_file, dst_file)


def prepare_isolated_journal(dst: Path) -> Path:
    """Copy the git-tracked fixture journal into dst and return the absolute path."""
    src = Path("tests/fixtures/journal").resolve()
    dst = dst.resolve()
    copytree_tracked(src, dst)
    return dst


@contextmanager
def isolated_app_env(journal: Path) -> Iterator[Path]:
    """Patch env so create_app(journal) is fully isolated."""

    journal = Path(journal).resolve()
    overrides = {
        "SOLSTONE_JOURNAL": str(journal),
        # Match tests/conftest.py's deterministic unit-test runtime. Without
        # these, baseline regeneration probes host GPU/supervisor state while
        # test_api_baselines compares against the isolated test contract.
        "SOL_SKIP_SUPERVISOR_CHECK": "1",
        "SOLSTONE_DISABLE_CONVEY_SIDE_RUNTIMES": "1",
    }
    previous = {key: os.environ.get(key) for key in overrides}
    os.environ.update(overrides)
    try:
        from solstone.think.providers import local_cuda, local_install, local_vulkan

        deterministic_devices = [
            local_vulkan.VulkanDevice(
                1,
                "NVIDIA GeForce GTX 1660 Ti",
                local_vulkan.VK_TYPE_DISCRETE,
                6390,
            )
        ]
        with (
            patch.object(
                local_cuda,
                "probe_nvidia_gpu",
                lambda: local_cuda.NvidiaProbe(
                    index=None,
                    compute_cap=None,
                    driver_cuda_version=None,
                    vram_mib=None,
                    tiering_memory_mib=None,
                    memory_source=local_cuda.MEMORY_SOURCE_UNAVAILABLE,
                    detected=False,
                ),
            ),
            patch.object(
                local_install,
                "probe_cuda_runtime_artifact_trust",
                lambda _pin, **_kwargs: local_cuda.ArtifactTrust.ABSENT,
            ),
            patch.object(
                local_install,
                "has_persisted_installed_cuda_target",
                lambda **_kwargs: False,
            ),
            patch.object(
                local_vulkan,
                "detect_gpus",
                lambda: deterministic_devices,
            ),
        ):
            yield journal
    finally:
        for key, value in previous.items():
            if value is None:
                os.environ.pop(key, None)
            else:
                os.environ[key] = value


def make_test_client(journal: Path):
    """Create a Flask test client for an isolated journal."""
    from solstone.convey import create_app

    app = create_app(journal=str(Path(journal).resolve()))
    app.config["TESTING"] = True
    client = app.test_client()
    return client


def mark_setup_complete(journal: Path, completed_at: int = 1700000000000) -> None:
    """Mark a minimal test journal as past first-run setup."""

    config_path = Path(journal) / "config" / "journal.json"
    config_path.parent.mkdir(parents=True, exist_ok=True)
    if config_path.exists():
        config = json.loads(config_path.read_text(encoding="utf-8"))
    else:
        config = {}
    config["setup"] = {"completed_at": completed_at}
    config_path.write_text(json.dumps(config, indent=2) + "\n", encoding="utf-8")
