feat(cp): FleetBinding config + per-fleet credential switch (ADR #19 slice 2)

crates/oab-mcp/src/main.rs

@@ -22,15 +22,23 @@ use rmcp::model::{
 use rmcp::service::{RequestContext, RoleServer};
 use rmcp::{ErrorData as McpError, ServiceExt};
 use serde_json::{json, Map, Value};
-use std::sync::Arc;
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
 use studio_cp as scp;
 
 /// The control-plane server. Holds the shared AWS config and the default
 /// cluster; cheap to clone (one handler per session).
 #[derive(Clone)]
 struct OabMcp {
+    /// Fallback AWS config (the default credential chain), used when no fleet
+    /// binding governs the target cluster.
     aws: aws_config::SdkConfig,
     default_cluster: String,
+    /// Declarative fleet → managing-credential bindings (ADR: Per-Fleet
+    /// managing identity).
+    bindings: Arc<scp::FleetBindings>,
+    /// Per-cluster resolved configs, memoized so a binding is resolved once.
+    resolved: Arc<Mutex<HashMap<String, aws_config::SdkConfig>>>,
 }
 
 fn as_map(v: Value) -> Arc<Map<String, Value>> {
@@ -134,10 +142,12 @@ fn tools() -> Vec<Tool> {
         ),
         Tool::new(
             "runtime_context",
-            "Show the effective runtime identity/context this control plane resolved: the acting principal (STS caller ARN), its kind (role vs static user), account (scope), region (location), and a best-effort credential-source hint. 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: the acting principal (STS caller ARN), its kind (role vs static user), account (scope), region (location), a best-effort credential-source hint, and the fleet binding in effect (if any). Read-only; answers \"who am I acting as, against what account?\" and surfaces silent credential fallback.",
             as_map(json!({
                 "type": "object",
-                "properties": {}
+                "properties": {
+                    "cluster": { "type": "string", "description": "Resolve the identity the fleet binding for this cluster selects (defaults to the server's configured cluster)." }
+                }
             })),
         ),
     ]
@@ -178,9 +188,30 @@ impl OabMcp {
             .unwrap_or_else(|| self.default_cluster.clone())
     }
 
+    /// 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
+    /// cluster's calls run under its credential, not whatever ambient
+    /// `[default]` the chain resolves first.
+    async fn aws_for(&self, cluster: &str) -> aws_config::SdkConfig {
+        let binding = match self.bindings.for_cluster(cluster) {
+            Some(b) if b.profile.is_some() || b.region.is_some() => b.clone(),
+            _ => return self.aws.clone(),
+        };
+        if let Some(cfg) = self.resolved.lock().unwrap().get(cluster) {
+            return cfg.clone();
+        }
+        let cfg = scp::resolve_binding_config(&binding).await;
+        self.resolved
+            .lock()
+            .unwrap()
+            .insert(cluster.to_string(), cfg.clone());
+        cfg
+    }
+
     async fn t_list(&self, args: &Map<String, Value>) -> Result<Value> {
         let cluster = self.cluster(args);
-        let svcs = scp::observe_services(&self.aws, &cluster).await?;
+        let svcs = scp::observe_services(&self.aws_for(&cluster).await, &cluster).await?;
         let deployments: Vec<Value> = svcs
             .iter()
             .map(|s| {
@@ -205,7 +236,7 @@ impl OabMcp {
             .get("service")
             .and_then(Value::as_str)
             .ok_or_else(|| anyhow::anyhow!("missing required arg: service"))?;
-        match scp::observe_deployment(&self.aws, &cluster, service).await? {
+        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 })),
         }
@@ -215,15 +246,17 @@ impl OabMcp {
         let cluster = self.cluster(args);
         let services: Vec<String> = match args.get("service").and_then(Value::as_str) {
             Some(s) => vec![s.to_string()],
-            None => scp::observe_services(&self.aws, &cluster)
+            None => scp::observe_services(&self.aws_for(&cluster).await, &cluster)
                 .await?
                 .into_iter()
                 .map(|s| s.name)
                 .collect(),
         };
         let mut instances = Vec::new();
         for svc in services {
-            if let Some(d) = scp::observe_deployment(&self.aws, &cluster, &svc).await? {
+            if let Some(d) =
+                scp::observe_deployment(&self.aws_for(&cluster).await, &cluster, &svc).await?
+            {
                 for inst in &d.instances {
                     instances.push(json!({
                         "service": d.name,
@@ -265,8 +298,15 @@ impl OabMcp {
             .unwrap_or(0);
         let since_ms = now_ms - since_minutes * 60_000;
 
-        let events =
-            scp::observe_events(&self.aws, &log_group, &cluster, service, since_ms, limit).await?;
+        let events = scp::observe_events(
+            &self.aws_for(&cluster).await,
+            &log_group,
+            &cluster,
+            service,
+            since_ms,
+            limit,
+        )
+        .await?;
         Ok(json!({
             "cluster": cluster,
             "log_group": log_group,
@@ -283,7 +323,8 @@ impl OabMcp {
             .and_then(Value::as_str)
             .ok_or_else(|| anyhow::anyhow!("missing required arg: manifest_yaml"))?;
         let wait = args.get("wait").and_then(Value::as_bool).unwrap_or(false);
-        let report = scp::apply_deployment(&self.aws, manifest, &cluster, wait).await?;
+        let report =
+            scp::apply_deployment(&self.aws_for(&cluster).await, manifest, &cluster, wait).await?;
         Ok(json!({ "ok": true, "services_applied": report.services.len() }))
     }
 
@@ -301,21 +342,40 @@ impl OabMcp {
             args.get("size")
                 .and_then(Value::as_i64)
                 .ok_or_else(|| anyhow::anyhow!("missing or invalid arg: size"))? as i32;
-        scp::scale_deployment(&self.aws, &cluster, namespace, name, size).await?;
+        scp::scale_deployment(
+            &self.aws_for(&cluster).await,
+            &cluster,
+            namespace,
+            name,
+            size,
+        )
+        .await?;
         Ok(
             json!({ "ok": true, "cluster": cluster, "namespace": namespace, "name": name, "size": size }),
         )
     }
 
-    async fn t_runtime_context(&self, _args: &Map<String, Value>) -> Result<Value> {
-        let ctx = scp::observe_identity(&self.aws).await?;
+    async fn t_runtime_context(&self, args: &Map<String, Value>) -> Result<Value> {
+        let cluster = self.cluster(args);
+        let aws = self.aws_for(&cluster).await;
+        let ctx = scp::observe_identity(&aws).await?;
+        let binding = self.bindings.for_cluster(&cluster).map(|b| {
+            json!({
+                "name": b.name,
+                "profile": b.profile,
+                "region": b.region,
+                "expected_principal": b.expected_principal,
+            })
+        });
         Ok(json!({
+            "cluster": cluster,
             "principal": ctx.principal,
             "principal_kind": ctx.principal_kind,
             "scope": ctx.scope,
             "location": ctx.location,
             "source": ctx.source,
             "caller_id": ctx.caller_id,
+            "binding": binding,
         }))
     }
 
@@ -333,7 +393,14 @@ impl OabMcp {
             .get("namespace")
             .and_then(Value::as_str)
             .unwrap_or("default");
-        scp::delete_deployment(&self.aws, resource, name, &cluster, namespace).await?;
+        scp::delete_deployment(
+            &self.aws_for(&cluster).await,
+            resource,
+            name,
+            &cluster,
+            namespace,
+        )
+        .await?;
         Ok(json!({ "ok": true, "resource": resource, "name": name }))
     }
 }
@@ -398,9 +465,23 @@ impl ServerHandler for OabMcp {
 async fn main() -> anyhow::Result<()> {
     let aws = aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await;
     let default_cluster = std::env::var("OAB_CLUSTER").unwrap_or_else(|_| "oab".to_string());
+    // Fleet → managing-credential bindings are strictly opt-in: a missing file
+    // yields an empty set. Warn on stderr only (stdout is the MCP JSON-RPC wire).
+    let bindings = match scp::default_bindings_path() {
+        Some(path) => scp::load_bindings(&path).unwrap_or_else(|e| {
+            eprintln!(
+                "warning: failed to load fleet bindings from {}: {e:#}",
+                path.display()
+            );
+            scp::FleetBindings::default()
+        }),
+        None => scp::FleetBindings::default(),
+    };
     let server = OabMcp {
         aws,
         default_cluster,
+        bindings: Arc::new(bindings),
+        resolved: Arc::new(Mutex::new(HashMap::new())),
     };
     let service = server.serve(rmcp::transport::stdio()).await?;
     service.waiting().await?;

crates/studio-cp/Cargo.toml

@@ -11,4 +11,7 @@ agent-lifecycle = { path = "../agent-lifecycle" }
 aws-config = "1.5"
 aws-sdk-sts = "1"
 anyhow = "1.0"
+serde = { version = "1.0", features = ["derive"] }
+toml = "0.8"
+dirs = "6"
 tokio = { version = "1.40", features = ["rt-multi-thread", "macros"] }

crates/studio-cp/src/lib.rs

@@ -235,6 +235,83 @@ pub async fn observe_identity(
     })
 }
 
+// ---- Fleet → managing-credential binding (ADR: Per-Fleet managing identity) --
+//
+// The *declarative* side of the loop: which credential should manage which
+// fleet/cluster. Operator config, deliberately separate from the Fleet Store
+// (observed membership/lease state); the two may fold together later. Selecting
+// a binding is credential *selection*, not per-caller authz.
+
+/// A declarative binding of a managed fleet/cluster to the credential that
+/// should manage it. Profile-first (assume-role is later work).
+#[derive(Debug, Clone, serde::Deserialize)]
+pub struct FleetBinding {
+    /// Fleet label (display / selection).
+    #[serde(default)]
+    pub name: String,
+    /// ECS cluster this binding governs — the match key against a call's target.
+    pub cluster: String,
+    /// Region to pin for this fleet.
+    #[serde(default)]
+    pub region: Option<String>,
+    /// Named AWS profile that supplies the managing credential (profile-first).
+    #[serde(default)]
+    pub profile: Option<String>,
+    /// Expected effective principal ARN — reconciled against the resolved
+    /// identity later (IdentityMismatch). Optional.
+    #[serde(default)]
+    pub expected_principal: Option<String>,
+}
+
+/// Parsed fleet-binding file: a list of `[[fleet]]` tables.
+#[derive(Debug, Clone, Default, serde::Deserialize)]
+pub struct FleetBindings {
+    #[serde(default, rename = "fleet")]
+    pub fleets: Vec<FleetBinding>,
+}
+
+impl FleetBindings {
+    /// The binding governing `cluster`, if any (first match wins).
+    pub fn for_cluster(&self, cluster: &str) -> Option<&FleetBinding> {
+        self.fleets.iter().find(|b| b.cluster == cluster)
+    }
+}
+
+/// Default fleet-binding config path: `$OAB_FLEETS_CONFIG`, else
+/// `<config-dir>/oab-studio/fleets.toml` (`~/.config/oab-studio/fleets.toml`).
+pub fn default_bindings_path() -> Option<std::path::PathBuf> {
+    if let Ok(p) = std::env::var("OAB_FLEETS_CONFIG") {
+        return Some(std::path::PathBuf::from(p));
+    }
+    dirs::config_dir().map(|d| d.join("oab-studio").join("fleets.toml"))
+}
+
+/// Load fleet bindings from `path`. A missing file is **not** an error — it
+/// yields an empty set (every target falls back to the default credential
+/// chain), so bindings are strictly opt-in.
+pub fn load_bindings(path: &std::path::Path) -> anyhow::Result<FleetBindings> {
+    match std::fs::read_to_string(path) {
+        Ok(content) => Ok(toml::from_str(&content)?),
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(FleetBindings::default()),
+        Err(e) => Err(e.into()),
+    }
+}
+
+/// Resolve the AWS config a binding selects — its named profile (profile-first)
+/// and/or pinned region, layered on the standard chain. This is the **switch**:
+/// calls for this fleet act as the bound credential instead of whatever the
+/// ambient `[default]` resolves first.
+pub async fn resolve_binding_config(binding: &FleetBinding) -> aws_config::SdkConfig {
+    let mut loader = aws_config::defaults(aws_config::BehaviorVersion::latest());
+    if let Some(profile) = &binding.profile {
+        loader = loader.profile_name(profile.as_str());
+    }
+    if let Some(region) = &binding.region {
+        loader = loader.region(aws_config::Region::new(region.clone()));
+    }
+    loader.load().await
+}
+
 // ---- Write side (ADR-2 write model) ------------------------------------
 //
 // The read side above observes; these mutate. Each is a thin passthrough to
@@ -373,4 +450,26 @@ mod tests {
         );
         assert_eq!(principal_kind("arn:aws:iam::1:root"), "unknown");
     }
+
+    #[test]
+    fn bindings_parse_and_match_by_cluster() {
+        let doc = r#"
+[[fleet]]
+name = "prod"
+cluster = "oab"
+region = "ap-east-2"
+profile = "orca-prod"
+
+[[fleet]]
+name = "sg"
+cluster = "oab-sg"
+profile = "appier-sg"
+"#;
+        let b: FleetBindings = toml::from_str(doc).expect("parse");
+        assert_eq!(b.fleets.len(), 2);
+        let prod = b.for_cluster("oab").expect("prod binding");
+        assert_eq!(prod.profile.as_deref(), Some("orca-prod"));
+        assert_eq!(prod.region.as_deref(), Some("ap-east-2"));
+        assert!(b.for_cluster("nope").is_none());
+    }
 }