diff --git a/backend/alembic/versions/j1k2l3m4n5o6_add_audit_logs.py b/backend/alembic/versions/j1k2l3m4n5o6_add_audit_logs.py
new file mode 100644
index 0000000..c8522a6
--- /dev/null
+++ b/backend/alembic/versions/j1k2l3m4n5o6_add_audit_logs.py
@@ -0,0 +1,81 @@
+"""add_audit_logs
+
+Revision ID: j1k2l3m4n5o6
+Revises: i1j2k3l4m5n6
+Create Date: 2026-09-21
+
+操作审计日志表（audit_logs）
+--------------------------
+新增一张独立的审计表，用于记录 task_logs 覆盖不到的操作：
+登录、导出、产品增删改、收口、权限/配置变更等与单个任务无关的动作。
+
+为什么另起一张表而不复用 task_logs：
+task_logs.task_id 是 NOT NULL 外键，只能挂在任务上，无法表达「张三导出了
+产品清单」这类动作；且缺少来源 IP / UA / 结果状态等审计必需字段。
+
+存量数据无需回填（本表从上线时刻开始记录）。
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+revision: str = "j1k2l3m4n5o6"
+down_revision: Union[str, None] = "i1j2k3l4m5n6"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "audit_logs",
+        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, nullable=False),
+        # 操作人
+        sa.Column("user_id", sa.String(64), nullable=True, comment="操作人账号(逻辑外键→MOM)"),
+        sa.Column("display_name", sa.String(100), nullable=True, comment="操作人显示名"),
+        sa.Column("role", sa.String(50), nullable=True, comment="操作时角色快照"),
+        # 业务语义
+        sa.Column("action", sa.String(50), nullable=False, comment="动作"),
+        sa.Column("module", sa.String(50), nullable=False, comment="业务模块"),
+        sa.Column("target_type", sa.String(50), nullable=True),
+        sa.Column("target_id", sa.String(100), nullable=True),
+        sa.Column("target_name", sa.String(200), nullable=True),
+        sa.Column("details", postgresql.JSONB(), nullable=True, comment="变更详情"),
+        # 请求上下文
+        sa.Column("ip_address", sa.String(50), nullable=True),
+        sa.Column("user_agent", sa.String(500), nullable=True),
+        sa.Column("method", sa.String(10), nullable=True),
+        sa.Column("url", sa.String(500), nullable=True),
+        sa.Column("status_code", sa.Integer(), nullable=True),
+        sa.Column("error_message", sa.Text(), nullable=True),
+        # 与结构化日志对账
+        sa.Column("request_id", sa.String(64), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.func.now(),
+        ),
+    )
+
+    # 索引：按审计页最常用的检索维度建
+    op.create_index("ix_audit_logs_created_at", "audit_logs", ["created_at"])
+    op.create_index("ix_audit_logs_user_id", "audit_logs", ["user_id"])
+    op.create_index("ix_audit_logs_module", "audit_logs", ["module"])
+    op.create_index("ix_audit_logs_action", "audit_logs", ["action"])
+    op.create_index("ix_audit_logs_target_id", "audit_logs", ["target_id"])
+    op.create_index("ix_audit_logs_request_id", "audit_logs", ["request_id"])
+    # 组合索引：审计页默认「按时间倒序 + 按模块/动作过滤」
+    op.create_index("ix_audit_logs_module_created", "audit_logs", ["module", "created_at"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_audit_logs_module_created", table_name="audit_logs")
+    op.drop_index("ix_audit_logs_request_id", table_name="audit_logs")
+    op.drop_index("ix_audit_logs_target_id", table_name="audit_logs")
+    op.drop_index("ix_audit_logs_action", table_name="audit_logs")
+    op.drop_index("ix_audit_logs_module", table_name="audit_logs")
+    op.drop_index("ix_audit_logs_user_id", table_name="audit_logs")
+    op.drop_index("ix_audit_logs_created_at", table_name="audit_logs")
+    op.drop_table("audit_logs")
diff --git a/backend/app/api/v1/endpoints/audit.py b/backend/app/api/v1/endpoints/audit.py
new file mode 100644
index 0000000..939cda2
--- /dev/null
+++ b/backend/app/api/v1/endpoints/audit.py
@@ -0,0 +1,99 @@
+"""审计日志 API —— 查看系统操作审计记录
+
+与 MOM(KCGL) /audit/logs 的接口保持同构的筛选维度（操作人/模块/动作/目标/
+时间区间），便于两端运维习惯统一；额外提供 request_id 筛选，可凭它直接跳到
+结构化日志里的那一次请求。
+"""
+from __future__ import annotations
+
+from datetime import datetime, time, timedelta
+
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.database import get_db
+from app.core.deps import require_admin
+from app.core.time_utils import BEIJING_TZ
+from app.schemas.audit import (
+    AuditLogListResponse,
+    AuditLogResponse,
+    AuditOption,
+    AuditOptionsResponse,
+)
+from app.services import audit_service
+from app.services.audit_service import ACTION_LABELS, MODULE_LABELS
+
+router = APIRouter(prefix="/audit", tags=["审计日志"])
+
+
+def _parse_day(value: str | None, *, end_of_day: bool = False) -> datetime | None:
+    """解析 YYYY-MM-DD 为北京时间。
+
+    结束日期取次日 00:00 作为上界（配合 < 判断）—— 直接取当天 23:59:59 会
+    漏掉该秒内的记录，是日期区间筛选最常见的差一错误。
+    """
+    if not value:
+        return None
+    try:
+        day = datetime.strptime(value, "%Y-%m-%d").date()
+    except ValueError:
+        return None
+    if end_of_day:
+        return datetime.combine(day + timedelta(days=1), time.min, tzinfo=BEIJING_TZ)
+    return datetime.combine(day, time.min, tzinfo=BEIJING_TZ)
+
+
+@router.get("/logs", response_model=AuditLogListResponse)
+async def get_audit_logs(
+    user_id: str | None = Query(None, description="操作人账号（模糊匹配）"),
+    module: str | None = Query(None, description="业务模块"),
+    action: str | None = Query(None, description="动作类型"),
+    target_id: str | None = Query(None, description="目标ID"),
+    request_id: str | None = Query(None, description="请求ID（与接口日志对账）"),
+    status_code: int | None = Query(None, description="响应状态码"),
+    start_date: str | None = Query(None, description="起始日期 YYYY-MM-DD"),
+    end_date: str | None = Query(None, description="结束日期 YYYY-MM-DD（含当天）"),
+    page: int = Query(1, ge=1),
+    page_size: int = Query(50, ge=1, le=200),
+    db: AsyncSession = Depends(get_db),
+    current_user: dict = Depends(require_admin),
+) -> AuditLogListResponse:
+    """审计日志分页查询（按时间倒序）"""
+    start = _parse_day(start_date)
+    # 结束日期用「次日 00:00」作为开区间上界，避免漏掉当天最后几条
+    end_exclusive = _parse_day(end_date, end_of_day=True)
+
+    rows, total = await audit_service.list_audit_logs(
+        db,
+        user_id=user_id,
+        module=module,
+        action=action,
+        target_id=target_id,
+        request_id=request_id,
+        status_code=status_code,
+        start=start,
+        end=end_exclusive - timedelta(microseconds=1) if end_exclusive else None,
+        skip=(page - 1) * page_size,
+        limit=page_size,
+    )
+
+    items = []
+    for row in rows:
+        item = AuditLogResponse.model_validate(row)
+        # 中文标签由服务端补，避免前端为每个枚举再维护一份映射
+        item.module_label = MODULE_LABELS.get(row.module, row.module)
+        item.action_label = ACTION_LABELS.get(row.action, row.action)
+        items.append(item)
+
+    return AuditLogListResponse(items=items, total=total)
+
+
+@router.get("/options", response_model=AuditOptionsResponse)
+async def get_audit_options(
+    current_user: dict = Depends(require_admin),
+) -> AuditOptionsResponse:
+    """筛选项：模块与动作的中文下拉"""
+    return AuditOptionsResponse(
+        modules=[AuditOption(value=k, label=v) for k, v in MODULE_LABELS.items()],
+        actions=[AuditOption(value=k, label=v) for k, v in ACTION_LABELS.items()],
+    )
diff --git a/backend/app/api/v1/endpoints/auth.py b/backend/app/api/v1/endpoints/auth.py
index c6308c3..d40d7bb 100644
--- a/backend/app/api/v1/endpoints/auth.py
+++ b/backend/app/api/v1/endpoints/auth.py
@@ -1,5 +1,5 @@
 """认证 API — 对接 MOM sys_user + 双 Token 刷新"""
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, Request
 from app.schemas.user import (
     LoginRequest,
     LoginResponse,
@@ -13,8 +13,13 @@ router = APIRouter(prefix="/auth", tags=["认证"])
 
 
 @router.post("/login", response_model=LoginResponse)
-def login_endpoint(data: LoginRequest):
+def login_endpoint(data: LoginRequest, request: Request):
     """登录 — 验证 MOM sys_user 表，返回 Access + Refresh 双 Token"""
+    # 登录请求本身尚未认证，中间件拿不到操作人。但「谁在尝试登录、失败了多少次」
+    # 恰恰是审计里最该有的信息，所以在校验之前就把尝试的账号写进 state：
+    # 登录失败时同样留痕，且能按账号追踪暴力破解。
+    # 注意：绝不把 data.password 写进 state / 审计，密码不落库。
+    request.state.audit_user = data.username
     return login(data.username, data.password)
 
 
diff --git a/backend/app/api/v1/router.py b/backend/app/api/v1/router.py
index da9efc9..9134603 100644
--- a/backend/app/api/v1/router.py
+++ b/backend/app/api/v1/router.py
@@ -17,6 +17,7 @@ from app.api.v1.endpoints.holidays import router as holidays_router
 from app.api.v1.endpoints.webhooks import router as webhooks_router
 from app.api.v1.endpoints.external_products import router as external_products_router
 from app.api.v1.endpoints.screen import router as screen_router
+from app.api.v1.endpoints.audit import router as audit_router
 
 api_router = APIRouter()
 
@@ -37,3 +38,4 @@ api_router.include_router(holidays_router)
 api_router.include_router(webhooks_router)
 api_router.include_router(external_products_router)
 api_router.include_router(screen_router)
+api_router.include_router(audit_router)
diff --git a/backend/app/core/audit_middleware.py b/backend/app/core/audit_middleware.py
new file mode 100644
index 0000000..cfa8969
--- /dev/null
+++ b/backend/app/core/audit_middleware.py
@@ -0,0 +1,195 @@
+"""审计采集中间件
+
+在响应生成后，把「谁 / 何时 / 从哪来 / 调了哪个接口 / 做了什么 / 结果如何」
+落进 audit_logs。
+
+为什么用中间件自动采集，而不是在每个业务函数里手写 record_audit
+------------------------------------------------------------------
+1. 手写必然漏。新加的端点很容易忘记补审计，而审计的价值恰恰建立在「完整」上。
+   现状可佐证：task_logs 全项目只有 4 处写入点，凡是不挂在任务上的动作
+   （登录、导出、改产品）全都没有留痕。
+2. 中间件能拿到业务函数拿不到的事实：真实来源 IP、UA、最终状态码、
+   以及与结构化日志对齐的 request_id。
+3. 业务语义（module / target）由路径推导，不如手写精确，但对「谁动了什么」
+   的追责场景已经够用；关键动作后续可再调 record_audit 补 details 做增强。
+
+采集范围
+--------
+- 所有写操作（POST/PUT/PATCH/DELETE）
+- 少数**读但敏感**的操作：导出、下载、打印（本项目 GET /people-history/export
+  就是导出，只按方法过滤会漏掉）
+
+明确不采集：GET /health*、/docs、/openapi.json —— 探针与文档的噪声没有审计价值。
+"""
+from __future__ import annotations
+
+import logging
+
+from fastapi import Request
+from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
+from starlette.responses import Response
+
+from app.services.audit_service import record_audit
+
+logger = logging.getLogger("track.audit")
+
+# 写操作一律采集
+_MUTATING_METHODS = frozenset({"POST", "PUT", "PATCH", "DELETE"})
+
+# 读操作里需要留痕的（导出/下载/打印属于「读」，但把数据带出了系统）
+_SENSITIVE_READ_KEYWORDS = frozenset({"export", "download", "print"})
+
+# 永久忽略的路径前缀
+_IGNORED_PREFIXES = ("/health", "/docs", "/redoc", "/openapi.json")
+
+# 路径段 → 审计模块
+_PATH_MODULE: dict[str, str] = {
+    "products": "product",
+    "tasks": "task",
+    "orders": "order",
+    "records": "record",
+    "print": "print",
+    "materials": "material",
+    "users": "user",
+    "upload": "upload",
+    "notifications": "notification",
+    "app-version": "app",
+    "analytics": "analytics",
+    "dashboard": "dashboard",
+    "holidays": "holiday",
+    "screen": "screen",
+    "webhooks": "external",
+    "external": "external",
+    "audit": "audit",
+    "auth": "auth",
+}
+
+# 路径段 → 动作（优先于按 HTTP 方法推断）
+_SEGMENT_ACTION: dict[str, str] = {
+    "login": "login",
+    "logout": "logout",
+    "refresh": "refresh",
+    "export": "export",
+    "download": "export",
+    "print": "print",
+    "upload": "upload",
+    "finalize": "finalize",
+    "receive": "receive",
+    "transfer": "transfer",
+    "reject": "reject",
+    "recall": "recall",
+    "spawn": "spawn",
+    "complete": "complete",
+    "end": "end",
+}
+
+_METHOD_ACTION: dict[str, str] = {
+    "POST": "create",
+    "PUT": "update",
+    "PATCH": "update",
+    "DELETE": "delete",
+    "GET": "read",
+}
+
+# 不可能是业务 ID 的路径段，避免把动作词误当成 target_id
+_NON_ID_SEGMENTS = frozenset(
+    set(_SEGMENT_ACTION) | {"api", "v1", "me", "options", "export", "lookup", "batch"}
+)
+
+
+def _derive_module_and_action(path: str, method: str) -> tuple[str, str, str | None]:
+    """由请求路径与 HTTP 方法推导 (module, action, target_id)"""
+    parts = [p for p in path.split("/") if p]
+
+    module = "other"
+    module_idx = -1
+    for i, seg in enumerate(parts):
+        if seg in _PATH_MODULE:
+            module = _PATH_MODULE[seg]
+            module_idx = i
+            break
+
+    action = None
+    for seg in reversed(parts):
+        if seg in _SEGMENT_ACTION:
+            action = _SEGMENT_ACTION[seg]
+            break
+    if action is None:
+        action = _METHOD_ACTION.get(method, method.lower())
+
+    target_id = None
+    if module_idx >= 0 and module_idx + 1 < len(parts):
+        candidate = parts[module_idx + 1]
+        if candidate not in _NON_ID_SEGMENTS:
+            target_id = candidate
+
+    return module, action, target_id
+
+
+class AuditMiddleware(BaseHTTPMiddleware):
+    """写操作审计采集。
+
+    必须注册在 RequestContextMiddleware **内层**，因为它依赖后者写入
+    request.state 的 request_id 才能与结构化日志对账。
+    """
+
+    def _should_audit(self, request: Request) -> bool:
+        path = request.url.path
+        if path.startswith(_IGNORED_PREFIXES):
+            return False
+        if request.method in _MUTATING_METHODS:
+            return True
+        if request.method == "GET":
+            lowered = path.lower()
+            return any(kw in lowered for kw in _SENSITIVE_READ_KEYWORDS)
+        return False
+
+    async def dispatch(
+        self, request: Request, call_next: RequestResponseEndpoint
+    ) -> Response:
+        if not self._should_audit(request):
+            return await call_next(request)
+
+        status_code = 500
+        error_message: str | None = None
+        try:
+            response = await call_next(request)
+            status_code = response.status_code
+            return response
+        except Exception as exc:
+            # 异常最终由 ServerErrorMiddleware 转成 500；这里先标记，
+            # 保证「失败的操作也有审计」——这正是选用独立 session 的目的
+            error_message = f"{type(exc).__name__}: {exc}"[:1000]
+            raise
+        finally:
+            await self._write(request, status_code, error_message)
+
+    async def _write(
+        self, request: Request, status_code: int, error_message: str | None
+    ) -> None:
+        try:
+            module, action, target_id = _derive_module_and_action(
+                request.url.path, request.method
+            )
+            client = request.client
+            await record_audit(
+                action=action,
+                module=module,
+                user_id=getattr(request.state, "audit_user", None),
+                display_name=getattr(request.state, "audit_display_name", None),
+                role=getattr(request.state, "audit_role", None),
+                target_type=module,
+                target_id=target_id,
+                # 对产品而言路径里的 ID 就是身份证号，本身即人可读的标识
+                target_name=target_id if module == "product" else None,
+                ip_address=client.host if client else None,
+                user_agent=request.headers.get("user-agent"),
+                method=request.method,
+                url=request.url.path,
+                status_code=status_code,
+                error_message=error_message,
+                request_id=getattr(request.state, "request_id", None),
+            )
+        except Exception:
+            # record_audit 内部已兜底；这里再兜一层，确保审计绝不冒泡成 500
+            logger.exception("审计采集失败（已忽略）")
diff --git a/backend/app/core/deps.py b/backend/app/core/deps.py
new file mode 100644
index 0000000..a532b59
--- /dev/null
+++ b/backend/app/core/deps.py
@@ -0,0 +1,35 @@
+"""通用 FastAPI 依赖"""
+from __future__ import annotations
+
+from fastapi import Depends, HTTPException, status
+
+from app.core.roles import ADMIN_ROLES
+from app.services.auth_service import get_current_user
+
+
+def require_roles(*roles: str):
+    """生成「限定角色」依赖，避免同一个内联判断被复制到每个端点。
+
+    用法::
+
+        @router.get("/x")
+        async def x(current_user: dict = Depends(require_admin)):
+            ...
+
+    失败一律 403 且不透露允许的角色集合（避免给探测者提供线索）。
+    """
+    allowed = frozenset(roles)
+
+    async def _guard(current_user: dict = Depends(get_current_user)) -> dict:
+        if (current_user or {}).get("role") not in allowed:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="当前角色无权访问该接口",
+            )
+        return current_user
+
+    return _guard
+
+
+# 审计日志等高权限接口复用同一实例
+require_admin = require_roles(*ADMIN_ROLES)
diff --git a/backend/app/core/roles.py b/backend/app/core/roles.py
new file mode 100644
index 0000000..cca0ff3
--- /dev/null
+++ b/backend/app/core/roles.py
@@ -0,0 +1,31 @@
+"""角色定义与管理员判定 —— 单一事实来源
+
+背景：角色字符串此前散落在至少三处 —— task_service.ADMIN_ROLES、
+products.py 的内联判断、以及前端 constants/task.ts。同一份规则抄多份的后果
+已经发生过：前端 constants/task.ts:233 的注释记录了一次「移动端只判了
+SUPER_ADMIN、漏了 SUPERVISOR，导致主管被误挡」的事故。
+
+本模块把**角色常量与管理员判定**先收敛到一处，供后端统一引用。
+完整的「角色 × 权限点」可配置矩阵是后续工作；但任何推进都应从这里出发，
+不要再新增第四份副本。
+"""
+from __future__ import annotations
+
+SUPER_ADMIN = "SUPER_ADMIN"
+SUPERVISOR = "SUPERVISOR"
+# 注意：MOM 登录返回的默认角色是小写 operator（见 auth_service.login）
+OPERATOR = "OPERATOR"
+
+# 管理员角色：可执行收口、审计查看等高权限动作
+ADMIN_ROLES: frozenset[str] = frozenset({SUPER_ADMIN, SUPERVISOR})
+
+ROLE_LABELS: dict[str, str] = {
+    SUPER_ADMIN: "超级管理员",
+    SUPERVISOR: "主管",
+    OPERATOR: "操作员",
+}
+
+
+def is_admin(role: str | None) -> bool:
+    """role 为 None / 未知值一律视为无权限（fail-closed，不做兜底放行）"""
+    return role in ADMIN_ROLES
diff --git a/backend/app/main.py b/backend/app/main.py
index f672de9..efbba30 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -4,6 +4,7 @@ from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 from app.core.config import settings
+from app.core.audit_middleware import AuditMiddleware
 from app.core.health import router as health_router
 from app.core.logging import request_id_var, setup_logging
 from app.core.middleware import RequestContextMiddleware
@@ -72,8 +73,11 @@ app.add_middleware(
     expose_headers=["X-Request-ID"],
 )
 
-# Starlette 的 add_middleware 是「后添加者在外层」，故 RequestContextMiddleware
-# 最后注册 → 最先进入请求，才能覆盖包括 CORS 预检在内的全部请求并回写 X-Request-ID
+# Starlette 的 add_middleware 是「后添加者在外层」。执行顺序（由外到内）：
+#   RequestContextMiddleware -> AuditMiddleware -> CORS -> 路由
+# AuditMiddleware 必须在 RequestContext 内层，才能读到后者写入 request.state
+# 的 request_id，从而把审计记录与结构化日志对上。
+app.add_middleware(AuditMiddleware)
 app.add_middleware(RequestContextMiddleware)
 
 
@@ -96,6 +100,10 @@ async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONR
     return JSONResponse(
         status_code=500,
         content={"detail": "服务器内部错误", "request_id": request_id},
+        # 该响应由 ServerErrorMiddleware（位于 RequestContextMiddleware 外层）
+        # 生成，中间件没机会再往响应头写 X-Request-ID，故在此显式补上，
+        # 保证报障时前端从响应头就能拿到可对账的 ID。
+        headers={"X-Request-ID": request_id} if request_id else None,
     )
 
 
diff --git a/backend/app/models/__init__.py b/backend/app/models/__init__.py
index 70e033a..ac5bba5 100644
--- a/backend/app/models/__init__.py
+++ b/backend/app/models/__init__.py
@@ -8,6 +8,7 @@ from app.models.notification import Notification
 from app.models.app_version import AppVersion
 from app.models.message import ProductMessage
 from app.models.holiday import Holiday
+from app.models.audit_log import AuditLog
 __all__ = [
     "Base",
     "ProductionOrder",
@@ -19,4 +20,5 @@ __all__ = [
     "AppVersion",
     "ProductMessage",
     "Holiday",
+    "AuditLog",
 ]
diff --git a/backend/app/models/audit_log.py b/backend/app/models/audit_log.py
new file mode 100644
index 0000000..dc9d7de
--- /dev/null
+++ b/backend/app/models/audit_log.py
@@ -0,0 +1,83 @@
+"""操作审计日志模型
+
+设计参考 MOM(KCGL) 的 audit_logs，但按 Track 的技术栈与诉求做了取舍：
+
+- 主键用 UUID（与库内其它表一致），而非 MOM 的自增 int。
+- 增加 request_id：与 core/logging.py 的结构化日志打通 —— 凭一个 ID 就能把
+  「接口访问日志」和「审计记录」对上，排障时不用再猜。MOM 无此字段。
+- 保留 module / action / target_* 的业务语义，使审计能按业务维度检索，
+  而不是只能按时间翻。
+- 绝不记录请求体：登录等接口 body 含明文密码，一旦落库就成了长期泄露面。
+
+与既有 task_logs 的分工：task_logs 是「任务流转轨迹」（有 task_id 非空约束，
+只能挂在任务上，供流转树渲染）；本表是「操作审计」，覆盖登录、导出、
+产品增删改、权限变更等与单个任务无关的动作，且额外记录来源 IP / UA / 耗时结果。
+"""
+import uuid
+from datetime import datetime
+
+from sqlalchemy import DateTime, Integer, String, Text
+from sqlalchemy.dialects.postgresql import JSONB, UUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.models.base import Base
+from app.core.time_utils import get_beijing_time
+
+
+class AuditLog(Base):
+    __tablename__ = "audit_logs"
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=uuid.uuid4,
+    )
+
+    # ---- 操作人（逻辑外键 → MOM sys_user，仅存账号，无物理约束）----
+    user_id: Mapped[str | None] = mapped_column(
+        String(64), nullable=True, index=True, comment="操作人账号(逻辑外键→MOM)",
+    )
+    display_name: Mapped[str | None] = mapped_column(
+        String(100), nullable=True, comment="操作人显示名",
+    )
+    role: Mapped[str | None] = mapped_column(
+        String(50), nullable=True, comment="操作时角色快照",
+    )
+
+    # ---- 业务语义 ----
+    action: Mapped[str] = mapped_column(
+        String(50), nullable=False, index=True, comment="动作: create/update/delete/export/login/...",
+    )
+    module: Mapped[str] = mapped_column(
+        String(50), nullable=False, index=True, comment="业务模块: product/task/order/auth/print/...",
+    )
+    target_type: Mapped[str | None] = mapped_column(
+        String(50), nullable=True, comment="目标类型(表名或实体名)",
+    )
+    target_id: Mapped[str | None] = mapped_column(
+        String(100), nullable=True, index=True, comment="目标ID",
+    )
+    target_name: Mapped[str | None] = mapped_column(
+        String(200), nullable=True, comment="目标显示名(如产品身份证/工单号)",
+    )
+    details: Mapped[dict | None] = mapped_column(
+        JSONB, nullable=True, comment="变更详情 {old:{}, new:{}}；禁止写入密码等敏感字段",
+    )
+
+    # ---- 请求上下文（由中间件自动填充）----
+    ip_address: Mapped[str | None] = mapped_column(String(50), nullable=True, comment="来源IP")
+    user_agent: Mapped[str | None] = mapped_column(String(500), nullable=True, comment="浏览器UA")
+    method: Mapped[str | None] = mapped_column(String(10), nullable=True, comment="HTTP方法")
+    url: Mapped[str | None] = mapped_column(String(500), nullable=True, comment="请求路径")
+    status_code: Mapped[int | None] = mapped_column(Integer, nullable=True, comment="响应状态码")
+    error_message: Mapped[str | None] = mapped_column(Text, nullable=True, comment="错误信息(如有)")
+
+    # ---- 与结构化日志对账用 ----
+    request_id: Mapped[str | None] = mapped_column(
+        String(64), nullable=True, index=True, comment="关联 core/logging 的 request_id",
+    )
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=get_beijing_time, index=True, comment="操作时间",
+    )
+
+    def __repr__(self) -> str:
+        return f"<AuditLog {self.action} {self.module} by {self.user_id}>"
diff --git a/backend/app/schemas/audit.py b/backend/app/schemas/audit.py
new file mode 100644
index 0000000..0e26eaa
--- /dev/null
+++ b/backend/app/schemas/audit.py
@@ -0,0 +1,57 @@
+"""审计日志 Pydantic Schema"""
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+
+from pydantic import BaseModel
+
+
+class AuditLogResponse(BaseModel):
+    """单条审计记录"""
+    id: uuid.UUID
+    user_id: str | None = None
+    display_name: str | None = None
+    role: str | None = None
+
+    action: str
+    action_label: str | None = None      # 服务端补的中文标签，避免前端各处硬编码
+    module: str
+    module_label: str | None = None
+
+    target_type: str | None = None
+    target_id: str | None = None
+    target_name: str | None = None
+    details: dict | None = None
+
+    ip_address: str | None = None
+    user_agent: str | None = None
+    method: str | None = None
+    url: str | None = None
+    status_code: int | None = None
+    error_message: str | None = None
+
+    # 与结构化日志对账用：拿着它就能捞到对应的接口日志
+    request_id: str | None = None
+
+    created_at: datetime
+
+    model_config = {"from_attributes": True}
+
+
+class AuditLogListResponse(BaseModel):
+    """审计日志分页列表"""
+    items: list[AuditLogResponse]
+    total: int
+
+
+class AuditOption(BaseModel):
+    """筛选项（value/label 结构，直接喂给前端下拉）"""
+    value: str
+    label: str
+
+
+class AuditOptionsResponse(BaseModel):
+    """筛选项集合"""
+    modules: list[AuditOption]
+    actions: list[AuditOption]
diff --git a/backend/app/services/audit_service.py b/backend/app/services/audit_service.py
new file mode 100644
index 0000000..532e17c
--- /dev/null
+++ b/backend/app/services/audit_service.py
@@ -0,0 +1,200 @@
+"""审计服务 — 写入与检索
+
+写入方案的取舍（与 MOM/KCGL 不同，理由如下）
+--------------------------------------------------
+MOM 用 SQLAlchemy event listener + **同事务**写入：优点是全自动、业务代码零改动；
+缺点是业务事务回滚时审计记录一起被回滚掉 —— 而失败/被拒的操作恰恰是最需要
+留痕的（比如越权尝试、参数错误导致的 4xx）。
+
+Track 改为：响应生成后，用**独立 session** 写入审计。
+- 业务回滚不影响审计，失败操作照样留痕
+- 审计写入失败也不影响业务（全包裹 try/except，仅记日志）
+- 代价：审计与业务不是原子提交，极端情况（响应后进程立即被 kill）可能丢一条。
+  对内部系统的操作审计，这个取舍划算。
+"""
+from __future__ import annotations
+
+import logging
+import uuid
+from datetime import datetime
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.database import AsyncSessionLocal
+from app.models.audit_log import AuditLog
+
+logger = logging.getLogger("track.audit")
+
+# 绝不落库的敏感字段名（命中即替换为 ***）
+# 登录请求体含明文密码，一旦进审计表就成了长期泄露面
+_SENSITIVE_KEYS = frozenset(
+    {"password", "passwd", "pwd", "token", "access_token", "refresh_token",
+     "secret", "api_key", "authorization", "password_hash"}
+)
+
+# 模块 / 动作 的中文标签（前端下拉与列表展示用）
+MODULE_LABELS: dict[str, str] = {
+    "auth": "认证登录",
+    "product": "产品管理",
+    "task": "任务流转",
+    "order": "订单管理",
+    "record": "任务记录",
+    "print": "标签打印",
+    "material": "物料",
+    "user": "用户",
+    "notification": "消息通知",
+    "upload": "文件上传",
+    "dashboard": "看板统计",
+    "analytics": "效能分析",
+    "screen": "数据大屏",
+    "holiday": "节假日配置",
+    "app": "App版本",
+    "external": "外部系统对接",
+    "audit": "审计日志",
+    "other": "其它",
+}
+
+ACTION_LABELS: dict[str, str] = {
+    "create": "新增",
+    "update": "修改",
+    "delete": "删除",
+    "read": "查询",
+    "export": "导出",
+    "login": "登录",
+    "logout": "登出",
+    "refresh": "刷新令牌",
+    "print": "打印",
+    "upload": "上传",
+    "finalize": "收口",
+    "receive": "接收",
+    "transfer": "转交",
+    "reject": "驳回",
+    "recall": "撤回",
+    "spawn": "派发",
+    "end": "结束分支",
+    "complete": "完结",
+}
+
+
+def sanitize_details(details: dict | None) -> dict | None:
+    """递归剔除敏感字段，避免密码/令牌落库"""
+    if not details:
+        return details
+
+    def _clean(value):
+        if isinstance(value, dict):
+            return {
+                k: ("***" if str(k).lower() in _SENSITIVE_KEYS else _clean(v))
+                for k, v in value.items()
+            }
+        if isinstance(value, list):
+            return [_clean(v) for v in value]
+        return value
+
+    return _clean(details)
+
+
+async def record_audit(
+    *,
+    action: str,
+    module: str,
+    user_id: str | None = None,
+    display_name: str | None = None,
+    role: str | None = None,
+    target_type: str | None = None,
+    target_id: str | None = None,
+    target_name: str | None = None,
+    details: dict | None = None,
+    ip_address: str | None = None,
+    user_agent: str | None = None,
+    method: str | None = None,
+    url: str | None = None,
+    status_code: int | None = None,
+    error_message: str | None = None,
+    request_id: str | None = None,
+) -> None:
+    """写入一条审计记录。**绝不抛异常**：审计失败不能影响业务。"""
+    try:
+        async with AsyncSessionLocal() as session:
+            session.add(
+                AuditLog(
+                    id=uuid.uuid4(),
+                    user_id=user_id,
+                    display_name=display_name,
+                    role=role,
+                    action=action,
+                    module=module,
+                    target_type=target_type,
+                    target_id=str(target_id) if target_id is not None else None,
+                    target_name=target_name,
+                    details=sanitize_details(details),
+                    ip_address=ip_address,
+                    user_agent=user_agent[:500] if user_agent else None,
+                    method=method,
+                    url=url[:500] if url else None,
+                    status_code=status_code,
+                    error_message=error_message,
+                    request_id=request_id,
+                )
+            )
+            await session.commit()
+    except Exception:
+        # 用 exception 级别但吞掉异常：保证调用方业务流程不受影响
+        logger.exception(
+            "审计写入失败（已忽略，不影响业务）",
+            extra={"extra_fields": {"action": action, "module": module, "url": url}},
+        )
+
+
+async def list_audit_logs(
+    db: AsyncSession,
+    *,
+    user_id: str | None = None,
+    module: str | None = None,
+    action: str | None = None,
+    target_id: str | None = None,
+    request_id: str | None = None,
+    status_code: int | None = None,
+    start: datetime | None = None,
+    end: datetime | None = None,
+    skip: int = 0,
+    limit: int = 50,
+) -> tuple[list[AuditLog], int]:
+    """审计日志检索（按时间倒序）。返回 (当前页, 真实总数)。
+
+    真实总数走独立 COUNT —— 前端分页器依赖它，不能用 len(当前页)。
+    """
+    filters = []
+    if user_id:
+        filters.append(AuditLog.user_id.ilike(f"%{user_id}%"))
+    if module:
+        filters.append(AuditLog.module == module)
+    if action:
+        filters.append(AuditLog.action == action)
+    if target_id:
+        filters.append(AuditLog.target_id == target_id)
+    if request_id:
+        filters.append(AuditLog.request_id == request_id)
+    if status_code is not None:
+        filters.append(AuditLog.status_code == status_code)
+    if start:
+        filters.append(AuditLog.created_at >= start)
+    if end:
+        filters.append(AuditLog.created_at <= end)
+
+    total = await db.scalar(
+        select(func.count()).select_from(AuditLog).where(*filters)
+    ) or 0
+
+    rows = (
+        await db.execute(
+            select(AuditLog)
+            .where(*filters)
+            .order_by(AuditLog.created_at.desc())
+            .offset(skip)
+            .limit(limit)
+        )
+    ).scalars().all()
+
+    return list(rows), total
diff --git a/backend/app/services/task_service.py b/backend/app/services/task_service.py
index a2bf8d1..6da7220 100644
--- a/backend/app/services/task_service.py
+++ b/backend/app/services/task_service.py
@@ -22,6 +22,7 @@ from app.core.lifecycle import (
 )
 from app.models.product import Product
 from app.models.task_log import TaskLog
+from app.core.roles import ADMIN_ROLES
 from app.schemas.task import (
     TaskCreate,
     TaskUpdate,
@@ -42,7 +43,8 @@ from app.schemas.task import (
 VIRTUAL_WAREHOUSE = "virtual_warehouse"
 
 # 管理员/主管角色白名单 — 拥有上帝视角操作权限
-ADMIN_ROLES = {"SUPER_ADMIN", "SUPERVISOR"}
+# 定义已收敛到 app.core.roles（单一事实来源）；本模块继续以同名导出，
+# 兼容 products.py 等处 `from app.services.task_service import ADMIN_ROLES` 的既有引用
 
 
 async def _recalc_product_location(