From a5324d39711e683817c7dd0afe7ce96ec01409e5 Mon Sep 17 00:00:00 2001 From: faligam Date: Mon, 7 Sep 2026 11:16:36 +0200 Subject: [PATCH] Fix research pipeline runtime failures --- API.md | 4 +- HANDOFF.md | 36 +++++++++++- src/nsct/api/rest_research.py | 2 +- src/nsct/crawler/fetcher.py | 3 +- src/nsct/crawler/manager.py | 12 ++-- src/nsct/orchestration/budget.py | 5 +- src/nsct/orchestration/orchestrator.py | 57 ++++++++++++++---- tests/test_backend_pipeline_repairs.py | 80 +++++++++++++++++++++++++- tests/test_crawler.py | 9 ++- 9 files changed, 184 insertions(+), 24 deletions(-) diff --git a/API.md b/API.md index 1fe7d03..b7f9037 100644 --- a/API.md +++ b/API.md @@ -148,7 +148,7 @@ Die Antwort kommt sofort — das eigentliche Ergebnis ist über | Tiefe | max_search_queries | max_sources | max_llm_requests | max_duration | max_context | |-------|--------------------|-------------|-------------------|--------------|-------------| -| `quick` | 10 | 15 | 15 | 120s | 16k | +| `quick` | 10 | 15 | 15 | 300s | 16k | | `normal` | 20 | 30 | 30 | 300s | 24k | | `deep` | 40 | 60 | 60 | 600s | 32k | @@ -411,4 +411,4 @@ Alle Fehler folgen einem konsistenten Format: ## Authentifizierung Stage 22: Keine Authentifizierung. Dies kann in späteren Stages -hinzugefügt werden (API-Key, OAuth, etc.). \ No newline at end of file +hinzugefügt werden (API-Key, OAuth, etc.). diff --git a/HANDOFF.md b/HANDOFF.md index 64a6eb7..cca2975 100644 --- a/HANDOFF.md +++ b/HANDOFF.md @@ -5,7 +5,41 @@ --- -## Aktueller Stand — 2026-09-06 +## Aktueller Stand — 2026-09-07 + +### Browser-E2E: Such- und Crawler-Pipeline repariert — 2026-09-07 + +Während des Browser-E2E-Tests traten drei aufeinanderfolgende Backend-Fehler +auf. Alle sind lokal repariert, durch gezielte Tests abgedeckt und mit +`docker compose up -d --build nsct-api` deployed: + +1. SearXNG liefert `NormalizedResult`-Pydantic-Objekte. Der Orchestrator + behandelte optionale leere Felder fälschlich als Dictionary und rief `.get()` + auf. Die Ergebnisnormalisierung unterscheidet nun sauber Modelle, Mappings + und ungültige Werte. +2. `AsyncFetcher` konfigurierte bei `httpx.Timeout` connect/read/write, aber + keinen `pool`-Timeout. Der Crawler konnte deshalb nicht initialisiert + werden. `pool` ist nun konfigurierbar über `NSCT_CRAWLER_POOL_TIMEOUT` + (Default 10 Sekunden). +3. Die Budgetierung buchte für jede Ergebnis-URL vor dem Abruf pauschal 500 KB. + Ein Quick-Run mit 22 URLs überschritt so fälschlich das 500-KB-Budget, obwohl + die tatsächlichen Downloads deutlich kleiner waren. Abrufe werden jetzt auf + verbleibende Quellen- und Bytebudgets begrenzt; die Buchung erfolgt nach + tatsächlich abgerufenen Response-Bytes. Budgetobergrenzen sind inklusiv. + Das Quick-Zeitlimit beträgt nun 300 Sekunden (API-Dokumentation angepasst). + +Validierung: 63 gezielte Tests (`test_backend_pipeline_repairs.py`, +`test_crawler.py`, `test_search.py`) erfolgreich; Docker-Container und +`/health` erfolgreich. Die komplette REST-Suite startet echte externe +Background-Retries und wurde deshalb nicht abgewartet. Pytest ist lokal in der +ignorierten `.nsct-test-venv` installiert. + +Wichtig für den nächsten Browser-Test: Research-Runs werden derzeit nur in +`_research_store` im Arbeitsspeicher gehalten. Jeder API-Neubau/-neustart +entfernt bisherige Run-IDs; die alten Runs liefern danach erwartungsgemäß 404. +Persistente Research-Runs sind ein offener Production-Readiness-Punkt. Vor dem +erneuten Test zuerst eine neue Recherche starten, nicht eine alte Detail-URL +wiederverwenden. ### Laufende Deployments diff --git a/src/nsct/api/rest_research.py b/src/nsct/api/rest_research.py index 179fae1..d992cb9 100644 --- a/src/nsct/api/rest_research.py +++ b/src/nsct/api/rest_research.py @@ -66,7 +66,7 @@ DEPTH_CONFIGS: dict[str, DepthConfig] = { max_pages_per_domain=2, max_total_download_bytes=500_000, max_llm_requests=15, - max_research_duration_seconds=120, + max_research_duration_seconds=300, max_context_per_llm_call=16_000, ), "normal": DepthConfig( diff --git a/src/nsct/crawler/fetcher.py b/src/nsct/crawler/fetcher.py index 4d5e70e..d59732f 100644 --- a/src/nsct/crawler/fetcher.py +++ b/src/nsct/crawler/fetcher.py @@ -98,6 +98,7 @@ class AsyncFetcher: connect=float(os.environ.get("NSCT_CRAWLER_CONNECT_TIMEOUT", "10")), read=float(os.environ.get("NSCT_CRAWLER_READ_TIMEOUT", "30")), write=float(os.environ.get("NSCT_CRAWLER_WRITE_TIMEOUT", "10")), + pool=float(os.environ.get("NSCT_CRAWLER_POOL_TIMEOUT", "10")), ) self._client = httpx.AsyncClient( timeout=timeout_config, @@ -304,4 +305,4 @@ class AsyncFetcher: @staticmethod def compute_content_hash(content: bytes) -> str: """Compute SHA256 hash of content.""" - return hashlib.sha256(content).hexdigest() \ No newline at end of file + return hashlib.sha256(content).hexdigest() diff --git a/src/nsct/crawler/manager.py b/src/nsct/crawler/manager.py index 9f5864c..e28e62c 100644 --- a/src/nsct/crawler/manager.py +++ b/src/nsct/crawler/manager.py @@ -36,7 +36,7 @@ class CrawlerManager: # Public API # ------------------------------------------------------------------ - async def fetch_and_extract(self, url: str) -> NormalizedDocument: + async def fetch_and_extract(self, url: str, max_size: int = 0) -> NormalizedDocument: """Fetch a single URL and extract main content. Pipeline: @@ -73,7 +73,7 @@ class CrawlerManager: logger.warning("DNS check warning for %s: %s", url, exc) # 3. HTTP fetch - result = await self.fetcher.fetch(url) + result = await self.fetcher.fetch(url, max_size=max_size) if result.status in (FetchStatus.BLOCKED, FetchStatus.ERROR, FetchStatus.TIMEOUT, FetchStatus.SIZE_LIMIT_EXCEEDED, FetchStatus.CONTENT_TYPE_BLOCKED): @@ -86,7 +86,7 @@ class CrawlerManager: # 4. Content-based extraction text = "" extraction_tool = "" - metadata: dict[str, Any] = {} + metadata: dict[str, Any] = {"download_bytes": len(content)} links: list[str] = [] title = "" @@ -131,7 +131,7 @@ class CrawlerManager: return doc - async def fetch_and_extract_many(self, urls: list[str]) -> list[NormalizedDocument]: + async def fetch_and_extract_many(self, urls: list[str], max_size: int = 0) -> list[NormalizedDocument]: """Fetch and extract multiple URLs in parallel. Each URL is processed independently — failures do not block others. @@ -143,7 +143,7 @@ class CrawlerManager: List of NormalizedDocument, one per input URL (in order). Failed URLs produce error documents with error info in metadata. """ - tasks = [self.fetch_and_extract(url) for url in urls] + tasks = [self.fetch_and_extract(url, max_size=max_size) for url in urls] results = await asyncio.gather(*tasks, return_exceptions=True) final_docs: list[NormalizedDocument] = [] @@ -262,4 +262,4 @@ class CrawlerManager: if lang_match: meta["language"] = lang_match.group(1) - return meta \ No newline at end of file + return meta diff --git a/src/nsct/orchestration/budget.py b/src/nsct/orchestration/budget.py index ee60189..22c4f0c 100644 --- a/src/nsct/orchestration/budget.py +++ b/src/nsct/orchestration/budget.py @@ -160,7 +160,10 @@ class BudgetTracker: counter_name = field.replace("max_", "", 1) limit = getattr(self._config, field) usage = self._counters.get(counter_name, 0) - if limit > 0 and usage >= limit: + # Limits are inclusive: exactly max_sources or exactly the byte + # budget is permitted; only use beyond the configured maximum + # exhausts the budget. + if limit > 0 and usage > limit: exceeded.append(field) logger.warning( "Budget limit exceeded: %s (%.2f / %s)", diff --git a/src/nsct/orchestration/orchestrator.py b/src/nsct/orchestration/orchestrator.py index 7e05fb6..cdd6dc3 100644 --- a/src/nsct/orchestration/orchestrator.py +++ b/src/nsct/orchestration/orchestrator.py @@ -11,6 +11,7 @@ import asyncio import json import logging import time +from collections.abc import Mapping from datetime import datetime, timezone from typing import Any from uuid import UUID, uuid4 @@ -603,14 +604,30 @@ class ResearchOrchestrator: seen_urls = set() self._search_results = [] for r in all_results: - url = r.url if hasattr(r, "url") else r.get("url", "") + # SearchProvider returns NormalizedResult instances. Keep the + # mapping branch for test doubles and legacy providers, but do + # not call ``.get()`` on a Pydantic model when an optional + # field such as ``snippet`` is empty. + if hasattr(r, "url"): + url = r.url + title = r.title + snippet = r.snippet + provider = r.provider + elif isinstance(r, Mapping): + url = r.get("url", "") + title = r.get("title", "") + snippet = r.get("snippet", "") + provider = r.get("provider", "") + else: + logger.warning("Ignoring unsupported search result type: %s", type(r).__name__) + continue if url and url not in seen_urls: seen_urls.add(url) result_dict = { "url": url, - "title": getattr(r, "title", "") or r.get("title", ""), - "snippet": getattr(r, "snippet", "") or r.get("snippet", ""), - "provider": getattr(r, "provider", "") or r.get("provider", ""), + "title": title, + "snippet": snippet, + "provider": provider, } self._search_results.append(result_dict) @@ -635,11 +652,31 @@ class ResearchOrchestrator: crawler = self._get_crawler() urls = [r["url"] for r in self._search_results] - # Budget: Abschätzung der Download-Größe - self._budget_tracker.increment_download_bytes(len(urls) * 500_000) # ~500KB pro Seite + # Restrict the batch before fetching. Charging each URL a + # fictitious 500 KB exhausted a quick run before any bytes had + # actually been downloaded. + usage = self._budget_tracker.get_usage() + remaining_sources = self._budget_config.max_sources - int( + usage["max_sources"]["usage"] + ) + urls = urls[:max(0, remaining_sources)] + if not urls: + return { + "success": False, + "error": "Source budget exhausted before fetching", + "data": {}, + } + + remaining_download_bytes = self._budget_config.max_total_download_bytes - int( + usage["max_total_download_bytes"]["usage"] + ) + max_size_per_url = 0 + if self._budget_config.max_total_download_bytes > 0: + max_size_per_url = max(1, remaining_download_bytes // len(urls)) + self._budget_tracker.increment_sources(len(urls)) - docs = await crawler.fetch_and_extract_many(urls) + docs = await crawler.fetch_and_extract_many(urls, max_size=max_size_per_url) self._budget_tracker.record_time_elapsed() # In sources-Dicts umwandeln @@ -657,9 +694,9 @@ class ResearchOrchestrator: } self._sources.append(src) - # Track bytes - content_len = len(doc.text or "") - self._budget_tracker.increment_download_bytes(content_len) + # Account for response bytes, not only extracted text. + downloaded_bytes = int(doc.metadata.get("download_bytes", len(doc.text.encode("utf-8")))) + self._budget_tracker.increment_download_bytes(downloaded_bytes) successful = [s for s in self._sources if not s.get("error")] metrics.increment(C_SOURCES_FETCHED_TOTAL, len(self._sources)) diff --git a/tests/test_backend_pipeline_repairs.py b/tests/test_backend_pipeline_repairs.py index 2fd522c..f56e2ff 100644 --- a/tests/test_backend_pipeline_repairs.py +++ b/tests/test_backend_pipeline_repairs.py @@ -6,14 +6,17 @@ import asyncio from uuid import uuid4 import httpx -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, Mock from nsct.config import AppSettings, AudioConfig, DatabaseConfig, LLMConfig, VisionConfig from nsct.orchestration.orchestrator import ResearchOrchestrator +from nsct.orchestration.budget import BudgetExhaustedError, BudgetTracker, HardBudgetConfig from nsct.orchestration.state import ResearchRunState +from nsct.providers.abstract import NormalizedResult from nsct.providers.llm import _LLMProviderImpl from nsct.providers.metrics import ProviderMetrics from nsct.providers.searxng import SearXNGProvider +from nsct.crawler.normalize import NormalizedDocument def _config(*, searxng_base_url: str | None = "http://searxng:8080") -> AppSettings: @@ -71,6 +74,81 @@ def test_orchestrator_registers_searxng_provider() -> None: assert isinstance(provider, SearXNGProvider) +def test_searching_accepts_normalized_result_with_empty_snippet() -> None: + """A Pydantic result must not be treated as a mapping during conversion.""" + async def run() -> None: + orchestrator = ResearchOrchestrator(_config(), uuid4(), "test") + result = NormalizedResult.from_raw( + provider_name="searxng", + title="Example", + url="https://example.org/article", + snippet="", + ) + multi_search = Mock() + multi_search.enabled_providers.return_value = ["searxng"] + multi_search.search = AsyncMock(return_value=[result]) + orchestrator._multi_search = multi_search + + response = await orchestrator._step_searching() + + assert response["success"] is True + assert response["url_count"] == 1 + assert orchestrator._search_results == [{ + "url": "https://example.org/article", + "title": "Example", + "snippet": "", + "provider": "searxng", + }] + + asyncio.run(run()) + + +def test_fetching_uses_actual_bytes_and_respects_remaining_source_budget() -> None: + async def run() -> None: + budget = HardBudgetConfig(max_sources=2, max_total_download_bytes=1_000) + orchestrator = ResearchOrchestrator(_config(), uuid4(), "test", budget_config=budget) + orchestrator._search_results = [ + {"url": "https://example.org/one"}, + {"url": "https://example.org/two"}, + {"url": "https://example.org/three"}, + ] + doc_one = NormalizedDocument.from_text( + "https://example.org/one", "one", metadata={"download_bytes": 120} + ) + doc_two = NormalizedDocument.from_text( + "https://example.org/two", "two", metadata={"download_bytes": 180} + ) + crawler = Mock() + crawler.fetch_and_extract_many = AsyncMock(return_value=[doc_one, doc_two]) + orchestrator._crawler = crawler + + response = await orchestrator._step_fetching() + + assert response["success"] is True + crawler.fetch_and_extract_many.assert_awaited_once_with( + ["https://example.org/one", "https://example.org/two"], max_size=500 + ) + usage = orchestrator.budget_tracker.get_usage() + assert usage["max_sources"]["usage"] == 2 + assert usage["max_total_download_bytes"]["usage"] == 300 + + asyncio.run(run()) + + +def test_hard_budget_allows_exactly_the_configured_limit() -> None: + tracker = BudgetTracker(HardBudgetConfig(max_sources=2)) + tracker.increment_sources(2) + tracker.check_budget() + + tracker.increment_sources() + try: + tracker.check_budget() + except BudgetExhaustedError as exc: + assert exc.exhausted_limits == ["max_sources"] + else: + raise AssertionError("Source usage above the limit must exhaust the budget") + + def test_pipeline_executes_extracting_between_fetching_and_analyzing() -> None: async def run() -> None: orchestrator = ResearchOrchestrator(_config(), uuid4(), "test") diff --git a/tests/test_crawler.py b/tests/test_crawler.py index f47789d..19b5f7d 100644 --- a/tests/test_crawler.py +++ b/tests/test_crawler.py @@ -176,6 +176,13 @@ def test_url_dedup_no_duplicate() -> None: # --------------------------------------------------------------------------- +async def test_fetcher_configures_all_httpx_timeout_phases() -> None: + """httpx requires a pool timeout when the other phases are explicit.""" + fetcher = AsyncFetcher() + assert fetcher._client.timeout.pool == 10.0 + await fetcher.close() + + async def test_fetcher_blocked_url() -> None: """Fetcher must return BLOCKED for SSRF-blocked URLs.""" fetcher = AsyncFetcher() @@ -266,4 +273,4 @@ def test_normalized_document_hash_matches() -> None: text="test content for hash", ) expected = hashlib.sha256(b"test content for hash").hexdigest() - assert doc.content_hash == expected \ No newline at end of file + assert doc.content_hash == expected