diff --git a/backend/src/alphapulse/news_research/api.py b/backend/src/alphapulse/news_research/api.py index 35c05fd..48920d5 100644 --- a/backend/src/alphapulse/news_research/api.py +++ b/backend/src/alphapulse/news_research/api.py @@ -19,10 +19,8 @@ from alphapulse.news_research.models import ( Membership, Version, Reference, - BatchEvent, - Batch, ) -from alphapulse.news_research import service, batches, scheduling, vocabulary as vocab +from alphapulse.news_research import service, scheduling, vocabulary as vocab, coverage as reports from alphapulse.news_research.models import VocabularyVersion from alphapulse.news_research.contracts import ( INDEX_ASSIGN_SKILL, @@ -120,7 +118,7 @@ async def settings(session: DB): {"config": row.config, "enabled": row.enabled, "start_at": row.start_at, "control_revision": row.control_revision, "realtime_from": row.realtime_from, "timezone": "Asia/Shanghai", "queue": await scheduling.queue_status(session, row), - "batch": await batches.progress(session)} + "batch": None} ) @@ -131,7 +129,7 @@ async def coverage(session: DB, start_at: AwareDatetime | None = None): if start > datetime.now(UTC): raise HTTPException(422, "回补起始时间不能晚于当前时间") effective = min(start, row.realtime_from) if row.realtime_from else start - report = await batches.coverage(session, effective) + report = await reports.coverage(session, effective) return success({**report, 'backfill_start_at': start, 'realtime_from': row.realtime_from}) @@ -268,11 +266,9 @@ async def enable(session: DB, body: EnableInput | None = None): ) if row.control_revision != control_revision or row.config != original_config: raise HTTPException(409, "验证期间已暂停或设置变化，请重新验证") - report = await batches.coverage(session, start_at) + report = await reports.coverage(session, start_at) if row.realtime_from is None: row.realtime_from = datetime.now(UTC) - await session.execute(update(Batch).where(Batch.phase.in_(["assigning", "researching"])) - .values(phase="rescheduled", finished_at=datetime.now(UTC))) row.enabled = True row.control_revision += 1 await session.commit() @@ -497,12 +493,16 @@ async def event(identifier: int, session: DB): if row.state == 'pending': settings = await settings_row(session) unfinished = [(p, n) for p, n in members if p.state != 'done' or p.content_hash != n.content_hash] - if any(n.published_at < scheduling.boundary(settings) for p, n in unfinished): + if any(v.state == 'processing' for v in versions): + wait_reason = '等待上一轮研究完成后处理事件新变化' + elif any(n.published_at < scheduling.boundary(settings) for p, n in unfinished): wait_reason = '范围外成员尚未完成，扩大回补范围后可继续研究' elif any(p.state.endswith('_failed') for p, n in unfinished): wait_reason = '成员包处理失败，需手动重试后才能研究' elif unfinished: wait_reason = '等待成员包完成关键词与事件归属' + elif await session.scalar(select(Event.id).where(Event.id == row.id, scheduling.manual_review())): + wait_reason = '当前事件版本需手动重试，不会自动重复调用' elif row.due_at > datetime.now(UTC): wait_reason = '正在汇集事件变化，聚合窗口结束后研究' latest = next((v for v in versions if v.state == "done"), None) @@ -544,9 +544,8 @@ async def retry_event(identifier: int, session: DB): raise HTTPException(409, "事件正在处理") 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)) + deferred = await session.scalar(select(Event.id).where( + Event.id == row.id, scheduling.manual_review())) 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) diff --git a/backend/src/alphapulse/news_research/batches.py b/backend/src/alphapulse/news_research/batches.py deleted file mode 100644 index 273c878..0000000 --- a/backend/src/alphapulse/news_research/batches.py +++ /dev/null @@ -1,191 +0,0 @@ -"""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_, and_, exists, func, any_, case -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 -from alphapulse.news_research.scheduling import scoped_event, ready_event - - -async def coverage(session, start_at, cutoff=None): - """Count each original once, including unfinished research after assignment.""" - cutoff = cutoff or datetime.now(UTC) - research = select( - Membership.package_id, - func.bool_or(Event.state == 'failed').label('failed'), - func.bool_or(Event.state == 'processing').label('processing'), - func.bool_or(Event.state.not_in(['done', 'retired'])).label('unfinished'), - ).join(Event, Event.id == Membership.event_id).group_by(Membership.package_id).subquery() - state = case( - (Package.id.is_(None), 'missing'), - (Package.content_hash != NewsItem.content_hash, 'stale'), - (Package.state.in_(['index_failed', 'assign_failed']), 'failed'), - (Package.state.in_(['index_processing', 'assign_processing']), 'processing'), - (Package.state != 'done', 'pending'), - (research.c.failed, 'failed'), - (research.c.processing, 'processing'), - (or_(research.c.package_id.is_(None), research.c.unfinished), 'pending'), - else_='completed', - ) - in_range = and_(NewsItem.published_at >= start_at, NewsItem.published_at <= cutoff) - rows = (await session.execute( - select(in_range, state, func.count()).select_from(NewsItem) - .outerjoin(Package, Package.news_id == NewsItem.id) - .outerjoin(research, research.c.package_id == Package.id) - .where(or_(in_range, Package.id.is_not(None))) - .group_by(in_range, state) - )).all() - counts = {status: count for selected, status, count in rows if selected} - outside_counts = {status: count for selected, status, count in rows if not selected} - outside = {name: outside_counts.get(name, 0) for name in - ('stale', 'pending', 'processing', 'failed')} - result = {name: counts.get(name, 0) for name in - ('missing', 'stale', 'pending', 'processing', 'failed', 'completed')} - return {'start_at': start_at, 'checked_at': cutoff, 'total': sum(result.values()), - 'unprocessed': sum(result.values()) - result['completed'], **result, - 'outside': {'unprocessed': sum(outside.values()), **outside}} - - -async def freeze(session, settings): - cutoff = datetime.now(UTC) - scope = [NewsItem.published_at >= settings.start_at, NewsItem.published_at <= cutoff] - realtime_from = getattr(settings, 'realtime_from', None) - if realtime_from is not None: - scope.append(NewsItem.published_at < realtime_from) - candidates = select(NewsItem.id).outerjoin(Package, Package.news_id == NewsItem.id).where( - *scope, or_(Package.id.is_(None), Package.content_hash != NewsItem.content_hash, - Package.state.in_(['index_pending', 'assign_pending']))) - ids = list(await session.scalars(candidates.order_by(NewsItem.published_at, NewsItem.id).limit(30))) - await enroll(session, ids) - existing_ids = list(await session.scalars(select(Package.id).where(Package.news_id.in_(ids)))) - deferred = exists(select(BatchEvent.batch_id).where(BatchEvent.event_id == Event.id, - BatchEvent.revision == Event.revision, BatchEvent.state == 'deferred')) - event_query = select(Event.id).where(Event.state == 'pending', scoped_event(settings), ready_event(), Event.due_at <= cutoff, ~deferred) - if realtime_from is not None: - event_query = event_query.where(~Event.realtime_pending) - event_cutoff = list(await session.scalars(event_query.order_by(Event.due_at, Event.id).limit(30))) - # Failed items are reported once; only explicit retries or new source revisions rerun them. - previous_failure = exists(select(BatchPackage.batch_id).where( - 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 == 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)))) - 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) - session.add(batch) - await session.flush() - session.add_all(BatchPackage(batch_id=batch.id, package_id=p.id, content_hash=p.content_hash, - state='failed' if p.state.endswith('_failed') else 'pending') for p in packages) - event_ids = {e.id for e in events} - if packages: - event_ids.update(await session.scalars(select(Membership.event_id).where( - 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 - - -async def actions(session, batch, active, slots, settings=None): - """Return stage/ID pairs; callers commit before launching and never overlap phases.""" - if batch.phase == 'assigning': - rows = list(await session.execute(select(BatchPackage, Package, NewsItem).join( - Package, Package.id == BatchPackage.package_id).join(NewsItem, NewsItem.id == Package.news_id) - .where(BatchPackage.batch_id == batch.id).order_by(NewsItem.published_at, NewsItem.id))) - result = [] - assigning = any(stage == 'assign' for stage, _ in active) - for item, package, news in rows: - if item.state != 'pending': - continue - if settings is not None and news.published_at < settings.start_at: - item.state = 'range_paused' - elif item.content_hash != news.content_hash or item.content_hash != package.content_hash: - item.state = 'deferred' - elif package.state == 'done': - item.state = 'done' - elif package.state.endswith('_failed'): - item.state = 'failed' - elif package.state in ('index_pending', 'assign_pending'): - stage = 'index' if package.state == 'index_pending' else 'assign' - if not any(identifier == package.id and task != 'event' for task, identifier in active) and len(result) < slots and (stage != 'assign' or not assigning): - result.append((stage, package.id)) - assigning |= stage == 'assign' - package_ids = {package.id for _, package, _ in rows} - batch_active = any(stage != 'event' and identifier in package_ids for stage, identifier in active) - if any(item.state == 'pending' for item, _, _ in rows) or batch_active: - return result - # Memberships now include events created by this batch. Existing affected events - # were frozen up front, including memberships that have since been removed. - 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 == 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( - BatchPackage.batch_id == batch.id, Attempt.stage == 'assign', Attempt.state == 'done').group_by(Attempt.package_id) - urgent = {int(event_id) for attempt in await session.scalars(select(Attempt).where(Attempt.id.in_(latest))) - for event_id, target in attempt.rules.get('membership_targets', {}).items() if target.get('urgent')} - for item, event in await session.execute(select(BatchEvent, Event).join(Event).where(BatchEvent.batch_id == batch.id)): - if item.event_id in urgent: - item.state = 'urgent_pending' - item.revision = event.revision - batch.phase = 'researching' - return [] - result = [] - rows = list(await session.execute(select(BatchEvent, Event).join(Event).where( - BatchEvent.batch_id == batch.id).order_by((BatchEvent.state == 'urgent_pending').desc(), Event.due_at, Event.id))) - for item, event in rows: - if item.state not in ('pending', 'urgent_pending', 'processing') or ('event', event.id) in active: - continue - if settings is not None and not await session.scalar(select(Event.id).where(Event.id == event.id, scoped_event(settings))): - item.state = 'range_paused' - elif settings is not None and event.realtime_pending: - item.state = 'realtime' - elif event.due_at > datetime.now(UTC): - item.state = 'aggregating' - elif item.revision != event.revision: - item.state = 'deferred' - elif event.state == 'done' or event.state == 'retired': - item.state = 'done' - elif event.state == 'failed': - item.state = 'failed' - elif item.state == 'processing': - item.state = 'deferred' # A completed call was superseded; do not spend again in this batch. - else: - invalid = await session.scalar(select(Membership.event_id).join(Package).join(NewsItem).where( - Membership.event_id == event.id, or_(Package.state != 'done', Package.content_hash != NewsItem.content_hash)).limit(1)) - if invalid: - item.state = 'blocked' - elif len(result) < slots: - item.state = 'processing' - result.append(('event', event.id)) - event_ids = {event.id for _, event in rows} - batch_active = any(stage == 'event' and identifier in event_ids for stage, identifier in active) - active_packages = {identifier for stage, identifier in active if stage != 'event'} - if active_packages: - batch_active |= bool(await session.scalar(select(BatchPackage.package_id).where( - BatchPackage.batch_id == batch.id, BatchPackage.package_id.in_(active_packages)).limit(1))) - if not result and not batch_active and all(item.state not in ('pending', 'urgent_pending', 'processing') for item, _ in rows): - bad = any(item.state in ('failed', 'deferred') for item, _ in rows) - bad |= bool(await session.scalar(select(BatchPackage.batch_id).where( - BatchPackage.batch_id == batch.id, BatchPackage.state.in_(['failed', 'deferred'])).limit(1))) - batch.phase = 'partial' if bad else 'done' - batch.finished_at = datetime.now(UTC) - return result - - -async def progress(session): - batch = await session.scalar(select(Batch).order_by(Batch.id.desc()).limit(1)) - if batch is None: - return None - result = {'id': batch.id, 'phase': batch.phase, 'cutoff_at': batch.cutoff_at} - for name, model in [('packages', BatchPackage), ('events', BatchEvent)]: - counts = Counter(await session.scalars(select(model.state).where(model.batch_id == batch.id))) - result.update({name + '_total': sum(counts.values()), **{name + '_' + s: counts[s] for s in ('done', 'failed', 'deferred', 'range_paused', 'blocked', 'realtime')}}) - return result diff --git a/backend/src/alphapulse/news_research/coverage.py b/backend/src/alphapulse/news_research/coverage.py new file mode 100644 index 0000000..13f8302 --- /dev/null +++ b/backend/src/alphapulse/news_research/coverage.py @@ -0,0 +1,44 @@ +"""Coverage reporting over durable news and research states.""" +from datetime import UTC, datetime +from sqlalchemy import select, or_, and_, func, case +from alphapulse.models import NewsItem +from alphapulse.news_research.models import Package, Event, Membership + + +async def coverage(session, start_at, cutoff=None): + """Count each original once, including unfinished research after assignment.""" + cutoff = cutoff or datetime.now(UTC) + research = select( + Membership.package_id, + func.bool_or(Event.state == 'failed').label('failed'), + func.bool_or(Event.state == 'processing').label('processing'), + func.bool_or(Event.state.not_in(['done', 'retired'])).label('unfinished'), + ).join(Event, Event.id == Membership.event_id).group_by(Membership.package_id).subquery() + state = case( + (Package.id.is_(None), 'missing'), + (Package.content_hash != NewsItem.content_hash, 'stale'), + (Package.state.in_(['index_failed', 'assign_failed']), 'failed'), + (Package.state.in_(['index_processing', 'assign_processing']), 'processing'), + (Package.state != 'done', 'pending'), + (research.c.failed, 'failed'), + (research.c.processing, 'processing'), + (or_(research.c.package_id.is_(None), research.c.unfinished), 'pending'), + else_='completed', + ) + in_range = and_(NewsItem.published_at >= start_at, NewsItem.published_at <= cutoff) + rows = (await session.execute( + select(in_range, state, func.count()).select_from(NewsItem) + .outerjoin(Package, Package.news_id == NewsItem.id) + .outerjoin(research, research.c.package_id == Package.id) + .where(or_(in_range, Package.id.is_not(None))) + .group_by(in_range, state) + )).all() + counts = {status: count for selected, status, count in rows if selected} + outside_counts = {status: count for selected, status, count in rows if not selected} + outside = {name: outside_counts.get(name, 0) for name in + ('stale', 'pending', 'processing', 'failed')} + result = {name: counts.get(name, 0) for name in + ('missing', 'stale', 'pending', 'processing', 'failed', 'completed')} + return {'start_at': start_at, 'checked_at': cutoff, 'total': sum(result.values()), + 'unprocessed': sum(result.values()) - result['completed'], **result, + 'outside': {'unprocessed': sum(outside.values()), **outside}} diff --git a/backend/src/alphapulse/news_research/scheduling.py b/backend/src/alphapulse/news_research/scheduling.py index 73c7443..5f143a6 100644 --- a/backend/src/alphapulse/news_research/scheduling.py +++ b/backend/src/alphapulse/news_research/scheduling.py @@ -1,8 +1,8 @@ """Priority selection over durable package and event states; no paid work here.""" from datetime import UTC, datetime -from sqlalchemy import select, exists, or_, and_, func, case +from sqlalchemy import select, exists, or_, and_, func, case, literal from alphapulse.models import NewsItem -from alphapulse.news_research.models import Package, Event, Membership, Attempt, Version, ResearchSettings +from alphapulse.news_research.models import Package, Event, Membership, Attempt, Version, ResearchSettings, BatchEvent def boundary(settings): @@ -47,57 +47,124 @@ async def mark_priority(session, event, news): event.realtime_pending = True -async def realtime_actions(session, settings, active, slots): - if settings.realtime_from is None or slots <= 0: - return [] - now = datetime.now(UTC) - busy_packages = {identifier for stage, identifier in active if stage != 'event'} - busy_events = {identifier for stage, identifier in active if stage == 'event'} - result = [] - events = await session.scalars(select(Event).where( - Event.state == 'pending', Event.realtime_pending, Event.due_at <= now, - scoped_event(settings), ready_event(), Event.id.not_in(busy_events), - ~exists(select(Version.id).where(Version.event_id == Event.id, Version.state == 'processing')), - ).order_by(Event.due_at, Event.id).limit(slots)) - result.extend(('event', event.id) for event in events) - # Known historical dependencies inherit the waiting realtime event's priority. +def manual_review(): + """Legacy deferred revisions and rejected current snapshots require explicit retry.""" + return or_( + exists(select(BatchEvent.event_id).where(BatchEvent.event_id == Event.id, + BatchEvent.revision == Event.revision, BatchEvent.state == 'deferred')), + exists(select(Version.id).where(Version.event_id == Event.id, + Version.revision == Event.revision, Version.state == 'superseded')), + ) + + +def priority_package(settings): + if settings.realtime_from is None: + return literal(False) dependency = exists(select(Membership.package_id).join(Event).where( Membership.package_id == Package.id, Event.realtime_pending, Event.state == 'pending')) - priority = or_(NewsItem.published_at >= settings.realtime_from, dependency) - candidates = select(NewsItem.id).outerjoin(Package, Package.news_id == NewsItem.id).where( - eligible(settings), priority, or_(Package.id.is_(None), Package.content_hash != NewsItem.content_hash), - or_(Package.id.is_(None), Package.id.not_in(busy_packages))).order_by(NewsItem.published_at, NewsItem.id).limit(30) - ids = list(await session.scalars(candidates)) - if ids: - from alphapulse.news_research.service import enroll - await enroll(session, ids) - assigning = any(stage == 'assign' for stage, _ in active) - for stage in ('assign', 'index'): - if stage == 'assign' and assigning: - continue - remaining = slots - len(result) - if remaining <= 0: - break - query = select(Package.id).join(NewsItem).where( - eligible(settings), priority, Package.state == stage + '_pending', - Package.id.not_in(busy_packages), - ~exists(select(Attempt.id).where(Attempt.package_id == Package.id, Attempt.state == 'processing')), - ).order_by(NewsItem.published_at, NewsItem.id).limit(1 if stage == 'assign' else remaining) - result.extend((stage, identifier) for identifier in await session.scalars(query)) + return or_(NewsItem.published_at >= settings.realtime_from, dependency) + + +def active_research(): + return exists(select(Version.id).where(Version.event_id == Event.id, Version.state == 'processing')) + + +def runnable_event(settings): + return and_(Event.state == 'pending', scoped_event(settings), ready_event(), + Event.due_at <= datetime.now(UTC), ~manual_review(), + ~active_research()) + + +async def lane_actions(session, settings, active, slots, realtime): + if slots <= 0 or (realtime and settings.realtime_from is None): + return [] + busy_packages = {identifier for stage, identifier in active if stage != 'event'} + busy_events = {identifier for stage, identifier in active if stage == 'event'} + scope = priority_package(settings) if realtime else ~priority_package(settings) + result = [] + # The chronological head includes missing and still-indexing originals. Failed + # source revisions are skipped, but changed originals become eligible again. + if not any(stage == 'assign' for stage, _ in active): + head = (await session.execute(select(Package.id, Package.state, Package.content_hash, + NewsItem.content_hash).select_from(NewsItem).outerjoin(Package, Package.news_id == NewsItem.id) + .where(eligible(settings), scope, or_(Package.id.is_(None), + Package.content_hash != NewsItem.content_hash, + Package.state.in_(['index_pending', 'index_processing', 'assign_pending', 'assign_processing']))) + .order_by(NewsItem.published_at, NewsItem.id).limit(1))).first() + if head and head[1] == 'assign_pending' and head[2] == head[3] and head[0] not in busy_packages: + if not await session.scalar(select(Attempt.id).where( + Attempt.package_id == head[0], Attempt.state == 'processing').limit(1)): + result.append(('assign', head[0])) + events = await session.scalars(select(Event.id).where( + runnable_event(settings), Event.realtime_pending == realtime, + Event.id.not_in(busy_events)).order_by(Event.due_at, Event.id).limit(slots - len(result))) + result.extend(('event', identifier) for identifier in events) + remaining = slots - len(result) + if remaining > 0: + ids = list(await session.scalars(select(NewsItem.id).outerjoin(Package, Package.news_id == NewsItem.id) + .where(eligible(settings), scope, + or_(Package.id.is_(None), Package.content_hash != NewsItem.content_hash, Package.state == 'index_pending'), + or_(Package.id.is_(None), Package.id.not_in(busy_packages)), + ~exists(select(Attempt.id).where(Attempt.package_id == Package.id, Attempt.state == 'processing'))) + .order_by(NewsItem.published_at, NewsItem.id).limit(remaining))) + if ids: + from alphapulse.news_research.service import enroll + await enroll(session, ids) + result.extend(('index', identifier) for identifier in await session.scalars( + select(Package.id).join(NewsItem).where(Package.news_id.in_(ids), Package.state == 'index_pending') + .order_by(NewsItem.published_at, NewsItem.id))) + return result + + +async def realtime_actions(session, settings, active, slots): + return await lane_actions(session, settings, active, slots, True) + + +async def actions(session, settings, active, slots): + result = await realtime_actions(session, settings, active, slots) + result += await lane_actions(session, settings, active | set(result), slots - len(result), False) return result async def queue_status(session, settings): - running = await session.scalar(select(func.count()).select_from(Attempt).where(Attempt.state == 'processing')) - running += await session.scalar(select(func.count()).select_from(Version).where(Version.state == 'processing')) - realtime_from = settings.realtime_from - lane = case((NewsItem.published_at >= realtime_from, 'realtime'), else_='backfill') if realtime_from else 'backfill' - if realtime_from: - groups = (await session.execute(select(lane, Package.state, func.count()).join(NewsItem) - .where(eligible(settings)).group_by(lane, Package.state))).all() - else: - groups = [('backfill', state, count) for state, count in (await session.execute( - select(Package.state, func.count()).join(NewsItem).where(eligible(settings)).group_by(Package.state))).all()] - queues = {name: {state: count for selected, state, count in groups if selected == name} + stages = dict((await session.execute(select(Attempt.stage, func.count()).where( + Attempt.state == 'processing').group_by(Attempt.stage))).all()) + stages['research'] = await session.scalar(select(func.count()).select_from(Version).where(Version.state == 'processing')) + stages = {stage: stages.get(stage, 0) for stage in ('index', 'assign', 'research')} + running = sum(stages.values()) + capacity = settings.config.get('concurrency', 3) + lane = case((priority_package(settings), 'realtime'), else_='backfill') + state = case((Package.id.is_(None), 'missing'), + (Package.content_hash != NewsItem.content_hash, 'stale'), else_=Package.state) + groups = (await session.execute(select(lane, state, func.count()).select_from(NewsItem) + .outerjoin(Package, Package.news_id == NewsItem.id).where(eligible(settings)).group_by(lane, state))).all() + queues = {name: {status: count for selected, status, count in groups if selected == name} for name in ('realtime', 'backfill')} - return {'running': running, 'draining': not settings.enabled and running > 0, **queues} + event_state = case((Event.state == 'processing', 'processing'), + (active_research(), 'waiting_previous'), (Event.state == 'failed', 'failed'), + (manual_review(), 'manual_review'), (~ready_event(), 'blocked'), + (Event.due_at > datetime.now(UTC), 'aggregating'), else_='ready') + event_groups = (await session.execute(select(Event.realtime_pending, event_state, func.count()).where( + Event.state.in_(['pending', 'processing', 'failed']), scoped_event(settings)) + .group_by(Event.realtime_pending, event_state))).all() + events = {name: {status: count for realtime, status, count in event_groups if realtime == (name == 'realtime')} + for name in ('realtime', 'backfill')} + reasons = [] + if running < capacity: + if not settings.enabled: + reasons.append('draining' if running else 'paused') + else: + if any(q.get('assign_pending', 0) for q in queues.values()): + reasons.append('assignment_serial' if stages['assign'] else 'preceding_index') + for status, reason in [('waiting_previous', 'waiting_previous'), ('aggregating', 'aggregation'), ('blocked', 'dependencies'), + ('manual_review', 'manual_retry'), ('failed', 'manual_retry')]: + if any(q.get(status, 0) for q in events.values()) and reason not in reasons: + reasons.append(reason) + if any(count for q in queues.values() for state, count in q.items() if state.endswith('_failed')) and 'manual_retry' not in reasons: + reasons.append('manual_retry') + if not reasons: + reasons.append('no_work' if not any( + q.get(s, 0) for q in queues.values() for s in ('missing', 'stale', 'index_pending', 'assign_pending') + ) and not any(q.get('ready', 0) for q in events.values()) else 'dispatching') + return {'running': running, 'capacity': capacity, 'running_by_stage': stages, + 'draining': not settings.enabled and running > 0, 'events': events, 'idle_reasons': reasons, **queues} diff --git a/backend/src/alphapulse/news_research/service.py b/backend/src/alphapulse/news_research/service.py index 8409f90..65abbb8 100644 --- a/backend/src/alphapulse/news_research/service.py +++ b/backend/src/alphapulse/news_research/service.py @@ -38,6 +38,7 @@ from alphapulse.news_research.contracts import ( ) from alphapulse.news_research import vocabulary as vocab +from alphapulse.news_research.scheduling import manual_review, active_research publication_lock = asyncio.Lock() assignment_lock = asyncio.Lock() @@ -826,7 +827,9 @@ async def process_event(identifier, config, force=False): or (not force and event.due_at > datetime.now(UTC)) ): return - if await session.scalar(select(Version.id).where(Version.event_id == identifier, Version.state == 'processing').limit(1)): + if await session.scalar(select(Event.id).where(Event.id == identifier, active_research())): + return + if await session.scalar(select(Event.id).where(Event.id == identifier, manual_review())): return members = list( await session.execute( diff --git a/backend/src/alphapulse/news_research/worker.py b/backend/src/alphapulse/news_research/worker.py index 46d3ec2..b43db36 100644 --- a/backend/src/alphapulse/news_research/worker.py +++ b/backend/src/alphapulse/news_research/worker.py @@ -2,7 +2,7 @@ import asyncio import logging -from sqlalchemy import select, text, update +from sqlalchemy import text, update from alphapulse.core.db import SessionLocal, engine from alphapulse.news_research.models import ( ResearchSettings, @@ -10,10 +10,9 @@ from alphapulse.news_research.models import ( Event, Attempt, Version, - Batch, BatchEvent, ) from alphapulse.news_research.service import process_package, process_event -from alphapulse.news_research import batches, scheduling +from alphapulse.news_research import scheduling logger = logging.getLogger(__name__) @@ -39,7 +38,6 @@ async def recover_interrupted(): .where(Version.state == "processing") .values(state="failed", error="进程中断，调用结果不明，请手动重试") ) - await session.execute(update(BatchEvent).where(BatchEvent.state == 'processing').values(state='pending')) await session.commit() @@ -64,20 +62,17 @@ async def run(): if settings and settings.enabled: active = set(tasks.values()) slots = max(0, settings.config.get('concurrency', 3) - len(tasks)) - actions = await scheduling.realtime_actions(session, settings, active, slots) - batch = await session.scalar(select(Batch).where( - Batch.phase.in_(['assigning', 'researching'])).order_by(Batch.id).limit(1)) - if batch is None: - batch = await batches.freeze(session, settings) - if batch is not None: - actions += await batches.actions(session, batch, active | set(actions), slots - len(actions), settings) + actions = await scheduling.actions(session, settings, active, slots) await session.commit() config = {**settings.config, '_control_revision': settings.control_revision} for stage, identifier in actions: call = (process_event(identifier, config) if stage == 'event' else process_package(identifier, config)) tasks[asyncio.create_task(call)] = (stage, identifier) - await asyncio.sleep(1) + if tasks: + await asyncio.wait(tasks, timeout=1, return_when=asyncio.FIRST_COMPLETED) + else: + await asyncio.sleep(1) finally: await asyncio.gather(*tasks, return_exceptions=True) diff --git a/backend/tests/test_news_research_workspace.py b/backend/tests/test_news_research_workspace.py index a427f8a..6fcf1e9 100644 --- a/backend/tests/test_news_research_workspace.py +++ b/backend/tests/test_news_research_workspace.py @@ -948,3 +948,78 @@ def test_enable_cannot_apply_an_unsaved_backfill_time(trading): assert error.value.status_code == 422 session.scalar.assert_not_awaited() session.commit.assert_not_awaited() + + +def test_worker_refills_on_completion_without_waiting_for_other_tasks(trading, monkeypatch): + """A slow peer must not turn the shared pool back into a batch barrier.""" + import asyncio + from unittest.mock import AsyncMock + from alphapulse.news_research import worker + + async def scenario(): + settings = N(enabled=True, config={'concurrency': 8}, control_revision=1) + + class Session: + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + pass + + async def get(self, *args): + return settings + + async def commit(self): + pass + + class Lock(Session): + async def scalar(self, *args): + return True + + release = asyncio.Event() + first_done = asyncio.Event() + initial_full = asyncio.Event() + refilled = asyncio.Event() + entered = [] + live = 0 + peak = 0 + next_id = 0 + + async def actions(session, settings, active, slots): + nonlocal next_id + result = [('index', i) for i in range(next_id, next_id + slots)] + next_id += slots + return result + + async def process(identifier, config): + nonlocal live, peak + live += 1 + peak = max(peak, live) + entered.append(identifier) + if len(entered) == 8: + initial_full.set() + if len(entered) > 8: + refilled.set() + try: + await (first_done.wait() if identifier == 0 else release.wait()) + finally: + live -= 1 + + monkeypatch.setattr(worker, 'engine', N(connect=Lock)) + monkeypatch.setattr(worker, 'SessionLocal', Session) + monkeypatch.setattr(worker, 'recover_interrupted', AsyncMock()) + monkeypatch.setattr(worker.scheduling, 'actions', actions) + monkeypatch.setattr(worker, 'process_package', process) + task = asyncio.create_task(worker.run()) + try: + await asyncio.wait_for(initial_full.wait(), 2) + first_done.set() + await asyncio.wait_for(refilled.wait(), 2) + assert peak == 8 and live == 8 and not release.is_set() + assert len(entered) == 9 + finally: + task.cancel() + release.set() + await asyncio.gather(task, return_exceptions=True) + + asyncio.run(scenario()) diff --git a/deploy/README.md b/deploy/README.md index c4f64fd..76c5ac3 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -100,9 +100,9 @@ New event timeline day/month values are UTC-midnight calendar markers (first day Long-input compaction stops when the final research Skill, payload, tools and result schema fit the configured context budget. Extraction and consolidation keep their event-boundary instructions and check their own call budgets; the final fit check does not count member metadata a second time in those instructions. -Ordinary facts, independent evidence, corrections and urgent developments wait for their frozen batch to finish assignment; urgent events sort first within research. Pure reposts retain their own packages and citations without requesting a new successful event review. Indexing, assignment and research share the configured concurrency pool (1–16, default 3). Pausing stops new admissions; in-flight work drains. Interrupted tasks recover from PostgreSQL. Third-stage reading/trends, the legacy queue/backfill jobs, old three-model configuration, API, ORM models and unused frontend components have been removed. Reused model submission, evidence, time and retrieval helpers now belong to `news_research`; the public HTTP reader belongs to `core.public_http`. Historical Alembic migrations and inactive legacy tables are retained; autogeneration excludes those archived tables. The three retired news Skills and their old bindings are removed. There is no legacy reset command. +Facts, independent evidence and corrections aggregate per event; urgent developments become ready immediately. Completed assignments do not wait for an unrelated batch before research. Pure reposts retain their own packages and citations without requesting a new successful event review. Indexing, assignment and research share the configured concurrency pool (1–16, default 3). Pausing stops new admissions; in-flight work drains. Interrupted tasks recover from PostgreSQL. Third-stage reading/trends, the legacy queue/backfill jobs, old three-model configuration, API, ORM models and unused frontend components have been removed. Reused model submission, evidence, time and retrieval helpers now belong to `news_research`; the public HTTP reader belongs to `core.public_http`. Historical Alembic migrations and inactive legacy tables are retained; autogeneration excludes those archived tables. The three retired news Skills and their old bindings are removed. There is no legacy reset command. -The homepage provides event and package searches, pagination, membership details, timelines, evidence and version history. `GET /api/risk/settings` and `PUT /api/risk/settings` manage independent `index` and `research` model/effort/Skill bindings and concurrency; saving pauses admission. Clearing effort inherits provider settings. `POST /api/risk/enable` validates the two models with synthetic material, including separate index/assignment contracts. `POST /api/risk/packages` enrolls up to 500 explicit news IDs; while enabled the worker additionally admits news published since November 1, 2026, 00:00 Asia/Shanghai. Failed assignment retries reuse indexing; event retries never repeat indexing. Earlier source revisions are not shown with current revision annotations. +The homepage provides event and package searches, pagination, membership details, timelines, evidence and version history. `GET /api/risk/settings` and `PUT /api/risk/settings` manage independent `index` and `research` model/effort/Skill bindings and concurrency; saving pauses admission. Clearing effort inherits provider settings. `POST /api/risk/enable` validates the two models with synthetic material, including separate index/assignment contracts. `POST /api/risk/packages` enrolls up to 500 explicit news IDs; while enabled the worker admits eligible news using the saved historical start and persistent realtime boundary. Failed assignment retries reuse indexing; event retries never repeat indexing. Earlier source revisions are not shown with current revision annotations. Create the two default Skill drafts in homepage settings, review and publish them in Skills, then bind them before enabling. Template generation never publishes or changes existing Chat/trading Skills. The additive `a71234567890` migration follows `f60123456789` and creates seven `news_research_*` tables; `b81234567890` adds the eighth table for immutable vocabulary snapshots. There is no conversion, truncation or deletion of original/legacy tables. Production schema execution, code deployment, commits and pushes require the corresponding task authorization; preparing a migration alone does not grant it. @@ -164,7 +164,12 @@ Resume requires the disposable database as well as report files. If that databas News research uses only stored news, packages, event versions and economic releases. External search and webpage-reading tools, their settings and the SearXNG service are removed. Collector HTTP access is unchanged. Historical external evidence remains readable but cannot be restored as evidence read by a new research attempt. -The worker freezes a durable batch of eligible originals and queued work. It completes indexing and serial assignment before researching any batch event; newly arriving work waits for the next batch. Failed items are isolated, source revisions are deferred, and successful stages are reused across pauses and restarts. Each event is scheduled once per batch. `/api/risk/settings` includes the latest `batch` progress. Batch tables are additive and retain original news and research history. Deployment keeps research paused; enabling remains an explicit action. +The worker continuously fills the shared pool from durable news, package and event +states. Indexing and ready research run alongside serial assignment; no batch +barrier delays new arrivals. Successful stages survive pauses and restarts, and +failed or uncertain calls require explicit retry. Settings return queue counts +and a null `batch` field for compatibility. Historical batch records are retained. +Deployment keeps research paused; enabling remains an explicit action. The crawler console accepts administrator-managed public RSS/Atom subscriptions, names, intervals and enable/disable controls. New subscriptions are paused, no default feeds are added, and the manager discovers new subscriptions without a restart. Only feed-provided content is stored; a summary is labeled as a summary. Arbitrary private-network feed URLs are rejected. @@ -475,13 +480,26 @@ a differing legacy enable body returns 422. The first successful enable fixes Publication time, not ingestion time, determines realtime membership. Realtime work takes the next available slot without cancelling accepted calls. Historical -work uses batches of at most 30 originals, ordered chronologically. Saving a new -start pauses dispatch; enabling after drain retires unfinished batch schedules -and rebuilds from durable stage states without repeating successes. Shrinking -history does not exclude realtime backlog. Known in-range dependencies inherit -a waiting realtime event's priority. Out-of-range or failed members block event -research without silently dropping evidence or retrying paid failures. +work continuously fills free slots from durable news/package/event states, with +no thirty-news admission barrier or batch phase. Assignment remains globally +serial and chronological within a priority lane; a preceding indexing task must +finish before later news is assigned. Failed revisions do not block the lane. +Within each lane, assignment is selected first, then ready event research, then +indexing; realtime work and its in-range known dependencies precede historical +work. Completion wakes dispatch immediately; new arrivals and timers are polled +at most once per second. Existing calls finish normally when concurrency shrinks. +Saving a new start pauses dispatch; the next enable uses the saved scope without +repeating successful stages. Shrinking history does not exclude realtime backlog. +Out-of-range or failed members block event research without dropping evidence. +Legacy batch tables remain historical records, with no new batches created. +Current-revision deferred records and superseded research require explicit retry; +a new event revision can proceed. Settings retain a null `batch` compatibility +field, while `queue` exposes capacity, running_by_stage, per-lane event states and +idle_reasons. A pending event with an active older research version is counted +as waiting_previous, with an explicit wait reason, rather than as ready work. +Missing and stale originals are included in queue counts. Coverage +remains a read-only report. Ordinary event research uses the existing five-minute quiet window with a 15-minute maximum; urgent events become ready immediately. A running research call is not duplicated when newer members arrive. New changes start another @@ -490,7 +508,7 @@ After a worker interruption, started calls become failed with an unknown-result message and need manual retry; unclaimed work resumes normally. Strict realtime priority can starve historical work under sustained overload. -CI additionally runs `scripts/smoke_priority_queue.py` in its disposable internal +CI runs `scripts/smoke_dynamic_queue.py` and `scripts/smoke_priority_queue.py` in its disposable internal network with synthetic credentials and no model calls. It covers preemption, late historical ingestion, stable realtime eligibility, range changes, dependencies, aggregation, control revisions and interrupted-call recovery. diff --git a/scripts/ci.py b/scripts/ci.py index df395a8..53ec989 100644 --- a/scripts/ci.py +++ b/scripts/ci.py @@ -181,7 +181,7 @@ def smoke(build, label): wait_command('docker', 'exec', names['api'], 'python', '-c', 'import urllib.request; urllib.request.urlopen("http://localhost:8000/api/health", timeout=3)') print(run('docker', 'exec', '-i', names['api'], 'python', '-', - input=(ROOT / 'scripts/smoke_news_batches.py').read_text()), flush=True) + input=(ROOT / 'scripts/smoke_dynamic_queue.py').read_text()), flush=True) print(run('docker', 'exec', '-i', names['api'], 'python', '-', input=(ROOT / 'scripts/smoke_priority_queue.py').read_text()), flush=True) print(run('docker', 'exec', '-i', names['api'], 'python', '-', diff --git a/scripts/smoke_dynamic_queue.py b/scripts/smoke_dynamic_queue.py new file mode 100644 index 0000000..b9ca387 --- /dev/null +++ b/scripts/smoke_dynamic_queue.py @@ -0,0 +1,289 @@ +"""Validate continuous scheduling with synthetic PostgreSQL data and no model calls.""" +import asyncio +import os +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace +from urllib.parse import urlparse +from unittest.mock import patch +from sqlalchemy import select, text, func, delete +from alphapulse.core.db import SessionLocal, engine +from alphapulse.models import NewsItem +from alphapulse.news_research.models import Package, Event, Membership, Attempt, Version, Batch, BatchEvent, BatchPackage +from alphapulse.news_research import scheduling, service, coverage as reports +from alphapulse.news_research.api import retry_event + + +async def main(): + url = urlparse(os.environ['DATABASE_URL']) + assert (url.username, url.password, url.path) == ('ci', 'ci-only', '/ci') + now = datetime.now(UTC) + config = SimpleNamespace(start_at=now-timedelta(days=5), realtime_from=now-timedelta(hours=1), + config={'concurrency': 8}, enabled=True) + async with SessionLocal() as s: + async def clear(): + await s.execute(text('TRUNCATE news_research_batches, news_research_packages, news_research_events, news_items CASCADE')) + await s.commit() + s.expunge_all() + + def original(i, live=False): + return NewsItem(source_code='wscn_news', external_id='dynamic-'+str(i), title='source', summary='', + content_text='source', content_hash='a'*64, url='https://example.invalid/'+str(i), + published_at=now-timedelta(minutes=30) if live else now-timedelta(days=3)+timedelta(seconds=i)) + + # More than thirty originals: refill while seven older calls remain active. + s.add_all([original(i) for i in range(80)]) + await s.commit() + actions = await scheduling.actions(s, config, set(), 8) + assert len(actions) == 8 and all(stage == 'index' for stage, _ in actions) + first = await s.get(Package, actions[0][1]); first.state='assign_pending' + await s.commit() + active=set(actions[1:]) + selected=await scheduling.actions(s, config, active, 1) + assert selected == [('assign', first.id)] + first.state='done'; await s.commit() + # Each completion admits another original, independent of the slow seven. + for _ in range(35): + selected=await scheduling.actions(s, config, active, 1) + assert len(selected)==1 and selected[0][0]=='index' + p=await s.get(Package, selected[0][1]); p.state='done' + await s.commit() + assert await s.scalar(select(func.count()).select_from(Package)) > 30 + assert await s.scalar(select(func.count()).select_from(Batch)) == 0 + status=await scheduling.queue_status(s, config) + assert status['backfill']['missing'] > 0 and status['capacity']==8 + report=await reports.coverage(s, config.start_at) + assert report['total']==80 and report['unprocessed']==80 + await clear() + + # One serial assignment plus seven ready studies, not an entire batch barrier. + news=[original(i) for i in range(60)] + s.add_all(news); await s.flush() + packages=[Package(news_id=n.id,content_hash=n.content_hash,state='done',keywords=[]) for n in news] + s.add_all(packages); await s.flush() + events=[Event(title='event-'+str(i),tags=[],revision=1,state='pending',due_at=now-timedelta(minutes=1)) for i in range(54)] + s.add_all(events); await s.flush() + for e,p in zip(events,packages,strict=False): + s.add(Membership(event_id=e.id,package_id=p.id,evidence='source',reason='source',change_kind='new_fact')) + packages[54].state='assign_pending' + packages[55].state='assign_pending' + await s.commit() + selected=await scheduling.actions(s, config, set(), 8) + assert selected[0]==('assign',packages[54].id) + assert len(selected)==8 and sum(stage=='event' for stage,_ in selected)==7 + # Another study fills a freed slot even while that assignment remains active. + active=set(selected[:-1]) + refill=await scheduling.actions(s, config, active, 1) + assert len(refill)==1 and refill[0][0]=='event' and refill[0] not in active + # Realtime joins the next free slot; zero capacity never enrolls work. + live=original(100,True); s.add(live); await s.commit() + assert await scheduling.actions(s, config, active, 0)==[] + assert await s.scalar(select(Package.id).where(Package.news_id==live.id)) is None + refill=await scheduling.actions(s, config, active, 1) + live_package=await s.scalar(select(Package).where(Package.news_id==live.id)) + assert refill==[('index',live_package.id)] + # Stale source enrollment happens only after its existing call is released. + live_package.state='index_processing' + s.add(Attempt(package_id=live_package.id,stage='index',state='processing',content_hash='a'*64,input={},rules={})) + live.content_hash='b'*64; await s.commit() + assert await scheduling.realtime_actions(s,config,set(),8)==[] + await clear() + + # Failed/deferred heads must be filtered before LIMIT, including realtime. + n=original(1); s.add(n); await s.flush() + p=Package(news_id=n.id,content_hash=n.content_hash,state='done',keywords=[]); s.add(p); await s.flush() + old=Batch(phase='partial',cutoff_at=now); s.add(old); await s.flush() + events=[Event(title='manual-'+str(i),tags=[],revision=7,state='pending',due_at=now-timedelta(minutes=1)) for i in range(45)] + s.add_all(events); await s.flush() + for i,e in enumerate(events): + s.add(Membership(event_id=e.id,package_id=p.id,evidence='source',reason='source',change_kind='new_fact')) + if i<35: s.add(BatchEvent(batch_id=old.id,event_id=e.id,revision=7,state='deferred')) + s.add(Version(event_id=events[35].id,revision=7,state='superseded',input={},rules={})) + events[36].due_at=now+timedelta(minutes=5) + events[37].state='failed' + await s.commit() + selected=await scheduling.actions(s,config,set(),8) + assert {i for stage,i in selected}=={e.id for e in events[38:]} + status=await scheduling.queue_status(s,config) + assert status['events']['backfill']=={'manual_review':36,'aggregating':1,'failed':1,'ready':7} + assert 'aggregation' in status['idle_reasons'] and 'manual_retry' in status['idle_reasons'] + # Final claim independently enforces manual retry, not only selection. + with patch.object(service, 'model_config', side_effect=AssertionError('Must not call model')): + await service.process_event(events[35].id,{}) + await retry_event(events[35].id,s) + assert events[35].revision==8 + assert ('event',events[35].id) in await scheduling.actions(s,config,set(),8) + # Moving to a new revision naturally releases legacy deferred records. + events[0].revision=8; events[0].realtime_pending=True + await s.commit() + assert (await scheduling.actions(s,config,set(),1))==[('event',events[0].id)] + await clear() + + # Failed earlier news does not block; a slow earlier index does block later assignment. + news=[original(i) for i in range(4)]; s.add_all(news); await s.flush() + ps=[Package(news_id=n.id,content_hash=n.content_hash,state=state,keywords=[]) for n,state in zip(news, + ['index_failed','index_processing','assign_pending','index_pending'],strict=True)] + s.add_all(ps); await s.commit() + selected=await scheduling.actions(s,config,{('index',ps[1].id)},7) + assert selected==[('index',ps[3].id)] + ps[1].state='assign_pending'; await s.commit() + selected=await scheduling.actions(s,config,set(),8) + assert selected[0]==('assign',ps[1].id) and sum(stage=='assign' for stage,_ in selected)==1 + config.start_at=now-timedelta(days=1) + assert await scheduling.actions(s,config,set(),8)==[] + config.start_at=now-timedelta(days=5) + assert await scheduling.actions(s,config,set(),8) + await clear() + # A pending revision cannot run while the preceding revision is still paid/in flight. + async with SessionLocal() as s: + n=original(1,True); s.add(n); await s.flush() + p=Package(news_id=n.id,content_hash=n.content_hash,state='done',keywords=[]) + e=Event(title='changing event',tags=[],revision=2,state='pending',realtime_pending=True, + due_at=now-timedelta(minutes=1)) + s.add_all([p,e]); await s.flush() + s.add(Membership(event_id=e.id,package_id=p.id,evidence='source',reason='source',change_kind='new_fact')) + version=Version(event_id=e.id,revision=1,state='processing',input={},rules={}) + s.add(version); await s.commit() + status=await scheduling.queue_status(s,config) + assert status['running']==1 and status['events']['realtime']=={'waiting_previous':1} + assert status['idle_reasons']==['waiting_previous'] + assert await scheduling.actions(s,config,set(),7)==[] + from alphapulse.news_research.api import event as event_detail + detail=await event_detail(e.id,s) + assert detail.data['wait_reason']=='等待上一轮研究完成后处理事件新变化' + # Releasing the old call automatically makes the new revision ready again. + version.state='superseded'; await s.commit() + status=await scheduling.queue_status(s,config) + assert status['running']==0 and status['events']['realtime']=={'ready':1} + assert await scheduling.actions(s,config,set(),8)==[('event',e.id)] + # Apply the same guard to historical events, independent of the priority lane. + e.realtime_pending=False; version.state='processing'; await s.commit() + status=await scheduling.queue_status(s,config) + assert status['events']['backfill']=={'waiting_previous':1} + assert await scheduling.actions(s,config,set(),8)==[] + await s.execute(text('TRUNCATE news_research_packages, news_research_events, news_items CASCADE')) + await s.commit() + # Large history scans remain bounded by free capacity, without huge IN lists. + async with SessionLocal() as s: + await s.execute(text("""INSERT INTO news_items + (source_code,external_id,url,title,summary,content_html,content_text,tags,is_paywalled, + published_at,content_hash,revision_no,extra) + SELECT 'wscn_news','dynamic-stress-'||i,'https://example.invalid/stress','source','','','source', + '[]'::jsonb,false,:published,:hash,1,'{}'::jsonb FROM generate_series(1,33000) i"""), + {'published':now-timedelta(days=2),'hash':'a'*64}) + await s.commit() + for capacity in [1,3,8,16]: + selected=await scheduling.actions(s,config,set(),capacity) + assert len(selected)==capacity and all(stage=='index' for stage,_ in selected) + assert await s.scalar(select(func.count()).select_from(Package)) <= 16 + await s.execute(text('TRUNCATE news_research_packages, news_items CASCADE')) + await s.commit() + await backfill_regressions(now) + await carryover_regressions(now) + await engine.dispose() + print('Dynamic queue passed: continuous refill beyond 30 originals, serial assignment with 54 ready studies, realtime priority, stale active calls, chronology, coverage and manual retry barriers.') + + +async def backfill_regressions(now): + start = now - timedelta(days=2) + async with SessionLocal() as s: + def original(key, published_at): + return NewsItem(source_code='wscn_news', external_id='coverage-'+key, + title=key, summary='', content_text=key, content_hash='a'*64, + url='https://example.invalid/'+key, published_at=published_at) + keys = ['boundary', 'done', 'research-pending', 'research-failed', + 'index-failed', 'processing', 'assign-pending', 'stale'] + news = {key: original(key, start if key == 'boundary' else now) for key in keys} + older = original('older', start-timedelta(microseconds=1)) + future = original('future', now+timedelta(days=1)) + s.add_all([*news.values(), older, future]) + await s.flush() + states = {'index-failed': 'index_failed', 'processing': 'index_processing', + 'assign-pending': 'assign_pending'} + packages = {key: Package(news_id=row.id, content_hash='b'*64 if key == 'stale' else row.content_hash, + state=states.get(key, 'done'), keywords=[]) + for key, row in news.items() if key != 'boundary'} + events = {state: Event(title='coverage-'+state, state=state, tags=[], revision=1, + result={} if state == 'done' else None, due_at=now) + for state in ['done', 'pending', 'failed']} + s.add_all([*packages.values(), *events.values()]) + await s.flush() + for key, state in [('done', 'done'), ('research-pending', 'pending'), + ('research-failed', 'failed'), ('research-failed', 'done'), ('stale', 'done')]: + s.add(Membership(package_id=packages[key].id, event_id=events[state].id, + evidence=key, reason='coverage fixture', change_kind='new_fact')) + await s.commit() + before = await s.scalar(select(func.count()).select_from(Package)) + report = await reports.coverage(s, start, now) + assert {k: report[k] for k in ['total', 'completed', 'missing', 'stale', 'pending', 'processing', 'failed', 'unprocessed']} == { + 'total': 8, 'completed': 1, 'missing': 1, 'stale': 1, 'pending': 2, + 'processing': 1, 'failed': 2, 'unprocessed': 7}, report + assert await s.scalar(select(func.count()).select_from(Package)) == before + # Late historical arrivals remain discoverable without a fixed cutoff. + late = original('late-arrival', start+timedelta(hours=1)) + s.add(late) + packages['processing'].state = 'index_pending' + await s.commit() + report = await reports.coverage(s, start, now) + assert report['total'] == 9 and report['missing'] == 2 + config = SimpleNamespace(start_at=start, realtime_from=None) + selected = await scheduling.actions(s, config, set(), 16) + admitted = set(await s.scalars(select(Package.news_id).where(Package.id.in_([i for stage,i in selected if stage=='index'])))) + assert {news['boundary'].id, late.id} <= admitted + assert not {news['done'].id, older.id, future.id} & admitted + assert packages['stale'].state == 'index_pending' + assert packages['stale'].content_hash == news['stale'].content_hash + config.start_at=older.published_at + selected=await scheduling.actions(s, config, set(), 16) + admitted=set(await s.scalars(select(Package.news_id).where(Package.id.in_([i for stage,i in selected if stage=='index'])))) + assert older.id in admitted and future.id not in admitted + assert all(row.content_hash == 'a'*64 for row in [*news.values(), older, future, late]) + for model in (BatchEvent, BatchPackage, Batch, Membership, Attempt, Package, Event, NewsItem): + await s.execute(delete(model)) + await s.commit() + print('Backfill coverage passed: inclusive boundary, pending research, distinct news counts, late arrivals, source revisions and expanded history without model calls.') + +async def carryover_regressions(now): + start = now - timedelta(days=2) + async with SessionLocal() as s: + keys = ['inside', 'pending', 'processing', 'failed', 'stale', 'research', 'done', 'unselected', 'future'] + news = {key: NewsItem(source_code='wscn_news', external_id='carryover-'+key, + title=key, summary='', content_text=key, content_hash='a'*64, + url='https://example.invalid/'+key, + published_at=now if key == 'inside' else + now+timedelta(days=1) if key == 'future' else start-timedelta(days=1)) + for key in keys} + s.add_all(news.values()) + await s.flush() + states = {'pending': 'index_pending', 'processing': 'assign_processing', + 'failed': 'index_failed', 'future': 'assign_pending'} + packages = {key: Package(news_id=row.id, content_hash='b'*64 if key == 'stale' else row.content_hash, + state=states.get(key, 'done'), keywords=[]) + for key, row in news.items() if key != 'unselected'} + done = Event(title='carryover-done', tags=[], state='done', result={}, revision=1, due_at=now) + pending = Event(title='carryover-pending', tags=[], state='pending', revision=1, due_at=now) + batch = Batch(phase='assigning', cutoff_at=now) + s.add_all([*packages.values(), done, pending, batch]) + await s.flush() + for key, event in [('inside', done), ('done', done), ('research', done), ('research', pending)]: + s.add(Membership(package_id=packages[key].id, event_id=event.id, + evidence=key, reason='carryover fixture', change_kind='new_fact')) + for key in ['pending', 'processing', 'stale']: + s.add(BatchPackage(batch_id=batch.id, package_id=packages[key].id, + content_hash=packages[key].content_hash, state='pending')) + s.add(BatchEvent(batch_id=batch.id, event_id=pending.id, revision=1, state='pending')) + await s.commit() + wide = await reports.coverage(s, start-timedelta(days=2), now) + assert wide['outside']['unprocessed'] == 1 # Explicitly enrolled future original. + narrow = await reports.coverage(s, start, now) + assert narrow['total'] == 1 and narrow['completed'] == 1 and narrow['unprocessed'] == 0 + assert narrow['outside'] == {'unprocessed': 6, 'pending': 3, 'processing': 1, 'failed': 1, 'stale': 1}, narrow + assert await s.scalar(select(func.count()).select_from(BatchPackage)) == 3 + assert await s.scalar(select(func.count()).select_from(BatchEvent)) == 1 + assert await s.scalar(select(func.count()).select_from(Package)) == 8 + assert batch.phase == 'assigning' and packages['pending'].state == 'index_pending' + for model in (BatchEvent, BatchPackage, Batch, Membership, Attempt, Package, Event, NewsItem): + await s.execute(delete(model)) + await s.commit() + print('Carryover coverage passed: narrowing preserves and reports outside queued news, active work, changed originals, pending research and explicit future enrollment without counting completed or unselected originals.') + +asyncio.run(main()) diff --git a/scripts/smoke_news_batches.py b/scripts/smoke_news_batches.py deleted file mode 100644 index 9b78970..0000000 --- a/scripts/smoke_news_batches.py +++ /dev/null @@ -1,358 +0,0 @@ -"""Exercise durable batch barriers in the disposable CI database without model calls.""" -import asyncio -import os -from datetime import UTC, datetime, timedelta -from types import SimpleNamespace -from urllib.parse import urlparse -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, service -from alphapulse.news_research.api import retry_event - - -async def main(): - url = urlparse(os.environ['DATABASE_URL']) - assert (url.username, url.password, url.path) == ('ci', 'ci-only', '/ci') - now = datetime.now(UTC) - settings = SimpleNamespace(start_at=now - timedelta(days=1)) - async with SessionLocal() as s: - assert await s.scalar(select(Batch.id).limit(1)) is None - def news(key): - return NewsItem(source_code='wscn_news', external_id='batch-'+key, - title=key, summary='', content_text=key, content_hash='a'*64, - url='https://example.invalid/'+key, published_at=now) - originals = [news(str(i)) for i in range(3)] - s.add_all(originals) - await s.commit() - batch = await batches.freeze(s, settings) - identifier = batch.id - assert batch.phase == 'assigning' - packages = list(await s.scalars(select(Package).order_by(Package.id))) - assert len(packages) == 3 - later = news('later') - s.add(later) - await s.commit() - actions = await batches.actions(s, batch, set(), 8) - assert len(actions) == 3 and all(stage == 'index' for stage, _ in actions) - for p in packages: - p.state = 'assign_pending' - await s.commit() - actions = await batches.actions(s, batch, set(), 8) - assert len(actions) == 1 and actions[0][0] == 'assign' - assert not await batches.actions(s, batch, {actions[0]}, 8) - event = Event(title='One shared event', tags=[], revision=3, state='pending', due_at=now) - s.add(event) - await s.flush() - for p in packages[:2]: - p.state = 'done' - s.add(Membership(event_id=event.id, package_id=p.id, evidence='original', reason='same event', change_kind='new_fact')) - s.add(Attempt(package_id=packages[0].id, content_hash=packages[0].content_hash, stage='assign', state='done', input={}, rules={'membership_targets': {str(event.id): {'urgent': True}}})) - packages[2].state = 'index_failed' - await s.commit() - assert not await batches.actions(s, batch, {('assign', packages[0].id)}, 8) - assert batch.phase == 'assigning' - assert not await batches.actions(s, batch, set(), 8) - await s.commit() - assert batch.phase == 'researching' - assert (await s.get(BatchEvent, (batch.id, event.id))).state == 'urgent_pending' - # Reopen the session to prove the admission barrier is durable. - async with SessionLocal() as s: - batch = await s.get(Batch, identifier) - actions = await batches.actions(s, batch, set(), 8) - assert actions == [('event', event.id)] - await s.commit() - assert not await batches.actions(s, batch, set(actions), 8) - e = await s.get(Event, event.id) - e.state = 'done' - await s.commit() - assert not await batches.actions(s, batch, set(), 8) - await s.commit() - assert batch.phase == 'partial' - progress = await batches.progress(s) - assert progress['packages_failed'] == 1 and progress['events_done'] == 1 - next_batch = await batches.freeze(s, settings) - items = list(await s.scalars(select(BatchPackage).where(BatchPackage.batch_id == next_batch.id))) - assert len(items) == 1 - p = await s.get(Package, items[0].package_id) - assert p.news_id == later.id - original = await s.get(NewsItem, later.id) - original.content_hash = 'b'*64 - await s.commit() - assert not await batches.actions(s, next_batch, set(), 8) - await s.commit() - assert items[0].state == 'deferred' - # Leave no fixtures for other isolated acceptance scripts. - for model in (BatchEvent, BatchPackage, Batch, Membership, Attempt, Package, Event, NewsItem): - await s.execute(delete(model)) - await s.commit() - await backfill_regressions(now) - await carryover_regressions(now) - await regressions(now, settings) - await queue_review_regressions(now, settings) - await engine.dispose() - print('Frozen batches passed: assignment barrier, single research, failures, new arrivals, restart and source revisions.') - - -async def backfill_regressions(now): - start = now - timedelta(days=2) - async with SessionLocal() as s: - def original(key, published_at): - return NewsItem(source_code='wscn_news', external_id='coverage-'+key, - title=key, summary='', content_text=key, content_hash='a'*64, - url='https://example.invalid/'+key, published_at=published_at) - keys = ['boundary', 'done', 'research-pending', 'research-failed', - 'index-failed', 'processing', 'assign-pending', 'stale'] - news = {key: original(key, start if key == 'boundary' else now) for key in keys} - older = original('older', start-timedelta(microseconds=1)) - future = original('future', now+timedelta(days=1)) - s.add_all([*news.values(), older, future]) - await s.flush() - states = {'index-failed': 'index_failed', 'processing': 'index_processing', - 'assign-pending': 'assign_pending'} - packages = {key: Package(news_id=row.id, content_hash='b'*64 if key == 'stale' else row.content_hash, - state=states.get(key, 'done'), keywords=[]) - for key, row in news.items() if key != 'boundary'} - events = {state: Event(title='coverage-'+state, state=state, tags=[], revision=1, - result={} if state == 'done' else None, due_at=now) - for state in ['done', 'pending', 'failed']} - s.add_all([*packages.values(), *events.values()]) - await s.flush() - for key, state in [('done', 'done'), ('research-pending', 'pending'), - ('research-failed', 'failed'), ('research-failed', 'done'), ('stale', 'done')]: - s.add(Membership(package_id=packages[key].id, event_id=events[state].id, - evidence=key, reason='coverage fixture', change_kind='new_fact')) - await s.commit() - before = await s.scalar(select(func.count()).select_from(Package)) - report = await batches.coverage(s, start, now) - assert {k: report[k] for k in ['total', 'completed', 'missing', 'stale', 'pending', 'processing', 'failed', 'unprocessed']} == { - 'total': 8, 'completed': 1, 'missing': 1, 'stale': 1, 'pending': 2, - 'processing': 1, 'failed': 2, 'unprocessed': 7}, report - assert await s.scalar(select(func.count()).select_from(Package)) == before - # Historical news arriving after a scan must still enter the next frozen batch. - late = original('late-arrival', start+timedelta(hours=1)) - s.add(late) - packages['processing'].state = 'index_pending' - await s.commit() - report = await batches.coverage(s, start, now) - assert report['total'] == 9 and report['missing'] == 2 - batch = await batches.freeze(s, SimpleNamespace(start_at=start)) - admitted = set(await s.scalars(select(Package.news_id).join(BatchPackage).where(BatchPackage.batch_id == batch.id))) - assert {news['boundary'].id, late.id} <= admitted - assert not {news['done'].id, older.id, future.id} & admitted - assert packages['stale'].state == 'index_pending' - assert packages['stale'].content_hash == news['stale'].content_hash - # Moving the start backwards discovers originals missed by the former boundary. - expanded = await batches.freeze(s, SimpleNamespace(start_at=older.published_at)) - admitted = set(await s.scalars(select(Package.news_id).join(BatchPackage).where(BatchPackage.batch_id == expanded.id))) - assert older.id in admitted and future.id not in admitted - assert all(row.content_hash == 'a'*64 for row in [*news.values(), older, future, late]) - for model in (BatchEvent, BatchPackage, Batch, Membership, Attempt, Package, Event, NewsItem): - await s.execute(delete(model)) - await s.commit() - print('Backfill coverage passed: inclusive boundary, pending research, distinct news counts, late arrivals, source revisions and expanded history without model calls.') - - -async def carryover_regressions(now): - start = now - timedelta(days=2) - async with SessionLocal() as s: - keys = ['inside', 'pending', 'processing', 'failed', 'stale', 'research', 'done', 'unselected', 'future'] - news = {key: NewsItem(source_code='wscn_news', external_id='carryover-'+key, - title=key, summary='', content_text=key, content_hash='a'*64, - url='https://example.invalid/'+key, - published_at=now if key == 'inside' else - now+timedelta(days=1) if key == 'future' else start-timedelta(days=1)) - for key in keys} - s.add_all(news.values()) - await s.flush() - states = {'pending': 'index_pending', 'processing': 'assign_processing', - 'failed': 'index_failed', 'future': 'assign_pending'} - packages = {key: Package(news_id=row.id, content_hash='b'*64 if key == 'stale' else row.content_hash, - state=states.get(key, 'done'), keywords=[]) - for key, row in news.items() if key != 'unselected'} - done = Event(title='carryover-done', tags=[], state='done', result={}, revision=1, due_at=now) - pending = Event(title='carryover-pending', tags=[], state='pending', revision=1, due_at=now) - batch = Batch(phase='assigning', cutoff_at=now) - s.add_all([*packages.values(), done, pending, batch]) - await s.flush() - for key, event in [('inside', done), ('done', done), ('research', done), ('research', pending)]: - s.add(Membership(package_id=packages[key].id, event_id=event.id, - evidence=key, reason='carryover fixture', change_kind='new_fact')) - for key in ['pending', 'processing', 'stale']: - s.add(BatchPackage(batch_id=batch.id, package_id=packages[key].id, - content_hash=packages[key].content_hash, state='pending')) - s.add(BatchEvent(batch_id=batch.id, event_id=pending.id, revision=1, state='pending')) - await s.commit() - wide = await batches.coverage(s, start-timedelta(days=2), now) - assert wide['outside']['unprocessed'] == 1 # Explicitly enrolled future original. - narrow = await batches.coverage(s, start, now) - assert narrow['total'] == 1 and narrow['completed'] == 1 and narrow['unprocessed'] == 0 - assert narrow['outside'] == {'unprocessed': 6, 'pending': 3, 'processing': 1, 'failed': 1, 'stale': 1}, narrow - assert await s.scalar(select(func.count()).select_from(BatchPackage)) == 3 - assert await s.scalar(select(func.count()).select_from(BatchEvent)) == 1 - assert await s.scalar(select(func.count()).select_from(Package)) == 8 - assert batch.phase == 'assigning' and packages['pending'].state == 'index_pending' - for model in (BatchEvent, BatchPackage, Batch, Membership, Attempt, Package, Event, NewsItem): - await s.execute(delete(model)) - await s.commit() - print('Carryover coverage passed: narrowing preserves and reports outside queued news, active work, changed originals, pending research and explicit future enrollment without counting completed or unselected originals.') - - -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) - original = NewsItem(source_code='wscn_news', external_id='retry-member', title='retry', summary='', content_text='retry', content_hash='a'*64, url='https://example.invalid/retry', published_at=now) - s.add_all([event, previous, original]) - await s.flush() - member = Package(news_id=original.id, content_hash=original.content_hash, state='done', keywords=[]) - s.add(member) - await s.flush() - s.add(Membership(event_id=event.id, package_id=member.id, evidence='retry', reason='retry', change_kind='new_fact')) - 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]) - assert await batches.freeze(s, settings) is None - package = await s.scalar(select(Package).where(Package.news_id == news.id)) - news.content_hash = 'b'*64 - await s.commit() - assert await batches.freeze(s, settings) is None - assert package.content_hash != news.content_hash - expanded = SimpleNamespace(start_at=news.published_at) - batch = await batches.freeze(s, expanded) - assert package.content_hash == news.content_hash and package.state == 'index_pending' - assert ('index', package.id) in await batches.actions(s, batch, set(), 8) - 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)) == 30 - assert await s.scalar(select(func.count()).select_from(Package)) == 30 - # 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)) == 30 - 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.') - - -async def queue_review_regressions(now, settings): - async def clear(session): - 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: - news = NewsItem(source_code='wscn_news', external_id='review-barrier', title='historical', - summary='', content_text='historical', content_hash='a'*64, - url='https://example.invalid/review-barrier', published_at=now) - s.add(news) - await s.flush() - package = Package(news_id=news.id, content_hash=news.content_hash, state='assign_pending', keywords=[]) - event = Event(title='historical event', tags=[], revision=1, state='pending', due_at=now) - batch = Batch(phase='assigning', cutoff_at=now) - s.add_all([package, event, batch]) - await s.flush() - s.add(Membership(event_id=event.id, package_id=package.id, - evidence='historical', reason='same event', change_kind='new_fact')) - s.add(BatchPackage(batch_id=batch.id, package_id=package.id, content_hash=package.content_hash, state='pending')) - await s.commit() - unrelated = package.id + 1000 - # Assignment remains globally serial even when the other call is not in this batch. - assert await batches.actions(s, batch, {('assign', unrelated)}, 7) == [] - assert batch.phase == 'assigning' - # Active work on this batch still holds its assignment barrier. - package.state = 'done' - await s.commit() - assert await batches.actions(s, batch, {('assign', package.id)}, 7) == [] - assert batch.phase == 'assigning' - # Unrelated realtime package and event calls must not hold the barrier. - realtime = {('index', unrelated), ('event', event.id + 1000)} - assert await batches.actions(s, batch, realtime, 6) == [] - await s.commit() - assert batch.phase == 'researching' - assert await batches.actions(s, batch, realtime, 6) == [('event', event.id)] - await s.commit() - event.state = 'done' - await s.commit() - # A completed result can be visible before the owning task has returned. - assert await batches.actions(s, batch, realtime | {('event', event.id)}, 5) == [] - assert batch.phase == 'researching' - assert await batches.actions(s, batch, realtime | {('index', package.id)}, 5) == [] - assert batch.phase == 'researching' - assert await batches.actions(s, batch, realtime, 6) == [] - await s.commit() - assert batch.phase == 'done' - # Completed batches release admission of the next historical originals. - news2 = NewsItem(source_code='wscn_news', external_id='review-next', title='next', summary='', - content_text='next', content_hash='b'*64, - url='https://example.invalid/review-next', published_at=now) - s.add(news2) - await s.commit() - assert (await batches.freeze(s, settings)).id != batch.id - await clear(s) - - # Thirty manual-only deferred heads cannot consume the thirty runnable slots. - news = NewsItem(source_code='wscn_news', external_id='review-deferred', title='source', summary='', - content_text='source', content_hash='a'*64, - url='https://example.invalid/review-deferred', published_at=now) - s.add(news) - await s.flush() - package = Package(news_id=news.id, content_hash=news.content_hash, state='done', keywords=[]) - previous = Batch(phase='partial', cutoff_at=now) - events = [Event(title='deferred-head-'+str(i), tags=[], revision=7, state='pending', - due_at=now-timedelta(minutes=35-i)) for i in range(35)] - s.add_all([package, previous, *events]) - await s.flush() - for i, event in enumerate(events): - s.add(Membership(event_id=event.id, package_id=package.id, evidence='source', reason='same source', change_kind='new_fact')) - if i <= 30: - s.add(BatchEvent(batch_id=previous.id, event_id=event.id, - revision=7 if i < 30 else 6, state='deferred')) - await s.commit() - batch = await batches.freeze(s, settings) - assert batch is not None - selected = set(await s.scalars(select(BatchEvent.event_id).where(BatchEvent.batch_id == batch.id))) - assert selected == {event.id for event in events[30:]}, selected - assert all(event.state == 'pending' and event.revision == 7 for event in events[:30]) - await batches.actions(s, batch, set(), 8) - await s.commit() - actions = await batches.actions(s, batch, set(), 8) - assert set(actions) == {('event', event.id) for event in events[30:]} - await clear(s) - print('Queue review regressions passed: unrelated realtime calls allow both batch barriers to advance, owning calls still block, assignment stays serial, and deferred heads cannot starve runnable events.') - - -asyncio.run(main()) diff --git a/scripts/smoke_priority_queue.py b/scripts/smoke_priority_queue.py index fbce80e..e147877 100644 --- a/scripts/smoke_priority_queue.py +++ b/scripts/smoke_priority_queue.py @@ -10,7 +10,7 @@ from unittest.mock import patch from alphapulse.core.db import SessionLocal, engine from alphapulse.models import NewsItem from alphapulse.news_research.models import Package, Event, Membership, Attempt, Version, ResearchSettings -from alphapulse.news_research import scheduling, batches, service +from alphapulse.news_research import scheduling, service from alphapulse.news_research.worker import recover_interrupted @@ -29,7 +29,7 @@ async def main(): live = original('live', now-timedelta(minutes=30)) late = original('late', now-timedelta(days=3)) s.add_all([old, live, late]); await s.commit() - batch = await batches.freeze(s, config) + await scheduling.lane_actions(s, config, set(), 8, False) packages = list(await s.scalars(select(Package))) assert len(packages) == 2 and all(p.news_id != live.id for p in packages) actions = await scheduling.realtime_actions(s, config, {('index', packages[0].id)}, 1) @@ -40,10 +40,9 @@ async def main(): config.start_at = now-timedelta(minutes=5) actions = await scheduling.realtime_actions(s, config, set(), 1) assert actions == [('index', live_package.id)] - await batches.actions(s, batch, set(), 8, config) - assert not await batches.freeze(s, config) + assert not await scheduling.lane_actions(s, config, set(), 8, False) config.start_at = now-timedelta(days=5) - assert await batches.freeze(s, config) + assert await scheduling.lane_actions(s, config, set(), 8, False) assert packages[0].keywords == [] # A realtime event's historical dependencies inherit priority. event = Event(title='shared', tags=[], state='pending', revision=1, realtime_pending=True, due_at=now) diff --git a/web/src/common/apis/news-research/type.ts b/web/src/common/apis/news-research/type.ts index df388b2..9ae6a3e 100644 --- a/web/src/common/apis/news-research/type.ts +++ b/web/src/common/apis/news-research/type.ts @@ -1,8 +1,8 @@ export interface Keyword { canonical: string, aliases: string[], quote: string } export interface ModelConfig { provider_id: number | null, model: string, reasoning_effort: string | null, skill_id: number | null } export interface Config { index: ModelConfig, research: ModelConfig, concurrency: number } -export interface BatchProgress { id: number, phase: string, cutoff_at: string, packages_total: number, packages_done: number, packages_failed: number, packages_deferred: number, events_total: number, events_done: number, events_failed: number, events_deferred: number } -export interface Settings { control_revision: number, realtime_from: string | null, queue: { running: number, draining: boolean, realtime: Record, backfill: Record }, batch?: BatchProgress | null, config: Partial, enabled: boolean, start_at: string } +export interface QueueStatus { running: number, capacity: number, running_by_stage: { index: number, assign: number, research: number }, draining: boolean, realtime: Record, backfill: Record, events: { realtime: Record, backfill: Record }, idle_reasons: string[] } +export interface Settings { control_revision: number, realtime_from: string | null, queue: QueueStatus, batch?: null, config: Partial, enabled: boolean, start_at: string } export interface Coverage { start_at: string, checked_at: string, total: number, unprocessed: number, missing: number, stale: number, pending: number, processing: number, failed: number, completed: number, outside: { unprocessed: number, stale: number, pending: number, processing: number, failed: number } } export interface Package { id: number, news_id: number, title: string, url: string, state: string, error: string | null, keywords: Keyword[], events: { id: number, title: string, reason: string, quote: string }[] } export interface Point { time_precision: "minute" | "day" | "month" | "unknown", planned_precision: "minute" | "day" | "month" | "unknown", package_ids: number[], time_text: string, description: string, occurred_at: string | null, planned_at: string | null, reported_at: string | null, period_text: string, evidence_refs: { article_id: number, quote: string }[] } diff --git a/web/src/pages/home/index.vue b/web/src/pages/home/index.vue index 5830b86..004b5cd 100644 --- a/web/src/pages/home/index.vue +++ b/web/src/pages/home/index.vue @@ -1,5 +1,5 @@