[verified] align CRM frontend contracts

.env.example

@@ -12,6 +12,11 @@ BOOTSTRAP_ADMIN_EMAIL=
 BOOTSTRAP_ADMIN_PASSWORD=
 # Hard safety default; this release has no delivery capability.
 AUTOMATED_OUTREACH_ENABLED=false
+# Optional criteria-first discovery provider. All three values are secret/config inputs:
+# endpoint must be HTTPS and its exact hostname must appear in the allowlist.
+SEARCH_PROVIDER_URL=
+SEARCH_PROVIDER_ALLOWED_HOSTS=
+SEARCH_PROVIDER_API_KEY=
 # Backup operations (host-side, never mounted into the web container).
 BACKUP_DIR=/var/backups/prospect-platform
 BACKUP_RETENTION=30

README.md

@@ -78,6 +78,10 @@ Compose passes the one-time `BOOTSTRAP_ADMIN_EMAIL` and `BOOTSTRAP_ADMIN_PASSWOR
 
 Authenticated browser requests use a server-side session cookie; login creates a session and logout invalidates it. The liveness endpoints (`GET /api/v1/health/live` and `GET /healthz`) intentionally remain unauthenticated so Docker, ingress, and monitoring health checks can use them. Authentication is not a substitute for tenant/authorization checks: protected routes must enforce the session and organization boundary server-side.
 
+### Criteria-first discovery
+
+`POST /api/v1/discovery` accepts a bounded `criteria` object and may omit `seed_urls`. In that mode, the API calls the configured generic JSON search provider, bounds returned URLs to at most 50, and feeds them into the existing SSRF-checked, same-origin crawler and tenant-scoped persistence. Explicit `seed_urls` (1–5 public URLs) remain supported for controlled runs. Without a provider, the criteria-first request returns `503 {"error":"not_configured"}`. Check `GET /api/v1/discovery/provider-status` for redacted readiness. Configure `SEARCH_PROVIDER_URL` (HTTPS endpoint), `SEARCH_PROVIDER_ALLOWED_HOSTS` (comma-separated exact hostname allowlist containing the endpoint host), and optional `SEARCH_PROVIDER_API_KEY`; unsafe or missing configuration is never called. The endpoint must return JSON `{"results":[{"url":"https://example.test"}]}` (or `items`/`website`/`link` equivalents).
+
 ## Phase 5 source boundary and remaining limitations
 
 Phase 5 defines a source adapter contract and registry; Phase 8 adds a bounded website-observation adapter, but it does not implement general network discovery, enrichment scheduling, or a live external-source adapter. A source adapter must declare its identity, terms owner, permitted purpose, rate limits, retention class, query/result schema, dry-run behavior, and health/circuit controls. CSV and manual reference adapters may be used for operator-supplied data; they must preserve source attribution and raw source records, and must not silently turn preview data into outreach or verified facts.

apps/api/app/ai_assistance.py

@@ -1,8 +1,8 @@
-"""Evidence-bounded AI assistance primitives for Phase 13.
+"""Evidence-bounded AI enrichment with a deterministic, network-free provider.
 
-This module deliberately has no network or model dependency.  The local provider is
-an auditable formatter over stored records; other providers are reported as
-not_configured rather than guessed at.
+Remote providers are intentionally only a status/configuration concept here.  No
+network client is present, so an unconfigured or unreviewed remote provider fails
+closed and cannot accidentally perform outreach or exfiltrate tenant data.
 """
 from __future__ import annotations
 
@@ -15,68 +15,70 @@ from typing import Any
 MAX_INPUT_ITEMS = 100
 MAX_FIELD_CHARS = 500
 MAX_OUTPUT_CHARS = 12_000
-SUPPORTED_KINDS = ("summary", "qualification_explanation", "missing_data_questions", "research_note")
 _SECRET_RE = re.compile(r"(?i)(password|passwd|secret|token|api[_-]?key|authorization|private[_-]?key|credential)\s*[:=]\s*[^\s,;]+")
+_SECRET_KEY_RE = re.compile(r"(?i)(password|passwd|secret|token|api[_-]?key|authorization|private[_-]?key|credential)")
 
 
 def _text(value: Any, limit: int = MAX_FIELD_CHARS) -> str:
-    value = "" if value is None else str(value)
-    value = _SECRET_RE.sub(r"\1: [REDACTED]", value)
-    return value[:limit]
+    return _SECRET_RE.sub(r"\1: [REDACTED]", "" if value is None else str(value))[:limit]
 
 
 def redact(value: Any) -> Any:
     if isinstance(value, dict):
-        return {str(k)[:80]: ("[REDACTED]" if re.search(r"(?i)(password|passwd|secret|token|api[_-]?key|authorization|private[_-]?key|credential)", str(k)) else redact(v)) for k, v in list(value.items())[:100]}
-    if isinstance(value, list):
-        return [redact(v) for v in value[:MAX_INPUT_ITEMS]]
-    if isinstance(value, str):
-        return _text(value)
+        return {str(k)[:80]: ("[REDACTED]" if _SECRET_KEY_RE.search(str(k)) else redact(v)) for k, v in list(value.items())[:MAX_INPUT_ITEMS]}
+    if isinstance(value, list): return [redact(v) for v in value[:MAX_INPUT_ITEMS]]
+    if isinstance(value, str): return _text(value)
     return value
 
 
+def _hash(item: Any) -> str:
+    return hashlib.sha256(json.dumps(redact(item), sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()).hexdigest()
+
+
 def evidence_hashes(evidence: list[dict[str, Any]]) -> list[str]:
-    return [hashlib.sha256(json.dumps(redact(item), sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()).hexdigest() for item in evidence]
+    return [_hash(item) for item in evidence[:MAX_INPUT_ITEMS]]
 
 
-def _citation(item: dict[str, Any]) -> dict[str, Any]:
-    return {"evidence_id": int(item["id"]), "kind": _text(item.get("kind", "evidence"), 80), "url": _text(item.get("url", ""), 500)}
+def input_fingerprint(business: dict[str, Any], scans: list[dict[str, Any]], contacts: list[dict[str, Any]], evidence: list[dict[str, Any]], max_items: int = MAX_INPUT_ITEMS) -> str:
+    payload = {"business": redact(business), "scans": [redact(x) for x in scans[:max_items]], "contacts": [redact(x) for x in contacts[:max_items]], "evidence": [redact(x) for x in evidence[:max_items]]}
+    return _hash(payload)
 
 
-def _claim(item: dict[str, Any]) -> str:
-    return _text(item.get("claim", ""), MAX_FIELD_CHARS).strip()
+def _citation(item: dict[str, Any], source_type: str, provenance: str) -> dict[str, Any]:
+    return {"source_type": source_type, "source_id": int(item["id"]) if str(item.get("id", "")).isdigit() else None,
+            "kind": _text(item.get("kind", source_type), 80), "url": _text(item.get("url", item.get("input_url", item.get("source_url", ""))), 500),
+            "provenance": provenance, "hash": _hash(item), "policy": "stored_evidence_only"}
 
 
 def build_local_suggestions(business: dict[str, Any], scans: list[dict[str, Any]], contacts: list[dict[str, Any]], evidence: list[dict[str, Any]], score_history: list[dict[str, Any]] | None = None) -> dict[str, Any]:
-    """Create deterministic suggestions using only supplied stored data.
-
-    Every claim-bearing item cites one or more rows from ``evidence``.  No
-    contact details are emitted, and contacts are used only as aggregate counts.
-    """
-    evidence = [redact(x) for x in evidence[:MAX_INPUT_ITEMS] if _claim(x)]
-    citations = [_citation(x) for x in evidence]
-    claims = [_claim(x) for x in evidence]
-    suggestions: list[dict[str, Any]] = []
+    evidence = [redact(x) for x in evidence[:MAX_INPUT_ITEMS] if _text(x.get("claim", "")).strip()]
+    scans = [redact(x) for x in scans[:MAX_INPUT_ITEMS]]
+    contacts = [redact(x) for x in contacts[:MAX_INPUT_ITEMS]]
+    citations = [_citation(x, "evidence", "discovery_evidence") for x in evidence]
+    citations += [_citation(x, "website_scan", "website_scanner") for x in scans]
+    citations += [_citation(x, "contact_extraction", "contact_extractor") for x in contacts]
+    claims = [_text(x.get("claim", "")).strip() for x in evidence]
     name = _text(business.get("name", "this business"), 200)
-    if claims:
-        joined = " ".join(f"{claim} [evidence:{item['id']}]" for claim, item in zip(claims[:5], evidence[:5]))
-        suggestions.append({"type": "summary", "text": f"Stored evidence for {name}: {joined}", "citations": citations[:5]})
-        score = business.get("score")
-        if score is not None:
-            suggestions.append({"type": "qualification_explanation", "text": f"The stored qualification score is {_text(score, 30)}; review the cited evidence before relying on it. [evidence:{evidence[0]['id']}]", "citations": citations[:1]})
-        missing = []
-        if not _text(business.get("website", "")).strip(): missing.append("official website")
-        if not contacts: missing.append("public contact evidence")
-        if missing:
-            suggestions.append({"type": "missing_data_questions", "text": "Confirm whether the following data is available: " + ", ".join(missing) + f". [evidence:{evidence[0]['id']}]", "citations": citations[:1]})
-        suggestions.append({"type": "research_note", "text": f"Draft note: independently verify the stored claims for {name}; do not infer facts beyond the cited records. [evidence:{evidence[0]['id']}]", "citations": citations[:1]})
-    else:
-        # No claim is fabricated.  A question is safe but has no citation, so
-        # return no suggestions and let the caller expose the missing-data state.
-        suggestions = []
-    output = {"provider": "local", "version": "deterministic-v1", "suggestions": suggestions, "grounded": True, "claim_policy": "stored_evidence_only"}
+    score = business.get("score")
+    priority = ("high" if score is not None and int(score) >= 70 else "medium" if score is not None and int(score) >= 40 else "low")
+    uncertainty = []
+    if not claims: uncertainty.append("no_claim_bearing_evidence")
+    if not contacts: uncertainty.append("no_public_contact_extraction")
+    conflicts = []
+    classifications = [str(x.get("classification", "")).lower() for x in scans if x.get("classification")]
+    if len(set(classifications)) > 1: conflicts.append("website_scan_classifications_disagree")
+    classification = "business_prospect" if claims or business.get("website") else "insufficient_evidence"
+    summary = f"Stored evidence for {name}: " + (" ".join(f"{claim} [evidence:{item['source_id']}]" for claim, item in zip(claims[:5], citations[:5])) if claims else "insufficient claim-bearing evidence")
+    output = {
+        "classification": classification, "summary": _text(summary, 2000),
+        "priority_recommendation": priority, "confidence": round(min(0.95, 0.45 + 0.1 * len(claims) + (0.1 if scans else 0)), 2),
+        "uncertainty": uncertainty, "conflicts": conflicts, "citations": citations[:MAX_INPUT_ITEMS],
+        "policy": {"claim_policy": "stored_evidence_only", "no_outreach": True, "redacted_inputs": True},
+        "suggestions": ([{"type": "summary", "text": _text(summary), "citations": citations[:5]}] if claims else []),
+        "grounded": True, "provider": "local", "version": "deterministic-v2",
+    }
     encoded = json.dumps(output, sort_keys=True, ensure_ascii=False)
-    return json.loads(encoded[:MAX_OUTPUT_CHARS]) if len(encoded) <= MAX_OUTPUT_CHARS else {"provider": "local", "version": "deterministic-v1", "suggestions": suggestions[:1], "grounded": True, "claim_policy": "stored_evidence_only"}
+    return output if len(encoded) <= MAX_OUTPUT_CHARS else {**output, "suggestions": output["suggestions"][:1], "citations": citations[:20]}
 
 
 def provider_name() -> str | None:
@@ -84,10 +86,14 @@ def provider_name() -> str | None:
     return value or None
 
 
+def provider_status(configured: str | None = None) -> dict[str, Any]:
+    provider = configured if configured is not None else provider_name()
+    local = provider in {"local", "deterministic"}
+    return {"provider": provider or "", "status": "ready" if local else "not_configured", "network_enabled": False, "reviewed": local, "outbound_calls": False}
+
+
 def generate(business: dict[str, Any], scans: list[dict[str, Any]], contacts: list[dict[str, Any]], evidence: list[dict[str, Any]], score_history: list[dict[str, Any]] | None = None) -> tuple[str, str, str, dict[str, Any]]:
     provider = provider_name()
-    hashes = evidence_hashes(evidence)
-    metadata = {"input_counts": {"business": 1, "scans": min(len(scans), MAX_INPUT_ITEMS), "contacts": min(len(contacts), MAX_INPUT_ITEMS), "evidence": min(len(evidence), MAX_INPUT_ITEMS)}, "redacted": True, "max_input_items": MAX_INPUT_ITEMS, "max_field_chars": MAX_FIELD_CHARS, "evidence_hashes": hashes}
-    if provider not in {"local", "deterministic"}:
-        return "not_configured", provider or "", "", metadata
-    return "succeeded", "local", "deterministic-v1", {**metadata, "output": build_local_suggestions(business, scans, contacts, evidence, score_history)}
+    metadata = {"input_counts": {"business": 1, "scans": min(len(scans), MAX_INPUT_ITEMS), "contacts": min(len(contacts), MAX_INPUT_ITEMS), "evidence": min(len(evidence), MAX_INPUT_ITEMS)}, "redacted": True, "max_input_items": MAX_INPUT_ITEMS, "max_field_chars": MAX_FIELD_CHARS, "evidence_hashes": evidence_hashes(evidence), "input_fingerprint": input_fingerprint(business, scans, contacts, evidence)}
+    if provider not in {"local", "deterministic"}: return "not_configured", provider or "", "", metadata
+    return "succeeded", "local", "deterministic-v2", {**metadata, "output": build_local_suggestions(business, scans, contacts, evidence, score_history)}

apps/api/app/discovery.py

@@ -0,0 +1,138 @@
+"""Bounded, public-only prospect discovery with optional criteria search.
+
+There is deliberately no general web search or arbitrary URL input here: callers provide
+at most a small set of public seed pages. When seeds are omitted, a configured search
+provider supplies bounded seed URLs. Links found on those seeds become candidate sites,
+and subsequent crawling is same-origin, bounded, and SSRF-checked by the scanner.
+"""
+from __future__ import annotations
+
+import re
+from html.parser import HTMLParser
+from urllib.parse import urljoin, urldefrag, urlparse
+
+from .contact_extractor import extract_contacts
+from .website_scanner import MAX_BYTES, MAX_REDIRECTS, _fetch, validate_url
+from .search_provider import search as search_provider
+
+MAX_SEEDS = 5
+MAX_PAGES = 20
+MAX_CANDIDATES = 50
+MAX_HTML = MAX_BYTES
+
+
+class _Links(HTMLParser):
+    def __init__(self):
+        super().__init__(convert_charrefs=True)
+        self.links = []
+        self.title = ""
+        self.headings = []
+        self._tag = ""
+        self._buf = []
+
+    def handle_starttag(self, tag, attrs):
+        tag = tag.lower(); self._tag = tag
+        if tag in {"a", "link"}:
+            attrs = {str(k).lower(): str(v or "") for k, v in attrs}
+            if attrs.get("href"): self.links.append(attrs["href"])
+        if tag in {"title", "h1", "h2", "h3"}: self._buf = []
+
+    def handle_data(self, data):
+        if self._tag in {"title", "h1", "h2", "h3"}: self._buf.append(data)
+
+    def handle_endtag(self, tag):
+        tag = tag.lower()
+        if tag == "title" and self._buf: self.title = " ".join("".join(self._buf).split())[:500]
+        if tag in {"h1", "h2", "h3"} and self._buf: self.headings.append(" ".join("".join(self._buf).split())[:300])
+        self._tag = ""
+
+
+def _html(value: bytes) -> str:
+    return value[:MAX_HTML].decode("utf-8", "replace")
+
+
+def _same_site(url, root):
+    return (urlparse(url).hostname or "").lower().rstrip(".") == (urlparse(root).hostname or "").lower().rstrip(".")
+
+
+def _criteria_match(text, criteria):
+    keywords = criteria.get("keywords", criteria.get("keyword", []))
+    if isinstance(keywords, str): keywords = [keywords]
+    if not isinstance(keywords, list) or len(keywords) > 20: raise ValueError("invalid_criteria")
+    haystack = text.lower()
+    return not keywords or all(str(k).strip().lower() in haystack for k in keywords if str(k).strip())
+
+
+def discover(criteria, seed_urls=None, *, max_pages=MAX_PAGES, max_candidates=MAX_CANDIDATES):
+    if not isinstance(criteria, dict) or len(criteria) > 20: raise ValueError("invalid_criteria")
+    try: max_pages = int(max_pages); max_candidates = int(max_candidates)
+    except (TypeError, ValueError): raise ValueError("invalid_limits")
+    if not 1 <= max_pages <= MAX_PAGES or not 1 <= max_candidates <= MAX_CANDIDATES: raise ValueError("invalid_limits")
+    if seed_urls is None:
+        seeds = search_provider(criteria, max_candidates)
+        mechanism = "criteria_search_provider"
+    else:
+        if not isinstance(seed_urls, list) or not 0 < len(seed_urls) <= MAX_SEEDS: raise ValueError("seed_urls_required")
+        seeds = list(seed_urls)
+        mechanism = "explicit_seed_allowlist"
+    raw_seeds = list(seeds)
+    seeds = []
+    for raw in raw_seeds:
+        try: safe = validate_url(raw)
+        except ValueError as exc: raise ValueError("unsafe_seed_url") from exc
+        if safe not in seeds: seeds.append(safe)
+    candidates = []
+    for seed in seeds:
+        fetched = _fetch(seed, max_bytes=MAX_HTML, max_redirects=MAX_REDIRECTS)
+        if fetched.get("content_type") not in {"text/html", "application/xhtml+xml"}: continue
+        parser = _Links(); parser.feed(_html(fetched["body"]))
+        seed_text = " ".join([parser.title, *parser.headings])
+        linked = []
+        for link in parser.links:
+            target = urldefrag(urljoin(seed, link))[0]
+            if not target or target.startswith(("mailto:", "tel:", "javascript:")): continue
+            try: target = validate_url(target)
+            except ValueError: continue
+            if target not in linked and target not in seeds: linked.append(target)
+            if len(linked) >= max_candidates: break
+        # A seed acts as a directory/index when it links outward. If it has no
+        # usable links, it may itself be the explicitly allowlisted business site.
+        if linked:
+            candidates.extend(x for x in linked if x not in candidates)
+        elif _criteria_match(seed_text, criteria):
+            candidates.append(seed)
+    results = []
+    seen = set()
+    for candidate in candidates[:max_candidates]:
+        root = candidate; queue = [candidate]; pages = []
+        while queue and len(pages) < max_pages:
+            page = queue.pop(0)
+            if page in {x["url"] for x in pages}: continue
+            try: fetched = _fetch(page, max_bytes=MAX_HTML, max_redirects=MAX_REDIRECTS)
+            except ValueError: continue
+            if fetched.get("content_type") not in {"text/html", "application/xhtml+xml"}: continue
+            html = _html(fetched["body"]); parser = _Links(); parser.feed(html)
+            pages.append({"url": page, "final_url": fetched.get("final_url") or page, "html": html, "title": parser.title, "headings": parser.headings, "status": fetched.get("status")})
+            for link in parser.links:
+                target = urldefrag(urljoin(page, link))[0]
+                if target and _same_site(target, root):
+                    try: target = validate_url(target)
+                    except ValueError: continue
+                    if target not in {x["url"] for x in pages} and target not in queue: queue.append(target)
+            if len(queue) + len(pages) >= max_pages: queue = queue[:max(0, max_pages - len(pages))]
+        if not pages: continue
+        text = " ".join(x["title"] + " " + " ".join(x["headings"]) for x in pages)
+        if not _criteria_match(text, criteria): continue
+        domain = (urlparse(root).hostname or "").lower().removeprefix("www.")
+        if domain in seen: continue
+        seen.add(domain)
+        contacts = []
+        evidence = []
+        for page in pages:
+            contacts.extend(extract_contacts(page["html"], page["url"], max_results=100))
+            claim = page["title"] or (page["headings"][0] if page["headings"] else "Public business page")
+            evidence.append({"kind": "discovery_page", "url": page["url"], "claim": claim, "provenance": "scoped_discovery"})
+        deduped = {(x["kind"], x["value"]): x for x in contacts}
+        name = next((x["headings"][0] for x in pages if x["headings"]), next((x["title"] for x in pages if x["title"]), domain))
+        results.append({"name": name[:200], "website": root, "website_domain": domain, "description": text[:1000], "contacts": list(deduped.values())[:100], "evidence": evidence, "pages": pages, "pages_crawled": len(pages), "provenance": {"mechanism": mechanism, "seed_urls": seeds, "root_url": root}})
+    return {"candidates": results, "seeds": seeds, "pages_limit": max_pages, "candidate_limit": max_candidates}

apps/api/app/main.py

@@ -13,7 +13,9 @@ if __package__ in (None, ""):
     from app.website_scanner import scan_website, validate_url
     from app.contact_extractor import extract_contacts, MAX_HTML_BYTES, MAX_RESULTS
     from app.scoring import DEFAULT_RULES, signals_for_business, evaluate_score, SCORE_VERSION
-    from app.ai_assistance import generate as generate_ai, MAX_INPUT_ITEMS, MAX_OUTPUT_CHARS
+    from app.ai_assistance import generate as generate_ai, input_fingerprint, provider_status, MAX_INPUT_ITEMS, MAX_OUTPUT_CHARS
+    from app.discovery import discover as scoped_discover
+    from app.search_provider import provider_status as search_provider_status
     from app.config import load_config
 else:
     from .domain import deduplication_key, deduplicate_businesses, is_suppressed, normalize_business, score_business, normalize_domain, normalize_phone, match_businesses
@@ -22,14 +24,16 @@ else:
     from .website_scanner import scan_website, validate_url
     from .contact_extractor import extract_contacts, MAX_HTML_BYTES, MAX_RESULTS
     from .scoring import DEFAULT_RULES, signals_for_business, evaluate_score, SCORE_VERSION
-    from .ai_assistance import generate as generate_ai, MAX_INPUT_ITEMS, MAX_OUTPUT_CHARS
+    from .ai_assistance import generate as generate_ai, input_fingerprint, provider_status, MAX_INPUT_ITEMS, MAX_OUTPUT_CHARS
+    from .discovery import discover as scoped_discover
+    from .search_provider import provider_status as search_provider_status
 from .config import load_config
 ORGANIZATION_ID = "demo-tenant"
 SCHEMA = Path(__file__).resolve().parents[1] / "schema.sql"
 SESSION_DAYS = 7
 PBKDF2_ITERATIONS = 300_000
 MUTATING_ROLES = {"owner", "admin", "researcher"}
-JOB_TYPES = {"noop", "prospect_recalculate", "source_discovery", "domain_check"}
+JOB_TYPES = {"noop", "prospect_recalculate", "source_discovery", "domain_check", "scoped_discovery"}
 JOB_PAGE_SIZE = 100
 WEBSITE_SCAN_PAGE_SIZE = 100
 WEBSITE_SCAN_CACHE_SECONDS = 3600
@@ -121,8 +125,8 @@ class ApiHandler(BaseHTTPRequestHandler):
         db.execute("INSERT INTO audit_log (organization_id,user_id,action,details) VALUES (?,?,?,?)",(user["organization_id"],user["id"],action,details))
     def business(self, db, ident, org): return db.execute("SELECT * FROM businesses WHERE id=? AND organization_id=?",(ident,org)).fetchone()
     def nested(self, db, bid, org):
-        result={"contacts":[],"domains":[],"websites":[],"evidence":[],"pipeline":[],"interactions":[],"notes":[]}
-        tables={"contacts":"contacts","domains":"domains","websites":"websites","evidence":"evidence","pipeline":"pipeline_entries","interactions":"interactions","notes":"notes"}
+        result={"contacts":[],"contact_extractions":[],"domains":[],"websites":[],"evidence":[],"pipeline":[],"interactions":[],"notes":[]}
+        tables={"contacts":"contacts","contact_extractions":"contact_extractions","domains":"domains","websites":"websites","evidence":"evidence","pipeline":"pipeline_entries","interactions":"interactions","notes":"notes"}
         for key, table in tables.items():
             result[key]=[row_json(r) for r in db.execute(f"SELECT * FROM {table} WHERE business_id=? AND organization_id=? ORDER BY id",(bid,org))]
         return result
@@ -470,6 +474,40 @@ class ApiHandler(BaseHTTPRequestHandler):
         reasons.append("network_send_disabled"); self.audit(db, user, "outreach_draft.send_blocked", f"{did}:network_send_disabled"); db.commit()
         return self.send_json(409, {"status": "blocked", "blocked_reasons": reasons, "network_send": False, "id": did})
 
+    def ai_provider_config(self, db, user, payload=None):
+        org = user["organization_id"]
+        row = db.execute("SELECT * FROM ai_provider_configs WHERE organization_id=?", (org,)).fetchone()
+        if payload is not None:
+            provider = str(payload.get("provider", "local")).strip().lower()
+            if not provider or len(provider) > 80: return self.send_json(400, {"error": "invalid_provider"})
+            reviewed = bool(payload.get("reviewed", False))
+            enabled = bool(payload.get("enabled", True))
+            # A remote provider cannot be enabled by configuration alone: this
+            # service has no reviewed transport abstraction and never sends data.
+            if provider not in {"local", "deterministic"} and (enabled or reviewed):
+                return self.send_json(409, {"error": "provider_not_reviewed", "network_enabled": False})
+            db.execute("INSERT INTO ai_provider_configs(organization_id,provider,enabled,reviewed) VALUES(?,?,?,?) ON CONFLICT(organization_id) DO UPDATE SET provider=excluded.provider,enabled=excluded.enabled,reviewed=excluded.reviewed,updated_at=CURRENT_TIMESTAMP", (org, provider, int(enabled), int(reviewed)))
+            self.audit(db, user, "ai_provider.updated", provider); db.commit()
+            row = db.execute("SELECT * FROM ai_provider_configs WHERE organization_id=?", (org,)).fetchone()
+        configured = row["provider"] if row and row["enabled"] else None
+        result = provider_status(configured if configured is not None else None)
+        result.update({"organization_id": org, "configured": bool(row), "enabled": bool(row and row["enabled"]), "reviewed": bool(row and row["reviewed"])})
+        return self.send_json(200, result)
+
+    def _ai_current_fingerprint(self, db, row):
+        if not row["business_id"]: return ""
+        try: limit = int(json.loads(row["prompt_metadata_json"] or "{}").get("request", {}).get("max_items", MAX_INPUT_ITEMS))
+        except (TypeError, ValueError): limit = MAX_INPUT_ITEMS
+        limit = max(1, min(limit, MAX_INPUT_ITEMS))
+        org, bid = row["organization_id"], row["business_id"]
+        business = self.business(db, bid, org)
+        if not business: return ""
+        scans = [dict(x) for x in db.execute("SELECT id,input_url,classification,result_json,scanned_at FROM website_scans WHERE business_id=? AND organization_id=? ORDER BY id DESC LIMIT ?", (bid, org, limit))]
+        evidence = [dict(x) for x in db.execute("SELECT id,kind,url,claim,created_at FROM evidence WHERE business_id=? AND organization_id=? ORDER BY id DESC LIMIT ?", (bid, org, limit))]
+        contacts = [dict(x) for x in db.execute("SELECT * FROM contacts WHERE business_id=? AND organization_id=? ORDER BY id DESC LIMIT ?", (bid, org, limit))]
+        extracted = [dict(x) for x in db.execute("SELECT * FROM contact_extractions WHERE business_id=? AND organization_id=? ORDER BY id DESC LIMIT ?", (bid, org, limit))]
+        return input_fingerprint(row_json(business), scans, contacts + extracted, evidence, limit)
+
     def suggest_ai(self, bid, payload, db, user):
         org = user["organization_id"]
         business = self.business(db, bid, org)
@@ -487,7 +525,7 @@ class ApiHandler(BaseHTTPRequestHandler):
         scans = [dict(r) for r in db.execute("SELECT id,input_url,classification,result_json,scanned_at FROM website_scans WHERE business_id=? AND organization_id=? ORDER BY id DESC LIMIT ?", (bid, org, requested)).fetchall()]
         evidence = [dict(r) for r in db.execute("SELECT id,kind,url,claim,created_at FROM evidence WHERE business_id=? AND organization_id=? ORDER BY id DESC LIMIT ?", (bid, org, requested)).fetchall()]
         history = [dict(r) for r in db.execute("SELECT score,eligible,priority_band,score_version,explanations_json,created_at FROM score_history WHERE business_id=? AND organization_id=? ORDER BY id DESC LIMIT ?", (bid, org, requested)).fetchall()]
-        status, provider, version, metadata = generate_ai(business_dict, scans[:requested], contacts[:requested], evidence[:requested], history[:requested])
+        status, provider, version, metadata = generate_ai(business_dict, scans[:requested], contacts[:requested] + extracted[:requested], evidence[:requested], history[:requested])
         output = metadata.pop("output", {}) if status == "succeeded" else {"suggestions": [], "grounded": True, "claim_policy": "stored_evidence_only"}
         hashes = metadata.get("evidence_hashes", [])
         cur = db.execute("INSERT INTO ai_runs(organization_id,business_id,input_evidence_hashes_json,model,provider,version,prompt_metadata_json,data_minimization_json,status,approval_state,output_json,actor_user_id) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)", (org, bid, json.dumps(hashes, sort_keys=True), "local-deterministic" if provider else "", provider, version, json.dumps({"request": {"max_items": requested}, "output_limit": MAX_OUTPUT_CHARS}, sort_keys=True), json.dumps(metadata, sort_keys=True), status, "pending", json.dumps(output, sort_keys=True), user["id"]))
@@ -515,6 +553,12 @@ class ApiHandler(BaseHTTPRequestHandler):
         if not row: return self.send_json(404, {"error": "not_found"})
         if decision == "approve" and row["status"] != "succeeded": return self.send_json(409, {"error": "run_not_approvable"})
         if row["approval_state"] != "pending": return self.send_json(409, {"error": "already_decided"})
+        if decision == "approve":
+            try: metadata = json.loads(row["data_minimization_json"] or "{}")
+            except (TypeError, ValueError): metadata = {}
+            expected = metadata.get("input_fingerprint")
+            if expected and expected != self._ai_current_fingerprint(db, row):
+                return self.send_json(409, {"error": "ai_run_stale", "approval_state": "pending", "reason": "source_hash_changed"})
         now = datetime.now(timezone.utc).replace(microsecond=0).isoformat()
         if decision == "approve": db.execute("UPDATE ai_runs SET approval_state='approved',approved_at=? WHERE id=? AND organization_id=?", (now, run_id, user["organization_id"]))
         else: db.execute("UPDATE ai_runs SET approval_state='rejected',rejected_at=? WHERE id=? AND organization_id=?", (now, run_id, user["organization_id"]))
@@ -552,6 +596,8 @@ class ApiHandler(BaseHTTPRequestHandler):
             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/discovery-queries": return self.list_queries(db,org)
+            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, search_provider_status())
             if path=="/api/v1/source-records": return self.list_source_records(db,org,parse_qs(parsed.query))
             if path=="/api/v1/jobs": return self.list_jobs(db,org,parse_qs(parsed.query))
             if path=="/api/v1/domain-checks": return self.list_domain_checks(db,org,parse_qs(parsed.query))
@@ -563,6 +609,7 @@ class ApiHandler(BaseHTTPRequestHandler):
             if path=="/api/v1/interactions": return self.list_interactions(db,org,parse_qs(parsed.query))
             if path=="/api/v1/suppressions": return self.list_suppressions(db,org,parse_qs(parsed.query))
             if path=="/api/v1/ai-runs": return self.list_ai_runs(db,user,parse_qs(parsed.query))
+            if path=="/api/v1/ai/provider-config": return self.ai_provider_config(db,user)
             if path=="/api/v1/outreach/drafts": return self.list_outreach_drafts(db,user,parse_qs(parsed.query))
             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))
@@ -746,6 +793,52 @@ class ApiHandler(BaseHTTPRequestHandler):
         rows=db.execute("SELECT * FROM jobs WHERE organization_id=? ORDER BY id DESC LIMIT ? OFFSET ?",(org,limit+1,offset)).fetchall(); more=len(rows)>limit
         return self.send_json(200,{"organization_id":org,"items":[job_json(r) for r in rows[:limit]],"limit":limit,"offset":offset,"has_more":more})
 
+    def _discovery_run_json(self, row):
+        item = row_json(row)
+        for field, default in (("criteria_json", {}), ("seed_urls_json", []), ("result_json", {})):
+            try: item[field[:-5]] = json.loads(item.pop(field) or json.dumps(default))
+            except (TypeError, ValueError): item[field[:-5]] = default
+        return item
+
+    def list_discovery_runs(self, db, org, query):
+        try: limit = max(1, min(int((query.get("page_size") or [50])[0]), 100)); offset = max(0, int((query.get("offset") or [0])[0]))
+        except (ValueError, TypeError): return self.send_json(400, {"error": "invalid_pagination"})
+        rows = db.execute("SELECT * FROM discovery_runs WHERE organization_id=? ORDER BY id DESC LIMIT ? OFFSET ?", (org, limit + 1, offset)).fetchall()
+        return self.send_json(200, {"organization_id": org, "items": [self._discovery_run_json(r) for r in rows[:limit]], "limit": limit, "offset": offset, "has_more": len(rows) > limit})
+
+    def create_scoped_discovery(self, payload, db, user):
+        criteria = payload.get("criteria", {}); seeds = payload.get("seed_urls")
+        criteria_only = seeds is None
+        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 criteria_only:
+            status = search_provider_status()
+            if status["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 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:
+            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)
+        # 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 [])))
+            self.audit(db, user, "discovery.created", str(job["id"])); db.commit()
+        getattr(self.server, "job_wakeup", threading.Event()).set()
+        return result
+
     def get_job_route(self, db, org, path, query):
         bits=path.split("/")
         if len(bits)<5 or not bits[4].isdigit(): return self.send_json(404,{"error":"not_found"})
@@ -772,7 +865,8 @@ class ApiHandler(BaseHTTPRequestHandler):
         safe=json.dumps(redact(data),sort_keys=True,separators=(",",":")); max_attempts=max(1,min(int(payload.get("max_attempts",3)),5)) if str(payload.get("max_attempts",3)).isdigit() else 3
         try:
             cur=db.execute("INSERT INTO jobs(organization_id,idempotency_key,type,payload,max_attempts) VALUES(?,?,?,?,?)",(user["organization_id"],key,kind,safe,max_attempts)); jid=cur.lastrowid
-            self.add_job_event(db,jid,user["organization_id"],"queued","Job queued",0); self.audit(db,user,"job.created",str(jid)); db.commit(); getattr(self.server,"job_wakeup",threading.Event()).set()
+            self.add_job_event(db,jid,user["organization_id"],"queued","Job queued",0); self.audit(db,user,"job.created",str(jid)); db.commit()
+            if not payload.get("_defer_wakeup"): getattr(self.server,"job_wakeup",threading.Event()).set()
             return self.send_json(202 if payload.get("_accepted") else 201,job_json(db.execute("SELECT * FROM jobs WHERE id=?",(jid,)).fetchone()))
         except sqlite3.IntegrityError:
             row=db.execute("SELECT * FROM jobs WHERE organization_id=? AND idempotency_key=?",(user["organization_id"],key)).fetchone(); return self.send_json(200,job_json(row))
@@ -925,6 +1019,7 @@ class ApiHandler(BaseHTTPRequestHandler):
             payload=self.read_json(); org=user["organization_id"]
             if path=="/api/v1/saved-filters": return self.save_filter(payload,db,user)
             if path=="/api/v1/outreach/provider-config": return self.provider_config(db,user,payload)
+            if path=="/api/v1/ai/provider-config": return self.ai_provider_config(db,user,payload)
             if path=="/api/v1/businesses/bulk-review": return self.bulk_review(payload,db,user)
             bits_ai=path.split("/")
             if len(bits_ai)==7 and bits_ai[:4]==["","api","v1","businesses"] and bits_ai[5]=="ai" and bits_ai[6]=="suggest": return self.suggest_ai(int(bits_ai[4]) if bits_ai[4].isdigit() else -1,payload,db,user)
@@ -938,6 +1033,7 @@ class ApiHandler(BaseHTTPRequestHandler):
             bits_outreach=path.split("/")
             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=="/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)
@@ -1262,6 +1358,41 @@ class ApiHandler(BaseHTTPRequestHandler):
 
     def log_message(self,*_):pass
 
+def _run_scoped_discovery(db, job, handler):
+    payload = json.loads(job["payload"] or "{}")
+    result = scoped_discover(payload.get("criteria", {}), payload.get("seed_urls", []), max_pages=payload.get("max_pages", 20), max_candidates=payload.get("max_candidates", 50))
+    org = job["organization_id"]; persisted = []
+    for candidate in result["candidates"]:
+        b = normalize_business(candidate)
+        suppressed = is_suppressed(b, [dict(r) for r in db.execute("SELECT kind,value FROM suppressions WHERE organization_id=? AND active=1", (org,))])
+        if suppressed or not b["website_domain"]: continue
+        existing = db.execute("SELECT id FROM businesses WHERE organization_id=? AND website_domain=?", (org, b["website_domain"])).fetchone()
+        if existing: bid = existing["id"]
+        else:
+            scored = score_business(b)
+            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) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (org,b["name"],b["website"],b["website_domain"],b["email"],b["phone"],b.get("description", ""),b["province"],b["city"],b["suburb"],scored["score"],scored["score_version"],json.dumps(scored["factors"]),scored["website_class"]))
+            bid = cur.lastrowid
+            priority = "high" if scored["score"] >= 70 else "medium" if scored["score"] >= 40 else "low"
+            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,priority,scored["score_version"],json.dumps(scored["factors"]),json.dumps({"source": "scoped_discovery"}, sort_keys=True)))
+        scan_ids = {}
+        for page in candidate.get("pages", []):
+            scan_key = hashlib.sha256((org + ":" + page["url"]).encode()).hexdigest()
+            scan = db.execute("SELECT id FROM website_scans WHERE organization_id=? AND business_id=? AND cache_key=? ORDER BY id DESC LIMIT 1", (org, bid, scan_key)).fetchone()
+            if scan: scan_ids[page["url"]] = scan["id"]; continue
+            scan_result = {"input_url": page["url"], "final_url": page.get("final_url"), "status": page.get("status"), "title": page.get("title", ""), "headings": page.get("headings", []), "html": page.get("html", ""), "provenance": "scoped_discovery"}
+            cur_scan = db.execute("INSERT INTO website_scans(organization_id,business_id,website_id,input_url,classification,result_json,cache_key,scanned_at,cache_expires_at) VALUES(?,?,?,?,?,?,?,?,?)", (org,bid,None,page["url"],"healthy",json.dumps(scan_result, sort_keys=True),scan_key,datetime.now(timezone.utc).replace(microsecond=0).isoformat(),None))
+            scan_ids[page["url"]] = cur_scan.lastrowid
+        for page in candidate["evidence"]:
+            db.execute("INSERT INTO evidence(business_id,organization_id,kind,url,claim) VALUES(?,?,?,?,?)", (bid, org, page["kind"], page["url"], page["claim"]))
+        for contact in candidate["contacts"]:
+            key = hashlib.sha256((str(job["id"]) + contact["source_url"] + contact["kind"] + contact["value"]).encode()).hexdigest()
+            db.execute("INSERT OR IGNORE INTO contact_extractions(organization_id,business_id,website_scan_id,extraction_key,kind,value,label,classification,confidence,source_url,public_business,mx_status,suppressed,do_not_contact,provenance) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (org,bid,scan_ids.get(contact["source_url"]),key,contact["kind"],contact["value"],contact["label"],contact["classification"],contact["confidence"],contact["source_url"],1,"unknown",int(contact["suppressed"]),int(contact["do_not_contact"]),contact["provenance"]))
+        persisted.append({"business_id": bid, "domain": b["website_domain"], "provenance": candidate["provenance"]})
+    safe_result = dict(result); safe_result["candidates"] = persisted
+    db.execute("UPDATE discovery_runs SET result_json=?,result_count=?,updated_at=CURRENT_TIMESTAMP WHERE organization_id=? AND job_id=?", (json.dumps(safe_result, sort_keys=True), len(persisted), org, job["id"]))
+    handler.add_job_event(db, job["id"], org, "discovery.completed", f"Persisted {len(persisted)} candidates", 100)
+
+
 def _job_worker(server):
     while not server.job_stop.is_set():
         db=connect(server.db_path)
@@ -1275,6 +1406,13 @@ def _job_worker(server):
             server_handler.add_job_event(db,jid,org,"started","Job started",0); db.commit()
             try: payload=json.loads(job["payload"] or "{}")
             except ValueError: payload={}
+            if job["type"] == "scoped_discovery":
+                try:
+                    _run_scoped_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 "DISCOVERY_FAILED", jid)); server_handler.add_job_event(db,jid,org,"failed","Discovery failed",job["progress"],str(exc)[:80]); db.commit()
+                continue
             try: steps=1 if job["type"]=="noop" else max(1,min(int(payload.get("steps",5)),20))
             except (ValueError,TypeError): steps=5
             cancelled=False

apps/api/app/search_provider.py

@@ -0,0 +1,91 @@
+"""Fail-closed, allowlisted HTTP JSON search provider for criteria-first discovery."""
+from __future__ import annotations
+
+import json
+import os
+from urllib.parse import urlparse
+from urllib.request import Request, urlopen
+
+from .website_scanner import validate_url
+
+MAX_RESULTS = 50
+MAX_RESPONSE_BYTES = 64 * 1024
+TIMEOUT_SECONDS = 8
+
+
+class ProviderConfigError(ValueError):
+    """Provider is absent or configured in a way that is unsafe to call."""
+
+
+def _config() -> tuple[str, set[str], str]:
+    endpoint = os.environ.get("SEARCH_PROVIDER_URL", "").strip()
+    allowed = {x.strip().lower().rstrip(".") for x in os.environ.get("SEARCH_PROVIDER_ALLOWED_HOSTS", "").split(",") if x.strip()}
+    api_key = os.environ.get("SEARCH_PROVIDER_API_KEY", "").strip()
+    return endpoint, allowed, api_key
+
+
+def _validated_endpoint() -> tuple[str, str, str]:
+    endpoint, allowed, api_key = _config()
+    if not endpoint:
+        raise ProviderConfigError("not_configured")
+    parsed = urlparse(endpoint)
+    host = (parsed.hostname or "").lower().rstrip(".")
+    if parsed.scheme != "https" or not host or parsed.username or parsed.password or parsed.fragment or host not in allowed:
+        raise ProviderConfigError("unsafe_provider")
+    return endpoint, host, api_key
+
+
+def provider_status() -> dict[str, object]:
+    endpoint = os.environ.get("SEARCH_PROVIDER_URL", "").strip()
+    if not endpoint:
+        return {"provider": "generic_http_json", "status": "not_configured", "configured": False, "network_enabled": False}
+    try:
+        _, host, _ = _validated_endpoint()
+    except ProviderConfigError as exc:
+        return {"provider": "generic_http_json", "status": "unsafe_configured", "configured": False, "network_enabled": False, "error": str(exc)}
+    return {"provider": "generic_http_json", "status": "ready", "configured": True, "network_enabled": True, "host": host, "max_results": MAX_RESULTS}
+
+
+def _result_urls(payload: object, limit: int) -> list[str]:
+    items = payload.get("results", payload.get("items", [])) if isinstance(payload, dict) else []
+    if not isinstance(items, list):
+        raise ProviderConfigError("invalid_provider_response")
+    urls: list[str] = []
+    for item in items[:limit]:
+        raw = item.get("url", item.get("website", item.get("link", ""))) if isinstance(item, dict) else item
+        if not isinstance(raw, str) or not raw.strip():
+            continue
+        if urlparse(raw.strip()).scheme != "https":
+            continue
+        try:
+            safe = validate_url(raw.strip())
+        except (TypeError, ValueError):
+            continue
+        if safe not in urls:
+            urls.append(safe)
+    return urls
+
+
+def search(criteria: dict, limit: int) -> list[str]:
+    endpoint, _, api_key = _validated_endpoint()
+    try:
+        bounded = max(1, min(int(limit), MAX_RESULTS))
+    except (TypeError, ValueError) as exc:
+        raise ProviderConfigError("invalid_limits") from exc
+    body = json.dumps({"criteria": criteria, "limit": bounded}, separators=(",", ":"), ensure_ascii=False).encode()
+    headers = {"Content-Type": "application/json", "Accept": "application/json"}
+    if api_key:
+        headers["Authorization"] = "Bearer " + api_key
+    request = Request(endpoint, data=body, headers=headers, method="POST")
+    try:
+        with urlopen(request, timeout=TIMEOUT_SECONDS) as response:
+            raw = response.read(MAX_RESPONSE_BYTES + 1)
+    except Exception as exc:
+        raise ProviderConfigError("provider_unavailable") from exc
+    if len(raw) > MAX_RESPONSE_BYTES:
+        raise ProviderConfigError("provider_response_too_large")
+    try:
+        payload = json.loads(raw.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ProviderConfigError("invalid_provider_response") from exc
+    return _result_urls(payload, bounded)

apps/api/schema.sql

@@ -281,6 +281,17 @@ CREATE TABLE IF NOT EXISTS ai_suggestions (
 );
 CREATE INDEX IF NOT EXISTS idx_ai_suggestions_run ON ai_suggestions(organization_id,ai_run_id,id);
 
+-- AI provider settings contain no secrets. Remote execution remains unavailable
+-- unless a separately reviewed provider abstraction is added.
+CREATE TABLE IF NOT EXISTS ai_provider_configs (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ organization_id TEXT NOT NULL REFERENCES organizations(id) ON DELETE CASCADE,
+ provider TEXT NOT NULL DEFAULT 'local', enabled INTEGER NOT NULL DEFAULT 1,
+ reviewed INTEGER NOT NULL DEFAULT 0, policy_json TEXT NOT NULL DEFAULT '{"network_enabled":false}',
+ created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ UNIQUE(organization_id)
+);
+
 -- Phase 14 outreach preparation. Provider configuration is metadata plus a
 -- one-way secret fingerprint; outbound transport is intentionally disabled.
 CREATE TABLE IF NOT EXISTS outreach_provider_configs (
@@ -313,3 +324,16 @@ CREATE TABLE IF NOT EXISTS outreach_drafts (
 );
 CREATE INDEX IF NOT EXISTS idx_outreach_drafts_org ON outreach_drafts(organization_id,created_at DESC,id DESC);
 CREATE INDEX IF NOT EXISTS idx_outreach_drafts_business ON outreach_drafts(organization_id,business_id,id DESC);
+
+-- 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,
+ 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,
+ UNIQUE(organization_id,job_id)
+);
+CREATE INDEX IF NOT EXISTS idx_discovery_runs_org ON discovery_runs(organization_id,created_at DESC,id DESC);

apps/api/tests/test_phase13_api.py

@@ -49,6 +49,36 @@ class Phase13ApiTests(unittest.TestCase):
         self.assertEqual(status, "not_configured"); self.assertEqual(provider, ""); self.assertEqual(version, "")
         self.assertNotIn("output", metadata)
 
+    def test_structured_enrichment_contract_has_classification_priority_confidence_uncertainty_and_provenance(self):
+        status, business = self.request("POST", "/api/v1/businesses", {"name": "Structured Co", "website": "https://structured.test"}); self.assertEqual(status, 201)
+        bid = business["id"]
+        self.request("POST", f"/api/v1/businesses/{bid}/evidence", {"kind": "source", "url": "https://source.test", "claim": "Provides solar installation"})
+        status, run = self.request("POST", f"/api/v1/businesses/{bid}/ai/suggest", {}); self.assertEqual(status, 201)
+        output = run["output"]
+        for key in ("classification", "summary", "priority_recommendation", "confidence", "uncertainty", "conflicts", "citations", "policy"):
+            self.assertIn(key, output)
+        self.assertTrue(output["citations"][0]["provenance"])
+        self.assertTrue(output["citations"][0]["hash"])
+        self.assertEqual(output["policy"]["claim_policy"], "stored_evidence_only")
+
+    def test_approval_rejects_stale_source_hashes(self):
+        _, business = self.request("POST", "/api/v1/businesses", {"name": "Stale Co"})
+        bid = business["id"]
+        self.request("POST", f"/api/v1/businesses/{bid}/evidence", {"kind": "source", "url": "https://source.test", "claim": "Original claim"})
+        _, run = self.request("POST", f"/api/v1/businesses/{bid}/ai/suggest", {})
+        self.request("POST", f"/api/v1/businesses/{bid}/evidence", {"kind": "source", "url": "https://source.test", "claim": "Changed claim"})
+        status, result = self.request("POST", f"/api/v1/ai-runs/{run['id']}/approve", {})
+        self.assertEqual(status, 409); self.assertEqual(result["error"], "ai_run_stale")
+
+    def test_remote_provider_is_not_configured_and_status_never_makes_network_call(self):
+        os.environ["AI_PROVIDER"] = "openai"
+        status, provider, version, metadata = generate({"name": "Remote"}, [], [], [])
+        self.assertEqual(status, "not_configured"); self.assertEqual(provider, "openai"); self.assertNotIn("output", metadata)
+        status, result = self.request("GET", "/api/v1/ai/provider-config")
+        self.assertEqual(status, 200); self.assertEqual(result["status"], "not_configured"); self.assertFalse(result["network_enabled"])
+        status, result = self.request("POST", "/api/v1/ai/provider-config", {"provider": "openai", "enabled": True, "reviewed": True})
+        self.assertEqual(status, 409); self.assertFalse(result["network_enabled"])
+
     def test_endpoint_persists_citations_and_approval_without_crm_write(self):
         status, business = self.request("POST", "/api/v1/businesses", {"name": "Evidence Co", "website": "https://evidence.test"}); self.assertEqual(status, 201)
         bid = business["id"]

apps/api/tests/test_scoped_discovery.py

@@ -0,0 +1,94 @@
+import json
+import os
+import sqlite3
+import threading
+import time
+import unittest
+from http.client import HTTPConnection
+from tempfile import TemporaryDirectory
+from unittest.mock import patch
+
+from app.main import create_server, hash_password
+
+
+class ScopedDiscoveryApiTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = TemporaryDirectory()
+        for key in ('SEARCH_PROVIDER_URL', 'SEARCH_PROVIDER_ALLOWED_HOSTS', 'SEARCH_PROVIDER_API_KEY'):
+            os.environ.pop(key, None)
+        os.environ['BOOTSTRAP_ADMIN_EMAIL'] = 'discover-owner@example.test'
+        os.environ['BOOTSTRAP_ADMIN_PASSWORD'] = 'password'
+        self.server = create_server('127.0.0.1', 0, self.tmp.name + '/db.sqlite')
+        self.thread = threading.Thread(target=self.server.serve_forever, daemon=True); self.thread.start()
+        self.conn = HTTPConnection('127.0.0.1', self.server.server_port, timeout=4); self.cookie = None
+        self.request('POST', '/api/v1/auth/login', {'email': 'discover-owner@example.test', 'password': 'password'})
+
+    def tearDown(self):
+        self.server.shutdown(); self.server.server_close(); self.thread.join(2); self.tmp.cleanup()
+
+    def request(self, method, path, payload=None):
+        body = json.dumps(payload).encode() if payload is not None else None
+        headers = {'Content-Type': 'application/json'} if body else {}
+        if self.cookie: headers['Cookie'] = self.cookie
+        self.conn.request(method, path, body, headers); response = self.conn.getresponse()
+        cookie = response.getheader('Set-Cookie')
+        if cookie: self.cookie = cookie.split(';', 1)[0]
+        return response.status, json.loads(response.read() or b'{}')
+
+    def test_direct_criteria_job_crawls_allowlisted_site_and_persists_evidence(self):
+        pages = {
+            'https://directory.test/': {'status': 200, 'final_url': 'https://directory.test/', 'content_type': 'text/html', 'body': b'<h1>Acme Solar</h1><a href="https://acme.test/">Acme</a>'},
+            'https://acme.test/': {'status': 200, 'final_url': 'https://acme.test/', 'content_type': 'text/html', 'body': b'<title>Acme Solar</title><h1>Acme Solar</h1><p>Solar installers</p><a href="/contact">Contact</a>'},
+            'https://acme.test/contact': {'status': 200, 'final_url': 'https://acme.test/contact', 'content_type': 'text/html', 'body': b'<h1>Contact Acme</h1><a href="mailto:hello@acme.test">Email</a><p>+27 11 555 0100</p>'},
+        }
+        def fetch(url, **_):
+            value = pages[url]; return dict(value, redirect_chain=[], elapsed_ms=1, tls=url.startswith('https://'), certificate_status='valid')
+        with patch('app.discovery._fetch', side_effect=fetch), patch('app.discovery.validate_url', side_effect=lambda url, **_: url), patch('app.main.validate_url', side_effect=lambda url, **_: url):
+            status, created = self.request('POST', '/api/v1/discovery', {'criteria': {'keywords': ['solar'], 'location': 'Cape Town'}, 'seed_urls': ['https://directory.test/'], 'max_pages': 5, 'idempotency_key': 'scope-1'})
+            self.assertEqual(status, 202); self.assertEqual(created['type'], 'scoped_discovery')
+            for _ in range(50):
+                _, job = self.request('GET', '/api/v1/jobs/' + str(created['id']))
+                if job['status'] in ('succeeded', 'failed'): break
+                time.sleep(.02)
+        self.assertEqual(job['status'], 'succeeded')
+        status, runs = self.request('GET', '/api/v1/discovery-runs')
+        self.assertEqual(status, 200); self.assertEqual(runs['items'][0]['criteria']['location'], 'Cape Town')
+        self.assertEqual(runs['items'][0]['result_count'], 1)
+        status, businesses = self.request('GET', '/api/v1/businesses')
+        self.assertEqual(status, 200); self.assertEqual(businesses['items'][0]['website_domain'], 'acme.test')
+        detail = self.request('GET', '/api/v1/businesses/' + str(businesses['items'][0]['id']))[1]
+        self.assertTrue(any(x['url'] == 'https://acme.test/contact' for x in detail['evidence']))
+        self.assertTrue(any(x['source_url'] == 'https://acme.test/contact' and x['provenance'] == 'mailto' for x in detail['contact_extractions']))
+        self.assertFalse(detail.get('outreach_enabled', False))
+
+    def test_requires_bounded_seed_allowlist_and_rejects_ssrf(self):
+        status, body = self.request('POST', '/api/v1/discovery', {'criteria': {'keywords': ['x']}})
+        self.assertEqual(status, 503); self.assertEqual(body['error'], 'not_configured')
+        status, body = self.request('POST', '/api/v1/discovery', {'criteria': {}, 'seed_urls': ['http://127.0.0.1/'], 'idempotency_key': 'bad'})
+        self.assertEqual(status, 400); self.assertEqual(body['error'], 'unsafe_seed_url')
+
+    def test_criteria_first_search_results_flow_through_existing_job_persistence(self):
+        os.environ['SEARCH_PROVIDER_URL'] = 'https://search.example.test/query'
+        os.environ['SEARCH_PROVIDER_ALLOWED_HOSTS'] = 'search.example.test'
+        pages = {'https://acme.test/': {'status': 200, 'final_url': 'https://acme.test/', 'content_type': 'text/html', 'body': b'<title>Acme Solar</title><h1>Acme Solar</h1><p>solar</p>'}}
+        def fetch(url, **_):
+            value = pages[url]; return dict(value, redirect_chain=[], elapsed_ms=1, tls=True, certificate_status='valid')
+        with patch('app.discovery.search_provider', return_value=['https://acme.test/']) as provider, patch('app.discovery._fetch', side_effect=fetch), patch('app.discovery.validate_url', side_effect=lambda url, **_: url), patch('app.main.validate_url', side_effect=lambda url, **_: url):
+            status, created = self.request('POST', '/api/v1/discovery', {'criteria': {'keywords': ['solar']}, 'max_candidates': 1, 'idempotency_key': 'criteria-1'})
+            self.assertEqual(status, 202)
+            for _ in range(50):
+                _, job = self.request('GET', '/api/v1/jobs/' + str(created['id']))
+                if job['status'] in ('succeeded', 'failed'): break
+                time.sleep(.02)
+        self.assertEqual(job['status'], 'succeeded'); provider.assert_called_once_with({'keywords': ['solar']}, 1)
+        run = self.request('GET', '/api/v1/discovery-runs')[1]['items'][0]
+        self.assertEqual(run['seed_urls'], []); self.assertEqual(run['result']['candidates'][0]['provenance']['mechanism'], 'criteria_search_provider')
+
+    def test_results_are_tenant_isolated(self):
+        ph, salt = hash_password('other-password')
+        db = sqlite3.connect(self.server.db_path); db.execute("INSERT INTO organizations VALUES ('other-tenant','Other',CURRENT_TIMESTAMP)"); db.execute("INSERT INTO users (organization_id,email,password_hash,password_salt,role) VALUES (?,?,?,?,?)", ('other-tenant','other@example.test',ph,salt,'owner')); db.commit(); db.close()
+        self.cookie = None; self.request('POST', '/api/v1/auth/login', {'email': 'other@example.test', 'password': 'other-password'})
+        self.assertEqual(self.request('GET', '/api/v1/discovery-runs')[1]['items'], [])
+
+
+if __name__ == '__main__': unittest.main()

apps/api/tests/test_search_provider.py

@@ -0,0 +1,44 @@
+import os
+import unittest
+from unittest.mock import patch
+
+from app.search_provider import ProviderConfigError, provider_status, search
+
+
+class SearchProviderTests(unittest.TestCase):
+    def tearDown(self):
+        for key in ("SEARCH_PROVIDER_URL", "SEARCH_PROVIDER_ALLOWED_HOSTS", "SEARCH_PROVIDER_API_KEY"):
+            os.environ.pop(key, None)
+
+    def test_not_configured_is_fail_closed(self):
+        self.assertEqual(provider_status()["status"], "not_configured")
+        with self.assertRaisesRegex(ProviderConfigError, "not_configured"):
+            search({"keywords": ["solar"]}, 5)
+
+    def test_unsafe_provider_is_rejected(self):
+        os.environ["SEARCH_PROVIDER_URL"] = "http://search.example.test/query"
+        os.environ["SEARCH_PROVIDER_ALLOWED_HOSTS"] = "search.example.test"
+        self.assertEqual(provider_status()["status"], "unsafe_configured")
+        with self.assertRaisesRegex(ProviderConfigError, "unsafe_provider"):
+            search({}, 5)
+
+    def test_successful_mocked_search_returns_bounded_https_urls(self):
+        os.environ["SEARCH_PROVIDER_URL"] = "https://search.example.test/query"
+        os.environ["SEARCH_PROVIDER_ALLOWED_HOSTS"] = "search.example.test"
+        response = type("Response", (), {"__enter__": lambda self: self, "__exit__": lambda self, *args: None, "read": lambda self, *_: b'{"results":[{"url":"https://acme.test"},{"url":"http://bad.test"},{"url":"https://acme.test"}]}'})()
+        with patch("app.search_provider.urlopen", return_value=response), patch("app.search_provider.validate_url", side_effect=lambda url, **_: url):
+            self.assertEqual(search({"keywords": ["solar"]}, 5), ["https://acme.test"])
+
+    def test_limit_is_bounded_and_sent_to_provider(self):
+        os.environ["SEARCH_PROVIDER_URL"] = "https://search.example.test/query"
+        os.environ["SEARCH_PROVIDER_ALLOWED_HOSTS"] = "search.example.test"
+        response = type("Response", (), {"__enter__": lambda self: self, "__exit__": lambda self, *args: None, "read": lambda self, *_: b'{"results":[]}'})()
+        with patch("app.search_provider.urlopen", return_value=response) as opened:
+            self.assertEqual(search({}, 500), [])
+            self.assertEqual(opened.call_args.kwargs["timeout"], 8)
+            request = opened.call_args.args[0]
+            self.assertIn(b'"limit":50', request.data)
+
+
+if __name__ == "__main__":
+    unittest.main()

apps/web/asset-manifest.json

@@ -1,13 +1,22 @@
 {
   "schema": 1,
-  "version": "phase-16",
-  "entrypoints": ["config.js", "app.js", "styles.css"],
-  "publicAssets": ["index.html", "health.html", "error.html", "healthz"],
+  "version": "phase-17",
+  "entrypoints": [
+    "config.js",
+    "app.js",
+    "styles.css"
+  ],
+  "publicAssets": [
+    "index.html",
+    "health.html",
+    "error.html",
+    "healthz"
+  ],
   "integrity": {
     "config.js": "sha256-f9c7b4db3eab4cf54146bd25891b5103b09ae75da93c57b548cf57ae93e4a3f6",
-    "app.js": "sha256-979a3efe97234febdd0a3c1c68c00ef346d1601326ac2202f223e3438532e236",
-    "styles.css": "sha256-7340dccc648fa917fb497ceeba7286d9c4712b6552960de54cef759cdbda6de4",
-    "index.html": "sha256-a40abc8af0b30ea9bb0005897d2231614eae3e26ceda0dc82dab1a16a78a634a",
+    "app.js": "sha256-cf0d8b00773877c09f03707030e3e8113b3f34ad389054e0a97ecb1dbddd139f",
+    "styles.css": "sha256-56ca598c8e8cbbcb91d81013c6a498f75193de79e33dd2d8d117df836fc43d57",
+    "index.html": "sha256-f3c9544c07c08c5248abafaf8075551ee641704f9cb3d516449188f74bf86644",
     "health.html": "sha256-c352a6f37aa24628cfc8d5709a70ff2d94181d192ed5fe00916cca9378a61d81",
     "error.html": "sha256-f3cc28d2dfc9e8af112d0257c6ee47b9e9146e5da8590a754dd589bf43702aaf",
     "healthz": "sha256-dc51b8c96c2d745df3bd5590d990230a482fd247123599548e0632fdbf97fc22"

apps/web/index.html

@@ -44,6 +44,16 @@
       <header class="topbar"><button class="mobile-menu" id="menuBtn" aria-label="Toggle navigation">☰</button><div class="crumb">Workspace <span>/</span> Growth pipeline</div><div class="top-actions"><span class="api-status" id="apiStatus">● Connecting…</span><span class="user-identity" id="userIdentity"></span><button class="icon-button" aria-label="Notifications">♢</button><button class="logout-button" id="logoutBtn" type="button">Log out</button><div class="avatar" id="userAvatar">?</div></div></header>
       <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-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>
+            <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>
+        </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>
           <a class="metric-card metric-link" data-dashboard-filter="review" href="#explorer"><div class="metric-icon amber">◌</div><div><p>Needs review</p><h2 id="metricReview">0</h2><span class="trend neutral">● Human verification</span></div></a>
@@ -55,7 +65,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="qualified">Qualified</option><option value="review">Review</option><option value="suppressed">Suppressed</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="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="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>
@@ -98,7 +108,7 @@
         </section>
         <section class="crm-section" id="crmActivity" aria-labelledby="crmActivityTitle" data-smoke="crm-interactions">
           <div class="crm-header panel"><div><p class="eyebrow">RELATIONSHIP HISTORY</p><h2 id="crmActivityTitle">Interactions &amp; follow-ups</h2><p class="muted">Capture outcomes and next steps without contacting anyone.</p></div></div>
-          <div class="crm-two-col"><article class="panel interaction-panel"><div id="interactionState" class="detail-loading">Select a prospect to load interactions.</div></article><article class="panel"><div class="panel-heading"><div><p class="eyebrow">ADD RECORD</p><h3>Log an interaction</h3></div><span class="small-label">Internal only</span></div><form id="interactionForm" class="crm-form"><label>Type<select name="type"><option value="note">Note</option><option value="call">Call</option><option value="meeting">Meeting</option><option value="email">Email (record only)</option></select></label><label>Outcome<select name="outcome"><option value="">Choose outcome</option><option value="no_response">No response</option><option value="interested">Interested</option><option value="not_a_fit">Not a fit</option><option value="follow_up">Follow-up requested</option></select></label><label>Follow-up date <span class="optional">optional</span><input name="follow_up_at" type="date"></label><label>Summary<textarea name="summary" rows="4" required placeholder="What happened? Keep this an internal record."></textarea></label><p id="interactionMessage" class="form-message" role="status"></p><button class="button primary" type="submit">Save interaction</button></form></article></div>
+          <div class="crm-two-col"><article class="panel interaction-panel"><div id="interactionState" class="detail-loading">Select a prospect to load interactions.</div></article><article class="panel"><div class="panel-heading"><div><p class="eyebrow">ADD RECORD</p><h3>Log an interaction</h3></div><span class="small-label">Internal only</span></div><form id="interactionForm" class="crm-form"><label>Channel<select name="kind"><option value="note">Note</option><option value="phone">Phone</option><option value="email">Email (record only)</option><option value="meeting">Meeting</option><option value="other">Other</option></select></label><label>Outcome<select name="outcome"><option value="other" selected>Other / not specified</option><option value="connected">Connected</option><option value="no_answer">No answer</option><option value="left_message">Left message</option><option value="meeting_booked">Meeting booked</option><option value="meeting_held">Meeting held</option><option value="qualified">Qualified</option><option value="disqualified">Disqualified</option><option value="won">Won</option><option value="lost">Lost</option></select></label><label>Follow-up date <span class="optional">optional</span><input name="follow_up_at" type="date"></label><label>Body<textarea name="body" rows="4" required placeholder="What happened? Keep this an internal record."></textarea></label><p id="interactionMessage" class="form-message" role="status"></p><button class="button primary" type="submit">Save interaction</button></form></article></div>
         </section>
         <section class="crm-section" id="crmReports" aria-labelledby="crmReportsTitle" data-smoke="crm-reports">
           <div class="crm-header panel"><div><p class="eyebrow">REPORTING</p><h2 id="crmReportsTitle">CRM reports</h2><p class="muted">Tenant-scoped pipeline, outcomes, and activity summaries from the API.</p></div><button class="button ghost" id="reportsRefreshBtn" type="button">↻ Refresh reports</button></div>

apps/web/scripts/final-acceptance.mjs

@@ -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', 'crmPipeline', 'pipelineBoard', 'crmActivity',
+  'sources', 'sourcesList', 'sourceRecordsList', 'discoveryForm', 'discoveryWorkspace', 'directDiscoveryForm', 'directDiscoveryRunsState', 'directDiscoveryResultState', 'crmPipeline', 'pipelineBoard', 'crmActivity',
   'interactionState', 'interactionForm', 'crmReports', 'pipelineReport', 'outcomesReport', 'activityReport',
   'suppressionCenter', 'suppressionForm', 'suppressionState', 'outreachSettings', 'providerPolicyPanel',
   'scoreRules', 'scoreRulesPanel', 'scoreDistributionPanel', 'mergeDialog', 'confirmMergeBtn'
@@ -73,7 +73,7 @@ check('dom.dynamic-markers', 'detail safety panels define dynamic smoke markers'
 const routeContracts = [
   '/api/v1/auth/me', '/api/v1/auth/login', '/api/v1/auth/logout', '/api/v1/dashboard/summary', '/api/v1/businesses',
   '/api/v1/review-queue', '/api/v1/saved-filters', '/api/v1/businesses/bulk-review', '/api/v1/jobs', '/api/v1/sources',
-  '/api/v1/source-records', '/api/v1/discovery-queries', '/api/v1/merge-history', '/api/v1/scoring/summary', '/api/v1/score-rules',
+  '/api/v1/source-records', '/api/v1/discovery', '/api/v1/discovery-runs', '/api/v1/discovery-queries', '/api/v1/merge-history', '/api/v1/scoring/summary', '/api/v1/score-rules',
   '/api/v1/pipeline-entries', '/api/v1/interactions', '/api/v1/reports/pipeline', '/api/v1/reports/outcomes', '/api/v1/reports/activity',
   '/api/v1/suppressions', '/api/v1/ai-runs', '/api/v1/outreach/drafts', '/api/v1/outreach/provider-config',
   '/matches', '/domains/check', '/domain-candidates', '/websites/scan', '/contacts/extract', '/score/recalculate', '/pipeline', '/verify'
@@ -82,6 +82,17 @@ check('routes.contracts', 'all critical API route contracts are referenced by th
 check('routes.authenticated', 'protected API requests include cookie credentials', js.includes("credentials:'include'") && js.includes('jsonRequest'));
 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'];
+const crmStagesMatch = js.match(/const crmStages = \[([^\]]+)\]/);
+const crmStages = crmStagesMatch ? [...crmStagesMatch[1].matchAll(/["']([^"']+)["']/g)].map(match => match[1]) : [];
+check('crm.stage-contract', 'CRM stage definitions exactly match the canonical pipeline', JSON.stringify(crmStages) === JSON.stringify(canonicalStages), `actual=${JSON.stringify(crmStages)}`);
+const pipelineControl = js.match(/<select name="stage"[\s\S]*?<\/select>/)?.[0] || '';
+check('crm.stage-control', 'detail stage control exposes only canonical stages', canonicalStages.every(stage => pipelineControl.includes(`value="${stage}"`)) && !pipelineControl.includes('value="review"') && !pipelineControl.includes('value="suppressed"'));
+const interactionControl = html.match(/<form id="interactionForm"[\s\S]*?<\/form>/)?.[0] || '';
+const canonicalOutcomes = ['connected', 'no_answer', 'left_message', 'meeting_booked', 'meeting_held', 'qualified', 'disqualified', 'won', 'lost', 'other'];
+check('crm.interaction-form-contract', 'interaction form uses kind/body and canonical outcomes', interactionControl.includes('name="kind"') && interactionControl.includes('name="body"') && canonicalOutcomes.every(outcome => interactionControl.includes(`value="${outcome}"`)) && !interactionControl.includes('name="type"') && !interactionControl.includes('name="summary"'));
+check('crm.interaction-payload-contract', 'interaction writes use the backend kind/body payload contract', /JSON\.stringify\(\{\s*kind:\s*data\.kind,\s*body:\s*data\.body/.test(js) && !/JSON\.stringify\(\{\.\.\.data,outreach:false\}\)/.test(js));
+
 const manifestAssets = manifest && [...new Set([...(manifest.entrypoints || []), ...(manifest.publicAssets || [])])];
 check('manifest.schema', 'asset manifest has a supported schema and asset lists', Boolean(manifest && manifest.schema === 1 && Array.isArray(manifest.entrypoints) && Array.isArray(manifest.publicAssets) && manifest.integrity && manifestAssets.length), manifest ? '' : 'manifest unavailable');
 if (manifestAssets) {

docker-compose.yml

@@ -15,6 +15,9 @@ services:
       # Optional first-run admin bootstrap; leave unset after provisioning.
       BOOTSTRAP_ADMIN_EMAIL: ${BOOTSTRAP_ADMIN_EMAIL:-}
       BOOTSTRAP_ADMIN_PASSWORD: ${BOOTSTRAP_ADMIN_PASSWORD:-}
+      SEARCH_PROVIDER_URL: ${SEARCH_PROVIDER_URL:-}
+      SEARCH_PROVIDER_ALLOWED_HOSTS: ${SEARCH_PROVIDER_ALLOWED_HOSTS:-}
+      SEARCH_PROVIDER_API_KEY: ${SEARCH_PROVIDER_API_KEY:-}
       AUTOMATED_OUTREACH_ENABLED: "false"
     volumes:
       - prospect_api_data:/data