feat(telemetry): expose LLM token usage coverage

openviking/models/vlm/backends/litellm_vlm.py

@@ -560,9 +560,11 @@ def _update_token_usage_from_response(
         duration_seconds: float = 0.0,
     ) -> None:
         """Update token usage from response."""
-        if hasattr(response, "usage") and response.usage:
-            prompt_tokens = response.usage.prompt_tokens
-            completion_tokens = response.usage.completion_tokens
+        usage = getattr(response, "usage", None)
+        self.record_token_usage_coverage(reported=usage is not None)
+        if usage:
+            prompt_tokens = usage.prompt_tokens
+            completion_tokens = usage.completion_tokens
             self.update_token_usage(
                 model_name=self.model or "unknown",
                 provider=self.provider,

openviking/models/vlm/backends/openai_vlm.py

@@ -205,12 +205,14 @@ def _update_token_usage_from_response(
         response,
         duration_seconds: float = 0.0,
     ):
-        if hasattr(response, "usage") and response.usage:
-            tracer.info(f"response.usage={response.usage}")
-            prompt_tokens = response.usage.prompt_tokens
-            completion_tokens = response.usage.completion_tokens
-            prompt_tokens_details = getattr(response.usage, "prompt_tokens_details", None)
-            completion_tokens_details = getattr(response.usage, "completion_tokens_details", None)
+        usage = getattr(response, "usage", None)
+        self.record_token_usage_coverage(reported=usage is not None)
+        if usage:
+            tracer.info(f"response.usage={usage}")
+            prompt_tokens = usage.prompt_tokens
+            completion_tokens = usage.completion_tokens
+            prompt_tokens_details = getattr(usage, "prompt_tokens_details", None)
+            completion_tokens_details = getattr(usage, "completion_tokens_details", None)
             prompt_cached_tokens = getattr(prompt_tokens_details, "cached_tokens", 0) or 0
             completion_reasoning_tokens = (
                 getattr(completion_tokens_details, "reasoning_tokens", 0) or 0

openviking/models/vlm/backends/volcengine_media.py

@@ -363,6 +363,7 @@ def _record_usage(
 ) -> None:
     try:
         usage = getattr(response, "usage", None)
+        vlm.record_token_usage_coverage(reported=usage is not None)
         input_details = getattr(usage, "input_tokens_details", None)
         output_details = getattr(usage, "output_tokens_details", None)
         vlm.update_token_usage(

openviking/models/vlm/base.py

@@ -213,6 +213,23 @@ def is_available(self) -> bool:
         return self.api_key is not None or self.api_base is not None
 
     # Token usage tracking methods
+    def record_token_usage_coverage(self, *, reported: bool) -> None:
+        """Record token-usage coverage for one successful provider response."""
+        try:
+            from openviking.telemetry import get_current_telemetry
+
+            get_current_telemetry().record_llm_token_usage_report(reported=reported)
+        except Exception as e:
+            # Telemetry must never break model inference.
+            if logger.isEnabledFor(logging.DEBUG):
+                logger.debug(
+                    "vlm token-usage coverage emit failed provider=%s model_name=%s err=%s: %s",
+                    self.provider,
+                    self.model,
+                    type(e).__name__,
+                    e,
+                )
+
     def update_token_usage(
         self,
         model_name: str,

openviking/telemetry/operation.py

@@ -64,6 +64,8 @@ class TelemetrySummaryBuilder:
 
     _PRUNED = object()
     _PRESERVED_ZERO_METRIC_PATHS = {
+        ("tokens", "llm", "coverage", "missing_report_count"),
+        ("tokens", "llm", "coverage", "reported_response_count"),
         ("vector", "cuvs", "memory", "estimated_peak_bytes_max"),
         ("vector", "cuvs", "memory", "free_bytes_min"),
         ("vector", "cuvs", "memory", "usable_bytes_min"),
@@ -229,6 +231,14 @@ def build(
         llm_completion_reasoning_tokens = cls._i(counters.get("tokens.llm.completion_reasoning"), 0)
         embedding_total_tokens = cls._i(counters.get("tokens.embedding.total"), 0)
         rerank_total_tokens = cls._i(counters.get("tokens.rerank.total"), 0)
+        llm_successful_response_count = max(
+            cls._i(counters.get("tokens.llm.coverage.successful_responses"), 0),
+            0,
+        )
+        llm_reported_response_count = min(
+            max(cls._i(counters.get("tokens.llm.coverage.reported_responses"), 0), 0),
+            llm_successful_response_count,
+        )
         stage_token_summary = cls._build_stage_token_summary(counters)
         vector_candidates_scored = cls._i(counters.get("vector.scored"), 0)
         vectors_scanned = gauges.get("vector.scanned")
@@ -257,6 +267,20 @@ def build(
         }
         if stage_token_summary:
             summary["tokens"]["stages"] = stage_token_summary
+        if llm_successful_response_count > 0:
+            llm_missing_report_count = llm_successful_response_count - llm_reported_response_count
+            if llm_reported_response_count == llm_successful_response_count:
+                coverage_status = "complete"
+            elif llm_reported_response_count == 0:
+                coverage_status = "missing"
+            else:
+                coverage_status = "partial"
+            summary["tokens"]["llm"]["coverage"] = {
+                "status": coverage_status,
+                "successful_response_count": llm_successful_response_count,
+                "reported_response_count": llm_reported_response_count,
+                "missing_report_count": llm_missing_report_count,
+            }
 
         if cls._has_metric_prefix("queue", counters, gauges):
             summary["queue"] = {
@@ -716,6 +740,15 @@ def add_token_usage(
             completion_reasoning_tokens=completion_reasoning_tokens,
         )
 
+    def record_llm_token_usage_report(self, *, reported: bool) -> None:
+        """Record whether one successful LLM response included token usage."""
+        if not self.enabled:
+            return
+        with self._lock:
+            self._counters["tokens.llm.coverage.successful_responses"] += 1
+            if reported:
+                self._counters["tokens.llm.coverage.reported_responses"] += 1
+
     def record_token_usage(
         self,
         source: str,

tests/models/vlm/test_token_usage_coverage.py

@@ -0,0 +1,115 @@
+# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
+# SPDX-License-Identifier: AGPL-3.0
+"""Tests for VLM token-usage coverage telemetry."""
+
+from types import SimpleNamespace
+
+import pytest
+
+from openviking.models.vlm.backends.litellm_vlm import LiteLLMVLMProvider
+from openviking.models.vlm.backends.openai_vlm import OpenAIVLM
+from openviking.models.vlm.backends.volcengine_media import _record_usage
+from openviking.models.vlm.backends.volcengine_vlm import VolcEngineVLM
+from openviking.telemetry.backends.memory import MemoryOperationTelemetry
+from openviking.telemetry.context import bind_telemetry
+
+
+def _response(prompt_tokens: int | None, completion_tokens: int = 0):
+    if prompt_tokens is None:
+        return SimpleNamespace(usage=None)
+    return SimpleNamespace(
+        usage=SimpleNamespace(
+            prompt_tokens=prompt_tokens,
+            completion_tokens=completion_tokens,
+            prompt_tokens_details=None,
+            completion_tokens_details=None,
+        )
+    )
+
+
+@pytest.mark.parametrize(
+    ("provider_type", "config"),
+    [
+        (OpenAIVLM, {"provider": "openai", "model": "test-model"}),
+        (LiteLLMVLMProvider, {"provider": "litellm", "model": "test-model"}),
+    ],
+    ids=["openai", "litellm"],
+)
+@pytest.mark.parametrize(
+    ("responses", "expected_status", "expected_counts"),
+    [
+        ([(None, 0)], "missing", (1, 0, 1)),
+        ([(0, 0)], "complete", (1, 1, 0)),
+        ([(10, 4), (None, 0)], "partial", (2, 1, 1)),
+    ],
+    ids=["missing", "explicit-zero", "partial"],
+)
+def test_chat_response_token_usage_coverage(
+    provider_type,
+    config,
+    responses,
+    expected_status,
+    expected_counts,
+):
+    telemetry = MemoryOperationTelemetry(operation="resources.add_resource", enabled=True)
+    provider = provider_type(config)
+
+    with bind_telemetry(telemetry):
+        for prompt_tokens, completion_tokens in responses:
+            provider._update_token_usage_from_response(_response(prompt_tokens, completion_tokens))
+
+    summary = telemetry.finish().summary
+    coverage = summary["tokens"]["llm"]["coverage"]
+    successful, reported, missing = expected_counts
+    assert coverage == {
+        "status": expected_status,
+        "successful_response_count": successful,
+        "reported_response_count": reported,
+        "missing_report_count": missing,
+    }
+    if responses == [(10, 4), (None, 0)]:
+        assert summary["tokens"]["llm"]["total"] == 14
+
+
+@pytest.mark.parametrize(
+    ("usage", "expected_status"),
+    [(None, "missing"), ((0, 0), "complete")],
+    ids=["missing", "explicit-zero"],
+)
+def test_volcengine_media_response_token_usage_coverage(usage, expected_status):
+    telemetry = MemoryOperationTelemetry(operation="resources.add_resource", enabled=True)
+    provider = VolcEngineVLM({"provider": "volcengine", "model": "test-model"})
+    if usage is None:
+        response = SimpleNamespace(usage=None)
+    else:
+        input_tokens, output_tokens = usage
+        response = SimpleNamespace(
+            usage=SimpleNamespace(
+                input_tokens=input_tokens,
+                output_tokens=output_tokens,
+                input_tokens_details=None,
+                output_tokens_details=None,
+            )
+        )
+
+    with bind_telemetry(telemetry):
+        _record_usage(provider, response, duration_seconds=0.1)
+
+    assert summary_coverage(telemetry) == {
+        "status": expected_status,
+        "successful_response_count": 1,
+        "reported_response_count": int(usage is not None),
+        "missing_report_count": int(usage is None),
+    }
+
+
+def test_operation_without_successful_llm_response_omits_coverage():
+    telemetry = MemoryOperationTelemetry(operation="resources.add_resource", enabled=True)
+
+    summary = telemetry.finish().summary
+
+    assert "tokens" not in summary
+
+
+def summary_coverage(telemetry: MemoryOperationTelemetry):
+    return telemetry.finish().summary["tokens"]["llm"]["coverage"]