diff --git a/subprojects/tooling-api-impl/src/main/java/com/itsaky/androidide/tooling/impl/ToolingApiServerImpl.kt b/subprojects/tooling-api-impl/src/main/java/com/itsaky/androidide/tooling/impl/ToolingApiServerImpl.kt
index dd5e0b8c26..9116e39908 100644
--- a/subprojects/tooling-api-impl/src/main/java/com/itsaky/androidide/tooling/impl/ToolingApiServerImpl.kt
+++ b/subprojects/tooling-api-impl/src/main/java/com/itsaky/androidide/tooling/impl/ToolingApiServerImpl.kt
@@ -63,8 +63,10 @@ import org.gradle.tooling.internal.consumer.DefaultGradleConnector
 import org.jetbrains.annotations.VisibleForTesting
 import org.slf4j.LoggerFactory
 import java.io.File
+import java.util.Properties
 import java.util.concurrent.CompletableFuture
 import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.locks.ReentrantLock
 import kotlin.concurrent.withLock
 
@@ -75,9 +77,35 @@ import kotlin.concurrent.withLock
  */
 internal class ToolingApiServerImpl : IToolingApiServer {
 	private var client: IToolingApiClient? = null
+
+	// Volatile because the reuse decision crosses threads: each initialize() runs on whichever
+	// commonPool worker CompletableFuture.supplyAsync hands it, and nothing else in this class
+	// establishes a happens-before edge between one call's writes and the next call's reads. A
+	// stale null read here reintroduces ADFA-5589 intermittently.
+	@Volatile
 	private var connector: GradleConnector? = null
+
+	@Volatile
 	private var connection: ProjectConnection? = null
-	private var lastInitParams: InitializeProjectParams? = null
+
+	@Volatile
+	@VisibleForTesting
+	internal var lastInitParams: InitializeProjectParams? = null
+
+	/** The `distributionUrl` the wrapper named when [connection] was opened; null if not a wrapper. */
+	@Volatile
+	private var lastWrapperDistribution: String? = null
+
+	/**
+	 * Whether the open connection has failed a build in a way that suggests it is dead.
+	 *
+	 * Reconnecting is driven from here rather than by dropping the pair at the point of failure,
+	 * so the replacement goes through [getOrConnectProject], which disconnects the old connector
+	 * before it opens a new one.
+	 */
+	@Volatile
+	@VisibleForTesting
+	internal var connectionSuspect: Boolean = false
 
 	@Suppress("ktlint:standard:backing-property-naming")
 	private var _buildCancellationToken: CancellationTokenSource? = null
@@ -88,18 +116,131 @@ internal class ToolingApiServerImpl : IToolingApiServer {
 		set(value) = cancellationTokenAccessLock.withLock { _buildCancellationToken = value }
 
 	/** Whether the project has been initialized or not. */
+	@Volatile
 	var isInitialized: Boolean = false
 		private set
 
-	/** Whether a build or project synchronization is in progress. */
-	private var isBuildInProgress: Boolean = false
+	/**
+	 * Whether a build or project synchronization is in progress, as a CAS rather than a
+	 * read-then-write.
+	 *
+	 * Same reason the fields above are @Volatile: every call runs on whichever commonPool worker
+	 * CompletableFuture.supplyAsync hands it. As a plain Boolean, two concurrent executeTasks calls
+	 * could both read false and both proceed -- two builds against one connection and one
+	 * cancellation token -- and a worker could go on seeing a stale true and refuse every build for
+	 * the life of the server. ([buildCancellationToken] needs nothing here; it already reads and
+	 * writes under its own lock.)
+	 */
+	private val buildInProgress = AtomicBoolean(false)
 
 	/** Whether the server has a live connection to Gradle. */
 	val isConnected: Boolean
 		get() = connector != null || connection != null
 
 	companion object {
+		private const val WRAPPER_PROPERTIES = "gradle/wrapper/gradle-wrapper.properties"
+
 		private val log = LoggerFactory.getLogger(ToolingApiServerImpl::class.java)
+
+		/**
+		 * Whether [a] and [b] name the same connection: the same project directory, served by the
+		 * same Gradle distribution. A connector is bound to those two and nothing else; every other
+		 * field of [InitializeProjectParams] is per-call.
+		 *
+		 * `a == b` cannot stand in for this. [InitializeProjectParams] declares no `equals`, so that
+		 * is reference equality between objects that arrive freshly deserialized on every call, and
+		 * it answered false every time (ADFA-5589).
+		 */
+		@VisibleForTesting
+		internal fun describesSameConnection(
+			a: InitializeProjectParams?,
+			b: InitializeProjectParams,
+		): Boolean =
+			a != null &&
+				sameDirectory(a.directory, b.directory) &&
+				a.gradleDistribution == b.gradleDistribution
+
+		/**
+		 * Whether [a] and [b] name the same directory.
+		 *
+		 * As paths, not as strings: `GradleConnector.forProjectDirectory` takes a [File], so a
+		 * trailing separator -- or `/sdcard` against `/storage/emulated/0`, both live on Android --
+		 * is one connector but two strings. Guessing wrong here only costs a needless reconnect.
+		 */
+		private fun sameDirectory(
+			a: String,
+			b: String,
+		): Boolean {
+			val first = File(a)
+			val second = File(b)
+			return first == second ||
+				runCatching { first.canonicalFile == second.canonicalFile }.getOrDefault(false)
+		}
+
+		/**
+		 * Whether a wrapper connection is still bound to the distribution the wrapper names.
+		 *
+		 * [GradleDistributionParams.WRAPPER] carries no version, so two wrapper params compare equal
+		 * across a wrapper upgrade. The distribution is resolved from `gradle-wrapper.properties`
+		 * inside `connect()` and frozen into the connection, and [Main.checkGradleWrapper] can
+		 * rewrite that file earlier in this same initialize -- so without this the initialize that
+		 * installs a new wrapper is exactly the one that reuses the connection bound to the old one.
+		 */
+		@VisibleForTesting
+		internal fun wrapperStillMatches(
+			params: InitializeProjectParams,
+			recorded: String?,
+			current: String? = wrapperDistributionUrl(params.directory),
+		): Boolean =
+			params.gradleDistribution.type != GradleDistributionType.GRADLE_WRAPPER ||
+				recorded == current
+
+		/** The `distributionUrl` named by [directory]'s Gradle wrapper, or null if unreadable. */
+		@VisibleForTesting
+		internal fun wrapperDistributionUrl(directory: String): String? =
+			runCatching {
+				File(directory, WRAPPER_PROPERTIES)
+					.takeIf(File::isFile)
+					?.inputStream()
+					?.use { stream -> Properties().apply { load(stream) } }
+					?.getProperty("distributionUrl")
+			}.getOrNull()
+	}
+
+	/**
+	 * Whether the connector already open can serve [params] without reconnecting.
+	 *
+	 * Not merely a slow path when false: reconnecting disconnects the open connector, and
+	 * `GradleConnector.disconnect()` sends the running daemon `StopWhenIdle` (ADFA-5589).
+	 */
+	@VisibleForTesting
+	internal fun canReuseConnector(params: InitializeProjectParams): Boolean =
+		connector != null &&
+			connection != null &&
+			!connectionSuspect &&
+			describesSameConnection(lastInitParams, params) &&
+			wrapperStillMatches(params, lastWrapperDistribution)
+
+	/**
+	 * The connection to build on, reconnecting first when the last build said it was dead.
+	 *
+	 * @throws IllegalStateException If nothing has been initialized, which is the caller's bug.
+	 */
+	@VisibleForTesting
+	internal fun connectionForBuild(): ProjectConnection {
+		val params = lastInitParams
+		if (connectionSuspect && params != null) {
+			log.info("Reconnecting to Gradle: the previous build reported a dead connection")
+			return getOrConnectProject(
+				projectDir = File(params.directory),
+				forceConnect = true,
+				initParams = params,
+			).second
+		}
+
+		return checkNotNull(this.connection) {
+			"ProjectConnection has not been initialized. Cannot execute tasks."
+		}
 	}
 
 	@VisibleForTesting
@@ -112,14 +253,31 @@ internal class ToolingApiServerImpl : IToolingApiServer {
 				?: GradleDistributionParams.WRAPPER,
 	): Pair<GradleConnector, ProjectConnection> =
 		withStopWatch("getOrConnectProject") {
-			if (!forceConnect && connector != null && connection != null) {
-				return@withStopWatch connector!! to connection!!
+			// Each field read once into a local. Four separate reads of two fields could pass the
+			// null checks and then throw on the !!: the reconnect below nulls both before it
+			// connects, so unlike before this fix they are null mid-flight on every reconnect, not
+			// only at shutdown.
+			val openConnector = connector
+			val openConnection = connection
+			if (!forceConnect && openConnector != null && openConnection != null) {
+				return@withStopWatch openConnector to openConnection
 			}
 
 			if (forceConnect) {
-				connector?.disconnect()
+				// The local, not the field. Re-reading it here is what the locals above exist to
+				// avoid: a concurrent reconnect can install a new connector between the two reads,
+				// and this would then disconnect that one and drop the one it meant to close.
+				openConnector?.disconnect()
 			}
 
+			// Dropped before the connect, not after it. A connect that throws -- a bad installation
+			// directory, an unreachable distribution -- would otherwise leave the disconnected pair in
+			// place, and the next initialize whose params match would reuse a dead connection and fail
+			// every build with CONNECTION_CLOSED until the server process restarts.
+			this.connector = null
+			this.connection = null
+			this.lastWrapperDistribution = null
+
 			val connector = GradleConnector.newConnector().forProjectDirectory(projectDir)
 			setupConnectorForGradleInstallation(connector, gradleDist)
 
@@ -127,6 +285,13 @@ internal class ToolingApiServerImpl : IToolingApiServer {
 
 			this.connector = connector
 			this.connection = connection
+			this.connectionSuspect = false
+			this.lastWrapperDistribution =
+				if (gradleDist.type == GradleDistributionType.GRADLE_WRAPPER) {
+					wrapperDistributionUrl(projectDir.path)
+				} else {
+					null
+				}
 
 			connector to connection
 		}
@@ -137,11 +302,9 @@ internal class ToolingApiServerImpl : IToolingApiServer {
 		}
 
 	override fun initialize(params: InitializeProjectParams): CompletableFuture<InitializeResult> {
-		return runBuild {
-			val start = System.currentTimeMillis()
-			try {
-				return@runBuild doInitialize(params, start)
-			} catch (err: Throwable) {
+		val start = System.currentTimeMillis()
+		return runBuild(
+			onFailure = { err ->
 				log.error("Failed to initialize project", err)
 				notifyBuildFailure(
 					BuildResult(
@@ -150,8 +313,12 @@ internal class ToolingApiServerImpl : IToolingApiServer {
 						durationMs = System.currentTimeMillis() - start,
 					),
 				)
-				return@runBuild InitializeResult.Failure(getTaskFailureType(err))
-			}
+				// No suspicion of the connection from here. A sync failure is about the project or
+				// its model, and condemning the connection over one costs a cold daemon start.
+				InitializeResult.Failure(getTaskFailureType(err))
+			},
+		) {
+			doInitialize(params, start)
 		}
 	}
 
@@ -179,8 +346,7 @@ internal class ToolingApiServerImpl : IToolingApiServer {
 		}
 
 		val stopWatch = StopWatch("Connection to project")
-		val isReinitializing =
-			connector != null && connection != null && params == lastInitParams
+		val isReinitializing = canReuseConnector(params)
 
 		if (isReinitializing) {
 			log.info("Project is being reinitialized")
@@ -260,8 +426,21 @@ internal class ToolingApiServerImpl : IToolingApiServer {
 	override fun isServerInitialized(): CompletableFuture<Boolean> = CompletableFuture.supplyAsync { isInitialized }
 
 	override fun executeTasks(message: TaskExecutionMessage): CompletableFuture<TaskExecutionResult> {
-		return runBuild {
-			val start = System.currentTimeMillis()
+		val start = System.currentTimeMillis()
+		return runBuild(
+			onFailure = { error ->
+				log.error("Failed to run tasks: {}", message.tasks, error)
+				notifyBuildFailure(
+					result =
+						BuildResult(
+							tasks = message.tasks,
+							buildId = message.buildId,
+							durationMs = System.currentTimeMillis() - start,
+						),
+				)
+				TaskExecutionResult(false, getTaskFailureType(error))
+			},
+		) {
 			if (!isServerInitialized().get()) {
 				log.error("Cannot execute tasks: {}", PROJECT_NOT_INITIALIZED)
 				return@runBuild TaskExecutionResult(false, PROJECT_NOT_INITIALIZED)
@@ -281,10 +460,11 @@ internal class ToolingApiServerImpl : IToolingApiServer {
 
 			Main.checkGradleWrapper()
 
-			val connection =
-				checkNotNull(this.connection) {
-					"ProjectConnection has not been initialized. Cannot execute tasks."
-				}
+			// Reconnects rather than asserting, when the last build said the connection was dead.
+			// A reconnect can itself throw -- an unreachable distribution, a bad installation dir --
+			// and runBuild's catch is what turns that into a classified failure instead of an
+			// exceptionally-completed future.
+			val connection = connectionForBuild()
 
 			val builder = connection.newBuild()
 
@@ -305,28 +485,28 @@ internal class ToolingApiServerImpl : IToolingApiServer {
 
 			try {
 				builder.run()
-				this.buildCancellationToken = null
-				notifyBuildSuccess(
-					result =
-						BuildResult(
-							tasks = message.tasks,
-							buildId = message.buildId,
-							durationMs = System.currentTimeMillis() - start,
-						),
-				)
-				return@runBuild TaskExecutionResult.SUCCESS
 			} catch (error: Throwable) {
-				log.error("Failed to run tasks: {}", message.tasks, error)
-				notifyBuildFailure(
-					result =
-						BuildResult(
-							tasks = message.tasks,
-							buildId = message.buildId,
-							durationMs = System.currentTimeMillis() - start,
-						),
-				)
-				return@runBuild TaskExecutionResult(false, getTaskFailureType(error))
+				// Marked here, not in runBuild's catch. That catch covers the whole action, which
+				// is what makes a setup failure into a classified result instead of an escaped
+				// exception -- but classifyTaskFailure reads *any* IllegalStateException as
+				// CONNECTION_CLOSED, so marking from there condemned the connection over a throw
+				// from checkGradleWrapper, doPrepareBuild or configureFrom. The next build then
+				// disconnected, which sends the running daemon StopWhenIdle: this ticket's own bug,
+				// reached from a setup failure. A build that actually ran against the connection is
+				// the only failure that says anything about it.
+				markSuspectIfConnectionFailure(error)
+				throw error
 			}
+			this.buildCancellationToken = null
+			notifyBuildSuccess(
+				result =
+					BuildResult(
+						tasks = message.tasks,
+						buildId = message.buildId,
+						durationMs = System.currentTimeMillis() - start,
+					),
+			)
+			TaskExecutionResult.SUCCESS
 		}
 	}
 
@@ -435,6 +615,7 @@ internal class ToolingApiServerImpl : IToolingApiServer {
 			this.client = null
 			this.buildCancellationToken = null
 			this.lastInitParams = null
+			this.lastWrapperDistribution = null
 
 			// wait for connections to close
 			connectionCloseFuture.get()
@@ -443,7 +624,22 @@ internal class ToolingApiServerImpl : IToolingApiServer {
 			null
 		}
 
-	private fun getTaskFailureType(error: Throwable): Failure =
+	/**
+	 * Classifies [error]. Pure: it reads nothing and changes nothing.
+	 *
+	 * It briefly marked the connection suspect, which was wrong twice over. The classification is
+	 * deliberately broad -- any [IllegalStateException] reads as CONNECTION_CLOSED -- and
+	 * `initialize`'s catch routes every sync failure through it, so a model builder throwing an
+	 * IllegalStateException condemned a connection that had just been opened successfully. And the
+	 * penalty was never the "one extra reconnect" its comment claimed: reconnecting calls
+	 * `GradleConnector.disconnect()`, which sends the running daemon `StopWhenIdle` -- the cold
+	 * start this whole ticket exists to prevent, on a new trigger.
+	 *
+	 * Recovery now lives at the one site that can tell a dead connection from a bad project: a
+	 * build that failed against it. See [executeTasks].
+	 */
+	@VisibleForTesting
+	internal fun getTaskFailureType(error: Throwable): Failure =
 		when (error) {
 			is BuildException -> BUILD_FAILED
 			is BuildCancelledException -> BUILD_CANCELLED
@@ -455,24 +651,54 @@ internal class ToolingApiServerImpl : IToolingApiServer {
 			else -> UNKNOWN
 		}
 
+	/**
+	 * Marks the connection suspect when [error] says it may be dead, so the next build replaces it
+	 * rather than reusing it. Only [executeTasks]'s build call reaches this; see the note there.
+	 */
+	private fun markSuspectIfConnectionFailure(error: Throwable) {
+		val failure = getTaskFailureType(error)
+		if (failure == CONNECTION_CLOSED || failure == CONNECTION_ERROR) {
+			log.warn("Marking the Gradle connection suspect after {}; the next build reconnects", failure)
+			connectionSuspect = true
+		}
+	}
+
 	private inline fun <T : Any?> supplyAsync(crossinline action: () -> T): CompletableFuture<T> =
 		CompletableFuture.supplyAsync {
 			action()
 		}
 
-	private inline fun <T : Any?> runBuild(crossinline action: () -> T): CompletableFuture<T> =
+	/**
+	 * Runs [action] as the one build in progress, turning anything it throws into [onFailure]'s
+	 * result rather than an exceptionally-completed future.
+	 *
+	 * The catch spans the whole action deliberately. Each caller used to guard only the part that
+	 * talks to Gradle, so everything before it -- resolving a connection, checking the wrapper,
+	 * preparing the build -- escaped as a raw exception, and the client got an ExecutionException
+	 * where it expected a classified failure. That is one bug this file has now had twice, in two
+	 * different statements on the same line, because the fix each time moved the throw instead of
+	 * covering it.
+	 *
+	 * "Build already in progress" is left to throw: it is a caller error rather than a build
+	 * outcome, and it is raised before this takes ownership of the flag.
+	 */
+	private inline fun <T : Any?> runBuild(
+		crossinline onFailure: (Throwable) -> T,
+		crossinline action: () -> T,
+	): CompletableFuture<T> =
 		supplyAsync {
-			if (isBuildInProgress) {
+			if (!buildInProgress.compareAndSet(false, true)) {
 				log.error("Cannot run build, build is already in progress!")
 				throw IllegalStateException("Build is already in progress")
 			}
 
-			isBuildInProgress = true
 			daemonWatcher.onBuildStarted()
 			try {
 				action()
+			} catch (error: Throwable) {
+				onFailure(error)
 			} finally {
-				isBuildInProgress = false
+				buildInProgress.set(false)
 			}
 		}
 
diff --git a/subprojects/tooling-api-impl/src/test/java/com/itsaky/androidide/tooling/impl/GradleDaemonWatcherTest.kt b/subprojects/tooling-api-impl/src/test/java/com/itsaky/androidide/tooling/impl/GradleDaemonWatcherTest.kt
index 2b6ac12aa5..470d042112 100644
--- a/subprojects/tooling-api-impl/src/test/java/com/itsaky/androidide/tooling/impl/GradleDaemonWatcherTest.kt
+++ b/subprojects/tooling-api-impl/src/test/java/com/itsaky/androidide/tooling/impl/GradleDaemonWatcherTest.kt
@@ -213,4 +213,16 @@ class GradleDaemonWatcherTest {
 		/** Every poll attempt, plus the initial schedule. */
 		const val MAX_SCHEDULES = GradleDaemonWatcher.MAX_POLL_ATTEMPTS + 1
 	}
+
+	@Test
+	fun `shutdown stops the scheduler`() {
+		// It had no caller at all, so the watcher's thread outlived server shutdown and an in-flight
+		// poll chain went on scanning descendants for up to a minute -- and onBuildStarted's note
+		// about the scheduler rejecting work after shutdown described a state nothing could reach.
+		val scheduler = mockk<ScheduledExecutorService>(relaxed = true)
+
+		watcher(scheduler = scheduler).shutdown()
+
+		verify(exactly = 1) { scheduler.shutdownNow() }
+	}
 }
diff --git a/subprojects/tooling-api-impl/src/test/java/com/itsaky/androidide/tooling/impl/ToolingApiServerImplTest.kt b/subprojects/tooling-api-impl/src/test/java/com/itsaky/androidide/tooling/impl/ToolingApiServerImplTest.kt
index d0b127de24..a846593702 100644
--- a/subprojects/tooling-api-impl/src/test/java/com/itsaky/androidide/tooling/impl/ToolingApiServerImplTest.kt
+++ b/subprojects/tooling-api-impl/src/test/java/com/itsaky/androidide/tooling/impl/ToolingApiServerImplTest.kt
@@ -2,7 +2,9 @@ package com.itsaky.androidide.tooling.impl
 
 import com.google.common.truth.Truth.assertThat
 import com.itsaky.androidide.tooling.api.messages.BuildId
+import com.itsaky.androidide.tooling.api.messages.GradleDistributionParams
 import com.itsaky.androidide.tooling.api.messages.InitializeProjectParams
+import com.itsaky.androidide.tooling.api.messages.TaskExecutionMessage
 import com.itsaky.androidide.tooling.api.messages.result.InitializeResult
 import com.itsaky.androidide.tooling.api.messages.result.TaskExecutionResult
 import com.itsaky.androidide.tooling.api.messages.result.isSuccessful
@@ -11,14 +13,19 @@ import com.itsaky.androidide.tooling.impl.sync.RootModelBuilder
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.mockkObject
+import io.mockk.mockkStatic
 import io.mockk.spyk
+import io.mockk.unmockkAll
 import io.mockk.verify
 import org.gradle.tooling.GradleConnector
 import org.gradle.tooling.ProjectConnection
+import org.junit.After
 import org.junit.Test
 import org.junit.runner.RunWith
 import org.junit.runners.JUnit4
 import java.io.File
+import java.nio.file.Files
+import java.util.concurrent.CompletableFuture
 import java.util.concurrent.TimeUnit
 
 /**
@@ -29,12 +36,19 @@ class ToolingApiServerImplTest {
 	private fun testInitParams(
 		directory: String = "/does/not/exist",
 		forceSync: Boolean = false,
+		gradleDistribution: GradleDistributionParams = GradleDistributionParams.WRAPPER,
 	) = InitializeProjectParams(
 		directory = directory,
+		gradleDistribution = gradleDistribution,
 		needsGradleSync = forceSync,
 		buildId = BuildId.Unknown,
 	)
 
+	@After
+	fun tearDown() {
+		unmockkAll()
+	}
+
 	private data class MockServer(
 		val server: ToolingApiServerImpl,
 		val connector: GradleConnector,
@@ -50,7 +64,7 @@ class ToolingApiServerImplTest {
 		every {
 			server.getOrConnectProject(
 				projectDir = any(),
-				forceConnect = true,
+				forceConnect = any(),
 				initParams = any(),
 				gradleDist = any(),
 			)
@@ -59,6 +73,178 @@ class ToolingApiServerImplTest {
 		return MockServer(server, connector, connection)
 	}
 
+	@Test
+	fun `GIVEN the same project twice THEN the connector can be reused`() {
+		// Two separately built objects describing the same connection, which is what the client
+		// sends: every initialize request arrives freshly deserialized from JSON-RPC. The guard used
+		// to be `params == lastInitParams`, which is reference equality on this type and so answered
+		// false for exactly this case (ADFA-5589).
+		val first = testInitParams()
+		val second = testInitParams()
+
+		assertThat(first).isNotSameInstanceAs(second)
+		assertThat(ToolingApiServerImpl.describesSameConnection(first, second)).isTrue()
+	}
+
+	@Test
+	fun `GIVEN the same directory spelled differently THEN the connector can be reused`() {
+		// forProjectDirectory takes a File, so a trailing separator is the same connector.
+		assertThat(
+			ToolingApiServerImpl.describesSameConnection(
+				testInitParams(directory = "/does/not/exist"),
+				testInitParams(directory = "/does/not/exist/"),
+			),
+		).isTrue()
+	}
+
+	@Test
+	fun `GIVEN a different project directory THEN the connector cannot be reused`() {
+		// A connector is bound to its project directory, so reusing one across projects would run
+		// the next build against the previous project's connection.
+		assertThat(
+			ToolingApiServerImpl.describesSameConnection(
+				testInitParams(directory = "/does/not/exist"),
+				testInitParams(directory = "/somewhere/else"),
+			),
+		).isFalse()
+	}
+
+	@Test
+	fun `GIVEN a different Gradle distribution THEN the connector cannot be reused`() {
+		// The other thing a connector is bound to. forInstallationDir is what the app actually sends
+		// for the gradleInstallationDir preference, so that is the case worth pinning: reusing a
+		// connector here would silently build with the distribution the user just changed away from.
+		val wrapper = testInitParams()
+		val installation =
+			testInitParams(gradleDistribution = GradleDistributionParams.forInstallationDir("/opt/gradle"))
+
+		assertThat(ToolingApiServerImpl.describesSameConnection(wrapper, installation)).isFalse()
+	}
+
+	@Test
+	fun `GIVEN nothing initialized yet THEN the connector cannot be reused`() {
+		assertThat(ToolingApiServerImpl.describesSameConnection(null, testInitParams())).isFalse()
+	}
+
+	@Test
+	fun `GIVEN the wrapper names a new distribution THEN the connector cannot be reused`() {
+		// Two WRAPPER params compare equal across a wrapper upgrade: the distribution is resolved
+		// from gradle-wrapper.properties inside connect() and frozen into the connection, so reusing
+		// it would keep building with the Gradle the user just upgraded away from.
+		assertThat(
+			ToolingApiServerImpl.wrapperStillMatches(
+				testInitParams(),
+				recorded = "https://example.invalid/gradle-8.14.3-bin.zip",
+				current = "https://example.invalid/gradle-9.0-bin.zip",
+			),
+		).isFalse()
+	}
+
+	@Test
+	fun `GIVEN the wrapper is unchanged THEN the connector can be reused`() {
+		val url = "https://example.invalid/gradle-8.14.3-bin.zip"
+		assertThat(
+			ToolingApiServerImpl.wrapperStillMatches(testInitParams(), recorded = url, current = url),
+		).isTrue()
+	}
+
+	@Test
+	fun `GIVEN a non-wrapper distribution THEN the wrapper properties do not matter`() {
+		// An installation-dir connection does not read gradle-wrapper.properties at all.
+		assertThat(
+			ToolingApiServerImpl.wrapperStillMatches(
+				testInitParams(gradleDistribution = GradleDistributionParams.forInstallationDir("/opt/gradle")),
+				recorded = null,
+				current = "https://example.invalid/gradle-9.0-bin.zip",
+			),
+		).isTrue()
+	}
+
+	@Test
+	fun `GIVEN a wrapper properties file THEN its distribution URL is read`() {
+		val project = Files.createTempDirectory("adfa5589").toFile()
+		val properties = File(project, "gradle/wrapper/gradle-wrapper.properties")
+		properties.parentFile.mkdirs()
+		properties.writeText("distributionUrl=https\\://example.invalid/gradle-8.14.3-bin.zip\n")
+
+		assertThat(ToolingApiServerImpl.wrapperDistributionUrl(project.path))
+			.isEqualTo("https://example.invalid/gradle-8.14.3-bin.zip")
+		assertThat(ToolingApiServerImpl.wrapperDistributionUrl("/does/not/exist")).isNull()
+
+		project.deleteRecursively()
+	}
+
+	@Test
+	fun `GIVEN the connector can be reused WHEN initializing THEN do not force a reconnect`() {
+		val (server) = mockkReusableServer(canReuse = true)
+
+		server.initialize(testInitParams()).get(5, TimeUnit.SECONDS)
+
+		// forceConnect is the whole point of the reuse check: it is what reaches
+		// connector.disconnect(), and that is what sends the running daemon StopWhenIdle. Asserting
+		// describesSameConnection alone would leave this wiring free to invert unnoticed.
+		verify(exactly = 1) {
+			server.getOrConnectProject(
+				projectDir = any(),
+				forceConnect = false,
+				initParams = any(),
+				gradleDist = any(),
+			)
+		}
+	}
+
+	@Test
+	fun `GIVEN the connector cannot be reused WHEN initializing THEN force a reconnect`() {
+		val (server) = mockkReusableServer(canReuse = false)
+
+		server.initialize(testInitParams()).get(5, TimeUnit.SECONDS)
+
+		verify(exactly = 1) {
+			server.getOrConnectProject(
+				projectDir = any(),
+				forceConnect = true,
+				initParams = any(),
+				gradleDist = any(),
+			)
+		}
+	}
+
+	@Test
+	fun `GIVEN a connect that fails THEN the dead connection is not left behind for reuse`() {
+		val server = ToolingApiServerImpl()
+		val connector = mockk<GradleConnector>(relaxed = true)
+		every { connector.forProjectDirectory(any()) } returns connector
+		every { connector.connect() } returns
+			mockk(relaxed = true) andThenThrows RuntimeException("connect failed")
+
+		mockkStatic(GradleConnector::class)
+		every { GradleConnector.newConnector() } returns connector
+
+		server.getOrConnectProject(File("/does/not/exist"), forceConnect = true)
+		assertThat(server.isConnected).isTrue()
+
+		// The reconnect disconnects the open connector before it opens the new one, so a connect
+		// that throws must not leave that dead pair in place: the next initialize whose params match
+		// would reuse it and fail every build with CONNECTION_CLOSED until the server restarts.
+		val reconnect = runCatching { server.getOrConnectProject(File("/does/not/exist"), forceConnect = true) }
+
+		assertThat(reconnect.isFailure).isTrue()
+		assertThat(server.isConnected).isFalse()
+	}
+
+	/** A server whose connect, project directory check and Gradle sync are all stubbed out. */
+	private fun mockkReusableServer(canReuse: Boolean): MockServer {
+		mockkObject(RootModelBuilder)
+		every {
+			RootModelBuilder.build(any(), any())
+		} returns ProjectSyncHelper.cacheFileForProject(File("/does/not/exist"))
+
+		val mocks = mockkToolingServer()
+		every { mocks.server.validateProjectDirectory(any()) } returns null
+		every { mocks.server.canReuseConnector(any()) } returns canReuse
+		return mocks
+	}
+
 	@Test
 	fun `GIVEN any initialization params WHEN project init fails THEN report as failure`() {
 		mockkObject(RootModelBuilder)
@@ -124,4 +310,127 @@ class ToolingApiServerImplTest {
 			RootModelBuilder.build(initParams, any())
 		}
 	}
+
+	@Test
+	fun `GIVEN a connection failure THEN classifying it changes nothing`() {
+		val server = ToolingApiServerImpl()
+		val connector = mockk<GradleConnector>(relaxed = true)
+		every { connector.forProjectDirectory(any()) } returns connector
+		every { connector.connect() } returns mockk(relaxed = true)
+
+		mockkStatic(GradleConnector::class)
+		every { GradleConnector.newConnector() } returns connector
+
+		server.getOrConnectProject(File("/does/not/exist"), forceConnect = true)
+
+		// Classification is pure. It briefly marked the connection suspect, and initialize's catch
+		// routes every sync failure through it -- so a model builder throwing an
+		// IllegalStateException condemned a connection opened moments earlier, and the next build's
+		// reconnect sent the warm daemon StopWhenIdle. ADFA-5589's own bug, on a new trigger.
+		assertThat(server.getTaskFailureType(IllegalStateException("connection closed")))
+			.isEqualTo(TaskExecutionResult.Failure.CONNECTION_CLOSED)
+		assertThat(server.connectionSuspect).isFalse()
+		assertThat(server.isConnected).isTrue()
+	}
+
+	@Test
+	fun `GIVEN a build failed on a dead connection THEN the next build reconnects rather than reusing it`() {
+		val server = ToolingApiServerImpl()
+		val connector = mockk<GradleConnector>(relaxed = true)
+		every { connector.forProjectDirectory(any()) } returns connector
+		val first = mockk<ProjectConnection>(relaxed = true)
+		val second = mockk<ProjectConnection>(relaxed = true)
+		every { connector.connect() } returns first andThen second
+
+		mockkStatic(GradleConnector::class)
+		every { GradleConnector.newConnector() } returns connector
+
+		server.getOrConnectProject(File("/does/not/exist"), forceConnect = true, initParams = testInitParams())
+		// Set directly: only doInitialize writes it, and that needs a real project directory.
+		// connectionForBuild reconnects to the project the last initialize named.
+		server.lastInitParams = testInitParams()
+		assertThat(server.connectionForBuild()).isSameInstanceAs(first)
+
+		// What the flag is for: the connection stays in place and the *next build* replaces it,
+		// which is what reaches connector.disconnect() and releases the old daemon client and its
+		// threads. Nulling the fields at the point of failure skipped that and stranded them.
+		server.connectionSuspect = true
+
+		assertThat(server.connectionForBuild()).isSameInstanceAs(second)
+		assertThat(server.connectionSuspect).isFalse()
+		verify(atLeast = 1) { connector.disconnect() }
+	}
+
+	@Test
+	fun `GIVEN a reconnect that throws THEN the build reports a failure rather than completing exceptionally`() {
+		val (server) = mockkToolingServer()
+		every { server.isServerInitialized() } returns CompletableFuture.completedFuture(true)
+		every { server.connectionForBuild() } throws IllegalStateException("cannot reconnect")
+
+		mockkObject(Main)
+		every { Main.checkGradleWrapper() } returns Unit
+
+		// Everything before the Gradle call used to sit outside the try, so a throw here escaped as
+		// an ExecutionException where the client expected a classified failure -- and the same line
+		// did it twice, first as a checkNotNull and then as the reconnect that replaced it.
+		val result =
+			server
+				.executeTasks(TaskExecutionMessage(tasks = listOf("assembleDebug"), buildId = BuildId.Unknown))
+				.get(5, TimeUnit.SECONDS)
+
+		assertThat(result.isSuccessful).isFalse()
+		assertThat(result.failure).isEqualTo(TaskExecutionResult.Failure.CONNECTION_CLOSED)
+	}
+
+	@Test
+	fun `GIVEN a failure before the build runs THEN the connection is not condemned for it`() {
+		val (server) = mockkToolingServer()
+		every { server.isServerInitialized() } returns CompletableFuture.completedFuture(true)
+
+		mockkObject(Main)
+		// A setup failure, not a build one. classifyTaskFailure reads any IllegalStateException as
+		// CONNECTION_CLOSED, so marking the connection from runBuild's whole-action catch condemned
+		// it over a throw from here -- and the next build's reconnect sends the running daemon
+		// StopWhenIdle, which is the cold start this ticket exists to prevent.
+		every { Main.checkGradleWrapper() } throws IllegalStateException("no wrapper")
+
+		val result =
+			server
+				.executeTasks(TaskExecutionMessage(tasks = listOf("assembleDebug"), buildId = BuildId.Unknown))
+				.get(5, TimeUnit.SECONDS)
+
+		assertThat(result.isSuccessful).isFalse()
+		assertThat(server.connectionSuspect).isFalse()
+	}
+
+	@Test
+	fun `GIVEN a build already running THEN a second is refused rather than sharing its connection`() {
+		// The refusal, not the atomicity. This passes against a read-then-write too -- the first
+		// build has set the flag long before the second reads it -- so it pins the behaviour and
+		// not the CAS. A genuine interleaving is not reproducible on demand, so the compareAndSet
+		// in runBuild is argued from the threading model rather than pinned here.
+		val (server) = mockkToolingServer()
+		every { server.isServerInitialized() } returns CompletableFuture.completedFuture(true)
+
+		val started = java.util.concurrent.CountDownLatch(1)
+		val release = java.util.concurrent.CountDownLatch(1)
+		mockkObject(Main)
+		every { Main.checkGradleWrapper() } answers {
+			started.countDown()
+			release.await(5, TimeUnit.SECONDS)
+			throw IllegalStateException("done")
+		}
+
+		val first = server.executeTasks(TaskExecutionMessage(tasks = listOf("a"), buildId = BuildId.Unknown))
+		assertThat(started.await(5, TimeUnit.SECONDS)).isTrue()
+
+		val second = server.executeTasks(TaskExecutionMessage(tasks = listOf("b"), buildId = BuildId.Unknown))
+		val refused = runCatching { second.get(5, TimeUnit.SECONDS) }.exceptionOrNull()
+
+		release.countDown()
+		first.get(5, TimeUnit.SECONDS)
+
+		assertThat(refused).isNotNull()
+		assertThat(refused).hasCauseThat().hasMessageThat().contains("already in progress")
+	}
 }