fix(runner,rbac,cli,routes): credential auth reliability and SSE improvements (#1231)

## Summary

- **runner**: On 401 from credential fetch, refresh the CP token and
retry once instead of aborting the turn — fixes intermittent turn
failures when the cached BOT_TOKEN expires mid-session
- **rbac**: Fix `pathToAction()` so `GET /credentials/{id}/token`
resolves to action `token` (matching `credential:token-reader` role)
rather than `read` — was causing systematic 401s for runner SAs
- **cli**: Improve `acpctl session send -f` output with `[thinking]`
reasoning blocks, `[ToolName] → result` tool call annotations, and
proper streaming newline handling
- **routes**: Add `haproxy.router.openshift.io/timeout: 10m` to all
`ambient-api-server` OpenShift Routes to prevent 504 Gateway Timeouts on
long-running SSE streams

## Test plan

- [ ] Send a message via `acpctl session send -f` and confirm
`[thinking]` and tool call output appears
- [ ] Verify credential fetches succeed on long-running sessions (no
intermittent 401 turn aborts)
- [ ] Confirm `GET /credentials/{id}/token` returns 200 for runner SA
(RBAC fix)
- [ ] Verify SSE streams survive beyond 30s without 504 after route
annotation is applied

🤖 Generated with [Claude Code](https://claude.ai/code)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **New Features**
* Automatic token refresh with a single retry on authentication failures
* Richer CLI streaming output with reasoning blocks, tool-call
annotations, and improved text formatting
* Optional support for alternate text-block types when collecting
observability output

* **Bug Fixes**
* Authorization now recognizes special path-based operations for correct
permission checks

* **Configuration**
  * Added 10-minute timeout to API server routes
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: Ambient Code Bot <bot@ambient-code.local>
Co-authored-by: Claude <noreply@anthropic.com>

Author: 3 people

components/ambient-api-server/pkg/rbac/middleware.go

@@ -57,7 +57,7 @@ func (m *DBAuthorizationMiddleware) AuthorizeApi(next http.Handler) http.Handler
 }
 
 func (m *DBAuthorizationMiddleware) isAllowed(g *gorm.DB, username, method, path string) (bool, error) {
-	action := httpMethodToAction(method)
+	action := pathToAction(method, path)
 	resource := pathToResource(path)
 
 	var bindings []roleBindingRow
@@ -130,6 +130,22 @@ func httpMethodToAction(method string) string {
 	}
 }
 
+func pathToAction(method, path string) string {
+	parts := strings.Split(strings.TrimPrefix(path, "/"), "/")
+	for i, p := range parts {
+		if p == "v1" && i+2 < len(parts) {
+			last := parts[len(parts)-1]
+			switch last {
+			case "token":
+				return "token"
+			case "start", "stop", "ignite", "ignition":
+				return last
+			}
+		}
+	}
+	return httpMethodToAction(method)
+}
+
 func pathToResource(path string) string {
 	parts := strings.Split(strings.TrimPrefix(path, "/"), "/")
 	for i, p := range parts {

components/ambient-cli/cmd/acpctl/session/send.go

@@ -78,6 +78,8 @@ func runSend(cmd *cobra.Command, args []string) error {
 
 	out := cmd.OutOrStdout()
 	scanner := bufio.NewScanner(stream)
+	var reasoningBuf strings.Builder
+	var inText bool
 	for scanner.Scan() {
 		line := scanner.Text()
 		if !strings.HasPrefix(line, "data: ") {
@@ -91,18 +93,53 @@ func runSend(cmd *cobra.Command, args []string) error {
 		}
 
 		var evt struct {
-			Type  string `json:"type"`
-			Delta string `json:"delta"`
+			Type         string `json:"type"`
+			Delta        string `json:"delta"`
+			ToolCallName string `json:"toolCallName"`
+			Content      string `json:"content"`
 		}
 		if err := json.Unmarshal([]byte(data), &evt); err != nil {
 			continue
 		}
-		if evt.Type == "TEXT_MESSAGE_CONTENT" && evt.Delta != "" {
-			fmt.Fprint(out, evt.Delta)
+		switch evt.Type {
+		case "REASONING_MESSAGE_CONTENT":
+			reasoningBuf.WriteString(evt.Delta)
+		case "REASONING_END":
+			if reasoningBuf.Len() > 0 {
+				fmt.Fprintf(out, "[thinking] %s\n", strings.TrimSpace(reasoningBuf.String()))
+				reasoningBuf.Reset()
+			}
+		case "TEXT_MESSAGE_CONTENT":
+			if evt.Delta != "" {
+				inText = true
+				fmt.Fprint(out, evt.Delta)
+			}
+		case "TEXT_MESSAGE_END":
+			if inText {
+				fmt.Fprintln(out)
+				inText = false
+			}
+		case "TOOL_CALL_START":
+			if evt.ToolCallName != "" {
+				fmt.Fprintf(out, "[%s] ", evt.ToolCallName)
+			}
+		case "TOOL_CALL_RESULT":
+			if evt.Content != "" {
+				var content string
+				if err := json.Unmarshal([]byte(evt.Content), &content); err != nil {
+					content = evt.Content
+				}
+				lines := strings.SplitN(strings.TrimSpace(content), "\n", 4)
+				preview := strings.Join(lines, " | ")
+				if len(lines) >= 4 {
+					preview += " ..."
+				}
+				fmt.Fprintf(out, "→ %s\n", preview)
+			}
 		}
 	}
 
-	if !sendFollowJSON {
+	if inText {
 		fmt.Fprintln(out)
 	}
 

components/manifests/overlays/local-dev/ambient-api-server-route.yaml

@@ -6,6 +6,8 @@ metadata:
   labels:
     app: ambient-api-server
     component: api
+  annotations:
+    haproxy.router.openshift.io/timeout: 10m
 spec:
   to:
     kind: Service
@@ -23,6 +25,8 @@ metadata:
   labels:
     app: ambient-api-server
     component: grpc
+  annotations:
+    haproxy.router.openshift.io/timeout: 10m
 spec:
   to:
     kind: Service

components/manifests/overlays/mpp-openshift/ambient-api-server-route.yaml

@@ -7,6 +7,8 @@ metadata:
     app: ambient-api-server
     component: api
     shard: internal
+  annotations:
+    haproxy.router.openshift.io/timeout: 10m
 spec:
   to:
     kind: Service
@@ -26,6 +28,8 @@ metadata:
     app: ambient-api-server
     component: grpc
     shard: internal
+  annotations:
+    haproxy.router.openshift.io/timeout: 10m
 spec:
   to:
     kind: Service

components/manifests/overlays/production/ambient-api-server-route.yaml

@@ -6,6 +6,8 @@ metadata:
   labels:
     app: ambient-api-server
     component: api
+  annotations:
+    haproxy.router.openshift.io/timeout: 10m
 spec:
   to:
     kind: Service
@@ -23,6 +25,8 @@ metadata:
   labels:
     app: ambient-api-server
     component: grpc
+  annotations:
+    haproxy.router.openshift.io/timeout: 10m
 spec:
   to:
     kind: Service

components/runners/ambient-runner/ambient_runner/observability.py

@@ -515,14 +515,19 @@ def end_turn(
             return
 
         try:
-            from claude_agent_sdk import TextBlock
+            # Extract text content — TextBlock is optional (claude_agent_sdk may not be installed)
+            try:
+                from claude_agent_sdk import TextBlock as _TextBlock
+            except ImportError:
+                _TextBlock = None
 
-            # Extract text content
             text_content = []
             message_content = getattr(message, "content", []) or []
             for blk in message_content:
-                if isinstance(blk, TextBlock):
+                if _TextBlock is not None and isinstance(blk, _TextBlock):
                     text_content.append(getattr(blk, "text", ""))
+                elif hasattr(blk, "text") and isinstance(getattr(blk, "text", None), str):
+                    text_content.append(blk.text)
 
             output_text = (
                 "\n".join(text_content) if text_content else "(no text output)"

components/runners/ambient-runner/ambient_runner/platform/auth.py

@@ -17,7 +17,7 @@
 from urllib.parse import urlparse
 
 from ambient_runner.platform.context import RunnerContext
-from ambient_runner.platform.utils import get_bot_token
+from ambient_runner.platform.utils import get_bot_token, refresh_bot_token
 
 logger = logging.getLogger(__name__)
 
@@ -183,18 +183,45 @@ def _do_req():
                         f"and BOT_TOKEN fallback also failed"
                     ) from fallback_err
             if e.code in (401, 403):
-                logger.warning(
-                    f"{credential_type} credential fetch failed with HTTP {e.code}: {e}"
-                )
-                raise PermissionError(
-                    f"{credential_type} authentication failed with HTTP {e.code}"
-                ) from e
+                # BOT_TOKEN may have expired — refresh from CP endpoint and retry once.
+                return _retry_with_fresh_bot_token(e.code)
             logger.warning(f"{credential_type} credential fetch failed: {e}")
             return ""
         except Exception as e:
             logger.warning(f"{credential_type} credential fetch failed: {e}")
             return ""
 
+    def _retry_with_fresh_bot_token(original_code: int):
+        logger.info(
+            f"{credential_type} got {original_code} with cached BOT_TOKEN — refreshing from CP endpoint and retrying"
+        )
+        try:
+            fresh_bot = refresh_bot_token()
+        except Exception as refresh_err:
+            logger.warning(f"{credential_type} CP token refresh failed: {refresh_err}")
+            raise PermissionError(
+                f"{credential_type} authentication failed with HTTP {original_code}"
+            ) from refresh_err
+        retry_req = _urllib_request.Request(url, method="GET")
+        if fresh_bot:
+            retry_req.add_header("Authorization", f"Bearer {fresh_bot}")
+        if context.current_user_id:
+            retry_req.add_header("X-Runner-Current-User", context.current_user_id)
+        try:
+            with _urllib_request.urlopen(retry_req, timeout=10) as resp:
+                logger.info(f"{credential_type} retry with fresh BOT_TOKEN succeeded")
+                return resp.read().decode("utf-8", errors="replace")
+        except _urllib_request.HTTPError as retry_err:
+            logger.warning(f"{credential_type} retry with fresh BOT_TOKEN failed: {retry_err}")
+            raise PermissionError(
+                f"{credential_type} authentication failed with HTTP {retry_err.code}"
+            ) from retry_err
+        except Exception as retry_err:
+            logger.warning(f"{credential_type} retry with fresh BOT_TOKEN failed: {retry_err}")
+            raise PermissionError(
+                f"{credential_type} authentication failed with HTTP {original_code}"
+            ) from retry_err
+
     resp_text = await loop.run_in_executor(None, _do_req)
     if not resp_text:
         return {}

components/runners/ambient-runner/ambient_runner/platform/utils.py

@@ -71,6 +71,27 @@ def get_bot_token() -> str:
     return (os.getenv("BOT_TOKEN") or "").strip()
 
 
+def refresh_bot_token() -> str:
+    """Fetch a fresh token from the CP token endpoint and update the in-process cache.
+
+    Returns the new token, or the current cached token if the CP endpoint is not
+    configured (local dev mode). Raises RuntimeError if the CP fetch fails.
+    """
+    cp_token_url = os.getenv("AMBIENT_CP_TOKEN_URL", "")
+    if not cp_token_url:
+        return get_bot_token()
+
+    public_key_pem = os.getenv("AMBIENT_CP_TOKEN_PUBLIC_KEY", "")
+    session_id = os.getenv("SESSION_ID", "")
+    if not public_key_pem or not session_id:
+        logger.warning("refresh_bot_token: CP env vars incomplete, skipping refresh")
+        return get_bot_token()
+
+    from ambient_runner._grpc_client import _fetch_token_from_cp
+
+    return _fetch_token_from_cp(cp_token_url, public_key_pem, session_id)
+
+
 def is_env_truthy(value: str) -> bool:
     """Return True for "1", "true", or "yes" (case-insensitive)."""
     return value.strip().lower() in _TRUTHY_VALUES