Incident System Architecture¶

class IncidentRecorderProtocol(Protocol):
    def record_incident(
        self,
        *,
        incident_type: IncidentType,
        severity: IncidentSeverity,
        message: str,
        interface_id: str | None = None,
        context: dict[str, Any] | None = None,
        journal: PingPongJournal | None = None,
    ) -> Coroutine[Any, Any, IncidentSnapshot]: ...

@dataclass(frozen=True, slots=True)
class IncidentSnapshot:
    id: str                              # UUID
    timestamp: datetime                  # When incident occurred
    incident_type: IncidentType          # Type classification
    severity: IncidentSeverity           # INFO, WARNING, ERROR
    message: str                         # Human-readable description
    interface_id: str | None             # Which interface (if applicable)
    context: dict[str, Any]              # Type-specific diagnostic data
    journal: list[JournalEntry] | None   # Event history (for PingPong)

{
    "pending_count": int,        # Current pending PONG count
    "threshold": int,            # Configured threshold
    # Plus: journal with last N PING/PONG events and timestamps
}

{
    "unknown_count": int,        # Current unknown PONG count
    "threshold": int,            # Configured threshold
    # Plus: journal with last N PING/PONG events and timestamps
}

{
    "old_state": str,            # State before trip (closed/half_open)
    "failure_count": int,        # Consecutive failures that caused trip
    "failure_threshold": int,    # Configured threshold
    "recovery_timeout": float,   # Seconds before half-open attempt
    "last_failure_time": str,    # ISO timestamp of last failure
    "total_requests": int,       # Total requests since start
}

{
    "success_count": int,        # Successful requests in half-open
    "success_threshold": int,    # Configured threshold for recovery
}

{
    "reason": str,                    # Reason for connection loss
    "detected_at": str,               # ISO timestamp when loss was detected
    "client_state": str | None,       # Client state at time of loss
    "circuit_breaker_state": str | None,  # Circuit breaker state
    "recovery_attempt_count": int,    # Number of recovery attempts so far
    "active_recoveries": list[str],   # List of interfaces currently recovering
    "in_failed_state": bool,          # Whether coordinator is in FAILED state
}

{
    "total_attempts": int,            # Total recovery attempts made
    "total_duration_ms": float,       # Total recovery duration in milliseconds
    "stages_completed": list[str],    # Recovery stages completed (e.g., ["TCP_CHECKING", "RPC_CHECKING", ...])
    "client_state": str | None,       # Client state after restoration
    "circuit_breaker_state": str | None,  # Circuit breaker state after restoration
    "was_in_failed_state": bool,      # Whether coordinator was in FAILED state before recovery
}

{
    "protocol": str,                  # Protocol type ("xml-rpc" or "json-rpc")
    "method": str,                    # RPC method that failed (e.g., "setValue", "getParamset")
    "error_type": str,                # Error type (e.g., "SSLError", "OSError", "XMLRPCFault", "JSONRPCError")
    "error_message": str,             # Sanitized error message (sensitive info removed)
    "tls_enabled": bool,              # Whether TLS is enabled for this connection
}

{
    "seconds_since_last_event": float,    # Time since last callback was received
    "callback_warn_interval": float,      # Configured threshold in seconds
    "last_event_time": str,               # ISO timestamp of last callback received
    "client_state": str,                  # Client state when timeout detected
    "circuit_breaker_state": str | None,  # Circuit breaker state (if available)
}

INCIDENT_STORE_MAX_PER_TYPE: Final = 20  # Max incidents per type
DEFAULT_MAX_AGE_DAYS: Final = 7          # 7-day retention

def _record_incident_sync(self, ...) -> None:
    if (incident_recorder := self._incident_recorder) is None:
        return

    # Capture current state for async closure
    captured_data = {...}

    async def _record() -> None:
        try:
            await incident_recorder.record_incident(...)
        except Exception as err:
            _LOGGER.debug("Failed to record incident: %s", err)

    # Fire and forget - suppress if no event loop
    with contextlib.suppress(RuntimeError):
        asyncio.get_running_loop().create_task(_record())

class CircuitBreaker:
    def __init__(
        self,
        *,
        incident_recorder: IncidentRecorderProtocol | None = None,
        ...
    ) -> None:
        self._incident_recorder = incident_recorder

async def get_diagnostics(self) -> dict[str, Any]:
    """Return diagnostic data for Home Assistant."""
    return {
        "incidents": [i.to_dict() for i in await self._load_incidents()],
        "total_count": len(self._incidents),
        "retention_days": self._retention_days,
    }

diag["incident_store"] = await central.cache_coordinator.incident_store.get_diagnostics()

# aiohomematic/store/types.py
class IncidentType(StrEnum):
    MY_NEW_INCIDENT = "MY_NEW_INCIDENT"
    """Description of when this incident occurs."""

def _record_my_incident(self) -> None:
    if (incident_recorder := self._incident_recorder) is None:
        return

    from aiohomematic.store.types import IncidentSeverity, IncidentType

    # Capture all relevant state
    context = {
        "relevant_field_1": self._field_1,
        "relevant_field_2": self._field_2,
        "configuration": self._config.some_setting,
        # Include anything that helps diagnose the issue
    }

    async def _record() -> None:
        try:
            await incident_recorder.record_incident(
                incident_type=IncidentType.MY_NEW_INCIDENT,
                severity=IncidentSeverity.ERROR,
                message=f"Descriptive message about what happened",
                interface_id=self._interface_id,
                context=context,
            )
        except Exception as err:
            _LOGGER.debug("Failed to record incident: %s", err)

    with contextlib.suppress(RuntimeError):
        asyncio.get_running_loop().create_task(_record())

@pytest.mark.asyncio
async def test_my_incident_recorded(self) -> None:
    from unittest.mock import AsyncMock, MagicMock

    incident_recorder = MagicMock()
    incident_recorder.record_incident = AsyncMock()

    # Trigger the incident
    ...

    # Wait for async task
    await asyncio.sleep(0.01)

    # Verify
    incident_recorder.record_incident.assert_called_once()
    call_kwargs = incident_recorder.record_incident.call_args.kwargs
    assert call_kwargs["incident_type"] == IncidentType.MY_NEW_INCIDENT