"""本地 HTTP daemon（SQLite + 任务队列 + 运行时管理）。

API:
  GET  /health          → {"status":"ok","rig":"name"}
  GET  /rig             → rig 定义（pods/members/edges）
  GET  /status          → 所有 member 状态
  GET  /tasks           → 任务列表（?status=&assignee=）
  POST /tasks           → 入队 {"assignee":"dev.lead","prompt":"..."}
  GET  /tasks/{id}      → 单个任务
  POST /tasks/{id}/claim  → 领取任务
  POST /tasks/{id}/done   → 完成 {"result":"..."}
  POST /tasks/{id}/fail   → 失败 {"error":"..."}
  POST /start           → 启动所有 member
  POST /stop            → 停止所有 member
  POST /start/{member}  → 启动单个 member
  POST /stop/{member}   → 停止单个 member
"""
from __future__ import annotations

import json
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any
from urllib.parse import urlparse, parse_qs

from .queue import TaskQueue
from .rig_schema import Rig, parse_rig
from .runtime_hermes import HermesCliRuntime
from .runtime_terminal import TerminalRuntime


class RigDaemon:
    """单机 rig daemon：管理 member 生命周期 + 任务队列。"""

    def __init__(self, rig_path: str | Path, db_path: str | Path | None = None,
                 host: str = "127.0.0.1", port: int = 27900):
        self.rig = parse_rig(rig_path)
        self.db_path = db_path or Path(rig_path).parent / "rig.sqlite3"
        self.queue = TaskQueue(self.db_path)
        self.host = host
        self.port = port
        self.terminal_rt = TerminalRuntime(self.rig.name)
        self.hermes_rt = HermesCliRuntime()
        self.hermes_procs: dict[str, Any] = {}  # member_id → HermesProcess
        self._http: ThreadingHTTPServer | None = None
        self._thread: threading.Thread | None = None
        self._sweep_stop = threading.Event()
        self._sweep_thread: threading.Thread | None = None

    # ── 生命周期 ──

    def start(self, background: bool = True):
        """启动 daemon HTTP 服务 + 超时扫描线程。"""
        handler = _make_handler(self)
        self._http = _ReusableHTTPServer((self.host, self.port), handler)
        self._thread = threading.Thread(target=self._http.serve_forever, daemon=True)
        self._thread.start()

        # 超时扫描（每 30s）
        self._sweep_stop.clear()
        self._sweep_thread = threading.Thread(target=self._sweep_loop, daemon=True)
        self._sweep_thread.start()

        self.queue.set_meta("daemon_started", str(time.time()))
        return self

    def stop(self):
        if self._http:
            self._http.shutdown()
            self._http.server_close()  # 关闭底层 socket，释放端口
        self._sweep_stop.set()
        self.stop_all_members()
        self.queue.set_meta("daemon_stopped", str(time.time()))
        self.queue.close()

    def _sweep_loop(self):
        while not self._sweep_stop.is_set():
            self.queue.sweep_timeouts()
            self._sweep_stop.wait(30)

    # ── member 管理 ──

    def start_member(self, member_id: str) -> str:
        """启动单个 member。member_id = 'pod.member' 格式。"""
        m = self.rig.all_members().get(member_id)
        if not m:
            return f"unknown member: {member_id}"

        if m.runtime == "terminal":
            try:
                self.terminal_rt.start(m.id, m.command or "", m.cwd)
                return f"started {member_id} (terminal)"
            except Exception as e:
                return f"terminal start failed for {member_id}: {e}"
        elif m.runtime == "hermes-cli":
            if not self.hermes_rt.available():
                return f"hermes CLI 不可用"
            # hermes-cli 用 on-demand spawn（hermes chat 需 TTY，不预起常驻）
            # 任务到达时由 dispatch 触发 spawn
            return f"ready {member_id} (hermes-cli, target={m.target}, profile={m.profile}, on-demand)"
        elif m.runtime == "hermes-acp":
            return f"hermes-acp 需要 tuttid（第一版未实现）"
        return f"unknown runtime: {m.runtime}"

    def stop_member(self, member_id: str) -> str:
        m = self.rig.all_members().get(member_id)
        if not m:
            return f"unknown member: {member_id}"
        if m.runtime == "terminal":
            self.terminal_rt.stop(m.id)
        elif m.runtime in ("hermes-cli", "hermes-acp"):
            hp = self.hermes_procs.pop(member_id, None)
            if hp:
                self.hermes_rt.stop(hp)
        return f"stopped {member_id}"

    def start_all(self) -> list[str]:
        results = []
        for mid in self.rig.all_members():
            results.append(self.start_member(mid))
        return results

    def stop_all_members(self):
        for mid in list(self.hermes_procs):
            self.stop_member(mid)
        self.terminal_rt.stop_all()

    def member_status(self, member_id: str) -> str:
        m = self.rig.all_members().get(member_id)
        if not m:
            return "unknown"
        if m.runtime == "terminal":
            return self.terminal_rt.status(m.id)
        hp = self.hermes_procs.get(member_id)
        if hp:
            return self.hermes_rt.status(hp)
        return "stopped"

    def all_status(self) -> dict[str, str]:
        return {mid: self.member_status(mid) for mid in self.rig.all_members()}

    # ── 任务 ──

    def resolve_member(self, assignee: str) -> str | None:
        """把 `hermes` / `dev.hermes` 都解析成 rig 的规范键 `pod.member`。

        【dsh@mac13 补，实测缺口】原来 `enqueue()` 直接 `all_members().get(assignee)`，
        而该 dict 的 key 是 **`pod.member`** ⇒ 传裸 member id（demo/rig.yaml 与桥的 target 都是这种写法）
        时查不到成员，on-demand 派发**静默不触发**：任务永远停在 pending，既不报错也不执行。
        """
        members = self.rig.all_members()
        if assignee in members:
            return assignee
        hits = [k for k in members if k.split(".", 1)[-1] == assignee]
        if len(hits) == 1:
            return hits[0]
        if len(hits) > 1:
            raise KeyError(f"assignee {assignee!r} 有歧义，请用完整键: {hits}")
        return None

    def enqueue(self, assignee: str, prompt: str, **kw) -> Any:
        key = self.resolve_member(assignee)
        if key is None:
            raise KeyError(f"rig 里没有成员 {assignee!r}（可用: {sorted(self.rig.all_members())}）")
        task = self.queue.enqueue(key, prompt, **kw)
        # 如果 assignee 是 hermes-cli，立即 on-demand spawn
        m = self.rig.all_members().get(key)
        if m and m.runtime == "hermes-cli":
            self._dispatch_hermes(key, task)
        return task

    def _dispatch_hermes(self, member_id: str, task: Any):
        """On-demand spawn hermes 子进程处理单个任务。"""
        import threading
        m = self.rig.all_members()[member_id]
        thread = threading.Thread(
            target=self._run_hermes_task,
            args=(member_id, m, task),
            daemon=True,
        )
        thread.start()

    def _run_hermes_task(self, member_id: str, m, task: Any):
        """在独立线程里跑 hermes 子进程（不阻塞 HTTP 线程）。"""
        import subprocess as _sp
        try:
            claimed = self.queue.claim(task_id=task.id)
            if not claimed:
                return
            # 构造命令：hermes [-z prompt] [--profile p]
            # -z 是全局 flag（单条 prompt 非交互模式），不能跟在 chat 后面
            cmd = [self.hermes_rt.hermes_bin]
            if m.profile and m.profile != "default":
                cmd += ["--profile", m.profile]
            # 【dsh@mac13 实测缺口】原来这里拼的是 `[bin, "--agent", target, "-z", prompt]`，
            # 但 hermes CLI **没有全局 --agent**（只有 profile 概念）：实测直接报
            #   `hermes: error: argument command: invalid choice: 'hermes'` ⇒ 任务必然 fail。
            # 现在按 CLI 真实形态构造：oneshot 用 `-z`，成员区分靠 `--profile`。
            if getattr(m, "model", ""):
                cmd += ["-m", m.model]          # 实测必需：不传 -m 会被 API 拒(HTTP 400)
            cmd += ["-z", task.prompt]  # 单条 prompt 模式

            r = _sp.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=600,
                cwd=m.cwd or ".",
            )
            output = (r.stdout or "").strip()
            if not output and r.stderr:
                output = f"[stderr] {r.stderr.strip()[:500]}"
            self.queue.complete(task.id, result=output[:2000] or "(no output)")
        except Exception as e:
            self.queue.fail(task.id, error=str(e))

    def claim(self, assignee: str | None = None, task_id: str | None = None):
        return self.queue.claim(assignee, task_id)


class _ReusableHTTPServer(ThreadingHTTPServer):
    allow_reuse_address = True


class _Handler(BaseHTTPRequestHandler):
    daemon: RigDaemon = None  # type: ignore

    def log_message(self, format, *args):  # noqa: A002
        pass  # 静默

    def _json(self, code: int, data: Any):
        body = json.dumps(data, ensure_ascii=False).encode()
        self.send_response(code)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _body(self) -> dict:
        length = int(self.headers.get("Content-Length", 0))
        if not length:
            return {}
        return json.loads(self.rfile.read(length))

    def do_GET(self):
        d = self.daemon
        path = urlparse(self.path).path.rstrip("/")
        qs = parse_qs(urlparse(self.path).query)

        if path == "/health":
            self._json(200, {"status": "ok", "rig": d.rig.name})
        elif path == "/rig":
            from dataclasses import asdict
            self._json(200, asdict(d.rig))
        elif path == "/status":
            self._json(200, d.all_status())
        elif path == "/tasks":
            tasks = d.queue.list(
                status=qs.get("status", [None])[0],
                assignee=qs.get("assignee", [None])[0],
                limit=int(qs.get("limit", ["50"])[0]))
            self._json(200, [
                {"id": t.id, "assignee": t.assignee, "prompt": t.prompt[:100],
                 "status": t.status, "result": t.result, "error": t.error,
                 "created_at": t.created_at}
                for t in tasks
            ])
        elif path.startswith("/tasks/"):
            tid = path.split("/")[-1]
            t = d.queue.get(tid)
            if t:
                # 【dsh@mac13 补】把 idem_key/context 也回出来：接通件与人都要能**看见**幂等键落库了没有
                self._json(200, {"id": t.id, "assignee": t.assignee,
                                 "prompt": t.prompt, "status": t.status,
                                 "result": t.result, "error": t.error,
                                 "idem_key": getattr(t, "idem_key", None),
                                 "context": getattr(t, "context", None)})
            else:
                self._json(404, {"error": "not found"})
        elif path == "/ui":
            ui_path = Path(__file__).parent / "ui.html"
            if ui_path.exists():
                body = ui_path.read_bytes()
                self.send_response(200)
                self.send_header("Content-Type", "text/html; charset=utf-8")
                self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                self.wfile.write(body)
            else:
                self._json(404, {"error": "ui.html not found"})
        else:
            self._json(404, {"error": "not found"})

    def do_POST(self):
        d = self.daemon
        path = urlparse(self.path).path.rstrip("/")
        body = self._body()

        if path == "/tasks":
            assignee = body.get("assignee", "")
            prompt = body.get("prompt", "")
            if not assignee or not prompt:
                self._json(400, {"error": "assignee 和 prompt 必填"})
                return
            # 【dsh@mac13 补，承 #43 审查 G1/G2】原来只往下传 context/deadline_s ⇒
            #   ① 桥传来的 `task_id` 被丢弃（rig 自己生成 id），回执关联就断了；
            #   ② `idem_key` 根本没进 HTTP 层，幂等只在 queue 层可用、经 HTTP 用不上。
            # 现在把两者透传下去（queue 已支持；缺省行为不变：不给就是 None → 自动生成 id、不去重）。
            try:
                resolved = d.resolve_member(assignee)
            except KeyError as e:
                self._json(400, {"error": str(e)})
                return
            if resolved is None:
                self._json(400, {"error": f"rig 里没有成员 {assignee!r}",
                                 "available": sorted(d.rig.all_members())})
                return
            assignee = resolved
            task = d.enqueue(assignee, prompt,
                             context=body.get("context"),
                             deadline_s=body.get("deadline_s", 600),
                             task_id=body.get("task_id") or None,
                             idem_key=body.get("idem_key") or None)
            self._json(201, {"id": task.id, "status": task.status})
        elif path.startswith("/tasks/") and path.endswith("/claim"):
            tid = path.split("/")[-2]
            task = d.claim(task_id=tid)
            if task:
                self._json(200, {"id": task.id, "status": "running"})
            else:
                self._json(404, {"error": "no pending task"})
        elif path.startswith("/tasks/") and path.endswith("/done"):
            tid = path.split("/")[-2]
            d.queue.complete(tid, result=body.get("result"))
            self._json(200, {"id": tid, "status": "done"})
        elif path.startswith("/tasks/") and path.endswith("/fail"):
            tid = path.split("/")[-2]
            d.queue.fail(tid, error=body.get("error", "unknown"))
            self._json(200, {"id": tid, "status": "failed"})
        elif path == "/start":
            results = d.start_all()
            self._json(200, results)
        elif path.startswith("/start/"):
            mid = path.split("/")[-1]
            self._json(200, {"result": d.start_member(mid)})
        elif path == "/stop":
            d.stop_all_members()
            self._json(200, {"result": "all stopped"})
        elif path.startswith("/stop/"):
            mid = path.split("/")[-1]
            self._json(200, {"result": d.stop_member(mid)})
        else:
            self._json(404, {"error": "not found"})


def _make_handler(daemon: RigDaemon):
    handler = type("BoundHandler", (_Handler,), {"daemon": daemon})
    return handler


def serve(rig_path: str | Path, host: str = "127.0.0.1", port: int = 27900,
          db_path: str | Path | None = None) -> RigDaemon:
    """创建并启动 daemon，返回 RigDaemon 实例。"""
    d = RigDaemon(rig_path, db_path=db_path, host=host, port=port)
    d.start(background=True)
    return d