diff --git a/backend/src/alphapulse/news_research/api.py b/backend/src/alphapulse/news_research/api.py index 7c9829a..c63ca67 100644 --- a/backend/src/alphapulse/news_research/api.py +++ b/backend/src/alphapulse/news_research/api.py @@ -19,6 +19,7 @@ from alphapulse.news_research.models import ( Membership, Version, Reference, + BatchEvent, ) from alphapulse.news_research import service, batches, vocabulary as vocab from alphapulse.news_research.models import VocabularyVersion @@ -496,7 +497,13 @@ async def retry_event(identifier: int, session: DB): raise HTTPException(404, "事件不存在") if row.state == "processing": raise HTTPException(409, "事件正在处理") - if row.state == "done": + deferred = None + if row.state == "pending": + deferred = await session.scalar(select(BatchEvent.batch_id).where( + BatchEvent.event_id == row.id, BatchEvent.revision == row.revision, + BatchEvent.state == "deferred").limit(1)) + if row.state == "done" or deferred is not None: + # Escape a deferred revision only on explicit retry; ordinary retries reuse evidence. service.dirty(row, True) else: row.state = "pending" diff --git a/backend/src/alphapulse/news_research/batches.py b/backend/src/alphapulse/news_research/batches.py index 6c6c64f..8ef87ef 100644 --- a/backend/src/alphapulse/news_research/batches.py +++ b/backend/src/alphapulse/news_research/batches.py @@ -1,7 +1,7 @@ """Durable admission barriers: finish this batch's assignments before any research.""" from collections import Counter from datetime import UTC, datetime -from sqlalchemy import select, or_, exists, func +from sqlalchemy import select, or_, and_, exists, func, any_ from alphapulse.models import NewsItem from alphapulse.news_research.models import Batch, BatchPackage, BatchEvent, Package, Event, Membership, Attempt from alphapulse.news_research.service import enroll @@ -10,8 +10,8 @@ from alphapulse.news_research.service import enroll async def freeze(session, settings): cutoff = datetime.now(UTC) ids = list(await session.scalars(select(NewsItem.id).outerjoin(Package, Package.news_id == NewsItem.id).where( - NewsItem.published_at >= settings.start_at, NewsItem.published_at <= cutoff, - or_(Package.id.is_(None), Package.content_hash != NewsItem.content_hash)))) + or_(and_(NewsItem.published_at >= settings.start_at, NewsItem.published_at <= cutoff, Package.id.is_(None)), + and_(Package.id.is_not(None), Package.content_hash != NewsItem.content_hash))))) existing_ids = list(await session.scalars(select(Package.id).where( Package.state.in_(['index_pending', 'assign_pending', 'index_failed', 'assign_failed'])))) event_cutoff = list(await session.scalars(select(Event.id).where(Event.state == "pending"))) @@ -21,12 +21,12 @@ async def freeze(session, settings): BatchPackage.package_id == Package.id, BatchPackage.content_hash == Package.content_hash, BatchPackage.state == 'failed')) packages = list(await session.scalars(select(Package).where( - or_(Package.id.in_(existing_ids), Package.news_id.in_(ids)), or_( + or_(Package.id == any_(existing_ids), Package.news_id == any_(ids)), or_( Package.state.in_(['index_pending', 'assign_pending']), Package.state.in_(['index_failed', 'assign_failed']) & ~previous_failure)))) deferred = exists(select(BatchEvent.batch_id).where(BatchEvent.event_id == Event.id, BatchEvent.revision == Event.revision, BatchEvent.state == 'deferred')) - events = list(await session.scalars(select(Event).where(Event.id.in_(event_cutoff), Event.state == 'pending', ~deferred))) + events = list(await session.scalars(select(Event).where(Event.id == any_(event_cutoff), Event.state == 'pending', ~deferred))) if not packages and not events: return None batch = Batch(phase='assigning', cutoff_at=cutoff) @@ -37,8 +37,8 @@ async def freeze(session, settings): event_ids = {e.id for e in events} if packages: event_ids.update(await session.scalars(select(Membership.event_id).where( - Membership.package_id.in_([p.id for p in packages])))) - for e in await session.scalars(select(Event).where(Event.id.in_(event_ids))): + Membership.package_id == any_([p.id for p in packages])))) + for e in await session.scalars(select(Event).where(Event.id == any_(list(event_ids)))): session.add(BatchEvent(batch_id=batch.id, event_id=e.id, revision=e.revision, state='pending')) await session.commit() return batch @@ -73,7 +73,7 @@ async def actions(session, batch, active, slots): ids = set(await session.scalars(select(Membership.event_id).join( BatchPackage, BatchPackage.package_id == Membership.package_id).where(BatchPackage.batch_id == batch.id))) existing = {x.event_id for x in await session.scalars(select(BatchEvent).where(BatchEvent.batch_id == batch.id))} - for e in await session.scalars(select(Event).where(Event.id.in_(ids - existing))): + for e in await session.scalars(select(Event).where(Event.id == any_(list(ids - existing)))): session.add(BatchEvent(batch_id=batch.id, event_id=e.id, revision=e.revision, state='pending')) await session.flush() latest = select(func.max(Attempt.id)).join(BatchPackage, BatchPackage.package_id == Attempt.package_id).where( diff --git a/backend/src/alphapulse/news_research/service.py b/backend/src/alphapulse/news_research/service.py index a87825d..76d9118 100644 --- a/backend/src/alphapulse/news_research/service.py +++ b/backend/src/alphapulse/news_research/service.py @@ -5,7 +5,7 @@ import json import re from datetime import UTC, datetime, timedelta from types import SimpleNamespace -from sqlalchemy import select, update, func, or_, case, column, exists, true, literal +from sqlalchemy import select, update, func, or_, case, column, exists, true, literal, any_ from sqlalchemy.dialects.postgresql import insert from alphapulse.core.db import SessionLocal from alphapulse.models import NewsItem @@ -233,7 +233,7 @@ async def model_config(session, config, stage): async def enroll(session, ids): - news = list(await session.scalars(select(NewsItem).where(NewsItem.id.in_(ids)))) + news = list(await session.scalars(select(NewsItem).where(NewsItem.id == any_(list(ids))))) for row in news: await session.execute( insert(Package) diff --git a/backend/tests/test_news_research_workspace.py b/backend/tests/test_news_research_workspace.py index 3df28ba..095da53 100644 --- a/backend/tests/test_news_research_workspace.py +++ b/backend/tests/test_news_research_workspace.py @@ -124,22 +124,22 @@ def test_new_event_waits_for_aggregation_unless_urgent(trading, urgent): assert event.due_at - event.dirty_since == timedelta(minutes=0 if urgent else 5) -@pytest.mark.parametrize("state", ["failed", "pending", "done"]) -def test_event_retry_preserves_content_revision_until_content_changes(trading, state): +@pytest.mark.parametrize("state,deferred", [("failed", False), ("pending", False), ("pending", True), ("done", False)]) +def test_event_retry_preserves_content_revision_unless_done_or_deferred(trading, state, deferred): import asyncio from unittest.mock import AsyncMock from alphapulse.news_research.api import retry_event event = N( - revision=7, + id=1, revision=7, state=state, dirty_since=None, due_at=datetime.now(UTC), error="Previous failure", ) - session = N(get=AsyncMock(return_value=event), commit=AsyncMock()) + session = N(get=AsyncMock(return_value=event), scalar=AsyncMock(return_value=12 if deferred else None), commit=AsyncMock()) asyncio.run(retry_event(1, session)) - assert event.revision == (8 if state == "done" else 7) + assert event.revision == (8 if state == "done" or deferred else 7) assert event.state == "pending" and event.error is None assert event.due_at <= datetime.now(UTC) diff --git a/scripts/smoke_news_batches.py b/scripts/smoke_news_batches.py index 06a2ed0..60c1c28 100644 --- a/scripts/smoke_news_batches.py +++ b/scripts/smoke_news_batches.py @@ -4,11 +4,12 @@ import os from datetime import UTC, datetime, timedelta from types import SimpleNamespace from urllib.parse import urlparse -from sqlalchemy import select, delete +from sqlalchemy import select, delete, insert, func, text from alphapulse.core.db import SessionLocal, engine from alphapulse.models import NewsItem from alphapulse.news_research.models import Batch, BatchPackage, BatchEvent, Package, Event, Membership, Attempt -from alphapulse.news_research import batches +from alphapulse.news_research import batches, service +from alphapulse.news_research.api import retry_event async def main(): @@ -86,8 +87,82 @@ async def main(): for model in (BatchEvent, BatchPackage, Batch, Membership, Attempt, Package, Event, NewsItem): await s.execute(delete(model)) await s.commit() + await regressions(now, settings) await engine.dispose() print('Frozen batches passed: assignment barrier, single research, failures, new arrivals, restart and source revisions.') +async def regressions(now, settings): + async def clear(session): + # This script runs first in an empty, disposable CI database. TRUNCATE avoids + # quadratic foreign-key checks when removing the 33,000-row stress fixture. + await session.execute(text('TRUNCATE news_research_batches, news_research_packages, news_research_events, news_items CASCADE')) + await session.commit() + + async with SessionLocal() as s: + event = Event(title='Deferred retry', tags=[], revision=7, state='pending', due_at=now) + previous = Batch(phase='partial', cutoff_at=now) + s.add_all([event, previous]) + await s.flush() + s.add(BatchEvent(batch_id=previous.id, event_id=event.id, revision=7, state='deferred')) + await s.commit() + assert await batches.freeze(s, settings) is None + await retry_event(event.id, s) + assert event.revision == 8 + batch = await batches.freeze(s, settings) + assert batch is not None + await batches.actions(s, batch, set(), 8) + await s.commit() + assert await batches.actions(s, batch, set(), 8) == [('event', event.id)] + # Failed retries retain their revision so cached evidence can still be reused. + event.state = 'failed' + await s.commit() + await retry_event(event.id, s) + assert event.revision == 8 and event.state == 'pending' + await clear(s) + + def original(key, published_at): + return dict(source_code='wscn_news', external_id=key, title=key, summary='', + content_text='Original', content_hash='a'*64, url='https://example.invalid/'+key, + published_at=published_at) + news = NewsItem(**original('manual-history', now-timedelta(days=365))) + excluded = NewsItem(**original('unselected-history', now-timedelta(days=365))) + s.add_all([news, excluded]) + await s.commit() + await service.enroll(s, [news.id]) + batch = await batches.freeze(s, settings) + package = await s.scalar(select(Package).where(Package.news_id == news.id)) + news.content_hash = 'b'*64 + await s.commit() + await batches.actions(s, batch, set(), 8) + await s.commit() + await batches.actions(s, batch, set(), 8) + await s.commit() + assert batch.phase == 'partial' + batch = await batches.freeze(s, settings) + assert package.content_hash == news.content_hash and package.state == 'index_pending' + assert await batches.actions(s, batch, set(), 8) == [('index', package.id)] + assert await s.scalar(select(func.count()).select_from(Package)) == 1 + # A previously completed manual original also refreshes outside the automatic boundary. + package.state = 'done' + news.content_hash = 'c'*64 + await s.commit() + await batches.freeze(s, settings) + assert package.content_hash == news.content_hash and package.state == 'index_pending' + await clear(s) + + count = 33000 + await s.execute(insert(NewsItem), [original('bulk-'+str(i), now) for i in range(count)]) + await s.commit() + batch = await batches.freeze(s, settings) + assert await s.scalar(select(func.count()).select_from(BatchPackage).where(BatchPackage.batch_id == batch.id)) == count + assert await s.scalar(select(func.count()).select_from(Package)) == count + # Exercise the existing-pending-ID and membership arrays above the driver limit too. + next_batch = await batches.freeze(s, settings) + assert await s.scalar(select(func.count()).select_from(BatchPackage).where(BatchPackage.batch_id == next_batch.id)) == count + assert len(await batches.actions(s, next_batch, set(), 8)) == 8 + await clear(s) + print('Batch regressions passed: deferred retries, failed-evidence reuse, manual source revisions, and 33,000 originals without parameter overflow.') + + asyncio.run(main())