fix: collector JWT expiry causes silent workstream data wipe (#126)

* fix: collector JWT expiry causes silent workstream data wipe

The console collector baked a one-time JWT snapshot into its httpx
client headers at startup. After 1 hour (JWT expiry), every poll to
server nodes returned 401. The error JSON was silently parsed as valid
empty data, wiping all workstream state while nodes still appeared
reachable — the cluster showed "10 nodes, 0 workstreams."

Root causes fixed:
- Collector: no auth baked into httpx.Client; per-request headers
  from ServiceTokenManager.token (auto-rotating) or static fallback
- Proxy: same pattern — proxy_client/proxy_sse_client created without
  auth headers; _proxy_auth_headers() injects fresh token per-request
- main(): static token snapshot only passed when no token_manager
  exists, preventing stale JWT from being stored anywhere
- _fetch_node: raise_for_status() before .json() so 401s throw
  instead of returning error JSON as "0 workstreams"
- Auth errors (401/403) logged at warning level for operator visibility

* fix: address PR #126 review — type annotation, regression tests, log messages

Tighten token_manager type from Any to ServiceTokenManager | None.
Add two regression tests verifying 401/403 poll responses preserve
existing workstream data and mark nodes unreachable. Fix misleading
log messages: "jwt_minted" → "token_manager_created" since
ServiceTokenManager mints lazily on first .token access.

tests/test_console.py

@@ -3,7 +3,7 @@
 import asyncio
 import json
 import queue
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 import pytest
 
@@ -264,6 +264,57 @@ class TestCollectorPolling:
         assert q.empty()
         assert len(c._nodes["node-a"].workstreams) == 0
 
+    def test_poll_401_preserves_workstreams_and_marks_unreachable(self):
+        """A 401 from the server must NOT wipe workstream data."""
+        c = _make_collector()
+        c._nodes["node-a"] = NodeSnapshot(
+            node_id="node-a",
+            server_url="http://a:8080",
+            reachable=True,
+            workstreams={"ws1": {"id": "ws1", "name": "existing", "state": "idle"}},
+        )
+
+        # Mock httpx to return 401
+        import httpx as _httpx
+
+        mock_response = _httpx.Response(
+            401,
+            json={"error": "Unauthorized"},
+            request=_httpx.Request("GET", "http://a:8080/v1/api/dashboard"),
+        )
+
+        with patch.object(c._http_client, "get", return_value=mock_response):
+            c._poll_all_nodes()
+
+        # Workstream data must be preserved, node marked unreachable
+        assert c._nodes["node-a"].reachable is False
+        assert "ws1" in c._nodes["node-a"].workstreams
+        assert c._nodes["node-a"].workstreams["ws1"]["name"] == "existing"
+
+    def test_poll_403_preserves_workstreams(self):
+        """A 403 should also preserve state and mark unreachable."""
+        c = _make_collector()
+        c._nodes["node-a"] = NodeSnapshot(
+            node_id="node-a",
+            server_url="http://a:8080",
+            reachable=True,
+            workstreams={"ws1": {"id": "ws1", "name": "keep-me", "state": "running"}},
+        )
+
+        import httpx as _httpx
+
+        mock_response = _httpx.Response(
+            403,
+            json={"error": "Forbidden"},
+            request=_httpx.Request("GET", "http://a:8080/v1/api/dashboard"),
+        )
+
+        with patch.object(c._http_client, "get", return_value=mock_response):
+            c._poll_all_nodes()
+
+        assert c._nodes["node-a"].reachable is False
+        assert "ws1" in c._nodes["node-a"].workstreams
+
 
 class TestCollectorEvents:
     """Real-time event handling from cluster channel."""

turnstone/console/collector.py

@@ -20,6 +20,7 @@ from typing import TYPE_CHECKING, Any
 import httpx
 
 if TYPE_CHECKING:
+    from turnstone.core.auth import ServiceTokenManager
     from turnstone.mq.broker import RedisBroker
 
 log = logging.getLogger("turnstone.console.collector")
@@ -58,6 +59,7 @@ class ClusterCollector:
         max_poll_workers: int = 50,
         http_timeout: float = 5.0,
         auth_token: str = "",
+        token_manager: ServiceTokenManager | None = None,
     ):
         self._broker = broker
         self._prefix = prefix
@@ -65,16 +67,20 @@ class ClusterCollector:
         self._discovery_interval = discovery_interval
         self._max_poll_workers = max_poll_workers
         self._http_timeout = http_timeout
+        self._token_manager = token_manager
+        # Static auth header — only used when no token_manager is present.
+        # When a token_manager exists, auth is injected per-request via
+        # extra_headers in _poll_all_nodes to avoid stale JWT expiry.
+        self._static_auth: dict[str, str] | None = None
+        if auth_token and token_manager is None:
+            self._static_auth = {"Authorization": f"Bearer {auth_token}"}
 
         self._lock = threading.Lock()
         self._nodes: dict[str, NodeSnapshot] = {}
         self._running = False
         self._threads: list[threading.Thread] = []
         self._poll_pool = ThreadPoolExecutor(max_workers=max_poll_workers)
-        headers = {}
-        if auth_token:
-            headers["Authorization"] = f"Bearer {auth_token}"
-        self._http_client = httpx.Client(timeout=http_timeout, headers=headers)
+        self._http_client = httpx.Client(timeout=http_timeout)
 
         # SSE fan-out to browser clients
         self._listeners: list[queue.Queue[dict[str, Any]]] = []
@@ -236,6 +242,14 @@ class ClusterCollector:
 
     def _poll_all_nodes(self) -> None:
         """Fetch dashboard data from all known nodes in parallel."""
+        # Snapshot current auth header for this poll cycle.  Per-request
+        # headers avoid mutating shared client state (thread-safe).
+        if self._token_manager is not None:
+            poll_headers: dict[str, str] | None = {
+                "Authorization": f"Bearer {self._token_manager.token}"
+            }
+        else:
+            poll_headers = self._static_auth
         with self._lock:
             targets = [
                 (n.node_id, n.server_url)
@@ -246,25 +260,44 @@ class ClusterCollector:
         if not targets:
             return
 
-        futures = {self._poll_pool.submit(self._fetch_node, nid, url): nid for nid, url in targets}
+        futures = {
+            self._poll_pool.submit(self._fetch_node, nid, url, poll_headers): nid
+            for nid, url in targets
+        }
         for future in as_completed(futures):
             nid = futures[future]
             try:
                 dashboard, health = future.result()
                 self._apply_poll(nid, dashboard, health)
+            except httpx.HTTPStatusError as exc:
+                if exc.response.status_code in (401, 403):
+                    log.warning(
+                        "Auth failure polling node %s: HTTP %d", nid, exc.response.status_code
+                    )
+                else:
+                    log.debug("Failed to poll node %s: HTTP %d", nid, exc.response.status_code)
+                with self._lock:
+                    if nid in self._nodes:
+                        self._nodes[nid].reachable = False
             except Exception:
                 log.debug("Failed to poll node %s", nid)
                 with self._lock:
                     if nid in self._nodes:
                         self._nodes[nid].reachable = False
 
-    def _fetch_node(self, node_id: str, server_url: str) -> tuple[dict[str, Any], dict[str, Any]]:
+    def _fetch_node(
+        self,
+        node_id: str,
+        server_url: str,
+        extra_headers: dict[str, str] | None = None,
+    ) -> tuple[dict[str, Any], dict[str, Any]]:
         """Fetch /v1/api/dashboard and /health from a single node."""
         base = server_url.rstrip("/")
-        dash_resp = self._http_client.get(f"{base}/v1/api/dashboard")
+        dash_resp = self._http_client.get(f"{base}/v1/api/dashboard", headers=extra_headers)
+        dash_resp.raise_for_status()
         dash_data: dict[str, Any] = dash_resp.json()
         try:
-            health_resp = self._http_client.get(f"{base}/health")
+            health_resp = self._http_client.get(f"{base}/health", headers=extra_headers)
             health_data: dict[str, Any] = health_resp.json()
         except Exception:
             health_data = {}

turnstone/console/server.py

@@ -679,17 +679,13 @@ async def _proxy_sse(
 
 @asynccontextmanager
 async def _lifespan(app: Starlette) -> AsyncGenerator[None, None]:
-    # Create async HTTP client for proxy routes
-    headers: dict[str, str] = {}
-    token = app.state.proxy_auth_token
-    if token:
-        headers["Authorization"] = f"Bearer {token}"
-    app.state.proxy_client = httpx.AsyncClient(timeout=30, headers=headers)
-    # Separate client for SSE streams — longer read timeout, shared connection pool
+    # Create async HTTP clients for proxy routes.  Auth headers are NOT baked
+    # in — _proxy_auth_headers() injects a fresh token per-request so JWTs
+    # auto-rotate via ServiceTokenManager instead of expiring after 1 hour.
+    app.state.proxy_client = httpx.AsyncClient(timeout=30)
     app.state.proxy_sse_client = httpx.AsyncClient(
         timeout=httpx.Timeout(connect=5, read=30, write=5, pool=5),
         limits=httpx.Limits(keepalive_expiry=30),
-        headers=headers,
     )
     # Start scheduler if configured
     scheduler = getattr(app.state, "scheduler", None)
@@ -4863,13 +4859,13 @@ def main() -> None:
             audience=JWT_AUD_SERVER,
             expiry_hours=1,
         )
-        collector_token = collector_token_mgr.token
-        log.info("console.collector_jwt_minted")
+        log.info("console.collector_token_manager_created")
 
     collector = ClusterCollector(
         broker=broker,
         poll_interval=args.poll_interval,
-        auth_token=collector_token,
+        auth_token=collector_token if collector_token_mgr is None else "",
+        token_manager=collector_token_mgr,
     )
     collector.start()
 
@@ -4907,8 +4903,7 @@ def main() -> None:
             audience=JWT_AUD_SERVER,
             expiry_hours=1,
         )
-        proxy_token = proxy_token_mgr.token
-        log.info("console.proxy_jwt_minted")
+        log.info("console.proxy_token_manager_created")
 
     from turnstone.core.web_helpers import parse_cors_origins
 
@@ -4920,7 +4915,7 @@ def main() -> None:
         auth_config=auth_config,
         jwt_secret=jwt_secret,
         auth_storage=auth_storage,
-        proxy_auth_token=proxy_token,
+        proxy_auth_token=proxy_token if proxy_token_mgr is None else "",
         proxy_token_mgr=proxy_token_mgr,
         cors_origins=cors_origins,
     )