fix(observability): persist model usage on unattended runs

backend/app/api/chat.py

@@ -51,6 +51,7 @@
     open_harness_artifact,
 )
 from app.llm import LLMClient, LLMError
+from app.observability import persist_spans
 from app.observability.spans import (
     bind_span_sink,
     llm_operation,
@@ -557,7 +558,12 @@ def _resume_human_handoff_worker(handoff_id: str) -> None:
                 channel="human_handoff_resume",
                 debug=False,
             )
-            AgentLoop(db).handle_turn(request)
+            with persist_spans(
+                db,
+                tenant_id=handoff.tenant_id,
+                session_id=handoff.session_id,
+            ):
+                AgentLoop(db).handle_turn(request)
             # resume turn 完成后不再写 resume_finished_at 标记:
             # _inject_handoff_context 已改为用 request.channel == "human_handoff_resume"
             # 判定 resume turn,时序可靠,无需事后标记。

backend/app/api/teams.py

@@ -14,6 +14,7 @@
 from app.async_jobs import enqueue_async_job
 from app.core import AgentLoop
 from app.db import get_session
+from app.observability import persist_spans
 from app.db.models import (
     AgentEvent,
     AgentProfile,
@@ -418,7 +419,13 @@ def tl_chat_endpoint(
         channel="team",
         interaction_mode="team_tl",
     )
-    response = AgentLoop(db).handle_turn(turn)
+    with persist_spans(
+        db,
+        tenant_id=team.tenant_id,
+        session_id=session.id,
+        client_turn_id=turn.client_turn_id,
+    ):
+        response = AgentLoop(db).handle_turn(turn)
     reply = response.reply or ""
     created = process_tl_reply(
         db,

backend/app/observability/__init__.py

@@ -1,4 +1,3 @@
-from app.observability.event_log import EventLog
-
-__all__ = ["EventLog"]
+from app.observability.event_log import EventLog, persist_spans
 
+__all__ = ["EventLog", "persist_spans"]

backend/app/observability/event_log.py

@@ -1,12 +1,14 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Callable
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager
 from typing import Any
 
 from sqlmodel import Session
 
 from app.db.models import AgentEvent
+from app.observability.spans import bind_span_sink
 
 logger = logging.getLogger(__name__)
 
@@ -47,3 +49,92 @@ def record(self, tenant_id: str, session_id: str, event_type: str, payload: dict
             except Exception:
                 logger.exception("event_sink 调用失败 event_type=%s", event_type)
         return event
+_LLM_SPAN_METRIC_FIELDS = frozenset(
+    {
+        "span_id",
+        "parent_span_id",
+        "operation",
+        "turn_id",
+        "user_message_id",
+        "client_turn_id",
+        "task_frame_id",
+        "iteration",
+        "started_at",
+        "finished_at",
+        "duration_ms",
+        "ttft_ms",
+        "provider_setup_ms",
+        "stream_duration_ms",
+        "model",
+        "model_name",
+        "endpoint",
+        "request_kind",
+        "stream",
+        "thinking_mode",
+        "max_output_tokens",
+        "response_mode",
+        "request_parameters",
+        "request_message_roles",
+        "request_prefix_fingerprints",
+        "provider_response_id",
+        "attempt",
+        "retry_count",
+        "max_attempts",
+        "json_attempt",
+        "json_max_attempts",
+        "json_retry_count",
+        "context_message_count",
+        "context_text_chars",
+        "payload_chars",
+        "request_message_chars",
+        "request_message_count",
+        "request_text_chars",
+        "system_prompt_chars",
+        "input_tokens",
+        "output_tokens",
+        "total_tokens",
+        "cached_input_tokens",
+        "uncached_input_tokens",
+        "status",
+        "finish_reason",
+        "error_type",
+        "error",
+        "output_chars",
+        "reasoning_chars",
+        "stream_chunks",
+    }
+)
+def _metrics_only(event_type: str, payload: dict[str, Any]) -> dict[str, Any]:
+    """llm_call 事件只保留标量指标;其他 span 原样返回。
+
+    裁剪过的行标记 bodies_omitted:审计视图对缺失的原文会渲染成空字符串与
+    空数组,与"模型确实返回了空内容"无法区分。留一个显式标记,排障时才不会
+    把"未留存"误读成"返回为空"。
+    """
+    if not event_type.startswith("llm_call_"):
+        return payload
+    kept = {key: value for key, value in payload.items() if key in _LLM_SPAN_METRIC_FIELDS}
+    if len(kept) != len(payload):
+        kept["bodies_omitted"] = True
+    return kept
+@contextmanager
+def persist_spans(
+    db: Session,
+    *,
+    tenant_id: str,
+    session_id: str,
+    client_turn_id: str | None = None,
+) -> Iterator[None]:
+    tenant_id = str(tenant_id or "").strip()
+    session_id = str(session_id or "").strip()
+    if not tenant_id or not session_id:
+        yield
+        return
+    event_log = EventLog(db)
+    if client_turn_id:
+        event_log.bind_turn("", client_turn_id)
+    def sink(event_type: str, payload: dict[str, Any]) -> None:
+        event_log.record(tenant_id, session_id, event_type, _metrics_only(event_type, payload))
+        db.commit()
+    with bind_span_sink(sink):
+        yield

backend/app/public_api/runs.py

@@ -15,6 +15,7 @@
 from app.core.cancellation import cancel_chat_turn
 from app.core.harness_session_cleanup import harness_task_workspace_path
 from app.db import engine, get_session
+from app.observability import persist_spans
 from app.db.models import (
     APIClient,
     APICredential,
@@ -204,7 +205,12 @@ def execute_run(db: Session, job: APIJob) -> dict[str, Any]:
 
     def execute_harness() -> None:
         try:
-            with Session(engine) as worker_db:
+            with Session(engine) as worker_db, persist_spans(
+                worker_db,
+                tenant_id=job.tenant_id,
+                session_id=session_id,
+                client_turn_id=job.id,
+            ):
                 for item in AgentLoop(worker_db).handle_turn_stream(request):
                     if item.get("event") != "complete":
                         continue

backend/app/scheduled_tasks/service.py

@@ -34,6 +34,7 @@
 )
 from app.llm import LLMClient, LLMError
 from app.observability.spans import llm_operation
+from app.observability import persist_spans
 from app.scheduled_tasks.schema import (
     ScheduledTaskCreateRequest,
     ScheduledTaskDraftRead,
@@ -545,10 +546,16 @@ def _execute_prepared_scheduled_task(
             client_timezone=task.timezone,
         )
         result: ChatTurnResponse | None = None
-        for seq, item in enumerate(AgentLoop(db).handle_turn_stream(request), start=1):
-            _record_scheduled_task_stream_event(db, run, run.session_id, seq, item)
-            if item.get("event") in {"complete", "done"} and isinstance(item.get("data"), dict):
-                result = ChatTurnResponse.model_validate(item["data"])
+        with persist_spans(
+            db,
+            tenant_id=task.tenant_id,
+            session_id=run.session_id,
+            client_turn_id=run.id,
+        ):
+            for seq, item in enumerate(AgentLoop(db).handle_turn_stream(request), start=1):
+                _record_scheduled_task_stream_event(db, run, run.session_id, seq, item)
+                if item.get("event") in {"complete", "done"} and isinstance(item.get("data"), dict):
+                    result = ChatTurnResponse.model_validate(item["data"])
         if result is None:
             raise RuntimeError("自动任务执行未返回完整结果")
         outcome = _scheduled_harness_outcome(db, run, result)

backend/app/teams/wakeup.py

@@ -28,6 +28,7 @@
     new_id,
     utc_now,
 )
+from app.observability import persist_spans
 from app.session.session_schema import (
     ChatTurnRequest,
     ChatTurnResponse,
@@ -755,9 +756,15 @@ def run_agent_turn(
         message_visibility=message_visibility,
     )
     result: ChatTurnResponse | None = None
-    for item in AgentLoop(db).handle_turn_stream(request):
-        if item.get("event") in {"complete", "done"} and isinstance(item.get("data"), dict):
-            result = ChatTurnResponse.model_validate(item["data"])
+    with persist_spans(
+        db,
+        tenant_id=team.tenant_id,
+        session_id=session_id,
+        client_turn_id=turn_id,
+    ):
+        for item in AgentLoop(db).handle_turn_stream(request):
+            if item.get("event") in {"complete", "done"} and isinstance(item.get("data"), dict):
+                result = ChatTurnResponse.model_validate(item["data"])
     if result is None:
         raise RuntimeError("团队唤醒执行未返回完整结果")
     outcome = _team_harness_outcome(