"""黑板写操作（CRUD）"""

from __future__ import annotations

import json
import logging
import sqlite3
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional

from .db import (
    VALID_TRANSITIONS,
    COMMENT_TYPES,
    EVENT_TYPES,
    OUTPUT_TYPES,
    REVIEW_TYPES,
    VERDICT_TYPES,
    EXPERIENCE_SOURCES,
    EXPERIENCE_CATEGORIES,
    EXPERIENCE_STATUSES,
    ATTEMPT_OUTCOMES,
    TERMINAL_STATUSES,
    get_connection,
    init_db,
)
from .models import (
    Task, Comment, Output, Decision, Observation, Event,
    Review, Experience,
)

logger = logging.getLogger("moziplus-v2.blackboard")


class Blackboard:
    """黑板操作封装（per-project）"""

    def __init__(self, db_path: Path):
        self.db_path = db_path
        init_db(db_path)

    def _conn(self) -> sqlite3.Connection:
        return get_connection(self.db_path)

    # ===================================================================
    # Task CRUD
    # ===================================================================

    def create_task(self, task: Task) -> Task:
        """创建任务"""
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            conn.execute(
                """INSERT INTO tasks
                (id, title, description, status, assignee, assigned_by,
                 depends_on, parent_task, priority, task_type, deadline,
                 retry_count, max_retries, must_haves, risk_level,
                 estimated_duration_minutes, escalated, stage, stages_json,
                 archived, archived_at)
                VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                (task.id, task.title, task.description, task.status,
                 task.assignee, task.assigned_by,
                 task.depends_on, task.parent_task, task.priority,
                 task.task_type, task.deadline,
                 task.retry_count, task.max_retries, task.must_haves,
                 task.risk_level, task.estimated_duration_minutes,
                 1 if task.escalated else 0,
                 task.stage, task.stages_json,
                 1 if task.archived else 0, task.archived_at),
            )
            conn.execute(
                "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)",
                (task.id, task.assigned_by, "task_created",
                 json.dumps({"title": task.title, "task_type": task.task_type})),
            )
            conn.commit()
            return task
        finally:
            conn.close()

    def get_task(self, task_id: str) -> Optional[Task]:
        """获取单个任务"""
        conn = self._conn()
        try:
            row = conn.execute(
                "SELECT * FROM tasks WHERE id=?", (task_id,)).fetchone()
            return Task.from_row(row) if row else None
        finally:
            conn.close()

    def update_task_status(self, task_id: str, new_status: str,
                           agent: Optional[str] = None,
                           detail: Optional[Dict] = None) -> bool:
        """更新任务状态（校验合法转换）"""
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            row = conn.execute(
                "SELECT status FROM tasks WHERE id=?", (task_id,)
            ).fetchone()
            if not row:
                return False

            old_status = row["status"]
            allowed = VALID_TRANSITIONS.get(old_status, set())

            # 终态不可转换；同状态不变
            if old_status in TERMINAL_STATUSES:
                return False
            if old_status == new_status:
                return True
            if new_status not in allowed:
                logger.warning("Invalid transition %s → %s for task %s",
                               old_status, new_status, task_id)
                return False

            now = datetime.utcnow().isoformat()
            updates = {"status": new_status, "updated_at": now}
            if new_status == "claimed":
                updates["claimed_at"] = now
                if agent:
                    updates["assignee"] = agent
            elif new_status == "working":
                updates["started_at"] = now
            elif new_status in ("done", "failed", "cancelled"):
                updates["completed_at"] = now
            elif new_status == "paused":
                updates["completed_at"] = now  # paused 也记录时间用于恢复
                updates["resumed_from"] = old_status  # 记录暂停前状态
            elif new_status == "pending":
                # 所有 →pending 转换都清空 assignee（与 ticker._transition_status L414
                # 对齐）
                updates["assignee"] = None
                updates["claimed_at"] = None
                updates["current_agent"] = None
                if old_status == "failed":
                    # 仅 failed→pending 递增 retry_count
                    updates["retry_count"] = (
                        conn.execute("SELECT retry_count FROM tasks WHERE id=?",
                                     (task_id,)).fetchone()["retry_count"] + 1
                    )

            set_clause = ", ".join(f"{k}=?" for k in updates)
            conn.execute(
                f"UPDATE tasks SET {set_clause} WHERE id=?",
                (*updates.values(), task_id),
            )

            event_type = f"task_{new_status}"
            if event_type not in EVENT_TYPES:
                event_type = "daemon_tick"
            conn.execute(
                "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)",
                (task_id, agent, event_type,
                 json.dumps({"from": old_status, "to": new_status, **(detail or {})})),
            )
            conn.commit()
            return True
        finally:
            conn.close()

    def claim_task(self, task_id: str, agent_id: str) -> bool:
        """原子 CAS 认领"""
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            cursor = conn.execute(
                "UPDATE tasks SET status='claimed', assignee=?, claimed_at=datetime('now'), "
                "updated_at=datetime('now') "
                "WHERE id=? AND status='pending' AND (assignee IS NULL OR assignee=?)",
                (agent_id, task_id, agent_id),
            )
            if cursor.rowcount > 0:
                conn.execute(
                    "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)",
                    (task_id, agent_id, "task_claimed",
                     json.dumps({"by": agent_id})),
                )
                conn.commit()
                return True
            conn.commit()
            return False
        finally:
            conn.close()

    def list_tasks(self, status: Optional[str] = None,
                   assignee: Optional[str] = None,
                   assigned_by: Optional[str] = None,
                   parent_task: Optional[str] = None) -> List[Task]:
        """列出任务"""
        conn = self._conn()
        try:
            query = "SELECT * FROM tasks"
            conditions = []
            params = []
            if status:
                conditions.append("status=?")
                params.append(status)
            if assignee:
                conditions.append("assignee=?")
                params.append(assignee)
            if assigned_by:
                conditions.append("assigned_by=?")
                params.append(assigned_by)
            if parent_task is not None:
                conditions.append("parent_task=?")
                params.append(parent_task)
            if conditions:
                query += " WHERE " + " AND ".join(conditions)
            query += " ORDER BY priority ASC, created_at ASC"
            rows = conn.execute(query, params).fetchall()
            return [Task.from_row(r) for r in rows]
        finally:
            conn.close()

    def update_must_haves(self, task_id: str, must_haves: str) -> bool:
        """更新 Task 的 must_haves 字段（用于 Mail 元数据等）"""
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            conn.execute(
                "UPDATE tasks SET must_haves=?, updated_at=datetime('now') WHERE id=?",
                (must_haves, task_id),
            )
            conn.commit()
            return True
        finally:
            conn.close()

    # ===================================================================
    # Comment
    # ===================================================================

    def add_comment(self, task_id: str, author: str, body: str,
                    comment_type: str = "general",
                    mentions: Optional[List[str]] = None) -> int:
        """添加评论"""
        if comment_type not in COMMENT_TYPES:
            raise ValueError(f"Invalid comment_type: {comment_type}")
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            cursor = conn.execute(
                "INSERT INTO comments (task_id, author, comment_type, body, mentions) "
                "VALUES (?,?,?,?,?)",
                (task_id, author, comment_type, body,
                 json.dumps(mentions or [])),
            )
            conn.execute(
                "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)",
                (task_id, author, "comment_added",
                 json.dumps({"comment_type": comment_type,
                             "body_preview": body[:100],
                             "mentions": mentions})),
            )
            conn.commit()
            return cursor.lastrowid
        finally:
            conn.close()

    def get_comments(self, task_id: str,
                     comment_type: Optional[str] = None) -> List[Comment]:
        """获取任务评论"""
        conn = self._conn()
        try:
            if comment_type:
                rows = conn.execute(
                    "SELECT * FROM comments WHERE task_id=? AND comment_type=? "
                    "ORDER BY created_at ASC",
                    (task_id, comment_type),
                ).fetchall()
            else:
                rows = conn.execute(
                    "SELECT * FROM comments WHERE task_id=? ORDER BY created_at ASC",
                    (task_id,),
                ).fetchall()
            return [Comment.from_row(r) for r in rows]
        finally:
            conn.close()

    # ===================================================================
    # Output
    # ===================================================================

    def write_output(self, task_id: str, agent: str, output_type: str,
                     title: str, content_path: Optional[str] = None,
                     summary: Optional[str] = None,
                     metadata: Optional[Dict] = None,
                     attempt_number: int = 1) -> int:
        """写入产出"""
        if output_type not in OUTPUT_TYPES:
            raise ValueError(f"Invalid output_type: {output_type}")
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            cursor = conn.execute(
                "INSERT INTO outputs "
                "(task_id, agent, output_type, title, content_path, summary, metadata, attempt_number) "
                "VALUES (?,?,?,?,?,?,?,?)",
                (task_id, agent, output_type, title, content_path, summary,
                 json.dumps(metadata or {}), attempt_number),
            )
            conn.execute(
                "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)",
                (task_id, agent, "output_written",
                 json.dumps({"output_id": title, "type": output_type})),
            )
            conn.commit()
            return cursor.lastrowid
        finally:
            conn.close()

    def get_outputs(self, task_id: str) -> List[Output]:
        """获取任务产出"""
        conn = self._conn()
        try:
            rows = conn.execute(
                "SELECT * FROM outputs WHERE task_id=? ORDER BY created_at ASC",
                (task_id,),
            ).fetchall()
            return [Output.from_row(r) for r in rows]
        finally:
            conn.close()

    # ===================================================================
    # Decision
    # ===================================================================

    def add_decision(self, task_id: str, decider: str, decision: str,
                     rationale: str,
                     alternatives: Optional[List[str]] = None) -> int:
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            cursor = conn.execute(
                "INSERT INTO decisions (task_id, decider, decision, rationale, alternatives) "
                "VALUES (?,?,?,?,?)",
                (task_id, decider, decision, rationale,
                 json.dumps(alternatives or [])),
            )
            conn.execute(
                "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)",
                (task_id, decider, "decision_recorded",
                 json.dumps({"decision": decision})),
            )
            conn.commit()
            return cursor.lastrowid
        finally:
            conn.close()

    def get_decisions(self, task_id: str) -> List[Decision]:
        conn = self._conn()
        try:
            rows = conn.execute(
                "SELECT * FROM decisions WHERE task_id=? ORDER BY created_at ASC",
                (task_id,),
            ).fetchall()
            return [Decision.from_row(r) for r in rows]
        finally:
            conn.close()

    # ===================================================================
    # Observation
    # ===================================================================

    def add_observation(self, task_id: Optional[str], observer: str, body: str,
                        severity: str = "info") -> int:
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            cursor = conn.execute(
                "INSERT INTO observations (task_id, observer, severity, body) VALUES (?,?,?,?)",
                (task_id, observer, severity, body),
            )
            conn.execute(
                "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)",
                (task_id, observer, "observation_added",
                 json.dumps({"severity": severity})),
            )
            conn.commit()
            return cursor.lastrowid
        finally:
            conn.close()

    def get_observations(self, task_id: str,
                         unresolved_only: bool = False) -> List[Observation]:
        conn = self._conn()
        try:
            query = "SELECT * FROM observations WHERE task_id=?"
            params: list = [task_id]
            if unresolved_only:
                query += " AND resolved_by IS NULL"
            query += " ORDER BY created_at ASC"
            rows = conn.execute(query, params).fetchall()
            return [Observation.from_row(r) for r in rows]
        finally:
            conn.close()

    # ===================================================================
    # Review
    # ===================================================================

    def add_review(self, review: Review) -> str:
        if review.review_type not in REVIEW_TYPES:
            raise ValueError(f"Invalid review_type: {review.review_type}")
        if review.verdict not in VERDICT_TYPES:
            raise ValueError(f"Invalid verdict: {review.verdict}")
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            conn.execute(
                """INSERT INTO reviews
                (id, task_id, output_id, reviewer, review_type, verdict,
                 confidence, round, max_rounds, consensus_reached,
                 summary, detail_path)
                VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""",
                (review.id, review.task_id, review.output_id, review.reviewer,
                 review.review_type, review.verdict,
                 review.confidence, review.round, review.max_rounds,
                 1 if review.consensus_reached else 0,
                 review.summary, review.detail_path),
            )
            conn.execute(
                "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)",
                (review.task_id, review.reviewer, "task_reviewed",
                 json.dumps({"verdict": review.verdict, "confidence": review.confidence})),
            )
            conn.commit()
            return review.id
        finally:
            conn.close()

    def get_reviews(self, task_id: str) -> List[Review]:
        conn = self._conn()
        try:
            rows = conn.execute(
                "SELECT * FROM reviews WHERE task_id=? ORDER BY created_at ASC",
                (task_id,),
            ).fetchall()
            return [Review.from_row(r) for r in rows]
        finally:
            conn.close()

    # ===================================================================
    # Experience
    # ===================================================================

    def add_experience(self, exp: Experience) -> str:
        if exp.source not in EXPERIENCE_SOURCES:
            raise ValueError(f"Invalid source: {exp.source}")
        if exp.category not in EXPERIENCE_CATEGORIES:
            raise ValueError(f"Invalid category: {exp.category}")
        if exp.status not in EXPERIENCE_STATUSES:
            raise ValueError(f"Invalid status: {exp.status}")
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            conn.execute(
                """INSERT INTO experiences
                (experience_id, source, task_id, summary, category, confidence,
                 status, skill_id, usage_count, last_used_at, created_at, created_by,
                 updated_at, deprecated_reason)
                VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                (exp.experience_id, exp.source, exp.task_id, exp.summary,
                 exp.category, exp.confidence, exp.status, exp.skill_id,
                 exp.usage_count, exp.last_used_at,
                 exp.created_at or datetime.utcnow().isoformat(),
                 exp.created_by, exp.updated_at, exp.deprecated_reason),
            )
            for tag in exp.tags:
                conn.execute(
                    "INSERT OR IGNORE INTO experience_tags (experience_id, tag) VALUES (?,?)",
                    (exp.experience_id, tag),
                )
            conn.commit()
            return exp.experience_id
        finally:
            conn.close()

    def query_experiences(self, tags: Optional[List[str]] = None,
                          status: str = "active",
                          limit: int = 10) -> List[Experience]:
        conn = self._conn()
        try:
            if tags:
                placeholders = ",".join("?" * len(tags))
                rows = conn.execute(
                    f"""SELECT DISTINCT e.* FROM experiences e
                    JOIN experience_tags et ON e.experience_id = et.experience_id
                    WHERE et.tag IN ({placeholders}) AND e.status=?
                    ORDER BY e.usage_count DESC, e.created_at DESC
                    LIMIT ?""",
                    (*tags, status, limit),
                ).fetchall()
            else:
                rows = conn.execute(
                    "SELECT * FROM experiences WHERE status=? "
                    "ORDER BY usage_count DESC, created_at DESC LIMIT ?",
                    (status, limit),
                ).fetchall()

            result = []
            for row in rows:
                exp = Experience.from_row(row)
                tag_rows = conn.execute(
                    "SELECT tag FROM experience_tags WHERE experience_id=?",
                    (exp.experience_id,),
                ).fetchall()
                exp.tags = [r["tag"] for r in tag_rows]
                result.append(exp)
            return result
        finally:
            conn.close()

    def touch_experience(self, experience_id: str) -> None:
        """增加引用计数"""
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            conn.execute(
                "UPDATE experiences SET usage_count=usage_count+1, "
                "last_used_at=datetime('now') WHERE experience_id=?",
                (experience_id,),
            )
            conn.commit()
        finally:
            conn.close()

    # ===================================================================
    # Event
    # ===================================================================

    def add_event(self, event_type: str, task_id: Optional[str] = None,
                  agent: Optional[str] = None,
                  detail: Optional[Dict] = None) -> int:
        conn = self._conn()
        try:
            cursor = conn.execute(
                "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)",
                (task_id, agent, event_type, json.dumps(detail or {})),
            )
            conn.commit()
            return cursor.lastrowid
        finally:
            conn.close()

    def get_events(self, task_id: Optional[str] = None,
                   limit: int = 50) -> List[Event]:
        conn = self._conn()
        try:
            if task_id:
                rows = conn.execute(
                    "SELECT * FROM events WHERE task_id=? ORDER BY created_at DESC LIMIT ?",
                    (task_id, limit),
                ).fetchall()
            else:
                rows = conn.execute(
                    "SELECT * FROM events ORDER BY created_at DESC LIMIT ?",
                    (limit,),
                ).fetchall()
            return [Event.from_row(r) for r in rows]
        finally:
            conn.close()

    # ===================================================================
    # Task Attempt
    # ===================================================================

    def add_task_attempt(self, task_id: str, attempt_number: int,
                         agent: str, outcome: str,
                         metadata: Optional[Dict] = None) -> int:
        if outcome not in ATTEMPT_OUTCOMES:
            raise ValueError(f"Invalid outcome: {outcome}")
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            cursor = conn.execute(
                "INSERT INTO task_attempts "
                "(task_id, attempt_number, agent, outcome, metadata) VALUES (?,?,?,?,?)",
                (task_id, attempt_number, agent, outcome,
                 json.dumps(metadata or {})),
            )
            conn.commit()
            return cursor.lastrowid
        finally:
            conn.close()

    # ===================================================================
    # Agent
    # ===================================================================

    def upsert_agent(self, agent_id: str, role: Optional[str] = None,
                     capabilities: Optional[List[str]] = None) -> None:
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            conn.execute(
                "INSERT OR REPLACE INTO agents (agent_id, role, capabilities, last_active) "
                "VALUES (?,?,?,datetime('now'))",
                (agent_id, role, json.dumps(capabilities or [])),
            )
            conn.commit()
        finally:
            conn.close()

    def get_agent(self, agent_id: str) -> Optional[Dict]:
        conn = self._conn()
        try:
            row = conn.execute(
                "SELECT * FROM agents WHERE agent_id=?", (agent_id,)
            ).fetchone()
            return dict(row) if row else None
        finally:
            conn.close()

    def update_agent_status(self, agent_id: str, status: str,
                            current_task: Optional[str] = None) -> None:
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            conn.execute(
                "UPDATE agents SET current_status=?, current_task=?, "
                "last_active=datetime('now') WHERE agent_id=?",
                (status, current_task, agent_id),
            )
            conn.commit()
        finally:
            conn.close()

    # ===================================================================
    # Archive (v2.8)
    # ===================================================================

    def archive_task(self, task_id: str) -> bool:
        """归档单个任务"""
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            conn.execute(
                "UPDATE tasks SET archived=1, archived_at=datetime('now'), "
                "updated_at=datetime('now') WHERE id=?",
                (task_id,),
            )
            conn.execute(
                "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)",
                (task_id, "daemon", "task_archived",
                 json.dumps({"action": "archive"})),
            )
            conn.commit()
            return True
        finally:
            conn.close()

    def unarchive_task(self, task_id: str) -> bool:
        """取消归档"""
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            conn.execute(
                "UPDATE tasks SET archived=0, archived_at=NULL, "
                "updated_at=datetime('now') WHERE id=?",
                (task_id,),
            )
            conn.execute(
                "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)",
                (task_id, "daemon", "task_unarchived",
                 json.dumps({"action": "unarchive"})),
            )
            conn.commit()
            return True
        finally:
            conn.close()

    def archive_done_tasks(self) -> int:
        """一键归档所有 done 状态的任务"""
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            cursor = conn.execute(
                "UPDATE tasks SET archived=1, archived_at=datetime('now'), "
                "updated_at=datetime('now') "
                "WHERE status='done' AND archived=0"
            )
            count = cursor.rowcount
            if count > 0:
                conn.execute(
                    "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)",
                    (None, "daemon", "task_archived",
                     json.dumps({"action": "archive_done_batch", "count": count})),
                )
            conn.commit()
            return count
        finally:
            conn.close()

    # ── Checkpoint CRUD（M3） ──

    def create_checkpoint(
        self,
        task_id: str,
        cp_type: str,
        title: str,
        payload: dict,
        description: str | None = None,
        checkpoint_id: str | None = None,
    ) -> dict:
        """创建 Checkpoint"""
        import uuid
        # BUG-33: 校验 payload 结构必须含 version 字段
        if not isinstance(payload, dict) or "version" not in payload:
            raise ValueError(
                "payload must be a dict containing 'version' field")
        cp_id = checkpoint_id or f"cp-{uuid.uuid4().hex[:8]}"
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            conn.execute(
                "INSERT INTO checkpoints (id, task_id, type, title, description, payload) "
                "VALUES (?,?,?,?,?,?)",
                (cp_id, task_id, cp_type, title, description, json.dumps(payload)),
            )
            conn.execute(
                "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)",
                (task_id, "daemon", "checkpoint_created",
                 json.dumps({"checkpoint_id": cp_id, "type": cp_type, "title": title})),
            )
            conn.commit()
            return {"id": cp_id, "task_id": task_id, "type": cp_type,
                    "title": title, "status": "pending"}
        finally:
            conn.close()

    def list_checkpoints(self, task_id: str) -> list[dict]:
        """列出 task 的所有 checkpoint"""
        conn = self._conn()
        try:
            rows = conn.execute(
                "SELECT * FROM checkpoints WHERE task_id=? ORDER BY created_at",
                (task_id,),
            ).fetchall()
            return [dict(r) for r in rows]
        finally:
            conn.close()

    def get_checkpoint(self, checkpoint_id: str) -> dict | None:
        """获取单个 checkpoint"""
        conn = self._conn()
        try:
            row = conn.execute(
                "SELECT * FROM checkpoints WHERE id=?", (checkpoint_id,)
            ).fetchone()
            return dict(row) if row else None
        finally:
            conn.close()

    def resolve_checkpoint(
        self,
        checkpoint_id: str,
        action: str,
        resolved_by: str = "user",
        note: str | None = None,
    ) -> dict | None:
        """通过/驳回 checkpoint，并自动推进 task 状态

        approve: verify → done, decision/action → working
        reject: 一律 → working
        """
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            row = conn.execute(
                "SELECT * FROM checkpoints WHERE id=?", (checkpoint_id,)
            ).fetchone()
            if not row:
                return None
            cp = dict(row)
            if cp["status"] != "pending":
                return {"error": f"Checkpoint already {cp['status']}"}

            # 更新 checkpoint 状态
            new_status = "approved" if action == "approve" else "rejected"
            conn.execute(
                "UPDATE checkpoints SET status=?, resolved_at=datetime('now'), "
                "resolved_by=?, resolve_note=? WHERE id=?",
                (new_status, resolved_by, note, checkpoint_id),
            )

            # 推进 task 状态
            task_id = cp["task_id"]
            cp_type = cp["type"]
            task_row = conn.execute(
                "SELECT status FROM tasks WHERE id=?", (task_id,)
            ).fetchone()
            if not task_row:
                conn.commit()
                return {"error": "Task not found"}

            task_status = task_row["status"]
            # BUG-32: 安全校验 — task 必须处于 waiting_human 才能推进
            if task_status != "waiting_human":
                conn.commit()
                return {"error": f"Task is {task_status}, expected waiting_human"}

            if action == "approve":
                if cp_type == "verify":
                    new_task_status = "done"
                else:
                    new_task_status = "working"
            else:
                new_task_status = "working"

            conn.execute(
                "UPDATE tasks SET status=?, updated_at=datetime('now') WHERE id=?",
                (new_task_status, task_id),
            )
            conn.execute(
                "INSERT INTO events (task_id, agent, event_type, detail) VALUES (?,?,?,?)",
                (task_id, resolved_by, f"checkpoint_{new_status}",
                 json.dumps({
                     "checkpoint_id": checkpoint_id,
                     "action": action,
                     "task_status": new_task_status,
                     "note": note,
                 })),
            )
            conn.commit()
            return {
                "checkpoint_id": checkpoint_id,
                "checkpoint_status": new_status,
                "task_id": task_id,
                "task_status": new_task_status,
            }
        finally:
            conn.close()

    # ===================================================================
    # 四相循环 (v2.9 #01)
    # ===================================================================

    def get_subtasks_summary(self, parent_id: str) -> Optional[Dict[str, Any]]:
        """获取 parent task 下所有 sub task 的状态摘要

        Returns:
            {"total": N, "done": N, "failed": N, "cancelled": N,
             "other": N, "all_terminal": bool}
            如果 parent 不存在或没有 sub task，返回 None
        """
        conn = self._conn()
        try:
            # 确认 parent 存在
            parent = conn.execute(
                "SELECT id, status, round_count FROM tasks WHERE id=?",
                (parent_id,)
            ).fetchone()
            if not parent:
                return None

            rows = conn.execute(
                "SELECT status, COUNT(*) as cnt FROM tasks WHERE parent_task=? GROUP BY status",
                (parent_id,)
            ).fetchall()

            if not rows:
                return None

            summary = {"parent_id": parent_id, "parent_status": parent["status"],
                       "round_count": parent["round_count"],
                       "total": 0, "done": 0, "failed": 0, "cancelled": 0, "other": 0}
            for row in rows:
                cnt = row["cnt"]
                summary["total"] += cnt
                if row["status"] in ("done", "failed", "cancelled"):
                    summary[row["status"]] += cnt
                else:
                    summary["other"] += cnt

            summary["all_terminal"] = summary["other"] == 0 and summary["total"] > 0
            return summary
        finally:
            conn.close()

    def get_aggregate_outputs(self, parent_id: str) -> List[Dict[str, Any]]:
        """聚合 parent task 下所有 sub task 的 outputs"""
        conn = self._conn()
        try:
            sub_ids = conn.execute(
                "SELECT id FROM tasks WHERE parent_task=?",
                (parent_id,)
            ).fetchall()
            if not sub_ids:
                return []

            ids = [r["id"] for r in sub_ids]
            placeholders = ",".join("?" * len(ids))
            rows = conn.execute(
                f"SELECT o.*, t.title as task_title FROM outputs o "
                f"JOIN tasks t ON o.task_id = t.id "
                f"WHERE o.task_id IN ({placeholders}) "
                f"ORDER BY o.created_at ASC",
                ids
            ).fetchall()
            return [dict(r) for r in rows]
        finally:
            conn.close()

    def get_round_comments(self, parent_id: str) -> List[Dict[str, Any]]:
        """获取 parent task 及其 sub task 的所有 comments"""
        conn = self._conn()
        try:
            # parent 自身的 comments + 所有 sub 的 comments
            rows = conn.execute(
                """SELECT c.*, t.title as task_title FROM comments c
                   JOIN tasks t ON c.task_id = t.id
                   WHERE c.task_id = ? OR c.task_id IN (SELECT id FROM tasks WHERE parent_task = ?)
                   ORDER BY c.created_at ASC""",
                (parent_id, parent_id)
            ).fetchall()
            return [dict(r) for r in rows]
        finally:
            conn.close()

    def increment_round_count(self, parent_id: str) -> int:
        """递增 parent task 的 round_count，返回新值"""
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            conn.execute(
                "UPDATE tasks SET round_count = round_count + 1, updated_at=datetime('now') WHERE id=?",
                (parent_id,)
            )
            row = conn.execute(
                "SELECT round_count FROM tasks WHERE id=?", (parent_id,)
            ).fetchone()
            conn.commit()
            return row["round_count"] if row else 0
        finally:
            conn.close()

    # ===================================================================
    # Mention Queue (v2.9 #01)
    # ===================================================================

    def record_mentions(self, comment_id: int, task_id: str,
                        mentioned_agents: List[str]) -> int:
        """写入 mention_queue 记录（comment 创建时调用）"""
        if not mentioned_agents:
            return 0
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            count = 0
            for agent_id in mentioned_agents:
                exists = conn.execute(
                    "SELECT id FROM mention_queue WHERE comment_id=? AND mentioned_agent=?",
                    (comment_id, agent_id)
                ).fetchone()
                if not exists:
                    conn.execute(
                        "INSERT INTO mention_queue (comment_id, task_id, mentioned_agent, status) "
                        "VALUES (?,?,?,'pending')",
                        (comment_id, task_id, agent_id)
                    )
                    count += 1
            conn.commit()
            return count
        finally:
            conn.close()

    def get_pending_mentions(
            self, max_retries: int = 5) -> List[Dict[str, Any]]:
        """获取所有 pending 且未超过重试上限的 mentions"""
        conn = self._conn()
        try:
            rows = conn.execute(
                """SELECT mq.*, c.body as comment_body, c.author as comment_author
                   FROM mention_queue mq
                   JOIN comments c ON mq.comment_id = c.id
                   WHERE mq.status = 'pending' AND mq.retry_count < ?
                   ORDER BY mq.created_at ASC""",
                (max_retries,)
            ).fetchall()
            return [dict(r) for r in rows]
        finally:
            conn.close()

    def mark_mention_notified(self, mention_id: int) -> bool:
        """标记 mention 为已通知"""
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            conn.execute(
                "UPDATE mention_queue SET status='notified', notified_at=datetime('now') WHERE id=?",
                (mention_id,)
            )
            conn.commit()
            return True
        finally:
            conn.close()

    def mark_mention_retry(self, mention_id: int) -> bool:
        """递增 mention 重试计数"""
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            conn.execute(
                "UPDATE mention_queue SET retry_count=retry_count+1 WHERE id=?", (mention_id,))
            conn.commit()
            return True
        finally:
            conn.close()

    def mark_mention_failed(self, mention_id: int) -> bool:
        """标记 mention 为失败（超过重试上限）"""
        conn = self._conn()
        try:
            conn.execute("BEGIN IMMEDIATE")
            conn.execute(
                "UPDATE mention_queue SET status='failed' WHERE id=?", (mention_id,))
            conn.commit()
            return True
        finally:
            conn.close()
