feat(credentials): Wave 5 — CP resolves credentials and injects CREDENTIAL_IDS into runner pods (RHOAIENG-55817) (#1128)

## Summary

- **CP**: `resolveCredentialIDs()` pages SDK credentials and builds a
`provider→id` map; `ensureCredentialRoleBindings()` grants
`credential:token-reader` RoleBinding per credential to the session SA;
`buildEnv()` injects `CREDENTIAL_IDS` JSON env var into runner pods
- **Runner**: `_fetch_credential()` migrated from legacy endpoint to
`GET /api/ambient/v1/credentials/{id}/token` using `CREDENTIAL_IDS` map;
Jira `apiToken`→`token`, Google `accessToken`→`token` (full SA JSON
written to `GOOGLE_APPLICATION_CREDENTIALS`)
- **Tests**: 8 existing credential tests updated for new
`CREDENTIAL_IDS` contract; 1 new test added; 29/29 credential tests
passing, 622 total pass

## Credential Flow

```
Session start → CP calls sdk.Credentials().ListAll()
             → builds {"github": "id1", "jira": "id2"} map
             → grants credential:token-reader RoleBinding per provider to session SA
             → injects CREDENTIAL_IDS={"github": "id1", ...} into pod env

Runner start → reads CREDENTIAL_IDS
            → calls GET /api/ambient/v1/credentials/{id}/token per provider
            → sets GITHUB_TOKEN, GITLAB_TOKEN, JIRA_API_TOKEN, GOOGLE_APPLICATION_CREDENTIALS
```

## Files Changed

 < /dev/null |  Component | File | Change |
|---|---|---|
| CP | `kube_reconciler.go` | `resolveCredentialIDs`,
`ensureCredentialRoleBindings`, `buildEnv` updated, `provisionSession`
wired |
| Runner | `platform/auth.py` | `_fetch_credential` URL + field mapping;
Google/Jira updated; dead stub removed |
| Tests | `test_shared_session_credentials.py` | Updated for
`CREDENTIAL_IDS` contract; 1 new test |
| Docs | `control-plane.spec.md` | Removed `other` from token response;
updated status to implemented |
| Docs | `control-plane.guide.md` | Restructured with dev context
reference; gap table updated |
| Docs | `.claude/context/control-plane-development.md` | Fixed
operator→CP refs; added `CREDENTIAL_IDS` section |

## Test plan

- [x] `cd components/runners/ambient-runner && python -m pytest tests/
-k credential` — 29/29 pass
- [x] `cd components/ambient-control-plane && go build ./... && go vet
./...` — clean
- [x] `gofmt -l internal/reconciler/kube_reconciler.go` — no output
(clean)
- [x] `uv run ruff check .` — no errors
- [ ] Deploy to MPP cluster and verify `CREDENTIAL_IDS` injected into
pod env
- [ ] Verify `credential-token-reader-{provider}` RoleBindings created
per session
- [ ] Verify `GITLAB_TOKEN`/`GITHUB_TOKEN` set in running pod

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

Author: ktdreyer

.claude/context/control-plane-development.md

@@ -9,7 +9,7 @@
 - **Protocol:** gRPC (proto definitions in `components/ambient-api-server/proto/ambient/v1/`)
 - **Runner entry:** `components/runners/ambient-runner/main.py`
 - **gRPC bridge:** `components/runners/ambient-runner/ambient_runner/bridges/claude/grpc_transport.py`
-- **Runner env:** controlled by operator via pod env vars — see `operator-development.md`
+- **Runner env:** controlled by CP via pod env vars — see `kube_reconciler.go:buildEnv()`
 
 ---
 
@@ -19,7 +19,7 @@ The CP was once reverted from upstream because it interfered with the runner's S
 
 | Concern | Runner expects | CP must preserve |
 |---|---|---|
-| Session start | Job pod scheduled by operator | CP does not reschedule |
+| Session start | Pod provisioned by CP | CP does not reschedule |
 | Event emission | Runner pushes AG-UI events via gRPC | CP forwards in order, never drops |
 | `RUN_FINISHED` | Emitted once, last | CP forwards exactly once — never duplicated |
 | `MESSAGES_SNAPSHOT` | Emitted periodically | CP forwards in order |
@@ -74,11 +74,27 @@ Two classes:
 
 **`GRPCMessageWriter`** — per-turn event consumer. Accumulates `MESSAGES_SNAPSHOT` content. On `RUN_FINISHED` or `RUN_ERROR`, calls `PushSessionMessage(event_type="assistant", payload=assistant_text)` — writes the durable DB record.
 
-Only active when `AMBIENT_GRPC_ENABLED=true` (set by operator when `AMBIENT_GRPC_URL` is non-empty).
+Only active when `AMBIENT_GRPC_ENABLED=true` (set by CP when `AMBIENT_GRPC_URL` is non-empty).
 
 ### Inbox drain at session start
 
-The runner drains the agent's inbox before starting the Claude Code session. All unread messages are assembled into `INITIAL_PROMPT` via `assembleInitialPrompt()` in the operator (see `kube_reconciler.go`). The runner receives this as the `INITIAL_PROMPT` env var.
+The runner drains the agent's inbox before starting the Claude Code session. All unread messages are assembled into `INITIAL_PROMPT` via `assembleInitialPrompt()` in the CP (`reconciler/kube_reconciler.go`). The runner receives this as the `INITIAL_PROMPT` env var.
+
+### Credential fetch (Wave 5)
+
+The CP resolves credentials for the session before pod creation. It calls `sdk.Credentials().ListAll()` — the API server applies RBAC-scoped filtering server-side, returning only credentials visible to the session's service account. The CP takes the first credential per provider (first-match wins; ordering is server-determined). It then:
+
+1. Builds `CREDENTIAL_IDS` — a JSON map of `provider → credential_id` — and injects it into the runner pod env
+2. Grants `credential:token-reader` on each credential ID to the runner pod's service account
+
+The runner reads `CREDENTIAL_IDS` at startup and calls `GET /api/ambient/v1/credentials/{id}/token` per provider. Response always uses `token` field (uniform across all providers). See `platform/auth.py:_fetch_credential()`.
+
+| Provider | Env var(s) set | File written |
+|----------|---------------|--------------|
+| `github` | `GITHUB_TOKEN` | `/tmp/.ambient_github_token` |
+| `gitlab` | `GITLAB_TOKEN` | `/tmp/.ambient_gitlab_token` |
+| `jira` | `JIRA_URL`, `JIRA_API_TOKEN`, `JIRA_EMAIL` | — |
+| `google` | `USER_GOOGLE_EMAIL` | `credentials.json` (token value is full SA JSON) |
 
 ### AG-UI event order (invariant)
 
@@ -102,7 +118,7 @@ The api-server does not have a built-in proxy to runner pods. Runner pods are ad
 http://session-{KubeCrName}.{KubeNamespace}.svc.cluster.local:8001
 ```
 
-The `Session` model stores `KubeCrName` and `KubeNamespace` — both available from the DB. The runner listens on port `8001` (set via `AGUI_PORT` env var by the operator; runner default is `8000` but the operator overrides it).
+The `Session` model stores `KubeCrName` and `KubeNamespace` — both available from the DB. The runner listens on port `8001` (set via `AGUI_PORT` env var by the CP; runner default is `8000` but the CP overrides it).
 
 This pattern is used by `components/backend/websocket/agui_proxy.go` (V1 backend). Any new proxy endpoint in the api-server must implement this same addressing.
 

components/ambient-control-plane/internal/reconciler/kube_reconciler.go

@@ -2,6 +2,7 @@ package reconciler
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"strings"
 	"time"
@@ -39,7 +40,6 @@ type KubeReconcilerConfig struct {
 	CPRuntimeNamespace    string
 }
 
-
 type SimpleKubeReconciler struct {
 	factory     *SDKClientFactory
 	kube        *kubeclient.KubeClient
@@ -150,7 +150,19 @@ func (r *SimpleKubeReconciler) provisionSession(ctx context.Context, session typ
 		return fmt.Errorf("ensuring service account: %w", err)
 	}
 
-	if err := r.ensurePod(ctx, namespace, session, sessionLabel, sdk); err != nil {
+	credentialIDs, err := r.resolveCredentialIDs(ctx, sdk)
+	if err != nil {
+		r.logger.Warn().Err(err).Str("session_id", session.ID).Msg("credential resolution failed; continuing without credentials")
+		credentialIDs = map[string]string{}
+	}
+
+	if len(credentialIDs) > 0 {
+		if err := r.ensureCredentialRoleBindings(ctx, namespace, serviceAccountName(session.ID), credentialIDs); err != nil {
+			return fmt.Errorf("ensuring credential role bindings: %w", err)
+		}
+	}
+
+	if err := r.ensurePod(ctx, namespace, session, sessionLabel, sdk, credentialIDs); err != nil {
 		return fmt.Errorf("ensuring pod: %w", err)
 	}
 
@@ -204,7 +216,6 @@ func (r *SimpleKubeReconciler) cleanupSession(ctx context.Context, session types
 	return nil
 }
 
-
 func (r *SimpleKubeReconciler) ensureService(ctx context.Context, namespace string, session types.Session, labelSelector string) error {
 	name := serviceName(session.ID)
 
@@ -361,7 +372,7 @@ func (r *SimpleKubeReconciler) ensureServiceAccount(ctx context.Context, namespa
 	return nil
 }
 
-func (r *SimpleKubeReconciler) ensurePod(ctx context.Context, namespace string, session types.Session, labelSelector string, sdk *sdkclient.Client) error {
+func (r *SimpleKubeReconciler) ensurePod(ctx context.Context, namespace string, session types.Session, labelSelector string, sdk *sdkclient.Client, credentialIDs map[string]string) error {
 	name := podName(session.ID)
 
 	if _, err := r.nsKube().GetPod(ctx, namespace, name); err == nil {
@@ -394,7 +405,7 @@ func (r *SimpleKubeReconciler) ensurePod(ctx context.Context, namespace string,
 				},
 			},
 			"volumeMounts": r.buildVolumeMounts(),
-			"env":          r.buildEnv(ctx, session, sdk, secretName, useMCPSidecar),
+			"env":          r.buildEnv(ctx, session, sdk, secretName, useMCPSidecar, credentialIDs),
 			"resources": map[string]interface{}{
 				"requests": map[string]interface{}{
 					"cpu":    "500m",
@@ -536,7 +547,7 @@ func (r *SimpleKubeReconciler) ensureVertexSecret(ctx context.Context, namespace
 	return nil
 }
 
-func (r *SimpleKubeReconciler) buildEnv(ctx context.Context, session types.Session, sdk *sdkclient.Client, credSecretName string, useMCPSidecar bool) []interface{} {
+func (r *SimpleKubeReconciler) buildEnv(ctx context.Context, session types.Session, sdk *sdkclient.Client, credSecretName string, useMCPSidecar bool, credentialIDs map[string]string) []interface{} {
 	useVertex := "0"
 	if r.cfg.VertexEnabled {
 		useVertex = "1"
@@ -603,9 +614,75 @@ func (r *SimpleKubeReconciler) buildEnv(ctx context.Context, session types.Sessi
 		env = append(env, envVar("REPOS_JSON", fmt.Sprintf(`[{"url":%q}]`, session.RepoURL)))
 	}
 
+	if len(credentialIDs) > 0 {
+		raw, err := json.Marshal(credentialIDs)
+		if err == nil {
+			env = append(env, envVar("CREDENTIAL_IDS", string(raw)))
+		}
+	}
+
 	return env
 }
 
+func (r *SimpleKubeReconciler) resolveCredentialIDs(ctx context.Context, sdk *sdkclient.Client) (map[string]string, error) {
+	result := map[string]string{}
+
+	it := sdk.Credentials().ListAll(ctx, &types.ListOptions{Size: 100})
+	for it.Next() {
+		cred := it.Item()
+		if cred.Provider == "" || cred.ID == "" {
+			continue
+		}
+		if _, already := result[cred.Provider]; !already {
+			result[cred.Provider] = cred.ID
+		}
+	}
+	if err := it.Err(); err != nil {
+		return nil, fmt.Errorf("listing credentials: %w", err)
+	}
+
+	r.logger.Info().Int("count", len(result)).Msg("resolved credential IDs for session")
+	return result, nil
+}
+
+func (r *SimpleKubeReconciler) ensureCredentialRoleBindings(ctx context.Context, namespace, saName string, credentialIDs map[string]string) error {
+	for provider, credID := range credentialIDs {
+		name := fmt.Sprintf("%s-credential-%s", saName, provider)
+		rb := &unstructured.Unstructured{
+			Object: map[string]interface{}{
+				"apiVersion": "rbac.authorization.k8s.io/v1",
+				"kind":       "RoleBinding",
+				"metadata": map[string]interface{}{
+					"name":      name,
+					"namespace": namespace,
+				},
+				"roleRef": map[string]interface{}{
+					"apiGroup": "rbac.authorization.k8s.io",
+					"kind":     "ClusterRole",
+					"name":     "credential:token-reader",
+				},
+				"subjects": []interface{}{
+					map[string]interface{}{
+						"kind":      "ServiceAccount",
+						"name":      saName,
+						"namespace": namespace,
+					},
+				},
+			},
+		}
+		if _, err := r.nsKube().CreateRoleBinding(ctx, namespace, rb); err != nil && !k8serrors.IsAlreadyExists(err) {
+			return fmt.Errorf("creating credential role binding %s for provider %s (id=%s): %w", name, provider, credID, err)
+		}
+		r.logger.Debug().
+			Str("role_binding", name).
+			Str("namespace", namespace).
+			Str("provider", provider).
+			Str("credential_id", credID).
+			Msg("credential token-reader role binding ensured")
+	}
+	return nil
+}
+
 func (r *SimpleKubeReconciler) assembleInitialPrompt(ctx context.Context, session types.Session, sdk *sdkclient.Client) string {
 	var parts []string
 

components/runners/ambient-runner/ag_ui_claude_sdk/adapter.py

@@ -607,6 +607,7 @@ def _emit_task_event(self, message: Any) -> "CustomEvent":
             TaskProgressMessage,
             TaskNotificationMessage,
         )
+
         if isinstance(message, TaskStartedMessage):
             return self._emit_task_started(message)
         elif isinstance(message, TaskProgressMessage):
@@ -633,6 +634,7 @@ def drain_hook_events(self) -> list:
                         sid = val.get("session_id", "")
                         if sid:
                             from pathlib import Path
+
                             base = Path.home() / ".claude" / "projects"
                             if base.exists():
                                 expected = f"agent-{agent_id}.jsonl"
@@ -670,7 +672,9 @@ def _emit_task_progress(self, message: Any) -> "CustomEvent":
         existing = self._task_registry.get(message.task_id, {})
         existing.update(progress_value)
         self._task_registry[message.task_id] = existing
-        return CustomEvent(type=EventType.CUSTOM, name="task:progress", value=progress_value)
+        return CustomEvent(
+            type=EventType.CUSTOM, name="task:progress", value=progress_value
+        )
 
     def _emit_task_notification(self, message: Any) -> "CustomEvent":
         usage = getattr(message, "usage", None)
@@ -687,7 +691,9 @@ def _emit_task_notification(self, message: Any) -> "CustomEvent":
         self._task_registry[message.task_id] = existing
         if output_file:
             self._task_outputs[message.task_id] = output_file
-        return CustomEvent(type=EventType.CUSTOM, name="task:completed", value=notification_value)
+        return CustomEvent(
+            type=EventType.CUSTOM, name="task:completed", value=notification_value
+        )
 
     async def _stream_claude_sdk(
         self,
@@ -1163,7 +1169,10 @@ def flush_pending_msg():
                             ):
                                 yield event
 
-                elif isinstance(message, (TaskStartedMessage, TaskProgressMessage, TaskNotificationMessage)):
+                elif isinstance(
+                    message,
+                    (TaskStartedMessage, TaskProgressMessage, TaskNotificationMessage),
+                ):
                     yield self._emit_task_event(message)
 
                 elif isinstance(message, SystemMessage):
@@ -1364,4 +1373,3 @@ def flush_pending_msg():
         # Re-raise to let run() emit RunErrorEvent
         if stream_error is not None:
             raise stream_error
-

components/runners/ambient-runner/ag_ui_claude_sdk/hooks.py

@@ -22,12 +22,14 @@
 logger = logging.getLogger(__name__)
 
 # Default hook event names to register (only what the UI consumes).
-_DEFAULT_HOOKS = frozenset({
-    "SubagentStart",
-    "SubagentStop",
-    "Notification",
-    "Stop",
-})
+_DEFAULT_HOOKS = frozenset(
+    {
+        "SubagentStart",
+        "SubagentStop",
+        "Notification",
+        "Stop",
+    }
+)
 
 # Keys stripped from payloads (internal paths the frontend should not see).
 _SANITIZE_KEYS = frozenset({"transcript_path", "cwd"})
@@ -51,7 +53,9 @@ async def _forward_hook_as_custom_event(
     event_name = hook_input.get("hook_event_name", "unknown")
     payload = {k: v for k, v in hook_input.items() if k not in _SANITIZE_KEYS}
 
-    logger.debug("[Hook] %s fired (agent_id=%s)", event_name, hook_input.get("agent_id", "n/a"))
+    logger.debug(
+        "[Hook] %s fired (agent_id=%s)", event_name, hook_input.get("agent_id", "n/a")
+    )
 
     await queue.put(
         CustomEvent(

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

@@ -189,7 +189,16 @@ class _InboxMessageProto:
       8: updated_at  (Timestamp, wire 2)
     """
 
-    __slots__ = ("id", "agent_id", "from_agent_id", "from_name", "body", "read", "created_at", "updated_at")
+    __slots__ = (
+        "id",
+        "agent_id",
+        "from_agent_id",
+        "from_name",
+        "body",
+        "read",
+        "created_at",
+        "updated_at",
+    )
 
     def __init__(self) -> None:
         self.id: str = ""

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

@@ -174,7 +174,6 @@ def watch(
 # ---------------------------------------------------------------------------
 
 
-
 def _encode_string(field_number: int, value: str) -> bytes:
     encoded = value.encode("utf-8")
     tag = (field_number << 3) | 2

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

@@ -151,9 +151,7 @@ async def lifespan(app: FastAPI):
             active_streams = getattr(bridge, "_active_streams", None)
             if active_streams is not None and session_id not in active_streams:
                 active_streams[session_id] = asyncio.Queue(maxsize=100)
-                logger.info(
-                    "Pre-registered SSE queue for session=%s", session_id
-                )
+                logger.info("Pre-registered SSE queue for session=%s", session_id)
 
         # Auto-execute prompts when present (skipped only for resumes,
         # where the conversation is continued rather than re-started).