Prioritize realtime news and make historical backfill resumable

scripts/ci.py

@@ -182,6 +182,8 @@ def smoke(build, label):
                      'import urllib.request; urllib.request.urlopen("http://localhost:8000/api/health", timeout=3)')
         print(run('docker', 'exec', '-i', names['api'], 'python', '-',
                   input=(ROOT / 'scripts/smoke_news_batches.py').read_text()), flush=True)
+        print(run('docker', 'exec', '-i', names['api'], 'python', '-',
+                  input=(ROOT / 'scripts/smoke_priority_queue.py').read_text()), flush=True)
         print(run('docker', 'exec', '-i', names['api'], 'python', '-',
                   input=(ROOT / 'scripts/smoke_news_research.py').read_text()), flush=True)
         print(run('docker', 'exec', '-i', names['api'], 'python', '-',

scripts/smoke_news_batches.py

@@ -90,6 +90,7 @@ async def main():
     await backfill_regressions(now)
     await carryover_regressions(now)
     await regressions(now, settings)
+    await queue_review_regressions(now, settings)
     await engine.dispose()
     print('Frozen batches passed: assignment barrier, single research, failures, new arrivals, restart and source revisions.')
 
@@ -208,7 +209,13 @@ async def regressions(now, settings):
     async with SessionLocal() as s:
         event = Event(title='Deferred retry', tags=[], revision=7, state='pending', due_at=now)
         previous = Batch(phase='partial', cutoff_at=now)
-        s.add_all([event, previous])
+        original = NewsItem(source_code='wscn_news', external_id='retry-member', title='retry', summary='', content_text='retry', content_hash='a'*64, url='https://example.invalid/retry', published_at=now)
+        s.add_all([event, previous, original])
+        await s.flush()
+        member = Package(news_id=original.id, content_hash=original.content_hash, state='done', keywords=[])
+        s.add(member)
+        await s.flush()
+        s.add(Membership(event_id=event.id, package_id=member.id, evidence='retry', reason='retry', change_kind='new_fact'))
         await s.flush()
         s.add(BatchEvent(batch_id=previous.id, event_id=event.id, revision=7, state='deferred'))
         await s.commit()
@@ -236,39 +243,116 @@ async def regressions(now, settings):
         s.add_all([news, excluded])
         await s.commit()
         await service.enroll(s, [news.id])
-        batch = await batches.freeze(s, settings)
+        assert await batches.freeze(s, settings) is None
         package = await s.scalar(select(Package).where(Package.news_id == news.id))
         news.content_hash = 'b'*64
         await s.commit()
-        await batches.actions(s, batch, set(), 8)
-        await s.commit()
-        await batches.actions(s, batch, set(), 8)
-        await s.commit()
-        assert batch.phase == 'partial'
-        batch = await batches.freeze(s, settings)
-        assert package.content_hash == news.content_hash and package.state == 'index_pending'
-        assert await batches.actions(s, batch, set(), 8) == [('index', package.id)]
-        assert await s.scalar(select(func.count()).select_from(Package)) == 1
-        # A previously completed manual original also refreshes outside the automatic boundary.
-        package.state = 'done'
-        news.content_hash = 'c'*64
-        await s.commit()
-        await batches.freeze(s, settings)
+        assert await batches.freeze(s, settings) is None
+        assert package.content_hash != news.content_hash
+        expanded = SimpleNamespace(start_at=news.published_at)
+        batch = await batches.freeze(s, expanded)
         assert package.content_hash == news.content_hash and package.state == 'index_pending'
+        assert ('index', package.id) in await batches.actions(s, batch, set(), 8)
         await clear(s)
 
         count = 33000
         await s.execute(insert(NewsItem), [original('bulk-'+str(i), now) for i in range(count)])
         await s.commit()
         batch = await batches.freeze(s, settings)
-        assert await s.scalar(select(func.count()).select_from(BatchPackage).where(BatchPackage.batch_id == batch.id)) == count
-        assert await s.scalar(select(func.count()).select_from(Package)) == count
+        assert await s.scalar(select(func.count()).select_from(BatchPackage).where(BatchPackage.batch_id == batch.id)) == 30
+        assert await s.scalar(select(func.count()).select_from(Package)) == 30
         # Exercise the existing-pending-ID and membership arrays above the driver limit too.
         next_batch = await batches.freeze(s, settings)
-        assert await s.scalar(select(func.count()).select_from(BatchPackage).where(BatchPackage.batch_id == next_batch.id)) == count
+        assert await s.scalar(select(func.count()).select_from(BatchPackage).where(BatchPackage.batch_id == next_batch.id)) == 30
         assert len(await batches.actions(s, next_batch, set(), 8)) == 8
         await clear(s)
     print('Batch regressions passed: deferred retries, failed-evidence reuse, manual source revisions, and 33,000 originals without parameter overflow.')
 
 
+async def queue_review_regressions(now, settings):
+    async def clear(session):
+        await session.execute(text('TRUNCATE news_research_batches, news_research_packages, news_research_events, news_items CASCADE'))
+        await session.commit()
+
+    async with SessionLocal() as s:
+        news = NewsItem(source_code='wscn_news', external_id='review-barrier', title='historical',
+                        summary='', content_text='historical', content_hash='a'*64,
+                        url='https://example.invalid/review-barrier', published_at=now)
+        s.add(news)
+        await s.flush()
+        package = Package(news_id=news.id, content_hash=news.content_hash, state='assign_pending', keywords=[])
+        event = Event(title='historical event', tags=[], revision=1, state='pending', due_at=now)
+        batch = Batch(phase='assigning', cutoff_at=now)
+        s.add_all([package, event, batch])
+        await s.flush()
+        s.add(Membership(event_id=event.id, package_id=package.id,
+                         evidence='historical', reason='same event', change_kind='new_fact'))
+        s.add(BatchPackage(batch_id=batch.id, package_id=package.id, content_hash=package.content_hash, state='pending'))
+        await s.commit()
+        unrelated = package.id + 1000
+        # Assignment remains globally serial even when the other call is not in this batch.
+        assert await batches.actions(s, batch, {('assign', unrelated)}, 7) == []
+        assert batch.phase == 'assigning'
+        # Active work on this batch still holds its assignment barrier.
+        package.state = 'done'
+        await s.commit()
+        assert await batches.actions(s, batch, {('assign', package.id)}, 7) == []
+        assert batch.phase == 'assigning'
+        # Unrelated realtime package and event calls must not hold the barrier.
+        realtime = {('index', unrelated), ('event', event.id + 1000)}
+        assert await batches.actions(s, batch, realtime, 6) == []
+        await s.commit()
+        assert batch.phase == 'researching'
+        assert await batches.actions(s, batch, realtime, 6) == [('event', event.id)]
+        await s.commit()
+        event.state = 'done'
+        await s.commit()
+        # A completed result can be visible before the owning task has returned.
+        assert await batches.actions(s, batch, realtime | {('event', event.id)}, 5) == []
+        assert batch.phase == 'researching'
+        assert await batches.actions(s, batch, realtime | {('index', package.id)}, 5) == []
+        assert batch.phase == 'researching'
+        assert await batches.actions(s, batch, realtime, 6) == []
+        await s.commit()
+        assert batch.phase == 'done'
+        # Completed batches release admission of the next historical originals.
+        news2 = NewsItem(source_code='wscn_news', external_id='review-next', title='next', summary='',
+                         content_text='next', content_hash='b'*64,
+                         url='https://example.invalid/review-next', published_at=now)
+        s.add(news2)
+        await s.commit()
+        assert (await batches.freeze(s, settings)).id != batch.id
+        await clear(s)
+
+        # Thirty manual-only deferred heads cannot consume the thirty runnable slots.
+        news = NewsItem(source_code='wscn_news', external_id='review-deferred', title='source', summary='',
+                        content_text='source', content_hash='a'*64,
+                        url='https://example.invalid/review-deferred', published_at=now)
+        s.add(news)
+        await s.flush()
+        package = Package(news_id=news.id, content_hash=news.content_hash, state='done', keywords=[])
+        previous = Batch(phase='partial', cutoff_at=now)
+        events = [Event(title='deferred-head-'+str(i), tags=[], revision=7, state='pending',
+                        due_at=now-timedelta(minutes=35-i)) for i in range(35)]
+        s.add_all([package, previous, *events])
+        await s.flush()
+        for i, event in enumerate(events):
+            s.add(Membership(event_id=event.id, package_id=package.id, evidence='source', reason='same source', change_kind='new_fact'))
+            if i <= 30:
+                s.add(BatchEvent(batch_id=previous.id, event_id=event.id,
+                                 revision=7 if i < 30 else 6, state='deferred'))
+        await s.commit()
+        batch = await batches.freeze(s, settings)
+        assert batch is not None
+        selected = set(await s.scalars(select(BatchEvent.event_id).where(BatchEvent.batch_id == batch.id)))
+        assert selected == {event.id for event in events[30:]}, selected
+        assert all(event.state == 'pending' and event.revision == 7 for event in events[:30])
+        await batches.actions(s, batch, set(), 8)
+        await s.commit()
+        actions = await batches.actions(s, batch, set(), 8)
+        assert set(actions) == {('event', event.id) for event in events[30:]}
+        await clear(s)
+    print('Queue review regressions passed: unrelated realtime calls allow both batch barriers to advance, owning calls still block, assignment stays serial, and deferred heads cannot starve runnable events.')
+
+
 asyncio.run(main())

scripts/smoke_news_research.py

@@ -393,14 +393,20 @@ async def main():
             assert report.json()["data"]["start_at"] == "2026-09-30T16:00:00Z"
             assert set(report.json()["data"]["outside"]) == {"unprocessed", "stale", "pending", "processing", "failed"}
         assert (await client.get("/api/risk/settings")).json()["data"]["start_at"] == previous_start
+        assert (await client.post("/api/risk/enable", json={"start_at": selected_start})).status_code == 422
+        settings = (await client.get("/api/risk/settings")).json()["data"]
+        saved = await client.put("/api/risk/settings", json={**config, "start_at": selected_start, "base_revision": settings["control_revision"]})
+        assert saved.status_code == 200, saved.text
         with patch.object(model_service, "complete", validation):
-            response = await client.post("/api/risk/enable", json={"start_at": selected_start})
+            response = await client.post("/api/risk/enable")
             assert response.status_code == 200, response.text
             assert response.json()["data"]["start_at"] == "2026-09-30T16:00:00Z"
         assert (await client.get("/api/risk/settings")).json()["data"][
             "enabled"
         ] is True
         assert (await client.get("/api/risk/settings")).json()["data"]["start_at"] == "2026-09-30T16:00:00Z"
+        fixed_anchor = (await client.get("/api/risk/settings")).json()["data"]["realtime_from"]
+        assert fixed_anchor is not None
         assert (await client.post("/api/risk/enable")).status_code == 409
         assert (await client.post("/api/risk/pause")).status_code == 200
         for action in ("pause", "save"):
@@ -421,12 +427,13 @@ async def main():
                 return await validation(provider, skill, payload, schema)
 
             with patch.object(model_service, "complete", interrupted_validation):
-                response = await client.post("/api/risk/enable", json={"start_at": "2026-09-01T00:00:00Z"})
+                response = await client.post("/api/risk/enable")
                 assert response.status_code == 409, response.text
             assert (await client.get("/api/risk/settings")).json()["data"][
                 "enabled"
             ] is False
             assert (await client.get("/api/risk/settings")).json()["data"]["start_at"] == "2026-09-30T16:00:00Z"
+            assert (await client.get("/api/risk/settings")).json()["data"]["realtime_from"] == fixed_anchor
         package_id = (await client.get(f"/api/risk/news/{news_id}/packages")).json()[
             "data"
         ][0]["id"]

scripts/smoke_priority_queue.py

@@ -0,0 +1,107 @@
+"""Exercise preemption and recovery using disposable PostgreSQL, never a model."""
+import asyncio
+import os
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+from urllib.parse import urlparse
+from sqlalchemy import select, text, func
+from sqlalchemy.exc import IntegrityError
+from unittest.mock import patch
+from alphapulse.core.db import SessionLocal, engine
+from alphapulse.models import NewsItem
+from alphapulse.news_research.models import Package, Event, Membership, Attempt, Version, ResearchSettings
+from alphapulse.news_research import scheduling, batches, service
+from alphapulse.news_research.worker import recover_interrupted
+
+
+async def main():
+    url = urlparse(os.environ['DATABASE_URL'])
+    assert (url.username, url.password, url.path) == ('ci', 'ci-only', '/ci')
+    now = datetime.now(UTC)
+    anchor = now - timedelta(hours=1)
+    config = SimpleNamespace(start_at=now-timedelta(days=5), realtime_from=anchor)
+    async with SessionLocal() as s:
+        assert await s.scalar(select(func.count()).select_from(Package)) == 0
+        def original(key, published):
+            return NewsItem(source_code='wscn_news', external_id='priority-'+key, title=key, summary='',
+                            content_text=key, content_hash='a'*64, url='https://example.invalid/'+key, published_at=published)
+        old = original('old', now-timedelta(days=4))
+        live = original('live', now-timedelta(minutes=30))
+        late = original('late', now-timedelta(days=3))
+        s.add_all([old, live, late]); await s.commit()
+        batch = await batches.freeze(s, config)
+        packages = list(await s.scalars(select(Package)))
+        assert len(packages) == 2 and all(p.news_id != live.id for p in packages)
+        actions = await scheduling.realtime_actions(s, config, {('index', packages[0].id)}, 1)
+        live_package = await s.scalar(select(Package).where(Package.news_id == live.id))
+        assert actions == [('index', live_package.id)]
+        assert await scheduling.realtime_actions(s, config, set(), 0) == []
+        # Shrinking the history range must not exclude accumulated realtime news.
+        config.start_at = now-timedelta(minutes=5)
+        actions = await scheduling.realtime_actions(s, config, set(), 1)
+        assert actions == [('index', live_package.id)]
+        await batches.actions(s, batch, set(), 8, config)
+        assert not await batches.freeze(s, config)
+        config.start_at = now-timedelta(days=5)
+        assert await batches.freeze(s, config)
+        assert packages[0].keywords == []
+        # A realtime event's historical dependencies inherit priority.
+        event = Event(title='shared', tags=[], state='pending', revision=1, realtime_pending=True, due_at=now)
+        s.add(event); await s.flush()
+        for p in [packages[0], live_package]:
+            s.add(Membership(event_id=event.id, package_id=p.id, evidence='fixture', reason='fixture', change_kind='new_fact'))
+        packages[0].state = 'assign_pending'
+        live_package.state = 'done'
+        await s.commit()
+        config.start_at = now-timedelta(minutes=5)
+        assert await scheduling.realtime_actions(s, config, set(), 1) == []
+        config.start_at = now-timedelta(days=5)
+        assert await scheduling.realtime_actions(s, config, set(), 1) == [('assign', packages[0].id)]
+        packages[0].state = 'done'; await s.commit()
+        assert await scheduling.realtime_actions(s, config, set(), 1) == [('event', event.id)]
+        event.due_at = now+timedelta(minutes=5); await s.commit()
+        assert ('event', event.id) not in await scheduling.realtime_actions(s, config, set(), 8)
+        # Pausing invalidates a selected but not-yet-claimed task.
+        s.add(ResearchSettings(id=1, config={}, enabled=False, control_revision=9, start_at=config.start_at, realtime_from=anchor))
+        await s.commit()
+        assert await scheduling.control(s, {'_control_revision': 8}) is False
+        await s.commit()
+        with patch.object(service, 'complete', side_effect=AssertionError('Paused selection must not call a model')):
+            await service.process_package(live_package.id, {'_control_revision': 8})
+        assert await s.scalar(select(func.count()).select_from(Attempt)) == 0
+        live_package = await s.get(Package, live_package.id)
+        live_package.state = 'index_processing'
+        s.add(Attempt(package_id=live_package.id, content_hash='a'*64, stage='index', state='processing', input={}, rules={}))
+        event = await s.get(Event, event.id); event.state='processing'
+        s.add(Version(event_id=event.id, revision=event.revision, state='processing', input={}, rules={}))
+        await s.commit()
+        for duplicate in [
+            Attempt(package_id=live_package.id, content_hash='a'*64, stage='assign', state='processing', input={}, rules={}),
+            Version(event_id=event.id, revision=event.revision, state='processing', input={}, rules={}),
+        ]:
+            try:
+                async with s.begin_nested():
+                    s.add(duplicate)
+                    await s.flush()
+            except IntegrityError:
+                pass
+            else:
+                raise AssertionError('Concurrent duplicate claim was accepted')
+    await recover_interrupted()
+    async with SessionLocal() as s:
+        assert (await s.get(Package, live_package.id)).state == 'index_failed'
+        assert (await s.get(Event, event.id)).state == 'failed'
+        assert await scheduling.realtime_actions(s, config, set(), 8) == []
+        empty = Event(title='removed members', tags=[], state='pending', revision=1, realtime_pending=True, due_at=now)
+        s.add(empty); await s.commit()
+        assert ('event', empty.id) in await scheduling.realtime_actions(s, config, set(), 8)
+        with patch.object(service.Research, 'run', side_effect=AssertionError('Empty events must not call a model')):
+            await service.process_event(empty.id, {})
+        await s.refresh(empty)
+        assert empty.state == 'retired'
+        await s.execute(text('TRUNCATE news_research_settings, news_research_batches, news_research_packages, news_research_events, news_items CASCADE'))
+        await s.commit()
+    await engine.dispose()
+    print('Priority queue passed: live preemption, stable boundary, late history, range changes, dependencies, aggregation, stale control and uncertain-call recovery.')
+
+asyncio.run(main())