diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/scraping/tools/index.py b/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/scraping/tools/index.py
index 43923dbb9..fb9018fb4 100644
--- a/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/scraping/tools/index.py
+++ b/surfsense_backend/app/agents/chat/multi_agent_chat/subagents/builtins/scraping/tools/index.py
@@ -8,8 +8,7 @@ from langchain_core.tools import BaseTool
 
 from app.agents.chat.multi_agent_chat.shared.permissions import Ruleset
 from app.capabilities.core.access.agent import build_capability_tools
-from app.capabilities.web.discover.definition import WEB_DISCOVER
-from app.capabilities.web.scrape.definition import WEB_SCRAPE
+from app.capabilities.web.crawl.definition import WEB_CRAWL
 from app.capabilities.youtube.comments.definition import YOUTUBE_COMMENTS
 from app.capabilities.youtube.scrape.definition import YOUTUBE_SCRAPE
 
@@ -21,7 +20,7 @@ NAME = "scraping"
 
 RULESET = Ruleset(origin=NAME, rules=[])
 
-_CI_VERBS = [WEB_DISCOVER, WEB_SCRAPE, YOUTUBE_SCRAPE, YOUTUBE_COMMENTS]
+_CI_VERBS = [WEB_CRAWL, YOUTUBE_SCRAPE, YOUTUBE_COMMENTS]
 
 
 def load_tools(
diff --git a/surfsense_backend/app/capabilities/web/__init__.py b/surfsense_backend/app/capabilities/web/__init__.py
index f3710d1cf..c0c4bef17 100644
--- a/surfsense_backend/app/capabilities/web/__init__.py
+++ b/surfsense_backend/app/capabilities/web/__init__.py
@@ -1,6 +1,5 @@
-"""``web.*`` namespace: the generic web scraping + discovery verbs."""
+"""``web.*`` namespace: the unified web content crawler verb."""
 
 from __future__ import annotations
 
-from app.capabilities.web.discover import definition as _discover  # noqa: F401
-from app.capabilities.web.scrape import definition as _scrape  # noqa: F401
+from app.capabilities.web.crawl import definition as _crawl  # noqa: F401
diff --git a/surfsense_backend/app/capabilities/web/crawl/__init__.py b/surfsense_backend/app/capabilities/web/crawl/__init__.py
new file mode 100644
index 000000000..8a13f0bee
--- /dev/null
+++ b/surfsense_backend/app/capabilities/web/crawl/__init__.py
@@ -0,0 +1,3 @@
+"""``web.crawl`` verb: single-URL scrape or same-site spider → one row per page."""
+
+from __future__ import annotations
diff --git a/surfsense_backend/app/capabilities/web/crawl/definition.py b/surfsense_backend/app/capabilities/web/crawl/definition.py
new file mode 100644
index 000000000..de3cef0ca
--- /dev/null
+++ b/surfsense_backend/app/capabilities/web/crawl/definition.py
@@ -0,0 +1,25 @@
+"""``web.crawl`` capability registration."""
+
+from __future__ import annotations
+
+from app.capabilities.core import BillingUnit, Capability, register_capability
+from app.capabilities.web.crawl.executor import build_crawl_executor
+from app.capabilities.web.crawl.schemas import CrawlInput, CrawlOutput
+
+WEB_CRAWL = Capability(
+    name="web.crawl",
+    description=(
+        "Scrape a single web page or crawl a whole website. Give it one or more "
+        "startUrls. Set maxCrawlDepth=0 to fetch just those URLs, or higher to "
+        "also follow the links on each page (depth 1 = the start pages plus the "
+        "pages they link to, and so on) — staying on the same site and stopping "
+        "at maxCrawlPages. Returns one item per fetched page with clean markdown "
+        "content, metadata (title, description), and crawl provenance."
+    ),
+    input_schema=CrawlInput,
+    output_schema=CrawlOutput,
+    executor=build_crawl_executor(),
+    billing_unit=BillingUnit.WEB_CRAWL,
+)
+
+register_capability(WEB_CRAWL)
diff --git a/surfsense_backend/app/capabilities/web/crawl/executor.py b/surfsense_backend/app/capabilities/web/crawl/executor.py
new file mode 100644
index 000000000..45f5b3004
--- /dev/null
+++ b/surfsense_backend/app/capabilities/web/crawl/executor.py
@@ -0,0 +1,65 @@
+"""``web.crawl`` executor: seeds → site spider → one cleaned item per fetched page.
+
+Boundary owned elsewhere: the crawl frontier/fetch live in the proprietary engine
+(``app.proprietary.web_crawler``). This executor only maps the engine's
+``CrawlPage`` list onto the typed ``CrawlOutput`` (status labels, truncation, and
+the captcha telemetry the billing seam reads).
+"""
+
+from __future__ import annotations
+
+from app.capabilities.core import Executor
+from app.capabilities.web.crawl.schemas import (
+    CrawlInput,
+    CrawlItem,
+    CrawlMeta,
+    CrawlOutput,
+)
+from app.proprietary.web_crawler import (
+    CrawlOutcomeStatus,
+    CrawlPage,
+    WebCrawlerConnector,
+    crawl_site,
+)
+
+_STATUS_LABEL: dict[CrawlOutcomeStatus, str] = {
+    CrawlOutcomeStatus.SUCCESS: "success",
+    CrawlOutcomeStatus.EMPTY: "empty",
+    CrawlOutcomeStatus.FAILED: "failed",
+}
+
+
+def build_crawl_executor(engine: WebCrawlerConnector | None = None) -> Executor:
+    """Build the ``web.crawl`` executor, optionally over an injected engine (tests)."""
+    crawler = engine or WebCrawlerConnector()
+
+    async def execute(payload: CrawlInput) -> CrawlOutput:
+        pages = await crawl_site(
+            crawler,
+            payload.startUrls,
+            max_crawl_depth=payload.maxCrawlDepth,
+            max_crawl_pages=payload.maxCrawlPages,
+        )
+        return CrawlOutput(
+            items=[_to_item(page, payload.maxLength) for page in pages],
+            captcha_attempts=sum(page.captcha_attempts for page in pages),
+            captcha_solved=sum(1 for page in pages if page.captcha_solved),
+        )
+
+    return execute
+
+
+def _to_item(page: CrawlPage, max_length: int) -> CrawlItem:
+    content = page.content[:max_length] if page.content is not None else None
+    return CrawlItem(
+        url=page.url,
+        status=_STATUS_LABEL[page.status],
+        crawl=CrawlMeta(
+            loadedUrl=page.loaded_url or page.url,
+            depth=page.depth,
+            referrerUrl=page.referrer,
+        ),
+        markdown=content,
+        metadata=page.metadata,
+        error=page.error,
+    )
diff --git a/surfsense_backend/app/capabilities/web/crawl/schemas.py b/surfsense_backend/app/capabilities/web/crawl/schemas.py
new file mode 100644
index 000000000..38ab0e7f5
--- /dev/null
+++ b/surfsense_backend/app/capabilities/web/crawl/schemas.py
@@ -0,0 +1,115 @@
+# ruff: noqa: N815 - public field names intentionally mirror the Apify camelCase spec
+"""``web.crawl`` I/O contracts.
+
+An Apify *Website Content Crawler*-style surface (see apify.com/apify/website-
+content-crawler and docs.firecrawl.dev): one verb that either scrapes the given
+URLs (``maxCrawlDepth == 0``) or spiders their site (``maxCrawlDepth > 0``),
+bounded by ``maxCrawlPages`` and kept on the seed's site.
+
+Fields are trimmed to what the proprietary engine honors today. Apify knobs the
+engine handles automatically (crawler type, proxy, dynamic-render waits) are
+intentionally omitted, as are features we haven't built (URL globs, output
+formats, click actions, PII handling).
+"""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+MAX_START_URLS = 20
+"""Per-call cap on seed URLs: bounds a synchronous request's fan-out (05)."""
+
+MAX_CRAWL_DEPTH = 5
+"""Deepest link distance a spider will follow from a start URL."""
+
+MAX_CRAWL_PAGES = 200
+"""Hard ceiling on pages fetched per call (protects the wallet and the run)."""
+
+
+class CrawlInput(BaseModel):
+    startUrls: list[str] = Field(
+        min_length=1,
+        max_length=MAX_START_URLS,
+        description=(
+            "Seed URLs to crawl. With maxCrawlDepth=0 only these are fetched; "
+            "with a higher depth they are also the entry points for the spider."
+        ),
+    )
+    maxCrawlDepth: int = Field(
+        default=0,
+        ge=0,
+        le=MAX_CRAWL_DEPTH,
+        description=(
+            "How many link-hops to follow from each start URL. 0 = scrape only "
+            "the start URLs (no spidering); 1 = also their linked pages; etc. "
+            "The spider stays on the start URL's site."
+        ),
+    )
+    maxCrawlPages: int = Field(
+        default=10,
+        ge=1,
+        le=MAX_CRAWL_PAGES,
+        description=(
+            "Maximum number of pages to fetch in total (start URLs included). "
+            "The crawl stops once this many pages have been fetched."
+        ),
+    )
+    maxLength: int = Field(
+        default=50_000,
+        ge=1,
+        description="Maximum characters of cleaned markdown kept per page (truncates beyond).",
+    )
+
+    @property
+    def estimated_units(self) -> int:
+        """Worst-case billable pages for the pre-flight gate (03c)."""
+        if self.maxCrawlDepth == 0:
+            return len(self.startUrls)
+        return self.maxCrawlPages
+
+
+class CrawlMeta(BaseModel):
+    loadedUrl: str = Field(description="The URL actually fetched for this page.")
+    depth: int = Field(
+        description="Link distance from a start URL (0 for a start URL itself)."
+    )
+    referrerUrl: str | None = Field(
+        default=None,
+        description="The page this URL was discovered on (null for start URLs).",
+    )
+
+
+class CrawlItem(BaseModel):
+    url: str = Field(description="The requested URL for this page.")
+    status: Literal["success", "empty", "failed"] = Field(
+        description="success = content returned; empty = fetched but no content; failed = could not fetch."
+    )
+    crawl: CrawlMeta | None = Field(
+        default=None, description="Crawl provenance (loaded URL, depth, referrer)."
+    )
+    markdown: str | None = Field(
+        default=None, description="Cleaned page content as markdown (null unless success)."
+    )
+    metadata: dict[str, str] | None = Field(
+        default=None, description="Page metadata such as title and description."
+    )
+    error: str | None = Field(
+        default=None, description="Failure reason when status is not success."
+    )
+
+
+class CrawlOutput(BaseModel):
+    items: list[CrawlItem] = Field(
+        default_factory=list,
+        description="One item per fetched page, in crawl (BFS) order.",
+    )
+    # Billing-only telemetry; excluded from the wire shape (mirrors web.scrape).
+    captcha_attempts: int = Field(default=0, exclude=True)
+    captcha_solved: int = Field(default=0, exclude=True)
+
+    @property
+    def billable_units(self) -> int:
+        """Successful pages are the metered unit (03c)."""
+        return sum(1 for item in self.items if item.status == "success")
diff --git a/surfsense_backend/app/capabilities/web/discover/__init__.py b/surfsense_backend/app/capabilities/web/discover/__init__.py
deleted file mode 100644
index 37b153535..000000000
--- a/surfsense_backend/app/capabilities/web/discover/__init__.py
+++ /dev/null
@@ -1,3 +0,0 @@
-"""``web.discover`` verb: a query → candidate URLs the agent can feed to ``web.scrape``."""
-
-from __future__ import annotations
diff --git a/surfsense_backend/app/capabilities/web/discover/definition.py b/surfsense_backend/app/capabilities/web/discover/definition.py
deleted file mode 100644
index f672c6bbb..000000000
--- a/surfsense_backend/app/capabilities/web/discover/definition.py
+++ /dev/null
@@ -1,23 +0,0 @@
-"""``web.discover`` capability registration (free — see 04-capabilities open item)."""
-
-from __future__ import annotations
-
-from app.capabilities.core import Capability, register_capability
-from app.capabilities.web.discover.executor import build_discover_executor
-from app.capabilities.web.discover.schemas import DiscoverInput, DiscoverOutput
-
-WEB_DISCOVER = Capability(
-    name="web.discover",
-    description=(
-        "Search the web for a query and return ranked results. Use it to find "
-        "pages when you don't already have exact URLs. Returns a list of hits "
-        "(url, title, snippet, provider); pass the chosen url(s) to web.scrape "
-        "to read their full content."
-    ),
-    input_schema=DiscoverInput,
-    output_schema=DiscoverOutput,
-    executor=build_discover_executor(),
-    billing_unit=None,
-)
-
-register_capability(WEB_DISCOVER)
diff --git a/surfsense_backend/app/capabilities/web/discover/executor.py b/surfsense_backend/app/capabilities/web/discover/executor.py
deleted file mode 100644
index 3660aedcd..000000000
--- a/surfsense_backend/app/capabilities/web/discover/executor.py
+++ /dev/null
@@ -1,42 +0,0 @@
-"""``web.discover`` executor: route a query to the first configured search provider."""
-
-from __future__ import annotations
-
-from collections.abc import Sequence
-
-from app.capabilities.core import Executor
-from app.capabilities.web.discover.providers.base import DiscoverProvider
-from app.capabilities.web.discover.schemas import DiscoverInput, DiscoverOutput
-
-
-class NoDiscoverProviderError(RuntimeError):
-    """Raised when no search provider is configured (no platform key/host set)."""
-
-
-def build_discover_executor(
-    providers: Sequence[DiscoverProvider] | None = None,
-) -> Executor:
-    """Bind the executor to a provider set (defaults to the real env-keyed providers)."""
-    registry = list(providers) if providers is not None else _default_providers()
-
-    async def execute(payload: DiscoverInput) -> DiscoverOutput:
-        provider = next((p for p in registry if p.is_available()), None)
-        if provider is None:
-            raise NoDiscoverProviderError(
-                "web.discover has no configured search provider "
-                "(set a SearXNG host or a Linkup/Baidu key)."
-            )
-        hits = await provider.search(payload.query, payload.top_k)
-        # Enforce the verb's documented cap here, once, for every provider:
-        # some backends (e.g. SearXNG) treat `top_k` as a hint and over-return.
-        return DiscoverOutput(hits=hits[: payload.top_k])
-
-    return execute
-
-
-def _default_providers() -> list[DiscoverProvider]:
-    from app.capabilities.web.discover.providers.baidu import BaiduProvider
-    from app.capabilities.web.discover.providers.linkup import LinkupProvider
-    from app.capabilities.web.discover.providers.searxng import SearxngProvider
-
-    return [SearxngProvider(), LinkupProvider(), BaiduProvider()]
diff --git a/surfsense_backend/app/capabilities/web/discover/providers/__init__.py b/surfsense_backend/app/capabilities/web/discover/providers/__init__.py
deleted file mode 100644
index 68ca2b374..000000000
--- a/surfsense_backend/app/capabilities/web/discover/providers/__init__.py
+++ /dev/null
@@ -1,3 +0,0 @@
-"""``web.discover`` search providers (env-keyed: SearXNG, Linkup, Baidu)."""
-
-from __future__ import annotations
diff --git a/surfsense_backend/app/capabilities/web/discover/providers/baidu.py b/surfsense_backend/app/capabilities/web/discover/providers/baidu.py
deleted file mode 100644
index fc632a9ea..000000000
--- a/surfsense_backend/app/capabilities/web/discover/providers/baidu.py
+++ /dev/null
@@ -1,53 +0,0 @@
-"""Baidu AI Search discover provider (env-keyed via ``BAIDU_API_KEY``)."""
-
-from __future__ import annotations
-
-import httpx
-
-from app.capabilities.web.discover.schemas import DiscoverHit
-from app.config import config
-
-_ENDPOINT = "https://qianfan.baidubce.com/v2/ai_search/chat/completions"
-_SNIPPET_MAX = 300
-
-
-class BaiduProvider:
-    name = "baidu"
-
-    def is_available(self) -> bool:
-        return bool(config.BAIDU_API_KEY)
-
-    async def search(self, query: str, top_k: int) -> list[DiscoverHit]:
-        max_per_type = min(top_k, 20)
-        payload = {
-            "messages": [{"role": "user", "content": query}],
-            "model": config.BAIDU_MODEL,
-            "search_source": config.BAIDU_SEARCH_SOURCE,
-            "resource_type_filter": [{"type": "web", "top_k": max_per_type}],
-            "stream": False,
-        }
-        headers = {
-            "X-Appbuilder-Authorization": f"Bearer {config.BAIDU_API_KEY}",
-            "Content-Type": "application/json",
-        }
-        async with httpx.AsyncClient(timeout=90.0) as client:
-            response = await client.post(_ENDPOINT, headers=headers, json=payload)
-            response.raise_for_status()
-
-        hits: list[DiscoverHit] = []
-        for reference in response.json().get("references", []):
-            url = reference.get("url", "")
-            if not url:
-                continue
-            content = reference.get("content", "") or ""
-            hits.append(
-                DiscoverHit(
-                    url=url,
-                    title=reference.get("title") or url,
-                    snippet=content[:_SNIPPET_MAX] or None,
-                    provider=self.name,
-                )
-            )
-            if len(hits) >= top_k:
-                break
-        return hits
diff --git a/surfsense_backend/app/capabilities/web/discover/providers/base.py b/surfsense_backend/app/capabilities/web/discover/providers/base.py
deleted file mode 100644
index c245072eb..000000000
--- a/surfsense_backend/app/capabilities/web/discover/providers/base.py
+++ /dev/null
@@ -1,22 +0,0 @@
-"""The seam every ``web.discover`` provider implements."""
-
-from __future__ import annotations
-
-from typing import Protocol, runtime_checkable
-
-from app.capabilities.web.discover.schemas import DiscoverHit
-
-
-@runtime_checkable
-class DiscoverProvider(Protocol):
-    """An env-keyed search backend that suggests candidate URLs for a query."""
-
-    name: str
-
-    def is_available(self) -> bool:
-        """True when this provider's platform key/host is configured."""
-        ...
-
-    async def search(self, query: str, top_k: int) -> list[DiscoverHit]:
-        """Return up to ``top_k`` candidate hits for ``query``."""
-        ...
diff --git a/surfsense_backend/app/capabilities/web/discover/providers/linkup.py b/surfsense_backend/app/capabilities/web/discover/providers/linkup.py
deleted file mode 100644
index 77393de08..000000000
--- a/surfsense_backend/app/capabilities/web/discover/providers/linkup.py
+++ /dev/null
@@ -1,40 +0,0 @@
-"""Linkup discover provider (env-keyed via ``LINKUP_API_KEY``)."""
-
-from __future__ import annotations
-
-import asyncio
-
-from linkup import LinkupClient
-
-from app.capabilities.web.discover.schemas import DiscoverHit
-from app.config import config
-
-
-class LinkupProvider:
-    name = "linkup"
-
-    def is_available(self) -> bool:
-        return bool(config.LINKUP_API_KEY)
-
-    async def search(self, query: str, top_k: int) -> list[DiscoverHit]:
-        client = LinkupClient(api_key=config.LINKUP_API_KEY)
-        response = await asyncio.to_thread(
-            client.search, query=query, depth="standard", output_type="searchResults"
-        )
-        hits: list[DiscoverHit] = []
-        for result in getattr(response, "results", None) or []:
-            url = getattr(result, "url", "") or ""
-            if not url:
-                continue
-            content = getattr(result, "content", "") or ""
-            hits.append(
-                DiscoverHit(
-                    url=url,
-                    title=getattr(result, "name", "") or url,
-                    snippet=content or None,
-                    provider=self.name,
-                )
-            )
-            if len(hits) >= top_k:
-                break
-        return hits
diff --git a/surfsense_backend/app/capabilities/web/discover/providers/searxng.py b/surfsense_backend/app/capabilities/web/discover/providers/searxng.py
deleted file mode 100644
index 92489862a..000000000
--- a/surfsense_backend/app/capabilities/web/discover/providers/searxng.py
+++ /dev/null
@@ -1,28 +0,0 @@
-"""SearXNG discover provider, wrapping the platform-env SearXNG search service."""
-
-from __future__ import annotations
-
-from app.capabilities.web.discover.schemas import DiscoverHit
-from app.services import web_search_service
-
-
-class SearxngProvider:
-    """Env-keyed via ``SEARXNG_DEFAULT_HOST`` (the platform SearXNG instance)."""
-
-    name = "searxng"
-
-    def is_available(self) -> bool:
-        return web_search_service.is_available()
-
-    async def search(self, query: str, top_k: int) -> list[DiscoverHit]:
-        result_object, _documents = await web_search_service.search(query, top_k)
-        return [
-            DiscoverHit(
-                url=source["url"],
-                title=source.get("title") or source["url"],
-                snippet=source.get("description") or None,
-                provider=self.name,
-            )
-            for source in result_object.get("sources", [])
-            if source.get("url")
-        ]
diff --git a/surfsense_backend/app/capabilities/web/discover/schemas.py b/surfsense_backend/app/capabilities/web/discover/schemas.py
deleted file mode 100644
index 229454e7e..000000000
--- a/surfsense_backend/app/capabilities/web/discover/schemas.py
+++ /dev/null
@@ -1,29 +0,0 @@
-"""``web.discover`` I/O contracts."""
-
-from __future__ import annotations
-
-from pydantic import BaseModel, Field
-
-
-class DiscoverInput(BaseModel):
-    query: str = Field(
-        description="What to search the web for, phrased in natural language."
-    )
-    top_k: int = Field(
-        default=10, ge=1, le=50, description="Maximum number of results to return (1-50)."
-    )
-
-
-class DiscoverHit(BaseModel):
-    url: str = Field(
-        description="The result's page URL; pass it to web.scrape to read it."
-    )
-    title: str = Field(description="The result's page title.")
-    snippet: str | None = Field(
-        default=None, description="A short extract summarizing the page."
-    )
-    provider: str = Field(description="Which search engine returned this hit.")
-
-
-class DiscoverOutput(BaseModel):
-    hits: list[DiscoverHit] = Field(description="Ranked search results, best first.")
diff --git a/surfsense_backend/app/capabilities/web/scrape/__init__.py b/surfsense_backend/app/capabilities/web/scrape/__init__.py
deleted file mode 100644
index 0a21b156e..000000000
--- a/surfsense_backend/app/capabilities/web/scrape/__init__.py
+++ /dev/null
@@ -1,3 +0,0 @@
-"""``web.scrape`` verb: a URL array → one cleaned row per URL."""
-
-from __future__ import annotations
diff --git a/surfsense_backend/app/capabilities/web/scrape/definition.py b/surfsense_backend/app/capabilities/web/scrape/definition.py
deleted file mode 100644
index f6ccafe02..000000000
--- a/surfsense_backend/app/capabilities/web/scrape/definition.py
+++ /dev/null
@@ -1,23 +0,0 @@
-"""``web.scrape`` capability registration."""
-
-from __future__ import annotations
-
-from app.capabilities.core import BillingUnit, Capability, register_capability
-from app.capabilities.web.scrape.executor import build_scrape_executor
-from app.capabilities.web.scrape.schemas import ScrapeInput, ScrapeOutput
-
-WEB_SCRAPE = Capability(
-    name="web.scrape",
-    description=(
-        "Fetch one or more web pages and return their clean, readable content. "
-        "Give it the exact URLs to read (use web.discover first if you don't have "
-        "them). Returns one row per URL with a status (success/empty/failed), the "
-        "page content, and metadata such as title and description."
-    ),
-    input_schema=ScrapeInput,
-    output_schema=ScrapeOutput,
-    executor=build_scrape_executor(),
-    billing_unit=BillingUnit.WEB_CRAWL,
-)
-
-register_capability(WEB_SCRAPE)
diff --git a/surfsense_backend/app/capabilities/web/scrape/executor.py b/surfsense_backend/app/capabilities/web/scrape/executor.py
deleted file mode 100644
index 8651ff738..000000000
--- a/surfsense_backend/app/capabilities/web/scrape/executor.py
+++ /dev/null
@@ -1,45 +0,0 @@
-"""``web.scrape`` executor: fetch each URL in the array → cleaned rows."""
-
-from __future__ import annotations
-
-from app.capabilities.core import Executor
-from app.capabilities.web.scrape.schemas import ScrapeInput, ScrapeOutput, ScrapeRow
-from app.proprietary.web_crawler import (
-    CrawlOutcome,
-    CrawlOutcomeStatus,
-    WebCrawlerConnector,
-)
-
-
-def build_scrape_executor(engine: WebCrawlerConnector | None = None) -> Executor:
-    """Bind the executor to a fetch engine (defaults to the proprietary crawler)."""
-    engine = engine or WebCrawlerConnector()
-
-    async def execute(payload: ScrapeInput) -> ScrapeOutput:
-        outcomes = [await engine.crawl_url(url) for url in payload.urls]
-        rows = [
-            _to_row(url, outcome, payload.max_length)
-            for url, outcome in zip(payload.urls, outcomes, strict=True)
-        ]
-        return ScrapeOutput(
-            rows=rows,
-            captcha_attempts=sum(o.captcha_attempts for o in outcomes),
-            captcha_solved=sum(1 for o in outcomes if o.captcha_solved),
-        )
-
-    return execute
-
-
-def _to_row(url: str, outcome: CrawlOutcome, max_length: int) -> ScrapeRow:
-    if outcome.status is CrawlOutcomeStatus.SUCCESS and outcome.result:
-        content = outcome.result.get("content")
-        if content is not None:
-            content = content[:max_length]
-        return ScrapeRow(
-            url=url,
-            status="success",
-            content=content,
-            metadata=outcome.result.get("metadata"),
-        )
-    status = "empty" if outcome.status is CrawlOutcomeStatus.EMPTY else "failed"
-    return ScrapeRow(url=url, status=status, error=outcome.error)
diff --git a/surfsense_backend/app/capabilities/web/scrape/schemas.py b/surfsense_backend/app/capabilities/web/scrape/schemas.py
deleted file mode 100644
index 04faaa516..000000000
--- a/surfsense_backend/app/capabilities/web/scrape/schemas.py
+++ /dev/null
@@ -1,67 +0,0 @@
-"""``web.scrape`` I/O contracts."""
-
-from __future__ import annotations
-
-from typing import Literal
-
-from pydantic import BaseModel, Field
-
-MAX_SCRAPE_URLS = 20
-"""Per-call batch cap: bounds a synchronous request's crawl fan-out (05)."""
-
-
-class ScrapeInput(BaseModel):
-    urls: list[str] = Field(
-        min_length=1,
-        max_length=MAX_SCRAPE_URLS,
-        description=(
-            "Full page URLs to fetch and read (1-20), each starting with "
-            "http:// or https://. Pass the exact URLs you want the content of."
-        ),
-    )
-    max_length: int = Field(
-        default=50_000,
-        description=(
-            "Maximum characters of cleaned content returned per page; "
-            "content longer than this is truncated."
-        ),
-    )
-
-    @property
-    def estimated_units(self) -> int:
-        """Worst-case billable crawls for pre-flight: one per requested URL."""
-        return len(self.urls)
-
-
-class ScrapeRow(BaseModel):
-    url: str = Field(description="The requested URL this result is for.")
-    status: Literal["success", "empty", "failed"] = Field(
-        description=(
-            "'success' = content returned; 'empty' = page reached but no "
-            "readable content; 'failed' = could not be fetched (see error)."
-        )
-    )
-    content: str | None = Field(
-        default=None, description="Cleaned, readable page text (present on success)."
-    )
-    metadata: dict[str, str] | None = Field(
-        default=None, description="Page metadata such as title and description."
-    )
-    error: str | None = Field(
-        default=None, description="Why the fetch failed (present on 'failed')."
-    )
-
-
-class ScrapeOutput(BaseModel):
-    rows: list[ScrapeRow] = Field(
-        description="One result per requested URL, in the same order."
-    )
-    # Billing-only telemetry (Phase 3d), excluded from the client-facing result:
-    # captcha solves are metered per attempt, independent of crawl success.
-    captcha_attempts: int = Field(default=0, exclude=True)
-    captcha_solved: int = Field(default=0, exclude=True)
-
-    @property
-    def billable_units(self) -> int:
-        """One billable unit per successful scrape."""
-        return sum(1 for row in self.rows if row.status == "success")
diff --git a/surfsense_backend/tests/unit/capabilities/access/test_rest_router.py b/surfsense_backend/tests/unit/capabilities/access/test_rest_router.py
index b874b8bba..e3bc387e9 100644
--- a/surfsense_backend/tests/unit/capabilities/access/test_rest_router.py
+++ b/surfsense_backend/tests/unit/capabilities/access/test_rest_router.py
@@ -95,8 +95,7 @@ def test_registered_verbs_appear_on_rest():
 
     router = rest.build_capabilities_router()
     paths = {route.path for route in router.routes}
-    assert "/workspaces/{workspace_id}/capabilities/web.scrape" in paths
-    assert "/workspaces/{workspace_id}/capabilities/web.discover" in paths
+    assert "/workspaces/{workspace_id}/capabilities/web.crawl" in paths
 
 
 @pytest.mark.asyncio
diff --git a/surfsense_backend/tests/unit/capabilities/test_billing.py b/surfsense_backend/tests/unit/capabilities/test_billing.py
index c0a74878b..ef5086587 100644
--- a/surfsense_backend/tests/unit/capabilities/test_billing.py
+++ b/surfsense_backend/tests/unit/capabilities/test_billing.py
@@ -14,7 +14,7 @@ import pytest
 import app.capabilities.core.billing as billing
 from app.capabilities.core.billing import charge_capability, gate_capability
 from app.capabilities.core.types import BillingUnit, CapabilityContext
-from app.capabilities.web.scrape.schemas import ScrapeInput, ScrapeOutput, ScrapeRow
+from app.capabilities.web.crawl.schemas import CrawlInput, CrawlItem, CrawlOutput
 from app.config import config
 from app.services.web_crawl_credit_service import InsufficientCreditsError
 
@@ -46,10 +46,10 @@ def _make_session(owner_id, balance_micros):
     return session, fake_user
 
 
-def _output(*statuses: str) -> ScrapeOutput:
-    return ScrapeOutput(
-        rows=[
-            ScrapeRow(url=f"https://{i}.com", status=status)
+def _output(*statuses: str) -> CrawlOutput:
+    return CrawlOutput(
+        items=[
+            CrawlItem(url=f"https://{i}.com", status=status)
             for i, status in enumerate(statuses)
         ]
     )
@@ -92,7 +92,7 @@ async def test_charges_workspace_owner_per_successful_crawl(monkeypatch, record_
     assert kwargs["cost_micros"] == 2000
 
 
-def _output_with_captcha(*statuses: str, attempts: int, solved: int) -> ScrapeOutput:
+def _output_with_captcha(*statuses: str, attempts: int, solved: int) -> CrawlOutput:
     out = _output(*statuses)
     out.captcha_attempts = attempts
     out.captcha_solved = solved
@@ -196,7 +196,7 @@ async def test_gate_blocks_when_worst_case_exceeds_balance(monkeypatch):
 
     with pytest.raises(InsufficientCreditsError):
         await gate_capability(
-            ScrapeInput(urls=["https://a.com", "https://b.com"]),
+            CrawlInput(startUrls=["https://a.com", "https://b.com"]),
             BillingUnit.WEB_CRAWL,
             _ctx(session),
         )
@@ -208,7 +208,7 @@ async def test_gate_passes_when_balance_covers_worst_case(monkeypatch):
     session = _gate_session(_OWNER, balance_micros=100_000)
 
     await gate_capability(
-        ScrapeInput(urls=["https://a.com", "https://b.com"]),
+        CrawlInput(startUrls=["https://a.com", "https://b.com"]),
         BillingUnit.WEB_CRAWL,
         _ctx(session),
     )
@@ -219,7 +219,7 @@ async def test_gate_is_noop_when_disabled(monkeypatch):
     session = _gate_session(_OWNER, balance_micros=0)
 
     await gate_capability(
-        ScrapeInput(urls=["https://a.com"]), BillingUnit.WEB_CRAWL, _ctx(session)
+        CrawlInput(startUrls=["https://a.com"]), BillingUnit.WEB_CRAWL, _ctx(session)
     )
 
 
@@ -233,7 +233,7 @@ async def test_gate_reserves_worst_case_captcha_when_solving_enabled(monkeypatch
 
     with pytest.raises(InsufficientCreditsError):
         await gate_capability(
-            ScrapeInput(urls=["https://a.com"]), BillingUnit.WEB_CRAWL, _ctx(session)
+            CrawlInput(startUrls=["https://a.com"]), BillingUnit.WEB_CRAWL, _ctx(session)
         )
 
 
@@ -245,7 +245,7 @@ async def test_gate_does_not_reserve_captcha_when_solving_disabled(monkeypatch):
 
     # Solving off → attempts can never happen → nothing to reserve → passes.
     await gate_capability(
-        ScrapeInput(urls=["https://a.com"]), BillingUnit.WEB_CRAWL, _ctx(session)
+        CrawlInput(startUrls=["https://a.com"]), BillingUnit.WEB_CRAWL, _ctx(session)
     )
 
 
@@ -253,6 +253,6 @@ async def test_gate_is_noop_for_free_verb(monkeypatch):
     monkeypatch.setattr(config, "WEB_CRAWL_CREDIT_BILLING_ENABLED", True)
     session = _gate_session(_OWNER, balance_micros=0)
 
-    await gate_capability(ScrapeInput(urls=["https://a.com"]), None, _ctx(session))
+    await gate_capability(CrawlInput(startUrls=["https://a.com"]), None, _ctx(session))
 
     session.execute.assert_not_called()
diff --git a/surfsense_backend/tests/unit/capabilities/test_registry.py b/surfsense_backend/tests/unit/capabilities/test_registry.py
index c603c02eb..ff8f6d931 100644
--- a/surfsense_backend/tests/unit/capabilities/test_registry.py
+++ b/surfsense_backend/tests/unit/capabilities/test_registry.py
@@ -9,25 +9,15 @@ from app.capabilities import (
 )
 from app.capabilities.core.store import get_capability
 from app.capabilities.core.types import BillingUnit
-from app.capabilities.web.discover.schemas import DiscoverInput, DiscoverOutput
-from app.capabilities.web.scrape.schemas import ScrapeInput, ScrapeOutput
+from app.capabilities.web.crawl.schemas import CrawlInput, CrawlOutput
 
 pytestmark = pytest.mark.unit
 
 
-def test_web_scrape_is_registered_with_its_schemas_and_billing_unit():
-    cap = get_capability("web.scrape")
+def test_web_crawl_is_registered_with_its_schemas_and_billing_unit():
+    cap = get_capability("web.crawl")
 
-    assert cap.name == "web.scrape"
-    assert cap.input_schema is ScrapeInput
-    assert cap.output_schema is ScrapeOutput
+    assert cap.name == "web.crawl"
+    assert cap.input_schema is CrawlInput
+    assert cap.output_schema is CrawlOutput
     assert cap.billing_unit is BillingUnit.WEB_CRAWL
-
-
-def test_web_discover_is_registered_and_free():
-    cap = get_capability("web.discover")
-
-    assert cap.name == "web.discover"
-    assert cap.input_schema is DiscoverInput
-    assert cap.output_schema is DiscoverOutput
-    assert cap.billing_unit is None
diff --git a/surfsense_backend/tests/unit/capabilities/web/crawl/test_executor.py b/surfsense_backend/tests/unit/capabilities/web/crawl/test_executor.py
new file mode 100644
index 000000000..2d550abbc
--- /dev/null
+++ b/surfsense_backend/tests/unit/capabilities/web/crawl/test_executor.py
@@ -0,0 +1,113 @@
+"""``web.crawl`` executor behavior: CrawlPage list → typed CrawlOutput items.
+
+Boundary mocked: the crawler engine (fake ``crawl_url`` + link graph). NOT
+mocked: the executor's page→item mapping, truncation, and captcha rollup.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from app.capabilities.web.crawl.executor import build_crawl_executor
+from app.capabilities.web.crawl.schemas import CrawlInput, CrawlOutput
+from app.proprietary.web_crawler import CrawlOutcome, CrawlOutcomeStatus
+
+pytestmark = pytest.mark.unit
+
+_SUCCESS = CrawlOutcomeStatus.SUCCESS
+
+
+class _FakeEngine:
+    def __init__(self, graph: dict[str, tuple[CrawlOutcomeStatus, list[str]]]):
+        self._graph = graph
+        self.calls: list[str] = []
+
+    async def crawl_url(self, url: str) -> CrawlOutcome:
+        self.calls.append(url)
+        status, links = self._graph[url]
+        if status is _SUCCESS:
+            return CrawlOutcome(
+                status=_SUCCESS,
+                result={
+                    "content": f"C:{url}",
+                    "metadata": {"title": url},
+                    "links": links,
+                },
+            )
+        return CrawlOutcome(status=status, error="boom")
+
+
+async def test_single_url_depth_zero_returns_one_item() -> None:
+    engine = _FakeEngine({"https://e.com/": (_SUCCESS, ["https://e.com/a"])})
+    execute = build_crawl_executor(engine=engine)
+
+    out = await execute(CrawlInput(startUrls=["https://e.com/"]))
+
+    assert isinstance(out, CrawlOutput)
+    assert len(out.items) == 1
+    item = out.items[0]
+    assert item.url == "https://e.com/"
+    assert item.status == "success"
+    assert item.markdown == "C:https://e.com/"
+    assert item.metadata == {"title": "https://e.com/"}
+    assert item.crawl is not None
+    assert item.crawl.depth == 0
+    assert item.crawl.referrerUrl is None
+
+
+async def test_spider_collects_multiple_pages_with_provenance() -> None:
+    engine = _FakeEngine(
+        {
+            "https://e.com/": (_SUCCESS, ["https://e.com/a"]),
+            "https://e.com/a": (_SUCCESS, []),
+        }
+    )
+    execute = build_crawl_executor(engine=engine)
+
+    out = await execute(
+        CrawlInput(startUrls=["https://e.com/"], maxCrawlDepth=1, maxCrawlPages=10)
+    )
+
+    by_url = {item.url: item for item in out.items}
+    assert set(by_url) == {"https://e.com/", "https://e.com/a"}
+    assert by_url["https://e.com/a"].crawl.referrerUrl == "https://e.com/"
+
+
+async def test_content_is_truncated_to_max_length() -> None:
+    engine = _FakeEngine({"https://e.com/": (_SUCCESS, [])})
+    execute = build_crawl_executor(engine=engine)
+
+    out = await execute(CrawlInput(startUrls=["https://e.com/"], maxLength=3))
+
+    assert out.items[0].markdown == "C:h"
+
+
+async def test_failed_page_has_no_markdown_but_keeps_error() -> None:
+    engine = _FakeEngine({"https://e.com/": (CrawlOutcomeStatus.FAILED, [])})
+    execute = build_crawl_executor(engine=engine)
+
+    out = await execute(CrawlInput(startUrls=["https://e.com/"]))
+
+    item = out.items[0]
+    assert item.status == "failed"
+    assert item.markdown is None
+    assert item.error == "boom"
+
+
+async def test_captcha_telemetry_is_rolled_up_for_billing() -> None:
+    class _CaptchaEngine:
+        async def crawl_url(self, url: str) -> CrawlOutcome:
+            return CrawlOutcome(
+                status=_SUCCESS,
+                result={"content": "ok", "metadata": {}, "links": []},
+                captcha_attempts=2,
+                captcha_solved=True,
+            )
+
+    execute = build_crawl_executor(engine=_CaptchaEngine())
+
+    out = await execute(CrawlInput(startUrls=["https://e.com/"]))
+
+    assert out.captcha_attempts == 2
+    assert out.captcha_solved == 1
+    assert out.billable_units == 1
diff --git a/surfsense_backend/tests/unit/capabilities/web/crawl/test_schemas.py b/surfsense_backend/tests/unit/capabilities/web/crawl/test_schemas.py
new file mode 100644
index 000000000..9e7503851
--- /dev/null
+++ b/surfsense_backend/tests/unit/capabilities/web/crawl/test_schemas.py
@@ -0,0 +1,67 @@
+"""``web.crawl`` I/O contract: camelCase surface, bounds, and billing counters."""
+
+from __future__ import annotations
+
+import pytest
+from pydantic import ValidationError
+
+from app.capabilities.web.crawl.schemas import (
+    CrawlInput,
+    CrawlItem,
+    CrawlMeta,
+    CrawlOutput,
+)
+
+pytestmark = pytest.mark.unit
+
+
+def test_requires_at_least_one_start_url() -> None:
+    with pytest.raises(ValidationError):
+        CrawlInput(startUrls=[])
+
+
+def test_camelcase_fields_and_defaults() -> None:
+    model = CrawlInput(startUrls=["https://e.com"])
+    assert model.startUrls == ["https://e.com"]
+    assert model.maxCrawlDepth == 0
+    assert model.maxCrawlPages == 10
+    assert model.maxLength == 50_000
+
+
+def test_depth_and_page_bounds_are_enforced() -> None:
+    with pytest.raises(ValidationError):
+        CrawlInput(startUrls=["https://e.com"], maxCrawlDepth=-1)
+    with pytest.raises(ValidationError):
+        CrawlInput(startUrls=["https://e.com"], maxCrawlDepth=99)
+    with pytest.raises(ValidationError):
+        CrawlInput(startUrls=["https://e.com"], maxCrawlPages=0)
+
+
+def test_estimated_units_for_single_url_is_seed_count() -> None:
+    model = CrawlInput(startUrls=["https://a.com", "https://b.com"], maxCrawlDepth=0)
+    assert model.estimated_units == 2
+
+
+def test_estimated_units_for_spider_is_max_pages() -> None:
+    model = CrawlInput(
+        startUrls=["https://a.com"], maxCrawlDepth=2, maxCrawlPages=25
+    )
+    assert model.estimated_units == 25
+
+
+def test_billable_units_counts_only_successes() -> None:
+    out = CrawlOutput(
+        items=[
+            CrawlItem(url="a", status="success", crawl=CrawlMeta(loadedUrl="a", depth=0)),
+            CrawlItem(url="b", status="empty", crawl=CrawlMeta(loadedUrl="b", depth=1)),
+            CrawlItem(url="c", status="failed", crawl=CrawlMeta(loadedUrl="c", depth=1)),
+        ]
+    )
+    assert out.billable_units == 1
+
+
+def test_captcha_counters_are_excluded_from_the_wire_shape() -> None:
+    out = CrawlOutput(items=[], captcha_attempts=3, captcha_solved=1)
+    dumped = out.model_dump()
+    assert "captcha_attempts" not in dumped
+    assert "captcha_solved" not in dumped
diff --git a/surfsense_backend/tests/unit/capabilities/web/discover/test_baidu_provider.py b/surfsense_backend/tests/unit/capabilities/web/discover/test_baidu_provider.py
deleted file mode 100644
index dc3035da7..000000000
--- a/surfsense_backend/tests/unit/capabilities/web/discover/test_baidu_provider.py
+++ /dev/null
@@ -1,91 +0,0 @@
-"""BaiduProvider maps the Baidu AI Search references to DiscoverHits, env-keyed.
-
-Boundary mocked: httpx.AsyncClient + config key. NOT mocked: reference→hit mapping.
-"""
-
-from __future__ import annotations
-
-import pytest
-
-import app.capabilities.web.discover.providers.baidu as baidu_module
-from app.capabilities.web.discover.providers.baidu import BaiduProvider
-from app.config import config
-
-pytestmark = pytest.mark.unit
-
-
-class _FakeResponse:
-    def __init__(self, payload):
-        self._payload = payload
-
-    def raise_for_status(self):
-        return None
-
-    def json(self):
-        return self._payload
-
-
-class _FakeAsyncClient:
-    """Minimal async-context httpx stand-in returning a canned payload."""
-
-    payload: dict = {}
-
-    def __init__(self, *args, **kwargs):
-        pass
-
-    async def __aenter__(self):
-        return self
-
-    async def __aexit__(self, *exc):
-        return False
-
-    async def post(self, url, headers=None, json=None):
-        return _FakeResponse(type(self).payload)
-
-
-def _install(monkeypatch, payload):
-    monkeypatch.setattr(config, "BAIDU_API_KEY", "k")
-    _FakeAsyncClient.payload = payload
-    monkeypatch.setattr(baidu_module.httpx, "AsyncClient", _FakeAsyncClient)
-
-
-def test_is_available_reflects_the_env_key(monkeypatch):
-    monkeypatch.setattr(config, "BAIDU_API_KEY", "k")
-    assert BaiduProvider().is_available() is True
-    monkeypatch.setattr(config, "BAIDU_API_KEY", None)
-    assert BaiduProvider().is_available() is False
-
-
-async def test_maps_references_to_hits(monkeypatch):
-    _install(
-        monkeypatch,
-        {
-            "references": [
-                {"title": "Acme", "url": "https://acme.cn", "content": "hello"},
-                {"title": "No URL", "url": "", "content": "skip"},
-            ]
-        },
-    )
-
-    hits = await BaiduProvider().search("acme", top_k=10)
-
-    assert [h.url for h in hits] == ["https://acme.cn"]
-    assert hits[0].title == "Acme"
-    assert hits[0].snippet == "hello"
-    assert hits[0].provider == "baidu"
-
-
-async def test_respects_top_k(monkeypatch):
-    _install(
-        monkeypatch,
-        {
-            "references": [
-                {"title": f"n{i}", "url": f"https://{i}.cn", "content": "c"}
-                for i in range(5)
-            ]
-        },
-    )
-
-    hits = await BaiduProvider().search("q", top_k=2)
-
-    assert len(hits) == 2
diff --git a/surfsense_backend/tests/unit/capabilities/web/discover/test_executor.py b/surfsense_backend/tests/unit/capabilities/web/discover/test_executor.py
deleted file mode 100644
index bfc93a31b..000000000
--- a/surfsense_backend/tests/unit/capabilities/web/discover/test_executor.py
+++ /dev/null
@@ -1,99 +0,0 @@
-"""`web.discover` executor: pick the first configured provider; self-disable when none.
-
-Boundary mocked: the providers (injected fakes). NOT mocked: the executor's
-provider-selection and self-disable behavior.
-"""
-
-from __future__ import annotations
-
-import pytest
-
-from app.capabilities.web.discover.executor import (
-    NoDiscoverProviderError,
-    build_discover_executor,
-)
-from app.capabilities.web.discover.schemas import (
-    DiscoverHit,
-    DiscoverInput,
-    DiscoverOutput,
-)
-
-pytestmark = pytest.mark.unit
-
-
-class _FakeProvider:
-    def __init__(
-        self, name: str, available: bool, hits: list[DiscoverHit] | None = None
-    ):
-        self.name = name
-        self._available = available
-        self._hits = hits or []
-        self.calls: list[tuple[str, int]] = []
-
-    def is_available(self) -> bool:
-        return self._available
-
-    async def search(self, query: str, top_k: int) -> list[DiscoverHit]:
-        self.calls.append((query, top_k))
-        return self._hits
-
-
-def _hit(url: str, provider: str) -> DiscoverHit:
-    return DiscoverHit(url=url, title=url, snippet="s", provider=provider)
-
-
-async def test_uses_the_first_available_provider():
-    first = _FakeProvider(
-        "searxng", available=True, hits=[_hit("https://a.com", "searxng")]
-    )
-    second = _FakeProvider(
-        "linkup", available=True, hits=[_hit("https://b.com", "linkup")]
-    )
-    execute = build_discover_executor(providers=[first, second])
-
-    out = await execute(DiscoverInput(query="acme pricing", top_k=5))
-
-    assert isinstance(out, DiscoverOutput)
-    assert [h.url for h in out.hits] == ["https://a.com"]
-    assert first.calls == [("acme pricing", 5)]
-    assert second.calls == []  # first available short-circuits
-
-
-async def test_skips_unavailable_providers():
-    off = _FakeProvider("searxng", available=False)
-    on = _FakeProvider("linkup", available=True, hits=[_hit("https://b.com", "linkup")])
-    execute = build_discover_executor(providers=[off, on])
-
-    out = await execute(DiscoverInput(query="q"))
-
-    assert [h.provider for h in out.hits] == ["linkup"]
-    assert off.calls == []
-
-
-async def test_self_disables_when_no_provider_is_configured():
-    execute = build_discover_executor(
-        providers=[_FakeProvider("searxng", available=False)]
-    )
-
-    with pytest.raises(NoDiscoverProviderError):
-        await execute(DiscoverInput(query="q"))
-
-
-async def test_caps_hits_to_top_k_when_provider_over_returns():
-    # SearXNG treats `limit` as a hint and can return more rows than asked; the
-    # verb must honor its own documented `top_k` cap regardless of the provider.
-    over = _FakeProvider(
-        "searxng",
-        available=True,
-        hits=[_hit(f"https://{i}.com", "searxng") for i in range(5)],
-    )
-    execute = build_discover_executor(providers=[over])
-
-    out = await execute(DiscoverInput(query="q", top_k=3))
-
-    assert len(out.hits) == 3
-    assert [h.url for h in out.hits] == [
-        "https://0.com",
-        "https://1.com",
-        "https://2.com",
-    ]
diff --git a/surfsense_backend/tests/unit/capabilities/web/discover/test_linkup_provider.py b/surfsense_backend/tests/unit/capabilities/web/discover/test_linkup_provider.py
deleted file mode 100644
index d846e0d94..000000000
--- a/surfsense_backend/tests/unit/capabilities/web/discover/test_linkup_provider.py
+++ /dev/null
@@ -1,75 +0,0 @@
-"""LinkupProvider maps the Linkup SDK results to DiscoverHits, env-keyed.
-
-Boundary mocked: the LinkupClient SDK + config key. NOT mocked: result→hit mapping.
-"""
-
-from __future__ import annotations
-
-import pytest
-
-import app.capabilities.web.discover.providers.linkup as linkup_module
-from app.capabilities.web.discover.providers.linkup import LinkupProvider
-from app.config import config
-
-pytestmark = pytest.mark.unit
-
-
-class _Result:
-    def __init__(self, name, url, content):
-        self.name = name
-        self.url = url
-        self.content = content
-        self.type = "text"
-
-
-class _Response:
-    def __init__(self, results):
-        self.results = results
-
-
-class _FakeClient:
-    def __init__(self, api_key):
-        self.api_key = api_key
-
-    def search(self, *, query, depth, output_type):
-        return _Response(
-            [
-                _Result("Acme", "https://acme.com", "acme home"),
-                _Result("No URL", "", "skip"),
-            ]
-        )
-
-
-def test_is_available_reflects_the_env_key(monkeypatch):
-    monkeypatch.setattr(config, "LINKUP_API_KEY", "k")
-    assert LinkupProvider().is_available() is True
-    monkeypatch.setattr(config, "LINKUP_API_KEY", None)
-    assert LinkupProvider().is_available() is False
-
-
-async def test_maps_results_to_hits(monkeypatch):
-    monkeypatch.setattr(config, "LINKUP_API_KEY", "k")
-    monkeypatch.setattr(linkup_module, "LinkupClient", _FakeClient)
-
-    hits = await LinkupProvider().search("acme", top_k=10)
-
-    assert [h.url for h in hits] == ["https://acme.com"]
-    assert hits[0].title == "Acme"
-    assert hits[0].snippet == "acme home"
-    assert hits[0].provider == "linkup"
-
-
-async def test_respects_top_k(monkeypatch):
-    monkeypatch.setattr(config, "LINKUP_API_KEY", "k")
-
-    class _ManyClient(_FakeClient):
-        def search(self, *, query, depth, output_type):
-            return _Response(
-                [_Result(f"n{i}", f"https://{i}.com", "c") for i in range(5)]
-            )
-
-    monkeypatch.setattr(linkup_module, "LinkupClient", _ManyClient)
-
-    hits = await LinkupProvider().search("q", top_k=2)
-
-    assert len(hits) == 2
diff --git a/surfsense_backend/tests/unit/capabilities/web/discover/test_schemas.py b/surfsense_backend/tests/unit/capabilities/web/discover/test_schemas.py
deleted file mode 100644
index f098014d8..000000000
--- a/surfsense_backend/tests/unit/capabilities/web/discover/test_schemas.py
+++ /dev/null
@@ -1,22 +0,0 @@
-"""``web.discover`` I/O contract bounds."""
-
-from __future__ import annotations
-
-import pytest
-from pydantic import ValidationError
-
-from app.capabilities.web.discover.schemas import DiscoverInput
-
-pytestmark = pytest.mark.unit
-
-
-def test_top_k_defaults_and_accepts_the_valid_range():
-    assert DiscoverInput(query="q").top_k == 10
-    assert DiscoverInput(query="q", top_k=1).top_k == 1
-    assert DiscoverInput(query="q", top_k=50).top_k == 50
-
-
-@pytest.mark.parametrize("bad", [0, -1, 51, 100])
-def test_top_k_outside_1_to_50_is_rejected(bad):
-    with pytest.raises(ValidationError):
-        DiscoverInput(query="q", top_k=bad)
diff --git a/surfsense_backend/tests/unit/capabilities/web/discover/test_searxng_provider.py b/surfsense_backend/tests/unit/capabilities/web/discover/test_searxng_provider.py
deleted file mode 100644
index a184666b3..000000000
--- a/surfsense_backend/tests/unit/capabilities/web/discover/test_searxng_provider.py
+++ /dev/null
@@ -1,58 +0,0 @@
-"""SearxngProvider maps the SearXNG service's sources to DiscoverHits.
-
-Boundary mocked: the web_search_service module. NOT mocked: the source→hit mapping.
-"""
-
-from __future__ import annotations
-
-from unittest.mock import AsyncMock
-
-import pytest
-
-import app.capabilities.web.discover.providers.searxng as searxng_module
-from app.capabilities.web.discover.providers.searxng import SearxngProvider
-
-pytestmark = pytest.mark.unit
-
-
-def _result(sources):
-    return ({"sources": sources}, [])
-
-
-async def test_maps_sources_to_hits(monkeypatch):
-    provider = SearxngProvider()
-    monkeypatch.setattr(
-        searxng_module.web_search_service,
-        "search",
-        AsyncMock(
-            return_value=_result(
-                [
-                    {"title": "Acme", "url": "https://acme.com", "description": "home"},
-                    {
-                        "title": "Docs",
-                        "url": "https://acme.com/docs",
-                        "description": "",
-                    },
-                    {"title": "no url", "url": "", "description": "skip me"},
-                ]
-            )
-        ),
-    )
-
-    hits = await provider.search("acme", top_k=5)
-
-    assert [h.url for h in hits] == ["https://acme.com", "https://acme.com/docs"]
-    assert hits[0].title == "Acme"
-    assert hits[0].snippet == "home"
-    assert hits[1].snippet is None  # empty description normalizes to None
-    assert all(h.provider == "searxng" for h in hits)
-
-
-def test_is_available_reflects_the_service(monkeypatch):
-    provider = SearxngProvider()
-    monkeypatch.setattr(searxng_module.web_search_service, "is_available", lambda: True)
-    assert provider.is_available() is True
-    monkeypatch.setattr(
-        searxng_module.web_search_service, "is_available", lambda: False
-    )
-    assert provider.is_available() is False
diff --git a/surfsense_backend/tests/unit/capabilities/web/scrape/test_executor.py b/surfsense_backend/tests/unit/capabilities/web/scrape/test_executor.py
deleted file mode 100644
index b37a0346c..000000000
--- a/surfsense_backend/tests/unit/capabilities/web/scrape/test_executor.py
+++ /dev/null
@@ -1,141 +0,0 @@
-"""`web.scrape` executor behavior: URLs in → cleaned rows out.
-
-Boundary mocked: the crawler (injected fake). NOT mocked: the executor's own
-CrawlOutcome → ScrapeRow mapping.
-"""
-
-from __future__ import annotations
-
-import pytest
-
-from app.capabilities.web.scrape.executor import build_scrape_executor
-from app.capabilities.web.scrape.schemas import ScrapeInput, ScrapeOutput
-from app.proprietary.web_crawler import CrawlOutcome, CrawlOutcomeStatus
-
-pytestmark = pytest.mark.unit
-
-
-class _FakeCrawler:
-    """Stand-in for WebCrawlerConnector: serves a canned outcome per URL."""
-
-    def __init__(self, outcomes: dict[str, CrawlOutcome]):
-        self._outcomes = outcomes
-        self.calls: list[str] = []
-
-    async def crawl_url(self, url: str) -> CrawlOutcome:
-        self.calls.append(url)
-        return self._outcomes[url]
-
-
-def _success(content: str, metadata: dict[str, str]) -> CrawlOutcome:
-    return CrawlOutcome(
-        status=CrawlOutcomeStatus.SUCCESS,
-        result={
-            "content": content,
-            "metadata": metadata,
-            "crawler_type": "scrapling-static",
-        },
-    )
-
-
-async def test_scrape_returns_one_cleaned_row_for_a_successful_url():
-    url = "https://example.com"
-    crawler = _FakeCrawler({url: _success("# Hello", {"title": "Hello"})})
-    execute = build_scrape_executor(engine=crawler)
-
-    out = await execute(ScrapeInput(urls=[url]))
-
-    assert isinstance(out, ScrapeOutput)
-    assert len(out.rows) == 1
-    row = out.rows[0]
-    assert row.url == url
-    assert row.status == "success"
-    assert row.content == "# Hello"
-    assert row.metadata == {"title": "Hello"}
-
-
-async def test_scrape_returns_one_row_per_url_in_input_order():
-    a, b, c = "https://a.com", "https://b.com", "https://c.com"
-    crawler = _FakeCrawler(
-        {
-            a: _success("A", {"title": "A"}),
-            b: _success("B", {"title": "B"}),
-            c: _success("C", {"title": "C"}),
-        }
-    )
-    execute = build_scrape_executor(engine=crawler)
-
-    out = await execute(ScrapeInput(urls=[a, b, c]))
-
-    assert [row.url for row in out.rows] == [a, b, c]
-    assert [row.content for row in out.rows] == ["A", "B", "C"]
-
-
-async def test_content_longer_than_max_length_is_truncated():
-    url = "https://long.com"
-    crawler = _FakeCrawler({url: _success("A" * 100, {"title": "Long"})})
-    execute = build_scrape_executor(engine=crawler)
-
-    out = await execute(ScrapeInput(urls=[url], max_length=10))
-
-    assert out.rows[0].content == "A" * 10
-
-
-async def test_content_within_max_length_is_untouched():
-    url = "https://short.com"
-    crawler = _FakeCrawler({url: _success("hello", {"title": "Short"})})
-    execute = build_scrape_executor(engine=crawler)
-
-    out = await execute(ScrapeInput(urls=[url], max_length=10))
-
-    assert out.rows[0].content == "hello"
-
-
-async def test_scrape_surfaces_total_captcha_attempts_for_billing():
-    ok, blocked = "https://ok.com", "https://blocked.com"
-    crawler = _FakeCrawler(
-        {
-            ok: CrawlOutcome(
-                status=CrawlOutcomeStatus.SUCCESS,
-                result={"content": "OK", "metadata": {}},
-                captcha_attempts=2,
-                captcha_solved=True,
-            ),
-            blocked: CrawlOutcome(
-                status=CrawlOutcomeStatus.FAILED,
-                error="blocked",
-                captcha_attempts=1,
-                captcha_solved=False,
-            ),
-        }
-    )
-    execute = build_scrape_executor(engine=crawler)
-
-    out = await execute(ScrapeInput(urls=[ok, blocked]))
-
-    # Attempts bill even when the crawl ultimately failed (Phase 3d).
-    assert out.captcha_attempts == 3
-    assert out.captcha_solved == 1
-
-
-async def test_partial_failure_keeps_the_batch_and_labels_each_url():
-    ok, empty, failed = "https://ok.com", "https://empty.com", "https://failed.com"
-    crawler = _FakeCrawler(
-        {
-            ok: _success("OK", {"title": "OK"}),
-            empty: CrawlOutcome(status=CrawlOutcomeStatus.EMPTY, error="no content"),
-            failed: CrawlOutcome(status=CrawlOutcomeStatus.FAILED, error="blocked"),
-        }
-    )
-    execute = build_scrape_executor(engine=crawler)
-
-    out = await execute(ScrapeInput(urls=[ok, empty, failed]))
-
-    by_url = {row.url: row for row in out.rows}
-    assert {u: r.status for u, r in by_url.items()} == {
-        ok: "success",
-        empty: "empty",
-        failed: "failed",
-    }
-    assert by_url[failed].content is None
-    assert by_url[failed].error == "blocked"
diff --git a/surfsense_backend/tests/unit/capabilities/web/scrape/test_schemas.py b/surfsense_backend/tests/unit/capabilities/web/scrape/test_schemas.py
deleted file mode 100644
index 9cc0eef23..000000000
--- a/surfsense_backend/tests/unit/capabilities/web/scrape/test_schemas.py
+++ /dev/null
@@ -1,47 +0,0 @@
-"""ScrapeOutput reports its own billable count; ScrapeInput bounds its batch size."""
-
-from __future__ import annotations
-
-import pytest
-from pydantic import ValidationError
-
-from app.capabilities.web.scrape.schemas import (
-    MAX_SCRAPE_URLS,
-    ScrapeInput,
-    ScrapeOutput,
-    ScrapeRow,
-)
-
-pytestmark = pytest.mark.unit
-
-
-def _output(*statuses: str) -> ScrapeOutput:
-    return ScrapeOutput(
-        rows=[
-            ScrapeRow(url=f"https://{i}.com", status=status)
-            for i, status in enumerate(statuses)
-        ]
-    )
-
-
-def test_billable_units_counts_successful_rows():
-    assert _output("success", "empty", "success", "failed").billable_units == 2
-
-
-def test_billable_units_is_zero_without_successes():
-    assert _output("empty", "failed").billable_units == 0
-
-
-def test_rejects_empty_url_batch():
-    with pytest.raises(ValidationError):
-        ScrapeInput(urls=[])
-
-
-def test_rejects_batch_over_the_cap():
-    with pytest.raises(ValidationError):
-        ScrapeInput(urls=[f"https://{i}.com" for i in range(MAX_SCRAPE_URLS + 1)])
-
-
-def test_accepts_batch_at_the_cap():
-    payload = ScrapeInput(urls=[f"https://{i}.com" for i in range(MAX_SCRAPE_URLS)])
-    assert payload.estimated_units == MAX_SCRAPE_URLS