fix(daily-paper): expose recovery details in failure answers

This commit is contained in:
adarshh347 2026-10-28 18:09:30 +05:30
parent 2ff331a338
commit c012baa930
5 changed files with 128 additions and 12 deletions

View file

@ -96,7 +96,7 @@ within each request. Downloads and PDF extraction are not repeated by this recov
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.
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

View file

@ -131,7 +131,7 @@ python -m pytest plugins/daily_paper -v
成功一篇或两篇时,仅根据成功的笔记生成部分简报,并列出失败的 arXiv ID。此时 `success=true`,
`metadata.partial=true`;`analysis_count`、`failure_count`、`failed_papers`、`note_paths` 和 `pdf_paths`
提供结果明细。全部失败时 `success=false`,不调用简报模型。若简报本身生成失败,已完成笔记的路径仍保留在响应元数据中。
提供结果明细。全部失败时 `success=false`,不调用简报模型。若简报本身生成失败,已完成笔记的路径仍保留在响应元数据中。失败回答还会列出已完成笔记路径、失败的 arXiv ID 和简短错误原因,默认 CLI 和 MCP 调用方无需显示元数据即可获得恢复信息。
已保存的部分简报沿用跳过策略;再次调用会返回 `partial=true` 和保存的失败 ID。使用
`reme daily_paper date=YYYY-MM-DD force=true` 重新生成,会重新运行正常选题流程,而非仅恢复失败论文。

View file

@ -240,7 +240,6 @@ class DailyPaperAnalyzeStep(DailyPaperStep):
"pdf_paths": [paper.pdf_path for paper in analyses],
},
)
self._set_state("analyses", analyses)
self.context.response.success = bool(analyses)
self.context.response.answer = (
f"Agent wrote {len(analyses)} detailed paper notes; {len(failures)} paper(s) failed"

View file

@ -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():
@ -73,11 +103,11 @@ class DailyPaperDigestStep(DailyPaperStep):
raise RuntimeError("An agent_wrapper is required for the daily brief")
analyses: list[AnalyzedPaper] = self._state("analyses") or []
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.success = False
self.context.response.answer = "Daily paper analysis failed for every selected paper; no brief generated"
self.context.response.metadata["partial"] = False
return self.context.response
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",
@ -128,8 +158,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 failures:
body += "\n\n## 未完成的论文\n\n" + "\n".join(f"- {item['arxiv_id']}" for item in failures)
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,
@ -140,8 +170,8 @@ class DailyPaperDigestStep(DailyPaperStep):
"kind": "daily-paper-brief",
"date": day,
"arxiv_ids": selected_ids,
"partial": bool(failures),
"failed_arxiv_ids": [item["arxiv_id"] for item in failures],
"partial": partial,
"failed_arxiv_ids": failed_arxiv_ids,
"selection_reasoning": [item.reasoning for item in analyses],
"source_notes": wikilinks,
"generated_at": utc_now_iso(),
@ -167,12 +197,12 @@ class DailyPaperDigestStep(DailyPaperStep):
self.logger.info(f"[{self.name}] refresh index done date={day}")
self.context.response.success = True
qualifier = f"partial ({len(analyses)}/{PAPER_COUNT} papers) " if failures else ""
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(
{
"date": day,
"partial": bool(failures),
"partial": partial,
"analysis_count": len(analyses),
"failure_count": len(failures),
"failed_papers": list(failures),

View file

@ -1126,3 +1126,90 @@ async def test_digest_reports_partial_and_total_failure(tmp_path, count):
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())