diff --git a/backend/src/alphapulse/api/risk.py b/backend/src/alphapulse/api/risk.py index 034e49c..7406354 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, Literal from fastapi import APIRouter, Body, Depends, HTTPException, Query from pydantic import BaseModel, Field, model_validator -from sqlalchemy import select, func, text +from sqlalchemy import select, func, text, BigInteger from sqlalchemy.ext.asyncio import AsyncSession from alphapulse.ai.llm import Effort from alphapulse.api.response import success @@ -437,7 +437,7 @@ async def reprocess_retry(identifier: int, session: DB): await session.execute(update(ReprocessItem).where(ReprocessItem.job_id==job.id,ReprocessItem.state=='failed').values(state='pending',error=None)) outlook_ids = {f['id'] for f in job.failures if f['stage'] == 'outlook'} snapshots = (await session.scalars(select(RiskSnapshot).where(RiskSnapshot.id.in_(outlook_ids), RiskSnapshot.state == 'failed', - RiskSnapshot.quotes['job_id'].as_integer() == job.id))).all() + RiskSnapshot.quotes['job_id'].astext.cast(BigInteger) == job.id))).all() for snapshot in snapshots: snapshot.quotes = {**snapshot.quotes, 'retry_requested': True} from alphapulse.models.risk import NewsEvent for stage, model in (('analysis', NewsSummary), ('events', NewsEvent)): diff --git a/backend/src/alphapulse/news_risk/queue.py b/backend/src/alphapulse/news_risk/queue.py index d8eaa82..b705ecc 100644 --- a/backend/src/alphapulse/news_risk/queue.py +++ b/backend/src/alphapulse/news_risk/queue.py @@ -1,7 +1,7 @@ """Queue presentation combines durable work with short-lived execution steps.""" import json from datetime import UTC, datetime -from sqlalchemy import select, func, or_, exists +from sqlalchemy import select, func, or_, exists, BigInteger from alphapulse.models import NewsItem from alphapulse.models.risk import (ReprocessItem, ReprocessTarget, ArticleAnalysis, NewsSummary, NewsEvent, RiskSnapshot, RiskSettings, ReprocessJob) from alphapulse.core.redis import redis @@ -61,15 +61,15 @@ async def listing(session, state, stage, origin, job_id, page, limit): if origin=='realtime':continue if job_id:query=query.where(model.job_id==job_id) elif kind in ('analysis','events'): - batch=model.review_config['job_id'].as_integer() + batch=model.review_config['job_id'].astext.cast(BigInteger) if origin:query=query.where(batch.is_not(None) if origin=='history' else batch.is_(None)) if job_id: tracked=exists(select(ReprocessTarget.id).where(ReprocessTarget.job_id==job_id, ReprocessTarget.kind==kind,ReprocessTarget.object_id==model.id)) query=query.where(or_(tracked,batch==job_id)) elif kind=='outlook': - if job_id:query=query.where(model.quotes['job_id'].as_integer()==job_id) - if origin:query=query.where(model.quotes['job_id'].as_integer().is_not(None) if origin=='history' else model.quotes['job_id'].as_integer().is_(None)) + if job_id:query=query.where(model.quotes['job_id'].astext.cast(BigInteger)==job_id) + if origin:query=query.where(model.quotes['job_id'].astext.cast(BigInteger).is_not(None) if origin=='history' else model.quotes['job_id'].astext.cast(BigInteger).is_(None)) elif job_id or origin=='history':continue count = await session.scalar(select(func.count()).select_from(query.subquery())) total += count diff --git a/backend/src/alphapulse/news_risk/reprocess.py b/backend/src/alphapulse/news_risk/reprocess.py index c5b7e22..f0303dc 100644 --- a/backend/src/alphapulse/news_risk/reprocess.py +++ b/backend/src/alphapulse/news_risk/reprocess.py @@ -1,5 +1,5 @@ """Explicit bounded historical jobs, sharing the normal news pipeline.""" -from sqlalchemy import select, text, update +from sqlalchemy import select, text, update, BigInteger from alphapulse.core.db import SessionLocal from alphapulse.models import NewsItem from alphapulse.models.risk import RiskSettings, ReprocessJob, ReprocessItem, ArticleAnalysis, NewsSummary, EventMember @@ -68,7 +68,7 @@ async def process_job(item_index=None): inputs = await reading_inputs(session, datetime.now(UTC)-timedelta(hours=72)) key = briefing.fingerprint(briefing.semantic_packages(inputs), job.settings['configurations']['outlook']['rules']) snapshot = await session.scalar(select(RiskSnapshot).where( - RiskSnapshot.quotes['job_id'].as_integer() == job.id, + RiskSnapshot.quotes['job_id'].astext.cast(BigInteger) == job.id, RiskSnapshot.quotes['fingerprint'].as_string() == key).order_by(RiskSnapshot.id.desc()).limit(1)) if snapshot is None or snapshot.state == 'processing' or snapshot.quotes.get('retry_requested'): return if snapshot.state == 'failed': diff --git a/scripts/smoke_backfill.py b/scripts/smoke_backfill.py index 6172ed1..1331575 100644 --- a/scripts/smoke_backfill.py +++ b/scripts/smoke_backfill.py @@ -324,6 +324,27 @@ async def main(): assert await session.scalar(select(func.count()).select_from(RiskSnapshot))==count+2 assert not await worker.outlook(prepare_only=True) no_model.assert_not_awaited() + # JSON batch references must preserve the BIGINT identity used by job rows. + async with SessionLocal() as session: + original=await session.get(ReprocessJob,job_id) + large=ReprocessJob(id=2147483648,state='failed',settings={**original.settings,'job_id':2147483648}, + items=[],position=0,failures=[]) + package=await session.get(NewsSummary,package_ids[0]) + package.review_config=large.settings; package.state='failed' + event=NewsEvent(title='Large batch',updated_at=now,due_at=now,state='failed',review_config=large.settings) + snapshot=RiskSnapshot(state='failed',model='senior',quotes={'job_id':large.id}) + session.add_all([large,event,snapshot]); await session.flush() + large.failures=[{'stage':'outlook','id':snapshot.id,'error':'Synthetic large batch failure'}] + await session.commit() + result=await queue.listing(session,'failed',None,'history',large.id,1,20) + assert {(r['stage'],r['object_id']) for r in result['items']}=={ + ('analysis',package.id),('events',event.id),('outlook',snapshot.id)} + for state in ('running','queued','waiting','failed','done',None): + result=await queue.listing(session,state,None,None,9223372036854775807,1,20) + assert result['total']==0 and result['items']==[] + from alphapulse.api.risk import reprocess_retry + await reprocess_retry(large.id,session) + assert (await session.get(RiskSnapshot,snapshot.id)).quotes['retry_requested'] async with SessionLocal() as session: for model in (ReprocessTarget,ReprocessItem,ReprocessJob,EventMember,NewsEvent,PackageCorrection,RiskSnapshot,GroupingVersion,SummaryArticle,NewsSummary,ArticleAnalysis): await session.execute(delete(model))