fix(sambanova): correct DeepSeek-V3.2 context window (32k -> 128k)

examples/lar1_ollama_config.yaml

@@ -0,0 +1,45 @@
+model_list:
+  - model_name: agent-router
+    litellm_params:
+      model: ollama/qwen3.5:9b
+      api_base: http://127.0.0.1:11434
+    model_info:
+      id: cloud-smart
+      type: cloud-smart
+
+  - model_name: agent-router
+    litellm_params:
+      model: ollama/phi4-mini:latest
+      api_base: http://127.0.0.1:11434
+    model_info:
+      id: cloud-fast
+      type: cloud-fast
+
+  - model_name: agent-router
+    litellm_params:
+      model: ollama/llama3.2:3b
+      api_base: http://127.0.0.1:11434
+    model_info:
+      id: local
+      type: local
+
+  - model_name: agent-router
+    litellm_params:
+      model: ollama/lfm2.5-thinking:latest
+      api_base: http://127.0.0.1:11434
+    model_info:
+      id: deep
+      type: deep
+
+router_settings:
+  routing_strategy: lar1
+  routing_strategy_args:
+    confidence_threshold_low: 0.3
+    confidence_threshold_medium: 0.5
+    confidence_threshold_high: 0.7
+
+general_settings:
+  master_key: sk-lar1-demo
+
+litellm_settings:
+  set_verbose: true

litellm/integrations/focus/destinations/mavvrik_destination.py

@@ -3,6 +3,7 @@
 Flow:
   1. GET /metrics/agent/ai/{connection_id}/upload-url → GCS signed URL
   2. PUT <signed_url> with CSV content
+  3. PATCH /metrics/agent/ai/{connection_id} → advance metricsMarker
 """
 
 from __future__ import annotations
@@ -33,8 +34,7 @@ def _validate_api_endpoint(api_endpoint: str) -> None:
     hostname = (urlparse(api_endpoint).hostname or "").lower()
     if not any(hostname.endswith(suffix) for suffix in _MAVVRIK_ALLOWED_SUFFIXES):
         raise ValueError(
-            "MAVVRIK_API_ENDPOINT host must be a Mavvrik domain "
-            "(e.g. https://api.mavvrik.dev/<tenant_id>)"
+            "MAVVRIK_API_ENDPOINT host must be a Mavvrik domain (e.g. https://api.mavvrik.dev/<tenant_id>)"
         )
 
 
@@ -50,8 +50,7 @@ def _validate_gcs_url(url: str, label: str) -> None:
         or hostname.endswith(".storage.googleapis.com")
     ):
         raise ValueError(
-            f"Mavvrik FOCUS destination: {label} must be a GCS endpoint "
-            f"(storage.googleapis.com), got '{hostname}'"
+            f"Mavvrik FOCUS destination: {label} must be a GCS endpoint (storage.googleapis.com), got '{hostname}'"
         )
 
 
@@ -127,17 +126,14 @@ async def _ensure_registered(self) -> Optional[int]:
             timeout=30.0,
         )
         if resp.status_code == 410:
-            # Connector has been disconnected in Mavvrik — reset flag so next
-            # delivery attempt re-registers after it becomes active again.
             self._registered = False
             raise RuntimeError(
                 "Mavvrik FOCUS destination: connector is disconnected (410). "
                 "Re-enable the connection in the Mavvrik dashboard."
             )
         if resp.status_code >= 400:
             raise RuntimeError(
-                f"Mavvrik FOCUS destination: register failed "
-                f"({resp.status_code}): {resp.text[:200]}"
+                f"Mavvrik FOCUS destination: register failed ({resp.status_code}): {resp.text[:200]}"
             )
         self._registered = True
         metrics_marker = resp.json().get("metricsMarker", 0)
@@ -159,8 +155,7 @@ async def _get_signed_url(self, date_str: str) -> str:
         )
         if resp.status_code >= 400:
             raise RuntimeError(
-                f"Mavvrik FOCUS destination: failed to get signed URL "
-                f"({resp.status_code}): {resp.text[:200]}"
+                f"Mavvrik FOCUS destination: failed to get signed URL ({resp.status_code}): {resp.text[:200]}"
             )
         signed_url = resp.json().get("url")
         if not signed_url:
@@ -205,8 +200,7 @@ async def _upload_to_gcs(self, signed_url: str, content: bytes) -> None:
         )
         if init_resp.status_code not in (200, 201):
             raise RuntimeError(
-                f"Mavvrik FOCUS destination: GCS session init failed "
-                f"({init_resp.status_code}): {init_resp.text[:400]}"
+                f"Mavvrik FOCUS destination: GCS session init failed ({init_resp.status_code}): {init_resp.text[:400]}"
             )
 
         session_uri = init_resp.headers.get("Location")
@@ -217,8 +211,7 @@ async def _upload_to_gcs(self, signed_url: str, content: bytes) -> None:
         _validate_gcs_url(session_uri, "session URI")
 
         verbose_logger.debug(
-            "Mavvrik FOCUS destination: GCS session started, uploading %d gzip bytes "
-            "in %d chunk(s)",
+            "Mavvrik FOCUS destination: GCS session started, uploading %d gzip bytes in %d chunk(s)",
             total,
             max(1, -(-total // _GCS_CHUNK_SIZE)),  # ceiling division
         )
@@ -273,14 +266,33 @@ async def _upload_to_gcs(self, signed_url: str, content: bytes) -> None:
                 pass
             raise
 
+    async def _update_metrics_marker(self, date_epoch: int) -> None:
+        """PATCH agent endpoint to advance metricsMarker after a successful upload."""
+        resp = await self._http.client.request(
+            method="PATCH",
+            url=self._agent_url,
+            headers=self._auth_headers,
+            json={"metricsMarker": date_epoch},
+            timeout=30.0,
+        )
+        if resp.status_code == 410:
+            self._registered = False
+            raise RuntimeError(
+                "Mavvrik FOCUS destination: connector is disconnected (410). "
+                "Re-enable the connection in the Mavvrik dashboard."
+            )
+        if resp.status_code >= 400:
+            raise RuntimeError(
+                f"Mavvrik FOCUS destination: failed to update metricsMarker "
+                f"({resp.status_code}): {resp.text[:200]}"
+            )
+        verbose_logger.debug(
+            "Mavvrik FOCUS destination: metricsMarker advanced to %s", date_epoch
+        )
+
     async def get_metrics_marker(self) -> Optional[int]:
         """Register with Mavvrik and return the current metricsMarker.
 
-        The metricsMarker is a Unix timestamp (seconds) representing the last
-        date Mavvrik has successfully ingested. Called on every scheduled run
-        so the logger can detect and catch up any dates missed due to previous
-        export failures.
-
         Always calls the Mavvrik register API — unlike deliver() which skips
         registration once _registered is True, catch-up requires a fresh
         marker value on every run.
@@ -300,8 +312,7 @@ async def get_metrics_marker(self) -> Optional[int]:
             )
         if resp.status_code >= 400:
             raise RuntimeError(
-                f"Mavvrik FOCUS destination: register failed "
-                f"({resp.status_code}): {resp.text[:200]}"
+                f"Mavvrik FOCUS destination: register failed ({resp.status_code}): {resp.text[:200]}"
             )
         self._registered = True
         metrics_marker = resp.json().get("metricsMarker", 0)
@@ -321,24 +332,29 @@ async def deliver(
 
         Uses the start date of the time window as the object date key.
         """
+        date_str = time_window.start_time.strftime("%Y-%m-%d")
+        date_epoch = int(time_window.start_time.timestamp())
+
+        await self._ensure_registered()
+
         if not content:
             verbose_logger.debug(
-                "Mavvrik FOCUS destination: empty content, skipping upload"
+                "Mavvrik FOCUS destination: empty content for date=%s, advancing marker",
+                date_str,
             )
+            await self._update_metrics_marker(date_epoch)
             return
 
-        date_str = time_window.start_time.strftime("%Y-%m-%d")
-
         verbose_logger.debug(
             "Mavvrik FOCUS destination: uploading %d bytes for date=%s (%s)",
             len(content),
             date_str,
             filename,
         )
 
-        await self._ensure_registered()
         signed_url = await self._get_signed_url(date_str)
         await self._upload_to_gcs(signed_url, content)
+        await self._update_metrics_marker(date_epoch)
 
         verbose_logger.debug(
             "Mavvrik FOCUS destination: upload complete for date=%s", date_str

litellm/integrations/mavvrik_focus/mavvrik_focus_logger.py

@@ -72,6 +72,16 @@ def _parse_metrics_marker(
     return None
 
 
+def _is_empty_metrics_marker(marker: Optional[object]) -> bool:
+    if marker is None:
+        return True
+    if isinstance(marker, (int, float)):
+        return marker == 0
+    if isinstance(marker, str):
+        return not marker.strip()
+    return False
+
+
 class MavvrikFocusLogger(FocusLogger):
     """FOCUS-based export logger that routes to the Mavvrik destination."""
 
@@ -122,19 +132,17 @@ async def _export_window(
                 window.start_time.date(),
                 window.end_time.date(),
             )
+        payload = b""
         if data.is_empty():
             verbose_proxy_logger.debug(
                 "Mavvrik FOCUS export: no usage data for window %s", window
             )
-            return
-        normalized = engine._transformer.transform(data)
-        if normalized.is_empty():
-            return
-        payload = engine._serializer.serialize(normalized)
-        if not payload:
-            return
+        else:
+            normalized = engine._transformer.transform(data)
+            if not normalized.is_empty():
+                payload = engine._serializer.serialize(normalized)
         await engine._destination.deliver(
-            content=payload,
+            content=payload or b"",
             time_window=window,
             filename=engine._build_filename(window),
         )
@@ -149,8 +157,8 @@ async def _run_scheduled_export(self) -> None:
 
         On each run:
         1. Register with Mavvrik → get metricsMarker (last successfully ingested date)
-        2. If metricsMarker is behind yesterday, catch up missed dates (capped at
-           _MAX_CATCHUP_DAYS to avoid runaway loops on long outages)
+        2. If metricsMarker is behind yesterday (or 0/None for a fresh connector),
+           catch up missed dates (capped at _MAX_CATCHUP_DAYS)
         3. Export yesterday (today's daily window)
 
         This ensures a failed export on day N is automatically retried on day N+1
@@ -177,13 +185,19 @@ async def _run_scheduled_export(self) -> None:
 
         last_ingested = _parse_metrics_marker(marker)
 
-        # Catch up missed dates, capped at _MAX_CATCHUP_DAYS
-        if last_ingested and last_ingested < yesterday:
-            # Never go further back than _MAX_CATCHUP_DAYS from yesterday
-            earliest_catchup = yesterday - timedelta(days=self._MAX_CATCHUP_DAYS - 1)
-            catch_up_date = max(last_ingested + timedelta(days=1), earliest_catchup)
+        is_empty_marker = _is_empty_metrics_marker(marker)
+        earliest_catchup = yesterday - timedelta(days=self._MAX_CATCHUP_DAYS - 1)
+        if is_empty_marker or (last_ingested is not None and last_ingested < yesterday):
+            catch_up_date = (
+                earliest_catchup
+                if last_ingested is None
+                else max(last_ingested + timedelta(days=1), earliest_catchup)
+            )
 
-            if last_ingested + timedelta(days=1) < earliest_catchup:
+            if (
+                last_ingested is not None
+                and last_ingested + timedelta(days=1) < earliest_catchup
+            ):
                 verbose_proxy_logger.warning(
                     "Mavvrik FOCUS export: metricsMarker is more than %d days behind "
                     "(%s). Catching up from %s only; earlier data will not be re-exported.",
@@ -197,18 +211,24 @@ async def _run_scheduled_export(self) -> None:
                     "Mavvrik FOCUS export: catching up missed date %s",
                     catch_up_date.date(),
                 )
+                # Use now as end_time for catch-up windows too — rows for old dates
+                # may have been flushed to DB well after their calendar day ended.
+                catch_up_end = min(catch_up_date + timedelta(days=1), now)
                 window = FocusTimeWindow(
                     start_time=catch_up_date,
-                    end_time=catch_up_date + timedelta(days=1),
+                    end_time=catch_up_end,
                     frequency="daily",
                 )
                 await self._export_window(window=window, limit=None)
                 catch_up_date += timedelta(days=1)
 
-        # Export yesterday's window (the normal daily run)
+        # Export yesterday's window (the normal daily run).
+        # Use `now` as end_time so spend rows flushed after midnight are included.
+        # LiteLLM's DailyUserSpend rows for a given date keep getting updated_at
+        # bumped as the flush job runs; capping at midnight would miss those updates.
         window = FocusTimeWindow(
             start_time=yesterday,
-            end_time=yesterday + timedelta(days=1),
+            end_time=now,
             frequency="daily",
         )
         await self._export_window(window=window, limit=None)
@@ -253,6 +273,21 @@ async def init_mavvrik_focus_background_job(
             )
             if type(cb) is MavvrikFocusLogger
         ]
+        if not loggers and "mavvrik" in litellm.callbacks:
+            # The logger is registered as the string "mavvrik" but hasn't been
+            # instantiated yet (lazy init happens on first LLM call). Force it now
+            # so the scheduler can register the daily export job at startup.
+            from litellm.litellm_core_utils.litellm_logging import (  # noqa: PLC0415
+                _init_custom_logger_compatible_class,
+            )
+
+            instance = _init_custom_logger_compatible_class(
+                logging_integration="mavvrik",
+                internal_usage_cache=None,
+                llm_router=None,
+            )
+            if isinstance(instance, MavvrikFocusLogger):
+                loggers = [instance]
         if not loggers:
             verbose_proxy_logger.debug(
                 "No MavvrikFocusLogger registered; skipping scheduler"