#!/usr/bin/env python3 """task-worktree — 多任务并行开发的空间管理（git worktree 生命周期） 执行体 = 本脚本（纯标准库）。规格见 docs/环境与部署/并行任务工作树方案.md， 用法与铁律见同目录 SKILL.md。 退出码：0=成功 / 1=操作失败 / 2=前置或环境错误 / 3=用法错误 """ import argparse import json import os import re import subprocess import sys import tempfile import time from pathlib import Path # 输出统一 UTF-8，避免被控制台代码页（cp936）转码破坏 for _stream in (sys.stdout, sys.stderr): try: _stream.reconfigure(encoding="utf-8") except Exception: pass EXIT_OK, EXIT_FAIL, EXIT_ENV, EXIT_USAGE = 0, 1, 2, 3 BRANCH_PREFIX = "task/" MAX_CONCURRENT = 8 LOCK_NAMES = ("build", "bench") # 按锁分设陈旧窗口（小时）——兜底窗口应与「操作真实时长」同量级： # build：编译是秒~分钟级（keil_build.py 自持自释，持有窗口 ≈ UV4 运行时长， # 上限 = --timeout，默认 600s）；崩溃泄漏时应在数十分钟内即可接管 → 1h。 # bench：会话级（flash → /serial-boot-test → 探针，天然几十分钟~数小时）→ 保留 6h。 # 另：锁 payload 里持有者可用 `ttl` 字段自报窗口（见 lock_acquire(ttl=...)）， # 优先于这里的默认值——这样 build 锁的窗口能自动跟随其 --timeout。 LOCK_TTL_HOURS = {"build": 1.0, "bench": 6.0} DEFAULT_LOCK_TTL_HOURS = 6.0 JUNCTION_SPECS = ( (".venv", "Python 环境"), ("simulator/SDL2", "模拟器 SDL2 SDK"), ) FILE_ATTRIBUTE_REPARSE_POINT = 0x400 ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*$") SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$") class Fail(Exception): """操作失败（退出码 1）""" class EnvError(Exception): """前置或环境错误（退出码 2）""" # --------------------------------------------------------------------- 仓库定位 def _main_worktree_of(start: Path): """借 git 反查「主工作树」路径——脚本可能被从任务工作树里调用，不能只靠 __file__""" for extra in (["--path-format=absolute"], []): r = subprocess.run( ["git", "-C", str(start), "rev-parse", *extra, "--git-common-dir"], capture_output=True, text=True, encoding="utf-8", errors="replace", stdin=subprocess.DEVNULL, ) if r.returncode == 0 and r.stdout.strip(): common = Path(r.stdout.strip()) if not common.is_absolute(): common = (start / common).resolve() return common.parent if common.name == ".git" else common return None _SCRIPT_ROOT = Path(__file__).resolve().parents[4] REPO_ROOT = _main_worktree_of(_SCRIPT_ROOT) or _SCRIPT_ROOT WT_ROOT = Path(os.environ.get("DTU_WT_ROOT") or (REPO_ROOT.parent / "_wt")) BASE_BRANCH = os.environ.get("DTU_TASK_BASE", "master") STATE_DIR = WT_ROOT / ".state" LOCKS_DIR = STATE_DIR / "locks" # bench-pass 记录文件（JSON lines，每行 {branch, timestamp, operator}） # 放在 .agents/worktrees/ 下，因为 git notes 在 Windows worktree 间共享有坑 BENCH_PASS_LOG = REPO_ROOT / ".agents" / "worktrees" / ".bench-pass-log" # 现役 DSH worktree 插件 = @deepseek-ai/dsh-experimental-tool-worktree 0.2.2-rc.1， # 模型工具 create_worktree 默认在仓库内 .agents/worktrees// 建同名分支并切换会话 cwd。 # 旧版 dsh-task-worktree 使用 .dsh-worktrees/worktree//；两处都与本 skill 的 # 仓外 _wt/ 刻意分开。现役插件只负责 create/restore，不合并、不删树、不管锁或共享联接。 # plugin_trees() 同时扫描两处，list 统一展示。 PLUGIN_ROOT_LEGACY = REPO_ROOT / ".dsh-worktrees" PLUGIN_NAMESPACE_LEGACY = "worktree" PLUGIN_SUBDIR_LEGACY = PLUGIN_ROOT_LEGACY / PLUGIN_NAMESPACE_LEGACY PLUGIN_ROOT_AGENTS = REPO_ROOT / ".agents" / "worktrees" PLUGIN_LOCK_PREFIX = "pw" # 插件树在锁 payload 里的持有者前缀 # 兼容旧引用：外部若直接读 PLUGIN_SUBDIR 仍指向 legacy 路径 PLUGIN_SUBDIR = PLUGIN_SUBDIR_LEGACY # --------------------------------------------------------------------- 基础工具 def git(*args, cwd=None, check=True, env=None): cmd = ["git"] + [str(a) for a in args] full_env = dict(os.environ) if env: full_env.update(env) r = subprocess.run( cmd, cwd=str(cwd or REPO_ROOT), capture_output=True, text=True, encoding="utf-8", errors="replace", stdin=subprocess.DEVNULL, env=full_env, ) if check and r.returncode != 0: raise Fail("git {} 失败（rc={}）\n{}".format( " ".join(cmd[1:]), r.returncode, (r.stdout + r.stderr).strip())) return r def git_out(*args, **kw): return git(*args, **kw).stdout.strip() def worktree_entries(): out = git_out("worktree", "list", "--porcelain") entries, cur = [], None for line in out.splitlines(): if line.startswith("worktree "): if cur: entries.append(cur) cur = {"path": Path(line[len("worktree "):]), "branch": None} elif cur is None: continue elif line.startswith("branch refs/heads/"): cur["branch"] = line[len("branch refs/heads/"):] if cur: entries.append(cur) return entries def main_worktree() -> Path: """集成点（合并发生的地方）。默认 = 主工作树；可用环境变量覆盖，便于演练/测试。""" override = os.environ.get("DTU_INTEGRATION_WORKTREE") if override: p = Path(override) if not p.is_absolute(): p = (REPO_ROOT / p).resolve() if not p.is_dir(): raise EnvError("DTU_INTEGRATION_WORKTREE 不存在：{}".format(p)) return p entries = worktree_entries() if not entries: raise EnvError("无法解析 git worktree 列表") return entries[0]["path"] def venv_python(tree: Path = None) -> Path: """优先用该工作树自己的 .venv（任务树里是指向主树的联接），否则退回主工作树""" for root in (tree, REPO_ROOT): if root is None: continue p = Path(root) / ".venv" / "Scripts" / "python.exe" if p.is_file(): return p return Path(sys.executable) def run_script(script: str, args, cwd: Path, label: str, tail: int = 40) -> bool: """跑项目脚本并回报。输出可能上千行，只回显尾部（结论与报错都在尾部） ⚠ 脚本必须取自 **cwd 那棵树**，不能固定用 REPO_ROOT： 项目脚本普遍以 `__file__` 反推仓库根（如 `scripts/ci_local.py`、 `scripts/check_registration.py`），若拿主树的脚本、却给集成树的 cwd， 脚本仍会去校验主树——门禁就测错了对象（2026-10-12 实测踩到： 报出主树才有的 iec104 未注册缺口，而集成树里根本没那些文件）。 """ script_path = Path(cwd) / script if not script_path.is_file(): script_path = REPO_ROOT / script cmd = [str(venv_python(Path(cwd))), str(script_path)] + [str(x) for x in args] r = subprocess.run( cmd, cwd=str(cwd), capture_output=True, text=True, encoding="utf-8", errors="replace", stdin=subprocess.DEVNULL, ) out = (r.stdout + r.stderr).strip() print("[{}] {}".format("ok" if r.returncode == 0 else "!!", label)) if out: lines = out.splitlines() if len(lines) > tail: print(" …（前 {} 行略）".format(len(lines) - tail)) lines = lines[-tail:] print(" " + "\n ".join(lines)) return r.returncode == 0 # --------------------------------------------------------------------- 台账 def state_file(task_id: str) -> Path: return STATE_DIR / (task_id + ".json") def load_state(task_id: str) -> dict: p = state_file(task_id) if not p.is_file(): raise EnvError("任务 {} 不存在（缺台账 {}）".format(task_id, p)) return json.loads(p.read_text(encoding="utf-8")) def save_state(st: dict) -> None: STATE_DIR.mkdir(parents=True, exist_ok=True) state_file(st["id"]).write_text( json.dumps(st, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def all_states(): if not STATE_DIR.is_dir(): return [] out = [] for p in sorted(STATE_DIR.glob("*.json")): try: out.append(json.loads(p.read_text(encoding="utf-8"))) except Exception: continue return out # --------------------------------------------------------------------- 独占锁 def lock_age_hours(info) -> float: try: return max(0.0, (time.time() - float(info.get("epoch") or 0)) / 3600.0) except (TypeError, ValueError): return float("inf") def lock_ttl_hours(name: str, info=None) -> float: """锁 `name` 的陈旧判定窗口（小时）。 优先用锁 payload 里持有者自报的 `ttl`（例如 keil_build.py 按其 --timeout 声明）， 否则回落到该锁的默认窗口（build 1h / bench 6h）。 """ if info: try: v = info.get("ttl") if v: return max(0.0, float(v)) except (TypeError, ValueError): pass return LOCK_TTL_HOURS.get(name, DEFAULT_LOCK_TTL_HOURS) def lock_is_stale(name: str, info) -> bool: return lock_age_hours(info) > lock_ttl_hours(name, info) def lock_path(name: str) -> Path: return LOCKS_DIR / (name + ".lock") def lock_read(name: str): p = lock_path(name) if not p.is_file(): return None try: return json.loads(p.read_text(encoding="utf-8")) except Exception: return {"task": "?", "pid": 0, "epoch": 0, "at": "?"} def lock_acquire(name: str, task: str, wait: int = 0, force: bool = False, note: str = "", ttl: float = None): """抢占独占资源。 ⚠ 锁**不**绑定进程存活：`lock acquire` 本身是短命进程，而用锁的烧录/构建 在另一个进程里跑。因此判「陈旧」只看**时效**（超过该锁的 TTL），外加显式 `--force`；pid 仅作记录便于排障。TTL 按锁分设（build 1h / bench 6h）， 持有者也可用 `ttl`（小时）自报更精确的窗口（如 build 按 --timeout 声明）。 """ LOCKS_DIR.mkdir(parents=True, exist_ok=True) p = lock_path(name) deadline = time.time() + max(0, wait) while True: try: fd = os.open(str(p), os.O_CREAT | os.O_EXCL | os.O_WRONLY) except FileExistsError: info = lock_read(name) or {} if info.get("task") == task: return True, info age = lock_age_hours(info) ttl_h = lock_ttl_hours(name, info) if force or age > ttl_h: print("[lock] 接管 {} 锁（原持有者 {}，已持有 {:.1f} 小时 / TTL {:.1f} 小时{}）".format( name, info.get("task"), age, ttl_h, "，--force" if force else "，视为陈旧")) try: p.unlink() except OSError: pass continue if time.time() >= deadline: return False, info print("[lock] {} 被 {} 占用（已持有 {:.1f} / TTL {:.1f} 小时），等待…".format( name, info.get("task"), age, ttl_h)) time.sleep(2) else: now = time.time() payload = { "name": name, "task": task, "pid": os.getpid(), "host": os.environ.get("COMPUTERNAME", "?"), "epoch": now, "at": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(now)), "note": note, "ttl": ttl, } os.write(fd, (json.dumps(payload, ensure_ascii=False) + "\n").encode("utf-8")) os.close(fd) return True, payload def lock_release(name: str, task=None, force: bool = False): p = lock_path(name) if not p.is_file(): return True, "锁 {} 未被持有".format(name) info = lock_read(name) or {} if not force and task and info.get("task") not in (task, None): return False, "锁 {} 由 {} 持有，与 {} 不符；强制释放请加 --force".format( name, info.get("task"), task) try: p.unlink() except OSError as e: return False, "释放锁 {} 失败：{}".format(name, e) return True, "已释放锁 {}（原持有者 {}）".format(name, info.get("task")) # --------------------------------------------------------------------- bench-pass 记录 def read_bench_pass_log() -> list: """读取 bench-pass 日志文件，返回记录列表（每行一个 JSON 对象）""" if not BENCH_PASS_LOG.is_file(): return [] records = [] try: for line in BENCH_PASS_LOG.read_text(encoding="utf-8").splitlines(): line = line.strip() if line: try: records.append(json.loads(line)) except json.JSONDecodeError: continue except OSError: pass return records def find_bench_pass(branch: str) -> dict: """查找指定分支的最近一次 bench-pass 记录，无记录返回 None""" records = read_bench_pass_log() # 从后往前找，返回最近的匹配 for rec in reversed(records): if rec.get("branch") == branch: return rec return None def write_bench_pass(branch: str, operator: str = None) -> None: """写入一条 bench-pass 记录""" BENCH_PASS_LOG.parent.mkdir(parents=True, exist_ok=True) record = { "branch": branch, "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), "operator": operator or os.environ.get("USERNAME", os.environ.get("USER", "?")), } with open(BENCH_PASS_LOG, "a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") # --------------------------------------------------------------------- 目录联接 def is_reparse(p: Path) -> bool: try: st = os.lstat(str(p)) except OSError: return False return bool(getattr(st, "st_file_attributes", 0) & FILE_ATTRIBUTE_REPARSE_POINT) def make_junction(link: Path, target: Path) -> None: """创建 Windows 目录联接（不需要管理员权限，也不是符号链接） 走 `mklink /J`——这是唯一不需要 SeCreateSymbolicLinkPrivilege 的目录链接形式。 ⚠ 实测（2026-10-12）：直接用 ctypes 调 `DeviceIoControl(FSCTL_SET_REPARSE_POINT)` 会被系统拒绝（Win32 4392 / ERROR_REPARSE_TAG_INVALID，缓冲区经 hexdump 核对无误）， 而 `mklink /J` 正常，故采用后者。参数以**列表**交给 subprocess， 含空格的路径（本项目 `DTU Code`）由 subprocess 自动加引号，不经手写引号。 """ if os.name != "nt": raise EnvError("目录联接是 Windows 专属能力；当前平台 {}".format(os.name)) if os.path.lexists(str(link)): raise Fail("联接位置已被占用：{}".format(link)) if not target.exists(): raise Fail("联接源不存在：{}".format(target)) link.parent.mkdir(parents=True, exist_ok=True) r = subprocess.run( ["cmd", "/c", "mklink", "/J", str(link), str(target)], capture_output=True, stdin=subprocess.DEVNULL, ) if r.returncode != 0 or not is_reparse(link): detail = (r.stdout or b"").decode("mbcs", "replace").strip() raise Fail("创建目录联接失败：{} -> {}\n rc={} {}".format( link, target, r.returncode, detail)) def remove_junction(link: Path) -> bool: """只摘掉联接本身，绝不触碰目标内容（os.rmdir = RemoveDirectoryW）""" if not is_reparse(link): return False try: os.rmdir(str(link)) return True except OSError as e: print("[warn] 摘除联接失败 {}：{}".format(link, e)) return False def bootstrap(wt: Path): """建立共享联接。单个联接失败只告警不中止——避免留下半成品任务空间""" made, failed = [], [] for rel, desc in JUNCTION_SPECS: target = REPO_ROOT / rel if not target.exists(): print("[skip] {}（{}）不存在，跳过联接".format(rel, desc)) continue try: make_junction(wt / rel, target) except (Fail, EnvError) as e: failed.append(rel) print("[WARN] {} 联接失败：{}".format(rel, e)) continue made.append(rel) print("[ok] 联接 {} -> {}".format(rel, target)) if failed: print() print("!! {} 个共享联接未建成；该任务里跑 `.venv\\Scripts\\python.exe` 会找不到解释器。".format(len(failed))) for rel in failed: print(" 手工补：cmd /c mklink /J \"{}\\{}\" \"{}\\{}\"".format(wt, rel, REPO_ROOT, rel)) return made # --------------------------------------------------------------------- DSH 插件工作树 # # 新旧插件建的树缺三样东西，下面的命令逐条补。 # 回收**不在这里**：done/abort 不该处理插件树。现役插件只 create/restore、保留目录与分支； # 要回收时先 unadopt 摘联接，再按本 skill 的 E 节非登记树安全规程处理。 # 1. 共享联接——没有 .venv 与 simulator 的 SDL2 SDK，插件树里「.venv\Scripts\python.exe」 # 直接找不到解释器：/keil-build、ci_local.py、所有 scripts/check_*.py 全部起跑失败； # 2. 可见性——插件树不在台账里，「list」看不见它，8 个活跃名额也只数登记任务； # 3. 独占资源归属——build 锁由 keil_build.py 自持自释（按 cwd 反推持有者名， # 插件树里算成 pw:），bench 锁仍要人占，占锁名得和「lock status」对得上。 def _plugin_name_parts(name: str): """把现役/旧版插件树名归一成同一组路径段；不静默吞掉 .. 等逃逸段。""" raw = (name or "").strip().replace("\\", "/").strip("/") if raw.startswith(".agents/worktrees/"): raw = raw[len(".agents/worktrees/"):] elif raw.startswith(".dsh-worktrees/"): raw = raw[len(".dsh-worktrees/"):] segs = [s for s in raw.split("/") if s] if segs and segs[0] == PLUGIN_NAMESPACE_LEGACY: segs = segs[1:] return segs def plugin_lock_owner(name: str) -> str: """插件树在锁 payload 里的持有者名：pw:（多段名压平成 -）。 必须与 keil_build.py 的 default_lock_task() 对同一种树算出的名字一致——这样构建 自持自释留下的归属，和手工「lock acquire bench --task pw:x」才是同一个可 grep 的标识。 新旧路径写法与裸名都归一到同一个 owner，避免 lock status 出现两个持有者名。 """ return "{}:{}".format(PLUGIN_LOCK_PREFIX, "-".join(_plugin_name_parts(name)) or "?") def plugin_dir_for(name: str) -> Path: """插件 worktree 名 → 树目录；现役 .agents/worktrees/ 与旧路径均可，禁止逃逸。 adopt/unadopt 会在解析出的目录上建/摘目录联接；不加这道闸，等于允许把主工作树或 任意 _wt 目录当插件树动手（联接会遮蔽真实内容，事后很难看出来）。同名树两处都有时 拒绝猜测，要求传带根路径的全名。 """ raw = (name or "").strip().replace("\\", "/").strip("/") if not raw: raise EnvError("缺插件 worktree 名（先跑「plugin list」看现有树）") if raw in (".agents", ".agents/worktrees", ".dsh-worktrees", ".dsh-worktrees/worktree"): raise EnvError("拒绝 {}：必须指定插件树名，不能只给根目录".format(raw)) segs = _plugin_name_parts(raw) if not segs or any(s in (".", "..") for s in segs): raise EnvError("拒绝 {}：插件树名不能为空，也不能含 . 或 ..".format(name)) explicit = raw.startswith(".agents/worktrees/") or raw.startswith(".dsh-worktrees/") bases = [PLUGIN_ROOT_AGENTS] if raw.startswith(".agents/worktrees/") else [] if raw.startswith(".dsh-worktrees/"): bases = [PLUGIN_SUBDIR_LEGACY] if not explicit: bases = [PLUGIN_ROOT_AGENTS, PLUGIN_SUBDIR_LEGACY] matches = [] for base in bases: p = (base.joinpath(*segs)).resolve() resolved_base = base.resolve() if p == resolved_base or resolved_base not in p.parents: raise EnvError("拒绝 {}：只接受插件命名空间里面的树".format(p)) if p.is_dir(): matches.append(p) if len(matches) > 1: raise EnvError("插件树名 {} 在新旧两处都存在；请传 .agents/worktrees/{} 或 " ".dsh-worktrees/worktree/{} 消歧".format(name, raw, raw)) # 未创建/已删除的树也要能给出确定的候选路径，让上层报「目录不存在」而非 NameError。 return matches[0] if matches else (bases[0].joinpath(*segs)).resolve() def junction_states(wt: Path): """每个共享联接的三态：link（目录联接）/ real（实体内容）/ absent（真缺）。 这是「联没联上」的**唯一**判定处——列表显示与 adopt 收尾都读它。早先列表只取样 JUNCTION_SPECS 的第一项，于是「有 .venv、缺 simulator/SDL2」的半联树被报成已联， 而 adopt 可以只成功一条（主线缺 SDL2 走 skip、联接失败走 warn）→ 说谎的正是这一列。 """ st = {} for rel, _desc in JUNCTION_SPECS: link = wt / rel if is_reparse(link): st[rel] = "link" elif os.path.lexists(str(link)): st[rel] = "real" else: st[rel] = "absent" return st def junction_label(st) -> str: """把三态压成一列短标签：缺什么都点名，实体内容也算「齐」但不算共享。""" tag = {rel: rel.rsplit("/", 1)[-1] for rel in st} absent = [tag[r] for r, s in st.items() if s == "absent"] real = [tag[r] for r, s in st.items() if s == "real"] if absent: return "缺 " + "+".join(absent) return "齐(实体" + "+".join(real) + ")" if real else "已联" def plugin_trees(): """扫描两处 DSH 插件工作树（事实源 = git worktree list，不读插件 manifest）。 位置： - .dsh-worktrees/worktree/ （旧版 dsh-task-worktree 插件） - .agents/worktrees/ （现役 dsh-experimental-tool-worktree 的 create_worktree 工具） manifest.json 是插件私有格式（MANIFEST_VERSION 1，字段随版本变），而 git 的注册表 同时覆盖「插件正常建的树」和「manifest 丢了 / 树被手工删」的残留——桥接命令要看的 是真实存在的东西。 """ bases = [] for candidate in (PLUGIN_SUBDIR_LEGACY, PLUGIN_ROOT_AGENTS): try: resolved = candidate.resolve() if resolved.is_dir(): bases.append(resolved) except OSError: continue if not bases: return [] trees = [] seen_paths = set() for ent in worktree_entries(): p = ent["path"] try: rp = p.resolve() except OSError: continue matched_base = None for base in bases: if rp == base or base in rp.parents: matched_base = base break if matched_base is None: continue # 去重（同一棵树不会出现在两个 base 下，但防御性检查） path_key = str(rp) if path_key in seen_paths: continue seen_paths.add(path_key) dirty = (len(git_out("status", "--porcelain", cwd=p, check=False).splitlines()) if p.is_dir() else -1) trees.append({ "name": rp.relative_to(matched_base).as_posix(), "path": str(p), "branch": ent.get("branch") or "(detached)", "dirty": dirty, "jn": junction_states(p), }) return trees def print_plugin_trees() -> int: """打印插件树一览（无树则静默），返回数量——「list」与「new」共用同一事实。""" trees = plugin_trees() if not trees: return 0 print() print("DSH 插件工作树（.agents/worktrees/ 与旧 .dsh-worktrees/worktree/；**不在本 skill " "台账** → 不占 {} 个名额、done/abort 管不到；现役插件保留目录与分支，回收先 " "unadopt 再走 E 节安全规程）：".format(MAX_CONCURRENT)) print("{:<26} {:<34} {:>4} {:>16}".format("名称", "分支", "脏", "联接")) for t in trees: print("{:<26} {:<34} {:>4} {:>16}".format( t["name"], t["branch"], "-" if t["dirty"] < 0 else str(t["dirty"]), junction_label(t["jn"]))) def miss(rel): return any(t["jn"].get(rel) == "absent" for t in trees) if miss(".venv"): print(" 缺 .venv：进去跑 /keil-build / ci_local.py 会**静默**落到机器级解释器（无 " "fontTools，generated-fresh 假红的老根因）") if miss("simulator/SDL2"): print(" 缺 simulator/SDL2：模拟器 build_sim.bat 直接报错——CMakeLists.txt 的 " "SDL2_DIR 是裸 set，没有环境变量逃生口") if miss(".venv") or miss("simulator/SDL2"): print(" 补法：「plugin adopt 」——只建缺失的联接，不碰树里任何文件") return len(trees) def adopt_junctions(wt: Path): """只补**缺失**的共享联接（已联的跳过、实体目录绝不碰），返回 (已有, 本次新建)。""" already, made = [], [] st = junction_states(wt) for rel, desc in JUNCTION_SPECS: link = wt / rel if st[rel] == "link": already.append(rel) continue if st[rel] == "real": print("[warn] {} 已是实体目录/文件（不是联接）——不动它".format(rel)) already.append(rel) continue target = REPO_ROOT / rel if not target.exists(): print("[skip] {}（{}）主线里不存在，跳过".format(rel, desc)) continue try: make_junction(link, target) except (Fail, EnvError) as err: print("[WARN] {} 联接失败：{}".format(rel, err)) continue made.append(rel) print("[ok] 联接 {} -> {}".format(rel, target)) return already, made def cmd_plugin(a) -> int: if a.action == "list": if print_plugin_trees() == 0: print("（.agents/worktrees/ 与旧 .dsh-worktrees/worktree/ 下没有插件工作树）") print(" 建法：让模型调现役插件的 create_worktree；默认建 .agents/worktrees// " "及同名分支，并切换会话 cwd") print(" 建完回来「plugin adopt 」补共享联接，才能在里面跑 /keil-build 与门禁。") return EXIT_OK wt = plugin_dir_for(a.name) if not wt.is_dir(): raise EnvError("目录不存在：{}（先用现役插件的 create_worktree 建树；旧树也可放在 " ".dsh-worktrees/worktree//）".format(wt)) if a.action == "adopt": if git("rev-parse", "--git-dir", cwd=wt, check=False).returncode != 0: raise EnvError("{} 不是 git 工作树（别拿普通目录当插件树）".format(wt)) print("为插件树补共享联接：{}".format(wt)) already, made = adopt_junctions(wt) owner = plugin_lock_owner(a.name) # 复算一次：adopt 里 skip/warn/失败都不算成功，「已有 N + 新建 M」两两相加可能对不上 # 总数——就绪判定只能按**现在的实际状态**说，否则又是把没联上的说成联上了 still = [r for r, s in junction_states(wt).items() if s == "absent"] print() print("[ok] 联接：已有 {} 个、本次新建 {} 个".format(len(already), len(made))) if still: print("[warn] 仍缺 {}：主线里该路径不存在或联接失败——先补主线再复跑 adopt" "（这棵树还不能算就绪）".format("、".join(still))) else: print("该目录现在可以跑 /keil-build / ci_local.py / /serial-boot-test") print("build 锁由 keil_build.py 自持自释，持有者名就是 {}".format(owner)) print("上台架前：$TW lock acquire bench --task {} --note <说明>".format(owner)) print("下台架后：$TW lock release bench --task {}".format(owner)) print("现役插件只 create/restore，不自动回收，会保留目录与分支；本 skill 不代删。") print("确需回收：先 plugin unadopt {} 摘联接，再按 E 节非登记树安全规程处理。".format(a.name)) return EXIT_OK removed = [rel for rel, _spec in JUNCTION_SPECS if remove_junction(wt / rel)] print("[ok] 摘掉联接：{}（只摘联接本身，绝不动目标内容与树里文件）".format( "、".join(removed) if removed else "无（该树没有联接着）")) return EXIT_OK # --------------------------------------------------------------------- bench-gate / merge（新统一模型） def cmd_bench_gate(a) -> int: """检查指定分支是否有 bench-pass 记录""" branch = a.branch rec = find_bench_pass(branch) if rec is None: print("分支 {} 无 bench-pass 记录，请先跑台架验证".format(branch), file=sys.stderr) return EXIT_FAIL print("[ok] 分支 {} 有 bench-pass 记录".format(branch)) print(" 最近通过时间：{}".format(rec.get("timestamp", "?"))) print(" 操作人：{}".format(rec.get("operator", "?"))) return EXIT_OK def cmd_merge(a) -> int: """在主树执行编排式合并""" branch = a.branch main = main_worktree() # 检查当前是否在主树 if Path.cwd().resolve() != main.resolve(): raise EnvError("merge 必须在主工作树执行：{}".format(main)) # 检查主树是否干净 if git_out("status", "--porcelain", cwd=main, check=False): raise Fail("主工作树不干净——合并会把未提交改动卷进来") cur = git_out("rev-parse", "--abbrev-ref", "HEAD", cwd=main) base = BASE_BRANCH if cur != base: raise Fail("主工作树当前在 {}，不是集成分支 {}；先切过去再 merge".format(cur, base)) # 检查分支是否存在 if git("rev-parse", "--verify", "--quiet", "refs/heads/" + branch, check=False).returncode != 0: raise EnvError("分支不存在：{}".format(branch)) # Step 1: bench-gate 检查（除非 --skip-bench） if not a.skip_bench: rec = find_bench_pass(branch) if rec is None: print("分支 {} 无 bench-pass 记录，请先跑台架验证".format(branch), file=sys.stderr) print("（加 --skip-bench 可跳过此检查）", file=sys.stderr) return EXIT_FAIL print("[ok] bench-gate 通过（{}）".format(rec.get("timestamp", "?"))) # Step 2: 预演合并（--no-commit） print("=== 合并预演（git merge --no-ff --no-commit）===") r = git("merge", "--no-ff", "--no-commit", branch, cwd=main, check=False) if r.returncode != 0: # 中止合并 git("merge", "--abort", cwd=main, check=False) out = (r.stdout + r.stderr).strip() print(out) raise Fail("合并有冲突：请回到任务分支解决冲突后重试\n" " cd <任务工作树> && git merge {} && ... && git commit".format(base)) # Step 3: 检查是否有实际改动（可能已经合入过了） staged = git_out("diff", "--cached", "--name-only", cwd=main, check=False) if not staged.strip(): # 没有改动，说明分支已合入或无差异 git("merge", "--abort", cwd=main, check=False) print("[i] 分支 {} 已合入或与 {} 无差异，无需合并".format(branch, base)) return EXIT_OK # Step 4: 完成合并提交 print("=== 完成合并提交 ===") title = "merge: {} 合入 {}".format(branch, base) if len(title) > 100: title = title[:97] + "..." msg_lines = [title, "", "经 task-worktree merge 命令合入。"] fd, msgpath = tempfile.mkstemp(prefix="taskwt-merge-", suffix=".txt") os.close(fd) Path(msgpath).write_text("\n".join(msg_lines) + "\n", encoding="utf-8", newline="\n") try: r = git("commit", "--no-edit", "-F", str(msgpath), cwd=main, check=False) if r.returncode != 0: raise Fail("合并提交失败：\n{}".format((r.stdout + r.stderr).strip())) finally: Path(msgpath).unlink(missing_ok=True) sha = git_out("rev-parse", "--short", "HEAD", cwd=main) print("[ok] 合并完成（{}）".format(sha)) print() print("合并完成，请在主树跑全量门禁") print("提示：任务分支和目录未自动删除，清场请走 plugin unadopt + 手动回收") return EXIT_OK # --------------------------------------------------------------------- 子进程脚本 def regen(cwd: Path) -> None: run_script("scripts/gen_task_table.py", ["--write"], cwd, "重生成 FreeRTOS 任务表") if list((Path(cwd) / "MDK-ARM").glob("**/*.map")): run_script("scripts/check_mem_numbers.py", ["--write"], cwd, "同步内存/Flash 快照") run_script("scripts/ram_baseline.py", ["--update"], cwd, "追加 RAM 趋势基线行") run_script("scripts/ram_baseline.py", ["--write"], cwd, "同步 RAM 预算台账受管区") else: print("[skip] 未找到 .map，跳过内存快照与 RAM 台账同步（需先 /keil-build）") def run_gates(cwd: Path, skip_build: bool) -> bool: # 先构建再跑门禁：ci_local.py 含 mem-budget 钩子，需要新鲜 .map 才能正确比对台账。 # 旧版先 ci 后 build 导致合并后 .map 仍是旧的 → mem-budget 必红。 ok = True if not skip_build: got, info = lock_acquire("build", "__done__", wait=900) if not got: print("!! build 锁被 {} 占用（UV4 不能并发构建），跳过 /keil-build".format(info.get("task"))) return False # 告知 /keil-build「调用方已持锁」：新版 keil_build.py 自持自释（见其 docstring）， # 读到本变量即跳过，避免自我死锁；**旧版根本不看此变量、也不会自锁**， # 故此处对「工作树里还是旧脚本」的在途任务同样安全（向后兼容）。 os.environ["DTU_LOCK_HELD"] = "build" try: if not run_script(".dsh/skills/keil-build/scripts/keil_build.py", [], cwd, "/keil-build（0 错 0 警）"): ok = False finally: os.environ.pop("DTU_LOCK_HELD", None) lock_release("build", "__done__", force=True) else: print("[skip] 按 --skip-build 跳过 /keil-build") # 构建完成（或跳过后）再跑静态门禁 + 单测，此时 .map 是新鲜的 if not run_script("scripts/ci_local.py", [], cwd, "静态门禁 + 主机侧纯逻辑单测 本地等价 CI"): ok = False return ok # --------------------------------------------------------------------- 合并辅助 def write_merge_message(st: dict, base: str, cwd: Path) -> Path: raw = git_out("log", "--format=%s", "{}..{}".format(base, st["branch"]), cwd=cwd) subjects = [s.strip() for s in raw.splitlines() if s.strip()] title = "merge: {} 合入 {}".format(st["branch"], base) if len(title) > 100: title = title[:97] + "..." lines = [title, "", "任务 {}（{} 个提交）经 task-worktree 回收：".format(st["id"], len(subjects))] for s in subjects: lines.append("- " + (s[:73] + "..." if len(s) > 76 else s)) if st.get("note"): note = st["note"] lines.append("") lines.append("备注：" + (note[:73] + "..." if len(note) > 76 else note)) fd, path = tempfile.mkstemp(prefix="taskwt-merge-", suffix=".txt") os.close(fd) Path(path).write_text("\n".join(lines) + "\n", encoding="utf-8", newline="\n") return Path(path) def audit_residue(base_before: str, branch: str, cwd: Path) -> None: """并集合并残留审计：双方都改过的文件会被 git 自动并成并集""" mb = git_out("merge-base", base_before, branch, cwd=cwd) task_files = set(filter(None, git_out("diff", "--name-only", mb, branch, cwd=cwd).splitlines())) base_files = set(filter(None, git_out("diff", "--name-only", mb, base_before, cwd=cwd).splitlines())) both = sorted(task_files & base_files) if not both: print("[ok] 没有「双方都改过」的文件，无并集残留风险") return print("以下 {} 个文件双方都改过，git 会自动并成并集，需人工核对：".format(len(both))) for f in both[:30]: print(" - " + f) if len(both) > 30: print(" …（其余 {} 个略）".format(len(both) - 30)) hits_dup = [] hits_marker = [] conflict_markers = ("<<<<<<<", "=======", ">>>>>>>") structured_exts = (".json", ".tsv", ".csv") for f in both: p = cwd / f if not p.is_file(): continue try: content = p.read_text(encoding="utf-8", errors="replace") lines = content.splitlines() except OSError: continue # 检查 git 冲突标记（结构化文件尤其危险，会静默损坏 JSON/TSV） ext = os.path.splitext(f)[1].lower() if ext in structured_exts or any(m in content for m in conflict_markers): marker_lines = [i + 1 for i, line in enumerate(lines) if any(line.strip().startswith(m) for m in conflict_markers)] if marker_lines: hits_marker.append((f, marker_lines)) # 检查连续重复行（并集合并的经典症状） dups = [i + 1 for i in range(1, len(lines)) if lines[i] == lines[i - 1] and lines[i].strip()] if dups: hits_dup.append((f, dups)) if hits_marker: print() print("!! 发现 git 冲突标记（结构化文件损坏风险极高）：") for f, lns in hits_marker: print(" - {} 第 {} 行".format(f, ", ".join(str(l) for l in lns[:8]))) if hits_dup: print() print("!! 连续重复行（并集合并残留嫌疑）：") for f, dups in hits_dup: print(" - {} 第 {} 行".format(f, ", ".join(str(d) for d in dups[:8]))) if not hits_marker and not hits_dup: print("[ok] 未检测到冲突标记或连续重复行") print("提示：权威判据是 base / 两个 tip 三方对比（git show :），别只看结果文件。") def harvest_memory(st: dict) -> None: """任务工作树会被删除，先把它的 dsh-mnemon 记忆归并回主工作树，避免丢上下文。 只处理 .mnemon/ 目录（dsh-mnemon 插件）。目录级合并：只拷贝目标不存在的子项， 不覆盖已有内容。若无 .mnemon/ 则静默跳过。 """ wt = Path(st["path"]) src_mn = wt / ".mnemon" if not src_mn.is_dir(): return dst_mn = REPO_ROOT / ".mnemon" dst_mn.mkdir(parents=True, exist_ok=True) import shutil count = 0 for item in sorted(src_mn.iterdir()): target = dst_mn / item.name try: if item.is_dir(): if not target.exists(): shutil.copytree(str(item), str(target)) count += 1 else: if not target.exists(): shutil.copy2(str(item), str(target)) count += 1 except OSError as e: print("[warn] 记忆拷贝失败 {}: {}".format(item.name, e)) if count: print("[ok] 记忆归并(mnemon) {} 个条目 -> {}".format(count, dst_mn)) def teardown_worktree(st: dict, force: bool = False, base: str = None) -> None: wt = Path(st["path"]) base = base or st.get("base") # 1) 先摘联接——绝不能让 git 递归进联接去删目标内容（.venv / SDL2 会被一起删掉） for rel in st.get("junctions", []): if remove_junction(wt / rel): print("[ok] 已摘除联接 {}".format(rel)) # 2) 归并任务侧记忆 if wt.is_dir(): harvest_memory(st) # 3) 移除工作树 r = git("worktree", "remove", *(["--force"] if force else []), str(wt), check=False) if r.returncode != 0: raise Fail("移除工作树失败：{}\n（提示：先处理未跟踪文件，或加 --force）".format( (r.stdout + r.stderr).strip())) # 4) 删分支。⚠ 不用 `git branch -d`——它按**环境 HEAD** 判断是否已合并， # 集成点被 DTU_INTEGRATION_WORKTREE 改走时会误判。改为显式核对 base。 if force or git("merge-base", "--is-ancestor", st["branch"], base, check=False).returncode == 0: r = git("branch", "-D", st["branch"], check=False) if r.returncode != 0: print("[warn] 删除分支 {} 失败：{}".format( st["branch"], (r.stdout + r.stderr).strip())) else: print("[ok] 已删除分支 {}".format(st["branch"])) else: print("[warn] 分支 {} 未并入 {}，保留不删（确认可弃再 `git branch -D {}`）".format( st["branch"], base, st["branch"])) state_file(st["id"]).unlink(missing_ok=True) # --------------------------------------------------------------------- 子命令 def cmd_list(a) -> int: states = all_states() if not states: print("（没有任务空间。用 `new ` 建一个）") print_plugin_trees() return EXIT_OK print("{:<12} {:<32} {:>5} {:>5} {:>4} {}".format( "任务", "分支", "落后", "领先", "脏", "台架")) for st in states: wt = Path(st["path"]) if not wt.is_dir(): print("{:<12} {:<32} {:>5} {:>5} {:>4} {}".format( st["id"], st["branch"], "-", "-", "-", "[工作树缺失]")) continue ab = git_out("rev-list", "--left-right", "--count", "{}...{}".format(st["base"], st["branch"]), cwd=wt, check=False).split() behind, ahead = (ab[0], ab[1]) if len(ab) == 2 else ("?", "?") dirty = len(git_out("status", "--porcelain", cwd=wt, check=False).splitlines()) bench = "PASS" if st.get("bench_pass") else ("需测" if st.get("needs_bench") else "-") print("{:<12} {:<32} {:>5} {:>5} {:>4} {}".format( st["id"], st["branch"], behind, ahead, dirty, bench)) print() for n in LOCK_NAMES: info = lock_read(n) if not info: print("锁 {:<6} 空闲".format(n)) else: print("锁 {:<6} {} 持有（已 {:.1f} 小时，于 {}）".format( n, info.get("task"), lock_age_hours(info), info.get("at"))) print_plugin_trees() return EXIT_OK def cmd_status(a) -> int: st = load_state(a.task_id) wt = Path(st["path"]) print("任务 {} ({})".format(st["id"], st["slug"])) print("分支 {} ← 基线 {}".format(st["branch"], st["base"])) print("工作树 {}{}".format(wt, "" if wt.is_dir() else " [缺失]")) print("建立于 {}".format(st.get("created_at"))) if st.get("note"): print("备注 {}".format(st["note"])) if wt.is_dir(): dirty = git_out("status", "--porcelain", cwd=wt, check=False).splitlines() print("未提交 {} 项".format(len(dirty))) for line in dirty[:10]: print(" " + line) ab = git_out("rev-list", "--left-right", "--count", "{}...{}".format(st["base"], st["branch"]), cwd=wt, check=False).split() if len(ab) == 2: print("相对基线 落后 {} / 领先 {} 个提交".format(ab[0], ab[1])) print("台架 {}{}".format( "需要，" if st.get("needs_bench") else "不涉及，", "已记录 PASS" if st.get("bench_pass") else "尚无 PASS 记录")) if st.get("junctions"): print("联接 {}".format(", ".join(st["junctions"]))) return EXIT_OK def cmd_sync(a) -> int: st = load_state(a.task_id) wt = Path(st["path"]) base = a.base or st["base"] if not wt.is_dir(): raise EnvError("工作树不存在：{}".format(wt)) if git_out("status", "--porcelain", cwd=wt, check=False): raise Fail("任务工作树有未提交改动，先提交或 stash 再 sync") r = git("rebase", base, cwd=wt, check=False) if r.returncode != 0: print((r.stdout + r.stderr).strip()) raise Fail("rebase 有冲突：在 {} 里解决后 `git rebase --continue`，放弃用 `git rebase --abort`".format(wt)) print("[ok] 任务 {} 已 rebase 到 {}".format(st["id"], base)) return EXIT_OK def cmd_lock(a) -> int: if a.action == "status": for n in LOCK_NAMES: info = lock_read(n) if not info: print("{:<6} 空闲（TTL {:.1f}h）".format(n, lock_ttl_hours(n))) else: age = lock_age_hours(info) ttl_h = lock_ttl_hours(n, info) mark = " [陈旧，可被自动接管]" if age > ttl_h else "" print("{:<6} {} 持有（pid={}，于 {}，已 {:.1f} / TTL {:.1f} 小时，备注 {}）{}".format( n, info.get("task"), info.get("pid"), info.get("at"), age, ttl_h, info.get("note") or "-", mark)) return EXIT_OK if a.name not in LOCK_NAMES: raise EnvError("未知资源 {}（可用：{}）".format(a.name, " / ".join(LOCK_NAMES))) if a.action == "acquire": if not a.task: raise EnvError("acquire 需要 --task ") got, info = lock_acquire(a.name, a.task, wait=a.wait, force=a.force, note=a.note or "", ttl=a.ttl) if not got: print("获取失败：{} 被 {} 占用（pid={}，于 {}）".format( a.name, info.get("task"), info.get("pid"), info.get("at"))) return EXIT_FAIL print("已获取 {} 锁（task={} pid={} ttl={}）".format( a.name, a.task, os.getpid(), "{:.2f}h".format(a.ttl) if a.ttl else "默认")) return EXIT_OK if a.bench_pass: if not a.task: raise EnvError("--bench-pass 需要 --task ") # 写入统一的 bench-pass-log（新模型） # 尝试从台账获取分支名，否则用 task ID 作为分支标识 branch_name = a.task if state_file(a.task).is_file(): try: st = load_state(a.task) st["bench_pass"] = True save_state(st) branch_name = st.get("branch", a.task) except Exception: pass write_bench_pass(branch_name) print("[ok] 已记录台架 PASS（分支 {}）到 {}".format(branch_name, BENCH_PASS_LOG)) ok, msg = lock_release(a.name, a.task, force=a.force) print(msg) return EXIT_OK if ok else EXIT_FAIL def release_stale_locks() -> None: """`done` 前置：陈旧锁自动接管；仍被有效占用的则拒绝收口。 「陈旧」= 持有时间 > 该锁 TTL（build 1h / bench 6h，或持有者自报的 ttl）。 崩溃/遗忘留下的锁不该阻塞收口——故自动释放而非硬失败（2026-10-12）。 """ for n in LOCK_NAMES: info = lock_read(n) if not info: continue age = lock_age_hours(info) ttl_h = lock_ttl_hours(n, info) if age > ttl_h: ok, msg = lock_release(n, force=True) print("[i] {} 锁已陈旧（{} 持有 {:.1f}h > TTL {:.1f}h），自动接管：{}".format( n, info.get("task"), age, ttl_h, msg)) continue raise Fail("独占资源 {} 仍被 {} 占用（已 {:.1f} / TTL {:.1f} 小时），先 release".format( n, info.get("task"), age, ttl_h)) def cmd_done(a) -> int: print("[DEPRECATED] done 已废弃，请用 merge 命令", file=sys.stderr) print() st = load_state(a.task_id) wt = Path(st["path"]) branch = st["branch"] base = a.base or st["base"] main = main_worktree() print("=== 前置检查 ===") if not wt.is_dir(): raise EnvError("工作树不存在：{}".format(wt)) if git_out("status", "--porcelain", cwd=wt, check=False): raise Fail("任务工作树有未提交改动，先提交") if git_out("status", "--porcelain", cwd=main, check=False): raise Fail("主工作树不干净（{}）——合并会把未提交改动卷进来".format(main)) cur = git_out("rev-parse", "--abbrev-ref", "HEAD", cwd=main) if cur != base: raise Fail("主工作树当前在 {}，不是集成分支 {}；先切过去再 done".format(cur, base)) release_stale_locks() ahead = int(git_out("rev-list", "--count", "{}..{}".format(base, branch), cwd=wt) or "0") already = git("merge-base", "--is-ancestor", branch, base, cwd=main, check=False).returncode == 0 if ahead == 0 and not already: raise Fail("任务分支相对 {} 没有新提交，无需合并".format(base)) if already and ahead == 0: print("[i] {} 已合入 {}，跳过合并，直接清场".format(branch, base)) print("[i] 该路径在门禁之前就返回——不替你复绿；上一次门禁没跑到的尾巴，" "清场后去主树跑 `postmerge`（或手工 ci_local + check_doc_links --all）") print("=== 清场 ===") teardown_worktree(st, force=a.force, base=base) print("\n完成（分支此前已合入）。") return EXIT_OK if st.get("needs_bench") and not st.get("bench_pass"): raise Fail("任务标记 --needs-bench 但无台架 PASS 记录；跑完 /serial-boot-test 后执行 " "`lock release bench --task {} --bench-pass`".format(st["id"])) print("[ok] 前置检查通过（领先 {} 共 {} 个提交）".format(base, ahead)) print("=== 合并预演（git merge-tree，不改动任何东西）===") r = git("merge-tree", "--write-tree", base, branch, cwd=main, check=False) if r.returncode != 0: print((r.stdout + r.stderr).strip()) raise Fail("预演发现冲突：到任务工作树里 `git merge {}` 解决并提交后重跑 done".format(base)) tree = (r.stdout.splitlines() or ["?"])[0].strip() print("[ok] 无冲突，预演结果树 {}".format(tree)) # 台账类生成物碰撞预检：双方都改过的已知生成文件在合并后大概率需要重算 ledger_files = ( "scripts/ram_baseline.tsv", "scripts/size_attestation.json", "docs/内存管理/RAM预算台账.md", "docs/内存管理/RAM使用分析与优化方案.md", ) mb = git_out("merge-base", base, branch, cwd=main, check=False) task_changed = set(filter(None, git_out("diff", "--name-only", mb, branch, cwd=main, check=False).splitlines())) base_changed = set(filter(None, git_out("diff", "--name-only", mb, base, cwd=main, check=False).splitlines())) collided = sorted(f for f in ledger_files if f in task_changed and f in base_changed) if collided: print() print("[warn] 以下 {} 个台账/凭证文件双方都改过，合并后需 postmerge 重算：".format(len(collided))) for f in collided: print(" - " + f) print(" （done 会自动调 postmerge 补绿；若 --dry-run 请留意此提示）") if a.dry_run: print() print("--dry-run：到此为止，未做任何改动。") return EXIT_OK print("=== 合入主干（{}，--no-ff）===".format(main)) base_before = git_out("rev-parse", "HEAD", cwd=main) msgfile = write_merge_message(st, base, wt) r = git("merge", "--no-ff", "--no-edit", branch, "-F", str(msgfile), cwd=main, check=False) msgfile.unlink(missing_ok=True) if r.returncode != 0: raise Fail("合并失败：\n{}".format((r.stdout + r.stderr).strip())) print((r.stdout or "").strip()) base_after = git_out("rev-parse", "HEAD", cwd=main) print("=== 合并残留审计 ===") audit_residue(base_before, branch, cwd=main) # 统一补绿链：regen + build + ci + postmerge 收敛到一条路径。 # regen 先跑（任务表/内存快照），然后 build+ci，最后 postmerge 兜底凭证与文档。 print("=== 重生成受管产物 ===") regen(main) if a.skip_gates: print("=== 门禁（--skip-gates 已跳过）===") gates_ok = True else: print("=== 门禁 ===") gates_ok = run_gates(main, skip_build=a.skip_build) # 无论门禁是否通过，都跑一次 postmerge 确保凭证/文档全库一致。 # 门禁已通过时 postmerge 检测到新鲜凭证会直接跳过（零开销）。 print("=== postmerge 补绿（凭证 + 文档全库）===") pm_args = argparse.Namespace(force=False, skip_mem=False, coverage=False, skip_docs=False) try: pm_rc = cmd_postmerge(pm_args) except Exception as e: print("[warn] postmerge 异常：{}".format(e)) pm_rc = EXIT_FAIL # postmerge 完成后检查是否有回填产物需要提交 dirty = git_out("diff", "--name-only", "HEAD", cwd=main, check=False).strip() if dirty and not gates_ok: print() print("!! 门禁未全绿且 postmerge 产生了回填文件，工作树与分支**保留**。") print(" 主干已含合并提交 {}".format(base_after[:12])) print("[i] 以下文件待提交（属 Ask first）：") for line in dirty.splitlines(): if line.strip(): print(" - " + line.strip()) print(" 建议：git add <上面列的文件> && git commit -F <正文文件>") print(" 提交后重跑 `done {}` 完成清场。".format(a.task_id)) return EXIT_FAIL elif not gates_ok: print() print("!! 门禁未全绿，工作树与分支**保留**，未清场。") print(" 主干已含合并提交 {}".format(base_after[:12])) print(" 请手动排查后重跑 `done {}`。".format(a.task_id)) print(" 或回退本次合并：git -C \"{}\" reset --hard {}".format(main, base_before[:12])) return EXIT_FAIL # 门禁通过 + postmerge 完成 → 若有回填文件则提示提交后再清场 if dirty: print() print("[i] postmerge 已回填，以下文件待提交（属 Ask first）：") for line in dirty.splitlines(): if line.strip(): print(" - " + line.strip()) print(" 建议：git add <上面列的文件> && git commit -F <正文文件>") print(" 提交后重跑 `done {}` 完成清场。".format(a.task_id)) return EXIT_FAIL print("=== 清场 ===") teardown_worktree(st, force=a.force, base=base) print() print("完成。主干 {} 现为 {}。".format(base, base_after[:12])) print("push 需人工许可，本次未推送。") return EXIT_OK # ------------------------------------------------- 树内预检 / 合并后补绿（机械化） HYGIENE_BINARY = (".png", ".jpg", ".gif", ".ico", ".ttf", ".bin", ".hex", ".elf", ".axf", ".o", ".crf", ".lst", ".zip", ".gz", ".pdf", ".cur") CR_EXEMPT = (".bat", ".cmd") # .gitattributes 强制 CRLF def _git_bytes(*args, cwd=None): r = subprocess.run(["git"] + [str(x) for x in args], cwd=str(cwd or REPO_ROOT), capture_output=True, stdin=subprocess.DEVNULL) return r.stdout if r.returncode == 0 else None def _changed_paths(cwd: Path): q = ("-c", "core.quotePath=false") # 中文文件名不许被八进制转义出去 tracked = git(*q, "diff", "--name-only", "HEAD", cwd=cwd, check=False).stdout others = git(*q, "ls-files", "--others", "--exclude-standard", cwd=cwd, check=False).stdout return ([p.strip() for p in tracked.splitlines() if p.strip()], [p.strip() for p in others.splitlines() if p.strip()]) def cmd_precheck(a) -> int: """树内自查（只读）：字节体检 + docs 逐档门禁 + 可选提交信息检查。绝不代改。""" cwd = Path.cwd() (cwd / "ci-artifacts").mkdir(exist_ok=True) # 新树没有 ci-artifacts（G 节：ENOENT 坑） tracked, untracked = _changed_paths(cwd) print("[i] {}：改动 {} 个（tracked {} + untracked {}）".format( cwd, len(tracked) + len(untracked), len(tracked), len(untracked))) bom3, cr, fffd = bytes.fromhex("efbbbf"), b"\r", bytes.fromhex("efbfbd") problems, docs = 0, [] for p in tracked + untracked: f = cwd / p if not f.is_file() or p.lower().endswith(HYGIENE_BINARY): continue b = f.read_bytes() bad = [] if cr in b and not p.lower().endswith(CR_EXEMPT): bad.append("CR") n = b.count(fffd) if n: bad.append("U+FFFD x{}".format(n)) if p in tracked: hb = _git_bytes("show", "HEAD:" + p, cwd=cwd) if hb is not None and hb.startswith(bom3) and not b.startswith(bom3): bad.append("BOM 漂移（HEAD 有 BOM、工作区没了——bom-guard 会拦）") if p.startswith("docs/") and p.endswith(".md"): docs.append(p) if bad: problems += 1 print("!! {}：{}".format(p, " / ".join(bad))) for d in docs: if not run_script("scripts/check_doc_links.py", [d], cwd, "文档门禁 " + d, tail=6): problems += 1 if a.msgfile: m = Path(a.msgfile) if not m.is_file(): raise EnvError("消息文件不存在：{}".format(m)) mb = m.read_bytes() if mb.startswith(bom3): problems += 1 print("!! 消息带 BOM（commit -F 要 UTF-8 无 BOM）") if cr in mb: problems += 1 print("!! 消息含 CRLF") ml = mb.decode("utf-8", "replace").splitlines() for i, ln in enumerate(ml): lim = 100 if i == 0 else 80 if len(ln) > lim: problems += 1 print("!! 消息第 {} 行长 {} > {}".format(i + 1, len(ln), lim)) break for w in a.ban: c = mb.decode("utf-8", "replace").lower().count(w.lower()) print("[i] 禁词 {!r}：{} 次".format(w, c)) if c: problems += 1 print("完成：问题 {} 个".format(problems)) return EXIT_OK if problems == 0 else EXIT_FAIL def cmd_postmerge(a) -> int: """主树合并后补绿：体积凭证自动诊断（短链/全链）+ 文档全量。列文件，不提交。 裁定外包给现有门禁：凭证红 → 先短链（--update 重写）→ 用只读的 check_mem_numbers 复检裁定数字是否真变——红了才升级全链（--write 族）。 """ main = main_worktree() if Path.cwd().resolve() != main.resolve(): raise EnvError("postmerge 必须在主工作树跑：{}".format(main)) if not list((main / "MDK-ARM").glob("**/*.map")): raise EnvError("主树没有 .map——先 /keil-build 0/0 再回来（C.5 的前提）") gate = "scripts/check_size_gate.py" if not a.force and run_script(gate, ["--local"], main, "体积凭证检查（同码）", tail=8): print("[ok] 凭证同码新鲜，无需补链") else: run_script("scripts/ram_baseline.py", ["--update"], main, "短链：重写体积凭证", tail=6) mem_ok = (a.skip_mem or run_script("scripts/check_mem_numbers.py", [], main, "受管区一致性复检（只读）", tail=8)) if not mem_ok: print("[i] 快照与 .map 真漂了 → 升级全链（数字变了才动受管区）") run_script("scripts/check_mem_numbers.py", ["--write"], main, "同步内存快照") run_script("scripts/ram_baseline.py", ["--write"], main, "同步 RAM 预算台账") if a.coverage: run_script("scripts/check_test_coverage.py", ["--update-baseline"], main, "覆盖率基线刷新", tail=6) if not run_script(gate, ["--local"], main, "凭证复检（期望绿）", tail=8): print("!! 凭证复检仍红——贴输出查 check_size_gate 原因，别硬提交") return EXIT_FAIL ok = True if not a.skip_docs: ok = run_script("scripts/check_doc_links.py", ["--all"], main, "文档全库检查", tail=8) dirty = git("-c", "core.quotePath=false", "diff", "--name-only", "HEAD", cwd=main, check=False).stdout.splitlines() dirty = [x.strip() for x in dirty if x.strip()] if dirty: print() print("[i] 回填产物（提交属 Ask first，本命令不代提交）：") for p in dirty: print(" - " + p) print(" 建议：git add <上面列的文件> && git commit -F <正文文件>") return EXIT_OK if ok else EXIT_FAIL # --------------------------------------------------------------------- 入口 def build_parser(): ap = argparse.ArgumentParser( prog="task_worktree.py", description="多任务并行开发的空间管理（git worktree 生命周期）") sub = ap.add_subparsers(dest="cmd") # bench-gate：检查分支是否有 bench-pass 记录 p = sub.add_parser("bench-gate", help="检查指定分支是否有 bench-pass 记录") p.add_argument("branch", help="要检查的分支名") p.set_defaults(func=cmd_bench_gate) # merge：在主树执行编排式合并 p = sub.add_parser("merge", help="在主树执行编排式合并（bench-gate → 预演 → 合并）") p.add_argument("branch", help="要合并的分支名") p.add_argument("--skip-bench", action="store_true", help="跳过 bench-gate 检查") p.set_defaults(func=cmd_merge) p = sub.add_parser("list", help="列出所有任务空间") p.set_defaults(func=cmd_list) p = sub.add_parser("status", help="单个任务详情") p.add_argument("task_id") p.set_defaults(func=cmd_status) p = sub.add_parser("sync", help="rebase 追平主干") p.add_argument("task_id") p.add_argument("--base", default=None) p.set_defaults(func=cmd_sync) p = sub.add_parser("lock", help="独占资源锁（build / bench）") p.add_argument("action", choices=["acquire", "release", "status"]) p.add_argument("name", nargs="?", choices=list(LOCK_NAMES)) p.add_argument("--task", default=None) p.add_argument("--wait", type=int, default=0, help="等待秒数（0=不等待）") p.add_argument("--force", action="store_true") p.add_argument("--note", default="") p.add_argument("--ttl", type=float, default=None, help="本次持有自报的陈旧窗口（小时）；缺省用该锁默认值（build 1h / bench 6h）") p.add_argument("--bench-pass", action="store_true", help="release bench 时顺带为该任务记录台架 PASS") p.set_defaults(func=cmd_lock) p = sub.add_parser("plugin", help="DSH 插件树（.agents/worktrees/，兼容旧路径）：list / adopt / unadopt") p.add_argument("action", choices=["list", "adopt", "unadopt"]) p.add_argument("name", nargs="?", default="", help="插件树名；可用裸名，或带 .agents/worktrees/ 与旧 .dsh-worktrees/worktree/ 前缀消歧") p.set_defaults(func=cmd_plugin) p = sub.add_parser("done", help="[DEPRECATED] 已废弃，请用 merge 命令") p.add_argument("task_id") p.add_argument("--base", default=None) p.add_argument("--dry-run", action="store_true", help="只预演，不做任何改动") p.add_argument("--skip-build", action="store_true", help="跳过 /keil-build") p.add_argument("--skip-gates", action="store_true", help="跳过全部门禁") p.add_argument("--force", action="store_true", help="清场时强制移除工作树") p.set_defaults(func=cmd_done) p = sub.add_parser("precheck", help="树内自查（只读）：字节体检 + docs 逐档门禁 + 可选消息检查") p.add_argument("--msgfile", default=None, help="附带检查提交信息文件（BOM/CRLF/行长/禁词）") p.add_argument("--ban", action="append", default=[], help="禁词，可重复") p.set_defaults(func=cmd_precheck) p = sub.add_parser("postmerge", help="主树合并后补绿：体积凭证自动诊断（短/全链）+ 文档全量；不代提交") p.add_argument("--force", action="store_true", help="凭证新鲜也强制走一遍 --update") p.add_argument("--coverage", action="store_true", help="附带覆盖率基线刷新") p.add_argument("--skip-docs", action="store_true", help="跳过 check_doc_links --all") p.add_argument("--skip-mem", action="store_true", help="只挪凭证，跳过受管区复检/全链") p.set_defaults(func=cmd_postmerge) return ap def main(argv=None) -> int: ap = build_parser() a = ap.parse_args(argv) if not getattr(a, "func", None): ap.print_help() return EXIT_USAGE if not (REPO_ROOT / ".git").exists(): print("!! 无法定位仓库根（推导为 {}）".format(REPO_ROOT), file=sys.stderr) return EXIT_ENV try: return a.func(a) except EnvError as e: print("前置/环境错误：{}".format(e), file=sys.stderr) return EXIT_ENV except Fail as e: print("失败：{}".format(e), file=sys.stderr) return EXIT_FAIL except KeyboardInterrupt: print("已中断", file=sys.stderr) return EXIT_FAIL if __name__ == "__main__": sys.exit(main())