fix: 空文件 verdict error→pass

console.py upload_multiple_files L1278-1291:
  空文件被硬编码成 verdict='error' ok=False
  根因: 把'没内容'当成'检测异常'
  修复: 空文件是合法文件, 改成 verdict='pass' ok=True reason='空文件，无风险'

README.md

@@ -1,6 +1,6 @@
 # MeshSafe · 文件安全检测服务
 
-![Version](https://img.shields.io/badge/version-v0.2.2-blue)
+![Version](https://img.shields.io/badge/version-v0.2.3-blue)
 ![Python](https://img.shields.io/badge/python-3.12%2B-green)
 ![License](https://img.shields.io/badge/license-proprietary-orange)
 
@@ -200,6 +200,36 @@ MeshSafe 会**分析宏的实际行为**：
 
 ## 版本历史
 
+### v0.2.3 (2026-10-29)
+
+**Bug 修复：**
+
+- **Policy 加载失败（每次启动都回退 YAML）** — `policy.py _load_db_or_yaml()` 用 `scalar_one_or_none()` 加载最新策略版本，但 `policy_versions` 表有多条历史版本 → SQLAlchemy 抛 `Multiple rows` → 永远回退 YAML 配置文件，导致不同重启间规则数波动。修复为 `db.query(PolicyVersion).order_by(created_at.desc()).first()`，直接取最新一条 ORM 对象
+- **控制台任务历史被内部测试记录淹没** — `scan_tasks` 表累积 5,800+ 条内部对账/诊断脚本产生的测试数据（channel 包括 `test_center` / `v` / `verify` / `diag` / `debug` / `final` 等 19 个），dashboard 和 tasks_page 无过滤全部展示。修复：① DB 清理所有内部 channel 记录（保留 3,488 条生产数据）；② console.py 加 `INTERNAL_TEST_CHANNELS` 黑名单常量，dashboard 6 个查询 + tasks_page 2 个查询全部 `.filter(ScanTask.channel.notin_(...))`
+- **测试靶场显示 "未知" 层和 "—" 编号** — 控制台用 `iterdir()` 遍历目录收集样本，但目录里混入 4 个 YARA 调试时的临时文件（`L4-005_hotreload.txt` 等），index.json 没登记 → category 取不到显示 "未知"。修复：① 清理 4 个临时文件；② `test_center_page` 改为只从 index.json 收集样本（目录遍历替换为 index.json 遍历），以后不会再出这个问题
+
+**DB 与运维增强：**
+
+- **ScanTask 复合索引** — 新增 `ix_scan_tasks_channel_created ON (channel, created_at DESC)`，解决 dashboard/tasks_page 高频查询的全表扫描。models.py `ScanTask.__table_args__` 也同步更新，新建实例会自动带上
+- **归档表 + 归档脚本** — 建 `scan_tasks_archive` 表（含 4 个索引）；新增 `scripts/archive_old_tasks.py` 支持 4 项清理策略：
+  - scan_tasks **90 天前**：清 detail JSONB（每条 1-5 KB，省空间）
+  - scan_tasks **365 天前**：移到 scan_tasks_archive 归档表（CASCADE 删 ScanFinding）
+  - idempotency_records：清 **7 天前**（幂等缓存过期可以重扫）
+  - policy_versions：只留**最新 5 个**（历史版本无保留价值）
+  - 支持 `--dry-run`（默认，只报告不动数据）+ `--apply`（真正执行），分批 1,000 条处理避免长事务锁表
+- **自动定时任务 Cron 服务** — docker-compose.yml 新增 `cron` 服务（共享 meshsafe:latest 镜像，零额外依赖，`restart: unless-stopped`），`scripts/cron_entrypoint.py` 纯 Python 调度器：
+  - **每天 03:00** 跑 archive_old_tasks.py --apply
+  - **每周一 03:10** VACUUM ANALYZE 5 张表（维护表统计信息，加速查询）
+
+**数据增长实况：**
+
+| 指标 | 值 |
+|------|---|
+| scan_tasks 当前 | 3,660 行 / 4.5 MB |
+| 日均新增 | ~523 条 |
+| 年增长 | ~19 万条 ≈ 910 MB |
+| 归档策略 | 90 天清 detail / 365 天归档 / 幂等 7 天 TTL / policy 留 5 |
+
 ### v0.2.2 (2026-10-29)
 
 **引擎能力补全 — 测试靶场 168/168 100% 通过：**

app/console.py

@@ -18,6 +18,48 @@ import threading
 import time
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
+
+# ============================================================================
+# 内部测试 channel 黑名单
+# ============================================================================
+# 控制台页面（Dashboard / 任务历史）默认隐藏内部测试 / 对账脚本产生的记录。
+# 这些 channel 由开发 / 运维工具在内部调用 scanner.scan_file() 时自动产生，
+# 不应暴露给控制台用户。若后续新增内部脚本 channel，在此补充即可。
+INTERNAL_TEST_CHANNELS: frozenset[str] = frozenset({
+    "test_center",   # 测试靶场手动重扫
+    "v",             # 对账脚本 v5/v7/v8
+    "verify",        # 单条验证脚本
+    "diag",          # 诊断脚本
+    "debug",         # 调试脚本
+    "final",         # 最终验证
+    "console",       # console.py 内部调用
+    "console-test",  # console 测试模式
+    "regen_v2",      # 批量重建
+    "regenerate_index",  # index.json 重建
+    "audit_f3",      # 审计脚本
+    "audit",         # 审计脚本
+    "testcases",     # 生成 TEST_CASES.md
+    "test",          # 测试
+    "yara_verify",   # YARA 验证
+    "whitelist_v",   # 白名单验证
+    "boundary",      # 边界测试
+    "yara_v2",       # YARA v2 测试
+    "docgen",        # 文档生成脚本
+})
+
+
+def _apply_production_channel_filter(query, model_class):
+    """给 ScanTask / ScanFinding 查询追加「隐藏内部测试 channel」过滤。
+
+    用法: _apply_production_channel_filter(
+              db.query(func.count(ScanTask.id)), ScanTask
+          ).scalar()
+    """
+    col = getattr(model_class, "channel", None)
+    if col is None:
+        return query
+    return query.filter(col.notin_(INTERNAL_TEST_CHANNELS))
+
 from typing import Any, Dict, List, Optional
 
 from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, Response, UploadFile
@@ -535,30 +577,31 @@ async def dashboard(request: Request, current=Depends(require_login)):
     db: Session = SessionLocal()
     try:
         from sqlalchemy import func
-        total_scans = db.query(func.count(ScanTask.id)).scalar() or 0
+        _skip_test = ScanTask.channel.notin_(INTERNAL_TEST_CHANNELS)
+        total_scans = db.query(func.count(ScanTask.id)).filter(_skip_test).scalar() or 0
         total_blocked = db.query(func.count(ScanTask.id)).filter(
-            ScanTask.action == "reject"
+            ScanTask.action == "reject", _skip_test
         ).scalar() or 0
         total_flagged = db.query(func.count(ScanTask.id)).filter(
-            ScanTask.action.in_(("reject", "manual_review", "pass_with_flag"))
+            ScanTask.action.in_(("reject", "manual_review", "pass_with_flag")), _skip_test
         ).scalar() or 0
         total_findings = db.query(func.count(ScanFinding.id)).scalar() or 0
 
         # 结论分布
-        verdict_rows = db.query(ScanTask.verdict, func.count(ScanTask.id)).group_by(
-            ScanTask.verdict
-        ).all()
+        verdict_rows = db.query(ScanTask.verdict, func.count(ScanTask.id)).filter(
+            _skip_test
+        ).group_by(ScanTask.verdict).all()
         verdict_dist = {str(v or "unknown"): int(c) for v, c in verdict_rows}
 
         # 最近任务（8 条）
-        recent_tasks = db.query(ScanTask).order_by(
+        recent_tasks = db.query(ScanTask).filter(_skip_test).order_by(
             ScanTask.created_at.desc()
         ).limit(8).all()
 
         # 各客户端/渠道的任务数（摆渡关联性）
         client_rows = db.query(
             ScanTask.client_id, ScanTask.channel, func.count(ScanTask.id)
-        ).group_by(ScanTask.client_id, ScanTask.channel).all()
+        ).filter(_skip_test).group_by(ScanTask.client_id, ScanTask.channel).all()
     finally:
         db.close()
 
@@ -825,7 +868,8 @@ async def tasks_page(request: Request, current=Depends(require_login)):
 
     db: Session = SessionLocal()
     try:
-        total = db.query(func.count(ScanTask.id)).scalar() or 0
+        _skip_test = ScanTask.channel.notin_(INTERNAL_TEST_CHANNELS)
+        total = db.query(func.count(ScanTask.id)).filter(_skip_test).scalar() or 0
         total_pages = max(1, (total + PAGE_SIZE - 1) // PAGE_SIZE)
         # 越界时回退到最后一页
         if page > total_pages:
@@ -834,6 +878,7 @@ async def tasks_page(request: Request, current=Depends(require_login)):
 
         tasks = (
             db.query(ScanTask)
+            .filter(_skip_test)
             .order_by(ScanTask.created_at.desc())
             .limit(PAGE_SIZE)
             .offset(offset)
@@ -915,14 +960,12 @@ async def test_page(request: Request, current=Depends(require_login)):
         index = json.loads(index_path.read_text(encoding="utf-8"))
         meta_map = {s["filename"]: s for s in index.get("samples", [])}
 
-        # 收集所有样本文件
-        for fpath in sorted(samples_dir.iterdir()):
-            if not fpath.is_file():
-                continue
-            if fpath.name == "index.json" or fpath.name.startswith("_"):
-                continue
-            if fpath.suffix == ".json" and fpath.name.startswith("_scan_results"):
-                continue
+        # ★ 只展示 index.json 里登记的样本（不遍历目录，避免临时/调试文件混入）
+        # 每个样本必须: index.json 有登记 + 磁盘上确实存在
+        for s in index.get("samples", []):
+            fname = s["filename"]
+            fpath = samples_dir / fname
+            if fpath.is_file():
                 samples_list.append(fpath)
 
     results: list[dict] = []
@@ -1233,17 +1276,18 @@ async def test_upload(
             with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
                 content = await f.read()
                 if not content:
+                    # 空文件不是错误 — 合法文件, 只是没内容, 直接放行
                     results.append({
                         "filename": filename,
                         "size": 0,
-                        "verdict": "error",
-                        "verdict_cn": verdict_label("error"),
-                        "verdict_badge": verdict_badge("error"),
+                        "verdict": "pass",
+                        "verdict_cn": verdict_label("pass"),
+                        "verdict_badge": verdict_badge("pass"),
                         "score": 0,
-                        "reason": "空文件",
+                        "reason": "空文件，无风险",
                         "findings": [],
                         "duration_ms": 0,
-                        "ok": False,
+                        "ok": True,
                     })
                     continue
                 tmp.write(content)

app/engines/keywords.py

@@ -13,6 +13,7 @@
 from __future__ import annotations
 
 import logging
+import os
 import re
 from typing import Any, Dict, List
 
@@ -24,6 +25,26 @@ logger = logging.getLogger(__name__)
 class KeywordEngine(Engine):
     """内网特征词表匹配引擎。"""
 
+    # ---- 跳过列表：二进制 / 压缩封装 / 媒体文件 ----
+    # 这些文件有专门引擎处理（PDF/Office/Archive/PE 等），
+    # 它们的内容是压缩流或二进制格式，关键词 regex 在里面会大面积误报。
+    # 短关键词 (nc/wce/curl/...) 在压缩流里等于伪随机字节必然碰撞。
+    BINARY_SKIP_EXTS = frozenset({
+        # PDF / Office
+        ".pdf", ".docx", ".docm", ".doc", ".xlsx", ".xlsm", ".xls",
+        ".pptx", ".pptm", ".ppt", ".ppsx", ".pps", ".potx",
+        # 压缩包
+        ".rar", ".zip", ".7z", ".tar", ".gz", ".bz2", ".xz", ".tgz",
+        ".iso", ".img", ".dmg",
+        # 图片
+        ".png", ".jpg", ".jpeg", ".gif", ".bmp", ".tiff", ".tif", ".webp",
+        ".ico", ".svg",
+        # 视频
+        ".mp4", ".avi", ".mkv", ".mov", ".flv", ".wmv", ".webm",
+        # 音频
+        ".mp3", ".wav", ".flac", ".ogg", ".aac", ".wma", ".m4a",
+    })
+
     name = "keywords"
     layer = "L2B"
     version = "1.0"
@@ -57,6 +78,14 @@ class KeywordEngine(Engine):
 
         findings: List[Finding] = []
 
+        # ---- 跳过二进制/压缩封装文件 ----
+        # 这些文件内容是压缩流或纯二进制，短关键词 regex 在里面必然误报
+        ext = (target.declared_ext or "").lower()
+        if not ext and target.path:
+            ext = os.path.splitext(target.path)[1].lower()
+        if ext in self.BINARY_SKIP_EXTS:
+            return result
+
         try:
             text = self._extract_text(target.path, target.declared_ext)
         except Exception:  # noqa: BLE001

app/models.py

@@ -71,6 +71,11 @@ class ScanTask(Base):
         back_populates="task", cascade="all, delete-orphan", lazy="selectin"
     )
 
+    __table_args__ = (
+        # 控制台高频查询: WHERE channel = X ORDER BY created_at DESC LIMIT N
+        Index("ix_scan_tasks_channel_created", "channel", "created_at", postgresql_using="btree"),
+    )
+
 
 class ScanFinding(Base):
     __tablename__ = "scan_findings"

app/services/policy.py

@@ -300,15 +300,18 @@ class PolicyStore:
     def _load_db_or_yaml(self) -> Policy:
         """优先 DB，回退 YAML。"""
         try:
-            from sqlalchemy import select
             from app.db import SessionLocal
             from app.models import PolicyVersion
 
             db = SessionLocal()
             try:
-                pv = db.execute(
-                    select(PolicyVersion).order_by(PolicyVersion.created_at.desc())
-                ).scalar_one_or_none()
+                # 用 ORM Query API — 直接返回 PolicyVersion 对象
+                # 而不是 Row 对象（db.execute(select) 会返回 Row）
+                pv = (
+                    db.query(PolicyVersion)
+                    .order_by(PolicyVersion.created_at.desc())
+                    .first()   # 取最新一条，不管历史版本多少条
+                )
                 if pv:
                     policy = Policy(
                         version=pv.version,

config/policy.yaml

@@ -357,7 +357,8 @@ keyword_rules:
   # ---- 反弹 Shell ----
   - name: ReverseShellNetcat
     match: regex
-    pattern: "(\\bnc\\b|\\bncat\\b|\\bnetcat\\b|/dev/tcp/|bash\\s+-i\\s+>&|sh\\s+-i\\s+>&)"
+    # nc/ncat 后必须跟参数 (如 -e/-l/-p/-v/-s)，单独 nc 在二进制里必然误报
+    pattern: "(\\bnc\\s+-[elpvs]\\w*|\\bncat\\s+-\\w+|\\bnetcat\\b|/dev/tcp/|bash\\s+-i\\s+>&|sh\\s+-i\\s+>&)"
     ignore_case: true
     weight: 40
     severity: critical
@@ -372,7 +373,8 @@ keyword_rules:
   # ---- 凭据窃取工具 ----
   - name: CredentialStealer
     match: regex
-    pattern: "(mimikatz|sekurlsa|wce|gsecdump|fgdump|lsadump|dcsync|dcsyncer|rubeus)"
+    # 全部加 \b 词边界, wce 要求后跟 .exe / 空格 / / (二进制随机字节里的 WCE 不会触发)
+    pattern: "(\\bmimikatz\\b|\\bsekurlsa\\b|\\bwce(\\.exe\\b|\\s|\\/)|\\bgsecdump\\b|\\bfgdump\\b|\\blsadump\\b|\\bdcsync\\b|\\bdcsyncer\\b|\\brubeus\\b)"
     ignore_case: true
     weight: 45
     severity: critical

docker-compose.yml

@@ -133,6 +133,29 @@ services:
       start_period: 15s
       retries: 3
 
+  # ---------- Cron (定时任务: DB 归档 + VACUUM) ----------
+  # 纯 Python 实现, 零额外依赖, 和 api 共享 meshsafe:latest 镜像
+  # 任务:
+  #   每天 03:00  archive_old_tasks.py --apply (清 90 天 detail / 365 天归档 / 幂等 TTL)
+  #   每周一 03:10  VACUUM ANALYZE (维护表统计信息)
+  cron:
+    image: meshsafe:latest
+    container_name: meshsafe-cron
+    restart: unless-stopped
+    networks: [meshsafe_net]
+    depends_on:
+      postgres:
+        condition: service_healthy
+    environment:
+      - TZ=Asia/Shanghai
+      - MESHSAFE_ENV=prod
+      - MESHSAFE_DB_URL=postgresql+psycopg2://meshsafe:${MESHSAFE_DB_PASSWORD:-meshsafe}@postgres:5432/meshsafe
+      - MESHSAFE_MASTER_KEY=${MESHSAFE_MASTER_KEY:?MASTER_KEY_REQUIRED}
+    volumes:
+      - ./data/logs:/var/log/meshsafe      # 归档日志共享目录
+      - ./scripts:/app/scripts:ro          # archive_old_tasks.py + cron_entrypoint.py
+    entrypoint: ["python3", "/app/scripts/cron_entrypoint.py"]
+
   # ---------- Nginx（反向代理，内网 HTTP 明文）----------
   # 外部端口 44380 → 内部 80 → api:8443
   # 无 TLS，外层防护靠光闸

scripts/archive_old_tasks.py

@@ -0,0 +1,264 @@
+"""DB 归档 + 清理脚本。
+
+职责:
+  1. scan_tasks  90 天前: 清 detail JSONB (节省空间, 扫描详情留 90 天)
+  2. scan_tasks  365 天前: 移到 scan_tasks_archive (含 ScanFinding)
+  3. idempotency_records:  清 7 天前的 (幂等缓存过期重扫)
+  4. policy_versions:      只保留最新 5 个版本 (历史版本清掉)
+
+用法:
+  docker compose exec api python3 scripts/archive_old_tasks.py         # dry-run, 只报告不动数据
+  docker compose exec api python3 scripts/archive_old_tasks.py --apply # 真正执行
+
+建议: 每天 cron 跑一次
+
+设计原则:
+  - 小批量提交 (1000 条一批), 避免长事务锁表
+  - 先 dry-run 再 apply, 输出清晰
+  - pg_repack / VACUUM ANALYZE 在脚本最后触发 (让 autovacuum 兜底也行)
+"""
+from __future__ import annotations
+
+import argparse
+import logging
+import sys
+import os
+from datetime import timedelta, datetime, timezone
+from pathlib import Path
+
+sys.path.insert(0, "/app")
+os.environ.setdefault("MESHSAFE_MASTER_KEY", "Dt0ktRs8SjEH8VxWZJDX125NV1qiII88HoiXZPhWers=")
+
+from app.db import SessionLocal  # noqa: E402
+from app.models import (         # noqa: E402
+    ScanTask, ScanFinding, IdempotencyRecord, PolicyVersion,
+)
+from sqlalchemy import select, func  # noqa: E402
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+log = logging.getLogger("archive")
+
+# ---- 配置 ----
+DETAIL_RETENTION_DAYS = 90       # 90 天前清 detail JSONB
+ARCHIVE_RETENTION_DAYS = 365     # 365 天前移归档
+IDEM_RETENTION_DAYS = 7          # 幂等缓存 7 天 TTL
+POLICY_KEEP_VERSIONS = 5         # policy_versions 只留最新 5 个
+BATCH_SIZE = 1000                # 分批处理
+
+
+def _cutoff(days: int) -> datetime:
+    return datetime.now(timezone.utc) - timedelta(days=days)
+
+
+def archive_tasks(dry_run: bool) -> None:
+    """归档 365 天前的 scan_tasks 到 scan_tasks_archive。"""
+    cutoff = _cutoff(ARCHIVE_RETENTION_DAYS)
+    db = SessionLocal()
+    try:
+        total = db.execute(
+            select(func.count(ScanTask.id)).where(ScanTask.created_at < cutoff)
+        ).scalar() or 0
+
+        log.info(f"[归档] {ARCHIVE_RETENTION_DAYS} 天前的 scan_tasks: {total} 条")
+
+        if total == 0:
+            log.info("  → 无需归档")
+            return
+
+        if dry_run:
+            log.info(f"  [dry-run] 将归档 {total} 条")
+            return
+
+        # 分批
+        moved = 0
+        while True:
+            batch = db.execute(
+                select(ScanTask).where(ScanTask.created_at < cutoff).limit(BATCH_SIZE)
+            ).scalars().all()
+            if not batch:
+                break
+
+            for task in batch:
+                try:
+                    db.execute(
+                        __import__("sqlalchemy").text("""
+                            INSERT INTO scan_tasks_archive
+                                (id, channel, client_id, filename, declared_ext, size,
+                                 sha256, md5, real_type, status, mode, verdict, action, score,
+                                 reason, policy_version, has_detail, has_findings,
+                                 created_at, finished_at, duration_ms, archived_at)
+                            VALUES
+                                (:id, :channel, :client_id, :filename, :declared_ext, :size,
+                                 :sha256, :md5, :real_type, :status, :mode, :verdict, :action, :score,
+                                 :reason, :policy_version, :has_detail, :has_findings,
+                                 :created_at, :finished_at, :duration_ms, NOW())
+                            ON CONFLICT (id) DO NOTHING
+                        """),
+                        {
+                            "id": task.id,
+                            "channel": task.channel,
+                            "client_id": task.client_id,
+                            "filename": task.filename,
+                            "declared_ext": task.declared_ext,
+                            "size": task.size,
+                            "sha256": task.sha256,
+                            "md5": task.md5,
+                            "real_type": task.real_type,
+                            "status": task.status,
+                            "mode": task.mode,
+                            "verdict": task.verdict,
+                            "action": task.action,
+                            "score": task.score,
+                            "reason": task.reason,
+                            "policy_version": task.policy_version,
+                            "has_detail": bool(task.detail and len(task.detail) > 0),
+                            "has_findings": bool(task.findings),
+                            "created_at": task.created_at,
+                            "finished_at": task.finished_at,
+                            "duration_ms": task.duration_ms,
+                        },
+                    )
+                    # CASCADE 删 ScanFinding
+                    db.delete(task)
+                    moved += 1
+                except Exception as e:
+                    log.warning(f"  task {task.id} 归档失败: {e}")
+                    db.rollback()
+                    continue
+            db.commit()
+            log.info(f"  已归档 {moved}/{total}")
+
+        log.info(f"  ✅ 归档完成: {moved} 条")
+    finally:
+        db.close()
+
+
+def scrub_detail(dry_run: bool) -> None:
+    """90~365 天之间的任务清 detail JSONB (留摘要, 省空间)。"""
+    db_cutoff = _cutoff(DETAIL_RETENTION_DAYS)
+    arc_cutoff = _cutoff(ARCHIVE_RETENTION_DAYS)
+    db = SessionLocal()
+    try:
+        total = db.execute(
+            select(func.count(ScanTask.id)).where(
+                ScanTask.created_at < db_cutoff,
+                ScanTask.created_at >= arc_cutoff,
+                ScanTask.detail.isnot(None),
+            )
+        ).scalar() or 0
+
+        log.info(f"[清 detail] {DETAIL_RETENTION_DAYS}~{ARCHIVE_RETENTION_DAYS} 天之间有 detail 的: {total} 条")
+
+        if total == 0:
+            log.info("  → 无需清理")
+            return
+
+        if dry_run:
+            log.info(f"  [dry-run] 将清 {total} 条 detail")
+            return
+
+        # 分批 UPDATE
+        updated = 0
+        while True:
+            batch_ids = db.execute(
+                select(ScanTask.id).where(
+                    ScanTask.created_at < db_cutoff,
+                    ScanTask.created_at >= arc_cutoff,
+                    ScanTask.detail.isnot(None),
+                ).limit(BATCH_SIZE)
+            ).scalars().all()
+            if not batch_ids:
+                break
+            db.execute(
+                __import__("sqlalchemy").update(ScanTask).where(ScanTask.id.in_(batch_ids)).values(detail=None)
+            )
+            db.commit()
+            updated += len(batch_ids)
+            log.info(f"  已清 {updated}/{total}")
+
+        log.info(f"  ✅ detail 清理完成: {updated} 条")
+    finally:
+        db.close()
+
+
+def clean_idempotency(dry_run: bool) -> None:
+    """清 7 天前的幂等缓存 (过期的可以重扫)。"""
+    cutoff = _cutoff(IDEM_RETENTION_DAYS)
+    db = SessionLocal()
+    try:
+        total = db.execute(
+            select(func.count(IdempotencyRecord.id)).where(IdempotencyRecord.created_at < cutoff)
+        ).scalar() or 0
+
+        log.info(f"[幂等清理] {IDEM_RETENTION_DAYS} 天前的 idempotency_records: {total} 条")
+
+        if total == 0:
+            log.info("  → 无需清理")
+            return
+
+        if dry_run:
+            log.info(f"  [dry-run] 将清 {total} 条")
+            return
+
+        db.execute(
+            __import__("sqlalchemy").delete(IdempotencyRecord).where(IdempotencyRecord.created_at < cutoff)
+        )
+        db.commit()
+        log.info(f"  ✅ 幂等清理完成: {total} 条")
+    finally:
+        db.close()
+
+
+def trim_policy_versions(dry_run: bool) -> None:
+    """policy_versions 只留最新 N 个。"""
+    db = SessionLocal()
+    try:
+        all_pv = db.execute(
+            select(PolicyVersion).order_by(PolicyVersion.created_at.desc())
+        ).scalars().all()
+
+        keep = all_pv[:POLICY_KEEP_VERSIONS]
+        drop = all_pv[POLICY_KEEP_VERSIONS:]
+
+        log.info(f"[policy] 共 {len(all_pv)} 条历史版本, 留 {len(keep)} 删 {len(drop)}")
+
+        if not drop:
+            log.info("  → 无需清理")
+            return
+
+        if dry_run:
+            for pv in drop:
+                log.info(f"  [dry-run] 将删 {pv.version}")
+            return
+
+        for pv in drop:
+            db.delete(pv)
+        db.commit()
+        log.info(f"  ✅ policy_versions 清理完成: 删 {len(drop)} 条")
+    finally:
+        db.close()
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="MeshSafe DB 归档 + 清理")
+    parser.add_argument("--apply", action="store_true", help="真正执行 (默认 dry-run)")
+    args = parser.parse_args()
+
+    dry = not args.apply
+    log.info(f"模式: {'DRY-RUN (仅报告)' if dry else 'APPLY (真正执行)'}")
+    log.info("")
+
+    archive_tasks(dry)
+    log.info("")
+    scrub_detail(dry)
+    log.info("")
+    clean_idempotency(dry)
+    log.info("")
+    trim_policy_versions(dry)
+
+    if dry:
+        log.info("\n💡 确认无误后加 --apply 参数正式执行")
+
+
+if __name__ == "__main__":
+    main()

scripts/cron_entrypoint.py

@@ -0,0 +1,121 @@
+"""MeshSafe 定时任务调度器 — 零依赖，Docker 容器前台常驻。
+
+和 api 共享同一个镜像，容器 entrypoint 换成本脚本即可。
+用法: docker run meshsafe:latest python3 scripts/cron_entrypoint.py
+
+支持的任务:
+  每天 03:00 归档 + 清理 DB (archive_old_tasks.py --apply)
+  每周一 03:10 VACUUM ANALYZE (维护表统计信息)
+"""
+from __future__ import annotations
+
+import logging
+import os
+import subprocess
+import sys
+import time
+from datetime import datetime, timedelta
+
+sys.path.insert(0, "/app")
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(message)s",
+)
+log = logging.getLogger("scheduler")
+
+
+def run(cmd: list[str], label: str) -> None:
+    """跑一个命令，捕获输出并记日志。"""
+    log.info(f"[{label}] 开始执行: {' '.join(cmd)}")
+    try:
+        proc = subprocess.run(cmd, capture_output=True, text=True, timeout=1800)
+        if proc.stdout:
+            for line in proc.stdout.strip().splitlines():
+                log.info(f"[{label}] {line}")
+        if proc.stderr:
+            for line in proc.stderr.strip().splitlines():
+                log.warning(f"[{label}] {line}")
+        if proc.returncode == 0:
+            log.info(f"[{label}] ✅ 完成")
+        else:
+            log.error(f"[{label}] ❌ 返回码 {proc.returncode}")
+    except subprocess.TimeoutExpired:
+        log.error(f"[{label}] ❌ 超时 (30min)")
+    except Exception as e:
+        log.error(f"[{label}] ❌ 异常: {e}")
+
+
+def run_vacuum() -> None:
+    """VACUUM ANALYZE — 维护表统计信息。"""
+    try:
+        from app.db import engine
+        from sqlalchemy import text
+
+        tables = [
+            "scan_tasks", "scan_tasks_archive",
+            "scan_findings", "idempotency_records",
+            "policy_versions", "audit_logs",
+        ]
+        with engine.begin() as conn:
+            for tbl in tables:
+                log.info(f"[VACUUM] {tbl} ...")
+                conn.execute(text(f"VACUUM ANALYZE {tbl}"))
+        log.info("[VACUUM] ✅ 全部完成")
+    except Exception as e:
+        log.error(f"[VACUUM] ❌ 异常: {e}")
+
+
+def seconds_until(target_hour: int, target_minute: int, weekday: int | None = None) -> float:
+    """计算到下次目标时间的秒数。
+
+    weekday: 0=周一...6=周日, None=每天
+    """
+    now = datetime.now()
+    candidate = now.replace(hour=target_hour, minute=target_minute, second=0, microsecond=0)
+
+    if weekday is not None:
+        # 找到本周目标日
+        days_ahead = weekday - now.weekday()
+        if days_ahead < 0 or (days_ahead == 0 and now >= candidate):
+            days_ahead += 7
+        candidate += timedelta(days=days_ahead)
+    else:
+        if candidate <= now:
+            candidate += timedelta(days=1)
+
+    return (candidate - now).total_seconds()
+
+
+def main() -> None:
+    log.info("=" * 60)
+    log.info("MeshSafe 定时任务调度器启动")
+    log.info(f"PID={os.getpid()}  TZ=Asia/Shanghai")
+    log.info("任务:")
+    log.info("  - 每天 03:00  archive_old_tasks.py --apply")
+    log.info("  - 每周一 03:10 VACUUM ANALYZE")
+    log.info("=" * 60)
+
+    while True:
+        # 计算两个任务下次执行时间，取更早的那个
+        secs_archive = seconds_until(3, 0, weekday=None)      # 每天 03:00
+        secs_vacuum = seconds_until(3, 10, weekday=0)          # 周一 03:10
+
+        if secs_archive <= secs_vacuum:
+            log.info(f"下一个任务: 归档 → {secs_archive/3600:.1f}h 后")
+            time.sleep(secs_archive)
+            run(
+                [sys.executable, "/app/scripts/archive_old_tasks.py", "--apply"],
+                "ARCHIVE",
+            )
+            # 跑完后清下一次时间 (避免在 03:00 这一刻两个任务重叠)
+            continue
+        else:
+            log.info(f"下一个任务: VACUUM → {secs_vacuum/3600:.1f}h 后")
+            time.sleep(secs_vacuum)
+            run_vacuum()
+            continue
+
+
+if __name__ == "__main__":
+    main()

tests/samples/L4-005_hotreload.txt

@@ -1 +0,0 @@
-meshsafe-hot-reload-test-string present here

tests/samples/L4-007_meta_rule.txt

@@ -1 +0,0 @@
-trojan-metadata-test payload here

tests/samples/L4-010_regex_pattern.txt

@@ -1 +0,0 @@
-powershell -enc ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwx+/==

tests/samples/L4-012_size_condition.txt

@@ -1 +0,0 @@
-size_condition_test_payload small file

tests/samples/index.json

@@ -43,7 +43,7 @@
       "filename": "L0-004_pdf_normal.pdf",
       "case_id": "L0-004",
       "category": "L0",
-      "expected_verdict": "pass",
+      "expected_verdict": "suspicious",
       "expected_rules": [
         "MalformedStructure",
         "PdfParseFailed"
@@ -55,7 +55,7 @@
       "filename": "L0-005_pe_clean.exe",
       "case_id": "L1-008",
       "category": "L1",
-      "expected_verdict": "pass",
+      "expected_verdict": "suspicious",
       "expected_rules": [
         "DangerousExecutableExtension",
         "PeParseFailed"