frontend-patterns/tests/test_research_bench.py

"""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]
sys.path.insert(0, str(REPO / "scripts"))
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):
    root = bench.initialize(FIXTURES / "source.md", tmp_path, "synthetic-probe")
    (root / "profile/execution.json").unlink()  # Exercise historical four-call compatibility.
    return root


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', reasoning_effort='low')
        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
            assert (root / 'reports/rejected-response.md').read_text() == content
        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['responses'][0]['content_characters'] == len(content)
        assert receipt['responses'][0]['content_sha256'] is not None
        assert receipt['service_requests_attempted'] == len(calls)
        assert calls[0][1].model_name == 'requested/model'
        assert calls[0][1].max_tokens == 4000
        assert calls[0][1].model_params['reasoning_effort'] == 'low'
        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()


def test_diagnostics_allowlist_excludes_raw_provider_data():
    raw = {'debug': {'provider_request': {'headers': {'Authorization': 'SECRET'},
        'payload': {'messages': ['SECRET'], 'max_tokens': 1000, 'reasoning_effort': 'low'}},
        'provider_response': {'body': {'choices': ['SECRET'], 'usage': {
            'total_tokens': 1200, 'cost': 0.01, 'private': 'SECRET',
            'completion_tokens_details': {'reasoning_tokens': 900, 'private': 'SECRET'}}}}}}
    summary = bench.diagnostic_summary(raw)
    assert summary['effective_settings']['reasoning_effort_present']
    assert summary['provider_usage']['completion_tokens_details']['reasoning_tokens'] == 900
    assert 'SECRET' not in json.dumps(summary)


def test_current_llm_connect_drops_reasoning_effort():
    from llm_connect._payload import merge_openai_chat_model_params
    payload = {'max_tokens': 1000}
    merge_openai_chat_model_params(payload, {'reasoning_effort': 'low'})
    assert 'reasoning_effort' not in payload


def conditional_root(tmp_path, slug='conditional'):
    return bench.initialize(FIXTURES / 'source.md', tmp_path, slug)


def structured_adapter(count=1, edges=None, malformed=False):
    from infospace_bench.workflow import AssistedGenerationResult
    candidates = {'candidates': [
        {'id': f'P{i+1:02}', 'known_uses': [], 'recurrence': 'unknown'} for i in range(count)]}
    return FixtureAssistedGenerationAdapter({
        ('claims', '*'): AssistedGenerationResult('# Synthetic claims', provider='fixture'),
        ('candidates', '*'): AssistedGenerationResult(
            'broken' if malformed else '```json\n' + json.dumps(candidates) + '\n```', provider='fixture'),
        ('relations', '*'): AssistedGenerationResult(
            '```json\n' + json.dumps(edges or []) + '\n```', provider='fixture'),
    })


@pytest.mark.parametrize('count,expected', [(0, 2), (1, 2), (2, 3)])
def test_conditional_stage_call_counts(tmp_path, count, expected):
    root = conditional_root(tmp_path)
    result = bench.execute(root, structured_adapter(count), max_calls=expected)
    assert result['status'] == 'completed'
    assert result['provider_calls_attempted'] == expected
    assert result['deterministic_stages'] == 4 - expected
    assert 'Editorial review required' in (root / 'output/evaluations/source-review.md').read_text()
    assert result['editorial_corrections'] is None
    assert result['review_status'] == 'unreviewed'


def test_invalid_candidates_stop_before_relations(tmp_path):
    root = conditional_root(tmp_path)
    with pytest.raises(ValueError, match='candidates structural screen'):
        bench.execute(root, structured_adapter(malformed=True))
    receipt = json.loads((root / 'reports/research-run.json').read_text())
    assert receipt['provider_calls_attempted'] == 2
    assert receipt['rejected_assertions'] > 0
    assert (root / 'reports/rejected-stage.json').exists()
    assert not (root / 'artifacts/generated/relation-proposals.md').exists()


def test_invalid_edges_stop_before_editorial_stage(tmp_path):
    root = conditional_root(tmp_path)
    with pytest.raises(ValueError, match='relations structural screen'):
        bench.execute(root, structured_adapter(2, [{'subject_candidate_id': 'P01',
            'object_candidate_id': 'C01', 'relation': 'requires'}]))
    receipt = json.loads((root / 'reports/research-run.json').read_text())
    assert receipt['provider_calls_attempted'] == 3
    assert not (root / 'output/evaluations/source-review.md').exists()


def test_explicit_reuse_makes_no_new_calls(tmp_path):
    first = conditional_root(tmp_path, 'first')
    bench.execute(first, structured_adapter())
    second = conditional_root(tmp_path, 'second')
    result = bench.execute(second, structured_adapter(), max_calls=0, reuse_from=first)
    assert result['provider_calls_attempted'] == 0
    assert result['cache_hits'] == 2
    assert 'reused_from' in (second / 'artifacts/index.yaml').read_text()


@pytest.mark.parametrize('change', ['prompt', 'source', 'fixture'])
def test_changed_inputs_never_reuse_stale_output(tmp_path, change):
    first = conditional_root(tmp_path, 'first')
    bench.execute(first, structured_adapter())
    source = FIXTURES / 'source.md'
    if change == 'source':
        source = tmp_path / 'different.md'
        source.write_text((FIXTURES / 'source.md').read_text() + '\nNew constraint.\n')
    second = bench.initialize(source, tmp_path, 'second')
    if change == 'prompt':
        template = second / 'profile/templates/claims.md'
        template.write_text(template.read_text() + '\nChanged question.\n')
    with pytest.raises(ValueError, match='call limit'):
        bench.execute(second, structured_adapter(2 if change == 'fixture' else 1),
                      max_calls=0, reuse_from=first)
    receipt = json.loads((second / 'reports/research-run.json').read_text())
    assert receipt['cache_hits'] == 0
    assert receipt['provider_calls_attempted'] == 0


def test_modified_cache_and_failed_run_are_not_reusable(tmp_path):
    first = conditional_root(tmp_path, 'first')
    bench.execute(first, structured_adapter())
    cache = first / 'reports/stage-cache.json'
    cache.write_text('{}')
    second = conditional_root(tmp_path, 'second')
    with pytest.raises(ValueError, match='matching cache digest'):
        bench.execute(second, structured_adapter(), reuse_from=first)
    assert not (second / 'reports/research-run.json').exists()


def test_two_candidate_relation_survives_screen(tmp_path):
    root = conditional_root(tmp_path)
    edge = {'subject_candidate_id': 'P01', 'object_candidate_id': 'P02',
            'relation': 'complements', 'rationale': 'Synthetic complement',
            'source_claim_and_locator': 'C01 Scenario',
            'counterexample': 'No persistence requirement',
            'observation_versus_hypothesis': 'hypothesis'}
    result = bench.execute(root, structured_adapter(2, [edge]))
    assert result['provider_calls_attempted'] == 3
    assert result['status'] == 'completed'


def test_service_reuse_and_setting_invalidation(tmp_path):
    from llm_connect.models import LLMResponse
    from llm_connect.server import LLMServer
    calls = []

    class Service:
        def validate_config(self, config):
            return True

        def execute_prompt(self, prompt, config):
            calls.append(prompt)
            return LLMResponse(content='```json\n{"candidates": []}\n```',
                               model='test', usage={'total_tokens': 42})

    server = LLMServer(Service(), port=0)
    server.start()
    endpoint = f'http://127.0.0.1:{server.port}'
    try:
        first = conditional_root(tmp_path, 'first')
        r = bench.execute(first, bench.LLMConnectAdapter(endpoint, 'test'))
        assert r['reported_tokens_this_run'] == 84
        second = conditional_root(tmp_path, 'second')
        r = bench.execute(second, bench.LLMConnectAdapter(endpoint, 'test'),
                          max_calls=0, reuse_from=first)
        assert r['reported_tokens_this_run'] == 0
        assert r['service_requests_attempted'] == 0
        assert len(calls) == 2
        third = conditional_root(tmp_path, 'third')
        with pytest.raises(ValueError, match='call limit'):
            bench.execute(third, bench.LLMConnectAdapter(endpoint, 'different-model'),
                          max_calls=0, reuse_from=first)
        receipt = json.loads((first / 'reports/research-run.json').read_text())
        receipt['status'] = 'failed'
        (first / 'reports/research-run.json').write_text(json.dumps(receipt))
        fourth = conditional_root(tmp_path, 'fourth')
        with pytest.raises(ValueError, match='completed run'):
            bench.execute(fourth, bench.LLMConnectAdapter(endpoint, 'test'), reuse_from=first)
    finally:
        server.stop()