Skip to content
79 changes: 60 additions & 19 deletions skills/zcode-tokenspeed/scripts/apply_after_exit.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
退出码约定(与 zcode_patcher.py 一致):0=成功、1=有项目失败/等待超时、2=预检失败(ZCode 仍在运行)。
"""

import os
import subprocess
import sys
import time
Expand All @@ -36,37 +37,60 @@
PATCH_TIMEOUT = 600


def _import_patcher():
"""导入主脚本模块;sys.path 只在缺失时插入。

看护轮询循环每 POLL_SEC 秒调一次 zcode_running() → 本函数被反复执行,
不能每次都 insert(sys.path 会无界增长——24h 上限约 2.9 万个重复项)。
"""
if str(HERE) not in sys.path:
sys.path.insert(0, str(HERE))
import zcode_patcher as zp
return zp


def log(msg: str) -> None:
with open(LOG, "a", encoding="utf-8") as f:
f.write(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}\n")


def zcode_running() -> bool:
# 用 bytes 检索,不走 text 解码:tasklist 输出是 GBK,而本机 Python 为 UTF-8
# 模式,text=True 会在读线程里抛 UnicodeDecodeError → stdout 变空 → 误判「已退出」。
# 同理不带 /FI:从 Git Bash/MSYS 环境启动时 "/FI" 会被路径转换破坏。
# 带 timeout:tasklist 在系统繁忙/WMI 打嗝时可能挂住,看护无人值守,不能无限等。
# 超时视为「仍在运行」(保守:宁可多等一轮,也不要在 ZCode 还锁着 asar 时动手)。
# 复用主脚本 zcode_patcher.zcode_running() 的跨平台检测(Windows: tasklist 检索
# ZCode.exe;POSIX: pgrep -f ZCode)。看护的判定必须与补丁预检同源,否则会出现
# 「看护以为退干净了、动手时又被预检拒绝」的永久错位。
# 历史问题:这里曾直接跑 ["tasklist"] 并检索 b"ZCode.exe" —— macOS/Linux 上没有
# tasklist,FileNotFoundError 未被捕获(except 只接 TimeoutExpired),看护在第一次
# 轮询就崩溃;启动方又把 stderr 定向到 DEVNULL,崩溃完全无声,表现为
# 「等退出 0 次、完成 0 次」、补丁永远写不进去。
try:
out = subprocess.run(["tasklist"], capture_output=True,
timeout=TASKLIST_TIMEOUT,
**no_window_kwargs()).stdout or b""
except subprocess.TimeoutExpired:
log(f"tasklist 超时({TASKLIST_TIMEOUT}s),本轮按「仍在运行」处理")
return _import_patcher().zcode_running()
except Exception as e:
log(f"运行检测失败: {type(e).__name__}: {e},本轮按「仍在运行」处理")
return True
return b"ZCode.exe" in out


def _find_exe(res: Path) -> Path | None:
"""从 asar 所在目录(<安装根>/resources)推出各平台的 ZCode 主程序路径。"""
root = res.parent # Windows: 安装根;macOS: ZCode.app/Contents
cands: list[Path] = [root / "ZCode.exe"] # Windows
if sys.platform == "darwin":
cands.insert(0, root / "MacOS" / "ZCode") # macOS .app 包
else:
cands += [root / "zcode", root / "ZCode"] # Linux 常见命名
for c in cands:
if c.is_file():
return c
return None


def resolve_install() -> tuple[Path | None, Path | None]:
"""复用主脚本的跨平台探测拿到 (asar 目录, ZCode 可执行文件)。"""
try:
sys.path.insert(0, str(HERE))
import zcode_patcher as zp
zp = _import_patcher()
for cjs in zp.discover():
res = cjs.parent.parent
if (res / "app.asar").is_file():
exe = res.parent / "ZCode.exe"
return res, (exe if exe.is_file() else None)
return res, _find_exe(res)
except Exception as e:
log(f"安装探测失败: {type(e).__name__}: {e}")
return None, None
Expand Down Expand Up @@ -143,15 +167,25 @@ def main() -> int:

res, exe = resolve_install()
if exe is None:
log("未找到 ZCode.exe(可用主脚本探测确认安装位置),请手动启动")
log("未找到 ZCode 主程序(Windows: ZCode.exe / macOS: .app 包内 MacOS/ZCode),请手动启动")
return 1
log(f"处理完成(失败 {failed} 项),重启 ZCode")
if zcode_running():
log("检测到 ZCode 已再次运行,跳过重启")
return 0
try:
subprocess.Popen([str(exe)], cwd=str(exe.parent),
creationflags=0x00000008) # DETACHED_PROCESS
if os.name == "nt":
subprocess.Popen([str(exe)], cwd=str(exe.parent),
creationflags=0x00000008) # DETACHED_PROCESS——保持上游原样,Windows 行为零变化
elif sys.platform == "darwin" and res is not None and res.parent.parent.suffix == ".app":
# macOS 经 Launch Services 打开 .app 包。注意 creationflags 是 Windows 专属
# 参数,POSIX 上传入会直接抛 ValueError,所以必须按平台分流。
subprocess.Popen(["open", str(res.parent.parent)],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
**no_window_kwargs())
else:
subprocess.Popen([str(exe)], cwd=str(exe.parent), start_new_session=True,
**no_window_kwargs())
except Exception as e:
log(f"重启 ZCode 失败(请手动启动): {e}")
return 1
Expand All @@ -160,4 +194,11 @@ def main() -> int:


if __name__ == "__main__":
sys.exit(main())
try:
sys.exit(main())
except Exception:
# 看护无人值守:崩溃必须留痕。此前 stderr 被 start_watchdog 定向到 DEVNULL,
# 崩溃无声 —— 这正是 macOS 上「等退出 0 次」长期没被发现的原因。
import traceback
log("看护异常退出:\n" + traceback.format_exc())
sys.exit(1)
12 changes: 8 additions & 4 deletions skills/zcode-tokenspeed/scripts/sync.py
Original file line number Diff line number Diff line change
Expand Up @@ -377,10 +377,14 @@ def run_patcher(args, revert: bool) -> str:
def start_watchdog(wanted: dict) -> None:
"""启动退出后看护:等 ZCode 退出 → 应用重打包级补丁。"""
args = [f"--want={k}={'on' if v else 'off'}" for k, v in wanted.items()]
flags = (DETACHED_PROCESS | CREATE_NO_WINDOW) if os.name == "nt" else 0
subprocess.Popen([sys.executable, str(WATCHDOG), *args], cwd=str(HERE),
creationflags=flags, stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL)
kw = {"cwd": str(HERE), "stdout": subprocess.DEVNULL, "stderr": subprocess.DEVNULL}
if os.name == "nt":
kw["creationflags"] = DETACHED_PROCESS | CREATE_NO_WINDOW
else:
# POSIX:自成会话组(与 spawn_detached 同一做法),否则看护留在钩子的
# 进程组里,应用退出/清理时可能被整组信号连带杀掉。
kw["start_new_session"] = True
subprocess.Popen([sys.executable, str(WATCHDOG), *args], **kw) # no-window-ok: kw 已按平台携带 DETACHED_PROCESS|CREATE_NO_WINDOW(Windows)/ start_new_session(POSIX)
log(f"已启动退出后看护: {' '.join(args)}")


Expand Down
70 changes: 59 additions & 11 deletions skills/zcode-tokenspeed/scripts/zcode_patcher.py
Original file line number Diff line number Diff line change
Expand Up @@ -121,7 +121,7 @@
import sys
import threading
import time
from pathlib import Path
from pathlib import Path, PureWindowsPath

try: # 控制台编码/窗口安全网(见 _console.py 的说明)
from _console import bad_mark, no_window_kwargs, ok_mark, safe_stdio
Expand Down Expand Up @@ -457,9 +457,13 @@ def _from_running_processes(found: list[Path]) -> None:
out = out.decode(errors="replace")
for line in out.splitlines():
line = line.strip()
# ZCode.exe / ZCode Skin Manager 等都指向安装根目录
if line.lower().endswith(".exe") and "zcode" in _norm(Path(line).name):
found.append(Path(line).parent)
# ZCode.exe / ZCode Skin Manager 等都指向安装根目录。
# 存 PureWindowsPath 而非 Path:单测把 os.name 伪造成 "nt"(zp.os 即全局 os
# 模块单例),此时 Path() 在 macOS/Linux 上会尝试构造 WindowsPath 并抛
# NotImplementedError;discover() 消费时统一 Path(root) 转回具体路径
# (真实 Windows 运行时才转换,PureWindowsPath 属性与 WindowsPath 一致)。
if line.lower().endswith(".exe") and "zcode" in _norm(PureWindowsPath(line).name):
found.append(PureWindowsPath(line).parent)


def _from_registry(found: list[Path]) -> None:
Expand Down Expand Up @@ -541,7 +545,7 @@ def discover() -> list[Path]:
if VERBOSE:
print(f"[·] 探测器 {probe.__name__} 命中 {len(roots) - before} 个候选目录")
seen, result = set(), []
for root in roots:
for root in map(Path, roots): # 探针可能存 PureWindowsPath(见 _from_running_processes)
cjs = root / "resources" / "glm" / "zcode.cjs"
try:
key = cjs.resolve()
Expand Down Expand Up @@ -1762,16 +1766,22 @@ def __enter__(self):
raise TimeoutError(
f"等待 asar 写入锁超时({self.timeout:.0f}s):{self.path.name}\n"
f" 本进程内已有注入流程正在写入,请稍后重试")
# ② 再拿跨进程文件锁
# ② 再拿跨进程文件锁(Windows msvcrt / POSIX fcntl —— 插件在 macOS/Linux 也会注入客户端)
try:
import msvcrt # Windows 专用;本工具只在 Windows 注入客户端
if os.name == "nt":
import msvcrt # Windows 专用
else:
import fcntl # POSIX 跨进程文件锁
deadline = time.time() + self.timeout
self.path.parent.mkdir(parents=True, exist_ok=True)
while True:
try:
fh = open(self.path, "a+b")
try:
msvcrt.locking(fh.fileno(), msvcrt.LK_NBLCK, 1)
if os.name == "nt":
msvcrt.locking(fh.fileno(), msvcrt.LK_NBLCK, 1)
else:
fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
except OSError:
fh.close()
raise
Expand Down Expand Up @@ -1805,9 +1815,13 @@ def _release_thread(self):
def __exit__(self, *exc):
if self._fh is not None:
try:
import msvcrt
self._fh.seek(0)
msvcrt.locking(self._fh.fileno(), msvcrt.LK_UNLCK, 1)
if os.name == "nt":
import msvcrt
msvcrt.locking(self._fh.fileno(), msvcrt.LK_UNLCK, 1)
else:
import fcntl
fcntl.flock(self._fh.fileno(), fcntl.LOCK_UN)
except OSError:
pass
try:
Expand Down Expand Up @@ -2709,6 +2723,37 @@ def _resolve_asars(target: str | None) -> list[Path]:
return asars


def _pid_commandline(pid: str) -> "str | None":
"""POSIX:返回进程完整命令行;进程已消失/查询失败返回 None。"""
try:
r = subprocess.run(["ps", "-o", "command=", "-p", pid],
capture_output=True, timeout=5, **no_window_kwargs())
if r.returncode != 0:
return None
return r.stdout.decode(errors="replace").strip()
except Exception:
return None


def _pgrep_hits_are_alive(pids: "list[str]", cmdline_of=None) -> bool:
"""对 pgrep -f ZCode 命中的 PID 逐个核对命令行,排除 crashpad 后判应用是否存活。

★ 为什么必须逐个核对(2026-09-24 macOS 实测):ZCode 每次退出都会泄漏一个
chrome_crashpad_handler(ppid=1 的崩溃报告进程,命令行仍含 ZCode.app 路径),
pgrep -f ZCode 因此永远非零——退出后看护等不到「退出」,重打包级补丁永远写不
进去。本机曾堆积 3.9.1 / 3.10.1 时代的多个泄漏进程,应用本体早已不在。应用真正
存活时主进程 / ZCode Helper 必然在列且非 crashpad,不受此排除影响。
cmdline_of 返回 None(进程在 pgrep 与 ps 之间消失)按已退出处理。
"""
cmdline_of = cmdline_of or _pid_commandline
for pid in pids:
cmd = cmdline_of(pid)
if cmd is None or "chrome_crashpad_handler" in cmd:
continue # 已消失的 / crashpad 僵尸:不代表应用存活
return True
return False


def zcode_running() -> bool:
"""ZCode 是否在运行(打补丁前预检:运行中会锁住 app.asar,配置也可能被回写覆盖)。"""
try:
Expand All @@ -2718,7 +2763,10 @@ def zcode_running() -> bool:
**no_window_kwargs()).stdout or b""
return b"ZCode.exe" in out
r = subprocess.run(["pgrep", "-f", "ZCode"], capture_output=True, timeout=10) # no-window-ok: 只在 POSIX 分支执行
return r.returncode == 0
if r.returncode != 0:
return False
pids = r.stdout.decode(errors="replace").split()
return _pgrep_hits_are_alive(pids)
except Exception:
return False

Expand Down
104 changes: 103 additions & 1 deletion tests/test_patcher.py
Original file line number Diff line number Diff line change
Expand Up @@ -567,7 +567,11 @@ def test_patch_after_upgrade_archives_stale_backup(self):
upgraded = make_cjs(zp.ANCHORS["3.9.2"], prefix="/*upgraded-kernel*/")
self.cjs.write_bytes(upgraded)
self.assertTrue(self.patch())
stale = list(self.tmp.glob("zcode.cjs.bak.stale-*"))
# glob 必须排除 .meta.json:_archive_backup 会把 bak 与 bak.meta.json 一起改名成
# stale-*,两者都命中该模式;目录枚举顺序(APFS 按名字哈希)随时间戳变化,
# stale[0] 取到谁是随机的——曾表现为同代码 10 跑 4~6 挂的 flaky。
stale = [p for p in self.tmp.glob("zcode.cjs.bak.stale-*")
if not p.name.endswith(".meta.json")]
self.assertTrue(stale, "旧备份未归档")
meta = json.loads((self.tmp / "zcode.cjs.bak.meta.json").read_text(encoding="utf-8"))
self.assertEqual(meta["sha256"], zp._sha256(upgraded), "备份不是升级后内核的原始副本")
Expand Down Expand Up @@ -1170,6 +1174,104 @@ def test_non_zcode_paths_are_ignored(self):
self.assertEqual(found, [])


@unittest.skipIf(os.name == "nt", "POSIX 分支(Windows 走 tasklist,不经过这段)")
class TestZcodeRunningPosix(unittest.TestCase):
"""zcode_running 的 POSIX 分支:泄漏的 crashpad 僵尸必须被排除。

macOS 实测(2026-09-24):ZCode 每次退出都泄漏一个 chrome_crashpad_handler
(ppid=1、命令行含 ZCode.app 路径),pgrep -f ZCode 永远非零,退出后看护
等不到「退出」、重打包级补丁永远写不进去——曾堆积 3.9.1/3.10.1 时代的进程。
"""

CRASHPAD = ("/Applications/ZCode.app/Contents/Frameworks/Electron Framework.framework"
"/Helpers/chrome_crashpad_handler --annotation=_productName=ZCode")
HELPER = ("/Applications/ZCode.app/Contents/Frameworks/ZCode Helper.app/Contents"
"/MacOS/ZCode Helper --type=gpu-process")

def setUp(self):
import zcode_patcher as zp
self.zp = zp
self._orig_run = zp.subprocess.run

def tearDown(self):
self.zp.subprocess.run = self._orig_run

def _install_fake(self, pgrep_rc, pgrep_out, cmdlines):
"""cmdlines: pid(str) -> 命令行字符串;None 表示 ps 查不到(进程已消失)。"""
class _R:
def __init__(self, rc, out):
self.returncode = rc
self.stdout = out

def fake_run(args, **kw):
if args[0] == "pgrep":
return _R(pgrep_rc, pgrep_out)
cmd = cmdlines.get(args[-1], "")
return _R(0 if cmd is not None else 1, (cmd or "").encode())

self.zp.subprocess.run = fake_run

def test_no_hits_means_not_running(self):
self._install_fake(1, b"", {})
self.assertFalse(self.zp.zcode_running())

def test_only_crashpad_zombies_means_not_running(self):
"""应用已退出、只剩泄漏的 crashpad(跨版本堆积)——必须判「未运行」。"""
self._install_fake(0, b"111 222 333\n", {
"111": self.CRASHPAD + " --version=3.14.3",
"222": self.CRASHPAD + " --version=3.10.1",
"333": self.CRASHPAD + " --version=3.9.1",
})
self.assertFalse(self.zp.zcode_running())

def test_crashpad_plus_real_helper_means_running(self):
"""crashpad 泄漏与真实 helper 并存——必须判「在运行」(crashpad 不影响判定)。"""
self._install_fake(0, b"111 222\n", {"111": self.HELPER, "222": self.CRASHPAD})
self.assertTrue(self.zp.zcode_running())

def test_vanished_pid_counts_as_gone(self):
"""pgrep 命中后进程在 ps 之前消失:按已退出处理,不得抛错或误判存活。"""
self._install_fake(0, b"111\n", {"111": None})
self.assertFalse(self.zp.zcode_running())

def test_pid_commandline_reads_real_process(self):
"""_pid_commandline 对真实子进程可用(ps 集成,非 mock)。"""
p = subprocess.Popen(["sleep", "5"])
try:
cmd = self.zp._pid_commandline(str(p.pid))
self.assertIsNotNone(cmd)
self.assertIn("sleep", cmd)
finally:
p.terminate()
p.wait()

def test_hits_are_alive_with_injected_cmdline(self):
"""纯函数入口:真实 cmdline_of 回调 + crashpad/真进程混合。"""
alive = self.zp._pgrep_hits_are_alive(
["1", "2"], cmdline_of=lambda pid: self.CRASHPAD if pid == "1" else self.HELPER)
self.assertTrue(alive)
gone = self.zp._pgrep_hits_are_alive(
["1"], cmdline_of=lambda pid: self.CRASHPAD)
self.assertFalse(gone)


class TestApplyAfterExitPathGuard(unittest.TestCase):
"""apply_after_exit 的看护轮询每 3s 调一次 zcode_running()(最长 24h)。

曾经的写法是每次 sys.path.insert(0, HERE)——列表无界增长(24h 约 2.9 万个
重复项)。_import_patcher 必须只在缺失时插入;同时不得破坏 import 缓存。
"""

def test_repeated_imports_do_not_grow_sys_path(self):
import apply_after_exit as aae
zp = aae._import_patcher()
self.assertIs(zp, sys.modules["zcode_patcher"], "应当复用缓存的模块对象")
baseline = len(sys.path)
for _ in range(200): # 模拟 10 分钟轮询量级的调用
aae._import_patcher()
self.assertEqual(len(sys.path), baseline, "sys.path 无界增长")


class TestNoConsoleWindowFlags(unittest.TestCase):
"""每个会起 console 子进程的调用都必须带「别弹控制台窗口」的标志。

Expand Down