code/self_healing_finetuning/data_pipeline/live_source_discovery.py

from __future__ import annotations

import html
import json
import os
import re
import urllib.parse
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from n21.config import write_json
from n21.config import load_structured
from n21.settings import CONFIG_ROOT
from observability.audit_log import utc_now
from data_pipeline.source_quality_certifier import certify_source_candidate


OFFICIAL_DOMAIN_HINTS = {
    "sec.gov": "US government public information",
    "investor.gov": "US government public information",
    "federalreserve.gov": "US government public information",
    "treasury.gov": "US government public information",
    "fasb.org": "Publicly available standards-source page; license terms must be reviewed before training use",
    "cfainstitute.org": "Publicly available education page; license terms must be reviewed before training use",
    "stern.nyu.edu": "Publicly available academic page; license terms must be reviewed before training use",
}


ROLE_TERMS = {
    "researcher": [
        "10-K 10-Q annual report financial statement analysis",
        "EDGAR public company disclosure MD&A risk factors",
        "equity valuation earnings per share price earnings ratio",
    ],
    "portfolio_manager": [
        "portfolio construction asset allocation equity portfolio management",
        "diversification risk return rebalancing public investor guide",
    ],
    "risk_manager": [
        "market risk liquidity risk financial statement red flags public company",
        "risk factors disclosure internal controls financial reporting",
    ],
    "performance_manager": [
        "investment performance attribution benchmark risk adjusted return",
        "portfolio performance measurement tracking error information ratio",
    ],
    "client_portfolio_manager": [
        "client investment reporting suitability disclosure portfolio risk",
        "investor education diversification risk tolerance asset allocation",
    ],
    "chief_investment_officer": [
        "capital market assumptions asset allocation investment policy",
        "market outlook financial stability risk premium investment committee",
    ],
}


ASSET_TERMS = {
    "equity": ["equity research", "public companies", "stock analysis"],
    "fixed_income": ["fixed income", "bond yield credit risk", "interest rates"],
    "multi_asset": ["multi asset", "asset allocation", "cross asset risk"],
}


RELEVANCE_TERMS = {
    "equity": [
        "10-k",
        "10-q",
        "8-k",
        "annual report",
        "financial statement",
        "edgar",
        "earnings",
        "valuation",
        "stock",
        "public compan",
        "disclosure",
    ],
    "fixed_income": [
        "bond",
        "yield",
        "duration",
        "credit",
        "interest rate",
        "treasury",
        "fixed income",
        "default",
        "spread",
    ],
    "multi_asset": [
        "asset allocation",
        "diversification",
        "portfolio",
        "risk",
        "financial stability",
        "capital market",
        "cross asset",
    ],
}


DEFAULT_SEED_PAGES = {
    "equity": [
        "https://www.investor.gov/search?keys=10-k",
        "https://www.investor.gov/search?keys=10-q",
        "https://www.investor.gov/search?keys=edgar",
        "https://www.investor.gov/search?keys=financial%20statements",
        "https://www.sec.gov/investor/pubs_annote.shtml",
    ],
    "fixed_income": [
        "https://www.investor.gov/search?keys=bonds",
        "https://www.investor.gov/search?keys=fixed%20income",
        "https://www.federalreserve.gov/publications.htm",
        "https://home.treasury.gov/policy-issues/financing-the-government/interest-rate-statistics",
    ],
    "multi_asset": [
        "https://www.investor.gov/search?keys=asset%20allocation",
        "https://www.investor.gov/search?keys=diversification",
        "https://www.federalreserve.gov/publications/financial-stability-report.htm",
    ],
}


SEC_API_SEED_CIKS = {
    "equity": [
        "0000320193",  # Apple
        "0000789019",  # Microsoft
        "0001045810",  # NVIDIA
        "0001018724",  # Amazon
        "0001318605",  # Tesla
    ]
}


DIRECT_FALLBACK_SOURCES = {
    ("equity", "researcher"): [
        {
            "title": "Investor.gov Form 10-K",
            "url": "https://www.investor.gov/additional-resources/general-resources/glossary/form-10-k",
            "source_type": "html",
            "rationale": "Official Investor.gov glossary source explaining Form 10-K annual reports and audited financial statements.",
        },
        {
            "title": "Investor.gov Form 10-Q",
            "url": "https://www.investor.gov/additional-resources/general-resources/glossary/form-10-q",
            "source_type": "html",
            "rationale": "Official Investor.gov glossary source explaining quarterly reports, unaudited financial statements, and EDGAR filtering.",
        },
        {
            "title": "Investor.gov Using EDGAR to Research Public Companies",
            "url": "https://www.investor.gov/researching-managing-investments/researching-investments/using-edgar-researching-public-companies",
            "source_type": "html",
            "rationale": "Official Investor.gov guide to researching 10-K, 10-Q, 8-K, proxy, and ownership filings.",
        },
        {
            "title": "Investor.gov Form 8-K",
            "url": "https://www.investor.gov/introduction-investing/investing-basics/glossary/form-8-k",
            "source_type": "html",
            "rationale": "Official Investor.gov source for material current-report events used in equity research monitoring.",
        },
        {
            "title": "Investor.gov Market Capitalization",
            "url": "https://www.investor.gov/additional-resources/general-resources/glossary/market-capitalization",
            "source_type": "html",
            "rationale": "Official Investor.gov source for market capitalization, a core public-equity sizing and valuation concept.",
        },
        {
            "title": "Investor.gov Net Income",
            "url": "https://www.investor.gov/introduction-investing/investing-basics/glossary/net-income",
            "source_type": "html",
            "rationale": "Official Investor.gov source for net income, a core profitability and valuation input.",
        },
    ],
    ("equity", "all"): [
        {
            "title": "Investor.gov Diversification",
            "url": "https://www.investor.gov/introduction-investing/investing-basics/glossary/diversification",
            "source_type": "html",
            "rationale": "Official Investor.gov source for diversification across equity super-agent workflows.",
        },
        {
            "title": "Investor.gov Risk",
            "url": "https://www.investor.gov/introduction-investing/investing-basics/glossary/risk",
            "source_type": "html",
            "rationale": "Official Investor.gov source for investment risk terminology.",
        },
    ],
    ("fixed_income", "all"): [
        {
            "title": "Investor.gov Bonds",
            "url": "https://www.investor.gov/introduction-investing/investing-basics/investment-products/bonds-or-fixed-income-products/bonds",
            "source_type": "html",
            "rationale": "Official Investor.gov source for bond structure, prices, yields, and fixed-income risks.",
        },
        {
            "title": "Investor.gov Interest Rate Risk",
            "url": "https://www.investor.gov/introduction-investing/investing-basics/glossary/interest-rate-risk",
            "source_type": "html",
            "rationale": "Official Investor.gov source for interest-rate risk in fixed-income analysis.",
        },
    ],
    ("multi_asset", "all"): [
        {
            "title": "Investor.gov Asset Allocation",
            "url": "https://www.investor.gov/introduction-investing/investing-basics/glossary/asset-allocation",
            "source_type": "html",
            "rationale": "Official Investor.gov source for asset allocation and portfolio construction.",
        },
        {
            "title": "Investor.gov Diversification",
            "url": "https://www.investor.gov/introduction-investing/investing-basics/glossary/diversification",
            "source_type": "html",
            "rationale": "Official Investor.gov source for diversification across asset classes.",
        },
    ],
}


@dataclass(frozen=True)
class Candidate:
    title: str
    url: str
    source_type: str
    license_hint: str
    rationale: str
    query: str
    provider: str
    rank: int
    score: float
    ai_certification: dict[str, Any]


def _request_text(url: str, *, timeout: int = 20) -> str:
    user_agent = os.environ.get(
        "SHFT_HTTP_USER_AGENT",
        "Linvest21-SHFT-live-source-discovery/1.0 (public-source transparency manifest; contact=linvest21)",
    )
    request = urllib.request.Request(
        url,
        headers={
            "User-Agent": user_agent,
            "Accept": "text/html,application/xhtml+xml,text/plain,*/*;q=0.8",
        },
    )
    with urllib.request.urlopen(request, timeout=timeout) as response:
        return response.read(2_000_000).decode("utf-8", errors="ignore")


def _extract_links(markup: str) -> list[tuple[str, str]]:
    links: list[tuple[str, str]] = []
    for match in re.finditer(r'<a\b[^>]*href=["\']([^"\']+)["\'][^>]*>(.*?)</a>', markup, flags=re.I | re.S):
        href = html.unescape(match.group(1))
        text = re.sub(r"<[^>]+>", " ", match.group(2))
        text = re.sub(r"\s+", " ", html.unescape(text)).strip()
        if href.startswith("/l/?") or "duckduckgo.com/l/?" in href:
            parsed = urllib.parse.urlparse(href)
            params = urllib.parse.parse_qs(parsed.query)
            href = params.get("uddg", [href])[0]
        links.append((href, text))
    return links


def _source_type(url: str) -> str:
    path = urllib.parse.urlparse(url).path.lower()
    if path.endswith(".pdf"):
        return "pdf"
    if path.endswith(".txt"):
        return "txt"
    if path.endswith(".md"):
        return "md"
    if path.endswith(".jsonl"):
        return "jsonl"
    return "html"


def _domain(url: str) -> str:
    host = urllib.parse.urlparse(url).netloc.lower()
    if host.startswith("www."):
        host = host[4:]
    return host


def _normalize_url(url: str) -> str:
    return url.split("#", 1)[0].strip()


def _absolute_url(base_url: str, href: str) -> str:
    return _normalize_url(urllib.parse.urljoin(base_url, href))


def _license_hint(url: str) -> str:
    host = _domain(url)
    for domain, hint in OFFICIAL_DOMAIN_HINTS.items():
        if host == domain or host.endswith(f".{domain}"):
            return hint
    return "Publicly available source; license terms must be reviewed before training use"


def _allowed_domain(url: str, domains: list[str]) -> bool:
    host = _domain(url)
    return any(host == domain or host.endswith(f".{domain}") for domain in domains)


def _relevant_to_asset_role(url: str, title: str, asset_class: str, role: str) -> bool:
    haystack = f"{url} {title}".lower().replace("%20", " ")
    asset_hits = RELEVANCE_TERMS.get(asset_class, [asset_class.replace("_", " ")])
    role_hits = ROLE_TERMS.get(role, [role.replace("_", " ")])
    if any(term.lower() in haystack for term in asset_hits):
        return True
    compact_role_terms = [token for phrase in role_hits for token in re.findall(r"[a-z0-9-]{4,}", phrase.lower())]
    return any(token in haystack for token in compact_role_terms[:12])


def _looks_like_navigation_or_index(url: str, title: str) -> bool:
    parsed = urllib.parse.urlparse(url)
    path = parsed.path.lower().rstrip("/")
    query = urllib.parse.parse_qs(parsed.query.lower())
    title_norm = re.sub(r"\s+", " ", title.lower()).strip()
    if path.endswith("/search") or path == "/search":
        return True
    if "keys" in query and "page" in query:
        return True
    if title_norm in {"skip to main content", "current page 1"}:
        return True
    if re.fullmatch(r"page \d+", title_norm):
        return True
    if title_norm in {"online publications", "filings forms", "sec filings", "company filings"}:
        return True
    if "formulario" in title_norm or "/espanol/" in path:
        return True
    return False


def _score(url: str, title: str, query: str, domains: list[str]) -> float:
    host = _domain(url)
    score = 0.0
    if _allowed_domain(url, domains):
        score += 5.0
    if host in {"sec.gov", "investor.gov"} or host.endswith(".sec.gov") or host.endswith(".investor.gov"):
        score += 3.0
    if _source_type(url) == "pdf":
        score += 1.0
    haystack = f"{title} {url}".lower()
    for term in re.findall(r"[a-z0-9-]{3,}", query.lower()):
        if term in haystack:
            score += 0.15
    return round(score, 4)


def _fallback_sources_for(asset_class: str, role: str) -> list[dict[str, Any]]:
    sources: list[dict[str, Any]] = []
    for key in ((asset_class, role), (asset_class, "all")):
        sources.extend(DIRECT_FALLBACK_SOURCES.get(key, []))
    return sources


def _reasoning_frame_terms(asset_class: str, role: str) -> list[str]:
    path = CONFIG_ROOT / "data" / "reasoning_frames.json"
    if not path.exists():
        return []
    try:
        frames = load_structured(path)
    except Exception:
        return []
    terms = frames.get(asset_class, {}).get(role, {}).get("discovery_terms", [])
    return [str(term) for term in terms if str(term).strip()]


def build_queries(
    asset_class: str,
    role: str,
    policy: dict[str, Any],
    quality_errors: list[str] | None = None,
    discovery_attempt: int = 0,
) -> list[str]:
    discovery = policy.get("live_discovery", {})
    templates = discovery.get("query_templates") or [
        "site:{domain} {asset_terms} {role_terms}",
        "site:{domain} investor bulletin {asset_terms} {role_terms}",
        "site:{domain} PDF {asset_terms} {role_terms}",
    ]
    domains = discovery.get("preferred_domains") or ["sec.gov", "investor.gov"]
    asset_terms = ASSET_TERMS.get(asset_class, [asset_class.replace("_", " ")])
    role_terms = ROLE_TERMS.get(role, [role.replace("_", " ")])
    frame_terms = _reasoning_frame_terms(asset_class, role)
    blockers = " ".join(quality_errors or [])
    if "critical_pass" in blockers:
        role_terms = [
            *frame_terms,
            *role_terms,
            "red flags checklist case study pass fail reasoning",
            "worked examples critical decision financial statement analysis",
            "reject approve because accounting quality warning signs",
        ]
    if discovery_attempt > 0:
        retry_terms = discovery.get("retry_query_terms") or [
            "annual report MD&A risk factors audited financial statements",
            "investor bulletin financial statements valuation disclosure",
            "public company analysis cash flow return on invested capital",
            "equity research accounting quality capital allocation moat",
        ]
        # Rotate broader retry terms forward after failed/no-trainable attempts so
        # later searches are not just the same exhausted candidate list.
        rotate = (discovery_attempt - 1) % len(retry_terms)
        ordered_retry_terms = list(retry_terms[rotate:]) + list(retry_terms[:rotate])
        role_terms = [*ordered_retry_terms[:3], *role_terms]
    queries: list[str] = []
    for domain in domains:
        for asset_term in asset_terms[:2]:
            for role_term in role_terms[:3]:
                for template in templates:
                    queries.append(
                        template.format(
                            domain=domain,
                            asset_terms=asset_term,
                            role_terms=role_term,
                            asset_class=asset_class.replace("_", " "),
                            role=role.replace("_", " "),
                        )
                    )
    seen: set[str] = set()
    unique: list[str] = []
    for query in queries:
        compact = re.sub(r"\s+", " ", query).strip()
        if compact and compact.lower() not in seen:
            seen.add(compact.lower())
            unique.append(compact)
    return unique[: int(discovery.get("max_queries", 24))]


def search_duckduckgo(query: str, *, max_results: int) -> list[tuple[str, str]]:
    url = "https://duckduckgo.com/html/?" + urllib.parse.urlencode({"q": query})
    markup = _request_text(url)
    links: list[tuple[str, str]] = []
    seen: set[str] = set()
    for href, text in _extract_links(markup):
        if not href.startswith("http://") and not href.startswith("https://"):
            continue
        clean = href.split("#", 1)[0]
        if clean.lower() in seen:
            continue
        seen.add(clean.lower())
        links.append((clean, text or clean))
        if len(links) >= max_results:
            break
    return links


def search_sec_submissions_api(asset_class: str, role: str, *, max_results: int) -> list[tuple[str, str]]:
    results: list[tuple[str, str]] = []
    for cik in SEC_API_SEED_CIKS.get(asset_class, [])[: max(1, max_results)]:
        api_url = f"https://data.sec.gov/submissions/CIK{cik}.json"
        try:
            payload = json.loads(_request_text(api_url))
        except Exception:
            continue
        filings = payload.get("filings", {}).get("recent", {})
        forms = filings.get("form", [])
        accessions = filings.get("accessionNumber", [])
        primary_docs = filings.get("primaryDocument", [])
        company = str(payload.get("name") or cik)
        for form, accession, primary_doc in zip(forms, accessions, primary_docs):
            if form not in {"10-K", "10-Q", "8-K"}:
                continue
            accession_compact = str(accession).replace("-", "")
            filing_url = f"https://www.sec.gov/Archives/edgar/data/{int(cik)}/{accession_compact}/{primary_doc}"
            title = f"SEC EDGAR {company} {form} filing"
            results.append((filing_url, title))
            if len(results) >= max_results:
                return results
    return results


def discover_public_sources(
    *,
    asset_class: str,
    role: str,
    policy: dict[str, Any],
    output_dir: Path,
    exclude_urls: set[str] | None = None,
    quality_errors: list[str] | None = None,
    discovery_attempt: int = 0,
) -> dict[str, Any]:
    discovery = policy.get("live_discovery", {})
    output_dir.mkdir(parents=True, exist_ok=True)
    preferred_domains = list(discovery.get("preferred_domains") or ["sec.gov", "investor.gov"])
    max_results = int(discovery.get("max_results", 40))
    max_results_per_query = int(discovery.get("max_results_per_query", 8))
    exclude = {url.strip().lower() for url in (exclude_urls or set()) if url.strip()}
    candidates: list[Candidate] = []
    search_errors: list[dict[str, Any]] = []
    query_diagnostics: list[dict[str, Any]] = []
    fallback_diagnostics: list[dict[str, Any]] = []
    provider_name = str(discovery.get("provider", "sec_api_first")).lower()
    duckduckgo_enabled = bool(
        discovery.get("duckduckgo_fallback_enabled", provider_name in {"duckduckgo", "duckduckgo_html", "search"})
    )
    queries = build_queries(
        asset_class,
        role,
        policy,
        quality_errors=quality_errors,
        discovery_attempt=discovery_attempt,
    )
    seen: set[str] = set(exclude)

    def consider_url(
        *,
        url: str,
        title: str,
        query: str,
        provider: str,
        rank: int,
        require_relevance: bool = False,
        diagnostics: dict[str, Any] | None = None,
        rationale: str | None = None,
        source_type: str | None = None,
    ) -> bool:
        clean = _normalize_url(url)
        key = clean.lower()
        if key in seen:
            if diagnostics is not None:
                diagnostics["skipped_duplicate_or_excluded"] = diagnostics.get("skipped_duplicate_or_excluded", 0) + 1
            return False
        parsed = urllib.parse.urlparse(clean)
        if parsed.scheme != "https":
            if diagnostics is not None:
                diagnostics["skipped_non_https"] = diagnostics.get("skipped_non_https", 0) + 1
            seen.add(key)
            return False
        if not _allowed_domain(clean, preferred_domains):
            if diagnostics is not None:
                diagnostics["skipped_domain"] = diagnostics.get("skipped_domain", 0) + 1
            seen.add(key)
            return False
        if _looks_like_navigation_or_index(clean, title):
            if diagnostics is not None:
                diagnostics["skipped_navigation_or_index"] = diagnostics.get("skipped_navigation_or_index", 0) + 1
            seen.add(key)
            return False
        if any(hint in clean.lower() for hint in policy.get("source_risk", {}).get("disallowed_url_hints", [])):
            if diagnostics is not None:
                diagnostics["skipped_disallowed_url_hint"] = diagnostics.get("skipped_disallowed_url_hint", 0) + 1
            seen.add(key)
            return False
        if require_relevance and not _relevant_to_asset_role(clean, title, asset_class, role):
            if diagnostics is not None:
                diagnostics["skipped_low_relevance"] = diagnostics.get("skipped_low_relevance", 0) + 1
            seen.add(key)
            return False
        seen.add(key)
        score = _score(clean, title, query, preferred_domains)
        if provider == "direct_official_fallback":
            score += 10.0
        source_rationale = rationale or f"Live-discovered public source for {asset_class}/{role}; query={query}"
        ai_certification = certify_source_candidate(
            asset_class=asset_class,
            role=role,
            title=title,
            url=clean,
            source_type=source_type or _source_type(clean),
            rationale=source_rationale,
            quality_errors=quality_errors,
            policy=policy,
        )
        if ai_certification.get("intended_use") == "reject":
            if diagnostics is not None:
                diagnostics["skipped_ai_rejected"] = diagnostics.get("skipped_ai_rejected", 0) + 1
            return False
        candidates.append(
            Candidate(
                title=title[:140] or parsed.path.rsplit("/", 1)[-1] or parsed.netloc,
                url=clean,
                source_type=source_type or _source_type(clean),
                license_hint=_license_hint(clean),
                rationale=source_rationale,
                query=query,
                provider=provider,
                rank=rank,
                score=score,
                ai_certification=ai_certification,
            )
        )
        if diagnostics is not None:
            diagnostics["accepted"] = diagnostics.get("accepted", 0) + 1
        return True

    for query in queries:
        if provider_name in {"sec_api_first", "api_first", "sec_api"}:
            diagnostics = {
                "query": query,
                "provider": "sec_submissions_api",
                "raw_result_count": 0,
                "accepted": 0,
                "skipped_duplicate_or_excluded": 0,
                "skipped_non_https": 0,
                "skipped_domain": 0,
                "skipped_navigation_or_index": 0,
                "skipped_disallowed_url_hint": 0,
                "skipped_ai_rejected": 0,
            }
            try:
                results = search_sec_submissions_api(asset_class, role, max_results=max_results_per_query)
            except Exception as exc:
                search_errors.append({"query": query, "provider": "sec_submissions_api", "error": str(exc)})
                diagnostics["error"] = str(exc)
                results = []
            diagnostics["raw_result_count"] = len(results)
            for rank, (url, title) in enumerate(results, start=1):
                consider_url(
                    url=url,
                    title=title,
                    query=query,
                    provider="sec_submissions_api",
                    rank=rank,
                    diagnostics=diagnostics,
                )
            query_diagnostics.append(diagnostics)
            candidates.sort(key=lambda item: item.score, reverse=True)
            if len(candidates) >= max_results:
                break
        if not duckduckgo_enabled:
            continue
        diagnostics = {
            "query": query,
            "provider": "duckduckgo_html",
            "raw_result_count": 0,
            "accepted": 0,
            "skipped_duplicate_or_excluded": 0,
            "skipped_non_https": 0,
            "skipped_domain": 0,
            "skipped_navigation_or_index": 0,
            "skipped_disallowed_url_hint": 0,
            "skipped_ai_rejected": 0,
        }
        try:
            results = search_duckduckgo(query, max_results=max_results_per_query)
        except Exception as exc:  # network/search failure is transparent, not fatal
            search_errors.append({"query": query, "provider": "duckduckgo_html", "error": str(exc)})
            diagnostics["error"] = str(exc)
            query_diagnostics.append(diagnostics)
            continue
        diagnostics["raw_result_count"] = len(results)
        for rank, (url, title) in enumerate(results, start=1):
            consider_url(
                url=url,
                title=title,
                query=query,
                provider="duckduckgo_html",
                rank=rank,
                diagnostics=diagnostics,
            )
        query_diagnostics.append(diagnostics)
        candidates.sort(key=lambda item: item.score, reverse=True)
        if len(candidates) >= max_results:
            break

    if len(candidates) < max_results:
        diagnostics = {
            "provider": "direct_official_fallback",
            "raw_source_count": 0,
            "accepted": 0,
            "skipped_duplicate_or_excluded": 0,
            "skipped_non_https": 0,
            "skipped_domain": 0,
            "skipped_navigation_or_index": 0,
            "skipped_disallowed_url_hint": 0,
            "skipped_ai_rejected": 0,
        }
        direct_sources = _fallback_sources_for(asset_class, role)
        diagnostics["raw_source_count"] = len(direct_sources)
        for rank, item in enumerate(direct_sources, start=1):
            consider_url(
                url=str(item.get("url") or ""),
                title=str(item.get("title") or ""),
                query="direct_official_fallback",
                provider="direct_official_fallback",
                rank=rank,
                diagnostics=diagnostics,
                rationale=str(item.get("rationale") or f"Direct official fallback source for {asset_class}/{role}"),
                source_type=str(item.get("source_type") or "") or None,
            )
            if len(candidates) >= max_results:
                break
        fallback_diagnostics.append(diagnostics)
        candidates.sort(key=lambda item: item.score, reverse=True)

    if len(candidates) < max_results:
        seed_pages = list(discovery.get("seed_pages") or DEFAULT_SEED_PAGES.get(asset_class, []))
        max_seed_links = int(discovery.get("max_seed_links_per_page", 30))
        for seed_index, seed_url in enumerate(seed_pages, start=1):
            diagnostics = {
                "seed_url": seed_url,
                "provider": "official_seed_page",
                "raw_link_count": 0,
                "accepted": 0,
                "skipped_duplicate_or_excluded": 0,
                "skipped_non_https": 0,
                "skipped_domain": 0,
                "skipped_navigation_or_index": 0,
                "skipped_disallowed_url_hint": 0,
                "skipped_low_relevance": 0,
                "skipped_ai_rejected": 0,
            }
            try:
                markup = _request_text(seed_url)
                links = _extract_links(markup)
            except Exception as exc:
                search_errors.append({"seed_url": seed_url, "provider": "official_seed_page", "error": str(exc)})
                diagnostics["error"] = str(exc)
                fallback_diagnostics.append(diagnostics)
                continue
            diagnostics["raw_link_count"] = len(links)
            for rank, (href, title) in enumerate(links[:max_seed_links], start=1):
                consider_url(
                    url=_absolute_url(seed_url, href),
                    title=title or href,
                    query=f"seed:{seed_url}",
                    provider="official_seed_page",
                    rank=rank,
                    require_relevance=True,
                    diagnostics=diagnostics,
                    rationale=f"Official seed-page link for {asset_class}/{role}; seed={seed_url}",
                )
                if len(candidates) >= max_results:
                    break
            fallback_diagnostics.append(diagnostics)
            candidates.sort(key=lambda item: item.score, reverse=True)
            if len(candidates) >= max_results:
                break

    sources = [
        {
            "asset_class": asset_class,
            "role": role,
            "title": candidate.title,
            "url": candidate.url,
            "source_type": candidate.source_type,
            "license_hint": candidate.license_hint,
            "rationale": candidate.rationale,
            "discovery": {
                "provider": candidate.provider,
                "query": candidate.query,
                "rank": candidate.rank,
                "score": candidate.score,
            },
            "ai_certification": candidate.ai_certification,
        }
        for candidate in candidates[:max_results]
    ]
    catalog = {"schema_version": "public_source_catalog_v1", "sources": sources}
    manifest = {
        "schema_version": "live_source_discovery_manifest_v1",
        "asset_class": asset_class,
        "role": role,
        "discovery_attempt": discovery_attempt,
        "queries": queries,
        "preferred_domains": preferred_domains,
        "candidate_count": len(sources),
        "catalog_path": str(output_dir / "live_discovered_public_source_catalog.json"),
        "query_diagnostics": query_diagnostics,
        "fallback_diagnostics": fallback_diagnostics,
        "errors": search_errors,
        "created_at": utc_now(),
    }
    write_json(output_dir / "live_discovered_public_source_catalog.json", catalog)
    write_json(output_dir / "live_source_discovery_manifest.json", manifest)
    return {"catalog": catalog, "manifest": manifest}