"""Append-only writer / reader for `consumers.jsonl` under a plan run's task root.

A stage's lifecycle position is the LAST lifecycle row written for it. An append
is redundant only when it would not move that position, so the same
(started / done / failed) record is never duplicated, while a `started` that
re-enters a stage whose last row is terminal does land.
force_reappend=True 인 보정 append 만 같은 tuple 을 다른 head_commit 으로 재기록할 수 있다."""

from __future__ import annotations

import json
import re
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List, Optional

from .json_boundary import JsonBoundaryError, load_owned_object
from .jsonl import append_jsonl
from .run_context import consumers_mutex

CONSUMERS_FILENAME = "consumers.jsonl"

# The rows that move a stage through its lifecycle. `started` claims the stage;
# `done` and `failed` both end that claim. `failed` is terminal WITHOUT
# completion: dependents stay blocked because the stage is not done, but the
# stage-key occupancy is released so the same stage number can be re-entered by
# a fix run. Without it a stage whose verifier returned FAIL stays `active`
# forever — `done` would be the only exit, and writing it would mark a stage
# carrying a confirmed regression as complete.
STAGE_LIFECYCLE_STATUSES = ("started", "done", "failed")

# The lead has already ruled on a stage whose last row is one of these — `done`
# closed it, `failed` deliberately did not. Neither may be re-derived from disk:
# closing a stage is the lead's verdict gate, not a file's existence.
_LEAD_SETTLED_STATUSES = ("done", "failed")


@dataclass(frozen=True)
class StageConsumerState:
    rows: List[Dict[str, Any]]
    done_rows: List[Dict[str, Any]]
    done_by_stage: Dict[int, Dict[str, Any]]
    done_stages: set[int]
    started_stages: set[int]
    verified_accepted_stages: set[int]
    pr_covered_stages: set[int]


def _path(plan_run_root: Path) -> Path:
    return plan_run_root / CONSUMERS_FILENAME


def read_consumers(plan_run_root: Path) -> List[Dict[str, Any]]:
    p = _path(plan_run_root)
    if not p.exists():
        return []
    out = []
    for line in p.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line:
            continue
        # 일반 open('a') append 는 crash/동시쓰기로 마지막 줄이 잘릴 수 있다.
        # 깨진 줄 하나가 stage 선택·reconcile 전체를 무너뜨리지 않도록,
        # worktree_registry._load 와 같은 방식으로 해당 줄만 건너뛴다.
        try:
            out.append(json.loads(line))
        except json.JSONDecodeError:
            continue
    return out


def latest_done_by_stage(rows: List[Dict[str, Any]]) -> Dict[int, Dict[str, Any]]:
    """stage → 마지막 done row. 리드가 `failed` 로 철회한 stage 는 제외한다.

    보정(reconciled) row 가 같은 stage 에 재-append 되므로 done 들 사이에서는
    last-wins 다. done 행만 훑던 동안은 뒤에 붙은 `failed` 가 아무 효과가 없어
    한 번 done 이 기록된 stage 를 되열 수단이 없었다 — `--stage N` 은 계속
    거부되고, 되돌리기가 필요해진 stage 를 okstra 밖에서 처리한 뒤 원장을 손으로
    맞춰야 했다. 그 보정을 놓치면 다음 stage 가 되돌리기 이전 트리에서 갈라진다.

    `started` 는 여기서 stage 를 내리지 않는다. fix run 이 done 인 stage 에
    `started` 를 다시 기록하는 것은 정상 흐름이고(`_equivalent_row_exists`),
    그동안 의존 stage 는 종전대로 마지막 done head 에서 base 를 잡는다.
    철회는 리드의 명시적 판정인 `failed` 하나로만 일어난다.
    """
    last = last_lifecycle_status_by_stage(rows)
    out: Dict[int, Dict[str, Any]] = {}
    for r in rows:
        if r.get("status") == "done" and isinstance(r.get("stage"), int):
            out[r["stage"]] = r
    return {stage: row for stage, row in out.items() if last.get(stage) != "failed"}


def read_stage_consumer_state(
    plan_run_root: Path, *, recover_from_carry: bool = False,
) -> StageConsumerState:
    if recover_from_carry:
        backfill_done_from_carry(plan_run_root)
    rows = read_consumers(plan_run_root)
    return stage_consumer_state_from_rows(rows)


def last_lifecycle_status_by_stage(
    rows: List[Dict[str, Any]],
) -> Dict[int, str]:
    """stage → 마지막 lifecycle row 의 status. 파일이 append-only 이므로
    읽기 순서가 곧 시간 순서다."""
    out: Dict[int, str] = {}
    for r in rows:
        stage = r.get("stage")
        status = r.get("status")
        if status in STAGE_LIFECYCLE_STATUSES and isinstance(stage, int):
            out[stage] = status
    return out


def stage_consumer_state_from_rows(rows: List[Dict[str, Any]]) -> StageConsumerState:
    done_rows = [r for r in rows if r.get("status") == "done"]
    done_by_stage = latest_done_by_stage(rows)
    last_status = last_lifecycle_status_by_stage(rows)
    return StageConsumerState(
        rows=rows,
        done_rows=done_rows,
        done_by_stage=done_by_stage,
        done_stages=set(done_by_stage.keys()),
        started_stages={n for n, s in last_status.items() if s == "started"},
        verified_accepted_stages=verified_accepted_stages(rows),
        pr_covered_stages=pr_covered_stages(rows),
    )


def append_consumer(plan_run_root: Path, *, impl_task_key: str, stage: int,
                    status: str, force_reappend: bool = False,
                    **fields: Any) -> None:
    if status not in STAGE_LIFECYCLE_STATUSES:
        allowed = " or ".join(repr(s) for s in STAGE_LIFECYCLE_STATUSES)
        raise ValueError(f"status must be {allowed}, got: {status!r}")
    with consumers_mutex(plan_run_root):
        if not _equivalent_row_exists(plan_run_root, impl_task_key, stage,
                                      status, force_reappend,
                                      fields.get("head_commit")):
            record: Dict[str, Any] = {
                "impl_task_key": impl_task_key,
                "stage": stage,
                "status": status,
                **fields,
            }
            _append_row(plan_run_root, record)
    if status == "done":
        _tag_stage_exit(plan_run_root, stage, fields.get("head_commit"))
    # 종결 status 는 점유 해제 이벤트이기도 하다 — 중복 append(no-op)에서도 풀어야
    # release 없이 done 만 기록된 과거 run 의 잔존 점유가 다음 호출에서 치유된다.
    if status == "done":
        _release_stage_reservation(impl_task_key, stage)
    elif status == "failed":
        _release_stage_occupancy_keeping_branch(impl_task_key, stage)


STAGE_EXIT_TAG = "stage-{stage}-exit"


def _repo_root(start: Path) -> Optional[Path]:
    for candidate in [start, *start.parents]:
        if (candidate / ".git").exists():
            return candidate
    return None


def _tag_stage_exit(plan_run_root: Path, stage: int, head_commit: Any) -> None:
    """Move `stage-<N>-exit` to the commit this ledger row records.

    The tag used to be a plan step, so it was written while the stage was still
    running — before the carry evidence the ledger reads. A stage then had three
    exit points that could disagree: the tag, the recorded `head_commit`, and
    the branch tip. okstra writes the tag at the moment it settles the stage, so
    the tag and the ledger cannot drift apart.

    Best effort by design: the ledger is the record and a tag is a convenience
    for the reader. A tree that is not a git repository, or a commit git cannot
    resolve, leaves the row written and says so on stderr.
    """
    if not isinstance(head_commit, str) or not head_commit.strip():
        return
    root = _repo_root(Path(plan_run_root).resolve())
    if root is None:
        return
    commit = head_commit.strip()
    tag = STAGE_EXIT_TAG.format(stage=stage)
    try:
        exists = subprocess.run(
            ["git", "-C", str(root), "cat-file", "-e", f"{commit}^{{commit}}"],
            capture_output=True,
        )
        if exists.returncode != 0:
            print(
                f"okstra: stage {stage} done recorded, but {commit[:12]} is not "
                f"a commit in {root} — `{tag}` not moved",
                file=sys.stderr,
            )
            return
        subprocess.run(
            ["git", "-C", str(root), "tag", "-f", tag, commit],
            capture_output=True,
            check=True,
        )
    except (OSError, subprocess.CalledProcessError) as exc:
        print(f"okstra: could not move `{tag}` to {commit[:12]}: {exc}", file=sys.stderr)


def _equivalent_row_exists(plan_run_root: Path, impl_task_key: str, stage: int,
                           status: str, force_reappend: bool,
                           head_commit: Any) -> bool:
    rows = read_consumers(plan_run_root)
    # 같은 tuple 이 이미 있어도, 그 뒤에 다른 lifecycle row 가 왔다면 이 append 는
    # stage 의 현재 위치를 옮기는 새 사실이다 — fix run 의 started 재기록이 그 경우다.
    if last_lifecycle_status_by_stage(rows).get(stage) != status:
        return False
    for row in rows:
        if (row.get("impl_task_key") == impl_task_key
                and row.get("stage") == stage
                and row.get("status") == status):
            if not force_reappend:
                return True
            if row.get("head_commit") == head_commit:
                return True  # 동일 보정의 중복 재-append 방지
    return False


def _stage_registry_coords(
    impl_task_key: str, stage: Any,
) -> Optional[tuple[str, str, str]]:
    """stage 점유의 registry 좌표. TASK_KEY(`project:group:task`) 각 segment 의
    safe-segment 와 같다(stage 예약이 그렇게 만들어진다). 형식이 다르면 점유
    주체가 아니므로 None."""
    parts = impl_task_key.split(":")
    if len(parts) != 3 or not isinstance(stage, int):
        return None
    from .ids import _safe_fs_segment
    return (_safe_fs_segment(parts[0]), _safe_fs_segment(parts[1]),
            _safe_fs_segment(parts[2]))


def _release_stage_reservation(impl_task_key: str, stage: Any) -> None:
    """done 이 기록된 stage 의 worktree-registry 점유(stage-key)를 해제하고
    브랜치 슬롯도 반납한다. worktree 디렉토리·브랜치 자체는 보존된다."""
    coords = _stage_registry_coords(impl_task_key, stage)
    if coords is None:
        return
    from . import worktree_registry
    worktree_registry.release(*coords, stage_number=stage)


def _release_stage_occupancy_keeping_branch(
    impl_task_key: str, stage: Any,
) -> None:
    """failed 이 기록된 stage 의 점유 표시만 푼다 — 브랜치 슬롯은 유지한다.

    fix run 은 같은 stage 워크트리·브랜치로 재진입하므로 슬롯이 계속 필요하다
    (`worktree_registry.release_status` 의 "slot is still needed" 경우)."""
    coords = _stage_registry_coords(impl_task_key, stage)
    if coords is None:
        return
    from . import worktree_registry
    worktree_registry.release_status(*coords, stage_number=stage)


def _append_row(plan_run_root: Path, record: Dict[str, Any]) -> None:
    append_jsonl(_path(plan_run_root), record, ensure_ascii=False, compact=False)


def append_verified(plan_run_root: Path, *, impl_task_key: str, stage: int,
                    verdict: str, report_path: str) -> None:
    """단독-stage final-verification 결과 기록. 같은 report_path 재기록은 멱등,
    다른 report_path 는 재검증으로 append 한다 (읽기는 last-wins)."""
    with consumers_mutex(plan_run_root):
        for row in read_consumers(plan_run_root):
            if (row.get("impl_task_key") == impl_task_key
                    and row.get("stage") == stage
                    and row.get("status") == "verified"
                    and row.get("report_path") == report_path):
                return
        _append_row(plan_run_root, {
            "impl_task_key": impl_task_key, "stage": stage,
            "status": "verified", "verdict": verdict,
            "report_path": report_path,
        })


def append_pr(plan_run_root: Path, *, impl_task_key: str, stages: List[int],
              branch: str, url: str) -> None:
    """handoff 의 PR 생성/재사용 기록. 같은 branch 의 pr 행이 이미 있으면 멱등."""
    with consumers_mutex(plan_run_root):
        for row in read_consumers(plan_run_root):
            if row.get("status") == "pr" and row.get("branch") == branch:
                return
        _append_row(plan_run_root, {
            "impl_task_key": impl_task_key, "stages": sorted(stages),
            "status": "pr", "branch": branch, "url": url,
        })


def verified_accepted_stages(rows: List[Dict[str, Any]]) -> set:
    """stage → 마지막 verified 행의 verdict 가 accepted 인 stage 집합 (last-wins)."""
    last: Dict[int, str] = {}
    for r in rows:
        if r.get("status") == "verified" and isinstance(r.get("stage"), int):
            last[r["stage"]] = (r.get("verdict") or "").strip().lower()
    return {n for n, v in last.items() if v == "accepted"}


def pr_covered_stages(rows: List[Dict[str, Any]]) -> set:
    out: set = set()
    for r in rows:
        if r.get("status") == "pr":
            out.update(n for n in (r.get("stages") or []) if isinstance(n, int))
    return out


# --- carry-as-SSOT done recovery ---------------------------------------------
#
# A stage's completion evidence is the verifier-authored sidecar at
# `runs/implementation/carry/stage-<N>.json`. The `done` row in consumers.jsonl
# is a derived index that the lead appends by hand (per the implementation
# profile) — so it can be missing even when the stage actually finished. The
# dependency gate (`stage_targets.resolve_stage_base_commit`) reads `done.head_commit`, so a
# missing `done` row wrongly blocks downstream stages. We treat the carry file
# as the source of truth and backfill the missing `done` rows from it before
# the gate runs. A stage with no carry, or an unfinished carry, is left blocked
# on purpose.


def _carry_stage_number(carry: Dict[str, Any], filename: str) -> Optional[int]:
    for key in ("stage", "stageNumber"):
        v = carry.get(key)
        if isinstance(v, int):
            return v
    m = re.search(r"stage-(\d+)", filename)
    return int(m.group(1)) if m else None


FAILED_CARRY_STATUSES = ("fail", "failed", "blocked", "error", "aborted")
"""carry 가 실패를 명시하는 status 값. 이 목록은 하나의 business fact 다 —
implementation_outcome 의 pass-grade 판정도 같은 목록을 본다."""


def _carry_is_complete(carry: Dict[str, Any]) -> bool:
    # A carry sidecar is written only after the stage's steps + Stage Validation
    # post commands all pass (spec §3.2), so its mere presence marks completion.
    # Treat it as complete unless it explicitly records a failure status. The
    # real backfill guard is whether a head commit can be extracted.
    status = carry.get("status")
    if status is not None and str(status).lower() in FAILED_CARRY_STATUSES:
        return False
    return True


def carry_head_commit(carry: Dict[str, Any]) -> str:
    rng = carry.get("stageCommitRange")
    if isinstance(rng, dict) and rng.get("head"):
        return str(rng["head"])
    for key in ("head_sha", "head_commit", "head"):
        v = carry.get(key)
        if v:
            return str(v)
    commits = carry.get("commits")
    if isinstance(commits, list) and commits:
        last = commits[-1]
        if isinstance(last, dict) and last.get("sha"):
            return str(last["sha"])
    return ""


def _carry_dir(plan_run_root: Path) -> Path:
    # consumers.jsonl lives at runs/implementation-planning/; the carry sidecars
    # live at the sibling runs/implementation/carry/.
    return plan_run_root.parent / "implementation" / "carry"


def backfill_done_from_carry(plan_run_root: Path) -> int:
    """Recover missing `done` rows from carry sidecars (carry is SSOT).

    Recovers the crash window between the lead writing the carry file and
    appending the `done` row: for every `runs/implementation/carry/stage-<N>.json`
    that is complete and whose stage has no terminal row yet, append a `done` row
    with the head commit read from the carry. Returns the number of rows
    recovered. Stages with no carry or an unfinished carry are skipped, so the
    dependency gate still legitimately blocks genuinely-unstarted stages.

    A stage the lead has already ruled on is never re-derived from disk. That
    includes `failed`: the executor emits its carry evidence when its own Tier 1/2
    validation passes, which happens BEFORE the verifier's gate, so a carry file
    exists for exactly the runs where the verifier then returned `FAIL`. Reading
    the file as completion would promote a stage carrying a confirmed regression
    to `done` and let its dependents proceed — and it would contradict this
    module's own rule that a stage's position is its last lifecycle row."""
    carry_dir = _carry_dir(plan_run_root)
    if not carry_dir.is_dir():
        return 0
    existing = read_consumers(plan_run_root)
    settled_stages = {
        stage
        for stage, status in last_lifecycle_status_by_stage(existing).items()
        if status in _LEAD_SETTLED_STATUSES
    }
    key_by_stage: Dict[Any, str] = {}
    fallback_key = ""
    for r in existing:
        k = r.get("impl_task_key")
        if k:
            fallback_key = k
            key_by_stage.setdefault(r.get("stage"), k)
    task_root = plan_run_root.parents[1]
    recovered = 0
    for cf in sorted(carry_dir.glob("stage-*.json")):
        try:
            carry = load_owned_object(cf, artifact="implementation stage carry")
        except (JsonBoundaryError, OSError):
            continue
        if not isinstance(carry, dict):
            continue
        stage = _carry_stage_number(carry, cf.name)
        if stage is None or stage in settled_stages:
            continue
        if not _carry_is_complete(carry):
            continue
        head = carry_head_commit(carry)
        if not head:
            continue
        impl_key = key_by_stage.get(stage) or carry.get("impl_task_key") or fallback_key
        if not impl_key:
            continue
        try:
            carry_path = str(cf.relative_to(task_root))
        except ValueError:
            carry_path = str(cf)
        append_consumer(
            plan_run_root,
            impl_task_key=impl_key,
            stage=stage,
            status="done",
            head_commit=head,
            carry_path=carry_path,
            source="carry-backfill",
        )
        settled_stages.add(stage)
        recovered += 1
    return recovered