fix(news): harden exclusive batch completion and timeline evidence

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)):

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):

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:

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)

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.