Merge 24db61e35a into c6049c3fc1

plugins/daily_paper/README.md

@@ -13,7 +13,7 @@ their Job configuration under `application_defaults`. Enable the installed plugi
 ### 1. Install ReMe and Daily Paper
 
 ```bash
-python -m pip install "reme-ai[core]>=0.4.1.12"
+python -m pip install "reme-ai[core]>=0.4.2.0"
 reme plugins install plugins/daily_paper
 ```
 
@@ -86,6 +86,23 @@ three source notes, and rebuilds the daily index. The workflow then runs `auto_t
 frontmatter of all three analyses and the final brief. The normal background file watcher observes those source-file
 changes and refreshes derived indexes.
 
+## Partial results and recovery
+
+Each selected paper is analyzed independently. An ordinary failure is recorded and later papers are still attempted;
+cancellation propagates. Invalid structured output gets one additional request asking for a shorter note. This is a
+prompt-level size target, not a provider output-token guarantee; the SDK may try its own structured-output strategies
+within each request. Downloads and PDF extraction are not repeated by this recovery.
+
+With one or two successful notes, the workflow writes a partial brief using only those notes and lists the failed
+arXiv IDs. A generated partial brief has `success=true` and `metadata.partial=true`; `analysis_count`, `failure_count`,
+`failed_papers`, `note_paths`, and `pdf_paths` report the outcome. An all-failed batch has `success=false` and makes no
+brief-generation call. If brief generation itself fails, completed note paths remain in response metadata. Failure answers also list completed-note paths, failed arXiv IDs and concise error reasons, so default CLI and MCP callers can recover without displaying metadata.
+
+Saved partial briefs follow the existing skip policy. A later call reports `partial=true` and the saved failed IDs;
+use `reme daily_paper date=YYYY-MM-DD force=true` to regenerate. This starts the usual selection pipeline again; it is
+not a failed-paper-only resume. Hosts should inspect `partial` rather than treating `success=true` as three successful
+analyses. The final tagging step can still report its own failure after a brief is written.
+
 ## Parameters
 
 | Parameter       | Default | Purpose                                                                           |

plugins/daily_paper/README_ZH.md

@@ -11,7 +11,7 @@ Step backend，并在 `application_defaults` 下提供 Job 配置；通过 `plug
 ### 1. 安装 ReMe 和每日论文插件
 
 ```bash
-python -m pip install "reme-ai[core]>=0.4.1.12"
+python -m pip install "reme-ai[core]>=0.4.2.0"
 reme plugins install plugins/daily_paper
 ```
 
@@ -121,3 +121,18 @@ python -m pytest plugins/daily_paper -v
 ```
 
 单元测试 mock Hugging Face、arXiv 和 AgentScope 边界，不访问外部服务。
+
+
+## 部分结果与恢复
+
+每篇论文独立处理，普通错误会被记录，后续论文仍会继续；任务取消会向上传播。结构化输出无效时，
+最多额外请求一次更短的笔记，不重复下载和 PDF 提取。这只是提示词中的长度目标，不是服务端输出 token
+限制；每次请求内部仍可能使用 SDK 的结构化输出策略回退。
+
+成功一篇或两篇时，仅根据成功的笔记生成部分简报，并列出失败的 arXiv ID。此时 `success=true`，
+`metadata.partial=true`；`analysis_count`、`failure_count`、`failed_papers`、`note_paths` 和 `pdf_paths`
+提供结果明细。全部失败时 `success=false`，不调用简报模型。若简报本身生成失败，已完成笔记的路径仍保留在响应元数据中。失败回答还会列出已完成笔记路径、失败的 arXiv ID 和简短错误原因，默认 CLI 和 MCP 调用方无需显示元数据即可获得恢复信息。
+
+已保存的部分简报沿用跳过策略；再次调用会返回 `partial=true` 和保存的失败 ID。使用
+`reme daily_paper date=YYYY-MM-DD force=true` 重新生成，会重新运行正常选题流程，而非仅恢复失败论文。
+调用方应检查 `partial`，不要把 `success=true` 理解为三篇论文全部成功。最终自动标签步骤仍可能报告自己的错误。

plugins/daily_paper/pyproject.toml

@@ -8,7 +8,7 @@ license-files = ["LICENSE"]
 requires-python = ">=3.11"
 dependencies = [
     "pypdf>=5.0.0",
-    "reme-ai>=0.4.1.12",
+    "reme-ai>=0.4.2.0",
 ]
 
 [project.entry-points."reme.plugins"]

plugins/daily_paper/src/reme_daily_paper/analyze.py

@@ -4,16 +4,17 @@ import asyncio
 import json
 from pathlib import Path
 
+
 from .arxiv import ArxivPdfClient
 from .base import (
     PAPER_COUNT,
     DailyPaperStep,
+    analysis_summary_metadata,
     iter_note_metadata,
+    markdown_output,
     normalize_chinese_title,
     replace_surrogates,
     reserve_unique_note_path,
-    strip_frontmatter,
-    structured_output,
     utc_now_iso,
     write_markdown,
 )
@@ -99,19 +100,33 @@ class DailyPaperAnalyzeStep(DailyPaperStep):
             f"chars={len(pdf_text)} truncated={truncated}",
         )
         self.logger.info(f"[{self.name}] agent start arxiv_id={paper.arxiv_id}")
-        result = await self.agent_wrapper.reply(
-            self.prompt_format(
-                "analyze_user",
-                paper_info=json.dumps(paper.model_dump(), ensure_ascii=False, indent=2),
-                selection_reason=selected.reasoning,
-                page_count=page_count,
-                truncated=str(truncated).lower(),
-                pdf_text=pdf_text,
-            ),
-            output_schema=DailyPaperMarkdownOutput,
+        prompt = self.prompt_format(
+            "analyze_user",
+            paper_info=json.dumps(paper.model_dump(), ensure_ascii=False, indent=2),
+            selection_reason=selected.reasoning,
+            page_count=page_count,
+            truncated=str(truncated).lower(),
+            pdf_text=pdf_text,
         )
+        # AgentScope is optional: Codex JSON parsing and Pydantic validation both raise ValueError.
+        recoverable = (ValueError,)
+        try:
+            from agentscope.exception import StructuredOutputError
+        except ImportError:
+            pass
+        else:
+            recoverable += (StructuredOutputError,)
+        for attempt in range(2):
+            try:
+                result = await self.agent_wrapper.reply(prompt, output_schema=DailyPaperMarkdownOutput)
+                output = markdown_output(result)
+                break
+            except recoverable:
+                if attempt:
+                    raise
+                self.logger.warning(f"[{self.name}] retry shorter structured note arxiv_id={paper.arxiv_id}")
+                prompt += "\n\n" + self.prompt_format("analyze_short_retry")
         self.logger.info(f"[{self.name}] agent done arxiv_id={paper.arxiv_id}")
-        output = structured_output(result, DailyPaperMarkdownOutput)
         title = normalize_chinese_title(output.title, f"论文解读-{paper.arxiv_id}")
         day_dir = self.workspace_path / daily_dir / day
         existing_note = self._find_existing_note(day_dir, paper.arxiv_id)
@@ -126,10 +141,7 @@ class DailyPaperAnalyzeStep(DailyPaperStep):
         ) as (title, note_path):
             used_titles.add(title)
             note_rel = note_path.relative_to(self.workspace_path).as_posix()
-            desc = replace_surrogates(output.desc.strip())
-            body = replace_surrogates(strip_frontmatter(output.body))
-            if not desc or not body:
-                raise ValueError(f"Agent returned an empty paper note for {paper.arxiv_id}")
+            desc, body = output.desc, output.body
             await write_markdown(
                 note_path,
                 body,
@@ -195,21 +207,31 @@ class DailyPaperAnalyzeStep(DailyPaperStep):
         self.logger.info(f"[{self.name}] start papers={len(selected)}")
 
         analyses: list[AnalyzedPaper] = []
+        failures: list[dict[str, str]] = []
+        self._set_state("analyses", analyses)
+        self._set_state("failures", failures)
         used_titles: set[str] = set()
         async with ArxivPdfClient(
             timeout=float(self._value("pdf_timeout", 600.0)),
             max_bytes=int(self._value("max_pdf_bytes", 50 * 1024 * 1024)),
         ) as downloader:
             for item in selected:
-                analyses.append(
-                    await self._analyze_one(
+                try:
+                    analysis = await self._analyze_one(
                         downloader,
                         candidate_map[item.arxiv_id],
                         item,
                         used_titles,
-                    ),
-                )
-        self._set_state("analyses", analyses)
-        self.context.response.answer = f"Agent wrote {len(analyses)} detailed paper notes"
+                    )
+                except Exception as exc:  # Cancellation must propagate, not become a paper failure.
+                    failures.append({"arxiv_id": item.arxiv_id, "error": f"{type(exc).__name__}: {exc}"})
+                    self.logger.warning(f"[{self.name}] paper failed arxiv_id={item.arxiv_id}: {exc}")
+                else:
+                    analyses.append(analysis)
+                self.context.response.metadata.update(analysis_summary_metadata(analyses, failures))
+        self.context.response.success = bool(analyses)
+        self.context.response.answer = (
+            f"Agent wrote {len(analyses)} detailed paper notes; {len(failures)} paper(s) failed"
+        )
         self.logger.info(f"[{self.name}] finish notes={len(analyses)}")
         return self.context.response

plugins/daily_paper/src/reme_daily_paper/analyze.yaml

@@ -31,3 +31,9 @@ analyze_user: |
   # PDF 分页文本
 
   {pdf_text}
+
+analyze_short_retry: |
+  上一次结构化输出无效或被截断。请重新生成更短的完整笔记：body 不超过 800 个中文字符，
+  desc 不超过 100 个中文字符，保留 title、desc、body 三个字段。
+  只保留核心问题、主要方法、一个有依据的结果及局限；省略冗长背景和重复内容。
+  不得补充输入材料之外的事实。优先确保 JSON 完整，不要重复上一次的长篇输出。

plugins/daily_paper/src/reme_daily_paper/base.py

@@ -1,26 +1,25 @@
 """Shared state and file helpers for daily-paper steps."""
 
 import datetime as dt
-import os
 import re
 import zoneinfo
 from collections.abc import Iterator
 from contextlib import asynccontextmanager
 from pathlib import Path
 from typing import Any, TypeVar
-from uuid import uuid4
 
-import aiofiles
 import frontmatter
 from pydantic import BaseModel
 
 from reme.steps import BaseStep
-from reme.steps.file_io import get_path_lock, validate_filename_component
+from reme.steps.file_io import get_path_lock, validate_filename_component, write_file_safe
+
+from .schema import AnalyzedPaper, DailyPaperMarkdownOutput
 
 # Number of papers selected, analyzed, and digested each run. Shared across steps.
 PAPER_COUNT = 3
 _STATE_PREFIX = "daily_paper_"
-_FRONTMATTER_PATTERN = re.compile(r"^---\s*\n.*?\n---\s*\n", re.DOTALL)
+_FRONTMATTER_PATTERN = re.compile(r"^---\s*\n.*?\n---(?:\s*\n|$)", re.DOTALL)
 _MARKDOWN_HEADING_PATTERN = re.compile(r"^#+\s*")
 _UNSAFE_FILENAME_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
 _CHINESE_PATTERN = re.compile(r"[\u3400-\u4dbf\u4e00-\u9fff]")
@@ -57,6 +56,30 @@ def _replace_surrogates_recursive(value: Any) -> Any:
     return value
 
 
+def markdown_output(result: dict[str, Any]) -> DailyPaperMarkdownOutput:
+    """Clean and validate usable Markdown before callers persist or retry it."""
+    output = structured_output(result, DailyPaperMarkdownOutput)
+    desc = replace_surrogates(output.desc.strip())
+    body = replace_surrogates(strip_frontmatter(output.body))
+    if not desc or not body:
+        raise ValueError("Agent returned an empty Markdown description or body")
+    return output.model_copy(update={"desc": desc, "body": body})
+
+
+def analysis_summary_metadata(
+    analyses: list[AnalyzedPaper],
+    failures: list[dict[str, str]],
+) -> dict[str, Any]:
+    """Build shared progress metadata for analyze and standalone digest steps."""
+    return {
+        "analysis_count": len(analyses),
+        "failure_count": len(failures),
+        "failed_papers": list(failures),
+        "note_paths": [item.note_path for item in analyses],
+        "pdf_paths": [item.pdf_path for item in analyses],
+    }
+
+
 def normalize_chinese_title(raw: str, fallback: str) -> str:
     """Return one safe Chinese title that can also be used as the filename stem."""
     title = replace_surrogates(str(raw or "").strip())
@@ -144,19 +167,12 @@ async def reserve_unique_note_path(
 
 
 async def write_atomic(path: Path, content: str | bytes) -> None:
-    """Write through a sibling temporary file under the repository path lock."""
+    """Write atomically under the path lock, cleaning surrogate text first."""
     path.parent.mkdir(parents=True, exist_ok=True)
     lock = await get_path_lock(path)
     async with lock:
-        temp_path = path.with_name(f".{path.name}.{uuid4().hex}.tmp")
-        payload = replace_surrogates(content).encode("utf-8") if isinstance(content, str) else content
-        try:
-            async with aiofiles.open(temp_path, "wb") as stream:
-                await stream.write(payload)
-            os.replace(temp_path, path)
-        finally:
-            if temp_path.exists():
-                temp_path.unlink()
+        payload = replace_surrogates(content) if isinstance(content, str) else content
+        await write_file_safe(path, payload)
 
 
 async def write_markdown(path: Path, body: str, metadata: dict[str, Any]) -> None:

plugins/daily_paper/src/reme_daily_paper/collect.py

@@ -63,11 +63,11 @@ class DailyPaperCollectStep(DailyPaperStep):
         return found
 
     @staticmethod
-    def find_saved_digest(day_dir: Path) -> Path | None:
-        """Find a generated brief by frontmatter instead of by its model-generated title."""
+    def find_saved_digest(day_dir: Path) -> tuple[Path, dict] | None:
+        """Find a generated brief and return its already-read frontmatter."""
         for path, metadata in iter_note_metadata(day_dir):
             if metadata.get("kind") == "daily-paper-brief" or path.name == "daily-paper-brief.md":
-                return path
+                return path, metadata
         return None
 
     @staticmethod
@@ -83,6 +83,7 @@ class DailyPaperCollectStep(DailyPaperStep):
         return PaperInfo.model_validate(values)
 
     async def execute(self):
+        """Load prior brief state or collect the current weekly and monthly papers."""
         assert self.context is not None
         timezone = self.app_context.app_config.timezone if self.app_context is not None else None
         raw_date = str(self._value("date", "") or "").strip()
@@ -93,17 +94,29 @@ class DailyPaperCollectStep(DailyPaperStep):
         daily_dir = str(self.config_value("daily_dir")).strip("/")
         force = bool(self._value("force", False))
         self.logger.info(f"[{self.name}] start date={day} force={force}")
-        digest_path = self.find_saved_digest(self.workspace_path / daily_dir / day)
-        if digest_path is not None:
+        saved_digest = self.find_saved_digest(self.workspace_path / daily_dir / day)
+        if saved_digest is not None:
+            digest_path, saved_metadata = saved_digest
             digest_rel = digest_path.relative_to(self.workspace_path).as_posix()
             self._set_state("existing_digest_path", digest_rel)
             if not force:
                 self._set_state("skip", True)
                 self._set_state("digest_path", digest_rel)
                 self.context.response.success = True
-                self.context.response.answer = f"Skipped: daily paper brief already exists at {digest_rel}"
+                partial = bool(saved_metadata.get("partial", False))
+                qualifier = "partial " if partial else ""
+                self.context.response.answer = (
+                    f"Skipped: {qualifier}daily paper brief already exists at {digest_rel}"
+                    + ("; use force=true to regenerate" if partial else "")
+                )
                 self.context.response.metadata.update(
-                    {"date": day, "digest_path": digest_rel, "skipped": True},
+                    {
+                        "date": day,
+                        "digest_path": digest_rel,
+                        "skipped": True,
+                        "partial": partial,
+                        "failed_arxiv_ids": saved_metadata.get("failed_arxiv_ids", []),
+                    },
                 )
                 self.logger.info(f"[{self.name}] skip existing digest path={digest_rel}")
                 return self.context.response

plugins/daily_paper/src/reme_daily_paper/digest.py

@@ -11,10 +11,10 @@ from reme.steps.file_io import refresh_day_index
 from .base import (
     PAPER_COUNT,
     DailyPaperStep,
+    analysis_summary_metadata,
+    markdown_output,
     normalize_chinese_title,
     reserve_unique_note_path,
-    strip_frontmatter,
-    structured_output,
     utc_now_iso,
     write_markdown,
 )
@@ -63,7 +63,37 @@ class DailyPaperDigestStep(DailyPaperStep):
 
         return _WIKILINK_RE.sub(replace, body)
 
+    def _failure(self, reason: str):
+        """Keep recovery information visible to callers that only display answer."""
+
+        def concise(value):
+            return " ".join(str(value).split())[:300]
+
+        lines = [f"Daily Paper failed: {concise(reason)}"]
+        analyses = self._state("analyses") or []
+        lines.append("Completed notes:")
+        lines.extend(f"- {item.note_path}" for item in analyses)
+        if not analyses:
+            lines.append("- None")
+        failures = self._state("failures") or []
+        if failures:
+            lines.append("Failed papers:")
+            lines.extend(f"- {item['arxiv_id']}: {concise(item['error'])}" for item in failures)
+        if digest_path := self._state("digest_path"):
+            lines.append(f"Brief already written: {digest_path}")
+        self.context.response.success = False
+        self.context.response.answer = "\n".join(lines)
+        return self.context.response
+
     async def execute(self):
+        """Render plugin failures before BaseJob replaces their recovery details."""
+        assert self.context is not None
+        try:
+            return await self._generate_digest()
+        except Exception as exc:  # Cancellation continues to propagate.
+            return self._failure(f"{type(exc).__name__}: {exc}")
+
+    async def _generate_digest(self):
         """Generate and persist the final brief from analyzed papers."""
         assert self.context is not None
         if self._skip():
@@ -72,13 +102,19 @@ class DailyPaperDigestStep(DailyPaperStep):
         if self.agent_wrapper is None:
             raise RuntimeError("An agent_wrapper is required for the daily brief")
         analyses: list[AnalyzedPaper] = self._state("analyses") or []
-        if len(analyses) != PAPER_COUNT:
+        failures = self._state("failures") or []
+        partial = bool(failures)
+        failed_arxiv_ids = [item["arxiv_id"] for item in failures]
+        if not analyses and failures:
+            self.context.response.metadata["partial"] = False
+            return self._failure("Every selected paper failed; no brief generated")
+        if not analyses or len(analyses) + len(failures) != PAPER_COUNT:
             raise RuntimeError(
                 "Detailed paper notes are missing before digest generation",
             )
         self.logger.info(f"[{self.name}] start notes={len(analyses)}")
 
-        documents = [{"title": item.title, "desc": item.desc, "body": item.body} for item in analyses]
+        documents = [item.model_dump(include={"title", "desc", "body"}) for item in analyses]
         wikilinks = [f"[[{item.note_path}]]" for item in analyses]
         run_day = dt.date.fromisoformat(self._run_day())
         daily_dir = str(self.config_value("daily_dir")).strip("/")
@@ -88,6 +124,7 @@ class DailyPaperDigestStep(DailyPaperStep):
                 "digest_user",
                 documents=json.dumps(documents, ensure_ascii=False, indent=2),
                 daily_dir=daily_dir,
+                paper_count=len(analyses),
             ),
             output_schema=DailyPaperMarkdownOutput,
             job_tools=list(self.kwargs.get("job_tools") or []),
@@ -99,10 +136,8 @@ class DailyPaperDigestStep(DailyPaperStep):
             },
         )
         self.logger.info(f"[{self.name}] agent done notes={len(analyses)}")
-        output = structured_output(result, DailyPaperMarkdownOutput)
-        body = strip_frontmatter(output.body)
-        if not output.desc.strip() or not body:
-            raise ValueError("Agent returned an empty daily paper brief")
+        output = markdown_output(result)
+        body = output.body
 
         day = run_day.isoformat()
         title = normalize_chinese_title(output.title, f"每日论文简报-{day}")
@@ -121,6 +156,8 @@ class DailyPaperDigestStep(DailyPaperStep):
             body = self._validate_historical_wikilinks(body, run_day, digest_path)
             body += "\n\n## 详细论文\n\n" + "\n".join(f"- {link}" for link in wikilinks)
             selected_ids = [item.arxiv_id for item in analyses]
+            if partial:
+                body += "\n\n## 未完成的论文\n\n" + "\n".join(f"- {arxiv_id}" for arxiv_id in failed_arxiv_ids)
             await write_markdown(
                 digest_path,
                 body,
@@ -131,6 +168,8 @@ class DailyPaperDigestStep(DailyPaperStep):
                     "kind": "daily-paper-brief",
                     "date": day,
                     "arxiv_ids": selected_ids,
+                    "partial": partial,
+                    "failed_arxiv_ids": failed_arxiv_ids,
                     "selection_reasoning": [item.reasoning for item in analyses],
                     "source_notes": wikilinks,
                     "generated_at": utc_now_iso(),
@@ -156,16 +195,17 @@ class DailyPaperDigestStep(DailyPaperStep):
         self.logger.info(f"[{self.name}] refresh index done date={day}")
 
         self.context.response.success = True
-        self.context.response.answer = f"Generated daily paper brief: {digest_rel}"
+        qualifier = f"partial ({len(analyses)}/{PAPER_COUNT} papers) " if partial else ""
+        self.context.response.answer = f"Generated {qualifier}daily paper brief: {digest_rel}"
         self.context.response.metadata.update(
             {
+                **analysis_summary_metadata(analyses, failures),
                 "date": day,
+                "partial": partial,
                 "week": self._state("week"),
                 "month": self._state("month"),
                 "selection_reasoning": [item.reasoning for item in analyses],
                 "selected_arxiv_ids": selected_ids,
-                "note_paths": [item.note_path for item in analyses],
-                "pdf_paths": [item.pdf_path for item in analyses],
                 "digest_path": digest_rel,
                 "source_counts": self._state("source_counts"),
                 "excluded_yesterday_count": len(

plugins/daily_paper/src/reme_daily_paper/digest.yaml

@@ -1,22 +1,22 @@
 # Prompt bundled with the Daily Paper plugin distribution.
 digest_user: |
-  你是中文 AI 研究资讯主编。请忠实综合下面三篇详细论文解读，生成一篇普通读者五分钟可以读懂的每日论文速读。
+  你是中文 AI 研究资讯主编。请忠实综合下面{paper_count} 篇详细论文解读，生成一篇普通读者五分钟可以读懂的每日论文速读。
   内容只能依据输入文档，不得补充文档中没有提供的事实。
-  保留技术准确性，同时解释三篇论文为什么值得关注，以及它们之间有什么联系。
+  保留技术准确性，同时解释{paper_count} 篇论文为什么值得关注，以及它们之间有什么联系。
 
-  下面提供的三篇详细论文解读是本次写作的主要输入，首要任务是基于它们生成每日论文速读。
-  历史搜索只用于帮助构建 Wikilink，最多调用两次 `search`：围绕三篇论文共同的核心问题、方法、关键词和
+  下面提供的{paper_count} 篇详细论文解读是本次写作的主要输入，首要任务是基于它们生成每日论文速读。
+  历史搜索只用于帮助构建 Wikilink，最多调用两次 `search`：围绕{paper_count} 篇论文共同的核心问题、方法、关键词和
   同义表达组织一次综合查询，搜索结果不必局限于 `{daily_dir}/`。只有 `{daily_dir}/` 下日期早于今天、
   且与本期内容确实相似或互补的 Markdown 文章才可作为正文中的历史链接候选；必要时调用 `read` 核验全文，
   不要仅凭标题判断。如果没有搜索到合适的历史文章，说明已有记忆中暂无相似文章，不添加 Wikilink 即可，
-  仍须基于三篇论文解读完成每日论文速读；不得将搜索无结果误判为论文输入缺失，也不得要求用户重新提供文档。
+  仍须基于{paper_count} 篇论文解读完成每日论文速读；不得将搜索无结果误判为论文输入缺失，也不得要求用户重新提供文档。
   将确认相关的旧文章以 Wikilink 自然织入正文，并用句子说明关联（延续、对比、补充或方法相似）；
   链接必须采用带 `.md` 的完整 workspace-relative 路径，例如
   `[[{daily_dir}/2026-07-01/旧文章.md|此前的相关解读]]`。不要输出裸链接、独立关系字段，也不要虚构搜索未命中的路径。
   旧文章只用于判断关联和建立链接，不得用来补充本期事实。如果没有真正相关的旧文章，不要强行添加；
-  当日三篇详细解读的链接会由系统统一附在文末。
+  当日{paper_count} 篇详细解读的链接会由系统统一附在文末。
 
-  # 三篇论文解读
+  # {paper_count} 篇论文解读
 
   {documents}
 
@@ -24,4 +24,4 @@ digest_user: |
   - title 必须是简洁准确的中文标题，不要包含 Markdown 标题符号、路径或 .md 后缀。
   - 只返回结构化结果中的 title、desc 和 body；body 不要包含 YAML frontmatter，也不要重复一级标题。
   - 总阅读时长约五分钟。
-  - 包含“今日一句话”“每篇一分钟读懂”“三篇论文之间的联系”“今天最值得关注什么”等部分。
+  - 包含“今日一句话”“每篇一分钟读懂”“今天最值得关注什么”等部分。多篇论文时比较它们的联系；只有一篇时不要虚构比较对象。

plugins/daily_paper/tests/test_daily_paper.py

@@ -623,6 +623,7 @@ def test_digest_prompt_uses_configured_daily_directory(tmp_path: Path):
     prompt = step.prompt_format(
         "digest_user",
         documents="[]",
+        paper_count=3,
         daily_dir=str(step.config_value("daily_dir")).strip("/"),
     )
 
@@ -1016,3 +1017,234 @@ async def test_digest_validates_model_generated_historical_wikilinks(tmp_path: P
     assert "非 daily 节点" in rendered and "[[digest/wiki/相关概念.md" not in rendered
     assert "越界路径" in rendered and "../outside.md" not in rendered
     assert "[[daily/2026-07-21/论文解读1.md]]" in rendered
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("failed_indices", [{0}, {2}, {0, 1, 2}, set()])
+async def test_analysis_isolates_paper_failures(tmp_path, monkeypatch, failed_indices):
+    """Ordinary failures retain completed notes and do not skip later selections."""
+    papers = [_paper(f"2607.1000{i}") for i in range(3)]
+    context = RuntimeContext(
+        daily_paper_run_date="2026-07-21",
+        daily_paper_selected=[PaperPick(arxiv_id=p.arxiv_id, reasoning="test") for p in papers],
+        daily_paper_candidates=papers,
+    )
+    step = DailyPaperAnalyzeStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))
+    attempted = []
+
+    async def analyze_one(_downloader, paper, _selected, _used_titles):
+        index = papers.index(paper)
+        attempted.append(index)
+        if index in failed_indices:
+            raise ValueError("invalid structured note")
+        note_path = f"daily/2026-07-21/{paper.arxiv_id}.md"
+        await write_markdown(tmp_path / note_path, "retained evidence", {})
+        return AnalyzedPaper(
+            arxiv_id=paper.arxiv_id,
+            reasoning="test",
+            title="论文",
+            desc="test",
+            body="retained evidence",
+            note_path=note_path,
+            pdf_path=f"resource/papers/{paper.arxiv_id}.pdf",
+        )
+
+    monkeypatch.setattr(step, "_analyze_one", analyze_one)
+    await step(context)
+    assert attempted == [0, 1, 2]
+    assert len(context["daily_paper_analyses"]) == 3 - len(failed_indices)
+    metadata = context.response.metadata
+    assert context.response.success == (len(failed_indices) < 3)
+    assert metadata["failure_count"] == len(failed_indices)
+    assert metadata["analysis_count"] == 3 - len(failed_indices)
+    assert [item["arxiv_id"] for item in metadata["failed_papers"]] == [
+        papers[i].arxiv_id for i in sorted(failed_indices)
+    ]
+    assert all((tmp_path / path).is_file() for path in metadata["note_paths"])
+
+
+@pytest.mark.asyncio
+async def test_analysis_does_not_swallow_cancellation(tmp_path, monkeypatch):
+    """Cancellation stops the batch instead of reporting an ordinary failed paper."""
+    import asyncio
+
+    papers = [_paper(f"2607.1000{i}") for i in range(3)]
+    context = RuntimeContext(
+        daily_paper_selected=[PaperPick(arxiv_id=p.arxiv_id, reasoning="test") for p in papers],
+        daily_paper_candidates=papers,
+    )
+    step = DailyPaperAnalyzeStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))
+    analyze_one = AsyncMock(side_effect=asyncio.CancelledError)
+    monkeypatch.setattr(step, "_analyze_one", analyze_one)
+    with pytest.raises(asyncio.CancelledError):
+        await step(context)
+    assert analyze_one.await_count == 1
+    assert context["daily_paper_failures"] == []
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("count", [0, 1, 2, 3])
+async def test_digest_reports_partial_and_total_failure(tmp_path, count):
+    """Generate a useful brief from any successes and avoid a model call on total failure."""
+    analyses = [
+        AnalyzedPaper(
+            arxiv_id=f"2607.1000{i}",
+            reasoning="test",
+            title=f"论文{i}",
+            desc="test",
+            body="evidence",
+            note_path=f"daily/2026-07-21/论文{i}.md",
+            pdf_path=f"resource/papers/{i}.pdf",
+        )
+        for i in range(count)
+    ]
+    failures = [{"arxiv_id": f"2607.1000{i}", "error": "invalid output"} for i in range(count, 3)]
+    context = RuntimeContext(
+        daily_paper_run_date="2026-07-21",
+        daily_paper_analyses=analyses,
+        daily_paper_failures=failures,
+    )
+    agent = _QueuedAgentWrapper([{"title": "今日简报", "desc": "summary", "body": "Grounded brief"}])
+    await DailyPaperDigestStep(
+        app_context=ApplicationContext(workspace_dir=str(tmp_path)),
+        agent_wrapper=agent,
+    )(context)
+    assert context.response.success == bool(count)
+    if not count:
+        assert not agent.calls
+        assert not list(tmp_path.rglob("*.md"))
+        return
+    assert context.response.metadata["partial"] == (count < 3)
+    assert context.response.metadata["failure_count"] == 3 - count
+    assert context.response.metadata["note_paths"] == [item.note_path for item in analyses]
+    brief = frontmatter.load(tmp_path / context.response.metadata["digest_path"])
+    assert brief.metadata["failed_arxiv_ids"] == [item["arxiv_id"] for item in failures]
+    assert all(item["arxiv_id"] in brief.content for item in failures)
+    assert f"{count} 篇" in agent.calls[0]["inputs"]
+    if count < 3:
+        rerun = RuntimeContext(date="2026-07-21")
+        await DailyPaperCollectStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))(rerun)
+        assert rerun.response.metadata["partial"] is True
+        assert "force=true" in rerun.response.answer
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("note_count", [0, 2])
+async def test_failure_recovery_details_reach_job_cli_and_mcp(tmp_path, capsys, note_count):
+    """Real job and service adapters expose recovery details without showing metadata."""
+    from types import SimpleNamespace
+
+    from fastmcp.exceptions import ToolError
+    from reme.components.job import BaseJob
+    from reme.components.service.cli_service import CliService
+    from reme.components.service.mcp_tools import add_mcp_job
+
+    analyses = [
+        AnalyzedPaper(
+            arxiv_id=f"2607.1000{i}",
+            reasoning="test",
+            title=f"论文{i}",
+            desc="test",
+            body="evidence",
+            note_path=f"daily/2026-07-21/论文{i}.md",
+            pdf_path=f"resource/papers/{i}.pdf",
+        )
+        for i in range(note_count)
+    ]
+    for item in analyses:
+        await write_markdown(tmp_path / item.note_path, item.body, {})
+    failures = [{"arxiv_id": f"2607.1000{i}", "error": "ValueError: invalid note"} for i in range(note_count, 3)]
+    app_context = ApplicationContext(workspace_dir=str(tmp_path))
+    agent = _QueuedAgentWrapper([])
+    agent.reply = AsyncMock(side_effect=RuntimeError("brief provider failed"))
+    job = BaseJob(
+        name="daily_paper",
+        app_context=app_context,
+        daily_paper_run_date="2026-07-21",
+        daily_paper_analyses=analyses,
+        daily_paper_failures=failures,
+    )
+    job.step_specs = [(DailyPaperDigestStep, {"app_context": app_context, "agent_wrapper": agent})]
+    response = await job()
+    assert response.success is False
+    expected = [item.note_path for item in analyses] + [item["arxiv_id"] for item in failures] + ["invalid note"]
+    if note_count:
+        expected.append("brief provider failed")
+    assert all(value in response.answer for value in expected)
+
+    async def run_job(_name, **kwargs):
+        return await job(**kwargs)
+
+    app = SimpleNamespace(start=AsyncMock(), close=AsyncMock(), run_job=run_job)
+    capsys.readouterr()
+    with pytest.raises(SystemExit) as stopped:
+        await CliService(job="daily_paper")._run_job(app)  # pylint: disable=protected-access
+    assert stopped.value.code == 1
+    stderr = capsys.readouterr().err
+    assert all(value in stderr for value in expected)
+    app.close.assert_awaited_once()
+
+    for error_mode in (False, True):
+        registered = []
+        add_mcp_job(
+            SimpleNamespace(add_tool=registered.append),
+            job,
+            injected_job_kwargs={},
+            tool_error_on_failure=error_mode,
+        )
+        if error_mode:
+            with pytest.raises(ToolError) as error:
+                await registered[0].fn()
+            rendered = str(error.value)
+        else:
+            rendered = await registered[0].fn()
+        assert all(value in rendered for value in expected)
+    if not note_count:
+        agent.reply.assert_not_awaited()
+    assert all((tmp_path / item.note_path).is_file() for item in analyses)
+
+
+@pytest.mark.asyncio
+async def test_digest_failure_boundary_preserves_cancellation(tmp_path):
+    """The answer-rendering boundary must not turn cancellation into a failed response."""
+    import asyncio
+
+    step = DailyPaperDigestStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)))
+    step._generate_digest = AsyncMock(side_effect=asyncio.CancelledError)  # pylint: disable=protected-access
+    with pytest.raises(asyncio.CancelledError):
+        await step(RuntimeContext())
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "invalid",
+    [
+        {"title": "论文", "desc": " ", "body": "content"},
+        {"title": "论文", "desc": "summary", "body": "\n "},
+        {"title": "论文", "desc": "summary", "body": "---\ntitle: generated\n---\n"},
+    ],
+)
+@pytest.mark.parametrize("recover", [True, False])
+async def test_blank_note_retries_once_before_writing(tmp_path, monkeypatch, invalid, recover):
+    """Blank or frontmatter-only content uses the same bounded shorter-note recovery."""
+    from types import SimpleNamespace
+
+    valid = {"title": "论文", "desc": " summary ", "body": "---\ntitle: model\n---\nEvidence"}
+    agent = _QueuedAgentWrapper([invalid, valid if recover else invalid])
+    step = DailyPaperAnalyzeStep(app_context=ApplicationContext(workspace_dir=str(tmp_path)), agent_wrapper=agent)
+    step.context = RuntimeContext(daily_paper_run_date="2026-07-21")
+    monkeypatch.setattr(step, "_extract_pdf_text_sync", lambda *_: ("evidence", 1, False))
+    downloader = SimpleNamespace(download=AsyncMock())
+    paper = _paper("2607.10001")
+    pick = PaperPick(arxiv_id=paper.arxiv_id, reasoning="test")
+    if recover:
+        result = await step._analyze_one(downloader, paper, pick, set())  # pylint: disable=protected-access
+        assert result.desc == "summary" and result.body == "Evidence"
+        assert (tmp_path / result.note_path).is_file()
+    else:
+        with pytest.raises(ValueError, match="empty Markdown"):
+            await step._analyze_one(downloader, paper, pick, set())  # pylint: disable=protected-access
+        assert not list(tmp_path.rglob("*.md"))
+    assert len(agent.calls) == 2
+    assert step.prompt_format("analyze_short_retry") in agent.calls[1]["inputs"]
+    downloader.download.assert_awaited_once()

plugins/daily_paper/tests/test_structured_recovery.py

@@ -0,0 +1,136 @@
+"""Offline recovery tests through the actual AgentScope structured-output parser."""
+
+# pylint: disable=protected-access
+
+import json
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+from pydantic import BaseModel
+
+from reme_daily_paper.analyze import DailyPaperAnalyzeStep
+from reme_daily_paper.schema import PaperInfo, PaperPick
+from reme.components import ApplicationContext, RuntimeContext
+from reme.components.agent_wrapper import AsAgentWrapper
+
+# The plugin also supports wrappers without the optional AgentScope installation.
+ChatModelBase = pytest.importorskip("agentscope.model").ChatModelBase
+ChatResponse = pytest.importorskip("agentscope.model").ChatResponse
+ChatUsage = pytest.importorskip("agentscope.model").ChatUsage
+OpenAIChatFormatter = pytest.importorskip("agentscope.formatter").OpenAIChatFormatter
+TextBlock = pytest.importorskip("agentscope.message").TextBlock
+ToolCallBlock = pytest.importorskip("agentscope.message").ToolCallBlock
+
+
+class _Parameters(BaseModel):
+    """No provider parameters are needed for this offline model."""
+
+
+class _TruncatedModel(ChatModelBase):
+    """Emit malformed tool arguments until the plugin asks for a shorter note."""
+
+    def __init__(self, recover):
+        super().__init__(credential=None, model="offline", parameters=_Parameters(), stream=False)
+        self.formatter = OpenAIChatFormatter()
+        self.recover = recover
+        self.retry_instruction = ""
+        self.analysis_calls = 0
+        self.structured_calls = 0
+
+    async def _call_api(self, *_args, **kwargs):
+        """Replace provider I/O while retaining real agent and parser behavior."""
+        messages = kwargs.get("messages") or []
+        for message in messages:
+            if message.role == "user":
+                message.id = "forced-user-id-800"
+        tools = kwargs.get("tools") or []
+        structured = any(item.get("function", {}).get("name") == "generate_structured_output" for item in tools)
+        if structured:
+            self.structured_calls += 1
+            shortened = any(
+                message.role == "user" and self.retry_instruction in message.get_text_content() for message in messages
+            )
+            raw = json.dumps({"title": "论文摘要", "desc": "summary", "body": "Supported result"})
+            if not (self.recover and shortened):
+                raw = '{"body":"' + "unfinished note " * 2000
+            content = [ToolCallBlock(id="offline-call", name="generate_structured_output", input=raw)]
+        else:
+            self.analysis_calls += 1
+            content = [TextBlock(text="Supported analysis")]
+        return ChatResponse(
+            content=content,
+            is_last=True,
+            usage=ChatUsage(input_tokens=10, output_tokens=10, time=0.01),
+        )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("recover", [True, False])
+async def test_shorter_retry_uses_real_structured_parser(tmp_path, monkeypatch, recover):
+    """A shorter second request recovers or fails without starting a third attempt."""
+    app = ApplicationContext(workspace_dir=str(tmp_path))
+    model = _TruncatedModel(recover)
+    agent = AsAgentWrapper(app_context=app, as_llm="")
+    agent.as_llm = SimpleNamespace(model=model)
+    step = DailyPaperAnalyzeStep(app_context=app, agent_wrapper=agent)
+    model.retry_instruction = step.prompt_format("analyze_short_retry")
+    step.context = RuntimeContext(daily_paper_run_date="2026-07-21")
+    monkeypatch.setattr(step, "_extract_pdf_text_sync", lambda *_: ("Supported result [p. 1]", 1, False))
+    downloader = SimpleNamespace(download=AsyncMock())
+    paper = PaperInfo(arxiv_id="2607.10001")
+    pick = PaperPick(arxiv_id=paper.arxiv_id, reasoning="test")
+    if recover:
+        result = await step._analyze_one(downloader, paper, pick, set())
+        assert (tmp_path / result.note_path).is_file()
+    else:
+        from agentscope.exception import StructuredOutputError
+
+        with pytest.raises(StructuredOutputError):
+            await step._analyze_one(downloader, paper, pick, set())
+        assert not list(tmp_path.rglob("*.md"))
+    assert model.analysis_calls == 2
+    assert model.structured_calls >= 2
+    assert downloader.download.await_count == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("recover", [True, False])
+async def test_codex_json_parse_failure_gets_one_shorter_retry(tmp_path, monkeypatch, recover):
+    """Use the real Codex reply parser with only SDK/lifecycle boundaries replaced."""
+    from reme.components.agent_wrapper import CodexAgentWrapper
+
+    app = ApplicationContext(workspace_dir=str(tmp_path))
+    wrapper = CodexAgentWrapper(app_context=app)
+    valid = json.dumps({"title": "论文摘要", "desc": "summary", "body": "Supported result"})
+    thread = SimpleNamespace(
+        id="offline-thread",
+        run=AsyncMock(
+            side_effect=[
+                SimpleNamespace(final_response='{"body":"unfinished', usage=None),
+                SimpleNamespace(final_response=valid if recover else '{"body":"unfinished', usage=None),
+            ],
+        ),
+    )
+    monkeypatch.setattr(wrapper, "start", AsyncMock())
+    monkeypatch.setattr(wrapper, "_get_codex", AsyncMock(return_value=object()))
+    monkeypatch.setattr(wrapper, "_open_thread", AsyncMock(return_value=thread))
+    monkeypatch.setattr(wrapper, "_release_thread", AsyncMock())
+    monkeypatch.setattr(wrapper, "_serialize", lambda _: {})
+    step = DailyPaperAnalyzeStep(app_context=app, agent_wrapper=wrapper)
+    step.context = RuntimeContext(daily_paper_run_date="2026-07-21")
+    monkeypatch.setattr(step, "_extract_pdf_text_sync", lambda *_: ("Evidence [p. 1]", 1, False))
+    downloader = SimpleNamespace(download=AsyncMock())
+    paper = PaperInfo(arxiv_id="2607.10001")
+    pick = PaperPick(arxiv_id=paper.arxiv_id, reasoning="test")
+    if recover:
+        result = await step._analyze_one(downloader, paper, pick, set())
+        assert (tmp_path / result.note_path).is_file()
+    else:
+        with pytest.raises(ValueError, match="Codex returned invalid JSON"):
+            await step._analyze_one(downloader, paper, pick, set())
+    assert thread.run.await_count == 2
+    instruction = step.prompt_format("analyze_short_retry")
+    assert instruction not in thread.run.await_args_list[0].args[0]
+    assert instruction in thread.run.await_args_list[1].args[0]
+    downloader.download.assert_awaited_once()

tests/unit/test_package_versions.py

@@ -224,7 +224,7 @@ def test_daily_paper_declares_runtime_dependencies() -> None:
 
     assert not by_name["reme-ai"].extras
     assert Version("0.4.1.11") not in by_name["reme-ai"].specifier
-    assert Version("0.4.1.12") in by_name["reme-ai"].specifier
+    assert Version("0.4.2.0") in by_name["reme-ai"].specifier
     assert "pypdf" in by_name