443 lines
18 KiB
Python
443 lines
18 KiB
Python
#!/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())
|