Implement client-side run

**Problem**
`run` task has been emulated via function call inside of a sandboxed classloader,
and blocking the command processing of sbt server loop.
This poses isolation and availability issues.

**Solution**
This implements client-side run where the server creates a sandbox environment, and sends the information to the client,
and the client forks a new JVM to perform the run.
The client-side behavior has been implemented in sbtn side already.

main-command/src/main/scala/sbt/State.scala

@@ -61,10 +61,13 @@ final case class State(
     }
 
   def source: Option[CommandSource] =
-    currentCommand match {
+    currentCommand match
       case Some(x) => x.source
       case _       => None
-    }
+  def isNetworkCommand: Boolean =
+    source match
+      case Some(s) => s.channelName.startsWith("network")
+      case _       => false
 }
 
 /**

main/src/main/scala/sbt/Defaults.scala

@@ -52,7 +52,6 @@ import sbt.internal.server.{
   BspCompileTask,
   BuildServerProtocol,
   BuildServerReporter,
-  ClientJob,
   Definition,
   LanguageServerProtocol,
   ServerHandler,
@@ -141,6 +140,7 @@ import xsbti.compile.{
   TransactionalManagerType
 }
 import sbt.internal.IncrementalTest
+import sbt.internal.RunUtil
 
 object Defaults extends BuildCommon {
   final val CacheDirectoryName = "cache"
@@ -241,7 +241,7 @@ object Defaults extends BuildCommon {
         getRootPaths(out, app) + ("CSR_CACHE" -> coursierCache)
       },
       fileConverter := MappedFileConverter(rootPaths.value, allowMachinePath.value)
-    ) ++ BuildServerProtocol.globalSettings ++ ClientJob.globalSettings
+    ) ++ BuildServerProtocol.globalSettings
 
   private[sbt] def getRootPaths(out: NioPath, app: AppConfiguration): ListMap[String, NioPath] =
     val base = app.baseDirectory.getCanonicalFile.toPath
@@ -391,6 +391,7 @@ object Defaults extends BuildCommon {
       aggregate :== true,
       maxErrors :== 100,
       fork :== false,
+      clientSide :== true,
       initialize :== {},
       templateResolverInfos :== Nil,
       templateDescriptions :== TemplateCommandUtil.defaultTemplateDescriptions,
@@ -1035,27 +1036,12 @@ object Defaults extends BuildCommon {
         })
         pickMainClassOrWarn(discoveredMainClasses.value, streams.value.log, logWarning)
       },
-      runMain := foregroundRunMainTask.evaluated,
-      run := foregroundRunTask.evaluated,
-      runBlock := {
-        val r = run.evaluated
-        val service = bgJobService.value
-        service.waitForTry(r.handle).get
-        ()
-      },
-      runMainBlock := {
-        val r = runMain.evaluated
-        val service = bgJobService.value
-        service.waitForTry(r.handle).get
-        ()
-      },
       fgRun := runTask(fullClasspath, (run / mainClass), (run / runner)).evaluated,
       fgRunMain := runMainTask(fullClasspath, (run / runner)).evaluated,
       copyResources := copyResourcesTask.value,
-      // note that we use the same runner and mainClass as plain run
+    ) ++ RunUtil.configTasks(This) ++ inTask(run)(
-      mainBgRunMainTaskForConfig(This),
+      runnerSettings ++ newRunnerSettings
-      mainBgRunTaskForConfig(This)
+    ) ++ compileIncrementalTaskSettings
-    ) ++ inTask(run)(runnerSettings ++ newRunnerSettings) ++ compileIncrementalTaskSettings
 
   private lazy val configGlobal = globalDefaults(
     Seq(
@@ -1869,122 +1855,6 @@ object Defaults extends BuildCommon {
   /** Implements `cleanFiles` task. */
   private[sbt] def cleanFilesTask: Initialize[Task[Vector[File]]] = Def.task { Vector.empty[File] }
 
-  private def termWrapper(canonical: Boolean, echo: Boolean): (() => Unit) => (() => Unit) =
-    (f: () => Unit) =>
-      () => {
-        val term = ITerminal.get
-        if (!canonical) {
-          term.enterRawMode()
-          if (echo) term.setEchoEnabled(echo)
-        } else if (!echo) term.setEchoEnabled(false)
-        try f()
-        finally {
-          if (!canonical) term.exitRawMode()
-          if (!echo) term.setEchoEnabled(true)
-        }
-      }
-  def bgRunMainTask(
-      products: Initialize[Task[Classpath]],
-      classpath: Initialize[Task[Classpath]],
-      copyClasspath: Initialize[Boolean],
-      scalaRun: Initialize[Task[ScalaRun]]
-  ): Initialize[InputTask[JobHandle]] = {
-    val parser = Defaults.loadForParser(discoveredMainClasses)((s, names) =>
-      Defaults.runMainParser(s, names getOrElse Nil)
-    )
-    Def.inputTask {
-      val service = bgJobService.value
-      val (mainClass, args) = parser.parsed
-      val hashClasspath = (bgRunMain / bgHashClasspath).value
-      val wrapper = termWrapper(canonicalInput.value, echoInput.value)
-      val converter = fileConverter.value
-      service.runInBackgroundWithLoader(resolvedScoped.value, state.value) { (logger, workingDir) =>
-        val cp =
-          if copyClasspath.value then
-            service.copyClasspath(
-              products.value,
-              classpath.value,
-              workingDir,
-              hashClasspath,
-              converter,
-            )
-          else classpath.value
-        given FileConverter = fileConverter.value
-        scalaRun.value match
-          case r: Run =>
-            val loader = r.newLoader(cp.files)
-            (
-              Some(loader),
-              wrapper(() => r.runWithLoader(loader, cp.files, mainClass, args, logger).get)
-            )
-          case sr =>
-            (None, wrapper(() => sr.run(mainClass, cp.files, args, logger).get))
-      }
-    }
-  }
-
-  def bgRunTask(
-      products: Initialize[Task[Classpath]],
-      classpath: Initialize[Task[Classpath]],
-      mainClassTask: Initialize[Task[Option[String]]],
-      copyClasspath: Initialize[Boolean],
-      scalaRun: Initialize[Task[ScalaRun]]
-  ): Initialize[InputTask[JobHandle]] =
-    val parser = Def.spaceDelimited()
-    Def.inputTask {
-      val args = parser.parsed
-      val service = bgJobService.value
-      val mainClass = mainClassTask.value getOrElse sys.error("No main class detected.")
-      val hashClasspath = (bgRun / bgHashClasspath).value
-      val wrapper = termWrapper(canonicalInput.value, echoInput.value)
-      val converter = fileConverter.value
-      service.runInBackgroundWithLoader(resolvedScoped.value, state.value) { (logger, workingDir) =>
-        val cp =
-          if copyClasspath.value then
-            service.copyClasspath(
-              products.value,
-              classpath.value,
-              workingDir,
-              hashClasspath,
-              converter
-            )
-          else classpath.value
-        given FileConverter = converter
-        scalaRun.value match
-          case r: Run =>
-            val loader = r.newLoader(cp.files)
-            (
-              Some(loader),
-              wrapper(() => r.runWithLoader(loader, cp.files, mainClass, args, logger).get)
-            )
-          case sr =>
-            (None, wrapper(() => sr.run(mainClass, cp.files, args, logger).get))
-      }
-    }
-
-  // `runMain` calls bgRunMain in the background and pauses the current channel
-  def foregroundRunMainTask: Initialize[InputTask[EmulateForeground]] =
-    Def.inputTask {
-      val handle = bgRunMain.evaluated
-      handle match
-        case threadJobHandle: AbstractBackgroundJobService#ThreadJobHandle =>
-          threadJobHandle.isAutoCancel = true
-        case _ => ()
-      EmulateForeground(handle)
-    }
-
-  // `run` task calls bgRun in the background and pauses the current channel
-  def foregroundRunTask: Initialize[InputTask[EmulateForeground]] =
-    Def.inputTask {
-      val handle = bgRun.evaluated
-      handle match {
-        case threadJobHandle: AbstractBackgroundJobService#ThreadJobHandle =>
-          threadJobHandle.isAutoCancel = true
-        case _ =>
-      }
-      EmulateForeground(handle)
-    }
-
   def runMainTask(
       classpath: Initialize[Task[Classpath]],
       scalaRun: Initialize[Task[ScalaRun]]
@@ -2005,15 +1875,7 @@ object Defaults extends BuildCommon {
       classpath: Initialize[Task[Classpath]],
       mainClassTask: Initialize[Task[Option[String]]],
       scalaRun: Initialize[Task[ScalaRun]]
-  ): Initialize[InputTask[Unit]] =
+  ): Initialize[InputTask[Unit]] = RunUtil.serverSideRunTask(classpath, mainClassTask, scalaRun)
-    val parser = Def.spaceDelimited()
-    Def.inputTask {
-      val in = parser.parsed
-      val mainClass = mainClassTask.value getOrElse sys.error("No main class detected.")
-      val cp = classpath.value
-      given FileConverter = fileConverter.value
-      scalaRun.value.run(mainClass, cp.files, in, streams.value.log).get
-    }
 
   def runnerTask: Setting[Task[ScalaRun]] = runner := runnerInit.value
 
@@ -2173,26 +2035,6 @@ object Defaults extends BuildCommon {
       )
     )
 
-  def mainBgRunTask = mainBgRunTaskForConfig(Select(Runtime))
-  def mainBgRunMainTask = mainBgRunMainTaskForConfig(Select(Runtime))
-
-  private def mainBgRunTaskForConfig(c: ScopeAxis[ConfigKey]) =
-    bgRun := bgRunTask(
-      exportedProductJars,
-      This / c / This / fullClasspathAsJars,
-      run / mainClass,
-      bgRun / bgCopyClasspath,
-      run / runner
-    ).evaluated
-
-  private def mainBgRunMainTaskForConfig(c: ScopeAxis[ConfigKey]) =
-    bgRunMain := bgRunMainTask(
-      exportedProductJars,
-      This / c / This / fullClasspathAsJars,
-      bgRunMain / bgCopyClasspath,
-      run / runner
-    ).evaluated
-
   def discoverMainClasses(analysis: CompileAnalysis): Seq[String] = analysis match {
     case analysis: Analysis =>
       analysis.infos.allInfos.values.map(_.getMainClasses).flatten.toSeq.sorted
@@ -2649,10 +2491,10 @@ object Defaults extends BuildCommon {
   lazy val configSettings: Seq[Setting[?]] =
     Classpaths.configSettings ++ configTasks ++ configPaths ++ packageConfig ++
       Classpaths.compilerPluginConfig ++ deprecationSettings ++
-      BuildServerProtocol.configSettings ++ ClientJob.configSettings
+      BuildServerProtocol.configSettings
 
   lazy val compileSettings: Seq[Setting[?]] =
-    configSettings ++ (mainBgRunMainTask +: mainBgRunTask) ++ Classpaths.addUnmanagedLibrary
+    configSettings ++ RunUtil.configTasks(Select(Runtime)) ++ Classpaths.addUnmanagedLibrary
 
   lazy val testSettings: Seq[Setting[?]] = configSettings ++ testTasks
 
@@ -4685,19 +4527,6 @@ trait BuildExtra extends BuildCommon with DefExtra {
       r.run(mainClass, cp.files, baseArguments ++ args, streams.value.log).get
     }
 
-  def runTask(
-      config: Configuration,
-      mainClass: String,
-      arguments: String*
-  ): Initialize[Task[Unit]] =
-    Def.task {
-      given FileConverter = fileConverter.value
-      val cp = (config / fullClasspath).value
-      val r = (config / run / runner).value
-      val s = streams.value
-      r.run(mainClass, cp.files, arguments, s.log).get
-    }
-
   // public API
   /** Returns a vector of settings that create custom run input task. */
   @nowarn

main/src/main/scala/sbt/Keys.scala

@@ -316,10 +316,8 @@ object Keys {
   // Run Keys
   val selectMainClass = taskKey[Option[String]]("Selects the main class to run.").withRank(BMinusTask)
   val mainClass = taskKey[Option[String]]("Defines the main class for packaging or running.").withRank(BPlusTask)
-  val run = inputKey[EmulateForeground]("Runs a main class, passing along arguments provided on the command line.").withRank(APlusTask)
+  val run = inputKey[Unit | ClientJobParams]("Runs a main class, passing along arguments provided on the command line.").withRank(APlusTask)
-  val runBlock = inputKey[Unit]("Runs a main class, and blocks until it's done.").withRank(DTask)
+  val runMain = inputKey[Unit | ClientJobParams]("Runs the main class selected by the first argument, passing the remaining arguments to the main method.").withRank(ATask)
-  val runMain = inputKey[EmulateForeground]("Runs the main class selected by the first argument, passing the remaining arguments to the main method.").withRank(ATask)
-  val runMainBlock = inputKey[Unit]("Runs the main class selected by the first argument, passing the remaining arguments to the main method.").withRank(DTask)
   val discoveredMainClasses = taskKey[Seq[String]]("Auto-detects main classes.").withRank(BMinusTask)
   val runner = taskKey[ScalaRun]("Implementation used to run a main class.").withRank(DTask)
   val trapExit = settingKey[Boolean]("If true, enables exit trapping and thread management for 'run'-like tasks. This was removed in sbt 1.6.0 due to JDK 17 deprecating Security Manager.").withRank(CSetting)
@@ -332,6 +330,7 @@ object Keys {
   val discoveredJavaHomes = settingKey[Map[String, File]]("Discovered Java home directories")
   val javaHomes = settingKey[Map[String, File]]("The user-defined additional Java home directories")
   val fullJavaHomes = settingKey[Map[String, File]]("Combines discoveredJavaHomes and custom javaHomes.").withRank(CTask)
+  val clientSide = settingKey[Boolean]("If true, takes the action on the client-side")
 
   val javaOptions = taskKey[Seq[String]]("Options passed to a new JVM when forking.").withRank(BPlusTask)
   val envVars = taskKey[Map[String, String]]("Environment variables used when forking a new JVM").withRank(BTask)
@@ -484,8 +483,6 @@ object Keys {
 
   @cacheLevel(include = Array.empty)
   val bspReporter = taskKey[BuildServerReporter]("").withRank(DTask)
-  val clientJob = inputKey[ClientJobParams]("Translates a task into a job specification").withRank(Invisible)
-  val clientJobRunInfo = inputKey[ClientJobParams]("Translates the run task into a job specification").withRank(Invisible)
 
   val csrCacheDirectory = settingKey[File]("Coursier cache directory. Uses -Dsbt.coursier.home or Coursier's default.").withRank(CSetting)
   val csrMavenProfiles = settingKey[Set[String]]("").withRank(CSetting)

main/src/main/scala/sbt/internal/RunUtil.scala

@@ -0,0 +1,302 @@
+package sbt
+package internal
+
+import java.io.File
+import sbt.BuildExtra.*
+import sbt.Def.*
+import sbt.Keys.*
+import sbt.ScopeAxis.This
+import sbt.SlashSyntax0.*
+import sbt.internal.util.{ Terminal as ITerminal }
+import sbt.internal.worker.{ ClientJobParams, FilePath, JvmRunInfo, RunInfo }
+import sbt.io.IO
+import sbt.protocol.Serialization
+import sbt.util.CacheImplicits.given
+import xsbti.FileConverter
+
+object RunUtil:
+  /**
+   * Conventional server-side run implementation.
+   */
+  def serverSideRunTask(
+      classpath: Initialize[Task[Classpath]],
+      mainClassTask: Initialize[Task[Option[String]]],
+      scalaRun: Initialize[Task[ScalaRun]]
+  ): Initialize[InputTask[Unit]] =
+    val parser = Def.spaceDelimited()
+    Def.inputTask {
+      val in = parser.parsed
+      val mainClass = getMainClass(mainClassTask.value)
+      val cp = classpath.value
+      given FileConverter = fileConverter.value
+      scalaRun.value.run(mainClass, cp.files, in, streams.value.log).get
+    }
+
+  def configTasks(c: ScopeAxis[ConfigKey]): Seq[Setting[?]] = Seq(
+    bgRunMain := bgRunMainTask(
+      exportedProductJars,
+      This / c / This / fullClasspathAsJars,
+      bgRunMain / bgCopyClasspath,
+      run / runner
+    ).evaluated,
+    // note that we use the same runner and mainClass as plain run
+    bgRun := bgRunTask(
+      exportedProductJars,
+      This / c / This / fullClasspathAsJars,
+      run / mainClass,
+      bgRun / bgCopyClasspath,
+      run / runner
+    ).evaluated,
+    runMain := defaultRunMainTask(
+      exportedProductJars,
+      This / c / This / fullClasspathAsJars,
+      run / runner,
+      runMain / clientSide
+    ).evaluated,
+    run := defaultRunTask(
+      exportedProductJars,
+      This / c / This / fullClasspathAsJars,
+      run / mainClass,
+      run / runner,
+      run / clientSide
+    ).evaluated,
+    run / connectInput := true,
+  )
+
+  private def termWrapper(canonical: Boolean, echo: Boolean): (() => Unit) => (() => Unit) =
+    (f: () => Unit) =>
+      () => {
+        val term = ITerminal.get
+        if (!canonical) {
+          term.enterRawMode()
+          if (echo) term.setEchoEnabled(echo)
+        } else if (!echo) term.setEchoEnabled(false)
+        try f()
+        finally {
+          if (!canonical) term.exitRawMode()
+          if (!echo) term.setEchoEnabled(true)
+        }
+      }
+
+  private def getMainClass(value: Option[String]): String =
+    value.getOrElse(sys.error("no main class detected"))
+
+  private def mkRunInfo(
+      args: Vector[String],
+      mainClass: String,
+      cp: Classpath,
+      fo: ForkOptions,
+      conv: FileConverter
+  ): RunInfo =
+    val strategy = fo.outputStrategy.map(_.getClass().getSimpleName().filter(_ != '$'))
+    // sbtn doesn't set java.home, so we need to do the fallback here
+    val javaHome =
+      fo.javaHome.map(IO.toURI).orElse(sys.props.get("java.home").map(x => IO.toURI(new File(x))))
+    val jvmRunInfo = JvmRunInfo(
+      args = args,
+      classpath = cp.map(x => IO.toURI(conv.toPath(x.data).toFile)).map(FilePath(_, "")).toVector,
+      mainClass = mainClass,
+      connectInput = fo.connectInput,
+      javaHome = javaHome,
+      outputStrategy = strategy,
+      workingDirectory = fo.workingDirectory.map(IO.toURI),
+      jvmOptions = fo.runJVMOptions,
+      environmentVariables = fo.envVars.toMap,
+    )
+    RunInfo(
+      jvm = true,
+      jvmRunInfo = jvmRunInfo,
+    )
+
+  def defaultRunMainTask(
+      products: Initialize[Task[Classpath]],
+      classpath: Initialize[Task[Classpath]],
+      scalaRun: Initialize[Task[ScalaRun]],
+      clientRun: Initialize[Boolean],
+  ): Initialize[InputTask[Unit | ClientJobParams]] =
+    val parser = Defaults.loadForParser(discoveredMainClasses)((s, names) =>
+      Defaults.runMainParser(s, names getOrElse Nil)
+    )
+    Def.inputTask {
+      val conv = fileConverter.value
+      given FileConverter = conv
+      val service = bgJobService.value
+      val (mainClass, args) = parser.parsed
+      val hashClasspath = (bgRunMain / bgHashClasspath).value
+      val fo = (run / forkOptions).value
+      val state = Keys.state.value
+      if clientRun.value && state.isNetworkCommand then
+        val workingDir = service.createWorkingDirectory
+        val cp = service.copyClasspath(
+          products.value,
+          classpath.value,
+          workingDir,
+          conv,
+        )
+        val info = mkRunInfo(args.toVector, mainClass, cp, fo, conv)
+        val result = ClientJobParams(
+          runInfo = info
+        )
+        import sbt.internal.worker.codec.JsonProtocol.*
+        state.notifyEvent(Serialization.clientJob, result)
+        result
+      else
+        val wrapper = termWrapper(canonicalInput.value, echoInput.value)
+        val handle = service.runInBackgroundWithLoader(Keys.resolvedScoped.value, state):
+          (logger, workingDir) =>
+            val cp = service.copyClasspath(
+              products.value,
+              classpath.value,
+              workingDir,
+              hashClasspath,
+              conv,
+            )
+            scalaRun.value match
+              case r: Run =>
+                val loader = r.newLoader(cp.files)
+                (
+                  Some(loader),
+                  wrapper(() => r.runWithLoader(loader, cp.files, mainClass, args, logger).get)
+                )
+              case sr =>
+                (None, wrapper(() => sr.run(mainClass, cp.files, args, logger).get))
+        service.waitForTry(handle).get
+        ()
+    }
+
+  def defaultRunTask(
+      products: Initialize[Task[Classpath]],
+      classpath: Initialize[Task[Classpath]],
+      mainClassTask: Initialize[Task[Option[String]]],
+      scalaRun: Initialize[Task[ScalaRun]],
+      clientRun: Initialize[Boolean]
+  ): Initialize[InputTask[Unit | ClientJobParams]] =
+    val parser = Def.spaceDelimited()
+    Def.inputTask {
+      val conv = fileConverter.value
+      given FileConverter = conv
+      val args = parser.parsed
+      val service = bgJobService.value
+      val mainClass = getMainClass(mainClassTask.value)
+      val hashClasspath = (bgRun / bgHashClasspath).value
+      val fo = (run / forkOptions).value
+      val state = Keys.state.value
+      if clientRun.value && state.isNetworkCommand then
+        val workingDir = service.createWorkingDirectory
+        val cp = service.copyClasspath(
+          products.value,
+          classpath.value,
+          workingDir,
+          conv,
+        )
+        val info = mkRunInfo(args.toVector, mainClass, cp, fo, conv)
+        val result = ClientJobParams(
+          runInfo = info
+        )
+        import sbt.internal.worker.codec.JsonProtocol.*
+        state.notifyEvent(Serialization.clientJob, result)
+        result
+      else
+        val wrapper = termWrapper(canonicalInput.value, echoInput.value)
+        val handle = service.runInBackgroundWithLoader(Keys.resolvedScoped.value, state):
+          (logger, workingDir) =>
+            val cp = service.copyClasspath(
+              products.value,
+              classpath.value,
+              workingDir,
+              hashClasspath,
+              conv
+            )
+            scalaRun.value match
+              case r: Run =>
+                val loader = r.newLoader(cp.files)
+                (
+                  Some(loader),
+                  wrapper(() => r.runWithLoader(loader, cp.files, mainClass, args, logger).get)
+                )
+              case sr =>
+                (None, wrapper(() => sr.run(mainClass, cp.files, args, logger).get))
+        service.waitForTry(handle).get
+        ()
+    }
+
+  def bgRunMainTask(
+      products: Initialize[Task[Classpath]],
+      classpath: Initialize[Task[Classpath]],
+      copyClasspath: Initialize[Boolean],
+      scalaRun: Initialize[Task[ScalaRun]]
+  ): Initialize[InputTask[JobHandle]] =
+    val parser = Defaults.loadForParser(discoveredMainClasses)((s, names) =>
+      Defaults.runMainParser(s, names getOrElse Nil)
+    )
+    Def.inputTask {
+      val service = bgJobService.value
+      val (mainClass, args) = parser.parsed
+      val hashClasspath = (bgRunMain / bgHashClasspath).value
+      val wrapper = termWrapper(canonicalInput.value, echoInput.value)
+      val converter = fileConverter.value
+      service.runInBackgroundWithLoader(Keys.resolvedScoped.value, state.value) {
+        (logger, workingDir) =>
+          val cp =
+            if copyClasspath.value then
+              service.copyClasspath(
+                products.value,
+                classpath.value,
+                workingDir,
+                hashClasspath,
+                converter,
+              )
+            else classpath.value
+          given FileConverter = fileConverter.value
+          scalaRun.value match
+            case r: Run =>
+              val loader = r.newLoader(cp.files)
+              (
+                Some(loader),
+                wrapper(() => r.runWithLoader(loader, cp.files, mainClass, args, logger).get)
+              )
+            case sr =>
+              (None, wrapper(() => sr.run(mainClass, cp.files, args, logger).get))
+      }
+    }
+
+  def bgRunTask(
+      products: Initialize[Task[Classpath]],
+      classpath: Initialize[Task[Classpath]],
+      mainClassTask: Initialize[Task[Option[String]]],
+      copyClasspath: Initialize[Boolean],
+      scalaRun: Initialize[Task[ScalaRun]]
+  ): Initialize[InputTask[JobHandle]] =
+    val parser = Def.spaceDelimited()
+    Def.inputTask {
+      val args = parser.parsed
+      val service = bgJobService.value
+      val mainClass = getMainClass(mainClassTask.value)
+      val hashClasspath = (bgRun / bgHashClasspath).value
+      val wrapper = termWrapper(canonicalInput.value, echoInput.value)
+      val converter = fileConverter.value
+      service.runInBackgroundWithLoader(Keys.resolvedScoped.value, state.value) {
+        (logger, workingDir) =>
+          val cp =
+            if copyClasspath.value then
+              service.copyClasspath(
+                products.value,
+                classpath.value,
+                workingDir,
+                hashClasspath,
+                converter
+              )
+            else classpath.value
+          given FileConverter = converter
+          scalaRun.value match
+            case r: Run =>
+              val loader = r.newLoader(cp.files)
+              (
+                Some(loader),
+                wrapper(() => r.runWithLoader(loader, cp.files, mainClass, args, logger).get)
+              )
+            case sr =>
+              (None, wrapper(() => sr.run(mainClass, cp.files, args, logger).get))
+      }
+    }
+end RunUtil

main/src/main/scala/sbt/internal/server/ClientJob.scala

@@ -1,95 +0,0 @@
-/*
- * sbt
- * Copyright 2023, Scala center
- * Copyright 2011 - 2022, Lightbend, Inc.
- * Copyright 2008 - 2010, Mark Harrah
- * Licensed under Apache License 2.0 (see LICENSE)
- */
-
-package sbt
-package internal
-package server
-
-import java.io.File
-import sbt.Def.*
-import sbt.Keys.*
-import sbt.UpperStateOps.*
-import sbt.internal.util.complete.Parser
-import sbt.internal.worker.{ ClientJobParams, FilePath, JvmRunInfo, RunInfo }
-import sbt.io.IO
-import sbt.protocol.Serialization
-import sbt.Keys.fileConverter
-
-/**
- * A ClientJob represents a unit of work that sbt server process
- * can outsourse back to the client. Initially intended for sbtn client-side run.
- */
-object ClientJob {
-  lazy val globalSettings: Seq[Def.Setting[?]] = Seq(
-    clientJob := clientJobTask.evaluated,
-    clientJob / aggregate := false,
-  )
-
-  private def clientJobTask: Def.Initialize[InputTask[ClientJobParams]] = Def.inputTaskDyn {
-    val tokens = spaceDelimited().parsed
-    val state = Keys.state.value
-    val p = Act.aggregatedKeyParser(state)
-    if (tokens.isEmpty) {
-      sys.error("expected an argument, for example foo/run")
-    }
-    val scopedKey = Parser.parse(tokens.head, p) match {
-      case Right(x :: Nil) => x
-      case Right(xs)       => sys.error("too many keys")
-      case Left(err)       => sys.error(err)
-    }
-    if (scopedKey.key == run.key)
-      clientJobRunInfo.rescope(scopedKey.scope).toTask(" " + tokens.tail.mkString(" "))
-    else sys.error(s"unsupported task for clientJob $scopedKey")
-  }
-
-  // This will be scoped to Compile, Test, etc
-  lazy val configSettings: Seq[Def.Setting[?]] = Seq(
-    clientJobRunInfo := clientJobRunInfoTask.evaluated,
-  )
-
-  private def clientJobRunInfoTask: Def.Initialize[InputTask[ClientJobParams]] = Def.inputTask {
-    val state = Keys.state.value
-    val args = spaceDelimited().parsed
-    val mainClass = (Keys.run / Keys.mainClass).value
-    val service = bgJobService.value
-    val fo = (Keys.run / Keys.forkOptions).value
-    val workingDir = service.createWorkingDirectory
-    val conv = fileConverter.value
-    val cp = service.copyClasspath(
-      exportedProductJars.value,
-      fullClasspathAsJars.value,
-      workingDir,
-      conv,
-    )
-    val strategy = fo.outputStrategy.map(_.getClass().getSimpleName().filter(_ != '$'))
-    // sbtn doesn't set java.home, so we need to do the fallback here
-    val javaHome =
-      fo.javaHome.map(IO.toURI).orElse(sys.props.get("java.home").map(x => IO.toURI(new File(x))))
-    val jvmRunInfo = JvmRunInfo(
-      args = args.toVector,
-      classpath = cp.map(x => IO.toURI(conv.toPath(x.data).toFile)).map(FilePath(_, "")).toVector,
-      mainClass = mainClass.getOrElse(sys.error("no main class")),
-      connectInput = fo.connectInput,
-      javaHome = javaHome,
-      outputStrategy = strategy,
-      workingDirectory = fo.workingDirectory.map(IO.toURI),
-      jvmOptions = fo.runJVMOptions,
-      environmentVariables = fo.envVars.toMap,
-    )
-    val info = RunInfo(
-      jvm = true,
-      jvmRunInfo = jvmRunInfo,
-    )
-    val result = ClientJobParams(
-      runInfo = info
-    )
-    import sbt.internal.worker.codec.JsonProtocol.*
-    state.notifyEvent(Serialization.clientJob, result)
-    result
-  }
-}

sbt-app/src/sbt-test/actions/add-alias/build.sbt

@@ -1,3 +1,3 @@
-addCommandAlias("demo-success", "runBlock true")
+addCommandAlias("demo-success", "run true")
-addCommandAlias("demo-failure", "runBlock false")
+addCommandAlias("demo-failure", "run false")
 addCommandAlias("z", "scalaVersion")

sbt-app/src/sbt-test/actions/input-task-dyn/build.sbt

@@ -12,7 +12,7 @@ lazy val root = (project in file(".")).
     name := "run-test",
     runFoo := Def.inputTaskDyn {
       val args = Def.spaceDelimited().parsed
-      (Compile / runMainBlock).toTask(s" Foo " + args.mkString(" "))
+      (Compile / runMain).toTask(s" Foo " + args.mkString(" "))
     }.evaluated,
     check := {
       val x = runFoo.toTask(" hi ho").value

sbt-app/src/sbt-test/classloader-cache/jni/build.sbt

@@ -35,6 +35,6 @@ val root = (project in file(".")).settings(
     val cp = System.getProperty("java.library.path", "").split(":").dropRight(1)
     System.setProperty("java.library.path", cp.mkString(":"))
   },
-  wrappedRun := wrap(Runtime / runBlock).value,
+  wrappedRun := wrap(Runtime / run).value,
   wrappedTest := wrap(Test / testOnly).value
 )

sbt-app/src/sbt-test/dependency-management/aar-packaging/test

@@ -1,3 +1,3 @@
 $ delete output
-> runBlock
+> run
 $ exists output

sbt-app/src/sbt-test/dependency-management/exclude-dependencies2/test

@@ -1,3 +1,3 @@
 $ delete output
-> runBlock
+> run
 $ exists output

sbt-app/src/sbt-test/dependency-management/profiles/test

@@ -1,3 +1,3 @@
 $ delete output
-> runBlock
+> run
 $ exists output

sbt-app/src/sbt-test/dependency-management/stdlib-unfreeze-warn/test

@@ -6,6 +6,6 @@
 # macro expansion fails
 -> b/compile
 
-> c/runBlock
+> c/run
 $ exists s2.13.13.txt
 $ delete s2.13.13.txt

sbt-app/src/sbt-test/dependency-management/stdlib-unfreeze/test

@@ -1,12 +1,12 @@
 > a/checkLibs
 > b/checkLibs
 
-> b/runBlock
+> b/run
 $ exists s2.13.12.txt
 $ delete s2.13.12.txt
 
 # don't crash when expanding the macro
-> b3/runBlock
+> b3/run
 $ exists s2.13.14.txt
 $ delete s2.13.14.txt
 

sbt-app/src/sbt-test/dependency-management/url-no-head/test

@@ -1,3 +1,3 @@
 $ delete output
-> runBlock
+> run
 $ exists output

sbt-app/src/sbt-test/lm-coursier/classifiers/test

@@ -1 +1 @@
-> runBlock
+> run

sbt-app/src/sbt-test/lm-coursier/from-no-head/test

@@ -1,3 +1,3 @@
 $ delete output
-> runBlock
+> run
 $ exists output

sbt-app/src/sbt-test/lm-coursier/from-wrong-url/test

@@ -1,3 +1,3 @@
 $ delete output
-> runBlock
+> run
 $ exists output

sbt-app/src/sbt-test/lm-coursier/from/test

@@ -1,4 +1,4 @@
 $ delete output
-> runBlock
+> run
 $ exists output
 $ delete shapeless_2.11-2.3.0.jar

sbt-app/src/sbt-test/lm-coursier/maven-compatible/test

@@ -1,3 +1,3 @@
 $ delete output
-> runBlock
+> run
 $ exists output

sbt-app/src/sbt-test/lm-coursier/per-config-resolution/test

@@ -1,2 +1,2 @@
-> runBlock
+> run
 > Test/run

sbt-app/src/sbt-test/lm-coursier/scala-jars/test

@@ -1,3 +1,3 @@
 $ delete output
-> runBlock
+> run
 $ exists output

sbt-app/src/sbt-test/lm-coursier/scala-sources-javadoc-jars/test

@@ -1,4 +1,4 @@
 $ delete output
-> runBlock
+> run
 $ exists output
 > updateClassifiersCheck

sbt-app/src/sbt-test/lm-coursier/simple/test

@@ -1,4 +1,4 @@
 $ delete output
-> runBlock
+> run
 $ exists output
 > updateSbtClassifiers

sbt-app/src/sbt-test/lm-coursier/zookeeper/test

@@ -1,3 +1,3 @@
 $ delete output
-> runBlock
+> run
 $ exists output

sbt-app/src/sbt-test/package/resources/test

@@ -4,7 +4,7 @@
 
 # This should fail because the Main object is in package jartest and the resource is directly
 # in src/main/resources
--> runBlock
+-> run
 
 > package
 
@@ -18,7 +18,7 @@ $ copy-file src/main/resources/main_resource_test src/main/resources/jartest/mai
 $ delete src/main/resources/main_resource_test
 
 # This should succeed because sbt should put the resource on the runClasspath
-> runBlock
+> run
 
 # This is necessary because package bases whether or not to run on last modified times, which don't have
 # high enough resolution to notice the above move of main_resource_test

sbt-app/src/sbt-test/source-dependencies/constants/test

@@ -3,14 +3,14 @@
 $ copy-file changes/B.scala B.scala
 
 $ copy-file changes/A1.scala A.scala
-> runBlock 1
+> run 1
 $ copy-file changes/A2.scala A.scala
-> runBlock 2
+> run 2
 
 > clean
 > ++2.13.12!
 
 $ copy-file changes/A1.scala A.scala
-> runBlock 1
+> run 1
 $ copy-file changes/A2.scala A.scala
-> runBlock 2
+> run 2

sbt-app/src/sbt-test/source-dependencies/implicit-search/test

@@ -2,10 +2,10 @@ $ copy-file changes/A1.scala A.scala
 $ copy-file changes/B.scala B.scala
 $ copy-file changes/C.scala C.scala
 > compile
--> runBlock
+-> run
 
 $ copy-file changes/A2.scala A.scala
 $ sleep 1000
 
 > compile
-> runBlock
+> run

sbt-app/src/sbt-test/source-dependencies/java-basic/test

@@ -36,10 +36,10 @@ $ delete src/main/java/a/A.java
 # It shouldn't run though, because it doesn't have a main method
 $ copy-file changes/B1.java src/main/java/a/b/B.java
 > compile
--> runBlock
+-> run
 
 
 # Replace B with a new B that has a main method and should therefore run
 # if the main method was properly detected
 $ copy-file changes/B3.java src/main/java/a/b/B.java
-> runBlock
+> run

sbt-app/src/sbt-test/source-dependencies/linearization/test

@@ -1,7 +1,7 @@
 > compile
 
 # the value of F.x should be 16
-> runBlock 16
+> run 16
 
 # modify D.scala so that the linearization changes
 $ copy-file changes/D.scala D.scala
@@ -12,4 +12,4 @@ $ sleep 1000
 
 # if F is recompiled, the value of x should be 11, otherwise it will still be 16
 # and this will fail
-> runBlock 11
+> run 11

sbt-app/src/sbt-test/source-dependencies/named/test

@@ -4,7 +4,7 @@
 > compile
 
 # result should be 1
-> runBlock 1
+> run 1
 
 # change order of arguments in A.x
 $ copy-file changes/A.scala A.scala
@@ -13,4 +13,4 @@ $ copy-file changes/A.scala A.scala
 > compile
 
 # Should still get 1 and not -1
-> runBlock 1
+> run 1

sbt-app/src/sbt-test/source-dependencies/specialized/test

@@ -2,7 +2,7 @@
 > compile
 
 # verify that erased A.x can be called normally and reflectively
-> runBlock false
+> run false
 
 # make A.x specialized
 $ copy-file changes/A.scala A.scala
@@ -12,4 +12,4 @@ $ copy-file changes/A.scala A.scala
 
 # verify that specialized A.x can be called normally and reflectively
 # NOTE: this test doesn't actually work correctly: have to check the output to see that B.scala was recompiled
-> runBlock true
+> run true

server-test/src/test/scala/testpkg/ClientTest.scala

@@ -122,6 +122,14 @@ class ClientTest extends AbstractServerTest {
   test("three commands with middle failure") {
     assert(client("compile;willFail;willSucceed") == 1)
   }
+  test("run") {
+    val (exitCode, lines) = clientWithStdoutLines("run")
+    assert(exitCode == 0)
+    assert(
+      lines.toList.exists(_.contains("running (fork) hello")),
+      lines.toList.mkString(",")
+    )
+  }
   test("compi completions") {
     val expected = Vector(
       "compile",