"""An owner restart leaves a started admission `unknown`; a client must be able to acknowledge it."""
import asyncio
import queue
from types import SimpleNamespace

import pytest

from gateway.config import GatewayConfig, Platform
from gateway.session import SessionSource, SessionStore
from gateway.session_authority import LiveSession, initialize_session_authority
from gateway.session_contract import SessionRef, Submission
from gateway.session_controls import AuthorityConnection
from hermes_state_runtime import RuntimeStoreError, claim_session_input, list_session_admissions
from tests.waiting import async_wait_until


class Peer:
    def __init__(self):
        self.frames = queue.Queue()

    def write(self, frame):
        self.frames.put(frame)
        return True


async def _restarted_owner_with_unknown_head(tmp_path, monkeypatch, executed):
    """Real authority + SQLite: a started turn interrupted by an owner epoch change plus a queued follower."""
    monkeypatch.setenv('HERMES_HOME', str(tmp_path))
    store = SessionStore(tmp_path / 'sessions', GatewayConfig())
    db = store._db

    async def answer(event):
        from gateway.session_results import execution_result
        executed.append(event.text)
        execution_result.get()['result'] = {'final_response': 'ACK_' + event.text, 'completed': True, 'messages': []}
        return 'ACK_' + event.text

    runner = SimpleNamespace(_session_db=db, session_store=store, _draining=False,
                             _handle_message=answer, _adapter_for_source=lambda source: None)
    source = SessionSource(platform=Platform.TELEGRAM, chat_id='unknown-head')
    db.create_session('s', source='telegram')
    first = await initialize_session_authority(runner, profile_id='p', instance_id='owner-1')
    first.sessions['s'] = LiveSession(source, 's')
    ref = SessionRef('p', 's')
    actor = AuthorityConnection(first, Peer(), {'user_id': 'human'}).actor
    # Admit the head and claim it without draining: the owner dies mid-turn.
    first._schedule = lambda ref: None
    head = await first.submit(actor, Submission('head', ref, {'text': 'HEAD'}, 'queue'))
    follower = await first.submit(actor, Submission('follower', ref, {'text': 'FOLLOWER'}, 'queue'))
    claimed = claim_session_input(db, epoch=first.epoch, session_id='s')
    assert claimed['admission_id'] == head.admission_id
    authority = await initialize_session_authority(runner, profile_id='p', instance_id='owner-2')
    authority.sessions['s'] = LiveSession(source, 's')
    rows = {r['admission_id']: r for r in list_session_admissions(db, session_id='s')}
    assert rows[head.admission_id]['status'] == 'unknown' and rows[follower.admission_id]['status'] == 'queued'
    return store, authority, ref, rows[head.admission_id], follower


@pytest.mark.asyncio
async def test_resolve_unknown_releases_the_paused_follower_exactly_once(tmp_path, monkeypatch):
    executed = []
    store, authority, _ref, unknown, follower = await _restarted_owner_with_unknown_head(tmp_path, monkeypatch, executed)
    peer = Peer()
    viewer = AuthorityConnection(authority, peer, {'user_id': 'human'})
    try:
        resumed = (await viewer.dispatch({'id': 1, 'method': 'session.resume', 'params': {'session_id': 's'}}))['result']
        assert [r['status'] for r in resumed['pending']] == ['unknown', 'queued']
        # The follower stays paused behind the unknown head; nothing runs on its own. Only a drain
        # runs a turn, so once no drain is live (none scheduled, or it paused) none can run later.
        live = authority.sessions['s']
        await async_wait_until(lambda: live.task is None or live.task.done(), what='no live drain')
        assert executed == []
        reply = await viewer.dispatch({'id': 2, 'method': 'prompt.resolve_unknown', 'params': {
            'session_id': 's', 'admission_id': unknown['admission_id'],
            'execution_generation': unknown['generation']}})
        receipt = reply['result']
        assert receipt['status'] == 'terminal' and receipt['outcome'] == 'interrupted', reply
        await authority.sessions['s'].task
        assert executed == ['FOLLOWER']
        statuses = {r['admission_id']: (r['status'], r['outcome']) for r in
                    list_session_admissions(authority.db, session_id='s', pending_only=False)}
        assert statuses[follower.admission_id] == ('terminal', 'completed')
        # The viewer learns about the release through the ordinary pending fanout.
        infos = []
        while not infos or infos[-1]['params']['type'] != 'message.complete':
            infos.append(await asyncio.to_thread(peer.frames.get, True, 5))
        first_info = next(f for f in infos if f['params']['type'] == 'session.info')
        assert all(r['status'] != 'unknown' for r in first_info['params']['payload']['pending'])
        # Lost-ACK retries return the committed resolution without another turn.
        again = await viewer.dispatch({'id': 3, 'method': 'prompt.resolve_unknown', 'params': {
            'session_id': 's', 'admission_id': unknown['admission_id'],
            'execution_generation': unknown['generation']}})
        assert again['result'] == receipt
        assert executed == ['FOLLOWER']
    finally:
        await viewer.close()
        store.close_all_db_handles()


@pytest.mark.asyncio
async def test_resolve_unknown_is_fenced_by_generation_and_capability(tmp_path, monkeypatch):
    executed = []
    store, authority, _ref, unknown, follower = await _restarted_owner_with_unknown_head(tmp_path, monkeypatch, executed)
    owner = AuthorityConnection(authority, Peer(), {'user_id': 'human'})
    reader = AuthorityConnection(authority, Peer(), {'user_id': 'human', 'capabilities': ['session:read', 'session:submit']})
    try:
        for connection in (owner, reader):
            await connection.dispatch({'id': 1, 'method': 'session.resume', 'params': {'session_id': 's'}})
        params = {'session_id': 's', 'admission_id': unknown['admission_id'], 'execution_generation': unknown['generation']}
        denied = await reader.dispatch({'id': 2, 'method': 'prompt.resolve_unknown', 'params': params})
        assert denied['error']['data']['reason'] == 'permission_denied'
        stale = await owner.dispatch({'id': 3, 'method': 'prompt.resolve_unknown',
                                     'params': {**params, 'execution_generation': unknown['generation'] + 1}})
        assert stale['error']['data']['reason'] == 'stale_generation'
        foreign = await owner.dispatch({'id': 4, 'method': 'prompt.resolve_unknown',
                                       'params': {**params, 'session_id': 'other'}})
        assert 'error' in foreign
        live = authority.sessions['s']
        await async_wait_until(lambda: live.task is None or live.task.done(), what='no live drain')
        rows = {r['admission_id']: r['status'] for r in list_session_admissions(authority.db, session_id='s')}
        assert rows[unknown['admission_id']] == 'unknown' and rows[follower.admission_id] == 'queued'
        assert executed == []
    finally:
        await owner.close()
        await reader.close()
        store.close_all_db_handles()


@pytest.mark.asyncio
async def test_discard_and_its_transcript_boundary_commit_atomically_on_the_physical_tip(tmp_path, monkeypatch):
    """R1: a failure writing the boundary must not leave the lost admission terminal (its exact
    Discard retry refused, the follower claimable onto an open user tail). The boundary lands on
    the rotated physical transcript, not the logical owner."""
    store, authority, _ref, unknown, follower = await _restarted_owner_with_unknown_head(tmp_path, monkeypatch, [])
    db = authority.db
    db.create_session('s-tip', source='telegram', parent_session_id='s')
    db.end_session('s', 'compression')
    db.append_message('s-tip', 'user', 'LOST_OPERATION')
    authority._schedule = lambda ref: None
    db._execute_write(lambda c: c.execute("CREATE TRIGGER deny_boundary BEFORE INSERT ON messages "
                                          "WHEN NEW.role='assistant' BEGIN SELECT RAISE(ABORT,'boundary failed'); END"))
    viewer = AuthorityConnection(authority, Peer(), {'user_id': 'human'})
    params = {'session_id': 's', 'admission_id': unknown['admission_id'], 'execution_generation': unknown['generation']}
    try:
        await viewer.dispatch({'id': 1, 'method': 'session.resume', 'params': {'session_id': 's'}})
        failed = await viewer.dispatch({'id': 2, 'method': 'prompt.resolve_unknown', 'params': params})
        assert 'error' in failed, failed
        rows = {r['admission_id']: r['status'] for r in list_session_admissions(db, session_id='s')}
        assert rows.get(unknown['admission_id']) == 'unknown', 'Discard committed without its transcript boundary'
        with pytest.raises(RuntimeStoreError, match='unknown_execution'):
            claim_session_input(db, epoch=authority.epoch, session_id='s')
        db._execute_write(lambda c: c.execute('DROP TRIGGER deny_boundary'))
        retried = await viewer.dispatch({'id': 3, 'method': 'prompt.resolve_unknown', 'params': params})
        assert retried['result']['outcome'] == 'interrupted', retried
        assert db.latest_conversation_role('s-tip') == 'assistant'
        assert [m['role'] for m in db.get_messages('s') if m['role'] == 'assistant'] == []
        assert claim_session_input(db, epoch=authority.epoch, session_id='s')['admission_id'] == follower.admission_id
    finally:
        await viewer.close()
        store.close_all_db_handles()


@pytest.mark.asyncio
async def test_discarded_lost_turn_is_closed_and_never_merged_into_the_follower_request(tmp_path, monkeypatch):
    """Discard keeps the lost text visible for resending, but the next provider request must not
    glue it onto the follower (consecutive-user merge): it is closed like a failed turn."""
    from agent.turn_failure_copy import FAILED_TURN_DISPLAY_KIND, PARTIAL_FAILED_TURN_NOTICE
    from run_agent import AIAgent
    from tests.acp_adapter.test_failed_turn_closure import _LoopbackProvider
    monkeypatch.setenv('HERMES_DISABLE_PLUGINS', '1')
    monkeypatch.setenv('NO_PROXY', '127.0.0.1,localhost')
    store, authority, _ref, unknown, _follower = await _restarted_owner_with_unknown_head(tmp_path, monkeypatch, [])
    db = authority.db
    db.append_message('s', 'user', 'HEAD lost in restart')  # the lost turn's turn-start flush
    viewer = AuthorityConnection(authority, Peer(), {'user_id': 'human'})
    provider = _LoopbackProvider()
    try:
        await viewer.dispatch({'id': 1, 'method': 'session.resume', 'params': {'session_id': 's'}})
        reply = await viewer.dispatch({'id': 2, 'method': 'prompt.resolve_unknown', 'params': {
            'session_id': 's', 'admission_id': unknown['admission_id'],
            'execution_generation': unknown['generation']}})
        assert reply['result']['outcome'] == 'interrupted', reply
        await authority.sessions['s'].task
        rows = [m for m in db.get_messages('s') if m['role'] in ('user', 'assistant')]
        assert [(m['role'], m['content']) for m in rows[:2]] == [
            ('user', 'HEAD lost in restart'), ('assistant', PARTIAL_FAILED_TURN_NOTICE)]
        assert rows[1]['display_kind'] == FAILED_TURN_DISPLAY_KIND
        # The follower's real provider request, built from the live-replay transcript.
        history = [m for m in store.load_transcript('s') if m.get('content') != 'ACK_FOLLOWER'
                   and m.get('content') != 'FOLLOWER']
        provider.script.extend({'finish_reason': 'stop', 'content': 'ok'} for _ in range(4))
        agent = AIAgent(api_key='fixture-only', base_url=provider.base_url, provider='openai-compat',
                        model='fixture-model', max_iterations=2, enabled_toolsets=[], quiet_mode=True,
                        skip_context_files=True, skip_memory=True, save_trajectories=False, platform='cli')
        await asyncio.to_thread(agent.run_conversation, 'FOLLOWER', conversation_history=history)
        chat = [r for r in provider.requests if 'messages' in r]
        sent = [m for m in chat[0]['messages'] if m['role'] != 'system']
        assert [m['content'] for m in sent if m['role'] == 'user'] == ['HEAD lost in restart', 'FOLLOWER'], sent
        assert sent[1] == {'role': 'assistant', 'content': PARTIAL_FAILED_TURN_NOTICE}
    finally:
        provider.shutdown()
        await viewer.close()
        store.close_all_db_handles()