From a7b9166c0e195c88b274cd493ad6d5b2e35dd336 Mon Sep 17 00:00:00 2001 From: adarshh347 Date: Tue, 27 Oct 2026 10:32:43 +0530 Subject: [PATCH] fix(daily-paper): isolate per-paper analysis failures --- .../src/reme_daily_paper/analyze.py | 27 ++++++-- plugins/daily_paper/tests/test_daily_paper.py | 63 +++++++++++++++++++ 2 files changed, 86 insertions(+), 4 deletions(-) diff --git a/plugins/daily_paper/src/reme_daily_paper/analyze.py b/plugins/daily_paper/src/reme_daily_paper/analyze.py index a56ccddd..5fc10ecb 100644 --- a/plugins/daily_paper/src/reme_daily_paper/analyze.py +++ b/plugins/daily_paper/src/reme_daily_paper/analyze.py @@ -194,21 +194,40 @@ 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, - ), + ) + 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_count": len(analyses), + "failure_count": len(failures), + "failed_papers": list(failures), + "note_paths": [paper.note_path for paper in analyses], + "pdf_paths": [paper.pdf_path for paper in analyses], + }, ) self._set_state("analyses", analyses) - self.context.response.answer = f"Agent wrote {len(analyses)} detailed paper notes" + 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/tests/test_daily_paper.py b/plugins/daily_paper/tests/test_daily_paper.py index a6d514a9..77c83dc2 100644 --- a/plugins/daily_paper/tests/test_daily_paper.py +++ b/plugins/daily_paper/tests/test_daily_paper.py @@ -1016,3 +1016,66 @@ 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"] == []