add governed source adapter discovery framework

apps/api/app/main.py

@@ -8,7 +8,7 @@ from urllib.parse import parse_qs, urlparse
 if __package__ in (None, ""):
     sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
     from app.domain import deduplication_key, deduplicate_businesses, is_suppressed, normalize_business, score_business, normalize_domain, normalize_phone, match_businesses
-    from app.sources import adapter_for, contains_secret
+    from app.sources import adapter_for, contains_secret, available_adapters, normalize_record, circuit_is_open
     from app.domain_intelligence import normalize_registrable_domain, resolve_domain, generate_candidate_domains
     from app.website_scanner import scan_website, validate_url
     from app.contact_extractor import extract_contacts, MAX_HTML_BYTES, MAX_RESULTS
@@ -21,7 +21,7 @@ if __package__ in (None, ""):
     from app.provider_config import validate_payload as validate_remote_provider, encrypt as encrypt_provider_secret, decrypt as decrypt_provider_secret, safe_status as remote_provider_status, test_connectivity as test_remote_connectivity
 else:
     from .domain import deduplication_key, deduplicate_businesses, is_suppressed, normalize_business, score_business, normalize_domain, normalize_phone, match_businesses
-    from .sources import adapter_for, contains_secret
+    from .sources import adapter_for, contains_secret, available_adapters, normalize_record, circuit_is_open
     from .domain_intelligence import normalize_registrable_domain, resolve_domain, generate_candidate_domains
     from .website_scanner import scan_website, validate_url
     from .contact_extractor import extract_contacts, MAX_HTML_BYTES, MAX_RESULTS
@@ -75,6 +75,10 @@ def connect(db_path: str) -> sqlite3.Connection:
         "pipeline_entries": (("notes", "TEXT NOT NULL DEFAULT ''"), ("next_action", "TEXT NOT NULL DEFAULT ''"), ("follow_up_at", "TEXT"), ("actor_user_id", "INTEGER"), ("idempotency_key", "TEXT"), ("version", "INTEGER NOT NULL DEFAULT 1")),
         "interactions": (("outcome", "TEXT NOT NULL DEFAULT 'other'"), ("notes", "TEXT NOT NULL DEFAULT ''"), ("next_action", "TEXT NOT NULL DEFAULT ''"), ("follow_up_at", "TEXT"), ("actor_user_id", "INTEGER"), ("idempotency_key", "TEXT")),
         "suppressions": (("active", "INTEGER NOT NULL DEFAULT 1"), ("updated_at", "TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP"), ("actor_user_id", "INTEGER")),
+        "sources": (("source_code", "TEXT NOT NULL DEFAULT ''"), ("display_name", "TEXT NOT NULL DEFAULT ''"), ("approved", "INTEGER NOT NULL DEFAULT 0"), ("policy_json", "TEXT NOT NULL DEFAULT '{}'"), ("quota_json", "TEXT NOT NULL DEFAULT '{}'")),
+        "discovery_queries": (("selected_adapters_json", "TEXT NOT NULL DEFAULT '[]'"), ("location", "TEXT NOT NULL DEFAULT ''"), ("category", "TEXT NOT NULL DEFAULT ''"), ("max_records", "INTEGER NOT NULL DEFAULT 100"), ("daily_limit", "INTEGER NOT NULL DEFAULT 1000"), ("schedule", "TEXT NOT NULL DEFAULT ''"), ("dry_run", "INTEGER NOT NULL DEFAULT 0"), ("lifecycle", "TEXT NOT NULL DEFAULT 'draft'")),
+        "discovery_runs": (("selected_adapters_json", "TEXT NOT NULL DEFAULT '[]'"), ("location", "TEXT NOT NULL DEFAULT ''"), ("category", "TEXT NOT NULL DEFAULT ''"), ("max_records", "INTEGER NOT NULL DEFAULT 100"), ("daily_limit", "INTEGER NOT NULL DEFAULT 1000"), ("schedule", "TEXT NOT NULL DEFAULT ''"), ("dry_run", "INTEGER NOT NULL DEFAULT 0"), ("lifecycle", "TEXT NOT NULL DEFAULT 'draft'"), ("paused_at", "TEXT")),
+        "source_records": (("discovery_run_id", "INTEGER"), ("normalized_key", "TEXT NOT NULL DEFAULT ''"), ("provenance_json", "TEXT NOT NULL DEFAULT '{}'")),
     }.items():
         existing = {r[1] for r in db.execute(f"PRAGMA table_info({table})")}
         for col, definition in additions:
@@ -631,7 +635,13 @@ class ApiHandler(BaseHTTPRequestHandler):
             if path=="/api/v1/businesses": return self.list_businesses(db,org,parse_qs(parsed.query))
             if path=="/api/v1/merge-history": return self.list_merge_history(db,org)
             if path=="/api/v1/sources": return self.list_sources(db,org)
+            if path=="/api/v1/sources/adapters": return self.send_json(200,{"items":available_adapters()})
             if path=="/api/v1/discovery-queries": return self.list_queries(db,org)
+            if path.startswith("/api/v1/discovery-runs/"):
+                bits=path.split("/")
+                if len(bits)==5 and bits[4].isdigit():
+                    run=db.execute("SELECT * FROM discovery_runs WHERE id=? AND organization_id=?",(int(bits[4]),org)).fetchone()
+                    return self.send_json(200,self._discovery_run_json(run)) if run else self.send_json(404,{"error":"not_found"})
             if path=="/api/v1/discovery-runs": return self.list_discovery_runs(db,org,parse_qs(parsed.query))
             if path=="/api/v1/discovery/provider-status": return self.send_json(200, ai_research_provider_status())
             if path=="/api/v1/discovery/ai-provider-status": return self.send_json(200, ai_research_provider_status())
@@ -652,6 +662,9 @@ class ApiHandler(BaseHTTPRequestHandler):
             if path=="/api/v1/outreach/provider-config": return self.provider_config(db,user)
             if path in ("/api/v1/reports/pipeline","/api/v1/reports/outcomes","/api/v1/reports/activity"): return self.report(db,org,path.rsplit('/',1)[1],parse_qs(parsed.query))
             if path.startswith("/api/v1/jobs/"): return self.get_job_route(db,org,path,parse_qs(parsed.query))
+            if path.startswith("/api/v1/sources/"):
+                bits=path.split("/")
+                if len(bits)==6 and bits[4].isdigit() and bits[5] == "health": return self.source_health(int(bits[4]),db,user)
             if path.startswith("/api/v1/businesses/"):
                 bits=path.split("/"); ident=bits[4] if len(bits)>4 else ""
                 if not ident.isdigit(): return self.send_json(404,{"error":"not_found"})
@@ -833,7 +846,7 @@ class ApiHandler(BaseHTTPRequestHandler):
 
     def _discovery_run_json(self, row):
         item = row_json(row)
-        for field, default in (("criteria_json", {}), ("seed_urls_json", []), ("result_json", {})):
+        for field, default in (("criteria_json", {}), ("seed_urls_json", []), ("result_json", {}), ("selected_adapters_json", [])):
             try: item[field[:-5]] = json.loads(item.pop(field) or json.dumps(default))
             except (TypeError, ValueError): item[field[:-5]] = default
         return item
@@ -846,9 +859,11 @@ class ApiHandler(BaseHTTPRequestHandler):
 
     def create_scoped_discovery(self, payload, db, user):
         criteria = payload.get("criteria", {}); seeds = payload.get("seed_urls")
-        criteria_only = seeds is None
+        selected_adapters = payload.get("selected_adapters", payload.get("sources", []))
+        source_mode = bool(selected_adapters) and seeds is None
+        criteria_only = seeds is None and not source_mode
         if not isinstance(criteria, dict): return self.send_json(400, {"error": "invalid_criteria"})
-        if not criteria_only and (not isinstance(seeds, list) or not seeds): return self.send_json(400, {"error": "seed_urls_required"})
+        if not criteria_only and not source_mode and (not isinstance(seeds, list) or not seeds): return self.send_json(400, {"error": "seed_urls_required"})
         if criteria_only:
             try: validate_ai_research_criteria(criteria)
             except AIResearchConfigError as exc:
@@ -859,26 +874,26 @@ class ApiHandler(BaseHTTPRequestHandler):
             legacy = search_provider_status()
             if status["status"] != "ready" and legacy["status"] != "ready": return self.send_json(503, {"error": status["status"], "provider": status["provider"]})
         try:
-            if not criteria_only and len(seeds) > 5: raise ValueError("invalid_criteria")
+            if not criteria_only and not source_mode and len(seeds) > 5: raise ValueError("invalid_criteria")
             if len(json.dumps(criteria).encode()) > 8192: raise ValueError("invalid_criteria")
             if not isinstance(criteria.get("keywords", criteria.get("keyword", [])), (list, str)): raise ValueError("invalid_criteria")
             max_pages = int(payload.get("max_pages", 20)); max_candidates = int(payload.get("max_candidates", 50))
             if not 1 <= max_pages <= 20 or not 1 <= max_candidates <= 50: raise ValueError("invalid_limits")
         except (ValueError, TypeError) as exc: return self.send_json(400, {"error": str(exc) or "invalid_criteria"})
-        if not criteria_only:
+        if not criteria_only and not source_mode:
             try:
                 for url in seeds: validate_url(url)
             except (ValueError, TypeError):
                 return self.send_json(400, {"error": "unsafe_seed_url"})
         key = str(payload.get("idempotency_key", "")).strip()
         if not key or len(key) > 200: return self.send_json(400, {"error": "invalid_idempotency_key"})
-        job_payload = {"criteria": criteria, "seed_urls": seeds if not criteria_only else None, "max_pages": max_pages, "max_candidates": max_candidates}
+        job_payload = {"criteria": criteria, "seed_urls": seeds if not criteria_only else None, "selected_adapters": selected_adapters, "location": payload.get("location", ""), "category": payload.get("category", ""), "max_records": payload.get("max_records", 100), "daily_limit": payload.get("daily_limit", 1000), "dry_run": bool(payload.get("dry_run", False)), "max_pages": max_pages, "max_candidates": max_candidates}
-        result = self.create_job({"type": "scoped_discovery", "payload": job_payload, "idempotency_key": key, "_accepted": True, "_defer_wakeup": True}, db, user)
+        result = self.create_job({"type": "source_discovery" if source_mode else "scoped_discovery", "payload": job_payload, "idempotency_key": key, "_accepted": True, "_defer_wakeup": True}, db, user)
         # create_job has already committed; read its id from the response is not available,
         # so resolve by the tenant-scoped idempotency key.
         job = db.execute("SELECT * FROM jobs WHERE organization_id=? AND idempotency_key=?", (user["organization_id"], key)).fetchone()
         if not db.execute("SELECT id FROM discovery_runs WHERE organization_id=? AND job_id=?", (user["organization_id"], job["id"])).fetchone():
-            db.execute("INSERT INTO discovery_runs(organization_id,job_id,criteria_json,seed_urls_json) VALUES(?,?,?,?)", (user["organization_id"], job["id"], json.dumps(criteria, sort_keys=True), json.dumps(seeds if not criteria_only else [])))
+            db.execute("INSERT INTO discovery_runs(organization_id,job_id,selected_adapters_json,location,category,max_records,daily_limit,schedule,dry_run,lifecycle,criteria_json,seed_urls_json) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)", (user["organization_id"], job["id"], json.dumps(selected_adapters), str(payload.get("location", "")), str(payload.get("category", "")), int(payload.get("max_records", 100)), int(payload.get("daily_limit", 1000)), str(payload.get("schedule", "")), int(bool(payload.get("dry_run", False))), "queued", json.dumps(criteria, sort_keys=True), json.dumps(seeds if not criteria_only else [])))
             self.audit(db, user, "discovery.created", str(job["id"])); db.commit()
         getattr(self.server, "job_wakeup", threading.Event()).set()
         return result
@@ -1080,6 +1095,13 @@ class ApiHandler(BaseHTTPRequestHandler):
             if len(bits_outreach)==7 and bits_outreach[:4]==["","api","v1","outreach"] and bits_outreach[4]=="drafts" and bits_outreach[5].isdigit() and bits_outreach[6] in {"approve","send"}: return self.approve_outreach_draft(int(bits_outreach[5]),db,user) if bits_outreach[6]=="approve" else self.send_outreach_draft(int(bits_outreach[5]),db,user)
             if path=="/api/v1/sources":return self.create_source(payload,db,user)
             if path=="/api/v1/discovery":return self.create_scoped_discovery(payload,db,user)
+            if path.startswith("/api/v1/discovery-runs/"):
+                bits=path.split("/")
+                if len(bits)==6 and bits[4].isdigit() and bits[5] in {"pause","resume","cancel"}: return self.discovery_run_action(int(bits[4]),bits[5],db,user)
+            if path=="/api/v1/discovery-runs":
+                payload.setdefault('criteria',{})
+                payload.setdefault('idempotency_key', 'discovery-run-' + hashlib.sha256(json.dumps(payload,sort_keys=True).encode()).hexdigest()[:24])
+                return self.create_scoped_discovery(payload,db,user)
             if path=="/api/v1/discovery-queries":return self.create_query(payload,db,user)
             if path=="/api/v1/suppressions":return self.create_suppression(payload,db,user)
             if path=="/api/v1/suppressions/import":return self.import_suppressions(payload,db,user)
@@ -1288,7 +1310,7 @@ class ApiHandler(BaseHTTPRequestHandler):
             else:seen.add(key);accepted.append(b)
         return self.send_json(200,{"accepted":len(accepted),"duplicates":len(rows)-len(normalized)+len(normalized)-len(accepted)-suppressed,"suppressed":suppressed,"rows":accepted})
     def list_sources(self,db,org):
-        cols='id,organization_id,name,kind,enabled,health_status,consecutive_failures,circuit_open,last_success_at,last_failure_at,last_error,created_at,updated_at'
+        cols='id,organization_id,name,kind,source_code,display_name,enabled,approved,policy_json,quota_json,health_status,consecutive_failures,circuit_open,last_success_at,last_failure_at,last_error,created_at,updated_at'
         return self.send_json(200,{"organization_id":org,"items":[row_json(r) for r in db.execute(f"SELECT {cols} FROM sources WHERE organization_id=? ORDER BY id",(org,))]})
     def list_queries(self,db,org):
         return self.send_json(200,{"organization_id":org,"items":[row_json(r) for r in db.execute("SELECT * FROM discovery_queries WHERE organization_id=? ORDER BY id",(org,))]})
@@ -1306,24 +1328,30 @@ class ApiHandler(BaseHTTPRequestHandler):
             out.append(x)
         return self.send_json(200,{"organization_id":org,"items":out,"limit":limit,"offset":offset,"has_more":len(rows)>limit})
     def create_source(self,payload,db,user):
-        name=str(payload.get('name','')).strip(); kind=str(payload.get('kind','')).strip().lower(); config=payload.get('config',{})
+        name=str(payload.get('name','')).strip(); kind=str(payload.get('source_code',payload.get('kind',''))).strip().lower(); config=payload.get('config',{})
-        if not name or kind not in ('csv','manual') or not isinstance(config,dict):return self.send_json(400,{"error":"invalid_source"})
+        optional = kind not in ('csv','manual')
+        if not name or kind not in ('csv','manual','google_places','bing_local','approved_directory','public_website','permitted_social','ct_logs','dns','rdap') or not isinstance(config,dict):return self.send_json(400,{"error":"invalid_source"})
         if contains_secret(config):return self.send_json(400,{"error":"secret_not_permitted"})
         try:
-            validation=adapter_for(kind).validate(config)
+            validation=adapter_for(kind).validate_config(config)
-            if config and not validation.valid:return self.send_json(400,{"error":"invalid_source_config","details":validation.errors})
+            if (config or optional) and not validation.valid:return self.send_json(400,{"error":"invalid_source_config","details":validation.errors})
-            cur=db.execute("INSERT INTO sources(organization_id,name,kind,enabled,config_json) VALUES(?,?,?,?,?)",(user['organization_id'],name,kind,int(bool(payload.get('enabled',False))),json.dumps(config,sort_keys=True)))
+            cur=db.execute("INSERT INTO sources(organization_id,name,kind,source_code,display_name,enabled,approved,config_json,policy_json,quota_json) VALUES(?,?,?,?,?,?,?,?,?,?)",(user['organization_id'],name,kind,kind,str(payload.get('display_name') or adapter_for(kind).display_name),int(bool(payload.get('enabled',False))),int(bool(payload.get('approved',config.get('approved',False)))),json.dumps(config,sort_keys=True),json.dumps(payload.get('policy',{}),sort_keys=True),json.dumps(payload.get('quota',{}),sort_keys=True)))
         except sqlite3.IntegrityError:return self.send_json(409,{"error":"duplicate_source"})
-        self.audit(db,user,'source.created',str(cur.lastrowid));db.commit();return self.send_json(201,row_json(db.execute("SELECT id,organization_id,name,kind,enabled,health_status,consecutive_failures,circuit_open,last_success_at,last_failure_at,last_error,created_at,updated_at FROM sources WHERE id=?",(cur.lastrowid,)).fetchone()))
+        self.audit(db,user,'source.created',str(cur.lastrowid));db.commit();return self.send_json(201,row_json(db.execute("SELECT id,organization_id,name,kind,source_code,display_name,enabled,approved,policy_json,quota_json,health_status,consecutive_failures,circuit_open,last_success_at,last_failure_at,last_error,created_at,updated_at FROM sources WHERE id=?",(cur.lastrowid,)).fetchone()))
     def update_source(self,sid,payload,db,user):
         if not db.execute("SELECT id FROM sources WHERE id=? AND organization_id=?",(sid,user['organization_id'])).fetchone():return self.send_json(404,{"error":"not_found"})
         if 'enabled' not in payload:return self.send_json(400,{"error":"enabled_required"})
         value=int(bool(payload['enabled']));db.execute("UPDATE sources SET enabled=?,updated_at=CURRENT_TIMESTAMP WHERE id=?",(value,sid));self.audit(db,user,'source.enabled' if value else 'source.disabled',str(sid));db.commit();return self.send_json(200,row_json(db.execute("SELECT * FROM sources WHERE id=?",(sid,)).fetchone()))
     def create_query(self,payload,db,user):
         sid=payload.get('source_id');name=str(payload.get('name','')).strip();query=payload.get('query',{})
+        selected=payload.get('selected_adapters',payload.get('sources',[])); location=str(payload.get('location','')).strip(); category=str(payload.get('category','')).strip(); schedule=str(payload.get('schedule','')).strip()
+        try: max_records=int(payload.get('max_records',100)); daily_limit=int(payload.get('daily_limit',1000))
+        except (TypeError,ValueError): return self.send_json(400,{"error":"invalid_limits"})
+        if not isinstance(selected,list) or any(str(x) not in {a["source_code"] for a in available_adapters()} for x in selected): return self.send_json(400,{"error":"invalid_adapters"})
+        if max_records<1 or max_records>10000 or daily_limit<1 or daily_limit>100000: return self.send_json(400,{"error":"invalid_limits"})
         if not isinstance(sid,int) or not name or not isinstance(query,dict) or contains_secret(query):return self.send_json(400,{"error":"invalid_query"})
         if not db.execute("SELECT id FROM sources WHERE id=? AND organization_id=?",(sid,user['organization_id'])).fetchone():return self.send_json(404,{"error":"not_found"})
-        try:cur=db.execute("INSERT INTO discovery_queries(organization_id,source_id,name,query_json) VALUES(?,?,?,?)",(user['organization_id'],sid,name,json.dumps(query,sort_keys=True)))
+        try:cur=db.execute("INSERT INTO discovery_queries(organization_id,source_id,name,query_json,selected_adapters_json,location,category,max_records,daily_limit,schedule,dry_run,lifecycle) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)",(user['organization_id'],sid,name,json.dumps(query,sort_keys=True),json.dumps(selected),location,category,max_records,daily_limit,schedule,int(bool(payload.get('dry_run',False))),str(payload.get('lifecycle','draft'))))
         except sqlite3.IntegrityError:return self.send_json(409,{"error":"duplicate_query"})
         self.audit(db,user,'discovery_query.created',str(cur.lastrowid));db.commit();return self.send_json(201,row_json(db.execute("SELECT * FROM discovery_queries WHERE id=?",(cur.lastrowid,)).fetchone()))
     def run_query(self,qid,db,user):
@@ -1350,10 +1378,27 @@ class ApiHandler(BaseHTTPRequestHandler):
         except (ValueError,KeyError) as exc:return self.send_json(400,{"error":"invalid_ingest","detail":str(exc)})
         inserted=0
         for record in page.records[:1000]:
-            raw=json.dumps(record,sort_keys=True,separators=(',',':'));digest=hashlib.sha256(raw.encode()).hexdigest()
+            raw=json.dumps(record,sort_keys=True,separators=(',',':'));digest=hashlib.sha256(raw.encode()).hexdigest(); normalized_key=hashlib.sha256(json.dumps(normalize_record(record),sort_keys=True,separators=(',',':')).encode()).hexdigest()
-            try:db.execute("INSERT INTO source_records(organization_id,source_id,content_hash,raw_json,normalized_json,source_url,query_context_json,cursor_json,rate_policy_json) VALUES(?,?,?,?,?,?,?,?,?)",(user['organization_id'],sid,digest,raw,raw,str(payload.get('source_url','')),json.dumps(payload.get('query_context',{}),sort_keys=True),json.dumps(payload.get('cursor',{}),sort_keys=True),json.dumps(payload.get('rate_policy',{}),sort_keys=True)));inserted+=1
+            try:
+                db.execute("INSERT INTO source_records(organization_id,source_id,content_hash,raw_json,normalized_json,normalized_key,source_url,provenance_json,query_context_json,cursor_json,rate_policy_json) VALUES(?,?,?,?,?,?,?,?,?,?,?)",(user['organization_id'],sid,digest,raw,json.dumps(normalize_record(record),sort_keys=True),normalized_key,str(payload.get('source_url','')),json.dumps({'adapter':source['kind']},sort_keys=True),json.dumps(payload.get('query_context',{}),sort_keys=True),json.dumps(payload.get('cursor',{}),sort_keys=True),json.dumps(payload.get('rate_policy',{}),sort_keys=True)))
+                record_id=db.execute("SELECT last_insert_rowid()").fetchone()[0]; db.execute("INSERT OR IGNORE INTO enrichment_queue(organization_id,source_record_id) VALUES(?,?)",(user['organization_id'],record_id)); inserted+=1
             except sqlite3.IntegrityError:pass
         db.execute("UPDATE sources SET health_status='healthy',consecutive_failures=0,last_success_at=CURRENT_TIMESTAMP,last_error=NULL WHERE id=?",(sid,));self.audit(db,user,'source.ingested',f'{sid}:{inserted}');db.commit();return self.send_json(201 if inserted else 200,{"inserted":inserted,"records":len(page.records)})
+    def source_health(self,sid,db,user):
+        source=db.execute("SELECT * FROM sources WHERE id=? AND organization_id=?",(sid,user["organization_id"])).fetchone()
+        if not source:return self.send_json(404,{"error":"not_found"})
+        try: config=json.loads(source["config_json"] or "{}")
+        except (TypeError,ValueError): config={}
+        health=adapter_for(source["kind"]).health_check(config)
+        return self.send_json(200,{"id":sid,"source_code":source["source_code"] or source["kind"],"display_name":source["display_name"] or source["name"],"status":source["health_status"],"configured":health.status=="healthy","circuit_open":bool(source["circuit_open"]),"consecutive_failures":source["consecutive_failures"],"last_error":source["last_error"]})
+    def discovery_run_action(self,rid,action,db,user):
+        run=db.execute("SELECT * FROM discovery_runs WHERE id=? AND organization_id=?",(rid,user["organization_id"])).fetchone()
+        if not run:return self.send_json(404,{"error":"not_found"})
+        lifecycle={"pause":"paused","resume":"queued","cancel":"cancelled"}[action]
+        db.execute("UPDATE discovery_runs SET lifecycle=?,paused_at=CASE WHEN ?='paused' THEN CURRENT_TIMESTAMP ELSE paused_at END,updated_at=CURRENT_TIMESTAMP WHERE id=? AND organization_id=?",(lifecycle,lifecycle,rid,user["organization_id"]))
+        if action=="cancel": db.execute("UPDATE jobs SET status='cancelled',completed_at=CURRENT_TIMESTAMP,updated_at=CURRENT_TIMESTAMP WHERE id=? AND organization_id=? AND status IN ('queued','running')",(run["job_id"],user["organization_id"]))
+        self.audit(db,user,"discovery."+action,str(rid)); db.commit()
+        return self.send_json(200,self._discovery_run_json(db.execute("SELECT * FROM discovery_runs WHERE id=?",(rid,)).fetchone()))
     def matches(self,bid,db,org):
         source=self.business(db,bid,org)
         if not source:return self.send_json(404,{"error":"not_found"})
@@ -1440,6 +1485,24 @@ def _run_scoped_discovery(db, job, handler):
     handler.add_job_event(db, job["id"], org, "discovery.completed", f"Persisted {len(persisted)} candidates", 100)
 
 
+def _run_source_discovery(db, job, handler):
+    payload=json.loads(job["payload"] or "{}"); org=job["organization_id"]; run=db.execute("SELECT * FROM discovery_runs WHERE organization_id=? AND job_id=?",(org,job["id"])).fetchone(); selected=payload.get("selected_adapters") or []
+    sources=db.execute("SELECT * FROM sources WHERE organization_id=? AND enabled=1 AND (source_code IN ("+(','.join('?'*len(selected)) or "NULL")+") OR kind IN ("+(','.join('?'*len(selected)) or "NULL")+"))",[org]+list(selected)+list(selected)).fetchall() if selected else []
+    total=0
+    for source in sources:
+        try: config=json.loads(source["config_json"] or "{}"); page=adapter_for(source["kind"]).discover(config)
+        except Exception as exc:
+            handler.add_job_event(db,job["id"],org,"source.blocked",f"{source['kind']} unavailable",0,"SOURCE_NOT_CONFIGURED"); continue
+        for record in page.records[:max(0,int(payload.get("max_records",100))-total)]:
+            raw=json.dumps(record,sort_keys=True,separators=(",",":")); digest=hashlib.sha256(raw.encode()).hexdigest(); norm=json.dumps(normalize_record(record),sort_keys=True); nkey=hashlib.sha256(norm.encode()).hexdigest()
+            try:
+                cur=db.execute("INSERT INTO source_records(organization_id,source_id,discovery_run_id,content_hash,raw_json,normalized_json,normalized_key,source_url,provenance_json) VALUES(?,?,?,?,?,?,?,?,?)",(org,source["id"],run["id"] if run else None,digest,raw,norm,nkey,str(config.get("source_url","")),json.dumps({"adapter":source["kind"]})))
+                db.execute("INSERT OR IGNORE INTO enrichment_queue(organization_id,source_record_id) VALUES(?,?)",(org,cur.lastrowid)); total+=1
+            except sqlite3.IntegrityError: pass
+    if run: db.execute("UPDATE discovery_runs SET result_count=?,lifecycle='succeeded',updated_at=CURRENT_TIMESTAMP WHERE id=?",(total,run["id"]))
+    handler.add_job_event(db,job["id"],org,"discovery.completed",f"Persisted {total} source records",100)
+
+
 def _job_worker(server):
     while not server.job_stop.is_set():
         db=connect(server.db_path)
@@ -1460,6 +1523,13 @@ def _job_worker(server):
                 except Exception as exc:
                     db.execute("UPDATE jobs SET status='failed',error_code=?,completed_at=CURRENT_TIMESTAMP,updated_at=CURRENT_TIMESTAMP WHERE id=?", (str(exc)[:80] or "DISCOVERY_FAILED", jid)); server_handler.add_job_event(db,jid,org,"failed","Discovery failed",job["progress"],str(exc)[:80]); db.commit()
                 continue
+            if job["type"] == "source_discovery":
+                try:
+                    _run_source_discovery(db, job, server_handler)
+                    db.execute("UPDATE jobs SET status='succeeded',progress=100,completed_at=CURRENT_TIMESTAMP,updated_at=CURRENT_TIMESTAMP WHERE id=?", (jid,)); db.commit()
+                except Exception as exc:
+                    db.execute("UPDATE jobs SET status='failed',error_code=?,completed_at=CURRENT_TIMESTAMP,updated_at=CURRENT_TIMESTAMP WHERE id=?", (str(exc)[:80] or "SOURCE_DISCOVERY_FAILED", jid)); server_handler.add_job_event(db,jid,org,"failed","Source discovery failed",job["progress"],str(exc)[:80]); db.commit()
+                continue
             try: steps=1 if job["type"]=="noop" else max(1,min(int(payload.get("steps",5)),20))
             except (ValueError,TypeError): steps=5
             cancelled=False

apps/api/app/sources.py

@@ -1,10 +1,17 @@
-"""Deterministic, network-free discovery source contracts and adapters."""
+"""Safe, tenant-neutral source adapter contracts.
+
+Network adapters are intentionally capability gated: configuration must explicitly
+approve public access, terms, rate limits and credentials (where required).
+Adapters never emit or persist credential values.
+"""
 from __future__ import annotations
 from dataclasses import dataclass, field
 from typing import Any, Mapping, Protocol, Sequence
-import csv, io, re
+import csv, io, random, time
 
 SECRET_KEYS = {"password", "passwd", "secret", "token", "api_key", "apikey", "authorization", "private_key", "credential"}
+NETWORK_KINDS = {"google_places", "bing_local", "approved_directory", "public_website", "permitted_social", "ct_logs", "dns", "rdap"}
+
 
 def contains_secret(value: Any, path: str = "") -> str | None:
     if isinstance(value, Mapping):
@@ -38,57 +45,140 @@ class SourceHealth:
     circuit_open: bool = False
     last_error: str | None = None
 
+@dataclass(frozen=True)
+class NormalizedRecord:
+    name: str = ""
+    website: str = ""
+    email: str = ""
+    phone: str = ""
+    description: str = ""
+    location: str = ""
+    source_url: str = ""
+    provenance: str = ""
+    raw: Mapping[str, Any] = field(default_factory=dict)
+
 class DiscoverySource(Protocol):
-    kind: str
+    source_code: str
-    def validate(self, config: Mapping[str, Any]) -> ValidationResult: ...
+    display_name: str
+    def validate_config(self, config: Mapping[str, Any]) -> ValidationResult: ...
     def discover(self, config: Mapping[str, Any], cursor: str | None = None) -> DiscoveryPage: ...
+    def health_check(self, config: Mapping[str, Any]) -> SourceHealth: ...
 
-_FIELDS = ("name", "website", "email", "phone", "description")
+_FIELDS = ("name", "website", "email", "phone", "description", "location")
-
 def normalize_record(row: Mapping[str, Any]) -> dict[str, str]:
-    result = {field: str(row.get(field, "")).strip() for field in _FIELDS}
+    result = {field: str(row.get(field, "") or "").strip() for field in _FIELDS}
-    # Accept common CSV spellings without retaining arbitrary sensitive fields.
+    aliases = {"company":"name", "business":"name", "url":"website", "domain":"website", "address":"location", "address_line":"location"}
-    aliases = {"company": "name", "url": "website", "domain": "website"}
     for key, target in aliases.items():
         if not result[target] and row.get(key) is not None: result[target] = str(row[key]).strip()
     return result
 
-class ManualSource:
+def normalized_record(row: Mapping[str, Any], *, source_url="", provenance="") -> NormalizedRecord:
-    kind = "manual"
+    x = normalize_record(row)
-    def validate(self, config: Mapping[str, Any]) -> ValidationResult:
+    return NormalizedRecord(**x, source_url=source_url, provenance=provenance, raw=dict(row))
-        if not isinstance(config, Mapping): return ValidationResult(False, ["config must be an object"])
-        found = contains_secret(config)
-        if found: return ValidationResult(False, [f"secret field is not permitted: {found}"])
-        rows = config.get("rows")
-        if not isinstance(rows, list): return ValidationResult(False, ["rows must be a list"])
-        return ValidationResult(True)
-    def discover(self, config: Mapping[str, Any], cursor: str | None = None) -> DiscoveryPage:
-        validation = self.validate(config)
-        if not validation.valid: raise ValueError(validation.errors[0])
-        rows = [normalize_record(row) for row in config["rows"] if isinstance(row, Mapping)]
-        return DiscoveryPage(rows, None, {"adapter": self.kind})
 
-class CsvSource:
+def exponential_backoff(attempt: int, base: float = 0.5, maximum: float = 30.0, jitter: float = 0.25) -> float:
-    kind = "csv"
+    """Bounded exponential delay with symmetric jitter; no sleeping occurs here."""
-    def validate(self, config: Mapping[str, Any]) -> ValidationResult:
+    delay = min(maximum, base * (2 ** max(0, int(attempt))))
+    return max(0.0, delay + random.uniform(-jitter * delay, jitter * delay))
+
+def backoff_delay(attempt: int, base: float = 0.5, maximum: float = 30.0, jitter: float = 0.25) -> float:
+    return exponential_backoff(attempt, base, maximum, jitter)
+
+def circuit_is_open(consecutive_failures: int, threshold: int = 3) -> bool:
+    return int(consecutive_failures) >= threshold
+
+def quota_allowed(used: int, limit: int | None) -> bool:
+    return limit is None or (limit >= 0 and used < limit)
+
+def rate_limit_delay(last_request: float | None, min_interval: float) -> float:
+    if last_request is None: return 0.0
+    return max(0.0, float(min_interval) - (time.monotonic() - last_request))
+
+class _Base:
+    kind = ""
+    source_code = ""
+    display_name = ""
+    def validate_config(self, config):
         if not isinstance(config, Mapping): return ValidationResult(False, ["config must be an object"])
         found = contains_secret(config)
         if found: return ValidationResult(False, [f"secret field is not permitted: {found}"])
+        return ValidationResult(True)
+    validate = validate_config
+    def health_check(self, config):
+        result = self.validate_config(config)
+        return SourceHealth("healthy" if result.valid else "unhealthy", last_error=None if result.valid else "; ".join(result.errors))
+    def discover(self, config, cursor=None):
+        result = self.validate_config(config)
+        if not result.valid: raise ValueError(result.errors[0])
+        raise RuntimeError("source_not_configured")
+
+class ManualSource(_Base):
+    kind = source_code = "manual"; display_name = "Manual records"
+    def validate_config(self, config):
+        result = super().validate_config(config)
+        if not result.valid: return result
+        if not isinstance(config.get("rows"), list): return ValidationResult(False, ["rows must be a list"])
+        return ValidationResult(True)
+    def discover(self, config, cursor=None):
+        result = self.validate_config(config)
+        if not result.valid: raise ValueError(result.errors[0])
+        return DiscoveryPage([normalize_record(row) for row in config["rows"] if isinstance(row, Mapping)], metadata={"adapter":self.source_code})
+
+class CsvSource(_Base):
+    kind = source_code = "csv"; display_name = "CSV import"
+    def validate_config(self, config):
+        result = super().validate_config(config)
+        if not result.valid: return result
         if not isinstance(config.get("csv"), str): return ValidationResult(False, ["csv must be text"])
         try:
-            reader = csv.DictReader(io.StringIO(config["csv"]));
+            reader = csv.DictReader(io.StringIO(config["csv"]))
             if not reader.fieldnames: return ValidationResult(False, ["CSV header is required"])
         except csv.Error as exc: return ValidationResult(False, [f"invalid CSV: {exc}"])
         return ValidationResult(True)
-    def discover(self, config: Mapping[str, Any], cursor: str | None = None) -> DiscoveryPage:
+    def discover(self, config, cursor=None):
-        validation = self.validate(config)
+        result = self.validate_config(config)
-        if not validation.valid: raise ValueError(validation.errors[0])
+        if not result.valid: raise ValueError(result.errors[0])
         reader = csv.DictReader(io.StringIO(config["csv"].replace("\r\n", "\n")))
-        records = [normalize_record({str(k).strip().lower(): v for k, v in row.items()}) for row in reader]
+        return DiscoveryPage([normalize_record({str(k).strip().lower():v for k,v in row.items()}) for row in reader], metadata={"adapter":self.source_code,"columns":reader.fieldnames or []})
-        return DiscoveryPage(records, None, {"adapter": self.kind, "columns": reader.fieldnames or []})
 
-ADAPTERS = {"manual": ManualSource, "csv": CsvSource}
+class GatedSource(_Base):
+    required = "approved"
+    def validate_config(self, config):
+        result = super().validate_config(config)
+        if not result.valid: return result
+        if config.get("approved") is not True: return ValidationResult(False, ["source approval is required"])
+        if config.get("public_access") is not True: return ValidationResult(False, ["public_access approval is required"])
+        if config.get("terms_accepted") is not True: return ValidationResult(False, ["terms_accepted is required"])
+        if self.source_code in {"google_places", "bing_local"} and not config.get("credential_ref"):
+            return ValidationResult(False, ["approved credential_ref is required"])
+        if not isinstance(config.get("rate_limit", 1), (int, float)) or config.get("rate_limit", 1) <= 0:
+            return ValidationResult(False, ["positive rate_limit is required"])
+        return ValidationResult(True)
+    def discover(self, config, cursor=None):
+        result = self.validate_config(config)
+        if not result.valid: raise ValueError(result.errors[0])
+        # Network execution is delegated to an explicitly approved provider; never guess.
+        raise RuntimeError("network_adapter_not_configured")
+
+
+def _gated(code, name):
+    return type(name.replace(" ", ""), (GatedSource,), {"kind":code, "source_code":code, "display_name":name})
+GooglePlacesSource = _gated("google_places", "Google Places")
+BingLocalSource = _gated("bing_local", "Bing / approved local API")
+ApprovedDirectorySource = _gated("approved_directory", "Approved directory")
+PublicWebsiteSource = _gated("public_website", "Public website")
+PermittedSocialSource = _gated("permitted_social", "Permitted social")
+CtLogsSource = _gated("ct_logs", "Certificate transparency logs")
+DnsSource = _gated("dns", "DNS")
+RdapSource = _gated("rdap", "RDAP")
+
+ADAPTERS = {x.source_code: x for x in (ManualSource, CsvSource, GooglePlacesSource, BingLocalSource, ApprovedDirectorySource, PublicWebsiteSource, PermittedSocialSource, CtLogsSource, DnsSource, RdapSource)}
+# common aliases used by clients
+ADAPTER_REGISTRY = ADAPTERS
 
 def adapter_for(kind: str) -> DiscoverySource:
-    try: return ADAPTERS[kind]()
+    try: return ADAPTERS[str(kind).strip().lower()]()
     except KeyError: raise ValueError("unsupported source kind")
+
+def available_adapters() -> list[dict[str, str]]:
+    return [{"source_code": cls.source_code, "display_name": cls.display_name} for cls in ADAPTERS.values()]

apps/api/schema.sql

@@ -136,8 +136,8 @@ CREATE INDEX IF NOT EXISTS idx_job_events_job_sequence ON job_events(job_id,sequ
 -- Phase 5 source framework (additive-safe; credentials contain metadata only).
 CREATE TABLE IF NOT EXISTS sources (
  id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id),
- name TEXT NOT NULL, kind TEXT NOT NULL CHECK(kind IN ('csv','manual')), enabled INTEGER NOT NULL DEFAULT 0,
+ name TEXT NOT NULL, kind TEXT NOT NULL CHECK(kind IN ('csv','manual','google_places','bing_local','approved_directory','public_website','permitted_social','ct_logs','dns','rdap')), source_code TEXT NOT NULL DEFAULT '', display_name TEXT NOT NULL DEFAULT '', enabled INTEGER NOT NULL DEFAULT 0, approved INTEGER NOT NULL DEFAULT 0,
- config_json TEXT NOT NULL DEFAULT '{}', health_status TEXT NOT NULL DEFAULT 'unknown',
+ config_json TEXT NOT NULL DEFAULT '{}', policy_json TEXT NOT NULL DEFAULT '{}', quota_json TEXT NOT NULL DEFAULT '{}', health_status TEXT NOT NULL DEFAULT 'unknown',
  consecutive_failures INTEGER NOT NULL DEFAULT 0, circuit_open INTEGER NOT NULL DEFAULT 0,
  last_success_at TEXT, last_failure_at TEXT, last_error TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE(organization_id,name)
@@ -151,18 +151,25 @@ CREATE TABLE IF NOT EXISTS source_credentials (
 );
 CREATE TABLE IF NOT EXISTS discovery_queries (
  id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id), source_id INTEGER NOT NULL REFERENCES sources(id) ON DELETE CASCADE,
- name TEXT NOT NULL, query_json TEXT NOT NULL DEFAULT '{}', enabled INTEGER NOT NULL DEFAULT 1, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ name TEXT NOT NULL, query_json TEXT NOT NULL DEFAULT '{}', enabled INTEGER NOT NULL DEFAULT 1, selected_adapters_json TEXT NOT NULL DEFAULT '[]', location TEXT NOT NULL DEFAULT '', category TEXT NOT NULL DEFAULT '', max_records INTEGER NOT NULL DEFAULT 100, daily_limit INTEGER NOT NULL DEFAULT 1000, schedule TEXT NOT NULL DEFAULT '', dry_run INTEGER NOT NULL DEFAULT 0, lifecycle TEXT NOT NULL DEFAULT 'draft', created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE(organization_id,name)
 );
 CREATE INDEX IF NOT EXISTS idx_discovery_queries_org ON discovery_queries(organization_id,id);
 CREATE TABLE IF NOT EXISTS source_records (
  id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id), source_id INTEGER NOT NULL REFERENCES sources(id) ON DELETE CASCADE,
- discovery_query_id INTEGER REFERENCES discovery_queries(id) ON DELETE SET NULL, content_hash TEXT NOT NULL, raw_json TEXT NOT NULL,
+ discovery_query_id INTEGER REFERENCES discovery_queries(id) ON DELETE SET NULL, discovery_run_id INTEGER REFERENCES discovery_runs(id) ON DELETE SET NULL, content_hash TEXT NOT NULL, raw_json TEXT NOT NULL,
- normalized_json TEXT NOT NULL, source_url TEXT NOT NULL DEFAULT '', query_context_json TEXT NOT NULL DEFAULT '{}', processing_status TEXT NOT NULL DEFAULT 'raw',
+ normalized_json TEXT NOT NULL, normalized_key TEXT NOT NULL DEFAULT '', source_url TEXT NOT NULL DEFAULT '', provenance_json TEXT NOT NULL DEFAULT '{}', query_context_json TEXT NOT NULL DEFAULT '{}', processing_status TEXT NOT NULL DEFAULT 'raw',
  cursor_json TEXT NOT NULL DEFAULT '{}', rate_policy_json TEXT NOT NULL DEFAULT '{}', created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
  UNIQUE(organization_id,source_id,content_hash)
 );
 CREATE INDEX IF NOT EXISTS idx_source_records_org ON source_records(organization_id,id DESC);
+CREATE TABLE IF NOT EXISTS enrichment_queue (
+ id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id) ON DELETE CASCADE, source_record_id INTEGER NOT NULL REFERENCES source_records(id) ON DELETE CASCADE, status TEXT NOT NULL DEFAULT 'queued', attempts INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE(organization_id,source_record_id)
+);
+CREATE TABLE IF NOT EXISTS source_events (
+ id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id) ON DELETE CASCADE, source_id INTEGER REFERENCES sources(id) ON DELETE SET NULL, discovery_run_id INTEGER REFERENCES discovery_runs(id) ON DELETE SET NULL, event_type TEXT NOT NULL, message TEXT NOT NULL DEFAULT '', metadata_json TEXT NOT NULL DEFAULT '{}', created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
+CREATE INDEX IF NOT EXISTS idx_source_events_org ON source_events(organization_id,created_at DESC,id DESC);
 
 -- Phase 7 domain intelligence (additive-safe; DNS results are explicitly stateful).
 CREATE TABLE IF NOT EXISTS domain_checks (
@@ -341,9 +348,8 @@ CREATE INDEX IF NOT EXISTS idx_outreach_drafts_business ON outreach_drafts(organ
 -- Built-in scoped discovery runs. Results are reviewable business records only;
 -- this feature never sends outreach.
 CREATE TABLE IF NOT EXISTS discovery_runs (
- id INTEGER PRIMARY KEY AUTOINCREMENT,
+ id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id), job_id INTEGER NOT NULL REFERENCES jobs(id) ON DELETE CASCADE,
- organization_id TEXT NOT NULL REFERENCES organizations(id) ON DELETE CASCADE,
+ selected_adapters_json TEXT NOT NULL DEFAULT '[]', location TEXT NOT NULL DEFAULT '', category TEXT NOT NULL DEFAULT '', max_records INTEGER NOT NULL DEFAULT 100, daily_limit INTEGER NOT NULL DEFAULT 1000, schedule TEXT NOT NULL DEFAULT '', dry_run INTEGER NOT NULL DEFAULT 0, lifecycle TEXT NOT NULL DEFAULT 'draft', paused_at TEXT,
- job_id INTEGER NOT NULL REFERENCES jobs(id) ON DELETE CASCADE,
  criteria_json TEXT NOT NULL DEFAULT '{}', seed_urls_json TEXT NOT NULL DEFAULT '[]',
  result_json TEXT NOT NULL DEFAULT '{}', result_count INTEGER NOT NULL DEFAULT 0,
  created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,