diff --git a/backend/app/api/chat.py b/backend/app/api/chat.py
index 8c18d1e32..a84c3147e 100644
--- a/backend/app/api/chat.py
+++ b/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,时序可靠,无需事后标记。
diff --git a/backend/app/api/teams.py b/backend/app/api/teams.py
index b875f95ac..b97f9f7de 100644
--- a/backend/app/api/teams.py
+++ b/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,
diff --git a/backend/app/observability/__init__.py b/backend/app/observability/__init__.py
index 28fbb36df..8db6d2569 100644
--- a/backend/app/observability/__init__.py
+++ b/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"]
diff --git a/backend/app/observability/event_log.py b/backend/app/observability/event_log.py
index fcc5c2cf3..2d8c9d008 100644
--- a/backend/app/observability/event_log.py
+++ b/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
diff --git a/backend/app/public_api/runs.py b/backend/app/public_api/runs.py
index a6facfb1a..bc3c7b705 100644
--- a/backend/app/public_api/runs.py
+++ b/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
diff --git a/backend/app/scheduled_tasks/service.py b/backend/app/scheduled_tasks/service.py
index 6f61a1349..b0136ae57 100644
--- a/backend/app/scheduled_tasks/service.py
+++ b/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)
diff --git a/backend/app/teams/wakeup.py b/backend/app/teams/wakeup.py
index 7b13a8f0d..203de5dfc 100644
--- a/backend/app/teams/wakeup.py
+++ b/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(
diff --git a/backend/tests/test_observability_span_persistence.py b/backend/tests/test_observability_span_persistence.py
new file mode 100644
index 000000000..d7cab4673
--- /dev/null
+++ b/backend/tests/test_observability_span_persistence.py
@@ -0,0 +1,269 @@
+from __future__ import annotations
+
+import json
+
+import pytest
+from sqlalchemy.pool import StaticPool
+from sqlmodel import Session, SQLModel, create_engine, select
+
+from app.db.models import AgentEvent, Tenant
+from app.observability import EventLog, persist_spans
+from app.observability.spans import emit_span_event, start_llm_call
+
+USAGE = {
+    "input_tokens": 212,
+    "output_tokens": 10,
+    "total_tokens": 222,
+    "cached_input_tokens": 154,
+}
+
+def _engine():
+    engine = create_engine(
+        "sqlite://",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+    SQLModel.metadata.create_all(engine)
+    return engine
+
+def _seed(db: Session) -> None:
+    db.add(Tenant(id="tenant_demo", name="Demo"))
+    db.commit()
+
+def _events(db: Session) -> list[AgentEvent]:
+    return list(db.exec(select(AgentEvent)).all())
+
+def test_llm_usage_is_persisted_inside_the_context() -> None:
+    """上下文内的模型调用留下带 token 用量的 llm_call_finished 行。"""
+    engine = _engine()
+    with Session(engine) as db:
+        _seed(db)
+        with persist_spans(
+            db,
+            tenant_id="tenant_demo",
+            session_id="session_1",
+            client_turn_id="run_1",
+        ):
+            span = start_llm_call(model="demo-model")
+            span.finish(**USAGE)
+
+        rows = _events(db)
+        finished = [row for row in rows if row.event_type == "llm_call_finished"]
+        assert len(finished) == 1
+        payload = finished[0].payload_json
+        assert payload["input_tokens"] == 212
+        assert payload["output_tokens"] == 10
+        assert payload["cached_input_tokens"] == 154
+        assert payload["model"] == "demo-model"
+        # 关联字段:用量要能按运行归集
+        assert payload["client_turn_id"] == "run_1"
+        assert finished[0].tenant_id == "tenant_demo"
+        assert finished[0].session_id == "session_1"
+
+def test_no_sink_bound_means_no_record() -> None:
+    """回归基线:不绑定时同样的调用不留任何痕迹——这正是被修复的缺口。"""
+    engine = _engine()
+    with Session(engine) as db:
+        _seed(db)
+        span = start_llm_call(model="demo-model")
+        span.finish(**USAGE)
+        assert _events(db) == []
+
+def test_sink_is_released_after_the_context() -> None:
+    """离开上下文后不再捕获,避免把无关调用记到该会话名下。"""
+    engine = _engine()
+    with Session(engine) as db:
+        _seed(db)
+        with persist_spans(db, tenant_id="tenant_demo", session_id="session_1"):
+            start_llm_call(model="demo-model").finish(**USAGE)
+        before = len(_events(db))
+
+        start_llm_call(model="demo-model").finish(**USAGE)
+        assert len(_events(db)) == before
+
+@pytest.mark.parametrize(
+    ("tenant_id", "session_id"),
+    [("", "session_1"), ("tenant_demo", ""), ("", "")],
+)
+def test_missing_identifiers_degrade_to_noop(tenant_id: str, session_id: str) -> None:
+    """缺租户或会话时静默跳过:可观测性不能让业务执行失败。"""
+    engine = _engine()
+    with Session(engine) as db:
+        _seed(db)
+        with persist_spans(db, tenant_id=tenant_id, session_id=session_id):
+            start_llm_call(model="demo-model").finish(**USAGE)
+        assert _events(db) == []
+
+def test_sink_failure_does_not_break_execution(monkeypatch) -> None:
+    """写入失败不得中断业务:span 发射侧已兜底,这里锁定该保证。"""
+    engine = _engine()
+    with Session(engine) as db:
+        _seed(db)
+        def boom(*_args, **_kwargs):
+            raise RuntimeError("事件写入失败")
+        monkeypatch.setattr(EventLog, "record", boom)
+        with persist_spans(db, tenant_id="tenant_demo", session_id="session_1"):
+            emit_span_event("llm_call_finished", dict(USAGE))
+
+def test_scheduled_task_run_persists_model_usage(monkeypatch) -> None:
+    from datetime import datetime
+
+    import app.scheduled_tasks.service as service
+    from app.db.models import AgentProfile, ChatSession, ScheduledTask, ScheduledTaskRun
+    from app.session.session_schema import ChatTurnResponse
+
+    engine = _engine()
+    with Session(engine) as db:
+        _seed(db)
+        db.add(
+            AgentProfile(
+                id="agent_demo",
+                tenant_id="tenant_demo",
+                name="巡检员",
+                status="active",
+            )
+        )
+        db.add(
+            ChatSession(
+                id="session_sched",
+                tenant_id="tenant_demo",
+                agent_id="agent_demo",
+                status="active",
+            )
+        )
+        task = ScheduledTask(
+            id="task_1",
+            tenant_id="tenant_demo",
+            agent_id="agent_demo",
+            created_by_user_id="user_1",
+            title="每日巡检",
+            prompt="巡检",
+        )
+        run = ScheduledTaskRun(
+            id="run_1",
+            tenant_id="tenant_demo",
+            scheduled_task_id="task_1",
+            agent_id="agent_demo",
+            user_id="user_1",
+            scheduled_for=datetime(2026, 8, 26, 9, 0, 0),
+            session_id="session_sched",
+        )
+        db.add(task)
+        db.add(run)
+        db.commit()
+
+        class FakeLoop:
+            def __init__(self, _db):
+                pass
+
+            def handle_turn_stream(self, request):
+                # 模拟一次真实的模型调用
+                start_llm_call(model="demo-model").finish(**USAGE)
+                yield {
+                    "event": "complete",
+                    "data": ChatTurnResponse(
+                        reply="巡检完成",
+                        session_id=request.session_id,
+                        session_state="active",
+                    ).model_dump(),
+                }
+        monkeypatch.setattr(service, "AgentLoop", FakeLoop)
+        service._execute_prepared_scheduled_task(db, task, run, manual=True)
+
+        finished = [
+            row for row in _events(db) if row.event_type == "llm_call_finished"
+        ]
+        assert len(finished) == 1, "定时任务运行未落下模型用量"
+        assert finished[0].session_id == "session_sched"
+        assert finished[0].payload_json["total_tokens"] == 222
+        assert finished[0].payload_json["client_turn_id"] == "run_1"
+
+def test_llm_span_bodies_are_not_persisted() -> None:
+    engine = _engine()
+    with Session(engine) as db:
+        _seed(db)
+        with persist_spans(db, tenant_id="tenant_demo", session_id="session_1"):
+            span = start_llm_call(model="demo-model")
+            span.finish(
+                **USAGE,
+                request_messages=[{"role": "user", "content": "机密提示词"}],
+                request_payload={"messages": [{"content": "机密提示词"}]},
+                response_text="机密回复",
+                response_message={"content": "机密回复"},
+                response_payload={"choices": [{"text": "机密回复"}]},
+            )
+
+        payload = _events(db)[-1].payload_json
+        for dropped in (
+            "request_messages",
+            "request_payload",
+            "response_text",
+            "response_message",
+            "response_payload",
+        ):
+            assert dropped not in payload, f"{dropped} 不应落库"
+        assert "机密" not in json.dumps(payload, ensure_ascii=False)
+        assert payload["total_tokens"] == 222
+        assert payload["cached_input_tokens"] == 154
+        assert payload["model"] == "demo-model"
+        assert payload["span_id"]
+
+def test_non_llm_spans_pass_through_unchanged() -> None:
+    engine = _engine()
+    with Session(engine) as db:
+        _seed(db)
+        with persist_spans(db, tenant_id="tenant_demo", session_id="session_1"):
+            emit_span_event(
+                "knowledge_span_finished",
+                {"operation": "knowledge.search", "query_chars": 12, "max_chunks": 5},
+            )
+
+        payload = _events(db)[-1].payload_json
+        assert payload["query_chars"] == 12
+        assert payload["max_chunks"] == 5
+
+def test_projection_keeps_fields_existing_consumers_need() -> None:
+    required = {
+        "span_id",
+        "operation",
+        "model_name",
+        "task_frame_id",
+        "iteration",
+        "attempt",
+        "max_attempts",
+        "json_attempt",
+        "json_max_attempts",
+        "started_at",
+        "duration_ms",
+        "model",
+        "error_type",
+        "error",
+        "request_parameters",
+    }
+    engine = _engine()
+    with Session(engine) as db:
+        _seed(db)
+        with persist_spans(db, tenant_id="tenant_demo", session_id="session_1"):
+            emit_span_event(
+                "llm_call_finished",
+                {key: f"v_{key}" for key in required} | {"request_payload": {"x": 1}},
+            )
+        payload = _events(db)[-1].payload_json
+        assert required <= set(payload), f"白名单遗漏:{sorted(required - set(payload))}"
+        assert "request_payload" not in payload
+
+
+def test_trimmed_rows_are_marked() -> None:
+    """裁剪过的行带 bodies_omitted,避免"未留存"被误读成"模型返回为空"。"""
+    engine = _engine()
+    with Session(engine) as db:
+        _seed(db)
+        with persist_spans(db, tenant_id="tenant_demo", session_id="session_1"):
+            emit_span_event("llm_call_finished", {**USAGE, "response_text": "原文"})
+            emit_span_event("llm_call_finished", dict(USAGE))  # 无原文可裁
+            emit_span_event("knowledge_span_finished", {"query_chars": 3})
+
+        trimmed, untouched, other = [row.payload_json for row in _events(db)]
+        assert trimmed["bodies_omitted"] is True
+        assert "bodies_omitted" not in untouched, "没裁掉任何东西时不应加标记"
+        assert "bodies_omitted" not in other, "非 llm_call 不应加标记"