fix source discovery egress and failures

apps/api/app/main.py

@@ -1649,7 +1649,13 @@ def _run_source_discovery(db, job, handler):
                 selected=[linked["kind"]] if linked else []
     placeholders=','.join('?'*len(selected)) or "NULL"
     sources=db.execute("SELECT * FROM sources WHERE organization_id=? AND enabled=1 AND circuit_open=0 AND (source_code IN ("+placeholders+") OR kind IN ("+placeholders+"))",[org]+list(selected)+list(selected)).fetchall() if selected else []
-    total=0; max_records=max(0,int(payload.get("max_records", query["max_records"] if query else 100)))
+    if selected and not sources:
+        blocked=db.execute("SELECT source_code,kind,circuit_open,enabled FROM sources WHERE organization_id=? AND (source_code IN ("+placeholders+") OR kind IN ("+placeholders+"))",[org]+list(selected)+list(selected)).fetchall()
+        code="SOURCE_CIRCUIT_OPEN" if any(row["circuit_open"] for row in blocked) else "SOURCE_DISABLED"
+        handler.add_job_event(db,job["id"],org,"source.blocked","No selected source is eligible to run",10,code)
+        if run: db.execute("UPDATE discovery_runs SET lifecycle='failed',updated_at=CURRENT_TIMESTAMP WHERE id=?",(run["id"],))
+        raise RuntimeError(code)
+    total=0; blocked_count=0; max_records=max(0,int(payload.get("max_records", query["max_records"] if query else 100)))
     for source in sources:
         handler.add_job_event(db,job["id"],org,"source.started",f"Starting {source['display_name'] or source['kind']}",5)
         try:
@@ -1664,9 +1670,10 @@ def _run_source_discovery(db, job, handler):
                 handler.add_job_event(db,job["id"],org,"source.quota_exceeded",f"Quota reached for {source['kind']}",10,"SOURCE_QUOTA_EXCEEDED"); continue
             page=adapter_for(source["source_code"] or source["kind"]).discover(source_config_with_credentials(db, source, config))
         except Exception as exc:
-            failures=int(source["consecutive_failures"])+1
+            blocked_count+=1; failures=int(source["consecutive_failures"])+1
             db.execute("UPDATE sources SET health_status='unhealthy',consecutive_failures=?,circuit_open=?,last_failure_at=CURRENT_TIMESTAMP,last_error=?,updated_at=CURRENT_TIMESTAMP WHERE id=?",(failures,int(circuit_is_open(failures)),str(exc)[:300],source["id"]))
-            handler.add_job_event(db,job["id"],org,"source.blocked",f"{source['kind']} unavailable",10,"SOURCE_NOT_CONFIGURED"); continue
+            detail=str(exc).lower(); code="SOURCE_NETWORK_ERROR" if any(token in detail for token in ("urlopen", "gaierror", "timed out", "temporary failure", "network is unreachable")) else "SOURCE_EXECUTION_FAILED"
+            handler.add_job_event(db,job["id"],org,"source.blocked",f"{source['display_name'] or source['kind']} could not be reached",10,code); continue
         limit=min(max_records-total, per_run_limit, max(0,daily_limit-used_today))
         for record in page.records[:limit]:
             raw=json.dumps(record,sort_keys=True,separators=(",",":")); digest=hashlib.sha256(raw.encode()).hexdigest(); normalized=normalize_business(normalize_record(record)); norm=json.dumps(normalized,sort_keys=True); nkey=hashlib.sha256(norm.encode()).hexdigest()
@@ -1697,7 +1704,10 @@ def _run_source_discovery(db, job, handler):
                 if not db.execute("SELECT 1 FROM contacts WHERE organization_id=? AND business_id=? AND email=? AND phone=?",(org,bid,normalized["email"],normalized["phone"])).fetchone(): db.execute("INSERT INTO contacts(business_id,organization_id,email,phone) VALUES(?,?,?,?)",(bid,org,normalized["email"],normalized["phone"]))
             if record_id: db.execute("UPDATE source_records SET processing_status='processed',normalized_json=?,normalized_key=? WHERE id=? AND organization_id=?",(norm,nkey,record_id,org)); db.execute("INSERT OR IGNORE INTO enrichment_queue(organization_id,source_record_id,status) VALUES(?,?,?)",(org,record_id,"completed")); db.execute("UPDATE enrichment_queue SET status='completed',updated_at=CURRENT_TIMESTAMP WHERE source_record_id=? AND organization_id=?",(record_id,org))
             handler.add_job_event(db,job["id"],org,"enrichment.queued",f"Enrichment complete for business {bid}",75); handler.add_job_event(db,job["id"],org,"review.queued",f"Business {bid} queued for review",90)
-    if run: db.execute("UPDATE discovery_runs SET result_count=?,lifecycle='succeeded',updated_at=CURRENT_TIMESTAMP WHERE id=?",(total,run["id"]))
+    if blocked_count and total==0:
+        if run: db.execute("UPDATE discovery_runs SET lifecycle='failed',updated_at=CURRENT_TIMESTAMP WHERE id=?",(run["id"],))
+        raise RuntimeError("SOURCE_NETWORK_ERROR")
+    if run: db.execute("UPDATE discovery_runs SET result_count=?,lifecycle=?,updated_at=CURRENT_TIMESTAMP WHERE id=?",(total,'partial' if blocked_count else 'succeeded',run["id"]))
     handler.add_job_event(db,job["id"],org,"discovery.completed",f"Persisted {total} source records",100)
 
 

apps/api/tests/test_sources_phase5.py

@@ -113,6 +113,16 @@ class SourceApiTests(unittest.TestCase):
         self.assertEqual(status, 200)
         self.assertFalse(health['configured'])
 
+    def test_disabled_selected_source_fails_instead_of_succeeding_with_zero_records(self):
+        status, source=self.req('POST','/api/v1/sources',{'name':'Disabled manual source','kind':'manual','config':{'rows':[]}}); self.assertEqual(status,201)
+        status, query=self.req('POST','/api/v1/discovery-queries',{'source_id':source['id'],'name':'disabled-source-run','query':{},'selected_adapters':['manual']}); self.assertEqual(status,201)
+        status, job=self.req('POST',f"/api/v1/discovery-queries/{query['id']}/run",{}); self.assertEqual(status,202)
+        for _ in range(100):
+            _, current=self.req('GET',f"/api/v1/jobs/{job['id']}")
+            if current['status'] in ('succeeded','failed'): break
+            threading.Event().wait(.01)
+        self.assertEqual(current['status'],'failed'); self.assertEqual(current['error_code'],'SOURCE_DISABLED')
+
     def test_source_discovery_persists_pipeline_and_is_idempotent(self):
         status, source = self.req('POST', '/api/v1/sources', {
             'name': 'Manual leads', 'kind': 'manual', 'enabled': True,

docker-compose.yml

@@ -56,6 +56,7 @@ services:
     restart: unless-stopped
     networks:
       - prospect_internal
+      - source_egress
 
   web:
     build:
@@ -115,5 +116,6 @@ volumes:
 networks:
   prospect_internal:
     internal: true
+  source_egress:
   searxng_egress:
   web_ingress: