Feat(core): DDD + Hexagonal Architecture — 身份统一 + 统一搜索 + Enrichment Pipeline#50
Conversation
…ion, unified search, enrichment pipeline P0 — Identity Unification: - Add paper_identifiers table (Alembic 0009) + PaperIdentifierModel - Add PaperIdentity value object, IdentityStore CRUD, IdentityResolver service - Dual-write canonical_paper_id on paper_feedback + paper_identifiers on paper upsert - Feature-flagged single-FK JOIN path for get_user_library (PAPERBOT_USE_CANONICAL_FK) - Backfill script (scripts/backfill_identifiers.py) P1 — Unified PaperSearchService: - SearchPort protocol + 5 adapters (S2, arXiv, papers.cool, HF Daily, OpenAlex) - PaperSearchService facade with concurrent fan-out, dedup, optional persistence - ContextEngine refactored to prefer PaperSearchService with legacy fallback - RegistryPort, FeedbackPort, EnrichmentPort protocol interfaces P2 — Enrichment Pipeline: - EnrichmentPipeline (Chain of Responsibility) for judge/summary/repo steps Domain: - PaperCandidate (normalized ACL output), FeedbackAction, JudgeScore, LLMSummary, ResearchTrack Fix: remove PaperModel.external_url reference in _resolve_paper_ref_id (column doesn't exist)
📝 WalkthroughWalkthroughThis PR introduces a multi-source paper search system with unified identifier tracking. It adds domain models for papers, feedback, enrichment, and research tracks; creates Protocol-based port interfaces for search, registry, enrichment, and feedback operations; implements adapters for multiple paper search sources (Semantic Scholar, ArXiv, OpenAlex, Hugging Face Daily, Papers.Cool); and establishes an identity resolution system to map external IDs to canonical paper records. Database schema is extended with a new paper_identifiers table and canonical_paper_id field on feedback records, with a backfill script for migration support. Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant SearchService as PaperSearchService
participant Adapters as SearchPort<br/>Adapters
participant Dedup as Deduplicator
participant Registry as PaperRegistry
participant IdentityResolver as IdentityResolver
Client->>SearchService: search(query, sources, persist)
activate SearchService
SearchService->>SearchService: _select_adapters(sources)
par Parallel Searches
SearchService->>Adapters: search(query)
Adapters-->>SearchService: List[PaperCandidate]
end
SearchService->>SearchService: Build provenance map
SearchService->>Dedup: Deduplicate by title_hash
alt persist=true and registry provided
SearchService->>Registry: upsert_many(papers)
activate Registry
Registry-->>SearchService: {paper_ids, ...}
deactivate Registry
SearchService->>IdentityResolver: resolve() for identities
activate IdentityResolver
IdentityResolver-->>SearchService: canonical_id
deactivate IdentityResolver
end
SearchService->>SearchService: Build SearchResult
SearchService-->>Client: SearchResult{papers, provenance, totals}
deactivate SearchService
sequenceDiagram
participant App
participant IdentityResolver as IdentityResolver
participant IdentityStore as IdentityStore
participant DB as PaperModel<br/>PaperIdentifierModel
App->>IdentityResolver: resolve(external_id, hints)
activate IdentityResolver
IdentityResolver->>IdentityStore: resolve(source, external_id)
activate IdentityStore
IdentityStore->>DB: Query paper_identifiers
DB-->>IdentityStore: Optional[paper_id]
deactivate IdentityStore
alt Found
IdentityStore-->>IdentityResolver: paper_id
else Not Found
IdentityResolver->>IdentityResolver: Normalize ID (arXiv/DOI)
IdentityResolver->>DB: Query papers by arxiv_id/doi
DB-->>IdentityResolver: Candidates
alt Still no match
IdentityResolver->>IdentityResolver: URL & title matching
IdentityResolver->>DB: Query by URLs/title
DB-->>IdentityResolver: Fallback candidates
end
end
IdentityResolver-->>App: Optional[canonical_paper_id]
deactivate IdentityResolver
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 1 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (1 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Summary of ChangesHello @jerry609, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! 此拉取请求旨在通过采用领域驱动设计（DDD）和六边形架构（Hexagonal Architecture）来重构核心模块和数据层，从而统一 Research、DailyPaper 和 Papers Library 三个模块的数据管道。这些更改显著提升了系统处理论文身份、搜索和富集的能力，使其更具可扩展性和可维护性，为未来的功能扩展奠定了坚实基础。 Highlights
🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Changelog
Activity
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a significant and well-structured refactoring towards DDD and Hexagonal Architecture, unifying paper identity management, creating a new search service, and an enrichment pipeline. A critical security vulnerability was identified: the recurring pattern of using scalar_one_or_none() on database columns without unique constraints creates a Denial of Service risk when duplicate data is encountered. Switching to .first() is recommended to improve robustness. Additionally, a critical bug in the new IdentityResolver could cause a runtime error due to a reference to a non-existent model attribute, and several high-severity performance issues were found in the data backfill script and the PaperStore related to N+1 queries and inefficient batch updates. Specific suggestions are provided to address these issues, enhance performance, and improve maintainability.
| select(PaperModel).where( | ||
| or_( | ||
| PaperModel.url.in_(url_candidates), | ||
| PaperModel.external_url.in_(url_candidates), |
There was a problem hiding this comment.
The query references PaperModel.external_url, but this attribute does not exist on the PaperModel class defined in src/paperbot/infrastructure/stores/models.py. This will cause an AttributeError at runtime. The PR description mentions fixing this exact issue in research_store.py, so it seems this one was missed.
| with provider.session() as session: | ||
| papers = session.execute(select(PaperModel)).scalars().all() | ||
| for paper in papers: | ||
| pairs: list[tuple[str, str]] = [] | ||
| if paper.semantic_scholar_id: | ||
| pairs.append(("semantic_scholar", paper.semantic_scholar_id)) | ||
| if paper.arxiv_id: | ||
| pairs.append(("arxiv", paper.arxiv_id)) | ||
| if paper.openalex_id: | ||
| pairs.append(("openalex", paper.openalex_id)) | ||
| if paper.doi: | ||
| pairs.append(("doi", paper.doi)) | ||
|
|
||
| for source, eid in pairs: | ||
| existing = session.execute( | ||
| select(PaperIdentifierModel).where( | ||
| PaperIdentifierModel.source == source, | ||
| PaperIdentifierModel.external_id == eid, | ||
| ) | ||
| ).scalar_one_or_none() | ||
| if existing is None: | ||
| session.add( | ||
| PaperIdentifierModel( | ||
| paper_id=paper.id, | ||
| source=source, | ||
| external_id=eid, | ||
| created_at=now, | ||
| ) | ||
| ) | ||
| created += 1 | ||
| else: | ||
| skipped += 1 | ||
|
|
||
| session.commit() |
There was a problem hiding this comment.
This function has a performance issue due to N+1 queries. It fetches all papers and then, inside a loop, executes a query for each potential identifier to check for its existence. This can be very slow for a large number of papers.
Consider optimizing this by:
- Fetching all existing
(source, external_id)pairs frompaper_identifiersinto a Pythonsetbefore the loop. - In the loop, checking for existence against this set in memory.
- Collecting all new
PaperIdentifierModelobjects in a list and usingsession.add_all()at the end for a bulk insert.
| with provider.session() as session: | ||
| rows = ( | ||
| session.execute( | ||
| select(PaperFeedbackModel).where( | ||
| PaperFeedbackModel.canonical_paper_id.is_(None), | ||
| PaperFeedbackModel.paper_ref_id.is_not(None), | ||
| ) | ||
| ) | ||
| .scalars() | ||
| .all() | ||
| ) | ||
| for row in rows: | ||
| row.canonical_paper_id = row.paper_ref_id | ||
| updated += 1 | ||
| session.commit() | ||
|
|
There was a problem hiding this comment.
This function loads all matching PaperFeedbackModel rows into memory using .all() before updating them one by one. For a large paper_feedback table, this can lead to high memory consumption. A more efficient and memory-friendly approach is to perform a bulk UPDATE statement directly in the database. You'll need to import update from sqlalchemy for this.
with provider.session() as session:
stmt = update(PaperFeedbackModel).where(
PaperFeedbackModel.canonical_paper_id.is_(None),
PaperFeedbackModel.paper_ref_id.is_not(None),
).values(canonical_paper_id=PaperFeedbackModel.paper_ref_id)
result = session.execute(stmt)
updated = result.rowcount
session.commit()| def _sync_identifiers(session, row: PaperModel) -> None: | ||
| """Write known external IDs to paper_identifiers (idempotent).""" | ||
| pairs: list[tuple[str, str]] = [] | ||
| if row.semantic_scholar_id: | ||
| pairs.append(("semantic_scholar", row.semantic_scholar_id)) | ||
| if row.arxiv_id: | ||
| pairs.append(("arxiv", row.arxiv_id)) | ||
| if row.openalex_id: | ||
| pairs.append(("openalex", row.openalex_id)) | ||
| if row.doi: | ||
| pairs.append(("doi", row.doi)) | ||
|
|
||
| for source, eid in pairs: | ||
| existing = session.execute( | ||
| select(PaperIdentifierModel).where( | ||
| PaperIdentifierModel.source == source, | ||
| PaperIdentifierModel.external_id == eid, | ||
| ) | ||
| ).scalar_one_or_none() | ||
| if existing is None: | ||
| session.add( | ||
| PaperIdentifierModel( | ||
| paper_id=row.id, | ||
| source=source, | ||
| external_id=eid, | ||
| created_at=_utcnow(), | ||
| ) | ||
| ) | ||
| elif existing.paper_id != row.id: | ||
| existing.paper_id = row.id | ||
| try: | ||
| session.flush() | ||
| except Exception: | ||
| session.rollback() |
There was a problem hiding this comment.
The _sync_identifiers method has a performance issue due to N+1 queries. Inside the loop, it executes a SELECT query for each identifier type. For a paper with multiple identifiers, this is inefficient.
Additionally, this logic duplicates functionality from IdentityStore.upsert_identity. It would be better to refactor this to either reuse IdentityStore or optimize the queries by fetching all existing identifiers for the paper in a single query before the loop.
| # Dual-write identifiers for all papers in this batch | ||
| for paper in papers: | ||
| existing = self._find_existing(session, paper) | ||
| if existing: | ||
| self._sync_identifiers(session, existing) | ||
|
|
There was a problem hiding this comment.
This block iterates over all papers in the batch a second time just to sync their identifiers. This is inefficient as it requires another database lookup (_find_existing) for each paper. The identifier synchronization should be performed within the main loop (lines 437-451) for both new and updated papers, right after they have been assigned an ID (for new papers) or retrieved (for updated papers). This would avoid the second loop and the repeated _find_existing calls.
| row = session.execute( | ||
| select(PaperIdentifierModel).where( | ||
| PaperIdentifierModel.external_id == external_id, | ||
| ) | ||
| ).scalar_one_or_none() |
There was a problem hiding this comment.
The use of scalar_one_or_none() on the external_id column is potentially dangerous because external_id is not unique across different sources (it is only unique when combined with source). If the same external_id exists for multiple sources (e.g., a numeric ID that happens to be used by both arXiv and Semantic Scholar), this call will raise a MultipleResultsFound exception, leading to a 500 Internal Server Error. This can be exploited as a Denial of Service (DoS) vector by an attacker who can cause duplicate external IDs to be registered in the system.
Recommendation: Use .scalars().first() instead to safely retrieve the first matching record if multiple exist.
| row = session.execute( | |
| select(PaperIdentifierModel).where( | |
| PaperIdentifierModel.external_id == external_id, | |
| ) | |
| ).scalar_one_or_none() | |
| row = session.execute( | |
| select(PaperIdentifierModel).where( | |
| PaperIdentifierModel.external_id == external_id, | |
| ).limit(1) | |
| ).scalars().first() |
| row = session.execute( | ||
| select(PaperModel).where( | ||
| func.lower(PaperModel.title) == title.lower() | ||
| ) | ||
| ).scalar_one_or_none() |
There was a problem hiding this comment.
The use of scalar_one_or_none() when querying by paper title is risky because titles are not guaranteed to be unique in the papers table. If multiple papers with the same title (but different identifiers) are harvested, this call will raise a MultipleResultsFound exception, causing the request to fail with a 500 error. This presents a Denial of Service (DoS) risk if an attacker can trigger the ingestion of papers with duplicate titles.
Recommendation: Use .scalars().first() and include a .limit(1) in the query to ensure the application handles potential duplicates gracefully.
| row = session.execute( | |
| select(PaperModel).where( | |
| func.lower(PaperModel.title) == title.lower() | |
| ) | |
| ).scalar_one_or_none() | |
| row = session.execute( | |
| select(PaperModel).where( | |
| func.lower(PaperModel.title) == title.lower() | |
| ).limit(1) | |
| ).scalars().first() |
| try: | ||
| return bool(context.is_offline_mode()) | ||
| except Exception: | ||
| return False |
There was a problem hiding this comment.
Using a broad except Exception: can mask unexpected errors and make debugging more difficult. If the goal is to handle cases where the script runs outside an Alembic context and context is not defined, it's better to catch a more specific exception like NameError.
except NameError:
return False| deduplicator=None, | ||
| registry=None, | ||
| identity_store=None, | ||
| ): | ||
| self._adapters = adapters | ||
| self._deduplicator = deduplicator | ||
| self._registry = registry | ||
| self._identity_store = identity_store |
There was a problem hiding this comment.
The __init__ method accepts deduplicator and identity_store as arguments, but they are assigned to instance variables that are never used within the class. This can be confusing for future maintenance. If they are intended for future use, a comment explaining this would be helpful. Otherwise, they should be removed to clean up the interface.
| deduplicator=None, | |
| registry=None, | |
| identity_store=None, | |
| ): | |
| self._adapters = adapters | |
| self._deduplicator = deduplicator | |
| self._registry = registry | |
| self._identity_store = identity_store | |
| deduplicator=None, # TODO: Unused parameter, for future implementation? | |
| registry=None, | |
| identity_store=None, # TODO: Unused parameter, for future implementation? | |
| ): | |
| self._adapters = adapters | |
| self._deduplicator = deduplicator | |
| self._registry = registry | |
| self._identity_store = identity_store |
There was a problem hiding this comment.
Pull request overview
This PR restructures the core data/search pipeline toward DDD + Hexagonal Architecture, introducing a unified paper identity layer (canonical papers.id) and a new multi-source search facade intended to serve Research / DailyPaper / Library flows consistently.
Changes:
- Adds
paper_identifiersmapping + dual-write path forpaper_feedback.canonical_paper_id, plus a backfill script and Alembic migration. - Introduces
PaperSearchService+SearchPortand multiple source adapters to unify paper search and optional persistence. - Adds new domain value objects and an
EnrichmentPipelinescaffold for shared enrichment steps.
Reviewed changes
Copilot reviewed 25 out of 25 changed files in this pull request and generated 17 comments.
Show a summary per file
| File | Description |
|---|---|
| src/paperbot/infrastructure/stores/research_store.py | Dual-write canonical_paper_id; removes nonexistent external_url reference in resolver path. |
| src/paperbot/infrastructure/stores/paper_store.py | Dual-write identifiers; feature-flagged canonical library join. |
| src/paperbot/infrastructure/stores/models.py | Adds canonical_paper_id column + paper_identifiers ORM model/relationship. |
| src/paperbot/infrastructure/stores/identity_store.py | CRUD store for paper_identifiers. |
| src/paperbot/infrastructure/adapters/s2_search_adapter.py | S2 SearchPort adapter. |
| src/paperbot/infrastructure/adapters/paperscool_adapter.py | papers.cool SearchPort adapter. |
| src/paperbot/infrastructure/adapters/openalex_adapter.py | OpenAlex SearchPort adapter. |
| src/paperbot/infrastructure/adapters/hf_daily_adapter.py | HF Daily Papers SearchPort adapter. |
| src/paperbot/infrastructure/adapters/arxiv_search_adapter.py | arXiv SearchPort adapter. |
| src/paperbot/infrastructure/adapters/init.py | Adapter registry builder. |
| src/paperbot/domain/track.py | Extracts ResearchTrack aggregate. |
| src/paperbot/domain/paper.py | Adds PaperCandidate ACL output + title hashing. |
| src/paperbot/domain/identity.py | Adds PaperIdentity value object. |
| src/paperbot/domain/feedback.py | Adds feedback value objects/enums. |
| src/paperbot/domain/enrichment.py | Adds enrichment value objects. |
| src/paperbot/context_engine/engine.py | Prefers injected PaperSearchService-style search over legacy searcher. |
| src/paperbot/application/services/paper_search_service.py | Unified async fan-out search + dedup + optional persistence. |
| src/paperbot/application/services/identity_resolver.py | New canonical-id resolution service (identifiers table + fallbacks). |
| src/paperbot/application/services/enrichment_pipeline.py | Chain-of-responsibility enrichment pipeline scaffold. |
| src/paperbot/application/ports/paper_search_port.py | SearchPort protocol. |
| src/paperbot/application/ports/paper_registry_port.py | RegistryPort protocol. |
| src/paperbot/application/ports/feedback_port.py | FeedbackPort protocol. |
| src/paperbot/application/ports/enrichment_port.py | EnrichmentPort protocol. |
| scripts/backfill_identifiers.py | Backfill script for identifiers + canonical FK column. |
| alembic/versions/0009_paper_identifiers.py | Migration creating paper_identifiers + adding canonical_paper_id column/index. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| # 3. Persist if requested | ||
| if persist and self._registry: | ||
| for paper in unique: | ||
| try: | ||
| result_dict = self._registry.upsert_paper( | ||
| paper=paper.to_dict(), | ||
| source_hint=provenance.get(paper.title_hash, ["unknown"])[0], | ||
| ) | ||
| paper.canonical_id = result_dict.get("id") | ||
| except Exception as e: | ||
| logger.warning(f"Failed to persist paper '{paper.title[:50]}': {e}") | ||
|
|
||
| return SearchResult( | ||
| papers=unique[:max_results], | ||
| provenance=provenance, | ||
| total_raw=total_raw, | ||
| duplicates_removed=duplicates_removed, | ||
| ) |
There was a problem hiding this comment.
PaperSearchService.search() persists every deduplicated paper in unique when persist=True, but returns only unique[:max_results]. For multi-source fan-out, unique can be much larger than max_results, leading to unexpected extra DB writes and slower requests. Consider persisting only the papers you will return (or add a separate persist_limit/persist_all option).
| from sqlalchemy import Integer, String, cast, desc, func, or_, select | ||
|
|
||
| from paperbot.domain.harvest import HarvestedPaper, HarvestSource | ||
| from paperbot.domain.identity import PaperIdentity |
There was a problem hiding this comment.
PaperIdentity is imported but not used in this module; removing it will avoid confusion and keep the dependency surface small.
| from paperbot.domain.identity import PaperIdentity |
|
|
||
| # Canonical FK (dual-write migration — will replace paper_id + paper_ref_id) | ||
| canonical_paper_id: Mapped[Optional[int]] = mapped_column( | ||
| Integer, nullable=True, index=True |
There was a problem hiding this comment.
PaperFeedbackModel.canonical_paper_id is described as a “Canonical FK”, but it’s declared as a plain Integer without a ForeignKey("papers.id"). Without the FK constraint you lose referential integrity and may get worse query planning/index usage. If this is intended to become the canonical join key, it should be declared with an explicit FK (even if nullable during migration).
| Integer, nullable=True, index=True | |
| Integer, ForeignKey("papers.id"), nullable=True, index=True |
| # Optional: PaperSearchService for unified search | ||
| try: | ||
| from paperbot.application.services.paper_search_service import PaperSearchService | ||
| except ImportError: # pragma: no cover | ||
| PaperSearchService = None # type: ignore | ||
|
|
There was a problem hiding this comment.
PaperSearchService is imported behind a try/except but never used in this module (the code checks self.search_service directly). This dead import block makes it harder to understand the actual dependency path; consider removing it or using it to type-check/instantiate a default PaperSearchService when search_service is not provided.
| # Optional: PaperSearchService for unified search | |
| try: | |
| from paperbot.application.services.paper_search_service import PaperSearchService | |
| except ImportError: # pragma: no cover | |
| PaperSearchService = None # type: ignore |
| # --- paper_feedback.canonical_paper_id (nullable FK for dual-write) --- | ||
| if not _is_offline() and "canonical_paper_id" in _get_columns("paper_feedback"): | ||
| pass # column already exists | ||
| else: | ||
| op.add_column( | ||
| "paper_feedback", | ||
| sa.Column("canonical_paper_id", sa.Integer(), nullable=True), | ||
| ) | ||
| _create_index( | ||
| "ix_paper_feedback_canonical_paper_id", | ||
| "paper_feedback", | ||
| ["canonical_paper_id"], | ||
| ) |
There was a problem hiding this comment.
Migration docstring says paper_feedback.canonical_paper_id is a “nullable FK column”, but the migration only adds a plain integer column + index (no FK constraint to papers.id). If you intend FK semantics, add the foreign key constraint in this revision (or clarify in the migration/docstring that it’s intentionally not enforced during the expand phase).
|
|
||
| from dataclasses import dataclass | ||
| from datetime import datetime | ||
| from typing import Any, Dict, Optional |
There was a problem hiding this comment.
Import of 'Dict' is not used.
Import of 'Any' is not used.
| from typing import Any, Dict, Optional | |
| from typing import Optional |
|
|
||
| from __future__ import annotations | ||
|
|
||
| from typing import Any, Dict, List, Optional, Protocol, runtime_checkable |
There was a problem hiding this comment.
Import of 'List' is not used.
Import of 'Optional' is not used.
| from typing import Any, Dict, List, Optional, Protocol, runtime_checkable | |
| from typing import Any, Dict, Protocol, runtime_checkable |
|
|
||
| from __future__ import annotations | ||
|
|
||
| from typing import Any, Dict, List, Optional |
There was a problem hiding this comment.
Import of 'Dict' is not used.
Import of 'Any' is not used.
| from typing import Any, Dict, List, Optional | |
| from typing import List, Optional |
| def close(self) -> None: | ||
| try: | ||
| self._provider.engine.dispose() | ||
| except Exception: |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
| except Exception: | ||
| pass |
There was a problem hiding this comment.
'except' clause does nothing but pass and there is no explanatory comment.
| except Exception: | |
| pass | |
| except Exception as e: | |
| logger.warning("Failed to close adapter %r: %s", adapter, e) |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 10
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/paperbot/infrastructure/stores/research_store.py (1)
1376-1396:⚠️ Potential issue | 🟠 Major
scalar_one_or_none()will raiseMultipleResultsFoundat runtime for duplicate URLs or titles.The TODO on line 1376 correctly identifies this. Both the URL-based lookup (line 1386) and the title-based lookup (line 1394) are susceptible. Title collisions are especially likely in practice. Consider switching to
.first()now to avoid unhandled exceptions bubbling up during feedback recording or paper resolution.Proposed fix
- row = session.execute( - select(PaperModel).where( - or_( - PaperModel.url.in_(url_candidates), - PaperModel.pdf_url.in_(url_candidates), - ) - ) - ).scalar_one_or_none() + row = session.execute( + select(PaperModel).where( + or_( + PaperModel.url.in_(url_candidates), + PaperModel.pdf_url.in_(url_candidates), + ) + ).limit(1) + ).scalar_one_or_none() if row is not None: return int(row.id) title = str(metadata.get("title") or "").strip() if title: - row = session.execute( - select(PaperModel).where(func.lower(PaperModel.title) == title.lower()) - ).scalar_one_or_none() + row = session.execute( + select(PaperModel).where(func.lower(PaperModel.title) == title.lower()).limit(1) + ).scalar_one_or_none() if row is not None: return int(row.id)
🤖 Fix all issues with AI agents
In `@alembic/versions/0009_paper_identifiers.py`:
- Around line 97-100: The downgrade() currently drops an index, a column, and a
table without existence checks which makes re-running a partially-applied
downgrade fail; update downgrade() to be idempotent by guarding each operation
(the op.drop_index("ix_paper_feedback_canonical_paper_id", ...),
op.drop_column("paper_feedback", "canonical_paper_id"), and
op.drop_table("paper_identifiers")) — either query the DB via the migration
bind/Inspector to check for the index/column/table before calling each op.* or
wrap each call in a narrow try/except that ignores "not found" errors so the
downgrade can be safely re-run.
In `@src/paperbot/application/services/identity_resolver.py`:
- Around line 102-114: The OR condition in identity resolution references a
removed column PaperModel.external_url causing AttributeError; update the URL
match in the function that queries PaperModel (the block using
session.execute(select(PaperModel).where(or_( ... ))).first()) to remove the
PaperModel.external_url operand and only check PaperModel.url and
PaperModel.pdf_url (i.e., drop the external_url reference from the or_ call) so
the select uses existing columns only.
In `@src/paperbot/application/services/paper_search_service.py`:
- Around line 100-117: The synchronous call to self._registry.upsert_paper
inside the async method blocks the event loop; change the persistence call to
run in a thread by awaiting asyncio.to_thread(self._registry.upsert_paper,
paper=paper.to_dict(), source_hint=provenance.get(paper.title_hash,
["unknown"])[0]) and assign paper.canonical_id from the returned dict, keeping
the existing exception handling but catching exceptions from the awaited
to_thread call; update imports to include asyncio if needed and ensure this
change is applied where upsert_paper is called in the same async method that
returns SearchResult (references: self._registry.upsert_paper, paper.to_dict(),
provenance, paper.canonical_id, SearchResult).
In `@src/paperbot/context_engine/engine.py`:
- Around line 522-545: Add unit tests covering the new search_service path: mock
an instance of search_service (and its search method) so when Engine.search flow
hits the branch where search_service is not None it calls
search_service.search(merged_query, sources=["semantic_scholar"],
max_results=fetch_limit, persist=True); return a mock search_result with a
.papers list of objects that expose .to_dict(), .canonical_id and
.get_identity("semantic_scholar") to exercise both canonical_id and fallback
identity branches, then assert the code injects paper_id into each returned dict
(verifying values produced from p_obj.canonical_id and p_obj.get_identity), and
assert Logger/info or method call count if relevant; place tests targeting the
Engine class/method that uses merged_query and fetch_limit so the
PaperSearchService path and the paper_id injection loop (p_obj.canonical_id,
p_obj.get_identity) are covered.
In `@src/paperbot/infrastructure/adapters/arxiv_search_adapter.py`:
- Around line 23-34: The search method currently ignores year_from/year_to
because _connector.search doesn't accept them; modify Paperbot's
ArxivSearchAdapter.search to post-filter the records returned by
self._connector.search by checking each record's published date year (use a
helper like _year_in_range(record) that extracts the year from the record's
published ISO date string and validates it against year_from/year_to) and only
convert remaining records with self._to_candidate; ensure the filter handles
missing/invalid published values gracefully (treat as out-of-range or skip) and
preserve max_results by applying the year filter before slicing to max_results.
In `@src/paperbot/infrastructure/stores/identity_store.py`:
- Around line 84-92: The current resolve_any method uses scalar_one_or_none()
which raises MultipleResultsFound when the same external_id exists across
different sources; change the query to fetch the first matching row instead (use
the session.execute(...).scalars().first() pattern) and return int(row.paper_id)
if row else None; optionally add an explicit order_by on
PaperIdentifierModel.source to make which source wins deterministic. Ensure
references to resolve_any and PaperIdentifierModel are updated accordingly.
In `@src/paperbot/infrastructure/stores/models.py`:
- Around line 469-472: The canonical_paper_id column is missing a ForeignKey
constraint, so either add ForeignKey("papers.id") to the mapped_column for
canonical_paper_id (mirroring paper_ref_id) to enforce referential integrity, or
if the omission is intentional during the dual-write migration, annotate the
field with a clear TODO comment (mentioning adding ForeignKey("papers.id") after
backfill and referencing the PAPERBOT_USE_CANONICAL_FK flag) so future readers
know this is deliberate; update the canonical_paper_id Mapped definition
accordingly.
In `@src/paperbot/infrastructure/stores/paper_store.py`:
- Around line 824-884: _get_user_library_canonical currently only sorts by
"saved_at", "citation_count" or default and therefore drops support for "title"
and "year" sorting; update the sorting branch in _get_user_library_canonical to
handle sort_by == "title" (sort by paper.title, using a case-insensitive
fallback like "" for None) and sort_by == "year" (sort by paper.year with a
numeric fallback like 0), honoring sort_order ("asc"/"desc") just like the other
branches; operate on unique_results (tuples of (PaperModel, PaperFeedbackModel))
and use x[0].title / x[0].year and the existing min_ts logic for saved_at so
behavior matches get_user_library.
- Around line 454-461: After session.flush(), avoid re-querying each paper with
_find_existing (extra N round-trips) and the risk that _sync_identifiers()
rollback will undo the whole batch; instead use the ORM model instances you
already have (the items in papers or the objects returned by the upsert/insert
logic) and pass those directly into _sync_identifiers to dual-write identifiers,
and wrap each per-paper identifier sync in a DB savepoint/independent
transaction (e.g., session.begin_nested() or equivalent) or catch and handle
exceptions locally so a failing _sync_identifiers does not rollback the entire
batch. Ensure you remove the call sites to _find_existing in the loop and
reference the existing paper instances when calling _sync_identifiers.
- Around line 238-278: _sync_identifiers currently flushes identifier rows but
never commits, and its broad except does session.rollback() which can undo
unrelated work in upsert_papers_batch; fix by making the identifier write atomic
and not able to rollback the whole outer transaction: inside _sync_identifiers
use a nested transaction/savepoint (session.begin_nested()) when inserting
PaperIdentifierModel rows, flush/commit that nested transaction (or let it
close) and on exception rollback only the nested transaction and re-raise the
error (do not call session.rollback() on the outer session); alternatively, call
session.commit() after the identifier sync when called from single-upsert paths
like upsert_paper, but for batch paths prefer the begin_nested() approach and
remove the unconditional session.rollback() so upsert_papers_batch is not
invalidated.
🧹 Nitpick comments (11)
src/paperbot/application/ports/enrichment_port.py (1)
8-14: Consider whetherLLMSummarypersistence should also be part of this port.The domain layer defines both
JudgeScoreandLLMSummaryas enrichment value objects, but this port only covers judge scores. If summary persistence is planned, adding it here now would keep the port aligned with the enrichment domain boundary. Fine to defer if summaries aren't persisted yet.src/paperbot/application/services/enrichment_pipeline.py (1)
41-54: Pipeline error handling is appropriate but silently swallows failures.The catch-all
Exceptionis the right pattern here for pipeline resilience. However,run()returnsNonewith no indication of which papers/steps failed. Consider returning a summary (e.g., list of failures) or accumulating errors inEnrichmentContext.extraso callers can act on partial failures.src/paperbot/infrastructure/adapters/__init__.py (1)
15-22: Usesource_nameproperty as registry key to avoid key/name drift.The hardcoded string keys duplicate each adapter's
source_nameproperty. If asource_namechanges in the adapter but not here, lookups will silently break.Proposed fix
def build_adapter_registry() -> Dict[str, SearchPort]: - return { - "semantic_scholar": S2SearchAdapter(), - "arxiv": ArxivSearchAdapter(), - "papers_cool": PapersCoolAdapter(), - "hf_daily": HFDailyAdapter(), - "openalex": OpenAlexAdapter(), - } + adapters = [ + S2SearchAdapter(), + ArxivSearchAdapter(), + PapersCoolAdapter(), + HFDailyAdapter(), + OpenAlexAdapter(), + ] + return {a.source_name: a for a in adapters}Also note: if any adapter's constructor fails (e.g., missing config), the entire registry build will fail. Consider wrapping each instantiation in a try/except with a warning log so remaining adapters stay available.
scripts/backfill_identifiers.py (3)
35-76: Performance: N+1 queries and full table load — will be slow on large datasets.
backfill_identifiersloads everyPaperModelinto memory (line 42) and then issues aSELECTper(source, external_id)pair per paper. For a database with, say, 100k papers averaging 2 identifiers each, that's ~200k individual queries in a single transaction.Consider:
- Chunking the paper load with
yield_per()orLIMIT/OFFSET.- Pre-loading existing identifiers into a set to avoid per-row lookups.
- Using
INSERT … ON CONFLICT DO NOTHINGfor bulk inserts instead of check-then-insert.Sketch using bulk insert with conflict handling
def backfill_identifiers(provider: SessionProvider) -> dict: """Populate paper_identifiers from papers columns.""" now = _utcnow() created = 0 skipped = 0 with provider.session() as session: - papers = session.execute(select(PaperModel)).scalars().all() + # Process in chunks to limit memory usage + query = select(PaperModel) + papers = session.execute(query).scalars() for paper in papers: pairs: list[tuple[str, str]] = [] if paper.semantic_scholar_id: pairs.append(("semantic_scholar", paper.semantic_scholar_id)) if paper.arxiv_id: pairs.append(("arxiv", paper.arxiv_id)) if paper.openalex_id: pairs.append(("openalex", paper.openalex_id)) if paper.doi: pairs.append(("doi", paper.doi)) for source, eid in pairs: - existing = session.execute( - select(PaperIdentifierModel).where( - PaperIdentifierModel.source == source, - PaperIdentifierModel.external_id == eid, - ) - ).scalar_one_or_none() - if existing is None: - session.add( - PaperIdentifierModel( - paper_id=paper.id, - source=source, - external_id=eid, - created_at=now, - ) - ) - created += 1 - else: - skipped += 1 + try: + session.add( + PaperIdentifierModel( + paper_id=paper.id, + source=source, + external_id=eid, + created_at=now, + ) + ) + session.flush() + created += 1 + except IntegrityError: + session.rollback() + skipped += 1 session.commit() return {"identifiers_created": created, "identifiers_skipped": skipped}
79-98: Use a single bulk UPDATE instead of loading all rows into Python.
backfill_canonical_paper_idloads all matching rows into memory and updates them one by one in Python. This can be replaced with a single SQLUPDATEstatement, which is significantly faster and uses negligible memory.Bulk UPDATE
def backfill_canonical_paper_id(provider: SessionProvider) -> dict: """Populate paper_feedback.canonical_paper_id from paper_ref_id.""" - updated = 0 with provider.session() as session: - rows = ( - session.execute( - select(PaperFeedbackModel).where( - PaperFeedbackModel.canonical_paper_id.is_(None), - PaperFeedbackModel.paper_ref_id.is_not(None), - ) - ) - .scalars() - .all() - ) - for row in rows: - row.canonical_paper_id = row.paper_ref_id - updated += 1 + from sqlalchemy import update + result = session.execute( + update(PaperFeedbackModel) + .where( + PaperFeedbackModel.canonical_paper_id.is_(None), + PaperFeedbackModel.paper_ref_id.is_not(None), + ) + .values(canonical_paper_id=PaperFeedbackModel.paper_ref_id) + ) + updated = result.rowcount session.commit() return {"feedback_rows_updated": updated}
21-28: Remove unusednoqadirectives.Ruff reports that the
# noqa: E402directives are unnecessary since the E402 rule is not enabled. Remove them for cleanliness.Fix
-from sqlalchemy import select # noqa: E402 -from paperbot.infrastructure.stores.models import ( # noqa: E402 +from sqlalchemy import select +from paperbot.infrastructure.stores.models import ( Base, PaperFeedbackModel, PaperIdentifierModel, PaperModel, ) -from paperbot.infrastructure.stores.sqlalchemy_db import SessionProvider, get_db_url # noqa: E402 +from paperbot.infrastructure.stores.sqlalchemy_db import SessionProvider, get_db_urlsrc/paperbot/application/services/identity_resolver.py (1)
24-31:IdentityResolveralways creates its ownSessionProvider, even when anIdentityStoreis injected.Line 31 creates a new
SessionProvider(self._db_url)unconditionally. If the caller already injects anIdentityStore(which has its own engine), this results in two separate DB engines for the same database. Consider reusing the provider from the injected store or accepting aSessionProviderdirectly.Reuse the store's provider
def __init__( self, identity_store: Optional[IdentityStore] = None, db_url: Optional[str] = None, + provider: Optional[SessionProvider] = None, ): self._identity_store = identity_store or IdentityStore(db_url=db_url) self._db_url = db_url or get_db_url() - self._provider = SessionProvider(self._db_url) + self._provider = provider or getattr(self._identity_store, '_provider', None) or SessionProvider(self._db_url)src/paperbot/domain/paper.py (1)
182-186: Title hash logic is duplicated withHarvestedPaper.compute_title_hash.
_compute_title_hashhere is identical toHarvestedPaper.compute_title_hash()indomain/harvest.py. Both normalize the same way and produce SHA256 hashes. Consider extracting a shared utility to avoid drift if one is updated without the other.src/paperbot/infrastructure/adapters/paperscool_adapter.py (1)
23-38:year_from/year_toare silently ignored — consistent with other adapters, but worth documenting.The papers.cool connector doesn't support year-range filtering, so these parameters are present only for
SearchPortconformance. This matches the pattern inArxivSearchAdapter. A brief inline comment would prevent future confusion.Add a brief note
async def search( self, query: str, *, max_results: int = 30, - year_from: Optional[int] = None, - year_to: Optional[int] = None, + year_from: Optional[int] = None, # Not supported by papers.cool + year_to: Optional[int] = None, # Not supported by papers.cool ) -> List[PaperCandidate]:src/paperbot/application/services/paper_search_service.py (1)
128-133: Silently swallowing exceptions inclose()hinders debugging.Per static analysis (S110),
try-except-passloses all diagnostic information.Proposed fix
async def close(self) -> None: for adapter in self._adapters.values(): try: await adapter.close() - except Exception: - pass + except Exception: + logger.debug("Error closing adapter %s", adapter.source_name, exc_info=True)src/paperbot/infrastructure/stores/identity_store.py (1)
107-111: Consider logging inclose()instead of silently passing.Same pattern as other stores — a
logger.debug(...)call would help diagnose teardown issues.
| def downgrade() -> None: | ||
| op.drop_index("ix_paper_feedback_canonical_paper_id", table_name="paper_feedback") | ||
| op.drop_column("paper_feedback", "canonical_paper_id") | ||
| op.drop_table("paper_identifiers") |
There was a problem hiding this comment.
Downgrade lacks idempotency guards — minor risk for partial rollback scenarios.
If the downgrade fails midway (e.g., after dropping the index but before dropping the column), re-running it will fail on drop_index. Since the upgrade is carefully guarded, consider mirroring that pattern in the downgrade, or at least noting this is by design.
This is low-priority since downgrades are rarely re-run in practice.
Suggested defensive downgrade
def downgrade() -> None:
- op.drop_index("ix_paper_feedback_canonical_paper_id", table_name="paper_feedback")
- op.drop_column("paper_feedback", "canonical_paper_id")
- op.drop_table("paper_identifiers")
+ if not _is_offline():
+ if "ix_paper_feedback_canonical_paper_id" in _get_indexes("paper_feedback"):
+ op.drop_index("ix_paper_feedback_canonical_paper_id", table_name="paper_feedback")
+ if "canonical_paper_id" in _get_columns("paper_feedback"):
+ op.drop_column("paper_feedback", "canonical_paper_id")
+ if _has_table("paper_identifiers"):
+ op.drop_table("paper_identifiers")
+ else:
+ op.drop_index("ix_paper_feedback_canonical_paper_id", table_name="paper_feedback")
+ op.drop_column("paper_feedback", "canonical_paper_id")
+ op.drop_table("paper_identifiers")🤖 Prompt for AI Agents
In `@alembic/versions/0009_paper_identifiers.py` around lines 97 - 100, The
downgrade() currently drops an index, a column, and a table without existence
checks which makes re-running a partially-applied downgrade fail; update
downgrade() to be idempotent by guarding each operation (the
op.drop_index("ix_paper_feedback_canonical_paper_id", ...),
op.drop_column("paper_feedback", "canonical_paper_id"), and
op.drop_table("paper_identifiers")) — either query the DB via the migration
bind/Inspector to check for the index/column/table before calling each op.* or
wrap each call in a narrow try/except that ignores "not found" errors so the
downgrade can be safely re-run.
| # 3. URL match | ||
| if url_candidates: | ||
| row = session.execute( | ||
| select(PaperModel).where( | ||
| or_( | ||
| PaperModel.url.in_(url_candidates), | ||
| PaperModel.external_url.in_(url_candidates), | ||
| PaperModel.pdf_url.in_(url_candidates), | ||
| ) | ||
| ) | ||
| ).first() | ||
| if row: | ||
| return int(row[0].id) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# Find and examine PaperModel definition
find . -name "models.py" -type f | grep -E "(models|model)" | head -20Repository: jerry609/PaperBot
Length of output: 157
🏁 Script executed:
# Search for PaperModel class definition and external_url references
rg -n "class PaperModel" --type=pyRepository: jerry609/PaperBot
Length of output: 133
🏁 Script executed:
# Check for external_url in the codebase
rg -n "external_url" --type=py -B 2 -A 2Repository: jerry609/PaperBot
Length of output: 20000
🏁 Script executed:
# Read PaperModel class definition
sed -n '693,750p' src/paperbot/infrastructure/stores/models.pyRepository: jerry609/PaperBot
Length of output: 2686
🏁 Script executed:
# Search for external_url field definition in PaperModel
rg -n "external_url" src/paperbot/infrastructure/stores/models.py -B 2 -A 2Repository: jerry609/PaperBot
Length of output: 43
🏁 Script executed:
# Check if there's a migration that drops the external_url column
rg -n "external_url" alembic/versions/ -A 2 -B 2Repository: jerry609/PaperBot
Length of output: 734
🏁 Script executed:
# Check for all migrations in order
ls -la alembic/versions/ | grep ".py"Repository: jerry609/PaperBot
Length of output: 754
🏁 Script executed:
# Check the latest migration files for any drops or alterations to external_url
for file in alembic/versions/000[4-9]*.py; do
echo "=== $file ==="
rg -n "external_url\|drop_column\|alter" "$file" -B 1 -A 1 || echo "No matches"
doneRepository: jerry609/PaperBot
Length of output: 446
🏁 Script executed:
# Look at the migration 0007 which mentions "paper_harvest_tables"
cat alembic/versions/0007_paper_harvest_tables.pyRepository: jerry609/PaperBot
Length of output: 5822
🏁 Script executed:
# Verify there's no external_url field in the current models.py PaperModel
grep -n "external_url" src/paperbot/infrastructure/stores/models.pyRepository: jerry609/PaperBot
Length of output: 43
Remove reference to non-existent PaperModel.external_url column.
Line 108 references PaperModel.external_url, but the current PaperModel schema only has url and pdf_url columns. The external_url column was removed from the database schema in migration 0007. This will raise an AttributeError at runtime whenever url_candidates is non-empty.
🐛 Fix: remove the non-existent column reference
# 3. URL match
if url_candidates:
row = session.execute(
select(PaperModel).where(
or_(
PaperModel.url.in_(url_candidates),
- PaperModel.external_url.in_(url_candidates),
PaperModel.pdf_url.in_(url_candidates),
)
)
).first()
if row:
return int(row[0].id)🤖 Prompt for AI Agents
In `@src/paperbot/application/services/identity_resolver.py` around lines 102 -
114, The OR condition in identity resolution references a removed column
PaperModel.external_url causing AttributeError; update the URL match in the
function that queries PaperModel (the block using
session.execute(select(PaperModel).where(or_( ... ))).first()) to remove the
PaperModel.external_url operand and only check PaperModel.url and
PaperModel.pdf_url (i.e., drop the external_url reference from the or_ call) so
the select uses existing columns only.
| # 3. Persist if requested | ||
| if persist and self._registry: | ||
| for paper in unique: | ||
| try: | ||
| result_dict = self._registry.upsert_paper( | ||
| paper=paper.to_dict(), | ||
| source_hint=provenance.get(paper.title_hash, ["unknown"])[0], | ||
| ) | ||
| paper.canonical_id = result_dict.get("id") | ||
| except Exception as e: | ||
| logger.warning(f"Failed to persist paper '{paper.title[:50]}': {e}") | ||
|
|
||
| return SearchResult( | ||
| papers=unique[:max_results], | ||
| provenance=provenance, | ||
| total_raw=total_raw, | ||
| duplicates_removed=duplicates_removed, | ||
| ) |
There was a problem hiding this comment.
Synchronous upsert_paper blocks the event loop.
self._registry.upsert_paper(...) (backed by PaperStore) is a synchronous DB call executed inside an async method without asyncio.to_thread(). When persisting many papers, this blocks the event loop for the duration of all writes.
Proposed fix — offload to thread
if persist and self._registry:
for paper in unique:
try:
- result_dict = self._registry.upsert_paper(
- paper=paper.to_dict(),
- source_hint=provenance.get(paper.title_hash, ["unknown"])[0],
+ result_dict = await asyncio.to_thread(
+ self._registry.upsert_paper,
+ paper=paper.to_dict(),
+ source_hint=provenance.get(paper.title_hash, ["unknown"])[0],
)
paper.canonical_id = result_dict.get("id")
except Exception as e:
logger.warning(f"Failed to persist paper '{paper.title[:50]}': {e}")🧰 Tools
🪛 Ruff (0.15.0)
[warning] 109-109: Do not catch blind exception: Exception
(BLE001)
🤖 Prompt for AI Agents
In `@src/paperbot/application/services/paper_search_service.py` around lines 100 -
117, The synchronous call to self._registry.upsert_paper inside the async method
blocks the event loop; change the persistence call to run in a thread by
awaiting asyncio.to_thread(self._registry.upsert_paper, paper=paper.to_dict(),
source_hint=provenance.get(paper.title_hash, ["unknown"])[0]) and assign
paper.canonical_id from the returned dict, keeping the existing exception
handling but catching exceptions from the awaited to_thread call; update imports
to include asyncio if needed and ensure this change is applied where
upsert_paper is called in the same async method that returns SearchResult
(references: self._registry.upsert_paper, paper.to_dict(), provenance,
paper.canonical_id, SearchResult).
|
|
||
| # Prefer PaperSearchService if available | ||
| if self.search_service is not None: | ||
| Logger.info( | ||
| f"Using PaperSearchService for query='{merged_query}'", | ||
| file=LogFiles.HARVEST, | ||
| ) | ||
| search_result = await self.search_service.search( | ||
| merged_query, | ||
| sources=["semantic_scholar"], | ||
| max_results=fetch_limit, | ||
| persist=True, | ||
| ) | ||
| raw = [p.to_dict() for p in search_result.papers] | ||
| # Inject paper_id from canonical_id or first identity | ||
| for p_dict, p_obj in zip(raw, search_result.papers): | ||
| pid = str(p_obj.canonical_id or "") | ||
| if not pid: | ||
| pid = p_obj.get_identity("semantic_scholar") or "" | ||
| p_dict["paper_id"] = pid | ||
| Logger.info( | ||
| f"PaperSearchService returned {len(raw)} papers", | ||
| file=LogFiles.HARVEST, | ||
| ) |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# First, let's read the actual file at the specified lines
head -545 src/paperbot/context_engine/engine.py | tail -30Repository: jerry609/PaperBot
Length of output: 1496
🏁 Script executed:
# Search for test files that might cover search_service integration
rg -n "search_service" --type=py -g '*test*' -A 3 -B 3Repository: jerry609/PaperBot
Length of output: 43
🏁 Script executed:
# Check for TODO comments around the search_service code
rg -n "TODO.*search|search.*TODO|sources.*semantic_scholar" src/paperbot/context_engine/engine.pyRepository: jerry609/PaperBot
Length of output: 118
🏁 Script executed:
# Check if there are any config files or parameters for sources
rg -n "sources.*=" --type=py src/paperbot/context_engine/engine.py -B 2 -A 2Repository: jerry609/PaperBot
Length of output: 329
🏁 Script executed:
# Find test files related to ContextEngine
fd -e py -path '*test*' | xargs grep -l "ContextEngine\|context_engine" 2>/dev/nullRepository: jerry609/PaperBot
Length of output: 231
🏁 Script executed:
# Check if there are any tests in the repository at all for this module
find . -name '*test*.py' -o -name 'test_*.py' | head -20Repository: jerry609/PaperBot
Length of output: 934
🏁 Script executed:
# Search for the search_service attribute definition to understand its nature
rg -n "self.search_service\s*=" --type=py src/paperbot/context_engine/engine.pyRepository: jerry609/PaperBot
Length of output: 109
🏁 Script executed:
# Check if there's a broader pattern of how sources are handled in search
rg -n "sources\s*=" --type=py src/paperbot/ -B 2 -A 2 | head -50Repository: jerry609/PaperBot
Length of output: 3693
🏁 Script executed:
# Check test_harvest_pipeline.py for search_service coverage
rg -n "search_service\|PaperSearchService" tests/integration/test_harvest_pipeline.py -B 2 -A 2Repository: jerry609/PaperBot
Length of output: 43
🏁 Script executed:
# Search for test files that might mock or test ContextEngine.search method
rg -n "ContextEngine\|search_service" tests/ --type py -lRepository: jerry609/PaperBot
Length of output: 43
🏁 Script executed:
# Check if the search_service integration is new (look for comments or recent changes)
rg -n "Prefer PaperSearchService\|search_service is not None" src/paperbot/context_engine/engine.py -B 5 -A 5Repository: jerry609/PaperBot
Length of output: 43
🏁 Script executed:
# Get more context around the search_service conditional to understand the fallback
sed -n '515,560p' src/paperbot/context_engine/engine.pyRepository: jerry609/PaperBot
Length of output: 2351
🏁 Script executed:
# Check for SemanticScholarSearch mentioned in the review
rg -n "SemanticScholarSearch" src/paperbot/context_engine/engine.py -B 3 -A 3Repository: jerry609/PaperBot
Length of output: 723
🏁 Script executed:
# Look for all search-related method calls in the engine to understand the flow
rg -n "\.search\(" src/paperbot/context_engine/engine.py -B 2 -A 2Repository: jerry609/PaperBot
Length of output: 311
Add test coverage for the new search_service code path.
The hardcoded sources=["semantic_scholar"] at line 531 is confirmed, which appears intentional for a safe rollout. However, per coding guidelines, behavior changes in src/**/*.py must include test coverage. The new search_service integration path has no corresponding tests. Add tests to cover the search_service flow, including the paper_id injection logic (lines 535-540).
As an optional enhancement, consider adding a configuration option to enable additional sources beyond semantic_scholar for future multi-source fan-out.
🤖 Prompt for AI Agents
In `@src/paperbot/context_engine/engine.py` around lines 522 - 545, Add unit tests
covering the new search_service path: mock an instance of search_service (and
its search method) so when Engine.search flow hits the branch where
search_service is not None it calls search_service.search(merged_query,
sources=["semantic_scholar"], max_results=fetch_limit, persist=True); return a
mock search_result with a .papers list of objects that expose .to_dict(),
.canonical_id and .get_identity("semantic_scholar") to exercise both
canonical_id and fallback identity branches, then assert the code injects
paper_id into each returned dict (verifying values produced from
p_obj.canonical_id and p_obj.get_identity), and assert Logger/info or method
call count if relevant; place tests targeting the Engine class/method that uses
merged_query and fetch_limit so the PaperSearchService path and the paper_id
injection loop (p_obj.canonical_id, p_obj.get_identity) are covered.
| async def search( | ||
| self, | ||
| query: str, | ||
| *, | ||
| max_results: int = 30, | ||
| year_from: Optional[int] = None, | ||
| year_to: Optional[int] = None, | ||
| ) -> List[PaperCandidate]: | ||
| records = await asyncio.to_thread( | ||
| self._connector.search, query=query, max_results=max_results | ||
| ) | ||
| return [self._to_candidate(r) for r in records] |
There was a problem hiding this comment.
year_from and year_to are silently ignored — results may include out-of-range papers.
The SearchPort protocol exposes year_from/year_to, and PaperSearchService forwards them, but ArxivConnector.search doesn't accept these parameters, so they're dropped. Callers relying on year filtering will get unfiltered results from this adapter.
Consider adding client-side post-filtering on the published date:
Proposed fix
async def search(
self,
query: str,
*,
max_results: int = 30,
year_from: Optional[int] = None,
year_to: Optional[int] = None,
) -> List[PaperCandidate]:
records = await asyncio.to_thread(
self._connector.search, query=query, max_results=max_results
)
- return [self._to_candidate(r) for r in records]
+ candidates = [self._to_candidate(r) for r in records]
+ if year_from is not None or year_to is not None:
+ candidates = [
+ c for c in candidates
+ if _year_in_range(c.publication_date, year_from, year_to)
+ ]
+ return candidatesWhere _year_in_range extracts the year from the ISO date string and checks bounds.
🧰 Tools
🪛 Ruff (0.15.0)
[warning] 28-28: Unused method argument: year_from
(ARG002)
[warning] 29-29: Unused method argument: year_to
(ARG002)
🤖 Prompt for AI Agents
In `@src/paperbot/infrastructure/adapters/arxiv_search_adapter.py` around lines 23
- 34, The search method currently ignores year_from/year_to because
_connector.search doesn't accept them; modify Paperbot's
ArxivSearchAdapter.search to post-filter the records returned by
self._connector.search by checking each record's published date year (use a
helper like _year_in_range(record) that extracts the year from the record's
published ISO date string and validates it against year_from/year_to) and only
convert remaining records with self._to_candidate; ensure the filter handles
missing/invalid published values gracefully (treat as out-of-range or skip) and
preserve max_results by applying the year filter before slicing to max_results.
| def resolve_any(self, external_id: str) -> Optional[int]: | ||
| """Resolve an external_id across all sources.""" | ||
| with self._provider.session() as session: | ||
| row = session.execute( | ||
| select(PaperIdentifierModel).where( | ||
| PaperIdentifierModel.external_id == external_id, | ||
| ) | ||
| ).scalar_one_or_none() | ||
| return int(row.paper_id) if row else None |
There was a problem hiding this comment.
resolve_any will raise MultipleResultsFound if external_id is not unique across sources.
The unique constraint is on (source, external_id), not external_id alone. If the same external_id string exists for different sources, scalar_one_or_none() throws sqlalchemy.exc.MultipleResultsFound.
Proposed fix
def resolve_any(self, external_id: str) -> Optional[int]:
"""Resolve an external_id across all sources."""
with self._provider.session() as session:
row = session.execute(
select(PaperIdentifierModel).where(
PaperIdentifierModel.external_id == external_id,
)
- ).scalar_one_or_none()
+ ).scalars().first()
return int(row.paper_id) if row else None📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def resolve_any(self, external_id: str) -> Optional[int]: | |
| """Resolve an external_id across all sources.""" | |
| with self._provider.session() as session: | |
| row = session.execute( | |
| select(PaperIdentifierModel).where( | |
| PaperIdentifierModel.external_id == external_id, | |
| ) | |
| ).scalar_one_or_none() | |
| return int(row.paper_id) if row else None | |
| def resolve_any(self, external_id: str) -> Optional[int]: | |
| """Resolve an external_id across all sources.""" | |
| with self._provider.session() as session: | |
| row = session.execute( | |
| select(PaperIdentifierModel).where( | |
| PaperIdentifierModel.external_id == external_id, | |
| ) | |
| ).scalars().first() | |
| return int(row.paper_id) if row else None |
🤖 Prompt for AI Agents
In `@src/paperbot/infrastructure/stores/identity_store.py` around lines 84 - 92,
The current resolve_any method uses scalar_one_or_none() which raises
MultipleResultsFound when the same external_id exists across different sources;
change the query to fetch the first matching row instead (use the
session.execute(...).scalars().first() pattern) and return int(row.paper_id) if
row else None; optionally add an explicit order_by on
PaperIdentifierModel.source to make which source wins deterministic. Ensure
references to resolve_any and PaperIdentifierModel are updated accordingly.
| # Canonical FK (dual-write migration — will replace paper_id + paper_ref_id) | ||
| canonical_paper_id: Mapped[Optional[int]] = mapped_column( | ||
| Integer, nullable=True, index=True | ||
| ) |
There was a problem hiding this comment.
canonical_paper_id lacks a ForeignKey constraint.
Unlike paper_ref_id (Line 464 which has ForeignKey("papers.id")), canonical_paper_id has no FK constraint. This means the DB won't enforce referential integrity — stale or invalid IDs can accumulate silently.
If the omission is intentional for the migration phase (to allow writing IDs before the referenced paper exists), consider adding the FK once the backfill is verified and the flag is enabled. A TODO comment documenting this intent would help.
Suggested improvement
# Canonical FK (dual-write migration — will replace paper_id + paper_ref_id)
canonical_paper_id: Mapped[Optional[int]] = mapped_column(
- Integer, nullable=True, index=True
+ Integer, ForeignKey("papers.id"), nullable=True, index=True
)Or, if intentionally deferred:
# Canonical FK (dual-write migration — will replace paper_id + paper_ref_id)
# TODO: Add ForeignKey("papers.id") after backfill is verified and PAPERBOT_USE_CANONICAL_FK is stable
canonical_paper_id: Mapped[Optional[int]] = mapped_column(
Integer, nullable=True, index=True
)🤖 Prompt for AI Agents
In `@src/paperbot/infrastructure/stores/models.py` around lines 469 - 472, The
canonical_paper_id column is missing a ForeignKey constraint, so either add
ForeignKey("papers.id") to the mapped_column for canonical_paper_id (mirroring
paper_ref_id) to enforce referential integrity, or if the omission is
intentional during the dual-write migration, annotate the field with a clear
TODO comment (mentioning adding ForeignKey("papers.id") after backfill and
referencing the PAPERBOT_USE_CANONICAL_FK flag) so future readers know this is
deliberate; update the canonical_paper_id Mapped definition accordingly.
|
|
||
| # Dual-write: also populate paper_identifiers | ||
| self._sync_identifiers(session, row) | ||
|
|
||
| return payload | ||
|
|
||
| @staticmethod | ||
| def _sync_identifiers(session, row: PaperModel) -> None: | ||
| """Write known external IDs to paper_identifiers (idempotent).""" | ||
| pairs: list[tuple[str, str]] = [] | ||
| if row.semantic_scholar_id: | ||
| pairs.append(("semantic_scholar", row.semantic_scholar_id)) | ||
| if row.arxiv_id: | ||
| pairs.append(("arxiv", row.arxiv_id)) | ||
| if row.openalex_id: | ||
| pairs.append(("openalex", row.openalex_id)) | ||
| if row.doi: | ||
| pairs.append(("doi", row.doi)) | ||
|
|
||
| for source, eid in pairs: | ||
| existing = session.execute( | ||
| select(PaperIdentifierModel).where( | ||
| PaperIdentifierModel.source == source, | ||
| PaperIdentifierModel.external_id == eid, | ||
| ) | ||
| ).scalar_one_or_none() | ||
| if existing is None: | ||
| session.add( | ||
| PaperIdentifierModel( | ||
| paper_id=row.id, | ||
| source=source, | ||
| external_id=eid, | ||
| created_at=_utcnow(), | ||
| ) | ||
| ) | ||
| elif existing.paper_id != row.id: | ||
| existing.paper_id = row.id | ||
| try: | ||
| session.flush() | ||
| except Exception: | ||
| session.rollback() |
There was a problem hiding this comment.
_sync_identifiers flushes but never commits — identifier rows may be silently lost.
In upsert_paper, session.commit() runs at Line 233. After that, _sync_identifiers (Line 240) adds rows and calls session.flush() (Line 276). These flushed rows live in a new implicit transaction. When the with block exits at return payload (Line 242), if the session context manager doesn't auto-commit, this transaction is rolled back and the identifiers are lost.
Additionally, the except Exception: session.rollback() at Lines 277-278 is dangerous in the upsert_papers_batch path (Line 460), where multiple papers share a single session — a rollback here undoes the entire batch's flushed state.
Proposed fix — commit after sync, and isolate rollback in batch
# Dual-write: also populate paper_identifiers
self._sync_identifiers(session, row)
+ session.commit()
return payload
`@staticmethod`
def _sync_identifiers(session, row: PaperModel) -> None:
"""Write known external IDs to paper_identifiers (idempotent)."""
pairs: list[tuple[str, str]] = []
if row.semantic_scholar_id:
pairs.append(("semantic_scholar", row.semantic_scholar_id))
if row.arxiv_id:
pairs.append(("arxiv", row.arxiv_id))
if row.openalex_id:
pairs.append(("openalex", row.openalex_id))
if row.doi:
pairs.append(("doi", row.doi))
for source, eid in pairs:
existing = session.execute(
select(PaperIdentifierModel).where(
PaperIdentifierModel.source == source,
PaperIdentifierModel.external_id == eid,
)
).scalar_one_or_none()
if existing is None:
session.add(
PaperIdentifierModel(
paper_id=row.id,
source=source,
external_id=eid,
created_at=_utcnow(),
)
)
elif existing.paper_id != row.id:
existing.paper_id = row.id
try:
session.flush()
- except Exception:
- session.rollback()
+ except Exception as exc:
+ logger.warning("Failed to sync identifiers for paper %s: %s", row.id, exc)
+ session.rollback()🧰 Tools
🪛 Ruff (0.15.0)
[warning] 277-277: Do not catch blind exception: Exception
(BLE001)
🤖 Prompt for AI Agents
In `@src/paperbot/infrastructure/stores/paper_store.py` around lines 238 - 278,
_sync_identifiers currently flushes identifier rows but never commits, and its
broad except does session.rollback() which can undo unrelated work in
upsert_papers_batch; fix by making the identifier write atomic and not able to
rollback the whole outer transaction: inside _sync_identifiers use a nested
transaction/savepoint (session.begin_nested()) when inserting
PaperIdentifierModel rows, flush/commit that nested transaction (or let it
close) and on exception rollback only the nested transaction and re-raise the
error (do not call session.rollback() on the outer session); alternatively, call
session.commit() after the identifier sync when called from single-upsert paths
like upsert_paper, but for batch paths prefer the begin_nested() approach and
remove the unconditional session.rollback() so upsert_papers_batch is not
invalidated.
| session.flush() | ||
|
|
||
| # Dual-write identifiers for all papers in this batch | ||
| for paper in papers: | ||
| existing = self._find_existing(session, paper) | ||
| if existing: | ||
| self._sync_identifiers(session, existing) | ||
|
|
There was a problem hiding this comment.
Redundant _find_existing queries and rollback risk in batch dual-write.
After session.flush() at Line 454, each paper is re-queried via _find_existing (Line 458) to locate the row for _sync_identifiers. This is an extra N DB round-trips when the models are already in the session's identity map.
More critically, if _sync_identifiers triggers a rollback (Line 278), it undoes the entire batch's flushed state — including all paper inserts/updates from lines 437-451.
Consider tracking the models during the loop and passing them directly:
Proposed fix sketch
+ touched: list[PaperModel] = []
for paper in papers:
...
if existing:
self._update_paper(existing, paper, now)
+ touched.append(existing)
updated_count += 1
else:
model = self._create_model(paper, now)
session.add(model)
+ touched.append(model)
new_count += 1
session.flush()
- for paper in papers:
- existing = self._find_existing(session, paper)
- if existing:
- self._sync_identifiers(session, existing)
+ for model in touched:
+ self._sync_identifiers(session, model)
session.commit()🤖 Prompt for AI Agents
In `@src/paperbot/infrastructure/stores/paper_store.py` around lines 454 - 461,
After session.flush(), avoid re-querying each paper with _find_existing (extra N
round-trips) and the risk that _sync_identifiers() rollback will undo the whole
batch; instead use the ORM model instances you already have (the items in papers
or the objects returned by the upsert/insert logic) and pass those directly into
_sync_identifiers to dual-write identifiers, and wrap each per-paper identifier
sync in a DB savepoint/independent transaction (e.g., session.begin_nested() or
equivalent) or catch and handle exceptions locally so a failing
_sync_identifiers does not rollback the entire batch. Ensure you remove the call
sites to _find_existing in the loop and reference the existing paper instances
when calling _sync_identifiers.
| def _get_user_library_canonical( | ||
| self, | ||
| user_id: str, | ||
| *, | ||
| session_provider, | ||
| track_id: Optional[int] = None, | ||
| actions: List[str], | ||
| sort_by: str, | ||
| sort_order: str, | ||
| limit: int, | ||
| offset: int, | ||
| ) -> Tuple[List[LibraryPaper], int]: | ||
| """Single FK JOIN path using paper_feedback.canonical_paper_id.""" | ||
| with session_provider.session() as session: | ||
| stmt = ( | ||
| select(PaperModel, PaperFeedbackModel) | ||
| .join( | ||
| PaperFeedbackModel, | ||
| PaperModel.id == PaperFeedbackModel.canonical_paper_id, | ||
| ) | ||
| .where( | ||
| PaperFeedbackModel.user_id == user_id, | ||
| PaperFeedbackModel.action.in_(actions), | ||
| PaperFeedbackModel.canonical_paper_id.is_not(None), | ||
| PaperModel.deleted_at.is_(None), | ||
| ) | ||
| ) | ||
| if track_id is not None: | ||
| stmt = stmt.where(PaperFeedbackModel.track_id == track_id) | ||
|
|
||
| all_results = session.execute(stmt).all() | ||
|
|
||
| paper_map: Dict[int, Tuple[PaperModel, PaperFeedbackModel]] = {} | ||
| for row in all_results: | ||
| paper, feedback = row[0], row[1] | ||
| if paper.id not in paper_map or feedback.ts > paper_map[paper.id][1].ts: | ||
| paper_map[paper.id] = (paper, feedback) | ||
|
|
||
| unique_results = list(paper_map.values()) | ||
| 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") | ||
| ) | ||
| elif sort_by == "citation_count": | ||
| unique_results.sort( | ||
| key=lambda x: x[0].citation_count 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") | ||
| ) | ||
|
|
||
| total = len(unique_results) | ||
| paginated = unique_results[offset : offset + limit] | ||
| return [ | ||
| LibraryPaper( | ||
| paper=row[0], saved_at=row[1].ts, track_id=row[1].track_id, action=row[1].action | ||
| ) | ||
| for row in paginated | ||
| ], total |
There was a problem hiding this comment.
_get_user_library_canonical is missing "title" and "year" sort options.
The non-canonical get_user_library (Lines 791-802) supports sorting by "title" and "year", but _get_user_library_canonical only handles "saved_at", "citation_count", and a default fallback. Users toggling PAPERBOT_USE_CANONICAL_FK=true will silently lose sort-by-title and sort-by-year behavior.
Proposed fix
if sort_by == "saved_at":
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")
+ )
+ elif sort_by == "year":
+ unique_results.sort(
+ key=lambda x: x[0].year or 0, 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")
)
else:🤖 Prompt for AI Agents
In `@src/paperbot/infrastructure/stores/paper_store.py` around lines 824 - 884,
_get_user_library_canonical currently only sorts by "saved_at", "citation_count"
or default and therefore drops support for "title" and "year" sorting; update
the sorting branch in _get_user_library_canonical to handle sort_by == "title"
(sort by paper.title, using a case-insensitive fallback like "" for None) and
sort_by == "year" (sort by paper.year with a numeric fallback like 0), honoring
sort_order ("asc"/"desc") just like the other branches; operate on
unique_results (tuples of (PaperModel, PaperFeedbackModel)) and use x[0].title /
x[0].year and the existing min_ts logic for saved_at so behavior matches
get_user_library.
概述
按 DDD 上下文拆分 + Hexagonal Architecture 重构数据层与模块，统一 Research / DailyPaper / Papers Library 三个模块的数据管道。
改动内容
P0: 身份统一（数据层根基）
paper_identifiers表 (Alembic 0009)：统一(source, external_id) → papers.id映射，替代原来paper_feedback.paper_id存外部 ID 的四路 OR JOINPaperIdentity值对象 +IdentityStoreCRUD +IdentityResolver服务（从_resolve_paper_ref_id提取）paper_store.upsert_paper()同时写paper_identifiers；research_store.add_paper_feedback()同时写canonical_paper_idPAPERBOT_USE_CANONICAL_FK：get_user_library()可切换到单路 FK JOINscripts/backfill_identifiers.pyP1: PaperSearchService 上线
SearchPortProtocol + 5 个 adapter（S2、arXiv、papers.cool、HF Daily、OpenAlex）PaperSearchService门面：并发扇出搜索 → 去重 → 可选持久化ContextEngine重构：优先使用PaperSearchService，保留SemanticScholarSearch兜底RegistryPort、FeedbackPort、EnrichmentPortP2: Enrichment 共享
EnrichmentPipeline（Chain of Responsibility）：可插拔的 Judge / Summary / Repo 步骤Domain 值对象
PaperCandidate（ACL 输出）、FeedbackAction、JudgeScore、LLMSummary、ResearchTrackBug Fix
_resolve_paper_ref_id中引用不存在的PaperModel.external_url列测试
pytest tests/unit/ -q→ 324 passed, 0 failed迁移步骤
alembic upgrade headpython scripts/backfill_identifiers.pyPAPERBOT_USE_CANONICAL_FK=true切换读路径Summary by CodeRabbit
New Features
Chores