Files
workbuddy_skills/session-mechanism/install.py
T

573 lines
26 KiB
Python
Raw Normal View History

#!/usr/bin/env python3
"""
session-mechanism · 一键配置(换机器只需跑这一个)
用法
python install.py --dry-run # 只打印将改什么(settings.json 的 diff),⛔ 不写盘
python install.py --apply # 真装:写 roots.env → 按声明表接线全局钩子 → 初始化工作区
python install.py --verify # 装完自检:每个钩子空载荷 rc=0 + collabd --where + selftest
python install.py --uninstall # 还原 settings.json(与装前备份逐字节相同)
python install.py --manifest # 重算 references/manifest.md 的逐文件表(改完包**必跑**)
设计要点(都由实测倒逼,⛔ 不要"优化"掉)
1. **自解析**:解释器一律 `sys.executable`;配置目录按 `CODEBUDDY_CONFIG_DIR` 推导。
⛔ 不硬编码 python 路径、⛔ 不硬编码盘符 —— 这正是「换机器必碎」的根因。
2. **声明表驱动**:本包 **10 条**接线收敛成一张 HOOKS 表(`--verify` 逐条空载荷复测)。
⛔ 表里**不含** `decision_bridge.py`(属 `ai1net-decision-laya` 另一条线)—— 它在 settings.json 里
另有 4 条、与本包同处一个 `hooks` 段 ⇒ 极易误读成"本包 14 条"(`--dry-run` 会打印被排除条数供目视核对)。
3. **幂等**:先删「本包自己的旧条目」再插;装两遍结果相同。
4. **可逆**:首次安装先留**原状**备份 `settings.json.bak-session-mechanism-orig`(⛔ 已存在不覆盖),
其后每次 `--apply` 另写**带微秒**的时间戳备份(防同秒同名互相覆盖);
`--uninstall` **优先**用 `-orig`,还原后做**逐字节**比对,不符即报错。
5. **外置根目录**:写 `<包根>/roots.env` —— 因为 `settings.json` 的 hook 条目**没有 env 字段**,
包内脚本搬一次就会静默指错(历史事故:台账写到别处、测试却全绿)。
6. **清单自算**:`--manifest` 只重写 `references/manifest.md` 的**表 + 计数行 + 重算时间**,
⛔ 不碰上面的「最近改动」散文 —— 那是人写的结论,脚本代笔会把它冲成流水账。
包一改就得跑一次,否则表里的字节/md5 立刻过期(本项目已因此误判过两次)。
"""
from __future__ import annotations
import argparse
import difflib
import hashlib
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import time
from pathlib import Path
PKG = Path(__file__).resolve().parent
ROOTS_ENV = PKG / "roots.env"
LOG = PKG / "install.log"
# 🔴 原状备份的**固定名**(首次安装、且当前 settings.json 无本包痕迹时才写;⛔ 已存在不覆盖)。
# 有了它,`--uninstall` 不需要"从一堆时间戳备份里猜哪份是原状"。
ORIG_BAK_NAME = "settings.json.bak-session-mechanism-orig"
BAK_PREFIX = "settings.json.bak-session-mechanism-"
# 🔴 判定"是不是本包的"一律**按脚本名**,⛔ 不用路径片段(`session-mechanism`)——
# 路径片段在"包被改名 / 拷到别的目录"时**认不出自己**(实测教训:§2 #4)。
# ⚠️ 这些脚本名**必须连旧落点一起认**(文档库 `07-scripts/` + 工作区 `.workbuddy/tools/`),
# 否则旧接线删不掉 ⇒ 同一件事挂两条钩子、同一秒各跑一次(`agent-operating-rules §10.2` 实测故障)。
# ⛔ `decision_bridge.py` 不在此列。
OWN_BASENAMES = (
"session-log-guard.py",
"lock-guard-hook.py",
"bash-output-guard.py",
"stop-dialog-guard.py",
"skill-load-guard.py",
"wb-result-hook.py",
)
# ── 声明表:本包负责的钩子(⛔ decision_bridge 不在此表内)───────────────────
# (事件, matcher, 包内脚本相对路径, 额外参数, 超时秒)
HOOKS: list[tuple[str, str | None, str, list[str], int]] = [
("PostToolUse", None, "scripts/hooks/session-log-guard.py", ["-S"], 10),
("PreToolUse", "^Bash$", "scripts/hooks/wb-result-hook.py", [], 30),
("PreToolUse", "Write|Edit", "scripts/hooks/lock-guard-hook.py", [], 10),
("PreToolUse", "Bash|Read", "scripts/hooks/bash-output-guard.py", [], 10),
("SessionEnd", None, "scripts/hooks/wb-result-hook.py", [], 10),
("SessionStart", "startup|resume", "scripts/hooks/lock-guard-hook.py", [], 10),
("UserPromptSubmit", None, "scripts/hooks/wb-result-hook.py", [], 20),
("UserPromptSubmit", None, "scripts/hooks/stop-dialog-guard.py", [], 10),
("UserPromptSubmit", None, "scripts/hooks/skill-load-guard.py", [], 10),
("UserPromptSubmit", None, "scripts/hooks/session-log-guard.py", ["-S"], 10),
]
def config_dir() -> Path:
v = os.environ.get("CODEBUDDY_CONFIG_DIR")
if v:
return Path(v)
return Path.home() / ".workbuddy"
def settings_path() -> Path:
return config_dir() / "settings.json"
def log(msg: str) -> None:
line = f"{time.strftime('%Y-%m-%d %H:%M:%S')} {msg}"
print(line)
try:
with open(LOG, "a", encoding="utf-8") as f:
f.write(line + "\n")
except Exception:
pass
def read_roots() -> dict[str, str]:
d: dict[str, str] = {}
if ROOTS_ENV.is_file():
for ln in ROOTS_ENV.read_text(encoding="utf-8").splitlines():
ln = ln.strip()
if ln and not ln.startswith("#") and "=" in ln:
k, v = ln.split("=", 1)
d[k.strip()] = v.strip()
return d
def detect_workspace(explicit: str | None) -> Path | None:
if explicit:
return Path(explicit).resolve()
for k in ("DSH_WS_ROOT", "COLLABD_WORKSPACE"):
if os.environ.get(k):
return Path(os.environ[k]).resolve()
prev = read_roots().get("DSH_WS_ROOT")
if prev and Path(prev).is_dir():
return Path(prev)
cwd = Path.cwd().resolve()
for cand in (cwd, *cwd.parents):
if (cand / ".workbuddy").is_dir() and (cand / "state.py").is_file():
return cand
return None
def detect_docs_root(explicit: str | None) -> Path | None:
if explicit:
return Path(explicit).resolve()
if os.environ.get("DSH_DOCS_ROOT"):
return Path(os.environ["DSH_DOCS_ROOT"]).resolve()
prev = read_roots().get("DSH_DOCS_ROOT")
if prev and Path(prev).is_dir():
return Path(prev)
# 启发式:文档库必须同时具备「05-交接单」与「07-scripts」两个标志目录
# ⚠️ 只看同名会误选到工作区里的 `dsh-server-docs/` 副本 ⇒ 两个条件都要满足。
seeds = [Path.cwd(), *Path.cwd().parents]
for k in ("DSH_CODE_REPO",):
if os.environ.get(k):
seeds.insert(0, Path(os.environ[k]))
for s in seeds[:6]:
for cand in (s / "dsh-server-docs", s.parent / "dsh-server-docs"):
if (cand / "05-交接单").is_dir() and (cand / "07-scripts").is_dir():
return cand.resolve()
return None
def py() -> str:
return sys.executable
def build_command(rel: str, extra: list[str]) -> str:
script = (PKG / rel).resolve()
parts = [f'"{py()}"', *[f'"{a}"' for a in extra], f'"{script}"']
return " ".join(parts)
def has_our_trace(text: str) -> bool:
"""**这条 hook 归不归本包管** —— 按**脚本名**判。
⚠️ 必须连旧落点一起认(文档库 `07-scripts/` + 工作区 `.workbuddy/tools/`),否则旧接线删不掉。
⚠️ ⛔ 不可拿它当"这份文件是否已装过本包"的判据 —— 真机 settings.json **本来就有**同名脚本
挂在 07-scripts 上(实测:6 个脚本名全 True)⇒ 会**永远写不出 `-orig`**、`--uninstall` **永远拒做**。
那个判据用 `points_into_pkg()`。"""
return any(b in text for b in OWN_BASENAMES)
def points_into_pkg(text: str) -> bool:
"""**这份文件是否已被本包接管** —— 有没有钩子**指向本包目录**。
🔴 路径**运行时从 `__file__` 推导**(`PKG`)⇒ 包被改名 / 拷到别的目录都认得出自己。
这正是 §3.1b 要修的病:⛔ 旧写法硬编码路径片段 `session-mechanism`。
旧落点(07-scripts / tools)**不算**"已装本包"。"""
return PKG.as_posix().lower() in text.replace("\\", "/").lower()
def is_ours(entry: dict) -> bool:
for h in entry.get("hooks", []):
if has_our_trace(str(h.get("command", ""))):
return True
return False
def unique_backup_path(sp: Path) -> Path:
"""备份落点:**带微秒**的时间戳 ⇒ 连装两遍不会同秒同名互相覆盖(§6 第 4 条实测踩到)。
极端情况下同名仍存在 ⇒ 追加序号,⛔ 绝不覆盖既有备份。"""
ts = time.strftime("%Y%m%d-%H%M%S") + f"-{time.time_ns() // 1000 % 1_000_000:06d}"
cand = sp.with_name(f"{BAK_PREFIX}{ts}")
n = 1
while cand.exists():
cand = sp.with_name(f"{BAK_PREFIX}{ts}-{n}")
n += 1
return cand
def make_backup(sp: Path, before_text: str) -> tuple[Path, Path | None]:
"""写备份。返回 (本次时间戳备份, 本次新写的原状备份或 None)。
· 当前 settings.json **尚未被本包接管**(无钩子指向本包)⇒ 另写固定名 `-orig`,已存在则⛔不覆盖。
· 每次 `--apply` 都另写一份带微秒的时间戳备份(逐次可回滚)。
"""
orig = sp.with_name(ORIG_BAK_NAME)
wrote_orig: Path | None = None
if not points_into_pkg(before_text):
if orig.exists():
log(f"⚠ 原状备份已存在,⛔ 不覆盖(若怀疑原件已换,请人工核对):{orig}")
else:
shutil.copy2(sp, orig)
wrote_orig = orig
log(f"已写**原状**备份(本包接管前)→ {orig}")
else:
log(f"ℹ 当前 settings.json 已被本包接管 ⇒ 跳过 `-orig`(原状备份只在首次安装时留)")
bak = unique_backup_path(sp)
shutil.copy2(sp, bak)
log(f"已备份 → {bak}")
return bak, wrote_orig
def desired_hooks() -> dict[str, list[dict]]:
out: dict[str, list[dict]] = {}
for event, matcher, rel, extra, timeout in HOOKS:
item: dict = {"hooks": [{"type": "command", "command": build_command(rel, extra), "timeout": timeout}]}
if matcher is not None:
item["matcher"] = matcher
out.setdefault(event, []).append(item)
return out
def merge(settings: dict) -> dict:
"""先删本包旧条目 → 再插本包新条目;⛔ 其他条目(含 decision_bridge)原样保留。"""
new = json.loads(json.dumps(settings))
hooks = new.setdefault("hooks", {})
want = desired_hooks()
for event in set(list(hooks.keys()) + list(want.keys())):
keep = [e for e in hooks.get(event, []) if not is_ours(e)]
keep.extend(want.get(event, []))
if keep:
hooks[event] = keep
elif event in hooks:
del hooks[event]
return new
def write_roots(ws: Path, docs: Path | None, code: str | None, dry: bool) -> None:
lines = [
"# session-mechanism · 根目录(由 install.py 生成,⛔ 请勿手改)",
"# 说明:settings.json 的 hook 条目没有 env 字段 ⇒ 包内脚本靠本文件定位根目录。",
"# 读取:各脚本头部 `_sm_load_roots()` 以 setdefault 注入 ⇒ **宿主 env 优先,本文件兜底**。",
f"DSH_WS_ROOT={ws.as_posix()}",
f"WB_RESULT_HOOK_WS={ws.as_posix()}",
# 🔴 2026-10-04 换机演练实测后加:`selftest.py` 的「生产工作区」判据要定位
# **使用方的部署配置**(它验的是生产数据)。⛔ 不可拿 `DSH_WS_ROOT` 拼 ——
# 那条是**自测自己的测试工作区根**(`WS = ... or DSH_WS_ROOT`)⇒ 会把夹具当生产。
f"COLLABD_PROD_CONFIG={(ws / '.workbuddy' / 'collab' / 'collabd.config.json').as_posix()}",
"CODEBUDDY_CONFIG_DIR=%s" % (os.environ.get("CODEBUDDY_CONFIG_DIR")
or os.path.expanduser("~/.workbuddy")).replace(os.sep, "/"),
]
# 可选键:仅当宿主 env 里已有值才落盘(⛔ 不猜、⛔ 不写死盘符)
for _k in ("DSH_OVERLAY_LOG_GLOB", "DSH_OVERLAY_DEV_ROOT"):
if os.environ.get(_k):
lines.append("%s=%s" % (_k, os.environ[_k].replace(os.sep, "/")))
if docs:
lines.append(f"DSH_DOCS_ROOT={docs.as_posix()}")
if code:
lines.append(f"DSH_CODE_REPO={code.as_posix()}")
text = "\n".join(lines) + "\n"
if dry:
log(f"[dry-run] 将写 {ROOTS_ENV}:\n{text}")
return
ROOTS_ENV.write_text(text, encoding="utf-8")
log(f"已写 {ROOTS_ENV}")
def init_workspace(ws: Path, dry: bool) -> None:
"""由 example 生成 collabd.config.json(已存在则不覆盖 ⇒ 保护现场活配置)。"""
src = PKG / "scripts" / "collabd.config.example.json"
dst = ws / ".workbuddy" / "collab" / "collabd.config.json"
if dst.is_file():
log(f"工作区配置已存在,⛔ 不覆盖:{dst}")
return
if not src.is_file():
log(f"⚠ 未找到模板 {src} ⇒ 跳过工作区初始化")
return
try:
cfg = json.loads(src.read_text(encoding="utf-8"))
except Exception as e:
log(f"⚠ 模板解析失败({e})⇒ 跳过")
return
cfg["workspace"] = ws.as_posix()
if dry:
log(f"[dry-run] 将写 {dst}")
return
dst.parent.mkdir(parents=True, exist_ok=True)
dst.write_text(json.dumps(cfg, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
log(f"已写 {dst}")
def cmd_apply(args) -> int:
sp = settings_path()
if not sp.is_file():
log(f"🔴 找不到 {sp} ⇒ 停手")
return 2
ws = detect_workspace(args.workspace)
if ws is None:
log("🔴 未能推断工作区 ⇒ 请显式 `--workspace <路径>`(⛔ 不猜)")
return 2
docs = detect_docs_root(args.docs_root)
log(f"包根={PKG} 工作区={ws} 文档库根={docs} 配置目录={config_dir()}")
raw = sp.read_text(encoding="utf-8")
before = json.loads(raw)
after = merge(before)
diff = "\n".join(
difflib.unified_diff(
json.dumps(before, ensure_ascii=False, indent=2).splitlines(),
json.dumps(after, ensure_ascii=False, indent=2).splitlines(),
fromfile="settings.json (before)", tofile="settings.json (after)", lineterm="",
)
)
print(diff)
excluded = [
str(h.get("command", ""))
for arr in before.get("hooks", {}).values()
for it in arr
for h in it.get("hooks", [])
if "decision_bridge" in str(h.get("command", ""))
]
log(f"(目视核对)⛔ 被排除、未被本包触碰的接线条数:{len(excluded)}(应 >0,且都在 decision_bridge 线上)")
write_roots(ws, docs, args.code_repo, args.dry_run)
init_workspace(ws, args.dry_run)
if args.dry_run:
log("[dry-run] ⛔ 未写 settings.json")
return 0
bak, wrote_orig = make_backup(sp, raw)
sp.write_text(json.dumps(after, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
log(f"已写 {sp}")
with open(LOG, "a", encoding="utf-8") as f:
f.write(f"BACKUP={bak}\n")
if wrote_orig is not None:
f.write(f"ORIG_BACKUP={wrote_orig}\n")
return 0
def cmd_verify(args) -> int:
fails: list[str] = []
# 🔴 ①②③ 一律在**临时目录**里跑:钩子与 collabd 都会按 cwd 推导"工作区"。
# 若直接用调用者的 cwd(常见=包根),它们会把**技能包本身**当成工作区,
# 于是在 `<包>/tmp/supervise-inbox/` 里写出运行态日志 ⇒ **污染技能包**(实测踩到)。
scratch = tempfile.mkdtemp(prefix="sm-verify-")
try:
print("== ① 包内钩子空载荷自检(期望 rc=0)==")
for event, _m, rel, extra, _t in HOOKS:
p = PKG / rel
if not p.is_file():
fails.append(f"{rel} 不存在")
print(f" 🔴 {event:16} {rel} ← 文件不存在")
continue
r = subprocess.run([py(), *extra, str(p)], input=b"{}", capture_output=True, cwd=scratch)
tag = "ok" if r.returncode == 0 else f"rc={r.returncode}"
if r.returncode != 0:
fails.append(f"{rel} rc={r.returncode}")
print(f" {tag:6} {event:16} {rel}")
print("== ② collabd --where ==")
collabd = PKG / "scripts" / "collabd.py"
r = subprocess.run([py(), str(collabd), "--where"], capture_output=True,
cwd=scratch, env={**os.environ, **read_roots()})
out = (r.stdout or b"").decode("utf-8", "replace").strip()
print(f" rc={r.returncode}\n {out[:300]}")
if r.returncode != 0:
fails.append("collabd --where 非 0")
print("== ③ selftest ==")
st = PKG / "scripts" / "selftest.py"
if st.is_file():
r = subprocess.run([py(), str(st)], capture_output=True,
cwd=scratch, env={**os.environ, **read_roots()})
tail = (r.stdout or b"").decode("utf-8", "replace").strip().splitlines()[-6:]
for ln in tail:
print(" " + ln)
if r.returncode != 0 or "FAIL 0" not in " ".join(tail):
fails.append("selftest 未通过(未见 FAIL 0)")
else:
print(" ⚠ 无 selftest.py")
finally:
shutil.rmtree(scratch, ignore_errors=True)
print("== 结论 ==")
if fails:
for f in fails:
print(" 🔴 " + f)
log(f"--verify 失败:{fails}")
return 1
print(" ✅ 全绿")
log("--verify 全绿")
return 0
def cmd_uninstall(args) -> int:
sp = settings_path()
orig = sp.with_name(ORIG_BAK_NAME)
ts_baks = sorted(p for p in sp.parent.glob(f"{BAK_PREFIX}*") if p.name != ORIG_BAK_NAME)
# 🔴 优先级:① 固定名 `-orig`(本包接管前的原状,最可信)→ ② 最早那份「未被本包接管」的时间戳备份。
# ⛔ 不能取 `sorted(...)[-1]`:`--apply` 每次都备份 ⇒ 最新那份其实是**已装状态**,
# 拿它还原 = 还原了个寂寞,而且脚本自比自还会报"逐字节相同"(**假绿**,上一棒实测撞到)。
pristine = orig if orig.is_file() else None
if pristine is not None:
log(f"采用固定名原状备份:{pristine.name}")
else:
for b in ts_baks:
try:
if not points_into_pkg(b.read_text(encoding="utf-8")):
pristine = b
break
except Exception:
continue
if pristine is not None:
log(f"⚠ 无 `-orig` 备份 ⇒ 回落用最早的原状时间戳备份:{pristine.name}")
if pristine is None:
log(f"🔴 {len(ts_baks)} 份时间戳备份里没有一份是「未被本包接管」的原状,且无 `-orig` ⇒ 停手(⛔ 不拿已装状态糊弄)")
return 2
# 还原前先把"当前状态"留一份 ⇒ `--uninstall` 本身也可回退。
stamp = unique_backup_path(sp).name[len(BAK_PREFIX):]
pre = sp.with_name(f"{BAK_PREFIX}preuninstall-{stamp}")
n = 1
while pre.exists():
pre = sp.with_name(f"{BAK_PREFIX}preuninstall-{stamp}-{n}")
n += 1
shutil.copy2(sp, pre)
log(f"(还原前存照)→ {pre}")
shutil.copy2(pristine, sp)
same = hashlib.md5(pristine.read_bytes()).hexdigest() == hashlib.md5(sp.read_bytes()).hexdigest()
clean = not points_into_pkg(sp.read_text(encoding="utf-8"))
log(f"已用 {pristine.name} 还原 {sp}")
log(f" · 与备份逐字节相同:{'✅' if same else '🔴'}")
log(f" · 还原后已无指向本包的钩子:{'✅' if clean else '🔴'}")
return 0 if (same and clean) else 1
# ── --manifest:重算 references/manifest.md 的逐文件表 ──────────────────────
# 🔴 为什么固化进 install.py:包一改,表里的字节/md5 立刻过期 ⇒ **每轮都得重算**。
# 以前每轮临时手搓一个脚本、跑完即删 —— 既浪费又会漏(本项目已因此误判过两次)。
MANIFEST = PKG / "references" / "manifest.md"
# ⛔ 不入表:运行日志 + 表自身(写完即失真)
MANIFEST_SKIP = {"install.log", "references/manifest.md"}
def _manifest_syntax(p: Path) -> str:
if p.suffix == ".py":
try:
compile(p.read_text(encoding="utf-8"), str(p), "exec")
return "ok"
except Exception:
return "🔴 py"
if p.suffix == ".json":
try:
json.loads(p.read_text(encoding="utf-8"))
return "ok"
except Exception:
return "🔴 json"
return "—"
def manifest_files() -> list[Path]:
keep: list[Path] = []
for p in PKG.rglob("*"):
if not p.is_file():
continue
rel = p.relative_to(PKG).as_posix()
if rel in MANIFEST_SKIP or "__pycache__" in p.parts:
continue
if p.suffix in (".pyc", ".pyo") or rel.startswith("tmp/") or "/tmp/" in rel:
continue
keep.append(p)
return sorted(keep, key=lambda q: q.relative_to(PKG).as_posix())
def cmd_manifest(args) -> int:
if not MANIFEST.is_file():
log(f"🔴 找不到 {MANIFEST}")
return 2
files = manifest_files()
rows: list[str] = []
bad: list[str] = []
for p in files:
rel = p.relative_to(PKG).as_posix()
b = p.read_bytes()
st = _manifest_syntax(p)
if st.startswith("🔴"):
bad.append(rel)
rows.append(f"| `{rel}` | {len(b)} | `{hashlib.md5(b).hexdigest()}` | {st} |")
# 🔴 行尾按**原文件**来:本表历史上是 CRLF,写完变 LF 会成为一次无声的全文件 diff。
raw = MANIFEST.read_bytes()
crlf = b"\r\n" in raw
lines = raw.decode("utf-8").replace("\r\n", "\n").split("\n")
# ① 计数行
for i, ln in enumerate(lines):
if ln.startswith("文件总数:"):
lines[i] = f"文件总数:**{len(files)}** | 语法 / 结构检查失败:**{len(bad)}**"
break
# ② 重算时间(`--note` 给括注;不给则保留原括注,⛔ 不抹掉人写的结论)
stamp = time.strftime("%Y-%m-%d %H:%M")
note = getattr(args, "note", None)
# 🔴🔴 2026-10-01 修一处**逐轮累积的写坏**(实测把这一行套成了 4 层):
# 原写法 `re.sub(r"最近一次全量重算:.*?)", …)` —— `.*?` **非贪婪**,只吃到**第一个 `)`**。
# `--note` 里一旦出现全角括号(例如「…行(会说出假话)+ …」),第一个 `)` 就是
# **括注内部**的那个 ⇒ 只替换了前半截,**旧括注的尾巴原样留在原地**,
# 下一轮再套一层 ⇒ 越写越长,新结论和旧结论粘成一句(再看分不清哪句是本次的)。
# ⇒ 改成**用 `**` 收尾界定整段**(模板里 `**最近一次全量重算:…**` 的那对 `**` 是唯一的),
# `--note` 含什么括号都不受影响。
# ⚠️ 不给 `--note` 时**仍保留原括注**(那是人写的结论,⛔ 不抹),但顺手把已经累积出来的
# 多余 `)` 收敛成一个 —— 否则坏行会一直坏下去。
for i, ln in enumerate(lines):
if "最近一次全量重算:" in ln:
_k = ln.find("最近一次全量重算:")
_end = ln.rfind("**")
_tail = ln[_end:] if _end > _k else "" # 正常=`**`;模板被改过则不吃行尾
_body = ln[_k + len("最近一次全量重算:"):(_end if _end > _k else len(ln))]
if note:
_body = f"{stamp} ({note})"
else:
_body = re.sub(r"){2,}", ")", re.sub(r"^\S+ \S+", stamp, _body))
lines[i] = ln[:_k] + "最近一次全量重算:" + _body + _tail
break
# ③ 表:自 `| 包内路径 |` 表头起整段替换(⛔ 上方散文一字不动)
head = next((i for i, ln in enumerate(lines) if ln.startswith("| 包内路径 |")), None)
if head is None:
log("🔴 manifest.md 里找不到 `| 包内路径 |` 表头 ⇒ 停手(⛔ 不猜、不追加)")
return 2
new = lines[:head] + ["| 包内路径 | 字节 | md5 | 语法检查 |", "|---|---|---|---|", *rows]
out = "\n".join(new) + "\n"
if crlf:
out = out.replace("\n", "\r\n")
MANIFEST.write_bytes(out.encode("utf-8"))
log(f"--manifest 已重算:{len(files)} 份文件,语法失败 {len(bad)}")
for r in bad:
log(f" 🔴 {r}")
return 0 if not bad else 1
def main() -> int:
ap = argparse.ArgumentParser(description="session-mechanism 一键配置")
g = ap.add_mutually_exclusive_group(required=True)
g.add_argument("--apply", action="store_true", help="真装")
g.add_argument("--dry-run", action="store_true", help="只打印 diff")
g.add_argument("--verify", action="store_true", help="装完自检")
g.add_argument("--uninstall", action="store_true", help="还原 settings.json")
g.add_argument("--manifest", action="store_true", help="重算 references/manifest.md 的逐文件表")
ap.add_argument("--note", help="写进「最近一次全量重算」括注(仅 --manifest 用)")
ap.add_argument("--workspace", help="工作区根(不传则自动推断)")
ap.add_argument("--docs-root", help="文档库根(不传则自动推断/沿用 roots.env)")
ap.add_argument("--code-repo", help="代码仓根(可选)")
args = ap.parse_args()
if args.apply or args.dry_run:
return cmd_apply(args)
if args.verify:
return cmd_verify(args)
if args.manifest:
return cmd_manifest(args)
return cmd_uninstall(args)
if __name__ == "__main__":
raise SystemExit(main())