"""Room worker threads use the real authority ledger, never TUI dispatch."""
import asyncio
import threading
from types import SimpleNamespace

import pytest


@pytest.mark.parametrize('member_target', [{}, {'target': None}], ids=['missing-target', 'null-target'])
@pytest.mark.parametrize('revocation', [None, 'member', 'room_epoch', 'owner', 'task_generation', 'stopping', 'disbanded', 'malformed_target'])
def test_local_hosted_member_revoked_during_preparation_is_not_admitted(hosted_owner, monkeypatch, revocation, member_target):
    """Revocation must fence admission, not merely pause execution afterward."""
    import concurrent.futures
    import json
    from pathlib import Path
    import time

    from gateway import hosted_room_driver as tasks, session_hosted_attachments
    from gateway.hosted_rooms import create_room, local_authority_gateway_id
    from gateway.session_hosted_service import CanonicalHostedRoomService
    from hermes_state_runtime import RuntimeStoreError, list_session_admissions
    from tui_gateway.hosted_room_driver import HostedRoomBinding

    authority, loop, _, _ = hosted_owner
    service = CanonicalHostedRoomService(authority, loop)
    # The component fixture uses a synthetic profile identifier, not a daemon home.
    monkeypatch.setattr(service, 'profile_homes', lambda: {'default': Path(authority.profile_id)})
    service.authorize_room('alice', 'room', create=True)
    gateway = local_authority_gateway_id()
    members = [
        {'member_id': 'one', 'profile': 'default', 'handle': 'one', **member_target},
        {'member_id': 'two', 'profile': 'other', 'handle': 'two'},
    ]
    create_room(authority.db.db_path, room_id='room', name='Room',
                authority_gateway_id=gateway, members=members)
    identity = tasks.TaskIdentity('room', 'task', 'thread', 'turn')
    payload = {'target_profile': 'default', 'target_member_id': 'one',
               'source_event_seq': 1, 'prompt': 'frozen'}
    tasks.admit_task(authority.db.db_path, identity, payload=payload, clock=time.time)
    lease = tasks.acquire_lease(authority.db.db_path, room_id='room', gateway_id=gateway,
        authority_epoch=1, process_generation='test', ttl_seconds=120, clock=time.time)
    tasks.start_task(authority.db.db_path, identity, lease,
                     expected_cancel_generation=0, clock=time.time)
    task, = tasks.list_tasks(authority.db.db_path, room_id='room')
    rpc = service._resolve_member_transport(HostedRoomBinding('room', gateway, 1), task)
    coords = {'profile': 'default', 'source': 'bot_room'}
    sid = rpc.create(**coords, title='Group: room')['session_id']
    preparing, release = threading.Event(), threading.Event()
    original = session_hosted_attachments.submission_payload

    def paused_preparation(*args, **kwargs):
        preparing.set()
        assert release.wait(10), 'test did not release preparation'
        return original(*args, **kwargs)

    monkeypatch.setattr(session_hosted_attachments, 'submission_payload', paused_preparation)
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
        submitted = executor.submit(rpc.submit, **coords, session_id=sid, prompt='frozen',
            task=identity, execution_generation=task['execution_generation'],
            on_terminal=lambda receipt: None)
        try:
            assert preparing.wait(10), 'producer never reached preparation'
            if revocation is not None:
                if revocation == 'malformed_target':
                    members[0]['target'] = []
                else:
                    members[0] = {'member_id': 'replacement', 'profile': 'default', 'handle': 'replacement'}
                mutations = {
                    'member': ('UPDATE hosted_rooms SET members_json=? WHERE room_id=?', (json.dumps(members), 'room')),
                    'room_epoch': ('UPDATE hosted_rooms SET authority_epoch=2 WHERE room_id=?', ('room',)),
                    'owner': ('UPDATE state_meta SET value=? WHERE key=?', ('bob', 'gateway.hosted.owner.v1:room')),
                    'task_generation': ('UPDATE hosted_room_driver_tasks SET execution_generation=2 WHERE room_id=?', ('room',)),
                    'stopping': ("UPDATE hosted_room_driver_tasks SET status='stopping' WHERE room_id=?", ('room',)),
                    'disbanded': ('UPDATE hosted_rooms SET disbanded_at=? WHERE room_id=?', (time.time(), 'room')),
                    'malformed_target': ('UPDATE hosted_rooms SET members_json=? WHERE room_id=?', (json.dumps(members), 'room')),
                }
                sql, args = mutations[revocation]
                authority.db._execute_write(lambda conn: conn.execute(sql, args))
        finally:
            release.set()
        if revocation is None:
            receipt = submitted.result(timeout=10)
            retry = rpc.submit(**coords, session_id=sid, prompt='frozen', task=identity,
                execution_generation=task['execution_generation'], on_terminal=lambda receipt: None)
            assert retry['admission_id'] == receipt['admission_id']
        else:
            with pytest.raises(RuntimeStoreError, match='permission_denied'):
                submitted.result(timeout=10)
    rows = list_session_admissions(authority.db, session_id=sid, pending_only=False)
    assert len(rows) == (1 if revocation is None else 0)


def test_room_binding_exact_retry_terminal_history_and_unknown(hosted_owner):
    from gateway.session_hosted_rpc import HostedRoomAuthorityRPC
    from gateway.hosted_room_driver import TaskIdentity
    from hermes_state_runtime import RuntimeStoreError, list_session_admissions, claim_session_input, settle_session_input, begin_runtime_epoch, recover_session_inputs
    authority, loop, principal, _ = hosted_owner
    allowed = [True]
    seen = []
    def authorize(operation, task, generation):
        seen.append(threading.get_ident())
        return allowed[0]
    def adapter(room='room'):
        return HostedRoomAuthorityRPC(authority, loop, room_id=room, member_id='member', profile='default', principal=principal, authorize=authorize)
    rpc = adapter()
    coords = dict(profile='default', source='bot_room')
    assert rpc.resolve_exact(**coords, title='Group: room') is None
    sid = rpc.create(**coords, title='Group: room')['session_id']
    stored = authority.db.get_session(sid)
    assert stored['source'] == 'bot_room'
    assert stored['title'] == 'Group: room'
    assert stored['hidden']
    assert adapter().resolve_exact(**coords, title='Group: room')['session_id'] == sid
    with pytest.raises(RuntimeStoreError):
        adapter('other').resume(**coords, session_id=sid)
    args = dict(**coords, session_id=sid, prompt='input', task=TaskIdentity('room','task','thread','turn'), execution_generation=1)
    callbacks = []
    first = rpc.submit(**args, on_terminal=callbacks.append)
    assert rpc.submit(**args, on_terminal=callbacks.append)['admission_id'] == first['admission_id']
    with pytest.raises(RuntimeStoreError, match='admission_conflict'):
        rpc.submit(**{**args, 'prompt': 'changed'}, on_terminal=callbacks.append)
    row = claim_session_input(authority.db, epoch=authority.epoch, session_id=sid)
    settle_session_input(authority.db, epoch=authority.epoch, admission_id=row['admission_id'], generation=row['generation'], outcome='completed', result={'result': {'final_response': 'reply'}, 'usage': {}})
    history = rpc.history(**coords, session_id=sid)
    assert history[-1]['content'] == 'reply'
    assert history[-1]['task_id'] == 'task'
    assert history[-1]['execution_generation'] == 1
    assert callbacks and callbacks[0]['status'] == 'settled'
    args['task'] = TaskIdentity('room','unknown','thread','turn2')
    rpc.submit(**args, on_terminal=callbacks.append)
    claim_session_input(authority.db, epoch=authority.epoch, session_id=sid)
    authority.epoch = begin_runtime_epoch(authority.db, instance_id='cold')
    recover_session_inputs(authority.db, epoch=authority.epoch)
    with pytest.raises(RuntimeStoreError, match='unknown_execution') as exc:
        adapter().submit(**args, on_terminal=callbacks.append)
    assert not getattr(exc.value, 'not_admitted', False)
    assert len(list_session_admissions(authority.db, session_id=sid, pending_only=False)) == 2
    allowed[0] = False
    for operation in (lambda: rpc.history(**coords, session_id=sid), lambda: rpc.info(**coords, session_id=sid), lambda: rpc.create(**coords, title='Group: room')):
        with pytest.raises(RuntimeStoreError, match='permission_denied'):
            operation()
    assert seen and all(t != threading.get_ident() for t in seen)


def test_controls_are_exact_current_admission_and_loop_safe(hosted_owner):
    from gateway.session_hosted_rpc import HostedRoomAuthorityRPC
    from gateway.hosted_room_driver import TaskIdentity
    from hermes_state_runtime import RuntimeStoreError, claim_session_input
    authority, loop, principal, agent = hosted_owner
    rpc = HostedRoomAuthorityRPC(authority, loop, room_id='room', member_id='member', profile='default', principal=principal, authorize=lambda *args: True)
    coords = dict(profile='default', source='bot_room')
    sid = rpc.create(**coords, title='Group: room')['session_id']
    rpc.submit(**coords, session_id=sid, prompt='input', task=TaskIdentity('room','task','thread','turn'), execution_generation=1, on_terminal=lambda receipt: None)
    row = claim_session_input(authority.db, epoch=authority.epoch, session_id=sid)
    assert rpc.info(**coords, session_id=sid)['task_id'] == 'task'
    live = authority.sessions[sid]
    answers = []
    authority.register_approval(sid, row['generation'], live.route, {'request_id': 'approve-me', 'command': 'owned fixture'})
    live.controls.remote_responders['approve-me'] = lambda *answer: answers.append(answer)
    pending = rpc.info(**coords, session_id=sid)['pending_approval']
    assert pending['request_id'] == 'approve-me'
    assert pending['choices'] == ['once', 'deny']
    with pytest.raises(RuntimeStoreError, match='invalid_params'):
        rpc.approve(session_id=sid, request_id='approve-me', choice='always')
    assert not answers
    assert rpc.approve(session_id=sid, request_id='approve-me', choice='once')['status'] == 'resolved'
    assert answers == [('approval', 'approve-me', 'once')]
    with pytest.raises(RuntimeStoreError, match='stale_generation'):
        rpc.interrupt(**coords, session_id=sid, expected_task_id='other')
    assert not agent.interrupted
    assert rpc.interrupt(**coords, session_id=sid, expected_task_id='task')['interrupted']
    assert agent.interrupted
    with pytest.raises(RuntimeStoreError):
        rpc.approve(session_id=sid, request_id='missing', choice='once')
    async def same_loop():
        with pytest.raises(RuntimeStoreError, match='invalid_params'):
            rpc.info(**coords, session_id=sid)
    asyncio.run_coroutine_threadsafe(same_loop(), loop).result()


def test_terminal_callback_follows_canonical_drain_without_polling(hosted_owner, monkeypatch):
    from gateway.session_hosted_rpc import HostedRoomAuthorityRPC
    from gateway.session_authority import SessionAuthority
    from gateway.hosted_room_driver import TaskIdentity
    from gateway import session_finite
    from hermes_state_runtime import list_session_admissions
    authority, loop, principal, _ = hosted_owner
    rpc = HostedRoomAuthorityRPC(authority, loop, room_id='room', member_id='member', profile='default', principal=principal, authorize=lambda *args: True)
    coords = dict(profile='default', source='bot_room')
    sid = rpc.create(**coords, title='Group: room')['session_id']
    async def execute(authority, ref, row):
        return 'canonical reply'
    # This RPC component fixture supplies the service gate explicitly; daemon
    # coverage exercises the real durable member/task authorizer.
    authority.runner._adapter_for_source = lambda source: authority.runner.adapters[source.platform]
    authority.hosted_room_service = SimpleNamespace(check_admission=lambda ref, row: True)
    monkeypatch.setattr(session_finite, 'execute_finite_admission', execute)
    monkeypatch.setattr(authority, '_schedule', SessionAuthority._schedule.__get__(authority))
    done, receipts = threading.Event(), []
    def terminal(receipt):
        receipts.append(receipt)
        done.set()
    rpc.submit(**coords, session_id=sid, prompt='input', task=TaskIdentity('room','task','thread','turn'), execution_generation=1, on_terminal=terminal)
    assert done.wait(3), 'terminal callback requires no driver polling'
    assert receipts[0]['text'] == 'canonical reply'
    rows = list_session_admissions(authority.db, session_id=sid, pending_only=False)
    assert rows[0]['status'] == 'terminal'


def test_unknown_info_is_inactive_with_hosted_not_canonical_generation(hosted_owner):
    from gateway.session_hosted_rpc import HostedRoomAuthorityRPC
    from gateway.hosted_room_driver import TaskIdentity
    from hermes_state_runtime import claim_session_input, begin_runtime_epoch, recover_session_inputs
    authority, loop, principal, _ = hosted_owner
    rpc = HostedRoomAuthorityRPC(authority, loop, room_id='room', member_id='member', profile='default', principal=principal, authorize=lambda *args: True)
    coords = dict(profile='default', source='bot_room')
    coords['session_id'] = rpc.create(**coords, title='Group: room')['session_id']
    rpc.submit(**coords, prompt='lost', task=TaskIdentity('room', 'lost', 'thread', 'turn'), execution_generation=17, on_terminal=lambda receipt: None)
    row = claim_session_input(authority.db, epoch=authority.epoch, session_id=coords['session_id'])
    assert row['generation'] != 17
    authority.epoch = begin_runtime_epoch(authority.db, instance_id='cold-info')
    recover_session_inputs(authority.db, epoch=authority.epoch)
    info = rpc.info(**coords)
    assert info['active'] is False
    assert info['status'] == 'unknown'
    assert info['task_id'] == 'lost'
    assert info['execution_generation'] == 17


def test_discard_requires_exact_owned_unknown_tuple_without_replay(hosted_owner, monkeypatch):
    from gateway.session_hosted_rpc import HostedRoomAuthorityRPC
    from gateway.hosted_room_driver import TaskIdentity
    from hermes_state_runtime import RuntimeStoreError, list_session_admissions, claim_session_input, begin_runtime_epoch, recover_session_inputs
    authority, loop, principal, _ = hosted_owner
    allowed = [True]
    rpc = HostedRoomAuthorityRPC(authority, loop, room_id='room', member_id='member', profile='default', principal=principal, authorize=lambda *args: allowed[0])
    coords = dict(profile='default', source='bot_room')
    sid = rpc.create(**coords, title='Group: room')['session_id']
    coords['session_id'] = sid
    callbacks = []
    rpc.submit(**coords, prompt='lost', task=TaskIdentity('room', 'lost', 'thread', 'turn'), execution_generation=17, on_terminal=callbacks.append)
    exact = dict(expected_task_id='lost', execution_generation=17)
    with pytest.raises(RuntimeStoreError, match='stale_generation'):
        rpc.discard(**coords, **exact)  # queued is not unknown
    row = claim_session_input(authority.db, epoch=authority.epoch, session_id=sid)
    with pytest.raises(RuntimeStoreError, match='stale_generation'):
        rpc.discard(**coords, **exact)  # started is not unknown
    follower = rpc.submit(**coords, prompt='next', task=TaskIdentity('room', 'next', 'thread', 'next'), execution_generation=18, on_terminal=callbacks.append)
    authority.epoch = begin_runtime_epoch(authority.db, instance_id='cold-discard')
    recover_session_inputs(authority.db, epoch=authority.epoch)
    before = list_session_admissions(authority.db, session_id=sid, pending_only=False)
    for changed in ({'expected_task_id': 'other'}, {'execution_generation': row['generation']}, {'execution_generation': True}):
        with pytest.raises(RuntimeStoreError, match='stale_generation|invalid_params'):
            rpc.discard(**coords, **{**exact, **changed})
    allowed[0] = False
    with pytest.raises(RuntimeStoreError, match='permission_denied'):
        rpc.discard(**coords, **exact)
    allowed[0] = True
    with pytest.raises(RuntimeStoreError, match='permission_denied'):
        rpc.discard(**{**coords, 'session_id': 'foreign'}, **exact)
    authority.db._execute_write(lambda conn: conn.execute('UPDATE session_admissions SET principal_id=? WHERE admission_id=?', ('foreign', row['admission_id'])))
    with pytest.raises(RuntimeStoreError, match='stale_generation'):
        rpc.discard(**coords, **exact)
    authority.db._execute_write(lambda conn: conn.execute('UPDATE session_admissions SET principal_id=? WHERE admission_id=?', (principal.subject, row['admission_id'])))
    assert list_session_admissions(authority.db, session_id=sid, pending_only=False) == before
    assert not callbacks
    scheduled = []
    monkeypatch.setattr(authority, '_schedule', scheduled.append)
    receipt = rpc.discard(**coords, **exact)
    assert receipt['discarded'] is True
    assert receipt['task_id'] == 'lost'
    assert receipt['execution_generation'] == 17
    after = list_session_admissions(authority.db, session_id=sid, pending_only=False)
    assert [(r['admission_id'], r['status']) for r in after] == [(row['admission_id'], 'terminal'), (follower['admission_id'], 'queued')]
    assert after[0]['outcome'] == 'interrupted'
    assert scheduled == [rpc.ref]
    assert rpc.discard(**coords, **exact) == receipt
    assert scheduled == [rpc.ref]
    assert rpc.info(**coords)['task_id'] == 'next'
    assert rpc.history(**coords)[-1]['status'] == 'cancelled'


def test_queued_cancellation_is_a_cancelled_receipt_not_storage_unavailable(hosted_owner):
    """A queued task that is stopped never ran, so it has no execution result; history/info
    must project it as cancelled instead of failing the whole room read. A completed row with
    a missing result is still storage corruption."""
    from gateway.session_hosted_rpc import HostedRoomAuthorityRPC
    from gateway.hosted_room_driver import TaskIdentity
    from hermes_state_runtime import RuntimeStoreError, claim_session_input, settle_session_input
    authority, loop, principal, _ = hosted_owner
    rpc = HostedRoomAuthorityRPC(authority, loop, room_id='room', member_id='member', profile='default', principal=principal, authorize=lambda *args: True)
    coords = dict(profile='default', source='bot_room')
    sid = rpc.create(**coords, title='Group: room')['session_id']
    receipts = []
    rpc.submit(**coords, session_id=sid, prompt='input', task=TaskIdentity('room', 'task', 'thread', 'turn'), execution_generation=1, on_terminal=receipts.append)
    assert rpc.interrupt(**coords, session_id=sid, expected_task_id='task')['interrupted']
    history = rpc.history(**coords, session_id=sid)
    assert history[-1]['status'] == 'cancelled'
    assert history[-1]['task_id'] == 'task'
    assert rpc.info(**coords, session_id=sid)['status'] == 'idle'
    assert receipts and receipts[0]['status'] == 'cancelled'
    # Negative control: a completed admission whose result row vanished is still unreadable.
    rpc.submit(**coords, session_id=sid, prompt='input', task=TaskIdentity('room', 'task2', 'thread', 'turn2'), execution_generation=2, on_terminal=receipts.append)
    row = claim_session_input(authority.db, epoch=authority.epoch, session_id=sid)
    settle_session_input(authority.db, epoch=authority.epoch, admission_id=row['admission_id'], generation=row['generation'], outcome='completed', result=None)
    with pytest.raises(RuntimeStoreError, match='storage_unavailable'):
        rpc.history(**coords, session_id=sid)


def test_room_member_turns_cannot_block_on_an_unanswerable_clarify(hosted_owner, monkeypatch):
    """A room has no clarification reply and hosted principals lack session:respond, so a
    member turn that called clarify would block for clarify_timeout with its question never
    shown. The frozen member policy must exclude the tool even when the profile's toolsets
    (here the hermes-cli bundle) include it; the member asks in its room reply instead."""
    from gateway import run
    from gateway.session_hosted_rpc import HostedRoomAuthorityRPC
    from gateway.session_local_recovery import local_adapter_map
    from gateway.config import Platform
    from agent.skill_utils import parse_config_string_list
    from model_tools import get_tool_definitions

    authority, loop, principal, _ = hosted_owner
    monkeypatch.setattr(run, '_load_gateway_config', lambda: {
        'model': {'default': 'fixture'}, 'platform_toolsets': {'cli': ['hermes-cli']}})
    rpc = HostedRoomAuthorityRPC(authority, loop, room_id='room', member_id='one', profile='default',
                                 principal=principal, authorize=lambda *args: True)
    sid = rpc.create(profile='default', title='Group: room', source='bot_room')['session_id']
    policy = local_adapter_map(authority)[Platform.LOCAL].policies[sid]
    # The same derivation the turn uses: frozen toolsets minus the frozen agent.disabled_toolsets.
    disabled = parse_config_string_list((policy.config().get('agent') or {}).get('disabled_toolsets'))
    tools = {t['function']['name'] for t in get_tool_definitions(
        enabled_toolsets=list(policy.toolsets), disabled_toolsets=disabled or None, quiet_mode=True)}
    assert 'terminal' in tools
    assert 'clarify' not in tools


def test_history_read_during_a_room_session_delete_is_not_found(hosted_owner, monkeypatch):
    """A delete whose commit lands while the history read is off the loop, before its retirement runs,
    answers ``not_found``: the read waits for the delete's mutation lock before it uses the entry."""
    from gateway import session_local_recovery, session_mutations
    from gateway.session_contract import SessionRef
    from gateway.session_hosted_rpc import HostedRoomAuthorityRPC
    from gateway.session_mutations import mutate_session
    from hermes_state_runtime import RuntimeStoreError
    authority, loop, principal, _ = hosted_owner
    rpc = HostedRoomAuthorityRPC(authority, loop, room_id='room', member_id='member', profile='default',
                                 principal=principal, authorize=lambda *args: True)
    coords = dict(profile='default', source='bot_room')
    sid = rpc.create(**coords, title='Group: room')['session_id']
    ref = SessionRef(authority.profile_id, sid)
    prepare, write, deletes = session_local_recovery.prepare_history, session_mutations.mutate_runtime_session, []
    committed, release = threading.Event(), threading.Event()

    def write_then_hold(*args, **kwargs):
        result = write(*args, **kwargs)
        if kwargs.get('operation') == 'delete':
            committed.set()
            assert release.wait(10), 'the delete was never released'
        return result

    async def prepare_then_delete(authority_, ref_):
        prepared = await prepare(authority_, ref_)
        handle = authority._handle(ref)
        deletes.append(asyncio.ensure_future(mutate_session(authority, principal, ref, {
            'session_id': sid, 'request_id': 'delete', 'operation': 'delete', 'payload': {},
            'expected_revision': handle.revision, 'expected_generation': handle.execution_generation})))
        assert await asyncio.to_thread(committed.wait, 10)
        assert authority.db.get_session(sid) is None and sid in authority.sessions
        release.set()  # the commit has landed; the retirement runs once the loop gets to it
        return prepared
    monkeypatch.setattr(session_mutations, 'mutate_runtime_session', write_then_hold)
    monkeypatch.setattr(session_local_recovery, 'prepare_history', prepare_then_delete)
    try:
        with pytest.raises(RuntimeStoreError, match='not_found'):
            rpc.history(**coords, session_id=sid)
    finally:
        release.set()

    async def deleted():
        return await deletes[0]
    asyncio.run_coroutine_threadsafe(deleted(), loop).result(10)
    assert sid not in authority.sessions