diff --git a/console/index.html b/console/index.html
index f3321b1..6329d00 100644
--- a/console/index.html
+++ b/console/index.html
@@ -92,17 +92,43 @@
                 <input id="deploy-fleet-name" class="compose-input" type="text" placeholder="e.g. support-fleet" spellcheck="false" />
               </label>
               <label
-                >Region
-                <input id="deploy-region" class="compose-input" type="text" placeholder="e.g. ap-east-2" spellcheck="false" />
-              </label>
-              <label
-                >Credential profile
-                <input id="deploy-profile" class="compose-input" type="text" placeholder="e.g. oab-fleet" spellcheck="false" />
-              </label>
-              <label
-                >Principal (optional)
-                <input id="deploy-principal" class="compose-input" type="text" placeholder="arn:aws:iam::…" spellcheck="false" />
+                >Provider
+                <select id="deploy-provider" class="compose-select">
+                  <option value="aws" selected>AWS (ECS)</option>
+                  <option value="k8s">Kubernetes</option>
+                </select>
               </label>
+              <div id="deploy-aws-fields">
+                <label
+                  >Region
+                  <input id="deploy-region" class="compose-input" type="text" placeholder="e.g. ap-east-2" spellcheck="false" />
+                </label>
+                <label
+                  >Credential profile
+                  <input id="deploy-profile" class="compose-input" type="text" placeholder="e.g. oab-fleet" spellcheck="false" />
+                </label>
+                <label
+                  >Principal (optional)
+                  <input id="deploy-principal" class="compose-input" type="text" placeholder="arn:aws:iam::…" spellcheck="false" />
+                </label>
+              </div>
+              <div id="deploy-k8s-fields" hidden>
+                <label
+                  >Context
+                  <select id="deploy-k8s-context" class="compose-select"></select>
+                </label>
+                <label
+                  >Namespace
+                  <input id="deploy-k8s-namespace" class="compose-input" type="text" list="deploy-k8s-namespace-options" placeholder="pick existing or type a new one" spellcheck="false" />
+                  <datalist id="deploy-k8s-namespace-options"></datalist>
+                </label>
+                <label
+                  >Service account (optional)
+                  <select id="deploy-k8s-service-account" class="compose-select">
+                    <option value="">— namespace default —</option>
+                  </select>
+                </label>
+              </div>
               <div class="compose-actions">
                 <button type="submit">Next: first instance &rarr;</button>
                 <span class="compose-status" id="deploy-identity-status"></span>
diff --git a/console/src/deploy.ts b/console/src/deploy.ts
index 0e181bf..5f3dc0d 100644
--- a/console/src/deploy.ts
+++ b/console/src/deploy.ts
@@ -37,6 +37,19 @@ function fillOptions(sel: HTMLSelectElement, names: string[], keepNoneFirst: boo
   sel.innerHTML = opts.join("");
 }
 
+// list_k8s_contexts / list_namespaces response shapes (oab-mcp, studio#104) —
+// kept minimal (just what this panel reads), not the tools' full contract.
+interface K8sContextsResponse {
+  contexts: { name: string }[];
+  current_context: string | null;
+}
+interface K8sNamespacesResponse {
+  namespaces: string[];
+}
+interface K8sServiceAccountsResponse {
+  service_accounts: string[];
+}
+
 export type DeployMode = { kind: "new-fleet" } | { kind: "add-instance"; fleetName: string };
 
 // What the panel reports back once a deploy + fleets.toml write both succeed —
@@ -75,9 +88,22 @@ export function initDeployPanel(deps: DeployPanelDeps): DeployPanelHandle | null
   const cancelBtn = document.getElementById("deploy-cancel") as HTMLButtonElement | null;
   const identityForm = document.getElementById("deploy-identity-form") as HTMLFormElement | null;
   const nameInput = document.getElementById("deploy-fleet-name") as HTMLInputElement | null;
+  const providerSel = document.getElementById("deploy-provider") as HTMLSelectElement | null;
+  const awsFieldsEl = document.getElementById("deploy-aws-fields");
   const regionInput = document.getElementById("deploy-region") as HTMLInputElement | null;
   const profileInput = document.getElementById("deploy-profile") as HTMLInputElement | null;
   const principalInput = document.getElementById("deploy-principal") as HTMLInputElement | null;
+  const k8sFieldsEl = document.getElementById("deploy-k8s-fields");
+  const k8sContextSel = document.getElementById("deploy-k8s-context") as HTMLSelectElement | null;
+  // Namespace is a text input with a <datalist> of what already exists (not a
+  // <select>) — per studio#104's design, a brand-new namespace is a valid
+  // choice and a plain select can't express "not in this list yet".
+  const k8sNamespaceInput = document.getElementById("deploy-k8s-namespace") as HTMLInputElement | null;
+  const k8sNamespaceOptions = document.getElementById("deploy-k8s-namespace-options") as HTMLDataListElement | null;
+  // Service account, unlike namespace, must already exist for k8s to accept
+  // it as a pod's serviceAccountName — so (unlike namespace) a plain <select>
+  // is the right shape here, no free-text escape hatch needed.
+  const k8sServiceAccountSel = document.getElementById("deploy-k8s-service-account") as HTMLSelectElement | null;
   const identityStatusEl = document.getElementById("deploy-identity-status");
   const composeSection = document.getElementById("deploy-compose");
   const composeHeading = document.getElementById("deploy-compose-heading");
@@ -97,9 +123,16 @@ export function initDeployPanel(deps: DeployPanelDeps): DeployPanelHandle | null
     !cancelBtn ||
     !identityForm ||
     !nameInput ||
+    !providerSel ||
+    !awsFieldsEl ||
     !regionInput ||
     !profileInput ||
     !principalInput ||
+    !k8sFieldsEl ||
+    !k8sContextSel ||
+    !k8sNamespaceInput ||
+    !k8sNamespaceOptions ||
+    !k8sServiceAccountSel ||
     !composeSection ||
     !tmplSel ||
     !ovlSel ||
@@ -130,8 +163,98 @@ export function initDeployPanel(deps: DeployPanelDeps): DeployPanelHandle | null
     setStatus(identityStatusEl, "");
     setStatus(previewStatusEl, "");
     setStatus(deployStatusEl, "");
+    // identityForm.reset() puts <select id="deploy-provider"> back to its
+    // `selected` default ("aws"), but doesn't touch the field-group `hidden`
+    // attributes this panel manages by hand — sync those too.
+    showProviderFields(providerSel.value);
+  };
+
+  // Toggle the AWS/k8s field groups per studio#104's design: switching
+  // providers resets which group is visible; field *values* aren't cleared
+  // here (identityForm.reset() already did that on open/close) since the
+  // two groups have no overlapping semantics to accidentally carry over.
+  const showProviderFields = (provider: string): void => {
+    awsFieldsEl.hidden = provider !== "aws";
+    k8sFieldsEl.hidden = provider !== "k8s";
+  };
+
+  const loadK8sNamespaces = async (): Promise<void> => {
+    const invoke = tauriInvoke();
+    if (!invoke) return;
+    const context = k8sContextSel.value || undefined;
+    try {
+      const res = await invoke<K8sNamespacesResponse>(
+        "list_namespaces",
+        context ? { context } : {},
+      );
+      k8sNamespaceOptions.innerHTML = res.namespaces
+        .map((n) => `<option value="${escapeHtml(n)}"></option>`)
+        .join("");
+    } catch (e) {
+      // Non-fatal: the namespace field is a free-text input either way (list_
+      // namespaces failing just means no autocomplete suggestions).
+      setStatus(identityStatusEl, `namespace list unavailable: ${errText(e)}`, "err");
+    }
+  };
+
+  // Service account is scoped to (context, namespace) and per #104's design
+  // fails *silently* — unlike context/namespace, any error here (including an
+  // RBAC-denied list, which is common against a scoped-down cluster identity)
+  // means "leave it unset" (the namespace's default service account applies),
+  // not something worth surfacing a status message for.
+  const loadK8sServiceAccounts = async (): Promise<void> => {
+    const defaultOption = '<option value="">— namespace default —</option>';
+    const invoke = tauriInvoke();
+    const namespace = k8sNamespaceInput.value.trim();
+    if (!invoke || !namespace) {
+      k8sServiceAccountSel.innerHTML = defaultOption;
+      return;
+    }
+    const context = k8sContextSel.value || undefined;
+    try {
+      const res = await invoke<K8sServiceAccountsResponse>(
+        "list_service_accounts",
+        context ? { context, namespace } : { namespace },
+      );
+      k8sServiceAccountSel.innerHTML =
+        defaultOption +
+        res.service_accounts
+          .map((sa) => `<option value="${escapeHtml(sa)}">${escapeHtml(sa)}</option>`)
+          .join("");
+    } catch {
+      k8sServiceAccountSel.innerHTML = defaultOption;
+    }
   };
 
+  const loadK8sContexts = async (): Promise<void> => {
+    const invoke = tauriInvoke();
+    if (!invoke) return;
+    try {
+      const res = await invoke<K8sContextsResponse>("list_k8s_contexts");
+      const opts = ['<option value="">— kubeconfig current-context —</option>'];
+      for (const c of res.contexts) {
+        const label = c.name === res.current_context ? `${c.name} (current)` : c.name;
+        opts.push(`<option value="${escapeHtml(c.name)}">${escapeHtml(label)}</option>`);
+      }
+      k8sContextSel.innerHTML = opts.join("");
+    } catch (e) {
+      setStatus(identityStatusEl, `k8s context list unavailable: ${errText(e)}`, "err");
+    }
+    void loadK8sNamespaces();
+  };
+
+  providerSel.addEventListener("change", () => {
+    showProviderFields(providerSel.value);
+    if (providerSel.value === "k8s") void loadK8sContexts();
+  });
+  k8sContextSel.addEventListener("change", () => {
+    void loadK8sNamespaces();
+    void loadK8sServiceAccounts();
+  });
+  // "change" (fires on commit/blur), not "input" (every keystroke) — avoids a
+  // tool call per character typed into the namespace field.
+  k8sNamespaceInput.addEventListener("change", () => void loadK8sServiceAccounts());
+
   const loadLibraryAndPickers = async (): Promise<void> => {
     const invoke = tauriInvoke();
     if (!invoke) {
@@ -184,6 +307,18 @@ export function initDeployPanel(deps: DeployPanelDeps): DeployPanelHandle | null
       setStatus(identityStatusEl, "fleet name is required", "err");
       return;
     }
+    // k8s provisioning isn't wired end-to-end yet (deploy_provision has no
+    // k8s dispatch — see openabdev/studio#104's discussion of why this isn't
+    // just "swap the driver"). Block here rather than let the wizard proceed
+    // into a Compose step that would fail at the final deploy_provision call.
+    if (providerSel.value === "k8s") {
+      setStatus(
+        identityStatusEl,
+        "Kubernetes provisioning isn't available yet — tracked in openabdev/studio#104",
+        "err",
+      );
+      return;
+    }
     identityForm.hidden = true;
     composeSection.hidden = false;
     if (composeHeading) composeHeading.textContent = "Step 2 — first instance";
diff --git a/console/src/fleetToml.ts b/console/src/fleetToml.ts
index 824da9c..dc4a348 100644
--- a/console/src/fleetToml.ts
+++ b/console/src/fleetToml.ts
@@ -7,7 +7,10 @@
 // regex-based (not a full TOML parser) so it's unit-testable and only ever
 // touches the one array/block it means to.
 
-function quote(s: string): string {
+// Exported so fleetsK8sToml.ts (fleets-k8s.toml's client-side edits, same
+// `[fleet.<name>]` shape) can reuse it instead of duplicating a one-line
+// helper.
+export function quote(s: string): string {
   return JSON.stringify(s);
 }
 
diff --git a/console/src/fleetsK8sToml.test.ts b/console/src/fleetsK8sToml.test.ts
new file mode 100644
index 0000000..8b42b82
--- /dev/null
+++ b/console/src/fleetsK8sToml.test.ts
@@ -0,0 +1,57 @@
+import { describe, it, expect } from "vitest";
+import { appendMember, appendK8sFleetBlock } from "./fleetsK8sToml";
+
+describe("appendMember (reused from fleetToml.ts)", () => {
+  it("works against fleets-k8s.toml's [fleet.<name>] shape too", () => {
+    const text = `[fleet.orbstack-dev]
+context = "orbstack"
+namespace = "dev"
+members = ["scratch-agent"]
+`;
+    const out = appendMember(text, "orbstack-dev", "scratch-agent-2");
+    expect(out).toContain('members = ["scratch-agent", "scratch-agent-2"]');
+    expect(out).toContain('context = "orbstack"');
+    expect(out).toContain('namespace = "dev"');
+  });
+});
+
+describe("appendK8sFleetBlock", () => {
+  it("appends a new [fleet.<name>] block with context, namespace, members, expected_principal", () => {
+    const out = appendK8sFleetBlock('default_cluster = "oab"\n', {
+      name: "orbstack-dev",
+      member: "oab-dev-scratch-agent",
+      context: "orbstack",
+      namespace: "dev",
+      expectedPrincipal: "system:serviceaccount:dev:oab-agent",
+    });
+    expect(out).toContain("[fleet.orbstack-dev]");
+    expect(out).toContain('context = "orbstack"');
+    expect(out).toContain('namespace = "dev"');
+    expect(out).toContain('members = ["oab-dev-scratch-agent"]');
+    expect(out).toContain('expected_principal = "system:serviceaccount:dev:oab-agent"');
+  });
+
+  it("omits context and expected_principal when not provided, but always writes namespace", () => {
+    const out = appendK8sFleetBlock("", {
+      name: "orca-k8s",
+      member: "oab-prod-orca",
+      context: null,
+      namespace: "prod",
+      expectedPrincipal: null,
+    });
+    expect(out).not.toContain("context =");
+    expect(out).not.toContain("expected_principal =");
+    expect(out).toContain('namespace = "prod"');
+  });
+
+  it("separates the new block from existing content with exactly one blank line", () => {
+    const out = appendK8sFleetBlock('default_cluster = "oab"\n', {
+      name: "x",
+      member: "m",
+      context: null,
+      namespace: "ns",
+      expectedPrincipal: null,
+    });
+    expect(out).toBe('default_cluster = "oab"\n\n[fleet.x]\nnamespace = "ns"\nmembers = ["m"]\n');
+  });
+});
diff --git a/console/src/fleetsK8sToml.ts b/console/src/fleetsK8sToml.ts
new file mode 100644
index 0000000..37552b2
--- /dev/null
+++ b/console/src/fleetsK8sToml.ts
@@ -0,0 +1,39 @@
+// Pure text-level edits to fleets-k8s.toml's `[fleet.<name>]` blocks — the
+// k8s counterpart to fleetToml.ts, same rationale (studio#104: k8s_fleet_
+// config_write has no partial/append primitive, so the client computes the
+// new/edited TOML and calls it with the full updated text).
+//
+// `[fleet.<name>]` block lookup/append-member is identical between fleets.
+// toml and fleets-k8s.toml (same table shape, same `members = [...]` array —
+// neither `findFleetBlock` nor `appendMember` reference any AWS-specific
+// field), so this module reuses fleetToml.ts's `appendMember` rather than
+// duplicating it. Only "create a brand-new fleet block" differs, since the
+// two files' required/optional fields differ (context+namespace vs
+// region+profile).
+
+import { quote, appendMember } from "./fleetToml";
+
+export { appendMember };
+
+export interface NewK8sFleetEntry {
+  name: string;
+  member: string;
+  context: string | null;
+  namespace: string;
+  expectedPrincipal: string | null;
+}
+
+// Append a brand-new `[fleet.<name>]` block to the end of the file, with the
+// one member — the first instance just deployed. `context` and
+// `expected_principal` are optional fields, omitted rather than written as
+// empty strings (mirrors fleetToml.ts's appendFleetBlock).
+export function appendK8sFleetBlock(text: string, entry: NewK8sFleetEntry): string {
+  const lines = [`[fleet.${entry.name}]`];
+  if (entry.context) lines.push(`context = ${quote(entry.context)}`);
+  lines.push(`namespace = ${quote(entry.namespace)}`);
+  lines.push(`members = [${quote(entry.member)}]`);
+  if (entry.expectedPrincipal) lines.push(`expected_principal = ${quote(entry.expectedPrincipal)}`);
+  const block = `${lines.join("\n")}\n`;
+  const trimmed = text.replace(/\s*$/, "");
+  return trimmed.length ? `${trimmed}\n\n${block}` : block;
+}
diff --git a/crates/oab-mcp/src/lib.rs b/crates/oab-mcp/src/lib.rs
index 3b8d8cd..82f9707 100644
--- a/crates/oab-mcp/src/lib.rs
+++ b/crates/oab-mcp/src/lib.rs
@@ -223,6 +223,39 @@ pub fn tools() -> Vec<Tool> {
                 "properties": {}
             })),
         ),
+        Tool::new(
+            "list_namespaces",
+            "List namespaces in the cluster a kubeconfig context resolves to. Read-only; backs the New Fleet wizard's namespace <select> (with a manual-entry fallback for a namespace that doesn't exist yet — this can only list what's already there).",
+            as_map(json!({
+                "type": "object",
+                "properties": {
+                    "context": { "type": "string", "description": "Kubeconfig context name. Omit to use the kubeconfig's current-context." }
+                }
+            })),
+        ),
+        Tool::new(
+            "list_service_accounts",
+            "List service accounts in one namespace of a kubeconfig context. Read-only; backs the New Fleet wizard's optional service-account <select>. Any failure here (including an RBAC-denied list) should be treated by the caller as \"leave it unset\" — the namespace's `default` service account applies — not surfaced as an error.",
+            as_map(json!({
+                "type": "object",
+                "properties": {
+                    "context": { "type": "string", "description": "Kubeconfig context name. Omit to use the kubeconfig's current-context." },
+                    "namespace": { "type": "string", "description": "k8s namespace to list service accounts in." }
+                },
+                "required": ["namespace"]
+            })),
+        ),
+        Tool::new(
+            "k8s_fleet_config_write",
+            "Persist the whole k8s fleet-binding config (fleets-k8s.toml, separate from AWS's fleets.toml) from raw TOML `text`. Validates the text parses before writing — a bad edit never lands on disk — and the bytes are stored verbatim, so comments/layout are preserved. Returns the parsed fleets (name, context, namespace, members, expected_principal) plus the raw text. Write tool: overwrites the operator's fleets-k8s.toml.",
+            as_map(json!({
+                "type": "object",
+                "properties": {
+                    "text": { "type": "string", "description": "Full TOML document for fleets-k8s.toml (a list of [fleet.<name>] tables)." }
+                },
+                "required": ["text"]
+            })),
+        ),
     ]
 }
 
@@ -337,6 +370,9 @@ impl OabMcp {
             "fleet_config_write" => self.t_fleet_write(args),
             "list_aws_profiles" => self.t_list_aws_profiles(args),
             "list_k8s_contexts" => self.t_list_k8s_contexts(args),
+            "list_namespaces" => self.t_list_namespaces(args).await,
+            "list_service_accounts" => self.t_list_service_accounts(args).await,
+            "k8s_fleet_config_write" => self.t_k8s_fleet_write(args),
             other => anyhow::bail!("unknown tool {other:?}"),
         }
     }
@@ -748,6 +784,60 @@ impl OabMcp {
         }))
     }
 
+    /// Namespace discovery (studio#104): backs the New Fleet wizard's
+    /// namespace `<select>`.
+    async fn t_list_namespaces(&self, args: &Map<String, Value>) -> Result<Value> {
+        let context = args.get("context").and_then(Value::as_str);
+        let namespaces = scp::list_namespaces(context).await?;
+        Ok(json!({ "namespaces": namespaces }))
+    }
+
+    /// Service-account discovery (studio#104): backs the New Fleet wizard's
+    /// optional service-account `<select>`. Errors (including RBAC-denied)
+    /// propagate as a normal tool error — per the design, the caller treats
+    /// any failure here as "leave it unset", not something to surface.
+    async fn t_list_service_accounts(&self, args: &Map<String, Value>) -> Result<Value> {
+        let context = args.get("context").and_then(Value::as_str);
+        let namespace = args
+            .get("namespace")
+            .and_then(Value::as_str)
+            .ok_or_else(|| anyhow::anyhow!("missing required arg: namespace"))?;
+        let service_accounts = scp::list_service_accounts(context, namespace).await?;
+        Ok(json!({ "service_accounts": service_accounts }))
+    }
+
+    /// Write tool: persist the whole `fleets-k8s.toml` from the editor's
+    /// `text` after validating it parses, mirroring `t_fleet_write`'s
+    /// AWS-side shape. Unlike AWS bindings, k8s bindings aren't cached
+    /// anywhere in `OabMcp` yet (nothing here dispatches provisioning to
+    /// `K8sDriver` yet either — that's a separate item), so this is a plain
+    /// validate-then-write with no in-memory state to invalidate.
+    fn t_k8s_fleet_write(&self, args: &Map<String, Value>) -> Result<Value> {
+        let text = args
+            .get("text")
+            .and_then(Value::as_str)
+            .ok_or_else(|| anyhow::anyhow!("missing required arg: text"))?;
+        let path = scp::default_k8s_bindings_path()
+            .ok_or_else(|| anyhow::anyhow!("no k8s fleet config path resolved; cannot write bindings"))?;
+        let bindings = scp::save_k8s_bindings_text(&path, text)?;
+        let fleets: Vec<Value> = bindings
+            .fleets
+            .iter()
+            .map(|b| json!({
+                "name": b.name,
+                "context": b.context,
+                "namespace": b.namespace,
+                "members": b.members,
+                "expected_principal": b.expected_principal,
+            }))
+            .collect();
+        Ok(json!({
+            "path": path.display().to_string(),
+            "fleets": fleets,
+            "text": text,
+        }))
+    }
+
     async fn t_delete(&self, args: &Map<String, Value>) -> Result<Value> {
         let t = self.target(args)?;
         let cluster = t.cluster.clone();
@@ -830,7 +920,7 @@ mod tests {
             .iter()
             .map(|t| t["name"].as_str().expect("tool has a name").to_string())
             .collect();
-        assert_eq!(names.len(), 13);
+        assert_eq!(names.len(), 16);
         for expected in [
             "deploy_list",
             "deploy_get",
@@ -845,6 +935,9 @@ mod tests {
             "fleet_config_write",
             "list_aws_profiles",
             "list_k8s_contexts",
+            "list_namespaces",
+            "list_service_accounts",
+            "k8s_fleet_config_write",
         ] {
             assert!(names.contains(&expected.to_string()), "missing {expected}");
         }
diff --git a/crates/studio-cp/src/lib.rs b/crates/studio-cp/src/lib.rs
index bbfd49e..a9204c8 100644
--- a/crates/studio-cp/src/lib.rs
+++ b/crates/studio-cp/src/lib.rs
@@ -295,14 +295,9 @@ pub async fn observe_k8s_identity(context: Option<&str>) -> anyhow::Result<Runti
         None => String::new(),
     };
 
-    let options = kube::config::KubeConfigOptions {
-        context: context.map(str::to_string),
-        ..Default::default()
-    };
-    let config = kube::Config::from_kubeconfig(&options)
+    let client = k8s_client_for(context)
         .await
         .map_err(|e| anyhow::anyhow!("failed to resolve kubeconfig context '{context_name}': {e}"))?;
-    let client = kube::Client::try_from(config).map_err(|e| anyhow::anyhow!("failed to build k8s client: {e}"))?;
 
     let api: Api<SelfSubjectReview> = Api::all(client);
     let review = api
@@ -519,6 +514,53 @@ pub fn list_k8s_contexts() -> K8sContextsResult {
     }
 }
 
+async fn k8s_client_for(context: Option<&str>) -> anyhow::Result<kube::Client> {
+    let options = kube::config::KubeConfigOptions {
+        context: context.map(str::to_string),
+        ..Default::default()
+    };
+    let config = kube::Config::from_kubeconfig(&options)
+        .await
+        .map_err(|e| anyhow::anyhow!("failed to resolve kubeconfig context: {e}"))?;
+    kube::Client::try_from(config).map_err(|e| anyhow::anyhow!("failed to build k8s client: {e}"))
+}
+
+/// List namespaces in the cluster the given kubeconfig context (or the
+/// ambient current-context) resolves to. Backs the New Fleet wizard's
+/// namespace `<select>` — a manual-entry fallback covers a namespace that
+/// doesn't exist yet (this can only list what's already there).
+pub async fn list_namespaces(context: Option<&str>) -> anyhow::Result<Vec<String>> {
+    use k8s_openapi::api::core::v1::Namespace;
+    use kube::api::{Api, ListParams};
+
+    let client = k8s_client_for(context).await?;
+    let api: Api<Namespace> = Api::all(client);
+    let list = api
+        .list(&ListParams::default())
+        .await
+        .map_err(|e| anyhow::anyhow!("failed to list namespaces: {e}"))?;
+    Ok(list.items.into_iter().filter_map(|ns| ns.metadata.name).collect())
+}
+
+/// List service accounts in one namespace of the given kubeconfig context.
+/// Backs the New Fleet wizard's (optional) service-account `<select>` — the
+/// caller falls back to leaving it unset (the namespace's `default` service
+/// account applies) on any error here, including an RBAC-denied `list`, so
+/// this deliberately doesn't distinguish failure reasons the way
+/// `list_aws_profiles`/`list_k8s_contexts` do.
+pub async fn list_service_accounts(context: Option<&str>, namespace: &str) -> anyhow::Result<Vec<String>> {
+    use k8s_openapi::api::core::v1::ServiceAccount;
+    use kube::api::{Api, ListParams};
+
+    let client = k8s_client_for(context).await?;
+    let api: Api<ServiceAccount> = Api::namespaced(client, namespace);
+    let list = api
+        .list(&ListParams::default())
+        .await
+        .map_err(|e| anyhow::anyhow!("failed to list service accounts: {e}"))?;
+    Ok(list.items.into_iter().filter_map(|sa| sa.metadata.name).collect())
+}
+
 // ---- Fleet → managing-credential binding (ADR: Per-Fleet managing identity) --
 //
 // The *declarative* side of the loop: which credential should manage which
@@ -767,6 +809,15 @@ pub struct K8sFleetBinding {
     /// `FleetBinding`'s empty-members-means-everything convention).
     #[serde(default)]
     pub members: Vec<String>,
+    /// Expected principal to verify the resolved k8s identity against —
+    /// same name and verify semantics as `FleetBinding::expected_principal`
+    /// (`observe_identity`/`identity_matches` for AWS), just compared against
+    /// `observe_k8s_identity`'s `SelfSubjectReview`-derived principal instead
+    /// of an STS caller ARN. Typically `system:serviceaccount:<ns>:<name>`
+    /// for a service account, or a plain username. `None` = no identity
+    /// verification for this fleet (same "unset = don't check" contract).
+    #[serde(default)]
+    pub expected_principal: Option<String>,
 }
 
 impl K8sFleetBinding {
@@ -786,6 +837,8 @@ struct K8sFleetBody {
     namespace: String,
     #[serde(default)]
     members: Vec<String>,
+    #[serde(default)]
+    expected_principal: Option<String>,
 }
 
 #[derive(serde::Deserialize)]
@@ -805,6 +858,7 @@ impl From<K8sFleetsDoc> for K8sFleetBindings {
                     context: b.context,
                     namespace: b.namespace,
                     members: b.members,
+                    expected_principal: b.expected_principal,
                 })
                 .collect(),
         }
@@ -1347,6 +1401,24 @@ namespace = "prod"
         assert_eq!(prod.context, None);
     }
 
+    #[test]
+    fn k8s_fleet_expected_principal_parses_and_defaults_to_none() {
+        let doc = r#"
+[fleet.dev]
+namespace = "dev"
+expected_principal = "system:serviceaccount:dev:oab-agent"
+
+[fleet.unset]
+namespace = "prod"
+"#;
+        let b: K8sFleetBindings = toml::from_str(doc).expect("parse");
+        assert_eq!(
+            b.get("dev").expect("dev fleet").expected_principal.as_deref(),
+            Some("system:serviceaccount:dev:oab-agent")
+        );
+        assert_eq!(b.get("unset").expect("unset fleet").expected_principal, None);
+    }
+
     #[test]
     fn k8s_binding_includes_matches_by_name_or_whole_namespace() {
         let scoped = K8sFleetBinding {
@@ -1354,6 +1426,7 @@ namespace = "prod"
             context: Some("orbstack".into()),
             namespace: "dev".into(),
             members: vec!["scratch-agent".into()],
+            expected_principal: None,
         };
         assert!(scoped.includes("scratch-agent"));
         assert!(!scoped.includes("other-agent"));
@@ -1363,6 +1436,7 @@ namespace = "prod"
             context: None,
             namespace: "prod".into(),
             members: vec![],
+            expected_principal: None,
         };
         assert!(whole.includes("anything"));
     }