fix(logging): 按 loguru 契约传递占位符和异常信息

backend/test/unit/server/test_loguru_logging_contract.py

@@ -0,0 +1,85 @@
+"""静态校验 loguru 调用不使用标准 logging 的占位符与异常约定。"""
+
+from __future__ import annotations
+
+import ast
+import re
+from pathlib import Path
+
+import pytest
+
+LOGURU_LOGGER_MODULES = frozenset({"yuxi.infrastructure.observability.logging", "loguru"})
+LOG_LEVELS = frozenset({"debug", "info", "warning", "error", "exception", "critical"})
+PERCENT_PLACEHOLDER = re.compile(r"%[sdifr]")
+YUXI_ROOT = Path(__file__).resolve().parents[3] / "yuxi"
+
+
+def _imports_loguru_logger(tree: ast.Module) -> bool:
+    """判断模块是否导入了 loguru 的 logger。"""
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.ImportFrom):
+            continue
+        names = {alias.name for alias in node.names}
+        if "logger" in names and (node.module or "") in LOGURU_LOGGER_MODULES:
+            return True
+    return False
+
+
+def _log_level(node: ast.Call) -> str | None:
+    """返回 logger.<level>() 或 logger.opt(...).<level>() 的级别名。"""
+    func = node.func
+    if not isinstance(func, ast.Attribute):
+        return None
+    if isinstance(func.value, ast.Name) and func.value.id == "logger":
+        return func.attr
+    if isinstance(func.value, ast.Call):
+        inner = func.value.func
+        if isinstance(inner, ast.Attribute) and isinstance(inner.value, ast.Name) and inner.value.id == "logger":
+            return func.attr
+    return None
+
+
+def find_loguru_violations(source: str) -> list[tuple[int, str]]:
+    """返回 loguru 调用中误用 % 占位符或 exc_info 的位置与原因。"""
+    tree = ast.parse(source)
+    if not _imports_loguru_logger(tree):
+        return []
+
+    violations: list[tuple[int, str]] = []
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.Call):
+            continue
+        level = _log_level(node)
+        if level not in LOG_LEVELS:
+            continue
+        message = node.args[0] if node.args else None
+        if isinstance(message, ast.Constant) and isinstance(message.value, str) and PERCENT_PLACEHOLDER.search(message.value):
+            violations.append((node.lineno, f"logger.{level} 使用 % 占位符"))
+        if any(keyword.arg == "exc_info" for keyword in node.keywords):
+            violations.append((node.lineno, f"logger.{level} 使用 exc_info 关键字"))
+    return violations
+
+
+def test_backend_loguru_calls_follow_loguru_formatting() -> None:
+    """生产代码中的 loguru 调用必须使用 {} 占位符和 opt(exception=True)。"""
+    violations = []
+    for path in sorted(YUXI_ROOT.rglob("*.py")):
+        source = path.read_text(encoding="utf-8")
+        for line, reason in find_loguru_violations(source):
+            violations.append(f"{path.relative_to(YUXI_ROOT)}:{line}: {reason}")
+
+    assert not violations, "\n".join(violations)
+
+
+@pytest.mark.parametrize(
+    ("source", "expected"),
+    [
+        ('from loguru import logger\nlogger.warning("run=%s", 1)\n', 1),
+        ('from loguru import logger\nlogger.error("failed", exc_info=True)\n', 1),
+        ('from yuxi.infrastructure.observability.logging import logger\nlogger.opt(exception=True).error("run={}", 1)\n', 0),
+        ('import logging\nlogger = logging.getLogger(__name__)\nlogger.warning("run=%s", 1)\n', 0),
+    ],
+)
+def test_loguru_violation_detector_distinguishes_logging_conventions(source: str, expected: int) -> None:
+    """负向样例证明检测器能发现旧写法，正向样例证明不误报标准 logging。"""
+    assert len(find_loguru_violations(source)) == expected

backend/yuxi/bootstrap/worker.py

@@ -123,7 +123,7 @@ async def _reconcile_agent_runs_once() -> None:
             await operation()
         except Exception:
             healthy = False
-            logger.exception("Agent reconciliation failed: phase=%s", phase)
+            logger.exception("Agent reconciliation failed: phase={}", phase)
     if healthy:
         try:
             await _publish_reconciliation_health()
@@ -148,12 +148,12 @@ async def _reconcile_background_jobs_forever() -> None:
         try:
             reconciled = await reconcile_and_publish_jobs()
             if reconciled:
-                logger.warning("Reconciled expired durable jobs: count=%s", len(reconciled))
+                logger.warning("Reconciled expired durable jobs: count={}", len(reconciled))
             await _publish_job_reconciliation_health()
         except asyncio.CancelledError:
             raise
         except Exception:
-            logger.error("Failed to reconcile durable jobs", exc_info=True)
+            logger.opt(exception=True).error("Failed to reconcile durable jobs")
 
 
 async def _publish_job_reconciliation_health() -> None:

backend/yuxi/modules/agents/runtime/callbacks/model_request_timing.py

@@ -52,4 +52,4 @@ async def persist(self, *, run_id: str, worker_id: str) -> None:
                 )
             self._persisted = True
         except Exception:
-            logger.warning(f"Failed to persist first model request timing: run={run_id}", exc_info=True)
+            logger.opt(exception=True).warning(f"Failed to persist first model request timing: run={run_id}")

backend/yuxi/modules/agents/services/attachments.py

@@ -159,7 +159,7 @@ async def parse_draft_file(*, file_id: str, parse_method: str | None, scope: Act
             pass
         if isinstance(exc, (asyncio.CancelledError, DomainError)):
             raise
-        logger.warning("Draft attachment parse failed: id=%s error=%s", file_id, exc)
+        logger.warning("Draft attachment parse failed: id={} error={}", file_id, exc)
         raise InvalidOperationError("附件解析失败") from exc
     return {**_file_response(attachment), "parse_method": method}
 
@@ -231,7 +231,7 @@ async def prepare_input_attachments(*, db, agent_session, input_item, binding) -
             await db.refresh(attachment)
             attachment.error = "附件准备失败，等待恢复重试"
         await db.flush()
-        logger.exception("附件准备失败，保留 Input 等待重试: %s", input_item.id)
+        logger.exception("附件准备失败，保留 Input 等待重试: {}", input_item.id)
         return False
     return True
 
@@ -246,7 +246,7 @@ async def cleanup_prepared_sources(*, db, thread_id: str) -> None:
         try:
             await _delete_draft_storage(client, attachment)
         except (StorageError, OSError):
-            logger.exception("就绪附件临时内容清理失败: %s", attachment.id)
+            logger.exception("就绪附件临时内容清理失败: {}", attachment.id)
             continue
         attachment.object_name = None
         attachment.parsed_source = None
@@ -307,7 +307,7 @@ async def delete_thread_attachment_view(
         except FileNotFoundError:
             pass
         except OSError:
-            logger.exception("附件记录已删除，Workdir 内容清理失败: session=%s scope=%s", thread_id, scope)
+            logger.exception("附件记录已删除，Workdir 内容清理失败: session={} scope={}", thread_id, scope)
     return {"message": "附件已删除"}
 
 
@@ -359,7 +359,7 @@ async def _cleanup_expired_drafts(scope, db) -> None:
         try:
             await _delete_draft_storage(client, attachment)
         except (StorageError, OSError):
-            logger.exception("过期草稿清理失败: %s", attachment.id)
+            logger.exception("过期草稿清理失败: {}", attachment.id)
             continue
         await db.delete(attachment)
     await db.commit()

backend/yuxi/modules/agents/services/cooperation.py

@@ -491,7 +491,7 @@ async def recover_cooperation_waits() -> None:
                 await _recover_cooperation_wait(turn_id)
             except Exception:
                 failures += 1
-                logger.exception("Cooperation recovery failed: turn_id=%s", turn_id)
+                logger.exception("Cooperation recovery failed: turn_id={}", turn_id)
         after = ids[-1]
     if failures:
         raise RuntimeError(f"协作等待恢复失败：{failures} 条")
@@ -697,13 +697,13 @@ async def reconcile_stopped_trees(*, tree_root_thread_id: str | None = None) ->
                         waiting.append((target, turn.id))
         except Exception:
             failures += 1
-            logger.exception("Stopped tree recovery failed: thread_id=%s", member.thread_id)
+            logger.exception("Stopped tree recovery failed: thread_id={}", member.thread_id)
     for target, turn_id in waiting:
         try:
             await settle_waiting_cancel(thread_id=target.thread_id, turn_id=turn_id, uid=target.uid, app_id=target.app_id)
         except Exception:
             failures += 1
-            logger.exception("Stopped waiting cleanup failed: turn_id=%s", turn_id)
+            logger.exception("Stopped waiting cleanup failed: turn_id={}", turn_id)
     if failures:
         raise RuntimeError(f"停止会话恢复失败：{failures} 条")
 

backend/yuxi/modules/agents/services/event_writer.py

@@ -46,15 +46,15 @@ async def flush(self) -> None:
         try:
             await append_run_stream_events(self.run_id, events)
         except Exception:
-            logger.warning("公开事件发布失败: run=%s", self.run_id, exc_info=True)
+            logger.opt(exception=True).warning("公开事件发布失败: run={}", self.run_id)
 
 
 async def append_run_event_best_effort(run_id: str, event: dict) -> bool:
     """发布已适配的事件，PostgreSQL 终态不依赖 Redis 可用性。"""
     try:
         await append_run_stream_events(run_id, [event])
     except Exception:
-        logger.warning("公开事件发布失败: run=%s type=%s", run_id, event.get("type"), exc_info=True)
+        logger.opt(exception=True).warning("公开事件发布失败: run={} type={}", run_id, event.get("type"))
         return False
     return True
 

backend/yuxi/modules/agents/services/execution.py

@@ -506,7 +506,7 @@ async def stream_agent_execution(
         try:
             await adapter.persist_incomplete()
         except Exception:
-            logger.warning("失败执行的部分 item 未能持久化", exc_info=True)
+            logger.opt(exception=True).warning("失败执行的部分 item 未能持久化")
         await _persist_model_request_timing(model_request_recorder, meta)
         async with pg_manager.get_async_session_context() as new_db:
             output = await save_partial_message(

backend/yuxi/modules/agents/services/leases.py

@@ -95,7 +95,7 @@ async def reconcile_expired_run_leases(*, now: datetime | None = None) -> list[s
                 reconciled.append(run.id)
         except Exception:
             failures += 1
-            logger.exception("Expired lease recovery failed: run_id=%s", run_id)
+            logger.exception("Expired lease recovery failed: run_id={}", run_id)
     for turn_id in terminal_turn_ids:
         await finish_turn_observation_if_terminal(turn_id)
     await reconcile_pending_runtime_cleanups()
@@ -123,7 +123,7 @@ async def reconcile_pending_runtime_cleanups() -> list[str]:
             cleaned.append(run.id)
         except Exception:
             failures += 1
-            logger.exception("Run cleanup recovery failed: run_id=%s", run.id)
+            logger.exception("Run cleanup recovery failed: run_id={}", run.id)
     await reconcile_cancelling_turns()
     if failures:
         raise RuntimeError(f"执行清理恢复失败：{failures} 条")

backend/yuxi/modules/agents/services/run_lifecycle.py

@@ -304,7 +304,7 @@ async def reconcile_cancelling_turns() -> list[str]:
                     settled.append(candidate.id)
         except Exception:
             failures += 1
-            logger.exception("Failed to reconcile cancelling Turn: %s", candidate.id)
+            logger.exception("Failed to reconcile cancelling Turn: {}", candidate.id)
     if failures:
         raise RuntimeError(f"cancelling Turn recovery failed for {failures} records")
     return settled

backend/yuxi/modules/agents/services/runner.py

@@ -187,7 +187,7 @@ async def _watch_durable_cancel(self) -> None:
             try:
                 cancelled = await _is_cancel_requested(self.run_id)
             except Exception:
-                logger.error(f"Failed to read durable AgentRun cancellation: run={self.run_id}", exc_info=True)
+                logger.opt(exception=True).error(f"Failed to read durable AgentRun cancellation: run={self.run_id}")
                 self.cancel_event.set()
                 return
             else:
@@ -213,7 +213,7 @@ async def _heartbeat_lease(self) -> None:
                     # 终态事务已清除 lease；本 attempt 仍需完成流收尾、清理和事件发布。
                     return
             except Exception:
-                logger.error(f"Failed to renew AgentRun lease: run={self.run_id}", exc_info=True)
+                logger.opt(exception=True).error(f"Failed to renew AgentRun lease: run={self.run_id}")
                 renewed = False
             if not renewed:
                 self.lease_lost = True
@@ -231,7 +231,7 @@ async def _cancel_sandbox_commands(self) -> None:
         try:
             await await_io(asyncio.to_thread(get_sandbox_provider().cleanup_run, self.uid, self.run_id, self.worker_id))
         except Exception:
-            logger.exception("Run 外部命令尚未确认停止，保留持久清理责任: run=%s", self.run_id)
+            logger.exception("Run 外部命令尚未确认停止，保留持久清理责任: run={}", self.run_id)
 
 
 async def _require_runtime_cleanup(run: AgentRun, message: str) -> None:
@@ -306,7 +306,7 @@ async def _record_run_timing_best_effort(
             else:
                 raise ValueError(f"不支持的 AgentRun timing phase: {phase}")
     except Exception:
-        logger.warning("Failed to persist AgentRun timing: run=%s, phase=%s", run_id, phase, exc_info=True)
+        logger.opt(exception=True).warning("Failed to persist AgentRun timing: run={}, phase={}", run_id, phase)
 
 
 async def _load_user(uid: str):
@@ -326,7 +326,7 @@ async def _confirmed_user_cancel(run_id: str) -> bool:
     try:
         return await _is_cancel_requested(run_id)
     except Exception:
-        logger.error(f"Failed to confirm durable AgentRun cancellation: run={run_id}", exc_info=True)
+        logger.opt(exception=True).error(f"Failed to confirm durable AgentRun cancellation: run={run_id}")
         return False
 
 
@@ -342,7 +342,7 @@ async def _read_run_token_usage_from_state(*, run_id: str, thread_id: str, curre
                 include_relations=False,
             )
     except Exception:
-        logger.warning(f"Failed to read token usage from state for run {run_id}", exc_info=True)
+        logger.opt(exception=True).warning(f"Failed to read token usage from state for run {run_id}")
         return None
 
     agent_state = view.get("agent_state") if isinstance(view, dict) else None
@@ -613,7 +613,7 @@ async def _process_agent_run(ctx, run_id: str, execution_timeout: asyncio.Timeou
             except Exception as manifest_error:
                 if await _is_cancel_requested(run_id):
                     raise asyncio.CancelledError(f"run {run_id} cancelled during preparation")
-                logger.error(f"Failed to persist AgentRun manifest: run={run_id}", exc_info=True)
+                logger.opt(exception=True).error(f"Failed to persist AgentRun manifest: run={run_id}")
                 raise RunAdmissionRejected("manifest_persist_failed", f"运行清单固化失败，执行未开始：{manifest_error}")
 
         except RunAdmissionRejected as rejection:
@@ -764,7 +764,7 @@ async def record_prepared() -> None:
         try:
             released = await release_run_lease_for_retry(run_id, worker_id)
         except Exception:
-            logger.error(f"Infrastructure cancellation could not release AgentRun lease: run={run_id}", exc_info=True)
+            logger.opt(exception=True).error(f"Infrastructure cancellation could not release AgentRun lease: run={run_id}")
             raise cancellation
         if not released and await _confirmed_user_cancel(run_id):
             transition = await _settle_uncommitted_run(
@@ -810,7 +810,7 @@ async def record_prepared() -> None:
                 ),
             )
             raise RetryableRunError(str(exc)) from exc
-        logger.error("Run failed %s: %s", run_id, exc)
+        logger.error("Run failed {}: {}", run_id, exc)
         await _settle_uncommitted_run(
             run_id,
             "failed",
@@ -827,7 +827,7 @@ async def record_prepared() -> None:
         try:
             final_run = await _get_run(run_id)
         except Exception:
-            logger.error(f"Failed to load AgentRun during lifecycle cleanup: run={run_id}", exc_info=True)
+            logger.opt(exception=True).error(f"Failed to load AgentRun during lifecycle cleanup: run={run_id}")
             final_run = None
         if final_run:
             await finish_run_lifecycle(final_run)

backend/yuxi/modules/agents/services/scheduler.py

@@ -98,7 +98,7 @@ async def deliver(dispatch: Dispatch) -> bool:
     try:
         await enqueue_agent_run(dispatch.run_id)
     except Exception:
-        logger.exception("Failed to publish committed AgentRun: %s", dispatch.run_id)
+        logger.exception("Failed to publish committed AgentRun: {}", dispatch.run_id)
         return False
     return True
 
@@ -150,15 +150,15 @@ async def recover_pending_dispatches() -> None:
                 failures += 1
         except Exception:
             failures += 1
-            logger.exception("Failed to republish pending AgentRun: %s", run.id)
+            logger.exception("Failed to republish pending AgentRun: {}", run.id)
 
     for uid, agent_slug, thread_id in scopes:
         try:
             if not await dispatch_next_input(uid=uid, agent_slug=agent_slug, thread_id=thread_id):
                 failures += 1
         except Exception:
             failures += 1
-            logger.exception("Failed to recover ready AgentInput: %s", thread_id)
+            logger.exception("Failed to recover ready AgentInput: {}", thread_id)
 
     if failures:
         raise RuntimeError(f"待执行任务补投失败：{failures} 条")

backend/yuxi/modules/agents/services/subscriptions.py

@@ -142,7 +142,7 @@ async def stream_thread_events(
                     events = await list_run_stream_events(run.id, after_seq=cursor.event_seq)
                     oldest = await list_run_stream_events(run.id, after_seq="0-0", limit=1)
                 except Exception as exc:
-                    logger.warning("读取 Run 增量失败: run=%s error=%s", run.id, exc)
+                    logger.warning("读取 Run 增量失败: run={} error={}", run.id, exc)
                     events, oldest = [], []
                 expired = (cursor.event_seq != "0-0" and (not oldest or _redis_id(oldest[0]["seq"]) > _redis_id(cursor.event_seq))) or (
                     run.status in TERMINAL_RUN_STATUSES and not events and not oldest