vent_agent_2.0/release/deploy.py

#!/usr/bin/env python3
"""Linux 部署包安装、版本切换与单进程服务管理。"""
from __future__ import annotations

import argparse
from contextlib import contextmanager
import fcntl
import json
import os
from pathlib import Path
import shutil
import signal
import subprocess
import sys
import tempfile
import time
import urllib.request

sys.dont_write_bytecode = True
from common import validate_package, validation_env, version

PACKAGE = Path(__file__).resolve().parents[1]
SHARED_DIRS = ("data", "workspace", "conversation_history", "large_tool_results")


def default_prefix() -> Path:
    return PACKAGE.parent.parent if PACKAGE.parent.name == "releases" else Path.home() / "vent-agent"


@contextmanager
def locked(path: Path):
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a+") as handle:
        try:
            fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            raise ValueError(f"已有操作或实例占用：{path}") from None
        try:
            yield
        finally:
            fcntl.flock(handle, fcntl.LOCK_UN)


def running(prefix: Path) -> bool:
    try:
        with locked(prefix / "shared/run/instance.lock"):
            return False
    except ValueError:
        return True


def record_path(prefix: Path) -> Path:
    return prefix / "shared/run/server.json"


def read_record(prefix: Path) -> dict:
    try:
        return json.loads(record_path(prefix).read_text())
    except (OSError, ValueError):
        return {}


def start_ticks(pid: int) -> str | None:
    try:
        # comm 可包含空格和括号；starttime 是右括号后第 20 个字段。
        fields = Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split()
        return None if fields[0] == "Z" else fields[19]
    except (OSError, IndexError):
        return None


def same_process(record: dict) -> bool:
    pid = record.get("pid", 0)
    return isinstance(pid, int) and pid > 1 and bool(record.get("start_ticks")) and start_ticks(pid) == record["start_ticks"]


def current_release(prefix: Path) -> Path:
    link = prefix / "current"
    if not link.is_symlink():
        raise ValueError("请先 activate 一个已安装版本")
    root = link.resolve(strict=True)
    if root.parent != prefix / "releases":
        raise ValueError("current 必须指向当前部署目录的 releases/<版本>")
    return root


def install(prefix: Path):
    manifest = validate_package(PACKAGE)
    target = prefix / "releases" / version(manifest["version"])
    if target.exists() or target.is_symlink():
        raise ValueError(f"版本已安装，不覆盖：{target}")
    if prefix == PACKAGE or PACKAGE in prefix.parents or prefix in PACKAGE.parents:
        raise ValueError("安装目录与解压目录必须分开")
    target.parent.mkdir(parents=True, exist_ok=True)
    target.mkdir()
    try:
        for name in [*manifest["files"], "release-manifest.json"]:
            dest = target / name
            dest.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(PACKAGE / name, dest)
        python = target / ".venv/bin/python"
        subprocess.run([sys.executable, "-m", "venv", str(target / ".venv")], check=True)
        subprocess.run([str(python), "-m", "pip", "install", "--disable-pip-version-check",
                        "--require-hashes", "--only-binary=:all:", "-r", str(target / "release/requirements.lock")], check=True)
        subprocess.run([str(python), "-m", "pip", "check"], check=True)
        # 自检只操作临时数据；在连接 shared 前执行，升级安装不会触碰线上数据库。
        with tempfile.TemporaryDirectory(prefix="vent-install-") as temporary:
            subprocess.run([str(python), "scripts/selfcheck.py"], cwd=target,
                           env=validation_env(Path(temporary)), check=True)
        shared = prefix / "shared"
        shared.mkdir(exist_ok=True)
        for name in (*SHARED_DIRS, "logs", "run"):
            (shared / name).mkdir(exist_ok=True)
        env_file = shared / ".env"
        if not env_file.exists():
            fd = os.open(env_file, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
            with os.fdopen(fd, "wb") as stream:
                stream.write((target / ".env.example").read_bytes())
        for name in (".env", *SHARED_DIRS):
            path = target / name
            if path.exists():  # 仅为这个新版本的隔离自检生成的目录
                shutil.rmtree(path) if path.is_dir() else path.unlink()
            path.symlink_to(shared / name, target_is_directory=name != ".env")
        validate_package(target, installed=True)
    except BaseException:
        shutil.rmtree(target)
        raise
    print(f"已安装 {manifest['version']}：{target}\n配置：{prefix / 'shared/.env'}\n请配置后执行 activate，再 start。")


def replace_link(path: Path, destination: Path):
    temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
    try:
        temporary.symlink_to(destination)
        os.replace(temporary, path)
    finally:
        temporary.unlink(missing_ok=True)


def activate(prefix: Path, selected: str):
    if running(prefix):
        raise ValueError("请先 stop，再切换版本")
    target = prefix / "releases" / version(selected)
    if target.is_symlink() or not (target / ".venv/bin/python").is_file():
        raise ValueError(f"版本尚未安装：{selected}")
    validate_package(target, installed=True)
    with locked(prefix / "shared/run/instance.lock"):
        current = prefix / "current"
        if current.exists() and not current.is_symlink():
            raise ValueError("current 已存在且不是符号链接")
        if current.is_symlink() and current.resolve() != target:
            replace_link(prefix / "previous", current_release(prefix))
        replace_link(current, target)
    print(f"当前版本：{selected}")


def serve(prefix: Path):
    with locked(prefix / "shared/run/instance.lock"):
        if current_release(prefix) != PACKAGE:
            raise ValueError("启动期间版本已切换，请重试")
        validate_package(PACKAGE, installed=True)
        os.chdir(PACKAGE)
        sys.path.insert(0, str(PACKAGE))
        import config
        import uvicorn
        record = {"pid": os.getpid(), "start_ticks": start_ticks(os.getpid()),
                  "version": config.APP_VERSION, "release": str(PACKAGE), "host": config.HOST, "port": config.PORT}
        path = record_path(prefix)
        temporary = path.with_suffix(".tmp")
        temporary.write_text(json.dumps(record))
        os.replace(temporary, path)
        try:
            uvicorn.run("main:app", host=config.HOST, port=config.PORT, workers=1,
                        timeout_graceful_shutdown=60, log_level=config.LOG_LEVEL.lower(),
                        log_config=None, access_log=False)
        finally:
            if read_record(prefix).get("pid") == os.getpid():
                path.unlink(missing_ok=True)


def health(record: dict) -> bool:
    if not same_process(record):
        return False
    host = record["host"]
    host = "127.0.0.1" if host in ("0.0.0.0", "") else "::1" if host == "::" else host
    if ":" in host:
        host = f"[{host}]"
    try:
        opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
        with opener.open(f"http://{host}:{record['port']}/health", timeout=2) as response:
            result = json.load(response)
        return result.get("status") == "ok" and result.get("pid") == record["pid"] and result.get("version") == record["version"]
    except (OSError, ValueError):
        return False


def stop(prefix: Path, timeout: float):
    if not running(prefix):
        record_path(prefix).unlink(missing_ok=True)
        print("服务已停止")
        return
    record = read_record(prefix)
    if not same_process(record):
        raise ValueError("实例锁被占用，但无法确认进程身份；请检查日志，不自动终止其他进程")
    # pidfd 避免检查后 PID 被回收再复用时误杀其他进程。
    descriptor = os.pidfd_open(record["pid"])
    try:
        if not same_process(record):
            raise ValueError("进程身份已变化，请重新检查状态")
        signal.pidfd_send_signal(descriptor, signal.SIGTERM)
    finally:
        os.close(descriptor)
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if not running(prefix):
            print("服务已停止")
            return
        time.sleep(0.2)
    raise ValueError("停止超时；服务可能仍在收尾，请检查日志后重试")


def start(prefix: Path, timeout: float):
    if running(prefix):
        raise ValueError("服务已经运行，请使用 status 查看")
    root = current_release(prefix)
    validate_package(root, installed=True)
    log = prefix / "shared/logs/server.log"
    log.parent.mkdir(parents=True, exist_ok=True)
    with log.open("ab") as stream:
        child = subprocess.Popen([str(root / ".venv/bin/python"), "-B", str(root / "release/deploy.py"),
                                  "_serve", "--prefix", str(prefix)], cwd=root,
                                 stdin=subprocess.DEVNULL, stdout=stream, stderr=subprocess.STDOUT, start_new_session=True)
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if child.poll() is not None:
            raise ValueError(f"启动失败（退出码 {child.returncode}），请查看 {log}")
        record = read_record(prefix)
        if record.get("pid") == child.pid and health(record):
            print(f"服务已启动：版本 {record['version']}，PID {child.pid}，端口 {record['port']}\n日志：{log}")
            return
        time.sleep(0.3)
    # 只终止自己创建且尚未回收的子进程。
    child.terminate()
    try:
        child.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        pass
    raise ValueError(f"启动健康检查超时，已请求停止；请查看 {log}")


def main():
    os.umask(0o077)
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("command", choices=["install", "activate", "start", "stop", "status", "logs", "run", "_serve"])
    parser.add_argument("version", nargs="?")
    parser.add_argument("--prefix", type=Path, default=default_prefix())
    parser.add_argument("--timeout", type=float, default=90)
    args = parser.parse_args()
    prefix = args.prefix.expanduser().resolve()
    try:
        if sys.platform != "linux" or sys.version_info[:2] != (3, 12):
            raise ValueError("部署脚本需要 Linux + Python 3.12")
        if args.timeout <= 0:
            raise ValueError("timeout 必须大于 0")
        if args.command == "_serve":
            serve(prefix)
        elif args.command == "run":
            root = current_release(prefix)
            python = root / ".venv/bin/python"
            os.execv(python, [str(python), "-B", str(root / "release/deploy.py"), "_serve", "--prefix", str(prefix)])
        elif args.command == "logs":
            subprocess.run(["tail", "-n", "100", "-f", str(prefix / "shared/logs/server.log")], check=True)
        elif args.command == "status":
            active = running(prefix)
            record = read_record(prefix)
            ready = active and health(record)
            print(json.dumps({"running": active, "healthy": ready, "process": record}, ensure_ascii=False, indent=2))
            return 0 if ready else 1
        else:
            with locked(prefix / "shared/run/control.lock"):
                if args.command == "install":
                    install(prefix)
                elif args.command == "activate":
                    if not args.version:
                        raise ValueError("用法：service.sh activate <版本>")
                    activate(prefix, args.version)
                elif args.command == "start":
                    start(prefix, args.timeout)
                elif args.command == "stop":
                    stop(prefix, args.timeout)
    except (OSError, ValueError, subprocess.CalledProcessError) as exc:
        parser.exit(1, f"操作失败：{exc}\n")
    return 0


if __name__ == "__main__":
    sys.exit(main())