diff --git a/.github/workflows/actions.yml b/.github/workflows/actions.yml
index d47bf7378cb..eb938897da8 100644
--- a/.github/workflows/actions.yml
+++ b/.github/workflows/actions.yml
@@ -34,6 +34,7 @@ jobs:
       java-version: 11
       millargs: __.compile
       populate_cache: true
+
   build-windows:
     uses: ./.github/workflows/run-mill-action.yml
     with:
@@ -41,6 +42,30 @@ jobs:
       java-version: 11
       millargs: __.compile
       populate_cache: true
+
+  test-docs:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with: { fetch-depth: 0 }
+
+      - run: ./mill -i docs.githubPages
+
+  compiler-bridge:
+    needs: build-linux
+    uses: ./.github/workflows/run-mill-action.yml
+    with:
+      java-version: '8'
+      millargs: bridge.__.publishLocal
+      env-bridge-versions: 'essential'
+
+  lint-autofix:
+    needs: build-linux
+    uses: ./.github/workflows/run-mill-action.yml
+    with:
+      java-version: '11'
+      buildcmd: ./mill -i mill.scalalib.scalafmt.ScalafmtModule/checkFormatAll __.sources + __.mimaReportBinaryIssues + __.fix --check
+
   itest:
     needs: build-linux
     strategy:
@@ -99,31 +124,11 @@ jobs:
           - java-version: 17
             millargs: "'integration.invalidation.__.server.testCached'"
 
-          # Check docsite compiles
-          - java-version: 11
-            millargs: docs.githubPages
-
-
     uses: ./.github/workflows/run-mill-action.yml
     with:
       java-version: ${{ matrix.java-version }}
       millargs: ${{ matrix.millargs }}
 
-  compiler-bridge:
-    needs: build-linux
-    uses: ./.github/workflows/run-mill-action.yml
-    with:
-      java-version: '8'
-      millargs: bridge.__.publishLocal
-      env-bridge-versions: 'essential'
-
-  format-scalafix-bincompat:
-    needs: build-linux
-    uses: ./.github/workflows/run-mill-action.yml
-    with:
-      java-version: '11'
-      buildcmd: ./mill -i mill.scalalib.scalafmt.ScalafmtModule/checkFormatAll __.sources + __.mimaReportBinaryIssues + __.fix --check
-
   windows:
     needs: build-windows
     strategy:
@@ -152,7 +157,7 @@ jobs:
   publish-sonatype:
     # when in master repo, publish all tags and manual runs on main
     if: github.repository == 'com-lihaoyi/mill' && (startsWith( github.ref, 'refs/tags/') || (github.ref == 'refs/heads/main' && github.event_name == 'workflow_dispatch' ) )
-    needs: [linux, windows, compiler-bridge, format-scalafix-bincompat, itest]
+    needs: [linux, windows, compiler-bridge, lint-autofix, itest, test-docs]
 
     runs-on: ubuntu-latest
 
@@ -171,8 +176,7 @@ jobs:
 
     steps:
       - uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
+        with: {fetch-depth: 0}
 
       - uses: coursier/cache-action@v6
 
@@ -194,8 +198,7 @@ jobs:
 
     steps:
       - uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
+        with: {fetch-depth: 0}
 
       - uses: coursier/cache-action@v6
 
diff --git a/.github/workflows/run-mill-action.yml b/.github/workflows/run-mill-action.yml
index 85b0085b967..7b62509a032 100644
--- a/.github/workflows/run-mill-action.yml
+++ b/.github/workflows/run-mill-action.yml
@@ -29,8 +29,7 @@ on:
         type: string
 
 jobs:
-  build:
-
+  run:
     runs-on: ${{ inputs.os }}
     continue-on-error: ${{ inputs.continue-on-error }}
     timeout-minutes: ${{ inputs.timeout-minutes }}
@@ -39,8 +38,6 @@ jobs:
 
     steps:
       - uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
         if: ${{ inputs.populate_cache }}
 
       - uses: actions/download-artifact@v4
diff --git a/build.mill b/build.mill
index 1a02d6470eb..06bb7f8d430 100644
--- a/build.mill
+++ b/build.mill
@@ -149,7 +149,7 @@ object Deps {
   val commonsIO = ivy"commons-io:commons-io:2.16.1"
   val lambdaTest = ivy"de.tototec:de.tobiasroeser.lambdatest:0.8.0"
   val log4j2Core = ivy"org.apache.logging.log4j:log4j-core:2.23.1"
-  val osLib = ivy"com.lihaoyi::os-lib:0.10.5"
+  val osLib = ivy"com.lihaoyi::os-lib:0.10.7-M1"
   val pprint = ivy"com.lihaoyi::pprint:0.9.0"
   val mainargs = ivy"com.lihaoyi::mainargs:0.7.4"
   val millModuledefsVersion = "0.11.0-M2"
diff --git a/docs/modules/ROOT/nav.adoc b/docs/modules/ROOT/nav.adoc
index 71b2cb90619..d5f4a49ed78 100644
--- a/docs/modules/ROOT/nav.adoc
+++ b/docs/modules/ROOT/nav.adoc
@@ -39,6 +39,7 @@
 * xref:Structuring_Large_Builds.adoc[]
 
 * xref:The_Mill_Evaluation_Model.adoc[]
+* xref:Mill_Sandboxing.adoc[]
 
 // This section talks about Mill plugins. While it could theoretically fit in
 // either section above, it is probably an important enough topic it is worth
diff --git a/docs/modules/ROOT/pages/Mill_Sandboxing.adoc b/docs/modules/ROOT/pages/Mill_Sandboxing.adoc
new file mode 100644
index 00000000000..6652e5bc482
--- /dev/null
+++ b/docs/modules/ROOT/pages/Mill_Sandboxing.adoc
@@ -0,0 +1,20 @@
+= Mill Sandboxing
+
+== Task Sandboxing
+
+include::example/depth/sandbox/1-task.adoc[]
+
+== Test Sandboxing
+
+include::example/depth/sandbox/2-test.adoc[]
+
+== Limitations
+
+Mill's approach to filesystem sandboxing is designed to avoid accidental interference
+between different Mill tasks. It is not designed to block intentional misbehavior, and
+tasks are always able to traverse the filesystem and do whatever they want. Furthermore,
+Mill's redirection of `os.pwd` does not apply to `java.io` or `java.nio` APIs, which are
+outside of Mill's control.
+
+However, by setting `os.pwd` to safe sandbox folders, we hope to minimize the cases where
+someone accidentally causes issues with their build by doing the wrong thing.
\ No newline at end of file
diff --git a/docs/modules/ROOT/pages/Tasks.adoc b/docs/modules/ROOT/pages/Tasks.adoc
index b1416790804..0cabefdd335 100644
--- a/docs/modules/ROOT/pages/Tasks.adoc
+++ b/docs/modules/ROOT/pages/Tasks.adoc
@@ -80,6 +80,6 @@ xref:Scala_Builtin_Commands.adoc#_show[show].
 If you want a way to run Mill commands and programmatically manipulate the
 tasks and outputs, you do so with your own evaluator command.
 
-=== Using ScalaModule.run as a task
+== Using ScalaModule.run as a task
 
 include::example/depth/tasks/11-module-run-task.adoc[]
diff --git a/example/depth/sandbox/1-task/build.mill b/example/depth/sandbox/1-task/build.mill
new file mode 100644
index 00000000000..6133eaf2013
--- /dev/null
+++ b/example/depth/sandbox/1-task/build.mill
@@ -0,0 +1,75 @@
+// In order to help manage your build, Mill performs some rudimentary filesystem
+// sandboxing to keep different tasks and modules from interfering with each other.
+// This tries to ensure your tasks only read and write from their designated `.dest/`
+// folders, which are unique to each task and thus guaranteed not to collide with
+// the filesystem operations of other tasks that may be occurring in parallel.
+//
+//
+// === `T.dest`
+// The standard way of working with a task's `.dest/` folder is through the `T.dest`
+// property. This is available within any task, and gives you access to the
+// `out/<module-names>/<task-name>.dest/` folder to use. The `.dest/` folder for
+// each task is lazily initialized when `T.dest` is referenced and used:
+
+package build
+import mill._
+
+object foo extends Module{
+  def tDestTask = T { println(T.dest.toString) }
+}
+
+/** Usage
+> ./mill foo.tDestTask
+.../out/foo/tDestTask.dest
+*/
+
+
+// === Task `os.pwd` redirection
+// Mill also redirects the `os.pwd` property from https://github.com/com-lihaoyi/os-lib[OS-Lib],
+// such that that also points towards a running task's own `.dest/` folder
+
+def osPwdTask = T { println(os.pwd.toString) }
+
+/** Usage
+> ./mill osPwdTask
+.../out/osPwdTask.dest
+*/
+
+// The redirection of `os.pwd` applies to `os.proc`, `os.call`, and `os.spawn` methods
+// as well. In the example below, we can see the `python3` subprocess we spawn prints
+// its `os.getcwd()`, which is our `osProcTask.dest/` sandbox folder:
+
+def osProcTask = T {
+  println(os.call(("python3", "-c", "import os; print(os.getcwd())"), cwd = T.dest).out.trim())
+}
+
+/** Usage
+> ./mill osProcTask
+.../out/osProcTask.dest
+*/
+
+// === Non-task `os.pwd` redirection
+//
+// Lastly, there is the possibily of calling `os.pwd` outside of a task. When outside of
+// a task there is no `.dest/` folder associated, so instead Mill will redirect `os.pwd`
+// towards an empty `sandbox/` folder in `out/mill-worker.../`:
+
+val externalPwd = os.pwd
+def externalPwdTask = T { println(externalPwd.toString) }
+
+/** Usage
+> ./mill externalPwdTask
+.../out/mill-worker-.../sandbox/sandbox
+*/
+
+
+// === Limitations of Mill's Sandboxing
+//
+// Mill's approach to filesystem sandboxing is designed to avoid accidental interference
+// between different Mill tasks. It is not designed to block intentional misbehavior, and
+// tasks are always able to traverse the filesystem and do whatever they want. Furthermore,
+// Mill's redirection of `os.pwd` does not apply to `java.io` or `java.nio` APIs, which are
+// outside of Mill's control.
+//
+// However, by setting `os.pwd` to safe sandbox folders, we hope to minimize the cases where
+// someone accidentally causes issues with their build by doing the wrong thing.
\ No newline at end of file
diff --git a/example/depth/sandbox/2-test/bar/src/bar/Bar.java b/example/depth/sandbox/2-test/bar/src/bar/Bar.java
new file mode 100644
index 00000000000..291650df9bf
--- /dev/null
+++ b/example/depth/sandbox/2-test/bar/src/bar/Bar.java
@@ -0,0 +1,7 @@
+package bar;
+
+public class Bar {
+    public static String generateHtml(String text) {
+        return "<p>" + text + "</p>";
+    }
+}
\ No newline at end of file
diff --git a/example/depth/sandbox/2-test/bar/test/src/bar/BarTests.java b/example/depth/sandbox/2-test/bar/test/src/bar/BarTests.java
new file mode 100644
index 00000000000..20139d0967b
--- /dev/null
+++ b/example/depth/sandbox/2-test/bar/test/src/bar/BarTests.java
@@ -0,0 +1,15 @@
+package bar;
+
+import static org.junit.Assert.assertEquals;
+import org.junit.Test;
+import java.nio.file.*;
+
+public class BarTests {
+    @Test
+    public void simple() throws Exception {
+        String result = Bar.generateHtml("world");
+        Path path = Paths.get("generated.html");
+        Files.write(path, result.getBytes());
+        assertEquals("<p>world</p>", new String(Files.readAllBytes(path)));
+    }
+}
\ No newline at end of file
diff --git a/example/depth/sandbox/2-test/build.mill b/example/depth/sandbox/2-test/build.mill
new file mode 100644
index 00000000000..08449fb18d8
--- /dev/null
+++ b/example/depth/sandbox/2-test/build.mill
@@ -0,0 +1,71 @@
+// Mill also creates sandbox folders for test suites to run in. Consider the
+// following build with two modules `foo` and `bar`, and their test suites
+// `foo.test` and `bar.test`:
+
+package build
+import mill._, javalib._
+
+trait MyModule extends JavaModule{
+  object test extends JavaTests with TestModule.Junit4
+}
+
+object foo extends MyModule{
+  def moduleDeps = Seq(bar)
+}
+
+object bar extends MyModule
+
+// For the sake of the example, both test modules contain tests that exercise the
+// logic in their corresponding non-test module, but also do some basic filesystem
+// operations at the same time, writing out a `generated.html` file and then reading it:
+
+/** See Also: foo/src/foo/Foo.java */
+/** See Also: foo/test/src/foo/FooTests.java */
+/** See Also: bar/src/bar/Bar.java */
+/** See Also: bar/test/src/bar/BarTests.java */
+
+// Both test suites can be run via
+
+/** Usage
+> ./mill __.test
+*/
+
+// Without sandboxing, due to the tests running in parallel, there is a race condition:
+// it's possible that `FooTests` may write the file, `BarTests` write over it, before
+// `FooTests` reads the output from `BarTests`. That would cause non-deterministic
+// flaky failures in your test suite that can be very difficult to debug and resolve.
+//
+// With Mill's test sandboxing, each test runs in a separate folder: the `.dest` folder
+// of the respective task and module. For example:
+//
+// - `foo.test` runs in `out/foo/test/test.dest/`
+// - `bar.test` runs in `out/bar/test/test.dest/`
+//
+// As a result, each test's `generated.html` file is written to its own dedicated
+// working directory, without colliding with each other on disk:
+
+/** Usage
+
+> find . | grep generated.html
+.../out/foo/test/test.dest/sandbox/generated.html
+.../out/bar/test/test.dest/sandbox/generated.html
+
+> cat out/foo/test/test.dest/sandbox/generated.html
+<h1>hello</h1>
+
+> cat out/bar/test/test.dest/sandbox/generated.html
+<p>world</p>
+
+*/
+
+// As each test suite runs in a different working directory by default, naive usage
+// reading and writing to the filesystem does not cause tests to interefere with
+// one another, which helps keep tests stable and deterministic even when run in
+// parallel
+//
+// Like Mill's Task sandboxing, Mill's Test sandboxing does not guard against
+// intentional misbehavior: tests can still walk the filesystem from the
+// sandbox folder via `..` or from the root folder `/` or home folder `~/`.
+// Nevertheless, it should add some simple guardrails to prevent many common
+// causes of inter-test interference, letting your test suite run in parallel both
+// quickly and reliably
\ No newline at end of file
diff --git a/example/depth/sandbox/2-test/foo/src/foo/Foo.java b/example/depth/sandbox/2-test/foo/src/foo/Foo.java
new file mode 100644
index 00000000000..809cfde1092
--- /dev/null
+++ b/example/depth/sandbox/2-test/foo/src/foo/Foo.java
@@ -0,0 +1,8 @@
+package foo;
+
+public class Foo {
+    public static String generateHtml(String text) {
+        return "<h1>" + text + "</h1>";
+    }
+}
+
diff --git a/example/depth/sandbox/2-test/foo/test/src/foo/FooTests.java b/example/depth/sandbox/2-test/foo/test/src/foo/FooTests.java
new file mode 100644
index 00000000000..5acd9084315
--- /dev/null
+++ b/example/depth/sandbox/2-test/foo/test/src/foo/FooTests.java
@@ -0,0 +1,15 @@
+package foo;
+
+import static org.junit.Assert.assertEquals;
+import org.junit.Test;
+import java.nio.file.*;
+
+public class FooTests {
+    @Test
+    public void simple() throws Exception {
+        String result = Foo.generateHtml("hello");
+        Path path = Paths.get("generated.html");
+        Files.write(path, result.getBytes());
+        assertEquals("<h1>hello</h1>", new String(Files.readAllBytes(path)));
+    }
+}
\ No newline at end of file
diff --git a/example/depth/tasks/11-module-run-task/bar/src/Bar.scala b/example/depth/tasks/11-module-run-task/bar/src/Bar.scala
index 72b88d2ad8f..c47ffa7f1fd 100644
--- a/example/depth/tasks/11-module-run-task/bar/src/Bar.scala
+++ b/example/depth/tasks/11-module-run-task/bar/src/Bar.scala
@@ -1,9 +1,8 @@
 package bar
 object Bar {
   def main(args: Array[String]) = {
-    val Array(destStr, sourceStrs @ _*) = args
-    val dest = os.Path(destStr)
-    for(sourceStr <- sourceStrs){
+    val dest = os.pwd
+    for(sourceStr <- args){
       val sourcePath = os.Path(sourceStr)
       for(p <- os.walk(sourcePath) if p.ext == "scala"){
         val text = os.read(p)
diff --git a/example/depth/tasks/11-module-run-task/build.mill b/example/depth/tasks/11-module-run-task/build.mill
index ead760076d7..3bcf03bd52c 100644
--- a/example/depth/tasks/11-module-run-task/build.mill
+++ b/example/depth/tasks/11-module-run-task/build.mill
@@ -1,15 +1,15 @@
 package build
 import mill._, scalalib._
+import mill.util.Jvm
 
 object foo extends ScalaModule {
   def scalaVersion = "2.13.8"
   def moduleDeps = Seq(bar)
   def ivyDeps = Agg(ivy"com.lihaoyi::mainargs:0.4.0")
 
-  def barWorkingDir = T{ T.dest }
   def sources = T{
-    bar.run(T.task(Args(barWorkingDir(), super.sources().map(_.path))))()
-    Seq(PathRef(barWorkingDir()))
+    bar.runner().run(args = super.sources())
+    Seq(PathRef(T.dest))
   }
 }
 
@@ -22,9 +22,9 @@ object bar extends ScalaModule{
 // than defining the task logic in the `build.mill`, we instead put the build
 // logic within the `bar` module as `bar/src/Bar.scala`. In this example, we use
 // `Bar.scala` as a source-code pre-processor on the `foo` module source code:
-// we override `foo.sources`, passing the `super.sources()` to `bar.run` along
-// with a `barWorkingDir`, and returning a `PathRef(barWorkingDir())` as the
-// new `foo.sources`.
+// we override `foo.sources`, passing the `super.sources()` and `bar.runClasspath`
+// to `bar.runner().run` along with a `T.dest`, and returning a `PathRef(T.dest)`
+// as the new `foo.sources`.
 
 /** Usage
 
@@ -39,4 +39,24 @@ Foo.value: HELLO
 // own arbitrarily complex transformations. This is useful for build logic that
 // may not fit nicely inside a `build.mill` file, whether due to the sheer lines
 // of code or due to dependencies that may conflict with the Mill classpath
-// present in `build.mill`
\ No newline at end of file
+// present in `build.mill`
+//
+// `bar.runner().run` by default inherits the `mainClass`, `forkEnv`, `forkArgs`,
+// from the owning module `bar`, and the working directory from the calling task's
+// `T.dest`. You can also pass in these parameters explicitly to `run()` as named
+// arguments if you wish to override the defaults
+//
+// [source,scala]
+// ----
+// trait Runner{
+//   def run(args: os.Shellable,
+//           mainClass: String = null,
+//           forkArgs: Seq[String] = null,
+//           forkEnv: Map[String, String] = null,
+//           workingDir: os.Path = null,
+//           useCpPassingJar: java.lang.Boolean = null)
+//          (implicit ctx: Ctx): Unit
+// }
+// ----
+
+
diff --git a/example/package.mill b/example/package.mill
index 56a538d8c2d..df1392d1565 100644
--- a/example/package.mill
+++ b/example/package.mill
@@ -48,6 +48,7 @@ object `package` extends RootModule with Module {
     object modules extends Cross[ExampleCrossModule](build.listIn(millSourcePath / "modules"))
     object cross extends Cross[ExampleCrossModule](build.listIn(millSourcePath / "cross"))
     object large extends Cross[ExampleCrossModule](build.listIn(millSourcePath / "large"))
+    object sandbox extends Cross[ExampleCrossModule](build.listIn(millSourcePath / "sandbox"))
   }
 
   object extending extends Module {
diff --git a/integration/feature/docannotations/src/DocAnnotationsTests.scala b/integration/feature/docannotations/src/DocAnnotationsTests.scala
index b23fdeb9b36..bfd237a9ae8 100644
--- a/integration/feature/docannotations/src/DocAnnotationsTests.scala
+++ b/integration/feature/docannotations/src/DocAnnotationsTests.scala
@@ -76,12 +76,13 @@ object DocAnnotationsTests extends UtestIntegrationTestSuite {
             |    args <str>...
             |
             |Inputs:
-            |    core.finalMainClass
+            |    core.finalMainClassOpt
             |    core.runClasspath
             |    core.forkArgs
             |    core.forkEnv
-            |    core.forkWorkingDir
             |    core.runUseArgsFile
+            |    core.finalMainClass
+            |    core.forkWorkingDir
             |""",
           run
         )
diff --git a/main/api/src/mill/api/PathRef.scala b/main/api/src/mill/api/PathRef.scala
index 5951ebdee46..ad1ae5a44eb 100644
--- a/main/api/src/mill/api/PathRef.scala
+++ b/main/api/src/mill/api/PathRef.scala
@@ -45,6 +45,7 @@ case class PathRef private (
 }
 
 object PathRef {
+  implicit def shellable(p: PathRef): os.Shellable = p.path
 
   /**
    * This class maintains a cache of already validated paths.
diff --git a/main/eval/src/mill/eval/EvaluatorCore.scala b/main/eval/src/mill/eval/EvaluatorCore.scala
index 8a85cf7bbac..777640d7eba 100644
--- a/main/eval/src/mill/eval/EvaluatorCore.scala
+++ b/main/eval/src/mill/eval/EvaluatorCore.scala
@@ -41,7 +41,7 @@ private[mill] trait EvaluatorCore extends GroupEvaluator {
 
       def contextLoggerMsg(threadId: Int) =
         if (effectiveThreadCount == 1) ""
-        else s"[#${if (effectiveThreadCount > 9) f"$threadId%02d" else threadId}] "
+        else s"#${if (effectiveThreadCount > 9) f"$threadId%02d" else threadId} "
 
       try evaluate0(goals, logger, reporter, testReporter, ec, contextLoggerMsg, serialCommandExec)
       finally ec.close()
diff --git a/main/eval/src/mill/eval/GroupEvaluator.scala b/main/eval/src/mill/eval/GroupEvaluator.scala
index 6f563252ec6..fafd4eb6ba3 100644
--- a/main/eval/src/mill/eval/GroupEvaluator.scala
+++ b/main/eval/src/mill/eval/GroupEvaluator.scala
@@ -12,7 +12,8 @@ import java.lang.reflect.Method
 import scala.collection.mutable
 import scala.reflect.NameTransformer.encode
 import scala.util.control.NonFatal
-import scala.util.{DynamicVariable, Using}
+import scala.util.hashing.MurmurHash3
+import scala.util.DynamicVariable
 
 /**
  * Logic around evaluating a single group, which is a collection of [[Task]]s
@@ -43,32 +44,6 @@ private[mill] trait GroupEvaluator {
   val effectiveThreadCount: Int =
     this.threadCount.getOrElse(Runtime.getRuntime().availableProcessors())
 
-  /**
-   * Synchronize evaluations of the same terminal task.
-   * This isn't necessarily needed for normal Mill executions,
-   * but in an BSP context, where multiple requests where handled concurrently in the same Mill instance,
-   * evaluating the same task concurrently  can happen.
-   *
-   * We don't synchronize multiple Mill-instances (e.g. run in two shells)
-   * or multiple evaluator-instances (which should have different `out`-dirs anyway.
-   */
-  private object synchronizedEval {
-    private val keyLock = new KeyLock[Segments]()
-    def apply[T](terminal: Terminal, onCollision: Option[() => Unit] = None)(f: => T): T =
-      terminal match {
-        case t: Terminal.Task[_] =>
-          // A un-labelled terminal task won't be synchronized
-          // as there is no filesystem cache region assigned to it
-          f
-        case Terminal.Labelled(_, segments) =>
-          // A labelled terminal needs synchronization due to
-          // a shared cache region in the filesystem
-          Using.resource(keyLock.lock(segments, onCollision)) { _ =>
-            f
-          }
-      }
-  }
-
   // those result which are inputs but not contained in this terminal group
   def evaluateGroupCached(
       terminal: Terminal,
@@ -80,20 +55,14 @@ private[mill] trait GroupEvaluator {
       logger: ColorLogger,
       classToTransitiveClasses: Map[Class[_], IndexedSeq[Class[_]]],
       allTransitiveClassMethods: Map[Class[_], Map[String, Method]]
-  ): GroupEvaluator.Results = synchronizedEval(
-    terminal,
-    onCollision =
-      Some(() => logger.debug(s"Waiting for concurrently executing task ${terminal.render}"))
-  ) {
+  ): GroupEvaluator.Results = {
 
-    val externalInputsHash = scala.util.hashing.MurmurHash3.orderedHash(
+    val externalInputsHash = MurmurHash3.orderedHash(
       group.items.flatMap(_.inputs).filter(!group.contains(_))
         .flatMap(results(_).result.asSuccess.map(_.value._2))
     )
 
-    val sideHashes = scala.util.hashing.MurmurHash3.orderedHash(
-      group.iterator.map(_.sideHash)
-    )
+    val sideHashes = MurmurHash3.orderedHash(group.iterator.map(_.sideHash))
 
     val scriptsHash = group
       .iterator
@@ -162,7 +131,8 @@ private[mill] trait GroupEvaluator {
           counterMsg = counterMsg,
           zincProblemReporter,
           testReporter,
-          logger
+          logger,
+          terminal.task.asWorker.nonEmpty
         )
         GroupEvaluator.Results(newResults, newEvaluated.toSeq, null, inputsHash, -1)
 
@@ -211,7 +181,8 @@ private[mill] trait GroupEvaluator {
                   counterMsg = counterMsg,
                   zincProblemReporter,
                   testReporter,
-                  logger
+                  logger,
+                  terminal.task.asWorker.nonEmpty
                 )
               }
 
@@ -251,7 +222,8 @@ private[mill] trait GroupEvaluator {
       counterMsg: String,
       reporter: Int => Option[CompileProblemReporter],
       testReporter: TestReporter,
-      logger: mill.api.Logger
+      logger: mill.api.Logger,
+      isWorker: Boolean
   ): (Map[Task[_], TaskResult[(Val, Int)]], mutable.Buffer[Task[_]]) = {
 
     def computeAll(enableTicker: Boolean) = {
@@ -283,32 +255,31 @@ private[mill] trait GroupEvaluator {
 
         override def rawOutputStream: PrintStream = logger.rawOutputStream
       }
-      // This is used to track the usage of `T.dest` in more than one Task
-      // But it's not really clear what issue we try to prevent here
-      // Vice versa, being able to use T.dest in multiple `T.task`
-      // is rather essential to split up larger tasks into small parts
-      // So I like to disable this detection for now
-      var usedDest = Option.empty[(Task[_], Array[StackTraceElement])]
+
+      var usedDest = Option.empty[os.Path]
       for (task <- nonEvaluatedTargets) {
         newEvaluated.append(task)
         val targetInputValues = task.inputs
           .map { x => newResults.getOrElse(x, results(x).result) }
           .collect { case Result.Success((v, _)) => v }
 
+        def makeDest() = this.synchronized {
+          paths match {
+            case Some(dest) =>
+              if (usedDest.isEmpty) os.makeDir.all(dest.dest)
+              usedDest = Some(dest.dest)
+              dest.dest
+
+            case None => throw new Exception("No `dest` folder available here")
+          }
+        }
+
         val res = {
           if (targetInputValues.length != task.inputs.length) Result.Skipped
           else {
             val args = new mill.api.Ctx(
               args = targetInputValues.map(_.value).toIndexedSeq,
-              dest0 = () =>
-                paths match {
-                  case Some(dest) =>
-                    if (usedDest.isEmpty) os.makeDir.all(dest.dest)
-                    usedDest = Some((task, new Exception().getStackTrace))
-                    dest.dest
-
-                  case None => throw new Exception("No `dest` folder available here")
-                },
+              dest0 = () => makeDest(),
               log = multiLogger,
               home = home,
               env = env,
@@ -320,15 +291,17 @@ private[mill] trait GroupEvaluator {
               override def jobs: Int = effectiveThreadCount
             }
 
-            mill.api.SystemStreams.withStreams(multiLogger.systemStreams) {
-              try task.evaluate(args).map(Val(_))
-              catch {
-                case f: Result.Failing[Val] => f
-                case NonFatal(e) =>
-                  Result.Exception(
-                    e,
-                    new OuterStack(new Exception().getStackTrace.toIndexedSeq)
-                  )
+            os.dynamicPwdFunction.withValue(() => makeDest()) {
+              mill.api.SystemStreams.withStreams(multiLogger.systemStreams) {
+                try task.evaluate(args).map(Val(_))
+                catch {
+                  case f: Result.Failing[Val] => f
+                  case NonFatal(e) =>
+                    Result.Exception(
+                      e,
+                      new OuterStack(new Exception().getStackTrace.toIndexedSeq)
+                    )
+                }
               }
             }
           }
@@ -342,6 +315,7 @@ private[mill] trait GroupEvaluator {
           )
         }
       }
+
       multiLogger.close()
       (newResults, newEvaluated)
     }
@@ -351,7 +325,7 @@ private[mill] trait GroupEvaluator {
     if (!failFast) maybeTargetLabel.foreach { targetLabel =>
       val taskFailed = newResults.exists(task => !task._2.isInstanceOf[Success[_]])
       if (taskFailed) {
-        logger.error(s"[${counterMsg}] ${targetLabel} failed")
+        logger.error(s"[$counterMsg] $targetLabel failed")
       }
     }
 
diff --git a/main/eval/src/mill/eval/ThreadNumberer.scala b/main/eval/src/mill/eval/ThreadNumberer.scala
index 1f0c66e641e..de95f722c49 100644
--- a/main/eval/src/mill/eval/ThreadNumberer.scala
+++ b/main/eval/src/mill/eval/ThreadNumberer.scala
@@ -7,6 +7,7 @@ private class ThreadNumberer() {
   private val threadIds = collection.mutable.Map.empty[Thread, Int]
 
   def getThreadId(thread: Thread): Int = synchronized {
-    threadIds.getOrElseUpdate(thread, threadIds.size)
+    // start thread IDs from 1 so they're easier to count
+    threadIds.getOrElseUpdate(thread, threadIds.size + 1)
   }
 }
diff --git a/main/src/mill/main/MainModule.scala b/main/src/mill/main/MainModule.scala
index 27e2a7060c4..54421527574 100644
--- a/main/src/mill/main/MainModule.scala
+++ b/main/src/mill/main/MainModule.scala
@@ -317,12 +317,6 @@ trait MainModule extends BaseModule0 {
    * will clean everything.
    */
   def clean(evaluator: Evaluator, targets: String*): Command[Seq[PathRef]] = Target.command {
-    if (evaluator.effectiveThreadCount > 1) {
-      evaluator.baseLogger.error(
-        "The clean target in parallel mode might result in unexpected effects"
-      )
-    }
-
     val rootDir = evaluator.outPath
 
     val KeepPattern = "(mill-.+)".r.anchored
diff --git a/main/util/src/mill/util/Jvm.scala b/main/util/src/mill/util/Jvm.scala
index 78417d35ed7..b026adabed8 100644
--- a/main/util/src/mill/util/Jvm.scala
+++ b/main/util/src/mill/util/Jvm.scala
@@ -83,7 +83,8 @@ object Jvm extends CoursierSupport {
       mainArgs: Seq[String] = Seq.empty,
       workingDir: os.Path = null,
       background: Boolean = false,
-      useCpPassingJar: Boolean = false
+      useCpPassingJar: Boolean = false,
+      runBackgroundLogToConsole: Boolean = false
   )(implicit ctx: Ctx): Unit = {
     runSubprocessWithBackgroundOutputs(
       mainClass,
@@ -92,11 +93,35 @@ object Jvm extends CoursierSupport {
       envArgs,
       mainArgs,
       workingDir,
-      if (background) defaultBackgroundOutputs(ctx.dest) else None,
+      if (!background) None
+      else if (runBackgroundLogToConsole) Some((os.Inherit, os.Inherit))
+      else Jvm.defaultBackgroundOutputs(ctx.dest),
       useCpPassingJar
     )
   }
 
+  // bincompat shim
+  def runSubprocess(
+      mainClass: String,
+      classPath: Agg[os.Path],
+      jvmArgs: Seq[String],
+      envArgs: Map[String, String],
+      mainArgs: Seq[String],
+      workingDir: os.Path,
+      background: Boolean,
+      useCpPassingJar: Boolean
+  )(implicit ctx: Ctx): Unit =
+    runSubprocess(
+      mainClass,
+      classPath,
+      jvmArgs,
+      envArgs,
+      mainArgs,
+      workingDir,
+      background,
+      useCpPassingJar
+    )
+
   /**
    * Runs a JVM subprocess with the given configuration and streams
    * it's stdout and stderr to the console.
diff --git a/scalalib/src/mill/scalalib/JavaModule.scala b/scalalib/src/mill/scalalib/JavaModule.scala
index 508793a2664..184a79ad2f2 100644
--- a/scalalib/src/mill/scalalib/JavaModule.scala
+++ b/scalalib/src/mill/scalalib/JavaModule.scala
@@ -907,6 +907,7 @@ trait JavaModule
     super.run(args)
   }
 
+  @deprecated("Binary compat shim, use `.runner().run(..., background=true)`", "Mill 0.12.0")
   override protected def doRunBackground(
       taskDest: Path,
       runClasspath: Seq[PathRef],
diff --git a/scalalib/src/mill/scalalib/RunModule.scala b/scalalib/src/mill/scalalib/RunModule.scala
index df07b386dcd..c1a69970bd1 100644
--- a/scalalib/src/mill/scalalib/RunModule.scala
+++ b/scalalib/src/mill/scalalib/RunModule.scala
@@ -4,7 +4,6 @@ import mainargs.arg
 import mill.api.JsonFormatters.pathReadWrite
 import mill.api.{Ctx, PathRef, Result}
 import mill.define.{Command, Task}
-import mill.main.client.EnvVars
 import mill.util.Jvm
 import mill.{Agg, Args, T}
 import os.{Path, ProcessOutput}
@@ -123,21 +122,23 @@ trait RunModule extends WithZincWorker {
   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()
-          )
+          runner().run(args = args().value, mainClass = mainClass(), workingDir = forkWorkingDir())
         )
       catch {
         case NonFatal(_) => Result.Failure("Subprocess failed")
       }
     }
 
+  def runner: Task[RunModule.Runner] = T.task {
+    new RunModule.RunnerImpl(
+      finalMainClassOpt(),
+      runClasspath().map(_.path),
+      forkArgs(),
+      forkEnv(),
+      runUseArgsFile()
+    )
+  }
+
   def runLocalTask(mainClass: Task[String], args: Task[Args] = T.task(Args())): Task[Unit] =
     T.task {
       Jvm.runLocal(
@@ -149,24 +150,15 @@ trait RunModule extends WithZincWorker {
 
   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())
-
-      // Make sure to sleep a bit in the Mill test suite to allow the servers we
-      // start time to initialize before we proceed with the following commands
-      if (T.env.contains(EnvVars.MILL_TEST_SUITE)) {
-        println("runBackgroundTask SLEEPING 10000")
-        Thread.sleep(5000)
-      }
+      val (procId, procTombstone, token) = backgroundSetup(T.dest)
+      runner().run(
+        args = Seq(procId.toString, procTombstone.toString, token, mainClass()) ++ args().value,
+        mainClass = "mill.scalalib.backgroundwrapper.BackgroundWrapper",
+        workingDir = forkWorkingDir(),
+        extraRunClasspath = zincWorker().backgroundWrapperClasspath().map(_.path).toSeq,
+        background = true,
+        runBackgroundLogToConsole = runBackgroundLogToConsole
+      )
     }
 
   /**
@@ -180,11 +172,7 @@ trait RunModule extends WithZincWorker {
   // 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)
-  }
-
+  @deprecated("Binary compat shim, use `.runner().run(..., background=true)`", "Mill 0.12.0")
   protected def doRunBackground(
       taskDest: Path,
       runClasspath: Seq[PathRef],
@@ -249,3 +237,54 @@ trait RunModule extends WithZincWorker {
   }
 
 }
+
+object RunModule {
+  trait Runner {
+    def run(
+        args: os.Shellable,
+        mainClass: String = null,
+        forkArgs: Seq[String] = null,
+        forkEnv: Map[String, String] = null,
+        workingDir: os.Path = null,
+        useCpPassingJar: java.lang.Boolean = null,
+        extraRunClasspath: Seq[os.Path] = Nil,
+        background: Boolean = false,
+        runBackgroundLogToConsole: Boolean = false
+    )(implicit ctx: Ctx): Unit
+  }
+  private class RunnerImpl(
+      mainClass0: Either[String, String],
+      runClasspath: Seq[os.Path],
+      forkArgs0: Seq[String],
+      forkEnv0: Map[String, String],
+      useCpPassingJar0: Boolean
+  ) extends Runner {
+
+    def run(
+        args: os.Shellable,
+        mainClass: String = null,
+        forkArgs: Seq[String] = null,
+        forkEnv: Map[String, String] = null,
+        workingDir: os.Path = null,
+        useCpPassingJar: java.lang.Boolean = null,
+        extraRunClasspath: Seq[os.Path] = Nil,
+        background: Boolean = false,
+        runBackgroundLogToConsole: Boolean = false
+    )(implicit ctx: Ctx): Unit = {
+      Jvm.runSubprocess(
+        Option(mainClass).getOrElse(mainClass0.fold(sys.error, identity)),
+        runClasspath ++ extraRunClasspath,
+        Option(forkArgs).getOrElse(forkArgs0),
+        Option(forkEnv).getOrElse(forkEnv0),
+        args.value,
+        Option(workingDir).getOrElse(ctx.dest),
+        background = background,
+        Option(useCpPassingJar) match {
+          case Some(b) => b
+          case None => useCpPassingJar0
+        },
+        runBackgroundLogToConsole = runBackgroundLogToConsole
+      )
+    }
+  }
+}