fix(telemetry): record LLM cycles when the answer is queued

The stream writer can read the first answer text after the worker starts
its next LLM call, for example while it saves earlier packets. The cycle
count then included the later call. The emitter now records each model's
count in the worker thread when it queues its first answer text. The
writer only selects the first model that answered.

Tests hold the writer on the answer until the next cycle starts. The
docs state that model comparisons without an answer report the deepest
model's calls, because the models run in parallel.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

.agents/feature-map/components/observability.md

@@ -419,7 +419,10 @@ every five minutes.
   fails with a 4xx `OnyxError` sends no event. This includes a chat that a
   query-processing hook rejects (`QUERY_REJECTED`): the chat stream sends the
   error as a `StreamingError` packet, and `QueryObservation.error_packet` reads
-  only its code.
+  only its code. Chat queries also report `llm_cycles`: `chat/llm_loop.py:run_llm_loop`
+  calls `Emitter.llm_step_started` before each LLM call, and `Emitter.emit` records
+  each model's count when it queues its first answer text. Deep research reports
+  the `deep_research` mode and no cycle count.
 - Indexing counters: the fetch, embed, and write steps call `emit_stage_counter`
   ([[indexing-pipeline]] §4.10). The sender thread sums the deltas for one
   attempt and stage over at most 30 seconds.
@@ -664,10 +667,10 @@ See `backend/AGENTS.md` for authoritative commands and required env.
 - **`utils/telemetry.py` is not fleet telemetry.** It holds the Cloud PostHog
   analytics calls (`mt_cloud_telemetry`, `mt_cloud_identify_user`), which run only
   under `MULTI_TENANT`. `DISABLE_TELEMETRY` does not affect them.
-- **`telemetry_chat` reads the channel from a keyword argument.** It reads
-  `new_msg_req` from the call's keywords (`utils/fleet_query_telemetry.py:_channel`).
-  If a caller passes it by position, the query event reports the channel `web`.
-  Nothing fails.
+- **`telemetry_chat` reads the channel and mode from a keyword argument.** It reads
+  `new_msg_req` from the call's keywords (`utils/fleet_query_telemetry.py:_channel`
+  and `_mode`). If a caller passes it by position, the query event reports the
+  channel `web` and the mode `chat`. Nothing fails.
 - **Discord queries report the channel `api`.** The send-message endpoint sets
   the origin to `api` for an API key or PAT caller
   (`server/query_and_chat/chat_backend.py:handle_send_chat_message`), and the

.agents/feature-map/components/streaming-protocol.md

@@ -144,10 +144,13 @@ string values.
 
 1. No-ops immediately if `drain_done` is set, so a cancelled turn's workers stop
    pushing packets without blocking on the queue.
-2. Copies the packet's `Placement` with `model_index` overwritten
+2. Passes the packet to the turn's fleet `QueryObservation`, if any
+   (`packet_queued`). In the worker thread, it records the model's LLM cycle
+   count at its first answer text, per [[observability]].
+3. Copies the packet's `Placement` with `model_index` overwritten
    (`base.model_copy(update={"model_index": self._model_idx})`). The producer's
    own `model_index`, if any, is discarded here.
-3. Puts `(model_idx, tagged_packet)` onto the shared `merged_queue`, the same
+4. Puts `(model_idx, tagged_packet)` onto the shared `merged_queue`, the same
    queue every model's worker thread writes to, per [[core-chat-loop]].
 
 `NullEmitter` (`emitter.py:NullEmitter`) implements the same interface but

backend/onyx/chat/emitter.py

@@ -46,6 +46,8 @@ class Emitter:
     def emit(self, packet: Packet) -> None:
         if self._drain_done is not None and self._drain_done.is_set():
             return
+        if self._query_observation is not None:
+            self._query_observation.packet_queued(packet, model_idx=self._model_idx)
         base = packet.placement or Placement(turn_index=0)
         tagged = Packet(
             placement=base.model_copy(update={"model_index": self._model_idx}),

backend/onyx/utils/fleet_query_telemetry.py

@@ -15,6 +15,7 @@ from onyx.utils.fleet_telemetry import emit_telemetry, error_category
 
 if TYPE_CHECKING:
     from onyx.chat.models import AnswerStreamPart, StreamingError
+    from onyx.server.query_and_chat.streaming_models import Packet
 
 F = TypeVar("F", bound=Callable[..., Any])
 
@@ -65,6 +66,12 @@ def _mode(kwargs: dict[str, Any]) -> QueryMode:
     return QueryMode.CHAT
 
 
+def _is_answer_text(packet: "Packet") -> bool:
+    from onyx.server.query_and_chat.streaming_models import AgentResponseDelta
+
+    return isinstance(packet.obj, AgentResponseDelta) and bool(packet.obj.content)
+
+
 class QueryToolUsage(str, Enum):
     NONE = "none"
     SEARCH_ONLY = "search_only"
@@ -88,7 +95,9 @@ class QueryObservation:
         self._used_other_tool = False
         # Chat-loop LLM calls per model index; model comparisons share one observation.
         self._llm_steps: dict[int, int] = {}
-        self._answer_llm_steps: int | None = None
+        # Each model's count when it queued its first answer text.
+        self._answer_llm_steps: dict[int, int] = {}
+        self._answer_model: int | None = None
         self._finished = False
 
     def llm_step_started(self, *, model_idx: int) -> None:
@@ -97,13 +106,24 @@ class QueryObservation:
                 return
             self._llm_steps[model_idx] = self._llm_steps.get(model_idx, 0) + 1
 
-    def _llm_cycles(self) -> int:
-        """LLM calls up to the first answer token, or every call when none arrived.
-        Text streamed before a tool call is the first answer, so later cycles
-        are not part of ``first_answer_ms``."""
-        if self._answer_llm_steps is not None:
-            return self._answer_llm_steps
-        return max(self._llm_steps.values(), default=0)
+    def packet_queued(self, packet: "Packet", *, model_idx: int) -> None:
+        """Runs in the model's worker thread before the writer can read the packet,
+        so a first answer keeps the count of the call that produced it."""
+        if not _is_answer_text(packet):
+            return
+        with self._lock:
+            if not self._finished:
+                self._answer_llm_steps.setdefault(
+                    model_idx, self._llm_steps.get(model_idx, 0)
+                )
+
+    def _llm_cycles(self) -> int | None:
+        """The first answering model's LLM calls up to its first answer text, which
+        can come before a tool call; None if that text bypassed the emitter. With no
+        answer, the most calls one model made: comparison models run in parallel."""
+        if self._answer_model is None:
+            return max(self._llm_steps.values(), default=0)
+        return self._answer_llm_steps.get(self._answer_model)
 
     def tool_started(self, *, is_search: bool) -> None:
         with self._lock:
@@ -128,14 +148,10 @@ class QueryObservation:
 
     def observe_packet(self, packet: "AnswerStreamPart") -> None:
         from onyx.chat.models import StreamingError
-        from onyx.server.query_and_chat.streaming_models import (
-            AgentResponseDelta,
-            OverallStop,
-            Packet,
-        )
+        from onyx.server.query_and_chat.streaming_models import OverallStop, Packet
 
         if isinstance(packet, Packet):
-            if isinstance(packet.obj, AgentResponseDelta) and packet.obj.content:
+            if _is_answer_text(packet):
                 self.answer(model_idx=packet.placement.model_index or 0)
             elif isinstance(packet.obj, OverallStop):
                 stop_outcome = _STOP_OUTCOMES.get(packet.obj.stop_reason or "")
@@ -147,10 +163,8 @@ class QueryObservation:
     def answer(self, *, model_idx: int) -> None:
         if self.first_answer_ms is None:
             self.first_answer_ms = max(0, (time.monotonic() - self.started) * 1000)
-            # The writer drains packets as workers queue them, and a model's next
-            # cycle waits for a tool round, so this count is still current.
             with self._lock:
-                self._answer_llm_steps = self._llm_steps.get(model_idx, 0)
+                self._answer_model = model_idx
 
     def results(self) -> None:
         if self.time_to_results_ms is None:

backend/tests/unit/onyx/chat/test_multi_model_streaming.py

@@ -1202,7 +1202,7 @@ def test_writer_reports_tool_usage_once_after_reader_disconnect(
 
 
 @pytest.mark.parametrize("n_models", [1, 2])
-def test_writer_reports_llm_cycles_at_the_first_answer(n_models: int) -> None:
+def test_lagging_writer_reports_the_cycle_of_the_first_answer(n_models: int) -> None:
     from onyx.chat.process_message import _run_models
     from onyx.utils.fleet_query_telemetry import (
         QueryMode,
@@ -1211,10 +1211,18 @@ def test_writer_reports_llm_cycles_at_the_first_answer(n_models: int) -> None:
     )
 
     finished = threading.Event()
+    next_cycle_started = threading.Event()
     observation = QueryObservation(channel="api", mode=QueryMode.CHAT)
     setup = _make_setup(n_models=n_models)
+    observe_packet = observation.observe_packet
 
-    def answer_after_one_tool_round(**kwargs: Any) -> None:
+    def lagging_observe(packet: Any) -> None:
+        # Hold the writer on the answer until its model starts another cycle.
+        if isinstance(packet, Packet) and isinstance(packet.obj, AgentResponseDelta):
+            assert next_cycle_started.wait(timeout=5)
+        observe_packet(packet)
+
+    def answer_then_keep_working(**kwargs: Any) -> None:
         emitter = kwargs["emitter"]
         emitter.llm_step_started()
         emitter.llm_step_started()
@@ -1224,13 +1232,15 @@ def test_writer_reports_llm_cycles_at_the_first_answer(n_models: int) -> None:
                 obj=AgentResponseDelta(content="private answer"),
             )
         )
+        emitter.llm_step_started()
+        next_cycle_started.set()
 
     with (
         patch("onyx.chat.process_message.load_settings"),
         patch("onyx.chat.process_message.get_session_with_current_tenant"),
         patch(
             "onyx.chat.process_message.run_llm_loop",
-            side_effect=answer_after_one_tool_round,
+            side_effect=answer_then_keep_working,
         ),
         patch("onyx.chat.process_message.construct_tools", return_value={}),
         patch("onyx.chat.process_message.llm_loop_completion_handle"),
@@ -1241,6 +1251,7 @@ def test_writer_reports_llm_cycles_at_the_first_answer(n_models: int) -> None:
             "onyx.chat.process_message.set_processing_status",
             side_effect=lambda **_: finished.set(),
         ),
+        patch.object(observation, "observe_packet", side_effect=lagging_observe),
         patch("onyx.utils.fleet_query_telemetry.emit_telemetry") as sink,
     ):
         list(

backend/tests/unit/onyx/utils/test_query_llm_cycles.py

@@ -1,5 +1,6 @@
 """Chat queries report the answering model's chat-loop LLM calls up to its first
-answer token, so cycle counts match the work that ``first_answer_ms`` measured."""
+answer token, so cycle counts match the work that ``first_answer_ms`` measured,
+even when the stream writer reads the answer after the next cycle starts."""
 
 from queue import Queue
 from unittest.mock import Mock
@@ -20,6 +21,19 @@ def _answer(model_idx: int = 0) -> Packet:
     )
 
 
+# The emitter's queue type; the stream writer reads it.
+StreamQueue = Queue[tuple[int, Packet | Exception | object]]
+
+
+def _observe_queued(
+    queue: StreamQueue, observation: telemetry.QueryObservation
+) -> None:
+    while not queue.empty():
+        _, packet = queue.get_nowait()
+        assert isinstance(packet, Packet)
+        observation.observe_packet(packet)
+
+
 @pytest.fixture
 def sink(monkeypatch: pytest.MonkeyPatch) -> Mock:
     sink = Mock()
@@ -37,21 +51,32 @@ def sink(monkeypatch: pytest.MonkeyPatch) -> Mock:
         (1, 3, 1),
     ],
 )
+@pytest.mark.parametrize("writer_lags", [False, True])
 def test_cycles_stop_at_the_first_answer_token(
-    sink: Mock, calls_before_answer: int, calls_after_answer: int, expected: int
+    sink: Mock,
+    calls_before_answer: int,
+    calls_after_answer: int,
+    expected: int,
+    writer_lags: bool,
 ) -> None:
+    queue: StreamQueue = Queue()
     observation = telemetry.QueryObservation(
         channel="web", mode=telemetry.QueryMode.CHAT
     )
-    emitter = Emitter(Queue(), query_observation=observation)
+    emitter = Emitter(queue, query_observation=observation)
     for _ in range(calls_before_answer):
         emitter.llm_step_started()
-    observation.observe_packet(_answer())
+    emitter.emit(_answer())
+    if not writer_lags:
+        _observe_queued(queue, observation)
+    # A lagging writer reads the answer after the worker starts more cycles.
     for _ in range(calls_after_answer):
         emitter.llm_step_started()
+    _observe_queued(queue, observation)
     observation.finish()
     event_type, data = sink.call_args.args
     assert data["llm_cycles"] == expected
+    assert data["first_answer_ms"] is not None
     assert sanitize_data(event_type, data) is not None
     assert "private" not in str(data)
 
@@ -74,21 +99,48 @@ def test_unanswered_queries_report_every_cycle_until_finish(sink: Mock) -> None:
 
 
 def test_model_comparisons_report_the_first_answering_model(sink: Mock) -> None:
+    queue: StreamQueue = Queue()
     observation = telemetry.QueryObservation(
         channel="web", mode=telemetry.QueryMode.CHAT
     )
     slow, fast = (
-        Emitter(Queue(), model_idx=i, query_observation=observation) for i in range(2)
+        Emitter(queue, model_idx=i, query_observation=observation) for i in range(2)
     )
     for _ in range(4):
         slow.llm_step_started()
     fast.llm_step_started()
-    observation.observe_packet(_answer(model_idx=1))
-    observation.observe_packet(_answer(model_idx=0))
+    fast.emit(_answer())
+    slow.emit(_answer())
+    _observe_queued(queue, observation)
     observation.finish()
     assert sink.call_args.args[1]["llm_cycles"] == 1
 
 
+def test_unanswered_model_comparisons_report_the_deepest_model(sink: Mock) -> None:
+    # The models run in parallel, so the wait is the deepest model's cycles, not the sum.
+    observation = telemetry.QueryObservation(
+        channel="web", mode=telemetry.QueryMode.CHAT
+    )
+    for model_idx, calls in ((0, 2), (1, 3)):
+        for _ in range(calls):
+            observation.llm_step_started(model_idx=model_idx)
+    observation.outcome = "canceled"
+    observation.finish()
+    assert sink.call_args.args[1]["llm_cycles"] == 3
+
+
+def test_an_answer_that_skipped_the_emitter_has_no_cycle_count(sink: Mock) -> None:
+    observation = telemetry.QueryObservation(
+        channel="web", mode=telemetry.QueryMode.CHAT
+    )
+    observation.llm_step_started(model_idx=0)
+    observation.observe_packet(_answer())
+    observation.finish()
+    data = sink.call_args.args[1]
+    assert data["first_answer_ms"] is not None
+    assert data["llm_cycles"] is None
+
+
 @pytest.mark.parametrize(
     "mode", [telemetry.QueryMode.SEARCH, telemetry.QueryMode.DEEP_RESEARCH]
 )

deployment/fleet-telemetry.md

@@ -47,10 +47,12 @@ Cancellation freezes usage at the server's stop point. Search-mode requests leav
 Outcome remains separate from tool usage. Deploy the receiving server's optional `tool_usage` field before this instrumentation.
 
 Chat queries report `llm_cycles`: the chat-loop LLM calls up to and including the call that produced the
-first answer token. If no answer token arrived, the count includes every call. 1 means the model answered
-directly. 2 means one round of tool calls came first. Text that a model streams before a tool call is the
-first answer, so the count always matches the work that `first_answer_ms` measures. Model comparisons
-report the count of the model that answered first. LLM calls inside tools do not count.
+first answer token. 1 means the model answered directly. 2 means one round of tool calls came first. Text
+that a model streams before a tool call is the first answer, so the count always matches the work that
+`first_answer_ms` measures. The emitter records the count when it queues that text, so a slow stream writer
+cannot change it. Model comparisons report the count of the model that answered first. If no answer token
+arrived, the count is the highest number of calls that one model made: the models run in parallel, so their
+counts are not added. LLM calls inside tools do not count.
 
 Deep research requests report the `deep_research` mode instead of `chat`. They report tool usage, but not
 `llm_cycles`, because their orchestration is not a chat loop. Deploy the receiving server's optional