sanguo_moziplus_v2/src/blackboard/models.py

"""黑板数据模型"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, 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
    # v2.6.1 路由扩展字段
    current_agent: Optional[str] = None
    previous_agent: Optional[str] = None
    next_capability: Optional[str] = None
    # v2.7 SubTask 层
    stage: Optional[str] = None              # 所属 Stage 标签
    stages_json: str = "[]"                   # 动态 Stage 定义（仅顶层 Task）
    # v3.1 暂停恢复
    resumed_from: Optional[str] = None       # 暂停前状态，恢复时回到原状态
    # v2.9 四相循环
    round_count: int = 0                     # 庞统 review 轮次计数
    # v2.8 归档
    archived: bool = False
    archived_at: Optional[str] = None

    @classmethod
    def from_row(cls, row: Any) -> Task:
        valid_keys = {f.name for f in cls.__dataclass_fields__.values()}
        return cls(**{k: row[k] for k in row.keys() if k in valid_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:
        valid_keys = {f.name for f in cls.__dataclass_fields__.values()}
        return cls(**{k: row[k] for k in row.keys() if k in valid_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:
        valid_keys = {f.name for f in cls.__dataclass_fields__.values()}
        return cls(**{k: row[k] for k in row.keys() if k in valid_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:
        valid_keys = {f.name for f in cls.__dataclass_fields__.values()}
        return cls(**{k: row[k] for k in row.keys() if k in valid_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:
        valid_keys = {f.name for f in cls.__dataclass_fields__.values()}
        return cls(**{k: row[k] for k in row.keys() if k in valid_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:
        valid_keys = {f.name for f in cls.__dataclass_fields__.values()}
        return cls(**{k: row[k] for k in row.keys() if k in valid_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:
        valid_keys = {f.name for f in cls.__dataclass_fields__.values()}
        d = {k: row[k] for k in row.keys() if k in valid_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:
        valid_keys = {f.name for f in cls.__dataclass_fields__.values()}
        d = {k: row[k] for k in row.keys() if k in valid_keys}
        d.pop("tags", None)  # tags queried separately
        return cls(**d)