252 lines
18 KiB
Python
252 lines
18 KiB
Python
"""Exercise event aggregation and correction transactions in the disposable CI DB."""
|
|
import asyncio
|
|
from datetime import UTC, datetime, timedelta
|
|
import os
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
from urllib.parse import urlparse
|
|
import httpx
|
|
from fastapi import FastAPI
|
|
from sqlalchemy import select, func, delete, null
|
|
from alphapulse.core.db import SessionLocal, engine
|
|
from alphapulse.models import AiProvider, NewsItem, SkillNode, SkillVersion
|
|
from alphapulse.models.risk import NewsSummary, ArticleAnalysis, SummaryArticle, RiskSettings, NewsEvent, EventMember, EventVersion, ReprocessJob, PackageCorrection, GroupingVersion, SummaryVersion, PackageEvidence, RiskSnapshot, EventLink
|
|
from alphapulse.ai.skills import create_node, add_version
|
|
from alphapulse.news_risk import pipeline, events, reprocess, worker, briefing
|
|
from alphapulse.news_risk.contracts import Correction
|
|
from alphapulse.news_risk.corrections import apply_correction
|
|
|
|
|
|
async def main():
|
|
url=urlparse(os.environ['DATABASE_URL'])
|
|
assert url.username=='ci' and url.password=='ci-only' and url.path=='/ci'
|
|
now=datetime.now(UTC)
|
|
async with SessionLocal() as session:
|
|
assert await session.scalar(select(func.count()).select_from(RiskSettings))==0
|
|
provider=AiProvider(name='event-ci',protocol='openai',base_url='https://example.invalid',api_key_encrypted=b'fake',api_key_hint='fake',model='shared',max_tokens=8192,context_window=200000)
|
|
session.add(provider); await session.flush()
|
|
ids={}
|
|
for role,(name,content) in pipeline.TEMPLATES.items():
|
|
node=await create_node(session,None,'skill',name,scope='news')
|
|
version=await add_version(session,node,content,'user','CI'); node.published_version_id=version.id
|
|
ids[role+'_skill_id']=node.id
|
|
settings=RiskSettings(id=1,enabled=True,start_at=now-timedelta(minutes=5),provider_id=provider.id,second_provider_id=provider.id,third_provider_id=provider.id,model='luna',second_model='reviewer',third_model='senior',**ids)
|
|
session.add(settings)
|
|
packages=[]; analyses=[]
|
|
for i in (1,2):
|
|
news=NewsItem(source_code='wscn_news',external_id='ci-event-'+str(i),url='https://example.invalid/'+str(i),title='Incident' if i==1 else 'Response',summary='',content_text='Original evidence '+str(i),content_hash=str(i),published_at=now)
|
|
session.add(news); await session.flush()
|
|
analysis=ArticleAnalysis(news_id=news.id,content_hash=news.content_hash,state='done',result={'previous':True},input={'title':news.title,'summary':'','content':news.content_text,'url':news.url,'source':'wscn_news','revision_no':1,'published_at':now.isoformat(),'progress_at':now.isoformat()})
|
|
session.add(analysis); await session.flush(); analyses.append(analysis)
|
|
result={'title':news.title,'summary':news.title,'importance_stars':4,'risk_stars':3,'category':'security','evidence_refs':[{'article_id':analysis.id,'quote':news.content_text}],'claims':[]}
|
|
package=NewsSummary(title=news.title,fact=news.title,category='security',tags=['exchange'],entities=['Exchange'],content_kind='report',published_at=now,updated_at=now,changed_at=now,due_at=now,state='done',revision=1,published_revision=1,result=result,importance_stars=4)
|
|
session.add(package); await session.flush(); packages.append(package)
|
|
session.add(SummaryArticle(summary_id=package.id,analysis_id=analysis.id,active=True,unit={'fact':news.title,'evidence':news.content_text}))
|
|
await session.flush()
|
|
result=SimpleNamespace(related_events=[],event_proposal=pipeline.EventProposal(title='Exchange incident',package_ids=[packages[0].id],reason='Official response to the same incident'))
|
|
await events.associate(session,packages[1],result,{packages[0].id})
|
|
event=await session.scalar(select(NewsEvent)); event.due_at=now
|
|
package_ids=[p.id for p in packages]; event_id=event.id
|
|
withdrawn = NewsSummary(title='Withdrawn',fact='Withdrawn',category='security',tags=['exchange'],entities=[],
|
|
content_kind='report',published_at=now,updated_at=now,changed_at=now,due_at=now,state='withdrawn',revision=1,published_revision=0)
|
|
session.add(withdrawn); await session.flush()
|
|
session.add(EventMember(package_id=withdrawn.id,event_id=event_id,reason='Previous member'))
|
|
await session.commit()
|
|
async with SessionLocal() as session:
|
|
nested = await session.begin_nested()
|
|
await session.execute(NewsEvent.__table__.update().where(NewsEvent.id == event_id).values(result=None))
|
|
await session.execute(NewsSummary.__table__.update().where(NewsSummary.id.in_(package_ids)).values(state='pending'))
|
|
for empty in (None, null()):
|
|
session.add(NewsSummary(title='Unreviewed',fact='Pending',category='security',entities=[],
|
|
published_at=now,updated_at=now,changed_at=now,due_at=now,state='pending',result=empty))
|
|
session.add(NewsEvent(title='Unreviewed',updated_at=now,due_at=now,state='pending',result=empty))
|
|
for empty in (None, null(), {'inputs':None}):
|
|
session.add(RiskSnapshot(state='done',result=empty,quotes={},model='synthetic'))
|
|
await session.flush()
|
|
inputs = await events.reading_inputs(session,now-timedelta(hours=72))
|
|
assert len(briefing.semantic_packages(inputs)) == 2
|
|
assert {(p['kind'],p['id']) for p in inputs} == {('package',i) for i in package_ids}
|
|
from alphapulse.api import risk as api
|
|
app = FastAPI()
|
|
app.include_router(api.router)
|
|
app.dependency_overrides[api.require_admin] = lambda: None
|
|
app.dependency_overrides[api.get_session] = lambda: session
|
|
async with httpx.AsyncClient(transport=httpx.ASGITransport(app),base_url='http://ci') as client:
|
|
for hours in (24,48,72,168):
|
|
response = await client.get('/api/risk/overview',params={'hours':hours})
|
|
assert response.status_code == 200, response.text
|
|
data = response.json()['data']
|
|
assert data['briefing'] is None and data['stale']
|
|
assert sum(data['counts'].values()) == 2
|
|
response = await client.get('/api/risk/summaries')
|
|
assert response.status_code == 200, response.text
|
|
data = response.json()['data']
|
|
assert data['total'] == 2 and {p['id'] for p in data['items']} == set(package_ids)
|
|
await nested.rollback()
|
|
print('Reading API: SQL NULL and JSON null results excluded; pending events preserve independent packages')
|
|
async def complete(provider,system,payload,schema):
|
|
points=[]
|
|
for p in payload['packages']:
|
|
points.append(events.TimelinePoint(occurred_at=None,time_precision='unknown',time_text='Time unknown',reported_at=now,description=p['result']['summary'],verification='partial',package_id=p['id'],package_revision=p['revision'],evidence_refs=p['result']['evidence_refs']))
|
|
return schema(title='Exchange incident',summary='Disclosure followed by response',category='security',importance_stars=4,risk_stars=3,uncertainty='Time unknown',opinions='Wait',counter_evidence='Unknown',watchpoints='Investigation',timeline_points=points),{}
|
|
with patch.object(events,'complete',complete):
|
|
await events.organize()
|
|
await events.organize()
|
|
async with SessionLocal() as session:
|
|
event=await session.get(NewsEvent,event_id)
|
|
assert event.state=='done' and len(event.result['timeline_points'])==2
|
|
inputs=await events.reading_inputs(session,now-timedelta(hours=72))
|
|
assert len(inputs)==1 and inputs[0]['kind']=='event'
|
|
assert await session.scalar(select(func.count()).select_from(EventVersion))==1
|
|
counts,_=await reprocess.preview(session,now-timedelta(seconds=1),now+timedelta(seconds=1))
|
|
assert counts=={'articles':2,'packages':2,'events':1}
|
|
started=await reprocess.start_job(session,now-timedelta(seconds=1),now+timedelta(seconds=1))
|
|
job=await session.get(ReprocessJob,started['id'])
|
|
assert len(job.items)==2
|
|
provider=await session.get(AiProvider,provider.id); provider.model='changed'; await session.commit()
|
|
frozen,_,_=await reprocess.frozen_configuration(session,job.settings,'grouping')
|
|
assert frozen.model=='luna'
|
|
job_id = job.id
|
|
await session.commit()
|
|
async with SessionLocal() as session:
|
|
event = await session.get(NewsEvent, event_id)
|
|
event.state = 'pending'; event.revision += 1; event.due_at = now
|
|
published = event.published_revision
|
|
await session.commit()
|
|
async def change_during_timeline(*args):
|
|
async with SessionLocal() as session:
|
|
package = await session.get(NewsSummary, package_ids[0])
|
|
worker.schedule(package, now, True)
|
|
await session.commit()
|
|
return await complete(*args)
|
|
with patch.object(events, 'complete', change_during_timeline): await events.organize()
|
|
async with SessionLocal() as session:
|
|
event = await session.get(NewsEvent, event_id)
|
|
assert event.state == 'pending' and event.published_revision == published
|
|
latest = await session.scalar(select(EventVersion).order_by(EventVersion.id.desc()).limit(1))
|
|
assert latest.state == 'superseded'
|
|
package = await session.get(NewsSummary, package_ids[0])
|
|
package.state = 'done'; package.published_revision = package.revision
|
|
await session.commit()
|
|
with patch.object(events, 'complete', complete): await events.organize()
|
|
analyze = worker.analyze
|
|
async def pause_before_grouping(*args):
|
|
async with SessionLocal() as session:
|
|
settings = await session.get(RiskSettings, 1); settings.enabled = False
|
|
await session.commit()
|
|
return await analyze(*args)
|
|
with patch.object(worker, 'analyze', pause_before_grouping): await reprocess.process_job()
|
|
async with SessionLocal() as session:
|
|
job = await session.get(ReprocessJob, job_id)
|
|
assert job.position == 0 and not job.failures
|
|
assert (await session.get(ArticleAnalysis, analyses[0].id)).result == {'previous':True}
|
|
settings = await session.get(RiskSettings, 1); settings.enabled = True
|
|
await session.commit()
|
|
async def fail_grouping(*args):
|
|
from alphapulse.api.risk import reprocess_cancel
|
|
async with SessionLocal() as other:
|
|
job = await other.get(ReprocessJob, job_id)
|
|
if job.state != 'cancelled': await reprocess_cancel(job_id, other)
|
|
raise ValueError('Synthetic grouping failure during cancellation')
|
|
with patch.object(worker, 'complete', fail_grouping): await reprocess.process_job()
|
|
async with SessionLocal() as session:
|
|
job = await session.get(ReprocessJob, job_id)
|
|
assert job.state == 'cancelled' and job.position == 1 and len(job.failures) == 1
|
|
old = await session.get(ArticleAnalysis, analyses[0].id)
|
|
assert old.result == {'previous':True} and old.state == 'done' and old.error
|
|
assert await session.scalar(select(func.count()).select_from(SummaryArticle).where(SummaryArticle.active.is_(True))) == 2
|
|
assert await session.scalar(select(func.count()).select_from(GroupingVersion).where(GroupingVersion.state == 'failed')) == 1
|
|
second = await reprocess.start_job(session, now-timedelta(seconds=1), now+timedelta(seconds=1))
|
|
job = await session.get(ReprocessJob, second['id'])
|
|
job.state = 'running'; job.position = len(job.items)
|
|
package = await session.get(NewsSummary, package_ids[0])
|
|
package.review_config = job.settings; package.state = 'failed'; package.error = 'Synthetic review failure'
|
|
await session.commit()
|
|
await reprocess.process_job()
|
|
async with SessionLocal() as session:
|
|
job = await session.get(ReprocessJob, second['id'])
|
|
assert job.state == 'failed' and job.failures == [{'stage':'analysis','id':package_ids[0],'error':'Synthetic review failure'}]
|
|
from alphapulse.api.risk import reprocess_retry
|
|
await reprocess_retry(job.id, session)
|
|
assert job.position == len(job.items) and job.state == 'running'
|
|
assert (await session.get(NewsSummary, package_ids[1])).state == 'done'
|
|
assert (await session.get(NewsSummary, package_ids[0])).state == 'pending'
|
|
assert await session.scalar(select(func.count()).select_from(GroupingVersion)) == 1
|
|
job.state = 'cancelled'
|
|
package = await session.get(NewsSummary, package_ids[0])
|
|
package.state = 'done'; package.error = None; package.review_config = None
|
|
news = NewsItem(source_code='wscn_news', external_id='ci-event-repost', url='https://example.invalid/repost',
|
|
title='Incident', summary='', content_text='Original evidence 1', content_hash='repost', published_at=now)
|
|
session.add(news); await session.flush()
|
|
repost = ArticleAnalysis(news_id=news.id, content_hash=news.content_hash, state='pending', input={
|
|
'title':news.title,'summary':'','content':news.content_text,'url':news.url,'source':'wscn_news',
|
|
'revision_no':1,'published_at':now.isoformat(),'progress_at':now.isoformat()})
|
|
session.add(repost); await session.commit(); analyses.append(repost)
|
|
old_revision, old_progress = package.revision, package.updated_at
|
|
async def group_repost(provider, system, payload, schema):
|
|
return schema(reason='Same concrete fact', packages=[dict(package_id=package_ids[0], group_key=None,
|
|
title='Incident', category='security', entities=['Exchange'], fact='Incident', evidence='Original evidence 1',
|
|
urgent=False, tags=['exchange'], content_kind='report', change_kind='repost')]), {}
|
|
with patch.object(worker, 'complete', group_repost): await worker.analyze(repost.id)
|
|
async with SessionLocal() as session:
|
|
package = await session.get(NewsSummary, package_ids[0])
|
|
assert package.revision == old_revision and package.updated_at == old_progress and package.state == 'done'
|
|
third = await reprocess.start_job(session, now-timedelta(seconds=1), now+timedelta(seconds=1))
|
|
job = await session.get(ReprocessJob, third['id'])
|
|
job.state = 'running'; job.position = len(job.items)
|
|
await session.commit()
|
|
await reprocess.process_job()
|
|
async with SessionLocal() as session:
|
|
assert (await session.get(ReprocessJob, third['id'])).state == 'running'
|
|
async def senior(provider, system, payload, schema):
|
|
assert provider.model == 'senior'
|
|
if schema is briefing.Ranked:
|
|
return schema(reading=[dict(kind=p['kind'], id=p['id'], revision=p['revision'], priority=80, reason='Synthetic')
|
|
for p in payload['packages']], notes='Synthetic'), {}
|
|
trend = briefing.Trend(summary='Unknown', conditions='Unknown', counter_evidence='Unknown', watchpoints='Updates')
|
|
return schema(summary='Synthetic', market='Unknown', short_term=trend, medium_term=trend, references=[]), {}
|
|
async def failed_senior(*args): raise ValueError('Synthetic outlook failure')
|
|
with patch.object(briefing, 'complete', failed_senior): await worker.outlook()
|
|
await reprocess.process_job()
|
|
async with SessionLocal() as session:
|
|
job = await session.get(ReprocessJob, third['id'])
|
|
assert job.state == 'failed' and len(job.failures) == 1 and job.failures[0]['stage'] == 'outlook'
|
|
groups_before = await session.scalar(select(func.count()).select_from(GroupingVersion))
|
|
await reprocess_retry(job.id, session)
|
|
assert job.position == len(job.items) == 3
|
|
with patch.object(briefing, 'complete', senior): await worker.outlook()
|
|
await reprocess.process_job()
|
|
async with SessionLocal() as session:
|
|
assert (await session.get(ReprocessJob, third['id'])).state == 'done'
|
|
assert await session.scalar(select(func.count()).select_from(GroupingVersion)) == groups_before
|
|
packages=[await session.get(NewsSummary,i) for i in package_ids]
|
|
proposal=Correction(packages=[{'id':p.id,'revision':p.revision} for p in packages],replacements=[{'keep_id':min(package_ids),'title':'Corrected shared fact','fact':'Same fact','category':'security','tags':['exchange'],'content_kind':'report','assignments':[{'analysis_id':a.id,'evidence':a.input['content']} for a in analyses]}],reason='Synthetic duplicate correction')
|
|
output=await apply_correction(session,proposal,package_ids[0],{package_ids[1]},{'test':True})
|
|
await session.commit()
|
|
assert output==[min(package_ids)]
|
|
assert (await session.get(NewsSummary,max(package_ids))).replacement_ids==output
|
|
assert await session.scalar(select(func.count()).select_from(PackageCorrection))==1
|
|
assert await session.scalar(select(func.count()).select_from(SummaryArticle).where(SummaryArticle.active.is_(True)))==3
|
|
assert await session.scalar(select(func.count()).select_from(NewsItem).where(NewsItem.external_id.startswith('ci-event-')))==3
|
|
await events.organize()
|
|
async with SessionLocal() as session:
|
|
event=await session.get(NewsEvent,event_id); event.due_at=now; await session.commit()
|
|
await events.organize()
|
|
async with SessionLocal() as session:
|
|
assert (await session.get(NewsEvent,event_id)).state=='retired'
|
|
assert len(await events.reading_inputs(session,now-timedelta(hours=72)))==1
|
|
async with SessionLocal() as session:
|
|
for model in (EventLink, EventVersion, EventMember, NewsEvent, ReprocessJob, PackageCorrection,
|
|
GroupingVersion, PackageEvidence, SummaryVersion, SummaryArticle, NewsSummary, ArticleAnalysis, RiskSnapshot, RiskSettings):
|
|
await session.execute(delete(model))
|
|
await session.execute(delete(NewsItem).where(NewsItem.external_id.startswith('ci-event-')))
|
|
await session.execute(delete(AiProvider).where(AiProvider.id == provider.id))
|
|
await session.execute(SkillNode.__table__.update().where(SkillNode.id.in_(ids.values())).values(published_version_id=None))
|
|
await session.execute(delete(SkillVersion).where(SkillVersion.skill_id.in_(ids.values())))
|
|
await session.execute(delete(SkillNode).where(SkillNode.id.in_(ids.values())))
|
|
await session.commit()
|
|
await engine.dispose()
|
|
print('Events: original evidence, independent packages, aggregation, timeline, preview, frozen config and atomic merge history passed')
|
|
|
|
asyncio.run(main())
|