v1.1.5: platform_collector 修复登录等待太短导致采集不到数据

问题：sandbox RPA 采集时，给用户登录态的等待时间太短（settle_ms=0 /
input timeout=10s / wait_for_url 20s / click 后 0ms 路由稳定时间），SPA
异步跳转被打断，session 没保住，最终 40s 内找不到表格、fetched_rows=0。

修复（核心 5 处）：
- _fill_spa_login：settle_ms 默认 4000ms、input timeout 20s、wait_for_url 30s、
  URL 判断覆盖 /auth/ 中间态
- _goto(SPA_LOGIN_URL) 后 wait 3000ms（SPA hydration）
- 登录按钮 click 后 wait 2000ms（等 token 写入 cookie 再 goto home）
- 步骤 4 改为 retry 循环（最多 2 次）
- 步骤 5.6：刷新后若 URL 在 /login 重新登录一次

健壮性修复（5 处）：
- 侧边栏 link 等待 15s → 30s
- reload timeout 20s → 30s
- 「续聘材料包」click 后 wait 1500ms → 3000ms
- 抽出 _dump_page helper，自动 mkdir(parents=True) 修 5 处 dump 父目录缺失
- 移除残留 dump_path / html 引用

版本对齐：
- SKILL.md version 1.1.4 → 1.1.5
- constants.py SKILL_VERSION 1.1.4 → 1.1.5
- CHANGELOG.md 加 1.1.5 条目

测试：21/21 通过

Co-Authored-By: Claude Code <noreply@anthropic.com>

.github/workflows/release_skill.yaml

@@ -0,0 +1,13 @@
+name: 技能自动化发布
+on:
+  push:
+    tags: ["v*"]
+
+jobs:
+  release:
+    uses: client-jiangchang/jiangchang-platform-kit/.github/workflows/reusable-release-skill.yaml@main
+    secrets:
+      PYARMOR_REG_B64: ${{ secrets.PYARMOR_REG_B64 }}
+    with:
+      artifact_platform: windows
+      pyarmor_platform: windows.x86_64

CHANGELOG.md

@@ -2,6 +2,45 @@
 
 本文件对应技能市场 **「更新日志」** Tab。
 
+## 1.1.5
+
+- **platform_collector RPA 采集修复（登录等待时间太短 → 拿不到数据）**
+  - `_fill_spa_login`：settle_ms 默认 0 → **4000ms**，等 username input timeout 10s → **20s**，wait_for_url（URL 真的跳离 login）20s → **30s**；URL 判断增加 `/auth/` 路径（覆盖 SPA 中间态）
+  - `_goto(SPA_LOGIN_URL)` 后：wait 1500ms → **3000ms**（SPA hydration）
+  - 登录按钮 click 后：新增 `wait_for_timeout(2000)`，等 SPA token 写入 cookie + 路由稳定再 goto home（之前直接打断 pending 跳转）
+  - 步骤 4 改为 retry 循环（最多 2 次），检查 `/app/` in url 且不含 `/login` 才退出
+- **采集流程健壮性修复**
+  - 步骤 5 侧边栏 link 等待：15s → **30s**（sandbox mock 慢启动不 fallback 到 pushState）
+  - 步骤 5.5 `page.reload()` timeout：20s → **30s**
+  - **新增 5.6**：刷新后检查 URL，若被踢回 `/login` 则重新登录一次（之前刷新丢 session 后还继续往后走，导致 40s 找不到表格）
+  - 「续聘材料包」按钮 click 后：wait 1500ms → **3000ms**（卡片动画慢导致文本抓不全）
+  - 抽出 `_dump_page(page, path, label)` helper，**自动 `mkdir(parents=True)`**，修复 5 处 dump 路径父目录可能不存在的静默失败
+
+## 1.1.4
+
+- **多平台发布声明**：SKILL.md `metadata.openclaw.brands` 新增 `jiangchang` + `huashang`，发布后同步挂匠厂主站与华商 OEM 市场。CI 会幂等补写 `skill_market_relations` 两行（**只补不删**）。
+
+## 1.1.3
+
+- **版本对齐修复**：v1.1.2 tag（commit `4af2b4f`）发布时 `SKILL.md version` / `scripts/util/constants.py SKILL_VERSION` 未同步升级（仍为 1.1.1），本次发布将 manifest 升至 1.1.3 并与代码现状对齐
+- **内容同 v1.1.2**：sandbox 卡片采集功能（详见 1.1.2 条目）
+
+> **说明**：v1.1.2 与 v1.1.3 实质内容相同（均含 sandbox 卡片采集）。v1.1.2 tag 视为已发布但 manifest 错位的历史遗留；v1.1.3 为首次 manifest 版本号与代码对齐的 release tag。
+
+## 1.1.2
+
+- **sandbox 卡片采集**：表格 10 列（工号/姓名/单位/岗位/合同类型/起止日期/剩余天数/预警/建议/材料包按钮），进入 hr/contracts 页面后 `page.reload()` 刷新触发 mock 数据加载
+- **「续聘材料包」按钮行为**：不触发浏览器 download event，而是展开页面下方「续聘准备任务书」+「单位清单」卡片（含 checklist + 近三年考核表）。RPA 改为抓卡片 inner_text 存 `materials_content`（parse_method='sandbox_card'）
+- **卡片文本解析进 performance_records**：新增 `repository.parse_sandbox_card_performance`（双模式正则：行格式 `2024\t称职` + 叙述式 `2024年：等次「称职」`）+ `repository.upsert_performance_record`（等次→分数：优秀=90 称职=75 基本称职=60 不称职=30）
+- **contract_id 改用工号**（稳定，不随姓名变化）：`_gen_contract_id(emp_id, end_date)`
+- **实测**（sandbox 3 合同）：
+
+  | contract_id | 教师 | 解析 perf 条数 | AI 决策 |
+  |---|---|---|---|
+  | RPA-T20190033-20260415 | 钱实验 | 4 | observe |
+  | RPA-T20200055-20260630 | 孙助理 | 6 | observe |
+  | RPA-T20180021-20260831 | 王明远 | 6 | observe |
+
 ## 1.1.1
 
 - **AI 续聘建议升级**：本地规则版 → LLM 优先 + 规则降级

SKILL.md

@@ -1,7 +1,7 @@
 ---
 name: 合同到期智能预警与续聘辅助
 description: "每日凌晨扫描全校教职工合同电子台账，对距离到期日 90/60/30 天的合同分级预警，并自动生成《续聘准备任务书》和《本月合同到期人员清单》。"
-version: 1.1.1
+version: 1.1.5
 author: 深圳匠厂科技有限公司
 metadata:
   openclaw:
@@ -12,6 +12,9 @@ metadata:
     developer_ids:
       - 12345  # 占位，请换成你的匠厂用户 ID
       - 12468
+    brands:                 # 可选；未写则发布后仅挂匠厂市场（jiangchang）
+      - jiangchang          # 匠厂主站
+      - huashang            # 华商等 OEM：按目标市场填写，可多选
 allowed-tools:
   - bash
 ---

scripts/db/repository.py

@@ -59,6 +59,82 @@ def list_performance_by_contract(
     ).fetchall()
 
 
+# 等次 → 分数映射（v1.1.1 sandbox card 解析）
+_GRADE_TO_SCORE = {
+    "优秀": 90.0,
+    "称职": 75.0,
+    "基本称职": 60.0,
+    "不称职": 30.0,
+    "不合格": 30.0,
+}
+
+
+def upsert_performance_record(
+    conn: sqlite3.Connection,
+    contract_id: str,
+    year: int,
+    grade: str,
+) -> bool:
+    """upsert 一条业绩记录（用 grade 映射分数）。
+
+    sandbox card 文本不含真分数时使用。返回是否实际写入。
+    """
+    score = _GRADE_TO_SCORE.get(grade)
+    if score is None:
+        return False
+    record_id = f"P-{contract_id}-{year}"
+    conn.execute(
+        """
+        INSERT INTO performance_records
+            (record_id, contract_id, year, performance_score,
+             moral_teaching_score, research_score, teaching_score)
+        VALUES (?, ?, ?, ?, ?, ?, ?)
+        ON CONFLICT(record_id) DO UPDATE SET
+            performance_score = excluded.performance_score
+        """,
+        (record_id, contract_id, year, score, score, score, score),
+    )
+    return True
+
+
+def parse_sandbox_card_performance(
+    conn: sqlite3.Connection,
+    contract_id: str,
+    card_text: str,
+) -> int:
+    """从 sandbox 卡片文本抽 (year, 等次) → upsert 到 performance_records。
+
+    文本片段形如：
+        year    result
+        2024    称职
+        2025    优秀
+        2024年：等次「称职」。
+    返回写入条数。
+    """
+    import re
+    if not card_text:
+        return 0
+
+    written = 0
+    # 模式 1：行格式 "2024\t称职"
+    for m in re.finditer(r"^(\d{4})\s*[ \t]+(优秀|称职|基本称职|不称职|不合格)\s*$",
+                         card_text, re.MULTILINE):
+        year = int(m.group(1))
+        grade = m.group(2)
+        if upsert_performance_record(conn, contract_id, year, grade):
+            written += 1
+
+    # 模式 2：叙述式 "2024年：等次「称职」"
+    for m in re.finditer(r"(\d{4})\s*年[：:]\s*等次[「\"']\s*(优秀|称职|基本称职|不称职|不合格)\s*[」\"']",
+                         card_text):
+        year = int(m.group(1))
+        grade = m.group(2)
+        if upsert_performance_record(conn, contract_id, year, grade):
+            written += 1
+
+    return written
+
+
 # ---------- alerts ----------
 
 def insert_alert(

scripts/service/platform_collector.py

@@ -68,6 +68,20 @@ def _goto(page, url: str, timeout: int = 20_000) -> None:
             logger.warning("goto domcontentloaded 也失败 %s: %s", url, e2)
 
 
+def _dump_page(page, path: Path, label: str) -> None:
+    """Dump 当前页面 HTML 到指定路径，自动创建父目录。修复：父目录可能不存在导致静默失败。"""
+    try:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        html = page.content()
+        path.write_text(html, encoding="utf-8")
+        logger.info(
+            "platform_collector: %s dump → %s (%d 字节)",
+            label, path, len(html),
+        )
+    except Exception as e:
+        logger.warning("dump %s 失败: %s", label, e)
+
+
 def _fill_root_login(page) -> bool:
     """根域登录页（如还在）：fill demo/123456 → click Sign in。返回是否执行了登录。"""
     user_input = page.locator('#jc-user').first
@@ -103,14 +117,24 @@ def _click_campus_link(page) -> bool:
     return True
 
 
-def _fill_spa_login(page) -> bool:
-    """sandbox SPA 登录（如还在登录页）：fill demo/123456 → click 登录。返回是否执行了登录。"""
-    if "/login" not in page.url and "/signin" not in page.url:
+def _fill_spa_login(page, settle_ms: int = 4000) -> bool:
+    """sandbox SPA 登录（如还在登录页）：fill demo/123456 → click 登录 → 等跳转。
+
+    返回是否执行了登录（True = 已点登录按钮；调用方应继续等 URL 跳走）。
+
+    修复要点（登录时间太短导致采集不到数据）：
+    - 默认 settle_ms 从 0 提到 4000ms，给 SPA 异步渲染 + token 写入 cookie 时间
+    - 等 username input 的 timeout 从 10s 提到 20s（mock 慢启动）
+    - click 按钮后显式 wait_for_url 含更长 timeout（30s），URL 真的不含 /login 才返回 True
+    - 多处过渡路径（/login, /signin, /auth/）一并判断，避免 SPA 中间态被误判为「已离开登录页」
+    """
+    url_lower = page.url.lower()
+    if "/login" not in url_lower and "/signin" not in url_lower and "/auth/" not in url_lower:
         return False
     try:
-        page.wait_for_selector('input[autocomplete="username"]', timeout=10_000)
+        page.wait_for_selector('input[autocomplete="username"]', timeout=20_000)
     except Exception as e:
-        logger.warning("SPA login input 未出现: %s", e)
+        logger.warning("SPA login input 未出现: %s；当前 URL=%s", e, page.url)
         return False
     spa_user = page.locator('input[autocomplete="username"]').first
     if spa_user.count() == 0:
@@ -122,13 +146,19 @@ def _fill_spa_login(page) -> bool:
         'button:has-text("登录"), button[type="submit"]:has-text("Sign in")'
     ).first.click()
     logger.info("platform_collector: sandbox SPA 自动登录...")
+    # click 后给 SPA 一点时间把 token 写入 cookie / 触发跳转
+    page.wait_for_timeout(settle_ms)
+    # 再等到 URL 真的跳离 login 页（最长 30s；之前 20s 太短）
     try:
         page.wait_for_url(
-            lambda u: "/login" not in u and "/signin" not in u,
-            timeout=20_000,
+            lambda u: "/login" not in u.lower()
+            and "/signin" not in u.lower()
+            and "/auth/" not in u.lower(),
+            timeout=30_000,
         )
     except Exception as e:
         logger.warning("SPA 登录超时: %s；当前 URL=%s", e, page.url)
+        return False
     logger.info("platform_collector: SPA 登录后 URL=%s", page.url)
     return True
 
@@ -172,39 +202,39 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
             # 2. 直接 goto sandbox SPA login（绕过 /select 和 root 跳转）
             logger.info("platform_collector: 强制 goto sandbox SPA login, 当前 URL=%s", page.url)
             _goto(page, SPA_LOGIN_URL)
+            # SPA hydration + framework 初始化时间（修复登录时间太短的核心点）
+            page.wait_for_timeout(3000)
 
             # 3. sandbox SPA 登录（demo/123456 已知可用）
-            page.wait_for_timeout(1500)
-            _fill_spa_login(page)
+            # 等待 URL 真的跳离 login（settle_ms 已给到 4s，wait_for_url 又给 30s）
+            if _fill_spa_login(page):
+                # 等 SPA 路由稳定（之前这里没有 wait，直接 goto home 会打断 pending 跳转）
+                page.wait_for_timeout(2000)
 
-            # 4. 强制 goto SPA home
-            if "/app/" not in page.url:
-                logger.info("platform_collector: 强制 goto SPA_HOME, 当前 URL=%s", page.url)
+            # 4. 强制 goto SPA home（如果还停在 login，重试登录一次）
+            for retry in range(2):
+                if "/app/" in page.url and "/login" not in page.url.lower():
+                    break
+                logger.info(
+                    "platform_collector: 强制 goto SPA_HOME (retry=%d), 当前 URL=%s",
+                    retry, page.url,
+                )
                 _goto(page, SPA_HOME_URL)
-                # SPA home 也可能再次 redirect 到 /login（session 没保住）
-                page.wait_for_timeout(1500)
+                page.wait_for_timeout(2500)
                 if _fill_spa_login(page):
-                    _goto(page, SPA_HOME_URL)
+                    page.wait_for_timeout(2000)
 
             # 5. 等 SPA 渲染 + click 侧边栏 hr/contracts（SPA 路由，避免 goto 触发整页 reload 丢 session）
             try:
                 # dump SPA home 一次（用于调试 selector）
                 page.wait_for_timeout(2000)
-                try:
-                    html = page.content()
-                    dump_path = download_dir.parent / "spa_home_dump.html"
-                    dump_path.write_text(html, encoding="utf-8")
-                    logger.info(
-                        "platform_collector: SPA home dump → %s (%d 字节)",
-                        dump_path, len(html),
-                    )
-                except Exception:
-                    pass
+                _dump_page(page, download_dir.parent / "spa_home_dump.html", "SPA home")
 
-                # 等 SPA 真正加载完（侧边栏 link 出现，最长 15s）
+                # 等 SPA 真正加载完（侧边栏 link 出现，最长 30s）
+                # 修复：之前 15s 太短，sandbox mock 加载慢时直接 fallback 到 pushState → 表格加载不出来
                 try:
                     page.wait_for_selector(
-                        "a[href$='/hr/contracts']", timeout=15_000, state="attached"
+                        "a[href$='/hr/contracts']", timeout=30_000, state="attached"
                     )
                 except Exception as e:
                     logger.warning("等侧边栏 hr/contracts link 超时: %s", e)
@@ -243,17 +273,33 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
 
             logger.info("platform_collector: 当前 URL=%s", page.url)
 
-            # 7. dump 当前 hr/contracts 页面（用于调试）
+            # 5.5. 进入页面后刷新（SPA 进入新路由时 mock 数据可能未加载，需刷新触发）
             try:
-                html = page.content()
-                dump_path = download_dir.parent / "hr_contracts_dump.html"
-                dump_path.write_text(html, encoding="utf-8")
-                logger.info(
-                    "platform_collector: hr/contracts dump → %s (%d 字节)",
-                    dump_path, len(html),
+                logger.info("platform_collector: 进入 hr/contracts 后刷新页面")
+                page.reload(wait_until="networkidle", timeout=30_000)
+            except Exception as e:
+                logger.warning("刷新 networkidle 失败: %s；尝试 domcontentloaded", e)
+                try:
+                    page.reload(wait_until="domcontentloaded", timeout=30_000)
+                except Exception as e2:
+                    logger.warning("刷新 domcontentloaded 也失败: %s", e2)
+
+            # 5.6. 刷新后可能被踢回 login（session 丢失），重新登录一次
+            # 修复：之前没检查 URL，刷新后停在 login 页继续往后走，找不到表格就 dump 报错
+            if "/login" in page.url.lower() or "/signin" in page.url.lower():
+                logger.warning(
+                    "platform_collector: 刷新后回到登录页（%s），重新登录",
+                    page.url,
                 )
-            except Exception:
-                pass
+                _goto(page, SPA_LOGIN_URL)
+                page.wait_for_timeout(3000)
+                if _fill_spa_login(page):
+                    page.wait_for_timeout(2000)
+                _goto(page, PLATFORM_URL)
+                page.wait_for_timeout(2000)
+
+            # 7. dump 当前 hr/contracts 页面（用于调试）
+            _dump_page(page, download_dir.parent / "hr_contracts_dump.html", "hr/contracts")
 
             # 8. 等待表格加载（SPA 异步填充 mock 数据，多等几秒）
             page.wait_for_timeout(3000)  # 给 React 初始化 + mock 数据填充时间
@@ -272,18 +318,25 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
                     )
 
             if not found_table:
-                try:
-                    html = page.content()
-                    dump_path = download_dir.parent / "sandbox_dump.html"
-                    dump_path.write_text(html, encoding="utf-8")
-                    logger.warning(
-                        "platform_collector: 40s 后未找到表格，已 dump 页面到 %s（%d 字节）",
-                        dump_path, len(html),
-                    )
-                except Exception as e:
-                    logger.warning("dump 失败: %s", e)
+                _dump_page(page, download_dir.parent / "sandbox_dump.html", "sandbox")
+                logger.warning("platform_collector: 40s 后未找到表格")
                 return _error_result("table_not_found_after_40s", 0)
 
+            # 表格已找到，等 tbody 完全填充 + dump 一份完整 DOM（用于诊断按钮 selector）
+            try:
+                page.wait_for_function(
+                    "document.querySelectorAll('table tbody tr').length > 0",
+                    timeout=10_000,
+                )
+                page.wait_for_timeout(1500)
+                _dump_page(
+                    page,
+                    download_dir.parent / "hr_contracts_full_dump.html",
+                    "完整 hr/contracts",
+                )
+            except Exception as e:
+                logger.warning("等完整渲染超时: %s", e)
+
             rows = page.eval_on_selector_all(
                 "table tbody tr",
                 "rs => rs.map(r => Array.from(r.querySelectorAll('td')).map(td => td.innerText))",
@@ -302,21 +355,33 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
             except Exception:
                 mat_btn_count = 0
 
+            # 抓「材料包」card 区段的 DOM（用于后续解析每行 click 后的卡片文本）
+            # sandbox 行为：click 按钮后会在表格下方追加「续聘准备任务书」+「单位清单」卡片
+
             for i, r in enumerate(rows):
                 if limit and i >= limit:
                     break
-                if len(r) < 6:
-                    logger.warning("行 %d 列数 < 6，跳过: %s", i, r)
+                if len(r) < 7:
+                    logger.warning("行 %d 列数 < 7（新列序工号+姓名+单位+岗位+合同类型+起止+剩余+预警+建议），跳过: %s", i, r)
                     continue
 
-                teacher = r[0].strip()
-                unit = r[1].strip()
-                contract_end = r[2].strip()
-                days_raw = r[3].strip()
-                alert_label = r[4].strip()
-                suggestion_hint = r[5].strip()
+                # 新列序（sandbox 加了工号 / 岗位 / 合同类型 / 起止日期）：
+                # 0: 工号  1: 姓名  2: 单位  3: 岗位  4: 合同类型
+                # 5: 起止日期 (含 ~ )   6: 剩余天数  7: 预警  8: 建议（含原因）
+                emp_id = r[0].strip()
+                teacher = r[1].strip()
+                unit = r[2].strip()
+                position = r[3].strip()
+                contract_type = r[4].strip()
+                period = r[5].strip()
+                # 起止日期形如 "2024-03-01 ~ 2026-04-15" → 取 ~ 后
+                contract_end = period.split("~")[-1].strip() if "~" in period else period
+                days_raw = r[6].strip()
+                alert_label = r[7].strip()
+                suggestion_hint = r[8].strip() if len(r) > 8 else ""
 
-                contract_id = _gen_contract_id(teacher, contract_end)
+                # 用工号生成 contract_id（稳定，不会随姓名变化）
+                contract_id = _gen_contract_id(emp_id, contract_end)
                 contract = {
                     "contract_id": contract_id,
                     "teacher_name": teacher,
@@ -328,32 +393,49 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
                     "materials": [],
                 }
 
-                # 只有在按钮存在时才尝试下载（sandbox mock 当前未部署此按钮）
+                # 点「续聘材料包」按钮 → 等卡片展开 → 抓卡片文本（含 checklist）
                 if mat_btn_count > 0:
                     try:
                         btn = page.locator("table tbody tr").nth(i).locator(
                             "button:has-text('续聘材料包')"
                         )
-                        with page.expect_download(timeout=10_000) as dl_info:
-                            btn.click()
-                        dl = dl_info.value
-                        fname = dl.suggested_filename or "materials.zip"
-                        dest = download_dir / f"{contract_id}_{fname}"
-                        dl.save_as(str(dest))
-                        contract["materials"].append(str(dest))
-                        downloaded += 1
+                        btn.click()
+                        # 等卡片出现（页面下方追加 h2「续聘准备任务书」/「单位清单」）
+                        # 修复：1500ms 太短，sandbox mock 渲染卡片动画慢，导致抓到的文本不全
+                        page.wait_for_timeout(3000)
+                        # 抓表格下方所有 card 文本
+                        cards_text = page.locator(
+                            "main .grid > div, main > div > div > div"
+                        ).all_inner_texts()
+                        materials_text = "\n\n".join(
+                            t.strip() for t in cards_text if t.strip()
+                        )
+                        if materials_text:
+                            # 存到 materials_content（sandbox mock 卡片文本作为材料包内容）
+                            contract["materials"].append({
+                                "kind": "sandbox_card",
+                                "file_name": f"{contract_id}_materials.txt",
+                                "text": materials_text,
+                            })
+                            downloaded += 1
+                            logger.info(
+                                "platform_collector: 抓到卡片文本 %d 字 (%s)",
+                                len(materials_text), contract_id,
+                            )
+                        # dump 一次（i==0）方便调试
+                        if i == 0:
+                            _dump_page(
+                                page,
+                                download_dir.parent / "after_click_dump.html",
+                                "首行 click 后",
+                            )
                     except Exception as e:
-                        logger.warning("材料下载失败 %s: %s", contract_id, e)
+                        logger.warning("抓卡片失败 %s: %s", contract_id, e)
                 else:
-                    logger.info(
-                        "platform_collector: sandbox 未部署「续聘材料包」按钮，"
-                        "跳过材料下载（仅采集表格列）",
-                    )
-                    # 仅警告一次
                     if i == 0:
-                        logger.warning(
-                            "platform_collector: 提示：材料包缺失时，"
-                            "parse-materials CLI 将标记为「无材料」"
+                        logger.info(
+                            "platform_collector: sandbox 未部署「续聘材料包」按钮，"
+                            "跳过材料抓取（仅采集表格列）",
                         )
 
                 contracts.append(contract)
@@ -413,27 +495,56 @@ def _persist_contracts(contracts: list[dict]) -> int:
                     ),
                 )
 
-                # 写 materials_content（占位：file_path 写入，extracted_text 待 parse-materials）
+                # 写 materials_content（sandbox card 文本 + parse-materials 解析后文件路径）
                 from db.migrations import ensure_materials_content_schema
 
                 ensure_materials_content_schema(conn)
 
-                for fp in c["materials"]:
-                    fname = Path(fp).name
-                    conn.execute(
-                        """
-                        INSERT OR REPLACE INTO materials_content
-                            (contract_id, file_name, file_path, extracted_text,
-                             parse_method, parsed_at)
-                        VALUES (?, ?, ?, NULL, 'pending', ?)
-                        """,
-                        (
-                            c["contract_id"],
-                            fname,
-                            fp,
-                            datetime.now().isoformat(timespec="seconds"),
-                        ),
-                    )
+                for m in c["materials"]:
+                    if isinstance(m, dict) and m.get("kind") == "sandbox_card":
+                        # sandbox mock 卡片文本，直接写入 extracted_text
+                        conn.execute(
+                            """
+                            INSERT OR REPLACE INTO materials_content
+                                (contract_id, file_name, file_path, extracted_text,
+                                 parse_method, parsed_at)
+                            VALUES (?, ?, ?, ?, 'sandbox_card', ?)
+                            """,
+                            (
+                                c["contract_id"],
+                                m["file_name"],
+                                f"sandbox://hr/contracts/{c['contract_id']}",
+                                m["text"],
+                                datetime.now().isoformat(timespec="seconds"),
+                            ),
+                        )
+                        # 同步解析 → 抽 year + 等次 → 写入 performance_records
+                        perf_n = repo.parse_sandbox_card_performance(
+                            conn, c["contract_id"], m["text"],
+                        )
+                        if perf_n > 0:
+                            logger.info(
+                                "platform_collector: %s 解析卡片 → performance %d 条",
+                                c["contract_id"], perf_n,
+                            )
+                    else:
+                        # 本地文件路径（parse-materials 解析后写入）
+                        fp = m if isinstance(m, str) else m.get("file_path")
+                        fname = Path(fp).name if fp else "unknown"
+                        conn.execute(
+                            """
+                            INSERT OR REPLACE INTO materials_content
+                                (contract_id, file_name, file_path, extracted_text,
+                                 parse_method, parsed_at)
+                            VALUES (?, ?, ?, NULL, 'pending', ?)
+                            """,
+                            (
+                                c["contract_id"],
+                                fname,
+                                fp,
+                                datetime.now().isoformat(timespec="seconds"),
+                            ),
+                        )
         return len(contracts)
     finally:
         conn.close()
@@ -450,11 +561,14 @@ def _error_result(reason: str, fetched: int) -> dict:
     }
 
 
-def _gen_contract_id(teacher: str, end_date: str) -> str:
-    """从 (姓名, 到期日) 派 contract_id，确保幂等。"""
-    safe_teacher = re.sub(r"[^\w]", "", teacher)[:8] or "T"
+def _gen_contract_id(emp_id: str, end_date: str) -> str:
+    """从 (工号, 到期日) 派 contract_id，确保幂等。
+
+    sandbox v1.1.1 起改用「工号」作主键（之前用姓名）。
+    """
+    safe_emp = re.sub(r"[^\w]", "", emp_id)[:10] or "T"
     safe_end = end_date.replace("-", "")[:8]
-    return f"RPA-{safe_teacher}-{safe_end}"
+    return f"RPA-{safe_emp}-{safe_end}"
 
 
 def _normalize_date(s: str) -> str:

scripts/util/constants.py

@@ -6,7 +6,7 @@
 """
 
 SKILL_SLUG = "alert-contract-expiring-hr"
-SKILL_VERSION = "1.1.1"
+SKILL_VERSION = "1.1.5"
 LOG_LOGGER_NAME = "openclaw.skill.alert_contract_expiring_hr"
 PLATFORM_KIT_MIN_VERSION = "1.2.11"
 

tests/run_tests.py

@@ -0,0 +1,612 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""默认测试套件：合同到期智能预警与续聘辅助。
+
+设计原则：
+- 仅用标准库（不依赖 pytest / jinja2），CI 与本地均能直接 `python tests/run_tests.py` 跑
+- 用 sqlite3 ":memory:" 隔离 DB，不污染仓库 data/ 目录
+- 覆盖：util.date_util 纯函数 / util.constants 版本对齐 / repository 新增功能
+  （parse_sandbox_card_performance + upsert_performance_record）/ scanner 分级 / task_generator
+  规则版 / html_renderer 降级路径
+- 退出码：0 = 全部通过，1 = 有失败
+- -v 开关：详细输出每个用例的状态
+
+release.ps1 在打 tag 前会自动跑 `python tests/run_tests.py -v`，任何一个
+test_* 返回 (False, msg) 即 abort 发布。
+"""
+
+from __future__ import annotations
+
+import argparse
+import os
+import re
+import shutil
+import sqlite3
+import sys
+import tempfile
+import traceback
+from datetime import date
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any, Callable
+
+# 让 scripts/ 下的模块可导入
+_REPO_ROOT = Path(__file__).resolve().parent.parent
+_SCRIPTS_DIR = _REPO_ROOT / "scripts"
+sys.path.insert(0, str(_SCRIPTS_DIR))
+
+
+# ---------- 测试基础设施 ----------
+
+# 测试结果收集器
+_RESULTS: list[tuple[str, bool, str]] = []
+_VERBOSE = False
+
+
+def _run(name: str, fn: Callable[[], tuple[bool, str]]) -> None:
+    """运行单个 test_*，记录结果。"""
+    try:
+        ok, msg = fn()
+    except Exception as e:  # 测试本身抛异常也算失败
+        tb = traceback.format_exc(limit=3)
+        ok, msg = False, f"exception: {e}\n{tb}"
+    _RESULTS.append((name, ok, msg))
+    if _VERBOSE or not ok:
+        marker = "PASS" if ok else "FAIL"
+        print(f"  [{marker}] {name}: {msg}", flush=True)
+
+
+def _make_conn() -> sqlite3.Connection:
+    """建一张 in-memory DB，应用所有 DDL + 迁移（不加载 seed）。"""
+    from db.connection import DDL_STATEMENTS
+    from db.migrations import apply_all
+
+    conn = sqlite3.connect(":memory:")
+    conn.row_factory = sqlite3.Row
+    conn.execute("PRAGMA foreign_keys = ON;")
+    for ddl in DDL_STATEMENTS:
+        conn.execute(ddl)
+    apply_all(conn)
+    conn.commit()
+    return conn
+
+
+class _FakeRow:
+    """Mock 一行 sqlite3.Row：同时支持下标访问 (r["x"]) 和属性访问 (r.x)。
+
+    production 代码里 task_generator.suggest() 用的是 r["moral_teaching_score"]
+    下标语法，所以 fake row 必须也支持。
+    """
+
+    __slots__ = ("year", "performance_score", "moral_teaching_score",
+                 "research_score", "teaching_score")
+
+    def __init__(self, year: int, score: float, moral: float):
+        self.year = year
+        self.performance_score = score
+        self.moral_teaching_score = moral
+        self.research_score = score
+        self.teaching_score = score
+
+    def __getitem__(self, key: str) -> Any:
+        return getattr(self, key)
+
+
+def _fake_perf(year: int, score: float, moral: float | None = None) -> _FakeRow:
+    """构造一个 performance_records-like 行（用于纯函数测试，不入 DB）。"""
+    m = moral if moral is not None else score
+    return _FakeRow(year=year, score=score, moral=m)
+
+
+def _insert_stub_contract(conn: sqlite3.Connection, contract_id: str) -> None:
+    """插入一份占位合同（performance_records 的 FK 目标）。"""
+    conn.execute(
+        """INSERT INTO contracts
+           (contract_id, employee_id, teacher_name, unit_id, unit_name,
+            position, contract_type, contract_start, contract_end,
+            trial_period_end, status, last_material_id, renewed_at)
+           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '可续聘', NULL, NULL)""",
+        (contract_id, f"E-{contract_id}", f"占位-{contract_id}", "U-TEST",
+         "测试系", "讲师", "正式", "2022-01-01", "2099-12-31", "2022-07-01"),
+    )
+    conn.commit()
+
+
+# ---------- util.date_util ----------
+
+def test_parse_iso_date() -> tuple[bool, str]:
+    from util.date_util import parse_iso_date
+    cases = [
+        ("2024-03-15", date(2024, 3, 15)),
+        ("2024-03-15T08:30:00", date(2024, 3, 15)),
+        ("  2024-03-15  ", date(2024, 3, 15)),
+        (date(2024, 3, 15), date(2024, 3, 15)),  # 已为 date
+    ]
+    for inp, want in cases:
+        got = parse_iso_date(inp)
+        if got != want:
+            return False, f"parse_iso_date({inp!r}) = {got}, want {want}"
+    return True, "4 cases ok"
+
+
+def test_days_to() -> tuple[bool, str]:
+    from util.date_util import days_to
+    if days_to(date(2024, 3, 15), today=date(2024, 1, 1)) != 74:
+        return False, "days_to 正向计算错"
+    if days_to(date(2023, 12, 31), today=date(2024, 1, 1)) != -1:
+        return False, "days_to 已过期返回负数错"
+    # 默认 today=今天
+    if days_to(date(2099, 1, 1)) <= 0:
+        return False, "days_to 默认 today 应为今天"
+    return True, "正/负/默认 today 三种场景"
+
+
+def test_classify_alert() -> tuple[bool, str]:
+    from util.date_util import classify_alert
+    t = (90, 60, 30)
+    cases = [
+        (100, None),    # 超出预警范围
+        (91,  None),    # 边界外
+        (90,  "yellow"),
+        (61,  "yellow"),
+        (60,  "orange"),
+        (31,  "orange"),
+        (30,  "red"),
+        (1,   "red"),
+        (0,   "red"),
+        (-5,  "red"),   # 已过期仍属 red
+    ]
+    for days, want in cases:
+        got = classify_alert(days, t)
+        if got != want:
+            return False, f"classify_alert({days}) = {got}, want {want}"
+    return True, "10 边界 + 越界场景"
+
+
+def test_month_match() -> tuple[bool, str]:
+    from util.date_util import month_match
+    d = date(2024, 3, 15)
+    if not month_match(d, "2024-03"):
+        return False, "同月应命中"
+    if month_match(d, "2024-04"):
+        return False, "不同月不应命中"
+    if not month_match(d, ""):
+        return False, "空 month_str 应命中全部"
+    if month_match(d, "garbage"):
+        return False, "非法 month_str 不应命中"
+    return True, "命中 / 未命中 / 空 / 非法 四种"
+
+
+# ---------- util.constants 版本对齐（POLICY-RELEASE-001） ----------
+
+def test_skill_md_version_alignment() -> tuple[bool, str]:
+    """POLICY-RELEASE-001：SKILL.md version == SKILL_VERSION == CHANGELOG 顶部版本"""
+    from util.constants import SKILL_VERSION
+
+    skill_md = (_REPO_ROOT / "SKILL.md").read_text(encoding="utf-8")
+    m = re.search(r'(?m)^version:\s*["\']?(\d+\.\d+\.\d+)', skill_md)
+    if not m:
+        return False, "SKILL.md 缺顶层 version: X.Y.Z"
+    md_ver = m.group(1)
+    if md_ver != SKILL_VERSION:
+        return False, f"SKILL.md={md_ver} != SKILL_VERSION={SKILL_VERSION}"
+
+    # CHANGELOG 顶部应有 ## X.Y.Z
+    cl = (_REPO_ROOT / "CHANGELOG.md").read_text(encoding="utf-8")
+    cl_m = re.search(r'(?m)^##\s+(\d+\.\d+\.\d+)', cl)
+    if not cl_m:
+        return False, "CHANGELOG.md 缺顶层 ## X.Y.Z"
+    cl_ver = cl_m.group(1)
+    if cl_ver != SKILL_VERSION:
+        return False, f"CHANGELOG 顶部={cl_ver} != SKILL_VERSION={SKILL_VERSION}"
+
+    return True, f"SKILL.md={md_ver} == CHANGELOG={cl_ver} == SKILL_VERSION={SKILL_VERSION}"
+
+
+def test_skill_md_slug_alignment() -> tuple[bool, str]:
+    """SKILL.md metadata.openclaw.slug == SKILL_SLUG"""
+    from util.constants import SKILL_SLUG
+
+    skill_md = (_REPO_ROOT / "SKILL.md").read_text(encoding="utf-8")
+    m = re.search(r'(?m)^\s*slug:\s*["\']?([a-z0-9_-]+)', skill_md)
+    if not m:
+        return False, "SKILL.md 缺 metadata.openclaw.slug"
+    md_slug = m.group(1)
+    if md_slug != SKILL_SLUG:
+        return False, f"SKILL.md slug={md_slug} != SKILL_SLUG={SKILL_SLUG}"
+    return True, f"slug={md_slug}"
+
+
+# ---------- db.repository：v1.1.2 新增的 sandbox card 解析 ----------
+
+def test_parse_sandbox_card_row_format() -> tuple[bool, str]:
+    """v1.1.2：行格式 '2024\\t称职'"""
+    from db import repository as repo
+
+    conn = _make_conn()
+    _insert_stub_contract(conn, "C001")
+    card_text = "year\tresult\n2023\t优秀\n2024\t称职\n2025\t基本称职\n"
+    n = repo.parse_sandbox_card_performance(conn, "C001", card_text)
+    if n != 3:
+        return False, f"写入 {n} 条，期望 3"
+
+    rows = repo.list_performance_by_contract(conn, "C001")
+    if len(rows) != 3:
+        return False, f"DB 行数 {len(rows)}，期望 3"
+    by_year = {r["year"]: r["performance_score"] for r in rows}
+    expected = {2023: 90.0, 2024: 75.0, 2025: 60.0}
+    for y, want in expected.items():
+        if by_year.get(y) != want:
+            return False, f"{y} 年分数={by_year.get(y)}，期望 {want}"
+    return True, "3 行（优秀/称职/基本称职）解析 + 写入正确"
+
+
+def test_parse_sandbox_card_narrative_format() -> tuple[bool, str]:
+    """v1.1.2：叙述式 '2024年：等次「称职」'"""
+    from db import repository as repo
+
+    conn = _make_conn()
+    _insert_stub_contract(conn, "C002")
+    card_text = "考核情况：\n2023年：等次「优秀」。\n2024年：等次「称职」。\n2025年：等次「不称职」。\n"
+    n = repo.parse_sandbox_card_performance(conn, "C002", card_text)
+    if n != 3:
+        return False, f"写入 {n} 条，期望 3"
+
+    rows = repo.list_performance_by_contract(conn, "C002")
+    by_year = {r["year"]: r["performance_score"] for r in rows}
+    expected = {2023: 90.0, 2024: 75.0, 2025: 30.0}
+    for y, want in expected.items():
+        if by_year.get(y) != want:
+            return False, f"{y} 年分数={by_year.get(y)}，期望 {want}"
+    return True, "3 行（优秀/称职/不称职）叙述式解析正确"
+
+
+def test_parse_sandbox_card_empty_and_invalid() -> tuple[bool, str]:
+    """空字符串 / 无匹配应返回 0，不报错"""
+    from db import repository as repo
+
+    conn = _make_conn()
+    _insert_stub_contract(conn, "C001")
+    if repo.parse_sandbox_card_performance(conn, "C001", "") != 0:
+        return False, "空字符串应返回 0"
+    if repo.parse_sandbox_card_performance(conn, "C001", "没有任何年份/等次信息") != 0:
+        return False, "无匹配应返回 0"
+    rows = repo.list_performance_by_contract(conn, "C001")
+    if rows:
+        return False, "无匹配不应写入任何行"
+    return True, "空 / 无匹配场景"
+
+
+def test_parse_sandbox_card_mixed_formats() -> tuple[bool, str]:
+    """同卡片混用两种格式时按行去重（PK = P-<contract>-<year>）"""
+    from db import repository as repo
+
+    conn = _make_conn()
+    _insert_stub_contract(conn, "C003")
+    card_text = (
+        "2024年：等次「称职」。\n"   # 叙述式
+        "2025\t优秀\n"              # 行格式
+        "2026年：等次「基本称职」\n"
+    )
+    n = repo.parse_sandbox_card_performance(conn, "C003", card_text)
+    if n != 3:
+        return False, f"混合解析写入 {n} 条，期望 3"
+    return True, "叙述式 + 行格式混合解析"
+
+
+def test_upsert_performance_record_grade_mapping() -> tuple[bool, str]:
+    """grade → score 映射 + ON CONFLICT upsert"""
+    from db import repository as repo
+
+    conn = _make_conn()
+    _insert_stub_contract(conn, "C001")
+    if not repo.upsert_performance_record(conn, "C001", 2024, "优秀"):
+        return False, "优秀应返回 True"
+    if not repo.upsert_performance_record(conn, "C001", 2024, "优秀"):
+        return False, "upsert 已存在应仍返回 True"
+    if not repo.upsert_performance_record(conn, "C001", 2025, "称职"):
+        return False, "称职应返回 True"
+    if not repo.upsert_performance_record(conn, "C001", 2026, "基本称职"):
+        return False, "基本称职应返回 True"
+    if not repo.upsert_performance_record(conn, "C001", 2027, "不称职"):
+        return False, "不称职应返回 True"
+    if not repo.upsert_performance_record(conn, "C001", 2028, "不合格"):
+        return False, "不合格应返回 True"
+    if repo.upsert_performance_record(conn, "C001", 2029, "未知等次"):
+        return False, "未知等次应返回 False"
+
+    rows = repo.list_performance_by_contract(conn, "C001")
+    # 期望 5 行：2024/2025/2026/2027/2028
+    # 第二次 (2024, 优秀) 是 upsert 覆盖不新增；2029 未知等次不入库
+    if len(rows) != 5:
+        return False, f"DB 行数 {len(rows)}，期望 5（upsert 同年不新增；未知等次不入库）"
+    by_year = {r["year"]: r["performance_score"] for r in rows}
+    expected = {2024: 90.0, 2025: 75.0, 2026: 60.0, 2027: 30.0, 2028: 30.0}
+    for y, want in expected.items():
+        if by_year.get(y) != want:
+            return False, f"{y} 年分数={by_year.get(y)}，期望 {want}"
+    # 验证 2024 的分数确实被 upsert 覆盖（不是 30/60/75 等其他值）
+    if by_year[2024] != 90.0:
+        return False, f"upsert 未保持分数：2024={by_year[2024]}，want 90.0"
+    return True, "grade 映射 + upsert 幂等 + 未知等次返回 False"
+
+
+# ---------- service.task_generator.suggest 规则版 ----------
+
+def test_suggest_empty_records() -> tuple[bool, str]:
+    """无业绩记录 → observe（数据缺失）"""
+    from service.task_generator import suggest
+    s, r = suggest([])
+    if s != "observe":
+        return False, f"suggest([]) = {s}, want observe"
+    if "缺失" not in r:
+        return False, f"reason 应提示缺失，实际 {r!r}"
+    return True, "observe + '缺失'"
+
+
+def test_suggest_low_moral() -> tuple[bool, str]:
+    """师德 < 60 → non_renew"""
+    from service.task_generator import suggest
+    rows = [_fake_perf(2024, 85, moral=50)]
+    s, r = suggest(rows)
+    if s != "non_renew":
+        return False, f"suggest(师德 50) = {s}, want non_renew"
+    if "师德" not in r:
+        return False, f"reason 应含「师德」，实际 {r!r}"
+    return True, "non_renew + '师德'"
+
+
+def test_suggest_fail_year() -> tuple[bool, str]:
+    """任意一年考核 < 60 → observe（复核）"""
+    from service.task_generator import suggest
+    rows = [
+        _fake_perf(2023, 80, moral=85),
+        _fake_perf(2024, 50, moral=85),  # 不及格
+        _fake_perf(2025, 85, moral=85),
+    ]
+    s, r = suggest(rows)
+    if s != "observe":
+        return False, f"suggest(有不及格年) = {s}, want observe"
+    if "复核" not in r and "不合格" not in r:
+        return False, f"reason 应提示不合格/复核，实际 {r!r}"
+    return True, "observe + '不合格'"
+
+
+def test_suggest_renew_upward() -> tuple[bool, str]:
+    """全部 >= 80 且逐年上升 → renew"""
+    from service.task_generator import suggest
+    rows = [
+        _fake_perf(2023, 80, moral=85),
+        _fake_perf(2024, 85, moral=85),
+        _fake_perf(2025, 90, moral=85),
+    ]
+    s, r = suggest(rows)
+    if s != "renew":
+        return False, f"suggest(全部 80+ 上升) = {s}, want renew"
+    return True, "renew + 上升趋势"
+
+
+def test_suggest_observe_default() -> tuple[bool, str]:
+    """达标但非逐年上升 → observe（常规评审）"""
+    from service.task_generator import suggest
+    rows = [
+        _fake_perf(2023, 85, moral=85),
+        _fake_perf(2024, 80, moral=85),  # 回落
+        _fake_perf(2025, 85, moral=85),
+    ]
+    s, r = suggest(rows)
+    if s != "observe":
+        return False, f"suggest(回落) = {s}, want observe"
+    return True, "observe（常规评审）"
+
+
+# ---------- service.scanner 端到端 ----------
+
+def test_scanner_no_alerts() -> tuple[bool, str]:
+    """所有合同都在 90 天外 → 0 预警"""
+    from service import scanner
+
+    conn = _make_conn()
+    conn.executemany(
+        """INSERT INTO contracts
+           (contract_id, employee_id, teacher_name, unit_id, unit_name,
+            position, contract_type, contract_start, contract_end,
+            trial_period_end, status, last_material_id, renewed_at)
+           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '可续聘', NULL, NULL)""",
+        [
+            ("C-FAR-1", "T001", "张三", "U1", "数学系", "教授", "正式",
+             "2022-01-01", "2099-12-31", "2022-07-01"),
+        ],
+    )
+    conn.commit()
+    result = scanner.scan(conn, today=date(2024, 1, 15))
+    if result["scanned_count"] != 1:
+        return False, f"scanned={result['scanned_count']}, want 1"
+    if result["yellow_count"] + result["orange_count"] + result["red_count"] != 0:
+        return False, f"应无预警，实测 yellow={result['yellow_count']} orange={result['orange_count']} red={result['red_count']}"
+    return True, "1 合同全在 90 天外 → 0 预警"
+
+
+def test_scanner_three_levels() -> tuple[bool, str]:
+    """3 份合同分别命中黄/橙/红"""
+    from service import scanner
+
+    conn = _make_conn()
+    conn.executemany(
+        """INSERT INTO contracts
+           (contract_id, employee_id, teacher_name, unit_id, unit_name,
+            position, contract_type, contract_start, contract_end,
+            trial_period_end, status, last_material_id, renewed_at)
+           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '可续聘', NULL, NULL)""",
+        [
+            ("C-YEL", "T001", "黄老师", "U1", "数学系", "教授", "正式",
+             "2022-01-01", "2024-05-15", "2022-07-01"),  # 距 2024-02-15 = 90 天 → yellow
+            ("C-ORA", "T002", "橙老师", "U1", "物理系", "讲师", "正式",
+             "2022-01-01", "2024-04-15", "2022-07-01"),  # 距 2024-02-15 = 60 天 → orange
+            ("C-RED", "T003", "红老师", "U1", "化学系", "副教授", "正式",
+             "2022-01-01", "2024-03-15", "2022-07-01"),  # 距 2024-02-15 = 29 天 → red
+        ],
+    )
+    conn.commit()
+    result = scanner.scan(conn, today=date(2024, 2, 15))
+    if result["scanned_count"] != 3:
+        return False, f"scanned={result['scanned_count']}, want 3"
+    if result["yellow_count"] != 1 or result["orange_count"] != 1 or result["red_count"] != 1:
+        return False, (f"分级数 yellow={result['yellow_count']} "
+                       f"orange={result['orange_count']} red={result['red_count']}, "
+                       f"want 1/1/1")
+
+    # 验证 alerts 表实际写入
+    alerts = conn.execute("SELECT contract_id, alert_level FROM alerts").fetchall()
+    by_cid = {a["contract_id"]: a["alert_level"] for a in alerts}
+    if by_cid.get("C-YEL") != "yellow":
+        return False, f"C-YEL 写入等级 {by_cid.get('C-YEL')}, want yellow"
+    if by_cid.get("C-ORA") != "orange":
+        return False, f"C-ORA 写入等级 {by_cid.get('C-ORA')}, want orange"
+    if by_cid.get("C-RED") != "red":
+        return False, f"C-RED 写入等级 {by_cid.get('C-RED')}, want red"
+    return True, "3 合同 → 1 黄 + 1 橙 + 1 红，alerts 表写入正确"
+
+
+def test_scanner_skips_non_active_contracts() -> tuple[bool, str]:
+    """status != '可续聘' 的合同不应被扫描"""
+    from service import scanner
+
+    conn = _make_conn()
+    conn.execute(
+        """INSERT INTO contracts
+           (contract_id, employee_id, teacher_name, unit_id, unit_name,
+            position, contract_type, contract_start, contract_end,
+            trial_period_end, status, last_material_id, renewed_at)
+           VALUES ('C-DONE', 'T001', '已续签老师', 'U1', '数学系', '教授', '正式',
+                   '2022-01-01', '2024-03-15', '2022-07-01', '已续签', NULL, NULL)"""
+    )
+    conn.commit()
+    result = scanner.scan(conn, today=date(2024, 2, 15))
+    if result["scanned_count"] != 0:
+        return False, f"已续签合同不应被扫描，scanned={result['scanned_count']}"
+    alerts = conn.execute("SELECT COUNT(*) AS n FROM alerts").fetchone()["n"]
+    if alerts != 0:
+        return False, f"应无 alerts 写入，实测 {alerts}"
+    return True, "已续签合同被跳过"
+
+
+# ---------- service.html_renderer 降级路径 ----------
+
+def test_html_render_basic() -> tuple[bool, str]:
+    """HTML 渲染（无 jinja2 走 str.replace 降级路径）"""
+    from service import html_renderer
+
+    tmpdir = Path(tempfile.mkdtemp(prefix="alert-hr-test-"))
+    saved = os.environ.get("HTML_OUTPUT_DIR")
+    os.environ["HTML_OUTPUT_DIR"] = str(tmpdir)
+    try:
+        conn = _make_conn()
+        conn.execute(
+            """INSERT INTO contracts
+               (contract_id, employee_id, teacher_name, unit_id, unit_name,
+                position, contract_type, contract_start, contract_end,
+                trial_period_end, status, last_material_id, renewed_at)
+               VALUES ('C-HTML', 'T100', '渲染测试', 'U9', '测试系', '教授', '正式',
+                       '2022-01-01', '2024-06-30', '2022-07-01', '可续聘', NULL, NULL)"""
+        )
+        conn.execute(
+            """INSERT INTO alerts
+               VALUES ('A-HTML', 'C-HTML', 'orange', 60, 'test',
+                       '2024-04-01T00:00:00', 'open')"""
+        )
+        conn.execute(
+            """INSERT INTO renewal_tasks
+               VALUES ('T-HTML', 'C-HTML', 'A-HTML', 'renew', '近 3 年考核良好',
+                       '业绩良好', '["材料 A", "材料 B"]',
+                       '2024-04-01T00:00:00', 'pending',
+                       'rules', '')"""
+        )
+        conn.commit()
+
+        path = html_renderer.render_task_html(conn, "T-HTML")
+        if not path.exists():
+            return False, f"输出文件不存在 {path}"
+
+        text = path.read_text(encoding="utf-8")
+        if "渲染测试" not in text:
+            return False, "HTML 应含教师姓名"
+        if "续聘准备任务书" not in text:
+            return False, "HTML 应含 TITLE"
+        if "橙色" not in text:
+            return False, "HTML 应含「橙色」预警等级标签"
+        if "建议续聘" not in text:
+            return False, "HTML 应含「建议续聘」建议标签"
+        return True, f"HTML 渲染 OK → {path.name}"
+    finally:
+        if saved is None:
+            os.environ.pop("HTML_OUTPUT_DIR", None)
+        else:
+            os.environ["HTML_OUTPUT_DIR"] = saved
+        shutil.rmtree(tmpdir, ignore_errors=True)
+
+
+def test_html_render_unknown_task_raises() -> tuple[bool, str]:
+    """渲染不存在的 task_id 应抛 ValueError"""
+    from service import html_renderer
+
+    conn = _make_conn()
+    try:
+        html_renderer.render_task_html(conn, "T-NONEXISTENT")
+    except ValueError as e:
+        return True, f"ValueError OK: {e}"
+    return False, "应抛 ValueError，实际未抛"
+
+
+# ---------- 入口 ----------
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="默认 release 测试套件")
+    parser.add_argument("-v", "--verbose", action="store_true",
+                        help="详细输出每个用例结果")
+    args = parser.parse_args()
+
+    global _VERBOSE
+    _VERBOSE = args.verbose
+
+    print("=== alert-contract-expiring-hr release tests ===", flush=True)
+
+    _run("util/parse_iso_date",        test_parse_iso_date)
+    _run("util/days_to",               test_days_to)
+    _run("util/classify_alert",        test_classify_alert)
+    _run("util/month_match",           test_month_match)
+    _run("constants/skill_md_version_alignment", test_skill_md_version_alignment)
+    _run("constants/skill_md_slug_alignment",    test_skill_md_slug_alignment)
+    _run("db/parse_sandbox_card_row",          test_parse_sandbox_card_row_format)
+    _run("db/parse_sandbox_card_narrative",    test_parse_sandbox_card_narrative_format)
+    _run("db/parse_sandbox_card_empty_invalid", test_parse_sandbox_card_empty_and_invalid)
+    _run("db/parse_sandbox_card_mixed",         test_parse_sandbox_card_mixed_formats)
+    _run("db/upsert_performance_record",       test_upsert_performance_record_grade_mapping)
+    _run("task_generator/suggest_empty",       test_suggest_empty_records)
+    _run("task_generator/suggest_low_moral",   test_suggest_low_moral)
+    _run("task_generator/suggest_fail_year",   test_suggest_fail_year)
+    _run("task_generator/suggest_renew",       test_suggest_renew_upward)
+    _run("task_generator/suggest_observe",     test_suggest_observe_default)
+    _run("scanner/no_alerts",                 test_scanner_no_alerts)
+    _run("scanner/three_levels",              test_scanner_three_levels)
+    _run("scanner/skips_non_active",          test_scanner_skips_non_active_contracts)
+    _run("html_renderer/basic",               test_html_render_basic)
+    _run("html_renderer/unknown_task_raises", test_html_render_unknown_task_raises)
+
+    total = len(_RESULTS)
+    passed = sum(1 for _, ok, _ in _RESULTS if ok)
+    failed = total - passed
+
+    print()
+    print(f"=== Summary: {passed}/{total} passed, {failed} failed ===", flush=True)
+    if failed:
+        print("\nFailures:")
+        for name, ok, msg in _RESULTS:
+            if not ok:
+                print(f"  - {name}: {msg}")
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())