import asyncio

import pytest

import main
from agent.memory.store import Store, model_cache_signature
from agent.providers.errors import (
    ModelNotFoundError,
    ProviderAuthenticationError,
    ProviderRateLimitError,
    ProviderTimeoutError,
    ProviderUnavailableError,
)
from agent.routing.fallback import with_fallback
from agent.routing.hybrid_router import Route, RoutingDecision
from agent.routing.model_selector import Selection

def test_fallback_tries_next_capable_candidate():
    candidates=[Selection('first','a',1,'test'),Selection('second','b',1,'test')]
    async def execute(candidate):
        if candidate.provider=='first': raise ProviderUnavailableError('offline')
        return 'ok'
    result,errors=asyncio.run(with_fallback(candidates,execute))
    assert result=='ok' and errors[0]['provider']=='first'


def _model(provider, model_id):
    return {
        'provider': provider, 'model_id': model_id,
        'capabilities': ['general', 'coding', 'reasoning'],
        'capability_profile': {
            'instruction_score': .9, 'coding_score': .9, 'reasoning_score': .9,
        },
        'availability': 'verified', 'health_status': 'healthy',
        'input_price': 0, 'output_price': 0, 'supports_tools': True,
    }


def test_cloud_completion_refreshes_empty_registry_and_uses_discovered_provider(
    tmp_path, monkeypatch
):
    models = []
    refreshed = []

    class Registry:
        def list(self):
            return list(models)

    class Discovery:
        def __init__(self, registry):
            assert registry is main.model_registry

        async def refresh(self, provider_name=None):
            refreshed.append(provider_name)
            models.append(_model('openai', 'discovered-model'))
            return [{'provider': 'openai', 'health_status': 'healthy'}]

    class Provider:
        async def complete_for_model(self, _prompt, _system, model_id):
            assert model_id == 'discovered-model'
            return 'automatic response', {'input_tokens': 2, 'output_tokens': 3}

    monkeypatch.setattr(main, 'store', Store(tmp_path / 'agent.db'))
    monkeypatch.setattr(main, 'model_registry', Registry())
    monkeypatch.setattr(main, 'ProviderDiscovery', Discovery)
    monkeypatch.setattr(main, 'get_provider', lambda _name: Provider())
    monkeypatch.setattr(main.settings, 'cloud_fallback', True)
    monkeypatch.setattr(
        'agent.routing.hybrid_router.provider_policy',
        lambda _name: {'enabled': True, 'routing_priority': 50, 'default_model': ''},
    )

    result = asyncio.run(main._cloud_completion('explain this', 'be concise'))

    assert result['response'] == 'automatic response'
    assert result['provider'] == 'openai'
    assert refreshed == [None]


def test_routing_candidates_bounds_slow_discovery(monkeypatch):
    class Registry:
        def list(self):
            return []

    class Discovery:
        def __init__(self, _registry):
            pass

        async def refresh(self, _provider_name=None):
            await asyncio.sleep(1)

    monkeypatch.setattr(main, 'model_registry', Registry())
    monkeypatch.setattr(main, 'ProviderDiscovery', Discovery)
    monkeypatch.setattr(main.settings, 'discovery_timeout_seconds', .01)

    models, candidates = asyncio.run(main._routing_candidates('explain this'))

    assert models == []
    assert candidates == []


def test_cloud_completion_does_not_refresh_when_registry_has_healthy_candidate(
    tmp_path, monkeypatch
):
    model = _model('openai', 'ready-model')

    class Registry:
        def list(self):
            return [model]

    class Discovery:
        def __init__(self, _registry):
            pass

        async def refresh(self, _provider_name=None):
            raise AssertionError('healthy registry must not be refreshed per chat')

    class Provider:
        async def complete_for_model(self, _prompt, _system, _model_id):
            return 'ready response', {'input_tokens': 1, 'output_tokens': 1}

    monkeypatch.setattr(main, 'store', Store(tmp_path / 'agent.db'))
    monkeypatch.setattr(main, 'model_registry', Registry())
    monkeypatch.setattr(main, 'ProviderDiscovery', Discovery)
    monkeypatch.setattr(main, 'get_provider', lambda _name: Provider())
    monkeypatch.setattr(main.settings, 'cloud_fallback', True)
    monkeypatch.setattr(
        'agent.routing.hybrid_router.provider_policy',
        lambda _name: {'enabled': True, 'routing_priority': 50, 'default_model': ''},
    )

    result = asyncio.run(main._cloud_completion('explain this', 'be concise'))

    assert result['response'] == 'ready response'


def test_cloud_completion_tries_another_model_from_same_provider(tmp_path, monkeypatch):
    models = [_model('first', 'model-a'), _model('first', 'model-b')]

    class Registry:
        def list(self):
            return models

    class Provider:
        async def complete_for_model(self, _prompt, _system, model_id):
            if model_id == 'model-a':
                raise RuntimeError('primary unavailable')
            return 'recovered response', {'input_tokens': 2, 'output_tokens': 3}

    monkeypatch.setattr(main, 'store', Store(tmp_path / 'agent.db'))
    monkeypatch.setattr(main, 'model_registry', Registry())
    monkeypatch.setattr(main, 'get_provider', lambda _name: Provider())
    monkeypatch.setattr(main.settings, 'cloud_fallback', True)
    monkeypatch.setattr(
        'agent.routing.hybrid_router.provider_policy',
        lambda _name: {'enabled': True, 'routing_priority': 50, 'default_model': 'model-a'},
    )

    result = asyncio.run(main._cloud_completion('explain this', 'be concise'))

    assert result['model'] == 'model-b'
    assert [item['status'] for item in result['fallback_trace']] == ['failed', 'success']
    assert result['fallback_trace'][0]['error'] == 'RuntimeError'


def test_cloud_completion_emits_ordered_attempt_events_for_fallback(tmp_path, monkeypatch):
    models = [_model('first', 'model-a'), _model('second', 'model-b'), _model('third', 'model-c')]

    class Registry:
        def list(self):
            return models

    class Provider:
        async def complete_for_model(self, _prompt, _system, model_id):
            if model_id != 'model-c':
                raise RuntimeError(f'{model_id} unavailable')
            return 'recovered response', {'input_tokens': 2, 'output_tokens': 3}

    events = []

    async def capture(event):
        events.append(event)

    monkeypatch.setattr(main, 'store', Store(tmp_path / 'agent.db'))
    monkeypatch.setattr(main, 'model_registry', Registry())
    monkeypatch.setattr(main, 'get_provider', lambda _name: Provider())
    monkeypatch.setattr(main.settings, 'cloud_fallback', True)
    monkeypatch.setattr(main.settings, 'routing_max_attempts', 3)
    monkeypatch.setattr(
        'agent.routing.hybrid_router.provider_policy',
        lambda _name: {'enabled': True, 'routing_priority': 50, 'default_model': ''},
    )

    callback_token = main._stream_event_callback.set(capture)
    request_token = main._workflow_request_id.set('fallback-test')
    main._progress_start('fallback-test')
    try:
        result = asyncio.run(main._cloud_completion('explain this', 'be concise'))
    finally:
        main._workflow_request_id.reset(request_token)
        main._stream_event_callback.reset(callback_token)

    assert result['response'] == 'recovered response'
    assert [
        (event['event'], event['data']['model'], event['data']['status'])
        for event in events
    ] == [
        ('provider_selected', 'model-a', 'running'),
        ('provider_fallback', 'model-a', 'failed'),
        ('provider_selected', 'model-b', 'running'),
        ('provider_fallback', 'model-b', 'failed'),
        ('provider_selected', 'model-c', 'running'),
        ('provider_selected', 'model-c', 'success'),
    ]


def test_cloud_completion_reports_local_and_all_failed_alternatives(tmp_path, monkeypatch):
    models = [_model('first', 'model-a'), _model('second', 'model-b')]

    class Registry:
        def list(self):
            return models

    class Provider:
        async def complete_for_model(self, _prompt, _system, _model_id):
            raise ConnectionError('secret provider detail')

    monkeypatch.setattr(main, 'store', Store(tmp_path / 'agent.db'))
    monkeypatch.setattr(main, 'model_registry', Registry())
    monkeypatch.setattr(main, 'get_provider', lambda _name: Provider())
    monkeypatch.setattr(main.settings, 'cloud_fallback', True)
    monkeypatch.setattr(
        'agent.routing.hybrid_router.provider_policy',
        lambda _name: {'enabled': True, 'routing_priority': 50, 'default_model': ''},
    )

    with pytest.raises(main.HTTPException) as caught:
        asyncio.run(main._cloud_completion('explain this', 'be concise'))

    detail = caught.value.detail
    assert detail['code'] == 'AI_EXECUTION_ERROR'
    assert detail['message'] == (
        '2 provider cloud (first, second) semuanya gagal merespons. '
        'Local Intelligence tidak dicoba karena dinonaktifkan atau tidak dikonfigurasi.'
    )
    assert len(detail['fallback_trace']) == 2
    assert all(item['route'] == 'CLOUD' for item in detail['fallback_trace'])
    assert all(item['source'] == 'cloud_provider' for item in detail['fallback_trace'])
    assert all(item['status'] == 'failed' for item in detail['fallback_trace'])
    assert all(item['failure_reason'] == 'NETWORK_ERROR' for item in detail['fallback_trace'])
    assert all(item['failure_message'] == 'The provider could not be reached.' for item in detail['fallback_trace'])
    assert 'secret provider detail' not in str(detail)


@pytest.mark.parametrize(('route', 'error', 'reason'), [
    (Route.CLOUD, ProviderAuthenticationError('secret'), 'AUTH_ERROR'),
    (Route.CLOUD, ProviderRateLimitError('secret'), 'RATE_LIMIT'),
    (Route.CLOUD, ProviderTimeoutError('secret'), 'TIMEOUT'),
    (Route.CLOUD, ProviderUnavailableError('secret'), 'NETWORK_ERROR'),
    (Route.LOCAL, ModelNotFoundError('secret'), 'LOCAL_MODEL_UNAVAILABLE'),
    (Route.CLOUD, RuntimeError('secret'), 'UNKNOWN'),
])
def test_attempt_record_classifies_secret_free_failure(route, error, reason):
    decision = RoutingDecision(
        route, 'local' if route is Route.LOCAL else 'cloud', 'model', 'test',
        ['general'], 0.1, [], 0.0,
    )

    record = main._attempt_record(decision, 'failed', error)

    assert record['source'] == ('local_model' if route is Route.LOCAL else 'cloud_provider')
    assert record['failure_reason'] == reason
    assert 'secret' not in str(record)


def test_local_to_cloud_fallback_after_quality_rejection(tmp_path, monkeypatch):
    models = [_model('local', 'zevora'), _model('cloud', 'cloud-model')]

    class Registry:
        def list(self):
            return models

    calls = []

    class Provider:
        async def complete_for_model(self, _prompt, _system, model_id):
            calls.append(model_id)
            if model_id == 'zevora':
                raise ProviderUnavailableError('local model is not loaded')
            return 'cloud recovery', {'input_tokens': 1, 'output_tokens': 2}

    monkeypatch.setattr(main, 'store', Store(tmp_path / 'agent.db'))
    monkeypatch.setattr(main, 'model_registry', Registry())
    monkeypatch.setattr(main, 'get_provider', lambda _name: Provider())
    monkeypatch.setattr(main.settings, 'cloud_fallback', True)
    monkeypatch.setattr(
        'agent.routing.hybrid_router.provider_policy',
        lambda _name: {'enabled': True, 'routing_priority': 50, 'default_model': ''},
    )

    result = asyncio.run(main._cloud_completion('explain this', 'be concise'))

    assert result['response'] == 'cloud recovery'
    assert calls == ['zevora', 'cloud-model']
    assert [item['route'] for item in result['fallback_trace']] == ['LOCAL', 'CLOUD']
    assert [item['status'] for item in result['fallback_trace']] == ['failed', 'success']
    assert result['fallback_trace'][0]['failure_reason'] == 'LOCAL_MODEL_UNAVAILABLE'


def test_auto_uses_healthy_local_first_for_simple_prompt(tmp_path, monkeypatch):
    models = [_model('local', 'zevora'), _model('cloud', 'cloud-model')]
    calls = []

    class Registry:
        def list(self):
            return models

    class Provider:
        async def complete_for_model(self, _prompt, _system, model_id):
            calls.append(model_id)
            return f'{model_id} response', {'input_tokens': 1, 'output_tokens': 2}

    monkeypatch.setattr(main, 'store', Store(tmp_path / 'agent.db'))
    monkeypatch.setattr(main, 'model_registry', Registry())
    monkeypatch.setattr(main, 'get_provider', lambda _name: Provider())
    monkeypatch.setattr(main.settings, 'cloud_fallback', True)
    monkeypatch.setattr(main.settings, 'routing_mode', 'AUTO')
    monkeypatch.setattr(
        'agent.routing.hybrid_router.provider_policy',
        lambda _name: {'enabled': True, 'routing_priority': 50, 'default_model': ''},
    )

    result = asyncio.run(main._cloud_completion('explain this', 'be concise'))

    assert result['response'] == 'zevora response'
    assert calls == ['zevora']
    assert result['fallback_trace'][0]['source'] == 'local_model'


def test_auto_reports_local_and_cloud_failures_together(tmp_path, monkeypatch):
    models = [_model('local', 'zevora'), _model('cloud', 'cloud-model')]

    class Registry:
        def list(self):
            return models

    class Provider:
        async def complete_for_model(self, _prompt, _system, model_id):
            if model_id == 'zevora':
                raise ProviderUnavailableError('local secret')
            raise ProviderAuthenticationError('cloud secret')

    monkeypatch.setattr(main, 'store', Store(tmp_path / 'agent.db'))
    monkeypatch.setattr(main, 'model_registry', Registry())
    monkeypatch.setattr(main, 'get_provider', lambda _name: Provider())
    monkeypatch.setattr(main.settings, 'cloud_fallback', True)
    monkeypatch.setattr(main.settings, 'routing_mode', 'AUTO')
    monkeypatch.setattr(
        'agent.routing.hybrid_router.provider_policy',
        lambda _name: {'enabled': True, 'routing_priority': 50, 'default_model': ''},
    )

    with pytest.raises(main.HTTPException) as caught:
        asyncio.run(main._cloud_completion('explain this', 'be concise'))

    detail = caught.value.detail
    trace = detail['fallback_trace']
    assert detail['message'] == (
        'Local Intelligence dan 1 provider cloud (cloud) semuanya gagal merespons.'
    )
    assert [item['source'] for item in trace] == ['local_model', 'cloud_provider']
    assert [item['failure_reason'] for item in trace] == [
        'LOCAL_MODEL_UNAVAILABLE', 'AUTH_ERROR',
    ]
    assert 'secret' not in str(trace)


def test_local_only_failure_message_does_not_claim_cloud_was_tried():
    trace = [{
        'source': 'local_model', 'route': 'LOCAL', 'provider': 'local',
        'model': 'zevora', 'status': 'failed',
    }]

    message = main._fallback_failure_message(trace)

    assert message == (
        'Local Intelligence gagal, dan tidak ada provider cloud yang terkonfigurasi '
        'untuk fallback. Tambahkan API key di halaman Providers agar ada cadangan otomatis.'
    )


def test_cloud_to_local_fallback_after_cloud_failure(tmp_path, monkeypatch):
    models = [_model('local', 'zevora'), _model('cloud', 'cloud-model')]

    class Registry:
        def list(self):
            return models

    calls = []

    class Provider:
        async def complete_for_model(self, _prompt, _system, model_id):
            calls.append(model_id)
            if model_id == 'cloud-model':
                raise ConnectionError('offline')
            return 'local recovery', {'input_tokens': 1, 'output_tokens': 2}

    monkeypatch.setattr(main, 'store', Store(tmp_path / 'agent.db'))
    monkeypatch.setattr(main, 'model_registry', Registry())
    monkeypatch.setattr(main, 'get_provider', lambda _name: Provider())
    monkeypatch.setattr(main.settings, 'cloud_fallback', True)
    monkeypatch.setattr(main.settings, 'routing_mode', 'AUTO')
    monkeypatch.setattr(
        'agent.routing.hybrid_router.provider_policy',
        lambda _name: {'enabled': True, 'routing_priority': 50, 'default_model': ''},
    )

    result = asyncio.run(main._cloud_completion(
        'migrate and redesign the whole project architecture', 'be concise'
    ))

    assert result['response'] == 'local recovery'
    assert calls == ['cloud-model', 'zevora']
    assert [item['route'] for item in result['fallback_trace']] == ['CLOUD', 'LOCAL']
    assert [item['status'] for item in result['fallback_trace']] == ['failed', 'success']


def test_task_exact_cache_resolves_locally_without_provider(tmp_path, monkeypatch):
    isolated_store = Store(tmp_path / 'agent.db')
    isolated_store.put_cache(
        'cached question', 'cached answer', 'openai', 'model-a', 'general',
        model_signature=model_cache_signature(),
    )
    monkeypatch.setattr(main, 'store', isolated_store)
    monkeypatch.setattr(
        main, 'get_provider',
        lambda _name: (_ for _ in ()).throw(AssertionError('provider must not run')),
    )

    result = asyncio.run(main.task(main.TaskRequest(prompt='cached question')))

    assert result['route'] == 'CACHE'
    assert result['response'] == 'cached answer'
    assert result['context_status'] == 'CACHE_SUFFICIENT'
    assert result['project_discovery'] is None
    assert result['flow'] == {
        'workspace': 'NOT_SELECTED', 'discovery': 'SKIPPED',
        'context': 'CACHE_SUFFICIENT', 'route': 'CACHE',
        'action': 'SKIPPED', 'verification': 'SKIPPED',
        'knowledge': 'RETRIEVED',
    }
    assert result['fallback_trace'] == [
        {'source': 'local', 'status': 'success', 'kind': 'exact_cache'}
    ]


def test_task_generation_reports_discovery_context_and_flow(tmp_path, monkeypatch):
    (tmp_path / 'pyproject.toml').write_text(
        '[project]\nname = "flow-test"\n', encoding='utf-8'
    )
    (tmp_path / 'app.py').write_text('print("hello")\n', encoding='utf-8')
    isolated_store = Store(tmp_path / 'agent.db')
    extracted = []
    provider_systems = []
    model = _model('local', 'zevora')
    decision = RoutingDecision(
        Route.LOCAL, 'local', 'zevora', 'BEST_LOCAL_MATCH',
        ['general'], 0.1, [], 0.0,
    )

    class Registry:
        def list(self):
            return [model]

    class Intelligence:
        def build_context(self, *_args, **_kwargs):
            return 'Relevant knowledge:\n- Use the project manifest.'

        def extract_knowledge(self, *args):
            extracted.append(args)

    async def complete(_candidate, _prompt, system, _images):
        provider_systems.append(system)
        return 'The project manifest defines the application package.', {
            'input_tokens': 4, 'output_tokens': 7,
        }

    monkeypatch.setattr(main, 'store', isolated_store)
    monkeypatch.setattr(main, 'model_registry', Registry())
    monkeypatch.setattr(main, 'intelligence_engine', Intelligence())
    monkeypatch.setattr(main, '_cloud_candidates', lambda *_args: [decision])
    monkeypatch.setattr(main, '_provider_completion', complete)

    result = asyncio.run(main.task(main.TaskRequest(
        prompt='Explain this project setup', project=str(tmp_path),
    )))

    assert result['route'] == 'LOCAL'
    assert result['context_status'] == 'RETRIEVAL_ENRICHED'
    assert result['project_discovery']['languages'] == ['Python']
    assert 'pyproject.toml' in result['project_discovery']['manifests']
    assert result['flow'] == {
        'workspace': 'SELECTED', 'discovery': 'COMPLETE',
        'context': 'RETRIEVAL_ENRICHED', 'route': 'LOCAL',
        'action': 'SKIPPED', 'verification': 'SKIPPED',
        'knowledge': 'EXTRACTED',
    }
    assert 'Relevant knowledge' in provider_systems[0]
    assert 'Project discovery (authoritative local index):' in provider_systems[0]
    assert 'Languages: Python' in provider_systems[0]
    assert extracted[0][3:5] == ('local', 'zevora')


def test_exact_cache_does_not_replay_across_model_signature_change(tmp_path):
    isolated_store = Store(tmp_path / 'agent.db')
    sig_a = 'a' * 64
    sig_b = 'b' * 64
    isolated_store.put_cache(
        'same prompt', 'answer under model A', 'local', 'qwen3.8-flash-next',
        'coding', model_signature=sig_a,
    )

    assert isolated_store.get_cache('same prompt', '', sig_a) is not None
    assert isolated_store.get_cache('same prompt', '', sig_b) is None
    assert isolated_store.get_cache('same prompt', '', sig_a)['response'] == 'answer under model A'


def test_model_cache_signature_changes_when_quant_changes(monkeypatch):
    from agent.config import settings as live_settings
    from agent.memory.store import model_cache_signature as sig
    monkeypatch.setattr(live_settings, 'local_model_quant', 'UD-Q4_K_XL')
    sig_base = sig()
    monkeypatch.setattr(live_settings, 'local_model_quant', 'Q4_K_M')
    sig_alt = sig()
    assert sig_base != sig_alt