"""Oracle and metric helpers for ``shallow_worker_branch_refresh_v1``.

The scenario asks: starting from a fresh shallow worker, create a named task

branch at the server's current ``main`` and perform the first parent refresh.

Success is judged from durable on-disk state and from the server, never from

the measured command's own output:

- Oak: ``.oak/oak.db`` is read with SQLite in query-only mode (current

  branch, branch head, stored commit rows, ancestry closure).

- Git: ``git rev-parse`` / ``symbolic-ref`` in the worktree (stock Git is the

  known-good reader for Git state).

- Expected head comes from the server (Oak branch API / bare repo rev-parse);

  expected tree comes from the fixture model, not from any VCS.

"""

from __future__ import annotations

import sqlite3

import subprocess

from pathlib import Path

from typing import Any

from oakbench.byteproxy import ByteProxyStats

from oakbench.httpcount import RequestStats

from oakbench.merge_history_fixture import tree_content_hash, worktree_files

SERVE_DEPTH_IGNORED = "serve_depth_ignored"

SERVE_DEPTH_HONORED = "serve_depth_honored"

def oak_disk_state(root: Path, branch: str) -> dict[str, Any]:

    """Read Oak checkout state from ``.oak/oak.db``; unobserved fields are None."""

    state: dict[str, Any] = {

        "current_branch": None,

        "head": None,

        "branch_head": None,

        "commit_rows": None,

        "ancestry_missing_edges": None,

        "error": None,

    }

    db = root / ".oak" / "oak.db"

    if not db.is_file():

        state["error"] = "oak_db_missing"

        return state

    try:

        # Same access pattern as task_loop.stat_cache_snapshot: a WAL database

        # cannot be opened with mode=ro before its -shm exists.

        connection = sqlite3.connect(str(db))

    except sqlite3.Error as error:

        state["error"] = f"oak_db_open:{error}"

        return state

    try:

        connection.execute("PRAGMA query_only = ON")

        meta = dict(connection.execute("SELECT key, value FROM metadata").fetchall())

        state["current_branch"] = meta.get("current_branch")

        state["head"] = meta.get("head")

        row = connection.execute(

            "SELECT head_hash FROM branch_heads WHERE branch_name = ?", (branch,)

        ).fetchone()

        state["branch_head"] = row[0] if row else None

        edges = connection.execute(

            "SELECT hash, parent_hash, merge_parent_hash FROM commits"

        ).fetchall()

    except sqlite3.Error as error:

        state["error"] = f"oak_db_read:{error}"

        return state

    finally:

        connection.close()

    state["commit_rows"] = len(edges)

    state["ancestry_missing_edges"] = missing_ancestry_edges(edges, state["branch_head"])

    return state

def missing_ancestry_edges(

    edges: list[tuple[str, str | None, str | None]], head: str | None

) -> int | None:

    """Count parent edges reachable from ``head`` whose commit row is absent."""

    if head is None:

        return None

    parents = {commit: (parent, merge) for commit, parent, merge in edges}

    missing = 0

    stack = [head]

    seen: set[str] = set()

    while stack:

        commit = stack.pop()

        if commit in seen:

            continue

        seen.add(commit)

        if commit not in parents:

            missing += 1

            continue

        stack.extend(parent for parent in parents[commit] if parent)

    return missing

def git_disk_state(root: Path, git_bin: str, env: dict[str, str] | None = None) -> dict[str, Any]:

    def read(*args: str) -> str | None:

        try:

            result = subprocess.run(

                [git_bin, *args], cwd=root, env=env, text=True,

                stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=30, check=False,

            )

        except (OSError, subprocess.TimeoutExpired):

            return None

        return result.stdout.strip() if result.returncode == 0 else None

    symbolic = read("symbolic-ref", "-q", "HEAD")

    commits = read("rev-list", "--count", "HEAD")

    return {

        "current_branch": symbolic.removeprefix("refs/heads/") if symbolic else None,

        "head": read("rev-parse", "HEAD"),

        "commit_rows": int(commits) if commits and commits.isdigit() else None,

        "shallow": read("rev-parse", "--is-shallow-repository"),

    }

def serve_depth_behavior(oak_commit_rows_after_clone: int | None) -> str | None:

    """Classify the served shallow clone from the rows it actually stored."""

    if oak_commit_rows_after_clone is None:

        return None

    return SERVE_DEPTH_HONORED if oak_commit_rows_after_clone <= 1 else SERVE_DEPTH_IGNORED

def evaluate_oracle(

    *,

    expected_branch: str,

    expected_head: str | None,

    expected_tree_hash: str,

    observed_branch: str | None,

    observed_branch_head: str | None,

    observed_head: str | None,

    worktree_tree_hash: str | None,

) -> dict[str, Any]:

    """Final-state oracle; an unobserved expected head can never pass."""

    checks = {

        "oracle_branch_current": observed_branch == expected_branch,

        "oracle_branch_head_is_server_main": bool(expected_head)

        and observed_branch_head == expected_head,

        "oracle_checkout_head_is_server_main": bool(expected_head) and observed_head == expected_head,

        "oracle_worktree_tree_matches_fixture": worktree_tree_hash == expected_tree_hash,

    }

    return {**checks, "oracle_passed": all(checks.values())}

def worktree_tree_hash(root: Path) -> str | None:

    if not root.is_dir():

        return None

    return tree_content_hash(worktree_files(root))

def request_metric_fields(

    stats: RequestStats | None, wire: ByteProxyStats | None, commits_info_suffix: str

) -> dict[str, Any]:

    """Per-step request/wire metrics; null (never 0) when unobserved."""

    return {

        "http_request_count": stats.request_count if stats else None,

        "http_commits_info_request_count": stats.count_suffix(commits_info_suffix) if stats else None,

        "http_requests_by_method": stats.methods if stats else None,

        "http_framing_errors": stats.framing_errors if stats else None,

        "injected_request_delay_ms": stats.injected_request_delay_ms if stats else None,

        "network_connections_observed": wire.connections if wire else None,

        "network_client_to_server_bytes_observed": wire.client_to_upstream_bytes if wire else None,

        "network_server_to_client_bytes_observed": wire.upstream_to_client_bytes if wire else None,

        "network_wire_bytes_observed": wire.total_bytes if wire else None,

        "request_measurement_source": "loopback_http1_request_framing_proxy" if stats else None,

    }

__all__ = [

    "SERVE_DEPTH_HONORED",

    "SERVE_DEPTH_IGNORED",

    "evaluate_oracle",

    "git_disk_state",

    "missing_ancestry_edges",

    "oak_disk_state",

    "request_metric_fields",

    "serve_depth_behavior",

    "worktree_tree_hash",

]