zapret-kvn/xray_fluent/engines/hysteria/manager.py

from __future__ import annotations

from collections import deque
from concurrent.futures import Future, ThreadPoolExecutor
from copy import deepcopy
import json
import os
from pathlib import Path
import time
from typing import Any, Callable

from PyQt6.QtCore import QObject, QProcess, QTimer, pyqtSignal

from ...application.async_steps import (
    TransitionRunner,
    TransitionSteps,
    run_in_worker,
    run_steps_blocking,
    sleep_ms,
    wait_process_finished,
    wait_process_started,
)

from ...constants import HYSTERIA_CONFIG_FILE, HYSTERIA_PATH_DEFAULT, PROXY_HOST, RUNTIME_DIR
from ...diagnostics.export import capture_runtime_config
from ..socks_probe import HTTPS_ENDPOINTS, probe_https
from ..health_check import BackgroundHealthCheck
from ..sidecar import wait_for_loopback_relay_steps
from .runtime_contract import (
    SECURITY_FAILURES,
    HysteriaFailureCode,
    classify_hysteria_failure,
)
from ...diagnostics.runtime_logging import RuntimeNodeIdentity, redact_runtime_log, strip_terminal_controls
from ...platform.windows.subprocess_utils import decode_output, kill_processes_by_path


_FUNCTIONAL_HTTPS_ENDPOINTS = HTTPS_ENDPOINTS

# Проверка живого туннеля после ошибки в логе: одна волна HTTPS-проб через
# relay и число волн. Клиент Hysteria сам переподключается при следующем
# запросе, так что проба одновременно и вызывает переподключение.
_TUNNEL_CHECK_TIMEOUT = 6.0
_TUNNEL_CHECK_WAVES = 2

# start_steps(): the in-flight attempt saw a Chrome-parrot TLS rejection and
# must be repeated once with disableChromeParrot (no nested start()).
_RETRY_WITHOUT_CHROME_PARROT = object()


def _still_wanted(should_continue: Callable[[], bool] | None) -> bool:
    return should_continue is None or bool(should_continue())


def _kill_orphaned_blocking(exe: Path) -> bool:
    """Worker-only: kill an orphaned hysteria.exe of this installation."""
    try:
        return bool(kill_processes_by_path(exe.name, exe, timeout=5, pump=False))
    except Exception:
        return False


class HysteriaManager(QObject):
    """Run the unmodified official Hysteria client as a local SOCKS sidecar."""

    started = pyqtSignal()
    stopped = pyqtSignal(int)
    log_received = pyqtSignal(str)
    error = pyqtSignal(str)
    warning = pyqtSignal(str)
    failure = pyqtSignal(str, str, int)
    state_changed = pyqtSignal(bool)

    def __init__(self, parent: QObject | None = None):
        super().__init__(parent)
        self._process = QProcess(self)
        self._process.setProcessChannelMode(QProcess.ProcessChannelMode.MergedChannels)
        self._process.readyReadStandardOutput.connect(self._on_ready_read)
        self._process.started.connect(self._on_started)
        self._process.errorOccurred.connect(self._on_error)
        self._process.finished.connect(self._on_finished)
        self._running = False
        self._starting = False
        self._stop_requested = False
        self._last_output_lines: deque[str] = deque(maxlen=100)
        self._context: RuntimeNodeIdentity | None = None
        self._attempt = 0
        self._failure_reported = False
        self._stdout_buffer = ""
        self._secret_values: tuple[str, ...] = ()
        self._compatibility_generation = 0
        self._compatibility_config: dict[str, Any] | None = None
        self._compatibility_relay_port = 0
        self._compatibility_context: RuntimeNodeIdentity | None = None
        self._chrome_fallback_pending = False
        self._chrome_fallback_used = False
        self._chrome_fallback_in_progress = False
        self._suppress_state_change = False
        self._process_generation = 0
        self._config_path = HYSTERIA_CONFIG_FILE
        self._last_failure_code: HysteriaFailureCode | None = None
        self.diagnostic_config: dict[str, Any] | None = None
        self._compatibility_allow_parallel = False
        self._compatibility_verify_remote = True
        self._attempt_started_at = 0.0
        self._remote_authenticated = False
        self.stats: dict[str, Any] = {}
        self._health = BackgroundHealthCheck(self)
        # Ошибка транспорта в логе работающего туннеля — повод для проверки,
        # а не приговор серверу: failure уходит, только если проверка провалена.
        self._tunnel_check = BackgroundHealthCheck(self)
        self._tunnel_check_wave = 0
        self._tunnel_check_evidence: tuple[str, str, HysteriaFailureCode] | None = None
        self._relay_probe: tuple[int, str, str] | None = None
        self._start_depth = 0
        self._fallback_runner: TransitionRunner | None = None
        # A crash can leave a short-lived config behind. It is never reusable:
        # every start writes a fresh one, so remove stale secrets immediately.
        self._cleanup_config()
        self._cleanup_stale_generation_configs()

    @property
    def is_running(self) -> bool:
        return self._running

    @property
    def process_generation(self) -> int:
        return self._process_generation

    @property
    def last_failure_code(self) -> HysteriaFailureCode | None:
        return self._last_failure_code

    @property
    def process_alive(self) -> bool:
        """A spawned process exists, even if readiness was never confirmed."""
        return self._process.state() != QProcess.ProcessState.NotRunning

    def start(
        self,
        config: dict[str, Any],
        relay_port: int,
        *,
        context: RuntimeNodeIdentity | None = None,
        process_generation: int = 0,
        allow_parallel: bool = False,
        verify_remote: bool = True,
        _compatibility_retry: bool = False,
    ) -> bool:
        # Синхронный драйвер тех же шагов — только для shutdown/тестов.
        return bool(
            run_steps_blocking(
                self.start_steps(
                    config,
                    relay_port,
                    context=context,
                    process_generation=process_generation,
                    allow_parallel=allow_parallel,
                    verify_remote=verify_remote,
                    _compatibility_retry=_compatibility_retry,
                )
            )
        )

    def start_steps(
        self,
        config: dict[str, Any],
        relay_port: int,
        *,
        context: RuntimeNodeIdentity | None = None,
        process_generation: int = 0,
        allow_parallel: bool = False,
        verify_remote: bool = True,
        should_continue: Callable[[], bool] | None = None,
        _compatibility_retry: bool = False,
    ) -> TransitionSteps:
        """Start the sidecar; all waits are steps, probes run in workers.

        A Chrome-parrot compatibility retry requested while this start is in
        flight is performed here, inline, instead of re-entering ``start``.
        """
        self._start_depth += 1
        try:
            started = yield from self._start_attempt_steps(
                config,
                relay_port,
                context=context,
                process_generation=process_generation,
                allow_parallel=allow_parallel,
                verify_remote=verify_remote,
                should_continue=should_continue,
                compatibility_retry=_compatibility_retry,
            )
            while started is _RETRY_WITHOUT_CHROME_PARROT:
                retry_config = self._chrome_fallback_config()
                if retry_config is None:
                    return False
                self._chrome_fallback_pending = False
                self._chrome_fallback_in_progress = True
                try:
                    started = yield from self._start_attempt_steps(
                        retry_config,
                        relay_port,
                        context=context,
                        process_generation=self._process_generation,
                        allow_parallel=self._compatibility_allow_parallel,
                        verify_remote=self._compatibility_verify_remote,
                        should_continue=should_continue,
                        compatibility_retry=True,
                    )
                finally:
                    self._chrome_fallback_in_progress = False
                if started is False:
                    self._emit_error(
                        "Не удалось повторно запустить Hysteria2 в режиме совместимости сертификата.",
                        stage="compatibility_retry",
                    )
                    self._clear_compatibility_state()
            return bool(started)
        finally:
            self._start_depth -= 1

    def _start_attempt_steps(
        self,
        config: dict[str, Any],
        relay_port: int,
        *,
        context: RuntimeNodeIdentity | None,
        process_generation: int,
        allow_parallel: bool,
        verify_remote: bool,
        should_continue: Callable[[], bool] | None,
        compatibility_retry: bool,
    ) -> TransitionSteps:
        if not compatibility_retry:
            self._compatibility_generation += 1
            self._chrome_fallback_pending = False
            self._chrome_fallback_used = False
            self._chrome_fallback_in_progress = False
            self._process_generation = max(0, int(process_generation))
            self._config_path = (
                HYSTERIA_CONFIG_FILE.with_name(
                    f"{HYSTERIA_CONFIG_FILE.stem}-{self._process_generation}.json"
                )
                if allow_parallel and self._process_generation > 0
                else HYSTERIA_CONFIG_FILE
            )
            self._compatibility_allow_parallel = bool(allow_parallel)
            self._compatibility_verify_remote = bool(verify_remote)
            self._last_failure_code = None
        self._compatibility_config = deepcopy(config)
        self._compatibility_relay_port = relay_port
        socks = config.get("socks5")
        socks = socks if isinstance(socks, dict) else {}
        self._relay_probe = (
            relay_port,
            str(socks.get("username") or ""),
            str(socks.get("password") or ""),
        )
        self._compatibility_context = context
        exe = HYSTERIA_PATH_DEFAULT.resolve()
        if not exe.is_file():
            self._begin_attempt(context, config)
            self._emit_error(
                f"hysteria.exe не найден: {exe}. Переустановите или обновите Zapret KVN.",
                stage="validate",
            )
            self._clear_compatibility_state()
            return False
        if relay_port <= 0:
            self._begin_attempt(context, config)
            self._emit_error("Некорректный локальный порт Hysteria sidecar", stage="validate")
            self._clear_compatibility_state()
            return False

        # Keep the old identity active until its process is fully stopped. This
        # prevents a late finished/error signal from being attributed to the
        # connection that is only about to start.
        if self._process.state() != QProcess.ProcessState.NotRunning:
            self._failure_reported = False
            if not (yield from self.stop_steps(expected=True, _preserve_compatibility=True)):
                self._emit_error(
                    "Не удалось остановить предыдущий процесс Hysteria",
                    stage="stop_previous",
                )
                self._clear_compatibility_state()
                return False
        elif self._running:
            self._running = False
            self.state_changed.emit(False)

        self._begin_attempt(context, config)
        attempt_generation = self._compatibility_generation
        if not allow_parallel:
            killed = yield run_in_worker(lambda: _kill_orphaned_blocking(exe))
            if killed:
                yield sleep_ms(500)
            if attempt_generation != self._compatibility_generation or not _still_wanted(should_continue):
                return False
        self._cleanup_config()
        temporary = self._config_path.with_suffix(".json.tmp")
        try:
            RUNTIME_DIR.mkdir(parents=True, exist_ok=True)
            temporary.write_text(json.dumps(config, ensure_ascii=True, indent=2), encoding="utf-8")
            temporary.chmod(0o600)
            temporary.replace(self._config_path)
            self.diagnostic_config = capture_runtime_config(exe, config)
        except OSError as exc:
            self._cleanup_config()
            self._emit_error(f"Не удалось записать временный конфиг: {exc}", stage="write_config")
            self._clear_compatibility_state()
            return False

        self._starting = True
        self._stop_requested = False
        self._last_output_lines.clear()
        self._emit_log(
            f"launch relay={PROXY_HOST}:{relay_port} remote_handshake=deferred(lazy)",
            stage="spawn",
        )
        self._process.setWorkingDirectory(str(exe.parent))
        self._process.setProgram(str(exe))
        self._process.setArguments(
            [
                "--config",
                str(self._config_path),
                "--disable-update-check",
                "--log-level",
                "info",
                "client",
            ]
        )
        self._process.start()
        if not (yield wait_process_started(self._process, 4000)):
            self._starting = False
            self._cleanup_config()
            self._emit_error(
                f"Не удалось запустить Hysteria: {self._process.errorString()}",
                stage="spawn",
            )
            self._clear_compatibility_state()
            return False

        relay_ready = yield from self._wait_until_relay_ready_steps(
            relay_port, should_continue=should_continue
        )
        if self._chrome_fallback_pending and attempt_generation == self._compatibility_generation:
            return _RETRY_WITHOUT_CHROME_PARROT
        if not relay_ready:
            if attempt_generation != self._compatibility_generation or not _still_wanted(should_continue):
                self._starting = False
                if self._process.state() != QProcess.ProcessState.NotRunning:
                    yield from self.stop_steps(expected=True, _preserve_compatibility=True)
                return False
            details = self._last_output_lines[-1] if self._last_output_lines else "локальный SOCKS не открылся"
            yield from self.stop_steps(expected=True)
            self._starting = False
            self._emit_error(f"Hysteria sidecar не запустился: {details}", stage="wait_ready")
            return False

        self._emit_log(
            f"local relay ready in {int((time.monotonic() - self._attempt_started_at) * 1000)} ms",
            stage="relay_ready",
        )

        socks = config.get("socks5")
        socks = socks if isinstance(socks, dict) else {}
        readiness_generation = self._compatibility_generation
        if verify_remote:
            remote_ready = yield from self._wait_until_remote_ready_steps(
                relay_port,
                username=str(socks.get("username") or ""),
                password=str(socks.get("password") or ""),
                should_continue=should_continue,
            )
            if self._chrome_fallback_pending and readiness_generation == self._compatibility_generation:
                return _RETRY_WITHOUT_CHROME_PARROT
            if not remote_ready:
                if readiness_generation != self._compatibility_generation:
                    return False
                if not _still_wanted(should_continue):
                    self._starting = False
                    yield from self.stop_steps(expected=True, _preserve_compatibility=True)
                    return False
                details = self._last_output_lines[-1] if self._last_output_lines else "HTTPS probe через relay не завершился"
                remote_failure = self._last_failure_code is not None
                yield from self.stop_steps(expected=True)
                self._starting = False
                self._emit_error(
                    ("Соединение Hysteria с сервером не установлено: " if remote_failure
                     else "Проверка HTTPS через Hysteria не завершилась: ") + details,
                    stage="remote_handshake" if remote_failure else "functional_ready",
                    code=self._last_failure_code
                    or classify_hysteria_failure(details)
                    or HysteriaFailureCode.TARGET_NETWORK_TIMEOUT,
                )
                return False
            self._emit_log(
                "functional readiness completed in "
                f"{int((time.monotonic() - self._attempt_started_at) * 1000)} ms",
                stage="functional_ready",
            )

        # Hysteria has parsed the config by the time its SOCKS listener is
        # ready. Do not leave the URI/passwords on disk for the whole session.
        self._cleanup_config()
        self._starting = False
        self._mark_running()
        return True

    def _begin_attempt(
        self,
        context: RuntimeNodeIdentity | None,
        config: dict[str, Any],
    ) -> None:
        self._cancel_health()
        self._remote_authenticated = False
        self.stats = {"remote_authenticated": False, "https_check": "pending"}
        self._attempt += 1
        self._attempt_started_at = time.monotonic()
        self._context = context
        self._failure_reported = False
        self._stdout_buffer = ""
        self._secret_values = self._collect_secret_values(config)

    def stop(self, expected: bool = True, *, _preserve_compatibility: bool = False) -> bool:
        # Синхронный драйвер — shutdown и тесты; переходы используют stop_steps().
        return bool(
            run_steps_blocking(
                self.stop_steps(expected=expected, _preserve_compatibility=_preserve_compatibility)
            )
        )

    def request_stop(self, expected: bool = True) -> None:
        """Close admission now without waiting (signal handlers)."""
        self._cancel_health()
        self._clear_compatibility_state()
        self._cleanup_config()
        if self._process.state() == QProcess.ProcessState.NotRunning:
            return
        self._stop_requested = expected
        self._process.kill()

    def stop_steps(self, expected: bool = True, *, _preserve_compatibility: bool = False) -> TransitionSteps:
        self._cancel_health()
        if not _preserve_compatibility:
            self._clear_compatibility_state()
        self._cleanup_config()
        if self._process.state() == QProcess.ProcessState.NotRunning:
            self._stop_requested = False
            self._starting = False
            if self._running:
                self._running = False
                self.state_changed.emit(False)
            return True

        self._stop_requested = expected
        self._process.kill()
        if not (yield wait_process_finished(self._process, 2000)):
            self._stop_requested = False
            self._emit_error("Не удалось вовремя остановить процесс Hysteria", stage="stop")
            return False
        self._starting = False
        return True

    def _wait_until_relay_ready(self, relay_port: int, timeout: float = 10.0) -> bool:
        return bool(run_steps_blocking(self._wait_until_relay_ready_steps(relay_port, timeout=timeout)))

    def _wait_until_relay_ready_steps(
        self,
        relay_port: int,
        timeout: float = 10.0,
        *,
        should_continue: Callable[[], bool] | None = None,
    ) -> TransitionSteps:
        # Shared sidecar seam: the loopback SOCKS listener must accept a TCP
        # connection before the sing-box front can dial it.
        generation = self._compatibility_generation
        return (yield from wait_for_loopback_relay_steps(
            relay_port,
            timeout=timeout,
            should_continue=lambda: (
                self._process.state() != QProcess.ProcessState.NotRunning
                and generation == self._compatibility_generation
                and not self._chrome_fallback_pending
                and _still_wanted(should_continue)
            ),
        ))

    def _wait_until_remote_ready(
        self,
        relay_port: int,
        *,
        username: str,
        password: str,
        timeout: float = 15.0,
    ) -> bool:
        return bool(
            run_steps_blocking(
                self._wait_until_remote_ready_steps(
                    relay_port, username=username, password=password, timeout=timeout
                )
            )
        )

    def _wait_until_remote_ready_steps(
        self,
        relay_port: int,
        *,
        username: str,
        password: str,
        timeout: float = 15.0,
        should_continue: Callable[[], bool] | None = None,
    ) -> TransitionSteps:
        """Prove HTTPS egress without making one external provider authoritative."""

        deadline = time.monotonic() + timeout
        generation = self._compatibility_generation
        failures: dict[str, str] = {}
        attempts = 0

        def abandoned() -> bool:
            return (
                self._stop_requested
                or generation != self._compatibility_generation
                or self._process.state() == QProcess.ProcessState.NotRunning
                or self._last_failure_code in SECURITY_FAILURES
                or self._chrome_fallback_pending
                or not _still_wanted(should_continue)
            )

        while time.monotonic() < deadline and attempts < 3:
            attempts += 1
            if abandoned():
                return False
            probe_timeout = min(4.0, max(0.2, deadline - time.monotonic()))
            executor = ThreadPoolExecutor(
                max_workers=len(_FUNCTIONAL_HTTPS_ENDPOINTS),
                thread_name_prefix="hysteria-ready",
            )
            futures: dict[Future[None], tuple[str, str, str]] = {
                executor.submit(
                    self._probe_remote_endpoint,
                    relay_port,
                    username=username,
                    password=password,
                    endpoint=endpoint,
                    timeout=probe_timeout,
                ): endpoint
                for endpoint in _FUNCTIONAL_HTTPS_ENDPOINTS
            }
            succeeded: tuple[str, str, str] | None = None
            while futures and time.monotonic() < deadline:
                if abandoned():
                    executor.shutdown(wait=False, cancel_futures=True)
                    return False
                if self._remote_authenticated:
                    self._monitor_health(executor, futures, failures, generation)
                    return True
                completed = [future for future in futures if future.done()]
                for future in completed:
                    endpoint = futures.pop(future)
                    error = future.exception()
                    if error is None:
                        succeeded = endpoint
                        break
                    failures[endpoint[1]] = f"{type(error).__name__}: {error}"
                if succeeded is not None:
                    break
                yield sleep_ms(50)
            if (succeeded is None and self._remote_authenticated and generation == self._compatibility_generation
                    and self._last_failure_code not in SECURITY_FAILURES
                    and self._process.state() != QProcess.ProcessState.NotRunning
                    and not self._stop_requested):
                self._monitor_health(executor, futures, failures, generation)
                return True
            executor.shutdown(wait=False, cancel_futures=True)
            if self._last_failure_code in SECURITY_FAILURES:
                return False
            if succeeded is not None:
                self.stats["https_check"] = "passed"
                self._emit_log(
                    f"functional HTTPS probe succeeded via {succeeded[1]}",
                    stage="functional_ready",
                )
                return True
            yield sleep_ms(100)
        if failures:
            summary = "; ".join(
                f"{host}={detail}" for host, detail in sorted(failures.items())
            )
            self._emit_log(f"functional HTTPS probes failed: {summary}", stage="functional_ready")
        return False

    def _monitor_health(self, executor, futures, failures, generation):
        """Finish the existing probe wave off the startup path, without new traffic."""
        self._cancel_health()
        self.stats["https_check"] = "pending"
        self._health.adopt(executor, {future: endpoint[1] for future, endpoint in futures.items()}, failures,
            current=lambda: (generation == self._compatibility_generation
                and self._running and not self._stop_requested
                and self._process.state() != QProcess.ProcessState.NotRunning
                and self._last_failure_code not in SECURITY_FAILURES),
            complete=self._health_complete)

    def _cancel_health(self):
        self._health.cancel()
        self._tunnel_check.cancel()
        self._tunnel_check_evidence = None
        if self.stats.get("https_check") == "pending":
            self.stats["https_check"] = "cancelled"

    def _poll_health(self):
        self._health.poll()
        if not self._health.active and self.stats.get("https_check") == "pending":
            self.stats["https_check"] = "cancelled"

    def _health_complete(self, winner, failures):
        if winner is not None:
            self.stats["https_check"] = "passed"
            self._emit_log(f"HTTPS check succeeded via {winner}", stage="health_check")
            return
        self.stats["https_check"] = "warning"
        summary = "; ".join(f"{host}={error}" for host, error in sorted(failures.items()))
        self._emit_log("authenticated server retained; HTTPS check endpoints did not respond: " + summary,
                       stage="health_check")
        # The server handshake is already authenticated (the official client
        # reported "connected to server"), so the connection is proven. Probe
        # destinations can still be filtered by an exit's policy, so their
        # failure is not a user-facing problem and never a teardown reason.
        # Only warn if the handshake itself was never confirmed.
        if not self._remote_authenticated:
            self.warning.emit("Hysteria подключена к серверу, но проверочные HTTPS-адреса не ответили. "
                              "Соединение сохранено; доступность сайтов пока не подтверждена.")

    def _begin_tunnel_check(self, message: str, stage: str, code: HysteriaFailureCode) -> None:
        """Проверить работающий туннель вместо немедленного отказа сервера.

        Таймаут одного запроса, обрыв простаивавшей QUIC-сессии или мёртвый
        сайт выглядят в логе одинаково с гибелью сервера. Различает их только
        проверка: если сервер снова авторизовал клиента или HTTPS-проба через
        relay прошла, сервер остаётся; иначе публикуется исходная ошибка.
        """

        if self._tunnel_check.active:
            return
        if self._relay_probe is None or self._relay_probe[0] <= 0:
            self._emit_error(message, stage=stage, code=code)
            return
        self._tunnel_check_evidence = (message, stage, code)
        self._tunnel_check_wave = 0
        self.stats["tunnel_checks"] = int(self.stats.get("tunnel_checks", 0)) + 1
        self._emit_log(
            f"transport error observed ({code.value}); verifying the tunnel before any recovery",
            stage="tunnel_check",
        )
        self._start_tunnel_check_wave()

    def _start_tunnel_check_wave(self) -> None:
        relay_port, username, password = self._relay_probe or (0, "", "")
        generation = self._compatibility_generation
        self._tunnel_check_wave += 1
        executor = ThreadPoolExecutor(
            max_workers=len(_FUNCTIONAL_HTTPS_ENDPOINTS),
            thread_name_prefix="hysteria-tunnel-check",
        )
        futures = {
            executor.submit(
                self._probe_remote_endpoint,
                relay_port,
                username=username,
                password=password,
                endpoint=endpoint,
                timeout=_TUNNEL_CHECK_TIMEOUT,
            ): endpoint[1]
            for endpoint in _FUNCTIONAL_HTTPS_ENDPOINTS
        }
        self._tunnel_check.adopt(
            executor,
            futures,
            {},
            current=lambda: (
                generation == self._compatibility_generation
                and self._running
                and not self._stop_requested
                and self._process.state() != QProcess.ProcessState.NotRunning
            ),
            complete=self._tunnel_check_complete,
        )

    def _tunnel_check_complete(self, winner, failures) -> None:
        evidence = self._tunnel_check_evidence
        if evidence is None:
            return
        if winner is not None:
            self._tunnel_check_passed(f"HTTPS check succeeded via {winner}")
            return
        if self._tunnel_check_wave < _TUNNEL_CHECK_WAVES:
            self._start_tunnel_check_wave()
            return
        self._tunnel_check_evidence = None
        summary = "; ".join(f"{host}={error}" for host, error in sorted(failures.items()))
        self._emit_log(f"tunnel did not recover: {summary}", stage="tunnel_check")
        message, stage, code = evidence
        self._emit_error(message, stage=stage, code=code)

    def _tunnel_check_passed(self, reason: str) -> None:
        self._tunnel_check.cancel()
        self._tunnel_check_evidence = None
        self._last_failure_code = None
        self.stats["tunnel_recoveries"] = int(self.stats.get("tunnel_recoveries", 0)) + 1
        self._emit_log(f"tunnel is alive, server retained: {reason}", stage="tunnel_check")

    def _is_health_probe_error(self, line):
        if not self._remote_authenticated or not (
            self._starting or self._health.active or self._tunnel_check.active
        ):
            return False
        parts = strip_terminal_controls(line).split("\t")
        if len(parts) < 4 or parts[-2] != "SOCKS5 TCP error":
            return False
        try:
            fields = json.loads(parts[-1])
        except (ValueError, TypeError):
            return False
        targets = {f"{address}:443" for address, _, _ in _FUNCTIONAL_HTTPS_ENDPOINTS}
        return isinstance(fields, dict) and fields.get("reqAddr") in targets

    def _observe_authenticated_server(self, line):
        # Parse the official CLI event before log redaction replaces JSON fields.
        # A listener message or a quoted error cannot establish remote readiness.
        parts = strip_terminal_controls(line).split("\t")
        checking = self._tunnel_check.active
        if (len(parts) < 4 or parts[-3:-1] != ["INFO", "connected to server"]
                or not (self._starting or checking) or self._stop_requested):
            return
        try:
            fields = json.loads(parts[-1])
        except (ValueError, TypeError):
            return
        if (not isinstance(fields, dict) or not isinstance(fields.get("addr"), str)
                or not fields["addr"] or type(fields.get("count")) is not int or fields["count"] < 1):
            return
        self._remote_authenticated = True
        self.stats["remote_authenticated"] = True
        if checking:
            # Клиент сам переподключился и сервер его авторизовал — это
            # сильнее любой пробы: проверочные адреса выход может фильтровать.
            self._tunnel_check_passed("server re-authenticated the client")

    def _probe_remote_endpoint(
        self,
        relay_port: int,
        *,
        username: str,
        password: str,
        endpoint: tuple[str, str, str],
        timeout: float,
    ) -> None:
        probe_https(relay_port, username=username, password=password,
                    endpoint=endpoint, timeout=timeout)

    def _cleanup_config(self) -> None:
        try:
            self._config_path.unlink(missing_ok=True)
            self._config_path.with_suffix(".json.tmp").unlink(missing_ok=True)
        except OSError:
            pass

    @staticmethod
    def _cleanup_stale_generation_configs() -> None:
        try:
            for path in HYSTERIA_CONFIG_FILE.parent.glob(
                f"{HYSTERIA_CONFIG_FILE.stem}-*.json*"
            ):
                if path.is_file():
                    path.unlink(missing_ok=True)
        except OSError:
            pass

    @staticmethod
    def redact_log_line(line: str) -> str:
        return redact_runtime_log(line)

    @staticmethod
    def _collect_secret_values(config: dict[str, Any]) -> tuple[str, ...]:
        secret_keys = {
            "auth",
            "password",
            "username",
            "obfs-password",
            "obfs_password",
            "pinsha256",
            "pin_sha256",
            "ech",
            "clientkey",
            "client_key",
        }
        values: list[str] = []

        def visit(value: Any, key: str = "") -> None:
            if isinstance(value, dict):
                for item_key, item in value.items():
                    visit(item, str(item_key).lower())
            elif isinstance(value, (list, tuple)):
                for item in value:
                    visit(item, key)
            elif key in secret_keys:
                text = str(value or "")
                if len(text) >= 4:
                    values.append(text)

        visit(config)
        return tuple(dict.fromkeys(values))

    def _format_message(self, message: str, *, stage: str) -> str:
        clean = redact_runtime_log(message, secrets=self._secret_values)
        fields = f"attempt={self._attempt} stage={stage}"
        if self._context is not None:
            fields += f" {self._context.fields()}"
        return f"[hysteria][{fields}] {clean}".strip()

    def _emit_log(self, message: str, *, stage: str) -> None:
        formatted = self._format_message(message, stage=stage)
        if not formatted:
            return
        self._last_output_lines.append(formatted)
        self.log_received.emit(formatted)

    def _emit_error(
        self,
        message: str,
        *,
        stage: str,
        code: HysteriaFailureCode | None = None,
    ) -> None:
        resolved = code or classify_hysteria_failure(message)
        if resolved is None:
            resolved = {
                "validate": HysteriaFailureCode.LOCAL_CONFIG_INVALID,
                "write_config": HysteriaFailureCode.LOCAL_CONFIG_INVALID,
                "spawn": HysteriaFailureCode.LOCAL_PROCESS_START_FAILED,
                "startup_exit": HysteriaFailureCode.LOCAL_PROCESS_EXITED,
                "unexpected_exit": HysteriaFailureCode.LOCAL_PROCESS_EXITED,
                "wait_ready": HysteriaFailureCode.LOCAL_RELAY_NOT_READY,
                "stop": HysteriaFailureCode.LOCAL_PROCESS_EXITED,
            }.get(stage, HysteriaFailureCode.CORE_UNCLASSIFIED)
        if self._failure_reported and not (
            resolved in SECURITY_FAILURES and self._last_failure_code not in SECURITY_FAILURES
        ):
            return
        # A transient timeout can be logged by one parallel probe before
        # another reports a definitive TLS/auth rejection. Preserve both raw
        # log entries, but security must take precedence for recovery policy.
        self._failure_reported = True
        self._last_failure_code = resolved
        formatted = self._format_message(message, stage=stage)
        # The typed cause is published before generic process/state callbacks,
        # so exit code 62097 cannot replace the original failure episode.
        self.failure.emit(resolved.value, formatted, self._process_generation)
        self.error.emit(formatted)

    def _on_ready_read(self) -> None:
        chunk = self._process.readAllStandardOutput()
        raw = getattr(chunk, "data")()
        text = decode_output(bytes(raw)) if isinstance(raw, (bytes, bytearray)) else str(raw)
        self._stdout_buffer += text
        lines = self._stdout_buffer.splitlines(keepends=True)
        self._stdout_buffer = ""
        for item in lines:
            if not item.endswith(("\n", "\r")):
                self._stdout_buffer = item
                continue
            self._emit_process_line(item.rstrip("\r\n"))

    def _emit_process_line(self, line: str) -> None:
        self._observe_authenticated_server(line)
        from ...diagnostics.runtime_errors import DESTINATION_UNREACHABLE_CODE, classify_core_error
        clean = redact_runtime_log(line, secrets=self._secret_values)
        code, action = classify_core_error(clean)
        if self._is_health_probe_error(line) and classify_hysteria_failure(clean) not in SECURITY_FAILURES:
            self._emit_log(clean, stage="health_check")
            return
        if action == "record_only":
            # Evidence only, never a transport failure or recovery: either a
            # loopback SOCKS client closed its socket, or the authenticated
            # server reported that it could not reach one destination (node
            # policy, dead site) while the tunnel itself stays intact.
            stage = "destination" if code == DESTINATION_UNREACHABLE_CODE else "local_client"
            self._emit_log(clean, stage=stage)
            return
        lowered = clean.lower()
        stage = (
            "remote_handshake"
            if any(
                marker in lowered
                for marker in (
                    "crypto_error",
                    "no recent network activity",
                    "handshake",
                    "failed to initialize client",
                    "tls:",
                    "certificate",
                    "pinned",
                    "x509",
                    "authentication failed",
                    "server rejected",
                    "connection refused",
                    "actively refused",
                    "forcibly closed",
                )
            )
            else "runtime"
        )
        if clean:
            self._emit_log(clean, stage=stage)
        if self._is_chrome_parrot_compatibility_error(clean):
            quic = (self._compatibility_config or {}).get("quic", {})
            if quic.get("disableChromeParrot"):
                self._emit_error(clean, stage="remote_handshake")
            elif not self._chrome_fallback_used and not self._chrome_fallback_pending:
                self._schedule_chrome_parrot_fallback()
            elif (
                self._chrome_fallback_used
                and not self._chrome_fallback_pending
                and not self._chrome_fallback_in_progress
            ):
                self._emit_error(clean, stage="remote_handshake")
        else:
            failure = classify_hysteria_failure(clean)
            if failure is not None:
                if self._starting and failure not in SECURITY_FAILURES:
                    # One failed probe is not the verdict of the whole start.
                    # Keep the cause and raw log until the readiness gate ends.
                    if self._last_failure_code not in SECURITY_FAILURES:
                        self._last_failure_code = failure
                elif failure in SECURITY_FAILURES or not self._running:
                    self._emit_error(clean, stage=stage, code=failure)
                else:
                    self._begin_tunnel_check(clean, stage, failure)

    @staticmethod
    def _is_chrome_parrot_compatibility_error(line: str) -> bool:
        lowered = str(line or "").lower()
        return (
            "crypto_error 0x150" in lowered
            and "(remote)" in lowered
            and "tls: internal error" in lowered
        )

    def _schedule_chrome_parrot_fallback(self) -> None:
        if self._compatibility_config is None or self._compatibility_relay_port <= 0:
            return
        self._chrome_fallback_pending = True
        self._chrome_fallback_used = True
        generation = self._compatibility_generation
        self._emit_log(
            "remote TLS internal_error; scheduling one compatibility retry "
            "without Chrome QUIC parroting",
            stage="compatibility_retry",
        )
        QTimer.singleShot(0, lambda: self._run_chrome_parrot_fallback(generation))

    def _chrome_fallback_config(self) -> dict[str, Any] | None:
        if self._compatibility_config is None or self._compatibility_relay_port <= 0:
            return None
        config = deepcopy(self._compatibility_config)
        quic = config.get("quic")
        if not isinstance(quic, dict):
            quic = {}
            config["quic"] = quic
        quic["disableChromeParrot"] = True
        return config

    def _run_chrome_parrot_fallback(self, generation: int) -> None:
        if generation != self._compatibility_generation:
            return
        if self._start_depth > 0:
            # A start is in flight: its readiness loop sees the pending flag
            # and performs the retry inline (no nested start on one QProcess).
            return
        self._chrome_fallback_pending = False
        if (
            self._compatibility_config is None
            or self._compatibility_relay_port <= 0
            or self._stop_requested
        ):
            return

        config = self._chrome_fallback_config()
        if config is None:
            return
        relay_port = self._compatibility_relay_port
        context = self._compatibility_context
        self._chrome_fallback_in_progress = True
        self._suppress_state_change = True
        runner = TransitionRunner(
            self.start_steps(
                config,
                relay_port,
                context=context,
                process_generation=self._process_generation,
                allow_parallel=self._compatibility_allow_parallel,
                verify_remote=self._compatibility_verify_remote,
                _compatibility_retry=True,
            ),
            # stop()/новый start() сбрасывают поколение совместимости —
            # тогда повтор закрывается на ближайшем шаге.
            is_current=lambda: generation == self._compatibility_generation,
            on_finished=self._on_chrome_parrot_fallback_finished,
            parent=self,
        )
        self._fallback_runner = runner
        runner.start()

    def _on_chrome_parrot_fallback_finished(self, runner: TransitionRunner) -> None:
        if self._fallback_runner is runner:
            self._fallback_runner = None
        runner.deleteLater()
        self._suppress_state_change = False
        self._chrome_fallback_in_progress = False
        started = bool(runner.result) and runner.error is None and not runner.cancelled
        if runner.cancelled:
            return
        if not started:
            if not self._running:
                self.state_changed.emit(False)
            self._emit_error(
                "Не удалось повторно запустить Hysteria2 в режиме совместимости сертификата.",
                stage="compatibility_retry",
            )
            self._clear_compatibility_state()

    def _clear_compatibility_state(self) -> None:
        self._compatibility_generation += 1
        self._compatibility_config = None
        self._compatibility_relay_port = 0
        self._compatibility_context = None
        self._chrome_fallback_pending = False
        self._chrome_fallback_used = False
        self._chrome_fallback_in_progress = False

    def _flush_stdout_buffer(self) -> None:
        self._on_ready_read()
        if self._stdout_buffer:
            tail = self._stdout_buffer
            self._stdout_buffer = ""
            self._emit_process_line(tail)

    def _on_started(self) -> None:
        # Readiness is established by the SOCKS probe in start(); do not expose
        # the process as a healthy sidecar merely because CreateProcess worked.
        self._emit_log("process started; waiting for local SOCKS relay", stage="process_started")
        self.started.emit()

    def _mark_running(self) -> None:
        if self._running:
            return
        self._running = True
        self._last_failure_code = None
        self._failure_reported = False
        self._emit_log(
            ("local SOCKS relay and authenticated Hysteria server are ready"
             if self._remote_authenticated else "local SOCKS relay and functional HTTPS handshake are ready"),
            stage="remote_ready" if self._remote_authenticated else "functional_ready",
        )
        if not self._suppress_state_change:
            self.state_changed.emit(True)

    def _on_error(self, process_error: QProcess.ProcessError) -> None:
        if self._stop_requested and process_error == QProcess.ProcessError.Crashed:
            return
        self._emit_error(
            f"Ошибка процесса Hysteria: {process_error.name} ({self._process.errorString()})",
            stage="spawn" if self._starting else "process_error",
        )

    def _on_finished(self, exit_code: int, _status: QProcess.ExitStatus) -> None:
        self._cancel_health()
        was_running = self._running
        was_starting = self._starting
        expected = self._stop_requested
        self._flush_stdout_buffer()
        self._cleanup_config()
        self._running = False
        self._starting = False
        self._stop_requested = False
        compatibility_pending = self._chrome_fallback_pending
        if not expected and not compatibility_pending:
            details = self._last_output_lines[-1] if self._last_output_lines else "без диагностического сообщения"
            self._emit_error(
                f"Hysteria неожиданно завершилась (код {exit_code}): {details}",
                stage="startup_exit" if was_starting else "unexpected_exit",
                code=self._last_failure_code or classify_hysteria_failure(details, process_exited=True),
            )
            self._clear_compatibility_state()
        elif not expected:
            self._emit_log(
                f"process exited with code {exit_code}; continuing scheduled compatibility retry",
                stage="compatibility_retry",
            )
        if was_running and not self._suppress_state_change and not compatibility_pending:
            self.state_changed.emit(False)
        self.stopped.emit(exit_code)