feat: agent-lifetime heartbeat sender

src/glassflow/client.py

@@ -16,6 +16,7 @@
 
 from . import __version__
 from .config import GlassflowConfig, resolve_config
+from .heartbeat import HeartbeatSender, OpenRootSpanTracker
 from .instrumentation import enable_instrumentations
 from .masking import MaskingSpanExporter
 from .semconv import TRACER_NAME
@@ -50,9 +51,15 @@ class GlassflowClient:
     configuration is available as ``client.config``.
     """
 
-    def __init__(self, provider: TracerProvider, config: GlassflowConfig) -> None:
+    def __init__(
+        self,
+        provider: TracerProvider,
+        config: GlassflowConfig,
+        heartbeat: HeartbeatSender | None = None,
+    ) -> None:
         self._provider = provider
         self.config = config
+        self._heartbeat = heartbeat
         self._is_shutdown = False
 
     def get_tracer(self, name: str = TRACER_NAME) -> trace.Tracer:
@@ -68,8 +75,14 @@ def flush(self, timeout_millis: int = 30_000) -> bool:
         return self._provider.force_flush(timeout_millis)
 
     def shutdown(self) -> None:
-        """Drain pending spans and stop. Releases the global init() slot."""
+        """Drain pending spans and stop. Releases the global init() slot.
+
+        Also stops the heartbeat thread and sends its final ``stopped`` ping,
+        so the backend can tell a clean shutdown from a vanished agent.
+        """
         global _current_client
+        if self._heartbeat is not None:
+            self._heartbeat.stop()
         self._provider.shutdown()
         self._is_shutdown = True
         with _lock:
@@ -89,6 +102,10 @@ def init(
     mask: Callable[[Any], Any] | None = None,
     instruments: Sequence[str] | None = None,
     span_exporter: SpanExporter | None = None,
+    heartbeat: bool | None = None,
+    heartbeat_interval: float | None = None,
+    agent_name: str | None = None,
+    heartbeat_transport: Callable[[dict[str, Any]], None] | None = None,
     set_global: bool = True,
 ) -> GlassflowClient:
     """Initialize the SDK: build a tracer provider that exports OTLP traces.
@@ -112,6 +129,16 @@ def init(
             Instrumentors are process-global, so with ``set_global=False`` they
             are only enabled when ``instruments`` is passed explicitly.
         span_exporter: Override the default OTLP exporter (useful for testing).
+        heartbeat: Enable the agent-lifetime heartbeat thread
+            (``GLASSFLOW_HEARTBEAT``; default off this release). Pings
+            ``<endpoint>/v1/heartbeat`` from init until process exit so the
+            platform can tell a live-but-idle agent from a vanished one.
+        heartbeat_interval: Seconds between pings (default 15, clamped to
+            ``[5, 300]`` — the backend derives staleness from this).
+        agent_name: Identity heartbeats group under; defaults to
+            ``service_name``.
+        heartbeat_transport: Override the heartbeat HTTP transport
+            (useful for testing, like ``span_exporter``).
         set_global: Register the provider as the global OpenTelemetry provider.
     """
     global _current_client
@@ -134,6 +161,10 @@ def init(
             mask=mask,
             instruments=instruments,
             span_exporter=span_exporter,
+            heartbeat=heartbeat,
+            heartbeat_interval=heartbeat_interval,
+            agent_name=agent_name,
+            heartbeat_transport=heartbeat_transport,
             set_global=set_global,
         )
 
@@ -150,6 +181,10 @@ def _do_init(
     mask: Callable[[Any], Any] | None,
     instruments: Sequence[str] | None,
     span_exporter: SpanExporter | None,
+    heartbeat: bool | None,
+    heartbeat_interval: float | None,
+    agent_name: str | None,
+    heartbeat_transport: Callable[[dict[str, Any]], None] | None,
     set_global: bool,
 ) -> GlassflowClient:
     global _current_client
@@ -161,6 +196,9 @@ def _do_init(
         disabled=disabled,
         sample_rate=sample_rate,
         capture_content=capture_content,
+        heartbeat=heartbeat,
+        heartbeat_interval=heartbeat_interval,
+        agent_name=agent_name,
     )
     # telemetry.sdk.* is reserved for the OTel SDK itself (Resource.create fills
     # it); we identify as a distribution via telemetry.distro.*.
@@ -197,7 +235,24 @@ def _do_init(
     if not config.disabled and (set_global or instruments is not None):
         enable_instrumentations(provider, instruments)
 
-    client = GlassflowClient(provider, config)
+    # Heartbeat: process-lifetime liveness, independent of trace
+    # traffic. The tracker rides the provider as a span processor so payloads
+    # can carry the currently-open root trace ids; disabled kills it too.
+    sender: HeartbeatSender | None = None
+    if config.heartbeat and not config.disabled:
+        tracker = OpenRootSpanTracker()
+        provider.add_span_processor(tracker)
+        sender = HeartbeatSender(
+            url=config.heartbeat_endpoint,
+            headers=config.headers,
+            interval=config.heartbeat_interval,
+            agent_name=config.agent_name,
+            tracker=tracker,
+            transport=heartbeat_transport,
+        )
+        sender.start()
+
+    client = GlassflowClient(provider, config, heartbeat=sender)
     if set_global:
         _current_client = client
     return client

src/glassflow/config.py

@@ -21,6 +21,15 @@
 ENV_DISABLED = "GLASSFLOW_DISABLED"
 ENV_SAMPLE_RATE = "GLASSFLOW_SAMPLE_RATE"
 ENV_CAPTURE_CONTENT = "GLASSFLOW_CAPTURE_CONTENT"
+ENV_HEARTBEAT = "GLASSFLOW_HEARTBEAT"
+ENV_HEARTBEAT_INTERVAL = "GLASSFLOW_HEARTBEAT_INTERVAL"
+ENV_AGENT_NAME = "GLASSFLOW_AGENT_NAME"
+
+# The backend expresses staleness as multiples of the interval, so the clamp
+# bounds are part of the heartbeat wire contract.
+HEARTBEAT_INTERVAL_MIN = 5.0
+HEARTBEAT_INTERVAL_MAX = 300.0
+DEFAULT_HEARTBEAT_INTERVAL = 15.0
 
 _TRUENESS = frozenset({"1", "true", "yes", "on"})
 
@@ -57,12 +66,20 @@ class GlassflowConfig:
     disabled: bool = False
     sample_rate: float = 1.0
     capture_content: bool = True
+    heartbeat: bool = False
+    heartbeat_interval: float = DEFAULT_HEARTBEAT_INTERVAL
+    agent_name: str = DEFAULT_SERVICE_NAME
 
     @property
     def traces_endpoint(self) -> str:
         """Full OTLP/HTTP traces URL (``<endpoint>/v1/traces``)."""
         return self.endpoint.rstrip("/") + "/v1/traces"
 
+    @property
+    def heartbeat_endpoint(self) -> str:
+        """Heartbeat URL (``<endpoint>/v1/heartbeat``) — same host as traces."""
+        return self.endpoint.rstrip("/") + "/v1/heartbeat"
+
 
 def _clamp_sample_rate(value: float) -> float:
     """Clamp to [0.0, 1.0] — an out-of-range value must degrade, not crash init()."""
@@ -73,6 +90,21 @@ def _clamp_sample_rate(value: float) -> float:
     return clamped
 
 
+def _clamp_heartbeat_interval(value: float) -> float:
+    """Clamp to the contract bounds — out-of-range degrades, never crashes init()."""
+    if HEARTBEAT_INTERVAL_MIN <= value <= HEARTBEAT_INTERVAL_MAX:
+        return value
+    clamped = min(max(value, HEARTBEAT_INTERVAL_MIN), HEARTBEAT_INTERVAL_MAX)
+    logger.warning(
+        "heartbeat_interval %s is outside [%s, %s]; clamped to %s",
+        value,
+        HEARTBEAT_INTERVAL_MIN,
+        HEARTBEAT_INTERVAL_MAX,
+        clamped,
+    )
+    return clamped
+
+
 def resolve_config(
     *,
     endpoint: str | None = None,
@@ -82,6 +114,9 @@ def resolve_config(
     disabled: bool | None = None,
     sample_rate: float | None = None,
     capture_content: bool | None = None,
+    heartbeat: bool | None = None,
+    heartbeat_interval: float | None = None,
+    agent_name: str | None = None,
 ) -> GlassflowConfig:
     """Resolve SDK configuration from arguments, environment, then defaults.
 
@@ -104,6 +139,15 @@ def resolve_config(
             (``GLASSFLOW_SAMPLE_RATE``).
         capture_content: When ``False``, content attributes are stripped at
             export (``GLASSFLOW_CAPTURE_CONTENT``).
+        heartbeat: Enable the agent-lifetime heartbeat thread
+            (``GLASSFLOW_HEARTBEAT``). Off by default this release.
+        heartbeat_interval: Seconds between pings
+            (``GLASSFLOW_HEARTBEAT_INTERVAL``), clamped to ``[5, 300]`` —
+            the backend derives staleness from this, so the bounds are part
+            of the wire contract.
+        agent_name: Identity heartbeats group under (``GLASSFLOW_AGENT_NAME``);
+            defaults to ``service_name`` so the agents view and the traces
+            view agree on what an "agent" is.
 
     Returns:
         The resolved, immutable ``GlassflowConfig``.
@@ -119,6 +163,14 @@ def resolve_config(
         _env_bool(ENV_CAPTURE_CONTENT, default=True) if capture_content is None else capture_content
     )
 
+    resolved_heartbeat = _env_bool(ENV_HEARTBEAT, default=False) if heartbeat is None else heartbeat
+    resolved_heartbeat_interval = _clamp_heartbeat_interval(
+        _env_float(ENV_HEARTBEAT_INTERVAL, default=DEFAULT_HEARTBEAT_INTERVAL)
+        if heartbeat_interval is None
+        else heartbeat_interval
+    )
+    resolved_agent_name = agent_name or os.getenv(ENV_AGENT_NAME) or resolved_service_name
+
     resolved_headers = dict(headers or {})
     has_auth = any(key.lower() == "authorization" for key in resolved_headers)
     if resolved_api_key and not has_auth:
@@ -132,4 +184,7 @@ def resolve_config(
         disabled=resolved_disabled,
         sample_rate=resolved_sample_rate,
         capture_content=resolved_capture_content,
+        heartbeat=resolved_heartbeat,
+        heartbeat_interval=resolved_heartbeat_interval,
+        agent_name=resolved_agent_name,
     )