zhifu-platform/scripts/verify_host_channel_worker.py

import asyncio
import json
import signal
import subprocess
import sys
import time

from verify_channel_worker import MAX_SECONDS, ROOT, inspect_environment


def main():
    enabled, pending, initial = asyncio.run(inspect_environment())
    if enabled or pending:
        raise RuntimeError("生命周期验收要求当前没有启用渠道和等待投递记录")
    windows = sys.platform == "win32"
    started = time.monotonic()
    # 独立进程组仅接收本次测试的停止信号；Windows 需要从终端运行以共享控制台。
    process = subprocess.Popen(
        [sys.executable, "-m", "zhifu.modules.platform.delivery"],
        cwd=ROOT,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        creationflags=subprocess.CREATE_NEW_PROCESS_GROUP if windows else 0,
    )
    try:
        while time.monotonic() - started < MAX_SECONDS:
            if process.poll() is not None:
                stdout, stderr = process.communicate(timeout=5)
                raise RuntimeError(f"渠道工作进程提前退出：{stdout}\n{stderr}")
            _, _, current = asyncio.run(inspect_environment())
            if current - initial:
                break
            time.sleep(0.25)
        else:
            raise RuntimeError("渠道工作进程未在 25 秒期限内登记心跳")
        assert len(current - initial) == 1, "渠道工作进程应只登记一份心跳"
        # 完成一次空队列轮询后发送真实控制台事件或 POSIX 信号，验证完整清理过程。
        time.sleep(1)
        process.send_signal(signal.CTRL_BREAK_EVENT if windows else signal.SIGTERM)
        stdout, stderr = process.communicate(timeout=10)
        assert process.returncode == 0, (process.returncode, stdout, stderr)
        assert "worker=channel-delivery stopped" in stdout, stdout
        _, _, after = asyncio.run(inspect_environment())
        assert after == initial, "正常停止后应立即清除本进程心跳"
        report = {
            "role": "channel-delivery",
            "platform": sys.platform,
            "heartbeat_count": 1,
            "signal": "CTRL_BREAK_EVENT" if windows else "SIGTERM",
            "exit_code": process.returncode,
            "heartbeat_removed": True,
            "seconds": round(time.monotonic() - started, 2),
            "generated_model_requests": 0,
            "external_delivery_requests": 0,
        }
        target = ROOT / ".work/channel-worker-host-lifecycle.json"
        target.parent.mkdir(exist_ok=True)
        target.write_text(json.dumps(report, indent=2), encoding="utf-8")
        print(json.dumps(report))
    finally:
        if process.poll() is None:
            process.kill()
            process.communicate(timeout=5)


if __name__ == "__main__":
    main()