Lorenze/ephemeral trace ask (#3530)

* feat(tracing): implement first-time trace handling and improve event management

- Added FirstTimeTraceHandler for managing first-time user trace collection and display.
- Enhanced TraceBatchManager to support ephemeral trace URLs and improved event buffering.
- Updated TraceCollectionListener to utilize the new FirstTimeTraceHandler.
- Refactored type annotations across multiple files for consistency and clarity.
- Improved error handling and logging for trace-related operations.
- Introduced utility functions for trace viewing prompts and first execution checks.

* brought back crew finalize batch events

* refactor(trace): move instance variables to __init__ in TraceBatchManager

- Refactored TraceBatchManager to initialize instance variables in the constructor instead of as class variables.
- Improved clarity and encapsulation of the class state.

* fix(tracing): improve error handling in user data loading and saving

- Enhanced error handling in _load_user_data and _save_user_data functions to log warnings for JSON decoding and file access issues.
- Updated documentation for trace usage to clarify the addition of tracing parameters in Crew and Flow initialization.
- Refined state management in Flow class to ensure proper handling of state IDs when persistence is enabled.

* add some tests

* fix test

* fix tests

* refactor(tracing): enhance user input handling for trace viewing

- Replaced signal-based timeout handling with threading for user input in prompt_user_for_trace_viewing function.
- Improved user experience by allowing a configurable timeout for viewing execution traces.
- Updated tests to mock threading behavior and verify timeout handling correctly.

* fix(tracing): improve machine ID retrieval with error handling

- Added error handling to the _get_machine_id function to log warnings when retrieving the machine ID fails.
- Ensured that the function continues to provide a stable, privacy-preserving machine fingerprint even in case of errors.

* refactor(flow): streamline state ID assignment in Flow class

- Replaced direct attribute assignment with setattr for improved flexibility in handling state IDs.
- Enhanced code readability by simplifying the logic for setting the state ID when persistence is enabled.

Author: lorenzejay

src/crewai/crew.py

@@ -37,6 +37,7 @@
 )
 from crewai.events.listeners.tracing.utils import (
     is_tracing_enabled,
+    should_auto_collect_first_time_traces,
 )
 from crewai.events.types.crew_events import (
     CrewKickoffCompletedEvent,
@@ -89,8 +90,8 @@ class Crew(FlowTrackable, BaseModel):
     tasks they should perform.
 
     Attributes:
-        tasks: List of tasks assigned to the crew.
-        agents: List of agents part of this crew.
+        tasks: list of tasks assigned to the crew.
+        agents: list of agents part of this crew.
         manager_llm: The language model that will run manager agent.
         manager_agent: Custom agent that will be used as manager.
         memory: Whether the crew should use memory to store memories of it's
@@ -238,11 +239,11 @@ class Crew(FlowTrackable, BaseModel):
     )
     task_execution_output_json_files: list[str] | None = Field(
         default=None,
-        description="List of file paths for task execution JSON files.",
+        description="list of file paths for task execution JSON files.",
     )
     execution_logs: list[dict[str, Any]] = Field(
         default=[],
-        description="List of execution logs for tasks",
+        description="list of execution logs for tasks",
     )
     knowledge_sources: list[BaseKnowledgeSource] | None = Field(
         default=None,
@@ -296,12 +297,16 @@ def check_config_type(cls, v: Json | dict[str, Any]) -> Json | dict[str, Any]:
 
     @model_validator(mode="after")
     def set_private_attrs(self) -> "Crew":
-        """Set private attributes."""
+        """set private attributes."""
 
         self._cache_handler = CacheHandler()
         event_listener = EventListener()
 
-        if is_tracing_enabled() or self.tracing:
+        if (
+            is_tracing_enabled()
+            or self.tracing
+            or should_auto_collect_first_time_traces()
+        ):
             trace_listener = TraceCollectionListener()
             trace_listener.setup_listeners(crewai_event_bus)
         event_listener.verbose = self.verbose

src/crewai/events/listeners/tracing/first_time_trace_handler.py

@@ -0,0 +1,174 @@
+import logging
+import uuid
+
+from rich.console import Console
+from rich.panel import Panel
+
+from crewai.events.listeners.tracing.trace_batch_manager import TraceBatchManager
+from crewai.events.listeners.tracing.utils import (
+    mark_first_execution_completed,
+    prompt_user_for_trace_viewing,
+    should_auto_collect_first_time_traces,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class FirstTimeTraceHandler:
+    """Handles the first-time user trace collection and display flow."""
+
+    def __init__(self):
+        self.is_first_time: bool = False
+        self.collected_events: bool = False
+        self.trace_batch_id: str | None = None
+        self.ephemeral_url: str | None = None
+        self.batch_manager: TraceBatchManager | None = None
+
+    def initialize_for_first_time_user(self) -> bool:
+        """Check if this is first time and initialize collection."""
+        self.is_first_time = should_auto_collect_first_time_traces()
+        return self.is_first_time
+
+    def set_batch_manager(self, batch_manager: TraceBatchManager):
+        """Set reference to batch manager for sending events."""
+        self.batch_manager = batch_manager
+
+    def mark_events_collected(self):
+        """Mark that events have been collected during execution."""
+        self.collected_events = True
+
+    def handle_execution_completion(self):
+        """Handle the completion flow as shown in your diagram."""
+        if not self.is_first_time or not self.collected_events:
+            return
+
+        try:
+            user_wants_traces = prompt_user_for_trace_viewing(timeout_seconds=20)
+
+            if user_wants_traces:
+                self._initialize_backend_and_send_events()
+
+                if self.ephemeral_url:
+                    self._display_ephemeral_trace_link()
+
+            mark_first_execution_completed()
+
+        except Exception as e:
+            self._gracefully_fail(f"Error in trace handling: {e}")
+            mark_first_execution_completed()
+
+    def _initialize_backend_and_send_events(self):
+        """Initialize backend batch and send collected events."""
+        if not self.batch_manager:
+            return
+
+        try:
+            if not self.batch_manager.backend_initialized:
+                original_metadata = (
+                    self.batch_manager.current_batch.execution_metadata
+                    if self.batch_manager.current_batch
+                    else {}
+                )
+
+                user_context = {
+                    "privacy_level": "standard",
+                    "user_id": "first_time_user",
+                    "session_id": str(uuid.uuid4()),
+                    "trace_id": self.batch_manager.trace_batch_id,
+                }
+
+                execution_metadata = {
+                    "execution_type": original_metadata.get("execution_type", "crew"),
+                    "crew_name": original_metadata.get(
+                        "crew_name", "First Time Execution"
+                    ),
+                    "flow_name": original_metadata.get("flow_name"),
+                    "agent_count": original_metadata.get("agent_count", 1),
+                    "task_count": original_metadata.get("task_count", 1),
+                    "crewai_version": original_metadata.get("crewai_version"),
+                }
+
+                self.batch_manager._initialize_backend_batch(
+                    user_context=user_context,
+                    execution_metadata=execution_metadata,
+                    use_ephemeral=True,
+                )
+                self.batch_manager.backend_initialized = True
+
+            if self.batch_manager.event_buffer:
+                self.batch_manager._send_events_to_backend()
+
+            self.batch_manager.finalize_batch()
+            self.ephemeral_url = self.batch_manager.ephemeral_trace_url
+
+            if not self.ephemeral_url:
+                self._show_local_trace_message()
+
+        except Exception as e:
+            self._gracefully_fail(f"Backend initialization failed: {e}")
+
+    def _display_ephemeral_trace_link(self):
+        """Display the ephemeral trace link to the user."""
+        console = Console()
+
+        panel_content = f"""
+🎉 Your First CrewAI Execution Trace is Ready!
+
+View your execution details here:
+{self.ephemeral_url}
+
+This trace shows:
+• Agent decisions and interactions
+• Task execution timeline
+• Tool usage and results
+• LLM calls and responses
+
+To use traces add tracing=True to your Crew(tracing=True) / Flow(tracing=True)
+
+📝 Note: This link will expire in 24 hours.
+        """.strip()
+
+        panel = Panel(
+            panel_content,
+            title="🔍 Execution Trace Generated",
+            border_style="bright_green",
+            padding=(1, 2),
+        )
+
+        console.print("\n")
+        console.print(panel)
+        console.print()
+
+    def _gracefully_fail(self, error_message: str):
+        """Handle errors gracefully without disrupting user experience."""
+        console = Console()
+        console.print(f"[yellow]Note: {error_message}[/yellow]")
+
+        logger.debug(f"First-time trace error: {error_message}")
+
+    def _show_local_trace_message(self):
+        """Show message when traces were collected locally but couldn't be uploaded."""
+        console = Console()
+
+        panel_content = f"""
+📊 Your execution traces were collected locally!
+
+Unfortunately, we couldn't upload them to the server right now, but here's what we captured:
+• {len(self.batch_manager.event_buffer)} trace events
+• Execution duration: {self.batch_manager.calculate_duration("execution")}ms
+• Batch ID: {self.batch_manager.trace_batch_id}
+
+The traces include agent decisions, task execution, and tool usage.
+Try running with CREWAI_TRACING_ENABLED=true next time for persistent traces.
+    """.strip()
+
+        panel = Panel(
+            panel_content,
+            title="🔍 Local Traces Collected",
+            border_style="yellow",
+            padding=(1, 2),
+        )
+
+        console.print("\n")
+        console.print(panel)
+        console.print()