feat(audit): skip audit log for excluded health/root paths

健康检查与根路径（/health/live、/health/ready、/api/v1/health、
/、/health/storage）没有用户、没业务动作，每秒被 K8s/LB
探针刷一次只会灌进无意义噪音。命中排除集即跳过审计行；
诊断日志（method/path/status/ms 走 stderr）照常打，对容器
运维排错仍有用。

* settings.audit_excluded_paths: list[str] 默认覆盖 5 条
  基础设施路径，env AUDIT_EXCLUDED_PATHS 用逗号分隔
  （pydantic NoDecode + field_validator 兼容 str/list）
* main.py 模块级 _AUDIT_EXCLUDED = frozenset(...)，
  access_log 的 success/exception 两条审计行各加守卫
  诊断无条件打
* 测试用 _AccessLogReplica 复刻 access_log 契约（不 import
  真实 main.py），新增 4 个 case：排除根路径、排除 /health/live、
  不排除路径照写审计、自定义排除集

顺带 schedule 模块：ExecutionResult 与 context 已迁到
schedule.domain.*（execution.py / orchestrator.py /
scheduler.py / worker.py），调用点跟进；schedule 自身
18 个测试在改前改后均通过。

.env.example

@@ -36,6 +36,11 @@ LOG_LEVEL=INFO
 # container). AUDIT_LOG_RETENTION_DAYS=0 disables cleanup of old files.
 AUDIT_LOG_DIR=
 AUDIT_LOG_RETENTION_DAYS=30
+# Exact paths excluded from the audit line (health/root probes carry no
+# business value but fire every second from K8s/LB). The default already
+# covers /health/live /health/ready /api/v1/health / /health/storage;
+# leave empty to keep the default. Comma-separated, e.g. /health/live,/api/v1/health.
+AUDIT_EXCLUDED_PATHS=
 
 # Object storage. Two modes are supported:
 #   STORAGE_BACKEND=s3    — connects to an S3-compatible service (MinIO,

backend/src/backend/main.py

@@ -53,6 +53,12 @@ configure_audit_logging(
 )
 
 
+# 审计排除的精确路径集（不含 query）：命中即跳过审计行，诊断日志照常打。
+# 健康检查 / 根路径探针每秒刷审计文件但无业务价值；可用
+# settings.audit_excluded_paths / AUDIT_EXCLUDED_PATHS 覆盖默认值。
+_AUDIT_EXCLUDED: frozenset[str] = frozenset(settings.audit_excluded_paths)
+
+
 @asynccontextmanager
 async def lifespan(app: Any) -> AsyncIterator[None]:
     # 生命周期内创建的对象挂在 app.state 上，路由通过 Depends 或 Request
@@ -137,7 +143,9 @@ async def access_log(request: Request, call_next):
     # 诊断：方法/路径/状态码/耗时 走 stderr（loguru default sink）
     # 合规：时间/用户/方法/路径/状态码 走独立 audit 文件 sink
     # 两条 logger.info() 共用一个出口，便于排查
+    # 排除集是精确路径匹配（不含 query），命中即跳过审计行；诊断日志照常。
     start = time.perf_counter()
+    skip_audit = request.url.path in _AUDIT_EXCLUDED
     try:
         response = await call_next(request)
     except Exception:
@@ -147,12 +155,13 @@ async def access_log(request: Request, call_next):
             method=request.method, path=request.url.path, ms=elapsed_ms,
         )
         # 异常路径：审计行也要写（status=500 由 unhandled_exception_handler 返回）
-        logger.bind(
-            user_id=_audit_user_id(request),
-            method=request.method,
-            path=request.url.path,
-            status=500,
-        ).info("audit")
+        if not skip_audit:
+            logger.bind(
+                user_id=_audit_user_id(request),
+                method=request.method,
+                path=request.url.path,
+                status=500,
+            ).info("audit")
         raise
     elapsed_ms = (time.perf_counter() - start) * 1000
     logger.info(
@@ -160,12 +169,13 @@ async def access_log(request: Request, call_next):
         method=request.method, path=request.url.path,
         status=response.status_code, ms=elapsed_ms,
     )
-    logger.bind(
-        user_id=_audit_user_id(request),
-        method=request.method,
-        path=request.url.path,
-        status=response.status_code,
-    ).info("audit")
+    if not skip_audit:
+        logger.bind(
+            user_id=_audit_user_id(request),
+            method=request.method,
+            path=request.url.path,
+            status=response.status_code,
+        ).info("audit")
     return response
 
 

backend/tests/test_audit_logging.py

@@ -53,29 +53,38 @@ class _AccessLogReplica(BaseHTTPMiddleware):
     """复制 main.py access_log 写审计行的契约（不 import 真实 main.py）。
 
     测试只覆盖 access_log 的审计行为（success / exception 两条路径 +
-    user_id 解析），避免触发 main.py 的 lifespan（MySQL / 路由初始化）。
-    未来 main.py 改 access_log 字段时，这里同步改即可，测试不会假阳/假阴。
+    user_id 解析 + 排除集守卫），避免触发 main.py 的 lifespan（MySQL /
+    路由初始化）。``excluded_paths`` 对应 main.py 的 ``_AUDIT_EXCLUDED``：
+    精确路径命中即跳过审计行（诊断 stderr 照常）。未来 main.py 改
+    access_log 字段时，这里同步改即可，测试不会假阳/假阴。
     """
 
+    def __init__(self, app, excluded_paths: frozenset[str] = frozenset()):
+        super().__init__(app)
+        self.excluded_paths = excluded_paths
+
     async def dispatch(
         self, request: Request, call_next: Callable[..., Awaitable[Response]]
     ) -> Response:
+        skip_audit = request.url.path in self.excluded_paths
         try:
             response = await call_next(request)
         except Exception:
+            if not skip_audit:
+                logger.bind(
+                    user_id=_audit_user_id(request),
+                    method=request.method,
+                    path=request.url.path,
+                    status=500,
+                ).info("audit")
+            raise
+        if not skip_audit:
             logger.bind(
                 user_id=_audit_user_id(request),
                 method=request.method,
                 path=request.url.path,
-                status=500,
+                status=response.status_code,
             ).info("audit")
-            raise
-        logger.bind(
-            user_id=_audit_user_id(request),
-            method=request.method,
-            path=request.url.path,
-            status=response.status_code,
-        ).info("audit")
         return response
 
 
@@ -95,12 +104,17 @@ def _reset_audit_logging():
 
 
 def _build_client(
-    log_dir: str, *, raise_server_exceptions: bool = True
+    log_dir: str,
+    *,
+    raise_server_exceptions: bool = True,
+    excluded_paths: frozenset[str] = frozenset(),
 ) -> TestClient:
     """配置审计日志并返回挂上 access_log 契约复刻中间件的测试 app 客户端。
 
-    只注册空路由，不碰数据库与业务逻辑；中间件在路由之后注册，与 main.py
-    的 access_log 行为一致。``/boom`` 用于验证 exception 路径（500）。
+    只注册少量测试路由，不碰数据库与业务逻辑；中间件在路由之后注册，与
+    main.py 的 access_log 行为一致。``/boom`` 用于验证 exception 路径
+    （500）；``/health/live`` ``/api/v1/scripts`` ``/custom`` ``/other``
+    ``/`` 供排除集用例使用。
     """
     audit.configure_audit_logging(log_dir, retention_days=30)
 
@@ -114,7 +128,27 @@ def _build_client(
     def boom() -> dict:
         raise RuntimeError("boom")
 
-    app.add_middleware(_AccessLogReplica)
+    @app.get("/health/live")
+    def health_live() -> dict:
+        return {"ok": True}
+
+    @app.get("/api/v1/scripts")
+    def scripts() -> dict:
+        return {"ok": True}
+
+    @app.get("/custom")
+    def custom() -> dict:
+        return {"ok": True}
+
+    @app.get("/other")
+    def other() -> dict:
+        return {"ok": True}
+
+    @app.get("/")
+    def root() -> dict:
+        return {"ok": True}
+
+    app.add_middleware(_AccessLogReplica, excluded_paths=excluded_paths)
     return TestClient(app, raise_server_exceptions=raise_server_exceptions)
 
 
@@ -206,6 +240,61 @@ def test_audit_log_bearer_header_resolves_user(tmp_path: Path) -> None:
     assert f"| {user_id} | GET /x -> 200" in lines[0]
 
 
+def test_access_log_skips_audit_for_excluded_path(tmp_path: Path) -> None:
+    """命中排除集（/health/live）：审计行不写，诊断 stderr 照常。
+
+    stderr 走 loguru default sink，难以用 caplog 抓取，这里直接断言
+    审计文件无新行即可。
+    """
+    client = _build_client(
+        str(tmp_path), excluded_paths=frozenset({"/health/live"})
+    )
+    response = client.get("/health/live")
+    assert response.status_code == 200
+
+    assert _audit_lines(tmp_path) == []
+
+
+def test_access_log_skips_audit_for_root_path(tmp_path: Path) -> None:
+    """根路径 `/` 命中排除集：审计行不写。"""
+    client = _build_client(str(tmp_path), excluded_paths=frozenset({"/"}))
+    response = client.get("/")
+    assert response.status_code == 200
+
+    assert _audit_lines(tmp_path) == []
+
+
+def test_access_log_still_writes_audit_for_non_excluded_path(
+    tmp_path: Path,
+) -> None:
+    """未排除路径（/api/v1/scripts）照常写审计行。"""
+    client = _build_client(
+        str(tmp_path), excluded_paths=frozenset({"/health/live", "/"})
+    )
+    response = client.get("/api/v1/scripts")
+    assert response.status_code == 200
+
+    lines = _audit_lines(tmp_path)
+    assert lines
+    assert "GET /api/v1/scripts -> 200" in lines[0]
+
+
+def test_access_log_custom_excluded_paths_from_settings(tmp_path: Path) -> None:
+    """自定义排除集：/custom 命中跳过、/other 未命中照常写。"""
+    client = _build_client(
+        str(tmp_path), excluded_paths=frozenset({"/custom"})
+    )
+    response = client.get("/custom")
+    assert response.status_code == 200
+    assert _audit_lines(tmp_path) == []
+
+    response = client.get("/other")
+    assert response.status_code == 200
+    lines = _audit_lines(tmp_path)
+    assert lines
+    assert "GET /other -> 200" in lines[0]
+
+
 def test_retention_cleanup_removes_old_files(tmp_path: Path) -> None:
     old = tmp_path / "audit-2024-01-01.log"
     recent = tmp_path / "audit-2024-06-01.log"

common/src/common/config.py

@@ -18,9 +18,10 @@ Rules for adding a new variable:
 from __future__ import annotations
 
 from functools import lru_cache
+from typing import Annotated
 
-from pydantic import Field
-from pydantic_settings import BaseSettings, SettingsConfigDict
+from pydantic import Field, field_validator
+from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
 
 
 class Settings(BaseSettings):
@@ -79,6 +80,29 @@ class Settings(BaseSettings):
         default=30,
         description="审计日志保留天数；过期文件启动时清理。设 0 关闭清理。",
     )
+    audit_excluded_paths: Annotated[list[str], NoDecode] = Field(
+        default=[
+            "/health/live",
+            "/health/ready",
+            "/api/v1/health",
+            "/",
+            "/health/storage",
+        ],
+        description=(
+            "审计排除的精确路径列表（不含 query）。命中即不写审计行。"
+            "诊断日志（method/path/status/ms）仍写 stderr。"
+            "环境变量 AUDIT_EXCLUDED_PATHS 用逗号分隔，例如"
+            " '/health/live,/api/v1/health'。"
+        ),
+    )
+
+    @field_validator("audit_excluded_paths", mode="before")
+    @classmethod
+    def _split_audit_paths(cls, v):
+        # env 进来是 "a,b,c"；代码里直接传 list 也行
+        if isinstance(v, str):
+            return [s.strip() for s in v.split(",") if s.strip()]
+        return v
 
     # ── runtime container endpoint ───────────────────────────────
     runtime_api_url: str = Field(

schedule/src/schedule/execution.py

@@ -4,10 +4,10 @@ import asyncio
 import json
 import sys
 import tempfile
-from dataclasses import dataclass
 from pathlib import Path, PurePosixPath
 
 from loguru import logger
+from schedule.domain.execution import ExecutionResult
 
 MAX_LOG_BYTES = 4 * 1024 * 1024
 
@@ -26,18 +26,6 @@ def _resolve_python(version: str) -> str:
         raise ValueError(f"unsupported python_version: {version}")
 
 
-@dataclass(frozen=True)
-class ExecutionResult:
-    status: str
-    exit_code: int | None
-    logs: bytes
-    result: bytes
-    result_file_name: str
-    result_content_type: str
-    error_code: str | None = None
-    error_message: str | None = None
-
-
 def _limited_log(value: str) -> bytes:
     encoded = value.encode("utf-8", errors="replace")
     if len(encoded) <= MAX_LOG_BYTES:

schedule/src/schedule/orchestrator.py

@@ -30,7 +30,7 @@ from loguru import logger
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
-from schedule.context import (
+from schedule.domain.context import (
     FAILED_NODE_STATES,
     TERMINAL_NODE_STATES,
     TERMINAL_RUN_STATES,

schedule/src/schedule/scheduler.py

@@ -28,7 +28,7 @@ from loguru import logger
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
-from schedule.context import naive_utc
+from schedule.domain.context import naive_utc
 
 # APScheduler's persistent SQLAlchemy job store pickles each job.  A bound
 # ``CronScheduler`` method captures this instance (including SQLAlchemy engine

schedule/src/schedule/worker.py

@@ -41,8 +41,9 @@ from common.scheduler.trigger import SYSTEM_CRON_USER_ID
 from loguru import logger
 from sqlalchemy import select
 
-from schedule.context import TERMINAL_NODE_STATES
-from schedule.execution import ExecutionResult, execute_artifact
+from schedule.domain.context import TERMINAL_NODE_STATES
+from schedule.domain.execution import ExecutionResult
+from schedule.execution import execute_artifact
 
 NODE_EXECUTE_EVENT = schedule_event_type("job.node.execute")
 NODE_FINISHED_EVENT = schedule_event_type("job.node.finished")