"""Offline cancellation/recovery regressions; no HTTP, NAS, secrets or model calls."""
import asyncio
import functools
import sqlite3
import threading

import pytest


def offline_async(function):
    @functools.wraps(function)
    def run(*args, **kwargs):
        return asyncio.run(function(*args, **kwargs))
    return run

from control.accounts import Accounts, token_hash
from control.chat_runs import ChatRuns
from control.chat_store import ChatConflict, ChatStore


class BlockedAdapter:
    available = True
    approvals = False

    def __init__(self, *, block_start=False, fail_stop=False):
        self.block_start = block_start
        self.fail_stop = fail_stop
        self.start_entered = asyncio.Event()
        self.start_release = asyncio.Event()
        self.stop_entered = asyncio.Event()
        self.stream_entered = asyncio.Event()
        self.stream_release = asyncio.Event()
        self.starts = []
        self.stops = []

    async def start(self, owner, rid, payload):
        self.starts.append(rid)
        self.start_entered.set()
        if self.block_start:
            await self.start_release.wait()
        return {'run_id': 'offline-upstream'}

    async def stop(self, owner, upstream):
        self.stops.append(upstream)
        self.stop_entered.set()
        if self.fail_stop:
            raise RuntimeError('Offline simulated unavailable stop')

    async def status(self, owner, upstream):
        return {'status': 'cancelled' if self.stops and not self.fail_stop else 'running'}

    async def events(self, owner, upstream):
        self.stream_entered.set()
        await self.stream_release.wait()
        yield {'event': 'run.cancelled'}


@pytest.fixture
def execution(tmp_path):
    accounts = Accounts(str(tmp_path / 'offline.sqlite'))
    user = accounts.create('review', 'Offline Review', 'user', 'offline-review-password')
    accounts.update(user['id'], status='ready')
    token, _ = accounts.issue_session(user['id'], 3600, user['password_hash'])
    store = ChatStore(accounts)
    cid = store.create_conversation(user['id'])['id']
    operation = store.send(user['id'], cid, 'Offline request', 'offline-key')
    yield accounts, store, user['id'], token, cid, operation['run']['id']
    accounts.close()


@offline_async
async def test_cancelled_sqlite_worker_is_joined_by_store_close(execution, monkeypatch, tmp_path):
    """Task cancellation is not thread completion: close must serialize with SQL."""
    accounts, store, owner, _, _, rid = execution
    entered = threading.Event()
    release = threading.Event()
    close_attempted = threading.Event()
    close_entered = threading.Event()
    original_lock, original_db = accounts.lock, accounts.db
    original_row = store._row
    closing_thread = threading.local()

    class ObservedLock:
        def __enter__(self):
            if getattr(closing_thread, 'closing', False):
                close_attempted.set()
            original_lock.acquire()
            return self

        def __exit__(self, *args):
            original_lock.release()

    class ObservedConnection:
        def close(self):
            close_entered.set()
            close_attempted.set()
            # Fail safely on the old implementation, without racing native SQL
            # close in the test process and potentially segfaulting pytest.
            assert original_lock._is_owned(), 'SQLite close must hold the shared store lock'
            original_db.close()

    def blocked_row(*args):
        entered.set()
        assert release.wait(5), 'Offline worker release timed out'
        return original_row(*args)

    def close():
        closing_thread.closing = True
        accounts.close()

    accounts.lock = store.lock = ObservedLock()
    monkeypatch.setattr(accounts, 'db', ObservedConnection())
    monkeypatch.setattr(store, '_row', blocked_row)
    worker = asyncio.create_task(asyncio.to_thread(store.append_event, owner, rid, 'done', {'offline': True}))
    closer = None
    try:
        assert await asyncio.to_thread(entered.wait, 2)
        worker.cancel()
        with pytest.raises(asyncio.CancelledError):
            await worker
        closer = asyncio.create_task(asyncio.to_thread(close))
        assert await asyncio.to_thread(close_attempted.wait, 2)
        assert not close_entered.is_set(), 'Close must wait for the cancelled task\'s SQLite worker'
    finally:
        release.set()
        try:
            if closer is not None:
                await closer
        finally:
            monkeypatch.setattr(accounts, 'db', original_db)
            accounts.lock = store.lock = original_lock
    assert close_entered.is_set()
    # The worker's terminal replay event committed before close; no test bypass
    # or change to the durable run/event contract is needed.
    with sqlite3.connect(str(tmp_path / 'offline.sqlite')) as reopened:
        assert reopened.execute('SELECT event, data FROM chat_events WHERE run_id=?', (rid,)).fetchall() == [('done', '{"offline": true}')]
    accounts.close()  # Teardown close remains idempotent.


def coordinator(execution, adapter, *, upstream=None):
    accounts, store, owner, token, cid, rid = execution
    runs = ChatRuns(accounts, store, adapter=adapter)
    runs.metadata(owner, rid, {'conversation_id': cid, 'payload': {'input': 'Offline request'},
                             'session_hash': token_hash(token), 'room': None,
                             'upstream_id': upstream, 'assistant_id': None})
    return runs


@offline_async
async def test_cancel_during_blocked_admission_stops_learned_upstream(execution):
    _, store, owner, _, _, rid = execution
    adapter = BlockedAdapter(block_start=True)
    runs = coordinator(execution, adapter)
    runs.schedule(owner, rid)
    try:
        await asyncio.wait_for(adapter.start_entered.wait(), 2)
        await runs.cancel(owner, rid)
        assert not runs.tasks[rid].done(), 'Do not abort admission before learning its upstream ID'
        adapter.start_release.set()
        await asyncio.wait_for(runs.tasks[rid], 2)
        assert adapter.starts == [rid]
        assert adapter.stops, 'Cancellation must stop the execution admitted during cancellation'
        assert store.get_run(owner, rid)['run']['status'] == 'cancelled'
    finally:
        adapter.start_release.set()
        await runs.stop()


@offline_async
async def test_stop_failure_keeps_execution_slot_and_durable_intent(execution):
    _, store, owner, _, _, rid = execution
    adapter = BlockedAdapter(fail_stop=True)
    runs = coordinator(execution, adapter, upstream='offline-upstream')
    store.update_run(owner, rid, 'running')
    runs.schedule(owner, rid)
    try:
        await asyncio.wait_for(adapter.stream_entered.wait(), 2)
        await runs.cancel(owner, rid)
        assert runs.metadata(owner, rid).get('stop_requested') is True
        assert store.get_run(owner, rid)['run']['status'] not in runs.store.TERMINAL
        assert runs.locks[owner].locked(), 'Do not admit new account work while old work is unconfirmed'
        assert not runs.tasks[rid].done()
    finally:
        await runs.stop()


@offline_async
async def test_revoked_session_recovery_stops_remote_before_terminalizing(execution):
    accounts, store, owner, token, _, rid = execution
    adapter = BlockedAdapter()
    runs = coordinator(execution, adapter, upstream='offline-upstream')
    store.update_run(owner, rid, 'running')
    accounts.revoke(token)
    try:
        await asyncio.wait_for(runs.execute(owner, rid), 2)
        assert adapter.starts == [], 'Recovery must never replay a known admission'
        assert adapter.stops, 'A revoked session does not mean its remote execution stopped'
        assert store.get_run(owner, rid)['run']['status'] == 'cancelled'
    finally:
        await runs.stop()


def test_stale_stream_metadata_cannot_erase_cancellation_intent(execution):
    _, _, owner, _, _, rid = execution
    runs = coordinator(execution, BlockedAdapter(), upstream='offline-upstream')
    stale = runs.metadata(owner, rid)
    cancel = runs.metadata(owner, rid)
    cancel['stop_requested'] = True
    runs.metadata(owner, rid, cancel)
    stale['content'] = 'A delta racing with cancellation'
    runs.metadata(owner, rid, stale)
    assert runs.metadata(owner, rid).get('stop_requested') is True


@offline_async
async def test_restart_after_cancel_during_admission_reconciles_remote(execution):
    _, store, owner, _, _, rid = execution
    adapter = BlockedAdapter(block_start=True, fail_stop=True)
    original = coordinator(execution, adapter)
    original.schedule(owner, rid)
    try:
        await asyncio.wait_for(adapter.start_entered.wait(), 2)
        await original.cancel(owner, rid)
        adapter.start_release.set()
        await asyncio.wait_for(adapter.stop_entered.wait(), 2)
    finally:
        adapter.start_release.set()
        await original.stop()
    # The remote run exists, its stop failed, and the local projection may already
    # be terminal. A restart must still reconcile that durable remote stop intent.
    recovered_adapter = BlockedAdapter()
    recovered = ChatRuns(execution[0], store, adapter=recovered_adapter)
    try:
        await recovered.start()
        await asyncio.wait_for(recovered_adapter.stop_entered.wait(), 2)
        if rid in recovered.tasks:
            await asyncio.wait_for(recovered.tasks[rid], 2)
        assert recovered_adapter.starts == []
        assert store.get_run(owner, rid)['run']['status'] == 'cancelled'
    finally:
        await recovered.stop()


@offline_async
async def test_delete_cannot_discard_unconfirmed_remote_cancellation(execution):
    _, store, owner, _, cid, rid = execution
    adapter = BlockedAdapter(fail_stop=True)
    runs = coordinator(execution, adapter, upstream='offline-upstream')
    store.update_run(owner, rid, 'running')
    runs.schedule(owner, rid)
    try:
        await asyncio.wait_for(adapter.stream_entered.wait(), 2)
        with pytest.raises(ChatConflict):
            await runs.cancel_conversation(owner, cid)
            store.delete_conversation(owner, cid)
        assert store.get_run(owner, rid)['run']['status'] not in store.TERMINAL
        assert runs.metadata(owner, rid).get('stop_requested') is True
    finally:
        await runs.stop()