diff --git a/build.sbt b/build.sbt
index b278b9114..4d07252d5 100644
--- a/build.sbt
+++ b/build.sbt
@@ -7,6 +7,7 @@ import java.util.Locale
 import sbt.internal.inc.Analysis
 import sbt.Tags
 import com.eed3si9n.jarjarabrams.ModuleCoordinate
+import Utils.JDK17
 
 // ThisBuild settings take lower precedence,
 // but can be shared across the multi projects.
@@ -466,16 +467,30 @@ lazy val testingProj = (project in file("testing"))
 
 lazy val workerProj = (project in file("worker"))
   .dependsOn(exampleWorkProj % Test)
+  .configs(JDK17)
   .settings(
     name := "worker",
+    inConfig(JDK17)(Defaults.compileSettings),
+    exportJars := true,
     testedBaseSettings,
     Compile / doc / javacOptions := Nil,
     crossPaths := false,
     autoScalaLibrary := false,
     libraryDependencies ++= Seq(gson, testInterface),
     libraryDependencies += "org.scala-lang" %% "scala3-library" % scalaVersion.value % Test,
-    // run / fork := false,
     Test / fork := true,
+    Compile / javacOptions := Seq("--release", "8"),
+    JDK17 / javacOptions := Seq("--release", "17"),
+    Compile / packageBin / packageOptions += Package.ManifestAttributes("Multi-Release" -> "true"),
+    Compile / packageBin / mappings ++= {
+      val _ = (JDK17 / compile).value
+      val dir = (Utils.JDK17 / classDirectory).value
+      fileTreeView.value
+        .list(Glob(dir) / **)
+        .map(_._1)
+        .map(_.toFile())
+        .pair(Path.rebase(dir, "META-INF/versions/17"))
+    },
     mimaSettings,
     mimaBinaryIssueFilters ++= Vector(
     ),
@@ -600,6 +615,11 @@ lazy val actionsProj = (project in file("main-actions"))
     Test / classLoaderLayeringStrategy := ClassLoaderLayeringStrategy.Flat,
     mimaSettings,
     mimaBinaryIssueFilters ++= Vector(
+      // WorkerConnection gained an Ipc(path) case; mixing a parameterized case into
+      // the enum drops the synthetic values()/valueOf() Java-enum forwarders. This is
+      // an internal (sbt.internal) type not meant for external consumption.
+      exclude[DirectMissingMethodProblem]("sbt.internal.WorkerConnection.valueOf"),
+      exclude[DirectMissingMethodProblem]("sbt.internal.WorkerConnection.values"),
     ),
   )
   .dependsOn(lmCore)
diff --git a/launcher-package/citest/build.sbt b/launcher-package/citest/build.sbt
index 9afef5b42..12f5f108b 100644
--- a/launcher-package/citest/build.sbt
+++ b/launcher-package/citest/build.sbt
@@ -1,5 +1,8 @@
 lazy val check = taskKey[Unit]("")
 lazy val check2 = taskKey[Unit]("")
+lazy val checkEvalArgHandling = taskKey[Unit]("")
+lazy val checkDArgHandling = taskKey[Unit]("")
+lazy val checkXXArgHandling = taskKey[Unit]("")
 
 lazy val root = (project in file("."))
   .settings(
@@ -18,5 +21,47 @@ lazy val root = (project in file("."))
       val ys = IO.readLines(file("err.txt")).toVector.distinct
 
       assert(ys.isEmpty, s"there's an stderr: $ys")
+    },
+    // Regression check for https://github.com/sbt/sbt/issues/9660, run as a
+    // real .bat command by test.bat rather than via a JVM-constructed command
+    // line, so cmd.exe parses the argument the same way it would if a person
+    // had typed it at a prompt.
+    checkEvalArgHandling := {
+      val evalOut = IO.readLines(file("evalOutput.txt")).toVector
+      println(evalOut)
+      assert(
+        evalOut.exists(_.contains("barqux")),
+        s"""eval ("bar") ++ ("qux") should print barqux, but got: $evalOut"""
+      )
+      assert(
+        !evalOut.exists(l =>
+          l.contains("was unexpected at this time") || l.contains("is not recognized")
+        ),
+        s"eval with quoted parens should not trigger a cmd.exe parse error: $evalOut"
+      )
+
+      assert(
+        !file("injected.txt").exists,
+        "the & in the eval argument must not escape sbt.bat's quoting and run as a separate command"
+      )
+    },
+    // Regression check for https://github.com/sbt/sbt/issues/9660, matching a
+    // reporter-confirmed case: "sbt" "-Dfoo=()&calc" pops calc when typed at a
+    // real cmd.exe prompt, even though the whole value sits inside one clean,
+    // matched pair of quotes (no embedded quote characters to lose parity).
+    checkDArgHandling := {
+      assert(
+        !file("injected2.txt").exists,
+        "the & in a -D argument value must not escape sbt.bat's quoting and run as a separate command"
+      )
+    },
+    // Regression check for https://github.com/sbt/sbt/issues/9660: the -XX
+    // handling in args_loop has the same shape as -D, so the same reporter-
+    // confirmed bypass applies to it.
+    checkXXArgHandling := {
+      assert(
+        !file("injected3.txt").exists,
+        "the & in a -XX argument value must not escape sbt.bat's quoting and run as a separate command"
+      )
     }
   )
diff --git a/launcher-package/citest/test.bat b/launcher-package/citest/test.bat
index c35ddae49..608abff66 100755
--- a/launcher-package/citest/test.bat
+++ b/launcher-package/citest/test.bat
@@ -28,4 +28,24 @@ SET SBT_OPTS=-Xmx4g -Dfile.encoding=UTF8
 "freshly-baked\sbt\bin\sbt" -Dsbt.no.format=true --version > version.txt
 "freshly-baked\sbt\bin\sbt" -Dsbt.no.format=true checkVersion
 
+rem Regression test for https://github.com/sbt/sbt/issues/9660, run as a real
+rem .bat command (parsed by cmd.exe the same way as if typed at a prompt),
+rem rather than via a JVM-constructed command line.
+"freshly-baked\sbt\bin\sbt" -Dsbt.no.format=true "eval (\"bar\") ++ (\"qux\")" 1> evalOutput.txt 2> evalErr.txt
+
+"freshly-baked\sbt\bin\sbt" -Dsbt.no.format=true "eval (\"foo\") & echo INJECTED>injected.txt & rem (" 1> evalInjOutput.txt 2> evalInjErr.txt
+
+"freshly-baked\sbt\bin\sbt" -Dsbt.no.format=true checkEvalArgHandling
+
+rem "about" comes before the risky argument so sbt.bat always receives a real
+rem command and cannot fall into an interactive shell (which would hang this
+rem script) even if the & below does escape its quoting.
+"freshly-baked\sbt\bin\sbt" -Dsbt.no.format=true about "-Dfoo=()&copy nul injected2.txt"
+
+"freshly-baked\sbt\bin\sbt" -Dsbt.no.format=true checkDArgHandling
+
+"freshly-baked\sbt\bin\sbt" -Dsbt.no.format=true about "-XXbar=()&copy nul injected3.txt"
+
+"freshly-baked\sbt\bin\sbt" -Dsbt.no.format=true checkXXArgHandling
+
 ENDLOCAL
diff --git a/launcher-package/integration-test/src/test/scala/RunnerMemoryScriptTest.scala b/launcher-package/integration-test/src/test/scala/RunnerMemoryScriptTest.scala
index 7f2c3ffd0..1a68cae65 100644
--- a/launcher-package/integration-test/src/test/scala/RunnerMemoryScriptTest.scala
+++ b/launcher-package/integration-test/src/test/scala/RunnerMemoryScriptTest.scala
@@ -75,12 +75,14 @@ object RunnerMemoryScriptTest extends verify.BasicTestSuite with ShellScriptUtil
     assert(out.contains[String]("-Xss12m"))
 
   // Test for issue #5742: -X options passed directly on command line
+  // Note: the -v preview quotes these (see #9660), so match by substring rather
+  // than exact line equality.
   testOutput("sbt -Xmx1G directly on command line")("-Xmx1G", "-v"): (out: List[String]) =>
-    assert(out.contains[String]("-Xmx1G"))
+    assert(out.exists(_.contains("-Xmx1G")))
 
   testOutput("sbt -Xms512M -Xmx1G directly on command line")("-Xms512M", "-Xmx1G", "-v"):
     (out: List[String]) =>
-      assert(out.contains[String]("-Xms512M"))
-      assert(out.contains[String]("-Xmx1G"))
+      assert(out.exists(_.contains("-Xms512M")))
+      assert(out.exists(_.contains("-Xmx1G")))
 
 end RunnerMemoryScriptTest
diff --git a/launcher-package/integration-test/src/test/scala/RunnerScriptTest.scala b/launcher-package/integration-test/src/test/scala/RunnerScriptTest.scala
index ad5774e77..ebf8af331 100644
--- a/launcher-package/integration-test/src/test/scala/RunnerScriptTest.scala
+++ b/launcher-package/integration-test/src/test/scala/RunnerScriptTest.scala
@@ -24,7 +24,8 @@ object RunnerScriptTest extends verify.BasicTestSuite with ShellScriptUtil:
     assert(out.contains[String]("-Dsbt.log.noformat=true"))
 
   testOutput("sbt --color=false")("compile", "--color=false", "-v"): (out: List[String]) =>
-    assert(out.contains[String]("-Dsbt.color=false"))
+    // Note: the -v preview quotes this (see #9660), so match by substring.
+    assert(out.exists(_.contains("-Dsbt.color=false")))
 
   testOutput("sbt --no-colors in SBT_OPTS", sbtOpts = "--no-colors")("compile", "-v"):
     (out: List[String]) =>
@@ -38,16 +39,21 @@ object RunnerScriptTest extends verify.BasicTestSuite with ShellScriptUtil:
     assert(out.contains[String]("-Dxsbt.inc.debug=true"))
 
   testOutput("sbt --supershell=never")("compile", "--supershell=never", "-v"):
-    (out: List[String]) => assert(out.contains[String]("-Dsbt.supershell=never"))
+    (out: List[String]) =>
+      // Note: the -v preview quotes this (see #9660), so match by substring.
+      assert(out.exists(_.contains("-Dsbt.supershell=never")))
 
   testOutput("sbt --timings")("compile", "--timings", "-v"): (out: List[String]) =>
     assert(out.contains[String]("-Dsbt.task.timings=true"))
 
   testOutput("sbt -D arguments")("-Dsbt.supershell=false", "compile", "-v"): (out: List[String]) =>
-    assert(out.contains[String]("-Dsbt.supershell=false"))
+    // Note: the -v preview quotes CLI -D arguments (see #9660), so match by
+    // substring rather than exact line equality.
+    assert(out.exists(_.contains("-Dsbt.supershell=false")))
 
   testOutput("sbt --sbt-version")("--sbt-version", "1.3.13", "-v"): (out: List[String]) =>
-    assert(out.contains[String]("-Dsbt.version=1.3.13"))
+    // Note: the -v preview quotes this (see #9660), so match by substring.
+    assert(out.exists(_.contains("-Dsbt.version=1.3.13")))
 
   testOutput(
     name = "sbt with -Dhttp.proxyHost=proxy -Dhttp.proxyPort=8080 in SBT_OPTS",
@@ -365,6 +371,12 @@ object RunnerScriptTest extends verify.BasicTestSuite with ShellScriptUtil:
       s"Should not have shell expansion errors, but found: ${errorMessages.mkString(", ")}"
     )
 
+  // NOTE on https://github.com/sbt/sbt/issues/9660: an argument with an
+  // unquoted &, (, or ) can be split apart by cmd.exe's parse of the command
+  // line used to *invoke* sbt.bat, before any line of sbt.bat runs -- no code
+  // inside the script can intervene there. Closing that would require a real
+  // executable entry point instead of a .bat file (see the sbtw project).
+
   // Test for issue #8755: Inline comments should be supported in .jvmopts
   testOutput(
     "sbt with inline comments in .jvmopts",
@@ -411,6 +423,8 @@ object RunnerScriptTest extends verify.BasicTestSuite with ShellScriptUtil:
     )
 
   testOutput("sbt --experimental_execution_log=true")("--experimental_execution_log=true", "-v"):
-    (out: List[String]) => assert(out.contains[String]("-Dsbt.experimental_execution_log=true"))
+    (out: List[String]) =>
+      // Note: the -v preview quotes this (see #9660), so match by substring.
+      assert(out.exists(_.contains("-Dsbt.experimental_execution_log=true")))
 
 end RunnerScriptTest
diff --git a/launcher-package/src/universal/bin/sbt.bat b/launcher-package/src/universal/bin/sbt.bat
index 9c5fe2252..68f8d26cb 100755
--- a/launcher-package/src/universal/bin/sbt.bat
+++ b/launcher-package/src/universal/bin/sbt.bat
@@ -537,71 +537,75 @@ if "%~0" == "init" (
   )
 )
 
-if "%g:~0,2%" == "-D" (
-  rem special handling for -D since '=' gets parsed away
-  for /F "tokens=1 delims==" %%a in ("%g%") do (
-    rem make sure it doesn't have the '=' already
-    if "%g%" == "%%a" (
-      if not "%~1" == "" (
-        call :dlog [args_loop] -D argument %~0=%~1
-        set "SBT_ARGS=!SBT_ARGS! %~0=%~1"
-        shift
-        goto args_loop
-      ) else (
-        echo %g% is missing a value
-        goto error
-      )
-    ) else (
-      call :dlog [args_loop] -D argument %~0
-      set "SBT_ARGS=!SBT_ARGS! %~0"
-      goto args_loop
-    )
-  )
-)
+rem NOTE: -D handling below is intentionally flat (goto-based, single-line
+rem `for` bodies) rather than nested if/for blocks. cmd.exe finds a multi-line
+rem block's closing paren by counting every literal ( and ) across the lines
+rem it spans, including ones arriving via variable substitution -- splicing an
+rem argument value (which may itself contain unbalanced-looking parens, see
+rem #9660) into a deeply nested multi-line construct risks miscounting that.
+rem A flat, single-line-per-command shape does not put cmd.exe in a position
+rem where it needs to hunt across multiple lines for a matching paren.
+if not "%g:~0,2%" == "-D" goto args_loop_after_D
+rem special handling for -D since '=' gets parsed away
+set "g_key="
+for /F "tokens=1 delims==" %%a in ("%g%") do set "g_key=%%a"
+if not "%g%" == "%g_key%" goto args_loop_D_has_value
+if "%~1" == "" goto args_loop_D_missing_value
+call :dlog [args_loop] -D argument %~0=%~1
+set SBT_ARGS=!SBT_ARGS! "%~0=%~1"
+shift
+goto args_loop
+:args_loop_D_missing_value
+echo %g% is missing a value
+goto error
+:args_loop_D_has_value
+call :dlog [args_loop] -D argument %~0
+set SBT_ARGS=!SBT_ARGS! "%~0"
+goto args_loop
+:args_loop_after_D
 
-if not "%g:~0,5%" == "-XX:+" if not "%g:~0,5%" == "-XX:-" if "%g:~0,3%" == "-XX" (
-  rem special handling for -XX since '=' gets parsed away
-  for /F "tokens=1 delims==" %%a in ("%g%") do (
-    rem make sure it doesn't have the '=' already
-    if "%g%" == "%%a" (
-      if not "%~1" == "" (
-        call :dlog [args_loop] -XX argument %~0=%~1
-        set "SBT_ARGS=!SBT_ARGS! %~0=%~1"
-        shift
-        goto args_loop
-      ) else (
-        echo %g% is missing a value
-        goto error
-      )
-    ) else (
-      call :dlog [args_loop] -XX argument %~0
-      set "SBT_ARGS=!SBT_ARGS! %~0"
-      goto args_loop
-    )
-  )
-)
+rem See the -D handling above for why this is flat rather than nested blocks.
+if "%g:~0,5%" == "-XX:+" goto args_loop_after_XX
+if "%g:~0,5%" == "-XX:-" goto args_loop_after_XX
+if not "%g:~0,3%" == "-XX" goto args_loop_after_XX
+rem special handling for -XX since '=' gets parsed away
+set "g_key="
+for /F "tokens=1 delims==" %%a in ("%g%") do set "g_key=%%a"
+if not "%g%" == "%g_key%" goto args_loop_XX_has_value
+if "%~1" == "" goto args_loop_XX_missing_value
+call :dlog [args_loop] -XX argument %~0=%~1
+set SBT_ARGS=!SBT_ARGS! "%~0=%~1"
+shift
+goto args_loop
+:args_loop_XX_missing_value
+echo %g% is missing a value
+goto error
+:args_loop_XX_has_value
+call :dlog [args_loop] -XX argument %~0
+set SBT_ARGS=!SBT_ARGS! "%~0"
+goto args_loop
+:args_loop_after_XX
 
 rem handle -X JVM options (e.g., -Xmx1G, -Xms512M, -Xss4M) - fixes #5742
 if "%g:~0,2%" == "-X" (
   call :dlog [args_loop] -X JVM argument %~0
-  call :addJava %~0
+  call :addJava "%~0"
   goto args_loop
 )
 
-if defined sbt_new if "%g:~0,2%" == "--" (
-  rem special handling for -- template arguments since '=' gets parsed away on Windows
-  for /F "tokens=1 delims==" %%a in ("%g%") do (
-    rem make sure it doesn't have the '=' already
-    if "%g%" == "%%a" (
-      if not "%~1" == "" (
-        call :dlog [args_loop] -- argument %~0=%~1
-        set "SBT_ARGS=!SBT_ARGS! %~0=%~1"
-        shift
-        goto args_loop
-      )
-    )
-  )
-)
+rem See the -D handling above for why this is flat rather than nested blocks.
+if not defined sbt_new goto args_loop_after_dashdash
+if not "%g:~0,2%" == "--" goto args_loop_after_dashdash
+rem special handling for -- template arguments since '=' gets parsed away on Windows
+set "g_key="
+for /F "tokens=1 delims==" %%a in ("%g%") do set "g_key=%%a"
+if not "%g%" == "%g_key%" goto args_loop_after_dashdash
+if "%~1" == "" goto args_loop_after_dashdash
+call :dlog [args_loop] -- argument %~0=%~1
+set SBT_ARGS=!SBT_ARGS! "%~0=%~1"
+shift
+goto args_loop
+:args_loop_after_dashdash
 
 rem the %0 (instead of %~0) preserves original argument quoting
 set sbt_args_seen_command=1
@@ -655,7 +659,11 @@ if !sbt_args_print_sbt_script_version! equ 1 (
 call :checkjava
 
 if !run_native_client! equ 1 if not defined sbt_args_print_version (
-  goto :runnative !SBT_ARGS!
+  rem Do not append !SBT_ARGS! here: `goto` ignores it, it is not a real
+  rem argument-passing mechanism, but the text is still spliced into this
+  rem line and re-scanned by cmd.exe, so it is live to shell operators ^(#9660^).
+  rem :runnative reads SBT_ARGS via delayed expansion instead.
+  goto :runnative
   goto :eof
 )
 
@@ -675,7 +683,11 @@ if defined JVM_DEBUG_PORT (
 
 call :sync_preloaded
 
-call :run !SBT_ARGS!
+rem Do not append !SBT_ARGS! to this call: splicing it onto a `call` command
+rem line makes cmd.exe re-tokenize it, letting shell operators inside a
+rem quoted user argument escape their quoting and run ^(#9660^).
+rem :run reads SBT_ARGS via delayed expansion instead.
+call :run
 
 if ERRORLEVEL 1 goto error
 goto end
@@ -700,34 +712,38 @@ if defined sbt_args_no_share (
   set _SBT_OPTS=-Dsbt.global.base=project/.sbtboot -Dsbt.boot.directory=project/.boot -Dsbt.ivy.home=project/.ivy !_SBT_OPTS!
 )
 
+rem NOTE: each user-supplied value below is wrapped in its own quotes before
+rem being appended to _SBT_OPTS. _SBT_OPTS is later spliced unquoted into
+rem exec lines ^(:copyrt, the final java invocation^), so an unquoted shell
+rem operator here would be live to cmd.exe at that point ^(#9660^).
 if defined sbt_args_supershell (
-  set _SBT_OPTS=-Dsbt.supershell=!sbt_args_supershell! !_SBT_OPTS!
+  set _SBT_OPTS="-Dsbt.supershell=!sbt_args_supershell!" !_SBT_OPTS!
 )
 
 if defined sbt_args_sbt_version (
-  set _SBT_OPTS=-Dsbt.version=!sbt_args_sbt_version! !_SBT_OPTS!
+  set _SBT_OPTS="-Dsbt.version=!sbt_args_sbt_version!" !_SBT_OPTS!
 )
 
 if defined sbt_args_sbt_dir (
-  set _SBT_OPTS=-Dsbt.global.base=!sbt_args_sbt_dir! !_SBT_OPTS!
+  set _SBT_OPTS="-Dsbt.global.base=!sbt_args_sbt_dir!" !_SBT_OPTS!
 ) else if defined LOCALAPPDATA (
-  set _SBT_OPTS=-Dsbt.global.base=!LOCALAPPDATA!\sbt !_SBT_OPTS!
+  set _SBT_OPTS="-Dsbt.global.base=!LOCALAPPDATA!\sbt" !_SBT_OPTS!
 )
 
 if defined sbt_args_sbt_boot (
-  set _SBT_OPTS=-Dsbt.boot.directory=!sbt_args_sbt_boot! !_SBT_OPTS!
+  set _SBT_OPTS="-Dsbt.boot.directory=!sbt_args_sbt_boot!" !_SBT_OPTS!
 )
 
 if defined sbt_args_sbt_cache (
-  set _SBT_OPTS=-Dsbt.global.localcache=!sbt_args_sbt_cache! !_SBT_OPTS!
+  set _SBT_OPTS="-Dsbt.global.localcache=!sbt_args_sbt_cache!" !_SBT_OPTS!
 )
 
 if defined sbt_args_ivy (
-  set _SBT_OPTS=-Dsbt.ivy.home=!sbt_args_ivy! !_SBT_OPTS!
+  set _SBT_OPTS="-Dsbt.ivy.home=!sbt_args_ivy!" !_SBT_OPTS!
 )
 
 if defined sbt_args_color (
-  set _SBT_OPTS=-Dsbt.color=!sbt_args_color! !_SBT_OPTS!
+  set _SBT_OPTS="-Dsbt.color=!sbt_args_color!" !_SBT_OPTS!
 )
 
 if defined sbt_args_mem (
@@ -755,7 +771,7 @@ if not defined sbt_args_no_hide_jdk_warnings (
 )
 
 if defined sbt_args_experimental_execution_log (
-  set _SBT_OPTS=-Dsbt.experimental_execution_log=!sbt_args_experimental_execution_log! !_SBT_OPTS!
+  set _SBT_OPTS="-Dsbt.experimental_execution_log=!sbt_args_experimental_execution_log!" !_SBT_OPTS!
 )
 
 rem TODO: _SBT_OPTS needs to be processed as args and diffed against SBT_ARGS
@@ -792,11 +808,14 @@ if defined sbt_args_verbose (
   echo -cp
   echo "!sbt_jar!"
   echo xsbt.boot.Boot
-  if not "%~1" == "" ( call :echolist %* )
+  if defined SBT_ARGS ( call :echolist !SBT_ARGS! )
   echo.
 )
 
-"!_JAVACMD!" !_JAVA_OPTS! !_SBT_OPTS! %JAVA_TOOL_OPTIONS% %JDK_JAVA_OPTIONS% -cp "!sbt_jar!" xsbt.boot.Boot %*
+rem one-hop marker: drop it before the server JVM
+set "SBT_EXPLICIT_JAVA_HOME="
+
+"!_JAVACMD!" !_JAVA_OPTS! !_SBT_OPTS! %JAVA_TOOL_OPTIONS% %JDK_JAVA_OPTIONS% -cp "!sbt_jar!" xsbt.boot.Boot !SBT_ARGS!
 
 goto :eof
 
@@ -804,14 +823,19 @@ goto :eof
 
 set "_SBTNCMD=!SBT_BIN_DIR!sbtn-x86_64-pc-win32.exe"
 
+rem NOTE: SBT_ARGS entries may carry their own embedded quotes, see the
+rem args_loop -D/-XX/catch-all handling above, to keep shell operators safe
+rem across the re-scan below. Do not wrap these set lines in an outer pair
+rem of quotes: that would pair up with an embedded quote instead of its
+rem match, breaking the quoting it is meant to preserve ^(#9660^).
 if defined sbt_args_verbose (
   echo # running native client
   if not "%~1" == "" ( call :echolist %* )
-  set "SBT_ARGS=-v !SBT_ARGS!"
+  set SBT_ARGS=-v !SBT_ARGS!
 )
 
 for %%I in ("!SBT_BIN_DIR!sbt.bat") do set "SBT_SCRIPT=%%~sI"
-set "SBT_ARGS=--sbt-script=!SBT_SCRIPT! %SBT_ARGS%"
+set SBT_ARGS=--sbt-script=!SBT_SCRIPT! %SBT_ARGS%
 
 rem Microsoft Visual C++ 2010 SP1 Redistributable Package (x64) is required
 rem https://www.microsoft.com/en-us/download/details.aspx?id=13523
@@ -885,7 +909,7 @@ exit /B 0
 
 :addJava
   call :dlog [addJava] arg = '%*'
-  set "_JAVA_OPTS=!_JAVA_OPTS! %*"
+  set _JAVA_OPTS=!_JAVA_OPTS! %*
 exit /B 0
 
 :addMemory
diff --git a/main-actions/src/main/scala/sbt/ForkTests.scala b/main-actions/src/main/scala/sbt/ForkTests.scala
index b18ff55f4..510450eee 100755
--- a/main-actions/src/main/scala/sbt/ForkTests.scala
+++ b/main-actions/src/main/scala/sbt/ForkTests.scala
@@ -28,7 +28,6 @@ import scala.util.Random
 import scala.util.control.NonFatal
 import scala.jdk.CollectionConverters.*
 import scala.sys.process.Process
-import sbt.internal.WorkerConnection
 
 /**
  * This implements forked testing, in cooperation with the worker CLI,
@@ -141,8 +140,7 @@ private[sbt] object ForkTests:
       )
       testListeners.foreach(_.doInit())
       val result =
-        val ct = WorkerConnection.Tcp
-        val w = WorkerExchange.startWorker(fork, if virtualClasspath then Nil else cpFiles, ct)
+        val w = WorkerExchange.startWorker(fork, if virtualClasspath then Nil else cpFiles)
         val wl = React(randomId, log, opts.testListeners, resultsAcc, w.process)
         try
           WorkerExchange.registerListener(wl)
@@ -152,7 +150,9 @@ private[sbt] object ForkTests:
           if wl.blockForResponse() != 0 then
             throw MessageOnlyException("Forked test harness failed")
           testOutputResult
-        finally WorkerExchange.unregisterListener(wl)
+        finally
+          w.close()
+          WorkerExchange.unregisterListener(wl)
       testListeners.foreach(_.doComplete(result.overall))
       result
     } // end task
diff --git a/main-actions/src/main/scala/sbt/Tests.scala b/main-actions/src/main/scala/sbt/Tests.scala
index 40e448f35..dddca5f1e 100644
--- a/main-actions/src/main/scala/sbt/Tests.scala
+++ b/main-actions/src/main/scala/sbt/Tests.scala
@@ -50,11 +50,21 @@ object Tests {
    * @param events The result of each test group (suite) executed during this test run.
    * @param summaries Explicit summaries directly provided by test frameworks.  This may be empty, in which case a default summary will be generated.
    */
-  private[sbt] final case class Output(
+  final case class Output(
       overall: TestResult,
       events: Map[String, SuiteResult],
       summaries: Iterable[Summary]
-  )
+  ) {
+
+    /**
+     * Returns a copy with the throwables removed from every suite result.
+     *
+     * Use this before retaining test results beyond the lifetime of the test task.
+     * See [[SuiteResult.throwables]] for why retaining those exceptions can keep the test class loader alive.
+     */
+    def withoutThrowables: Output =
+      copy(events = events.view.mapValues(_.withoutThrowables).toMap)
+  }
 
   /**
    * Summarizes a test run.
diff --git a/main-actions/src/main/scala/sbt/internal/WorkerExchange.scala b/main-actions/src/main/scala/sbt/internal/WorkerExchange.scala
index 7d89d6f9e..fe22ca67c 100644
--- a/main-actions/src/main/scala/sbt/internal/WorkerExchange.scala
+++ b/main-actions/src/main/scala/sbt/internal/WorkerExchange.scala
@@ -11,21 +11,62 @@ package internal
 
 import org.scalasbt.shadedgson.com.google.gson.Gson
 import java.io.*
-import java.net.{ InetAddress, ServerSocket }
+import java.net.{ InetAddress, ServerSocket, StandardProtocolFamily, UnixDomainSocketAddress }
+import java.nio.channels.{ ServerSocketChannel, SocketChannel }
+import java.nio.file.{ Files, Path as NioPath }
 import java.util.Scanner
 import sbt.io.IO
 import sbt.internal.io.Retry
 import sbt.internal.worker1.*
+import sbt.protocol.DuplexChannels
 import sbt.testing.Framework
 import scala.sys.process.{ BasicIO, Process, ProcessIO }
 import scala.collection.mutable
+import scala.collection.concurrent.TrieMap
 import scala.collection.mutable.ListBuffer
 import scala.concurrent.{ Await, Promise }
 import scala.concurrent.duration.*
+import scala.util.control.NonFatal
 
 object WorkerExchange:
   val listeners: mutable.ListBuffer[WorkerResponseListener] = ListBuffer.empty
   private val loopback = InetAddress.getByName(null)
+  private val jdkIpcSupportCache = TrieMap.empty[Option[File], Boolean]
+
+  /**
+   * Start a worker process.
+   */
+  def startWorker(fo: ForkOptions, extraCp: Seq[File]): WorkerProxy =
+    val ct =
+      if supportsUnixDomainSockets(fo.javaHome) then WorkerConnection.Ipc(newIpcSocketPath())
+      else WorkerConnection.Stdio
+    startWorker(fo, extraCp, ct)
+
+  /**
+   * True if `javaHome` (None meaning the JDK currently running sbt) is JDK 16+.
+   */
+  private def supportsUnixDomainSockets(javaHome: Option[File]): Boolean =
+    def doDetect: Boolean =
+      javaHome match
+        case None       => true // the JDK running sbt itself, which requires 17+
+        case Some(home) =>
+          try
+            val releaseFile = File(home, "release")
+            val props = java.util.Properties()
+            val in = FileInputStream(releaseFile)
+            try props.load(in)
+            finally in.close()
+            val raw = Option(props.getProperty("JAVA_VERSION")).getOrElse("")
+            val version = raw.stripPrefix("\"").stripSuffix("\"")
+            val digits =
+              version.takeWhile(c => c.isDigit || c == '.').split('.').flatMap(_.toIntOption)
+            val major = digits match
+              case Array(1, minor, _*) => minor // legacy 1.8-style versioning
+              case Array(m, _*)        => m
+              case _                   => 0
+            major >= 16
+          catch case NonFatal(_) => false
+    jdkIpcSupportCache.getOrElseUpdate(javaHome, doDetect)
 
   /**
    * Start a worker process.
@@ -42,26 +83,50 @@ object WorkerExchange:
       IO.classLocationPath(classOf[Gson]).toFile,
     )
     val inputRef = Promise[OutputStream]()
-    val socketOpt = connectionType match
+    def runAccepter(out: OutputStream, in: InputStream): Unit =
+      inputRef.success(out)
+      val scanner = Scanner(in, "UTF-8")
+      while scanner.hasNextLine() do notifyListeners(scanner.nextLine())
+    val (connArgs, closer): (Seq[String], Option[AutoCloseable]) = connectionType match
       case WorkerConnection.Tcp =>
         val serverSocket = Retry(ServerSocket(0, 1, loopback))
         val accepter = Thread(() => {
           val socket = serverSocket.accept()
-          inputRef.success(socket.getOutputStream())
-          val scanner = Scanner(socket.getInputStream(), "UTF-8")
-          while scanner.hasNextLine() do notifyListeners(scanner.nextLine())
+          runAccepter(socket.getOutputStream(), socket.getInputStream())
         })
         accepter.start()
-        Some(serverSocket)
-      case _ => None
+        (Seq("--tcp", serverSocket.getLocalPort().toString()), Some(serverSocket))
+      case WorkerConnection.Ipc(path) =>
+        val serverChannel = Retry {
+          Files.deleteIfExists(path)
+          val ch = ServerSocketChannel.open(StandardProtocolFamily.UNIX)
+          ch.bind(UnixDomainSocketAddress.of(path))
+          ch
+        }
+        @volatile var acceptedChannel: SocketChannel = null
+        val accepter = Thread(() => {
+          val channel = serverChannel.accept()
+          acceptedChannel = channel
+          runAccepter(
+            DuplexChannels.newOutputStream(channel),
+            DuplexChannels.newInputStream(channel)
+          )
+        })
+        accepter.setName("sbt-fork-test-response-reader")
+        accepter.setPriority(Thread.NORM_PRIORITY + 1)
+        accepter.start()
+        val closer: AutoCloseable = () => {
+          if acceptedChannel != null then acceptedChannel.close()
+          serverChannel.close()
+          Files.deleteIfExists(path)
+        }
+        (Seq("--ipc", path.toString()), Some(closer))
+      case WorkerConnection.Stdio => (Nil, None)
     val options = Seq(
       "-classpath",
       fullCp.mkString(File.pathSeparator),
       classOf[WorkerMain].getCanonicalName,
-    ) ++
-      (socketOpt match
-        case Some(s) => Seq("--tcp", s.getLocalPort().toString())
-        case _       => Nil)
+    ) ++ connArgs
     val onStdoutLine: String => Unit = connectionType match
       case WorkerConnection.Stdio => notifyListeners
       case _                      => (line) => scala.Console.out.println(line)
@@ -78,7 +143,17 @@ object WorkerExchange:
     val p = Fork.java.fork(forkWithIo, options)
     val forkTimeout = fo.connectionTimeout.getOrElse(30.seconds)
     val input = Await.result(inputRef.future, forkTimeout)
-    WorkerProxy(input, p, options, socketOpt)
+    WorkerProxy(input, p, options, closer)
+
+  /** Generates a fresh path suitable for binding a `WorkerConnection.Ipc` socket. */
+  def newIpcSocketPath(): NioPath =
+    val dir = NioPath
+      .of(sys.env.getOrElse("XDG_RUNTIME_DIR", sys.props("java.io.tmpdir")))
+      .resolve(".sbt-fork-ipc")
+    Files.createDirectories(dir)
+    val path = Files.createTempFile(dir, "fork-", ".sock")
+    Files.deleteIfExists(path)
+    path
 
   def registerListener(listener: WorkerResponseListener): Unit =
     synchronized:
@@ -102,12 +177,12 @@ class WorkerProxy(
     input: OutputStream,
     val process: Process,
     val options: Seq[String],
-    serverSocket: Option[ServerSocket],
+    closer: Option[AutoCloseable],
 ) extends AutoCloseable:
   lazy val inputStream = PrintStream(input)
   def close(): Unit =
     input.close()
-    serverSocket.foreach(_.close())
+    closer.foreach(_.close())
   def blockForExitCode(): Int =
     if !process.isAlive() then process.exitValue()
     else Fork.blockForExitCode(process)
@@ -130,3 +205,4 @@ abstract class WorkerResponseListener extends Function1[String, Unit]:
 enum WorkerConnection:
   case Stdio
   case Tcp
+  case Ipc(path: NioPath)
diff --git a/main-actions/src/test/scala/sbt/internal/WorkerExchangeTest.scala b/main-actions/src/test/scala/sbt/internal/WorkerExchangeTest.scala
index fd64d7fa1..52f9bca5d 100644
--- a/main-actions/src/test/scala/sbt/internal/WorkerExchangeTest.scala
+++ b/main-actions/src/test/scala/sbt/internal/WorkerExchangeTest.scala
@@ -9,13 +9,18 @@ import scala.sys.process.Process
 
 object WorkerExchangeTest extends Properties:
   given Gen[WorkerConnection] =
-    Gen.choice1(Gen.constant(WorkerConnection.Stdio), Gen.constant(WorkerConnection.Tcp))
+    Gen.choice1(
+      Gen.constant(WorkerConnection.Stdio),
+      Gen.constant(WorkerConnection.Tcp),
+      Gen.constant(WorkerConnection.Ipc(WorkerExchange.newIpcSocketPath())),
+    )
 
   def gen[A1: Gen]: Gen[A1] = summon[Gen[A1]]
 
   override lazy val tests: List[Test] = List(
     propertyN("non-jsonrpc should return exit code 1", propBadInput, 10),
     propertyN("bye should return response json with a result", propBye, 10),
+    example("startWorker(fo, extraCp) auto-detects a working connection type", exampleAutoDetect),
   )
 
   def propertyN(name: String, result: => Property, n: Int): Test =
@@ -47,6 +52,17 @@ object WorkerExchangeTest extends Properties:
         .and(Result.assert(l.sb.toString() == s"""{ "jsonrpc": "2.0", "result": 0, "id": $i }"""))
         .log(s"\"${l.sb.toString()}\"")
 
+  def exampleAutoDetect: Result =
+    val w = WorkerExchange.startWorker(ForkOptions(), Nil)
+    withListener: l =>
+      w.println("""{"jsonrpc": "2.0", "method": "bye", "params": {}, "id": 1}""")
+      val exitCode = w.blockForExitCode()
+      l.awaitResponse()
+      Result
+        .assert(exitCode == 0)
+        .and(Result.assert(l.sb.toString() == """{ "jsonrpc": "2.0", "result": 0, "id": 1 }"""))
+        .log(s"\"${l.sb.toString()}\"")
+
   def withListener[A1](f: ConcreteListener => A1) =
     val l = ConcreteListener()
     try
diff --git a/main-command/src/main/scala/sbt/State.scala b/main-command/src/main/scala/sbt/State.scala
index b1afeb374..ec575bc9a 100644
--- a/main-command/src/main/scala/sbt/State.scala
+++ b/main-command/src/main/scala/sbt/State.scala
@@ -248,12 +248,40 @@ object State {
     val app = state.configuration.provider
     new Reboot(
       app.scalaProvider.version,
-      state.remainingCommands map { case e: Exec => e.commandLine },
+      addPluginSbtFileArguments(state) ::: state.remainingCommands.map(_.commandLine),
       app.id,
       state.configuration.baseDirectory
     )
   }
 
+  /**
+   * Builds the `early(...)` commands that add the extra plugin sbt files back after a reboot.
+   * A reboot clears the state, so sbt forgets these files. The commands are passed to the new
+   * sbt as arguments, and they run before it loads the build.
+   *
+   * The `early(addPluginSbtFile=...)` form is used, not `--addPluginSbtFile=...`, because the
+   * `--` form loses the quotes around the path, and then a path with a space in it fails.
+   * A path with a space can be added using `addPluginSbtFile "<path>"`.
+   */
+  private[sbt] def addPluginSbtFileArguments(state: State): List[String] =
+    state.get(BasicKeys.extraMetaSbtFiles).toList.flatten.distinct.map { vf =>
+      val path = vf match {
+        case f: xsbti.PathBasedFile => f.toPath.toString
+        case f                      => f.id
+      }
+      val command = s"${BasicCommandStrings.AddPluginSbtFileCommand}=${quote(path)}"
+      s"${BasicCommandStrings.EarlyCommand}($command)"
+    }
+
+  /**
+   * Puts `path` in quotes so the new sbt session reads it back as one whole string. There the path goes
+   * through the `Parsers.StringBasic` in `BasicCommands.addPluginSbtFileParser`, which stops at
+   * the first space when there are no quotes. A path with a space can be added, so it also has
+   * to come back after a reboot.
+   */
+  private def quote(path: String): String =
+    s"\"${path.replace("\\", "\\\\").replace("\"", "\\\"")}\""
+
   @deprecated("Import State._ or State.StateOpsImpl to access state extension methods", "1.3.0")
   def stateOps(s: State): StateOps = new StateOpsImpl(s)
 
@@ -331,8 +359,9 @@ object State {
           StartServer :: remaining.dropWhile(!_.startsWith(ReportResult)).tail ::: "shell" :: Nil
         case _ => remaining
       }
-      if (currentOnly) throw new RebootCurrent(fullRemaining)
-      else throw new xsbti.FullReload(fullRemaining.toArray, full)
+      val arguments = State.addPluginSbtFileArguments(s) ::: fullRemaining
+      if (currentOnly) throw new RebootCurrent(arguments)
+      else throw new xsbti.FullReload(arguments.toArray, full)
     }
 
     def reload = runExitHooks().setNext(new Return(defaultReload(s)))
diff --git a/main/src/main/scala/sbt/internal/Load.scala b/main/src/main/scala/sbt/internal/Load.scala
index 15d810942..8d2937c3e 100755
--- a/main/src/main/scala/sbt/internal/Load.scala
+++ b/main/src/main/scala/sbt/internal/Load.scala
@@ -15,7 +15,7 @@ import sbt.Keys.*
 import sbt.Project.inScope
 import sbt.ProjectExtra.{ prefixConfigs, setProject, showLoadingKey, structure }
 import sbt.Scope.GlobalScope
-import sbt.ScopeAxis.{ Select, Zero }
+import sbt.ScopeAxis.{ Select, This, Zero }
 import sbt.SlashSyntax0.*
 import sbt.internal.BuildStreams.*
 import sbt.internal.inc.classpath.ClasspathUtil
@@ -1056,6 +1056,11 @@ private[sbt] object Load {
         log = log,
       )
 
+    // Excludes settings already scoped explicitly (ThisBuild, Global); those apply once (#9668).
+    def thisProjectScoped(settings: Seq[Setting[?]]): Seq[Setting[?]] =
+      settings.filter: s =>
+        s.key.scope.project == This || s.key.scope.project == Select(ThisProject)
+
     // load all relevant configuration files (.sbt, as .scala already exists at this point)
     def discover(base: File): DiscoveredProjects = {
       val auto =
@@ -1118,7 +1123,9 @@ private[sbt] object Load {
       val newAcc = acc :+ finalRoot
       val newGenerated = generated ++ generatedConfigClassFiles
       // only root-level settings are build-wide; a submodule's own settings must not leak to siblings (#9517).
-      val cs = if isRootPath(p.base, buildBase) then finalRoot.commonSettings else commonSettings
+      val cs =
+        if isRootPath(p.base, buildBase) then thisProjectScoped(finalRoot.commonSettings)
+        else commonSettings
       loadTransitive1(newProjects, newAcc, newGenerated, cs)
     }
 
@@ -1154,7 +1161,9 @@ private[sbt] object Load {
                   acc = acc,
                   generated = Nil,
                   commonSettings0 = commonSettings
-                    ++ expandCommonSettingsPerBase1(buildBase.getCanonicalFile()),
+                    ++ thisProjectScoped(
+                      expandCommonSettingsPerBase1(buildBase.getCanonicalFile())
+                    ),
                 )
               val existingIds = otherProjects.projects.map(_.id)
               val refs = existingIds.map(id => ProjectRef(buildUri, id))
@@ -1167,7 +1176,12 @@ private[sbt] object Load {
         val newAcc = finalRoot +: (acc ++ otherProjects.projects)
         val newGenerated =
           generated ++ otherProjects.generatedConfigClassFiles ++ generatedConfigClassFiles
-        loadTransitive1(newProjects, newAcc, newGenerated, finalRoot.commonSettings)
+        loadTransitive1(
+          newProjects,
+          newAcc,
+          newGenerated,
+          thisProjectScoped(finalRoot.commonSettings)
+        )
       case Nil =>
         val projectIds = acc.map(_.id).mkString("(", ", ", ")")
         log.debug(s"[Loading] Done in $buildBase, returning: $projectIds")
diff --git a/project/Utils.scala b/project/Utils.scala
index 9a66ba076..0abb747e5 100644
--- a/project/Utils.scala
+++ b/project/Utils.scala
@@ -7,6 +7,8 @@ import sbt.internal.inc.Analysis
 
 object Utils {
   val version2_13 = settingKey[String]("version number")
+  val JDK17 = config("jdk17")
+
   val ExclusiveTest: Tags.Tag = Tags.Tag("exclusive-test")
 
   val componentID: SettingKey[Option[String]] = settingKey[Option[String]]("")
diff --git a/protocol/src/main/java/sbt/protocol/DuplexChannels.java b/protocol/src/main/java/sbt/protocol/DuplexChannels.java
new file mode 100644
index 000000000..0ea53307c
--- /dev/null
+++ b/protocol/src/main/java/sbt/protocol/DuplexChannels.java
@@ -0,0 +1,91 @@
+/*
+ * 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.protocol;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+
+/**
+ * java.nio.channels.Channels.newInputStream/newOutputStream both synchronize on the channel's
+ * blockingLock() for the duration of each blocking call, so a thread parked in a blocking read
+ * holds that lock for as long as the read blocks, and a concurrent writer on the same channel can
+ * never acquire it. These factories talk to the channel directly instead, so a SocketChannel can
+ * safely be read and written from different threads at the same time.
+ */
+public final class DuplexChannels {
+  private DuplexChannels() {}
+
+  public static OutputStream newOutputStream(SocketChannel ch) {
+    return new OutputStream() {
+      @Override
+      public void write(int b) throws IOException {
+        ByteBuffer bb = ByteBuffer.wrap(new byte[] {(byte) b});
+        while (bb.hasRemaining()) ch.write(bb);
+      }
+
+      @Override
+      public void write(byte[] b, int off, int len) throws IOException {
+        ByteBuffer bb = ByteBuffer.wrap(b, off, len);
+        while (bb.hasRemaining()) ch.write(bb);
+      }
+    };
+  }
+
+  public static InputStream newInputStream(SocketChannel ch) {
+    return new InputStream() {
+      @Override
+      public int read() throws IOException {
+        ByteBuffer bb = ByteBuffer.allocate(1);
+        int n = ch.read(bb);
+        return n <= 0 ? -1 : (bb.get(0) & 0xff);
+      }
+
+      @Override
+      public int read(byte[] b, int off, int len) throws IOException {
+        if (len == 0) return 0;
+        return ch.read(ByteBuffer.wrap(b, off, len));
+      }
+    };
+  }
+
+  /**
+   * Wraps a connected SocketChannel as a Socket backed by {@link #newInputStream}/{@link
+   * #newOutputStream}.
+   */
+  public static Socket newSocket(SocketChannel ch) {
+    return new Socket() {
+      private final InputStream in = newInputStream(ch);
+      private final OutputStream out = newOutputStream(ch);
+
+      @Override
+      public InputStream getInputStream() {
+        return in;
+      }
+
+      @Override
+      public OutputStream getOutputStream() {
+        return out;
+      }
+
+      @Override
+      public void close() throws IOException {
+        ch.close();
+      }
+
+      @Override
+      public boolean isClosed() {
+        return !ch.isOpen();
+      }
+    };
+  }
+}
diff --git a/protocol/src/main/scala/sbt/protocol/ClientSocket.scala b/protocol/src/main/scala/sbt/protocol/ClientSocket.scala
index 79f333cc5..cffd4cba4 100644
--- a/protocol/src/main/scala/sbt/protocol/ClientSocket.scala
+++ b/protocol/src/main/scala/sbt/protocol/ClientSocket.scala
@@ -10,7 +10,8 @@ package sbt
 package protocol
 
 import java.io.File
-import java.net.{ Socket, URI, InetAddress }
+import java.net.{ InetAddress, Socket, StandardProtocolFamily, URI, UnixDomainSocketAddress }
+import java.nio.channels.SocketChannel
 import sjsonnew.BasicJsonProtocol
 import sjsonnew.support.scalajson.unsafe.{ Parser, Converter }
 import sjsonnew.shaded.scalajson.ast.unsafe.JValue
@@ -45,4 +46,9 @@ object ClientSocket {
   def localSocket(name: String, useJNI: Boolean): Socket =
     if (isWindows) new Win32NamedPipeSocket(s"\\\\.\\pipe\\$name", useJNI)
     else new UnixDomainSocket(name, useJNI)
+
+  def bootSocket(path: String): Socket =
+    val ch = SocketChannel.open(StandardProtocolFamily.UNIX)
+    ch.connect(UnixDomainSocketAddress.of(path))
+    DuplexChannels.newSocket(ch)
 }
diff --git a/protocol/src/test/scala/sbt/protocol/ClientSocketDuplexTest.scala b/protocol/src/test/scala/sbt/protocol/ClientSocketDuplexTest.scala
new file mode 100644
index 000000000..604a4d954
--- /dev/null
+++ b/protocol/src/test/scala/sbt/protocol/ClientSocketDuplexTest.scala
@@ -0,0 +1,130 @@
+/*
+ * 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.protocol
+
+import hedgehog.{ Gen, Property, Result }
+import hedgehog.core.{ ShrinkLimit, SuccessCount }
+import hedgehog.runner.*
+import java.io.{ EOFException, InputStream }
+import java.net.{ StandardProtocolFamily, UnixDomainSocketAddress }
+import java.nio.ByteBuffer
+import java.nio.channels.{ ServerSocketChannel, SocketChannel }
+import java.util.concurrent.{ CountDownLatch, LinkedBlockingQueue, TimeUnit }
+import scala.util.Using
+import scala.util.control.NonFatal
+import sbt.io.IO
+
+/**
+ * Regression test: bootSocket used to wrap its channel with
+ * Channels.newInputStream/newOutputStream, which share the channel's blockingLock() and deadlock
+ * a blocking read against a concurrent write on the same channel. It now uses DuplexChannels,
+ * which talks to the channel directly and has no such shared lock. What matters for reproducing
+ * the deadlock is timing, not payload content, so the reader and writer threads are each given an
+ * independent startup delay drawn from {0, 100, 300}ms to exercise the read starting well before,
+ * around the same time as, and well after the write.
+ */
+object ClientSocketDuplexTest extends Properties:
+  override def tests: List[Test] = List(
+    propertyN(
+      "bootSocket: a concurrent read and write on the same channel do not deadlock",
+      propDuplex,
+      20,
+    ),
+  )
+
+  def propertyN(name: String, result: => Property, n: Int): Test =
+    Test(name, result)
+      .config(_.copy(testLimit = SuccessCount(n), shrinkLimit = ShrinkLimit(n * 10)))
+
+  private val toServer: Array[Byte] = Array[Byte](1)
+  private val toClient: Array[Byte] = Array[Byte](2)
+
+  val sleepMsGen: Gen[Int] = Gen.element1(0, 100, 300)
+
+  def propDuplex: Property =
+    for
+      readerSleepMs <- sleepMsGen.log("reader startup delay (ms)")
+      writerSleepMs <- sleepMsGen.log("writer startup delay (ms)")
+    yield runDuplexRound(readerSleepMs, writerSleepMs)
+
+  private def runDuplexRound(readerSleepMs: Int, writerSleepMs: Int): Result =
+    IO.withTemporaryDirectory: dir =>
+      val path = dir.toPath.resolve("boot.sock")
+      Using.resource(ServerSocketChannel.open(StandardProtocolFamily.UNIX)): serverChannel =>
+        serverChannel.bind(UnixDomainSocketAddress.of(path))
+        Using.resource(ClientSocket.bootSocket(path.toString)): client =>
+          Using.resource(serverChannel.accept()): serverSide =>
+            // The client's reader is parked waiting for toClient before the server has sent
+            // anything, so it's mid-read (and would be holding blockingLock() under the old
+            // Channels-based implementation) while we race the write below against it.
+            val readOutcome = new LinkedBlockingQueue[Either[Throwable, Array[Byte]]]()
+            val reader = new Thread(() =>
+              readOutcome.put(
+                try
+                  Thread.sleep(readerSleepMs.toLong)
+                  Right(readNBytes(client.getInputStream(), toClient.length))
+                catch case NonFatal(e) => Left(e)
+              )
+            )
+            reader.setDaemon(true)
+            reader.start()
+
+            val writeDone = new CountDownLatch(1)
+            val writer = new Thread(() =>
+              try
+                Thread.sleep(writerSleepMs.toLong)
+                client.getOutputStream().write(toServer)
+              catch case NonFatal(_) => ()
+              finally writeDone.countDown()
+            )
+            writer.setDaemon(true)
+            writer.start()
+
+            if !writeDone.await(3, TimeUnit.SECONDS) then
+              Result.failure.log(
+                "write blocked behind the concurrent read: possible regression of the " +
+                  "Channels.newInputStream/newOutputStream blockingLock() deadlock"
+              )
+            else
+              val fromClient = readNBytes(serverSide, toServer.length)
+              serverSide.write(ByteBuffer.wrap(toClient))
+              readOutcome.poll(3, TimeUnit.SECONDS) match
+                case null =>
+                  Result.failure.log(
+                    "client's blocked read never completed after the server replied"
+                  )
+                case Left(e)           => Result.failure.log(s"client read failed: $e")
+                case Right(fromServer) =>
+                  Result.all(
+                    List(
+                      Result
+                        .assert(fromClient.sameElements(toServer))
+                        .log("server received a different payload than the client sent"),
+                      Result
+                        .assert(fromServer.sameElements(toClient))
+                        .log("client received a different payload than the server sent"),
+                    )
+                  )
+
+  private def readNBytes(in: InputStream, n: Int): Array[Byte] =
+    val buf = new Array[Byte](n)
+    var total = 0
+    while total < n do
+      val r = in.read(buf, total, n - total)
+      if r < 0 then throw new EOFException(s"expected $n bytes, got $total")
+      total += r
+    buf
+
+  private def readNBytes(ch: SocketChannel, n: Int): Array[Byte] =
+    val bb = ByteBuffer.allocate(n)
+    while bb.hasRemaining() do
+      val r = ch.read(bb)
+      if r < 0 then throw new EOFException(s"expected $n bytes, got ${bb.position()}")
+    bb.array()
+end ClientSocketDuplexTest
diff --git a/sbt-app/src/sbt-test/global-plugin/global-plugin/test b/sbt-app/src/sbt-test/global-plugin/global-plugin/test
index ae785f180..bd79c7bb5 100644
--- a/sbt-app/src/sbt-test/global-plugin/global-plugin/test
+++ b/sbt-app/src/sbt-test/global-plugin/global-plugin/test
@@ -13,3 +13,9 @@ $ copy-file changes/global-plugins.sbt global/plugins/plugins.sbt
 $ copy-file changes/plugins.sbt project/plugins.sbt
 > reload
 > check
+
+# The tests of one scripted run share a directory, and `global` is the part of it that is kept between
+# them, so delete what this test installed there.
+$ delete global/plugins global/useGlobalAutoPlugin.sbt
+$ absent global/plugins global/useGlobalAutoPlugin.sbt
+> reload
diff --git a/sbt-app/src/sbt-test/project/addPluginSbtFile-reboot/build.sbt b/sbt-app/src/sbt-test/project/addPluginSbtFile-reboot/build.sbt
new file mode 100644
index 000000000..5a1a82fcf
--- /dev/null
+++ b/sbt-app/src/sbt-test/project/addPluginSbtFile-reboot/build.sbt
@@ -0,0 +1,46 @@
+import sbt.internal.LoadedBuild
+
+lazy val root = project.in(file("."))
+
+def detectedPlugins(lb: LoadedBuild): Seq[String] =
+  lb.units(lb.root).unit.plugins.detected.autoPlugins.map(_.name)
+
+InputKey[Unit]("checkPlugins") := {
+  val args = Def.spaceDelimited("<names>").parsed
+  val detected = detectedPlugins(loadedBuild.value)
+  args.foreach { name =>
+    assert(
+      detected.exists(_.contains(name)),
+      s"expected plugin $name to be detected, got: ${detected.mkString(", ")}"
+    )
+  }
+}
+
+InputKey[Unit]("checkPluginsAbsent") := {
+  val args = Def.spaceDelimited("<names>").parsed
+  val detected = detectedPlugins(loadedBuild.value)
+  args.foreach { name =>
+    assert(
+      !detected.exists(_.contains(name)),
+      s"expected plugin $name not to be detected, got: ${detected.mkString(", ")}"
+    )
+  }
+}
+
+// Compares the whole list of registered paths, in order. This also catches a path that comes
+// back from a reboot changed, doubled or in a different place, not only one that is lost.
+// With no arguments it asserts that no file is registered.
+InputKey[Unit]("checkFiles") := {
+  val expected = Def.spaceDelimited("<paths>").parsed.toList
+  val actual = state.value.get(BasicKeys.extraMetaSbtFiles).toList.flatten.map(_.id)
+  assert(
+    actual == expected,
+    s"expected registered files to be [${expected.mkString(", ")}], got: [${actual.mkString(", ")}]"
+  )
+}
+
+// Tests in a scripted group share one sbt session, and the extra files now survive reboot on
+// purpose, so they have to be dropped before the next test.
+commands += Command.command("clearExtraPluginSbtFiles") { s =>
+  s.remove(BasicKeys.extraMetaSbtFiles)
+}
diff --git a/sbt-app/src/sbt-test/project/addPluginSbtFile-reboot/temp with spaces/extraC.sbt b/sbt-app/src/sbt-test/project/addPluginSbtFile-reboot/temp with spaces/extraC.sbt
new file mode 100644
index 000000000..ecbbaada7
--- /dev/null
+++ b/sbt-app/src/sbt-test/project/addPluginSbtFile-reboot/temp with spaces/extraC.sbt	
@@ -0,0 +1 @@
+addSbtPlugin("org.scala-native" % "sbt-scala-native" % "0.5.11")
diff --git a/sbt-app/src/sbt-test/project/addPluginSbtFile-reboot/temp/extraA.sbt b/sbt-app/src/sbt-test/project/addPluginSbtFile-reboot/temp/extraA.sbt
new file mode 100644
index 000000000..ddfa827f9
--- /dev/null
+++ b/sbt-app/src/sbt-test/project/addPluginSbtFile-reboot/temp/extraA.sbt
@@ -0,0 +1 @@
+addSbtPlugin("com.eed3si9n" % "sbt-buildinfo" % "0.13.1")
diff --git a/sbt-app/src/sbt-test/project/addPluginSbtFile-reboot/temp/extraB.sbt b/sbt-app/src/sbt-test/project/addPluginSbtFile-reboot/temp/extraB.sbt
new file mode 100644
index 000000000..fdcd4e64d
--- /dev/null
+++ b/sbt-app/src/sbt-test/project/addPluginSbtFile-reboot/temp/extraB.sbt
@@ -0,0 +1 @@
+addSbtPlugin("org.scala-js" % "sbt-scalajs" % "1.22.0")
diff --git a/sbt-app/src/sbt-test/project/addPluginSbtFile-reboot/test b/sbt-app/src/sbt-test/project/addPluginSbtFile-reboot/test
new file mode 100644
index 000000000..42f3eada7
--- /dev/null
+++ b/sbt-app/src/sbt-test/project/addPluginSbtFile-reboot/test
@@ -0,0 +1,43 @@
+# Regression test for sbt/sbt#4303: files added with --addPluginSbtFile must survive reboot.
+
+# Neither plugin is on the meta-build classpath to begin with, and nothing is registered.
+> checkPluginsAbsent BuildInfoPlugin ScalaJSPlugin ScalaNativePlugin
+> checkFiles
+
+# Add the first extra plugin sbt file and check that its plugin is picked up.
+> early(addPluginSbtFile=temp/extraA.sbt); reload
+> checkPlugins BuildInfoPlugin
+> checkFiles temp/extraA.sbt
+
+# The first reboot must not lose it.
+> reboot
+> checkPlugins BuildInfoPlugin
+> checkFiles temp/extraA.sbt
+
+# Add a second extra plugin sbt file on top of the first one.
+> early(addPluginSbtFile=temp/extraB.sbt); reload
+> checkPlugins BuildInfoPlugin ScalaJSPlugin
+> checkFiles temp/extraA.sbt temp/extraB.sbt
+
+# The second reboot must keep both of them, and must not add either of them twice.
+> reboot
+> checkPlugins BuildInfoPlugin ScalaJSPlugin
+> checkFiles temp/extraA.sbt temp/extraB.sbt
+
+# A third file, this one in a directory whose name contains spaces. The path has to be quoted
+# for sbt to parse it as one argument, and the reboot has to keep it quoted to replay it.
+> addPluginSbtFile "temp with spaces/extraC.sbt"
+> reload
+> checkPlugins BuildInfoPlugin ScalaJSPlugin ScalaNativePlugin
+> checkFiles temp/extraA.sbt temp/extraB.sbt "temp with spaces/extraC.sbt"
+
+# The third reboot must keep all three, in order, with the spaced path intact.
+> reboot
+> checkPlugins BuildInfoPlugin ScalaJSPlugin ScalaNativePlugin
+> checkFiles temp/extraA.sbt temp/extraB.sbt "temp with spaces/extraC.sbt"
+
+# Drop the extra files, because the sbt session can be shared with the other scripted tests.
+> clearExtraPluginSbtFiles
+> reload
+> checkPluginsAbsent BuildInfoPlugin ScalaJSPlugin ScalaNativePlugin
+> checkFiles
diff --git a/sbt-app/src/sbt-test/project/common-settings-synthetic-root/build.sbt b/sbt-app/src/sbt-test/project/common-settings-synthetic-root/build.sbt
index 8a7a122e4..ff9984941 100644
--- a/sbt-app/src/sbt-test/project/common-settings-synthetic-root/build.sbt
+++ b/sbt-app/src/sbt-test/project/common-settings-synthetic-root/build.sbt
@@ -5,6 +5,16 @@ scalaVersion := scala212
 val o = "com.example"
 organization := o
 
+lazy val aa = settingKey[Seq[String]]("")
+Global / aa := Seq("initial-value")
+// explicitly ThisBuild-scoped settings must apply exactly once, not once per project (#9668)
+ThisBuild / aa += "added-value"
+
+lazy val bb = settingKey[Seq[String]]("")
+Global / bb := Seq("initial-value")
+// ThisProject means "current project", so it must still apply once per project, unlike ThisBuild
+ThisProject / bb += "added-value"
+
 lazy val foo = project
 lazy val bar = project
   .settings(
@@ -24,5 +34,17 @@ check := {
   assert((bar / organization).value == "com.example.bar", s"unexpected bar / organization = {(bar / organization).value}")
   // Test that baz/build.sbt bare settings get loaded
   assert((baz / organization).value == "com.example.baz", s"unexpected baz/organization")
+
+  // Test that ThisBuild / aa += is applied exactly once, not once per project (#9668)
+  val expectedAa = Seq("initial-value", "added-value")
+  assert((foo / aa).value == expectedAa, s"(foo / aa).value: ${(foo / aa).value}")
+  assert((bar / aa).value == expectedAa, s"(bar / aa).value: ${(bar / aa).value}")
+  assert((baz / aa).value == expectedAa, s"(baz / aa).value: ${(baz / aa).value}")
+
+  // Test that ThisProject / bb += still applies once per project (#9668)
+  val expectedBb = Seq("initial-value", "added-value")
+  assert((foo / bb).value == expectedBb, s"(foo / bb).value: ${(foo / bb).value}")
+  assert((bar / bb).value == expectedBb, s"(bar / bb).value: ${(bar / bb).value}")
+  assert((baz / bb).value == expectedBb, s"(baz / bb).value: ${(baz / bb).value}")
 }
 check / aggregate := false
diff --git a/sbt-app/src/sbt-test/project/common-settings/build.sbt b/sbt-app/src/sbt-test/project/common-settings/build.sbt
index d734f3509..2c79c2ea5 100644
--- a/sbt-app/src/sbt-test/project/common-settings/build.sbt
+++ b/sbt-app/src/sbt-test/project/common-settings/build.sbt
@@ -6,6 +6,16 @@ scalaVersion := scala212
 val o = "com.example"
 organization := o
 
+lazy val aa = settingKey[Seq[String]]("")
+Global / aa := Seq("initial-value")
+// explicitly ThisBuild-scoped settings must apply exactly once, not once per project (#9668)
+ThisBuild / aa += "added-value"
+
+lazy val bb = settingKey[Seq[String]]("")
+Global / bb := Seq("initial-value")
+// ThisProject means "current project", so it must still apply once per project, unlike ThisBuild
+ThisProject / bb += "added-value"
+
 lazy val root = rootProject
   .autoAggregate
 
@@ -34,5 +44,21 @@ LocalRootProject / check := {
   assert((baz / organization).value == "com.example.baz")
   // Test that baz/build.sbt settings don't leak onto qux, processed right after it (#9517)
   assert((qux / organization).value == o, s"(qux / organization).value: ${(qux / organization).value}")
+
+  // Test that ThisBuild / aa += is applied exactly once, not once per project (#9668)
+  val expectedAa = Seq("initial-value", "added-value")
+  assert((root / aa).value == expectedAa, s"(root / aa).value: ${(root / aa).value}")
+  assert((foo / aa).value == expectedAa, s"(foo / aa).value: ${(foo / aa).value}")
+  assert((bar / aa).value == expectedAa, s"(bar / aa).value: ${(bar / aa).value}")
+  assert((baz / aa).value == expectedAa, s"(baz / aa).value: ${(baz / aa).value}")
+  assert((qux / aa).value == expectedAa, s"(qux / aa).value: ${(qux / aa).value}")
+
+  // Test that ThisProject / bb += still applies once per project (#9668)
+  val expectedBb = Seq("initial-value", "added-value")
+  assert((root / bb).value == expectedBb, s"(root / bb).value: ${(root / bb).value}")
+  assert((foo / bb).value == expectedBb, s"(foo / bb).value: ${(foo / bb).value}")
+  assert((bar / bb).value == expectedBb, s"(bar / bb).value: ${(bar / bb).value}")
+  assert((baz / bb).value == expectedBb, s"(baz / bb).value: ${(baz / bb).value}")
+  assert((qux / bb).value == expectedBb, s"(qux / bb).value: ${(qux / bb).value}")
 }
 check / aggregate := false
diff --git a/sbt-app/src/sbt-test/tests/fork-shutdown-hook/build.sbt b/sbt-app/src/sbt-test/tests/fork-shutdown-hook/build.sbt
new file mode 100644
index 000000000..1b63fb338
--- /dev/null
+++ b/sbt-app/src/sbt-test/tests/fork-shutdown-hook/build.sbt
@@ -0,0 +1,21 @@
+Global / localCacheDirectory := baseDirectory.value / "diskcache"
+scalaVersion := "3.8.4"
+
+Test / fork := true
+
+libraryDependencies += "org.scalameta" %% "munit" % "1.0.4" % Test
+
+val check = TaskKey[Unit]("check", "Verify the shutdown hook could load classes.")
+
+check := Def.uncached {
+  val file = baseDirectory.value / "hook-result.txt"
+  val deadline = System.currentTimeMillis + 30000
+  def content: String = if (file.exists) IO.read(file).trim else ""
+  while (content.isEmpty || content.startsWith("PENDING")) {
+    if (System.currentTimeMillis > deadline)
+      sys.error(s"shutdown hook never completed: '$content'")
+    Thread.sleep(500)
+  }
+  val result = content
+  if (!result.startsWith("OK")) sys.error(s"shutdown hook failed: '$result'")
+}
diff --git a/sbt-app/src/sbt-test/tests/fork-shutdown-hook/src/test/scala/LazilyLoaded.scala b/sbt-app/src/sbt-test/tests/fork-shutdown-hook/src/test/scala/LazilyLoaded.scala
new file mode 100644
index 000000000..1708e1fc1
--- /dev/null
+++ b/sbt-app/src/sbt-test/tests/fork-shutdown-hook/src/test/scala/LazilyLoaded.scala
@@ -0,0 +1,5 @@
+package repro
+
+object LazilyLoaded {
+  val marker: String = "loaded"
+}
diff --git a/sbt-app/src/sbt-test/tests/fork-shutdown-hook/src/test/scala/ShutdownHookTest.scala b/sbt-app/src/sbt-test/tests/fork-shutdown-hook/src/test/scala/ShutdownHookTest.scala
new file mode 100644
index 000000000..e741ab970
--- /dev/null
+++ b/sbt-app/src/sbt-test/tests/fork-shutdown-hook/src/test/scala/ShutdownHookTest.scala
@@ -0,0 +1,20 @@
+package repro
+
+import java.nio.file.{ Files, Paths }
+
+class ShutdownHookTest extends munit.FunSuite {
+  test("shutdown hook can load classes after the test run") {
+    val out = Paths.get("hook-result.txt")
+    Files.writeString(out, "PENDING: shutdown hook did not run" + System.lineSeparator)
+    val _ = sys.addShutdownHook {
+      val result =
+        try {
+          Class.forName("repro.LazilyLoaded$")
+          "OK: classloader still works at JVM shutdown"
+        } catch {
+          case t: Throwable => s"FAIL: $t"
+        }
+      Files.writeString(out, result + System.lineSeparator)
+    }
+  }
+}
diff --git a/sbt-app/src/sbt-test/tests/fork-shutdown-hook/test b/sbt-app/src/sbt-test/tests/fork-shutdown-hook/test
new file mode 100644
index 000000000..21544ae1c
--- /dev/null
+++ b/sbt-app/src/sbt-test/tests/fork-shutdown-hook/test
@@ -0,0 +1,2 @@
+> testFull
+> check
diff --git a/sbt-app/src/sbt-test/tests/test-result-logger-api/build.sbt b/sbt-app/src/sbt-test/tests/test-result-logger-api/build.sbt
new file mode 100644
index 000000000..0d4fc4abe
--- /dev/null
+++ b/sbt-app/src/sbt-test/tests/test-result-logger-api/build.sbt
@@ -0,0 +1,28 @@
+import sbt.*
+import sbt.Tests.Output
+import sbt.util.Logger
+
+ThisBuild / scalaVersion := "3.8.4"
+
+val marker = file("test-result-logger-ran")
+val verify = "com.eed3si9n.verify" %% "verify" % "1.0.0"
+
+libraryDependencies += verify % Test
+testFrameworks += new TestFramework("verify.runner.Framework")
+
+Test / testResultLogger := new TestResultLogger:
+  def run(log: Logger, results: Output, taskName: String): Unit =
+    val suiteResults: Iterable[SuiteResult] =
+      results.withoutThrowables.events.values.map(_.withoutThrowables)
+    val suite = suiteResults.head
+    IO.write(
+      marker,
+      Seq(
+        s"overall=${results.overall}",
+        s"suites=${suiteResults.size}",
+        s"suiteResult=${suite.result}",
+        s"passed=${suite.passedCount}",
+        s"failed=${suite.failureCount}",
+        s"errors=${suite.errorCount}"
+      ).mkString("", "\n", "\n")
+    )
diff --git a/sbt-app/src/sbt-test/tests/test-result-logger-api/expected-result b/sbt-app/src/sbt-test/tests/test-result-logger-api/expected-result
new file mode 100644
index 000000000..d4b21c8af
--- /dev/null
+++ b/sbt-app/src/sbt-test/tests/test-result-logger-api/expected-result
@@ -0,0 +1,6 @@
+overall=Passed
+suites=1
+suiteResult=Passed
+passed=1
+failed=0
+errors=0
diff --git a/sbt-app/src/sbt-test/tests/test-result-logger-api/src/test/scala/LoggerApiTest.scala b/sbt-app/src/sbt-test/tests/test-result-logger-api/src/test/scala/LoggerApiTest.scala
new file mode 100644
index 000000000..8b9ce963b
--- /dev/null
+++ b/sbt-app/src/sbt-test/tests/test-result-logger-api/src/test/scala/LoggerApiTest.scala
@@ -0,0 +1,3 @@
+object LoggerApiTest extends verify.BasicTestSuite:
+  test("test result logger receives a suite result"):
+    assert(true)
diff --git a/sbt-app/src/sbt-test/tests/test-result-logger-api/test b/sbt-app/src/sbt-test/tests/test-result-logger-api/test
new file mode 100644
index 000000000..da233b362
--- /dev/null
+++ b/sbt-app/src/sbt-test/tests/test-result-logger-api/test
@@ -0,0 +1,2 @@
+> test
+$ must-mirror expected-result test-result-logger-ran
diff --git a/testing/src/main/scala/sbt/TestReportListener.scala b/testing/src/main/scala/sbt/TestReportListener.scala
index 3870e1786..3dafa81f5 100644
--- a/testing/src/main/scala/sbt/TestReportListener.scala
+++ b/testing/src/main/scala/sbt/TestReportListener.scala
@@ -55,11 +55,12 @@ trait TestsListener extends TestReportListener {
  *   a `Class` strongly references its defining class loader. Holding a `SuiteResult` with a
  *   non-empty `throwables` therefore keeps the test class loader -- and every jar handle it has
  *   open -- alive. That is fine for the duration of the test task, which is where these are
- *   consumed, but anything that outlives the task must drop them first. `TestRecap.collect` does
- *   exactly that before stashing a copy on `State.attributes`; see the note on `TestRecap.recapKey`.
+ *   consumed, but anything that outlives the task must call [[withoutThrowables]] first.
+ *   `TestSummary.append` does exactly that before stashing a copy on `State.attributes`; see the
+ *   note on `TestSummary.entriesKey`.
  *   On Windows a leaked handle makes the underlying jar undeletable (e.g. by `clearCaches`).
  */
-private[sbt] final class SuiteResult(
+final class SuiteResult(
     val result: TestResult,
     val passedCount: Int,
     val failureCount: Int,
@@ -91,13 +92,36 @@ private[sbt] final class SuiteResult(
       pendingCount,
       Nil
     )
+
+  /**
+   * Returns an equivalent result without retaining test-thrown exceptions.
+   *
+   * Use this before retaining test results beyond the lifetime of the test task. See
+   * `throwables` for why retaining those exceptions can keep the test class loader alive.
+   */
+  def withoutThrowables: SuiteResult =
+    if throwables.isEmpty then this
+    else
+      new SuiteResult(
+        result,
+        passedCount,
+        failureCount,
+        errorCount,
+        skippedCount,
+        ignoredCount,
+        canceledCount,
+        pendingCount,
+      )
+
   def +(other: SuiteResult): SuiteResult = {
     val combinedTestResult =
       (result, other.result) match {
-        case (TestResult.Passed, TestResult.Passed) => TestResult.Passed: TestResult
-        case (_, TestResult.Error)                  => TestResult.Error: TestResult
-        case (TestResult.Error, _)                  => TestResult.Error: TestResult
-        case _                                      => TestResult.Failed: TestResult
+        case (TestResult.Empty, TestResult.Empty) => TestResult.Empty: TestResult
+        case (TestResult.Passed | TestResult.Empty, TestResult.Passed | TestResult.Empty) =>
+          TestResult.Passed: TestResult
+        case (_, TestResult.Error) => TestResult.Error: TestResult
+        case (TestResult.Error, _) => TestResult.Error: TestResult
+        case _                     => TestResult.Failed: TestResult
       }
     new SuiteResult(
       combinedTestResult,
@@ -113,7 +137,7 @@ private[sbt] final class SuiteResult(
   }
 }
 
-private[sbt] object SuiteResult {
+object SuiteResult {
 
   /**
    * Computes the overall result and counts for a suite with individual test results in `events`.
diff --git a/testing/src/test/scala/sbt/SuiteResultSpec.scala b/testing/src/test/scala/sbt/SuiteResultSpec.scala
new file mode 100644
index 000000000..bc3068e95
--- /dev/null
+++ b/testing/src/test/scala/sbt/SuiteResultSpec.scala
@@ -0,0 +1,72 @@
+/*
+ * sbt
+ * Copyright 2026, Scala center
+ * Copyright 2011 - 2022, Lightbend, Inc.
+ * Copyright 2008 - 2010, Mark Harrah
+ * Licensed under Apache License 2.0 (see LICENSE)
+ */
+
+package sbt
+
+import sbt.protocol.testing.TestResult
+import verify.BasicTestSuite
+
+object SuiteResultSpec extends BasicTestSuite:
+
+  private def emptySuite(result: TestResult): SuiteResult =
+    new SuiteResult(result, 0, 0, 0, 0, 0, 0, 0)
+
+  private def assertCombined(
+      left: TestResult,
+      right: TestResult
+  )(expected: TestResult): Unit =
+    val actual = (emptySuite(left) + emptySuite(right)).result
+    assert(actual == expected)
+
+  test("SuiteResult combines Passed and Passed as Passed") {
+    assertCombined(TestResult.Passed, TestResult.Passed)(TestResult.Passed)
+  }
+
+  test("SuiteResult combines Passed and Empty as Passed") {
+    assertCombined(TestResult.Passed, TestResult.Empty)(TestResult.Passed)
+    assertCombined(TestResult.Empty, TestResult.Passed)(TestResult.Passed)
+  }
+
+  test("SuiteResult combines Passed and Failed as Failed") {
+    assertCombined(TestResult.Passed, TestResult.Failed)(TestResult.Failed)
+    assertCombined(TestResult.Failed, TestResult.Passed)(TestResult.Failed)
+  }
+
+  test("SuiteResult combines Passed and Error as Error") {
+    assertCombined(TestResult.Passed, TestResult.Error)(TestResult.Error)
+    assertCombined(TestResult.Error, TestResult.Passed)(TestResult.Error)
+  }
+
+  test("SuiteResult combines Empty and Empty as Empty") {
+    assertCombined(TestResult.Empty, TestResult.Empty)(TestResult.Empty)
+  }
+
+  test("SuiteResult combines Empty and Failed as Failed") {
+    assertCombined(TestResult.Empty, TestResult.Failed)(TestResult.Failed)
+    assertCombined(TestResult.Failed, TestResult.Empty)(TestResult.Failed)
+  }
+
+  test("SuiteResult combines Empty and Error as Error") {
+    assertCombined(TestResult.Empty, TestResult.Error)(TestResult.Error)
+    assertCombined(TestResult.Error, TestResult.Empty)(TestResult.Error)
+  }
+
+  test("SuiteResult combines Failed and Failed as Failed") {
+    assertCombined(TestResult.Failed, TestResult.Failed)(TestResult.Failed)
+  }
+
+  test("SuiteResult combines Failed and Error as Error") {
+    assertCombined(TestResult.Failed, TestResult.Error)(TestResult.Error)
+    assertCombined(TestResult.Error, TestResult.Failed)(TestResult.Error)
+  }
+
+  test("SuiteResult combines Error and Error as Error") {
+    assertCombined(TestResult.Error, TestResult.Error)(TestResult.Error)
+  }
+
+end SuiteResultSpec
diff --git a/worker/src/jdk17/java/sbt/internal/worker1/JdkCompat.java b/worker/src/jdk17/java/sbt/internal/worker1/JdkCompat.java
new file mode 100644
index 000000000..1e29477d2
--- /dev/null
+++ b/worker/src/jdk17/java/sbt/internal/worker1/JdkCompat.java
@@ -0,0 +1,23 @@
+/*
+ * 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.internal.worker1;
+
+import java.io.IOException;
+import java.net.StandardProtocolFamily;
+import java.net.UnixDomainSocketAddress;
+import java.nio.channels.SocketChannel;
+import java.nio.file.Path;
+
+public class JdkCompat {
+  public static SocketChannel connectUnixSocket(Path socketPath) throws IOException {
+    SocketChannel client = SocketChannel.open(StandardProtocolFamily.UNIX);
+    client.connect(UnixDomainSocketAddress.of(socketPath));
+    return client;
+  }
+}
diff --git a/worker/src/main/java/sbt/internal/worker1/DuplexChannels.java b/worker/src/main/java/sbt/internal/worker1/DuplexChannels.java
new file mode 100644
index 000000000..8c2629826
--- /dev/null
+++ b/worker/src/main/java/sbt/internal/worker1/DuplexChannels.java
@@ -0,0 +1,91 @@
+/*
+ * 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.internal.worker1;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+
+/**
+ * java.nio.channels.Channels.newInputStream/newOutputStream both synchronize on the channel's
+ * blockingLock() for the duration of each blocking call, so a thread parked in a blocking read
+ * holds that lock for as long as the read blocks, and a concurrent writer on the same channel can
+ * never acquire it. These factories talk to the channel directly instead, so a SocketChannel can
+ * safely be read and written from different threads at the same time.
+ */
+public final class DuplexChannels {
+  private DuplexChannels() {}
+
+  public static OutputStream newOutputStream(SocketChannel ch) {
+    return new OutputStream() {
+      @Override
+      public void write(int b) throws IOException {
+        ByteBuffer bb = ByteBuffer.wrap(new byte[] {(byte) b});
+        while (bb.hasRemaining()) ch.write(bb);
+      }
+
+      @Override
+      public void write(byte[] b, int off, int len) throws IOException {
+        ByteBuffer bb = ByteBuffer.wrap(b, off, len);
+        while (bb.hasRemaining()) ch.write(bb);
+      }
+    };
+  }
+
+  public static InputStream newInputStream(SocketChannel ch) {
+    return new InputStream() {
+      @Override
+      public int read() throws IOException {
+        ByteBuffer bb = ByteBuffer.allocate(1);
+        int n = ch.read(bb);
+        return n <= 0 ? -1 : (bb.get(0) & 0xff);
+      }
+
+      @Override
+      public int read(byte[] b, int off, int len) throws IOException {
+        if (len == 0) return 0;
+        return ch.read(ByteBuffer.wrap(b, off, len));
+      }
+    };
+  }
+
+  /**
+   * Wraps a connected SocketChannel as a Socket backed by {@link #newInputStream}/{@link
+   * #newOutputStream}.
+   */
+  public static Socket newSocket(SocketChannel ch) {
+    return new Socket() {
+      private final InputStream in = newInputStream(ch);
+      private final OutputStream out = newOutputStream(ch);
+
+      @Override
+      public InputStream getInputStream() {
+        return in;
+      }
+
+      @Override
+      public OutputStream getOutputStream() {
+        return out;
+      }
+
+      @Override
+      public void close() throws IOException {
+        ch.close();
+      }
+
+      @Override
+      public boolean isClosed() {
+        return !ch.isOpen();
+      }
+    };
+  }
+}
diff --git a/worker/src/main/java/sbt/internal/worker1/JdkCompat.java b/worker/src/main/java/sbt/internal/worker1/JdkCompat.java
new file mode 100644
index 000000000..e3f7b50a1
--- /dev/null
+++ b/worker/src/main/java/sbt/internal/worker1/JdkCompat.java
@@ -0,0 +1,25 @@
+/*
+ * 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.internal.worker1;
+
+import java.io.IOException;
+import java.nio.channels.SocketChannel;
+import java.nio.file.Path;
+
+/**
+ * Base (Java 8) fallback. The Multi-Release variant under src/jdk17/java implements this using JDK
+ * 16+ Unix domain socket APIs (StandardProtocolFamily.UNIX, UnixDomainSocketAddress); it's the one
+ * actually loaded when the worker runs on Java 17+.
+ */
+public class JdkCompat {
+  public static SocketChannel connectUnixSocket(Path socketPath) throws IOException {
+    throw new UnsupportedOperationException(
+        "Unix domain sockets require Java 16+; this worker JVM is running on an older version");
+  }
+}
diff --git a/worker/src/main/java/sbt/internal/worker1/WorkerMain.java b/worker/src/main/java/sbt/internal/worker1/WorkerMain.java
index 0caa6d04d..7cd0617f6 100644
--- a/worker/src/main/java/sbt/internal/worker1/WorkerMain.java
+++ b/worker/src/main/java/sbt/internal/worker1/WorkerMain.java
@@ -8,6 +8,22 @@
 
 package sbt.internal.worker1;
 
+import java.io.ByteArrayOutputStream;
+import java.io.InputStream;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.lang.reflect.Method;
+import java.net.InetAddress;
+import java.net.MalformedURLException;
+import java.net.Socket;
+import java.net.URL;
+import java.net.URLClassLoader;
+import java.nio.channels.SocketChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Scanner;
 import org.scalasbt.shadedgson.com.google.gson.Gson;
 import org.scalasbt.shadedgson.com.google.gson.GsonBuilder;
 import org.scalasbt.shadedgson.com.google.gson.JsonElement;
@@ -15,21 +31,6 @@ import org.scalasbt.shadedgson.com.google.gson.JsonObject;
 import org.scalasbt.shadedgson.com.google.gson.JsonParser;
 import org.scalasbt.shadedgson.com.google.gson.JsonPrimitive;
 import org.scalasbt.shadedgson.com.google.gson.typeadapters.RuntimeTypeAdapterFactory;
-import java.io.ByteArrayOutputStream;
-import java.io.InputStream;
-import java.io.IOException;
-import java.io.PrintStream;
-import java.net.InetAddress;
-import java.net.Socket;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.lang.reflect.Method;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.net.URLClassLoader;
-import java.util.ArrayList;
-import java.util.Scanner;
 import sbt.testing.*;
 
 /**
@@ -79,6 +80,10 @@ public final class WorkerMain {
         int serverPort = Integer.parseInt(args[1]);
         app.socketWork(serverPort);
         System.exit(0);
+      } else if (args.length == 2 && args[0].equals("--ipc")) {
+        WorkerMain app = new WorkerMain();
+        app.ipcWork(Paths.get(args[1]));
+        System.exit(0);
       } else {
         System.err.println("missing args");
         System.exit(1);
@@ -124,6 +129,16 @@ public final class WorkerMain {
     }
   }
 
+  void ipcWork(Path socketPath) throws Exception {
+    SocketChannel client = JdkCompat.connectUnixSocket(socketPath);
+    this.jsonOut = new PrintStream(DuplexChannels.newOutputStream(client), true, "UTF-8");
+    this.inScanner = new Scanner(DuplexChannels.newInputStream(client), "UTF-8");
+    if (this.inScanner.hasNextLine()) {
+      String line = this.inScanner.nextLine();
+      process(line);
+    }
+  }
+
   /** This processes single request of supposed JSON line. */
   void process(String json) throws Exception {
     JsonElement elem = JsonParser.parseString(json);
@@ -173,12 +188,11 @@ public final class WorkerMain {
         throw new RuntimeException("missing jvmRunInfo element");
       }
       RunInfo.JvmRunInfo jvmRunInfo = info.jvmRunInfo;
-      try (URLClassLoader cl = createClassLoader(jvmRunInfo, ClassLoader.getSystemClassLoader())) {
-        Class<?> mainClass = cl.loadClass(jvmRunInfo.mainClass);
-        Method mainMethod = mainClass.getMethod("main", String[].class);
-        String[] mainArgs = jvmRunInfo.args.stream().toArray(String[]::new);
-        mainMethod.invoke(null, (Object) mainArgs);
-      }
+      URLClassLoader cl = createClassLoader(jvmRunInfo, ClassLoader.getSystemClassLoader());
+      Class<?> mainClass = cl.loadClass(jvmRunInfo.mainClass);
+      Method mainMethod = mainClass.getMethod("main", String[].class);
+      String[] mainArgs = jvmRunInfo.args.stream().toArray(String[]::new);
+      mainMethod.invoke(null, (Object) mainArgs);
     } else {
       throw new RuntimeException("only jvm is supported");
     }
@@ -192,9 +206,8 @@ public final class WorkerMain {
       if (jvmRunInfo.classpath.isEmpty()) {
         ForkTestMain.main(id, info, this.jsonOut, parent);
       } else {
-        try (URLClassLoader cl = createClassLoader(jvmRunInfo, parent)) {
-          ForkTestMain.main(id, info, this.jsonOut, cl);
-        }
+        URLClassLoader cl = createClassLoader(jvmRunInfo, parent);
+        ForkTestMain.main(id, info, this.jsonOut, cl);
       }
     } else {
       throw new RuntimeException("only jvm is supported");
diff --git a/worker/src/test/scala/sbt/internal/worker1/WorkerTest.scala b/worker/src/test/scala/sbt/internal/worker1/WorkerTest.scala
index 8285f3997..51b70559b 100644
--- a/worker/src/test/scala/sbt/internal/worker1/WorkerTest.scala
+++ b/worker/src/test/scala/sbt/internal/worker1/WorkerTest.scala
@@ -1,10 +1,23 @@
 package sbt.internal.worker1
 
+import java.net.{ StandardProtocolFamily, UnixDomainSocketAddress }
+import java.nio.channels.ServerSocketChannel
+import scala.util.Using
 import sbt.io.IO
 
 object WorkerTest extends verify.BasicTestSuite:
   val main = WorkerMain()
 
+  test("JDK UNIX domain socket connects via Multi-Release JAR"):
+    IO.withTemporaryDirectory: dir =>
+      val path = dir.toPath.resolve("test.sock")
+      Using.resource(ServerSocketChannel.open(StandardProtocolFamily.UNIX)): server =>
+        server.bind(UnixDomainSocketAddress.of(path))
+        Using.resource(JdkCompat.connectUnixSocket(path)): client =>
+          Using.resource(server.accept()): accepted =>
+            assert(client.isConnected)
+            assert(accepted != null)
+
   test("process") {
     val u0 = IO.classLocationPath(classOf[example.Hello]).toUri()
     val u1 = IO.classLocationPath(classOf[scala.quoted.Quotes]).toUri()