chore: bump version to 0.9.9

Multiple browser tabs open to the same server could see workstream
names, states, and content mixed up between workstreams when creating,
closing, and switching tabs rapidly.

Root causes and fixes:
- Global SSE ws_created events were never handled — other tabs never
  learned about new workstreams, causing blank names and stale tab bars
- SSE reconnection assigned all stale panes to the first workstream
  instead of deduplicating; now uses two-pass assignment with tracking
- switchTab left the old EventSource open while reassigning pane.wsId,
  creating a window for events to leak; now disconnects SSE first
- Per-workstream events carried no ws_id — server now stamps ws_id on
  all events via _enqueue (shallow copy); client handleEvent drops
  events with mismatched ws_id as defense-in-depth
- Plan dialog used pane.wsId at resolve time (could drift after tab
  switch); now captures ws_id when the dialog opens
- Global ws_closed could reassign panes before per-ws SSE finished
  draining; now disconnects per-ws SSE immediately on close

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "turnstone"
-version = "0.9.8"
+version = "0.9.9"
 description = "Multi-node AI orchestration platform with tool use, agent routing, and cluster simulation."
 readme = "README.md"
 license = "BUSL-1.1"

tests/test_workstream.py

@@ -730,13 +730,23 @@ class TestWebUIFanOut:
         ui._enqueue({"type": "content", "text": "hello"})  # should not raise
 
     def test_enqueue_single_listener(self):
-        """Single listener receives the event."""
+        """Single listener receives the event with ws_id stamped."""
         from turnstone.server import WebUI
 
         ui = WebUI(ws_id="test")
         q = ui._register_listener()
         ui._enqueue({"type": "content", "text": "hello"})
-        assert q.get_nowait() == {"type": "content", "text": "hello"}
+        assert q.get_nowait() == {"type": "content", "text": "hello", "ws_id": "test"}
+
+    def test_enqueue_does_not_mutate_input(self):
+        """_enqueue must not mutate the caller's dict."""
+        from turnstone.server import WebUI
+
+        ui = WebUI(ws_id="test")
+        ui._register_listener()
+        original = {"type": "content", "text": "hello"}
+        ui._enqueue(original)
+        assert "ws_id" not in original
 
     def test_enqueue_multiple_listeners(self):
         """All registered listeners receive an identical copy."""
@@ -747,12 +757,12 @@ class TestWebUIFanOut:
         q2 = ui._register_listener()
         q3 = ui._register_listener()
 
-        event = {"type": "content", "text": "world"}
-        ui._enqueue(event)
+        ui._enqueue({"type": "content", "text": "world"})
 
-        assert q1.get_nowait() == event
-        assert q2.get_nowait() == event
-        assert q3.get_nowait() == event
+        expected = {"type": "content", "text": "world", "ws_id": "test"}
+        assert q1.get_nowait() == expected
+        assert q2.get_nowait() == expected
+        assert q3.get_nowait() == expected
 
     def test_unregister_stops_delivery(self):
         """After unregister, the queue receives no further events."""
@@ -784,11 +794,10 @@ class TestWebUIFanOut:
         assert fast.qsize() == 0
 
         # Enqueue via fan-out — slow drops (full), fast receives
-        event = {"type": "content", "text": "overflow"}
-        ui._enqueue(event)
+        ui._enqueue({"type": "content", "text": "overflow"})
         assert slow.qsize() == 500  # still full, overflow dropped
         assert fast.qsize() == 1
-        assert fast.get_nowait() == event
+        assert fast.get_nowait() == {"type": "content", "text": "overflow", "ws_id": "test"}
 
     def test_unregister_idempotent(self):
         """Double unregister does not raise."""

turnstone/__init__.py

@@ -1,3 +1,3 @@
 """turnstone - Multi-node AI orchestration platform with tool use, agent routing, and cluster simulation."""
 
-__version__ = "0.9.8"
+__version__ = "0.9.9"

turnstone/channels/discord/bot.py

@@ -180,6 +180,7 @@ class TurnstoneBot:
             server_token_factory=server_token_factory,
         )
 
+        self._commands_synced: bool = False
         self._subscribed_ws: set[str] = set()
         self._sse_tasks: dict[str, asyncio.Task[None]] = {}
         self._streaming: dict[str, StreamingMessage] = {}
@@ -204,12 +205,17 @@ class TurnstoneBot:
         # response message can be re-tracked for multi-turn DM conversations.
         self._notify_reply_channels: dict[str, tuple[discord.abc.Messageable, str]] = {}
 
-        # Shared HTTP client for SSE connections (long-lived, no timeout).
+        # Shared HTTP client for SSE connections.
+        # Read timeout detects half-open connections (server sends ping=5s
+        # keepalives, so 90s is very conservative).
         # Token factory provides auto-rotating JWTs; static token is fallback.
         headers: dict[str, str] = {}
         if api_token and not server_token_factory:
             headers["Authorization"] = f"Bearer {api_token}"
-        self._http_client = httpx.AsyncClient(headers=headers, timeout=None)
+        self._http_client = httpx.AsyncClient(
+            headers=headers,
+            timeout=httpx.Timeout(connect=10.0, read=90.0, write=10.0, pool=10.0),
+        )
 
         intents = discord.Intents.default()
         intents.message_content = True
@@ -230,6 +236,10 @@ class TurnstoneBot:
         async def on_ready() -> None:
             await self._on_ready()
 
+        @self._bot.event
+        async def on_resumed() -> None:
+            await self._on_resumed()
+
     # -- lifecycle -----------------------------------------------------------
 
     async def _setup_hook(self) -> None:
@@ -247,23 +257,57 @@ class TurnstoneBot:
         log.info("discord.setup_hook_complete")
 
     async def _on_ready(self) -> None:
-        """Sync slash commands and recover existing routes."""
+        """Sync slash commands (once) and recover existing routes."""
         import discord
 
         bot = self._bot
         log.info("discord.ready", user=str(bot.user), guild_count=len(bot.guilds))
 
-        if self.config.guild_id:
-            guild = discord.Object(id=self.config.guild_id)
-            bot.tree.copy_global_to(guild=guild)
-            await bot.tree.sync(guild=guild)
-            log.info("discord.commands_synced", guild_id=self.config.guild_id)
-        else:
-            await bot.tree.sync()
-            log.info("discord.commands_synced_global")
+        if not self._commands_synced:
+            if self.config.guild_id:
+                guild = discord.Object(id=self.config.guild_id)
+                bot.tree.copy_global_to(guild=guild)
+                await bot.tree.sync(guild=guild)
+                log.info("discord.commands_synced", guild_id=self.config.guild_id)
+            else:
+                await bot.tree.sync()
+                log.info("discord.commands_synced_global")
+            self._commands_synced = True
 
+        self._purge_dead_sse_tasks("ready")
         await self._recover_routes()
 
+    async def _on_resumed(self) -> None:
+        """Recover dead SSE tasks after a gateway session resume.
+
+        Unlike ``on_ready``, ``on_resumed`` fires when discord.py resumes
+        an existing session after a brief disconnect — ``on_ready`` is NOT
+        called in that case.  Any SSE listener tasks that died during the
+        blip need to be cleaned up and re-subscribed.
+        """
+        self._purge_dead_sse_tasks("resumed")
+        await self._recover_routes()
+
+    def _purge_dead_sse_tasks(self, trigger: str) -> None:
+        """Remove completed/failed SSE tasks so they can be re-subscribed."""
+        dead = [ws_id for ws_id, task in self._sse_tasks.items() if task.done()]
+        for ws_id in dead:
+            task = self._sse_tasks.pop(ws_id)
+            self._subscribed_ws.discard(ws_id)
+            # Retrieve exception to suppress "Task exception was never
+            # retrieved" warnings and log the underlying failure.
+            if not task.cancelled():
+                exc = task.exception()
+                if exc is not None:
+                    log.warning(
+                        "discord.sse_task_failed",
+                        trigger=trigger,
+                        ws_id=ws_id,
+                        error=str(exc),
+                    )
+        if dead:
+            log.info("discord.purged_dead_tasks", trigger=trigger, count=len(dead), ws_ids=dead)
+
     async def _recover_routes(self) -> None:
         """Re-subscribe to event channels for existing discord routes.
 
@@ -361,14 +405,15 @@ class TurnstoneBot:
         """
         import httpx_sse
 
-        # When routing through the console, connect SSE directly to the
-        # assigned server node (node_url from the create response).
-        node_base = await self.router.get_node_url(ws_id)
-        url = f"{node_base}/v1/api/events"
         delay = _SSE_RECONNECT_DELAY
+        url = ""  # set before loop so exception handlers can reference it
 
         while True:
             try:
+                # Re-resolve node URL on each attempt so reconnects pick up
+                # changes after bot restarts or router cache expiry.
+                node_base = await self.router.get_node_url(ws_id)
+                url = f"{node_base}/v1/api/events"
                 # Refresh auth header per-connection (token may have rotated)
                 sse_headers: dict[str, str] | None = None
                 if self._token_factory is not None:
@@ -392,7 +437,13 @@ class TurnstoneBot:
                             ws_id=ws_id,
                             status=status,
                         )
-                        # Fall through to backoff/retry for transient errors.
+                        # Don't try to parse a non-SSE error body —
+                        # fall through to backoff/retry below.
+                        raise httpx.HTTPStatusError(
+                            f"SSE upstream {status}",
+                            request=event_source.response.request,
+                            response=event_source.response,
+                        )
                     delay = _SSE_RECONNECT_DELAY  # reset on successful connect
                     async for sse in event_source.aiter_sse():
                         if sse.event == "message" or not sse.event:
@@ -406,12 +457,27 @@ class TurnstoneBot:
                                 )
                                 continue
                             event = ServerEvent.from_dict(data)
-                            await self._on_ws_event(ws_id, thread, event)
+                            try:
+                                await self._on_ws_event(ws_id, thread, event)
+                            except Exception:
+                                # Discord API failures (rate limits, outages)
+                                # must not kill the SSE connection.
+                                log.warning(
+                                    "discord.event_dispatch_failed",
+                                    ws_id=ws_id,
+                                    exc_info=True,
+                                )
+            except httpx.HTTPStatusError:
+                pass  # already logged above; fall through to backoff
             except httpx.RemoteProtocolError:
                 # Server closed connection (normal on stream_end or shutdown).
                 log.debug("discord.sse_remote_closed", ws_id=ws_id)
             except asyncio.CancelledError:
                 return  # unsubscribe or shutdown
+            except httpx.ReadTimeout:
+                # No data received within read timeout — likely a half-open
+                # connection.  Reconnect to recover.
+                log.info("discord.sse_read_timeout", ws_id=ws_id)
             except (httpx.ConnectError, httpx.ConnectTimeout) as exc:
                 log.warning(
                     "discord.sse_connect_failed",

turnstone/console/server.py

@@ -5580,7 +5580,7 @@ async def admin_list_prompt_policies(request: Request) -> JSONResponse:
     storage, err = require_storage_or_503(request)
     if err:
         return err
-    err = require_permission(request, "admin.prompt_policies")
+    err = require_permission(request, "admin.policies")
     if err:
         return err
 
@@ -5599,7 +5599,7 @@ async def admin_create_prompt_policy(request: Request) -> JSONResponse:
     storage, err = require_storage_or_503(request)
     if err:
         return err
-    err = require_permission(request, "admin.prompt_policies")
+    err = require_permission(request, "admin.policies")
     if err:
         return err
 
@@ -5656,7 +5656,7 @@ async def admin_get_prompt_policy(request: Request) -> JSONResponse:
     storage, err = require_storage_or_503(request)
     if err:
         return err
-    err = require_permission(request, "admin.prompt_policies")
+    err = require_permission(request, "admin.policies")
     if err:
         return err
 
@@ -5676,7 +5676,7 @@ async def admin_update_prompt_policy(request: Request) -> JSONResponse:
     storage, err = require_storage_or_503(request)
     if err:
         return err
-    err = require_permission(request, "admin.prompt_policies")
+    err = require_permission(request, "admin.policies")
     if err:
         return err
 
@@ -5729,7 +5729,7 @@ async def admin_delete_prompt_policy(request: Request) -> JSONResponse:
     storage, err = require_storage_or_503(request)
     if err:
         return err
-    err = require_permission(request, "admin.prompt_policies")
+    err = require_permission(request, "admin.policies")
     if err:
         return err
 

turnstone/server.py

@@ -119,6 +119,11 @@ class WebUI:
         self._ws_turn_content_size: int = 0
 
     def _enqueue(self, data: dict[str, Any]) -> None:
+        # Stamp ws_id on every per-workstream event so the client can
+        # validate it belongs to the pane's current workstream.
+        # Shallow copy to avoid mutating caller's dict (e.g. _pending_approval).
+        if "ws_id" not in data:
+            data = {**data, "ws_id": self.ws_id}
         with self._listeners_lock:
             snapshot = list(self._listeners)
         for lq in snapshot:

turnstone/ui/static/app.js

@@ -310,29 +310,53 @@ Pane.prototype.connectSSE = function (wsId) {
               workstreams[ws.id] = { name: ws.name, state: ws.state };
             });
             renderTabBar();
-            // Reconnect all disconnected panes, reassigning stale ws_ids
+            // Reconnect all disconnected panes, reassigning stale ws_ids.
+            // Two passes: (1) reassign stale panes, (2) reconnect all.
+            // Track assigned ws_ids to avoid multiple panes on the same ws.
+            var remaining = Object.keys(workstreams);
+            if (!remaining.length) {
+              showDashboard();
+              return;
+            }
+            var usedWsIds = {};
             for (var pid in panes) {
-              var p = panes[pid];
-              if (p.wsId && !workstreams[p.wsId]) {
-                var ids = Object.keys(workstreams);
-                if (ids.length) {
-                  p.wsId = ids[0];
-                  p.messagesEl.innerHTML = "";
-                  p.showEmptyState();
-                  p.updateWsName();
-                } else {
-                  showDashboard();
-                  return;
+              if (panes[pid].wsId && workstreams[panes[pid].wsId])
+                usedWsIds[panes[pid].wsId] = true;
+            }
+            for (var pid2 in panes) {
+              var p2 = panes[pid2];
+              if (p2.wsId && !workstreams[p2.wsId]) {
+                var newWsId = null;
+                for (var ri = 0; ri < remaining.length; ri++) {
+                  if (!usedWsIds[remaining[ri]]) {
+                    newWsId = remaining[ri];
+                    break;
+                  }
                 }
+                if (newWsId) {
+                  p2.disconnectSSE();
+                  p2.wsId = newWsId;
+                  usedWsIds[newWsId] = true;
+                  while (p2.messagesEl.firstChild)
+                    p2.messagesEl.removeChild(p2.messagesEl.firstChild);
+                  p2.showEmptyState();
+                  p2.updateWsName();
+                }
+                // else: more panes than workstreams — leave pane stale,
+                // connectSSE below will pick it up or it stays disconnected.
               }
-              if (pid === focusedPaneId) currentWsId = p.wsId;
-              if (!p.evtSource) {
+            }
+            // Pass 2: reconnect all panes and sync focused pane
+            for (var pid3 in panes) {
+              var p3 = panes[pid3];
+              if (pid3 === focusedPaneId) currentWsId = p3.wsId;
+              if (!p3.evtSource && p3.wsId && workstreams[p3.wsId]) {
                 setTimeout(
                   (function (pp) {
                     return function () {
                       pp.connectSSE(pp.wsId);
                     };
-                  })(p),
+                  })(p3),
                   self.retryDelay,
                 );
               }
@@ -357,6 +381,9 @@ Pane.prototype.connectSSE = function (wsId) {
 };
 
 Pane.prototype.handleEvent = function (evt) {
+  // Guard: drop events that belong to a different workstream.
+  // This prevents cross-contamination during tab switches and reconnects.
+  if (evt.ws_id && evt.ws_id !== this.wsId) return;
   var self = this;
   switch (evt.type) {
     case "thinking_start":
@@ -2304,10 +2331,12 @@ function switchTab(wsId) {
     }
   }
 
+  pane.disconnectSSE();
   pane.reset();
   pane.wsId = wsId;
   currentWsId = wsId;
-  pane.messagesEl.innerHTML = "";
+  while (pane.messagesEl.firstChild)
+    pane.messagesEl.removeChild(pane.messagesEl.firstChild);
   pane.showEmptyState();
   pane.updateWsName();
   renderTabBar();
@@ -2957,8 +2986,18 @@ function connectGlobalSSE() {
       for (var id in panes) {
         if (panes[id].wsId === data.ws_id) panes[id].updateWsName();
       }
+    } else if (data.type === "ws_created") {
+      workstreams[data.ws_id] = workstreams[data.ws_id] || {};
+      workstreams[data.ws_id].name = data.name || data.ws_id.slice(0, 6);
+      workstreams[data.ws_id].state = "idle";
+      renderTabBar();
     } else if (data.type === "ws_closed") {
       var wsId = data.ws_id;
+      // Disconnect per-ws SSE on affected panes immediately so stale
+      // events from the dying workstream don't leak into reassigned panes.
+      for (var cid in panes) {
+        if (panes[cid].wsId === wsId) panes[cid].disconnectSSE();
+      }
       delete workstreams[wsId];
       renderTabBar();
       if (data.reason === "evicted") {
@@ -3137,10 +3176,13 @@ function makeCollapsible(el) {
 
 var _planContent = "";
 var _planPaneId = null;
+var _planWsId = null;
 
 function showPlanDialog(content) {
   _planContent = content;
   _planPaneId = focusedPaneId;
+  var paneNow = panes[_planPaneId];
+  _planWsId = paneNow ? paneNow.wsId : currentWsId;
   document.getElementById("plan-content").textContent = content;
   var feedbackEl = document.getElementById("plan-feedback");
   feedbackEl.value = "";
@@ -3186,7 +3228,10 @@ function resolvePlan(defaultFeedback) {
   }
 
   // Critical: fire the API call first — this unblocks the server.
-  var wsId = pane ? pane.wsId : currentWsId;
+  // Use the ws_id captured when the dialog opened, not the current pane
+  // (user may have switched tabs while the dialog was open).
+  var wsId = _planWsId || (pane ? pane.wsId : currentWsId);
+  _planWsId = null;
   authFetch("/v1/api/plan", {
     method: "POST",
     headers: { "Content-Type": "application/json" },

uv.lock

@@ -2485,7 +2485,7 @@ wheels = [
 
 [[package]]
 name = "turnstone"
-version = "0.9.8"
+version = "0.9.9"
 source = { editable = "." }
 dependencies = [
     { name = "alembic" },