fix: unified logging probe before file handler attach

src/jiangchang_skill_core/unified_logging.py

@@ -10,13 +10,16 @@ import os
 import sys
 import uuid
 from logging.handlers import TimedRotatingFileHandler
-from typing import Optional
+from typing import Optional, Tuple
 
 from .runtime_env import get_data_root, get_user_id
 
 _skill_slug: str = ""
 _logger_name: str = ""
 
+# 避免 logging 内部错误再向 stderr 打印 "--- Logging error ---"
+logging.raiseExceptions = False
+
 
 def get_unified_logs_dir() -> str:
     path = os.path.join(get_data_root(), get_user_id(), "logs")
@@ -41,6 +44,19 @@ def get_skill_log_file_path() -> str:
     return os.path.join(get_unified_logs_dir(), "jiangchang.log")
 
 
+def _can_open_log_file(log_path: str) -> Tuple[bool, str]:
+    """探测日志文件是否可追加写入（真实 open，非 os.access）。"""
+    try:
+        parent = os.path.dirname(os.path.abspath(log_path))
+        if parent:
+            os.makedirs(parent, exist_ok=True)
+        with open(log_path, "a", encoding="utf-8"):
+            pass
+        return True, ""
+    except Exception as exc:
+        return False, str(exc)
+
+
 def ensure_trace_for_process() -> str:
     """本进程调用链 ID：沿用 JIANGCHANG_TRACE_ID，否则生成并写入环境（子进程可继承）。"""
     existing = (os.getenv("JIANGCHANG_TRACE_ID") or "").strip()
@@ -87,8 +103,17 @@ def _backup_count() -> int:
         return 30
 
 
-def _attach_fallback_handler(log: logging.Logger, fmt: logging.Formatter) -> None:
+def _clear_logger_handlers(log: logging.Logger) -> None:
-    """文件日志不可用时仅挂内存 handler，不向 stderr 打印 logging 异常。"""
+    for h in list(log.handlers):
+        log.removeHandler(h)
+        try:
+            h.close()
+        except Exception:
+            pass
+
+
+def _attach_fallback_handler(log: logging.Logger) -> None:
+    """文件日志不可用时仅挂 NullHandler，不向 stderr 打印 logging 异常。"""
     log.addHandler(logging.NullHandler())
 
 
@@ -96,37 +121,39 @@ def setup_skill_logging(skill_slug: str, logger_name: str) -> None:
     """
     幂等：为指定 logger 挂载统一文件日志与可选 stderr。
     须在进程早期调用（如 CLI main 在业务日志之前）。
-    文件日志目录无权限或不可写时降级为 no-op，不影响 CLI 返回码。
+    文件日志不可写时降级为 NullHandler，不影响 CLI 返回码。
     """
     global _skill_slug, _logger_name
+    logging.raiseExceptions = False
     ensure_trace_for_process()
     _skill_slug = skill_slug
     _logger_name = logger_name
 
     log = logging.getLogger(logger_name)
-    if log.handlers:
+    _clear_logger_handlers(log)
-        return
     log.setLevel(_log_level_from_env())
     fmt = logging.Formatter(_FORMAT, datefmt=_DATEFMT)
     filt = _SkillContextFilter()
 
-    try:
+    log_path = get_skill_log_file_path()
-        path = get_skill_log_file_path()
+    ok, _err = _can_open_log_file(log_path)
-        fh = TimedRotatingFileHandler(
+    if ok:
-            path,
+        try:
-            when="midnight",
+            fh = TimedRotatingFileHandler(
-            interval=1,
+                log_path,
-            backupCount=_backup_count(),
+                when="midnight",
-            encoding="utf-8",
+                interval=1,
-            delay=True,
+                backupCount=_backup_count(),
-        )
+                encoding="utf-8",
-        fh.setFormatter(fmt)
+                delay=False,
-        fh.addFilter(filt)
+            )
-        log.addHandler(fh)
+            fh.setFormatter(fmt)
-    except (OSError, PermissionError, ValueError):
+            fh.addFilter(filt)
-        _attach_fallback_handler(log, fmt)
+            log.addHandler(fh)
-    except Exception:
+        except Exception:
-        _attach_fallback_handler(log, fmt)
+            _attach_fallback_handler(log)
+    else:
+        _attach_fallback_handler(log)
 
     if (os.getenv("JIANGCHANG_LOG_TO_STDERR") or "").strip().lower() in (
         "1",
@@ -162,23 +189,25 @@ def attach_unified_file_handler(
     独立子进程内追加同一日志文件（如 login 检测子进程），格式与主进程一致。
     """
     global _skill_slug
+    logging.raiseExceptions = False
     ensure_trace_for_process()
     _skill_slug = skill_slug
     lg = logging.getLogger(logger_name)
-    lg.handlers.clear()
+    _clear_logger_handlers(lg)
     lg.setLevel(level)
     fmt = logging.Formatter(_FORMAT, datefmt=_DATEFMT)
-    try:
+    filt = _SkillContextFilter()
-        parent = os.path.dirname(os.path.abspath(log_path))
+
-        if parent:
+    ok, _err = _can_open_log_file(log_path)
-            os.makedirs(parent, exist_ok=True)
+    if ok:
-        fh = logging.FileHandler(log_path, encoding="utf-8")
+        try:
-        fh.setFormatter(fmt)
+            fh = logging.FileHandler(log_path, encoding="utf-8", delay=False)
-        fh.addFilter(_SkillContextFilter())
+            fh.setFormatter(fmt)
-        lg.addHandler(fh)
+            fh.addFilter(filt)
-    except (OSError, PermissionError, ValueError):
+            lg.addHandler(fh)
-        lg.addHandler(logging.NullHandler())
+        except Exception:
-    except Exception:
+            _attach_fallback_handler(lg)
-        lg.addHandler(logging.NullHandler())
+    else:
+        _attach_fallback_handler(lg)
     lg.propagate = False
     return lg

tests/test_unified_logging.py

@@ -0,0 +1,64 @@
+# -*- coding: utf-8 -*-
+"""统一日志：文件不可写时降级，首次 emit 不抛 PermissionError。"""
+from __future__ import annotations
+
+import logging
+import os
+import tempfile
+import unittest
+from logging.handlers import TimedRotatingFileHandler
+from unittest.mock import patch
+
+from jiangchang_skill_core import unified_logging as ul
+
+
+class TestUnifiedLogging(unittest.TestCase):
+    def setUp(self) -> None:
+        ul._skill_slug = ""
+        ul._logger_name = ""
+
+    def test_can_open_log_file_writable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "test.log")
+            ok, err = ul._can_open_log_file(path)
+            self.assertTrue(ok)
+            self.assertEqual(err, "")
+            self.assertTrue(os.path.isfile(path))
+
+    def test_setup_skill_logging_writable(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            log_path = os.path.join(tmp, "skill.log")
+            logger_name = "test.logger.writable"
+            with patch.object(ul, "get_skill_log_file_path", return_value=log_path):
+                ul.setup_skill_logging("receive-order", logger_name)
+                ul.get_skill_logger().info("hello")
+                self.assertTrue(os.path.isfile(log_path))
+                ul._clear_logger_handlers(logging.getLogger(logger_name))
+
+    def test_setup_skill_logging_unwritable_no_file_handler(self) -> None:
+        logger_name = "test.logger.unwritable"
+        with patch.object(ul, "_can_open_log_file", return_value=(False, "permission denied")):
+            ul.setup_skill_logging("receive-order", logger_name)
+            lg = logging.getLogger(logger_name)
+            lg.info("hello")
+        handlers = logging.getLogger(logger_name).handlers
+        self.assertFalse(any(isinstance(h, TimedRotatingFileHandler) for h in handlers))
+        self.assertTrue(any(isinstance(h, logging.NullHandler) for h in handlers))
+
+    def test_attach_unified_file_handler_unwritable(self) -> None:
+        logger_name = "test.logger.attach.unwritable"
+        with patch.object(ul, "_can_open_log_file", return_value=(False, "permission denied")):
+            lg = ul.attach_unified_file_handler(
+                "/no/access/test.log",
+                skill_slug="receive-order",
+                logger_name=logger_name,
+            )
+            lg.info("hello")
+        self.assertFalse(
+            any(isinstance(h, logging.FileHandler) for h in lg.handlers)
+        )
+        self.assertTrue(any(isinstance(h, logging.NullHandler) for h in lg.handlers))
+
+
+if __name__ == "__main__":
+    unittest.main()