feat: three-phase batch job worker (submit → poll → harvest)

Generated with [Claude Code](https://claude.ai/code)
via [Happy](https://happy.engineering)

Co-Authored-By: Claude <noreply@anthropic.com>
Co-Authored-By: Happy <yesreply@happy.engineering>

app/worker.py

@@ -3,6 +3,7 @@ import asyncio
 import json
 import logging
 import os
+import time
 import traceback
 from app.database import Database
 from app.indicators.base import IndicatorRegistry
@@ -17,6 +18,9 @@ from app.core.email import send_completion_email
 
 logger = logging.getLogger(__name__)
 
+BATCH_POLL_INTERVAL = 30  # seconds between status checks
+BATCH_TIMEOUT = 1200  # 20 minutes maximum wait
+
 
 def _save_spatial_json(spatial, status_value: str, path: str) -> None:
     """Serialize spatial data to JSON for the frontend."""
@@ -51,19 +55,97 @@ async def process_job(job_id: str, db: Database, registry: IndicatorRegistry) ->
         return
     await db.update_job_status(job_id, JobStatus.PROCESSING)
     try:
-        # Track spatial data per indicator for map generation
         spatial_cache = {}
 
+        # Separate batch vs non-batch indicators
+        batch_indicators = {}
+        process_indicators = []
         for indicator_id in job.request.indicator_ids:
-            await db.update_job_progress(job_id, indicator_id, "processing")
             indicator = registry.get(indicator_id)
-            result = await indicator.process(
-                job.request.aoi,
-                job.request.time_range,
-                season_months=job.request.season_months(),
-            )
+            if indicator.uses_batch:
+                batch_indicators[indicator_id] = indicator
+            else:
+                process_indicators.append((indicator_id, indicator))
+
+        # -- Phase 1: Submit batch jobs --
+        batch_submissions = {}
+        fallback_ids = set()
+        for indicator_id, indicator in batch_indicators.items():
+            await db.update_job_progress(job_id, indicator_id, "submitting")
+            try:
+                jobs = await indicator.submit_batch(
+                    job.request.aoi,
+                    job.request.time_range,
+                    season_months=job.request.season_months(),
+                )
+                batch_submissions[indicator_id] = jobs
+                await db.update_job_progress(job_id, indicator_id, "processing on CDSE")
+            except Exception as exc:
+                logger.warning("Batch submit failed for %s, will use fallback: %s", indicator_id, exc)
+                fallback_ids.add(indicator_id)
+
+        # -- Phase 2: Poll until all batch jobs finish --
+        poll_start = time.monotonic()
+        pending = dict(batch_submissions)
+
+        while pending:
+            # Check current statuses before sleeping
+            for indicator_id in list(pending.keys()):
+                jobs = pending[indicator_id]
+                statuses = [j.status() for j in jobs]
+                if all(s == "finished" for s in statuses):
+                    logger.info("Batch jobs finished for %s", indicator_id)
+                    del pending[indicator_id]
+                elif any(s in ("error", "canceled") for s in statuses):
+                    logger.warning("Batch job failed for %s: %s", indicator_id, statuses)
+                    del pending[indicator_id]
+
+            if not pending:
+                break
+
+            elapsed = time.monotonic() - poll_start
+            if elapsed >= BATCH_TIMEOUT:
+                logger.warning("Batch poll timeout after %.0fs, remaining: %s", elapsed, list(pending.keys()))
+                fallback_ids.update(pending.keys())
+                break
+
+            await asyncio.sleep(BATCH_POLL_INTERVAL)
+
+        # -- Phase 3: Harvest batch results + process non-batch indicators --
+        for indicator_id in job.request.indicator_ids:
+            indicator = registry.get(indicator_id)
+
+            if indicator_id in fallback_ids:
+                await db.update_job_progress(job_id, indicator_id, "processing")
+                result = await indicator.process(
+                    job.request.aoi,
+                    job.request.time_range,
+                    season_months=job.request.season_months(),
+                )
+            elif indicator_id in batch_submissions:
+                await db.update_job_progress(job_id, indicator_id, "downloading")
+                try:
+                    result = await indicator.harvest(
+                        job.request.aoi,
+                        job.request.time_range,
+                        season_months=job.request.season_months(),
+                        batch_jobs=batch_submissions[indicator_id],
+                    )
+                except Exception as exc:
+                    logger.warning("Harvest failed for %s, using fallback: %s", indicator_id, exc)
+                    result = await indicator.process(
+                        job.request.aoi,
+                        job.request.time_range,
+                        season_months=job.request.season_months(),
+                    )
+            else:
+                await db.update_job_progress(job_id, indicator_id, "processing")
+                result = await indicator.process(
+                    job.request.aoi,
+                    job.request.time_range,
+                    season_months=job.request.season_months(),
+                )
 
-            # Capture spatial data before it's lost
             spatial = indicator.get_spatial_data()
             if spatial is not None:
                 spatial_cache[indicator_id] = spatial

tests/test_worker.py

@@ -1,5 +1,6 @@
 import pytest
 from datetime import date
+from unittest.mock import MagicMock
 from app.worker import process_job
 from app.database import Database
 from app.models import JobStatus, AOI, TimeRange, JobRequest, IndicatorResult, StatusLevel, TrendDirection, ConfidenceLevel
@@ -66,3 +67,112 @@ async def test_process_job_handles_unknown_indicator(temp_db_path):
     job = await db.get_job(job_id)
     assert job.status == JobStatus.FAILED
     assert "nonexistent" in job.error
+
+
+class MockBatchIndicator(BaseIndicator):
+    """Batch indicator for testing the three-phase worker."""
+    id = "ndvi"
+    name = "Vegetation (NDVI)"
+    category = "D2"
+    question = "Is vegetation cover declining?"
+    estimated_minutes = 8
+    uses_batch = True
+
+    async def process(self, aoi, time_range, season_months=None):
+        return IndicatorResult(
+            indicator_id="ndvi", headline="placeholder",
+            status=StatusLevel.GREEN, trend=TrendDirection.STABLE,
+            confidence=ConfidenceLevel.LOW, map_layer_path="",
+            chart_data={"dates": ["2025"], "values": [0.3], "label": "NDVI"},
+            data_source="placeholder",
+            summary="Fallback.", methodology="Placeholder.", limitations=[],
+        )
+
+    async def submit_batch(self, aoi, time_range, season_months=None):
+        mock_job = MagicMock()
+        mock_job.job_id = "j-test"
+        mock_job.status.return_value = "finished"
+        return [mock_job, mock_job, mock_job]
+
+    async def harvest(self, aoi, time_range, season_months=None, batch_jobs=None):
+        return IndicatorResult(
+            indicator_id="ndvi", headline="Real NDVI data",
+            status=StatusLevel.GREEN, trend=TrendDirection.STABLE,
+            confidence=ConfidenceLevel.HIGH, map_layer_path="",
+            chart_data={"dates": ["2025-01"], "values": [0.45], "label": "NDVI"},
+            data_source="satellite",
+            summary="Real.", methodology="Sentinel-2.", limitations=[],
+        )
+
+
+@pytest.mark.asyncio
+async def test_process_job_uses_batch_flow(temp_db_path):
+    """Worker uses submit_batch -> poll -> harvest for batch indicators."""
+    db = Database(temp_db_path)
+    await db.init()
+    reg = IndicatorRegistry()
+    reg.register(MockBatchIndicator())
+    request = JobRequest(
+        aoi=AOI(name="Test", bbox=[32.45, 15.65, 32.65, 15.80]),
+        time_range=TimeRange(start=date(2025, 3, 1), end=date(2026, 3, 1)),
+        indicator_ids=["ndvi"],
+        email="test@example.com",
+    )
+    job_id = await db.create_job(request)
+    await process_job(job_id, db, reg)
+    job = await db.get_job(job_id)
+    assert job.status == JobStatus.COMPLETE
+    assert len(job.results) == 1
+    assert job.results[0].data_source == "satellite"
+    assert job.results[0].headline == "Real NDVI data"
+
+
+@pytest.mark.asyncio
+async def test_process_job_mixes_batch_and_process(temp_db_path):
+    """Worker handles batch and non-batch indicators in the same job."""
+    db = Database(temp_db_path)
+    await db.init()
+    reg = IndicatorRegistry()
+    reg.register(MockBatchIndicator())
+    reg.register(MockFiresIndicator())
+    request = JobRequest(
+        aoi=AOI(name="Test", bbox=[32.45, 15.65, 32.65, 15.80]),
+        time_range=TimeRange(start=date(2025, 3, 1), end=date(2026, 3, 1)),
+        indicator_ids=["ndvi", "fires"],
+        email="test@example.com",
+    )
+    job_id = await db.create_job(request)
+    await process_job(job_id, db, reg)
+    job = await db.get_job(job_id)
+    assert job.status == JobStatus.COMPLETE
+    assert len(job.results) == 2
+
+    ndvi_result = next(r for r in job.results if r.indicator_id == "ndvi")
+    fires_result = next(r for r in job.results if r.indicator_id == "fires")
+    assert ndvi_result.data_source == "satellite"
+    assert fires_result.headline == "3 fire events detected"
+
+
+@pytest.mark.asyncio
+async def test_process_job_batch_submit_failure_falls_back(temp_db_path):
+    """If submit_batch() fails, worker falls back to process()."""
+
+    class FailingBatchIndicator(MockBatchIndicator):
+        async def submit_batch(self, aoi, time_range, season_months=None):
+            raise ConnectionError("CDSE unreachable")
+
+    db = Database(temp_db_path)
+    await db.init()
+    reg = IndicatorRegistry()
+    reg.register(FailingBatchIndicator())
+    request = JobRequest(
+        aoi=AOI(name="Test", bbox=[32.45, 15.65, 32.65, 15.80]),
+        time_range=TimeRange(start=date(2025, 3, 1), end=date(2026, 3, 1)),
+        indicator_ids=["ndvi"],
+        email="test@example.com",
+    )
+    job_id = await db.create_job(request)
+    await process_job(job_id, db, reg)
+    job = await db.get_job(job_id)
+    assert job.status == JobStatus.COMPLETE
+    assert job.results[0].data_source == "placeholder"