from __future__ import annotations

import argparse
import asyncio
import csv
from dataclasses import dataclass
from datetime import datetime, timezone
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import time
from typing import Any

import httpx


ROOT = Path(__file__).resolve().parents[1]
DATA_ROOT = ROOT / ".data_uat_dry_run"
DEFAULT_OUTPUT_ROOT = ROOT / "docs" / "uat" / "evidence"

SCENARIO_SERVICE_MAP = {
    "SYS": "platform",
    "S0": "auth-service/gateway",
    "S1": "customer-service",
    "S2": "interaction-service",
    "S3": "interaction-service/routing-service",
    "S4": "routing-service/interaction-service",
    "S5": "voice-adapter-service",
    "S6": "telegram-adapter-service",
    "S7": "kb-service",
    "S8": "supervisor-service",
    "S9": "reporting-service",
}

SEVERITY_RISK_MAP = {
    "P1": "High",
    "P2": "Medium",
    "P3": "Medium",
    "P4": "Low",
}

SERVICE_SPECS = [
    {"name": "auth", "module": "services.auth_service.app:app", "port": 58001},
    {"name": "audit", "module": "services.audit_service.app:app", "port": 58002},
    {"name": "customer", "module": "services.customer_service.app:app", "port": 58003},
    {"name": "interaction", "module": "services.interaction_service.app:app", "port": 58004},
    {"name": "routing", "module": "services.routing_service.app:app", "port": 58005},
    {"name": "voice", "module": "services.voice_adapter_service.app:app", "port": 58006},
    {"name": "telegram", "module": "services.telegram_adapter_service.app:app", "port": 58007},
    {"name": "kb", "module": "services.kb_service.app:app", "port": 58008},
    {"name": "reporting", "module": "services.reporting_service.app:app", "port": 58009},
    {"name": "supervisor", "module": "services.supervisor_service.app:app", "port": 58010},
    {"name": "gateway", "module": "gateway.app:app", "port": 58080},
]


@dataclass
class ScenarioResult:
    scenario_id: str
    title: str
    passed: bool
    severity_on_fail: str
    details: str
    evidence: dict[str, Any]


def utc_now() -> datetime:
    return datetime.now(timezone.utc)


def now_compact() -> str:
    return utc_now().strftime("%Y%m%d_%H%M%S")


def iso_now() -> str:
    return utc_now().isoformat()


async def wait_for_health(base_url: str, retries: int = 80, delay: float = 0.2) -> None:
    async with httpx.AsyncClient(timeout=2) as client:
        for _ in range(retries):
            try:
                response = await client.get(f"{base_url}/health")
                if response.status_code == 200:
                    return
            except Exception:
                pass
            await asyncio.sleep(delay)
    raise RuntimeError(f"Service not healthy: {base_url}/health")


def start_one_service(spec: dict[str, Any], data_dir: Path) -> subprocess.Popen:
    env = os.environ.copy()
    env["CC_DATA_DIR"] = str(data_dir)

    if spec["name"] == "gateway":
        env["AUTH_SERVICE_URL"] = "http://127.0.0.1:58001"
        env["AUDIT_SERVICE_URL"] = "http://127.0.0.1:58002"
        env["CUSTOMER_SERVICE_URL"] = "http://127.0.0.1:58003"
        env["INTERACTION_SERVICE_URL"] = "http://127.0.0.1:58004"
        env["ROUTING_SERVICE_URL"] = "http://127.0.0.1:58005"
        env["VOICE_ADAPTER_SERVICE_URL"] = "http://127.0.0.1:58006"
        env["TELEGRAM_ADAPTER_SERVICE_URL"] = "http://127.0.0.1:58007"
        env["KB_SERVICE_URL"] = "http://127.0.0.1:58008"
        env["REPORTING_SERVICE_URL"] = "http://127.0.0.1:58009"
        env["SUPERVISOR_SERVICE_URL"] = "http://127.0.0.1:58010"

    cmd = [
        sys.executable,
        "-m",
        "uvicorn",
        spec["module"],
        "--host",
        "127.0.0.1",
        "--port",
        str(spec["port"]),
    ]
    return subprocess.Popen(
        cmd,
        cwd=str(ROOT),
        env=env,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )


async def start_services_sequential(data_dir: Path) -> list[subprocess.Popen]:
    data_dir.mkdir(parents=True, exist_ok=True)
    processes: list[subprocess.Popen] = []
    for spec in SERVICE_SPECS:
        proc = start_one_service(spec, data_dir)
        processes.append(proc)
        await wait_for_health(f"http://127.0.0.1:{spec['port']}")
    return processes


def stop_services(processes: list[subprocess.Popen]) -> None:
    for proc in processes:
        if proc.poll() is not None:
            continue
        try:
            proc.terminate()
        except Exception:
            pass

    for proc in processes:
        try:
            proc.wait(timeout=2)
        except Exception:
            try:
                proc.kill()
            except Exception:
                pass


def scenario_result(
    scenario_id: str,
    title: str,
    passed: bool,
    severity_on_fail: str,
    details: str,
    evidence: dict[str, Any] | None = None,
) -> ScenarioResult:
    return ScenarioResult(
        scenario_id=scenario_id,
        title=title,
        passed=passed,
        severity_on_fail=severity_on_fail,
        details=details,
        evidence=evidence or {},
    )


async def run_scenarios(base_url: str) -> tuple[list[ScenarioResult], dict[str, Any]]:
    admin = {"X-User": "admin", "X-Role": "admin"}
    supervisor = {"X-User": "supervisor", "X-Role": "supervisor"}
    operator = {"X-User": "operator", "X-Role": "operator"}
    analyst = {"X-User": "analyst", "X-Role": "analyst"}
    results: list[ScenarioResult] = []
    context: dict[str, Any] = {
        "started_at": iso_now(),
        "baseline_doc": "docs/gates/mvp-pilot-baseline.md",
        "wave2_backlog": "docs/roadmap/05-wave2-backlog.md",
    }

    async with httpx.AsyncClient(base_url=base_url, timeout=12) as client:
        health = await client.get("/health")
        registry = await client.get("/registry")
        if health.status_code != 200 or registry.status_code != 200:
            details = f"health={health.status_code}, registry={registry.status_code}"
            return [scenario_result("SYS", "System precheck", False, "P1", details)], context
        services = registry.json().get("services", {})
        context["reporting_url"] = services.get("reporting", "")

        login = await client.post("/proxy/auth/auth/login", json={"username": "admin", "password": "admin123"})
        login_ok = False
        if login.status_code == 200:
            login_ok = bool(login.json().get("access_token"))
        denied = await client.post(
            "/proxy/auth/users",
            headers=operator,
            json={
                "username": f"uat_denied_{now_compact()}",
                "password": "secret123",
                "full_name": "UAT Denied",
                "role": "operator",
            },
        )
        results.append(
            scenario_result(
                "S0",
                "Login and RBAC deny",
                login_ok and denied.status_code == 403,
                "P1",
                f"login={login.status_code}, denied={denied.status_code}",
                {"user": "admin", "denied_status": denied.status_code},
            )
        )

        queue = await client.post(
            "/proxy/routing/queues",
            headers=admin,
            json={
                "name": f"UAT Dry Queue {now_compact()}",
                "description": "UAT dry-run queue",
                "rules": [
                    {"channel": "voice", "priority": 3, "strategy": "round_robin", "sla_seconds": 30},
                    {"channel": "telegram", "priority": 3, "strategy": "round_robin", "sla_seconds": 45},
                ],
            },
        )
        if queue.status_code != 200:
            results.append(
                scenario_result(
                    "SYS",
                    "System precheck",
                    False,
                    "P1",
                    f"queue create failed status={queue.status_code}",
                )
            )
            context["completed_at"] = iso_now()
            return results, context
        queue_id = queue.json()["queue_id"]
        context["queue_id"] = queue_id

        customer = await client.post(
            "/proxy/customer/customers",
            json={
                "display_name": "UAT Dry Customer",
                "phones": ["+77017770000"],
                "preferred_phone": "+77017770000",
                "tags": ["uat", "dry-run"],
            },
        )
        customer_id = ""
        search_ok = False
        if customer.status_code == 200:
            customer_id = customer.json()["customer_id"]
            search = await client.get("/proxy/customer/customers?query=dry-run")
            if search.status_code == 200:
                search_ok = any(item.get("customer_id") == customer_id for item in search.json())
        context["customer_id"] = customer_id
        results.append(
            scenario_result(
                "S1",
                "Customer create and search",
                customer.status_code == 200 and search_ok,
                "P1",
                f"create={customer.status_code}, search={search_ok}",
                {"customer_id": customer_id, "query": "dry-run"},
            )
        )

        voice_interaction_id = ""
        interaction = await client.post(
            "/proxy/interaction/interactions",
            headers=operator,
            json={
                "channel": "voice",
                "subject": "UAT dry-run S2 voice",
                "customer_id": customer_id or None,
                "queue_id": queue_id,
                "priority": 3,
            },
        )
        if interaction.status_code != 200:
            results.append(
                scenario_result(
                    "S2",
                    "Voice interaction lifecycle",
                    False,
                    "P1",
                    f"create status={interaction.status_code}",
                )
            )
        else:
            voice_interaction_id = interaction.json()["interaction_id"]
            context["voice_interaction_id"] = voice_interaction_id
            assign = await client.patch(
                f"/proxy/interaction/interactions/{voice_interaction_id}/assign",
                headers=supervisor,
                json={"assignee": "operator_a"},
            )
            close = await client.patch(
                f"/proxy/interaction/interactions/{voice_interaction_id}/status",
                headers=operator,
                json={"status": "closed"},
            )
            timeline = await client.get(f"/proxy/interaction/interactions/{voice_interaction_id}/timeline")
            actions: set[str] = set()
            if timeline.status_code == 200:
                actions = {event.get("action", "") for event in timeline.json().get("events", [])}
            expected = {"interaction.created", "interaction.assigned", "interaction.status_changed"}
            results.append(
                scenario_result(
                    "S2",
                    "Voice interaction lifecycle",
                    assign.status_code == 200 and close.status_code == 200 and expected.issubset(actions),
                    "P1",
                    f"assign={assign.status_code}, close={close.status_code}, actions={sorted(actions)}",
                    {"interaction_id": voice_interaction_id},
                )
            )

        esc_create = await client.post(
            "/proxy/interaction/interactions",
            headers=operator,
            json={
                "channel": "voice",
                "subject": "UAT dry-run S3 escalation",
                "customer_id": customer_id or None,
                "queue_id": queue_id,
                "priority": 3,
            },
        )
        if esc_create.status_code != 200:
            results.append(
                scenario_result(
                    "S3",
                    "Assignment and escalation to second line",
                    False,
                    "P1",
                    f"create status={esc_create.status_code}",
                )
            )
        else:
            esc_interaction_id = esc_create.json()["interaction_id"]
            assign = await client.patch(
                f"/proxy/interaction/interactions/{esc_interaction_id}/assign",
                headers=supervisor,
                json={"assignee": "operator_b"},
            )
            escalated = await client.post(
                f"/proxy/interaction/interactions/{esc_interaction_id}/escalate",
                headers=operator,
                json={"target_queue_id": "line2"},
            )
            timeline = await client.get(f"/proxy/interaction/interactions/{esc_interaction_id}/timeline")
            actions = []
            if timeline.status_code == 200:
                actions = [event.get("action", "") for event in timeline.json().get("events", [])]
            results.append(
                scenario_result(
                    "S3",
                    "Assignment and escalation to second line",
                    assign.status_code == 200
                    and escalated.status_code == 200
                    and escalated.json().get("status") == "escalated"
                    and escalated.json().get("queue_id") == "line2"
                    and "interaction.escalated" in actions,
                    "P1",
                    f"assign={assign.status_code}, escalate={escalated.status_code}, queue={escalated.json().get('queue_id') if escalated.status_code == 200 else ''}",
                    {"interaction_id": esc_interaction_id, "target_queue": "line2"},
                )
            )

        route = await client.post(f"/proxy/routing/queues/{queue_id}/route?channel=voice&priority=3")
        timeline = (
            await client.get(f"/proxy/interaction/interactions/{voice_interaction_id}/timeline")
            if voice_interaction_id
            else None
        )
        timeline_ok = False
        timeline_count = 0
        if timeline and timeline.status_code == 200:
            timeline_count = len(timeline.json().get("events", []))
            timeline_ok = timeline_count >= 3
        route_ok = False
        if route.status_code == 200:
            route_ok = bool(route.json().get("assignee"))
        results.append(
            scenario_result(
                "S4",
                "Routing and timeline verification",
                route_ok and timeline_ok,
                "P2",
                f"route={route.status_code}, timeline_ok={timeline_ok}, timeline_events={timeline_count}",
                {
                    "queue_id": queue_id,
                    "routed_assignee": route.json().get("assignee") if route.status_code == 200 else "",
                    "interaction_id": voice_interaction_id,
                },
            )
        )

        voice_event = await client.post(
            "/proxy/voice/integrations/voice/events",
            headers=operator,
            json={
                "event_type": "call.started",
                "call_id": f"uat_call_{now_compact()}",
                "interaction_id": voice_interaction_id or None,
                "payload": {"source": "uat_dry_run"},
            },
        )
        voice_event_id = voice_event.json().get("event_id") if voice_event.status_code == 200 else ""
        voice_list_ok = False
        if voice_event.status_code == 200:
            voice_events = await client.get(
                "/proxy/voice/integrations/voice/events?limit=10",
                headers=supervisor,
            )
            if voice_events.status_code == 200:
                voice_list_ok = any(item.get("event_id") == voice_event_id for item in voice_events.json())
        results.append(
            scenario_result(
                "S5",
                "Voice event intake",
                voice_event.status_code == 200 and voice_list_ok,
                "P2",
                f"create={voice_event.status_code}, listed={voice_list_ok}",
                {"event_id": voice_event_id},
            )
        )

        telegram = await client.post(
            "/proxy/telegram/integrations/telegram/webhook",
            json={
                "chat_id": f"uat_dry_chat_{now_compact()}",
                "text": "UAT dry-run telegram",
                "payload": {"source": "uat_dry_run"},
            },
        )
        telegram_id = telegram.json().get("message_id") if telegram.status_code == 200 else ""
        telegram_list_ok = False
        if telegram.status_code == 200:
            messages = await client.get("/proxy/telegram/integrations/telegram/messages?limit=10")
            if messages.status_code == 200:
                telegram_list_ok = any(item.get("message_id") == telegram_id for item in messages.json())
        results.append(
            scenario_result(
                "S6",
                "Telegram interaction lifecycle",
                telegram.status_code == 200 and telegram_list_ok,
                "P1",
                f"create={telegram.status_code}, listed={telegram_list_ok}",
                {"message_id": telegram_id},
            )
        )

        category = await client.post(
            "/proxy/kb/knowledge/categories",
            headers=analyst,
            json={"name": f"UAT Dry {now_compact()}", "description": "dry-run"},
        )
        if category.status_code != 200:
            results.append(
                scenario_result(
                    "S7",
                    "KB usage in active handling",
                    False,
                    "P2",
                    f"category create status={category.status_code}",
                )
            )
        else:
            category_id = category.json()["category_id"]
            article = await client.post(
                "/proxy/kb/knowledge/articles",
                headers=analyst,
                json={
                    "category_id": category_id,
                    "title": "UAT dry KB article",
                    "body": "KB article body for dry run",
                    "tags": ["uat", "kb", "dry"],
                },
            )
            search = await client.get("/proxy/kb/knowledge/search?q=dry")
            article_id = article.json().get("article_id") if article.status_code == 200 else ""
            found = False
            if search.status_code == 200:
                found = any(item.get("article_id") == article_id for item in search.json())
            results.append(
                scenario_result(
                    "S7",
                    "KB usage in active handling",
                    article.status_code == 200 and search.status_code == 200 and found,
                    "P2",
                    f"article={article.status_code}, search={search.status_code}, found={found}",
                    {"article_id": article_id, "keyword": "dry"},
                )
            )

        up1 = await client.post(
            "/proxy/supervisor/supervisor/agent-states",
            json={"agent_id": "uat_dry_a1", "state": "READY", "queue_id": queue_id},
        )
        up2 = await client.post(
            "/proxy/supervisor/supervisor/agent-states",
            json={"agent_id": "uat_dry_a2", "state": "BUSY", "queue_id": queue_id},
        )
        metrics = await client.post(
            f"/proxy/supervisor/supervisor/queue-metrics?queue_id={queue_id}&in_queue=2&avg_wait_seconds=19"
        )
        realtime = await client.get("/proxy/supervisor/supervisor/realtime")
        queue_seen = False
        agent_total = 0
        if realtime.status_code == 200:
            body = realtime.json()
            agent_total = int(body.get("agents", {}).get("total", 0))
            queue_seen = any(row.get("queue_id") == queue_id for row in body.get("queues", []))
        results.append(
            scenario_result(
                "S8",
                "Supervisor realtime",
                up1.status_code == 200
                and up2.status_code == 200
                and metrics.status_code == 200
                and realtime.status_code == 200
                and agent_total >= 2
                and queue_seen,
                "P2",
                f"state_updates=({up1.status_code},{up2.status_code}), metrics={metrics.status_code}, realtime={realtime.status_code}",
                {"queue_id": queue_id, "agent_total": agent_total},
            )
        )

        kpi_rows = [
            {
                "queue_id": queue_id,
                "answered": True,
                "wait_seconds": 15,
                "handle_seconds": 90,
                "abandoned": False,
                "resolved_first_contact": True,
            },
            {
                "queue_id": queue_id,
                "answered": True,
                "wait_seconds": 33,
                "handle_seconds": 110,
                "abandoned": False,
                "resolved_first_contact": False,
            },
            {
                "queue_id": queue_id,
                "answered": False,
                "wait_seconds": 10,
                "handle_seconds": 0,
                "abandoned": True,
                "resolved_first_contact": False,
            },
        ]
        ingest_statuses: list[int] = []
        for row in kpi_rows:
            ingested = await client.post("/proxy/reporting/reports/events", json=row)
            ingest_statuses.append(ingested.status_code)
        kpi = await client.get(f"/proxy/reporting/reports/kpi?queue_id={queue_id}&sl_threshold_seconds=30")
        has_kpi = False
        if kpi.status_code == 200:
            payload = kpi.json().get("kpi", {})
            has_kpi = {"SL", "ASA", "AHT", "Abandon", "FCR"}.issubset(set(payload.keys()))
        exported = await client.get("/proxy/reporting/reports/export")
        export_raw = exported.json().get("raw", "") if exported.status_code == 200 else ""
        csv_ok = (
            exported.status_code == 200
            and "queue_id,answered,wait_seconds,handle_seconds,abandoned,resolved_first_contact,created_at" in export_raw
            and queue_id in export_raw
        )
        results.append(
            scenario_result(
                "S9",
                "KPI report and export validation",
                all(code == 200 for code in ingest_statuses) and kpi.status_code == 200 and has_kpi and csv_ok,
                "P2",
                f"ingest={ingest_statuses}, kpi={kpi.status_code}, csv={csv_ok}",
                {"queue_id": queue_id},
            )
        )

    context["completed_at"] = iso_now()
    return results, context


def build_defects(results: list[ScenarioResult]) -> list[dict[str, str]]:
    defects: list[dict[str, str]] = []
    stamp = iso_now()
    counter = 1
    for row in results:
        if row.passed:
            continue
        defect_id = f"UAT-DRY-{counter:03d}"
        counter += 1
        severity = row.severity_on_fail
        defects.append(
            {
                "defect_id": defect_id,
                "severity": severity,
                "status": "Open",
                "scenario": row.scenario_id,
                "service": SCENARIO_SERVICE_MAP.get(row.scenario_id, "platform"),
                "risk_level": SEVERITY_RISK_MAP.get(severity, "Medium"),
                "summary": row.title,
                "repro_steps": "See scenario checklist and automated dry-run logs",
                "actual_result": row.details,
                "expected_result": "Scenario should pass",
                "verification_step": f"Re-run scenario {row.scenario_id} after fix",
                "owner": "TBD",
                "opened_at": stamp,
                "closed_at": "",
                "backlog_bucket": "MVP" if severity in {"P1", "P2"} else "Wave2",
                "notes": "Generated by scripts/uat_dry_run.py",
            }
        )
    return defects


def write_scenario_json(output_dir: Path, results: list[ScenarioResult], context: dict[str, Any]) -> Path:
    payload = {
        "generated_at": iso_now(),
        "context": context,
        "scenarios": [
            {
                "scenario_id": row.scenario_id,
                "title": row.title,
                "passed": row.passed,
                "severity_on_fail": row.severity_on_fail,
                "details": row.details,
                "evidence": row.evidence,
            }
            for row in results
        ],
    }
    path = output_dir / "scenario-results.json"
    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
    return path


def write_defect_csv(output_dir: Path, defects: list[dict[str, str]]) -> Path:
    path = output_dir / "defect-log.csv"
    fieldnames = [
        "defect_id",
        "severity",
        "status",
        "scenario",
        "service",
        "risk_level",
        "summary",
        "repro_steps",
        "actual_result",
        "expected_result",
        "verification_step",
        "owner",
        "opened_at",
        "closed_at",
        "backlog_bucket",
        "notes",
    ]
    with path.open("w", newline="", encoding="utf-8") as fp:
        writer = csv.DictWriter(fp, fieldnames=fieldnames)
        writer.writeheader()
        for row in defects:
            writer.writerow(row)
    return path


def write_session_protocol(
    output_dir: Path,
    session_id: str,
    base_url: str,
    results: list[ScenarioResult],
    defects: list[dict[str, str]],
    build_version: str,
    deployment_date: str,
    cluster_namespace: str,
) -> Path:
    passed = sum(1 for row in results if row.passed)
    failed = sum(1 for row in results if not row.passed)
    p1 = sum(1 for row in defects if row["severity"] == "P1")
    p2 = sum(1 for row in defects if row["severity"] == "P2")
    p3 = sum(1 for row in defects if row["severity"] == "P3")
    p4 = sum(1 for row in defects if row["severity"] == "P4")

    lines: list[str] = []
    lines.append("# UAT Session Protocol (Automated Dry Run)")
    lines.append("")
    lines.append("## Session Metadata")
    lines.append("")
    lines.append(f"- Session ID: {session_id}")
    lines.append(f"- Date: {utc_now().date().isoformat()}")
    lines.append(f"- Start time (UTC): {iso_now()}")
    lines.append(f"- End time (UTC): {iso_now()}")
    lines.append(f"- Environment URL: {base_url}")
    lines.append(f"- Build/version: {build_version}")
    lines.append(f"- Deployment date: {deployment_date}")
    lines.append(f"- Cluster/namespace: {cluster_namespace}")
    lines.append("")
    lines.append("## Scope Confirmation")
    lines.append("")
    lines.append("- [x] MVP pilot scope reviewed against `docs/gates/mvp-pilot-baseline.md`")
    lines.append("- [x] Out-of-scope requests remain deferred to `docs/roadmap/05-wave2-backlog.md`")
    lines.append("- [x] API stabilization rules remain in effect (backward-compatible fixes only)")
    lines.append("")
    lines.append("## Participants")
    lines.append("")
    lines.append("- Operators: automated dry-run actor")
    lines.append("- Supervisor: automated dry-run actor")
    lines.append("- Analyst: automated dry-run actor")
    lines.append("- Admin: automated dry-run actor")
    lines.append("- Business owner: pending manual UAT")
    lines.append("- IT owner: pending manual UAT")
    lines.append("")
    lines.append("## Preconditions")
    lines.append("")
    lines.append("- [x] Environment is reachable")
    lines.append("- [x] Test accounts are active")
    lines.append("- [x] Test queue is configured")
    lines.append("- [x] Audit and reporting endpoints are reachable")
    lines.append("")
    lines.append("## Execution Summary")
    lines.append("")
    lines.append(f"- Mandatory scenarios executed: {len(results)}")
    lines.append(f"- Passed: {passed}")
    lines.append(f"- Failed: {failed}")
    lines.append("- Blocked: 0")
    lines.append("")
    lines.append("## Scenario Results")
    lines.append("")
    for row in results:
        mark = "PASS" if row.passed else "FAIL"
        lines.append(f"- [{mark}] {row.scenario_id} {row.title}: {row.details}")
    lines.append("")
    lines.append("## Defect Summary")
    lines.append("")
    lines.append(f"- P1: {p1}")
    lines.append(f"- P2: {p2}")
    lines.append(f"- P3: {p3}")
    lines.append(f"- P4: {p4}")
    lines.append("- Defect log attachment: defect-log.csv")
    lines.append("")
    lines.append("## Decision")
    lines.append("")
    if failed == 0:
        lines.append("- [x] Accepted with conditions")
        lines.append("- [ ] Accepted for pilot completion")
        lines.append("- [ ] Not accepted")
    else:
        lines.append("- [ ] Accepted with conditions")
        lines.append("- [ ] Accepted for pilot completion")
        lines.append("- [x] Not accepted")
    lines.append("")
    lines.append("## Comments")
    lines.append("")
    lines.append("- Automated dry-run completed.")
    lines.append("- Manual UAT with real operators/supervisor is still required for final sign-off.")
    lines.append("- Any future scope additions must be recorded in `docs/roadmap/05-wave2-backlog.md`.")
    lines.append("")
    lines.append("## Signatures")
    lines.append("")
    lines.append("- Business owner: pending")
    lines.append("- IT owner: pending")
    lines.append("- Supervisor representative: pending")
    lines.append(f"- Date: {utc_now().date().isoformat()}")
    lines.append("")
    path = output_dir / "session-protocol.md"
    path.write_text("\n".join(lines), encoding="utf-8")
    return path


def write_signoff_draft(output_dir: Path, session_id: str, failed: int) -> Path:
    lines: list[str] = []
    lines.append("# UAT Sign-off Sheet (Draft)")
    lines.append("")
    lines.append(f"- Session ID: {session_id}")
    lines.append(f"- Date: {utc_now().date().isoformat()}")
    lines.append(f"- Scenario failures: {failed}")
    lines.append("")
    lines.append("## Scope Confirmation")
    lines.append("")
    lines.append("- MVP pilot scope confirmed against `docs/gates/mvp-pilot-baseline.md`")
    lines.append("- Out-of-scope requests remain deferred to `docs/roadmap/05-wave2-backlog.md`")
    lines.append("")
    lines.append("## Acceptance Statement")
    lines.append("")
    lines.append("This draft is generated by automated dry-run and cannot replace real UAT signatures.")
    lines.append("")
    lines.append("## Required Manual Signatures")
    lines.append("")
    lines.append("- Business owner: pending")
    lines.append("- IT owner: pending")
    lines.append("- Supervisor representative: pending")
    lines.append("")
    path = output_dir / "signoff-draft.md"
    path.write_text("\n".join(lines), encoding="utf-8")
    return path


def write_summary(
    output_dir: Path,
    session_id: str,
    base_url: str,
    scenario_file: Path,
    defect_file: Path,
    protocol_file: Path,
    signoff_file: Path,
    results: list[ScenarioResult],
    defects: list[dict[str, str]],
) -> Path:
    passed = sum(1 for row in results if row.passed)
    failed = sum(1 for row in results if not row.passed)
    p1 = sum(1 for row in defects if row["severity"] == "P1")
    p2 = sum(1 for row in defects if row["severity"] == "P2")

    lines: list[str] = []
    lines.append("# UAT Dry-Run Summary")
    lines.append("")
    lines.append(f"- Session ID: {session_id}")
    lines.append(f"- Generated at (UTC): {iso_now()}")
    lines.append(f"- Environment URL: {base_url}")
    lines.append(f"- Scenarios: {len(results)} total, {passed} passed, {failed} failed")
    lines.append(f"- Defects: P1={p1}, P2={p2}")
    lines.append("")
    lines.append("## Files")
    lines.append("")
    lines.append(f"- Scenario results: `{scenario_file.name}`")
    lines.append(f"- Session protocol: `{protocol_file.name}`")
    lines.append(f"- Defect log: `{defect_file.name}`")
    lines.append(f"- Sign-off draft: `{signoff_file.name}`")
    lines.append("")
    lines.append("## Next Step")
    lines.append("")
    lines.append("Run manual UAT with real participants, fix only P1/P2 items, and collect signatures.")
    lines.append("Route feature requests and deferred scope to `docs/roadmap/05-wave2-backlog.md`.")
    lines.append("")
    path = output_dir / "summary.md"
    path.write_text("\n".join(lines), encoding="utf-8")
    return path


def update_p1p2_register(defects: list[dict[str, str]], session_id: str) -> None:
    p1 = sum(1 for row in defects if row["severity"] == "P1" and row["status"] == "Open")
    p2 = sum(1 for row in defects if row["severity"] == "P2" and row["status"] == "Open")
    path = ROOT / "docs" / "gates" / "p1-p2-defects.md"
    lines: list[str] = []
    lines.append("# P1/P2 Defect Register (Pilot)")
    lines.append("")
    lines.append(f"Last update: {utc_now().date().isoformat()} ({session_id})")
    lines.append("")
    lines.append("## Current Status")
    lines.append(f"- Open P1: {p1}")
    lines.append(f"- Open P2: {p2}")
    lines.append("- Source: `scripts/uat_dry_run.py` and automated checks")
    lines.append(f"- Latest dry-run evidence: `docs/uat/evidence/dry_run_{session_id}/`")
    lines.append("")
    lines.append("## Triage Policy")
    lines.append("- Only `P1` and `P2` issues belong to MVP remediation.")
    lines.append("- `P3` and `P4` issues move to `docs/roadmap/05-wave2-backlog.md` unless they block sign-off.")
    lines.append("")
    lines.append("## Pilot Note")
    lines.append("- Final P1/P2 closure is confirmed only after the real-operator UAT cycle and sign-off.")
    path.write_text("\n".join(lines), encoding="utf-8")


async def main() -> None:
    parser = argparse.ArgumentParser(description="Run automated UAT dry-run and generate session artifacts")
    parser.add_argument("--base-url", default="http://localhost:8080", help="Gateway base URL")
    parser.add_argument("--auto-start", action="store_true", help="Auto-start local services for dry-run")
    parser.add_argument("--keep-data", action="store_true", help="Keep temporary auto-start data directory")
    parser.add_argument(
        "--output-dir",
        default="",
        help="Output root directory (default: docs/uat/evidence)",
    )
    parser.add_argument("--session-id", default="", help="Optional session ID override")
    parser.add_argument("--build-version", default="v1.0.0-mvp", help="Build/version for protocol")
    parser.add_argument(
        "--deployment-date",
        default=utc_now().date().isoformat(),
        help="Deployment date for protocol",
    )
    parser.add_argument("--cluster-namespace", default="local/standalone", help="Cluster/namespace for protocol")
    parser.add_argument(
        "--update-defect-register",
        action="store_true",
        help="Update docs/gates/p1-p2-defects.md using dry-run result",
    )
    args = parser.parse_args()

    output_root = Path(args.output_dir) if args.output_dir else DEFAULT_OUTPUT_ROOT
    output_root = output_root if output_root.is_absolute() else (ROOT / output_root).resolve()
    session_id = args.session_id.strip() or f"UAT-DRY-{now_compact()}"
    session_dir = output_root / f"dry_run_{session_id}"
    session_dir.mkdir(parents=True, exist_ok=True)

    run_data_dir = DATA_ROOT / f"run_{int(time.time() * 1000)}"
    processes: list[subprocess.Popen] = []
    base_url = args.base_url

    try:
        if args.auto_start:
            processes = await start_services_sequential(run_data_dir)
            base_url = "http://127.0.0.1:58080"

        results, context = await run_scenarios(base_url)
        defects = build_defects(results)

        scenario_file = write_scenario_json(session_dir, results, context)
        defect_file = write_defect_csv(session_dir, defects)
        protocol_file = write_session_protocol(
            session_dir,
            session_id,
            base_url,
            results,
            defects,
            args.build_version,
            args.deployment_date,
            args.cluster_namespace,
        )
        signoff_file = write_signoff_draft(session_dir, session_id, sum(1 for row in results if not row.passed))
        summary_file = write_summary(
            session_dir,
            session_id,
            base_url,
            scenario_file,
            defect_file,
            protocol_file,
            signoff_file,
            results,
            defects,
        )

        if args.update_defect_register:
            update_p1p2_register(defects, session_id)

        passed = sum(1 for row in results if row.passed)
        failed = sum(1 for row in results if not row.passed)
        print(f"UAT dry-run session: {session_id}")
        print(f"Output directory: {session_dir}")
        print(f"Scenarios: total={len(results)} passed={passed} failed={failed}")
        print(f"Summary: {summary_file}")
        sys.exit(0 if failed == 0 else 1)
    finally:
        stop_services(processes)
        if args.auto_start and not args.keep_data:
            try:
                if run_data_dir.exists():
                    shutil.rmtree(run_data_dir)
            except Exception:
                pass


if __name__ == "__main__":
    asyncio.run(main())