diff --git a/metals/src/main/scala/scala/meta/internal/metals/Compilers.scala b/metals/src/main/scala/scala/meta/internal/metals/Compilers.scala
index a18e7b62a05e..aacb0695ef3d 100644
--- a/metals/src/main/scala/scala/meta/internal/metals/Compilers.scala
+++ b/metals/src/main/scala/scala/meta/internal/metals/Compilers.scala
@@ -7,6 +7,7 @@ import java.time.Duration
 import java.util.concurrent.ScheduledExecutorService
 import java.util.concurrent.ScheduledFuture
 import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicLong
 import java.{util => ju}
 
 import scala.annotation.nowarn
@@ -313,6 +314,7 @@ class Compilers(
   override def cancel(): Unit = {
     presentationCompilerCache.invalidateAll()
     presentationCompilerWorksheetsCache.invalidateAll()
+    presentationCompilerGeneration.incrementAndGet()
     worksheetsDigests.clear()
     outlineFilesProvider.clear()
   }
@@ -353,7 +355,11 @@ class Compilers(
     loadCompiler(path).foreach(_.didClose(path.toNIO.toUri()))
   }
 
-  def didFocus(path: AbsolutePath): Future[List[Diagnostic]] = {
+  def didFocus(
+      path: AbsolutePath,
+      retryStale: Boolean = true,
+  ): Future[List[Diagnostic]] = {
+    val generation = presentationCompilerGeneration.get()
     val maybeDiagnostics =
       for (pc <- loadCompiler(path); contents <- buffers.get(path))
         yield {
@@ -385,24 +391,37 @@ class Compilers(
               )
               result
             }
+            .flatMap { result =>
+              if (generation == presentationCompilerGeneration.get())
+                Future.successful(result)
+              else if (retryStale) didFocus(path, retryStale = false)
+              else Future.successful(Nil)
+            }
         }
 
     maybeDiagnostics.getOrElse(Future.successful(List.empty))
   }
 
+  private val presentationCompilerGeneration = new AtomicLong(0)
   private val inFlightDidChange =
     TrieMap.empty[AbsolutePath, CompletableCancelToken]
+  private case class DidChangeRequest(
+      path: AbsolutePath,
+      generation: Long,
+  )
   private val diagnosticsDebouncerDelay: FiniteDuration =
     if (Testing.isEnabled) 0.millis
     else sys.Prop[Int]("metals.errors-delay").option.getOrElse(500).millis
-  private val fileDidChange: BatchedFunction[AbsolutePath, Unit] =
-    BatchedFunction.fromFuture[AbsolutePath, Unit](
-      changedFiles => {
+  private val fileDidChange: BatchedFunction[DidChangeRequest, Unit] =
+    BatchedFunction.fromFuture[DidChangeRequest, Unit](
+      requests => {
         for {
           _ <- sh.sleep(diagnosticsDebouncerDelay)
+          files = requests
+            .groupMapReduce(_.path)(_.generation)(math.max)
           futures =
             for {
-              file <- changedFiles.distinct
+              (file, generation) <- files
               pc <- this.loadCompiler(file).toList
               contents <- buffers.get(file).toList
             } yield {
@@ -426,15 +445,20 @@ class Compilers(
                   pc.didChange(params).asScala
                 }
                 .map { case (timer, reportedDiagnostics) =>
-                  diagnostics.publishDiagnosticsNotAdjusted(
-                    file,
-                    reportedDiagnostics.asScala.toList,
-                  )
-                  metrics.recordEvent(
-                    Event
-                      .duration("diagnostics", timer.elapsed)
-                      .withLanguage(file.toJLanguage)
-                  )
+                  if (generation == presentationCompilerGeneration.get()) {
+                    diagnostics.publishDiagnosticsNotAdjusted(
+                      file,
+                      reportedDiagnostics.asScala.toList,
+                    )
+                    metrics.recordEvent(
+                      Event
+                        .duration("diagnostics", timer.elapsed)
+                        .withLanguage(file.toJLanguage)
+                    )
+                  }
+                }
+                .andThen { case _ =>
+                  inFlightDidChange.remove(file, token)
                 }
             }
           _ <- Future.sequence(futures)
@@ -553,7 +577,9 @@ class Compilers(
   def didChange(path: AbsolutePath): Future[Unit] = {
     if (userConfig().presentationCompilerDiagnostics)
       // Batch/debounce these requests since they can arrive in bursts
-      fileDidChange(Seq(path))
+      fileDidChange(
+        Seq(DidChangeRequest(path, presentationCompilerGeneration.get()))
+      )
     else
       didChangeBSPDiagnostics(path, shouldReturnDiagnostics = false).ignoreValue
   }
@@ -653,7 +679,11 @@ class Compilers(
     // Restart PC for all build targets that depend on this target
     for {
       target <- buildTargets.allInverseDependencies(target)
-      compiler <- buildTargetPCFromCache(target)
+      key <- List(
+        PresentationCompilerKey.ScalaBuildTarget(target),
+        PresentationCompilerKey.JavaBuildTarget(target),
+      )
+      compiler <- cache.get(key).map(_.await)
     } {
       scribe.debug(s"Restarting PC for target ${target.getUri}")
       compiler.restart()
diff --git a/metals/src/main/scala/scala/meta/internal/metals/ConnectionProvider.scala b/metals/src/main/scala/scala/meta/internal/metals/ConnectionProvider.scala
index 7cb7ffb1ea22..3a2230f65951 100644
--- a/metals/src/main/scala/scala/meta/internal/metals/ConnectionProvider.scala
+++ b/metals/src/main/scala/scala/meta/internal/metals/ConnectionProvider.scala
@@ -70,6 +70,7 @@ class ConnectionProvider(
     indexProviders: IndexProviders,
     syncStatusReporter: SyncStatusReporter,
     mbtBuild: () => MbtBuild,
+    refreshMbtStateAfterIndex: () => Future[Unit],
     mbtDebugStarter: () => Option[MbtDebugSessionStarter] = () => None,
 )(implicit ec: ExecutionContextExecutorService, rc: ReportContext)
     extends Indexer(indexProviders, mbtBuild)
@@ -253,17 +254,22 @@ class ConnectionProvider(
       case Some(session) if session.canReloadWorkspace =>
         workDoneProgress.trackProgressFuture(
           "Sync",
-          progress =>
-            for {
-              _ <- session.workspaceReload()
-              _ <- connect(new ImportBuildAndIndex(session), progress)
-            } yield (),
+          progress => reloadAndImport(session, progress).ignoreValue,
           metricName = Some("reload_build_server"),
         )
       case _ =>
         fullConnect()
     }
 
+  private def reloadAndImport(
+      session: BspSession,
+      progress: TaskProgress,
+  ): Future[BuildChange] =
+    for {
+      _ <- session.workspaceReload()
+      buildChange <- connect(new ImportBuildAndIndex(session), progress)
+    } yield buildChange
+
   private def isBspAvailable(buildTool: BuildTool) =
     buildTool.isBspGenerated(folder) || bspGlobalDirectories.exists(
       _.resolve(s"${buildTool.buildServerName}.json").isFile
@@ -285,6 +291,10 @@ class ConnectionProvider(
         else runMbtReimport(mbtImporters)
       runImport.flatMap { _ =>
         bspSession match {
+          case Some(session)
+              if MbtBuildServer.isMbtServer(session.main.name) &&
+                session.canReloadWorkspace =>
+            reloadAndImport(session, progress)
           case Some(session) =>
             connect(new ImportBuildAndIndex(session), progress)
           case None =>
@@ -680,32 +690,27 @@ class ConnectionProvider(
         }
         _ = compilers.cancel()
         buildChange <- index(check, progress)
-        // When testing we need to make sure the classpath is refreshed after mbt.json is generated
-        _ <- {
-          if (MetalsServerConfig.isTesting)
-            refreshMbtTurbineClasspath(session).withInterrupt
-          else
-            Future {
-              refreshMbtTurbineClasspath(session)
-            }.withInterrupt
-        }
+        _ <- refreshMbtStateAfterIndexIfNeeded(session)
       } yield {
         syncStatusReporter.importFinished(focusedDocument.map(_.toURI.toString))
         buildChange
       }
     }
 
-    private def refreshMbtTurbineClasspath(
+    private def refreshMbtStateAfterIndexIfNeeded(
         session: BspSession
-    ): Future[Unit] =
-      if (
-        MbtBuildServer.isMbtServer(session.main.name) &&
-        userConfig.javaSymbolLoader.isTurbineClasspath
-      ) {
-        mbtSymbolSearch.scheduleRecompileTurbineClasspath()
-      } else {
-        Future.unit
-      }
+    ): Interruptable[Unit] = {
+      val refresh =
+        if (MbtBuildServer.isMbtServer(session.main.name))
+          refreshMbtStateAfterIndex()
+        else Future.unit
+
+      if (MetalsServerConfig.isTesting) refresh.withInterrupt
+      else
+        refresh.recover { case error =>
+          scribe.warn("failed to refresh MBT diagnostics", error)
+        }.withInterrupt
+    }
 
     private def saveProjectReferencesInfo(
         bspBuilds: List[BspSession.BspBuild]
diff --git a/metals/src/main/scala/scala/meta/internal/metals/MetalsLspService.scala b/metals/src/main/scala/scala/meta/internal/metals/MetalsLspService.scala
index 7eb1374765ca..18c0af16a926 100644
--- a/metals/src/main/scala/scala/meta/internal/metals/MetalsLspService.scala
+++ b/metals/src/main/scala/scala/meta/internal/metals/MetalsLspService.scala
@@ -1157,20 +1157,31 @@ abstract class MetalsLspService(
   private def refreshAllDiagnostics(): Future[Unit] = {
     refreshDiagnostics(_ => true)
   }
+
+  protected def refreshMbtStateAfterIndex(): Future[Unit] =
+    for {
+      _ <-
+        if (userConfig.javaSymbolLoader.isTurbineClasspath)
+          mbt2.recompileTurbineClasspath()
+        else Future.unit
+      _ = compilers.cancel()
+      _ = diagnostics.reset(buffers.open.toSeq)
+      _ <- refreshAllDiagnostics()
+    } yield ()
+
   protected def refreshDiagnostics(
       isIncludedPath: AbsolutePath => Boolean
-  ): Future[Unit] = {
+  ): Future[Unit] =
     // rerun diagnostics for all open documents
-    val futures =
-      buffers.open.filter(isIncludedPath).map { path =>
+    Future
+      .traverse(buffers.open.filter(isIncludedPath)) { path =>
         for {
           reportedDiagnostics <- compilers.didFocus(path)
           _ = diagnostics
             .publishDiagnosticsNotAdjusted(path, reportedDiagnostics)
         } yield ()
       }
-    Future.sequence(futures).map(_ => ())
-  }
+      .ignoreValue
 
   def resetPresentationCompilers(): Future[Unit] = {
     compilers.restartAll()
diff --git a/metals/src/main/scala/scala/meta/internal/metals/ProjectMetalsLspService.scala b/metals/src/main/scala/scala/meta/internal/metals/ProjectMetalsLspService.scala
index 59e756adcfce..ada08b04d927 100644
--- a/metals/src/main/scala/scala/meta/internal/metals/ProjectMetalsLspService.scala
+++ b/metals/src/main/scala/scala/meta/internal/metals/ProjectMetalsLspService.scala
@@ -237,6 +237,7 @@ class ProjectMetalsLspService(
       this,
       syncStatusReporter,
       () => mbtBuild,
+      () => refreshMbtStateAfterIndex(),
       mbtDebugStarter = () => mbtDebugStarter,
     )
     provider.buildServerPromise.future.onComplete(_ => moduleStatus.refresh())
diff --git a/metals/src/main/scala/scala/meta/internal/metals/mbt/TurbineCompiler.scala b/metals/src/main/scala/scala/meta/internal/metals/mbt/TurbineCompiler.scala
index 83a5c819f2b9..417967d0a737 100644
--- a/metals/src/main/scala/scala/meta/internal/metals/mbt/TurbineCompiler.scala
+++ b/metals/src/main/scala/scala/meta/internal/metals/mbt/TurbineCompiler.scala
@@ -5,6 +5,7 @@ import java.nio.file.Path
 import java.util.Optional
 import java.util.concurrent.ConcurrentLinkedQueue
 import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicLong
 import java.{util => ju}
 import javax.tools.JavaFileManager
 import javax.tools.JavaFileObject
@@ -156,6 +157,8 @@ class TurbineCompiler[T](
   // In this mode, we rely entirely on SOURCE_PATH fallback for updated sources.
   private def isRecompilationDisabled: Boolean =
     debounceDelay.toMillis >= 3600000
+  private val compileGeneration = new AtomicLong(0L)
+  private val compileLock = new Object
 
   private val doCompile =
     BatchedFunction.fromFuture[Unit, TurbineCompileResult](
@@ -166,12 +169,14 @@ class TurbineCompiler[T](
           Future.successful(result)
         } else {
           val toCompile = sourcepathSources()
+          val generation = compileGeneration.get()
           for {
             _ <- sleeper.sleep(debounceDelay)
           } yield {
-            val result = doCompileNow()
-            toCompile.foreach(_.isCompiled.set(true))
-            result
+            doCompileNow(
+              expectedGeneration = generation,
+              markCompiled = toCompile,
+            )
           }
         }
       },
@@ -183,20 +188,31 @@ class TurbineCompiler[T](
     )
   }
 
-  var result = TurbineCompiler.emptyResult
-  def doCompileNow(): TurbineCompileResult = {
-    result = TurbineCompiler.compileClassfiles(
-      allCompilationUnits(),
-      parseUnit,
-      classpath(),
-      progressBars,
-    )
-    cleanup()
-    // Clear deleted binary names after recompile - they are no longer in the compiled output
-    deletedBinaryNames.clear()
-    onIndexingDone()
-    result
-  }
+  @volatile var result = TurbineCompiler.emptyResult
+  private def doCompileNow(
+      expectedGeneration: Long,
+      markCompiled: Seq[SourcepathJavaFileObject],
+  ): TurbineCompileResult =
+    compileLock.synchronized {
+      if (expectedGeneration != compileGeneration.get()) result
+      else {
+        val compiled = TurbineCompiler.compileClassfiles(
+          allCompilationUnits(),
+          parseUnit,
+          classpath(),
+          progressBars,
+        )
+        if (expectedGeneration == compileGeneration.get()) {
+          result = compiled
+          markCompiled.foreach(_.isCompiled.set(true))
+          cleanup()
+          // Clear deleted binary names after recompile - they are no longer in the compiled output
+          deletedBinaryNames.clear()
+          onIndexingDone()
+        }
+        result
+      }
+    }
 
   /**
    * Called when a file is deleted. Tracks the binary names of the deleted classes
@@ -225,8 +241,16 @@ class TurbineCompiler[T](
     deletedBinaryNames.contains(binaryName)
   }
 
-  def compileNow(): Future[TurbineCompileResult] = Future {
-    doCompileNow()
+  def compileNow(): Future[TurbineCompileResult] = {
+    val toCompile = sourcepathSources()
+    val generation = compileGeneration.incrementAndGet()
+    doCompile.cancelAll()
+    Future {
+      doCompileNow(
+        expectedGeneration = generation,
+        markCompiled = toCompile,
+      )
+    }
   }
 
   def scheduleCompile(): Future[TurbineCompileResult] = {
diff --git a/mtags-java/src/main/scala/scala/meta/internal/jpc/JavaPruneCompiler.scala b/mtags-java/src/main/scala/scala/meta/internal/jpc/JavaPruneCompiler.scala
index d6095abae2c9..e668a4e33786 100644
--- a/mtags-java/src/main/scala/scala/meta/internal/jpc/JavaPruneCompiler.scala
+++ b/mtags-java/src/main/scala/scala/meta/internal/jpc/JavaPruneCompiler.scala
@@ -48,6 +48,12 @@ class JavaPruneCompiler(
 ) extends Closeable {
 
   private val isDebugEnabled = reportsLevel == ReportLevel.Debug
+  // Requests are serialized within one Java PC, but different build targets
+  // compile concurrently and must not mutate the same javac name table.
+  private[jpc] val namesTable: Names = new Names(new Context()) {
+    // Task contexts may dispose Names while this compiler still reuses it.
+    override def dispose(): Unit = ()
+  }
 
   // Only used for testing purposes
   var isCacheEnabled = true
@@ -206,22 +212,22 @@ class JavaPruneCompiler(
 
   /**
    * Creates a new context with minimal pre-registration.
-   * Only SharedNames is registered before getTask() - following NetBeans' pattern
-   * where most custom components are registered AFTER getTask() when the
+   * Only the compiler's Names is registered before getTask() - following NetBeans'
+   * pattern where most custom components are registered AFTER getTask() when the
    * classpath is already configured.
    */
   private def hotContext(): Context = {
     val context = new Context()
     try {
-      // Only register SharedNames before getTask() - this is safe because
+      // Only register Names before getTask() - this is safe because
       // Names doesn't depend on classpath configuration
       if (servicesOverrides.names()) {
-        context.put(Names.namesKey, JavaPruneCompiler.sharedNames)
+        context.put(Names.namesKey, namesTable)
       }
     } catch {
       case _: IllegalAccessError =>
         logger.warn(
-          "Failed to pre-register SharedNames. To fix this problem, make sure you include all the required --add-exports VM options. The full list is defined in META-INF/metals-required-vm-options.txt"
+          "Failed to pre-register Names. To fix this problem, make sure you include all the required --add-exports VM options. The full list is defined in META-INF/metals-required-vm-options.txt"
         )
     }
     context
@@ -387,9 +393,3 @@ class JavaPruneCompiler(
   }
 
 }
-
-object JavaPruneCompiler {
-  val sharedNames: Names = new Names(new Context()) {
-    override def dispose(): Unit = ()
-  }
-}
diff --git a/tests/unit/src/test/scala/scala/meta/internal/jpc/JavaPruneCompilerConcurrencySuite.scala b/tests/unit/src/test/scala/scala/meta/internal/jpc/JavaPruneCompilerConcurrencySuite.scala
new file mode 100644
index 000000000000..0ef3b8c47ccd
--- /dev/null
+++ b/tests/unit/src/test/scala/scala/meta/internal/jpc/JavaPruneCompilerConcurrencySuite.scala
@@ -0,0 +1,42 @@
+package scala.meta.internal.jpc
+
+import scala.meta.internal.metals.Configs.JavacServicesOverrides
+import scala.meta.internal.metals.Embedded
+import scala.meta.internal.metals.ReportLevel
+import scala.meta.pc.JavaFileManagerFactory
+import scala.meta.pc.ProgressBars
+
+import munit.AnyFixture
+import org.slf4j.LoggerFactory
+
+class JavaPruneCompilerConcurrencySuite extends munit.FunSuite {
+  private val tmp = new tests.TemporaryDirectoryFixture()
+  override def munitFixtures: Seq[AnyFixture[_]] = List(tmp)
+
+  test("names-table-is-scoped-to-compiler") {
+    val embedded = new Embedded(tmp())
+    val first = newCompiler(embedded)
+    val second = newCompiler(embedded)
+    try {
+      assertNotEquals(
+        first.namesTable,
+        second.namesTable,
+        "Concurrent Java presentation compilers must not share javac Names",
+      )
+    } finally {
+      first.close()
+      second.close()
+    }
+  }
+
+  private def newCompiler(embedded: Embedded): JavaPruneCompiler =
+    new JavaPruneCompiler(
+      logger =
+        LoggerFactory.getLogger(classOf[JavaPruneCompilerConcurrencySuite]),
+      reportsLevel = ReportLevel.Info,
+      javaFileManagerFactory = JavaFileManagerFactory.EMPTY,
+      embedded = embedded,
+      progressBars = ProgressBars.EMPTY,
+      servicesOverrides = JavacServicesOverrides.default,
+    )
+}
diff --git a/tests/unit/src/test/scala/scala/meta/internal/metals/CompilersRaceLspSuite.scala b/tests/unit/src/test/scala/scala/meta/internal/metals/CompilersRaceLspSuite.scala
new file mode 100644
index 000000000000..d2ee97d7d239
--- /dev/null
+++ b/tests/unit/src/test/scala/scala/meta/internal/metals/CompilersRaceLspSuite.scala
@@ -0,0 +1,78 @@
+package scala.meta.internal.metals
+
+import tests.BaseCompletionLspSuite
+
+class CompilersRaceLspSuite extends BaseCompletionLspSuite("compilers-race") {
+
+  override def userConfig: UserConfiguration =
+    super.userConfig.copy(
+      presentationCompilerDiagnostics = true,
+      buildOnChange = false,
+      buildOnFocus = false,
+    )
+
+  private val filename = "a/src/main/scala/a/A.scala"
+  private val goodText =
+    """|package a
+       |object A {
+       |  val value: String = "ok"
+       |}
+       |""".stripMargin
+  private val badText =
+    """|package a
+       |object A {
+       |  val value: String = 1
+       |}
+       |""".stripMargin
+
+  test("cancel-retries-stale-did-focus-diagnostics") {
+    cleanWorkspace()
+
+    for {
+      _ <- initialize(
+        s"""/metals.json
+           |{
+           |  "a": {}
+           |}
+           |/$filename
+           |$goodText
+           |""".stripMargin
+      )
+      _ <- server.didOpen(filename)
+      _ = assertNoDiagnostics()
+      path = server.toPath(filename)
+      _ = server.buffers.put(path, badText)
+      didFocus = server.didFocus(filename)
+      _ = server.buffers.put(path, goodText)
+      _ = server.server.compilers.cancel()
+      _ = server.server.diagnostics.reset(Seq(path))
+      _ <- didFocus
+      _ = assertNoDiagnostics()
+    } yield ()
+  }
+
+  test("cancel-drops-stale-did-change-diagnostics") {
+    cleanWorkspace()
+
+    for {
+      _ <- initialize(
+        s"""/metals.json
+           |{
+           |  "a": {}
+           |}
+           |/$filename
+           |$goodText
+           |""".stripMargin
+      )
+      _ <- server.didOpen(filename)
+      _ = assertNoDiagnostics()
+      path = server.toPath(filename)
+      didChange = server.didChange(filename)(_ => badText)
+      _ = server.buffers.put(path, goodText)
+      _ = server.server.compilers.cancel()
+      _ = server.server.diagnostics.reset(Seq(path))
+      _ <- didChange
+      _ = assertNoDiagnostics()
+    } yield ()
+  }
+}
diff --git a/tests/unit/src/test/scala/scala/meta/internal/metals/mbt/TurbineCompilerConcurrencySuite.scala b/tests/unit/src/test/scala/scala/meta/internal/metals/mbt/TurbineCompilerConcurrencySuite.scala
new file mode 100644
index 000000000000..7da0f64dbf7f
--- /dev/null
+++ b/tests/unit/src/test/scala/scala/meta/internal/metals/mbt/TurbineCompilerConcurrencySuite.scala
@@ -0,0 +1,143 @@
+package scala.meta.internal.metals.mbt
+
+import java.util.concurrent.CancellationException
+import java.util.concurrent.Executors
+import java.util.concurrent.atomic.AtomicInteger
+
+import scala.collection.parallel.mutable.ParArray
+import scala.concurrent.Await
+import scala.concurrent.ExecutionContext
+import scala.concurrent.Future
+import scala.concurrent.Promise
+import scala.concurrent.duration.DurationInt
+import scala.concurrent.duration.FiniteDuration
+import scala.jdk.CollectionConverters._
+
+import scala.meta.internal.metals.Configs.TurbineRecompileDelayConfig
+import scala.meta.internal.metals.EmptyReportContext
+import scala.meta.internal.metals.ReportContext
+import scala.meta.internal.metals.Sleeper
+import scala.meta.pc.ProgressBars
+
+import com.google.turbine.diag.SourceFile
+
+class TurbineCompilerConcurrencySuite extends munit.FunSuite {
+  private val executor = Executors.newFixedThreadPool(4)
+  private implicit val ec: ExecutionContext =
+    ExecutionContext.fromExecutorService(executor)
+  private implicit val rc: ReportContext = EmptyReportContext
+
+  override def afterAll(): Unit = executor.shutdownNow()
+
+  test("compile-now-supersedes-running-scheduled-compile") {
+    val sleeper = new ControllableSleeper()
+    val firstCompileStarted = Promise[Unit]()
+    val releaseFirstCompile = Promise[Unit]()
+    val compileCount = new AtomicInteger()
+    val indexingCount = new AtomicInteger()
+    val compiler = newCompiler(
+      () => {
+        if (compileCount.incrementAndGet() == 1) {
+          firstCompileStarted.trySuccess(())
+          Await.result(releaseFirstCompile.future, 10.seconds)
+          sources("Stale")
+        } else sources("Current")
+      },
+      sleeper,
+      indexingCount,
+    )
+
+    val scheduled = compiler.scheduleCompile()
+    sleeper.release()
+    for {
+      _ <- firstCompileStarted.future
+      current = compiler.compileNow()
+      _ = releaseFirstCompile.trySuccess(())
+      result <- current
+      cancellation <- scheduled.failed
+    } yield {
+      assert(cancellation.isInstanceOf[CancellationException])
+      assertEquals(compileCount.get(), 2)
+      assertEquals(indexingCount.get(), 1)
+      assertContains(result, "Current")
+      assertNotContains(result, "Stale")
+    }
+  }
+
+  test("scheduled-compile-publishes-current-generation") {
+    val sleeper = new ControllableSleeper()
+    val indexingCount = new AtomicInteger()
+    val compiler = newCompiler(
+      () => sources("Scheduled"),
+      sleeper,
+      indexingCount,
+    )
+
+    val scheduled = compiler.scheduleCompile()
+    sleeper.release()
+    scheduled.map { result =>
+      assertEquals(indexingCount.get(), 1)
+      assertContains(result, "Scheduled")
+    }
+  }
+
+  private def newCompiler(
+      allCompilationUnits: () => ParArray[String],
+      sleeper: Sleeper,
+      indexingCount: AtomicInteger,
+  ): TurbineCompiler[String] =
+    new TurbineCompiler[String](
+      allCompilationUnits = allCompilationUnits,
+      parseUnit = name =>
+        Seq(
+          new SourceFile(
+            s"$name.java",
+            s"package test; public class $name {}",
+          )
+        ),
+      classpath = () => Nil,
+      progressBars = ProgressBars.EMPTY,
+      turbineRecompileDelay = () => TurbineRecompileDelayConfig(1.millis),
+      listProtoJavaOutlinesForPackage = _ => Iterator.empty,
+      sleeper = sleeper,
+      onIndexingDone = () => indexingCount.incrementAndGet(),
+      onNewProjectClasspath = _ => (),
+    )
+
+  private def sources(name: String): ParArray[String] =
+    ParArray.fromSpecific(List(name))
+
+  private def assertContains(
+      result: TurbineCompileResult,
+      name: String,
+  ): Unit = {
+    val names = binaryNames(result)
+    assert(
+      names.exists(_.endsWith(name)),
+      names.mkString(", "),
+    )
+  }
+
+  private def assertNotContains(
+      result: TurbineCompileResult,
+      name: String,
+  ): Unit = {
+    val names = binaryNames(result)
+    assert(
+      !names.exists(_.endsWith(name)),
+      names.mkString(", "),
+    )
+  }
+
+  private def binaryNames(result: TurbineCompileResult): Set[String] =
+    result.lowered.symbols().asScala.map(_.binaryName()).toSet
+
+  private class ControllableSleeper extends Sleeper {
+    private val sleeping = Promise[Unit]()
+
+    override def sleep(duration: FiniteDuration): Future[Unit] =
+      sleeping.future
+
+    def release(): Unit = sleeping.trySuccess(())
+  }
+}
diff --git a/tests/unit/src/test/scala/tests/mbt/MbtBuildServerManualImportLspSuite.scala b/tests/unit/src/test/scala/tests/mbt/MbtBuildServerManualImportLspSuite.scala
new file mode 100644
index 000000000000..88408a6304c0
--- /dev/null
+++ b/tests/unit/src/test/scala/tests/mbt/MbtBuildServerManualImportLspSuite.scala
@@ -0,0 +1,96 @@
+package tests.mbt
+
+import scala.concurrent.Future
+import scala.jdk.CollectionConverters._
+import scala.util.Properties
+
+import scala.meta.internal.metals.AutoImportBuildKind
+import scala.meta.internal.metals.Configs.FallbackSourcepathConfig
+import scala.meta.internal.metals.Configs.ReferenceProviderConfig
+import scala.meta.internal.metals.Configs.WorkspaceSymbolProviderConfig
+import scala.meta.internal.metals.ServerCommands
+import scala.meta.internal.metals.UserConfiguration
+import scala.meta.internal.metals.mbt.MbtBuildServer
+
+import tests.BaseCompletionLspSuite
+import tests.BuildInfo
+import tests.MbtJsonBuilder
+
+class MbtBuildServerManualImportLspSuite
+    extends BaseCompletionLspSuite("mbt-build-server-manual-import") {
+
+  override def userConfig: UserConfiguration =
+    super.userConfig.copy(
+      fallbackScalaVersion = Some(BuildInfo.scalaVersion),
+      presentationCompilerDiagnostics = true,
+      buildOnChange = false,
+      buildOnFocus = false,
+      workspaceSymbolProvider = WorkspaceSymbolProviderConfig.mbt,
+      referenceProvider = ReferenceProviderConfig.mbt,
+      fallbackSourcepath = FallbackSourcepathConfig("all-sources"),
+      preferredBuildServer = Some(MbtBuildServer.name),
+      automaticImportBuild = AutoImportBuildKind.Off,
+    )
+
+  if (!Properties.isWin)
+    test("script-import-clears-java-diagnostics") {
+      runScriptImportClearsJavaDiagnostics()
+    }
+
+  private def runScriptImportClearsJavaDiagnostics(): Future[Unit] = {
+    cleanWorkspace()
+    val mainFile = "src/main/java/a/SampleProfileApplication.java"
+    val firstExtraFile = "src/main/java/a/FirstExtra.java"
+    val secondExtraFile = "src/main/java/a/SecondExtra.java"
+    val mbtJson = new MbtJsonBuilder(BuildInfo.scalaVersion)
+      .addJavaDependency("com.google.guava", "guava", "33.5.0-jre")
+      .addNamespace("core", List("src/**"))
+      .build()
+    val script =
+      s"""|#!/bin/sh
+          |sleep 1
+          |printf '%s' '$mbtJson' > "$$MBT_OUTPUT_FILE"
+          |""".stripMargin
+    def fileInput(className: String): String =
+      s"""|package a;
+          |
+          |import com.google.common.collect.ImmutableList;
+          |
+          |public class $className {
+          |  public static ImmutableList<String> names = ImmutableList.of("Alice", "Bob");
+          |}
+          |""".stripMargin
+
+    client.showMessageRequestHandler = params =>
+      if (params.getMessage.startsWith("New MBT"))
+        params.getActions.asScala.find(_.getTitle == "Not now")
+      else None
+
+    for {
+      _ <- initialize(
+        s"""|/build.mbt.sh
+            |$script
+            |/$mainFile
+            |${fileInput("SampleProfileApplication")}
+            |/$firstExtraFile
+            |${fileInput("FirstExtra")}
+            |/$secondExtraFile
+            |${fileInput("SecondExtra")}
+            |""".stripMargin,
+        expectError = true,
+      )
+      _ <- server.didOpen(mainFile)
+      _ <- server.didFocus(mainFile)
+      _ = assert(
+        client.workspaceDiagnostics.nonEmpty,
+        "Expected diagnostics before MBT import",
+      )
+      importBuild = server.executeCommand(ServerCommands.ImportBuild)
+      _ <- server.didOpen(firstExtraFile)
+      _ <- server.didOpen(secondExtraFile)
+      _ <- importBuild
+      _ = assertConnectedToBuildServer("MBT")
+      _ = assertNoDiagnostics()
+    } yield ()
+  }
+}