""" Operation failure model for tracking non-command failures. Tracks failures from API calls, file operations, network requests, and other operations (distinct from command execution failures tracked in command_runs). """ from datetime import datetime from typing import TYPE_CHECKING, Optional from sqlalchemy import Boolean, CHAR, CheckConstraint, ForeignKey, Index, String, Text, TIMESTAMP from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.sql import func from .base import Base, UUIDMixin if TYPE_CHECKING: from .session import Session from .work_item import WorkItem class OperationFailure(Base, UUIDMixin): """ Operation failure model for non-command failures. Tracks failures from API calls, file operations, network requests, database queries, and external integrations. Used for troubleshooting, pattern detection, and system reliability monitoring. Distinct from CommandRun failures which track shell command execution. This tracks programmatic operations and API interactions. Attributes: session_id: Foreign key to sessions table work_item_id: Foreign key to work_items table operation_type: Type of operation that failed operation_description: Detailed description of what was attempted target_system: Host, URL, or service name that was targeted error_message: Error message from the failure error_code: HTTP status, exit code, or error number failure_category: Category of failure (timeout, authentication, etc.) stack_trace: Stack trace if available resolution_applied: Description of how the failure was resolved resolved: Whether the failure has been resolved resolved_at: When the failure was resolved request_data: JSON data of what was attempted response_data: JSON data of error response environment_snapshot: JSON snapshot of relevant environment variables/versions created_at: When the failure occurred session: Relationship to Session model work_item: Relationship to WorkItem model """ __tablename__ = "operation_failures" # Foreign keys session_id: Mapped[Optional[str]] = mapped_column( CHAR(36), ForeignKey("sessions.id", ondelete="CASCADE"), doc="Foreign key to sessions table" ) work_item_id: Mapped[Optional[str]] = mapped_column( CHAR(36), ForeignKey("work_items.id", ondelete="CASCADE"), doc="Foreign key to work_items table" ) # Operation details operation_type: Mapped[str] = mapped_column( String(100), nullable=False, doc="Type of operation: api_call, file_operation, network_request, database_query, external_integration, service_restart" ) operation_description: Mapped[str] = mapped_column( Text, nullable=False, doc="Detailed description of what was attempted" ) target_system: Mapped[Optional[str]] = mapped_column( String(255), doc="Host, URL, or service name that was targeted" ) # Failure details error_message: Mapped[str] = mapped_column( Text, nullable=False, doc="Error message from the failure" ) error_code: Mapped[Optional[str]] = mapped_column( String(50), doc="HTTP status code, exit code, or error number" ) failure_category: Mapped[Optional[str]] = mapped_column( String(100), doc="Category of failure: timeout, authentication, not_found, permission_denied, etc." ) stack_trace: Mapped[Optional[str]] = mapped_column( Text, doc="Stack trace if available" ) # Resolution tracking resolution_applied: Mapped[Optional[str]] = mapped_column( Text, doc="Description of how the failure was resolved" ) resolved: Mapped[bool] = mapped_column( Boolean, default=False, server_default="0", nullable=False, doc="Whether the failure has been resolved" ) resolved_at: Mapped[Optional[datetime]] = mapped_column( TIMESTAMP, doc="When the failure was resolved" ) # Context data (JSON stored as text) request_data: Mapped[Optional[str]] = mapped_column( Text, doc="JSON data of what was attempted" ) response_data: Mapped[Optional[str]] = mapped_column( Text, doc="JSON data of error response" ) environment_snapshot: Mapped[Optional[str]] = mapped_column( Text, doc="JSON snapshot of relevant environment variables, versions, etc." ) # Timestamp created_at: Mapped[datetime] = mapped_column( TIMESTAMP, nullable=False, server_default=func.now(), doc="When the failure occurred" ) # Relationships session: Mapped[Optional["Session"]] = relationship( "Session", back_populates="operation_failures", doc="Relationship to Session model" ) work_item: Mapped[Optional["WorkItem"]] = relationship( "WorkItem", doc="Relationship to WorkItem model" ) # Constraints and indexes __table_args__ = ( CheckConstraint( "operation_type IN ('api_call', 'file_operation', 'network_request', 'database_query', 'external_integration', 'service_restart')", name="ck_operation_failures_type" ), Index("idx_op_failure_session", "session_id"), Index("idx_op_failure_type", "operation_type"), Index("idx_op_failure_category", "failure_category"), Index("idx_op_failure_resolved", "resolved"), ) def __repr__(self) -> str: """String representation of the operation failure.""" return f""