feat: wire agent runtime metrics

app/agent/orchestrator.py

@@ -3,6 +3,7 @@ import hashlib
 import inspect
 import json
 import re
+import time
 import traceback
 import uuid
 import warnings
@@ -69,6 +70,7 @@ from app.agent.tools.impl.query_system_settings import QuerySystemSettingsTool
 from app.chain.agent import AgentChain
 from app.runtime.config import settings
 from app.runtime.events import eventmanager
+from app.runtime.observability import record_metric
 from app.application.plugin.runtime import get_plugin_manager
 
 
@@ -90,6 +92,37 @@ from app.foundation.identity import SYSTEM_INTERNAL_USER_ID
 warnings.filterwarnings("ignore", message=".*allowed_objects.*")
 
 
+_KNOWN_AGENT_PROVIDER_TYPES = (
+    "anthropic",
+    "azure",
+    "deepseek",
+    "gemini",
+    "ollama",
+    "openai",
+)
+
+
+def _agent_provider_metric_type(provider: object) -> str:
+    """把可配置 provider 名称收敛为有限指标类别，避免泄露自定义名称。"""
+    normalized = str(provider or "").strip().lower()
+    if not normalized:
+        return "unknown"
+    for provider_type in _KNOWN_AGENT_PROVIDER_TYPES:
+        if provider_type in normalized:
+            return provider_type
+    return "custom"
+
+
+def _agent_task_metric_type(source: object, channel: object = None) -> str:
+    """把 Agent 来源归一为交互、调度、后台三类稳定标签。"""
+    normalized = str(source or "").strip().lower()
+    if normalized in {"scheduler", "scheduled", "heartbeat", "agent_task"}:
+        return "scheduled"
+    if channel:
+        return "interactive"
+    return "background"
+
+
 def _finish_processing_status(status: Optional[dict], user_id: Optional[str] = None) -> None:
     """结束入站消息的渠道处理状态。"""
     if not status:
@@ -707,6 +740,24 @@ class MoviePilotAgent:
         self._session_usage.total_cache_write_input_tokens += cache_write_input_tokens
         self._session_usage.total_uncached_input_tokens += uncached_input_tokens
         self._session_usage.cache_usage_available |= cache_usage_available
+        provider_type = _agent_provider_metric_type(
+            (self._llm_provider_selection or {}).get("provider")
+            or settings.LLM_PROVIDER
+        )
+        if input_tokens:
+            record_metric(
+                "agent.token_usage",
+                input_tokens,
+                provider_type=provider_type,
+                direction="input",
+            )
+        if output_tokens:
+            record_metric(
+                "agent.token_usage",
+                output_tokens,
+                provider_type=provider_type,
+                direction="output",
+            )
 
         if not is_current_request:
             return
@@ -2279,6 +2330,7 @@ class MoviePilotAgent:
         """
         execution_success = False
         execution_error: Optional[str] = None
+        metric_started_at = time.perf_counter()
         self._agent_started_at = datetime.now()
         self._llm_runtime_config = None
         self._llm_provider_selection = {}
@@ -2435,6 +2487,15 @@ class MoviePilotAgent:
             await self._dispatch_execution_notice(friendly_message)
             return friendly_message, {}
         finally:
+            selection = self._llm_provider_selection or {}
+            record_metric(
+                "agent.provider.duration",
+                time.perf_counter() - metric_started_at,
+                provider_type=_agent_provider_metric_type(
+                    selection.get("provider") or settings.LLM_PROVIDER
+                ),
+                outcome="success" if execution_success else "error",
+            )
             self._send_agent_tokens_usage_event(
                 success=execution_success,
                 error=execution_error,
@@ -2885,6 +2946,8 @@ class AgentManager:
                     logger.debug(f"会话 {session_id} 的消息队列空闲，worker退出")
                     break
 
+                task_type = _agent_task_metric_type(task.source, task.channel)
+                active_metric_recorded = False
                 try:
                     if task.enqueued_at is not None:
                         queue_wait_ms = max(
@@ -2900,6 +2963,12 @@ class AgentManager:
                             3,
                         )
                     await self._start_task_processing_status(task)
+                    record_metric(
+                        "agent.active_tasks",
+                        1,
+                        task_type=task_type,
+                    )
+                    active_metric_recorded = True
                     result = await self._process_message_internal(task)
                     if task.completion_future and not task.completion_future.done():
                         if (
@@ -2923,6 +2992,12 @@ class AgentManager:
                     if task.completion_future and not task.completion_future.done():
                         task.completion_future.set_exception(e)
                 finally:
+                    if active_metric_recorded:
+                        record_metric(
+                            "agent.active_tasks",
+                            -1,
+                            task_type=task_type,
+                        )
                     await self._finish_task_processing_status(task)
                     queue.task_done()
                 if session_id in self._session_cancel_requested:
@@ -3060,6 +3135,15 @@ class AgentManager:
     async def _stop_current_task_locked(self, session_id: str):
         """在 lifecycle 互斥域内停止会话 worker。"""
         stopped = False
+        active_agent = self.active_agents.get(session_id)
+        task_type = (
+            _agent_task_metric_type(
+                getattr(active_agent, "source", None),
+                getattr(active_agent, "channel", None),
+            )
+            if active_agent
+            else "unknown"
+        )
 
         worker = self._session_workers.get(session_id)
         queue = self._session_queues.get(session_id)
@@ -3101,6 +3185,12 @@ class AgentManager:
         else:
             logger.debug(f"会话 {session_id} 没有正在执行的Agent任务")
 
+        record_metric(
+            "agent.cancel",
+            task_type=task_type,
+            outcome="stopped" if stopped else "not_found",
+        )
+
         return stopped
 
     async def clear_session(self, session_id: str, user_id: str):

docs/refactor/backend-architecture-next-stage.md

@@ -866,6 +866,10 @@ OTel 初始化只能位于 Startup/Adapter；Domain/Application 只依赖 no-op-
 - `app.adapters.observability.otel` 只在组合根显式读取 `MOVIEPILOT_OTEL_METRICS=1` 后懒加载 OTel API；
   未安装可选包时稳定回退 no-op，不给核心层增加 SDK 依赖。HTTP Adapter 通过路由匹配输出模板，绝不以原始
   request path 充当 label。
+- 2026-08-22 扩展接线覆盖 SQLAlchemy checkout/checkin、异步回退配额 wait/timeout、Module 真实
+  `TimeoutError`、插件 start/initialize/stop/reload，以及 Agent 活跃任务、取消结果、供应商耗时和输入/
+  输出 token。自定义 Agent provider 统一归类为 `custom`，不会暴露配置名称；Scheduler retry/dead-letter
+  属于本轮明确暂停的 Outbox worker 范围，目录合同保留但不在本轮接线。
 - 专项测试覆盖 exporter 缺失、非法标签、全目录高基数审计、成功/失败 outcome、动态 URL 路由模板；
   既有 API、Event、Module、Scheduler 与健康探针回归保持通过。
 

tests/test_agent_tokens_events.py

@@ -94,6 +94,36 @@ def test_initialize_llm_uses_chain_event_selection(monkeypatch) -> None:
     assert agent._llm_provider_selection["selected_provider_id"] == "provider-1"
 
 
+def test_record_usage_emits_low_cardinality_token_metrics() -> None:
+    """Token 指标只暴露归一供应商类别与输入输出方向。"""
+    agent = MoviePilotAgent(session_id="usage-metrics", user_id="user-1")
+    agent._llm_provider_selection = {"provider": "private-provider-name"}
+
+    with patch("app.agent.orchestrator.record_metric") as record_metric:
+        agent._record_usage(
+            {
+                "has_usage": True,
+                "input_usage_available": True,
+                "input_tokens": 120,
+                "output_tokens": 30,
+                "total_tokens": 150,
+            }
+        )
+
+    record_metric.assert_any_call(
+        "agent.token_usage",
+        120,
+        provider_type="custom",
+        direction="input",
+    )
+    record_metric.assert_any_call(
+        "agent.token_usage",
+        30,
+        provider_type="custom",
+        direction="output",
+    )
+
+
 def test_execute_agent_broadcasts_usage_on_success() -> None:
     """Agent 执行成功后应广播聚合 token 用量事件。"""
     agent = MoviePilotAgent(session_id="usage-success", user_id="user-1")

tests/test_async_db_pooling.py

@@ -14,6 +14,7 @@ Connection 与 aiosqlite 的线程都绑定在创建它的循环上。因此池
 """
 import asyncio
 import threading
+from unittest.mock import patch
 
 import pytest
 
@@ -163,8 +164,16 @@ def test_fallback_slot_times_out_when_exhausted(monkeypatch):
 
     async def run():
         db_module._fallback_slots.acquire()  # 占满唯一名额
-        with pytest.raises(TimeoutError):
+        with patch("app.db.session.record_metric") as record_metric:
-            await db_module._acquire_fallback_slot()
+            with pytest.raises(TimeoutError):
+                await db_module._acquire_fallback_slot()
+        record_metric.assert_any_call("db.pool.timeout", backend="sqlite")
+        record_metric.assert_any_call(
+            "db.pool.wait",
+            pytest.approx(0.05, abs=0.03),
+            backend="sqlite",
+            outcome="timeout",
+        )
 
     asyncio.run(run())
 

tests/test_observability.py

@@ -5,12 +5,16 @@ from typing import Mapping
 
 import httpx
 import pytest
+from sqlalchemy import create_engine
 from starlette.applications import Starlette
 from starlette.responses import PlainTextResponse
 from starlette.routing import Route
 
 from app.adapters.observability import otel
 from app.adapters.web.metrics import HttpMetricsMiddleware
+from app.db.engine import _register_database_pool_metrics
+from app.runtime.extensions.plugin.lifecycle import observe_plugin_lifecycle
+from app.schemas.plugin import PluginRuntimeStatus
 from app.runtime.observability import (
     METRIC_SPECS,
     MetricSpec,
@@ -137,3 +141,36 @@ def test_optional_otel_adapter_falls_back_to_noop(monkeypatch) -> None:
     monkeypatch.setattr(otel.importlib, "import_module", missing)
 
     assert isinstance(otel.build_observation_port(), NoopObservationPort)
+
+
+def test_database_pool_checkout_updates_gauge() -> None:
+    """真实 SQLAlchemy checkout/checkin 应成对维护连接借出量。"""
+    port = RecordingObservationPort()
+    configure_observation(port)
+    engine = create_engine("sqlite://")
+    _register_database_pool_metrics(engine)
+
+    with engine.connect():
+        pass
+    engine.dispose()
+
+    records = [record for record in port.records if record[0].name == "db.pool.checked_out"]
+    assert [record[1] for record in records] == [1.0, -1.0]
+    assert all(record[2] == {"backend": "sqlite"} for record in records)
+
+
+def test_plugin_lifecycle_failed_status_records_error_outcome() -> None:
+    """被插件生命周期内部收敛的加载失败仍应记录 error。"""
+    port = RecordingObservationPort()
+    configure_observation(port)
+
+    @observe_plugin_lifecycle("start")
+    def load_plugin() -> dict[str, PluginRuntimeStatus]:
+        """返回插件加载失败状态。"""
+        return {"Example": PluginRuntimeStatus.LOAD_FAILED}
+
+    load_plugin()
+
+    spec, _, labels = port.records[-1]
+    assert spec.name == "plugin.lifecycle.duration"
+    assert labels == {"operation": "start", "outcome": "error"}