PlantAI-rag-new/bioflow_service/app/services/bioflow/runner.py

"""app/services/bioflow/runner.py —— 任务状态机（逐阶段推进）+ 并发信号量 + 重启恢复。

状态机（job）：created → await_inputs → running → paused / done / failed / archiving / archived
状态机（stage）：waiting → await_files → waiting_confirm → queued → running
              → done / failed / skipped；failed →（rerun）waiting_confirm/running 或（skip）skipped

职责：
- run_job：逐阶段检查输入→等待确认（手动模式）→docker run→读 manifest 判成败→推进/暂停。
- 事件总线：每个状态变化 publish（12 类事件全清单）。
- 并发信号量（BIOFLOW_CONCURRENCY 默认 3，FR-BIO-031）：docker 执行前获取，
  无槽则排队（stage=queued）；任务失败/完成/暂停释放槽。
- 服务重启恢复：扫描 DB，running/running 阶段任务从断点续跑
  （已完成阶段不重跑；按 manifest 判成败——FR-BIO-031）。
- 执行器注入：默认 Kernel.run_stage（真 docker）；测试注入 fake（mock docker）。

服务层纪律：纯 dataclass/dict/stdlib，无 pydantic。
"""
from __future__ import annotations

import asyncio
import logging
import re
import time
import uuid
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Awaitable, Callable

from .catalog import StageCatalog
from .events import EventBus
from .manifests import ManifestError, locate_manifest, parse_manifest

logger = logging.getLogger(__name__)

# ---- 状态常量 ----

STAGE_WAITING = "waiting"
STAGE_AWAIT_FILES = "await_files"
STAGE_WAITING_CONFIRM = "waiting_confirm"
STAGE_QUEUED = "queued"
STAGE_RUNNING = "running"
STAGE_DONE = "done"
STAGE_FAILED = "failed"
STAGE_SKIPPED = "skipped"
STAGE_PAUSED = "paused"

STAGE_TERMINAL = {STAGE_DONE, STAGE_FAILED, STAGE_SKIPPED}
STAGE_RUNNABLE = {STAGE_WAITING, STAGE_AWAIT_FILES, STAGE_WAITING_CONFIRM, STAGE_QUEUED}

JOB_CREATED = "created"
JOB_AWAIT_INPUTS = "await_inputs"
JOB_RUNNING = "running"
JOB_PAUSED = "paused"
JOB_DONE = "done"
JOB_FAILED = "failed"
JOB_ARCHIVING = "archiving"
JOB_ARCHIVED = "archived"

JOB_TERMINAL = {JOB_DONE, JOB_FAILED, JOB_ARCHIVED}

# 首阶段默认输入文件名（FR-BIO-005 文件引导提示用）
DEFAULT_INPUT_HINTS = ("pep.fa", "pep.fasta", ".fasta", ".fa")

# manifest 判成败兜底：退出码 0 但 manifest status=failed → 失败（TC-BIO-RUN-016）
# manifest 缺必需字段但退出码 0 → 按退出码兜底（TC-BIO-KERN-017）


class RunnerError(Exception):
    """runner 操作错误（状态机非法迁移等；路由层转 409 信封）。"""

    def __init__(self, code: str, message: str) -> None:
        super().__init__(message)
        self.code = code
        self.message = message


# ---- 阶段输入检查 ----

def _stage_input_satisfied(stage: str, job_input_dir: Path) -> tuple[bool, list[str]]:
    """检查阶段输入文件是否就绪（目录内存在生信数据文件）。

    - 首阶段（identify/genome_prepare/seq 提取类）：要求 input/ 下有 fasta/gz 类文件；
      genome_prepare 放宽（允许 gff3/cds）。
    - 后续阶段：跨阶段输入（上游 final/ 产物），有上游 done 即满足（无显式缺名）。
    返回 (satisfied, missing_hints)。实现为保守启发：目录内任一数据文件即视为满足。
    """
    if not job_input_dir.is_dir():
        return False, list(DEFAULT_INPUT_HINTS)
    data_exts = (".fa", ".fasta", ".faa", ".fna", ".pep", ".cds", ".gz", ".gff", ".gff3", ".gtf", ".txt", ".tsv")
    files = [p for p in job_input_dir.rglob("*") if p.is_file()]
    data_files = [p for p in files if p.suffix.lower() in data_exts or ".fasta" in p.name.lower()]
    if data_files:
        return True, []
    return False, list(DEFAULT_INPUT_HINTS)


def ensure_samples_yaml(job_input_dir: Path, source_roots: list[Path] | None = None) -> list[str]:
    """为 rnaseq 系阶段生成/修正 samples.yaml,返回需挂载的宿主目录列表。

    逻辑(用户要求,勿做变体猜测):
    1. 描述文件发现:扫描登记目录树与任务 input(限深度 4)找 tsv/csv > xlsx >
       yaml/yml > md,解析样本清单(fastq 路径 + run 标识)。
    2. 文件实况分析:扫描源目录(登记目录本身)与任务 input,分类原始序列
       (.fq/.fastq 系)与压缩包(.tar.gz/.tar/.zip);压缩包解压到
       input/archives/(幂等);序列文件用于配对(不假设命名规律,只按 R1/R2
       语义:文件名含 f1/r1/1/forward 与 f2/r2/2/reverse)。
    3. 描述文件路径校正:逐条校验存在性;失效路径按 run 标识在实况序列文件中
       查找(文件名包含 run),唯一命中替换(按 R1/R2 语义分边);零/多命中保留
       原路径(CLI 报错)。
    4. 生成/重写 input/samples.yaml(引用真实存在路径),返回引用 fastq 的父目录
       (供执行容器挂载)。
    幂等:重跑时读入已是正确路径,校验通过后重写同内容。
    """
    import re as _re
    import yaml as _yaml

    # 描述文件解析函数(候选择优用;定义在循环前供嵌套闭包使用)
    def _parse_tsv_csv(desc: Path, seq_exts_local: tuple) -> list[dict]:
        import csv as _csv
        out: list[dict] = []
        with desc.open(encoding="utf-8", errors="replace", newline="") as fh:
            reader = _csv.DictReader(
                fh, delimiter="\t" if desc.name.lower().endswith(".tsv") else ","
            )
            for row in reader:
                r1 = str(row.get("fastq_1") or "").strip()
                r2 = str(row.get("fastq_2") or "").strip()
                run = str(row.get("run") or "").strip()
                sid = str(row.get("sample_id") or "").strip()
                if not r1 and not r2:
                    for v in row.values():
                        if str(v).strip().lower().endswith(seq_exts_local):
                            if not r1:
                                r1 = str(v).strip()
                            elif not r2:
                                r2 = str(v).strip()
                if r1 or r2 or run:
                    out.append({
                        "name": sid or run or (r1 and Path(r1).name.split(".")[0]) or "sample",
                        "run": run, "r1": r1, "r2": r2,
                        "meta": {k: v for k, v in row.items()
                                 if k not in ("fastq_1", "fastq_2", "sample_id", "run")},
                    })
        return out

    def _parse_xlsx(desc: Path) -> list[dict]:
        import openpyxl as _xl
        out: list[dict] = []
        wb = _xl.load_workbook(desc, read_only=True, data_only=True)
        try:
            ws = wb.active
            rows = list(ws.iter_rows(values_only=True))
            if rows:
                header = [str(h or "").strip() for h in rows[0]]
                for row in rows[1:]:
                    d = dict(zip(header, row))
                    run = str(d.get("run") or "").strip()
                    if not run:
                        continue
                    out.append({
                        "name": run, "run": run, "r1": "", "r2": "",
                        "meta": {k: v for k, v in d.items() if k != "run"},
                    })
        finally:
            wb.close()
        return out

    def _parse_yaml(desc: Path) -> list[dict]:
        raw = _yaml.safe_load(desc.read_text(encoding="utf-8", errors="replace"))
        out: list[dict] = []
        samples = raw.get("samples", {}) if isinstance(raw, dict) else {}
        for sname, spaths in samples.items():
            r1 = r2 = run = ""
            if isinstance(spaths, dict):
                r1 = str(spaths.get("r1") or spaths.get("read1") or "")
                r2 = str(spaths.get("r2") or spaths.get("read2") or "")
                run = str(spaths.get("run") or "")
            elif isinstance(spaths, (list, tuple)) and len(spaths) >= 2:
                r1, r2 = str(spaths[0]), str(spaths[1])
            if r1 or r2:
                out.append({
                    "name": str(sname), "run": run, "r1": r1, "r2": r2, "meta": {},
                })
        return out

    yaml_path = job_input_dir / "samples.yaml"
    seq_exts = (".fq.gz", ".fastq.gz", ".fq", ".fastq")
    roots: list[Path] = [job_input_dir] + [Path(r) for r in (source_roots or []) if Path(r).exists()]

    def _is_seq(p: Path) -> bool:
        return p.is_file() and p.name.lower().endswith(seq_exts)

    def _r1_sem(name: str) -> bool:
        n = name.lower()
        return bool(_re.search(r"[._-](f1|r1|1|forward)([._-]|$)", n))

    def _r2_sem(name: str) -> bool:
        n = name.lower()
        return bool(_re.search(r"[._-](f2|r2|2|reverse)([._-]|$)", n))

    def _walk(root: Path):
        yield from root.rglob("*")

    # ---- 2. 实况扫描:源目录(真实文件)+ 任务 input(跳过 symlink 重复) ----
    # 同文件去重:符号链接与其目标(samefile)只留一个 —— 优先保留真实文件(非链接),
    # 避免链接的别名文件名(如 CRR809753_2.fq.gz → _r2.fq.gz)被当成第二个样本。
    seq_files: list[Path] = []
    archives: list[Path] = []
    seen_inodes: set[tuple[int, int]] = set()

    def _seq_dedup(p: Path, pool: list[Path]) -> bool:
        try:
            key = (p.stat().st_dev, p.stat().st_ino)
        except OSError:
            key = None
        if key is None or key not in seen_inodes:
            if key is not None:
                seen_inodes.add(key)
            pool.append(p)
            return True
        # 已存在同 inode 文件:若新候选是真实文件、旧的是链接,用真实文件替换
        for i, q in enumerate(pool):
            if q.samefile(p):
                if p.is_symlink() and not q.is_symlink():
                    return False
                if not p.is_symlink() and q.is_symlink():
                    pool[i] = p
                    return True
                return False
        return False

    for root in roots:
        for p in sorted(_walk(root)):
            if not p.is_file():
                continue
            if p.is_symlink() and root == job_input_dir:
                continue  # input 里是登记 symlink,源目录已有真实文件
            if _is_seq(p):
                _seq_dedup(p, seq_files)
            elif p.name.lower().endswith((".tar.gz", ".tar", ".zip", ".tgz")):
                _seq_dedup(p, archives)
    seq_files = sorted({str(p): p for p in seq_files}.values(), key=lambda p: str(p))
    archives = sorted({str(p): p for p in archives}.values(), key=lambda p: str(p))

    # 解压压缩包(幂等:archives/ 已存在同名解出物则跳过)
    archive_dir = job_input_dir / "archives"
    for a in archives:
        target_dir = archive_dir / (a.name.rsplit(".", 1)[0])
        if target_dir.exists():
            continue
        try:
            import tarfile
            import zipfile
            target_dir.mkdir(parents=True, exist_ok=True)
            if a.name.endswith((".tar.gz", ".tar", ".tgz")):
                with tarfile.open(a, "r:*") as tf:
                    tf.extractall(target_dir)
            else:
                with zipfile.ZipFile(a) as zf:
                    zf.extractall(target_dir)
        except Exception:  # noqa: BLE001
            pass  # 解压失败不阻断;后续扫描自然不含其内容
    if archive_dir.exists():
        for p in sorted(archive_dir.rglob("*")):
            if _is_seq(p):
                seq_files.append(p)
        seq_files = sorted({str(p): p for p in seq_files}.values(), key=lambda p: str(p))

    # ---- 1. 描述文件发现:候选全收 → 逐份解析 → 按解析出的样本条目数择优 ----
    def _depth_ok(p: Path, root: Path) -> bool:
        try:
            return len(p.relative_to(root).parts) <= 4
        except ValueError:
            return False

    candidates: list[Path] = []
    for root in roots:
        for p in sorted(root.rglob("*")):
            if not p.is_file() or p.name == "samples.yaml":
                continue
            if not _depth_ok(p, root):
                continue
            if p.name.lower().endswith((".tsv", ".csv", ".xlsx", ".xls", ".yaml", ".yml")):
                candidates.append(p)

    described: list[dict] = []
    best: tuple[int, list[dict]] | None = None
    for desc in candidates:
        try:
            if desc.name.lower().endswith((".tsv", ".csv")):
                parsed = _parse_tsv_csv(desc, seq_exts)
            elif desc.name.lower().endswith((".xlsx", ".xls")):
                parsed = _parse_xlsx(desc)
            else:
                parsed = _parse_yaml(desc)
        except Exception:  # noqa: BLE001
            parsed = []
        # 样本条目数最多者胜;平手时 tsv/csv 优先(语义最全),再按路径排序稳定
        is_table = desc.name.lower().endswith((".tsv", ".csv"))
        if best is None or len(parsed) > best[0] or (len(parsed) == best[0] and is_table and not best_is_table):
            best = (len(parsed), parsed)
            best_is_table = is_table
    if best and best[0] > 0:
        described = best[1]

    # ---- 3. 描述文件路径校正(实况为准,按 run 标识,不做变体猜测) ----
    seq_by_run: dict[str, list[Path]] = {}
    for p in seq_files:
        m = _re.search(r"([A-Za-z0-9]{6,})", p.stem)
        if m:
            seq_by_run.setdefault(m.group(1), []).append(p)
    if described:
        for d in described:
            for key in ("r1", "r2"):
                p = Path(d[key]) if d[key] else None
                if p and p.exists():
                    continue
                run = d.get("run") or ""
                cands = [q for q in seq_by_run.get(run, [])] if run else []
                if not cands:
                    cands = [q for q in seq_files if d["name"] and d["name"] in q.name]
                if len(cands) == 1:
                    d[key] = str(cands[0])
                elif len(cands) >= 2:
                    mate = [q for q in cands if (_r2_sem(q.name) if key == "r2" else _r1_sem(q.name))]
                    if len(mate) == 1:
                        d[key] = str(mate[0])
                # 零命中:保留原路径,CLI 报错
        # 实况中还有描述未覆盖的序列 → 追加
        known = {Path(d["r1"]).name for d in described if d["r1"]}
        known |= {Path(d["r2"]).name for d in described if d["r2"]}
        for p in seq_files:
            if p.name not in known:
                described.append({"name": p.stem, "run": "", "r1": str(p), "r2": "", "meta": {}})

    def _pair_key(name: str) -> str:
        """归一配对比较键(小写):_f1/_r2/_1/_2 别名归一到同一样本。"""
        return _re.sub(
            r"[._-](f1|r1|1|forward|f2|r2|2|reverse)([._-]|$).*", "", name.lower()
        )

    def _pair_base(name: str) -> str:
        """保留原大小写的样本名:剥离 R1/R2 语义后缀。"""
        return _re.sub(
            r"[._-](f1|r1|1|forward|f2|r2|2|reverse)([._-]|$).*", "", name,
            flags=_re.IGNORECASE,
        )

    # ---- 无描述文件:实况配对(不假设命名规律,按 R1/R2 语义) ----
    if not described and seq_files:
        r1s = [p for p in seq_files if _r1_sem(p.name)]
        r2s = [p for p in seq_files if _r2_sem(p.name)]
        used_r2: set[str] = set()  # 一对一配对:已被认领的 r2 不再参与
        for p in r1s:
            base = _pair_key(p.name)
            mate_cands = [
                q for q in r2s if _pair_key(q.name) == base and str(q) not in used_r2
            ]
            mate: Path | None = None
            if len(mate_cands) == 1:
                mate = mate_cands[0]
            elif len(mate_cands) > 1:
                # 多个同名 stem 候选:优先真实 f2/r2/reverse 命名,纯数字 _2 别名靠后
                pref = [
                    q for q in mate_cands
                    if _re.search(r"[._-](f2|r2|reverse)([._-]|$)", q.name.lower())
                ]
                mate = pref[0] if pref else mate_cands[0]
            described.append({
                "name": _pair_base(p.name) or p.stem, "run": "",
                "r1": str(p), "r2": str(mate) if mate else "", "meta": {},
            })
            if mate is not None:
                used_r2.add(str(mate))
        for p in r2s:
            if str(p) not in used_r2:
                described.append({"name": p.stem, "run": "", "r1": "", "r2": str(p), "meta": {}})
        for p in seq_files:
            if not _r1_sem(p.name) and not _r2_sem(p.name):
                described.append({"name": p.stem, "run": "", "r1": str(p), "r2": "", "meta": {}})

    # ---- 4. 生成/重写 samples.yaml + 返回挂载目录 ----
    mounts: set[str] = set()
    samples_out: dict[str, dict] = {}
    for d in described:
        if not (d.get("r1") or d.get("r2")):
            continue
        entry: dict = {}
        for key in ("r1", "r2"):
            p = Path(d[key]) if d[key] else None
            if p and p.exists():
                entry[key] = str(p)
                mounts.add(str(p.parent))
        if not entry:
            continue
        for k, v in d.get("meta", {}).items():
            if v not in (None, ""):
                entry[str(k)] = v
        if d.get("run"):
            entry.setdefault("run", d["run"])
        samples_out[d["name"]] = entry
    if samples_out:
        yaml_path.write_text(
            _yaml.safe_dump({"samples": samples_out}, allow_unicode=True), encoding="utf-8"
        )
    return sorted(mounts)


def inspect_unknown_files(paths: list[Path], max_files: int = 200) -> list[dict]:
    """可观测特征函数:对无法按规则归类的文件,收集结构化特征供 LLM 兜底。

    每项:{path, size, ext, magic(首 64 字节 hex + file 类型), parent,
          archive_kind(tar|zip|gz|None), archive_members(≤20 内嵌名)}。
    纯 stdlib + 子进程 file/tar/zcat/unzip,失败静默跳过该文件。
    """
    import subprocess
    features: list[dict] = []
    for p in sorted(paths, key=lambda q: (q.stat().st_size if q.is_file() else 0), reverse=True)[:max_files]:
        if not p.is_file():
            continue
        try:
            size = p.stat().st_size
            head = p.read_bytes()[:64]
            magic = head.hex()
            ext = p.suffix.lower() or ""
            parent = str(p.parent)
            archive_kind = None
            members: list[str] = []
            file_type = ""
            try:
                file_type = subprocess.run(
                    ["file", "-b", str(p)], capture_output=True, text=True, timeout=10
                ).stdout.strip()
            except Exception:  # noqa: BLE001
                pass
            if file_type.startswith("gzip") or magic.startswith("1f8b"):
                archive_kind = "gz"
                try:
                    out = subprocess.run(
                        ["gzip", "-l", str(p)], capture_output=True, text=True, timeout=10
                    ).stdout
                    members = out.strip().splitlines()[1:]
                except Exception:  # noqa: BLE001
                    pass
            elif file_type.startswith("POSIX tar") or magic.startswith(("7573746172",)):
                archive_kind = "tar"
                try:
                    out = subprocess.run(
                        ["tar", "-tzf", str(p)], capture_output=True, text=True, timeout=15
                    ).stdout
                    members = [m for m in out.splitlines() if m.strip()][:20]
                except Exception:  # noqa: BLE001
                    pass
            elif file_type.startswith("Zip archive") or magic.startswith("504b"):
                archive_kind = "zip"
                try:
                    out = subprocess.run(
                        ["unzip", "-l", str(p)], capture_output=True, text=True, timeout=15
                    ).stdout
                    members = [m.strip().split()[-1] for m in out.splitlines()[3:-2] if m.strip()][:20]
                except Exception:  # noqa: BLE001
                    pass
            features.append({
                "path": str(p), "size": size, "ext": ext, "magic": magic[:64],
                "file_type": file_type, "parent": parent,
                "archive_kind": archive_kind, "archive_members": members,
            })
        except Exception:  # noqa: BLE001
            continue
    return features


async def llm_sample_fallback(job_input_dir: Path, source_roots: list[Path]) -> bool:
    """LLM 兜底:规则无法归纳时,把文件特征清单交给 DeepSeek 生成样本描述。

    触发条件:描述文件解析 0 条且实况配对 0 样本(调用方判断)。
    返回是否成功生成 samples.yaml。LLM 失败/未配置/输出无法回验 → False,
    不影响登记主链。r1/r2 必须回验存在(清单路径或解压后可解析),否则丢弃。
    """
    import yaml as _yaml
    from .llm import LLMError, complete_json

    # 收集未知文件(排除已识别的序列/描述文件)
    known_exts = (".tsv", ".csv", ".xlsx", ".xls", ".yaml", ".yml", ".md",
                  ".fq.gz", ".fastq.gz", ".fq", ".fastq")
    unknown: list[Path] = []
    for root in [job_input_dir, *[Path(r) for r in source_roots if Path(r).exists()]]:
        for p in root.rglob("*"):
            if not p.is_file() or p.name == "samples.yaml":
                continue
            if p.name.lower().endswith(known_exts):
                continue
            unknown.append(p)
    if not unknown:
        return False
    features = inspect_unknown_files(unknown)
    if not features:
        return False

    # system prompt(独立文件,可迭代)
    prompt_path = Path(__file__).resolve().parent.parent / "prompts" / "llm_sample_analysis.txt"
    try:
        system = prompt_path.read_text(encoding="utf-8")
    except OSError:
        system = "你是生信数据文件分析助手,只输出 JSON。"
    import json as _json
    user = (
        "以下是一个数据目录的文件特征清单,请推断其中的 RNA-seq FASTQ 样本并输出 JSON。\n"
        f"{_json.dumps(features, ensure_ascii=False, default=str)[:24000]}"
    )
    try:
        result = await complete_json(system, user, temperature=0.2, max_tokens=4000, timeout=30.0)
    except (LLMError, Exception):  # noqa: BLE001
        return False
    samples = result.get("samples") if isinstance(result, dict) else None
    if not isinstance(samples, list) or not samples:
        return False

    # 回验 r1/r2:必须是清单路径(存在)或压缩包内嵌名(解压后可解析)
    valid_paths = {f["path"] for f in features if Path(f["path"]).exists()}
    archive_members: dict[str, str] = {}
    for f in features:
        for m in (f.get("archive_members") or []):
            archive_members.setdefault(m, f["path"])
    yaml_samples: dict[str, dict] = {}
    mounts: set[str] = set()
    for s in samples:
        if not isinstance(s, dict):
            continue
        r1 = str(s.get("r1") or "").strip()
        r2 = str(s.get("r2") or "").strip()
        entry: dict = {}
        for key, val in (("r1", r1), ("r2", r2)):
            if not val:
                continue
            if val in valid_paths:
                entry[key] = val
                mounts.add(str(Path(val).parent))
            elif val in archive_members:
                # 解压该压缩包(幂等),用内嵌名构造解出路径
                arch = Path(archive_members[val])
                target_dir = job_input_dir / "archives" / (arch.name.rsplit(".", 1)[0])
                try:
                    import tarfile
                    import zipfile
                    if not target_dir.exists():
                        target_dir.mkdir(parents=True, exist_ok=True)
                        if arch.name.endswith((".tar.gz", ".tar", ".tgz")):
                            with tarfile.open(arch, "r:*") as tf:
                                tf.extractall(target_dir)
                        else:
                            with zipfile.ZipFile(arch) as zf:
                                zf.extractall(target_dir)
                    resolved = target_dir / val
                    if resolved.exists():
                        entry[key] = str(resolved)
                        mounts.add(str(resolved.parent))
                except Exception:  # noqa: BLE001
                    pass
        if not entry:
            continue
        for k in ("run", "group_role", "condition", "genotype", "replicate"):
            v = str(s.get(k) or "").strip()
            if v:
                entry[k] = v
        yaml_samples[str(s.get("name") or Path(entry.get("r1") or "sample").stem)] = entry
    if not yaml_samples:
        return False
    (job_input_dir / "samples.yaml").write_text(
        _yaml.safe_dump({"samples": yaml_samples}, allow_unicode=True), encoding="utf-8"
    )
    return True


# ---- runner ----

ExecutorFn = Callable[..., Awaitable[dict[str, Any]]]

# 增量日志事件节流：每 N 行或每 S 秒发布一次 stage_log(增量)
LOG_EMIT_LINES = 20
LOG_EMIT_SECS = 0.5
# 进度事件节流(秒)：stage_progress 最低发布间隔
PROGRESS_THROTTLE_SECS = 2.0
# 进度解析白名单：只有这些阶段的 CLI 有 [PROGRESS] 标记(其余阶段不编造数字)
PROGRESS_STAGES = ("rnaseq_qc", "rnaseq_sample_qc")


# 阶段进度解析：CLI 日志中的 [PROGRESS] 标记 → 进度增量信息。
# 保守策略：只识别 rnaseq 系 CLI 的既有标记，非白名单阶段不编造数字。
_PROGRESS_SAMPLE_RE = re.compile(
    r"\[PROGRESS\]\s*Processing:\s*(.+?)\s*$"
)
_PROGRESS_COPY_RE = re.compile(
    r"\[PROGRESS\]\s*Copying\s+results"
)


def parse_progress_marker(line: str) -> dict[str, Any] | None:
    """解析单行日志中的 [PROGRESS] 标记（纯函数，可单测）。

    - `[PROGRESS] Processing: X` → {"kind": "sample", "message": "X"}
    - `[PROGRESS] Copying results ...` → {"kind": "copy", "percent": 95}
    - 其余行 → None
    """
    if "[PROGRESS]" not in line:
        return None
    m = _PROGRESS_SAMPLE_RE.search(line)
    if m:
        return {"kind": "sample", "message": m.group(1).strip()}
    if _PROGRESS_COPY_RE.search(line):
        return {"kind": "copy", "percent": 95}
    return None


@dataclass
class JobRecord:
    """任务内存记录（DB 行 + 派生路径）。"""

    id: str
    session_id: str = ""
    title: str = ""
    status: str = JOB_CREATED
    stage_chain: list[str] = field(default_factory=list)
    params: dict[str, dict[str, Any]] = field(default_factory=dict)
    auto_continue: bool = False
    archive_id: str | None = None
    pause_requested: bool = False
    registered_server_paths: list[str] = field(default_factory=list)  # 登记路径（自动挂载）
    _running_container: str | None = None  # 当前执行容器名(暂停时按名 kill)
    _stage_states: dict[str, dict[str, Any]] = field(default_factory=dict)

    @property
    def job_dir(self) -> Path:
        from ...config import get_settings
        return Path(get_settings().jobs_dir_abs) / self.id

    @property
    def input_dir(self) -> Path:
        return self.job_dir / "input"

    @property
    def workspace_dir(self) -> Path:
        return self.job_dir / "workspace"

    @property
    def logs_dir(self) -> Path:
        return self.job_dir / "logs"

    def progress(self) -> int:
        total = len(self.stage_chain)
        if not total:
            return 100
        done = sum(
            1 for s in self.stage_chain
            if self._stage_states.get(s, {}).get("status") in (STAGE_DONE, STAGE_SKIPPED)
        )
        return int(done * 100 / total)


class Runner:
    """任务状态机执行器（每任务一个实例；共享 db/events/kernel/catalog/semaphore）。"""

    def __init__(
        self,
        db: Any,
        events: EventBus,
        executor: ExecutorFn,
        catalog: StageCatalog,
        semaphore: asyncio.Semaphore,
        *,
        force_input_check: bool = True,
        store: Any = None,          # 存储层（测试可注入；生产用 app/services/bioflow/store）
        task_factory: Any = None,   # asyncio.create_task 替身（测试注入）
        auto_mode_default: bool = False,
    ) -> None:
        self.db = db
        self.events = events
        self.executor = executor
        self.catalog = catalog
        self.semaphore = semaphore
        self.force_input_check = force_input_check
        self.store = store
        self._task_factory = task_factory or asyncio.create_task
        self.auto_mode_default = auto_mode_default
        self._tasks: set[asyncio.Task] = set()

    # ---- 任务生命周期 ----

    async def create_job(
        self,
        session_id: str,
        title: str,
        stages: list[str],
        params: dict[str, dict[str, Any]] | None = None,
        *,
        auto_continue: bool = False,
        archive_id: str | None = None,
        auto_start: bool = True,
    ) -> JobRecord:
        """创建任务（job_created 事件 + 三表落行 + 可选自动启动）。"""
        unknown = self.catalog.validate_chain(stages)
        if unknown:
            raise RunnerError("UNKNOWN_STAGE", f"未知阶段: {', '.join(unknown)}")
        ordered, _added = self.catalog.auto_fill_prerequisites(stages)
        job_id = uuid.uuid4().hex[:16]
        record = JobRecord(
            id=job_id, session_id=session_id, title=title or "生信任务",
            status=JOB_CREATED, stage_chain=ordered,
            params=params or {}, auto_continue=auto_continue or self.auto_mode_default,
            archive_id=archive_id,
        )
        record.job_dir.mkdir(parents=True, exist_ok=True)
        record.input_dir.mkdir(parents=True, exist_ok=True)
        record.workspace_dir.mkdir(parents=True, exist_ok=True)
        record.logs_dir.mkdir(parents=True, exist_ok=True)
        for stage in ordered:
            record._stage_states[stage] = {"status": STAGE_WAITING, "params": params.get(stage) or {} if params else {}}
        await self._persist_job(record)
        for stage in ordered:
            await self._persist_stage(record, stage)
        self.events.publish("job_created", {
            "job_id": record.id,
            "session_id": record.session_id,
            "title": record.title,
            "stages": [{"stage": s, "status": STAGE_WAITING} for s in ordered],
            "auto_continue": record.auto_continue,
        })
        if auto_start:
            self.spawn(self.run_job(record))
        return record

    def spawn(self, coro: Awaitable[Any]) -> asyncio.Task:
        task = self._task_factory(coro)
        self._tasks.add(task)
        task.add_done_callback(self._tasks.discard)
        return task

    # ---- DB 持久化 ----

    async def _persist_job(self, record: JobRecord) -> None:
        if self.store is not None:
            await self.store.upsert_job(record)
            return
        from .store import upsert_job
        await upsert_job(self.db, record)

    async def _persist_stage(self, record: JobRecord, stage: str) -> None:
        if self.store is not None:
            await self.store.upsert_stage(record, stage)
            return
        from .store import upsert_stage
        await upsert_stage(self.db, record, stage)

    # ---- 阶段状态读写 ----

    def stage_status(self, record: JobRecord, stage: str) -> str:
        return str(record._stage_states.get(stage, {}).get("status") or STAGE_WAITING)

    def _set_stage(self, record: JobRecord, stage: str, **fields: Any) -> None:
        st = record._stage_states.setdefault(stage, {"status": STAGE_WAITING})
        st.update(fields)

    # ---- 主循环 ----

    async def run_job(self, record: JobRecord) -> None:
        """逐阶段推进（成功/失败/暂停/等待确认）。"""
        try:
            record.status = JOB_RUNNING
            await self._persist_job(record)
            for idx, stage in enumerate(record.stage_chain):
                if record.pause_requested:
                    await self._pause_job(record, stage)
                    return
                if record.status == JOB_PAUSED:
                    return
                outcome = await self._run_stage(record, stage, idx)
                if outcome == "paused":
                    return
                if outcome == "failed":
                    record.status = JOB_FAILED
                    await self._persist_job(record)
                    return
                if outcome == "await":
                    # 等待确认/文件：任务进入 await_inputs（可删除；确认/补齐后续跑）
                    record.status = JOB_AWAIT_INPUTS
                    await self._persist_job(record)
                    return
            # 全链完成
            if record.pause_requested:
                await self._pause_job(record, None)
                return
            record.status = JOB_DONE
            await self._persist_job(record)
            self.events.publish("job_done", {
                "job_id": record.id,
                "title": record.title,
                "progress": 100,
                "elapsed_sec": None,
            })
        except asyncio.CancelledError:
            # 任务级取消：不落脏状态（阶段取消由 _run_stage 单独处理）
            raise
        except Exception:  # noqa: BLE001 - 防御：任何异常不挂服务
            logger.exception("runner: 任务执行异常 %s", record.id)
            record.status = JOB_FAILED
            await self._persist_job(record)

    async def _run_stage(self, record: JobRecord, stage: str, idx: int) -> str:
        """执行单个阶段（状态机核心）。返回 "done"|"await"|"failed"|"paused"|"skipped"。"""
        status = self.stage_status(record, stage)
        if status == STAGE_DONE:
            return "done"
        if status == STAGE_SKIPPED:
            return "skipped"

        # 0. confirm 已确认旁路：状态停留 waiting_confirm 但参数已被确认 → 直接执行
        st = record._stage_states.setdefault(stage, {})
        if status == STAGE_WAITING_CONFIRM and st.get("_confirmed"):
            pass  # 跳过下方确认门
        else:
            # 1. 输入检查（FR-BIO-005：缺失→await_files；force_input_check=False 时跳过）
            if self.force_input_check:
                satisfied, missing = _stage_input_satisfied(stage, record.input_dir)
                if not satisfied:
                    self._set_stage(record, stage, status=STAGE_AWAIT_FILES, error="")
                    await self._persist_stage(record, stage)
                    self.events.publish("stage_await_files", {
                        "job_id": record.id, "stage": stage,
                        "required_files": missing,
                        "message": "请将数据文件拖入窗口上传至服务器，或在聊天窗口中指定数据文件的具体路径",
                    })
                    return "await"

            # 2. 手动模式等待确认（FR-BIO-008）
            if not record.auto_continue and status != STAGE_QUEUED:
                self._set_stage(record, stage, status=STAGE_WAITING_CONFIRM)
                await self._persist_stage(record, stage)
                defaults = self._form_defaults(stage, record)
                self.events.publish("stage_await_confirm", {
                    "job_id": record.id, "stage": stage,
                    "form_defaults": defaults,
                })
                return "await"

        # 3. 并发信号量（FR-BIO-031；槽满时内部置 queued 并阻塞排队）
        await self._acquire(record, stage)

        # 排队期间收到暂停：释放槽，停在阶段边界（TC-BIO-API-122）
        if record.pause_requested:
            self.semaphore.release()
            self._set_stage(record, stage, status=STAGE_WAITING_CONFIRM if not record.auto_continue else STAGE_WAITING)
            await self._persist_stage(record, stage)
            await self._pause_job(record, stage)
            return "paused"

        # 4. docker 执行
        try:
            params = record._stage_states.get(stage, {}).get("params") or {}
            attempt = int(record._stage_states.get(stage, {}).get("attempt") or 0) + 1
            self._set_stage(record, stage, status=STAGE_RUNNING, attempt=attempt,
                            started_at=_now_iso(), error="")
            await self._persist_stage(record, stage)
            self.events.publish("stage_started", {
                "job_id": record.id, "stage": stage, "attempt": attempt,
            })
            attempt_no = int(record._stage_states.get(stage, {}).get("attempt") or 0) + 1
            record._running_container = f"plantai-bio-{record.id}-{stage}-{attempt_no}"

            # 进度/增量日志状态(每次 attempt 重置)
            emit_buf: list[str] = []
            emit_last = time.monotonic()
            prog_state: dict[str, Any] = {"seen": 0, "total": 0, "last": 0.0}
            track_progress = stage in PROGRESS_STAGES
            total_re = re.compile(r"Processing\s+(\d+)\s+samples")
            valid_re = re.compile(r"Valid\s+samples:\s*(\d+)")

            async def _on_log(chunk: str) -> None:
                nonlocal emit_last
                emit_buf.append(chunk)
                now = time.monotonic()
                if len(emit_buf) >= LOG_EMIT_LINES or now - emit_last >= LOG_EMIT_SECS:
                    self.events.publish("stage_log", {
                        "job_id": record.id, "stage": stage, "log": "".join(emit_buf),
                    })
                    emit_buf.clear()
                    emit_last = now
                # [PROGRESS] 解析(白名单内才跟踪)
                if track_progress:
                    for line in chunk.splitlines():
                        marker = parse_progress_marker(line)
                        if not marker:
                            continue
                        if marker["kind"] == "copy":
                            prog_state["copying"] = True
                        elif marker["kind"] == "sample":
                            prog_state["seen"] += 1
                            prog_state["sample"] = marker["message"]
                    tm = total_re.search(chunk)
                    if tm:
                        prog_state["total"] = int(tm.group(1))
                    vm = valid_re.search(chunk)
                    if vm:
                        prog_state["total"] = int(vm.group(1))
                    if (prog_state["seen"] and prog_state["total"]
                            and now - prog_state["last"] >= PROGRESS_THROTTLE_SECS):
                        prog_state["last"] = now
                        percent = min(int(prog_state["seen"] * 100 / prog_state["total"]), 99)
                        self.events.publish("stage_progress", {
                            "job_id": record.id, "stage": stage,
                            "percent": percent,
                            "message": str(prog_state.get("sample") or ""),
                            "sample": str(prog_state.get("sample") or ""),
                        })

            result = await self.executor(stage, record.input_dir, record.workspace_dir,
                                         params, log_path=record.logs_dir / f"{stage}.log",
                                         extra_mounts=self._registered_mounts(record),
                                         container_name=record._running_container,
                                         on_log=_on_log)
            record._running_container = None
            # 尾块兜底(增量 stage_log 与日志文件对齐)
            if emit_buf:
                self.events.publish("stage_log", {
                    "job_id": record.id, "stage": stage, "log": "".join(emit_buf),
                })
            log_tail = str(result.get("log_tail") or "")
            self._set_stage(record, stage, log_tail=log_tail)
            await self._persist_stage(record, stage)
            self.events.publish("stage_log", {
                "job_id": record.id, "stage": stage, "log": log_tail,
            })
            if record.pause_requested:
                # 暂停请求在阶段执行期间到达(容器已被 pause_job kill)：
                # 不按 exit 137 判 failed,走暂停分支
                self._set_stage(record, stage, status=STAGE_PAUSED, error="")
                await self._persist_stage(record, stage)
                await self._pause_job(record, stage)
                return "paused"
            outcome = await self._judge_stage(record, stage, result)
            return outcome
        finally:
            record._running_container = None
            self.semaphore.release()

    async def _acquire(self, record: JobRecord, stage: str) -> bool:
        """获取并发槽（FR-BIO-031）。

        - 槽空闲：立即获取，返回 True。
        - 槽满：阶段置 queued（事件 stage_log 说明排队），随后阻塞等待
          asyncio.Semaphore 内建 FIFO 等待队列（先入队者先执行）。
        竞态：locked() 检查与 acquire() 之间可能被其他任务抢占，最坏情况是
        排队阶段未标记 queued 直接执行——状态正确性不受影响。
        """
        if self.semaphore.locked():
            self._set_stage(record, stage, status=STAGE_QUEUED)
            await self._persist_stage(record, stage)
            self.events.publish("stage_log", {
                "job_id": record.id, "stage": stage,
                "log": "并发槽已满，阶段排队等待中…（queue）",
            })
        await self.semaphore.acquire()
        return True

    async def _judge_stage(self, record: JobRecord, stage: str, result: dict[str, Any]) -> str:
        """执行结果 + manifest 判成败（TC-BIO-RUN-012/014/015/016）。"""
        exit_code = result.get("exit_code")
        timed_out = bool(result.get("timed_out"))
        if timed_out:
            await self._fail_stage(record, stage, exit_code=exit_code, reason="timeout")
            return "failed"
        if exit_code not in (0, None):
            await self._fail_stage(record, stage, exit_code=exit_code, reason="exit_code")
            return "failed"

        # 退出码 0：读 manifest 判成败
        prefix = self._prefix_for(record, stage)
        manifest_path = locate_manifest(record.workspace_dir, stage, prefix)
        if manifest_path is None:
            await self._fail_stage(record, stage, exit_code=exit_code, reason="no_manifest")
            return "failed"
        try:
            manifest = parse_manifest(record.workspace_dir, stage, prefix)
        except ManifestError as exc:
            # manifest 解析失败：退出码 0 且 JSON 损坏 → 失败（KERN-016）
            await self._fail_stage(record, stage, exit_code=exit_code, reason=f"manifest_error: {exc}")
            return "failed"
        if manifest.status == "failed":
            await self._fail_stage(record, stage, exit_code=exit_code, reason="manifest_failed")
            return "failed"

        # 成功
        self._set_stage(record, stage, status=STAGE_DONE, manifest_path=manifest.path,
                        finished_at=_now_iso())
        await self._persist_stage(record, stage)
        # 终态强制 100%(白名单阶段;非白名单阶段前端会因 done 状态显示完成)
        if stage in PROGRESS_STAGES:
            self.events.publish("stage_progress", {
                "job_id": record.id, "stage": stage, "percent": 100, "message": "",
            })
        self.events.publish("stage_done", {
            "job_id": record.id, "stage": stage, "exit_code": exit_code,
            "manifest_ref": manifest.path,
            "metrics": manifest.metrics,
        })
        return "done"

    async def _fail_stage(self, record: JobRecord, stage: str, *,
                          exit_code: int | None, reason: str) -> None:
        self._set_stage(record, stage, status=STAGE_FAILED, finished_at=_now_iso(),
                        error=reason)
        await self._persist_stage(record, stage)
        self.events.publish("stage_failed", {
            "job_id": record.id, "stage": stage,
            "exit_code": exit_code if exit_code is not None else -1,
            "reason": reason,
        })

    def _prefix_for(self, record: JobRecord, stage: str) -> str | None:
        params = record._stage_states.get(stage, {}).get("params") or {}
        return str(params.get("prefix") or "") or None

    def _form_defaults(self, stage: str, record: JobRecord) -> list[dict[str, Any]]:
        """参数确认表单默认值（目录 schema + 任务参数/存档参数预填 + 来源标注）。"""
        entry = self.catalog.get(stage)
        schema = entry.params if entry else []
        prefill = record.params.get(stage) or {}
        st = record._stage_states.get(stage, {})
        existing = st.get("params") or {}
        merged = {p.get("name"): p for p in schema}
        defaults: list[dict[str, Any]] = []
        for p in schema:
            name = p.get("name")
            value = existing.get(name, prefill.get(name, p.get("default")))
            defaults.append({
                "name": name,
                "type": p.get("type"),
                "default": value,
                "source": p.get("source", "contract"),
                "help": p.get("help", ""),
                "choices": p.get("choices"),
            })
        return defaults

    # ---- 控制操作（路由层调用）----

    async def confirm_stage(self, record: JobRecord, stage: str, params: dict[str, Any]) -> None:
        """阶段确认（FR-BIO-009）：校验参数→存表→执行（并发排队）。"""
        status = self.stage_status(record, stage)
        # 幂等：已确认且正在执行/排队（如前端重复点击「继续」），直接放行不报 409
        st_existing = record._stage_states.get(stage, {})
        if status in (STAGE_RUNNING, STAGE_QUEUED) and st_existing.get("_confirmed"):
            return
        if status not in (STAGE_WAITING_CONFIRM, STAGE_QUEUED, STAGE_WAITING):
            raise RunnerError("STAGE_NOT_CONFIRMABLE", f"阶段 {stage} 当前状态 {status} 不可确认")
        allowed = self._allowed_param_names(stage)
        unknown = [k for k in params if k not in allowed]
        if unknown:
            raise RunnerError("UNKNOWN_PARAMS", f"未知参数: {', '.join(unknown)}")
        st = record._stage_states.setdefault(stage, {"status": STAGE_WAITING})
        st["params"] = {**st.get("params", {}), **params}
        st["_confirmed"] = True  # 确认旁路标记（_run_stage 跳过确认门直接执行）
        await self._persist_stage(record, stage)
        record.pause_requested = False
        if record.status in (JOB_PAUSED, JOB_CREATED, JOB_FAILED, JOB_AWAIT_INPUTS):
            record.status = JOB_RUNNING
            await self._persist_job(record)
        self.spawn(self._run_confirmed_stage(record, stage))

    async def _run_confirmed_stage(self, record: JobRecord, stage: str) -> None:
        """确认后执行单个阶段并推进后续链（done→继续；failed→停）。"""
        try:
            outcome = await self._run_stage(record, stage, 0)
            if outcome == "failed":
                record.status = JOB_FAILED
                await self._persist_job(record)
                return
            if outcome == "await":
                return
            # 后续阶段继续推进
            idx = record.stage_chain.index(stage) if stage in record.stage_chain else 0
            for nxt in record.stage_chain[idx + 1:]:
                if record.pause_requested or record.status == JOB_PAUSED:
                    return
                res = await self._run_stage(record, nxt, idx)
                if res in ("failed", "await", "paused"):
                    if res == "failed":
                        record.status = JOB_FAILED
                        await self._persist_job(record)
                    return
            if not record.pause_requested:
                record.status = JOB_DONE
                await self._persist_job(record)
                self.events.publish("job_done", {"job_id": record.id, "progress": 100})
        except asyncio.CancelledError:
            raise
        except Exception:  # noqa: BLE001
            logger.exception("runner: confirm 推进异常 %s", record.id)

    def _allowed_param_names(self, stage: str) -> set[str]:
        entry = self.catalog.get(stage)
        if entry is None:
            return set()
        return {p.get("name") for p in entry.params if p.get("name")} | {"cpu", "memory", "env"}

    async def rerun_stage(self, record: JobRecord, stage: str, params: dict[str, Any] | None) -> None:
        """失败阶段重跑（FR-BIO-012）。带参数 → waiting_confirm 再确认；否则直接跑。"""
        status = self.stage_status(record, stage)
        if status != STAGE_FAILED:
            raise RunnerError("STAGE_NOT_FAILED", f"阶段 {stage} 当前状态 {status} 不可重跑")
        if params is None:
            params = {}
        st = record._stage_states.setdefault(stage, {})
        st["params"] = {**st.get("params", {}), **params}
        if params:
            st["status"] = STAGE_WAITING_CONFIRM
            await self._persist_stage(record, stage)
            defaults = self._form_defaults(stage, record)
            self.events.publish("stage_await_confirm", {
                "job_id": record.id, "stage": stage, "form_defaults": defaults,
                "rerun_with_params": True,
            })
            record.status = JOB_RUNNING
            await self._persist_job(record)
            return
        st["status"] = STAGE_WAITING
        await self._persist_stage(record, stage)
        record.status = JOB_RUNNING
        record.pause_requested = False
        await self._persist_job(record)
        self.spawn(self._run_confirmed_stage(record, stage))

    async def skip_stage(self, record: JobRecord, stage: str) -> None:
        """失败阶段跳过（FR-BIO-012）：置 skipped + 事件 + 后续阶段推进。"""
        status = self.stage_status(record, stage)
        if status != STAGE_FAILED:
            raise RunnerError("STAGE_NOT_FAILED", f"阶段 {stage} 当前状态 {status} 不可跳过")
        self._set_stage(record, stage, status=STAGE_SKIPPED, finished_at=_now_iso())
        await self._persist_stage(record, stage)
        self.events.publish("stage_skipped", {"job_id": record.id, "stage": stage})
        record.status = JOB_RUNNING
        record.pause_requested = False
        await self._persist_job(record)
        self.spawn(self._run_after_stage(record, stage))

    async def _run_after_stage(self, record: JobRecord, stage: str) -> None:
        try:
            idx = record.stage_chain.index(stage) if stage in record.stage_chain else 0
            for nxt in record.stage_chain[idx + 1:]:
                if record.pause_requested or record.status == JOB_PAUSED:
                    return
                res = await self._run_stage(record, nxt, idx)
                if res in ("failed", "await", "paused"):
                    if res == "failed":
                        record.status = JOB_FAILED
                        await self._persist_job(record)
                    return
            if not record.pause_requested and record.status != JOB_PAUSED:
                record.status = JOB_DONE
                await self._persist_job(record)
                self.events.publish("job_done", {"job_id": record.id, "progress": 100})
        except asyncio.CancelledError:
            raise
        except Exception:  # noqa: BLE001
            logger.exception("runner: skip 后推进异常 %s", record.id)

    async def pause_job(self, record: JobRecord) -> None:
        """暂停（FR-BIO-010）：running 阶段有执行容器时立即 kill,任务即刻暂停。"""
        if record.status not in (JOB_RUNNING, JOB_AWAIT_INPUTS):
            raise RunnerError("JOB_NOT_RUNNING", f"任务 {record.id} 当前状态 {record.status} 不可暂停")
        record.pause_requested = True
        await self._persist_job(record)
        # 有 running 阶段 → kill 执行容器(立即生效;失败仅日志,不阻断暂停语义)
        running = [s for s in record.stage_chain
                   if self.stage_status(record, s) == STAGE_RUNNING]
        if running and getattr(record, "_running_container", None):
            name = record._running_container
            try:
                proc = await asyncio.create_subprocess_shell(
                    f"docker kill {name}",
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.STDOUT,
                )
                await asyncio.wait_for(proc.communicate(), timeout=30.0)
            except Exception as exc:  # noqa: BLE001 - kill 失败不阻断(仍走阶段边界暂停)
                logger.warning("pause_job: docker kill %s 失败: %s", name, exc)
        if not running:
            await self._pause_job(record, None)

    async def _pause_job(self, record: JobRecord, current_stage: str | None) -> None:
        record.status = JOB_PAUSED
        record.pause_requested = False
        await self._persist_job(record)
        if current_stage:
            st = record._stage_states.get(current_stage)
            if st and st.get("status") not in (STAGE_DONE, STAGE_FAILED, STAGE_SKIPPED, STAGE_WAITING_CONFIRM):
                st["status"] = STAGE_PAUSED
                await self._persist_stage(record, current_stage)
        self.events.publish("job_paused", {
            "job_id": record.id,
            "current_stage": current_stage,
        })

    async def resume_job(self, record: JobRecord) -> None:
        """恢复（FR-BIO-010）：从下一阶段继续，不重跑已完成；确认点不跳过。"""
        if record.status != JOB_PAUSED:
            raise RunnerError("JOB_NOT_PAUSED", f"任务 {record.id} 当前状态 {record.status} 不可恢复")
        # paused 阶段复位为等待确认（不跳过确认，TC-BIO-API-069）
        for s, st in record._stage_states.items():
            if st.get("status") == STAGE_PAUSED:
                st["status"] = STAGE_WAITING_CONFIRM
                await self._persist_stage(record, s)
        record.status = JOB_RUNNING
        record.pause_requested = False
        await self._persist_job(record)
        self.spawn(self._resume_loop(record))

    def _registered_mounts(self, record: JobRecord) -> list[str]:
        """登记路径 → 挂载清单：取每个登记文件的父目录（去重）。

        另并入 samples.yaml 引用的 fastq 父目录(描述文件路径校正/实况分析产物)。
        """
        parents: list[str] = []
        seen: set[str] = set()
        for p in getattr(record, "registered_server_paths", None) or []:
            parent = str(Path(p).parent)
            if parent not in seen:
                seen.add(parent)
                parents.append(parent)
        try:
            for m in ensure_samples_yaml(record.input_dir, source_roots=[
                Path(p).parent for p in getattr(record, "registered_server_paths", None) or []
            ]):
                if m not in seen:
                    seen.add(m)
                    parents.append(m)
        except Exception:  # noqa: BLE001 - 挂载补充失败不阻断
            pass
        return parents

    async def retry_after_files(self, record: JobRecord) -> None:
        """文件上传/路径登记后的自动续跑（BUG-P7-01：文件引导闭环 FR-BIO-005/006）。

        此前任务首阶段缺输入停在 await_files 时 run_job 已 "await" 返回，
        上传成功后无任何重推机制（confirm 对 await_files 抛 409）——任务卡死。
        本方法：
        - 仅当 record.status == JOB_AWAIT_INPUTS 时生效（其余状态 no-op，
          防重复 spawn）；
        - 置 JOB_RUNNING、pause_requested=False、persist_job；
        - spawn(_resume_loop)：跳过 done 阶段，await_files 阶段经 _run_stage
          重新输入检查——文件齐备则 auto_continue 直跑或发 stage_await_confirm。
        """
        if record.status != JOB_AWAIT_INPUTS:
            return
        record.status = JOB_RUNNING
        record.pause_requested = False
        await self._persist_job(record)
        self.spawn(self._resume_loop(record))

    async def _resume_loop(self, record: JobRecord) -> None:
        try:
            for stage in record.stage_chain:
                if record.pause_requested or record.status == JOB_PAUSED:
                    return
                status = self.stage_status(record, stage)
                if status in (STAGE_DONE, STAGE_SKIPPED):
                    continue
                if status == STAGE_FAILED:
                    record.status = JOB_FAILED
                    await self._persist_job(record)
                    return
                res = await self._run_stage(record, stage, 0)
                if res in ("failed", "await", "paused"):
                    if res == "failed":
                        record.status = JOB_FAILED
                        await self._persist_job(record)
                    return
            if not record.pause_requested and record.status != JOB_PAUSED:
                record.status = JOB_DONE
                await self._persist_job(record)
                self.events.publish("job_done", {"job_id": record.id, "progress": 100})
        except asyncio.CancelledError:
            raise
        except Exception:  # noqa: BLE001
            logger.exception("runner: resume 循环异常 %s", record.id)

    # ---- 重启恢复（FR-BIO-031）----

    async def recover(self) -> list[JobRecord]:
        """服务重启恢复：扫 DB，running/created 任务从断点续跑。

        - running 阶段：按 manifest 判成败（有 success manifest→done，否则重跑）。
        - done 阶段不重跑（attempt 不变，TC-BIO-API-111）。
        - 已归档任务跳过（TC-BIO-RUN-030）。
        """
        from .store import load_running_jobs, load_job_stages
        rows = await load_running_jobs(self.db)
        recovered: list[JobRecord] = []
        for row in rows:
            stages = await load_job_stages(self.db, row["id"])
            record = JobRecord(
                id=row["id"], session_id=row["session_id"] or "",
                title=row["title"] or "生信任务",
                status=row["status"],
                stage_chain=stages_list(row["stage_chain"]),
                params=params_dict(row["params"]),
                auto_continue=bool(row["auto_continue"]),
                archive_id=row["archive_id"],
            )
            # 登记路径恢复（server_registered.json）
            try:
                import json as _json
                reg_file = record.job_dir / "server_registered.json"
                if reg_file.is_file():
                    record.registered_server_paths = (
                        _json.loads(reg_file.read_text(encoding="utf-8"))
                        .get("registered_server_paths") or []
                    )
            except Exception:  # noqa: BLE001
                pass
            for s in stages:
                record._stage_states[s["stage"]] = {
                    "status": s["status"], "params": params_dict(s["params"]),
                    "attempt": int(s["attempt"] or 0),
                    "manifest_path": s["manifest_path"], "log_tail": s["log_tail"] or "",
                    "error": s["error"],
                }
            # running 阶段按 manifest 判成败（断点续跑不重跑已成功部分）
            for stage, st in record._stage_states.items():
                if st.get("status") == STAGE_RUNNING:
                    prefix = self._prefix_for(record, stage)
                    if locate_manifest(record.workspace_dir, stage, prefix):
                        try:
                            manifest = parse_manifest(record.workspace_dir, stage, prefix)
                            if manifest.status == "success":
                                st["status"] = STAGE_DONE
                                st["manifest_path"] = manifest.path
                            else:
                                st["status"] = STAGE_FAILED
                                st["error"] = "manifest_failed"
                        except ManifestError:
                            st["status"] = STAGE_FAILED
                            st["error"] = "manifest_error"
                    else:
                        st["status"] = STAGE_WAITING
                    await self._persist_stage(record, stage)
            self.events.publish("job_recovered", {
                "job_id": record.id, "status": record.status,
            })
            self.spawn(self._resume_loop(record))
            recovered.append(record)
        return recovered


def stages_list(raw: Any) -> list[str]:
    if isinstance(raw, list):
        return [str(s) for s in raw]
    import json
    try:
        value = json.loads(raw or "[]")
        return [str(s) for s in value] if isinstance(value, list) else []
    except (ValueError, TypeError):
        return []


def params_dict(raw: Any) -> dict[str, dict[str, Any]]:
    import json
    if isinstance(raw, dict):
        return raw
    try:
        value = json.loads(raw or "{}")
        return value if isinstance(value, dict) else {}
    except (ValueError, TypeError):
        return {}


def _now_iso() -> str:
    from datetime import datetime, timezone
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")