From 0a150eca0f926c47de58ef0e1728ee7fac340e42 Mon Sep 17 00:00:00 2001 From: eezd Date: Sun, 1 Nov 2026 15:13:44 +0800 Subject: [PATCH] fix: preserve bigint batch identities in news AI queries Use BIGINT casts for JSON batch references in queue filters, retries and legacy job reconciliation. Cover real large batch references and absent 64-bit IDs in isolated PostgreSQL acceptance. --- backend/src/alphapulse/api/risk.py | 4 ++-- backend/src/alphapulse/news_risk/queue.py | 8 +++---- backend/src/alphapulse/news_risk/reprocess.py | 4 ++-- scripts/smoke_backfill.py | 21 +++++++++++++++++++ 4 files changed, 29 insertions(+), 8 deletions(-) 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))