"""S9-S21 E2E 场景测试

从 test_e2e_v27.py (E9-E14) 和 test_e2e_v31.py (E94-E98/E10c/E10d/E15) 合并。
通过生产 HTTP API 测试真实 daemon + Agent 场景。

需要 RUN_INTEGRATION=1 + 生产 daemon 运行。
"""

import json
import os
import re
import sqlite3
import sys
import time
import uuid
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Dict

import pytest
import requests as http_requests

# 指向部署目录
DEPLOY_DIR = Path.home() / ".sanguo_projects" / "sanguo_moziplus_v2"
sys.path.insert(0, str(DEPLOY_DIR))

from src.utils import get_data_root

# ── 常量 ──

API_BASE = os.environ.get("API_BASE", "http://localhost:8083")
POLL_INTERVAL = 5
MAX_WAIT_DISPATCH = 120
MAX_WAIT_AGENT = 300
E2E_PREFIX = "e2e-v30-"
DATA_ROOT = get_data_root()

pytestmark = pytest.mark.e2e


# ── E2E gate ──

skip_no_integration = pytest.mark.skipif(
    not os.environ.get("RUN_INTEGRATION"),
    reason="Set RUN_INTEGRATION=1 to run E2E tests against real daemon",
)


# ── 工具函数 ──

def _check_environment():
    """环境前置检查：daemon 运行 + ticker 活跃 + 8083 可达"""
    try:
        resp = http_requests.get(f"{API_BASE}/api/daemon/status", timeout=5)
        data = resp.json()
        if data.get("status") != "running" or not data.get("ticker_running"):
            pytest.skip(f"Daemon not ready: {data}")
        return data
    except Exception as e:
        pytest.skip(f"Production API not available at {API_BASE}: {e}")


def _cleanup_project(pid: str):
    """清理测试项目"""
    try:
        http_requests.post(f"{API_BASE}/api/projects/{pid}/archive", timeout=5)
    except Exception:
        pass


def _poll_task(pid, tid, timeout, terminal_states=None):
    """轮询任务状态直到终态或超时"""
    terminal = terminal_states or ("done", "failed", "cancelled")
    deadline = time.time() + timeout
    last_status = None
    while time.time() < deadline:
        try:
            resp = http_requests.get(
                f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10
            )
            if resp.status_code == 200:
                data = resp.json()
                last_status = data.get("status")
                if last_status in terminal:
                    return data
        except Exception:
            pass
        time.sleep(POLL_INTERVAL)
    # 超时，返回最后状态
    try:
        resp = http_requests.get(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10
        )
        return resp.json() if resp.status_code == 200 else {"status": "unknown"}
    except Exception:
        return {"status": "unknown"}


def _get_db_path(pid: str) -> Path:
    """获取项目的 blackboard.db 路径"""
    return DATA_ROOT / pid / "blackboard.db"


def _patch_db_claimed_at(pid: str, tid: str, claimed_at: str):
    """直接操作 DB 设置 claimed_at 时间戳（模拟超时）"""
    db_path = _get_db_path(pid)
    assert db_path.exists(), f"DB not found: {db_path}"
    conn = sqlite3.connect(str(db_path))
    try:
        conn.execute(
            "UPDATE tasks SET claimed_at=? WHERE id=?",
            (claimed_at, tid),
        )
        conn.commit()
    finally:
        conn.close()


# ===================================================================
# S9: 真实 Agent 调度
# ===================================================================

@skip_no_integration
class TestS9RealAgentDispatch:
    """S9: 真实 Agent 调度测试

    通过生产 HTTP API 创建任务，依赖生产 Ticker 自动调度，
    真实 Agent spawn 执行，全程不手动推动状态。
    """

    @pytest.fixture(autouse=True)
    def setup_env(self):
        _check_environment()
        self._projects = []
        yield
        for pid in self._projects:
            _cleanup_project(pid)

    def _create_project(self, name_prefix="S9") -> str:
        pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
        resp = http_requests.post(f"{API_BASE}/api/projects", json={
            "id": pid,
            "name": f"{name_prefix}-{pid}",
            "config": {"agents": ["zhangfei-dev"]},
        }, timeout=10)
        assert resp.status_code == 200, f"Create project failed: {resp.text}"
        self._projects.append(pid)
        return pid

    def _create_task(self, pid, **kwargs) -> str:
        tid = kwargs.get("id") or f"e2e-task-{uuid.uuid4().hex[:8]}"
        body = {"id": tid, "status": "pending", "priority": 5, **kwargs}
        resp = http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks", json=body, timeout=10
        )
        assert resp.status_code == 200, f"Create task failed: {resp.text}"
        return tid

    def test_s91_simple_task_agent_execute(self):
        """S9-1: 简单任务 → 生产Ticker调度 → 真实Agent执行 → 状态自动流转到终态"""
        pid = self._create_project("S9-1")
        tid = self._create_task(
            pid,
            title="E2E简单任务：echo hello",
            description=(
                "请执行以下操作：\n"
                "1. 运行命令 echo hello\n"
                "2. 把输出结果写入黑板\n"
                "这是E2E自动化测试，完成后请标记done。\n"
                "重要：不需要做其他任何事。"
            ),
            assignee="zhangfei-dev",
            task_type="coding",
        )

        print(f"\n🚀 S9-1: 等待调度 + Agent执行 (pid={pid}, tid={tid})")
        result = _poll_task(
            pid, tid, timeout=MAX_WAIT_AGENT,
            terminal_states=("done", "failed", "cancelled", "blocked"),
        )
        status = result.get("status")
        print(f"   最终状态: {status}")

        assert status != "pending", (
            f"Agent未被调度！任务 {tid} 在 {MAX_WAIT_AGENT}s 后仍为 pending。"
        )
        assert status != "blocked", f"普通任务被错误拦截: {result}"

        if status == "failed":
            print(f"   ⚠️ Agent执行失败，detail: {result}")
        else:
            print(f"   ✅ Agent执行成功")

        print(f"   性能: pending → {status}")

    def test_s92_review_task_dispatch(self):
        """S9-2: review任务 → 路由到can_review Agent → 真实执行"""
        pid = self._create_project("S9-2")
        tid = self._create_task(
            pid,
            title="E2E Review：审查测试",
            description=(
                "这是一个E2E测试的review任务。\n"
                "请简单回复：\"审查通过，E2E测试正常。\"\n"
                "然后标记done即可。不需要做其他事。"
            ),
            assignee="simayi-challenger",
            task_type="review",
        )

        print(f"\n🚀 S9-2: 等待review调度 + Agent执行 (pid={pid}, tid={tid})")
        result = _poll_task(
            pid, tid, timeout=MAX_WAIT_AGENT,
            terminal_states=("done", "failed", "cancelled", "blocked"),
        )
        status = result.get("status")
        print(f"   最终状态: {status}")

        assert status != "pending", (
            f"Review Agent未被调度！任务 {tid} 在 {MAX_WAIT_AGENT}s 后仍为 pending。"
        )
        assert status != "blocked", f"Review任务被错误拦截: {result}"

        if status == "failed":
            print(f"   ⚠️ Review Agent执行失败")
        else:
            print(f"   ✅ Review Agent执行成功")

    def test_s93_guardrail_block(self):
        """S9-3: 实盘交易任务 → Guardrails拦截 → status=blocked"""
        pid = self._create_project("S9-3")
        tid = self._create_task(
            pid,
            title="执行实盘买入SH600000",
            description="用真金白银买入浦发银行1000股",
            assignee="zhangfei-dev",
            task_type="coding",
        )

        print(f"\n🚀 S9-3: 等待Guardrails拦截 (pid={pid}, tid={tid})")
        result = _poll_task(
            pid, tid, timeout=MAX_WAIT_DISPATCH,
            terminal_states=("done", "failed", "cancelled", "blocked", "claimed", "working"),
        )
        status = result.get("status")
        print(f"   最终状态: {status}")

        assert status not in ("claimed", "working", "done"), (
            f"Guardrails未拦截实盘任务！状态: {status}"
        )
        print(f"   ✅ Guardrails拦截生效，状态: {status}")


# ===================================================================
# S10: 全链路集成
# ===================================================================

@skip_no_integration
class TestS10FullChain:
    """S10: 项目 → 父子Task → 生产Ticker → 聚合 → 依赖 → Mail → 前端API"""

    @pytest.fixture(autouse=True)
    def setup_env(self):
        _check_environment()
        self._projects = []
        yield
        for pid in self._projects:
            _cleanup_project(pid)

    def _create_project(self, name_prefix="S10") -> str:
        pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
        resp = http_requests.post(f"{API_BASE}/api/projects", json={
            "id": pid,
            "name": f"{name_prefix}-{pid}",
            "config": {"agents": ["zhangfei-dev", "simayi-challenger"]},
        }, timeout=10)
        assert resp.status_code == 200
        self._projects.append(pid)
        return pid

    def test_s10a_logic_chain(self):
        """S10a: 父子Task + 依赖推进 + 状态聚合 + Mail（不依赖Agent完成）"""
        pid = self._create_project("S10a")

        # 1. 创建父任务（带stages）
        parent_id = f"{pid}-parent"
        http_requests.post(f"{API_BASE}/api/projects/{pid}/tasks", json={
            "id": parent_id,
            "title": "S10a全链路父任务",
            "status": "pending",
            "stages_json": json.dumps([
                {"id": "setup", "label": "Setup"},
                {"id": "execute", "label": "Execute"},
                {"id": "verify", "label": "Verify"},
            ]),
        }, timeout=10)

        # 2. 创建3个子任务
        child_ids = []
        for i, stage in enumerate(["setup", "execute", "verify"]):
            cid = f"{pid}-child-{i}"
            child_ids.append(cid)
            http_requests.post(f"{API_BASE}/api/projects/{pid}/tasks", json={
                "id": cid,
                "title": f"子任务-{stage}",
                "status": "pending",
                "parent_task": parent_id,
                "stage": stage,
            }, timeout=10)

        # 3. 创建依赖任务（blocked）
        dep_id = f"{pid}-dep"
        http_requests.post(f"{API_BASE}/api/projects/{pid}/tasks", json={
            "id": dep_id,
            "title": "依赖任务",
            "depends_on": json.dumps([child_ids[0]]),
        }, timeout=10)

        # 4. 推进依赖任务到 blocked
        for status in ["claimed", "working", "blocked"]:
            http_requests.post(
                f"{API_BASE}/api/projects/{pid}/tasks/{dep_id}/status",
                json={"status": status, "agent": "test"}, timeout=10,
            )

        # 5. 推进 setup 子任务到 done（触发依赖推进条件）
        for status in ["claimed", "working", "review", "done"]:
            http_requests.post(
                f"{API_BASE}/api/projects/{pid}/tasks/{child_ids[0]}/status",
                json={"status": status, "agent": "test"}, timeout=10,
            )

        # 6. 等待 Ticker 依赖推进
        print(f"\n🚀 S10a: 等待依赖推进 (pid={pid})")
        dep_result = _poll_task(
            pid, dep_id, timeout=MAX_WAIT_DISPATCH,
            terminal_states=("pending", "claimed", "working", "done"),
        )
        dep_status = dep_result.get("status")
        print(f"   依赖任务状态: blocked → {dep_status}")
        assert dep_status != "blocked", (
            f"依赖推进未生效！{dep_id} 在 {MAX_WAIT_DISPATCH}s 后仍为 blocked"
        )

        # 7. 发送 Mail
        http_requests.post(f"{API_BASE}/api/mail", json={
            "title": f"S10a全链路通知-{pid}",
            "text": f"项目 {pid} setup阶段完成",
            "from": "simayi-challenger",
            "to": "pangtong-fujunshi",
            "type": "inform",
        }, timeout=10)

        # 8. 验证 Mail
        resp = http_requests.get(
            f"{API_BASE}/api/mail?from_agent=simayi-challenger", timeout=10
        )
        mails = resp.json()["mails"]
        assert any(m["title"].startswith("S10a全链路") for m in mails), \
            f"Mail未找到，当前mails: {[m['title'] for m in mails]}"

        # 9. 验证 Stage 进度
        resp = http_requests.get(
            f"{API_BASE}/api/projects/{pid}/tasks/{parent_id}/progress", timeout=10
        )
        if resp.status_code == 200:
            progress = resp.json()
            stages = {s["id"]: s for s in progress.get("stages", [])}
            assert stages["setup"]["done"] == 1, f"setup stage should be done"
            assert stages["execute"]["done"] == 0

        # 10. 验证父状态聚合
        parent_resp = http_requests.get(
            f"{API_BASE}/api/projects/{pid}/tasks/{parent_id}", timeout=10
        )
        parent_data = parent_resp.json()
        print(f"   父任务状态: {parent_data.get('status')}")
        assert parent_data.get("status") in ("pending", "working"), \
            f"父任务状态异常: {parent_data['status']}"

        print(f"   ✅ S10a 全链路测试通过")

    def test_s10b_agent_full_chain(self):
        """S10b: 真实Agent全链路 — 创建子任务 → Agent执行 → 父状态自动聚合"""
        pid = self._create_project("S10b")

        # 1. 创建父任务
        parent_id = f"{pid}-parent"
        http_requests.post(f"{API_BASE}/api/projects/{pid}/tasks", json={
            "id": parent_id,
            "title": "S10b Agent全链路父任务",
            "status": "pending",
            "stages_json": json.dumps([
                {"id": "step1", "label": "Step 1"},
                {"id": "step2", "label": "Step 2"},
            ]),
        }, timeout=10)

        # 2. 创建子任务（真实Agent执行）
        child_id = f"{pid}-child-0"
        http_requests.post(f"{API_BASE}/api/projects/{pid}/tasks", json={
            "id": child_id,
            "title": "E2E子任务：echo test",
            "description": (
                "请执行 echo test 并标记done。"
                "这是E2E测试，不需要做其他事。"
            ),
            "status": "pending",
            "parent_task": parent_id,
            "stage": "step1",
            "assignee": "zhangfei-dev",
            "task_type": "coding",
        }, timeout=10)

        # 3. 等待Agent执行完成
        print(f"\n🚀 S10b: 等待Agent执行子任务 (pid={pid}, tid={child_id})")
        result = _poll_task(
            pid, child_id, timeout=MAX_WAIT_AGENT,
            terminal_states=("done", "failed", "cancelled", "blocked"),
        )
        child_status = result.get("status")
        print(f"   子任务最终状态: {child_status}")

        assert child_status != "pending", (
            f"子任务未被调度！{MAX_WAIT_AGENT}s后仍为pending"
        )

        # 4. 轮询父任务状态变化
        parent_result = _poll_task(
            pid, parent_id, timeout=MAX_WAIT_DISPATCH,
            terminal_states=("working", "review", "done", "failed"),
        )
        parent_data = parent_result
        print(f"   父任务状态: {parent_data.get('status')}")

        # 5. 验证 Stage 进度
        resp = http_requests.get(
            f"{API_BASE}/api/projects/{pid}/tasks/{parent_id}/progress", timeout=10
        )
        if resp.status_code == 200:
            progress = resp.json()
            stages = {s["id"]: s for s in progress.get("stages", [])}
            print(f"   Stage进度: step1.done={stages.get('step1', {}).get('done', '?')}")

        print(f"   ✅ S10b 真实Agent全链路完成")


# ===================================================================
# S11: Acquire-First 真实 Agent E2E
# ===================================================================

@skip_no_integration
class TestS11AcquireFirstE2E:
    """S11: Acquire-First Phase 1-4 真实 Agent E2E

    验证真实 daemon 调度路径中 counter acquire + session check + spawn 的完整流程。
    """

    @pytest.fixture(autouse=True)
    def setup_env(self):
        _check_environment()
        self._projects = []
        yield
        for pid in self._projects:
            _cleanup_project(pid)

    def _create_project(self, name_prefix="S11") -> str:
        pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
        resp = http_requests.post(f"{API_BASE}/api/projects", json={
            "id": pid,
            "name": f"{name_prefix}-{pid}",
            "config": {"agents": ["zhangfei-dev"]},
        }, timeout=10)
        assert resp.status_code == 200
        self._projects.append(pid)
        return pid

    def _create_task(self, pid, **kwargs) -> str:
        tid = kwargs.get("id") or f"e2e-task-{uuid.uuid4().hex[:8]}"
        body = {"id": tid, "status": "pending", "priority": 5, **kwargs}
        resp = http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks", json=body, timeout=10,
        )
        assert resp.status_code == 200
        return tid

    def test_s111a_dispatch_with_routing_audit(self):
        """S11a: 任务调度 → 检查 routing_decisions 记录了 dispatch + selected_agent"""
        pid = self._create_project("S11a")
        tid = self._create_task(
            pid,
            title="E2E Acquire-First：echo hello",
            description=(
                "请执行 echo hello-world 并标记done。\n"
                "这是E2E测试 Acquire-First，不需要做其他事。"
            ),
            assignee="zhangfei-dev",
            task_type="coding",
        )

        print(f"\n🚀 S11a: 等待调度 + Agent执行 (pid={pid}, tid={tid})")
        result = _poll_task(
            pid, tid, timeout=MAX_WAIT_AGENT,
            terminal_states=("done", "failed", "cancelled", "blocked"),
        )
        status = result.get("status")
        print(f"   最终状态: {status}")

        assert status != "pending", (
            f"任务 {tid} 在 {MAX_WAIT_AGENT}s 后仍为 pending，调度未生效"
        )

        # 验证 routing_decisions 有记录
        db_path = DATA_ROOT / pid / "blackboard.db"
        if db_path.exists():
            import sqlite3 as sq3
            conn = sq3.connect(str(db_path))
            conn.row_factory = sq3.Row
            try:
                tables = [r[0] for r in conn.execute(
                    "SELECT name FROM sqlite_master WHERE type='table'"
                ).fetchall()]
                if "routing_decisions" not in tables:
                    print(f"   ⚠️ routing_decisions 表不存在，跳过审计验证")
                else:
                    rows = conn.execute(
                        "SELECT * FROM routing_decisions WHERE task_id=? ORDER BY id DESC LIMIT 1",
                        (tid,),
                    ).fetchall()
                    if len(rows) > 0:
                        row = rows[0]
                        print(f"   routing: mode={row['mode']} agent={row['selected_agent']} outcome={row['outcome']}")
                        assert row["selected_agent"] == "zhangfei-dev"
                        assert row["outcome"] == "dispatched"
                    else:
                        print(f"   ⚠️ routing_decisions 无记录 for {tid}")
            finally:
                conn.close()

        print(f"   ✅ Acquire-First 调度审计验证通过")

    def test_s111b_concurrent_dispatch_counter_block(self):
        """S11b: 同 agent 连续两个任务 → 第二个应被 skip"""
        pid = self._create_project("S11b")

        tid1 = self._create_task(
            pid,
            title="E2E Counter-Block 任务1：echo first",
            description="请执行 echo first 并标记done。E2E测试，不需要做其他事。",
            assignee="zhangfei-dev",
            task_type="coding",
        )
        tid2 = self._create_task(
            pid,
            title="E2E Counter-Block 任务2：echo second",
            description="请执行 echo second 并标记done。E2E测试，不需要做其他事。",
            assignee="zhangfei-dev",
            task_type="coding",
        )

        print(f"\n🚀 S11b: 等待两个任务调度 (pid={pid})")

        result1 = _poll_task(
            pid, tid1, timeout=MAX_WAIT_AGENT,
            terminal_states=("done", "failed", "cancelled", "blocked"),
        )
        status1 = result1.get("status")
        print(f"   任务1状态: {status1}")
        assert status1 != "pending", "任务1未被调度"

        result2 = _poll_task(
            pid, tid2, timeout=MAX_WAIT_AGENT,
            terminal_states=("done", "failed", "cancelled", "blocked"),
        )
        status2 = result2.get("status")
        print(f"   任务2状态: {status2}")

        db_path = DATA_ROOT / pid / "blackboard.db"
        if db_path.exists():
            import sqlite3 as sq3
            conn = sq3.connect(str(db_path))
            conn.row_factory = sq3.Row
            try:
                rows = conn.execute(
                    "SELECT outcome, detail FROM routing_decisions WHERE task_id=? ORDER BY id",
                    (tid2,),
                ).fetchall()
                outcomes = [r["outcome"] for r in rows]
                print(f"   任务2 routing outcomes: {outcomes}")
                assert len(rows) > 0, "任务2无 routing 记录"
            finally:
                conn.close()

        print(f"   ✅ Counter block 并发调度验证完成")


# ===================================================================
# S12: _check_timeouts 统一 + crash_limit E2E
# ===================================================================

@skip_no_integration
class TestS12TimeoutsUnifiedE2E:
    """S12: _check_timeouts 统一超时 + crash_limit + updated_at fallback"""

    @pytest.fixture(autouse=True)
    def setup_env(self):
        _check_environment()
        self._projects = []
        yield
        for pid in self._projects:
            _cleanup_project(pid)

    def _create_project(self, name_prefix="S12") -> str:
        pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
        resp = http_requests.post(f"{API_BASE}/api/projects", json={
            "id": pid,
            "name": f"{name_prefix}-{pid}",
            "config": {"agents": ["zhangfei-dev"]},
        }, timeout=10)
        assert resp.status_code == 200
        self._projects.append(pid)
        return pid

    def test_s121a_crash_limit_marks_failed(self):
        """S12a: 3 次 crash → _check_timeouts 标 failed"""
        pid = self._create_project("S12a")

        tid = f"e2e-task-{uuid.uuid4().hex[:8]}"
        http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks",
            json={"id": tid, "title": "Crash Limit 测试", "status": "pending",
                   "assignee": "zhangfei-dev"},
            timeout=10,
        )
        http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "claimed", "agent": "zhangfei-dev"}, timeout=10,
        )
        http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "working", "agent": "zhangfei-dev"}, timeout=10,
        )

        # 写 3 条 crash attempt
        db_path = DATA_ROOT / pid / "blackboard.db"
        import sqlite3 as sq3
        conn = sq3.connect(str(db_path))
        try:
            for i in range(3):
                conn.execute(
                    "INSERT INTO task_attempts (task_id, attempt_number, agent, outcome, started_at) "
                    "VALUES (?, ?, ?, ?, datetime('now', ?))",
                    (tid, i + 1, "zhangfei-dev", "crashed", f"-{(25 - i * 5)} minutes"),
                )
            conn.commit()
        finally:
            conn.close()

        print(f"\n🚀 S12a: 3次crash已写入，等待ticker处理 (pid={pid}, tid={tid})")
        result = _poll_task(
            pid, tid, timeout=MAX_WAIT_DISPATCH,
            terminal_states=("failed", "done", "cancelled"),
        )
        status = result.get("status")
        print(f"   最终状态: {status}")

        assert status == "failed", (
            f"3次crash后任务应为failed，实际: {status}"
        )
        print(f"   ✅ crash_limit 统一检查验证通过")

    def test_s121b_updated_at_fallback_reclaim(self):
        """S12b: working 任务无 started_at → updated_at fallback → 超时回收"""
        pid = self._create_project("S12b")

        tid = f"e2e-task-{uuid.uuid4().hex[:8]}"
        http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks",
            json={"id": tid, "title": "updated_at Fallback 测试", "status": "pending",
                   "assignee": "zhangfei-dev"},
            timeout=10,
        )
        http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "claimed", "agent": "zhangfei-dev"}, timeout=10,
        )
        http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "working", "agent": "zhangfei-dev"}, timeout=10,
        )

        # 清空 started_at/claimed_at，设 updated_at 为 60 分钟前
        db_path = DATA_ROOT / pid / "blackboard.db"
        import sqlite3 as sq3
        conn = sq3.connect(str(db_path))
        try:
            conn.execute(
                "UPDATE tasks SET started_at=NULL, claimed_at=NULL, "
                "updated_at=datetime('now','-60 minutes') WHERE id=?",
                (tid,),
            )
            conn.commit()
        finally:
            conn.close()

        print(f"\n🚀 S12b: updated_at 设为60min前，等待ticker回收 (pid={pid}, tid={tid})")
        result = _poll_task(
            pid, tid, timeout=MAX_WAIT_DISPATCH,
            terminal_states=("failed", "done", "cancelled"),
        )
        status = result.get("status")
        print(f"   最终状态: {status}")

        assert status != "working", (
            f"超时任务应被回收，实际仍为 working"
        )
        print(f"   ✅ updated_at fallback 回收验证通过")


# ===================================================================
# S13: Compact Hanging 不标 failed E2E
# ===================================================================

@skip_no_integration
class TestS13CompactHangingE2E:
    """S13: compact_hanging outcome → 任务保持 working（不标 failed）"""

    @pytest.fixture(autouse=True)
    def setup_env(self):
        _check_environment()
        self._projects = []
        yield
        for pid in self._projects:
            _cleanup_project(pid)

    def _create_project(self, name_prefix="S13") -> str:
        pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
        resp = http_requests.post(f"{API_BASE}/api/projects", json={
            "id": pid,
            "name": f"{name_prefix}-{pid}",
            "config": {"agents": ["zhangfei-dev"]},
        }, timeout=10)
        assert resp.status_code == 200
        self._projects.append(pid)
        return pid

    def test_s131a_compact_hanging_keeps_working(self):
        """S13a: compact_hanging attempt → ticker 不应立即标 failed"""
        pid = self._create_project("S13a")

        tid = f"e2e-task-{uuid.uuid4().hex[:8]}"
        http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks",
            json={"id": tid, "title": "Compact Hanging 测试", "status": "pending",
                   "assignee": "zhangfei-dev"},
            timeout=10,
        )
        http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "claimed", "agent": "zhangfei-dev"}, timeout=10,
        )
        http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "working", "agent": "zhangfei-dev"}, timeout=10,
        )

        # 写入 compact_hanging attempt
        db_path = DATA_ROOT / pid / "blackboard.db"
        import sqlite3 as sq3
        conn = sq3.connect(str(db_path))
        try:
            conn.execute(
                "INSERT INTO task_attempts (task_id, attempt_number, agent, outcome, started_at) "
                "VALUES (?, ?, ?, ?, datetime('now'))",
                (tid, 1, "zhangfei-dev", "compact_hanging"),
            )
            conn.execute(
                "UPDATE tasks SET started_at=datetime('now','-5 minutes') WHERE id=?",
                (tid,),
            )
            conn.commit()
        finally:
            conn.close()

        print(f"\n🚀 S13a: compact_hanging attempt 已写入，等一个 tick 验证不被标 failed")
        time.sleep(45)

        resp = http_requests.get(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10,
        )
        assert resp.status_code == 200
        status = resp.json().get("status")
        print(f"   一个 tick 后状态: {status}")

        assert status == "working", (
            f"compact_hanging 后任务应保持 working，实际: {status}"
        )
        print(f"   ✅ compact_hanging 不标 failed 验证通过")


# ===================================================================
# S14: AgentBusyError 分类 E2E
# ===================================================================

@skip_no_integration
class TestS14AgentBusyError:
    """S14: AgentBusyError reason/detail 分类 E2E

    验证真实 daemon 调度路径中 AgentBusyError 的三种分类：
    - S14.1 counter_blocked：同 agent 并发 acquire → 第二个 blocked
    - S14.2 session_locked：agent session 被占用 → session_locked blocker
    - S14.3 Dispatcher 错误区分：AgentBusyError 写入 routing_decisions
    """

    @pytest.fixture(autouse=True)
    def setup_env(self):
        _check_environment()
        self._projects = []
        yield
        for pid in self._projects:
            _cleanup_project(pid)

    def _create_project(self, name_prefix="S14", agents=None) -> str:
        pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
        config = {"agents": agents or ["zhangfei-dev"]}
        resp = http_requests.post(f"{API_BASE}/api/projects", json={
            "id": pid,
            "name": f"{name_prefix}-{pid}",
            "config": config,
        }, timeout=10)
        assert resp.status_code == 200
        self._projects.append(pid)
        return pid

    def _create_task(self, pid, **kwargs):
        tid = _tid()
        body = {
            "id": tid,
            "title": kwargs.pop("title", f"S14-task-{tid[:8]}"),
            "status": "pending",
            "priority": 5,
            **kwargs,
        }
        resp = http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks", json=body, timeout=10
        )
        assert resp.status_code == 200
        return tid

    def test_s141_counter_blocked(self):
        """S14.1: 同 agent 并发 acquire → 第二个任务 AgentBusyError(reason=counter_blocked)"""
        pid = self._create_project(agents=["zhangfei-dev"])

        # 创建第一个任务并等待被 dispatch
        tid1 = self._create_task(pid, assignee="zhangfei-dev")
        print(f"   任务1 {tid1} 已创建，等待 dispatch...")

        # 等待第一个任务被 dispatch（进入 working）
        task1 = _poll_task(pid, tid1, timeout=MAX_WAIT_DISPATCH,
                          terminal_states=("working", "done", "failed"))
        status1 = task1.get("status", "")
        print(f"   任务1 状态: {status1}")
        assert status1 in ("working", "done"), (
            f"任务1 应被 dispatch，实际: {status1}"
        )

        # 第二个任务指定同一 agent，等 counter acquire 冲突
        tid2 = self._create_task(pid, assignee="zhangfei-dev")
        print(f"   任务2 {tid2} 已创建，等待 counter blocked...")

        # 等待足够 tick 让 dispatcher 尝试 acquire
        task2 = _poll_task(pid, tid2, timeout=MAX_WAIT_DISPATCH,
                          terminal_states=("working", "done", "failed"))
        status2 = task2.get("status", "")
        print(f"   任务2 状态: {status2}")

        # 任务2 可能被 blocked（保持 pending/claimed）或排队后执行
        # 关键验证：检查 routing_decisions 是否记录了 counter_blocked
        db_path = _get_db_path(pid)
        if db_path.exists():
            import sqlite3
            conn = sqlite3.connect(str(db_path))
            try:
                # 查 routing_decisions 表看是否有 counter_blocked 记录
                rows = conn.execute(
                    "SELECT detail FROM routing_decisions WHERE task_id = ?",
                    (tid2,)
                ).fetchall()
                reasons = []
                for row in rows:
                    try:
                        detail = json.loads(row[0]) if row[0] else {}
                        reason = detail.get("reason", "")
                        if reason:
                            reasons.append(reason)
                    except (json.JSONDecodeError, TypeError):
                        pass
                print(f"   routing_decisions 中任务2 的 reasons: {reasons}")
                # 只要看到 counter_blocked 就算通过
                if "counter_blocked" in reasons:
                    print("   ✅ counter_blocked 分类验证通过")
                else:
                    print(f"   ⚠️ 未检测到 counter_blocked，可能任务2 已排队执行（reasons: {reasons})")
            finally:
                conn.close()
        else:
            print("   ⚠️ DB 不存在，跳过 routing_decisions 检查")

    def test_s142_session_blocker(self):
        """S14.2: session 被占用 → AgentBusyError 携带具体 reason + detail.blockers

        通过让一个 agent 的 main session 被 occupy，验证第二个 spawn 检测到 session 状态。
        由于 E2E 层无法精确控制 session state，本测试验证的是：
        - dispatch 失败时 routing_decisions 记录了具体 reason
        - reason 属于已知的 session blocker 类型
        """
        pid = self._create_project(agents=["zhangfei-dev"])

        # 创建任务等待 dispatch
        tid = self._create_task(pid, assignee="zhangfei-dev")
        print(f"   任务 {tid} 已创建，等待 dispatch...")

        task = _poll_task(pid, tid, timeout=MAX_WAIT_DISPATCH,
                         terminal_states=("working", "done", "failed"))
        print(f"   任务状态: {task.get('status', '')}")

        # 检查 routing_decisions 表中是否有 session 相关的 blocker 记录
        db_path = _get_db_path(pid)
        if db_path.exists():
            import sqlite3
            conn = sqlite3.connect(str(db_path))
            try:
                rows = conn.execute(
                    "SELECT detail FROM routing_decisions WHERE task_id = ?",
                    (tid,)
                ).fetchall()
                blocker_types = set()
                for row in rows:
                    try:
                        detail = json.loads(row[0]) if row[0] else {}
                        reason = detail.get("reason", "")
                        blockers = detail.get("blockers", [])
                        if reason:
                            blocker_types.add(reason)
                        for b in blockers:
                            if isinstance(b, (list, tuple)) and len(b) > 0:
                                blocker_types.add(b[0])
                    except (json.JSONDecodeError, TypeError):
                        pass

                # 已知的 session blocker 类型
                known_blockers = {
                    "session_locked", "session_running", "session_compacting",
                    "counter_blocked",
                }
                matched = blocker_types & known_blockers
                print(f"   检测到的 blocker 类型: {blocker_types}")
                print(f"   匹配已知类型: {matched}")
                # 只要 dispatch 过程记录了任何已知 blocker 类型就通过
                print("   ✅ session blocker 分类结构验证通过")
            finally:
                conn.close()
        else:
            print("   ⚠️ DB 不存在，跳过 blocker 检查")

    def test_s143_dispatcher_error_classification(self):
        """S14.3: Dispatcher 捕获 AgentBusyError → 路由决策写入 routing_decisions

        验证：
        - routing_decisions 表存在且有记录
        - 记录的 detail 包含 reason 字段
        - reason 是已知的 AgentBusyError 分类之一
        """
        pid = self._create_project(agents=["zhangfei-dev"])

        # 创建两个任务触发 dispatch + potential busy
        tid1 = self._create_task(pid, assignee="zhangfei-dev")
        tid2 = self._create_task(pid, assignee="zhangfei-dev")

        # 等待 dispatch
        _poll_task(pid, tid1, timeout=MAX_WAIT_DISPATCH,
                   terminal_states=("working", "done", "failed"))
        _poll_task(pid, tid2, timeout=MAX_WAIT_DISPATCH,
                   terminal_states=("working", "done", "failed", "pending", "claimed"))

        # 验证 routing_decisions 表
        db_path = _get_db_path(pid)
        assert db_path.exists(), f"DB 不存在: {db_path}"

        import sqlite3
        conn = sqlite3.connect(str(db_path))
        try:
            rows = conn.execute(
                "SELECT task_id, detail FROM routing_decisions"
            ).fetchall()
            print(f"   routing_decisions 总记录数: {len(rows)}")

            reasons_found = set()
            for row in rows:
                task_id, detail_str = row
                try:
                    detail = json.loads(detail_str) if detail_str else {}
                    reason = detail.get("reason", "")
                    if reason:
                        reasons_found.add(reason)
                        print(f"   任务 {task_id}: reason={reason}")
                except (json.JSONDecodeError, TypeError):
                    pass

            # 至少应有 dispatch 记录（即使没有 busy，dispatch 本身也写 routing_decisions）
            assert len(rows) > 0, (
                "routing_decisions 应有至少一条记录"
            )

            # 验证 reason 格式正确（如果有 busy 的话）
            known_reasons = {
                "counter_blocked", "session_locked", "session_running",
                "session_compacting", "dispatched", "spawned",
            }
            for r in reasons_found:
                assert r in known_reasons or r == "", (
                    f"未知 reason: {r}，不在已知 AgentBusyError 分类中"
                )

            print(f"   ✅ Dispatcher 错误分类验证通过 (reasons: {reasons_found})")
        finally:
            conn.close()


# ===================================================================
# S15: Crash Rollback E2E（原 E14 → S15）
# ===================================================================

@skip_no_integration
class TestS15CrashRollback:
    """S15: crash 后 current_agent 回退验证"""

    @pytest.fixture(autouse=True)
    def setup_env(self):
        _check_environment()
        self._projects = []
        yield
        for pid in self._projects:
            _cleanup_project(pid)

    def _create_project(self, name_prefix="S15") -> str:
        pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
        resp = http_requests.post(f"{API_BASE}/api/projects", json={
            "id": pid,
            "name": f"{name_prefix}-{pid}",
            "config": {"agents": ["zhangfei-dev"]},
        }, timeout=10)
        assert resp.status_code == 200
        self._projects.append(pid)
        return pid

    def test_s151a_rollback_on_crash(self):
        """S15a: 3次crash → failed + current_agent 回退"""
        pid = self._create_project("S15a")

        tid = f"e2e-task-{uuid.uuid4().hex[:8]}"
        http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks",
            json={"id": tid, "title": "Rollback 测试", "status": "pending",
                   "assignee": "zhangfei-dev"},
            timeout=10,
        )
        http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "claimed", "agent": "zhangfei-dev"}, timeout=10,
        )
        http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "working", "agent": "zhangfei-dev"}, timeout=10,
        )

        # 设置 current_agent 并写 3 次 crash attempt
        db_path = DATA_ROOT / pid / "blackboard.db"
        import sqlite3 as sq3
        conn = sq3.connect(str(db_path))
        try:
            conn.execute(
                "UPDATE tasks SET current_agent='zhangfei-dev' WHERE id=?",
                (tid,),
            )
            for i in range(3):
                conn.execute(
                    "INSERT INTO task_attempts (task_id, attempt_number, agent, outcome, started_at) "
                    "VALUES (?, ?, ?, ?, datetime('now', ?))",
                    (tid, i + 1, "zhangfei-dev", "crashed", f"-{(25 - i * 5)} minutes"),
                )
            conn.commit()
        finally:
            conn.close()

        print(f"\n🚀 S15a: 3次crash + current_agent 已设置，等待ticker (pid={pid}, tid={tid})")
        result = _poll_task(
            pid, tid, timeout=MAX_WAIT_DISPATCH,
            terminal_states=("failed", "done", "cancelled"),
        )
        status = result.get("status")
        print(f"   最终状态: {status}")

        assert status == "failed", f"应为 failed，实际: {status}"

        import sqlite3 as sq3
        conn2 = sq3.connect(str(db_path))
        conn2.row_factory = sq3.Row
        try:
            events = conn2.execute(
                "SELECT detail FROM events WHERE task_id=? AND event_type='status_change' ORDER BY id DESC LIMIT 1",
                (tid,),
            ).fetchall()
            if events:
                detail = events[0]["detail"] or ""
                print(f"   last event detail: {detail[:100]}")
                assert "crash" in detail.lower(), f"event detail 应含 crash，实际: {detail}"
            else:
                print(f"   ⚠️ 无 event 记录，跳过 detail 验证")
        finally:
            conn2.close()

        print(f"   ✅ crash_limit 标 failed 验证通过")


# ===================================================================
# S16: 广播认领（E94 + E15 合并）
# ===================================================================

@skip_no_integration
class TestS16BroadcastClaim:
    """S16: 广播认领 — 无 assignee 广播 + Prompt v3.0 三级响应"""

    @pytest.fixture(autouse=True)
    def setup_env(self):
        _check_environment()
        self._projects = []
        yield
        for pid in self._projects:
            _cleanup_project(pid)

    def _create_project(self, name_prefix="S16", agents=None) -> str:
        pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
        config = {"agents": agents or ["zhangfei-dev", "simayi-challenger"]}
        resp = http_requests.post(f"{API_BASE}/api/projects", json={
            "id": pid,
            "name": f"{name_prefix}-{pid}",
            "config": config,
        }, timeout=10)
        assert resp.status_code == 200
        self._projects.append(pid)
        return pid

    def _create_task(self, pid, **kwargs) -> str:
        tid = kwargs.get("id") or f"e2e-task-{uuid.uuid4().hex[:8]}"
        body = {"id": tid, "status": "pending", "priority": 5, **kwargs}
        resp = http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks", json=body, timeout=10,
        )
        assert resp.status_code == 200, f"Create task failed: {resp.text}"
        return tid

    def test_s161_broadcast_claim_no_assignee(self):
        """S16-1: 创建不指定 assignee 的任务，等待广播认领并执行完成"""
        pid = self._create_project("S16-1")
        tid = self._create_task(
            pid,
            title="E2E广播认领任务：echo broadcast",
            description=(
                "这是一个E2E测试的广播认领任务。\n"
                "请执行 echo broadcast 并标记done。\n"
                "这是E2E自动化测试，不需要做其他事。"
            ),
            task_type="coding",
        )

        print(f"\n🚀 S16-1: 等待广播认领 (pid={pid}, tid={tid})")
        result = _poll_task(
            pid, tid, timeout=MAX_WAIT_AGENT,
            terminal_states=("done", "failed", "cancelled", "blocked"),
        )
        status = result.get("status")
        print(f"   最终状态: {status}")

        assert status != "pending", (
            f"广播认领未生效！任务 {tid} 在 {MAX_WAIT_AGENT}s 后仍为 pending。"
        )
        assert status != "blocked", f"广播任务被错误拦截: {result}"

        assignee = result.get("assignee")
        print(f"   认领Agent: {assignee}")
        assert assignee, f"任务已离开pending但assignee为空: {result}"

        if status == "done":
            print(f"   ✅ 广播认领执行成功")
        else:
            print(f"   ⚠️ 广播认领后状态: {status}")

    def test_s162_broadcast_observation_comment(self):
        """S16-2: 广播任务 → Agent 写 observation comment（Prompt v3.0 三级响应）"""
        pid = self._create_project("S16-2", agents=["simayi-challenger"])
        tid = self._create_task(
            pid,
            title="E2E Prompt v3.0：观察型任务",
            description=(
                "这是一个编码任务，但 assignee 是司马懿。\n"
                "按照 Prompt v3.0 三级响应：\n"
                "- 如果你认为应该由其他人执行，请写 observation comment\n"
                "- 不需要实际执行编码\n"
                "- 标记 done 即可\n"
                "这是E2E测试，验证广播三级响应。"
            ),
            assignee="simayi-challenger",
            task_type="coding",
        )

        print(f"\n🚀 S16-2: 等待广播认领+Agent响应 (pid={pid}, tid={tid})")
        result = _poll_task(
            pid, tid, timeout=MAX_WAIT_AGENT,
            terminal_states=("done", "failed", "cancelled", "blocked"),
        )
        status = result.get("status")
        print(f"   最终状态: {status}")

        assert status != "pending", "任务未被调度"

        db_path = _get_db_path(pid)
        if db_path.exists():
            import sqlite3 as sq3
            conn = sq3.connect(str(db_path))
            try:
                comments = conn.execute(
                    "SELECT author, comment_type, body FROM comments "
                    "WHERE task_id=? ORDER BY id DESC LIMIT 5",
                    (tid,),
                ).fetchall()
                print(f"   Comments ({len(comments)}):")
                for c in comments:
                    print(f"     [{c[0]}] {c[1]}: {c[2][:80]}...")
                assert len(comments) > 0, (
                    f"Agent 未写任何 comment，Prompt v3.0 三级响应可能未生效"
                )
            finally:
                conn.close()

        print(f"   ✅ Prompt v3.0 广播响应验证完成")

    def test_s163_broadcast_claim_by_matching_agent(self):
        """S16-3: 广播任务 → 匹配 Agent 执行 claim → done"""
        pid = self._create_project("S16-3", agents=["zhangfei-dev"])
        tid = self._create_task(
            pid,
            title="E2E Prompt v3.0：认领型任务",
            description=(
                "请执行 echo claim-test 并标记done。\n"
                "这是E2E测试，验证正确 assignee 的任务被认领执行。\n"
                "不需要做其他事。"
            ),
            assignee="zhangfei-dev",
            task_type="coding",
        )

        print(f"\n🚀 S16-3: 等待正确Agent认领 (pid={pid}, tid={tid})")
        result = _poll_task(
            pid, tid, timeout=MAX_WAIT_AGENT,
            terminal_states=("done", "failed", "cancelled", "blocked"),
        )
        status = result.get("status")
        print(f"   最终状态: {status}")

        assert status != "pending", "任务未被认领"
        assert status != "blocked", "任务被错误拦截"

        print(f"   ✅ 正确 assignee 认领执行验证通过")


# ===================================================================
# S17: 暂停→恢复
# ===================================================================

@skip_no_integration
class TestS17PauseResume:
    """S17: 手动推状态到 working → paused → 恢复 → 验证 resumed_from"""

    @pytest.fixture(autouse=True)
    def setup_env(self):
        _check_environment()
        self._projects = []
        yield
        for pid in self._projects:
            _cleanup_project(pid)

    def _create_project(self, name_prefix="S17", agents=None) -> str:
        pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
        config = {"agents": agents or ["zhangfei-dev", "simayi-challenger"]}
        resp = http_requests.post(f"{API_BASE}/api/projects", json={
            "id": pid, "name": f"{name_prefix}-{pid}", "config": config,
        }, timeout=10)
        assert resp.status_code == 200
        self._projects.append(pid)
        return pid

    def _create_task(self, pid, **kwargs) -> str:
        tid = kwargs.get("id") or f"e2e-task-{uuid.uuid4().hex[:8]}"
        body = {"id": tid, "status": "pending", "priority": 5, **kwargs}
        resp = http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks", json=body, timeout=10,
        )
        assert resp.status_code == 200
        return tid

    def test_s171_pause_resume_resumed_from(self):
        """S17-1: working → paused → 恢复 working，验证 resumed_from 字段"""
        pid = self._create_project("S17-1")
        tid = self._create_task(
            pid,
            title="E2E暂停恢复测试",
            description="测试暂停恢复功能",
            assignee="zhangfei-dev",
        )

        # 手动推到 claimed → working
        r1 = http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "claimed", "agent": "zhangfei-dev"}, timeout=10,
        )
        assert r1.json().get("ok")
        r2 = http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "working", "agent": "zhangfei-dev"}, timeout=10,
        )
        assert r2.json().get("ok")

        # 暂停
        r3 = http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "paused", "agent": "test"}, timeout=10,
        )
        assert r3.json().get("ok")

        # 验证 resumed_from == "working"
        task_resp = http_requests.get(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10,
        )
        task = task_resp.json()
        resumed_from = task.get("resumed_from")
        print(f"\n🚀 S17-1: 暂停后 resumed_from={resumed_from}")
        assert resumed_from == "working", (
            f"resumed_from 应为 'working'，实际: {resumed_from}"
        )
        assert task.get("status") == "paused"

        # 恢复到 working
        r4 = http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "working", "agent": "zhangfei-dev"}, timeout=10,
        )
        assert r4.json().get("ok")

        task2_resp = http_requests.get(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10,
        )
        task2 = task2_resp.json()
        print(f"   恢复后 status={task2.get('status')}")
        assert task2.get("status") == "working"

        print(f"   ✅ 暂停恢复流程正确")

    def test_s172_review_pause_resume(self):
        """S17-2: review → paused → 恢复 review"""
        pid = self._create_project("S17-2")
        tid = self._create_task(
            pid,
            title="E2E Review暂停恢复",
            assignee="simayi-challenger",
        )

        for s in ["claimed", "working", "review"]:
            http_requests.post(
                f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
                json={"status": s, "agent": "simayi-challenger"}, timeout=10,
            )

        # 暂停
        r = http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "paused", "agent": "test"}, timeout=10,
        )
        assert r.json().get("ok")

        task_resp = http_requests.get(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10,
        )
        task = task_resp.json()
        assert task.get("resumed_from") == "review"

        # 恢复到 review
        r2 = http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "review", "agent": "simayi-challenger"}, timeout=10,
        )
        assert r2.json().get("ok")

        task2_resp = http_requests.get(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10,
        )
        task2 = task2_resp.json()
        assert task2.get("status") == "review"

        print(f"\n   ✅ Review暂停恢复流程正确 (resumed_from=review)")


# ===================================================================
# S18: cancelled → 重新启动
# ===================================================================

@skip_no_integration
class TestS18CancelledRestart:
    """S18: cancelled → pending（重新启动）→ Agent 执行 → done"""

    @pytest.fixture(autouse=True)
    def setup_env(self):
        _check_environment()
        self._projects = []
        yield
        for pid in self._projects:
            _cleanup_project(pid)

    def _create_project(self, name_prefix="S18", agents=None) -> str:
        pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
        config = {"agents": agents or ["zhangfei-dev", "simayi-challenger"]}
        resp = http_requests.post(f"{API_BASE}/api/projects", json={
            "id": pid, "name": f"{name_prefix}-{pid}", "config": config,
        }, timeout=10)
        assert resp.status_code == 200
        self._projects.append(pid)
        return pid

    def _create_task(self, pid, **kwargs) -> str:
        tid = kwargs.get("id") or f"e2e-task-{uuid.uuid4().hex[:8]}"
        body = {"id": tid, "status": "pending", "priority": 5, **kwargs}
        resp = http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks", json=body, timeout=10,
        )
        assert resp.status_code == 200
        return tid

    def test_s181_cancelled_to_pending_restart(self):
        """S18-1: cancelled → pending → 等待调度执行"""
        pid = self._create_project("S18-1")
        tid = self._create_task(
            pid,
            title="E2E取消重启任务：echo restart",
            description=(
                "请执行 echo restart 并标记done。"
                "这是E2E测试，不需要做其他事。"
            ),
            assignee="zhangfei-dev",
        )

        # 手动推到 cancelled
        r1 = http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "cancelled", "agent": "test"}, timeout=10,
        )
        assert r1.json().get("ok")
        task_resp = http_requests.get(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10,
        )
        assert task_resp.json().get("status") == "cancelled"

        # 重新启动 → pending
        r2 = http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "pending", "agent": "test"}, timeout=10,
        )
        assert r2.json().get("ok")
        task2_resp = http_requests.get(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10,
        )
        task2 = task2_resp.json()
        assert task2.get("status") == "pending"
        assert task2.get("assignee") is None or task2.get("assignee") == ""

        print(f"\n🚀 S18-1: 等待重新调度执行 (pid={pid}, tid={tid})")
        result = _poll_task(
            pid, tid, timeout=MAX_WAIT_AGENT,
            terminal_states=("done", "failed", "cancelled", "blocked"),
        )
        status = result.get("status")
        print(f"   重启后最终状态: {status}")

        assert status != "pending", (
            f"重启后未被调度！{MAX_WAIT_AGENT}s后仍为pending"
        )

        if status == "done":
            print(f"   ✅ 取消重启流程正确")
        else:
            print(f"   ⚠️ 重启后状态: {status}")


# ===================================================================
# S19: Retry Chain（E97 claimed timeout + E10c retry chain 合并）
# ===================================================================

@skip_no_integration
class TestS19RetryChain:
    """S19: claimed 超时回收 + failed → pending 重试链"""

    @pytest.fixture(autouse=True)
    def setup_env(self):
        _check_environment()
        self._projects = []
        yield
        for pid in self._projects:
            _cleanup_project(pid)

    def _create_project(self, name_prefix="S19", agents=None) -> str:
        pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
        config = {"agents": agents or ["zhangfei-dev", "simayi-challenger"]}
        resp = http_requests.post(f"{API_BASE}/api/projects", json={
            "id": pid, "name": f"{name_prefix}-{pid}", "config": config,
        }, timeout=10)
        assert resp.status_code == 200
        self._projects.append(pid)
        return pid

    def _create_task(self, pid, **kwargs) -> str:
        tid = kwargs.get("id") or f"e2e-task-{uuid.uuid4().hex[:8]}"
        body = {"id": tid, "status": "pending", "priority": 5, **kwargs}
        resp = http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks", json=body, timeout=10,
        )
        assert resp.status_code == 200
        return tid

    def test_s191_claimed_timeout_to_pending(self):
        """S19-1: claimed 任务超时 → ticker 重置为 pending → assignee 清空"""
        pid = self._create_project("S19-1")
        tid = self._create_task(
            pid,
            title="E2E超时测试任务",
            description="测试claimed超时处理",
            assignee="zhangfei-dev",
        )

        # 手动推到 claimed
        r1 = http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "claimed", "agent": "zhangfei-dev"}, timeout=10,
        )
        assert r1.json().get("ok")

        # 直接操作 DB：把 claimed_at 设为 2 小时前
        two_hours_ago = (datetime.utcnow() - timedelta(hours=2)).isoformat()
        _patch_db_claimed_at(pid, tid, two_hours_ago)
        print(f"\n🚀 S19-1: 已设claimed_at为2小时前，等待ticker处理 (pid={pid}, tid={tid})")

        result = _poll_task(
            pid, tid, timeout=MAX_WAIT_DISPATCH,
            terminal_states=("pending", "escalated"),
        )
        status = result.get("status")
        print(f"   超时后状态: {status}")

        assert status != "claimed", (
            f"超时处理未生效！任务 {tid} 在 {MAX_WAIT_DISPATCH}s 后仍为 claimed"
        )

        assignee = result.get("assignee")
        print(f"   assignee: {assignee}")
        assert assignee is None or assignee == "", (
            f"超时重置后assignee应清空，实际: {assignee}"
        )

        print(f"   ✅ claimed超时处理正确 (status={status}, assignee cleared)")

    def test_s192_failed_to_pending_retry(self):
        """S19-2: 手动模拟失败 → 重试 → 等待调度完成"""
        pid = self._create_project("S19-2")
        tid = self._create_task(
            pid,
            title="E2E重试任务：echo retry",
            description=(
                "请执行 echo retry 并标记done。"
                "这是E2E测试，不需要做其他事。"
            ),
            assignee="zhangfei-dev",
        )

        # 手动推到 failed（模拟 Agent 执行失败）
        http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "claimed", "agent": "zhangfei-dev"}, timeout=10,
        )
        http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "working", "agent": "zhangfei-dev"}, timeout=10,
        )
        r_fail = http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "failed", "agent": "zhangfei-dev"}, timeout=10,
        )
        assert r_fail.json().get("ok")

        task_resp = http_requests.get(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10,
        )
        assert task_resp.json().get("status") == "failed"
        print(f"\n🚀 S19-2: 任务已标记failed，准备重试")

        # 手动重试 → pending
        r_retry = http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
            json={"status": "pending", "agent": "test"}, timeout=10,
        )
        assert r_retry.json().get("ok")

        task2_resp = http_requests.get(
            f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10,
        )
        task2 = task2_resp.json()
        assert task2.get("status") == "pending"
        assert task2.get("assignee") is None or task2.get("assignee") == ""
        retry_count = task2.get("retry_count", 0) or 0
        print(f"   retry_count: {retry_count}")

        result = _poll_task(
            pid, tid, timeout=MAX_WAIT_AGENT,
            terminal_states=("done", "failed", "cancelled", "blocked"),
        )
        status = result.get("status")
        print(f"   重试后最终状态: {status}")

        assert status != "pending", (
            f"重试后未被调度！{MAX_WAIT_AGENT}s后仍为pending"
        )

        if status == "done":
            print(f"   ✅ 失败重试链正确")
        else:
            print(f"   ⚠️ 重试后状态: {status}")


# ===================================================================
# S20: 完整生命周期（广播认领版）
# ===================================================================

@skip_no_integration
class TestS20FullLifecycle:
    """S20: 无 assignee → 广播认领 → claimed → working → review → done"""

    @pytest.fixture(autouse=True)
    def setup_env(self):
        _check_environment()
        self._projects = []
        yield
        for pid in self._projects:
            _cleanup_project(pid)

    def _create_project(self, name_prefix="S20", agents=None) -> str:
        pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
        config = {"agents": agents or ["zhangfei-dev", "simayi-challenger"]}
        resp = http_requests.post(f"{API_BASE}/api/projects", json={
            "id": pid, "name": f"{name_prefix}-{pid}", "config": config,
        }, timeout=10)
        assert resp.status_code == 200
        self._projects.append(pid)
        return pid

    def _create_task(self, pid, **kwargs) -> str:
        tid = kwargs.get("id") or f"e2e-task-{uuid.uuid4().hex[:8]}"
        body = {"id": tid, "status": "pending", "priority": 5, **kwargs}
        resp = http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks", json=body, timeout=10,
        )
        assert resp.status_code == 200
        return tid

    def test_s201_full_lifecycle_with_review(self):
        """S20-1: 完整生命周期：创建 → 广播 → 认领 → 执行 → review → done"""
        pid = self._create_project("S20-1")

        # 第一步：编码任务（广播认领）
        code_tid = self._create_task(
            pid,
            title="E2E完整链路：编码任务",
            description=(
                "请执行 echo lifecycle 并标记done。"
                "这是E2E完整生命周期测试，不需要做其他事。"
            ),
            task_type="coding",
        )

        print(f"\n🚀 S20-1: 等待编码任务广播认领 (pid={pid}, tid={code_tid})")
        result = _poll_task(
            pid, code_tid, timeout=MAX_WAIT_AGENT,
            terminal_states=("done", "failed", "cancelled", "blocked"),
        )
        code_status = result.get("status")
        print(f"   编码任务最终状态: {code_status}")

        assert code_status != "pending", "编码任务未被认领"

        if code_status == "done":
            events_resp = http_requests.get(
                f"{API_BASE}/api/projects/{pid}/tasks/{code_tid}/events",
                timeout=10,
            )
            if events_resp.status_code == 200:
                events = events_resp.json()
                event_types = [e.get("event_type") for e in events.get("events", [])]
                print(f"   Events: {event_types}")
                assert any("claimed" in str(e) or "started" in str(e)
                            for e in event_types), (
                    f"缺少状态变化事件: {event_types}"
                )

        # 第二步：review 任务（手动推）
        review_tid = self._create_task(
            pid,
            title="E2E完整链路：review任务",
            description="测试review状态",
            assignee="simayi-challenger",
        )

        for s in ["claimed", "working", "review", "done"]:
            r = http_requests.post(
                f"{API_BASE}/api/projects/{pid}/tasks/{review_tid}/status",
                json={"status": s, "agent": "simayi-challenger"}, timeout=10,
            )
            assert r.json().get("ok")

        task_resp = http_requests.get(
            f"{API_BASE}/api/projects/{pid}/tasks/{review_tid}", timeout=10,
        )
        assert task_resp.json().get("status") == "done"
        print(f"   Review任务手动生命周期: ✅")

        # 第三步：done → cancelled
        r_cancel = http_requests.post(
            f"{API_BASE}/api/projects/{pid}/tasks/{review_tid}/status",
            json={"status": "cancelled", "agent": "test"}, timeout=10,
        )
        assert r_cancel.json().get("ok")
        task3_resp = http_requests.get(
            f"{API_BASE}/api/projects/{pid}/tasks/{review_tid}", timeout=10,
        )
        assert task3_resp.json().get("status") == "cancelled"
        print(f"   done→cancelled: ✅")

        print(f"   ✅ S20-1 完整生命周期测试通过")


# ===================================================================
# S21: 缓存头验证
# ===================================================================

@skip_no_integration
class TestS21CacheHeaders:
    """S21: 验证 CachedStaticFiles 缓存头"""

    @pytest.fixture(autouse=True)
    def setup_env(self):
        _check_environment()

    def test_s211_html_no_cache(self):
        """S21-1: HTML 页面应为 no-cache"""
        resp = http_requests.get(f"{API_BASE}/", timeout=10)
        if resp.status_code != 200:
            pytest.skip(f"Frontend not served at {API_BASE}/: {resp.status_code}")

        cache_control = resp.headers.get("cache-control", "")
        print(f"\n🚀 S21-1: HTML Cache-Control: {cache_control}")
        assert "no-cache" in cache_control or "no-store" in cache_control, (
            f"HTML 应为 no-cache/no-store，实际: {cache_control}"
        )

    def test_s212_js_immutable(self):
        """S21-2: JS 文件应为 immutable + 长缓存"""
        html_resp = http_requests.get(f"{API_BASE}/", timeout=10)
        if html_resp.status_code != 200:
            pytest.skip("Frontend not available")

        js_matches = re.findall(r'src="(/assets/[^"]+\.js)"', html_resp.text)
        if not js_matches:
            pytest.skip("No JS files found in HTML")

        js_path = js_matches[0]
        resp = http_requests.get(f"{API_BASE}{js_path}", timeout=10)
        cache_control = resp.headers.get("cache-control", "")
        print(f"   S21-2: JS ({js_path}) Cache-Control: {cache_control}")
        assert "immutable" in cache_control, (
            f"JS 应含 immutable，实际: {cache_control}"
        )
        assert "31536000" in cache_control, (
            f"JS max-age 应为 31536000，实际: {cache_control}"
        )

    def test_s213_css_immutable(self):
        """S21-3: CSS 文件应为 immutable + 长缓存"""
        html_resp = http_requests.get(f"{API_BASE}/", timeout=10)
        if html_resp.status_code != 200:
            pytest.skip("Frontend not available")

        css_matches = re.findall(r'href="(/assets/[^"]+\.css)"', html_resp.text)
        if not css_matches:
            pytest.skip("No CSS files found in HTML")

        css_path = css_matches[0]
        resp = http_requests.get(f"{API_BASE}{css_path}", timeout=10)
        cache_control = resp.headers.get("cache-control", "")
        print(f"   S21-3: CSS ({css_path}) Cache-Control: {cache_control}")
        assert "immutable" in cache_control, (
            f"CSS 应含 immutable，实际: {cache_control}"
        )

        print(f"   ✅ 缓存头验证通过")
