efi-kernel/内核/进程.py

"""通用进程库: 启动 / 停止 / 判活 / 日志.

[谁用它]
  内核用它管**驱动**进程, 引导器用它管**内核**进程 -- 只此一份实现, 绝不复制
  (设计 02 §7 原话: "只此一份实现, 不复制").两套实现 = 各有一套坑, 还各自以为对方对.

[为什么全走 /proc, 不用 ps]
  * busybox 的 ps 会截断命令行 (软路由上踩过), 长 cmdline 根本看不全
  * `ps | grep` 会把**你自己的排查命令**也算进去, 得到假计数 (踩过好几次)
  * /proc/<pid>/cmdline 是 NUL 分隔的 argv 原文, 精确点名, 不用二次解析

[四条铁律 (都是踩过的坑, 改这个文件前先读一遍)]
  1. 发信号前必须校验 cmdline -- pid 会被系统复用, 裸 kill <pid> 可能杀到别人的进程
     (老板的 GUI 程序就这么被误杀过一次, 被点名批评)
  2. start_new_session=True 起独立进程组 -- 停止时能连子树一起收 (杀父不等于杀子树:
     python 死了, 它拉起的子进程会变孤儿继续跑,继续写同一个文件)
  3. 启动与停止分两条命令, 一次只起一份 -- 同一个 shell 里连着做会留孤儿互抢端口
  4. 排查动作本身会破坏现场: 反复起/杀会攒出一堆孤儿, 让人误判成"程序不稳定"
     (曾经攒出 13 份隧道孤儿互抢端口).所以本库只做"一次一件事", 不自动重试.

[依赖]
  纯 stdlib.内核和引导器都直接 import 这个文件.
"""

from __future__ import annotations

import os
import signal
import subprocess
import sys
import time
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path

# ─────────────────────── 判活结果 (判活() 的返回值, 也是 driver_state.state 的词汇表) ───────────────────────
# 五个值各有明确含义, 调用方按它决定"认领 / 收尸 / 不认领":
运行中: str = "running"  # /proc/<pid> 在, 且 cmdline 校验通过 -> 是我们那个进程, 认领
已停止: str = "stopped"  # 没有 pid 记录, 或 /proc/<pid> 根本不在了 -> 正常态, 幂等
已崩: str = "crashed"  # PG 里记着 pid 但进程没了 (断电 / 被杀) -> 该收尸, 按 restart 策略决定要不要拉
被复用: str = "exited"  # pid 还在, 但 cmdline 不是我们的 (系统把 pid 分给别人了) -> 不认领, 更不许杀
僵尸: str = "zombie"  # 进程其实已死, 但父进程没 wait 回收 (Z 态) -> 杀不动, 只能等父进程收

# /proc/<pid>/stat 里的 utime / stime / starttime 单位是"节拍"(jiffies), 除以它才是秒
时钟频率: int = os.sysconf("SC_CLK_TCK")
# /proc/<pid>/statm 的单位是"页", 乘它才是字节 (拿来做 RSS)
页大小: int = os.sysconf("SC_PAGE_SIZE")


@dataclass
class 进程信息:
    """从 /proc 直读出来的一份进程快照 (读不到就返回 None, 不抛异常).

    字段全部来自 /proc/<pid>/{stat,statm,cmdline}, 不来自 PG 里的旧 pid 记录.
    """

    pid: int  # 进程号
    ppid: int  # 父进程号 (看进程归属)
    pgid: int  # 进程组号 (收子树的关键: killpg(-pgid))
    state: str  # 单字母状态: R 运行 / S 睡眠 / D 不可中断 / Z 僵尸 / T 停止
    cmdline: list[str]  # argv 原文 (NUL 分隔切好的); 空表 = 读不到 (权限 / 内核线程 / 僵尸)
    rss_kb: int  # 常驻内存, KB (来自 statm 的第 2 列 * 页大小)
    cpu秒: float  # 用户态 + 内核态 CPU 时间, 秒 (utime + stime)
    启动时刻: float  # unix 时间戳 (由 开机秒数 - starttime 反推)
    存活秒: float  # 已经跑了多久 (秒)

    @property
    def cmdline文本(self) -> str:
        """空格拼起来的命令行, 给人看 / 写日志用."""
        return " ".join(self.cmdline)

    @property
    def 内存文本(self) -> str:
        """RSS 的人读格式 (KB / MB / GB)."""
        if self.rss_kb >= 1024 * 1024:
            return f"{self.rss_kb / 1024 / 1024:.2f} GB"
        if self.rss_kb >= 1024:
            return f"{self.rss_kb / 1024:.1f} MB"
        return f"{self.rss_kb} KB"

    @property
    def 存活文本(self) -> str:
        """存活时长的人读格式 (HH:MM:SS)."""
        return 时长文本(self.存活秒)


@dataclass
class 启动结果:
    """启动() 的返回.秒退也算返回 (不是异常), 由调用方决定怎么记账.

    字段:
        ok:        起成功了 (探活期内没退).
        pid:       子进程 pid; spawn 都失败时为 None.
        pgid:      子进程的进程组号 (独立会话所以通常 == pid); 没活下来时 None.
        exit_code: 秒退时的退出码; 没退则 None.
        seconds:   从 spawn 到判定的秒数.
        detail:    人读的一句话结论 (给 PG 的 last_error / detail 用, 带原文不吞错).
    """

    ok: bool
    pid: int | None
    pgid: int | None
    exit_code: int | None
    seconds: float
    detail: str


@dataclass
class 停止结果:
    """停止() 的返回.

    字段:
        ok:     结果干净 (进程本来就不在 / SIGTERM 收工 / SIGKILL 后收干净).
        forced: 是否升过级 (用过 SIGKILL).
        seconds: 总共等了多久.
        detail: 人读结论; pid 被复用时说明"不发信号"的原因.
    """

    ok: bool
    forced: bool
    seconds: float
    detail: str


@dataclass
class 命令结果:
    """跑命令() / 跑并转发() 的返回 (给 uv / pip / psql / 探版本这类短命令用).

    字段:
        code:    退出码 (超时 124, 跑不起来 127, 跟 shell 惯例一致).
        seconds: 耗时.
        lines:   输出行 (已去掉空行; 跑并转发 时同时转发到了 stdout).
        detail:  出错时的一句话原因 (超时才填).
    """

    code: int
    seconds: float
    lines: list[str]
    detail: str = ""


# ─────────────────────────────── 底层读 /proc ───────────────────────────────


def 时长文本(秒: float) -> str:
    """秒 -> HH:MM:SS.负数按 0 处理 (时钟抖动时别出现 -00:00:01)."""
    整 = int(max(秒, 0.0))
    时 = 整 // 3600
    分 = (整 % 3600) // 60
    剩 = 整 % 60
    return f"{时:02d}:{分:02d}:{剩:02d}"


def 读cmdline(pid: int) -> list[str]:
    """读 /proc/<pid>/cmdline: argv 的原文, 元素之间是 NUL 字节 (\0).

    参数:
        pid: 进程号.
    返回:
        argv 列表; 进程不在 / 读不到 (权限,内核线程,僵尸) 时返回空表.

    说明:
        空表**不等于**"进程不在": 僵尸进程的 cmdline 就是空的.
        所以要判在不在, 用 判活() / Path("/proc/<pid>").exists(), 别只看这个.
    """
    try:
        原始 = Path(f"/proc/{pid}/cmdline").read_bytes()
    except OSError:
        return []
    return [段.decode("utf-8", "replace") for 段 in 原始.split(b"\0") if 段]


def 读stat(pid: int) -> list[str] | None:
    """读 /proc/<pid>/stat, 返回**去掉 pid 和 comm 之后**的字段列表.

    参数:
        pid: 进程号.
    返回:
        字段列表, 下标 0 == stat 的第 3 个字段 (state); 读不到返回 None.

    为什么这样切:
        stat 的第 2 个字段 comm 是括号包着的进程名, **里面可以带空格和右括号**
        (如 "(my prog),v2)"), 用 split() 切必错.唯一可靠的切法是取**最后一个**右括号,
        后面的部分再 split -- 所以下标要整体减 2 (下标 i 对应 stat 的第 i+3 个字段).

    常用下标 (下标 = 字段号 - 3):
        [0]  state      进程状态 (R/S/D/Z/T)
        [1]  ppid       父进程
        [2]  pgrp       进程组 (收子树用)
        [11] utime      用户态 jiffies
        [12] stime      内核态 jiffies
        [19] starttime  进程启动时刻 (相对开机, jiffies)
    """
    try:
        原文 = Path(f"/proc/{pid}/stat").read_text()
    except OSError:
        return None
    右括号 = 原文.rfind(")")
    if 右括号 < 0:
        return None
    return 原文[右括号 + 1 :].split()


def 读开机秒() -> float:
    """系统开机到现在的秒数 (/proc/uptime 的第 1 列).读不到返回 0."""
    try:
        return float(Path("/proc/uptime").read_text().split()[0])
    except (OSError, ValueError, IndexError):
        return 0.0


def 读进程信息(pid: int) -> 进程信息 | None:
    """把 /proc/<pid> 下几份文件拼成一份快照.

    参数:
        pid: 进程号.
    返回:
        进程信息; 进程不在 / 字段残缺 / 权限不够时返回 None (不抛).

    启动时刻怎么算的:
        stat 只给"相对开机时刻"的 starttime (jiffies), 换算:
            存活秒 = 开机秒数 - starttime / 时钟频率
            启动时刻(unix) = 当前时间 - 存活秒
        uptime 和某进程的 starttime 是两个不同时刻读的, 会有毫秒级误差, 够用.
    """
    if pid <= 0:
        return None
    字段 = 读stat(pid)
    if 字段 is None or len(字段) < 20:
        return None
    try:
        ppid = int(字段[1])
        pgid = int(字段[2])
        utime = int(字段[11])
        stime = int(字段[12])
        启动tick = int(字段[19])
    except (ValueError, IndexError):
        return None
    rss_kb = 0
    try:
        # statm 第 1 列是虚拟内存页数, 第 2 列才是常驻 (RSS) 页数
        页 = Path(f"/proc/{pid}/statm").read_text().split()
        rss_kb = int(页[1]) * 页大小 // 1024
    except (OSError, ValueError, IndexError):
        rss_kb = 0
    存活秒 = 读开机秒() - 启动tick / 时钟频率
    存活秒 = max(存活秒, 0.0)
    启动时刻 = time.time() - 存活秒
    return 进程信息(
        pid=pid,
        ppid=ppid,
        pgid=pgid,
        state=字段[0],
        cmdline=读cmdline(pid),
        rss_kb=rss_kb,
        cpu秒=(utime + stime) / 时钟频率,
        启动时刻=启动时刻,
        存活秒=存活秒,
    )


# ─────────────────────────────── 判活 / 认领 ───────────────────────────────


def 规范化(路径: str | Path) -> str:
    """转成 realpath 绝对路径, 专门用于"这条 cmdline 指的是不是那个文件"的精确比对.

    为什么要 realpath:
        启动时可能给的是相对路径,带软链的路径,带 ../ 的路径; 而 /proc 里记的是当时的写法.
        两边都 realpath 到真实文件再比, 才判得准.
    """
    return os.path.realpath(os.fspath(路径))


def 匹配入口(pid: int, 入口: Path) -> bool:
    """argv 里有没有哪个元素 realpath 后正好等于这个入口文件.

    参数:
        pid: 要校验的进程.
        入口: 入口文件的绝对路径 (如 内核/内核.py,驱动/<名>/json解码.py).
    返回:
        命中 True / 不命中 False.

    为什么不写成子串匹配:
        子串匹配会把"我的排查命令里提到了这个文件"也算命中 (踩过: 宽松匹配 `*aria2c*x*`
        连自己那条 ssh 一起匹配上, 真把自己 shell 杀了).这里只认 realpath 完全相等.

    局限:
        只认"argv 里直接出现入口文件"这一种写法.`python -m 包` 这类入口 (argv 里没有文件路径)
        匹配不上 -- 本项目不用这种写法, 驱动/内核都是 `解释器 + 入口文件` 的形态.
    """
    目标 = 规范化(入口)
    for 段 in 读cmdline(pid):
        if 规范化(段) == 目标:
            return True
    return False


def 判活(pid: int | None, 入口: Path | None = None) -> str:
    """判一个进程现在是什么状态 (本库最核心的判定).

    参数:
        pid: 要判的进程号; None / <=0 直接当"没在跑".
        入口: 给了就做 cmdline 校验 (防 pid 复用); 不给只判在不在.
    返回:
        五个状态之一 (见文件头的常量注释): 运行中 / 已停止 / 已崩 / 被复用 / 僵尸.

    判定顺序 (顺序本身有讲究):
        1. 没 pid -> 已停止
        2. /proc/<pid> 不存在 -> 已崩 (我们以为它在跑, 其实没了)
        3. state == Z -> 僵尸 (已经死了, 只是父进程还没回收; 如实报, 不谎报运行中)
        4. 给了入口但 cmdline 对不上 -> 被复用 (pid 被系统分给了别人)
        5. 其余 -> 运行中
        第 4 步放最后: pid 复用概率低但后果严重 (误杀别人), 前面几步都是便宜判断.
    """
    if pid is None or pid <= 0:
        return 已停止
    if not Path(f"/proc/{pid}").exists():
        return 已崩
    字段 = 读stat(pid)
    if 字段 is not None and 字段[0] == "Z":
        return 僵尸
    if 入口 is not None and not 匹配入口(pid, 入口):
        return 被复用
    return 运行中


def 按入口找进程(入口: Path) -> list[int]:
    """扫 /proc 所有进程, 把 cmdline 里有这个入口文件的 pid 全找出来.

    参数:
        入口: 入口文件绝对路径.
    返回:
        升序 pid 列表; 没有则空表.

    为什么不用 ps / pgrep:
        见文件头.这里逐 pid 读 /proc/<pid>/cmdline, 顺便跳过自己 (否则"查找命令自己"
        可能被算成一个匹配项, 得到假计数).

    用途:
        引导器拿它找内核进程 (不依赖 PG, PG 挂了也能找); 内核拿它找驱动进程.
    """
    出: list[int] = []
    自己 = os.getpid()
    for 项 in Path("/proc").iterdir():
        if not 项.name.isdigit():
            continue
        pid = int(项.name)
        if pid == 自己:
            continue
        if 匹配入口(pid, 入口):
            出.append(pid)
    出.sort()
    return 出


def 进程组成员(pgid: int) -> list[int]:
    """列出某个进程组里**还活着**的成员 pid (收残用).

    参数:
        pgid: 进程组号 (一般来自 进程信息.pgid).
    返回:
        升序 pid 列表, 不含自己, **不含僵尸**.

    为什么跳过僵尸:
        僵尸是"已死但没被父进程 wait 回收"的空壳, SIGKILL 也收不掉.
        如果算进来, 停止就会永远报"还有 N 个没收掉", 让人以为没停干净 (踩过).

    局限:
        进程组号会被复用.这个函数只在"刚对自己确认过的进程组发完信号"这个窗口里用,
        不要拿去当长期的身份判据.
    """
    出: list[int] = []
    自己 = os.getpid()
    for 项 in Path("/proc").iterdir():
        if not 项.name.isdigit():
            continue
        pid = int(项.name)
        if pid == 自己:
            continue
        字段 = 读stat(pid)
        if 字段 is None or len(字段) < 3:
            continue
        if 字段[0] == "Z":
            continue
        try:
            if int(字段[2]) == pgid:
                出.append(pid)
        except ValueError:
            continue
    出.sort()
    return 出


# ─────────────────────────────── 日志读取 ───────────────────────────────
# 注意分工 (2026-09-16 日志系统落地后):
#   内核 / 引导器 里的 "日志 <谁> 看几行 / -f 跟 / 按级别关键词筛 / 认轮转历史份" 都走
#   内核/日志.py (尾/跟) -- 那边带过滤与轮转感知, 是日志系统的实现 (设计 04).
#   本节这两个函数仍是**进程库自己的通用能力**: 从位置读日志 是"秒退只读本次输出"的判据
#   (启动() 在用, 别动), 尾日志 / 跟日志 保留给进程层的诊断 (自测进程.py 用 从位置读日志).


def 从位置读日志(路径: Path, 起始: int, 行数: int = 200) -> list[str]:
    """读日志文件里"第 起始 字节之后"的内容.

    参数:
        路径: 日志文件.
        起始: 字节偏移 (启动前记下的文件大小).
        行数: 最多返回末尾几行; <=0 表示不限.
    返回:
        行列表 (读不到返回空表).

    为什么需要它:
        日志是**追加**写的 (append).进程秒退时要回答"它临死说了什么", 如果直接读整个文件,
        会把上一次运行的输出也当成本次死因 (踩过: 退出码 3 的进程, 死因栏写着上一次的 "hi").
        所以启动前记下文件大小, 秒退时只读新增的那一段.
    """
    try:
        with open(路径, "rb") as 句柄:
            句柄.seek(max(起始, 0))
            数据 = 句柄.read()
    except OSError:
        return []
    行 = 数据.decode("utf-8", "replace").splitlines()
    if 行数 <= 0:
        return 行
    return 行[-行数:]


def 尾日志(路径: Path, 行数: int = 200) -> list[str]:
    """读日志末尾 N 行 (等价 tail -n).

    参数:
        路径: 日志文件.
        行数: 要几行; <=0 返回空表.
    返回:
        行列表; 文件不存在/读不到 返回空表 (不抛).

    说明:
        小文件直接整读再切片 (本项目日志没有几十 MB 的, 不做 seek 倒读那套复杂度).
    """
    if 行数 <= 0:
        return []
    try:
        原文 = 路径.read_text(errors="replace")
    except OSError:
        return []
    return 原文.splitlines()[-行数:]


def 跟日志(路径: Path) -> int:
    """`-f` 用的实时跟日志: 先吐末尾 10 行, 然后一直跟着新内容打印.

    参数:
        路径: 日志文件 (可以先不存在, 等它被创建).
    返回:
        0 (Ctrl-C 中断也返回 0, 不当失败).

    实现方式:
        轮询文件大小 + 从上次位置读新增字节, 每轮睡 0.3s.
        不用 subprocess 起 `tail -f`: 少一个外部依赖, 也好控制"文件被轮转/清空"的情况.
    文件被清空/轮转:
        大小比上次位置小 -> 位置归 0, 从头读 (否则会一直卡在旧偏移上什么都不出).
    """
    for 行 in 尾日志(路径, 10):
        print(行, flush=True)
    try:
        位置 = 路径.stat().st_size
    except OSError:
        位置 = 0
    try:
        while True:
            try:
                大小 = 路径.stat().st_size
            except OSError:
                大小 = 位置
            if 大小 < 位置:
                位置 = 0
            if 大小 > 位置:
                try:
                    with open(路径, "rb") as 句柄:
                        句柄.seek(位置)
                        数据 = 句柄.read()
                        位置 = 句柄.tell()
                except OSError:
                    数据 = b""
                if 数据:
                    sys.stdout.write(数据.decode("utf-8", "replace"))
                    sys.stdout.flush()
            else:
                time.sleep(0.3)
    except KeyboardInterrupt:
        return 0
    return 0


# ─────────────────────────────── 启动 ───────────────────────────────


def 启动(
    argv: list[str],
    cwd: Path,
    日志: Path,
    env: dict[str, str] | None = None,
    入口: Path | None = None,
    探活秒: float = 0.6,
    分隔: str = "",
) -> 启动结果:
    """拉起一个进程: 独立进程组, stdout/stderr 追加进日志文件, 探活一段时间再判定.

    参数:
        argv: 完整命令行 (第一个元素是解释器或可执行文件), **不经过 shell**.
        cwd: 工作目录 (驱动就起在驱动文件夹里; 内核起在项目根).
        env: 完整环境变量字典; None = 继承当前进程的.
       日志: stdout/stderr 追加到这个文件 (文件不存在会被创建, 父目录会自动建).
       入口: 入口文件绝对路径, 用于探活后做 cmdline 校验 (确认起的是我们要的那个).
       探活秒: spawn 后等多久再 poll (默认 0.6s -- 够 "入口不存在/解释器缺" 这类错暴露出来).
       分隔: 非空时先在日志里写一条分隔头 (如 "启动 样板常驻 <命令>") -- 日志是追加的,
             多轮启动的输出连成一片就分不清"这句是哪一次说的"; 分隔头把它切开,
             也顺手给人一个 tail 定位锚点.

    返回:
        启动结果 (见其注释).**秒退不抛异常**, 而是 ok=False + exit_code + 本次的日志尾巴,
        让调用方自己决定怎么记账.

    关键决定:
        start_new_session=True  -> 子进程自己开一个会话/进程组, 停止时能 killpg 连子树一起收
        stdin=DEVNULL           -> 别让子进程去抢终端输入 (后台跑着还要读键盘会挂住)
        stdout/stderr 同一个文件 -> 驱动/内核的原始输出都留着, 出问题时有据可查
        先记 起始字节偏移       -> 秒退时只读"这一次"的新增输出 (别冤枉上一次)
    """
    起点 = time.monotonic()
    if not argv:
        return 启动结果(False, None, None, None, 0.0, "argv 为空")
    try:
        日志.parent.mkdir(parents=True, exist_ok=True)
        句柄 = open(日志, "ab")
    except OSError as 错:
        return 启动结果(False, None, None, None, time.monotonic() - 起点, f"日志打不开: {错}")
    if 分隔.strip():
        try:
            句柄.write(f"\n==== {分隔}  {datetime.now().astimezone().isoformat(timespec='seconds')} ====\n".encode("utf-8"))
            句柄.flush()
        except OSError:
            pass
    try:
        起始 = 日志.stat().st_size  # 分隔头之后取, 秒退死因才只读"这一次"的输出
    except OSError:
        起始 = 0
    try:
        进程 = subprocess.Popen(  # noqa: S603 (argv 由本程序拼, 不经 shell)
            argv,
            cwd=str(cwd),
            stdin=subprocess.DEVNULL,
            stdout=句柄,
            stderr=subprocess.STDOUT,
            start_new_session=True,
            env=env,
        )
    except (OSError, ValueError) as 错:
        句柄.close()
        return 启动结果(False, None, None, None, time.monotonic() - 起点, f"spawn 失败: {错}")
    句柄.close()  # 子进程已经持有 fd, 这边及时关掉 (否则孤儿进程会把日志句柄一直拽着)
    time.sleep(max(探活秒, 0.0))
    码 = 进程.poll()
    秒 = time.monotonic() - 起点
    if 码 is not None:
        尾部 = " | ".join(从位置读日志(日志, 起始, 5))
        if 码 == 0:
            说明 = f"启动后 {秒:.1f}s 就退出 (退出码 0) -- 不像常驻进程"
        else:
            说明 = f"秒退, 退出码 {码}"
        if 尾部:
            说明 += f", 日志末尾: {尾部}"
        return 启动结果(False, 进程.pid, None, 码, 秒, 说明)
    信息 = 读进程信息(进程.pid)
    if 信息 is None:
        return 启动结果(False, 进程.pid, None, None, 秒, "读不到 /proc, 状态未知")
    校验 = ""
    if 入口 is not None and not 匹配入口(信息.pid, 入口):
        校验 = " (cmdline 里没看到入口文件 -- 起错了?)"
    return 启动结果(
        True, 信息.pid, 信息.pgid, None, 秒, f"已拉起 pid={信息.pid} pgid={信息.pgid}{校验}"
    )


# ─────────────────────────────── 停止 ───────────────────────────────


def _没了(pid: int) -> bool:
    """进程是不是"死透了"(/proc 不在, 或已成僵尸).

    僵尸死透了等回收, 再发信号也没意义 -- 所以"停干净"的判据是 _没了() 为真, 不是"kill 成功了".
    """
    if not Path(f"/proc/{pid}").exists():
        return True
    字段 = 读stat(pid)
    return 字段 is not None and 字段[0] == "Z"


def _等死(pid: int, 超时: float) -> list[int]:
    """等 pid 死透, 最多等 超时 秒.

    参数:
        pid: 目标进程.
        超时: 最多等多少秒.
    返回:
        死透了 -> 空表.
        超时了 -> 它所在进程组里**还活着**的 pid 列表 (收残目标; 不含自己, 不含僵尸).
    """
    结束 = time.monotonic() + max(超时, 0.0)
    while True:
        if _没了(pid):
            return []
        if time.monotonic() >= 结束:
            break
        time.sleep(0.1)
    信息 = 读进程信息(pid)
    组 = 信息.pgid if 信息 is not None else pid
    return 进程组成员(组)


def 停止(
    pid: int,
    入口: Path | None = None,
    超时: float = 10.0,
) -> 停止结果:
    """停一个进程 (连同它的子树), 全过程按"宁可不杀, 不可误杀"来.

    参数:
        pid: 目标进程号.
        入口: 给了就先做 cmdline 校验 -- 对不上就判"pid 被复用", **一个信号都不发**.
        超时: SIGTERM 之后等多久再升级 SIGKILL, 默认 10s (设计文档的 stop_timeout).

    返回:
        停止结果 (见其注释).

    流程:
        1. 判活: 已经不在 (已停止/已崩/僵尸) -> 直接当成功 (幂等, 重复停不报错);
           被复用 -> 不杀, 返回 ok=False + 原因;
        2. SIGTERM 发给**进程组** killpg(-pgid) -- 连子树一起收 (杀父不等于杀子树);
        3. 等 超时 秒;
        4. 还活着 -> SIGKILL 给进程组, 再按 pid 逐个收残 (只补刀, 不猜);
        5. 复查, 收干净才算 ok.

    为什么先 SIGTERM 不直接 SIGKILL:
        给进程一个收尾的机会 (刷缓冲区,写最后一条 events,删自己的 pid 文件).
        只有赖着不走的才升级.
    """
    起点 = time.monotonic()
    状态 = 判活(pid, 入口)
    if 状态 in (已停止, 已崩, 僵尸):
        return 停止结果(True, False, 0.0, "进程本来就不在 (幂等)")
    if 状态 == 被复用:
        return 停止结果(False, False, 0.0, f"pid {pid} 的 cmdline 不是我们的 (pid 被复用), 不发信号")
    信息 = 读进程信息(pid)
    组 = 信息.pgid if 信息 is not None else pid
    # ── 自杀防线 (2026-09-16 真踩过) ──
    # 目标如果跟我们在**同一个进程组** (典型场景: 引导器用 subprocess 前台起的常驻内核,
    # 它的进程组是从调用方继承来的), 对它 killpg 就等于给自己也来一下 --
    # 当时 `试跑引导器.py` 里的 `内核 停止` 把整条进程组收了, 脚本自己 exit -15.
    # 同组时只发**单进程**信号: 目标的子树归它自己收尾 (驱动都是独立会话, 不会漏).
    要打组 = 组 != os.getpgrp()

    def 发信号(号: int) -> None:
        """平时打整个进程组 (连子树一起收); 同组时只打目标自己 (免得把自己也收了)."""
        if 要打组:
            os.killpg(组, 号)
        else:
            os.kill(pid, 号)

    try:
        发信号(signal.SIGTERM)
    except ProcessLookupError:
        return 停止结果(True, False, time.monotonic() - 起点, "发信号前进程已退出")
    except PermissionError as 错:
        return 停止结果(False, False, time.monotonic() - 起点, f"没权限发信号: {错}")
    剩 = _等死(pid, 超时)
    if not 剩:
        哪打的 = f"进程组 {组}" if 要打组 else f"单进程 {pid} (跟我同组, 没打组)"
        return 停止结果(True, False, time.monotonic() - 起点, f"SIGTERM 收工 ({哪打的})")
    强制: list[int] = []
    if 要打组:
        try:
            os.killpg(组, signal.SIGKILL)
            强制.append(组)
        except (ProcessLookupError, PermissionError):
            pass
    for 目标 in 剩:
        if 目标 in (os.getpid(), os.getpgrp()):
            continue
        try:
            os.kill(目标, signal.SIGKILL)
            强制.append(目标)
        except (ProcessLookupError, PermissionError):
            continue
    剩 = _等死(pid, 3.0)
    if 剩:
        return 停止结果(
            False,
            True,
            time.monotonic() - 起点,
            f"SIGKILL 之后还有 {len(剩)} 个没收掉: {剩}",
        )
    return 停止结果(True, True, time.monotonic() - 起点, f"SIGTERM 超时 -> SIGKILL 收干净 (进程组 {组})")


# ─────────────────────────────── 跑命令 ───────────────────────────────


def 跑命令(
    argv: list[str],
    cwd: Path | None = None,
    超时: float = 60.0,
    env: dict[str, str] | None = None,
) -> 命令结果:
    """跑一条短命令, 收输出, 不转发给用户 (uv pip list / psql / 问解释器版本这类).

    参数:
        argv: 命令行.
        cwd: 工作目录; None = 当前目录.
        超时: 秒.超了返回 code=124 (跟 timeout(1) 惯例一致).
        env: 环境变量; None = 继承.
    返回:
        命令结果 (stdout + stderr 合并进 lines, 空行已去掉).
    """
    起点 = time.monotonic()
    try:
        完成 = subprocess.run(  # noqa: S603
            argv,
            cwd=None if cwd is None else str(cwd),
            stdin=subprocess.DEVNULL,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            timeout=超时,
            env=env,
            check=False,
        )
    except subprocess.TimeoutExpired:
        return 命令结果(124, time.monotonic() - 起点, [], f"超时 ({超时:.0f}s)")
    except (OSError, ValueError) as 错:
        return 命令结果(127, time.monotonic() - 起点, [], f"跑不起来: {错}")
    行 = [段 for 段 in 完成.stdout.splitlines() if 段.strip()]
    return 命令结果(完成.returncode, time.monotonic() - 起点, 行)


def 跑并转发(
    argv: list[str],
    cwd: Path,
    尾部行数: int = 200,
    env: dict[str, str] | None = None,
) -> 命令结果:
    """跑命令并把输出**实时转发**到 stdout, 同时留末尾若干行给台账.

    参数:
        argv: 命令行 (内核/驱动的启动命令).
        cwd: 工作目录.
        尾部行数: 保留多少行给调用方 (给 PG 的 detail 栏用; 失败时贴日志尾巴).
        env: 环境变量; 调用方一般会带 PYTHONUNBUFFERED=1 (见下).
    返回:
        命令结果 (lines = 保留的那几行).

    为什么要转发而不是重定向到文件:
        老板要能看实时进度, 不接受黑盒等待.管道的另一头是人.

    为什么要 PYTHONUNBUFFERED=1:
        子进程 (尤其是 python) 往管道写时默认是**块缓冲**, 不 flush 的话几百行输出会憋到进程
        退出才一起出来, "实时"就没了.由调用方在 env 里带上这个变量 (本库不擅自改别人的 env).

    注意:
        这里逐行读管道直到 EOF, 所以会一直阻塞到子进程结束 -- 这正是"前台跑一次"的语义.
        要"后台常驻不等它"的话用 启动(), 别用这个.
    """
    起点 = time.monotonic()
    保留: list[str] = []
    try:
        进程 = subprocess.Popen(  # noqa: S603
            argv,
            cwd=str(cwd),
            stdin=subprocess.DEVNULL,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            bufsize=1,  # 行缓冲: 子进程一 flush 我们就能收到
            env=env,
        )
    except (OSError, ValueError) as 错:
        return 命令结果(127, time.monotonic() - 起点, [], f"跑不起来: {错}")
    管道 = 进程.stdout
    if 管道 is not None:
        for 行 in 管道:
            文本行 = 行.rstrip("\n")
            print(文本行, flush=True)
            保留.append(文本行)
            if len(保留) > 尾部行数:
                保留.pop(0)
    码 = 进程.wait()
    return 命令结果(码, time.monotonic() - 起点, 保留)