diff --git a/build.sc b/build.sc
index fe375522bd0..0c1f9e4a070 100644
--- a/build.sc
+++ b/build.sc
@@ -165,6 +165,7 @@ object Deps {
   def scalaCompiler(scalaVersion: String) = ivy"org.scala-lang:scala-compiler:${scalaVersion}"
   // last scalafmt release supporting Java 8 is 3.7.15
   val scalafmtDynamic = ivy"org.scalameta::scalafmt-dynamic:3.7.15" // scala-steward:off
+  def scalap(scalaVersion: String) = ivy"org.scala-lang:scalap:${scalaVersion}"
   def scalaReflect(scalaVersion: String) = ivy"org.scala-lang:scala-reflect:${scalaVersion}"
   val scalacScoveragePlugin = ivy"org.scoverage:::scalac-scoverage-plugin:1.4.11"
   val scoverage2Version = "2.1.0"
@@ -448,7 +449,7 @@ trait MillStableScalaModule extends MillPublishScalaModule with Mima {
     ProblemFilter.exclude[ReversedMissingMethodProblem](
       "mill.scalanativelib.ScalaNativeModule.mill$scalanativelib$ScalaNativeModule$$super$zincAuxiliaryClassFileExtensions"
     ),
-    // (7x) See https://github.com/com-lihaoyi/mill/pull/3064
+    // (6x) See https://github.com/com-lihaoyi/mill/pull/3064
     // Moved targets up in trait hierarchy, but also call them via super, which I think is safe
     ProblemFilter.exclude[ReversedMissingMethodProblem](
       "mill.scalalib.JavaModule.mill$scalalib$JavaModule$$super$zincWorker"
@@ -459,9 +460,6 @@ trait MillStableScalaModule extends MillPublishScalaModule with Mima {
     ProblemFilter.exclude[ReversedMissingMethodProblem](
       "mill.scalalib.JavaModule.mill$scalalib$JavaModule$$super$runUseArgsFile"
     ),
-    ProblemFilter.exclude[ReversedMissingMethodProblem](
-      "mill.scalalib.JavaModule#JavaModuleTests.mill$scalalib$JavaModule$JavaModuleTests$$super$testClasspath"
-    ),
     ProblemFilter.exclude[ReversedMissingMethodProblem](
       "mill.scalalib.JavaModule.mill$scalalib$JavaModule$$super$forkArgs"
     ),
@@ -470,6 +468,32 @@ trait MillStableScalaModule extends MillPublishScalaModule with Mima {
     ),
     ProblemFilter.exclude[ReversedMissingMethodProblem](
       "mill.scalalib.JavaModule.mill$scalalib$JavaModule$$super$forkWorkingDir"
+    ),
+    // (8x)
+    // Moved targets up in trait hierarchy, but also call them via super, which I think is safe
+    ProblemFilter.exclude[ReversedMissingMethodProblem](
+      "mill.scalalib.JavaModule.mill$scalalib$JavaModule$$super$localRunClasspath"
+    ),
+    ProblemFilter.exclude[ReversedMissingMethodProblem](
+      "mill.scalalib.JavaModule.mill$scalalib$JavaModule$$super$runLocal"
+    ),
+    ProblemFilter.exclude[ReversedMissingMethodProblem](
+      "mill.scalalib.JavaModule.mill$scalalib$JavaModule$$super$run"
+    ),
+    ProblemFilter.exclude[ReversedMissingMethodProblem](
+      "mill.scalalib.JavaModule.mill$scalalib$JavaModule$$super$doRunBackground"
+    ),
+    ProblemFilter.exclude[ReversedMissingMethodProblem](
+      "mill.scalalib.JavaModule.mill$scalalib$JavaModule$$super$runBackgroundLogToConsole"
+    ),
+    ProblemFilter.exclude[ReversedMissingMethodProblem](
+      "mill.scalalib.JavaModule.mill$scalalib$JavaModule$$super$runMainBackground"
+    ),
+    ProblemFilter.exclude[ReversedMissingMethodProblem](
+      "mill.scalalib.JavaModule.mill$scalalib$JavaModule$$super$runMainLocal"
+    ),
+    ProblemFilter.exclude[ReversedMissingMethodProblem](
+      "mill.scalalib.JavaModule.mill$scalalib$JavaModule$$super$runMain"
     )
   )
   def mimaPreviousVersions: T[Seq[String]] = Settings.mimaBaseVersions
@@ -740,7 +764,7 @@ object scalalib extends MillStableScalaModule {
 
   object worker extends MillPublishScalaModule with BuildInfo {
     def moduleDeps = Seq(scalalib.api)
-    def ivyDeps = Agg(Deps.zinc, Deps.log4j2Core)
+    def ivyDeps = Agg(Deps.zinc, Deps.log4j2Core, Deps.scalap(scalaVersion()))
     def buildInfoPackageName = "mill.scalalib.worker"
     def buildInfoObjectName = "Versions"
     def buildInfoMembers = Seq(
diff --git a/example/basic/4-builtin-commands/build.sc b/example/basic/4-builtin-commands/build.sc
index 8902380e8ab..5d9b005830b 100644
--- a/example/basic/4-builtin-commands/build.sc
+++ b/example/basic/4-builtin-commands/build.sc
@@ -82,7 +82,7 @@ foo.artifactName
 /** Usage
 
 > mill inspect foo.run
-foo.run(JavaModule.scala:...)
+foo.run(RunModule.scala:...)
     Runs this module's code in a subprocess and waits for it to finish
 Inputs:
     foo.finalMainClass
diff --git a/integration/feature/docannotations/test/src/DocAnnotationsTests.scala b/integration/feature/docannotations/test/src/DocAnnotationsTests.scala
index f86cdb04f2d..07428e428e7 100644
--- a/integration/feature/docannotations/test/src/DocAnnotationsTests.scala
+++ b/integration/feature/docannotations/test/src/DocAnnotationsTests.scala
@@ -67,7 +67,7 @@ object DocAnnotationsTests extends IntegrationTestSuite {
 
       assert(
         globMatches(
-          """core.run(JavaModule.scala:...)
+          """core.run(RunModule.scala:...)
             |    Runs this module's code in a subprocess and waits for it to finish
             |
             |    args <str>...
diff --git a/scalalib/api/src/mill/scalalib/api/ZincWorkerApi.scala b/scalalib/api/src/mill/scalalib/api/ZincWorkerApi.scala
index 04dfc1d0e98..37050202d44 100644
--- a/scalalib/api/src/mill/scalalib/api/ZincWorkerApi.scala
+++ b/scalalib/api/src/mill/scalalib/api/ZincWorkerApi.scala
@@ -144,6 +144,9 @@ trait ZincWorkerApi {
       auxiliaryClassFileExtensions = Seq.empty[String]
     )
 
+  /**
+   * Find main classes by inspecting the Zinc compilation analysis file.
+   */
   def discoverMainClasses(compilationResult: CompilationResult): Seq[String]
 
   def docJar(
@@ -153,4 +156,12 @@ trait ZincWorkerApi {
       scalacPluginClasspath: Agg[PathRef],
       args: Seq[String]
   )(implicit ctx: ZincWorkerApi.Ctx): Boolean
+
+  /**
+   * Discover main classes by inspecting the classpath.
+   */
+  def discoverMainClasses(classpath: Seq[os.Path]): Seq[String] = {
+    // We need this default-impl to keep binary compatinility (0.11.x)
+    Seq.empty
+  }
 }
diff --git a/scalalib/src/mill/scalalib/JavaModule.scala b/scalalib/src/mill/scalalib/JavaModule.scala
index cd5e64f64d2..3a7bad49b5c 100644
--- a/scalalib/src/mill/scalalib/JavaModule.scala
+++ b/scalalib/src/mill/scalalib/JavaModule.scala
@@ -53,15 +53,6 @@ trait JavaModule
       }
     }
 
-    /**
-     * The classpath containing the tests. This defaults to the compilation output.
-     */
-    def testClasspath: T[Seq[PathRef]] = T {
-      // bin-compat-shim: keep the super.call in the classfile
-      super.testClasspath
-      Seq(compile().classes)
-    }
-
     /**
      * JavaModule and its derivates define inner test modules.
      * To avoid unexpected misbehavior due to the use of the wrong inner test trait
@@ -458,14 +449,38 @@ trait JavaModule
       }
     }
 
+  /**
+   * The part of the [[localClasspath]] which is available "after compilation".
+   *
+   * Keep in sync with [[bspLocalRunClasspath]]
+   */
+  override def localRunClasspath: T[Seq[PathRef]] = T {
+    super.localRunClasspath() ++ resources() ++
+      Agg(compile().classes)
+  }
+
+  /**
+   * Same as [[localRunClasspath]] but for use in BSP server.
+   *
+   * Keep in sync with [[localRunClasspath]]
+   */
+  def bspLocalRunClasspath: T[Agg[UnresolvedPath]] = T {
+    Agg.from(super.localRunClasspath() ++ resources())
+      .map(p => UnresolvedPath.ResolvedPath(p.path)) ++
+      Agg(bspCompileClassesPath())
+  }
+
   /**
    * The *output* classfiles/resources from this module, used for execution,
    * excluding upstream modules and third-party dependencies, but including unmanaged dependencies.
    *
+   * This is build from [[localCompileClasspath]] and [[localRunClasspath]]
+   * as the parts available "before compilation" and "after compiliation".
+   *
    * Keep in sync with [[bspLocalClasspath]]
    */
   def localClasspath: T[Seq[PathRef]] = T {
-    localCompileClasspath().toSeq ++ resources() ++ Agg(compile().classes)
+    localCompileClasspath().toSeq ++ localRunClasspath()
   }
 
   /**
@@ -476,8 +491,8 @@ trait JavaModule
    */
   @internal
   def bspLocalClasspath: T[Agg[UnresolvedPath]] = T {
-    (localCompileClasspath() ++ resources()).map(p => UnresolvedPath.ResolvedPath(p.path)) ++
-      Agg(bspCompileClassesPath())
+    (localCompileClasspath()).map(p => UnresolvedPath.ResolvedPath(p.path)) ++
+      bspLocalRunClasspath()
   }
 
   /**
@@ -533,7 +548,7 @@ trait JavaModule
    * All classfiles and resources from upstream modules and dependencies
    * necessary to run this module's code after compilation
    */
-  def runClasspath: T[Seq[PathRef]] = T {
+  override def runClasspath: T[Seq[PathRef]] = T {
     super.runClasspath() ++
       resolvedRunIvyDeps().toSeq ++
       transitiveLocalClasspath() ++
@@ -688,7 +703,7 @@ trait JavaModule
    * Any command-line parameters you want to pass to the forked JVM under `run`,
    * `test` or `repl`
    */
-  def forkArgs: T[Seq[String]] = T {
+  override def forkArgs: T[Seq[String]] = T {
     // overridden here for binary compatibility (0.11.x)
     super.forkArgs()
   }
@@ -697,7 +712,7 @@ trait JavaModule
    * Any environment variables you want to pass to the forked JVM under `run`,
    * `test` or `repl`
    */
-  def forkEnv: T[Map[String, String]] = T {
+  override def forkEnv: T[Map[String, String]] = T {
     // overridden here for binary compatibility (0.11.x)
     super.forkEnv()
   }
@@ -815,80 +830,22 @@ trait JavaModule
     }
   }
 
-  def runUseArgsFile: T[Boolean] = T {
+  override def runUseArgsFile: T[Boolean] = T {
     // overridden here for binary compatibility (0.11.x)
     super.runUseArgsFile()
   }
 
-  /**
-   * Runs this module's code in-process within an isolated classloader. This is
-   * faster than `run`, but in exchange you have less isolation between runs
-   * since the code can dirty the parent Mill process and potentially leave it
-   * in a bad state.
-   */
-  def runLocal(args: Task[Args] = T.task(Args())): Command[Unit] = T.command {
-    Jvm.runLocal(
-      finalMainClass(),
-      runClasspath().map(_.path),
-      args().value
-    )
-  }
-
-  /**
-   * Runs this module's code in a subprocess and waits for it to finish
-   */
-  def run(args: Task[Args] = T.task(Args())): Command[Unit] = T.command {
-    try Result.Success(
-        Jvm.runSubprocess(
-          finalMainClass(),
-          runClasspath().map(_.path),
-          forkArgs(),
-          forkEnv(),
-          args().value,
-          workingDir = forkWorkingDir(),
-          useCpPassingJar = runUseArgsFile()
-        )
-      )
-    catch {
-      case e: Exception =>
-        Result.Failure("subprocess failed")
-    }
+  override def runLocal(args: Task[Args] = T.task(Args())): Command[Unit] = {
+    // overridden here for binary compatibility (0.11.x)
+    super.runLocal(args)
   }
 
-  private[this] def backgroundSetup(dest: os.Path): (Path, Path, String) = {
-    val token = java.util.UUID.randomUUID().toString
-    val procId = dest / ".mill-background-process-id"
-    val procTombstone = dest / ".mill-background-process-tombstone"
-    // The backgrounded subprocesses poll the procId file, and kill themselves
-    // when the procId file is deleted. This deletion happens immediately before
-    // the body of these commands run, but we cannot be sure the subprocess has
-    // had time to notice.
-    //
-    // To make sure we wait for the previous subprocess to
-    // die, we make the subprocess write a tombstone file out when it kills
-    // itself due to procId being deleted, and we wait a short time on task-start
-    // to see if such a tombstone appears. If a tombstone appears, we can be sure
-    // the subprocess has killed itself, and can continue. If a tombstone doesn't
-    // appear in a short amount of time, we assume the subprocess exited or was
-    // killed via some other means, and continue anyway.
-    val start = System.currentTimeMillis()
-    while ({
-      if (os.exists(procTombstone)) {
-        Thread.sleep(10)
-        os.remove.all(procTombstone)
-        true
-      } else {
-        Thread.sleep(10)
-        System.currentTimeMillis() - start < 100
-      }
-    }) ()
-
-    os.write(procId, token)
-    os.write(procTombstone, token)
-    (procId, procTombstone, token)
+  override def run(args: Task[Args] = T.task(Args())): Command[Unit] = {
+    // overridden here for binary compatibility (0.11.x)
+    super.run(args)
   }
 
-  protected def doRunBackground(
+  override protected def doRunBackground(
       taskDest: Path,
       runClasspath: Seq[PathRef],
       zwBackgroundWrapperClasspath: Agg[PathRef],
@@ -898,40 +855,26 @@ trait JavaModule
       forkWorkingDir: Path,
       runUseArgsFile: Boolean,
       backgroundOutputs: Option[Tuple2[ProcessOutput, ProcessOutput]]
-  )(args: String*): Ctx => Result[Unit] = ctx => {
-    val (procId, procTombstone, token) = backgroundSetup(taskDest)
-    try Result.Success(
-        Jvm.runSubprocessWithBackgroundOutputs(
-          "mill.scalalib.backgroundwrapper.BackgroundWrapper",
-          (runClasspath ++ zwBackgroundWrapperClasspath).map(_.path),
-          forkArgs,
-          forkEnv,
-          Seq(procId.toString, procTombstone.toString, token, finalMainClass) ++ args,
-          workingDir = forkWorkingDir,
-          backgroundOutputs,
-          useCpPassingJar = runUseArgsFile
-        )(ctx)
-      )
-    catch {
-      case e: Exception =>
-        Result.Failure("subprocess failed")
-    }
+  )(args: String*): Ctx => Result[Unit] = {
+    // overridden here for binary compatibility (0.11.x)
+    super.doRunBackground(
+      taskDest,
+      runClasspath,
+      zwBackgroundWrapperClasspath,
+      forkArgs,
+      forkEnv,
+      finalMainClass,
+      forkWorkingDir,
+      runUseArgsFile,
+      backgroundOutputs
+    )(args: _*)
+  }
+
+  override def runBackgroundLogToConsole: Boolean = {
+    // overridden here for binary compatibility (0.11.x)
+    super.runBackgroundLogToConsole
   }
 
-  /**
-   * If true, stdout and stderr of the process executed by `runBackground`
-   * or `runMainBackground` is sent to mill's stdout/stderr (which usualy
-   * flow to the console).
-   *
-   * If false, output will be directed to files `stdout.log` and `stderr.log`
-   * in `runBackground.dest` (or `runMainBackground.dest`)
-   */
-  def runBackgroundLogToConsole: Boolean = true
-
-  private def backgroundOutputs(dest: os.Path) =
-    if (runBackgroundLogToConsole) Some((os.Inherit, os.Inherit))
-    else Jvm.defaultBackgroundOutputs(dest)
-
   /**
    * Runs this module's code in a background process, until it dies or
    * `runBackground` is used again. This lets you continue using Mill while
@@ -943,72 +886,33 @@ trait JavaModule
    * when ready. This is useful when working on long-running server processes
    * that would otherwise run forever
    */
-  def runBackground(args: String*): Command[Unit] = T.command {
-    val ctx = implicitly[Ctx]
-
-    doRunBackground(
-      taskDest = T.dest,
-      runClasspath = runClasspath(),
-      zwBackgroundWrapperClasspath = zincWorker().backgroundWrapperClasspath(),
-      forkArgs = forkArgs(),
-      forkEnv = forkEnv(),
-      finalMainClass = finalMainClass(),
-      forkWorkingDir = forkWorkingDir(),
-      runUseArgsFile = runUseArgsFile(),
-      backgroundOutputs = backgroundOutputs(T.dest)
-    )(args: _*)(ctx)
+  def runBackground(args: String*): Command[Unit] = {
+    val task = runBackgroundTask(finalMainClass, T.task { Args(args) })
+    T.command { task }
   }
 
   /**
    * Same as `runBackground`, but lets you specify a main class to run
    */
-  def runMainBackground(mainClass: String, args: String*): Command[Unit] = T.command {
-    val ctx = implicitly[Ctx]
-
-    doRunBackground(
-      taskDest = T.dest,
-      runClasspath = runClasspath(),
-      zwBackgroundWrapperClasspath = zincWorker().backgroundWrapperClasspath(),
-      forkArgs = forkArgs(),
-      forkEnv = forkEnv(),
-      finalMainClass = mainClass,
-      forkWorkingDir = forkWorkingDir(),
-      runUseArgsFile = runUseArgsFile(),
-      backgroundOutputs = backgroundOutputs(T.dest)
-    )(args: _*)(ctx)
+  override def runMainBackground(mainClass: String, args: String*): Command[Unit] = {
+    // overridden here for binary compatibility (0.11.x)
+    super.runMainBackground(mainClass, args: _*)
   }
 
   /**
    * Same as `runLocal`, but lets you specify a main class to run
    */
-  def runMainLocal(mainClass: String, args: String*): Command[Unit] =
-    T.command {
-      Jvm.runLocal(
-        mainClass,
-        runClasspath().map(_.path),
-        args
-      )
-    }
+  override def runMainLocal(mainClass: String, args: String*): Command[Unit] = {
+    // overridden here for binary compatibility (0.11.x)
+    super.runMainLocal(mainClass, args: _*)
+  }
 
   /**
    * Same as `run`, but lets you specify a main class to run
    */
-  def runMain(mainClass: String, args: String*): Command[Unit] = T.command {
-    try Result.Success(
-        Jvm.runSubprocess(
-          mainClass,
-          runClasspath().map(_.path),
-          forkArgs(),
-          forkEnv(),
-          args,
-          workingDir = forkWorkingDir(),
-          useCpPassingJar = runUseArgsFile()
-        )
-      )
-    catch {
-      case e: Exception =>
-        Result.Failure("subprocess failed")
-    }
+  override def runMain(mainClass: String, args: String*): Command[Unit] = {
+    // overridden here for binary compatibility (0.11.x)
+    super.runMain(mainClass, args: _*)
   }
 
   /**
@@ -1032,7 +936,7 @@ trait JavaModule
    */
   def artifactSuffix: T[String] = platformSuffix()
 
-  def forkWorkingDir: T[Path] = T {
+  override def forkWorkingDir: T[Path] = T {
     // overridden here for binary compatibility (0.11.x)
     super.forkWorkingDir()
   }
diff --git a/scalalib/src/mill/scalalib/RunModule.scala b/scalalib/src/mill/scalalib/RunModule.scala
index 6dc8a249975..a8e94c8b9d0 100644
--- a/scalalib/src/mill/scalalib/RunModule.scala
+++ b/scalalib/src/mill/scalalib/RunModule.scala
@@ -1,11 +1,15 @@
 package mill.scalalib
 
-import mill.T
-import mill.define.Module
 import mill.api.JsonFormatters.pathReadWrite
-import mill.api.PathRef
+import mill.api.{Ctx, PathRef, Result}
+import mill.define.{Command, Task}
+import mill.util.Jvm
+import mill.{Agg, Args, T}
+import os.{Path, ProcessOutput}
 
-trait RunModule extends Module {
+import scala.util.control.NonFatal
+
+trait RunModule extends WithZincWorker {
 
   /**
    * Any command-line parameters you want to pass to the forked JVM.
@@ -25,11 +29,214 @@ trait RunModule extends Module {
    */
   def runClasspath: T[Seq[PathRef]] = T { Seq.empty[PathRef] }
 
+  /**
+   * The elements of the run classpath which are local to this module.
+   * This is typically the output of a compilation step and bundles runtime resources.
+   */
+  def localRunClasspath: T[Seq[PathRef]] = T { Seq.empty[PathRef] }
+
+  /**
+   * Allows you to specify an explicit main class to use for the `run` command.
+   * If none is specified, the classpath is searched for an appropriate main
+   * class to use if one exists.
+   */
+  def mainClass: T[Option[String]] = None
+
+  def allLocalMainClasses: T[Seq[String]] = T {
+    zincWorker().worker().discoverMainClasses(localRunClasspath().map(_.path))
+  }
+
+  def finalMainClassOpt: T[Either[String, String]] = T {
+    mainClass() match {
+      case Some(m) => Right(m)
+      case None =>
+        allLocalMainClasses() match {
+          case Seq() => Left("No main class specified or found")
+          case Seq(main) => Right(main)
+          case mains =>
+            Left(
+              s"Multiple main classes found (${mains.mkString(",")}) " +
+                "please explicitly specify which one to use by overriding mainClass"
+            )
+        }
+    }
+  }
+
+  def finalMainClass: T[String] = T {
+    finalMainClassOpt() match {
+      case Right(main) => Result.Success(main)
+      case Left(msg) => Result.Failure(msg)
+    }
+  }
+
   /**
    * Control whether `run*`-targets should use an args file to pass command line args, if possible.
    */
   def runUseArgsFile: T[Boolean] = T { scala.util.Properties.isWin }
 
-//  def zincWorker: ModuleRef[ZincWorkerModule] = ModuleRef(mill.scalalib.ZincWorkerModule)
+  /**
+   * Runs this module's code in a subprocess and waits for it to finish
+   */
+  def run(args: Task[Args] = T.task(Args())): Command[Unit] = T.command {
+    runForkedTask(finalMainClass, args)
+  }
+
+  /**
+   * Runs this module's code in-process within an isolated classloader. This is
+   * faster than `run`, but in exchange you have less isolation between runs
+   * since the code can dirty the parent Mill process and potentially leave it
+   * in a bad state.
+   */
+  def runLocal(args: Task[Args] = T.task(Args())): Command[Unit] = T.command {
+    runLocalTask(finalMainClass, args)
+  }
+
+  /**
+   * Same as `run`, but lets you specify a main class to run
+   */
+  def runMain(mainClass: String, args: String*): Command[Unit] = {
+    val task = runForkedTask(T.task { mainClass }, T.task { Args(args) })
+    T.command { task }
+  }
+
+  /**
+   * Same as `runBackground`, but lets you specify a main class to run
+   */
+  def runMainBackground(mainClass: String, args: String*): Command[Unit] = {
+    val task = runBackgroundTask(T.task { mainClass }, T.task { Args(args) })
+    T.command { task }
+  }
+
+  /**
+   * Same as `runLocal`, but lets you specify a main class to run
+   */
+  def runMainLocal(mainClass: String, args: String*): Command[Unit] = {
+    val task = runLocalTask(T.task { mainClass }, T.task { Args(args) })
+    T.command { task }
+  }
+
+  /**
+   * Runs this module's code in a subprocess and waits for it to finish
+   */
+  def runForkedTask(mainClass: Task[String], args: Task[Args] = T.task(Args())): Task[Unit] =
+    T.task {
+      try Result.Success(
+          Jvm.runSubprocess(
+            mainClass(),
+            runClasspath().map(_.path),
+            forkArgs(),
+            forkEnv(),
+            args().value,
+            workingDir = forkWorkingDir(),
+            useCpPassingJar = runUseArgsFile()
+          )
+        )
+      catch {
+        case NonFatal(_) => Result.Failure("Subprocess failed")
+      }
+    }
+
+  def runLocalTask(mainClass: Task[String], args: Task[Args] = T.task(Args())): Task[Unit] =
+    T.task {
+      Jvm.runLocal(
+        mainClass(),
+        runClasspath().map(_.path),
+        args().value
+      )
+    }
+
+  def runBackgroundTask(mainClass: Task[String], args: Task[Args] = T.task(Args())): Task[Unit] =
+    T.task {
+      doRunBackground(
+        taskDest = T.dest,
+        runClasspath = runClasspath(),
+        zwBackgroundWrapperClasspath = zincWorker().backgroundWrapperClasspath(),
+        forkArgs = forkArgs(),
+        forkEnv = forkEnv(),
+        finalMainClass = mainClass(),
+        forkWorkingDir = forkWorkingDir(),
+        runUseArgsFile = runUseArgsFile(),
+        backgroundOutputs = backgroundOutputs(T.dest)
+      )(args().value: _*)(T.ctx())
+    }
+
+  /**
+   * If true, stdout and stderr of the process executed by `runBackground`
+   * or `runMainBackground` is sent to mill's stdout/stderr (which usualy
+   * flow to the console).
+   *
+   * If false, output will be directed to files `stdout.log` and `stderr.log`
+   * in `runBackground.dest` (or `runMainBackground.dest`)
+   */
+  // TODO: make this a task, to be more dynamic
+  def runBackgroundLogToConsole: Boolean = true
+
+  private def backgroundOutputs(dest: os.Path): Option[(ProcessOutput, ProcessOutput)] = {
+    if (runBackgroundLogToConsole) Some((os.Inherit, os.Inherit))
+    else Jvm.defaultBackgroundOutputs(dest)
+  }
+
+  protected def doRunBackground(
+      taskDest: Path,
+      runClasspath: Seq[PathRef],
+      zwBackgroundWrapperClasspath: Agg[PathRef],
+      forkArgs: Seq[String],
+      forkEnv: Map[String, String],
+      finalMainClass: String,
+      forkWorkingDir: Path,
+      runUseArgsFile: Boolean,
+      backgroundOutputs: Option[Tuple2[ProcessOutput, ProcessOutput]]
+  )(args: String*): Ctx => Result[Unit] = ctx => {
+    val (procId, procTombstone, token) = backgroundSetup(taskDest)
+    try Result.Success(
+        Jvm.runSubprocessWithBackgroundOutputs(
+          "mill.scalalib.backgroundwrapper.BackgroundWrapper",
+          (runClasspath ++ zwBackgroundWrapperClasspath).map(_.path),
+          forkArgs,
+          forkEnv,
+          Seq(procId.toString, procTombstone.toString, token, finalMainClass) ++ args,
+          workingDir = forkWorkingDir,
+          backgroundOutputs,
+          useCpPassingJar = runUseArgsFile
+        )(ctx)
+      )
+    catch {
+      case e: Exception =>
+        Result.Failure("subprocess failed")
+    }
+  }
+
+  private[this] def backgroundSetup(dest: os.Path): (Path, Path, String) = {
+    val token = java.util.UUID.randomUUID().toString
+    val procId = dest / ".mill-background-process-id"
+    val procTombstone = dest / ".mill-background-process-tombstone"
+    // The background subprocesses poll the procId file, and kill themselves
+    // when the procId file is deleted. This deletion happens immediately before
+    // the body of these commands run, but we cannot be sure the subprocess has
+    // had time to notice.
+    //
+    // To make sure we wait for the previous subprocess to
+    // die, we make the subprocess write a tombstone file out when it kills
+    // itself due to procId being deleted, and we wait a short time on task-start
+    // to see if such a tombstone appears. If a tombstone appears, we can be sure
+    // the subprocess has killed itself, and can continue. If a tombstone doesn't
+    // appear in a short amount of time, we assume the subprocess exited or was
+    // killed via some other means, and continue anyway.
+    val start = System.currentTimeMillis()
+    while ({
+      if (os.exists(procTombstone)) {
+        Thread.sleep(10)
+        os.remove.all(procTombstone)
+        true
+      } else {
+        Thread.sleep(10)
+        System.currentTimeMillis() - start < 100
+      }
+    }) ()
+
+    os.write(procId, token)
+    os.write(procTombstone, token)
+    (procId, procTombstone, token)
+  }
 
 }
diff --git a/scalalib/src/mill/scalalib/TestModule.scala b/scalalib/src/mill/scalalib/TestModule.scala
index c3c25f007f2..12944e7a3f3 100644
--- a/scalalib/src/mill/scalalib/TestModule.scala
+++ b/scalalib/src/mill/scalalib/TestModule.scala
@@ -20,9 +20,9 @@ trait TestModule
 
   /**
    * The classpath containing the tests. This is most likely the output of the compilation target.
-   * Return by default the empty `Seq` for compatibility (0.11.x).
+   * By default this uses the result of [[localRunClasspath]], which is most likely the result of a local compilation.
    */
-  def testClasspath: T[Seq[PathRef]] = T { Seq.empty[PathRef] }
+  def testClasspath: T[Seq[PathRef]] = T { localRunClasspath() }
 
   /**
    * The test framework to use.
diff --git a/scalalib/test/src/mill/scalalib/HelloWorldTests.scala b/scalalib/test/src/mill/scalalib/HelloWorldTests.scala
index a124f18504e..220ddf6ce14 100644
--- a/scalalib/test/src/mill/scalalib/HelloWorldTests.scala
+++ b/scalalib/test/src/mill/scalalib/HelloWorldTests.scala
@@ -760,7 +760,7 @@ object HelloWorldTests extends TestSuite {
       }
 
       "notRunInvalidMainObject" - workspaceTest(HelloWorld) { eval =>
-        val Left(Result.Failure("subprocess failed", _)) =
+        val Left(Result.Failure("Subprocess failed", _)) =
           eval.apply(HelloWorld.core.runMain("Invalid"))
       }
       "notRunWhenCompileFailed" - workspaceTest(HelloWorld) { eval =>
diff --git a/scalalib/worker/src/mill/scalalib/worker/ZincProblemPosition.scala b/scalalib/worker/src/mill/scalalib/worker/ZincProblemPosition.scala
index 99abe811212..65e0c3d1e4d 100644
--- a/scalalib/worker/src/mill/scalalib/worker/ZincProblemPosition.scala
+++ b/scalalib/worker/src/mill/scalalib/worker/ZincProblemPosition.scala
@@ -2,31 +2,34 @@ package mill.scalalib.worker
 
 import java.io.File
 import java.util.Optional
-
 import mill.api.{ProblemPosition, internal}
 
+import scala.jdk.OptionConverters.RichOptional
+
 @internal
 class ZincProblemPosition(base: xsbti.Position) extends ProblemPosition {
 
-  object JavaOptionConverter {
-    implicit def convertInt(x: Optional[Integer]): Option[Int] =
-      if (x.isPresent) Some(x.get().intValue()) else None
-    implicit def convert[T](x: Optional[T]): Option[T] = if (x.isPresent) Some(x.get()) else None
-  }
+  import ZincProblemPosition.ToIntOption
 
-  import JavaOptionConverter._
-
-  override def line: Option[Int] = base.line()
+  override def line: Option[Int] = base.line().toIntOption
   override def lineContent: String = base.lineContent()
-  override def offset: Option[Int] = base.offset()
-  override def pointer: Option[Int] = base.pointer()
-  override def pointerSpace: Option[String] = base.pointerSpace()
-  override def sourcePath: Option[String] = base.sourcePath()
-  override def sourceFile: Option[File] = base.sourceFile()
-  override def startOffset: Option[Int] = base.startOffset()
-  override def endOffset: Option[Int] = base.endOffset()
-  override def startLine: Option[Int] = base.startLine()
-  override def startColumn: Option[Int] = base.startColumn()
-  override def endLine: Option[Int] = base.endLine()
-  override def endColumn: Option[Int] = base.endColumn()
+  override def offset: Option[Int] = base.offset().toIntOption
+  override def pointer: Option[Int] = base.pointer().toIntOption
+  override def pointerSpace: Option[String] = base.pointerSpace().toScala
+  override def sourcePath: Option[String] = base.sourcePath().toScala
+  override def sourceFile: Option[File] = base.sourceFile().toScala
+  override def startOffset: Option[Int] = base.startOffset().toIntOption
+  override def endOffset: Option[Int] = base.endOffset().toIntOption
+  override def startLine: Option[Int] = base.startLine().toIntOption
+  override def startColumn: Option[Int] = base.startColumn().toIntOption
+  override def endLine: Option[Int] = base.endLine().toIntOption
+  override def endColumn: Option[Int] = base.endColumn().toIntOption
+}
+
+object ZincProblemPosition {
+
+  private implicit class ToIntOption(val opt: Optional[Integer]) extends AnyVal {
+    def toIntOption: Option[Int] = if (opt.isPresent()) Option(opt.get().intValue()) else None
+  }
+
 }
diff --git a/scalalib/worker/src/mill/scalalib/worker/ZincWorkerImpl.scala b/scalalib/worker/src/mill/scalalib/worker/ZincWorkerImpl.scala
index 4dec3759702..8dd93778750 100644
--- a/scalalib/worker/src/mill/scalalib/worker/ZincWorkerImpl.scala
+++ b/scalalib/worker/src/mill/scalalib/worker/ZincWorkerImpl.scala
@@ -1,8 +1,15 @@
 package mill.scalalib.worker
 
 import mill.api.Loose.Agg
-import mill.api.{CompileProblemReporter, KeyedLockedCache, PathRef, Result, internal}
-import mill.scalalib.api.{CompilationResult, ZincWorkerApi, ZincWorkerUtil, Versions}
+import mill.api.{
+  CompileProblemReporter,
+  DummyOutputStream,
+  KeyedLockedCache,
+  PathRef,
+  Result,
+  internal
+}
+import mill.scalalib.api.{CompilationResult, Versions, ZincWorkerApi, ZincWorkerUtil}
 import sbt.internal.inc.{
   Analysis,
   CompileFailed,
@@ -32,11 +39,16 @@ import xsbti.compile.{
 }
 import xsbti.{PathBasedFile, VirtualFile}
 
-import java.io.File
+import java.io.{File, PrintWriter}
 import java.util.Optional
+import scala.annotation.tailrec
 import scala.collection.mutable
 import scala.ref.SoftReference
+import scala.tools.nsc.{CloseableRegistry, Settings}
+import scala.tools.nsc.classpath.{AggregateClassPath, ClassPathFactory}
+import scala.tools.scalap.{ByteArrayReader, Classfile, JavaWriter}
 import scala.util.Properties.isWin
+import scala.util.Using
 
 @internal
 class ZincWorkerImpl(
@@ -275,6 +287,47 @@ class ZincWorkerImpl(
 
   }
 
+  /**
+   * Discover main classes by inspecting the classpath.
+   *
+   * This implementation uses the Scala API of `scalap` to inspect the classfiles for `public static main` methods.
+   *
+   * In contrast to [[discoverMainClasses()]], this version does not need a successful zinc compilation,
+   * which makes it independent of the actual used compiler.
+   * It should also work for JVM bytecode generated by Kotlin and other langauges.
+   *
+   * This implementation is only in this "zinc"-specific module, because this module is already shared between all `JavaModule`s.
+   */
+  override def discoverMainClasses(classpath: Seq[os.Path]): Seq[String] = {
+    val cp = classpath.map(_.toNIO.toString()).mkString(File.pathSeparator)
+
+    val settings = new Settings()
+    Using.resource(new CloseableRegistry) { registry =>
+      val path = AggregateClassPath(
+        new ClassPathFactory(settings, registry).classesInExpandedPath(cp)
+      )
+
+      val mainClasses = for {
+        foundPackage <- ZincWorkerImpl.recursive("", (p: String) => path.packages(p).map(_.name))
+        classFile <- path.classes(foundPackage)
+        cf = {
+          val bytes = os.read.bytes(os.Path(classFile.file.file))
+          val reader = new ByteArrayReader(bytes)
+          new Classfile(reader)
+        }
+        jw = new JavaWriter(cf, new PrintWriter(DummyOutputStream))
+        method <- cf.methods
+        static = jw.isStatic(method.flags)
+        methodName = jw.getName(method.name)
+        methodType = jw.getType(method.tpe)
+        if static && methodName == "main" && methodType == "(scala.Array[java.lang.String]): scala.Unit"
+        className = jw.getClassName(cf.classname)
+      } yield className
+
+      mainClasses
+    }
+  }
+
   def discoverMainClasses(compilationResult: CompilationResult): Seq[String] = {
     def toScala[A](o: Optional[A]): Option[A] = if (o.isPresent) Some(o.get) else None
 
@@ -582,3 +635,40 @@ class ZincWorkerImpl(
     javaOnlyCompilersCache.clear()
   }
 }
+
+object ZincWorkerImpl {
+  // copied from ModuleUtils
+  private def recursive[T <: String](start: T, deps: T => Seq[T]): Seq[T] = {
+
+    @tailrec def rec(
+        seenModules: List[T],
+        toAnalyze: List[(List[T], List[T])]
+    ): List[T] = {
+      toAnalyze match {
+        case Nil => seenModules
+        case traces :: rest =>
+          traces match {
+            case (_, Nil) => rec(seenModules, rest)
+            case (trace, cand :: remaining) =>
+              if (trace.contains(cand)) {
+                // cycle!
+                val rendered =
+                  (cand :: (cand :: trace.takeWhile(_ != cand)).reverse).mkString(" -> ")
+                val msg = s"cycle detected: ${rendered}"
+                println(msg)
+                throw sys.error(msg)
+              }
+              rec(
+                seenModules ++ Seq(cand),
+                toAnalyze = ((cand :: trace, deps(cand).toList)) :: (trace, remaining) :: rest
+              )
+          }
+      }
+    }
+
+    rec(
+      seenModules = List(),
+      toAnalyze = List((List(start), deps(start).toList))
+    ).reverse
+  }
+}