[WIP] Add durable API-cost outbox and canonical usage tracking

src/youtube_extension/services/ai/gemini_service.py

@@ -305,6 +305,7 @@ class GeminiResult:
     model_name: str
     backend: str  # "api" or "vertex"
     error: Optional[str] = None
+    usage_metadata: Optional[Any] = None
 
 
 class GeminiService:
@@ -588,7 +589,8 @@ async def process_image(
                 response=response.text,
                 latency=latency,
                 model_name=self.config.model_name,
-                backend="vertex" if self._use_vertex else "api"
+                backend="vertex" if self._use_vertex else "api",
+                usage_metadata=getattr(response, "usage_metadata", None),
             )
 
         except Exception as e:
@@ -676,6 +678,7 @@ async def process_text(
                 latency=time.time() - start_time,
                 model_name=self.config.model_name,
                 backend=self._backend_kind,
+                usage_metadata=getattr(response, "usage_metadata", None),
             )
 
         except Exception as exc:
@@ -825,7 +828,8 @@ async def process_video(
                 response=response.text,
                 latency=latency,
                 model_name=self.config.model_name,
-                backend="vertex" if self._use_vertex else "api"
+                backend="vertex" if self._use_vertex else "api",
+                usage_metadata=getattr(response, "usage_metadata", None),
             )
 
         except Exception as e:
@@ -892,6 +896,7 @@ async def process_audio(
                 latency=latency,
                 model_name=self.config.model_name,
                 backend="vertex" if self._use_vertex else "api",
+                usage_metadata=getattr(response, "usage_metadata", None),
             )
 
         except Exception as e:
@@ -1173,7 +1178,8 @@ async def process_youtube(
                 response=response.text,
                 latency=latency,
                 model_name=self.config.model_name,
-                backend="api"
+                backend="api",
+                usage_metadata=getattr(response, "usage_metadata", None),
             )
 
         except Exception as e:

src/youtube_extension/services/ai/hybrid_processor_service.py

@@ -26,6 +26,13 @@
 from .gemini_service import GeminiConfig, GeminiResult, GeminiService
 
 
+async def _record_api_usage(*args: Any, **kwargs: Any) -> Any:
+    """Load cost tracking only when provider usage is actually available."""
+    from youtube_extension.backend.services.api_cost_monitor import track_api_call
+
+    return await track_api_call(*args, **kwargs)
+
+
 class ProcessingMode(Enum):
     """Processing mode roadmap retained for compatibility."""
 
@@ -261,6 +268,11 @@ async def process(
                     **kwargs,
                 )
 
+            await self._track_gemini_usage(
+                cloud_result,
+                routing_decision.task_type,
+            )
+
             hybrid_result = HybridResult(
                 success=cloud_result.success,
                 response=cloud_result.response,
@@ -290,6 +302,46 @@ async def process(
                 error=str(exc),
             )
 
+    async def _track_gemini_usage(
+        self,
+        result: GeminiResult,
+        task_type: TaskType,
+    ) -> None:
+        """Persist provider-reported usage without delaying a paid result."""
+        if not result.success or result.backend not in {"api", "vertex", "gemini"}:
+            return
+
+        usage = result.usage_metadata
+        if usage is None:
+            self.logger.warning(
+                "Gemini response omitted usage metadata; cost record skipped"
+            )
+            return
+
+        input_tokens = int(getattr(usage, "prompt_token_count", 0) or 0)
+        output_tokens = int(getattr(usage, "candidates_token_count", 0) or 0)
+        output_tokens += int(getattr(usage, "thoughts_token_count", 0) or 0)
+        if input_tokens <= 0 and output_tokens <= 0:
+            self.logger.warning(
+                "Gemini usage metadata contained no billable token counts"
+            )
+            return
+
+        try:
+            await _record_api_usage(
+                "google",
+                "hybrid/process",
+                input_tokens,
+                model=result.model_name,
+                output_tokens=output_tokens,
+                request_type=task_type.value,
+                success=True,
+            )
+        except Exception:
+            self.logger.exception(
+                "Gemini usage tracking failed after provider completion"
+            )
+
     async def _call_gemini(
         self,
         input_data: str | Path | Image.Image,

tests/unit/test_api_cost_database_substrate.py

@@ -630,8 +630,9 @@ async def tracking_to_thread(
 
     await monitor.record_usage("openai", "/chat", 100, model="gpt-4o")
 
-    assert "_record_usage_sync" in calls
-    assert "_get_daily_cost_sync" in calls
+    # Usage persistence and UTC-day aggregation now share one worker-thread
+    # transaction; a second daily-cost query would reopen the crash boundary.
+    assert calls == ["_record_usage_sync"]
 
 
 async def test_telemetry_database_failure_does_not_fail_paid_api_result(

tests/unit/test_api_cost_monitor.py

@@ -89,12 +89,24 @@ def test_youtube_quota_cost(self, monitor):
     def test_unknown_service_returns_zero(self, monitor):
         assert monitor.calculate_cost("nonexistent", "model", 1000) == 0.0
 
-    def test_unknown_model_falls_back_to_first_model(self, monitor):
+    def test_unknown_model_fails_closed(self, monitor):
+        with pytest.raises(ValueError, match="Unknown pricing model"):
+            monitor.calculate_cost(
+                "anthropic", "unknown-model", input_tokens=1000, output_tokens=0
+            )
+
+    def test_default_model_preserves_legacy_service_costing(self, monitor):
         cost = monitor.calculate_cost(
-            "anthropic", "unknown-model", input_tokens=1000, output_tokens=0
+            "openai", "default", input_tokens=1000, output_tokens=0
         )
         assert cost > 0.0
 
+    def test_google_gemini_35_flash_cost(self, monitor):
+        cost = monitor.calculate_cost(
+            "google", "gemini-3.5-flash", input_tokens=1000, output_tokens=1000
+        )
+        assert pytest.approx(cost, rel=1e-6) == 0.0015 + 0.009
+
     def test_zero_tokens_returns_zero_cost(self, monitor):
         cost = monitor.calculate_cost(
             "anthropic", "claude-opus-4-8", input_tokens=0, output_tokens=0
@@ -345,6 +357,56 @@ async def test_record_failure_usage(self, monitor):
         assert record.success is False
         assert record.error_message == "rate limited"
 
+    async def test_usage_and_crossed_alert_commit_atomically(self, monitor):
+        from youtube_extension.backend.models.api_cost import (
+            APIUsage,
+            DailyBudget,
+            WebhookOutbox,
+        )
+
+        monitor.alert_threshold = 0.001
+        monitor.daily_budget = 100.0
+        record = await monitor.record_usage(
+            service="anthropic",
+            endpoint="/messages",
+            tokens_used=1000,
+            model="claude-opus-4-8",
+        )
+
+        with monitor._session_scope() as session:
+            assert session.query(APIUsage).count() == 1
+            budget = session.query(DailyBudget).one()
+            alert = session.query(WebhookOutbox).one()
+
+            assert budget.total_cost == pytest.approx(record.cost)
+            assert budget.alert_sent is True
+            assert alert.alert_type == "threshold"
+            assert alert.current_cost == pytest.approx(record.cost)
+
+    async def test_alert_staging_failure_rolls_back_usage(self, monitor, monkeypatch):
+        from youtube_extension.backend.models.api_cost import (
+            APIUsage,
+            DailyBudget,
+            WebhookOutbox,
+        )
+
+        def fail_staging(session, timestamp):
+            raise RuntimeError("simulated crash boundary")
+
+        monkeypatch.setattr(monitor, "_stage_budget_alerts", fail_staging)
+        record = await monitor.record_usage(
+            service="anthropic",
+            endpoint="/messages",
+            tokens_used=1000,
+            model="claude-opus-4-8",
+        )
+
+        assert record is not None
+        with monitor._session_scope() as session:
+            assert session.query(APIUsage).count() == 0
+            assert session.query(DailyBudget).count() == 0
+            assert session.query(WebhookOutbox).count() == 0
+
 
 # ===========================================================================
 # APICostMonitor — get_daily_cost
@@ -740,7 +802,7 @@ async def fake_notification(message):
 
         # Attempt 2, 3, 4, 5
         for expected_retry in [2, 3, 4, 5]:
-            await monitor.process_outbox()
+            await monitor.process_outbox(force=True)
             session = monitor.Session()
             try:
                 item = (
@@ -754,7 +816,7 @@ async def fake_notification(message):
                 session.close()
 
         # Attempt 6 (should not be retried because retry count reached 5)
-        await monitor.process_outbox()
+        await monitor.process_outbox(force=True)
         session = monitor.Session()
         try:
             item = (