"""Live hot-serve contract of a real ``python -m gateway.run`` multiplexer (Linux; loopback model).

A profile created under the running multiplexer is served by it without a restart: ``identify``
lists the new home, a ticket minted for it opens a session owned by this process, and ``profiles/``
(not the boot-time reservation) is what the next restart reserves — so ``hermes -p new`` finds an
owner instead of spawning a stray per-profile daemon that would later collide with the
multiplexer's all-or-nothing reserve. A secondary whose ``state.db`` is unusable parks only itself.
"""
import asyncio
import contextlib
import json
import os
from pathlib import Path
import signal
import socket
import subprocess
import sys
import time

import pytest
from websockets.asyncio.client import connect

from tests.waiting import wait_until

ROOT = Path(__file__).resolve().parents[2]
# The daemon is a foreground child of the test (never detached), boots a disposable custom root
# with HOME redirected (no systemd unit, no webhook port) and is reaped in every path below.
pytestmark = [pytest.mark.platforms("linux"), pytest.mark.spawns_gateway_lookalike]


def control(home, verb, params=None, timeout=10):
    request = json.dumps({'protocol': 1, 'id': 1, 'verb': verb, 'params': params or {}}).encode() + b'\n'
    from gateway.control_socket import resolve_client_socket_path
    path = resolve_client_socket_path(home)
    if path is None:
        raise FileNotFoundError('control socket absent')
    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as peer:
        peer.settimeout(timeout)
        peer.connect(str(path))
        peer.sendall(request)
        with peer.makefile('rb') as stream:
            raw = stream.readline()
    reply = json.loads(raw.split(b'\n', 1)[0])
    assert reply.get('ok'), reply
    return reply['result']


@pytest.fixture
def peer():
    from http.server import ThreadingHTTPServer
    import threading
    from tests.gateway.fixtures.shared_authority_peer import ModelPeer
    server = ThreadingHTTPServer(('127.0.0.1', 0), ModelPeer)
    server.requests, server.metadata_requests = [], []
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    try:
        yield server
    finally:
        server.shutdown()
        server.server_close()
        thread.join(timeout=5)


@pytest.fixture
def mux(tmp_path, peer):
    """Disposable custom root (``<root>/profiles/<name>`` layout, as in Docker) with
    ``multiplex_profiles`` on, plus one boot-time secondary."""
    user = tmp_path / 'user'
    user.mkdir()
    root = tmp_path / 'state'
    root.mkdir(mode=0o700)
    url = f'http://127.0.0.1:{peer.server_port}/v1'
    config = {'gateway': {'multiplex_profiles': True},
              'model': {'provider': 'custom', 'default': 'local-wire-stub', 'base_url': url},
              'auxiliary': {'title_generation': {'enabled': False}}}
    (root / 'config.yaml').write_text(json.dumps(config))
    boot = root / 'profiles' / 'boot'
    boot.mkdir(parents=True, mode=0o700)
    (boot / 'config.yaml').write_text(json.dumps(config))
    env = {k: os.environ[k] for k in ('PATH', 'LANG', 'TZ') if k in os.environ}
    env.update(HOME=str(user), USERPROFILE=str(user), HERMES_HOME=str(root), PYTHONPATH=str(ROOT),
               PYTHONUNBUFFERED='1', OPENAI_API_KEY='loopback-only', OPENAI_BASE_URL=url)
    return {'root': root, 'boot': boot, 'env': env, 'config': config, 'log': tmp_path / 'gateway.log'}


def start_daemon(mux, timeout=90, *, extra_args=()):
    """Boot the multiplexer the way ``hermes gateway run`` does; the canonical gateway command-line
    matcher (``gateway.status.looks_like_gateway_command_line``) is what the CLI's liveness probes
    apply to the recorded PID, so the fixture must spawn a recognisable launcher."""
    log = mux['log'].open('a+', encoding='utf-8', errors='replace')
    proc = subprocess.Popen([sys.executable, '-m', 'hermes_cli.main', 'gateway', 'run', '--quiet', *extra_args], cwd=ROOT,
                            env=mux['env'], stdin=subprocess.DEVNULL, stdout=log, stderr=subprocess.STDOUT)
    desc = {}
    deadline = time.monotonic() + timeout
    while proc.poll() is None and time.monotonic() < deadline:
        try:
            desc = control(mux['root'], 'identify', timeout=2)
            if desc.get('state') == 'ready':
                break
        except (OSError, ValueError, AssertionError):
            pass
        time.sleep(.1)
    log.flush()
    assert desc.get('state') == 'ready', (desc, proc.poll(), tail(mux))
    return proc, desc


def tail(mux, n=6000):
    """stdout capture plus the daemon's own ``logs/gateway.log`` (``--quiet`` sends records there)."""
    texts = [p.read_text(encoding='utf-8', errors='replace') for p in (mux['log'], mux['root'] / 'logs' / 'gateway.log')
             if p.exists()]
    return '\n'.join(t[-n:] for t in texts)


def stop(proc, expect=0):
    if proc.poll() is None:
        proc.send_signal(signal.SIGINT)
        try:
            code = proc.wait(timeout=45)
        except subprocess.TimeoutExpired:
            proc.kill()
            proc.wait(timeout=10)
            raise
        if expect is not None:
            assert code == expect, code


def served_homes(desc):
    return {Path(entry['home']).resolve() for entry in desc.get('served_profiles', [])}


def recorded_served(root):
    return json.loads((root / 'gateway_state.json').read_text(encoding='utf-8')).get('served_profiles')


def wait_recorded(root, predicate, what):
    """gateway_state.json is written by the runtime-status writer thread after the control reply
    (publish_runtime_status enqueues, never waits), so the recorded set is eventually consistent."""
    wait_until(lambda: predicate(recorded_served(root) or []), timeout=20, interval=.05, what=what)
    return recorded_served(root)


async def open_session_for(root, home, desc):
    """A ticket bound to *home* opens a session the multiplexer commits into *home*'s state.db."""
    grant = control(root, 'session-ticket', {'profile_id': str(home.resolve()),
                                             'instance_id': desc['instance_id'], 'purpose': 'interactive'})
    url = desc['api_origin'].replace('http:', 'ws:') + '/api/ws'
    async with connect(url, subprotocols=['hermes-gateway-v1', 'hermes-gateway-ticket.' + grant['ticket']]) as ws:
        await ws.send(json.dumps({'jsonrpc': '2.0', 'id': 1, 'method': 'session.create',
                                  'params': {'request_id': 'hot-served', 'source': 'cli'}}))
        async with asyncio.timeout(20):
            while True:
                reply = json.loads(await ws.recv())
                if reply.get('id') == 1:
                    assert 'result' in reply, reply
                    return reply['result']['session_id']


def test_profile_created_under_live_multiplexer_is_served_and_owned(mux):
    root, env = mux['root'], mux['env']
    proc, desc = start_daemon(mux)
    try:
        assert served_homes(desc) == {root.resolve(), mux['boot'].resolve()}
        # ``hermes profile create`` under the multiplexer's root: the CLI signals the control socket.
        created = subprocess.run([sys.executable, '-m', 'hermes_cli.main', 'profile', 'create', 'fresh'],
                                 cwd=ROOT, env=env, capture_output=True, text=True, timeout=120, check=False)
        assert created.returncode == 0, (created.stdout, created.stderr)
        fresh = root / 'profiles' / 'fresh'
        assert fresh.is_dir()
        deadline = time.monotonic() + 20
        while time.monotonic() < deadline and fresh.resolve() not in served_homes(desc):
            time.sleep(.2)
            desc = control(root, 'identify')
        assert fresh.resolve() in served_homes(desc), (desc, tail(mux))
        assert wait_recorded(root, lambda r: r == ['default', 'boot', 'fresh'], 'fresh recorded') == ['default', 'boot', 'fresh']
        # The new profile's sessions are this multiplexer's: same instance, its own state.db.
        sid = asyncio.run(open_session_for(root, fresh, desc))
        import sqlite3
        with sqlite3.connect(f"file:{fresh / 'state.db'}?mode=ro", uri=True) as db:
            assert db.execute('SELECT instance_id FROM runtime_epoch').fetchone() == (desc['instance_id'],)
            assert db.execute('SELECT 1 FROM sessions WHERE id=?', (sid,)).fetchone()
        # ``hermes -p fresh`` resolves to the multiplexer instead of spawning a daemon of its own.
        ensured = subprocess.run([sys.executable, '-m', 'hermes_cli.main', '-p', 'fresh', 'gateway', 'ensure',
                                  '--timeout', '15'], cwd=ROOT, env=env, capture_output=True, text=True, timeout=60, check=False)
        payload = json.loads(ensured.stdout.strip().splitlines()[-1])
        assert payload['state'] == 'ready' and payload['endpoint']['instance_id'] == desc['instance_id'], (payload, ensured.stderr)
        assert not (fresh / 'gateway.pid').exists()
        stop(proc)
        # The restart reserves what exists now, the hot-served profile included: no OwnershipConflict.
        proc, desc = start_daemon(mux)
        assert served_homes(desc) == {root.resolve(), mux['boot'].resolve(), fresh.resolve()}
        assert 'Cannot reserve gateway profiles' not in tail(mux)
    finally:
        stop(proc, expect=None)


def kill_recorded_gateway(root, timeout=45):
    """SIGINT the gateway whose PID ``gateway_state.json`` records (a detached spawn is not our child)."""
    pid = json.loads((root / 'gateway_state.json').read_text(encoding='utf-8')).get('pid')
    if pid:
        with contextlib.suppress(ProcessLookupError):
            os.kill(pid, signal.SIGINT)
        def gone():
            try:
                os.kill(pid, 0)
            except ProcessLookupError:
                return True
            return False
        with contextlib.suppress(AssertionError):
            wait_until(gone, timeout=timeout, interval=.2, what=f'gateway {pid} exited')
            return
        with contextlib.suppress(ProcessLookupError):
            os.kill(pid, signal.SIGKILL)


@pytest.mark.live_system_guard_bypass
def test_named_profile_ensure_during_multiplexer_downtime_starts_the_multiplexer(mux):
    """The multiplexer is down (restart/update) and ``hermes -p boot gateway ensure`` runs: the client
    must start the MULTIPLEXER, never a per-profile daemon (which would answer this client and then
    block the multiplexer's next all-or-nothing reserve with ``Cannot reserve gateway profiles``)."""
    root, env, boot = mux['root'], mux['env'], mux['boot']
    proc, desc = start_daemon(mux)
    stop(proc)
    assert recorded_served(root) == ['default', 'boot']
    try:
        ensured = subprocess.run([sys.executable, '-m', 'hermes_cli.main', '-p', 'boot', 'gateway', 'ensure',
                                  '--json', '--timeout', '90'], cwd=ROOT, env=env, capture_output=True, text=True, timeout=120, check=False)
        payload = json.loads(ensured.stdout.strip().splitlines()[-1])
        assert payload['state'] == 'ready', (payload, ensured.stderr)
        # The answering owner is the multiplexer at the root, serving boot as a secondary.
        assert Path(payload['endpoint']['control_home']).resolve() == root.resolve(), payload
        assert served_homes(control(root, 'identify')) == {root.resolve(), boot.resolve()}
        assert not (boot / 'gateway.pid').exists() and not (boot / 'gateway.sock').exists()
    finally:
        kill_recorded_gateway(root)
    # A fresh multiplexer boot reserves every profile: nothing stray owns boot's home.
    proc, desc = start_daemon(mux)
    try:
        assert served_homes(desc) == {root.resolve(), boot.resolve()}
        assert 'Cannot reserve gateway profiles' not in tail(mux)
        stop(proc)
    finally:
        stop(proc, expect=None)


def test_broken_secondary_state_db_parks_only_that_profile(mux):
    root = mux['root']
    broken = root / 'profiles' / 'broken'
    broken.mkdir(parents=True, mode=0o700)
    (broken / 'config.yaml').write_text(json.dumps(mux['config']))
    # Not a SQLite file and not the zero-byte case the quarantine repairs: an unopenable store.
    (broken / 'state.db').mkdir()
    proc, desc = start_daemon(mux)
    try:
        assert served_homes(desc) == {root.resolve(), mux['boot'].resolve()}, desc
        assert recorded_served(root) == ['default', 'boot']
        assert "Profile 'broken' not served" in tail(mux), tail(mux)
        # The park is published (identify + gateway_state.json), so the profile's clients fail fast
        # with the real reason instead of waiting the whole deadline for a service that never comes.
        assert 'broken' in desc.get('parked_profiles', {}), desc
        recorded = json.loads((root / 'gateway_state.json').read_text(encoding='utf-8'))
        assert 'broken' in recorded.get('parked_profiles', {}), recorded
        started = time.monotonic()
        ensured = subprocess.run([sys.executable, '-m', 'hermes_cli.main', '-p', 'broken', 'gateway', 'ensure',
                                  '--json', '--timeout', '30'], cwd=ROOT, env=mux['env'], capture_output=True,
                                 text=True, timeout=60, check=False)
        payload = json.loads(ensured.stdout.strip().splitlines()[-1])
        assert (payload['state'], payload['reason_code']) == ('inaccessible', 'profile_parked'), (payload, ensured.stderr)
        assert 'unusable' in (payload.get('detail') or ''), payload
        assert ensured.returncode == 7 and time.monotonic() - started < 20, (ensured.returncode, ensured.stderr)
        stop(proc)
    finally:
        stop(proc, expect=None)


def test_invalid_hot_profile_config_is_parked_and_unserved(mux):
    """A runtime MultiplexConfigError withdraws both new and previously served secondaries.

    This exercises the real gateway runtime: authority/ticket publication happens before adapter
    config validation, so the regression left an invalid profile routable even though its adapters
    never started.
    """
    root = mux['root']
    proc, desc = start_daemon(mux)
    try:
        invalid = json.loads(json.dumps(mux['config']))
        invalid['whatsapp'] = {'enabled': True, 'dm_policy': 'open'}

        added = root / 'profiles' / 'invalid-hot'
        added.mkdir(parents=True, mode=0o700)
        (added / 'config.yaml').write_text(json.dumps(invalid), encoding='utf-8')
        result = control(root, 'rescan-profiles')
        desc = control(root, 'identify')
        assert result['parked'] == ['invalid-hot'], (result, tail(mux))
        assert added.resolve() not in served_homes(desc), desc
        assert 'invalid-hot' in desc.get('parked_profiles', {}), desc
        wait_recorded(root, lambda r: 'invalid-hot' not in r, 'invalid-hot never recorded')

        # The same contract applies to a secondary that was already published and then edited
        # into an invalid configuration.
        assert mux['boot'].resolve() in served_homes(desc)
        (mux['boot'] / 'config.yaml').write_text(json.dumps(invalid), encoding='utf-8')
        result = control(root, 'rescan-profiles')
        desc = control(root, 'identify')
        # An explicit control-socket rescan retries parked profiles by design, so the still-invalid
        # 'invalid-hot' is re-parked alongside 'boot' (never served in between).
        assert sorted(result['parked']) == ['boot', 'invalid-hot'], (result, tail(mux))
        assert added.resolve() not in served_homes(desc), desc
        assert mux['boot'].resolve() not in served_homes(desc), desc
        assert 'boot' in desc.get('parked_profiles', {}), desc
        wait_recorded(root, lambda r: 'boot' not in r, 'parked boot withdrawn from the recorded set')
    finally:
        stop(proc, expect=None)


async def chat_for(root, home, desc, *, during_turn=None):
    """Exercise the real session/agent path; only the model HTTP wire is a fixture."""
    grant = control(root, 'session-ticket', {'profile_id': str(home.resolve()),
                    'instance_id': desc['instance_id'], 'purpose': 'interactive'})
    url = desc['api_origin'].replace('http:', 'ws:') + '/api/ws'
    async with connect(url, subprotocols=['hermes-gateway-v1', 'hermes-gateway-ticket.' + grant['ticket']]) as ws:
        async def until(predicate):
            async with asyncio.timeout(40):
                while True:
                    reply = json.loads(await ws.recv())
                    if predicate(reply):
                        return reply

        await ws.send(json.dumps({'jsonrpc': '2.0', 'id': 1, 'method': 'session.create',
            'params': {'request_id': 'profile-stop-chat', 'source': 'cli'}}))
        created = await until(lambda r: r.get('id') == 1)
        assert 'result' in created, created
        sid = created['result']['session_id']
        # Keep the same connection and conversation for a second turn after the stop.
        for index in range(2):
            text = 'WS_SHARED BLOCK_FIFO' if during_turn and index == 0 else 'WS_SHARED'
            await ws.send(json.dumps({'jsonrpc': '2.0', 'id': index + 2, 'method': 'prompt.submit',
                'params': {'session_id': sid, 'text': text, 'input_id': f'profile-stop-{index}'}}))
            submitted = await until(lambda r: r.get('id') == index + 2)
            assert 'result' in submitted, submitted
            if during_turn and index == 0:
                await during_turn()
            terminal = await until(lambda r: r.get('params', {}).get('type') in {'message.complete', 'error'})
            assert 'LOCAL_ACK_WS_SHARED' in json.dumps(terminal), terminal


def test_stop_secondary_keeps_host_and_survives_host_restart(mux, peer):
    """A satellite's runtime PID is its serving host, never its process to kill."""
    import threading
    peer.blocked, peer.release = threading.Event(), threading.Event()
    root, boot = mux['root'], mux['boot']
    proc, desc = start_daemon(mux)

    async def stop_during_primary_turn():
        assert await asyncio.to_thread(peer.blocked.wait, 10), 'primary turn never reached model'
        try:
            stopped = await asyncio.to_thread(subprocess.run,
                [sys.executable, '-m', 'hermes_cli.main', '-p', 'boot', 'gateway', 'stop'],
                cwd=ROOT, env=mux['env'], capture_output=True, text=True, timeout=45, check=False)
            assert stopped.returncode == 0, (stopped.stdout, stopped.stderr)
            assert (boot / 'gateway.parked').exists(), (stopped.stdout, stopped.stderr)
            assert proc.poll() is None, tail(mux)
            after = control(root, 'identify')
            assert after['instance_id'] == desc['instance_id']
            assert served_homes(after) == {root.resolve()}
        finally:
            peer.release.set()

    try:
        asyncio.run(chat_for(root, root, desc, during_turn=stop_during_primary_turn))
        # An explicit host restart exercises persisted stop intent, not merely cached status.
        stop(proc)
        proc, desc = start_daemon(mux)
        assert served_homes(desc) == {root.resolve()}
        started = subprocess.run(
            [sys.executable, '-m', 'hermes_cli.main', '-p', 'boot', 'gateway', 'start'],
            cwd=ROOT, env=mux['env'], capture_output=True, text=True, timeout=45, check=False,
        )
        assert started.returncode == 0, (started.stdout, started.stderr)
        assert not (boot / 'gateway.parked').exists()
        after = control(root, 'identify')
        assert after['instance_id'] == desc['instance_id']
        assert served_homes(after) == {root.resolve(), boot.resolve()}
        asyncio.run(chat_for(root, boot, after))
    finally:
        peer.release.set()
        stop(proc, expect=None)


def test_standalone_secondary_stops_without_stopping_host(mux):
    """Positive OS ownership still permits stopping a real independent gateway."""
    root, boot = mux['root'], mux['boot']
    config = {**mux['config'], 'gateway': {'standalone': True, 'multiplex_profiles': False}}
    (boot / 'config.yaml').write_text(json.dumps(config), encoding='utf-8')
    host, desc = start_daemon(mux)
    child = None
    try:
        assert served_homes(desc) == {root.resolve()}
        standalone = {**mux, 'root': boot, 'env': {**mux['env'], 'HERMES_HOME': str(boot)},
                      'log': mux['log'].with_name('standalone.log')}
        child, _ = start_daemon(standalone)
        stopped = subprocess.run(
            [sys.executable, '-m', 'hermes_cli.main', '-p', 'boot', 'gateway', 'stop'],
            cwd=ROOT, env=mux['env'], capture_output=True, text=True, timeout=45, check=False)
        assert stopped.returncode == 0, (stopped.stdout, stopped.stderr)
        child.wait(timeout=30)
        assert host.poll() is None
        assert control(root, 'identify')['instance_id'] == desc['instance_id']
        assert not (boot / 'gateway.parked').exists()
    finally:
        if child is not None:
            stop(child, expect=None)
        stop(host, expect=None)


def test_unsupervised_host_without_profile_flag_can_stop(mux):
    proc, _ = start_daemon(mux, extra_args=('--no-supervise',))
    try:
        result = subprocess.run(
            [sys.executable, '-m', 'hermes_cli.main', 'gateway', 'stop'],
            cwd=ROOT, env=mux['env'], capture_output=True, text=True, timeout=45, check=False)
        assert result.returncode == 0, (result.stdout, result.stderr)
        assert 'Stopped gateway for this profile' in result.stdout, (result.stdout, result.stderr)
        proc.wait(timeout=30)
    finally:
        stop(proc, expect=None)