feat: complete DDD Phase 2 - add domain services, infrastructure layers, and shared components

- domain/value_objects: Add Topic, UserTitle, GoldenQuote, Statistics value objects
- domain/services: Add StatisticsCalculator, ReportGenerator domain services
- domain/exceptions: Add comprehensive domain exception hierarchy
- infrastructure/persistence: Add HistoryRepository for data storage
- infrastructure/llm: Add LLMClient wrapper for AstrBot providers
- infrastructure/config: Add ConfigManager for centralized configuration
- infrastructure/resilience: Add CircuitBreaker, RateLimiter, retry utilities
- application: Add SchedulingService, ReportingService application services
- shared: Add constants and TraceContext for request tracing

All imports verified in Docker container.

src/application/__init__.py

@@ -1,5 +1,12 @@
 # Application Layer - Orchestration and Use Cases
 from .analysis_orchestrator import AnalysisOrchestrator
 from .message_converter import MessageConverter
+from .scheduling_service import SchedulingService
+from .reporting_service import ReportingService
 
-__all__ = ["AnalysisOrchestrator", "MessageConverter"]
+__all__ = [
+    "AnalysisOrchestrator",
+    "MessageConverter",
+    "SchedulingService",
+    "ReportingService",
+]

src/application/reporting_service.py

@@ -0,0 +1,263 @@
+"""
+Reporting Service - Application service for generating and sending reports
+
+This service coordinates report generation and delivery to groups.
+"""
+
+from datetime import datetime
+from typing import Any, Dict, List, Optional
+
+from astrbot.api import logger
+
+from ..domain.services import ReportGenerator
+from ..domain.value_objects.topic import Topic
+from ..domain.value_objects.user_title import UserTitle
+from ..domain.value_objects.golden_quote import GoldenQuote
+from ..domain.value_objects.statistics import GroupStatistics
+from ..infrastructure.config import ConfigManager
+from ..infrastructure.persistence import HistoryRepository
+
+
+class ReportingService:
+    """
+    Application service for generating and managing reports.
+
+    This service coordinates between domain services and infrastructure
+    to produce and deliver analysis reports.
+    """
+
+    def __init__(
+        self,
+        config: ConfigManager,
+        history_repository: HistoryRepository,
+    ):
+        """
+        Initialize the reporting service.
+
+        Args:
+            config: Configuration manager
+            history_repository: Repository for storing reports
+        """
+        self.config = config
+        self.history = history_repository
+
+    def generate_report(
+        self,
+        group_id: str,
+        group_name: str,
+        statistics: GroupStatistics,
+        topics: List[Topic],
+        user_titles: List[UserTitle],
+        golden_quotes: List[GoldenQuote],
+        date_str: Optional[str] = None,
+    ) -> str:
+        """
+        Generate a complete analysis report.
+
+        Args:
+            group_id: Group identifier
+            group_name: Group display name
+            statistics: Group statistics
+            topics: List of discussion topics
+            user_titles: List of user titles
+            golden_quotes: List of golden quotes
+            date_str: Report date (defaults to today)
+
+        Returns:
+            Formatted report string
+        """
+        date_str = date_str or datetime.now().strftime("%Y-%m-%d")
+
+        generator = ReportGenerator(
+            group_name=group_name,
+            date_str=date_str,
+        )
+
+        # Generate report based on configuration
+        report = generator.generate_full_report(
+            statistics=statistics,
+            topics=topics if self.config.get_include_topics() else [],
+            user_titles=user_titles if self.config.get_include_user_titles() else [],
+            golden_quotes=golden_quotes if self.config.get_include_golden_quotes() else [],
+            include_header=True,
+            include_footer=True,
+        )
+
+        return report
+
+    def generate_summary(
+        self,
+        group_id: str,
+        statistics: GroupStatistics,
+        top_topic: Optional[Topic] = None,
+        top_quote: Optional[GoldenQuote] = None,
+        date_str: Optional[str] = None,
+    ) -> str:
+        """
+        Generate a brief summary report.
+
+        Args:
+            group_id: Group identifier
+            statistics: Group statistics
+            top_topic: Most significant topic
+            top_quote: Best golden quote
+            date_str: Report date
+
+        Returns:
+            Brief summary string
+        """
+        date_str = date_str or datetime.now().strftime("%Y-%m-%d")
+
+        generator = ReportGenerator(date_str=date_str)
+        return generator.generate_summary_report(
+            statistics=statistics,
+            top_topic=top_topic,
+            top_quote=top_quote,
+        )
+
+    def save_report(
+        self,
+        group_id: str,
+        report_data: Dict[str, Any],
+        date_str: Optional[str] = None,
+    ) -> bool:
+        """
+        Save a report to history.
+
+        Args:
+            group_id: Group identifier
+            report_data: Report data dictionary
+            date_str: Report date
+
+        Returns:
+            True if saved successfully
+        """
+        date_str = date_str or datetime.now().strftime("%Y-%m-%d")
+
+        return self.history.save_analysis_result(
+            group_id=group_id,
+            result=report_data,
+            date_str=date_str,
+        )
+
+    def get_report(
+        self,
+        group_id: str,
+        date_str: str,
+    ) -> Optional[Dict[str, Any]]:
+        """
+        Get a saved report.
+
+        Args:
+            group_id: Group identifier
+            date_str: Report date
+
+        Returns:
+            Report data or None
+        """
+        return self.history.get_analysis_result(group_id, date_str)
+
+    def get_recent_reports(
+        self,
+        group_id: str,
+        limit: int = 7,
+    ) -> List[Dict[str, Any]]:
+        """
+        Get recent reports for a group.
+
+        Args:
+            group_id: Group identifier
+            limit: Maximum number of reports
+
+        Returns:
+            List of report data dictionaries
+        """
+        return self.history.get_recent_results(group_id, limit)
+
+    def has_report_for_today(self, group_id: str) -> bool:
+        """
+        Check if a report exists for today.
+
+        Args:
+            group_id: Group identifier
+
+        Returns:
+            True if report exists
+        """
+        today = datetime.now().strftime("%Y-%m-%d")
+        return self.history.has_analysis_for_date(group_id, today)
+
+    def format_for_platform(
+        self,
+        report: str,
+        platform: str,
+        format_type: Optional[str] = None,
+    ) -> str:
+        """
+        Format a report for a specific platform.
+
+        Args:
+            report: Raw report text
+            platform: Target platform
+            format_type: Override format type
+
+        Returns:
+            Platform-formatted report
+        """
+        format_type = format_type or self.config.get_report_format()
+
+        # For now, return as-is. Can be extended for platform-specific formatting
+        if format_type == "markdown":
+            return report
+        elif format_type == "text":
+            # Strip markdown formatting
+            return self._strip_markdown(report)
+        else:
+            return report
+
+    def _strip_markdown(self, text: str) -> str:
+        """Strip markdown formatting from text."""
+        # Simple markdown stripping
+        import re
+
+        # Remove bold
+        text = re.sub(r"\*\*(.*?)\*\*", r"\1", text)
+        # Remove italic
+        text = re.sub(r"\*(.*?)\*", r"\1", text)
+        # Remove headers
+        text = re.sub(r"^#+\s*", "", text, flags=re.MULTILINE)
+
+        return text
+
+    def create_report_data(
+        self,
+        group_id: str,
+        group_name: str,
+        statistics: GroupStatistics,
+        topics: List[Topic],
+        user_titles: List[UserTitle],
+        golden_quotes: List[GoldenQuote],
+    ) -> Dict[str, Any]:
+        """
+        Create a report data dictionary for storage.
+
+        Args:
+            group_id: Group identifier
+            group_name: Group display name
+            statistics: Group statistics
+            topics: List of topics
+            user_titles: List of user titles
+            golden_quotes: List of golden quotes
+
+        Returns:
+            Report data dictionary
+        """
+        return {
+            "group_id": group_id,
+            "group_name": group_name,
+            "timestamp": datetime.now().isoformat(),
+            "statistics": statistics.to_dict(),
+            "topics": [t.to_dict() for t in topics],
+            "user_titles": [u.to_dict() for u in user_titles],
+            "golden_quotes": [q.to_dict() for q in golden_quotes],
+        }

src/application/scheduling_service.py

@@ -0,0 +1,264 @@
+"""
+Scheduling Service - Application service for scheduled analysis
+
+This service manages scheduled analysis tasks and coordinates
+with the analysis orchestrator.
+"""
+
+import asyncio
+from datetime import datetime, timedelta
+from typing import Any, Callable, Dict, List, Optional, Set
+
+from astrbot.api import logger
+
+from ..infrastructure.config import ConfigManager
+from ..shared.constants import TASK_STATE_PENDING, TASK_STATE_RUNNING, TASK_STATE_COMPLETED
+
+
+class ScheduledTask:
+    """Represents a scheduled analysis task."""
+
+    def __init__(
+        self,
+        task_id: str,
+        group_id: str,
+        scheduled_time: str,  # HH:MM format
+        callback: Callable,
+        enabled: bool = True,
+    ):
+        self.task_id = task_id
+        self.group_id = group_id
+        self.scheduled_time = scheduled_time
+        self.callback = callback
+        self.enabled = enabled
+        self.last_run: Optional[datetime] = None
+        self.next_run: Optional[datetime] = None
+        self._calculate_next_run()
+
+    def _calculate_next_run(self) -> None:
+        """Calculate the next run time."""
+        if not self.enabled:
+            self.next_run = None
+            return
+
+        try:
+            hours, minutes = map(int, self.scheduled_time.split(":"))
+            now = datetime.now()
+            next_run = now.replace(hour=hours, minute=minutes, second=0, microsecond=0)
+
+            # If the time has passed today, schedule for tomorrow
+            if next_run <= now:
+                next_run += timedelta(days=1)
+
+            self.next_run = next_run
+        except ValueError:
+            logger.error(f"Invalid scheduled time format: {self.scheduled_time}")
+            self.next_run = None
+
+    def should_run(self) -> bool:
+        """Check if the task should run now."""
+        if not self.enabled or not self.next_run:
+            return False
+
+        now = datetime.now()
+
+        # Check if we're within the execution window (5 minute tolerance)
+        if self.next_run <= now <= self.next_run + timedelta(minutes=5):
+            # Check if we haven't run today
+            if self.last_run is None or self.last_run.date() != now.date():
+                return True
+
+        return False
+
+    def mark_completed(self) -> None:
+        """Mark the task as completed and schedule next run."""
+        self.last_run = datetime.now()
+        self._calculate_next_run()
+
+
+class SchedulingService:
+    """
+    Application service for managing scheduled analysis tasks.
+
+    This service runs a background loop that checks for and
+    executes scheduled tasks.
+    """
+
+    def __init__(self, config: ConfigManager):
+        """
+        Initialize the scheduling service.
+
+        Args:
+            config: Configuration manager
+        """
+        self.config = config
+        self._tasks: Dict[str, ScheduledTask] = {}
+        self._running = False
+        self._task: Optional[asyncio.Task] = None
+        self._callbacks: Dict[str, Callable] = {}
+
+    def register_callback(self, name: str, callback: Callable) -> None:
+        """
+        Register a callback for scheduled tasks.
+
+        Args:
+            name: Callback name
+            callback: Async callback function
+        """
+        self._callbacks[name] = callback
+
+    def add_task(
+        self,
+        group_id: str,
+        scheduled_time: Optional[str] = None,
+        callback_name: str = "analyze",
+    ) -> str:
+        """
+        Add a scheduled task for a group.
+
+        Args:
+            group_id: Group identifier
+            scheduled_time: Time in HH:MM format (uses config default if not provided)
+            callback_name: Name of registered callback to use
+
+        Returns:
+            Task ID
+        """
+        scheduled_time = scheduled_time or self.config.get_analysis_time()
+        task_id = f"task_{group_id}"
+
+        callback = self._callbacks.get(callback_name)
+        if not callback:
+            logger.warning(f"Callback '{callback_name}' not registered")
+            return task_id
+
+        task = ScheduledTask(
+            task_id=task_id,
+            group_id=group_id,
+            scheduled_time=scheduled_time,
+            callback=callback,
+            enabled=True,
+        )
+
+        self._tasks[task_id] = task
+        logger.info(f"Added scheduled task {task_id} for {scheduled_time}")
+
+        return task_id
+
+    def remove_task(self, task_id: str) -> bool:
+        """
+        Remove a scheduled task.
+
+        Args:
+            task_id: Task identifier
+
+        Returns:
+            True if task was removed
+        """
+        if task_id in self._tasks:
+            del self._tasks[task_id]
+            logger.info(f"Removed scheduled task {task_id}")
+            return True
+        return False
+
+    def enable_task(self, task_id: str) -> bool:
+        """Enable a scheduled task."""
+        if task_id in self._tasks:
+            self._tasks[task_id].enabled = True
+            self._tasks[task_id]._calculate_next_run()
+            return True
+        return False
+
+    def disable_task(self, task_id: str) -> bool:
+        """Disable a scheduled task."""
+        if task_id in self._tasks:
+            self._tasks[task_id].enabled = False
+            self._tasks[task_id].next_run = None
+            return True
+        return False
+
+    def get_task_status(self, task_id: str) -> Optional[Dict[str, Any]]:
+        """
+        Get status of a scheduled task.
+
+        Args:
+            task_id: Task identifier
+
+        Returns:
+            Task status dictionary or None
+        """
+        task = self._tasks.get(task_id)
+        if not task:
+            return None
+
+        return {
+            "task_id": task.task_id,
+            "group_id": task.group_id,
+            "scheduled_time": task.scheduled_time,
+            "enabled": task.enabled,
+            "last_run": task.last_run.isoformat() if task.last_run else None,
+            "next_run": task.next_run.isoformat() if task.next_run else None,
+        }
+
+    def list_tasks(self) -> List[Dict[str, Any]]:
+        """List all scheduled tasks."""
+        return [self.get_task_status(tid) for tid in self._tasks.keys()]
+
+    async def start(self) -> None:
+        """Start the scheduling service."""
+        if self._running:
+            return
+
+        self._running = True
+        self._task = asyncio.create_task(self._run_loop())
+        logger.info("Scheduling service started")
+
+    async def stop(self) -> None:
+        """Stop the scheduling service."""
+        self._running = False
+        if self._task:
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+        logger.info("Scheduling service stopped")
+
+    async def _run_loop(self) -> None:
+        """Main scheduling loop."""
+        while self._running:
+            try:
+                await self._check_and_run_tasks()
+                # Check every minute
+                await asyncio.sleep(60)
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                logger.error(f"Error in scheduling loop: {e}")
+                await asyncio.sleep(60)
+
+    async def _check_and_run_tasks(self) -> None:
+        """Check for and execute due tasks."""
+        for task in list(self._tasks.values()):
+            if task.should_run():
+                try:
+                    logger.info(f"Executing scheduled task {task.task_id}")
+                    await task.callback(task.group_id)
+                    task.mark_completed()
+                    logger.info(f"Completed scheduled task {task.task_id}")
+                except Exception as e:
+                    logger.error(f"Failed to execute task {task.task_id}: {e}")
+
+    def setup_from_config(self) -> None:
+        """Set up scheduled tasks from configuration."""
+        if not self.config.get_auto_analysis_enabled():
+            logger.info("Auto analysis is disabled")
+            return
+
+        enabled_groups = self.config.get_enabled_groups()
+        analysis_time = self.config.get_analysis_time()
+
+        for group_id in enabled_groups:
+            self.add_task(group_id, analysis_time)
+
+        logger.info(f"Set up {len(enabled_groups)} scheduled tasks")

src/domain/exceptions.py

@@ -0,0 +1,234 @@
+"""
+Domain Exceptions - Custom exceptions for the domain layer
+
+This module contains all domain-specific exceptions used throughout
+the plugin. These exceptions are platform-agnostic and represent
+business logic errors.
+"""
+
+
+class DomainException(Exception):
+    """Base exception for all domain errors."""
+
+    def __init__(self, message: str, code: str = "DOMAIN_ERROR"):
+        self.message = message
+        self.code = code
+        super().__init__(self.message)
+
+
+# ============================================================================
+# Analysis Exceptions
+# ============================================================================
+
+
+class AnalysisException(DomainException):
+    """Base exception for analysis-related errors."""
+
+    def __init__(self, message: str, code: str = "ANALYSIS_ERROR"):
+        super().__init__(message, code)
+
+
+class InsufficientDataException(AnalysisException):
+    """Raised when there is not enough data to perform analysis."""
+
+    def __init__(self, message: str = "Insufficient data for analysis"):
+        super().__init__(message, "INSUFFICIENT_DATA")
+
+
+class AnalysisTimeoutException(AnalysisException):
+    """Raised when analysis takes too long."""
+
+    def __init__(self, message: str = "Analysis timed out"):
+        super().__init__(message, "ANALYSIS_TIMEOUT")
+
+
+class LLMException(AnalysisException):
+    """Raised when LLM API call fails."""
+
+    def __init__(self, message: str = "LLM API call failed", provider: str = ""):
+        self.provider = provider
+        super().__init__(f"{message} (provider: {provider})" if provider else message, "LLM_ERROR")
+
+
+class LLMRateLimitException(LLMException):
+    """Raised when LLM API rate limit is exceeded."""
+
+    def __init__(self, message: str = "LLM rate limit exceeded", provider: str = ""):
+        super().__init__(message, provider)
+        self.code = "LLM_RATE_LIMIT"
+
+
+class LLMQuotaExceededException(LLMException):
+    """Raised when LLM API quota is exceeded."""
+
+    def __init__(self, message: str = "LLM quota exceeded", provider: str = ""):
+        super().__init__(message, provider)
+        self.code = "LLM_QUOTA_EXCEEDED"
+
+
+# ============================================================================
+# Platform Exceptions
+# ============================================================================
+
+
+class PlatformException(DomainException):
+    """Base exception for platform-related errors."""
+
+    def __init__(self, message: str, platform: str = "", code: str = "PLATFORM_ERROR"):
+        self.platform = platform
+        super().__init__(f"[{platform}] {message}" if platform else message, code)
+
+
+class PlatformNotSupportedException(PlatformException):
+    """Raised when a platform is not supported."""
+
+    def __init__(self, platform: str):
+        super().__init__(f"Platform '{platform}' is not supported", platform, "PLATFORM_NOT_SUPPORTED")
+
+
+class PlatformConnectionException(PlatformException):
+    """Raised when connection to platform fails."""
+
+    def __init__(self, message: str = "Failed to connect to platform", platform: str = ""):
+        super().__init__(message, platform, "PLATFORM_CONNECTION_ERROR")
+
+
+class PlatformAPIException(PlatformException):
+    """Raised when platform API call fails."""
+
+    def __init__(self, message: str = "Platform API call failed", platform: str = ""):
+        super().__init__(message, platform, "PLATFORM_API_ERROR")
+
+
+class MessageFetchException(PlatformException):
+    """Raised when fetching messages fails."""
+
+    def __init__(self, message: str = "Failed to fetch messages", platform: str = "", group_id: str = ""):
+        self.group_id = group_id
+        super().__init__(f"{message} (group: {group_id})" if group_id else message, platform, "MESSAGE_FETCH_ERROR")
+
+
+class MessageSendException(PlatformException):
+    """Raised when sending a message fails."""
+
+    def __init__(self, message: str = "Failed to send message", platform: str = "", group_id: str = ""):
+        self.group_id = group_id
+        super().__init__(f"{message} (group: {group_id})" if group_id else message, platform, "MESSAGE_SEND_ERROR")
+
+
+# ============================================================================
+# Configuration Exceptions
+# ============================================================================
+
+
+class ConfigurationException(DomainException):
+    """Base exception for configuration-related errors."""
+
+    def __init__(self, message: str, code: str = "CONFIG_ERROR"):
+        super().__init__(message, code)
+
+
+class InvalidConfigurationException(ConfigurationException):
+    """Raised when configuration is invalid."""
+
+    def __init__(self, message: str = "Invalid configuration", key: str = ""):
+        self.key = key
+        super().__init__(f"{message}: {key}" if key else message, "INVALID_CONFIG")
+
+
+class MissingConfigurationException(ConfigurationException):
+    """Raised when required configuration is missing."""
+
+    def __init__(self, key: str):
+        self.key = key
+        super().__init__(f"Missing required configuration: {key}", "MISSING_CONFIG")
+
+
+# ============================================================================
+# Repository Exceptions
+# ============================================================================
+
+
+class RepositoryException(DomainException):
+    """Base exception for repository-related errors."""
+
+    def __init__(self, message: str, code: str = "REPOSITORY_ERROR"):
+        super().__init__(message, code)
+
+
+class DataNotFoundException(RepositoryException):
+    """Raised when requested data is not found."""
+
+    def __init__(self, message: str = "Data not found", entity_type: str = "", entity_id: str = ""):
+        self.entity_type = entity_type
+        self.entity_id = entity_id
+        super().__init__(f"{entity_type} not found: {entity_id}" if entity_type else message, "DATA_NOT_FOUND")
+
+
+class DataPersistenceException(RepositoryException):
+    """Raised when data persistence fails."""
+
+    def __init__(self, message: str = "Failed to persist data"):
+        super().__init__(message, "DATA_PERSISTENCE_ERROR")
+
+
+# ============================================================================
+# Scheduling Exceptions
+# ============================================================================
+
+
+class SchedulingException(DomainException):
+    """Base exception for scheduling-related errors."""
+
+    def __init__(self, message: str, code: str = "SCHEDULING_ERROR"):
+        super().__init__(message, code)
+
+
+class TaskAlreadyScheduledException(SchedulingException):
+    """Raised when trying to schedule an already scheduled task."""
+
+    def __init__(self, task_id: str):
+        self.task_id = task_id
+        super().__init__(f"Task already scheduled: {task_id}", "TASK_ALREADY_SCHEDULED")
+
+
+class TaskNotFoundException(SchedulingException):
+    """Raised when a scheduled task is not found."""
+
+    def __init__(self, task_id: str):
+        self.task_id = task_id
+        super().__init__(f"Scheduled task not found: {task_id}", "TASK_NOT_FOUND")
+
+
+# ============================================================================
+# Validation Exceptions
+# ============================================================================
+
+
+class ValidationException(DomainException):
+    """Base exception for validation errors."""
+
+    def __init__(self, message: str, field: str = "", code: str = "VALIDATION_ERROR"):
+        self.field = field
+        super().__init__(f"{field}: {message}" if field else message, code)
+
+
+class InvalidGroupIdException(ValidationException):
+    """Raised when group ID is invalid."""
+
+    def __init__(self, group_id: str):
+        super().__init__(f"Invalid group ID: {group_id}", "group_id", "INVALID_GROUP_ID")
+
+
+class InvalidUserIdException(ValidationException):
+    """Raised when user ID is invalid."""
+
+    def __init__(self, user_id: str):
+        super().__init__(f"Invalid user ID: {user_id}", "user_id", "INVALID_USER_ID")
+
+
+class InvalidMessageException(ValidationException):
+    """Raised when message format is invalid."""
+
+    def __init__(self, message: str = "Invalid message format"):
+        super().__init__(message, "message", "INVALID_MESSAGE")

src/domain/services/__init__.py

@@ -0,0 +1,14 @@
+"""
+Domain Services - Business logic services for analysis
+
+This module exports all domain services that encapsulate core business logic
+for analyzing group chat data. These services are platform-agnostic.
+"""
+
+from .statistics_calculator import StatisticsCalculator
+from .report_generator import ReportGenerator
+
+__all__ = [
+    "StatisticsCalculator",
+    "ReportGenerator",
+]

src/domain/services/report_generator.py

@@ -0,0 +1,188 @@
+"""
+Report Generator - Domain service for generating analysis reports
+
+This service generates formatted reports from analysis results.
+It is platform-agnostic and produces text/markdown reports.
+"""
+
+from datetime import datetime
+from typing import List, Optional
+
+from ..value_objects.topic import Topic
+from ..value_objects.user_title import UserTitle
+from ..value_objects.golden_quote import GoldenQuote
+from ..value_objects.statistics import GroupStatistics, TokenUsage
+
+
+class ReportGenerator:
+    """
+    Domain service for generating analysis reports.
+
+    This service takes analysis results and produces formatted
+    text reports that can be sent to any platform.
+    """
+
+    def __init__(self, group_name: str = "", date_str: str = ""):
+        """
+        Initialize the report generator.
+
+        Args:
+            group_name: Name of the group for report header
+            date_str: Date string for the report
+        """
+        self.group_name = group_name
+        self.date_str = date_str or datetime.now().strftime("%Y-%m-%d")
+
+    def generate_full_report(
+        self,
+        statistics: GroupStatistics,
+        topics: List[Topic],
+        user_titles: List[UserTitle],
+        golden_quotes: List[GoldenQuote],
+        include_header: bool = True,
+        include_footer: bool = True,
+    ) -> str:
+        """
+        Generate a complete analysis report.
+
+        Args:
+            statistics: Group chat statistics
+            topics: List of discussion topics
+            user_titles: List of user titles/badges
+            golden_quotes: List of golden quotes
+            include_header: Whether to include report header
+            include_footer: Whether to include report footer
+
+        Returns:
+            Formatted report string
+        """
+        sections = []
+
+        if include_header:
+            sections.append(self._generate_header())
+
+        sections.append(self._generate_statistics_section(statistics))
+
+        if topics:
+            sections.append(self._generate_topics_section(topics))
+
+        if user_titles:
+            sections.append(self._generate_user_titles_section(user_titles))
+
+        if golden_quotes:
+            sections.append(self._generate_golden_quotes_section(golden_quotes))
+
+        if include_footer:
+            sections.append(self._generate_footer(statistics.token_usage))
+
+        return "\n\n".join(sections)
+
+    def _generate_header(self) -> str:
+        """Generate report header."""
+        title = f"📊 Group Analysis Report"
+        if self.group_name:
+            title += f" - {self.group_name}"
+
+        return f"{title}\n📅 Date: {self.date_str}\n{'=' * 40}"
+
+    def _generate_statistics_section(self, stats: GroupStatistics) -> str:
+        """Generate statistics section."""
+        lines = [
+            "📈 **Statistics Overview**",
+            f"• Total Messages: {stats.message_count}",
+            f"• Total Characters: {stats.total_characters}",
+            f"• Participants: {stats.participant_count}",
+            f"• Average Message Length: {stats.average_message_length:.1f} chars",
+            f"• Most Active Period: {stats.most_active_period}",
+        ]
+
+        if stats.emoji_count > 0:
+            lines.append(f"• Emoji Used: {stats.emoji_count}")
+
+        return "\n".join(lines)
+
+    def _generate_topics_section(self, topics: List[Topic]) -> str:
+        """Generate topics section."""
+        lines = ["💬 **Discussion Topics**"]
+
+        for i, topic in enumerate(topics, 1):
+            contributors_str = ", ".join(topic.contributors[:3])
+            if len(topic.contributors) > 3:
+                contributors_str += f" +{len(topic.contributors) - 3} more"
+
+            lines.append(f"\n{i}. **{topic.name}**")
+            lines.append(f"   Contributors: {contributors_str}")
+            if topic.detail:
+                # Truncate long details
+                detail = topic.detail[:200] + "..." if len(topic.detail) > 200 else topic.detail
+                lines.append(f"   {detail}")
+
+        return "\n".join(lines)
+
+    def _generate_user_titles_section(self, titles: List[UserTitle]) -> str:
+        """Generate user titles section."""
+        lines = ["🏆 **User Titles & Badges**"]
+
+        for title in titles:
+            lines.append(f"\n👤 **{title.name}**")
+            lines.append(f"   🎖️ Title: {title.title}")
+            if title.mbti:
+                lines.append(f"   🧠 MBTI: {title.mbti}")
+            if title.reason:
+                reason = title.reason[:150] + "..." if len(title.reason) > 150 else title.reason
+                lines.append(f"   💡 Reason: {reason}")
+
+        return "\n".join(lines)
+
+    def _generate_golden_quotes_section(self, quotes: List[GoldenQuote]) -> str:
+        """Generate golden quotes section."""
+        lines = ["✨ **Golden Quotes**"]
+
+        for i, quote in enumerate(quotes, 1):
+            lines.append(f"\n{i}. \"{quote.content}\"")
+            lines.append(f"   — {quote.sender}")
+            if quote.reason:
+                reason = quote.reason[:100] + "..." if len(quote.reason) > 100 else quote.reason
+                lines.append(f"   ({reason})")
+
+        return "\n".join(lines)
+
+    def _generate_footer(self, token_usage: Optional[TokenUsage] = None) -> str:
+        """Generate report footer."""
+        lines = ["─" * 40]
+        lines.append(f"Generated at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+
+        if token_usage and token_usage.total_tokens > 0:
+            lines.append(f"Token Usage: {token_usage.total_tokens} tokens")
+
+        return "\n".join(lines)
+
+    def generate_summary_report(
+        self,
+        statistics: GroupStatistics,
+        top_topic: Optional[Topic] = None,
+        top_quote: Optional[GoldenQuote] = None,
+    ) -> str:
+        """
+        Generate a brief summary report.
+
+        Args:
+            statistics: Group chat statistics
+            top_topic: Most significant topic (optional)
+            top_quote: Best golden quote (optional)
+
+        Returns:
+            Brief summary string
+        """
+        lines = [
+            f"📊 Daily Summary ({self.date_str})",
+            f"Messages: {statistics.message_count} | Participants: {statistics.participant_count}",
+        ]
+
+        if top_topic:
+            lines.append(f"🔥 Hot Topic: {top_topic.name}")
+
+        if top_quote:
+            lines.append(f"✨ Quote: \"{top_quote.content}\" — {top_quote.sender}")
+
+        return "\n".join(lines)

src/domain/services/statistics_calculator.py

@@ -0,0 +1,266 @@
+"""
+Statistics Calculator - Domain service for computing chat statistics
+
+This service calculates various statistics from unified messages.
+It is platform-agnostic and works with the domain value objects.
+"""
+
+from datetime import datetime
+from typing import Dict, List, Optional
+
+from ..value_objects import UnifiedMessage
+from ..value_objects.statistics import (
+    GroupStatistics,
+    UserStatistics,
+    EmojiStatistics,
+    ActivityVisualization,
+    TokenUsage,
+)
+
+
+class StatisticsCalculator:
+    """
+    Domain service for calculating group chat statistics.
+
+    This service processes UnifiedMessage objects and produces
+    platform-agnostic statistics.
+    """
+
+    def __init__(self, bot_user_ids: Optional[List[str]] = None):
+        """
+        Initialize the statistics calculator.
+
+        Args:
+            bot_user_ids: List of bot user IDs to filter out from statistics
+        """
+        self.bot_user_ids = set(bot_user_ids or [])
+
+    def calculate_group_statistics(
+        self,
+        messages: List[UnifiedMessage],
+        token_usage: Optional[TokenUsage] = None,
+    ) -> GroupStatistics:
+        """
+        Calculate comprehensive group statistics from messages.
+
+        Args:
+            messages: List of unified messages to analyze
+            token_usage: Optional token usage from LLM analysis
+
+        Returns:
+            GroupStatistics object with computed statistics
+        """
+        if not messages:
+            return GroupStatistics()
+
+        # Filter out bot messages
+        filtered_messages = [
+            msg for msg in messages if msg.sender_id not in self.bot_user_ids
+        ]
+
+        if not filtered_messages:
+            return GroupStatistics()
+
+        # Calculate basic statistics
+        message_count = len(filtered_messages)
+        total_characters = sum(len(msg.text_content) for msg in filtered_messages)
+        unique_senders = set(msg.sender_id for msg in filtered_messages)
+        participant_count = len(unique_senders)
+
+        # Calculate emoji statistics
+        emoji_stats = self._calculate_emoji_statistics(filtered_messages)
+
+        # Calculate activity visualization
+        activity_viz = self._calculate_activity_visualization(filtered_messages)
+
+        # Determine most active period
+        most_active_period = self._determine_most_active_period(activity_viz)
+
+        return GroupStatistics(
+            message_count=message_count,
+            total_characters=total_characters,
+            participant_count=participant_count,
+            most_active_period=most_active_period,
+            emoji_statistics=emoji_stats,
+            activity_visualization=activity_viz,
+            token_usage=token_usage or TokenUsage(),
+        )
+
+    def calculate_user_statistics(
+        self, messages: List[UnifiedMessage]
+    ) -> Dict[str, UserStatistics]:
+        """
+        Calculate per-user statistics from messages.
+
+        Args:
+            messages: List of unified messages to analyze
+
+        Returns:
+            Dictionary mapping user_id to UserStatistics
+        """
+        user_stats: Dict[str, UserStatistics] = {}
+
+        for msg in messages:
+            # Skip bot messages
+            if msg.sender_id in self.bot_user_ids:
+                continue
+
+            user_id = msg.sender_id
+
+            if user_id not in user_stats:
+                user_stats[user_id] = UserStatistics(
+                    user_id=user_id,
+                    nickname=msg.sender_name,
+                )
+
+            stats = user_stats[user_id]
+            stats.message_count += 1
+            stats.char_count += len(msg.text_content)
+            stats.emoji_count += msg.emoji_count
+
+            # Count replies
+            if msg.reply_to_id:
+                stats.reply_count += 1
+
+            # Track hourly activity
+            hour = msg.timestamp.hour
+            stats.hours[hour] = stats.hours.get(hour, 0) + 1
+
+        return user_stats
+
+    def get_top_users(
+        self,
+        user_stats: Dict[str, UserStatistics],
+        limit: int = 10,
+        min_messages: int = 5,
+    ) -> List[Dict]:
+        """
+        Get top users by message count.
+
+        Args:
+            user_stats: Dictionary of user statistics
+            limit: Maximum number of users to return
+            min_messages: Minimum messages required to be included
+
+        Returns:
+            List of top user dictionaries sorted by message count
+        """
+        eligible_users = [
+            stats for stats in user_stats.values() if stats.message_count >= min_messages
+        ]
+
+        sorted_users = sorted(
+            eligible_users, key=lambda x: x.message_count, reverse=True
+        )
+
+        return [
+            {
+                "user_id": u.user_id,
+                "nickname": u.nickname,
+                "name": u.nickname,  # Backward compatibility
+                "message_count": u.message_count,
+                "avg_chars": round(u.average_chars, 1),
+                "emoji_ratio": round(u.emoji_ratio, 2),
+                "night_ratio": round(u.night_ratio, 2),
+                "reply_ratio": round(u.reply_ratio, 2),
+            }
+            for u in sorted_users[:limit]
+        ]
+
+    def _calculate_emoji_statistics(
+        self, messages: List[UnifiedMessage]
+    ) -> EmojiStatistics:
+        """Calculate emoji usage statistics from messages."""
+        standard_count = 0
+        custom_count = 0
+        animated_count = 0
+        sticker_count = 0
+        other_count = 0
+        emoji_details: Dict[str, int] = {}
+
+        for msg in messages:
+            for content in msg.contents:
+                if content.type.value == "emoji":
+                    emoji_id = content.metadata.get("emoji_id", "unknown")
+                    emoji_details[emoji_id] = emoji_details.get(emoji_id, 0) + 1
+
+                    emoji_type = content.metadata.get("emoji_type", "standard")
+                    if emoji_type == "standard":
+                        standard_count += 1
+                    elif emoji_type == "custom":
+                        custom_count += 1
+                    elif emoji_type == "animated":
+                        animated_count += 1
+                    elif emoji_type == "sticker":
+                        sticker_count += 1
+                    else:
+                        other_count += 1
+
+        return EmojiStatistics(
+            standard_emoji_count=standard_count,
+            custom_emoji_count=custom_count,
+            animated_emoji_count=animated_count,
+            sticker_count=sticker_count,
+            other_emoji_count=other_count,
+            emoji_details=tuple(emoji_details.items()),
+        )
+
+    def _calculate_activity_visualization(
+        self, messages: List[UnifiedMessage]
+    ) -> ActivityVisualization:
+        """Calculate activity visualization data from messages."""
+        hourly: Dict[int, int] = {h: 0 for h in range(24)}
+        daily: Dict[str, int] = {}
+        user_counts: Dict[str, int] = {}
+
+        for msg in messages:
+            # Hourly activity
+            hour = msg.timestamp.hour
+            hourly[hour] += 1
+
+            # Daily activity
+            date_str = msg.timestamp.strftime("%Y-%m-%d")
+            daily[date_str] = daily.get(date_str, 0) + 1
+
+            # User activity
+            user_counts[msg.sender_id] = user_counts.get(msg.sender_id, 0) + 1
+
+        # Calculate peak hours (top 3)
+        sorted_hours = sorted(hourly.items(), key=lambda x: x[1], reverse=True)
+        peak_hours = [h for h, _ in sorted_hours[:3]]
+
+        # User activity ranking
+        sorted_users = sorted(user_counts.items(), key=lambda x: x[1], reverse=True)
+        user_ranking = [
+            {"user_id": uid, "count": count} for uid, count in sorted_users[:20]
+        ]
+
+        return ActivityVisualization(
+            hourly_activity=tuple(hourly.items()),
+            daily_activity=tuple(daily.items()),
+            user_activity_ranking=tuple(user_ranking),
+            peak_hours=tuple(peak_hours),
+            heatmap_data=tuple(),  # Can be extended for heatmap visualization
+        )
+
+    def _determine_most_active_period(
+        self, activity: ActivityVisualization
+    ) -> str:
+        """Determine the most active time period description."""
+        hourly = dict(activity.hourly_activity)
+
+        if not hourly:
+            return "Unknown"
+
+        # Find peak hour
+        peak_hour = max(hourly, key=hourly.get)
+
+        # Categorize time periods
+        if 6 <= peak_hour < 12:
+            return "Morning (6:00-12:00)"
+        elif 12 <= peak_hour < 18:
+            return "Afternoon (12:00-18:00)"
+        elif 18 <= peak_hour < 24:
+            return "Evening (18:00-24:00)"
+        else:
+            return "Late Night (0:00-6:00)"

src/domain/value_objects/__init__.py

@@ -2,13 +2,37 @@
 from .unified_message import UnifiedMessage, MessageContent, MessageContentType
 from .platform_capabilities import PlatformCapabilities, PLATFORM_CAPABILITIES
 from .unified_group import UnifiedGroup, UnifiedMember
+from .topic import Topic, TopicCollection
+from .user_title import UserTitle, UserTitleCollection
+from .golden_quote import GoldenQuote, GoldenQuoteCollection
+from .statistics import (
+    TokenUsage,
+    EmojiStatistics,
+    ActivityVisualization,
+    GroupStatistics,
+    UserStatistics,
+)
 
 __all__ = [
+    # Core platform abstractions
     "UnifiedMessage",
-    "MessageContent", 
+    "MessageContent",
     "MessageContentType",
     "PlatformCapabilities",
     "PLATFORM_CAPABILITIES",
     "UnifiedGroup",
     "UnifiedMember",
+    # Analysis value objects
+    "Topic",
+    "TopicCollection",
+    "UserTitle",
+    "UserTitleCollection",
+    "GoldenQuote",
+    "GoldenQuoteCollection",
+    # Statistics
+    "TokenUsage",
+    "EmojiStatistics",
+    "ActivityVisualization",
+    "GroupStatistics",
+    "UserStatistics",
 ]

src/domain/value_objects/golden_quote.py

@@ -0,0 +1,137 @@
+"""
+GoldenQuote Value Object - Platform-agnostic golden quote representation
+
+This value object represents a memorable quote extracted from group chat messages.
+It is immutable and contains no platform-specific logic.
+"""
+
+from dataclasses import dataclass, field
+from typing import List
+
+
+@dataclass(frozen=True)
+class GoldenQuote:
+    """
+    GoldenQuote value object for group chat analysis.
+
+    Represents a memorable/interesting quote from the chat.
+    Immutable by design (frozen=True).
+
+    Attributes:
+        content: The actual quote content
+        sender: Display name of the person who said it
+        reason: Why this quote was selected as golden
+        user_id: Platform-agnostic user identifier (stored as string)
+    """
+
+    content: str
+    sender: str
+    reason: str = ""
+    user_id: str = ""
+
+    def __post_init__(self):
+        """Validate and normalize golden quote data after initialization."""
+        # Ensure user_id is always a string
+        if not isinstance(self.user_id, str):
+            object.__setattr__(self, "user_id", str(self.user_id))
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "GoldenQuote":
+        """
+        Create GoldenQuote from dictionary data.
+
+        Args:
+            data: Dictionary with golden quote data
+
+        Returns:
+            GoldenQuote instance
+        """
+        # Handle both 'qq' and 'user_id' keys for backward compatibility
+        user_id = data.get("user_id", data.get("qq", ""))
+
+        return cls(
+            content=data.get("content", "").strip(),
+            sender=data.get("sender", "").strip(),
+            reason=data.get("reason", "").strip(),
+            user_id=str(user_id) if user_id else "",
+        )
+
+    def to_dict(self) -> dict:
+        """
+        Convert GoldenQuote to dictionary.
+
+        Returns:
+            Dictionary representation
+        """
+        return {
+            "content": self.content,
+            "sender": self.sender,
+            "reason": self.reason,
+            "user_id": self.user_id,
+            "qq": int(self.user_id) if self.user_id.isdigit() else 0,  # Backward compat
+        }
+
+    @property
+    def is_valid(self) -> bool:
+        """Check if golden quote has valid data."""
+        return bool(
+            self.content and self.content.strip() and self.sender and self.sender.strip()
+        )
+
+    @property
+    def qq(self) -> int:
+        """Get QQ number for backward compatibility."""
+        try:
+            return int(self.user_id)
+        except (ValueError, TypeError):
+            return 0
+
+    def with_user_id(self, user_id: str) -> "GoldenQuote":
+        """
+        Create a new GoldenQuote with updated user_id.
+
+        Since GoldenQuote is frozen, we need to create a new instance.
+
+        Args:
+            user_id: The user ID to set
+
+        Returns:
+            New GoldenQuote instance with updated user_id
+        """
+        return GoldenQuote(
+            content=self.content,
+            sender=self.sender,
+            reason=self.reason,
+            user_id=str(user_id),
+        )
+
+
+@dataclass
+class GoldenQuoteCollection:
+    """
+    Collection of golden quotes with utility methods.
+
+    This is mutable to allow building up a collection of quotes.
+    """
+
+    quotes: List[GoldenQuote] = field(default_factory=list)
+
+    def add(self, quote: GoldenQuote) -> None:
+        """Add a golden quote to the collection."""
+        if quote.is_valid:
+            self.quotes.append(quote)
+
+    def add_from_dict(self, data: dict) -> None:
+        """Add a golden quote from dictionary data."""
+        quote = GoldenQuote.from_dict(data)
+        self.add(quote)
+
+    def to_list(self) -> List[dict]:
+        """Convert all quotes to list of dictionaries."""
+        return [q.to_dict() for q in self.quotes]
+
+    def __len__(self) -> int:
+        return len(self.quotes)
+
+    def __iter__(self):
+        return iter(self.quotes)

src/domain/value_objects/statistics.py

@@ -0,0 +1,317 @@
+"""
+Statistics Value Objects - Platform-agnostic statistics representations
+
+This module contains value objects for various statistics collected during
+group chat analysis. All objects are immutable and platform-agnostic.
+"""
+
+from dataclasses import dataclass, field
+from typing import Dict, List
+
+
+@dataclass(frozen=True)
+class TokenUsage:
+    """
+    Token usage statistics for LLM API calls.
+
+    Immutable by design (frozen=True).
+
+    Attributes:
+        prompt_tokens: Number of tokens in the prompt
+        completion_tokens: Number of tokens in the completion
+        total_tokens: Total tokens used
+    """
+
+    prompt_tokens: int = 0
+    completion_tokens: int = 0
+    total_tokens: int = 0
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "TokenUsage":
+        """Create TokenUsage from dictionary."""
+        return cls(
+            prompt_tokens=data.get("prompt_tokens", 0),
+            completion_tokens=data.get("completion_tokens", 0),
+            total_tokens=data.get("total_tokens", 0),
+        )
+
+    def to_dict(self) -> dict:
+        """Convert to dictionary."""
+        return {
+            "prompt_tokens": self.prompt_tokens,
+            "completion_tokens": self.completion_tokens,
+            "total_tokens": self.total_tokens,
+        }
+
+    def __add__(self, other: "TokenUsage") -> "TokenUsage":
+        """Add two TokenUsage objects together."""
+        if not isinstance(other, TokenUsage):
+            return NotImplemented
+        return TokenUsage(
+            prompt_tokens=self.prompt_tokens + other.prompt_tokens,
+            completion_tokens=self.completion_tokens + other.completion_tokens,
+            total_tokens=self.total_tokens + other.total_tokens,
+        )
+
+
+@dataclass(frozen=True)
+class EmojiStatistics:
+    """
+    Emoji usage statistics.
+
+    Platform-agnostic representation of emoji usage in messages.
+    Immutable by design (frozen=True).
+
+    Attributes:
+        standard_emoji_count: Standard unicode emoji count
+        custom_emoji_count: Platform-specific custom emoji count
+        animated_emoji_count: Animated emoji count
+        sticker_count: Sticker count
+        other_emoji_count: Other emoji types count
+        emoji_details: Detailed breakdown by emoji ID/name
+    """
+
+    standard_emoji_count: int = 0
+    custom_emoji_count: int = 0
+    animated_emoji_count: int = 0
+    sticker_count: int = 0
+    other_emoji_count: int = 0
+    emoji_details: tuple = field(default_factory=tuple)
+
+    @property
+    def total_count(self) -> int:
+        """Get total emoji count."""
+        return (
+            self.standard_emoji_count
+            + self.custom_emoji_count
+            + self.animated_emoji_count
+            + self.sticker_count
+            + self.other_emoji_count
+        )
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "EmojiStatistics":
+        """Create EmojiStatistics from dictionary."""
+        details = data.get("face_details", data.get("emoji_details", {}))
+        if isinstance(details, dict):
+            details = tuple(details.items())
+
+        return cls(
+            standard_emoji_count=data.get("face_count", data.get("standard_emoji_count", 0)),
+            custom_emoji_count=data.get("mface_count", data.get("custom_emoji_count", 0)),
+            animated_emoji_count=data.get("bface_count", data.get("animated_emoji_count", 0)),
+            sticker_count=data.get("sface_count", data.get("sticker_count", 0)),
+            other_emoji_count=data.get("other_emoji_count", 0),
+            emoji_details=details,
+        )
+
+    def to_dict(self) -> dict:
+        """Convert to dictionary."""
+        return {
+            "standard_emoji_count": self.standard_emoji_count,
+            "custom_emoji_count": self.custom_emoji_count,
+            "animated_emoji_count": self.animated_emoji_count,
+            "sticker_count": self.sticker_count,
+            "other_emoji_count": self.other_emoji_count,
+            "total_emoji_count": self.total_count,
+            "emoji_details": dict(self.emoji_details),
+            # Backward compatibility
+            "face_count": self.standard_emoji_count,
+            "mface_count": self.custom_emoji_count,
+            "bface_count": self.animated_emoji_count,
+            "sface_count": self.sticker_count,
+        }
+
+
+@dataclass(frozen=True)
+class ActivityVisualization:
+    """
+    Activity visualization data.
+
+    Platform-agnostic representation of chat activity patterns.
+    Immutable by design (frozen=True).
+
+    Attributes:
+        hourly_activity: Message count by hour (0-23)
+        daily_activity: Message count by date
+        user_activity_ranking: Ranked list of user activity
+        peak_hours: List of peak activity hours
+        heatmap_data: Data for activity heatmap visualization
+    """
+
+    hourly_activity: tuple = field(default_factory=tuple)
+    daily_activity: tuple = field(default_factory=tuple)
+    user_activity_ranking: tuple = field(default_factory=tuple)
+    peak_hours: tuple = field(default_factory=tuple)
+    heatmap_data: tuple = field(default_factory=tuple)
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "ActivityVisualization":
+        """Create ActivityVisualization from dictionary."""
+        hourly = data.get("hourly_activity", {})
+        daily = data.get("daily_activity", {})
+        ranking = data.get("user_activity_ranking", [])
+        peaks = data.get("peak_hours", [])
+        heatmap = data.get("activity_heatmap_data", data.get("heatmap_data", {}))
+
+        return cls(
+            hourly_activity=tuple(hourly.items()) if isinstance(hourly, dict) else tuple(hourly),
+            daily_activity=tuple(daily.items()) if isinstance(daily, dict) else tuple(daily),
+            user_activity_ranking=tuple(ranking),
+            peak_hours=tuple(peaks),
+            heatmap_data=tuple(heatmap.items()) if isinstance(heatmap, dict) else tuple(heatmap),
+        )
+
+    def to_dict(self) -> dict:
+        """Convert to dictionary."""
+        return {
+            "hourly_activity": dict(self.hourly_activity),
+            "daily_activity": dict(self.daily_activity),
+            "user_activity_ranking": list(self.user_activity_ranking),
+            "peak_hours": list(self.peak_hours),
+            "activity_heatmap_data": dict(self.heatmap_data),
+        }
+
+
+@dataclass(frozen=True)
+class GroupStatistics:
+    """
+    Comprehensive group chat statistics.
+
+    Platform-agnostic representation of group chat statistics.
+    Immutable by design (frozen=True).
+
+    Attributes:
+        message_count: Total number of messages
+        total_characters: Total character count across all messages
+        participant_count: Number of unique participants
+        most_active_period: Description of the most active time period
+        emoji_statistics: Emoji usage statistics
+        activity_visualization: Activity pattern data
+        token_usage: LLM token usage for analysis
+    """
+
+    message_count: int = 0
+    total_characters: int = 0
+    participant_count: int = 0
+    most_active_period: str = ""
+    emoji_statistics: EmojiStatistics = field(default_factory=EmojiStatistics)
+    activity_visualization: ActivityVisualization = field(default_factory=ActivityVisualization)
+    token_usage: TokenUsage = field(default_factory=TokenUsage)
+
+    @property
+    def average_message_length(self) -> float:
+        """Calculate average message length."""
+        if self.message_count == 0:
+            return 0.0
+        return self.total_characters / self.message_count
+
+    @property
+    def emoji_count(self) -> int:
+        """Get total emoji count for backward compatibility."""
+        return self.emoji_statistics.total_count
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "GroupStatistics":
+        """Create GroupStatistics from dictionary."""
+        emoji_data = data.get("emoji_statistics", {})
+        if not emoji_data:
+            # Backward compatibility: construct from flat fields
+            emoji_data = {
+                "face_count": data.get("emoji_count", 0),
+            }
+
+        activity_data = data.get("activity_visualization", {})
+        token_data = data.get("token_usage", {})
+
+        return cls(
+            message_count=data.get("message_count", 0),
+            total_characters=data.get("total_characters", 0),
+            participant_count=data.get("participant_count", 0),
+            most_active_period=data.get("most_active_period", ""),
+            emoji_statistics=EmojiStatistics.from_dict(emoji_data),
+            activity_visualization=ActivityVisualization.from_dict(activity_data),
+            token_usage=TokenUsage.from_dict(token_data),
+        )
+
+    def to_dict(self) -> dict:
+        """Convert to dictionary."""
+        return {
+            "message_count": self.message_count,
+            "total_characters": self.total_characters,
+            "participant_count": self.participant_count,
+            "most_active_period": self.most_active_period,
+            "emoji_count": self.emoji_count,  # Backward compatibility
+            "emoji_statistics": self.emoji_statistics.to_dict(),
+            "activity_visualization": self.activity_visualization.to_dict(),
+            "token_usage": self.token_usage.to_dict(),
+        }
+
+
+@dataclass
+class UserStatistics:
+    """
+    Per-user statistics (mutable for accumulation during analysis).
+
+    Attributes:
+        user_id: Platform-agnostic user identifier
+        nickname: User's display name
+        message_count: Number of messages sent
+        char_count: Total characters sent
+        emoji_count: Number of emojis used
+        reply_count: Number of replies made
+        hours: Message count by hour (0-23)
+    """
+
+    user_id: str
+    nickname: str = ""
+    message_count: int = 0
+    char_count: int = 0
+    emoji_count: int = 0
+    reply_count: int = 0
+    hours: Dict[int, int] = field(default_factory=lambda: {h: 0 for h in range(24)})
+
+    @property
+    def average_chars(self) -> float:
+        """Calculate average characters per message."""
+        if self.message_count == 0:
+            return 0.0
+        return self.char_count / self.message_count
+
+    @property
+    def emoji_ratio(self) -> float:
+        """Calculate emoji per message ratio."""
+        if self.message_count == 0:
+            return 0.0
+        return self.emoji_count / self.message_count
+
+    @property
+    def night_ratio(self) -> float:
+        """Calculate night activity ratio (0-6 hours)."""
+        if self.message_count == 0:
+            return 0.0
+        night_messages = sum(self.hours.get(h, 0) for h in range(6))
+        return night_messages / self.message_count
+
+    @property
+    def reply_ratio(self) -> float:
+        """Calculate reply ratio."""
+        if self.message_count == 0:
+            return 0.0
+        return self.reply_count / self.message_count
+
+    def to_dict(self) -> dict:
+        """Convert to dictionary."""
+        return {
+            "user_id": self.user_id,
+            "nickname": self.nickname,
+            "message_count": self.message_count,
+            "char_count": self.char_count,
+            "emoji_count": self.emoji_count,
+            "reply_count": self.reply_count,
+            "avg_chars": round(self.average_chars, 1),
+            "emoji_ratio": round(self.emoji_ratio, 2),
+            "night_ratio": round(self.night_ratio, 2),
+            "reply_ratio": round(self.reply_ratio, 2),
+            "hours": self.hours,
+        }

src/domain/value_objects/topic.py

@@ -0,0 +1,112 @@
+"""
+Topic Value Object - Platform-agnostic topic representation
+
+This value object represents a discussion topic extracted from group chat messages.
+It is immutable and contains no platform-specific logic.
+"""
+
+from dataclasses import dataclass, field
+from typing import List
+
+
+@dataclass(frozen=True)
+class Topic:
+    """
+    Topic value object for group chat analysis.
+
+    Represents a discussion topic with contributors and details.
+    Immutable by design (frozen=True).
+
+    Attributes:
+        name: Topic title/name
+        contributors: List of usernames who participated in this topic
+        detail: Detailed description or summary of the topic discussion
+    """
+
+    name: str
+    contributors: tuple[str, ...] = field(default_factory=tuple)
+    detail: str = ""
+
+    def __post_init__(self):
+        """Validate topic data after initialization."""
+        if not self.name or not self.name.strip():
+            object.__setattr__(self, "name", "Unknown Topic")
+
+        # Ensure contributors is a tuple for immutability
+        if isinstance(self.contributors, list):
+            object.__setattr__(self, "contributors", tuple(self.contributors))
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "Topic":
+        """
+        Create Topic from dictionary data.
+
+        Args:
+            data: Dictionary with topic data
+
+        Returns:
+            Topic instance
+        """
+        contributors = data.get("contributors", [])
+        if isinstance(contributors, list):
+            contributors = tuple(contributors)
+
+        return cls(
+            name=data.get("topic", data.get("name", "")).strip(),
+            contributors=contributors,
+            detail=data.get("detail", "").strip(),
+        )
+
+    def to_dict(self) -> dict:
+        """
+        Convert Topic to dictionary.
+
+        Returns:
+            Dictionary representation
+        """
+        return {
+            "topic": self.name,
+            "contributors": list(self.contributors),
+            "detail": self.detail,
+        }
+
+    @property
+    def contributor_count(self) -> int:
+        """Get the number of contributors."""
+        return len(self.contributors)
+
+    @property
+    def is_valid(self) -> bool:
+        """Check if topic has valid data."""
+        return bool(self.name and self.name.strip() and self.detail and self.detail.strip())
+
+
+@dataclass
+class TopicCollection:
+    """
+    Collection of topics with utility methods.
+
+    This is mutable to allow building up a collection of topics.
+    """
+
+    topics: List[Topic] = field(default_factory=list)
+
+    def add(self, topic: Topic) -> None:
+        """Add a topic to the collection."""
+        if topic.is_valid:
+            self.topics.append(topic)
+
+    def add_from_dict(self, data: dict) -> None:
+        """Add a topic from dictionary data."""
+        topic = Topic.from_dict(data)
+        self.add(topic)
+
+    def to_list(self) -> List[dict]:
+        """Convert all topics to list of dictionaries."""
+        return [t.to_dict() for t in self.topics]
+
+    def __len__(self) -> int:
+        return len(self.topics)
+
+    def __iter__(self):
+        return iter(self.topics)

src/domain/value_objects/user_title.py

@@ -0,0 +1,134 @@
+"""
+UserTitle Value Object - Platform-agnostic user title representation
+
+This value object represents a user's title/badge assigned based on their
+chat behavior analysis. It is immutable and contains no platform-specific logic.
+"""
+
+from dataclasses import dataclass, field
+from typing import List
+
+
+@dataclass(frozen=True)
+class UserTitle:
+    """
+    UserTitle value object for group chat analysis.
+
+    Represents a title/badge assigned to a user based on their behavior.
+    Immutable by design (frozen=True).
+
+    Attributes:
+        name: User's display name
+        user_id: Platform-agnostic user identifier (stored as string)
+        title: The title/badge assigned to the user
+        mbti: MBTI personality type assessment
+        reason: Explanation for why this title was assigned
+    """
+
+    name: str
+    user_id: str
+    title: str
+    mbti: str = ""
+    reason: str = ""
+
+    def __post_init__(self):
+        """Validate and normalize user title data after initialization."""
+        # Ensure user_id is always a string
+        if not isinstance(self.user_id, str):
+            object.__setattr__(self, "user_id", str(self.user_id))
+
+    @classmethod
+    def from_dict(cls, data: dict) -> "UserTitle":
+        """
+        Create UserTitle from dictionary data.
+
+        Args:
+            data: Dictionary with user title data
+
+        Returns:
+            UserTitle instance
+        """
+        # Handle both 'qq' and 'user_id' keys for backward compatibility
+        user_id = data.get("user_id", data.get("qq", ""))
+
+        return cls(
+            name=data.get("name", "").strip(),
+            user_id=str(user_id),
+            title=data.get("title", "").strip(),
+            mbti=data.get("mbti", "").strip().upper(),
+            reason=data.get("reason", "").strip(),
+        )
+
+    def to_dict(self) -> dict:
+        """
+        Convert UserTitle to dictionary.
+
+        Returns:
+            Dictionary representation
+        """
+        return {
+            "name": self.name,
+            "user_id": self.user_id,
+            "qq": int(self.user_id) if self.user_id.isdigit() else 0,  # Backward compat
+            "title": self.title,
+            "mbti": self.mbti,
+            "reason": self.reason,
+        }
+
+    @property
+    def is_valid(self) -> bool:
+        """Check if user title has valid data."""
+        return bool(
+            self.name
+            and self.name.strip()
+            and self.title
+            and self.title.strip()
+            and self.user_id
+        )
+
+    @property
+    def qq(self) -> int:
+        """Get QQ number for backward compatibility."""
+        try:
+            return int(self.user_id)
+        except (ValueError, TypeError):
+            return 0
+
+
+@dataclass
+class UserTitleCollection:
+    """
+    Collection of user titles with utility methods.
+
+    This is mutable to allow building up a collection of titles.
+    """
+
+    titles: List[UserTitle] = field(default_factory=list)
+
+    def add(self, title: UserTitle) -> None:
+        """Add a user title to the collection."""
+        if title.is_valid:
+            self.titles.append(title)
+
+    def add_from_dict(self, data: dict) -> None:
+        """Add a user title from dictionary data."""
+        title = UserTitle.from_dict(data)
+        self.add(title)
+
+    def get_by_user_id(self, user_id: str) -> UserTitle | None:
+        """Get title by user ID."""
+        user_id_str = str(user_id)
+        for title in self.titles:
+            if title.user_id == user_id_str:
+                return title
+        return None
+
+    def to_list(self) -> List[dict]:
+        """Convert all titles to list of dictionaries."""
+        return [t.to_dict() for t in self.titles]
+
+    def __len__(self) -> int:
+        return len(self.titles)
+
+    def __iter__(self):
+        return iter(self.titles)

src/infrastructure/__init__.py

@@ -1 +1,24 @@
 # Infrastructure Layer
+from .platform import PlatformAdapter, PlatformAdapterFactory, OneBotAdapter
+from .persistence import HistoryRepository
+from .llm import LLMClient
+from .config import ConfigManager
+from .resilience import CircuitBreaker, RateLimiter, retry_async, RetryConfig
+
+__all__ = [
+    # Platform
+    "PlatformAdapter",
+    "PlatformAdapterFactory",
+    "OneBotAdapter",
+    # Persistence
+    "HistoryRepository",
+    # LLM
+    "LLMClient",
+    # Config
+    "ConfigManager",
+    # Resilience
+    "CircuitBreaker",
+    "RateLimiter",
+    "retry_async",
+    "RetryConfig",
+]

src/infrastructure/config/__init__.py

@@ -0,0 +1,7 @@
+"""
+Config Module - Configuration management
+"""
+
+from .config_manager import ConfigManager
+
+__all__ = ["ConfigManager"]

src/infrastructure/config/config_manager.py

@@ -0,0 +1,239 @@
+"""
+Config Manager - Centralized configuration management
+
+This module provides a unified interface for accessing plugin configuration,
+wrapping the existing config module with additional validation and defaults.
+"""
+
+from typing import Any, Dict, List, Optional
+
+from astrbot.api import logger
+
+
+class ConfigManager:
+    """
+    Centralized configuration manager for the plugin.
+
+    Provides typed access to configuration values with defaults
+    and validation.
+    """
+
+    def __init__(self, config: Dict[str, Any]):
+        """
+        Initialize the configuration manager.
+
+        Args:
+            config: Raw configuration dictionary
+        """
+        self._config = config or {}
+
+    def get(self, key: str, default: Any = None) -> Any:
+        """
+        Get a configuration value.
+
+        Args:
+            key: Configuration key (supports dot notation)
+            default: Default value if key not found
+
+        Returns:
+            Configuration value or default
+        """
+        try:
+            keys = key.split(".")
+            value = self._config
+            for k in keys:
+                if isinstance(value, dict):
+                    value = value.get(k)
+                else:
+                    return default
+                if value is None:
+                    return default
+            return value
+        except Exception:
+            return default
+
+    def set(self, key: str, value: Any) -> None:
+        """
+        Set a configuration value.
+
+        Args:
+            key: Configuration key
+            value: Value to set
+        """
+        keys = key.split(".")
+        config = self._config
+        for k in keys[:-1]:
+            if k not in config:
+                config[k] = {}
+            config = config[k]
+        config[keys[-1]] = value
+
+    # ========================================================================
+    # Group Configuration
+    # ========================================================================
+
+    def get_enabled_groups(self) -> List[str]:
+        """Get list of enabled group IDs."""
+        groups = self.get("enabled_groups", [])
+        return [str(g) for g in groups] if groups else []
+
+    def is_group_enabled(self, group_id: str) -> bool:
+        """Check if a group is enabled for analysis."""
+        enabled = self.get_enabled_groups()
+        return str(group_id) in enabled or not enabled  # Empty means all enabled
+
+    def get_bot_qq_ids(self) -> List[str]:
+        """Get list of bot QQ IDs to filter out."""
+        ids = self.get("bot_qq_ids", [])
+        return [str(i) for i in ids] if ids else []
+
+    # ========================================================================
+    # Analysis Configuration
+    # ========================================================================
+
+    def get_max_topics(self) -> int:
+        """Get maximum number of topics to extract."""
+        return int(self.get("max_topics", 5))
+
+    def get_max_user_titles(self) -> int:
+        """Get maximum number of user titles to generate."""
+        return int(self.get("max_user_titles", 10))
+
+    def get_max_golden_quotes(self) -> int:
+        """Get maximum number of golden quotes to extract."""
+        return int(self.get("max_golden_quotes", 5))
+
+    def get_min_messages_for_analysis(self) -> int:
+        """Get minimum messages required for analysis."""
+        return int(self.get("min_messages", 50))
+
+    # ========================================================================
+    # LLM Configuration
+    # ========================================================================
+
+    def get_topic_provider_id(self) -> Optional[str]:
+        """Get provider ID for topic analysis."""
+        return self.get("topic_provider_id")
+
+    def get_user_title_provider_id(self) -> Optional[str]:
+        """Get provider ID for user title analysis."""
+        return self.get("user_title_provider_id")
+
+    def get_golden_quote_provider_id(self) -> Optional[str]:
+        """Get provider ID for golden quote analysis."""
+        return self.get("golden_quote_provider_id")
+
+    def get_topic_max_tokens(self) -> int:
+        """Get max tokens for topic analysis."""
+        return int(self.get("topic_max_tokens", 2000))
+
+    def get_user_title_max_tokens(self) -> int:
+        """Get max tokens for user title analysis."""
+        return int(self.get("user_title_max_tokens", 2000))
+
+    def get_golden_quote_max_tokens(self) -> int:
+        """Get max tokens for golden quote analysis."""
+        return int(self.get("golden_quote_max_tokens", 1500))
+
+    # ========================================================================
+    # Prompt Configuration
+    # ========================================================================
+
+    def get_topic_analysis_prompt(self) -> Optional[str]:
+        """Get custom prompt template for topic analysis."""
+        return self.get("prompts.topic_analysis")
+
+    def get_user_title_analysis_prompt(self) -> Optional[str]:
+        """Get custom prompt template for user title analysis."""
+        return self.get("prompts.user_title_analysis")
+
+    def get_golden_quote_analysis_prompt(self) -> Optional[str]:
+        """Get custom prompt template for golden quote analysis."""
+        return self.get("prompts.golden_quote_analysis")
+
+    # ========================================================================
+    # Scheduling Configuration
+    # ========================================================================
+
+    def get_auto_analysis_enabled(self) -> bool:
+        """Check if auto analysis is enabled."""
+        return bool(self.get("auto_analysis_enabled", False))
+
+    def get_analysis_time(self) -> str:
+        """Get scheduled analysis time (HH:MM format)."""
+        return str(self.get("analysis_time", "23:00"))
+
+    def get_analysis_timezone(self) -> str:
+        """Get timezone for scheduled analysis."""
+        return str(self.get("timezone", "Asia/Shanghai"))
+
+    # ========================================================================
+    # Report Configuration
+    # ========================================================================
+
+    def get_report_format(self) -> str:
+        """Get report format (text, markdown, image)."""
+        return str(self.get("report_format", "text"))
+
+    def get_include_statistics(self) -> bool:
+        """Check if statistics should be included in reports."""
+        return bool(self.get("include_statistics", True))
+
+    def get_include_topics(self) -> bool:
+        """Check if topics should be included in reports."""
+        return bool(self.get("include_topics", True))
+
+    def get_include_user_titles(self) -> bool:
+        """Check if user titles should be included in reports."""
+        return bool(self.get("include_user_titles", True))
+
+    def get_include_golden_quotes(self) -> bool:
+        """Check if golden quotes should be included in reports."""
+        return bool(self.get("include_golden_quotes", True))
+
+    # ========================================================================
+    # Utility Methods
+    # ========================================================================
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Get the raw configuration dictionary."""
+        return self._config.copy()
+
+    def update(self, updates: Dict[str, Any]) -> None:
+        """
+        Update configuration with new values.
+
+        Args:
+            updates: Dictionary of updates to apply
+        """
+        self._config.update(updates)
+
+    def validate(self) -> List[str]:
+        """
+        Validate the configuration.
+
+        Returns:
+            List of validation error messages (empty if valid)
+        """
+        errors = []
+
+        # Validate numeric ranges
+        if self.get_max_topics() < 1 or self.get_max_topics() > 20:
+            errors.append("max_topics must be between 1 and 20")
+
+        if self.get_max_user_titles() < 1 or self.get_max_user_titles() > 50:
+            errors.append("max_user_titles must be between 1 and 50")
+
+        if self.get_max_golden_quotes() < 1 or self.get_max_golden_quotes() > 20:
+            errors.append("max_golden_quotes must be between 1 and 20")
+
+        # Validate time format
+        time_str = self.get_analysis_time()
+        try:
+            hours, minutes = time_str.split(":")
+            if not (0 <= int(hours) <= 23 and 0 <= int(minutes) <= 59):
+                errors.append("analysis_time must be in HH:MM format (00:00-23:59)")
+        except ValueError:
+            errors.append("analysis_time must be in HH:MM format")
+
+        return errors

src/infrastructure/llm/__init__.py

@@ -0,0 +1,7 @@
+"""
+LLM Module - LLM client implementations
+"""
+
+from .llm_client import LLMClient
+
+__all__ = ["LLMClient"]

src/infrastructure/llm/llm_client.py

@@ -0,0 +1,186 @@
+"""
+LLM Client - Wrapper for AstrBot's LLM provider system
+
+This module provides a clean interface to AstrBot's LLM capabilities,
+abstracting away the provider management details.
+"""
+
+from typing import Any, Dict, List, Optional, Tuple
+
+from astrbot.api import logger
+
+from ...domain.value_objects.statistics import TokenUsage
+from ...domain.exceptions import LLMException, LLMRateLimitException
+
+
+class LLMClient:
+    """
+    Client for interacting with LLM providers.
+
+    This class wraps AstrBot's provider system and provides
+    a clean interface for making LLM calls.
+    """
+
+    def __init__(self, context: Any):
+        """
+        Initialize the LLM client.
+
+        Args:
+            context: AstrBot plugin context with provider access
+        """
+        self.context = context
+        self._provider_cache: Dict[str, Any] = {}
+
+    def get_provider(self, provider_id: Optional[str] = None) -> Any:
+        """
+        Get an LLM provider by ID.
+
+        Args:
+            provider_id: Specific provider ID, or None for default
+
+        Returns:
+            Provider instance
+
+        Raises:
+            LLMException: If provider not found
+        """
+        try:
+            if provider_id and provider_id in self._provider_cache:
+                return self._provider_cache[provider_id]
+
+            if provider_id:
+                provider = self.context.get_provider_by_id(provider_id)
+            else:
+                # Get default provider
+                providers = self.context.get_all_providers()
+                if not providers:
+                    raise LLMException("No LLM providers available")
+                provider = providers[0]
+
+            if provider:
+                self._provider_cache[provider_id or "default"] = provider
+
+            return provider
+
+        except Exception as e:
+            raise LLMException(f"Failed to get provider: {e}")
+
+    async def chat_completion(
+        self,
+        prompt: str,
+        provider_id: Optional[str] = None,
+        max_tokens: int = 2000,
+        temperature: float = 0.7,
+        system_prompt: Optional[str] = None,
+    ) -> Tuple[str, TokenUsage]:
+        """
+        Make a chat completion request.
+
+        Args:
+            prompt: The user prompt
+            provider_id: Specific provider ID (optional)
+            max_tokens: Maximum tokens in response
+            temperature: Sampling temperature
+            system_prompt: Optional system prompt
+
+        Returns:
+            Tuple of (response_text, token_usage)
+
+        Raises:
+            LLMException: If the request fails
+        """
+        try:
+            provider = self.get_provider(provider_id)
+            if not provider:
+                raise LLMException("No provider available", provider_id or "default")
+
+            # Build messages
+            messages = []
+            if system_prompt:
+                messages.append({"role": "system", "content": system_prompt})
+            messages.append({"role": "user", "content": prompt})
+
+            # Make the request
+            response = await provider.text_chat(
+                prompt=prompt,
+                session_id=None,  # Stateless
+            )
+
+            # Extract response text
+            if hasattr(response, "completion_text"):
+                response_text = response.completion_text
+            elif isinstance(response, dict):
+                response_text = response.get("completion_text", response.get("text", ""))
+            else:
+                response_text = str(response)
+
+            # Extract token usage
+            token_usage = TokenUsage()
+            if hasattr(response, "usage"):
+                usage = response.usage
+                if hasattr(usage, "prompt_tokens"):
+                    token_usage = TokenUsage(
+                        prompt_tokens=usage.prompt_tokens or 0,
+                        completion_tokens=usage.completion_tokens or 0,
+                        total_tokens=usage.total_tokens or 0,
+                    )
+
+            return response_text, token_usage
+
+        except Exception as e:
+            error_msg = str(e).lower()
+            if "rate limit" in error_msg or "429" in error_msg:
+                raise LLMRateLimitException(str(e), provider_id or "default")
+            raise LLMException(f"Chat completion failed: {e}", provider_id or "default")
+
+    async def analyze_with_json_output(
+        self,
+        prompt: str,
+        provider_id: Optional[str] = None,
+        max_tokens: int = 2000,
+        temperature: float = 0.7,
+    ) -> Tuple[str, TokenUsage]:
+        """
+        Make a completion request expecting JSON output.
+
+        Args:
+            prompt: The analysis prompt
+            provider_id: Specific provider ID (optional)
+            max_tokens: Maximum tokens in response
+            temperature: Sampling temperature
+
+        Returns:
+            Tuple of (response_text, token_usage)
+        """
+        # Add JSON instruction to prompt if not present
+        json_instruction = "\nRespond with valid JSON only."
+        if "json" not in prompt.lower():
+            prompt = prompt + json_instruction
+
+        return await self.chat_completion(
+            prompt=prompt,
+            provider_id=provider_id,
+            max_tokens=max_tokens,
+            temperature=temperature,
+        )
+
+    def list_available_providers(self) -> List[Dict[str, str]]:
+        """
+        List all available LLM providers.
+
+        Returns:
+            List of provider info dictionaries
+        """
+        try:
+            providers = self.context.get_all_providers()
+            return [
+                {
+                    "id": getattr(p, "id", str(i)),
+                    "name": getattr(p, "name", f"Provider {i}"),
+                    "type": getattr(p, "type", "unknown"),
+                }
+                for i, p in enumerate(providers)
+            ]
+        except Exception as e:
+            logger.error(f"Failed to list providers: {e}")
+            return []

src/infrastructure/persistence/__init__.py

@@ -0,0 +1,7 @@
+"""
+Persistence Module - Data storage implementations
+"""
+
+from .history_repository import HistoryRepository
+
+__all__ = ["HistoryRepository"]

src/infrastructure/persistence/history_repository.py

@@ -0,0 +1,212 @@
+"""
+History Repository - Implementation for storing analysis history
+
+This module provides persistent storage for analysis results and history.
+It wraps the existing history_manager functionality.
+"""
+
+import json
+import os
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from astrbot.api import logger
+
+
+class HistoryRepository:
+    """
+    Repository for storing and retrieving analysis history.
+
+    This implementation stores history as JSON files, maintaining
+    backward compatibility with the existing history_manager.
+    """
+
+    def __init__(self, data_dir: str):
+        """
+        Initialize the history repository.
+
+        Args:
+            data_dir: Base directory for storing history data
+        """
+        self.data_dir = Path(data_dir)
+        self.history_dir = self.data_dir / "history"
+        self._ensure_directories()
+
+    def _ensure_directories(self) -> None:
+        """Ensure required directories exist."""
+        self.history_dir.mkdir(parents=True, exist_ok=True)
+
+    def _get_group_history_path(self, group_id: str) -> Path:
+        """Get the history file path for a group."""
+        return self.history_dir / f"group_{group_id}.json"
+
+    def save_analysis_result(
+        self,
+        group_id: str,
+        result: Dict[str, Any],
+        date_str: Optional[str] = None,
+    ) -> bool:
+        """
+        Save an analysis result to history.
+
+        Args:
+            group_id: The group identifier
+            result: Analysis result dictionary
+            date_str: Date string (defaults to today)
+
+        Returns:
+            True if saved successfully
+        """
+        try:
+            date_str = date_str or datetime.now().strftime("%Y-%m-%d")
+            history = self.load_group_history(group_id)
+
+            # Add timestamp if not present
+            if "timestamp" not in result:
+                result["timestamp"] = datetime.now().isoformat()
+
+            # Store by date
+            if "daily" not in history:
+                history["daily"] = {}
+
+            history["daily"][date_str] = result
+            history["last_updated"] = datetime.now().isoformat()
+
+            # Write to file
+            history_path = self._get_group_history_path(group_id)
+            with open(history_path, "w", encoding="utf-8") as f:
+                json.dump(history, f, ensure_ascii=False, indent=2)
+
+            logger.debug(f"Saved analysis result for group {group_id} on {date_str}")
+            return True
+
+        except Exception as e:
+            logger.error(f"Failed to save analysis result: {e}")
+            return False
+
+    def load_group_history(self, group_id: str) -> Dict[str, Any]:
+        """
+        Load history for a group.
+
+        Args:
+            group_id: The group identifier
+
+        Returns:
+            History dictionary
+        """
+        try:
+            history_path = self._get_group_history_path(group_id)
+            if history_path.exists():
+                with open(history_path, "r", encoding="utf-8") as f:
+                    return json.load(f)
+            return {"daily": {}, "group_id": group_id}
+        except Exception as e:
+            logger.error(f"Failed to load group history: {e}")
+            return {"daily": {}, "group_id": group_id}
+
+    def get_analysis_result(
+        self, group_id: str, date_str: str
+    ) -> Optional[Dict[str, Any]]:
+        """
+        Get analysis result for a specific date.
+
+        Args:
+            group_id: The group identifier
+            date_str: Date string (YYYY-MM-DD format)
+
+        Returns:
+            Analysis result or None if not found
+        """
+        history = self.load_group_history(group_id)
+        return history.get("daily", {}).get(date_str)
+
+    def get_recent_results(
+        self, group_id: str, limit: int = 7
+    ) -> List[Dict[str, Any]]:
+        """
+        Get recent analysis results.
+
+        Args:
+            group_id: The group identifier
+            limit: Maximum number of results to return
+
+        Returns:
+            List of recent analysis results
+        """
+        history = self.load_group_history(group_id)
+        daily = history.get("daily", {})
+
+        # Sort by date descending
+        sorted_dates = sorted(daily.keys(), reverse=True)[:limit]
+        return [daily[date] for date in sorted_dates]
+
+    def has_analysis_for_date(self, group_id: str, date_str: str) -> bool:
+        """
+        Check if analysis exists for a specific date.
+
+        Args:
+            group_id: The group identifier
+            date_str: Date string (YYYY-MM-DD format)
+
+        Returns:
+            True if analysis exists
+        """
+        result = self.get_analysis_result(group_id, date_str)
+        return result is not None
+
+    def delete_old_history(self, group_id: str, keep_days: int = 30) -> int:
+        """
+        Delete history older than specified days.
+
+        Args:
+            group_id: The group identifier
+            keep_days: Number of days of history to keep
+
+        Returns:
+            Number of entries deleted
+        """
+        try:
+            history = self.load_group_history(group_id)
+            daily = history.get("daily", {})
+
+            cutoff_date = datetime.now().strftime("%Y-%m-%d")
+            # Calculate cutoff (simple string comparison works for YYYY-MM-DD format)
+            from datetime import timedelta
+
+            cutoff = (datetime.now() - timedelta(days=keep_days)).strftime("%Y-%m-%d")
+
+            # Find dates to delete
+            dates_to_delete = [date for date in daily.keys() if date < cutoff]
+
+            for date in dates_to_delete:
+                del daily[date]
+
+            if dates_to_delete:
+                history["daily"] = daily
+                history_path = self._get_group_history_path(group_id)
+                with open(history_path, "w", encoding="utf-8") as f:
+                    json.dump(history, f, ensure_ascii=False, indent=2)
+
+            return len(dates_to_delete)
+
+        except Exception as e:
+            logger.error(f"Failed to delete old history: {e}")
+            return 0
+
+    def list_groups_with_history(self) -> List[str]:
+        """
+        List all groups that have history.
+
+        Returns:
+            List of group IDs
+        """
+        try:
+            groups = []
+            for file_path in self.history_dir.glob("group_*.json"):
+                group_id = file_path.stem.replace("group_", "")
+                groups.append(group_id)
+            return groups
+        except Exception as e:
+            logger.error(f"Failed to list groups: {e}")
+            return []

src/infrastructure/platform/__init__.py

@@ -1,5 +1,6 @@
 # Platform Adapters
 from .factory import PlatformAdapterFactory
 from .base import PlatformAdapter
+from .adapters.onebot_adapter import OneBotAdapter
 
-__all__ = ["PlatformAdapterFactory", "PlatformAdapter"]
+__all__ = ["PlatformAdapterFactory", "PlatformAdapter", "OneBotAdapter"]

src/infrastructure/resilience/__init__.py

@@ -0,0 +1,15 @@
+"""
+Resilience Module - Circuit breaker, rate limiter, and retry utilities
+"""
+
+from .circuit_breaker import CircuitBreaker, CircuitState
+from .rate_limiter import RateLimiter
+from .retry import retry_async, RetryConfig
+
+__all__ = [
+    "CircuitBreaker",
+    "CircuitState",
+    "RateLimiter",
+    "retry_async",
+    "RetryConfig",
+]

src/infrastructure/resilience/circuit_breaker.py

@@ -0,0 +1,138 @@
+"""
+Circuit Breaker - Prevents cascading failures
+
+Implements the circuit breaker pattern to prevent repeated calls
+to failing services.
+"""
+
+import time
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Callable, Optional
+
+from astrbot.api import logger
+
+
+class CircuitState(Enum):
+    """Circuit breaker states."""
+
+    CLOSED = "closed"  # Normal operation
+    OPEN = "open"  # Failing, reject calls
+    HALF_OPEN = "half_open"  # Testing if service recovered
+
+
+@dataclass
+class CircuitBreaker:
+    """
+    Circuit breaker implementation.
+
+    Prevents cascading failures by tracking failure rates and
+    temporarily blocking calls to failing services.
+    """
+
+    name: str
+    failure_threshold: int = 5
+    recovery_timeout: float = 30.0
+    half_open_max_calls: int = 3
+
+    # Internal state
+    _state: CircuitState = field(default=CircuitState.CLOSED, init=False)
+    _failure_count: int = field(default=0, init=False)
+    _success_count: int = field(default=0, init=False)
+    _last_failure_time: float = field(default=0, init=False)
+    _half_open_calls: int = field(default=0, init=False)
+
+    @property
+    def state(self) -> CircuitState:
+        """Get current circuit state, checking for recovery."""
+        if self._state == CircuitState.OPEN:
+            if time.time() - self._last_failure_time >= self.recovery_timeout:
+                self._transition_to(CircuitState.HALF_OPEN)
+        return self._state
+
+    def _transition_to(self, new_state: CircuitState) -> None:
+        """Transition to a new state."""
+        old_state = self._state
+        self._state = new_state
+
+        if new_state == CircuitState.CLOSED:
+            self._failure_count = 0
+            self._success_count = 0
+        elif new_state == CircuitState.HALF_OPEN:
+            self._half_open_calls = 0
+
+        logger.debug(f"Circuit {self.name}: {old_state.value} -> {new_state.value}")
+
+    def record_success(self) -> None:
+        """Record a successful call."""
+        if self._state == CircuitState.HALF_OPEN:
+            self._success_count += 1
+            if self._success_count >= self.half_open_max_calls:
+                self._transition_to(CircuitState.CLOSED)
+        elif self._state == CircuitState.CLOSED:
+            # Reset failure count on success
+            self._failure_count = 0
+
+    def record_failure(self) -> None:
+        """Record a failed call."""
+        self._failure_count += 1
+        self._last_failure_time = time.time()
+
+        if self._state == CircuitState.HALF_OPEN:
+            self._transition_to(CircuitState.OPEN)
+        elif self._state == CircuitState.CLOSED:
+            if self._failure_count >= self.failure_threshold:
+                self._transition_to(CircuitState.OPEN)
+
+    def can_execute(self) -> bool:
+        """Check if a call can be executed."""
+        state = self.state  # This may trigger state transition
+
+        if state == CircuitState.CLOSED:
+            return True
+        elif state == CircuitState.OPEN:
+            return False
+        elif state == CircuitState.HALF_OPEN:
+            self._half_open_calls += 1
+            return self._half_open_calls <= self.half_open_max_calls
+
+        return False
+
+    def reset(self) -> None:
+        """Reset the circuit breaker to closed state."""
+        self._transition_to(CircuitState.CLOSED)
+
+    async def execute(
+        self,
+        func: Callable,
+        *args,
+        fallback: Optional[Callable] = None,
+        **kwargs,
+    ):
+        """
+        Execute a function with circuit breaker protection.
+
+        Args:
+            func: Async function to execute
+            *args: Function arguments
+            fallback: Optional fallback function if circuit is open
+            **kwargs: Function keyword arguments
+
+        Returns:
+            Function result or fallback result
+
+        Raises:
+            Exception: If circuit is open and no fallback provided
+        """
+        if not self.can_execute():
+            if fallback:
+                return await fallback(*args, **kwargs)
+            raise Exception(f"Circuit {self.name} is open")
+
+        try:
+            result = await func(*args, **kwargs)
+            self.record_success()
+            return result
+        except Exception as e:
+            self.record_failure()
+            raise

src/infrastructure/resilience/rate_limiter.py

@@ -0,0 +1,142 @@
+"""
+Rate Limiter - Controls request rates
+
+Implements token bucket rate limiting to prevent overwhelming services.
+"""
+
+import asyncio
+import time
+from dataclasses import dataclass, field
+from typing import Optional
+
+from astrbot.api import logger
+
+
+@dataclass
+class RateLimiter:
+    """
+    Token bucket rate limiter.
+
+    Controls the rate of operations by using a token bucket algorithm.
+    """
+
+    name: str
+    rate: float  # Tokens per second
+    burst: int  # Maximum burst size (bucket capacity)
+
+    # Internal state
+    _tokens: float = field(default=0, init=False)
+    _last_update: float = field(default=0, init=False)
+    _lock: asyncio.Lock = field(default_factory=asyncio.Lock, init=False)
+
+    def __post_init__(self):
+        """Initialize the token bucket."""
+        self._tokens = float(self.burst)
+        self._last_update = time.time()
+
+    def _refill(self) -> None:
+        """Refill tokens based on elapsed time."""
+        now = time.time()
+        elapsed = now - self._last_update
+        self._tokens = min(self.burst, self._tokens + elapsed * self.rate)
+        self._last_update = now
+
+    async def acquire(self, tokens: int = 1, timeout: Optional[float] = None) -> bool:
+        """
+        Acquire tokens from the bucket.
+
+        Args:
+            tokens: Number of tokens to acquire
+            timeout: Maximum time to wait (None = wait forever)
+
+        Returns:
+            True if tokens acquired, False if timeout
+        """
+        start_time = time.time()
+
+        async with self._lock:
+            while True:
+                self._refill()
+
+                if self._tokens >= tokens:
+                    self._tokens -= tokens
+                    return True
+
+                if timeout is not None:
+                    elapsed = time.time() - start_time
+                    if elapsed >= timeout:
+                        return False
+
+                # Calculate wait time for enough tokens
+                tokens_needed = tokens - self._tokens
+                wait_time = tokens_needed / self.rate
+
+                if timeout is not None:
+                    remaining = timeout - (time.time() - start_time)
+                    wait_time = min(wait_time, remaining)
+
+                if wait_time > 0:
+                    await asyncio.sleep(wait_time)
+
+    def try_acquire(self, tokens: int = 1) -> bool:
+        """
+        Try to acquire tokens without waiting.
+
+        Args:
+            tokens: Number of tokens to acquire
+
+        Returns:
+            True if tokens acquired, False otherwise
+        """
+        self._refill()
+
+        if self._tokens >= tokens:
+            self._tokens -= tokens
+            return True
+        return False
+
+    @property
+    def available_tokens(self) -> float:
+        """Get current available tokens."""
+        self._refill()
+        return self._tokens
+
+    def reset(self) -> None:
+        """Reset the rate limiter to full capacity."""
+        self._tokens = float(self.burst)
+        self._last_update = time.time()
+
+
+class RateLimiterGroup:
+    """
+    Group of rate limiters for different operations.
+    """
+
+    def __init__(self):
+        self._limiters: dict[str, RateLimiter] = {}
+
+    def get_or_create(
+        self,
+        name: str,
+        rate: float = 1.0,
+        burst: int = 5,
+    ) -> RateLimiter:
+        """
+        Get or create a rate limiter.
+
+        Args:
+            name: Limiter name
+            rate: Tokens per second
+            burst: Maximum burst size
+
+        Returns:
+            RateLimiter instance
+        """
+        if name not in self._limiters:
+            self._limiters[name] = RateLimiter(name=name, rate=rate, burst=burst)
+        return self._limiters[name]
+
+    def reset_all(self) -> None:
+        """Reset all rate limiters."""
+        for limiter in self._limiters.values():
+            limiter.reset()

src/infrastructure/resilience/retry.py

@@ -0,0 +1,176 @@
+"""
+Retry - Retry utilities with exponential backoff
+
+Provides retry decorators and utilities for handling transient failures.
+"""
+
+import asyncio
+import random
+from dataclasses import dataclass
+from functools import wraps
+from typing import Callable, Optional, Tuple, Type, Union
+
+from astrbot.api import logger
+
+
+@dataclass
+class RetryConfig:
+    """Configuration for retry behavior."""
+
+    max_attempts: int = 3
+    base_delay: float = 1.0
+    max_delay: float = 60.0
+    exponential_base: float = 2.0
+    jitter: bool = True
+    retry_exceptions: Tuple[Type[Exception], ...] = (Exception,)
+
+
+def calculate_delay(
+    attempt: int,
+    base_delay: float,
+    max_delay: float,
+    exponential_base: float,
+    jitter: bool,
+) -> float:
+    """
+    Calculate delay for a retry attempt.
+
+    Args:
+        attempt: Current attempt number (0-based)
+        base_delay: Base delay in seconds
+        max_delay: Maximum delay in seconds
+        exponential_base: Base for exponential backoff
+        jitter: Whether to add random jitter
+
+    Returns:
+        Delay in seconds
+    """
+    delay = base_delay * (exponential_base**attempt)
+    delay = min(delay, max_delay)
+
+    if jitter:
+        delay = delay * (0.5 + random.random())
+
+    return delay
+
+
+def retry_async(
+    max_attempts: int = 3,
+    base_delay: float = 1.0,
+    max_delay: float = 60.0,
+    exponential_base: float = 2.0,
+    jitter: bool = True,
+    retry_exceptions: Tuple[Type[Exception], ...] = (Exception,),
+    on_retry: Optional[Callable[[Exception, int], None]] = None,
+):
+    """
+    Decorator for retrying async functions with exponential backoff.
+
+    Args:
+        max_attempts: Maximum number of attempts
+        base_delay: Base delay between retries
+        max_delay: Maximum delay between retries
+        exponential_base: Base for exponential backoff
+        jitter: Whether to add random jitter
+        retry_exceptions: Tuple of exceptions to retry on
+        on_retry: Optional callback on retry (exception, attempt)
+
+    Returns:
+        Decorated function
+    """
+
+    def decorator(func: Callable):
+        @wraps(func)
+        async def wrapper(*args, **kwargs):
+            last_exception = None
+
+            for attempt in range(max_attempts):
+                try:
+                    return await func(*args, **kwargs)
+                except retry_exceptions as e:
+                    last_exception = e
+
+                    if attempt < max_attempts - 1:
+                        delay = calculate_delay(
+                            attempt, base_delay, max_delay, exponential_base, jitter
+                        )
+
+                        if on_retry:
+                            on_retry(e, attempt + 1)
+
+                        logger.debug(
+                            f"Retry {attempt + 1}/{max_attempts} for {func.__name__} "
+                            f"after {delay:.2f}s: {e}"
+                        )
+                        await asyncio.sleep(delay)
+                    else:
+                        logger.warning(
+                            f"All {max_attempts} attempts failed for {func.__name__}: {e}"
+                        )
+
+            raise last_exception
+
+        return wrapper
+
+    return decorator
+
+
+class RetryExecutor:
+    """
+    Executor for running functions with retry logic.
+    """
+
+    def __init__(self, config: Optional[RetryConfig] = None):
+        """
+        Initialize the retry executor.
+
+        Args:
+            config: Retry configuration
+        """
+        self.config = config or RetryConfig()
+
+    async def execute(
+        self,
+        func: Callable,
+        *args,
+        config: Optional[RetryConfig] = None,
+        **kwargs,
+    ):
+        """
+        Execute a function with retry logic.
+
+        Args:
+            func: Async function to execute
+            *args: Function arguments
+            config: Optional override config
+            **kwargs: Function keyword arguments
+
+        Returns:
+            Function result
+
+        Raises:
+            Exception: If all retries fail
+        """
+        cfg = config or self.config
+        last_exception = None
+
+        for attempt in range(cfg.max_attempts):
+            try:
+                return await func(*args, **kwargs)
+            except cfg.retry_exceptions as e:
+                last_exception = e
+
+                if attempt < cfg.max_attempts - 1:
+                    delay = calculate_delay(
+                        attempt,
+                        cfg.base_delay,
+                        cfg.max_delay,
+                        cfg.exponential_base,
+                        cfg.jitter,
+                    )
+                    logger.debug(
+                        f"Retry {attempt + 1}/{cfg.max_attempts} after {delay:.2f}s: {e}"
+                    )
+                    await asyncio.sleep(delay)
+
+        raise last_exception

src/shared/__init__.py

@@ -0,0 +1,11 @@
+"""
+Shared Module - Common utilities and constants
+"""
+
+from .constants import *
+from .trace_context import TraceContext
+
+__all__ = [
+    "TraceContext",
+    # Constants are exported via *
+]

src/shared/constants.py

@@ -0,0 +1,91 @@
+"""
+Constants - Shared constants used across the plugin
+"""
+
+# Plugin metadata
+PLUGIN_NAME = "astrbot_plugin_qq_group_daily_analysis"
+PLUGIN_VERSION = "2.0.0"
+
+# Platform identifiers
+PLATFORM_ONEBOT = "onebot"
+PLATFORM_TELEGRAM = "telegram"
+PLATFORM_DISCORD = "discord"
+PLATFORM_SLACK = "slack"
+PLATFORM_LARK = "lark"
+
+SUPPORTED_PLATFORMS = [
+    PLATFORM_ONEBOT,
+    # Future platforms
+    # PLATFORM_TELEGRAM,
+    # PLATFORM_DISCORD,
+    # PLATFORM_SLACK,
+    # PLATFORM_LARK,
+]
+
+# Analysis defaults
+DEFAULT_MAX_TOPICS = 5
+DEFAULT_MAX_USER_TITLES = 10
+DEFAULT_MAX_GOLDEN_QUOTES = 5
+DEFAULT_MIN_MESSAGES = 50
+DEFAULT_MAX_TOKENS = 2000
+
+# Time periods
+HOUR_RANGES = {
+    "morning": (6, 12),
+    "afternoon": (12, 18),
+    "evening": (18, 24),
+    "night": (0, 6),
+}
+
+# Report formats
+REPORT_FORMAT_TEXT = "text"
+REPORT_FORMAT_MARKDOWN = "markdown"
+REPORT_FORMAT_IMAGE = "image"
+REPORT_FORMAT_HTML = "html"
+
+# Message content types
+CONTENT_TYPE_TEXT = "text"
+CONTENT_TYPE_IMAGE = "image"
+CONTENT_TYPE_EMOJI = "emoji"
+CONTENT_TYPE_STICKER = "sticker"
+CONTENT_TYPE_FILE = "file"
+CONTENT_TYPE_AUDIO = "audio"
+CONTENT_TYPE_VIDEO = "video"
+CONTENT_TYPE_REPLY = "reply"
+CONTENT_TYPE_AT = "at"
+CONTENT_TYPE_UNKNOWN = "unknown"
+
+# Analysis task states
+TASK_STATE_PENDING = "pending"
+TASK_STATE_RUNNING = "running"
+TASK_STATE_COMPLETED = "completed"
+TASK_STATE_FAILED = "failed"
+TASK_STATE_CANCELLED = "cancelled"
+
+# Error codes
+ERROR_INSUFFICIENT_DATA = "INSUFFICIENT_DATA"
+ERROR_LLM_FAILED = "LLM_FAILED"
+ERROR_PLATFORM_ERROR = "PLATFORM_ERROR"
+ERROR_CONFIG_ERROR = "CONFIG_ERROR"
+ERROR_TIMEOUT = "TIMEOUT"
+
+# Cache TTL (in seconds)
+CACHE_TTL_SHORT = 60  # 1 minute
+CACHE_TTL_MEDIUM = 300  # 5 minutes
+CACHE_TTL_LONG = 3600  # 1 hour
+CACHE_TTL_DAY = 86400  # 24 hours
+
+# Rate limiting defaults
+RATE_LIMIT_LLM_CALLS = 10  # calls per minute
+RATE_LIMIT_API_CALLS = 60  # calls per minute
+RATE_LIMIT_BURST = 5  # burst size
+
+# Retry defaults
+RETRY_MAX_ATTEMPTS = 3
+RETRY_BASE_DELAY = 1.0
+RETRY_MAX_DELAY = 30.0
+
+# File paths
+HISTORY_DIR = "history"
+CACHE_DIR = "cache"
+TEMP_DIR = "temp"

src/shared/trace_context.py

@@ -0,0 +1,161 @@
+"""
+Trace Context - Request tracing and correlation
+
+Provides context for tracking requests across the plugin.
+"""
+
+import uuid
+from contextvars import ContextVar
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Any, Dict, Optional
+
+# Context variable for current trace
+_current_trace: ContextVar[Optional["TraceContext"]] = ContextVar(
+    "current_trace", default=None
+)
+
+
+@dataclass
+class TraceContext:
+    """
+    Context for tracing requests through the plugin.
+
+    Provides correlation IDs and timing information for debugging
+    and monitoring.
+    """
+
+    trace_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
+    group_id: str = ""
+    platform: str = ""
+    operation: str = ""
+    start_time: datetime = field(default_factory=datetime.now)
+    metadata: Dict[str, Any] = field(default_factory=dict)
+
+    # Timing data
+    _checkpoints: Dict[str, datetime] = field(default_factory=dict, init=False)
+
+    def checkpoint(self, name: str) -> None:
+        """
+        Record a timing checkpoint.
+
+        Args:
+            name: Checkpoint name
+        """
+        self._checkpoints[name] = datetime.now()
+
+    def elapsed_ms(self, from_checkpoint: Optional[str] = None) -> float:
+        """
+        Get elapsed time in milliseconds.
+
+        Args:
+            from_checkpoint: Optional checkpoint to measure from
+
+        Returns:
+            Elapsed time in milliseconds
+        """
+        start = self.start_time
+        if from_checkpoint and from_checkpoint in self._checkpoints:
+            start = self._checkpoints[from_checkpoint]
+
+        delta = datetime.now() - start
+        return delta.total_seconds() * 1000
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert trace context to dictionary."""
+        return {
+            "trace_id": self.trace_id,
+            "group_id": self.group_id,
+            "platform": self.platform,
+            "operation": self.operation,
+            "start_time": self.start_time.isoformat(),
+            "elapsed_ms": self.elapsed_ms(),
+            "metadata": self.metadata,
+            "checkpoints": {k: v.isoformat() for k, v in self._checkpoints.items()},
+        }
+
+    def __enter__(self) -> "TraceContext":
+        """Enter context manager."""
+        _current_trace.set(self)
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
+        """Exit context manager."""
+        _current_trace.set(None)
+
+    @classmethod
+    def current(cls) -> Optional["TraceContext"]:
+        """Get the current trace context."""
+        return _current_trace.get()
+
+    @classmethod
+    def get_or_create(
+        cls,
+        group_id: str = "",
+        platform: str = "",
+        operation: str = "",
+    ) -> "TraceContext":
+        """
+        Get current trace or create a new one.
+
+        Args:
+            group_id: Group identifier
+            platform: Platform name
+            operation: Operation name
+
+        Returns:
+            TraceContext instance
+        """
+        current = cls.current()
+        if current:
+            return current
+
+        return cls(
+            group_id=group_id,
+            platform=platform,
+            operation=operation,
+        )
+
+
+def get_trace_id() -> str:
+    """
+    Get current trace ID or generate a new one.
+
+    Returns:
+        Trace ID string
+    """
+    trace = TraceContext.current()
+    if trace:
+        return trace.trace_id
+    return str(uuid.uuid4())[:8]
+
+
+def with_trace(
+    group_id: str = "",
+    platform: str = "",
+    operation: str = "",
+):
+    """
+    Decorator to add trace context to a function.
+
+    Args:
+        group_id: Group identifier
+        platform: Platform name
+        operation: Operation name
+
+    Returns:
+        Decorated function
+    """
+
+    def decorator(func):
+        async def wrapper(*args, **kwargs):
+            with TraceContext(
+                group_id=group_id,
+                platform=platform,
+                operation=operation or func.__name__,
+            ):
+                return await func(*args, **kwargs)
+
+        return wrapper
+
+    return decorator