add governed source discovery system

complete source discovery processing lifecycle
add governed source adapter discovery framework
2026-09-03 23:04:37 +02:00 · 2026-09-03 23:01:25 +02:00 · 2026-09-03 22:49:05 +02:00 · 2026-09-03 22:48:28 +02:00
10 changed files with 429 additions and 93 deletions
@@ -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 '{}'"), ("response_metadata_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}
-        result = self.create_job({"type": "scoped_discovery", "payload": job_payload, "idempotency_key": key, "_accepted": True, "_defer_wakeup": True}, db, user)
+        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": "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,))]})
@@ -1300,35 +1322,49 @@ class ApiHandler(BaseHTTPRequestHandler):
        rows=db.execute("SELECT * FROM source_records WHERE organization_id=? ORDER BY id DESC LIMIT ? OFFSET ?",(org,limit+1,offset)).fetchall(); out=[]
        for r in rows[:limit]:
            x=row_json(r)
-            for k in ('raw_json','normalized_json','query_context_json','cursor_json','rate_policy_json'):
+            for k in ('raw_json','normalized_json','query_context_json','response_metadata_json','cursor_json','rate_policy_json'):
                try:x[k]=json.loads(x[k])
                except (ValueError,TypeError):pass
            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',{})
-        if not name or kind not in ('csv','manual') or not isinstance(config,dict):return self.send_json(400,{"error":"invalid_source"})
+        name=str(payload.get('name','')).strip(); kind=str(payload.get('source_code',payload.get('kind',''))).strip().lower(); config=payload.get('config',{})
+        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)
+            # Registration may precede the actual local payload or optional provider
+            # approval. Discovery/ingest still validates the effective configuration.
            if config 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):
-        if not db.execute("SELECT id FROM discovery_queries WHERE id=? AND organization_id=?",(qid,user['organization_id'])).fetchone():return self.send_json(404,{"error":"not_found"})
-        return self.create_job({"type":"source_discovery","_accepted":True,"payload":{"discovery_query_id":qid},"idempotency_key":f"discovery-query-{qid}-{int(time.time())}"},db,user)
+        query=db.execute("SELECT * FROM discovery_queries WHERE id=? AND organization_id=?",(qid,user["organization_id"])).fetchone()
+        if not query:return self.send_json(404,{"error":"not_found"})
+        key=f"discovery-query-{qid}-{int(time.time())}"
+        result=self.create_job({"type":"source_discovery","_accepted":True,"payload":{"discovery_query_id":qid,"selected_adapters":json.loads(query["selected_adapters_json"] or "[]"),"max_records":query["max_records"],"daily_limit":query["daily_limit"]},"idempotency_key":key,"_defer_wakeup":True},db,user)
+        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,selected_adapters_json,location,category,max_records,daily_limit,schedule,dry_run,lifecycle,criteria_json,seed_urls_json) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)",(user["organization_id"],job["id"],query["selected_adapters_json"],query["location"],query["category"],query["max_records"],query["daily_limit"],query["schedule"],query["dry_run"],"queued",query["query_json"],"[]")); db.commit()
+        getattr(self.server,"job_wakeup",threading.Event()).set(); return result
    def test_source(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"})
@@ -1350,10 +1386,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()
-            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
+            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,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 +1493,71 @@ 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 []
+    query=None
+    if payload.get("discovery_query_id"):
+        query=db.execute("SELECT * FROM discovery_queries WHERE id=? AND organization_id=?",(payload["discovery_query_id"],org)).fetchone()
+        if query:
+            selected=json.loads(query["selected_adapters_json"] or "[]")
+            if not selected:
+                linked=db.execute("SELECT kind FROM sources WHERE id=? AND organization_id=?",(query["source_id"],org)).fetchone()
+                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)))
+    for source in sources:
+        handler.add_job_event(db,job["id"],org,"source.started",f"Starting {source['display_name'] or source['kind']}",5)
+        try:
+            config=json.loads(source["config_json"] or "{}")
+            if query and query["source_id"]==source["id"]:
+                config.update(json.loads(query["query_json"] or "{}"))
+            quota=json.loads(source["quota_json"] or "{}")
+            daily_limit=int(quota.get("daily_limit", payload.get("daily_limit", 100000)))
+            per_run_limit=int(quota.get("per_run_limit", max_records))
+            used_today=db.execute("SELECT COUNT(*) FROM source_records WHERE organization_id=? AND source_id=? AND date(created_at)=date('now')",(org,source["id"])).fetchone()[0]
+            if used_today >= daily_limit or per_run_limit <= 0:
+                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["kind"]).discover(config)
+        except Exception as exc:
+            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
+        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()
+            handler.add_job_event(db,job["id"],org,"source.raw_persisted",f"Persisting {source['kind']} record",20)
+            try:
+                cur=db.execute("INSERT INTO source_records(organization_id,source_id,discovery_query_id,discovery_run_id,content_hash,raw_json,normalized_json,normalized_key,source_url,provenance_json,query_context_json,response_metadata_json,processing_status) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?)",(org,source["id"],query["id"] if query else None,run["id"] if run else None,digest,raw,norm,nkey,str(config.get("source_url","")),json.dumps({"adapter":source["kind"],"metadata":page.metadata},sort_keys=True),json.dumps(payload.get("criteria",{}),sort_keys=True),json.dumps(page.metadata,sort_keys=True),"raw"))
+                record_id=cur.lastrowid; total+=1
+            except sqlite3.IntegrityError:
+                existing=db.execute("SELECT id,processing_status FROM source_records WHERE organization_id=? AND source_id=? AND content_hash=?",(org,source["id"],digest)).fetchone()
+                if existing and existing["processing_status"] in ("processed","matched"): continue
+                record_id=existing["id"] if existing else None
+            handler.add_job_event(db,job["id"],org,"source.normalized",f"Normalized {normalized.get('name','')}",35)
+            suppressed=is_suppressed(normalized,[dict(r) for r in db.execute("SELECT kind,value FROM suppressions WHERE organization_id=? AND active=1",(org,))])
+            if suppressed:
+                if record_id: db.execute("UPDATE source_records SET processing_status='skipped' WHERE id=? AND organization_id=?",(record_id,org))
+                handler.add_job_event(db,job["id"],org,"review.skipped",f"Suppressed source record {record_id}",40,"SUPPRESSED"); continue
+            existing=db.execute("SELECT * FROM businesses WHERE organization_id=? AND ((website_domain<>'' AND website_domain=?) OR (email<>'' AND email=?) OR (phone<>'' AND phone=?) OR (name=? AND city=?)) ORDER BY id LIMIT 1",(org,normalized["website_domain"],normalized["email"],normalized["phone"],normalized["name"],normalized["city"])).fetchone()
+            if existing: bid=existing["id"]; handler.add_job_event(db,job["id"],org,"business.matched",f"Matched business {bid}",50)
+            else:
+                scored=score_business(normalized); cur=db.execute("INSERT INTO businesses(organization_id,name,website,website_domain,email,phone,description,province,city,suburb,score,score_version,score_factors,website_class,review_status) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",(org,normalized["name"],normalized["website"],normalized["website_domain"],normalized["email"],normalized["phone"],normalized.get("description",""),normalized["province"],normalized["city"],normalized["suburb"],scored["score"],scored["score_version"],json.dumps(scored["factors"]),scored["website_class"],"pending")); bid=cur.lastrowid
+                db.execute("INSERT INTO score_history(organization_id,business_id,score,eligible,priority_band,score_version,explanations_json,signals_json) VALUES(?,?,?,?,?,?,?,?)",(org,bid,scored["score"],1,"high" if scored["score"]>=70 else "medium" if scored["score"]>=40 else "low",scored["score_version"],json.dumps(scored["factors"]),json.dumps({"source":source["kind"]})))
+                handler.add_job_event(db,job["id"],org,"business.created",f"Created business {bid}",60)
+            if normalized["website_domain"] and not db.execute("SELECT 1 FROM domains WHERE organization_id=? AND business_id=? AND domain=?",(org,bid,normalized["website_domain"])).fetchone(): db.execute("INSERT INTO domains(business_id,organization_id,domain,kind) VALUES(?,?,?,?)",(bid,org,normalized["website_domain"],"website"))
+            if normalized["website"] and not db.execute("SELECT 1 FROM websites WHERE organization_id=? AND business_id=? AND url=?",(org,bid,normalized["website"])).fetchone(): db.execute("INSERT INTO websites(business_id,organization_id,url,website_class) VALUES(?,?,?,?)",(bid,org,normalized["website"],"business_site"))
+            if normalized["website"] or normalized["website_domain"]: db.execute("INSERT OR IGNORE INTO evidence(business_id,organization_id,kind,url,claim) VALUES(?,?,?,?,?)",(bid,org,"source_record",normalized["website"],"Discovered by "+source["kind"]))
+            if normalized["email"] or normalized["phone"]:
+                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"]))
+    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 +1578,18 @@ 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_state=db.execute("SELECT lifecycle FROM discovery_runs WHERE organization_id=? AND job_id=?",(org,jid)).fetchone()
+                    if run_state and run_state["lifecycle"] == "cancelled":
+                        db.execute("UPDATE jobs SET status='cancelled',completed_at=CURRENT_TIMESTAMP,updated_at=CURRENT_TIMESTAMP WHERE id=?",(jid,)); server_handler.add_job_event(db,jid,org,"cancelled","Discovery cancelled",job["progress"]); db.commit(); continue
+                    if run_state and run_state["lifecycle"] == "paused":
+                        db.execute("UPDATE jobs SET status='queued',updated_at=CURRENT_TIMESTAMP WHERE id=?",(jid,)); server_handler.add_job_event(db,jid,org,"paused","Discovery paused",job["progress"]); db.commit(); continue
+                    _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
@@ -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,146 @@ 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
-    def validate(self, config: Mapping[str, Any]) -> ValidationResult: ...
+    source_code: str
+    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}
-    # Accept common CSV spellings without retaining arbitrary sensitive fields.
-    aliases = {"company": "name", "url": "website", "domain": "website"}
+    result = {field: str(row.get(field, "") or "").strip() for field in _FIELDS}
+    aliases = {"company":"name", "business":"name", "url":"website", "domain":"website", "address":"location", "address_line":"location"}
    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:
-    kind = "manual"
-    def validate(self, config: Mapping[str, Any]) -> ValidationResult:
-        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})
+def normalized_record(row: Mapping[str, Any], *, source_url="", provenance="") -> NormalizedRecord:
+    x = normalize_record(row)
+    return NormalizedRecord(**x, source_url=source_url, provenance=provenance, raw=dict(row))

-class CsvSource:
-    kind = "csv"
-    def validate(self, config: Mapping[str, Any]) -> ValidationResult:
+def exponential_backoff(attempt: int, base: float = 0.5, maximum: float = 30.0, jitter: float = 0.25) -> float:
+    """Bounded exponential delay with symmetric jitter; no sleeping occurs here."""
+    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) >= max(1, int(threshold))
+
+def quota_remaining(used: int, limit: int | None) -> int | None:
+    """Return remaining quota, clamped so malformed values fail closed."""
+    if limit is None: return None
+    return max(0, int(limit) - max(0, int(used)))
+
+def quota_allowed(used: int, limit: int | None) -> bool:
+    remaining = quota_remaining(used, limit)
+    return remaining is None or remaining > 0
+
+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:
-        validation = self.validate(config)
-        if not validation.valid: raise ValueError(validation.errors[0])
+    def discover(self, config, cursor=None):
+        result = self.validate_config(config)
+        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(records, None, {"adapter": self.kind, "columns": reader.fieldnames or []})
+        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 []})

-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()]
@@ -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,
- config_json TEXT NOT NULL DEFAULT '{}', health_status TEXT NOT NULL DEFAULT 'unknown',
+ 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 '{}', 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,26 @@ 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,
- 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',
+ 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, 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 '{}', response_metadata_json TEXT NOT NULL DEFAULT '{}', processing_status TEXT NOT NULL DEFAULT 'raw' CHECK(processing_status IN ('raw','processed','matched','failed','skipped')),
 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 INDEX IF NOT EXISTS idx_source_records_processing ON source_records(organization_id,processing_status,created_at 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 +349,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,
- organization_id TEXT NOT NULL REFERENCES organizations(id) ON DELETE CASCADE,
- job_id INTEGER NOT NULL REFERENCES jobs(id) ON DELETE CASCADE,
+ id INTEGER PRIMARY KEY AUTOINCREMENT, organization_id TEXT NOT NULL REFERENCES organizations(id), job_id INTEGER NOT NULL REFERENCES jobs(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,
 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,
@@ -47,8 +47,80 @@ class SourceApiTests(unittest.TestCase):
        _,q=self.req('POST','/api/v1/discovery-queries',{'source_id':source['id'],'name':'q','query':{'csv':'name\nA'}})
        status,job=self.req('POST',f"/api/v1/discovery-queries/{q['id']}/run",{})
        self.assertEqual(status,202); self.assertEqual(job['type'],'source_discovery')
+        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'],'succeeded')
+        self.assertEqual(len(self.req('GET','/api/v1/businesses')[1]['items']),1)
+        self.assertEqual(self.req('GET','/api/v1/source-records')[1]['items'][0]['discovery_query_id'],q['id'])
        self.assertEqual(self.req('GET','/api/v1/source-records?page_size=101')[0],400)
    def test_sources_require_auth(self):
        self.cookie=None; self.assertEqual(self.req('GET','/api/v1/sources',cookie=False)[0],401)

+    def test_fresh_schema_accepts_optional_source_kind_fail_closed(self):
+        status, source = self.req('POST', '/api/v1/sources', {'name': 'RDAP', 'kind': 'rdap', 'config': {}})
+        self.assertEqual(status, 201)
+        self.assertEqual(source['kind'], 'rdap')
+        self.assertEqual(self.req('POST', f"/api/v1/sources/{source['id']}/test", {})[0], 200)
+        status, health = self.req('GET', f"/api/v1/sources/{source['id']}/health")
+        self.assertEqual(status, 200)
+        self.assertFalse(health['configured'])
+
+    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,
+            'config': {'rows': [{'name': 'Acme Solar', 'website': 'https://acme.test',
+                                 'email': 'hello@acme.test', 'phone': '011 555 0100',
+                                 'description': 'solar installers', 'location': 'Cape Town'}]}})
+        self.assertEqual(status, 201)
+        payload = {'criteria': {'keywords': ['solar']}, 'selected_adapters': ['manual'],
+                   'idempotency_key': 'source-run-1', 'max_records': 10}
+        status, job = self.req('POST', '/api/v1/discovery', payload)
+        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'], 'succeeded')
+        events = self.req('GET', f"/api/v1/jobs/{job['id']}/events")[1]['items']
+        event_types = [event['event_type'] for event in events]
+        for stage in ('source.started', 'source.raw_persisted', 'source.normalized',
+                      'business.created', 'enrichment.queued', 'review.queued', 'discovery.completed'):
+            self.assertIn(stage, event_types)
+        businesses = self.req('GET', '/api/v1/businesses')[1]['items']
+        self.assertEqual(len(businesses), 1)
+        detail = self.req('GET', f"/api/v1/businesses/{businesses[0]['id']}")[1]
+        self.assertTrue(detail['domains']); self.assertTrue(detail['websites']); self.assertTrue(detail['evidence'])
+        self.assertTrue(detail['contacts']); self.assertEqual(detail['review_status'], 'pending')
+        db = sqlite3.connect(self.tmp.name + '/x.db')
+        self.assertEqual(db.execute('SELECT processing_status FROM source_records').fetchone()[0], 'processed')
+        self.assertEqual(db.execute('SELECT status FROM enrichment_queue').fetchone()[0], 'completed')
+        db.close()
+        status, second = self.req('POST', '/api/v1/discovery', {**payload, 'idempotency_key': 'source-run-2'})
+        self.assertEqual(status, 202)
+        for _ in range(100):
+            _, current = self.req('GET', f"/api/v1/jobs/{second['id']}")
+            if current['status'] in ('succeeded', 'failed'): break
+            threading.Event().wait(.01)
+        self.assertEqual(current['status'], 'succeeded')
+        self.assertEqual(len(self.req('GET', '/api/v1/businesses')[1]['items']), 1)
+        self.assertEqual(self.req('GET', '/api/v1/source-records')[1]['items'].__len__(), 1)
+
+    def test_source_limits_and_run_lifecycle_are_enforced(self):
+        status, source = self.req('POST', '/api/v1/sources', {'name': 'Limited', 'kind': 'manual', 'enabled': True,
+            'config': {'rows': [{'name': 'A'}, {'name': 'B'}]}, 'quota': {'daily_limit': 1, 'per_run_limit': 1}})
+        self.assertEqual(status, 201)
+        status, job = self.req('POST', '/api/v1/discovery', {'criteria': {}, 'selected_adapters': ['manual'],
+            'idempotency_key': 'limited-1', 'max_records': 10, 'daily_limit': 1})
+        self.assertEqual(status, 202)
+        runs = self.req('GET', '/api/v1/discovery-runs')[1]['items']; rid = runs[0]['id']
+        self.assertEqual(self.req('POST', f'/api/v1/discovery-runs/{rid}/pause', {})[0], 200)
+        self.assertEqual(self.req('POST', f'/api/v1/discovery-runs/{rid}/resume', {})[0], 200)
+        self.assertEqual(self.req('POST', f'/api/v1/discovery-runs/{rid}/cancel', {})[0], 200)
+        db = sqlite3.connect(self.tmp.name + '/x.db')
+        self.assertEqual(db.execute("SELECT lifecycle FROM discovery_runs WHERE id=?", (rid,)).fetchone()[0], 'cancelled')
+        self.assertIn(db.execute("SELECT status FROM jobs WHERE id=?", (job['id'],)).fetchone()[0], ('cancelled', 'succeeded'))
+        db.close()
+
 if __name__=='__main__': unittest.main()
@@ -133,7 +133,9 @@ Phase 14 is not implemented as a live outreach workflow. The current static clie

 ## Remaining limitations

-The static client has no client-side crawler, scanner, contact extractor, enrichment scheduler, outreach integration, availability provider, or SSE delivery. It can display server-provided Phase 9 observations, but production still requires server-side official-site scoping, SSRF/DNS-rebinding/redirect controls, hard extraction/page/byte/time/candidate budgets, durable history/cache isolation and retention/deletion, abuse/rate controls, suppression regression tests, and authenticated provenance/audit coverage. For domain intelligence, display registrable-domain/PSL version and unresolved reasons, DNS status and freshness/TTL (not “available”), independent MX/NS/TXT uncertainty, and association confidence with explainable evidence. Never auto-attach candidate domains or treat `nxdomain`/`no_data` as availability. CSV preview is capped for display and is not an import workflow.
+The Discovery Runs workspace is a static client over the existing authenticated contracts: `POST /api/v1/discovery` creates a bounded run, `GET /api/v1/discovery-runs` lists run history, and `GET /api/v1/jobs/{id}` supplies progress/events. It sends category/geography criteria, selected approved source IDs, max records, daily limit, schedule metadata, dry-run intent, optional seeds, and an idempotency key. The UI polls persisted job state, renders independently scrollable logs, metrics, partial/error states, and source-health telemetry when returned. CSV/manual source management continues to use `GET/POST/PATCH /api/v1/sources`, `POST /api/v1/sources/{id}/test`, and `GET /api/v1/source-records`; provider configuration remains read/write-only through the existing AI provider routes.
+
+Pause/resume and per-source retry/circuit buttons are deliberately fail-closed because the current backend exposes no matching mutation routes. The UI documents the expected future names (`POST /api/v1/jobs/{id}/pause`, `POST /api/v1/jobs/{id}/resume`, and source-run retry/circuit endpoints) instead of pretending those actions succeeded. Cancel uses the existing `POST /api/v1/jobs/{id}/cancel` contract. Prospect explorer filters use server query fields where available and apply safe client-side matching for returned source, geography, category, website/domain, and contact fields. The browser never fetches targets or stores secrets in localStorage.

 ## Phase 13 optional AI assistance UI contract

@@ -6,14 +6,14 @@
  const endpoint = (path) => `${API_BASE}${path}`;
  let prospects = [], selectedId = null, selectedDetail = null, currentUser = null;
  let page = 1, pageSize = 10, hasNextPage = false, savedFilters = [], reviewQueue = [], selectedReviewIds = new Set();
-  const filterKeys = ['q','score','status','website_class','pipeline_stage','freshness'];
+  const filterKeys = ['q','score','status','website_class','pipeline_stage','freshness','source','geography','category','contact_status'];
  const reviewItems = payload => payloadItems(payload, ['items','businesses','prospects','queue']);
  const savedItems = payload => payloadItems(payload, ['items','saved_filters','filters','views']);
  const apiError = error => error?.message || 'Request failed.';
  function setExplorerState(text='', error=false){const el=$('explorerState');if(el){el.textContent=text;el.className=`explorer-state${error?' error':''}`;}}
  function setSavedMessage(text,error=false){const el=$('savedFilterMessage');if(el){el.textContent=text;el.className=`form-message${error?' error':''}`;}}
  function currentFilters(){return filterValues();}
-  function applyFilters(filters={}){filterKeys.forEach(key=>{const id={q:'searchInput',score:'scoreFilter',status:'statusFilter',website_class:'websiteClassFilter',pipeline_stage:'pipelineFilter'}[key];if(filters[key]!==undefined&&$(id))$(id).value=filters[key];});if($('explorerState'))$('explorerState').dataset.filter=filters.freshness||'all';page=1;loadData();}
+  function applyFilters(filters={}){filterKeys.forEach(key=>{const id={q:'searchInput',score:'scoreFilter',status:'statusFilter',website_class:'websiteClassFilter',pipeline_stage:'pipelineFilter',source:'sourceFilter',geography:'geographyFilter',category:'categoryFilter',contact_status:'contactStatusFilter'}[key];if(filters[key]!==undefined&&$(id))$(id).value=filters[key];});if($('explorerState'))$('explorerState').dataset.filter=filters.freshness||'all';page=1;loadData();}
  function selectedReviewProspects(){return [...selectedReviewIds].filter(id=>reviewQueue.some(p=>String(p.id)===String(id)));}
  function updateBulkState(){const count=selectedReviewProspects().length; if($('selectedReviewCount'))$('selectedReviewCount').textContent=`${count} selected`;['bulkVerifyBtn','bulkRejectBtn'].forEach(id=>{if($(id))$(id).disabled=!count;});if($('selectAllReview'))$('selectAllReview').checked=Boolean(reviewQueue.length&&count===reviewQueue.length);}
  function renderReviewQueue(payload){reviewQueue=reviewItems(payload).slice(0,100);$('reviewQueueCount').textContent=Number(payload?.count??payload?.total??reviewQueue.length);const state=$('reviewQueueState');if(!state)return;if(!reviewQueue.length){state.innerHTML='<p class="muted">No prospects currently require human review.</p>';updateBulkState();return;}state.innerHTML=`<div class="queue-list">${reviewQueue.map(p=>`<label class="queue-row"><input type="checkbox" data-review-id="${esc(p.id)}" ${selectedReviewIds.has(String(p.id))?'checked':''}><span><strong>${esc(p.name||`Prospect ${p.id}`)}</strong><small>${esc(p.website_domain||p.location||'Evidence pending')}</small></span><span class="status review">Needs review</span></label>`).join('')}</div><p class="queue-boundary">Showing up to 100 queued prospects. Bulk actions apply only to selected visible records.</p>`;updateBulkState();}
@@ -37,12 +37,13 @@
  function showLogin(message=''){currentUser=null;$('dashboardShell').hidden=true;$('loginScreen').hidden=false;$('loginMessage').textContent=message;$('loginMessage').className=`form-message${message?' error':''}`;}
  function showDashboard(user){currentUser=user||{};const name=currentUser.name||currentUser.full_name||currentUser.email||'Workspace member';const role=currentUser.role||currentUser.roles?.[0]||'Member';$('userIdentity').textContent=`${name} · ${role}`;$('userAvatar').textContent=name.split(/\s+/).map(x=>x[0]).join('').slice(0,2).toUpperCase();$('loginScreen').hidden=true;$('dashboardShell').hidden=false;updateJobPermissions();}
  function renderMetrics(summary){const total=Number(summary?.businesses??summary?.total??prospects.length),high=Number(summary?.high_fit??summary?.high_fit_count??prospects.filter(p=>scoreFor(p)>=80).length),review=Number(summary?.needs_review??summary?.review_count??prospects.filter(p=>statusOf(p)==='review').length),suppressed=Number(summary?.suppressed??summary?.suppressed_count??prospects.filter(p=>statusOf(p)==='suppressed').length),fresh=prospects.length?Math.round(prospects.filter(p=>freshness(p).cls==='good').length/prospects.length*100):0;$('metricTotal').textContent=total;$('heroCount').textContent=`${total} prospects`;$('metricReview').textContent=review;$('metricHigh').textContent=high;$('metricFresh').textContent=`${summary?.freshness_under_7d??summary?.fresh_count??fresh}%`;if($('metricSuppressed'))$('metricSuppressed').textContent=suppressed;}
-  function filterValues(){return {q:$('searchInput').value.trim(),score:$('scoreFilter').value,status:$('statusFilter').value,website_class:$('websiteClassFilter').value,pipeline_stage:$('pipelineFilter').value,freshness:$('explorerState')?.dataset?.filter||'all'};}
-  function filtered(){const {q,score:sf,status:st,website_class:wc,pipeline_stage:ps,freshness}=filterValues();return prospects.filter(p=>{const s=scoreFor(p),text=`${p.name} ${p.website_domain||p.website||''} ${p.location||''}`.toLowerCase(),stage=p.pipeline_stage||p.pipeline?.stage||(Array.isArray(p.pipeline)?p.pipeline.at(-1)?.stage:'')||'new';return(!q||text.includes(q.toLowerCase()))&&(sf==='all'||(sf==='high'&&s>=80)||(sf==='medium'&&s>=60&&s<80)||(sf==='low'&&s<60))&&(st==='all'||statusOf(p)===st)&&(wc==='all'||(p.website_class||'missing')===wc)&&(ps==='all'||stage===ps)&&(freshness!=='under_7d'||freshnessValue(p));});}
+  function filterValues(){return {q:$('searchInput').value.trim(),score:$('scoreFilter').value,status:$('statusFilter').value,website_class:$('websiteClassFilter').value,pipeline_stage:$('pipelineFilter').value,source:$('sourceFilter')?.value||'all',geography:$('geographyFilter')?.value||'all',category:$('categoryFilter')?.value||'all',contact_status:$('contactStatusFilter')?.value||'all',freshness:$('explorerState')?.dataset?.filter||'all'};}
+  function filtered(){const {q,score:sf,status:st,website_class:wc,pipeline_stage:ps,source,geography,category,contact_status,freshness}=filterValues();return prospects.filter(p=>{const s=scoreFor(p),text=`${p.name} ${p.website_domain||p.website||''} ${p.location||''}`.toLowerCase(),stage=p.pipeline_stage||p.pipeline?.stage||(Array.isArray(p.pipeline)?p.pipeline.at(-1)?.stage:'')||'new',contacts=p.contacts||p.contact_extractions||[], pSource=String(p.source_name||p.source||p.source_id||'all'), geo=String(p.city||p.province||p.country||p.location||'all'), cat=String(p.category||p.industry||p.categories?.[0]||'all'), contactState=contacts.length||p.email||p.phone?'present':statusOf(p)==='suppressed'?'suppressed':'missing';return(!q||text.includes(q.toLowerCase()))&&(source==='all'||pSource===source||String(p.source_id)===source)&&(geography==='all'||geo.toLowerCase().includes(geography.toLowerCase()))&&(category==='all'||cat===category)&&(contact_status==='all'||contactState===contact_status)&&(sf==='all'||(sf==='high'&&s>=80)||(sf==='medium'&&s>=60&&s<80)||(sf==='low'&&s<60))&&(st==='all'||statusOf(p)===st)&&(wc==='all'||(p.website_class||'missing')===wc)&&(ps==='all'||stage===ps)&&(freshness!=='under_7d'||freshnessValue(p));});}
  const freshnessValue=p=>freshness(p).cls==='good';
  function renderRows(){const rows=filtered();$('resultCount').textContent=`Showing ${rows.length} prospect${rows.length===1?'':'s'} · page ${page}`;$('prospectRows').innerHTML=rows.length?rows.map(p=>{const s=scoreFor(p),f=freshness(p),st=statusOf(p),factors=p.score_factors||p.factors||[];return `<tr data-id="${esc(p.id)}" class="${Number(p.id)===Number(selectedId)?'selected':''}"><td>${esc(p.name)}<span class="company-sub">${esc(p.website_domain||'no detected website')}</span></td><td><span class="score ${scoreClass(s)}">${s} <small>/ 100</small></span></td><td class="evidence"><strong>${factors.length} signal${factors.length===1?'':'s'}</strong>${esc(factors.slice(0,2).map(labelFactor).join(' · ')||'Limited evidence')}</td><td><span class="fresh ${f.cls}">${f.label}</span></td><td><span class="status ${st}">${st==='review'?'Needs review':st[0].toUpperCase()+st.slice(1)}</span></td><td class="row-arrow">›</td></tr>`;}).join(''):'<tr><td colspan="6" class="muted">No prospects match these filters.</td></tr>';$('nextPageBtn').disabled=!hasNextPage;document.querySelectorAll('#prospectRows tr[data-id]').forEach(row=>row.addEventListener('click',()=>selectProspect(row.dataset.id)));}
-  function listQuery(){const f=filterValues(),params=new URLSearchParams({page:String(page),page_size:String(pageSize)});if(f.q)params.set('q',f.q);if(f.score!=='all')params.set('score_band',f.score);if(f.status!=='all')params.set('status',f.status);if(f.website_class!=='all')params.set('website_class',f.website_class);if(f.pipeline_stage!=='all')params.set('pipeline_stage',f.pipeline_stage);if(f.freshness!=='all')params.set('freshness',f.freshness);return `?${params}`;}
-  async function loadData(){ $('apiStatus').textContent='● Connecting…';$('apiStatus').classList.remove('live');setExplorerState('Loading prospects…');try{const [listRes,summaryRes]=await Promise.all([request(`/api/v1/businesses${listQuery()}`),request('/api/v1/dashboard/summary')]);if(!listRes.ok||!summaryRes.ok)throw new Error('API request failed');const list=await listRes.json(),summary=await summaryRes.json();prospects=Array.isArray(list)?list:(list.businesses||list.items||[]);hasNextPage=Boolean(list.has_next??list.next_page??list.next??list.next_cursor);renderMetrics(summary);$('apiStatus').textContent='● API connected';$('apiStatus').classList.add('live');setExplorerState('');}catch(error){if(error.message==='unauthorized')return;prospects=[];hasNextPage=false;renderMetrics(null);$('apiStatus').textContent='● API unavailable';setExplorerState(`Unable to load prospects: ${apiError(error)}`,true);}renderRows();if(selectedId)loadDetail(selectedId);}
+  function listQuery(){const f=filterValues(),params=new URLSearchParams({page:String(page),page_size:String(pageSize)});if(f.q)params.set('q',f.q);if(f.score!=='all')params.set('score_band',f.score);if(f.status!=='all')params.set('status',f.status);if(f.website_class!=='all')params.set('website_class',f.website_class);if(f.pipeline_stage!=='all')params.set('pipeline_stage',f.pipeline_stage);if(f.source!=='all')params.set('source',f.source);if(f.geography!=='all')params.set('geography',f.geography);if(f.category!=='all')params.set('category',f.category);if(f.contact_status!=='all')params.set('contact_status',f.contact_status);if(f.freshness!=='all')params.set('freshness',f.freshness);return `?${params}`;}
+  function populateProspectFilters(){const values={source:new Set(),geography:new Set(),category:new Set()};prospects.forEach(p=>{values.source.add(String(p.source_name||p.source||p.source_id||''));values.geography.add(String(p.city||p.province||p.country||p.location||''));values.category.add(String(p.category||p.industry||p.categories?.[0]||''));});[['sourceFilter',values.source,'All sources'],['geographyFilter',values.geography,'All geographies'],['categoryFilter',values.category,'All categories']].forEach(([id,set,placeholder])=>{const el=$(id),current=el?.value||'all';if(!el)return;el.innerHTML='<option value="all">'+placeholder+'</option>'+[...set].filter(Boolean).sort().map(value=>'<option value="'+esc(value)+'">'+esc(value)+'</option>').join('');if([...el.options].some(option=>option.value===current))el.value=current;});}
+  async function loadData(){ $('apiStatus').textContent='● Connecting…';$('apiStatus').classList.remove('live');setExplorerState('Loading prospects…');try{const [listRes,summaryRes]=await Promise.all([request(`/api/v1/businesses${listQuery()}`),request('/api/v1/dashboard/summary')]);if(!listRes.ok||!summaryRes.ok)throw new Error('API request failed');const list=await listRes.json(),summary=await summaryRes.json();prospects=Array.isArray(list)?list:(list.businesses||list.items||[]);populateProspectFilters();hasNextPage=Boolean(list.has_next??list.next_page??list.next??list.next_cursor);renderMetrics(summary);$('apiStatus').textContent='● API connected';$('apiStatus').classList.add('live');setExplorerState('');}catch(error){if(error.message==='unauthorized')return;prospects=[];hasNextPage=false;renderMetrics(null);$('apiStatus').textContent='● API unavailable';setExplorerState(`Unable to load prospects: ${apiError(error)}`,true);}renderRows();if(selectedId)loadDetail(selectedId);}
  async function selectProspect(id){selectedId=Number(id);selectedDetail=null;renderRows();$('detailPanel').innerHTML='<div class="detail-loading" aria-live="polite">Loading prospect detail…</div>';await loadDetail(selectedId);}
  async function loadDetail(id){try{const detail=await jsonRequest(`/api/v1/businesses/${encodeURIComponent(id)}`);selectedDetail=detail;const index=prospects.findIndex(p=>Number(p.id)===Number(id));if(index>=0)prospects[index]={...prospects[index],...detail};renderDetail(detail);}catch(error){if(error.message!=='unauthorized')$('detailPanel').innerHTML=`<div class="detail-error" role="alert"><h3>Unable to load detail</h3><p>${esc(error.message)}</p><button class="button ghost" id="retryDetailBtn" type="button">Try again</button></div>`;}}
  const listItems=(items,empty,label)=>Array.isArray(items)&&items.length?`<ul class="detail-list">${items.map(item=>`<li>${esc(typeof item==='string'?item:item[label]||item.value||item.name||JSON.stringify(item))}</li>`).join('')}</ul>`:`<p class="muted">${empty}</p>`;
@@ -274,7 +275,7 @@
  const sourceStatus = source => sourceState(source) ? 'enabled' : 'disabled';
  const sourceItems = payload => Array.isArray(payload) ? payload : (payload?.sources || payload?.items || payload?.records || []);
  function sourceMessage(text, error = false) { const el = $('sourcesMessage'); if (el) { el.textContent = text || ''; el.className = `sources-message${error ? ' error' : ''}`; } }
-  function renderSourceSelect() { const select = $('discoverySource'); if (!select) return; select.innerHTML = `<option value="">Select a source</option>${sources.map(s => `<option value="${esc(s.id)}">${esc(s.name || s.label || `Source ${s.id}`)} · ${sourceStatus(s)}</option>`).join('')}`; }
+  function renderSourceSelect() { const select = $('discoverySource'); if (select) select.innerHTML = `<option value="">Select a source</option>${sources.map(s => `<option value="${esc(s.id)}">${esc(s.name || s.label || `Source ${s.id}`)} · ${sourceStatus(s)}</option>`).join('')}`; const multi=$('directDiscoverySources'); if(multi) multi.innerHTML=sources.filter(sourceState).map(s=>`<option value="${esc(s.id)}">${esc(s.name||s.label||`Source ${s.id}`)} · ${esc(s.kind||'source')}</option>`).join('') || '<option value="" disabled>No approved sources enabled</option>'; }
  function renderSources() { renderSourceSelect(); const list = $('sourcesList'); if (!sources.length) { list.innerHTML = '<div class="source-empty">No sources returned by the workspace.</div>'; return; } list.innerHTML = sources.map(source => { const status = sourceStatus(source), health = source.health || source.health_status || 'Not tested', terms = source.terms_reviewed ?? source.terms_status ?? 'Not reviewed', config = source.config || {}; return `<article class="source-row" data-source-id="${esc(source.id)}"><div class="source-row-main"><strong>${esc(source.name || source.label || `Source ${source.id}`)}</strong><small>${esc(source.kind || source.source_type || source.type || 'manual')}${source.url || config.url ? ` · ${esc(source.url || config.url)}` : ''}</small></div><span class="source-status ${status}">${status}</span><dl class="source-facts"><div><dt>Owner</dt><dd>${esc(source.owner || source.owner_name || config.owner || 'Not assigned')}</dd></div><div><dt>Terms</dt><dd>${esc(String(source.terms_reviewed ?? source.terms_status ?? (config.terms_url ? 'Provided' : 'Not reviewed')))}</dd></div><div><dt>Rate limit</dt><dd>${esc(source.rate_limit || source.rate_limit_label || config.rate_limit || 'Not set')}</dd></div><div><dt>Health</dt><dd>${esc(String(health))}</dd></div></dl><div class="source-actions"><button class="button ghost compact" type="button" data-source-action="test" data-source-id="${esc(source.id)}">Test</button><button class="button ${status === 'enabled' ? 'danger' : 'primary'} compact" type="button" data-source-action="toggle" data-source-id="${esc(source.id)}">${status === 'enabled' ? 'Disable' : 'Enable'}</button></div></article>`; }).join(''); }
  function renderSourceRecords(items) { const list = $('sourceRecordsList'); if (!items.length) { list.innerHTML = '<div class="source-empty">No source records returned by the workspace.</div>'; return; } list.innerHTML = `<div class="source-record-table"><table><thead><tr><th>Record</th><th>Source</th><th>Status</th><th>Observed</th></tr></thead><tbody>${items.slice(0,25).map(record => `<tr><td>${esc(record.name || record.title || record.external_id || record.id || 'Unnamed record')}</td><td>${esc(record.source_name || record.source || 'Unknown source')}</td><td><span class="status">${esc(record.status || 'Pending')}</span></td><td>${esc(record.observed_at || record.created_at || 'Time unavailable')}</td></tr>`).join('')}</tbody></table></div>`; }
  async function loadSources() { sourceMessage('Loading sources…'); $('sourcesList').innerHTML = '<div class="detail-loading" aria-live="polite">Loading source registry…</div>'; $('sourceRecordsList').innerHTML = '<div class="detail-loading" aria-live="polite">Loading source records…</div>'; try { const [sourcePayload, recordPayload] = await Promise.all([jsonRequest('/api/v1/sources'), jsonRequest('/api/v1/source-records?page_size=25')]); sources = sourceItems(sourcePayload); renderSources(); renderSourceRecords(sourceItems(recordPayload)); $('sourcesUpdatedAt').textContent = `Updated ${new Date().toLocaleTimeString([], {hour:'2-digit', minute:'2-digit'})}`; sourceMessage(sources.some(sourceState) ? '' : 'No live source is enabled.'); } catch (error) { sources = []; renderSources(); renderSourceRecords([]); if (error.message !== 'unauthorized') sourceMessage(error.message || 'Unable to load sources.', true); } }
@@ -283,13 +284,34 @@
  async function runDiscovery(dryRun) { const form = $('discoveryForm'), data = Object.fromEntries(new FormData(form).entries()); data.dry_run = Boolean(dryRun); if (!data.source_id || !data.query.trim()) { message('discoveryMessage', 'Select a source and enter a query.', true); return; } const source = sources.find(item => String(item.id) === String(data.source_id)); if (!dryRun && !sourceState(source)) { message('discoveryMessage', 'This source is disabled. Enable it only after review.', true); return; } const button = dryRun ? $('discoveryDryRunBtn') : $('discoveryRunBtn'); button.disabled = true; message('discoveryMessage', dryRun ? 'Validating query…' : 'Starting discovery…'); try { const query = await jsonRequest('/api/v1/discovery-queries', {method:'POST', headers:{'Content-Type':'application/json'}, body:JSON.stringify({source_id:Number(data.source_id), name:data.query.trim().slice(0,80), query:{text:data.query.trim()}, dry_run:data.dry_run})}); if (!dryRun) await jsonRequest(`/api/v1/discovery-queries/${encodeURIComponent(query.id)}/run`, {method:'POST', headers:{'Content-Type':'application/json'}, body:JSON.stringify({})}); message('discoveryMessage', dryRun ? 'Dry run completed; no discovery job was started.' : 'Discovery request accepted. Check Jobs for progress.'); } catch (error) { if (error.message !== 'unauthorized') message('discoveryMessage', error.message || 'Discovery request failed.', true); } finally { button.disabled = false; } }
  let discoveryRuns = [], selectedDiscoveryRunId = null, discoveryPollTimer = null;
  const discoveryStatus = run => String(run?.status || run?.state || 'queued').toLowerCase().replaceAll('_','-');
-  const discoveryLabel = state => ({queued:'Queued',running:'Running',succeeded:'Complete',failed:'Failed',cancelled:'Cancelled',partial:'Partial result',stale:'Stale'})[state] || state.replaceAll('-',' ');
+  const discoveryLabel = state => ({queued:'Queued',running:'Running',succeeded:'Complete',failed:'Failed',cancelled:'Cancelled',partial:'Partial result',paused:'Paused',stale:'Stale'})[state] || state.replaceAll('-',' ');
  const discoveryCandidates = run => payloadItems(run?.result||{}, ['items','candidates','businesses','prospects','results']);
-  function renderDiscoveryResult(run, job) { const el=$('directDiscoveryResultState'), status=discoveryStatus(job||run), candidates=discoveryCandidates(run), result=run?.result||{}; $('directDiscoveryResultTitle').textContent=`Run ${run.id} · ${run.criteria?.keywords||run.criteria?.keyword||'Discovery brief'}`; $('directDiscoveryResultStatus').textContent=discoveryLabel(status); if(status==='running'||status==='queued'){el.innerHTML=`<div class="discovery-progress"><strong>${esc(discoveryLabel(status))}</strong><p>${esc(job?.progress_message||'The bounded discovery job is processing approved seed URLs.')}</p><div class="progress-track"><span style="width:${Math.max(0,Math.min(100,Number(job?.progress||0)))}%"></span></div></div>`;return;} const partial=status==='partial'||run.partial===true||result.partial===true; el.innerHTML=`${partial?'<div class="discovery-partial" role="status"><strong>Partial results</strong><p>The job reached a bounded limit or returned an incomplete source response. Review provenance before using any candidate.</p></div>':''}${status==='failed'?`<div class="detail-error" role="alert"><strong>Discovery failed</strong><p>${esc(job?.error?.message||job?.message||run.error||'The job returned no safe result.')}</p></div>`:''}<div class="discovery-result-meta"><span>${candidates.length} candidate${candidates.length===1?'':'s'}</span><span>Seed URLs: ${Array.isArray(run.seed_urls)?run.seed_urls.length:0}</span><span>${run.completed_at||run.updated_at||'Completion time unavailable'}</span></div>${candidates.length?`<div class="discovery-candidate-list">${candidates.slice(0,50).map(item=>`<article class="discovery-candidate"><strong>${esc(item.name||item.business_name||item.title||'Unnamed candidate')}</strong><span>${esc(item.website||item.url||item.domain||'No URL returned')}</span><small>${esc(item.source_url||item.source||'Source provenance unavailable')} · ${esc(item.confidence||item.score||'Confidence unknown')}</small></article>`).join('')}</div>`:'<p class="muted">No candidates returned. This is distinct from a successful empty result.</p>'}`; }
-  async function loadDiscoveryRun(run) { selectedDiscoveryRunId=run.id; renderDiscoveryResult(run); try { const job=await jsonRequest(`/api/v1/jobs/${encodeURIComponent(run.job_id)}`); renderDiscoveryResult(run,job); if(['queued','running'].includes(discoveryStatus(job))){clearTimeout(discoveryPollTimer);discoveryPollTimer=setTimeout(loadDiscoveryRuns,3000);} } catch(error){if(error.message!=='unauthorized')renderDiscoveryResult(run,{status:'error',message:error.message});} }
-  function renderDiscoveryRuns() { const el=$('directDiscoveryRunsState'); if(!discoveryRuns.length){el.innerHTML='<div class="discovery-empty">No discovery runs yet. Start with a market brief and approved public seed URLs.</div>';return;} el.innerHTML=discoveryRuns.map(run=>`<button class="discovery-run-row ${String(run.id)===String(selectedDiscoveryRunId)?'selected':''}" type="button" data-discovery-run-id="${esc(run.id)}"><span><strong>Run ${esc(run.id)}</strong><small>${esc(run.criteria?.keywords||run.criteria?.keyword||'Discovery brief')}</small></span><span class="discovery-run-status ${esc(discoveryStatus(run))}">${esc(discoveryLabel(discoveryStatus(run)))}</span><small>${esc(run.result_count??discoveryCandidates(run).length)} results</small></button>`).join(''); }
+  const selectedSourceIds = () => [...($('directDiscoverySources')?.selectedOptions||[])].map(option=>Number(option.value)).filter(Number.isInteger);
+  function renderDiscoveryResult(run, job) {
+    const el = $('directDiscoveryResultState');
+    const status = discoveryStatus(job || run);
+    const candidates = discoveryCandidates(run);
+    const result = run?.result || {};
+    const progress = Math.max(0, Math.min(100, Number(job?.progress ?? run.progress ?? 0)));
+    const events = job?.events || job?.event_timeline || run?.events || [];
+    const errors = run?.errors || result.errors || [];
+    $('directDiscoveryResultTitle').textContent = 'Run ' + run.id + ' · ' + (run.criteria?.keywords || run.criteria?.keyword || 'Discovery brief');
+    $('directDiscoveryResultStatus').textContent = discoveryLabel(status);
+    const controls = '<div class="discovery-controls"><button class="button ghost compact" type="button" data-run-action="pause" ' + (!['queued','running'].includes(status) ? 'disabled' : '') + '>Ⅱ Pause</button><button class="button ghost compact" type="button" data-run-action="resume" ' + (status !== 'paused' ? 'disabled' : '') + '>▶ Resume</button><button class="button danger compact" type="button" data-run-action="cancel" ' + (!['queued','running','paused'].includes(status) ? 'disabled' : '') + '>Cancel</button><span class="small-label">Pause/resume require the job-control API.</span></div>';
+    const sourceRows = run.sources || run.source_health || [];
+    const health = sourceRows.length ? '<section class="source-health"><div class="subheading"><h4>Source health</h4><span class="small-label">' + sourceRows.length + ' sources</span></div><div class="source-health-grid">' + sourceRows.map(source => '<article><strong>' + esc(source.name || source.source_name || ('Source ' + (source.source_id || ''))) + '</strong><span class="source-health-status ' + esc(source.status || source.health || 'unknown') + '">' + esc(source.status || source.health || 'Unknown') + '</span><small>' + esc(source.result_count ?? source.results ?? 0) + ' results · ' + esc(source.error || 'No errors') + '</small><div class="source-health-actions"><button class="button ghost compact" type="button" data-source-retry="' + esc(source.source_id || source.id || '') + '">Retry</button><button class="button ghost compact" type="button" data-source-circuit="' + esc(source.source_id || source.id || '') + '">Circuit</button></div></article>').join('') + '</div></section>' : '<section class="source-health"><div class="subheading"><h4>Source health</h4></div><p class="muted">Per-source health will appear when the API returns source telemetry.</p></section>';
+    const log = events.length ? events.map(event => '<li><span class="timeline-dot"></span><div><strong>' + esc(event.message || event.type || 'Run event') + '</strong><small>' + esc(event.created_at || event.timestamp || '') + '</small></div></li>').join('') : '<li class="muted">No persisted events returned yet.</li>';
+    const stats = '<div class="discovery-stat-grid"><div><small>Progress</small><strong>' + progress + '%</strong></div><div><small>Records</small><strong>' + esc(run.result_count ?? candidates.length) + '</strong></div><div><small>Daily limit</small><strong>' + esc(run.daily_limit ?? '—') + '</strong></div><div><small>Errors</small><strong>' + (Array.isArray(errors) ? errors.length : 0) + '</strong></div></div>';
+    const cards = candidates.length ? candidates.slice(0, 50).map(item => '<article class="discovery-candidate"><strong>' + esc(item.name || item.business_name || item.title || 'Unnamed candidate') + '</strong><span>' + esc(item.website || item.url || item.domain || 'No URL returned') + '</span><small>' + esc(item.source_url || item.source || 'Source provenance unavailable') + ' · ' + esc(item.confidence || item.score || 'Confidence unknown') + '</small></article>').join('') : '<p class="muted">No candidates returned by this run.</p>';
+    if (['running','queued','paused'].includes(status)) { el.innerHTML = controls + stats + '<p>' + esc(job?.progress_message || 'Processing approved sources within bounded limits.') + '</p><div class="progress-track"><span style="width:' + progress + '%"></span></div><div class="discovery-two-col"><section class="live-log"><h4>Live log</h4><ol>' + log + '</ol></section>' + health + '</div>'; return; }
+    const partial = status === 'partial' || run.partial === true || result.partial === true;
+    el.innerHTML = controls + (partial ? '<div class="discovery-partial" role="status"><strong>Partial results</strong><p>Review provenance before using any candidate.</p></div>' : '') + (status === 'failed' ? '<div class="detail-error" role="alert"><strong>Discovery failed</strong><p>' + esc(job?.error?.message || job?.message || run.error || 'The job returned no safe result.') + '</p></div>' : '') + stats + health + '<div class="discovery-two-col"><section class="live-log"><h4>Run log</h4><ol>' + log + '</ol></section><section><h4>Results</h4><div class="discovery-candidate-list">' + cards + '</div></section></div>';
+  }
+  async function discoveryAction(action,run){ if(!run?.id)return; if(action!=='cancel'&&!['pause','resume'].includes(action))return; if(action==='cancel'&&!window.confirm('Cancel this discovery run? The current partial result will remain reviewable.'))return; try{await jsonRequest(`/api/v1/discovery-runs/${encodeURIComponent(run.id)}/${action}`,{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({})});message('directDiscoveryMessage',`Discovery run ${action} requested.`);await loadDiscoveryRuns();}catch(error){if(error.message!=='unauthorized')message('directDiscoveryMessage',error.message,true);} }
+  async function loadDiscoveryRun(run) { selectedDiscoveryRunId=run.id; renderDiscoveryResult(run); try { const job=await jsonRequest(`/api/v1/jobs/${encodeURIComponent(run.job_id)}`); renderDiscoveryResult(run,job); if(['queued','running','paused'].includes(discoveryStatus(job))){clearTimeout(discoveryPollTimer);discoveryPollTimer=setTimeout(loadDiscoveryRuns,3000);} } catch(error){if(error.message!=='unauthorized')renderDiscoveryResult(run,{status:'error',message:error.message});} }
+  function renderDiscoveryRuns() { const el=$('directDiscoveryRunsState'); if(!discoveryRuns.length){el.innerHTML='<div class="discovery-empty">No discovery runs yet. Start with a market brief and approved sources.</div>';return;} el.innerHTML=discoveryRuns.map(run=>`<button class="discovery-run-row ${String(run.id)===String(selectedDiscoveryRunId)?'selected':''}" type="button" data-discovery-run-id="${esc(run.id)}"><span><strong>Run ${esc(run.id)}</strong><small>${esc(run.criteria?.keywords||run.criteria?.keyword||'Discovery brief')} · ${esc(run.criteria?.city||run.criteria?.location||'Global')}</small></span><span class="discovery-run-status ${esc(discoveryStatus(run))}">${esc(discoveryLabel(discoveryStatus(run)))}</span><small>${esc(run.result_count??discoveryCandidates(run).length)} records</small></button>`).join(''); }
  async function loadDiscoveryRuns() { const state=$('directDiscoveryRunsState'); if(state)state.innerHTML='<div class="detail-loading" aria-live="polite">Loading discovery runs…</div>'; try { const payload=await jsonRequest('/api/v1/discovery-runs?page_size=50'); discoveryRuns=payloadItems(payload,['items','runs']); const jobsById=new Map(await Promise.all(discoveryRuns.map(async run=>{try{return [String(run.job_id),await jsonRequest(`/api/v1/jobs/${encodeURIComponent(run.job_id)}`)]}catch{return [String(run.job_id),{}]}}))); discoveryRuns=discoveryRuns.map(run=>({...run,status:jobsById.get(String(run.job_id))?.status||run.status||'queued'})); renderDiscoveryRuns(); const selected=discoveryRuns.find(run=>String(run.id)===String(selectedDiscoveryRunId))||discoveryRuns[0]; if(selected)loadDiscoveryRun(selected); } catch(error) { if(error.message!=='unauthorized'&&state)state.innerHTML=`<div class="detail-error" role="alert"><strong>Unable to load discovery runs</strong><p>${esc(error.message)}</p><button class="button ghost compact" id="retryDiscoveryRunsBtn" type="button">Try again</button></div>`; } }
-  async function submitDirectDiscovery(event) { event.preventDefault(); const form=event.currentTarget, button=$('directDiscoveryRunBtn'), fields=Object.fromEntries(new FormData(form).entries()), seeds=fields.seed_urls.split(/\r?\n|,/).map(value=>value.trim()).filter(Boolean); if(!fields.keywords.trim()||!seeds.length||seeds.length>5){message('directDiscoveryMessage','Enter criteria and 1–5 seed URLs.',true);return;} button.disabled=true; message('directDiscoveryMessage','Starting bounded discovery job…'); try { const job=await jsonRequest('/api/v1/discovery',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({criteria:{keywords:fields.keywords.trim(),location:fields.location.trim()},seed_urls:seeds,max_pages:Number(fields.max_pages),max_candidates:Number(fields.max_candidates),idempotency_key:`discovery-${Date.now()}-${Math.random().toString(36).slice(2)}`})}); message('directDiscoveryMessage','Discovery accepted. Tracking job status and partial results below.'); await loadDiscoveryRuns(); const run=discoveryRuns.find(item=>String(item.job_id)===String(job.id)); if(run)loadDiscoveryRun(run); } catch(error) { if(error.message!=='unauthorized')message('directDiscoveryMessage',error.message||'Discovery request failed.',true); } finally {button.disabled=false;} }
+  async function submitDirectDiscovery(event) { event.preventDefault(); const form=event.currentTarget, button=$('directDiscoveryRunBtn'), fields=Object.fromEntries(new FormData(form).entries()), seeds=fields.seed_urls.split(/\r?\n|,/).map(value=>value.trim()).filter(Boolean), dryRun=$('directDiscoveryDryRun')?.checked; if(!fields.keywords.trim()){message('directDiscoveryMessage','Enter a category or keyword.',true);return;} if(seeds.length>5){message('directDiscoveryMessage','Use no more than 5 public seed URLs.',true);return;} if(dryRun){message('directDiscoveryMessage',`Plan validated: ${selectedSourceIds().length||'criteria-first'} source selection, max ${fields.max_candidates} records, ${fields.daily_limit}/day. No job created.`);return;} button.disabled=true; message('directDiscoveryMessage','Starting bounded discovery job…'); try { const criteria={keywords:fields.keywords.trim(),city:fields.city.trim(),province:fields.province.trim(),country:fields.country.trim()}; const payload={criteria,source_ids:selectedSourceIds(),daily_limit:Number(fields.daily_limit),schedule:fields.schedule,max_pages:Number(fields.max_pages),max_candidates:Number(fields.max_candidates),idempotency_key:`discovery-${Date.now()}-${Math.random().toString(36).slice(2)}`}; if(seeds.length)payload.seed_urls=seeds; const job=await jsonRequest('/api/v1/discovery',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify(payload)}); message('directDiscoveryMessage','Discovery accepted. Tracking job status and partial results below.'); await loadDiscoveryRuns(); const run=discoveryRuns.find(item=>String(item.job_id)===String(job.id)); if(run)loadDiscoveryRun(run); } catch(error) { if(error.message!=='unauthorized')message('directDiscoveryMessage',error.message||'Discovery request failed.',true); } finally {button.disabled=false;} }

  function parseCsv(text){const lines=text.trim().split(/\r?\n/).filter(Boolean),cells=line=>line.match(/("[^"]*(?:""[^"]*)*"|[^,]+)(?=,|$)/g)?.map(x=>x.replace(/^"|"$/g,'').replaceAll('""','"'))||[];if(!lines.length)return[];const headers=cells(lines[0]);return lines.slice(1,11).map(l=>Object.fromEntries(cells(l).map((v,i)=>[headers[i]||`column_${i+1}`,v])));}
  function renderCsv(rows){if(!rows.length){$('csvPreview').innerHTML='<span>⊞</span><p>No data rows found</p>';return;}const h=Object.keys(rows[0]);$('csvPreview').className='csv-table';$('csvPreview').innerHTML=`<table><thead><tr>${h.map(x=>`<th>${esc(x)}</th>`).join('')}</tr></thead><tbody>${rows.map(r=>`<tr>${h.map(x=>`<td>${esc(r[x])}</td>`).join('')}</tr>`).join('')}</tbody></table><small class="muted">Showing up to 10 rows · Preview only; nothing added yet.</small>`;}
@@ -300,6 +322,8 @@
  document.addEventListener('click',e=>{if(e.target.id==='verifyBtn')verify();if(e.target.id==='recalculateScoreBtn')recalculateScore();if(e.target.id==='generateAiSuggestionBtn')generateAiSuggestion();if(e.target.id==='createOutreachDraftBtn')createOutreachDraft();const outreachButton=e.target.closest?.('[data-outreach-action]');if(outreachButton)approveOutreachDraft(outreachButton.dataset.outreachId);const aiButton=e.target.closest?.('[data-ai-action]');if(aiButton)aiAction(aiButton.dataset.aiAction,aiButton.dataset.aiId);const scoreEdit=e.target.closest?.('[data-score-edit]');if(scoreEdit)updateScoreRule(scoreEdit.dataset.scoreEdit,scoreEdit.dataset.scorePoints);if(e.target.id==='extractContactsBtn'&&selectedId)loadContactExtraction(selectedId,{extract:true});if(e.target.id==='refreshContactExtractionBtn'&&selectedId)loadContactExtraction(selectedId,{extract:true});if(e.target.id==='retryContactExtractionBtn'&&selectedId)loadContactExtraction(selectedId);if(e.target.id==='scanWebsiteBtn'&&selectedId)loadWebsiteScan(selectedId,{scan:true});if(e.target.id==='refreshWebsiteScanBtn'&&selectedId)loadWebsiteScan(selectedId);if(e.target.id==='retryWebsiteScanBtn'&&selectedId)loadWebsiteScan(selectedId);if(e.target.id==='runDomainCheckBtn')runDomainCheck();if(e.target.id==='retryDomainBtn'&&selectedId)loadDomainIntelligence(selectedId);const availability=e.target.closest?.('[data-domain-availability]');if(availability)checkDomainAvailability(availability.dataset.domain,availability);if(e.target.id==='retryDetailBtn'&&selectedId)loadDetail(selectedId);if(e.target.id==='retryDedupBtn'&&selectedId)loadMatchSuggestions(selectedId);if(e.target.id==='retryHistoryBtn'&&selectedId)loadMergeHistory(selectedId);if(e.target.id==='cancelMergeBtn'||e.target.id==='cancelMergeBtnSecondary')closeMergeDialog();if(e.target.id==='confirmMergeBtn')confirmMerge();const mergeButton=e.target.closest?.('[data-merge-target]');if(mergeButton)openMergeDialog(mergeButton.dataset.mergeTarget,mergeButton.dataset.mergeTargetName);const reverseButton=e.target.closest?.('[data-reverse-merge]');if(reverseButton)reverseMerge(reverseButton.dataset.reverseMerge);if(e.target.id==='retryJobDetailBtn'&&selectedJobId)loadJobDetail(selectedJobId);if(e.target.id==='cancelJobBtn')jobAction('cancel');if(e.target.id==='retryJobBtn')jobAction('retry');const row=e.target.closest?.('[data-job-id]');if(row)loadJobDetail(row.dataset.jobId);});
  $('loginForm').addEventListener('submit',login);$('logoutBtn').addEventListener('click',logout);$('directDiscoveryForm').addEventListener('submit',submitDirectDiscovery);$('directDiscoveryRefreshBtn').addEventListener('click',loadDiscoveryRuns);$('directDiscoveryRunsState').addEventListener('click',e=>{const row=e.target.closest?.('[data-discovery-run-id]');if(row){const run=discoveryRuns.find(item=>String(item.id)===String(row.dataset.discoveryRunId));if(run)loadDiscoveryRun(run);}if(e.target.id==='retryDiscoveryRunsBtn')loadDiscoveryRuns();});$('searchInput').addEventListener('input',()=>{page=1;renderRows();});['scoreFilter','statusFilter','websiteClassFilter','pipelineFilter'].forEach(id=>$(id).addEventListener('change',()=>{page=1;loadData();}));$('pageSize').addEventListener('change',e=>{pageSize=Number(e.target.value);page=1;loadData();});$('nextPageBtn').addEventListener('click',()=>{if(hasNextPage){page+=1;loadData();}});$('refreshBtn').addEventListener('click',loadData);$('jobsRefreshBtn').addEventListener('click',()=>loadJobs());$('startDemoJobBtn').addEventListener('click',startDemoJob);$('sourcesRefreshBtn').addEventListener('click',loadSources);$('sourceForm').addEventListener('submit',saveSource);$('sourceType').addEventListener('change',e=>{$('sourceCsvField').hidden=e.target.value!=='csv';});$('discoveryForm').addEventListener('submit',e=>{e.preventDefault();runDiscovery(true);});$('discoveryRunBtn').addEventListener('click',()=>runDiscovery(false));$('sourcesList').addEventListener('click',e=>{const button=e.target.closest?.('[data-source-action]');if(button)sourceAction(button.dataset.sourceId,button.dataset.sourceAction);});$('addForm').addEventListener('submit',addProspect);$('csvInput').addEventListener('change',e=>{const file=e.target.files[0];if(file){const reader=new FileReader();reader.onload=()=>renderCsv(parseCsv(reader.result));reader.readAsText(file);}});$('menuBtn').addEventListener('click',()=>document.querySelector('.sidebar').classList.toggle('open'));document.querySelectorAll('[data-scroll]').forEach(b=>b.addEventListener('click',()=>document.querySelector(b.dataset.scroll)?.scrollIntoView()));
  $('savedFilterForm').addEventListener('submit',saveCurrentFilters);$('savedFilterSelect').addEventListener('change',e=>{ $('deleteSavedFilterBtn').disabled=!e.target.value;if(e.target.value)loadSavedFilter(e.target.value);});$('deleteSavedFilterBtn').addEventListener('click',deleteSavedFilter);$('reviewQueueState').addEventListener('change',e=>{const input=e.target.closest?.('[data-review-id]');if(input){if(input.checked)selectedReviewIds.add(String(input.dataset.reviewId));else selectedReviewIds.delete(String(input.dataset.reviewId));updateBulkState();}});$('selectAllReview').addEventListener('change',e=>{reviewQueue.forEach(p=>e.target.checked?selectedReviewIds.add(String(p.id)):selectedReviewIds.delete(String(p.id)));renderReviewQueue({items:reviewQueue,count:$('reviewQueueCount').textContent});});$('bulkVerifyBtn').addEventListener('click',()=>bulkReview('verify'));$('bulkRejectBtn').addEventListener('click',()=>bulkReview('reject'));$('reviewQueueState').addEventListener('click',e=>{if(e.target.id==='retryReviewQueueBtn')loadReviewQueue();});document.querySelectorAll('[data-dashboard-filter]').forEach(card=>card.addEventListener('click',()=>{const kind=card.dataset.dashboardFilter;if(kind==='review'||kind==='suppressed')applyFilters({status:kind});else if(kind==='high')applyFilters({score:'high'});else if(kind==='fresh')applyFilters({});else applyFilters({status:'all',score:'all'});}));
-  $('interactionForm').addEventListener('submit',saveInteraction);$('suppressionForm').addEventListener('submit',addSuppression);$('aiProviderForm').addEventListener('submit',saveAiProviderSettings);$('aiProvider').addEventListener('change',updateAiProviderForm);$('aiProviderRefreshBtn').addEventListener('click',loadAiProviderSettings);$('testAiProviderBtn').addEventListener('click',testAiProviderConnection);$('crmRefreshBtn').addEventListener('click',()=>{loadCrmPipeline();loadInteractions(selectedId);});$('reportsRefreshBtn').addEventListener('click',loadReports);$('suppressionRefreshBtn').addEventListener('click',loadSuppressions);$('pipelineViewToggle').addEventListener('click',()=>{crmListMode=!crmListMode;$('pipelineViewToggle').textContent=crmListMode?'▦ Board view':'☷ List view';$('pipelineViewToggle').setAttribute('aria-pressed',String(crmListMode));renderPipeline();});$('pipelineBoard').addEventListener('click',e=>{const selectButton=e.target.closest?.('[data-crm-select]');if(selectButton){selectedId=Number(selectButton.dataset.crmSelect);selectProspect(selectedId);loadInteractions(selectedId);$('crmActivity').scrollIntoView({behavior:'smooth',block:'start'});}const stageButton=e.target.closest?.('[data-crm-save-stage]');if(stageButton)saveCrmStage(stageButton.dataset.crmSaveStage);if(e.target.id==='retryCrmBtn')loadCrmPipeline();if(e.target.id==='retryInteractionsBtn'&&selectedId)loadInteractions(selectedId);});$('suppressionState').addEventListener('change',e=>{const input=e.target.closest?.('[data-suppression-id]');if(input){if(input.checked)selectedSuppressionIds.add(String(input.dataset.suppressionId));else selectedSuppressionIds.delete(String(input.dataset.suppressionId));updateSuppressionSelection();}});$('suppressionState').addEventListener('click',e=>{const button=e.target.closest?.('[data-remove-suppression]');if(button)removeSuppression(button.dataset.removeSuppression);if(e.target.id==='retrySuppressionsBtn')loadSuppressions();});$('selectAllSuppressions').addEventListener('change',e=>{suppressions.forEach(item=>e.target.checked?selectedSuppressionIds.add(String(item.id)):selectedSuppressionIds.delete(String(item.id)));renderSuppressions();});$('bulkReviewSuppressionsBtn').addEventListener('click',bulkReviewSuppressions);$('outreachPolicyRefreshBtn').addEventListener('click',loadProviderPolicy);
+  ['sourceFilter','geographyFilter','categoryFilter','contactStatusFilter','scoreFilter','statusFilter','websiteClassFilter','pipelineFilter'].forEach(id=>$(id)?.addEventListener('change',()=>{page=1;loadData();}));
+  $('interactionForm').addEventListener('submit',saveInteraction);$('suppressionForm').addEventListener('submit',addSuppression);$('aiProviderForm').addEventListener('submit',saveAiProviderSettings);$('aiProvider').addEventListener('change',updateAiProviderForm);$('aiProviderRefreshBtn').addEventListener('click',loadAiProviderSettings);$('testAiProviderBtn').addEventListener('click',testAiProviderConnection);$('crmRefreshBtn').addEventListener('click',()=>{loadCrmPipeline();loadInteractions(selectedId);});$('reportsRefreshBtn').addEventListener('click',loadReports);$('suppressionRefreshBtn').addEventListener('click',loadSuppressions);$('pipelineViewToggle').addEventListener('click',()=>{crmListMode=!crmListMode;$('pipelineViewToggle').textContent=crmListMode?'▦ Board view':'☷ List view';$('pipelineViewToggle').setAttribute('aria-pressed',String(crmListMode));renderPipeline();});$('pipelineBoard').addEventListener('click',e=>{const selectButton=e.target.closest?.('[data-crm-select]');if(selectButton){selectedId=Number(selectButton.dataset.crmSelect);selectProspect(selectedId);loadInteractions(selectedId);$('crmActivity').scrollIntoView({behavior:'smooth',block:'start'});}const stageButton=e.target.closest?.('[data-crm-save-stage]');if(stageButton)saveCrmStage(stageButton.dataset.crmSaveStage);if(e.target.id==='retryCrmBtn')loadCrmPipeline();if(e.target.id==='retryInteractionsBtn'&&selectedId)loadInteractions(selectedId);});$('suppressionState').addEventListener('change',e=>{const input=e.target.closest?.('[data-suppression-id]');if(input){if(input.checked)selectedSuppressionIds.add(String(input.dataset.suppressionId));else selectedSuppressionIds.delete(String(input.dataset.suppressionId));updateSuppressionSelection();}});$('suppressionState').addEventListener('click',e=>{const button=e.target.closest?.('[data-remove-suppression]');if(button)removeSuppression(button.dataset.removeSuppression);if(e.target.id==='retrySuppressionsBtn')loadSuppressions();});$('selectAllSuppressions').addEventListener('change',e=>{suppressions.forEach(item=>e.target.checked?selectedSuppressionIds.add(String(item.id)):selectedSuppressionIds.delete(String(item.id)));renderSuppressions();});$('bulkReviewSuppressionsBtn').addEventListener('click',bulkReviewSuppressions);$('discoveryWorkspace').addEventListener('click',e=>{const action=e.target.closest?.('[data-run-action]');if(action){const run=discoveryRuns.find(item=>String(item.id)===String(selectedDiscoveryRunId));discoveryAction(action.dataset.runAction,run);}});
+$('aiProviderForm').addEventListener('submit',saveAiProviderSettings);$('aiProvider').addEventListener('change',updateAiProviderForm);$('aiProviderRefreshBtn').addEventListener('click',loadAiProviderSettings);$('testAiProviderBtn').addEventListener('click',testAiProviderConnection);$('crmRefreshBtn').addEventListener('click',()=>{loadCrmPipeline();loadInteractions(selectedId);});$('reportsRefreshBtn').addEventListener('click',loadReports);$('suppressionRefreshBtn').addEventListener('click',loadSuppressions);$('pipelineViewToggle').addEventListener('click',()=>{crmListMode=!crmListMode;$('pipelineViewToggle').textContent=crmListMode?'▦ Board view':'☷ List view';$('pipelineViewToggle').setAttribute('aria-pressed',String(crmListMode));renderPipeline();});$('pipelineBoard').addEventListener('click',e=>{const selectButton=e.target.closest?.('[data-crm-select]');if(selectButton){selectedId=Number(selectButton.dataset.crmSelect);selectProspect(selectedId);loadInteractions(selectedId);$('crmActivity').scrollIntoView({behavior:'smooth',block:'start'});}const stageButton=e.target.closest?.('[data-crm-save-stage]');if(stageButton)saveCrmStage(stageButton.dataset.crmSaveStage);if(e.target.id==='retryCrmBtn')loadCrmPipeline();if(e.target.id==='retryInteractionsBtn'&&selectedId)loadInteractions(selectedId);});$('suppressionState').addEventListener('change',e=>{const input=e.target.closest?.('[data-suppression-id]');if(input){if(input.checked)selectedSuppressionIds.add(String(input.dataset.suppressionId));else selectedSuppressionIds.delete(String(input.dataset.suppressionId));updateSuppressionSelection();}});$('suppressionState').addEventListener('click',e=>{const button=e.target.closest?.('[data-remove-suppression]');if(button)removeSuppression(button.dataset.removeSuppression);if(e.target.id==='retrySuppressionsBtn')loadSuppressions();});$('selectAllSuppressions').addEventListener('change',e=>{suppressions.forEach(item=>e.target.checked?selectedSuppressionIds.add(String(item.id)):selectedSuppressionIds.delete(String(item.id)));renderSuppressions();});$('bulkReviewSuppressionsBtn').addEventListener('click',bulkReviewSuppressions);$('outreachPolicyRefreshBtn').addEventListener('click',loadProviderPolicy);
  bootstrap();
 })();
@@ -1,6 +1,6 @@
 {
  "schema": 1,
-  "version": "phase-18",
+  "version": "phase-19",
  "entrypoints": [
    "config.js",
    "app.js",
@@ -14,9 +14,9 @@
  ],
  "integrity": {
    "config.js": "sha256-f9c7b4db3eab4cf54146bd25891b5103b09ae75da93c57b548cf57ae93e4a3f6",
-    "app.js": "sha256-b8e1708b5a35123b645d85cddc836cae96424e18ab1f282e64f02a500ded3f0f",
-    "styles.css": "sha256-ca618b1a44a1f310029ab9d6986a4120e9147cd61784c3e7a0ff826cd2d42696",
-    "index.html": "sha256-7a17c1568c471db9ed2cccae1529004e26c4a2abc58f4a83a592a702c5613726",
+    "app.js": "sha256-ab3d9046b53fd87950bedeec66da479da8b935bc070b60490a1aab7b22fff2a8",
+    "styles.css": "sha256-423d0f9489061aff6420bea3d854e49aaf1c35d94a8c0e0bca2f82434781a2b7",
+    "index.html": "sha256-1979f265a29009f7bd5380401c2185c6473394eb2412b592c50ca90271cb6d3f",
    "health.html": "sha256-c352a6f37aa24628cfc8d5709a70ff2d94181d192ed5fe00916cca9378a61d81",
    "error.html": "sha256-f3cc28d2dfc9e8af112d0257c6ee47b9e9146e5da8590a754dd589bf43702aaf",
    "healthz": "sha256-dc51b8c96c2d745df3bd5590d990230a482fd247123599548e0632fdbf97fc22"
@@ -5,7 +5,7 @@
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>ProspectOS · Pipeline intelligence</title>
  <meta name="description" content="Prospect discovery and review dashboard">
-  <link rel="stylesheet" href="styles.css?v=phase-18">
+  <link rel="stylesheet" href="styles.css?v=phase-19">
 </head>
 <body>
  <section class="login-screen" id="loginScreen" aria-labelledby="loginTitle">
@@ -30,6 +30,7 @@
        <a class="nav-item active" href="#dashboard"><span>▦</span> Dashboard</a>
        <a class="nav-item" href="#explorer"><span>⌕</span> Prospect explorer</a>
        <a class="nav-item" href="#add"><span>＋</span> Add prospects</a>
+        <a class="nav-item" href="#discoveryWorkspace" data-nav="discovery"><span>⌁</span> Discovery runs</a>
        <a class="nav-item" href="#jobs" data-nav="jobs"><span>◷</span> Jobs</a>
        <a class="nav-item" href="#sources" data-nav="sources"><span>⌁</span> Sources</a>
        <a class="nav-item" href="#crmPipeline" data-nav="crm"><span>◫</span> CRM pipeline</a>
@@ -46,14 +47,14 @@
      <div class="content">
        <section class="hero" id="dashboard"><div><p class="eyebrow">EVIDENCE-LED PROSPECTING</p><h1>Good morning, Alex <span>✦</span></h1><p class="hero-sub">Your pipeline has <strong id="heroCount">0 prospects</strong> ready for review.</p></div><button class="button primary" data-scroll="#add">＋ Add prospect</button></section>
        <section class="direct-discovery-section" id="discoveryWorkspace" aria-labelledby="discoveryWorkspaceTitle" data-smoke="direct-discovery">
-          <div class="discovery-hero panel"><div><p class="eyebrow">BUILT-IN DISCOVERY</p><h2 id="discoveryWorkspaceTitle">Find prospects from approved seed sites</h2><p class="muted">Describe the market, provide up to five public seed URLs, and let the bounded workspace job return reviewable candidates. No manual source registration required.</p></div><span class="discovery-badge">Bounded · review first</span></div>
+          <div class="discovery-hero panel"><div><p class="eyebrow">SOURCE INTELLIGENCE</p><h2 id="discoveryWorkspaceTitle">Discovery runs</h2><p class="muted">Build a criteria-first run across approved sources, monitor it live, and review every result with provenance before it enters your pipeline.</p></div><span class="discovery-badge">Bounded · review first</span></div>
          <div class="discovery-workspace-grid">
-            <article class="panel direct-discovery-form-panel"><div class="panel-heading"><div><p class="eyebrow">DISCOVERY BRIEF</p><h3>Set your criteria</h3></div><span class="small-label">Max 20 pages · 50 candidates</span></div>
-              <form id="directDiscoveryForm"><label>Keywords or market<input id="directDiscoveryKeywords" name="keywords" required maxlength="240" placeholder="e.g. solar installers, Cape Town"></label><label>Location <span class="optional">optional</span><input id="directDiscoveryLocation" name="location" maxlength="120" placeholder="Western Cape, South Africa"></label><label>Seed URLs <span class="optional">up to 5 · one per line</span><textarea id="directDiscoverySeeds" name="seed_urls" rows="4" required placeholder="https://directory.example/&#10;https://association.example/members"></textarea></label><div class="discovery-limit-grid"><label>Page limit<select id="directDiscoveryMaxPages" name="max_pages"><option value="5">5 pages</option><option value="10" selected>10 pages</option><option value="20">20 pages</option></select></label><label>Candidate limit<select id="directDiscoveryMaxCandidates" name="max_candidates"><option value="10">10 candidates</option><option value="25" selected>25 candidates</option><option value="50">50 candidates</option></select></label></div><div class="form-footer"><p id="directDiscoveryMessage" class="form-message" role="status" aria-live="polite"></p><button class="button primary" id="directDiscoveryRunBtn" type="submit">Run discovery <span aria-hidden="true">→</span></button></div></form>
+            <article class="panel direct-discovery-form-panel"><div class="panel-heading"><div><p class="eyebrow">RUN BUILDER</p><h3>Define your market</h3></div><span class="small-label">Max 20 pages · 50 candidates</span></div>
+              <form id="directDiscoveryForm"><div class="form-grid"><label>Category / keyword<input id="directDiscoveryKeywords" name="keywords" required maxlength="240" placeholder="e.g. solar installers"></label><label>City<input id="directDiscoveryCity" name="city" maxlength="80" placeholder="Cape Town"></label><label>Province / state<input id="directDiscoveryProvince" name="province" maxlength="80" placeholder="Western Cape"></label><label>Country<input id="directDiscoveryCountry" name="country" maxlength="80" placeholder="South Africa"></label></div><label>Approved sources <span class="optional">multi-select · only enabled sources can run</span><select id="directDiscoverySources" name="source_ids" multiple size="4" aria-label="Approved discovery sources"><option value="">Loading approved sources…</option></select></label><div class="discovery-limit-grid"><label>Max records<select id="directDiscoveryMaxCandidates" name="max_candidates"><option value="10">10 records</option><option value="25" selected>25 records</option><option value="50">50 records</option></select></label><label>Daily limit<input id="directDiscoveryDailyLimit" name="daily_limit" type="number" min="1" max="1000" value="100"></label><label>Schedule<select id="directDiscoverySchedule" name="schedule"><option value="now">Run now</option><option value="daily">Daily · 09:00</option><option value="weekdays">Weekdays · 09:00</option><option value="weekly">Weekly · Monday</option></select></label><label>Page limit<select id="directDiscoveryMaxPages" name="max_pages"><option value="5">5 pages</option><option value="10" selected>10 pages</option><option value="20">20 pages</option></select></label></div><details class="advanced-discovery"><summary>Advanced run options</summary><label>Public seed URLs <span class="optional">optional · up to 5, one per line</span><textarea id="directDiscoverySeeds" name="seed_urls" rows="3" placeholder="https://directory.example/"></textarea></label><label class="checkbox-label"><input type="checkbox" name="dry_run" id="directDiscoveryDryRun"> Dry run · validate the plan without creating a job</label></details><div class="form-footer"><p id="directDiscoveryMessage" class="form-message" role="status" aria-live="polite"></p><button class="button primary" id="directDiscoveryRunBtn" type="submit">Start discovery <span aria-hidden="true">→</span></button></div></form>
            </article>
            <aside class="panel discovery-runs-panel"><div class="panel-heading"><div><p class="eyebrow">RUN HISTORY</p><h3>Discovery runs</h3></div><button class="button ghost compact" id="directDiscoveryRefreshBtn" type="button">↻ Refresh</button></div><div id="directDiscoveryRunsState" class="discovery-runs-state" aria-live="polite"><div class="detail-loading">Sign in to load discovery runs.</div></div></aside>
          </div>
-          <div class="panel discovery-results-panel" id="directDiscoveryResultsPanel"><div class="panel-heading"><div><p class="eyebrow">RESULT STATUS</p><h3 id="directDiscoveryResultTitle">Select a run to inspect results</h3></div><span id="directDiscoveryResultStatus" class="small-label">No run selected</span></div><div id="directDiscoveryResultState" aria-live="polite"><p class="muted">Results appear here with job state, partial-result handling, and candidate provenance.</p></div></div>
+          <div class="panel discovery-results-panel" id="directDiscoveryResultsPanel"><div class="panel-heading"><div><p class="eyebrow">ACTIVE RUN</p><h3 id="directDiscoveryResultTitle">Select a run to inspect results</h3></div><span id="directDiscoveryResultStatus" class="small-label">No run selected</span></div><div id="directDiscoveryResultState" aria-live="polite"><p class="muted">Progress, metrics, live events, and per-source health appear here.</p></div></div>
        </section>
        <section class="metrics" aria-label="Dashboard metrics">
          <a class="metric-card metric-link" data-dashboard-filter="all" href="#explorer"><div class="metric-icon violet">◎</div><div><p>Total prospects</p><h2 id="metricTotal">0</h2><span class="trend neutral">● Current workspace</span></div></a>
@@ -66,7 +67,7 @@
        <section class="score-rules-section" id="scoreRules" aria-labelledby="scoreRulesTitle"><article class="panel" id="scoreRulesPanel" data-smoke="score-rules"><div class="detail-loading" aria-live="polite">Sign in to load score rules…</div></article></section>
        <section class="workspace-grid" id="explorer">
          <div class="explorer-panel panel"><div class="panel-heading"><div><p class="eyebrow">PIPELINE</p><h2>Prospect explorer</h2></div><button class="button ghost" id="refreshBtn">↻ Refresh</button></div>
-            <div class="filters"><label class="search-wrap"><span>⌕</span><input id="searchInput" type="search" placeholder="Search companies, domains, locations…" autocomplete="off"></label><select id="scoreFilter" aria-label="Filter by score"><option value="all">All scores</option><option value="high">High fit · 80+</option><option value="medium">Medium · 60–79</option><option value="low">Low · under 60</option></select><select id="statusFilter" aria-label="Filter by status"><option value="all">All statuses</option><option value="review">Needs review</option><option value="reviewed">Reviewed</option><option value="suppressed">Suppressed</option></select><select id="websiteClassFilter" aria-label="Filter by website class"><option value="all">All website classes</option><option value="business_site">Business site</option><option value="social_profile">Social profile</option><option value="missing">Missing website</option></select><select id="pipelineFilter" aria-label="Filter by pipeline stage"><option value="all">All pipeline stages</option><option value="new">New</option><option value="contacted">Contacted</option><option value="qualified">Qualified</option><option value="proposal">Proposal</option><option value="negotiation">Negotiation</option><option value="won">Won</option><option value="lost">Lost</option></select></div>
+            <div class="filters"><label class="search-wrap"><span>⌕</span><input id="searchInput" type="search" placeholder="Search companies, domains, locations…" autocomplete="off"></label><select id="sourceFilter" aria-label="Filter by source"><option value="all">All sources</option></select><select id="geographyFilter" aria-label="Filter by geography"><option value="all">All geographies</option></select><select id="categoryFilter" aria-label="Filter by category"><option value="all">All categories</option></select><select id="websiteClassFilter" aria-label="Filter by website or domain status"><option value="all">Website / domain status</option><option value="business_site">Business site</option><option value="social_profile">Social profile</option><option value="missing">Missing website</option></select><select id="contactStatusFilter" aria-label="Filter by contact status"><option value="all">All contact status</option><option value="present">Contact present</option><option value="missing">Contact missing</option><option value="suppressed">Do not contact</option></select><select id="scoreFilter" aria-label="Filter by score"><option value="all">All scores</option><option value="high">High fit · 80+</option><option value="medium">Medium · 60–79</option><option value="low">Low · under 60</option></select><select id="statusFilter" aria-label="Filter by review status"><option value="all">All review states</option><option value="review">Needs review</option><option value="reviewed">Reviewed</option><option value="suppressed">Suppressed</option></select><select id="pipelineFilter" aria-label="Filter by pipeline stage"><option value="all">All pipeline stages</option><option value="new">New</option><option value="contacted">Contacted</option><option value="qualified">Qualified</option><option value="proposal">Proposal</option><option value="negotiation">Negotiation</option><option value="won">Won</option><option value="lost">Lost</option></select></div>
            <div class="saved-view-controls" data-smoke="saved-views"><form id="savedFilterForm" class="inline-form"><input id="savedFilterName" name="name" placeholder="Save current filters as…" maxlength="80" required><button class="button ghost compact" type="submit">Save view</button></form><select id="savedFilterSelect" aria-label="Load saved view"><option value="">Saved views</option></select><button class="button ghost compact" id="deleteSavedFilterBtn" type="button" disabled>Delete view</button><p id="savedFilterMessage" class="form-message" role="status" aria-live="polite"></p></div>
            <section class="review-queue" data-smoke="review-queue"><div class="queue-heading"><div><p class="eyebrow">OPERATOR QUEUE</p><h3>Review queue <span class="count" id="reviewQueueCount">0</span></h3></div><span class="small-label">Human decision required</span></div><div id="reviewQueueState" class="queue-state" aria-live="polite">Loading review queue…</div><div class="bulk-actions"><label class="checkbox-label"><input id="selectAllReview" type="checkbox"> Select visible</label><span id="selectedReviewCount" class="small-label">0 selected</span><button class="button primary compact" id="bulkVerifyBtn" type="button" disabled>Verify selected</button><button class="button danger compact" id="bulkRejectBtn" type="button" disabled>Reject selected</button></div></section>
            <div class="table-meta"><span id="resultCount">Showing 0 prospects</span><span class="legend"><span class="legend-dot high-dot"></span> High fit <span class="legend-dot review-dot"></span> Needs review</span><label class="page-size">Rows <select id="pageSize" aria-label="Page size"><option>10</option><option>25</option><option>50</option></select></label><button class="button ghost compact" id="nextPageBtn" type="button">Next page →</button></div>
@@ -129,7 +130,7 @@
  <div class="merge-dialog" id="mergeDialog" hidden role="dialog" aria-modal="true" aria-labelledby="mergeDialogTitle">
    <div class="merge-dialog-card"><div class="panel-heading"><div><p class="eyebrow">REVIEW REQUIRED</p><h2 id="mergeDialogTitle">Confirm merge</h2></div><button class="icon-button" id="cancelMergeBtn" type="button" aria-label="Close merge confirmation">×</button></div><p id="mergeDialogCopy"></p><div class="merge-warning"><strong>This action is reversible.</strong> The merge will be recorded in history and can be reversed later.</div><p id="mergeDialogMessage" class="form-message" role="alert" aria-live="polite"></p><div class="merge-dialog-actions"><button class="button ghost" id="cancelMergeBtnSecondary" type="button">Cancel</button><button class="button primary" id="confirmMergeBtn" type="button">Confirm merge</button></div></div>
  </div>
-  <script src="config.js?v=phase-18"></script>
-  <script src="app.js?v=phase-18"></script>
+  <script src="config.js?v=phase-19"></script>
+  <script src="app.js?v=phase-19"></script>
 </body>
 </html>
@@ -57,7 +57,7 @@ const expectedIds = [
  'loginScreen', 'loginForm', 'loginEmail', 'loginPassword', 'dashboardShell', 'logoutBtn', 'apiStatus',
  'explorer', 'detailPanel', 'reviewQueueCount', 'reviewQueueState', 'savedFilterForm', 'savedFilterSelect',
  'deleteSavedFilterBtn', 'bulkVerifyBtn', 'bulkRejectBtn', 'nextPageBtn', 'jobs', 'jobsList', 'jobDetailPanel',
-  'sources', 'sourcesList', 'sourceRecordsList', 'discoveryForm', 'discoveryWorkspace', 'directDiscoveryForm', 'directDiscoveryRunsState', 'directDiscoveryResultState', 'crmPipeline', 'pipelineBoard', 'crmActivity',
+  'sources', 'sourcesList', 'sourceRecordsList', 'discoveryForm', 'discoveryWorkspace', 'directDiscoveryForm', 'directDiscoveryRunsState', 'directDiscoveryResultState', 'directDiscoverySources', 'directDiscoveryDailyLimit', 'directDiscoverySchedule', 'directDiscoveryDryRun', 'sourceFilter', 'geographyFilter', 'categoryFilter', 'contactStatusFilter', 'crmPipeline', 'pipelineBoard', 'crmActivity',
  'interactionState', 'interactionForm', 'crmReports', 'pipelineReport', 'outcomesReport', 'activityReport',
  'suppressionCenter', 'suppressionForm', 'suppressionState', 'outreachSettings', 'providerPolicyPanel', 'aiProviderSettings', 'aiProviderForm', 'aiProviderStatus', 'aiProviderRefreshBtn', 'testAiProviderBtn', 'saveAiProviderBtn',
  'scoreRules', 'scoreRulesPanel', 'scoreDistributionPanel', 'mergeDialog', 'confirmMergeBtn'
@@ -80,6 +80,8 @@ const routeContracts = [
 ];
 check('routes.contracts', 'all critical API route contracts are referenced by the client', all(routeContracts, route => js.includes(route)), listMissing(routeContracts, route => js.includes(route)).join(', '));
 check('routes.authenticated', 'protected API requests include cookie credentials', js.includes("credentials:'include'") && js.includes('jsonRequest'));
+check('discovery.operator-controls', 'discovery builder and run controls are represented', all(['data-run-action="pause"', 'data-run-action="resume"', 'data-run-action="cancel"', 'live-log', 'source-health', 'daily_limit', 'source_ids', 'schedule'], token => `${html}\n${js}\n${css}`.includes(token)));
+check('prospects.filter-contract', 'prospect explorer exposes source, geography, category, and contact filters', all(['sourceFilter', 'geographyFilter', 'categoryFilter', 'contactStatusFilter', 'contact_status'], token => `${html}\n${js}\n${css}`.includes(token)));
 check('routes.no-arbitrary-target-fetch', 'browser does not fetch arbitrary target URLs directly', !/fetch\s*\(\s*(?:url|target|website|href)\b/i.test(js));

 const canonicalStages = ['new', 'contacted', 'qualified', 'proposal', 'negotiation', 'won', 'lost'];
@@ -126,6 +126,7 @@ h4 { font-size: 14px; }
 .workspace-grid { display: grid; grid-template-columns: minmax(0, 1.5fr) minmax(330px, .8fr); gap: 18px; margin-top: 18px; scroll-margin-top: 90px; }
 .explorer-panel { min-width: 0; padding: 20px; }
 .filters { display: grid; grid-template-columns: minmax(210px, 1.5fr) repeat(4, minmax(120px, 1fr)); gap: 8px; margin: 20px 0 12px; }
+.filters select { min-width: 0; }
 .search-wrap { position: relative; display: flex; align-items: center; }
 .search-wrap span { position: absolute; left: 11px; color: var(--muted-soft); font-size: 18px; }
 .search-wrap input { padding-left: 34px !important; }
@@ -265,6 +266,7 @@ footer { display: flex; justify-content: space-between; gap: 16px; margin-top: 3
 .direct-discovery-section { margin-top: 22px; scroll-margin-top: 90px; }
 .discovery-hero { display: flex; align-items: flex-start; justify-content: space-between; gap: 20px; padding: 20px; }
 .discovery-hero h2 { margin: 2px 0 5px; }.discovery-hero p { max-width: 700px; margin: 0; }.discovery-badge { padding: 7px 10px; border: 1px solid #c8e2de; border-radius: 999px; color: var(--teal); background: var(--teal-soft); font-size: 11px; font-weight: 800; white-space: nowrap; }
-.discovery-workspace-grid { display: grid; grid-template-columns: minmax(0, .95fr) minmax(330px, 1.05fr); gap: 18px; margin-top: 12px; }.direct-discovery-form-panel, .discovery-runs-panel, .discovery-results-panel { padding: 20px; }.direct-discovery-form-panel form { display: grid; gap: 13px; margin-top: 16px; }.discovery-limit-grid { display: grid; grid-template-columns: 1fr 1fr; gap: 10px; }.discovery-runs-state { margin-top: 14px; border-top: 1px solid var(--line); }.discovery-run-row { width: 100%; display: grid; grid-template-columns: minmax(0, 1fr) auto auto; gap: 10px; align-items: center; padding: 13px 0; border: 0; border-bottom: 1px solid var(--line); color: inherit; background: transparent; text-align: left; }.discovery-run-row:hover, .discovery-run-row.selected { background: #f4f9f8; }.discovery-run-row > span:first-child { display: grid; min-width: 0; gap: 2px; }.discovery-run-row small { overflow: hidden; color: var(--muted); font-size: 11px; text-overflow: ellipsis; white-space: nowrap; }.discovery-run-status { padding: 4px 8px; border-radius: 999px; color: var(--blue); background: var(--blue-soft); font-size: 10px; font-weight: 800; white-space: nowrap; }.discovery-run-status.failed { color: var(--red); background: var(--red-soft); }.discovery-run-status.succeeded { color: var(--green); background: var(--green-soft); }.discovery-results-panel { margin-top: 12px; }.discovery-result-meta { display: flex; flex-wrap: wrap; gap: 14px; margin: 16px 0 10px; color: var(--muted); font-size: 11px; }.discovery-candidate-list { display: grid; grid-template-columns: repeat(auto-fit, minmax(210px, 1fr)); gap: 9px; }.discovery-candidate { display: grid; gap: 3px; padding: 11px; border: 1px solid var(--line); border-radius: 9px; background: #fff; }.discovery-candidate span, .discovery-candidate small { overflow-wrap: anywhere; color: var(--muted); font-size: 11px; }.discovery-partial { margin: 12px 0; padding: 11px; border: 1px solid #ead6a9; border-radius: 9px; color: #77500c; background: var(--amber-soft); font-size: 12px; }.discovery-partial p { margin: 4px 0 0; }.discovery-empty { padding: 30px 10px; color: var(--muted); text-align: center; }
+.discovery-workspace-grid { display: grid; grid-template-columns: minmax(0, .95fr) minmax(330px, 1.05fr); gap: 18px; margin-top: 12px; }.direct-discovery-form-panel, .discovery-runs-panel, .discovery-results-panel { padding: 20px; }.direct-discovery-form-panel form { display: grid; gap: 13px; margin-top: 16px; }.discovery-limit-grid { display: grid; grid-template-columns: 1fr 1fr; gap: 10px; }.discovery-runs-state { margin-top: 14px; border-top: 1px solid var(--line); }.discovery-run-row { width: 100%; display: grid; grid-template-columns: minmax(0, 1fr) auto auto; gap: 10px; align-items: center; padding: 13px 0; border: 0; border-bottom: 1px solid var(--line); color: inherit; background: transparent; text-align: left; }.discovery-run-row:hover, .discovery-run-row.selected { background: #f4f9f8; }.discovery-run-row > span:first-child { display: grid; min-width: 0; gap: 2px; }.discovery-run-row small { overflow: hidden; color: var(--muted); font-size: 11px; text-overflow: ellipsis; white-space: nowrap; }.discovery-run-status { padding: 4px 8px; border-radius: 999px; color: var(--blue); background: var(--blue-soft); font-size: 10px; font-weight: 800; white-space: nowrap; }.discovery-run-status.failed { color: var(--red); background: var(--red-soft); }.discovery-run-status.succeeded { color: var(--green); background: var(--green-soft); }.discovery-results-panel { margin-top: 12px; }.discovery-result-meta { display: flex; flex-wrap: wrap; gap: 14px; margin: 16px 0 10px; color: var(--muted); font-size: 11px; }.discovery-candidate-list { display: grid; grid-template-columns: repeat(auto-fit, minmax(210px, 1fr)); gap: 9px; }.discovery-candidate { display: grid; gap: 3px; padding: 11px; border: 1px solid var(--line); border-radius: 9px; background: #fff; }.discovery-candidate span, .discovery-candidate small { overflow-wrap: anywhere; color: var(--muted); font-size: 11px; }.discovery-partial { margin: 12px 0; padding: 11px; border: 1px solid #ead6a9; border-radius: 9px; color: #77500c; background: var(--amber-soft); font-size: 12px; }
+.discovery-controls { display:flex; align-items:center; flex-wrap:wrap; gap:8px; margin-bottom:14px; padding-bottom:12px; border-bottom:1px solid var(--line); }.discovery-progress { padding:15px 0; }.discovery-stat-grid { display:grid; grid-template-columns:repeat(4,minmax(0,1fr)); gap:8px; margin:0 0 14px; }.discovery-stat-grid > div { padding:11px; border:1px solid var(--line); border-radius:9px; background:#f8fbfa; }.discovery-stat-grid small { display:block; color:var(--muted); font-size:10px; text-transform:uppercase; letter-spacing:.06em; }.discovery-stat-grid strong { display:block; margin-top:3px; color:var(--ink-strong); font-size:20px; }.discovery-two-col { display:grid; grid-template-columns:minmax(0,.9fr) minmax(0,1.1fr); gap:14px; margin-top:16px; }.live-log { min-width:0; }.live-log h4, .source-health h4 { margin:0 0 10px; }.live-log ol { max-height:240px; overflow:auto; display:grid; gap:11px; margin:0; padding:12px; border:1px solid var(--line); border-radius:9px; background:#f8fbfa; list-style:none; }.live-log li { display:flex; gap:9px; font-size:12px; }.live-log small { display:block; color:var(--muted); font-size:10px; }.source-health { padding:13px; border:1px solid #c8e2de; border-radius:10px; background:var(--teal-soft); }.subheading { display:flex; justify-content:space-between; gap:10px; align-items:center; }.source-health-grid { display:grid; gap:8px; }.source-health-grid article { display:grid; gap:3px; padding:10px; border:1px solid #c8e2de; border-radius:8px; background:#fff; }.source-health-grid small { color:var(--muted); font-size:10px; }.source-health-status { width:fit-content; padding:3px 7px; border-radius:99px; color:var(--green); background:var(--green-soft); font-size:10px; font-weight:800; }.source-health-actions { display:flex; gap:6px; margin-top:5px; }.advanced-discovery { padding:10px 0; border-top:1px solid var(--line); border-bottom:1px solid var(--line); }.advanced-discovery summary { cursor:pointer; color:var(--teal); font-size:12px; font-weight:750; margin-bottom:9px; }.discovery-partial p { margin: 4px 0 0; }.discovery-empty { padding: 30px 10px; color: var(--muted); text-align: center; }
 .ai-facts { display: grid; grid-template-columns: repeat(4, 1fr); gap: 7px; margin: 13px 0; }.ai-facts div { padding: 8px; border: 1px solid var(--line); border-radius: 8px; background: #fff; }.ai-facts dt { color: var(--muted); font-size: 9px; font-weight: 800; text-transform: uppercase; }.ai-facts dd { margin: 3px 0 0; overflow-wrap: anywhere; font-size: 11px; }
-@media(max-width:700px) { .discovery-hero { flex-direction: column; }.discovery-workspace-grid { grid-template-columns: 1fr; }.discovery-limit-grid, .ai-facts { grid-template-columns: 1fr 1fr; }.discovery-run-row { grid-template-columns: 1fr auto; }.discovery-run-row > small { grid-column: 1 / -1; } }
+@media(max-width:700px) { .discovery-hero { flex-direction: column; }.discovery-workspace-grid { grid-template-columns: 1fr; }.discovery-limit-grid, .ai-facts, .discovery-stat-grid { grid-template-columns: 1fr 1fr; }.discovery-two-col { grid-template-columns: 1fr; }.discovery-run-row { grid-template-columns: 1fr auto; }.discovery-run-row > small { grid-column: 1 / -1; } }
Author	SHA1	Message	Date
Marco0300	1541d8f6c2	add governed source discovery system CI / compose (push) Successful in 14m5s Details	2026-09-03 23:04:37 +02:00
Marco0300	58c4d93208	complete source discovery processing lifecycle	2026-09-03 23:01:25 +02:00
Marco0300	a9213c0282	add governed source adapter discovery framework	2026-09-03 22:49:05 +02:00
Marco0300	a99e6b26dc	[verified] build discovery runs workspace	2026-09-03 22:48:28 +02:00