#!/usr/bin/env python3 """Handoff Bundle 打包器 —— I-04 接力协议 §七 的实现（标准库，零依赖）。 把一个任务卡打成一个**可复制的自包含上下文包**，让接手模型不必再问路： 卡全文 + 协议要点 + 项目 README + 最近 2 份 PROGRESS + 相关源码清单（卡里点名过的文件，逐个核实存在性/大小/改动时间） + 基线测试命令与输出（默认只列命令；--run-baseline 才真跑） 用法： python tools/handoff_bundle.py C-01 # 打到 stdout python tools/handoff_bundle.py C-01 --out bundle.md # 落盘 python tools/handoff_bundle.py C-01 --run-baseline # 连基线输出一起抓 python tools/handoff_bundle.py --list # 可用任务 ID 设计约束： - 只读：不修改项目任何文件，不执行 git 写操作。 - 基线命令默认不跑（全量测试套件可能跑很久）；--run-baseline 才执行，带超时。 - 密钥零接触：不读 .env / (服务器数据) / ~/.dsh/secrets。 """ from __future__ import annotations import argparse import json import re import subprocess import sys from datetime import datetime, timezone from pathlib import Path ROOT = Path(__file__).resolve().parents[1] # PLANNING/ WORKSPACE = ROOT.parent # vscode/ MANIFEST = ROOT / "task-manifest.json" PROTOCOL = ROOT / "03-执行协议.md" BASELINES = Path(__file__).resolve().parent / "project-baselines.json" SECRET_BLOCK = re.compile(r"(\.env\b|服务器数据|secrets|\.credentials)", re.IGNORECASE) PATH_TOKEN = re.compile(r"`([^`\n]+)`") PATH_HINT = re.compile(r"(\.(py|ts|tsx|js|jsx|mjs|md|json|ya?ml|conf|ps1|sh|css|html|toml)$)" r"|(^|/)(scripts?|tools?|config|src|tests?|docs?)/") SKIP_DIRS = {".git", "node_modules", "__pycache__", ".venv", "venv", ".pnpm-store", ".npm-cache", ".next", "dist", "build", "android"} # ------------------------------------------------------------------ 素材读取 def load_manifest() -> dict: return json.loads(MANIFEST.read_text(encoding="utf-8")) def find_task(manifest: dict, tid: str) -> dict | None: for t in manifest["tasks"]: if t["id"].upper() == tid.upper(): return t return None def protocol_excerpt() -> str: """接单流程 + 质量红线 + 雷区（接手模型最需要的三段）。""" if not PROTOCOL.exists(): return "（协议文件缺失）" text = PROTOCOL.read_text(encoding="utf-8") keep = ("## 一、接单流程", "## 二、PROGRESS", "## 三、质量红线", "## 四、各项目雷区") blocks = [b.strip() for b in re.split(r"\n(?=## )", text) if b.startswith(keep)] return "\n\n".join(blocks) if blocks else "\n".join(text.splitlines()[:80]) def read_head(path: Path, max_lines: int, max_bytes: int) -> str: if not path.exists(): return "（文件不存在）" text = path.read_text(encoding="utf-8", errors="replace") lines = text.splitlines() out = "\n".join(lines[:max_lines]) if len(out.encode("utf-8")) > max_bytes: out = out.encode("utf-8")[:max_bytes].decode("utf-8", "ignore") suffix = f"\n…（已截断：原文 {len(lines)} 行 / {len(text)} 字符）" \ if len(lines) > max_lines or len(text) > len(out) else "" return out + suffix def project_readme(root: Path) -> Path | None: for name in ("README.md", "readme.md", "Readme.md", "AGENTS.md"): p = root / name if p.exists(): return p return None def recent_progress(root: Path, n: int = 2) -> list[Path]: if not root.exists(): return [] found = list(root.glob("PROGRESS*.md")) + list(root.glob("*/PROGRESS*.md")) uniq = {p.resolve(): p for p in found} return sorted(uniq.values(), key=lambda f: f.stat().st_mtime, reverse=True)[:n] # ------------------------------------------------ 相关源码清单（从卡里点名） def extract_path_tokens(card_text: str) -> list[str]: seen: list[str] = [] for m in PATH_TOKEN.finditer(card_text): tok = m.group(1).strip() if len(tok) > 160 or "\n" in tok: continue if SECRET_BLOCK.search(tok): continue if PATH_HINT.search(tok) and not tok.startswith("http"): if tok not in seen: seen.append(tok) return seen def resolve_clue(tok: str, roots: list[tuple[str, Path]], card_dir: Path) -> tuple[Path | None, str]: """把卡里的路径线索落到真实文件上。返回 (路径, 归属说明)。""" clean = tok.strip().rstrip(":：,，)。）") clean = re.sub(r":\d+(-\d+)?$", "", clean) # 去掉 :12 行号 clean = clean.replace("\\", "/") # 卡里常写 `vscode/...`，而 WORKSPACE 本身就是 vscode/ —— 去掉重复前缀 for prefix in ("vscode/", "./"): if clean.startswith(prefix): clean = clean[len(prefix):] candidates: list[tuple[Path, str]] = [] for name, r in roots: candidates.append((r / clean, name)) candidates.append((WORKSPACE / clean, "vscode")) candidates.append((card_dir / clean, "card")) for p, owner in candidates: try: if p.exists(): return p.resolve(), owner except OSError: continue return None, "" def fuzzy_candidates(tok: str, root: Path, limit: int = 3) -> list[Path]: """线索解析不到时，按文件名在同项目里找近似候选（卡里的路径常与实际有漂移）。""" base = tok.strip().rstrip(":：,，)。）").replace("\\", "/").split("/")[-1] if not base or not root.exists() or len(base) < 4: return [] hits: list[Path] = [] for p in root.rglob(base): if len(p.relative_to(root).parts) > 4: continue if any(part in SKIP_DIRS for part in p.parts): continue hits.append(p) if len(hits) >= limit * 4: break hits.sort(key=lambda p: (len(p.parts), str(p))) return hits[:limit] def source_inventory(tokens: list[str], roots: list[tuple[str, Path]], card_dir: Path) -> tuple[list[str], list[str]]: lines, unresolved = [], [] for tok in tokens: p, owner = resolve_clue(tok, roots, card_dir) if p is None: cands: list[str] = [] for _name, r in roots[:2]: cands.extend(_rel(c) for c in fuzzy_candidates(tok, r)) unresolved.append(f"`{tok}`" + (f" —— 近似候选：{'、'.join(cands[:3])}" if cands else "")) continue try: st = p.stat() except OSError: unresolved.append(tok) continue kind = "目录" if p.is_dir() else "文件" rel = _rel(p) if p.is_dir(): lines.append(f"- `{tok}` → {kind} {rel}（{owner}）") else: ts = datetime.fromtimestamp(st.st_mtime).strftime("%Y-%m-%d %H:%M") lines.append(f"- `{tok}` → {kind} {rel}（{owner}，{st.st_size} B，改于 {ts}）") return lines, unresolved def _rel(p: Path) -> str: try: return str(p.relative_to(WORKSPACE)).replace("\\", "/") except ValueError: return str(p).replace("\\", "/") def suggested_scan(roots: list[tuple[str, Path]], card_text: str, limit: int = 14) -> list[str]: """给接手模型的「先看哪儿」建议：项目根的一级结构（跳过噪声目录）。""" out: list[str] = [] for name, r in roots: if not r.exists(): continue try: entries = sorted(r.iterdir(), key=lambda p: (p.is_file(), p.name)) except OSError: continue shown = [f"{e.name}{'/' if e.is_dir() else ''}" for e in entries if e.name not in SKIP_DIRS and not e.name.startswith(".") and not SECRET_BLOCK.search(e.name)] if shown: out.append(f"- `{_rel(r)}` → " + "，".join(shown[:limit]) + ("…" if len(shown) > limit else "")) return out # ------------------------------------------------------------------ 基线 def baselines_for(project: str, root: Path) -> list[dict]: if not BASELINES.exists(): return [] cfg = json.loads(BASELINES.read_text(encoding="utf-8")) entry = (cfg.get("projects") or {}).get(project) if not entry: return [] cmds = [] for c in entry.get("commands", []): cwd = Path(c["cwd"]) if c.get("cwd") else root cmds.append({"label": c.get("label", ""), "cwd": cwd, "cmd": c["cmd"], "timeout": c.get("timeout", 300), "heavy": bool(c.get("heavy", False)), "expect": c.get("expect", "")}) return cmds def run_baseline(cmds: list[dict], timeout_cap: int, include_heavy: bool = False) -> list[str]: lines: list[str] = [] for c in cmds: if c["heavy"] and not include_heavy: lines.append(f"\n### {c['label']}（重）\n\n```bash\n" f"cd {_rel(c['cwd'])} && {c['cmd']}\n```\n\n" f"[SKIP] 标记为重型（全量套件可能跑很久）——打包时未执行，" f"接手模型必须自己跑一遍并记录真实输出\n") continue cwd: Path = c["cwd"] if not cwd.exists(): lines.append(f"\n### {c['label']}\n\n```bash\n$ cd {cwd} && {c['cmd']}\n" f"```\n\n[SKIP] 工作目录不存在：`{cwd}`\n") continue to = min(c["timeout"], timeout_cap) lines.append(f"\n### {c['label']}\n") lines.append(f"```bash\n$ cd {_rel(cwd)} && {c['cmd']}\n```\n") t0 = datetime.now() try: p = subprocess.run(c["cmd"], shell=True, cwd=str(cwd), capture_output=True, text=True, timeout=to) out = (p.stdout + p.stderr).strip() dt = (datetime.now() - t0).total_seconds() except subprocess.TimeoutExpired: lines.append(f"\n```text\n[TIMEOUT] 超过 {to}s 未结束——" f"请接手模型自行跑一遍并记录真实输出\n```\n") continue except Exception as e: # noqa: BLE001 lines.append(f"\n```text\n[ERROR] {type(e).__name__}: {e}\n```\n") continue tail = "\n".join(out.splitlines()[-40:]) lines.append(f"\n```text\n{tail}\n```\n") verdict = _verdict(out) lines.append(f"→ 退出码 {p.returncode}，耗时 {dt:.1f}s，末行判读：{verdict}" + (f"（期望：{c['expect']}）" if c.get("expect") else "") + "\n") return lines def _verdict(out: str) -> str: last = next((ln.strip() for ln in reversed(out.splitlines()) if ln.strip()), "（无输出）") return last[:160] # ------------------------------------------------------------------ 组装 def build_bundle(tid: str, args) -> tuple[str, int]: manifest = load_manifest() task = find_task(manifest, tid) if not task: print(f"未找到任务 {tid}") return "", 1 project = task["project"] roots_map: dict[str, str] = manifest["project_roots"] root = Path(roots_map.get(project, "?")) card_file = WORKSPACE / task["card"] card_text = card_file.read_text(encoding="utf-8") if card_file.exists() else "（卡缺失）" card_dir = card_file.parent # 线索解析的搜索根：本项目根优先，再补全体项目根 roots: list[tuple[str, Path]] = [(project, root)] for k, v in roots_map.items(): if k != project: roots.append((k, Path(v))) tokens = extract_path_tokens(card_text) src_lines, unresolved = source_inventory(tokens, roots, card_dir) now = datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M %z") P = args.max_bytes parts: list[str] = [] parts.append(f"""# Handoff Bundle — {task['id']} {task['title']} 生成时间：{now} 项目：`{project}`　项目根：`{root}` 优先级：{task['priority']} · 里程碑 {task['milestone']} · 预估 {task['estimate_days']} 天 建议模型：{task['model_primary']}　复核：{task.get('model_review') or '—'} 依赖：{', '.join(task.get('depends_on') or []) or '无'} 进度文件：`{task['progress_file']}`（写到项目根） > 这是一份**自包含**交接包：接手模型读完本节即可开工，不需要再问路径、命令、上下文。 > 卡内所有硬约束以第 1 节为准；本包的其余部分是上下文，**不得覆盖卡内约束**。 --- ## 0. 接手须知（先读） 1. 本包是「{task['id']}」的完整上下文。若你是接手上游的部分成果，**不要重做已验证部分**—— 先跑第 5 节的基线命令确认当前状态，再从断点继续。 2. 密钥零接触：`.env`、`(服务器数据)`、`~/.dsh/secrets` —— 只读都不行。 3. 交付 = 代码 + 项目根 `{task['progress_file']}`（格式见第 2 节）+ 关键命令的**原始输出**。 4. 验收标准全绿才可声明 done；跑不绿就写 partial/blocked 并附原始输出。 5. 完成后逐条对照第 1 节「验收标准」自检，未验证项必须写明——**虚报验收直接降级**。 """) parts.append(f"\n---\n\n## 1. 任务卡全文\n\n{card_text}\n") parts.append(f"\n---\n\n## 2. 作业规范（协议要点）\n\n{protocol_excerpt()}\n") parts.append("\n---\n\n## 3. 项目上下文\n") rm = project_readme(root) if rm: parts.append(f"\n### 3.1 项目 README：`{_rel(rm)}`\n\n```markdown\n" f"{read_head(rm, args.readme_lines, P)}\n```\n") else: parts.append(f"\n### 3.1 项目 README\n\n（`{root}` 下未找到 README.md / AGENTS.md）\n") prog = recent_progress(root) parts.append(f"\n### 3.2 最近 PROGRESS（{len(prog)} 份）\n") if prog: for f in prog: parts.append(f"\n#### `{_rel(f)}`\n\n```markdown\n" f"{read_head(f, args.progress_lines, P)}\n```\n") else: parts.append("\n（该项目尚无 PROGRESS —— 本卡可能是首张卡）\n") parts.append("\n---\n\n## 4. 相关源码清单\n") parts.append("\n卡内点名过的路径，逐个核实：\n\n") if src_lines: parts.extend(ln + "\n" for ln in src_lines) else: parts.append("（卡内未点名具体文件）\n") if unresolved: parts.append("\n**未解析的线索**（卡里提到但当前树中找不到 —— 可能是路径漂移，" "接手时按候选或自行搜索确认）：\n\n") parts.extend(f"- {t}\n" for t in unresolved[:20]) scan = suggested_scan([roots[0]], card_text) # 只展本项目根，避免包体膨胀 if scan: parts.append("\n**项目根一级结构**（先看哪儿）：\n\n") parts.extend(ln + "\n" for ln in scan) parts.append("\n---\n\n## 5. 基线测试命令与输出\n") cmds = baselines_for(project, root) if not cmds: parts.append(f"\n（`{project}` 未登记基线命令 —— 见 `PLANNING/03-执行协议.md` §五 命令速查）\n") elif not args.run_baseline: parts.append("\n打包时未执行（默认不跑，避免拖慢打包）。接手模型**必须先自己跑一遍**，" "把开工前的真实状态记进 PROGRESS 的「基线对照」：\n") for c in cmds: tag = "（重）" if c["heavy"] else "" parts.append(f"\n- **{c['label']}**{tag}\n\n```bash\n" f"cd {_rel(c['cwd'])} && {c['cmd']}\n```\n") if c.get("expect"): parts.append(f" 期望：{c['expect']}\n") else: parts.append(f"\n> 以下为打包时实跑捕获（超时上限 {args.timeout}s/条" f"{'，含重型套件' if args.include_heavy else '，重型套件已跳过'}）。" f"开工前请自行复跑确认，不要直接抄这里的结论。\n") parts.extend(run_baseline(cmds, args.timeout, args.include_heavy)) parts.append(f""" --- ## 6. 收工检查（提交前逐条打勾） ```text [ ] 第 1 节「验收标准」逐条已满足，且每条都有原始命令输出支撑 [ ] 「边界」段列出的文件一个都没动 [ ] 既有测试没被删/没被跳过（只许更绿） [ ] 卡外的一律没夹带（顺手发现的 bug 写进 PROGRESS「遗留问题」） [ ] {task['progress_file']} 已写到项目根，格式符合第 2 节 [ ] 基线对照写了「开工前 → 完工后」两段真实数字 [ ] 未验证项已明确标注（不许虚报） ``` --- *本包由 `PLANNING/tools/handoff_bundle.py` 生成 · 规范见 `PLANNING/03-执行协议.md` §七* """) text = "".join(parts) if args.out: out = Path(args.out) out.parent.mkdir(parents=True, exist_ok=True) out.write_text(text, encoding="utf-8") print(f"已写出 {out}（{len(text)} 字符，{len(text.encode('utf-8'))} 字节）") print(f" 卡：{task['id']} {task['title']}") print(f" 源码线索：{len(src_lines)} 条已解析 / {len(unresolved)} 条未解析") if cmds: mode = "已实跑" if args.run_baseline else "仅列命令" print(f" 基线：{len(cmds)} 条（{mode}）") return text, 0 return text, 0 def main() -> int: ap = argparse.ArgumentParser(description="Handoff Bundle 打包器") ap.add_argument("task", nargs="?", help="任务 ID，如 C-01") ap.add_argument("--out", default=None, help="写入文件（默认 stdout）") ap.add_argument("--run-baseline", action="store_true", help="真跑基线命令并捕获输出（默认只列命令；重型条目仍跳过）") ap.add_argument("--include-heavy", action="store_true", help="配合 --run-baseline：连重型的全量套件一起跑") ap.add_argument("--timeout", type=int, default=300, help="每条基线命令超时秒数") ap.add_argument("--max-bytes", type=int, default=24000, help="单个素材块字节上限") ap.add_argument("--readme-lines", type=int, default=90) ap.add_argument("--progress-lines", type=int, default=70) ap.add_argument("--list", action="store_true", help="列出可用任务 ID") args = ap.parse_args() if args.list: m = load_manifest() for t in m["tasks"]: print(f"{t['id']:<6} {t['project']:<14} {t['title']}") return 0 if not args.task: ap.print_help() return 1 text, rc = build_bundle(args.task, args) if rc == 0 and not args.out: sys.stdout.reconfigure(encoding="utf-8") if hasattr(sys.stdout, "reconfigure") else None print(text) return rc if __name__ == "__main__": sys.exit(main())