From 535e9c5dc1de37b6b9b18e871ff70b6f02c7b041 Mon Sep 17 00:00:00 2001 From: retoor Date: Sun, 26 Jul 2026 19:05:21 +0200 Subject: [PATCH] Keep DeepSearch crawling when the Playwright driver cannot start The degradation guard in crawl() wrapped only chromium.launch(), while the driver start sat outside it in the async context manager. A failure to start the driver therefore propagated out of crawl() and failed the whole DeepSearch job, contradicting the warning it logs on that path ("pages will use httpx only"). Start the driver inside the guarded block and stop it in the finally, so an unavailable driver degrades to httpx-only fetching as intended. The crawl loop body is unchanged apart from indentation. --- devplacepy/services/jobs/deepsearch/crawl.py | 173 ++++++++++--------- tests/unit/services/jobs/deepsearch/crawl.py | 25 +++ 2 files changed, 113 insertions(+), 85 deletions(-) diff --git a/devplacepy/services/jobs/deepsearch/crawl.py b/devplacepy/services/jobs/deepsearch/crawl.py index ca963031..74d917f3 100644 --- a/devplacepy/services/jobs/deepsearch/crawl.py +++ b/devplacepy/services/jobs/deepsearch/crawl.py @@ -302,99 +302,102 @@ async def crawl( total = min(len(level_candidates), max_pages) cancelled = False + pw = None browser = None - async with async_playwright() as pw: - try: - browser = await pw.chromium.launch( - headless=True, args=["--no-sandbox", "--disable-dev-shm-usage"] - ) - except Exception as exc: - logger.warning("deepsearch playwright launch failed, pages will use httpx only: %s", exc) + try: + pw = await async_playwright().start() + browser = await pw.chromium.launch( + headless=True, args=["--no-sandbox", "--disable-dev-shm-usage"] + ) + except Exception as exc: + logger.warning("deepsearch playwright launch failed, pages will use httpx only: %s", exc) - try: - for level in range(max(1, depth)): - if cancelled or fetched >= max_pages or not level_candidates: + try: + for level in range(max(1, depth)): + if cancelled or fetched >= max_pages or not level_candidates: + break + next_candidates: list[dict] = [] + for start in range(0, len(level_candidates), CRAWL_CONCURRENCY): + if fetched >= max_pages: break - next_candidates: list[dict] = [] - for start in range(0, len(level_candidates), CRAWL_CONCURRENCY): + if await should_stop(): + emit({"type": "stage", "stage": "cancelled", "message": "Crawl cancelled"}) + cancelled = True + break + batch = level_candidates[start : start + CRAWL_CONCURRENCY][: max_pages - fetched] + for candidate in batch: + emit( + { + "type": "progress", + "done": fetched, + "total": total, + "url": candidate["url"], + "depth": level, + "message": f"Reading {candidate['url']}", + } + ) + if is_cached(candidate["url"]): + emit({"type": "page_cached", "url": candidate["url"], "reason": "seen in a prior run"}) + fetch_start = time.perf_counter() + results = await asyncio.gather( + *(_resolve_candidate(candidate, level, browser) for candidate in batch), + return_exceptions=True, + ) + elapsed_ms = int((time.perf_counter() - fetch_start) * 1000) + for candidate, page in zip(batch, results): + url = candidate["url"] + if isinstance(page, BaseException): + logger.info("deepsearch fetch crashed for %s: %s", url, page) + page = None + if page is None: + emit( + { + "type": "page_skipped", + "url": url, + "reason": "no readable content", + "elapsed_ms": elapsed_ms, + } + ) + continue if fetched >= max_pages: break - if await should_stop(): - emit({"type": "stage", "stage": "cancelled", "message": "Crawl cancelled"}) - cancelled = True - break - batch = level_candidates[start : start + CRAWL_CONCURRENCY][: max_pages - fetched] - for candidate in batch: + digest = content_hash(page.text) + if digest in outcome.seen_hashes: emit( { - "type": "progress", - "done": fetched, - "total": total, - "url": candidate["url"], - "depth": level, - "message": f"Reading {candidate['url']}", - } - ) - if is_cached(candidate["url"]): - emit({"type": "page_cached", "url": candidate["url"], "reason": "seen in a prior run"}) - fetch_start = time.perf_counter() - results = await asyncio.gather( - *(_resolve_candidate(candidate, level, browser) for candidate in batch), - return_exceptions=True, - ) - elapsed_ms = int((time.perf_counter() - fetch_start) * 1000) - for candidate, page in zip(batch, results): - url = candidate["url"] - if isinstance(page, BaseException): - logger.info("deepsearch fetch crashed for %s: %s", url, page) - page = None - if page is None: - emit( - { - "type": "page_skipped", - "url": url, - "reason": "no readable content", - "elapsed_ms": elapsed_ms, - } - ) - continue - if fetched >= max_pages: - break - digest = content_hash(page.text) - if digest in outcome.seen_hashes: - emit( - { - "type": "page_duplicate", - "url": url, - "reason": "duplicate content", - "elapsed_ms": elapsed_ms, - } - ) - continue - outcome.seen_hashes.add(digest) - outcome.pages.append(page) - fetched += 1 - emit( - { - "type": "page_loaded", - "url": page.url, - "title": page.title, - "source": page.source, - "depth": level, - "render": page.source == "playwright", + "type": "page_duplicate", + "url": url, + "reason": "duplicate content", "elapsed_ms": elapsed_ms, - "done": fetched, - "total": total, } ) - if level + 1 < depth: - for link in relevant_links(page.links, query, LINKS_PER_PAGE): - if link not in seen_urls: - seen_urls.add(link) - next_candidates.append({"url": link}) - level_candidates = next_candidates - total = min(total + len(next_candidates), max_pages) - finally: - if browser is not None: - await browser.close() + continue + outcome.seen_hashes.add(digest) + outcome.pages.append(page) + fetched += 1 + emit( + { + "type": "page_loaded", + "url": page.url, + "title": page.title, + "source": page.source, + "depth": level, + "render": page.source == "playwright", + "elapsed_ms": elapsed_ms, + "done": fetched, + "total": total, + } + ) + if level + 1 < depth: + for link in relevant_links(page.links, query, LINKS_PER_PAGE): + if link not in seen_urls: + seen_urls.add(link) + next_candidates.append({"url": link}) + level_candidates = next_candidates + total = min(total + len(next_candidates), max_pages) + finally: + if browser is not None: + await browser.close() + if pw is not None: + await pw.stop() return outcome diff --git a/tests/unit/services/jobs/deepsearch/crawl.py b/tests/unit/services/jobs/deepsearch/crawl.py index cbe38353..949b9d11 100644 --- a/tests/unit/services/jobs/deepsearch/crawl.py +++ b/tests/unit/services/jobs/deepsearch/crawl.py @@ -110,3 +110,28 @@ def test_crawl_falls_back_to_snippet_when_fetch_empty(monkeypatch): ) assert len(outcome.pages) == 1 assert outcome.pages[0].source == "search" + + +def test_crawl_degrades_to_httpx_when_the_playwright_driver_cannot_start(monkeypatch): + import playwright.async_api as playwright_api + + class FailingDriver: + async def start(self): + raise RuntimeError("driver missing") + + monkeypatch.setattr(playwright_api, "async_playwright", lambda: FailingDriver()) + + async def fake_fetch(url, depth, browser=None): + return CrawledPage( + url=url, title="T", text="body " * 100, source="httpx", status=200, depth=depth + ) + + monkeypatch.setattr(crawl_module, "fetch_page", fake_fetch) + candidates = [ + {"url": "https://example.com/a", "title": "A", "description": "", "content": ""} + ] + outcome = run_async( + crawl(candidates, 6, lambda f: None, lambda u: False, _no_stop, query="q", depth=1) + ) + assert len(outcome.pages) == 1 + assert outcome.pages[0].source == "httpx"