@@ -490,6 +490,8 @@ def list_traces(
         errors: bool = False,
         models: bool = False,
         plugins: bool = False,
+        start_time: int | None = None,
+        end_time: int | None = None,
     ) -> dict:
         """Read one numbered or keyset page, never all stored span bodies.
 
@@ -501,6 +503,8 @@ def list_traces(
             errors: Require an error.
             models: Require an Agent call.
             plugins: Require a plugin call.
+            start_time: Inclusive minimum trace start time, in Unix seconds.
+            end_time: Inclusive maximum trace start time, in Unix seconds.
 
         Returns:
             Summaries, filtered count, next cursor, and storage health.
@@ -515,6 +519,12 @@ def list_traces(
         if page < 1 or (cursor and page != 1):
             raise ValueError("Invalid trace page.")
         conditions, values = [], []
+        if start_time is not None:
+            conditions.append("start_ns >= ?")
+            values.append(start_time * 1_000_000_000)
+        if end_time is not None:
+            conditions.append("start_ns <= ?")
+            values.append(end_time * 1_000_000_000)
         for key, enabled in (
             ("has_errors", errors),
             ("has_models", models),
 
@@ -126,7 +126,12 @@ def create_dashboard_asgi_app(
         files=FileService(),
         knowledge_bases=KnowledgeBaseService(core_lifecycle),
         live_chat=LiveChatService(db, core_lifecycle),
-        logs=LogService(log_broker, core_lifecycle.astrbot_config, core=core_lifecycle),
+        logs=LogService(
+            log_broker,
+            core_lifecycle.astrbot_config,
+            core=core_lifecycle,
+            dashboard_static_folder=static_folder,
+        ),
         bots=BotConfigService(core_lifecycle),
         platforms=PlatformService(core_lifecycle),
         providers=ProviderConfigService(core_lifecycle),
 
@@ -87,11 +87,18 @@ async def live_logs(
 @router.get("/logs/export")
 async def export_logs(
     redact: bool = Query(default=True),
+    start_time: int | None = Query(default=None, ge=0),
+    end_time: int | None = Query(default=None, ge=0),
     _username: str = Depends(require_dashboard_user),
     service: LogService = Depends(get_service),
 ):
     try:
-        archive_path = await asyncio.to_thread(service.export_logs, redact=redact)
+        archive_path = await asyncio.to_thread(
+            service.export_logs,
+            redact=redact,
+            start_time=start_time,
+            end_time=end_time,
+        )
     except LogServiceError as exc:
         _raise_log_error(exc)
     return FileResponse(
 
@@ -8,6 +8,7 @@
 import platform
 import re
 import subprocess
+from datetime import datetime
 from pathlib import Path
 from typing import Any, BinaryIO
 from urllib.parse import quote, quote_plus
@@ -17,7 +18,7 @@
 
 REDACTED = "[REDACTED]"
 _SECRET_KEY = re.compile(
-    r"password|passwd|secret|token$|api[_-]?key|access[_-]?key|private[_-]?key|"
+    r"password|passwd|secret|token$|api[_-]?key|access[_-]?key|private[_-]?key|aes[_-]?key|"
     r"authorization|cookie|credential|(?:client|app|signing|encrypt(?:ion)?)[_-]?key|"
     r"recovery[_-]?code|(?:^|\.)key$|^headers$|^env$",
     re.IGNORECASE,
@@ -26,13 +27,14 @@
 _AUTH = re.compile(r"\b(Bearer|Basic|ApiKey)\s+[^\s\"'<>]+", re.IGNORECASE)
 _COOKIE = re.compile(r"(\b(?:set-cookie|cookie)\s*:\s*)[^\r\n]+", re.IGNORECASE)
 _URL_AUTH = re.compile(r"(https?://)[^/\s@]+@", re.IGNORECASE)
-_ASSIGNMENT = re.compile(
-    r"((?:[\w.-]*(?:token|secret|password|passwd|api[_-]?key|access[_-]?key|"
-    r"(?:client|app|signing|encrypt(?:ion)?)[_-]?key|recovery[_-]?code)"
-    r"[\w.-]*|authorization|cookie|\bkey)[\"']?\s*[:=]\s*)"
-    r"(\"[^\"]*\"|'[^']*'|[^\s&,;}]+)",
+_ASSIGNMENT = re.compile(r"(?<![\w.-])([\w.-]+)[\"']?\s*[:=]\s*")
+_ASSIGNMENT_KEY = re.compile(
+    r"token|secret|password|passwd|api[_-]?key|access[_-]?key|aes[_-]?key|"
+    r"(?:client|app|signing|encrypt(?:ion)?)[_-]?key|recovery[_-]?code|"
+    r"(?:authorization|cookie|\bkey)$",
     re.IGNORECASE,
 )
+_ASSIGNMENT_VALUE = re.compile(r"\"[^\"]*\"|'[^']*'|[^\s&,;}]+")
 _TOKEN = re.compile(
     r"\b(?:sk-[A-Za-z0-9_-]{16,}|gh[pousr]_[A-Za-z0-9_]{20,}|"
     r"eyJ[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,}\.[A-Za-z0-9_-]{8,})"
@@ -49,6 +51,10 @@
     r")(?!\w|\.\w)"
 )
 MAX_LOG_LINE_BYTES = 1024 * 1024
+_LOG_TIMESTAMP = re.compile(
+    r"^\[?(\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2}"
+    r"(?:[.,]\d+)?(?:Z|[+-]\d{2}:?\d{2})?)(?:\]|\s|$)"
+)
 
 
 class DiagnosticRedactor:
@@ -57,6 +63,7 @@ class DiagnosticRedactor:
     def __init__(self, *, enabled: bool = True) -> None:
         self.enabled = enabled
         self.secrets: set[str] = set()
+        self._sorted_secrets: list[str] | None = None
         self.public_ips: dict[str, str] = {}
 
     def collect(self, value: Any, sensitive: bool = False) -> None:
@@ -79,6 +86,7 @@ def collect(self, value: Any, sensitive: bool = False) -> None:
             self.secrets.update(
                 (text, quote(text, safe=""), quote_plus(text), json.dumps(text)[1:-1])
             )
+            self._sorted_secrets = None
 
     def redact(self, value: Any, sensitive: bool = False) -> Any:
         """Return a sanitized copy while retaining containers and boolean flags.
@@ -118,15 +126,59 @@ def text(self, value: str) -> str:
         """
         if not self.enabled:
             return value
-        value = _ANSI.sub("", value)
-        value = _PRIVATE_KEY.sub(REDACTED, value)
-        for secret in sorted(self.secrets, key=len, reverse=True):
-            value = value.replace(secret, REDACTED)
-        value = _AUTH.sub(lambda match: f"{match[1]} {REDACTED}", value)
-        value = _COOKIE.sub(lambda match: f"{match[1]}{REDACTED}", value)
-        value = _URL_AUTH.sub(lambda match: f"{match[1]}{REDACTED}@", value)
-        value = _ASSIGNMENT.sub(lambda match: f"{match[1]}{REDACTED}", value)
-        value = _TOKEN.sub(REDACTED, value)
+        if "\x1b[" in value:
+            value = _ANSI.sub("", value)
+        if "PRIVATE KEY-----" in value:
+            value = _PRIVATE_KEY.sub(REDACTED, value)
+        if self.secrets:
+            if self._sorted_secrets is None:
+                self._sorted_secrets = sorted(self.secrets, key=len, reverse=True)
+            for secret in self._sorted_secrets:
+                value = value.replace(secret, REDACTED)
+        # These cheap substring checks avoid running every regex on ordinary
+        # ASCII log lines. Unicode case-insensitive regex matches can include
+        # characters that lower() does not normalize to the ASCII hint.
+        lowered = value.lower()
+        ascii_only = value.isascii()
+        if not ascii_only or any(
+            prefix in lowered for prefix in ("bearer", "basic", "apikey")
+        ):
+            value = _AUTH.sub(lambda match: f"{match[1]} {REDACTED}", value)
+        if not ascii_only or "cookie" in lowered:
+            value = _COOKIE.sub(lambda match: f"{match[1]}{REDACTED}", value)
+        if "://" in value and "@" in value:
+            value = _URL_AUTH.sub(lambda match: f"{match[1]}{REDACTED}@", value)
+        if not ascii_only or any(
+            hint in lowered
+            for hint in (
+                "token",
+                "secret",
+                "password",
+                "passwd",
+                "key",
+                "code",
+                "authorization",
+                "cookie",
+            )
+        ):
+            # Scan assignment prefixes without consuming ordinary values, which
+            # may themselves contain a nested credential assignment.
+            parts = []
+            end = 0
+            for match in _ASSIGNMENT.finditer(value):
+                if match.start() < end or not _ASSIGNMENT_KEY.search(match[1]):
+                    continue
+                assigned = _ASSIGNMENT_VALUE.match(value, match.end())
+                if assigned is not None:
+                    parts.extend((value[end : match.end()], REDACTED))
+                    end = assigned.end()
+            parts.append(value[end:])
+            value = "".join(parts)
+        if any(
+            prefix in value
+            for prefix in ("sk-", "ghp_", "gho_", "ghu_", "ghs_", "ghr_", "eyJ")
+        ):
+            value = _TOKEN.sub(REDACTED, value)
         parts = []
         end = 0
         for match in _IP_LITERAL.finditer(value):
@@ -136,6 +188,12 @@ def text(self, value: str) -> str:
                 literal[1:-1] if literal.startswith("[") else literal.rstrip(".")
             )
             host, separator, scope = candidate.partition("%")
+            if (
+                ":" in host
+                and "::" not in host
+                and host.count(":") != (6 if "." in host else 7)
+            ):
+                continue
             try:
                 address = ipaddress.ip_address(host)
             except ValueError:
@@ -156,18 +214,29 @@ def text(self, value: str) -> str:
         parts.append(value[end:])
         return "".join(parts)
 
-    def copy_log(self, source: BinaryIO, target: BinaryIO, limit: int) -> bool:
+    def copy_log(
+        self,
+        source: BinaryIO,
+        target: BinaryIO,
+        limit: int,
+        *,
+        start_time: int | None = None,
+        end_time: int | None = None,
+    ) -> bool:
         """Stream complete text lines, skipping oversized lines and private keys.
 
         Args:
             source: An already verified regular file, positioned at its start.
             target: Open ZIP entry for redacted text.
             limit: Maximum source bytes to consume, fixed before reading.
+            start_time: Inclusive minimum record time, in Unix seconds.
+            end_time: Inclusive maximum record time, in Unix seconds.
 
         Returns:
             Whether content was omitted to respect line or file limits.
         """
-        if not self.enabled:
+        filtering = start_time is not None or end_time is not None
+        if not self.enabled and not filtering:
             remaining = limit
             while remaining:
                 chunk = source.read(min(remaining, MAX_LOG_LINE_BYTES))
@@ -179,42 +248,96 @@ def copy_log(self, source: BinaryIO, target: BinaryIO, limit: int) -> bool:
         truncated = False
         oversized = False
         private_key = False
+        pending: list[str] = []
+        pending_bytes = 0
         remaining = limit
+        include_record = not filtering
         while remaining:
             line = source.readline(min(remaining, MAX_LOG_LINE_BYTES + 1))
             if not line:
                 break
             remaining -= len(line)
+            if filtering and not oversized and not private_key:
+                # Timestamp-free continuation lines inherit their record's
+                # selection; content before the first dated record is excluded.
+                prefix = line[:160].decode("utf-8", errors="replace")
+                match = _LOG_TIMESTAMP.match(_ANSI.sub("", prefix))
+                timestamp = None
+                dated_record = match is not None
+                try:
+                    if match is not None:
+                        timestamp = datetime.fromisoformat(
+                            match[1].replace("Z", "+00:00").replace(",", ".")
+                        ).timestamp()
+                    elif prefix.startswith("{"):
+                        record = json.loads(line)
+                        if isinstance(record, dict):
+                            dated_record = True
+                            if "time_unix_nano" in record:
+                                timestamp = int(record["time_unix_nano"]) / 1e9
+                            elif isinstance(record.get("time"), (int, float)):
+                                timestamp = record["time"]
+                            elif "start_time_unix_nano" in record:
+                                timestamp = int(record["start_time_unix_nano"]) / 1e9
+                except (ValueError, TypeError, OverflowError, OSError):
+                    pass
+                if dated_record:
+                    include_record = (
+                        timestamp is not None
+                        and (start_time is None or timestamp >= start_time)
+                        and (end_time is None or timestamp <= end_time)
+                    )
             complete = line.endswith(b"\n") or len(line) <= MAX_LOG_LINE_BYTES
             if oversized or not complete:
                 oversized = not line.endswith(b"\n")
-                truncated = True
-                continue
+                if self.enabled:
+                    truncated = True
+                    continue
             # Never emit a fragment cut at the export boundary: it could be a token.
             if not remaining and not line.endswith(b"\n") and source.read(1):
                 truncated = True
                 break
+            if not self.enabled:
+                if include_record:
+                    target.write(line)
+                continue
             text = line.decode("utf-8", errors="replace")
             if "-----BEGIN " in text and "PRIVATE KEY-----" in text:
-                private_key = True
-                target.write((REDACTED + "\n").encode())
-            if private_key:
+                private_key = not ("-----END " in text and "PRIVATE KEY-----" in text)
+                text = REDACTED + "\n"
+            elif private_key:
                 if "-----END " in text and "PRIVATE KEY-----" in text:
                     private_key = False
                 continue
-            target.write(self.text(text).encode("utf-8"))
+            if not include_record:
+                continue
+            pending.append(text)
+            pending_bytes += len(line)
+            # Keep complete lines together while amortizing regex and ZIP writes
+            # across a bounded batch instead of repeating them for every line.
+            if pending_bytes >= 64 * 1024:
+                target.write(self.text("".join(pending)).encode("utf-8"))
+                pending.clear()
+                pending_bytes = 0
+        if pending:
+            target.write(self.text("".join(pending)).encode("utf-8"))
         return truncated
 
 
 def collect_diagnostics(
-    config: dict, core: Any = None, *, redact: bool = True
+    config: dict,
+    core: Any = None,
+    *,
+    redact: bool = True,
+    webui_version: str | None = None,
 ) -> tuple[dict, DiagnosticRedactor]:
     """Snapshot configuration, plugin status and environment for a log export.
 
     Args:
         config: Running global configuration.
         core: Running lifecycle, when metadata is available.
         redact: Whether to mask credentials throughout the bundle.
+        webui_version: Version declared by the active WebUI assets, if available.
 
     Returns:
         JSON sections and a redactor seeded from all available credentials.
@@ -281,6 +404,7 @@ def collect_diagnostics(
         pass
     sections["environment.json"] = {
         "astrbot_version": __version__,
+        "webui_version": webui_version,
         "commit": commit,
         "python_version": platform.python_version(),
         "os": platform.system(),