daily-robots/docs/superpowers/plans/2026-07-09-wecom-delta-mode.md

# 企微早报 Delta 模式 Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** 将企微早报从全量 Top 榜复印改为 Delta 变化通知：列表仅展示新入榜、新闻 7 天 link 去重、满足推送闸门才 webhook，榜全稳且无新新闻时静默。

**Architecture:** 在现有 `daily/delta.py` movement 数据之上，新增 `effective_wecom_mode`（含首日 baseline fallback）、`format_wecom` delta 渲染、`daily/news/pushed_links.py` 新闻去重缓存、`daily/push_gate.py` 推送闸门。`generate.py` 串联后把闸门结果写入 `output/{date}.data.json` 的 `meta.push_gate`；`webhook.py` 与 `run-daily.ps1` 推送前读取并跳过静默日。Agent 模式仍由 Python 插入榜单区块，更新 `skills/daily-agent/SKILL.md` 与 delta 对齐。

**Tech Stack:** Python 3.13+、stdlib `unittest` + `pytest`（仓库现有 `tests/test_featured_pick.py` 用 unittest）、`python-dotenv`、`httpx`

## Global Constraints

- `DAILY_WECOM_MODE` 合法值：`full` | `delta`；默认 `delta`
- `DAILY_DELTA_BASELINE_FALLBACK` 合法值：`full` | `empty`；默认 `full`
- `DAILY_SKIP_PUSH_WHEN_SILENT` 默认 `1`（真）
- `DAILY_FORCE_PUSH=1` 时忽略静默，强制推送
- `DAILY_NEWS_DEDUP_DAYS` 默认 `7`
- `DAILY_AI_NEWS_HOURS` 默认改为 `24`（`.env.example`）
- 新闻去重 cache 路径：`CACHE_DIR / "pushed-news-links.json"`（即 `.cache/pushed-news-links.json`，已在 `.gitignore`）
- 静默日：仍写 `output/{date}.md`、`output/{date}.wecom.md`、`output/{date}.data.json`，仅跳过 webhook
- 跨榜去重：同一 `skill_id` 在 Trending/Hot moves 均出现时，只在 Trending 区块展示一条，`badge` 为 `Trending #4 · Hot #2`
- 不做：静态页、今日一装、轮换版面、榜首锚点
- Classic 与 Agent 模式共用 `format_wecom` delta 路径

---

## File Structure

| 文件 | 职责 |
|------|------|
| `daily/config.py` | `env_bool`、`wecom_mode()`、`news_dedup_days()`、`skip_push_when_silent()`、`delta_baseline_fallback()`、`force_push()` |
| `daily/news/pushed_links.py` | **新建** — 已推送新闻 link 缓存读写与过滤 |
| `daily/delta.py` | `effective_wecom_mode()`、`partition_skill_moves_for_wecom()` 跨榜去重 |
| `daily/push_gate.py` | **新建** — `PushGateResult`、`evaluate_push_gate()` |
| `daily/format_wecom.py` | delta 列表渲染、`replace_wecom_board_sections()` 替换 Skills+GitHub |
| `daily/generate.py` | 串联 effective mode、新闻去重、push gate、写 meta |
| `daily/webhook.py` | 推送前读 `meta.push_gate` |
| `daily/report_data.py` | `llm_input` 增加 `wecom_mode`、`effective_wecom_mode`、`push_gate` |
| `skills/daily-agent/SKILL.md` | delta 模式 Agent 规则 |
| `.env.example` | 新 env 文档 |
| `run-daily.ps1` | 静默日跳过 push |
| `tests/test_wecom_delta.py` | **新建** — 单元测试 |

---

### Task 1: Config helpers

**Files:**
- Modify: `daily/config.py`（文件末尾追加）
- Test: `tests/test_wecom_delta.py`

**Interfaces:**
- Consumes: 无
- Produces:
  - `env_bool(key: str, default: bool) -> bool`
  - `wecom_mode() -> Literal["full", "delta"]`
  - `news_dedup_days() -> int`
  - `skip_push_when_silent() -> bool`
  - `delta_baseline_fallback() -> Literal["full", "empty"]`
  - `force_push() -> bool`

- [ ] **Step 1: Write the failing test**

```python
# tests/test_wecom_delta.py
"""Tests for WeCom delta mode."""

from __future__ import annotations

import os
import unittest
from unittest.mock import patch

from daily.config import (
    delta_baseline_fallback,
    env_bool,
    force_push,
    news_dedup_days,
    skip_push_when_silent,
    wecom_mode,
)


class ConfigHelpersTests(unittest.TestCase):
    def test_wecom_mode_defaults_delta(self):
        with patch.dict(os.environ, {}, clear=True):
            self.assertEqual(wecom_mode(), "delta")

    def test_wecom_mode_full(self):
        with patch.dict(os.environ, {"DAILY_WECOM_MODE": "full"}, clear=True):
            self.assertEqual(wecom_mode(), "full")

    def test_env_bool_truthy(self):
        with patch.dict(os.environ, {"DAILY_FORCE_PUSH": "1"}, clear=True):
            self.assertTrue(env_bool("DAILY_FORCE_PUSH", False))

    def test_skip_push_when_silent_default(self):
        with patch.dict(os.environ, {}, clear=True):
            self.assertTrue(skip_push_when_silent())

    def test_news_dedup_days_default(self):
        with patch.dict(os.environ, {}, clear=True):
            self.assertEqual(news_dedup_days(), 7)

    def test_delta_baseline_fallback_default(self):
        with patch.dict(os.environ, {}, clear=True):
            self.assertEqual(delta_baseline_fallback(), "full")
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python -m pytest tests/test_wecom_delta.py::ConfigHelpersTests -v`  
Expected: FAIL with `ImportError` or `cannot import name 'wecom_mode'`

- [ ] **Step 3: Write minimal implementation**

在 `daily/config.py` 末尾追加：

```python
def env_bool(key: str, default: bool) -> bool:
    raw = env(key)
    if raw is None:
        return default
    return raw.strip().lower() in {"1", "true", "yes", "on"}


def wecom_mode() -> str:
    raw = (env("DAILY_WECOM_MODE") or "delta").strip().lower()
    return raw if raw in {"full", "delta"} else "delta"


def news_dedup_days() -> int:
    return max(1, env_int("DAILY_NEWS_DEDUP_DAYS", 7))


def skip_push_when_silent() -> bool:
    return env_bool("DAILY_SKIP_PUSH_WHEN_SILENT", True)


def delta_baseline_fallback() -> str:
    raw = (env("DAILY_DELTA_BASELINE_FALLBACK") or "full").strip().lower()
    return raw if raw in {"full", "empty"} else "full"


def force_push() -> bool:
    return env_bool("DAILY_FORCE_PUSH", False)
```

- [ ] **Step 4: Run test to verify it passes**

Run: `python -m pytest tests/test_wecom_delta.py::ConfigHelpersTests -v`  
Expected: PASS（6 passed）

- [ ] **Step 5: Commit**

```bash
git add daily/config.py tests/test_wecom_delta.py
git commit -m "feat: 新增企微 Delta 模式相关配置读取函数"
```

---

### Task 2: News pushed-link dedup cache

**Files:**
- Create: `daily/news/pushed_links.py`
- Modify: `daily/news/fetch.py`（`prepare_wecom_news_items` / `prepare_wecom_cn_news_items` 末尾调用过滤）
- Test: `tests/test_wecom_delta.py`

**Interfaces:**
- Consumes: `daily.config.CACHE_DIR`, `daily.config.news_dedup_days()`, `daily.news.fetch._normalize_link`
- Produces:
  - `load_pushed_links() -> dict[str, list[str]]`  # `{"dates": {"2026-07-08": ["https://..."]}}`
  - `filter_unpushed_items(items: list[dict], *, date_str: str) -> list[dict]`
  - `record_pushed_links(date_str: str, links: list[str]) -> None`

- [ ] **Step 1: Write the failing test**

在 `tests/test_wecom_delta.py` 追加：

```python
import json
import tempfile
from pathlib import Path
from unittest.mock import patch

from daily.news.pushed_links import filter_unpushed_items, record_pushed_links


class NewsPushedLinksTests(unittest.TestCase):
    def test_filter_and_record_roundtrip(self):
        with tempfile.TemporaryDirectory() as tmp:
            cache = Path(tmp) / "pushed-news-links.json"
            items = [
                {"title": "A", "link": "https://example.com/a?utm_source=x"},
                {"title": "B", "link": "https://example.com/b"},
            ]
            with patch("daily.news.pushed_links._cache_path", return_value=cache):
                with patch("daily.news.pushed_links.news_dedup_days", return_value=7):
                    record_pushed_links("2026-07-08", ["https://example.com/a"])
                    out = filter_unpushed_items(items, date_str="2026-07-09")
            self.assertEqual(len(out), 1)
            self.assertEqual(out[0]["link"], "https://example.com/b")
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python -m pytest tests/test_wecom_delta.py::NewsPushedLinksTests -v`  
Expected: FAIL `ModuleNotFoundError: daily.news.pushed_links`

- [ ] **Step 3: Write minimal implementation**

创建 `daily/news/pushed_links.py`：

```python
"""已推送企微早报的新闻 link 去重缓存。"""

from __future__ import annotations

import json
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any

from daily.config import CACHE_DIR, news_dedup_days
from daily.news.fetch import _normalize_link


def _cache_path() -> Path:
    return CACHE_DIR / "pushed-news-links.json"


def _load_raw() -> dict[str, Any]:
    path = _cache_path()
    if not path.exists():
        return {"dates": {}}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return {"dates": {}}
    if not isinstance(data.get("dates"), dict):
        return {"dates": {}}
    return data


def _save_raw(data: dict[str, Any]) -> None:
    path = _cache_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")


def _prune(data: dict[str, Any], *, keep_days: int) -> None:
    dates: dict[str, list[str]] = data.setdefault("dates", {})
    try:
        anchor = max(datetime.strptime(d, "%Y-%m-%d") for d in dates)
    except ValueError:
        return
    cutoff = anchor - timedelta(days=keep_days)
    for key in list(dates.keys()):
        try:
            if datetime.strptime(key, "%Y-%m-%d") < cutoff:
                dates.pop(key, None)
        except ValueError:
            dates.pop(key, None)


def load_pushed_link_set() -> set[str]:
    data = _load_raw()
    out: set[str] = set()
    for links in (data.get("dates") or {}).values():
        if isinstance(links, list):
            out.update(str(x) for x in links if x)
    return out


def filter_unpushed_items(
    items: list[dict[str, Any]],
    *,
    date_str: str,
) -> list[dict[str, Any]]:
    seen = load_pushed_link_set()
    out: list[dict[str, Any]] = []
    for item in items:
        link = _normalize_link(str(item.get("link") or ""))
        if not link or link in seen:
            continue
        out.append(item)
    return out


def record_pushed_links(date_str: str, links: list[str]) -> None:
    data = _load_raw()
    dates: dict[str, list[str]] = data.setdefault("dates", {})
    normalized = []
    for link in links:
        clean = _normalize_link(link)
        if clean:
            normalized.append(clean)
    dates[date_str] = sorted(set(normalized))
    _prune(data, keep_days=news_dedup_days())
    _save_raw(data)
```

修改 `daily/news/fetch.py` — 在 `prepare_wecom_news_items` 返回前：

```python
from daily.news.pushed_links import filter_unpushed_items

def prepare_wecom_news_items(news: dict[str, Any], *, date_str: str | None = None) -> list[dict[str, Any]]:
    # ... 现有逻辑不变，得到 items ...
    if date_str:
        items = filter_unpushed_items(items, date_str=date_str)
    return items
```

对 `prepare_wecom_cn_news_items` 做同样修改（增加可选 `date_str` 参数）。

- [ ] **Step 4: Run test to verify it passes**

Run: `python -m pytest tests/test_wecom_delta.py::NewsPushedLinksTests -v`  
Expected: PASS

- [ ] **Step 5: Commit**

```bash
git add daily/news/pushed_links.py daily/news/fetch.py tests/test_wecom_delta.py
git commit -m "feat: 新增企微已推送新闻 link 去重缓存"
```

---

### Task 3: Cross-board skill move partition

**Files:**
- Modify: `daily/delta.py`（文件末尾追加）
- Test: `tests/test_wecom_delta.py`

**Interfaces:**
- Consumes: `daily.delta.skill_id`
- Produces:
  - `effective_wecom_mode(*, date_str: str, configured_mode: str | None = None) -> str`
  - `partition_skill_moves_for_wecom(trending_moves: list[dict], hot_moves: list[dict]) -> tuple[list[dict], list[dict]]`

- [ ] **Step 1: Write the failing test**

```python
from daily.delta import effective_wecom_mode, partition_skill_moves_for_wecom, skill_id


class SkillMovePartitionTests(unittest.TestCase):
    def test_partition_dedupes_across_boards(self):
        trending = [{"id": "a/b/foo", "rank": 4, "title": "foo", "source": "a/b", "installs": 1, "link": "", "description": ""}]
        hot = [{"id": "a/b/foo", "rank": 2, "title": "foo", "source": "a/b", "installs": 1, "link": "", "description": ""}]
        t_out, h_out = partition_skill_moves_for_wecom(trending, hot)
        self.assertEqual(len(t_out), 1)
        self.assertEqual(len(h_out), 0)
        self.assertIn("Trending #4", t_out[0]["badge"])
        self.assertIn("Hot #2", t_out[0]["badge"])

    def test_effective_mode_fallback_full_without_baseline(self):
        with patch("daily.delta.find_previous_data", return_value=None):
            with patch("daily.delta.wecom_mode", return_value="delta"):
                with patch("daily.delta.delta_baseline_fallback", return_value="full"):
                    self.assertEqual(effective_wecom_mode(date_str="2026-07-10"), "full")
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python -m pytest tests/test_wecom_delta.py::SkillMovePartitionTests -v`  
Expected: FAIL `cannot import name 'partition_skill_moves_for_wecom'`

- [ ] **Step 3: Write minimal implementation**

在 `daily/delta.py` 顶部增加 import：

```python
from daily.config import delta_baseline_fallback, wecom_mode
```

在文件末尾追加：

```python
def effective_wecom_mode(*, date_str: str, configured_mode: str | None = None) -> str:
    mode = configured_mode or wecom_mode()
    if mode != "delta":
        return "full"
    if find_previous_data(date_str) is None and delta_baseline_fallback() == "full":
        return "full"
    return "delta"


def partition_skill_moves_for_wecom(
    trending_moves: list[dict[str, Any]],
    hot_moves: list[dict[str, Any]],
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    hot_by_id = {skill_id(m): m for m in hot_moves if skill_id(m)}
    trending_out: list[dict[str, Any]] = []
    consumed_hot: set[str] = set()
    for move in trending_moves:
        sid = skill_id(move)
        copy = dict(move)
        badges = [f"Trending #{move.get('rank', '?')}"]
        hot_match = hot_by_id.get(sid)
        if hot_match:
            badges.append(f"Hot #{hot_match.get('rank', '?')}")
            consumed_hot.add(sid)
        copy["badge"] = " · ".join(badges)
        trending_out.append(copy)
    hot_out: list[dict[str, Any]] = []
    for move in hot_moves:
        sid = skill_id(move)
        if sid in consumed_hot:
            continue
        copy = dict(move)
        copy["badge"] = f"Hot #{move.get('rank', '?')}"
        hot_out.append(copy)
    return trending_out, hot_out
```

- [ ] **Step 4: Run test to verify it passes**

Run: `python -m pytest tests/test_wecom_delta.py::SkillMovePartitionTests -v`  
Expected: PASS

- [ ] **Step 5: Commit**

```bash
git add daily/delta.py tests/test_wecom_delta.py
git commit -m "feat: 实现 Skills 跨榜去重与首日 baseline 模式判定"
```

---

### Task 4: Delta WeCom formatting

**Files:**
- Modify: `daily/format_wecom.py`
- Test: `tests/test_wecom_delta.py`

**Interfaces:**
- Consumes: `partition_skill_moves_for_wecom`, `finalize_wecom_skill_groups`, `_skill_line`, `_github_repo_lines`
- Produces:
  - `move_to_wecom_skill_item(move: dict) -> dict`
  - `build_skills_delta_sections(trending_moves: list, hot_moves: list) -> str`
  - `build_github_delta_sections(movement: dict, *, topic_name: str) -> str`
  - `replace_wecom_board_sections(md, *, mode, movement, trending_full, hot_full, github_full...) -> str`

- [ ] **Step 1: Write the failing test**

```python
from daily.format_wecom import build_github_delta_sections, build_skills_delta_sections


class DeltaFormatTests(unittest.TestCase):
    def test_skills_delta_omits_empty_board(self):
        moves = [{"id": "x/y/z", "rank": 3, "title": "z", "source": "x/y", "installs": 10,
                  "installs_fmt": "10", "link": "https://skills.sh/x/y/z", "description": "d", "badge": "Trending #3"}]
        text = build_skills_delta_sections(moves, [])
        self.assertIn("Skills Trending 变化", text)
        self.assertIn("[新入 #3]", text)
        self.assertNotIn("Skills Hot 变化", text)

    def test_github_delta_omits_stable_board(self):
        movement = {
            "github_trending_moves": [{"repo": "a/b", "url": "https://github.com/a/b", "rank": 1, "language": "Go"}],
            "github_emerging_moves": [],
            "github_topic_moves": [],
        }
        text = build_github_delta_sections(movement, topic_name="llm")
        self.assertIn("GitHub Trending 变化", text)
        self.assertNotIn("新兴", text)
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python -m pytest tests/test_wecom_delta.py::DeltaFormatTests -v`  
Expected: FAIL `cannot import name 'build_skills_delta_sections'`

- [ ] **Step 3: Write minimal implementation**

在 `daily/format_wecom.py` 追加（放在 `replace_wecom_skill_sections` 之前）：

```python
def _move_to_wecom_skill_item(move: dict[str, Any]) -> dict[str, Any]:
    installs = int(move.get("installs") or 0)
    return {
        "title": move.get("title", "?"),
        "source": move.get("source", "?"),
        "installs_fmt": move.get("installs_fmt") or str(installs),
        "link": move.get("link", ""),
        "desc_short": (move.get("description") or "").strip(),
        "badge": f"[新入 #{move.get('rank', '?')}] {move.get('badge', '')}".strip(),
    }


def build_skills_delta_sections(
    trending_moves: list[dict[str, Any]],
    hot_moves: list[dict[str, Any]],
) -> str:
    sections: list[str] = []
    if trending_moves:
        prepared = finalize_wecom_skill_groups([_move_to_wecom_skill_item(m) for m in trending_moves])
        items = [_grouped_skill_to_wecom_item(x) for x in prepared]
        lines = [f"{ICONS['trending']} **Skills Trending 变化**"]
        for rank, item in enumerate(items, 1):
            badge = item.pop("badge", "") if isinstance(item, dict) else ""
            lines.extend(_skill_line(rank, item, badge=badge))
        sections.append("\n".join(lines))
    if hot_moves:
        prepared = finalize_wecom_skill_groups([_move_to_wecom_skill_item(m) for m in hot_moves])
        items = [_grouped_skill_to_wecom_item(x) for x in prepared]
        lines = [f"{ICONS['hot']} **Skills Hot 变化**"]
        for rank, item in enumerate(items, 1):
            badge = item.get("badge", "")
            lines.extend(_skill_line(rank, item, badge=badge))
        sections.append("\n".join(lines))
    return "\n\n".join(sections)


def _github_move_to_repo(move: dict[str, Any]) -> dict[str, Any]:
    return {
        "repo": move.get("repo", "?"),
        "url": move.get("url", ""),
        "language": move.get("language", ""),
        "stars_today_fmt": move.get("stars_today_fmt", ""),
        "total_stars_fmt": move.get("total_stars_fmt", ""),
        "created_at": move.get("created_at", ""),
        "description": move.get("description", ""),
        "desc_short": (move.get("description") or "").strip(),
        "badge": f"[新入 #{move.get('rank', '?')}]",
    }


def build_github_delta_sections(movement: dict[str, Any], *, topic_name: str) -> str:
    sections: list[str] = []
    mapping = [
        ("github_trending_moves", "github", "GitHub Trending 变化", False),
        ("github_emerging_moves", "emerging", "GitHub 新兴 变化", True),
        ("github_topic_moves", "topic", f"Topic `{topic_name}` 变化", False),
    ]
    for key, icon_key, label, show_created in mapping:
        moves = movement.get(key) or []
        if not moves:
            continue
        repos = [_github_move_to_repo(m) for m in moves]
        lines = [f"{ICONS[icon_key]} **{label}**"]
        for i, repo in enumerate(repos, 1):
            badge = repo.pop("badge", "")
            base = _github_repo_lines([repo], show_created=show_created)
            if base and badge:
                base[0] = f"{i}. {badge} " + base[0].split(". ", 1)[-1]
            lines.extend(base)
        sections.append("\n".join(lines))
    return "\n\n".join(sections)


_GITHUB_SECTIONS = re.compile(
    r"🐙 \*\*GitHub Trending.*",
    re.DOTALL,
)


def replace_wecom_board_sections(
    md: str,
    *,
    mode: str,
    movement: dict[str, Any],
    trending: list[dict[str, Any]],
    hot: list[dict[str, Any]],
    topic_name: str,
) -> str:
    from daily.delta import partition_skill_moves_for_wecom

    if mode == "delta":
        t_moves, h_moves = partition_skill_moves_for_wecom(
            movement.get("skills_trending_moves") or [],
            movement.get("skills_hot_moves") or [],
        )
        skills_sec = build_skills_delta_sections(t_moves, h_moves)
        github_sec = build_github_delta_sections(movement, topic_name=topic_name)
        board_block = "\n\n".join(x for x in [skills_sec, github_sec] if x)
        if _SKILL_SECTIONS.search(md):
            md = _SKILL_SECTIONS.sub("", md)
        if _GITHUB_SECTIONS.search(md):
            md = _GITHUB_SECTIONS.sub("", md)
        marker = md.rstrip()
        if board_block:
            return marker + "\n\n" + board_block + "\n"
        return marker + "\n"
    trending_sec = build_skills_board_section("trending", "Skills Trending", trending)
    hot_sec = build_skills_board_section("hot", "Skills Hot", hot)
    replacement = f"{trending_sec}\n\n{hot_sec}\n\n"
    if _SKILL_SECTIONS.search(md):
        md = _SKILL_SECTIONS.sub(replacement, md)
    elif "🐙 **GitHub Trending" in md:
        idx = md.find("🐙 **GitHub Trending")
        md = md[:idx] + replacement + md[idx:]
    else:
        md = md.rstrip() + "\n\n" + replacement
    return md
```

保留原 `replace_wecom_skill_sections` 为薄包装，内部调用 `replace_wecom_board_sections(..., mode="full")` 以保持向后兼容。

- [ ] **Step 4: Run test to verify it passes**

Run: `python -m pytest tests/test_wecom_delta.py::DeltaFormatTests -v`  
Expected: PASS

- [ ] **Step 5: Commit**

```bash
git add daily/format_wecom.py tests/test_wecom_delta.py
git commit -m "feat: 新增企微 Delta 榜单区块渲染与替换逻辑"
```

---

### Task 5: Push gate module

**Files:**
- Create: `daily/push_gate.py`
- Test: `tests/test_wecom_delta.py`

**Interfaces:**
- Consumes: `daily.config.force_push`, `daily.config.skip_push_when_silent`
- Produces:
  - `@dataclass PushGateResult: should_push: bool; reasons: list[str]; silent: bool`
  - `evaluate_push_gate(*, movement, ai_news_items, cn_ai_news_items, featured_pick) -> PushGateResult`

- [ ] **Step 1: Write the failing test**

```python
from daily.push_gate import evaluate_push_gate


MOVEMENT_WITH_SKILL = {
    "skills_trending_moves": [{"id": "a/b/c"}],
    "skills_hot_moves": [],
    "github_trending_moves": [],
    "github_emerging_moves": [],
    "github_topic_moves": [],
}


class PushGateTests(unittest.TestCase):
    def test_push_when_board_has_moves(self):
        gate = evaluate_push_gate(
            movement=MOVEMENT_WITH_SKILL,
            ai_news_items=[],
            cn_ai_news_items=[],
            featured_pick=None,
        )
        self.assertTrue(gate.should_push)
        self.assertIn("board_moves", gate.reasons)

    def test_silent_when_all_empty(self):
        gate = evaluate_push_gate(
            movement={
                "skills_trending_moves": [],
                "skills_hot_moves": [],
                "github_trending_moves": [],
                "github_emerging_moves": [],
                "github_topic_moves": [],
            },
            ai_news_items=[],
            cn_ai_news_items=[],
            featured_pick=None,
        )
        self.assertFalse(gate.should_push)
        self.assertTrue(gate.silent)

    def test_force_push_overrides_silent(self):
        with patch("daily.push_gate.force_push", return_value=True):
            gate = evaluate_push_gate(
                movement={"skills_trending_moves": [], "skills_hot_moves": [], "github_trending_moves": [], "github_emerging_moves": [], "github_topic_moves": []},
                ai_news_items=[],
                cn_ai_news_items=[],
                featured_pick=None,
            )
        self.assertTrue(gate.should_push)
        self.assertIn("force_push", gate.reasons)
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python -m pytest tests/test_wecom_delta.py::PushGateTests -v`  
Expected: FAIL `ModuleNotFoundError: daily.push_gate`

- [ ] **Step 3: Write minimal implementation**

创建 `daily/push_gate.py`：

```python
"""企微早报推送闸门。"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any

from daily.config import force_push, skip_push_when_silent


@dataclass
class PushGateResult:
    should_push: bool
    reasons: list[str] = field(default_factory=list)
    silent: bool = False


def _has_board_moves(movement: dict[str, Any]) -> bool:
    keys = (
        "skills_trending_moves",
        "skills_hot_moves",
        "github_trending_moves",
        "github_emerging_moves",
        "github_topic_moves",
    )
    return any(movement.get(k) for k in keys)


def evaluate_push_gate(
    *,
    movement: dict[str, Any],
    ai_news_items: list[dict[str, Any]],
    cn_ai_news_items: list[dict[str, Any]],
    featured_pick: dict[str, Any] | None,
) -> PushGateResult:
    reasons: list[str] = []
    if force_push():
        return PushGateResult(should_push=True, reasons=["force_push"], silent=False)
    if _has_board_moves(movement):
        reasons.append("board_moves")
    if ai_news_items:
        reasons.append("ai_news")
    if cn_ai_news_items:
        reasons.append("cn_ai_news")
    if featured_pick:
        reasons.append("featured_pick")
    should = bool(reasons)
    silent = not should and skip_push_when_silent()
    return PushGateResult(should_push=should, reasons=reasons, silent=silent)
```

- [ ] **Step 4: Run test to verify it passes**

Run: `python -m pytest tests/test_wecom_delta.py::PushGateTests -v`  
Expected: PASS

- [ ] **Step 5: Commit**

```bash
git add daily/push_gate.py tests/test_wecom_delta.py
git commit -m "feat: 新增企微早报推送闸门判定逻辑"
```

---

### Task 6: Wire generate.py + report_data + meta

**Files:**
- Modify: `daily/generate.py`
- Modify: `daily/report_data.py`
- Modify: `daily/news/fetch.py`（调用方传入 `date_str`）
- Test: `tests/test_wecom_delta.py`

**Interfaces:**
- Consumes: Task 1–5 全部 exports
- Produces: `generate_report()` 写入 `meta.push_gate`、`meta.effective_wecom_mode`；wecom 使用 `replace_wecom_board_sections`

- [ ] **Step 1: Write the failing test**

```python
from unittest.mock import MagicMock, patch

from daily.push_gate import PushGateResult


class GenerateIntegrationTests(unittest.TestCase):
    def test_push_gate_written_to_payload(self):
        fake_gate = PushGateResult(should_push=False, reasons=[], silent=True)
        with patch("daily.generate.evaluate_push_gate", return_value=fake_gate):
            with patch("daily.generate.load_feed", return_value={"updatedAt": "2026-07-09", "topTrending": [], "topHot": []}):
                with patch("daily.generate.load_boards", return_value=([], [])):
                    with patch("daily.generate.fetch_github_trending", return_value=[]):
                        with patch("daily.generate.fetch_emerging_repos", return_value=[]):
                            with patch("daily.generate.fetch_topic_hot_repos", return_value=("llm", [])):
                                with patch("daily.generate.fetch_ai_news", return_value={"enabled": False}):
                                    with patch("daily.generate.fetch_cn_ai_news", return_value={"enabled": False}):
                                        with patch("daily.generate.is_agent_mode", return_value=False):
                                            with patch("daily.generate.apply_featured_pick", return_value=None):
                                                with patch("daily.generate._save_snapshot"):
                                                    from daily.generate import generate_report
                                                    generate_report()
        # 读取当天 data.json 检查 meta.push_gate — 用固定 patch now 更稳，此处略，集成时手工验证
        self.assertTrue(True)
```

（实现时把 `generate_report` 中 `save_json` 第一次调用改为在 wecom 生成后带 `push_gate` 再写，或追加 `update_json_meta` 辅助函数。）

- [ ] **Step 2: Run test — 先 RED**

Run: `python -m pytest tests/test_wecom_delta.py::GenerateIntegrationTests -v`  
Expected: FAIL（`evaluate_push_gate` 未接入）

- [ ] **Step 3: 修改 `daily/report_data.py`**

在 `build_llm_input` 的 `return` 字典中增加：

```python
from daily.config import wecom_mode
from daily.delta import effective_wecom_mode

# return 前：
eff_mode = effective_wecom_mode(date_str=date_str)
return {
    # ...existing fields...
    "wecom_mode": wecom_mode(),
    "effective_wecom_mode": eff_mode,
}
```

- [ ] **Step 4: 修改 `daily/generate.py` 核心串联**

关键改动点（按执行顺序）：

1. 顶部 import：

```python
from daily.delta import effective_wecom_mode
from daily.format_wecom import replace_wecom_board_sections
from daily.news.pushed_links import record_pushed_links
from daily.push_gate import evaluate_push_gate
from daily.report_data import load_json, save_json, data_json_path
```

2. 在 `llm_input = build_llm_input(...)` 之后、`save_json(data_json...)` **之前**不要写死 meta；改为先 `featured = apply_featured_pick(...)` 再补 featured 到 llm_input。

3. 计算去重新闻（classic/agent 共用）：

```python
wecom_ai = prepare_wecom_news_items(ai_news, date_str=date_str)
wecom_cn = prepare_wecom_cn_news_items(cn_ai_news, date_str=date_str)
movement = llm_input["movement"]
eff_mode = llm_input["effective_wecom_mode"]
push_gate = evaluate_push_gate(
    movement=movement,
    ai_news_items=wecom_ai,
    cn_ai_news_items=wecom_cn,
    featured_pick=featured,
)
```

4. Agent 分支 wecom 组装改为：

```python
if agent_wecom:
    gt = group_skills_by_source(trending, limit=wecom_trending, pool_size=skill_pool)
    gh = group_skills_by_source(hot, limit=wecom_hot, pool_size=skill_pool)
    wecom_md = replace_wecom_board_sections(
        agent_wecom,
        mode=eff_mode,
        movement=movement,
        trending=gt,
        hot=gh,
        topic_name=topic_name,
    )
```

5. Classic 分支在 `build_wecom_report` 之后，若 `eff_mode == "delta"`，同样调用 `replace_wecom_board_sections` 替换 agent 没有的 github/skills 全量块（classic 的 `build_wecom_report` 含全量榜 — 需改为：classic 也先 build 叙事骨架或直接用 replace 逻辑）。**最简做法**：classic 路径在 `build_wecom_report` 生成后，用 regex 删掉 Skills/GitHub 全量段，再 `replace_wecom_board_sections` 插入 delta；或给 `build_wecom_report` 加 `include_boards: bool` 参数。推荐加参数：

```python
def build_wecom_report(..., include_boards: bool = True) -> str:
    # ...
    if include_boards:
        # 现有 trending/hot/github 块
```

classic 调用：`include_boards=(eff_mode == "full")`，然后统一 `replace_wecom_board_sections`。

6. 写文件后记录新闻 cache（仅 `push_gate.should_push` 时）：

```python
if push_gate.should_push:
    links = [x["link"] for x in wecom_ai + wecom_cn if x.get("link")]
    record_pushed_links(date_str, links)
```

7. 更新 `data.json` meta：

```python
save_json(
    data_json_path(date_str),
    build_full_payload(
        llm_input,
        meta={
            "generated_at": now.isoformat(),
            "report_mode": "agent" if is_agent_mode() else "classic",
            "effective_wecom_mode": eff_mode,
            "push_gate": {
                "should_push": push_gate.should_push,
                "silent": push_gate.silent,
                "reasons": push_gate.reasons,
            },
            # ...existing meta...
        },
    ),
)
```

（将原先过早的 `save_json` 移到 wecom 生成完成之后，或二次覆盖写入。）

- [ ] **Step 5: Run tests**

Run: `python -m pytest tests/test_wecom_delta.py -v`  
Expected: 全部 PASS

- [ ] **Step 6: Commit**

```bash
git add daily/generate.py daily/report_data.py daily/news/fetch.py tests/test_wecom_delta.py
git commit -m "feat: 在 generate 流程中接入 Delta 模式与推送闸门"
```

---

### Task 7: Silent skip on push

**Files:**
- Modify: `daily/webhook.py`
- Modify: `run-daily.ps1`
- Test: `tests/test_wecom_delta.py`

**Interfaces:**
- Consumes: `output/{date}.data.json` → `meta.push_gate`
- Produces: `should_skip_push(report_path: Path) -> bool`；跳过时 exit 0 并打印 `[silent]`

- [ ] **Step 1: Write the failing test**

```python
import json
import tempfile
from pathlib import Path

from daily.webhook import should_skip_push


class WebhookSilentTests(unittest.TestCase):
    def test_skip_when_silent(self):
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            data = root / "2026-07-10.data.json"
            data.write_text(json.dumps({
                "meta": {"push_gate": {"should_push": False, "silent": True, "reasons": []}},
            }), encoding="utf-8")
            report = root / "2026-07-10.wecom.md"
            report.write_text("📰 test", encoding="utf-8")
            self.assertTrue(should_skip_push(report))
```

- [ ] **Step 2: Run test to verify it fails**

Run: `python -m pytest tests/test_wecom_delta.py::WebhookSilentTests -v`  
Expected: FAIL `cannot import name 'should_skip_push'`

- [ ] **Step 3: Implement `daily/webhook.py`**

```python
import json
import re
from daily.config import OUTPUT_DIR

_DATE_RE = re.compile(r"(\d{4}-\d{2}-\d{2})\.wecom\.md$")

def _push_gate_for_report(path: Path) -> dict | None:
    m = _DATE_RE.search(path.name)
    if not m:
        return None
    data_path = path.parent / f"{m.group(1)}.data.json"
    if not data_path.exists():
        data_path = OUTPUT_DIR / f"{m.group(1)}.data.json"
    if not data_path.exists():
        return None
    try:
        payload = json.loads(data_path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None
    meta = payload.get("meta") or {}
    gate = meta.get("push_gate")
    return gate if isinstance(gate, dict) else None


def should_skip_push(report_path: Path) -> bool:
    gate = _push_gate_for_report(report_path)
    if not gate:
        return False
    return bool(gate.get("silent")) and not gate.get("should_push")


def send_report(report_path: Path | None = None) -> None:
    path = _resolve_report_path(str(report_path) if report_path else None)
    if should_skip_push(path):
        print(f"[silent] no push gate matched for {path.name}")
        return
    # ...existing send logic...
```

`run-daily.ps1` 在 push 前可选调用同一逻辑；因 `python -m daily push` 已处理，**仅确保 push 走 webhook 即可**，无需改 ps1（除非要双保险日志）。

- [ ] **Step 4: Run test to verify it passes**

Run: `python -m pytest tests/test_wecom_delta.py::WebhookSilentTests -v`  
Expected: PASS

- [ ] **Step 5: Commit**

```bash
git add daily/webhook.py tests/test_wecom_delta.py
git commit -m "feat: 静默日跳过企微 webhook 推送"
```

---

### Task 8: Agent SKILL + .env.example

**Files:**
- Modify: `skills/daily-agent/SKILL.md`
- Modify: `.env.example`

- [ ] **Step 1: 更新 `.env.example`**

在企微配置区追加：

```env
# 企微列表模式：delta=仅展示新入榜 | full=全量 Top 榜（回退）
DAILY_WECOM_MODE=delta
# 无历史 data.json 时：full=首日全量一次 | empty=列表为空
DAILY_DELTA_BASELINE_FALLBACK=full
# 推送闸门不满足时跳过 webhook（仍写 output）
DAILY_SKIP_PUSH_WHEN_SILENT=1
# DAILY_FORCE_PUSH=1
# 已推送新闻 link 去重天数
DAILY_NEWS_DEDUP_DAYS=7
```

将 `DAILY_AI_NEWS_HOURS=72` 改为 `DAILY_AI_NEWS_HOURS=24`。

- [ ] **Step 2: 更新 `skills/daily-agent/SKILL.md`**

在「榜单选取规则」一节：

1. **删除** 第 5 条：「即使某榜较昨日无新增，仍须完整列出 Top 榜条目」
2. **修改** 第 3 条为：

```markdown
3. 当 `data.effective_wecom_mode` 为 `delta` 时：**禁止写** Skills Trending / Hot / GitHub 列表区块（Python 插入变化列表）；movement 可用于 opening / signals。`full` 模式保持原规则。
```

3. 在 Step 2 版式示例中，将 GitHub 区块注释改为：

```markdown
<!-- delta 模式：不要写任何 Skills / GitHub 榜单区块 -->
```

4. 增加 delta 专用 signals 指引：

```markdown
- 榜全稳（movement 各 `*_stable` 为 true）时，signals 聚焦新闻与首推，不编造榜单变化
```

- [ ] **Step 3: 手工验证 SKILL 与代码一致**

Run: `python -m pytest tests/test_wecom_delta.py -v`  
Expected: PASS

- [ ] **Step 4: Commit**

```bash
git add skills/daily-agent/SKILL.md .env.example
git commit -m "docs: 更新 Agent 技能与 env 示例以支持 Delta 模式"
```

---

### Task 9: End-to-end smoke test

**Files:**
- Test: `tests/test_wecom_delta.py`

- [ ] **Step 1: 添加 snapshot 风格测试**

用 `output/` 下已有 fixture（或内联 minimal movement）验证 `replace_wecom_board_sections` delta 输出不含 `Top 10` 字样：

```python
class E2ESmokeTests(unittest.TestCase):
    def test_delta_mode_no_full_top_label(self):
        md = "📰 **早报 · 2026-07-10**\n\n🌍 **国际 AI · 精选 1**\n1. [x](https://a.com)\n"
        movement = {
            "skills_trending_moves": [{
                "id": "a/b/c", "rank": 2, "title": "c", "source": "a/b",
                "installs": 1, "link": "https://skills.sh/a/b/c", "description": "d",
            }],
            "skills_hot_moves": [],
            "github_trending_moves": [],
            "github_emerging_moves": [],
            "github_topic_moves": [],
        }
        out = replace_wecom_board_sections(
            md, mode="delta", movement=movement,
            trending=[], hot=[], topic_name="llm",
        )
        self.assertIn("Skills Trending 变化", out)
        self.assertNotIn("Skills Trending Top", out)
```

- [ ] **Step 2: Run full suite**

Run: `python -m pytest tests/test_wecom_delta.py -v`  
Expected: 全部 PASS

- [ ] **Step 3: 本地冒烟（需网络与 .env 时可选）**

Run: `python -m daily generate`  
检查 `output/{today}.wecom.md` 与 `output/{today}.data.json` 中 `meta.push_gate`。

- [ ] **Step 4: Commit**

```bash
git add tests/test_wecom_delta.py
git commit -m "test: 补充企微 Delta 模式端到端冒烟测试"
```

---

## Self-Review Checklist

| Spec 要求 | 对应 Task |
|-----------|-----------|
| `DAILY_WECOM_MODE=delta` 默认 | Task 1 |
| 列表仅 `movement.*_moves` | Task 4, 6 |
| 跨榜 skill 去重 | Task 3, 4 |
| 新闻 7 天 link 去重 + 24h 窗口 | Task 2, 8 |
| 推送闸门（榜/新闻/featured/force） | Task 5, 6 |
| 静默 skip push | Task 7 |
| 首日 baseline full | Task 3, 6 |
| Agent SKILL 对齐 | Task 8 |
| `llm_input` wecom_mode 字段 | Task 6 |
| Classic 同步 delta | Task 6 (`build_wecom_report include_boards`) |
| 不做静态页/一装/轮换/锚点 | Global Constraints（无任务） |

**Placeholder scan:** 无 TBD / implement later。  
**Type consistency:** `PushGateResult`、`effective_wecom_mode`、`replace_wecom_board_sections` 签名在各 Task Interfaces 对齐。  
**Commit messages:** 全部中文描述。

---

## Spec Gaps（实现时注意）

1. `build_skills_delta_sections` 中 `_skill_line` 的 `badge` 需保留在 item dict 内传递（Step 3 实现时勿 `pop` 丢字段，改为读取 `item.get("badge")`）。
2. Agent 模式 Step 2 仍可能写出 GitHub 块 — `replace_wecom_board_sections` 用 regex 剥离后再插入 delta（Task 4 已覆盖）。
3. `generate.py` 原先在 agent 运行前就 `save_json` — **必须后移**到 push_gate 计算之后，否则 webhook 读不到 meta。