feat(reader): add async article summary jobs

2026-04-10 16:37:43 +08:00
parent 87d18e4263
commit 2053cccfef
8 changed files with 1269 additions and 15 deletions
@@ -9,7 +9,7 @@ pip install -e .
 summary-mcp
 ```
-服务当前暴露 11 个工具。
+服务当前暴露 14 个工具。
 正式 workflow service 相关工具：
@@ -26,7 +26,10 @@ summary-mcp
 - `extract_url_content`
 - `extract_item_content`
 - `filter_summary_result`
- `generate_article_summaries`
+- `generate_article_summaries`（同步模式）
 - `start_article_summary_job`
 - `get_article_summary_job_status`
 - `get_article_summary_job_result`
 ## 正式能力边界
@@ -34,7 +37,9 @@ summary-mcp
 - 每次 FreshRSS 主流水线 run 都会在 `outputs/freshrss/rerun/<run_dir>/run-state.json` 落地运行真相
 - OpenClaw 正式读取结果应优先使用 `get_delivery_payload` 与 `get_run_report`，而不是自己拼输出目录路径
 - `digest-brief.json` 当前会随主流水线产出，但还没有独立的 MCP 读取工具；如需定位它，应通过 `list_run_artifacts` 或 `get_run_report` 返回的信息发现
- `run_freshrss_openclaw_pipeline` 与 `resume_run` 当前都是同步 MCP 调用；仓库里还没有后台队列 / worker / 异步任务管理
+- `run_freshrss_openclaw_pipeline` 与 `resume_run` 当前都是同步 MCP 调用；仓库里还没有针对 FreshRSS 主流程的后台队列 / worker / 异步任务管理
 - 单篇总结已补上最小异步 job 形态：`start_article_summary_job` / `get_article_summary_job_status` / `get_article_summary_job_result`
 - `generate_article_summaries` 仍保留，但定位是同步 debug 路径，而不是 OpenClaw 的正式生产集成入口
 ## OpenClaw 推荐调用路径
@@ -70,8 +75,10 @@ summary-mcp
 相关能力：
- `generate_article_summaries` MCP 工具（基于已有 extracted payload 做单篇总结）
+- `start_article_summary_job` / `get_article_summary_job_status` / `get_article_summary_job_result`（正式推荐的最小异步 job 路径）
 - `generate_article_summaries` MCP 工具（同步 debug 路径）
 - `scripts/run_article_summaries.py` CLI 辅助脚本
 - `scripts/run_article_summary_job.py` 后台 runner 入口
 ## 校验 LLM 摘要结果
@@ -257,7 +264,7 @@ python scripts/build_keyword_index.py ^
 - `skills/keyword-cleanup-review/`
-构建给 LLM skill 使用的评审数据包（review bundle）：
+构建给关键词治理流程使用的评审数据包（review bundle，临时工作文件）：
 ```bash
 python skills/keyword-cleanup-review/scripts/build_review_bundle.py ^
@@ -266,17 +273,44 @@ python skills/keyword-cleanup-review/scripts/build_review_bundle.py ^
  --output outputs/term_index/review/keyword-cleanup-bundle.json
 ```
-现在这个评审数据包（review bundle）还会额外携带治理上下文：
+这个评审数据包（review bundle）会额外携带治理上下文：
 - 来自 `configs/term_cleanup_policy.json` 的清理阈值
 - 当前 watch list（`configs/term_watchlist.json`）
 - 最近已应用的变更（`configs/term_change_log.json`）
-这个 skill 只负责生成 review 输入与建议，不会自动修改：
+接下来可以把 bundle 渲染成正式建议产物（默认走确定性规则，不把 LLM 作为默认路径）：
- `term_aliases`
+```bash
- `term_stopwords`
+python scripts/generate_term_cleanup_suggestions.py ^
- `filter_context.personal.json`
+  --bundle outputs/term_index/review/keyword-cleanup-bundle.json
 ```
 默认只生成：
 - `outputs/term_index/review/term-cleanup-suggestions-YYYY-MM-DD.json`（唯一正式建议产物，建议短期保留）
 如需人工审阅展示稿，再显式加：
 ```bash
 python scripts/generate_term_cleanup_suggestions.py ^
  --bundle outputs/term_index/review/keyword-cleanup-bundle.json ^
  --emit-markdown
 ```
 这时才会额外生成：
 - `outputs/term_index/review/term-cleanup-suggestions-YYYY-MM-DD.md`（临时展示稿，可按需生成，不必作为长期资产保留）
 其中本轮最小版本优先覆盖 `interest_keyword_suggestions` 和 `watch_terms` 主链路；`alias_suggestions` / `stopword_suggestions` 先保持保守。
 产物保留策略建议：
 - `data/term_index/daily/YYYY-MM-DD.json`、`data/term_index/term_stats.json` 作为事实层长期保留
 - `configs/filter_context.personal.json`、`configs/term_watchlist.json`、`configs/term_aliases.json`、`configs/term_stopwords.json`、`configs/term_change_log.json` 作为状态层长期保留
 - `term-cleanup-suggestions-YYYY-MM-DD.json` 作为正式建议产物短期保留（最近少量几份或仅保留已应用过的）
 - `keyword-cleanup-bundle.json` 仅作为临时工作文件，默认只保留当前最新一份
 - `term-cleanup-suggestions-YYYY-MM-DD.md` 仅作为临时展示稿，优先按需生成，不建议默认长期归档
 如果你想先预览已接受建议，再决定是否写配置文件：
@@ -324,12 +358,26 @@ python scripts/run_article_summaries.py ^
 也可以通过 `summary_mcp.server` 暴露的 MCP 工具 `generate_article_summaries` 调用：
 - `extracted_path`（string）：单篇 extracted JSON 路径（例如 `outputs/freshrss/rerun/<run_id>/extracted/item-01.extracted.json`），或者包含 `results` 数组的 batch extracted JSON
- `selected_ids`（array of strings）：要总结的一个或多个 `item_id`。如果传空数组，则对文件中的全部条目做总结
+- `selected_ids`（array of strings）：必填，至少传一个 `item_id`；如果传入的 ID 在 extracted payload 中一个都匹配不到，会直接报错
 - `output_dir`（optional string）：Markdown 输出目录；若不传，则默认写到 extracted 文件旁边的 `single_summaries/` 目录
 - `llm_api_key` / `llm_model` / `llm_api_url`（optional strings）：单篇总结 LLM 的覆盖配置；不传时会按前文规则回退到 `ARTICLE_SUMMARY_*` 或主 `LLM_*`
 该工具返回一个 JSON 数组，内容为生成好的 Markdown 文件路径。
 OpenClaw / 正式集成建议优先走异步 job：
 - 调 `start_article_summary_job` 启动任务，立即拿到 `job_id`
 - 轮询 `get_article_summary_job_status(job_id)`，直到 `status` 进入 `success` 或 `failed`
 - 成功后调用 `get_article_summary_job_result(job_id)` 读取 `written_paths` 与结构化结果
 - 失败时优先查看 `error_summary` 与 job 目录中的 `job-report.json`
 异步 job 状态目录固定落在 `outputs/freshrss/article_summary_jobs/<job_id>/`，最小会包含：
 - `run-state.json`
 - `input.json`
 - `result.json`（成功时）
 - `job-report.json`
 单篇总结使用独立 prompt：`outputs/prompts/article-summary-prompt.txt`。
 它与日报 prompt 完全独立，输出的是中文结构化知识笔记，包含这些部分：
@@ -187,6 +187,47 @@
 ---
 ### [DOING][P1] 单篇总结改为最小真异步 job
 目标：
 - 解决 `generate_article_summaries` 在 OpenClaw → MCP 同步链路里易 timeout 的问题
 - 将单篇总结正式升级为可启动、可轮询、可读取结果的异步 job
 要求：
 - 新增最小异步接口：
  - `start_article_summary_job`
  - `get_article_summary_job_status`
  - `get_article_summary_job_result`
 - 状态目录固定落到：
  - `outputs/freshrss/article_summary_jobs/<job_id>/`
 - 至少包含：
  - `run-state.json`
  - `input.json`
  - `result.json`（成功时）
  - `job-report.json`
 - 执行模型优先使用后台子进程，不使用线程
 - 业务逻辑继续复用 `summarize_selected_articles(...)`，不要重写正文总结核心逻辑
 - 对 OpenClaw / reader-digest-flow 而言，异步 job 成功后应可继续接 IMA 沉淀闭环
 当前进展：
 - 2026-04-10：已完成方案文档 `plans/article-summary-async-job-plan.md`
 - 2026-04-10：已落地最小代码骨架：
  - `src/summary_mcp/runtime/article_summary_jobs.py`
  - `scripts/run_article_summary_job.py`
  - `src/summary_mcp/server.py` 已新增 3 个 async job tools
 - 2026-04-10：已用真实 extracted 文件验证最小异步链路可跑通，job 能成功进入 `running → success`，并可读回结果
 - 2026-04-10：已补 README / OpenClaw handoff 文档，并对外统一为 `start_article_summary_job` / `get_article_summary_job_status` / `get_article_summary_job_result`
 - 2026-04-10：已完成聚焦自检：
  - MCP tool 注册名校验通过
  - stubbed async job 成功路径通过
  - stubbed async job 失败路径通过，`error_summary` 与 `job-report.json` 可回读
 下一步：
 - 将 `reader-digest-flow` 正式默认路径切到 async job
 - 用真实 LLM 配置再做一次非 stub 的服务端冒烟验证
 ---
 ### [TODO][P2] 评估 `rerun_stage` 是否值得进入第一阶段
 目标：
@@ -37,7 +37,7 @@ OpenClaw should treat the returned `run_id` as the only stable handle for follow
 ## Supported MCP Tools
-Current MCP tools: 11 total.
+Current MCP tools: 14 total, including the FreshRSS workflow set plus article-summary async job tools.
 Workflow service tools:
@@ -49,12 +49,34 @@ Workflow service tools:
 - `get_run_report`
 - `resume_run`
 Article-summary tools:
 - `start_article_summary_job`
 - `get_article_summary_job_status`
 - `get_article_summary_job_result`
 Single-step / debug tools:
 - `extract_url_content`
 - `extract_item_content`
 - `filter_summary_result`
- `generate_article_summaries`
+- `generate_article_summaries`（同步模式，仅适合轻量调试）
 ## Recommended Selected-Article Flow
 For OpenClaw selected-article follow-up, prefer the async job path:
 1. Call `start_article_summary_job` with a real extracted file path plus a non-empty `selected_ids` list.
 2. Poll `get_article_summary_job_status` until `status` becomes `success` or `failed`.
 3. On success, call `get_article_summary_job_result` and continue downstream processing from `written_paths`.
 4. Use `generate_article_summaries` only as a synchronous debug fallback, not as the default production path.
 Job state is written under `outputs/freshrss/article_summary_jobs/<job_id>/` and will minimally contain:
 - `run-state.json`
 - `input.json`
 - `result.json` on success
 - `job-report.json`
 ## Required Environment Variables
@@ -144,8 +166,9 @@ reader 当前正式 MCP workflow service 的边界如下：
 - state query tools: `get_run_status`, `list_runs`, `list_run_artifacts`
 - result read tools: `get_delivery_payload`, `get_run_report`
 - `digest-brief.json` is generated and registered as an artifact, but there is no standalone `get_digest_brief` tool yet
- `run_freshrss_openclaw_pipeline` and `resume_run` are synchronous MCP calls today; there is no background queue / worker model yet
+- `run_freshrss_openclaw_pipeline` and `resume_run` are synchronous MCP calls today; there is no background queue / worker model for the FreshRSS daily workflow yet
- `generate_article_summaries` is supported, but it is outside the formal `resume_run` scope and not part of the FreshRSS workflow-state model
+- article summary now has a minimal asynchronous job model with `start_article_summary_job` / `get_article_summary_job_status` / `get_article_summary_job_result`
 - `generate_article_summaries` is still supported, but it is a synchronous debug path and outside the formal `resume_run` scope
 Historical compatibility note:
@@ -0,0 +1,757 @@
 # 单篇总结异步 job 最小版落地方案
 ## 1. 背景与问题定义
 当前 reader 已经把日更 FreshRSS 主流程做成了带 `run-state.json` 的 runtime 模型：
 - `src/summary_mcp/runtime/state_models.py`
 - `src/summary_mcp/runtime/run_store.py`
 - `src/summary_mcp/runtime/query_service.py`
 - `src/summary_mcp/workflows/freshrss_pipeline.py`
 这条主链路已经具备：
 - run / stage / artifact 的结构化状态
 - MCP 查询接口：`get_run_status` / `list_runs` / `list_run_artifacts`
 - 结果读取接口：`get_delivery_payload` / `get_run_report`
 但“单篇总结”这条线目前还是同步调用：
 - 核心逻辑：`src/summary_mcp/workflows/article_summary.py`
 - MCP 暴露：`src/summary_mcp/server.py` 中的 `generate_article_summaries`
 - CLI 辅助：`scripts/run_article_summaries.py`
 现状问题已经很明确：
 - 在 OpenClaw → MCP tool 这条链路里，`generate_article_summaries` 可能因为 tool 调用时长而 timeout
 - 但 reader 项目本体在 `.venv` 下直接跑 article summary，大约 37.5 秒即可成功
 - 这说明问题不一定在 summary 业务本身，而更可能在“同步工具调用 + 上层等待模型”这个包装层
 所以目标不是先继续调 timeout，而是把单篇总结也纳入 **真正异步、可轮询、可落盘、可恢复基本状态** 的最小 job 模型里。
 ---
 ## 2. 为什么同步 MCP 不适合这一步
 `generate_article_summaries` 当前在 `server.py` 里直接同步执行 `summarize_selected_articles(...)`，调用方必须一直阻塞等待，直到：
 1. 读取 extracted payload
 2. 调用 LLM 生成总结
 3. 可选 repair retry
 4. 渲染 Markdown
 5. 写文件完成
 6. MCP tool 返回生成路径
 这个模式对“几十秒级、依赖外部 LLM、可能重试”的任务不稳，核心问题有三层：
 ### 2.1 tool 调用时长不可控
 `run_loop_payload()` 内部会发起外部 HTTP 请求，还可能做 repair retry。即便单次平均 37.5 秒，也已经接近很多上层编排系统的心理和技术超时边界。
 ### 2.2 调用方看不到中间状态
 现在如果卡住，调用方只能等：
 - 不知道是在读输入
 - 不知道是在调 LLM
 - 不知道是在重试
 - 不知道是否已经写出部分结果
 这也是同步接口最烦的点：失败时只能看到“tool timeout / tool failed”，而不是“业务跑到哪一步了”。
 ### 2.3 与 reader 已有 runtime 风格不一致
 FreshRSS 主流程已经是“run truth + status query + artifact read”的思路，而单篇总结仍然是黑箱同步函数。继续维持两套风格，只会让 SOP 更复杂：
 - 日报主链路用 `run_id`
 - 单篇总结却要么同步等，要么退回 CLI fallback
 这不利于后续把 `reader-digest-flow` 稳定成正式 SOP。
 ---
 ## 3. 本轮目标：最小真异步，不做大而全
 这次只做 **单篇总结异步 job 最小版**，目标是：
 > 让 OpenClaw 或其他调用方能先“启动单篇总结 job”，立即拿到 `job_id`，再通过状态接口轮询，最后读取输出文件/结果。
 ### 3.1 本轮必须做到的范围
 1. 新增单篇总结 job 的 start/status/result 最小接口
 2. job 真正在后台执行，而不是 MCP 请求线程里阻塞到完成
 3. 状态落盘到文件，遵循 reader 当前 runtime 风格
 4. 复用现有 `summarize_selected_articles` 逻辑，不重写业务
 5. 输出仍然是现有 Markdown 文件，不改知识内容 schema
 ### 3.2 本轮明确不做
 1. **不做通用队列系统**
 2. **不做数据库**
 3. **不做多 worker / 分布式调度**
 4. **不做取消 job / kill job**
 5. **不做并发配额控制**
 6. **不做 resume/retry from stage**
 7. **不把 article summary 一次性并入 freshrss `resume_run` 体系**
 8. **不改 summary prompt / validator / 输出格式**
 9. **不处理批量高吞吐场景优化**
 一句话：这轮只解“同步 tool 容易 timeout，但业务本身能跑完”这个问题，不顺手扩成任务调度平台。
 ---
 ## 4. 接入当前 reader 结构的建议
 ### 4.1 复用现有 runtime 设计，但单独建 article summary job 命名空间
 不建议把 article summary job 粗暴塞进现有 `freshrss_daily_digest` run 查询里混用一个 schema；更合适的是：
 - 复用 `RunState / StageState / ArtifactRecord / RunStore` 这套思维
 - 但给单篇总结定义独立 workflow 名称与存储目录
 建议：
 - workflow: `article_summary_job`
 - run_type: `article_summary`
 - output root: `outputs/freshrss/article_summary_jobs/<job_id>/`
 这样有几个好处：
 - 不污染 `outputs/freshrss/rerun/`
 - 语义清楚：这是独立 job，不是假装自己是日报 rerun
 - 查询和排查时更直观
 ### 4.2 job 与结果 Markdown 解耦
 job 目录只负责：
 - 状态文件
 - 输入快照
 - artifact 索引
 - 执行报告
 真正生成的总结 Markdown，仍然可以写到用户指定的 `output_dir`（或默认 `single_summaries/`）。
 这样不破坏当前下游 SOP：
 - `reader-digest-flow` 依然从原来的单篇总结输出目录拿 `.md`
 - job 目录只提供状态与索引，不强迫下游改结果路径约定
 ---
 ## 5. 新增工具 / API 设计
 本轮建议新增 3 个 MCP tool，名字尽量和当前 runtime 风格一致。
 ## 5.1 `start_article_summary_job`
 ### 作用
 启动一个后台 job，立即返回 `job_id`，不等待总结完成。
 ### 输入建议
 ```json
 {
  "extracted_path": "outputs/freshrss/rerun/<run_id>/extracted/item-01.extracted.json",
  "selected_ids": ["12345"],
  "output_dir": "outputs/freshrss/single_summaries/2026-04-10",
  "max_retries": 2,
  "timeout_seconds": 120,
  "llm_api_key": null,
  "llm_model": null,
  "llm_api_url": null
 }
 ```
 ### 返回建议
 ```json
 {
  "job_id": "article-summary-20260410-144500-ab12cd34",
  "workflow": "article_summary_job",
  "run_type": "article_summary",
  "status": "running",
  "output_dir": "outputs/freshrss/article_summary_jobs/article-summary-20260410-144500-ab12cd34",
  "message": "Article summary job started successfully. Use get_article_summary_job_status to poll progress."
 }
 ```
 ### 约束建议
 - `selected_ids` 第一版允许多个，但建议由 OpenClaw 每次只传一篇或少量篇，避免一个 job 干太多事
 - `extracted_path` 必须存在，否则直接拒绝启动
 - `output_dir` 不传则按当前默认逻辑推导
 ---
 ## 5.2 `get_article_summary_job_status`
 ### 作用
 查询 job 当前状态、阶段、进度、错误摘要、已注册 artifact。
 ### 输入
 ```json
 {
  "job_id": "article-summary-20260410-144500-ab12cd34"
 }
 ```
 ### 返回建议
 ```json
 {
  "job_id": "article-summary-20260410-144500-ab12cd34",
  "workflow": "article_summary_job",
  "run_type": "article_summary",
  "status": "running",
  "current_stage": "generate_markdown",
  "started_at": "2026-04-10T14:45:00+08:00",
  "updated_at": "2026-04-10T14:45:23+08:00",
  "finished_at": null,
  "output_dir": "outputs/freshrss/article_summary_jobs/article-summary-20260410-144500-ab12cd34",
  "progress": {
    "completed_stage_count": 2,
    "running_stage_count": 1,
    "failed_stage_count": 0,
    "pending_stage_count": 1,
    "total_stage_count": 4
  },
  "artifacts": [...],
  "error_summary": null
 }
 ```
 ---
 ## 5.3 `get_article_summary_job_result`
 ### 作用
 当 job 成功后，返回结构化结果，供 OpenClaw 继续下游 IMA 沉淀。
 ### 输入
 ```json
 {
  "job_id": "article-summary-20260410-144500-ab12cd34"
 }
 ```
 ### 返回建议
 ```json
 {
  "job_id": "article-summary-20260410-144500-ab12cd34",
  "status": "success",
  "written_paths": [
    "outputs/freshrss/single_summaries/2026-04-10/某篇文章标题.md"
  ],
  "artifact": {
    "name": "job_result",
    "path": "outputs/freshrss/article_summary_jobs/article-summary-20260410-144500-ab12cd34/result.json",
    "kind": "json",
    "stage": "write_result"
  },
  "result": {
    "selected_ids": ["12345"],
    "written_paths": [
      "outputs/freshrss/single_summaries/2026-04-10/某篇文章标题.md"
    ]
  }
 }
 ```
 ### 行为建议
 - 若 job 还没完成，返回当前状态 + 提示“not ready”
 - 若 job 失败，返回失败摘要，不硬抛文件不存在异常
 ---
 ## 6. job 状态文件设计
 建议直接复用现有 `RunState` 模型，不另造一套 schema。
 job 目录示例：
 ```text
 outputs/freshrss/article_summary_jobs/
  article-summary-20260410-144500-ab12cd34/
    run-state.json
    input.json
    result.json
    job-report.json
 ```
 ## 6.1 `run-state.json`
 建议直接沿用当前字段：
 ```json
 {
  "run_id": "article-summary-20260410-144500-ab12cd34",
  "workflow": "article_summary_job",
  "run_type": "article_summary",
  "status": "running",
  "current_stage": "generate_markdown",
  "started_at": "2026-04-10T14:45:00+08:00",
  "updated_at": "2026-04-10T14:45:23+08:00",
  "finished_at": null,
  "input": {
    "extracted_path": "outputs/freshrss/rerun/<run_id>/extracted/item-01.extracted.json",
    "selected_ids": ["12345"],
    "output_dir": "outputs/freshrss/single_summaries/2026-04-10",
    "max_retries": 2,
    "timeout_seconds": 120
  },
  "stages": [...],
  "artifacts": [...],
  "error": null,
  "recovery": {
    "resumable": false,
    "resume_from_stage": null,
    "last_success_stage": "load_input"
  }
 }
 ```
 ### 第一版 stage 建议
 建议只切 4 个 stage，够看即可：
 1. `prepare_job`
   - 校验输入
   - 解析路径
   - 写 `input.json`
 2. `load_input`
   - 读取 extracted payload
   - 确认 `selected_ids` 可匹配条目
 3. `generate_markdown`
   - 调 `summarize_selected_articles(...)`
   - 这是主要耗时阶段
 4. `write_result`
   - 写 `result.json`
   - 注册输出 artifact
 这里不要把 LLM 调用再拆更多细 stage，否则最小版反而过度设计。
 ---
 ## 6.2 `input.json`
 作用：保留启动请求快照，便于排查。
 建议内容与 `run-state.input` 基本一致。
 ---
 ## 6.3 `result.json`
 成功时写：
 ```json
 {
  "job_id": "article-summary-20260410-144500-ab12cd34",
  "selected_ids": ["12345"],
  "written_paths": [
    "outputs/freshrss/single_summaries/2026-04-10/某篇文章标题.md"
  ],
  "completed_at": "2026-04-10T14:45:41+08:00"
 }
 ```
 失败时可以不写，或只写失败快照都行。最小版建议：
 - 成功写 `result.json`
 - 失败只依赖 `run-state.json`
 避免双份失败状态不一致。
 ---
 ## 7. 执行模型建议：优先子进程，不建议线程
 ### 7.1 推荐：子进程后台执行
 最小真异步推荐模型：
 - `start_article_summary_job` 负责：
  - 创建 job 目录
  - 初始化 `run-state.json`
  - 通过 `subprocess.Popen(...)` 启动一个独立 Python 进程执行 job runner
  - 立即返回 `job_id`
 后台 runner 再去：
 - 读取 `input.json`
 - 用 `RunStore` 更新状态
 - 调用 `summarize_selected_articles(...)`
 - 写 `result.json`
 - finish/fail run
 ### 7.2 为什么不推荐线程
 虽然线程实现看起来更省事，但不适合作为 reader 的正式最小异步落地：
 1. **MCP server 进程重启后线程直接丢失**
 2. 线程状态不天然可恢复，容易出现“状态文件还在 running，但线程没了”
 3. 未来要做健康检查/孤儿 job 检测时，线程模型更难收口
 ### 7.3 为什么子进程更贴当前项目风格
 reader 现在本来就偏“文件产物 + runtime 状态真相”风格。子进程模式有天然优势：
 - 和 CLI/fallback 思维一致
 - 进程边界清楚
 - `run-state.json` 由实际执行者写，职责清晰
 - 将来如果要做 orphan detection / stale running job 修复，也容易补
 ### 7.4 本轮不做进程管理增强
 最小版里，不要求：
 - 记录 PID 后做 kill/cancel
 - 自动清理僵尸进程
 - 守护进程/worker 池
 但建议在 `input.json` 或 `run-state.input` 里附带：
 - `launcher_pid`
 - `runner_command`
 方便排障。
 ---
 ## 8. 与现有 `summarize_selected_articles` 的复用关系
 核心原则：**不重写总结业务，只包一层 job runner。**
 ### 8.1 直接复用的部分
 `src/summary_mcp/workflows/article_summary.py` 已经做了：
 - 读取 extracted payload
 - 根据 `selected_ids` 找条目
 - 调用 `run_loop_payload(...)`
 - 渲染 Markdown
 - 写到 `output_dir`
 - 返回 `list[Path]`
 这些都继续用。
 ### 8.2 最小新增建议
 建议只新增一层 runtime/service，例如：
 - `src/summary_mcp/runtime/article_summary_jobs.py`
 职责：
 - 生成 `job_id`
 - 创建 job 目录
 - 初始化 `RunStore`
 - 启动 runner 子进程
 - 提供 status/result 查询
 - 在 runner 里调用 `summarize_selected_articles`
 ### 8.3 是否需要改 `summarize_selected_articles`
 最小版尽量少改，只建议加两类低风险增强：
 1. **可选输入校验增强**
   - 如果 `selected_ids` 一个都匹配不到，显式报错
   - 避免“成功返回空列表”却让 job 看起来像成功
 2. **可选 hook / telemetry（非必须）**
   - 如果后面需要更细粒度写 stage 输出，可再加
   - 但第一版没必要为了观测性重构函数
 结论：
 - 第一版优先保持 `summarize_selected_articles` 基本不动
 - job 层只把它作为黑盒业务函数调用
 ---
 ## 9. 建议新增代码组织
 建议新增文件：
 ```text
 src/summary_mcp/runtime/article_summary_jobs.py
 scripts/run_article_summary_job.py
 ```
 ### 9.1 `article_summary_jobs.py`
 建议包含：
 - `start_article_summary_job(...)`
 - `run_article_summary_job(...)`
 - `get_article_summary_job_status(...)`
 - `get_article_summary_job_result(...)`
 - 若干私有 helper：
  - job id 生成
  - job dir 解析
  - result artifact 读取
 ### 9.2 `scripts/run_article_summary_job.py`
 作用：作为子进程 runner 入口。
 例如：
 ```bash
 python scripts/run_article_summary_job.py --job-id article-summary-... 
 ```
 runner 只做一件事：
 - 根据 `job_id` 找到 job 目录和 `input.json`
 - 真正执行 job
 这样避免在 `Popen("python -c ...")` 里塞长字符串，也方便本地调试。
 ---
 ## 10. 对 `reader-digest-flow` SOP 的影响
 这块是重点，因为老大的实际痛点就在这。
 ### 10.1 现状 SOP
 当前 skill 已经有硬规则：
 - 优先走 MCP `generate_article_summaries`
 - MCP timeout 时，立刻 fallback 到 reader 本地 `.venv`
 这个 fallback 现在是必要的，但它本质是在补“单篇总结没有正式异步接口”。
 ### 10.2 引入异步 job 后的推荐 SOP
 建议调整为：
 1. OpenClaw 在用户确认保留文章后
 2. 调 `start_article_summary_job`
 3. 拿到 `job_id`
 4. 轮询 `get_article_summary_job_status`
 5. 成功后调 `get_article_summary_job_result`
 6. 再继续 IMA 格式化与上传
 ### 10.3 对 skill 文档的影响
 `reader-digest-flow` 需要后续补一条新规则：
 - 单篇总结的正式生产路径从“同步 MCP + 本地 fallback”升级为“异步 job + 状态轮询”
 - 本地 `.venv` CLI fallback 仍保留，但降级为：
  - job 启动失败
  - job runner 异常
  - reader 服务端出现系统性问题时的应急路径
 ### 10.4 用户体验改善
 异步 job 后，OpenClaw 可给出更像正式系统的反馈：
 - “已启动单篇总结任务，正在生成”
 - “当前状态：generate_markdown”
 - “已完成，准备继续沉淀到 IMA”
 而不是现在的：
 - 直接卡住几十秒
 - 然后 tool timeout
 - 再走一套 fallback
 ---
 ## 11. 验证方案
 本轮验证不要追求大而全，按 4 层做就够。
 ### 11.1 单元级验证
 目标：确保 job 状态文件和结果文件行为正确。
 建议覆盖：
 1. `start_article_summary_job` 能创建 job 目录与 `run-state.json`
 2. 输入路径不存在时，启动直接失败
 3. `get_article_summary_job_status` 能正确读取状态
 4. job 成功后 `get_article_summary_job_result` 返回 `written_paths`
 5. job 失败后状态为 `failed`，并带错误摘要
 ### 11.2 本地集成验证
 用一个真实 extracted 文件跑：
 1. 启动 job
 2. 轮询 status
 3. 成功后检查：
   - `result.json` 存在
   - Markdown 文件存在
   - 路径正确
 ### 11.3 OpenClaw 链路验证
 在实际 `reader-digest-flow` 环节，用一篇已确认保留的文章做：
 1. 启动 async job
 2. 等待成功
 3. 继续做 IMA markdown 整理与上传
 4. 确认整个链路不再因为同步 tool timeout 中断
 ### 11.4 异常验证
 至少测 3 类异常：
 1. `selected_ids` 不存在
 2. LLM 接口失败 / 超时
 3. runner 进程异常退出
 预期：
 - `run-state.json` 最终为 `failed`
 - `error_summary` 可读
 - 调用方能明确知道失败，而不是只看到 transport timeout
 ---
 ## 12. 风险与回滚方案
 ## 12.1 风险
 ### 风险 1：后台子进程成功启动，但状态长期卡在 running
 常见原因：
 - runner 进程崩了
 - server 重启时某些路径没写全
 - 子进程命令不对
 缓解：
 - runner 启动前就先写 `run-state.json`
 - runner 一进来先更新 `prepare_job` / `load_input`
 - 后续可加“stale running 超时判定”，但第一版先不做自动修复
 ### 风险 2：复用旧函数导致“空输出也算成功”
 当前 `summarize_selected_articles()` 如果没匹配到文章或全部失败，存在返回空列表的可能。
 缓解：
 - job runner 里把“`written_paths` 为空”视为失败
 - 或者顺手在 `summarize_selected_articles` 里补显式校验
 ### 风险 3：同一时刻大量 job 并发，LLM 调用被打爆
 第一版不解决系统级并发控制。
 缓解：
 - SOP 层先按单篇/少量串行用
 - skill 层避免一口气启动很多 job
 ### 风险 4：状态目录与结果目录分离，排查时容易迷路
 缓解：
 - 在 `result.json` 和 artifact 元数据里明确记录 `written_paths`
 - 在 `run-state.input.output_dir` 中保留结果目录
 ---
 ## 12.2 回滚方案
 这个方案很好回滚，因为是“新增，不替换”。
 ### 回滚原则
 - 保留现有 `generate_article_summaries`
 - 保留现有 `scripts/run_article_summaries.py`
 - 新增 async job 接口如果不稳定，直接停止在 OpenClaw 层使用即可
 ### 回滚路径
 1. 停止调用 `start_article_summary_job`
 2. 恢复回原 SOP：
   - 先尝试同步 MCP `generate_article_summaries`
   - 失败则本地 `.venv` fallback
 3. async job 相关代码保留但不作为正式入口
 也就是说，这轮改动不会堵死当前生产路径，风险可控。
 ---
 ## 13. 实施顺序（建议按这个顺序落地）
 ### Phase A：先把方案落地成最小代码骨架
 1. 新增 `src/summary_mcp/runtime/article_summary_jobs.py`
 2. 新增 job 目录常量与 helper
 3. 新增 `scripts/run_article_summary_job.py`
 4. 实现 runner 内部对 `summarize_selected_articles` 的调用
 5. 先本地命令验证 job 可跑通
 ### Phase B：再把 MCP 接口接上
 6. 在 `server.py` 新增：
   - `start_article_summary_job`
   - `get_article_summary_job_status`
   - `get_article_summary_job_result`
 7. 本地 MCP 调用验证
 ### Phase C：最后接 OpenClaw SOP
 8. 更新 `reader-digest-flow` skill，把正式路径切到 async job
 9. 保留本地 `.venv` fallback 作为应急方案
 10. 跑一次真实日报保留文章沉淀闭环
 ---
 ## 14. 推荐的最小返回/状态语义
 为了跟现有 runtime 风格一致，建议继续用：
 - `status`: `running` / `success` / `failed`
 - `current_stage`: 当前 stage 名
 - `artifacts`: 注册产物列表
 - `error_summary`: 结构化错误
 第一版不必引入：
 - `queued`
 - `cancelled`
 - `retrying`
 - `paused`
 避免状态机一开始就复杂化。
 ---
 ## 15. 结论 / 拍板建议
 拍板建议很直接：
 1. **这件事值得做，而且优先级高**，因为它正好卡在当前日报 SOP 的真实痛点上
 2. **不建议继续先修同步 timeout**，因为同步模型本身就不适合几十秒级、外部 LLM 驱动的任务
 3. **最小真异步应采用“文件状态 + 后台子进程 + start/status/result 三接口”**
 4. **业务层严格复用 `summarize_selected_articles`**，不要为异步化重写 summary 核心逻辑
 5. **先把单篇总结 job 做成独立小 runtime 命名空间**，不要急着并入 freshrss 主 run 的 resume 体系
 如果只允许做一轮最小落地，我建议就做到：
 - `start_article_summary_job`
 - `get_article_summary_job_status`
 - `get_article_summary_job_result`
 - `outputs/freshrss/article_summary_jobs/<job_id>/run-state.json`
 - 子进程 runner
 这套已经足够把当前 OpenClaw timeout 问题从“同步等待”改成“正式异步轮询”，并且几乎不碰无关模块。
@@ -0,0 +1,24 @@
 from __future__ import annotations
 import argparse
 import sys
 from pathlib import Path
 REPO_ROOT = Path(__file__).resolve().parents[1]
 SRC_ROOT = REPO_ROOT / "src"
 if str(SRC_ROOT) not in sys.path:
    sys.path.insert(0, str(SRC_ROOT))
 from summary_mcp.runtime.article_summary_jobs import run_article_summary_job
 def main() -> None:
    parser = argparse.ArgumentParser(description="Run a background article-summary job by job_id.")
    parser.add_argument("--job-id", required=True, help="Article summary job id")
    args = parser.parse_args()
    run_article_summary_job(job_id=args.job_id)
 if __name__ == "__main__":
    main()
@@ -0,0 +1,315 @@
 from __future__ import annotations
 import json
 import os
 import subprocess
 import sys
 from datetime import datetime
 from pathlib import Path
 from typing import Any
 from uuid import uuid4
 from .run_store import RunStore
 from .state_models import RunState
 REPO_ROOT = Path(__file__).resolve().parents[3]
 OUTPUT_ROOT = REPO_ROOT / "outputs" / "freshrss"
 ARTICLE_SUMMARY_JOBS_ROOT = OUTPUT_ROOT / "article_summary_jobs"
 WORKFLOW_NAME = "article_summary_job"
 RUN_TYPE = "article_summary"
 RUN_STATE_FILENAME = "run-state.json"
 INPUT_FILENAME = "input.json"
 RESULT_FILENAME = "result.json"
 JOB_REPORT_FILENAME = "job-report.json"
 DEFAULT_STAGES = [
    "prepare_job",
    "load_input",
    "generate_markdown",
    "write_result",
 ]
 def _now() -> datetime:
    return datetime.now().astimezone()
 def _new_job_id() -> str:
    ts = _now().strftime("%Y%m%d-%H%M%S")
    return f"article-summary-{ts}-{uuid4().hex[:8]}"
 def _job_dir(job_id: str) -> Path:
    return ARTICLE_SUMMARY_JOBS_ROOT / job_id
 def _run_state_path(job_id: str) -> Path:
    return _job_dir(job_id) / RUN_STATE_FILENAME
 def _input_path(job_id: str) -> Path:
    return _job_dir(job_id) / INPUT_FILENAME
 def _result_path(job_id: str) -> Path:
    return _job_dir(job_id) / RESULT_FILENAME
 def _job_report_path(job_id: str) -> Path:
    return _job_dir(job_id) / JOB_REPORT_FILENAME
 def _normalize_repo_path(path: Path) -> str:
    try:
        return str(path.resolve().relative_to(REPO_ROOT.resolve()))
    except ValueError:
        return str(path)
 def _load_run_store(job_id: str) -> RunStore:
    return RunStore.load(path=_run_state_path(job_id), repo_root=REPO_ROOT)
 def _load_result(job_id: str) -> dict[str, Any] | None:
    path = _result_path(job_id)
    if not path.exists():
        return None
    return json.loads(path.read_text(encoding="utf-8-sig"))
 def start_article_summary_job(
    *,
    extracted_path: Path,
    selected_ids: list[str],
    output_dir: Path | None = None,
    max_retries: int = 2,
    timeout_seconds: float = 120.0,
    llm_api_key: str | None = None,
    llm_model: str | None = None,
    llm_api_url: str | None = None,
 ) -> dict[str, Any]:
    if not extracted_path.exists():
        raise FileNotFoundError(f"extracted_path does not exist: {extracted_path}")
    if not selected_ids:
        raise ValueError("selected_ids must not be empty")
    job_id = _new_job_id()
    job_dir = _job_dir(job_id)
    job_dir.mkdir(parents=True, exist_ok=True)
    started_at = _now()
    resolved_output_dir = output_dir or (extracted_path.parent / "single_summaries")
    input_payload = {
        "extracted_path": str(extracted_path),
        "selected_ids": selected_ids,
        "output_dir": str(resolved_output_dir),
        "max_retries": max_retries,
        "timeout_seconds": timeout_seconds,
        "llm_api_key": llm_api_key,
        "llm_model": llm_model,
        "llm_api_url": llm_api_url,
        "launcher_pid": os.getpid(),
    }
    store = RunStore.create(
        path=_run_state_path(job_id),
        run_id=job_id,
        workflow=WORKFLOW_NAME,
        run_type=RUN_TYPE,
        started_at=started_at,
        input_payload=input_payload,
        repo_root=REPO_ROOT,
    )
    for stage_name in DEFAULT_STAGES:
        store._get_or_create_stage(stage_name)
    store.save()
    store.start_stage("prepare_job")
    input_file = _input_path(job_id)
    input_file.write_text(json.dumps(input_payload, ensure_ascii=False, indent=2), encoding="utf-8")
    store.register_artifact(name="job_input", path=input_file, kind="json", stage="prepare_job")
    runner_script = REPO_ROOT / "scripts" / "run_article_summary_job.py"
    cmd = [sys.executable, str(runner_script), "--job-id", job_id]
    try:
        proc = subprocess.Popen(
            cmd,
            cwd=str(REPO_ROOT),
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
        )
    except Exception as exc:
        store.fail_stage("prepare_job", error=exc)
        raise
    store.finish_stage("prepare_job", outputs={"runner_pid": proc.pid, "runner_command": cmd})
    return {
        "job_id": job_id,
        "workflow": WORKFLOW_NAME,
        "run_type": RUN_TYPE,
        "status": "running",
        "output_dir": _normalize_repo_path(job_dir),
        "message": "Article summary job started successfully. Use get_article_summary_job_status to poll progress.",
    }
 def run_article_summary_job(*, job_id: str) -> dict[str, Any]:
    from summary_mcp.workflows.article_summary import ArticleSummaryConfig, summarize_selected_articles
    store = _load_run_store(job_id)
    input_payload = json.loads(_input_path(job_id).read_text(encoding="utf-8-sig"))
    try:
        store.start_stage("load_input")
        extracted_path = Path(input_payload["extracted_path"])
        output_dir = Path(input_payload["output_dir"])
        selected_ids = list(input_payload["selected_ids"])
        if not extracted_path.exists():
            raise FileNotFoundError(f"extracted_path does not exist: {extracted_path}")
        if not selected_ids:
            raise ValueError("selected_ids must not be empty")
        store.finish_stage(
            "load_input",
            outputs={
                "extracted_path": _normalize_repo_path(extracted_path),
                "selected_id_count": len(selected_ids),
                "output_dir": _normalize_repo_path(output_dir),
            },
        )
        store.start_stage("generate_markdown")
        config = ArticleSummaryConfig(
            max_retries=int(input_payload.get("max_retries") or 2),
            timeout_seconds=float(input_payload.get("timeout_seconds") or 120.0),
        )
        written_paths = summarize_selected_articles(
            extracted_path=extracted_path,
            selected_ids=selected_ids,
            output_dir=output_dir,
            config=config,
            api_key=input_payload.get("llm_api_key"),
            model=input_payload.get("llm_model"),
            api_url=input_payload.get("llm_api_url"),
        )
        if not written_paths:
            raise RuntimeError("Article summary job produced no Markdown outputs.")
        normalized_paths = [_normalize_repo_path(Path(p)) for p in written_paths]
        store.finish_stage(
            "generate_markdown",
            outputs={
                "written_count": len(written_paths),
                "written_paths": normalized_paths,
            },
        )
        for idx, p in enumerate(written_paths, start=1):
            store.register_artifact(
                name=f"summary_markdown_{idx}",
                path=Path(p),
                kind="markdown",
                stage="generate_markdown",
                metadata={"output_type": "single_summary"},
            )
        store.start_stage("write_result")
        result = {
            "job_id": job_id,
            "selected_ids": selected_ids,
            "written_paths": normalized_paths,
            "completed_at": _now().isoformat(),
        }
        result_file = _result_path(job_id)
        result_file.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
        store.register_artifact(name="job_result", path=result_file, kind="json", stage="write_result")
        report_file = _job_report_path(job_id)
        report_file.write_text(
            json.dumps(
                {
                    "job_id": job_id,
                    "status": "success",
                    "written_count": len(normalized_paths),
                    "written_paths": normalized_paths,
                },
                ensure_ascii=False,
                indent=2,
            ),
            encoding="utf-8",
        )
        store.register_artifact(name="job_report", path=report_file, kind="json", stage="write_result")
        store.finish_stage("write_result", outputs={"result_path": _normalize_repo_path(result_file)})
        store.finish_run(status="success")
        return result
    except Exception as exc:
        current_stage = store.state.current_stage or "generate_markdown"
        store.fail_stage(current_stage, error=exc)
        report_file = _job_report_path(job_id)
        report_file.write_text(
            json.dumps(
                {
                    "job_id": job_id,
                    "status": "failed",
                    "error_type": type(exc).__name__,
                    "error_message": str(exc),
                    "failed_stage": current_stage,
                },
                ensure_ascii=False,
                indent=2,
            ),
            encoding="utf-8",
        )
        try:
            store.register_artifact(name="job_report", path=report_file, kind="json", stage=current_stage)
        except Exception:
            pass
        raise
 def get_article_summary_job_status(*, job_id: str) -> dict[str, Any]:
    store = _load_run_store(job_id)
    state = store.state
    completed_stage_count = sum(1 for s in state.stages if s.status == "success")
    running_stage_count = sum(1 for s in state.stages if s.status == "running")
    failed_stage_count = sum(1 for s in state.stages if s.status == "failed")
    pending_stage_count = sum(1 for s in state.stages if s.status == "pending")
    return {
        "job_id": state.run_id,
        "workflow": state.workflow,
        "run_type": state.run_type,
        "status": state.status,
        "current_stage": state.current_stage,
        "started_at": state.started_at.isoformat(),
        "updated_at": state.updated_at.isoformat(),
        "finished_at": state.finished_at.isoformat() if state.finished_at else None,
        "output_dir": _normalize_repo_path(_job_dir(job_id)),
        "progress": {
            "completed_stage_count": completed_stage_count,
            "running_stage_count": running_stage_count,
            "failed_stage_count": failed_stage_count,
            "pending_stage_count": pending_stage_count,
            "total_stage_count": len(state.stages),
        },
        "artifacts": [artifact.model_dump(mode="json") for artifact in state.artifacts],
        "error_summary": state.error.model_dump(mode="json") if state.error else None,
    }
 def get_article_summary_job_result(*, job_id: str) -> dict[str, Any]:
    store = _load_run_store(job_id)
    state = store.state
    result = _load_result(job_id)
    if state.status != "success" or result is None:
        return {
            "job_id": state.run_id,
            "status": state.status,
            "message": "Article summary job result is not ready.",
            "error_summary": state.error.model_dump(mode="json") if state.error else None,
        }
    artifact = next((a.model_dump(mode="json") for a in state.artifacts if a.name == "job_result"), None)
    return {
        "job_id": state.run_id,
        "status": state.status,
        "written_paths": result.get("written_paths", []),
        "artifact": artifact,
        "result": result,
    }
@@ -20,6 +20,11 @@ from summary_mcp.runtime import get_run_status as load_run_status
 from summary_mcp.runtime import get_run_report as load_run_report
 from summary_mcp.runtime import list_run_artifacts as load_run_artifacts
 from summary_mcp.runtime import list_runs as load_runs
 from summary_mcp.runtime.article_summary_jobs import (
    get_article_summary_job_result as load_article_summary_job_result,
    get_article_summary_job_status as load_article_summary_job_status,
    start_article_summary_job as launch_article_summary_job,
 )
 from summary_mcp.runtime.resume_service import resume_run as resume_existing_run
 from summary_mcp.workflows import run_freshrss_pipeline
 from summary_mcp.workflows.article_summary import ArticleSummaryConfig, summarize_selected_articles
@@ -164,6 +169,42 @@ def resume_run(run_id: str) -> dict:
    """Resume a failed or interrupted FreshRSS workflow run from its latest supported recovery point."""
    return resume_existing_run(run_id=run_id)
@mcp.tool()
 def start_article_summary_job(
    *,
    extracted_path: str,
    selected_ids: list[str],
    output_dir: str | None = None,
    max_retries: int = 2,
    timeout_seconds: float = 120.0,
    llm_api_key: str | None = None,
    llm_model: str | None = None,
    llm_api_url: str | None = None,
 ) -> dict:
    """Start an asynchronous article-summary job and return a job_id immediately."""
    return launch_article_summary_job(
        extracted_path=Path(extracted_path),
        selected_ids=selected_ids,
        output_dir=Path(output_dir) if output_dir else None,
        max_retries=max_retries,
        timeout_seconds=timeout_seconds,
        llm_api_key=llm_api_key,
        llm_model=llm_model,
        llm_api_url=llm_api_url,
    )
@mcp.tool()
 def get_article_summary_job_status(job_id: str) -> dict:
    """Get the current status of an asynchronous article-summary job."""
    return load_article_summary_job_status(job_id=job_id)
@mcp.tool()
 def get_article_summary_job_result(job_id: str) -> dict:
    """Get the final result of an asynchronous article-summary job."""
    return load_article_summary_job_result(job_id=job_id)
@mcp.tool()
 def generate_article_summaries(
@@ -167,10 +167,12 @@ def summarize_selected_articles(
    output_dir.mkdir(parents=True, exist_ok=True)
    written_paths: list[Path] = []
    matched_ids: set[str] = set()
    from summary_mcp.core.summary_loop import build_summary_input
    for item_id, entry in _iter_selected_items(payload, selected_ids):
        matched_ids.add(item_id)
        # Prefer the real project format where each entry has ``item`` and
        # ``extraction.article``; fall back to legacy layout where the
        # article fields live directly on the element.
@@ -288,4 +290,7 @@ def summarize_selected_articles(
        output_path.write_text("\n".join(lines), encoding="utf-8")
        written_paths.append(output_path)
    if not matched_ids:
        raise ValueError(f"No extracted entries matched selected_ids: {list(selected_ids)}")
    return written_paths