"""Atomic attempt evidence seals and orchestrator-owned host event streams."""
from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Mapping, Sequence

from .domain.worker_stream import host_event_row
from .execution_identity import Attempt, ExecutionManifestError
from .execution_manifest import read_execution_manifest
from .execution_manifest import _replace_manifest, _write_manifest_over_existing
from .jsonl import append_jsonl
from .run_context import task_mutex


class AttemptEvidenceError(ValueError):
    """Raised when an attempt cannot be sealed from the observed terminal state."""


def _sha(data: bytes) -> str:
    return "sha256:" + hashlib.sha256(data).hexdigest()


def _utc_now() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


@dataclass(frozen=True)
class TerminalObservation:
    status: str
    source_diff_digest: str
    git_projection_digest: str
    changed_paths: tuple[str, ...]
    artifact_paths: tuple[Path, ...]
    host_event_stream_path: Path
    project_root: Path
    worktree: Path | None
    interrupted_before_record: bool = False


@dataclass(frozen=True)
class AttemptEvidenceSeal:
    seal_ref: str
    attempt_ref: str
    captured_at: str
    source_diff_digest: str
    git_projection_digest: str
    changed_paths: tuple[str, ...]
    host_event_stream: Mapping[str, Any]
    artifacts: tuple[Mapping[str, Any], ...]

    def to_payload(self) -> dict[str, Any]:
        return {
            "sealRef": self.seal_ref,
            "attemptRef": self.attempt_ref,
            "capturedAt": self.captured_at,
            "sourceDiffDigest": self.source_diff_digest,
            "gitProjectionDigest": self.git_projection_digest,
            "changedPaths": list(self.changed_paths),
            "hostEventStream": dict(self.host_event_stream),
            "hostEventStreamRef": self.host_event_stream.get("path"),
            "artifacts": [dict(row) for row in self.artifacts],
        }

    @classmethod
    def from_payload(cls, payload: Mapping[str, Any]) -> "AttemptEvidenceSeal":
        stream = payload.get("hostEventStream")
        if not isinstance(stream, Mapping):
            stream = {}
        return cls(
            seal_ref=str(payload["sealRef"]),
            attempt_ref=str(payload["attemptRef"]),
            captured_at=str(payload["capturedAt"]),
            source_diff_digest=str(payload["sourceDiffDigest"]),
            git_projection_digest=str(payload["gitProjectionDigest"]),
            changed_paths=tuple(payload.get("changedPaths") or ()),
            host_event_stream=dict(stream),
            artifacts=tuple(payload.get("artifacts") or ()),
        )


class HostEventStreamWriter:
    """Append-only JSONL writer for orchestrator-owned host events."""

    def __init__(
        self, path: Path, *, invocation_ref: str, attempt: int
    ) -> None:
        self.path = path
        self.invocation_ref = invocation_ref
        self.attempt = attempt
        self._sequence = 0
        path.parent.mkdir(parents=True, exist_ok=True)

    def append(self, event_type: str, payload: Mapping[str, Any]) -> dict[str, Any]:
        self._sequence += 1
        row = host_event_row(
            event_id=f"evt-{self._sequence:04d}",
            sequence=self._sequence,
            event_type=event_type,
            invocation_ref=self.invocation_ref,
            attempt=self.attempt,
            payload=payload,
            timestamp=_utc_now(),
        )
        append_jsonl(self.path, row, ensure_ascii=False, compact=False)
        return row


def _split_attempt_ref(attempt_ref: str) -> tuple[str, int]:
    invocation_ref, sep, raw = attempt_ref.rpartition(":")
    kind, dash, number = raw.partition("-")
    if (
        not sep
        or kind != "attempt"
        or dash != "-"
        or not invocation_ref
        or not number.isdigit()
    ):
        raise AttemptEvidenceError("attemptRef is malformed")
    return invocation_ref, int(number)


def _artifact_rows(paths: Sequence[Path]) -> tuple[dict[str, Any], ...]:
    rows: list[dict[str, Any]] = []
    for path in paths:
        if path.is_symlink():
            raise AttemptEvidenceError("sealed artifact must not be a symlink")
        data = path.read_bytes() if path.is_file() else b""
        rows.append({
            "kind": "artifact",
            "normalizedPath": str(path),
            "byteLength": len(data),
            "sha256": _sha(data),
        })
    return tuple(rows)


def _stream_payload(path: Path) -> dict[str, Any]:
    data = path.read_bytes() if path.is_file() else b""
    sequences = []
    event_ids = []
    for line in data.decode("utf-8").splitlines():
        if not line.strip():
            continue
        row = json.loads(line)
        sequences.append(int(row["sequence"]))
        event_ids.append(str(row["eventId"]))
    low = sequences[0] if sequences else 0
    high = sequences[-1] if sequences else 0
    return {
        "path": str(path),
        "schemaVersion": "1.0",
        "byteLength": len(data),
        "sha256": _sha(data),
        "sequenceRange": [low, high],
        "eventIds": event_ids,
    }


def _build_seal(
    attempt_ref: str, terminal: TerminalObservation
) -> AttemptEvidenceSeal:
    stream = _stream_payload(terminal.host_event_stream_path)
    captured = _utc_now()
    digest = _sha(
        json.dumps(
            {
                "attemptRef": attempt_ref,
                "source": terminal.source_diff_digest,
                "git": terminal.git_projection_digest,
                "stream": stream["sha256"],
            },
            sort_keys=True,
            separators=(",", ":"),
        ).encode("utf-8")
    )
    return AttemptEvidenceSeal(
        seal_ref=f"seal-{digest[7:23]}",
        attempt_ref=attempt_ref,
        captured_at=captured,
        source_diff_digest=terminal.source_diff_digest,
        git_projection_digest=terminal.git_projection_digest,
        changed_paths=terminal.changed_paths,
        host_event_stream=stream,
        artifacts=_artifact_rows(terminal.artifact_paths),
    )


def _same_observation(seal: AttemptEvidenceSeal, terminal: TerminalObservation) -> bool:
    stream = _stream_payload(terminal.host_event_stream_path)
    return (
        seal.source_diff_digest == terminal.source_diff_digest
        and seal.git_projection_digest == terminal.git_projection_digest
        and seal.host_event_stream.get("sha256") == stream["sha256"]
    )


def finalize_attempt(
    manifest_path: Path,
    attempt_ref: str,
    terminal: TerminalObservation,
    *,
    task_key: str,
) -> AttemptEvidenceSeal:
    """Seal host events, artifacts, source diff and git projection atomically."""
    invocation_ref, attempt_number = _split_attempt_ref(attempt_ref)
    with task_mutex(task_key):
        manifest = read_execution_manifest(manifest_path)
        index = next(
            (
                offset
                for offset, row in enumerate(manifest.attempts)
                if row.invocation_ref == invocation_ref and row.attempt == attempt_number
            ),
            None,
        )
        if index is None:
            raise AttemptEvidenceError("attempt to seal is unknown")
        existing = manifest.attempts[index]
        if terminal.interrupted_before_record:
            closed = Attempt(
                invocation_ref=existing.invocation_ref,
                attempt=existing.attempt,
                started_at=existing.started_at,
                finished_at=_utc_now(),
                status="evidence-unsealed",
                result_path=existing.result_path,
                error_path=existing.error_path,
                change_summary=dict(existing.change_summary),
                evidence_seal_ref=None,
            )
            attempts = list(manifest.attempts)
            attempts[index] = closed
            _write_manifest_over_existing(
                manifest_path,
                _replace_manifest(manifest, attempts=tuple(attempts)),
            )
            raise AttemptEvidenceError("attempt closed as evidence-unsealed")
        seal = _build_seal(attempt_ref, terminal)
        existing_seal = next(
            (
                AttemptEvidenceSeal.from_payload(row)
                for row in manifest.attempt_evidence_seals
                if row.get("attemptRef") == attempt_ref
            ),
            None,
        )
        if existing_seal is not None:
            if not _same_observation(existing_seal, terminal):
                raise AttemptEvidenceError("attempt already sealed with different observation")
            return existing_seal
        closed = Attempt(
            invocation_ref=existing.invocation_ref,
            attempt=existing.attempt,
            started_at=existing.started_at,
            finished_at=existing.finished_at or _utc_now(),
            status=terminal.status,
            result_path=existing.result_path,
            error_path=existing.error_path,
            change_summary=dict(existing.change_summary),
            evidence_seal_ref=seal.seal_ref,
        )
        attempts = list(manifest.attempts)
        attempts[index] = closed
        updated = _replace_manifest(
            manifest,
            attempts=tuple(attempts),
            attempt_evidence_seals=(
                *manifest.attempt_evidence_seals,
                seal.to_payload(),
            ),
        )
        _write_manifest_over_existing(manifest_path, updated)
        return seal


def reject_unsealed_attempt(
    manifest_path: Path,
    attempt_ref: str,
    original_baseline_digest: str,
    *,
    task_key: str,
):
    """Close an unsealed attempt and record a cleanup-only reject."""
    from .mutation_recovery import MutationRecoveryDecision

    invocation_ref, attempt_number = _split_attempt_ref(attempt_ref)
    with task_mutex(task_key):
        manifest = read_execution_manifest(manifest_path)
        index = next(
            (
                offset
                for offset, row in enumerate(manifest.attempts)
                if row.invocation_ref == invocation_ref and row.attempt == attempt_number
            ),
            None,
        )
        if index is None:
            raise AttemptEvidenceError("attempt to reject is unknown")
        existing = manifest.attempts[index]
        closed = Attempt(
            invocation_ref=existing.invocation_ref,
            attempt=existing.attempt,
            started_at=existing.started_at,
            finished_at=existing.finished_at or _utc_now(),
            status="evidence-unsealed",
            result_path=existing.result_path,
            error_path=existing.error_path,
            change_summary=dict(existing.change_summary),
            evidence_seal_ref=None,
        )
        decision = MutationRecoveryDecision(
            recovery_ref=f"recovery-unsealed-{attempt_number}",
            source_invocation_ref=invocation_ref,
            source_attempt_ref=attempt_ref,
            source_evidence_seal_ref=None,
            original_baseline_digest=original_baseline_digest,
            terminal_source_diff_digest="",
            terminal_git_chain_digest="",
            verifier_role_execution_ref=None,
            verifier_invocation_ref=None,
            recovered_evidence_refs=(),
            recovered_out_of_plan_edits=(),
            unmet_obligations=("evidence-unsealed",),
            decision="reject",
            reason="evidence-unsealed",
        )
        attempts = list(manifest.attempts)
        attempts[index] = closed
        updated = _replace_manifest(
            manifest,
            attempts=tuple(attempts),
            mutation_recovery_decisions=(
                *manifest.mutation_recovery_decisions,
                decision.to_payload(),
            ),
        )
        _write_manifest_over_existing(manifest_path, updated)
        return decision