From 71595e89cedddd4908880735645392037f01d7e9 Mon Sep 17 00:00:00 2001 From: jerry <1772030600@qq.com> Date: Wed, 11 Feb 2026 22:00:43 +0800 Subject: [PATCH 1/8] docs: add agent-browser roadmap and issue backlog --- docs/PROJECT_ISSUE_BACKLOG.md | 89 +++++++++++++++++++++++++++++++++++ docs/ROADMAP_TODO.md | 49 +++++++++++++++++++ 2 files changed, 138 insertions(+) diff --git a/docs/PROJECT_ISSUE_BACKLOG.md b/docs/PROJECT_ISSUE_BACKLOG.md index 7fd223db..befafc02 100644 --- a/docs/PROJECT_ISSUE_BACKLOG.md +++ b/docs/PROJECT_ISSUE_BACKLOG.md @@ -265,3 +265,92 @@ Custom fields: - `Priority` (P0/P1/P2) - `Area` (Backend/Frontend/Data/Workflow) - `Size` (S/M/L/XL) + + +## Issue 15 - Agent Browser Source Runner + +- Title: `[Feature] Add agent-browser source runner with fallback connectors` +- Labels: `roadmap`, `phase-4`, `backend`, `integration`, `priority-p1` +- Source TODO: + - `新增 Browser Source Runner` + - `失败后 fallback 到 API connector` + - `DOM 抽取模板化` +- Scope: + - Integrate `vercel-labs/agent-browser` as optional source collector + - Support structured extraction for HF Papers / arXiv / OpenReview + - Persist capture traces (steps/screenshots) for debugging +- Acceptance: + - Browser collector can produce normalized paper candidates + - Fallback path is observable and does not block workflow + +## Issue 16 - Browser-Driven Workflow E2E in CI + +- Title: `[Feature] Add browser-agent E2E for workflow streaming UX` +- Labels: `roadmap`, `phase-4`, `frontend`, `infra`, `priority-p1` +- Source TODO: + - `SSE 增量渲染 E2E` + - `关键截图和性能指标` + - `CI artifacts` +- Scope: + - Run end-to-end test for Search → DailyPaper → Analyze + - Assert DAG restore, non-blank loading state, incremental judge/trend render + - Upload run traces and screenshots in CI +- Acceptance: + - Failing UX regressions are detectable in CI + - Artifacts are attached for triage + +## Issue 17 - Platform Benchmark Monitor Agent + +- Title: `[Feature] Add benchmark monitor agent for HF/AlphaXiv parity tracking` +- Labels: `roadmap`, `phase-4`, `product`, `automation`, `priority-p2` +- Source TODO: + - `对标监测 Agent` + - `能力差距报告 docs/benchmark/` +- Scope: + - Periodically crawl public product pages and extract capability signals + - Generate versioned markdown reports under `docs/benchmark/` +- Acceptance: + - Weekly benchmark report updates automatically + - Capability diff is auditable over time + +## Issue 18 - Push Preview Validation Agent + +- Title: `[Feature] Add browser-agent push preview validation` +- Labels: `roadmap`, `phase-4`, `backend`, `qa`, `priority-p2` +- Source TODO: + - `邮件/Slack/钉钉渲染预览` + - `多端一致性截图` +- Scope: + - Open rendered previews via browser agent and capture screenshots + - Check mandatory sections and formatting constraints +- Acceptance: + - Template breakages are detected before daily dispatch + +## Issue 19 - Browser Extension Smoke Automation + +- Title: `[Feature] Add browser extension smoke tests on arXiv pages` +- Labels: `roadmap`, `phase-4`, `frontend`, `qa`, `priority-p2` +- Source TODO: + - `Browser Extension smoke test` +- Scope: + - Validate content script injection and action flows on arXiv paper pages + - Cover CTA button, detail sheet, and API roundtrip +- Acceptance: + - Core extension flow is tested in CI/nightly + +## Issue 20 - Browser Agent Security & Rate Control + +- Title: `[Feature] Harden browser-agent secrets, auditing, and rate control` +- Labels: `roadmap`, `phase-4`, `security`, `infra`, `priority-p1` +- Source TODO: + - `session 密钥管理` + - `Agent 审计日志` + - `速率限制/并发隔离` +- Scope: + - Implement secure secret injection for browser sessions + - Add auditable logs for domain/action/latency/error + - Add source-level concurrency and throttling policies +- Acceptance: + - No plaintext secrets in repo/runtime logs + - Browser automation runs are traceable and throttled + diff --git a/docs/ROADMAP_TODO.md b/docs/ROADMAP_TODO.md index c57f6c3d..8b2efeee 100644 --- a/docs/ROADMAP_TODO.md +++ b/docs/ROADMAP_TODO.md @@ -208,6 +208,53 @@ --- + +## Phase 3.7 — Agent Browser 自动化能力（基于 vercel-labs/agent-browser） + +> 参考仓库： +> 目标：把"网页交互型"任务从静态 API 拉取升级为可观测、可回放的浏览器 Agent 流程。 + +### 4.1 Source Capture Agent（网页采集增强） + +- [ ] 新增 Browser Source Runner：支持登录后抓取（HF Papers、arXiv、OpenReview） + - 输入：source 配置、cookies/session、抓取策略 + - 输出：结构化 paper candidates + 抓取轨迹（screenshots + step logs） +- [ ] 新增反爬/失败回退链路 + - 失败后自动 fallback 到现有 API connector（papers.cool / arXiv API） + - 记录 fallback 原因到 run metadata +- [ ] 新增 DOM 语义抽取模板 + - 把标题/作者/摘要/链接抽取规则模板化，支持 source 版本升级时快速修复 + +### 4.2 Workflow UX Agent（前端交互与 E2E 验证） + +- [ ] 为 Search → DailyPaper → Analyze 流程增加 browser-driven E2E 回归 + - 覆盖 SSE 增量渲染（Judge/Trend/Insight）和 DAG 状态恢复 +- [ ] 自动录制关键节点截图与性能指标 + - 首屏可见时间、分析阶段空白时长、首条增量结果时间（TTFR） +- [ ] 将 E2E 结果接入 CI artifacts + - 每次 PR 自动上传步骤日志和失败页面快照 + +### 4.3 Community/Platform Agent（平台对标能力） + +- [ ] 新增 HF/AlphaXiv 对标监测 Agent + - 周期性抓取公开页面能力矩阵（发现/排序/交互/推送） + - 生成差距报告写入 `docs/benchmark/`（markdown） +- [ ] 新增 Daily push 预览 Agent + - 自动打开邮件/Slack/钉钉渲染预览页面并截图 + - 验证 BestBlogs 风格模板在多端一致性 +- [ ] 新增 Browser Extension smoke test + - 校验 arXiv 页面注入按钮、详情弹层、跳转链路 + +### 4.4 运维与安全 + +- [ ] 新增 Browser session 密钥管理 + - cookies/token 通过环境变量或密钥服务注入，禁止明文入库 +- [ ] 新增 Agent 审计日志 + - 记录访问域名、操作步骤、耗时、失败原因（可用于问题追踪） +- [ ] 新增速率限制与并发隔离策略 + - 避免批量采集触发封禁，支持 source 级并发控制 + + ## 多智能体系统现状与 OpenClaw 评估 ### 现有多智能体管线 @@ -526,3 +573,5 @@ OpenClaw Skill ── 独立，仅依赖 PaperBot REST API（已有） - 2025-02-10: 创建 ROADMAP_TODO.md，完成对标分析与功能规划 - 2025-02-10: 新增多智能体系统现状盘点（5 套管线 + 15 个 Agent）与 OpenClaw 迁移评估 +- 2026-02-11: 对齐远端 `origin/master` 的 Harvest 基线，保留旧实现到 `backup/feat-dailypaper-sse-stream-pre-harvest-20260211` +- 2026-02-11: 新增 Phase 4（Agent Browser 自动化）任务清单，覆盖采集、E2E、对标监测、安全与限流 From a145ac13572c0bcca67d01fb3b8ffbb1382d25e7 Mon Sep 17 00:00:00 2001 From: jerry <1772030600@qq.com> Date: Wed, 11 Feb 2026 22:20:48 +0800 Subject: [PATCH 2/8] feat: complete phase1 repo enrichment persistence flow --- alembic/versions/0008_paper_repos.py | 100 +++++++ docs/ROADMAP_TODO.md | 9 +- src/paperbot/api/routes/paperscool.py | 135 +++++++-- src/paperbot/api/routes/research.py | 37 ++- src/paperbot/infrastructure/stores/models.py | 98 ++++++- .../infrastructure/stores/paper_store.py | 220 +++++++++------ .../infrastructure/stores/research_store.py | 261 ++++++++++++++++-- tests/unit/test_paper_repo_persistence.py | 96 +++++++ tests/unit/test_paperscool_route.py | 85 ++++++ .../test_research_paper_registry_routes.py | 37 +++ 10 files changed, 926 insertions(+), 152 deletions(-) create mode 100644 alembic/versions/0008_paper_repos.py create mode 100644 tests/unit/test_paper_repo_persistence.py diff --git a/alembic/versions/0008_paper_repos.py b/alembic/versions/0008_paper_repos.py new file mode 100644 index 00000000..60d400b1 --- /dev/null +++ b/alembic/versions/0008_paper_repos.py @@ -0,0 +1,100 @@ +"""paper repos enrichment table + +Revision ID: 0008_paper_repos +Revises: 0007_paper_harvest_tables +Create Date: 2026-02-11 + +Adds paper_repos table to persist repository enrichment metadata linked to +canonical paper registry rows. +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import context, op + +revision = "0008_paper_repos" +down_revision = "0007_paper_harvest_tables" +branch_labels = None +depends_on = None + + +def _is_offline() -> bool: + try: + return bool(context.is_offline_mode()) + except Exception: + return False + + +def _insp(): + return sa.inspect(op.get_bind()) + + +def _has_table(name: str) -> bool: + return _insp().has_table(name) + + +def _get_indexes(table: str) -> set[str]: + idx = set() + for i in _insp().get_indexes(table): + idx.add(str(i.get("name") or "")) + return idx + + +def _create_index(name: str, table: str, cols: list[str]) -> None: + if _is_offline(): + op.create_index(name, table, cols) + return + if name in _get_indexes(table): + return + op.create_index(name, table, cols) + + +def upgrade() -> None: + if _is_offline() or not _has_table("paper_repos"): + op.create_table( + "paper_repos", + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("paper_id", sa.Integer(), sa.ForeignKey("papers.id"), nullable=False), + sa.Column("repo_url", sa.String(length=512), nullable=False, server_default=""), + sa.Column("full_name", sa.String(length=256), nullable=False, server_default=""), + sa.Column("description", sa.Text(), nullable=False, server_default=""), + sa.Column("stars", sa.Integer(), nullable=False, server_default="0"), + sa.Column("forks", sa.Integer(), nullable=False, server_default="0"), + sa.Column("open_issues", sa.Integer(), nullable=False, server_default="0"), + sa.Column("watchers", sa.Integer(), nullable=False, server_default="0"), + sa.Column("language", sa.String(length=64), nullable=False, server_default=""), + sa.Column("license", sa.String(length=64), nullable=False, server_default=""), + sa.Column("archived", sa.Boolean(), nullable=False, server_default=sa.false()), + sa.Column("html_url", sa.String(length=512), nullable=False, server_default=""), + sa.Column("topics_json", sa.Text(), nullable=False, server_default="[]"), + sa.Column("updated_at_remote", sa.DateTime(timezone=True), nullable=True), + sa.Column("pushed_at_remote", sa.DateTime(timezone=True), nullable=True), + sa.Column("query", sa.String(length=256), nullable=False, server_default=""), + sa.Column( + "source", + sa.String(length=32), + nullable=False, + server_default="paperscool_repo_enrich", + ), + sa.Column("metadata_json", sa.Text(), nullable=False, server_default="{}"), + sa.Column("synced_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.UniqueConstraint("paper_id", "repo_url", name="uq_paper_repos_paper_repo"), + ) + + _create_index("ix_paper_repos_paper_id", "paper_repos", ["paper_id"]) + _create_index("ix_paper_repos_repo_url", "paper_repos", ["repo_url"]) + _create_index("ix_paper_repos_full_name", "paper_repos", ["full_name"]) + _create_index("ix_paper_repos_stars", "paper_repos", ["stars"]) + _create_index("ix_paper_repos_archived", "paper_repos", ["archived"]) + _create_index("ix_paper_repos_query", "paper_repos", ["query"]) + _create_index("ix_paper_repos_source", "paper_repos", ["source"]) + _create_index("ix_paper_repos_synced_at", "paper_repos", ["synced_at"]) + _create_index("ix_paper_repos_updated_at_remote", "paper_repos", ["updated_at_remote"]) + _create_index("ix_paper_repos_pushed_at_remote", "paper_repos", ["pushed_at_remote"]) + + +def downgrade() -> None: + op.drop_table("paper_repos") diff --git a/docs/ROADMAP_TODO.md b/docs/ROADMAP_TODO.md index 8b2efeee..4ca97871 100644 --- a/docs/ROADMAP_TODO.md +++ b/docs/ROADMAP_TODO.md @@ -63,19 +63,19 @@ - [x] API：`GET /api/research/papers/saved`（用户收藏列表，支持排序：judge_score / saved_at / published_at） - [x] API：`POST /api/research/papers/{paper_id}/status`（更新阅读状态） - [x] API：`GET /api/research/papers/{paper_id}`（论文详情，聚合 judge + feedback + summary） -- [ ] 前端：收藏列表页面组件 +- [x] 前端：收藏列表页面组件 - 文件：`web/src/components/research/SavedPapersList.tsx` ### 1.3 GitHub Repo 关联（Repo Enrichment） -- [ ] 新增 `PaperRepoModel` 表 +- [x] 新增 `PaperRepoModel` 表 - 字段：`paper_id`、`repo_url`、`owner`、`name`、`stars`、`forks`、`last_commit_at`、`language`、`description`、`fetched_at` - [x] Enrichment 服务：从论文 abstract/url/external_url 中提取 GitHub 链接并补元数据 - 当前实现：`src/paperbot/api/routes/paperscool.py`（后续可下沉到 service） - 提取来源：`github_url/external_url/url/pdf_url/alternative_urls + snippet/abstract` - 调用 GitHub API 补元数据（stars/forks/language/updated_at） -- [ ] DailyPaper 生成后异步调用 repo enrichment -- [ ] API：`GET /api/research/papers/{paper_id}/repos` +- [x] DailyPaper 生成后异步调用 repo enrichment +- [x] API：`GET /api/research/papers/{paper_id}/repos` - [x] API：`POST /api/research/paperscool/repos`（批量，含 stars/活跃度） --- @@ -575,3 +575,4 @@ OpenClaw Skill ── 独立，仅依赖 PaperBot REST API（已有） - 2025-02-10: 新增多智能体系统现状盘点（5 套管线 + 15 个 Agent）与 OpenClaw 迁移评估 - 2026-02-11: 对齐远端 `origin/master` 的 Harvest 基线，保留旧实现到 `backup/feat-dailypaper-sse-stream-pre-harvest-20260211` - 2026-02-11: 新增 Phase 4（Agent Browser 自动化）任务清单，覆盖采集、E2E、对标监测、安全与限流 +- 2026-02-11: 完成 Phase 1 收尾（PaperRepoModel + /papers/{paper_id}/repos + DailyPaper 异步 repo enrichment），并修复 harvest 基线下 paper store 兼容性 diff --git a/src/paperbot/api/routes/paperscool.py b/src/paperbot/api/routes/paperscool.py index aa611987..7da23b62 100644 --- a/src/paperbot/api/routes/paperscool.py +++ b/src/paperbot/api/routes/paperscool.py @@ -3,11 +3,11 @@ import copy import os import re +from threading import Thread from typing import Any, Dict, List, Optional from urllib.parse import urlparse import requests - from fastapi import APIRouter, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel, Field @@ -29,6 +29,7 @@ select_judge_candidates, ) from paperbot.application.workflows.paperscool_topic_search import PapersCoolTopicSearchWorkflow +from paperbot.infrastructure.stores.research_store import SqlAlchemyResearchStore from paperbot.utils.text_processing import extract_github_url router = APIRouter() @@ -90,7 +91,9 @@ class DailyPaperRequest(BaseModel): top_n: int = Field(10, ge=1, le=200) formats: List[str] = Field(default_factory=lambda: ["both"]) save: bool = False - output_dir: str = Field("./reports/dailypaper", description="Relative path under project root for saving reports") + output_dir: str = Field( + "./reports/dailypaper", description="Relative path under project root for saving reports" + ) enable_llm_analysis: bool = False llm_features: List[str] = Field(default_factory=lambda: ["summary"]) enable_judge: bool = False @@ -126,6 +129,7 @@ class PapersCoolReposRequest(BaseModel): papers: List[Dict[str, Any]] = Field(default_factory=list) max_items: int = Field(100, ge=1, le=1000) include_github_api: bool = True + persist: bool = False class PapersCoolReposResponse(BaseModel): @@ -133,6 +137,7 @@ class PapersCoolReposResponse(BaseModel): matched_repos: int github_api_used: bool repos: List[Dict[str, Any]] + persist_summary: Optional[Dict[str, int]] = None @router.post("/research/paperscool/search", response_model=PapersCoolSearchResponse) @@ -214,7 +219,11 @@ async def _dailypaper_stream(req: DailyPaperRequest): yield StreamEvent( type="progress", - data={"phase": "llm", "message": "Starting LLM enrichment...", "total": summary_total}, + data={ + "phase": "llm", + "message": "Starting LLM enrichment...", + "total": summary_total, + }, ) for query in report.get("queries") or []: @@ -241,7 +250,9 @@ async def _dailypaper_stream(req: DailyPaperRequest): if "relevance" in features: for item in top_items: - item["relevance"] = llm_service.assess_relevance(paper=item, query=query_name) + item["relevance"] = llm_service.assess_relevance( + paper=item, query=query_name + ) if "summary" not in features: summary_done += 1 @@ -259,7 +270,10 @@ async def _dailypaper_stream(req: DailyPaperRequest): ) if "insight" in features: - yield StreamEvent(type="progress", data={"phase": "insight", "message": "Generating daily insight..."}) + yield StreamEvent( + type="progress", + data={"phase": "insight", "message": "Generating daily insight..."}, + ) llm_block["daily_insight"] = llm_service.generate_daily_insight(report) yield StreamEvent(type="insight", data={"analysis": llm_block["daily_insight"]}) @@ -384,13 +398,15 @@ async def _dailypaper_stream(req: DailyPaperRequest): kept.append(item) else: removed.append(item) - filter_log.append({ - "query": query_name, - "title": item.get("title") or "Untitled", - "recommendation": rec, - "overall": j.get("overall"), - "action": "removed", - }) + filter_log.append( + { + "query": query_name, + "title": item.get("title") or "Untitled", + "recommendation": rec, + "overall": j.get("overall"), + "action": "removed", + } + ) else: # No judge score — keep by default (unjudged papers) kept.append(item) @@ -442,6 +458,8 @@ async def _dailypaper_stream(req: DailyPaperRequest): except Exception as exc: report["judge_registry_ingest"] = {"error": str(exc)} + _enqueue_repo_enrichment_async(report) + markdown = render_daily_paper_markdown(report) markdown_path = None @@ -459,7 +477,9 @@ async def _dailypaper_stream(req: DailyPaperRequest): json_path = artifacts.json_path if req.notify: - yield StreamEvent(type="progress", data={"phase": "notify", "message": "Sending notifications..."}) + yield StreamEvent( + type="progress", data={"phase": "notify", "message": "Sending notifications..."} + ) notify_service = DailyPushService.from_env() notify_result = notify_service.push_dailypaper( report=report, @@ -523,6 +543,8 @@ def _sync_daily_report(req: DailyPaperRequest, cleaned_queries: List[str]): except Exception as exc: report["registry_ingest"] = {"error": str(exc)} + _enqueue_repo_enrichment_async(report) + markdown = render_daily_paper_markdown(report) markdown_path = None @@ -692,16 +714,19 @@ def _fetch_github_repo_metadata(repo_url: str, token: Optional[str]) -> Dict[str } -@router.post("/research/paperscool/repos", response_model=PapersCoolReposResponse) -def enrich_papers_with_repo_data(req: PapersCoolReposRequest): - papers: List[Dict[str, Any]] = [] - if isinstance(req.report, dict): - papers.extend(_flatten_report_papers(req.report)) - papers.extend(list(req.papers or [])) +def _env_flag(name: str, default: bool = False) -> bool: + raw = os.getenv(name) + if raw is None: + return bool(default) + return str(raw).strip().lower() not in {"", "0", "false", "off", "no"} - if not papers: - raise HTTPException(status_code=400, detail="report or papers is required") +def _collect_repo_enrichment_rows( + *, + papers: List[Dict[str, Any]], + max_items: int, + include_github_api: bool, +) -> tuple[List[Dict[str, Any]], List[Dict[str, Any]]]: deduped: List[Dict[str, Any]] = [] seen: set[str] = set() for item in papers: @@ -711,7 +736,7 @@ def enrich_papers_with_repo_data(req: PapersCoolReposRequest): seen.add(key) deduped.append(item) - selected = deduped[: max(1, int(req.max_items))] + selected = deduped[: max(1, int(max_items))] token = os.getenv("GITHUB_TOKEN") or os.getenv("GH_TOKEN") # TODO: GitHub API calls are sequential — switch to concurrent.futures or @@ -729,21 +754,83 @@ def enrich_papers_with_repo_data(req: PapersCoolReposRequest): "paper_url": item.get("url") or item.get("external_url") or "", "repo_url": repo_url, } - if req.include_github_api: + if include_github_api: row["github"] = _fetch_github_repo_metadata(repo_url=repo_url, token=token) repos.append(row) - if req.include_github_api: + if include_github_api: repos.sort( key=lambda row: int(((row.get("github") or {}).get("stars") or -1)), reverse=True, ) + return selected, repos + + +def _persist_repo_enrichment_async(report: Dict[str, Any]) -> None: + try: + max_items_raw = os.getenv("PAPERBOT_REPO_ENRICH_MAX_ITEMS", "100") + max_items = max(1, int(max_items_raw)) + except Exception: + max_items = 100 + + include_github_api = _env_flag("PAPERBOT_REPO_ENRICH_INCLUDE_GITHUB_API", default=True) + + try: + papers = _flatten_report_papers(report) + if not papers: + return + _, repos = _collect_repo_enrichment_rows( + papers=papers, + max_items=max_items, + include_github_api=include_github_api, + ) + if not repos: + return + store = SqlAlchemyResearchStore() + store.ingest_repo_enrichment_rows(rows=repos, source="paperscool_daily_async") + except Exception: + # Async best-effort hook: ignore failures to avoid affecting daily report flow. + return + + +def _enqueue_repo_enrichment_async(report: Dict[str, Any]) -> None: + if not _env_flag("PAPERBOT_REPO_ENRICH_ASYNC", default=True): + return + Thread( + target=_persist_repo_enrichment_async, args=(copy.deepcopy(report),), daemon=True + ).start() + + +@router.post("/research/paperscool/repos", response_model=PapersCoolReposResponse) +def enrich_papers_with_repo_data(req: PapersCoolReposRequest): + papers: List[Dict[str, Any]] = [] + if isinstance(req.report, dict): + papers.extend(_flatten_report_papers(req.report)) + papers.extend(list(req.papers or [])) + + if not papers: + raise HTTPException(status_code=400, detail="report or papers is required") + + selected, repos = _collect_repo_enrichment_rows( + papers=papers, + max_items=req.max_items, + include_github_api=bool(req.include_github_api), + ) + + persist_summary: Optional[Dict[str, int]] = None + if req.persist: + store = SqlAlchemyResearchStore() + persist_summary = store.ingest_repo_enrichment_rows( + rows=repos, source="paperscool_repos_api" + ) + return PapersCoolReposResponse( total_candidates=len(selected), matched_repos=len(repos), github_api_used=bool(req.include_github_api), repos=repos, + persist_summary=persist_summary, ) diff --git a/src/paperbot/api/routes/research.py b/src/paperbot/api/routes/research.py index 60f8c428..a4f6bcca 100644 --- a/src/paperbot/api/routes/research.py +++ b/src/paperbot/api/routes/research.py @@ -1,9 +1,9 @@ from __future__ import annotations -from collections import Counter -from datetime import datetime, timezone import os import re +from collections import Counter +from datetime import datetime, timezone from typing import Any, Dict, List, Optional, Tuple from fastapi import APIRouter, BackgroundTasks, HTTPException, Query @@ -11,13 +11,13 @@ from paperbot.context_engine import ContextEngine, ContextEngineConfig from paperbot.context_engine.track_router import TrackRouter -from paperbot.utils.logging_config import Logger, LogFiles, set_trace_id -from paperbot.infrastructure.stores.memory_store import SqlAlchemyMemoryStore from paperbot.infrastructure.api_clients.semantic_scholar import SemanticScholarClient +from paperbot.infrastructure.stores.memory_store import SqlAlchemyMemoryStore from paperbot.infrastructure.stores.research_store import SqlAlchemyResearchStore from paperbot.memory.eval.collector import MemoryMetricCollector from paperbot.memory.extractor import extract_memories from paperbot.memory.schema import MemoryCandidate, NormalizedMessage +from paperbot.utils.logging_config import LogFiles, Logger, set_trace_id router = APIRouter() @@ -676,7 +676,9 @@ def add_paper_feedback(req: PaperFeedbackRequest): # If action is "save" and we have paper metadata, insert into papers table if req.action == "save" and req.paper_title: - Logger.info("Save action detected, inserting paper into papers table", file=LogFiles.HARVEST) + Logger.info( + "Save action detected, inserting paper into papers table", file=LogFiles.HARVEST + ) try: from paperbot.domain.harvest import HarvestedPaper, HarvestSource @@ -696,7 +698,9 @@ def add_paper_feedback(req: PaperFeedbackRequest): source=source, abstract=req.paper_abstract or "", authors=req.paper_authors or [], - semantic_scholar_id=req.paper_id if source == HarvestSource.SEMANTIC_SCHOLAR else None, + semantic_scholar_id=( + req.paper_id if source == HarvestSource.SEMANTIC_SCHOLAR else None + ), arxiv_id=req.paper_id if source == HarvestSource.ARXIV else None, openalex_id=req.paper_id if source == HarvestSource.OPENALEX else None, year=req.paper_year, @@ -713,7 +717,9 @@ def add_paper_feedback(req: PaperFeedbackRequest): library_paper_id = result.id # Store library_paper_id in metadata for joins, keep paper_id as external ID meta["library_paper_id"] = library_paper_id - Logger.info(f"Paper saved to library with id={library_paper_id}", file=LogFiles.HARVEST) + Logger.info( + f"Paper saved to library with id={library_paper_id}", file=LogFiles.HARVEST + ) except Exception as e: Logger.warning(f"Failed to save paper to library: {e}", file=LogFiles.HARVEST) @@ -772,6 +778,11 @@ class PaperDetailResponse(BaseModel): detail: Dict[str, Any] +class PaperRepoListResponse(BaseModel): + paper_id: str + repos: List[Dict[str, Any]] + + @router.post("/research/papers/{paper_id}/status", response_model=PaperReadingStatusResponse) def update_paper_status(paper_id: str, req: PaperReadingStatusRequest): status = _research_store.set_paper_reading_status( @@ -804,6 +815,14 @@ def get_paper_detail(paper_id: str, user_id: str = "default"): return PaperDetailResponse(detail=detail) +@router.get("/research/papers/{paper_id}/repos", response_model=PaperRepoListResponse) +def get_paper_repos(paper_id: str): + repos = _research_store.list_paper_repos(paper_id=paper_id) + if repos is None: + raise HTTPException(status_code=404, detail="Paper not found in registry") + return PaperRepoListResponse(paper_id=paper_id, repos=repos) + + class RouterSuggestRequest(BaseModel): user_id: str = "default" query: str = Field(..., min_length=1) @@ -883,7 +902,9 @@ async def build_context(req: ContextRequest): include_cross_track=req.include_cross_track, ) paper_count = len(pack.get("paper_recommendations", [])) - Logger.info(f"Context pack built successfully, found {paper_count} papers", file=LogFiles.HARVEST) + Logger.info( + f"Context pack built successfully, found {paper_count} papers", file=LogFiles.HARVEST + ) return ContextResponse(context_pack=pack) finally: await engine.close() diff --git a/src/paperbot/infrastructure/stores/models.py b/src/paperbot/infrastructure/stores/models.py index f6f58660..0cb436af 100644 --- a/src/paperbot/infrastructure/stores/models.py +++ b/src/paperbot/infrastructure/stores/models.py @@ -4,7 +4,7 @@ from datetime import datetime from typing import Any, Dict, Optional -from sqlalchemy import DateTime, Float, ForeignKey, Integer, String, Text, UniqueConstraint +from sqlalchemy import Boolean, DateTime, Float, ForeignKey, Integer, String, Text, UniqueConstraint from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship @@ -449,6 +449,7 @@ class ResearchMilestoneModel(Base): track = relationship("ResearchTrackModel", back_populates="milestones") + class PaperFeedbackModel(Base): """User feedback on recommended/seen papers (track-scoped).""" @@ -503,6 +504,64 @@ class PaperJudgeScoreModel(Base): paper = relationship("PaperModel", back_populates="judge_scores") +class PaperRepoModel(Base): + """Repository enrichment metadata linked to canonical papers.""" + + __tablename__ = "paper_repos" + __table_args__ = (UniqueConstraint("paper_id", "repo_url", name="uq_paper_repos_paper_repo"),) + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + paper_id: Mapped[int] = mapped_column(Integer, ForeignKey("papers.id"), index=True) + + repo_url: Mapped[str] = mapped_column(String(512), default="", index=True) + full_name: Mapped[str] = mapped_column(String(256), default="", index=True) + description: Mapped[str] = mapped_column(Text, default="") + + stars: Mapped[int] = mapped_column(Integer, default=0, index=True) + forks: Mapped[int] = mapped_column(Integer, default=0) + open_issues: Mapped[int] = mapped_column(Integer, default=0) + watchers: Mapped[int] = mapped_column(Integer, default=0) + + language: Mapped[str] = mapped_column(String(64), default="") + license: Mapped[str] = mapped_column(String(64), default="") + archived: Mapped[bool] = mapped_column(Boolean, default=False, index=True) + + html_url: Mapped[str] = mapped_column(String(512), default="") + topics_json: Mapped[str] = mapped_column(Text, default="[]") + + updated_at_remote: Mapped[Optional[datetime]] = mapped_column( + DateTime(timezone=True), nullable=True, index=True + ) + pushed_at_remote: Mapped[Optional[datetime]] = mapped_column( + DateTime(timezone=True), nullable=True, index=True + ) + + query: Mapped[str] = mapped_column(String(256), default="", index=True) + source: Mapped[str] = mapped_column(String(32), default="paperscool_repo_enrich", index=True) + metadata_json: Mapped[str] = mapped_column(Text, default="{}") + + synced_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True) + + paper = relationship("PaperModel", back_populates="repo_rows") + + def set_topics(self, values: Optional[list[str]]) -> None: + self.topics_json = json.dumps( + [str(v) for v in (values or []) if str(v).strip()], + ensure_ascii=False, + ) + + def get_topics(self) -> list[str]: + try: + data = json.loads(self.topics_json or "[]") + if isinstance(data, list): + return [str(v) for v in data if str(v).strip()] + except Exception: + pass + return [] + + class PaperReadingStatusModel(Base): """Per-user reading lifecycle state for a paper.""" @@ -635,9 +694,15 @@ class PaperModel(Base): # Canonical identifiers (for deduplication) doi: Mapped[Optional[str]] = mapped_column(String(128), unique=True, nullable=True, index=True) - arxiv_id: Mapped[Optional[str]] = mapped_column(String(32), unique=True, nullable=True, index=True) - semantic_scholar_id: Mapped[Optional[str]] = mapped_column(String(64), unique=True, nullable=True, index=True) - openalex_id: Mapped[Optional[str]] = mapped_column(String(64), unique=True, nullable=True, index=True) + arxiv_id: Mapped[Optional[str]] = mapped_column( + String(32), unique=True, nullable=True, index=True + ) + semantic_scholar_id: Mapped[Optional[str]] = mapped_column( + String(64), unique=True, nullable=True, index=True + ) + openalex_id: Mapped[Optional[str]] = mapped_column( + String(64), unique=True, nullable=True, index=True + ) title_hash: Mapped[str] = mapped_column(String(64), index=True) # SHA256 of normalized title # Core metadata @@ -658,18 +723,27 @@ class PaperModel(Base): fields_of_study_json: Mapped[str] = mapped_column(Text, default="[]") # Source tracking - primary_source: Mapped[str] = mapped_column(String(32), default="") # First source that found this paper - sources_json: Mapped[str] = mapped_column(Text, default="[]") # All sources that returned this paper + primary_source: Mapped[str] = mapped_column( + String(32), default="" + ) # First source that found this paper + sources_json: Mapped[str] = mapped_column( + Text, default="[]" + ) # All sources that returned this paper # Timestamps - created_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True, index=True) + created_at: Mapped[Optional[datetime]] = mapped_column( + DateTime(timezone=True), nullable=True, index=True + ) updated_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) - deleted_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) # Soft delete + deleted_at: Mapped[Optional[datetime]] = mapped_column( + DateTime(timezone=True), nullable=True + ) # Soft delete # Relationships feedback_rows = relationship("PaperFeedbackModel", back_populates="paper") judge_scores = relationship("PaperJudgeScoreModel", back_populates="paper") reading_status_rows = relationship("PaperReadingStatusModel", back_populates="paper") + repo_rows = relationship("PaperRepoModel", back_populates="paper") def get_authors(self) -> list: try: @@ -720,14 +794,18 @@ class HarvestRunModel(Base): max_results_per_source: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) # Results - status: Mapped[Optional[str]] = mapped_column(String(32), default="running", index=True) # running/success/partial/failed + status: Mapped[Optional[str]] = mapped_column( + String(32), default="running", index=True + ) # running/success/partial/failed papers_found: Mapped[Optional[int]] = mapped_column(Integer, default=0) papers_new: Mapped[Optional[int]] = mapped_column(Integer, default=0) papers_deduplicated: Mapped[Optional[int]] = mapped_column(Integer, default=0) error_json: Mapped[str] = mapped_column(Text, default="{}") # Timestamps - started_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True, index=True) + started_at: Mapped[Optional[datetime]] = mapped_column( + DateTime(timezone=True), nullable=True, index=True + ) ended_at: Mapped[Optional[datetime]] = mapped_column(DateTime(timezone=True), nullable=True) def get_keywords(self) -> list: diff --git a/src/paperbot/infrastructure/stores/paper_store.py b/src/paperbot/infrastructure/stores/paper_store.py index b26327a1..9252a990 100644 --- a/src/paperbot/infrastructure/stores/paper_store.py +++ b/src/paperbot/infrastructure/stores/paper_store.py @@ -1,29 +1,24 @@ from __future__ import annotations -from datetime import datetime, timezone -from typing import Any, Dict, Iterable, List, Optional - -from sqlalchemy import desc, func, select - -from paperbot.domain.paper_identity import normalize_arxiv_id, normalize_doi -from paperbot.infrastructure.stores.models import Base, PaperJudgeScoreModel, PaperModel - +import hashlib import json from dataclasses import dataclass from datetime import datetime, timezone -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, Dict, Iterable, List, Optional, Tuple -from sqlalchemy import Integer, String, cast, func, or_, select +from sqlalchemy import Integer, String, cast, desc, func, or_, select -from paperbot.utils.logging_config import Logger, LogFiles from paperbot.domain.harvest import HarvestedPaper, HarvestSource +from paperbot.domain.paper_identity import normalize_arxiv_id, normalize_doi from paperbot.infrastructure.stores.models import ( Base, HarvestRunModel, PaperFeedbackModel, + PaperJudgeScoreModel, PaperModel, ) from paperbot.infrastructure.stores.sqlalchemy_db import SessionProvider, get_db_url +from paperbot.utils.logging_config import LogFiles, Logger def _utcnow() -> datetime: @@ -72,6 +67,8 @@ def _as_utc(value: Optional[datetime]) -> Optional[datetime]: class SqlAlchemyPaperStore: """Canonical paper registry with idempotent upsert for daily workflows.""" + + @dataclass class LibraryPaper: """Paper with library metadata (saved_at, track_id, action).""" @@ -107,7 +104,6 @@ def upsert_paper( seen_at: Optional[datetime] = None, ) -> Dict[str, Any]: now = _utcnow() - first_seen = seen_at or now title = str(paper.get("title") or "").strip() url = str(paper.get("url") or "").strip() @@ -133,30 +129,39 @@ def upsert_paper( source_hint or (paper.get("sources") or [None])[0] or paper.get("source") + or paper.get("primary_source") or "papers_cool" ) venue = str(paper.get("subject_or_venue") or paper.get("venue") or "").strip() - published_at = _parse_datetime( - paper.get("published_at") or paper.get("published") or paper.get("publicationDate") + publication_date = str( + paper.get("publication_date") + or paper.get("published_at") + or paper.get("published") + or "" + ).strip() + + year_raw = ( + paper.get("year") if paper.get("year") is not None else paper.get("published_year") ) + try: + year = int(year_raw) if year_raw is not None else None + except Exception: + year = None + + citation_raw = paper.get("citation_count") + try: + citation_count = int(citation_raw) if citation_raw is not None else 0 + except Exception: + citation_count = 0 authors = _safe_list(paper.get("authors")) keywords = _safe_list(paper.get("keywords")) + fields_of_study = _safe_list(paper.get("fields_of_study")) - metadata = { - "paper_id": paper.get("paper_id"), - "matched_queries": _safe_list(paper.get("matched_queries")), - "branches": _safe_list(paper.get("branches")), - "score": paper.get("score"), - "pdf_stars": paper.get("pdf_stars"), - "kimi_stars": paper.get("kimi_stars"), - "alternative_urls": _safe_list(paper.get("alternative_urls")), - } + normalized_title = title.lower().strip() or "untitled" + title_hash = hashlib.sha256(normalized_title.encode("utf-8")).hexdigest() with self._provider.session() as session: - # TODO: title+url fallback query uses scalar_one_or_none() which - # raises MultipleResultsFound if duplicates exist. Switch to - # .first() or add .limit(1) for safety. row = None if arxiv_id: row = session.execute( @@ -171,50 +176,55 @@ def upsert_paper( select(PaperModel).where(PaperModel.url == url) ).scalar_one_or_none() if row is None and title: - row = session.execute( - select(PaperModel).where( - func.lower(PaperModel.title) == title.lower(), - PaperModel.url == url, + row = ( + session.execute( + select(PaperModel) + .where(func.lower(PaperModel.title) == title.lower()) + .limit(1) ) - ).scalar_one_or_none() + .scalars() + .first() + ) created = row is None if row is None: row = PaperModel( - first_seen_at=_as_utc(first_seen) or now, + title_hash=title_hash, created_at=now, updated_at=now, ) session.add(row) - # Keep earliest first_seen_at for existing records. - existing_seen = _as_utc(row.first_seen_at) - candidate_seen = _as_utc(first_seen) or now - if not existing_seen or candidate_seen < existing_seen: - row.first_seen_at = candidate_seen - if arxiv_id: row.arxiv_id = arxiv_id if doi: row.doi = doi - + row.title_hash = title_hash row.title = title or row.title or "" row.abstract = abstract or row.abstract or "" - row.url = url or row.url or "" - row.external_url = external_url or row.external_url or "" - row.pdf_url = pdf_url or row.pdf_url or "" - row.source = str(source or row.source or "papers_cool") - row.venue = venue or row.venue or "" - row.published_at = _as_utc(published_at) or _as_utc(row.published_at) - # TODO: unconditional set_authors/set_keywords/set_metadata may wipe - # existing data when new paper dict has empty values. Consider - # preserving existing values when incoming data is empty: - # row.set_authors(authors or row.get_authors()) - row.set_authors(authors) - row.set_keywords(keywords) - row.set_metadata(metadata) - row.updated_at = now + row.url = url or row.url or None + row.pdf_url = pdf_url or row.pdf_url or None + row.venue = venue or row.venue or None + row.year = year if year is not None else row.year + row.publication_date = publication_date or row.publication_date + row.citation_count = max(citation_count, int(row.citation_count or 0)) + + if authors: + row.authors_json = json.dumps(authors, ensure_ascii=False) + if keywords: + row.keywords_json = json.dumps(keywords, ensure_ascii=False) + if fields_of_study: + row.fields_of_study_json = json.dumps(fields_of_study, ensure_ascii=False) + + source_text = str(source or "").strip() or "papers_cool" + row.primary_source = source_text + existing_sources = row.get_sources() + merged_sources = ( + sorted({*existing_sources, source_text}) if source_text else existing_sources + ) + row.set_sources(merged_sources) + row.updated_at = now session.commit() session.refresh(row) @@ -247,10 +257,10 @@ def upsert_many( def list_recent(self, *, limit: int = 50, source: Optional[str] = None) -> List[Dict[str, Any]]: with self._provider.session() as session: - stmt = select(PaperModel) + stmt = select(PaperModel).where(PaperModel.deleted_at.is_(None)) if source: - stmt = stmt.where(PaperModel.source == source) - stmt = stmt.order_by(desc(PaperModel.first_seen_at), desc(PaperModel.id)).limit( + stmt = stmt.where(PaperModel.primary_source == source) + stmt = stmt.order_by(desc(PaperModel.updated_at), desc(PaperModel.id)).limit( max(1, int(limit)) ) rows = session.execute(stmt).scalars().all() @@ -332,25 +342,37 @@ def upsert_judge_scores_from_report(self, report: Dict[str, Any]) -> Dict[str, i @staticmethod def _paper_to_dict(row: PaperModel) -> Dict[str, Any]: + publication_date = row.publication_date + published_at = publication_date + return { "id": int(row.id), "arxiv_id": row.arxiv_id, "doi": row.doi, + "semantic_scholar_id": row.semantic_scholar_id, + "openalex_id": row.openalex_id, "title": row.title, "authors": row.get_authors(), "abstract": row.abstract, "url": row.url, - "external_url": row.external_url, + "external_url": row.url, "pdf_url": row.pdf_url, - "source": row.source, + "source": row.primary_source, + "primary_source": row.primary_source, "venue": row.venue, - "published_at": row.published_at.isoformat() if row.published_at else None, - "first_seen_at": row.first_seen_at.isoformat() if row.first_seen_at else None, + "year": row.year, + "publication_date": publication_date, + "published_at": published_at, + "first_seen_at": row.created_at.isoformat() if row.created_at else None, "keywords": row.get_keywords(), - "metadata": row.get_metadata(), + "fields_of_study": row.get_fields_of_study(), + "sources": row.get_sources(), + "citation_count": int(row.citation_count or 0), + "metadata": {}, "created_at": row.created_at.isoformat() if row.created_at else None, "updated_at": row.updated_at.isoformat() if row.updated_at else None, } + def upsert_papers_batch( self, papers: List[HarvestedPaper], @@ -376,7 +398,9 @@ def upsert_papers_batch( self._update_paper(existing, paper, now) updated_count += 1 else: - Logger.info("No existing paper found, creating new record", file=LogFiles.HARVEST) + Logger.info( + "No existing paper found, creating new record", file=LogFiles.HARVEST + ) model = self._create_model(paper, now) session.add(model) new_count += 1 @@ -384,7 +408,10 @@ def upsert_papers_batch( Logger.info("Committing transaction to database", file=LogFiles.HARVEST) session.commit() - Logger.info(f"Batch upsert complete: {new_count} new, {updated_count} updated", file=LogFiles.HARVEST) + Logger.info( + f"Batch upsert complete: {new_count} new, {updated_count} updated", + file=LogFiles.HARVEST, + ) return new_count, updated_count def _find_existing(self, session, paper: HarvestedPaper) -> Optional[PaperModel]: @@ -452,9 +479,7 @@ def _create_model(self, paper: HarvestedPaper, now: datetime) -> PaperModel: updated_at=now, ) - def _update_paper( - self, existing: PaperModel, paper: HarvestedPaper, now: datetime - ) -> None: + def _update_paper(self, existing: PaperModel, paper: HarvestedPaper, now: datetime) -> None: """Update existing paper with new data.""" # Fill in missing identifiers if not existing.doi and paper.doi: @@ -546,9 +571,7 @@ def search_papers( # Keyword filter (search in keywords_json) if keywords: - keyword_conditions = [ - PaperModel.keywords_json.ilike(f"%{kw}%") for kw in keywords - ] + keyword_conditions = [PaperModel.keywords_json.ilike(f"%{kw}%") for kw in keywords] stmt = stmt.where(or_(*keyword_conditions)) # Year filters (use explicit None check to allow year_from=0 if needed) @@ -598,9 +621,7 @@ def get_paper_by_id(self, paper_id: int) -> Optional[PaperModel]: ) ).scalar_one_or_none() - def get_paper_by_source_id( - self, source: HarvestSource, source_id: str - ) -> Optional[PaperModel]: + def get_paper_by_source_id(self, source: HarvestSource, source_id: str) -> Optional[PaperModel]: """ Get a paper by its source-specific ID. @@ -649,7 +670,9 @@ def get_user_library( # 1. Integer ID as string (from harvest saves): "123" -> join on papers.id # 2. Semantic Scholar ID (from recommendation saves): "abc123" -> join on papers.semantic_scholar_id - Logger.info("Executing database query to join papers with feedback", file=LogFiles.HARVEST) + Logger.info( + "Executing database query to join papers with feedback", file=LogFiles.HARVEST + ) # First, get all matching paper-feedback pairs # Join on external IDs (semantic_scholar_id, arxiv_id, openalex_id) # This avoids CAST errors on PostgreSQL for non-numeric paper_ids @@ -678,7 +701,10 @@ def get_user_library( # Execute and deduplicate in Python by paper.id (keeping latest feedback) all_results = session.execute(base_stmt).all() - Logger.info(f"Query returned {len(all_results)} results before deduplication", file=LogFiles.HARVEST) + Logger.info( + f"Query returned {len(all_results)} results before deduplication", + file=LogFiles.HARVEST, + ) # Deduplicate by paper.id, keeping the one with latest timestamp Logger.info("Deduplicating results by paper id", file=LogFiles.HARVEST) @@ -691,26 +717,38 @@ def get_user_library( # Convert to list and sort unique_results = list(paper_map.values()) - Logger.info(f"After deduplication: {len(unique_results)} unique papers", file=LogFiles.HARVEST) + Logger.info( + f"After deduplication: {len(unique_results)} unique papers", file=LogFiles.HARVEST + ) # Sort min_ts = datetime.min.replace(tzinfo=timezone.utc) if sort_by == "saved_at": - unique_results.sort(key=lambda x: x[1].ts or min_ts, reverse=(sort_order.lower() == "desc")) + unique_results.sort( + key=lambda x: x[1].ts or min_ts, reverse=(sort_order.lower() == "desc") + ) elif sort_by == "title": - unique_results.sort(key=lambda x: x[0].title or "", reverse=(sort_order.lower() == "desc")) + unique_results.sort( + key=lambda x: x[0].title or "", reverse=(sort_order.lower() == "desc") + ) elif sort_by == "citation_count": - unique_results.sort(key=lambda x: x[0].citation_count or 0, reverse=(sort_order.lower() == "desc")) + unique_results.sort( + key=lambda x: x[0].citation_count or 0, reverse=(sort_order.lower() == "desc") + ) elif sort_by == "year": - unique_results.sort(key=lambda x: x[0].year or 0, reverse=(sort_order.lower() == "desc")) + unique_results.sort( + key=lambda x: x[0].year or 0, reverse=(sort_order.lower() == "desc") + ) else: - unique_results.sort(key=lambda x: x[1].ts or min_ts, reverse=(sort_order.lower() == "desc")) + unique_results.sort( + key=lambda x: x[1].ts or min_ts, reverse=(sort_order.lower() == "desc") + ) # Get total count before pagination total = len(unique_results) # Apply pagination - paginated_results = unique_results[offset:offset + limit] + paginated_results = unique_results[offset : offset + limit] return [ LibraryPaper( @@ -725,12 +763,10 @@ def get_user_library( def remove_from_library(self, user_id: str, paper_id: int) -> bool: """Remove paper from user's library by deleting 'save' feedback.""" with self._provider.session() as session: - stmt = ( - PaperFeedbackModel.__table__.delete().where( - PaperFeedbackModel.user_id == user_id, - PaperFeedbackModel.paper_id == str(paper_id), - PaperFeedbackModel.action == "save", - ) + stmt = PaperFeedbackModel.__table__.delete().where( + PaperFeedbackModel.user_id == user_id, + PaperFeedbackModel.paper_id == str(paper_id), + PaperFeedbackModel.action == "save", ) result = session.execute(stmt) session.commit() @@ -830,9 +866,9 @@ def get_paper_count(self) -> int: with self._provider.session() as session: return ( session.execute( - select(func.count()).select_from(PaperModel).where( - PaperModel.deleted_at.is_(None) - ) + select(func.count()) + .select_from(PaperModel) + .where(PaperModel.deleted_at.is_(None)) ).scalar() or 0 ) @@ -845,6 +881,10 @@ def close(self) -> None: pass +# Backward-compatible alias: older workflows/tests import SqlAlchemyPaperStore. +SqlAlchemyPaperStore = PaperStore + + def paper_to_dict(paper: PaperModel) -> Dict[str, Any]: """Convert PaperModel to dictionary for API response.""" return { diff --git a/src/paperbot/infrastructure/stores/research_store.py b/src/paperbot/infrastructure/stores/research_store.py index 7e654ec1..db4572ef 100644 --- a/src/paperbot/infrastructure/stores/research_store.py +++ b/src/paperbot/infrastructure/stores/research_store.py @@ -9,18 +9,14 @@ from sqlalchemy.exc import IntegrityError from paperbot.domain.paper_identity import normalize_arxiv_id, normalize_doi - -from paperbot.domain.paper_identity import normalize_arxiv_id, normalize_doi - -from paperbot.utils.logging_config import Logger, LogFiles - from paperbot.infrastructure.stores.models import ( Base, PaperFeedbackModel, + PaperImpressionModel, PaperJudgeScoreModel, PaperModel, - PaperImpressionModel, PaperReadingStatusModel, + PaperRepoModel, ResearchContextRunModel, ResearchMilestoneModel, ResearchTaskModel, @@ -28,6 +24,7 @@ ResearchTrackModel, ) from paperbot.infrastructure.stores.sqlalchemy_db import SessionProvider, get_db_url +from paperbot.utils.logging_config import LogFiles, Logger def _utcnow() -> datetime: @@ -54,6 +51,32 @@ def _load_list(raw: str) -> List[str]: return [] +def _safe_int(value: Any, default: int = 0) -> int: + try: + return int(value) + except Exception: + return int(default) + + +def _parse_datetime(value: Any) -> Optional[datetime]: + if isinstance(value, datetime): + return value if value.tzinfo else value.replace(tzinfo=timezone.utc) + if not value: + return None + + text = str(value).strip() + if not text: + return None + + if text.endswith("Z"): + text = f"{text[:-1]}+00:00" + try: + parsed = datetime.fromisoformat(text) + return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) + except Exception: + return None + + class SqlAlchemyResearchStore: """ Track/progress store for personalized paper recommendation. @@ -347,7 +370,6 @@ def add_paper_feedback( Logger.error("Track not found", file=LogFiles.HARVEST) return None - resolved_paper_ref_id = self._resolve_paper_ref_id( session=session, paper_id=(paper_id or "").strip(), @@ -570,6 +592,90 @@ def list_saved_papers( return rows[: max(1, int(limit))] + def ingest_repo_enrichment_rows( + self, + *, + rows: List[Dict[str, Any]], + source: str = "paperscool_repo_enrich", + ) -> Dict[str, int]: + now = _utcnow() + created = 0 + updated = 0 + skipped = 0 + unresolved = 0 + + with self._provider.session() as session: + for raw in rows or []: + if not isinstance(raw, dict): + skipped += 1 + continue + + github = raw.get("github") if isinstance(raw.get("github"), dict) else {} + repo_url = str(raw.get("repo_url") or github.get("repo_url") or "").strip() + if not repo_url: + skipped += 1 + continue + + paper_meta = { + "title": raw.get("title"), + "paper_url": raw.get("paper_url"), + "url": raw.get("paper_url"), + } + paper_hint = str(raw.get("paper_id") or raw.get("paper_ref_id") or "").strip() + paper_ref_id = self._resolve_paper_ref_id( + session=session, + paper_id=paper_hint, + metadata=paper_meta, + ) + if not paper_ref_id: + unresolved += 1 + continue + + was_created = self._upsert_paper_repo_row( + session=session, + paper_ref_id=int(paper_ref_id), + repo_row=raw, + source=source, + now=now, + ) + if was_created is None: + skipped += 1 + elif was_created: + created += 1 + else: + updated += 1 + + session.commit() + + return { + "total": created + updated, + "created": created, + "updated": updated, + "skipped": skipped, + "unresolved_paper": unresolved, + } + + def list_paper_repos(self, *, paper_id: str) -> Optional[List[Dict[str, Any]]]: + with self._provider.session() as session: + paper_ref_id = self._resolve_paper_ref_id( + session=session, + paper_id=(paper_id or "").strip(), + metadata={}, + ) + if not paper_ref_id: + return None + + rows = ( + session.execute( + select(PaperRepoModel) + .where(PaperRepoModel.paper_id == int(paper_ref_id)) + .order_by(desc(PaperRepoModel.stars), desc(PaperRepoModel.synced_at)) + ) + .scalars() + .all() + ) + return [self._repo_to_dict(row) for row in rows] + def get_paper_detail( self, *, paper_id: str, user_id: str = "default" ) -> Optional[Dict[str, Any]]: @@ -619,6 +725,16 @@ def get_paper_detail( .all() ) + repo_rows = ( + session.execute( + select(PaperRepoModel) + .where(PaperRepoModel.paper_id == int(paper_ref_id)) + .order_by(desc(PaperRepoModel.stars), desc(PaperRepoModel.synced_at)) + ) + .scalars() + .all() + ) + feedback_summary: Dict[str, int] = {} for row in feedback_rows: action = str(row.action or "") @@ -635,6 +751,7 @@ def get_paper_detail( self._judge_score_to_dict(judge_scores[0]) if judge_scores else None ), "judge_scores": [self._judge_score_to_dict(row) for row in judge_scores], + "repos": [self._repo_to_dict(row) for row in repo_rows], "feedback_summary": feedback_summary, "feedback_rows": [self._feedback_to_dict(row) for row in feedback_rows], } @@ -975,28 +1092,43 @@ def _upsert_reading_status_row( @staticmethod def _paper_to_dict(p: PaperModel) -> Dict[str, Any]: + metadata_raw = getattr(p, "metadata_json", "{}") or "{}" try: - metadata = json.loads(p.metadata_json or "{}") + metadata = json.loads(metadata_raw) if not isinstance(metadata, dict): metadata = {} except Exception: metadata = {} + published_at = None + if getattr(p, "publication_date", None): + published_at = str(getattr(p, "publication_date")) + + source = getattr(p, "primary_source", None) or getattr(p, "source", "") + return { "id": int(p.id), "arxiv_id": p.arxiv_id, "doi": p.doi, + "semantic_scholar_id": getattr(p, "semantic_scholar_id", None), + "openalex_id": getattr(p, "openalex_id", None), "title": p.title, "authors": p.get_authors(), "abstract": p.abstract, "url": p.url, - "external_url": p.external_url, + "external_url": p.url, "pdf_url": p.pdf_url, - "source": p.source, + "source": source, + "primary_source": source, "venue": p.venue, - "published_at": p.published_at.isoformat() if p.published_at else None, - "first_seen_at": p.first_seen_at.isoformat() if p.first_seen_at else None, + "year": getattr(p, "year", None), + "publication_date": getattr(p, "publication_date", None), + "published_at": published_at, + "first_seen_at": p.created_at.isoformat() if p.created_at else None, "keywords": p.get_keywords(), + "fields_of_study": p.get_fields_of_study(), + "sources": p.get_sources(), + "citation_count": int(getattr(p, "citation_count", 0) or 0), "metadata": metadata, } @@ -1047,6 +1179,105 @@ def _reading_status_to_dict(row: PaperReadingStatusModel) -> Dict[str, Any]: "metadata": metadata, } + @staticmethod + def _repo_to_dict(row: PaperRepoModel) -> Dict[str, Any]: + try: + metadata = json.loads(row.metadata_json or "{}") + if not isinstance(metadata, dict): + metadata = {} + except Exception: + metadata = {} + + return { + "id": int(row.id), + "paper_id": int(row.paper_id), + "repo_url": row.repo_url, + "full_name": row.full_name, + "description": row.description, + "stars": int(row.stars or 0), + "forks": int(row.forks or 0), + "open_issues": int(row.open_issues or 0), + "watchers": int(row.watchers or 0), + "language": row.language, + "license": row.license, + "archived": bool(row.archived), + "html_url": row.html_url, + "topics": row.get_topics(), + "updated_at_remote": ( + row.updated_at_remote.isoformat() if row.updated_at_remote else None + ), + "pushed_at_remote": row.pushed_at_remote.isoformat() if row.pushed_at_remote else None, + "query": row.query, + "source": row.source, + "synced_at": row.synced_at.isoformat() if row.synced_at else None, + "metadata": metadata, + } + + def _upsert_paper_repo_row( + self, + *, + session, + paper_ref_id: int, + repo_row: Dict[str, Any], + source: str, + now: datetime, + ) -> Optional[bool]: + github = repo_row.get("github") if isinstance(repo_row.get("github"), dict) else {} + repo_url = str(repo_row.get("repo_url") or github.get("repo_url") or "").strip() + if not repo_url: + return None + + row = session.execute( + select(PaperRepoModel).where( + PaperRepoModel.paper_id == int(paper_ref_id), + PaperRepoModel.repo_url == repo_url, + ) + ).scalar_one_or_none() + created = row is None + if row is None: + row = PaperRepoModel( + paper_id=int(paper_ref_id), + repo_url=repo_url, + created_at=now, + updated_at=now, + synced_at=now, + ) + session.add(row) + + row.full_name = str(github.get("full_name") or repo_row.get("full_name") or "").strip() + row.description = str(github.get("description") or repo_row.get("description") or "") + row.stars = _safe_int(github.get("stars") or repo_row.get("stars"), 0) + row.forks = _safe_int(github.get("forks") or repo_row.get("forks"), 0) + row.open_issues = _safe_int(github.get("open_issues") or repo_row.get("open_issues"), 0) + row.watchers = _safe_int(github.get("watchers") or repo_row.get("watchers"), 0) + row.language = str(github.get("language") or repo_row.get("language") or "").strip() + row.license = str(github.get("license") or repo_row.get("license") or "").strip() + row.archived = bool(github.get("archived") or repo_row.get("archived")) + row.html_url = str(github.get("html_url") or repo_row.get("html_url") or repo_url).strip() + row.updated_at_remote = _parse_datetime( + github.get("updated_at") or repo_row.get("updated_at") + ) + row.pushed_at_remote = _parse_datetime(github.get("pushed_at") or repo_row.get("pushed_at")) + row.query = str(repo_row.get("query") or "").strip() + row.source = (str(source or "").strip() or "paperscool_repo_enrich")[:32] + + topics = github.get("topics") or repo_row.get("topics") or [] + if not isinstance(topics, list): + topics = [] + row.set_topics([str(v) for v in topics if str(v).strip()]) + + metadata = { + "title": repo_row.get("title"), + "paper_url": repo_row.get("paper_url"), + "github": github, + } + row.metadata_json = json.dumps(metadata, ensure_ascii=False) + row.synced_at = now + row.updated_at = now + + session.add(row) + return created + @staticmethod def _resolve_paper_ref_id( *, @@ -1055,8 +1286,6 @@ def _resolve_paper_ref_id( metadata: Dict[str, Any], ) -> Optional[int]: pid = (paper_id or "").strip() - if not pid: - return None if pid.isdigit(): row = session.execute( @@ -1065,8 +1294,8 @@ def _resolve_paper_ref_id( if row is not None: return int(row.id) - arxiv_id = normalize_arxiv_id(pid) - doi = normalize_doi(pid) + arxiv_id = normalize_arxiv_id(pid) if pid else None + doi = normalize_doi(pid) if pid else None url_candidates = [] for key in ("paper_url", "url", "external_url", "pdf_url"): diff --git a/tests/unit/test_paper_repo_persistence.py b/tests/unit/test_paper_repo_persistence.py new file mode 100644 index 00000000..72b34e5e --- /dev/null +++ b/tests/unit/test_paper_repo_persistence.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +from pathlib import Path + +from paperbot.infrastructure.stores.paper_store import SqlAlchemyPaperStore +from paperbot.infrastructure.stores.research_store import SqlAlchemyResearchStore + + +def test_ingest_and_list_paper_repos(tmp_path: Path): + db_path = tmp_path / "paper-repos.db" + db_url = f"sqlite:///{db_path}" + + paper_store = SqlAlchemyPaperStore(db_url=db_url) + research_store = SqlAlchemyResearchStore(db_url=db_url) + + paper = paper_store.upsert_paper( + paper={ + "title": "UniICL", + "url": "https://arxiv.org/abs/2501.12345", + "pdf_url": "https://arxiv.org/pdf/2501.12345.pdf", + } + ) + + first = research_store.ingest_repo_enrichment_rows( + rows=[ + { + "title": "UniICL", + "paper_url": "https://arxiv.org/abs/2501.12345", + "repo_url": "https://github.com/example/unicicl", + "query": "icl compression", + "github": { + "full_name": "example/unicicl", + "stars": 321, + "forks": 12, + "open_issues": 1, + "watchers": 18, + "language": "Python", + "license": "MIT", + "topics": ["icl", "llm"], + "html_url": "https://github.com/example/unicicl", + }, + } + ], + source="test_repo_enrich", + ) + + assert first["total"] == 1 + assert first["created"] == 1 + assert first["updated"] == 0 + + repos = research_store.list_paper_repos(paper_id=str(paper["id"])) + assert repos is not None + assert len(repos) == 1 + assert repos[0]["repo_url"] == "https://github.com/example/unicicl" + assert repos[0]["stars"] == 321 + assert repos[0]["source"] == "test_repo_enrich" + + second = research_store.ingest_repo_enrichment_rows( + rows=[ + { + "title": "UniICL", + "paper_url": "https://arxiv.org/abs/2501.12345", + "repo_url": "https://github.com/example/unicicl", + "github": {"full_name": "example/unicicl", "stars": 500}, + } + ], + source="test_repo_enrich", + ) + + assert second["total"] == 1 + assert second["created"] == 0 + assert second["updated"] == 1 + + updated_repos = research_store.list_paper_repos(paper_id=str(paper["id"])) + assert updated_repos is not None + assert updated_repos[0]["stars"] == 500 + + +def test_ingest_repo_rows_unresolved_paper(tmp_path: Path): + db_path = tmp_path / "paper-repos-missing.db" + db_url = f"sqlite:///{db_path}" + _ = SqlAlchemyPaperStore(db_url=db_url) + research_store = SqlAlchemyResearchStore(db_url=db_url) + + result = research_store.ingest_repo_enrichment_rows( + rows=[ + { + "title": "Unknown", + "paper_url": "https://arxiv.org/abs/9999.99999", + "repo_url": "https://github.com/example/unknown", + } + ] + ) + + assert result["total"] == 0 + assert result["unresolved_paper"] == 1 diff --git a/tests/unit/test_paperscool_route.py b/tests/unit/test_paperscool_route.py index c4f9fb8a..decf90d4 100644 --- a/tests/unit/test_paperscool_route.py +++ b/tests/unit/test_paperscool_route.py @@ -7,6 +7,7 @@ def _parse_sse_events(text: str): """Parse SSE text into a list of event dicts.""" import json + events = [] for line in text.split("\n"): if line.startswith("data: "): @@ -654,3 +655,87 @@ def test_dailypaper_sync_path_no_llm_no_judge(monkeypatch): assert payload["report"]["stats"]["unique_items"] == 1 # No filter block in sync path assert "filter" not in payload["report"] + + +def test_paperscool_repos_route_can_persist(monkeypatch): + class _FakeResp: + status_code = 200 + + def json(self): + return { + "full_name": "owner/repo", + "stargazers_count": 42, + "forks_count": 7, + "open_issues_count": 1, + "watchers_count": 5, + "language": "Python", + "license": {"spdx_id": "MIT"}, + "updated_at": "2026-02-01T00:00:00Z", + "pushed_at": "2026-02-02T00:00:00Z", + "archived": False, + "topics": ["llm"], + "html_url": "https://github.com/owner/repo", + } + + class _FakeStore: + def __init__(self): + self.rows = [] + + def ingest_repo_enrichment_rows(self, *, rows, source): + self.rows.extend(rows) + return { + "total": len(rows), + "created": len(rows), + "updated": 0, + "skipped": 0, + "unresolved_paper": 0, + } + + fake_store = _FakeStore() + monkeypatch.setattr(paperscool_route.requests, "get", lambda *args, **kwargs: _FakeResp()) + monkeypatch.setattr(paperscool_route, "SqlAlchemyResearchStore", lambda: fake_store) + + with TestClient(api_main.app) as client: + resp = client.post( + "/api/research/paperscool/repos", + json={ + "papers": [ + { + "title": "Repo Paper", + "url": "https://papers.cool/arxiv/1234", + "external_url": "https://github.com/owner/repo", + } + ], + "include_github_api": True, + "persist": True, + }, + ) + + assert resp.status_code == 200 + payload = resp.json() + assert payload["persist_summary"]["total"] == 1 + assert len(fake_store.rows) == 1 + + +def test_paperscool_daily_route_enqueues_repo_enrichment(monkeypatch): + monkeypatch.setattr(paperscool_route, "PapersCoolTopicSearchWorkflow", _FakeWorkflow) + + called = {"count": 0} + + def _fake_enqueue(report): + called["count"] += 1 + + monkeypatch.setattr(paperscool_route, "_enqueue_repo_enrichment_async", _fake_enqueue) + + with TestClient(api_main.app) as client: + resp = client.post( + "/api/research/paperscool/daily", + json={ + "queries": ["ICL压缩"], + "enable_llm_analysis": False, + "enable_judge": False, + }, + ) + + assert resp.status_code == 200 + assert called["count"] == 1 diff --git a/tests/unit/test_research_paper_registry_routes.py b/tests/unit/test_research_paper_registry_routes.py index daa5fd2a..5c65d7eb 100644 --- a/tests/unit/test_research_paper_registry_routes.py +++ b/tests/unit/test_research_paper_registry_routes.py @@ -32,6 +32,27 @@ def _prepare_db(tmp_path: Path): action="save", metadata={"title": "UniICL"}, ) + research_store.ingest_repo_enrichment_rows( + rows=[ + { + "title": "UniICL", + "paper_url": "https://arxiv.org/abs/2501.12345", + "repo_url": "https://github.com/example/unicicl", + "query": "icl compression", + "github": { + "full_name": "example/unicicl", + "stars": 321, + "forks": 12, + "open_issues": 1, + "watchers": 18, + "language": "Python", + "license": "MIT", + "topics": ["icl", "llm"], + "html_url": "https://github.com/example/unicicl", + }, + } + ] + ) return research_store, int(paper["id"]) @@ -50,6 +71,8 @@ def test_saved_and_detail_routes(tmp_path, monkeypatch): payload = detail.json()["detail"] assert payload["paper"]["id"] == paper_id assert payload["paper"]["title"] == "UniICL" + assert len(payload["repos"]) == 1 + assert payload["repos"][0]["repo_url"] == "https://github.com/example/unicicl" def test_update_status_route(tmp_path, monkeypatch): @@ -66,3 +89,17 @@ def test_update_status_route(tmp_path, monkeypatch): payload = resp.json()["status"] assert payload["paper_id"] == paper_id assert payload["status"] == "reading" + + +def test_paper_repos_route(tmp_path, monkeypatch): + store, paper_id = _prepare_db(tmp_path) + monkeypatch.setattr(research_route, "_research_store", store) + + with TestClient(api_main.app) as client: + resp = client.get(f"/api/research/papers/{paper_id}/repos") + + assert resp.status_code == 200 + payload = resp.json() + assert payload["paper_id"] == str(paper_id) + assert len(payload["repos"]) == 1 + assert payload["repos"][0]["full_name"] == "example/unicicl" From 8be54179c4a42d00b2b278541c0bc0000ebdc8e5 Mon Sep 17 00:00:00 2001 From: jerry <1772030600@qq.com> Date: Wed, 11 Feb 2026 22:21:29 +0800 Subject: [PATCH 3/8] docs: scope agent-browser to advanced workflows only --- docs/ROADMAP_TODO.md | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/ROADMAP_TODO.md b/docs/ROADMAP_TODO.md index 4ca97871..f7c90e45 100644 --- a/docs/ROADMAP_TODO.md +++ b/docs/ROADMAP_TODO.md @@ -213,6 +213,7 @@ > 参考仓库： > 目标：把"网页交互型"任务从静态 API 拉取升级为可观测、可回放的浏览器 Agent 流程。 +> 定位：**仅用于复杂/高交互采集与回归验证场景**，默认流程保持现有 API connector，不强制改造。 ### 4.1 Source Capture Agent（网页采集增强） From e5f19c08c2c86f2e7f09f635deefddb87e2f94cf Mon Sep 17 00:00:00 2001 From: jerry <1772030600@qq.com> Date: Wed, 11 Feb 2026 22:29:46 +0800 Subject: [PATCH 4/8] docs: add agent runtime unification todo and issue cards --- docs/PROJECT_ISSUE_BACKLOG.md | 62 +++++++++++++++++++++++++++++++++++ docs/ROADMAP_TODO.md | 48 +++++++++++++++++++++++++++ 2 files changed, 110 insertions(+) diff --git a/docs/PROJECT_ISSUE_BACKLOG.md b/docs/PROJECT_ISSUE_BACKLOG.md index befafc02..ebdd4742 100644 --- a/docs/PROJECT_ISSUE_BACKLOG.md +++ b/docs/PROJECT_ISSUE_BACKLOG.md @@ -2,6 +2,11 @@ This file splits unfinished items in `docs/ROADMAP_TODO.md` into issue-ready cards for GitHub Projects. +## Delivery Rule + +- One issue -> one commit (`1 issue = 1 commit`) +- Commit message include issue id, e.g. `feat: agent runtime contract (#41)` + ## Suggested Labels - `roadmap` @@ -354,3 +359,60 @@ Custom fields: - No plaintext secrets in repo/runtime logs - Browser automation runs are traceable and throttled +## Issue 21 - Agent Inventory and Boundary Map + +- Title: `[Chore] Build agent inventory and responsibility boundary map` +- Labels: `roadmap`, `phase-3`, `architecture`, `priority-p1` +- Source TODO: + - `盘点现有 Agent 入口与责任边界` + - `标记必须 Agent vs 普通 service` +- Scope: + - Inventory all agent entrypoints (routes/workflows) + - Document MUST-use-agent vs service-only decision rules + - Output `docs/agent_inventory.md` +- Acceptance: + - Every current agent entry has owner, input/output, SLA and fallback + +## Issue 22 - AgentRuntime Contract and SourceCollector Port + +- Title: `[Feature] Introduce unified AgentRuntime and SourceCollector contracts` +- Labels: `roadmap`, `phase-3`, `backend`, `architecture`, `priority-p1` +- Source TODO: + - `定义 AgentRuntime 接口` + - `定义 AgentMessage/AgentResult/AgentError schema` + - `定义 SourceCollector 接口` +- Scope: + - Add `core/abstractions/agent_runtime.py` + - Add `application/ports/source_collector.py` + - Add compatibility adapters for current agent classes +- Acceptance: + - Existing agent flows compile and run through adapter layer + +## Issue 23 - Unified Event Envelope for SSE + Trace Pipeline + +- Title: `[Feature] Unify SSE/event-log envelope across workflows and agents` +- Labels: `roadmap`, `phase-3`, `backend`, `frontend`, `observability`, `priority-p1` +- Source TODO: + - `SSE 事件统一` + - `trace_id 全链路贯穿` + - `前端统一 event parser` +- Scope: + - Define shared event envelope schema + - Map workflow events to the schema + - Add frontend parser utility and migrate key pages +- Acceptance: + - One parser handles Search/Daily/Analyze streams consistently + +## Issue 24 - AgentRuntime Migration Wave 1 (Analyze/Review) + +- Title: `[Refactor] Migrate analyze/review routes to AgentRuntime` +- Labels: `roadmap`, `phase-3`, `backend`, `priority-p1` +- Source TODO: + - `Step 1: analyze + review` +- Scope: + - Route-level migration for `analyze` and `review` + - Keep API contract backward-compatible + - Add regression tests for lifecycle events +- Acceptance: + - No behavior regressions in analyze/review, with unified runtime events + diff --git a/docs/ROADMAP_TODO.md b/docs/ROADMAP_TODO.md index f7c90e45..f4fb4892 100644 --- a/docs/ROADMAP_TODO.md +++ b/docs/ROADMAP_TODO.md @@ -256,6 +256,53 @@ - 避免批量采集触发封禁，支持 source 级并发控制 + +## Phase 3.8 — Agent Runtime 统一（接口契约） + +> 目标：把现有多套 Agent 调用方式收敛到统一契约，避免 API/Workflow 各写一套生命周期。 + +### 3.8.1 Agent 模块盘点（当前） + +- [ ] 盘点现有 Agent 入口与责任边界（形成清单） + - API 直接调用：`analyze/review/track/research/gen_code`（5 个入口） + - Agent 类：`src/paperbot/agents` 13 个业务 Agent（不含 Base），`src/paperbot/repro/agents` 4 个业务 Agent（不含 Base） + - 输出文档：`docs/agent_inventory.md` +- [ ] 标记哪些场景必须 Agent、哪些保持普通 service + - 必须 Agent：复杂推理、长链路、多步工具调用 + - 非必须：纯 CRUD、低延迟同步 API、简单规则处理 + +### 3.8.2 契约抽象（必须先做） + +- [ ] 定义统一 `AgentRuntime` 接口 + - 生命周期：`input -> plan -> execute -> emit events -> finalize` + - 文件：`src/paperbot/core/abstractions/agent_runtime.py` +- [ ] 定义统一 `AgentMessage` / `AgentResult` / `AgentError` schema + - 兼容当前 runbook/event_log 字段：`run_id`、`trace_id`、`stage`、`agent_name` +- [ ] 定义 `SourceCollector` 接口（API Source / Browser Source 同契约） + - 文件：`src/paperbot/application/ports/source_collector.py` + +### 3.8.3 观测统一（SSE + 日志 + 事件） + +- [ ] 把 workflow SSE 事件映射到统一事件总线 + - Search/DailyPaper/Analyze/Judge/Trend 都输出同一 event envelope +- [ ] 打通 `trace_id` 贯穿：API -> workflow -> agent -> store/event_log +- [ ] 前端统一消费事件协议（避免每个页面写一套 event parser） + +### 3.8.4 迁移顺序（按风险从低到高） + +- [ ] Step 1：`analyze` + `review` 接入 `AgentRuntime`（低耦合） +- [ ] Step 2：`track` + `research` 接入 `AgentRuntime`（中耦合） +- [ ] Step 3：`gen_code`（Paper2Code）迁移到统一 Runtime 适配层（高耦合） +- [ ] Step 4：为每步补回归测试 + runbook 对账测试 + +### 3.8.5 工程规则 + +- [ ] 执行策略：**每个 issue 对应 1 个 commit**（禁止跨 issue 混提） +- [ ] commit message 必须包含 issue 编号（如 `feat: xxx (#41)`） + +--- + + ## 多智能体系统现状与 OpenClaw 评估 ### 现有多智能体管线 @@ -577,3 +624,4 @@ OpenClaw Skill ── 独立，仅依赖 PaperBot REST API（已有） - 2026-02-11: 对齐远端 `origin/master` 的 Harvest 基线，保留旧实现到 `backup/feat-dailypaper-sse-stream-pre-harvest-20260211` - 2026-02-11: 新增 Phase 4（Agent Browser 自动化）任务清单，覆盖采集、E2E、对标监测、安全与限流 - 2026-02-11: 完成 Phase 1 收尾（PaperRepoModel + /papers/{paper_id}/repos + DailyPaper 异步 repo enrichment），并修复 harvest 基线下 paper store 兼容性 +- 2026-02-11: 新增 Phase 3.8 Agent Runtime 统一 TODO（契约/事件总线/迁移顺序），并明确 1 issue = 1 commit 规则 From 3efceb744c68016331c1f6a2cec0884098061025 Mon Sep 17 00:00:00 2001 From: jerry <1772030600@qq.com> Date: Wed, 11 Feb 2026 22:49:40 +0800 Subject: [PATCH 5/8] docs: add agent inventory and boundary map Closes #44 --- docs/ROADMAP_TODO.md | 5 ++- docs/agent_inventory.md | 89 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 92 insertions(+), 2 deletions(-) create mode 100644 docs/agent_inventory.md diff --git a/docs/ROADMAP_TODO.md b/docs/ROADMAP_TODO.md index f4fb4892..f1799a60 100644 --- a/docs/ROADMAP_TODO.md +++ b/docs/ROADMAP_TODO.md @@ -263,11 +263,11 @@ ### 3.8.1 Agent 模块盘点（当前） -- [ ] 盘点现有 Agent 入口与责任边界（形成清单） +- [x] 盘点现有 Agent 入口与责任边界（形成清单） - API 直接调用：`analyze/review/track/research/gen_code`（5 个入口） - Agent 类：`src/paperbot/agents` 13 个业务 Agent（不含 Base），`src/paperbot/repro/agents` 4 个业务 Agent（不含 Base） - 输出文档：`docs/agent_inventory.md` -- [ ] 标记哪些场景必须 Agent、哪些保持普通 service +- [x] 标记哪些场景必须 Agent、哪些保持普通 service - 必须 Agent：复杂推理、长链路、多步工具调用 - 非必须：纯 CRUD、低延迟同步 API、简单规则处理 @@ -625,3 +625,4 @@ OpenClaw Skill ── 独立，仅依赖 PaperBot REST API（已有） - 2026-02-11: 新增 Phase 4（Agent Browser 自动化）任务清单，覆盖采集、E2E、对标监测、安全与限流 - 2026-02-11: 完成 Phase 1 收尾（PaperRepoModel + /papers/{paper_id}/repos + DailyPaper 异步 repo enrichment），并修复 harvest 基线下 paper store 兼容性 - 2026-02-11: 新增 Phase 3.8 Agent Runtime 统一 TODO（契约/事件总线/迁移顺序），并明确 1 issue = 1 commit 规则 +- 2026-02-11: 完成 Issue #44（Agent inventory + 边界决策文档），新增 `docs/agent_inventory.md` diff --git a/docs/agent_inventory.md b/docs/agent_inventory.md new file mode 100644 index 00000000..5254ce68 --- /dev/null +++ b/docs/agent_inventory.md @@ -0,0 +1,89 @@ +# Agent Inventory and Boundary Map + +> Scope: current state on harvest-aligned baseline (`origin/master` + branch work) +> Date: 2026-02-11 + +## 1) Inventory Summary + +- API entrypoints that directly invoke agents: **5** + - `analyze`, `review`, `track`, `research` (scholar identity resolve), `gen_code` +- Agent classes: + - `src/paperbot/agents`: **13** business agents (excluding `BaseAgent`) + - `src/paperbot/repro/agents`: **4** business agents (excluding `BaseAgent`) +- Primary orchestration layers: + - `ScholarPipeline` -> `ScholarWorkflowCoordinator` (scholar analysis chain) + - `ReproAgent` -> `Orchestrator` (Paper2Code chain) + +## 2) Entry Boundary Matrix + +| Entrypoint | Owner | Agent Path | Input | Output | SLA (target) | Current Fallback | Boundary Decision | +|---|---|---|---|---|---|---|---| +| `POST /api/analyze` (`src/paperbot/api/routes/analyze.py`) | Research API | `ResearchAgent.analyze_paper` | title, abstract | SSE progress + paper analysis result | P50 < 15s, P95 < 45s | Emits SSE `error` event | **Must use agent** (unstructured paper reasoning) | +| `POST /api/review` (`src/paperbot/api/routes/review.py`) | Research API | `ReviewerAgent.review` | title, abstract | SSE progress + review decision payload | P50 < 20s, P95 < 60s | Emits SSE `error` event | **Must use agent** (deep critique + recommendation) | +| `GET /api/track` (`src/paperbot/api/routes/track.py`) | Scholar Tracking | `PaperTrackerAgent.track_scholar` + `ScholarPipeline.analyze_paper` | scholar id/name, flags | SSE run status + per-paper analysis | P50 < 60s for <=5 papers | Partial result or SSE `error`; closes S2 client in `finally` | **Must use agent** (multi-stage pipeline + external APIs) | +| `POST /api/research/scholar/network|trends` identity resolve (`src/paperbot/api/routes/research.py`) | Research API | `ScholarProfileAgent` (lookup only) | scholar_id or scholar_name | resolved scholar id then service stats | P50 < 2s lookup + downstream client time | HTTP 4xx/5xx | **Service preferred** (simple profile lookup; agent optional) | +| `POST /api/gen-code` (`src/paperbot/api/routes/gen_code.py`) | Paper2Code | `ReproAgent.reproduce_from_paper` -> `Orchestrator` (`Planning/Coding/Verification/Debugging` agents) | title, abstract, method section, flags | SSE progress + generated files/verification status | P50 < 180s (model/executor dependent) | Emits SSE `error`; returns partial generation when possible | **Must use agent** (planning + coding + repair loop) | + +## 3) Workflow-Level Agent Entry Points + +| Workflow Entry | Owner | Agent Chain | Notes | +|---|---|---|---| +| `ScholarPipeline.analyze_paper` (`src/paperbot/application/workflows/scholar_pipeline.py`) | Scholar Tracking | delegates to `ScholarWorkflowCoordinator` | Stable app-layer boundary; currently thin wrapper | +| `ScholarWorkflowCoordinator.run_paper_pipeline` (`src/paperbot/core/workflow_coordinator.py`) | Scholar Tracking | `ResearchAgent` -> `CodeAnalysisAgent` -> `QualityAgent` -> influence/report | Has Fail-Fast + ScoreShareBus integration | +| `ReproAgent.reproduce_from_paper` (`src/paperbot/repro/repro_agent.py`) | Paper2Code | node pipeline or orchestrator mode | Supports Docker/E2B executor choices | +| `Orchestrator.run` (`src/paperbot/repro/orchestrator.py`) | Paper2Code | `PlanningAgent` -> `CodingAgent` -> `VerificationAgent` -> `DebuggingAgent` loop | Emits event_log envelopes with run/trace IDs | + +## 4) Must-Use-Agent vs Service-Preferred Rules + +### Must-Use-Agent + +Use agents when all of the following are true: +- Task is primarily **unstructured reasoning** (summary/review/planning/repair) +- Requires **multi-step tool-calling** or iterative feedback loops +- Result quality benefits from chain-level context sharing (memory/score bus) +- Latency is acceptable in async/SSE workflows + +### Service-Preferred + +Use normal services (non-agent) when any of the following dominates: +- Task is deterministic CRUD/query/transform +- Strict low-latency sync API path is required +- Output schema is fixed and does not benefit from long reasoning +- Failure mode should be simple retries, not planner-style adaptation + +## 5) Known Gaps (for Issue #45-#47) + +- Lifecycle contracts are fragmented (route-level SSE events differ by feature) +- Some agent paths emit `trace_id`, others do not expose consistent envelopes +- Agent invocation style differs between `agents/*` and `repro/agents/*` +- No single `AgentRuntime` interface yet for plan/execute/finalize semantics + +## 6) Agent Class Inventory (Business Agents) + +`src/paperbot/agents`: +- `CodeAnalysisAgent` +- `ConferenceResearchAgent` +- `DeepResearchAgent` +- `DocumentationAgent` +- `HuggingFaceAgent` +- `OpenReviewAgent` +- `PaperTrackerAgent` +- `QualityAgent` +- `ResearchAgent` +- `ReviewerAgent` +- `ScholarProfileAgent` +- `SemanticScholarAgent` +- `VerificationAgent` + +`src/paperbot/repro/agents`: +- `PlanningAgent` +- `CodingAgent` +- `DebuggingAgent` +- `VerificationAgent` + +## 7) Ownership and Next Actions + +- Research API owner: normalize analyze/review event contracts before runtime migration +- Scholar Tracking owner: isolate profile lookup from agent dependency where possible +- Paper2Code owner: keep orchestrator stable while introducing runtime adapter +- Platform owner: introduce unified `AgentRuntime` + `SourceCollector` contracts in next issue From 432a088ccd30ffcc91a7255766068c354295733a Mon Sep 17 00:00:00 2001 From: jerry <1772030600@qq.com> Date: Wed, 11 Feb 2026 22:56:36 +0800 Subject: [PATCH 6/8] feat: add unified agent runtime and source collector contracts Closes #45 --- docs/ROADMAP_TODO.md | 7 +- src/paperbot/api/routes/gen_code.py | 59 +++-- src/paperbot/application/ports/__init__.py | 16 ++ .../application/ports/source_collector.py | 50 ++++ src/paperbot/core/abstractions/__init__.py | 17 +- .../core/abstractions/agent_runtime.py | 237 ++++++++++++++++++ tests/unit/test_agent_runtime_contract.py | 58 +++++ 7 files changed, 421 insertions(+), 23 deletions(-) create mode 100644 src/paperbot/application/ports/source_collector.py create mode 100644 src/paperbot/core/abstractions/agent_runtime.py create mode 100644 tests/unit/test_agent_runtime_contract.py diff --git a/docs/ROADMAP_TODO.md b/docs/ROADMAP_TODO.md index f1799a60..4b991deb 100644 --- a/docs/ROADMAP_TODO.md +++ b/docs/ROADMAP_TODO.md @@ -273,12 +273,12 @@ ### 3.8.2 契约抽象（必须先做） -- [ ] 定义统一 `AgentRuntime` 接口 +- [x] 定义统一 `AgentRuntime` 接口 - 生命周期：`input -> plan -> execute -> emit events -> finalize` - 文件：`src/paperbot/core/abstractions/agent_runtime.py` -- [ ] 定义统一 `AgentMessage` / `AgentResult` / `AgentError` schema +- [x] 定义统一 `AgentMessage` / `AgentResult` / `AgentError` schema - 兼容当前 runbook/event_log 字段：`run_id`、`trace_id`、`stage`、`agent_name` -- [ ] 定义 `SourceCollector` 接口（API Source / Browser Source 同契约） +- [x] 定义 `SourceCollector` 接口（API Source / Browser Source 同契约） - 文件：`src/paperbot/application/ports/source_collector.py` ### 3.8.3 观测统一（SSE + 日志 + 事件） @@ -626,3 +626,4 @@ OpenClaw Skill ── 独立，仅依赖 PaperBot REST API（已有） - 2026-02-11: 完成 Phase 1 收尾（PaperRepoModel + /papers/{paper_id}/repos + DailyPaper 异步 repo enrichment），并修复 harvest 基线下 paper store 兼容性 - 2026-02-11: 新增 Phase 3.8 Agent Runtime 统一 TODO（契约/事件总线/迁移顺序），并明确 1 issue = 1 commit 规则 - 2026-02-11: 完成 Issue #44（Agent inventory + 边界决策文档），新增 `docs/agent_inventory.md` +- 2026-02-11: 完成 Issue #45（AgentRuntime/SourceCollector 契约 + 兼容适配器 + contract tests） diff --git a/src/paperbot/api/routes/gen_code.py b/src/paperbot/api/routes/gen_code.py index a61d4987..8a7af0f4 100644 --- a/src/paperbot/api/routes/gen_code.py +++ b/src/paperbot/api/routes/gen_code.py @@ -2,15 +2,18 @@ Paper2Code Generation API Route """ +import tempfile +from pathlib import Path +from typing import Optional + from fastapi import APIRouter, Request from fastapi.responses import StreamingResponse from pydantic import BaseModel -from typing import Optional -from pathlib import Path -import tempfile -from ..streaming import StreamEvent, wrap_generator from paperbot.application.collaboration.message_schema import new_run_id, new_trace_id +from paperbot.core.abstractions import AgentRunContext + +from ..streaming import StreamEvent, wrap_generator router = APIRouter() @@ -24,7 +27,9 @@ class GenCodeRequest(BaseModel): output_dir: Optional[str] = None -async def gen_code_stream(request: GenCodeRequest, *, event_log=None, run_id: str = "", trace_id: str = ""): +async def gen_code_stream( + request: GenCodeRequest, *, event_log=None, run_id: str = "", trace_id: str = "" +): """Stream code generation progress""" try: if not run_id: @@ -32,18 +37,25 @@ async def gen_code_stream(request: GenCodeRequest, *, event_log=None, run_id: st if not trace_id: trace_id = new_trace_id() + runtime_context = AgentRunContext( + run_id=run_id, + trace_id=trace_id, + workflow="paper2code", + agent_name="ReproAgent", + ) + yield StreamEvent( type="progress", data={ "phase": "Initializing", "message": "Setting up code generation...", - "run_id": run_id, - "trace_id": trace_id, + "run_id": runtime_context.run_id, + "trace_id": runtime_context.trace_id, }, ) # Import repro modules - from ...repro import ReproAgent, PaperContext + from ...repro import PaperContext, ReproAgent # Create paper context paper_context = PaperContext( @@ -114,7 +126,7 @@ async def gen_code_stream(request: GenCodeRequest, *, event_log=None, run_id: st # Build result files = [] for filename, content in result.generated_files.items(): - lines = len(content.split('\n')) if content else 0 + lines = len(content.split("\n")) if content else 0 purpose = "Generated code" if "config" in filename.lower(): purpose = "Configuration" @@ -127,11 +139,13 @@ async def gen_code_stream(request: GenCodeRequest, *, event_log=None, run_id: st elif "main" in filename.lower(): purpose = "Entry point" - files.append({ - "name": filename, - "lines": lines, - "purpose": purpose, - }) + files.append( + { + "name": filename, + "lines": lines, + "purpose": purpose, + } + ) yield StreamEvent( type="result", @@ -140,12 +154,17 @@ async def gen_code_stream(request: GenCodeRequest, *, event_log=None, run_id: st "outputDir": str(output_dir), "files": files, "blueprint": { - "architectureType": result.blueprint.architecture_type if result.blueprint else "unknown", + "architectureType": ( + result.blueprint.architecture_type if result.blueprint else "unknown" + ), "domain": result.blueprint.domain if result.blueprint else "unknown", }, - "verificationPassed": len(result.verification_results) > 0 and all( - v.passed for v in result.verification_results - ) if result.verification_results else False, + "verificationPassed": ( + len(result.verification_results) > 0 + and all(v.passed for v in result.verification_results) + if result.verification_results + else False + ), }, ) @@ -164,7 +183,9 @@ async def generate_code(request: GenCodeRequest, http_request: Request): run_id = new_run_id() trace_id = new_trace_id() return StreamingResponse( - wrap_generator(gen_code_stream(request, event_log=event_log, run_id=run_id, trace_id=trace_id)), + wrap_generator( + gen_code_stream(request, event_log=event_log, run_id=run_id, trace_id=trace_id) + ), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", diff --git a/src/paperbot/application/ports/__init__.py b/src/paperbot/application/ports/__init__.py index b947dfb0..260f071d 100644 --- a/src/paperbot/application/ports/__init__.py +++ b/src/paperbot/application/ports/__init__.py @@ -1,3 +1,19 @@ """Application ports (interfaces) used by the application layer.""" +from .event_log_port import EventLogPort +from .harvester_port import HarvesterPort +from .source_collector import ( + NullSourceCollector, + SourceCollector, + SourceCollectRequest, + SourceCollectResult, +) +__all__ = [ + "EventLogPort", + "HarvesterPort", + "SourceCollector", + "SourceCollectRequest", + "SourceCollectResult", + "NullSourceCollector", +] diff --git a/src/paperbot/application/ports/source_collector.py b/src/paperbot/application/ports/source_collector.py new file mode 100644 index 00000000..9e8d2265 --- /dev/null +++ b/src/paperbot/application/ports/source_collector.py @@ -0,0 +1,50 @@ +"""Source collector contract shared by API and browser-based collectors.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Protocol, runtime_checkable + + +@dataclass +class SourceCollectRequest: + source: str + query: str + max_results: int = 100 + session: Dict[str, Any] = field(default_factory=dict) + strategy: Dict[str, Any] = field(default_factory=dict) + + +@dataclass +class SourceCollectResult: + source: str + items: List[Dict[str, Any]] = field(default_factory=list) + metadata: Dict[str, Any] = field(default_factory=dict) + trace: Dict[str, Any] = field(default_factory=dict) + + +@runtime_checkable +class SourceCollector(Protocol): + """Unified source collection interface. + + Implementations can be plain API connectors or browser-driven collectors. + """ + + async def collect(self, req: SourceCollectRequest) -> SourceCollectResult: ... + + async def close(self) -> None: ... + + +class NullSourceCollector: + """No-op collector for fallback paths and tests.""" + + async def collect(self, req: SourceCollectRequest) -> SourceCollectResult: + return SourceCollectResult( + source=req.source, + items=[], + metadata={"reason": "null_source_collector"}, + trace={"strategy": req.strategy}, + ) + + async def close(self) -> None: # pragma: no cover + return None diff --git a/src/paperbot/core/abstractions/__init__.py b/src/paperbot/core/abstractions/__init__.py index 8dc3c8f4..cf50699e 100644 --- a/src/paperbot/core/abstractions/__init__.py +++ b/src/paperbot/core/abstractions/__init__.py @@ -2,11 +2,26 @@ 核心抽象：统一 Agent/Node 的执行契约与结果结构。 """ +from .agent_runtime import ( + AgentError, + AgentMessage, + AgentResult, + AgentRunContext, + AgentRuntime, + BaseAgentRuntime, + LegacyMethodRuntime, +) from .executable import Executable, ExecutionResult, ensure_execution_result __all__ = [ + "AgentRuntime", + "AgentRunContext", + "AgentMessage", + "AgentResult", + "AgentError", + "BaseAgentRuntime", + "LegacyMethodRuntime", "Executable", "ExecutionResult", "ensure_execution_result", ] - diff --git a/src/paperbot/core/abstractions/agent_runtime.py b/src/paperbot/core/abstractions/agent_runtime.py new file mode 100644 index 00000000..ce9aaec4 --- /dev/null +++ b/src/paperbot/core/abstractions/agent_runtime.py @@ -0,0 +1,237 @@ +""" +Unified runtime contract for agent-like execution. + +This module introduces a stable lifecycle contract: +input -> plan -> execute -> emit events -> finalize + +It is intentionally framework-agnostic and can wrap existing legacy agents. +""" + +from __future__ import annotations + +import inspect +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from datetime import datetime, timezone +from typing import Any, Callable, Dict, Generic, Optional, Protocol, TypeVar + +TInput = TypeVar("TInput") +TOutput = TypeVar("TOutput") + + +@dataclass +class AgentRunContext: + """Execution context propagated across routes/workflows/runtime.""" + + run_id: str + trace_id: str + workflow: str + agent_name: str = "" + stage: str = "input" + metadata: Dict[str, Any] = field(default_factory=dict) + + def with_stage(self, stage: str) -> "AgentRunContext": + return AgentRunContext( + run_id=self.run_id, + trace_id=self.trace_id, + workflow=self.workflow, + agent_name=self.agent_name, + stage=stage, + metadata=dict(self.metadata), + ) + + +@dataclass +class AgentMessage: + """Unified runtime event payload.""" + + kind: str + context: AgentRunContext + payload: Dict[str, Any] = field(default_factory=dict) + ts: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat()) + + def to_dict(self) -> Dict[str, Any]: + return { + "kind": self.kind, + "run_id": self.context.run_id, + "trace_id": self.context.trace_id, + "workflow": self.context.workflow, + "agent_name": self.context.agent_name, + "stage": self.context.stage, + "payload": self.payload, + "ts": self.ts, + } + + +@dataclass +class AgentError(Exception): + """Structured runtime error.""" + + message: str + code: str = "agent_runtime_error" + retryable: bool = False + details: Dict[str, Any] = field(default_factory=dict) + + def __str__(self) -> str: + return self.message + + +@dataclass +class AgentResult(Generic[TOutput]): + """Standardized runtime result.""" + + ok: bool + output: Optional[TOutput] = None + error: Optional[AgentError] = None + metadata: Dict[str, Any] = field(default_factory=dict) + + @classmethod + def success(cls, output: TOutput, **metadata: Any) -> "AgentResult[TOutput]": + return cls(ok=True, output=output, metadata=metadata or {}) + + @classmethod + def failure( + cls, + message: str, + *, + code: str = "agent_runtime_error", + retryable: bool = False, + details: Optional[Dict[str, Any]] = None, + **metadata: Any, + ) -> "AgentResult[TOutput]": + return cls( + ok=False, + error=AgentError( + message=message, + code=code, + retryable=retryable, + details=details or {}, + ), + metadata=metadata or {}, + ) + + +EmitFn = Callable[[AgentMessage], None] + + +class AgentRuntime(Protocol, Generic[TInput, TOutput]): + """Unified runtime protocol for all agent executors.""" + + async def run( + self, + input_data: TInput, + *, + context: AgentRunContext, + emit: Optional[EmitFn] = None, + ) -> AgentResult[TOutput]: ... + + +class BaseAgentRuntime(ABC, Generic[TInput, TOutput]): + """Template implementation for agent runtime lifecycle.""" + + async def plan( + self, + input_data: TInput, + *, + context: AgentRunContext, + ) -> Dict[str, Any]: + return {} + + @abstractmethod + async def execute( + self, + input_data: TInput, + *, + context: AgentRunContext, + plan: Dict[str, Any], + ) -> TOutput: + raise NotImplementedError + + async def finalize( + self, + result: AgentResult[TOutput], + *, + context: AgentRunContext, + ) -> AgentResult[TOutput]: + return result + + async def run( + self, + input_data: TInput, + *, + context: AgentRunContext, + emit: Optional[EmitFn] = None, + ) -> AgentResult[TOutput]: + def _emit(kind: str, stage: str, payload: Optional[Dict[str, Any]] = None) -> None: + if emit is None: + return + emit( + AgentMessage( + kind=kind, + context=context.with_stage(stage), + payload=payload or {}, + ) + ) + + _emit("input", "input") + + try: + plan = await self.plan(input_data, context=context.with_stage("plan")) + _emit("plan", "plan", {"plan": plan}) + + output = await self.execute( + input_data, + context=context.with_stage("execute"), + plan=plan, + ) + result = AgentResult.success(output) + _emit("result", "result") + except AgentError as exc: + result = AgentResult(ok=False, error=exc) + _emit("error", "error", {"message": exc.message, "code": exc.code}) + except Exception as exc: # noqa: BLE001 + result = AgentResult.failure(str(exc)) + _emit("error", "error", {"message": str(exc), "code": "unhandled_exception"}) + + finalized = await self.finalize(result, context=context.with_stage("finalize")) + _emit("finalize", "finalize", {"ok": finalized.ok}) + return finalized + + +class LegacyMethodRuntime(BaseAgentRuntime[Dict[str, Any], Any]): + """Compatibility adapter for existing agent methods. + + Expected input format: + { + "args": [...], + "kwargs": {...} + } + """ + + def __init__(self, *, agent: Any, method_name: str): + self._agent = agent + self._method_name = method_name + + async def execute( + self, + input_data: Dict[str, Any], + *, + context: AgentRunContext, + plan: Dict[str, Any], + ) -> Any: + method = getattr(self._agent, self._method_name, None) + if method is None: + raise AgentError( + message=f"method '{self._method_name}' not found on legacy agent", + code="method_not_found", + ) + + args = input_data.get("args") if isinstance(input_data, dict) else None + kwargs = input_data.get("kwargs") if isinstance(input_data, dict) else None + args = args if isinstance(args, list) else [] + kwargs = kwargs if isinstance(kwargs, dict) else {} + + result = method(*args, **kwargs) + if inspect.isawaitable(result): + return await result + return result diff --git a/tests/unit/test_agent_runtime_contract.py b/tests/unit/test_agent_runtime_contract.py new file mode 100644 index 00000000..b549c55d --- /dev/null +++ b/tests/unit/test_agent_runtime_contract.py @@ -0,0 +1,58 @@ +from __future__ import annotations + +import pytest + +from paperbot.core.abstractions import AgentRunContext, BaseAgentRuntime, LegacyMethodRuntime + + +class _EchoRuntime(BaseAgentRuntime[dict, dict]): + async def plan(self, input_data: dict, *, context: AgentRunContext): + return {"planned": True, "keys": sorted(input_data.keys())} + + async def execute(self, input_data: dict, *, context: AgentRunContext, plan: dict): + return {"input": input_data, "plan": plan, "stage": context.stage} + + +class _LegacyAgent: + def sync_method(self, value: int) -> int: + return value + 1 + + async def async_method(self, value: int) -> int: + return value + 2 + + +@pytest.mark.asyncio +async def test_base_agent_runtime_run_emits_events(): + runtime = _EchoRuntime() + context = AgentRunContext(run_id="r1", trace_id="t1", workflow="wf", agent_name="Echo") + + events = [] + + result = await runtime.run({"a": 1}, context=context, emit=events.append) + + assert result.ok is True + assert result.output is not None + assert result.output["stage"] == "execute" + assert [e.kind for e in events] == ["input", "plan", "result", "finalize"] + + +@pytest.mark.asyncio +async def test_legacy_method_runtime_supports_sync_and_async_methods(): + legacy = _LegacyAgent() + context = AgentRunContext(run_id="r2", trace_id="t2", workflow="wf", agent_name="Legacy") + + sync_runtime = LegacyMethodRuntime(agent=legacy, method_name="sync_method") + sync_result = await sync_runtime.run( + {"args": [3], "kwargs": {}}, + context=context, + ) + assert sync_result.ok is True + assert sync_result.output == 4 + + async_runtime = LegacyMethodRuntime(agent=legacy, method_name="async_method") + async_result = await async_runtime.run( + {"args": [3], "kwargs": {}}, + context=context, + ) + assert async_result.ok is True + assert async_result.output == 5 From 7e3c7b9375a44457e04b6752512dfdfe042e8dc1 Mon Sep 17 00:00:00 2001 From: jerry <1772030600@qq.com> Date: Wed, 11 Feb 2026 23:02:51 +0800 Subject: [PATCH 7/8] feat: unify streaming envelope and trace propagation Closes #46 --- docs/ROADMAP_TODO.md | 7 +- src/paperbot/api/routes/analyze.py | 17 ++-- src/paperbot/api/routes/chat.py | 22 +++-- src/paperbot/api/routes/gen_code.py | 5 +- src/paperbot/api/routes/harvest.py | 12 +-- src/paperbot/api/routes/paperscool.py | 4 +- src/paperbot/api/routes/review.py | 16 ++-- src/paperbot/api/routes/sandbox.py | 34 ++++++-- src/paperbot/api/routes/track.py | 8 +- src/paperbot/api/streaming.py | 85 +++++++++++++++++-- tests/unit/test_streaming_envelope.py | 40 +++++++++ .../research/TopicWorkflowDashboard.tsx | 11 ++- web/src/lib/sse.ts | 51 ++++++++++- 13 files changed, 256 insertions(+), 56 deletions(-) create mode 100644 tests/unit/test_streaming_envelope.py diff --git a/docs/ROADMAP_TODO.md b/docs/ROADMAP_TODO.md index 4b991deb..a02de3a5 100644 --- a/docs/ROADMAP_TODO.md +++ b/docs/ROADMAP_TODO.md @@ -283,10 +283,10 @@ ### 3.8.3 观测统一（SSE + 日志 + 事件） -- [ ] 把 workflow SSE 事件映射到统一事件总线 +- [x] 把 workflow SSE 事件映射到统一事件总线 - Search/DailyPaper/Analyze/Judge/Trend 都输出同一 event envelope -- [ ] 打通 `trace_id` 贯穿：API -> workflow -> agent -> store/event_log -- [ ] 前端统一消费事件协议（避免每个页面写一套 event parser） +- [x] 打通 `trace_id` 贯穿：API -> workflow -> agent -> store/event_log +- [x] 前端统一消费事件协议（避免每个页面写一套 event parser） ### 3.8.4 迁移顺序（按风险从低到高） @@ -627,3 +627,4 @@ OpenClaw Skill ── 独立，仅依赖 PaperBot REST API（已有） - 2026-02-11: 新增 Phase 3.8 Agent Runtime 统一 TODO（契约/事件总线/迁移顺序），并明确 1 issue = 1 commit 规则 - 2026-02-11: 完成 Issue #44（Agent inventory + 边界决策文档），新增 `docs/agent_inventory.md` - 2026-02-11: 完成 Issue #45（AgentRuntime/SourceCollector 契约 + 兼容适配器 + contract tests） +- 2026-02-11: 完成 Issue #46（SSE envelope 统一 + trace_id 贯穿 + 前端 normalize parser） diff --git a/src/paperbot/api/routes/analyze.py b/src/paperbot/api/routes/analyze.py index 49209162..3e5a9a8f 100644 --- a/src/paperbot/api/routes/analyze.py +++ b/src/paperbot/api/routes/analyze.py @@ -2,10 +2,11 @@ Paper Analysis API Route """ +from typing import Optional + from fastapi import APIRouter from fastapi.responses import StreamingResponse from pydantic import BaseModel -from typing import Optional from ..streaming import StreamEvent, wrap_generator @@ -51,11 +52,13 @@ async def analyze_paper_stream(request: AnalyzeRequest): type="result", data={ "title": request.title, - "summary": result.summary if hasattr(result, 'summary') else "Analysis complete", - "keyContributions": result.key_contributions if hasattr(result, 'key_contributions') else [], - "methodology": result.methodology if hasattr(result, 'methodology') else "", - "strengths": result.strengths if hasattr(result, 'strengths') else [], - "weaknesses": result.weaknesses if hasattr(result, 'weaknesses') else [], + "summary": result.summary if hasattr(result, "summary") else "Analysis complete", + "keyContributions": ( + result.key_contributions if hasattr(result, "key_contributions") else [] + ), + "methodology": result.methodology if hasattr(result, "methodology") else "", + "strengths": result.strengths if hasattr(result, "strengths") else [], + "weaknesses": result.weaknesses if hasattr(result, "weaknesses") else [], }, ) @@ -71,7 +74,7 @@ async def analyze_paper(request: AnalyzeRequest): Returns Server-Sent Events with analysis updates. """ return StreamingResponse( - wrap_generator(analyze_paper_stream(request)), + wrap_generator(analyze_paper_stream(request), workflow="analyze"), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", diff --git a/src/paperbot/api/routes/chat.py b/src/paperbot/api/routes/chat.py index 1b40af26..6202dad4 100644 --- a/src/paperbot/api/routes/chat.py +++ b/src/paperbot/api/routes/chat.py @@ -2,10 +2,11 @@ Chat API Route - Interactive conversation with AI about papers """ +from typing import List, Optional + from fastapi import APIRouter from fastapi.responses import StreamingResponse from pydantic import BaseModel -from typing import List, Optional from ..streaming import StreamEvent, wrap_generator @@ -40,7 +41,9 @@ async def chat_stream(request: ChatRequest): # Build conversation messages = [ - {"role": "system", "content": """You are PaperBot, an AI assistant specialized in academic research. + { + "role": "system", + "content": """You are PaperBot, an AI assistant specialized in academic research. You help users: - Find and analyze research papers - Track scholars and their publications @@ -48,19 +51,24 @@ async def chat_stream(request: ChatRequest): - Generate code implementations from papers - Review papers for quality and novelty -Be concise and helpful. When discussing papers, cite specific details when available."""}, +Be concise and helpful. When discussing papers, cite specific details when available.""", + }, ] # Optional long-term memory augmentation (cross-platform). if request.use_memory and request.user_id: try: from paperbot.infrastructure.stores.memory_store import SqlAlchemyMemoryStore - from paperbot.memory.schema import MemoryCandidate from paperbot.memory.extractor import build_memory_context + from paperbot.memory.schema import MemoryCandidate store = SqlAlchemyMemoryStore() - items = store.search_memories(user_id=request.user_id, query=request.message, limit=8) - store.touch_usage(item_ids=[int(i["id"]) for i in items if i.get("id")], actor_id=request.user_id) + items = store.search_memories( + user_id=request.user_id, query=request.message, limit=8 + ) + store.touch_usage( + item_ids=[int(i["id"]) for i in items if i.get("id")], actor_id=request.user_id + ) cands = [ MemoryCandidate( kind=i.get("kind") or "fact", # type: ignore[arg-type] @@ -117,7 +125,7 @@ async def chat(request: ChatRequest): Returns Server-Sent Events with streaming text. """ return StreamingResponse( - wrap_generator(chat_stream(request)), + wrap_generator(chat_stream(request), workflow="chat"), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", diff --git a/src/paperbot/api/routes/gen_code.py b/src/paperbot/api/routes/gen_code.py index 8a7af0f4..a33d3ad4 100644 --- a/src/paperbot/api/routes/gen_code.py +++ b/src/paperbot/api/routes/gen_code.py @@ -184,7 +184,10 @@ async def generate_code(request: GenCodeRequest, http_request: Request): trace_id = new_trace_id() return StreamingResponse( wrap_generator( - gen_code_stream(request, event_log=event_log, run_id=run_id, trace_id=trace_id) + gen_code_stream(request, event_log=event_log, run_id=run_id, trace_id=trace_id), + workflow="gen_code", + run_id=run_id, + trace_id=trace_id, ), media_type="text/event-stream", headers={ diff --git a/src/paperbot/api/routes/harvest.py b/src/paperbot/api/routes/harvest.py index 491134cc..c0fe7388 100644 --- a/src/paperbot/api/routes/harvest.py +++ b/src/paperbot/api/routes/harvest.py @@ -24,8 +24,8 @@ HarvestPipeline, HarvestProgress, ) -from paperbot.utils.logging_config import Logger, LogFiles, set_trace_id, clear_trace_id from paperbot.infrastructure.stores.paper_store import PaperStore, paper_to_dict +from paperbot.utils.logging_config import LogFiles, Logger, clear_trace_id, set_trace_id router = APIRouter() @@ -64,9 +64,7 @@ class HarvestRequest(BaseModel): venues: Optional[List[str]] = Field(None, description="Filter to specific venues") year_from: Optional[int] = Field(None, ge=1900, le=2100, description="Start year") year_to: Optional[int] = Field(None, ge=1900, le=2100, description="End year") - max_results_per_source: int = Field( - 50, ge=1, le=200, description="Max papers per source" - ) + max_results_per_source: int = Field(50, ge=1, le=200, description="Max papers per source") sources: Optional[List[str]] = Field( None, description="Sources to harvest (arxiv, semantic_scholar, openalex)" ) @@ -129,7 +127,7 @@ async def harvest_papers(request: HarvestRequest): trace_id = set_trace_id() Logger.info(f"Starting harvest request: keywords={request.keywords}", file=LogFiles.HARVEST) return StreamingResponse( - wrap_generator(harvest_stream(request)), + wrap_generator(harvest_stream(request), workflow="harvest"), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", @@ -367,7 +365,9 @@ def get_user_library( offset=offset, ) - Logger.info(f"Retrieved {len(library_papers)} papers from library, total={total}", file=LogFiles.HARVEST) + Logger.info( + f"Retrieved {len(library_papers)} papers from library, total={total}", file=LogFiles.HARVEST + ) return LibraryResponse( papers=[ LibraryPaperResponse( diff --git a/src/paperbot/api/routes/paperscool.py b/src/paperbot/api/routes/paperscool.py index 7da23b62..d44f20bb 100644 --- a/src/paperbot/api/routes/paperscool.py +++ b/src/paperbot/api/routes/paperscool.py @@ -514,7 +514,7 @@ async def generate_daily_report(req: DailyPaperRequest): # SSE streaming path for long-running operations return StreamingResponse( - wrap_generator(_dailypaper_stream(req)), + wrap_generator(_dailypaper_stream(req), workflow="paperscool_daily"), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "Connection": "keep-alive"}, ) @@ -1006,7 +1006,7 @@ async def analyze_daily_report(req: PapersCoolAnalyzeRequest): raise HTTPException(status_code=400, detail="report with queries is required") return StreamingResponse( - wrap_generator(_paperscool_analyze_stream(req)), + wrap_generator(_paperscool_analyze_stream(req), workflow="paperscool_analyze"), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", diff --git a/src/paperbot/api/routes/review.py b/src/paperbot/api/routes/review.py index 6c59c5c8..c3123ef9 100644 --- a/src/paperbot/api/routes/review.py +++ b/src/paperbot/api/routes/review.py @@ -54,13 +54,15 @@ async def review_paper_stream(request: ReviewRequest): type="result", data={ "title": request.title, - "summary": result.summary if hasattr(result, 'summary') else "", - "keyContributions": result.contributions if hasattr(result, 'contributions') else [], + "summary": result.summary if hasattr(result, "summary") else "", + "keyContributions": ( + result.contributions if hasattr(result, "contributions") else [] + ), "methodology": "", - "strengths": result.strengths if hasattr(result, 'strengths') else [], - "weaknesses": result.weaknesses if hasattr(result, 'weaknesses') else [], - "noveltyScore": result.novelty_score if hasattr(result, 'novelty_score') else None, - "recommendation": result.decision if hasattr(result, 'decision') else None, + "strengths": result.strengths if hasattr(result, "strengths") else [], + "weaknesses": result.weaknesses if hasattr(result, "weaknesses") else [], + "noveltyScore": result.novelty_score if hasattr(result, "novelty_score") else None, + "recommendation": result.decision if hasattr(result, "decision") else None, }, ) @@ -76,7 +78,7 @@ async def review_paper(request: ReviewRequest): Returns Server-Sent Events with review updates. """ return StreamingResponse( - wrap_generator(review_paper_stream(request)), + wrap_generator(review_paper_stream(request), workflow="review"), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", diff --git a/src/paperbot/api/routes/sandbox.py b/src/paperbot/api/routes/sandbox.py index 91557fbd..e79769c8 100644 --- a/src/paperbot/api/routes/sandbox.py +++ b/src/paperbot/api/routes/sandbox.py @@ -12,7 +12,8 @@ from __future__ import annotations from typing import Any, Dict, List, Optional -from fastapi import APIRouter, Query, Request, HTTPException + +from fastapi import APIRouter, HTTPException, Query, Request from fastapi.responses import StreamingResponse from pydantic import BaseModel @@ -23,8 +24,10 @@ # --- Request/Response Models --- + class SubmitRequest(BaseModel): """Request body for job submission""" + type: str = "paper2code" paper_url: Optional[str] = None paper_id: Optional[str] = None @@ -34,6 +37,7 @@ class SubmitRequest(BaseModel): class CancelResponse(BaseModel): """Response for job cancellation""" + status: str job_id: str message: str = "" @@ -41,6 +45,7 @@ class CancelResponse(BaseModel): class RetryResponse(BaseModel): """Response for job retry""" + status: str old_job_id: str new_job_id: Optional[str] = None @@ -49,6 +54,7 @@ class RetryResponse(BaseModel): # --- Queue Management --- + @router.get("/sandbox/queue") async def get_queue_status( http_request: Request, @@ -70,7 +76,13 @@ async def get_queue_status( finally: await manager.close() except ImportError: - return {"error": "Redis/ARQ not configured", "pending": [], "running": [], "completed": [], "stats": {}} + return { + "error": "Redis/ARQ not configured", + "pending": [], + "running": [], + "completed": [], + "stats": {}, + } except Exception as e: return {"error": str(e), "pending": [], "running": [], "completed": [], "stats": {}} @@ -111,9 +123,15 @@ async def cancel_job(job_id: str, http_request: Request) -> CancelResponse: try: success = await manager.cancel_job(job_id) if success: - return CancelResponse(status="cancelled", job_id=job_id, message="Job cancelled successfully") + return CancelResponse( + status="cancelled", job_id=job_id, message="Job cancelled successfully" + ) else: - return CancelResponse(status="failed", job_id=job_id, message="Cannot cancel job (may be running or completed)") + return CancelResponse( + status="failed", + job_id=job_id, + message="Cannot cancel job (may be running or completed)", + ) finally: await manager.close() except Exception as e: @@ -151,6 +169,7 @@ async def retry_job(job_id: str, http_request: Request) -> RetryResponse: # --- Job Submission --- + @router.post("/sandbox/submit") async def submit_job(body: SubmitRequest, http_request: Request): """ @@ -189,6 +208,7 @@ async def submit_job(body: SubmitRequest, http_request: Request): # --- Log Streaming --- + async def _log_stream_generator(run_id: str): """Generate SSE events for log streaming.""" from paperbot.infrastructure.logging.execution_logger import get_execution_logger @@ -212,7 +232,7 @@ async def stream_logs(run_id: str, http_request: Request): Returns Server-Sent Events with log entries. """ return StreamingResponse( - wrap_generator(_log_stream_generator(run_id)), + wrap_generator(_log_stream_generator(run_id), workflow="sandbox_logs"), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", @@ -242,6 +262,7 @@ async def get_logs( # --- Resource Metrics --- + async def _metrics_stream_generator(run_id: str): """Generate SSE events for metrics streaming.""" from paperbot.infrastructure.monitoring.resource_monitor import get_resource_monitor @@ -265,7 +286,7 @@ async def stream_metrics(run_id: str, http_request: Request): Returns Server-Sent Events with CPU/memory metrics. """ return StreamingResponse( - wrap_generator(_metrics_stream_generator(run_id)), + wrap_generator(_metrics_stream_generator(run_id), workflow="sandbox_metrics"), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", @@ -313,6 +334,7 @@ async def get_metrics_history( # --- System Status --- + @router.get("/sandbox/status") async def get_system_status(http_request: Request): """ diff --git a/src/paperbot/api/routes/track.py b/src/paperbot/api/routes/track.py index 90e35698..e54fc18e 100644 --- a/src/paperbot/api/routes/track.py +++ b/src/paperbot/api/routes/track.py @@ -2,9 +2,10 @@ Scholar Tracking API Route """ +from typing import Any, Dict, List, Optional + from fastapi import APIRouter, Query, Request from fastapi.responses import StreamingResponse -from typing import Optional, Dict, Any, List from ..streaming import StreamEvent, wrap_generator @@ -23,9 +24,9 @@ async def track_scholar_stream( ): """Stream scholar tracking progress""" # Imports here to avoid circular imports and reduce cold-start cost. + from paperbot.agents.scholar_tracking.paper_tracker_agent import PaperTrackerAgent from paperbot.application.collaboration.message_schema import new_run_id, new_trace_id from paperbot.application.workflows.scholar_pipeline import ScholarPipeline - from paperbot.agents.scholar_tracking.paper_tracker_agent import PaperTrackerAgent from paperbot.domain.paper import PaperMeta run_id = new_run_id() @@ -218,7 +219,8 @@ async def track_scholar( max_new_papers=max_new_papers, persist_report=persist_report, offline=offline, - ) + ), + workflow="track", ), media_type="text/event-stream", headers={ diff --git a/src/paperbot/api/streaming.py b/src/paperbot/api/streaming.py index 58cd99bd..2c870637 100644 --- a/src/paperbot/api/streaming.py +++ b/src/paperbot/api/streaming.py @@ -1,42 +1,109 @@ """ -Streaming utilities for Server-Sent Events (SSE) +Streaming utilities for Server-Sent Events (SSE). + +Provides a normalized envelope for stream observability: +- workflow +- run_id +- trace_id +- seq +- phase +- ts """ +from __future__ import annotations + import json -from typing import Any, AsyncGenerator, Optional from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any, AsyncGenerator, Dict, Optional +from uuid import uuid4 + + +def _new_stream_id(prefix: str) -> str: + return f"{prefix}_{uuid4().hex[:12]}" @dataclass class StreamEvent: - """SSE event structure""" + """SSE event structure.""" + type: str # progress, result, error, done data: Any = None message: Optional[str] = None + envelope: Optional[Dict[str, Any]] = None def to_sse(self) -> str: - """Convert to SSE format""" + """Convert to SSE format.""" payload = { "type": self.type, "data": self.data, "message": self.message, + "envelope": self.envelope, } return f"data: {json.dumps(payload)}\n\n" def sse_done() -> str: - """Return SSE done signal""" + """Return SSE done signal.""" return "data: [DONE]\n\n" +def _with_envelope( + event: StreamEvent, + *, + workflow: str, + run_id: str, + trace_id: str, + seq: int, +) -> StreamEvent: + if event.envelope: + return event + + phase = None + if isinstance(event.data, dict): + phase = event.data.get("phase") + + event.envelope = { + "workflow": workflow or "unknown", + "run_id": run_id, + "trace_id": trace_id, + "seq": seq, + "phase": phase, + "ts": datetime.now(timezone.utc).isoformat(), + } + return event + + async def wrap_generator( - generator: AsyncGenerator[StreamEvent, None] + generator: AsyncGenerator[StreamEvent, None], + *, + workflow: str = "", + run_id: Optional[str] = None, + trace_id: Optional[str] = None, ) -> AsyncGenerator[str, None]: - """Wrap a StreamEvent generator to SSE strings""" + """Wrap a StreamEvent generator to SSE strings with a normalized envelope.""" + resolved_run_id = run_id or _new_stream_id("run") + resolved_trace_id = trace_id or _new_stream_id("trace") + seq = 0 + try: async for event in generator: - yield event.to_sse() + seq += 1 + yield _with_envelope( + event, + workflow=workflow, + run_id=resolved_run_id, + trace_id=resolved_trace_id, + seq=seq, + ).to_sse() yield sse_done() except Exception as e: - yield StreamEvent(type="error", message=str(e)).to_sse() + seq += 1 + yield _with_envelope( + StreamEvent(type="error", message=str(e)), + workflow=workflow, + run_id=resolved_run_id, + trace_id=resolved_trace_id, + seq=seq, + ).to_sse() yield sse_done() diff --git a/tests/unit/test_streaming_envelope.py b/tests/unit/test_streaming_envelope.py new file mode 100644 index 00000000..3d498168 --- /dev/null +++ b/tests/unit/test_streaming_envelope.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +import json + +import pytest + +from paperbot.api.streaming import StreamEvent, wrap_generator + + +async def _simple_stream(): + yield StreamEvent(type="progress", data={"phase": "judge", "message": "running"}) + yield StreamEvent(type="result", data={"ok": True}) + + +@pytest.mark.asyncio +async def test_wrap_generator_injects_envelope(): + payloads = [] + async for raw in wrap_generator( + _simple_stream(), + workflow="paperscool_analyze", + run_id="run_x", + trace_id="trace_x", + ): + if not raw.startswith("data: "): + continue + data = raw.removeprefix("data: ").strip() + if data == "[DONE]": + continue + payloads.append(json.loads(data)) + + assert len(payloads) == 2 + for idx, payload in enumerate(payloads, start=1): + env = payload["envelope"] + assert env["workflow"] == "paperscool_analyze" + assert env["run_id"] == "run_x" + assert env["trace_id"] == "trace_x" + assert env["seq"] == idx + assert isinstance(env["ts"], str) + + assert payloads[0]["envelope"]["phase"] == "judge" diff --git a/web/src/components/research/TopicWorkflowDashboard.tsx b/web/src/components/research/TopicWorkflowDashboard.tsx index b030bcbc..5a2de3c2 100644 --- a/web/src/components/research/TopicWorkflowDashboard.tsx +++ b/web/src/components/research/TopicWorkflowDashboard.tsx @@ -47,7 +47,7 @@ import { SheetTrigger, } from "@/components/ui/sheet" import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs" -import { readSSE } from "@/lib/sse" +import { normalizeSSEMessage, readSSE } from "@/lib/sse" import { useWorkflowStore } from "@/lib/stores/workflow-store" import type { DailyResult, WorkflowPhase } from "@/lib/stores/workflow-store" @@ -930,7 +930,8 @@ export default function TopicWorkflowDashboard() { // SSE streaming path if (!res.body) throw new Error("No response body for SSE stream") - for await (const event of readSSE(res.body)) { + for await (const rawEvent of readSSE(res.body)) { + const event = normalizeSSEMessage(rawEvent, "paperscool_daily") if (event.type === "progress") { const d = (event.data || {}) as { phase?: string; message?: string; total?: number } const p = (d.phase || "search") as StreamPhase @@ -1182,10 +1183,12 @@ export default function TopicWorkflowDashboard() { }) if (!res.ok || !res.body) throw new Error(await res.text()) - for await (const event of readSSE(res.body)) { + for await (const rawEvent of readSSE(res.body)) { + const event = normalizeSSEMessage(rawEvent, "paperscool_analyze") if (event.type === "progress") { const d = (event.data || {}) as { phase?: string; message?: string; total?: number } - store.addAnalyzeLog(`[${d.phase || "step"}] ${d.message || "running"}`) + const trace = event.envelope.trace_id ? ` trace=${event.envelope.trace_id}` : "" + store.addAnalyzeLog(`[${d.phase || "step"}] ${d.message || "running"}${trace}`) if (d.phase === "judge" && (d.total || 0) > 0) { setAnalyzeProgress({ done: 0, total: d.total || 0 }) } diff --git a/web/src/lib/sse.ts b/web/src/lib/sse.ts index b76a6ca7..9b5ecc29 100644 --- a/web/src/lib/sse.ts +++ b/web/src/lib/sse.ts @@ -1,7 +1,57 @@ +export type StreamEnvelope = { + workflow?: string + run_id?: string + trace_id?: string + seq?: number + phase?: string | null + ts?: string +} + export type SSEMessage = { type?: string data?: unknown message?: string | null + envelope?: StreamEnvelope | null +} + +export type NormalizedSSEEvent = { + type: string + data: unknown + message: string | null + envelope: StreamEnvelope +} + +function asEnvelope(raw: unknown): StreamEnvelope { + if (!raw || typeof raw !== "object") return {} + const obj = raw as Record + return { + workflow: typeof obj.workflow === "string" ? obj.workflow : undefined, + run_id: typeof obj.run_id === "string" ? obj.run_id : undefined, + trace_id: typeof obj.trace_id === "string" ? obj.trace_id : undefined, + seq: typeof obj.seq === "number" ? obj.seq : undefined, + phase: typeof obj.phase === "string" ? obj.phase : null, + ts: typeof obj.ts === "string" ? obj.ts : undefined, + } +} + +export function normalizeSSEMessage(message: SSEMessage, fallbackWorkflow = "unknown"): NormalizedSSEEvent { + const dataObj = message.data && typeof message.data === "object" ? (message.data as Record) : null + const envelope = asEnvelope(message.envelope) + const derivedPhase = typeof dataObj?.phase === "string" ? dataObj.phase : envelope.phase || null + + return { + type: typeof message.type === "string" && message.type.length > 0 ? message.type : "unknown", + data: message.data, + message: typeof message.message === "string" ? message.message : null, + envelope: { + workflow: envelope.workflow || fallbackWorkflow, + run_id: envelope.run_id, + trace_id: envelope.trace_id, + seq: envelope.seq, + phase: derivedPhase, + ts: envelope.ts, + }, + } } export async function* readSSE(stream: ReadableStream): AsyncGenerator { @@ -36,4 +86,3 @@ export async function* readSSE(stream: ReadableStream): AsyncGenerat } } } - From 58aa6e5507e504b54fa8e8a05115b02b6a11e76e Mon Sep 17 00:00:00 2001 From: jerry <1772030600@qq.com> Date: Wed, 11 Feb 2026 23:06:17 +0800 Subject: [PATCH 8/8] refactor: migrate analyze and review routes to agent runtime Closes #47 --- docs/ROADMAP_TODO.md | 3 +- src/paperbot/api/routes/analyze.py | 72 ++++++++++-- src/paperbot/api/routes/review.py | 79 ++++++++++--- .../test_analyze_review_runtime_routes.py | 108 ++++++++++++++++++ 4 files changed, 236 insertions(+), 26 deletions(-) create mode 100644 tests/unit/test_analyze_review_runtime_routes.py diff --git a/docs/ROADMAP_TODO.md b/docs/ROADMAP_TODO.md index a02de3a5..614587d2 100644 --- a/docs/ROADMAP_TODO.md +++ b/docs/ROADMAP_TODO.md @@ -290,7 +290,7 @@ ### 3.8.4 迁移顺序（按风险从低到高） -- [ ] Step 1：`analyze` + `review` 接入 `AgentRuntime`（低耦合） +- [x] Step 1：`analyze` + `review` 接入 `AgentRuntime`（低耦合） - [ ] Step 2：`track` + `research` 接入 `AgentRuntime`（中耦合） - [ ] Step 3：`gen_code`（Paper2Code）迁移到统一 Runtime 适配层（高耦合） - [ ] Step 4：为每步补回归测试 + runbook 对账测试 @@ -628,3 +628,4 @@ OpenClaw Skill ── 独立，仅依赖 PaperBot REST API（已有） - 2026-02-11: 完成 Issue #44（Agent inventory + 边界决策文档），新增 `docs/agent_inventory.md` - 2026-02-11: 完成 Issue #45（AgentRuntime/SourceCollector 契约 + 兼容适配器 + contract tests） - 2026-02-11: 完成 Issue #46（SSE envelope 统一 + trace_id 贯穿 + 前端 normalize parser） +- 2026-02-11: 完成 Issue #47（analyze/review 路由迁移到 AgentRuntime，保持 SSE 兼容） diff --git a/src/paperbot/api/routes/analyze.py b/src/paperbot/api/routes/analyze.py index 3e5a9a8f..44c70e88 100644 --- a/src/paperbot/api/routes/analyze.py +++ b/src/paperbot/api/routes/analyze.py @@ -8,6 +8,9 @@ from fastapi.responses import StreamingResponse from pydantic import BaseModel +from paperbot.application.collaboration.message_schema import new_run_id, new_trace_id +from paperbot.core.abstractions import AgentRunContext, LegacyMethodRuntime + from ..streaming import StreamEvent, wrap_generator router = APIRouter() @@ -19,33 +22,65 @@ class AnalyzeRequest(BaseModel): doi: Optional[str] = None -async def analyze_paper_stream(request: AnalyzeRequest): - """Stream paper analysis progress""" +async def analyze_paper_stream(request: AnalyzeRequest, *, run_id: str, trace_id: str): + """Stream paper analysis progress via AgentRuntime contract.""" try: yield StreamEvent( type="progress", - data={"phase": "Fetching", "message": "Retrieving paper information..."}, + data={ + "phase": "Fetching", + "message": "Retrieving paper information...", + "run_id": run_id, + "trace_id": trace_id, + }, ) - # Import agents from ...agents.research import ResearchAgent agent = ResearchAgent({}) + runtime = LegacyMethodRuntime(agent=agent, method_name="analyze_paper") + runtime_context = AgentRunContext( + run_id=run_id, + trace_id=trace_id, + workflow="analyze", + agent_name="ResearchAgent", + ) yield StreamEvent( type="progress", - data={"phase": "Analyzing", "message": "Extracting key contributions..."}, + data={ + "phase": "Analyzing", + "message": "Extracting key contributions...", + "run_id": run_id, + "trace_id": trace_id, + }, ) - # Analyze paper - result = await agent.analyze_paper( - title=request.title, - abstract=request.abstract or "", + runtime_result = await runtime.run( + { + "args": [], + "kwargs": { + "title": request.title, + "abstract": request.abstract or "", + }, + }, + context=runtime_context, ) + if not runtime_result.ok: + message = runtime_result.error.message if runtime_result.error else "Analysis failed" + raise RuntimeError(message) + + result = runtime_result.output + yield StreamEvent( type="progress", - data={"phase": "Summarizing", "message": "Generating summary..."}, + data={ + "phase": "Summarizing", + "message": "Generating summary...", + "run_id": run_id, + "trace_id": trace_id, + }, ) yield StreamEvent( @@ -59,11 +94,17 @@ async def analyze_paper_stream(request: AnalyzeRequest): "methodology": result.methodology if hasattr(result, "methodology") else "", "strengths": result.strengths if hasattr(result, "strengths") else [], "weaknesses": result.weaknesses if hasattr(result, "weaknesses") else [], + "run_id": run_id, + "trace_id": trace_id, }, ) except Exception as e: - yield StreamEvent(type="error", message=str(e)) + yield StreamEvent( + type="error", + message=str(e), + data={"run_id": run_id, "trace_id": trace_id}, + ) @router.post("/analyze") @@ -73,8 +114,15 @@ async def analyze_paper(request: AnalyzeRequest): Returns Server-Sent Events with analysis updates. """ + run_id = new_run_id() + trace_id = new_trace_id() return StreamingResponse( - wrap_generator(analyze_paper_stream(request), workflow="analyze"), + wrap_generator( + analyze_paper_stream(request, run_id=run_id, trace_id=trace_id), + workflow="analyze", + run_id=run_id, + trace_id=trace_id, + ), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", diff --git a/src/paperbot/api/routes/review.py b/src/paperbot/api/routes/review.py index c3123ef9..7ad1f1e3 100644 --- a/src/paperbot/api/routes/review.py +++ b/src/paperbot/api/routes/review.py @@ -6,6 +6,9 @@ from fastapi.responses import StreamingResponse from pydantic import BaseModel +from paperbot.application.collaboration.message_schema import new_run_id, new_trace_id +from paperbot.core.abstractions import AgentRunContext, LegacyMethodRuntime + from ..streaming import StreamEvent, wrap_generator router = APIRouter() @@ -16,38 +19,75 @@ class ReviewRequest(BaseModel): abstract: str -async def review_paper_stream(request: ReviewRequest): - """Stream deep review progress""" +async def review_paper_stream(request: ReviewRequest, *, run_id: str, trace_id: str): + """Stream deep review progress via AgentRuntime contract.""" try: yield StreamEvent( type="progress", - data={"phase": "Initializing", "message": "Starting deep review..."}, + data={ + "phase": "Initializing", + "message": "Starting deep review...", + "run_id": run_id, + "trace_id": trace_id, + }, ) - # Import reviewer agent from ...agents.review import ReviewerAgent agent = ReviewerAgent({}) + runtime = LegacyMethodRuntime(agent=agent, method_name="review") + runtime_context = AgentRunContext( + run_id=run_id, + trace_id=trace_id, + workflow="review", + agent_name="ReviewerAgent", + ) yield StreamEvent( type="progress", - data={"phase": "Screening", "message": "Initial screening..."}, + data={ + "phase": "Screening", + "message": "Initial screening...", + "run_id": run_id, + "trace_id": trace_id, + }, ) yield StreamEvent( type="progress", - data={"phase": "Critiquing", "message": "Deep critique analysis..."}, + data={ + "phase": "Critiquing", + "message": "Deep critique analysis...", + "run_id": run_id, + "trace_id": trace_id, + }, ) - # Run review - result = await agent.review( - title=request.title, - abstract=request.abstract, + runtime_result = await runtime.run( + { + "args": [], + "kwargs": { + "title": request.title, + "abstract": request.abstract, + }, + }, + context=runtime_context, ) + if not runtime_result.ok: + message = runtime_result.error.message if runtime_result.error else "Review failed" + raise RuntimeError(message) + + result = runtime_result.output + yield StreamEvent( type="progress", - data={"phase": "Decision", "message": "Generating recommendation..."}, + data={ + "phase": "Decision", + "message": "Generating recommendation...", + "run_id": run_id, + "trace_id": trace_id, + }, ) yield StreamEvent( @@ -63,11 +103,17 @@ async def review_paper_stream(request: ReviewRequest): "weaknesses": result.weaknesses if hasattr(result, "weaknesses") else [], "noveltyScore": result.novelty_score if hasattr(result, "novelty_score") else None, "recommendation": result.decision if hasattr(result, "decision") else None, + "run_id": run_id, + "trace_id": trace_id, }, ) except Exception as e: - yield StreamEvent(type="error", message=str(e)) + yield StreamEvent( + type="error", + message=str(e), + data={"run_id": run_id, "trace_id": trace_id}, + ) @router.post("/review") @@ -77,8 +123,15 @@ async def review_paper(request: ReviewRequest): Returns Server-Sent Events with review updates. """ + run_id = new_run_id() + trace_id = new_trace_id() return StreamingResponse( - wrap_generator(review_paper_stream(request), workflow="review"), + wrap_generator( + review_paper_stream(request, run_id=run_id, trace_id=trace_id), + workflow="review", + run_id=run_id, + trace_id=trace_id, + ), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", diff --git a/tests/unit/test_analyze_review_runtime_routes.py b/tests/unit/test_analyze_review_runtime_routes.py new file mode 100644 index 00000000..58831dea --- /dev/null +++ b/tests/unit/test_analyze_review_runtime_routes.py @@ -0,0 +1,108 @@ +from __future__ import annotations + +import json +from dataclasses import dataclass + +from fastapi.testclient import TestClient + +from paperbot.api import main as api_main + + +def _parse_sse_payloads(raw_text: str): + payloads = [] + for line in raw_text.splitlines(): + if not line.startswith("data: "): + continue + payload = line[len("data: ") :].strip() + if payload == "[DONE]": + continue + payloads.append(json.loads(payload)) + return payloads + + +@dataclass +class _AnalyzeResult: + summary: str + key_contributions: list[str] + methodology: str + strengths: list[str] + weaknesses: list[str] + + +@dataclass +class _ReviewResult: + summary: str + contributions: list[str] + strengths: list[str] + weaknesses: list[str] + novelty_score: float + decision: str + + +class _FakeResearchAgent: + def __init__(self, _config): + pass + + async def analyze_paper(self, *, title: str, abstract: str): + return _AnalyzeResult( + summary=f"summary::{title}", + key_contributions=["k1"], + methodology=f"m::{abstract[:8]}", + strengths=["s1"], + weaknesses=["w1"], + ) + + +class _FakeReviewerAgent: + def __init__(self, _config): + pass + + async def review(self, *, title: str, abstract: str): + return _ReviewResult( + summary=f"review::{title}", + contributions=["c1"], + strengths=["s1"], + weaknesses=["w1"], + novelty_score=4.2, + decision="accept", + ) + + +def test_analyze_route_uses_runtime_contract(monkeypatch): + import paperbot.agents.research as research_pkg + + monkeypatch.setattr(research_pkg, "ResearchAgent", _FakeResearchAgent) + + with TestClient(api_main.app) as client: + resp = client.post( + "/api/analyze", + json={"title": "UniICL", "abstract": "context compression"}, + ) + + assert resp.status_code == 200 + payloads = _parse_sse_payloads(resp.text) + assert payloads + assert payloads[-1]["type"] == "result" + assert payloads[-1]["data"]["summary"] == "summary::UniICL" + assert payloads[-1]["envelope"]["workflow"] == "analyze" + assert payloads[-1]["envelope"]["trace_id"] + + +def test_review_route_uses_runtime_contract(monkeypatch): + import paperbot.agents.review as review_pkg + + monkeypatch.setattr(review_pkg, "ReviewerAgent", _FakeReviewerAgent) + + with TestClient(api_main.app) as client: + resp = client.post( + "/api/review", + json={"title": "VL-Cache", "abstract": "kv cache"}, + ) + + assert resp.status_code == 200 + payloads = _parse_sse_payloads(resp.text) + assert payloads + assert payloads[-1]["type"] == "result" + assert payloads[-1]["data"]["recommendation"] == "accept" + assert payloads[-1]["envelope"]["workflow"] == "review" + assert payloads[-1]["envelope"]["trace_id"]