"""黑板数据模型""" 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 # v2.6.1 路由扩展字段(司马懿 BUG-1:不改 assignee 语义) current_agent: Optional[str] = None # 当前阶段执行者(随状态流转更新) previous_agent: Optional[str] = None # 前一阶段执行者(审计追溯) next_capability: Optional[str] = None # Agent 声明的下一步需要的能力(Mode B) @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)