From d8ad0be94d1129502218c15b1952d4e06c1a02fd Mon Sep 17 00:00:00 2001 From: kyle <308708576+swfuzkwu-ui@users.noreply.github.com> Date: Mon, 24 Aug 2026 10:33:31 +0800 Subject: [PATCH] feat(reader): add traceable M06 paper reading --- .ai-team/TASK.md | 113 +- README.md | 1 + docs/API.md | 19 +- docs/ARCHITECTURE.md | 16 +- docs/FUNCTION_MATRIX.md | 5 +- docs/M06_PAPER_READER.md | 100 ++ docs/WORKFLOW_AND_STATE.md | 10 +- src/autoresearch/agents/base.py | 8 +- src/autoresearch/agents/paper_reader.py | 255 ++- src/autoresearch/api.py | 53 +- src/autoresearch/application.py | 265 ++- src/autoresearch/cli.py | 52 +- src/autoresearch/contracts.py | 572 ++++++- src/autoresearch/evidence.py | 7 +- src/autoresearch/handoffs.py | 23 +- src/autoresearch/reader_parser.py | 997 +++++++++++ src/autoresearch/reader_service.py | 1998 +++++++++++++++++++++-- src/autoresearch/writing_service.py | 29 +- tests/test_reader_graph.py | 212 +++ tests/test_reader_m06.py | 1368 ++++++++++++++++ tests/test_workflow.py | 82 +- 21 files changed, 5917 insertions(+), 268 deletions(-) create mode 100644 docs/M06_PAPER_READER.md create mode 100644 src/autoresearch/reader_parser.py create mode 100644 tests/test_reader_graph.py create mode 100644 tests/test_reader_m06.py diff --git a/.ai-team/TASK.md b/.ai-team/TASK.md index 8982ec2..1e9b1df 100644 --- a/.ai-team/TASK.md +++ b/.ai-team/TASK.md @@ -1,61 +1,110 @@ # Current Task -- ID: `AR-003` -- Title: `Simplify the assignment plan to module level` -- Status: `done` -- Owner: `zzg` -- Next owner: `user/team` +- ID: `AR-010` +- Title: `将可追溯的 M06 论文读取接入主 Research LangGraph` +- Status: `handoff` +- Owner: `kyle` +- Next owner: `kyle/项目评审` ## Goal -根据用户反馈，把任务拆解入口从 122 个子任务收敛为模块级分配。保留此前详细版作为附录，但当前团队先按模块认领负责人、边界、主要产出、依赖和验收目标。 +把调用方显式提供并声明有权处理的本地、可提取文本论文转换为可定位、可校验、可追问的阅读证据， +并接入主 +Research LangGraph。当前切片承诺全文解析、稳定 locator、标准 ReadingCard 十项合同、 +claim-evidence 绑定、单次证据问答和论文级失败恢复；章节级阅读陪跑仍是独立功能块。 ## Acceptance scenarios -- [x] `docs/AutoResearch_任务拆解书.md` 只作为模块级分配入口，覆盖 M00–M17。 -- [x] 每个模块说明目标、边界、主要交付物、前置模块、负责人建议和当前状态。 -- [x] 文档提供模块负责人验收模板和第一批模块分配建议。 -- [x] 原 122 子任务版本保留为 `docs/AutoResearch_任务拆解书_详细版.md`，不丢失后续拆解素材。 -- [x] README、Project-to-Act 和 VibeCollab 任务状态保持一致，不改变五业务 Agent 和证据门禁边界。 +- [x] 多页 PDF 被解析为稳定文档块，第二页结果保留第二页 locator 和全文 checksum。 +- [x] ReadingCard 中的结构化 claim 可通过 evidence ID 回溯到同一论文和有效原文块；仅有元数据或 + 摘要时不会冒充全文证据。 +- [x] 单次论文问答只从可定位原文句子中选择答案；来源不支持时返回 `unresolved=true`。 +- [x] Paper Reader 子图在 `ResearchState` 中保存有界的论文级进度，重放时不复制已完成阅读产物， + checkpoint 中不保存全文。 +- [x] 缺失、损坏、空白、加密、需要 OCR 或不支持的文档产生结构化诊断；同批一篇失败不会抹掉 + 已成功论文。 +- [x] 现有 CLI/API 保持兼容，业务 Agent 仍严格为五个，离线 Ruff、pytest、doctor 和任务检查通过。 +- [x] 标准 ReadingCard 覆盖规范引用、版本、M05 OA/许可快照、全文 hash、研究问题/作者主张、 + 方法与实验要素、精确结果定位、适用边界、复现资源、idea 关系、创新分栏、待补读项、置信度及 + 审核状态；事实项均复用可回查的 claim/evidence/locator。 +- [ ] 章节级 checkpoint、暂停/恢复和按当前段落装配上下文属于独立功能块，当前没有实现，也不计入 + 本切片完成项。 ## Invariants -- 当前入口不要求团队一次性拆到函数、接口或工单；模块负责人后续再自行细化。 -- 详细版是参考材料，不代表所有子任务都已实现或已验收。 -- 真实论文试点、生产数据库、Web UI、灾备和性能仍保持原有“外部阻塞/待开发”状态。 -- 不提交 token、密钥、私有来源、完整论文、原始提示或虚构外部结果。 +- 业务 Agent 始终保持五个。解析、locator 校验、证据持久化和 checkpoint 记账是确定性服务或节点， + 不伪装成额外 Agent。 +- 不绕过付费墙，不提交凭据、私有/完整论文、原始提示词或模型原始输出；测试只使用运行时生成的 + fixture。 +- 全文和无界文本块不得进入 `ResearchState` 或 `HandoffEnvelope`；图状态只保存稳定 ID、有界摘要、 + 进度和诊断。 +- 模型可以理解给定原文块，但不能编造页码、章节、block locator、evidence ID、证据等级或状态迁移。 +- 来源缺失或 provenance 无效时必须 fail closed；摘要/题录降级必须显式标注，不能取得全文证据等级。 +- 当前不重做 M05 检索/OA/许可，不重做 M07 创新挖掘，也不实现 OCR、复杂表格/公式解析、 + Web UI 或生产存储。 +- 现有 `POST /papers/{id}/questions` 仅为无会话的单次证据问答兼容接口，不得在文档或验收中称为 + “章节级阅读陪跑”。 ## Decisions -- 使用 M00–M17 作为稳定模块 ID，先按模块分配负责人。 -- 保留详细拆解版，避免未来重新整理时丢失依赖和验收素材。 -- `.project-to-act` 继续作为长期项目事实源；`.ai-team/TASK.md` 只记录本次短周期交付。 +- 本任务基于 `main` 提交 `46fbd7a`；开放中的 AR-008 V2 仍是明确集成风险，不能静默采用其合同设计。 +- 首个切片支持调用方显式提供并声明有权处理的本地 born-digital PDF、UTF-8 TXT 和 Markdown。扫描 PDF 返回 + `ocr_required`，不伪造正文。 +- 复用现有父图和 SQLite checkpointer，在原 Paper Reader 内扩展子图，不增加第六个业务 Agent。 +- 保持已有 `ReadingCard`/`ReadingAnswer` 消费方兼容，同时增加有类型的 provenance 和 claim 绑定。 +- 按项目负责人确认的口径把章节级阅读陪跑视为独立功能块：不删除现有单次问答接口，不在本任务中实现陪读会话， + 并保留 parser/block/locator 作为未来陪读模块可复用的底座。 +- 按 M05 PR 当前合同，M05 只保存独立 `paper_open_access` 记录，包含 OA 状态、合法全文入口和 + license，不修改 `PaperRecord`、不下载 PDF。M06 只按 `paper_id` 读取该记录作为卡片元数据，绝不把 + `fulltext_url` 当作 `full_text_path`；正文仍只来自显式提供的本地文件。 ## Completed -- 将原 `docs/AutoResearch_任务拆解书.md` 移为 `docs/AutoResearch_任务拆解书_详细版.md`。 -- 新建模块级 `docs/AutoResearch_任务拆解书.md`，包含 18 个模块、执行阶段、负责人模板和首批分配建议。 -- 保留 README 任务书入口，并将 AR-003 的状态、范围和下一步写入协作账本。 +- 增加解析文档、文档块、locator、读取诊断、claim 绑定、ReadingCard/ReadingAnswer 扩展和 Reader 进度合同。 +- 增加基于内容寻址的 PDF/TXT/Markdown 解析器，保存页/块定位、原文 checksum、缓存完整性校验、 + 大小/页数/字符上限、摘要降级和显式失败类型。 +- 重构 `PaperReaderService`：缓存抽取 profile，验证模型返回的 block/quote/claim，构建逐 claim 证据， + 拒绝失效证据，并只从可定位原文句子回答单次问题。 +- 把原单节点 Reader 改为 `prepare_batch -> read_one_paper -> finalize_batch`，记录逐篇游标并通过父图 + checkpointer 恢复；未知异常使用同一 `thread_id` 的 continue 路径。 +- 增加 `POST /runs/{run_id}/continue` 和 `autoresearch continue-run`；`resume` 仍只处理人工中断。 +- 增加双页 PDF 精确定位、解析失败、缓存篡改、摘要降级、失效证据、对抗模型输出、问答负例、 + 批次部分/全部失败、幂等重放和应用重启恢复测试。 +- 补齐标准 ReadingCard 十项结构：规范引用快照、显式论文版本、M05 OA/许可快照、全文 SHA-256、 + 研究问题/作者主张、方法/数据集/基线/变量/指标、关键结果、适用范围/假设/局限/失败案例、复现资源 + 与缺失项、idea 关系、作者创新/Reader 推断创新分栏、待补读引用、置信度和待审核 evidence 集。 +- locator 新增章节、表和图标签；只有原文能识别标签时才填写，并与页码、block、字符偏移一起回查。 ## Pending -- 用户/团队需要为模块负责人模板填写实际 owner、分支、目标日期和 evidence ID。 -- 模块负责人确认后，再选择需要展开的模块，不默认展开全部模块。 -- 真实论文试点仍等待 idea、领域、目标 venue、授权全文和实验资源边界。 +- 项目负责人需确认非 `pending` 的阅读卡审核由哪个角色写入、允许哪些状态迁移；当前 Reader 只建立 + `pending` 状态和待审核 evidence ID，不冒充人工或 Reviewer 已审核。 +- 项目负责人需确认是否在 M06 验收中要求独立的用户授权记录及“本地解析/发送外部 LLM”分 scope 门禁；当前 + `full_text_path` 只执行调用方声明式合同，没有独立授权审计记录，E2 仅表示全文证据等级而非版权许可。 +- 项目负责人和检索负责人需确认论文版本由检索、显式导入还是人工补录；当前合同支持版本枚举，来源未提供时 + 明确保存 `unknown` 和待补读项，不用 DOI/URL 猜测正式版本。 +- 需为章节级阅读陪读创建独立任务范围，明确 session、章节 cursor、暂停点、恢复请求、公式/图表 + 解析依赖和上下文装配规则。 +- 合并前与检索负责人协调 `contracts.py`、`application.py`、API 文档和 `.ai-team/TASK.md` 的交叉改动， + 先 rebase 再共同确认 M05→M06 合同。 +- 生产多 worker 的抽取占用锁和事务 outbox 仍在 M04/生产存储范围，不能宣称 exactly-once。 ## Next step -先认领 M01–M04、M05–M07、M08、M09–M10、M11–M12、M13–M15/M17、M16 这 7 个模块包；完成模块边界确认后，再只对当前瓶颈模块做子任务拆解。 +先完成全量回归和代码评审，再由 kyle 把“版本来源、阅读卡审核状态迁移、摘要降级策略”三个剩余产品 +口径提交项目评审。章节级陪读另开功能任务，不删除也不复用当前问答接口冒充完成。 ## Verification -- [x] Project-to-Act `--check` — managed 配置有效，唯一账本保持不变。 -- [x] 模块文档结构检查 — M00–M17 全部存在，详细版文件可访问。 -- [x] `node .ai-team/check.mjs --base main` — AR-003 acceptance 5/5，任务范围和变更同步有效。 -- [x] Project-to-Act `--validate` — managed 配置 valid，issues 为空。 +- [x] `.venv\Scripts\python.exe -m ruff check src tests`：通过。 +- [x] `.venv\Scripts\python.exe -m pytest -W error -q`：40 项测试通过。 +- [x] `.venv\Scripts\autoresearch.exe doctor`：通过；恰好注册五个业务 Agent。 +- [x] `node .ai-team/check.mjs --base origin/main`：通过（使用工作区内置 Node）。 +- [x] `git diff --check`：通过；仅出现 Git 的 LF/CRLF checkout 提示。 ## Handoff note -- From: `zzg` -- To: `user/team` -- Summary: 当前分配入口已经降为模块级；详细拆解保留在附录。先分模块负责人，后续只对真正需要的模块继续细化。 +- From: `kyle` +- To: `项目负责人/检索负责人/团队评审` +- Summary: 当前交付是可追溯读取底座、标准阅读卡十项、单次证据问答和论文级恢复；章节级阅读陪跑 + 仍未实现且保持独立。M05 OA/license 只读接入，不下载全文；版本来源和非 pending 审核流程待项目负责人确认。 diff --git a/README.md b/README.md index 8c38e3e..46b37ec 100644 --- a/README.md +++ b/README.md @@ -31,6 +31,7 @@ AutoResearch 是一个以证据为边界、以项目文件夹为账本的 LangGr - docs/ARCHITECTURE.md：总体架构与五 Agent。 - docs/FUNCTION_MATRIX.md：基础版功能表与生产边界。 - docs/WORKFLOW_AND_STATE.md：状态机、handoff、checkpoint/resume。 +- docs/M06_PAPER_READER.md：可定位阅读证据、claim map、单次证据问答与逐篇恢复。 - docs/EVIDENCE_AND_GATES.md：证据等级与 L0–L4 门禁。 - docs/KNOWLEDGE_AND_EVOLUTION.md：Wiki+Graph、画像、经验与自进化。 - docs/API.md、docs/OPERATIONS.md、docs/DEVELOPMENT.md：接口、运行和开发。 diff --git a/docs/API.md b/docs/API.md index 31ac246..526b80f 100644 --- a/docs/API.md +++ b/docs/API.md @@ -29,6 +29,7 @@ project_id 只接受小写字母、数字、下划线和连字符。已有目录 | GET | /runs/{run_id} | 查看运行记录 | | GET | /runs/{run_id}?include_checkpoint=true | 同时查看 checkpoint/next/interrupt | | POST | /runs/{run_id}/resume | 恢复 L4 人工 interrupt | +| POST | /runs/{run_id}/continue | 从普通节点异常的 checkpoint 重试 | POST /runs 可包含 seed_papers。没有真实论文时返回 201，但 run.status=blocked，这是有效的 fail-closed 业务结果，不是 HTTP 故障。 @@ -42,6 +43,15 @@ POST /runs/{id}/resume： 非 interrupted run 恢复返回 409。 +`resume` 与 `continue` 不是同一个动作：`resume` 必须携带人工审批数据，只恢复 +`interrupt()`；`continue` 不接收审批数据，只重试保留在同一 thread_id 中的失败节点。 +工作流节点抛出未知异常时，`POST /runs` 返回 503，响应 detail 中包含 `run_id` 和 +`resumable`。仅当该 run 为 `failed` 且仍有待执行 checkpoint 时，`continue` 才接受请求。 +人工 resume 已提交但节点出现瞬时系统错误时，run 保持 `interrupted`、保留原 interrupt， +并返回 `retry_mode=resume`；客户端应使用同一审批内容再次调用 `/resume`，不能改走 +`/continue`。重试请求必须与首次 approval、reviewer、note 完全一致；不同 payload 返回 +409，避免 LangGraph 重放已消费审批时忽略后来改变的决定。 + ## 证据与执行 | 方法 | 路径 | 作用 | @@ -58,11 +68,18 @@ POST /runs/{id}/resume： | 方法 | 路径 | 作用 | |---|---|---| -| POST | /papers/{paper_id}/questions | 证据限定的陪读问答 | +| POST | /papers/{paper_id}/questions | 无会话的单次论文证据问答（非章节级陪读） | | POST | /manuscripts/{manuscript_id}/revisions | 生成修改或润色版本 | 阅读材料不足时返回 unresolved=true，不生成猜测答案。修订不会覆盖原稿，继承 evidence IDs 和 unresolved gaps，并强制 release_ready=false。 +单次问答答案中的 `citations` 包含 document checksum、block ID、页码（PDF）、字符偏移和 +精确引文。相关证据失效后，同一问题不会继续返回旧的 resolved 答案。 + +该接口没有章节 cursor、暂停点、恢复请求或阅读会话；它不能作为计划书中“章节级阅读陪跑” +的验收证据。章节级陪读应作为独立功能块接入 LangGraph，并复用现有 document/block/locator， +而不是把本接口改名后冒充完成。 + ## Wiki、Graph 与画像 | 方法 | 路径 | 作用 | diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 4842961..382358b 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -35,7 +35,7 @@ LLM 是可选认知增强器，不是授权器。LLM 不可覆盖 GateDecision |---|---|---|---| | 总控 Agent | idea、项目约束、证据引用 | research charter、检索式、工作包、结构化派单 | 自行确认创新或实验结果 | | 论文搜索 Agent | 检索式、seed paper | 去重 PaperRecord、题录证据、检索诊断 | 无结果时编造论文 | -| 论文读取 Agent | PaperRecord、全文或摘要 | ReadingCard、陪读回答、InnovationCandidate | 把摘要推断写成全文事实；把候选写成已证实创新 | +| 论文读取 Agent | PaperRecord、M05 OA/许可记录、全文或摘要 | 标准 ReadingCard、单次证据回答、InnovationCandidate | 把摘要推断写成全文事实；把候选写成已证实创新 | | 写作 Agent | 放行的阅读卡、证据、实验记录 | 论文草稿、修改/润色版本、缺口清单 | 新增无证据数字；把草稿标记为可发布 | | 审核 Agent | 候选、稿件、Evidence IDs | ReviewReport、Gate 建议、定向打回 | 修改证据后自批；越过人工外发 Gate | @@ -56,6 +56,12 @@ LLM 是可选认知增强器，不是授权器。LLM 不可覆盖 GateDecision 每个业务节点调用对应 Agent 的 LangGraph 子图。SQLite checkpointer 以 run_id 作为 thread_id。L4 节点调用 interrupt；恢复必须用同一个 run_id 和 Command(resume=...)。 +Paper Reader 是其中一个业务 Agent，不是解析器的别名。其子图逐篇调度阅读；PDF/TXT +解析、checksum、block/locator 校验、EvidenceItem 持久化和 reader_cursor 更新均为确定性 +组件。配置 LLM 时，模型只提出绑定到既有 block 和逐字 quote 的结构化候选；代码验证 +block、quote 和 offset 后才允许形成 supported claim。普通节点异常使用同一 thread_id 的 +continue，人工 release interrupt 仍使用 Command(resume=...)。 + 父状态只保留 typed state、稳定 ID、短诊断和不超过 2000 字符的 bounded_context。论文全文、完整聊天和大产物不进入 handoff。 ## 5. 数据与持久化 @@ -67,6 +73,14 @@ LLM 是可选认知增强器，不是授权器。LLM 不可覆盖 GateDecision RecordStore 使用 kind + record_id 作为主键，project_id 和 partition_name 作为检索边界。关键操作同时写 append-only audit_events。 +M06 额外保存 ParsedDocument、DocumentBlock、ReaderDiagnostic 和 ReadingCard。ReadingCard 包含 +规范引用/版本/访问快照/全文 hash、研究问题与作者主张、方法和实验要素、关键结果、边界与失败、 +复现资源、idea 关系、创新分栏、待补读项及审核状态。M05 的 `fulltext_url` 只作为访问元数据， +不会被 Reader 当成本地正文或触发下载。 +ReadingClaim 与 ClaimEvidenceBinding 当前嵌在 ReadingCard JSON 中，并不是可独立检索的规范记录。 +ParsedDocument/DocumentBlock 使用内容寻址 ID；ReadingCard 同时保留旧字段和嵌套的结构化 +claim map，以兼容现有 Writer/Reviewer 消费者。 + 基础版 SQLite 适合本地开发和单进程验证。生产部署目标仍是 PostgreSQL/pgvector 规范源、对象存储大产物和可重建的 Neo4j 投影；当前代码没有把 SQLite 冒充该生产拓扑。 ## 6. 运行边界 diff --git a/docs/FUNCTION_MATRIX.md b/docs/FUNCTION_MATRIX.md index 157e953..d1741c7 100644 --- a/docs/FUNCTION_MATRIX.md +++ b/docs/FUNCTION_MATRIX.md @@ -10,8 +10,9 @@ | idea 深化 | 已实现 | OrchestratorAgent.scope | 可证伪问题、非目标、工作包 | 尚无领域专用模板库 | | 相关论文检索 | 已实现 | PaperSearchService | PaperRecord、题录证据、诊断 | 三个连接器已编码；默认关闭网络，live 质量未作为本地测试结论 | | 论文去重 | 已实现 | PaperSearchService._dedupe_key | DOI/标题归一结果 | 尚未实现预印本到期刊版本图 | -| 论文总结 | 已实现 | PaperReaderService.read | ReadingCard、locator、置信度 | 结构化抽取为基础算法；图表/公式解析待增强 | -| 阅读陪跑 | 已实现 | POST /papers/{id}/questions、ask-paper | ReadingAnswer、locator、evidence IDs | 只回答已供应文本；无材料时 unresolved=true | +| 标准论文阅读卡 | 已实现 | PaperReaderService.read | 十项 ReadingCard、claim/evidence/locator、置信度 | 识别正文中的章节/表图标签；复杂表格、公式语义解析待开发 | +| 单次论文证据问答 | 已实现 | POST /papers/{id}/questions、ask-paper | ReadingAnswer、locator、evidence IDs | 无会话、无章节 cursor；只回答已供应文本，不等同章节级陪读 | +| 章节级阅读陪跑 | 未实现 | 独立功能块待定 | 章节 checkpoint、当前段落上下文、暂停/恢复 | 需项目负责人确认任务、会话合同及图表/公式依赖 | | 创新点挖掘 | 已实现 | PaperReaderService.mine_innovations | InnovationCandidate、反证测试 | 永远先标记 hypothesis，不自动宣称 novelty | | 工作量辅助 | 已实现 | WorkPackage、ExecutionService | 四类工作包、依赖、验收、状态 | 完成状态必须绑定有效证据；资源估算器待增强 | | 论文写作 | 已实现 | WriterAgent、WritingService.draft | MANUSCRIPT_DRAFT.md | 缺实验时 Results 固定为待真实实验 | diff --git a/docs/M06_PAPER_READER.md b/docs/M06_PAPER_READER.md new file mode 100644 index 0000000..127d53e --- /dev/null +++ b/docs/M06_PAPER_READER.md @@ -0,0 +1,100 @@ +# M06 论文读取 Agent + +## 交付范围 + +M06 已接入主 Research LangGraph 的 `paper_reader` 节点，保持业务 Agent 总数为五个。 +首个切片支持调用方显式提供并声明有权处理的本地 born-digital PDF、UTF-8 TXT 和 Markdown；扫描件只返回 +`ocr_required`，不在本任务中自动 OCR。摘要降级会明确标为 `source_scope=abstract` 和 E1， +不会冒充 E2 全文证据。 + +读取链路为： + + PaperRecord + -> ParsedDocument + content-addressed DocumentBlock + -> validated SourceLocator + EvidenceItem + -> ReadingClaim / ClaimEvidenceBinding + -> backward-compatible ReadingCard + -> Reviewer handoff + +PDF locator 包含文档 checksum、block ID、block ordinal、页码、可识别的章节/表/图标签和 quote 的 +字符偏移。TXT/MD +没有伪造页码，使用稳定 block locator。缓存命中时会重新校验 block 文本 checksum 和归属； +缓存缺 block 或 checksum 不一致时，缓存会被判为无效，并从仍可读取的原始文件重建。只有重建 +和 locator 再校验成功后才生成证据；原始文件也不可用时才返回结构化失败。 + +## 标准 ReadingCard + +ReadingCard 保留旧字段供 Writer/Reviewer 兼容使用，同时增加十项有类型字段： + +1. 从 PaperRecord 规范化的引用快照、显式版本、M05 OA/license 快照和全文 SHA-256； +2. 研究问题与作者主张； +3. 方法、数据集、基线、变量和评价指标； +4. 关键结果及页码、章节、表/图、block 和字符偏移； +5. 适用范围、假设、局限和失败案例； +6. 可复现资源与未定位到的资源； +7. 与当前 idea 的支持、冲突、正交、不可比或 unknown 关系；模型分类初始只标为 hypothesis； +8. 作者声称的创新与 Reader 推断的 hypothesis 分栏； +9. 未解决问题、待补读引用及逐项置信度； +10. 初始 `pending` 审核状态及待审核 evidence ID。 + +正文事实字段使用 ReadingClaim，不复制成无法回查的自由文本；每个 supported claim 都带逐字 quote、 +evidence ID 和 SourceLocator。缺失项使用带 `kind` 的 ReadingGap 显式记录，表示“本次未定位到”， +不推断论文一定没有；`standard_v1` 要求每个核心栏目满足“有值或有对应 typed gap”。M05 的 +`paper_open_access` 只读装配为访问快照；合法入口 URL 不会触发下载，也不会冒充本地全文。 + +## 模型与确定性边界 + +离线模式使用保守的确定性抽取。配置 LLM 时，最多抽取 24 个有界 block 作为认知输入， +模型只能返回已给出的 block ID、逐字 quote，以及 quote 内的原文 claim；任何不存在的 +block、quote 或释义型 claim 都被拒绝并记录 diagnostic。模型不能生成 locator、checksum、 +EvidenceGrade、ClaimStatus 或状态迁移。 + +提取结果按 provider、model、reader 版本和 prompt 版本缓存。重放不会再次调用模型，也 +不会重复生成 ReadingCard、EvidenceItem、handoff 或完成事件。已失效 evidence 不能再次 +形成 supported claim；单次证据问答遇到失效引用会返回 `unresolved=true`。 + +## Checkpoint 与批处理 + +Reader 子图包含三个节点： + +1. `prepare_batch`：验证来自 Paper Search 的 handoff，定位首篇未完成论文。 +2. `read_one_paper`：每次只处理一篇，成功或已知解析失败后推进 cursor。 +3. `finalize_batch`：有成功 ReadingCard 时生成现有的 innovation hypothesis 并交给 Reviewer； + 全部失败时进入 `WAITING_EVIDENCE`。 + +缺失、坏文件、空文档、加密、OCR-required 和不支持格式属于结构化业务失败，会继续下一篇。 +数据库、模型客户端或未知程序异常会向上抛出并保留当前论文 checkpoint。应用返回可恢复的 +run_id；`POST /runs/{id}/continue` 或 `autoresearch continue-run RUN_ID` 使用同一 thread_id +重试。`/resume` 仍只用于人工 release interrupt。 + +## 本地验证 + +在仓库根目录运行： + + .venv\Scripts\python.exe -m ruff check src tests + .venv\Scripts\python.exe -m pytest -W error -q + .venv\Scripts\autoresearch.exe doctor + node .ai-team/check.mjs --base origin/main + git diff --check + +M06 测试运行时生成 PDF/TXT，不包含真实论文或二进制 fixture。覆盖标准阅读卡十项、M05 OA/license +只读装配、全文 hash、idea 关系证据、双页 PDF 的 page-2/section/table/figure +locator、稳定 ID、缓存篡改修复、解析失败矩阵、claim-evidence 链、resolved/unresolved +单次证据问答、失效证据、对抗性模型输出、部分批次失败、全部失败，以及应用重启后的逐篇恢复。 + +## 合并前需项目负责人确认的产品口径 + +- 部分成功策略：当前只要至少一篇生成 ReadingCard 就继续 Reviewer，同时保留失败清单； + 若要求“任一失败即整批阻断”，需调整 finalize policy。 +- 摘要降级：当前为兼容已有流程允许 E1 ReadingCard；若 M06 验收只接受授权全文，应改为 + 无全文即 `WAITING_EVIDENCE`。 +- 模型数据边界：配置 LLM 后会把有界论文原文块发送给该 OpenAI-compatible provider；上线前 + 必须确认供应商、论文授权和数据合规，或增加 Reader 专用禁用开关。 +- 授权审计：当前 `full_text_path` 遵循调用方“有权处理”的声明式输入合同，但没有独立授权记录，也 + 没有区分本地解析与发送外部 LLM 的授权 scope。若项目负责人把它列入 M06 验收，需要先确定记录字段、 + 撤销语义和默认阻断策略；E2 只表示全文证据等级，不等于版权许可。 +- 章节级陪读拆分：按项目负责人确认的口径，它是独立功能块。当前保留单次问答兼容接口，不把它接入或冒充 + 陪读状态机；未来独立模块复用 document/block/locator。任务编号、负责人、session 合同、 + 公式/图表解析依赖和验收样例仍需项目负责人确认。 +- 多 worker：当前 SQLite/profile 面向本地单进程；并发 extraction reservation 和事务 outbox + 属于生产存储/M04-05 范围，不能把本实现宣称为多 worker exactly-once。 diff --git a/docs/WORKFLOW_AND_STATE.md b/docs/WORKFLOW_AND_STATE.md index 9811049..29b7c0b 100644 --- a/docs/WORKFLOW_AND_STATE.md +++ b/docs/WORKFLOW_AND_STATE.md @@ -23,6 +23,8 @@ - FAILED：系统错误，不能当作证据不足。 - RunStatus.BLOCKED：可通过补证据并启动新 run 重新评估。 - RunStatus.INTERRUPTED：必须用同一 thread_id 执行 resume。 +- RunStatus.FAILED 且 resumable=true：系统异常停在 checkpoint，必须用同一 thread_id + 执行 continue；不得伪造人工 `Command(resume=...)`。 ## 2. 关键 Gate @@ -49,9 +51,15 @@ ## 4. Checkpoint 与恢复 - 初次运行用 run_id 作为 LangGraph thread_id。 -- 每个 super-step 由 SqliteSaver 持久化。 +- 每个 super-step 由 SqliteSaver 以 `durability=sync` 持久化；所有业务子图继承父图 + checkpointer，避免子图禁用持久化时产生伪 checkpoint。 - release_gate 调用 interrupt，payload 必须可 JSON 序列化。 - resume 接收 approval、reviewer、note。 +- Paper Reader 子图按 `prepare_batch -> read_one_paper -> finalize_batch` 执行。每成功或 + 业务失败一篇才推进 reader_cursor；未知异常向上抛出，使当前论文留在 checkpoint 中。 +- continue 不携带业务输入，使用 `graph.invoke(None, same_thread_config)` 重试失败节点。 +- Reader checkpoint 只保存 document/card/evidence/diagnostic IDs、游标和短诊断；全文块 + 只存在规范记录库，不进入 ResearchState。 - interrupt 前不执行非幂等写操作；人工证据 ID 对同一 run 保持稳定。 - 普通证据阻断运行保持不可变审计历史；补证据后创建新 run，避免静默重写旧结论。 diff --git a/src/autoresearch/agents/base.py b/src/autoresearch/agents/base.py index d3072e9..98b2242 100644 --- a/src/autoresearch/agents/base.py +++ b/src/autoresearch/agents/base.py @@ -17,7 +17,13 @@ class WorkflowState(TypedDict, total=False): search_queries: list[str] seed_papers: list[dict[str, Any]] paper_ids: list[str] + parsed_document_ids: list[str] reading_card_ids: list[str] + reader_cursor: int + reader_current_paper_id: str | None + reader_completed_paper_ids: list[str] + reader_failed_paper_ids: list[str] + reader_diagnostic_ids: list[str] innovation_ids: list[str] work_package_ids: list[str] manuscript_id: str | None @@ -36,4 +42,4 @@ def single_node_subgraph(name: str, function: Callable[[WorkflowState], dict]) - builder.add_node(name, function) builder.add_edge(START, name) builder.add_edge(name, END) - return builder.compile(checkpointer=False) + return builder.compile() diff --git a/src/autoresearch/agents/paper_reader.py b/src/autoresearch/agents/paper_reader.py index 5a68213..9743ece 100644 --- a/src/autoresearch/agents/paper_reader.py +++ b/src/autoresearch/agents/paper_reader.py @@ -1,14 +1,21 @@ from __future__ import annotations -from autoresearch.agents.base import WorkflowState, single_node_subgraph +from langgraph.graph import END, START, StateGraph + +from autoresearch.agents.base import WorkflowState from autoresearch.contracts import ( AgentId, ArtifactRef, HandoffEnvelope, + KnowledgePartition, LifecycleState, PaperRecord, + ReaderDiagnostic, + ReaderDiagnosticCode, + ReadingCard, ResearchState, RunStatus, + stable_id, utc_now, ) from autoresearch.handoffs import HandoffService @@ -32,7 +39,18 @@ def __init__( self.handoffs = handoffs self.state_machine = state_machine - def run(self, raw_state: WorkflowState) -> dict: + @staticmethod + def _next_cursor(state: ResearchState) -> int: + finished = { + *state.reader_completed_paper_ids, + *state.reader_failed_paper_ids, + } + for index, paper_id in enumerate(state.paper_ids): + if paper_id not in finished: + return index + return len(state.paper_ids) + + def prepare_batch(self, raw_state: WorkflowState) -> dict: state = ResearchState.model_validate(raw_state) self.handoffs.accept( state.handoff or {}, @@ -40,12 +58,140 @@ def run(self, raw_state: WorkflowState) -> dict: project_id=state.project_id, run_id=state.run_id, ) - papers = [] - for paper_id in state.paper_ids: - raw = self.store.get("paper", paper_id) - if raw is not None: - papers.append(PaperRecord.model_validate(raw)) - if not papers: + cursor = self._next_cursor(state) + current = state.paper_ids[cursor] if cursor < len(state.paper_ids) else None + return state.model_copy( + update={ + "reader_cursor": cursor, + "reader_current_paper_id": current, + "last_agent": self.agent_id, + "updated_at": utc_now(), + } + ).model_dump(mode="json") + + def _missing_paper_diagnostic(self, state: ResearchState, paper_id: str) -> ReaderDiagnostic: + diagnostic = ReaderDiagnostic( + diagnostic_id=stable_id( + "reader_diag", + state.project_id, + paper_id, + ReaderDiagnosticCode.PAPER_RECORD_MISSING.value, + "record_load", + ), + project_id=state.project_id, + paper_id=paper_id, + code=ReaderDiagnosticCode.PAPER_RECORD_MISSING, + stage="record_load", + message="The registered paper record disappeared before reading.", + resolution="Restore the paper record or rerun paper search.", + ) + raw_existing = self.store.get("reader_diagnostic", diagnostic.diagnostic_id) + if raw_existing is None: + self.store.put( + "reader_diagnostic", + diagnostic.diagnostic_id, + diagnostic, + project_id=state.project_id, + partition=KnowledgePartition.PAPERS.value, + ) + return diagnostic + existing = ReaderDiagnostic.model_validate(raw_existing) + expected = diagnostic.model_dump(mode="json", exclude={"created_at"}) + actual = existing.model_dump(mode="json", exclude={"created_at"}) + if actual != expected: + raise ValueError(f"stable reader diagnostic collision: {diagnostic.diagnostic_id}") + return existing + + def _diagnostic_messages(self, diagnostic_ids: list[str]) -> list[str]: + messages: list[str] = [] + for diagnostic_id in diagnostic_ids: + raw = self.store.get("reader_diagnostic", diagnostic_id) + if raw is None: + continue + item = ReaderDiagnostic.model_validate(raw) + messages.append(f"reader[{item.paper_id}] {item.code.value}: {item.message}") + return messages + + def read_one_paper(self, raw_state: WorkflowState) -> dict: + state = ResearchState.model_validate(raw_state) + if state.reader_cursor >= len(state.paper_ids): + return state.model_copy( + update={"reader_current_paper_id": None, "updated_at": utc_now()} + ).model_dump(mode="json") + + paper_id = state.paper_ids[state.reader_cursor] + raw_paper = self.store.get("paper", paper_id) + if raw_paper is None: + missing = self._missing_paper_diagnostic(state, paper_id) + document_ids = state.parsed_document_ids + card_ids = state.reading_card_ids + evidence_ids = state.evidence_ids + completed_ids = state.reader_completed_paper_ids + failed_ids = list(dict.fromkeys([*state.reader_failed_paper_ids, paper_id])) + diagnostic_ids = list( + dict.fromkeys([*state.reader_diagnostic_ids, missing.diagnostic_id]) + ) + new_diagnostic_ids = [missing.diagnostic_id] + else: + paper = PaperRecord.model_validate(raw_paper) + outcome = self.reader.read(paper, idea=state.idea) + document_ids = list( + dict.fromkeys([*state.parsed_document_ids, outcome.document_id]) + ) + card_ids = list( + dict.fromkeys( + [*state.reading_card_ids, *([outcome.card_id] if outcome.card_id else [])] + ) + ) + evidence_ids = list(dict.fromkeys([*state.evidence_ids, *outcome.evidence_ids])) + diagnostic_ids = list( + dict.fromkeys([*state.reader_diagnostic_ids, *outcome.diagnostic_ids]) + ) + new_diagnostic_ids = outcome.diagnostic_ids + if outcome.succeeded: + completed_ids = list( + dict.fromkeys([*state.reader_completed_paper_ids, paper_id]) + ) + failed_ids = state.reader_failed_paper_ids + else: + completed_ids = state.reader_completed_paper_ids + failed_ids = list(dict.fromkeys([*state.reader_failed_paper_ids, paper_id])) + + next_cursor = state.reader_cursor + 1 + next_paper = ( + state.paper_ids[next_cursor] if next_cursor < len(state.paper_ids) else None + ) + new_diagnostics = self._diagnostic_messages(new_diagnostic_ids) + diagnostics = list(dict.fromkeys([*state.diagnostics, *new_diagnostics]))[-100:] + return state.model_copy( + update={ + "parsed_document_ids": document_ids, + "reading_card_ids": card_ids, + "reader_cursor": next_cursor, + "reader_current_paper_id": next_paper, + "reader_completed_paper_ids": completed_ids, + "reader_failed_paper_ids": failed_ids, + "reader_diagnostic_ids": diagnostic_ids, + "evidence_ids": evidence_ids, + "diagnostics": diagnostics, + "last_agent": self.agent_id, + "updated_at": utc_now(), + } + ).model_dump(mode="json") + + @staticmethod + def _route_batch(raw_state: WorkflowState) -> str: + state = ResearchState.model_validate(raw_state) + return "read" if state.reader_cursor < len(state.paper_ids) else "finalize" + + def finalize_batch(self, raw_state: WorkflowState) -> dict: + state = ResearchState.model_validate(raw_state) + cards = [ + ReadingCard.model_validate(raw) + for card_id in state.reading_card_ids + if (raw := self.store.get("reading_card", card_id)) is not None + ] + if not cards: target = self.state_machine.transition( state.lifecycle_state, LifecycleState.WAITING_EVIDENCE ) @@ -53,69 +199,98 @@ def run(self, raw_state: WorkflowState) -> dict: update={ "lifecycle_state": target, "run_status": RunStatus.BLOCKED, - "blockers": [*state.blockers, "Paper records disappeared before reading."], + "blockers": [ + *state.blockers, + "No paper produced an evidence-backed reading card.", + ], "handoff": None, + "reader_current_paper_id": None, "last_agent": self.agent_id, "updated_at": utc_now(), } ).model_dump(mode="json") - cards = [] - reading_evidence_ids = [] - for paper in papers: - card, evidence = self.reader.read(paper) - cards.append(card) - reading_evidence_ids.append(evidence.evidence_id) innovations = self.reader.mine_innovations(state.project_id, state.idea, cards) target = self.state_machine.transition( state.lifecycle_state, LifecycleState.LITERATURE_READ ) - combined_evidence = list(dict.fromkeys([*state.evidence_ids, *reading_evidence_ids])) + artifacts = [ + ArtifactRef( + artifact_id=card.card_id, + kind="reading_card", + summary=card.findings[0][:300], + ) + for card in cards + ] + [ + ArtifactRef( + artifact_id=item.innovation_id, + kind="innovation_candidate", + summary=item.statement[:300], + ) + for item in innovations + ] + reading_evidence_ids = list( + dict.fromkeys(evidence_id for card in cards for evidence_id in card.evidence_ids) + ) + handoff_evidence_ids = list( + dict.fromkeys([*reading_evidence_ids, *state.evidence_ids]) + )[:100] handoff = self.handoffs.issue( HandoffEnvelope( + handoff_id=stable_id( + "handoff", + state.project_id, + state.run_id, + self.agent_id.value, + AgentId.REVIEWER.value, + *state.reading_card_ids, + *[item.innovation_id for item in innovations], + ), project_id=state.project_id, run_id=state.run_id, from_agent=self.agent_id, to_agent=AgentId.REVIEWER, - objective=( - "Audit reading provenance and decide whether innovation planning may proceed." - ), + objective="Audit reading provenance and decide whether planning may proceed.", expected_output="Gate-backed review report with required changes.", - artifact_refs=[ - ArtifactRef( - artifact_id=card.card_id, - kind="reading_card", - summary=card.findings[0][:300], - ) - for card in cards - ] - + [ - ArtifactRef( - artifact_id=item.innovation_id, - kind="innovation_candidate", - summary=item.statement[:300], - ) - for item in innovations - ], - evidence_ids=combined_evidence, + artifact_refs=artifacts[:20], + evidence_ids=handoff_evidence_ids, constraints=[ "A hypothesis is not a novelty finding", + "Every reading claim must retain a validated source locator", "Insufficient evidence must block progression", ], - bounded_context=f"cards={len(cards)}; candidates={len(innovations)}", + bounded_context=( + f"cards={len(cards)}; candidates={len(innovations)}; " + f"reader_failures={len(state.reader_failed_paper_ids)}" + ), ) ) return state.model_copy( update={ "lifecycle_state": target, - "reading_card_ids": [card.card_id for card in cards], "innovation_ids": [item.innovation_id for item in innovations], - "evidence_ids": combined_evidence, "handoff": handoff.model_dump(mode="json"), + "reader_current_paper_id": None, "last_agent": self.agent_id, "updated_at": utc_now(), } ).model_dump(mode="json") def build_graph(self): - return single_node_subgraph("read_and_mine", self.run) + builder = StateGraph(WorkflowState) + builder.add_node("prepare_batch", self.prepare_batch) + builder.add_node("read_one_paper", self.read_one_paper) + builder.add_node("finalize_batch", self.finalize_batch) + builder.add_edge(START, "prepare_batch") + builder.add_conditional_edges( + "prepare_batch", + self._route_batch, + {"read": "read_one_paper", "finalize": "finalize_batch"}, + ) + builder.add_conditional_edges( + "read_one_paper", + self._route_batch, + {"read": "read_one_paper", "finalize": "finalize_batch"}, + ) + builder.add_edge("finalize_batch", END) + return builder.compile() diff --git a/src/autoresearch/api.py b/src/autoresearch/api.py index d908f6c..b9691a2 100644 --- a/src/autoresearch/api.py +++ b/src/autoresearch/api.py @@ -2,7 +2,7 @@ from fastapi import FastAPI, HTTPException, Query -from autoresearch.application import AutoResearchApplication +from autoresearch.application import AutoResearchApplication, RunExecutionError from autoresearch.contracts import ( EvidenceInvalidationRequest, EvidenceItem, @@ -54,6 +54,20 @@ def get_project(project_id: str): def start_run(request: RunRequest): try: return runtime.run(request) + except RunExecutionError as exc: + raise HTTPException( + status_code=( + 503 + if exc.record["resumable"] or exc.record.get("retry_mode") == "resume" + else 500 + ), + detail={ + "message": str(exc), + "run_id": exc.record["run_id"], + "resumable": exc.record["resumable"], + "retry_mode": exc.record.get("retry_mode"), + }, + ) from exc except KeyError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc @@ -70,6 +84,43 @@ def get_run(run_id: str, include_checkpoint: bool = False): def resume_run(run_id: str, request: ResumeRequest): try: return runtime.resume(run_id, request) + except RunExecutionError as exc: + raise HTTPException( + status_code=( + 503 + if exc.record["resumable"] or exc.record.get("retry_mode") == "resume" + else 500 + ), + detail={ + "message": str(exc), + "run_id": exc.record["run_id"], + "resumable": exc.record["resumable"], + "retry_mode": exc.record.get("retry_mode"), + }, + ) from exc + except KeyError as exc: + raise HTTPException(status_code=404, detail="run not found") from exc + except ValueError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + + @app.post("/runs/{run_id}/continue") + def continue_run(run_id: str): + try: + return runtime.continue_run(run_id) + except RunExecutionError as exc: + raise HTTPException( + status_code=( + 503 + if exc.record["resumable"] or exc.record.get("retry_mode") == "resume" + else 500 + ), + detail={ + "message": str(exc), + "run_id": exc.record["run_id"], + "resumable": exc.record["resumable"], + "retry_mode": exc.record.get("retry_mode"), + }, + ) from exc except KeyError as exc: raise HTTPException(status_code=404, detail="run not found") from exc except ValueError as exc: diff --git a/src/autoresearch/application.py b/src/autoresearch/application.py index 1da0cfc..84f9ce2 100644 --- a/src/autoresearch/application.py +++ b/src/autoresearch/application.py @@ -32,6 +32,7 @@ WorkPackage, WorkPackageUpdate, new_id, + stable_id, ) from autoresearch.evidence import EvidenceService from autoresearch.evolution_service import EvolutionService, ExperienceService @@ -50,6 +51,23 @@ from autoresearch.writing_service import WritingService +class RunExecutionError(RuntimeError): + def __init__(self, record: dict[str, Any]): + self.record = record + run_id = record.get("run_id", "unknown") + if record.get("resumable"): + message = f"research run {run_id} stopped at a resumable checkpoint" + elif record.get("retry_mode") == "resume": + message = f"research run {run_id} failed while resuming; retry the human resume" + else: + message = f"research run {run_id} failed without a resumable checkpoint" + super().__init__(message) + + +class ResumableRunError(RunExecutionError): + pass + + class AutoResearchApplication: def __init__(self, settings: Settings | None = None): self.settings = settings or get_settings() @@ -72,7 +90,12 @@ def __init__(self, settings: Settings | None = None): self.knowledge, network_enabled=self.settings.network_enabled, ) - self.reader = PaperReaderService(self.store, self.evidence, self.knowledge) + self.reader = PaperReaderService( + self.store, + self.evidence, + self.knowledge, + llm=self.llm, + ) self.writing = WritingService( self.store, self.evidence, @@ -165,6 +188,10 @@ def _save_run( "status": status, "state": state, "interrupts": interrupts or [], + "resumable": False, + "retry_mode": None, + "resume_request_fingerprint": None, + "error": None, } self.store.put( "run", @@ -186,6 +213,175 @@ def _save_run( self.projects.sync_run_projection(record) return record + @staticmethod + def _snapshot_reader_progress(snapshot: Any) -> tuple[dict[str, Any], list[str]]: + reader_keys = { + "parsed_document_ids", + "reading_card_ids", + "reader_cursor", + "reader_current_paper_id", + "reader_completed_paper_ids", + "reader_failed_paper_ids", + "reader_diagnostic_ids", + "evidence_ids", + "diagnostics", + } + progress: dict[str, Any] = {} + errors: list[str] = [] + pending = [snapshot] + seen: set[int] = set() + while pending: + item = pending.pop() + if id(item) in seen: + continue + seen.add(id(item)) + values = getattr(item, "values", None) + if isinstance(values, dict) and "reader_cursor" in values: + candidate = {key: values[key] for key in reader_keys if key in values} + if candidate.get("reader_cursor", -1) >= progress.get("reader_cursor", -1): + progress = candidate + for task in getattr(item, "tasks", ()): + error = getattr(task, "error", None) + if error: + errors.append(str(error)[:1000]) + nested = getattr(task, "state", None) + if hasattr(nested, "values"): + pending.append(nested) + return progress, errors + + def _save_failed_run( + self, + state: dict[str, Any], + snapshot: Any, + exc: Exception, + ) -> dict[str, Any]: + progress, task_errors = self._snapshot_reader_progress(snapshot) + projected_state = {**state, **progress} + projected_state["run_status"] = RunStatus.FAILED.value + error_message = f"{type(exc).__name__}: {str(exc)}"[:1000] + diagnostics = list( + dict.fromkeys( + [ + *projected_state.get("diagnostics", []), + f"workflow_checkpoint_error: {error_message}", + ] + ) + )[-100:] + projected_state["diagnostics"] = diagnostics + resumable = bool(getattr(snapshot, "next", ())) + record = { + "run_id": projected_state["run_id"], + "project_id": projected_state["project_id"], + "status": RunStatus.FAILED.value, + "state": projected_state, + "interrupts": [], + "resumable": resumable, + "retry_mode": "continue" if resumable else None, + "resume_request_fingerprint": None, + "error": { + "type": type(exc).__name__, + "message": str(exc)[:1000], + "checkpoint_errors": task_errors[-20:], + }, + } + self.store.put( + "run", + projected_state["run_id"], + record, + project_id=projected_state["project_id"], + partition="runs", + ) + self.store.append_event( + "run.failed_at_checkpoint", + { + "run_id": projected_state["run_id"], + "resumable": resumable, + "error_type": type(exc).__name__, + }, + project_id=projected_state["project_id"], + actor="application", + ) + self.projects.sync_run_projection(record) + return record + + def _save_resume_failure( + self, + existing: dict[str, Any], + snapshot: Any, + exc: Exception, + request: ResumeRequest, + ) -> dict[str, Any]: + checkpoint_state = dict(snapshot.values) or existing["state"] + projected_state = dict(checkpoint_state) + projected_state["run_status"] = RunStatus.INTERRUPTED.value + error_message = f"{type(exc).__name__}: {str(exc)}"[:1000] + projected_state["diagnostics"] = list( + dict.fromkeys( + [ + *projected_state.get("diagnostics", []), + f"human_resume_checkpoint_error: {error_message}", + ] + ) + )[-100:] + record = { + "run_id": projected_state["run_id"], + "project_id": projected_state["project_id"], + "status": RunStatus.INTERRUPTED.value, + "state": projected_state, + "interrupts": existing.get("interrupts", []), + "resumable": False, + "retry_mode": "resume", + "resume_request_fingerprint": self._resume_request_fingerprint( + projected_state["run_id"], + request, + ), + "error": { + "type": type(exc).__name__, + "message": str(exc)[:1000], + "checkpoint_errors": [str(getattr(task, "error", ""))[:1000] + for task in snapshot.tasks + if getattr(task, "error", None)], + }, + } + self.store.put( + "run", + projected_state["run_id"], + record, + project_id=projected_state["project_id"], + partition="runs", + ) + self.store.append_event( + "run.human_resume_failed", + { + "run_id": projected_state["run_id"], + "error_type": type(exc).__name__, + }, + project_id=projected_state["project_id"], + actor="application", + ) + self.projects.sync_run_projection(record) + return record + + @staticmethod + def _resume_request_fingerprint(run_id: str, request: ResumeRequest) -> str: + return stable_id( + "resume_request", + run_id, + request.approval, + request.reviewer, + request.note, + ) + + def _snapshot(self, run_id: str): + config = {"configurable": {"thread_id": run_id}} + return self.graph.get_state(config, subgraphs=True) + + @staticmethod + def _execution_error(record: dict[str, Any]) -> RunExecutionError: + if record.get("resumable"): + return ResumableRunError(record) + return RunExecutionError(record) + def create_project(self, request: ProjectCreate, *, owner: str = "user") -> dict[str, Any]: path = self.projects.create(request, owner=owner) return {"project_id": request.project_id, "path": str(path), "status": "active"} @@ -210,8 +406,42 @@ def run(self, request: RunRequest) -> dict[str, Any]: project_id=state.project_id, actor="user", ) + initial_state = state.model_dump(mode="json") + self._save_run(initial_state) + try: + with self._graph_lock: + result = self.graph.invoke(initial_state, config=config, durability="sync") + except Exception as exc: + with self._graph_lock: + snapshot = self._snapshot(state.run_id) + checkpoint_state = dict(snapshot.values) or initial_state + record = self._save_failed_run(checkpoint_state, snapshot, exc) + raise self._execution_error(record) from exc + interrupts = self._interrupts(result) + return self._save_run(result, interrupts=interrupts) + + def continue_run(self, run_id: str) -> dict[str, Any]: + existing = self.get_run(run_id) + if existing is None: + raise KeyError(run_id) + if existing.get("interrupts"): + raise ValueError("run is waiting for a human interrupt; use resume instead") + if existing.get("status") != RunStatus.FAILED.value or not existing.get("resumable"): + raise ValueError("run has no resumable failed checkpoint") + config = {"configurable": {"thread_id": run_id}} with self._graph_lock: - result = self.graph.invoke(state.model_dump(mode="json"), config=config) + snapshot = self._snapshot(run_id) + if not snapshot.next: + raise ValueError("run checkpoint has no pending graph task") + try: + with self._graph_lock: + result = self.graph.invoke(None, config=config, durability="sync") + except Exception as exc: + with self._graph_lock: + failed_snapshot = self._snapshot(run_id) + state = dict(failed_snapshot.values) or existing["state"] + record = self._save_failed_run(state, failed_snapshot, exc) + raise self._execution_error(record) from exc interrupts = self._interrupts(result) return self._save_run(result, interrupts=interrupts) @@ -221,12 +451,24 @@ def resume(self, run_id: str, request: ResumeRequest) -> dict[str, Any]: raise KeyError(run_id) if not existing["interrupts"]: raise ValueError("run is not waiting for a human interrupt") + expected_fingerprint = existing.get("resume_request_fingerprint") + if existing.get("retry_mode") == "resume" and expected_fingerprint: + actual_fingerprint = self._resume_request_fingerprint(run_id, request) + if actual_fingerprint != expected_fingerprint: + raise ValueError("human resume retry must use the identical approval payload") config = {"configurable": {"thread_id": run_id}} - with self._graph_lock: - result = self.graph.invoke( - Command(resume=request.model_dump(mode="json")), - config=config, - ) + try: + with self._graph_lock: + result = self.graph.invoke( + Command(resume=request.model_dump(mode="json")), + config=config, + durability="sync", + ) + except Exception as exc: + with self._graph_lock: + failed_snapshot = self._snapshot(run_id) + record = self._save_resume_failure(existing, failed_snapshot, exc, request) + raise RunExecutionError(record) from exc interrupts = self._interrupts(result) return self._save_run(result, interrupts=interrupts) @@ -234,9 +476,10 @@ def get_run(self, run_id: str) -> dict[str, Any] | None: return self.store.get("run", run_id) def graph_state(self, run_id: str) -> dict[str, Any]: - config = {"configurable": {"thread_id": run_id}} with self._graph_lock: - snapshot = self.graph.get_state(config) + snapshot = self._snapshot(run_id) + reader_progress, errors = self._snapshot_reader_progress(snapshot) + record = self.get_run(run_id) or {} return { "values": dict(snapshot.values), "next": list(snapshot.next), @@ -246,6 +489,10 @@ def graph_state(self, run_id: str) -> dict[str, Any]: for task in snapshot.tasks for interrupt_item in task.interrupts ], + "reader_progress": reader_progress, + "errors": errors, + "resumable": bool(record.get("resumable")), + "retry_mode": record.get("retry_mode"), } def add_evidence(self, item: EvidenceItem, *, actor: str = "user") -> EvidenceItem: diff --git a/src/autoresearch/cli.py b/src/autoresearch/cli.py index 643a45b..7402a01 100644 --- a/src/autoresearch/cli.py +++ b/src/autoresearch/cli.py @@ -7,7 +7,7 @@ import typer import uvicorn -from autoresearch.application import AutoResearchApplication +from autoresearch.application import AutoResearchApplication, RunExecutionError from autoresearch.contracts import ( EvidenceGrade, EvidenceItem, @@ -89,16 +89,20 @@ def run( seeds = [PaperRecord.model_validate({**raw, "project_id": project_id}) for raw in raw_items] runtime = AutoResearchApplication() try: - _echo( - runtime.run( - RunRequest( - project_id=project_id, - idea=idea, - seed_papers=seeds, - request_release=request_release, + try: + _echo( + runtime.run( + RunRequest( + project_id=project_id, + idea=idea, + seed_papers=seeds, + request_release=request_release, + ) ) ) - ) + except RunExecutionError as exc: + _echo({"error": str(exc), **exc.record}) + raise typer.Exit(code=1) from exc finally: runtime.close() @@ -113,12 +117,32 @@ def resume( """Resume an L4 human interrupt using the same durable thread ID.""" runtime = AutoResearchApplication() try: - _echo( - runtime.resume( - run_id, - ResumeRequest(approval=approve, reviewer=reviewer, note=note), + try: + _echo( + runtime.resume( + run_id, + ResumeRequest(approval=approve, reviewer=reviewer, note=note), + ) ) - ) + except RunExecutionError as exc: + _echo({"error": str(exc), **exc.record}) + raise typer.Exit(code=1) from exc + finally: + runtime.close() + + +@app.command("continue-run") +def continue_run( + run_id: Annotated[str, typer.Argument(help="Failed resumable LangGraph run ID")], +) -> None: + """Retry the failed graph node from the same durable checkpoint.""" + runtime = AutoResearchApplication() + try: + try: + _echo(runtime.continue_run(run_id)) + except RunExecutionError as exc: + _echo({"error": str(exc), **exc.record}) + raise typer.Exit(code=1) from exc finally: runtime.close() diff --git a/src/autoresearch/contracts.py b/src/autoresearch/contracts.py index 34158de..4fd21ba 100644 --- a/src/autoresearch/contracts.py +++ b/src/autoresearch/contracts.py @@ -2,7 +2,8 @@ from datetime import UTC, datetime from enum import StrEnum -from typing import Any +from hashlib import sha256 +from typing import Any, Literal from uuid import uuid4 from pydantic import BaseModel, Field, model_validator @@ -12,6 +13,13 @@ def new_id(prefix: str) -> str: return f"{prefix}_{uuid4().hex[:16]}" +def stable_id(prefix: str, *parts: object) -> str: + """Build a content-addressed ID for replay-safe domain artifacts.""" + + payload = "\x1f".join(str(part) for part in parts).encode("utf-8") + return f"{prefix}_{sha256(payload).hexdigest()[:16]}" + + def utc_now() -> datetime: return datetime.now(UTC) @@ -97,6 +105,181 @@ class ClaimStatus(StrEnum): REFUTED = "refuted" +class PaperVersionKind(StrEnum): + UNKNOWN = "unknown" + PREPRINT = "preprint" + ACCEPTED_MANUSCRIPT = "accepted_manuscript" + VERSION_OF_RECORD = "version_of_record" + CORRECTED_VERSION = "corrected_version" + + +class IdeaRelation(StrEnum): + SUPPORTS = "supports" + CONFLICTS = "conflicts" + ORTHOGONAL = "orthogonal" + INCOMPARABLE = "incomparable" + UNKNOWN = "unknown" + + +class ReadingReviewStatus(StrEnum): + PENDING = "pending" + APPROVED = "approved" + NEEDS_REVISION = "needs_revision" + REJECTED = "rejected" + + +class DocumentFormat(StrEnum): + PDF = "pdf" + TEXT = "text" + ABSTRACT = "abstract" + METADATA = "metadata" + + +class DocumentParseStatus(StrEnum): + PARSED = "parsed" + PARTIAL = "partial" + METADATA_ONLY = "metadata_only" + MISSING = "missing" + MALFORMED = "malformed" + EMPTY = "empty" + ENCRYPTED = "encrypted" + OCR_REQUIRED = "ocr_required" + UNSUPPORTED = "unsupported" + + +class ReadingSourceScope(StrEnum): + LEGACY = "legacy" + FULL_TEXT = "full_text" + ABSTRACT = "abstract" + METADATA = "metadata" + + +class ReaderDiagnosticCode(StrEnum): + PAPER_RECORD_MISSING = "paper_record_missing" + FULL_TEXT_MISSING = "full_text_missing" + UNSUPPORTED_MEDIA = "unsupported_media" + DOCUMENT_TOO_LARGE = "document_too_large" + MALFORMED_DOCUMENT = "malformed_document" + EMPTY_TEXT = "empty_text" + ENCRYPTED_PDF = "encrypted_pdf" + OCR_REQUIRED = "ocr_required" + PARTIAL_PARSE = "partial_parse" + TEXT_DECODE_REPLACED = "text_decode_replaced" + INSUFFICIENT_EVIDENCE = "insufficient_evidence" + LOCATOR_VALIDATION_FAILED = "locator_validation_failed" + INVALIDATED_EVIDENCE = "invalidated_evidence" + LLM_EXTRACTION_FAILED = "llm_extraction_failed" + + +class EvidenceRelation(StrEnum): + SUPPORTS = "supports" + REFUTES = "refutes" + MENTIONS = "mentions" + + +class ReaderDiagnostic(BaseModel): + diagnostic_id: str = Field(default_factory=lambda: new_id("reader_diag")) + project_id: str + paper_id: str + code: ReaderDiagnosticCode + stage: str = Field(max_length=100) + message: str = Field(max_length=2000) + document_id: str | None = None + source_uri: str | None = None + source_checksum: str | None = None + retryable: bool = False + resolution: str = Field(default="", max_length=1000) + affected_pages: list[int] = Field(default_factory=list) + created_at: datetime = Field(default_factory=utc_now) + + +class SourceLocator(BaseModel): + document_id: str + document_checksum: str + block_id: str + block_ordinal: int = Field(ge=0) + page_number: int | None = Field(default=None, ge=1) + section: str | None = Field(default=None, max_length=500) + figure_label: str | None = Field(default=None, max_length=100) + table_label: str | None = Field(default=None, max_length=100) + start_offset: int | None = Field(default=None, ge=0) + end_offset: int | None = Field(default=None, ge=0) + + @model_validator(mode="after") + def offsets_are_ordered(self) -> SourceLocator: + if (self.start_offset is None) != (self.end_offset is None): + raise ValueError("locator offsets must either both be present or both be absent") + if ( + self.start_offset is not None + and self.end_offset is not None + and self.end_offset < self.start_offset + ): + raise ValueError("locator end_offset precedes start_offset") + return self + + +class DocumentBlock(BaseModel): + block_id: str + project_id: str + paper_id: str + document_id: str + ordinal: int = Field(ge=0) + text: str = Field(min_length=1) + text_checksum: str + page_number: int | None = Field(default=None, ge=1) + section: str | None = Field(default=None, max_length=500) + + +class ParsedDocument(BaseModel): + document_id: str + project_id: str + paper_id: str + format: DocumentFormat + checksum: str + status: DocumentParseStatus + source_scope: ReadingSourceScope + source_uri: str | None = None + parser_name: str + parser_version: str + block_ids: list[str] = Field(default_factory=list) + page_count: int | None = Field(default=None, ge=0) + character_count: int = Field(default=0, ge=0) + diagnostic_ids: list[str] = Field(default_factory=list) + created_at: datetime = Field(default_factory=utc_now) + + +class ClaimEvidenceBinding(BaseModel): + evidence_id: str + relation: EvidenceRelation = EvidenceRelation.SUPPORTS + locator: SourceLocator + quote: str = Field(min_length=1, max_length=4000) + quote_checksum: str + comparison_fingerprint: str | None = None + + +class ReadingClaim(BaseModel): + claim_id: str + kind: str = Field(min_length=1, max_length=100) + text: str = Field(min_length=1, max_length=2000) + status: ClaimStatus = ClaimStatus.UNKNOWN + bindings: list[ClaimEvidenceBinding] = Field(default_factory=list) + confidence: float = Field(ge=0, le=1) + + @model_validator(mode="after") + def decided_claims_require_evidence(self) -> ReadingClaim: + if self.status in {ClaimStatus.SUPPORTED, ClaimStatus.REFUTED} and not self.bindings: + raise ValueError("supported or refuted reading claims require evidence bindings") + if self.status == ClaimStatus.SUPPORTED and any( + binding.relation != EvidenceRelation.SUPPORTS for binding in self.bindings + ): + raise ValueError("supported reading claims require supporting evidence bindings") + if self.status == ClaimStatus.REFUTED and any( + binding.relation != EvidenceRelation.REFUTES for binding in self.bindings + ): + raise ValueError("refuted reading claims require refuting evidence bindings") + return self + + class ArtifactRef(BaseModel): artifact_id: str kind: str @@ -119,6 +302,7 @@ class EvidenceItem(BaseModel): independent_source: str = Field(min_length=1, max_length=300) valid: bool = True supersedes: str | None = None + source_locators: list[SourceLocator] = Field(default_factory=list) created_at: datetime = Field(default_factory=utc_now) metadata: dict[str, Any] = Field(default_factory=dict) @@ -199,10 +383,136 @@ class PaperRecord(BaseModel): url: str | None = None source: str source_record_id: str | None = None + version: PaperVersionKind = PaperVersionKind.UNKNOWN + version_label: str | None = Field(default=None, max_length=200) full_text_path: str | None = None retrieved_at: datetime = Field(default_factory=utc_now) +class CanonicalCitation(BaseModel): + source_paper_id: str = Field(min_length=1, max_length=200) + text: str = Field(min_length=1, max_length=2000) + title: str = Field(min_length=1, max_length=1000) + authors: list[str] = Field(default_factory=list) + year: int | None = None + doi: str | None = None + url: str | None = None + + +class ReadingAccessSnapshot(BaseModel): + """Read-only projection of M05's paper_open_access record.""" + + record_available: bool = False + record_kind: Literal["paper_open_access"] = "paper_open_access" + record_id: str | None = Field(default=None, max_length=200) + status: Literal[ + "open_access", + "no_open_fulltext_found", + "unknown", + ] = "unknown" + fulltext_url: str | None = None + license: str | None = None + source: str | None = None + checked_sources: list[str] = Field(default_factory=list) + diagnostics: list[str] = Field(default_factory=list) + + @model_validator(mode="after") + def open_access_requires_a_location(self) -> ReadingAccessSnapshot: + if self.status == "open_access" and not self.fulltext_url: + raise ValueError("open_access records require fulltext_url") + if self.status != "open_access" and (self.fulltext_url or self.license): + raise ValueError("non-open access records cannot claim a location or license") + if ( + self.record_available + and self.status == "no_open_fulltext_found" + and not self.checked_sources + and not self.diagnostics + ): + raise ValueError( + "no_open_fulltext_found requires checked_sources or diagnostics" + ) + if not self.record_available and self.status != "unknown": + raise ValueError("missing access records must have unknown status") + if self.record_available and not self.record_id: + raise ValueError("available access records require record_id") + if not self.record_available and self.record_id is not None: + raise ValueError("missing access records cannot claim a record_id") + return self + + +class ReadingGap(BaseModel): + kind: str = Field(default="unspecified", min_length=1, max_length=100) + description: str = Field(min_length=1, max_length=2000) + reason: str = Field(min_length=1, max_length=2000) + confidence: float = Field(ge=0, le=1) + evidence_ids: list[str] = Field(default_factory=list) + + +class FollowUpReference(BaseModel): + citation_text: str = Field(min_length=1, max_length=2000) + identifier: str | None = Field(default=None, max_length=500) + reason: str = Field(min_length=1, max_length=1000) + evidence_ids: list[str] = Field(min_length=1) + confidence: float = Field(ge=0, le=1) + + +class ReaderInnovationInference(BaseModel): + statement: str = Field(min_length=1, max_length=2000) + rationale: str = Field(min_length=1, max_length=2000) + evidence_ids: list[str] = Field(min_length=1) + confidence: float = Field(ge=0, le=1) + status: ClaimStatus = ClaimStatus.HYPOTHESIS + + +class IdeaRelationAssessment(BaseModel): + compared_idea: str = Field(default="", max_length=2000) + compared_idea_fingerprint: str | None = None + relation: IdeaRelation = IdeaRelation.UNKNOWN + status: ClaimStatus = ClaimStatus.UNKNOWN + rationale: str = Field(default="", max_length=2000) + evidence_ids: list[str] = Field(default_factory=list) + confidence: float = Field(default=0, ge=0, le=1) + + @model_validator(mode="after") + def decided_relations_require_evidence(self) -> IdeaRelationAssessment: + if self.compared_idea: + expected = stable_id("idea", self.compared_idea) + if self.compared_idea_fingerprint != expected: + raise ValueError("idea relation fingerprint does not match compared_idea") + elif self.compared_idea_fingerprint is not None: + raise ValueError("empty compared_idea cannot have a fingerprint") + if self.relation != IdeaRelation.UNKNOWN and ( + not self.evidence_ids or not self.compared_idea + ): + raise ValueError("decided idea relations require idea and evidence") + if self.relation == IdeaRelation.UNKNOWN and self.status != ClaimStatus.UNKNOWN: + raise ValueError("unknown idea relations must retain unknown status") + if self.relation != IdeaRelation.UNKNOWN and self.status not in { + ClaimStatus.HYPOTHESIS, + ClaimStatus.SUPPORTED, + }: + raise ValueError("classified idea relations must be hypothesis or supported") + return self + + +class ReadingReview(BaseModel): + status: ReadingReviewStatus = ReadingReviewStatus.PENDING + evidence_ids: list[str] = Field(default_factory=list) + reviewer: str | None = Field(default=None, max_length=200) + notes: str = Field(default="", max_length=2000) + reviewed_at: datetime | None = None + + @model_validator(mode="after") + def completed_reviews_require_a_reviewer(self) -> ReadingReview: + if self.status != ReadingReviewStatus.PENDING and ( + not self.reviewer or self.reviewed_at is None or not self.evidence_ids + ): + raise ValueError( + "completed reading reviews require evidence, reviewer, and reviewed_at" + ) + return self + + class ReadingCard(BaseModel): card_id: str = Field(default_factory=lambda: new_id("card")) project_id: str @@ -215,6 +525,258 @@ class ReadingCard(BaseModel): locators: list[str] evidence_ids: list[str] confidence: float = Field(ge=0, le=1) + schema_version: Literal["legacy", "standard_v1"] = "legacy" + document_id: str | None = None + document_checksum: str | None = None + source_scope: ReadingSourceScope = ReadingSourceScope.LEGACY + extraction_profile: str = "legacy" + claims: list[ReadingClaim] = Field(default_factory=list) + diagnostic_ids: list[str] = Field(default_factory=list) + citation: CanonicalCitation | None = None + paper_version: PaperVersionKind = PaperVersionKind.UNKNOWN + paper_version_label: str | None = None + access: ReadingAccessSnapshot = Field(default_factory=ReadingAccessSnapshot) + full_text_hash: str | None = Field( + default=None, + pattern=r"^sha256:[0-9a-f]{64}$", + ) + research_questions: list[ReadingClaim] = Field(default_factory=list) + author_claims: list[ReadingClaim] = Field(default_factory=list) + methods: list[ReadingClaim] = Field(default_factory=list) + datasets: list[ReadingClaim] = Field(default_factory=list) + baselines: list[ReadingClaim] = Field(default_factory=list) + variables: list[ReadingClaim] = Field(default_factory=list) + evaluation_metrics: list[ReadingClaim] = Field(default_factory=list) + key_results: list[ReadingClaim] = Field(default_factory=list) + applicability: list[ReadingClaim] = Field(default_factory=list) + assumptions: list[ReadingClaim] = Field(default_factory=list) + limitation_claims: list[ReadingClaim] = Field(default_factory=list) + failure_cases: list[ReadingClaim] = Field(default_factory=list) + reproducibility_resources: list[ReadingClaim] = Field(default_factory=list) + missing_resources: list[ReadingGap] = Field(default_factory=list) + idea_relation: IdeaRelationAssessment = Field(default_factory=IdeaRelationAssessment) + author_claimed_innovations: list[ReadingClaim] = Field(default_factory=list) + reader_inferred_innovations: list[ReaderInnovationInference] = Field( + default_factory=list + ) + unresolved_questions: list[ReadingGap] = Field(default_factory=list) + follow_up_references: list[FollowUpReference] = Field(default_factory=list) + review: ReadingReview = Field(default_factory=ReadingReview) + + @model_validator(mode="after") + def standard_sections_preserve_lineage(self) -> ReadingCard: + if self.citation is not None and self.citation.source_paper_id != self.paper_id: + raise ValueError("citation source does not match reading-card paper") + if self.access.record_available and self.access.record_id != self.paper_id: + raise ValueError("access source does not match reading-card paper") + if self.full_text_hash is not None and ( + self.source_scope != ReadingSourceScope.FULL_TEXT + or self.full_text_hash != self.document_checksum + ): + raise ValueError("full_text_hash must identify the parsed full-text document") + if self.schema_version == "standard_v1": + if self.citation is None: + raise ValueError("standard reading cards require a citation") + if self.source_scope not in { + ReadingSourceScope.FULL_TEXT, + ReadingSourceScope.ABSTRACT, + }: + raise ValueError("standard reading cards require text or abstract provenance") + if not self.document_id or not self.document_checksum: + raise ValueError("standard reading cards require document provenance") + if not self.claims or not self.evidence_ids: + raise ValueError("standard reading cards require located claims and evidence") + if not self.review.evidence_ids: + raise ValueError("standard reading cards require evidence pending review") + if ( + self.source_scope == ReadingSourceScope.FULL_TEXT + and self.full_text_hash is None + ): + raise ValueError("standard full-text cards require full_text_hash") + + gap_kinds = {item.kind for item in self.unresolved_questions} + requirements = ( + (self.research_questions, "research_question"), + (self.author_claims, "author_claim"), + (self.methods, "method"), + (self.datasets, "dataset"), + (self.baselines, "baseline"), + (self.variables, "variable"), + (self.evaluation_metrics, "evaluation_metric"), + (self.key_results, "key_result"), + (self.applicability, "applicability"), + (self.assumptions, "assumption"), + (self.limitation_claims, "limitation"), + (self.failure_cases, "failure_case"), + (self.author_claimed_innovations, "author_innovation"), + ) + if any(not values and kind not in gap_kinds for values, kind in requirements): + raise ValueError("standard reading-card sections require a value or typed gap") + if any(values and kind in gap_kinds for values, kind in requirements): + raise ValueError("standard reading-card values cannot retain matching typed gaps") + if ( + not self.reproducibility_resources + and not any( + item.kind == "reproducibility_resource" + for item in self.missing_resources + ) + ): + raise ValueError("standard reading cards require resource findings or gaps") + if ( + self.idea_relation.relation == IdeaRelation.UNKNOWN + and "idea_relation" not in gap_kinds + ): + raise ValueError("unknown idea relations require a typed gap") + if ( + self.paper_version == PaperVersionKind.UNKNOWN + and "paper_version" not in gap_kinds + ): + raise ValueError("unknown paper versions require a typed gap") + if ( + ( + not self.access.record_available + or self.access.status in {"unknown", "no_open_fulltext_found"} + ) + and "access" not in gap_kinds + ): + raise ValueError("unknown access status requires a typed gap") + if ( + self.access.status == "open_access" + and self.access.license is None + and "access_license" not in gap_kinds + ): + raise ValueError("missing open-access license requires a typed gap") + if ( + self.source_scope == ReadingSourceScope.ABSTRACT + and "full_text_hash" not in gap_kinds + ): + raise ValueError("abstract cards require a typed full-text hash gap") + + specialized_sections = ( + (self.research_questions, "research_question"), + (self.author_claims, "author_claim"), + (self.methods, "method"), + (self.datasets, "dataset"), + (self.baselines, "baseline"), + (self.variables, "variable"), + (self.evaluation_metrics, "evaluation_metric"), + (self.key_results, "finding"), + (self.applicability, "applicability"), + (self.assumptions, "assumption"), + (self.limitation_claims, "limitation"), + (self.failure_cases, "failure_case"), + (self.reproducibility_resources, "reproducibility_resource"), + (self.author_claimed_innovations, "author_innovation"), + ) + specialized = [claim for values, _ in specialized_sections for claim in values] + claims_by_id = {claim.claim_id: claim for claim in self.claims} + if len(claims_by_id) != len(self.claims): + raise ValueError("reading-card claim IDs must be unique") + if any(claims_by_id.get(claim.claim_id) != claim for claim in specialized): + raise ValueError("standard reading-card sections must reuse exact card claims") + if self.schema_version == "standard_v1" and any( + {claim.claim_id for claim in values} + != { + claim.claim_id + for claim in self.claims + if claim.kind == expected_kind + } + for values, expected_kind in specialized_sections + ): + raise ValueError("standard reading-card claims must project into matching sections") + if self.schema_version == "standard_v1" and any( + claim.kind != expected_kind + or claim.status != ClaimStatus.SUPPORTED + or not claim.bindings + or any( + binding.relation != EvidenceRelation.SUPPORTS + for binding in claim.bindings + ) + for values, expected_kind in specialized_sections + for claim in values + ): + raise ValueError( + "standard reading-card values require matching kinds and decided evidence" + ) + + if self.schema_version == "standard_v1": + for claim in self.claims: + for binding in claim.bindings: + if ( + binding.locator.document_id != self.document_id + or binding.locator.document_checksum != self.document_checksum + ): + raise ValueError("claim locator does not belong to the card document") + expected_quote_checksum = ( + "sha256:" + sha256(binding.quote.encode("utf-8")).hexdigest() + ) + if binding.quote_checksum != expected_quote_checksum: + raise ValueError("claim quote checksum mismatch") + + relation_claims = [ + claim for claim in self.claims if claim.kind.startswith("idea_relation:") + ] + if self.idea_relation.relation == IdeaRelation.UNKNOWN: + if relation_claims: + raise ValueError("unknown idea relations cannot retain relation claims") + else: + expected_kind = f"idea_relation:{self.idea_relation.relation.value}" + relation_evidence_ids = { + binding.evidence_id + for claim in relation_claims + for binding in claim.bindings + } + if ( + len(relation_claims) != 1 + or relation_claims[0].kind != expected_kind + or relation_claims[0].status != self.idea_relation.status + or relation_evidence_ids != set(self.idea_relation.evidence_ids) + or any( + binding.relation != EvidenceRelation.SUPPORTS + or binding.comparison_fingerprint + != self.idea_relation.compared_idea_fingerprint + for binding in relation_claims[0].bindings + ) + ): + raise ValueError("idea relation does not match its internal claim evidence") + + evidence_ids = set(self.evidence_ids) + claim_evidence_ids = { + binding.evidence_id + for claim in self.claims + for binding in claim.bindings + } + referenced_evidence_ids = { + binding.evidence_id + for claim in specialized + for binding in claim.bindings + } + referenced_evidence_ids.update(self.idea_relation.evidence_ids) + referenced_evidence_ids.update(self.review.evidence_ids) + gap_evidence_ids = { + evidence_id + for item in [*self.unresolved_questions, *self.missing_resources] + for evidence_id in item.evidence_ids + } + referenced_evidence_ids.update(gap_evidence_ids) + derived_evidence_ids = set(self.idea_relation.evidence_ids) + for item in self.reader_inferred_innovations: + referenced_evidence_ids.update(item.evidence_ids) + derived_evidence_ids.update(item.evidence_ids) + for item in self.follow_up_references: + referenced_evidence_ids.update(item.evidence_ids) + derived_evidence_ids.update(item.evidence_ids) + referenced_evidence_ids.update(claim_evidence_ids) + if not referenced_evidence_ids.issubset(evidence_ids): + raise ValueError("standard reading-card sections reference unknown evidence") + review_required_evidence_ids = ( + claim_evidence_ids | gap_evidence_ids | derived_evidence_ids + ) + if self.schema_version == "standard_v1" and not review_required_evidence_ids.issubset( + set(self.review.evidence_ids) + ): + raise ValueError("pending review evidence must cover all card claims") + return self class PaperQuestion(BaseModel): @@ -231,6 +793,8 @@ class ReadingAnswer(BaseModel): evidence_ids: list[str] confidence: float = Field(ge=0, le=1) unresolved: bool = False + document_id: str | None = None + citations: list[ClaimEvidenceBinding] = Field(default_factory=list) class InnovationCandidate(BaseModel): @@ -386,7 +950,13 @@ class ResearchState(BaseModel): search_queries: list[str] = Field(default_factory=list) seed_papers: list[dict[str, Any]] = Field(default_factory=list) paper_ids: list[str] = Field(default_factory=list) + parsed_document_ids: list[str] = Field(default_factory=list) reading_card_ids: list[str] = Field(default_factory=list) + reader_cursor: int = Field(default=0, ge=0) + reader_current_paper_id: str | None = None + reader_completed_paper_ids: list[str] = Field(default_factory=list) + reader_failed_paper_ids: list[str] = Field(default_factory=list) + reader_diagnostic_ids: list[str] = Field(default_factory=list) innovation_ids: list[str] = Field(default_factory=list) work_package_ids: list[str] = Field(default_factory=list) manuscript_id: str | None = None diff --git a/src/autoresearch/evidence.py b/src/autoresearch/evidence.py index 9fb1492..51b52c6 100644 --- a/src/autoresearch/evidence.py +++ b/src/autoresearch/evidence.py @@ -10,7 +10,12 @@ def __init__(self, store: RecordStore): def add(self, item: EvidenceItem, *, actor: str = "system") -> EvidenceItem: existing = self.store.get("evidence", item.evidence_id) - if existing is not None and existing != item.model_dump(mode="json"): + normalized_existing = ( + EvidenceItem.model_validate(existing).model_dump(mode="json") + if existing is not None + else None + ) + if normalized_existing is not None and normalized_existing != item.model_dump(mode="json"): raise ValueError(f"evidence is immutable: {item.evidence_id}") if existing is None: self.store.put( diff --git a/src/autoresearch/handoffs.py b/src/autoresearch/handoffs.py index befcbe4..669c105 100644 --- a/src/autoresearch/handoffs.py +++ b/src/autoresearch/handoffs.py @@ -9,6 +9,14 @@ def __init__(self, store: RecordStore): self.store = store def issue(self, envelope: HandoffEnvelope) -> HandoffEnvelope: + raw_existing = self.store.get("handoff", envelope.handoff_id) + if raw_existing is not None: + existing = HandoffEnvelope.model_validate(raw_existing) + expected = envelope.model_dump(mode="json", exclude={"created_at"}) + actual = existing.model_dump(mode="json", exclude={"created_at"}) + if actual != expected: + raise ValueError(f"stable handoff collision: {envelope.handoff_id}") + return existing self.store.put( "handoff", envelope.handoff_id, @@ -37,10 +45,15 @@ def accept( raise ValueError(f"handoff addressed to {envelope.to_agent}, not {expected_receiver}") if envelope.project_id != project_id or envelope.run_id != run_id: raise ValueError("handoff project/run identity mismatch") - self.store.append_event( + if self.store.remember_idempotent( "handoff.accepted", - {"handoff_id": envelope.handoff_id, "receiver": expected_receiver.value}, - project_id=project_id, - actor=expected_receiver.value, - ) + f"{envelope.handoff_id}:{expected_receiver.value}", + {"run_id": run_id}, + ): + self.store.append_event( + "handoff.accepted", + {"handoff_id": envelope.handoff_id, "receiver": expected_receiver.value}, + project_id=project_id, + actor=expected_receiver.value, + ) return envelope diff --git a/src/autoresearch/reader_parser.py b/src/autoresearch/reader_parser.py new file mode 100644 index 0000000..d34cf6b --- /dev/null +++ b/src/autoresearch/reader_parser.py @@ -0,0 +1,997 @@ +from __future__ import annotations + +import re +from dataclasses import dataclass +from hashlib import sha256 +from io import BytesIO +from pathlib import Path + +from pypdf import PdfReader + +from autoresearch.contracts import ( + DocumentBlock, + DocumentFormat, + DocumentParseStatus, + PaperRecord, + ParsedDocument, + ReaderDiagnostic, + ReaderDiagnosticCode, + ReadingSourceScope, + SourceLocator, + stable_id, +) +from autoresearch.storage import RecordStore + +PARSER_NAME = "autoresearch.reader_parser" +PARSER_VERSION = "2" +MAX_DOCUMENT_BYTES = 25 * 1024 * 1024 +MAX_BLOCK_CHARACTERS = 8_000 +MAX_PDF_PAGES = 1_000 +MAX_EXTRACTED_CHARACTERS = 5_000_000 + + +@dataclass(frozen=True) +class ParseResult: + document: ParsedDocument + blocks: list[DocumentBlock] + diagnostics: list[ReaderDiagnostic] + + @property + def usable(self) -> bool: + return bool(self.blocks) and self.document.source_scope in { + ReadingSourceScope.FULL_TEXT, + ReadingSourceScope.ABSTRACT, + } + + +class PaperDocumentParser: + """Deterministically parse supplied paper files into content-addressed blocks.""" + + def __init__(self, store: RecordStore): + self.store = store + + @staticmethod + def checksum(data: bytes) -> str: + return "sha256:" + sha256(data).hexdigest() + + @staticmethod + def _section_label(text: str) -> str | None: + heading = re.compile( + r"^(?:" + r"(?:\d+(?:\.\d+)*[.)]?\s+[^.!?。！？]{2,100})" + r"|abstract|introduction|background|related work|methods?|methodology" + r"|materials and methods|experiments?|results?|discussion|conclusions?" + r"|limitations?|references|摘要|引言|背景|相关工作|方法|实验|结果" + r"|讨论|结论|局限|参考文献" + r")$", + flags=re.IGNORECASE, + ) + for line in text[:1000].splitlines()[:12]: + candidate = re.sub(r"\s+", " ", line).strip() + if candidate and heading.fullmatch(candidate): + return candidate[:500] + return None + + @staticmethod + def _section_at(text: str, offset: int) -> str | None: + heading = re.compile( + r"^(?:" + r"(?:\d+(?:\.\d+)*[.)]?\s+[^.!?。！？]{2,100})" + r"|abstract|introduction|background|related work|methods?|methodology" + r"|materials and methods|experiments?|results?|discussion|conclusions?" + r"|limitations?|references|摘要|引言|背景|相关工作|方法|实验|结果" + r"|讨论|结论|局限|参考文献" + r")$", + flags=re.IGNORECASE, + ) + current: str | None = None + for line in text[:offset].splitlines(): + candidate = re.sub(r"\s+", " ", line).strip() + if candidate and heading.fullmatch(candidate): + current = candidate[:500] + return current + + @staticmethod + def _reference_label(text: str, kind: str) -> str | None: + labels = { + "figure": r"(?:\bfig(?:ure)?\.?\s*\d+[a-z]?\b|图\s*\d+[a-z]?)", + "table": r"(?:\btable\s*\d+[a-z]?\b|表\s*\d+[a-z]?)", + } + label = labels[kind] + prefix = ( + rf"(?i)(?:results?|values?|scores?|findings?|shown|reported|presented|listed|" + rf"see|according\s+to)(?:\s+\w+){{0,4}}\s+(?P