feat: record time to first token on the OpenTelemetry chat span

CHANGELOG.md

@@ -14,6 +14,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 * The `snowflake` extra (i.e., `pip install "chatlas[snowflake]"`) no longer installs `snowflake-ml-python`, which pulled in the heavy torch ecosystem and blocked installation on Python 3.13+. `ChatSnowflake()` has used the Cortex REST API (`snowflake.core`) since v0.11.0, so it only needs `snowflake-core` and `snowflake-snowpark-python`.
 
+### New features
+
+* Streamed responses now record the time to first token (in seconds) as the `gen_ai.response.time_to_first_chunk` attribute on the OpenTelemetry `chat` span.
+
 ### Bug fixes
 
 * `ChatBedrock()` (with `api="messages"` or `api="responses"`) no longer sends a URL for `content_pdf_url()`/`content_document_url()` content, which `bedrock-mantle` rejects since it can't fetch URLs itself. The bytes are downloaded and sent instead, as they already were before URL passthrough was added. `ChatBedrockAnthropic()` gets the same fix, since Bedrock's InvokeModel API can't fetch `url` sources either. (#410)

chatlas/_chat.py

@@ -65,6 +65,7 @@
     activate_span,
     end_span,
     record_chat_result,
+    record_chat_ttft_attr,
     record_error,
     start_agent_span,
     start_chat_span,
@@ -3007,6 +3008,9 @@ def emit(x: str | Content):
                 # provider's HTTP instrumentor spans nest under it -- never
                 # across a `yield`, which would leak the context into the
                 # consumer's scope.
+                # Measured from request issuance; chat_perform() blocks until
+                # headers arrive.
+                stream_start = time.monotonic()
                 with activate_span(chat_span):
                     response = self.provider.chat_perform(
                         stream=True,
@@ -3034,8 +3038,12 @@ def emit(x: str | Content):
                         for content in self.provider.stream_content(
                             chunk, result, turns=provider_turns
                         ):
+                            text = display_text(content)
+                            if stream_start is not None and text:
+                                record_chat_ttft_attr(chat_span, stream_start)
+                                stream_start = None
                             yield from acc.process_content(
-                                content, display_text(content), content_mode, emit
+                                content, text, content_mode, emit
                             )
 
                     yield from acc.flush_thinking(content_mode, emit)
@@ -3167,6 +3175,9 @@ def emit(x: str | Content):
                 # provider's HTTP instrumentor spans nest under it -- never
                 # across a `yield`, which would leak the context into the
                 # consumer's scope.
+                # Measured from request issuance; chat_perform_async() blocks
+                # until headers arrive.
+                stream_start = time.monotonic()
                 with activate_span(chat_span):
                     response = await self.provider.chat_perform_async(
                         stream=True,
@@ -3194,8 +3205,12 @@ def emit(x: str | Content):
                         for content in self.provider.stream_content(
                             chunk, result, turns=provider_turns
                         ):
+                            text = display_text(content)
+                            if stream_start is not None and text:
+                                record_chat_ttft_attr(chat_span, stream_start)
+                                stream_start = None
                             for item in acc.process_content(
-                                content, display_text(content), content_mode, emit
+                                content, text, content_mode, emit
                             ):
                                 yield item
 

chatlas/_otel.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import os
+import time
 from contextlib import AbstractContextManager, nullcontext
 from typing import TYPE_CHECKING, Any, Optional
 
@@ -101,6 +102,17 @@ def start_tool_span(
     )
 
 
+def record_chat_ttft_attr(span: Span, start: float) -> None:
+    """Record the time to first token (in seconds) on the chat span as the
+    `gen_ai.response.time_to_first_chunk` semconv attribute."""
+    if not span.is_recording():
+        return
+    span.set_attribute(
+        "gen_ai.response.time_to_first_chunk",
+        time.monotonic() - start,
+    )
+
+
 def record_chat_result(
     span: Span,
     turn: AssistantTurn[Any],

chatlas/types/anthropic/_client.py

@@ -7,6 +7,7 @@
     Any,
     Awaitable,
     Callable,
+    Literal,
     Mapping,
     Optional,
     Sequence,
@@ -37,6 +38,7 @@ class ChatClientArgs(TypedDict, total=False):
     max_retries: int
     default_headers: Optional[Mapping[str, str]]
     default_query: Optional[Mapping[str, object]]
+    open_telemetry: Union[anthropic.OpenTelemetryOptions, Literal[False], None]
     http_client: httpx2.AsyncClient | None
     middleware: Optional[
         Sequence[

chatlas/types/anthropic/_client_bedrock.py

@@ -6,6 +6,7 @@
     Any,
     Awaitable,
     Callable,
+    Literal,
     Mapping,
     Optional,
     Sequence,
@@ -30,6 +31,7 @@ class ChatBedrockClientArgs(TypedDict, total=False):
     max_retries: int
     default_headers: Optional[Mapping[str, str]]
     default_query: Optional[Mapping[str, object]]
+    open_telemetry: Union[anthropic.OpenTelemetryOptions, Literal[False], None]
     http_client: httpx2.AsyncClient | None
     middleware: Optional[
         Sequence[

chatlas/types/anthropic/_submit.py

@@ -48,6 +48,7 @@ class SubmitInputArgs(TypedDict, total=False):
     messages: Iterable[anthropic.types.message_param.MessageParam]
     model: Union[
         Literal[
+            "claude-haiku-5-5",
             "claude-sonnet-5-5",
             "claude-fable-5-1",
             "claude-opus-5-5",
@@ -86,6 +87,7 @@ class SubmitInputArgs(TypedDict, total=False):
         anthropic.types.output_config_param.OutputConfigParam | anthropic.Omit
     )
     service_tier: Union[Literal["auto", "standard_only"], anthropic.Omit]
+    speed: Union[Literal["fast", "standard"], None, anthropic.Omit]
     stop_sequences: Union[Sequence[str], anthropic.Omit]
     stream: Union[Literal[False], Literal[True], anthropic.Omit]
     system: Union[

pyproject.toml

@@ -104,7 +104,9 @@ mcp = [
     "mcp>=2.0.0;python_version>='3.10'"
 ]
 eval = [
-    "inspect-ai;python_version>='3.10'"
+    "inspect-ai;python_version>='3.10'",
+    # Inspect's Anthropic adapter constructs MessageParam, a union in SDK 2.0.
+    "anthropic<2.0;python_version>='3.10'",
 ]
 # Provider extras ----
 anthropic = ["anthropic>=1.0.0"]

tests/test_otel.py

@@ -1,16 +1,25 @@
 from __future__ import annotations
 
 import json
-from typing import Any, cast
+from collections.abc import AsyncIterator, Iterator, Sequence
+from types import SimpleNamespace
+from typing import cast
 
 import pytest
 
 pytest.importorskip("opentelemetry.sdk")
 
-from chatlas import ChatOpenAI, _otel
-from chatlas._content import ContentText, ContentToolRequest, ContentToolResult, ToolInfo
-from chatlas._turn import AssistantTurn
+from chatlas import ChatOpenAI, _chat, _otel
+from chatlas._content import (
+    Content,
+    ContentText,
+    ContentThinkingDelta,
+    ContentToolRequest,
+    ContentToolResult,
+    ToolInfo,
+)
 from chatlas._tools import ToolBuiltIn
+from chatlas._turn import AssistantTurn, Turn
 from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import SimpleSpanProcessor
 from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
@@ -269,6 +278,93 @@ def test_streaming_span_lifecycle(otel_setup: InMemorySpanExporter):
     assert chat_span.status.status_code.name == "UNSET", (
         f"Successful chat span should be UNSET, got {chat_span.status.status_code.name}"
     )
+    # Time to first token is recorded on the streamed chat span.
+    attrs = chat_span.attributes or {}
+    ttft = attrs.get("gen_ai.response.time_to_first_chunk")
+    assert isinstance(ttft, float) and ttft > 0, (
+        f"Expected gen_ai.response.time_to_first_chunk > 0, got {ttft!r}"
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"])
+@pytest.mark.parametrize(
+    "first_content",
+    [ContentText(text="hi"), ContentThinkingDelta(thinking="hmm")],
+    ids=["text", "thinking"],
+)
+async def test_ttft_recorded_at_first_non_empty_token(
+    otel_setup: InMemorySpanExporter,
+    monkeypatch: pytest.MonkeyPatch,
+    async_mode: bool,
+    first_content: ContentText | ContentThinkingDelta,
+) -> None:
+    chat = ChatOpenAI(model="gpt-4o-mini")
+    now = 0.0
+    clock = SimpleNamespace(monotonic=lambda: now)
+    monkeypatch.setattr(_chat, "time", clock)
+    monkeypatch.setattr(_otel, "time", clock)
+
+    def chunks() -> Iterator[object]:
+        nonlocal now
+        for timestamp in (2.0, 5.0, 9.0):
+            now = timestamp
+            yield object()
+
+    def streaming_perform(*args: object, **kwargs: object) -> Iterator[object]:
+        nonlocal now
+        now = 1.0
+        return chunks()
+
+    async def streaming_perform_async(
+        *args: object, **kwargs: object
+    ) -> AsyncIterator[object]:
+        nonlocal now
+        now = 1.0
+
+        async def gen() -> AsyncIterator[object]:
+            for chunk in chunks():
+                yield chunk
+
+        return gen()
+
+    content_chunks: list[list[Content]] = [
+        [ContentText.model_construct(text="")],
+        [first_content],
+        [ContentText(text="there")],
+    ]
+    stream_contents = iter(content_chunks)
+
+    def stream_content(
+        chunk: object, completion: object, turns: Sequence[Turn] = ()
+    ) -> list[Content]:
+        return next(stream_contents)
+
+    def stream_merge_chunks(result: object, chunk: object) -> object:
+        return chunk
+
+    def stream_turn(
+        completion: object, has_data_model: bool, turns: Sequence[Turn] = ()
+    ) -> AssistantTurn[object]:
+        return AssistantTurn("hi there")
+
+    monkeypatch.setattr(chat.provider, "chat_perform", streaming_perform)
+    monkeypatch.setattr(chat.provider, "chat_perform_async", streaming_perform_async)
+    monkeypatch.setattr(chat.provider, "stream_content", stream_content)
+    monkeypatch.setattr(chat.provider, "stream_merge_chunks", stream_merge_chunks)
+    monkeypatch.setattr(chat.provider, "stream_turn", stream_turn)
+
+    if async_mode:
+        async for _ in await chat.stream_async("Say hello."):
+            pass
+    else:
+        list(chat.stream("Say hello."))
+
+    spans = otel_setup.get_finished_spans()
+    chat_spans = [s for s in spans if s.name.startswith("chat ")]
+    assert len(chat_spans) == 1
+    attrs = chat_spans[0].attributes or {}
+    assert attrs.get("gen_ai.response.time_to_first_chunk") == 5.0
 
 
 def test_chat_error_recorded(otel_setup: InMemorySpanExporter):
@@ -534,6 +630,12 @@ async def gen():
     assert _parent_span_id(iter_span) == chat_span.context.span_id, (
         "provider span created while iterating the async stream should nest under the chat span"
     )
+    # Time to first token is recorded on the async streamed chat span.
+    attrs = chat_span.attributes or {}
+    ttft = attrs.get("gen_ai.response.time_to_first_chunk")
+    assert isinstance(ttft, float) and ttft > 0, (
+        f"Expected gen_ai.response.time_to_first_chunk > 0, got {ttft!r}"
+    )
 
 
 @pytest.mark.vcr