Add Conversation Analytics with HF Dataset persistence

- Add conversation_analytics.py module with metrics tracking
- Track message count, tool usage, response times, errors
- Add analytics dashboard to Gradio UI with summary view
- Add API endpoints for analytics stats and history
- Support HF Dataset sync for conversation persistence
- Integrate analytics tracking into chat_with_brain function

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

.openclaw/agents/conversation_analytics.py

@@ -0,0 +1,500 @@
+#!/usr/bin/env python3
+"""
+Conversation Analytics Module for HuggingClaw Cain
+
+Tracks conversation metrics and provides analytics including:
+- Message count and statistics
+- Tool usage tracking
+- Response time metrics
+- HF Dataset persistence
+- Conversation summaries
+"""
+
+import json
+import os
+import time
+from typing import Dict, List, Any, Optional
+from pathlib import Path
+from datetime import datetime, timedelta
+from dataclasses import dataclass, asdict
+from collections import defaultdict
+import threading
+
+
+# Analytics storage paths
+OPENCLAW_AGENTS_DIR = Path(__file__).parent
+ANALYTICS_DIR = OPENCLAW_AGENTS_DIR / "analytics"
+ANALYTICS_FILE = ANALYTICS_DIR / "conversation_analytics.json"
+CONVERSATION_HISTORY_FILE = ANALYTICS_DIR / "conversation_history.jsonl"
+
+
+@dataclass
+class ConversationMetrics:
+    """Metrics for a single conversation"""
+    conversation_id: str
+    timestamp: str
+    message_count: int
+    tool_calls: Dict[str, int]
+    avg_response_time: float
+    total_duration: float
+    status: str
+    error_count: int
+
+
+@dataclass
+class MessageRecord:
+    """Record of a single message"""
+    conversation_id: str
+    timestamp: str
+    role: str  # "user" or "assistant"
+    content: str
+    tools_used: List[str]
+    response_time: float
+    success: bool
+
+
+class ConversationAnalytics:
+    """
+    Tracks conversation analytics with HF Dataset persistence support.
+    Thread-safe for concurrent access.
+    """
+
+    def __init__(self, hf_dataset_repo: Optional[str] = None):
+        """
+        Initialize conversation analytics.
+
+        Args:
+            hf_dataset_repo: HF Dataset repo ID for persistence (e.g., "tao-shen/HuggingClaw-Cain-data")
+        """
+        self.hf_dataset_repo = hf_dataset_repo or os.getenv("OPENCLAW_DATASET_REPO")
+        self._lock = threading.Lock()
+
+        # Analytics data
+        self._conversations: Dict[str, ConversationMetrics] = {}
+        self._message_records: List[MessageRecord] = []
+        self._tool_usage: Dict[str, int] = defaultdict(int)
+        self._daily_stats: Dict[str, Dict[str, Any]] = {}
+
+        # Current conversation state
+        self._current_conversation_id: Optional[str] = None
+        self._current_conversation_start: Optional[float] = None
+        self._current_message_count: int = 0
+        self._current_tools: Dict[str, int] = defaultdict(int)
+        self._current_errors: int = 0
+
+        # Load existing analytics
+        self._ensure_dirs()
+        self._load_analytics()
+
+    def _ensure_dirs(self):
+        """Ensure analytics directories exist."""
+        ANALYTICS_DIR.mkdir(parents=True, exist_ok=True)
+
+    def _load_analytics(self):
+        """Load existing analytics from disk."""
+        if ANALYTICS_FILE.exists():
+            try:
+                with open(ANALYTICS_FILE, 'r') as f:
+                    data = json.load(f)
+                    for conv_id, conv_data in data.get("conversations", {}).items():
+                        self._conversations[conv_id] = ConversationMetrics(**conv_data)
+                    self._tool_usage = defaultdict(int, data.get("tool_usage", {}))
+                    self._daily_stats = data.get("daily_stats", {})
+            except Exception as e:
+                print(f"[ConversationAnalytics] Failed to load analytics: {e}")
+
+        # Load conversation history
+        if CONVERSATION_HISTORY_FILE.exists():
+            try:
+                with open(CONVERSATION_HISTORY_FILE, 'r') as f:
+                    for line in f:
+                        if line.strip():
+                            data = json.loads(line)
+                            self._message_records.append(MessageRecord(**data))
+            except Exception as e:
+                print(f"[ConversationAnalytics] Failed to load history: {e}")
+
+    def _save_analytics(self):
+        """Save analytics to disk."""
+        try:
+            data = {
+                "last_updated": datetime.utcnow().isoformat() + "Z",
+                "conversations": {
+                    k: asdict(v) for k, v in self._conversations.items()
+                },
+                "tool_usage": dict(self._tool_usage),
+                "daily_stats": self._daily_stats,
+                "total_conversations": len(self._conversations),
+                "total_messages": len(self._message_records)
+            }
+            with open(ANALYTICS_FILE, 'w') as f:
+                json.dump(data, f, indent=2)
+        except Exception as e:
+            print(f"[ConversationAnalytics] Failed to save analytics: {e}")
+
+    def _save_message_record(self, record: MessageRecord):
+        """Append a message record to history file."""
+        try:
+            with open(CONVERSATION_HISTORY_FILE, 'a') as f:
+                f.write(json.dumps(asdict(record)) + "\n")
+        except Exception as e:
+            print(f"[ConversationAnalytics] Failed to save message record: {e}")
+
+    def start_conversation(self, conversation_id: Optional[str] = None) -> str:
+        """
+        Start a new conversation tracking session.
+
+        Args:
+            conversation_id: Optional conversation ID (auto-generated if None)
+
+        Returns:
+            The conversation ID
+        """
+        with self._lock:
+            # End current conversation if any
+            if self._current_conversation_id:
+                self.end_conversation(status="interrupted")
+
+            # Generate new conversation ID
+            if conversation_id is None:
+                conversation_id = f"conv_{int(time.time() * 1000)}"
+
+            self._current_conversation_id = conversation_id
+            self._current_conversation_start = time.time()
+            self._current_message_count = 0
+            self._current_tools = defaultdict(int)
+            self._current_errors = 0
+
+            return conversation_id
+
+    def track_message(
+        self,
+        role: str,
+        content: str,
+        tools_used: List[str] = None,
+        response_time: float = 0.0,
+        success: bool = True
+    ):
+        """
+        Track a message in the current conversation.
+
+        Args:
+            role: Message role ("user" or "assistant")
+            content: Message content
+            tools_used: List of tools used in response
+            response_time: Response time in seconds
+            success: Whether the message was processed successfully
+        """
+        with self._lock:
+            if self._current_conversation_id is None:
+                self.start_conversation()
+
+            # Update current conversation stats
+            self._current_message_count += 1
+            if not success:
+                self._current_errors += 1
+
+            for tool in (tools_used or []):
+                self._current_tools[tool] += 1
+                self._tool_usage[tool] += 1
+
+            # Create message record
+            record = MessageRecord(
+                conversation_id=self._current_conversation_id,
+                timestamp=datetime.utcnow().isoformat() + "Z",
+                role=role,
+                content=content[:1000],  # Truncate long messages
+                tools_used=tools_used or [],
+                response_time=response_time,
+                success=success
+            )
+            self._message_records.append(record)
+
+            # Save to history file
+            self._save_message_record(record)
+
+    def end_conversation(self, status: str = "completed") -> Optional[ConversationMetrics]:
+        """
+        End the current conversation and calculate metrics.
+
+        Args:
+            status: Conversation completion status ("completed", "error", "interrupted")
+
+        Returns:
+            The conversation metrics, or None if no conversation was active
+        """
+        with self._lock:
+            if self._current_conversation_id is None:
+                return None
+
+            # Calculate metrics
+            duration = time.time() - self._current_conversation_start
+            avg_response_time = 0.0
+
+            if self._message_records:
+                response_times = [
+                    m.response_time for m in self._message_records
+                    if m.role == "assistant" and m.response_time > 0
+                ]
+                if response_times:
+                    avg_response_time = sum(response_times) / len(response_times)
+
+            metrics = ConversationMetrics(
+                conversation_id=self._current_conversation_id,
+                timestamp=datetime.utcnow().isoformat() + "Z",
+                message_count=self._current_message_count,
+                tool_calls=dict(self._current_tools),
+                avg_response_time=avg_response_time,
+                total_duration=duration,
+                status=status,
+                error_count=self._current_errors
+            )
+
+            # Store metrics
+            self._conversations[self._current_conversation_id] = metrics
+
+            # Update daily stats
+            today = datetime.utcnow().strftime("%Y-%m-%d")
+            if today not in self._daily_stats:
+                self._daily_stats[today] = {
+                    "conversation_count": 0,
+                    "message_count": 0,
+                    "total_duration": 0.0,
+                    "errors": 0
+                }
+
+            self._daily_stats[today]["conversation_count"] += 1
+            self._daily_stats[today]["message_count"] += self._current_message_count
+            self._daily_stats[today]["total_duration"] += duration
+            self._daily_stats[today]["errors"] += self._current_errors
+
+            # Reset current conversation
+            self._current_conversation_id = None
+            self._current_conversation_start = None
+            self._current_message_count = 0
+            self._current_tools = defaultdict(int)
+            self._current_errors = 0
+
+            # Save analytics
+            self._save_analytics()
+
+            return metrics
+
+    def get_stats(self) -> Dict[str, Any]:
+        """
+        Get current analytics statistics.
+
+        Returns:
+            Dictionary containing analytics statistics
+        """
+        with self._lock:
+            total_conversations = len(self._conversations)
+            total_messages = len(self._message_records)
+            total_errors = sum(c.error_count for c in self._conversations.values())
+
+            avg_response_time = 0.0
+            if self._conversations:
+                response_times = [c.avg_response_time for c in self._conversations.values() if c.avg_response_time > 0]
+                if response_times:
+                    avg_response_time = sum(response_times) / len(response_times)
+
+            # Get recent conversations (last 10)
+            recent_conversations = sorted(
+                self._conversations.values(),
+                key=lambda c: c.timestamp,
+                reverse=True
+            )[:10]
+
+            return {
+                "total_conversations": total_conversations,
+                "total_messages": total_messages,
+                "total_errors": total_errors,
+                "avg_response_time": round(avg_response_time, 3),
+                "tool_usage": dict(sorted(self._tool_usage.items(), key=lambda x: x[1], reverse=True)),
+                "daily_stats": dict(sorted(self._daily_stats.items(), reverse=True)),
+                "recent_conversations": [asdict(c) for c in recent_conversations],
+                "current_conversation": self._current_conversation_id
+            }
+
+    def get_conversation_history(self, limit: int = 50) -> List[Dict[str, Any]]:
+        """
+        Get recent conversation history.
+
+        Args:
+            limit: Maximum number of message records to return
+
+        Returns:
+            List of message record dictionaries
+        """
+        with self._lock:
+            recent = self._message_records[-limit:]
+            return [asdict(r) for r in recent]
+
+    def get_summary(self) -> str:
+        """
+        Get a formatted summary of analytics.
+
+        Returns:
+            Formatted markdown summary
+        """
+        stats = self.get_stats()
+
+        lines = [
+            "## 📊 Conversation Analytics",
+            "",
+            f"**Total Conversations:** {stats['total_conversations']}",
+            f"**Total Messages:** {stats['total_messages']}",
+            f"**Total Errors:** {stats['total_errors']}",
+            f"**Avg Response Time:** {stats['avg_response_time']}s",
+            "",
+        ]
+
+        if stats['tool_usage']:
+            lines.append("### Tool Usage")
+            lines.append("")
+            for tool, count in list(stats['tool_usage'].items())[:10]:
+                lines.append(f"- `{tool}`: {count} uses")
+            lines.append("")
+
+        if stats['daily_stats']:
+            lines.append("### Daily Stats (Last 7 Days)")
+            lines.append("")
+            for date, day_stats in list(stats['daily_stats'].items())[:7]:
+                lines.append(f"- **{date}:** {day_stats['conversation_count']} convs, {day_stats['message_count']} msgs")
+            lines.append("")
+
+        if stats['recent_conversations']:
+            lines.append("### Recent Conversations")
+            lines.append("")
+            for conv in stats['recent_conversations'][:5]:
+                status_emoji = {
+                    "completed": "✅",
+                    "error": "❌",
+                    "interrupted": "⏸️"
+                }.get(conv['status'], "❓")
+                lines.append(
+                    f"- {status_emoji} `{conv['conversation_id']}`: "
+                    f"{conv['message_count']} msgs, "
+                    f"{conv['avg_response_time']:.3f}s avg"
+                )
+
+        return "\n".join(lines)
+
+    def sync_to_dataset(self) -> Dict[str, Any]:
+        """
+        Sync analytics data to HF Dataset.
+
+        Returns:
+            Result dict with success status
+        """
+        if not self.hf_dataset_repo:
+            return {
+                "success": False,
+                "error": "No HF dataset repository configured"
+            }
+
+        try:
+            from huggingface_hub import HfApi
+
+            api = HfApi()
+            timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
+
+            # Upload analytics file
+            analytics_upload = api.upload_file(
+                path_or_fileobj=str(ANALYTICS_FILE),
+                path_in_repo=f"analytics/conversation_analytics_{timestamp}.json",
+                repo_id=self.hf_dataset_repo,
+                repo_type="dataset"
+            )
+
+            # Upload conversation history
+            history_upload = api.upload_file(
+                path_or_fileobj=str(CONVERSATION_HISTORY_FILE),
+                path_in_repo=f"analytics/conversation_history_{timestamp}.jsonl",
+                repo_id=self.hf_dataset_repo,
+                repo_type="dataset"
+            )
+
+            return {
+                "success": True,
+                "analytics_file": analytics_upload,
+                "history_file": history_upload,
+                "timestamp": timestamp,
+                "repo_id": self.hf_dataset_repo
+            }
+
+        except ImportError:
+            return {
+                "success": False,
+                "error": "huggingface_hub not available"
+            }
+        except Exception as e:
+            return {
+                "success": False,
+                "error": str(e)
+            }
+
+    def reset(self):
+        """Reset all analytics data."""
+        with self._lock:
+            self._conversations.clear()
+            self._message_records.clear()
+            self._tool_usage = defaultdict(int)
+            self._daily_stats.clear()
+            self._current_conversation_id = None
+            self._current_conversation_start = None
+            self._current_message_count = 0
+            self._current_tools = defaultdict(int)
+            self._current_errors = 0
+            self._save_analytics()
+
+
+# Global instance
+_global_analytics: Optional[ConversationAnalytics] = None
+
+
+def get_analytics(hf_dataset_repo: Optional[str] = None) -> ConversationAnalytics:
+    """
+    Get the global conversation analytics instance.
+
+    Args:
+        hf_dataset_repo: Optional HF Dataset repo ID
+
+    Returns:
+        ConversationAnalytics instance
+    """
+    global _global_analytics
+    if _global_analytics is None:
+        _global_analytics = ConversationAnalytics(hf_dataset_repo)
+    return _global_analytics
+
+
+if __name__ == "__main__":
+    # Test the analytics module
+    print("=== Conversation Analytics Test ===\n")
+
+    analytics = ConversationAnalytics()
+
+    # Start a conversation
+    conv_id = analytics.start_conversation("test_conv_001")
+    print(f"Started conversation: {conv_id}")
+
+    # Track some messages
+    analytics.track_message("user", "Hello, how are you?", [], 0, True)
+    time.sleep(0.1)
+    analytics.track_message("assistant", "I'm doing well!", ["conversation_process"], 0.1, True)
+    time.sleep(0.1)
+    analytics.track_message("user", "What can you do?", [], 0, True)
+    time.sleep(0.15)
+    analytics.track_message("assistant", "I can help with many things.", ["conversation_process", "memory_read"], 0.15, True)
+
+    # End conversation
+    metrics = analytics.end_conversation("completed")
+    print(f"\nConversation metrics: {metrics}")
+
+    # Get stats
+    stats = analytics.get_stats()
+    print(f"\nStats: {json.dumps(stats, indent=2, default=str)}")
+
+    # Get summary
+    print(f"\nSummary:\n{analytics.get_summary()}")

app.py

@@ -15,6 +15,7 @@ import sys
 import json
 import asyncio
 import threading
+import time
 from pathlib import Path
 from datetime import datetime
 from typing import Dict, Any, List, Optional, Tuple
@@ -200,6 +201,18 @@ except ImportError as e:
     print(f"[Agent Office] Warning: rbac not available - {e}")
     RBAC_AVAILABLE = False
 
+# Import conversation analytics
+try:
+    from conversation_analytics import get_analytics, ConversationAnalytics
+    ANALYTICS_AVAILABLE = True
+    # Initialize analytics with HF dataset repo from env
+    hf_dataset = os.getenv("OPENCLAW_DATASET_REPO", "tao-shen/HuggingClaw-Cain-data")
+    analytics = get_analytics(hf_dataset)
+except ImportError as e:
+    print(f"[Agent Office] Warning: conversation_analytics not available - {e}")
+    ANALYTICS_AVAILABLE = False
+    analytics = None
+
 # ========== Data Loading Functions ==========
 
 def load_cain_status() -> Dict[str, Any]:
@@ -390,6 +403,10 @@ def chat_with_brain(
     if not message.strip():
         return history, "Please enter a message."
 
+    # Track start time for analytics
+    start_time = time.time()
+    tools_used = []
+
     # Emit thinking event
     emit_agent_thought(
         ThoughtEventType.THINKING,
@@ -433,6 +450,7 @@ def chat_with_brain(
             )
 
             result = brain_instance.execute_tool("conversation_process", message)
+            tools_used = ["conversation_process"]
 
             if result.get("success"):
                 response = result.get("response", f"Processed: {message}")
@@ -468,6 +486,13 @@ def chat_with_brain(
             )
 
         history.append((message, response))
+
+        # Track analytics for user and assistant messages
+        response_time = time.time() - start_time
+        if ANALYTICS_AVAILABLE and analytics:
+            analytics.track_message("user", message, [], 0, True)
+            analytics.track_message("assistant", response, tools_used, response_time, "Error" not in response)
+
         return history, status_msg
 
     except Exception as e:
@@ -480,6 +505,12 @@ def chat_with_brain(
             metadata={"exception": str(e), "type": type(e).__name__}
         )
 
+        # Track error in analytics
+        response_time = time.time() - start_time
+        if ANALYTICS_AVAILABLE and analytics:
+            analytics.track_message("user", message, [], 0, True)
+            analytics.track_message("assistant", error_response, tools_used, response_time, False)
+
         return history, f"❌ Exception: {str(e)}"
 
 
@@ -512,6 +543,55 @@ def get_brain_info() -> str:
         return f"Error getting brain info: {str(e)}"
 
 
+# ========== Analytics Functions ==========
+
+def get_analytics_summary() -> str:
+    """Get the conversation analytics summary"""
+    if not ANALYTICS_AVAILABLE or analytics is None:
+        return "### 📊 Analytics\n\nAnalytics module not available."
+    return analytics.get_summary()
+
+
+def refresh_analytics() -> str:
+    """Refresh the analytics display"""
+    return get_analytics_summary()
+
+
+def sync_analytics_to_dataset() -> str:
+    """Sync analytics to HF Dataset"""
+    if not ANALYTICS_AVAILABLE or analytics is None:
+        return "❌ Analytics not available"
+
+    result = analytics.sync_to_dataset()
+    if result.get("success"):
+        return f"✅ Synced to dataset `{result['repo_id']}` at `{result['timestamp']}`"
+    else:
+        return f"❌ Sync failed: {result.get('error', 'Unknown error')}"
+
+
+def reset_analytics_data() -> str:
+    """Reset all analytics data"""
+    if not ANALYTICS_AVAILABLE or analytics is None:
+        return "❌ Analytics not available"
+
+    analytics.reset()
+    return "✅ Analytics data has been reset"
+
+
+def get_analytics_stats_json() -> Dict[str, Any]:
+    """Get analytics stats as JSON for API"""
+    if not ANALYTICS_AVAILABLE or analytics is None:
+        return {"error": "Analytics not available"}
+    return analytics.get_stats()
+
+
+def get_conversation_history_json(limit: int = 50) -> List[Dict[str, Any]]:
+    """Get conversation history as JSON for API"""
+    if not ANALYTICS_AVAILABLE or analytics is None:
+        return [{"error": "Analytics not available"}]
+    return analytics.get_conversation_history(limit)
+
+
 # ========== Refresh Functions ==========
 
 def refresh_status() -> str:
@@ -526,19 +606,30 @@ def refresh_agent_directory() -> Tuple[List[List[str]], str]:
     return format_agent_directory(registry), f"Last updated: {datetime.utcnow().isoformat()}Z"
 
 
-def refresh_all() -> Tuple[str, List[List[str]], str, str]:
+def refresh_all() -> Tuple:
     """Refresh all dynamic components"""
     status = load_cain_status()
     registry = load_agent_registry()
     brain_info = get_brain_info()
     timestamp = f"Last updated: {datetime.utcnow().isoformat()}Z"
 
-    return (
-        format_status_header(status),
-        format_agent_directory(registry),
-        brain_info,
-        timestamp
-    )
+    # Include analytics if available
+    if ANALYTICS_AVAILABLE:
+        analytics_summary = get_analytics_summary()
+        return (
+            format_status_header(status),
+            format_agent_directory(registry),
+            brain_info,
+            analytics_summary,
+            timestamp
+        )
+    else:
+        return (
+            format_status_header(status),
+            format_agent_directory(registry),
+            brain_info,
+            timestamp
+        )
 
 
 # ========== Custom CSS ==========
@@ -676,6 +767,24 @@ def create_agent_office():
                         label="Current Brain Status"
                     )
 
+                # ========== Conversation Analytics ==========
+                if ANALYTICS_AVAILABLE:
+                    with gr.Accordion("📊 Conversation Analytics", open=False):
+                        analytics_display = gr.Markdown(
+                            value=get_analytics_summary(),
+                            label="Analytics Summary"
+                        )
+                        refresh_analytics_btn = gr.Button("🔄 Refresh Analytics", size="sm")
+
+                        with gr.Row():
+                            sync_analytics_btn = gr.Button("💾 Sync to Dataset", size="sm")
+                            reset_analytics_btn = gr.Button("🗑️ Reset Analytics", size="sm", variant="stop")
+
+                        analytics_status = gr.Markdown(
+                            value="",
+                            label="Sync Status"
+                        )
+
         # ========== Footer ==========
         gr.Markdown("---")
         with gr.Row():
@@ -727,21 +836,55 @@ def create_agent_office():
                 cain_status_display,
                 agent_directory_table,
                 brain_info_display,
-                last_updated_display
+                analytics_display if ANALYTICS_AVAILABLE else last_updated_display,
+                last_updated_display if ANALYTICS_AVAILABLE else brain_info_display
             ]
         )
 
+        # Analytics handlers (if available)
+        if ANALYTICS_AVAILABLE:
+            refresh_analytics_btn.click(
+                fn=refresh_analytics,
+                inputs=[],
+                outputs=[analytics_display]
+            )
+
+            sync_analytics_btn.click(
+                fn=sync_analytics_to_dataset,
+                inputs=[],
+                outputs=[analytics_status]
+            )
+
+            reset_analytics_btn.click(
+                fn=reset_analytics_data,
+                inputs=[],
+                outputs=[analytics_status]
+            )
+
         # Auto-refresh on load
-        app.load(
-            fn=refresh_all,
-            inputs=[],
-            outputs=[
-                cain_status_display,
-                agent_directory_table,
-                brain_info_display,
-                last_updated_display
-            ]
-        )
+        if ANALYTICS_AVAILABLE:
+            app.load(
+                fn=refresh_all,
+                inputs=[],
+                outputs=[
+                    cain_status_display,
+                    agent_directory_table,
+                    brain_info_display,
+                    analytics_display,
+                    last_updated_display
+                ]
+            )
+        else:
+            app.load(
+                fn=refresh_all,
+                inputs=[],
+                outputs=[
+                    cain_status_display,
+                    agent_directory_table,
+                    brain_info_display,
+                    last_updated_display
+                ]
+            )
 
     return app
 
@@ -783,6 +926,36 @@ def create_agent_office_with_ws():
         """Get agent registry as JSON."""
         return load_agent_registry()
 
+    # ========== Analytics API Endpoints ==========
+    if ANALYTICS_AVAILABLE:
+        @fastapi_app.get("/api/analytics/stats")
+        async def api_analytics_stats():
+            """Get conversation analytics statistics."""
+            return get_analytics_stats_json()
+
+        @fastapi_app.get("/api/analytics/summary")
+        async def api_analytics_summary():
+            """Get analytics summary as markdown."""
+            return {"summary": get_analytics_summary()}
+
+        @fastapi_app.get("/api/analytics/history")
+        async def api_analytics_history(limit: int = 50):
+            """Get conversation history."""
+            return get_conversation_history_json(limit)
+
+        @fastapi_app.post("/api/analytics/sync")
+        async def api_analytics_sync():
+            """Sync analytics to HF Dataset."""
+            result = analytics.sync_to_dataset()
+            return result
+
+        @fastapi_app.post("/api/analytics/reset")
+        async def api_analytics_reset():
+            """Reset analytics data."""
+            if analytics:
+                analytics.reset()
+            return {"status": "reset"}
+
     # Create Gradio app
     gradio_app = create_agent_office()