diff --git a/backend/alembic/versions/20261106_dd4567890123_priority_queue.py b/backend/alembic/versions/20261106_dd4567890123_priority_queue.py new file mode 100644 index 0000000..2680a29 --- /dev/null +++ b/backend/alembic/versions/20261106_dd4567890123_priority_queue.py @@ -0,0 +1,24 @@ +"""Persist realtime boundaries and enforce one active call per subject.""" +from alembic import op +import sqlalchemy as sa + +revision = 'dd4567890123' +down_revision = 'dc3456789012' +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column('news_research_settings', sa.Column('realtime_from', sa.DateTime(timezone=True))) + op.add_column('news_research_events', sa.Column('realtime_pending', sa.Boolean(), nullable=False, server_default=sa.false())) + op.create_index('uq_news_attempt_active', 'news_research_attempts', ['package_id'], unique=True, + postgresql_where=sa.text("state = 'processing'")) + op.create_index('uq_news_version_active', 'news_research_versions', ['event_id'], unique=True, + postgresql_where=sa.text("state = 'processing'")) + + +def downgrade(): + op.drop_index('uq_news_version_active', table_name='news_research_versions') + op.drop_index('uq_news_attempt_active', table_name='news_research_attempts') + op.drop_column('news_research_events', 'realtime_pending') + op.drop_column('news_research_settings', 'realtime_from') diff --git a/backend/src/alphapulse/news_research/api.py b/backend/src/alphapulse/news_research/api.py index b92558a..35c05fd 100644 --- a/backend/src/alphapulse/news_research/api.py +++ b/backend/src/alphapulse/news_research/api.py @@ -20,8 +20,9 @@ from alphapulse.news_research.models import ( Version, Reference, BatchEvent, + Batch, ) -from alphapulse.news_research import service, batches, vocabulary as vocab +from alphapulse.news_research import service, batches, scheduling, vocabulary as vocab from alphapulse.news_research.models import VocabularyVersion from alphapulse.news_research.contracts import ( INDEX_ASSIGN_SKILL, @@ -90,6 +91,8 @@ class SettingsInput(Strict): index: ModelConfig research: ModelConfig concurrency: int = Field(default=3, ge=1, le=16) + start_at: AwareDatetime | None = None + base_revision: int | None = Field(default=None, ge=0) async def settings_row(session): @@ -101,7 +104,7 @@ async def settings_row(session): id=1, config={}, enabled=False, - start_at=datetime(2026, 10, 31, 16, tzinfo=UTC), + start_at=datetime(2026, 11, 4, 16, tzinfo=UTC), ) .on_conflict_do_nothing(index_elements=["id"]) ) @@ -115,6 +118,8 @@ async def settings(session: DB): row = await settings_row(session) return success( {"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)} ) @@ -125,21 +130,24 @@ async def coverage(session: DB, start_at: AwareDatetime | None = None): start = (start_at if start_at is not None else row.start_at).astimezone(UTC) if start > datetime.now(UTC): raise HTTPException(422, "回补起始时间不能晚于当前时间") - return success(await batches.coverage(session, start)) + effective = min(start, row.realtime_from) if row.realtime_from else start + report = await batches.coverage(session, effective) + return success({**report, 'backfill_start_at': start, 'realtime_from': row.realtime_from}) @router.put("/settings") async def save(body: SettingsInput, session: DB): await settings_row(session) - await session.execute( - update(ResearchSettings) - .where(ResearchSettings.id == 1) - .values( - config=body.model_dump(), - enabled=False, - control_revision=ResearchSettings.control_revision + 1, - ) - ) + row = await session.get(ResearchSettings, 1, populate_existing=True, with_for_update=True) + if body.base_revision is not None and body.base_revision != row.control_revision: + raise HTTPException(409, "设置已变化，请重新读取后保存") + if body.start_at is not None: + if body.start_at > datetime.now(UTC): + raise HTTPException(422, "回补起始时间不能晚于当前时间") + row.start_at = body.start_at.astimezone(UTC) + row.config = body.model_dump(exclude={"start_at", "base_revision"}) + row.enabled = False + row.control_revision += 1 await session.commit() return success(None) @@ -164,6 +172,8 @@ class EnableInput(Strict): async def enable(session: DB, body: EnableInput | None = None): row = await settings_row(session) start_at = (body.start_at if body is not None else row.start_at).astimezone(UTC) + if start_at != row.start_at: + raise HTTPException(422, "请在模型与 Skills 中保存回补时间并暂停后再启用") if start_at > datetime.now(UTC): raise HTTPException(422, "回补起始时间不能晚于当前时间") if not await session.scalar(text("SELECT pg_try_advisory_xact_lock(20261103, 2)")): @@ -172,11 +182,11 @@ async def enable(session: DB, body: EnableInput | None = None): raise HTTPException(409, "请先暂停两阶段处理") running = await session.scalar( select(func.count()) - .select_from(Package) - .where(Package.state.in_(["index_processing", "assign_processing"])) + .select_from(Attempt) + .where(Attempt.state == "processing") ) running += await session.scalar( - select(func.count()).select_from(Event).where(Event.state == "processing") + select(func.count()).select_from(Version).where(Version.state == "processing") ) if running: raise HTTPException(409, "请等待已接收任务完成后再验证") @@ -259,7 +269,10 @@ 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) - row.start_at = 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() @@ -480,6 +493,18 @@ async def event(identifier: int, session: DB): .order_by(Version.id.desc()) ) ) + wait_reason = None + 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): + wait_reason = '范围外成员尚未完成，扩大回补范围后可继续研究' + elif any(p.state.endswith('_failed') for p, n in unfinished): + wait_reason = '成员包处理失败，需手动重试后才能研究' + elif unfinished: + wait_reason = '等待成员包完成关键词与事件归属' + elif row.due_at > datetime.now(UTC): + wait_reason = '正在汇集事件变化，聚合窗口结束后研究' latest = next((v for v in versions if v.state == "done"), None) references = await version_references(session, latest.id) if latest else [] return success( @@ -490,6 +515,7 @@ async def event(identifier: int, session: DB): "result": row.result, "error": row.error, "revision": row.revision, + "wait_reason": wait_reason, "packages": [await package_out(session, p, n) for p, n in members], "references": references, "versions": [ diff --git a/backend/src/alphapulse/news_research/batches.py b/backend/src/alphapulse/news_research/batches.py index 0f2b8b3..273c878 100644 --- a/backend/src/alphapulse/news_research/batches.py +++ b/backend/src/alphapulse/news_research/batches.py @@ -5,6 +5,7 @@ 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): @@ -48,13 +49,22 @@ async def coverage(session, start_at, cutoff=None): 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( - 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"))) + 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, @@ -63,8 +73,6 @@ async def freeze(session, settings): 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 == any_(event_cutoff), Event.state == 'pending', ~deferred))) if not packages and not events: return None @@ -83,7 +91,7 @@ async def freeze(session, settings): return batch -async def actions(session, batch, active, slots): +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( @@ -94,7 +102,9 @@ async def actions(session, batch, active, slots): for item, package, news in rows: if item.state != 'pending': continue - if item.content_hash != news.content_hash or item.content_hash != package.content_hash: + 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' @@ -102,10 +112,12 @@ async def actions(session, batch, active, slots): item.state = 'failed' elif package.state in ('index_pending', 'assign_pending'): stage = 'index' if package.state == 'index_pending' else 'assign' - if (stage, package.id) not in active and len(result) < slots and (stage != 'assign' or not assigning): + 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' - if any(item.state == 'pending' for item, _, _ in rows) or active: + 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. @@ -131,7 +143,13 @@ async def actions(session, batch, active, slots): for item, event in rows: if item.state not in ('pending', 'urgent_pending', 'processing') or ('event', event.id) in active: continue - if item.revision != event.revision: + 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' @@ -143,11 +161,17 @@ async def actions(session, batch, active, slots): 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 = 'deferred' + item.state = 'blocked' elif len(result) < slots: item.state = 'processing' result.append(('event', event.id)) - if not result and not active and all(item.state not in ('pending', 'urgent_pending', 'processing') for item, _ in rows): + 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))) @@ -163,5 +187,5 @@ async def progress(session): 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')}}) + 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/models.py b/backend/src/alphapulse/news_research/models.py index bdecd23..ed07167 100644 --- a/backend/src/alphapulse/news_research/models.py +++ b/backend/src/alphapulse/news_research/models.py @@ -6,6 +6,8 @@ from sqlalchemy import ( DateTime, ForeignKey, Integer, + Index, + text, String, Text, UniqueConstraint, @@ -32,6 +34,7 @@ class ResearchSettings(Base): Integer, default=0, server_default="0" ) start_at: Mapped[datetime] = mapped_column(DateTime(timezone=True)) + realtime_from: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) class Package(Base): @@ -49,6 +52,7 @@ class Package(Base): class Attempt(Base): __tablename__ = "news_research_attempts" + __table_args__ = (Index("uq_news_attempt_active", "package_id", unique=True, postgresql_where=text("state = 'processing'")),) id: Mapped[int] = mapped_column(BigInteger, primary_key=True) package_id: Mapped[int] = mapped_column( ForeignKey("news_research_packages.id"), index=True @@ -75,6 +79,7 @@ class Event(Base): state: Mapped[str] = mapped_column(String(16), default="pending", index=True) result: Mapped[dict | None] = mapped_column(JSONB) error: Mapped[str | None] = mapped_column(Text) + realtime_pending: Mapped[bool] = mapped_column(default=False, server_default="false") dirty_since: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) due_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), index=True @@ -99,6 +104,7 @@ class Membership(Base): class Version(Base): __tablename__ = "news_research_versions" + __table_args__ = (Index("uq_news_version_active", "event_id", unique=True, postgresql_where=text("state = 'processing'")),) id: Mapped[int] = mapped_column(BigInteger, primary_key=True) event_id: Mapped[int] = mapped_column( ForeignKey("news_research_events.id"), index=True diff --git a/backend/src/alphapulse/news_research/scheduling.py b/backend/src/alphapulse/news_research/scheduling.py new file mode 100644 index 0000000..73c7443 --- /dev/null +++ b/backend/src/alphapulse/news_research/scheduling.py @@ -0,0 +1,103 @@ +"""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 alphapulse.models import NewsItem +from alphapulse.news_research.models import Package, Event, Membership, Attempt, Version, ResearchSettings + + +def boundary(settings): + realtime_from = getattr(settings, 'realtime_from', None) + return min(settings.start_at, realtime_from) if realtime_from else settings.start_at + + +def eligible(settings): + return and_(NewsItem.published_at >= boundary(settings), NewsItem.published_at <= datetime.now(UTC)) + + +def members(event_id): + return select(Membership.package_id).where(Membership.event_id == event_id) + + +def ready_event(): + return ~exists(select(Membership.package_id).join(Package).join(NewsItem).where( + Membership.event_id == Event.id, + or_(Package.state != 'done', Package.content_hash != NewsItem.content_hash))) + + +def scoped_event(settings): + # Empty events must still reach the existing no-model retirement path. + return or_(exists(select(Membership.package_id).join(Package).join(NewsItem).where( + Membership.event_id == Event.id, eligible(settings))), + ~exists(select(Membership.package_id).where(Membership.event_id == Event.id))) + + +async def control(session, config): + """The control lock spans the durable claim, never the remote model call.""" + if '_control_revision' not in config: + return None # Explicit bounded/offline callers retain their existing contract. + row = await session.get(ResearchSettings, 1, populate_existing=True, with_for_update=True) + if not row.enabled or row.control_revision != config['_control_revision']: + return False + return row + + +async def mark_priority(session, event, news): + settings = await session.get(ResearchSettings, 1) + if settings and settings.realtime_from and news.published_at >= settings.realtime_from: + 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. + 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 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} + for name in ('realtime', 'backfill')} + return {'running': running, 'draining': not settings.enabled and running > 0, **queues} diff --git a/backend/src/alphapulse/news_research/service.py b/backend/src/alphapulse/news_research/service.py index ef136ce..8409f90 100644 --- a/backend/src/alphapulse/news_research/service.py +++ b/backend/src/alphapulse/news_research/service.py @@ -27,6 +27,7 @@ from alphapulse.news_research.models import ( Version, Reference, ) +from alphapulse.news_research.scheduling import control, mark_priority, boundary from alphapulse.news_research.contracts import ( INDEX_ASSIGN_SKILL, task_skill, @@ -280,6 +281,7 @@ async def enroll(session, ids): .with_for_update() ): dirty(event) + await mark_priority(session, event, row) await session.commit() @@ -429,6 +431,9 @@ async def process_package(identifier, config): async def _process_package(identifier, config, expected_stage): async with SessionLocal() as session: + settings = await control(session, config) + if settings is False: + return package = await session.get(Package, identifier, with_for_update=True) if package is None: return @@ -438,6 +443,10 @@ async def _process_package(identifier, config, expected_stage): if package.content_hash != news.content_hash: await enroll(session, [news.id]) return + if settings is not None and not boundary(settings) <= news.published_at <= datetime.now(UTC): + return + if await session.scalar(select(Attempt.id).where(Attempt.package_id == identifier, Attempt.state == 'processing').limit(1)): + return source = article(news) source_hash = package.content_hash stage = "index" if package.state == "index_pending" else "assign" @@ -566,12 +575,14 @@ async def _process_package(identifier, config, expected_stage): existing.change_kind = target.change_kind if target.change_kind != "repost" or event.result is None: dirty(event, target.urgent) + await mark_priority(session, event, news) for old in previous: if old.event_id not in assigned: event = await session.get( Event, old.event_id, with_for_update=True ) dirty(event) + await mark_priority(session, event, news) await session.delete(old) affected[event.id] = event package.state = "done" @@ -805,6 +816,9 @@ def validate_timeline(result, package_news): async def process_event(identifier, config, force=False): async with SessionLocal() as session: + settings = await control(session, config) + if settings is False: + return event = await session.get(Event, identifier, with_for_update=True) if ( event is None @@ -812,6 +826,8 @@ 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)): + return members = list( await session.execute( select(Package, Membership, NewsItem) @@ -821,6 +837,8 @@ async def process_event(identifier, config, force=False): .order_by(Package.id) ) ) + if settings is not None and members and not any(boundary(settings) <= n.published_at <= datetime.now(UTC) for p, m, n in members): + return if not members: event.state = "retired" await session.commit() @@ -859,6 +877,7 @@ async def process_event(identifier, config, force=False): ) session.add(version) event.state = "processing" + event.dirty_since = None await session.commit() version_id, revision = version.id, event.revision researcher = Research(event={"id": identifier, "revision": revision}) @@ -977,6 +996,7 @@ async def process_event(identifier, config, force=False): current.title = output.title current.state = "done" current.dirty_since = None + current.realtime_pending = False current.error = None for ref in output.references: session.add( diff --git a/backend/src/alphapulse/news_research/worker.py b/backend/src/alphapulse/news_research/worker.py index 2e3fd8d..46d3ec2 100644 --- a/backend/src/alphapulse/news_research/worker.py +++ b/backend/src/alphapulse/news_research/worker.py @@ -13,38 +13,42 @@ from alphapulse.news_research.models import ( Batch, BatchEvent, ) from alphapulse.news_research.service import process_package, process_event -from alphapulse.news_research import batches +from alphapulse.news_research import batches, scheduling logger = logging.getLogger(__name__) +async def recover_interrupted(): + async with SessionLocal() as session: + for stage in ("index", "assign"): + await session.execute( + update(Package) + .where(Package.state == stage + "_processing") + .values(state=stage + "_failed", error="进程中断，调用结果不明，请手动重试") + ) + await session.execute( + update(Attempt) + .where(Attempt.state == "processing") + .values(state="failed", error="进程中断，调用结果不明，请手动重试") + ) + await session.execute( + update(Event).where(Event.state == "processing").values(state="failed", error="进程中断，调用结果不明，请手动重试") + ) + await session.execute( + update(Version) + .where(Version.state == "processing") + .values(state="failed", error="进程中断，调用结果不明，请手动重试") + ) + await session.execute(update(BatchEvent).where(BatchEvent.state == 'processing').values(state='pending')) + await session.commit() + + async def run(): async with engine.connect() as lock: if not await lock.scalar(text("SELECT pg_try_advisory_lock(20261030, 1)")): raise RuntimeError("新闻AI worker 已运行") await lock.commit() - async with SessionLocal() as session: - for stage in ("index", "assign"): - await session.execute( - update(Package) - .where(Package.state == stage + "_processing") - .values(state=stage + "_pending") - ) - await session.execute( - update(Attempt) - .where(Attempt.state == "processing") - .values(state="failed", error="进程中断，待重试") - ) - await session.execute( - update(Event).where(Event.state == "processing").values(state="pending") - ) - await session.execute( - update(Version) - .where(Version.state == "processing") - .values(state="failed", error="进程中断，待重试") - ) - await session.execute(update(BatchEvent).where(BatchEvent.state == 'processing').values(state='pending')) - await session.commit() + await recover_interrupted() tasks = {} try: while True: @@ -58,20 +62,21 @@ async def run(): async with SessionLocal() as session: settings = await session.get(ResearchSettings, 1) 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) - settings = await session.get(ResearchSettings, 1, populate_existing=True, with_for_update=True) - if batch is not None and settings.enabled: - active = set(tasks.values()) - actions = await batches.actions(session, batch, active, - max(0, settings.config.get('concurrency', 3) - len(tasks))) - await session.commit() - for stage, identifier in actions: - call = (process_event(identifier, settings.config, force=True) - if stage == 'event' else process_package(identifier, settings.config)) - tasks[asyncio.create_task(call)] = (stage, identifier) + if batch is not None: + actions += await batches.actions(session, batch, active | set(actions), slots - len(actions), settings) + 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) 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 e13d2e5..a427f8a 100644 --- a/backend/tests/test_news_research_workspace.py +++ b/backend/tests/test_news_research_workspace.py @@ -302,7 +302,7 @@ def test_failed_enable_does_not_save_the_selected_backfill_start(trading, monkey session = N(get=AsyncMock(return_value=row), scalar=AsyncMock(side_effect=[True, 0, 0]), commit=AsyncMock()) monkeypatch.setattr(service, "model_config", AsyncMock(side_effect=ValueError("Skill 不可用"))) with pytest.raises(HTTPException) as error: - asyncio.run(api.enable(session, api.EnableInput(start_at=previous - timedelta(days=30)))) + asyncio.run(api.enable(session, api.EnableInput(start_at=previous))) assert error.value.status_code == 502 assert row.start_at == previous and not row.enabled and row.control_revision == 7 session.commit.assert_not_called() @@ -842,6 +842,7 @@ def test_alias_adjustment_audit_survives_index_failures(trading, monkeypatch, fa session.get = AsyncMock(side_effect=lambda model, *args, **kwargs: attempts[0] if model is Attempt else package if model is Package else news) session.execute = AsyncMock() + session.scalar = AsyncMock(return_value=None) monkeypatch.setattr(service, "SessionLocal", lambda: session) monkeypatch.setattr(service.vocab, "snapshot", AsyncMock(return_value={"entries": []})) monkeypatch.setattr(service, "model_config", AsyncMock(return_value=(None, "索引规则", {}))) @@ -903,3 +904,47 @@ def test_task_skill_keeps_custom_rules_and_accepts_crlf(trading): content = "## 归属规则\r\n自定义归属\r\n### 细则\r\n保留规则\r\n## 索引规则\r\n自定义索引" assert task_skill(content, "assign") == "自定义归属\n### 细则\n保留规则" assert task_skill(content, "index") == "自定义索引" + + +def test_settings_save_atomically_pauses_and_preserves_model_contract(trading): + import asyncio + from unittest.mock import AsyncMock + from alphapulse.news_research import api + row = N(start_at=datetime(2026, 11, 4, 16, tzinfo=UTC), config={}, enabled=True, control_revision=3) + session = N(get=AsyncMock(return_value=row), commit=AsyncMock()) + model = dict(provider_id=1, model='test', skill_id=1) + body = api.SettingsInput(index=model, research=model, start_at='2026-10-20T00:00:00+08:00', base_revision=3) + asyncio.run(api.save(body, session)) + assert not row.enabled and row.control_revision == 4 + assert row.start_at == datetime(2026, 10, 19, 16, tzinfo=UTC) + assert set(row.config) == {'index', 'research', 'concurrency'} + session.commit.assert_awaited_once() + + +def test_settings_save_rejects_other_tabs_revision_without_changes(trading): + import asyncio + from unittest.mock import AsyncMock + from fastapi import HTTPException + from alphapulse.news_research import api + row = N(start_at=datetime(2026, 11, 4, 16, tzinfo=UTC), config={}, enabled=True, control_revision=4) + session = N(get=AsyncMock(return_value=row), commit=AsyncMock()) + model = dict(provider_id=1, model='test', skill_id=1) + with pytest.raises(HTTPException) as error: + asyncio.run(api.save(api.SettingsInput(index=model, research=model, base_revision=3), session)) + assert error.value.status_code == 409 + assert row.enabled and row.control_revision == 4 + session.commit.assert_not_awaited() + + +def test_enable_cannot_apply_an_unsaved_backfill_time(trading): + import asyncio + from unittest.mock import AsyncMock + from fastapi import HTTPException + from alphapulse.news_research import api + row = N(start_at=datetime(2026, 11, 4, 16, tzinfo=UTC)) + session = N(get=AsyncMock(return_value=row), scalar=AsyncMock(), commit=AsyncMock()) + with pytest.raises(HTTPException) as error: + asyncio.run(api.enable(session, api.EnableInput(start_at='2026-10-20T00:00:00+08:00'))) + assert error.value.status_code == 422 + session.scalar.assert_not_awaited() + session.commit.assert_not_awaited() diff --git a/deploy/README.md b/deploy/README.md index d096014..c4f64fd 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -455,3 +455,42 @@ some documents were imported: ECB dynamic index version coverage, inaccessible BoE speeches and CoinDesk access blocks are reported rather than labeled complete. The deployment lock serializes writes; crawler/calendar processes are restored in finally blocks. Do not start a second driver or remove its state to renew quotas. + +### Realtime priority and historical AI backfill + +Migration `dd4567890123` adds the persistent realtime boundary and pending-event +priority, plus partial unique indexes for active package attempts and event +versions. Pause and drain news AI before applying it; duplicate active records +must be investigated rather than deleted to force migration success. Preserve +originals, attempts and version history. Deployment requires the candidate's +schema head and must leave automatic processing paused during verification. + +The model/Skills drawer saves the historical start time with model settings via +**Save and pause**. New settings default to 2026-11-05 00:00 Asia/Shanghai; existing +saved values remain intact. Settings responses expose `control_revision`, +`realtime_from`, `timezone` and queue counts. PUT accepts optional `start_at` and +`base_revision`; version conflicts return 409. Enable uses the saved start only; +a differing legacy enable body returns 422. The first successful enable fixes +`realtime_from`; later saves, pauses and restarts never move it. + +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. + +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 +aggregation window and obsolete output cannot overwrite current conclusions. +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 +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 ea63643..df395a8 100644 --- a/scripts/ci.py +++ b/scripts/ci.py @@ -182,6 +182,8 @@ def smoke(build, label): '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) + 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', '-', input=(ROOT / 'scripts/smoke_news_research.py').read_text()), flush=True) print(run('docker', 'exec', '-i', names['api'], 'python', '-', diff --git a/scripts/smoke_news_batches.py b/scripts/smoke_news_batches.py index 38e1839..9b78970 100644 --- a/scripts/smoke_news_batches.py +++ b/scripts/smoke_news_batches.py @@ -90,6 +90,7 @@ async def main(): 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.') @@ -208,7 +209,13 @@ async def regressions(now, settings): 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]) + 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() @@ -236,39 +243,116 @@ async def regressions(now, settings): s.add_all([news, excluded]) await s.commit() await service.enroll(s, [news.id]) - batch = await batches.freeze(s, settings) + 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() - 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 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)) == count - assert await s.scalar(select(func.count()).select_from(Package)) == count + 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)) == count + 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_news_research.py b/scripts/smoke_news_research.py index f6f6977..09c7a14 100644 --- a/scripts/smoke_news_research.py +++ b/scripts/smoke_news_research.py @@ -393,14 +393,20 @@ async def main(): assert report.json()["data"]["start_at"] == "2026-09-30T16:00:00Z" assert set(report.json()["data"]["outside"]) == {"unprocessed", "stale", "pending", "processing", "failed"} assert (await client.get("/api/risk/settings")).json()["data"]["start_at"] == previous_start + assert (await client.post("/api/risk/enable", json={"start_at": selected_start})).status_code == 422 + settings = (await client.get("/api/risk/settings")).json()["data"] + saved = await client.put("/api/risk/settings", json={**config, "start_at": selected_start, "base_revision": settings["control_revision"]}) + assert saved.status_code == 200, saved.text with patch.object(model_service, "complete", validation): - response = await client.post("/api/risk/enable", json={"start_at": selected_start}) + response = await client.post("/api/risk/enable") assert response.status_code == 200, response.text assert response.json()["data"]["start_at"] == "2026-09-30T16:00:00Z" assert (await client.get("/api/risk/settings")).json()["data"][ "enabled" ] is True assert (await client.get("/api/risk/settings")).json()["data"]["start_at"] == "2026-09-30T16:00:00Z" + fixed_anchor = (await client.get("/api/risk/settings")).json()["data"]["realtime_from"] + assert fixed_anchor is not None assert (await client.post("/api/risk/enable")).status_code == 409 assert (await client.post("/api/risk/pause")).status_code == 200 for action in ("pause", "save"): @@ -421,12 +427,13 @@ async def main(): return await validation(provider, skill, payload, schema) with patch.object(model_service, "complete", interrupted_validation): - response = await client.post("/api/risk/enable", json={"start_at": "2026-09-01T00:00:00Z"}) + response = await client.post("/api/risk/enable") assert response.status_code == 409, response.text assert (await client.get("/api/risk/settings")).json()["data"][ "enabled" ] is False assert (await client.get("/api/risk/settings")).json()["data"]["start_at"] == "2026-09-30T16:00:00Z" + assert (await client.get("/api/risk/settings")).json()["data"]["realtime_from"] == fixed_anchor package_id = (await client.get(f"/api/risk/news/{news_id}/packages")).json()[ "data" ][0]["id"] diff --git a/scripts/smoke_priority_queue.py b/scripts/smoke_priority_queue.py new file mode 100644 index 0000000..fbce80e --- /dev/null +++ b/scripts/smoke_priority_queue.py @@ -0,0 +1,107 @@ +"""Exercise preemption and recovery using disposable PostgreSQL, never a model.""" +import asyncio +import os +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace +from urllib.parse import urlparse +from sqlalchemy import select, text, func +from sqlalchemy.exc import IntegrityError +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.worker import recover_interrupted + + +async def main(): + url = urlparse(os.environ['DATABASE_URL']) + assert (url.username, url.password, url.path) == ('ci', 'ci-only', '/ci') + now = datetime.now(UTC) + anchor = now - timedelta(hours=1) + config = SimpleNamespace(start_at=now-timedelta(days=5), realtime_from=anchor) + async with SessionLocal() as s: + assert await s.scalar(select(func.count()).select_from(Package)) == 0 + def original(key, published): + return NewsItem(source_code='wscn_news', external_id='priority-'+key, title=key, summary='', + content_text=key, content_hash='a'*64, url='https://example.invalid/'+key, published_at=published) + old = original('old', now-timedelta(days=4)) + 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) + 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) + live_package = await s.scalar(select(Package).where(Package.news_id == live.id)) + assert actions == [('index', live_package.id)] + assert await scheduling.realtime_actions(s, config, set(), 0) == [] + # Shrinking the history range must not exclude accumulated realtime news. + 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) + config.start_at = now-timedelta(days=5) + assert await batches.freeze(s, config) + 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) + s.add(event); await s.flush() + for p in [packages[0], live_package]: + s.add(Membership(event_id=event.id, package_id=p.id, evidence='fixture', reason='fixture', change_kind='new_fact')) + packages[0].state = 'assign_pending' + live_package.state = 'done' + await s.commit() + config.start_at = now-timedelta(minutes=5) + assert await scheduling.realtime_actions(s, config, set(), 1) == [] + config.start_at = now-timedelta(days=5) + assert await scheduling.realtime_actions(s, config, set(), 1) == [('assign', packages[0].id)] + packages[0].state = 'done'; await s.commit() + assert await scheduling.realtime_actions(s, config, set(), 1) == [('event', event.id)] + event.due_at = now+timedelta(minutes=5); await s.commit() + assert ('event', event.id) not in await scheduling.realtime_actions(s, config, set(), 8) + # Pausing invalidates a selected but not-yet-claimed task. + s.add(ResearchSettings(id=1, config={}, enabled=False, control_revision=9, start_at=config.start_at, realtime_from=anchor)) + await s.commit() + assert await scheduling.control(s, {'_control_revision': 8}) is False + await s.commit() + with patch.object(service, 'complete', side_effect=AssertionError('Paused selection must not call a model')): + await service.process_package(live_package.id, {'_control_revision': 8}) + assert await s.scalar(select(func.count()).select_from(Attempt)) == 0 + live_package = await s.get(Package, live_package.id) + live_package.state = 'index_processing' + s.add(Attempt(package_id=live_package.id, content_hash='a'*64, stage='index', state='processing', input={}, rules={})) + event = await s.get(Event, event.id); event.state='processing' + s.add(Version(event_id=event.id, revision=event.revision, state='processing', input={}, rules={})) + await s.commit() + for duplicate in [ + Attempt(package_id=live_package.id, content_hash='a'*64, stage='assign', state='processing', input={}, rules={}), + Version(event_id=event.id, revision=event.revision, state='processing', input={}, rules={}), + ]: + try: + async with s.begin_nested(): + s.add(duplicate) + await s.flush() + except IntegrityError: + pass + else: + raise AssertionError('Concurrent duplicate claim was accepted') + await recover_interrupted() + async with SessionLocal() as s: + assert (await s.get(Package, live_package.id)).state == 'index_failed' + assert (await s.get(Event, event.id)).state == 'failed' + assert await scheduling.realtime_actions(s, config, set(), 8) == [] + empty = Event(title='removed members', tags=[], state='pending', revision=1, realtime_pending=True, due_at=now) + s.add(empty); await s.commit() + assert ('event', empty.id) in await scheduling.realtime_actions(s, config, set(), 8) + with patch.object(service.Research, 'run', side_effect=AssertionError('Empty events must not call a model')): + await service.process_event(empty.id, {}) + await s.refresh(empty) + assert empty.state == 'retired' + await s.execute(text('TRUNCATE news_research_settings, news_research_batches, news_research_packages, news_research_events, news_items CASCADE')) + await s.commit() + await engine.dispose() + print('Priority queue passed: live preemption, stable boundary, late history, range changes, dependencies, aggregation, stale control and uncertain-call recovery.') + +asyncio.run(main()) diff --git a/web/src/common/apis/news-research/index.ts b/web/src/common/apis/news-research/index.ts index 46a6462..91c15b7 100644 --- a/web/src/common/apis/news-research/index.ts +++ b/web/src/common/apis/news-research/index.ts @@ -2,8 +2,8 @@ import type * as Research from "./type" import { request } from "@/http/axios" export const getSettingsApi = () => request({ url: "/risk/settings", method: "get" }) -export const saveSettingsApi = (data: Research.Config) => request({ url: "/risk/settings", method: "put", data }) -export const enableApi = (start_at: string) => request({ url: "/risk/enable", method: "post", data: { start_at } }) +export const saveSettingsApi = (data: Research.Config & { start_at: string, base_revision: number }) => request({ url: "/risk/settings", method: "put", data }) +export const enableApi = () => request({ url: "/risk/enable", method: "post" }) export const getCoverageApi = (start_at: string) => request({ url: "/risk/coverage", method: "get", params: { start_at } }) export const pauseApi = () => request({ url: "/risk/pause", method: "post" }) export const createSkillsApi = () => request<{ index: number, research: number }>({ url: "/risk/skill-templates", method: "post" }) diff --git a/web/src/common/apis/news-research/type.ts b/web/src/common/apis/news-research/type.ts index b4aff7e..df388b2 100644 --- a/web/src/common/apis/news-research/type.ts +++ b/web/src/common/apis/news-research/type.ts @@ -2,13 +2,13 @@ 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 { batch?: BatchProgress | null, config: Partial, enabled: boolean, start_at: string } +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 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 }[] } export interface Result { title: string, summary: string, facts: string, reasoning: string, uncertainty: string, opinions: string, counter_evidence: string, watchpoints: string, asset_impacts: string, importance_stars: number, risk_stars: number, confidence: string, verification: string, timeline_points: Point[] } export interface Event { id: number, title: string, state: string, revision: number, error: string | null, tags: string[], result: Result | null } -export interface EventDetail extends Event { packages: Package[], references: { event_id: number, title: string, version_id: number, role: string, reason: string, changed: boolean }[], versions: { id: number, revision: number, state: string, result: Result | null, error: string | null, usage: unknown, evidence: unknown }[] } +export interface EventDetail extends Event { wait_reason?: string | null, packages: Package[], references: { event_id: number, title: string, version_id: number, role: string, reason: string, changed: boolean }[], versions: { id: number, revision: number, state: string, result: Result | null, error: string | null, usage: unknown, evidence: unknown }[] } export interface PackageDetail extends Package { article: { title: string, summary: string, content: string }, attempts: { id: number, stage: string, state: string, error: string | null, output: unknown, usage: unknown }[] } export interface VocabularyEntry { canonical: string, aliases: string[] } diff --git a/web/src/pages/home/index.vue b/web/src/pages/home/index.vue index 69a96d1..5830b86 100644 --- a/web/src/pages/home/index.vue +++ b/web/src/pages/home/index.vue @@ -6,8 +6,15 @@ import * as api from "@@/apis/news-research" import { listProviderModelsApi, listProvidersApi } from "@@/apis/settings" import { listSkillTreeApi } from "@@/apis/skills" import { EFFORTS } from "@@/constants/ai" +import dayjs from "dayjs" + +import timezone from "dayjs/plugin/timezone" +import utc from "dayjs/plugin/utc" import { useUserStore } from "@/pinia/stores/user" +dayjs.extend(utc) +dayjs.extend(timezone) + const vocabularyOpened = ref(false) const vocabulary = ref({ revision: 0, entries: [] }) const vocabularyBusy = ref(false) @@ -90,8 +97,19 @@ const settingsOpened = ref(false) const enabled = ref(false) const startAt = ref(null) const savedStartAt = ref(null) +const settingsRevision = ref(0) +const queue = ref<{ running: number, draining: boolean, realtime?: Record, backfill?: Record } | null>(null) +function disabledBackfillDate(date: Date) { + return dayjs.tz(dayjs(date).format("YYYY-MM-DD"), "Asia/Shanghai").valueOf() > Date.now() +} +const startText = computed({ + get: () => startAt.value ? dayjs(startAt.value).tz("Asia/Shanghai").format("YYYY-MM-DD HH:mm:ss") : "", + set: (value: string) => { + startAt.value = value ? dayjs.tz(value, "Asia/Shanghai").toDate() : null + } +}) const selectedStartAt = computed({ - get: () => enabled.value ? savedStartAt.value : startAt.value, + get: () => enabled.value && !settingsOpened.value ? savedStartAt.value : startAt.value, set: (value: Date | null) => { startAt.value = value } @@ -134,6 +152,7 @@ async function load() { const [settings, rows] = await Promise.all([api.getSettingsApi(), tab.value === "events" ? api.listEventsApi(q.value, page.value) : api.listPackagesApi(q.value, page.value)]) if (!valid() || seq !== sequence) return enabled.value = settings.enabled + queue.value = settings.queue ?? null if (!startAt.value || startAt.value.getTime() === savedStartAt.value?.getTime()) startAt.value = new Date(settings.start_at) savedStartAt.value = new Date(settings.start_at) batch.value = settings.batch ?? null @@ -166,13 +185,7 @@ async function checkCoverage() { } async function enable() { if (!valid()) return - if (!startAt.value || !Number.isFinite(startAt.value.getTime()) || startAt.value.getTime() > Date.now()) { - ElMessage.error("请选择不晚于当前时间的回补起始时间") - return - } - const start = startAt.value.toISOString() - const result = await api.enableApi(start) - if (valid() && selectedStartAt.value?.toISOString() === start) coverage.value = result + await api.enableApi() } async function openEvent(id: number) { eventId = id @@ -248,6 +261,8 @@ async function settings() { providers.value = choices skills.value = nodes.filter(s => s.scope === "news" && s.kind === "skill") if (value.config.index && value.config.research) config.value = value.config as Config + startAt.value = new Date(value.start_at) + settingsRevision.value = value.control_revision settingsOpened.value = true } async function action(fn: () => Promise) { @@ -275,7 +290,12 @@ async function retryPackage() { if (id) await action(() => api.retryPackageApi(id)) } async function save() { - await action(() => api.saveSettingsApi(config.value)) + if (!valid() || busy.value) return + if (!startAt.value || !Number.isFinite(startAt.value.getTime()) || startAt.value.getTime() > Date.now()) { + ElMessage.error("请选择不晚于当前时间的回补起始时间") + return + } + await action(() => api.saveSettingsApi({ ...config.value, start_at: startAt.value!.toISOString(), base_revision: settingsRevision.value })) if (valid()) settingsOpened.value = false } async function prepareSkills() { @@ -364,7 +384,7 @@ onBeforeUnmount(() => { 模型与 Skills - + {{ enabled ? '暂停新任务' : '启用两阶段处理' }} @@ -377,45 +397,20 @@ onBeforeUnmount(() => {

新闻包保存原文和关键词；事件汇集相关报道，持续更新判断与时间线。

-
-
- - - 按浏览器本地时区 - - 核查遗漏 - -
-

- 启用时保存起点，自动核查并补齐该时间以来已采集的新闻；先完成关键词与归属，再集中研究事件。调整起点不会取消已有队列。 -

-

- 范围内 {{ coverage.total }} 条，全部阶段已完成 {{ coverage.completed }} 条，未完成 {{ coverage.unprocessed }} 条： - 未入队 {{ coverage.missing }}，原文更新需重判 {{ coverage.stale }}，待处理 {{ coverage.pending }}，处理中 {{ coverage.processing }}，失败 {{ coverage.failed }}。 - 失败项需手动重试，当前批次以外的遗漏排入下一批。 -

-

- 范围外另有已入队但未完成的新闻 {{ coverage.outside.unprocessed }} 条： - 原文更新需重判 {{ coverage.outside.stale }}，待处理 {{ coverage.outside.pending }}，处理中 {{ coverage.outside.processing }}，失败 {{ coverage.outside.failed }}。 - 已有任务继续按批次处理，失败项需手动重试。 -

-

- 核查失败，请重试。 -

-
+

+ 实时任务优先取得空闲并发；历史回补至 {{ savedStartAt ? dayjs(savedStartAt).tz("Asia/Shanghai").format("YYYY-MM-DD HH:mm:ss") : '—' }}（北京时间）。 + 正在完成 {{ queue.running }} 个已接收调用，收尾后可重新启用。 +

+

+ 实时队列：等待索引 {{ queue.realtime?.index_pending ?? 0 }}，等待归属 {{ queue.realtime?.assign_pending ?? 0 }}； + 回补队列：等待索引 {{ queue.backfill?.index_pending ?? 0 }}，等待归属 {{ queue.backfill?.assign_pending ?? 0 }}。 + 正在调用 {{ queue.running }}；范围外暂停、失败和中断结果不明的任务不会自动重试。 +

批次 #{{ batch.id }} · {{ batchPhases[batch.phase] ?? batch.phase }} · 新闻归属 {{ batch.packages_done }}/{{ batch.packages_total }}，失败 {{ batch.packages_failed }}，延期 {{ batch.packages_deferred }} · 事件研究 {{ batch.events_done }}/{{ batch.events_total }}，失败 {{ batch.events_failed }}，延期 {{ batch.events_deferred }}。 - 新新闻和重试请求排入下一批，研究仅使用站内资料。 + 实时任务可中途插队；回补任务分批归属后集中研究，研究仅使用站内资料。

@@ -474,6 +469,9 @@ onBeforeUnmount(() => {