// Copyright 2000-2024 JetBrains s.r.o. and contributors. Use of this source code is governed by the Apache 2.0 license.
package com.jetbrains.python.debugger;

import com.google.common.collect.Maps;
import com.intellij.debugger.ui.DebuggerContentInfo;
import com.intellij.execution.configurations.RunProfile;
import com.intellij.execution.process.ProcessEvent;
import com.intellij.execution.process.ProcessHandler;
import com.intellij.execution.process.ProcessListener;
import com.intellij.execution.runners.ExecutionEnvironment;
import com.intellij.execution.ui.ConsoleView;
import com.intellij.execution.ui.ConsoleViewContentType;
import com.intellij.execution.ui.ExecutionConsole;
import com.intellij.execution.ui.RunnerLayoutUi;
import com.intellij.execution.ui.layout.LayoutAttractionPolicy;
import com.intellij.notification.NotificationGroup;
import com.intellij.notification.NotificationGroupManager;
import com.intellij.notification.NotificationType;
import com.intellij.openapi.actionSystem.*;
import com.intellij.openapi.application.ApplicationInfo;
import com.intellij.openapi.application.ReadAction;
import com.intellij.openapi.diagnostic.Logger;
import com.intellij.openapi.editor.Document;
import com.intellij.openapi.fileEditor.FileDocumentManager;
import com.intellij.openapi.fileTypes.FileTypeRegistry;
import com.intellij.openapi.progress.ProgressIndicator;
import com.intellij.openapi.progress.ProgressManager;
import com.intellij.openapi.progress.Task;
import com.intellij.openapi.project.Project;
import com.intellij.openapi.util.NlsActions;
import com.intellij.openapi.util.NlsContexts.ProgressText;
import com.intellij.openapi.util.NlsContexts.ProgressTitle;
import com.intellij.openapi.util.Pair;
import com.intellij.openapi.util.registry.Registry;
import com.intellij.openapi.util.text.StringUtil;
import com.intellij.openapi.vfs.VirtualFile;
import com.intellij.psi.PsiElement;
import com.intellij.psi.util.PsiTreeUtil;
import com.intellij.remote.RemoteSdkException;
import com.intellij.ui.ExperimentalUI;
import com.intellij.util.containers.ContainerUtil;
import com.intellij.util.ui.UIUtil;
import com.intellij.xdebugger.*;
import com.intellij.xdebugger.breakpoints.*;
import com.intellij.xdebugger.evaluation.XDebuggerEditorsProvider;
import com.intellij.xdebugger.frame.*;
import com.intellij.xdebugger.impl.ui.XDebuggerUIConstants;
import com.intellij.xdebugger.impl.ui.tree.nodes.XValueNodeEx;
import com.intellij.xdebugger.stepping.XSmartStepIntoHandler;
import com.intellij.xdebugger.ui.XDebugTabLayouter;
import com.jetbrains.python.PyBundle;
import com.jetbrains.python.PythonFileType;
import com.jetbrains.python.console.PythonConsoleView;
import com.jetbrains.python.console.PythonDebugLanguageConsoleView;
import com.jetbrains.python.console.pydev.PydevCompletionVariant;
import com.jetbrains.python.debugger.containerview.PyViewNumericContainerAction;
import com.jetbrains.python.debugger.pydev.*;
import com.jetbrains.python.debugger.pydev.dataviewer.DataViewerCommandBuilder;
import com.jetbrains.python.debugger.pydev.dataviewer.DataViewerCommandResult;
import com.jetbrains.python.debugger.settings.PyDebuggerSettings;
import com.jetbrains.python.debugger.smartstepinto.PySmartStepIntoContext;
import com.jetbrains.python.debugger.smartstepinto.PySmartStepIntoHandler;
import com.jetbrains.python.debugger.smartstepinto.PySmartStepIntoVariant;
import com.jetbrains.python.debugger.variablesview.usertyperenderers.ConfigureTypeRenderersHyperLink;
import com.jetbrains.python.debugger.variablesview.usertyperenderers.PyUserNodeRenderer;
import com.jetbrains.python.debugger.variablesview.usertyperenderers.PyUserTypeRenderersSettings;
import com.jetbrains.python.psi.PyFunction;
import com.jetbrains.python.remote.RemoteProcessControl;
import com.jetbrains.python.tables.TableCommandParameters;
import com.jetbrains.python.tables.TableCommandType;
import com.jetbrains.python.testing.AbstractPythonTestRunConfiguration;
import org.jetbrains.annotations.Nls;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;

import java.io.IOException;
import java.net.ServerSocket;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

import static com.jetbrains.python.debugger.variablesview.usertyperenderers.ConfigureTypeRenderersActionKt.getTypeRenderer;
import static com.jetbrains.python.debugger.variablesview.usertyperenderers.ConfigureTypeRenderersActionKt.loadTypeRendererChildren;
import static com.jetbrains.python.statistics.PythonDebuggerIdsHolder.CONNECTION_FAILED;


public class PyDebugProcess extends XDebugProcess implements IPyDebugProcess, ProcessListener, PyDebugProcessWithConsole {
  private static final Logger LOG = Logger.getInstance(PyDebugProcess.class);
  private static final int CONNECTION_TIMEOUT = 60000;

  public static NotificationGroup getNotificationGroup() {
    return NotificationGroupManager.getInstance().getNotificationGroup("Python Debugger");
  }

  private final ProcessDebugger myDebugger;
  private final XBreakpointHandler<?>[] myBreakpointHandlers;
  private final PyDebuggerEditorsProvider myEditorsProvider;
  private final ProcessHandler myProcessHandler;
  private final ExecutionConsole myExecutionConsole;
  private final Map<PySourcePosition, XLineBreakpoint<?>> myRegisteredBreakpoints = new ConcurrentHashMap<>();
  private final Map<String, XBreakpoint<? extends ExceptionBreakpointProperties<?>>> myRegisteredExceptionBreakpoints =
    new ConcurrentHashMap<>();

  private final Set<PyThreadInfo> mySuspendedThreads = Collections.synchronizedSet(new HashSet<>());

  private record BreakpointHitContext(@NotNull XBreakpoint<?> breakpoint,
                                      @Nullable String evaluatedLogExpression,
                                      @NotNull XSuspendContext suspendContext) {
  }

  private final List<BreakpointHitContext> myBreakpointHits = new LinkedList<>();

  private final Map<String, XValueChildrenList> myStackFrameCache = Maps.newConcurrentMap();
  private final Object myFrameCacheObject = new Object();
  private final Map<String, PyDebugValue> myNewVariableValue = Maps.newHashMap();
  private boolean myDownloadSources = false;

  protected PyPositionConverter myPositionConverter;
  private final @NotNull XSmartStepIntoHandler<?> mySmartStepIntoHandler;
  private boolean myWaitingForConnection = false;
  private PyStackFrame myConsoleContextFrame = null;
  private PyReferrersLoader myReferrersProvider;
  private final List<PyFrameListener> myFrameListeners = ContainerUtil.createLockFreeCopyOnWriteList();
  private @Nullable XCompositeNode myCurrentRootNode;

  private final Map<String, Map<String, PyDebugValueDescriptor>> myDescriptorsCache = Maps.newConcurrentMap();

  private final AtomicBoolean myInlayForFailedTestIsAlreadyShownSentinel = new AtomicBoolean();

  public PyDebugProcess(@NotNull XDebugSession session,
                        @NotNull ServerSocket serverSocket,
                        @NotNull ExecutionConsole executionConsole,
                        @Nullable ProcessHandler processHandler, boolean multiProcess) {
    this(session, multiProcess ? process -> process.createMultiprocessDebugger(serverSocket)
                               : process -> new RemoteDebugger(process, serverSocket, process.getConnectTimeout()),
         executionConsole, processHandler);
  }

  public PyDebugProcess(final @NotNull XDebugSession session,
                        final @NotNull ExecutionConsole executionConsole,
                        final @Nullable ProcessHandler processHandler,
                        @NotNull String serverHost, int serverPort) {
    this(session, process -> new ClientModeMultiProcessDebugger(process, serverHost, serverPort), executionConsole, processHandler);
  }

  private PyDebugProcess(@NotNull XDebugSession session,
                         @NotNull DebuggerFactory debuggerFactory,
                         @NotNull ExecutionConsole executionConsole,
                         @Nullable ProcessHandler processHandler) {
    super(session);

    session.setPauseActionSupported(true);

    myDebugger = debuggerFactory.createDebugger(this);

    List<XBreakpointHandler<?>> breakpointHandlers = new ArrayList<>();
    breakpointHandlers.add(new PyLineBreakpointHandler(this));
    breakpointHandlers.add(new PyExceptionBreakpointHandler(this));
    for (PyBreakpointHandlerFactory factory : PyBreakpointHandlerFactory.EP_NAME.getExtensionList()) {
      breakpointHandlers.add(factory.createBreakpointHandler(this));
    }
    myBreakpointHandlers = breakpointHandlers.toArray(XBreakpointHandler.EMPTY_ARRAY);

    myEditorsProvider = new PyDebuggerEditorsProvider();
    myProcessHandler = processHandler;
    myExecutionConsole = executionConsole;
    if (myProcessHandler != null) {
      myProcessHandler.addProcessListener(this);
    }
    if (processHandler instanceof PositionConverterProvider) {
      myPositionConverter = ((PositionConverterProvider)processHandler).createPositionConverter(this);
    }
    else {
      myPositionConverter = new PyLocalPositionConverter();
    }

    mySmartStepIntoHandler = new PySmartStepIntoHandler(this);

    PyDebugValueExecutionService executionService = PyDebugValueExecutionService.getInstance(getProject());
    executionService.sessionStarted(this);
    session.addSessionListener(new XDebugSessionListener() {
      @Override
      public void sessionStopped() {
        executionService.sessionStopped(PyDebugProcess.this);
      }
    });

    myDebugger.addCloseListener(new RemoteDebuggerCloseListener() {
      @Override
      public void closed() {
        handleStop();
      }

      @Override
      public void communicationError() {
        detachDebuggedProcess();
      }

      @Override
      public void detached() {
        detachDebuggedProcess();
      }
    });

    session.addSessionListener(new XDebugSessionListener() {
      @Override
      public void stackFrameChanged() {
        String currentFrameThreadId = null;
        final XStackFrame currentFrame = session.getCurrentStackFrame();
        if (currentFrame instanceof PyStackFrame) {
          currentFrameThreadId = ((PyStackFrame)currentFrame).getThreadId();
        }
        var suspendContext = session.getSuspendContext();
        if (suspendContext == null) {
          return;
        }
        XExecutionStack activeStack = suspendContext.getActiveExecutionStack();
        if ((activeStack == null) || (currentFrameThreadId == null)) {
          return;
        }
        final XStackFrame frameFromSuspendContext = activeStack.getTopFrame();
        String activeStackThreadId = null;
        if (frameFromSuspendContext instanceof PyStackFrame) {
          activeStackThreadId = ((PyStackFrame)frameFromSuspendContext).getThreadId();
        }
        if (!currentFrameThreadId.equals(activeStackThreadId)) {
          // another thread was selected, we should update suspendContext
          PyThreadInfo threadInfo = null;
          for (PyThreadInfo info : mySuspendedThreads) {
            if (info.getId().equals(currentFrameThreadId)) {
              threadInfo = info;
              break;
            }
          }
          if (threadInfo != null) {
            getSession().positionReached(createSuspendContext(threadInfo));
          }
        }
        PyFrameListener.publisher().frameChanged();
        for (PyFrameListener listener : myFrameListeners) {
          listener.frameChanged();
        }
      }

      @Override
      public void sessionStopped() {
        PyFrameListener.publisher().sessionStopped(null);
        XDebugSessionListener.super.sessionStopped();
        for (PyFrameListener listener : myFrameListeners) {
          listener.sessionStopped(null);
        }
      }
    });

    session.addSessionListener(new XDebugSessionListener() {
      @Override
      public void sessionStopped() {
        PyUnitTestsDebuggingService.removeInlaysAssociatedWithSession(session);
      }
    });
  }

  private MultiProcessDebugger createMultiprocessDebugger(ServerSocket serverSocket) {
    boolean useDispatcher = Registry.get("python.debugger.use.dispatcher").asBoolean();
    MultiProcessDebugger debugger = new MultiProcessDebugger(this, serverSocket, getConnectTimeout(), useDispatcher);
    debugger.addOtherDebuggerCloseListener(new MultiProcessDebugger.DebuggerProcessListener() {
      @Override
      public void threadsClosed(Set<String> threadIds) {
        for (PyThreadInfo t : mySuspendedThreads) {
          if (threadIds.contains(t.getId())) {
            if (getSession().isSuspended()) {
              getSession().resume();
              break;
            }
          }
        }
      }
    });
    return debugger;
  }

  protected void detachDebuggedProcess() {
    // We can get here because the debugger reader socket is closed, which means the debugger is exiting.
    // Now we have to wait for the main debuggee process to finish. Otherwise, `XDebugSession` will terminate it,
    // which causes the process to finish with a non-zero exit code or the `KeyboardInterrupt` exception.
    // This issue happens frequently with multiprocess debugging.
    handleStop(); // In case of normal debug, we stop the session
  }

  protected void handleStop() {
    getSession().stop();
  }

  public void setPositionConverter(PyPositionConverter positionConverter) {
    myPositionConverter = positionConverter;
  }


  @Override
  public PyPositionConverter getPositionConverter() {
    return myPositionConverter;
  }

  @Override
  public XBreakpointHandler<?> @NotNull [] getBreakpointHandlers() {
    return myBreakpointHandlers;
  }

  @Override
  public @NotNull XDebuggerEditorsProvider getEditorsProvider() {
    return myEditorsProvider;
  }

  @Override
  protected @Nullable ProcessHandler doGetProcessHandler() {
    return myProcessHandler;
  }

  @Override
  public @NotNull ExecutionConsole createConsole() {
    return myExecutionConsole;
  }

  @Override
  public @NotNull XSmartStepIntoHandler<?> getSmartStepIntoHandler() {
    return mySmartStepIntoHandler;
  }

  @Override
  public void sessionInitialized() {
    if (PyDebuggerOptionsProvider.getInstance(getProject()).isRunDebuggerInServerMode() &&
        Registry.is("python.debug.use.single.port")) {
      // In the case of a single port, the debugger is already connected at this point.
      // An additional connection attempt will result in an error.
      return;
    }
    waitForConnection(getConnectionMessage(), getConnectionTitle());
  }

  @Override
  public @NotNull XDebugTabLayouter createTabLayouter() {
    return new XDebugTabLayouter() {
      @Override
      public void registerAdditionalContent(@NotNull RunnerLayoutUi ui) {
        if (ExperimentalUI.isNewUI()) {
          ui.getDefaults().initContentAttraction(DebuggerContentInfo.CONSOLE_CONTENT,
                                                 XDebuggerUIConstants.LAYOUT_VIEW_FINISH_CONDITION,
                                                 new LayoutAttractionPolicy.FocusOnce());
        }
      }
    };
  }

  protected void waitForConnection(final @ProgressText String connectionMessage, @ProgressTitle String connectionTitle) {
    ProgressManager.getInstance().run(new Task.Backgroundable(getSession().getProject(), connectionTitle, false) {
      @Override
      public void run(final @NotNull ProgressIndicator indicator) {
        indicator.setText(connectionMessage);
        try {
          beforeConnect();
          myWaitingForConnection = true;
          myDebugger.waitForConnect();
          myWaitingForConnection = false;
          afterConnect();

          handshake();
          init();
          myDebugger.run();
        }
        catch (final Exception e) {
          myWaitingForConnection = false;
          if (myProcessHandler != null) {
            myProcessHandler.destroyProcess();
          }
          if (shouldLogConnectionException(e)) {
            getNotificationGroup()
              .createNotification(PyBundle.message("debug.notification.title.connection.failed"), e.getMessage(), NotificationType.ERROR)
              .setDisplayId(CONNECTION_FAILED)
              .notify(myProject);
          }
        }
      }
    });
  }

  protected boolean shouldLogConnectionException(final Exception e) {
    return true;
  }

  @Override
  public void init() {
    getSession().rebuildViews();
    registerBreakpoints();
    setUserTypeRenderersSettings();
    setShowReturnValues(PyDebuggerSettings.getInstance().isWatchReturnValues());
    setUnitTestDebuggingMode();
  }

  @Override
  public int handleDebugPort(int localPort) throws IOException {
    if (myProcessHandler instanceof RemoteProcessControl) {
      return getRemoteHostPortForDebuggerConnection(localPort, (RemoteProcessControl)myProcessHandler).getSecond();
    }
    else {
      return localPort;
    }
  }

  protected static Pair<String, Integer> getRemoteHostPortForDebuggerConnection(int localPort, @NotNull RemoteProcessControl handler)
    throws IOException {
    try {
      return handler.getRemoteSocket(localPort);
    }
    catch (RemoteSdkException e) {
      throw new IOException(e);
    }
  }

  @Override
  public void recordSignature(PySignature signature) {
    PySignatureCacheManager.getInstance(getSession().getProject()).recordSignature(myPositionConverter.convertSignature(signature));
  }

  @Override
  public void recordLogEvent(PyConcurrencyEvent event) {
    PyConcurrencyService.getInstance(getSession().getProject()).recordEvent(getSession(), event, event.isAsyncio());
  }

  @Override
  public void showConsole(PyThreadInfo thread) {
    myConsoleContextFrame = new PyExecutionStack(this, thread).getTopFrame();
    if (myExecutionConsole instanceof PythonDebugLanguageConsoleView consoleView) {
      UIUtil.invokeLaterIfNeeded(() -> {
        consoleView.enableConsole(false);
        consoleView.getPydevConsoleView().setConsoleEnabled(true);
      });
    }
  }

  @Override
  public void consoleInputRequested(boolean isStarted) {
    if (myExecutionConsole instanceof PythonDebugLanguageConsoleView) {
      PythonConsoleView consoleView = ((PythonDebugLanguageConsoleView)myExecutionConsole).getPydevConsoleView();
      if (isStarted) {
        consoleView.inputRequested();
      }
      else {
        consoleView.inputReceived();
      }
    }
  }

  @Override
  public void showWarning(String warningId) {
    if (warningId.equals("cython")) {
      if (!PyDebugShowingNotificationsService.getInstance().isCythonNotificationShown()) {
        PyDebugNotificationForCythonExtension.showCythonExtensionWarning(getSession().getProject());
        PyDebugShowingNotificationsService.getInstance().setCythonNotificationShown(true);
      }
    }
  }

  protected void afterConnect() {
  }

  protected void beforeConnect() {
  }

  protected @ProgressText String getConnectionMessage() {
    return PyBundle.message("python.debugger.settings.waiting.for.connection");
  }

  protected @ProgressTitle String getConnectionTitle() {
    return PyBundle.message("python.debugger.settings.connecting.to.debugger");
  }

  private void handshake() throws PyDebuggerException {
    String remoteVersion = myDebugger.handshake();
    String currentBuild = ApplicationInfo.getInstance().getBuild().asStringWithoutProductCode();
    if ("@@BUILD_NUMBER@@".equals(remoteVersion)) {
      remoteVersion = currentBuild;
    }
    else {
      remoteVersion = StringUtil.trimStart(remoteVersion, "PY-");
    }
    printToConsole(PyBundle.message("debugger.connected.to.pydev.debugger.build", remoteVersion), ConsoleViewContentType.SYSTEM_OUTPUT);

    if (!remoteVersion.startsWith(currentBuild)) {
      LOG.warn(String.format("Wrong debugger version. Remote version: %s Current build: %s", remoteVersion, currentBuild));
      printToConsole(PyBundle.message("debugger.warning.wrong.debugger.version", currentBuild), ConsoleViewContentType.ERROR_OUTPUT);
    }
  }

  @Override
  public void printToConsole(@Nls String text, ConsoleViewContentType contentType) {
    ((ConsoleView)myExecutionConsole).print(text, contentType);
  }

  private void registerBreakpoints() {
    registerLineBreakpoints();
    registerExceptionBreakpoints();
  }

  private void registerExceptionBreakpoints() {
    for (XBreakpoint<? extends ExceptionBreakpointProperties<?>> bp : myRegisteredExceptionBreakpoints.values()) {
      addExceptionBreakpoint(bp);
    }
  }

  @Override
  public void setUserTypeRenderersSettings() {
    PyUserTypeRenderersSettings settings = PyUserTypeRenderersSettings.getInstance();
    if (settings == null) {
      return;
    }
    if (isConnected()) {
      List<PyUserTypeRenderer> renderersList = settings.getApplicableRenderers();
      if (!renderersList.isEmpty()) {
        myDebugger.setUserTypeRenderers(renderersList);
      }
    }
  }

  @Override
  public @Nullable XDebuggerTreeNodeHyperlink getUserTypeRenderersLink(@NotNull String typeRendererId) {
    return new ConfigureTypeRenderersHyperLink(typeRendererId, getProject(), null);
  }

  public void registerLineBreakpoints() {
    for (Map.Entry<PySourcePosition, XLineBreakpoint<?>> entry : myRegisteredBreakpoints.entrySet()) {
      addBreakpoint(entry.getKey(), entry.getValue());
    }
  }

  @Override
  public void registerAdditionalActions(@NotNull DefaultActionGroup leftToolbar,
                                        @NotNull DefaultActionGroup topToolbar,
                                        @NotNull DefaultActionGroup settings) {
    super.registerAdditionalActions(leftToolbar, topToolbar, settings);
    settings.add(new WatchReturnValuesAction(this));
    settings.add(new PyVariableViewSettings.SimplifiedView(this));
    settings.add(new PyVariableViewSettings.VariablesPolicyGroup());
    settings.add(new PyVariableViewSettings.QuotingPolicyGroup());
  }

  private static final class WatchReturnValuesAction extends ToggleAction {
    private volatile boolean myWatchesReturnValues;
    private final PyDebugProcess myProcess;
    private final @NlsActions.ActionText String myText;

    WatchReturnValuesAction(@NotNull PyDebugProcess debugProcess) {
      super("", PyBundle.message("debugger.watch.return.values.description"), null);
      myWatchesReturnValues = PyDebuggerSettings.getInstance().isWatchReturnValues();
      myProcess = debugProcess;
      myText = PyBundle.message("debugger.watch.show.return.values");
    }

    @Override
    public void update(final @NotNull AnActionEvent e) {
      super.update(e);
      final Presentation presentation = e.getPresentation();
      presentation.setEnabled(true);
      presentation.setText(myText);
    }

    @Override
    public @NotNull ActionUpdateThread getActionUpdateThread() {
      return ActionUpdateThread.BGT;
    }

    @Override
    public boolean isSelected(@NotNull AnActionEvent e) {
      return myWatchesReturnValues;
    }

    @Override
    public void setSelected(@NotNull AnActionEvent e, boolean watch) {
      myWatchesReturnValues = watch;
      PyDebuggerSettings.getInstance().setWatchReturnValues(watch);
      final Project project = e.getProject();
      if (project != null) {
        myProcess.setShowReturnValues(myWatchesReturnValues);
        myProcess.getSession().rebuildViews();
      }
    }
  }

  public void setShowReturnValues(boolean showReturnValues) {
    myDebugger.setShowReturnValues(showReturnValues);
  }

  public void setUnitTestDebuggingMode() {
    ExecutionEnvironment environment = getSession().getExecutionEnvironment();
    if (environment == null) return;
    RunProfile runProfile = environment.getRunProfile();
    if (runProfile instanceof AbstractPythonTestRunConfiguration
        && PyDebuggerOptionsProvider.getInstance(getProject()).isDropIntoDebuggerOnFailedTest()) {
      myDebugger.setUnitTestDebuggingMode();
    }
  }

  @Override
  public void startStepOver(@Nullable XSuspendContext context) {
    passToCurrentThread(context, ResumeOrStepCommand.Mode.STEP_OVER);
  }

  @Override
  public void startStepInto(@Nullable XSuspendContext context) {
    passToCurrentThread(context, ResumeOrStepCommand.Mode.STEP_INTO);
  }

  public void startStepIntoMyCode(@Nullable XSuspendContext context) {
    if (!checkCanPerformCommands()) return;
    getSession().sessionResumed();
    passToCurrentThread(context, ResumeOrStepCommand.Mode.STEP_INTO_MY_CODE);
  }

  public void startSetNextStatement(@Nullable XSuspendContext context,
                                    @NotNull XSourcePosition sourcePosition,
                                    @NotNull PyDebugCallback<Pair<Boolean, String>> callback) {
    if (!checkCanPerformCommands()) return;
    dropFrameCaches();
    if (isConnected()) {
      String threadId = threadIdBeforeResumeOrStep(context);
      for (PyThreadInfo suspendedThread : mySuspendedThreads) {
        if (threadId != null && threadId.equals(suspendedThread.getId())) {
          myDebugger.setNextStatement(threadId, sourcePosition, getFunctionName(sourcePosition), callback);
          break;
        }
      }
    }
  }

  @Override
  public void startStepOut(@Nullable XSuspendContext context) {
    passToCurrentThread(context, ResumeOrStepCommand.Mode.STEP_OUT);
  }

  public void startSmartStepInto(@NotNull PySmartStepIntoVariant variant) {
    String threadId = variant.getContext().getFrame().getThreadId();
    String frameId = variant.getContext().getFrame().getFrameId();
    if (isConnected()) {
      dropFrameCaches();
      for (PyThreadInfo suspendedThread : mySuspendedThreads) {
        if (threadId.equals(suspendedThread.getId())) {
          PySmartStepIntoContext context = variant.getContext();
          myDebugger.smartStepInto(threadId, frameId, variant.getFunctionName(), variant.getCallOrder(),
                                   context.getStartLine(), context.getEndLine());
          break;
        }
      }
    }
  }

  public @NotNull List<Pair<String, Boolean>> getSmartStepIntoVariants(int startContextLine, int endContextLine) {
    try {
      PyStackFrame frame = currentFrame();
      var smartStepIntoVariants = myDebugger.getSmartStepIntoVariants(
        frame.getThreadId(), frame.getFrameId(), startContextLine, endContextLine);
      return smartStepIntoVariants != null ? smartStepIntoVariants : Collections.emptyList();
    }
    catch (PyDebuggerException e) {
      return Collections.emptyList();
    }
  }

  @Override
  public void stop() {
    myDebugger.close();
  }

  @Override
  public void resume(@Nullable XSuspendContext context) {
    if (myBreakpointHits.isEmpty()) {
      passResumeToAllThreads();
    }
    else {
      var breakpointHitContext = myBreakpointHits.removeFirst();
      var shouldStop = getSession().breakpointReached(breakpointHitContext.breakpoint, breakpointHitContext.evaluatedLogExpression,
                                                      breakpointHitContext.suspendContext);
      if (!shouldStop) {
        resume(breakpointHitContext.suspendContext);
      }
    }
  }

  @Override
  public void startPausing() {
    if (isConnected()) {
      myDebugger.suspendAllThreads();
    }
  }

  @Override
  public void suspendAllOtherThreads(PyThreadInfo thread) {
    myDebugger.suspendOtherThreads(thread);
  }

  /**
   * Check if there is the thread suspended on the breakpoint with the "Suspend all" policy
   *
   * @return true if this thread exists
   */
  @Override
  public boolean isSuspendedOnAllThreadsPolicy() {
    if (getSession().isSuspended()) {
      for (PyThreadInfo threadInfo : getThreads()) {
        final List<PyStackFrameInfo> frames = threadInfo.getFrames();
        if ((threadInfo.getState() == PyThreadInfo.State.SUSPENDED) && (frames != null)) {
          XBreakpoint<?> breakpoint = null;
          if (threadInfo.isStopOnBreakpoint()) {
            final PySourcePosition position = frames.getFirst().getPosition();
            breakpoint = myRegisteredBreakpoints.get(position);
          }
          else if (threadInfo.isExceptionBreak()) {
            String exceptionName = threadInfo.getMessage();
            if (exceptionName != null) {
              breakpoint = myRegisteredExceptionBreakpoints.get(exceptionName);
            }
          }
          if ((breakpoint != null) && (breakpoint.getType().isSuspendThreadSupported()) &&
              (breakpoint.getSuspendPolicy() == SuspendPolicy.ALL)) {
            return true;
          }
        }
      }
    }
    return false;
  }

  private void passResumeToAllThreads() {
    dropFrameCaches();
    myInlayForFailedTestIsAlreadyShownSentinel.set(false);
    if (isConnected()) {
      for (PyThreadInfo thread : myDebugger.getThreads()) {
        myDebugger.resumeOrStep(thread.getId(), ResumeOrStepCommand.Mode.RESUME);
      }
    }
  }

  private void passToCurrentThread(@Nullable XSuspendContext context, final ResumeOrStepCommand.Mode mode) {
    dropFrameCaches();
    myBreakpointHits.clear();
    if (isConnected()) {
      String threadId = threadIdBeforeResumeOrStep(context);

      for (PyThreadInfo suspendedThread : mySuspendedThreads) {
        if (threadId == null || threadId.equals(suspendedThread.getId())) {
          myDebugger.resumeOrStep(suspendedThread.getId(), mode);
          break;
        }
      }
    }
  }

  private static @Nullable String threadIdBeforeResumeOrStep(@Nullable XSuspendContext context) {
    if (context instanceof PySuspendContext) {
      return ((PySuspendContext)context).getActiveExecutionStack().getThreadId();
    }
    else {
      return null;
    }
  }

  public boolean isConnected() {
    return myDebugger.isConnected();
  }

  protected void disconnect() {
    myDebugger.disconnect();
    cleanUp();
  }

  public boolean isDownloadSources() {
    return myDownloadSources;
  }

  public void setDownloadSources(boolean downloadSources) {
    myDownloadSources = downloadSources;
  }

  protected void cleanUp() {
    mySuspendedThreads.clear();
    myDownloadSources = false;
  }

  @Override
  public void runToPosition(final @NotNull XSourcePosition position, @Nullable XSuspendContext context) {
    dropFrameCaches();
    if (isConnected() && !mySuspendedThreads.isEmpty()) {
      final PySourcePosition pyPosition = myPositionConverter.convertToPython(position);
      String type =
        ReadAction.compute(() -> {
          String breakpointTypeId = PyLineBreakpointType.ID;
          final Document document = FileDocumentManager.getInstance().getDocument(position.getFile());
          if (document != null) {
            for (XBreakpointType<?, ?> breakpointType : XBreakpointType.EXTENSION_POINT_NAME.getExtensionList()) {
              if (breakpointType instanceof PyBreakpointType &&
                  ((PyBreakpointType)breakpointType).isBreakpointTypeAllowedInDocument(getSession().getProject(), document)) {
                breakpointTypeId = breakpointType.getId();
                break;
              }
            }
          }
          return breakpointTypeId;
        });
      myDebugger.setTempBreakpoint(type, pyPosition.getFile(), pyPosition.getLine());

      passToCurrentThread(context, ResumeOrStepCommand.Mode.RESUME);
    }
  }

  @Override
  public PyDebugValue evaluate(final String expression, final boolean execute, boolean doTrunc) throws PyDebuggerException {
    dropFrameCaches();
    final PyStackFrame frame = currentFrame();
    return evaluate(expression, execute, frame, doTrunc);
  }

  private PyDebugValue evaluate(String expression, boolean execute, PyStackFrame frame, boolean trimResult) throws PyDebuggerException {
    PyDebuggerOptionsProvider settings = PyDebuggerOptionsProvider.getInstance(getProject());
    return myDebugger.evaluate(frame.getThreadId(), frame.getFrameId(), expression, execute,
                               settings.getEvaluationResponseTimeout(), trimResult);
  }

  @Override
  public void consoleExec(String command, @NotNull PyDebugCallback<String> callback) {
    dropFrameCaches();
    try {
      final PyStackFrame frame = currentFrame();
      myDebugger.consoleExec(frame.getThreadId(), frame.getFrameId(), command, callback);
    }
    catch (PyDebuggerException e) {
      callback.error(e);
    }
  }

  @Override
  public String execTableCommand(String command, TableCommandType commandType, TableCommandParameters tableCommandParameters)
    throws PyDebuggerException {
    final PyStackFrame frame = currentFrame();
    return myDebugger.execTableCommand(frame.getThreadId(), frame.getFrameId(), command, commandType, tableCommandParameters);
  }

  @Override
