feat(observability): 接入 Prometheus 指标与 OpenTelemetry/Tempo 链路 (#360)

* feat(observability): 接入 Prometheus 指标与 OpenTelemetry/Tempo 链路

HTTP/WebSocket/Agent/外部 SDK 只打点和打 span，方便定位一轮对话卡在 ASR、LLM 还是 TTS；Grafana 只负责把数据画出来。

* feat(observability): 记录语音会话占用阶段与挂断原因

Grafana 需要当前卡在 ASR/LLM/TTS 的人数，以及 UI 挂断、超时、道别的区分；session_id 只留进程内，不进 Prometheus label。

* feat(observability): 在 Tempo span 上记录 APK 登录用户名

Grafana 按登录页账号筛 turn/tool 轨迹；username 只作为 span 属性，不进 Prometheus label。

* feat(observability): 把 Grafana 看板纳入版本库，展示 APK 登录账号

此前整棵 /observability/ 被 ignore，登录账号筛选只留在本机 Grafana。把看板 JSON 纳入仓库，Tempo 表选出 timeflow.username。

* style: 按 ruff 折行 Qwen Audio 连续模式测试断言

合入 main 后该行超长，CI 的 ruff format --check 失败。

* fix(observability): 保留 VOICE_TELEMETRY 具体类型以绑定用户名查找

mypy 将 Protocol 标注视为对外类型，导致 composition root 无法调用 bind_username_lookup。

* test(observability): 补齐会话占用、用户名查找与错误路径覆盖

覆盖 composition root 的账户查找、NoOp 占用接口，以及握手/播放/数据库监听的边界分支，把 patch 覆盖率抬过 Codecov 门槛。

* fix(observability): 用户名查找失败时回退 unknown，不打断语音回合

数据库短暂不可用时记录脱敏诊断，span 使用 unknown 且不写入缓存，恢复后下一轮仍可补上登录名。

* feat(observability): 用 Compose 在同一台机器部署 Grafana/Prometheus/Tempo

配置改为容器路径和 Docker 服务名，去掉本机绝对路径，便于和 API 一起上云。

Author: gac0812

.env.example

@@ -1,11 +1,34 @@
 POSTGRES_DB=timeapp
 POSTGRES_USER=timeapp
 POSTGRES_PASSWORD=timeapp
+POSTGRES_HOST=0.0.0.0
 POSTGRES_PORT=5432
+API_HOST=0.0.0.0
 API_PORT=8000
 TIMEFLOW_ENVIRONMENT=development
 # Generate a private value of at least 32 UTF-8 bytes before starting the API.
 TIMEFLOW_JWT_SECRET=
 # 访问令牌有效期（秒），默认一个月（30 天）。
 TIMEFLOW_JWT_ACCESS_TTL_SECONDS=2592000
 TIMEFLOW_CORS_ALLOWED_ORIGINS=http://localhost:8081,http://127.0.0.1:8081
+
+# 云上同一台机器跑 API 时填真实语音密钥；非 development 必须配置，否则 /ws 拒绝启动。
+TIMEFLOW_VOICE_AGENT_MODE=1
+TIMEFLOW_ALIYUN_AUDIO_API_KEY=
+TIMEFLOW_ALIYUN_AUDIO_WORKSPACE_ID=
+TIMEFLOW_OPENAI_BASE_URL=
+TIMEFLOW_OPENAI_API_KEY=
+TIMEFLOW_OPENAI_MODEL=qwen3.6-flash
+TIMEFLOW_TENCENT_MAP_KEY=
+
+# 同一台机器跑 Grafana / Prometheus / Tempo 时：
+# docker compose -f docker-compose.yml -f docker-compose.observability.yml up -d --build
+# 云上建议：API_HOST=127.0.0.1  POSTGRES_HOST=127.0.0.1  TIMEFLOW_ENVIRONMENT=production
+GRAFANA_HOST=127.0.0.1
+GRAFANA_PORT=3000
+GRAFANA_ADMIN_USER=admin
+GRAFANA_ADMIN_PASSWORD=
+GRAFANA_ROOT_URL=http://127.0.0.1:3000
+SENTRY_AUTH_TOKEN=
+TIMEFLOW_OTEL_SERVICE_NAME=timeflow-backend
+TIMEFLOW_OTEL_EXPORTER_OTLP_ENDPOINT=

.gitignore

@@ -2,6 +2,13 @@
 .idea/
 .cursor/
 scripts/
+# Local Grafana/Prometheus/Tempo install (binaries, WAL, plugins). Config is tracked.
+/observability/data/
+/observability/downloads/
+/observability/grafana/
+/observability/prometheus/prometheus
+/observability/prometheus/promtool
+/observability/tempo/tempo
 .env
 .venv/
 node_modules/

backend/.env.example

@@ -83,3 +83,7 @@ TIMEFLOW_TENCENT_MAP_TIMEOUT_SECONDS=5
 # Which voice agent backend /ws uses: "1" = end-to-end realtime model (above),
 # "2" = ASR + LLM + TTS conversation pipeline (requires ASR, OpenAI-compatible LLM, and TTS).
 TIMEFLOW_VOICE_AGENT_MODE=1
+
+# Optional Tempo export. Empty keeps traces off; Prometheus still scrapes GET /metrics.
+TIMEFLOW_OTEL_SERVICE_NAME=timeflow-backend
+TIMEFLOW_OTEL_EXPORTER_OTLP_ENDPOINT=

backend/pyproject.toml

@@ -22,6 +22,10 @@ dependencies = [
     "sqlalchemy>=2.0,<3",
     "uvicorn>=0.51,<1",
     "websockets>=15,<16",
+    "opentelemetry-api>=1.27,<2",
+    "opentelemetry-exporter-otlp-proto-http>=1.27,<2",
+    "opentelemetry-sdk>=1.27,<2",
+    "prometheus-client>=0.21,<1",
 ]
 
 [dependency-groups]

backend/src/timeflow/composition.py

@@ -21,6 +21,7 @@
 from timeflow.intelligence.location import ClientLocation, LocationSearchService
 from timeflow.intelligence.ports import ResultSink
 from timeflow.intelligence.schedule_category import LlmScheduleCategoryClassifier
+from timeflow.observability import VOICE_TELEMETRY
 
 
 def build_composed_voice_agent(
@@ -63,6 +64,7 @@ def agent_factory(
                 client_location=client_location,
             ),
             max_tool_rounds=settings.agent_max_tool_rounds,
+            telemetry=VOICE_TELEMETRY,
         )
 
     return ComposedVoiceAgent(
@@ -71,6 +73,7 @@ def agent_factory(
         QwenAudioTts(settings),
         result_sink,
         location_service=location_service,
+        telemetry=VOICE_TELEMETRY,
     )
 
 

backend/src/timeflow/data/database.py

@@ -1,16 +1,29 @@
 """SQLAlchemy primitives for persistence adapters."""
 
-from sqlalchemy import Engine, create_engine
+from sqlalchemy import Engine, create_engine, text
 from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker
 
+from timeflow.infrastructure.observability.database import instrument_engine
+
 
 class Base(DeclarativeBase):
     """Declarative base for TimeFlow-owned database models."""
 
 
 def build_engine(database_url: str) -> Engine:
     """Create a database engine without opening a connection eagerly."""
-    return create_engine(database_url, pool_pre_ping=True, hide_parameters=True)
+    engine = create_engine(database_url, pool_pre_ping=True, hide_parameters=True)
+    return instrument_engine(engine)
+
+
+def ping_database(engine: Engine) -> bool:
+    """Return whether the engine can execute a trivial readiness query."""
+    try:
+        with engine.connect() as connection:
+            connection.execute(text("SELECT 1"))
+        return True
+    except Exception:
+        return False
 
 
 def build_session_factory(engine: Engine) -> sessionmaker[Session]:

backend/src/timeflow/data/repositories/account.py

@@ -23,6 +23,10 @@ def get_by_username(self, username: str) -> AccountRecord | None:
         model = self._require_session().scalar(select(Account).where(Account.username == username))
         return None if model is None else _to_record(model)
 
+    def get_by_id(self, account_id: str) -> AccountRecord | None:
+        model = self._require_session().get(Account, account_id)
+        return None if model is None else _to_record(model)
+
     def add(self, account: NewAccount) -> AccountRecord:
         session = self._require_session()
         model = Account(

backend/src/timeflow/gateway/http/auth.py

@@ -1,6 +1,7 @@
 """账户访问与稳定认证错误的 HTTP 适配器。"""
 
 import logging
+import time
 from collections.abc import Callable, Coroutine
 from dataclasses import dataclass
 from typing import Any, Protocol
@@ -15,6 +16,7 @@
 from timeflow.business.auth import AuthAccessResult, AuthError, AuthErrorCode
 from timeflow.gateway.auth_diagnostics import log_sanitized_exception
 from timeflow.gateway.http.rate_limit import AuthRateLimiter
+from timeflow.gateway.observability.http import record_auth_access
 
 logger = logging.getLogger(__name__)
 
@@ -180,9 +182,11 @@ def get_route_handler(
         original_handler = super().get_route_handler()
 
         async def route_handler(request: Request) -> Response:
+            started = time.perf_counter()
             try:
                 return await original_handler(request)
             except RequestValidationError as error:
+                record_auth_access("failure", time.perf_counter() - started)
                 error_code = _single_invalid_request_field(error)
                 if error_code is None:
                     return _uncontracted_validation_response()
@@ -213,17 +217,21 @@ def access(
         http_request: Request,
         request: AuthAccessRequest,
     ) -> AuthAccessResponse | Response:
+        started = time.perf_counter()
         client = http_request.client
         client_key = client.host if client is not None else "unknown"
         if not limiter.allow(client_key):
+            record_auth_access("rate_limited", time.perf_counter() - started)
             return _error_response(
                 AUTH_RATE_LIMITED,
                 retry_after_seconds=limiter.retry_after_seconds,
             )
         try:
             result = auth_access.access(request.username, request.password)
         except Exception as error:
+            record_auth_access("failure", time.perf_counter() - started)
             return _service_error_response(error)
+        record_auth_access("success", time.perf_counter() - started)
         return AuthAccessResponse(
             account_id=result.account_id,
             access_token=result.access_token,

backend/src/timeflow/gateway/http/reminder_state.py

@@ -1,6 +1,7 @@
 """HTTP adapter for syncing a schedule reminder's final confirmed state."""
 
 import logging
+import time
 from collections.abc import Callable, Coroutine
 from datetime import UTC, datetime
 from typing import Annotated, Any, Literal
@@ -24,6 +25,7 @@
     AuthenticatedAccount,
     AuthenticatedAccountDependency,
 )
+from timeflow.gateway.observability.http import record_reminder_state
 
 logger = logging.getLogger(__name__)
 
@@ -91,9 +93,11 @@ def get_route_handler(
         original_handler = super().get_route_handler()
 
         async def route_handler(request: Request) -> Response:
+            started = time.perf_counter()
             try:
                 return await original_handler(request)
             except RequestValidationError:
+                record_reminder_state("invalid", time.perf_counter() - started)
                 response = ReminderStateInvalidRequestResponse()
                 return JSONResponse(
                     status_code=422,
@@ -135,6 +139,7 @@ def confirm_reminder_state(
         request: ReminderStateRequest,
         account: Annotated[AuthenticatedAccount, Security(trusted_account)],
     ) -> Response:
+        started = time.perf_counter()
         try:
             result = confirmer.confirm(
                 account_id=account.account_id,
@@ -145,18 +150,23 @@ def confirm_reminder_state(
                 disposition_state=result.disposition_state,
                 updated_at=_as_utc(result.updated_at),
             )
+            record_reminder_state("confirmed", time.perf_counter() - started)
             return JSONResponse(status_code=200, content=response.model_dump(mode="json"))
         except ScheduleBusinessError as error:
             if error.code is ScheduleErrorCode.SCHEDULE_NOT_FOUND:
+                record_reminder_state("not_found", time.perf_counter() - started)
                 return _error_response(404, "SCHEDULE_NOT_FOUND", "Schedule not found")
             if error.code is ScheduleErrorCode.REMINDER_NOT_CONFIGURED:
+                record_reminder_state("conflict", time.perf_counter() - started)
                 return _error_response(
                     409,
                     "REMINDER_NOT_CONFIGURED",
                     "Reminder not configured",
                 )
+            record_reminder_state("error", time.perf_counter() - started)
             return _internal_error_response(error)
         except Exception as error:
+            record_reminder_state("error", time.perf_counter() - started)
             return _internal_error_response(error)
 
     return router

backend/src/timeflow/gateway/http/schedule_snapshot.py

@@ -1,6 +1,7 @@
 """受保护的账号日程全量快照 HTTP 适配器。"""
 
 import logging
+import time
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, Request, Security
@@ -27,6 +28,7 @@
     AuthenticatedAccount,
     AuthenticatedAccountDependency,
 )
+from timeflow.gateway.observability.http import record_schedule_snapshot
 
 logger = logging.getLogger(__name__)
 
@@ -203,11 +205,14 @@ def authenticate_snapshot_account(
     def get_schedule_snapshot(
         account: Annotated[AuthenticatedAccount, Depends(authenticate_snapshot_account)],
     ) -> JSONResponse:
+        started = time.perf_counter()
         try:
             snapshot = reader.get_account_snapshot(account_id=account.account_id)
             response = _build_response(account.account_id, snapshot)
+            record_schedule_snapshot("success", time.perf_counter() - started, server_error=False)
             return JSONResponse(status_code=200, content=response.model_dump(mode="json"))
         except Exception as error:
+            record_schedule_snapshot("error", time.perf_counter() - started, server_error=True)
             return _internal_error_response(error)
 
     return router