admin
19101acd65
- 按用户指示清空原有 25 技能内容，只提交 session-mechanism（57 文件） - 附 .gitignore（产物 + 本机凭据） - 令牌明文已脱敏（历史 .neodata_token 与 pitfalls 引用均不入库） - 本提交为孤儿提交（父提交为空），历史自此重新开始
350 lines
17 KiB
Python
350 lines
17 KiB
Python
# -*- 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())
|