feat: 统一日志格式与执行链路追踪

.env.example

@@ -29,7 +29,9 @@ DEEPAGENT_THINKING=off
 # ============================================================
 HOST=0.0.0.0
 PORT=8080
+# INFO：启动、任务及请求汇总；DEBUG：含并发位、健康检查等细节
 LOG_LEVEL=INFO
+# text：本地阅读；json：每行一条结构化日志，适合采集
 LOG_FORMAT=text
 
 # ============================================================

README.md

@@ -130,6 +130,8 @@ DEEP_AGENT_ENV_PRIORITY=process LOG_LEVEL=DEBUG ~/vent-agent/current/release/ser
 
 **目前没有自动的数据迁移、兼容性检查或备份。** 回退只切换代码和依赖，不回滚数据库或 Agent checkpoint；结构不兼容时不能直接切回旧版。完整命令见 [升级与回退](release/README.md#升级与回退)。
 
+日志默认使用 `INFO` + `text`；设置 `LOG_FORMAT=json` 可输出结构化日志。请求与任务通过 `trace_id` 关联，任务还带 `session_id`、`run_id`、状态和耗时。健康检查、静态资源成功请求及并发位细节在 `DEBUG` 查看。配置与排查示例见 [日志说明](docs/DEVELOPMENT.md#日志与排查)。
+
 ## 7. 业务扩展与开发验证
 
 聊天、定时任务和批处理共用 `runtime/` 执行层；HTTP 层负责 SSE 编码。独立脚本使用 `RuntimeEnvironment` 自动初始化和清理资源，默认不启动调度器。调用方式见 [执行层与入口](docs/DEVELOPMENT.md#执行层与入口)。

agents/agent_registry.py

@@ -50,7 +50,7 @@ class _AgentSlot:
         with self._lock:
             # 二次检查：可能在等锁期间已被别的线程编译好
             if self._agent is None:
-                logger.info("首次编译 agent（generation=%d）", self._generation + 1)
+                logger.debug("首次编译 agent（generation=%d）", self._generation + 1)
                 self._agent = self._compile()
                 self._generation += 1
                 logger.info("agent 编译完成（generation=%d）", self._generation)
@@ -66,7 +66,7 @@ class _AgentSlot:
         with self._lock:
             self._agent = None
             self._generation += 1
-            logger.info("agent 已失效，下次请求将重建（generation=%d）", self._generation)
+            logger.debug("agent 已失效，下次请求将重建（generation=%d）", self._generation)
             return self._generation
 
 
@@ -106,5 +106,5 @@ async def prewarm() -> bool:
         await asyncio.to_thread(get_agent)
         return True
     except Exception as exc:
-        logger.error("agent 预热失败，将在首次请求时重试：%s: %s", type(exc).__name__, exc)
+        logger.exception("agent 预热失败，将在首次请求时重试", extra={"kv": {"event": "agent.prewarm_failed"}})
         return False

api/request_logging.py

@@ -0,0 +1,71 @@
+"""ASGI 请求日志：覆盖完整 SSE 生命周期，不读取请求正文或查询参数。"""
+import asyncio
+import logging
+import time
+
+from obs import bind_trace_id, unbind_trace_id, normalize_trace_id, get_logger
+
+logger = get_logger('api.access')
+
+
+class RequestLogMiddleware:
+    def __init__(self, app, *, trace_enabled=True):
+        self.app = app
+        self.trace_enabled = trace_enabled
+
+    async def __call__(self, scope, receive, send):
+        if scope['type'] != 'http':
+            return await self.app(scope, receive, send)
+        started = time.perf_counter()
+        headers = dict(scope.get('headers', []))
+        trace_id = normalize_trace_id(headers.get(b'x-trace-id', b'').decode('latin1'))
+        token = bind_trace_id(trace_id) if self.trace_enabled else None
+        if self.trace_enabled:
+            scope.setdefault('state', {})['trace_id'] = trace_id
+        status, completed, disconnected, failure = None, False, False, None
+
+        async def logged_receive():
+            nonlocal disconnected
+            message = await receive()
+            if message['type'] == 'http.disconnect':
+                disconnected = True
+            return message
+
+        async def logged_send(message):
+            nonlocal status, completed
+            if message['type'] == 'http.response.start':
+                status = message['status']
+                if self.trace_enabled:
+                    message = {**message, 'headers': [(k, v) for k, v in message.get('headers', [])
+                                                       if k.lower() != b'x-trace-id']
+                               + [(b'x-trace-id', trace_id.encode('ascii'))]}
+            await send(message)
+            if message['type'] == 'http.response.body' and not message.get('more_body', False):
+                completed = True
+
+        try:
+            await self.app(scope, logged_receive, logged_send)
+        except BaseException as exc:
+            failure = exc
+            raise
+        finally:
+            try:
+                outcome = ('cancelled' if isinstance(failure, asyncio.CancelledError)
+                           else 'error' if failure else 'completed' if completed
+                           else 'disconnected' if disconnected else 'incomplete')
+                code = status if status is not None else (500 if failure and outcome == 'error' else None)
+                level = logging.ERROR if outcome == 'error' or (code and code >= 500) else (
+                    logging.WARNING if outcome != 'completed' or (code and code >= 400) else logging.INFO)
+                path = scope.get('path', '')
+                if level == logging.INFO and (path == '/health' or path.startswith('/static/') or path == '/favicon.ico'):
+                    level = logging.DEBUG
+                route = scope.get('route')
+                logger.log(level, '请求结束', extra={'kv': {
+                    'event': 'http.request.completed', 'method': scope.get('method'),
+                    'path': getattr(route, 'path', path), 'status_code': code, 'outcome': outcome,
+                    'duration_ms': round((time.perf_counter() - started) * 1000, 1),
+                    **({'error_type': type(failure).__name__} if failure else {}),
+                }})
+            finally:
+                if token is not None:
+                    unbind_trace_id(token)

billing/meter.py

@@ -99,7 +99,7 @@ class UsageMeter:
                 self.model, self.input_tokens, self.output_tokens
             )
         except Exception as exc:
-            logger.error(
+            logger.exception(
                 "计价失败未入账 user=%s model=%s tokens=%d/%d err=%s: %s",
                 self.user_name, self.model, self.input_tokens, self.output_tokens,
                 type(exc).__name__, exc,
@@ -120,7 +120,7 @@ class UsageMeter:
                 usage_id=self._usage_id,
             )
         except Exception as exc:
-            logger.error(
+            logger.exception(
                 "入账失败 user=%s session=%s tokens=%d/%d err=%s: %s",
                 self.user_name, self.session_id, self.input_tokens, self.output_tokens,
                 type(exc).__name__, exc,
@@ -137,6 +137,8 @@ class UsageMeter:
             "入账完成 user=%s kind=%s model=%s tokens=%d/%d cost=%.6f%s source=%s",
             self.user_name, self.kind, self.model,
             self.input_tokens, self.output_tokens, cost, currency, self.usage_source,
+            extra={"kv": {"event": "billing.settled", "usage_id": usage_id,
+                          "session_id": self.session_id, "cost": cost, "currency": currency}},
         )
         return {
             "settled": True,
@@ -160,7 +162,13 @@ def retry_pending() -> int:
             source = payload.pop("source")
             meter = UsageMeter(**payload)
             meter.add(inputs, outputs, source)
-            settled += bool(meter.settle().get("settled"))
+            from obs import bind_trace_id, unbind_trace_id, log_context
+            token = bind_trace_id(meter.trace_id)
+            try:
+                with log_context(run_id=meter._usage_id, session_id=meter.session_id, kind=meter.kind):
+                    settled += bool(meter.settle().get("settled"))
+            finally:
+                unbind_trace_id(token)
         except Exception:
             logger.exception("待结算记录恢复失败 id=%s", row["id"])
     return settled

docs/DEVELOPMENT.md

@@ -230,6 +230,55 @@ if __name__ == "__main__":
 
 ---
 
+## 日志与排查
+
+继续使用 `LOG_LEVEL`（默认 `INFO`）与 `LOG_FORMAT`（`text` / `json`）。业务日志和 Uvicorn 日志使用相同格式、毫秒时间戳与时区；发布脚本也读取配置中的日志级别。输出仍走标准输出，部署脚本写入 `shared/logs/server.log`，systemd 使用 journal。
+
+| 级别 | 内容 |
+|---|---|
+| INFO | 运行环境/扩展服务就绪、执行开始与结束、计费、业务请求汇总 |
+| WARNING | 额度不足、取消、4xx、估算用量等需要留意的情况 |
+| ERROR | 执行或收尾失败、5xx、落库与结算异常；关键异常保留堆栈 |
+| DEBUG | 并发位申请释放、缓存失效、健康检查及静态资源成功请求 |
+
+常见第三方 HTTP 客户端、数据库驱动和调度器内部日志仍保持 WARNING，避免调试业务时输出大量底层细节。
+
+### 按一次执行定位问题
+
+1. 从响应头 `X-Trace-Id` 或 SSE 事件取得 `trace_id`。也可以传入由字母、数字、点、下划线、冒号和连字符组成的 1–128 位 ID；非法值会替换为新 ID。
+2. 按该 ID 搜索日志，再按 `run_id` 区分具体执行、按 `session_id` 关联会话。定时任务还包含 `schedule_id`，其调度记录 ID 为 `schedule_run_id`。
+3. 查看 `execution.completed` 的 `status`、`duration_ms`、输入/输出 token 和 `error_code`，异常详情看同一 trace 下的堆栈。
+
+`http.request.completed` 在响应流和应用清理结束后记录一次，因此 SSE 耗时包含整个连接过程。HTTP 200 只代表流已建立，Agent 是否成功要看 `execution.completed`。健康接口当前可能以 HTTP 200 返回 `status=down`；日志降噪按 HTTP 状态码判断，监控仍需检查响应内容。
+
+独立批处理自动生成 trace；定时执行和后台结算重试也会携带关联 ID。并发请求的上下文互相隔离，传到 `asyncio.to_thread()` 的数据库/计费日志沿用同一上下文。
+
+JSON 日志可这样筛选（服务器装有 `jq` 时）：
+
+```bash
+jq -c 'select(.trace_id == "这里填写trace_id")' ~/vent-agent/shared/logs/server.log
+jq -c 'select(.event == "execution.completed" and .status == "failed")' ~/vent-agent/shared/logs/server.log
+```
+
+以上命令适用于文件内容均为 JSON 的情况；切换格式前留下的文本日志可用 `rg 'trace_id值' 文件路径` 搜索。生产文件的保留与轮转由部署环境管理，本次没有新增第二份应用日志文件。
+
+### 扩展模块如何记日志
+
+```python
+from obs import get_logger
+
+logger = get_logger(__name__)
+logger.info("检索完成", extra={"kv": {
+    "event": "rag.search.completed", "hits": 5, "duration_ms": 38,
+}})
+```
+
+`kv` 在文本和 JSON 中都会输出。`ts`、`level`、`logger`、`trace_id`、`message`、`exc` 是保留字段，不能通过 `kv` 覆盖；当前绑定的执行身份优先。异常处理使用 `logger.exception()` 保留堆栈。
+
+请求日志不记录正文、查询参数、Cookie 或 Authorization；任务摘要不记录提示词、回答和工具参数。框架输出会遮蔽结构化凭证字段，以及消息/堆栈中常见的密钥赋值、Bearer 凭证和 URL 用户密码。扩展仍应只记录排查所需元数据，避免把整段业务数据或客户端配置直接写入日志。
+
+---
+
 ## 扩展方式
 
 ### 新增业务 Agent

extensions/services.py

@@ -4,6 +4,11 @@ from __future__ import annotations
 from contextlib import AsyncExitStack
 from dataclasses import dataclass
 import inspect
+import time
+
+from obs import get_logger
+
+logger = get_logger(__name__)
 from types import MappingProxyType
 from typing import Any, AsyncContextManager, Callable, Mapping
 
@@ -56,6 +61,8 @@ class ServiceContainer:
         order = ordered_names({name: set(spec.requires) for name, spec in self._specs.items()}, "服务")
         for name in order:
             spec = self._specs[name]
+            started = time.perf_counter()
+            logger.debug("扩展服务启动", extra={"kv": {"event": "service.starting", "service": name}})
             dependencies = self.view(spec.requires)
             context = spec.factory(dependencies)
             if not hasattr(context, "__aenter__") or not hasattr(context, "__aexit__"):
@@ -63,6 +70,8 @@ class ServiceContainer:
                     context.close()
                 raise TypeError(f"服务 {name} 的工厂必须返回异步上下文管理器")
             self._values[name] = await self._stack.enter_async_context(context)
+            logger.info("扩展服务就绪", extra={"kv": {"event": "service.ready", "service": name,
+                        "duration_ms": round((time.perf_counter() - started) * 1000, 1)}})
         _active = self
 
     def view(self, names) -> Mapping[str, Any]:

main.py

@@ -11,11 +11,12 @@ from fastapi.staticfiles import StaticFiles
 
 import config
 from api import router as api_router
+from api.request_logging import RequestLogMiddleware
 from runtime import concurrency
 from runtime.environment import RuntimeEnvironment
 from billing import guard
 from db import db_conn
-from obs import bind_trace_id, get_logger, new_trace_id, setup_logging, unbind_trace_id
+from obs import bind_trace_id, get_logger, setup_logging, unbind_trace_id
 
 setup_logging(config.LOG_LEVEL, config.LOG_FORMAT)
 logger = get_logger(__name__)
@@ -45,7 +46,8 @@ async def quota_error(request: Request, exc: guard.QuotaExceededError):
 
 @app.exception_handler(guard.BillingConfigError)
 async def billing_error(request: Request, exc: guard.BillingConfigError):
-    logger.error("计费准入失败：%s", exc)
+    logger.error("计费准入失败", exc_info=(type(exc), exc, exc.__traceback__),
+                 extra={"kv": {"event": "billing.admission_failed"}})
     return JSONResponse(status_code=503, content={"detail": {"code": "billing_unavailable", "message": "计费暂不可用"}})
 
 
@@ -68,25 +70,7 @@ async def session_missing(request: Request, exc: concurrency.SessionMissingError
                         content={"detail": {"code": "session_not_found", "message": str(exc)}})
 
 
-if config.TRACE_MIDDLEWARE:
-
-    @app.middleware("http")
-    async def trace_middleware(request: Request, call_next):
-        """把 ``X-Trace-Id`` 透传进来或新生成一个，回写响应头。
-
-        注意：这里绑定的 ContextVar 在 StreamingResponse 的生成器里**不保证**
-        可见（响应体迭代发生在中间件调用栈之外），所以 SSE 生成器内部会自己
-        再绑一次。这个中间件的作用是让普通 HTTP 接口与访问日志带上 trace_id。
-        """
-        trace_id = request.headers.get("X-Trace-Id") or new_trace_id()
-        request.state.trace_id = trace_id
-        token = bind_trace_id(trace_id)
-        try:
-            response = await call_next(request)
-        finally:
-            unbind_trace_id(token)
-        response.headers["X-Trace-Id"] = trace_id
-        return response
+app.add_middleware(RequestLogMiddleware, trace_enabled=config.TRACE_MIDDLEWARE)
 
 
 @app.exception_handler(Exception)
@@ -96,10 +80,17 @@ async def unhandled_exception_handler(request: Request, exc: Exception):
     不返回堆栈——它可能含连接串、内网地址。堆栈只进日志，用 trace_id 关联。
     """
     trace_id = getattr(request.state, "trace_id", "")
-    logger.exception("未处理异常 path=%s trace=%s", request.url.path, trace_id)
+    token = bind_trace_id(trace_id)
+    try:
+        logger.exception("请求处理异常", extra={"kv": {"event": "http.request.failed",
+                         "method": request.method, "path": request.url.path,
+                         "error_type": type(exc).__name__}})
+    finally:
+        unbind_trace_id(token)
     return JSONResponse(
         status_code=500,
         content={"code": "internal_error", "message": f"服务内部错误：{type(exc).__name__}"},
+        headers={"X-Trace-Id": trace_id} if trace_id else None,
     )
 
 
@@ -177,4 +168,5 @@ if __name__ == "__main__":
         workers=1,  # 单进程是硬约束，见 _check_single_process
         reload=False,  # 开 reload 会起子进程，同样破坏进程内状态
         log_level=config.LOG_LEVEL.lower(),
+        log_config=None, access_log=False,
     )

obs/__init__.py

@@ -3,6 +3,7 @@
 from obs.log_setup import get_logger, setup_logging
 from obs.tracing import (
     bind_trace_id,
+    bind_log_context, unbind_log_context, get_log_context, log_context, normalize_trace_id,
     get_trace_id,
     new_trace_id,
     trace_headers,
@@ -11,6 +12,7 @@ from obs.tracing import (
 
 __all__ = [
     "get_logger",
+    "bind_log_context", "unbind_log_context", "get_log_context", "log_context", "normalize_trace_id",
     "setup_logging",
     "bind_trace_id",
     "get_trace_id",

obs/log_setup.py

@@ -1,85 +1,109 @@
-"""日志装配。
-
-两个格式档位：
-- ``text``（默认）：人类可读，本地开发用 ``tail`` 看
-- ``json``：一行一条 JSON，便于采集到 ELK/Loki
-
-trace_id 通过 ``obs.tracing.get_trace_id`` 动态注入——用 filter 而非
-formatter 参数，是为了让每条日志自动带上当前上下文的 trace_id，调用方
-不需要在每条 log 里手动传。
-"""
-
+"""统一业务与服务器日志：文本便于阅读，JSON 便于采集。"""
 from __future__ import annotations
 
 import json
 import logging
+import re
 import sys
 from datetime import datetime
 
-from obs.tracing import get_trace_id
+from obs.tracing import get_trace_id, get_log_context
 
-_TEXT_FMT = "%(asctime)s %(levelname)-5s [%(trace_id)s] %(name)s | %(message)s"
-_DATE_FMT = "%Y-%m-%d %H:%M:%S"
+_NOISY_LOGGERS = ('httpx', 'httpcore', 'urllib3', 'openai', 'aiosqlite',
+                  'apscheduler.executors.default', 'apscheduler.scheduler',
+                  'langchain_openai.chat_models._client_utils')
+_RESERVED = {'ts', 'level', 'logger', 'trace_id', 'message', 'exc'}
+_SECRET_KEYS = {'authorization', 'proxy_authorization', 'api_key', 'openai_api_key',
+                'password', 'secret', 'token', 'access_token', 'refresh_token', 'cookie', 'set_cookie'}
+_ASSIGNMENT = re.compile(r'(?i)((?:api[_-]?key|password|access[_-]?token|refresh[_-]?token|secret)\s*[=:]\s*[\"\']?)[^\s,;\"\'&]+')
+_BEARER = re.compile(r'(?i)(bearer\s+)[A-Za-z0-9._~+/-]+=*')
+_URL_CREDENTIALS = re.compile(r'(://)[^\s/@]+:[^\s/@]+@')
 
-# 这些第三方 logger 在 INFO 级别过于聒噪，统一压到 WARNING
-_NOISY_LOGGERS = (
-    "httpx",
-    "httpcore",
-    "urllib3",
-    "apscheduler.executors.default",
-    "apscheduler.scheduler",
-    "openai",
-)
 
+def _safe(value):
+    if isinstance(value, dict):
+        return {str(k): '[REDACTED]' if str(k).lower().replace('-', '_') in _SECRET_KEYS else _safe(v)
+                for k, v in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [_safe(item) for item in value]
+    if isinstance(value, str):
+        value = _ASSIGNMENT.sub(r'\1[REDACTED]', value)
+        value = _BEARER.sub(r'\1[REDACTED]', value)
+        return _URL_CREDENTIALS.sub(r'\1[REDACTED]@', value)
+    return value
 
-class _TraceIdFilter(logging.Filter):
-    """把当前上下文的 trace_id 注入 LogRecord，缺省为 '-'。"""
 
-    def filter(self, record: logging.LogRecord) -> bool:
-        record.trace_id = get_trace_id() or "-"
+class _TraceIdFilter(logging.Filter):
+    def filter(self, record):
+        record.trace_id = get_trace_id() or '-'
+        record.log_context = get_log_context()
         return True
 
 
+def _payload(record):
+    payload = {
+        'ts': datetime.fromtimestamp(record.created).astimezone().isoformat(timespec='milliseconds'),
+        'level': record.levelname,
+        'logger': record.name,
+        'trace_id': getattr(record, 'trace_id', '-'),
+        'message': _safe(record.getMessage()),
+    }
+    context = getattr(record, 'log_context', {})
+    kv = getattr(record, 'kv', {})
+    if isinstance(kv, dict):
+        payload.update({k: v for k, v in _safe(kv).items() if k not in _RESERVED})
+    # 绑定的执行身份优先于临时字段，防止覆盖关联信息。
+    payload.update({k: v for k, v in _safe(context).items() if k not in _RESERVED})
+    if record.exc_info:
+        payload['exc'] = _safe(logging.Formatter().formatException(record.exc_info))
+    if record.stack_info:
+        payload['stack'] = _safe(record.stack_info)
+    return payload
+
+
 class _JsonFormatter(logging.Formatter):
-    """一行一条 JSON 的格式化器。"""
-
-    def format(self, record: logging.LogRecord) -> str:
-        payload = {
-            "ts": datetime.fromtimestamp(record.created).isoformat(timespec="milliseconds"),
-            "level": record.levelname,
-            "logger": record.name,
-            "trace_id": getattr(record, "trace_id", "-"),
-            "message": record.getMessage(),
-        }
-        if record.exc_info:
-            payload["exc"] = self.formatException(record.exc_info)
-        # 允许调用方通过 logger.info("...", extra={"kv": {...}}) 附加结构化字段
-        kv = getattr(record, "kv", None)
-        if isinstance(kv, dict):
-            payload.update(kv)
-        return json.dumps(payload, ensure_ascii=False)
-
-
-def setup_logging(level: str = "INFO", fmt: str = "text") -> None:
-    """初始化根 logger。重复调用安全（先清空已有 handler）。"""
+    def format(self, record):
+        return json.dumps(_payload(record), ensure_ascii=False, default=str)
+
+
+class _TextFormatter(logging.Formatter):
+    def format(self, record):
+        p = _payload(record)
+        # JSON 转义保持一条普通日志一行；异常堆栈在下一行完整呈现。
+        message = str(p['message']).replace('\r', r'\r').replace('\n', r'\n')
+        fields = ' '.join(f'{key}={json.dumps(value, ensure_ascii=False, default=str)}'
+                          for key, value in p.items() if key not in _RESERVED)
+        line = f"{p['ts']} {p['level']:<7} [{p['trace_id']}] {p['logger']} | {message}"
+        if fields:
+            line += ' | ' + fields
+        if p.get('exc'):
+            line += '\n' + p['exc']
+        return line
+
+
+def setup_logging(level: str = 'INFO', fmt: str = 'text') -> None:
+    """重复配置只替换自身 handler，保留测试/宿主添加的采集器。"""
     root = logging.getLogger()
     root.setLevel(getattr(logging, str(level).upper(), logging.INFO))
-
     for handler in list(root.handlers):
-        root.removeHandler(handler)
-
+        if getattr(handler, '_vent_handler', False):
+            root.removeHandler(handler)
+            handler.close()
     handler = logging.StreamHandler(sys.stdout)
+    handler._vent_handler = True
     handler.addFilter(_TraceIdFilter())
-    if str(fmt).lower() == "json":
-        handler.setFormatter(_JsonFormatter())
-    else:
-        handler.setFormatter(logging.Formatter(_TEXT_FMT, datefmt=_DATE_FMT))
+    handler.setFormatter(_JsonFormatter() if str(fmt).lower() == 'json' else _TextFormatter())
     root.addHandler(handler)
-
     for name in _NOISY_LOGGERS:
         logging.getLogger(name).setLevel(logging.WARNING)
+    for name in ('uvicorn', 'uvicorn.error', 'uvicorn.access'):
+        server = logging.getLogger(name)
+        server.handlers.clear()
+        server.propagate = True
+        server.setLevel(logging.NOTSET)
+    # 请求汇总由 ASGI 中间件记录，包含流结束时间，避免 Uvicorn 再记一遍。
+    logging.getLogger('uvicorn.access').disabled = True
 
 
 def get_logger(name: str) -> logging.Logger:
-    """取 logger 的统一入口（留一层薄封装，将来换实现只改这里）。"""
     return logging.getLogger(name)

obs/tracing.py

@@ -9,11 +9,43 @@ ContextVar 自身保证。
 from __future__ import annotations
 
 import uuid
+import re
+from contextlib import contextmanager
 from contextvars import ContextVar
 from typing import Any
 
 # 当前请求的 trace_id；未绑定时为空串（而非 None，省掉调用方的判空）
 _trace_id: ContextVar[str] = ContextVar("trace_id", default="")
+_context: ContextVar[dict[str, Any]] = ContextVar("log_context", default={})
+_TRACE_PATTERN = re.compile(r"[A-Za-z0-9._:-]{1,128}\Z")
+
+
+def normalize_trace_id(value: str | None) -> str:
+    return value if isinstance(value, str) and _TRACE_PATTERN.fullmatch(value) else new_trace_id()
+
+
+def get_log_context() -> dict[str, Any]:
+    return dict(_context.get())
+
+
+def bind_log_context(**fields):
+    return _context.set({**_context.get(), **fields})
+
+
+def unbind_log_context(token) -> None:
+    try:
+        _context.reset(token)
+    except (ValueError, LookupError):
+        pass
+
+
+@contextmanager
+def log_context(**fields):
+    token = bind_log_context(**fields)
+    try:
+        yield
+    finally:
+        unbind_log_context(token)
 
 
 def new_trace_id() -> str:
@@ -32,7 +64,7 @@ def bind_trace_id(trace_id: str | None) -> Any:
     调用方应在 finally 中 ``unbind_trace_id(token)``。用 token 还原而不是
     直接重置为空串，是为了支持嵌套场景（外层已有 trace 时内层不要抹掉）。
     """
-    return _trace_id.set(trace_id or new_trace_id())
+    return _trace_id.set(normalize_trace_id(trace_id))
 
 
 def unbind_trace_id(token: Any) -> None:

release/deploy.py

@@ -171,7 +171,8 @@ def serve(prefix: Path):
         os.replace(temporary, path)
         try:
             uvicorn.run("main:app", host=config.HOST, port=config.PORT, workers=1,
-                        timeout_graceful_shutdown=60, log_level="info")
+                        timeout_graceful_shutdown=60, log_level=config.LOG_LEVEL.lower(),
+                        log_config=None, access_log=False)
         finally:
             if read_record(prefix).get("pid") == os.getpid():
                 path.unlink(missing_ok=True)

runtime/concurrency.py

@@ -114,7 +114,7 @@ class TaskSlot:
                 _per_user_count.pop(self.user_name, None)
             else:
                 _per_user_count[self.user_name] = user_count - 1
-        logger.info(
+        logger.debug(
             "释放并发位 user=%s 持有时长=%.1fs 剩余全局=%d 剩余用户=%d",
             self.user_name, self.held_seconds, _global_count[0],
             _per_user_count.get(self.user_name, 0),
@@ -145,7 +145,7 @@ def acquire_task_slot(user_name: str) -> TaskSlot:
             )
         _global_count[0] += 1
         _per_user_count[user] = current_user + 1
-        logger.info(
+        logger.debug(
             "占用并发位 user=%s 全局=%d/%d 用户=%d/%d",
             user, _global_count[0], config.MAX_CONCURRENT_TASKS,
             _per_user_count[user], config.MAX_USER_TASKS,

runtime/engine.py

@@ -24,7 +24,7 @@ from runtime.source_tracker import SourceTracker
 from runtime.subagent_tracker import SubagentTracker
 from billing.meter import UsageMeter
 from db import chat_store, db_conn, run_store
-from obs import get_logger, get_trace_id
+from obs import get_logger, get_trace_id, normalize_trace_id, bind_log_context, unbind_log_context
 
 logger = get_logger(__name__)
 
@@ -118,6 +118,7 @@ class Execution:
 
     def __init__(self, request: ExecutionRequest) -> None:
         self.req = request
+        request.trace_id = normalize_trace_id(request.trace_id or get_trace_id())
         self.writer = EventBuilder(request.session_id, request.trace_id or get_trace_id(), encode=lambda event: event)
         self.meter = _meter_for_request()
         self.tracker = SubagentTracker()
@@ -176,7 +177,11 @@ class Execution:
         if self._consumed:
             raise RuntimeError("同一任务只能执行一次，请创建新的 Execution")
         self._consumed = True
-        token = bind_trace_id(self.req.trace_id or None)
+        token = bind_trace_id(self.req.trace_id)
+        context_token = bind_log_context(run_id=self.req.run_id, session_id=self.req.session_id,
+                                         user_name=self.req.user_name, kind=self.req.kind)
+        started = time.perf_counter()
+        logger.info("任务开始", extra={"kv": {"event": "execution.started", "model": self.req.model or config.MODEL}})
         stream = None
         try:
             await asyncio.to_thread(run_store.start, self.req.run_id, self.req.session_id,
@@ -213,14 +218,32 @@ class Execution:
             async for frame in self._error_events(exc):
                 yield frame
         finally:
-            # 清理与发帧独立。ASGI 断连的取消作用域不能打断保存和释放。
-            with anyio.CancelScope(shield=True):
-                cleanup = asyncio.create_task(self._cleanup(stream))
+            cleanup_failed = False
+            try:
+                # 清理与发帧独立。ASGI 断连的取消作用域不能打断保存和释放。
+                with anyio.CancelScope(shield=True):
+                    cleanup = asyncio.create_task(self._cleanup(stream))
+                    try:
+                        await asyncio.shield(cleanup)
+                    except asyncio.CancelledError:
+                        await cleanup
+            except Exception:
+                cleanup_failed = True
+                logger.exception("任务收尾失败", extra={"kv": {"event": "execution.cleanup_failed"}})
+                raise
+            finally:
                 try:
-                    await asyncio.shield(cleanup)
-                except asyncio.CancelledError:
-                    await cleanup
-            unbind_trace_id(token)
+                    status = "cleanup_failed" if cleanup_failed else self.status
+                    level = "error" if status in {"failed", "cleanup_failed"} else "warning" if status == "cancelled" else "info"
+                    getattr(logger, level)("任务结束", extra={"kv": {
+                        "event": "execution.completed", "status": status,
+                        "duration_ms": round((time.perf_counter() - started) * 1000, 1),
+                        "input_tokens": self.billing.input_tokens, "output_tokens": self.billing.output_tokens,
+                        "tool_calls": len(self.tool_calls), "error_code": self.error or None,
+                    }})
+                finally:
+                    unbind_log_context(context_token)
+                    unbind_trace_id(token)
 
     async def _cleanup(self, stream: Any) -> None:
         try:
@@ -285,7 +308,7 @@ class Execution:
                     "冷线程补种历史 session=%s 条数=%d", self.req.session_id, len(messages)
                 )
         else:
-            logger.info("热线程，仅喂当轮消息 session=%s", self.req.session_id)
+            logger.debug("热线程，仅喂当轮消息 session=%s", self.req.session_id)
 
         # 用户消息先落库：即使这一轮流到一半挂了，历史里也得有用户问了什么
         await asyncio.to_thread(chat_store.save_message, self.req.session_id, "user",
@@ -601,7 +624,7 @@ class Execution:
             except Exception as exc:
                 self.status = "failed"
                 self.error = "message_persistence_failed"
-                logger.error("落库失败 session=%s：%s: %s", self.req.session_id, type(exc).__name__, exc)
+                logger.exception("回复落库失败", extra={"kv": {"event": "execution.persistence_failed"}})
 
         # 4) 上下文用量 + 压缩存档
         await self._persist_context(snapshot)
@@ -704,7 +727,7 @@ class Execution:
                 },
             )
         except Exception as exc:
-            logger.error(
+            logger.exception(
                 "上下文用量落库失败 session=%s：%s: %s",
                 self.req.session_id, type(exc).__name__, exc,
             )

runtime/environment.py

@@ -5,6 +5,7 @@ import asyncio
 from contextlib import AsyncExitStack
 import os
 import threading
+import time
 
 import config
 from agents import agent_memory, agent_registry
@@ -40,7 +41,7 @@ async def _stop_billing(stop: asyncio.Event, task: asyncio.Task) -> None:
 
 async def _prewarm() -> None:
     ok = await agent_registry.prewarm()
-    logger.info("agent 预热%s", "完成" if ok else "失败（将在首次请求时重试）")
+    logger.debug("agent 预热%s", "完成" if ok else "失败（将在首次请求时重试）")
 
 
 async def _join(task: asyncio.Task) -> None:
@@ -83,12 +84,19 @@ class RuntimeEnvironment:
             raise RuntimeError("已有活动的 RuntimeEnvironment，请复用外层环境")
         stack = self._stack = AsyncExitStack()
         stack.callback(_owner_lock.release)
+        started = time.perf_counter()
         try:
             setup_logging(config.LOG_LEVEL, config.LOG_FORMAT)
+            logger.info("运行环境启动", extra={"kv": {"event": "runtime.starting",
+                        "version": config.APP_VERSION, "dialect": config.DB_DIALECT, "model": config.MODEL,
+                        "scheduler": self.scheduler}})
             config.log_config_sources()
             _check_single_process()
             catalog = load_extensions()
             approval_names()
+            logger.info("扩展加载完成", extra={"kv": {"event": "extensions.loaded",
+                        "modules": catalog.modules, "disabled": catalog.disabled_modules,
+                        "tools": len(catalog.tools), "agents": len(catalog.agents)}})
 
             stack.callback(db_conn.close)
             db_conn.init_schema()
@@ -129,9 +137,14 @@ class RuntimeEnvironment:
 
             if not config.OPENAI_API_KEY:
                 logger.warning("OPENAI_API_KEY 未配置，对话会失败——请在 .env 里设置")
-            logger.info("运行环境就绪 scheduler=%s", self.scheduler)
+            logger.info("运行环境就绪", extra={"kv": {"event": "runtime.ready", "scheduler": self.scheduler,
+                        "duration_ms": round((time.perf_counter() - started) * 1000, 1)}})
             return self
-        except BaseException:
+        except BaseException as exc:
+            if isinstance(exc, asyncio.CancelledError):
+                logger.warning("运行环境启动已取消", extra={"kv": {"event": "runtime.cancelled"}})
+            else:
+                logger.exception("运行环境启动失败", extra={"kv": {"event": "runtime.startup_failed"}})
             await self._close()
             raise
 
@@ -151,7 +164,7 @@ class RuntimeEnvironment:
             task.result()
         finally:
             self._stack = None
-        logger.info("运行环境已关闭")
+        logger.info("运行环境已关闭", extra={"kv": {"event": "runtime.closed"}})
         if cancelled:
             raise asyncio.CancelledError
 

runtime/execution.py

@@ -62,17 +62,23 @@ async def execute(*, user_name: str, session_id: str, message: str, agent=None,
 
     独立批量任务应各自创建会话；额度与并发限制和聊天入口一致。
     """
-    from copy import deepcopy
-    admission = await admit(user_name, session_id)
+    from obs import bind_trace_id, unbind_trace_id, get_trace_id, log_context
+    token = bind_trace_id(trace_id or get_trace_id())
     try:
-        options = deepcopy(graph_config) if graph_config is not None else {}
-        options.setdefault("configurable", {})["thread_id"] = session_id
-        execution = await build_execution(admission, agent=agent, session_id=session_id,
-            thread_id=session_id, user_name=user_name, message=message, original_user_message=message,
-            kind=kind, config=options, trace_id=trace_id)
-        from runtime.interrupts import pending_interrupts, ResumeError
-        if await pending_interrupts(execution.req.agent, options):
-            raise ResumeError("pending_interrupt", "请先处理当前待确认请求")
-        return await execution.run()
+        with log_context(session_id=session_id, user_name=user_name, kind=kind):
+            from copy import deepcopy
+            admission = await admit(user_name, session_id)
+            try:
+                options = deepcopy(graph_config) if graph_config is not None else {}
+                options.setdefault("configurable", {})["thread_id"] = session_id
+                execution = await build_execution(admission, agent=agent, session_id=session_id,
+                    thread_id=session_id, user_name=user_name, message=message, original_user_message=message,
+                    kind=kind, config=options, trace_id=get_trace_id())
+                from runtime.interrupts import pending_interrupts, ResumeError
+                if await pending_interrupts(execution.req.agent, options):
+                    raise ResumeError("pending_interrupt", "请先处理当前待确认请求")
+                return await execution.run()
+            finally:
+                admission.release()
     finally:
-        admission.release()
+        unbind_trace_id(token)

runtime/subagent_tracker.py

@@ -103,7 +103,7 @@ class SubagentTracker:
             call_id = call.get("id") or ""
             if not call_id:
                 # 没有 id 就无法把后续子图事件关联回来，跳过而不是编一个
-                logger.debug("task 调用缺少 id，跳过跟踪：%s", call)
+                logger.debug("task 调用缺少 id，跳过跟踪")
                 continue
             args = call.get("args") or {}
             if not isinstance(args, dict):
@@ -122,8 +122,8 @@ class SubagentTracker:
                 self._order.append(call_id)
                 touched.append(run)
                 logger.info(
-                    "子智能体启动 id=%s type=%s desc=%s",
-                    call_id[:8], subagent_type, description[:60],
+                    "子智能体启动", extra={"kv": {"event": "subagent.started",
+                    "tool_call_id": call_id, "agent": subagent_type}},
                 )
             elif existing.status == "started":
                 existing.status = "running"
@@ -156,6 +156,8 @@ class SubagentTracker:
         logger.info(
             "子智能体结束 id=%s type=%s status=%s 耗时=%sms 工具调用=%d",
             run.call_id[:8], run.subagent_type, run.status, run.duration_ms, run.tool_calls,
+            extra={"kv": {"event": "subagent.completed", "tool_call_id": run.call_id,
+                          "agent": run.subagent_type, "status": run.status, "duration_ms": run.duration_ms}},
         )
         return run
 

scheduler/runner.py

@@ -12,12 +12,15 @@
 from __future__ import annotations
 
 import asyncio
+import logging
+import time
 from typing import Any
 
 import config
-from billing import pricing
+from billing import pricing, guard
+from runtime import concurrency
 from db import chat_store, db_conn, schedule_store
-from obs import get_logger
+from obs import get_logger, get_trace_id, bind_trace_id, unbind_trace_id, log_context
 
 logger = get_logger(__name__)
 
@@ -27,6 +30,32 @@ SCHEDULED_RECURSION_LIMIT = 30
 
 
 async def run_schedule(schedule_id: str, trigger: str = "cron") -> dict[str, Any]:
+    token = bind_trace_id(get_trace_id())
+    started = time.perf_counter()
+    try:
+        with log_context(schedule_id=schedule_id, trigger=trigger):
+            try:
+                result = await _run_schedule(schedule_id, trigger)
+            except asyncio.CancelledError:
+                logger.warning("定时执行已取消", extra={"kv": {"event": "schedule.cancelled"}})
+                raise
+            except (guard.QuotaExceededError, concurrency.TooManyTasksError, concurrency.SessionBusyError) as exc:
+                logger.warning("定时执行未获准", extra={"kv": {"event": "schedule.rejected", "error_type": type(exc).__name__}})
+                raise
+            except Exception:
+                logger.exception("定时执行失败", extra={"kv": {"event": "schedule.failed"}})
+                raise
+            logger.log(logging.INFO if result.get("ok") else logging.WARNING, "定时执行结束", extra={"kv": {
+                "event": "schedule.completed", "status": result.get("status", result.get("error")),
+                "schedule_run_id": result.get("run_id"), "session_id": result.get("session_id"),
+                "duration_ms": round((time.perf_counter() - started) * 1000, 1),
+            }})
+            return result
+    finally:
+        unbind_trace_id(token)
+
+
+async def _run_schedule(schedule_id: str, trigger: str) -> dict[str, Any]:
     """执行并持久化状态；准入失败由 HTTP 或调度回调处理。"""
     schedule = await asyncio.to_thread(schedule_store.get_schedule, schedule_id)
     if schedule is None:

scheduler/scheduler.py

@@ -42,8 +42,8 @@ async def _invoke(schedule_id: str) -> None:
     _active.add(task)
     try:
         await run_schedule(schedule_id, trigger="cron")
-    except Exception:  # pragma: no cover - run_schedule 内部已兜底
-        logger.exception("定时任务回调异常 id=%s", schedule_id)
+    except Exception:  # runner 已记录带 schedule_id / trace_id 的错误，避免重复堆栈
+        pass
     finally:
         _active.discard(task)
 

scripts/selfcheck.py

@@ -49,7 +49,7 @@ MODULES = [
     "api.source_tracker", "api.concurrency", "api.sse_core", "api.models",
     "runtime.environment", "runtime.engine", "runtime.execution", "runtime.events", "runtime.interrupts",
     "runtime.concurrency", "runtime.context_meter", "runtime.source_tracker", "runtime.subagent_tracker",
-    "api.chat_routes", "api.session_routes", "api.model_routes",
+    "api.request_logging", "api.chat_routes", "api.session_routes", "api.model_routes",
     "api.usage_routes", "api.schedule_routes", "api",
     "scheduler.runner", "scheduler.scheduler", "main",
 ]

tests/test_logging.py

@@ -0,0 +1,284 @@
+"""日志格式、关联信息、SSE 请求汇总和异常脱敏。"""
+import asyncio
+import io
+import json
+import logging
+from pathlib import Path
+
+import pytest
+
+from obs import (setup_logging, get_logger, bind_trace_id, unbind_trace_id, get_trace_id,
+                 get_log_context, log_context, normalize_trace_id)
+from obs.log_setup import _NOISY_LOGGERS
+from api.request_logging import RequestLogMiddleware
+
+
+@pytest.fixture
+def logs(monkeypatch):
+    root = logging.getLogger()
+    previous_handlers, previous_level = root.handlers[:], root.level
+    names = (*_NOISY_LOGGERS, 'uvicorn', 'uvicorn.error', 'uvicorn.access')
+    states = {name: (logging.getLogger(name).handlers[:], logging.getLogger(name).level,
+                     logging.getLogger(name).propagate, logging.getLogger(name).disabled) for name in names}
+    root.handlers = []
+    stream = io.StringIO()
+    monkeypatch.setattr('sys.stdout', stream)
+    def configure(level='INFO', fmt='json'):
+        monkeypatch.setattr('sys.stdout', stream)
+        setup_logging(level, fmt)
+    stream.configure = configure
+    configure()
+    try:
+        yield stream
+    finally:
+        for handler in root.handlers:
+            handler.close()
+        root.handlers = previous_handlers
+        root.setLevel(previous_level)
+        for name, (handlers, level, propagate, disabled) in states.items():
+            target = logging.getLogger(name)
+            target.handlers = handlers
+            target.setLevel(level)
+            target.propagate = propagate
+            target.disabled = disabled
+
+
+def records(logs):
+    return [json.loads(line) for line in logs.getvalue().splitlines() if line]
+
+
+def test_json_context_fields_redaction_and_exception(logs):
+    token = bind_trace_id('request-123')
+    try:
+        with log_context(run_id='run-1', session_id='session-1'):
+            try:
+                raise RuntimeError('api_key=private-key password=private-password Bearer private-token')
+            except RuntimeError:
+                get_logger('test').exception('连接失败 https://user:private-url-password@host/path', extra={'kv': {
+                    'event': 'test.failed', 'trace_id': 'forged', 'run_id': 'forged', 'message': 'forged',
+                    'api_key': 'structured-key', 'nested': {'Authorization': 'structured-bearer'}, 'path': Path('/tmp/test'),
+                }})
+    finally:
+        unbind_trace_id(token)
+    result, = records(logs)
+    assert result['trace_id'] == 'request-123' and result['run_id'] == 'run-1'
+    assert result['session_id'] == 'session-1' and result['event'] == 'test.failed'
+    assert 'Traceback' in result['exc'] and 'RuntimeError' in result['exc']
+    assert result['api_key'] == '[REDACTED]'
+    assert result['nested']['Authorization'] == '[REDACTED]'
+    assert result['path'] == '/tmp/test'
+    assert all(secret not in logs.getvalue() for secret in (
+        'private-key', 'private-password', 'private-token', 'private-url-password', 'structured-key', 'structured-bearer'))
+    assert get_log_context() == {} and not get_trace_id()
+    assert result['ts'][-6] in '+-'
+
+
+def test_text_fields_and_reconfiguration_do_not_duplicate(logs):
+    foreign = logging.NullHandler()
+    logging.getLogger().addHandler(foreign)
+    logs.configure('INFO', 'text')
+    logs.configure('INFO', 'text')
+    assert foreign in logging.getLogger().handlers
+    assert sum(bool(getattr(h, '_vent_handler', False)) for h in logging.getLogger().handlers) == 1
+    get_logger('uvicorn.error').info('服务器就绪', extra={'kv': {'event': 'server.ready', 'duration_ms': 12}})
+    get_logger('uvicorn.access').info('duplicate request')
+    get_logger('runtime.concurrency').debug('并发细节')
+    assert logs.getvalue().count('服务器就绪') == 1
+    assert 'event="server.ready"' in logs.getvalue() and 'duration_ms=12' in logs.getvalue()
+    assert 'duplicate request' not in logs.getvalue() and '并发细节' not in logs.getvalue()
+
+
+@pytest.mark.parametrize('value', ['line\nbreak', 'a' * 129, '中文', '', None])
+def test_invalid_trace_is_replaced(value):
+    result = normalize_trace_id(value)
+    assert len(result) == 32 and all(c in '0123456789abcdef' for c in result)
+
+
+@pytest.mark.asyncio
+async def test_context_isolated_between_tasks_and_propagates_to_threads(logs):
+    async def run(name):
+        token = bind_trace_id(name)
+        try:
+            with log_context(session_id=name):
+                await asyncio.sleep(0)
+                await asyncio.to_thread(get_logger('thread').info, '线程日志')
+        finally:
+            unbind_trace_id(token)
+    await asyncio.gather(run('one'), run('two'))
+    assert {(r['trace_id'], r['session_id']) for r in records(logs)} == {('one', 'one'), ('two', 'two')}
+    assert not get_trace_id() and not get_log_context()
+
+
+def scope(path='/api/chat'):
+    return {'type': 'http', 'method': 'POST', 'path': path, 'headers': [(b'x-trace-id', b'http-test')],
+            'query_string': b'api_key=never-log-query'}
+
+
+@pytest.mark.asyncio
+async def test_sse_request_logged_only_after_stream_and_cleanup(logs):
+    first, finish = asyncio.Event(), asyncio.Event()
+    sent = []
+    async def app(scope, receive, send):
+        await send({'type': 'http.response.start', 'status': 200, 'headers': []})
+        await send({'type': 'http.response.body', 'body': b'data: secret-body\n\n', 'more_body': True})
+        first.set()
+        await finish.wait()
+        await send({'type': 'http.response.body', 'body': b'', 'more_body': False})
+        get_logger('cleanup').info('清理完成')
+    async def receive():
+        return {'type': 'http.request', 'body': b'never-log-request'}
+    async def send(message):
+        sent.append(message)
+    task = asyncio.create_task(RequestLogMiddleware(app)(scope(), receive, send))
+    await first.wait()
+    assert records(logs) == []
+    finish.set()
+    await task
+    result = records(logs)
+    assert result[-1]['event'] == 'http.request.completed'
+    assert result[-1]['outcome'] == 'completed' and result[-1]['status_code'] == 200
+    assert result[-1]['duration_ms'] >= 0
+    assert all(r['trace_id'] == 'http-test' for r in result)
+    assert (b'x-trace-id', b'http-test') in sent[0]['headers']
+    assert 'never-log' not in logs.getvalue() and 'secret-body' not in logs.getvalue()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('mode,status,outcome', [('error', 500, 'error'), ('cancel', None, 'cancelled'), ('disconnect', None, 'disconnected')])
+async def test_request_failures_logged_once_and_restore_context(logs, mode, status, outcome):
+    async def app(scope, receive, send):
+        if mode == 'error':
+            raise ValueError('failed')
+        if mode == 'cancel':
+            raise asyncio.CancelledError
+        await receive()
+    async def receive():
+        return {'type': 'http.disconnect'}
+    async def send(message):
+        pass
+    request = RequestLogMiddleware(app)(scope(), receive, send)
+    if mode in {'error', 'cancel'}:
+        with pytest.raises(ValueError if mode == 'error' else asyncio.CancelledError):
+            await request
+    else:
+        await request
+    row, = records(logs)
+    assert row['status_code'] == status and row['outcome'] == outcome
+    assert not get_trace_id()
+
+
+@pytest.mark.asyncio
+async def test_health_noise_is_debug_but_errors_are_visible(logs):
+    logs.configure('INFO', 'json')
+    status = 200
+    async def app(scope, receive, send):
+        await send({'type': 'http.response.start', 'status': status})
+        await send({'type': 'http.response.body'})
+    async def noop(*args):
+        pass
+    await RequestLogMiddleware(app)(scope('/health'), noop, noop)
+    assert records(logs) == []
+    status = 503
+    await RequestLogMiddleware(app)(scope('/health'), noop, noop)
+    assert records(logs)[0]['level'] == 'ERROR'
+
+
+@pytest.mark.asyncio
+async def test_batch_logs_events_and_billing_share_trace(logs, monkeypatch, tmp_path, pricing_ready):
+    import config
+    from db import chat_store, billing_store, db_conn
+    from runtime.execution import execute
+    from runtime import concurrency
+    from test_execution import Agent
+    monkeypatch.setattr(config, 'ROOT', tmp_path)
+    concurrency.reset_for_tests()
+    sessions = [chat_store.create_session('log-user')['id'] for _ in range(2)]
+    await asyncio.gather(*(execute(user_name='log-user', session_id=sid, message='never-log-prompt', agent=Agent()) for sid in sessions))
+    completed = [r for r in records(logs) if r.get('event') == 'execution.completed']
+    assert len(completed) == 2
+    assert {r['session_id'] for r in completed} == set(sessions)
+    assert len({r['trace_id'] for r in completed}) == 2
+    for row in completed:
+        assert row['status'] == 'success' and row['input_tokens'] == 8
+        usage = db_conn.query_one('SELECT trace_id FROM billing_usage WHERE id = ?', (row['run_id'],))
+        assert usage['trace_id'] == row['trace_id']
+        start = next(r for r in records(logs) if r.get('event') == 'execution.started' and r['run_id'] == row['run_id'])
+        assert start['trace_id'] == row['trace_id']
+    assert 'never-log-prompt' not in logs.getvalue()
+    assert not get_trace_id() and not get_log_context()
+
+
+@pytest.mark.asyncio
+async def test_real_http_response_header_and_access_log_match(logs, monkeypatch):
+    import httpx
+    from main import app
+    # main 导入可能重新配置日志，将本例格式明确设回 JSON。
+    logs.configure('INFO', 'json')
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url='http://test') as client:
+        response = await client.get('/api/no-such-route', headers={'X-Trace-Id': 'invalid trace'})
+    rows = [r for r in records(logs) if r.get('event') == 'http.request.completed']
+    assert len(rows) == 1 and rows[0]['status_code'] == 404
+    assert rows[0]['trace_id'] == response.headers['x-trace-id']
+    assert len(rows[0]['trace_id']) == 32
+
+
+@pytest.mark.asyncio
+async def test_http_unhandled_exception_keeps_trace_and_stack(logs):
+    import httpx
+    from fastapi import FastAPI
+    from main import unhandled_exception_handler
+    app = FastAPI()
+    app.add_middleware(RequestLogMiddleware)
+    app.add_exception_handler(Exception, unhandled_exception_handler)
+    @app.get('/broken')
+    async def broken():
+        raise RuntimeError('password=hidden-error-password')
+    logs.configure()
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app, raise_app_exceptions=False), base_url='http://test') as client:
+        response = await client.get('/broken', headers={'X-Trace-Id': 'broken-request'})
+    assert response.status_code == 500
+    assert response.headers['x-trace-id'] == 'broken-request'
+    related = [r for r in records(logs) if r.get('event', '').startswith('http.request.')]
+    assert len(related) == 2  # 一条请求汇总、一条错误堆栈。
+    assert all(r['trace_id'] == 'broken-request' for r in related)
+    assert 'Traceback' in next(r for r in related if r['event'] == 'http.request.failed')['exc']
+    assert 'hidden-error-password' not in logs.getvalue()
+
+
+@pytest.mark.asyncio
+async def test_execution_event_trace_matches_logs_when_not_provided(logs, monkeypatch, tmp_path, pricing_ready):
+    import config
+    from db import chat_store
+    from runtime.execution import admit, build_execution
+    from test_execution import Agent
+    monkeypatch.setattr(config, 'ROOT', tmp_path)
+    sid = chat_store.create_session('event-user')['id']
+    admission = await admit('event-user', sid)
+    execution = await build_execution(admission, agent=Agent(), session_id=sid, thread_id=sid,
+        user_name='event-user', message='问题', config={'configurable': {'thread_id': sid}})
+    events = [event async for event in execution.events()]
+    assert events and all(event['trace_id'] == execution.req.trace_id for event in events)
+    completed = [r for r in records(logs) if r.get('event') == 'execution.completed']
+    assert len(completed) == 1 and completed[0]['trace_id'] == execution.req.trace_id
+    assert not get_trace_id() and not get_log_context()
+
+
+@pytest.mark.asyncio
+async def test_schedule_and_execution_share_trace(logs, monkeypatch, tmp_path, pricing_ready):
+    import config
+    from db import schedule_store
+    from scheduler.runner import run_schedule
+    from test_execution import Agent
+    monkeypatch.setattr(config, 'ROOT', tmp_path)
+    monkeypatch.setattr('agents.agent_registry.get_agent', lambda: Agent())
+    schedule = schedule_store.create_schedule('log-user', 'report', '0 9 * * *', 'never-log-schedule-prompt')
+    result = await run_schedule(schedule['id'])
+    assert result['ok']
+    run = next(r for r in records(logs) if r.get('event') == 'execution.completed')
+    scheduled = next(r for r in records(logs) if r.get('event') == 'schedule.completed')
+    assert run['trace_id'] == scheduled['trace_id'] != '-'
+    assert run['schedule_id'] == scheduled['schedule_id'] == schedule['id']
+    assert run['session_id'] == scheduled['session_id'] == result['session_id']
+    assert scheduled['schedule_run_id'] == result['run_id']
+    assert 'never-log-schedule-prompt' not in logs.getvalue()
+    assert not get_trace_id() and not get_log_context()