feat(capabilities): replace web.scrape/web.discover with unified web.crawl

web.crawl scrapes a single URL (maxCrawlDepth=0) or spiders a whole site,
backed by the proprietary site_crawler engine. Rewires the scraping subagent
tools and capability tests onto the new verb.
This commit is contained in:
CREDO23 2026-07-03 10:51:05 +02:00
parent f82fae3973
commit 9fe9c5b71d
31 changed files with 411 additions and 950 deletions

View file

@ -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

View file

@ -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