WorkBuddy 自动化任务网关（workbuddy-task-gateway）· v1.0.0
> 把"分散在各自动化任务里的调度 + 容错 + 推送"收敛成一个共享 IP。所有任务共用同一条稳定出口。
一、它解决什么（先结论）
| 痛点 | 本方案解法 |
|---|---|
| 每个任务各自写调度、分散难维护 | 统一入口 run_task / serve，任务集中在 tasks/registry.json |
| 任务跑挂了没人兜底 | 四重容错：重试退避 / 超时熔断 / 断路器 / 死信队列 |
| 推送逻辑重复且易因出口 IP 漂移 40164 | 复用 wechat_gateway（零依赖/强制IPv4/双通道降级/预审分级/红线内置），推送失败隔离进死信可重放 |
| 进程重启丢状态 | 运行历史、下次运行时间、断路器状态全落 JSON 文件 |
定位边界：本技能是"任务网关"，长在既有 wechat_gateway/gateway.py（公众号统一发布网关）之上。出口 IP 漂移、双栈 IPv6 隐性 40164、预审字数门槛、群发红线的兜底，全部已在上游网关里解决，本网关只管"任务怎么被统一调度、跑挂了怎么办"，不复刻这些逻辑。
二、触发场景
- 用户有 ≥2 个自动化任务（研究/抓取/分析/选题/发文），希望统一调度而非各自为政。
- 用户抱怨"任务跑挂没人管""推送零散难维护""公众号草稿要手动贴"。
- 用户要在 WorkBuddy 把所有栏目（如一千零一技/工具早班车/单干手册/树洞23:59）挂成共享出口的常驻调度。
- 用户想把"跑完任务自动进草稿箱"作为通用能力沉淀为可复用方案。
三、参考实现（已实测通过，可直接用）
技能内置完整参考实现于 scripts/，与 _shared/task_gateway/ 真实运行态一致：
scripts/
├── task_gateway.py # 核心：统一入口 + cron 调度 + 四重容错 + 结果推送 + CLI
├── gateway_bridge.py # 对接 wechat_gateway/gateway.py（含 mock 安全降级）
├── task_gateway_config.json # 全局配置（超时/重试/断路器/推送开关）
├── tasks/
│ ├── registry.json # 任务注册表（cron / handler / 推送配置）
│ ├── example_research.py # 示例研究任务：产出 md+封面 → 推草稿
│ └── __init__.py
└── README.md # 架构与接入指南
部署一步到位（装/挂即用）：
# 1) 落到共享目录（与 wechat_gateway 平级，bridge 用 ../wechat_gateway 自动定位上游）
DST="C:/Users/刘鑫伟/WorkBuddy/_shared/task_gateway"
mkdir -p "$DST/tasks" && cp -r scripts/. "$DST/"
# 2) 只跑任务逻辑（不触网，mock 推送）：
python "$DST/task_gateway.py" run --task example_research
# 3) 挂成 WorkBuddy 自动化（常驻按 cron 调度全部任务）：prompt 一句即可
# "运行 python <DST>/task_gateway.py serve，以统一网关调度全部任务并把成果推成公众号草稿"
四、四重容错（破坏性测试已逐一验证）
- 重试退避：瞬时错误指数退避重试（默认
max_retries=2，退避backoff_base_sec × 2^attempt）。 - 超时熔断：单任务超时被线程强制终止（
_run_in_thread+join(timeout)），不拖死调度器。 - 断路器：某任务连续失败达
breaker_failures（默认 3）即 OPEN 冷却breaker_cooldown_sec（默认 1800s），冷却后才放一次 HALF_OPEN 探活。 - 死信队列：最终失败的任务/推送写入
deadletter/dead_*.json，retry-dead精准重放（draft 死信直接重推，task 死信重跑），成功即清理。
运行态持久化：state/run_history.json（最近 300 条）、state/breakers.json、state/next_runs.json、state/mock_drafts/。
五、安全默认与红线（强制）
- 默认 mock：无真实
wechat_gateway/gateway_config.json凭据，或环境变量TASKGW_DRAFT_MOCK=1时，桥接层自动走 mock（仅落盘记录，不向微信发任何请求）。端到端可测、绝不误推。 - 真推三步：①补齐
../wechat_gateway/gateway_config.json；②确保本网关draft.enabled=true（默认已开）；③去掉TASKGW_DRAFT_MOCK；预览用run --dry先预审。 - 红线只在 draft/add：上游
wechat_gateway代码内FORBIDDEN_PATHS = ("/cgi-bin/message/mass", "/cgi-bin/freepublish")，本技能及上游均不存在群发/发布接口调用。机器止于草稿箱，群发必须人工在公众号后台点击。 - 推送失败隔离：推送异常只进死信，不污染任务成功态——任务算成功、
pushed=false，可独立重放。
六、接入一个新任务（三步走）
- 在
tasks/写模块，定义task_main(context) -> dict：
python def task_main(context): ctx = context md = ctx.out_dir + "/article.md" open(md, "w", encoding="utf-8").write("# 标题\n正文…") cover = ctx.make_cover() # 纯标准库生成合规封面（不依赖 Pillow） return {"draft": {"title": "…", "md_path": md, "cover_path": cover, "digest": "…", "author": "阿笃", "preset": "standard"}}
不推草稿就返回{}或不含draft的 dict。 - 在
tasks/registry.json加一条：id / name / handler(tasks.xxx) / cron / draft / timeout_sec / max_retries。 python task_gateway.py run --task <id>验证，serve即自动纳入调度。
七、CLI 速查
python task_gateway.py list # 任务 + 下次运行时间
python task_gateway.py run --task <id> [--force] [--dry]
python task_gateway.py serve # 常驻按 cron 调度
python task_gateway.py health # 微信通道 + 各任务断路器
python task_gateway.py status [--task X] [--n 20]
python task_gateway.py retry-dead [--kind task|draft] [--all]
八、已知限制 / 边界
serve为单进程常驻调度器，不负责跨进程高可用（崩溃需靠平台自动化重启）；大规模集群建议改用 WorkBuddy 原生自动化逐个挂载run --task。- cron 解析为逐分钟扫描（5 年内查找），低频任务足够；秒级精度不支持。
- 运行态为本地 JSON 文件，多实例并行写同一
state/会竞争——同一网关目录只跑一个进程。 - 上游
wechat_gateway必须先存在并配置；本技能不替你建公众号白名单/IP 固定出口（那是ip-drift-hunter/ 固定出口 IP 方案的事）。 - 封面
make_cover为纯色兜底图，正式栏目请替换为真实封面生成（如ai-image-deai去 AI 味后落盘）。
九、与外部技能的协同
| 技能 | 关系 |
|---|---|
wechat_gateway（既有，非本技能产物） |
上游发布网关，本技能桥接它 |
ip-drift-hunter |
根治出口 IP 漂移 40164（上游网关已强制 IPv4 吸收，本层无需关心） |
md-to-wechat / wechat-oa / wechat-publisher-pro |
单篇发布型技能；本技能是"多任务统一调度 + 自动推送"层，二者可并存 |
ai-image-deai |
生成合规封面并去 AI 味，替换 make_cover 兜底图 |
skill-ops-monitor |
把本网关的 health/status 纳入技能运维巡检 |
十、参考来源（R1 真实性）
- 本报告原始研究对话（WorkBuddy 分享页）：
https://workbuddy.link/p/gf8Npj4Z0IRLjIvCCFHgdP?ext2=copy_link（课题：共建 WorkBuddy 自动化任务网关并推送公众号草稿）。 - 参考实现蓝本：
C:/Users/刘鑫伟/WorkBuddy/_shared/wechat_gateway/gateway.py（公众号统一发布网关，v 实测 34KB）+ 其上已落地的C:/Users/刘鑫伟/WorkBuddy/_shared/task_gateway/（v1.0.0 实测态）。 - 上游契约：网关
push(title, content_md_path, cover_path, digest, author, preset, account, prefer, dry_run)，FORBIDDEN_PATHS禁群发/发布；40164=出口 IP 不在白名单。 - 以上均为一手/可查源；本技能未做任何待证实假设。
十一、验收标准
- [ ]
scripts/7 个文件齐备且可python task_gateway.py list列出 example_research。 - [ ]
run --task example_research（mock）成功产出_outputs/example_research/article.md + cover.png，推送记state/mock_drafts/不触网。 - [ ] 四重容错经破坏性测试（注入故障任务）验证：超时终止、重试退避、断路器 OPEN、死信落盘与重放均通过。
- [ ]
health输出含wechat.mock=true或真实通道；retry-dead可清理死信。 - [ ] 真推联调（仅在你提供 gateway_config.json 后）走
draft/add，全程无 mass/freepublish 调用。
