auto-sync: 2026-05-18 11:37:25

src/blackboard/queries.py

@@ -6,7 +6,7 @@ import json
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
-from .db import get_connection
+from .db import get_connection, MANUAL_STATUSES
 from .models import Task
 
 
@@ -73,36 +73,25 @@ class Queries:
         finally:
             conn.close()
 
-    def tasks_by_status(self, status: str,
-                         card_id: Optional[str] = None) -> List[Task]:
+    def tasks_by_status(self, status: str) -> List[Task]:
         """查询指定状态的所有任务"""
         conn = self._conn()
         try:
-            if card_id:
-                rows = conn.execute(
-                    "SELECT * FROM tasks WHERE status=? AND card_id=? ORDER BY priority ASC",
-                    (status, card_id),
-                ).fetchall()
-            else:
-                rows = conn.execute(
-                    "SELECT * FROM tasks WHERE status=? ORDER BY priority ASC",
-                    (status,),
-                ).fetchall()
+            rows = conn.execute(
+                "SELECT * FROM tasks WHERE status=? ORDER BY priority ASC",
+                (status,),
+            ).fetchall()
             return [Task.from_row(r) for r in rows]
         finally:
             conn.close()
 
-    def pending_dispatchable(self, card_id: Optional[str] = None) -> List[Task]:
+    def pending_dispatchable(self) -> List[Task]:
         """查询可调度的 pending 任务（依赖已满足）"""
         conn = self._conn()
         try:
-            query = "SELECT * FROM tasks WHERE status='pending'"
-            params: list = []
-            if card_id:
-                query += " AND card_id=?"
-                params.append(card_id)
-            query += " ORDER BY priority ASC"
-            rows = conn.execute(query, params).fetchall()
+            rows = conn.execute(
+                "SELECT * FROM tasks WHERE status='pending' ORDER BY priority ASC"
+            ).fetchall()
             result = []
             for r in rows:
                 deps = json.loads(r["depends_on"] or "[]")
@@ -193,5 +182,161 @@ class Queries:
         finally:
             conn.close()
 
+    # ===================================================================
+    # v2.7 父子关系查询
+    # ===================================================================
+
+    def list_subtasks(self, parent_task_id: str) -> List[Task]:
+        """列出某个父 Task 的所有子 Task"""
+        conn = self._conn()
+        try:
+            rows = conn.execute(
+                "SELECT * FROM tasks WHERE parent_task=? ORDER BY priority ASC, created_at ASC",
+                (parent_task_id,),
+            ).fetchall()
+            return [Task.from_row(r) for r in rows]
+        finally:
+            conn.close()
+
+    def top_level_tasks(self) -> List[Task]:
+        """列出所有顶层 Task（parent_task IS NULL）"""
+        conn = self._conn()
+        try:
+            rows = conn.execute(
+                "SELECT * FROM tasks WHERE parent_task IS NULL ORDER BY priority ASC, created_at ASC"
+            ).fetchall()
+            return [Task.from_row(r) for r in rows]
+        finally:
+            conn.close()
+
+    def compute_parent_status(self, parent_task_id: str) -> Optional[str]:
+        """从子 Task 聚合推导父 Task 状态
+
+        优先级：review > working > pending > blocked > failed
+        手动状态（cancelled）不参与聚合
+        """
+        conn = self._conn()
+        try:
+            # 检查父 Task 是否有手动状态
+            parent_row = conn.execute(
+                "SELECT status FROM tasks WHERE id=?", (parent_task_id,)
+            ).fetchone()
+            if not parent_row:
+                return None
+            if parent_row["status"] in MANUAL_STATUSES:
+                return parent_row["status"]
+
+            # 聚合子 Task 状态（排除 cancelled）
+            rows = conn.execute(
+                "SELECT status, COUNT(*) as cnt FROM tasks "
+                "WHERE parent_task=? AND status != 'cancelled' "
+                "GROUP BY status",
+                (parent_task_id,),
+            ).fetchall()
+
+            if not rows:
+                # 无子 Task，保持原状态
+                return parent_row["status"]
+
+            status_counts = {r["status"]: r["cnt"] for r in rows}
+            total = sum(status_counts.values())
+            done_count = status_counts.get("done", 0)
+
+            # 所有 done → done
+            if done_count == total:
+                return "done"
+
+            # 有 review → review
+            if status_counts.get("review", 0) > 0:
+                return "review"
+
+            # 有 working/claimed → working
+            if status_counts.get("working", 0) > 0 or status_counts.get("claimed", 0) > 0:
+                return "working"
+
+            # 有 pending → pending
+            if status_counts.get("pending", 0) > 0:
+                return "pending"
+
+            # 有 blocked → blocked
+            if status_counts.get("blocked", 0) > 0:
+                return "blocked"
+
+            # 有 failed → failed
+            if status_counts.get("failed", 0) > 0:
+                return "failed"
+
+            return parent_row["status"]
+        finally:
+            conn.close()
+
+    def parent_task_progress(self, parent_task_id: str) -> Dict[str, Any]:
+        """父 Task 的 Stage 进度信息"""
+        conn = self._conn()
+        try:
+            parent_row = conn.execute(
+                "SELECT * FROM tasks WHERE id=?", (parent_task_id,)
+            ).fetchone()
+            if not parent_row:
+                return {}
+
+            parent = dict(parent_row)
+            stages = json.loads(parent.get("stages_json") or "[]")
+
+            # 子 Task 统计
+            total_row = conn.execute(
+                "SELECT COUNT(*) as cnt FROM tasks WHERE parent_task=? AND status != 'cancelled'",
+                (parent_task_id,),
+            ).fetchone()
+            done_row = conn.execute(
+                "SELECT COUNT(*) as cnt FROM tasks WHERE parent_task=? AND status='done'",
+                (parent_task_id,),
+            ).fetchone()
+            total = total_row["cnt"] if total_row else 0
+            done = done_row["cnt"] if done_row else 0
+
+            # 按 stage 分组
+            stage_progress = []
+            for stage in stages:
+                stage_id = stage.get("id", "")
+                s_total = conn.execute(
+                    "SELECT COUNT(*) as cnt FROM tasks WHERE parent_task=? AND stage=? AND status != 'cancelled'",
+                    (parent_task_id, stage_id),
+                ).fetchone()["cnt"]
+                s_done = conn.execute(
+                    "SELECT COUNT(*) as cnt FROM tasks WHERE parent_task=? AND stage=? AND status='done'",
+                    (parent_task_id, stage_id),
+                ).fetchone()["cnt"]
+                s_active = conn.execute(
+                    "SELECT COUNT(*) as cnt FROM tasks WHERE parent_task=? AND stage=? AND status IN ('working','review','claimed')",
+                    (parent_task_id, stage_id),
+                ).fetchone()["cnt"]
+                stage_progress.append({
+                    "id": stage_id,
+                    "label": stage.get("label", stage_id),
+                    "order": stage.get("order", 0),
+                    "total": s_total,
+                    "done": s_done,
+                    "active": s_active,
+                })
+
+            # 当前活跃 stage
+            active_stage = None
+            for sp in stage_progress:
+                if sp["active"] > 0 or (sp["total"] > 0 and sp["done"] < sp["total"]):
+                    if not active_stage and sp["done"] < sp["total"]:
+                        active_stage = sp["label"]
+
+            return {
+                "task_id": parent_task_id,
+                "title": parent.get("title", ""),
+                "total_subtasks": total,
+                "done_subtasks": done,
+                "active_stage": active_stage,
+                "stages": stage_progress,
+            }
+        finally:
+            conn.close()
+
     def db_size_bytes(self) -> int:
         return self.db_path.stat().st_size if self.db_path.exists() else 0