feat: persist message traces in bounded SQLite storage

astrbot/core/__init__.py

@@ -35,7 +35,6 @@ html_renderer = HtmlRenderer(t2i_base_url)
 """**Deprecated**. For plugin developers, please use `self.context.html_renderer` instead."""
 logger = LogManager.GetLogger(log_name="astrbot")
 LogManager.configure_logger(logger, astrbot_config)
-LogManager.configure_trace_logger(astrbot_config)
 db_helper = SQLiteDatabase(DB_PATH)
 sp = SharedPreferences(db_helper=db_helper)
 file_token_service = FileTokenService()

astrbot/core/config/default.py

@@ -309,9 +309,9 @@ DEFAULT_CONFIG = {
     "log_file_max_mb": 20,
     "temp_dir_max_size": 1024,
     "trace_enable": False,
-    "trace_log_enable": False,
-    "trace_log_path": "logs/astrbot.trace.log",
-    "trace_log_max_mb": 20,
+    "trace_db_path": "logs/traces.db",
+    "trace_db_max_mb": 80,
+    "trace_retention_days": 7,
     "pip_install_arg": "",
     "pypi_index_url": "https://mirrors.aliyun.com/pypi/simple/",
     "persona": [],  # deprecated
@@ -3236,15 +3236,9 @@ CONFIG_METADATA_2 = {
             "log_file_path": {"type": "string", "condition": {"log_file_enable": True}},
             "log_file_max_mb": {"type": "int", "condition": {"log_file_enable": True}},
             "temp_dir_max_size": {"type": "int"},
-            "trace_log_enable": {"type": "bool"},
-            "trace_log_path": {
-                "type": "string",
-                "condition": {"trace_log_enable": True},
-            },
-            "trace_log_max_mb": {
-                "type": "int",
-                "condition": {"trace_log_enable": True},
-            },
+            "trace_db_path": {"type": "string"},
+            "trace_db_max_mb": {"type": "int"},
+            "trace_retention_days": {"type": "int"},
             "t2i_strategy": {
                 "type": "string",
                 "options": ["remote", "local"],
@@ -4805,20 +4799,20 @@ CONFIG_METADATA_3_SYSTEM = {
                         "type": "int",
                         "hint": "用于限制 data/temp 目录总大小，单位为 MB。系统每 10 分钟检查一次，超限时按文件修改时间从旧到新删除，释放约 30% 当前体积。",
                     },
-                    "trace_log_enable": {
-                        "description": "启用 Trace 文件日志",
-                        "type": "bool",
-                        "hint": "将 Trace 事件写入独立文件（不影响控制台输出）。",
-                    },
-                    "trace_log_path": {
-                        "description": "Trace 日志文件路径",
+                    "trace_db_path": {
+                        "description": "追踪数据库路径",
                         "type": "string",
-                        "hint": "相对路径以 data 目录为基准，例如 logs/astrbot.trace.log；支持绝对路径。",
+                        "hint": "独立 SQLite 数据库，相对 data 目录。默认 logs/traces.db；更改路径后重启生效。开启追踪即持久化，不再输出 JSONL。",
                     },
-                    "trace_log_max_mb": {
-                        "description": "Trace 日志大小上限 (MB)",
+                    "trace_db_max_mb": {
+                        "description": "追踪数据库容量预算 (MB)",
                         "type": "int",
-                        "hint": "超过大小后自动轮转，默认 20MB。",
+                        "hint": "正整数，默认 80 MB。按完整追踪淘汰旧数据；WAL 与检查点可能暂时增加磁盘占用，非瞬时硬上限。不支持无限容量。",
+                    },
+                    "trace_retention_days": {
+                        "description": "追踪保留天数",
+                        "type": "int",
+                        "hint": "正整数，默认 7 天；保留期限或容量预算先达到者触发清理。运行中的追踪不会被常规清理。",
                     },
                     "pip_install_arg": {
                         "description": "pip 安装额外参数",

astrbot/core/core_lifecycle.py

@@ -165,10 +165,8 @@ class AstrBotCoreLifecycle:
             LogManager.configure_logger(
                 logger, self.astrbot_config, override_level="DEBUG"
             )
-            LogManager.configure_trace_logger(self.astrbot_config)
         else:
             LogManager.configure_logger(logger, self.astrbot_config)
-            LogManager.configure_trace_logger(self.astrbot_config)
 
         await self.db.initialize()
         if sp.db_helper is self.db:
@@ -420,6 +418,9 @@ class AstrBotCoreLifecycle:
                 logger.error(f"任务 {task.get_name()} 发生错误: {e}")
 
         # 释放数据库引擎连接池
+        from astrbot.core.utils.trace_store import close_trace_store
+
+        await asyncio.to_thread(close_trace_store)
         try:
             await self.db.engine.dispose()
         except Exception as e:
@@ -431,6 +432,9 @@ class AstrBotCoreLifecycle:
         await self.provider_manager.terminate()
         await self.platform_manager.terminate()
         await self.kb_manager.terminate()
+        from astrbot.core.utils.trace_store import close_trace_store
+
+        await asyncio.to_thread(close_trace_store)
         if sp.db_helper is self.db:
             await sp.close()
         self.dashboard_shutdown_event.set()

astrbot/core/event_bus.py

@@ -42,7 +42,8 @@ class EventBus:
             conf_info = self.astrbot_config_mgr.get_conf_info(event.unified_msg_origin)
             conf_id = conf_info["id"]
             conf_name = conf_info.get("name") or conf_id
-            self._print_event(event, conf_name)
+            with event.trace.context():
+                self._print_event(event, conf_name)
             scheduler = self.pipeline_scheduler_mapping.get(conf_id)
             if not scheduler:
                 logger.error(

astrbot/core/log.py

@@ -6,6 +6,7 @@ import logging
 import os
 import sys
 import tempfile
+import threading
 import time
 from asyncio import Queue
 from collections import deque
@@ -23,6 +24,11 @@ from astrbot.core.utils.astrbot_path import (
     get_astrbot_config_path,
     get_astrbot_data_path,
 )
+from astrbot.core.utils.telemetry import (
+    bounded_attributes,
+    bounded_text,
+    current_trace_context,
+)
 
 CACHED_SIZE = 500
 
@@ -115,6 +121,10 @@ class _RecordEnricherFilter(logging.Filter):
         record.platform_type = getattr(record, "platform_type", None) or (
             context.platform_type if context_is_active else None
         )
+        trace_context = current_trace_context.get()
+        if trace_context is not None and trace_context.active:
+            record.trace_id = trace_context.trace_id
+            record.span_id = trace_context.span_id
         return True
 
 
@@ -176,6 +186,14 @@ def _patch_record(record: "Record") -> None:
     extra.setdefault("plugin_name", None)
     extra.setdefault("platform_id", None)
     extra.setdefault("platform_type", None)
+    extra.setdefault(
+        "trace_tag",
+        (
+            f" [trace_id={extra['trace_id']} span_id={extra.get('span_id', '')}]"
+            if extra.get("trace_id")
+            else ""
+        ),
+    )
 
 
 _loguru = _raw_loguru_logger.patch(_patch_record)
@@ -208,6 +226,8 @@ class _LoguruInterceptHandler(logging.Handler):
             "plugin_name": getattr(record, "plugin_name", None),
             "platform_id": getattr(record, "platform_id", None),
             "platform_type": getattr(record, "platform_type", None),
+            "trace_id": getattr(record, "trace_id", None),
+            "span_id": getattr(record, "span_id", None),
         }
 
         _loguru.bind(**payload).opt(exception=record.exc_info).log(
@@ -222,22 +242,70 @@ class LogBroker:
     def __init__(self) -> None:
         self.log_cache = deque(maxlen=CACHED_SIZE)
         self.subscribers: list[Queue] = []
+        self._subscriber_lock = threading.Lock()
+        self._subscriber_loops: dict[Queue, asyncio.AbstractEventLoop | None] = {}
+        self._pending_callbacks: dict[Queue, int] = {}
 
     def register(self) -> Queue:
         q = Queue(maxsize=CACHED_SIZE + 10)
-        self.subscribers.append(q)
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            loop = None
+        with self._subscriber_lock:
+            self.subscribers.append(q)
+            self._subscriber_loops[q] = loop
+            self._pending_callbacks[q] = 0
         return q
 
     def unregister(self, q: Queue) -> None:
-        self.subscribers.remove(q)
+        with self._subscriber_lock:
+            self.subscribers.remove(q)
+            self._subscriber_loops.pop(q, None)
+            self._pending_callbacks.pop(q, None)
 
     def publish(self, log_entry: dict) -> None:
-        self.log_cache.append(log_entry)
-        for q in self.subscribers:
+        if log_entry.get("type") != "trace":
+            self.log_cache.append(log_entry)
+        try:
+            current_loop = asyncio.get_running_loop()
+        except RuntimeError:
+            current_loop = None
+        for q in tuple(self.subscribers):
+            owner = self._subscriber_loops.get(q)
+            if owner is None or owner is current_loop:
+                try:
+                    q.put_nowait(log_entry)
+                except asyncio.QueueFull:
+                    pass
+                continue
+            # Reserve space before scheduling: queued callbacks must also be
+            # bounded when a background writer reports errors to a slow loop.
+            with self._subscriber_lock:
+                if (
+                    q not in self._pending_callbacks
+                    or q.qsize() + self._pending_callbacks[q] >= q.maxsize
+                    or owner.is_closed()
+                ):
+                    continue
+                self._pending_callbacks[q] += 1
+
+            def enqueue(queue=q, entry=log_entry):
+                with self._subscriber_lock:
+                    if queue not in self._pending_callbacks:
+                        return
+                    self._pending_callbacks[queue] -= 1
+                try:
+                    queue.put_nowait(entry)
+                except asyncio.QueueFull:
+                    pass
+
             try:
-                q.put_nowait(log_entry)
-            except asyncio.QueueFull:
-                pass
+                owner.call_soon_threadsafe(enqueue)
+            except RuntimeError:
+                with self._subscriber_lock:
+                    if q in self._pending_callbacks:
+                        self._pending_callbacks[q] -= 1
 
 
 class LogQueueHandler(logging.Handler):
@@ -253,7 +321,42 @@ class LogQueueHandler(logging.Handler):
             {
                 "level": record.levelname,
                 "time": time.time(),
-                "data": log_entry,
+                "data": bounded_text(log_entry),
+                "timestamp": str(int(record.created * 10**9)),
+                "observed_timestamp": str(time.time_ns()),
+                "severity_text": record.levelname,
+                "severity_number": {10: 5, 20: 9, 30: 13, 40: 17, 50: 21}.get(
+                    record.levelno, 0
+                ),
+                "body": bounded_text(record.getMessage()),
+                **(
+                    {
+                        "trace_id": record.trace_id,
+                        "span_id": record.span_id,
+                        "trace_flags": 1,
+                    }
+                    if getattr(record, "trace_id", None)
+                    else {}
+                ),
+                "attributes": bounded_attributes(
+                    {
+                        "astrbot.category": getattr(record, "category", None)
+                        or "system",
+                        "astrbot.source": getattr(record, "source", "core"),
+                        "astrbot.plugin_name": getattr(record, "plugin_name", None),
+                        "astrbot.platform_id": getattr(record, "platform_id", None),
+                        "astrbot.platform_type": getattr(record, "platform_type", None),
+                        "code.file.path": record.pathname,
+                        "code.line.number": record.lineno,
+                    }
+                ),
+                "resource": {
+                    "attributes": {
+                        "service.name": "astrbot",
+                        "service.version": VERSION,
+                    }
+                },
+                "instrumentation_scope": {"name": record.name},
                 "category": getattr(record, "category", None) or "system",
                 "source": getattr(record, "source", "core"),
                 "plugin_name": getattr(record, "plugin_name", None),
@@ -270,7 +373,6 @@ class LogManager:
     _configured = False
     _console_sink_id: int | None = None
     _file_sink_id: int | None = None
-    _trace_sink_id: int | None = None
     _plugin_logger_names: set[str] = set()
     _plugin_level_overrides: dict[str, str] | None = None
     _log_broker: "LogBroker | None" = None
@@ -524,25 +626,12 @@ class LogManager:
         level: int,
         max_mb: int | None,
         backup_count: int,
-        trace: bool,
     ) -> int:
         os.makedirs(os.path.dirname(file_path) or ".", exist_ok=True)
         rotation = f"{max_mb} MB" if max_mb and max_mb > 0 else None
         retention = (
             backup_count if rotation and backup_count and backup_count > 0 else None
         )
-        if trace:
-            return _loguru.add(
-                file_path,
-                level="INFO",
-                format="[{time:YYYY-MM-DD HH:mm:ss.SSS}] {message}",
-                encoding="utf-8",
-                rotation=rotation,
-                retention=retention,
-                enqueue=True,
-                filter=lambda record: record["extra"].get("is_trace", False),
-            )
-
         logging_level_name = logging.getLevelName(level)
         if isinstance(logging_level_name, int):
             logging_level_name = "INFO"
@@ -552,7 +641,7 @@ class LogManager:
             format=(
                 "[{time:YYYY-MM-DD HH:mm:ss.SSS}] {extra[plugin_tag]} "
                 "[{extra[short_levelname]}]{extra[astrbot_version_tag]} "
-                "[{extra[source_file]}:{extra[source_line]}]: {message}"
+                "[{extra[source_file]}:{extra[source_line]}]{extra[trace_tag]}: {message}"
             ),
             encoding="utf-8",
             rotation=rotation,
@@ -636,43 +725,6 @@ class LogManager:
                 level=logger.level,
                 max_mb=max_mb,
                 backup_count=3,
-                trace=False,
             )
         except Exception as e:
             logger.error(f"Failed to add file sink: {e}")
-
-    @classmethod
-    def configure_trace_logger(cls, config: dict | None) -> None:
-        if not config:
-            return
-
-        enable = bool(
-            config.get("trace_log_enable")
-            or (config.get("log_file", {}) or {}).get("trace_enable", False)
-        )
-        path = config.get("trace_log_path")
-        max_mb = config.get("trace_log_max_mb")
-        if "log_file" in config:
-            legacy = config.get("log_file") or {}
-            path = path or legacy.get("trace_path")
-            max_mb = max_mb or legacy.get("trace_max_mb")
-
-        trace_logger = logging.getLogger("astrbot.trace")
-        cls._ensure_logger_enricher_filter(trace_logger)
-        cls._ensure_logger_intercept_handler(trace_logger)
-        trace_logger.setLevel(logging.INFO)
-        trace_logger.propagate = False
-
-        cls._remove_sink(cls._trace_sink_id)
-        cls._trace_sink_id = None
-
-        if not enable:
-            return
-
-        cls._trace_sink_id = cls._add_file_sink(
-            file_path=cls._resolve_log_path(path or "logs/astrbot.trace.log"),
-            level=logging.INFO,
-            max_mb=max_mb,
-            backup_count=3,
-            trace=True,
-        )

astrbot/core/pipeline/respond/stage.py

@@ -13,6 +13,7 @@ from astrbot.core.platform.astr_message_event import (
 )
 from astrbot.core.star.star_handler import EventType
 from astrbot.core.utils.path_util import path_Mapping
+from astrbot.core.utils.telemetry import bounded_text
 
 from ..context import PipelineContext, call_event_hook
 from ..stage import Stage, register_stage
@@ -267,6 +268,8 @@ class RespondStage(Stage):
             logger.info(f"Applying streaming output ({event.get_platform_id()}).")
             stream = result.async_stream
             stream_outline: list[str] = []
+            outline_bytes = 0
+            outline_truncated = False
 
             async def traced_stream() -> AsyncGenerator[MessageChain, None]:
                 """Collect response outlines as the platform consumes the stream.
@@ -274,6 +277,7 @@ class RespondStage(Stage):
                 Yields:
                     Original message chunks for platform delivery.
                 """
+                nonlocal outline_bytes, outline_truncated
                 iterator = stream.__aiter__()
                 while True:
                     # Tool/plugin sends during generation are independent responses.
@@ -291,7 +295,15 @@ class RespondStage(Stage):
                         "tool_call",
                         "agent_stats",
                     }:
-                        stream_outline.append(event._outline_chain(chain.chain))
+                        fragment = event._outline_chain(chain.chain)
+                        remaining = 4096 - outline_bytes
+                        if remaining >= 16:
+                            preview = bounded_text(fragment, remaining)
+                            stream_outline.append(preview)
+                            outline_bytes += len(preview.encode())
+                            outline_truncated |= preview != fragment
+                        elif fragment:
+                            outline_truncated = True
                     yield chain
 
             # Adapter chunks and protocol closing frames share one final trace.
@@ -308,6 +320,7 @@ class RespondStage(Stage):
                     source=event.get_extra("_trace_response_source"),
                     message=outline,
                     streaming=True,
+                    **({"truncated": True} if outline_truncated else {}),
                 )
             return
         if len(result.chain) > 0:

astrbot/core/pipeline/scheduler.py

@@ -89,7 +89,10 @@ class PipelineScheduler:
 
         """
         active_event_registry.register(event)
-        with log_context(event.get_platform_id(), event.get_platform_name()):
+        with (
+            log_context(event.get_platform_id(), event.get_platform_name()),
+            event.trace.context(),
+        ):
             pipeline_started = time.perf_counter()
             pipeline_status = "success"
             pipeline_error = None

astrbot/core/utils/migra_helper.py

@@ -311,6 +311,33 @@ def migrate_config_on_load(config: dict[str, Any], config_path: Path) -> bool:
     Returns:
         Whether the configuration changed.
     """
+    changed = False
+    legacy = config.get("log_file")
+    legacy = legacy if isinstance(legacy, dict) else {}
+    old_keys = ("trace_log_enable", "trace_log_path", "trace_log_max_mb")
+    if any(key in config for key in old_keys) or any(
+        key in legacy for key in ("trace_enable", "trace_path", "trace_max_mb")
+    ):
+        old_size = config.get("trace_log_max_mb", legacy.get("trace_max_mb", 20))
+        if "trace_db_max_mb" not in config:
+            config["trace_db_max_mb"] = (
+                old_size * 4
+                if isinstance(old_size, int)
+                and not isinstance(old_size, bool)
+                and old_size > 0
+                else 80
+            )
+        config.setdefault("trace_db_path", "logs/traces.db")
+        config.setdefault("trace_retention_days", 7)
+        for key in old_keys:
+            config.pop(key, None)
+        for key in ("trace_enable", "trace_path", "trace_max_mb"):
+            legacy.pop(key, None)
+        logger.warning(
+            "Trace file settings migrated to SQLite in %s. Recording now persists when trace_enable is on; old files are untouched and not imported.",
+            config_path,
+        )
+        changed = True
     fallback_config = None
     resolved_path = config_path.resolve()
     profile_root = Path(get_astrbot_config_path()).resolve()
@@ -327,7 +354,7 @@ def migrate_config_on_load(config: dict[str, Any], config_path: Path) -> bool:
                 resolved_path,
                 exc,
             )
-    return _migrate_agent_runner_config(config, fallback_config)
+    return _migrate_agent_runner_config(config, fallback_config) or changed
 
 
 def finalize_config_migrations(configs: list[dict[str, Any]]) -> bool:

astrbot/core/utils/storage_cleaner.py

@@ -155,7 +155,32 @@ class StorageCleaner:
         files = set(self._iter_files(self._data_dir / "logs"))
         for log_path in self._configured_log_paths():
             files.update(self._iter_log_family_files(log_path))
-        return files
+        trace_path = self._resolve_log_path(
+            self._get_optional_str("trace_db_path"),
+            default_relative_path="logs/traces.db",
+        ).resolve()
+        protected = {
+            Path(str(trace_path) + suffix)
+            for suffix in ("", "-wal", "-shm", "-journal")
+        }
+        candidates = set()
+        for path in files:
+            if (
+                path.resolve() in protected
+                or path.suffix in {".db", ".sqlite", ".sqlite3"}
+                or path.name.endswith(("-wal", "-shm", "-journal"))
+            ):
+                continue
+            try:
+                # Protect custom database names, including an old active path
+                # after a configuration edit that has not restarted the core.
+                with path.open("rb") as stream:
+                    if stream.read(16) == b"SQLite format 3\x00":
+                        continue
+            except OSError:
+                continue
+            candidates.add(path)
+        return candidates
 
     def _collect_cache_files(self) -> set[Path]:
         files = set(self._iter_files(self._temp_dir))

astrbot/core/utils/telemetry.py

@@ -0,0 +1,147 @@
+"""Bounded OTel-compatible values and local trace context (without an SDK)."""
+
+import json
+from collections.abc import Iterator
+from contextlib import contextmanager
+from contextvars import ContextVar
+from dataclasses import dataclass
+from itertools import islice
+from typing import Any
+
+MAX_ATTRIBUTE_BYTES = 16 * 1024
+MAX_BODY_BYTES = 16 * 1024
+
+
+def bounded_text(value: str, limit: int = MAX_BODY_BYTES) -> str:
+    """Bound text by UTF-8 bytes without splitting a character.
+
+    Args:
+        value: Original text.
+        limit: Maximum encoded length.
+
+    Returns:
+        Original text or text ending in a truncation marker.
+    """
+    encoded = value[:limit].encode("utf-8")
+    if len(value) <= limit and len(encoded) <= limit:
+        return value
+    return (
+        encoded[: max(0, limit - 16)].decode("utf-8", errors="ignore") + " [truncated]"
+    )
+
+
+def bounded_attributes(fields: dict[str, Any]) -> dict[str, Any]:
+    """Convert application fields to bounded OTel attribute values.
+
+    Nested objects are represented as JSON strings, not non-standard attribute
+    objects. A shared byte budget also bounds wide and deeply nested inputs.
+
+    Args:
+        fields: Application attributes with their final names.
+
+    Returns:
+        Primitive attributes with an optional truncation marker.
+    """
+    remaining = MAX_ATTRIBUTE_BYTES - 64
+    truncated = False
+    result = {}
+    for key, value in fields.items():
+        key = bounded_text(str(key), 128)
+        if value is None:
+            continue
+        if (
+            isinstance(value, (list, tuple))
+            and len(value) <= 128
+            and (not value or type(value[0]) in (str, bool, int, float))
+            and all(type(item) is type(value[0]) for item in value)
+        ):
+            original = value
+            value = [
+                bounded_text(item, 256) if isinstance(item, str) else item
+                for item in value
+            ]
+            truncated |= value != list(original)
+        elif not isinstance(value, (str, bool, int, float)):
+            # Bound containers before serialization, including nested values.
+            def compact(item: Any, depth: int = 0) -> Any:
+                """Bound nested values before serialization."""
+                nonlocal truncated
+                if isinstance(item, str):
+                    text = bounded_text(item, 2048)
+                    truncated |= text != item
+                    return text
+                if depth >= 3:
+                    truncated = True
+                    return bounded_text(str(type(item).__name__), 128)
+                if isinstance(item, dict):
+                    truncated |= len(item) > 16
+                    return {
+                        str(k)[:128]: compact(v, depth + 1)
+                        for k, v in islice(item.items(), 16)
+                    }
+                if isinstance(item, (list, tuple)):
+                    truncated |= len(item) > 16
+                    return [compact(v, depth + 1) for v in item[:16]]
+                return (
+                    item
+                    if isinstance(item, (bool, int, float)) or item is None
+                    else bounded_text(str(item), 2048)
+                )
+
+            try:
+                value = json.dumps(compact(value), ensure_ascii=False, allow_nan=False)
+            except (TypeError, ValueError):
+                truncated = True
+                continue
+        if isinstance(value, str):
+            original = value
+            value = bounded_text(value, min(4096, max(16, remaining - len(key) - 16)))
+            truncated |= value != original
+        try:
+            cost = len(
+                json.dumps({key: value}, ensure_ascii=False, allow_nan=False).encode()
+            )
+        except ValueError:
+            continue
+        if cost > remaining:
+            result["astrbot.truncated"] = True
+            break
+        result[key] = value
+        remaining -= cost
+    if truncated:
+        result["astrbot.truncated"] = True
+    return result
+
+
+@dataclass
+class TraceContext:
+    """Trace identifiers whose lifetime also applies to copied task contexts."""
+
+    trace_id: str
+    span_id: str
+    active: bool = True
+
+
+current_trace_context: ContextVar[TraceContext | None] = ContextVar(
+    "astrbot_trace_context", default=None
+)
+
+
+@contextmanager
+def bind_trace_context(trace_id: str, span_id: str) -> Iterator[TraceContext]:
+    """Bind IDs and invalidate detached copies on exit.
+
+    Args:
+        trace_id: Lowercase 32-character trace ID.
+        span_id: Lowercase 16-character span ID.
+
+    Yields:
+        The active context.
+    """
+    context = TraceContext(trace_id, span_id)
+    token = current_trace_context.set(context)
+    try:
+        yield context
+    finally:
+        context.active = False
+        current_trace_context.reset(token)

astrbot/core/utils/trace.py

@@ -1,15 +1,28 @@
+"""OTel-compatible local trace/span/event model, independent of OTLP."""
+
 import json
-import logging
+import secrets
 import time
-import uuid
+from collections.abc import Iterator
+from contextlib import contextmanager
 from typing import Any
 
-from astrbot import logger
-from astrbot.core import LogManager, astrbot_config
+from astrbot import __version__, logger
+from astrbot.core import astrbot_config
 from astrbot.core.log import LogQueueHandler
+from astrbot.core.utils.telemetry import (
+    TraceContext,
+    bind_trace_context,
+    bounded_attributes,
+    bounded_text,
+    current_trace_context,
+)
+from astrbot.core.utils.trace_store import get_trace_store
 
 _cached_log_broker = None
-_trace_logger = None
+MAX_TRACE_BYTES = 1024 * 1024
+MAX_SPANS = 128
+MAX_EVENTS = 256
 
 
 def _get_log_broker():
@@ -23,18 +36,9 @@ def _get_log_broker():
     return None
 
 
-def _get_trace_logger():
-    global _trace_logger
-    if _trace_logger is not None:
-        return _trace_logger
-
-    # 按配置初始化 trace 文件日志
-    LogManager.configure_trace_logger(astrbot_config)
-    _trace_logger = logging.getLogger("astrbot.trace")
-    return _trace_logger
-
-
 class TraceSpan:
+    """Own one message trace and its bounded operation spans and events."""
+
     def __init__(
         self,
         name: str,
@@ -42,42 +46,324 @@ class TraceSpan:
         sender_name: str | None = None,
         message_outline: str | None = None,
     ) -> None:
-        self.span_id = str(uuid.uuid4())
-        self.name = name
-        self.umo = umo
-        self.sender_name = sender_name
-        self.message_outline = message_outline
+        self.trace_id = secrets.token_hex(16)
+        self.span_id = secrets.token_hex(8)
+        self.name = bounded_text(name, 256)
+        self.umo = bounded_text(umo or "", 1024)
+        self.sender_name = bounded_text(sender_name or "", 256)
+        self.message_outline = bounded_text(message_outline or "", 4096)
         self.started_at = time.time()
-        self._sequence = 0
+        self._start_ns = time.time_ns()
+        self._started = False
+        self._finished = False
+        self._bytes = 0
+        self._events = 0
+        self._span_count = 1
+        self._pending: dict[str, tuple[dict, Any, TraceContext]] = {}
+        self._plugins: set[str] = set()
+        self._summary = {
+            "trace_id": self.trace_id,
+            "root_span_id": self.span_id,
+            "name": self.name,
+            "start_time_unix_nano": str(self._start_ns),
+            "end_time_unix_nano": None,
+            "state": "running",
+            "has_errors": False,
+            "has_models": False,
+            "has_plugins": False,
+            "span_count": 1,
+            "truncated": False,
+            "attributes": bounded_attributes(
+                {
+                    "astrbot.umo": self.umo,
+                    "astrbot.sender_name": self.sender_name,
+                    "astrbot.message_outline": self.message_outline,
+                }
+            ),
+            "resource": {
+                "attributes": {
+                    "service.name": "astrbot",
+                    "service.version": __version__,
+                }
+            },
+        }
+        self._root = {
+            "trace_id": self.trace_id,
+            "span_id": self.span_id,
+            "parent_span_id": None,
+            "name": self.name,
+            "kind": 1,
+            "start_time_unix_nano": str(self._start_ns),
+            "end_time_unix_nano": None,
+            "status": {"code": 0},
+            "attributes": {"astrbot.action": "event_received"},
+        }
 
-    def record(self, action: str, **fields: Any) -> None:
-        # Check if trace recording is enabled
-        if not astrbot_config.get("trace_enable", True):
+    @contextmanager
+    def context(self) -> Iterator[None]:
+        """Bind the message root without correlating unrecorded messages.
+
+        Yields:
+            Control while this root's IDs are active.
+        """
+        if not astrbot_config.get("trace_enable", False) and not self._started:
+            yield
             return
+        with bind_trace_context(self.trace_id, self.span_id):
+            yield
 
+    def _emit(
+        self,
+        *,
+        span: dict | None = None,
+        event: dict | None = None,
+        closing: bool = False,
+    ) -> None:
+        """Queue a bounded snapshot and separately notify live subscribers.
+
+        Args:
+            span: Standard span update.
+            event: Standard event associated with a span.
+            closing: Whether completion must survive the trace payload limit.
+        """
         payload = {
             "type": "trace",
-            "level": "TRACE",
-            "time": time.time(),
-            "record_id": str(uuid.uuid4()),
-            "sequence": self._sequence,
-            "elapsed_ms": round((time.time() - self.started_at) * 1000, 2),
-            "schema_version": 2,
-            "span_id": self.span_id,
-            "name": self.name,
-            "umo": self.umo,
-            "sender_name": self.sender_name,
-            "message_outline": self.message_outline,
-            "action": action,
-            "fields": fields,
+            "schema_version": 3,
+            "trace": self._summary,
+            "time_unix_nano": str(time.time_ns()),
         }
-        self._sequence += 1
-        log_broker = _get_log_broker()
-        if log_broker:
-            log_broker.publish(payload)
-        else:
-            logger.info(f"[trace] {payload}")
+        self._summary["updated_time_unix_nano"] = payload["time_unix_nano"]
+        if span is not None:
+            payload["span"] = span
+        if event is not None:
+            payload["event"] = event
+        size = len(json.dumps(payload, ensure_ascii=False).encode())
+        if self._bytes + size > MAX_TRACE_BYTES:
+            self._summary["truncated"] = True
+            self._bytes = MAX_TRACE_BYTES
+            if not closing:
+                return
+            if span is not None:
+                span["attributes"] = bounded_attributes(
+                    {
+                        key: bounded_text(value, 256)
+                        if isinstance(value, str)
+                        else value
+                        for key, value in span["attributes"].items()
+                        if key
+                        in {
+                            "astrbot.action",
+                            "astrbot.end_action",
+                            "astrbot.kind",
+                            "astrbot.plugin",
+                            "astrbot.handler",
+                            "astrbot.handler_full_name",
+                            "astrbot.handler_category",
+                            "astrbot.handler_type",
+                            "astrbot.duration_ms",
+                            "astrbot.runner_type",
+                        }
+                    }
+                )
+        self._bytes += size
+        try:
+            accepted = get_trace_store(astrbot_config).append(payload)
+        except Exception:
+            accepted = False
+            logger.error("Failed to queue trace persistence.", exc_info=True)
+        # Persistence acknowledgements are not implied by an SSE notification.
+        live = json.loads(
+            json.dumps({**payload, "storage_accepted": accepted}, ensure_ascii=False)
+        )
+        broker = _get_log_broker()
+        if broker:
+            broker.publish(live)
 
-        trace_logger = _get_trace_logger()
-        if trace_logger and trace_logger.handlers:
-            trace_logger.info(json.dumps(payload, ensure_ascii=False))
+    def record(self, action: str, **fields: Any) -> None:
+        """Record a span transition or event using bounded primitive attributes.
+
+        Args:
+            action: Existing instrumentation action.
+            **fields: AstrBot-specific metadata mapped to namespaced attributes.
+        """
+        if self._finished:
+            return
+        enabled = bool(astrbot_config.get("trace_enable", False))
+        if not enabled and not self._started:
+            return
+        now = time.time_ns()
+        attrs = bounded_attributes(
+            {"astrbot." + key: value for key, value in fields.items()}
+        )
+        if attrs.get("astrbot.truncated"):
+            self._summary["truncated"] = True
+        if fields.get("status") == "error":
+            self._summary["has_errors"] = True
+        if action == "event_received":
+            self._summary["attributes"].update(
+                {key: value for key, value in attrs.items() if key != "astrbot.message"}
+            )
+            self._root["attributes"].update(attrs)
+        if not self._started:
+            self._started = True
+            self._emit(span=self._root)
+        if action == "event_received":
+            return
+
+        current = current_trace_context.get()
+        parent = (
+            current.span_id
+            if current and current.active and current.trace_id == self.trace_id
+            else self.span_id
+        )
+        starts = {
+            "handler_start": "handler_end",
+            "agent_start": "agent_end",
+            "agent_tool_call": "agent_tool_result",
+        }
+        if action in starts:
+            if (
+                not enabled
+                or self._span_count >= MAX_SPANS
+                or self._bytes >= MAX_TRACE_BYTES
+            ):
+                self._summary["truncated"] = True
+                return
+            span_id = secrets.token_hex(8)
+            span = {
+                "trace_id": self.trace_id,
+                "span_id": span_id,
+                "parent_span_id": parent,
+                "name": bounded_text(
+                    str(
+                        fields.get("handler_full_name")
+                        or fields.get("tool_name")
+                        or action
+                    ),
+                    256,
+                ),
+                "kind": 1,
+                "start_time_unix_nano": str(now),
+                "end_time_unix_nano": None,
+                "status": {"code": 0},
+                "attributes": {
+                    **attrs,
+                    "astrbot.action": action,
+                    "astrbot.end_action": starts[action],
+                },
+            }
+            context = TraceContext(self.trace_id, span_id)
+            token = current_trace_context.set(context)
+            self._pending[span_id] = (span, token, context)
+            self._span_count += 1
+            self._summary["span_count"] = self._span_count
+            if action == "handler_start":
+                self._summary["has_plugins"] = True
+                if len(self._plugins) < MAX_SPANS:
+                    self._plugins.add(
+                        bounded_text(str(fields.get("plugin") or ""), 256)
+                    )
+                if len(self._plugins) < MAX_SPANS:
+                    self._plugins.add(
+                        bounded_text(
+                            str(
+                                fields.get("plugin_display_name")
+                                or fields.get("plugin")
+                                or ""
+                            ),
+                            256,
+                        )
+                    )
+                self._summary["attributes"]["astrbot.plugins"] = bounded_text(
+                    ", ".join(sorted(self._plugins)), 4096
+                )
+            if action == "agent_start":
+                self._summary["has_models"] = True
+            self._emit(span=span)
+            return
+
+        if action in starts.values() or action == "pipeline_end":
+            target = parent
+            while target in self._pending:
+                span, _, _ = self._pending[target]
+                if action == "pipeline_end" or (
+                    span["attributes"]["astrbot.end_action"] == action
+                    and (
+                        action != "handler_end"
+                        or span["attributes"].get("astrbot.handler_full_name")
+                        == fields.get("handler_full_name")
+                    )
+                ):
+                    break
+                target = span["parent_span_id"]
+            if action != "pipeline_end" and target not in self._pending:
+                return
+            # Close abandoned descendants before their ancestor. Their copied
+            # contexts are invalidated even when an operation is cancelled.
+            while parent in self._pending:
+                span, token, context = self._pending.pop(parent)
+                is_target = parent == target and action != "pipeline_end"
+                span["end_time_unix_nano"] = str(now)
+                if is_target:
+                    span["attributes"] = bounded_attributes(
+                        {**span["attributes"], **attrs}
+                    )
+                    span["status"] = {
+                        "code": 2 if fields.get("status") == "error" else 1
+                    }
+                    if fields.get("error"):
+                        span["status"]["message"] = bounded_text(
+                            str(fields["error"]), 1024
+                        )
+                else:
+                    span["status"] = {
+                        "code": 2,
+                        "message": "Parent operation ended before this span.",
+                    }
+                    self._summary["has_errors"] = True
+                context.active = False
+                current_trace_context.reset(token)
+                parent = span["parent_span_id"]
+                self._emit(span=span, closing=True)
+                if is_target:
+                    break
+            if action != "pipeline_end":
+                return
+            self._finished = True
+            for span, _, context in self._pending.values():
+                context.active = False
+                span["end_time_unix_nano"] = str(now)
+                span["status"] = {
+                    "code": 2,
+                    "message": "Message ended before this concurrent operation.",
+                }
+                self._summary["has_errors"] = True
+                self._emit(span=span, closing=True)
+            self._pending.clear()
+            self._summary["state"] = "completed"
+            self._summary["end_time_unix_nano"] = str(now)
+            self._root["end_time_unix_nano"] = str(now)
+            self._root["attributes"].update(attrs)
+            self._root["attributes"]["astrbot.end_action"] = "pipeline_end"
+            self._root["status"] = {"code": 2 if fields.get("status") == "error" else 1}
+            if fields.get("error"):
+                self._root["status"]["message"] = bounded_text(
+                    str(fields["error"]), 1024
+                )
+            self._emit(span=self._root, closing=True)
+            return
+
+        if not enabled or self._events >= MAX_EVENTS or self._bytes >= MAX_TRACE_BYTES:
+            self._summary["truncated"] = True
+            return
+        self._events += 1
+        self._emit(
+            event={
+                "event_id": secrets.token_hex(8),
+                "trace_id": self.trace_id,
+                "span_id": parent,
+                "name": action,
+                "time_unix_nano": str(now),
+                "attributes": attrs,
+            }
+        )

astrbot/core/utils/trace_store.py

@@ -0,0 +1,524 @@
+"""Independent, bounded SQLite storage for local trace history."""
+
+from __future__ import annotations
+
+import atexit
+import json
+import logging
+import sqlite3
+import threading
+import time
+from collections.abc import Mapping
+from contextlib import closing
+from pathlib import Path
+from queue import Empty, Full, Queue
+from typing import Any
+
+from astrbot.core.utils.astrbot_path import get_astrbot_data_path
+
+logger = logging.getLogger("astrbot")
+QUEUE_BYTES = 4 * 1024 * 1024
+MAX_FRAME_BYTES = 64 * 1024
+
+
+class TraceStore:
+    """Store span updates and events using one bounded background writer."""
+
+    def __init__(self, path: Path, config: Mapping[str, Any]) -> None:
+        self.path = path
+        self.config = config
+        self.queue: Queue = Queue(maxsize=128)
+        self._lock = threading.Lock()
+        self._pending_bytes = 0
+        self._stop = threading.Event()
+        self._ready = threading.Event()
+        self._failed = False
+        self.dropped = 0
+        self.last_error: str | None = None
+        self._last_warning = 0.0
+        self._thread = threading.Thread(
+            target=self._run, name="trace-sqlite", daemon=True
+        )
+        self._thread.start()
+
+    def append(self, payload: dict) -> bool:
+        """Queue an immutable snapshot without blocking message processing.
+
+        Args:
+            payload: A trace summary with a span update or event.
+
+        Returns:
+            Whether the bounded queue accepted the snapshot.
+        """
+        encoded = json.dumps(
+            payload, ensure_ascii=False, separators=(",", ":"), allow_nan=False
+        )
+        size = len(encoded.encode())
+        with self._lock:
+            if (
+                self._stop.is_set()
+                or self._failed
+                or size > MAX_FRAME_BYTES
+                or self._pending_bytes + size > QUEUE_BYTES
+            ):
+                accepted = False
+            else:
+                try:
+                    self.queue.put_nowait((encoded, size))
+                    self._pending_bytes += size
+                    accepted = True
+                except Full:
+                    accepted = False
+            if not accepted:
+                self.dropped += 1
+        if not accepted and time.monotonic() - self._last_warning > 30:
+            self._last_warning = time.monotonic()
+            logger.warning(
+                "Trace persistence rejected a record: queue/capacity unavailable (dropped=%s).",
+                self.dropped,
+            )
+        return accepted
+
+    def flush(self, timeout: float = 5) -> bool:
+        """Wait for accepted snapshots to commit.
+
+        Args:
+            timeout: Maximum seconds to wait.
+
+        Returns:
+            Whether the barrier committed without a storage error.
+        """
+        if self._failed or self._stop.is_set():
+            return False
+        done = threading.Event()
+        try:
+            self.queue.put((done, 0), timeout=timeout)
+        except Full:
+            return False
+        return done.wait(timeout) and self.last_error is None
+
+    def close(self) -> None:
+        """Drain the writer and close its SQLite connection."""
+        self._stop.set()
+        self._thread.join(timeout=5)
+        if self._thread.is_alive():
+            logger.error(
+                "Trace database writer did not finish within the shutdown deadline."
+            )
+
+    def _prune(self, connection: sqlite3.Connection, force: bool = False) -> None:
+        """Evict complete traces as units and reclaim reusable database pages.
+
+        Args:
+            connection: Writer-owned connection.
+            force: Whether to free additional space after a full-disk error.
+        """
+        max_mb = self.config.get("trace_db_max_mb", 80)
+        max_mb = (
+            max_mb
+            if isinstance(max_mb, int) and not isinstance(max_mb, bool) and max_mb > 0
+            else 80
+        )
+        days = self.config.get("trace_retention_days", 7)
+        days = (
+            days
+            if isinstance(days, int) and not isinstance(days, bool) and days > 0
+            else 7
+        )
+        connection.execute(
+            "DELETE FROM traces WHERE state != 'running' AND updated_ns < ?",
+            (time.time_ns() - days * 86400 * 10**9,),
+        )
+        logical = sum(
+            connection.execute(
+                f"SELECT COALESCE(SUM(size_bytes), 0) FROM {table}"
+            ).fetchone()[0]
+            for table in ("traces", "spans", "events")
+        )
+        physical = connection.execute("PRAGMA page_count").fetchone()[0] * 4096
+        free = connection.execute("PRAGMA freelist_count").fetchone()[0] * 4096
+        budget = max_mb * 1024 * 1024
+        if force or logical > budget * 0.75 or physical - free > budget * 0.85:
+            target = max(0, logical - max(budget // 4, logical - int(budget * 0.6)))
+            while logical > target:
+                oldest = connection.execute(
+                    "SELECT trace_id FROM traces WHERE state != 'running' ORDER BY start_ns, trace_id LIMIT 32"
+                ).fetchall()
+                if not oldest:
+                    break
+                connection.executemany("DELETE FROM traces WHERE trace_id = ?", oldest)
+                logical = sum(
+                    connection.execute(
+                        f"SELECT COALESCE(SUM(size_bytes), 0) FROM {table}"
+                    ).fetchone()[0]
+                    for table in ("traces", "spans", "events")
+                )
+        connection.commit()
+        connection.execute("PRAGMA incremental_vacuum(256)")
+        connection.execute("PRAGMA wal_checkpoint(TRUNCATE)")
+        connection.execute(f"PRAGMA max_page_count={max(256, budget // 4096)}")
+
+    def _write(self, connection: sqlite3.Connection, batch: list[dict]) -> None:
+        """Upsert bounded snapshots in one transaction.
+
+        Args:
+            connection: Writer-owned connection.
+            batch: Snapshots to persist atomically.
+        """
+        with connection:
+            for payload in batch:
+                summary = payload["trace"]
+                trace_id = summary["trace_id"]
+                data = json.dumps(summary, ensure_ascii=False, separators=(",", ":"))
+                attrs = summary["attributes"]
+                search = " ".join(
+                    str(attrs.get("astrbot." + key, ""))
+                    for key in ("umo", "sender_name", "message_outline", "plugins")
+                )
+                connection.execute(
+                    """INSERT INTO traces VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    ON CONFLICT(trace_id) DO UPDATE SET
+                    updated_ns=excluded.updated_ns, state=excluded.state,
+                    has_errors=MAX(traces.has_errors, excluded.has_errors),
+                    has_models=MAX(traces.has_models, excluded.has_models),
+                    has_plugins=MAX(traces.has_plugins, excluded.has_plugins),
+                    search=excluded.search, data=excluded.data, size_bytes=excluded.size_bytes
+                    WHERE excluded.updated_ns >= traces.updated_ns""",
+                    (
+                        trace_id,
+                        int(summary["start_time_unix_nano"]),
+                        int(payload["time_unix_nano"]),
+                        summary["state"],
+                        summary["has_errors"],
+                        summary["has_models"],
+                        summary["has_plugins"],
+                        search,
+                        data,
+                        len(data.encode()),
+                        summary["root_span_id"],
+                    ),
+                )
+                if "span" in payload:
+                    span = payload["span"]
+                    data = json.dumps(span, ensure_ascii=False, separators=(",", ":"))
+                    connection.execute(
+                        """INSERT INTO spans VALUES (?, ?, ?, ?, ?)
+                        ON CONFLICT(span_id) DO UPDATE SET data=excluded.data, size_bytes=excluded.size_bytes
+                        WHERE json_extract(spans.data, '$.end_time_unix_nano') IS NULL
+                            OR json_extract(excluded.data, '$.end_time_unix_nano') IS NOT NULL""",
+                        (
+                            span["span_id"],
+                            trace_id,
+                            span["parent_span_id"],
+                            data,
+                            len(data.encode()),
+                        ),
+                    )
+                if "event" in payload:
+                    event = payload["event"]
+                    data = json.dumps(event, ensure_ascii=False, separators=(",", ":"))
+                    connection.execute(
+                        "INSERT OR IGNORE INTO events VALUES (?, ?, ?, ?, ?, ?)",
+                        (
+                            event["event_id"],
+                            trace_id,
+                            event["span_id"],
+                            int(event["time_unix_nano"]),
+                            data,
+                            len(data.encode()),
+                        ),
+                    )
+
+    def _run(self) -> None:
+        """Own the connection, batch writes, and perform bounded retention."""
+        connection = None
+        try:
+            self.path.parent.mkdir(parents=True, exist_ok=True)
+            connection = sqlite3.connect(self.path, timeout=1)
+            tables = {
+                row[0]
+                for row in connection.execute(
+                    "SELECT name FROM sqlite_master WHERE type='table'"
+                )
+            }
+            if tables and "trace_meta" not in tables:
+                raise ValueError(
+                    "Trace path contains a different database; choose a separate path."
+                )
+            application_id = connection.execute("PRAGMA application_id").fetchone()[0]
+            if application_id not in (0, 0x41535452):
+                raise ValueError("Trace path belongs to a different application.")
+            if "trace_meta" in tables:
+                if (
+                    connection.execute("SELECT version FROM trace_meta").fetchone()[0]
+                    != 1
+                ):
+                    raise ValueError("Unsupported trace database schema version.")
+            connection.execute("PRAGMA application_id=1095980114")
+            connection.execute("PRAGMA page_size=4096")
+            connection.execute("PRAGMA auto_vacuum=INCREMENTAL")
+            connection.execute("PRAGMA foreign_keys=ON")
+            connection.execute("PRAGMA journal_mode=WAL")
+            connection.execute("PRAGMA journal_size_limit=4194304")
+            connection.execute("PRAGMA wal_autocheckpoint=128")
+            connection.executescript("""
+                CREATE TABLE IF NOT EXISTS trace_meta(version INTEGER NOT NULL);
+                INSERT INTO trace_meta SELECT 1 WHERE NOT EXISTS (SELECT 1 FROM trace_meta);
+                CREATE TABLE IF NOT EXISTS traces(
+                    trace_id TEXT PRIMARY KEY, start_ns INTEGER NOT NULL,
+                    updated_ns INTEGER NOT NULL, state TEXT NOT NULL,
+                    has_errors INTEGER NOT NULL, has_models INTEGER NOT NULL,
+                    has_plugins INTEGER NOT NULL, search TEXT NOT NULL,
+                    data TEXT NOT NULL, size_bytes INTEGER NOT NULL,
+                    root_span_id TEXT NOT NULL);
+                CREATE INDEX IF NOT EXISTS traces_time ON traces(start_ns DESC, trace_id DESC);
+                CREATE INDEX IF NOT EXISTS traces_errors ON traces(has_errors, start_ns DESC);
+                CREATE INDEX IF NOT EXISTS traces_models ON traces(has_models, start_ns DESC);
+                CREATE INDEX IF NOT EXISTS traces_plugins ON traces(has_plugins, start_ns DESC);
+                CREATE TABLE IF NOT EXISTS spans(
+                    span_id TEXT PRIMARY KEY, trace_id TEXT NOT NULL REFERENCES traces(trace_id) ON DELETE CASCADE,
+                    parent_span_id TEXT, data TEXT NOT NULL, size_bytes INTEGER NOT NULL);
+                CREATE INDEX IF NOT EXISTS spans_trace ON spans(trace_id);
+                CREATE TABLE IF NOT EXISTS events(
+                    event_id TEXT PRIMARY KEY, trace_id TEXT NOT NULL REFERENCES traces(trace_id) ON DELETE CASCADE,
+                    span_id TEXT NOT NULL, time_ns INTEGER NOT NULL,
+                    data TEXT NOT NULL, size_bytes INTEGER NOT NULL);
+                CREATE INDEX IF NOT EXISTS events_trace ON events(trace_id, time_ns);
+            """)
+            if connection.execute("SELECT version FROM trace_meta").fetchone()[0] != 1:
+                raise ValueError("Unsupported trace database schema version.")
+            # A process restart cannot resume spans from a previous process.
+            connection.execute("""UPDATE traces SET state='interrupted', has_errors=1,
+                data=json_set(data, '$.state', 'interrupted', '$.has_errors', json('true'),
+                    '$.end_time_unix_nano', CAST(updated_ns AS TEXT))
+                WHERE state='running'""")
+            connection.execute("""UPDATE spans SET data=json_set(data,
+                '$.end_time_unix_nano', CAST((SELECT updated_ns FROM traces WHERE traces.trace_id=spans.trace_id) AS TEXT),
+                '$.status.code', 2, '$.status.message', 'Process stopped before completion')
+                WHERE json_extract(data, '$.end_time_unix_nano') IS NULL""")
+            connection.commit()
+            self._prune(connection)
+            self._ready.set()
+            last_prune = time.monotonic()
+            while not self._stop.is_set() or not self.queue.empty():
+                try:
+                    first = self.queue.get(timeout=0.1)
+                except Empty:
+                    if time.monotonic() - last_prune >= 60:
+                        self._prune(connection)
+                        last_prune = time.monotonic()
+                    continue
+                items = [first]
+                for _ in range(31):
+                    try:
+                        items.append(self.queue.get_nowait())
+                    except Empty:
+                        break
+                batch = [json.loads(item) for item, _ in items if isinstance(item, str)]
+                try:
+                    wal = Path(str(self.path) + "-wal")
+                    if wal.exists() and wal.stat().st_size > QUEUE_BYTES:
+                        self._prune(connection)
+                        if wal.exists() and wal.stat().st_size > QUEUE_BYTES:
+                            raise sqlite3.OperationalError(
+                                "Trace WAL cannot checkpoint within its capacity budget."
+                            )
+                    try:
+                        self._write(connection, batch)
+                    except sqlite3.OperationalError as exc:
+                        # A lock or I/O error must not trigger early eviction.
+                        if "database or disk is full" not in str(exc).lower():
+                            raise
+                        self._prune(connection, force=True)
+                        self._write(connection, batch)
+                    if batch:
+                        self.last_error = None
+                    if time.monotonic() - last_prune >= 10:
+                        self._prune(connection)
+                        last_prune = time.monotonic()
+                except Exception as exc:
+                    self.last_error = str(exc)
+                    self.dropped += len(batch)
+                    logger.error("Trace database write failed: %s", exc)
+                finally:
+                    for item, size in items:
+                        with self._lock:
+                            self._pending_bytes -= size
+                        if isinstance(item, threading.Event):
+                            item.set()
+                        self.queue.task_done()
+        except Exception as exc:
+            self._failed = True
+            self.last_error = str(exc)
+            logger.error("Trace database initialization failed: %s", exc)
+            while True:
+                try:
+                    item, size = self.queue.get_nowait()
+                except Empty:
+                    break
+                with self._lock:
+                    self._pending_bytes -= size
+                    self.dropped += int(size > 0)
+                if isinstance(item, threading.Event):
+                    item.set()
+                self.queue.task_done()
+        finally:
+            self._ready.set()
+            if connection is not None:
+                connection.close()
+
+    def list_traces(
+        self,
+        *,
+        limit: int = 50,
+        cursor: str | None = None,
+        query: str = "",
+        errors: bool = False,
+        models: bool = False,
+        plugins: bool = False,
+    ) -> dict:
+        """Read one keyset page of summaries, never all stored span bodies.
+
+        Args:
+            limit: Maximum summaries, capped at 100.
+            cursor: Previous page's timestamp and trace ID.
+            query: Message, session, sender, or plugin search.
+            errors: Require an error.
+            models: Require an Agent call.
+            plugins: Require a plugin call.
+
+        Returns:
+            Summaries, filtered count, next cursor, and storage health.
+
+        Raises:
+            RuntimeError: If initialization failed.
+            ValueError: If the cursor is invalid.
+        """
+        if not self._ready.wait(5) or self._failed:
+            raise RuntimeError(self.last_error or "Trace database is not ready.")
+        limit = min(100, max(1, limit))
+        conditions, values = [], []
+        for key, enabled in (
+            ("has_errors", errors),
+            ("has_models", models),
+            ("has_plugins", plugins),
+        ):
+            if enabled:
+                conditions.append(key + "=1")
+        if query:
+            conditions.append("search LIKE ? ESCAPE '\\'")
+            values.append(
+                "%"
+                + query[:256]
+                .replace("\\", "\\\\")
+                .replace("%", "\\%")
+                .replace("_", "\\_")
+                + "%"
+            )
+        where = " WHERE " + " AND ".join(conditions) if conditions else ""
+        with closing(sqlite3.connect(self.path, timeout=1)) as connection:
+            connection.execute("BEGIN")
+            total = connection.execute(
+                "SELECT COUNT(*) FROM traces" + where, values
+            ).fetchone()[0]
+            if cursor:
+                stamp, trace_id = cursor.split(":", 1)
+                if len(trace_id) != 32 or any(
+                    c not in "0123456789abcdef" for c in trace_id
+                ):
+                    raise ValueError("Invalid trace cursor.")
+                conditions.append("(start_ns < ? OR (start_ns = ? AND trace_id < ?))")
+                values.extend((int(stamp), int(stamp), trace_id))
+            where = " WHERE " + " AND ".join(conditions) if conditions else ""
+            rows = connection.execute(
+                "SELECT data, start_ns, trace_id FROM traces"
+                + where
+                + " ORDER BY start_ns DESC, trace_id DESC LIMIT ?",
+                [*values, limit + 1],
+            ).fetchall()
+        return {
+            "traces": [json.loads(row[0]) for row in rows[:limit]],
+            "total": total,
+            "cursor": f"{rows[limit - 1][1]}:{rows[limit - 1][2]}"
+            if len(rows) > limit
+            else None,
+            "storage": {"dropped": self.dropped, "error": self.last_error},
+        }
+
+    def get_trace(self, trace_id: str) -> dict | None:
+        """Read one bounded trace and its spans/events.
+
+        Args:
+            trace_id: ID of the requested trace.
+
+        Returns:
+            Complete detail, or None when evicted or absent.
+
+        Raises:
+            RuntimeError: If storage initialization failed.
+        """
+        if not self._ready.wait(5) or self._failed:
+            raise RuntimeError(self.last_error or "Trace database is not ready.")
+        connection = sqlite3.connect(self.path, timeout=1)
+        try:
+            connection.execute("BEGIN")
+            row = connection.execute(
+                "SELECT data FROM traces WHERE trace_id=?", (trace_id,)
+            ).fetchone()
+            if not row:
+                return None
+            spans = [
+                json.loads(item[0])
+                for item in connection.execute(
+                    "SELECT data FROM spans WHERE trace_id=? LIMIT 128", (trace_id,)
+                )
+            ]
+            events = [
+                json.loads(item[0])
+                for item in connection.execute(
+                    "SELECT data FROM events WHERE trace_id=? ORDER BY time_ns, event_id LIMIT 256",
+                    (trace_id,),
+                )
+            ]
+            by_id = {span["span_id"]: span for span in spans}
+            for event in events:
+                owner = by_id.get(event["span_id"])
+                if owner is not None:
+                    owner.setdefault("events", []).append(event)
+            return {"trace": json.loads(row[0]), "spans": spans}
+        finally:
+            connection.close()
+
+
+_store: TraceStore | None = None
+_store_lock = threading.Lock()
+
+
+def get_trace_store(config: Mapping[str, Any] | None = None) -> TraceStore:
+    """Get the process-wide dedicated writer.
+
+    Args:
+        config: Global configuration, or None to resolve it lazily.
+
+    Returns:
+        The local trace store.
+    """
+    global _store
+    with _store_lock:
+        if _store is None:
+            if config is None:
+                from astrbot.core import astrbot_config
+
+                config = astrbot_config
+            path = Path(config.get("trace_db_path") or "logs/traces.db")
+            if not path.is_absolute():
+                path = Path(get_astrbot_data_path()) / path
+            _store = TraceStore(path.resolve(), config)
+            atexit.register(_store.close)
+        return _store
+
+
+def close_trace_store() -> None:
+    """Flush an initialized trace store without creating a new one."""
+    if _store is not None:
+        _store.close()

astrbot/dashboard/api/logs.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from fastapi import APIRouter, Depends, Header, Request
+from fastapi import APIRouter, Depends, Header, Path, Query, Request
 from fastapi.responses import StreamingResponse
 
 from astrbot.dashboard.responses import ApiError, ok
@@ -88,6 +88,49 @@ async def get_trace_settings(
     return _get_trace_settings(service)
 
 
+@router.get("/trace/history")
+async def list_traces(
+    limit: int = Query(default=50, ge=1, le=100),
+    cursor: str | None = Query(default=None, max_length=64),
+    query: str = Query(default="", max_length=256),
+    errors: bool = False,
+    models: bool = False,
+    plugins: bool = False,
+    _auth: AuthContext = Depends(require_system_scope),
+    service: LogService = Depends(get_service),
+):
+    """Return a page of persisted trace summaries."""
+    try:
+        return ok(
+            await service.get_trace_history(
+                limit=limit,
+                cursor=cursor,
+                query=query,
+                errors=errors,
+                models=models,
+                plugins=plugins,
+            )
+        )
+    except LogServiceError as exc:
+        _raise_log_error(exc)
+
+
+@router.get("/trace/{trace_id}")
+async def get_trace_detail(
+    trace_id: str = Path(pattern="^[0-9a-f]{32}$"),
+    _auth: AuthContext = Depends(require_system_scope),
+    service: LogService = Depends(get_service),
+):
+    """Return one trace's canonical spans and events."""
+    try:
+        detail = await service.get_trace_detail(trace_id)
+    except LogServiceError as exc:
+        _raise_log_error(exc)
+    if detail is None:
+        raise ApiError("Trace not found or already evicted.", status_code=404)
+    return ok(detail)
+
+
 @router.put("/trace/settings")
 async def update_trace_settings(
     payload: TraceSettingsRequest,

astrbot/dashboard/services/log_service.py

@@ -7,6 +7,7 @@ from collections.abc import AsyncGenerator
 
 from astrbot.core import LogBroker, logger
 from astrbot.core.config.astrbot_config import AstrBotConfig
+from astrbot.core.utils.trace_store import get_trace_store
 
 
 class LogServiceError(Exception):
@@ -76,6 +77,45 @@ class LogService:
             logger.error(f"获取 Trace 设置失败: {exc}")
             raise LogServiceError(f"获取 Trace 设置失败: {exc}") from exc
 
+    async def get_trace_history(self, **filters) -> dict:
+        """Read a bounded database page.
+
+        Args:
+            **filters: Validated cursor, keyword, and feature filters.
+
+        Returns:
+            Trace summaries and pagination metadata.
+
+        Raises:
+            LogServiceError: If the cursor or database is unavailable.
+        """
+        try:
+            store = get_trace_store(self.config)
+            return await asyncio.to_thread(store.list_traces, **filters)
+        except Exception as exc:
+            logger.error("Failed to query trace history: %s", exc)
+            raise LogServiceError(str(exc)) from exc
+
+    async def get_trace_detail(self, trace_id: str) -> dict | None:
+        """Read the spans and events of a single trace.
+
+        Args:
+            trace_id: Validated trace identifier.
+
+        Returns:
+            Trace detail, or None if evicted.
+
+        Raises:
+            LogServiceError: If database access fails.
+        """
+        try:
+            return await asyncio.to_thread(
+                get_trace_store(self.config).get_trace, trace_id
+            )
+        except Exception as exc:
+            logger.error("Failed to query trace detail: %s", exc)
+            raise LogServiceError(str(exc)) from exc
+
     def update_trace_settings(self, payload: dict | None) -> str:
         try:
             if payload is None:

changelogs/unreleased.md

@@ -16,9 +16,11 @@
 - Prefer session aliases and plugin display names with hover details; show platform/plugin logos and lazily load handler descriptions from plugin details.
 - Collapse consecutive same-type timeline nodes with stacked icons and visible error counts.
 - Record only the final streaming-response summary, excluding adapter chunks and protocol closing frames while retaining independent tool/plugin replies.
-- Keep the existing in-memory trace storage; no new database persistence.
+- Replace the in-memory trace record cache and JSONL sink with dedicated SQLite history at `data/logs/traces.db`, using bounded asynchronous writes, whole-trace retention, and a default 80 MB capacity target / 7-day retention. Preserve legacy files without importing or appending them.
+- Align trace/span IDs, parent relationships, timestamps, status, attributes, and events with the OpenTelemetry core model, and correlate structured logs with the active trace/span context. This does not add an OTel SDK, OTLP transport, or external export.
+- Migrate `trace_log_enable/path/max_mb` to `trace_db_path/max_mb` and `trace_retention_days`; keep `trace_enable`. New database settings take precedence. Protect SQLite files and sidecars from log cleanup.
 - Show execution mode, plugin names, and response ownership as timeline chips. Reuse profile runner labels and classify built-in Agent replies as system/core while preserving plugin senders.
 - Remove the outer gray trace card and tinted list shell, using a flat tab layout with toolbar/table separators in both themes.
 - Top hint and separate recording switch → a toolbar Start/Pause/Connecting button. Filters, recording, and refresh share gray icon-and-text styling and 36px input height; search fills available width and includes the message count in its placeholder.
-- Use Vuetify datatable with native rows, typography, spacing, and hover styling for the trace list; customize individual cells while retaining fixed headers, stable trace row IDs, keyboard activation, and explicit detail selection without adding pagination or automatic selection.
+- Use Vuetify datatable with native rows, typography, spacing, and hover styling for the trace list; retain fixed headers, stable trace row IDs, keyboard activation, and explicit detail selection. Read history in keyset pages, cap browser summaries at 300, and lazily fetch only the selected trace's detail. Spans now counts actual spans, excluding events.
 - Shorten the Chinese duration column label and prevent the platform column header from wrapping with a minimum width.

dashboard/src/api/generated/openapi-v1/types.gen.ts

@@ -3725,6 +3725,31 @@ export type GetLogHistoryResponse = (string);
 
 export type GetLogHistoryError = unknown;
 
+export type ListTracesData = {
+    query?: {
+        cursor?: string;
+        errors?: boolean;
+        limit?: number;
+        models?: boolean;
+        plugins?: boolean;
+        query?: string;
+    };
+};
+
+export type ListTracesResponse = (SuccessEnvelope);
+
+export type ListTracesError = unknown;
+
+export type GetTraceDetailData = {
+    path: {
+        trace_id: string;
+    };
+};
+
+export type GetTraceDetailResponse = (SuccessEnvelope);
+
+export type GetTraceDetailError = (unknown);
+
 export type GetTraceSettingsResponse = (SuccessEnvelope);
 
 export type GetTraceSettingsError = unknown;

dashboard/src/api/v1.ts

@@ -688,6 +688,12 @@ export const apiKeyApi = {
 };
 
 export const traceApi = {
+  history(query?: { limit?: number; cursor?: string; query?: string; errors?: boolean; models?: boolean; plugins?: boolean }) {
+    return typed<OpenConfig>(openApiV1.listTraces({ query }));
+  },
+  detail(traceId: string) {
+    return typed<OpenConfig>(openApiV1.getTraceDetail({ path: { trace_id: traceId } }));
+  },
   getSettings() {
     return typed<OpenConfig>(openApiV1.getTraceSettings());
   },

dashboard/src/components/shared/TraceDisplayer.vue

@@ -17,7 +17,7 @@ import {
 } from "@lucide/vue";
 import { computed, nextTick, onBeforeUnmount, onMounted, reactive, ref, watch } from "vue";
 import { EventSourcePolyfill } from "event-source-polyfill";
-import { logApi, pluginApi, sessionApi } from "@/api/v1";
+import { logApi, traceApi, pluginApi, sessionApi } from "@/api/v1";
 import { useModuleI18n } from "@/i18n/composables";
 import { getPlatformIcon } from "@/utils/platformUtils";
 import defaultPluginIcon from "/favicon.svg";
@@ -36,6 +36,18 @@ const searchText = ref("");
 const filtersOpen = ref(false);
 const selectedFilters = ref([]);
 const loading = ref(true);
+const historyBusy = ref(false);
+const detailBusy = ref(false);
+const historyError = ref("");
+const detailError = ref("");
+const totalRecords = ref(0);
+const nextCursor = ref(null);
+const storageWarning = ref(false);
+let metadataLoaded = false;
+let historyRequest = 0;
+let detailRequest = 0;
+let queryTimer = null;
+let historyRefreshTimer = null;
 const clockNow = ref(Date.now() / 1000);
 const durationTimer = ref(null);
 const connected = ref(false);
@@ -107,6 +119,7 @@ const filteredEvents = computed(() => {
       event.umo,
       umoNames.value[event.umo],
       event.sender_name,
+      event.plugin_names,
       ...event.records.map((record) => recordSearchText(record)),
     ]
       .filter(Boolean)
@@ -124,6 +137,8 @@ const filters = computed(() => [
 
 watch(filteredEvents, (visibleEvents) => {
   if (selectedSpanId.value && !visibleEvents.some((event) => event.span_id === selectedSpanId.value)) {
+    const previous = eventIndex.get(selectedSpanId.value);
+    if (previous?.model_version === 3) previous.records = [];
     selectedSpanId.value = null;
   }
 });
@@ -131,7 +146,7 @@ watch(filteredEvents, (visibleEvents) => {
 const traceRows = computed(() => filteredEvents.value.map((event) => ({
   event,
   duration: formatDuration(Math.max(0, ((event.completed ? event.last_time : Math.max(clockNow.value, event.last_time)) - event.first_time) * 1000)),
-  spans: timelineRecords(event).length,
+  spans: event.span_count ?? timelineRecords(event).length,
 })));
 
 const traceHeaders = computed(() => [
@@ -203,7 +218,9 @@ function timelineRecords(event) {
       const startAction = record.action === "handler_end" ? "handler_start" : "agent_start";
       const start = [...records].reverse().find((item) =>
         item.action === startAction && !item.completed &&
-        item.fields.handler_full_name === record.fields.handler_full_name,
+        (item.operation_span_id && record.operation_span_id
+          ? item.operation_span_id === record.operation_span_id
+          : item.fields.handler_full_name === record.fields.handler_full_name),
       );
       if (start) {
         start.fields = { ...start.fields, ...record.fields };
@@ -219,11 +236,11 @@ function timelineRecords(event) {
 function matchesFilter(event, filter) {
   if (filter === "errors") return event.status === "error";
   if (filter === "models") {
-    return event.records.some((record) => record.kind === "agent" ||
+    return event.has_models ?? event.records.some((record) => record.kind === "agent" ||
       ["astr_agent_prepare", "astr_agent_complete"].includes(record.action));
   }
   if (filter === "plugins") {
-    return event.records.some((record) => record.kind === "plugin_handler");
+    return event.has_plugins ?? event.records.some((record) => record.kind === "plugin_handler");
   }
   return true;
 }
@@ -243,60 +260,138 @@ function recordSearchText(record) {
     .join(" ");
 }
 
+function normalizeTracePayload(payload) {
+  const summary = payload.trace;
+  const attrs = summary.attributes || {};
+  const finished = summary.state !== "running";
+  const time = Number(payload.time_unix_nano || summary.updated_time_unix_nano || summary.start_time_unix_nano) / 1e9;
+  const meta = {
+    span_id: summary.trace_id, name: summary.name, model_version: 3,
+    umo: attrs["astrbot.umo"] || "", sender_name: attrs["astrbot.sender_name"] || "",
+    message_outline: attrs["astrbot.message_outline"] || "",
+    plugin_names: attrs["astrbot.plugins"] || "",
+    platform_id: attrs["astrbot.platform_id"] || "",
+    platform_name: attrs["astrbot.platform_name"] || "",
+    time: Number(summary.start_time_unix_nano) / 1e9,
+    last_time: Number(summary.end_time_unix_nano) / 1e9 || time,
+    completed: finished, status: summary.has_errors ? "error" : finished ? "success" : "running",
+    has_models: summary.has_models, has_plugins: summary.has_plugins,
+    span_count: summary.span_count, truncated: summary.truncated,
+    model_updated_ns: String(summary.updated_time_unix_nano || summary.start_time_unix_nano),
+  };
+  const records = [];
+  const spans = payload.spans || (payload.span ? [payload.span] : []);
+  const items = [
+    ...spans.map((span) => ({ span })),
+    ...spans.flatMap((span) => (span.events || []).map((event) => ({ event }))),
+    ...(payload.event ? [{ event: payload.event }] : []),
+  ];
+  for (const item of items) {
+    const model = item.span || item.event;
+    const inspectedModel = item.span ? { ...model, events: undefined } : model;
+    const fields = {};
+    for (const [key, value] of Object.entries(model.attributes || {})) {
+      if (!key.startsWith("astrbot.")) continue;
+      const name = key.slice(8);
+      fields[name] = value;
+      if (["chat_provider", "stats", "tools", "tool_name"].includes(name) && typeof value === "string") {
+        try { fields[name] = JSON.parse(value); } catch { /* Plain text is valid. */ }
+      }
+    }
+    if (item.event) {
+      records.push({
+        span_id: summary.trace_id, record_id: model.event_id,
+        action: model.name, time: Number(model.time_unix_nano) / 1e9,
+        fields, kind: fields.kind, model: inspectedModel,
+      });
+      continue;
+    }
+    const root = model.span_id === summary.root_span_id;
+    const action = root ? "event_received" : fields.action;
+    if (!action) continue;
+    const started = Number(model.start_time_unix_nano) / 1e9;
+    records.push({
+      span_id: summary.trace_id, record_id: model.span_id + ":start",
+      operation_span_id: model.span_id, parent_span_id: model.parent_span_id,
+      action, time: started,
+      fields: { ...fields, kind: root ? "message" : fields.kind, status: "running" },
+      kind: root ? "message" : fields.kind, model: inspectedModel,
+    });
+    if (model.end_time_unix_nano) {
+      const ended = Number(model.end_time_unix_nano) / 1e9;
+      records.push({
+        span_id: summary.trace_id, record_id: model.span_id + ":end",
+        operation_span_id: model.span_id, parent_span_id: model.parent_span_id,
+        action: root ? "pipeline_end" : fields.end_action,
+        time: ended, kind: fields.kind,
+        fields: { ...fields, duration_ms: Math.max(0, (ended - started) * 1000),
+          status: model.status?.code === 2 ? "error" : "success",
+          error: model.status?.message || fields.error },
+        model: inspectedModel,
+      });
+    }
+  }
+  return { meta, records };
+}
+
 function processNewTraces(newTraces) {
   if (!Array.isArray(newTraces)) return;
-
-  for (const trace of newTraces) {
+  for (const packet of newTraces) {
+    const normalized = packet?.trace ? normalizeTracePayload(packet) : null;
+    const trace = normalized?.meta || packet;
     if (!trace?.span_id) continue;
     if (["stage_start", "stage_end", "pipeline_start", "wake_evaluated"].includes(trace.action)) continue;
     let event = eventIndex.get(trace.span_id);
     if (!event) {
       event = reactive({
-        span_id: trace.span_id,
-        name: trace.name || "AstrMessageEvent",
-        umo: trace.umo || "",
-        sender_name: trace.sender_name || "",
+        span_id: trace.span_id, name: trace.name || "AstrMessageEvent",
+        umo: trace.umo || "", sender_name: trace.sender_name || "",
         message_outline: trace.message_outline || "",
-        platform_id: "",
-        platform_name: "",
-        platform_icon: "",
+        platform_id: "", platform_name: "", platform_icon: "",
         first_time: trace.time || Date.now() / 1000,
         last_time: trace.time || Date.now() / 1000,
-        status: "running",
-        completed: false,
-        records: [],
+        status: "running", completed: false, records: [],
       });
       eventIndex.set(trace.span_id, event);
       events.value.unshift(event);
     }
-
-    event.umo = event.umo || trace.umo || "";
-    event.sender_name = event.sender_name || trace.sender_name || "";
-    event.message_outline = event.message_outline || trace.message_outline || "";
-    event.platform_id = trace.fields?.platform_id || event.platform_id;
-    if (trace.fields?.platform_name) {
-      event.platform_name = trace.fields.platform_name;
-      event.platform_icon = getPlatformIcon(event.platform_name) || "";
+    if (normalized) {
+      if (!event.model_updated_ns || trace.model_updated_ns.padStart(20, "0") >= event.model_updated_ns.padStart(20, "0")) {
+        Object.assign(event, trace);
+        event.first_time = trace.time;
+        event.platform_icon = getPlatformIcon(trace.platform_name) || "";
+      }
+    } else {
+      event.umo = event.umo || trace.umo || "";
+      event.sender_name = event.sender_name || trace.sender_name || "";
+      event.message_outline = event.message_outline || trace.message_outline || "";
+      event.platform_id = trace.fields?.platform_id || event.platform_id;
+      if (trace.fields?.platform_name) {
+        event.platform_name = trace.fields.platform_name;
+        event.platform_icon = getPlatformIcon(event.platform_name) || "";
+      }
+      event.first_time = Math.min(event.first_time, trace.time || event.first_time);
+      event.last_time = Math.max(event.last_time, trace.time || event.last_time);
     }
-    event.first_time = Math.min(event.first_time, trace.time || event.first_time);
-    event.last_time = Math.max(event.last_time, trace.time || event.last_time);
-
-    const recordKey = trace.record_id || `${trace.span_id}:${trace.sequence ?? trace.time}:${trace.action}`;
-    if (!event.records.some((record) => record.key === recordKey)) {
-      event.records.push({ ...trace, key: recordKey, kind: trace.kind || trace.fields?.kind, fields: trace.fields || {} });
-      event.records.sort((a, b) => (a.time || 0) - (b.time || 0));
-    }
-    if (trace.fields?.message && !event.message_outline && trace.action === "event_received") {
-      event.message_outline = trace.fields.message;
-    }
-    if (trace.action === "pipeline_end") event.completed = true;
-    if (trace.status === "error" || trace.fields?.status === "error") {
-      event.status = "error";
-    } else if (trace.action === "pipeline_end") {
-      event.status = event.status === "error" ? "error" : "success";
+    // Only the inspected trace retains span bodies. Other live packets update
+    // bounded list summaries; history is paged independently from SQLite.
+    const records = normalized
+      ? selectedSpanId.value === trace.span_id || packet.spans ? normalized.records : []
+      : [trace];
+    for (const record of records) {
+      const recordKey = record.record_id || `${trace.span_id}:${record.sequence ?? record.time}:${record.action}`;
+      const value = { ...record, key: recordKey, kind: record.kind || record.fields?.kind, fields: record.fields || {} };
+      const existing = event.records.findIndex((item) => item.key === recordKey);
+      if (existing === -1) event.records.push(value);
+      else event.records[existing] = value;
+      if (!normalized) {
+        if (record.action === "pipeline_end") event.completed = true;
+        if (record.status === "error" || record.fields?.status === "error") event.status = "error";
+        else if (record.action === "pipeline_end") event.status = event.status === "error" ? "error" : "success";
+      }
     }
+    event.records.sort((a, b) => (a.time || 0) - (b.time || 0));
   }
-
   events.value.sort((a, b) => b.first_time - a.first_time);
   if (events.value.length > maxItems) {
     const removed = events.value.splice(maxItems);
@@ -304,36 +399,87 @@ function processNewTraces(newTraces) {
   }
 }
 
-async function fetchTraceHistory() {
-  try {
-    const response = await pluginApi.list();
-    plugins.value = Object.fromEntries(
-      (response.data?.data || []).filter((plugin) => plugin.name).map((plugin) => [plugin.name, plugin]),
-    );
-  } catch (error) {
-    console.error("Failed to fetch trace plugin metadata:", error);
+async function fetchTraceHistory(reset = true, reloadMetadata = false, preserveLoaded = false) {
+  if (!reset && (historyBusy.value || !nextCursor.value || events.value.length >= maxItems)) return;
+  const request = ++historyRequest;
+  historyBusy.value = true;
+  historyError.value = "";
+  if (!metadataLoaded || reloadMetadata) {
+    try {
+      const response = await pluginApi.list();
+      plugins.value = Object.fromEntries(
+        (response.data?.data || []).filter((plugin) => plugin.name).map((plugin) => [plugin.name, plugin]),
+      );
+    } catch (error) {
+      console.error("Failed to fetch trace plugin metadata:", error);
+    }
+    try {
+      const response = await sessionApi.activeUmos();
+      umoNames.value = Object.fromEntries(
+        (response.data?.data?.umo_infos || []).map((info) => [
+          info.umo, info.user_alias || info.auto_name || info.display_name || info.umo,
+        ]),
+      );
+    } catch (error) {
+      console.error("Failed to fetch session aliases:", error);
+    }
+    metadataLoaded = true;
   }
   try {
-    const response = await sessionApi.activeUmos();
-    umoNames.value = Object.fromEntries(
-      (response.data?.data?.umo_infos || []).map((info) => [
-        info.umo, info.user_alias || info.auto_name || info.display_name || info.umo,
-      ]),
-    );
-  } catch (error) {
-    console.error("Failed to fetch session aliases:", error);
-  }
-  try {
-    const response = await logApi.history();
-    const logs = response.data?.data?.logs || [];
-    processNewTraces(logs.filter((item) => item.type === "trace"));
+    const response = await traceApi.history({
+      limit: 50, cursor: reset ? undefined : nextCursor.value,
+      query: searchText.value.trim().slice(0, 256),
+      errors: selectedFilters.value.includes("errors"),
+      models: selectedFilters.value.includes("models"),
+      plugins: selectedFilters.value.includes("plugins"),
+    });
+    if (request !== historyRequest) return;
+    if (response.data?.status !== "ok") throw new Error("Trace history request failed");
+    const data = response.data.data;
+    if (reset && !preserveLoaded) {
+      const keep = new Set((data.traces || []).map((trace) => trace.trace_id));
+      if (selectedSpanId.value) keep.add(selectedSpanId.value);
+      events.value = events.value.filter((event) => keep.has(event.span_id));
+      for (const id of eventIndex.keys()) if (!keep.has(id)) eventIndex.delete(id);
+    }
+    processNewTraces((data.traces || []).map((trace) => ({ trace })));
+    totalRecords.value = data.total || 0;
+    nextCursor.value = data.cursor;
+    storageWarning.value = Boolean(data.storage?.error || data.storage?.dropped);
+    if (reset && selectedSpanId.value) loadTraceDetail(selectedSpanId.value);
   } catch (error) {
+    if (request === historyRequest) historyError.value = tm("history.loadFailed");
     console.error("Failed to fetch trace history:", error);
   } finally {
-    loading.value = false;
+    if (request === historyRequest) {
+      historyBusy.value = false;
+      loading.value = false;
+    }
   }
 }
 
+async function loadTraceDetail(traceId) {
+  const request = ++detailRequest;
+  detailBusy.value = true;
+  detailError.value = "";
+  try {
+    const response = await traceApi.detail(traceId);
+    if (request !== detailRequest || selectedSpanId.value !== traceId) return;
+    if (response.data?.status !== "ok") throw new Error("Trace detail request failed");
+    processNewTraces([response.data.data]);
+  } catch (error) {
+    if (request === detailRequest && selectedSpanId.value === traceId) detailError.value = tm("history.detailFailed");
+    console.error("Failed to fetch trace details:", error);
+  } finally {
+    if (request === detailRequest) detailBusy.value = false;
+  }
+}
+
+watch([searchText, selectedFilters], () => {
+  if (queryTimer) clearTimeout(queryTimer);
+  queryTimer = setTimeout(() => fetchTraceHistory(), 250);
+}, { deep: true });
+
 function connectSSE() {
   if (eventSource.value) eventSource.value.close();
   const token = localStorage.getItem("token");
@@ -346,12 +492,22 @@ function connectSSE() {
   source.onopen = () => {
     connected.value = true;
     retryAttempts.value = 0;
+    fetchTraceHistory(true, false, true);
   };
   source.onmessage = (event) => {
     if (event.lastEventId) lastEventId.value = event.lastEventId;
     try {
       const payload = JSON.parse(event.data);
-      if (payload?.type === "trace") processNewTraces([payload]);
+      if (payload?.type === "trace") {
+        processNewTraces([payload]);
+        if (payload.storage_accepted === false) storageWarning.value = true;
+        if (payload.span && (!payload.span.parent_span_id || selectedFilters.value.length) && !historyRefreshTimer) {
+          historyRefreshTimer = setTimeout(() => {
+            historyRefreshTimer = null;
+            fetchTraceHistory(true, false, true);
+          }, 1000);
+        }
+      }
     } catch (error) {
       console.error("Failed to parse trace payload:", error);
     }
@@ -371,7 +527,10 @@ function connectSSE() {
 
 function selectEvent(spanId) {
   const previousSpanId = selectedSpanId.value;
+  const previous = eventIndex.get(previousSpanId);
+  if (previous?.model_version === 3 && previousSpanId !== spanId) previous.records = [];
   selectedSpanId.value = spanId;
+  if (spanId && eventIndex.get(spanId)?.model_version === 3) loadTraceDetail(spanId);
   nextTick(() => {
     const row = document.querySelector(`[data-trace-id="${spanId || previousSpanId}"]`);
     row?.scrollIntoView({ block: "nearest" });
@@ -517,6 +676,10 @@ onMounted(async () => {
 });
 
 onBeforeUnmount(() => {
+  historyRequest++;
+  detailRequest++;
+  if (queryTimer) clearTimeout(queryTimer);
+  if (historyRefreshTimer) clearTimeout(historyRefreshTimer);
   if (durationTimer.value) clearInterval(durationTimer.value);
   eventSource.value?.close();
   if (retryTimer.value) clearTimeout(retryTimer.value);
@@ -540,7 +703,7 @@ onBeforeUnmount(() => {
       </v-btn>
       <div class="trace-search">
         <Search :size="16" aria-hidden="true" />
-        <input v-model="searchText" :placeholder="tm('search', { count: filteredEvents.length })" />
+        <input v-model="searchText" :placeholder="tm('search', { count: totalRecords })" :aria-label="tm('search', { count: totalRecords })" />
         <button v-if="searchText" type="button" class="icon-button" :aria-label="tm('clearSearch')" @click="clearSearch">
           <X :size="15" aria-hidden="true" />
         </button>
@@ -559,12 +722,14 @@ onBeforeUnmount(() => {
         <component :is="!recordingEnabled ? Play : connected ? Pause : LoaderCircle" :size="16" :class="{ 'trace-connecting-icon': recordingEnabled && !connected }" aria-hidden="true" />
         {{ recordingLabel }}
       </v-btn>
-      <v-btn class="refresh-button" size="small" variant="tonal" :aria-label="tm('refresh')" :title="tm('refresh')" @click="fetchTraceHistory">
+      <v-btn class="refresh-button" size="small" variant="tonal" :aria-label="tm('refresh')" :title="tm('refresh')" @click="fetchTraceHistory(true, true)">
         <RefreshCw :size="16" aria-hidden="true" />
         <span>{{ tm("refresh") }}</span>
       </v-btn>
     </div>
 
+    <v-alert v-if="historyError" type="error" variant="tonal" density="compact">{{ historyError }}</v-alert>
+    <v-alert v-if="storageWarning" type="warning" variant="tonal" density="compact">{{ tm("history.storageWarning") }}</v-alert>
     <div class="trace-workspace">
       <button v-if="filtersOpen" type="button" class="trace-filter-backdrop" :aria-label="tm('filters.close')" @click="filtersOpen = false" />
       <aside v-if="filtersOpen" id="trace-filter-panel" class="trace-filter-panel" :aria-label="tm('filters.title')" @keydown.esc="filtersOpen = false">
@@ -645,6 +810,10 @@ onBeforeUnmount(() => {
             <span :title="tm('columns.spansHint')">{{ item.spans }}</span>
           </template>
         </v-data-table>
+        <div v-if="nextCursor" class="trace-page-footer">
+          <v-btn v-if="events.length < maxItems" size="small" variant="tonal" :loading="historyBusy" @click="fetchTraceHistory(false)">{{ tm("history.loadMore") }}</v-btn>
+          <span v-else>{{ tm("history.limitHint", { count: maxItems }) }}</span>
+        </div>
       </section>
 
       <main v-if="selectedEvent" class="trace-detail" @keydown.esc.stop="selectEvent(null)">
@@ -667,7 +836,13 @@ onBeforeUnmount(() => {
           </div>
         </div>
 
-        <div class="trace-stat-grid">
+        <v-alert v-if="selectedEvent.truncated" class="mt-3" type="warning" variant="tonal" density="compact">{{ tm("history.truncated") }}</v-alert>
+        <div v-if="detailBusy" class="trace-empty-state" aria-live="polite"><LoaderCircle :size="20" class="trace-connecting-icon" /><span>{{ tm("loading") }}</span></div>
+        <v-alert v-if="detailError" class="mt-3" type="error" variant="tonal" density="compact">
+          {{ detailError }}
+          <v-btn size="small" variant="text" @click="loadTraceDetail(selectedSpanId)">{{ tm("refresh") }}</v-btn>
+        </v-alert>
+        <div v-if="!detailBusy && !detailError" class="trace-stat-grid">
           <div class="trace-stat"><Clock3 :size="15" /><span>{{ tm("duration") }}</span><strong>{{ selectedStats.duration }}</strong></div>
           <div class="trace-stat"><PlugZap :size="15" /><span>{{ tm("handlers") }}</span><strong>{{ selectedStats.handlers }}</strong></div>
           <div class="trace-stat"><Send :size="15" /><span>{{ tm("responses") }}</span><strong>{{ selectedStats.responses }}</strong></div>
@@ -755,7 +930,7 @@ onBeforeUnmount(() => {
                 <v-chip class="timeline-metadata-chip" size="x-small" variant="tonal" color="primary">{{ tm("chips.executionMode", { name: recordSubtitle(record) }) }}</v-chip>
               </div>
               <div v-else-if="recordSubtitle(record)" class="timeline-subtitle">{{ recordSubtitle(record) }}</div>
-              <details v-if="formatFields(record.fields)" class="timeline-fields"><summary>{{ tm("details") }}</summary><pre>{{ formatFields(record.fields) }}</pre></details>
+              <details v-if="formatFields(record.model || record.fields)" class="timeline-fields"><summary>{{ tm("details") }}</summary><pre>{{ formatFields(record.model || record.fields) }}</pre></details>
               </template>
             </div>
           </div>
@@ -815,9 +990,10 @@ onBeforeUnmount(() => {
 @keyframes trace-connecting-spin { to { transform: rotate(360deg); } }
 @media (prefers-reduced-motion: reduce) { .trace-connecting-icon { animation: none; } }
 .trace-explorer-body { display: flex; flex: 1; min-height: 0; min-width: 0; overflow: hidden; position: relative; }
-.trace-list { flex: 1; min-height: 0; min-width: 0; overflow: hidden; }
+.trace-list { display: flex; flex-direction: column; flex: 1; min-height: 0; min-width: 0; overflow: hidden; }
 .has-detail .trace-list { border-right: 1px solid var(--trace-border); }
-.trace-list-table { background: transparent; border-radius: 0; height: 100%; }
+.trace-list-table { background: transparent; border-radius: 0; flex: 1; min-height: 0; height: 100%; }
+.trace-page-footer { align-items: center; color: var(--trace-muted); display: flex; justify-content: center; padding: 8px; font-size: 12px; }
 .trace-list-table :deep(.trace-list-row.selected) { background: rgba(var(--v-theme-primary), 0.1); }
 .trace-list-table :deep(.trace-list-row:focus-visible) { outline: 2px solid rgb(var(--v-theme-primary)); outline-offset: -2px; }
 .trace-cell-text { display: block; max-width: 320px; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }

dashboard/src/i18n/locales/en-US/features/config-metadata.json

@@ -1252,17 +1252,17 @@
         "description": "Temp Directory Size Limit (MB)",
         "hint": "Limits total size of data/temp in MB. The system checks every 10 minutes, and when exceeded, deletes oldest files first to release about 30% of current size."
       },
-      "trace_log_enable": {
-        "description": "Enable Trace File Logging",
-        "hint": "Write trace events to a separate file (does not change console output)."
+      "trace_db_path": {
+        "description": "Trace Database Path",
+        "hint": "Dedicated SQLite relative to data, default logs/traces.db. Restart after changing the path."
       },
-      "trace_log_path": {
-        "description": "Trace Log File Path",
-        "hint": "Relative paths are resolved under the data directory, e.g. logs/astrbot.trace.log; absolute paths are supported."
+      "trace_db_max_mb": {
+        "description": "Trace Database Capacity Budget (MB)",
+        "hint": "Positive integer, default 80 MB. Evict old complete traces. WAL and checkpoints may temporarily add disk usage; this is not an instantaneous hard limit."
       },
-      "trace_log_max_mb": {
-        "description": "Trace Log Max Size (MB)",
-        "hint": "Rotate when exceeding this size; default 20MB."
+      "trace_retention_days": {
+        "description": "Trace Retention Days",
+        "hint": "Positive integer, default 7 days. Cleanup when age or capacity is reached."
       },
       "pip_install_arg": {
         "description": "Additional pip Installation Arguments",

dashboard/src/i18n/locales/en-US/features/trace.json

@@ -34,6 +34,14 @@
   "details": "View details",
   "selectTrace": "Select a trace to inspect the complete execution chain",
   "closeDetails": "Close details",
+  "history": {
+    "loadFailed": "Failed to load trace history. Please retry.",
+    "detailFailed": "Failed to load trace details; the trace may have been evicted. Please retry.",
+    "storageWarning": "Some traces could not be persisted. Check database errors or capacity limits.",
+    "loadMore": "Load more",
+    "limitHint": "Showing up to {count} traces. Refine your search.",
+    "truncated": "This trace exceeded content or node limits; some data was truncated."
+  },
   "controls": {
     "start": "Start",
     "pauseButton": "Pause",
@@ -49,7 +57,7 @@
     "timestamp": "Timestamp",
     "message": "Message",
     "sender": "Sender",
-    "spansHint": "Timeline nodes, with start/end paired; collapsing groups does not reduce the count."
+    "spansHint": "Actual spans including the message root, excluding events. Collapsing does not change the count."
   },
   "chips": {
     "executionMode": "Execution mode: {name}",

dashboard/src/i18n/locales/ja-JP/features/config-metadata.json

@@ -1353,17 +1353,17 @@
         "description": "一時ディレクトリの最大サイズ（MB）",
         "hint": "data/temp ディレクトリの合計サイズを MB 単位で制限します。システムは 10 分ごとに確認し、上限を超えると更新日時の古いファイルから削除して、現在のサイズの約 30% を解放します。"
       },
-      "trace_log_enable": {
-        "description": "Trace ファイルログを有効化",
-        "hint": "Trace イベントを個別のファイルへ書き込みます（コンソール出力には影響しません）。"
+      "trace_db_path": {
+        "description": "トレースDBのパス",
+        "hint": "data を基準とする専用 SQLite。既定は logs/traces.db。変更後は再起動が必要です。"
       },
-      "trace_log_path": {
-        "description": "Trace ログファイルのパス",
-        "hint": "相対パスは data ディレクトリを基準とします（例: logs/astrbot.trace.log）。絶対パスにも対応しています。"
+      "trace_db_max_mb": {
+        "description": "トレースDB容量 (MB)",
+        "hint": "正の整数。既定は 80 MB。古いトレース単位で削除。WAL とチェックポイントにより一時的に増えるため、厳密な瞬間上限ではありません。"
       },
-      "trace_log_max_mb": {
-        "description": "Trace ログの最大サイズ（MB）",
-        "hint": "サイズを超えると自動的にローテーションします。デフォルトは 20MB です。"
+      "trace_retention_days": {
+        "description": "トレース保持日数",
+        "hint": "正の整数。既定は 7 日。期限または容量で削除します。"
       },
       "pip_install_arg": {
         "description": "pip インストールの追加引数",

dashboard/src/i18n/locales/ja-JP/features/trace.json

@@ -34,6 +34,14 @@
   "details": "詳細を表示",
   "selectTrace": "トレースを選択すると実行チェーンを確認できます",
   "closeDetails": "詳細を閉じる",
+  "history": {
+    "loadFailed": "トレース履歴を読み込めません。再試行してください。",
+    "detailFailed": "詳細を読み込めません。削除済みの可能性があります。再試行してください。",
+    "storageWarning": "一部のトレースを保存できませんでした。DB エラーまたは容量を確認してください。",
+    "loadMore": "もっと読み込む",
+    "limitHint": "最大 {count} 件表示。検索条件を絞ってください。",
+    "truncated": "内容またはノード数が上限を超え、一部が省略されました。"
+  },
   "controls": {
     "start": "開始",
     "pauseButton": "一時停止",
@@ -49,7 +57,7 @@
     "timestamp": "タイムスタンプ",
     "message": "メッセージ",
     "sender": "送信者",
-    "spansHint": "開始と終了をまとめたタイムラインのノード数。グループを折りたたんでも件数は変わりません。"
+    "spansHint": "メッセージのルートを含む実際の Span 数。イベントは含まず、折りたたんでも変わりません。"
   },
   "chips": {
     "executionMode": "実行方式: {name}",

dashboard/src/i18n/locales/ru-RU/features/config-metadata.json

@@ -1357,17 +1357,17 @@
                 "description": "Лимит размера временной директории (МБ)",
                 "hint": "Лимит временной папки (МБ). Система проверяет каждые 10 минут и удаляет старое при переполнении."
             },
-            "trace_log_enable": {
-                "description": "Включить логирование трассировки",
-                "hint": "Записывать трассировку в отдельный файл."
+            "trace_db_path": {
+              "description": "Путь базы трассировок",
+              "hint": "SQLite относительно data, по умолчанию logs/traces.db. После изменения нужен перезапуск."
             },
-            "trace_log_path": {
-                "description": "Путь к файлу логов трассировки",
-                "hint": "Относительные пути определяются относительно директории данных, например, logs/astrbot.trace.log; абсолютные пути также поддерживаются."
+            "trace_db_max_mb": {
+              "description": "Бюджет базы трассировок (МБ)",
+              "hint": "Положительное целое число, по умолчанию 80 МБ. Старые трассы удаляются целиком. WAL и контрольные точки могут временно увеличить размер; это не жёсткий мгновенный предел."
             },
-            "trace_log_max_mb": {
-                "description": "Макс. размер файла логов трассировки (МБ)",
-                "hint": "Ротация при достижении размера. По умолчанию 20МБ."
+            "trace_retention_days": {
+              "description": "Хранение трассировок (дни)",
+              "hint": "Положительное целое число, по умолчанию 7 дней. Очистка по сроку или объёму."
             },
             "pip_install_arg": {
                 "description": "Дополнительные аргументы установки pip",
@@ -2096,4 +2096,4 @@
         "helpMiddle": "или",
         "helpSuffix": "."
     }
-}
+}

dashboard/src/i18n/locales/ru-RU/features/trace.json

@@ -34,6 +34,14 @@
   "details": "Подробности",
   "selectTrace": "Выберите трассировку, чтобы увидеть цепочку выполнения",
   "closeDetails": "Закрыть подробности",
+  "history": {
+    "loadFailed": "Не удалось загрузить историю трасс. Повторите попытку.",
+    "detailFailed": "Не удалось загрузить детали; трасса могла быть удалена. Повторите попытку.",
+    "storageWarning": "Часть трасс не сохранена. Проверьте ошибки БД и лимиты.",
+    "loadMore": "Загрузить ещё",
+    "limitHint": "Показано до {count} трасс. Уточните поиск.",
+    "truncated": "Превышен лимит содержимого или узлов; часть данных усечена."
+  },
   "controls": {
     "start": "Запустить",
     "pauseButton": "Пауза",
@@ -49,7 +57,7 @@
     "timestamp": "Время",
     "message": "Сообщение",
     "sender": "Отправитель",
-    "spansHint": "Число узлов временной шкалы с объединёнными началом и концом. Сворачивание групп не уменьшает число."
+    "spansHint": "Число Span, включая корневой Span сообщения, без событий. Сворачивание не меняет число."
   },
   "chips": {
     "executionMode": "Режим выполнения: {name}",

dashboard/src/i18n/locales/zh-CN/features/config-metadata.json

@@ -1273,17 +1273,17 @@
         "description": "临时目录大小上限 (MB)",
         "hint": "用于限制 data/temp 目录总大小，单位为 MB。系统每 10 分钟检查一次，超限时按文件修改时间从旧到新删除，释放约 30% 当前体积。"
       },
-      "trace_log_enable": {
-        "description": "启用 Trace 文件日志",
-        "hint": "将 Trace 事件写入独立文件（不影响控制台输出）。"
+      "trace_db_path": {
+        "description": "追踪数据库路径",
+        "hint": "独立 SQLite，相对 data 目录，默认 logs/traces.db；更改后重启生效。"
       },
-      "trace_log_path": {
-        "description": "Trace 日志文件路径",
-        "hint": "相对路径以 data 目录为基准，例如 logs/astrbot.trace.log；支持绝对路径。"
+      "trace_db_max_mb": {
+        "description": "追踪数据库容量预算 (MB)",
+        "hint": "正整数，默认 80 MB；按完整追踪淘汰旧数据。WAL 与检查点可能暂时增加磁盘占用，此值不是瞬时硬上限。"
       },
-      "trace_log_max_mb": {
-        "description": "Trace 日志大小上限 (MB)",
-        "hint": "超过大小后自动轮转，默认 20MB。"
+      "trace_retention_days": {
+        "description": "追踪保留天数",
+        "hint": "正整数，默认 7 天；期限或容量先达到者触发清理。"
       },
       "pip_install_arg": {
         "description": "pip 安装额外参数",

dashboard/src/i18n/locales/zh-CN/features/trace.json

@@ -34,6 +34,14 @@
   "details": "查看详情",
   "selectTrace": "选择一条追踪查看完整执行链",
   "closeDetails": "关闭详情",
+  "history": {
+    "loadFailed": "读取追踪历史失败，请重试",
+    "detailFailed": "读取追踪详情失败；记录可能已被清理，请重试",
+    "storageWarning": "部分追踪未能持久化，请检查数据库写入错误或容量限制。",
+    "loadMore": "加载更多",
+    "limitHint": "当前最多展示 {count} 条，请缩小搜索范围。",
+    "truncated": "此追踪内容或节点数量超过限制，部分数据已截断。"
+  },
   "controls": {
     "start": "启动",
     "pauseButton": "暂停",
@@ -49,7 +57,7 @@
     "timestamp": "时间戳",
     "message": "消息",
     "sender": "发送人",
-    "spansHint": "时间线节点数；开始/结束合并，同类型折叠不减少计数。"
+    "spansHint": "实际 Span 数（含消息根 Span），不包含事件；折叠不影响计数。"
   },
   "chips": {
     "executionMode": "执行方式：{name}",

dashboard/tests/traceList.test.mjs

@@ -12,15 +12,26 @@ const ast = ts.createSourceFile("trace.js", descriptor.scriptSetup.content, ts.S
 
 function setup(t) {
   const focused = [];
+  const historyRequests = [];
+  const detailRequests = [];
   const context = vm.createContext({
     computed, nextTick, reactive, ref, watch, tm: (key) => key,
     getPlatformIcon: (name) => name + ".png",
+    console: { error() {} },
+    traceApi: {
+      history: (options) => new Promise((resolve, reject) => historyRequests.push({ options, resolve, reject })),
+      detail: (id) => new Promise((resolve, reject) => detailRequests.push({ id, resolve, reject })),
+    },
+    pluginApi: { list: async () => ({ data: { data: [] } }) },
+    sessionApi: { activeUmos: async () => ({ data: { data: { umo_infos: [] } } }) },
     document: { querySelector: (selector) => ({ scrollIntoView() {}, focus() { focused.push(selector); } }) },
   });
   const names = new Set([
     "events", "selectedSpanId", "searchText", "selectedFilters", "umoNames", "clockNow",
     "eventIndex", "maxItems", "selectedEvent", "normalizedSearch", "filteredEvents", "traceRows", "traceHeaders",
-    "processNewTraces", "matchesFilter", "recordSearchText", "timelineRecords", "formatDuration", "selectEvent",
+    "processNewTraces", "normalizeTracePayload", "matchesFilter", "recordSearchText", "timelineRecords", "formatDuration", "selectEvent",
+    "fetchTraceHistory", "loadTraceDetail", "historyBusy", "detailBusy", "historyError", "detailError",
+    "totalRecords", "nextCursor", "storageWarning", "metadataLoaded", "historyRequest", "detailRequest", "plugins", "loading",
   ]);
   const statements = ast.statements.filter((node) =>
     ts.isFunctionDeclaration(node) ? names.has(node.name?.text) :
@@ -30,13 +41,95 @@ function setup(t) {
   vm.runInContext(statements.map((node) => node.getText(ast)).join("\n"), context);
   const watcher = ast.statements.find((node) => node.getText(ast).startsWith("watch(filteredEvents,"));
   t.after(vm.runInContext(watcher.getText(ast), context));
-  return { focused, ...vm.runInContext("({ events, selectedSpanId, selectedEvent, selectedFilters, traceRows, traceHeaders, clockNow, processNewTraces, selectEvent })", context) };
+  return { focused, historyRequests, detailRequests, ...vm.runInContext("({ events, selectedSpanId, selectedEvent, selectedFilters, traceRows, traceHeaders, clockNow, processNewTraces, selectEvent, searchText, fetchTraceHistory, loadTraceDetail, totalRecords, nextCursor, historyBusy, detailBusy, historyError, detailError })", context) };
 }
 
 function record(span, action, time, fields = {}) {
   return { span_id: span, record_id: span + ":" + action + ":" + time, action, time, fields };
 }
 
+function canonical(id, stamp = "2000000000", completed = true) {
+  return {
+    trace_id: id, root_span_id: id.slice(0, 16), name: "message",
+    start_time_unix_nano: "1000000000", updated_time_unix_nano: stamp,
+    end_time_unix_nano: completed ? stamp : null,
+    state: completed ? "completed" : "running", has_errors: false,
+    has_models: true, has_plugins: true, span_count: 3,
+    attributes: { "astrbot.message_outline": "Hello", "astrbot.plugins": "Weather" },
+  };
+}
+
+test("canonical live updates keep summaries only and count real spans, not events", async (t) => {
+  const state = setup(t);
+  const trace = canonical("a".repeat(32));
+  state.processNewTraces([{ trace, event: { event_id: "reply", name: "response_sent", time_unix_nano: "2000000000", attributes: { "astrbot.message": "Reply" } } }]);
+  assert.equal(state.events.value[0].records.length, 0);
+  assert.equal(state.traceRows.value[0].spans, 3);
+  assert.equal(state.events.value[0].completed, true);
+  state.processNewTraces([{ trace: canonical(trace.trace_id, "1500000000", false) }]);
+  assert.equal(state.events.value[0].completed, true);
+  state.selectEvent(trace.trace_id);
+  state.detailRequests[0].resolve({ data: { status: "ok", data: {
+    trace, spans: [{
+      trace_id: trace.trace_id, span_id: trace.root_span_id, parent_span_id: null,
+      start_time_unix_nano: "1000000000", end_time_unix_nano: "2000000000",
+      status: { code: 1 }, attributes: {},
+      events: [{ event_id: "reply", name: "response_sent", time_unix_nano: "2000000000", attributes: { "astrbot.message": "Reply" } }],
+    }],
+  } } });
+  await new Promise((resolve) => setImmediate(resolve));
+  assert.equal(state.events.value[0].records.length, 3);
+  assert.equal(state.traceRows.value[0].spans, 3);
+  state.selectEvent(null);
+  assert.equal(state.events.value[0].records.length, 0);
+});
+
+test("history uses keyset pages and ignores a superseded search response", async (t) => {
+  const state = setup(t);
+  const first = state.fetchTraceHistory();
+  await new Promise((resolve) => setImmediate(resolve));
+  state.historyRequests[0].resolve({ data: { status: "ok", data: {
+    traces: [canonical("a".repeat(32))], total: 51, cursor: "next", storage: {},
+  } } });
+  await first;
+  assert.equal(state.totalRecords.value, 51);
+  const more = state.fetchTraceHistory(false);
+  assert.equal(state.historyRequests[1].options.cursor, "next");
+  state.historyRequests[1].resolve({ data: { status: "ok", data: {
+    traces: [canonical("b".repeat(32))], total: 51, cursor: null, storage: {},
+  } } });
+  await more;
+  assert.equal(state.events.value.length, 2);
+  const oldRequest = state.fetchTraceHistory();
+  state.searchText.value = "New";
+  const newRequest = state.fetchTraceHistory();
+  state.historyRequests[3].resolve({ data: { status: "ok", data: { traces: [], total: 0, cursor: null } } });
+  await newRequest;
+  state.historyRequests[2].resolve({ data: { status: "ok", data: { traces: [canonical("c".repeat(32))], total: 500 } } });
+  await oldRequest;
+  assert.equal(state.totalRecords.value, 0);
+  assert.equal(state.events.value.length, 0);
+  assert.equal(state.historyBusy.value, false);
+});
+
+test("late detail responses cannot reopen a closed inspector or another trace", async (t) => {
+  const state = setup(t);
+  const a = canonical("a".repeat(32));
+  const b = canonical("b".repeat(32));
+  state.processNewTraces([{ trace: a }, { trace: b }]);
+  state.selectEvent(a.trace_id);
+  state.selectEvent(b.trace_id);
+  state.detailRequests[1].resolve({ data: { status: "ok", data: { trace: b, spans: [] } } });
+  await new Promise((resolve) => setImmediate(resolve));
+  state.detailRequests[0].resolve({ data: { status: "ok", data: { trace: a, spans: [] } } });
+  await new Promise((resolve) => setImmediate(resolve));
+  assert.equal(state.selectedEvent.value.span_id, b.trace_id);
+  assert.equal(state.detailBusy.value, false);
+  state.selectEvent(null);
+  assert.equal(state.selectedEvent.value, null);
+  assert.equal(state.events.value.find((event) => event.span_id === a.trace_id).records.length, 0);
+});
+
 test("history and new traces leave the full-width list unselected", async (t) => {
   const state = setup(t);
   state.processNewTraces([record("first", "event_received", 1, { platform_id: "qq-1", platform_name: "qq_official", message: "Hello" })]);

dashboard/tests/traceRecordingControls.test.mjs

@@ -117,7 +117,7 @@ test("toolbar has matching icon-and-text buttons, fluid input, and an embedded r
   assert.match(buttons[2], /tm\("refresh"\)/);
   assert.match(buttons[1], /:loading="recordingBusy"/);
   assert.match(buttons[1], /:disabled="recordingBusy"/);
-  assert.match(toolbar, /tm\('search', \{ count: filteredEvents.length \}\)/);
+  assert.match(toolbar, /tm\('search', \{ count: totalRecords \}\)/);
   const searchStyle = view.styles[0].content.match(/\.trace-search\s*\{([^}]+)\}/)[1];
   assert.match(searchStyle, /flex:\s*1/);
   assert.match(searchStyle, /height:\s*36px/);

docs/en/dev/astrbot-config.md

@@ -556,6 +556,27 @@ Log level. Default is `INFO`. Can be set to `DEBUG`, `INFO`, `WARNING`, `ERROR`,
 
 Whether to enable trace recording. Default is `false`. When enabled, AstrBot records execution traces, which can be viewed under `Data & Logs → Trace` in the admin panel.
 
+Enabling recording also persists traces. Pausing preserves history and lets existing traces finish. Logs correlate with the active `trace_id` / `span_id`; plugins continue using `astrbot.api.logger`. Core fields align with the OTel model, without an OTel SDK, OTLP, or external export.
+
+### `trace_db_path`
+
+Dedicated SQLite path, default `logs/traces.db`, relative to `data`. Restart after changing the path. Log cleanup protects the database, WAL, and SHM. Trace content may contain messages, plugin arguments, and tool results; restrict file access accordingly.
+
+### `trace_db_max_mb` / `trace_retention_days`
+
+Positive integers, default 80 MB capacity target and 7-day retention; unlimited capacity is not supported. Expired or oldest traces are evicted as complete units. WAL, checkpoints, and page reclamation may temporarily add disk usage, so this is not an instantaneous hard limit. Per-trace content, span, and event limits also apply; detail shows truncation warnings.
+
+Legacy settings migrate on load, with new settings taking precedence. Existing JSONL files remain untouched and are neither appended nor automatically imported:
+
+| Old setting | New setting/behavior |
+| --- | --- |
+| `trace_enable` | Unchanged; controls recording and persistence of new traces |
+| `trace_log_enable` | Removed; no separate JSONL toggle |
+| `trace_log_path` | Removed; replaced by `trace_db_path: logs/traces.db` by default |
+| `trace_log_max_mb` | When new capacity is absent, a positive integer is multiplied by 4 for `trace_db_max_mb`, approximating the old current file plus 3 backups; invalid values fall back to 80 MB |
+
+The new `trace_retention_days` defaults to 7 days.
+
 ### `pip_install_arg`
 
 Arguments for `pip install`. E.g., `-i https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple`.

docs/en/use/webui.md

@@ -174,12 +174,12 @@ The toolbar contains search, fullscreen, font-size controls, and `Install Pip pa
 
 The trace tab has no outer gray card; its toolbar and list sit directly on the page with only the necessary separators. The former hint and switch are replaced by a toolbar button showing Pause (enabled and connected), Connecting (enabled but disconnected), or Start (disabled). Filters, recording, and refresh use matching light-gray icon-and-text buttons sharing the input's background color, all 36px high. Search fills the remaining width, pushing actions to the right. Its placeholder includes the message count instead of a separate counter. Click to pause or resume. Updates show loading and prevent repeated clicks; failures preserve the previous state and display an error. Pausing retains the list and open details in both themes.
 
-The `Trace` tab shows message execution chains: receipt, plugin handlers, Agent requests, responses, and completion. It opens as a full-width Vuetify datatable with columns ordered as status chip, platform logo, timestamp, duration, message, UMO, sender, and Spans. Headers, row height, and spacing use Vuetify defaults; only cell content such as status, platform icons, and messages is customized. The platform column has a minimum width and a non-wrapping header. Duration updates while execution is active. Click a row to open its timeline details on the right; use the close button or Escape to return to the list. History loading, incoming records, and filtering never automatically open another trace. Scroll horizontally on narrow screens. Spans counts timeline nodes after pairing start/end records; collapsing groups does not reduce it. The search placeholder counts message traces only. Open `Filters` next to search to select `Has errors`, `Has model calls`, and `Has plugin calls`. Multiple selections require all conditions and can be combined with search. No selection shows all traces; `Reset` clears selections, and closing the sidebar preserves them. This sidebar replaces the former `All / Errors / Agent / Plugins` single-select tabs. Use the toolbar recording button to enable or disable trace recording.
+The `Trace` tab shows message execution chains: receipt, plugin handlers, Agent requests, responses, and completion. It opens as a full-width Vuetify datatable with columns ordered as status chip, platform logo, timestamp, duration, message, UMO, sender, and Spans. Headers, row height, and spacing use Vuetify defaults; only cell content such as status, platform icons, and messages is customized. The platform column has a minimum width and a non-wrapping header. Duration updates while execution is active. Click a row to open its timeline details on the right; use the close button or Escape to return to the list. History loading, incoming records, and filtering never automatically open another trace. Scroll horizontally on narrow screens. Spans counts actual spans including the message root, excluding span events; collapsing does not change the count. The search placeholder shows the total message traces matching the current filters. History is searched and filtered server-side in pages of 50; click Load more to fetch another page. The browser retains at most 300 summaries and loads complete detail only for the selected trace, releasing it on close. Open `Filters` next to search to select `Has errors`, `Has model calls`, and `Has plugin calls`. Multiple selections require all conditions and can be combined with search. No selection shows all traces; `Reset` clears selections, and closing the sidebar preserves them. This sidebar replaces the former `All / Errors / Agent / Plugins` single-select tabs. Use the toolbar recording button to enable or disable trace recording.
 
 Session aliases replace raw UMOs when available; hover over the dotted underline to see the full UMO. Plugins use their display name when available; hover to see the plugin ID. Handler names, event types, and duration are also shown. Plugin handler nodes use the plugin logo when available and the plugin card's default blue star icon if it is missing or fails to load. Titles show the category in parentheses, matching the plugin detail page: command/command group, event listener, event hook, or LLM tool.
 
 > [!NOTE]
-> Traces are kept in memory and are cleared on restart or cache eviction. Instead of expanding a message row, click a message in the list to open its timeline on the right.
+> Traces are persisted in dedicated SQLite at `data/logs/traces.db` by default, with 7-day retention and an 80 MB capacity target. Old traces are evicted as complete units. WAL and checkpoints may temporarily add disk usage; the target is not an instantaneous hard limit. Pausing leaves history readable and allows existing traces to finish. Traces unfinished at restart are marked errored/interrupted. Queue pressure or storage failures show a warning without blocking message processing. Legacy JSONL files are neither appended nor imported; ordinary logs keep their existing file settings.
 
 Handler names also have a dotted underline. Hover to lazily fetch plugin details and show that handler's description, including nested commands, using the same description as the plugin detail page. Details are cached for the current page; hover again to retry a failed request.
 

docs/zh/dev/astrbot-config.md

@@ -556,6 +556,27 @@ AstrBot WebUI 配置。
 
 是否启用追踪记录。默认为 `false`。启用后，AstrBot 会记录运行追踪信息，可以在管理面板的 `数据与日志 → 追踪` 页面查看。
 
+开启后自动持久化；暂停不删除历史，已开始的追踪仍会记录完成状态。日志会关联当前活动的 `trace_id` / `span_id`，插件仍使用原来的 `astrbot.api.logger`。数据模型对齐 OTel 核心字段，但没有 OTel SDK、OTLP 或外部导出。
+
+### `trace_db_path`
+
+独立 SQLite 路径，默认 `logs/traces.db`（相对 `data` 目录），更改路径后重启生效。数据库、WAL 和 SHM 不受“清理日志”影响。追踪正文可能包含消息、插件参数和工具结果，请限制数据库文件访问权限。
+
+### `trace_db_max_mb` / `trace_retention_days`
+
+正整数，默认容量目标 80 MB、保留 7 天，不支持无限容量。达到容量或过期后按完整 trace 淘汰旧数据；WAL、检查点和回收页面可能暂时增加磁盘占用，因此容量不是瞬时硬上限。单条追踪也有内容、Span 和事件数量限制，截断会在详情中提示。
+
+旧配置在加载时自动迁移，新配置优先，旧 JSONL 文件保留但不再写入或自动导入：
+
+| 旧配置 | 新配置/行为 |
+| --- | --- |
+| `trace_enable` | 保持不变，控制新追踪记录与持久化 |
+| `trace_log_enable` | 移除，不再单独开关 JSONL |
+| `trace_log_path` | 移除，默认改用 `trace_db_path: logs/traces.db` |
+| `trace_log_max_mb` | 未指定新容量时，正整数乘 4 迁移为 `trace_db_max_mb`（接近原当前文件加 3 份备份预算）；无效值改用 80 MB |
+
+新增 `trace_retention_days` 默认为 7 天。
+
 ### `pip_install_arg`
 
 `pip install` 的参数。如 `-i https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple`。

docs/zh/use/webui.md

@@ -174,12 +174,12 @@ ChatUI 支持以下常用能力：
 
 追踪标签页不使用外层灰色卡片，工具栏和列表直接铺在页面上，仅保留必要的分隔线。原来的顶部说明和独立记录开关已移除；工具栏按钮显示“暂停”（启用且已连接）、“连接中”（启用但未连接）或“启动”（停用）。过滤器、记录按钮、刷新均采用图标加文字的浅灰按钮，与搜索框共用背景色，统一为 36px 高；搜索框填满剩余宽度，将操作按钮靠右排列。记录数量并入占位文字：“在 X 条记录中搜索消息、会话、插件”，不单独显示。点击记录按钮可暂停或恢复，更新期间显示加载状态并禁用重复点击；失败时保持原状态并提示错误。暂停不会清除已有列表或关闭详情，明暗主题保持相同布局。
 
-`追踪` 页面实时显示消息执行链：收到消息、触发插件处理器、Agent 相关请求、发送响应和完成。默认只展示全宽 Vuetify datatable 追踪列表，列顺序为状态标签（运行中/已完成/错误）、平台 logo、时间戳、时长、消息、UMO、发送人和 Spans；表头、行高和间距采用 Vuetify 默认样式，仅定制状态、平台图标及消息等单元格内容。平台列设有最小宽度，表头不会换行。时长在执行期间持续更新。点击某行后才打开右侧时间线详情，点击关闭按钮或按 Escape 返回列表；加载历史记录、新记录到达及筛选变化都不会自动打开其他记录的详情。窄屏列表可横向滚动。Spans 统计开始/结束合并后的时间线节点，同类型折叠不减少该数量，搜索占位文字中的记录数仍只统计消息。点击搜索框左侧的“过滤器”，勾选“有错误”“有模型调用”“有插件调用”；多选需同时满足所有条件，并可叠加搜索。不勾选则显示全部，“重置”清除勾选，关闭侧栏仍保留条件。原先的“全部 / 错误 / Agent / 插件”单选标签已改为多选侧栏。工具栏记录按钮可启用或禁用追踪记录。
+`追踪` 页面实时显示消息执行链：收到消息、触发插件处理器、Agent 相关请求、发送响应和完成。默认只展示全宽 Vuetify datatable 追踪列表，列顺序为状态标签（运行中/已完成/错误）、平台 logo、时间戳、时长、消息、UMO、发送人和 Spans；表头、行高和间距采用 Vuetify 默认样式，仅定制状态、平台图标及消息等单元格内容。平台列设有最小宽度，表头不会换行。时长在执行期间持续更新。点击某行后才打开右侧时间线详情，点击关闭按钮或按 Escape 返回列表；加载历史记录、新记录到达及筛选变化都不会自动打开其他记录的详情。窄屏列表可横向滚动。Spans 统计实际 Span 数（包括消息根 Span），不包含挂载在 Span 上的事件，折叠不改变计数；搜索占位文字显示满足当前条件的消息 trace 总数。历史记录在服务端搜索和筛选，每页 50 条，可点击“加载更多”；浏览器最多保留 300 条摘要，只有当前打开的 trace 加载完整详情，关闭后释放详情。点击搜索框左侧的“过滤器”，勾选“有错误”“有模型调用”“有插件调用”；多选需同时满足所有条件，并可叠加搜索。不勾选则显示全部，“重置”清除勾选，关闭侧栏仍保留条件。原先的“全部 / 错误 / Agent / 插件”单选标签已改为多选侧栏。工具栏记录按钮可启用或禁用追踪记录。
 
 UMO 有别名时显示别名，带点划线的文字可悬停查看完整 UMO。插件优先显示 display name，悬停可查看插件 ID，同时展示 Handler 名称、事件类型及耗时。插件处理器节点优先使用插件 logo，没有 logo 或加载失败时使用插件卡片的默认蓝色星形图标；标题括号内展示与插件详情页一致的分类：指令/指令组、事件监听器、事件钩子或 LLM 工具。
 
 > [!NOTE]
-> 追踪数据暂存于内存，重启后清空，也会随缓存淘汰。原先展开消息查看记录的方式，改为点击列表中的消息后打开右侧时间线。
+> 追踪持久化到独立 SQLite：默认 `data/logs/traces.db`，保留 7 天，容量目标 80 MB，按完整 trace 淘汰最旧数据。WAL 和检查点可能暂时增加占用，容量不是瞬时硬上限。暂停后仍可查看历史，已开始的 trace 会完成记录；重启时未完成的 trace 标为错误/中断。写入队列满或存储失败时页面显示警告，不阻塞消息处理。旧 JSONL 不再写入，也不自动导入；普通日志仍沿用原有文件设置。
 
 处理器名称也带有点划线，悬停时才请求插件详情并展示该 Handler 的 description（包括指令组内的子指令），与插件详情页使用同一份说明。详情在当前页面内缓存；加载失败时再次悬停可重试。
 

openspec/openapi-v1.yaml

@@ -5085,6 +5085,52 @@ paths:
         "200":
           $ref: "#/components/responses/Text"
 
+  /api/v1/trace/history:
+    get:
+      tags: [Logs]
+      summary: List persisted trace summaries
+      operationId: listTraces
+      x-astrbot-scope: system
+      parameters:
+        - name: limit
+          in: query
+          schema: {type: integer, minimum: 1, maximum: 100, default: 50}
+        - name: cursor
+          in: query
+          schema: {type: string, maxLength: 64}
+        - name: query
+          in: query
+          schema: {type: string, maxLength: 256, default: ""}
+        - name: errors
+          in: query
+          schema: {type: boolean, default: false}
+        - name: models
+          in: query
+          schema: {type: boolean, default: false}
+        - name: plugins
+          in: query
+          schema: {type: boolean, default: false}
+      responses:
+        "200":
+          $ref: "#/components/responses/Ok"
+
+  /api/v1/trace/{trace_id}:
+    get:
+      tags: [Logs]
+      summary: Get persisted trace spans and events
+      operationId: getTraceDetail
+      x-astrbot-scope: system
+      parameters:
+        - name: trace_id
+          in: path
+          required: true
+          schema: {type: string, pattern: "^[0-9a-f]{32}$"}
+      responses:
+        "200":
+          $ref: "#/components/responses/Ok"
+        "404":
+          description: Trace not found or already evicted.
+
   /api/v1/trace/settings:
     get:
       tags: [Logs]

tests/test_fastapi_v1_dashboard.py

@@ -1,6 +1,7 @@
 import asyncio
 import copy
 import json
+import time
 from dataclasses import dataclass
 from pathlib import Path
 from types import SimpleNamespace
@@ -4312,3 +4313,79 @@ async def test_v1_platform_webhook_preserves_tuple_response(
     assert response.status_code == 202
     assert response.headers["content-type"] == "text/plain"
     assert response.text == "accepted"
+
+
+@pytest.mark.asyncio
+async def test_v1_persistent_trace_routes(asgi_client, monkeypatch, tmp_path):
+    """Verify durable history/detail validation and dashboard authentication."""
+    from astrbot.core.utils.trace_store import TraceStore
+    from astrbot.dashboard.services import log_service
+
+    store = TraceStore(tmp_path / "traces.db", {"trace_db_max_mb": 8})
+    monkeypatch.setattr(log_service, "get_trace_store", lambda *_: store)
+    stamp = str(time.time_ns())
+    trace_id = "a" * 32
+    span_id = "b" * 16
+    summary = {
+        "trace_id": trace_id,
+        "root_span_id": span_id,
+        "start_time_unix_nano": stamp,
+        "end_time_unix_nano": stamp,
+        "updated_time_unix_nano": stamp,
+        "state": "completed",
+        "has_errors": False,
+        "has_models": False,
+        "has_plugins": False,
+        "span_count": 1,
+        "attributes": {"astrbot.message_outline": "Hello"},
+    }
+    try:
+        assert store.append(
+            {
+                "trace": summary,
+                "time_unix_nano": stamp,
+                "span": {
+                    "trace_id": trace_id,
+                    "span_id": span_id,
+                    "parent_span_id": None,
+                    "name": "message",
+                    "start_time_unix_nano": stamp,
+                    "end_time_unix_nano": stamp,
+                    "attributes": {},
+                    "status": {"code": 1},
+                },
+            }
+        )
+        assert store.flush()
+        response = await asgi_client.get("/api/v1/trace/history")
+        assert response.status_code == 401
+        token = jwt.encode(
+            {"username": "astrbot", "exp": int(time.time()) + 600},
+            JWT_SECRET,
+            algorithm="HS256",
+        )
+        headers = {"Authorization": f"Bearer {token}"}
+        response = await asgi_client.get(
+            "/api/v1/trace/history",
+            params={"limit": 1, "query": "Hello"},
+            headers=headers,
+        )
+        assert response.status_code == 200
+        assert response.json()["data"]["total"] == 1
+        response = await asgi_client.get(f"/api/v1/trace/{trace_id}", headers=headers)
+        assert response.status_code == 200
+        assert response.json()["data"]["spans"][0]["span_id"] == span_id
+        assert (
+            await asgi_client.get("/api/v1/trace/history?limit=101", headers=headers)
+        ).status_code == 422
+        assert (
+            await asgi_client.get("/api/v1/trace/invalid", headers=headers)
+        ).status_code == 422
+        assert (
+            await asgi_client.get("/api/v1/trace/" + "c" * 32, headers=headers)
+        ).status_code == 404
+        assert (
+            await asgi_client.get("/api/v1/trace/settings", headers=headers)
+        ).status_code == 200
+    finally:
+        store.close()

tests/unit/test_astr_message_event.py

@@ -157,6 +157,38 @@ async def test_streaming_trace_context_restored_after_failure(
     assert streaming_response_event.get() is None
 
 
+@pytest.mark.asyncio
+async def test_streaming_outline_limit_does_not_truncate_delivery(astr_message_event):
+    """Bound diagnostic content without changing the outgoing stream."""
+    event = astr_message_event
+    event.trace = MagicMock()
+    fragment = "字" * 20000
+    delivered = []
+
+    async def stream():
+        yield MessageChain([Plain(fragment)])
+        yield MessageChain([Plain("End")])
+
+    async def deliver(generator, use_fallback):
+        async for chunk in generator:
+            delivered.append(chunk.chain[0].text)
+
+    event.send_streaming = AsyncMock(side_effect=deliver)
+    event.set_result(
+        MessageEventResult()
+        .set_result_content_type(ResultContentType.STREAMING_RESULT)
+        .set_async_stream(stream())
+    )
+    stage = RespondStage()
+    stage.config = {"provider_settings": {}}
+    await stage.process(event)
+    event.trace.record.assert_called_once()
+    fields = event.trace.record.call_args.kwargs
+    assert len(fields["message"].encode()) <= 4096
+    assert fields["truncated"] is True
+    assert delivered == [fragment, "End"]
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     "result_type",

tests/unit/test_trace_store.py

@@ -0,0 +1,433 @@
+"""Canonical trace model, correlation, bounded storage, and migration tests."""
+
+import asyncio
+import json
+import logging
+import re
+import sqlite3
+import threading
+import time
+
+import pytest
+
+from astrbot.core.log import LogBroker, LogManager
+from astrbot.core.utils import trace as trace_module
+from astrbot.core.utils.migra_helper import migrate_config_on_load
+from astrbot.core.utils.storage_cleaner import StorageCleaner
+from astrbot.core.utils.telemetry import (
+    MAX_ATTRIBUTE_BYTES,
+    bounded_attributes,
+    current_trace_context,
+)
+from astrbot.core.utils.trace import TraceSpan
+from astrbot.core.utils.trace_store import QUEUE_BYTES, TraceStore
+
+
+@pytest.fixture
+def capture(monkeypatch):
+    """Capture immutable snapshots without writing to the default path."""
+    frames = []
+
+    class CaptureStore:
+        def append(self, payload):
+            frames.append(json.loads(json.dumps(payload)))
+            return True
+
+    monkeypatch.setitem(trace_module.astrbot_config, "trace_enable", True)
+    monkeypatch.setattr(trace_module, "get_trace_store", lambda *_: CaptureStore())
+    monkeypatch.setattr(trace_module, "_get_log_broker", lambda: None)
+    return frames
+
+
+def test_standard_ids_parents_and_immutable_snapshots(capture):
+    trace = TraceSpan("message", "qq:Group:123", "Test", "Hello")
+    assert re.fullmatch("[0-9a-f]{32}", trace.trace_id)
+    assert re.fullmatch("[0-9a-f]{16}", trace.span_id)
+    with trace.context():
+        trace.record("event_received", platform_id="qq", kind="message")
+        trace.record(
+            "handler_start",
+            kind="plugin_handler",
+            plugin="test",
+            handler_full_name="test.handler",
+        )
+        handler_id = current_trace_context.get().span_id
+        trace.record("agent_start", kind="agent")
+        agent_id = current_trace_context.get().span_id
+        trace.record("agent_tool_call", tool_name="weather")
+        trace.record("agent_tool_result", tool_result="Sunny")
+        trace.record("agent_end", status="success")
+        assert current_trace_context.get().span_id == handler_id
+        trace.record("handler_end", handler_full_name="test.handler", status="success")
+        assert current_trace_context.get().span_id == trace.span_id
+        trace.record("response_sent", kind="response", source="core", message="Sunny")
+        trace.record("pipeline_end", status="success")
+    assert current_trace_context.get() is None
+    spans = {
+        frame["span"]["span_id"]: frame["span"] for frame in capture if "span" in frame
+    }
+    assert len(spans) == 4
+    assert spans[handler_id]["parent_span_id"] == trace.span_id
+    assert spans[agent_id]["parent_span_id"] == handler_id
+    assert all(span["end_time_unix_nano"] for span in spans.values())
+    assert capture[0]["span"]["end_time_unix_nano"] is None
+    assert capture[0]["trace"]["state"] == "running"
+    assert capture[-1]["trace"]["state"] == "completed"
+    assert not trace._pending
+
+
+@pytest.mark.asyncio
+async def test_logs_correlate_and_detached_contexts_expire(capture):
+    trace = TraceSpan("message")
+    broker = LogBroker()
+    log = logging.Logger("astrbot.trace_test")
+    LogManager.set_queue_handler(log, broker)
+    release = asyncio.Event()
+
+    async def late_log():
+        await release.wait()
+        log.info("Detached")
+
+    with trace.context():
+        trace.record("event_received")
+        log.info("Root")
+        trace.record("handler_start", handler_full_name="plugin.handler")
+        child_id = current_trace_context.get().span_id
+        task = asyncio.create_task(late_log())
+        log.warning("Handler")
+        trace.record(
+            "handler_end", handler_full_name="plugin.handler", status="success"
+        )
+        log.info("Root again")
+        trace.record("pipeline_end", status="success")
+    release.set()
+    await task
+    entries = list(broker.log_cache)
+    assert entries[0]["trace_id"] == trace.trace_id
+    assert entries[0]["span_id"] == trace.span_id
+    assert entries[1]["span_id"] == child_id
+    assert entries[1]["severity_number"] == 13
+    assert entries[1]["body"] == "Handler"
+    assert entries[2]["span_id"] == trace.span_id
+    assert "trace_id" not in entries[-1]
+
+
+@pytest.mark.asyncio
+async def test_concurrent_abandoned_spans_end(capture):
+    trace = TraceSpan("message")
+    release = asyncio.Event()
+
+    async def child(name):
+        trace.record("handler_start", handler_full_name=name)
+        context = current_trace_context.get()
+        await release.wait()
+        assert context.active is False
+
+    with trace.context():
+        trace.record("event_received")
+        tasks = [asyncio.create_task(child(name)) for name in ("a", "b")]
+        await asyncio.sleep(0)
+        trace.record("pipeline_end", status="error", error="Cancelled")
+    release.set()
+    await asyncio.gather(*tasks)
+    assert not trace._pending
+    ended = [
+        frame["span"]
+        for frame in capture
+        if frame.get("span", {}).get("end_time_unix_nano")
+    ]
+    assert len(ended) == 3
+    assert all(span["status"]["code"] == 2 for span in ended)
+
+
+def test_pause_preserves_completion(capture, monkeypatch):
+    trace = TraceSpan("message")
+    with trace.context():
+        trace.record("event_received")
+        trace.record("handler_start", handler_full_name="test.handler")
+        monkeypatch.setitem(trace_module.astrbot_config, "trace_enable", False)
+        trace.record("agent_start")
+        trace.record("handler_end", handler_full_name="test.handler", status="success")
+        trace.record("pipeline_end", status="success")
+    count = len(capture)
+    TraceSpan("disabled").record("event_received")
+    assert len(capture) == count
+    assert capture[-1]["trace"]["state"] == "completed"
+    assert capture[-1]["trace"]["span_count"] == 2
+
+
+def test_limits_preserve_completion(capture):
+    attrs = bounded_attributes(
+        {
+            "astrbot.body": "字" * 50000,
+            "astrbot.object": {"nested": ["x" * 50000] * 100},
+        }
+    )
+    assert len(json.dumps(attrs, ensure_ascii=False).encode()) <= MAX_ATTRIBUTE_BYTES
+    assert attrs["astrbot.truncated"] is True
+    assert all(isinstance(value, (str, bool, int, float)) for value in attrs.values())
+    trace = TraceSpan("message")
+    with trace.context():
+        trace.record("event_received")
+        for index in range(400):
+            trace.record("handler_start", handler_full_name=str(index))
+            trace.record("handler_end", handler_full_name=str(index), status="success")
+        trace.record("pipeline_end", status="success")
+    assert trace._span_count <= 128
+    assert trace._summary["truncated"] is True
+    assert capture[-1]["trace"]["state"] == "completed"
+
+
+@pytest.fixture
+def persisted(tmp_path, capture):
+    """Create an isolated dedicated database."""
+    store = TraceStore(
+        tmp_path / "traces.db", {"trace_db_max_mb": 8, "trace_retention_days": 7}
+    )
+    yield store
+    store.close()
+
+
+def test_database_pages_filters_details_and_restart(persisted, capture):
+    traces = []
+    for index in range(4):
+        trace = TraceSpan("message", "qq:Group:123", "Test", f"Message {index}")
+        with trace.context():
+            trace.record("event_received")
+            trace.record(
+                "handler_start",
+                plugin="test",
+                plugin_display_name="Display Test",
+                handler_full_name="test.handler",
+            )
+            trace.record(
+                "handler_end",
+                handler_full_name="test.handler",
+                status="error" if index == 2 else "success",
+            )
+            trace.record(
+                "response_sent", kind="response", source="plugin:test", message="Reply"
+            )
+            if index != 3:
+                trace.record("pipeline_end", status="success")
+        traces.append(trace)
+    for frame in capture:
+        assert persisted.append(frame)
+    assert persisted.flush()
+    first = persisted.list_traces(limit=2)
+    second = persisted.list_traces(limit=2, cursor=first["cursor"])
+    assert first["total"] == 4
+    assert len(first["traces"]) == len(second["traces"]) == 2
+    assert {item["trace_id"] for item in first["traces"]}.isdisjoint(
+        item["trace_id"] for item in second["traces"]
+    )
+    assert persisted.list_traces(errors=True)["total"] == 1
+    assert persisted.list_traces(plugins=True, query="Display Test")["total"] == 4
+    assert persisted.list_traces(models=True)["total"] == 0
+    detail = persisted.get_trace(traces[0].trace_id)
+    assert len(detail["spans"]) == 2
+    assert any(
+        event["name"] == "response_sent"
+        for span in detail["spans"]
+        for event in span.get("events", [])
+    )
+    assert persisted.get_trace("0" * 32) is None
+    persisted.close()
+    restarted = TraceStore(persisted.path, persisted.config)
+    try:
+        assert restarted.flush()
+        detail = restarted.get_trace(traces[3].trace_id)
+        assert detail["trace"]["state"] == "interrupted"
+        assert detail["trace"]["has_errors"] is True
+        assert all(span["end_time_unix_nano"] for span in detail["spans"])
+    finally:
+        restarted.close()
+
+
+def test_retention_deletes_whole_traces(persisted, capture):
+    trace = TraceSpan("message")
+    trace.record("event_received")
+    trace.record("response_sent", message="Reply")
+    trace.record("pipeline_end", status="success")
+    for frame in capture:
+        assert persisted.append(frame)
+    assert persisted.flush()
+    with sqlite3.connect(persisted.path) as connection:
+        connection.execute("PRAGMA foreign_keys=ON")
+        connection.execute(
+            "UPDATE traces SET updated_ns=?", (time.time_ns() - 8 * 86400 * 10**9,)
+        )
+        persisted._prune(connection)
+        for table in ("traces", "spans", "events"):
+            assert (
+                connection.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] == 0
+            )
+
+
+def test_oversized_queue_records_are_rejected(persisted):
+    assert persisted.append({"oversized": "x" * QUEUE_BYTES}) is False
+    assert persisted.dropped == 1
+    assert persisted._pending_bytes == 0
+
+
+def test_stale_updates_do_not_reopen_completed_trace(persisted, capture):
+    trace = TraceSpan("message", message_outline="Literal % _")
+    trace.record("event_received")
+    trace.record("pipeline_end", status="success")
+    for frame in reversed(capture):
+        assert persisted.append(frame)
+    assert persisted.flush()
+    detail = persisted.get_trace(trace.trace_id)
+    assert detail["trace"]["state"] == "completed"
+    assert detail["spans"][0]["end_time_unix_nano"]
+    assert persisted.list_traces(query="% _")["total"] == 1
+    assert persisted.list_traces(query="other%")["total"] == 0
+
+
+def test_capacity_prunes_complete_traces_with_cascade(persisted, capture):
+    trace = TraceSpan("message")
+    trace.record("event_received")
+    trace.record("pipeline_end", status="success")
+    for index in range(80):
+        frame = json.loads(json.dumps(capture[-1]))
+        frame["trace"]["trace_id"] = frame["span"]["trace_id"] = f"{index:032x}"
+        frame["trace"]["root_span_id"] = frame["span"]["span_id"] = f"{index:016x}"
+        frame["trace"]["attributes"]["astrbot.payload"] = "x" * 20000
+        assert persisted.append(frame)
+        if index % 16 == 0:
+            assert persisted.flush()
+    assert persisted.flush()
+    persisted.config["trace_db_max_mb"] = 1
+    with sqlite3.connect(persisted.path) as connection:
+        connection.execute("PRAGMA foreign_keys=ON")
+        persisted._prune(connection)
+        count = connection.execute("SELECT COUNT(*) FROM traces").fetchone()[0]
+        assert 0 < count < 80
+        assert connection.execute("SELECT COUNT(*) FROM spans").fetchone()[0] == count
+        assert (
+            sum(
+                connection.execute(
+                    f"SELECT COALESCE(SUM(size_bytes), 0) FROM {table}"
+                ).fetchone()[0]
+                for table in ("traces", "spans", "events")
+            )
+            < 1024 * 1024
+        )
+
+
+def test_writer_queue_has_byte_and_count_limits(tmp_path, monkeypatch, capture):
+    release = threading.Event()
+    original_run = TraceStore._run
+
+    def delayed_run(self):
+        """Hold the writer to simulate a slow disk."""
+        release.wait(5)
+        original_run(self)
+
+    monkeypatch.setattr(TraceStore, "_run", delayed_run)
+    trace = TraceSpan("message")
+    trace.record("event_received")
+    payload = capture[0]
+    payload["trace"]["attributes"]["astrbot.payload"] = "x" * 40000
+    store = TraceStore(tmp_path / "traces.db", {})
+    try:
+        accepted = sum(store.append(payload) for _ in range(200))
+        assert accepted < 128
+        assert store._pending_bytes <= QUEUE_BYTES
+        assert store.queue.qsize() == accepted
+        assert store.dropped == 200 - accepted
+        release.set()
+        assert store.flush()
+    finally:
+        release.set()
+        store.close()
+
+
+@pytest.mark.asyncio
+async def test_background_broker_delivery_is_bounded_and_wakes_reader():
+    broker = LogBroker()
+    queue = broker.register()
+    try:
+        waiting = asyncio.create_task(queue.get())
+        await asyncio.to_thread(broker.publish, {"type": "trace", "value": 1})
+        assert (await asyncio.wait_for(waiting, 1))["value"] == 1
+        thread = threading.Thread(
+            target=lambda: [broker.publish({"type": "trace"}) for _ in range(5000)]
+        )
+        thread.start()
+        thread.join()
+        assert broker._pending_callbacks[queue] <= queue.maxsize
+        await asyncio.sleep(0)
+        assert queue.qsize() <= queue.maxsize
+        assert not broker.log_cache
+    finally:
+        broker.unregister(queue)
+
+
+def test_foreign_database_is_not_modified(tmp_path):
+    path = tmp_path / "business.db"
+    with sqlite3.connect(path) as connection:
+        connection.execute("CREATE TABLE important(value TEXT)")
+        connection.execute("INSERT INTO important VALUES ('keep')")
+    store = TraceStore(path, {})
+    try:
+        assert store._ready.wait(5)
+        assert store._failed
+        assert not store.append({})
+        with sqlite3.connect(path) as connection:
+            assert connection.execute("SELECT * FROM important").fetchall() == [
+                ("keep",)
+            ]
+            assert (
+                connection.execute(
+                    "SELECT name FROM sqlite_master WHERE name='trace_meta'"
+                ).fetchone()
+                is None
+            )
+    finally:
+        store.close()
+
+
+@pytest.mark.parametrize(
+    "size,expected", [(20, 80), (50, 200), (0, 80), (-1, 80), ("bad", 80)]
+)
+def test_config_migration(tmp_path, size, expected):
+    config = {
+        "trace_enable": False,
+        "trace_log_enable": True,
+        "trace_log_path": "logs/old.jsonl",
+        "trace_log_max_mb": size,
+    }
+    old = tmp_path / "old.jsonl"
+    old.write_text("keep", encoding="utf-8")
+    assert migrate_config_on_load(config, tmp_path / "cmd_config.json")
+    assert config["trace_enable"] is False
+    assert config["trace_db_path"] == "logs/traces.db"
+    assert config["trace_db_max_mb"] == expected
+    assert not any(key.startswith("trace_log_") for key in config)
+    assert old.read_text() == "keep"
+    config["trace_db_max_mb"] = 123
+    config["trace_log_max_mb"] = 5
+    migrate_config_on_load(config, tmp_path / "cmd_config.json")
+    assert config["trace_db_max_mb"] == 123
+
+
+def test_log_cleanup_protects_databases(tmp_path):
+    logs = tmp_path / "logs"
+    logs.mkdir()
+    for name in ("traces.db", "traces.db-wal", "traces.db-shm", "astrbot.log"):
+        (logs / name).write_bytes(b"keep")
+    with sqlite3.connect(logs / "old-custom-name") as connection:
+        connection.execute("CREATE TABLE important(value TEXT)")
+    (logs / "old-custom-name-wal").write_bytes(b"keep")
+    cleaner = StorageCleaner(
+        {"trace_db_path": "logs/traces.db"},
+        data_dir=tmp_path,
+        temp_dir=tmp_path / "temp",
+    )
+    cleaner.cleanup("logs")
+    for name in ("traces.db", "traces.db-wal", "traces.db-shm"):
+        assert (logs / name).read_bytes() == b"keep"
+    assert not (logs / "astrbot.log").exists()
+    assert (logs / "old-custom-name").exists()
+    assert (logs / "old-custom-name-wal").exists()