diff --git a/crates/oab-mcp/src/lib.rs b/crates/oab-mcp/src/lib.rs
index 82f9707..48300ea 100644
--- a/crates/oab-mcp/src/lib.rs
+++ b/crates/oab-mcp/src/lib.rs
@@ -146,7 +146,7 @@ pub fn tools() -> Vec<Tool> {
         ),
         Tool::new(
             "deploy_provision",
-            "Provision an agent from the compose library: compose template ⊕ overlay into a file bundle, push it to the agent's S3 artifacts prefix, and redeploy the ECS service at the chosen image tag. Reuses the agent's stored manifest for networking/resources/secrets, so the agent must already have been created.",
+            "Provision an agent from the compose library: compose template ⊕ overlay into a file bundle and push it to the agent's S3 artifacts prefix (bundle carrier — shared regardless of provider). If the agent already has a stored manifest, patches its image/bundle and re-applies (networking/resources/secrets/runtime ride along unchanged). If not, builds a fresh manifest with sensible defaults and creates the agent — this now works for a genuinely brand-new agent, not just a redeploy of one already created via `oabctl create`. `provider` (default \"aws\") selects the target: \"aws\" applies via ECS (`fleet`/`cluster` select the credential); \"k8s\" applies via the given kubeconfig `context` instead, with `expected_principal` optionally naming a service account (`system:serviceaccount:<ns>:<name>` — the bare name becomes the pod's serviceAccountName; unset uses the namespace's default).",
             as_map(json!({
                 "type": "object",
                 "properties": {
@@ -156,8 +156,11 @@ pub fn tools() -> Vec<Tool> {
                     "name": { "type": "string", "description": "Agent / service name (service = oab-{namespace}-{name})." },
                     "namespace": { "type": "string", "description": "Namespace (default \"default\")." },
                     "image_tag": { "type": "string", "description": "Image tag override (defaults to the bundle's own image tag)." },
-                    "fleet": { "type": "string", "description": "Fleet name (see fleet_config): targets the fleet's cluster and managing credential; a write to a service outside the fleet's members is refused. Overrides the cluster arg." },
-                    "cluster": { "type": "string", "description": "ECS cluster (defaults to the server's configured cluster)." }
+                    "provider": { "type": "string", "description": "\"aws\" (default) or \"k8s\" — which driver applies the result." },
+                    "fleet": { "type": "string", "description": "AWS only. Fleet name (see fleet_config): targets the fleet's cluster and managing credential; a write to a service outside the fleet's members is refused. Overrides the cluster arg." },
+                    "cluster": { "type": "string", "description": "AWS only. ECS cluster (defaults to the server's configured cluster)." },
+                    "context": { "type": "string", "description": "k8s only. Kubeconfig context to apply through. Omit to use the kubeconfig's current-context." },
+                    "expected_principal": { "type": "string", "description": "k8s only, optional. `system:serviceaccount:<namespace>:<name>` to set the pod's service account; unset uses the namespace's default." }
                 },
                 "required": ["library", "template", "name"]
             })),
@@ -546,8 +549,6 @@ impl OabMcp {
     }
 
     async fn t_provision(&self, args: &Map<String, Value>) -> Result<Value> {
-        let t = self.target(args)?;
-        let cluster = t.cluster.clone();
         let namespace = args
             .get("namespace")
             .and_then(Value::as_str)
@@ -569,6 +570,44 @@ impl OabMcp {
         )
         .map_err(|e| anyhow::anyhow!("invalid library: {e}"))?;
 
+        // studio#104: k8s dispatch. `context`/`expected_principal` come as
+        // direct args — the console's identity form already collects them
+        // (context/namespace/service-account <select>s, #108/#109), there's
+        // no existing K8sFleetBinding to resolve them from for a brand-new
+        // fleet (unlike AWS's `fleet` + `self.target()` — a fleet-scoped k8s
+        // lookup is future work for redeploys into an *existing* k8s fleet,
+        // not needed for this dispatch to exist).
+        if args.get("provider").and_then(Value::as_str) == Some("k8s") {
+            let context = args.get("context").and_then(Value::as_str);
+            let expected_principal = args.get("expected_principal").and_then(Value::as_str);
+            let outcome = scp::provision_from_library_k8s(
+                &self.aws,
+                context,
+                namespace,
+                name,
+                &library,
+                template,
+                overlay,
+                image,
+                expected_principal,
+            )
+            .await?;
+            return Ok(json!({
+                "ok": true,
+                "context": context,
+                "namespace": namespace,
+                "name": name,
+                "image": outcome.image,
+                "digest": outcome.digest,
+                "objects": outcome.objects,
+                "action": outcome.action,
+                "services_applied": outcome.services_applied,
+            }));
+        }
+
+        let t = self.target(args)?;
+        let cluster = t.cluster.clone();
+
         // Same fleet-scope guard as scale/delete: a fleet handle only provisions
         // its own members, so a scoped call can't reach a co-located non-member.
         let service_name = format!("oab-{namespace}-{name}");
diff --git a/crates/oabctl/src/create.rs b/crates/oabctl/src/create.rs
index 7724128..27155dd 100644
--- a/crates/oabctl/src/create.rs
+++ b/crates/oabctl/src/create.rs
@@ -201,9 +201,14 @@ async fn store_secret(sm: &SmClient, name: &str, value: &str) -> Result<()> {
     }
 }
 
-struct VpcInfo { id: String, label: String }
-
-async fn list_vpcs(ec2: &Ec2Client) -> Result<Vec<VpcInfo>> {
+/// A VPC discovered in the account/region, with a human-readable label
+/// (id + Name tag + CIDR + default marker) for interactive selection.
+/// `pub` (studio#111): reused by the provision-from-scratch path to pick a
+/// sensible default VPC without prompting, same discovery `oabctl create`'s
+/// wizard uses interactively.
+pub struct VpcInfo { pub id: String, pub label: String, pub is_default: bool }
+
+pub async fn list_vpcs(ec2: &Ec2Client) -> Result<Vec<VpcInfo>> {
     let resp = ec2.describe_vpcs().send().await?;
     Ok(resp.vpcs().iter().map(|v| {
         let id = v.vpc_id().unwrap_or_default().to_string();
@@ -214,13 +219,37 @@ async fn list_vpcs(ec2: &Ec2Client) -> Result<Vec<VpcInfo>> {
             .and_then(|t| t.value())
             .unwrap_or("unnamed");
         let label = format!("{id} ({name}, {cidr}{})", if is_default { ", default" } else { "" });
-        VpcInfo { id, label }
+        VpcInfo { id, label, is_default }
     }).collect())
 }
 
-struct SubnetInfo { id: String, az: String, kind: String, has_nat: bool }
+/// Pick a VPC with **no interactive prompt**, for the provision-from-scratch
+/// path (studio#111): the account/region's default VPC, if there is exactly
+/// one. Unlike subnet/security-group, `run()`'s wizard has no default here
+/// (it always asks) — a "default VPC" is the closest zero-prompt equivalent,
+/// but it's genuinely ambiguous when there's none or more than one, so this
+/// errors rather than guessing at that point (a caller that has an explicit
+/// VPC choice — e.g. from per-fleet config — should skip this and pass it
+/// directly to `select_subnets`/`default_security_group` instead).
+pub async fn default_vpc(ec2: &Ec2Client) -> Result<VpcInfo> {
+    let vpcs = list_vpcs(ec2).await?;
+    let defaults: Vec<VpcInfo> = vpcs.into_iter().filter(|v| v.is_default).collect();
+    match defaults.len() {
+        1 => Ok(defaults.into_iter().next().unwrap()),
+        0 => anyhow::bail!("no default VPC in this account/region — configure a VPC explicitly"),
+        n => anyhow::bail!("{n} VPCs marked default — configure a VPC explicitly, can't pick automatically"),
+    }
+}
+
+/// A subnet candidate, already classified private/public + NAT reachability.
+/// `pub` (studio#111): see `VpcInfo`.
+pub struct SubnetInfo { pub id: String, pub az: String, pub kind: String, pub has_nat: bool }
 
-async fn select_subnets(ec2: &Ec2Client, vpc_id: &str) -> Result<Vec<SubnetInfo>> {
+/// Auto-select up to 3 subnets (one per AZ), preferring private+NAT >
+/// private > public — this already has zero interactive prompting, it's the
+/// exact "sensible default" the provision-from-scratch path (studio#111)
+/// needs, just needed to be reachable outside `create.rs`.
+pub async fn select_subnets(ec2: &Ec2Client, vpc_id: &str) -> Result<Vec<SubnetInfo>> {
     let subnets_resp = ec2.describe_subnets()
         .filters(aws_sdk_ec2::types::Filter::builder().name("vpc-id").values(vpc_id).build())
         .send().await?;
@@ -282,9 +311,10 @@ async fn select_subnets(ec2: &Ec2Client, vpc_id: &str) -> Result<Vec<SubnetInfo>
     Ok(selected)
 }
 
-struct SgInfo { id: String, name: String }
+/// `pub` (studio#111): see `VpcInfo`.
+pub struct SgInfo { pub id: String, pub name: String }
 
-async fn list_security_groups(ec2: &Ec2Client, vpc_id: &str) -> Result<Vec<SgInfo>> {
+pub async fn list_security_groups(ec2: &Ec2Client, vpc_id: &str) -> Result<Vec<SgInfo>> {
     let resp = ec2.describe_security_groups()
         .filters(aws_sdk_ec2::types::Filter::builder().name("vpc-id").values(vpc_id).build())
         .send().await?;
@@ -296,6 +326,54 @@ async fn list_security_groups(ec2: &Ec2Client, vpc_id: &str) -> Result<Vec<SgInf
     }).collect())
 }
 
+/// Pick a security group for a new agent with **no interactive prompt** —
+/// the fully-automatic counterpart to `run()`'s "Create new (oab-{name})"
+/// default choice (studio#111: the provision-from-scratch path needs a
+/// sensible default here, not a prompt). Reuses an existing `oab-{name}`
+/// group if one's already there (idempotent — safe to call again for the
+/// same agent), otherwise creates it. Unlike `run()`'s interactive flow,
+/// this never offers picking a different existing group; that choice stays
+/// CLI-only.
+pub async fn default_security_group(ec2: &Ec2Client, vpc_id: &str, name: &str) -> Result<String> {
+    let sg_name = format!("oab-{name}");
+    let existing = list_security_groups(ec2, vpc_id).await?;
+    if let Some(sg) = existing.iter().find(|sg| sg.name == sg_name) {
+        return Ok(sg.id.clone());
+    }
+    let resp = ec2.create_security_group()
+        .group_name(&sg_name)
+        .description(format!("OAB agent {name}"))
+        .vpc_id(vpc_id)
+        .send().await
+        .context("failed to create security group")?;
+    Ok(resp.group_id().unwrap_or_default().to_string())
+}
+
+/// Subnet + security-group IDs for a new agent's ECS networking, with no
+/// interactive prompt. The single entry point studio#111's provision-from-
+/// scratch path needs — takes just an `SdkConfig` (not an `Ec2Client`) so
+/// callers outside `oabctl` (e.g. `studio-cp`) don't need `aws-sdk-ec2` as a
+/// direct dependency just to reach this; ADR-2's "RuntimeDriver is the only
+/// layer with vendor terms" principle extends here too — `Ec2Client` stays
+/// an `oabctl`-internal detail.
+pub struct DefaultNetworking {
+    pub vpc_id: String,
+    pub subnets: Vec<String>,
+    pub security_groups: Vec<String>,
+}
+
+pub async fn default_networking(config: &aws_config::SdkConfig, name: &str) -> Result<DefaultNetworking> {
+    let ec2 = Ec2Client::new(config);
+    let vpc = default_vpc(&ec2).await?;
+    let subnets = select_subnets(&ec2, &vpc.id).await?;
+    let sg = default_security_group(&ec2, &vpc.id, name).await?;
+    Ok(DefaultNetworking {
+        vpc_id: vpc.id,
+        subnets: subnets.into_iter().map(|s| s.id).collect(),
+        security_groups: vec![sg],
+    })
+}
+
 fn generate_config(_backend: &str, name: &str, namespace: &str, stt_enabled: bool) -> String {
     let stt_section = if stt_enabled {
         r#"[stt]
diff --git a/crates/oabctl/src/lib.rs b/crates/oabctl/src/lib.rs
index ef552c3..02c3a49 100644
--- a/crates/oabctl/src/lib.rs
+++ b/crates/oabctl/src/lib.rs
@@ -51,7 +51,7 @@ mod bootstrap;
 mod cli;
 mod config;
 mod control_plane;
-mod create;
+pub mod create;
 mod delete;
 pub mod driver;
 pub mod events;
diff --git a/crates/oabctl/src/studio_api.rs b/crates/oabctl/src/studio_api.rs
index a73679a..5097ae0 100644
--- a/crates/oabctl/src/studio_api.rs
+++ b/crates/oabctl/src/studio_api.rs
@@ -229,6 +229,59 @@ pub async fn provision(
         .context("failed to apply manifest during provision")
 }
 
+/// [`provision`], but takes an already-built [`crate::manifest::OABServiceManifest`]
+/// instead of YAML text — for a caller (studio-cp's provision-from-scratch
+/// path, studio#111) that constructs one directly rather than starting from
+/// a stored manifest's YAML. Keeps the YAML serialization an `oabctl`-internal
+/// detail rather than making every caller depend on `serde_yaml` themselves.
+pub async fn provision_manifest(
+    config: &aws_config::SdkConfig,
+    cluster: &str,
+    manifest: &crate::manifest::OABServiceManifest,
+    objects: &[(String, Vec<u8>)],
+    control_plane_bucket: Option<&str>,
+) -> Result<crate::apply::ApplyReport> {
+    let yaml = serde_yaml::to_string(manifest).context("failed to serialize manifest")?;
+    provision(config, cluster, &yaml, objects, control_plane_bucket).await
+}
+
+/// [`provision_manifest`], but applies through [`crate::k8s_driver::K8sDriver`]
+/// instead of [`crate::driver::EcsDriver`] (studio#104's k8s `deploy_provision`
+/// dispatch, resumed once #111 landed a create-vs-redeploy branch point that
+/// doesn't care which driver applies the result).
+///
+/// Takes **two** separate credential contexts, unlike every other function in
+/// this module — this is inherent to how k8s provisioning works here, not
+/// something to simplify away: the bundle carrier (`hooks.pre_seed`, studio#97
+/// slice 3c) still goes through S3 regardless of runtime, so `aws_config` is
+/// still needed for [`push_bundle`]; `context` is the kubeconfig context
+/// [`crate::k8s_driver::K8sDriver::apply`] actually applies the Deployment
+/// through. A k8s-provisioned agent's bundle upload and its Deployment apply
+/// are two different systems with two different credentials — there's no way
+/// to collapse this to a single config the way the ECS path's one
+/// `aws_config::SdkConfig` covers both S3 and ECS.
+pub async fn provision_k8s(
+    aws_config: &aws_config::SdkConfig,
+    context: Option<&str>,
+    manifest: &crate::manifest::OABServiceManifest,
+    objects: &[(String, Vec<u8>)],
+    control_plane_bucket: Option<&str>,
+) -> Result<crate::apply::ApplyReport> {
+    // Same ordering as `provision()`: bundle first (idempotent puts), so the
+    // carrier is ready before the Deployment comes up and pre_seed runs.
+    push_bundle(aws_config, control_plane_bucket, objects).await?;
+
+    let driver = crate::k8s_driver::K8sDriver::from_context(context).await?;
+    let opts = crate::driver::ProvisionOptions {
+        control_plane_bucket: control_plane_bucket.map(str::to_string),
+        wait: false,
+    };
+    driver
+        .apply(std::slice::from_ref(manifest), &opts)
+        .await
+        .context("failed to apply k8s manifest during provision")
+}
+
 /// Load the desired `OABService` manifest oabctl persists at
 /// `manifests/{namespace}/{name}.yaml` in the control-plane bucket. Returns
 /// `Ok(None)` when the agent has no stored manifest yet (never applied); other
diff --git a/crates/studio-cp/src/lib.rs b/crates/studio-cp/src/lib.rs
index a9204c8..071861a 100644
--- a/crates/studio-cp/src/lib.rs
+++ b/crates/studio-cp/src/lib.rs
@@ -1021,12 +1021,74 @@ pub struct ProvisionOutcome {
     pub action: String,
 }
 
+/// Where a brand-new agent's `spec.configFrom` should point — the same
+/// `artifacts/{namespace}/{name}/config.toml` key `Bundle::artifact_objects`
+/// already uploads a copy of the composed config.toml to, and the same
+/// convention `oabctl create`'s wizard uses for its own generated manifest.
+fn default_config_from_uri(bucket: &str, namespace: &str, name: &str) -> String {
+    format!("s3://{bucket}/{}/config.toml", studio_compose::artifacts_prefix(namespace, name))
+}
+
+/// Build a fresh `OABServiceManifest` for an agent that has never been
+/// provisioned before — `redeploy()` can only patch an *already-stored*
+/// manifest (studio#111: it has no "create the first one" path). Fields not
+/// derivable from the compose bundle get sensible, zero-prompt defaults:
+/// networking from `oabctl::create::default_networking` (same discovery
+/// `oabctl create`'s CLI wizard uses, minus the interactive prompts),
+/// resources 256/512 (the CLI wizard's own default), `FARGATE`/`X86_64`
+/// (the schema's own `#[serde(default)]` values), empty `secrets` (valid —
+/// no ECS-level Secrets Manager env injection; whatever the template's own
+/// config.toml needs is the operator's concern, unchanged from how
+/// `oabctl create` already works), no ingress.
+async fn build_default_manifest(
+    aws_config: &aws_config::SdkConfig,
+    namespace: &str,
+    name: &str,
+    image: &str,
+    bucket: &str,
+) -> anyhow::Result<oabctl::manifest::OABServiceManifest> {
+    let net = oabctl::create::default_networking(aws_config, name).await?;
+    let config_from = default_config_from_uri(bucket, namespace, name);
+    Ok(oabctl::manifest::OABServiceManifest {
+        api_version: "oab.dev/v2".to_string(),
+        kind: "OABService".to_string(),
+        metadata: oabctl::manifest::Metadata {
+            name: name.to_string(),
+            namespace: namespace.to_string(),
+            generation: 0,
+        },
+        spec: oabctl::manifest::Spec {
+            image: image.to_string(),
+            resources: oabctl::manifest::Resources {
+                cpu: "256".to_string(),
+                memory: "512".to_string(),
+            },
+            config_from,
+            bundle_from: None,
+            bootstrap_from: None,
+            secrets: std::collections::HashMap::new(),
+            runtime: oabctl::manifest::Runtime::Ecs(oabctl::manifest::EcsRuntime {
+                capacity_provider: "FARGATE".to_string(),
+                architecture: "X86_64".to_string(),
+                task_role_arn: None,
+                networking: oabctl::manifest::EcsNetworking {
+                    subnets: net.subnets,
+                    security_groups: net.security_groups,
+                    assign_public_ip: false,
+                },
+            }),
+            ingress: None,
+        },
+    })
+}
+
 /// Provision an agent from the compose **library**: compose `template ⊕ overlay`,
 /// then **redeploy** — push the bundle to the agent's artifacts prefix and apply
 /// its stored manifest at the chosen image tag (agent-deployment ADR slice 2,
 /// path A). Networking/resources/secrets ride along from the stored manifest, so
-/// this is the "update this agent to new persona / skills / image" path; the
-/// agent must already have been `create`d.
+/// this is the "update this agent to new persona / skills / image" path — unless
+/// the agent has never been provisioned before, in which case a fresh manifest is
+/// built instead ([`build_default_manifest`], studio#111).
 ///
 /// `image_override` (when non-empty) wins over the bundle's own default image tag.
 pub async fn provision_from_library(
@@ -1075,16 +1137,175 @@ pub async fn provision_from_library(
 
     let digest = bundle.digest();
 
-    let report = oabctl::studio_api::redeploy(
-        aws_config,
-        cluster,
-        namespace,
-        name,
-        Some(&image),
-        &objects,
-        Some(&bucket),
-    )
-    .await?;
+    // studio#111: `redeploy()` only patches an *existing* stored manifest —
+    // it errors if this agent has never been provisioned. Check first and
+    // build a fresh manifest in that case, rather than surfacing that error
+    // to the console on every "+ New fleet" first deploy.
+    let existing_manifest =
+        oabctl::studio_api::load_manifest(aws_config, namespace, name, Some(&bucket)).await?;
+    let report = match existing_manifest {
+        Some(_) => {
+            oabctl::studio_api::redeploy(
+                aws_config,
+                cluster,
+                namespace,
+                name,
+                Some(&image),
+                &objects,
+                Some(&bucket),
+            )
+            .await?
+        }
+        None => {
+            let mut manifest = build_default_manifest(aws_config, namespace, name, &image, &bucket).await?;
+            manifest.spec.bundle_from = Some(oabctl::studio_api::bundle_from_uri(&bucket, namespace, name));
+            oabctl::studio_api::provision_manifest(aws_config, cluster, &manifest, &objects, Some(&bucket))
+                .await?
+        }
+    };
+
+    Ok(ProvisionOutcome {
+        image,
+        digest,
+        objects: objects.len(),
+        services_applied: report.services.len(),
+        action: report
+            .services
+            .first()
+            .map(|s| format!("{:?}", s.action))
+            .unwrap_or_default(),
+    })
+}
+
+/// Extract the bare service-account name from an `expected_principal` string
+/// in `system:serviceaccount:<namespace>:<name>` form — the format
+/// `K8sFleetBinding.expected_principal` holds when the "+ New fleet" wizard's
+/// service-account picker set it (studio#104). `KubernetesRuntime.
+/// service_account` wants just the bare name (`k8s_driver::build_deployment`
+/// sets it as `pod_spec.service_account_name` directly), not the qualified
+/// form `k8s_principal_kind` classifies against. `None` — or a plain
+/// username, not a service account — means the pod uses the namespace's
+/// default service account.
+fn k8s_service_account_from_principal(expected_principal: Option<&str>) -> Option<String> {
+    expected_principal
+        .and_then(|p| p.strip_prefix("system:serviceaccount:"))
+        .and_then(|rest| rest.split_once(':'))
+        .map(|(_namespace, name)| name.to_string())
+}
+
+/// Build a fresh k8s `OABServiceManifest` — the `Runtime::Kubernetes`
+/// counterpart to [`build_default_manifest`]. No VPC/subnet/security-group
+/// concept (that's ECS-specific networking); k8s's per-fleet placement is
+/// `context` (which cluster) and `namespace` (both already resolved by the
+/// caller from `K8sFleetBinding`, not part of the manifest itself — mirrors
+/// how AWS's `cluster` is a driver-construction parameter, not a manifest
+/// field). `node_selector`/`tolerations` default empty; `service_account`
+/// comes from `expected_principal` when it names one, else the namespace's
+/// default applies (same "unset = use default" contract `default_security_group`-
+/// style AWS defaults don't have an equivalent of, since k8s already has one
+/// built in).
+async fn build_default_k8s_manifest(
+    namespace: &str,
+    name: &str,
+    image: &str,
+    bucket: &str,
+    expected_principal: Option<&str>,
+) -> anyhow::Result<oabctl::manifest::OABServiceManifest> {
+    let config_from = default_config_from_uri(bucket, namespace, name);
+    Ok(oabctl::manifest::OABServiceManifest {
+        api_version: "oab.dev/v2".to_string(),
+        kind: "OABService".to_string(),
+        metadata: oabctl::manifest::Metadata {
+            name: name.to_string(),
+            namespace: namespace.to_string(),
+            generation: 0,
+        },
+        spec: oabctl::manifest::Spec {
+            image: image.to_string(),
+            resources: oabctl::manifest::Resources {
+                cpu: "256".to_string(),
+                memory: "512".to_string(),
+            },
+            config_from,
+            bundle_from: None,
+            bootstrap_from: None,
+            secrets: std::collections::HashMap::new(),
+            runtime: oabctl::manifest::Runtime::Kubernetes(oabctl::manifest::KubernetesRuntime {
+                node_selector: std::collections::HashMap::new(),
+                service_account: k8s_service_account_from_principal(expected_principal),
+                tolerations: Vec::new(),
+            }),
+            ingress: None,
+        },
+    })
+}
+
+/// [`provision_from_library`], but for a k8s-driven fleet — studio#104's
+/// `deploy_provision` k8s dispatch, resumed once #111 gave both drivers a
+/// shared, provider-agnostic create-vs-redeploy branch point
+/// (`load_manifest` is just an S3 key, it doesn't care which `Runtime` variant
+/// the stored manifest holds).
+///
+/// Compose/bundle-upload is identical to the AWS path (the bundle carrier —
+/// `hooks.pre_seed`, studio#97 slice 3c — is provider-agnostic, still S3
+/// regardless of where the agent actually runs) — some duplication with
+/// `provision_from_library` here is deliberate for now rather than risk
+/// reworking that already-landed function's shape again; worth revisiting
+/// once both paths are proven.
+///
+/// `expected_principal` is `K8sFleetBinding.expected_principal` — see
+/// [`k8s_service_account_from_principal`] for how it maps onto the manifest.
+pub async fn provision_from_library_k8s(
+    aws_config: &aws_config::SdkConfig,
+    context: Option<&str>,
+    namespace: &str,
+    name: &str,
+    library: &Library,
+    template: &str,
+    overlay: Option<&str>,
+    image_override: Option<&str>,
+    expected_principal: Option<&str>,
+) -> anyhow::Result<ProvisionOutcome> {
+    let mut bundle = studio_compose::compose_named(library, template, overlay)
+        .map_err(|e| anyhow::anyhow!("compose failed: {e}"))?;
+    let image = image_override
+        .filter(|s| !s.is_empty())
+        .map(str::to_string)
+        .unwrap_or_else(|| bundle.image_tag.clone());
+
+    let bucket = oabctl::resolve_bucket(aws_config, None).await?;
+    let zip_uri = oabctl::studio_api::bundle_zip_uri(&bucket, namespace, name);
+    match bundle.files.get_mut("config.toml") {
+        Some(bytes) => *bytes = oabctl::studio_api::inject_pre_seed_hook(bytes, &zip_uri)?,
+        None => anyhow::bail!("composed bundle for {namespace}/{name} has no config.toml — cannot wire hooks.pre_seed"),
+    }
+
+    let mut objects = bundle.artifact_objects(namespace, name);
+    let zip_key = format!(
+        "{}/{}",
+        studio_compose::artifacts_prefix(namespace, name),
+        oabctl::studio_api::BUNDLE_ZIP_FILENAME
+    );
+    objects.push((zip_key, bundle.zip_bytes()));
+
+    let digest = bundle.digest();
+
+    let existing_manifest =
+        oabctl::studio_api::load_manifest(aws_config, namespace, name, Some(&bucket)).await?;
+    let mut manifest = match existing_manifest {
+        // Redeploy: preserve everything else about the stored manifest
+        // (including its k8s runtime config), only bump the image — same
+        // guarantee the AWS path's `redeploy()` gives.
+        Some(mut stored) => {
+            stored.spec.image = image.clone();
+            stored
+        }
+        None => build_default_k8s_manifest(namespace, name, &image, &bucket, expected_principal).await?,
+    };
+    manifest.spec.bundle_from = Some(oabctl::studio_api::bundle_from_uri(&bucket, namespace, name));
+
+    let report =
+        oabctl::studio_api::provision_k8s(aws_config, context, &manifest, &objects, Some(&bucket)).await?;
 
     Ok(ProvisionOutcome {
         image,
@@ -1607,4 +1828,32 @@ aws_access_key_id = AKIA...
         let names = parse_aws_credentials_names(text);
         assert_eq!(names, vec!["default".to_string(), "oab-fleet".to_string()]);
     }
+
+    #[test]
+    fn default_config_from_uri_matches_artifact_objects_key() {
+        // Must land at the same key Bundle::artifact_objects uploads
+        // config.toml to (and the same convention oabctl create's wizard
+        // uses for its own generated manifest) — a brand-new agent's
+        // configFrom pointing anywhere else would read nothing at boot.
+        assert_eq!(
+            default_config_from_uri("oab-control-plane-123", "prod", "orca"),
+            "s3://oab-control-plane-123/artifacts/prod/orca/config.toml",
+        );
+    }
+
+    #[test]
+    fn k8s_service_account_from_principal_extracts_bare_name() {
+        assert_eq!(
+            k8s_service_account_from_principal(Some("system:serviceaccount:dev:oab-agent")),
+            Some("oab-agent".to_string()),
+        );
+    }
+
+    #[test]
+    fn k8s_service_account_from_principal_none_for_plain_username() {
+        // A plain username (not a service account) or unset both mean "use
+        // the namespace's default service account" — not an error.
+        assert_eq!(k8s_service_account_from_principal(Some("brett.chien")), None);
+        assert_eq!(k8s_service_account_from_principal(None), None);
+    }
 }