feat: debounced partial spans — delay emission, cancel on fast finish

src/glassflow/client.py

@@ -108,6 +108,7 @@ def init(
     agent_name: str | None = None,
     heartbeat_transport: Callable[[dict[str, Any]], None] | None = None,
     partial_spans: bool | None = None,
+    partial_spans_delay: float | None = None,
     set_global: bool = True,
 ) -> GlassflowClient:
     """Initialize the SDK: build a tracer provider that exports OTLP traces.
@@ -168,6 +169,7 @@ def init(
             agent_name=agent_name,
             heartbeat_transport=heartbeat_transport,
             partial_spans=partial_spans,
+            partial_spans_delay=partial_spans_delay,
             set_global=set_global,
         )
 
@@ -189,6 +191,7 @@ def _do_init(
     agent_name: str | None,
     heartbeat_transport: Callable[[dict[str, Any]], None] | None,
     partial_spans: bool | None,
+    partial_spans_delay: float | None,
     set_global: bool,
 ) -> GlassflowClient:
     global _current_client
@@ -204,6 +207,7 @@ def _do_init(
         heartbeat_interval=heartbeat_interval,
         agent_name=agent_name,
         partial_spans=partial_spans,
+        partial_spans_delay=partial_spans_delay,
     )
     # telemetry.sdk.* is reserved for the OTel SDK itself (Resource.create fills
     # it); we identify as a distribution via telemetry.distro.*.
@@ -227,7 +231,9 @@ def _do_init(
         if config.partial_spans:
             # Pending snapshots ride the SAME batch pipeline as final spans
             # (exporter, retries, masking); see pending.py for the contract.
-            provider.add_span_processor(PendingSpanProcessor(batch_processor))
+            provider.add_span_processor(
+                PendingSpanProcessor(batch_processor, delay=config.partial_spans_delay)
+            )
         provider.add_span_processor(batch_processor)
 
     if set_global and not config.disabled:

src/glassflow/config.py

@@ -25,13 +25,20 @@
 ENV_HEARTBEAT_INTERVAL = "GLASSFLOW_HEARTBEAT_INTERVAL"
 ENV_AGENT_NAME = "GLASSFLOW_AGENT_NAME"
 ENV_PARTIAL_SPANS = "GLASSFLOW_PARTIAL_SPANS"
+ENV_PARTIAL_SPANS_DELAY = "GLASSFLOW_PARTIAL_SPANS_DELAY"
 
 # 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
 
+# Debounce for partial spans (GLA2-244): 0 = emit immediately at span start;
+# N>0 = emit only if the span is still open after N seconds. Beyond 60s a
+# "live" view stops being live, so larger values are clamped.
+PARTIAL_SPANS_DELAY_MIN = 0.0
+PARTIAL_SPANS_DELAY_MAX = 60.0
+
 _TRUENESS = frozenset({"1", "true", "yes", "on"})
 
 
@@ -71,6 +78,7 @@ class GlassflowConfig:
     heartbeat_interval: float = DEFAULT_HEARTBEAT_INTERVAL
     agent_name: str = DEFAULT_SERVICE_NAME
     partial_spans: bool = False
+    partial_spans_delay: float = 0.0
 
     @property
     def traces_endpoint(self) -> str:
@@ -92,6 +100,21 @@ def _clamp_sample_rate(value: float) -> float:
     return clamped
 
 
+def _clamp_partial_spans_delay(value: float) -> float:
+    """Clamp to [0, 60] — out-of-range degrades, never crashes init()."""
+    if PARTIAL_SPANS_DELAY_MIN <= value <= PARTIAL_SPANS_DELAY_MAX:
+        return value
+    clamped = min(max(value, PARTIAL_SPANS_DELAY_MIN), PARTIAL_SPANS_DELAY_MAX)
+    logger.warning(
+        "partial_spans_delay %s is outside [%s, %s]; clamped to %s",
+        value,
+        PARTIAL_SPANS_DELAY_MIN,
+        PARTIAL_SPANS_DELAY_MAX,
+        clamped,
+    )
+    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:
@@ -120,6 +143,7 @@ def resolve_config(
     heartbeat_interval: float | None = None,
     agent_name: str | None = None,
     partial_spans: bool | None = None,
+    partial_spans_delay: float | None = None,
 ) -> GlassflowConfig:
     """Resolve SDK configuration from arguments, environment, then defaults.
 
@@ -155,6 +179,11 @@ def resolve_config(
             sampled span at span START (``GLASSFLOW_PARTIAL_SPANS``), so
             in-flight work is visible and crashes leave a record. Off by
             default until the backend's unfinished-spans storage ships.
+        partial_spans_delay: Debounce for pending snapshots
+            (``GLASSFLOW_PARTIAL_SPANS_DELAY``), clamped to ``[0, 60]``
+            seconds. ``0`` (default) emits at span start; ``N`` emits only if
+            the span is still open after N seconds — spans that finish
+            sooner cost no network at all.
 
     Returns:
         The resolved, immutable ``GlassflowConfig``.
@@ -180,6 +209,11 @@ def resolve_config(
     resolved_partial_spans = (
         _env_bool(ENV_PARTIAL_SPANS, default=False) if partial_spans is None else partial_spans
     )
+    resolved_partial_spans_delay = _clamp_partial_spans_delay(
+        _env_float(ENV_PARTIAL_SPANS_DELAY, default=0.0)
+        if partial_spans_delay is None
+        else partial_spans_delay
+    )
 
     resolved_headers = dict(headers or {})
     has_auth = any(key.lower() == "authorization" for key in resolved_headers)
@@ -198,4 +232,5 @@ def resolve_config(
         heartbeat_interval=resolved_heartbeat_interval,
         agent_name=resolved_agent_name,
         partial_spans=resolved_partial_spans,
+        partial_spans_delay=resolved_partial_spans_delay,
     )

src/glassflow/pending.py

@@ -18,14 +18,27 @@
 - Identity/taxonomy attributes only (``PENDING_IDENTITY_ATTRIBUTES`` /
   ``_PREFIXES``); NEVER content, whatever instrumentation set it.
 
-v1 emits immediately on start (Logfire-style). The emission seam
-(:meth:`PendingSpanProcessor._emit`) exists so a debounce ("only emit if
-still open after N seconds" — the volume escape valve) can be added later
-without changing the wire contract.
+Debounce (GLA2-244): with ``partial_spans_delay > 0`` the snapshot is held
+for N seconds and only emitted if the span is STILL OPEN then — a span that
+finishes first costs zero network. Most agent spans live milliseconds, so a
+small delay cuts pending volume drastically while keeping the live view
+useful (anything worth watching live is open longer than the delay). The
+snapshot is still built at ``on_start`` and held, never rebuilt at emit
+time: content set during the delay (``set_input`` etc.) can never leak onto
+a pending. A delayed pending is byte-identical to an immediate one — zero
+wire/backend/UI impact.
 """
 
 from __future__ import annotations
 
+import heapq
+import itertools
+import logging
+import os
+import threading
+import time
+import weakref
+from collections.abc import Callable
 from typing import Any
 
 from opentelemetry import context as otel_context
@@ -38,6 +51,136 @@
     PENDING_IDENTITY_PREFIXES,
 )
 
+logger = logging.getLogger(__name__)
+
+_SpanKey = tuple[int, int]  # (trace_id, span_id)
+
+# ``os.register_at_fork`` callbacks can never be unregistered, so the hook is
+# installed once at module level over a weak set of live schedulers — the same
+# pattern as the heartbeat sender. A forked child re-arms its scheduler thread
+# with an EMPTY registry: the parent's open spans are not the child's.
+_active_schedulers: weakref.WeakSet[PendingScheduler] = weakref.WeakSet()
+_fork_hook_installed = False
+_fork_lock = threading.Lock()
+
+
+def _reset_schedulers_in_child() -> None:  # pragma: no cover - exercised via fork
+    for scheduler in list(_active_schedulers):
+        scheduler._at_fork_reinit()
+
+
+def _install_fork_hook() -> None:
+    global _fork_hook_installed
+    with _fork_lock:
+        if _fork_hook_installed or not hasattr(os, "register_at_fork"):
+            return
+        os.register_at_fork(after_in_child=_reset_schedulers_in_child)
+        _fork_hook_installed = True
+
+
+class PendingScheduler:
+    """Delays snapshot emission; a span ending first cancels its snapshot.
+
+    ONE daemon thread regardless of span volume: deadlines live in a heap,
+    snapshots in a key->snapshot registry. ``cancel`` just drops the registry
+    entry (heap entries for cancelled keys are discarded lazily), so both
+    ``schedule`` and ``cancel`` are O(log n) / O(1) — safe on the span hot
+    path. ``clock`` and ``start_thread`` are injectable for tests.
+    """
+
+    def __init__(
+        self,
+        *,
+        emit: Callable[[ReadableSpan], None],
+        delay: float,
+        clock: Callable[[], float] = time.monotonic,
+        start_thread: bool = True,
+    ) -> None:
+        self._emit_fn = emit
+        self._delay = delay
+        self._clock = clock
+        self._cond = threading.Condition()
+        self._heap: list[tuple[float, int, _SpanKey]] = []
+        self._snapshots: dict[_SpanKey, ReadableSpan] = {}
+        self._counter = itertools.count()  # heap tiebreaker
+        self._stopped = False
+        self._thread: threading.Thread | None = None
+        if start_thread:
+            _active_schedulers.add(self)
+            _install_fork_hook()
+            self._start_thread()
+
+    def _start_thread(self) -> None:
+        self._thread = threading.Thread(
+            target=self._run, name="glassflow-pending-scheduler", daemon=True
+        )
+        self._thread.start()
+
+    def schedule(self, key: _SpanKey, snapshot: ReadableSpan) -> None:
+        with self._cond:
+            if self._stopped:
+                return
+            self._snapshots[key] = snapshot
+            heapq.heappush(self._heap, (self._clock() + self._delay, next(self._counter), key))
+            self._cond.notify_all()
+
+    def cancel(self, key: _SpanKey) -> None:
+        """Span ended before its deadline: the pending never hits the wire."""
+        with self._cond:
+            self._snapshots.pop(key, None)
+
+    def pop_due(self) -> None:
+        """Emit every snapshot whose deadline has passed (thread and tests)."""
+        due: list[ReadableSpan] = []
+        with self._cond:
+            now = self._clock()
+            while self._heap and self._heap[0][0] <= now:
+                _, _, key = heapq.heappop(self._heap)
+                snapshot = self._snapshots.pop(key, None)
+                if snapshot is not None:  # None = cancelled, discard lazily
+                    due.append(snapshot)
+        for snapshot in due:
+            try:
+                self._emit_fn(snapshot)
+            except Exception:  # noqa: BLE001 - never propagate into the SDK
+                logger.debug("pending snapshot emission failed", exc_info=True)
+
+    def shutdown(self) -> None:
+        """Drop everything not yet due: the final spans are being flushed at
+        this moment, so any pending emitted now would be instantly superseded."""
+        with self._cond:
+            self._stopped = True
+            self._snapshots.clear()
+            self._heap.clear()
+            self._cond.notify_all()
+        if self._thread is not None:
+            self._thread.join(timeout=1.0)
+
+    def _at_fork_reinit(self) -> None:  # pragma: no cover - exercised via fork
+        # Fresh lock (the parent's may be held mid-fork), empty registry, new
+        # thread: parent spans do not exist in the child.
+        self._cond = threading.Condition()
+        self._heap = []
+        self._snapshots = {}
+        if not self._stopped:
+            self._start_thread()
+
+    def _run(self) -> None:
+        while True:
+            with self._cond:
+                if self._stopped:
+                    return
+                # discard cancelled heads so the timeout tracks a LIVE deadline
+                while self._heap and self._heap[0][2] not in self._snapshots:
+                    heapq.heappop(self._heap)
+                timeout = None
+                if self._heap:
+                    timeout = max(0.0, self._heap[0][0] - self._clock())
+                self._cond.wait(timeout)
+                if self._stopped:
+                    return
+            self.pop_due()
+
 
 def _identity_attributes(attributes: Any) -> dict[str, Any]:
     """Filter a span's start-time attributes down to the pending allowlist."""
@@ -50,22 +193,37 @@ def _identity_attributes(attributes: Any) -> dict[str, Any]:
     }
 
 
+def _span_key(context: Any) -> _SpanKey:
+    return (context.trace_id, context.span_id)
+
+
 class PendingSpanProcessor(SpanProcessor):
     """Exports a pending snapshot of every sampled span at ``on_start``.
 
     Delegates the snapshot to the provider's existing batch processor
     (``delegate.on_end``), so pendings share the exporter, batching, retry,
     and masking pipeline with final spans — nothing bespoke on the wire path.
     ``on_start`` stays an in-memory enqueue: the never-block guarantee holds.
+
+    With ``delay > 0`` (GLA2-244) emission is debounced through a
+    :class:`PendingScheduler`; ``delay == 0`` keeps the emit-immediately
+    behavior with no scheduler thread at all.
     """
 
-    def __init__(self, delegate: SpanProcessor) -> None:
+    def __init__(self, delegate: SpanProcessor, *, delay: float = 0.0) -> None:
         self._delegate = delegate
+        self._scheduler: PendingScheduler | None = None
+        if delay > 0:
+            self._scheduler = PendingScheduler(emit=delegate.on_end, delay=delay)
 
     def on_start(self, span: Span, parent_context: otel_context.Context | None = None) -> None:
         if not span.is_recording():
             return
-        self._emit(self._snapshot(span))
+        snapshot = self._snapshot(span)
+        if self._scheduler is not None:
+            self._scheduler.schedule(_span_key(span.get_span_context()), snapshot)
+        else:
+            self._emit(snapshot)
 
     @staticmethod
     def _snapshot(span: Span) -> ReadableSpan:
@@ -87,15 +245,22 @@ def _snapshot(span: Span) -> ReadableSpan:
         )
 
     def _emit(self, snapshot: ReadableSpan) -> None:
-        # The debounce seam: a future timer wraps THIS call (delay + cancel on
-        # early end), leaving the snapshot construction and wire shape alone.
         self._delegate.on_end(snapshot)
 
-    def on_end(self, span: ReadableSpan) -> None:  # pragma: no cover - no-op
-        pass
+    def on_end(self, span: ReadableSpan) -> None:
+        # The debounce cancellation hook: a span that ends within the delay
+        # never sends its pending at all.
+        if self._scheduler is not None and span.context is not None:
+            self._scheduler.cancel(_span_key(span.context))
 
-    def shutdown(self) -> None:  # pragma: no cover - delegate owns the exporter
-        pass
+    def shutdown(self) -> None:
+        if self._scheduler is not None:
+            self._scheduler.shutdown()
 
     def force_flush(self, timeout_millis: int = 30000) -> bool:
+        # Deliberately NOT a drop (deviation from the ticket's prose, kept to
+        # its ACs): flush() happens mid-operation — killing scheduled pendings
+        # here would silently disable liveness for spans that stay open. The
+        # batch delegate flushes its own queue; not-yet-due pendings simply
+        # emit later if their spans are still open.
         return True