feat(studio-cp,oab-mcp): k8s roster observe + runtime_context dispatch

crates/oab-mcp/src/lib.rs

@@ -66,24 +66,24 @@ pub fn tools() -> Vec<Tool> {
     vec![
         Tool::new(
             "deploy_list",
-            "List all OAB deployments (ECS services) in the cluster with replica counts and status.",
+            "List all OAB deployments in the cluster (ECS) or namespace (k8s) with replica counts. For a k8s-runtime fleet, pass `fleet` — the fleet's own context/namespace govern the call, `cluster` is ignored.",
             as_map(json!({
                 "type": "object",
                 "properties": {
-                    "fleet": { "type": "string", "description": "Fleet name (see fleet_config): targets the fleet's cluster and managing credential and, for listing tools, restricts results to its members. Overrides the cluster arg." },
-                    "cluster": { "type": "string", "description": "ECS cluster (defaults to the server's configured cluster)." }
+                    "fleet": { "type": "string", "description": "Fleet name (see fleet_config): targets the fleet's cluster/context and managing credential and, for listing tools, restricts results to its members. Overrides the cluster arg. Required for a k8s-runtime fleet." },
+                    "cluster": { "type": "string", "description": "ECS cluster (defaults to the server's configured cluster). Ignored for a k8s-runtime fleet." }
                 }
             })),
         ),
         Tool::new(
             "deploy_get",
-            "Get one deployment's read-model: replica counters plus each instance's canonical lifecycle phase (6-state).",
+            "Get one deployment's read-model: replica counters plus each instance's canonical lifecycle phase (6-state). Works for both ECS and k8s-runtime fleets (pass `fleet` for k8s).",
             as_map(json!({
                 "type": "object",
                 "properties": {
-                    "service": { "type": "string", "description": "ECS service name (or bare agent name)." },
-                    "fleet": { "type": "string", "description": "Fleet name (see fleet_config): targets the fleet's cluster and managing credential and, for listing tools, restricts results to its members. Overrides the cluster arg." },
-                    "cluster": { "type": "string", "description": "ECS cluster (defaults to the server's configured cluster)." }
+                    "service": { "type": "string", "description": "ECS service name, k8s reconstructed oab-{namespace}-{name}, or bare agent name." },
+                    "fleet": { "type": "string", "description": "Fleet name (see fleet_config): targets the fleet's cluster/context and managing credential and, for listing tools, restricts results to its members. Overrides the cluster arg. Required for a k8s-runtime fleet." },
+                    "cluster": { "type": "string", "description": "ECS cluster (defaults to the server's configured cluster). Ignored for a k8s-runtime fleet." }
                 },
                 "required": ["service"]
             })),
@@ -207,7 +207,7 @@ pub fn tools() -> Vec<Tool> {
         ),
         Tool::new(
             "runtime_context",
-            "Show the effective runtime identity/context this control plane resolved for a cluster/fleet: the acting principal (STS caller ARN), its kind (role vs static user), account (scope), region (location), a best-effort credential-source hint, the fleet binding in effect (if any), and — when the binding declares an expected_principal — whether the resolved identity matches it (identity_matches; a non-blocking IdentityMismatch when false). Read-only; answers \"who am I acting as, against what account?\" and surfaces silent credential fallback.",
+            "Show the effective runtime identity/context this control plane resolved for a cluster/fleet: for ECS, the acting principal (STS caller ARN), its kind (role vs static user), account (scope), region (location), a best-effort credential-source hint; for a k8s-runtime fleet, the SelfSubjectReview-resolved principal against the fleet's kubeconfig context (cluster is null, context/namespace are set instead). Also returns the fleet binding in effect (if any) and — when the binding declares an expected_principal — whether the resolved identity matches it (identity_matches; a non-blocking IdentityMismatch when false). Read-only; answers \"who am I acting as, against what account?\" and surfaces silent credential fallback.",
             as_map(json!({
                 "type": "object",
                 "properties": {
@@ -469,6 +469,27 @@ impl OabMcp {
         })
     }
 
+    /// Like [`Self::target`], but doesn't reject a k8s-runtime fleet — for
+    /// the calls this made runtime-aware (studio#146): roster list/get and
+    /// `runtime_context`. Returns the named fleet's binding, or `None` for
+    /// an unscoped call (back-compat, same as `target()`'s bare-cluster
+    /// path). An unknown fleet name is still an error, same as `target()`.
+    /// `target()` itself stays ecs-only — every other call (scale/delete/
+    /// apply/provision/events) genuinely has no k8s-runtime dispatch yet, so
+    /// it should keep failing loudly rather than silently acting against an
+    /// empty cluster string.
+    fn named_fleet(&self, args: &Map<String, Value>) -> Result<Option<scp::FleetBinding>> {
+        let Some(name) = args.get("fleet").and_then(Value::as_str) else {
+            return Ok(None);
+        };
+        let guard = self.bindings.read().unwrap();
+        let binding = guard.get(name).cloned().ok_or_else(|| {
+            let known: Vec<&str> = guard.fleets.iter().map(|f| f.name.as_str()).collect();
+            anyhow::anyhow!("unknown fleet {name:?}; configured fleets: [{}]", known.join(", "))
+        })?;
+        Ok(Some(binding))
+    }
+
     /// The AWS config to act as for `cluster`: the fleet binding's credential
     /// when one governs it (resolved once, then memoized), else the default
     /// chain. This is where the per-fleet **switch** takes effect — a bound
@@ -496,6 +517,18 @@ impl OabMcp {
     }
 
     async fn t_list(&self, args: &Map<String, Value>) -> Result<Value> {
+        if let Some(b) = self.named_fleet(args)? {
+            if b.runtime == scp::FleetRuntime::K8s {
+                let namespace = b.namespace.clone().unwrap_or_else(|| "default".to_string());
+                let svcs = scp::observe_k8s_services(b.context.as_deref(), &namespace).await?;
+                let deployments: Vec<Value> = svcs
+                    .iter()
+                    .filter(|s| b.includes(&s.service_name, &s.name))
+                    .map(|s| json!({ "name": s.name, "namespace": s.namespace, "desired": s.desired }))
+                    .collect();
+                return Ok(json!({ "context": b.context, "namespace": namespace, "deployments": deployments }));
+            }
+        }
         let t = self.target(args)?;
         let cluster = t.cluster.clone();
         let svcs = scp::observe_services(&self.aws_for(&cluster).await, &cluster).await?;
@@ -519,11 +552,20 @@ impl OabMcp {
     }
 
     async fn t_get(&self, args: &Map<String, Value>) -> Result<Value> {
-        let cluster = self.target(args)?.cluster;
         let service = args
             .get("service")
             .and_then(Value::as_str)
             .ok_or_else(|| anyhow::anyhow!("missing required arg: service"))?;
+        if let Some(b) = self.named_fleet(args)? {
+            if b.runtime == scp::FleetRuntime::K8s {
+                let namespace = b.namespace.clone().unwrap_or_else(|| "default".to_string());
+                return match scp::observe_k8s_deployment(b.context.as_deref(), &namespace, service).await? {
+                    Some(d) => Ok(deployment_json(&d)),
+                    None => Ok(json!({ "found": false, "service": service })),
+                };
+            }
+        }
+        let cluster = self.target(args)?.cluster;
         match scp::observe_deployment(&self.aws_for(&cluster).await, &cluster, service).await? {
             Some(d) => Ok(deployment_json(&d)),
             None => Ok(json!({ "found": false, "service": service })),
@@ -848,6 +890,34 @@ impl OabMcp {
     }
 
     async fn t_runtime_context(&self, args: &Map<String, Value>) -> Result<Value> {
+        if let Some(b) = self.named_fleet(args)? {
+            if b.runtime == scp::FleetRuntime::K8s {
+                let ctx = scp::observe_k8s_identity(b.context.as_deref()).await?;
+                let expected = b.expected_principal.clone();
+                let identity_matches = expected
+                    .as_ref()
+                    .map(|e| scp::principal_matches(e, &ctx.principal));
+                return Ok(json!({
+                    "cluster": null,
+                    "context": b.context,
+                    "namespace": b.namespace,
+                    "principal": ctx.principal,
+                    "principal_kind": ctx.principal_kind,
+                    "scope": ctx.scope,
+                    "location": ctx.location,
+                    "source": ctx.source,
+                    "caller_id": ctx.caller_id,
+                    "binding": json!({
+                        "name": b.name,
+                        "context": b.context,
+                        "namespace": b.namespace,
+                        "expected_principal": b.expected_principal,
+                    }),
+                    "expected_principal": expected,
+                    "identity_matches": identity_matches,
+                }));
+            }
+        }
         let t = self.target(args)?;
         let cluster = t.cluster.clone();
         let aws = self.aws_for(&cluster).await;

crates/studio-cp/src/lib.rs

@@ -170,6 +170,216 @@ pub async fn observe_events(
     oabctl::fetch_ecs_events(aws_config, log_group, Some(cluster), service, since_ms, limit).await
 }
 
+// ---- k8s observe (studio#146) --------------------------------------------
+//
+// The k8s counterpart to the ECS read-model above (`observe_services` /
+// `observe_deployment` / `build_deployment`), producing the exact same
+// runtime-agnostic `Deployment`/`InstancePhase` types via
+// `agent_lifecycle::k8s::K8sDriver` instead of `EcsDriver`. This was
+// deliberately deferred when the k8s `ProvisionDriver` write path landed
+// (`oabctl::k8s_driver`'s module doc, studio#63 slice 3b) — this is that
+// follow-up (studio#146).
+
+/// One k8s Deployment discovered while listing a namespace — the k8s
+/// counterpart to ECS's [`oabctl::ServiceStatus`] enumeration step (only the
+/// fields fleet-membership filtering and the follow-up per-Deployment fetch
+/// need; full per-Pod detail comes from [`observe_k8s_deployment`], mirroring
+/// the ECS `deploy_list` → `deploy_get` two-step the console roster already
+/// does).
+pub struct K8sServiceStatus {
+    /// The agent's original (un-slugified) name — read back from the
+    /// Deployment's `oab/name` label (`k8s_driver.rs::build_deployment`
+    /// sets it verbatim, never the k8s-safe slug `k8s_deployment_name` uses
+    /// for the resource's own name).
+    pub name: String,
+    pub namespace: String,
+    /// Reconstructed `oab-{namespace}-{name}` — **not** the on-cluster
+    /// Deployment resource name (`oab-{slug(name)}`, no namespace embedded;
+    /// see `k8s_driver.rs`'s module doc on why the two differ). This is the
+    /// form the New Fleet / Add Instance wizard actually writes into
+    /// `fleets.toml`'s `members` array (`console/src/deploy.ts`'s `service =
+    /// oab-${namespace}-${name}`, unconditional for both runtimes) — carried
+    /// so `FleetBinding::includes` matches the same way it does for ECS.
+    pub service_name: String,
+    pub desired: i32,
+}
+
+fn k8s_oab_name(dep: &k8s_openapi::api::apps::v1::Deployment) -> Option<String> {
+    dep.metadata.labels.as_ref()?.get("oab/name").cloned()
+}
+
+/// List every OAB-managed Deployment (`oab/name` label present) in one
+/// `(context, namespace)` — the k8s counterpart to [`observe_services`].
+pub async fn observe_k8s_services(
+    context: Option<&str>,
+    namespace: &str,
+) -> anyhow::Result<Vec<K8sServiceStatus>> {
+    use k8s_openapi::api::apps::v1::Deployment as K8sDeployment;
+    use kube::api::{Api, ListParams};
+
+    let client = k8s_client_for(context).await?;
+    let api: Api<K8sDeployment> = Api::namespaced(client, namespace);
+    let list = api
+        .list(&ListParams::default().labels("oab/name"))
+        .await
+        .map_err(|e| anyhow::anyhow!("failed to list k8s deployments in '{namespace}': {e}"))?;
+    Ok(list
+        .items
+        .iter()
+        .filter_map(|dep| {
+            let name = k8s_oab_name(dep)?;
+            let desired = dep.spec.as_ref().and_then(|s| s.replicas).unwrap_or(0);
+            Some(K8sServiceStatus {
+                service_name: format!("oab-{namespace}-{name}"),
+                name,
+                namespace: namespace.to_string(),
+                desired,
+            })
+        })
+        .collect())
+}
+
+fn k8s_pod_phase(pod: &k8s_openapi::api::core::v1::Pod) -> agent_lifecycle::k8s::PodPhase {
+    use agent_lifecycle::k8s::PodPhase;
+    match pod.status.as_ref().and_then(|s| s.phase.as_deref()) {
+        Some("Running") => PodPhase::Running,
+        Some("Succeeded") => PodPhase::Succeeded,
+        Some("Failed") => PodPhase::Failed,
+        Some("Unknown") => PodPhase::Unknown,
+        // "Pending" or not yet reported — not started, not a fault.
+        _ => PodPhase::Pending,
+    }
+}
+
+fn k8s_pod_ready(pod: &k8s_openapi::api::core::v1::Pod) -> bool {
+    pod.status
+        .as_ref()
+        .and_then(|s| s.conditions.as_ref())
+        .into_iter()
+        .flatten()
+        .any(|c| c.type_ == "Ready" && c.status == "True")
+}
+
+fn k8s_pod_ready_check_defined(pod: &k8s_openapi::api::core::v1::Pod) -> bool {
+    pod.spec
+        .as_ref()
+        .into_iter()
+        .flat_map(|s| s.containers.iter())
+        .any(|c| c.readiness_probe.is_some())
+}
+
+fn k8s_pod_crash_loop_back_off(pod: &k8s_openapi::api::core::v1::Pod) -> bool {
+    pod.status
+        .as_ref()
+        .and_then(|s| s.container_statuses.as_ref())
+        .into_iter()
+        .flatten()
+        .any(|cs| {
+            cs.state
+                .as_ref()
+                .and_then(|st| st.waiting.as_ref())
+                .is_some_and(|w| w.reason.as_deref() == Some("CrashLoopBackOff"))
+        })
+}
+
+/// One-shot approximation of the `identity_verified` latch (same caveat as
+/// ECS's `latched_verified`: the real latch needs CP-persisted history) — a
+/// Pod counts as verified once it's reached Running, or is already
+/// Terminating (it must have run to get there).
+fn k8s_latched_verified(pod: &k8s_openapi::api::core::v1::Pod) -> bool {
+    matches!(k8s_pod_phase(pod), agent_lifecycle::k8s::PodPhase::Running)
+        || pod.metadata.deletion_timestamp.is_some()
+}
+
+/// Map a live k8s Pod onto the canonical [`AgentState`], via
+/// [`agent_lifecycle::k8s::K8sDriver`] — the k8s counterpart to
+/// `instance_phase`'s ECS derivation. `lease_valid`/`accepting_work` are
+/// CP-level, not yet k8s-observable, so they default to valid/admitting —
+/// same stance `instance_phase` takes for ECS today.
+pub fn k8s_instance_phase(pod: &k8s_openapi::api::core::v1::Pod, verified_before: bool) -> AgentState {
+    use agent_lifecycle::k8s::{K8sDriver, K8sPod};
+    use agent_lifecycle::RuntimeDriver;
+
+    let native = K8sPod {
+        phase: k8s_pod_phase(pod),
+        deletion_timestamp_set: pod.metadata.deletion_timestamp.is_some(),
+        ready: k8s_pod_ready(pod),
+        ready_check_defined: k8s_pod_ready_check_defined(pod),
+        crash_loop_back_off: k8s_pod_crash_loop_back_off(pod),
+        lease_valid: true,
+        accepting_work: true,
+    };
+    K8sDriver.project(&native, verified_before).classify()
+}
+
+/// Observe one k8s Deployment end-to-end: replica counters + per-Pod phase —
+/// the k8s counterpart to [`observe_deployment`]. `service` matches either
+/// the reconstructed `oab-{namespace}-{name}` form or the bare agent name
+/// (same dual-match spirit as ECS's `resolve_service`).
+pub async fn observe_k8s_deployment(
+    context: Option<&str>,
+    namespace: &str,
+    service: &str,
+) -> anyhow::Result<Option<Deployment>> {
+    use k8s_openapi::api::apps::v1::Deployment as K8sDeployment;
+    use k8s_openapi::api::core::v1::Pod;
+    use kube::api::{Api, ListParams};
+
+    let client = k8s_client_for(context).await?;
+    let dep_api: Api<K8sDeployment> = Api::namespaced(client.clone(), namespace);
+    let deployments = dep_api
+        .list(&ListParams::default().labels("oab/name"))
+        .await
+        .map_err(|e| anyhow::anyhow!("failed to list k8s deployments in '{namespace}': {e}"))?;
+    let Some(dep) = deployments.items.into_iter().find(|d| {
+        let Some(name) = k8s_oab_name(d) else { return false };
+        service == format!("oab-{namespace}-{name}") || service == name
+    }) else {
+        return Ok(None);
+    };
+    let name = k8s_oab_name(&dep).unwrap_or_default();
+    let desired = dep.spec.as_ref().and_then(|s| s.replicas).unwrap_or(0);
+
+    let selector = dep
+        .spec
+        .as_ref()
+        .and_then(|s| s.selector.match_labels.as_ref())
+        .map(|m| {
+            m.iter()
+                .map(|(k, v)| format!("{k}={v}"))
+                .collect::<Vec<_>>()
+                .join(",")
+        })
+        .unwrap_or_default();
+    let pod_api: Api<Pod> = Api::namespaced(client, namespace);
+    let pods = pod_api
+        .list(&ListParams::default().labels(&selector))
+        .await
+        .map_err(|e| anyhow::anyhow!("failed to list pods for k8s deployment '{name}': {e}"))?;
+
+    let instances: Vec<InstancePhase> = pods
+        .items
+        .iter()
+        .map(|p| InstancePhase {
+            id: p.metadata.uid.clone().unwrap_or_default(),
+            phase: k8s_instance_phase(p, k8s_latched_verified(p)),
+        })
+        .collect();
+    let ready = instances
+        .iter()
+        .filter(|p| p.phase == AgentState::Running)
+        .count() as i32;
+
+    Ok(Some(Deployment {
+        name,
+        namespace: namespace.to_string(),
+        desired,
+        current: instances.len() as i32,
+        ready,
+        instances,
+    }))
+}
+
 // ---- Effective runtime identity/context (ADR: Per-Fleet managing identity) --
 //
 // Read-only observation of *who this control plane is actually acting as*. The