""" Waiting period model for tracking think time and engagement. """ from datetime import datetime from typing import Optional from sqlalchemy import String, Text, Integer, DateTime, ForeignKey from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.sql import func from .base import Base, TimestampMixin class WaitingPeriod(Base, TimestampMixin): """ Represents a period when Claude is waiting for user input. These periods provide insight into user thinking time, engagement patterns, and workflow interruptions during development sessions. """ __tablename__ = "waiting_periods" # Primary key id: Mapped[int] = mapped_column(primary_key=True) # Foreign key to session session_id: Mapped[int] = mapped_column(ForeignKey("sessions.id"), nullable=False, index=True) # Timing start_time: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, default=func.now(), index=True ) end_time: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) duration_seconds: Mapped[Optional[int]] = mapped_column(Integer, nullable=True, index=True) # Context context_before: Mapped[Optional[str]] = mapped_column(Text, nullable=True) context_after: Mapped[Optional[str]] = mapped_column(Text, nullable=True) # Activity inference likely_activity: Mapped[Optional[str]] = mapped_column(String(50), nullable=True) # thinking, research, external_work, break # Relationships session: Mapped["Session"] = relationship("Session", back_populates="waiting_periods") def __repr__(self) -> str: duration_info = f", duration={self.duration_seconds}s" if self.duration_seconds else "" return f"" @property def is_active(self) -> bool: """Check if this waiting period is still active (not ended).""" return self.end_time is None @property def calculated_duration_seconds(self) -> Optional[int]: """Calculate waiting period duration in seconds.""" if self.end_time is None: # Still waiting, calculate current duration current_duration = datetime.utcnow() - self.start_time return int(current_duration.total_seconds()) else: # Finished waiting if self.duration_seconds is not None: return self.duration_seconds else: duration = self.end_time - self.start_time return int(duration.total_seconds()) @property def duration_minutes(self) -> Optional[float]: """Get duration in minutes.""" seconds = self.calculated_duration_seconds return seconds / 60 if seconds is not None else None def end_waiting(self, context_after: Optional[str] = None) -> None: """End the waiting period and calculate duration.""" if self.end_time is None: self.end_time = func.now() self.duration_seconds = self.calculated_duration_seconds if context_after: self.context_after = context_after def classify_activity(self) -> str: """ Classify the likely activity based on duration and context. Returns one of: 'thinking', 'research', 'external_work', 'break' """ if self.likely_activity: return self.likely_activity duration = self.calculated_duration_seconds if duration is None: return "unknown" # Classification based on duration if duration < 10: return "thinking" # Quick pause elif duration < 60: return "thinking" # Short contemplation elif duration < 300: # 5 minutes return "research" # Looking something up elif duration < 1800: # 30 minutes return "external_work" # Working on something else else: return "break" # Extended break @property def engagement_score(self) -> float: """ Calculate an engagement score based on waiting time. Returns a score from 0.0 (disengaged) to 1.0 (highly engaged). """ duration = self.calculated_duration_seconds if duration is None: return 0.5 # Default neutral score # Short waits indicate high engagement if duration <= 5: return 1.0 elif duration <= 30: return 0.9 elif duration <= 120: # 2 minutes return 0.7 elif duration <= 300: # 5 minutes return 0.5 elif duration <= 900: # 15 minutes return 0.3 else: return 0.1 # Long waits indicate low engagement def is_quick_response(self) -> bool: """Check if user responded quickly (< 30 seconds).""" duration = self.calculated_duration_seconds return duration is not None and duration < 30 def is_thoughtful_pause(self) -> bool: """Check if this was a thoughtful pause (30s - 2 minutes).""" duration = self.calculated_duration_seconds return duration is not None and 30 <= duration < 120 def is_research_break(self) -> bool: """Check if this was likely a research break (2 - 15 minutes).""" duration = self.calculated_duration_seconds return duration is not None and 120 <= duration < 900 def is_extended_break(self) -> bool: """Check if this was an extended break (> 15 minutes).""" duration = self.calculated_duration_seconds return duration is not None and duration >= 900