diagnostics.py

# -*- coding: utf-8 -*-
"""TaskDiagnostics：故障識別、分類、寫上下文、廣播。

對應架構教學文件 §5.4 / §10.5、interfaces.md §7 故障碼總表。
"""
from __future__ import annotations

import json
import logging
import threading
import time
from dataclasses import dataclass
from typing import Any, Callable, Dict, List, Optional

from .context import ContextStore, TaskContext
from .events import FaultRecord, SkillEvent


logger = logging.getLogger(__name__)


# 任務級故障碼表（interfaces.md §7.1）
FAULT_TABLE: Dict[str, Dict[str, str]] = {
    "TM-001": {"name": "PLAN_FAIL", "severity": "error", "default_action": "abort"},
    "TM-002": {"name": "SCENE_EMPTY", "severity": "info", "default_action": "none"},
    "TM-003": {"name": "GRASP_TIMEOUT", "severity": "warn", "default_action": "retry"},
    "TM-004": {"name": "REORIENT_FAIL", "severity": "warn", "default_action": "replan"},
    "TM-005": {
        "name": "PLACE_OUT_OF_ZONE",
        "severity": "warn",
        "default_action": "retry",
    },
    "TM-006": {"name": "SKILL_FAULT", "severity": "error", "default_action": "replan"},
    "TM-007": {
        "name": "SOFTWARE_FAULT",
        "severity": "critical",
        "default_action": "pause",
    },
    "TM-008": {"name": "MODE_BLOCKED", "severity": "info", "default_action": "none"},
}


@dataclass
class DiagnoseDecision:
    code: str
    action: str  # retry / replan / pause / abort / none
    reason: str = ""


class TaskDiagnostics:
    """負責把任意故障事件 → FaultRecord → 寫 Redis + 廣播 + 回給 Scheduler。

    Scheduler 在註冊時傳入回呼（`scheduler_action_callback`），
    這樣 Diagnostics 不需要直接 import Scheduler，避免循環依賴。
    """

    def __init__(
        self,
        collaborator: Any,
        context_store: ContextStore,
        scheduler_action_callback: Optional[Callable[[DiagnoseDecision], None]] = None,
        fault_channel: str = "task_faults",
    ) -> None:
        self.collaborator = collaborator
        self.context_store = context_store
        self.fault_channel = fault_channel
        self._scheduler_cb = scheduler_action_callback
        self._lock = threading.Lock()

    # --- 對外 API ---------------------------------------------------------
    def raise_fault(
        self,
        task_id: str,
        code: str,
        details: Optional[Dict[str, Any]] = None,
        subtask_id: Optional[str] = None,
        package_id: Optional[str] = None,
        action_override: Optional[str] = None,
    ) -> FaultRecord:
        info = FAULT_TABLE.get(code) or {
            "name": "UNKNOWN",
            "severity": "error",
            "default_action": "pause",
        }
        action = action_override or info["default_action"]
        record = FaultRecord(
            code=code,
            name=info["name"],
            severity=info["severity"],
            task_id=task_id,
            subtask_id=subtask_id,
            package_id=package_id,
            details=details or {},
            action_taken=action,
        )
        self._publish_and_persist(record)
        self._dispatch_action(record)
        return record

    def from_skill_event(self, evt: SkillEvent) -> Optional[FaultRecord]:
        """技能事件 → 任務級故障（自動翻譯 SK-* 為 TM-006）。"""
        if evt.status == "ok":
            return None
        details = {
            "skill_name": evt.skill_name,
            "phase": evt.phase,
            "status": evt.status,
            "upstream_code": evt.fault_code,
        }
        details.update(evt.details or {})
        return self.raise_fault(
            task_id=evt.task_id,
            code="TM-006",
            details=details,
            subtask_id=evt.subtask_id,
        )

    def soft_error(self, task_id: str, exc: BaseException) -> FaultRecord:
        return self.raise_fault(
            task_id=task_id,
            code="TM-007",
            details={"exception": repr(exc)},
        )

    def list_faults(self, task_id: str) -> List[Dict[str, Any]]:
        ctx = self.context_store.get(task_id)
        return list(ctx.faults) if ctx else []

    def clear_fault(self, task_id: str, code: str) -> int:
        with self._lock:
            ctx = self.context_store.get(task_id)
            if not ctx:
                return 0
            before = len(ctx.faults)
            ctx.faults = [f for f in ctx.faults if f.get("code") != code]
            self.context_store.save(ctx)
            return before - len(ctx.faults)

    # --- 內部 -------------------------------------------------------------
    def _publish_and_persist(self, record: FaultRecord) -> None:
        # 1) 廣播
        try:
            self.collaborator.send(self.fault_channel, json.dumps(record.to_dict()))
        except Exception as e:  # noqa: BLE001
            logger.warning("[Diagnostics] publish failed: %s", e)
        # 2) 寫 context.faults
        try:
            ctx: Optional[TaskContext] = self.context_store.get(record.task_id)
            if ctx:
                self.context_store.append_fault(ctx, record)
            else:
                logger.info(
                    "[Diagnostics] context for task_id=%s not found, skip persist",
                    record.task_id,
                )
        except Exception as e:  # noqa: BLE001
            logger.warning("[Diagnostics] persist failed: %s", e)
        # 3) 統計
        try:
            self.context_store.bump_stat(f"fault_{record.code}", 1)
        except Exception:  # noqa: BLE001
            pass
        logger.warning(
            "[Diagnostics] fault raised: %s (%s) action=%s details=%s",
            record.code,
            record.name,
            record.action_taken,
            record.details,
        )

    def _dispatch_action(self, record: FaultRecord) -> None:
        if not self._scheduler_cb:
            return
        decision = DiagnoseDecision(
            code=record.code,
            action=record.action_taken,
            reason=f"{record.code}:{record.name}",
        )
        try:
            self._scheduler_cb(decision)
        except Exception as e:  # noqa: BLE001
            logger.exception("[Diagnostics] scheduler callback error: %s", e)


__all__ = ["TaskDiagnostics", "DiagnoseDecision", "FAULT_TABLE"]