"""End-to-end invariants from the September 11 live audit."""
import json
import multiprocessing
import os
from pathlib import Path
from unittest.mock import patch

import pytest
import state_store
import trigger
import workflow
import worker
import completion_watcher
from types import SimpleNamespace


@pytest.fixture
def board(tmp_path, monkeypatch):
    for name in ('intent', 'research', 'inbox', 'processing', 'builder', 'builder-processing'):
        (tmp_path / name).mkdir()
    for name, value in {
        'BOARD_STATE_FILE': str(tmp_path / 'state.json'),
        'BOARD_INTENT_DIR': str(tmp_path / 'intent'),
        'BOARD_RESEARCH_INTENT_DIR': str(tmp_path / 'research'),
        'BOARD_INBOX_DIR': str(tmp_path / 'inbox'),
        'BOARD_PROCESSING_DIR': str(tmp_path / 'processing'),
        'AGENT_BUILDER_QUEUE_DIR': str(tmp_path / 'builder'),
        'AGENT_BUILDER_PROCESSING_DIR': str(tmp_path / 'builder-processing'),
    }.items():
        monkeypatch.setenv(name, value)
    (tmp_path / 'intent' / 'goal.md').write_text('# Goal\n')
    trigger.save_state({'cards': {}})
    return tmp_path


@pytest.mark.parametrize('phase', ['deployed', 'bereit_fuer_deployment', 'recherche_fertig'])
def test_manual_moves_cannot_claim_verified_outcomes(board, phase):
    with pytest.raises(PermissionError):
        trigger.update_phase('goal', phase)
    assert not workflow.verified_deployment({'phase': 'deployed'})
    assert trigger.deployment_evidence({'phase': 'deployed'})['status'] == 'unknown'


def test_failed_inspector_can_archive_and_reactivate(board):
    trigger.save_state({'cards': {'goal': {'phase': 'inspector_fehlgeschlagen', 'last_run_status': 'failed'}}})
    assert 'archive' in [a['id'] for a in trigger.available_actions('goal', 'intent', trigger.load_state()['cards']['goal'])]
    trigger.update_phase('goal', 'archiviert')
    trigger.update_phase('goal', 'geplant')
    assert trigger.load_state()['cards']['goal']['phase'] == 'geplant'


def test_research_reset_returns_without_partial_failure(board):
    (board / 'research' / 'study.md').write_text('# Study\n')
    trigger.save_state({'cards': {'study': {'source': 'research', 'phase': 'recherche_fehlgeschlagen'}}})
    assert trigger.update_phase('study', 'recherche_angefordert') is None
    assert trigger.load_state()['cards']['study']['phase'] == 'recherche_angefordert'


def test_downstream_builder_is_visible_to_all_trigger_guards(board):
    (board / 'builder-processing' / 'job.json').write_text(json.dumps({'board_slug': 'goal'}))
    assert trigger.builder_trigger_blocked('goal')
    with pytest.raises(PermissionError):
        trigger.update_phase('goal', 'archiviert')


def test_routing_to_builder_queues_real_job(board):
    trigger.save_state({'cards': {'goal': {'phase': 'inspector_fehlgeschlagen', 'routing_pending': True,
                                          'routing_recommendation': 'in_umsetzung'}}})
    assert trigger.apply_routing_recommendation('goal') == 'in_umsetzung'
    assert len(list((board / 'inbox').glob('*.json'))) == 1
    assert trigger.load_state()['cards']['goal']['last_run_status'] == 'queued'


def _increment(path, count):
    os.environ['BOARD_STATE_FILE'] = path
    for _ in range(count):
        with state_store.transaction():
            state = state_store.load()
            state['count'] = state.get('count', 0) + 1
            state_store.save(state)


def test_concurrent_processes_do_not_lose_updates(board):
    context = multiprocessing.get_context('fork')
    processes = [context.Process(target=_increment, args=(str(board / 'state.json'), 20)) for _ in range(3)]
    for process in processes:
        process.start()
    for process in processes:
        process.join(10)
        assert process.exitcode == 0
    assert state_store.load()['count'] == 60


def test_corrupt_state_fails_closed(board):
    (board / 'state.json').write_text('{broken')
    with pytest.raises(ValueError):
        trigger.create_idea('Do not erase state')
    assert (board / 'state.json').read_text() == '{broken'


def test_claimed_job_is_not_republished_after_outbox_crash(board):
    target = board / 'inbox' / 'durable-job.json'
    state_store.atomic_json(target, {'job_id': 'durable-job'})
    assert state_store.claim_job(target, board / 'processing' / target.name)
    state_store.atomic_json(board / 'state.json', {'cards': {}, '_pending_writes': {str(target): {'job_id': 'durable-job'}}})
    with state_store.transaction():
        pass
    assert not target.exists()
    assert (board / 'processing' / target.name).exists()


def test_changed_commit_invalidates_old_activation_receipt(board):
    trigger.save_state({'cards': {'goal': {'commit_hash': 'a' * 40, 'deployment_status': 'deployed',
                                         'deployed_commit_hash': 'a' * 40, 'activated_system': '/nix/store/old'}}})
    worker.update_state('goal', 'done', 'review-new', commit_hash='b' * 40)
    assert not workflow.verified_deployment(trigger.load_state()['cards']['goal'])


@pytest.mark.parametrize('verdict', [None, {'schema_version': 1, 'commit_hash': 'badbeef', 'decision': 'approve'},
                                     {'schema_version': True, 'commit_hash': 'abc1234', 'decision': 'approve'}])
def test_invalid_review_never_replaces_previous_report(board, monkeypatch, verdict):
    from test_board_agent import RunReviewTests
    for key, value in RunReviewTests()._environment(str(board)).items():
        monkeypatch.setenv(key, value)
    report_dir = board / 'docs' / 'review'
    report_dir.mkdir(parents=True)
    report = report_dir / 'goal.md'
    report.write_text('Previous verified report')
    response = '===REVIEW-REPORT===\nNot ready for deployment. Decision: back to planning.\n===END-REVIEW-REPORT==='
    if verdict is not None:
        response += '\n===REVIEW-RESULT===\n' + json.dumps(verdict) + '\n===END-REVIEW-RESULT==='
    output = json.dumps({'type': 'item.completed', 'item': {'type': 'agent_message', 'text': response}})
    with patch.object(worker.subprocess, 'run', return_value=SimpleNamespace(returncode=0, stdout=output, stderr='')):
        with pytest.raises(RuntimeError):
            worker.run_review('goal', 'abc1234', 'codex')
    assert report.read_text() == 'Previous verified report'


def test_deployment_retry_reuses_review_without_model(board, monkeypatch):
    monkeypatch.setenv('AGENT_BUILDER_DEPLOY_QUEUE_DIR', str(board / 'deploy'))
    monkeypatch.setenv('BOARD_REPO_DIR', str(board))
    monkeypatch.setenv('BOARD_BUILDER_RESULTS_DIR', str(board / 'results'))
    trigger.save_state({'cards': {'goal': {'source': 'intent', 'phase': 'fixer_fehlgeschlagen',
        'failure_stage': 'deployment', 'review_decision': 'bereit_fuer_deployment', 'commit_hash': 'abc1234'}}})
    job_id = trigger.retry_deployment('goal')
    with patch.object(worker, 'run_review', side_effect=AssertionError('No new model run')):
        worker.process_job(board / 'inbox' / (job_id + '.json'), board / 'failed', board / 'done')
    request = json.loads(next((board / 'deploy').glob('*.json')).read_text())
    assert request['commit_hash'] == 'abc1234'
    assert trigger.load_state()['cards']['goal']['deployment_status'] == 'queued'
    assert not workflow.verified_deployment(trigger.load_state()['cards']['goal'])


def test_archived_parent_does_not_restart_children(board):
    import orchestration
    trigger.save_state({'cards': {'parent': {'phase': 'archiviert', 'orchestration_kind': 'operator', 'depends_on': ['goal']},
        'goal': {'phase': 'geplant', 'parent_id': 'parent', 'operator_generated': True}}})
    with patch.object(trigger, 'write_job', side_effect=AssertionError('Archived family must remain inactive')):
        orchestration.progress_all(trigger.write_job, trigger.COLUMN_CONFIG_DEFAULTS)
    assert trigger.load_state()['cards']['parent']['phase'] == 'archiviert'


def test_outbox_publishes_only_after_state_commit(board):
    target = board / 'inbox' / 'unique.json'
    with state_store.transaction():
        state = state_store.load()
        state['cards']['goal'] = {'last_run_status': 'queued'}
        state_store.save(state)
        state_store.write_payload(target, {'job_id': 'unique'})
        assert not target.exists()
    assert target.exists()
    assert state_store.load()['cards']['goal']['last_run_status'] == 'queued'


def test_transaction_rollback_does_not_publish_job(board):
    target = board / 'inbox' / 'rollback.json'
    with pytest.raises(RuntimeError):
        with state_store.transaction():
            state_store.write_payload(target, {'job_id': 'rollback'})
            raise RuntimeError('aborted')
    assert not target.exists()


def test_completion_is_idempotent_and_stale_pipeline_is_ignored(board):
    record = board / 'record.json'
    record.write_text(json.dumps({'job_id':'board-goal-1', 'board_slug':'goal', 'status':'no_changes', 'pipeline_run_id':'old'}))
    trigger.save_state({'cards': {'goal': {'phase':'geplant', 'pipeline_run_id':'new'}}})
    completion_watcher.process_record(record)
    completion_watcher.process_record(record)
    assert trigger.load_state()['cards']['goal']['phase'] == 'geplant'
    assert len(trigger.load_state()['processed_completions']) == 1


def test_blocked_children_override_finished_operator_job():
    child = {'id':'child', 'phase':'in_umsetzung', 'last_run_status':'no_changes'}
    display = workflow.presentation({'phase':'wartet_auf_kinder', 'last_run_status':'done'}, [child])
    assert display['group'] == 'attention'
    assert display['label'] == 'Blocked by children'
    assert display['blocking_children'] == ['child']


def test_negative_review_and_not_applicable_are_not_success():
    assert workflow.presentation({'phase':'planung_mit_feedback', 'last_run_status':'done'})['group'] == 'attention'
    assert workflow.presentation({'phase':'klaerung_reparatur', 'last_run_status':'not_applicable'})['group'] == 'attention'


def test_idea_refinement_preserves_identity(board):
    card_id = trigger.create_idea('A useful goal')
    assert trigger.plan_card(card_id) == card_id
    assert trigger.load_state()['cards'][card_id]['source'] == 'draft'