"""The worker audit-sidecar contract, shared by Phase 7 and the mid-run check.

Phase 7 has always enforced this post-hoc, but by then the worker session is
gone and the only remedies left are editing the result after the fact — which
breaks the audit chain — or failing the run. `okstra worker-audit-check` runs
the same rules the moment a worker returns, while the worker is still listening
and can fix its own citation. Both consumers must agree on what a violation is,
so the rules live here rather than inside either one — the same split
`worker_heartbeat` makes for the heartbeat cadence.
"""
from __future__ import annotations

import json
import re
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path

from okstra_ctl.worker_prompt_headers import EVIDENCE_LEDGER_HEADER

# Worker-results filename pattern: `<worker-role>-<task-type>-<seq>.md`.
# Every analysis-worker role name ends in `-worker` (`claude-worker`,
# `codex-worker`, `antigravity-worker`, `report-writer-worker`), so anchor the
# split on that suffix — otherwise `antigravity-worker-error-analysis-001.md`
# ambiguously parses as `worker=antigravity, task=worker-error-analysis`.
# Audit sidecars (`*-audit-*`) and errors sidecars (`.json`) are not matched here.
_WORKER_RESULT_BASENAME_RE = re.compile(
    r"^(?P<worker>[a-z][a-z0-9-]*-worker)-(?P<task_type>[a-z][a-z-]*?)-(?P<seq>\d{3})\.md$"
)

READING_CONFIRMATION_HEADING_RE = re.compile(
    r"^##[ \t]+0\.[ \t]+Reading Confirmation\b", re.MULTILINE
)

# The row's prefix opens it and the end of the line closes it, so the backticks
# the template prescribes are decoration and are accepted either way. Requiring
# them once cost a run: a worker wrote six correct paths bare, none of the rows
# parsed, and every citation it made read as unbacked — a file it demonstrably
# opened counted as unread over markdown punctuation.
_EVIDENCE_READ_RE = re.compile(
    r"^- Evidence read: `?(?P<path>[^`\n]+?)`?\s*$",
    re.MULTILINE,
)
_EVIDENCE_COMMAND_RE = re.compile(
    r"^- Evidence command:[ \t]*(?P<payload>.*)$", re.MULTILINE
)
_EVIDENCE_COMMAND_FIELDS = ("command", "cwd", "exitCode", "outputSummary")
_EVIDENCE_COMMAND_IDENTITY_FIELDS = (
    "participantRef",
    "roleExecutionRef",
    "invocationRef",
    "attempt",
    "executionLabel",
)
_EVIDENCE_COMMAND_STRING_FIELDS = ("command", "cwd", "outputSummary")
# There is deliberately no secret detector here.
#
# One lived on this path and only ever cost runs. It matched the *name* half of
# an assignment, so `AUTH_MODE=off` and `TOKEN_TTL=60` were "sensitive material"
# and the row was refused; a worker that cannot record the command it ran has no
# way to satisfy the evidence ledger that demands it, and the run ends
# `contract-violated` over a switch. Judging the value half instead only moves
# the guesswork. What belongs in a row is the command's shape — the keys — and a
# value that must not be written down is the worker's to leave out or pass as a
# `$VAR` reference. A run does not need it: at execution time the value comes
# from the environment or from the person running it, and if it is absent the
# command fails then, where the failure is legible.
_FILE_LINE_CITATION_RE = re.compile(
    r"`(?P<path>(?!https?://)[^`\n]+?):(?P<line>\d+(?:-\d+)?)`"
)
_EXTENSIONLESS_SOURCE_FILENAMES = frozenset(
    {"Dockerfile", "Justfile", "Makefile", "Procfile", "Rakefile"}
)


def _looks_like_file_path(path: str) -> bool:
    if (
        not path
        or path.startswith(("-", "$"))
        or any(char.isspace() for char in path)
    ):
        return False
    if re.fullmatch(r"[0-9a-fA-F]{7,64}", path):
        return False
    # A dotted name is the usual filename signal, and an IPv4 literal is dots and
    # digits too. `127.0.0.1:13306` in a Read-only command log is a host and port —
    # the worker cannot record an Evidence read of it, so reading it as a citation
    # fails the run over something it has no way to satisfy.
    if re.fullmatch(r"\d{1,3}(?:\.\d{1,3}){3}", path):
        return False
    return (
        "/" in path
        or "." in Path(path).name
        or Path(path).name in _EXTENSIONLESS_SOURCE_FILENAMES
    )


def _cited_file_paths(content: str) -> set[str]:
    paths: set[str] = set()
    for match in _FILE_LINE_CITATION_RE.finditer(content):
        path = match.group("path")
        if _looks_like_file_path(path):
            paths.add(path)
    return paths


def _audit_evidence_read_paths(content: str) -> set[str]:
    return {
        match.group("path")
        for match in _EVIDENCE_READ_RE.finditer(content)
    }


@dataclass(frozen=True)
class EvidenceCommand:
    command: str
    cwd: str
    exit_code: int
    output_summary: str
    identity: dict[str, object] | None = None

    def to_record(self) -> dict[str, object]:
        record: dict[str, object] = {
            "command": self.command,
            "cwd": self.cwd,
            "exitCode": self.exit_code,
            "outputSummary": self.output_summary,
        }
        if self.identity:
            record.update(self.identity)
        return record


def _evidence_command_contract_failures(
    payload: Mapping[str, object], row_number: int
) -> list[str]:
    failures: list[str] = []
    missing = [key for key in _EVIDENCE_COMMAND_FIELDS if key not in payload]
    unexpected = sorted(
        set(payload) - set(_EVIDENCE_COMMAND_FIELDS) - set(_EVIDENCE_COMMAND_IDENTITY_FIELDS)
    )
    if missing:
        failures.append(
            f"Evidence command row {row_number} is missing {', '.join(missing)}"
        )
    if unexpected:
        failures.append(
            f"Evidence command row {row_number} has unexpected "
            f"{', '.join(unexpected)}"
        )
    for field_name in _EVIDENCE_COMMAND_STRING_FIELDS:
        if field_name in payload and not isinstance(payload[field_name], str):
            failures.append(
                f"Evidence command row {row_number} {field_name} must be a string"
            )
    if "exitCode" in payload and type(payload["exitCode"]) is not int:
        failures.append(
            f"Evidence command row {row_number} exitCode must be an integer"
        )
    return failures


def _evidence_command_from_payload(payload: Mapping[str, object]) -> EvidenceCommand:
    from okstra_ctl.execution_identity import stored_identity

    return EvidenceCommand(
        command=payload["command"],
        cwd=payload["cwd"],
        exit_code=payload["exitCode"],
        output_summary=payload["outputSummary"],
        identity=stored_identity(payload) or None,
    )


def parse_evidence_commands(
    content: str,
) -> tuple[tuple[EvidenceCommand, ...], tuple[str, ...]]:
    commands: list[EvidenceCommand] = []
    failures: list[str] = []
    for row_number, match in enumerate(
        _EVIDENCE_COMMAND_RE.finditer(content), start=1
    ):
        try:
            payload = json.loads(match.group("payload"))
        except json.JSONDecodeError:
            failures.append(f"Evidence command row {row_number} is not valid JSON")
            continue
        if not isinstance(payload, Mapping):
            failures.append(
                f"Evidence command row {row_number} must be a JSON object"
            )
            continue
        contract_failures = _evidence_command_contract_failures(payload, row_number)
        failures.extend(contract_failures)
        if contract_failures:
            continue
        commands.append(_evidence_command_from_payload(payload))
    return tuple(commands), tuple(failures)


def read_evidence_commands(
    audit_path: Path,
) -> tuple[tuple[EvidenceCommand, ...], tuple[str, ...]]:
    try:
        return parse_evidence_commands(audit_path.read_text(encoding="utf-8"))
    except OSError as exc:
        return (), (
            f"worker audit sidecar unreadable: {audit_path.name} ({exc})",
        )


def _worker_prompt_path(
    run_dir: Path,
    worker_role: str,
    task_type: str,
    seq: str,
) -> Path:
    return run_dir / "prompts" / f"{worker_role}-prompt-{task_type}-{seq}.md"


def _evidence_read_ledger_failures(
    *,
    run_dir: Path,
    worker_role: str,
    task_type: str,
    seq: str,
    result_name: str,
    result_content: str,
    audit_path: Path,
) -> list[str]:
    if worker_role == "report-writer-worker":
        return []
    prompt_path = _worker_prompt_path(run_dir, worker_role, task_type, seq)
    try:
        prompt_content = prompt_path.read_text(encoding="utf-8")
    except OSError:
        return []
    if EVIDENCE_LEDGER_HEADER not in prompt_content.splitlines():
        return []
    try:
        audit_content = audit_path.read_text(encoding="utf-8")
    except OSError as exc:
        return [f"worker audit sidecar unreadable: {audit_path.name} ({exc})"]

    read_paths = _audit_evidence_read_paths(audit_content)
    missing_paths = sorted(_cited_file_paths(result_content) - read_paths)
    return [
        f"worker `{worker_role}` result `{result_name}` cites "
        f"`{missing_path}:line` without an Evidence read row for "
        f"`{missing_path}` in `{audit_path.name}`"
        + _ledger_path_hint(missing_path, read_paths)
        for missing_path in missing_paths
    ]


def _ledger_path_hint(cited_path: str, read_paths: set[str]) -> str:
    """What to cite instead, when the ledger plainly holds the same file.

    A worker that cites the full path once and the bare filename afterwards
    reads to this check as never having opened the file, and the bare failure
    text sends the reader looking for a missing read that is not missing. One
    unambiguous ledger row makes the remedy a rewritten citation; two rows
    ending the same way make it a guess, so those say nothing.
    """
    if "/" in cited_path:
        return ""
    same_name = [path for path in read_paths if path.rsplit("/", 1)[-1] == cited_path]
    if len(same_name) != 1:
        return ""
    return (
        f" — the ledger records it as `{same_name[0]}`; cite that "
        f"project-relative path, not the bare filename"
    )


@dataclass(frozen=True)
class WorkerResultName:
    """The three fields a canonical worker-result basename carries."""

    worker_role: str
    task_type: str
    seq: str


def parse_worker_result_name(basename: str) -> WorkerResultName | None:
    """Split `<role>-worker-<task-type>-<seq>.md`, or None when non-canonical.

    Callers that already hold one worker's result path — the dispatcher settling
    that worker — read the audit check's arguments from here instead of
    re-deriving them from the manifest. The `worker_role` this returns carries
    the `-worker` suffix, which is what `check_worker_results_audit(worker=...)`
    matches on; a bare provider id like `claude` matches nothing.
    """
    if "-audit-" in basename:
        return None
    match = _WORKER_RESULT_BASENAME_RE.match(basename)
    if match is None:
        return None
    return WorkerResultName(
        worker_role=match.group("worker"),
        task_type=match.group("task_type"),
        seq=match.group("seq"),
    )


def normalize_worker_filter(worker: str | None) -> str | None:
    """Accept a worker id the way every other okstra surface spells it.

    Result files are named `<role>-worker-...`, so the filter matches
    `claude-worker`. But `claude` is what a worker is called everywhere a lead
    reads or types one — `--workers claude,codex`, `workerId`, the roster in the
    profile — so `--worker claude` was the natural thing to pass, and it matched
    nothing. A zero-match filter produces no failures, which the CLI reports as
    `{"ok": true}` with exit 0: indistinguishable from a real pass, at exactly
    the point `team-contract` tells the lead to run this check before deciding
    whether to re-dispatch. Accepting both spellings removes the trap rather
    than documenting it.
    """
    if worker is None or worker.endswith("-worker"):
        return worker
    return f"{worker}-worker"


def worker_result_files(
    run_dir: Path, task_type: str, seq: str | None, worker: str | None
):
    """Every worker-results file in *run_dir* this check owns, in name order."""
    worker = normalize_worker_filter(worker)
    for path in sorted((run_dir / "worker-results").glob("*.md")):
        match = parse_worker_result_name(path.name)
        if match is None:
            # Files that don't match the canonical pattern (e.g. ad-hoc notes
            # left by the operator) are out of contract scope.
            continue
        if match.task_type != task_type:
            # Cross-phase artifacts shouldn't appear here; skip rather than
            # fail to keep the check focused on the current phase.
            continue
        if seq is not None and match.seq != seq:
            # A prior run's artifact. Its contract was judged when it ran.
            continue
        if worker is not None and match.worker_role != worker:
            continue
        yield path, match.worker_role, match.seq


def check_worker_results_audit(
    run_dir: Path,
    task_type: str,
    seq: str | None,
    *,
    worker: str | None = None,
) -> list[str]:
    """Every audit-sidecar contract finding among this run's worker results.

    Blocking findings first, then the advisory citation-ledger ones. Mid-run
    callers want both in one list: the worker session is still alive and can fix
    either. Phase 7 separates them — see `worker_results_audit_findings`.
    """
    blocking, advisories = worker_results_audit_findings(
        run_dir, task_type, seq, worker=worker
    )
    return blocking + advisories


def worker_results_audit_findings(
    run_dir: Path,
    task_type: str,
    seq: str | None,
    *,
    worker: str | None = None,
) -> tuple[list[str], list[str]]:
    """The same checks, split into `(blocking, advisory)`.

    *run_dir* is `runs/<task-type>/`; `worker-results/` and `prompts/` hang off
    it. *seq* scopes the check to one run — `worker-results/` accumulates every
    run's artifacts, so scanning the whole directory judged a run by files it
    did not produce. Pass ``None`` only when the seq is genuinely unknown, which
    falls back to not filtering rather than silently checking nothing.

    Blocking: a result file that still carries the `## 0. Reading Confirmation`
    heading (that block moved to the sidecar), a missing sidecar, and a
    malformed Evidence command row. Each of those means the audit trail itself
    is absent or unreadable.

    Advisory: a `path:line` citation with no matching Evidence read row. The
    ledger exists and is readable; what the check found is a citation whose
    spelling does not match a row. That is worth reporting and not worth
    voiding a finished run over — one unmatched citation among hundreds used to
    take the whole phase to `contract-violated`, which stops the run without
    telling anyone whether the finding itself was wrong.
    """
    failures: list[str] = []
    advisories: list[str] = []
    if not (run_dir / "worker-results").is_dir():
        # No worker-results directory means no analysis workers ran (e.g.
        # `release-handoff`, which is single-lead). Nothing to enforce.
        return failures, advisories

    for path, worker_role, result_seq in worker_result_files(run_dir, task_type, seq, worker):
        rel = path.name
        try:
            content = path.read_text()
        except OSError as exc:
            failures.append(f"worker-results file unreadable: {rel} ({exc})")
            continue

        if READING_CONFIRMATION_HEADING_RE.search(content) is not None:
            failures.append(
                f"worker-results file `{rel}` contains a `## 0. Reading "
                f"Confirmation` heading — that block moved to the audit "
                f"sidecar (`{worker_role}-audit-{task_type}-{result_seq}.md`). "
                f"Remove the §0 heading + body from the main file and "
                f"write a fresh sidecar."
            )

        audit_path = (
            run_dir / "worker-results"
            / f"{worker_role}-audit-{task_type}-{result_seq}.md"
        )
        if not audit_path.exists():
            failures.append(
                f"worker `{worker_role}` produced `{rel}` but no audit sidecar "
                f"at `{audit_path.name}` — the sidecar must carry the Reading "
                f"Confirmation block (one short line per input file). Workers "
                f"write this in the same step as the main worker-results file."
            )
            continue

        _, command_failures = read_evidence_commands(audit_path)
        failures.extend(command_failures)
        advisories.extend(_evidence_read_ledger_failures(
            run_dir=run_dir,
            worker_role=worker_role,
            task_type=task_type,
            seq=result_seq,
            result_name=rel,
            result_content=content,
            audit_path=audit_path,
        ))
    return failures, advisories