feat: add dual eval review and selectable URL crawling

.env.example

@@ -13,8 +13,21 @@ NVIDIA_API_BASE_URL=https://integrate.api.nvidia.com/v1
 # Admin review backend
 ADMIN_REVIEW_PROVIDER=auto
 ADMIN_REVIEW_MODEL=gemma4:latest
+ADMIN_REVIEW_PRIMARY_PROVIDER=nvidia
+ADMIN_REVIEW_PRIMARY_MODEL=google/gemma-4-31b-it
+ADMIN_REVIEW_SECONDARY_PROVIDER=ollama
+ADMIN_REVIEW_SECONDARY_MODEL=gemma4:26b
+ADMIN_REVIEW_CONSENSUS_ENABLED=true
+ADMIN_REVIEW_CONSENSUS_MIN_CONFIDENCE=0.75
 URL_INGEST_ADMIN_ENABLED=true
 URL_INGEST_ALLOWED_HOSTS=
+URL_INGEST_EXTRACTOR=basic
+URL_INGEST_CRAWL4AI_ENABLED=false
+URL_INGEST_CRAWL4AI_FALLBACK=false
+URL_INGEST_DEFAULT_MODE=single_page
+URL_INGEST_MAX_DEPTH=1
+URL_INGEST_MAX_PAGES=5
+URL_INGEST_CRAWL4AI_TIMEOUT_S=20
 URL_INGEST_TIMEOUT_S=12
 URL_INGEST_MAX_BYTES=1500000
 

backend/api/admin.py

@@ -54,12 +54,18 @@ class SeedEvalDatasetPayload(BaseModel):
 
 class DraftReviewPayload(BaseModel):
     model: Optional[str] = None
+    primary_model: Optional[str] = None
+    secondary_model: Optional[str] = None
+    dual_review: bool = False
     force: bool = False
 
 
 class BatchDraftReviewPayload(BaseModel):
     limit: int = 10
     model: Optional[str] = None
+    primary_model: Optional[str] = None
+    secondary_model: Optional[str] = None
+    dual_review: bool = False
     force: bool = False
     reviewed_only: bool = False
 
@@ -72,6 +78,9 @@ class CodeGraphIndexPayload(BaseModel):
 class UrlIngestPayload(BaseModel):
     url: str
     label: Optional[str] = None
+    mode: Optional[str] = None
+    max_depth: Optional[int] = None
+    max_pages: Optional[int] = None
 
 
 def _admin_client():
@@ -815,21 +824,42 @@ def generate_eval_dataset_draft_review(
             "skipped": True,
         }
 
-    from backend.eval.draft_reviewer import generate_eval_dataset_draft_review as _generate
+    from backend.eval.draft_reviewer import (
+        generate_eval_dataset_draft_review as _generate,
+        generate_eval_dataset_dual_draft_review as _generate_dual,
+    )
 
     trace_row = _load_trace_for_dataset(dataset_row.get("trace_id"))
     feedback_rows = _load_feedback_for_dataset(dataset_row.get("trace_id"))
-    draft = _generate(
-        dataset_row,
-        trace_row=trace_row,
-        feedback_rows=feedback_rows,
-        model=payload.model,
-    )
+    if payload.dual_review and config.ADMIN_REVIEW_CONSENSUS_ENABLED:
+        draft = _generate_dual(
+            dataset_row,
+            trace_row=trace_row,
+            feedback_rows=feedback_rows,
+            primary_model=payload.primary_model,
+            secondary_model=payload.secondary_model,
+        )
+        model_label = ",".join(
+            model
+            for model in [
+                (draft.get("primary") or {}).get("model"),
+                (draft.get("secondary") or {}).get("model"),
+            ]
+            if model
+        )
+    else:
+        draft = _generate(
+            dataset_row,
+            trace_row=trace_row,
+            feedback_rows=feedback_rows,
+            model=payload.model,
+        )
+        model_label = draft.get("model")
     now_iso = datetime.now(timezone.utc).isoformat()
     _admin_client().table("evaluation_datasets").update(
         {
             "ai_review_draft": draft,
-            "ai_review_model": draft.get("model"),
+            "ai_review_model": model_label,
             "ai_review_generated_at": now_iso,
         }
     ).eq("id", dataset_id).execute()
@@ -857,7 +887,13 @@ def batch_generate_eval_dataset_draft_reviews(
     for row in candidates[:batch_limit]:
         result = generate_eval_dataset_draft_review(
             int(row["id"]),
-            DraftReviewPayload(model=payload.model, force=payload.force),
+            DraftReviewPayload(
+                model=payload.model,
+                primary_model=payload.primary_model,
+                secondary_model=payload.secondary_model,
+                dual_review=payload.dual_review,
+                force=payload.force,
+            ),
             x_admin_key=x_admin_key,
         )
         generated.append(
@@ -880,6 +916,13 @@ def apply_eval_dataset_draft_review(
     draft = row.get("ai_review_draft") or {}
     if not draft:
         raise HTTPException(status_code=404, detail="No AI draft review found for this row.")
+    if "consensus" in draft:
+        draft = draft.get("consensus") or {}
+        if not draft.get("apply_ready"):
+            raise HTTPException(
+                status_code=409,
+                detail="AI reviewers disagree. Manually review this dataset row before applying.",
+            )
 
     now_iso = datetime.now(timezone.utc).isoformat()
     update = {
@@ -947,6 +990,11 @@ def get_url_ingest_options(
         "ok": True,
         "enabled": bool(config.URL_INGEST_ADMIN_ENABLED),
         "allowed_hosts": list(config.URL_INGEST_ALLOWED_HOSTS),
+        "extractor": config.URL_INGEST_EXTRACTOR,
+        "default_mode": config.URL_INGEST_DEFAULT_MODE,
+        "max_depth": config.URL_INGEST_MAX_DEPTH,
+        "max_pages": config.URL_INGEST_MAX_PAGES,
+        "crawl4ai_enabled": bool(config.URL_INGEST_CRAWL4AI_ENABLED),
         "timeout_s": config.URL_INGEST_TIMEOUT_S,
         "max_bytes": config.URL_INGEST_MAX_BYTES,
         "user_id": user_id,
@@ -964,7 +1012,13 @@ def run_url_ingest(
     _check_admin(x_admin_key)
     _ensure_url_ingest_enabled()
     try:
-        fetched = fetch_url_to_tempfile(payload.url, label=payload.label)
+        fetched = fetch_url_to_tempfile(
+            payload.url,
+            label=payload.label,
+            mode=payload.mode,
+            max_depth=payload.max_depth,
+            max_pages=payload.max_pages,
+        )
     except UrlIngestionError as exc:
         raise HTTPException(status_code=400, detail=str(exc)) from exc
     except Exception as exc:
@@ -996,4 +1050,5 @@ def run_url_ingest(
         "content_type": fetched.content_type,
         "content_bytes": fetched.content_bytes,
         "title": fetched.title,
+        "pages": fetched.pages or [],
     }

backend/core/config.py

@@ -51,6 +51,24 @@ ADMIN_REVIEW_MODEL = os.getenv(
     "ADMIN_REVIEW_MODEL",
     os.getenv("ADMIN_REVIEW_OLLAMA_MODEL", "gemma4:latest"),
 ).strip()
+ADMIN_REVIEW_PRIMARY_PROVIDER = os.getenv(
+    "ADMIN_REVIEW_PRIMARY_PROVIDER", "nvidia"
+).strip().lower()
+ADMIN_REVIEW_PRIMARY_MODEL = os.getenv(
+    "ADMIN_REVIEW_PRIMARY_MODEL", "google/gemma-4-31b-it"
+).strip()
+ADMIN_REVIEW_SECONDARY_PROVIDER = os.getenv(
+    "ADMIN_REVIEW_SECONDARY_PROVIDER", "ollama"
+).strip().lower()
+ADMIN_REVIEW_SECONDARY_MODEL = os.getenv(
+    "ADMIN_REVIEW_SECONDARY_MODEL", "gemma4:26b"
+).strip()
+ADMIN_REVIEW_CONSENSUS_ENABLED = os.getenv(
+    "ADMIN_REVIEW_CONSENSUS_ENABLED", "true"
+).lower() in {"1", "true", "yes"}
+ADMIN_REVIEW_CONSENSUS_MIN_CONFIDENCE = float(
+    os.getenv("ADMIN_REVIEW_CONSENSUS_MIN_CONFIDENCE", "0.75")
+)
 ADMIN_REVIEW_OLLAMA_MODEL = ADMIN_REVIEW_MODEL
 ADMIN_REVIEW_OLLAMA_TEMPERATURE = float(
     os.getenv("ADMIN_REVIEW_OLLAMA_TEMPERATURE", "0.1")
@@ -302,6 +320,17 @@ URL_INGEST_ALLOWED_HOSTS = [
     for segment in os.getenv("URL_INGEST_ALLOWED_HOSTS", "").split(",")
     if segment.strip()
 ]
+URL_INGEST_EXTRACTOR = os.getenv("URL_INGEST_EXTRACTOR", "basic").strip().lower()
+URL_INGEST_CRAWL4AI_ENABLED = os.getenv(
+    "URL_INGEST_CRAWL4AI_ENABLED", "false"
+).lower() in {"1", "true", "yes"}
+URL_INGEST_CRAWL4AI_FALLBACK = os.getenv(
+    "URL_INGEST_CRAWL4AI_FALLBACK", "false"
+).lower() in {"1", "true", "yes"}
+URL_INGEST_DEFAULT_MODE = os.getenv("URL_INGEST_DEFAULT_MODE", "single_page").strip().lower()
+URL_INGEST_MAX_DEPTH = int(os.getenv("URL_INGEST_MAX_DEPTH", "1"))
+URL_INGEST_MAX_PAGES = int(os.getenv("URL_INGEST_MAX_PAGES", "5"))
+URL_INGEST_CRAWL4AI_TIMEOUT_S = float(os.getenv("URL_INGEST_CRAWL4AI_TIMEOUT_S", "20"))
 URL_INGEST_ALLOWED_CONTENT_TYPES = {
     "text/html",
     "application/xhtml+xml",

backend/core/url_ingestion.py

@@ -4,13 +4,14 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 from html.parser import HTMLParser
+import asyncio
 import ipaddress
 import os
 from pathlib import Path
 import socket
 import tempfile
 from typing import Iterable
-from urllib.parse import urlparse
+from urllib.parse import urljoin, urlparse
 
 import requests
 
@@ -30,6 +31,18 @@ class FetchedUrlDocument:
     content_type: str
     content_bytes: int
     title: str | None = None
+    pages: list[dict] | None = None
+
+
+@dataclass
+class UrlPage:
+    source_url: str
+    final_url: str
+    title: str | None
+    markdown: str
+    content_type: str
+    content_bytes: int
+    links: list[str]
 
 
 class _VisibleTextParser(HTMLParser):
@@ -164,6 +177,23 @@ def _render_markdown_from_remote(source_url: str, title: str | None, body: str)
     return "\n".join(parts).strip() + "\n"
 
 
+def _render_combined_markdown(label: str | None, pages: list[UrlPage]) -> str:
+    title = label or (pages[0].title if pages else None) or "URL Import"
+    parts = [f"# {title}", ""]
+    for index, page in enumerate(pages, start=1):
+        page_title = page.title or page.final_url
+        parts.extend(
+            [
+                f"## Page {index}: {page_title}",
+                f"Source URL: {page.final_url}",
+                "",
+                page.markdown.strip(),
+                "",
+            ]
+        )
+    return "\n".join(parts).strip() + "\n"
+
+
 def _extract_text_payload(content_type: str, raw_bytes: bytes) -> tuple[str | None, str]:
     lowered = content_type.lower()
     decoded = raw_bytes.decode("utf-8", errors="replace")
@@ -180,31 +210,30 @@ def _extract_text_payload(content_type: str, raw_bytes: bytes) -> tuple[str | No
     return None, text
 
 
-def fetch_url_to_tempfile(url: str, *, label: str | None = None) -> FetchedUrlDocument:
-    parsed = urlparse((url or "").strip())
-    _ensure_safe_host(parsed)
+def _extract_links_from_html(base_url: str, raw_bytes: bytes, content_type: str) -> list[str]:
+    if not content_type.lower().startswith("text/html"):
+        return []
+    html = raw_bytes.decode("utf-8", errors="replace")
 
-    with requests.Session() as session:
-        response = session.get(
-            parsed.geturl(),
-            allow_redirects=True,
-            timeout=config.URL_INGEST_TIMEOUT_S,
-            stream=True,
-            headers={"User-Agent": config.URL_INGEST_USER_AGENT},
-        )
+    class _LinkParser(HTMLParser):
+        def __init__(self) -> None:
+            super().__init__()
+            self.links: list[str] = []
 
-    try:
-        response.raise_for_status()
-    except requests.HTTPError as exc:
-        raise UrlIngestionError(f"URL fetch failed with status {response.status_code}.") from exc
+        def handle_starttag(self, tag: str, attrs) -> None:  # noqa: ANN001
+            if tag != "a":
+                return
+            attrs_dict = dict(attrs)
+            href = (attrs_dict.get("href") or "").strip()
+            if href:
+                self.links.append(urljoin(base_url, href))
 
-    final_url = response.url or parsed.geturl()
-    _ensure_safe_host(urlparse(final_url))
+    parser = _LinkParser()
+    parser.feed(html)
+    return parser.links
 
-    content_type = (response.headers.get("content-type") or "").split(";", 1)[0].strip().lower()
-    if content_type not in config.URL_INGEST_ALLOWED_CONTENT_TYPES:
-        raise UrlIngestionError(f"Unsupported URL content type: {content_type or 'unknown'}.")
 
+def _read_limited_response(response) -> bytes:  # noqa: ANN001
     declared_length = response.headers.get("content-length")
     if declared_length:
         try:
@@ -220,20 +249,209 @@ def fetch_url_to_tempfile(url: str, *, label: str | None = None) -> FetchedUrlDo
         body.extend(chunk)
         if len(body) > config.URL_INGEST_MAX_BYTES:
             raise UrlIngestionError("Remote URL response exceeds the configured size limit.")
+    return bytes(body)
+
+
+def _fetch_basic_page(url: str, *, label: str | None = None) -> UrlPage:
+    parsed = urlparse((url or "").strip())
+    _ensure_safe_host(parsed)
 
-    title, text = _extract_text_payload(content_type, bytes(body))
+    with requests.Session() as session:
+        response = session.get(
+            parsed.geturl(),
+            allow_redirects=True,
+            timeout=config.URL_INGEST_TIMEOUT_S,
+            stream=True,
+            headers={"User-Agent": config.URL_INGEST_USER_AGENT},
+        )
+        try:
+            response.raise_for_status()
+        except requests.HTTPError as exc:
+            raise UrlIngestionError(f"URL fetch failed with status {response.status_code}.") from exc
+
+        final_url = response.url or parsed.geturl()
+        _ensure_safe_host(urlparse(final_url))
+        content_type = (response.headers.get("content-type") or "").split(";", 1)[0].strip().lower()
+        if content_type not in config.URL_INGEST_ALLOWED_CONTENT_TYPES:
+            raise UrlIngestionError(f"Unsupported URL content type: {content_type or 'unknown'}.")
+        body = _read_limited_response(response)
+
+    title, text = _extract_text_payload(content_type, body)
     rendered = _render_markdown_from_remote(final_url, label or title, text)
+    return UrlPage(
+        source_url=parsed.geturl(),
+        final_url=final_url,
+        title=label or title,
+        markdown=rendered,
+        content_type=content_type,
+        content_bytes=len(body),
+        links=_extract_links_from_html(final_url, body, content_type),
+    )
+
+
+def _run_crawl4ai_single_page(url: str, *, label: str | None = None) -> UrlPage:
+    if not config.URL_INGEST_CRAWL4AI_ENABLED:
+        raise UrlIngestionError("Crawl4AI URL extraction is disabled.")
+    parsed = urlparse((url or "").strip())
+    _ensure_safe_host(parsed)
+    try:
+        from crawl4ai import AsyncWebCrawler
+    except Exception as exc:
+        if config.URL_INGEST_CRAWL4AI_FALLBACK:
+            return _fetch_basic_page(url, label=label)
+        raise UrlIngestionError("Crawl4AI is not installed or could not be imported.") from exc
+
+    async def _crawl():
+        async with AsyncWebCrawler() as crawler:
+            return await asyncio.wait_for(
+                crawler.arun(url=parsed.geturl()),
+                timeout=config.URL_INGEST_CRAWL4AI_TIMEOUT_S,
+            )
+
+    try:
+        result = asyncio.run(_crawl())
+    except Exception as exc:
+        if config.URL_INGEST_CRAWL4AI_FALLBACK:
+            return _fetch_basic_page(url, label=label)
+        raise UrlIngestionError(f"Crawl4AI extraction failed: {exc}") from exc
+
+    final_url = str(getattr(result, "url", None) or parsed.geturl())
+    _ensure_safe_host(urlparse(final_url))
+    markdown = str(getattr(result, "markdown", "") or "").strip()
+    if not markdown:
+        raise UrlIngestionError("Crawl4AI returned empty markdown.")
+    if len(markdown.encode("utf-8")) > config.URL_INGEST_MAX_BYTES:
+        raise UrlIngestionError("Crawl4AI markdown exceeds the configured size limit.")
+    metadata = getattr(result, "metadata", None) or {}
+    title = label or metadata.get("title") or None
+    links: list[str] = []
+    raw_links = getattr(result, "links", None)
+    if isinstance(raw_links, dict):
+        for values in raw_links.values():
+            if isinstance(values, list):
+                for item in values:
+                    if isinstance(item, dict) and item.get("href"):
+                        links.append(urljoin(final_url, str(item["href"])))
+                    elif isinstance(item, str):
+                        links.append(urljoin(final_url, item))
+    return UrlPage(
+        source_url=parsed.geturl(),
+        final_url=final_url,
+        title=title,
+        markdown=_render_markdown_from_remote(final_url, title, markdown),
+        content_type="text/markdown",
+        content_bytes=len(markdown.encode("utf-8")),
+        links=links,
+    )
+
+
+def _fetch_url_page(url: str, *, label: str | None = None) -> UrlPage:
+    if config.URL_INGEST_EXTRACTOR == "crawl4ai":
+        return _run_crawl4ai_single_page(url, label=label)
+    if config.URL_INGEST_EXTRACTOR != "basic":
+        raise UrlIngestionError("URL_INGEST_EXTRACTOR must be basic or crawl4ai.")
+    return _fetch_basic_page(url, label=label)
+
+
+def _same_allowed_host(candidate_url: str, root_hostname: str) -> bool:
+    parsed = urlparse(candidate_url)
+    try:
+        hostname = _normalize_host(parsed.hostname)
+    except UrlIngestionError:
+        return False
+    return hostname == root_hostname
+
+
+def _crawl_bounded_pages(
+    url: str,
+    *,
+    label: str | None,
+    max_depth: int,
+    max_pages: int,
+) -> list[UrlPage]:
+    if not config.URL_INGEST_ALLOWED_HOSTS:
+        raise UrlIngestionError("Bounded URL crawl requires URL_INGEST_ALLOWED_HOSTS.")
+    root = urlparse((url or "").strip())
+    root_host = _normalize_host(root.hostname)
+    _ensure_safe_host(root)
+
+    seen: set[str] = set()
+    queue: list[tuple[str, int]] = [(root.geturl(), 0)]
+    pages: list[UrlPage] = []
+    while queue and len(pages) < max_pages:
+        current_url, depth = queue.pop(0)
+        normalized = current_url.split("#", 1)[0]
+        if normalized in seen:
+            continue
+        seen.add(normalized)
+        page = _fetch_url_page(normalized, label=label if not pages else None)
+        pages.append(page)
+        if depth >= max_depth:
+            continue
+        for link in page.links:
+            link_without_fragment = link.split("#", 1)[0]
+            if link_without_fragment in seen:
+                continue
+            if not _same_allowed_host(link_without_fragment, root_host):
+                continue
+            try:
+                _ensure_safe_host(urlparse(link_without_fragment))
+            except UrlIngestionError:
+                continue
+            queue.append((link_without_fragment, depth + 1))
+            if len(queue) + len(pages) >= max_pages:
+                break
+    if not pages:
+        raise UrlIngestionError("No crawlable pages were fetched.")
+    return pages
+
+
+def fetch_url_to_tempfile(
+    url: str,
+    *,
+    label: str | None = None,
+    mode: str | None = None,
+    max_depth: int | None = None,
+    max_pages: int | None = None,
+) -> FetchedUrlDocument:
+    selected_mode = (mode or config.URL_INGEST_DEFAULT_MODE or "single_page").strip().lower()
+    page_limit = max(1, min(max_pages or config.URL_INGEST_MAX_PAGES, config.URL_INGEST_MAX_PAGES))
+    depth_limit = max(0, min(max_depth if max_depth is not None else config.URL_INGEST_MAX_DEPTH, config.URL_INGEST_MAX_DEPTH))
+    if selected_mode == "bounded_crawl":
+        pages = _crawl_bounded_pages(
+            url,
+            label=label,
+            max_depth=depth_limit,
+            max_pages=page_limit,
+        )
+    elif selected_mode == "single_page":
+        pages = [_fetch_url_page(url, label=label)]
+    else:
+        raise UrlIngestionError("URL ingest mode must be single_page or bounded_crawl.")
+
+    rendered = _render_combined_markdown(label, pages)
+    final_url = pages[0].final_url
     tmp_fd, tmp_path = tempfile.mkstemp(suffix=".md", prefix="morpheus_url_")
     os.close(tmp_fd)
     with open(tmp_path, "w", encoding="utf-8") as handle:
         handle.write(rendered)
 
     return FetchedUrlDocument(
-        source_url=parsed.geturl(),
+        source_url=urlparse((url or "").strip()).geturl(),
         final_url=final_url,
-        filename=_sanitize_filename(final_url, label or title),
+        filename=_sanitize_filename(final_url, label or pages[0].title),
         temp_path=tmp_path,
-        content_type=content_type,
-        content_bytes=len(body),
-        title=label or title,
+        content_type=pages[0].content_type,
+        content_bytes=sum(page.content_bytes for page in pages),
+        title=label or pages[0].title,
+        pages=[
+            {
+                "source_url": page.source_url,
+                "final_url": page.final_url,
+                "title": page.title,
+                "content_type": page.content_type,
+                "content_bytes": page.content_bytes,
+            }
+            for page in pages
+        ],
     )

backend/eval/draft_reviewer.py

@@ -1,4 +1,5 @@
 import json
+from concurrent.futures import ThreadPoolExecutor
 from typing import Any, Optional
 
 from backend.core import config
@@ -172,8 +173,8 @@ def _build_prompt(dataset_row: dict[str, Any], trace_row: Optional[dict[str, Any
     )
 
 
-def _resolve_admin_review_provider(model: str) -> str:
-    configured = (config.ADMIN_REVIEW_PROVIDER or "auto").strip().lower()
+def _resolve_admin_review_provider(model: str, provider: Optional[str] = None) -> str:
+    configured = (provider or config.ADMIN_REVIEW_PROVIDER or "auto").strip().lower()
     if configured in {"ollama", "nvidia"}:
         return configured
     if config.NVIDIA_API_KEY and "/" in (model or ""):
@@ -185,9 +186,10 @@ def _generate_review_text(
     *,
     prompt: str,
     model: str,
+    provider: Optional[str] = None,
 ) -> str:
-    provider = _resolve_admin_review_provider(model)
-    if provider == "nvidia":
+    resolved_provider = _resolve_admin_review_provider(model, provider)
+    if resolved_provider == "nvidia":
         return nvidia_generate(
             base_url=config.NVIDIA_API_BASE_URL,
             api_key=config.NVIDIA_API_KEY or "",
@@ -214,11 +216,13 @@ def generate_eval_dataset_draft_review(
     trace_row: Optional[dict[str, Any]] = None,
     feedback_rows: Optional[list[dict[str, Any]]] = None,
     model: Optional[str] = None,
+    provider: Optional[str] = None,
 ) -> dict[str, Any]:
     prompt = _build_prompt(dataset_row, trace_row, feedback_rows or [])
     selected_model = model or config.ADMIN_REVIEW_MODEL
-    raw = _generate_review_text(prompt=prompt, model=selected_model)
+    raw = _generate_review_text(prompt=prompt, model=selected_model, provider=provider)
     data = _extract_json_object(raw)
+    resolved_provider = _resolve_admin_review_provider(selected_model, provider)
     return {
         "manual_verdict": _normalize_verdict(data.get("manual_verdict")),
         "gold_answer_text": _clean_text(data.get("gold_answer_text"))
@@ -233,5 +237,133 @@ def generate_eval_dataset_draft_review(
         "confidence": _normalize_confidence(data.get("confidence")),
         "rationale": _clean_text(data.get("rationale")) or "Draft generated from trace, feedback, and existing evidence.",
         "model": selected_model,
-        "provider": _resolve_admin_review_provider(selected_model),
+        "provider": resolved_provider,
+    }
+
+
+def _token_set(text: str) -> set[str]:
+    return {
+        token
+        for token in "".join(ch.lower() if ch.isalnum() else " " for ch in text or "").split()
+        if len(token) > 2
+    }
+
+
+def _text_compatible(left: str, right: str) -> bool:
+    left_tokens = _token_set(left)
+    right_tokens = _token_set(right)
+    if not left_tokens or not right_tokens:
+        return False
+    overlap = len(left_tokens & right_tokens) / max(1, min(len(left_tokens), len(right_tokens)))
+    return overlap >= 0.25
+
+
+def _error_draft(*, model: str, provider: str, exc: Exception) -> dict[str, Any]:
+    return {
+        "manual_verdict": "partial",
+        "gold_answer_text": "",
+        "gold_evidence_text": "",
+        "failure_mode": "other",
+        "confidence": 0.0,
+        "rationale": f"Reviewer failed: {exc}",
+        "model": model,
+        "provider": provider,
+        "error": str(exc),
+    }
+
+
+def _build_consensus(primary: dict[str, Any], secondary: Optional[dict[str, Any]]) -> dict[str, Any]:
+    if primary.get("error") and secondary and not secondary.get("error"):
+        return {
+            **secondary,
+            "agreement_status": "secondary_only",
+            "apply_ready": True,
+            "confidence": min(float(secondary.get("confidence") or 0.0), 0.74),
+            "rationale": "Only the secondary reviewer produced a usable draft.",
+        }
+    if not secondary or secondary.get("error"):
+        return {
+            **primary,
+            "agreement_status": "primary_only",
+            "apply_ready": True,
+            "confidence": min(float(primary.get("confidence") or 0.0), 0.74),
+            "rationale": "Only the primary reviewer produced a usable draft.",
+        }
+
+    verdict_match = primary.get("manual_verdict") == secondary.get("manual_verdict")
+    answer_match = _text_compatible(
+        primary.get("gold_answer_text") or "",
+        secondary.get("gold_answer_text") or "",
+    )
+    evidence_match = _text_compatible(
+        primary.get("gold_evidence_text") or "",
+        secondary.get("gold_evidence_text") or "",
+    )
+    min_confidence = min(
+        float(primary.get("confidence") or 0.0),
+        float(secondary.get("confidence") or 0.0),
+    )
+    chosen = primary if float(primary.get("confidence") or 0.0) >= float(secondary.get("confidence") or 0.0) else secondary
+    compatible = verdict_match and (answer_match or evidence_match)
+    if compatible and min_confidence >= config.ADMIN_REVIEW_CONSENSUS_MIN_CONFIDENCE:
+        status = "agreement"
+    elif compatible:
+        status = "weak_agreement"
+    else:
+        status = "disagreement"
+    return {
+        "manual_verdict": chosen.get("manual_verdict") or "partial",
+        "gold_answer_text": chosen.get("gold_answer_text") or "",
+        "gold_evidence_text": chosen.get("gold_evidence_text") or "",
+        "failure_mode": chosen.get("failure_mode"),
+        "confidence": min_confidence,
+        "rationale": (
+            "Both reviewers agree enough for draft application."
+            if status in {"agreement", "weak_agreement"}
+            else "Reviewers disagreed; human edit is required before applying."
+        ),
+        "agreement_status": status,
+        "apply_ready": status in {"agreement", "weak_agreement"},
+        "primary_model": primary.get("model"),
+        "secondary_model": secondary.get("model"),
+    }
+
+
+def generate_eval_dataset_dual_draft_review(
+    dataset_row: dict[str, Any],
+    *,
+    trace_row: Optional[dict[str, Any]] = None,
+    feedback_rows: Optional[list[dict[str, Any]]] = None,
+    primary_model: Optional[str] = None,
+    secondary_model: Optional[str] = None,
+    primary_provider: Optional[str] = None,
+    secondary_provider: Optional[str] = None,
+) -> dict[str, Any]:
+    selected_primary_model = primary_model or config.ADMIN_REVIEW_PRIMARY_MODEL
+    selected_secondary_model = secondary_model or config.ADMIN_REVIEW_SECONDARY_MODEL
+    selected_primary_provider = primary_provider or config.ADMIN_REVIEW_PRIMARY_PROVIDER
+    selected_secondary_provider = secondary_provider or config.ADMIN_REVIEW_SECONDARY_PROVIDER
+
+    def _run(model: str, provider: str) -> dict[str, Any]:
+        try:
+            return generate_eval_dataset_draft_review(
+                dataset_row,
+                trace_row=trace_row,
+                feedback_rows=feedback_rows or [],
+                model=model,
+                provider=provider,
+            )
+        except Exception as exc:
+            return _error_draft(model=model, provider=provider, exc=exc)
+
+    with ThreadPoolExecutor(max_workers=2) as executor:
+        primary_future = executor.submit(_run, selected_primary_model, selected_primary_provider)
+        secondary_future = executor.submit(_run, selected_secondary_model, selected_secondary_provider)
+        primary = primary_future.result()
+        secondary = secondary_future.result()
+
+    return {
+        "primary": primary,
+        "secondary": secondary,
+        "consensus": _build_consensus(primary, secondary),
     }

docs/operations_playbook.md

@@ -5,6 +5,7 @@
 - Promote high-signal traces into `evaluation_datasets`
 - Review Gemma drafts, then manually activate trusted rows
 - If using hosted Gemma review, ensure `NVIDIA_API_KEY` is set and valid
+- If using dual review, compare consensus disagreements before applying drafts
 
 ## Monthly
 - Run retrieval-profile sweeps against the active reviewed baseline
@@ -22,10 +23,12 @@
 - Use Graph view for summary, search, path, and export
 - Code indexing is operator-only and restricted to allowed roots
 - URL ingestion is operator-only and restricted by host allowlist plus public-IP checks
+- Bounded URL crawl must stay within configured depth and page budgets
 - Keep PDF answer-first flow unchanged while graph features expand
 
 ## Admin Review Backends
 - `ADMIN_REVIEW_PROVIDER=auto` prefers NVIDIA hosted models when `NVIDIA_API_KEY` is present and the model looks hosted
 - `ADMIN_REVIEW_PROVIDER=ollama` forces local review generation
 - `ADMIN_REVIEW_PROVIDER=nvidia` forces NVIDIA hosted review generation
+- Dual review uses `ADMIN_REVIEW_PRIMARY_*` and `ADMIN_REVIEW_SECONDARY_*` settings, then stores consensus for human approval
 - Rotate and replace any leaked provider keys before putting them into runtime envs

docs/release_checklist.md

@@ -13,6 +13,8 @@
   - `MASTER_ADMIN_KEY`
   - graph/code index env vars if used
   - if hosted admin review is enabled: `ADMIN_REVIEW_PROVIDER`, `ADMIN_REVIEW_MODEL`, `NVIDIA_API_KEY`, `NVIDIA_API_BASE_URL`
+  - if dual review is enabled: `ADMIN_REVIEW_PRIMARY_MODEL`, `ADMIN_REVIEW_SECONDARY_MODEL`, `ADMIN_REVIEW_CONSENSUS_ENABLED`
+  - if Crawl4AI URL extraction is enabled: `URL_INGEST_EXTRACTOR`, `URL_INGEST_CRAWL4AI_ENABLED`, `URL_INGEST_ALLOWED_HOSTS`
 
 ## Post-Deploy Smoke
 - `GET /health`
@@ -21,8 +23,10 @@
 - Query streaming still returns sources and a trace id
 - Admin review loads traces and evaluation datasets
 - Admin draft review works with the configured provider (`ollama` or NVIDIA hosted)
+- Dual admin draft review shows primary, secondary, and consensus status when enabled
 - Graph summary/search/export endpoints return tenant-scoped data
 - Operator URL ingest accepts only safe public hosts and queues background ingestion
+- Bounded URL crawl is constrained by allowed hosts, depth, and page budget
 
 ## Operator Checks
 - Confirm reviewed eval rows and active eval rows are non-zero

frontend/index.html

@@ -515,6 +515,14 @@
               <div class="confirm-zone" id="graphUrlOperatorHelp">URL ingestion is unavailable.</div>
               <input type="text" id="graphUrlInput" placeholder="https://example.com/docs/page" style="margin-top:10px;" />
               <input type="text" id="graphUrlLabel" placeholder="Optional display label…" style="margin-top:8px;" />
+              <select id="graphUrlMode" onchange="syncGraphUrlMode()" style="margin-top:8px;">
+                <option value="single_page">Single page</option>
+                <option value="bounded_crawl">Bounded crawl</option>
+              </select>
+              <div id="graphUrlCrawlFields" style="display:none;gap:8px;flex-wrap:wrap;margin-top:8px;">
+                <input type="number" id="graphUrlMaxDepth" min="0" max="3" value="1" placeholder="Max depth" style="flex:1;min-width:100px;" />
+                <input type="number" id="graphUrlMaxPages" min="1" max="20" value="5" placeholder="Max pages" style="flex:1;min-width:100px;" />
+              </div>
               <div style="display:flex;gap:8px;flex-wrap:wrap;margin-top:10px;">
                 <button class="btn-primary" onclick="runOperatorUrlIngest()">INGEST URL</button>
               </div>

frontend/js/admin.js

@@ -157,6 +157,12 @@ function _renderEvalDatasetList(rows) {
     const categories = Array.isArray(row.document_types) ? row.document_types : [];
     const draft = row.ai_review_draft || {};
     const hasDraft = Object.keys(draft).length > 0;
+    const primaryDraft = draft.primary || draft;
+    const secondaryDraft = draft.secondary || null;
+    const consensus = draft.consensus || null;
+    const displayDraft = consensus || primaryDraft;
+    const agreementStatus = consensus?.agreement_status || (hasDraft ? 'single_reviewer' : null);
+    const agreementTone = agreementStatus === 'disagreement' ? 'danger' : agreementStatus === 'agreement' ? 'success' : 'muted';
     return `
       <div style="padding:10px;border:1px solid #22304a;border-radius:10px;background:rgba(7,12,24,0.72);margin-bottom:10px;">
         <div style="display:flex;justify-content:space-between;gap:12px;align-items:flex-start;">
@@ -177,12 +183,18 @@ function _renderEvalDatasetList(rows) {
           <div style="margin-top:10px;padding:10px;border:1px solid #243142;border-radius:8px;background:rgba(10,18,32,0.55);">
             <div style="font-size:0.76rem;color:#7dd3fc;letter-spacing:0.12em;text-transform:uppercase;">Gemma Draft</div>
             <div style="margin-top:6px;">
-              ${_adminBadge(`verdict ${draft.manual_verdict || 'partial'}`, draft.manual_verdict === 'wrong' ? 'danger' : draft.manual_verdict === 'correct' ? 'success' : 'muted')}
-              ${_adminBadge(`confidence ${Number(draft.confidence || 0).toFixed(2)}`)}
-              ${draft.failure_mode ? _adminBadge(`failure ${draft.failure_mode}`, 'danger') : ''}
+              ${_adminBadge(`verdict ${displayDraft.manual_verdict || 'partial'}`, displayDraft.manual_verdict === 'wrong' ? 'danger' : displayDraft.manual_verdict === 'correct' ? 'success' : 'muted')}
+              ${_adminBadge(`confidence ${Number(displayDraft.confidence || 0).toFixed(2)}`)}
+              ${agreementStatus ? _adminBadge(`agreement ${agreementStatus}`, agreementTone) : ''}
+              ${displayDraft.failure_mode ? _adminBadge(`failure ${displayDraft.failure_mode}`, 'danger') : ''}
             </div>
-            <div style="margin-top:8px;font-size:0.8rem;color:#cbd5e1;white-space:pre-wrap;">${esc(draft.gold_answer_text || '')}</div>
-            ${draft.rationale ? `<div style="margin-top:6px;font-size:0.78rem;color:#94a3b8;">${esc(draft.rationale)}</div>` : ''}
+            <div style="margin-top:8px;font-size:0.8rem;color:#cbd5e1;white-space:pre-wrap;">${esc(displayDraft.gold_answer_text || '')}</div>
+            ${displayDraft.rationale ? `<div style="margin-top:6px;font-size:0.78rem;color:#94a3b8;">${esc(displayDraft.rationale)}</div>` : ''}
+            <div style="margin-top:8px;font-size:0.74rem;color:#94a3b8;">
+              primary ${esc(primaryDraft.provider || 'unknown')}/${esc(primaryDraft.model || 'unknown')}
+              ${secondaryDraft ? ` · secondary ${esc(secondaryDraft.provider || 'unknown')}/${esc(secondaryDraft.model || 'unknown')}` : ''}
+            </div>
+            ${agreementStatus === 'disagreement' ? `<div style="margin-top:8px;color:#fb7185;font-size:0.78rem;">Reviewers disagreed. Edit manually before applying.</div>` : ''}
           </div>
         ` : ''}
         <div style="display:flex;gap:8px;flex-wrap:wrap;margin-top:10px;">
@@ -347,8 +359,18 @@ async function generateAdminEvalDraft(datasetId) {
   if (!STATE.adminKey) return;
   const modelRaw = window.prompt('Review model for draft review', 'gemma4:latest');
   if (modelRaw === null) return;
+  const dualReview = window.confirm('Run dual-review with NVIDIA + local Ollama when configured?');
+  let primaryModel = null;
+  let secondaryModel = null;
+  if (dualReview) {
+    primaryModel = window.prompt('Primary review model (blank = configured NVIDIA default)', '') || null;
+    secondaryModel = window.prompt('Secondary review model (blank = configured local default)', '') || null;
+  }
   await apiAdminGenerateEvalDraft(STATE.adminKey, datasetId, {
     model: modelRaw.trim() || null,
+    primary_model: primaryModel?.trim() || null,
+    secondary_model: secondaryModel?.trim() || null,
+    dual_review: dualReview,
     force: true,
   });
   toast('Gemma draft review generated.', 'success');
@@ -361,9 +383,19 @@ async function batchGenerateAdminEvalDrafts() {
   if (limitRaw === null) return;
   const modelRaw = window.prompt('Review model for batch drafts', 'gemma4:latest');
   if (modelRaw === null) return;
+  const dualReview = window.confirm('Run dual-review for this batch? Recommended when NVIDIA and local Ollama are configured.');
+  let primaryModel = null;
+  let secondaryModel = null;
+  if (dualReview) {
+    primaryModel = window.prompt('Primary review model (blank = configured NVIDIA default)', '') || null;
+    secondaryModel = window.prompt('Secondary review model (blank = configured local default)', '') || null;
+  }
   const result = await apiAdminBatchGenerateEvalDrafts(STATE.adminKey, {
     limit: Number(limitRaw) || 10,
     model: modelRaw.trim() || null,
+    primary_model: primaryModel?.trim() || null,
+    secondary_model: secondaryModel?.trim() || null,
+    dual_review: dualReview,
     force: false,
   });
   toast(`Generated ${Number(result.generated_count || 0)} Gemma draft(s).`, 'success');

frontend/js/graph.js

@@ -491,6 +491,7 @@ async function refreshGraphOperatorState() {
   STATE.graphAllowedRoots = options.allowed_roots || [];
   STATE.graphIndexDefaultRoot = options.default_root || '';
   STATE.graphAllowedHosts = urlOptions.allowed_hosts || [];
+  STATE.graphUrlOptions = urlOptions || {};
   if (document.getElementById('graphIndexRoot') && !document.getElementById('graphIndexRoot').value) {
     document.getElementById('graphIndexRoot').value = STATE.graphIndexDefaultRoot || '';
   }
@@ -498,9 +499,10 @@ async function refreshGraphOperatorState() {
     ? `Allowed roots:<br>${STATE.graphAllowedRoots.map(root => `<code>${esc(root)}</code>`).join('<br>')}`
     : 'No operator code roots are configured.';
   if (urlHelp) {
+    const limits = `mode ${esc(urlOptions.default_mode || 'single_page')} · max depth ${Number(urlOptions.max_depth || 0)} · max pages ${Number(urlOptions.max_pages || 1)} · extractor ${esc(urlOptions.extractor || 'basic')}`;
     urlHelp.innerHTML = STATE.graphAllowedHosts.length
-      ? `Allowed hosts:<br>${STATE.graphAllowedHosts.map(host => `<code>${esc(host)}</code>`).join('<br>')}`
-      : 'No host allowlist is configured. URL ingestion still blocks private and non-routable hosts.';
+      ? `${limits}<br>Allowed hosts:<br>${STATE.graphAllowedHosts.map(host => `<code>${esc(host)}</code>`).join('<br>')}`
+      : `${limits}<br>No host allowlist is configured. Single-page ingestion still blocks private and non-routable hosts; bounded crawl requires an allowlist.`;
   }
 }
 
@@ -577,6 +579,9 @@ async function runOperatorUrlIngest() {
   }
   const url = document.getElementById('graphUrlInput')?.value?.trim() || '';
   const label = document.getElementById('graphUrlLabel')?.value?.trim() || '';
+  const mode = document.getElementById('graphUrlMode')?.value || 'single_page';
+  const maxDepth = Number(document.getElementById('graphUrlMaxDepth')?.value || 1);
+  const maxPages = Number(document.getElementById('graphUrlMaxPages')?.value || 5);
   if (!url) {
     toast('Enter a URL to ingest first.', 'error');
     return;
@@ -584,10 +589,19 @@ async function runOperatorUrlIngest() {
   const result = await apiAdminIngestUrl(STATE.adminKey, {
     url,
     label: label || null,
+    mode,
+    max_depth: maxDepth,
+    max_pages: maxPages,
   });
   toast(`Queued URL ingestion for ${result.final_url || result.source_url}.`, 'success');
 }
 
+function syncGraphUrlMode() {
+  const mode = document.getElementById('graphUrlMode')?.value || 'single_page';
+  const crawlFields = document.getElementById('graphUrlCrawlFields');
+  if (crawlFields) crawlFields.style.display = mode === 'bounded_crawl' ? 'flex' : 'none';
+}
+
 function downloadGraphExport() {
   const snapshot = STATE.graphSnapshot;
   if (!snapshot) {
@@ -695,6 +709,7 @@ window.runGraphSearch = runGraphSearch;
 window.runGraphPath = runGraphPath;
 window.runOperatorCodeIndex = runOperatorCodeIndex;
 window.runOperatorUrlIngest = runOperatorUrlIngest;
+window.syncGraphUrlMode = syncGraphUrlMode;
 window.downloadGraphExport = downloadGraphExport;
 window.selectGraphNode = selectGraphNode;
 window.selectGraphNodeByKey = selectGraphNodeByKey;

frontend/js/state.js

@@ -37,6 +37,7 @@ const STATE = {
   graphAllowedRoots: [],
   graphIndexDefaultRoot: '',
   graphAllowedHosts: [],
+  graphUrlOptions: {},
 };
 
 function stateRefreshCategories() {

tests/test_pipeline_regressions.py

@@ -9,6 +9,8 @@ from pathlib import Path
 from types import SimpleNamespace
 
 from langchain_core.documents import Document
+import pytest
+from fastapi import HTTPException
 from starlette.requests import Request
 
 from backend.api import admin
@@ -3022,6 +3024,80 @@ def test_draft_reviewer_can_route_to_nvidia_hosted_model(monkeypatch):
     assert captured["kwargs"]["enable_thinking"] is True
 
 
+def test_draft_reviewer_dual_review_builds_agreement_consensus(monkeypatch):
+    import backend.eval.draft_reviewer as draft_reviewer
+
+    def _fake_generate(*, model, provider=None, **_kwargs):
+        del provider
+        if model == "google/gemma-4-31b-it":
+            answer = "The guide exists because definitions are not enough."
+        else:
+            answer = "The guide exists because knowing definitions is not enough."
+        return json.dumps(
+            {
+                "manual_verdict": "correct",
+                "gold_answer_text": answer,
+                "gold_evidence_text": "This guide exists because knowing definitions is not enough.",
+                "failure_mode": None,
+                "confidence": 0.91,
+                "rationale": "Grounded in the trace.",
+            }
+        )
+
+    monkeypatch.setattr(draft_reviewer, "_generate_review_text", _fake_generate)
+
+    result = draft_reviewer.generate_eval_dataset_dual_draft_review(
+        {
+            "question": "Why does this guide exist?",
+            "gold_answer_text": "",
+            "gold_evidence_text": "",
+        },
+        primary_model="google/gemma-4-31b-it",
+        secondary_model="gemma4:26b",
+        primary_provider="nvidia",
+        secondary_provider="ollama",
+    )
+
+    assert result["primary"]["provider"] == "nvidia"
+    assert result["secondary"]["provider"] == "ollama"
+    assert result["consensus"]["agreement_status"] == "agreement"
+    assert result["consensus"]["apply_ready"] is True
+
+
+def test_admin_apply_eval_dataset_dual_draft_blocks_disagreement(monkeypatch):
+    fake_service = FakeServiceSupabase()
+    fake_service.eval_rows.append(
+        {
+            "id": 33,
+            "trace_id": "trace-ai-disagree",
+            "question": "What is the title?",
+            "manual_verdict": None,
+            "is_active": False,
+            "ai_review_draft": {
+                "primary": {"manual_verdict": "correct", "model": "google/gemma-4-31b-it"},
+                "secondary": {"manual_verdict": "wrong", "model": "gemma4:26b"},
+                "consensus": {
+                    "manual_verdict": "partial",
+                    "gold_answer_text": "",
+                    "gold_evidence_text": "",
+                    "failure_mode": "other",
+                    "confidence": 0.4,
+                    "agreement_status": "disagreement",
+                    "apply_ready": False,
+                },
+            },
+        }
+    )
+
+    monkeypatch.setattr(admin, "_admin_client", lambda: fake_service)
+    monkeypatch.setenv("MASTER_ADMIN_KEY", "secret")
+
+    with pytest.raises(HTTPException) as exc_info:
+        admin.apply_eval_dataset_draft_review(33, x_admin_key="secret")
+
+    assert exc_info.value.status_code == 409
+
+
 def test_assess_profile_change_accepts_only_when_acceptance_rules_pass():
     dataset = [
         {"question": f"q{i}", "manual_verdict": "partial" if i else "correct"}

tests/test_url_ingestion.py

@@ -86,6 +86,81 @@ def test_fetch_url_to_tempfile_extracts_readable_html(monkeypatch):
     assert "Hybrid retrieval works." in rendered
 
 
+def test_fetch_url_to_tempfile_uses_crawl4ai_extractor_when_enabled(monkeypatch):
+    monkeypatch.setattr(url_ingestion.config, "URL_INGEST_EXTRACTOR", "crawl4ai")
+    monkeypatch.setattr(
+        url_ingestion,
+        "_run_crawl4ai_single_page",
+        lambda url, label=None: url_ingestion.UrlPage(
+            source_url=url,
+            final_url=url,
+            title=label or "Crawl4AI Page",
+            markdown="# Crawl4AI Page\n\nClean markdown.",
+            content_type="text/markdown",
+            content_bytes=32,
+            links=[],
+        ),
+    )
+
+    result = url_ingestion.fetch_url_to_tempfile("https://docs.example.com/guide")
+
+    assert result.title == "Crawl4AI Page"
+    with open(result.temp_path, "r", encoding="utf-8") as handle:
+        rendered = handle.read()
+    assert "Clean markdown." in rendered
+
+
+def test_fetch_url_to_tempfile_bounded_crawl_combines_pages(monkeypatch):
+    monkeypatch.setattr(url_ingestion.config, "URL_INGEST_ALLOWED_HOSTS", ["docs.example.com"])
+    monkeypatch.setattr(url_ingestion.config, "URL_INGEST_MAX_DEPTH", 1)
+    monkeypatch.setattr(url_ingestion.config, "URL_INGEST_MAX_PAGES", 3)
+    monkeypatch.setattr(
+        url_ingestion.socket,
+        "getaddrinfo",
+        lambda *args, **kwargs: [(None, None, None, None, ("93.184.216.34", 443))],
+    )
+
+    def _fake_fetch(url, label=None):
+        links = ["https://docs.example.com/second", "https://other.example.com/blocked"] if url.endswith("/start") else []
+        return url_ingestion.UrlPage(
+            source_url=url,
+            final_url=url,
+            title=label or url.rsplit("/", 1)[-1],
+            markdown=f"# {url}\n\nPage body.",
+            content_type="text/html",
+            content_bytes=100,
+            links=links,
+        )
+
+    monkeypatch.setattr(url_ingestion, "_fetch_url_page", _fake_fetch)
+
+    result = url_ingestion.fetch_url_to_tempfile(
+        "https://docs.example.com/start",
+        mode="bounded_crawl",
+        max_depth=1,
+        max_pages=3,
+    )
+
+    assert len(result.pages) == 2
+    with open(result.temp_path, "r", encoding="utf-8") as handle:
+        rendered = handle.read()
+    assert "https://docs.example.com/start" in rendered
+    assert "https://docs.example.com/second" in rendered
+    assert "other.example.com" not in rendered
+
+
+def test_fetch_url_to_tempfile_bounded_crawl_requires_allowlist(monkeypatch):
+    monkeypatch.setattr(url_ingestion.config, "URL_INGEST_ALLOWED_HOSTS", [])
+
+    with pytest.raises(url_ingestion.UrlIngestionError, match="requires URL_INGEST_ALLOWED_HOSTS"):
+        url_ingestion.fetch_url_to_tempfile(
+            "https://docs.example.com/start",
+            mode="bounded_crawl",
+            max_depth=1,
+            max_pages=2,
+        )
+
+
 def test_run_ingestion_with_url_override_persists_url_metadata(monkeypatch):
     from tests.test_pipeline_regressions import FakeIngestionSupabase
 
@@ -154,7 +229,7 @@ def test_run_url_ingest_requires_admin_and_queues_task(monkeypatch):
     monkeypatch.setattr(
         admin_api,
         "fetch_url_to_tempfile",
-        lambda url, label=None: SimpleNamespace(
+        lambda url, label=None, mode=None, max_depth=None, max_pages=None: SimpleNamespace(
             source_url=url,
             final_url=url,
             filename="Morpheus-Docs.md",
@@ -162,12 +237,17 @@ def test_run_url_ingest_requires_admin_and_queues_task(monkeypatch):
             content_type="text/html",
             content_bytes=1234,
             title=label or "Morpheus Docs",
+            pages=[{"final_url": url}],
         ),
     )
     monkeypatch.setattr(admin_api.process_document_task, "delay", lambda *args: _FakeTask())
 
     result = admin_api.run_url_ingest(
-        admin_api.UrlIngestPayload(url="https://docs.example.com/guide", label="Morpheus Docs"),
+        admin_api.UrlIngestPayload(
+            url="https://docs.example.com/guide",
+            label="Morpheus Docs",
+            mode="single_page",
+        ),
         x_admin_key="admin",
         x_auth_token="token",
         user_id="user-1",
@@ -176,3 +256,4 @@ def test_run_url_ingest_requires_admin_and_queues_task(monkeypatch):
     assert result["ok"] is True
     assert result["task_id"] == "task-123"
     assert result["filename"] == "Morpheus-Docs.md"
+    assert result["pages"] == [{"final_url": "https://docs.example.com/guide"}]