fix(clients): honour token_provider on the SSE-backed clients and never hang on a dead stream

robosystems_client/clients/README.md

@@ -115,6 +115,33 @@ from robosystems_client.clients import TokenClients
 extensions = TokenClients(token="your-jwt-token", base_url="https://api.robosystems.ai")
 ```
 
+### Rotating Credentials (`token_provider`)
+
+Short-lived JWTs rotate, and the backend revokes the previous token on every
+session refresh — a credential captured when the facade was built stops
+working the moment the session rotates. Pass a zero-arg callable instead and
+every client resolves the credential fresh: the GraphQL facades (`ledger` /
+`investor` / `library`) on each request, and the SSE-backed clients
+(`operator` / `operations` / `query`) on each REST call _and_ each stream
+connect. It wins over any static `token` or auth header in `headers`, and is
+routed by shape (`rfs…` keys as `X-API-Key`, anything else as a Bearer JWT).
+
+```python
+from robosystems_client.clients import RoboSystemsClients, RoboSystemsClientConfig
+
+extensions = RoboSystemsClients(
+  RoboSystemsClientConfig(
+    base_url="https://api.robosystems.ai",
+    token_provider=lambda: load_current_jwt(),  # or `lambda: manager.token`
+  )
+)
+```
+
+`OperatorClient` also follows a queued run over `/v1/operations/{id}/status`
+whenever its stream gives no verdict — it could not open, its reconnects ran
+out, or it ended before a terminal event — so a run that is already executing
+is never lost. `OperatorOptions.poll_interval` (seconds) tunes the interval.
+
 ### Environment-Specific Configurations
 
 ```python

robosystems_client/clients/auth_integration.py

@@ -6,26 +6,16 @@
 from typing import Dict, Any
 from ..client import Client, AuthenticatedClient
 from .facade import RoboSystemsClients, RoboSystemsClientConfig
+from .token_utils import apply_auth_header
 
 
 def _apply_auth_header(headers: Dict[str, str], credential: str) -> None:
   """Set the correct auth header for a credential, routed by shape.
 
-  The backend accepts two credential formats, and they go in DIFFERENT
-  headers — not interchangeable (see ``graphql/client.py``):
-
-  - Long-lived API keys (``rfs…`` prefix) → ``X-API-Key``. Validated
-    against the api_keys table.
-  - Short-lived JWTs → ``Authorization: Bearer …``. Validated by the
-    JWT middleware.
-
-  Sending a JWT as ``X-API-Key`` (or an API key as Bearer) both fail
-  with 401 "Invalid API key" — so exactly one header is set, never both.
+  Thin alias of :func:`token_utils.apply_auth_header`, which the per-call
+  header resolver in the SSE-backed clients shares — one routing rule.
   """
-  if credential.startswith("rfs"):
-    headers["X-API-Key"] = credential
-  else:
-    headers["Authorization"] = f"Bearer {credential}"
+  apply_auth_header(headers, credential)
 
 
 def _build_sdk_client(base_url: str, credential: str, headers: Dict[str, str]):

robosystems_client/clients/operation_client.py

@@ -10,6 +10,7 @@
 from enum import Enum
 
 from .sse_client import SSEClient, AsyncSSEClient, SSEConfig, EventType
+from .token_utils import resolve_auth_headers
 
 logger = logging.getLogger(__name__)
 
@@ -105,8 +106,11 @@ def monitor_operation(
     error = None
 
     # Set up SSE connection with event replay from the beginning
-    # This handles the race condition where the operation may have already completed
-    sse_config = SSEConfig(base_url=self.base_url, headers=self.headers)
+    # This handles the race condition where the operation may have already completed.
+    # Headers are resolved per connect so a rotated JWT reaches the stream.
+    sse_config = SSEConfig(
+      base_url=self.base_url, headers=resolve_auth_headers(self.config)
+    )
     sse_client = SSEClient(sse_config)
 
     def on_operation_started(data):
@@ -216,8 +220,8 @@ def get_operation_status(self, operation_id: str) -> Dict[str, Any]:
     )
     from ..client import Client
 
-    # Use regular Client with headers instead of AuthenticatedClient
-    client = Client(base_url=self.base_url, headers=self.headers)
+    # Plain Client with the headers current now (`token_provider` wins).
+    client = Client(base_url=self.base_url, headers=resolve_auth_headers(self.config))
     try:
       # Auth travels in self.headers (X-API-Key / Authorization). The generated
       # function takes no `token` kwarg — passing one raised TypeError, which
@@ -246,8 +250,8 @@ def cancel_operation(self, operation_id: str) -> bool:
     from ..api.operations.cancel_operation import sync_detailed as cancel_operation
     from ..client import Client
 
-    # Use regular Client with headers instead of AuthenticatedClient
-    client = Client(base_url=self.base_url, headers=self.headers)
+    # Plain Client with the headers current now (`token_provider` wins).
+    client = Client(base_url=self.base_url, headers=resolve_auth_headers(self.config))
     try:
       # See get_operation_status: no `token` kwarg on the generated function.
       response = cancel_operation(operation_id=operation_id, client=client)
@@ -308,8 +312,11 @@ async def monitor_operation(
     completed = False
     error = None
 
-    # Set up SSE connection
-    sse_config = SSEConfig(base_url=self.base_url, headers=self.headers)
+    # Set up SSE connection; headers resolved per connect so a rotated JWT
+    # reaches the stream.
+    sse_config = SSEConfig(
+      base_url=self.base_url, headers=resolve_auth_headers(self.config)
+    )
     sse_client = AsyncSSEClient(sse_config)
 
     def on_operation_started(data):