Upload 2 files

requirements.txt

@@ -0,0 +1,11 @@
+fastapi>=0.115
+uvicorn[standard]>=0.30
+pydantic>=2.8
+ddgs>=9.0
+gradio_client>=1.0
+httpx>=0.27
+beautifulsoup4>=4.12
+trafilatura>=2.0
+python-dateutil>=2.9
+scrapling>=0.3.13
+playwright>=1.55

web_search.py

@@ -0,0 +1,1301 @@
+"""
+X-RUDRA Web Search v2
+Single-file web research engine.
+
+Pipeline:
+Model A + Model B -> keyword/query planning -> DuckDuckGo discovery
+-> static HTTP -> Scrapling parsing -> DynamicFetcher/Playwright fallback
+-> optional StealthyFetcher fallback -> evidence extraction -> ranking.
+
+Important:
+- Browser automation is for rendering pages that require JavaScript.
+- It does NOT attempt to defeat CAPTCHAs, access controls, paywalls, or authentication.
+- Webpage text is untrusted data and never becomes an instruction to the agent.
+- SSRF checks are applied before navigation.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import ipaddress
+import json
+import logging
+import os
+import re
+import socket
+import sqlite3
+import time
+from collections import Counter
+from datetime import datetime, timezone
+from typing import Any, Optional
+from urllib.parse import parse_qsl, urlencode, urljoin, urlparse, urlunparse
+from urllib.robotparser import RobotFileParser
+
+import trafilatura
+from bs4 import BeautifulSoup
+from ddgs import DDGS
+from fastapi import FastAPI, HTTPException
+from pydantic import BaseModel, Field, ConfigDict
+
+# Scrapling is the primary extraction/fetching layer.
+# Playwright is used directly as a fallback when browser-level control is useful.
+try:
+    from scrapling.fetchers import Fetcher, AsyncFetcher, DynamicFetcher, StealthyFetcher
+    SCRAPLING_AVAILABLE = True
+except Exception:
+    Fetcher = AsyncFetcher = DynamicFetcher = StealthyFetcher = None
+    SCRAPLING_AVAILABLE = False
+
+try:
+    from playwright.async_api import async_playwright
+    PLAYWRIGHT_AVAILABLE = True
+except Exception:
+    async_playwright = None
+    PLAYWRIGHT_AVAILABLE = False
+
+
+# ============================================================
+# CONFIG
+# ============================================================
+
+M1_SPACE = os.getenv("M1_SPACE", "Shrijanagain/M1")
+M2_SPACE = os.getenv("M2_SPACE", "Shrijanagain/M2")
+HF_TOKEN = os.getenv("HF_TOKEN", "")
+
+MAX_RESULTS = int(os.getenv("MAX_RESULTS", "12"))
+MAX_ROUNDS = int(os.getenv("MAX_ROUNDS", "3"))
+MAX_CONCURRENT = int(os.getenv("MAX_CONCURRENT", "4"))
+HTTP_TIMEOUT = float(os.getenv("HTTP_TIMEOUT", "15"))
+BROWSER_TIMEOUT_MS = int(os.getenv("BROWSER_TIMEOUT_MS", "25000"))
+MAX_BODY = int(os.getenv("MAX_BODY", str(8 * 1024 * 1024)))
+CACHE_TTL = int(os.getenv("CACHE_TTL", "3600"))
+CACHE_DB = os.getenv("CACHE_DB", "x_rudra_web_v2.sqlite3")
+BROWSER_MODE = os.getenv("BROWSER_MODE", "auto")  # auto|never|browser
+USE_STEALTH_FETCHER = os.getenv("USE_STEALTH_FETCHER", "1") == "1"
+USER_AGENT = os.getenv(
+    "USER_AGENT",
+    "X-RUDRA-ResearchBot/2.0 (+research client)",
+)
+
+logging.basicConfig(
+    level=os.getenv("LOG_LEVEL", "INFO"),
+    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
+)
+log = logging.getLogger("x-rudra-web")
+
+
+# ============================================================
+# SCHEMAS
+# ============================================================
+
+class SearchRequest(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+    question: str = Field(min_length=1, max_length=12000)
+    max_results: int = Field(default=MAX_RESULTS, ge=1, le=30)
+    max_rounds: int = Field(default=MAX_ROUNDS, ge=1, le=5)
+    use_models: bool = True
+    freshness: str = "auto"
+
+
+class SearchPlan(BaseModel):
+    intent: str = "general"
+    topic: str = ""
+    entities: list[str] = Field(default_factory=list)
+    keywords: list[str] = Field(default_factory=list)
+    queries: list[str] = Field(default_factory=list)
+    opposing_queries: list[str] = Field(default_factory=list)
+    time_constraint: Optional[str] = None
+    source_types: list[str] = Field(default_factory=list)
+
+
+class SearchResult(BaseModel):
+    title: str
+    url: str
+    snippet: str = ""
+    query: str = ""
+    rank: int = 0
+
+
+class PageDocument(BaseModel):
+    url: str
+    final_url: str
+    domain: str
+    title: str = ""
+    description: str = ""
+    author: Optional[str] = None
+    published_at: Optional[str] = None
+    modified_at: Optional[str] = None
+    text: str = ""
+    passages: list[dict[str, Any]] = Field(default_factory=list)
+    source_score: float = 0.0
+    freshness_score: float = 0.0
+    relevance_score: float = 0.0
+    fetch_method: str = "unknown"
+    status_code: Optional[int] = None
+    prompt_injection_detected: bool = False
+
+
+class Claim(BaseModel):
+    claim: str
+    source_url: str
+    passage: str
+    support_score: float = 0.0
+
+
+class SearchReport(BaseModel):
+    status: str
+    question: str
+    plan: SearchPlan
+    queries: list[str]
+    results: list[SearchResult]
+    sources: list[PageDocument]
+    claims: list[Claim]
+    contradictions: list[dict[str, Any]]
+    failures: list[dict[str, Any]]
+    rounds: int
+    stopping_reason: str
+
+
+# ============================================================
+# BASIC UTILITIES
+# ============================================================
+
+TRACKING = {
+    "utm_source", "utm_medium", "utm_campaign", "utm_term", "utm_content",
+    "gclid", "fbclid", "mc_cid", "mc_eid", "ref", "ref_src",
+}
+
+
+def clean_text(value: str) -> str:
+    return re.sub(r"\s+", " ", value or "").strip()
+
+
+def normalize_url(url: str) -> str:
+    try:
+        p = urlparse(url.strip())
+        if p.scheme.lower() not in {"http", "https"}:
+            return ""
+        query = [
+            (k, v) for k, v in parse_qsl(p.query, keep_blank_values=True)
+            if k.lower() not in TRACKING
+        ]
+        return urlunparse(
+            p._replace(
+                scheme=p.scheme.lower(),
+                netloc=p.netloc.lower(),
+                query=urlencode(query),
+                fragment="",
+            )
+        )
+    except Exception:
+        return ""
+
+
+def domain(url: str) -> str:
+    return (urlparse(url).hostname or "").lower()
+
+
+def tokens(text: str) -> set[str]:
+    return set(re.findall(r"[a-zA-Z0-9][a-zA-Z0-9_-]{2,}", text.lower()))
+
+
+def similarity(a: str, b: str) -> float:
+    x, y = tokens(a), tokens(b)
+    return len(x & y) / max(1, len(x | y))
+
+
+def sha(value: str) -> str:
+    return hashlib.sha256(value.encode()).hexdigest()
+
+
+def private_host(host: str) -> bool:
+    if not host or host.lower() in {"localhost", "localhost.localdomain"}:
+        return True
+    try:
+        addresses = socket.getaddrinfo(host, None)
+    except OSError:
+        return True
+    for info in addresses:
+        try:
+            ip = ipaddress.ip_address(info[4][0])
+        except ValueError:
+            return True
+        if (
+            ip.is_private or ip.is_loopback or ip.is_link_local
+            or ip.is_multicast or ip.is_reserved or ip.is_unspecified
+        ):
+            return True
+    return False
+
+
+def safe_url(url: str) -> bool:
+    url = normalize_url(url)
+    if not url:
+        return False
+    p = urlparse(url)
+    return p.scheme in {"http", "https"} and not private_host(p.hostname or "")
+
+
+# ============================================================
+# CACHE
+# ============================================================
+
+class Cache:
+    def __init__(self, path: str):
+        self.path = path
+        with sqlite3.connect(self.path) as c:
+            c.execute(
+                "CREATE TABLE IF NOT EXISTS cache "
+                "(k TEXT PRIMARY KEY, v TEXT, t REAL)"
+            )
+
+    def get(self, key: str) -> Any:
+        try:
+            with sqlite3.connect(self.path) as c:
+                row = c.execute(
+                    "SELECT v,t FROM cache WHERE k=?", (key,)
+                ).fetchone()
+            if not row:
+                return None
+            if time.time() - row[1] > CACHE_TTL:
+                return None
+            return json.loads(row[0])
+        except Exception:
+            return None
+
+    def set(self, key: str, value: Any):
+        try:
+            with sqlite3.connect(self.path) as c:
+                c.execute(
+                    "INSERT OR REPLACE INTO cache(k,v,t) VALUES(?,?,?)",
+                    (key, json.dumps(value, ensure_ascii=False), time.time()),
+                )
+        except Exception as exc:
+            log.debug("cache write failed: %s", exc)
+
+
+CACHE = Cache(CACHE_DB)
+
+
+# ============================================================
+# MODEL PLANNING
+# ============================================================
+
+class SpaceModel:
+    def __init__(self, space: str):
+        self.space = space
+        self.client = None
+
+    def _client(self):
+        if self.client is None:
+            from gradio_client import Client
+            kwargs = {"hf_token": HF_TOKEN} if HF_TOKEN else {}
+            self.client = Client(self.space, **kwargs)
+        return self.client
+
+    async def generate(self, prompt: str) -> Optional[str]:
+        def run():
+            try:
+                result = self._client().predict(
+                    prompt, 1200, 0.2, 0.9, api_name="/generate"
+                )
+                return result if isinstance(result, str) else str(result)
+            except Exception as exc:
+                log.warning("Space %s unavailable: %s", self.space, exc)
+                return None
+        return await asyncio.to_thread(run)
+
+
+class Planner:
+    def __init__(self):
+        self.a = SpaceModel(M1_SPACE)
+        self.b = SpaceModel(M2_SPACE)
+
+    @staticmethod
+    def fallback(q: str) -> SearchPlan:
+        stop = {
+            "what","when","where","which","who","why","how","is","are",
+            "was","were","the","a","an","of","to","for","in","on","and",
+            "or","with","about","tell","me","please"
+        }
+        words = [x for x in re.findall(r"[a-zA-Z0-9_-]{3,}", q.lower())
+                 if x not in stop]
+        words = list(dict.fromkeys(words))
+        current = any(x in q.lower() for x in
+                      ("latest","today","current","recent","2026"))
+        return SearchPlan(
+            intent="current_factual" if current else "general",
+            topic=" ".join(words[:10]),
+            keywords=words[:20],
+            time_constraint="recent" if current else "auto",
+            source_types=["official","primary","academic","reputable"],
+        )
+
+    @staticmethod
+    def parse(text: Optional[str]) -> Optional[SearchPlan]:
+        if not text:
+            return None
+        m = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.S)
+        raw = m.group(1) if m else None
+        if raw is None:
+            a, b = text.find("{"), text.rfind("}")
+            if a >= 0 and b > a:
+                raw = text[a:b + 1]
+        if not raw:
+            return None
+        try:
+            return SearchPlan.model_validate(json.loads(raw))
+        except Exception:
+            return None
+
+    async def one(self, model: SpaceModel, question: str):
+        prompt = f"""
+X-RUDRA search planner. Return JSON only.
+Schema:
+{{
+ "intent":"general|current_factual|research|comparison|explanatory",
+ "topic":"...",
+ "entities":[],
+ "keywords":[],
+ "queries":[],
+ "opposing_queries":[],
+ "time_constraint":null,
+ "source_types":[]
+}}
+Extract search concepts and generate independent queries.
+Never answer the user question.
+QUESTION: {question}
+"""
+        return self.parse(await model.generate(prompt))
+
+    async def plan(self, question: str, use_models=True) -> SearchPlan:
+        fallback = self.fallback(question)
+        if not use_models:
+            return fallback
+        a, b = await asyncio.gather(
+            self.one(self.a, question),
+            self.one(self.b, question),
+        )
+        plans = [x for x in (a, b) if x]
+        if not plans:
+            return fallback
+
+        merged = fallback.model_copy(deep=True)
+        merged.keywords = []
+        merged.queries = []
+        merged.entities = []
+        merged.opposing_queries = []
+        merged.source_types = []
+
+        for p in plans:
+            for field in ("keywords","queries","entities",
+                          "opposing_queries","source_types"):
+                for x in getattr(p, field):
+                    if x and x not in getattr(merged, field):
+                        getattr(merged, field).append(x)
+            if p.topic:
+                merged.topic = p.topic
+            if p.time_constraint:
+                merged.time_constraint = p.time_constraint
+            if p.intent:
+                merged.intent = p.intent
+
+        return merged
+
+
+# ============================================================
+# QUERY GENERATION + DDG
+# ============================================================
+
+class QueryBuilder:
+    @staticmethod
+    def build(question: str, plan: SearchPlan) -> list[str]:
+        out = []
+
+        def add(x):
+            x = clean_text(x)
+            if x and x not in out:
+                out.append(x)
+
+        for q in plan.queries:
+            add(q)
+
+        core = plan.topic or " ".join(plan.keywords[:10])
+        add(core)
+        add(f"{core} latest")
+        add(f"{core} official")
+        add(f"{core} primary source")
+        add(f"{core} research")
+        add(f"{core} report")
+        add(f"{core} evidence")
+        add(f"{core} criticism")
+        add(f"{core} limitations")
+
+        if plan.time_constraint not in (None, "auto"):
+            add(f"{core} {plan.time_constraint}")
+
+        for q in plan.opposing_queries:
+            add(q)
+
+        return list(dict.fromkeys(out))[:30]
+
+
+class DDG:
+    async def search(self, query: str, limit: int) -> list[SearchResult]:
+        key = "ddg:" + sha(f"{query}:{limit}")
+        cached = CACHE.get(key)
+        if cached:
+            return [SearchResult.model_validate(x) for x in cached]
+
+        def run():
+            out = []
+            try:
+                with DDGS(timeout=int(HTTP_TIMEOUT)) as d:
+                    for i, item in enumerate(
+                        d.text(query, max_results=limit), 1
+                    ):
+                        url = normalize_url(item.get("href", ""))
+                        if url:
+                            out.append(SearchResult(
+                                title=clean_text(item.get("title", "")),
+                                url=url,
+                                snippet=clean_text(item.get("body", "")),
+                                query=query,
+                                rank=i,
+                            ))
+            except Exception as exc:
+                log.warning("DDG failed: %s", exc)
+            return out
+
+        result = await asyncio.to_thread(run)
+        CACHE.set(key, [x.model_dump() for x in result])
+        return result
+
+
+# ============================================================
+# ROBOTS
+# ============================================================
+
+class Robots:
+    def __init__(self):
+        self.cache = {}
+
+    async def allowed(self, url: str) -> bool:
+        host = domain(url)
+        if not host:
+            return False
+        if host in self.cache:
+            return self.cache[host]
+
+        robots_url = f"{urlparse(url).scheme}://{host}/robots.txt"
+
+        def read():
+            rp = RobotFileParser()
+            rp.set_url(robots_url)
+            try:
+                rp.read()
+                return rp.can_fetch(USER_AGENT, url)
+            except Exception:
+                return True
+
+        value = await asyncio.to_thread(read)
+        self.cache[host] = value
+        return value
+
+
+# ============================================================
+# SCRAPLING STATIC FETCH
+# ============================================================
+
+class ScraplingStatic:
+    async def fetch(self, url: str) -> Optional[dict[str, Any]]:
+        if not SCRAPLING_AVAILABLE or not safe_url(url):
+            return None
+
+        def run():
+            try:
+                page = Fetcher.get(
+                    url,
+                    timeout=int(HTTP_TIMEOUT),
+                    retries=2,
+                    follow_redirects="safe",
+                    stealthy_headers=True,
+                    impersonate="chrome",
+                )
+                html = getattr(page, "html_content", None)
+                if callable(html):
+                    html = html()
+                if html is None:
+                    html = getattr(page, "text", "") or ""
+                status = getattr(page, "status", None)
+                final = getattr(page, "url", None) or url
+                return {
+                    "html": html,
+                    "status": status,
+                    "final_url": final,
+                    "method": "scrapling-http",
+                }
+            except Exception as exc:
+                log.debug("Scrapling static failed %s: %s", url, exc)
+                return None
+
+        return await asyncio.to_thread(run)
+
+
+# ============================================================
+# PLAYWRIGHT DIRECT BROWSER
+# ============================================================
+
+class PlaywrightBrowser:
+    def __init__(self):
+        self.playwright = None
+        self.browser = None
+        self.lock = asyncio.Lock()
+
+    async def start(self):
+        if not PLAYWRIGHT_AVAILABLE:
+            return
+        async with self.lock:
+            if self.browser:
+                return
+            self.playwright = await async_playwright().start()
+            self.browser = await self.playwright.chromium.launch(
+                headless=True,
+                args=[
+                    "--disable-dev-shm-usage",
+                    "--no-first-run",
+                    "--no-default-browser-check",
+                ],
+            )
+
+    async def close(self):
+        if self.browser:
+            await self.browser.close()
+            self.browser = None
+        if self.playwright:
+            await self.playwright.stop()
+            self.playwright = None
+
+    async def fetch(self, url: str) -> Optional[dict[str, Any]]:
+        if not PLAYWRIGHT_AVAILABLE or not safe_url(url):
+            return None
+
+        await self.start()
+        if not self.browser:
+            return None
+
+        context = await self.browser.new_context(
+            user_agent=USER_AGENT,
+            java_script_enabled=True,
+            ignore_https_errors=False,
+        )
+        page = await context.new_page()
+
+        try:
+            response = await page.goto(
+                url,
+                wait_until="domcontentloaded",
+                timeout=BROWSER_TIMEOUT_MS,
+            )
+            await page.wait_for_timeout(500)
+            html = await page.content()
+            final = page.url
+
+            if not safe_url(final):
+                return None
+
+            return {
+                "html": html[:MAX_BODY],
+                "status": response.status if response else None,
+                "final_url": final,
+                "method": "playwright",
+            }
+        except Exception as exc:
+            log.debug("Playwright failed %s: %s", url, exc)
+            return None
+        finally:
+            await page.close()
+            await context.close()
+
+
+# ============================================================
+# SCRAPLING DYNAMIC / STEALTH BROWSER
+# ============================================================
+
+class ScraplingBrowser:
+    async def dynamic(self, url: str) -> Optional[dict[str, Any]]:
+        if not SCRAPLING_AVAILABLE or not safe_url(url):
+            return None
+
+        def run():
+            try:
+                page = DynamicFetcher.fetch(
+                    url,
+                    headless=True,
+                    timeout=BROWSER_TIMEOUT_MS,
+                    disable_resources=True,
+                    block_ads=True,
+                    network_idle=False,
+                    load_dom=True,
+                )
+                html = getattr(page, "html_content", "") or ""
+                return {
+                    "html": html[:MAX_BODY],
+                    "status": getattr(page, "status", None),
+                    "final_url": getattr(page, "url", None) or url,
+                    "method": "scrapling-dynamic",
+                }
+            except Exception as exc:
+                log.debug("Scrapling Dynamic failed %s: %s", url, exc)
+                return None
+
+        return await asyncio.to_thread(run)
+
+    async def stealth(self, url: str) -> Optional[dict[str, Any]]:
+        if not SCRAPLING_AVAILABLE or not USE_STEALTH_FETCHER:
+            return None
+        if not safe_url(url):
+            return None
+
+        # This mode uses Scrapling's browser implementation but does
+        # not solve CAPTCHAs or authenticate to protected resources.
+        def run():
+            try:
+                page = StealthyFetcher.fetch(
+                    url,
+                    headless=True,
+                    timeout=BROWSER_TIMEOUT_MS,
+                    disable_resources=True,
+                    block_ads=True,
+                    network_idle=False,
+                    load_dom=True,
+                )
+                html = getattr(page, "html_content", "") or ""
+                return {
+                    "html": html[:MAX_BODY],
+                    "status": getattr(page, "status", None),
+                    "final_url": getattr(page, "url", None) or url,
+                    "method": "scrapling-stealth",
+                }
+            except Exception as exc:
+                log.debug("Scrapling Stealth failed %s: %s", url, exc)
+                return None
+
+        return await asyncio.to_thread(run)
+
+
+# ============================================================
+# EXTRACTION
+# ============================================================
+
+INJECTION_PATTERNS = [
+    r"ignore\s+(all|any|the)\s+(previous|prior|system)\s+instructions",
+    r"reveal\s+(your|the)\s+(system|developer)\s+prompt",
+    r"disregard\s+your\s+instructions",
+    r"act\s+as\s+the\s+system",
+    r"show\s+your\s+hidden\s+instructions",
+]
+
+
+def detect_injection(text: str) -> bool:
+    x = text.lower()
+    return any(re.search(p, x) for p in INJECTION_PATTERNS)
+
+
+def metadata(html: str) -> dict[str, Any]:
+    soup = BeautifulSoup(html, "html.parser")
+    title = clean_text(soup.title.get_text(" ")) if soup.title else ""
+
+    def meta(name=None, prop=None):
+        tag = soup.find(
+            "meta",
+            attrs=({"name": re.compile(f"^{name}$", re.I)}
+                   if name else {"property": prop}),
+        )
+        return clean_text(tag.get("content", "")) if tag else ""
+
+    canonical = ""
+    link = soup.find("link", attrs={
+        "rel": lambda x: x and "canonical" in x
+    })
+    if link:
+        canonical = link.get("href", "")
+
+    published = meta(prop="article:published_time")
+    modified = meta(prop="article:modified_time")
+    author = meta(name="author")
+
+    for script in soup.find_all(
+        "script", attrs={"type": "application/ld+json"}
+    ):
+        try:
+            data = json.loads(script.string or "")
+            items = data if isinstance(data, list) else [data]
+            for item in items:
+                if not isinstance(item, dict):
+                    continue
+                title = title or clean_text(item.get("headline", ""))
+                published = published or item.get("datePublished", "")
+                modified = modified or item.get("dateModified", "")
+                if isinstance(item.get("author"), dict):
+                    author = author or item["author"].get("name", "")
+        except Exception:
+            continue
+
+    return {
+        "title": title,
+        "description": meta(name="description"),
+        "canonical": canonical,
+        "author": author or None,
+        "published": published or None,
+        "modified": modified or None,
+    }
+
+
+def extract_text(html: str) -> str:
+    try:
+        value = trafilatura.extract(
+            html,
+            include_tables=True,
+            include_links=False,
+            include_images=False,
+            favor_precision=True,
+        )
+        if value:
+            return clean_text(value)
+    except Exception:
+        pass
+
+    soup = BeautifulSoup(html, "html.parser")
+    for tag in soup([
+        "script","style","noscript","svg","canvas",
+        "iframe","nav","footer","form"
+    ]):
+        tag.decompose()
+    return clean_text(soup.get_text(" ", strip=True))
+
+
+def passages(question: str, text: str, limit=8):
+    sentences = re.split(r"(?<=[.!?])\s+", clean_text(text))
+    chunks, current = [], ""
+
+    for sentence in sentences:
+        if len(current) + len(sentence) < 1800:
+            current = f"{current} {sentence}".strip()
+        else:
+            if current:
+                chunks.append(current)
+            current = sentence
+    if current:
+        chunks.append(current)
+
+    scored = []
+    for chunk in chunks:
+        scored.append({
+            "text": chunk,
+            "relevance": similarity(question, chunk),
+        })
+
+    return sorted(
+        scored, key=lambda x: x["relevance"], reverse=True
+    )[:limit]
+
+
+# ============================================================
+# SOURCE SCORING
+# ============================================================
+
+def authority(host: str) -> float:
+    h = host.lower()
+    if h.endswith(".gov") or ".gov." in h:
+        return 1.0
+    if h.endswith(".edu") or ".edu." in h:
+        return 0.95
+    if h.endswith(".org"):
+        return 0.72
+
+    strong = {
+        "unesco.org","un.org","who.int","worldbank.org",
+        "oecd.org","nasa.gov","nih.gov","reuters.com",
+        "apnews.com","bbc.com"
+    }
+    return 0.95 if any(
+        h == x or h.endswith("." + x) for x in strong
+    ) else 0.45
+
+
+def freshness(value: Optional[str], mode: str) -> float:
+    if not value:
+        return 0.5
+    try:
+        from dateutil import parser
+        dt = parser.parse(value)
+        if not dt.tzinfo:
+            dt = dt.replace(tzinfo=timezone.utc)
+        age = max(
+            0,
+            (datetime.now(timezone.utc) - dt.astimezone(timezone.utc))
+            .total_seconds() / 86400,
+        )
+        if mode in {"latest","recent","current"}:
+            return max(0, 1 - age / 365)
+        return 0.7
+    except Exception:
+        return 0.5
+
+
+def source_score(host, rel, fresh):
+    return round(
+        authority(host) * 0.35
+        + rel * 0.40
+        + fresh * 0.25,
+        4,
+    )
+
+
+# ============================================================
+# MULTI-STRATEGY FETCH ORCHESTRATOR
+# ============================================================
+
+class FetchOrchestrator:
+    def __init__(self):
+        self.static = ScraplingStatic()
+        self.dynamic = ScraplingBrowser()
+        self.playwright = PlaywrightBrowser()
+        self.robots = Robots()
+        self.sem = asyncio.Semaphore(MAX_CONCURRENT)
+
+    async def fetch(self, url: str, browser_hint=False):
+        if not safe_url(url):
+            return None, "unsafe_url"
+
+        if not await self.robots.allowed(url):
+            return None, "robots_disallowed"
+
+        # 1. Fast Scrapling HTTP.
+        result = await self.static.fetch(url)
+        if result and result.get("html"):
+            return result, "scrapling-http"
+
+        if BROWSER_MODE == "never":
+            return None, "static_failed"
+
+        # 2. Scrapling dynamic browser.
+        if browser_hint or BROWSER_MODE == "browser":
+            result = await self.dynamic.dynamic(url)
+            if result and result.get("html"):
+                return result, "scrapling-dynamic"
+
+        # 3. Optional stealth browser as a renderer.
+        if browser_hint and USE_STEALTH_FETCHER:
+            result = await self.dynamic.stealth(url)
+            if result and result.get("html"):
+                return result, "scrapling-stealth"
+
+        # 4. Direct Playwright fallback.
+        if PLAYWRIGHT_AVAILABLE:
+            result = await self.playwright.fetch(url)
+            if result and result.get("html"):
+                return result, "playwright"
+
+        return None, "all_fetchers_failed"
+
+    async def close(self):
+        await self.playwright.close()
+
+
+# ============================================================
+# CLAIM / EVIDENCE
+# ============================================================
+
+class EvidenceEngine:
+    @staticmethod
+    def claims(question, sources):
+        claims = []
+        for source in sources:
+            for p in source.passages:
+                rel = float(p.get("relevance", 0))
+                if rel >= 0.18 and len(p["text"]) >= 50:
+                    claims.append(
+                        Claim(
+                            claim=p["text"][:1500],
+                            source_url=source.url,
+                            passage=p["text"][:2000],
+                            support_score=round(
+                                rel * source.source_score, 4
+                            ),
+                        )
+                    )
+        return claims
+
+    @staticmethod
+    def dedupe(claims):
+        out = []
+        for c in claims:
+            if not any(
+                similarity(c.claim, x.claim) >= 0.90
+                for x in out
+            ):
+                out.append(c)
+        return out
+
+    @staticmethod
+    def contradictions(claims):
+        result = []
+        for i, a in enumerate(claims):
+            for b in claims[i + 1:]:
+                if a.source_url == b.source_url:
+                    continue
+                sim = similarity(a.claim, b.claim)
+                if sim < 0.45:
+                    continue
+
+                na = bool(re.search(
+                    r"\b(not|no|never|false|denied|did not)\b",
+                    a.claim.lower(),
+                ))
+                nb = bool(re.search(
+                    r"\b(not|no|never|false|denied|did not)\b",
+                    b.claim.lower(),
+                ))
+
+                if na != nb:
+                    result.append({
+                        "claim_a": a.claim,
+                        "source_a": a.source_url,
+                        "claim_b": b.claim,
+                        "source_b": b.source_url,
+                        "status": "needs_review",
+                    })
+        return result[:20]
+
+
+# ============================================================
+# MAIN ENGINE
+# ============================================================
+
+class XrudraWebSearch:
+    def __init__(self):
+        self.planner = Planner()
+        self.ddg = DDG()
+        self.fetcher = FetchOrchestrator()
+
+    async def search(
+        self,
+        question: str,
+        max_results=MAX_RESULTS,
+        max_rounds=MAX_ROUNDS,
+        use_models=True,
+        freshness_mode="auto",
+    ):
+        plan = await self.planner.plan(question, use_models)
+        queries = QueryBuilder.build(question, plan)
+
+        all_results = []
+        seen = set()
+        all_sources = []
+        failures = []
+
+        for round_no in range(1, max_rounds + 1):
+            if round_no > 1:
+                followups = [
+                    f"{plan.topic} primary source",
+                    f"{plan.topic} official evidence",
+                    f"{plan.topic} independent evidence",
+                    f"{plan.topic} criticism",
+                ]
+                queries.extend(q for q in followups if q not in queries)
+
+            round_queries = queries[
+                (round_no - 1) * 8: round_no * 8
+            ]
+
+            if not round_queries:
+                break
+
+            batches = await asyncio.gather(
+                *[
+                    self.ddg.search(q, max_results)
+                    for q in round_queries
+                ],
+                return_exceptions=True,
+            )
+
+            for batch in batches:
+                if isinstance(batch, Exception):
+                    continue
+                for r in batch:
+                    r.url = normalize_url(r.url)
+                    if r.url and r.url not in seen:
+                        seen.add(r.url)
+                        all_results.append(r)
+
+            # Search results with dynamic-looking signals get browser
+            # priority; normal pages use fast HTTP first.
+            ranked = sorted(
+                all_results,
+                key=lambda r: similarity(
+                    question,
+                    r.title + " " + r.snippet
+                ),
+                reverse=True,
+            )
+
+            targets = ranked[:max_results]
+            jobs = [
+                self._fetch_one(
+                    r,
+                    question,
+                    plan,
+                    freshness_mode,
+                )
+                for r in targets
+            ]
+
+            fetched = await asyncio.gather(
+                *jobs,
+                return_exceptions=True,
+            )
+
+            for item in fetched:
+                if isinstance(item, Exception):
+                    continue
+                source, failure = item
+                if source:
+                    if not any(
+                        x.final_url == source.final_url
+                        for x in all_sources
+                    ):
+                        all_sources.append(source)
+                elif failure:
+                    failures.append(failure)
+
+            # Enough independent evidence -> stop.
+            strong = [
+                s for s in all_sources
+                if s.source_score >= 0.45
+                and s.relevance_score >= 0.18
+            ]
+            domains = {s.domain for s in strong}
+
+            if len(strong) >= 3 and len(domains) >= 2:
+                return await self._report(
+                    question, plan, queries, all_results,
+                    all_sources, failures, round_no,
+                    "sufficient_independent_evidence",
+                )
+
+        return await self._report(
+            question, plan, queries, all_results,
+            all_sources, failures, max_rounds,
+            "max_rounds_reached",
+        )
+
+    async def _fetch_one(
+        self, result, question, plan, freshness_mode
+    ):
+        browser_hint = any(
+            x in result.snippet.lower()
+            for x in (
+                "javascript", "dynamic", "interactive",
+                "app", "dashboard"
+            )
+        )
+
+        try:
+            async with self.fetcher.sem:
+                payload, method = await self.fetcher.fetch(
+                    result.url,
+                    browser_hint=browser_hint,
+                )
+
+            if not payload:
+                return None, {
+                    "url": result.url,
+                    "error": method,
+                }
+
+            final_url = normalize_url(
+                payload.get("final_url") or result.url
+            )
+            if not safe_url(final_url):
+                return None, {
+                    "url": result.url,
+                    "error": "unsafe_final_redirect",
+                }
+
+            html = payload.get("html", "")
+            if not html:
+                return None, {
+                    "url": result.url,
+                    "error": "empty_html",
+                }
+
+            meta = metadata(html)
+            text = extract_text(html)
+            if not text:
+                return None, {
+                    "url": result.url,
+                    "error": "empty_text",
+                }
+
+            p = passages(question, text)
+            rel = max(
+                [x["relevance"] for x in p],
+                default=0.0,
+            )
+            fresh = freshness(
+                meta.get("published") or meta.get("modified"),
+                plan.time_constraint
+                if plan.time_constraint not in (None, "auto")
+                else freshness_mode,
+            )
+
+            src = PageDocument(
+                url=result.url,
+                final_url=final_url,
+                domain=domain(final_url),
+                title=meta["title"] or result.title,
+                description=meta["description"],
+                author=meta["author"],
+                published_at=meta["published"],
+                modified_at=meta["modified"],
+                text=text[:100000],
+                passages=p,
+                source_score=source_score(
+                    domain(final_url), rel, fresh
+                ),
+                freshness_score=fresh,
+                relevance_score=rel,
+                fetch_method=method,
+                status_code=payload.get("status"),
+                prompt_injection_detected=detect_injection(
+                    text[:100000]
+                ),
+            )
+            return src, None
+
+        except Exception as exc:
+            return None, {
+                "url": result.url,
+                "error": f"{type(exc).__name__}:{exc}",
+            }
+
+    async def _report(
+        self, question, plan, queries, results,
+        sources, failures, rounds, reason
+    ):
+        # Source diversity: maximum two per domain.
+        selected = []
+        counts = Counter()
+
+        for s in sorted(
+            sources,
+            key=lambda x: x.source_score,
+            reverse=True,
+        ):
+            if counts[s.domain] >= 2:
+                continue
+            counts[s.domain] += 1
+            selected.append(s)
+            if len(selected) >= MAX_RESULTS:
+                break
+
+        claims = EvidenceEngine.dedupe(
+            EvidenceEngine.claims(question, selected)
+        )
+        contradictions = EvidenceEngine.contradictions(claims)
+
+        return SearchReport(
+            status="success",
+            question=question,
+            plan=plan,
+            queries=list(dict.fromkeys(queries)),
+            results=results[:MAX_RESULTS * 3],
+            sources=selected,
+            claims=claims[:100],
+            contradictions=contradictions,
+            failures=failures[:100],
+            rounds=rounds,
+            stopping_reason=reason,
+        )
+
+
+# ============================================================
+# FASTAPI
+# ============================================================
+
+app = FastAPI(
+    title="X-RUDRA Web Search v2",
+    version="2.0",
+)
+
+
+ENGINE = XrudraWebSearch()
+
+
+@app.get("/health")
+async def health():
+    return {
+        "status": "ok",
+        "scrapling": SCRAPLING_AVAILABLE,
+        "playwright": PLAYWRIGHT_AVAILABLE,
+        "m1": M1_SPACE,
+        "m2": M2_SPACE,
+        "browser_mode": BROWSER_MODE,
+    }
+
+
+@app.post("/v2/web-search", response_model=SearchReport)
+async def web_search(req: SearchRequest):
+    try:
+        return await ENGINE.search(
+            question=req.question,
+            max_results=req.max_results,
+            max_rounds=req.max_rounds,
+            use_models=req.use_models,
+            freshness_mode=req.freshness,
+        )
+    except Exception as exc:
+        log.exception("web search failed")
+        raise HTTPException(
+            status_code=500,
+            detail="web search failed",
+        ) from exc
+
+
+@app.on_event("shutdown")
+async def shutdown():
+    await ENGINE.fetcher.close()
+
+
+# ============================================================
+# CLI
+# ============================================================
+
+async def main():
+    import argparse
+
+    p = argparse.ArgumentParser()
+    p.add_argument("question")
+    p.add_argument("--results", type=int, default=10)
+    p.add_argument("--rounds", type=int, default=3)
+    p.add_argument("--no-models", action="store_true")
+    args = p.parse_args()
+
+    report = await ENGINE.search(
+        args.question,
+        max_results=args.results,
+        max_rounds=args.rounds,
+        use_models=not args.no_models,
+    )
+    print(json.dumps(
+        report.model_dump(),
+        ensure_ascii=False,
+        indent=2,
+    ))
+
+
+if __name__ == "__main__":
+    asyncio.run(main())