"""Integration checks use real bench workflows and deterministic responses.""" import importlib.util import json from pathlib import Path import subprocess import sys import pytest from infospace_bench import register_artifact from infospace_bench.workflow import FixtureAssistedGenerationAdapter REPO = Path(__file__).resolve().parents[1] spec = importlib.util.spec_from_file_location("research_bench", REPO / "scripts/research_bench.py") bench = importlib.util.module_from_spec(spec) spec.loader.exec_module(bench) FIXTURES = REPO / "research/bench/fixtures" def initialize(tmp_path): return bench.initialize(FIXTURES / "source.md", tmp_path, "synthetic-probe") def adapter(): return FixtureAssistedGenerationAdapter.from_file(FIXTURES / "responses.yaml") def test_pipeline_carries_source_and_previous_outputs_into_review(tmp_path): root = initialize(tmp_path) delegate = adapter() prompts = {} class Capture: def generate(self, request): prompts[request.stage_id] = request.prompt return delegate.generate(request) result = bench.execute(root, Capture()) assert result["provider_calls_attempted"] == 4 assert result["status"] == "completed" assert "server can reject submission" in prompts["critique"] assert "# Synthetic claims" in prompts["critique"] assert "# Local Draft State" in prompts["critique"] assert "only P01 exists" in prompts["critique"] assert "{{" not in prompts["critique"] assert (root / "output/evaluations/source-review.md").exists() index = (root / "artifacts/index.yaml").read_text() assert "provider: fixture" in index assert result["review_status"] == "unreviewed" assert result["evidence_kind"] == "synthetic" assert not list((root / "artifacts/entities").glob("*.md")) assert not (root / "patterns").exists() with pytest.raises(ValueError, match="already attempted"): bench.execute(root, Capture()) def test_mutated_source_and_multiple_sources_are_rejected(tmp_path): root = initialize(tmp_path) source = root / "artifacts/sources/source.md" original = source.read_bytes() source.write_text("changed") with pytest.raises(ValueError, match="Source changed"): bench.plan(root) source.write_bytes(original) register_artifact(root, artifact_id="source/second.md", path=source, kind="source", title="Second") with pytest.raises(ValueError, match="Exactly one"): bench.plan(root) def test_call_cap_prevents_provider_use(tmp_path): root = initialize(tmp_path) class NeverCall: def generate(self, request): pytest.fail("Provider must not be called") with pytest.raises(ValueError, match="exceed"): bench.execute(root, NeverCall(), max_calls=3) assert not (root / "reports/research-run.json").exists() def test_failure_is_recorded_and_cannot_be_silently_retried(tmp_path): root = initialize(tmp_path) class Fail: def generate(self, request): raise RuntimeError("fixture failure") with pytest.raises(RuntimeError): bench.execute(root, Fail()) receipt = json.loads((root / "reports/research-run.json").read_text()) assert receipt["status"] == "failed" assert receipt["provider_calls_attempted"] == 1 with pytest.raises(ValueError, match="already attempted"): bench.execute(root, adapter()) def test_provider_diagnostic_keeps_status_but_not_sensitive_detail(tmp_path): from infospace_bench.errors import InfospaceError root = initialize(tmp_path) class Denied: model = "example/model" def generate(self, request): raise InfospaceError("provider_failed", "Request rejected", { "error": "HTTP Error 401: Unauthorized", "headers": {"Authorization": "Bearer TEST-SECRET-MUST-NOT-PERSIST"}, }) with pytest.raises(InfospaceError): bench.execute(root, Denied()) text = (root / "reports/research-run.json").read_text() result = json.loads(text) assert result["http_status"] == 401 assert result["model"] == "example/model" assert "TEST-SECRET" not in text def test_unidentifiable_source_is_rejected_before_creating_workspace(tmp_path): source = tmp_path / "unidentified.md" source.write_text("# No bibliography") with pytest.raises(ValueError, match="frontmatter"): bench.initialize(source, tmp_path, "invalid") assert not (tmp_path / "infospaces/invalid").exists() def test_negative_extraction_is_a_valid_result(tmp_path): from infospace_bench.workflow import AssistedGenerationResult root = initialize(tmp_path) class Negative: def generate(self, request): return AssistedGenerationResult( markdown="# No supported candidates\n\nInsufficient synthetic evidence.\n", provider="fixture", ) assert bench.execute(root, Negative())["status"] == "completed" assert not list((root / "artifacts/entities").glob("*.md")) def test_cli_refuses_fixture_output_on_real_source(tmp_path): source = tmp_path / "source.md" source.write_text((FIXTURES / "source.md").read_text().replace( "evidence_kind: synthetic", "evidence_kind: primary")) root = bench.initialize(source, tmp_path, "primary-probe") result = subprocess.run([ sys.executable, str(REPO / "scripts/research_bench.py"), "run", str(root), "--fixture-responses", str(FIXTURES / "responses.yaml"), ], capture_output=True, text=True) assert result.returncode != 0 assert "only run against synthetic" in result.stderr assert not (root / "reports/research-run.json").exists() @pytest.mark.parametrize('finish,content,complete', [ ('stop', '# Supported claim\n\nProvisional.', True), ('length', '# Cut short', False), ('stop', '', False), ]) def test_llm_connect_service_contract(tmp_path, finish, content, complete): from llm_connect.models import LLMResponse from llm_connect.server import LLMServer calls = [] class ServiceAdapter: def validate_config(self, config): return True def execute_prompt(self, prompt, config): calls.append((prompt, config)) return LLMResponse(content=content, model='resolved/model', usage={'total_tokens': 42}, finish_reason=finish, metadata={'provider': 'test', 'response_id': 'r1', 'debug': 'DO-NOT-PERSIST'}) server = LLMServer(ServiceAdapter(), port=0) server.start() root = initialize(tmp_path) try: live = bench.LLMConnectAdapter(f'http://127.0.0.1:{server.port}', 'requested/model') if complete: assert bench.execute(root, live)['status'] == 'completed' assert len(calls) == 4 else: with pytest.raises(ValueError, match='empty or incomplete'): bench.execute(root, live) assert len(calls) == 1 receipt_text = (root / 'reports/research-run.json').read_text() receipt = json.loads(receipt_text) assert receipt['responses'][0]['model'] == 'resolved/model' assert receipt['responses'][0]['usage']['total_tokens'] == 42 assert receipt['responses'][0]['finish_reason'] == finish assert receipt['service_requests_attempted'] == len(calls) assert calls[0][1].model_name == 'requested/model' assert calls[0][1].max_tokens == 4000 assert 'DO-NOT-PERSIST' not in receipt_text finally: server.stop() def test_llm_connect_http_failure_is_safe_and_not_retried(tmp_path): from llm_connect.exceptions import LLMAPIError from llm_connect.server import LLMServer calls = [] class FailingService: def validate_config(self, config): return True def execute_prompt(self, prompt, config): calls.append(prompt) raise LLMAPIError('SECRET-SENTINEL', status_code=401, response_body='SECRET-SENTINEL') server = LLMServer(FailingService(), port=0) server.start() root = initialize(tmp_path) try: live = bench.LLMConnectAdapter(f'http://127.0.0.1:{server.port}', 'test') with pytest.raises(bench.InfospaceError) as raised: bench.execute(root, live) text = (root / 'reports/research-run.json').read_text() assert len(calls) == 1 assert json.loads(text)['http_status'] >= 400 assert 'SECRET-SENTINEL' not in text + str(raised.value) finally: server.stop()