from __future__ import annotations import asyncio from app.core.errors import EngineError from app.core.interface import EngineBackend from app.core.registry import BackendRegistry from app.models import ( CapabilitySet, EngineKind, EngineState, RuntimeRequest, RuntimeSnapshot, RuntimeStats, ) class EngineManager: """Serializes runtime operations and prevents FreeTurn/VPN conflicts.""" def __init__(self, registry: BackendRegistry) -> None: self.registry = registry self._active_kind: EngineKind | None = None self._backend: EngineBackend | None = None self._lock = asyncio.Lock() @property def active_kind(self) -> EngineKind | None: return self._active_kind def capabilities(self, kind: EngineKind) -> CapabilitySet: if self._backend is not None and self._active_kind is kind: return self._backend.capabilities return self.registry.create(kind).capabilities async def start(self, request: RuntimeRequest) -> RuntimeSnapshot: async with self._lock: if self._active_kind is not None and self._active_kind is not request.engine: raise EngineError("another engine is active; stop it before switching") backend = ( self._backend if self._backend is not None and self._active_kind is request.engine else self.registry.create(request.engine) ) self._backend = backend if request.mode.value in {"vpn", "both"} and not backend.capabilities.supports_vpn: raise EngineError(f"{request.engine} does not support VPN mode") if request.mode.value in {"proxy", "both"} and not backend.capabilities.supports_proxy: raise EngineError(f"{request.engine} does not support proxy mode") if request.mode.value == "both" and not backend.capabilities.supports_both: raise EngineError(f"{request.engine} does not support Both mode") snapshot = await backend.start(request) if snapshot.state in {EngineState.RUNNING, EngineState.STARTING}: self._active_kind = request.engine return snapshot async def stop(self) -> RuntimeSnapshot: async with self._lock: if self._backend is None: return RuntimeSnapshot() snapshot = await self._backend.stop() if snapshot.state in {EngineState.STOPPED, EngineState.ERROR, EngineState.UNSUPPORTED}: self._active_kind = None self._backend = None return snapshot async def restart(self, request: RuntimeRequest) -> RuntimeSnapshot: async with self._lock: if self._active_kind is not None and self._active_kind is not request.engine: raise EngineError("another engine is active; stop it before switching") backend = ( self._backend if self._backend is not None and self._active_kind is request.engine else self.registry.create(request.engine) ) self._backend = backend snapshot = await backend.restart(request) if snapshot.state in {EngineState.RUNNING, EngineState.STARTING}: self._active_kind = request.engine return snapshot async def get_state(self) -> RuntimeSnapshot: async with self._lock: if self._backend is None: # Native services can outlive the Flet process. Probe registered # adapters so a newly opened UI can discover the authoritative # runtime instead of interpreting missing Python state as STOPPED. for kind in self.registry.kinds(): candidate = self.registry.create(kind) snapshot = await candidate.get_state() if snapshot.engine is kind and snapshot.state in { EngineState.STARTING, EngineState.RUNNING, EngineState.STOPPING, }: self._backend = candidate self._active_kind = kind return snapshot return RuntimeSnapshot() return await self._backend.get_state() async def get_stats(self) -> RuntimeStats: async with self._lock: if self._backend is None: return RuntimeStats() return await self._backend.get_stats()