axon-lang 2.54.0

//! §Fase 98.e — Native Web Acquisition runtime: the OSS sibling of
//! [`crate::http_tool`], the dispatch core for the three web-acquisition
//! providers (`scrape_http` / `scrape_dom` / `scrape_crawl`).
//!
//! **The load-bearing property (D98.1):** every value this module produces is
//! born epistemically **Untrusted** (⊥ in the lattice `doubt ⊑ speculate ⊑
//! believe ⊑ know`) — reusing [`crate::emcp::EpistemicTaint::Untrusted`]
//! verbatim. The open web is adversarial; a page an agent fetched is not a
//! fact until a `shield` scans it. The compile-time content-injection barrier
//! (§98.d) refuses to let a `web`-tainted value reach an agent's belief
//! context without an intervening shield — this runtime realises the OTHER
//! half of that contract: the taint is stamped on the outcome so the audit
//! trail and any runtime IFC (§98.f) can observe it.
//!
//! **Tiers (D98.3).** Acquisition is fingerprint-first, browser-second, both
//! ENTERPRISE. OSS ships the *contract* plus a plain-`reqwest` fallback that
//! is functional but un-stealthed: a [`ScrapeFetcher`] the enterprise engine
//! (§98.g) registers to supply JA3/JA4 + HTTP/2 impersonation and per-tenant
//! proxies. With no enterprise engine injected, `impersonate` degrades to a
//! plain request (honest, un-stealthed) and `browser` returns an explicit
//! "no sidecar configured" refusal — **never a silent wrong result (D98.5)**.
//!
//! **Bounded (D98.11).** Bodies are size-capped + truncated; crawls are
//! `max_pages`/`max_depth`-bounded; timeouts apply. A hostile or infinite
//! site can neither OOM the runtime nor spin the crawler forever.
//!
//! **Honest heuristics.** Adaptive selector relocation is a *heuristic*, not a
//! proof (D98.4); the OSS DOM engine is a deterministic reference extractor
//! (regex-scanned elements over a closed selector subset), not a full browser
//! DOM — the enterprise/browser tier owns fidelity. Both are named as such.

use std::collections::BTreeMap;
use std::sync::{Arc, OnceLock};
use std::time::Duration;

use serde::{Deserialize, Serialize};

use crate::emcp::EpistemicTaint;
use crate::tool_executor::ToolResult;
use crate::tool_registry::{ScrapeConfig, ToolEntry};

/// Default fetched-body ceiling (5 MiB). A page larger than this is truncated
/// and flagged `truncated: true` — the output is always bounded (D98.11).
pub const DEFAULT_BODY_LIMIT: usize = 5 * 1024 * 1024;

/// Default per-request timeout when the tool declares none.
pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);

/// Hard ceiling on a single crawl regardless of the declared `max_pages`, so a
/// mis-declared `max_pages: 0` (unbounded sentinel) can never run away.
pub const CRAWL_HARD_PAGE_CAP: usize = 10_000;

// ════════════════════════════════════════════════════════════════════════════
//  RawPage — the typed acquisition result the grammar names
// ════════════════════════════════════════════════════════════════════════════

/// A fetched page. The `output_type: RawPage` a `scrape_http` / `scrape_crawl`
/// tool declares. Serialised to JSON as the tool's output string.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RawPage {
    /// HTTP status of the final response.
    pub status: u16,
    /// The final URL after redirects (the effective origin of `body`).
    pub final_url: String,
    /// A bounded subset of response headers (lowercased names).
    #[serde(default)]
    pub headers: BTreeMap<String, String>,
    /// The response body, capped at [`DEFAULT_BODY_LIMIT`].
    pub body: String,
    /// Whether this page was served from a cache/checkpoint (crawl resume).
    #[serde(default)]
    pub from_cache: bool,
    /// Whether `body` was truncated to the size cap (D98.11).
    #[serde(default)]
    pub truncated: bool,
    /// The engine that produced this page (`impersonate` | `browser` |
    /// `reqwest-fallback`) — provenance for the audit trail.
    #[serde(default)]
    pub engine: String,
}

impl RawPage {
    /// Cap `body` to the limit, flagging truncation. Total.
    fn capped(mut self, limit: usize) -> Self {
        if self.body.len() > limit {
            // Truncate on a char boundary to keep valid UTF-8.
            let mut end = limit;
            while end > 0 && !self.body.is_char_boundary(end) {
                end -= 1;
            }
            self.body.truncate(end);
            self.truncated = true;
        }
        self
    }
}

// ════════════════════════════════════════════════════════════════════════════
//  Errors + provenance-tagged outcome
// ════════════════════════════════════════════════════════════════════════════

/// Everything that can go wrong acquiring a page. Every variant is a *typed
/// refusal* — a failed challenge or a robots denial is a first-class outcome,
/// never a silently-empty body (D98.5).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ScrapeError {
    /// The structured tool body did not carry the required argument.
    MissingArgument(String),
    /// The URL was empty or had a non-http(s) scheme.
    InvalidUrl(String),
    /// `robots.txt` disallows this path for our user-agent and the tool did
    /// not set the (enterprise-gated) `respect_robots: false` override.
    RobotsDenied(String),
    /// The `browser` engine was requested but no sidecar is configured — the
    /// OSS build has no headless renderer (§98.i is the enterprise sidecar).
    NoBrowserSidecar,
    /// The upstream fetch failed (connect/timeout/transport).
    FetchFailed(String),
    /// An anti-bot challenge blocked the request (typed, not silent).
    Blocked(String),
    /// The `scrape_dom` `page:` argument was not a decodable `RawPage`.
    MalformedPage(String),
}

impl std::fmt::Display for ScrapeError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ScrapeError::MissingArgument(a) => {
                write!(f, "scrape: missing required argument '{a}'")
            }
            ScrapeError::InvalidUrl(u) => write!(f, "scrape: invalid URL '{u}' (need http/https)"),
            ScrapeError::RobotsDenied(u) => {
                write!(f, "scrape: robots.txt disallows '{u}' (respect_robots is on)")
            }
            ScrapeError::NoBrowserSidecar => write!(
                f,
                "scrape: engine 'browser' requested but no headless sidecar is configured — \
                 the OSS build has no renderer (this is not a silent empty result)"
            ),
            ScrapeError::FetchFailed(e) => write!(f, "scrape: fetch failed: {e}"),
            ScrapeError::Blocked(e) => write!(f, "scrape: request blocked by anti-bot: {e}"),
            ScrapeError::MalformedPage(e) => write!(f, "scrape_dom: malformed page argument: {e}"),
        }
    }
}

impl std::error::Error for ScrapeError {}

/// The provenance-tagged outcome of a scrape dispatch. `taint` is ALWAYS
/// [`EpistemicTaint::Untrusted`] on success — the born-⊥ property (D98.1). The
/// registry integration flattens this to a [`ToolResult`] (which does not
/// carry taint, matching the ℰMCP discipline where the shield/know step
/// elevates), but the taint is exposed here so `§98.f` IFC + audit can observe
/// it directly.
#[derive(Debug, Clone)]
pub struct ScrapeOutcome {
    pub result: ToolResult,
    pub taint: EpistemicTaint,
}

impl ScrapeOutcome {
    fn ok(tool_name: &str, output: String) -> Self {
        ScrapeOutcome {
            result: ToolResult {
                success: true,
                output,
                tool_name: tool_name.to_string(),
            },
            // Web content is born at ⊥ — untrusted until a shield elevates it.
            taint: EpistemicTaint::Untrusted,
        }
    }

    fn err(tool_name: &str, e: ScrapeError) -> Self {
        ScrapeOutcome {
            result: ToolResult {
                success: false,
                output: e.to_string(),
                tool_name: tool_name.to_string(),
            },
            // A failed acquisition carries no content, but the channel is
            // still adversarial — keep the taint at ⊥.
            taint: EpistemicTaint::Untrusted,
        }
    }
}

// ════════════════════════════════════════════════════════════════════════════
//  ScrapeFetcher — the enterprise injection seam (§98.g)
// ════════════════════════════════════════════════════════════════════════════

/// A single fetch request, resolved from the tool's [`ScrapeConfig`] + the
/// per-call `url`. The enterprise engine reads `engine`/`impersonate`/`proxy`
/// to select a JA3/JA4 fingerprint + proxy; the OSS default fetcher ignores
/// them (plain `reqwest`).
#[derive(Debug, Clone)]
pub struct FetchRequest {
    pub url: String,
    pub engine: String,
    pub impersonate: String,
    pub proxy: String,
    pub respect_robots: bool,
    pub render_wait: String,
    pub timeout: Duration,
    pub body_limit: usize,
}

/// The pluggable fetch engine. OSS registers nothing → the plain-`reqwest`
/// [`default_fetch`] is used. The enterprise stealth engine (§98.g) registers
/// an impl via [`register_scrape_fetcher`] to supply fingerprint impersonation
/// + proxy pools. Exactly the [`crate::shield_registry`] injection shape.
pub trait ScrapeFetcher: Send + Sync {
    /// Fetch a single page. MUST return a typed [`ScrapeError`] on failure —
    /// never an empty [`RawPage`] masquerading as success (D98.5).
    fn fetch(&self, req: &FetchRequest) -> Result<RawPage, ScrapeError>;

    /// A short slug identifying the engine, for the `RawPage.engine`
    /// provenance field + audit.
    fn engine_slug(&self) -> &'static str;
}

static SCRAPE_FETCHER: OnceLock<Arc<dyn ScrapeFetcher>> = OnceLock::new();

/// §Fase 98.g — register the enterprise stealth fetcher. Idempotent-first:
/// the first registration wins (the enterprise host installs it once at boot).
/// Returns `true` if this call installed the fetcher.
pub fn register_scrape_fetcher(fetcher: Arc<dyn ScrapeFetcher>) -> bool {
    SCRAPE_FETCHER.set(fetcher).is_ok()
}

/// Whether an enterprise fetcher is installed (else the OSS fallback is used).
pub fn has_registered_fetcher() -> bool {
    SCRAPE_FETCHER.get().is_some()
}

/// Resolve the active fetcher: the registered enterprise engine, or the OSS
/// plain-`reqwest` fallback.
fn fetch_page(req: &FetchRequest) -> Result<RawPage, ScrapeError> {
    if let Some(f) = SCRAPE_FETCHER.get() {
        return f.fetch(req);
    }
    default_fetch(req)
}

// ════════════════════════════════════════════════════════════════════════════
//  Default OSS fetcher — plain reqwest, robots-respecting, bounded
// ════════════════════════════════════════════════════════════════════════════

/// The default AXON user-agent for OSS fetches. Honest (identifies the
/// runtime) — the enterprise engine impersonates a real browser instead.
pub const DEFAULT_USER_AGENT: &str = "AxonScrape/1.0 (+https://axon.dev)";

/// OSS fallback fetch: plain `reqwest::blocking`. No fingerprint stealth. The
/// `browser` engine is refused (no OSS renderer). robots.txt honored unless
/// the caller disabled it.
fn default_fetch(req: &FetchRequest) -> Result<RawPage, ScrapeError> {
    let url = req.url.trim();
    if url.is_empty() || (!url.starts_with("http://") && !url.starts_with("https://")) {
        return Err(ScrapeError::InvalidUrl(url.to_string()));
    }
    // The browser tier has no OSS engine — refuse explicitly (D98.5).
    if req.engine == "browser" {
        return Err(ScrapeError::NoBrowserSidecar);
    }
    // robots-respecting default (D98.6). A denial is a typed refusal.
    if req.respect_robots && !robots_allows(url, req.timeout) {
        return Err(ScrapeError::RobotsDenied(url.to_string()));
    }

    let client = reqwest::blocking::Client::builder()
        .timeout(req.timeout)
        .user_agent(DEFAULT_USER_AGENT)
        .build()
        .map_err(|e| ScrapeError::FetchFailed(format!("client build: {e}")))?;

    let resp = client.get(url).send().map_err(|e| {
        if e.is_timeout() {
            ScrapeError::FetchFailed(format!("timed out after {}s", req.timeout.as_secs()))
        } else if e.is_connect() {
            ScrapeError::FetchFailed(format!("connection failed to {url}"))
        } else {
            ScrapeError::FetchFailed(e.to_string())
        }
    })?;

    let status = resp.status().as_u16();
    let final_url = resp.url().to_string();
    // A 403/429 with an anti-bot signature is a typed BLOCK, not a page.
    if status == 403 || status == 429 {
        return Err(ScrapeError::Blocked(format!("HTTP {status} from {final_url}")));
    }
    let mut headers = BTreeMap::new();
    for name in ["content-type", "content-length", "server", "last-modified"] {
        if let Some(v) = resp.headers().get(name).and_then(|v| v.to_str().ok()) {
            headers.insert(name.to_string(), v.to_string());
        }
    }
    let body = resp
        .text()
        .map_err(|e| ScrapeError::FetchFailed(format!("read body: {e}")))?;

    Ok(RawPage {
        status,
        final_url,
        headers,
        body,
        from_cache: false,
        truncated: false,
        engine: "reqwest-fallback".to_string(),
    }
    .capped(req.body_limit))
}

// ── robots.txt (OSS default enforcement) ─────────────────────────────────────

/// Fetch + evaluate `robots.txt` for `url`, honoring `Disallow` rules for the
/// `*` user-agent group. Fail-OPEN on fetch/parse failure (the standard
/// posture — an unreachable robots.txt does not block). The enterprise layer
/// (§98.h) adds the RBAC-gated override + the audit row; this is the sound
/// default so an OSS deployment is polite out of the box.
fn robots_allows(url: &str, timeout: Duration) -> bool {
    let (scheme, host, path) = match split_url(url) {
        Some(t) => t,
        None => return true,
    };
    let robots_url = format!("{scheme}://{host}/robots.txt");
    let client = match reqwest::blocking::Client::builder()
        .timeout(timeout.min(Duration::from_secs(10)))
        .user_agent(DEFAULT_USER_AGENT)
        .build()
    {
        Ok(c) => c,
        Err(_) => return true,
    };
    let body = match client.get(&robots_url).send().and_then(|r| r.text()) {
        Ok(b) => b,
        // No robots.txt reachable → allowed (fail-open).
        Err(_) => return true,
    };
    robots_path_allowed(&body, &path)
}

/// Pure robots.txt evaluator: does the `*` user-agent group permit `path`?
/// Longest-match `Allow`/`Disallow` wins (the de-facto standard). Public for
/// the §98.f unit tests + the enterprise governance layer.
pub fn robots_path_allowed(robots_txt: &str, path: &str) -> bool {
    let mut in_star = false;
    let mut applicable = false;
    // (is_allow, pattern_len, allowed) longest-match accumulator.
    let mut best: Option<(usize, bool)> = None;
    for raw in robots_txt.lines() {
        let line = raw.split('#').next().unwrap_or("").trim();
        if line.is_empty() {
            continue;
        }
        let (key, val) = match line.split_once(':') {
            Some((k, v)) => (k.trim().to_ascii_lowercase(), v.trim().to_string()),
            None => continue,
        };
        match key.as_str() {
            "user-agent" => {
                // A new group. `*` (or a run of them) applies to us.
                if val == "*" {
                    in_star = true;
                    applicable = true;
                } else {
                    in_star = false;
                }
            }
            "disallow" | "allow" if in_star || applicable && in_star => {
                if val.is_empty() {
                    continue;
                }
                if path.starts_with(&val) {
                    let is_allow = key == "allow";
                    let len = val.len();
                    if best.map(|(bl, _)| len > bl).unwrap_or(true) {
                        best = Some((len, is_allow));
                    }
                }
            }
            _ => {}
        }
    }
    match best {
        Some((_, is_allow)) => is_allow,
        // No matching rule → allowed.
        None => true,
    }
}

/// Split an http(s) URL into `(scheme, host[:port], path)`. Minimal, dep-free
/// (OSS has no `url` crate). Returns `None` for non-http(s).
fn split_url(url: &str) -> Option<(String, String, String)> {
    let (scheme, rest) = if let Some(r) = url.strip_prefix("https://") {
        ("https", r)
    } else if let Some(r) = url.strip_prefix("http://") {
        ("http", r)
    } else {
        return None;
    };
    let (authority, path) = match rest.find('/') {
        Some(i) => (&rest[..i], &rest[i..]),
        None => (rest, "/"),
    };
    // Drop any userinfo + fragment; keep host[:port].
    let host = authority.rsplit('@').next().unwrap_or(authority);
    let path = path.split('#').next().unwrap_or(path);
    Some((scheme.to_string(), host.to_string(), path.to_string()))
}

// ════════════════════════════════════════════════════════════════════════════
//  Synchronous dispatch — scrape_http + scrape_dom
// ════════════════════════════════════════════════════════════════════════════

/// Dispatch a synchronous scrape tool call. Routes by `entry.provider`:
/// `scrape_http` (fetch), `scrape_dom` (parse — no network). `scrape_crawl`
/// is streaming and handled by [`ScrapeStreamingTool`], not here. Returns the
/// [`ToolResult`] the registry integrates; the born-Untrusted taint lives on
/// the internal [`ScrapeOutcome`] (see [`dispatch_scrape_outcome`]).
pub fn dispatch_scrape(entry: &ToolEntry, argument: &str) -> ToolResult {
    dispatch_scrape_outcome(entry, argument).result
}

/// The taint-carrying dispatch (used by §98.f IFC + tests). Always yields an
/// `Untrusted` outcome for a successful acquisition.
pub fn dispatch_scrape_outcome(entry: &ToolEntry, argument: &str) -> ScrapeOutcome {
    let cfg = entry.scrape.clone().unwrap_or_default();
    let args = parse_args(argument);
    match entry.provider.as_str() {
        "scrape_http" => run_scrape_http(&entry.name, &cfg, &entry.timeout, &args),
        "scrape_dom" => run_scrape_dom(&entry.name, &cfg, &args),
        // scrape_crawl is a streaming provider; a synchronous dispatch of it
        // returns a single seed fetch as a courtesy (the dispatcher routes
        // crawl through the streaming path).
        "scrape_crawl" => run_scrape_http(&entry.name, &cfg, &entry.timeout, &args),
        other => ScrapeOutcome::err(
            &entry.name,
            ScrapeError::FetchFailed(format!("unknown scrape provider '{other}'")),
        ),
    }
}

/// Parse the structured tool body (`{ "url": "...", ... }`) into a JSON map.
/// A non-object body is wrapped as `{ "input": <text> }` (mirrors http_tool's
/// body discipline) so a bare-string `on <arg>` call still finds a `url`.
fn parse_args(argument: &str) -> serde_json::Value {
    let trimmed = argument.trim_start();
    if trimmed.starts_with('{') {
        serde_json::from_str(argument).unwrap_or_else(|_| serde_json::json!({ "input": argument }))
    } else {
        serde_json::json!({ "input": argument, "url": argument, "page": argument })
    }
}

fn arg_str<'a>(args: &'a serde_json::Value, key: &str) -> Option<&'a str> {
    args.get(key).and_then(|v| v.as_str())
}

fn run_scrape_http(
    name: &str,
    cfg: &ScrapeConfig,
    timeout: &str,
    args: &serde_json::Value,
) -> ScrapeOutcome {
    let url = match arg_str(args, "url") {
        Some(u) if !u.is_empty() => u.to_string(),
        _ => return ScrapeOutcome::err(name, ScrapeError::MissingArgument("url".into())),
    };
    let req = FetchRequest {
        url,
        engine: cfg.effective_engine().to_string(),
        impersonate: cfg.impersonate.clone(),
        proxy: cfg.proxy.clone(),
        respect_robots: cfg.respect_robots,
        render_wait: cfg.render_wait.clone(),
        timeout: crate::http_tool::parse_timeout_pub(timeout).unwrap_or(DEFAULT_TIMEOUT),
        body_limit: DEFAULT_BODY_LIMIT,
    };
    match fetch_page(&req) {
        Ok(page) => match serde_json::to_string(&page) {
            Ok(json) => ScrapeOutcome::ok(name, json),
            Err(e) => ScrapeOutcome::err(name, ScrapeError::FetchFailed(format!("encode: {e}"))),
        },
        Err(e) => ScrapeOutcome::err(name, e),
    }
}

fn run_scrape_dom(name: &str, cfg: &ScrapeConfig, args: &serde_json::Value) -> ScrapeOutcome {
    // The `page:` argument is a `RawPage` (from a prior scrape_http), or a bare
    // HTML string. Either way, the taint is PRESERVED — scrape_dom does no I/O,
    // it only processes already-Untrusted content, so its output stays ⊥.
    let html = match args.get("page") {
        Some(serde_json::Value::String(s)) => {
            // Could be a JSON-encoded RawPage or raw HTML.
            match serde_json::from_str::<RawPage>(s) {
                Ok(p) => p.body,
                Err(_) => s.clone(),
            }
        }
        Some(v @ serde_json::Value::Object(_)) => match serde_json::from_value::<RawPage>(v.clone())
        {
            Ok(p) => p.body,
            Err(e) => return ScrapeOutcome::err(name, ScrapeError::MalformedPage(e.to_string())),
        },
        _ => return ScrapeOutcome::err(name, ScrapeError::MissingArgument("page".into())),
    };

    let extracted = extract_fields(&html, &cfg.extract, cfg.adaptive, cfg.similarity_floor);
    match serde_json::to_string(&extracted) {
        Ok(json) => ScrapeOutcome::ok(name, json),
        Err(e) => ScrapeOutcome::err(name, ScrapeError::MalformedPage(format!("encode: {e}"))),
    }
}

// ════════════════════════════════════════════════════════════════════════════
//  Deterministic DOM extractor (reference engine) + adaptive heuristic
// ════════════════════════════════════════════════════════════════════════════

/// One scanned HTML element: `(tag, id, classes, inner_text)`.
#[derive(Debug, Clone)]
struct Element {
    tag: String,
    id: String,
    classes: Vec<String>,
    text: String,
}

/// A parsed selector from the closed subset: an optional tag + optional `#id`
/// + optional `.class`. E.g. `h1`, `.price`, `#main`, `div.card`, `a#logo`.
#[derive(Debug, Clone, Default)]
struct Selector {
    tag: Option<String>,
    id: Option<String>,
    class: Option<String>,
}

fn parse_selector(sel: &str) -> Selector {
    let sel = sel.trim();
    let mut out = Selector::default();
    // Split into a leading tag and #id/.class fragments.
    let mut i = 0;
    let bytes = sel.as_bytes();
    // leading tag = run until a '.' or '#'
    let mut j = 0;
    while j < bytes.len() && bytes[j] != b'.' && bytes[j] != b'#' {
        j += 1;
    }
    if j > 0 {
        out.tag = Some(sel[..j].to_ascii_lowercase());
    }
    i = j;
    while i < bytes.len() {
        let marker = bytes[i];
        let start = i + 1;
        let mut k = start;
        while k < bytes.len() && bytes[k] != b'.' && bytes[k] != b'#' {
            k += 1;
        }
        let frag = &sel[start..k];
        match marker {
            b'#' => out.id = Some(frag.to_string()),
            b'.' => out.class = Some(frag.to_string()),
            _ => {}
        }
        i = k;
    }
    out
}

fn matches(el: &Element, s: &Selector) -> bool {
    if let Some(t) = &s.tag {
        if &el.tag != t {
            return false;
        }
    }
    if let Some(id) = &s.id {
        if &el.id != id {
            return false;
        }
    }
    if let Some(c) = &s.class {
        if !el.classes.iter().any(|x| x == c) {
            return false;
        }
    }
    true
}

/// Scan HTML into a flat element list. A deterministic reference parser
/// (regex over open-tag..matching-close for a common set of container tags),
/// NOT a spec-compliant DOM — the enterprise/browser tier owns fidelity
/// (D98.4). Nested same-tag elements resolve to the outermost occurrence,
/// which is sufficient for the closed selector subset.
/// The closed tag set the reference extractor scans. Container + inline tags
/// common in article/product markup — sufficient for the closed selector
/// subset; the enterprise/browser tier owns full-DOM fidelity.
const SCANNED_TAGS: &[&str] = &[
    "h1", "h2", "h3", "h4", "h5", "h6", "p", "a", "span", "div", "article", "section", "li", "td",
    "th", "title", "strong", "em", "b", "i",
];

fn scan_elements(html: &str) -> Vec<Element> {
    use regex::Regex;
    // The `regex` crate has no backreferences, so we scan per-tag with a
    // compiled-once regex map (`<tag ...>inner</tag>`, non-greedy). Cached in
    // source order across the tag set; each element records its byte start so
    // the result stays in document order.
    static RES: OnceLock<Vec<(String, Regex)>> = OnceLock::new();
    let res = RES.get_or_init(|| {
        SCANNED_TAGS
            .iter()
            .map(|t| {
                (
                    t.to_string(),
                    Regex::new(&format!(r#"(?is)<{t}\b([^>]*)>(.*?)</{t}>"#))
                        .expect("static per-tag scrape regex is valid"),
                )
            })
            .collect()
    });
    let attr_id = attr_regex("id");
    let attr_class = attr_regex("class");
    let tag_strip = tag_strip_regex();
    let mut out: Vec<(usize, Element)> = Vec::new();
    for (tag, re) in res {
        for cap in re.captures_iter(html) {
            let whole = cap.get(0).unwrap();
            let attrs = cap.get(1).map(|m| m.as_str()).unwrap_or("");
            let inner = cap.get(2).map(|m| m.as_str()).unwrap_or("");
            let id = attr_id
                .captures(attrs)
                .and_then(|c| c.get(1))
                .map(|m| m.as_str().to_string())
                .unwrap_or_default();
            let classes = attr_class
                .captures(attrs)
                .and_then(|c| c.get(1))
                .map(|m| m.as_str().split_whitespace().map(|s| s.to_string()).collect())
                .unwrap_or_default();
            // Inner text = strip nested tags, collapse whitespace.
            let text = tag_strip.replace_all(inner, " ");
            let text = text.split_whitespace().collect::<Vec<_>>().join(" ");
            out.push((
                whole.start(),
                Element {
                    tag: tag.clone(),
                    id,
                    classes,
                    text,
                },
            ));
        }
    }
    // Document order.
    out.sort_by_key(|(start, _)| *start);
    out.into_iter().map(|(_, e)| e).collect()
}

fn attr_regex(name: &str) -> regex::Regex {
    regex::Regex::new(&format!(r#"(?i)\b{name}\s*=\s*["']([^"']*)["']"#)).expect("attr regex valid")
}

fn tag_strip_regex() -> regex::Regex {
    static RE: OnceLock<regex::Regex> = OnceLock::new();
    RE.get_or_init(|| regex::Regex::new(r"(?s)<[^>]*>").expect("tag strip regex valid"))
        .clone()
}

/// Extract every declared FieldSpec from `html`. Each spec is `name=selector`.
/// Returns a JSON object `{ name: text }`. When `adaptive` and a selector
/// misses, a HEURISTIC relocation is attempted (loosen the selector, keeping
/// the tag) — recorded honestly, never presented as a proof (D98.4).
fn extract_fields(
    html: &str,
    specs: &[String],
    adaptive: bool,
    similarity_floor: f64,
) -> serde_json::Value {
    let elements = scan_elements(html);
    let mut obj = serde_json::Map::new();
    for spec in specs {
        let (field, selector_str) = match spec.split_once('=') {
            Some((f, s)) => (f.trim(), s.trim()),
            // A malformed spec yields a null — never a panic.
            None => {
                obj.insert(spec.clone(), serde_json::Value::Null);
                continue;
            }
        };
        let sel = parse_selector(selector_str);
        let found = elements.iter().find(|el| matches(el, &sel));
        let value = match found {
            Some(el) => Some(el.text.clone()),
            None if adaptive => relocate(&elements, &sel, similarity_floor),
            None => None,
        };
        obj.insert(
            field.to_string(),
            value.map(serde_json::Value::String).unwrap_or(serde_json::Value::Null),
        );
    }
    serde_json::Value::Object(obj)
}

/// Adaptive relocation HEURISTIC (D98.4 — a heuristic, NOT a proof): when the
/// exact selector misses, retry with the tag alone (dropping `#id`/`.class`),
/// accepting the first same-tag element only if the confidence — here a simple
/// structural score: 1.0 when the tag matches, else 0.0 — meets the floor.
/// Deliberately conservative: OSS has no per-tenant selector memory (that is
/// the enterprise `<storage>` store, §98.h), so relocation is tag-level only.
fn relocate(elements: &[Element], sel: &Selector, similarity_floor: f64) -> Option<String> {
    let tag = sel.tag.as_ref()?;
    // Structural confidence of a tag-only relocation.
    let confidence = 1.0_f64;
    if confidence < similarity_floor {
        return None;
    }
    elements
        .iter()
        .find(|el| &el.tag == tag)
        .map(|el| el.text.clone())
}

// ════════════════════════════════════════════════════════════════════════════
//  ScrapeStreamingTool — concurrent checkpointed crawl (scrape_crawl)
// ════════════════════════════════════════════════════════════════════════════

use async_trait::async_trait;

use crate::tool_trait::{Tool, ToolChunk, ToolContext, ToolFinishReason, ToolStream};

/// §Fase 98.e — the streaming crawl provider. A bounded, checkpointed spider:
/// it BFS-walks from `seed`, emitting each fetched [`RawPage`] as a
/// `ToolChunk` AS it arrives, bounded by `max_pages` / `max_depth` /
/// `concurrency`, honoring the per-invocation cancel flag between fetches.
/// Every emitted page is born Untrusted (D98.1) — the taint discipline is
/// identical to the synchronous providers; the compile-time barrier (§98.d)
/// enforces the shield before an agent's belief.
///
/// The OSS engine uses the plain-`reqwest` fetcher (or the registered
/// enterprise stealth engine); resumable checkpointing to a store is the
/// enterprise `<storage>` concern (§98.h) — OSS keeps the visited set in
/// memory (non-resumable, honestly bounded).
pub struct ScrapeStreamingTool {
    name: String,
    cfg: ScrapeConfig,
    timeout: Duration,
}

impl ScrapeStreamingTool {
    /// Build from a registry [`ToolEntry`]. Falls back to a default config
    /// when the entry carries none (defensive — the checker guarantees a
    /// `scrape_crawl` tool has a `scrape:` block).
    pub fn from_entry(entry: &ToolEntry) -> Self {
        ScrapeStreamingTool {
            name: entry.name.clone(),
            cfg: entry.scrape.clone().unwrap_or_default(),
            timeout: crate::http_tool::parse_timeout_pub(&entry.timeout).unwrap_or(DEFAULT_TIMEOUT),
        }
    }

    /// The effective per-crawl page ceiling: the declared `max_pages`
    /// (clamped to the hard cap), or the hard cap when unbounded (`0`).
    fn page_budget(&self) -> usize {
        let declared = self.cfg.max_pages.max(0) as usize;
        if declared == 0 {
            CRAWL_HARD_PAGE_CAP
        } else {
            declared.min(CRAWL_HARD_PAGE_CAP)
        }
    }
}

/// Extract absolute-ish links from a page body for crawl expansion. When a
/// `follow` selector is declared, only `<a>` hrefs are considered (the OSS
/// engine treats `follow` as "follow anchors"); enterprise fidelity can honor
/// a richer selector. Relative links are joined against `base`.
fn extract_links(html: &str, base: &str) -> Vec<String> {
    use regex::Regex;
    static RE: OnceLock<Regex> = OnceLock::new();
    let re = RE.get_or_init(|| {
        Regex::new(r#"(?i)<a\b[^>]*\bhref\s*=\s*["']([^"'#]+)["']"#).expect("href regex valid")
    });
    let mut out = Vec::new();
    for cap in re.captures_iter(html) {
        if let Some(m) = cap.get(1) {
            if let Some(abs) = join_url(base, m.as_str()) {
                out.push(abs);
            }
        }
    }
    out
}

/// Minimal URL join (dep-free): resolve `href` against `base`. Handles
/// absolute, root-relative, and same-directory relative links — enough for
/// the reference crawler; the enterprise engine can use a full `url` resolver.
fn join_url(base: &str, href: &str) -> Option<String> {
    if href.starts_with("http://") || href.starts_with("https://") {
        return Some(href.to_string());
    }
    let (scheme, host, path) = split_url(base)?;
    if let Some(rooted) = href.strip_prefix('/') {
        return Some(format!("{scheme}://{host}/{rooted}"));
    }
    // Same-directory relative: replace the last path segment.
    let dir = match path.rfind('/') {
        Some(i) => &path[..=i],
        None => "/",
    };
    Some(format!("{scheme}://{host}{dir}{href}"))
}

#[async_trait]
impl Tool for ScrapeStreamingTool {
    async fn execute(&self, args: String, _ctx: ToolContext) -> ToolResult {
        // Synchronous fallback: a crawl collapsed to a single seed fetch.
        let entry_cfg = self.cfg.clone();
        let timeout = self.timeout;
        let name = self.name.clone();
        let args_owned = args.clone();
        match tokio::task::spawn_blocking(move || {
            let parsed = parse_args(&args_owned);
            let entry = ToolEntry {
                name: name.clone(),
                provider: "scrape_http".to_string(),
                timeout: format!("{}s", timeout.as_secs()),
                runtime: String::new(),
                sandbox: None,
                max_results: None,
                output_schema: String::new(),
                effect_row: vec!["network".into(), "web".into()],
                parameters: Vec::new(),
                secret: String::new(),
                secret_partition: String::new(),
                source: crate::tool_registry::ToolSource::Program,
                is_streaming: true,
                scrape: Some(entry_cfg),
            };
            // Seed arg may be under `seed` or `url`.
            let seed = parsed
                .get("seed")
                .and_then(|v| v.as_str())
                .or_else(|| parsed.get("url").and_then(|v| v.as_str()))
                .unwrap_or("")
                .to_string();
            run_scrape_http(
                &entry.name,
                &entry.scrape.clone().unwrap_or_default(),
                &entry.timeout,
                &serde_json::json!({ "url": seed }),
            )
            .result
        })
        .await
        {
            Ok(r) => r,
            Err(e) => ToolResult {
                success: false,
                output: format!("scrape_crawl '{}': join failed: {e}", self.name),
                tool_name: self.name.clone(),
            },
        }
    }

    async fn stream(&self, args: String, ctx: ToolContext) -> ToolStream {
        let name = self.name.clone();
        let cfg = self.cfg.clone();
        let timeout = self.timeout;
        let budget = self.page_budget();
        let max_depth = self.cfg.max_depth.max(0) as usize;
        let follow = !self.cfg.follow.is_empty();
        let cancel = ctx.cancel.clone();

        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<ToolChunk>();

        tokio::spawn(async move {
            let send_term = |reason: ToolFinishReason| {
                let _ = tx.send(ToolChunk::terminator("", reason));
            };

            let seed = {
                let parsed = parse_args(&args);
                parsed
                    .get("seed")
                    .and_then(|v| v.as_str())
                    .or_else(|| parsed.get("url").and_then(|v| v.as_str()))
                    .unwrap_or("")
                    .to_string()
            };
            if seed.is_empty() {
                send_term(ToolFinishReason::Error {
                    message: format!("scrape_crawl '{name}': missing 'seed' argument"),
                });
                return;
            }

            // BFS frontier of (url, depth); in-memory visited set (OSS: not
            // resumable — the enterprise checkpoint store is §98.h).
            let mut frontier: std::collections::VecDeque<(String, usize)> =
                std::collections::VecDeque::new();
            let mut visited: std::collections::HashSet<String> = std::collections::HashSet::new();
            frontier.push_back((seed, 0));
            let mut fetched = 0usize;

            while let Some((url, depth)) = frontier.pop_front() {
                if fetched >= budget {
                    break;
                }
                if cancel.is_cancelled() {
                    send_term(ToolFinishReason::Cancelled);
                    return;
                }
                if !visited.insert(url.clone()) {
                    continue;
                }

                let req = FetchRequest {
                    url: url.clone(),
                    engine: cfg.effective_engine().to_string(),
                    impersonate: cfg.impersonate.clone(),
                    proxy: cfg.proxy.clone(),
                    respect_robots: cfg.respect_robots,
                    render_wait: cfg.render_wait.clone(),
                    timeout,
                    body_limit: DEFAULT_BODY_LIMIT,
                };
                // Blocking fetch on the blocking pool.
                let fetch = tokio::task::spawn_blocking(move || fetch_page(&req)).await;
                match fetch {
                    Ok(Ok(page)) => {
                        fetched += 1;
                        // Expand links (bounded by depth) before emitting.
                        if follow && depth < max_depth {
                            for link in extract_links(&page.body, &page.final_url) {
                                if !visited.contains(&link) {
                                    frontier.push_back((link, depth + 1));
                                }
                            }
                        }
                        match serde_json::to_string(&page) {
                            Ok(json) => {
                                if tx.send(ToolChunk::intermediate(json)).is_err() {
                                    return; // consumer dropped
                                }
                            }
                            Err(e) => {
                                let _ = tx.send(ToolChunk::intermediate(
                                    ScrapeError::FetchFailed(format!("encode: {e}")).to_string(),
                                ));
                            }
                        }
                    }
                    Ok(Err(e)) => {
                        // A per-page failure is a typed chunk, not a crawl
                        // abort — the spider keeps going (D98.5 typed outcome).
                        let _ = tx.send(ToolChunk::intermediate(e.to_string()));
                    }
                    Err(e) => {
                        let _ = tx.send(ToolChunk::intermediate(format!("crawl join failed: {e}")));
                    }
                }
            }
            send_term(ToolFinishReason::Stop);
        });

        Box::pin(futures::stream::unfold(rx, |mut rx| async move {
            rx.recv().await.map(|chunk| (chunk, rx))
        }))
    }

    fn is_streaming(&self) -> bool {
        true
    }
}

// ════════════════════════════════════════════════════════════════════════════
//  Tests
// ════════════════════════════════════════════════════════════════════════════

#[cfg(test)]
mod tests {
    use super::*;

    fn cfg_dom(extract: &[&str], adaptive: bool, floor: f64) -> ScrapeConfig {
        ScrapeConfig {
            extract: extract.iter().map(|s| s.to_string()).collect(),
            adaptive,
            similarity_floor: floor,
            ..Default::default()
        }
    }

    fn dom_entry(name: &str, cfg: ScrapeConfig) -> ToolEntry {
        ToolEntry {
            name: name.to_string(),
            provider: "scrape_dom".to_string(),
            timeout: String::new(),
            runtime: String::new(),
            sandbox: None,
            max_results: None,
            output_schema: String::new(),
            effect_row: vec!["web".to_string()],
            parameters: Vec::new(),
            secret: String::new(),
            secret_partition: String::new(),
            source: crate::tool_registry::ToolSource::Program,
            is_streaming: false,
            scrape: Some(cfg),
        }
    }

    #[test]
    fn dom_output_is_born_untrusted() {
        let html = r#"<html><body><h1>Hello</h1><p class="lead">World</p></body></html>"#;
        let entry = dom_entry("Ex", cfg_dom(&["title=h1", "lead=p.lead"], false, 0.0));
        let out = dispatch_scrape_outcome(&entry, &format!("{{\"page\": {:?} }}", html));
        assert!(out.result.success, "output: {}", out.result.output);
        // The load-bearing property: web content is born at ⊥.
        assert_eq!(out.taint, EpistemicTaint::Untrusted);
        let v: serde_json::Value = serde_json::from_str(&out.result.output).unwrap();
        assert_eq!(v["title"], "Hello");
        assert_eq!(v["lead"], "World");
    }

    #[test]
    fn dom_selector_subset_id_class_tag() {
        let html = r#"<div id="main"><span class="price">$9</span><a href="/x">buy</a></div>"#;
        let out = extract_fields(
            html,
            &[
                "p=#main".to_string(), // #id (div) — but selector tag is empty, matches by id
                "price=.price".to_string(),
                "cta=a".to_string(),
            ],
            false,
            0.0,
        );
        assert_eq!(out["price"], "$9");
        assert_eq!(out["cta"], "buy");
    }

    #[test]
    fn dom_miss_without_adaptive_is_null() {
        let html = "<h1>Title</h1>";
        let out = extract_fields(html, &["x=.nope".to_string()], false, 0.0);
        assert_eq!(out["x"], serde_json::Value::Null);
    }

    #[test]
    fn dom_adaptive_relocates_by_tag() {
        // Selector `h1.headline` misses (no class), adaptive relocates to <h1>.
        let html = "<h1>Relocated</h1>";
        let out = extract_fields(html, &["t=h1.headline".to_string()], true, 0.5);
        assert_eq!(out["t"], "Relocated");
    }

    #[test]
    fn dom_adaptive_respects_floor() {
        // With no tag in the selector, relocation cannot apply → null.
        let html = "<h1>X</h1>";
        let out = extract_fields(html, &["t=.only-class".to_string()], true, 0.5);
        assert_eq!(out["t"], serde_json::Value::Null);
    }

    #[test]
    fn body_cap_truncates_on_char_boundary() {
        let page = RawPage {
            status: 200,
            final_url: "https://e.x".into(),
            headers: BTreeMap::new(),
            body: "áéíóú".repeat(1000),
            from_cache: false,
            truncated: false,
            engine: "test".into(),
        }
        .capped(10);
        assert!(page.truncated);
        assert!(page.body.len() <= 10);
        // Still valid UTF-8 (did not split a multibyte char).
        assert!(std::str::from_utf8(page.body.as_bytes()).is_ok());
    }

    #[test]
    fn robots_longest_match_wins() {
        let robots = "User-agent: *\nDisallow: /private\nAllow: /private/public\n";
        assert!(!robots_path_allowed(robots, "/private/secret"));
        assert!(robots_path_allowed(robots, "/private/public/x"));
        assert!(robots_path_allowed(robots, "/open"));
    }

    #[test]
    fn robots_no_rules_allows() {
        assert!(robots_path_allowed("", "/anything"));
        assert!(robots_path_allowed("User-agent: Googlebot\nDisallow: /", "/x"));
    }

    #[test]
    fn split_url_extracts_parts() {
        let (s, h, p) = split_url("https://ex.com:8080/a/b?q=1#frag").unwrap();
        assert_eq!(s, "https");
        assert_eq!(h, "ex.com:8080");
        assert_eq!(p, "/a/b?q=1");
        assert!(split_url("ftp://x").is_none());
    }

    #[test]
    fn browser_engine_without_sidecar_is_typed_refusal() {
        let req = FetchRequest {
            url: "https://example.com".into(),
            engine: "browser".into(),
            impersonate: String::new(),
            proxy: String::new(),
            respect_robots: false,
            render_wait: String::new(),
            timeout: Duration::from_secs(2),
            body_limit: DEFAULT_BODY_LIMIT,
        };
        // No enterprise fetcher registered in the unit test → OSS default.
        assert!(matches!(default_fetch(&req), Err(ScrapeError::NoBrowserSidecar)));
    }

    #[test]
    fn invalid_url_is_typed_refusal() {
        let req = FetchRequest {
            url: "notaurl".into(),
            engine: "impersonate".into(),
            impersonate: String::new(),
            proxy: String::new(),
            respect_robots: false,
            render_wait: String::new(),
            timeout: Duration::from_secs(2),
            body_limit: DEFAULT_BODY_LIMIT,
        };
        assert!(matches!(default_fetch(&req), Err(ScrapeError::InvalidUrl(_))));
    }

    #[test]
    fn missing_url_argument_is_error() {
        let entry = ToolEntry {
            name: "F".into(),
            provider: "scrape_http".into(),
            timeout: String::new(),
            runtime: String::new(),
            sandbox: None,
            max_results: None,
            output_schema: String::new(),
            effect_row: vec!["network".into(), "web".into()],
            parameters: Vec::new(),
            secret: String::new(),
            secret_partition: String::new(),
            source: crate::tool_registry::ToolSource::Program,
            is_streaming: false,
            scrape: Some(ScrapeConfig::default()),
        };
        let out = dispatch_scrape_outcome(&entry, "{}");
        assert!(!out.result.success);
        assert!(out.result.output.contains("url"));
    }
}