[2.x] feat: Notify sbtn client when command is queued (#8568)

Fixes #8356

**Problem**

When `sbtn` sends a command while another long-running task (like `console`) is already executing, the client silently blocks with no indication that the command is waiting in a queue.

**Solution**

When a new command arrives via the network channel and another command is currently running, the server now sends an `ExecStatusEvent` notification with status `"Queued"` to the client. The client displays a message like:

```
[info] waiting for: console
```

main-command/src/main/scala/sbt/internal/client/NetworkClient.scala

@@ -691,6 +691,14 @@ class NetworkClient(
           }
         case (`Shutdown`, Some(_))                => Vector.empty
         case (msg, _) if msg.startsWith("build/") => Vector.empty
+        case ("sbt/exec", Some(json)) =>
+          import sbt.protocol.codec.JsonProtocol.given
+          Converter.fromJson[ExecStatusEvent](json) match {
+            case Success(event) if event.status == "Queued" =>
+              event.message.foreach(m => errorStream.println(s"[info] $m"))
+              Vector.empty
+            case _ => Vector.empty
+          }
         case _ =>
           Vector(
             (

main/src/main/scala/sbt/internal/server/NetworkChannel.scala

@@ -131,10 +131,15 @@ final class NetworkChannel(
     def jsonRpcNotify[A: JsonFormat](method: String, params: A): Unit =
       self.jsonRpcNotify(method, params)
 
-    def appendExec(commandLine: String, execId: Option[String]): Boolean =
+    def appendExec(commandLine: String, execId: Option[String]): Boolean = {
+      self.notifyIfQueued(execId)
       self.appendExec(commandLine, execId)
+    }
 
-    def appendExec(exec: Exec): Boolean = self.append(exec)
+    def appendExec(exec: Exec): Boolean = {
+      self.notifyIfQueued(exec.execId)
+      self.append(exec)
+    }
 
     def log: Logger = self.log
     def name: String = self.name
@@ -166,6 +171,24 @@ final class NetworkChannel(
 
   protected def authOptions: Set[ServerAuthentication] = auth
 
+  private def notifyIfQueued(execId: Option[String]): Unit =
+    StandardMain.exchange.currentExec match {
+      case Some(currentExec) =>
+        val event = ExecStatusEvent(
+          "Queued",
+          Some(name),
+          execId,
+          Vector.empty,
+          None,
+          currentExec.commandLine match {
+            case cmd if cmd.length > 50 => Some(s"waiting for: ${cmd.take(50)}...")
+            case cmd                    => Some(s"waiting for: $cmd")
+          }
+        )
+        jsonRpcNotify("sbt/exec", event)
+      case None =>
+    }
+
   override def mkUIThread: (State, CommandChannel) => UITask = (state, command) => {
     if (interactive.get || ContinuousCommands.isInWatch(state, this)) mkUIThreadImpl(state, command)
     else

server-test/src/server-test/queued/build.sbt

@@ -0,0 +1,11 @@
+
+commands += Command.command("slowTask") { state =>
+  Thread.sleep(5000)
+  state
+}
+
+commands += Command.command("quickTask") { state =>
+  state
+}
+
+Global / cancelable := true

server-test/src/test/scala/testpkg/QueuedNotificationTest.scala

@@ -0,0 +1,42 @@
+/*
+ * sbt
+ * Copyright 2011 - 2018, Lightbend, Inc.
+ * Copyright 2008 - 2010, Mark Harrah
+ * Licensed under Apache License 2.0 (see LICENSE)
+ */
+
+package testpkg
+
+import scala.concurrent.duration.*
+import java.util.concurrent.atomic.AtomicInteger
+
+class QueuedNotificationTest extends AbstractServerTest {
+  override val testDirectory: String = "queued"
+  val currentID = new AtomicInteger(2000)
+
+  test("send Queued notification when command is queued behind another") {
+    val slowId = currentID.getAndIncrement()
+    svr.sendJsonRpc(
+      s"""{ "jsonrpc": "2.0", "id": $slowId, "method": "sbt/exec", "params": { "commandLine": "slowTask" } }"""
+    )
+
+    Thread.sleep(500)
+
+    val quickId = currentID.getAndIncrement()
+    svr.sendJsonRpc(
+      s"""{ "jsonrpc": "2.0", "id": $quickId, "method": "sbt/exec", "params": { "commandLine": "quickTask" } }"""
+    )
+
+    assert(svr.waitForString(10.seconds) { s =>
+      s.contains(""""status":"Queued"""") && s.contains(""""waiting for: slowTask"""")
+    })
+
+    assert(svr.waitForString(10.seconds) { s =>
+      s.contains(s""""id":$slowId""") && s.contains(""""status":"Done"""")
+    })
+
+    assert(svr.waitForString(10.seconds) { s =>
+      s.contains(s""""id":$quickId""") && s.contains(""""status":"Done"""")
+    })
+  }
+}