Files
workbuddy_skills/session-mechanism/scripts/workspace_mirror.py
T

349 lines
17 KiB
Python
Raw Normal View History

# -*- coding: utf-8 -*-
"""workspace_mirror —— **工作区副本的同步/漂移自检**(2026-10-04 建)
一句话:**全局技能改了,工作区副本怎么跟上、怎么证明没残留。**
🔴 为什么必须有这个工具(不是"又写个脚本")
· 副本是**整目录拷贝**出去的 ⇒ 全局删掉/新增的文件,副本**不知道**,会静默留着
· 副本的 `roots.env` 是**工作区专属**的 ⇒ ⛔ 不能盲目整目录覆盖(会把工作区指向冲回别人)
· 于是每次同步都变成"人肉算差异+人肉挑例外" ⇒ **迟早漏**(实测已漏过:mut_run.py 差点没补上)
⇒ 正确解法:**让"例外的例外"由工具自己知道**,并且**能自证没漂移**。
用法(三条,都只读或幂等)
1) 查漂移(⛔ 纯只读,⛔ 不写任何文件)—— 用它当体检项:
python workspace_mirror.py --check <工作区副本目录>
2) 同步(会先备份 ⛔ 备份落在 <WS>/交付物/,⛔ 绝不留在包内):
python workspace_mirror.py --sync <工作区副本目录>
3) 只列出该保留什么(自测要断言这个表):
python workspace_mirror.py --list-keep
🔴 三条白名单(**有意保留副本自己的东西,不是残留**)
1. `roots.env` —— 工作区专属键。被覆盖 ⇒ 副本会去读写**另一个仓库**
(`setdefault` 兜底:宿主 env 没设时注入本文件的值)
2. `副本使用说明.md` —— 副本独有的口径文档
3. `install.log` —— 副本自己的安装历史(追加式)
⚠️ 白名单是**按文件名**判的 ⇒ 换副本(别的文件)要改 `KEEP`,
别靠"我记得跳过哪个"—— 那正是会漏的地方。
"""
import argparse
import hashlib
import io
import os
import shutil
import sys
import time
from pathlib import Path
try:
sys.stdout.reconfigure(encoding="utf-8", errors="replace")
except Exception:
pass
# 🔴 副本里"这个包自己的"目录,永远不算漂移
SELF_DIRS = {"__pycache__", ".git", ".pytest_cache"}
SELF_SUFFIX = (".pyc", ".pyo", ".tmp", ".swp", ".mirtmp")
# 🔴🔴 身份标记:`do_sync` 靠它判断"这像不像技能副本"。⛔ **绝不能**把它当残留删掉 ——
# 删了副本就**再也同步不了**(每次都被判"不像副本"直接退出,rc=2),
# 而用户看到的现象只是"工具忽然不干活了",极难往这儿想。
# (2026-10-04 实测踩到:夹具 src 侧没放 SKILL.md ⇒ sync 把它当残留删了 ⇒ 后续全卡死。)
# ⚠️ 它**故意不在** KEEP 里:KEEP 是"工作区专属、不可被全局覆盖"的内容;
# 而 SKILL.md 应当**跟着全局走**(它是版本的一部分)⇒ 只需在残留判定里豁免。
IDENT_MARK = "SKILL.md"
NEVER_DELETE = (IDENT_MARK,) # ⛔ 算残留时跳过 ⇒ 删了副本就废了
KEEP = ("roots.env", "副本使用说明.md", "install.log")
# 备份目录名前缀(`_find_in_prior_backups` 靠它认出"这是历史备份" ⇒ ⛔ 改名要同步改那里)
BK_PREFIX = "session-mechanism-副本备份-"
GLOBAL_PKG = Path(__file__).resolve().parent.parent # 本包根
def _files(root: Path, keep: bool = True):
"""列相对路径 → 绝对路径。⛔ **不用 glob**(P0-49:`**` 不穿透点目录,会零命中)。"""
out = {}
root = Path(root)
for dp, dns, fns in os.walk(str(root)):
dns[:] = [d for d in dns if d not in SELF_DIRS]
for fn in fns:
if fn.endswith(SELF_SUFFIX):
continue
rel = (Path(dp) / fn).relative_to(root).as_posix()
if keep and rel in KEEP:
continue
out[rel] = Path(dp) / fn
return out
def _md5(p: Path) -> str:
return hashlib.md5(p.read_bytes()).hexdigest()
def _keep_state(copy_dir: Path, backup_dir=None):
"""覆盖/删除**前**抓白名单文件的现状,供 ④ 兜底写回。
🔴 2026-10-04 修(变异实测发现的死代码):早先只从 `copy_dir` 现抓 ⇒
**文件一旦已经丢了就抓不到** ⇒ 兜底层永远救不了"白名单已丢"这个
**最需要救**的场景(现抓时它已经不在了)。
⇒ 加 `backup_dir` 兜底源:上一次备份里若有,一并收进来。
"""
st = {}
for k in KEEP:
p = copy_dir / k
if p.exists():
st[k] = p.read_bytes()
elif backup_dir:
b = Path(backup_dir) / k
if b.exists():
st[k] = b.read_bytes()
return st
def _find_in_prior_backups(keep_lost, bk_root: Path):
"""🔴🔴 2026-10-04 修(**上一版修法的病根**):在**历次备份**里找已丢的白名单。
⚠️ **原写法是死代码,判据因此报红**(`selftest` 那条「有备份时能捞回」FAIL 1):
原实现把**本次刚建的备份**当兜底源 ⇒ 但 `copytree` 发生在**丢失之后**
⇒ 那份备份里**必然也没有**这个文件 ⇒ 兜底永远捞不到。
⇒ 正解=翻**先前**留下的备份,按时间**从新到旧**找第一份命中的。
⛔⛔ **绝不拿全局那份顶替** —— 全局 `roots.env` 指向**别的仓库**,
填进去 = 把"副本悄悄指错仓库"变成"副本明确指错仓库",⛔ 更坏。
"""
found = {}
if not bk_root.is_dir():
return found
# 备份目录名带秒级时间戳 ⇒ 字典序即时间序(新→旧)
cands = sorted((d for d in bk_root.iterdir()
if d.is_dir() and d.name.startswith(BK_PREFIX)),
key=lambda d: d.name, reverse=True)
for d in cands:
for k in keep_lost:
if k in found:
continue
f = d / k
if f.is_file():
found[k] = f.read_bytes()
print(" ↺ 白名单兜底源=**历史备份** %s" % d.name)
if len(found) == len(keep_lost):
break
return found
def diff(gpkg: Path, copy_dir: Path):
g, c = _files(gpkg), _files(copy_dir)
# 🔴 `NEVER_DELETE`(身份标记 SKILL.md)豁免:⛔ 删了副本就再也同步不了
only_c = sorted(x for x in (set(c) - set(g)) if x not in NEVER_DELETE) # 真残留
only_g = sorted(set(g) - set(c)) # 待补
changed = sorted(k for k in set(g) & set(c) if _md5(g[k]) != _md5(c[k]))
return g, c, only_c, only_g, changed
def do_check(gpkg: Path, copy_dir: Path) -> int:
g, c, only_c, only_g, changed = diff(gpkg, copy_dir)
print("=== 副本漂移体检(⛔ 只读,⛔ 不写任何文件)===")
print("全局 %s |副本 %s" % (gpkg, copy_dir))
for k in only_c:
print(" ⛔ 残留 %-48s %d B" % (k, c[k].stat().st_size))
for k in only_g:
print(" + 待补 %-48s %d B" % (k, g[k].stat().st_size))
for k in changed:
print(" ✎ 待更新 %-48s 全局 %d B / 副本 %d B" % (k, g[k].stat().st_size, c[k].stat().st_size))
drift = len(only_c) + len(only_g) + len(changed)
print("\n漂移合计:%d(残留 %d |待补 %d |待更新 %d |白名单 %s)"
% (drift, len(only_c), len(only_g), len(changed), ", ".join(KEEP)))
if drift:
print("⇒ 有漂移,跑 `--sync %s`" % copy_dir)
else:
print("✅ 无漂移")
return 1 if drift else 0
def _ws_guess(copy_dir: Path) -> Path:
"""备份落点:`<工作区>/交付物/`(⛔ 绝不放包内 —— t_pkg_hygiene 会判红)。
🔴 判据:**副本自己就在 `.workbuddy/skills/` 下**才往上找(这才是真副本);
找不到(自建夹具/自建目录)⇒ 落**副本的父目录**。
⚠️ 早先无条件往上找 `.workbuddy` ⇒ 夹具落在 `tmp/` 下时,备份会被
写进**真工作区的 `交付物/`**(自测污染真工作区,今天实测踩到)。
"""
p = copy_dir
for up in range(1, 5):
p = p.parent
if p.name == ".workbuddy" and p.parent.name:
return p.parent # …/<工作区>/.workbuddy/skills/<包> ⇒ 落 <工作区>
return copy_dir.parent
def _latest_backup(copy_dir: Path):
"""副本**最近一次**备份(⛔ 不存在返回 None)。
🔴 为什么要它:判「白名单是丢了」还是「**从来没有过**」,
光看副本现状**分不出来** —— 只能看副本自己**历史备份里有没有**。
"""
d = _ws_guess(copy_dir) / "交付物"
if not d.is_dir():
return None
cands = []
for e in d.iterdir():
if e.is_dir() and e.name.startswith("session-mechanism-副本备份-"):
cands.append(e)
if not cands:
return None
# 名字尾缀 `_2`/`_3` 也要能比:去掉后缀按字典序(时间戳定长)
return sorted(cands, key=lambda p: p.name.replace("_2", "").replace("_3", ""))[-1]
def do_sync(gpkg: Path, copy_dir: Path) -> int:
if not (copy_dir / IDENT_MARK).exists():
print("⛔ %s 不像技能副本(没有 SKILL.md)⇒ 不动" % copy_dir)
return 2
# 🔴 2026-10-04 修:白名单**缺失**也算漂移。
# `diff()` 刻意排除 KEEP(免得每次都报它们)⇒ 副本的 `roots.env` 没了
# ⇒ 差异表里**一个字段都不动** ⇒ 早先直接走"无漂移,幂等返回",
# 白名单兜底层**永远没机会跑**。而这恰恰是最该救的场景:
# `roots.env` 丢了 ⇒ 副本按位置推导 ⇒ 去读写**别的仓库**(静默指错)。
# ⇒ 这里单独查一遍。
# 🔴🔴 修第二版:「**全局侧有**」⛔ 不等于「副本**该**有」。
# 实测踩到:全新副本本来就没有 `install.log`(从未安装过)⇒ 却被判成
# "白名单已丢" ⇒ 每次 sync 都刷一条假告警,**假告警多了就没人看了**。
# ⇒ 判据收紧成「**副本自己的历史备份里有过**」=「曾经有过,现在没了」=真丢。
# 副作用(可接受):真丢且**从未备份过**时只告警不阻断,并提示人工恢复。
_hist = _latest_backup(copy_dir)
keep_lost = []
for k in KEEP:
if (copy_dir / k).exists():
continue
if _hist is not None and (_hist / k).exists():
keep_lost.append(k) # 副本自己有过 ⇒ 确定是丢
keep_state = _keep_state(copy_dir)
_, _, only_c, only_g, changed = diff(gpkg, copy_dir)
if keep_lost:
print("⛔ **白名单缺失**(副本历史里有过,现在没了):%s" % "、".join(keep_lost)
+ ("(⛔ `roots.env` 丢 ⇒ 副本会按位置推导去读写**别的仓库**)"
if "roots.env" in keep_lost else ""))
if not (only_c or only_g or changed or keep_lost):
print("✅ 无漂移,⛔ 什么也不做(幂等)")
return 0
# 🔴 破坏前预检:⛔ 先确认每个待写源文件都真能读出来。
# 理由:②③ 是破坏性动作,读到一半炸掉 ⇒ 副本停在**半同步态**,
# 那是比"不同步"更难查的状态(2026-10-04 实测踩到一次 KeyError 崩在半路)。
gpre = _files(gpkg)
for k in sorted(gpre):
try:
gpre[k].read_bytes()
except Exception as e:
print("⛔ 源文件读不出 ⇒ 拒绝动手:%s(%s)" % (k, e))
return 2
# ① 备份(⛔ 落在 <WS>/交付物/,⛔ 绝不放包内 —— t_pkg_hygiene 会判红)
ws_guess = _ws_guess(copy_dir)
# 🔴 备份目录名**秒级**会撞:同一秒跑第二次 `--sync` ⇒ `copytree` 抛
# `FileExistsError`(2026-10-04 自测撞到)。⇒ 加 `_2`/`_3` 后缀避让。
stamp = time.strftime("%Y%m%d-%H%M%S")
bk = ws_guess / "交付物" / (BK_PREFIX + stamp)
_n = 2
while bk.exists():
bk = ws_guess / "交付物" / ("%s%s_%d" % (BK_PREFIX, stamp, _n))
_n += 1
bk.parent.mkdir(parents=True, exist_ok=True)
if copy_dir.exists():
shutil.copytree(str(copy_dir), str(bk), ignore=shutil.ignore_patterns(*SELF_DIRS))
print("① 已备份 ⇒ %s" % bk)
# 🔴🔴 2026-10-04 修(**这才是能救回来的那版**):兜底源必须是**别的**备份。
# ⚠️ 原来这段只认 `bk`(**本次刚建的**那份)⇒ `copytree` 发生在**丢失之后**
# ⇒ 那份备份里必然也没有 ⇒ 「有备份时能捞回」这条判据**恒红**
# (实测:全局 selftest `PASS 77 / FAIL 1`,独立复现 rc=0 但文件没回来)。
# ⇒ 正解=扫 `交付物/` 下**历次**备份,按时间从新到旧找第一份命中的。
# ⛔⛔ 但**绝不**拿全局那份来填 —— 全局 `roots.env` 指向**别的仓库**,
# 填进去 = 把「副本悄悄指错仓库」变成「副本明确指错仓库」,⛔ 更坏。
# ⇒ 只能从**副本自己的历史备份**里捞;捞不到就**明说**,⛔ 不静默、⛔ 不代填。
if keep_lost:
_need = [k for k in keep_lost if k not in keep_state]
if _need:
keep_state.update(_find_in_prior_backups(_need, bk.parent))
for k in keep_lost:
if k not in keep_state:
print("⛔⛔ 白名单 **%s 已丢且所有历史备份里都没有** ⇒ 无法自动恢复。" % k)
print(" ⛔ **绝不拿全局那份顶替**(全局指向别的仓库)。")
print(" ⇒ 请人工恢复 %s ——⛔ 不代填。" % (copy_dir / k))
# ② 删残留(副本有、全局已删)
for k in only_c:
_files(copy_dir)[k].unlink()
print("② 删残留 %s" % k)
# ③ 覆盖 + 补入
# 🔴 只遍历**全局有的**。⛔ 残留(只在副本)已在 ② 删掉,⛔ 绝不能进这个循环
# —— 它在 `g` 里不存在,`g[k]` 会 KeyError(2026-10-04 实测踩到:sync 崩在半路)
g = _files(gpkg)
n_ov = n_new = 0
for k in sorted(g):
if k in KEEP:
continue
src = g[k]
dst = copy_dir / k
dst.parent.mkdir(parents=True, exist_ok=True)
data = src.read_bytes()
if dst.exists() and dst.read_bytes() == data:
continue
tmp = Path(str(dst) + ".mirtmp")
io.open(str(tmp), "wb").write(data)
os.replace(str(tmp), str(dst)) # 原子替换
if k in only_g:
n_new += 1
else:
n_ov += 1
print("③ 覆盖 %d |补入 %d" % (n_ov, n_new))
# ④ 白名单兜底:**核对+缺则补回**(⛔ 这是"兜底层",不是主防线)
# 主防线是 ③ 循环里那句 `if k in KEEP: continue`(判据锚在那句上)。
# 这层的作用:万一 ③ 的守卫被改坏/将来有人加新白名单忘了守卫,
# 这里仍能把副本自己的工作区指向**捞回来** ⇒ 失败模式从"静默指错仓库"
# 降级成"这次跑完就修好"。
# ⚠️ 故意的设计:**恒真无害**(内容没被动就零动作),⛔ 不做"逐字重写",
# 否则会给 install.log / roots.env 换来无谓的 mtime 抖动。
for k, data in keep_state.items():
p = copy_dir / k
if p.exists() and p.read_bytes() == data:
continue
tmp = Path(str(p) + ".mirtmp")
tmp.parent.mkdir(parents=True, exist_ok=True)
io.open(str(tmp), "wb").write(data)
os.replace(str(tmp), str(p)) # 缺文件时 p.parent 可能还没建
print("④ 白名单兜底恢复 %s" % k)
# ⑤ 复核(⛔ 复核工具自身出错 ⇒ 结论作废,必须让它抛而不是吞)
g2, c2, only_c2, only_g2, changed2 = diff(gpkg, copy_dir)
bad = [k for k in g2 if k not in c2 or _md5(g2[k]) != _md5(c2[k])]
print("⑤ 复核:应共享 %d,逐字不一致 = %d %s" % (len(g2), len(bad), bad if bad else "✅"))
print(" 副本额外 = %s" % (sorted(set(c2) - set(g2)) or "(无)✅"))
if bad or only_c2:
return 1
print("\n✅ 同步完成。⛔ 还没完 —— **必须**接着跑:\n"
" 1. 副本 selftest(须 FAIL 0):python <副本>/scripts/selftest.py\n"
" 2. env_check(看 deploy 段):python <副本>/scripts/hooks/_env.py <工作区>")
return 0
def main():
ap = argparse.ArgumentParser(description="工作区副本同步/漂移自检")
ap.add_argument("--check", metavar="副本目录")
ap.add_argument("--sync", metavar="副本目录")
ap.add_argument("--global", dest="gpkg", default=str(GLOBAL_PKG), help="全局包根(默认本包)")
ap.add_argument("--list-keep", action="store_true")
a = ap.parse_args()
if a.list_keep:
for k in KEEP:
print(k)
return 0
gpkg = Path(a.gpkg).resolve()
if a.check:
return do_check(gpkg, Path(a.check).resolve())
if a.sync:
return do_sync(gpkg, Path(a.sync).resolve())
ap.print_help()
return 0
if __name__ == "__main__":
sys.exit(main())