feat(cs_feed): 论文关联 PaperTopic + 触发引用 auto_link

apps/api/routers/cs_feeds.py

@@ -6,6 +6,7 @@
 
 from fastapi import APIRouter, Depends, Query, Request
 
+from packages.ai.cs_feed_orchestrator import CSFeedOrchestrator
 from packages.domain.task_tracker import global_tracker
 from packages.storage.db import SessionLocal
 from packages.storage.repositories import CSFeedRepository
@@ -165,19 +166,28 @@ def _fetch_fn(progress_callback=None):
             progress_callback(f"开始入库 ({total_papers} 篇)...", 50, 100)
 
         count = 0
+        paper_ids: list[str] = []
         with session_scope() as session:
             paper_repo = PaperRepository(session)
             for i, p in enumerate(papers):
-                paper_repo.upsert_paper(p)
+                saved = paper_repo.upsert_paper(p)
                 count += 1
+                paper_ids.append(saved.id)
                 if progress_callback:
                     progress_callback(
                         f"入库中 ({i + 1}/{total_papers})...",
                         50 + int((i + 1) / total_papers * 40),
                         100,
                     )
+            # 关联到 cs_feed 分类 topic + 触发 auto_link（与 orchestrator 行为一致）
+            if paper_ids:
+                CSFeedOrchestrator._link_cs_papers_to_topic(session, category_code, paper_ids)
             repo.update_run_status(category_code, count)
 
+        # 抓取后触发引用自动关联（与 orchestrator 对齐）
+        if paper_ids:
+            CSFeedOrchestrator._trigger_auto_link(paper_ids)
+
         if progress_callback:
             progress_callback("抓取完成", 95, 100)
         return {"fetched": count}

packages/ai/cs_feed_orchestrator.py

@@ -149,6 +149,12 @@ def run(self):
                     if title:
                         titles.append(title)
 
+                # 关联论文到 cs_feed 分类对应的 topic（每分类一个 disabled topic）
+                # 使 cs_feed 论文接入主题侧边栏/图谱/统计。enabled=False 防
+                # topic_dispatch 重复抓取同一分类（仅 cs_feed_dispatch 负责）
+                if paper_ids:
+                    self._link_cs_papers_to_topic(sub_session, category_code, paper_ids)
+
                 sub_repo.update_run_status(category_code, count)
                 sub_session.commit()
                 logger.info("[CSFeed] %s: ingested %d papers", category_code, count)
@@ -158,6 +164,8 @@ def run(self):
                     # 此前 cs_feed 论文处于 unread 无 embedding 无 topic，只能靠
                     # idle_processor 事后补——改为抓取即处理
                     self._process_cs_papers(paper_ids)
+                    # 抓取后触发引用自动关联（与 ingest_arxiv 对齐，此前 cs_feed 缺这步）
+                    self._trigger_auto_link(paper_ids)
             except Exception as e:
                 sub_session.rollback()
                 err_str = str(e)
@@ -217,6 +225,43 @@ def _process_cs_papers(self, paper_ids: list[str]) -> None:
             finally:
                 limiter.end_task()
 
+    @staticmethod
+    def _link_cs_papers_to_topic(session, category_code: str, paper_ids: list[str]) -> None:
+        """把 cs_feed 抓取的论文关联到对应分类的 topic（每分类一个 disabled topic）。
+
+        命名约定：topic name = `csfeed:{category_code}`，query = `cat:{category_code}`，
+        enabled=False（防 topic_dispatch 重复抓取同一分类，仅 cs_feed_dispatch 负责）。
+        upsert_topic 按 name 幂等，link_to_topic 受 uq_paper_topic 约束兜底幂等。
+        关联后 cs_feed 论文即接入主题侧边栏/图谱视图/统计/推荐 boost。
+        """
+        from packages.storage.repositories import PaperRepository, TopicRepository
+
+        topic_repo = TopicRepository(session)
+        topic = topic_repo.upsert_topic(
+            name=f"csfeed:{category_code}",
+            query=f"cat:{category_code}",
+            enabled=False,
+        )
+        paper_repo = PaperRepository(session)
+        for pid in paper_ids:
+            paper_repo.link_to_topic(pid, topic.id)
+
+    @staticmethod
+    def _trigger_auto_link(paper_ids: list[str]) -> None:
+        """抓取后触发引用自动关联（复用 paper_pipelines 的有界线程池）。
+
+        此前 cs_feed 只入库不 auto_link，论文间引用关系靠后续 sync_incremental 补。
+        与 ingest_arxiv 对齐，抓取即提交后台关联。
+        """
+        if not paper_ids:
+            return
+        try:
+            from packages.ai.pipelines.paper_pipelines import _auto_link_pool, _bg_auto_link
+
+            _auto_link_pool.submit(_bg_auto_link, paper_ids)
+        except Exception as exc:
+            logger.warning("[CSFeed] 触发 auto_link 失败: %s", exc)
+
     def _notify_digest(self, digest: list[tuple[str, int, list[str]]]) -> None:
         """抓取入库后发送邮件摘要；SMTP 未配置或无收件人时静默跳过"""
         from packages.config import get_settings

tests/test_repositories.py

@@ -355,3 +355,72 @@ def test_skim_report_roundtrip(self, db_session):
         assert "一句话总结" in report.summary_md
         assert report.skim_score == 0.85
         assert report.key_insights.get("skim_one_liner") == "一句话总结"
+
+
+class TestCSFeedTopicLink:
+    def test_link_creates_disabled_topic_and_links_papers(self, db_session):
+        """cs_feed 论文关联到每分类的 disabled topic（接入主题侧边栏/图谱/统计）"""
+        from packages.ai.cs_feed_orchestrator import CSFeedOrchestrator
+
+        repo = PaperRepository(db_session)
+        p1 = repo.upsert_paper(
+            PaperCreate(arxiv_id="2401.00201", title="t1", abstract="a", metadata={})
+        )
+        p2 = repo.upsert_paper(
+            PaperCreate(arxiv_id="2401.00202", title="t2", abstract="a", metadata={})
+        )
+        db_session.flush()
+
+        CSFeedOrchestrator._link_cs_papers_to_topic(db_session, "cs.AI", [p1.id, p2.id])
+        db_session.commit()
+
+        topic = TopicRepository(db_session).get_by_name("csfeed:cs.AI")
+        assert topic is not None
+        # enabled=False 防 topic_dispatch 重复抓取同一分类
+        assert topic.enabled is False
+        assert topic.query == "cat:cs.AI"
+        linked = PaperRepository(db_session).list_by_topic(topic.id)
+        assert len(linked) == 2
+
+    def test_link_idempotent_no_duplicate_rows(self, db_session):
+        """重复关联同一 (paper, topic) 不产生重复行（uq_paper_topic 兜底）"""
+        from packages.ai.cs_feed_orchestrator import CSFeedOrchestrator
+
+        repo = PaperRepository(db_session)
+        p = repo.upsert_paper(
+            PaperCreate(arxiv_id="2401.00203", title="t", abstract="a", metadata={})
+        )
+        db_session.flush()
+
+        CSFeedOrchestrator._link_cs_papers_to_topic(db_session, "cs.LG", [p.id])
+        db_session.commit()
+        # 再关联一次
+        CSFeedOrchestrator._link_cs_papers_to_topic(db_session, "cs.LG", [p.id])
+        db_session.commit()
+
+        topic = TopicRepository(db_session).get_by_name("csfeed:cs.LG")
+        linked = PaperRepository(db_session).list_by_topic(topic.id)
+        assert len(linked) == 1  # 幂等，不重复
+
+    def test_link_category_isolation(self, db_session):
+        """不同分类建独立 topic，论文不串"""
+        from packages.ai.cs_feed_orchestrator import CSFeedOrchestrator
+
+        repo = PaperRepository(db_session)
+        p_ai = repo.upsert_paper(
+            PaperCreate(arxiv_id="2401.00204", title="ai", abstract="a", metadata={})
+        )
+        p_lg = repo.upsert_paper(
+            PaperCreate(arxiv_id="2401.00205", title="lg", abstract="a", metadata={})
+        )
+        db_session.flush()
+
+        CSFeedOrchestrator._link_cs_papers_to_topic(db_session, "cs.AI", [p_ai.id])
+        CSFeedOrchestrator._link_cs_papers_to_topic(db_session, "cs.LG", [p_lg.id])
+        db_session.commit()
+
+        t_ai = TopicRepository(db_session).get_by_name("csfeed:cs.AI")
+        t_lg = TopicRepository(db_session).get_by_name("csfeed:cs.LG")
+        assert len(PaperRepository(db_session).list_by_topic(t_ai.id)) == 1
+        assert len(PaperRepository(db_session).list_by_topic(t_lg.id)) == 1
+        assert t_ai.id != t_lg.id