"""Real compression, real registered worker, real ordinary owner transport."""
import asyncio
from contextlib import closing
from http.server import ThreadingHTTPServer
import json
import os
from pathlib import Path
import sqlite3
import subprocess
import sys
import threading

import pytest

from tests.gateway.fixtures.local_recovery_probe import daemon, rpc, websocket
from tests.gateway.test_worker_agent_execution import Model


@pytest.mark.platforms("linux")
def test_ordinary_daemon_worker_compresses_without_canonical_sqlite_opens(tmp_path):
    root = Path(__file__).resolve().parents[2]
    home, user = tmp_path / 'state', tmp_path / 'user'
    home.mkdir(mode=0o700)
    user.mkdir()
    peer = ThreadingHTTPServer(('127.0.0.1', 0), Model)
    peer.requests = []
    thread = threading.Thread(target=peer.serve_forever, daemon=True)
    thread.start()
    url = f'http://127.0.0.1:{peer.server_port}/v1'
    (home / 'config.yaml').write_text(json.dumps({'gateway': {'multiplex_profiles': False},
        'model': {'provider': 'custom', 'default': 'worker-model', 'base_url': url},
        'auxiliary': {'title_generation': {'enabled': False},
                      'compression': {'provider': 'custom', 'model': 'worker-model', 'base_url': url}},
        'platform_toolsets': {'cli': []}}))
    env = {k: os.environ[k] for k in ('PATH', 'LANG', 'TZ', 'TIRITH_ENABLED') if k in os.environ}
    env.update(HOME=str(user), USERPROFILE=str(user), HERMES_HOME=str(home), PYTHONPATH=str(root),
               OPENAI_API_KEY='loopback-only', OPENAI_BASE_URL=url)
    async def create(desc, key):
        async with websocket(home, desc) as ws:
            response = await rpc(ws, 'session.create', request_id=key, source='cli', model='worker-model', toolsets=[])
            assert 'result' in response, response
            return response['result']['session_id']
    try:
        with daemon(root, home, env, barrier=False) as (_, desc):
            sid = asyncio.run(create(desc, 'compressed'))
            foreign = asyncio.run(create(desc, 'foreign'))
            receipt = tmp_path / 'receipt.json'
            result = subprocess.run([sys.executable, str(root / 'tests/gateway/fixtures/compression_persistence_worker.py')],
                cwd=root, env=env, input=json.dumps(dict(home=str(home), url=url, session_id=sid,
                    foreign=foreign, receipt=str(receipt))) + '\n', text=True, capture_output=True, timeout=120, check=False)
            assert result.returncode == 0, result.stdout + result.stderr
            proof = json.loads(receipt.read_text())
            assert proof['opens'] == proof['fds'] == []
            assert proof['compressed_count'] < proof['input_count']
            assert peer.requests, 'no actual summary inference'
            with closing(sqlite3.connect(f'file:{home / "state.db"}?mode=ro', uri=True)) as db:
                assignment = db.execute('SELECT session_id,status FROM worker_executions').fetchone()
                assert assignment == (proof['after'], 'terminal')
                assert db.execute('SELECT COUNT(*) FROM messages WHERE session_id=? AND active=1', (proof['after'],)).fetchone()[0] == len(proof['history'])
            print(json.dumps(dict(before=proof['before'], after=proof['after'], opens=proof['opens'],
                fds=proof['fds'], negatives=proof['negatives'], requests=len(peer.requests),
                compressed_count=proof['compressed_count'], input_count=proof['input_count'],
                persisted_rows=len(proof['history']))))
    finally:
        peer.shutdown()
        peer.server_close()
        thread.join(timeout=5)