Sprinter-SDCC/toolchain/sdbg/supervisor.py

"""Владелец автономного launcher для MCP без запущенного VS Code."""
from __future__ import annotations

import json
from pathlib import Path
import subprocess
import sys
import tempfile
import threading
import uuid

from .mcp_adapter import McpSession
from .session import SessionError


class SessionSupervisor:
    def __init__(self, build: str, socket_path: str | None = None,
                 launcher_options: list[str] | None = None):
        self.build = str(Path(build).expanduser().resolve())
        self._temporary = None
        if socket_path:
            self.socket_path = str(Path(socket_path).expanduser())
        else:
            self._temporary = tempfile.TemporaryDirectory(prefix='sprinter-mcp-', dir='/tmp')
            self.socket_path = str(Path(self._temporary.name) / 'session.sock')
        self.launcher_options = list(launcher_options or [])
        self.lock = threading.RLock()
        self.phase = 'idle'
        self.error: str | None = None
        self.entry = None
        self.mame_pid = None
        self.process: subprocess.Popen | None = None
        self.monitor: threading.Thread | None = None
        self.log = None
        self.client: McpSession | None = None
        self.launch_id = None

    def status(self) -> dict:
        with self.lock:
            result = {'phase': self.phase, 'socket': self.socket_path,
                      'launch_id': self.launch_id}
            if self.error:
                result['error'] = self.error
            if self.phase == 'ready' and self.client is not None:
                try:
                    result.update(self.client.status())
                    result['mame_pid'] = self.mame_pid
                    result['entry'] = self.entry
                except (SessionError, OSError) as error:
                    self.phase = 'failed'
                    self.error = str(error)
                    result.update(phase='failed', error=self.error)
            return result

    def require_client(self) -> McpSession:
        with self.lock:
            if self.phase != 'ready' or self.client is None:
                raise SessionError('C-сессия ещё не готова: ' + self.phase)
            return self.client

    def start(self) -> dict:
        with self.lock:
            if self.phase in ('starting', 'ready', 'stopping'):
                raise SessionError('Автономная сессия уже запущена: ' + self.phase)
            if self.process is not None and self.process.poll() is None:
                raise SessionError('Предыдущий launcher ещё работает')
            if self.monitor is not None and self.monitor.is_alive():
                raise SessionError('Предыдущий launcher ещё завершает работу')
            if not Path(self.build).is_dir():
                raise SessionError('Debug-пакет не найден: ' + self.build)
            command = [sys.executable, str(Path(__file__).resolve().parents[1] /
                       'sdbg_launcher.py'), '--build', self.build,
                       '--socket', self.socket_path, *self.launcher_options]
            if self.log is not None:
                self.log.close()
            self.log = tempfile.TemporaryFile(mode='w+t', encoding='utf-8')
            self.phase = 'starting'
            self.error = None
            self.entry = None
            self.mame_pid = None
            self.client = None
            self.launch_id = uuid.uuid4().hex
            try:
                self.process = subprocess.Popen(command, stdout=subprocess.PIPE,
                                                stderr=self.log, text=True)
            except OSError:
                self.phase = 'failed'
                raise
            self.monitor = threading.Thread(target=self._monitor, name='sdbg-supervisor',
                                            daemon=True)
            self.monitor.start()
            return {'accepted': True, 'phase': self.phase,
                    'socket': self.socket_path, 'launch_id': self.launch_id}

    def _diagnostics(self) -> str:
        if self.log is None:
            return ''
        self.log.flush()
        self.log.seek(0)
        return self.log.read()[-4000:].strip()

    def _monitor(self) -> None:
        process = self.process
        assert process is not None and process.stdout is not None
        try:
            line = process.stdout.readline()
            if not line:
                raise SessionError('Launcher завершился до main: ' + self._diagnostics())
            ready = json.loads(line)
            if not ready.get('ready') or ready.get('socket') != self.socket_path:
                raise SessionError('Launcher вернул неверный ready')
            client = McpSession(self.socket_path)
            client.status()
            with self.lock:
                if self.phase == 'starting':
                    self.client = client
                    self.entry = ready['entry']
                    self.mame_pid = ready['pid']
                    self.phase = 'ready'
            process.wait()
            with self.lock:
                if self.phase not in ('stopping', 'failed'):
                    self.phase = 'stopped'
                    self.error = 'Launcher завершился: ' + self._diagnostics()
                elif self.phase == 'stopping':
                    self.phase = 'stopped'
                self.client = None
        except (OSError, ValueError, KeyError, SessionError) as error:
            with self.lock:
                if self.phase == 'stopping':
                    self.phase = 'stopped'
                else:
                    self.phase = 'failed'
                    self.error = str(error)
                self.client = None

    def stop(self) -> dict:
        with self.lock:
            if self.phase not in ('starting', 'ready'):
                return {'accepted': False, 'phase': self.phase}
            self.phase = 'stopping'
            if self.process is not None and self.process.poll() is None:
                self.process.terminate()
            return {'accepted': True, 'phase': 'stopping',
                    'socket': self.socket_path}

    def close(self) -> None:
        self.stop()
        process = self.process
        if process is not None:
            try:
                process.wait(timeout=8)
            except subprocess.TimeoutExpired:
                process.kill()
                process.wait()
        if self.monitor is not None:
            self.monitor.join(timeout=2)
        if self.log is not None:
            self.log.close()
        if self._temporary is not None:
            self._temporary.cleanup()


class ManagedMcpSession:
    """Совместимый с McpSession фасад, пока launcher проходит DSS."""
    def __init__(self, supervisor: SessionSupervisor):
        self.supervisor = supervisor

    def status(self) -> dict:
        return self.supervisor.status()

    def __getattr__(self, name: str):
        return getattr(self.supervisor.require_client(), name)