PlantAI-rag-new/bioflow_service/tests/test_runner.py

"""tests/test_runner.py —— runner 状态机全路径（mock docker 执行器）。

覆盖（TC-BIO-RUN 系列）：成功推进/失败/重跑/跳过/暂停/恢复/重启续跑/超时/
日志截断/并发上限/输入缺失等待/确认等待/自动连续/信号量释放/事件成对。

注意：runner 内部用 asyncio.create_task 派生任务，测试必须「单场景单 loop」——
每个用例在一个新建 event loop 内完成 create→run→断言前的所有异步步骤。
"""
from __future__ import annotations

import asyncio

import pytest

from conftest import FakeExecutor, FakeTimedOutExecutor, make_runner
from app.services.bioflow.runner import (
    JOB_AWAIT_INPUTS,
    JOB_DONE,
    JOB_FAILED,
    JOB_PAUSED,
    STAGE_DONE,
    STAGE_PAUSED,
    STAGE_QUEUED,
    STAGE_RUNNING,
    STAGE_FAILED,
    STAGE_SKIPPED,
    STAGE_WAITING_CONFIRM,
    RunnerError,
)

pytestmark = pytest.mark.bioflow_unit


def _scenario(coro_factory):
    """单 loop 跑完整场景。"""
    loop = asyncio.new_event_loop()
    try:
        return loop.run_until_complete(coro_factory())
    finally:
        loop.close()


async def _drain(runner, seconds: float = 5.0) -> None:
    """等待 runner 派生的后台任务全部结束。"""
    while runner._tasks:
        await asyncio.wait_for(asyncio.gather(*list(runner._tasks)), timeout=seconds)
    await asyncio.sleep(0.01)


def _collected(events, job_id):
    return [e for e in events.replay(job_id) if e["type"] != "job_recovered"]


def test_full_chain_success(db, events, catalog, fake_executor):
    """TC-BIO-RUN-001：全链成功推进（自动连续）。"""

    async def scenario():
        runner = make_runner(db, events, fake_executor, catalog)
        record = await runner.create_job(
            "s1", "SPL 鉴定", ["identify", "phylogeny", "motif"],
            auto_continue=True, auto_start=False)
        await runner.run_job(record)
        return runner, record

    runner, record = _scenario(scenario)
    assert record.status == JOB_DONE
    ev_types = [e["type"] for e in _collected(events, record.id)]
    assert ev_types.count("stage_started") == 3
    assert ev_types.count("stage_done") == 3
    assert ev_types[-1] == "job_done"
    assert ev_types[0] == "job_created"
    assert all(runner.stage_status(record, s) == STAGE_DONE for s in record.stage_chain)


def test_stage_failure_stops_chain(db, events, catalog):
    """TC-BIO-RUN-002：阶段失败转 failed，后续阶段不执行。"""
    executor = FakeExecutor(exit_code=1, log="boom\n")

    async def scenario():
        runner = make_runner(db, events, executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify", "phylogeny", "motif"],
            auto_continue=True, auto_start=False)
        await runner.run_job(record)
        return runner, record

    runner, record = _scenario(scenario)
    assert record.status == JOB_FAILED
    assert runner.stage_status(record, "identify") == STAGE_FAILED
    assert runner.stage_status(record, "phylogeny") == "waiting"
    ev_types = [e["type"] for e in _collected(events, record.id)]
    assert "stage_failed" in ev_types
    failed_ev = [e for e in _collected(events, record.id) if e["type"] == "stage_failed"][0]
    assert failed_ev["data"]["stage"] == "identify"
    assert failed_ev["data"]["reason"] == "exit_code"
    assert [c["stage"] for c in executor.calls] == ["identify"]  # 后续阶段未执行


def test_rerun_after_failure(db, events, catalog):
    """TC-BIO-RUN-003：失败后 rerun 成功，attempt=2，继续后续阶段。"""
    executor = FakeExecutor(exit_code=1)

    async def scenario():
        runner = make_runner(db, events, executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify", "phylogeny"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        assert record.status == JOB_FAILED
        executor.exit_code = 0
        await runner.rerun_stage(record, "identify", None)
        await _drain(runner)
        return runner, record

    runner, record = _scenario(scenario)
    assert runner.stage_status(record, "identify") == STAGE_DONE
    assert record._stage_states["identify"]["attempt"] == 2
    assert record.status == JOB_DONE  # 后续阶段自动推进


def test_skip_after_failure(db, events, catalog):
    """TC-BIO-RUN-004：失败后 skip → skipped，后续阶段可执行。"""
    executor = FakeExecutor(exit_code=1)

    async def scenario():
        runner = make_runner(db, events, executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify", "phylogeny"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        assert runner.stage_status(record, "identify") == STAGE_FAILED
        executor.exit_code = 0
        await runner.skip_stage(record, "identify")
        await _drain(runner)
        return runner, record

    runner, record = _scenario(scenario)
    assert runner.stage_status(record, "identify") == STAGE_SKIPPED
    ev_types = [e["type"] for e in _collected(events, record.id)]
    assert "stage_skipped" in ev_types
    assert record.status == JOB_DONE
    assert runner.stage_status(record, "phylogeny") == STAGE_DONE


def test_rerun_with_params_goes_confirm(db, events, catalog):
    """TC-BIO-RUN-005：带参数重跑 → waiting_confirm 再确认；新参数进入执行。"""
    executor = FakeExecutor(exit_code=1)

    async def scenario():
        runner = make_runner(db, events, executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        await runner.rerun_stage(record, "identify", {"cpu": "8"})
        assert runner.stage_status(record, "identify") == STAGE_WAITING_CONFIRM
        executor.exit_code = 0
        await runner.confirm_stage(record, "identify", {"cpu": "8"})
        await _drain(runner)
        return runner, record

    runner, record = _scenario(scenario)
    assert record._stage_states["identify"]["params"]["cpu"] == "8"
    assert runner.stage_status(record, "identify") == STAGE_DONE
    assert executor.calls[-1]["cpu"] == "8"


def test_rerun_non_failed_rejected(db, events, catalog, fake_executor):
    """TC-BIO-RUN-024：done 阶段 rerun/skip 拒绝。"""

    async def scenario():
        runner = make_runner(db, events, fake_executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        return runner, record

    runner, record = _scenario(scenario)
    with pytest.raises(RunnerError, match="不可重跑"):
        _scenario(lambda: runner.rerun_stage(record, "identify", None))
    with pytest.raises(RunnerError, match="不可跳过"):
        _scenario(lambda: runner.skip_stage(record, "identify"))


def test_pause_at_stage_boundary(db, events, catalog):
    """TC-BIO-RUN-006：运行中暂停立即生效（当前阶段置 paused,不判失败）。"""
    executor = FakeExecutor(delay=0.05)

    async def scenario():
        runner = make_runner(db, events, executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify", "phylogeny"], auto_continue=True, auto_start=False)
        task = asyncio.create_task(runner.run_job(record))
        await asyncio.sleep(0.02)  # identify 已进入执行
        await runner.pause_job(record)
        await asyncio.wait_for(task, timeout=5)
        return runner, record

    runner, record = _scenario(scenario)
    assert record.status == JOB_PAUSED
    assert runner.stage_status(record, "identify") == STAGE_PAUSED  # 立即暂停语义
    assert all(c["stage"] != "phylogeny" for c in executor.calls)


def test_confirm_idempotent_while_running(db, events, catalog):
    """确认幂等：已确认且 running/queued 的重复 confirm 不报 409。"""
    executor = FakeExecutor(delay=0.1)

    async def scenario():
        runner = make_runner(db, events, executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=False, auto_start=False)
        await runner.run_job(record)  # 手动模式:停 waiting_confirm
        await runner.confirm_stage(record, "identify", {})
        # 立即二次 confirm:阶段已 running 且 _confirmed → 幂等放行,不抛异常
        await runner.confirm_stage(record, "identify", {})
        await _drain(runner)
        return runner, record

    runner, record = _scenario(scenario)
    assert runner.stage_status(record, "identify") in (STAGE_RUNNING, STAGE_DONE, STAGE_QUEUED)


def test_resume_continues_without_rerun(db, events, catalog):
    """TC-BIO-RUN-007：暂停后恢复续跑，不重跑已完成。"""
    executor = FakeExecutor(delay=0.02)

    async def scenario():
        runner = make_runner(db, events, executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify", "phylogeny"], auto_continue=True, auto_start=False)
        task = asyncio.create_task(runner.run_job(record))
        await asyncio.sleep(0.03)
        await runner.pause_job(record)
        await asyncio.wait_for(task, timeout=5)
        assert record.status == JOB_PAUSED
        await runner.resume_job(record)
        await _drain(runner)
        return runner, record

    runner, record = _scenario(scenario)
    assert record.status == JOB_DONE
    identify_calls = [c for c in executor.calls if c["stage"] == "identify"]
    assert len(identify_calls) == 1  # 不重跑


def test_timeout_fails_stage(db, events, catalog):
    """TC-BIO-RUN-012：超时 → failed（reason=timeout）。"""
    executor = FakeTimedOutExecutor()

    async def scenario():
        runner = make_runner(db, events, executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        return runner, record

    runner, record = _scenario(scenario)
    assert runner.stage_status(record, "identify") == STAGE_FAILED
    failed = [e for e in _collected(events, record.id) if e["type"] == "stage_failed"][0]
    assert failed["data"]["reason"] == "timeout"


def test_log_tail_truncated(db, events, catalog):
    """TC-BIO-RUN-013：日志截断只留尾 80 行。"""
    executor = FakeExecutor(log="\n".join(f"L{i}" for i in range(500)) + "\n")

    async def scenario():
        runner = make_runner(db, events, executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        return runner, record

    runner, record = _scenario(scenario)
    tail = record._stage_states["identify"]["log_tail"]
    assert tail.count("\n") + 1 == 80
    assert tail.startswith("L420")
    assert tail.rstrip().endswith("L499")


def test_exit_nonzero_judged_failed(db, events, catalog):
    """TC-BIO-RUN-014：退出码非 0 判失败，log_tail 含 stderr。"""
    executor = FakeExecutor(exit_code=3, log="error output\n")

    async def scenario():
        runner = make_runner(db, events, executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        return runner, record

    runner, record = _scenario(scenario)
    assert runner.stage_status(record, "identify") == STAGE_FAILED
    assert "error output" in record._stage_states["identify"]["log_tail"]


def test_manifest_missing_judged_failed(db, events, catalog):
    """TC-BIO-RUN-015：退出码 0 但 manifest 缺 → failed（reason=no_manifest）。"""
    executor = FakeExecutor(exit_code=0, manifest_status="missing")

    async def scenario():
        runner = make_runner(db, events, executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        return runner, record

    runner, record = _scenario(scenario)
    assert runner.stage_status(record, "identify") == STAGE_FAILED
    failed = [e for e in _collected(events, record.id) if e["type"] == "stage_failed"][0]
    assert failed["data"]["reason"] == "no_manifest"


def test_manifest_failed_status(db, events, catalog):
    """TC-BIO-RUN-016：manifest status=failed 判败（不因退出码 0 误判成功）。"""
    executor = FakeExecutor(exit_code=0, manifest_status="failed")

    async def scenario():
        runner = make_runner(db, events, executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        return runner, record

    runner, record = _scenario(scenario)
    assert runner.stage_status(record, "identify") == STAGE_FAILED
    failed = [e for e in _collected(events, record.id) if e["type"] == "stage_failed"][0]
    assert failed["data"]["reason"] == "manifest_failed"


def test_await_files_when_missing(db, events, catalog):
    """TC-BIO-RUN-017：输入缺失等待文件（force_input_check=True）。"""
    executor = FakeExecutor()

    async def scenario():
        runner = make_runner(db, events, executor, catalog, check_inputs=True)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        return runner, record

    runner, record = _scenario(scenario)
    assert runner.stage_status(record, "identify") == "await_files"
    ev = [e for e in _collected(events, record.id) if e["type"] == "stage_await_files"][0]
    assert ev["data"]["stage"] == "identify"
    assert ev["data"]["required_files"]
    assert executor.calls == []  # 不启动 docker


def test_await_files_then_resume(db, events, catalog):
    """TC-BIO-RUN-018：文件补齐自动推进。"""
    executor = FakeExecutor()

    async def scenario():
        runner = make_runner(db, events, executor, catalog, check_inputs=True)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        assert runner.stage_status(record, "identify") == "await_files"
        (record.input_dir / "genome.pep.fa").write_text(">x\nACGT\n", encoding="utf-8")
        record._stage_states["identify"]["status"] = "waiting"
        await runner.run_job(record)
        return runner, record

    runner, record = _scenario(scenario)
    assert record.status == JOB_DONE
    assert runner.stage_status(record, "identify") == STAGE_DONE


def test_retry_after_files_auto_continue_runs_to_done(db, events, catalog):
    """BUG-P7-01①：auto_continue 任务上传后 retry_after_files 自动续跑到 done。"""
    executor = FakeExecutor()

    async def scenario():
        runner = make_runner(db, events, executor, catalog, check_inputs=True)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        assert runner.stage_status(record, "identify") == "await_files"
        assert record.status == JOB_AWAIT_INPUTS
        assert executor.calls == []
        # 模拟上传落盘
        (record.input_dir / "genome.pep.fa").write_text(">x\nACGT\n", encoding="utf-8")
        # 上传端点调用点（BUG-P7-01 修法核心）
        await runner.retry_after_files(record)
        await _drain(runner)
        return runner, record

    runner, record = _scenario(scenario)
    assert record.status == JOB_DONE
    assert runner.stage_status(record, "identify") == STAGE_DONE
    assert executor.calls  # fake executor 记录执行调用


def test_retry_after_files_manual_mode_goes_confirm(db, events, catalog):
    """BUG-P7-01②：手动模式上传后进入 waiting_confirm 并收到 stage_await_confirm。"""
    executor = FakeExecutor()

    async def scenario():
        runner = make_runner(db, events, executor, catalog, check_inputs=True)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=False, auto_start=False)
        await runner.run_job(record)
        assert runner.stage_status(record, "identify") == "await_files"
        (record.input_dir / "genome.pep.fa").write_text(">x\nACGT\n", encoding="utf-8")
        await runner.retry_after_files(record)
        await _drain(runner)
        return runner, record

    runner, record = _scenario(scenario)
    assert runner.stage_status(record, "identify") == STAGE_WAITING_CONFIRM
    confirm_events = [e for e in _collected(events, record.id)
                      if e["type"] == "stage_await_confirm"]
    assert confirm_events
    assert confirm_events[-1]["data"]["stage"] == "identify"
    assert executor.calls == []  # 未确认不执行


def test_retry_after_files_non_await_state_is_noop(db, events, catalog):
    """BUG-P7-01③：非 await_inputs 状态调用为 no-op（不重复 spawn）。"""
    executor = FakeExecutor()

    async def scenario():
        runner = make_runner(db, events, executor, catalog, check_inputs=True)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        assert record.status == JOB_AWAIT_INPUTS
        (record.input_dir / "genome.pep.fa").write_text(">x\nACGT\n", encoding="utf-8")
        await runner.retry_after_files(record)
        await _drain(runner)
        assert record.status == JOB_DONE
        done_calls = len(executor.calls)
        tasks_before = len(runner._tasks)
        # 已完成状态再调用 → no-op：状态不变、不新增任务、无新执行
        await runner.retry_after_files(record)
        await _drain(runner)
        assert record.status == JOB_DONE
        assert len(executor.calls) == done_calls  # 不重复执行
        assert len(runner._tasks) <= tasks_before
        return runner, record

    _scenario(scenario)


def test_await_confirm_manual_mode(db, events, catalog):
    """TC-BIO-RUN-019/020：手动模式等待确认；确认后执行。"""
    executor = FakeExecutor()

    async def scenario():
        runner = make_runner(db, events, executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify", "phylogeny"], auto_continue=False, auto_start=False)
        await runner.run_job(record)
        assert runner.stage_status(record, "identify") == STAGE_WAITING_CONFIRM
        assert executor.calls == []  # 未确认不执行
        await runner.confirm_stage(record, "identify", {})
        await _drain(runner)
        return runner, record

    runner, record = _scenario(scenario)
    ev = [e for e in _collected(events, record.id) if e["type"] == "stage_await_confirm"][0]
    assert ev["data"]["stage"] == "identify"
    assert ev["data"]["form_defaults"]  # 表单默认值
    assert runner.stage_status(record, "identify") == STAGE_DONE
    assert runner.stage_status(record, "phylogeny") == STAGE_WAITING_CONFIRM


def test_confirm_unknown_param_rejected(db, events, catalog, fake_executor):
    """TC-BIO-API-058 对应服务层：confirm 拒绝非 schema 参数。"""

    async def scenario():
        runner = make_runner(db, events, fake_executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_start=False)
        await runner.run_job(record)
        with pytest.raises(RunnerError, match="未知参数"):
            await runner.confirm_stage(record, "identify", {"magic": "x"})
        return runner, record

    _scenario(scenario)


def test_confirm_non_confirmable(db, events, catalog, fake_executor):
    """TC-BIO-API-059：重复 confirm 幂等拒绝。"""

    async def scenario():
        runner = make_runner(db, events, fake_executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        assert runner.stage_status(record, "identify") == STAGE_DONE
        with pytest.raises(RunnerError, match="不可确认"):
            await runner.confirm_stage(record, "identify", {})
        return runner, record

    _scenario(scenario)


def test_auto_continue_skips_confirm(db, events, catalog, fake_executor):
    """TC-BIO-RUN-021：自动连续跳过确认全程推进。"""

    async def scenario():
        runner = make_runner(db, events, fake_executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify", "phylogeny"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        return runner, record

    runner, record = _scenario(scenario)
    assert record.status == JOB_DONE
    assert "stage_await_confirm" not in [e["type"] for e in _collected(events, record.id)]


def test_concurrency_limit(db, events, catalog):
    """TC-BIO-RUN-010：并发上限排队（4 任务并发，同时 running ≤3）。"""
    executor = FakeExecutor(delay=0.1)

    async def scenario():
        runner = make_runner(db, events, executor, catalog)
        records = []
        for i in range(4):
            r = await runner.create_job(
                f"s{i}", f"t{i}", ["identify"], auto_continue=True, auto_start=False)
            records.append(r)
        await asyncio.gather(*[runner.run_job(r) for r in records])
        return runner, records

    runner, records = _scenario(scenario)
    assert executor.max_running <= 3
    assert all(r.status == JOB_DONE for r in records)


def test_semaphore_released_on_failure(db, events, catalog):
    """TC-BIO-RUN-025：执行失败后信号量释放。"""
    executor = FakeExecutor(exit_code=1)

    async def scenario():
        runner = make_runner(db, events, executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        return runner, record

    runner, record = _scenario(scenario)
    assert record.status == JOB_FAILED
    assert not runner.semaphore.locked()


def test_restart_recovery_no_rerun(db, events, catalog, fake_executor):
    """TC-BIO-RUN-008：重启续跑不重跑已完成阶段（attempt 不变）。"""

    async def scenario():
        runner = make_runner(db, events, fake_executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify", "phylogeny"], auto_continue=True, auto_start=False)
        # 只完成第 1 阶段（模拟中断前状态）
        await runner._run_stage(record, "identify", 0)
        assert runner.stage_status(record, "identify") == STAGE_DONE
        # 模拟重启：新 runner 实例（进程重启后内存清空）
        runner2 = make_runner(db, events, fake_executor, catalog)
        recovered = await runner2.recover()
        await _drain(runner2)
        return runner2, record, recovered

    runner2, record, recovered = _scenario(scenario)
    assert any(r.id == record.id for r in recovered)
    from app.services.bioflow import store
    st = _scenario(lambda: store.get_stage(db, record.id, "identify"))
    assert st["status"] == STAGE_DONE
    assert st["attempt"] == 1  # 不重跑


def test_restart_recovers_inflight_by_manifest(db, events, catalog, fake_executor):
    """TC-BIO-RUN-009：重启恢复 in-flight running 阶段按 manifest 判成败。"""

    async def scenario():
        runner = make_runner(db, events, fake_executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify", "phylogeny"], auto_continue=True, auto_start=False)
        await runner._run_stage(record, "identify", 0)  # 第 1 阶段完成
        # phylogeny 标记 running 且无 manifest（模拟崩溃瞬间）
        record._stage_states["phylogeny"]["status"] = "running"
        await runner._persist_stage(record, "phylogeny")
        # 模拟重启
        runner2 = make_runner(db, events, fake_executor, catalog)
        await runner2.recover()
        await _drain(runner2)
        # 同一 loop 内校验（aiosqlite 连接绑定本 loop）
        from app.services.bioflow import store
        st = await store.get_stage(db, record.id, "phylogeny")
        assert st["status"] == STAGE_DONE
        assert st["attempt"] >= 1  # 无 manifest → 重跑；attempt 至少 1
        return runner2, record

    _scenario(scenario)


def test_restart_skips_archived(db, events, catalog, fake_executor):
    """TC-BIO-RUN-030：恢复时跳过已归档任务。"""

    async def scenario():
        runner = make_runner(db, events, fake_executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        # 归档（状态改为 archived）
        await runner._persist_job(record)
        from app.services.bioflow.store import update_job_status
        await update_job_status(db, record.id, "archived")
        runner2 = make_runner(db, events, fake_executor, catalog)
        recovered = await runner2.recover()
        return recovered

    recovered = _scenario(scenario)
    assert recovered == []


def test_empty_stages_done_immediately(db, events, catalog, fake_executor):
    """TC-BIO-RUN-039：空阶段任务直接 done，无 docker 调用。"""
    # 注意：创建任务要求 stages 非空（pydantic min_length=1）由路由层保证；
    # 服务层直接构造空链验证防御
    async def scenario():
        runner = make_runner(db, events, fake_executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=True, auto_start=False)
        record.stage_chain = []
        record._stage_states = {}
        await runner.run_job(record)
        return runner, record

    runner, record = _scenario(scenario)
    assert record.status == JOB_DONE
    assert fake_executor.calls == []
    assert record.progress() == 100


def test_progress_calculation(db, events, catalog, fake_executor):
    """TC-BIO-RUN-029：任务进度数值。"""

    async def scenario():
        runner = make_runner(db, events, fake_executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify", "phylogeny", "motif"],
            auto_continue=True, auto_start=False)
        assert record.progress() == 0
        # 全链跑完 → 100（进度公式按完成阶段数/总数）
        await runner.run_job(record)
        return runner, record

    runner, record = _scenario(scenario)
    assert record.progress() == 100


def test_unknown_stage_create_rejected(db, events, catalog, fake_executor):
    async def scenario():
        runner = make_runner(db, events, fake_executor, catalog)
        with pytest.raises(RunnerError, match="未知阶段"):
            await runner.create_job("s1", "t", ["ghost_stage"], auto_start=False)
        return None

    _scenario(scenario)


def test_prerequisite_autofill(db, events, catalog, fake_executor):
    """TC-BIO-RUN-034：前置补全后链完整（phylogeny 缺 identify → 自动补）。"""

    async def scenario():
        runner = make_runner(db, events, fake_executor, catalog)
        record = await runner.create_job("s1", "t", ["phylogeny", "motif"], auto_start=False)
        return record

    record = _scenario(scenario)
    assert record.stage_chain[0] == "identify"
    assert set(record.stage_chain) == {"identify", "phylogeny", "motif"}


def test_multi_job_isolation(db, events, catalog):
    """TC-BIO-RUN-028：多任务互不串扰。"""
    executor = FakeExecutor(log="job1 log\n")

    async def scenario():
        runner = make_runner(db, events, executor, catalog)
        r1 = await runner.create_job("s1", "t1", ["identify"], auto_continue=True, auto_start=False)
        r2 = await runner.create_job("s2", "t2", ["identify"], auto_continue=True, auto_start=False)
        await asyncio.gather(runner.run_job(r1), runner.run_job(r2))
        return runner, r1, r2

    runner, r1, r2 = _scenario(scenario)
    assert r1.id != r2.id
    assert r1.status == JOB_DONE and r2.status == JOB_DONE
    assert r1._stage_states["identify"]["log_tail"] == "job1 log"  # tail_lines 去尾换行


def test_stage_log_event(db, events, catalog, fake_executor):
    """TC-BIO-SSE-004：stage_log 事件随执行发布（log 非空）。"""

    async def scenario():
        runner = make_runner(db, events, fake_executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        return runner, record

    runner, record = _scenario(scenario)
    log_events = [e for e in _collected(events, record.id) if e["type"] == "stage_log"]
    assert log_events
    assert all(e["data"]["log"] for e in log_events)


def test_event_seq_strictly_increasing(db, events, catalog, fake_executor):
    """TC-BIO-SSE-013：seq 严格递增。"""

    async def scenario():
        runner = make_runner(db, events, fake_executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        return runner, record

    runner, record = _scenario(scenario)
    seqs = [e["seq"] for e in _collected(events, record.id)]
    assert seqs == sorted(seqs)
    assert len(set(seqs)) == len(seqs)


# ---------------------------------------------------------------------------
# ensure_samples_yaml 配对修复（2026-09-17 水稻 M34 真实缺陷回归）
# ---------------------------------------------------------------------------

def _mk_fq(dirp, name):
    p = dirp / name
    p.write_text(">seq\nACGT\n")
    return p


def test_samples_pairing_symlink_alias_dedup(tmp_path):
    """TC-BIO-FIX-01：目录含 f1/r2 真实文件 + _1/_2 符号链接别名时，
    只产出 N 个配对样本，无 _r2.fq 孤儿条目，无 _2 别名抢配。"""
    from app.services.bioflow.runner import ensure_samples_yaml

    src = tmp_path / "fastq"
    src.mkdir()
    for sid in ("CRR1", "CRR2", "CRR3"):
        sub = src / sid
        sub.mkdir()
        real_r1 = _mk_fq(sub, f"{sid}_f1.fq.gz")
        real_r2 = _mk_fq(sub, f"{sid}_r2.fq.gz")
        # 别名符号链接（指向真实文件；模拟 M34 水稻目录实况）
        (sub / f"{sid}_1.fq.gz").symlink_to(real_r1)
        (sub / f"{sid}_2.fq.gz").symlink_to(real_r2)

    job_in = tmp_path / "job" / "input"
    job_in.mkdir(parents=True)
    mounts = ensure_samples_yaml(job_in, source_roots=[src])

    import yaml
    samples = yaml.safe_load((job_in / "samples.yaml").read_text())["samples"]
    assert len(samples) == 3, f"期望 3 个配对样本,实际 {list(samples)}"
    assert all(not n.endswith("_r2.fq") for n in samples)
    for sid in ("CRR1", "CRR2", "CRR3"):
        s = samples[sid]
        assert s["r1"].endswith(f"{sid}_f1.fq.gz")
        assert s["r2"].endswith(f"{sid}_r2.fq.gz")
    assert mounts


def test_samples_pairing_real_r2_not_orphaned(tmp_path):
    """TC-BIO-FIX-02：真实 _r2 与别名 _2 并存时，r2 必须取真实文件，
    且 _2 别名不产生孤儿样本条目。"""
    from app.services.bioflow.runner import ensure_samples_yaml

    src = tmp_path / "fastq"
    src.mkdir()
    sub = src / "S1"
    sub.mkdir()
    real_r1 = _mk_fq(sub, "S1_f1.fq.gz")
    real_r2 = _mk_fq(sub, "S1_r2.fq.gz")
    (sub / "S1_1.fq.gz").symlink_to(real_r1)
    (sub / "S1_2.fq.gz").symlink_to(real_r2)

    job_in = tmp_path / "job" / "input"
    job_in.mkdir(parents=True)
    ensure_samples_yaml(job_in, source_roots=[src])

    import yaml
    samples = yaml.safe_load((job_in / "samples.yaml").read_text())["samples"]
    assert len(samples) == 1, f"期望 1 个配对样本,实际 {list(samples)}"
    s = samples["S1"]
    assert s["r1"].endswith("S1_f1.fq.gz")
    assert s["r2"].endswith("S1_r2.fq.gz")


def test_samples_pairing_unpaired_r2_kept(tmp_path):
    """TC-BIO-FIX-03：只有单端 r2（无 r1）时保留孤儿条目（原契约）。"""
    from app.services.bioflow.runner import ensure_samples_yaml

    src = tmp_path / "fastq"
    src.mkdir()
    _mk_fq(src, "WX_alone_r2.fq.gz")

    job_in = tmp_path / "job" / "input"
    job_in.mkdir(parents=True)
    ensure_samples_yaml(job_in, source_roots=[src])

    import yaml
    samples = yaml.safe_load((job_in / "samples.yaml").read_text())["samples"]
    assert len(samples) == 1
    s = next(iter(samples.values()))
    assert not s.get("r1")
    assert s["r2"].endswith("WX_alone_r2.fq.gz")


# ---------------------------------------------------------------------------
# 阶段进度条（2026-09-17：流式日志 + [PROGRESS] 标记解析 + stage_progress 事件）
# ---------------------------------------------------------------------------


def test_parse_progress_marker_table():
    """parse_progress_marker 表驱动:样本/拷贝标记识别,无关行 None。"""
    from app.services.bioflow.runner import parse_progress_marker

    cases = [
        ("[PROGRESS] Processing: SRR1", {"kind": "sample", "message": "SRR1"}),
        ("  [PROGRESS] Processing:  sample A ", {"kind": "sample", "message": "sample A"}),
        ("[PROGRESS] Copying results to final directory...",
         {"kind": "copy", "percent": 95}),
        ("[PROGRESS] Copying results to final", {"kind": "copy", "percent": 95}),
        ("[INFO] not progress", None),
        ("Processing: SRR1", None),  # 无标记前缀不识别
        ("", None),
    ]
    for line, expected in cases:
        assert parse_progress_marker(line) == expected, f"line={line!r}"


class FakeStreamingExecutor:
    """mock docker 流式执行器：记录 on_log 回调并在调用中分块推送进度标记。"""

    def __init__(self, chunks: list[str], exit_code: int = 0):
        self.chunks = chunks
        self.exit_code = exit_code
        self.calls: list[dict] = []
        self.on_log = None

    async def __call__(self, stage, input_dir, workspace, params=None,
                       log_path=None, **kwargs):
        self.calls.append({"stage": stage})
        self.on_log = kwargs.get("on_log")
        for chunk in self.chunks:
            if self.on_log is not None:
                await self.on_log(chunk)
        # 写成功 manifest(默认 FakeExecutor 语义:退出码 0 → success)
        if self.exit_code == 0:
            from conftest import FakeExecutor
            FakeExecutor._write_manifest(self, workspace, stage, "success")
        return {"exit_code": self.exit_code, "timed_out": False,
                "log_tail": "".join(self.chunks), "cmd": []}


def test_stage_progress_events_rnaseq(db, events, catalog):
    """TC-BIO-PROG-01：rnaseq_qc 白名单阶段收到 [PROGRESS] → 发布
    stage_progress(percent=seen/total,单调递增);成功后强制 100。"""
    chunks = [
        "Input validation passed: 2 sample(s) ready\n",
        "Processing 2 samples...\n",
        "[PROGRESS] Processing: A\n",
        "fastp done\n",
        "[PROGRESS] Processing: B\n",
    ]
    executor = FakeStreamingExecutor(chunks)

    async def scenario():
        runner = make_runner(db, events, executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["rnaseq_qc"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        return runner, record

    runner, record = _scenario(scenario)
    assert runner.stage_status(record, "rnaseq_qc") == STAGE_DONE
    prog = [e["data"] for e in _collected(events, record.id)
            if e["type"] == "stage_progress"]
    assert prog, "应发布 stage_progress 事件"
    percents = [p["percent"] for p in prog]
    assert percents[-1] == 100  # 终态强制 100
    assert percents[:-1] == sorted(percents[:-1])  # 中间值单调
    assert any(p["percent"] > 0 for p in prog)  # 有实际进度(非只有 0/100)
    assert prog[-1]["message"] == ""  # 终态清空 message


def test_stage_progress_not_for_non_rnaseq(db, events, catalog):
    """TC-BIO-PROG-02：非白名单阶段(identify)不发布 stage_progress。"""
    chunks = ["Processing 5 samples...\n", "[PROGRESS] Processing: A\n"]
    executor = FakeStreamingExecutor(chunks)

    async def scenario():
        runner = make_runner(db, events, executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        return runner, record

    runner, record = _scenario(scenario)
    assert runner.stage_status(record, "identify") == STAGE_DONE
    prog = [e for e in _collected(events, record.id) if e["type"] == "stage_progress"]
    assert prog == []


def test_incremental_stage_log_events(db, events, catalog):
    """TC-BIO-PROG-03：on_log 分块 → 增量 stage_log 发布 + 终态 log_tail 并存。"""
    executor = FakeStreamingExecutor(["chunk1\n", "chunk2\n"])

    async def scenario():
        runner = make_runner(db, events, executor, catalog)
        record = await runner.create_job(
            "s1", "t", ["identify"], auto_continue=True, auto_start=False)
        await runner.run_job(record)
        return runner, record

    runner, record = _scenario(scenario)
    log_events = [e["data"] for e in _collected(events, record.id)
                  if e["type"] == "stage_log"]
    assert log_events
    joined = "".join(e["log"] for e in log_events)
    assert "chunk1" in joined and "chunk2" in joined
    assert record._stage_states["identify"]["log_tail"]  # 终态 log_tail 仍在