fix(broker): sweep stale attached sessions; replace pre-v3 broker in place on upgrade

CHANGELOG.md

@@ -2,6 +2,12 @@
 
 ## Unreleased
 
+### Changed
+- Broker protocol bumped to v3 (no client-facing wire change). A pre-v3 broker still running after a package upgrade now fails the new health probe and is replaced in place, so the stale-session sweep below takes effect immediately on upgrade rather than only after the next natural broker restart. `EnsureRunning` also waits for the port to be released after shutting down its own recorded broker, so a same-port in-place replacement (which a protocol bump triggers) completes in a single call instead of bailing as "port in use" while the just-closed socket lingers.
+
+### Fixed
+- The keepalive broker now sweeps stale attached sessions instead of remembering them forever. A crashed or force-killed editor never sends a detach, so its session lingered in the broker's attached-session set indefinitely and defeated the fast-fail gate for client requests: new requests were queued and held up to the 300s hold deadline instead of failing fast once the only backend was gone. Sessions now carry a last-seen timestamp (refreshed on attach and every pull) and are swept after a staleness window (default = the 300s hold deadline, so a domain-reload/compile gap never evicts a live-but-reloading session; overridable via `FUNPLAY_BROKER_SESSION_STALE_MS`). The sweep skips the session whose long-poll is currently parked (provably alive) and reconciles a swept session's in-flight work exactly like an explicit detach — failing still-queued requests only when no backend remains, so a dead session swept while a healthy one is connected leaves those requests for the healthy session.
+
 ## [0.5.3] - 2026-07-18
 
 ### Added

Editor/MCP/Server/Broker/MCPBrokerProcessManager.cs

@@ -66,7 +66,20 @@ internal static bool EnsureRunning(int port, string monoPathOverride, MCPBrokerR
                         var shutdownAccepted = SendShutdown(existing.Port, existing.Token);
                         WaitForExit(existing.Pid, 2500);
                         if (shutdownAccepted)
+                        {
                             KillVerifiedProcess(existing.Pid);
+
+                            // The port frees a moment after the process exits (socket
+                            // teardown / TIME_WAIT). When the replacement reuses the same
+                            // port -- e.g. a protocol bump on a package upgrade forces this
+                            // broker to be replaced in place -- wait for it to actually free
+                            // so we bind it on this same call instead of bailing below.
+                            if (existing.Port == port)
+                            {
+                                for (var i = 0; i < 30 && IsTcpPortOpen(port); i++)
+                                    Thread.Sleep(100);
+                            }
+                        }
                     }
 
                     DeletePidFile(paths.PidFilePath);

Editor/MCP/Server/Broker/MCPBrokerProtocol.cs

@@ -7,7 +7,11 @@ internal static class MCPBrokerProtocol
         // v2: pull responses carry AcceptSseHeader (client's Accept: text/event-stream),
         //     push requests may carry ContentTypeHeader to override the client-facing
         //     response content type (used for SSE-piggybacked notifications).
-        public const int Version = 2;
+        // v3: no client-facing wire change. The broker now sweeps stale attached sessions
+        //     (crashed editors that never detached). Bumped so a pre-v3 broker still running
+        //     after a package upgrade fails the health probe and is replaced by the new one
+        //     (see MCPBrokerProcessManager.EnsureRunning's upgrade-cleanup path).
+        public const int Version = 3;
         public const string Name = "funplay-unity-mcp-broker";
         public const string HealthPath = "/_funplay/broker/health";
         public const string AttachPath = "/_funplay/broker/attach";

Editor/MCP/Server/Broker/keepalive-broker.cs.txt

@@ -17,7 +17,10 @@ namespace Funplay.UnityMcp.Broker
 {
     internal static class Program
     {
-        private const int ProtocolVersion = 2;
+        // Bumped 2 -> 3 for the stale-session sweep below (no client-facing wire change). A
+        // pre-v3 broker left running after a package upgrade fails the new health probe and is
+        // replaced with a freshly compiled one. Keep in sync with MCPBrokerProtocol.Version.
+        private const int ProtocolVersion = 3;
         private const int PullTimeoutMs = 25000;
         private const long HoldDeadlineMs = 300000;
         private const int MaxHeaderBytes = 64 * 1024;
@@ -38,13 +41,24 @@ namespace Funplay.UnityMcp.Broker
         private static readonly object Gate = new object();
         private static readonly Dictionary<long, PendingRequest> Pending = new Dictionary<long, PendingRequest>();
         private static readonly List<long> Queue = new List<long>();
-        private static readonly HashSet<string> AttachedSessions = new HashSet<string>();
+        // session -> last-seen epoch ms (attach and pull act as the heartbeat). Stale entries are
+        // swept in SweepLoop so a crashed editor -- which never sends detach -- cannot linger
+        // forever and defeat the fast-fail gate in HandleClientRequest.
+        private static readonly Dictionary<string, long> AttachedSessions = new Dictionary<string, long>();
         private static Waiter WaitingPull;
         private static TcpListener Listener;
         private static string Token;
         private static long Sequence;
         private static bool Stopping;
 
+        // A session that has not pulled within this window is treated as dead and swept.
+        // Defaults to HoldDeadlineMs: we already hold a client request up to that long waiting
+        // for a session to (re)appear, so a session silent longer than that is definitely gone.
+        // Must stay >= the longest domain-reload/compile gap, otherwise a reloading editor's
+        // session would be swept mid-reload and requests would fail instead of waiting for it.
+        // Overridable via FUNPLAY_BROKER_SESSION_STALE_MS (used by tests to shorten the window).
+        private static long SessionStaleMs = HoldDeadlineMs;
+
         private sealed class PendingRequest
         {
             public TcpClient Client;
@@ -87,6 +101,11 @@ namespace Funplay.UnityMcp.Broker
             if (port <= 0)
                 port = 8765;
 
+            var staleEnv = Environment.GetEnvironmentVariable("FUNPLAY_BROKER_SESSION_STALE_MS");
+            long parsedStale;
+            if (!string.IsNullOrEmpty(staleEnv) && long.TryParse(staleEnv, out parsedStale) && parsedStale > 0)
+                SessionStaleMs = parsedStale;
+
             if (string.IsNullOrEmpty(Token))
             {
                 Log("fatal: missing broker token");
@@ -296,7 +315,7 @@ namespace Funplay.UnityMcp.Broker
         {
             lock (Gate)
             {
-                AttachedSessions.Add(session);
+                AttachedSessions[session] = NowMs();
 
                 if (WaitingPull != null)
                 {
@@ -346,7 +365,7 @@ namespace Funplay.UnityMcp.Broker
                 return;
 
             lock (Gate)
-                AttachedSessions.Add(session);
+                AttachedSessions[session] = NowMs();
         }
 
         private static void HandleDetach(string session)
@@ -355,28 +374,39 @@ namespace Funplay.UnityMcp.Broker
                 return;
 
             lock (Gate)
-            {
-                AttachedSessions.Remove(session);
+                RemoveSessionLocked(session);
+        }
 
-                if (WaitingPull != null && string.Equals(WaitingPull.Session, session, StringComparison.Ordinal))
-                {
-                    TryWriteNoContent(WaitingPull.Stream);
-                    SafeClose(WaitingPull.Client);
-                    WaitingPull = null;
-                }
+        // Removes a session and reconciles its in-flight work. Caller must hold Gate.
+        // Used by both explicit detach and the staleness sweep in SweepLoop.
+        private static void RemoveSessionLocked(string session)
+        {
+            AttachedSessions.Remove(session);
 
-                foreach (var item in Pending)
-                {
-                    var pending = item.Value;
-                    if (!string.Equals(pending.ActiveSession, session, StringComparison.Ordinal))
-                        continue;
+            if (WaitingPull != null && string.Equals(WaitingPull.Session, session, StringComparison.Ordinal))
+            {
+                TryWriteNoContent(WaitingPull.Stream);
+                SafeClose(WaitingPull.Client);
+                WaitingPull = null;
+            }
 
-                    pending.ActiveSession = null;
-                    pending.Redelivered = true;
-                    if (!Queue.Contains(item.Key))
-                        Queue.Insert(0, item.Key);
-                }
+            foreach (var item in Pending)
+            {
+                var pending = item.Value;
+                if (!string.Equals(pending.ActiveSession, session, StringComparison.Ordinal))
+                    continue;
+
+                pending.ActiveSession = null;
+                pending.Redelivered = true;
+                if (!Queue.Contains(item.Key))
+                    Queue.Insert(0, item.Key);
+            }
 
+            // Fail still-queued (never-dispatched) requests only when no backend remains to
+            // serve them. When another session is still attached -- e.g. a dead session swept
+            // while a healthy one is connected -- leave them queued for that session to pull.
+            if (AttachedSessions.Count == 0)
+            {
                 var unavailable = new List<long>();
                 foreach (var item in Pending)
                 {
@@ -553,6 +583,23 @@ namespace Funplay.UnityMcp.Broker
                         }
                         SafeClose(pending.Client);
                     }
+
+                    var staleSessions = new List<string>();
+                    foreach (var item in AttachedSessions)
+                    {
+                        // The session whose pull is currently parked is provably alive even if
+                        // its last-seen is old (a parked long-poll does not refresh the timestamp).
+                        if (WaitingPull != null && string.Equals(item.Key, WaitingPull.Session, StringComparison.Ordinal))
+                            continue;
+                        if (now - item.Value > SessionStaleMs)
+                            staleSessions.Add(item.Key);
+                    }
+
+                    foreach (var session in staleSessions)
+                    {
+                        Log("session " + session + " swept after " + SessionStaleMs + "ms without activity");
+                        RemoveSessionLocked(session);
+                    }
                 }
             }
         }

Tests/Editor/MCPBrokerTransportTests.cs

@@ -23,6 +23,25 @@ namespace Funplay.Editor
 {
     public sealed class MCPBrokerTransportTests
     {
+        private string _priorStaleEnv;
+
+        // The staleness override is a process-wide env var inherited by launched broker
+        // processes. Capture/restore it reliably here: a [TearDown] runs even when a
+        // [UnityTest] throws, whereas an in-test try/finally can be skipped when an
+        // exception propagates out of a nested yielded IEnumerator (which would otherwise
+        // leak the override into every later test's broker).
+        [SetUp]
+        public void CaptureStaleEnvOverride()
+        {
+            _priorStaleEnv = Environment.GetEnvironmentVariable(SessionStaleEnvVar);
+        }
+
+        [TearDown]
+        public void RestoreStaleEnvOverride()
+        {
+            Environment.SetEnvironmentVariable(SessionStaleEnvVar, _priorStaleEnv);
+        }
+
         [UnityTest]
         public IEnumerator BrokerProcess_StartsWithHealthTokenAndStops()
         {
@@ -112,6 +131,43 @@ public IEnumerator BrokerProcess_PortChangeStopsRecordedBroker()
             yield return null;
         }
 
+        [UnityTest]
+        public IEnumerator BrokerProcess_ReplacesRecordedBrokerOnSamePortInOneCall()
+        {
+            var root = CreateTempRoot();
+            var paths = CreateBrokerPaths(root);
+            var port = GetFreeTcpPort();
+
+            try
+            {
+                Assume.That(!string.IsNullOrEmpty(MCPBrokerProcessManager.ResolveMono(string.Empty)),
+                    "Unity-bundled Mono is required for broker process tests.");
+
+                Assert.IsTrue(MCPBrokerProcessManager.EnsureRunning(port, string.Empty, paths), MCPBrokerProcessManager.LastError);
+                Assert.IsTrue(MCPBrokerProcessManager.TryGetConnectionInfo(paths, port, out var first));
+
+                // Force the recorded broker to fail the identity probe (bogus pid) while keeping
+                // its real token, so EnsureRunning must shut it down and relaunch on the SAME
+                // port -- the shape of an in-place replacement (e.g. a protocol bump on upgrade).
+                // Without the post-shutdown port-free wait this bails with "port in use" because
+                // the just-closed socket lingers; with it, replacement completes in one call.
+                WriteBrokerState(paths, pid: 999999, port: port, token: first.Token);
+
+                Assert.IsTrue(MCPBrokerProcessManager.EnsureRunning(port, string.Empty, paths), MCPBrokerProcessManager.LastError);
+                Assert.IsTrue(MCPBrokerProcessManager.TryGetConnectionInfo(paths, port, out var second));
+                Assert.IsTrue(MCPBrokerProcessManager.TryProbeBroker(port, second.Token, out var health));
+                Assert.AreEqual(second.Pid, health.Pid);
+                Assert.AreNotEqual(first.Pid, second.Pid, "The recorded broker must be replaced by a new process on the same port.");
+            }
+            finally
+            {
+                MCPBrokerProcessManager.Stop(paths);
+                DeleteTempRoot(root);
+            }
+
+            yield return null;
+        }
+
         [UnityTest]
         public IEnumerator BrokerProcess_DoesNotKillUnverifiedPidFromStaleState()
         {
@@ -353,6 +409,50 @@ public IEnumerator BrokerTransport_DetachMakesNewRequestsFailFast()
             }
         }
 
+        [UnityTest]
+        public IEnumerator BrokerTransport_StaleSessionIsSweptSoNewRequestsFailFast()
+        {
+            var root = CreateTempRoot();
+            var paths = CreateBrokerPaths(root);
+            var port = GetFreeTcpPort();
+
+            try
+            {
+                // Shorten the staleness window for the test; the child broker inherits this env var.
+                // [SetUp]/[TearDown] capture and restore it so it cannot leak into other tests.
+                Environment.SetEnvironmentVariable(SessionStaleEnvVar, "1500");
+
+                Assume.That(!string.IsNullOrEmpty(MCPBrokerProcessManager.ResolveMono(string.Empty)),
+                    "Unity-bundled Mono is required for broker process tests.");
+
+                Assert.IsTrue(MCPBrokerProcessManager.EnsureRunning(port, string.Empty, paths), MCPBrokerProcessManager.LastError);
+                Assert.IsTrue(MCPBrokerProcessManager.TryGetConnectionInfo(paths, port, out var connection));
+
+                // Simulate an editor that attached and then CRASHED: it registers a session (so
+                // AttachedSessions is non-empty) but never pulls and never sends detach.
+                var attachTask = SendAttachAsync(port, connection.Token, Guid.NewGuid().ToString("N"));
+                yield return WaitForTask(attachTask);
+
+                // Wait past the (shortened) staleness window so the sweep evicts the dead session.
+                yield return new WaitForSecondsRealtime(3f);
+
+                // A new client request must now fail fast instead of being queued until the hold deadline.
+                var startedAt = DateTime.UtcNow;
+                var requestTask = SendToolCallAsync(port, "get_editor_state");
+                yield return WaitForTask(requestTask, 4f);
+                var elapsed = DateTime.UtcNow - startedAt;
+
+                Assert.Less(elapsed.TotalSeconds, 2.0, "A crashed (never-detached) session must be swept so new requests fail fast, not hang.");
+                Assert.That(requestTask.Result, Does.Contain("\"code\":-32001"));
+                Assert.That(requestTask.Result, Does.Contain("\"retryable\":true"));
+            }
+            finally
+            {
+                MCPBrokerProcessManager.Stop(paths);
+                DeleteTempRoot(root);
+            }
+        }
+
         [UnityTest]
         public IEnumerator BrokerTransport_DetachRejectsQueuedRequestsBehindInterruptedSession()
         {
@@ -501,6 +601,20 @@ private static async Task<string> SendToolCallAsync(int port, string toolName)
             }
         }
 
+        private const string SessionStaleEnvVar = "FUNPLAY_BROKER_SESSION_STALE_MS";
+
+        private static async Task SendAttachAsync(int port, string token, string session)
+        {
+            using (var client = new HttpClient { Timeout = TimeSpan.FromSeconds(5) })
+            using (var request = new HttpRequestMessage(HttpMethod.Post, "http://127.0.0.1:" + port + "/_funplay/broker/attach"))
+            {
+                request.Headers.TryAddWithoutValidation(MCPBrokerProtocol.TokenHeader, token);
+                request.Headers.TryAddWithoutValidation(MCPBrokerProtocol.SessionHeader, session);
+                request.Content = new StringContent(string.Empty);
+                await client.SendAsync(request);
+            }
+        }
+
         private static IEnumerator WaitForTask(Task task, float timeoutSeconds = 5f)
         {
             var start = Time.realtimeSinceStartup;