feat: emit partial (pending) spans at span start

src/glassflow/client.py

@@ -19,6 +19,7 @@
 from .heartbeat import HeartbeatSender, OpenRootSpanTracker
 from .instrumentation import enable_instrumentations
 from .masking import MaskingSpanExporter
+from .pending import PendingSpanProcessor
 from .semconv import TRACER_NAME
 
 logger = logging.getLogger(__name__)
@@ -106,6 +107,8 @@ def init(
     heartbeat_interval: float | None = None,
     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.
@@ -165,6 +168,8 @@ def init(
             heartbeat_interval=heartbeat_interval,
             agent_name=agent_name,
             heartbeat_transport=heartbeat_transport,
+            partial_spans=partial_spans,
+            partial_spans_delay=partial_spans_delay,
             set_global=set_global,
         )
 
@@ -185,6 +190,8 @@ def _do_init(
     heartbeat_interval: float | None,
     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
@@ -199,6 +206,8 @@ def _do_init(
         heartbeat=heartbeat,
         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.*.
@@ -218,7 +227,14 @@ def _do_init(
             exporter = MaskingSpanExporter(
                 exporter, capture_content=config.capture_content, mask=mask
             )
-        provider.add_span_processor(BatchSpanProcessor(exporter))
+        batch_processor = BatchSpanProcessor(exporter)
+        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, delay=config.partial_spans_delay)
+            )
+        provider.add_span_processor(batch_processor)
 
     if set_global and not config.disabled:
         trace.set_tracer_provider(provider)

src/glassflow/config.py

@@ -24,13 +24,21 @@
 ENV_HEARTBEAT = "GLASSFLOW_HEARTBEAT"
 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"})
 
 
@@ -69,6 +77,8 @@ class GlassflowConfig:
     heartbeat: bool = False
     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:
@@ -90,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:
@@ -117,6 +142,8 @@ def resolve_config(
     heartbeat: bool | None = None,
     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.
 
@@ -148,6 +175,15 @@ def resolve_config(
         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.
+        partial_spans: Export a content-free pending snapshot of every
+            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``.
@@ -170,6 +206,14 @@ def resolve_config(
         else heartbeat_interval
     )
     resolved_agent_name = agent_name or os.getenv(ENV_AGENT_NAME) or resolved_service_name
+    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)
@@ -187,4 +231,6 @@ def resolve_config(
         heartbeat=resolved_heartbeat,
         heartbeat_interval=resolved_heartbeat_interval,
         agent_name=resolved_agent_name,
+        partial_spans=resolved_partial_spans,
+        partial_spans_delay=resolved_partial_spans_delay,
     )

src/glassflow/generation.py

@@ -31,6 +31,7 @@
     GEN_AI_USAGE_OUTPUT_TOKENS,
     TRACER_NAME,
     SpanKind,
+    kind_attributes,
     set_span_kind,
 )
 
@@ -234,6 +235,18 @@ def _configure(
         generation.set_input(input)
 
 
+def _creation_attributes(model: str | None, provider: str | None, operation: str) -> dict[str, str]:
+    """Identity attributes for an LLM span at CREATION (pending snapshots
+    are built at on_start; anything set later is invisible to them)."""
+    attributes = kind_attributes(SpanKind.LLM)
+    attributes[GEN_AI_OPERATION_NAME] = operation
+    if model is not None:
+        attributes[GEN_AI_REQUEST_MODEL] = model
+    if provider is not None:
+        attributes[GEN_AI_PROVIDER_NAME] = provider
+    return attributes
+
+
 def start_generation(
     name: str,
     *,
@@ -260,7 +273,9 @@ def start_generation(
     Returns:
         A ``Generation`` handle; call ``.end()`` when the call completes.
     """
-    span = trace.get_tracer(TRACER_NAME, __version__).start_span(name)
+    span = trace.get_tracer(TRACER_NAME, __version__).start_span(
+        name, attributes=_creation_attributes(model, provider, operation)
+    )
     generation = Generation(span)
     _configure(
         generation,
@@ -294,7 +309,9 @@ def start_as_current_generation(
         metadata; the span ends when the block exits.
     """
     tracer = trace.get_tracer(TRACER_NAME, __version__)
-    with tracer.start_as_current_span(name) as span:
+    with tracer.start_as_current_span(
+        name, attributes=_creation_attributes(model, provider, operation)
+    ) as span:
         generation = Generation(span)
         _configure(
             generation,