Merge branch 'main' into gl/fix/task-planning-order

lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py

@@ -71,6 +71,7 @@ from crewai.events.types.reasoning_events import (
     AgentReasoningFailedEvent,
     AgentReasoningStartedEvent,
 )
+from crewai.events.types.system_events import SignalEvent, on_signal
 from crewai.events.types.task_events import (
     TaskCompletedEvent,
     TaskFailedEvent,
@@ -159,6 +160,7 @@ class TraceCollectionListener(BaseEventListener):
         self._register_flow_event_handlers(crewai_event_bus)
         self._register_context_event_handlers(crewai_event_bus)
         self._register_action_event_handlers(crewai_event_bus)
+        self._register_system_event_handlers(crewai_event_bus)
 
         self._listeners_setup = True
 
@@ -458,6 +460,15 @@ class TraceCollectionListener(BaseEventListener):
         ) -> None:
             self._handle_action_event("knowledge_query_failed", source, event)
 
+    def _register_system_event_handlers(self, event_bus: CrewAIEventsBus) -> None:
+        """Register handlers for system signal events (SIGTERM, SIGINT, etc.)."""
+
+        @on_signal
+        def handle_signal(source: Any, event: SignalEvent) -> None:
+            """Flush trace batch on system signals to prevent data loss."""
+            if self.batch_manager.is_batch_initialized():
+                self.batch_manager.finalize_batch()
+
     def _initialize_crew_batch(self, source: Any, event: Any) -> None:
         """Initialize trace batch.
 

lib/crewai/src/crewai/events/types/system_events.py

@@ -0,0 +1,102 @@
+"""System signal event types for CrewAI.
+
+This module contains event types for system-level signals like SIGTERM,
+allowing listeners to perform cleanup operations before process termination.
+"""
+
+from collections.abc import Callable
+from enum import IntEnum
+import signal
+from typing import Annotated, Literal, TypeVar
+
+from pydantic import Field, TypeAdapter
+
+from crewai.events.base_events import BaseEvent
+
+
+class SignalType(IntEnum):
+    """Enumeration of supported system signals."""
+
+    SIGTERM = signal.SIGTERM
+    SIGINT = signal.SIGINT
+    SIGHUP = signal.SIGHUP
+    SIGTSTP = signal.SIGTSTP
+    SIGCONT = signal.SIGCONT
+
+
+class SigTermEvent(BaseEvent):
+    """Event emitted when SIGTERM is received."""
+
+    type: Literal["SIGTERM"] = "SIGTERM"
+    signal_number: SignalType = SignalType.SIGTERM
+    reason: str | None = None
+
+
+class SigIntEvent(BaseEvent):
+    """Event emitted when SIGINT is received."""
+
+    type: Literal["SIGINT"] = "SIGINT"
+    signal_number: SignalType = SignalType.SIGINT
+    reason: str | None = None
+
+
+class SigHupEvent(BaseEvent):
+    """Event emitted when SIGHUP is received."""
+
+    type: Literal["SIGHUP"] = "SIGHUP"
+    signal_number: SignalType = SignalType.SIGHUP
+    reason: str | None = None
+
+
+class SigTStpEvent(BaseEvent):
+    """Event emitted when SIGTSTP is received.
+
+    Note: SIGSTOP cannot be caught - it immediately suspends the process.
+    """
+
+    type: Literal["SIGTSTP"] = "SIGTSTP"
+    signal_number: SignalType = SignalType.SIGTSTP
+    reason: str | None = None
+
+
+class SigContEvent(BaseEvent):
+    """Event emitted when SIGCONT is received."""
+
+    type: Literal["SIGCONT"] = "SIGCONT"
+    signal_number: SignalType = SignalType.SIGCONT
+    reason: str | None = None
+
+
+SignalEvent = Annotated[
+    SigTermEvent | SigIntEvent | SigHupEvent | SigTStpEvent | SigContEvent,
+    Field(discriminator="type"),
+]
+
+signal_event_adapter: TypeAdapter[SignalEvent] = TypeAdapter(SignalEvent)
+
+SIGNAL_EVENT_TYPES: tuple[type[BaseEvent], ...] = (
+    SigTermEvent,
+    SigIntEvent,
+    SigHupEvent,
+    SigTStpEvent,
+    SigContEvent,
+)
+
+
+T = TypeVar("T", bound=Callable[[object, SignalEvent], None])
+
+
+def on_signal(func: T) -> T:
+    """Decorator to register a handler for all signal events.
+
+    Args:
+        func: Handler function that receives (source, event) arguments.
+
+    Returns:
+        The original function, registered for all signal event types.
+    """
+    from crewai.events.event_bus import crewai_event_bus
+
+    for event_type in SIGNAL_EVENT_TYPES:
+        crewai_event_bus.on(event_type)(func)
+    return func

lib/crewai/src/crewai/telemetry/telemetry.py

@@ -9,12 +9,14 @@ data is collected. Users can opt-in to share more complete data using the
 from __future__ import annotations
 
 import asyncio
+import atexit
 from collections.abc import Callable
 from importlib.metadata import version
 import json
 import logging
 import os
 import platform
+import signal
 import threading
 from typing import TYPE_CHECKING, Any
 
@@ -31,6 +33,14 @@ from opentelemetry.sdk.trace.export import (
 from opentelemetry.trace import Span
 from typing_extensions import Self
 
+from crewai.events.event_bus import crewai_event_bus
+from crewai.events.types.system_events import (
+    SigContEvent,
+    SigHupEvent,
+    SigIntEvent,
+    SigTStpEvent,
+    SigTermEvent,
+)
 from crewai.telemetry.constants import (
     CREWAI_TELEMETRY_BASE_URL,
     CREWAI_TELEMETRY_SERVICE_NAME,
@@ -121,6 +131,7 @@ class Telemetry:
             )
 
             self.provider.add_span_processor(processor)
+            self._register_shutdown_handlers()
             self.ready = True
         except Exception as e:
             if isinstance(
@@ -155,6 +166,71 @@ class Telemetry:
                 self.ready = False
                 self.trace_set = False
 
+    def _register_shutdown_handlers(self) -> None:
+        """Register handlers for graceful shutdown on process exit and signals."""
+        atexit.register(self._shutdown)
+
+        self._original_handlers: dict[int, Any] = {}
+
+        self._register_signal_handler(signal.SIGTERM, SigTermEvent, shutdown=True)
+        self._register_signal_handler(signal.SIGINT, SigIntEvent, shutdown=True)
+        self._register_signal_handler(signal.SIGHUP, SigHupEvent, shutdown=False)
+        self._register_signal_handler(signal.SIGTSTP, SigTStpEvent, shutdown=False)
+        self._register_signal_handler(signal.SIGCONT, SigContEvent, shutdown=False)
+
+    def _register_signal_handler(
+        self,
+        sig: signal.Signals,
+        event_class: type,
+        shutdown: bool = False,
+    ) -> None:
+        """Register a signal handler that emits an event.
+
+        Args:
+            sig: The signal to handle.
+            event_class: The event class to instantiate and emit.
+            shutdown: Whether to trigger shutdown on this signal.
+        """
+        try:
+            original_handler = signal.getsignal(sig)
+            self._original_handlers[sig] = original_handler
+
+            def handler(signum: int, frame: Any) -> None:
+                crewai_event_bus.emit(self, event_class())
+
+                if shutdown:
+                    self._shutdown()
+
+                if original_handler not in (signal.SIG_DFL, signal.SIG_IGN, None):
+                    if callable(original_handler):
+                        original_handler(signum, frame)
+                elif shutdown:
+                    raise SystemExit(0)
+
+            signal.signal(sig, handler)
+        except ValueError as e:
+            logger.warning(
+                f"Cannot register {sig.name} handler: not running in main thread",
+                exc_info=e,
+            )
+        except OSError as e:
+            logger.warning(f"Cannot register {sig.name} handler: {e}", exc_info=e)
+
+    def _shutdown(self) -> None:
+        """Flush and shutdown the telemetry provider on process exit.
+
+        Uses a short timeout to avoid blocking process shutdown.
+        """
+        if not self.ready:
+            return
+
+        try:
+            self.provider.force_flush(timeout_millis=5000)
+            self.provider.shutdown()
+            self.ready = False
+        except Exception as e:
+            logger.debug(f"Telemetry shutdown failed: {e}")
+
     def _safe_telemetry_operation(
         self, operation: Callable[[], Span | None]
     ) -> Span | None: