diff --git a/plugins/daily_paper/README.md b/plugins/daily_paper/README.md index 2f60ca25..fa837635 100644 --- a/plugins/daily_paper/README.md +++ b/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 | diff --git a/plugins/daily_paper/README_ZH.md b/plugins/daily_paper/README_ZH.md index cb809870..58350574 100644 --- a/plugins/daily_paper/README_ZH.md +++ b/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` 理解为三篇论文全部成功。最终自动标签步骤仍可能报告自己的错误。 diff --git a/plugins/daily_paper/pyproject.toml b/plugins/daily_paper/pyproject.toml index eb131365..16bf9437 100644 --- a/plugins/daily_paper/pyproject.toml +++ b/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"] diff --git a/plugins/daily_paper/src/reme_daily_paper/analyze.py b/plugins/daily_paper/src/reme_daily_paper/analyze.py index cc41de08..a2ecaf33 100644 --- a/plugins/daily_paper/src/reme_daily_paper/analyze.py +++ b/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 diff --git a/plugins/daily_paper/src/reme_daily_paper/analyze.yaml b/plugins/daily_paper/src/reme_daily_paper/analyze.yaml index 65b869f4..28f411f7 100644 --- a/plugins/daily_paper/src/reme_daily_paper/analyze.yaml +++ b/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 完整,不要重复上一次的长篇输出。 diff --git a/plugins/daily_paper/src/reme_daily_paper/base.py b/plugins/daily_paper/src/reme_daily_paper/base.py index bf61b2da..6dd2a033 100644 --- a/plugins/daily_paper/src/reme_daily_paper/base.py +++ b/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"\A---[ \t]*\r?\n.*?^---[ \t]*(?:\r?\n|$)", re.DOTALL | re.MULTILINE) _MARKDOWN_HEADING_PATTERN = re.compile(r"^#+\s*") _UNSAFE_FILENAME_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]') _CHINESE_PATTERN = re.compile(r"[\u3400-\u4dbf\u4e00-\u9fff]") @@ -31,7 +30,10 @@ _OutputT = TypeVar("_OutputT", bound=BaseModel) def structured_output(result: dict[str, Any], model: type[_OutputT]) -> _OutputT: """Validate an agent wrapper's structured output.""" if result["status"] != "completed": - raise RuntimeError(result["error"] or f"Agent reply ended: {result['status']}") + error = result["error"] or f"Agent reply ended: {result['status']}" + if error == "Codex returned invalid JSON for the requested output_schema": + raise ValueError(error) + raise RuntimeError(error) return model.model_validate(result["structured_output"]) @@ -58,6 +60,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()) @@ -145,19 +171,11 @@ 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.""" - path.parent.mkdir(parents=True, exist_ok=True) + """Write atomically under the path lock, cleaning surrogate text first.""" 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: diff --git a/plugins/daily_paper/src/reme_daily_paper/collect.py b/plugins/daily_paper/src/reme_daily_paper/collect.py index e6d5832e..4d150e29 100644 --- a/plugins/daily_paper/src/reme_daily_paper/collect.py +++ b/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 diff --git a/plugins/daily_paper/src/reme_daily_paper/digest.py b/plugins/daily_paper/src/reme_daily_paper/digest.py index 7638c83a..fb0563f0 100644 --- a/plugins/daily_paper/src/reme_daily_paper/digest.py +++ b/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,16 +156,20 @@ 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, { "name": title, "title": title, - "description": output.desc.strip(), + "description": output.desc, "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( diff --git a/plugins/daily_paper/src/reme_daily_paper/digest.yaml b/plugins/daily_paper/src/reme_daily_paper/digest.yaml index dd6c3b84..fad7d1fa 100644 --- a/plugins/daily_paper/src/reme_daily_paper/digest.yaml +++ b/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,也不要重复一级标题。 - 总阅读时长约五分钟。 - - 包含“今日一句话”“每篇一分钟读懂”“三篇论文之间的联系”“今天最值得关注什么”等部分。 + - 包含“今日一句话”“每篇一分钟读懂”“今天最值得关注什么”等部分。多篇论文时比较它们的联系;只有一篇时不要虚构比较对象。 diff --git a/plugins/daily_paper/tests/test_daily_paper.py b/plugins/daily_paper/tests/test_daily_paper.py index 2c11d81a..8c023035 100644 --- a/plugins/daily_paper/tests/test_daily_paper.py +++ b/plugins/daily_paper/tests/test_daily_paper.py @@ -628,6 +628,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("/"), ) @@ -1021,3 +1022,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() diff --git a/plugins/daily_paper/tests/test_structured_recovery.py b/plugins/daily_paper/tests/test_structured_recovery.py new file mode 100644 index 00000000..b483ef1b --- /dev/null +++ b/plugins/daily_paper/tests/test_structured_recovery.py @@ -0,0 +1,211 @@ +"""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.base import markdown_output +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 current Codex result normalization while replacing the app-server turn boundary.""" + from reme.components.agent_wrapper import CodexAgentWrapper + from openai_codex.generated.v2_all import AgentMessageThreadItem, Turn + + app = ApplicationContext(workspace_dir=str(tmp_path)) + wrapper = CodexAgentWrapper() + valid = json.dumps({"title": "论文摘要", "desc": "summary", "body": "Supported result"}) + responses = ['{"body":"unfinished', valid if recover else '{"body":"unfinished'] + calls = [] + + async def reply(_prompt, **kwargs): + calls.append(_prompt) + item = AgentMessageThreadItem(type="agentMessage", id=f"item-{len(calls)}", text=responses[len(calls) - 1]) + turn = Turn(id=f"turn-{len(calls)}", items=[item], status="completed", duration_ms=1, error=None) + return wrapper._finish_reply("offline-thread", turn, [], None, None, kwargs) + + step = DailyPaperAnalyzeStep(app_context=app) + step.agent_wrapper = SimpleNamespace(reply=reply) + 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 len(calls) == 2 + instruction = step.prompt_format("analyze_short_retry") + assert instruction not in calls[0] + assert instruction in calls[1] + downloader.download.assert_awaited_once() + + +@pytest.mark.parametrize("empty_frontmatter", ["---\n---", "---\n---\n", "---\r\n---\r\n"]) +@pytest.mark.asyncio +async def test_empty_frontmatter_retries_once_without_repeating_pdf_work(tmp_path, monkeypatch, empty_frontmatter): + """An empty frontmatter-only note triggers one shorter retry after one PDF extraction.""" + app = ApplicationContext(workspace_dir=str(tmp_path)) + valid = {"title": "论文摘要", "desc": "summary", "body": "Recovered result"} + reply = AsyncMock( + side_effect=[ + { + "status": "completed", + "error": None, + "structured_output": {"title": "论文摘要", "desc": "summary", "body": empty_frontmatter}, + }, + {"status": "completed", "error": None, "structured_output": valid}, + ], + ) + step = DailyPaperAnalyzeStep(app_context=app) + step.agent_wrapper = SimpleNamespace(reply=reply) + step.context = RuntimeContext(daily_paper_run_date="2026-07-21") + extraction_calls = [] + + def extract(*_args): + extraction_calls.append(True) + return "Evidence [p. 1]", 1, False + + monkeypatch.setattr(step, "_extract_pdf_text_sync", extract) + downloader = SimpleNamespace(download=AsyncMock()) + paper = PaperInfo(arxiv_id="2607.10001") + pick = PaperPick(arxiv_id=paper.arxiv_id, reasoning="test") + + result = await step._analyze_one(downloader, paper, pick, set()) + + note = (tmp_path / result.note_path).read_text(encoding="utf-8") + assert "Recovered result" in note + assert reply.await_count == 2 + downloader.download.assert_awaited_once() + assert extraction_calls == [True] + + +@pytest.mark.parametrize( + ("body", "expected"), + [ + ("---\n---\nUseful note", "Useful note"), + ("---\n---\n\nUseful note", "Useful note"), + ("---\r\ntitle: generated\r\n---\r\nUseful note", "Useful note"), + ], +) +def test_frontmatter_removal_preserves_content(body, expected): + """Removing a document-leading frontmatter block must retain following Markdown.""" + result = markdown_output( + { + "status": "completed", + "error": None, + "structured_output": {"title": "Title", "desc": "Description", "body": body}, + }, + ) + + assert result.body == expected + + +def test_horizontal_rules_in_markdown_body_are_not_frontmatter(): + """Frontmatter removal must only consider a delimiter at the document start.""" + result = markdown_output( + { + "status": "completed", + "error": None, + "structured_output": { + "title": "Title", + "desc": "Description", + "body": "# Findings\n\nIntroduction.\n\n---\nImportant supported finding.\n---\nConclusion.", + }, + }, + ) + + assert result.body == "# Findings\n\nIntroduction.\n\n---\nImportant supported finding.\n---\nConclusion." diff --git a/tests/unit/test_package_versions.py b/tests/unit/test_package_versions.py index f1c34c4a..d8d89277 100644 --- a/tests/unit/test_package_versions.py +++ b/tests/unit/test_package_versions.py @@ -224,7 +224,8 @@ 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.1.12") not in by_name["reme-ai"].specifier + assert Version("0.4.2.0") in by_name["reme-ai"].specifier assert "pypdf" in by_name