#!/usr/bin/env python3
"""Safe live execution-plan snapshots for CARBON's protected local viewer."""

from __future__ import annotations

import argparse
import datetime as dt
import http.server
import json
import os
import re
import secrets
import signal
import sys
import tempfile
import time
from pathlib import Path
from typing import Any
from urllib.parse import urlparse

from carbon_quality_confidence import assess_quality_confidence

TERMINAL = {"passed", "failed", "blocked", "skipped", "canceled"}
TEST_STATUSES = {
    "planned", "queued", "running", "passed", "failed", "blocked", "skipped", "canceled"
}
SECRET_PATTERNS = [
    re.compile(r"\bsk-[A-Za-z0-9_-]{16,}\b"),
    re.compile(r"\bghp_[A-Za-z0-9]{20,}\b"),
    re.compile(r"\bAKIA[0-9A-Z]{16}\b"),
    re.compile(r"\bBearer\s+[A-Za-z0-9._~+/-]{16,}\b", re.IGNORECASE),
    re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"),
    re.compile(r"\beyJ[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\b"),
]


def read_json(path: Path) -> dict[str, Any]:
    if path.is_symlink() or not path.is_file():
        raise ValueError(f"CARBON execution artifact must be a regular file: {path}")
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        raise ValueError(f"Invalid CARBON execution artifact: {path}") from error
    if not isinstance(value, dict):
        raise ValueError(f"CARBON execution artifact must contain an object: {path}")
    return value


def within(path: Path, parent: Path) -> bool:
    try:
        path.relative_to(parent)
        return True
    except ValueError:
        return False


def safe_artifact_path(raw: str | None, *, root: Path, default: Path, kind: str) -> Path:
    lexical = Path(raw).expanduser() if raw else default
    if not lexical.is_absolute():
        lexical = root / lexical
    if lexical.is_symlink():
        raise ValueError(f"{kind} cannot be a symbolic link")
    candidate = lexical.resolve()
    carbon_root = (root / ".carbon").resolve()
    if not within(candidate, carbon_root):
        raise ValueError(f"{kind} must remain inside {carbon_root}")
    return candidate


def parse_time(value: Any) -> dt.datetime | None:
    if not value:
        return None
    try:
        parsed = dt.datetime.fromisoformat(str(value).replace("Z", "+00:00"))
        return parsed if parsed.tzinfo else parsed.replace(tzinfo=dt.timezone.utc)
    except ValueError:
        return None


def seconds_between(start: Any, finish: Any) -> int | None:
    left, right = parse_time(start), parse_time(finish)
    if not left or not right:
        return None
    return max(0, round((right - left).total_seconds()))


def discover_run(root: Path, plan_path: Path) -> Path | None:
    runs_root = root / ".carbon" / "runs"
    if not runs_root.is_dir() or runs_root.is_symlink():
        return None
    matches: list[tuple[str, Path]] = []
    for state_path in runs_root.glob("*/run.json"):
        try:
            state = read_json(state_path)
            source = Path(str(state.get("sourcePlan") or "")).expanduser().resolve()
            if source == plan_path:
                matches.append((str(state.get("updatedAt") or state.get("createdAt") or ""), state_path.parent))
        except (ValueError, OSError):
            continue
    return max(matches, default=("", None))[1]


def clipped(value: Any, limit: int = 500) -> str:
    compact = " ".join(str(value or "").split())
    for pattern in SECRET_PATTERNS:
        compact = pattern.sub("[REDACTED]", compact)
    return compact if len(compact) <= limit else compact[: limit - 1] + "…"


def strings(value: Any, limit: int = 30) -> list[str]:
    if not isinstance(value, list):
        return []
    return [clipped(item, 500) for item in value[:limit] if clipped(item, 500)]


def test_row(raw: Any, index: int) -> dict[str, Any]:
    item = raw if isinstance(raw, dict) else {}
    status = clipped(item.get("status") or "planned", 30).lower()
    if status not in TEST_STATUSES:
        status = "planned"
    predicted = clipped(item.get("predictedOutcome") or "unknown", 20).lower()
    if predicted not in {"pass", "fail", "unknown"}:
        predicted = "unknown"
    estimated = max(1, min(86_400, int(item.get("estimatedSeconds") or 60)))
    actual_seconds = seconds_between(item.get("startedAt"), item.get("finishedAt"))
    return {
        "id": clipped(item.get("id") or f"TEST-{index + 1:03d}", 120),
        "title": clipped(item.get("title") or item.get("name") or item.get("check") or f"Test {index + 1}", 300),
        "priority": clipped(item.get("priority") or "P2", 12).upper(),
        "adapter": clipped(item.get("adapter") or "auto", 80),
        "domain": clipped(item.get("domain") or item.get("type") or "context-derived", 80),
        "status": status,
        "estimatedSeconds": estimated,
        "actualSeconds": actual_seconds,
        "actual": clipped(item.get("actual"), 1000) if status in TERMINAL else "",
        "oracle": clipped(item.get("oracle"), 1000),
        "oracleBasis": clipped(item.get("oracleBasis"), 500),
        "predictedOutcome": predicted,
        "predictionBasis": clipped(item.get("predictionBasis"), 1000),
    }


def snapshot(root_value: str, plan_value: str | None, run_value: str | None) -> dict[str, Any]:
    lexical_root = Path(root_value).expanduser()
    if lexical_root.is_symlink():
        raise ValueError("root must be an existing non-symlink project directory")
    root = lexical_root.resolve()
    if not root.is_dir():
        raise ValueError("root must be an existing non-symlink project directory")
    plan_path = safe_artifact_path(
        plan_value,
        root=root,
        default=root / ".carbon" / "framework" / "orchestration-plan.json",
        kind="planPath",
    )
    plan = read_json(plan_path)
    run_dir = safe_artifact_path(
        run_value,
        root=root,
        default=root / ".carbon" / "runs" / "not-started",
        kind="runDir",
    ) if run_value else discover_run(root, plan_path)
    run: dict[str, Any] | None = None
    evidence_manifest: dict[str, Any] = {}
    if run_dir:
        if run_dir.is_symlink() or not run_dir.is_dir():
            raise ValueError("runDir must be an existing non-symlink directory")
        run = read_json(run_dir / "run.json")
        source = Path(str(run.get("sourcePlan") or "")).expanduser().resolve()
        if source != plan_path:
            raise ValueError("runDir does not belong to the selected execution plan")
        evidence_path = run_dir / "evidence" / "manifest.json"
        if evidence_path.is_file() and not evidence_path.is_symlink():
            evidence_manifest = read_json(evidence_path)

    source_tests = (run or plan).get("tests")
    tests = [test_row(item, index) for index, item in enumerate(source_tests if isinstance(source_tests, list) else [])]
    if not tests:
        raise ValueError("Execution plan requires at least one test")
    counts = {status: 0 for status in sorted(TEST_STATUSES)}
    for test in tests:
        counts[test["status"]] += 1
    completed = sum(counts[name] for name in TERMINAL)
    total = len(tests)
    now = dt.datetime.now(dt.timezone.utc)
    status = clipped((run or {}).get("status") or "planned", 30).lower()
    started = parse_time((run or {}).get("startedAt"))
    finished = parse_time((run or {}).get("finishedAt"))
    elapsed = max(0, round(((finished or now) - started).total_seconds())) if started else 0
    planned_seconds = sum(test["estimatedSeconds"] for test in tests)
    observed_tests = [test for test in tests if test["actualSeconds"] is not None and test["status"] in TERMINAL]
    outstanding_estimates = sum(test["estimatedSeconds"] for test in tests if test["status"] not in TERMINAL)
    observed_seconds = sum(test["actualSeconds"] or 0 for test in observed_tests)
    observed_planned_seconds = sum(test["estimatedSeconds"] for test in observed_tests)
    pace_ratio = observed_seconds / observed_planned_seconds if observed_planned_seconds else None
    remaining = round(outstanding_estimates * pace_ratio) if pace_ratio is not None else outstanding_estimates
    if status in {"completed", "canceled"}:
        remaining = 0
    configured_minutes = max(1, int((run or plan).get("estimatedMinutes") or round(planned_seconds / 60) or 1))
    strategy = (run or plan).get("testStrategy")
    strategy = strategy if isinstance(strategy, dict) else {}
    updated_at = clipped((run or {}).get("updatedAt") or plan.get("updatedAt") or plan.get("createdAt") or "", 60)
    confidence = assess_quality_confidence({"targetRoot": str(root), **(run or plan)}, evidence_manifest)
    return {
        "schema": "carbon.execution-plan-view/v1",
        "project": clipped(root.name, 160),
        "title": clipped((run or plan).get("title") or "CARBON test pass", 300),
        "target": clipped((run or plan).get("target"), 1000),
        "mode": clipped((run or plan).get("mode") or plan.get("size") or "quick", 40),
        "status": status,
        "runId": clipped((run or {}).get("runId"), 160),
        "updatedAt": updated_at,
        "live": run is not None,
        "progress": {
            "total": total,
            "completed": completed,
            "percent": round((completed / total) * 100),
            "counts": counts,
            "currentTestId": clipped((run or {}).get("currentTestId"), 120),
        },
        "timing": {
            "estimatedMinutes": configured_minutes,
            "estimatedSeconds": planned_seconds,
            "elapsedSeconds": elapsed,
            "estimatedRemainingSeconds": max(0, remaining),
            "estimatedRemainingMinutes": max(0, round(remaining / 60)),
            "basis": "observed vs planned test durations" if pace_ratio is not None else "per-test plan estimates",
        },
        "strategy": {
            "selected": clipped(strategy.get("selected") or strategy.get("requested") or "auto", 40),
            "rationale": clipped(strategy.get("rationale"), 1000),
        },
        "candidatePool": {
            "generated": max(total, int(plan.get("generatedCandidates") or total)),
            "selected": total,
            "deferred": max(0, int(plan.get("deferredCandidates") or 0)),
            "budgetMinutes": max(1, int((plan.get("executionCutLine") or {}).get("budgetMinutes") or configured_minutes)),
            "reason": clipped((plan.get("executionCutLine") or {}).get("reason"), 1000),
            "moreCommand": "/carbon-more",
        },
        "businessValue": {
            "required": bool((plan.get("businessValue") or {}).get("required")),
            "candidateCount": max(0, int((plan.get("businessValue") or {}).get("candidateCount") or 0)),
            "selectedTestIds": strings((plan.get("businessValue") or {}).get("selectedTestIds")),
        },
        "coverageByDomain": {
            clipped(key, 80): max(0, int(value or 0))
            for key, value in (plan.get("coverageByDomain") or {}).items()
            if clipped(key, 80)
        },
        "scope": strings((run or plan).get("scope")),
        "exclusions": strings((run or plan).get("exclusions")),
        "stopConditions": strings((run or plan).get("stopConditions")),
        "approvalGates": strings(plan.get("approvalGates") or plan.get("approval_gates")),
        "tests": tests,
        "qualityConfidence": confidence,
    }


def serve(
    root_value: str,
    plan_value: str | None,
    run_value: str | None,
    ready_value: str,
    ttl_seconds: int,
) -> int:
    """Serve one execution plan independently of the invoking MCP process."""
    lexical_root = Path(root_value).expanduser()
    if lexical_root.is_symlink():
        raise ValueError("root must be an existing non-symlink project directory")
    root = lexical_root.resolve()
    if not root.is_dir():
        raise ValueError("root must be an existing non-symlink project directory")
    plan_path = safe_artifact_path(
        plan_value,
        root=root,
        default=root / ".carbon" / "framework" / "orchestration-plan.json",
        kind="planPath",
    )
    run_dir = safe_artifact_path(
        run_value,
        root=root,
        default=root / ".carbon" / "runs" / "not-started",
        kind="runDir",
    ) if run_value else None
    # Validate the initial snapshot before advertising a live endpoint.
    snapshot(str(root), str(plan_path), str(run_dir) if run_dir else None)

    ready_path = Path(ready_value).expanduser().resolve()
    runtime_root = (root / ".carbon" / "runtime").resolve()
    if not within(ready_path, runtime_root):
        raise ValueError(f"readyFile must remain inside {runtime_root}")
    runtime_root.mkdir(parents=True, exist_ok=True)
    token = os.environ.get("CARBON_PLAN_VIEW_TOKEN", "")
    if len(token) < 32:
        raise ValueError("CARBON plan viewer requires an ephemeral local access token")

    plugin_root = Path(__file__).resolve().parents[3]
    ui_path = plugin_root / "ui" / "execution-plan.html"
    if ui_path.is_symlink() or not ui_path.is_file():
        raise ValueError("CARBON execution-plan UI is unavailable")
    ui = ui_path.read_bytes()
    started = time.monotonic()
    stop = False
    instance_id = secrets.token_urlsafe(18)

    class Handler(http.server.BaseHTTPRequestHandler):
        server_version = "CARBONPlanViewer/1"

        def log_message(self, _format: str, *_args: Any) -> None:
            return

        def local_request(self) -> bool:
            host = str(self.headers.get("Host") or "").split(":", 1)[0].strip("[]").lower()
            return host in {"127.0.0.1", "localhost", "::1"}

        def json_response(self, status: int, value: dict[str, Any]) -> None:
            body = json.dumps(value).encode("utf-8")
            self.send_response(status)
            self.send_header("Content-Type", "application/json; charset=utf-8")
            self.send_header("Content-Length", str(len(body)))
            self.send_header("Cache-Control", "no-store")
            self.send_header("X-Content-Type-Options", "nosniff")
            self.end_headers()
            self.wfile.write(body)

        def do_GET(self) -> None:  # noqa: N802 - BaseHTTPRequestHandler API
            if not self.local_request():
                self.json_response(403, {"ok": False, "error": "Local CARBON viewer requests only."})
                return
            route = urlparse(self.path).path
            if route == "/health":
                self.json_response(200, {"ok": True, "schema": "carbon.execution-plan-viewer/v1", "instanceId": instance_id, "pid": os.getpid()})
                return
            if route == "/api/execution-plan":
                if self.headers.get("X-Carbon-Settings-Token") != token:
                    self.json_response(403, {"ok": False, "error": "Invalid CARBON execution-plan session."})
                    return
                try:
                    value = snapshot(str(root), str(plan_path), str(run_dir) if run_dir else None)
                    self.json_response(200, {"ok": True, "plan": value})
                except (OSError, ValueError) as error:
                    self.json_response(400, {"ok": False, "error": str(error)})
                return
            if route in {"/", "/run"}:
                self.send_response(200)
                self.send_header("Content-Type", "text/html; charset=utf-8")
                self.send_header("Content-Length", str(len(ui)))
                self.send_header("Cache-Control", "no-store")
                self.send_header("Content-Security-Policy", "default-src 'none'; style-src 'unsafe-inline'; script-src 'unsafe-inline'; connect-src 'self'; img-src 'self' data:; base-uri 'none'; form-action 'self'; frame-ancestors 'none'")
                self.send_header("Referrer-Policy", "no-referrer")
                self.send_header("X-Content-Type-Options", "nosniff")
                self.send_header("X-Frame-Options", "DENY")
                self.end_headers()
                self.wfile.write(ui)
                return
            if route == "/favicon.ico":
                self.send_response(204)
                self.end_headers()
                return
            self.json_response(404, {"ok": False, "error": "Not found."})

    server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    server.timeout = 1
    port = int(server.server_address[1])
    expires = dt.datetime.now(dt.timezone.utc) + dt.timedelta(seconds=ttl_seconds)
    ready = {
        "schema": "carbon.execution-plan-viewer/v1",
        "instanceId": instance_id,
        "pid": os.getpid(),
        "port": port,
        "baseUrl": f"http://127.0.0.1:{port}",
        "planPath": str(plan_path),
        "expiresAt": expires.replace(microsecond=0).isoformat().replace("+00:00", "Z"),
    }
    with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=runtime_root, delete=False) as handle:
        json.dump(ready, handle)
        temporary = Path(handle.name)
    os.chmod(temporary, 0o600)
    os.replace(temporary, ready_path)

    def request_stop(_signum: int, _frame: Any) -> None:
        nonlocal stop
        stop = True

    signal.signal(signal.SIGTERM, request_stop)
    signal.signal(signal.SIGINT, request_stop)
    try:
        while not stop and time.monotonic() - started < ttl_seconds:
            server.handle_request()
    finally:
        server.server_close()
        try:
            ready_path.unlink()
        except OSError:
            pass
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("command", nargs="?", choices=("snapshot", "serve"), default="snapshot")
    parser.add_argument("--root", required=True)
    parser.add_argument("--plan")
    parser.add_argument("--run-dir")
    parser.add_argument("--ready-file")
    parser.add_argument("--ttl-seconds", type=int, default=129_600)
    args = parser.parse_args()
    try:
        if args.command == "serve":
            if not args.ready_file:
                raise ValueError("serve requires --ready-file")
            if not 60 <= args.ttl_seconds <= 604_800:
                raise ValueError("ttlSeconds must be between 60 seconds and seven days")
            return serve(args.root, args.plan, args.run_dir, args.ready_file, args.ttl_seconds)
        print(json.dumps(snapshot(args.root, args.plan, args.run_dir), indent=2))
        return 0
    except (OSError, ValueError) as error:
        print(f"carbon_plan_view: {error}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())
