"""Shared remote plumbing for every lane that touches a remote.

One module owns: which disposable remote a lane talks to (env resolution with

structured skip reasons), local-bare git remotes (the core-equivalent track),

GitHub remotes (the agent-default track), collision-safe disposable branch

names, and the transport-honesty fields every remote-touching row must carry.

Transport honesty (AGENTS.md, ADR-0005): a network row and a local-file row

are different measurements. Every row built from this module records

`remote_transport` and `remote_server`, and reports must never aggregate or

delta across transports. This module makes the asymmetry explicit; it cannot

make it disappear.

Skips are work items, not absences (ADR-0002): an unresolved remote yields a

structured reason naming the exact env var to set, for the lane to emit as a

returncode-77 skip row.

"""

from __future__ import annotations

import itertools

import os

import secrets

import subprocess

import sys

from collections.abc import Mapping

from dataclasses import dataclass

from pathlib import Path

from .environment import base_env

# Purpose-scoped env vars keep lanes from polluting each other's disposable

# repos (fleet branch storms never land in the sync lane's history). Most

# purposes fall back to the generic OAK_BENCH_REMOTE; mirror intentionally

# does not because mount-vs-clone needs a large mirror, not a tiny disposable

# repo used by other lanes.

OAK_REMOTE_ENV = "OAK_BENCH_REMOTE"

_BRANCH_COUNTER = itertools.count()

OAK_PURPOSE_ENVS: dict[str, str] = {

    "fleet": "OAK_BENCH_FLEET_REPO",

    "sync": "OAK_BENCH_SYNC_REPO",

    "mirror": "OAK_BENCH_MIRROR_REPO",

}

OAK_PURPOSES_WITHOUT_GENERIC_FALLBACK = {"mirror"}

GIT_GITHUB_ENV = "GIT_BENCH_REMOTE"

GIT_MIRROR_ENV = "GIT_BENCH_MIRROR_URL"

TRANSPORT_NETWORK = "network"

TRANSPORT_LOCAL_FILE = "local_file"

# netshape lane (Linux netns + veth + tc/netem): both subjects served from

# local disk through the same shaped pipe. A shaped row is neither a real

# network row nor a local-file row — it is its own transport, and it is only

# comparable to another shaped row through the SAME pipe (same profile, same

# server identity). shaped_comparison_legal() is the one place that rule

# lives; existing transport rules are unchanged.

TRANSPORT_NETWORK_SHAPED = "network_shaped"

@dataclass(frozen=True)

class RemoteResolution:

    """Where a lane's remote lives, or exactly why it doesn't."""

    repo: str | None  # ORG/REPO, URL, or filesystem path

    transport: str | None  # TRANSPORT_NETWORK | TRANSPORT_LOCAL_FILE

    server: str | None  # host identity recorded on rows (e.g. "oak.space")

    env_var: str  # the env var that resolved (or would resolve) it

    skip_reason: str | None = None

    @property

    def resolved(self) -> bool:

        return self.repo is not None

    def row_fields(self) -> dict[str, str]:

        """Transport-honesty fields for every row touching this remote."""

        if not self.resolved:

            raise ValueError(f"unresolved remote ({self.skip_reason}); emit a skip row instead")

        return {

            "remote_repo": self.repo,

            "remote_transport": self.transport,

            "remote_server": self.server,

        }

@dataclass(frozen=True)

class RemoteRequirement:

    """One lane's need for an Oak remote before the run starts."""

    purpose: str

    lane: str

    operations: tuple[str, ...]

@dataclass(frozen=True)

class RemotePreflightWarning:

    """A remote skip that can be predicted before any fixture work starts."""

    purpose: str

    lane: str

    env_var: str

    operations: tuple[str, ...]

    skip_reason: str

def resolve_oak_remote(

    purpose: str = "default",

    env: Mapping[str, str] | None = None,

) -> RemoteResolution:

    """Resolve the disposable oak ORG/REPO for a lane purpose.

    Checks the purpose-scoped env var first, then OAK_BENCH_REMOTE, except

    for mirror which must be explicitly configured. The server identity is the

    configured oak host (oak.space unless OAK_SERVER overrides it for a test

    rig).

    """

    source_env = os.environ if env is None else env

    candidates = []

    scoped = OAK_PURPOSE_ENVS.get(purpose)

    if scoped:

        candidates.append(scoped)

    if purpose not in OAK_PURPOSES_WITHOUT_GENERIC_FALLBACK:

        candidates.append(OAK_REMOTE_ENV)

    for env_var in candidates:

        repo = source_env.get(env_var, "").strip()

        if repo:

            return RemoteResolution(

                repo=repo,

                transport=TRANSPORT_NETWORK,

                server=source_env.get("OAK_SERVER", "oak.space"),

                env_var=env_var,

            )

    primary = candidates[0]

    return RemoteResolution(

        repo=None,

        transport=None,

        server=None,

        env_var=primary,

        skip_reason=(

            f"oak_remote_not_configured: set {primary} to a DISPOSABLE ORG/REPO "

            f"(e.g. oak/bench-{purpose}-tmp) to measure real-server operations"

        ),

    )

def oak_remote_preflight_warnings(

    requirements: list[RemoteRequirement],

    env: Mapping[str, str] | None = None,

) -> list[RemotePreflightWarning]:

    """Resolve remote requirements without mutating state.

    Skip rows are still emitted by the lanes themselves. This is an early,

    human-visible forecast so a long benchmark run does not wait minutes before

    revealing that an Oak remote env var was missing.

    """

    grouped: dict[tuple[str, str, str, str], list[str]] = {}

    for requirement in requirements:

        resolution = resolve_oak_remote(requirement.purpose, env=env)

        if resolution.resolved:

            continue

        key = (

            requirement.purpose,

            requirement.lane,

            resolution.env_var,

            resolution.skip_reason or "",

        )

        grouped.setdefault(key, []).extend(requirement.operations)

    return [

        RemotePreflightWarning(

            purpose=purpose,

            lane=lane,

            env_var=env_var,

            operations=tuple(dict.fromkeys(operations)),

            skip_reason=skip_reason,

        )

        for (purpose, lane, env_var, skip_reason), operations in sorted(grouped.items())

    ]

def format_remote_preflight_warnings(warnings: list[RemotePreflightWarning]) -> list[str]:

    if not warnings:

        return []

    lines = [

        "[preflight] Oak remote configuration missing; these lanes will emit skip rows:",

    ]

    for warning in warnings:

        operations = ", ".join(warning.operations[:6])

        if len(warning.operations) > 6:

            operations += f", +{len(warning.operations) - 6} more"

        lines.append(

            f"[preflight] - {warning.lane}/{warning.purpose}: set {warning.env_var}; "

            f"affected: {operations}; skip_reason: {warning.skip_reason}"

        )

    return lines

def print_remote_preflight_warnings(warnings: list[RemotePreflightWarning]) -> None:

    for line in format_remote_preflight_warnings(warnings):

        print(line, file=sys.stderr, flush=True)

def resolve_git_github_remote(env_var: str = GIT_GITHUB_ENV) -> RemoteResolution:

    """Resolve git's agent-default network remote (a GitHub URL or org/repo).

    This is the transport-comparable counterpart to oak's network remote: git

    pays a real `git push`/`git clone` over the wire like oak pays a real

    `oak push`. Core-equivalent runs use make_git_bare_remote() instead.

    """

    url = os.environ.get(env_var, "").strip()

    if not url:

        return RemoteResolution(

            repo=None,

            transport=None,

            server=None,

            env_var=env_var,

            skip_reason=(

                f"git_network_remote_not_configured: set {env_var} to a DISPOSABLE "

                "GitHub repo (URL or owner/repo) for the agent-default track"

            ),

        )

    server = "github.com"

    if "://" in url:

        server = url.split("://", 1)[1].split("/", 1)[0]

    elif "@" in url and ":" in url:

        server = url.split("@", 1)[1].split(":", 1)[0]

    return RemoteResolution(

        repo=url,

        transport=TRANSPORT_NETWORK,

        server=server,

        env_var=env_var,

    )

def make_git_bare_remote(git_bin: str | Path, dest_dir: Path, name: str = "remote.git") -> RemoteResolution:

    """Create an untimed local bare repo: git's core-equivalent remote.

    Local-file transport isolates object-transfer/checkout cost from network

    jitter. Setup is a precondition, never the measurement — callers must not

    time this call.

    """

    remote_dir = dest_dir / name

    subprocess.run(

        [str(git_bin), "init", "--bare", str(remote_dir)],

        env=base_env(),

        stdout=subprocess.DEVNULL,

        stderr=subprocess.DEVNULL,

        check=False,

    )

    return RemoteResolution(

        repo=str(remote_dir),

        transport=TRANSPORT_LOCAL_FILE,

        server="local",

        env_var="",

    )

def attach_git_origin(git_bin: str | Path, repo: Path, remote: RemoteResolution) -> None:

    """Untimed: point `origin` at the resolved remote (precondition, not measurement)."""

    subprocess.run(

        [str(git_bin), "remote", "add", "origin", str(remote.repo)],

        cwd=repo,

        env=base_env(),

        stdout=subprocess.DEVNULL,

        stderr=subprocess.DEVNULL,

        check=False,

    )

def current_git_branch(git_bin: str | Path, repo: Path) -> str:

    """Untimed probe of the checked-out branch name (defaults to main)."""

    probe = subprocess.run(

        [str(git_bin), "rev-parse", "--abbrev-ref", "HEAD"],

        cwd=repo,

        env=base_env(),

        capture_output=True,

        text=True,

        check=False,

    )

    return probe.stdout.strip() or "main"

def point_git_bare_head(git_bin: str | Path, remote: RemoteResolution, branch: str) -> None:

    """Untimed: point the bare remote's HEAD at a branch so clones check out

    a working tree (precondition, not measurement)."""

    subprocess.run(

        [str(git_bin), "--git-dir", str(remote.repo), "symbolic-ref", "HEAD", f"refs/heads/{branch}"],

        env=base_env(),

        stdout=subprocess.DEVNULL,

        stderr=subprocess.DEVNULL,

        check=False,

    )

def shaped_comparison_legal(row_a: dict, row_b: dict) -> bool:

    """Pure rule: two shaped rows may be compared only same-pipe.

    Legal iff BOTH rows carry remote_transport == TRANSPORT_NETWORK_SHAPED,

    the same non-empty netshape_profile, and the same non-empty remote_server.

    Anything else — one side unshaped, profile mismatch, missing fields — is

    illegal here; comparisons at other transports are governed by the existing

    transport-honesty rules, which this helper does not relax.

    """

    if row_a.get("remote_transport") != TRANSPORT_NETWORK_SHAPED:

        return False

    if row_b.get("remote_transport") != TRANSPORT_NETWORK_SHAPED:

        return False

    profile_a = row_a.get("netshape_profile")

    profile_b = row_b.get("netshape_profile")

    if not profile_a or profile_a != profile_b:

        return False

    server_a = row_a.get("remote_server")

    server_b = row_b.get("remote_server")

    return bool(server_a) and server_a == server_b

def disposable_branch(

    bench_id: str,

    scenario_name: str,

    subject_name: str,

    run_index: int,

    *,

    pid: int | None = None,

    token: str | None = None,

) -> str:

    """Disposable remote branch name: readable prefix plus per-run uniquifier."""

    if pid is None:

        pid = os.getpid()

    if token is None:

        token = f"{secrets.token_hex(4)}-{next(_BRANCH_COUNTER):x}"

    # Same-second parallel runs must never collide on the server.

    return (

        f"bench-{bench_id}-{scenario_name}-{subject_name}-r{run_index}"

        f"-{pid:x}-{token}"

    )