"""后台进程(bg proc):detach + 文件系统状态,不引队列组件(DESIGN §8.12)。

解决的问题:模型写的长脚本(数据处理/模拟计算)在工具内同步跑会被超时掐死、
占死 run、蓝绿切换时随实例进程陪葬。方案是把进程从 zcbot 进程树上摘下来:

- host 模式:detach 启动 `core/proc_wrapper.py`(stdlib-only),wrapper 负责跑
  子进程、限时、写退出码 —— zcbot 重启不影响它
- docker 模式:每个 bg proc 一个专用容器 `zcbot-proc-<id>`(executor_docker 侧
  实现),dockerd 托管,与 sandbox 容器的 idle reaper / shutdown_all 生命周期解耦

目录布局(dotfile,/v1/files API 天然隐藏,用户文件浏览器不可见):
  <user_root>/.zcbot_procs/<task_id>/<proc_id>/
    proc.json    # 元数据(kind/command/backend/pid 或 container/timeout/created_at)
    output.log   # stdout+stderr 合流(结束时截到最后 10MB)
    exit_code    # 结束后由 wrapper(host)/runner.sh(docker)写入 —— 存在即已结束
    script.py    # run_python inline code 落盘(可选)
    cmd.sh       # docker 模式的用户命令(可选,绕开引号转义)

状态判定不靠常驻登记,全靠文件 + 系统探测(list/check 时现算):
  exit_code 存在 → finished;否则 host 查 wrapper pid 活着 / docker 查容器
  running → running;都探不到 → lost(宿主重启把进程带走了,产物文件仍在)。

边界(DESIGN §8.12,防滑坡):只覆盖「单个本地长进程」。job 链/依赖/重试策略
不进这里 —— 编排的唯一归属是 agent loop 本身(模型在 check 后自己决定下一步)。
"""
from __future__ import annotations

import json
import os
import re
import secrets
import shutil
import subprocess
import sys
import threading
import time
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

PROCS_SUBDIR = ".zcbot_procs"

# bg 进程默认/最大运行时长(秒);前台 60/120s 的默认值不放大 —— 它是逼模型
# 在"快命令"与"长任务"之间做选择的杠杆
DEFAULT_TIMEOUT_S = 7200
MAX_TIMEOUT_S = 86400

# 每用户并发 bg 进程上限(跨 task 合计);防模型失控起一堆
MAX_RUNNING_PER_USER = int(os.getenv("ZCBOT_MAX_BG_PROCS", "3"))

# 终态 proc 目录保留时长,sweep 超期删除
FINISHED_TTL_S = 7 * 86400

_ID_RE = re.compile(r"^[A-Za-z0-9_-]+$")


def new_proc_id() -> str:
    return "p" + secrets.token_hex(4)


def procs_root(anchor: Path) -> Path:
    """anchor = user_root(常规)或 base_dir(CLI --working-dir 无 user_root 兜底)。"""
    return Path(anchor) / PROCS_SUBDIR


def proc_dir(anchor: Path, task_id: str, proc_id: str) -> Optional[Path]:
    """proc_id 来自模型输入,先校验再拼路径(防 ../ 越界)。非法返 None。"""
    if not (_ID_RE.match(str(task_id) or "") and _ID_RE.match(proc_id or "")):
        return None
    return procs_root(anchor) / str(task_id) / proc_id


def write_meta(d: Path, meta: Dict[str, Any]) -> None:
    """原子写(tmp + replace):wrapper 与 check_process 并发读写不撕裂。"""
    tmp = d / "proc.json.tmp"
    tmp.write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8")
    os.replace(tmp, d / "proc.json")


def read_meta(d: Path) -> Optional[Dict[str, Any]]:
    try:
        return json.loads((d / "proc.json").read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None


def clamp_timeout(timeout: Optional[int]) -> int:
    try:
        t = int(timeout) if timeout else DEFAULT_TIMEOUT_S
    except (TypeError, ValueError):
        t = DEFAULT_TIMEOUT_S
    return max(60, min(t, MAX_TIMEOUT_S))


def count_running(anchor: Path) -> int:
    """该用户(anchor=user_root)当前 running 状态的 bg 进程数,跨 task 合计。"""
    root = procs_root(anchor)
    if not root.is_dir():
        return 0
    n = 0
    for tdir in root.iterdir():
        if not tdir.is_dir():
            continue
        for d in tdir.iterdir():
            meta = read_meta(d) if d.is_dir() else None
            if meta and status_of(meta, d)[0] == "running":
                n += 1
    return n


# ───────────── 状态探测 ─────────────

def _pid_alive(pid: int) -> bool:
    if pid <= 0:
        return False
    if os.name == "posix":
        try:
            os.kill(pid, 0)
            return True
        except ProcessLookupError:
            return False
        except PermissionError:
            return True
        except OSError:
            return False
    # Windows:OpenProcess 探测(PID 复用误报概率低,可接受)
    import ctypes
    PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
    h = ctypes.windll.kernel32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
    if not h:
        return False
    try:
        code = ctypes.c_ulong()
        ok = ctypes.windll.kernel32.GetExitCodeProcess(h, ctypes.byref(code))
        STILL_ACTIVE = 259
        return bool(ok) and code.value == STILL_ACTIVE
    finally:
        ctypes.windll.kernel32.CloseHandle(h)


def _container_running(name: str) -> bool:
    try:
        r = subprocess.run(
            ["docker", "inspect", "--type=container",
             "--format={{.State.Running}}", name],
            capture_output=True, text=True, timeout=15,
        )
        return r.returncode == 0 and r.stdout.strip() == "true"
    except (OSError, subprocess.TimeoutExpired):
        return False


def status_of(meta: Dict[str, Any], d: Path) -> Tuple[str, Optional[int]]:
    """→ ("running"|"finished"|"lost", exit_code|None)。exit_code 文件是唯一终态锚。"""
    try:
        raw = (d / "exit_code").read_text(encoding="utf-8").strip()
        return "finished", int(raw)
    except (OSError, ValueError):
        pass
    if meta.get("backend") == "docker":
        if _container_running(str(meta.get("container") or "")):
            return "running", None
        return "lost", None
    pid = int(meta.get("wrapper_pid") or 0)
    if _pid_alive(pid):
        return "running", None
    return "lost", None


def tail_log(d: Path, max_bytes: int = 8_000) -> str:
    p = d / "output.log"
    try:
        size = p.stat().st_size
        with open(p, "rb") as f:
            if size > max_bytes:
                f.seek(size - max_bytes)
            data = f.read()
        text = data.decode("utf-8", errors="replace")
        if size > max_bytes:
            text = f"[...前 {size - max_bytes} 字节省略...]\n" + text
        return text
    except OSError:
        return "(暂无输出)"


def list_procs(anchor: Path, task_id: str) -> List[Dict[str, Any]]:
    """该 task 的所有 bg proc,带现算 status/exit_code,按创建时间排。"""
    tdir = procs_root(anchor) / str(task_id)
    if not tdir.is_dir():
        return []
    out = []
    for d in sorted(tdir.iterdir()):
        if not d.is_dir():
            continue
        meta = read_meta(d)
        if not meta:
            continue
        st, ec = status_of(meta, d)
        meta["_status"], meta["_exit_code"], meta["_dir"] = st, ec, d
        out.append(meta)
    out.sort(key=lambda m: m.get("created_ts") or 0)
    return out


def list_all_procs(anchor: Path) -> List[Dict[str, Any]]:
    """该用户(anchor=user_root)全部 task 的 bg proc(web `/v1/procs` 用:跨 task
    通知——用户切到别的 task 也能收到完成提示)。"""
    root = procs_root(anchor)
    if not root.is_dir():
        return []
    out: List[Dict[str, Any]] = []
    for tdir in sorted(root.iterdir()):
        if tdir.is_dir():
            out.extend(list_procs(anchor, tdir.name))
    return out


def finished_at(d: Path) -> Optional[float]:
    """终态时刻 = exit_code 文件 mtime(结束时才写,天然就是完成时间戳)。"""
    try:
        return (d / "exit_code").stat().st_mtime
    except OSError:
        return None


# ───────────── host 模式启动 / 终止 ─────────────

def launch_host(
    anchor: Path,
    task_id: str,
    *,
    kind: str,                       # "shell" | "python"
    command_display: str,            # 给 LLM/用户看的命令描述
    argv: Optional[List[str]] = None,        # 二选一:argv(python)
    shell_cmd: Optional[str] = None,         # 或 shell 命令串
    cwd: Path,
    timeout_s: int,
    env: Optional[Dict[str, str]] = None,
    inline_code: Optional[str] = None,       # run_python inline → 落 script.py,argv 引用它
) -> Tuple[str, Path]:
    """detach 启动 wrapper,立即返回 (proc_id, proc_dir)。

    wrapper 是独立脚本(core/proc_wrapper.py,stdlib-only),用当前解释器直跑
    (非 -m,不依赖 PYTHONPATH),zcbot 进程退出/重启不影响它。
    """
    proc_id = new_proc_id()
    d = procs_root(anchor) / str(task_id) / proc_id
    d.mkdir(parents=True, exist_ok=True)

    if inline_code is not None:
        script = d / "script.py"
        script.write_text(inline_code, encoding="utf-8")
        argv = [sys.executable, str(script)]

    meta: Dict[str, Any] = {
        "proc_id": proc_id,
        "task_id": str(task_id),
        "kind": kind,
        "backend": "host",
        "command": command_display[:500],
        "cwd": str(cwd),
        "timeout_s": timeout_s,
        "created_at": datetime.now().isoformat(timespec="seconds"),
        "created_ts": time.time(),
    }
    if argv:
        meta["argv"] = argv
    else:
        meta["shell_cmd"] = shell_cmd
    write_meta(d, meta)

    wrapper = Path(__file__).parent / "proc_wrapper.py"
    spawn_kw: Dict[str, Any] = dict(
        stdin=subprocess.DEVNULL,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        env=env,
    )
    if os.name == "posix":
        spawn_kw["start_new_session"] = True
    else:
        spawn_kw["creationflags"] = (
            subprocess.DETACHED_PROCESS
            | subprocess.CREATE_NEW_PROCESS_GROUP
            | subprocess.CREATE_NO_WINDOW
        )
    p = subprocess.Popen([sys.executable, str(wrapper), str(d)], **spawn_kw)
    meta["wrapper_pid"] = p.pid
    write_meta(d, meta)
    # 后台线程 wait 掉 wrapper,避免 POSIX 僵尸挂在 web 进程下
    threading.Thread(target=p.wait, daemon=True).start()
    return proc_id, d


def kill_proc(meta: Dict[str, Any], d: Path) -> str:
    """强杀。docker → rm -f 容器;host → taskkill /T(win)/ killpg(posix)。
    杀完补写 exit_code=137,让状态收敛到 finished(killed)。"""
    st, _ = status_of(meta, d)
    if st == "finished":
        return "进程已结束,无需终止"
    if meta.get("backend") == "docker":
        name = str(meta.get("container") or "")
        if name:
            subprocess.run(["docker", "rm", "-f", name], capture_output=True, timeout=30)
    else:
        wrapper_pid = int(meta.get("wrapper_pid") or 0)
        child_pid = int(meta.get("child_pid") or 0)
        if os.name == "posix":
            import signal
            for pid in (child_pid, wrapper_pid):
                if pid > 0:
                    try:
                        os.killpg(pid, signal.SIGKILL)
                    except (ProcessLookupError, PermissionError, OSError):
                        try:
                            os.kill(pid, signal.SIGKILL)
                        except OSError:
                            pass
        else:
            for pid in (wrapper_pid, child_pid):
                if pid > 0:
                    subprocess.run(
                        ["taskkill", "/PID", str(pid), "/T", "/F"],
                        capture_output=True, timeout=30,
                    )
    try:
        (d / "exit_code").write_text("137", encoding="utf-8")
    except OSError:
        pass
    meta["killed"] = True
    write_meta(d, meta)
    return "已终止"


# ───────────── 清扫(web lifespan 周期调用) ─────────────

def sweep(user_root_base: Path, ttl_s: int = FINISHED_TTL_S) -> Dict[str, int]:
    """两件事:① 终态超 TTL 的 proc 目录删掉 ② 已结束(exit_code 在)但容器还挂着
    的 docker proc 容器 rm -f(runner 写完退出码后容器空转 sleep,等这里回收)。
    另兜底:label=zcbot.product=proc 的孤儿容器(目录已删/exited)一并清。
    幂等,蓝绿双实例同时跑无害(rm -f / rmtree 都幂等)。"""
    removed_dirs = reaped_containers = 0
    now = time.time()
    base = Path(user_root_base)
    if base.is_dir():
        for uroot in base.iterdir():
            # 单条目异常(权限/竞态删除)不打断整轮清扫
            try:
                root = uroot / PROCS_SUBDIR
                if not root.is_dir():
                    continue
                for tdir in list(root.iterdir()):
                    if not tdir.is_dir():
                        continue
                    for d in list(tdir.iterdir()):
                        if not d.is_dir():
                            continue
                        meta = read_meta(d)
                        if not meta:
                            continue
                        st, _ = status_of(meta, d)
                        if st != "running" and meta.get("backend") == "docker":
                            name = str(meta.get("container") or "")
                            if name and _container_exists(name):
                                subprocess.run(
                                    ["docker", "rm", "-f", name],
                                    capture_output=True, timeout=30,
                                )
                                reaped_containers += 1
                        age = now - float(meta.get("created_ts") or now)
                        if st != "running" and age > ttl_s:
                            shutil.rmtree(d, ignore_errors=True)
                            removed_dirs += 1
                    try:
                        tdir.rmdir()  # 空了顺手收掉,非空抛 OSError 忽略
                    except OSError:
                        pass
            except OSError:
                continue
    # 孤儿容器兜底(proc 目录已被删,或容器已 exited)
    try:
        r = subprocess.run(
            ["docker", "ps", "-aq", "--filter", "label=zcbot.product=proc",
             "--filter", "status=exited"],
            capture_output=True, text=True, timeout=30,
        )
        ids = r.stdout.split() if r.returncode == 0 else []
        if ids:
            subprocess.run(["docker", "rm", "-f", *ids], capture_output=True, timeout=60)
            reaped_containers += len(ids)
    except (OSError, subprocess.TimeoutExpired):
        pass  # docker 不在(host 模式部署)→ 只做文件清扫
    return {"removed_dirs": removed_dirs, "reaped_containers": reaped_containers}


def _container_exists(name: str) -> bool:
    try:
        r = subprocess.run(
            ["docker", "inspect", "--type=container", name],
            capture_output=True, timeout=15,
        )
        return r.returncode == 0
    except (OSError, subprocess.TimeoutExpired):
        return False


# ───────────── LLM 面向的描述格式化(check_process / 启动返回共用) ─────────────

def format_proc_line(meta: Dict[str, Any]) -> str:
    st = meta.get("_status", "?")
    ec = meta.get("_exit_code")
    elapsed = ""
    ts = meta.get("created_ts")
    if ts:
        elapsed = f" · 已运行 {_fmt_elapsed(time.time() - float(ts))}" if st == "running" else ""
    tail = f"(exit {ec})" if st == "finished" else ""
    return (
        f"- {meta.get('proc_id')} [{st}{tail}] {meta.get('kind')}: "
        f"{meta.get('command', '')[:120]}{elapsed}"
    )


def _fmt_elapsed(s: float) -> str:
    s = int(s)
    if s < 60:
        return f"{s}s"
    if s < 3600:
        return f"{s // 60}m{s % 60:02d}s"
    return f"{s // 3600}h{(s % 3600) // 60:02d}m"