diff --git a/backend/alembic/versions/20261101_e5f701234567_news_concurrency.py b/backend/alembic/versions/20261101_e5f701234567_news_concurrency.py new file mode 100644 index 0000000..554f370 --- /dev/null +++ b/backend/alembic/versions/20261101_e5f701234567_news_concurrency.py @@ -0,0 +1,28 @@ +"""Add the news pipeline concurrency limit and durable item states.""" +from alembic import op +import sqlalchemy as sa + +revision = 'e5f701234567' +down_revision = 'd4e6f8012345' +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column('risk_settings', sa.Column('max_concurrency', sa.Integer(), server_default='5', nullable=False)) + op.create_check_constraint('risk_concurrency_positive', 'risk_settings', 'max_concurrency > 0') + op.execute("""UPDATE news_ai_reprocess_jobs j SET items = ( + SELECT COALESCE(jsonb_agg(item || jsonb_build_object('state', CASE + WHEN ordinal > j.position THEN 'pending' + WHEN EXISTS (SELECT 1 FROM jsonb_array_elements(j.failures) f + WHERE f->>'stage' = 'grouping' AND f->>'id' = item->>'id') THEN 'failed' + ELSE 'done' END) ORDER BY ordinal), '[]'::jsonb) + FROM jsonb_array_elements(j.items) WITH ORDINALITY AS entries(item, ordinal))""") + + +def downgrade(): + op.execute("""UPDATE news_ai_reprocess_jobs SET items = ( + SELECT COALESCE(jsonb_agg(item - 'state' ORDER BY CASE WHEN item->>'state' IN ('done', 'failed') THEN 0 ELSE 1 END, ordinal), '[]'::jsonb) + FROM jsonb_array_elements(items) WITH ORDINALITY AS entries(item, ordinal))""") + op.drop_constraint('risk_concurrency_positive', 'risk_settings', type_='check') + op.drop_column('risk_settings', 'max_concurrency') diff --git a/backend/src/alphapulse/api/risk.py b/backend/src/alphapulse/api/risk.py index 2e2fded..6eae324 100644 --- a/backend/src/alphapulse/api/risk.py +++ b/backend/src/alphapulse/api/risk.py @@ -4,7 +4,7 @@ from enum import IntEnum from typing import Annotated from fastapi import APIRouter, Body, Depends, HTTPException, Query from pydantic import BaseModel, Field, model_validator -from sqlalchemy import select, func +from sqlalchemy import select, func, text from sqlalchemy.ext.asyncio import AsyncSession from alphapulse.ai.llm import Effort from alphapulse.api.response import success @@ -43,7 +43,7 @@ class SettingsIn(BaseModel): def config_out(value): - return {**{key: getattr(value, key) for key in SettingsIn.model_fields}, 'start_at': value.start_at} + return {**{key: getattr(value, key) for key in SettingsIn.model_fields}, 'start_at': value.start_at, 'max_concurrency': value.max_concurrency if value.max_concurrency is not None else 5} @router.get('/settings') @@ -53,14 +53,28 @@ async def settings(session: DB): @router.put('/settings') async def save_settings(body: SettingsIn, session: DB): - previous = await session.get(RiskSettings, 1) + previous = await session.get(RiskSettings, 1, with_for_update=True) if body.enabled: raise HTTPException(409, '请先使用测试并启用按钮验证三阶段模型和Skills') - value = RiskSettings(id=1, **body.model_dump(), start_at=previous.start_at if previous else None) + value = RiskSettings(id=1, **body.model_dump(), start_at=previous.start_at if previous else None, + max_concurrency=previous.max_concurrency if previous else 5) await session.merge(value); await session.commit() return success(None) +class ConcurrencyIn(BaseModel): + max_concurrency: int = Field(gt=0, strict=True) + + +@router.patch('/settings/concurrency') +async def save_concurrency(body: ConcurrencyIn, session: DB): + value = await default_settings(session) + value.max_concurrency = body.max_concurrency + session.add(value) + await session.commit() + return success({'max_concurrency': value.max_concurrency}) + + class TemplatesIn(BaseModel): refresh_existing: bool = False @@ -90,34 +104,56 @@ async def templates(session: DB, body: Annotated[TemplatesIn | None, Body()] = N @router.post('/enable') async def enable(body: SettingsIn, session: DB): + from alphapulse.core.db import engine + from alphapulse.news_risk.scheduler import GATE + async with engine.connect() as gate: + if not await gate.scalar(text('SELECT pg_try_advisory_lock(:gate, 1)'), {'gate': GATE}): + raise HTTPException(409, '分析仍在运行或正在验证，请暂停并等待在途任务完成') + await gate.commit() + try: + previous = await session.get(RiskSettings, 1, populate_existing=True) + if previous and previous.enabled: + raise HTTPException(409, '请先保存并暂停分析，再测试并启用') + return await validate_enable(body, session) + finally: + await gate.execute(text('SELECT pg_advisory_unlock(:gate, 1)'), {'gate': GATE}) + await gate.commit() + + +async def validate_enable(body: SettingsIn, session: DB): from alphapulse.news_risk.pipeline import configuration, Grouping, summarize_articles, Evidence, validate_evidence from alphapulse.news_risk import briefing before = await session.get(RiskSettings, 1) - before_config = config_out(before) if before else None + before_config = {k: v for k, v in config_out(before).items() if k != 'max_concurrency'} if before else None value = RiskSettings(id=1, **{**body.model_dump(), 'enabled': True}) fingerprints = {} fixture = {'id': 1, 'title': '虚构示例：示例公司发布维护通知', 'summary': '示例公司计划例行维护。', 'content': '本条为虚构测试，不是真实市场事件。示例公司计划例行维护，恢复时间尚未公布。'} try: provider, system, _, fingerprints['grouping'] = await configuration(session, value, 'grouping') + await session.commit() grouped, _ = await complete(provider, system, {'article': fixture, 'candidates': []}, Grouping) for event in grouped.packages: if event.package_id is not None: raise ValueError('测试没有已有候选事件') validate_evidence([Evidence(article_id=1, quote=event.evidence)], {1: fixture}) provider, system, _, fingerprints['analysis'] = await configuration(session, value, 'analysis') + await session.commit() await summarize_articles(provider, system, [fixture], {'id': 1, 'title': fixture['title'], 'fact': '示例公司计划例行维护', 'category': 'company', 'tags': ['维护'], 'content_kind': 'statement'}) provider, system, _, fingerprints['outlook'] = await configuration(session, value, 'outlook') + await session.commit() await briefing.generate(provider, system, [{'id': 1, 'revision': 1, 'importance_stars': 3, 'result': {'summary': '虚构测试'}}]) except Exception as error: raise HTTPException(502, f'三阶段验证失败：{error}') from error previous = await session.get(RiskSettings, 1, populate_existing=True, with_for_update=True) - if (config_out(previous) if previous else None) != before_config: + current_config = {k: v for k, v in config_out(previous).items() if k != 'max_concurrency'} if previous else None + if current_config != before_config: raise HTTPException(409, '验证期间设置已变化，请重新操作') for stage, fingerprint in fingerprints.items(): if (await configuration(session, value, stage))[3] != fingerprint: raise HTTPException(409, '验证期间模型或Skill已变化，请重新验证') value.start_at = previous.start_at if previous and previous.start_at else datetime.now(UTC) + value.max_concurrency = previous.max_concurrency if previous else 5 await session.merge(value); await session.commit() return success(None) @@ -152,7 +188,17 @@ async def overview(session: DB, hours: RiskWindow = RiskWindow.DAY): 'heartbeat': await redis.get('ap:risk:outlook:heartbeat')} from alphapulse.models.risk import NewsEvent event_states = dict((await session.execute(select(NewsEvent.state, func.count()).group_by(NewsEvent.state))).all()) - return success({'event_states': event_states, 'stage2': stage2, 'stage3': stage3, 'as_of': now, 'counts': counts, 'coverage': await coverage(session), 'scenario': scenario, 'briefing': snapshot.result if snapshot else None, + import json + runtime = await redis.get('ap:risk:concurrency') + settings = await session.get(RiskSettings, 1) + concurrency = {'limit': settings.max_concurrency if settings else 5, 'running': None, 'stages': {}, 'updated_at': None} + if runtime: concurrency.update(json.loads(runtime)) + from alphapulse.models.risk import ReprocessJob + jobs = (await session.scalars(select(ReprocessJob).where(ReprocessJob.state.in_(['pending', 'running'])))).all() + historical_pending = sum(i['state'] == 'pending' for job in jobs for i in job.items) + coverage_data = await coverage(session) + concurrency['queued'] = historical_pending + sum(stage2.get(k, 0) for k in ('pending',)) + event_states.get('pending', 0) + coverage_data['pending'] + return success({'concurrency': concurrency, 'event_states': event_states, 'stage2': stage2, 'stage3': stage3, 'as_of': now, 'counts': counts, 'coverage': coverage_data, 'scenario': scenario, 'briefing': snapshot.result if snapshot else None, 'snapshot_at': snapshot.created_at if snapshot else None, 'quotes': snapshot.quotes if snapshot else None, 'stale': not snapshot or changed or bool(last and last.error), 'error': last.error if last else None, 'heartbeat': await redis.get('ap:risk:heartbeat')}) @@ -390,7 +436,7 @@ async def reprocess_retry(identifier: int, session: DB): row.state = 'pending'; row.due_at = datetime.now(UTC); row.error = None ids = {f['id'] for f in job.failures if f['stage'] == 'grouping'} if ids: - job.items = [i for i in job.items if i['id'] in ids] + job.items = [{**i, 'state': 'pending'} for i in job.items if i['id'] in ids] job.position = 0 else: job.position = len(job.items) diff --git a/backend/src/alphapulse/models/risk.py b/backend/src/alphapulse/models/risk.py index 5eccd15..0a39327 100644 --- a/backend/src/alphapulse/models/risk.py +++ b/backend/src/alphapulse/models/risk.py @@ -10,6 +10,7 @@ class RiskSettings(Base): __tablename__ = 'risk_settings' id: Mapped[int] = mapped_column(primary_key=True) enabled: Mapped[bool] = mapped_column(Boolean, default=False) + max_concurrency: Mapped[int] = mapped_column(Integer, default=5, server_default='5') provider_id: Mapped[int | None] = mapped_column(ForeignKey('ai_providers.id')) model: Mapped[str] = mapped_column(String(128), default='') second_provider_id: Mapped[int | None] = mapped_column(ForeignKey('ai_providers.id')) diff --git a/backend/src/alphapulse/news_risk/events.py b/backend/src/alphapulse/news_risk/events.py index 4f7b743..8d862d7 100644 --- a/backend/src/alphapulse/news_risk/events.py +++ b/backend/src/alphapulse/news_risk/events.py @@ -91,11 +91,11 @@ async def associate(session, package, result, researched_ids, researched_events= event.updated_at = max(p.updated_at for p in members) -async def organize(): +async def organize(target_id=None): async with SessionLocal() as session: settings = await session.get(RiskSettings, 1) if not settings or not settings.enabled: return - event = await session.scalar(select(NewsEvent).where(NewsEvent.state == 'pending', NewsEvent.due_at <= datetime.now(UTC)).order_by(NewsEvent.due_at).limit(1)) + event = await session.scalar(select(NewsEvent).where(NewsEvent.state == 'pending', NewsEvent.due_at <= datetime.now(UTC), NewsEvent.id == target_id if target_id else True).order_by(NewsEvent.due_at).limit(1)) if event is None: return members = (await session.scalars(select(NewsSummary).join(EventMember, EventMember.package_id == NewsSummary.id) .where(EventMember.event_id == event.id))).all() @@ -125,18 +125,21 @@ async def organize(): result, usage = await complete(provider, system, {'task': 'event_timeline', 'previous': event.result, 'packages': packages}, EventResult) validate_timeline(result, packages) version.result = result.model_dump(mode='json'); version.usage = usage; version.state = 'done' - current_members = (await session.scalars(select(NewsSummary).join(EventMember, EventMember.package_id == NewsSummary.id) - .where(EventMember.event_id == event_id).order_by(NewsSummary.id).with_for_update(of=NewsSummary).execution_options(populate_existing=True))).all() - current = await session.get(NewsEvent, event_id, populate_existing=True, with_for_update=True) - current_members = [p for p in current_members if not p.replacement_ids and p.state != 'withdrawn'] - unchanged = {(p.id, p.published_revision) for p in current_members} == {(p['id'], p['revision']) for p in packages} - unchanged = unchanged and all(p.state == 'done' and not p.replacement_ids for p in current_members) - if current.revision == version.revision and unchanged: - current.published_revision = version.revision - current.result = version.result; current.title = result.title; current.state = 'done'; current.error = None - else: - version.state = 'superseded' - current.state = 'pending' + from alphapulse.news_risk.worker import publication_lock + async with publication_lock: + current_members = (await session.scalars(select(NewsSummary).join(EventMember, EventMember.package_id == NewsSummary.id) + .where(EventMember.event_id == event_id).order_by(NewsSummary.id).with_for_update(of=NewsSummary).execution_options(populate_existing=True))).all() + current = await session.get(NewsEvent, event_id, populate_existing=True, with_for_update=True) + current_members = [p for p in current_members if not p.replacement_ids and p.state != 'withdrawn'] + unchanged = {(p.id, p.published_revision) for p in current_members} == {(p['id'], p['revision']) for p in packages} + unchanged = unchanged and all(p.state == 'done' and not p.replacement_ids for p in current_members) + if current.revision == version.revision and unchanged: + current.published_revision = version.revision + current.result = version.result; current.title = result.title; current.state = 'done'; current.error = None + else: + version.state = 'superseded' + current.state = 'pending' + await session.commit() except Exception as error: await session.rollback() version = await session.get(EventVersion, version_id) diff --git a/backend/src/alphapulse/news_risk/reprocess.py b/backend/src/alphapulse/news_risk/reprocess.py index fba4787..4ecea13 100644 --- a/backend/src/alphapulse/news_risk/reprocess.py +++ b/backend/src/alphapulse/news_risk/reprocess.py @@ -21,6 +21,7 @@ async def capture(session, settings): from alphapulse.api.risk import config_out value = config_out(settings) value.pop('start_at') + value.pop('max_concurrency', None) configurations = {} for stage in ('grouping', 'analysis', 'outlook'): provider, system, rules, fingerprint = await configuration(session, settings, stage) @@ -39,16 +40,17 @@ async def frozen_configuration(session, frozen, stage): return provider, item['system'], item['rules'] -async def process_job(): +async def process_job(item_index=None): from alphapulse.news_risk import worker async with SessionLocal() as session: settings = await session.get(RiskSettings, 1) if not settings or not settings.enabled: return - if await session.scalar(select(ArticleAnalysis.id).where(ArticleAnalysis.state == 'pending').limit(1)): return - if await session.scalar(select(NewsSummary.id).where(NewsSummary.state.in_(['pending', 'processing'])).limit(1)): return + if item_index is None and await session.scalar(select(ArticleAnalysis.id).where(ArticleAnalysis.state == 'pending').limit(1)): return + if item_index is None and await session.scalar(select(NewsSummary.id).where(NewsSummary.state.in_(['pending', 'processing'])).limit(1)): return job = await session.scalar(select(ReprocessJob).where(ReprocessJob.state.in_(['pending', 'running'])).order_by(ReprocessJob.id).limit(1).with_for_update()) if job is None: return if job.position >= len(job.items): + if await session.scalar(select(NewsSummary.id).where(NewsSummary.state.in_(['pending', 'processing'])).limit(1)): return from alphapulse.models.risk import NewsEvent if await session.scalar(select(NewsEvent.id).where(NewsEvent.state.in_(['pending','processing'])).limit(1)): return from alphapulse.models.risk import RiskSnapshot @@ -72,7 +74,11 @@ async def process_job(): failures.append({'stage': 'outlook', 'id': snapshot.id, 'error': (snapshot.error or '阅读编排失败')[:1000]}) job.failures = [*job.failures, *failures] job.state = 'failed' if job.failures else 'done'; await session.commit(); return - item = job.items[job.position] + index = job.position if item_index is None else item_index + if index < 0: return + item = job.items[index] + if item.get('state') not in (None, 'pending'): return + job.items = [{**entry, 'state': 'processing'} if n == index else entry for n, entry in enumerate(job.items)] job_id = job.id try: for stage in ('grouping', 'analysis', 'outlook'): @@ -89,16 +95,22 @@ async def process_job(): session.add(row); await session.flush() row_id = row.id; job_id = job.id; job.state = 'running'; await session.commit() outcome = await worker.analyze(row_id, job.settings) - if outcome == 'deferred': return + if outcome == 'deferred': + job = await session.get(ReprocessJob, job_id, populate_existing=True, with_for_update=True) + job.items = [{**entry, 'state': 'pending'} if n == index else entry for n, entry in enumerate(job.items)] + await session.commit(); return row = await session.get(ArticleAnalysis, row_id, populate_existing=True) if outcome != 'done' or row.state != 'done' or row.error: raise ValueError(row.error or '原文已变化') + terminal = 'done' except Exception as error: + terminal = 'failed' await session.rollback() - job = await session.get(ReprocessJob, job_id, populate_existing=True) + job = await session.get(ReprocessJob, job_id, populate_existing=True, with_for_update=True) job.failures = [*job.failures, {'stage': 'grouping', 'id': item['id'], 'error': str(error)[:1000]}] await session.flush() job = await session.get(ReprocessJob, job_id, populate_existing=True, with_for_update=True) - job.position += 1 + job.items = [{**entry, 'state': terminal} if n == index else entry for n, entry in enumerate(job.items)] + job.position = sum(entry.get('state') in ('done', 'failed') for entry in job.items) await session.commit() @@ -111,7 +123,7 @@ async def start_job(session, start, end): if settings is None: raise ValueError('请先配置新闻AI') counts, items = await preview(session, start, end) if not items: raise ValueError('所选范围没有新闻') - job = ReprocessJob(state='pending', settings=await capture(session, settings), items=items, position=0, failures=[]) + job = ReprocessJob(state='pending', settings=await capture(session, settings), items=[{**i, 'state': 'pending'} for i in items], position=0, failures=[]) session.add(job); await session.flush() job.settings = {**job.settings, 'job_id': job.id} await session.commit() diff --git a/backend/src/alphapulse/news_risk/scheduler.py b/backend/src/alphapulse/news_risk/scheduler.py new file mode 100644 index 0000000..ddbe41c --- /dev/null +++ b/backend/src/alphapulse/news_risk/scheduler.py @@ -0,0 +1,152 @@ +"""Bounded, completion-driven dispatch for every background news AI stage.""" +import asyncio +import json +import logging +import time +from datetime import UTC, datetime +from sqlalchemy import select, text, func +from alphapulse.core.db import SessionLocal +from alphapulse.core.redis import redis +from alphapulse.models.risk import RiskSettings, NewsSummary, NewsEvent, ArticleAnalysis, ReprocessJob + +logger = logging.getLogger(__name__) +GATE = 20261102 + + +class Pool: + def __init__(self): + self.tasks = {} + + def start(self, key, action): + self.tasks[asyncio.create_task(action())] = key + + async def reap(self, timeout=1): + if not self.tasks: + await asyncio.sleep(timeout) + return + done, _ = await asyncio.wait(self.tasks, timeout=timeout, return_when=asyncio.FIRST_COMPLETED) + for task in done: + del self.tasks[task] + try: + task.result() + except asyncio.CancelledError: + pass + except Exception: + logger.exception('新闻AI任务异常') + + async def close(self): + for task in self.tasks: + task.cancel() + await asyncio.gather(*self.tasks, return_exceptions=True) + + +async def candidate(active, cursor, allow_outlook=True): + from alphapulse.news_risk import worker, events, reprocess + from alphapulse.models import NewsItem + async with SessionLocal() as session: + settings = await session.get(RiskSettings, 1) + if not settings or not settings.enabled or not settings.start_at: + return None + await worker.enqueue(session, settings) + history_news = [] + for kind, identity in active: + if kind == 'history' and identity[1] >= 0: + job = await session.get(ReprocessJob, identity[0]) + history_news.append(job.items[identity[1]]['id']) + stages = ('grouping', 'analysis', 'events', 'outlook') + for historical in (False, True): + for offset in range(len(stages)): + stage = stages[(cursor + offset) % len(stages)] + if historical and stage in ('grouping', 'outlook'): continue + busy = [key[1] for key in active if key[0] == stage] + if stage == 'outlook': + if not busy and allow_outlook: + return (stage, 0), worker.outlook, (cursor + offset + 1) % 4 + continue + model = {'grouping': ArticleAnalysis, 'analysis': NewsSummary, 'events': NewsEvent}[stage] + query = select(model.id).where(model.state == 'pending', model.id.not_in(busy)) + if stage == 'grouping': + query = query.join(NewsItem).where(model.content_hash == NewsItem.content_hash, NewsItem.id.not_in(history_news)).order_by(NewsItem.published_at.desc()) + else: + realtime = model.review_config.is_(None) | (func.jsonb_typeof(model.review_config) == 'null') + query = query.where(~realtime if historical else realtime, model.due_at <= datetime.now(UTC)).order_by(model.due_at, model.id) + if stage == 'events': + from alphapulse.models.risk import EventMember + blocked = select(EventMember.event_id).join(NewsSummary, NewsSummary.id == EventMember.package_id).where( + NewsSummary.replacement_ids == [], NewsSummary.state != 'withdrawn', + (NewsSummary.state != 'done') | NewsSummary.result.is_(None)) + query = query.where(model.id.not_in(blocked)) + identifier = await session.scalar(query.limit(1)) + if identifier is not None: + action = {'grouping': worker.analyze, 'analysis': worker.summarize, 'events': events.organize}[stage] + return (stage, identifier), lambda action=action, identifier=identifier: action(identifier), (cursor + offset + 1) % 4 + job = await session.scalar(select(ReprocessJob).where(ReprocessJob.state.in_(['pending', 'running'])).order_by(ReprocessJob.id).limit(1)) + if job: + for index, item in enumerate(job.items): + key = ('history', (job.id, index)) + if item.get('state', 'pending') == 'pending' and key not in active: + # Real-time and historical grouping share the article identity. + row_id = await session.scalar(select(ArticleAnalysis.id).where(ArticleAnalysis.news_id == item['id'], ArticleAnalysis.content_hash == item['hash'])) + if row_id is not None and ('grouping', row_id) in active: + continue + return key, lambda index=index: reprocess.process_job(index), cursor + if not any(k[0] == 'history' for k in active): + return ('history', (job.id, -1)), lambda: reprocess.process_job(-1), cursor + return None + + +async def run(lock): + pool = Pool() + cursor = 0 + gate = False + next_outlook = 0.0 + # Interrupted historical items are reclaimed independently of completion order. + async with SessionLocal() as session: + jobs = (await session.scalars(select(ReprocessJob).where(ReprocessJob.state.in_(['pending', 'running'])))).all() + for job in jobs: + for item in job.items: + if item['state'] == 'processing': + from sqlalchemy import update + await session.execute(update(ArticleAnalysis).where(ArticleAnalysis.news_id == item['id'], + ArticleAnalysis.content_hash == item['hash'], ArticleAnalysis.state == 'pending').values(state='reprocess')) + job.items = [{**i, 'state': 'pending' if i.get('state') == 'processing' else i['state']} for i in job.items] + await session.commit() + try: + while True: + if not pool.tasks and gate: + await lock.execute(text('SELECT pg_advisory_unlock_shared(:gate, 1)'), {'gate': GATE}) + await lock.commit(); gate = False + async with SessionLocal() as session: + settings = await session.get(RiskSettings, 1) + limit = settings.max_concurrency if settings else 5 + enabled = bool(settings and settings.enabled) + while enabled and len(pool.tasks) < limit: + if not gate: + gate = bool(await lock.scalar(text('SELECT pg_try_advisory_lock_shared(:gate, 1)'), {'gate': GATE})) + await lock.commit() + if not gate: + break + picked = await candidate(set(pool.tasks.values()), cursor, time.monotonic() >= next_outlook) + if picked is None: + break + key, action, cursor = picked + pool.start(key, action) + if key[0] == 'outlook': next_outlook = time.monotonic() + 1 + # Refresh operational settings for each admission, including pause. + async with SessionLocal() as session: + settings = await session.get(RiskSettings, 1) + limit, enabled = settings.max_concurrency, settings.enabled + now = datetime.now(UTC).isoformat() + counts = {stage: sum(k[0] == stage for k in pool.tasks.values()) for stage in ('grouping', 'analysis', 'events', 'outlook', 'history')} + try: + await redis.set('ap:risk:concurrency', json.dumps({'running': len(pool.tasks), 'stages': counts, 'updated_at': now}), ex=15) + for stage in ('', 'analysis:', 'events:', 'outlook:', 'reprocess:'): + await redis.set('ap:risk:' + stage + 'heartbeat', now, ex=15) + except Exception: + logger.exception('新闻AI运行状态暂时无法更新') + await pool.reap() + finally: + await pool.close() + if gate: + await lock.execute(text('SELECT pg_advisory_unlock_shared(:gate, 1)'), {'gate': GATE}) + await lock.commit() diff --git a/backend/src/alphapulse/news_risk/worker.py b/backend/src/alphapulse/news_risk/worker.py index 2e9734d..ea38b6b 100644 --- a/backend/src/alphapulse/news_risk/worker.py +++ b/backend/src/alphapulse/news_risk/worker.py @@ -1,11 +1,10 @@ -"""Independent, durable grouping and event analysis loops.""" +"""Durable news AI stages dispatched through the shared concurrency pool.""" import asyncio import logging from datetime import UTC, datetime, timedelta from sqlalchemy import select, update, func, exists, text, case from sqlalchemy.dialects.postgresql import insert from alphapulse.core.db import SessionLocal, engine -from alphapulse.core.redis import redis from alphapulse.models import NewsItem from alphapulse.models.risk import RiskSettings, RiskSnapshot, ArticleAnalysis, NewsSummary, SummaryArticle, SummaryVersion, PackageEvidence, GroupingVersion from alphapulse.news_risk.service import complete @@ -14,8 +13,8 @@ from alphapulse.news_risk.research import Research from alphapulse.news_risk import briefing logger = logging.getLogger(__name__) -analysis_lock = asyncio.Lock() -grouping_lock = asyncio.Lock() +publication_lock = asyncio.Lock() + def eligible(start): @@ -44,8 +43,7 @@ async def enqueue(session, settings): async def analyze(target_id=None, frozen=None): - async with grouping_lock: - return await _analyze(target_id, frozen) + return await _analyze(target_id, frozen) async def _analyze(target_id=None, frozen=None): @@ -64,6 +62,7 @@ async def _analyze(target_id=None, frozen=None): provider, system, rules = await frozen_configuration(session, frozen, 'grouping') else: provider, system, rules, _ = await configuration(session, settings, 'grouping') + catalog = await session.scalar(select(func.max(NewsSummary.id))) tag_score = sum((case((NewsSummary.tags.contains([tag.strip().casefold()]), 1), else_=0) for tag in row.input.get('source_tags', [])[:8] if isinstance(tag, str)), 0) candidates = (await session.scalars(select(NewsSummary).where(NewsSummary.replacement_ids == [], NewsSummary.state != 'withdrawn').order_by( (func.greatest(func.similarity(NewsSummary.title, row.input['title']), func.similarity(NewsSummary.fact, row.input['title'])) + tag_score).desc(), NewsSummary.updated_at.desc()).limit(80))).all() @@ -88,53 +87,56 @@ async def _analyze(target_id=None, frozen=None): if attempt: raise payload = pack_candidates(provider, system, {**payload, 'correction': str(error)}, payload['candidates']) - # Check the article again before changing event membership. - current = await session.get(NewsItem, row.news_id, populate_existing=True) - if current.content_hash != row.content_hash: - row.state = 'superseded'; await session.commit(); return 'superseded' - now = datetime.now(UTC) - old_ids = select(SummaryArticle.summary_id).join(ArticleAnalysis).where(ArticleAnalysis.news_id == row.news_id) - old_events = (await session.scalars(select(NewsSummary).where(NewsSummary.id.in_(old_ids)) - .with_for_update().execution_options(populate_existing=True))).all() - touched = {e.id: e for e in old_events} - previous_ids = set(touched) - await session.execute(update(SummaryArticle).where(SummaryArticle.analysis_id == row.id).values(active=False)) - groups = {} - changed = set() - for match in result.packages: - event = await session.get(NewsSummary, match.package_id, with_for_update=True, populate_existing=True) if match.package_id else groups.get(match.group_key) - if event is not None and (event.replacement_ids or event.state == 'withdrawn'): - raise ValueError('候选包已被纠错或撤回，请重试初筛') - progress = datetime.fromisoformat(row.input['progress_at']) - published = datetime.fromisoformat(row.input['published_at']) - if event is None: - event = NewsSummary(title=match.title, category=match.category, entities=match.entities, - published_at=published, updated_at=progress, changed_at=now, due_at=now, - revision=0, published_revision=0, state='pending') - session.add(event); await session.flush(); groups[match.group_key] = event - event.tags = sorted(set(event.tags or []) | {t.strip().casefold() for t in match.tags if t.strip()}) - if match.change_kind != 'repost' or not event.fact: - event.content_kind = match.content_kind; event.fact = match.fact - event.published_at = min(event.published_at, published) - if match.change_kind != 'repost' or row.input['revision_no'] > 1: - event.updated_at = max(event.updated_at, progress) - link = await session.scalar(select(SummaryArticle).where(SummaryArticle.summary_id == event.id, SummaryArticle.analysis_id == row.id)) - if link is None: - link = SummaryArticle(summary_id=event.id, analysis_id=row.id); session.add(link) - link.active = True; link.unit = match.model_dump() - if match.change_kind != 'repost' or event.result is None or row.input['revision_no'] > 1: - changed.add(event.id) - touched[event.id] = event - if target_id or row.input['revision_no'] > 1: - changed.update(previous_ids) - for event in touched.values(): - if event.id in changed: - event.review_config = frozen - schedule(event, now, any(m.urgent for m in result.packages)) - session.add(GroupingVersion(analysis_id=row.id, input=row.input, result=result.model_dump(mode='json'), rules=rules, state='done')) - row.result = result.model_dump(mode='json'); row.usage = {'attempts': usages}; row.state = 'done' - row.error = None; row.finished_at = now - await session.commit() + async with publication_lock: + # Check the article again before changing event membership. + current = await session.get(NewsItem, row.news_id, populate_existing=True) + if current.content_hash != row.content_hash: + row.state = 'superseded'; await session.commit(); return 'superseded' + if any(m.package_id is None for m in result.packages) and catalog != await session.scalar(select(func.max(NewsSummary.id))): + row.state = 'reprocess' if frozen else 'pending'; await session.commit(); return 'deferred' + now = datetime.now(UTC) + old_ids = select(SummaryArticle.summary_id).join(ArticleAnalysis).where(ArticleAnalysis.news_id == row.news_id) + old_events = (await session.scalars(select(NewsSummary).where(NewsSummary.id.in_(old_ids)) + .order_by(NewsSummary.id).with_for_update().execution_options(populate_existing=True))).all() + touched = {e.id: e for e in old_events} + previous_ids = set(touched) + await session.execute(update(SummaryArticle).where(SummaryArticle.analysis_id == row.id).values(active=False)) + groups = {} + changed = set() + for match in result.packages: + event = await session.get(NewsSummary, match.package_id, with_for_update=True, populate_existing=True) if match.package_id else groups.get(match.group_key) + if event is not None and (event.replacement_ids or event.state == 'withdrawn'): + raise ValueError('候选包已被纠错或撤回，请重试初筛') + progress = datetime.fromisoformat(row.input['progress_at']) + published = datetime.fromisoformat(row.input['published_at']) + if event is None: + event = NewsSummary(title=match.title, category=match.category, entities=match.entities, + published_at=published, updated_at=progress, changed_at=now, due_at=now, + revision=0, published_revision=0, state='pending') + session.add(event); await session.flush(); groups[match.group_key] = event + event.tags = sorted(set(event.tags or []) | {t.strip().casefold() for t in match.tags if t.strip()}) + if match.change_kind != 'repost' or not event.fact: + event.content_kind = match.content_kind; event.fact = match.fact + event.published_at = min(event.published_at, published) + if match.change_kind != 'repost' or row.input['revision_no'] > 1: + event.updated_at = max(event.updated_at, progress) + link = await session.scalar(select(SummaryArticle).where(SummaryArticle.summary_id == event.id, SummaryArticle.analysis_id == row.id)) + if link is None: + link = SummaryArticle(summary_id=event.id, analysis_id=row.id); session.add(link) + link.active = True; link.unit = match.model_dump() + if match.change_kind != 'repost' or event.result is None or row.input['revision_no'] > 1: + changed.add(event.id) + touched[event.id] = event + if target_id or row.input['revision_no'] > 1: + changed.update(previous_ids) + for event in touched.values(): + if event.id in changed: + event.review_config = frozen + schedule(event, now, any(m.urgent for m in result.packages)) + session.add(GroupingVersion(analysis_id=row.id, input=row.input, result=result.model_dump(mode='json'), rules=rules, state='done')) + row.result = result.model_dump(mode='json'); row.usage = {'attempts': usages}; row.state = 'done' + row.error = None; row.finished_at = now + await session.commit() return 'done' except Exception as error: await session.rollback() @@ -146,13 +148,14 @@ async def _analyze(target_id=None, frozen=None): return 'failed' -async def summarize(): - async with analysis_lock, SessionLocal() as session: +async def summarize(target_id=None): + async with SessionLocal() as session: settings = await session.get(RiskSettings, 1) if not settings or not settings.enabled: return event = await session.scalar(select(NewsSummary).where(NewsSummary.state == 'pending', - NewsSummary.due_at <= datetime.now(UTC)).order_by(NewsSummary.due_at, NewsSummary.id).limit(1).with_for_update()) + NewsSummary.due_at <= datetime.now(UTC), + NewsSummary.id == target_id if target_id else True).order_by(NewsSummary.due_at, NewsSummary.id).limit(1).with_for_update()) if event is None: return config_error = None @@ -194,45 +197,51 @@ async def summarize(): research = Research() evidence_rows = (await session.scalars(select(PackageEvidence).where(PackageEvidence.version_id == identifier).order_by(PackageEvidence.id))).all() research.restore([r.data for r in evidence_rows]) + await session.commit() try: if config_error: raise ValueError(config_error) result, usage = await summarize_articles(provider, system, articles, package, research) - event = await session.get(NewsSummary, event_id, populate_existing=True, with_for_update=True) - if result.correction is not None: - from alphapulse.news_risk.corrections import apply_correction - from alphapulse.news_risk.contracts import Correction - await apply_correction(session, Correction.model_validate(result.correction), event_id, research.read_package_ids, rules) - version.result = result.model_dump(mode='json'); version.usage = usage; version.state = 'corrected'; version.finished_at = datetime.now(UTC) + async with publication_lock: + event = await session.get(NewsSummary, event_id, populate_existing=True, with_for_update=True) + if event.replacement_ids or event.state == 'withdrawn': + version.state = 'superseded'; await session.commit(); return + if result.correction is not None and event.revision != revision: + version.state = 'superseded'; await session.commit(); return + if result.correction is not None: + from alphapulse.news_risk.corrections import apply_correction + from alphapulse.news_risk.contracts import Correction + await apply_correction(session, Correction.model_validate(result.correction), event_id, research.read_package_ids, rules) + version.result = result.model_dump(mode='json'); version.usage = usage; version.state = 'corrected'; version.finished_at = datetime.now(UTC) + for record in research.records: + session.add(PackageEvidence(summary_id=event_id, version_id=identifier, kind=record['tool'], reference=str(record['arguments']), data=record)) + await session.commit(); return + version.result = result.model_dump(mode='json'); version.usage = usage; version.state = 'done'; version.finished_at = datetime.now(UTC) + if event.revision == revision and revision > event.published_revision: + event.result = result.model_dump(mode='json') + event.stars = result.risk_stars; event.peak_stars = max(event.peak_stars, result.risk_stars) + event.importance_stars = result.importance_stars + event.published_revision = revision; event.analyzed_at = version.finished_at; event.error = None + event.state = 'done' if event.revision == revision else 'pending' + if event.state == 'done': event.dirty_since = None for record in research.records: session.add(PackageEvidence(summary_id=event_id, version_id=identifier, kind=record['tool'], reference=str(record['arguments']), data=record)) - await session.commit(); return - version.result = result.model_dump(mode='json'); version.usage = usage; version.state = 'done'; version.finished_at = datetime.now(UTC) - if event.revision == revision and revision > event.published_revision: - event.result = result.model_dump(mode='json') - event.stars = result.risk_stars; event.peak_stars = max(event.peak_stars, result.risk_stars) - event.importance_stars = result.importance_stars - event.published_revision = revision; event.analyzed_at = version.finished_at; event.error = None - event.state = 'done' if event.revision == revision else 'pending' - if event.state == 'done': event.dirty_since = None - for record in research.records: - session.add(PackageEvidence(summary_id=event_id, version_id=identifier, kind=record['tool'], reference=str(record['arguments']), data=record)) - from alphapulse.news_risk.events import associate - if event.revision == revision: - await associate(session, event, result, research.read_package_ids, research.read_event_ids) - await session.commit() + from alphapulse.news_risk.events import associate + if event.revision == revision: + await associate(session, event, result, research.read_package_ids, research.read_event_ids) + await session.commit() except Exception as error: await session.rollback() for record in research.records: session.add(PackageEvidence(summary_id=event_id, version_id=identifier, kind=record['tool'], reference=str(record['arguments']), data=record)) await session.execute(update(SummaryVersion).where(SummaryVersion.id == identifier).values( state='failed', error=str(error)[:2000], finished_at=datetime.now(UTC))) - await session.execute(update(NewsSummary).where(NewsSummary.id == event_id).values( + await session.execute(update(NewsSummary).where(NewsSummary.id == event_id, NewsSummary.replacement_ids == [], NewsSummary.state != 'withdrawn').values( state=case((NewsSummary.revision > revision, 'pending'), else_='failed'), error=str(error)[:2000])) await session.commit(); logger.exception('事件总结失败') async def outlook(): - async with analysis_lock, SessionLocal() as session: + async with SessionLocal() as session: settings = await session.get(RiskSettings, 1) if not settings or not settings.enabled: return now = datetime.now(UTC) @@ -293,16 +302,6 @@ async def outlook(): await session.commit() -async def loop(action, heartbeat): - while True: - try: - await redis.set(heartbeat, datetime.now(UTC).isoformat(), ex=1800) - await action() - except Exception: - logger.exception('新闻AI处理异常') - await asyncio.sleep(10) - - async def main(): async with engine.connect() as lock: if not await lock.scalar(text('SELECT pg_try_advisory_lock(20261030, 1)')): @@ -319,10 +318,8 @@ async def main(): await session.execute(update(NewsEvent).where(NewsEvent.state == 'processing').values(state='pending')) await session.execute(update(EventVersion).where(EventVersion.state == 'processing').values(state='failed', error='进程中断')) await session.commit() - from alphapulse.news_risk.events import organize - from alphapulse.news_risk.reprocess import process_job - await asyncio.gather(loop(organize, 'ap:risk:events:heartbeat'), loop(process_job, 'ap:risk:reprocess:heartbeat'), loop(analyze, 'ap:risk:heartbeat'), loop(summarize, 'ap:risk:analysis:heartbeat'), - loop(outlook, 'ap:risk:outlook:heartbeat')) + from alphapulse.news_risk.scheduler import run + await run(lock) if __name__ == '__main__': diff --git a/backend/tests/test_news_concurrency.py b/backend/tests/test_news_concurrency.py new file mode 100644 index 0000000..7f41108 --- /dev/null +++ b/backend/tests/test_news_concurrency.py @@ -0,0 +1,130 @@ +"""Concurrency behavior uses controlled tasks and isolated I/O boundaries.""" +import asyncio +from types import SimpleNamespace as N +from unittest.mock import AsyncMock, MagicMock +import pytest + + +def fake_session(settings): + session = MagicMock() + session.__aenter__ = AsyncMock(return_value=session) + session.__aexit__ = AsyncMock() + session.get = AsyncMock(return_value=settings) + session.scalars = AsyncMock(return_value=N(all=lambda: [])) + session.commit = AsyncMock() + return session + + +def test_dispatch_refills_without_waiting_for_slow_tasks_and_respects_live_limit(trading, monkeypatch): + from alphapulse.news_risk import scheduler + async def check(): + settings = N(enabled=True, max_concurrency=5) + session = fake_session(settings) + monkeypatch.setattr(scheduler, 'SessionLocal', lambda: session) + monkeypatch.setattr(scheduler.redis, 'set', AsyncMock()) + lock = N(scalar=AsyncMock(return_value=True), execute=AsyncMock(), commit=AsyncMock()) + releases = [asyncio.Event() for _ in range(10)] + started = [] + async def task(index): + started.append(index) + await releases[index].wait() + async def pick(active, cursor, allow): + assert len(active) < settings.max_concurrency + index = len(started) + # Reserve before the spawned coroutine starts. + index = max([key[1] for key in active] + [index-1]) + 1 + if index >= len(releases): return None + return (('grouping', 'analysis', 'events', 'outlook')[index % 4], index), lambda: task(index), cursor + monkeypatch.setattr(scheduler, 'candidate', pick) + run = asyncio.create_task(scheduler.run(lock)) + async def until(predicate): + async with asyncio.timeout(3): + while not predicate(): await asyncio.sleep(.001) + try: + await until(lambda: len(started) == 5) + releases[0].set(); releases[1].set() + await until(lambda: len(started) == 7) + assert all(not releases[i].is_set() for i in (2,3,4)) + settings.max_concurrency = 2 + for i in (2,3,4,5): releases[i].set() + await until(lambda: len(started) == 8) + assert not releases[6].is_set() + assert len(started) == 8 + settings.enabled = False + releases[6].set(); releases[7].set() + await asyncio.sleep(.03) + assert len(started) == 8 + settings.enabled = True; settings.max_concurrency = 5 + await until(lambda: len(started) == 10) + finally: + run.cancel() + with pytest.raises(asyncio.CancelledError): await run + assert lock.execute.await_count >= 1 + asyncio.run(check()) + + +def test_failed_tasks_release_capacity(trading): + from alphapulse.news_risk.scheduler import Pool + async def check(): + pool = Pool() + async def fail(): raise RuntimeError('Synthetic failure') + pool.start(('analysis', 1), fail) + await pool.reap() + assert not pool.tasks + await pool.close() + asyncio.run(check()) + + +@pytest.mark.parametrize('value', [0, -1, 1.5, '5', True, None]) +def test_concurrency_requires_a_positive_integer(trading, value): + from pydantic import ValidationError + from alphapulse.api.risk import ConcurrencyIn + with pytest.raises(ValidationError): ConcurrencyIn(max_concurrency=value) + + +def test_operational_update_does_not_pause_or_change_model_settings(trading, monkeypatch): + from alphapulse.api import risk + settings = N(enabled=True, max_concurrency=5, model='original') + monkeypatch.setattr(risk, 'default_settings', AsyncMock(return_value=settings)) + asyncio.run(risk.save_concurrency(risk.ConcurrencyIn(max_concurrency=3), N(commit=AsyncMock(), add=MagicMock()))) + assert settings.enabled and settings.model == 'original' and settings.max_concurrency == 3 + + +@pytest.mark.parametrize('busy,enabled', [(True,False), (False,True)]) +def test_validation_never_bypasses_background_tasks(trading, monkeypatch, busy, enabled): + from fastapi import HTTPException + from alphapulse.api import risk + from alphapulse.core import db + gate = MagicMock() + gate.__aenter__ = AsyncMock(return_value=gate); gate.__aexit__ = AsyncMock(return_value=None) + gate.scalar = AsyncMock(return_value=not busy) + gate.execute = AsyncMock(); gate.commit = AsyncMock() + monkeypatch.setattr(db, 'engine', N(connect=lambda: gate)) + validate = AsyncMock() + monkeypatch.setattr(risk, 'validate_enable', validate) + session = N(get=AsyncMock(return_value=N(enabled=enabled))) + with pytest.raises(HTTPException) as error: + asyncio.run(risk.enable(risk.SettingsIn(),session)) + assert error.value.status_code == 409 + validate.assert_not_awaited() + + +def test_history_completion_is_counted_independently_of_item_order(trading, monkeypatch): + from alphapulse.news_risk import reprocess, worker + from alphapulse.models.risk import RiskSettings, ArticleAnalysis, ReprocessJob + from alphapulse.models import NewsItem + job = N(id=4, state='running', position=0, settings={}, failures=[], + items=[{'id':i,'hash':'current','state':'pending'} for i in (1,2,3)]) + article = N(id=30, state='done', error=None) + session = fake_session(None) + session.scalar = AsyncMock(side_effect=[job, article]) + session.get = AsyncMock(side_effect=lambda model,*a,**kw: { + RiskSettings:N(enabled=True), ReprocessJob:job, + ArticleAnalysis:article, NewsItem:N(id=3, content_hash='current')}[model]) + monkeypatch.setattr(reprocess, 'SessionLocal', lambda:session) + monkeypatch.setattr(reprocess, 'frozen_configuration', AsyncMock()) + monkeypatch.setattr(worker, 'analyze', AsyncMock(return_value='done')) + asyncio.run(reprocess.process_job(2)) + assert job.position == 1 + assert [i['state'] for i in job.items] == ['pending','pending','done'] + assert job.failures == [] diff --git a/backend/tests/test_news_reprocess.py b/backend/tests/test_news_reprocess.py index 8b8b1a8..a3198bd 100644 --- a/backend/tests/test_news_reprocess.py +++ b/backend/tests/test_news_reprocess.py @@ -45,7 +45,7 @@ def test_downstream_failure_does_not_mark_unrelated_news_failed(trading, monkeyp failed = N(id=31,error='Synthetic failure',state='failed',quotes={}) session = session_mock() session.get.return_value = N(enabled=True) - session.scalar.side_effect = [None,None,job,None,failed] + session.scalar.side_effect = [None,None,job,None,None,failed] session.scalars.side_effect = [N(all=lambda:[failed] if stage == 'analysis' else []), N(all=lambda:[failed] if stage == 'events' else [])] monkeypatch.setattr(reprocess,'SessionLocal',lambda:session) @@ -70,7 +70,7 @@ def test_retry_resumes_only_failed_stage(trading, stage): asyncio.run(reprocess_retry(job.id,session)) assert job.state == 'running' and job.failures == [] if stage == 'grouping': - assert job.items == [{'id':11}] and job.position == 0 + assert job.items == [{'id':11, 'state':'pending'}] and job.position == 0 else: assert job.items == [{'id':11},{'id':12}] and job.position == 2 if stage == 'outlook': assert failed.quotes['retry_requested'] is True diff --git a/backend/tests/test_news_risk.py b/backend/tests/test_news_risk.py index e910b9b..d966951 100644 --- a/backend/tests/test_news_risk.py +++ b/backend/tests/test_news_risk.py @@ -48,7 +48,15 @@ def test_evidence_retains_quotes_present_in_source(risk): def mock_enable(risk, monkeypatch, valid=True, verification='reported', direction='neutral'): from types import SimpleNamespace - from unittest.mock import AsyncMock + from unittest.mock import AsyncMock, MagicMock + from alphapulse.core import db + gate = MagicMock() + gate.__aenter__ = AsyncMock(return_value=gate) + gate.__aexit__ = AsyncMock(return_value=None) + gate.scalar = AsyncMock(return_value=True) + gate.execute = AsyncMock() + gate.commit = AsyncMock() + monkeypatch.setattr(db, "engine", SimpleNamespace(connect=lambda: gate)) from alphapulse.news_risk import pipeline, briefing from alphapulse.api import risk as api provider = SimpleNamespace(model='news-only', max_tokens=1000, context_window=200000) @@ -101,7 +109,7 @@ def test_enable_validates_evidence_before_saving_settings(risk, monkeypatch, val if valid: saved = session.merge.call_args.args[0] assert saved.enabled and saved.start_at is not None - session.commit.assert_awaited_once() + assert session.commit.await_count == 4 else: assert '原文引用' in response.text session.merge.assert_not_awaited() @@ -233,13 +241,13 @@ def test_window_query_parameters_over_http(risk, monkeypatch, endpoint, hours, s import httpx from alphapulse.api import risk as api - session = SimpleNamespace(execute=AsyncMock(return_value=SimpleNamespace(all=lambda: [])), + session = SimpleNamespace(get=AsyncMock(return_value=None), execute=AsyncMock(return_value=SimpleNamespace(all=lambda: [])), scalar=AsyncMock(return_value=None), scalars=AsyncMock(return_value=SimpleNamespace(all=lambda: []))) app = FastAPI() app.include_router(api.router) app.dependency_overrides[api.require_admin] = lambda: None app.dependency_overrides[api.get_session] = lambda: session - monkeypatch.setattr(api, 'coverage', AsyncMock(return_value={})) + monkeypatch.setattr(api, 'coverage', AsyncMock(return_value={'pending': 0})) monkeypatch.setattr(api.redis, 'get', AsyncMock(return_value=None)) async def check(): diff --git a/deploy/README.md b/deploy/README.md index fc3e940..cd7b800 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -101,3 +101,24 @@ Each complete daily response withdraws missing records only within its own day a Official and Jin10 observations retain separate identities and revisions. The read API merges only verified indicator IDs, definitions, units, reference periods, release stages and exact publication times. Official nonmissing fields win independently; Jin10 fills missing fields and market forecasts with explicit provenance. Unverified definitions remain distinct records. Original and revised previous values remain separate. Forecast observations captured after publication are displayed with that limitation and excluded from historical surprise judgments. Jin10's gold/USD direction flags are not BTC/ETH signals. Tentative times and all-day holidays never display an invented exact midnight time. CI acceptance runs `scripts/smoke_calendar.py` against the isolated database using public fixtures and synthetic account material. It never starts a browser collector or connects a real Jin10 account. Local unit tests likewise isolate all external I/O. The live database migration, credential import/reconnection, history collection and dev deployment are separate operator actions; ordinary local validation does not authorize them. + +### News AI concurrency + +All background news AI stages share `risk_settings.max_concurrency` (default 5). +A slot covers one complete stage task, including its sequential model and verification +calls. Completed tasks release their slots immediately; runnable real-time stages rotate +before historical items use spare capacity. Pending work remains in PostgreSQL. + +The home settings dialog saves this operational limit independently of model settings. +Increasing it admits more work; decreasing it drains existing tasks without cancelling +paid requests. Pausing stops new admissions. The overview reports active slots and stage +counts; an expired worker heartbeat shows an unknown active count. + +`PATCH /api/risk/settings/concurrency` accepts a positive integer `max_concurrency`. +The limit is excluded from frozen reprocessing configurations and model fingerprints. +“Test and enable” requires paused analysis and an idle pipeline; an exclusive PostgreSQL +advisory gate prevents concurrent validation and background work. Historical items persist +individual states, so completion order does not change progress or skip unfinished items. + +Migration `e5f701234567` adds the limit and converts legacy historical checkpoints to +item states. Deployment and execution of this migration require separate authorization. diff --git a/web/src/common/apis/risk/index.ts b/web/src/common/apis/risk/index.ts index 272a86c..5b5a8a7 100644 --- a/web/src/common/apis/risk/index.ts +++ b/web/src/common/apis/risk/index.ts @@ -19,3 +19,5 @@ export const startReprocessApi = (data: Risk.ReprocessRange) => request<{ id: nu export const listReprocessApi = () => request({ url: "/risk/reprocess", method: "get" }) export const cancelReprocessApi = (id: number) => request({ url: `/risk/reprocess/${id}/cancel`, method: "post" }) export const retryReprocessApi = (id: number) => request({ url: `/risk/reprocess/${id}/retry`, method: "post" }) + +export const saveRiskConcurrencyApi = (max_concurrency: number) => request<{ max_concurrency: number }>({ url: "/risk/settings/concurrency", method: "patch", data: { max_concurrency } }) diff --git a/web/src/common/apis/risk/type.ts b/web/src/common/apis/risk/type.ts index da6a2b3..c15d930 100644 --- a/web/src/common/apis/risk/type.ts +++ b/web/src/common/apis/risk/type.ts @@ -43,6 +43,7 @@ export interface RiskEvent { updated_at: string } export interface Settings { + max_concurrency: number enabled: boolean provider_id: number | null model: string @@ -79,6 +80,7 @@ export interface Briefing { inputs: { kind?: "package" | "event", id: number, revision: number, category: string, tags: string[], result: RiskResult & { title: string } }[] } export interface Overview { + concurrency: { limit: number, running: number | null, queued: number, stages: Record, updated_at: string | null } briefing: Briefing | null stage2: Record stage3?: { state: string, pending: boolean, heartbeat: string | null } diff --git a/web/src/pages/home/index.vue b/web/src/pages/home/index.vue index 6c8090d..64d6e6c 100644 --- a/web/src/pages/home/index.vue +++ b/web/src/pages/home/index.vue @@ -59,6 +59,7 @@ const showFailures = ref(false) const loading = ref(false) const failed = ref(false) const saving = ref(false) +const concurrencySaving = ref(false) let sequence = 0 let detailSequence = 0 let modelsSequence = 0 @@ -157,6 +158,19 @@ async function createTemplates() { newsSkills.value = skills.filter(s => s.scope === "news" && s.kind === "skill") ElMessage.success("默认新闻 Skills 已准备好，请前往 Skills 页面审阅并发布") } +async function saveConcurrency() { + if (!settings.value || concurrencySaving.value) return + const version = user.sessionVersion + concurrencySaving.value = true + try { + await api.saveRiskConcurrencyApi(settings.value.max_concurrency) + if (version !== user.sessionVersion) return + ElMessage.success("并发上限已保存；降低上限时等待在途任务完成") + await load() + } finally { + if (version === user.sessionVersion) concurrencySaving.value = false + } +} async function save(enable: boolean) { if (!settings.value) return const version = user.sessionVersion @@ -207,6 +221,7 @@ watch(taskStage, () => { if (showFailures.value) void openFailures() }) watch(() => user.sessionVersion, () => { + concurrencySaving.value = false thirdSequence++ secondModelsSequence++ secondModels.value = [] @@ -284,6 +299,10 @@ onBeforeUnmount(() => { 失败 {{ overview.coverage.failed }}

+

+ AI 并发：{{ overview.concurrency.running ?? "未知" }} / {{ overview.concurrency.limit }} · 排队 {{ overview.concurrency.queued }} + {{ ({ grouping: "分包", analysis: "研判", events: "事件", outlook: "阅读", history: "历史" } as Record)[stage] }} {{ count }} +

包研判：完成 {{ overview.stage2.done ?? 0 }} · 待更新 {{ overview.stage2.pending ?? 0 }} · 处理中 {{ overview.stage2.processing ?? 0 }} · 失败 {{ overview.stage2.failed ?? 0 }}

阅读编排：{{ overview.stage3.state === 'processing' ? '处理中' : overview.stage3.state === 'failed' ? '失败' : overview.stage3.pending ? '等待聚合更新' : '已更新' }} @@ -441,6 +460,13 @@ onBeforeUnmount(() => { 编辑与发布 Skills + + + + 保存并发上限 + + +

所有新闻阶段共享名额，完成即补位；实时优先，历史使用空闲名额。测试并启用前请先暂停并等待在途任务完成。

当前状态：{{ settings.enabled ? '已启用' : '已关闭' }}

diff --git a/web/tests/workspaces.test.ts b/web/tests/workspaces.test.ts index 82d1d36..9c12768 100644 --- a/web/tests/workspaces.test.ts +++ b/web/tests/workspaces.test.ts @@ -836,3 +836,49 @@ test("重整理预览的旧会话完成不能解除新操作的忙碌状态", as assert.equal(page.busy.value, false) assert.equal(page.preview.value.articles, 2) }) + +test("新闻并发独立保存保留模型草稿，重复点击和旧会话响应不影响新操作", async () => { + const user = vue.reactive({ sessionVersion: 1 }) + const pending: ReturnType[] = [] + const values: number[] = [] + let reloads = 0 + const load = moduleLoader({ + "@@/apis/skills": { listSkillTreeApi: async () => [] }, + "@/common/apis/settings": {}, + "@/pinia/stores/user": { useUserStore: () => user }, + "@/common/apis/risk": { + saveRiskConcurrencyApi: (value: number) => { + values.push(value) + const request = Promise.withResolvers() + pending.push(request) + return request.promise + }, + getRiskOverviewApi: async () => { + reloads++ + return { briefing: null } + }, + listRiskEventsApi: async () => ({ items: [], total: 0 }) + } + }, globals) + const page = load("@/pages/home/index.vue").default.setup({}, { expose() {} }) + page.settings.value = { max_concurrency: 5, model: "unsaved-model" } + const first = page.saveConcurrency() + await page.saveConcurrency() + assert.deepEqual(values, [5]) + pending[0].resolve({ max_concurrency: 5 }) + await first + assert.equal(page.settings.value.model, "unsaved-model") + assert.equal(reloads, 1) + const old = page.saveConcurrency() + user.sessionVersion++ + await vue.nextTick() + page.settings.value = { max_concurrency: 2, model: "new-session" } + const current = page.saveConcurrency() + pending[1].resolve({ max_concurrency: 5 }) + await old + assert.equal(page.concurrencySaving.value, true) + assert.equal(page.settings.value.model, "new-session") + pending[2].resolve({ max_concurrency: 2 }) + await current + assert.equal(page.concurrencySaving.value, false) +})