workflow.py

"""实验流程层：四件套脚手架 + 生命周期状态机 + 一致性体检 + 原子收尾。

把两个实验仓库验证过的纪律固化成代码，新仓库 `explab init` 即可套用：

四件套：
- docs/STATE.md          唯一权威状态源（冠军数字/在跑/黑名单/已知坑）
- docs/LEDGER.md         每实验一行状态标签
- experiments/QUEUE.md   实验队列表（领实验的唯一入口）
- experiments/runs/*.md  每实验一份运行记录（RUN_TEMPLATE.md 复制而来）

状态机：todo -> running -> done / dead / blocked。
防崩坏机制：
- start 前队列行必须存在且状态合法，run 记录不存在才允许开工
- finish 一次性改完 QUEUE + run 记录再统一校验（四件套同一次操作更新）
- check 校验：四件套在场、根目录无散落文件、队列与记录一一对应、
  记录里引用的路径必须真实存在、blocked 行必须在 notes 写明卡住原因、
  done/dead 的 Decision 证据指针（Evidence 行）引用必须存在
- check 告警（不阻塞）：done/dead 缺证据指针、running 超 7 天无记录更新
项目特性（黑名单、口径规则、进程核对）不进 explab——留在各项目自己的
check_state.py / AGENTS.md，explab 只管通用纪律。
"""
import datetime as _dt
import re
from pathlib import Path

import yaml

__all__ = ["WorkflowError", "load_manifest", "init_project", "check_state",
           "start_run", "finish_run", "STATUSES", "render_agents_md",
           "upgrade_agents_md", "upgrade_project"]

STATUSES = ("todo", "running", "done", "dead", "blocked")
MANIFEST_NAME = "explab.yaml"

_DEFAULT_MANIFEST = {
    "state": "docs/STATE.md",
    "ledger": "docs/LEDGER.md",
    "queue": "experiments/QUEUE.md",
    "runs_dir": "experiments/runs",
    "run_template": "experiments/RUN_TEMPLATE.md",
    "configs_dir": "configs",
    "provenance": "docs/PROVENANCE.md",
    "third_party_dir": "third_party",
    # 根目录禁止散落的文件后缀（仓库根只放 README/AGENTS 等文档）
    "clutter_exts": [".yaml", ".yml"],
}

_RUN_TEMPLATE = """# {id} —— <一句话目的>

## Metadata

| 字段 | 值 |
|---|---|
| ID | `{id}` |
| Owner | `<agent/user>` |
| Status | `planned` |
| Started | `<YYYY-MM-DD HH:MM>` |
| Finished | `` |
| Machine | `<本机/服务器名——判定该实验状态时必须核对的环境>` |
| Queue row | `experiments/QUEUE.md::{id}` |

## Question

这次只回答一个问题：

> <写清楚要验证的假设，不要把多个实验揉在一起。>

## Protocol

| 项 | 值 |
|---|---|
| Config | `<configs/...yaml>` |
| Code change | `<commit/diff/none>` |
| Data split | `<子集/全量/物体或动作列表>` |
| Metrics | `<主指标>` |
| Baseline | `<必须可追溯到 STATE/LEDGER>` |
| Success line | `<达到什么才算有效>` |

## Commands

```bash
# 逐条写实际命令；不要写“同上”
```

## Live Log

- `<time>`：<启动/中断/恢复/异常/观察>

## Result

| 指标 | baseline | this run | delta | note |
|---|---:|---:|---:|---|
|  |  |  |  |  |

## Decision

- 结论：`keep/reject/retry/blocked`
- 原因：
- Evidence: `<结果文件/日志路径——结论必须可追溯到一个真实存在的文件>`
- 下一步：

## Sync Checklist

- [ ] `experiments/QUEUE.md` 状态已更新
- [ ] `docs/STATE.md` 冠军/在跑/下一步已更新
- [ ] `docs/LEDGER.md` 已新增或更新一行
- [ ] 结果文件路径写清楚
"""

_STATE_SKELETON = """# STATE —— 唯一权威状态源

## 冠军

| 字段 | 值 |
|---|---|
| 配置 | `<configs/...yaml>` |
| 主指标 | `<数字（必须带口径：子集/全量、含不含 TTA 等）>` |
| 复跑命令 | `<一行命令>` |

## 在跑

（无）

## 黑名单（不许重跑的路线 + 理由）

（无）

## 已知坑

（无）
"""

_LEDGER_SKELETON = """# LEDGER —— 每实验一行状态标签

| 轮次/配置 | 状态 | 结论 |
|---|---|---|
| `<config>` | `todo/running/done/dead` | `<一句话>` |
"""

_QUEUE_SKELETON = """# 实验队列

只从这里领实验。新增实验先加一行，状态从 `todo` 改 `running` 后才能开跑。

| ID | status | priority | config | run record | question | success line | notes |
|---|---|---:|---|---|---|---|---|
"""

_AGENTS_HEADER = """# AGENTS.md —— 实验 agent 操作协议

本仓库的实验流程由文件约定驱动，不依赖任何外部工具：四件套
（docs/STATE.md、docs/LEDGER.md、experiments/QUEUE.md、experiments/runs/*.md）
+ 单文件体检脚本 scripts/check_state.py（纯标准库，无需安装）。
新会话进来按顺序做，**不要跳步**：

1. 读 `docs/STATE.md`：冠军数字、在跑实验、黑名单、已知坑——唯一权威状态源
2. 读 `docs/LEDGER.md`：每个实验一行状态标签；判 `dead` 的路线不许重跑
3. 读 `experiments/QUEUE.md`：领实验的唯一入口
4. 跑 `python3 scripts/check_state.py`；失败先修流程，不要继续实验

> `<!-- explab:section ... -->` 标记之间的段落是 explab 托管段，
> 用 explab 工具升级模板时会替换它们。仓库特有的自定义约定
> 请写在标记之外的独立段落里，升级不会动。
"""

_AGENTS_SECTIONS = [
    ("先方案后入队", """## 先方案后入队（每个实验都要）

不论实验大小，先写方案再入队：

1. **写方案**：要回答什么问题、方法是什么、数据与口径、成功线
   （什么数字/现象算成立）、风险。方案写进该实验 run 记录的
   Question / Protocol 区域
2. **入队**：在 `experiments/QUEUE.md` 加一行 todo，question 与
   success line 两列填方案的一句话摘要版
3. **批准可选**：低风险（纯分析、小规模训练）可径行开工；高风险
   （长算力、改共享代码、改数据口径）在方案里写明风险点，等用户
   确认后再开工

空仓库也一样：方案需额外写清怎么搭这个仓库（任务定义/数据/模型/
目录结构），确认后按「构建期写代码纪律」搭建，搭完冒烟通过才进入
正常实验循环。"""),
    ("开一个实验", """## 开一个实验（三件事缺一不可）

1. QUEUE 里该实验行已存在且状态为 `todo`
2. 状态改 `running`；复制 `experiments/RUN_TEMPLATE.md` 到
   `experiments/runs/<实验ID>.md`，写清实际命令、成功线、基线出处
3. 以上没做完之前，不动代码与配置

禁止绕过队列直接跑；禁止复用旧配置改参数偷跑（账本会失去追溯）。"""),
    ("配置纪律", """## 配置纪律

- 新实验 = 新 yaml。配置支持 `base: <相对路径>` 继承：消融配置只写与
  主配置的差异，不要复制整份
- 不写代码不读的键：新增配置键之前先 grep 代码确认有人读它；
  半接线的键（yaml 里声明、代码里却写死值）比纯硬编码更危险
- 训练/评测脚本开头把解析后的完整配置快照存进 run 目录，
  复现不靠记忆"""),
    ("代码变更三类", """## 代码变更三类，处理方式不同

1. **调参数** → 只改 yaml，代码零改动
2. **同一可切换轴上的新方法、需要与旧版对比** → 新文件 + 登记进
   分发入口（dict 映射或工厂函数均可），旧实现一字不动
   （它还是对照基线）
3. **修复 / 共享数学 / 所有人都该用的改进** → 原地改，绝不新增变体。
   判断标准：旧版本是"值得对比的另一种做法"还是"错了的做法"？
   错了就原地修，不留档

同一个物理量/公式只允许一处实现。发现两处各写了一遍同样的数学，
视为 bug：抽成共享函数，不是"以后记得同步改"。"""),
    ("收尾", """## 收尾（四件套同一次操作更新）

1. QUEUE 行 `running`→`done`（判死用 `dead`，卡住用 `blocked`
   且必须在 notes 列写明卡住原因），run 记录的 Status 字段同步改、
   Finished 填完成时间戳
2. 同一次操作更新 `docs/STATE.md`（冠军/在跑/下一步）与
   `docs/LEDGER.md`（加一行或改状态）
3. run 记录里写清结果、口径、结论
4. done/dead 的结论必须带证据：run 记录 Decision 填 `Evidence:` 行
   （结果文件/日志路径），check_state 会验证引用真实存在——没有
   证据指针的结论不许进 STATE
5. 最后一条命令必须是 `python3 scripts/check_state.py`，失败就不算完成

防伪闭环：done/dead 记录的 Status 必须与 QUEUE 一致、必须有完成
时间戳——"完成"要留证明。done 还要过证据闸门：Result 表的
this run 列必须有可解析的数字指标，LEDGER 必须有对应行——
留不出数就是没跑完，账本没留痕就是没收尾。别手改 markdown 绕过
`explab finish`，失同步会被 check_state 当场揪出。

数字必须带口径（子集/全量、处理变体名等）——混口径 = 结论作废。
判 `dead` 的变体代码从分发登记里删掉，不是留着。"""),
    ("参考代码纪律", """## 参考代码纪律（从别人的代码起步时）

- 外部参考代码原样放 `third_party/<名字>/`，**钉死版本/commit，视为只读**：
  不许就地编辑。必须打补丁时用 patch 文件并登记进 `docs/PROVENANCE.md`
- 你的适配代码全部写在 `src/`，import third_party——"哪行是上游的、
  哪行是我的"永远有答案，上游修复来了能直接换版本重跑
- 每个 third_party 条目必须在 `docs/PROVENANCE.md` 登记：来源、版本、
  日期、用途、适配改动清单（`scripts/check_state.py` 会查）
- **动参考代码前先建立零点**：跑通它自带的测试，或复现它宣称的招牌数字，
  记进 run 记录。没有零点，之后坏了分不清是你改坏的还是本来就有病"""),
    ("构建期写代码纪律", """## 构建期写代码纪律

- **先搜后写**：写任何"仓库里概念上已经存在"的代码前，先搜现有实现，
  复用或在原处扩展；禁止照字面需求另起一份手写实现
- **同一物理量/公式只允许一处实现**：发现两处各写了一遍同样的数学，
  视为 bug，抽成共享函数——不是"以后记得同步改"，记不住才是常态
- **新增模块自带冒烟**：新文件要有最小可跑入口或测试，写完立刻跑通；
  "之后再补测试"= 永远不补
- **不留半成品**：跑不通的实验分支不合并进主线；暂时不用的代码直接删，
  历史有 git 管。死代码比没代码更危险——它会误导下一个读代码的人
- **依赖要显式**：新装第三方包必须登记进依赖清单（pyproject/requirements），
  不允许"我机器上有"式隐性依赖
- **报错给人看**：异常信息写清"哪里错了 + 下一步怎么办"，不写裸
  `raise Exception("error")`"""),
    ("Git 纪律", """## Git 纪律（仓库是 git 仓库时）

- **开工前确认工作区干净**（`git status`）：在脏树上跑实验，数字无法
  归属到任何一份代码，等于白跑
- **开工前若当前状态还没提交，先提交**：尤其是准备动共享代码/做大改
  之前——没提交就没有"回得去的过去"
- **提交粒度 = 一个语义变更**：一次 bug 修复、一次实验收尾（代码 +
  四件套同一个 commit）。不按时间、不按文件凑提交
- **四件套必须与产生它的代码同一次提交**：数字、记录、代码锁在一个
  commit 里，将来 checkout 任何一个 commit 都能还原那个数字的现场
- **提交信息格式**：`<实验ID或模块>: 一句话结论`，例如
  `opt-sgd-adam: 同预算 Adam 领先 1.2 个点，收敛快 30%`
- **不提交大文件**：数据 / 权重 / checkpoint / outputs 一律进
  .gitignore；产物在 run 记录里写路径引用，不进库
- **分支**：主线实验在主分支顺序推进即可；只有大重构/高风险改动
  （如架构迁移）才开分支
- **push 必须用户明确要求才做**；提交本身也只在用户要求或协议
  明确允许时执行"""),
]

_SECTION_BEGIN = "<!-- explab:section "
_SECTION_END = "<!-- /explab:section -->"


def _agents_blocks():
    """(段名, 正文, 是否允许在旧文件里追加)。头部只替换不追加——
    已有自己的 AGENTS.md 的仓库保留自己的开场白。"""
    yield "_header", _AGENTS_HEADER, False
    for name, body in _AGENTS_SECTIONS:
        yield name, body, True


def _section_block(name, body) -> str:
    return f"{_SECTION_BEGIN}{name} -->\n{body}\n{_SECTION_END}"


def render_agents_md() -> str:
    """生成完整 AGENTS.md（每个段落带 explab 托管标记）。"""
    return "\n\n".join(_section_block(n, b) for n, b, _ in _agents_blocks()) + "\n"


def _section_key(body) -> str:
    """取段落标题去掉括号注解后的关键词，用于旧文件里的同题检测。"""
    title = body.splitlines()[0].lstrip("#").strip()
    for sep in ("(", "（"):
        title = title.split(sep)[0].strip()
    return title


def _has_heading(text, key) -> bool:
    for line in text.splitlines():
        s = line.strip()
        if not s.startswith("#"):
            continue
        if _section_key(s) == key:
            return True
    return False


def upgrade_agents_md(root) -> dict:
    """把 AGENTS.md 的 explab 托管段更新到当前模板版本。

    只动 explab 标记包裹的段落：标记段被新版本替换；模板新增的段落
    追加到文件末尾（带标记）；旧文件里已有同题标题但没打标记的段落
    视为用户自写，跳过不动。文件不存在时整份写入。
    返回 {created, updated, added, skipped}。
    """
    p = _p(root, "AGENTS.md")
    if not p.exists():
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(render_agents_md(), encoding="utf-8")
        return {"created": True, "updated": [], "added": [], "skipped": []}
    text = p.read_text(encoding="utf-8")
    result = {"created": False, "updated": [], "added": [], "skipped": []}
    for name, body, append_missing in _agents_blocks():
        begin = f"{_SECTION_BEGIN}{name} -->"
        block = _section_block(name, body)
        pat = re.compile(re.escape(begin) + r".*?" + re.escape(_SECTION_END), re.S)
        if pat.search(text):
            new_text = pat.sub(lambda _m: block, text, count=1)
            if new_text != text:
                result["updated"].append(name)
            text = new_text
            continue
        if _has_heading(text, _section_key(body)):
            result["skipped"].append(name)
            continue
        if append_missing:
            if not text.endswith("\n"):
                text += "\n"
            text += "\n" + block + "\n"
            result["added"].append(name)
        else:
            result["skipped"].append(name)
    p.write_text(text, encoding="utf-8")
    return result


def _upgrade_check_script(root) -> str:
    """按托管标记契约更新 scripts/check_state.py。

    返回动作：created / updated / unchanged / skipped-customized。
    有托管标记行 = 未自定义，随模板替换；没有标记 = 用户自定义版，
    永远不碰。
    """
    p = _p(root, "scripts/check_state.py")
    if not p.exists():
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(_CHECK_SCRIPT, encoding="utf-8")
        return "created"
    text = p.read_text(encoding="utf-8")
    # 托管标记带版本号：历史上任何一版标记都算托管，随模板升级；
    # 只有完全没有标记的才是用户自定义版
    if not re.search(r"# explab-check-template v\d+", text):
        return "skipped-customized"
    if text != _CHECK_SCRIPT:
        p.write_text(_CHECK_SCRIPT, encoding="utf-8")
        return "updated"
    return "unchanged"


def upgrade_project(root) -> dict:
    """升级入口：AGENTS.md 托管段 + scripts/check_state.py（按标记契约）。

    不触碰任何其它文件，可用于已有仓库的安全升级。
    """
    r = upgrade_agents_md(root)
    r["check_script"] = _upgrade_check_script(root)
    return r


_CHECK_SCRIPT = '''#!/usr/bin/env python3
# explab-check-template v5 —— 托管标记行：explab upgrade 会随模板更新本脚本。
# 若你自定义了本脚本，删掉这一行，升级就永远不碰它。
"""流程体检（单文件、纯标准库、无需安装任何包）。

校验（失败退出码 1）：四件套在场 / 队列状态合法 / 单 running /
running 在 STATE 有记载 / blocked 行必须写明卡住原因 / 队列与记录
一一对应 / QUEUE config 列引用的配置真实存在（支持 {a,b} 与 * 展开）/
run 记录结构完整 / 记录引用路径真实存在（同样支持展开）/ done/dead 的
Decision 证据指针（Evidence 行）引用真实存在 / done/dead 防伪闭环
（记录 Status 与 QUEUE 一致、必须有完成时间戳）/ done 证据闸门
（Result 表必须有可解析数字指标、LEDGER 必须有对应行）/
config base 继承可达 / 根目录与 scripts、configs 无散落文件 /
third_party 已登记。
告警（不阻塞，退出码不变）：done/dead 缺证据指针 / running 超 7 天
无记录更新。
"""
import datetime as _dt
import re
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
REQUIRED = ["AGENTS.md", "docs/STATE.md", "docs/LEDGER.md",
            "experiments/QUEUE.md", "experiments/RUN_TEMPLATE.md"]
RUNS_DIR = ROOT / "experiments" / "runs"
STATUSES = ("todo", "running", "done", "dead", "blocked")
REQUIRED_RUN_HEADINGS = ["## Metadata", "## Question", "## Protocol",
                         "## Commands", "## Result", "## Decision"]
CLUTTER_EXTS = {".yaml", ".yml", ".ipynb", ".log"}
PATH_RE = re.compile(r"\\b((?:configs|scripts|experiments)/[^\\s`()（）]+)")
BRACE_RE = re.compile(r"^(.*)\\{([^{}]+)\\}(.*)$")
EVIDENCE_RE = re.compile(r"[-*]\\s*Evidence[:：]\\s*(.+)")
CAND_RE = re.compile(r"[\\w][\\w.\\-/]*\\.[A-Za-z0-9]{1,8}")
DATE_RE = re.compile(r"\\b(\\d{4}-\\d{2}-\\d{2})\\b")
METADATA_FIELD_RE = r"^\\|\\s*{field}\\s*\\|\\s*(.*?)\\s*\\|\\s*$"
STALE_DAYS = 7

errors = []
warnings = []


def expand_paths(token):
    """展开 {a,b} 与 * 通配符，返回相对路径列表；无通配时原样返回。"""
    m = BRACE_RE.match(token)
    if m:
        pre, items, post = m.groups()
        out = []
        for item in items.split(","):
            out.extend(expand_paths(pre + item + post))
        return out
    if "*" in token:
        return sorted(str(p.relative_to(ROOT)) for p in ROOT.glob(token))
    return [token]


def missing_expansions(rel):
    return [x for x in expand_paths(rel) if not (ROOT / x).exists()]


def parse_queue():
    rows = []
    queue = ROOT / "experiments/QUEUE.md"
    for line in queue.read_text(encoding="utf-8").splitlines():
        if not line.strip().startswith("|"):
            continue
        cells = [c.strip() for c in line.strip().strip("|").split("|")]
        if len(cells) < 5:
            continue
        rid = cells[0]
        if rid in ("ID", "") or set(rid) <= {"-", ":"}:
            continue
        record = cells[4].strip("`").strip() or f"experiments/runs/{rid}.md"
        notes = cells[7] if len(cells) > 7 else ""
        rows.append({"id": rid, "status": cells[1],
                     "config": cells[3] if len(cells) > 3 else "",
                     "record": record.split("::")[0].strip(),
                     "notes": notes.strip("`").strip()})
    return rows


def check_record_paths(record: Path):
    text = record.read_text(encoding="utf-8")
    for heading in REQUIRED_RUN_HEADINGS:
        if heading not in text:
            errors.append(f"{record.name} 缺少模板小节：{heading}")
    for raw in set(PATH_RE.findall(text)):
        rel = raw.rstrip(".,;:").split("::")[0]
        if "<" in rel or ">" in rel:
            continue
        missing = missing_expansions(rel)
        if missing:
            errors.append(f"{record.name} 引用了不存在的路径：{', '.join(missing)}")


def check_decision_evidence(record: Path, status: str):
    """Decision 的证据指针：done/dead 缺 Evidence 行记告警；
    Evidence 引用的路径不存在记错误（结论必须可追溯）。"""
    if status not in ("done", "dead"):
        return
    text = record.read_text(encoding="utf-8")
    idx = text.find("## Decision")
    if idx < 0:
        return
    m = EVIDENCE_RE.search(text[idx:])
    if not m:
        warnings.append(f"{record.name} 判 {status} 但 Decision 没写证据指针"
                        f"（Evidence 行）——结论不可追溯")
        return
    val = m.group(1).strip().strip("`").strip()
    if not val or "<" in val or ">" in val:
        warnings.append(f"{record.name} 的 Evidence 行还是模板占位，"
                        f"没填真实结果文件/日志路径")
        return
    for cand in CAND_RE.findall(val):
        cand = cand.strip("`\\\"'")
        if "/" not in cand or set("*{},") & set(cand):
            continue
        if not (ROOT / cand).exists():
            errors.append(f"{record.name} Evidence 引用了不存在的路径：{cand}")


def check_running_stale(record: Path, rid: str):
    """running 实验的记录超 7 天没有日期更新 → 告警（可能早已停跑）。"""
    dates = []
    for s in DATE_RE.findall(record.read_text(encoding="utf-8")):
        try:
            dates.append(_dt.datetime.strptime(s, "%Y-%m-%d").date())
        except ValueError:
            pass
    today = _dt.date.today()
    if not dates or (today - max(dates)).days > STALE_DAYS:
        latest = max(dates).isoformat() if dates else "无任何日期"
        warnings.append(f"running 实验 {rid} 已 {STALE_DAYS} 天没有记录更新"
                        f"（记录里最新日期 {latest}）——确认是否还在跑，"
                        f"否则改 blocked 并写明原因，或直接收尾")


def check_record_closure(record: Path, rid: str, status: str):
    """防伪闭环：done/dead 记录的状态字段必须与 QUEUE 一致，
    且必须带真实的完成时间戳——"完成"要留证明，不能只靠一句声明。"""
    text = record.read_text(encoding="utf-8")
    m = re.search(METADATA_FIELD_RE.format(field="Status"),
                  text, re.MULTILINE)
    if m:
        rec_status = m.group(1).strip().strip("`").strip()
        if rec_status != status:
            errors.append(f"QUEUE {rid} 状态 {status} 但记录 Metadata "
                          f"Status 是 {rec_status!r}——两处必须一致，"
                          f"用 explab finish 收尾，别只改一处")
    fm = re.search(METADATA_FIELD_RE.format(field="Finished"),
                   text, re.MULTILINE)
    val = fm.group(1).strip().strip("`").strip() if fm else ""
    if not val or "<" in val or ">" in val or not DATE_RE.search(val):
        errors.append(f"{record.name} 判 {status} 但 Metadata 没有真实完成"
                      f"时间戳（Finished 行）——完成必须留时间证明")


def check_done_metrics(record: Path, rid: str):
    """done 证据闸门：Result 表的 this run 列必须有至少一个数字——
    完成要留数，留不出数就是没跑完。"""
    text = record.read_text(encoding="utf-8")
    idx = text.find("## Result")
    if idx < 0:
        errors.append(f"{record.name} 判 done 但没有 ## Result 段")
        return
    section = text[idx:]
    nxt = section.find("\\n## ", 3)
    if nxt > 0:
        section = section[:nxt]
    n_metrics = 0
    for line in section.splitlines():
        line = line.strip()
        if not line.startswith("|"):
            continue
        cells = [c.strip() for c in line.strip("|").split("|")]
        if len(cells) < 3:
            continue
        if set(cells[0]) <= set("-: ") or cells[0] == "指标":
            continue
        if re.search(r"\\d", cells[2]):
            n_metrics += 1
    if n_metrics == 0:
        errors.append(f"{record.name} 判 done 但 Result 表里没有可解析的"
                      f"数字指标（this run 列）——完成必须留数")


def check_ledger_row(rid: str, status: str):
    """done/dead 证据闸门：LEDGER 必须有一行登记（第一列含 run id）——
    每个实验都要在账本留痕，不许只改 QUEUE。"""
    ledger = ROOT / "docs" / "LEDGER.md"
    if not ledger.exists():
        return  # 文件缺失已在四件套检查里报过
    for line in ledger.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line.startswith("|"):
            continue
        cells = [c.strip() for c in line.strip("|").split("|")]
        if cells and rid in cells[0]:
            return
    errors.append(f"QUEUE {rid} 判 {status} 但 LEDGER 没有对应行"
                  f"——每个实验都要在账本留痕")


def check_config_bases():
    for p in (ROOT / "configs").rglob("*.yaml"):
        if not p.is_file():
            continue
        for line_no, raw in enumerate(p.read_text(encoding="utf-8").splitlines(), 1):
            line = raw.strip()
            if not line.startswith("base:"):
                continue
            target = (p.parent / line.split(":", 1)[1].strip()).resolve()
            if not target.exists():
                errors.append(f"config base 断链：{p.relative_to(ROOT)}:{line_no} -> {target}")


def main():
    for rel in REQUIRED:
        if not (ROOT / rel).exists():
            errors.append(f"流程文件缺失：{rel}")
    if errors:
        return report()

    for f in ROOT.iterdir():
        if (f.is_file() and f.suffix in CLUTTER_EXTS
                and f.name != "explab.yaml"):
            errors.append(f"根目录散落文件：{f.name}（应放进 configs/ 或 docs/）")
    scripts_dir = ROOT / "scripts"
    for f in scripts_dir.iterdir():
        if f.is_file() and f.name not in ("check_state.py", "README.md"):
            errors.append(f"scripts/ 根散落文件：{f.name}（放进子目录，如 scripts/analysis/）")
    for f in (ROOT / "configs").glob("*.yaml"):
        errors.append(f"configs/ 根散落配置：{f.name}（按层放进子目录，如 configs/runs/）")

    check_config_bases()

    rows = parse_queue()
    seen = set()
    running = []
    state_text = (ROOT / "docs/STATE.md").read_text(encoding="utf-8")
    for r in rows:
        if r["id"] in seen:
            errors.append(f"QUEUE 重复 ID：{r['id']}")
        seen.add(r["id"])
        if r["status"] not in STATUSES:
            errors.append(f"QUEUE {r['id']}: 非法状态 {r['status']!r}"
                          f"（可选 {list(STATUSES)}）")
        if r["status"] == "blocked" and not r["notes"]:
            errors.append(f"QUEUE {r['id']}: 状态 blocked 但 notes 没写卡住原因"
                          f"（blocked 不许无声挂起）")
        if r["status"] == "running":
            running.append(r["id"])
            if r["id"] not in state_text:
                errors.append(f"running 实验 {r['id']} 未出现在 docs/STATE.md（STATE 与 QUEUE 失同步）")
        if r["status"] in ("running", "done", "dead"):
            cfg_tokens = [t for t in PATH_RE.findall(r["config"]) if t.startswith("configs/")]
            if not cfg_tokens:
                errors.append(f"QUEUE {r['id']}: config 列未引用任何 configs/ 路径，体检无法校验")
            for raw in cfg_tokens:
                rel = raw.rstrip(".,;:").split("::")[0]
                if "<" in rel or ">" in rel:
                    continue
                missing = missing_expansions(rel)
                if missing:
                    errors.append(f"QUEUE {r['id']}: config 列路径不存在：{', '.join(missing)}")
        p = ROOT / r["record"]
        if not p.exists():
            if r["status"] in ("running", "done", "dead"):
                errors.append(f"QUEUE {r['id']}: 状态 {r['status']} "
                              f"但 run 记录缺失 {r['record']}")
        else:
            check_record_paths(p)
            check_decision_evidence(p, r["status"])
            if r["status"] == "running":
                check_running_stale(p, r["id"])
            if r["status"] in ("done", "dead"):
                check_record_closure(p, r["id"], r["status"])
            if r["status"] == "done":
                check_done_metrics(p, r["id"])
        if r["status"] in ("done", "dead"):
            check_ledger_row(r["id"], r["status"])
    if len(running) > 1:
        errors.append("同一时刻只允许一个 running 实验：" + ", ".join(running))

    if RUNS_DIR.exists():
        ids = {r["id"] for r in rows}
        for f in sorted(RUNS_DIR.glob("*.md")):
            if f.stem not in ids:
                errors.append(f"孤儿 run 记录（QUEUE 里没有对应行）：{f.name}")

    tp = ROOT / "third_party"
    if tp.exists():
        prov = ROOT / "docs" / "PROVENANCE.md"
        if not prov.exists():
            errors.append("存在 third_party/ 但没有 docs/PROVENANCE.md 登记")
        else:
            prov_text = prov.read_text(encoding="utf-8")
            for entry in sorted(p for p in tp.iterdir()
                                if p.is_dir() and not p.name.startswith(".")):
                if entry.name not in prov_text:
                    errors.append(f"third_party/{entry.name} 未在 "
                                  f"PROVENANCE.md 登记（来源/版本/用途必须可追溯）")
    return report()


def report():
    if errors:
        print("STATE CHECK FAILED")
        for e in errors:
            print("- " + e)
    else:
        print("STATE CHECK OK")
    for w in warnings:
        print("⚠ " + w)
    return 1 if errors else 0


if __name__ == "__main__":
    sys.exit(main())
'''

CHECK_MARKER = "# explab-check-template v5"

_PROVENANCE_SKELETON = """# PROVENANCE —— 参考代码来源登记

third_party/ 里每一个外部代码条目都必须在这里登记。
纪律：third_party 内代码只读、钉死版本；适配写在 src/；
必须打补丁时用 patch 文件并在"适配/补丁"列登记。

| 名字 | 来源（URL/仓库） | 版本/commit | 引入日期 | 用途 | 适配/补丁 |
|---|---|---|---|---|---|
| `<目录名>` | `<url>` | `<commit或版本号>` | `<YYYY-MM-DD>` | `<拿它干什么>` | `<src 里的适配文件/patch 清单>` |
"""


class WorkflowError(RuntimeError):
    """流程违规：状态机非法 / 四件套缺失 / 引用路径不存在。"""


# ---------------------------------------------------------------- manifest
def load_manifest(root) -> dict:
    """读项目根 explab.yaml；缺省字段用默认值补齐。"""
    root = Path(root)
    m = dict(_DEFAULT_MANIFEST)
    p = root / MANIFEST_NAME
    if p.exists():
        with open(p) as f:
            user = yaml.safe_load(f) or {}
        m.update(user)
    return m


def _p(root, rel) -> Path:
    return Path(root) / rel


# ---------------------------------------------------------------- init
def init_project(root) -> list:
    """搭四件套脚手架。已存在的文件一律不覆盖（防崩坏），返回新建清单。"""
    root = Path(root)
    m = load_manifest(root)
    created = []

    def ensure(rel, content):
        p = _p(root, rel)
        if p.exists():
            return
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(content, encoding="utf-8")
        created.append(rel)

    ensure(MANIFEST_NAME,
           yaml.safe_dump(_DEFAULT_MANIFEST, allow_unicode=True, sort_keys=False))
    ensure("AGENTS.md", render_agents_md())
    ensure(m["state"], _STATE_SKELETON)
    ensure(m["ledger"], _LEDGER_SKELETON)
    ensure(m["provenance"], _PROVENANCE_SKELETON)
    ensure(m["queue"], _QUEUE_SKELETON)
    ensure(m["run_template"], _RUN_TEMPLATE.replace("{id}", "<实验ID>"))
    ensure("scripts/check_state.py", _CHECK_SCRIPT)
    runs_dir = _p(root, m["runs_dir"])
    if not runs_dir.exists():
        runs_dir.mkdir(parents=True)
        created.append(m["runs_dir"] + "/")
    cfg = _p(root, m["configs_dir"])
    if not cfg.exists():
        cfg.mkdir(parents=True)
        created.append(m["configs_dir"] + "/")
    return created


# ---------------------------------------------------------------- queue io
def _parse_queue(queue_path: Path):
    """解析队列表 → [{id, status, priority, record, raw, lineno}]。"""
    rows = []
    lines = queue_path.read_text(encoding="utf-8").splitlines()
    header = None
    for i, line in enumerate(lines, 1):
        s = line.strip()
        if not s.startswith("|"):
            continue
        cells = [c.strip() for c in s.strip("|").split("|")]
        if header is None:
            header = [c.lower() for c in cells]
            continue
        if set("".join(cells)) <= set("-: "):
            continue
        if len(cells) < 2:
            continue
        rid = cells[0].strip("`")
        status = cells[1].strip("`")
        record = ""
        notes = ""
        for h, c in zip(header, cells):
            if "record" in h and not record:
                record = c.strip().strip("`")
            if "note" in h and not notes:
                notes = c.strip().strip("`")
        rows.append({"id": rid, "status": status, "record": record,
                     "notes": notes, "raw": line, "lineno": i})
    return rows


def _replace_row(queue_path: Path, run_id: str, new_status: str):
    lines = queue_path.read_text(encoding="utf-8").splitlines()
    hit = False
    for i, line in enumerate(lines):
        s = line.strip()
        if s.startswith("|") and s.split("|")[1].strip().strip("`") == run_id:
            cells = s.strip("|").split("|")
            cells[1] = f" {new_status} "
            lines[i] = "|" + "|".join(cells) + "|"
            hit = True
            break
    if not hit:
        raise WorkflowError(f"QUEUE 里没有 ID={run_id} 的行")
    queue_path.write_text("\n".join(lines) + "\n", encoding="utf-8")


def _set_record_finished(record_path: Path):
    now = _dt.datetime.now().strftime("%Y-%m-%d %H:%M")
    text = record_path.read_text(encoding="utf-8")
    new = re.sub(r"(\|\s*Finished\s*\|\s*)`[^`]*`", rf"\1`{now}`", text, count=1)
    if new == text:
        raise WorkflowError(f"{record_path.name} 里找不到 Finished 行")
    record_path.write_text(new, encoding="utf-8")


def _set_record_status(record_path: Path, status: str):
    text = record_path.read_text(encoding="utf-8")
    new = re.sub(r"(\|\s*Status\s*\|\s*)`[^`]*`", rf"\1`{status}`", text, count=1)
    if new == text:
        raise WorkflowError(f"{record_path.name} 里找不到 Status 行")
    record_path.write_text(new, encoding="utf-8")


# ---------------------------------------------------------------- lifecycle
def start_run(root, run_id: str):
    """开工：QUEUE 行 todo→running + 从模板创建 run 记录。

    防崩坏：行必须存在且状态是 todo；run 记录不存在才允许创建。
    """
    m = load_manifest(root)
    queue = _p(root, m["queue"])
    rows = {r["id"]: r for r in _parse_queue(queue)}
    if run_id not in rows:
        raise WorkflowError(f"QUEUE 里没有 {run_id}——先加队列行再开工")
    row = rows[run_id]
    if row["status"] != "todo":
        raise WorkflowError(
            f"{run_id} 状态是 {row['status']}，只有 todo 能开工")
    record = _p(root, row["record"] or f"{m['runs_dir']}/{run_id}.md")
    if record.exists():
        raise WorkflowError(f"run 记录已存在：{record}（不许覆盖别人/自己的旧记录）")
    tmpl = _p(root, m["run_template"])
    content = tmpl.read_text(encoding="utf-8").replace("<实验ID>", run_id)
    content = re.sub(r"(\|\s*Status\s*\|\s*)`[^`]*`", r"\1`running`", content, count=1)
    now = _dt.datetime.now().strftime("%Y-%m-%d %H:%M")
    content = re.sub(r"(\|\s*Started\s*\|\s*)`[^`]*`", rf"\1`{now}`", content, count=1)
    record.parent.mkdir(parents=True, exist_ok=True)
    record.write_text(content, encoding="utf-8")
    _replace_row(queue, run_id, "running")
    return record


def finish_run(root, run_id: str, status: str):
    """收尾：QUEUE 状态 + run 记录 Status 一次改完。

    status 只接受 done/dead/blocked。四件套其余两件（STATE/LEDGER）按纪律
    由人/agent 同一次操作更新——explab check 会把关。
    """
    if status not in ("done", "dead", "blocked"):
        raise WorkflowError(f"收尾状态只能是 done/dead/blocked，得到 {status!r}")
    m = load_manifest(root)
    queue = _p(root, m["queue"])
    rows = {r["id"]: r for r in _parse_queue(queue)}
    if run_id not in rows:
        raise WorkflowError(f"QUEUE 里没有 {run_id}")
    if rows[run_id]["status"] != "running":
        raise WorkflowError(
            f"{run_id} 状态是 {rows[run_id]['status']}，只有 running 能收尾")
    record = _p(root, rows[run_id]["record"] or f"{m['runs_dir']}/{run_id}.md")
    _replace_row(queue, run_id, status)
    if record.exists():
        _set_record_status(record, status)
        if status in ("done", "dead"):
            _set_record_finished(record)


# ---------------------------------------------------------------- check
def check_state(root) -> list:
    """体检，返回问题清单（空 = 通过）。"""
    root = Path(root)
    m = load_manifest(root)
    errors = []

    for rel in ("AGENTS.md", m["state"], m["ledger"], m["queue"], m["run_template"]):
        if not _p(root, rel).exists():
            errors.append(f"流程文件缺失：{rel}（explab init 可补建）")
    if errors:
        return errors

    exts = set(m.get("clutter_exts", []))
    for f in root.iterdir():
        if f.is_file() and f.suffix in exts and f.name != MANIFEST_NAME:
            errors.append(f"根目录散落文件：{f.name}（应进 {m['configs_dir']}/）")

    rows = _parse_queue(_p(root, m["queue"]))
    seen = set()
    for r in rows:
        if r["id"] in seen:
            errors.append(f"QUEUE 重复 ID：{r['id']}")
        seen.add(r["id"])
        if r["status"] not in STATUSES:
            errors.append(f"QUEUE {r['id']}: 非法状态 {r['status']!r}"
                          f"（可选 {list(STATUSES)}）")
        if r["status"] == "blocked" and not r["notes"].strip():
            errors.append(f"QUEUE {r['id']}: 状态 blocked 但 notes 没写卡住原因"
                          f"（blocked 不许无声挂起）")
        rel = r["record"] or f"{m['runs_dir']}/{r['id']}.md"
        if not _p(root, rel).exists():
            if r["status"] in ("running", "done", "dead"):
                errors.append(f"QUEUE {r['id']}: 状态 {r['status']} 但 run 记录缺失 {rel}")
        else:
            _check_record_paths(root, _p(root, rel), errors)
            if r["status"] in ("done", "dead"):
                _check_decision_evidence(root, _p(root, rel), errors)
                _check_record_closure(root, _p(root, rel), r["id"],
                                      r["status"], errors)
            if r["status"] == "done":
                _check_done_metrics(_p(root, rel), r["id"], errors)
        if r["status"] in ("done", "dead"):
            _check_ledger_row(root, m["ledger"], r["id"], r["status"], errors)

    runs_dir = _p(root, m["runs_dir"])
    if runs_dir.exists():
        ids = {r["id"] for r in rows}
        for f in sorted(runs_dir.glob("*.md")):
            if f.stem not in ids:
                errors.append(f"孤儿 run 记录（QUEUE 里没有对应行）：{f.name}")

    tp = _p(root, m.get("third_party_dir", "third_party"))
    prov = _p(root, m["provenance"])
    if tp.exists():
        if not prov.exists():
            errors.append("存在 third_party/ 但没有 docs/PROVENANCE.md 登记")
        else:
            prov_text = prov.read_text(encoding="utf-8")
            for entry in sorted(p for p in tp.iterdir()
                                if p.is_dir() and not p.name.startswith(".")):
                if entry.name not in prov_text:
                    errors.append(
                        f"third_party/{entry.name} 未在 PROVENANCE.md 登记"
                        f"（来源/版本/用途必须可追溯）")
    return errors


_PATH_RE = re.compile(r"\b((?:configs|scripts|experiments)/[^\s`()（）]+)")
_BRACE_RE = re.compile(r"^(.*)\{([^{}]+)\}(.*)$")


def _expand_paths(root: Path, token: str) -> list:
    """展开 {a,b} 与 * 通配符，返回相对路径列表；无通配时原样返回。"""
    m = _BRACE_RE.match(token)
    if m:
        pre, items, post = m.groups()
        out = []
        for item in items.split(","):
            out.extend(_expand_paths(root, pre + item + post))
        return out
    if "*" in token:
        return sorted(str(p.relative_to(root)) for p in root.glob(token))
    return [token]


def _missing_expansions(root: Path, rel: str) -> list:
    return [x for x in _expand_paths(root, rel) if not (root / x).exists()]


def _check_record_paths(root: Path, record: Path, errors: list):
    """run 记录里引用的 configs/scripts/experiments 路径必须真实存在。

    跳过模板占位（含 <>）； Queue row 锚点写法 path::id 只取 path 部分；
    支持 {a,b} 与 * 简写（与托管检查脚本同一展开逻辑）。
    """
    text = record.read_text(encoding="utf-8")
    for raw in set(_PATH_RE.findall(text)):
        rel = raw.rstrip(".,;:")
        rel = rel.split("::")[0]
        if "<" in rel or ">" in rel:
            continue
        missing = _missing_expansions(root, rel)
        if missing:
            errors.append(
                f"{record.name} 引用了不存在的路径：{', '.join(missing)}")


_EVIDENCE_RE = re.compile(r"[-*]\s*Evidence[:：]\s*(.+)")
_EVIDENCE_CAND_RE = re.compile(r"[\w][\w.\-/]*\.[A-Za-z0-9]{1,8}")


def _check_decision_evidence(root: Path, record: Path, errors: list):
    """Decision 的证据指针：Evidence 行引用的路径必须真实存在。

    缺 Evidence 行只由托管检查脚本告警（不阻塞存量记录），这里只管
    「写了但指向不存在文件」的硬错误。
    """
    text = record.read_text(encoding="utf-8")
    idx = text.find("## Decision")
    if idx < 0:
        return
    m = _EVIDENCE_RE.search(text[idx:])
    if not m:
        return
    val = m.group(1).strip().strip("`").strip()
    if not val or "<" in val or ">" in val:
        return
    for cand in _EVIDENCE_CAND_RE.findall(val):
        cand = cand.strip("`\"'")
        if "/" not in cand or set("*{},") & set(cand):
            continue
        if not (root / cand).exists():
            errors.append(f"{record.name} Evidence 引用了不存在的路径：{cand}")


_META_FIELD_RE = r"^\|\s*{field}\s*\|\s*(.*?)\s*\|\s*$"
_FINISHED_DATE_RE = re.compile(r"\b(\d{4}-\d{2}-\d{2})\b")


def _check_record_closure(root: Path, record: Path, rid: str,
                          status: str, errors: list):
    """防伪闭环（与托管脚本 v4 对齐）：done/dead 记录的 Metadata Status
    必须与 QUEUE 状态一致（防绕过 CLI 只改一处），且必须带真实的
    Finished 时间戳——"完成"要留证明，不能只靠一句声明。"""
    text = record.read_text(encoding="utf-8")
    m = re.search(_META_FIELD_RE.format(field="Status"), text, re.MULTILINE)
    if m:
        rec_status = m.group(1).strip().strip("`").strip()
        if rec_status != status:
            errors.append(f"QUEUE {rid} 状态 {status} 但记录 {record.name} "
                          f"Metadata Status 是 {rec_status!r}——两处必须一致，"
                          f"用 explab finish 收尾，别只改一处")
    fm = re.search(_META_FIELD_RE.format(field="Finished"),
                   text, re.MULTILINE)
    val = fm.group(1).strip().strip("`").strip() if fm else ""
    if not val or "<" in val or ">" in val or not _FINISHED_DATE_RE.search(val):
        errors.append(f"{record.name} 判 {status} 但 Metadata 没有真实完成"
                      f"时间戳（Finished 行）——完成必须留时间证明")


def _check_done_metrics(record: Path, rid: str, errors: list):
    """done 证据闸门（与托管脚本 v5 对齐）：Result 表的 this run 列
    必须有至少一个数字——完成要留数，留不出数就是没跑完。"""
    text = record.read_text(encoding="utf-8")
    idx = text.find("## Result")
    if idx < 0:
        errors.append(f"{record.name} 判 done 但没有 ## Result 段")
        return
    section = text[idx:]
    nxt = section.find("\n## ", 3)
    if nxt > 0:
        section = section[:nxt]
    n_metrics = 0
    for line in section.splitlines():
        line = line.strip()
        if not line.startswith("|"):
            continue
        cells = [c.strip() for c in line.strip("|").split("|")]
        if len(cells) < 3:
            continue
        if set(cells[0]) <= set("-: ") or cells[0] == "指标":
            continue
        if re.search(r"\d", cells[2]):
            n_metrics += 1
    if n_metrics == 0:
        errors.append(f"{record.name} 判 done 但 Result 表里没有可解析的"
                      f"数字指标（this run 列）——完成必须留数")


def _check_ledger_row(root: Path, ledger_rel: str, rid: str,
                      status: str, errors: list):
    """done/dead 证据闸门（与托管脚本 v5 对齐）：LEDGER 必须有一行
    登记（第一列含 run id）——每个实验都要在账本留痕。"""
    ledger = _p(root, ledger_rel)
    if not ledger.exists():
        return
    for line in ledger.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line.startswith("|"):
            continue
        cells = [c.strip() for c in line.strip("|").split("|")]
        if cells and rid in cells[0]:
            return
    errors.append(f"QUEUE {rid} 判 {status} 但 LEDGER 没有对应行"
                  f"——每个实验都要在账本留痕")