610c716a97
Connect source-independence analysis and Stage 8 evidence scoring to the research orchestrator so completed runs retain a transparent score for every extracted claim. Forward those scores into Stage 9's evidence package, allowing synthesis to reason from actual provenance and exposing them through the research API. Replace the previously technical-only synthesis fallback with the report contract consumed by the UI: a summary, source statistics, methodology, and provenance-preserving uncertain findings. Add regression coverage for score generation, score hand-off to Stage 9, report methodology, and visible fallback findings; record the healthy container deployment in the hand-off.
412 lines
16 KiB
Python
412 lines
16 KiB
Python
"""Regression tests for the deployed backend research pipeline fixes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from uuid import uuid4
|
|
|
|
import httpx
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
|
|
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, StateMachine
|
|
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
|
|
from nsct.api.rest_research import DEPTH_CONFIGS
|
|
from nsct.models.claim import Claim, ClaimType
|
|
from nsct.stages.stage9_synthesis import _build_evidence_package_json
|
|
|
|
|
|
def _config(*, searxng_base_url: str | None = "http://searxng:8080") -> AppSettings:
|
|
return AppSettings(
|
|
llm=LLMConfig(base_url="http://llm.example/v1", model="test-model"),
|
|
vision=VisionConfig(base_url="http://vision.example/v1", model="vision"),
|
|
audio=AudioConfig(base_url="http://audio.example/v1"),
|
|
postgres=DatabaseConfig(url="sqlite+aiosqlite://"),
|
|
searxng_base_url=searxng_base_url,
|
|
)
|
|
|
|
|
|
def test_llm_base_url_does_not_duplicate_existing_v1() -> None:
|
|
provider = _LLMProviderImpl(_config(), ProviderMetrics())
|
|
provider._api_key = "test-key"
|
|
client = provider._create_client()
|
|
try:
|
|
assert str(client.base_url) == "http://llm.example/v1/"
|
|
finally:
|
|
asyncio.run(client.close())
|
|
|
|
|
|
def test_searxng_provider_normalizes_json_results() -> None:
|
|
async def run() -> None:
|
|
provider = SearXNGProvider("http://searxng:8080")
|
|
provider._client = httpx.AsyncClient(
|
|
transport=httpx.MockTransport(
|
|
lambda request: httpx.Response(
|
|
200,
|
|
json={
|
|
"results": [
|
|
{"url": "https://example.org/a", "title": "A", "content": "Snippet"},
|
|
{"url": "mailto:invalid@example.org", "title": "Invalid"},
|
|
]
|
|
},
|
|
)
|
|
)
|
|
)
|
|
try:
|
|
results = await provider.search("test")
|
|
assert len(results) == 1
|
|
assert results[0].provider == "searxng"
|
|
assert results[0].url == "https://example.org/a"
|
|
assert results[0].snippet == "Snippet"
|
|
assert provider._client is not None
|
|
finally:
|
|
await provider._client.aclose()
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_searxng_provider_sends_configured_engine_allow_list() -> None:
|
|
async def run() -> None:
|
|
seen_request: httpx.Request | None = None
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
nonlocal seen_request
|
|
seen_request = request
|
|
return httpx.Response(200, json={"results": []})
|
|
|
|
provider = SearXNGProvider("http://searxng:8080", engines=("bing", "yahoo"))
|
|
provider._client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
|
try:
|
|
await provider.search("test")
|
|
assert seen_request is not None
|
|
assert seen_request.url.params["engines"] == "bing,yahoo"
|
|
finally:
|
|
await provider._client.aclose()
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_orchestrator_registers_searxng_provider() -> None:
|
|
orchestrator = ResearchOrchestrator(_config(), uuid4(), "test")
|
|
provider = orchestrator._get_multi_search().get_provider("searxng")
|
|
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_searching_with_no_usable_results_fails_instead_of_completing_empty() -> None:
|
|
"""Provider-wide empty results must not become an empty completed report."""
|
|
async def run() -> None:
|
|
orchestrator = ResearchOrchestrator(_config(), uuid4(), "test")
|
|
multi_search = Mock()
|
|
multi_search.enabled_providers.return_value = ["searxng"]
|
|
multi_search.search = AsyncMock(return_value=[])
|
|
orchestrator._multi_search = multi_search
|
|
|
|
response = await orchestrator._step_searching()
|
|
|
|
assert response["success"] is False
|
|
assert response["url_count"] == 0
|
|
assert "no usable URLs" in response["error"]
|
|
assert orchestrator._search_results == []
|
|
|
|
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_time_tracking_records_only_each_new_elapsed_interval() -> None:
|
|
with patch("nsct.orchestration.budget.time.monotonic", side_effect=[100.0, 110.0, 125.0]):
|
|
tracker = BudgetTracker(HardBudgetConfig())
|
|
tracker.record_time_elapsed()
|
|
tracker.record_time_elapsed()
|
|
|
|
usage = tracker.get_usage()
|
|
assert usage["max_research_duration_seconds"]["usage"] == 25.0
|
|
|
|
|
|
def test_depth_time_budgets_allow_local_35b_inference() -> None:
|
|
assert DEPTH_CONFIGS["quick"].max_research_duration_seconds == 1_800
|
|
assert DEPTH_CONFIGS["normal"].max_research_duration_seconds == 3_600
|
|
assert DEPTH_CONFIGS["deep"].max_research_duration_seconds == 7_200
|
|
|
|
|
|
def test_synthesis_package_keeps_stage5_uuid_claim_provenance() -> None:
|
|
"""Stage 9 must accept the actual, UUID-bearing Stage 5 claim model."""
|
|
source_id = uuid4()
|
|
claim = Claim(
|
|
research_run_id=uuid4(),
|
|
source_id=source_id,
|
|
claim_text="Eine überprüfbare Behauptung.",
|
|
evidence_span="Die belegende Passage.",
|
|
claim_type=ClaimType.CLAIM,
|
|
source_url="https://example.org/source",
|
|
)
|
|
|
|
package = json.loads(_build_evidence_package_json([claim.model_dump()], [], "Thema"))
|
|
|
|
assert package["evidence"][0]["claim_id"] == str(claim.id)
|
|
assert package["evidence"][0]["source_id"] == str(source_id)
|
|
assert package["sources"][0]["source_id"] == str(source_id)
|
|
|
|
|
|
def test_orchestrator_synthesis_returns_a_report_for_uuid_claims() -> None:
|
|
"""The production hand-off to Stage 9 must not trigger UUID JSON errors."""
|
|
async def run() -> None:
|
|
orchestrator = ResearchOrchestrator(_config(), uuid4(), "Testthema")
|
|
await orchestrator.start()
|
|
orchestrator._state_machine = StateMachine(ResearchRunState.COMPARING)
|
|
orchestrator._claims = [
|
|
Claim(
|
|
research_run_id=uuid4(),
|
|
source_id=uuid4(),
|
|
claim_text="Eine überprüfbare Behauptung.",
|
|
evidence_span="Die belegende Passage.",
|
|
claim_type=ClaimType.CLAIM,
|
|
source_url="https://example.org/source",
|
|
)
|
|
]
|
|
provider = Mock()
|
|
provider.complete = AsyncMock(return_value=json.dumps({
|
|
"summary": "Neutraler Bericht.",
|
|
"confident_findings": [],
|
|
"uncertain_areas": [],
|
|
"contradictions": [],
|
|
}))
|
|
orchestrator._llm_provider = provider
|
|
orchestrator._comparison_data = {
|
|
"evidence_scores": {
|
|
str(orchestrator._claims[0].id): {
|
|
"claim_id": str(orchestrator._claims[0].id),
|
|
"evidence_type": "direct_observation",
|
|
"source_independence_score": 1.0,
|
|
"cross_source_support": 0.0,
|
|
"contradiction_level": 1.0,
|
|
"evidence_directness": 1.0,
|
|
"date_relevance_score": 0.5,
|
|
"primary_source_proximity": 0.3,
|
|
"relation_links": [],
|
|
}
|
|
}
|
|
}
|
|
|
|
response = await orchestrator._step_synthesizing()
|
|
|
|
assert response["success"] is True
|
|
assert response["report"]["summary"] == "Neutraler Bericht."
|
|
assert response["report"]["methodology"]
|
|
assert '"evidence_type": "direct_observation"' in provider.complete.await_args.kwargs["messages"][1]["content"]
|
|
provider.complete.assert_awaited_once()
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_comparing_creates_evidence_scores_for_extracted_claims() -> None:
|
|
async def run() -> None:
|
|
orchestrator = ResearchOrchestrator(_config(), uuid4(), "Testthema")
|
|
await orchestrator.start()
|
|
orchestrator._state_machine = StateMachine(ResearchRunState.ANALYZING)
|
|
claim = Claim(
|
|
research_run_id=uuid4(),
|
|
source_id=uuid4(),
|
|
claim_text="Die Untersuchung beobachtete einen Anstieg um 20 Prozent.",
|
|
evidence_span="beobachtete einen Anstieg um 20 Prozent",
|
|
claim_type=ClaimType.FACT,
|
|
source_url="https://example.org/source",
|
|
)
|
|
orchestrator._claims = [claim]
|
|
orchestrator._source_independence_data = {
|
|
str(claim.source_id): {"independence_score": 1.0}
|
|
}
|
|
|
|
response = await orchestrator._step_comparing()
|
|
|
|
assert response["success"] is True
|
|
assert response["evidence_count"] == 1
|
|
score = orchestrator._comparison_data["evidence_scores"][str(claim.id)]
|
|
assert score["source_independence_score"] == 1.0
|
|
assert score["claim_id"] == str(claim.id)
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_fallback_report_has_visible_findings_and_methodology() -> None:
|
|
orchestrator = ResearchOrchestrator(_config(), uuid4(), "Testthema")
|
|
claim = Claim(
|
|
research_run_id=uuid4(),
|
|
source_id=uuid4(),
|
|
claim_text="Eine überprüfbare Behauptung.",
|
|
evidence_span="Die belegende Passage.",
|
|
claim_type=ClaimType.CLAIM,
|
|
source_url="https://example.org/source",
|
|
)
|
|
orchestrator._claims = [claim]
|
|
|
|
report = orchestrator._get_report()
|
|
|
|
assert report["methodology"]
|
|
assert report["summary"]
|
|
assert report["uncertain_areas"][0]["claim_text"] == claim.claim_text
|
|
|
|
|
|
def test_start_does_not_charge_a_phantom_planner_request() -> None:
|
|
async def run() -> None:
|
|
orchestrator = ResearchOrchestrator(_config(), uuid4(), "test")
|
|
|
|
await orchestrator.start()
|
|
|
|
usage = orchestrator.budget_tracker.get_usage()
|
|
assert usage["max_llm_requests"]["usage"] == 0
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_claim_extraction_reserves_synthesis_budget_and_skips_failed_sources() -> None:
|
|
async def run() -> None:
|
|
budget = HardBudgetConfig(max_sources=15, max_llm_requests=15)
|
|
orchestrator = ResearchOrchestrator(_config(), uuid4(), "test", budget_config=budget)
|
|
await orchestrator.start()
|
|
orchestrator._state_machine = StateMachine(ResearchRunState.FETCHING)
|
|
orchestrator._budget_tracker.increment_llm_requests() # completed planning call
|
|
orchestrator._llm_provider = Mock()
|
|
orchestrator._sources = [
|
|
{"id": str(index), "url": f"https://example.org/{index}", "error": ""}
|
|
for index in range(14)
|
|
] + [{"id": "failed", "url": "https://example.org/failed", "error": "timeout"}]
|
|
captured: dict[str, object] = {}
|
|
|
|
class FakeExtractor:
|
|
def __init__(self, **kwargs: object) -> None:
|
|
captured["sources"] = kwargs["sources"]
|
|
|
|
async def extract(self) -> list[object]:
|
|
return []
|
|
|
|
with patch("nsct.stages.stage5_extract_claims.Stage5Extractor", FakeExtractor):
|
|
response = await orchestrator._step_extracting()
|
|
|
|
assert response["success"] is True
|
|
assert len(captured["sources"]) == 13
|
|
usage = orchestrator.budget_tracker.get_usage()
|
|
assert usage["max_llm_requests"]["usage"] == 14
|
|
|
|
# The reserved final request reaches, but does not exceed, the quick limit.
|
|
orchestrator._budget_tracker.increment_llm_requests()
|
|
orchestrator.budget_tracker.check_budget()
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_pipeline_executes_extracting_between_fetching_and_analyzing() -> None:
|
|
async def run() -> None:
|
|
orchestrator = ResearchOrchestrator(_config(), uuid4(), "test")
|
|
orchestrator._comparison_data = {}
|
|
for step in ("planning", "searching", "fetching", "extracting", "analyzing", "comparing", "synthesizing"):
|
|
async def handler(target: str = step) -> dict[str, object]:
|
|
orchestrator._transition_to(target)
|
|
return {"success": True, "data": {}}
|
|
|
|
setattr(orchestrator, f"_step_{step}", AsyncMock(side_effect=handler))
|
|
|
|
result = await orchestrator.run()
|
|
|
|
assert result["success"] is True
|
|
assert orchestrator._step_extracting.await_count == 1
|
|
assert orchestrator.state is ResearchRunState.COMPLETED
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_no_search_provider_marks_run_failed_with_reason() -> None:
|
|
async def run() -> None:
|
|
orchestrator = ResearchOrchestrator(_config(searxng_base_url=None), uuid4(), "test")
|
|
await orchestrator.start()
|
|
orchestrator._step_planning = AsyncMock(return_value={"success": True, "data": {}})
|
|
|
|
result = await orchestrator.run()
|
|
|
|
assert result["success"] is False
|
|
assert result["failed_step"] == "searching"
|
|
assert "No search provider" in result["error"]
|
|
assert orchestrator.state is ResearchRunState.FAILED
|
|
|
|
asyncio.run(run())
|