import asyncio
import json
from dataclasses import replace
from datetime import datetime, timedelta, timezone

import pytest

from viano_bug_analysis.api import APIConfig, Client, Project
from viano_bug_analysis.lifecycle import State
from viano_bug_analysis.lifecycle.locking import FileLock
from viano_bug_analysis.lifecycle.errors import LockBusy
from viano_bug_analysis.package import PackageReader
from viano_bug_analysis.worker import AnalysisWorker, WorkerConfig, WorkerError
from viano_bug_analysis.worker.config import binding
from test_case_builder import case_setup
from test_case_interpreter import gateway, FixtureAdapter

CASE = ('case-A', 'stable-package')


@pytest.fixture(autouse=True)
def project_manifest(manifest):
    manifest['project'] = {'id': 'test', 'type': 'wow', 'profile': 'vcore'}


@pytest.fixture
def setup(case_setup):
    store = case_setup[0]
    project = Project('test', 'wow', 'vcore', planning_mode='separate')
    config = WorkerConfig(APIConfig(store.workspace.root, (project,),
        (Client('capture', 'TEST_TOKEN', ('test',)),)), min_free_bytes=1)
    def loader(p):
        return gateway(FixtureAdapter()), {'planning_mode': 'separate'}, binding(p)
    return config, loader, store


def test_real_case_build_then_interpret_and_restart(setup):
    config, loader, store = setup
    worker = AnalysisWorker(config, loader=loader)
    assert asyncio.run(worker.tick()) == 1
    with PackageReader(store.latest(*CASE)) as reader:
        assert reader.manifest['state'] == State.BUILDING_CASE
        assert reader.manifest['artifacts']['case']
    # Fresh worker object has no in-memory queue. It recovers from managed packages.
    resumed = AnalysisWorker(config, loader=loader)
    assert asyncio.run(resumed.tick()) == 1
    with PackageReader(store.latest(*CASE)) as reader:
        assert reader.manifest['state'] == State.INTERPRETING_CASE
        assert reader.manifest['artifacts']['case_interpretation']


def test_scheduler_retry_is_durable_and_not_busy_loop(setup):
    config, loader, store = setup
    now = datetime.now(timezone.utc)
    calls = []
    async def dispatch(*args, **kwargs):
        calls.append(kwargs)
        return {'status': 'RETRY_PENDING', 'code': 'LLM_CONNECTION',
                'retry_not_before': (now + timedelta(minutes=5)).isoformat()}
    worker = AnalysisWorker(config, loader=loader, dispatcher=dispatch, clock=lambda: now)
    assert asyncio.run(worker.tick()) == 1
    again = AnalysisWorker(config, loader=loader, dispatcher=dispatch, clock=lambda: now)
    assert asyncio.run(again.tick()) == 0
    assert len(calls) == 1 and calls[0]['attempt_id'] == 'default'
    after = AnalysisWorker(config, loader=loader, dispatcher=dispatch,
        clock=lambda: now + timedelta(minutes=6))
    assert asyncio.run(after.tick()) == 1


@pytest.mark.parametrize('reason,automatic', [
    ('AWAITING_TEST_FEEDBACK', True), ('INVALID_LLM_RESPONSES', True),
    ('LLM_UNAVAILABLE', True), ('RESEARCH_UNAVAILABLE', False)])
def test_worker_never_silently_bypasses_block(setup, reason, automatic):
    config, loader, store = setup
    store.advance(*CASE, State.BUILDING_CASE, step_id='start')
    blocked = store.advance(*CASE, State.BLOCKED, step_id='block', reason=reason,
        retry_not_before='2020-01-01T00:00:00Z', details={'automatic_retry': automatic})
    async def forbidden(*args, **kwargs):
        raise AssertionError('blocked case was dispatched')
    worker = AnalysisWorker(config, loader=loader, dispatcher=forbidden)
    assert asyncio.run(worker.tick()) == 0
    assert store.latest(*CASE) == blocked
    if reason != 'RESEARCH_UNAVAILABLE':
        with pytest.raises(WorkerError, match='WORKER_RETRY_STATE'):
            worker.retry(*CASE)
    else:
        worker.retry(*CASE)
        with PackageReader(store.latest(*CASE)) as reader:
            assert reader.manifest['state'] == 'BUILDING_CASE'


def test_explicitly_permitted_due_retry_resumes(setup):
    config, loader, store = setup
    store.advance(*CASE, State.BUILDING_CASE, step_id='start')
    store.advance(*CASE, State.BLOCKED, step_id='block', reason='RESEARCH_UNAVAILABLE',
        retry_not_before='2020-01-01T00:00:00Z', details={'automatic_retry': True})
    seen = []
    async def dispatch(store, gateway, *identity, **kwargs):
        with PackageReader(store.latest(*identity)) as reader:
            seen.append(reader.manifest['state'])
        return {'status': 'YIELDED'}
    assert asyncio.run(AnalysisWorker(config, loader=loader, dispatcher=dispatch).tick()) == 1
    assert seen == ['BUILDING_CASE']


def test_one_service_and_trial_lease_exclusion(setup):
    config, loader, store = setup
    worker = AnalysisWorker(config, loader=loader)
    with FileLock(worker.workspace.runtime / 'locks/analysis-service.lock'):
        with pytest.raises(LockBusy):
            asyncio.run(worker.run(once=True))
    from viano_bug_analysis.lifecycle.storage import key
    with FileLock(worker.workspace.runtime / 'locks' / ('trial-' + key(CASE[0]) + '.lock')):
        assert asyncio.run(worker.tick()) == 0


def test_exception_backoff_and_explicit_scheduler_release(setup):
    config, loader, store = setup
    calls = []
    async def broken(*args, **kwargs):
        calls.append(1)
        raise ValueError('secret data must not appear in runtime state')
    worker = AnalysisWorker(config, loader=loader, dispatcher=broken)
    asyncio.run(worker.tick())
    asyncio.run(worker.tick())
    assert len(calls) == 1
    record = next(worker.records.glob('*.json')).read_text()
    assert 'secret data' not in record
    worker.retry(*CASE)
    asyncio.run(worker.tick())
    assert len(calls) == 2


def test_stop_requested_during_dispatch_finishes_step(setup):
    config, loader, store = setup
    async def dispatch(*args, **kwargs):
        worker.request_stop()
        return {'status': 'YIELDED'}
    worker = AnalysisWorker(config, loader=loader, dispatcher=dispatch)
    asyncio.run(worker.run())
    assert json.loads((worker.root / 'status.json').read_bytes())['status'] == 'STOPPED'
    assert next(worker.records.glob('*.json')).exists()


def test_config_binding_persisted_before_dispatch_and_detects_restart_change(setup):
    config, loader, store = setup
    async def dispatch(*args, **kwargs):
        assert next(worker.records.glob('*.json')).exists()
        return {'status': 'YIELDED'}
    worker = AnalysisWorker(config, loader=loader, dispatcher=dispatch)
    asyncio.run(worker.tick())
    changed = replace(config.intake.projects[0], planning_mode='joint')
    updated = replace(config, intake=replace(config.intake, projects=(changed,)))
    other = AnalysisWorker(updated, loader=loader, dispatcher=dispatch)
    assert asyncio.run(other.tick()) == 0
    record = json.loads(next(other.records.glob('*.json')).read_bytes())
    assert record['status'] == 'ATTENTION' and record['code'] == 'WORKER_CONFIG_CHANGED'


def test_unknown_project_held_without_model_call(setup):
    config, loader, store = setup
    changed = replace(config.intake.projects[0], id='other')
    updated = replace(config, intake=APIConfig(config.intake.workspace, (changed,),
        (Client('capture', 'TEST_TOKEN', ('other',)),)))
    def forbidden(project):
        raise AssertionError('unknown project')
    worker = AnalysisWorker(updated, loader=forbidden)
    assert asyncio.run(worker.tick()) == 0
    assert json.loads(next(worker.records.glob('*.json')).read_bytes())['code'] == 'WORKER_PROJECT_UNKNOWN'


def test_terminal_result_not_republished_every_pass(setup):
    config, loader, store = setup
    calls = []
    async def dispatch(*args, **kwargs):
        calls.append(1)
        return {'status': 'OUTPUT_READY'}
    worker = AnalysisWorker(config, loader=loader, dispatcher=dispatch)
    asyncio.run(worker.tick())
    asyncio.run(worker.tick())
    assert len(calls) == 1


def test_reopened_revision_does_not_skip_new_interpretation(setup):
    config, loader, store = setup
    worker = AnalysisWorker(config, loader=loader)
    asyncio.run(worker.tick())
    asyncio.run(worker.tick())
    for state in ('GENERATING_HYPOTHESES', 'INVESTIGATING_HYPOTHESES', 'BUILDING_DIAGNOSIS', 'OUTPUT_READY', 'HANDED_OFF'):
        store.advance(*CASE, state, step_id='test:' + state)
    store.advance(*CASE, 'REOPENED', step_id='test:reopen')
    asyncio.run(worker.tick())
    with PackageReader(store.latest(*CASE)) as reader:
        assert reader.manifest['analysis_revision'] == 2
        assert reader.manifest['state'] == 'BUILDING_CASE'
        # Old interpretation is deliberately retained in the append-only package.
        old = json.loads(reader.read_bytes(reader.manifest['artifacts']['case_interpretation']))
        assert old['analysis_revision'] == 1
    asyncio.run(worker.tick())
    with PackageReader(store.latest(*CASE)) as reader:
        doc = json.loads(reader.read_bytes(reader.manifest['artifacts']['case_interpretation']))
        assert doc['analysis_revision'] == 2


def test_multiple_projects_and_poison_case_do_not_starve_other_cases(setup, case_setup, tmp_path):
    from viano_bug_analysis.package.factory import PackageFactory
    config, loader, store = setup
    with PackageReader(case_setup[2]) as reader:
        manifest = reader.manifest
        contents = {ref: reader.read_bytes(ref) for ref in reader.files if ref != 'manifest.json'}
    manifest.update(case_id='case-B', package_id='other-package', project={'id': 'second', 'type': 'wow', 'profile': 'vcore'})
    PackageFactory().create(store.workspace.inbox / 'other.zip', manifest, contents)
    project2 = Project('second', 'wow', 'vcore', planning_mode='separate')
    intake = replace(config.intake, projects=config.intake.projects + (project2,))
    seen = []
    def load(project):
        return project.id, {'project_id': project.id}, binding(project)
    async def dispatch(store, gateway, case_id, package_id, **kwargs):
        seen.append((case_id, gateway, kwargs['policies']['project_id']))
        if case_id == 'case-A':
            raise ValueError('broken first case')
        return {'status': 'YIELDED'}
    worker = AnalysisWorker(replace(config, intake=intake), loader=load, dispatcher=dispatch)
    asyncio.run(worker.tick())
    assert seen == [('case-A', 'test', 'test'), ('case-B', 'second', 'second')]
    asyncio.run(worker.tick())
    assert [v[0] for v in seen].count('case-A') == 1
    assert [v[0] for v in seen].count('case-B') == 2


def test_cli_check_status_and_empty_once_do_not_need_api_token(tmp_path, monkeypatch, capsys):
    from viano_bug_analysis.worker.__main__ import main
    import viano_bug_analysis.worker.__main__ as cli
    monkeypatch.delenv('ABSENT_API_TOKEN', raising=False)
    intake = tmp_path / 'api.toml'
    intake.write_text('''[api]
workspace = "service"
[[projects]]
id = "test"
type = "wow"
profile = "vcore"
[[clients]]
id = "capture"
token_env = "ABSENT_API_TOKEN"
projects = ["test"]
''')
    config = tmp_path / 'worker.toml'
    config.write_text('[worker]\nintake_config = "api.toml"\nmin_free_bytes = 1\n')
    loaded = []
    monkeypatch.setattr(cli, 'load_project', lambda p: loaded.append(p.id))
    assert main(['check', '--config', str(config)]) == 0
    assert loaded == ['test']
    assert main(['run', '--once', '--config', str(config)]) == 0
    assert main(['status', '--config', str(config)]) == 0
    assert 'STOPPED' in capsys.readouterr().out


def test_storage_floor_pauses_without_dispatch(setup, monkeypatch):
    import viano_bug_analysis.worker.service as module
    from collections import namedtuple
    config, loader, store = setup
    Usage = namedtuple('Usage', 'total used free')
    worker = AnalysisWorker(config, loader=loader)
    monkeypatch.setattr(module.shutil, 'disk_usage', lambda *_: Usage(100, 100, 0))
    assert asyncio.run(worker.tick()) == 0
    assert json.loads((worker.root / 'status.json').read_bytes())['status'] == 'STORAGE_LOW'


def test_exception_after_checkpoint_holds_actual_latest_generation(setup):
    config, loader, store = setup
    calls = []
    async def dispatch(store, gateway, *identity, **kwargs):
        calls.append(1)
        store.advance(*identity, 'BUILDING_CASE', step_id='commit-before-error')
        raise ValueError('unexpected after checkpoint')
    worker = AnalysisWorker(config, loader=loader, dispatcher=dispatch)
    asyncio.run(worker.tick())
    asyncio.run(worker.tick())
    assert len(calls) == 1
    record = json.loads(next(worker.records.glob('*.json')).read_bytes())
    with PackageReader(store.latest(*CASE)) as reader:
        assert record['package_generation'] == reader.manifest['package_generation']
    assert record['status'] == 'ATTENTION'


def test_operator_pause_survives_restart_and_drains_started_cases(setup):
    config,loader,store=setup
    worker=AnalysisWorker(config,loader=loader)
    worker.control.pause('operator','test',True)
    assert asyncio.run(worker.tick())==0
    worker=AnalysisWorker(config,loader=loader)
    assert asyncio.run(worker.tick())==0
    worker.control.pause('operator','test',False)
    assert asyncio.run(worker.tick())==1
    worker.control.pause('operator','test',True)
    assert asyncio.run(worker.tick())==1 # already started case continues


def test_operator_commands_recheck_generation_and_forbid_validation_reset(setup):
    config,loader,store=setup
    worker=AnalysisWorker(config,loader=loader)
    store.advance(*CASE,State.BUILDING_CASE,step_id='operator:start')
    store.advance(*CASE,State.BLOCKED,step_id='operator:block',reason='RESEARCH_UNAVAILABLE',
        retry_not_before='2020-01-01T00:00:00Z',details={'automatic_retry':False})
    with PackageReader(store.latest(*CASE)) as reader:
        manifest=reader.manifest
    worker.control.enqueue('operator','test',manifest)
    worker._commands()
    assert not worker.control.read()['commands']
    assert worker.control.read()['events'][-1]['status']=='APPLIED'
    worker.control.enqueue('operator','test',manifest) # stale UI cannot release new head
    worker._commands()
    assert worker.control.read()['events'][-1]['status']=='REJECTED'
    store.advance(*CASE,State.BLOCKED,step_id='operator:invalid',reason='INVALID_LLM_RESPONSES',
        retry_not_before='2020-01-01T00:00:00Z')
    path=store.latest(*CASE)
    with PackageReader(path) as reader:
        worker.control.enqueue('operator','test',reader.manifest)
    worker._commands()
    assert store.latest(*CASE)==path
    assert worker.control.read()['events'][-1]['status']=='REJECTED'