[rd launcher] improve stability of backend output parsing

GitOrigin-RevId: 437a4aec9dc941bd73b879558df865b493ed3081

build/launch/src/com/intellij/tools/launch/Launcher.kt

@@ -6,11 +6,16 @@ import com.intellij.tools.launch.ide.environments.docker.legacyDockerRunCliComma
 import com.intellij.tools.launch.ide.environments.local.LocalIdeCommandLauncherFactory
 import com.intellij.tools.launch.ide.environments.local.localLaunchOptions
 import com.intellij.tools.launch.os.ProcessOutputStrategy
+import kotlinx.coroutines.CoroutineName
+import kotlinx.coroutines.CoroutineScope
 import org.jetbrains.jps.model.java.JpsJavaClasspathKind
 import java.net.InetAddress
 import java.net.ServerSocket
 
 object Launcher {
+  @Suppress("SSBasedInspection")
+  private val launcherLifespanScope = CoroutineScope(CoroutineName("RemoteDev Launcher"))
+
   fun launch(
     paths: PathsProvider,
     modulesToScopes: Map<String, JpsJavaClasspathKind>,
@@ -50,6 +55,8 @@ object Launcher {
         localLaunchOptions(
           beforeProcessStart = options.beforeProcessStart,
           processOutputStrategy = if (options.redirectOutputIntoParentProcess) ProcessOutputStrategy.InheritIO else ProcessOutputStrategy.RedirectToFiles(paths.logFolder),
+          processTitle = "IDE backend (local process)",
+          lifespanScope = launcherLifespanScope
         )
       )
       return IdeLauncher.launchCommand(localLauncherFactory, ideLaunchContext).process to null

build/launch/src/com/intellij/tools/launch/ide/environments/docker/DockerRunCliCommandLauncher.kt

@@ -9,16 +9,53 @@ import com.intellij.tools.launch.ide.ClassPathBuilder
 import com.intellij.tools.launch.ide.ClasspathCollector
 import com.intellij.tools.launch.ide.IdeCommandLauncherFactory
 import com.intellij.tools.launch.ide.IdePathsInLaunchEnvironment
+import com.intellij.tools.launch.os.ProcessOutputInfo
+import com.intellij.tools.launch.os.ProcessWrapper
+import com.intellij.tools.launch.os.asyncAwaitExit
+import com.intellij.tools.launch.os.produceOutputFlows
+import kotlinx.coroutines.CoroutineName
+import kotlinx.coroutines.CoroutineScope
+import org.jetbrains.annotations.ApiStatus.Obsolete
+import java.util.concurrent.locks.ReentrantLock
+import kotlin.concurrent.withLock
 
 interface LocalDockerRunResult {
+  /**
+   * Use [processWrapper] instead. If [ProcessWrapper] lacks certain properties feel consider adding them instead of switching to [process].
+   */
+  @get:Obsolete
   val process: Process
+  val processWrapper: ProcessWrapper
   val containerId: String
 }
 
+/**
+ * Note! [process] is still used in [com.intellij.tools.launch.Launcher]. [processWrapperFactory] factory allows on-demand consumption of
+ * the process output without interfering with the direct usages of input and error streams of the process.
+ */
 private data class LocalDockerRunResultImpl(
-  override val process: Process,
+  private val underlyingProcess: Process,
   override val containerId: String,
-) : LocalDockerRunResult
+  private val processWrapperFactory: (Process) -> ProcessWrapper,
+) : LocalDockerRunResult {
+  private var _processWrapper: ProcessWrapper? = null
+
+  /**
+   * Guards modification of [_processWrapper] and access to [process] and [processWrapper].
+   */
+  private val lock = ReentrantLock()
+
+  override val process: Process
+    get() = lock.withLock {
+      require(_processWrapper == null) { "`process` is wrapped, use `processWrapper` instead" }
+      underlyingProcess
+    }
+
+  override val processWrapper: ProcessWrapper
+    get() = lock.withLock {
+      _processWrapper ?: processWrapperFactory(underlyingProcess).also { _processWrapper = it }
+    }
+}
 
 fun dockerRunCliCommandLauncherFactory(dockerContainerOptions: DockerContainerOptions): DockerRunCliCommandLauncherFactory {
   val dockerLauncherOptions = DockerLauncherOptionsImpl(
@@ -28,8 +65,10 @@ fun dockerRunCliCommandLauncherFactory(dockerContainerOptions: DockerContainerOp
   return DockerRunCliCommandLauncherFactory(dockerContainerOptions, dockerLauncherOptions)
 }
 
-fun legacyDockerRunCliCommandLauncherFactory(dockerLauncherOptions: DockerLauncherOptions,
-                                             paths: PathsProvider): DockerRunCliCommandLauncherFactory {
+fun legacyDockerRunCliCommandLauncherFactory(
+  dockerLauncherOptions: DockerLauncherOptions,
+  paths: PathsProvider,
+): DockerRunCliCommandLauncherFactory {
   val dockerContainerOptions = DockerContainerOptions(
     image = dockerLauncherOptions.dockerImageName,
     containerName = dockerLauncherOptions.containerName,
@@ -42,10 +81,12 @@ fun legacyDockerRunCliCommandLauncherFactory(dockerLauncherOptions: DockerLaunch
 
 class DockerRunCliCommandLauncherFactory(
   private val dockerContainerOptions: DockerContainerOptions,
-  private val dockerLauncherOptions: DockerLauncherOptions
+  private val dockerLauncherOptions: DockerLauncherOptions,
 ) : IdeCommandLauncherFactory<LocalDockerRunResult> {
-  override fun create(localPaths: PathsProvider,
-                      classpathCollector: ClasspathCollector): Pair<AbstractCommandLauncher<LocalDockerRunResult>, IdePathsInLaunchEnvironment> {
+  override fun create(
+    localPaths: PathsProvider,
+    classpathCollector: ClasspathCollector,
+  ): Pair<AbstractCommandLauncher<LocalDockerRunResult>, IdePathsInLaunchEnvironment> {
     val dockerContainerEnvironment = DockerContainerEnvironment.createDefaultDockerContainerEnvironment(dockerContainerOptions, localPaths)
     // collect classpath with the paths within the Docker container (the container does not exist at this point, it is going to be created later)
     val classpathInDocker = classpathCollector.collect(dockerContainerEnvironment)
@@ -70,6 +111,18 @@ private class DockerRunCliCommandLauncher(
     val launchCommand = dockerContainerEnvironment.buildCommand()
     val dockerLauncher = DockerLauncher(localPaths, dockerLauncherOptions)
     val (process, containerId) = dockerLauncher.runInContainer(dockerContainerEnvironment, launchCommand, dockerContainerOptions)
-    return LocalDockerRunResultImpl(process, containerId)
+    return LocalDockerRunResultImpl(process, containerId, processWrapperFactory(processTitle = "Docker container ($containerId)"))
   }
-}
+}
+
+@Suppress("SSBasedInspection")
+private fun createLifespanScope(processTitle: String): CoroutineScope = CoroutineScope(CoroutineName(processTitle))
+
+private fun processWrapperFactory(processTitle: String): (Process) -> ProcessWrapper =
+  { process ->
+    val lifespanScope = createLifespanScope(processTitle)
+    ProcessWrapper(
+      processOutputInfo = ProcessOutputInfo.Piped(process.produceOutputFlows(lifespanScope)),
+      terminationDeferred = process.asyncAwaitExit(lifespanScope, processTitle)
+    )
+  }

build/launch/src/com/intellij/tools/launch/ide/environments/local/LocalLaunchEnvironment.kt

@@ -7,11 +7,14 @@ import com.intellij.tools.launch.ide.ClassPathBuilder
 import com.intellij.tools.launch.ide.ClasspathCollector
 import com.intellij.tools.launch.ide.IdeCommandLauncherFactory
 import com.intellij.tools.launch.ide.IdePathsInLaunchEnvironment
-import com.intellij.tools.launch.os.ProcessOutputInfo
 import com.intellij.tools.launch.os.ProcessOutputStrategy
+import com.intellij.tools.launch.os.ProcessWrapper
 import com.intellij.tools.launch.os.affixIO
+import com.intellij.tools.launch.os.asyncAwaitExit
 import com.intellij.util.SystemProperties
 import com.sun.security.auth.module.UnixSystem
+import kotlinx.coroutines.CoroutineScope
+import org.jetbrains.annotations.ApiStatus.Obsolete
 import java.nio.file.Path
 import java.nio.file.Paths
 
@@ -33,25 +36,35 @@ object LocalLaunchEnvironment : LaunchEnvironment {
 interface LocalLaunchOptions {
   val beforeProcessStart: () -> Unit
   val processOutputStrategy: ProcessOutputStrategy
+
+  /**
+   * Process title used for diagnostic purposes (f.e. log records, names of associated threads and coroutines).
+   */
+  val processTitle: String
+  val lifespanScope: CoroutineScope
 }
 
 private data class LocalLaunchOptionsImpl(
   override val beforeProcessStart: () -> Unit,
   override val processOutputStrategy: ProcessOutputStrategy,
+  override val processTitle: String,
+  override val lifespanScope: CoroutineScope,
 ) : LocalLaunchOptions
 
 fun localLaunchOptions(
   beforeProcessStart: () -> Unit = {},
   processOutputStrategy: ProcessOutputStrategy,
+  processTitle: String,
+  lifespanScope: CoroutineScope,
 ): LocalLaunchOptions =
-  LocalLaunchOptionsImpl(beforeProcessStart, processOutputStrategy)
+  LocalLaunchOptionsImpl(beforeProcessStart, processOutputStrategy, processTitle, lifespanScope)
 
 class LocalCommandLauncher(private val localLaunchOptions: LocalLaunchOptions) : AbstractCommandLauncher<LocalProcessLaunchResult> {
   override fun launch(buildCommand: LaunchEnvironment.() -> LaunchCommand): LocalProcessLaunchResult {
     val (commandLine, environment) = LocalLaunchEnvironment.buildCommand()
     val processBuilder = ProcessBuilder(commandLine)
 
-    val processOutputStrategy = processBuilder.affixIO(localLaunchOptions.processOutputStrategy)
+    val processOutputInfoFactory = processBuilder.affixIO(localLaunchOptions.processOutputStrategy)
     processBuilder.environment().putAll(environment)
     localLaunchOptions.beforeProcessStart()
 
@@ -59,7 +72,14 @@ class LocalCommandLauncher(private val localLaunchOptions: LocalLaunchOptions) :
     logger.info(processBuilder.command().joinToString("\n"))
     logger.info("-- END")
 
-    return LocalProcessLaunchResult(process = processBuilder.start(), processOutputStrategy)
+    val process = processBuilder.start()
+    val processOutputInfo = processOutputInfoFactory(localLaunchOptions.lifespanScope, process)
+    val processWrapper = ProcessWrapper(
+      processOutputInfo = processOutputInfo,
+      terminationDeferred = process.asyncAwaitExit(localLaunchOptions.lifespanScope, processTitle = "")
+    )
+
+    return LocalProcessLaunchResult(process, processWrapper)
   }
 
   companion object {
@@ -68,8 +88,9 @@ class LocalCommandLauncher(private val localLaunchOptions: LocalLaunchOptions) :
 }
 
 data class LocalProcessLaunchResult(
+  @Obsolete
   val process: Process,
-  val processOutputInfo: ProcessOutputInfo,
+  val processWrapper: ProcessWrapper,
 )
 
 class LocalIdeCommandLauncherFactory(private val localLaunchOptions: LocalLaunchOptions) : IdeCommandLauncherFactory<LocalProcessLaunchResult> {

build/launch/src/com/intellij/tools/launch/os/ProcessWrapper.kt

@@ -0,0 +1,11 @@
+package com.intellij.tools.launch.os
+
+import kotlinx.coroutines.Deferred
+
+/**
+ * This abstraction allows consuming the underlying process's streams by several consumers.
+ */
+data class ProcessWrapper(
+  val processOutputInfo: ProcessOutputInfo,
+  val terminationDeferred: Deferred<Int>,
+)

build/launch/src/com/intellij/tools/launch/os/Processes.kt

@@ -1,17 +1,16 @@
 package com.intellij.tools.launch.os
 
 import com.intellij.util.io.awaitExit
-import kotlinx.coroutines.CoroutineName
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Deferred
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.async
+import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.SharingStarted
+import kotlinx.coroutines.flow.asFlow
+import kotlinx.coroutines.flow.shareIn
 import java.io.File
 import java.nio.file.Path
 
 sealed interface ProcessOutputInfo {
-  data object Piped : ProcessOutputInfo
+  data class Piped(val outputFlows: ProcessOutputFlows) : ProcessOutputInfo
 
   data object InheritedByParent : ProcessOutputInfo
 
@@ -29,12 +28,14 @@ sealed interface ProcessOutputStrategy {
   data class RedirectToFiles(val logFolder: File) : ProcessOutputStrategy
 }
 
-fun ProcessBuilder.affixIO(strategy: ProcessOutputStrategy): ProcessOutputInfo =
-  when (strategy) {
-    ProcessOutputStrategy.Pipe -> ProcessOutputInfo.Piped
+fun ProcessBuilder.affixIO(strategy: ProcessOutputStrategy): CoroutineScope.(Process) -> ProcessOutputInfo {
+  return when (strategy) {
+    ProcessOutputStrategy.Pipe -> {
+      { process -> ProcessOutputInfo.Piped(process.produceOutputFlows(coroutineScope = this)) }
+    }
     ProcessOutputStrategy.InheritIO -> {
       this.inheritIO()
-      ProcessOutputInfo.InheritedByParent
+      return { ProcessOutputInfo.InheritedByParent }
     }
     is ProcessOutputStrategy.RedirectToFiles -> {
       strategy.logFolder.mkdirs()
@@ -43,11 +44,28 @@ fun ProcessBuilder.affixIO(strategy: ProcessOutputStrategy): ProcessOutputInfo =
       val stderrFile = strategy.logFolder.resolve("err-$ts.log")
       this.redirectOutput(stdoutFile)
       this.redirectError(stderrFile)
-      ProcessOutputInfo.RedirectedToFiles(stdoutFile.toPath(), stderrFile.toPath())
+      return { ProcessOutputInfo.RedirectedToFiles(stdoutFile.toPath(), stderrFile.toPath()) }
     }
   }
+}
 
-suspend fun Process.asyncAwaitExit(coroutineScope: CoroutineScope, processTitle: String): Deferred<Int> =
-  coroutineScope.async(Dispatchers.IO + SupervisorJob() + CoroutineName("$processTitle | await for termination")) {
+fun Process.asyncAwaitExit(lifespanScope: CoroutineScope, processTitle: String): Deferred<Int> =
+  lifespanScope.async(Dispatchers.IO + SupervisorJob() + CoroutineName("$processTitle | await for termination")) {
     awaitExit()
-  }
+  }
+
+data class ProcessOutputFlows(val stdout: Flow<String>, val stderr: Flow<String>)
+
+fun Process.produceOutputFlows(coroutineScope: CoroutineScope): ProcessOutputFlows =
+  ProcessOutputFlows(
+    stdout = inputStream
+      .bufferedReader()
+      .lineSequence()
+      .asFlow()
+      .shareIn(coroutineScope, SharingStarted.Lazily),
+    stderr = errorStream
+      .bufferedReader()
+      .lineSequence()
+      .asFlow()
+      .shareIn(coroutineScope, SharingStarted.Lazily),
+  )

build/launch/src/com/intellij/tools/launch/rd/RemoteDevLauncher.kt

@@ -3,7 +3,6 @@ package com.intellij.tools.launch.rd
 import com.intellij.openapi.diagnostic.logger
 import com.intellij.tools.launch.docker.BindMount
 import com.intellij.tools.launch.os.ProcessOutputInfo
-import com.intellij.tools.launch.os.asyncAwaitExit
 import com.intellij.tools.launch.os.terminal.AnsiColor
 import com.intellij.tools.launch.os.terminal.colorize
 import com.intellij.tools.launch.rd.RemoteDevLauncher.logger
@@ -11,20 +10,13 @@ import com.intellij.tools.launch.rd.components.BackendLaunchResult
 import com.intellij.tools.launch.rd.components.runCodeWithMeHostNoLobby
 import com.intellij.tools.launch.rd.components.runJetBrainsClientLocally
 import com.intellij.tools.launch.rd.dsl.*
-import kotlinx.coroutines.CoroutineName
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Deferred
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.coroutineScope
-import kotlinx.coroutines.launch
-import kotlinx.coroutines.withContext
+import kotlinx.coroutines.*
 import java.nio.file.Path
 
 suspend fun CoroutineScope.launchRemoteDev(init: LaunchRemoteDevBuilder.() -> Unit): LaunchRemoteDevResult =
   LaunchRemoteDevBuilderImpl().let {
     it.init()
-    it.launch(coroutineScope = this)
+    it.launch(remoteDevLifespanScope = this)
   }
 
 data class LaunchRemoteDevResult(val clientTerminationDeferred: Deferred<Int>, val backendTerminationDeferred: Deferred<Int>)
@@ -56,10 +48,10 @@ private class LaunchRemoteDevBuilderImpl : LaunchRemoteDevBuilder {
     )
   }
 
-  suspend fun launch(coroutineScope: CoroutineScope): LaunchRemoteDevResult {
+  suspend fun launch(remoteDevLifespanScope: CoroutineScope): LaunchRemoteDevResult {
     logger.info("Starting backend: $backendInEnvDescription")
-    val backendLaunchResult = runCodeWithMeHostNoLobby(backendInEnvDescription)
-    handleBackendProcessOutput(backendLaunchResult, coroutineScope)
+    val backendLaunchResult = runCodeWithMeHostNoLobby(backendInEnvDescription, remoteDevLifespanScope)
+    handleBackendProcessOutput(backendLaunchResult, remoteDevLifespanScope)
     val backendDebugPort = 5006
     logger.info("Attaching debugger to backend at port $backendDebugPort")
     backendInEnvDescription.backendDescription.attachDebuggerCallback?.invoke(backendDebugPort)
@@ -70,25 +62,23 @@ private class LaunchRemoteDevBuilderImpl : LaunchRemoteDevBuilder {
       error("Backend failed to start normally")
     }
     logger.info("Starting JetBrains client")
-    val (clientLocalProcessResult, clientDebugPort) = coroutineScope {
+    val (clientLocalProcessResult, clientDebugPort) =
       withContext(CoroutineName("JetBrains Client Launcher")) {
-        runJetBrainsClientLocally()
+        runJetBrainsClientLocally(remoteDevLifespanScope)
       }
-    }
     handleLocalProcessOutput(
-      clientLocalProcessResult.process,
-      clientLocalProcessResult.processOutputInfo,
-      processShortName = "JetBrains Client",
+      clientLocalProcessResult.processWrapper.processOutputInfo,
+      processShortName = "\uD83D\uDE80 JetBrains Client",
       processColor = AnsiColor.GREEN,
-      coroutineScope
+      remoteDevLifespanScope
     )
     clientResult.attachDebuggerCallback?.let {
       logger.info("Attaching debugger to client at port $clientDebugPort")
       it.invoke(clientDebugPort)
     }
     return LaunchRemoteDevResult(
-      clientTerminationDeferred = clientLocalProcessResult.process.asyncAwaitExit(coroutineScope, processTitle = "JetBrains Client"),
-      backendTerminationDeferred = backendLaunchResult.asyncAwaitExit(coroutineScope)
+      clientTerminationDeferred = clientLocalProcessResult.processWrapper.terminationDeferred,
+      backendTerminationDeferred = backendLaunchResult.asyncAwaitExit(remoteDevLifespanScope)
     )
   }
 }
@@ -96,15 +86,13 @@ private class LaunchRemoteDevBuilderImpl : LaunchRemoteDevBuilder {
 private suspend fun handleBackendProcessOutput(backendLaunchResult: BackendLaunchResult, coroutineScope: CoroutineScope) {
   when (backendLaunchResult) {
     is BackendLaunchResult.Local -> handleLocalProcessOutput(
-      backendLaunchResult.localProcessResult.process,
-      backendLaunchResult.localProcessResult.processOutputInfo,
+      backendLaunchResult.localProcessResult.processWrapper.processOutputInfo,
       processShortName = "IDE Backend",
       processColor = AnsiColor.PURPLE,
       coroutineScope
     )
     is BackendLaunchResult.Docker -> handleLocalProcessOutput(
-      backendLaunchResult.localDockerRunResult.process,
-      ProcessOutputInfo.Piped,
+      backendLaunchResult.localDockerRunResult.processWrapper.processOutputInfo,
       processShortName = "\uD83D\uDC33 IDE Backend",
       processColor = AnsiColor.CYAN,
       coroutineScope
@@ -112,8 +100,7 @@ private suspend fun handleBackendProcessOutput(backendLaunchResult: BackendLaunc
   }
 }
 
-private suspend fun handleLocalProcessOutput(
-  process: Process,
+private fun handleLocalProcessOutput(
   outputStrategy: ProcessOutputInfo,
   processShortName: String,
   processColor: AnsiColor,
@@ -121,14 +108,14 @@ private suspend fun handleLocalProcessOutput(
 ) {
   val prefix = colorize("$processShortName\t| ", processColor)
   when (outputStrategy) {
-    ProcessOutputInfo.Piped -> {
+    is ProcessOutputInfo.Piped -> {
       coroutineScope.launch(Dispatchers.IO + SupervisorJob() + CoroutineName("$processShortName | stdout")) {
-        process.inputReader().lines().forEach { line ->
+        outputStrategy.outputFlows.stdout.collect { line ->
           println("$prefix$line")
         }
       }
       coroutineScope.launch(Dispatchers.IO + SupervisorJob() + CoroutineName("$processShortName | stderr")) {
-        process.errorReader().lines().forEach { line ->
+        outputStrategy.outputFlows.stderr.collect { line ->
           println("$prefix${colorize(line, AnsiColor.RED)}")
         }
       }
@@ -144,8 +131,8 @@ private suspend fun handleLocalProcessOutput(
 
 private suspend fun BackendLaunchResult.asyncAwaitExit(coroutineScope: CoroutineScope): Deferred<Int> =
   when (this) {
-    is BackendLaunchResult.Local -> localProcessResult.process.asyncAwaitExit(coroutineScope, processTitle = "IDE Backend")
-    is BackendLaunchResult.Docker -> localDockerRunResult.process.asyncAwaitExit(coroutineScope, processTitle = "IDE Backend (Docker)")
+    is BackendLaunchResult.Local -> localProcessResult.processWrapper.terminationDeferred
+    is BackendLaunchResult.Docker -> localDockerRunResult.processWrapper.terminationDeferred
   }
 
 internal sealed interface BackendInEnvDescription {

build/launch/src/com/intellij/tools/launch/rd/components/JetBrainClient.kt

@@ -19,7 +19,7 @@ data class JetBrainsClientLaunchResult(
   val debugPort: Int,
 )
 
-suspend fun CoroutineScope.runJetBrainsClientLocally(): JetBrainsClientLaunchResult {
+fun runJetBrainsClientLocally(clientProcessLifespanScope: CoroutineScope): JetBrainsClientLaunchResult {
   JetBrainClient.logger.info("Starting JetBrains client")
   val paths = JetBrainsClientIdeaPathsProvider()
   val classpath = classpathCollector(
@@ -29,7 +29,11 @@ suspend fun CoroutineScope.runJetBrainsClientLocally(): JetBrainsClientLaunchRes
   )
   val debugPort = 5007
   val localProcessLaunchResult = IdeLauncher.launchCommand(
-    LocalIdeCommandLauncherFactory(localLaunchOptions(processOutputStrategy = ProcessOutputStrategy.Pipe)),
+    LocalIdeCommandLauncherFactory(localLaunchOptions(
+      processOutputStrategy = ProcessOutputStrategy.Pipe,
+      processTitle = "JetBrains Client",
+      lifespanScope = clientProcessLifespanScope
+    )),
     context = IdeLaunchContext(
       classpathCollector = classpath,
       // changed in Java 9, now we have to use *: to listen on all interfaces

build/launch/src/com/intellij/tools/launch/rd/components/RemoteDevBackend.kt

@@ -13,11 +13,15 @@ import com.intellij.tools.launch.ide.environments.docker.dockerRunCliCommandLaun
 import com.intellij.tools.launch.ide.environments.local.LocalIdeCommandLauncherFactory
 import com.intellij.tools.launch.ide.environments.local.LocalProcessLaunchResult
 import com.intellij.tools.launch.ide.environments.local.localLaunchOptions
+import com.intellij.tools.launch.os.ProcessOutputFlows
+import com.intellij.tools.launch.os.ProcessOutputInfo
 import com.intellij.tools.launch.os.ProcessOutputStrategy
 import com.intellij.tools.launch.rd.BackendInDockerContainer
 import com.intellij.tools.launch.rd.BackendInEnvDescription
 import com.intellij.tools.launch.rd.BackendOnLocalMachine
+import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.flow.firstOrNull
 import kotlinx.coroutines.withContext
 import java.io.File
 
@@ -32,7 +36,7 @@ internal sealed interface BackendLaunchResult {
   data class Docker(val localDockerRunResult: LocalDockerRunResult, override val backendStatus: BackendStatus) : BackendLaunchResult
 }
 
-internal fun runCodeWithMeHostNoLobby(backendDescription: BackendInEnvDescription): BackendLaunchResult {
+internal fun runCodeWithMeHostNoLobby(backendDescription: BackendInEnvDescription, coroutineScope: CoroutineScope): BackendLaunchResult {
   val projectPath = backendDescription.backendDescription.projectPath
   val mainModule = backendDescription.backendDescription.product.mainModule
   val paths = IdeaPathsProvider()
@@ -51,7 +55,11 @@ internal fun runCodeWithMeHostNoLobby(backendDescription: BackendInEnvDescriptio
   return when (backendDescription) {
     is BackendOnLocalMachine -> {
       val localProcessLaunchResult = IdeLauncher.launchCommand(
-        LocalIdeCommandLauncherFactory(localLaunchOptions(processOutputStrategy = ProcessOutputStrategy.RedirectToFiles(paths.logFolder))),
+        LocalIdeCommandLauncherFactory(localLaunchOptions(
+          processOutputStrategy = ProcessOutputStrategy.RedirectToFiles(paths.logFolder),
+          processTitle = "IDE Backend on Docker",
+          lifespanScope = coroutineScope
+        )),
         context = IdeLaunchContext(
           classpathCollector = classpathCollector,
           localPaths = paths,
@@ -62,7 +70,7 @@ internal fun runCodeWithMeHostNoLobby(backendDescription: BackendInEnvDescriptio
           specifyUserHomeExplicitly = false,
         )
       )
-      BackendLaunchResult.Local(localProcessLaunchResult, BackendStatusFromStdout(localProcessLaunchResult.process))
+      BackendLaunchResult.Local(localProcessLaunchResult, localProcessLaunchResult.processWrapper.processOutputInfo.toBackendStatus())
     }
     is BackendInDockerContainer -> {
       val localDockerRunResult = IdeLauncher.launchCommand(
@@ -85,7 +93,7 @@ internal fun runCodeWithMeHostNoLobby(backendDescription: BackendInEnvDescriptio
           specifyUserHomeExplicitly = false,
         )
       )
-      BackendLaunchResult.Docker(localDockerRunResult, BackendStatusFromStdout(localDockerRunResult.process))
+      BackendLaunchResult.Docker(localDockerRunResult, localDockerRunResult.processWrapper.processOutputInfo.toBackendStatus())
     }
   }
 }
@@ -94,21 +102,23 @@ internal interface BackendStatus {
   suspend fun waitForHealthy(): Boolean
 }
 
-private class BackendStatusFromStdout(private val process: Process) : BackendStatus {
-  override suspend fun waitForHealthy(): Boolean {
-    return withContext(Dispatchers.IO) {
-      // do not call `use` or `useLines` to prevent the process from stopping after closing its `stdout`
-      process.inputStream
-        .bufferedReader()
-        .lineSequence()
-        .forEach { line ->
-          if (line.contains("Join link:")) {
-            return@withContext true
-          }
-        }
-      false
+private fun ProcessOutputInfo.toBackendStatus(): BackendStatus =
+  when (this) {
+    is ProcessOutputInfo.Piped -> BackendStatusFromStdout(outputFlows)
+    ProcessOutputInfo.InheritedByParent -> {
+      throw NotImplementedError("We should look for the status in the parent's process standard output (if we stick to parsing stdout)")
+    }
+    is ProcessOutputInfo.RedirectedToFiles -> {
+      throw NotImplementedError("We should look for the status in the log: $stdoutLogPath")
     }
   }
+
+private class BackendStatusFromStdout(private val processOutputFlows: ProcessOutputFlows) : BackendStatus {
+  override suspend fun waitForHealthy(): Boolean =
+    withContext(Dispatchers.IO) {
+      // do not call `use` or `useLines` to prevent the process from stopping after closing its `stdout`
+      processOutputFlows.stdout.firstOrNull { it.contains("Join link:") } != null
+    }
 }
 
 private fun cwmHostNoLobby(bindToHost: String, projectPath: PathInLaunchEnvironment) =