diff --git a/backend/src/alphapulse/news_risk/backfill.py b/backend/src/alphapulse/news_risk/backfill.py index b7affb2..0874e62 100644 --- a/backend/src/alphapulse/news_risk/backfill.py +++ b/backend/src/alphapulse/news_risk/backfill.py @@ -76,6 +76,7 @@ async def start(session, identifier, run_mode="shared", validated=False): if not validated: raise ValueError('仅历史批次必须先验证三阶段配置') settings = await session.get(RiskSettings, 1) settings.enabled = True + if settings.start_at is None: settings.start_at = datetime.now(UTC) job = ReprocessJob(state='pending', items=[], position=0, failures=[], settings=value['config']) session.add(job); await session.flush() job.settings = {**job.settings, 'job_id':job.id, 'request':{**value['request'], 'run_mode':run_mode}} @@ -166,6 +167,23 @@ async def reconcile(session, job): await session.delete(row); waiting=True continue if obj.state=='retired': continue + if row.kind == 'events' and job.settings.get('request', {}).get('run_mode') == 'history_only': + # Retry busy members next pass instead of blocking their publication. + dependencies = (await session.scalars(select(NewsSummary).join(EventMember).where( + EventMember.event_id == obj.id, NewsSummary.replacement_ids == [], NewsSummary.state != 'withdrawn', + or_(NewsSummary.state != 'done', NewsSummary.result.is_(None), func.jsonb_typeof(NewsSummary.result) == 'null', + NewsSummary.published_revision < NewsSummary.revision)).order_by(NewsSummary.id) + .with_for_update(of=NewsSummary, skip_locked=True))).all() + for package in dependencies: + # Only unfinished members of this event join the frozen batch. + package.review_config = job.settings + if package.state == 'done': + from alphapulse.news_risk.worker import schedule + schedule(package, datetime.now(UTC), True) + if package.state == 'pending': package.due_at = datetime.now(UTC) + await target(session, job.id, 'analysis', package) + if package.state == 'failed': failures.append({'stage':'analysis','id':package.id,'error':package.error}) + else: waiting = True if obj.state=='failed': failures.append({'stage':row.kind,'id':obj.id,'error':obj.error}) elif row.kind == 'events' and await session.scalar(select(NewsSummary.id).join(EventMember).where( EventMember.event_id == obj.id, NewsSummary.state == 'failed', NewsSummary.replacement_ids == []).limit(1)): diff --git a/backend/src/alphapulse/news_risk/events.py b/backend/src/alphapulse/news_risk/events.py index fbc6503..e66e74a 100644 --- a/backend/src/alphapulse/news_risk/events.py +++ b/backend/src/alphapulse/news_risk/events.py @@ -44,6 +44,8 @@ def validate_timeline(result, packages): raise ValueError('事件发生时间必须来自组成包已核验时间点') if point.planned_at is not None and not any( PackageTimePoint.model_validate(p).planned_at == point.planned_at and p.get('planned_precision') == point.planned_precision + and p.get('time_text') == point.time_text and point.time_text.strip() + and any(ref in p['evidence_refs'] and point.time_text in ref.get('quote', '') for ref in point.evidence_refs) for p in package['result'].get('timeline_points', [])): raise ValueError('事件计划时间必须来自组成包已核验计划') if any(ref not in allowed for ref in point.evidence_refs): diff --git a/backend/src/alphapulse/news_risk/worker.py b/backend/src/alphapulse/news_risk/worker.py index 2e5658d..204c5e8 100644 --- a/backend/src/alphapulse/news_risk/worker.py +++ b/backend/src/alphapulse/news_risk/worker.py @@ -265,14 +265,14 @@ async def outlook(prepare_only=False): if job and job.settings.get('request', {}).get('run_mode') != 'history_only' and job.settings.get('request'): job = None if job and job.settings.get('request', {}).get('run_mode') == 'history_only': from alphapulse.models.risk import ReprocessItem - if await session.scalar(select(ReprocessItem.id).where(ReprocessItem.job_id == job.id, ReprocessItem.state.in_(['pending','processing'])).limit(1)): return + if await session.scalar(select(ReprocessItem.id).where(ReprocessItem.job_id == job.id, ReprocessItem.state.in_(['pending','processing','failed'])).limit(1)): return if job and job.position < len(job.items): if requested and requested.quotes.get('job_id') == job.id: return job = None if job: from alphapulse.models.risk import NewsEvent - if await session.scalar(select(NewsSummary.id).where(NewsSummary.state.in_(['pending', 'processing']), NewsSummary.review_config['job_id'].astext.cast(BigInteger) == job.id).limit(1)): return - if await session.scalar(select(NewsEvent.id).where(NewsEvent.state.in_(['pending', 'processing']), NewsEvent.review_config['job_id'].astext.cast(BigInteger) == job.id).limit(1)): return + if await session.scalar(select(NewsSummary.id).where(NewsSummary.state.in_(['pending', 'processing', 'failed']), NewsSummary.review_config['job_id'].astext.cast(BigInteger) == job.id).limit(1)): return + if await session.scalar(select(NewsEvent.id).where(NewsEvent.state.in_(['pending', 'processing', 'failed']), NewsEvent.review_config['job_id'].astext.cast(BigInteger) == job.id).limit(1)): return from alphapulse.news_risk.reprocess import frozen_configuration provider, system, rules = await frozen_configuration(session, job.settings, 'outlook') else: diff --git a/backend/tests/test_news_tuning.py b/backend/tests/test_news_tuning.py index bc84a77..3a724f2 100644 --- a/backend/tests/test_news_tuning.py +++ b/backend/tests/test_news_tuning.py @@ -1,5 +1,6 @@ """Workflow tuning preserves evidence boundaries and limits expensive work.""" import asyncio +from datetime import UTC, datetime from types import SimpleNamespace as N from unittest.mock import AsyncMock, MagicMock import pytest @@ -134,3 +135,134 @@ def test_large_synthesis_compacts_package_notes_in_batches(trading,monkeypatch): result=asyncio.run(briefing.generate(N(),'rules',inputs,rankings=rankings)) assert calls==['synthesis_notes']*3+['synthesis'] assert len(result['reading'])==3 + + +@pytest.mark.parametrize('original_start', [None, datetime(2026,11,1,tzinfo=UTC)]) +def test_exclusive_start_persists_dispatch_start_and_preserves_existing_boundary(trading,monkeypatch,original_start): + from alphapulse.news_risk import backfill,scheduler + settings=N(enabled=False,start_at=original_start) + session=MagicMock() + session.get=AsyncMock(return_value=settings);session.flush=AsyncMock();session.commit=AsyncMock() + session.add=lambda job:setattr(job,'id',7) + value={'items':[{'id':1,'hash':'current','mode':'review'}],'config':{},'request':{'selection':'range'}} + monkeypatch.setattr(backfill,'prepare_start',AsyncMock(return_value=value)) + monkeypatch.setattr(backfill.redis,'delete',AsyncMock()) + before=datetime.now(UTC) + result=asyncio.run(backfill.start(session,'preview','history_only',validated=True)) + assert result=={'id':7} and settings.enabled + if original_start is None: assert before<=settings.start_at<=datetime.now(UTC) + else: assert settings.start_at==original_start + session.__aenter__=AsyncMock(return_value=session);session.__aexit__=AsyncMock(return_value=False) + job=N(id=7,settings={'request':{'run_mode':'history_only'}}) + session.scalar=AsyncMock(side_effect=[job,1,job,None]) + session.scalars=AsyncMock(return_value=N(all=lambda:[N(id=1,news_id=1,content_hash='current')])) + monkeypatch.setattr(scheduler,'SessionLocal',lambda:session) + picked=asyncio.run(scheduler.candidate(set(),0)) + assert picked[0]==('backfill',1) + + +@pytest.mark.parametrize('dependency_state', ['pending','failed','done']) +def test_exclusive_event_enrolls_only_unfinished_dependencies_and_preserves_failures(trading,monkeypatch,dependency_state): + from alphapulse.news_risk import backfill + from alphapulse.models.risk import ReprocessJob,NewsEvent,RiskSettings + config={'request':{'run_mode':'history_only'}} + job=N(id=7,state='running',settings=config,failures=[]) + event=N(id=9,state='pending',published_revision=0) + dependency=N(id=2,state=dependency_state,revision=2,published_revision=1,result=None, + review_config=None,error='Synthetic' if dependency_state=='failed' else None,dirty_since=None) + target=N(kind='events',object_id=9,revision=1) + settings=N(enabled=True) + session=MagicMock() + session.get=AsyncMock(side_effect=lambda model,*a,**k:{ReprocessJob:job,NewsEvent:event,RiskSettings:settings}[model]) + session.scalar=AsyncMock(side_effect=[None,2 if dependency_state=='failed' else None]) + session.scalars=AsyncMock(side_effect=[N(all=lambda:[]),N(all=lambda:[target]),N(all=lambda:[dependency])]) + session.commit=AsyncMock() + enroll=AsyncMock();monkeypatch.setattr(backfill,'target',enroll) + asyncio.run(backfill.reconcile(session,job)) + assert dependency.review_config==config + enroll.assert_awaited_once_with(session,7,'analysis',dependency) + if dependency_state=='failed': + assert dependency.state=='failed' and job.state=='failed' and not settings.enabled + assert any(f['stage']=='analysis' and f['id']==2 for f in job.failures) + else: + assert dependency.state=='pending' and job.state=='running' and settings.enabled + assert dependency.due_at<=datetime.now(UTC) + assert dependency.revision==(3 if dependency_state=='done' else 2) + + +def test_exclusive_batch_finishes_after_event_dependency_is_published(trading,monkeypatch): + from alphapulse.news_risk import backfill,events + from alphapulse.models.risk import ReprocessJob,NewsEvent,NewsSummary,RiskSettings,EventMember + config={'request':{'run_mode':'history_only'},'configurations':{'outlook':{'rules':{}}}} + job=N(id=7,state='running',settings=config,failures=[]) + event=N(id=9,state='done',published_revision=1) + dependency=N(id=2,state='done',revision=2,published_revision=2,replacement_ids=[]) + targets=[N(kind='events',object_id=9,revision=1),N(kind='analysis',object_id=2,revision=2)] + settings=N(enabled=True) + session=MagicMock() + session.get=AsyncMock(side_effect=lambda model,*a,**k:{ + ReprocessJob:job,NewsEvent:event,NewsSummary:dependency,RiskSettings:settings,EventMember:None}[model]) + session.scalar=AsyncMock(side_effect=[None,None,N(id=3,state='done',quotes={})]) + session.scalars=AsyncMock(side_effect=[N(all=lambda:[]),N(all=lambda:targets),N(all=lambda:[])]) + session.commit=AsyncMock() + monkeypatch.setattr(events,'reading_inputs',AsyncMock(return_value=[])) + asyncio.run(backfill.reconcile(session,job)) + assert job.state=='done' and not settings.enabled + + +@pytest.mark.parametrize('failed_stage', ['grouping','analysis','events',None]) +@pytest.mark.parametrize('prepare_only', [True,False]) +def test_exclusive_outlook_never_admits_failed_batch_results(trading,monkeypatch,failed_stage,prepare_only): + from alphapulse.news_risk import worker,events,reprocess + from alphapulse.models.risk import ReprocessJob,ReprocessItem,NewsSummary,NewsEvent + job=N(id=7,state='running',position=0,items=[],settings={'request':{'run_mode':'history_only'}}) + failed_model={'grouping':ReprocessItem,'analysis':NewsSummary,'events':NewsEvent}.get(failed_stage) + session=MagicMock() + session.__aenter__=AsyncMock(return_value=session);session.__aexit__=AsyncMock(return_value=False) + session.get=AsyncMock(return_value=N(enabled=True));session.commit=AsyncMock() + async def scalar(query): + model=query.column_descriptions[0]['entity'] + if model is ReprocessJob: return job + if model is failed_model: + # Emulate filtering a stored failed row, without storage or model I/O. + return 1 if any(isinstance(value,list) and 'failed' in value for value in query.compile().params.values()) else None + return None + session.scalar=AsyncMock(side_effect=scalar) + inputs=[{'id':1,'kind':'package','revision':1,'importance_stars':3,'category':'company','result':{'summary':'Previous successful result'}}] + monkeypatch.setattr(worker,'SessionLocal',lambda:session) + monkeypatch.setattr(events,'reading_inputs',AsyncMock(return_value=inputs)) + configuration=AsyncMock(return_value=(N(model='advanced'),'rules',{})) + monkeypatch.setattr(reprocess,'frozen_configuration',configuration) + generate=AsyncMock(return_value={'reading':[],'synthesis':None,'usage':[]}) + monkeypatch.setattr(worker.briefing,'generate',generate) + ready=asyncio.run(worker.outlook(prepare_only=prepare_only)) + if failed_stage: + assert ready is None + configuration.assert_not_awaited();generate.assert_not_awaited();session.add.assert_not_called() + elif prepare_only: + assert ready is True + generate.assert_not_awaited() + else: + assert generate.await_count==2 + assert session.add.call_args.args[0].state=='done' + + +@pytest.mark.parametrize('case', ['valid','wrong_date','wrong_text','wrong_quote','extra_evidence']) +def test_event_plan_binds_date_precision_text_and_its_original_quote(trading,case): + from alphapulse.news_risk.events import EventResult,validate_timeline + def point(day): + return dict(occurred_at=None,time_precision='unknown',planned_at=f'2026-11-{day}T00:00:00Z', + planned_precision='day',time_text=f'11月{day}日',reported_at=None,description='Plan',verification='partial', + evidence_refs=[{'article_id':1,'quote':f'计划11月{day}日开始'}]) + first,second=point('10'),point('20') + submitted={**first,'package_id':1,'package_revision':1} + if case=='wrong_date': submitted={**second,'planned_at':first['planned_at'],'package_id':1,'package_revision':1} + if case=='wrong_text': submitted['time_text']=second['time_text'] + if case=='wrong_quote': submitted['evidence_refs']=second['evidence_refs'] + if case=='extra_evidence': submitted['evidence_refs']=first['evidence_refs']+second['evidence_refs'] + result=EventResult(title='Synthetic',summary='Synthetic',category='company',importance_stars=3,risk_stars=0, + uncertainty='Unknown',opinions='Unknown',counter_evidence='Unknown',watchpoints='Unknown',timeline_points=[submitted]) + packages=[{'id':1,'revision':1,'result':{'timeline_points':[first,second]}}] + if case in ('valid','extra_evidence'): validate_timeline(result,packages) + else: + with pytest.raises(ValueError,match='已核验计划'): validate_timeline(result,packages) diff --git a/deploy/README.md b/deploy/README.md index ebca0f1..4f2c3d1 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -207,3 +207,5 @@ Grouping retains each coherent publication as a complete news unit: related metr Reading ranks all important inputs, shortlists at most twelve low-star candidates across domains, selects up to three expansion entries, then synthesizes once. Saved priorities are reused by synthesis. Reprocess previews accept `selection=attempted` to freeze distinct previously attempted news IDs at their latest source hashes and force regrouping. `run_mode=history_only` requires paused, idle AI; startup uses the existing three-stage validation gate, enables only this durable batch, and pauses on completion, cancellation or failure. It does not consume realtime backlog. Grouping finishes before downstream review; completion also waits for the final reading snapshot. Existing range/shared behavior remains the default. These additions use existing JSON storage and require no schema migration. + +An exclusive batch also enrolls unfinished member packages required by its events, using the same frozen configuration. This may add package reviews beyond the selected news; unrelated realtime work stays queued. Failed members remain failed until explicitly retried. Any grouping, package or event failure blocks advanced reading generation. The first exclusive start initializes the dispatch boundary while preserving an existing boundary. Event plan dates must match the verified time point's precision, original time text and at least one corresponding time quote.