auto-sync: 2026-05-17 00:34:41

src/blackboard/models.py

@@ -0,0 +1,164 @@
+"""黑板数据模型"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+
+@dataclass
+class Task:
+    id: str
+    title: str
+    description: Optional[str] = None
+    status: str = "pending"
+    assignee: Optional[str] = None
+    assigned_by: Optional[str] = None
+    depends_on: Optional[str] = None  # JSON array string
+    parent_task: Optional[str] = None
+    priority: int = 5
+    task_type: Optional[str] = None
+    created_at: Optional[str] = None
+    updated_at: Optional[str] = None
+    claimed_at: Optional[str] = None
+    started_at: Optional[str] = None
+    completed_at: Optional[str] = None
+    deadline: Optional[str] = None
+    retry_count: int = 0
+    max_retries: int = 2
+    must_haves: Optional[str] = None  # JSON
+    risk_level: str = "standard"
+    estimated_duration_minutes: Optional[int] = None
+    escalated: bool = False
+
+    @classmethod
+    def from_row(cls, row: Any) -> Task:
+        return cls(**{k: row[k] for k in row.keys()})
+
+
+@dataclass
+class Comment:
+    id: Optional[int] = None
+    task_id: str = ""
+    author: str = ""
+    comment_type: str = "general"
+    body: str = ""
+    mentions: Optional[str] = None  # JSON array string
+    created_at: Optional[str] = None
+
+    @classmethod
+    def from_row(cls, row: Any) -> Comment:
+        return cls(**{k: row[k] for k in row.keys()})
+
+
+@dataclass
+class Output:
+    id: Optional[int] = None
+    task_id: str = ""
+    agent: str = ""
+    output_type: str = ""
+    title: str = ""
+    content_path: Optional[str] = None
+    summary: Optional[str] = None
+    metadata: Optional[str] = None  # JSON
+    attempt_number: int = 1
+    created_at: Optional[str] = None
+
+    @classmethod
+    def from_row(cls, row: Any) -> Output:
+        return cls(**{k: row[k] for k in row.keys()})
+
+
+@dataclass
+class Decision:
+    id: Optional[int] = None
+    task_id: str = ""
+    decider: str = ""
+    decision: str = ""
+    rationale: str = ""
+    alternatives: Optional[str] = None  # JSON
+    created_at: Optional[str] = None
+
+    @classmethod
+    def from_row(cls, row: Any) -> Decision:
+        return cls(**{k: row[k] for k in row.keys()})
+
+
+@dataclass
+class Observation:
+    id: Optional[int] = None
+    task_id: str = ""
+    observer: str = ""
+    severity: str = "info"
+    body: str = ""
+    resolved_by: Optional[str] = None
+    resolved_at: Optional[str] = None
+    created_at: Optional[str] = None
+
+    @classmethod
+    def from_row(cls, row: Any) -> Observation:
+        return cls(**{k: row[k] for k in row.keys()})
+
+
+@dataclass
+class Event:
+    id: Optional[int] = None
+    task_id: Optional[str] = None
+    agent: Optional[str] = None
+    event_type: str = ""
+    detail: Optional[str] = None  # JSON
+    created_at: Optional[str] = None
+
+    @classmethod
+    def from_row(cls, row: Any) -> Event:
+        return cls(**{k: row[k] for k in row.keys()})
+
+
+@dataclass
+class Review:
+    id: str = ""
+    task_id: str = ""
+    output_id: Optional[str] = None
+    reviewer: str = ""
+    review_type: str = ""
+    verdict: str = ""
+    confidence: Optional[float] = None
+    round: int = 1
+    max_rounds: int = 3
+    consensus_reached: bool = False
+    summary: str = ""
+    detail_path: Optional[str] = None
+    created_at: Optional[str] = None
+
+    @classmethod
+    def from_row(cls, row: Any) -> Review:
+        d = {k: row[k] for k in row.keys()}
+        # SQLite stores boolean as 0/1
+        if "consensus_reached" in d:
+            d["consensus_reached"] = bool(d["consensus_reached"])
+        return cls(**d)
+
+
+@dataclass
+class Experience:
+    experience_id: str = ""
+    source: str = ""
+    task_id: Optional[str] = None
+    summary: str = ""
+    category: str = ""
+    confidence: float = 0.8
+    status: str = "active"
+    skill_id: Optional[str] = None
+    usage_count: int = 0
+    last_used_at: Optional[str] = None
+    created_at: Optional[str] = None
+    created_by: str = ""
+    updated_at: Optional[str] = None
+    deprecated_reason: Optional[str] = None
+    tags: List[str] = field(default_factory=list)
+
+    @classmethod
+    def from_row(cls, row: Any) -> Experience:
+        d = {k: row[k] for k in row.keys()}
+        d.pop("tags", None)  # tags queried separately
+        return cls(**d)