diff --git a/README.md b/README.md
index 17313f2..a95e79d 100644
--- a/README.md
+++ b/README.md
@@ -214,14 +214,21 @@ hops config install --repo hops-ops/aws-auto-eks-cluster --version v0.11.0
 
 ### Cluster backends
 
-`hops local` supports two backends behind the same commands:
+`hops local` supports three backends behind the same commands:
 
 - **colima** — a VM running dockerd + k3s. macOS/Linux; supports `--cpus`,
   `--memory`, `--disk`, and `hops local resize`.
 - **kind** — cluster nodes as docker containers on any reachable docker
-  daemon: Docker Desktop, colima's dockerd, [dory](https://augani.github.io/dory),
-  or CI runners. No VM of its own, so sizing flags don't apply (size the
-  docker daemon instead); requires kind >= v0.27.
+  daemon: Docker Desktop, colima's dockerd, or CI runners. No VM of its own,
+  so sizing flags don't apply (size the docker daemon instead); requires
+  kind >= v0.27.
+- **dory** — [dory](https://augani.github.io/dory)'s built-in k3s, driven
+  headlessly through the `dory` CLI (`dory k8s enable/disable/status`).
+  Requires the Dory app running (it provides the engine and forwards
+  published ports to localhost) and a `dory` CLI with headless k8s support.
+  hops writes `~/.dory/k8s/registries.yaml` (k3s' native registry trust) and
+  publishes the registry NodePort at cluster create. The VM is sized in the
+  Dory app, so hops sizing flags don't apply.
 
 Select with the global `--backend` flag:
 
@@ -236,22 +243,31 @@ persisted choice > existing cluster detection (colima wins) > platform
 default (macOS: colima, otherwise kind).
 
 Unless `--context` is given, kubectl commands automatically use the backend's
-kubeconfig context (`colima` or `kind-hops`), regardless of your
-current-context.
+kubeconfig context (`colima`, `kind-hops`, or `dory`), regardless of your
+current-context. For dory, hops also prepends `~/.kube/dory-config` (where
+dory keeps its kubeconfig) to `KUBECONFIG` for its own kubectl/helm calls.
 
 #### Using dory
 
-[dory](https://augani.github.io/dory) exposes a real docker socket, so run the
-kind backend against it:
+With a `dory` CLI that supports `dory k8s enable` (headless Kubernetes),
+use the native backend:
+
+```bash
+hops local start --backend dory
+```
+
+For `hops provider install` / `hops config install` builds, point your docker
+CLI at dory's engine (`export DOCKER_HOST=unix://$HOME/.dory/engine.sock`) so
+image builds/pushes land on the daemon that reaches the registry.
+
+Without the headless CLI, dory still exposes a real docker socket, so the
+kind backend works against it:
 
 ```bash
 docker context use dory   # or: export DOCKER_HOST=unix://$HOME/.dory/dory.sock
 hops local start --backend kind
 ```
 
-Don't use dory's built-in Kubernetes for hops — it publishes only the API
-port, so the local package registry would be unreachable from the host.
-
 ### Local provider setup and auth
 
 `hops local aws`, `hops local github`, and `hops local zitadel` install the provider package and bootstrap auth into a local control plane. The exception is `--refresh`, which updates credentials only.
diff --git a/src/commands/config/install.rs b/src/commands/config/install.rs
index d3812c9..dd396c2 100644
--- a/src/commands/config/install.rs
+++ b/src/commands/config/install.rs
@@ -1,4 +1,4 @@
-use crate::commands::local::backend::{self, wire_local_registry};
+use crate::commands::local::backend::{self, Backend};
 use crate::commands::local::package_install::run_watch;
 use crate::commands::local::package_install::{
     docker_arch, ensure_cached_repo_checkout, ensure_registry, image_config_name,
@@ -6,9 +6,7 @@ use crate::commands::local::package_install::{
     rewrite_registry_with_tag, sanitize_name_component, short_hash, split_ref, strip_registry,
     unique_suffix, RepoInstallTarget, RepoSpec, REGISTRY_PULL, REGISTRY_PUSH,
 };
-use crate::commands::local::{
-    kubectl_apply_stdin, kubectl_command, run_cmd, run_cmd_output, HOPS_KUBE_CONTEXT_ENV,
-};
+use crate::commands::local::{kubectl_apply_stdin, kubectl_command, run_cmd, run_cmd_output};
 use clap::Args;
 use flate2::read::GzDecoder;
 use serde::Deserialize;
@@ -43,6 +41,10 @@ pub struct ConfigArgs {
     #[arg(long)]
     pub context: Option<String>,
 
+    /// Local cluster backend whose node should be wired for local package pulls.
+    #[arg(long, value_enum)]
+    pub backend: Option<Backend>,
+
     /// Watch the project directory for changes and re-run install automatically
     #[arg(long, conflicts_with = "repo")]
     pub watch: bool,
@@ -108,17 +110,22 @@ struct PackageResource {
 }
 
 pub fn run(args: &ConfigArgs) -> Result<(), Box<dyn Error>> {
-    if let Some(ctx) = &args.context {
-        std::env::set_var(HOPS_KUBE_CONTEXT_ENV, ctx);
-    }
+    let backend = backend::activate(args.backend, args.context.as_deref());
 
     match (args.repo.as_deref(), args.version.as_deref()) {
         (Some(repo), Some(version)) => {
             apply_repo_version(repo, version, args.skip_dependency_resolution)
         }
-        (Some(repo), None) => run_repo_install(repo, args.skip_dependency_resolution),
+        (Some(repo), None) => run_repo_install(
+            repo,
+            args.skip_dependency_resolution,
+            backend,
+            args.backend,
+            args.context.as_deref(),
+        ),
         (None, _) => {
             let path = args.path.as_deref().unwrap_or(".");
+            prepare_local_registry(backend, args.backend, args.context.as_deref())?;
             run_local_path(path, args.skip_dependency_resolution)?;
 
             if args.watch {
@@ -134,21 +141,49 @@ pub fn run(args: &ConfigArgs) -> Result<(), Box<dyn Error>> {
     }
 }
 
-fn run_repo_install(repo: &str, skip_dependency_resolution: bool) -> Result<(), Box<dyn Error>> {
+fn run_repo_install(
+    repo: &str,
+    skip_dependency_resolution: bool,
+    backend: Backend,
+    backend_flag: Option<Backend>,
+    context: Option<&str>,
+) -> Result<(), Box<dyn Error>> {
     let spec = parse_repo_spec(repo)?;
     match resolve_repo_install_target(&spec)? {
-        RepoInstallTarget::SourceBuild => run_repo_clone(&spec, skip_dependency_resolution),
+        RepoInstallTarget::SourceBuild => run_repo_clone(
+            &spec,
+            skip_dependency_resolution,
+            backend,
+            backend_flag,
+            context,
+        ),
         RepoInstallTarget::PublishedVersion(version) => {
             apply_repo_version_spec(&spec, &version, skip_dependency_resolution)
         }
     }
 }
 
-fn run_repo_clone(spec: &RepoSpec, skip_dependency_resolution: bool) -> Result<(), Box<dyn Error>> {
-    let cache_path = ensure_cached_repo_checkout(&spec)?;
+fn run_repo_clone(
+    spec: &RepoSpec,
+    skip_dependency_resolution: bool,
+    backend: Backend,
+    backend_flag: Option<Backend>,
+    context: Option<&str>,
+) -> Result<(), Box<dyn Error>> {
+    let cache_path = ensure_cached_repo_checkout(spec)?;
+    prepare_local_registry(backend, backend_flag, context)?;
     run_local_path(&cache_path.to_string_lossy(), skip_dependency_resolution)
 }
 
+fn prepare_local_registry(
+    backend: Backend,
+    backend_flag: Option<Backend>,
+    context: Option<&str>,
+) -> Result<(), Box<dyn Error>> {
+    ensure_registry()?;
+    backend::wire_local_registry_for_target(backend, backend_flag, context)
+}
+
 fn apply_repo_version_spec(
     spec: &RepoSpec,
     version: &str,
@@ -216,9 +251,6 @@ fn run_local_path(path: &str, skip_dependency_resolution: bool) -> Result<(), Bo
         return Err(format!("{} is not a directory", path).into());
     }
 
-    ensure_registry()?;
-    wire_local_registry(backend::resolve(None))?;
-
     // Build the Crossplane package
     log::info!("Building Crossplane package in {}...", path);
     let status = Command::new("up")
@@ -237,7 +269,7 @@ fn run_local_path(path: &str, skip_dependency_resolution: bool) -> Result<(), Bo
     let packages: Vec<_> = fs::read_dir(&output_dir)
         .map_err(|e| format!("Failed to read {}: {}", output_dir.display(), e))?
         .filter_map(|entry| entry.ok())
-        .filter(|entry| entry.path().extension().map_or(false, |ext| ext == "uppkg"))
+        .filter(|entry| entry.path().extension().is_some_and(|ext| ext == "uppkg"))
         .collect();
 
     if packages.is_empty() {
@@ -1038,6 +1070,15 @@ spec:
         assert!(!without_skip.contains("skipDependencyResolution: true"));
     }
 
+    #[test]
+    fn local_registry_wiring_skips_foreign_context_without_backend_flag() {
+        assert!(!backend::should_wire_local_registry(
+            None,
+            Some("kind-hops"),
+            Backend::Colima
+        ));
+    }
+
     #[test]
     fn package_source_strips_tag_and_digest() {
         assert_eq!(
diff --git a/src/commands/local/backend/dory.rs b/src/commands/local/backend/dory.rs
new file mode 100644
index 0000000..1c0dfed
--- /dev/null
+++ b/src/commands/local/backend/dory.rs
@@ -0,0 +1,741 @@
+//! dory backend: k3s in a container on dory's shared-VM dockerd
+//! (https://augani.github.io/dory), driven headlessly through the `dory` CLI
+//! (`dory k8s enable|disable|status`) and the engine docker socket.
+//!
+//! Registry plumbing differs from colima/kind because the cluster container
+//! has create-time config shared with the Dory app:
+//! - published NodePorts are static config: hops writes `~/.dory/k8s/ports`
+//!   BEFORE `dory k8s enable`, so GUI-side recreates keep the same port.
+//! - trust is static config: hops writes `~/.dory/k8s/registries.yaml`
+//!   BEFORE `dory k8s enable`; dory bind-mounts it and k3s reads it at boot,
+//!   aliasing both pull names to the registry Service hostname over HTTP.
+//! - name resolution is dynamic: `wire_registry` syncs the hostname ->
+//!   ClusterIP in the node container's /etc/hosts on every start (same
+//!   re-wire-on-start model as the other backends).
+//! - the host reaches `localhost:30500` because `dory k8s enable --publish
+//!   30500:30500` publishes the NodePort and the Dory app's port forwarder
+//!   maps published ports to host loopback.
+
+use super::SizeArgs;
+use crate::commands::local::package_install::{REGISTRY_HOSTNAME, REGISTRY_PULL, REGISTRY_PUSH};
+use crate::commands::local::{command_exists, run_cmd, run_cmd_output};
+use std::error::Error;
+use std::path::PathBuf;
+
+const NODE_CONTAINER: &str = "dory-k8s";
+/// hops' registry NodePort, published on the cluster container at create.
+const REGISTRY_PORT_PUBLISH: &str = "30500:30500";
+const REGISTRIES_BEGIN: &str = "# BEGIN hops-managed (do not edit inside)";
+const REGISTRIES_END: &str = "# END hops-managed";
+
+fn home() -> Result<PathBuf, Box<dyn Error>> {
+    Ok(PathBuf::from(std::env::var("HOME").map_err(|_| {
+        "HOME is not set; unable to locate dory's state directory"
+    })?))
+}
+
+fn engine_socket() -> Result<PathBuf, Box<dyn Error>> {
+    Ok(home()?.join(".dory/engine.sock"))
+}
+
+/// dory's side-file kubeconfig (context name `dory`). Current dory also
+/// merges the context into ~/.kube/config at enable time; the side file is
+/// the pre-merge fallback and dory's own `--kubeconfig` input.
+pub fn kubeconfig_path() -> Option<String> {
+    home()
+        .ok()
+        .map(|h| h.join(".kube/dory-config").to_string_lossy().into_owned())
+}
+
+fn registries_yaml_path() -> Result<PathBuf, Box<dyn Error>> {
+    Ok(home()?.join(".dory/k8s/registries.yaml"))
+}
+
+fn ports_file_path() -> Result<PathBuf, Box<dyn Error>> {
+    Ok(home()?.join(".dory/k8s/ports"))
+}
+
+/// k3s' native registry config: alias both pull names to the registry
+/// Service hostname over plain HTTP. The hostname resolves through the
+/// node's /etc/hosts, which `wire_registry` keeps pointed at the ClusterIP.
+fn registries_yaml() -> String {
+    format!(
+        "# Written by `hops local start --backend dory`.\n\
+         # k3s reads this at boot; edits require `hops local reset`.\n\
+         # Hops manages only the marked block; keep user mirrors outside it.\n\
+         mirrors:\n\
+         {block}",
+        block = registries_yaml_block(),
+    )
+}
+
+fn registries_yaml_block() -> String {
+    format!(
+        "  {begin}\n\
+         \x20\x20# hops: mirror {pull}\n\
+         \x20\x20\"{pull}\":\n\
+         \x20\x20  endpoint:\n\
+         \x20\x20    - \"http://{pull}\"\n\
+         \x20\x20# hops: mirror {push}\n\
+         \x20\x20\"{push}\":\n\
+         \x20\x20  endpoint:\n\
+         \x20\x20    - \"http://{pull}\"\n\
+         \x20\x20{end}\n",
+        begin = REGISTRIES_BEGIN,
+        end = REGISTRIES_END,
+        pull = REGISTRY_PULL,
+        push = REGISTRY_PUSH,
+    )
+}
+
+fn legacy_registries_yaml() -> String {
+    format!(
+        "# Written by `hops local start --backend dory`.\n\
+         # k3s reads this at boot; edits require `hops local reset`.\n\
+         mirrors:\n\
+         \x20 \"{pull}\":\n\
+         \x20   endpoint:\n\
+         \x20     - \"http://{pull}\"\n\
+         \x20 \"{push}\":\n\
+         \x20   endpoint:\n\
+         \x20     - \"http://{pull}\"\n",
+        pull = REGISTRY_PULL,
+        push = REGISTRY_PUSH,
+    )
+}
+
+/// Run docker against dory's engine socket (the daemon the Dory app manages).
+fn engine_docker(args: &[&str]) -> Result<(), Box<dyn Error>> {
+    let sock = format!("unix://{}", engine_socket()?.display());
+    let mut full = vec!["-H", sock.as_str()];
+    full.extend_from_slice(args);
+    run_cmd("docker", &full)
+}
+
+fn engine_docker_output(args: &[&str]) -> Result<String, Box<dyn Error>> {
+    let sock = format!("unix://{}", engine_socket()?.display());
+    let mut full = vec!["-H", sock.as_str()];
+    full.extend_from_slice(args);
+    run_cmd_output("docker", &full)
+}
+
+pub fn install() -> Result<(), Box<dyn Error>> {
+    log::info!("Installing Dory via Homebrew...");
+    run_cmd("brew", &["install", "--cask", "Augani/dory/dory"])?;
+    log::info!("Dory installed; launch the Dory app once so it provisions its engine");
+    Ok(())
+}
+
+pub fn uninstall() -> Result<(), Box<dyn Error>> {
+    log::info!("Uninstalling Dory...");
+    run_cmd("brew", &["uninstall", "--cask", "dory"])?;
+    log::info!("Dory uninstalled");
+    Ok(())
+}
+
+pub fn start(size: &SizeArgs) -> Result<(), Box<dyn Error>> {
+    if size.any_set() {
+        return Err(format!(
+            "the dory backend's VM is sized by the Dory app, not hops; drop{}",
+            size.command_suffix()
+        )
+        .into());
+    }
+
+    preflight()?;
+    write_ports_file()?;
+    write_registries_yaml()?;
+
+    // Creates, restarts, or reuses the dory-k8s container as needed. If the
+    // running container has create-time config drift, dory exits 3 rather than
+    // destroying state; surface that plus the hops reset path.
+    run_dory_enable(false)
+}
+
+pub fn stop() -> Result<(), Box<dyn Error>> {
+    log::info!("Stopping dory k8s node '{}'...", NODE_CONTAINER);
+    engine_docker(&["stop", NODE_CONTAINER])?;
+    log::info!("dory cluster stopped");
+    Ok(())
+}
+
+pub fn destroy() -> Result<(), Box<dyn Error>> {
+    log::info!("Deleting dory k8s cluster...");
+    run_cmd("dory", &["k8s", "disable"])?;
+    log::info!("dory cluster deleted");
+    Ok(())
+}
+
+/// The cluster container IS the cluster, so reset means recreate.
+pub fn reset() -> Result<(), Box<dyn Error>> {
+    preflight()?;
+    run_cmd("dory", &["k8s", "disable"])?;
+    write_ports_file()?;
+    write_registries_yaml()?;
+    run_dory_enable(true)
+}
+
+pub fn resize(_size: &SizeArgs) -> Result<(), Box<dyn Error>> {
+    Err("the dory backend has no hops-managed VM to resize; \
+         adjust resources in the Dory app instead"
+        .into())
+}
+
+/// Whether the hops-relevant dory cluster exists (running or stopped).
+/// Missing app/engine/CLI reads as "no cluster".
+pub fn cluster_exists() -> bool {
+    let Ok(sock) = engine_socket() else {
+        return false;
+    };
+    if !sock.exists() {
+        return false;
+    }
+    engine_docker_output(&["inspect", "-f", "{{.State.Running}}", NODE_CONTAINER]).is_ok()
+}
+
+fn preflight() -> Result<(), Box<dyn Error>> {
+    if !command_exists("dory") {
+        return Err("the `dory` CLI is not on PATH; link it from the dory repo \
+             (`ln -sf <dory>/scripts/dory /opt/homebrew/bin/dory`)"
+            .into());
+    }
+    let sock = engine_socket()?;
+    if !sock.exists() {
+        return Err(format!(
+            "dory's engine socket ({}) is missing; launch the Dory app and wait for its engine to start",
+            sock.display()
+        )
+        .into());
+    }
+    if !command_exists("docker") {
+        return Err("docker CLI not found; install it (dory provides the daemon)".into());
+    }
+    Ok(())
+}
+
+/// Write the static registry trust config read by k3s at boot. Must exist
+/// before `dory k8s enable` because dory binds it at container create.
+fn write_registries_yaml() -> Result<(), Box<dyn Error>> {
+    let path = registries_yaml_path()?;
+    let current = match std::fs::read_to_string(&path) {
+        Ok(content) => Some(content),
+        Err(err) if err.kind() == std::io::ErrorKind::NotFound => None,
+        Err(err) => return Err(err.into()),
+    };
+    let desired = merge_registries_yaml(current.as_deref())?;
+    if current.as_deref() == Some(desired.as_str()) {
+        return Ok(());
+    }
+    if let Some(parent) = path.parent() {
+        std::fs::create_dir_all(parent)?;
+    }
+    log::info!("Writing k3s registry config: {}", path.display());
+    std::fs::write(&path, desired)?;
+    Ok(())
+}
+
+fn write_ports_file() -> Result<(), Box<dyn Error>> {
+    let path = ports_file_path()?;
+    let current = match std::fs::read_to_string(&path) {
+        Ok(content) => content,
+        Err(err) if err.kind() == std::io::ErrorKind::NotFound => String::new(),
+        Err(err) => return Err(err.into()),
+    };
+    let desired = ports_file_with_publish(&current);
+    if current == desired {
+        return Ok(());
+    }
+    if let Some(parent) = path.parent() {
+        std::fs::create_dir_all(parent)?;
+    }
+    log::info!(
+        "Ensuring dory k8s port config includes {}: {}",
+        REGISTRY_PORT_PUBLISH,
+        path.display()
+    );
+    std::fs::write(&path, desired)?;
+    Ok(())
+}
+
+/// Seconds after the Dory app (re)creates its engine socket during which it is
+/// still provisioning the engine. A dockerd restart at the end of that window
+/// SIGTERMs every container — including a k3s node enabled meanwhile, which
+/// reports Ready and then dies under the bootstrap (observed ~90s on Dory
+/// 0.2.0; padded for slower machines).
+const ENGINE_LAUNCH_WINDOW_SECS: u64 = 180;
+
+/// Age of the current engine session: the app recreates the engine socket at
+/// launch, so its mtime marks when provisioning began. None when unreadable.
+fn engine_session_age() -> Option<std::time::Duration> {
+    let sock = engine_socket().ok()?;
+    let modified = std::fs::metadata(&sock).ok()?.modified().ok()?;
+    std::time::SystemTime::now().duration_since(modified).ok()
+}
+
+/// Time left inside the app's provisioning window for a given engine session
+/// age; None once the window has passed.
+fn launch_window_remaining(age: std::time::Duration) -> Option<std::time::Duration> {
+    std::time::Duration::from_secs(ENGINE_LAUNCH_WINDOW_SECS)
+        .checked_sub(age)
+        .filter(|remaining| !remaining.is_zero())
+}
+
+fn node_running() -> bool {
+    engine_docker_output(&["inspect", "-f", "{{.State.Running}}", NODE_CONTAINER])
+        .map(|state| state.trim() == "true")
+        .unwrap_or(false)
+}
+
+/// Hold a freshly-enabled cluster under observation while the Dory app may
+/// still be provisioning its engine, re-enabling if the engine restart takes
+/// the node down. Immediate no-op when the window has already passed, so
+/// steady-state starts pay one container inspect and nothing more.
+fn hold_through_engine_launch_window() -> Result<(), Box<dyn Error>> {
+    let in_window = |age: Option<std::time::Duration>| {
+        age.map(|a| launch_window_remaining(a).is_some())
+            .unwrap_or(false)
+    };
+    if in_window(engine_session_age()) {
+        log::info!(
+            "Dory engine session is younger than {}s; watching the k8s node through the app's provisioning window...",
+            ENGINE_LAUNCH_WINDOW_SECS
+        );
+    }
+    let mut reenables = 0;
+    loop {
+        match (node_running(), in_window(engine_session_age())) {
+            (true, false) => return Ok(()),
+            (true, true) => {}
+            (false, _) => {
+                if reenables >= 3 {
+                    return Err("the dory engine keeps stopping the k8s node during app startup; \
+                         wait for the Dory app to finish provisioning, then re-run `hops local start --backend dory`"
+                        .into());
+                }
+                reenables += 1;
+                log::warn!(
+                    "dory engine restart stopped the k8s node; re-enabling ({}/3)...",
+                    reenables
+                );
+                let args = dory_enable_args(false);
+                run_cmd("dory", &args)?;
+            }
+        }
+        std::thread::sleep(std::time::Duration::from_secs(3));
+    }
+}
+
+fn run_dory_enable(recreate: bool) -> Result<(), Box<dyn Error>> {
+    let args = dory_enable_args(recreate);
+    match run_cmd("dory", &args) {
+        Ok(()) => hold_through_engine_launch_window(),
+        Err(err) if !recreate && err.to_string().contains("exit status: 3") => Err(format!(
+            "{}\nhint: run `hops local reset --backend dory` to recreate the dory cluster and apply create-time config drift",
+            err
+        )
+        .into()),
+        Err(err) => Err(err),
+    }
+}
+
+fn dory_enable_args(recreate: bool) -> Vec<&'static str> {
+    let mut args = vec!["k8s", "enable"];
+    if recreate {
+        args.push("--recreate");
+    }
+    args.extend(["--publish", REGISTRY_PORT_PUBLISH]);
+    args
+}
+
+fn merge_registries_yaml(existing: Option<&str>) -> Result<String, Box<dyn Error>> {
+    let Some(existing) = existing else {
+        return Ok(registries_yaml());
+    };
+    if existing.trim().is_empty() || existing == legacy_registries_yaml() {
+        return Ok(registries_yaml());
+    }
+
+    match (
+        existing.find(REGISTRIES_BEGIN),
+        existing.find(REGISTRIES_END),
+    ) {
+        (Some(begin), Some(end)) if begin <= end => replace_registries_block(existing, begin, end),
+        (Some(_), Some(_)) | (Some(_), None) | (None, Some(_)) => {
+            Err(format!("malformed dory registries.yaml managed block; expected `{REGISTRIES_BEGIN}` before `{REGISTRIES_END}`").into())
+        }
+        (None, None) => insert_registries_block(existing),
+    }
+}
+
+fn replace_registries_block(
+    existing: &str,
+    begin: usize,
+    end: usize,
+) -> Result<String, Box<dyn Error>> {
+    let line_start = existing[..begin].rfind('\n').map_or(0, |idx| idx + 1);
+    let line_end = existing[end..]
+        .find('\n')
+        .map_or(existing.len(), |idx| end + idx + 1);
+    let mut merged = String::with_capacity(existing.len() + registries_yaml_block().len());
+    merged.push_str(&existing[..line_start]);
+    merged.push_str(&registries_yaml_block());
+    merged.push_str(&existing[line_end..]);
+    Ok(merged)
+}
+
+fn insert_registries_block(existing: &str) -> Result<String, Box<dyn Error>> {
+    if contains_hops_mirror_key(existing) {
+        return Err("dory registries.yaml already contains hops registry mirror entries without managed markers; remove those entries or wrap them in the hops-managed block"
+            .into());
+    }
+
+    let mut merged = ensure_trailing_newline(existing);
+    if let Some(insert_at) = mirrors_line_insert_position(&merged) {
+        merged.insert_str(insert_at, &registries_yaml_block());
+        return Ok(merged);
+    }
+
+    merged.push_str("mirrors:\n");
+    merged.push_str(&registries_yaml_block());
+    Ok(merged)
+}
+
+fn contains_hops_mirror_key(content: &str) -> bool {
+    content.contains(&format!("\"{}\":", REGISTRY_PULL))
+        || content.contains(&format!("\"{}\":", REGISTRY_PUSH))
+}
+
+fn mirrors_line_insert_position(content: &str) -> Option<usize> {
+    let mut offset = 0;
+    for line in content.split_inclusive('\n') {
+        let without_newline = line.trim_end_matches('\n').trim_end_matches('\r');
+        if is_top_level_mirrors_line(without_newline) {
+            return Some(offset + line.len());
+        }
+        offset += line.len();
+    }
+    None
+}
+
+fn is_top_level_mirrors_line(line: &str) -> bool {
+    line.starts_with("mirrors:") && line.split('#').next().unwrap_or("").trim_end() == "mirrors:"
+}
+
+fn ensure_trailing_newline(content: &str) -> String {
+    let mut out = content.to_string();
+    if !out.is_empty() && !out.ends_with('\n') {
+        out.push('\n');
+    }
+    out
+}
+
+#[derive(Debug, PartialEq, Eq)]
+struct PortPublish {
+    host: u16,
+    container: u16,
+    proto: String,
+}
+
+fn ports_file_with_publish(existing: &str) -> String {
+    if ports_file_contains_publish(existing, REGISTRY_PORT_PUBLISH) {
+        return existing.to_string();
+    }
+
+    let mut out = ensure_trailing_newline(existing);
+    out.push_str(REGISTRY_PORT_PUBLISH);
+    out.push('\n');
+    out
+}
+
+fn ports_file_contains_publish(existing: &str, desired: &str) -> bool {
+    let Some(desired) = parse_port_publish(desired) else {
+        return false;
+    };
+    existing
+        .lines()
+        .filter_map(parse_port_publish)
+        .any(|port| port == desired)
+}
+
+fn parse_port_publish(line: &str) -> Option<PortPublish> {
+    let cleaned: String = line
+        .split('#')
+        .next()
+        .unwrap_or("")
+        .chars()
+        .filter(|c| !c.is_whitespace())
+        .collect();
+    if cleaned.is_empty() {
+        return None;
+    }
+    let (ports, proto) = cleaned
+        .split_once('/')
+        .map_or((cleaned.as_str(), "tcp"), |(ports, proto)| (ports, proto));
+    if proto != "tcp" && proto != "udp" {
+        return None;
+    }
+    let (host, container) = ports.split_once(':')?;
+    let host = host.parse().ok().filter(|port| *port > 0)?;
+    let container = container.parse().ok().filter(|port| *port > 0)?;
+    Some(PortPublish {
+        host,
+        container,
+        proto: proto.to_string(),
+    })
+}
+
+/// Keep the node container's /etc/hosts pointing the registry hostname at
+/// the current ClusterIP (docker regenerates /etc/hosts on restart, and the
+/// ClusterIP changes if the Service is recreated — hence re-run per start).
+pub fn wire_registry(cluster_ip: &str) -> Result<(), Box<dyn Error>> {
+    let hostname = REGISTRY_HOSTNAME;
+    let current_ip = engine_docker_output(&[
+        "exec",
+        NODE_CONTAINER,
+        "sh",
+        "-c",
+        &format!("awk '$2 == \"{}\" {{print $1; exit}}' /etc/hosts", hostname),
+    ])
+    .unwrap_or_default();
+    if current_ip.trim() == cluster_ip {
+        return Ok(());
+    }
+
+    log::info!("Updating node hosts entry: {} -> {}", hostname, cluster_ip);
+
+    // /etc/hosts is a bind mount inside the container: `sed -i` fails with
+    // "Resource busy" (rename over a mount point), so rewrite it in place.
+    engine_docker(&[
+        "exec",
+        NODE_CONTAINER,
+        "sh",
+        "-c",
+        &format!(
+            "awk '$2 != \"{host}\"' /etc/hosts > /tmp/hosts.new && \
+             cat /tmp/hosts.new > /etc/hosts && rm -f /tmp/hosts.new && \
+             echo '{ip} {host}' >> /etc/hosts",
+            host = hostname,
+            ip = cluster_ip
+        ),
+    ])?;
+
+    Ok(())
+}
+
+/// Make `--context dory` resolvable. Current dory merges the context into
+/// ~/.kube/config at enable time, so normally there is nothing to do —
+/// mutating KUBECONFIG for every child process is then pure noise. Older
+/// dory versions only write the side file; for those, prepend it to
+/// KUBECONFIG (preserving whatever the user already has).
+pub fn export_kubeconfig_env() {
+    if effective_kubeconfig_has_dory_context() {
+        return;
+    }
+    let Some(dory_cfg) = kubeconfig_path() else {
+        return;
+    };
+    let existing = std::env::var("KUBECONFIG").unwrap_or_default();
+    if existing.split(':').any(|p| p == dory_cfg) {
+        return;
+    }
+    let rest = if existing.is_empty() {
+        match home() {
+            Ok(h) => h.join(".kube/config").to_string_lossy().into_owned(),
+            Err(_) => return,
+        }
+    } else {
+        existing
+    };
+    std::env::set_var("KUBECONFIG", format!("{}:{}", dory_cfg, rest));
+}
+
+/// Whether the kubeconfig(s) kubectl will read without our help — the
+/// $KUBECONFIG chain when set, else ~/.kube/config — already define a
+/// `dory` entry (i.e. dory's kubectl-merge ran against a file in scope).
+fn effective_kubeconfig_has_dory_context() -> bool {
+    let paths: Vec<PathBuf> = match std::env::var("KUBECONFIG") {
+        Ok(chain) if !chain.is_empty() => chain.split(':').map(PathBuf::from).collect(),
+        _ => match home() {
+            Ok(h) => vec![h.join(".kube/config")],
+            Err(_) => return false,
+        },
+    };
+    paths.iter().any(|path| {
+        std::fs::read_to_string(path)
+            .map(|content| has_dory_entry(&content))
+            .unwrap_or(false)
+    })
+}
+
+/// Line-anchored scan for a kubeconfig entry named `dory` — the mapping
+/// form (`name: dory`, as kubectl writes context/cluster names) or the
+/// sequence-item form (`- name: dory`, the users list). `name: dory-prod`
+/// or `username: dory` must not count.
+fn has_dory_entry(kubeconfig: &str) -> bool {
+    kubeconfig.lines().any(|line| {
+        let trimmed = line.trim();
+        trimmed == "name: dory" || trimmed == "- name: dory"
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn launch_window_remaining_covers_only_the_provisioning_window() {
+        use std::time::Duration;
+
+        assert_eq!(
+            launch_window_remaining(Duration::ZERO),
+            Some(Duration::from_secs(ENGINE_LAUNCH_WINDOW_SECS))
+        );
+        assert_eq!(
+            launch_window_remaining(Duration::from_secs(ENGINE_LAUNCH_WINDOW_SECS - 1)),
+            Some(Duration::from_secs(1))
+        );
+        assert_eq!(
+            launch_window_remaining(Duration::from_secs(ENGINE_LAUNCH_WINDOW_SECS)),
+            None
+        );
+        assert_eq!(launch_window_remaining(Duration::from_secs(3600)), None);
+    }
+
+    #[test]
+    fn has_dory_entry_matches_mapping_and_sequence_forms_only() {
+        let merged = "contexts:\n- context:\n    cluster: dory\n    user: dory\n  name: dory\n";
+        let users_list = "users:\n- name: dory\n  user: {}\n";
+        let near_misses = "name: dory-prod\nusername: dory\n# name: dory\nfullname: dory\n";
+
+        assert!(has_dory_entry(merged));
+        assert!(has_dory_entry(users_list));
+        assert!(!has_dory_entry(near_misses));
+        assert!(!has_dory_entry(""));
+    }
+
+    #[test]
+    fn registries_yaml_aliases_both_pull_names_to_the_service_over_http() {
+        let yaml = registries_yaml();
+
+        assert!(yaml.contains(REGISTRIES_BEGIN));
+        assert!(yaml.contains(REGISTRIES_END));
+        assert!(yaml.contains("\"registry.crossplane-system.svc.cluster.local:5000\":"));
+        assert!(yaml.contains("\"localhost:30500\":"));
+        // Both mirrors resolve to the same HTTP endpoint on the Service name.
+        assert_eq!(
+            yaml.matches("- \"http://registry.crossplane-system.svc.cluster.local:5000\"")
+                .count(),
+            2
+        );
+        assert!(!yaml.contains("https://"));
+    }
+
+    #[test]
+    fn registries_yaml_replaces_only_hops_managed_block() {
+        let existing = format!(
+            "configs:\n  example: value\nmirrors:\n  {begin}\n  old: value\n  {end}\n  \"user.local:5000\":\n    endpoint:\n      - \"http://user.local:5000\"\n",
+            begin = REGISTRIES_BEGIN,
+            end = REGISTRIES_END,
+        );
+
+        let merged = merge_registries_yaml(Some(&existing)).unwrap();
+
+        assert!(merged.starts_with("configs:\n  example: value\nmirrors:\n"));
+        assert!(merged.contains("  \"user.local:5000\":\n"));
+        assert!(!merged.contains("old: value"));
+        assert!(merged.contains(&format!("  \"{}\":", REGISTRY_PULL)));
+        assert_eq!(merged.matches(REGISTRIES_BEGIN).count(), 1);
+        assert_eq!(merged.matches(REGISTRIES_END).count(), 1);
+    }
+
+    #[test]
+    fn registries_yaml_inserts_block_under_existing_mirrors_key() {
+        let existing = "mirrors:\n  \"user.local:5000\":\n    endpoint:\n      - \"http://user.local:5000\"\nconfigs:\n  another: value\n";
+
+        let merged = merge_registries_yaml(Some(existing)).unwrap();
+
+        let block_pos = merged.find(REGISTRIES_BEGIN).unwrap();
+        let user_pos = merged.find("\"user.local:5000\"").unwrap();
+        assert!(block_pos < user_pos);
+        assert!(merged.contains("configs:\n  another: value\n"));
+    }
+
+    #[test]
+    fn registries_yaml_appends_mirrors_section_when_missing() {
+        let existing = "configs:\n  example: value\n";
+
+        let merged = merge_registries_yaml(Some(existing)).unwrap();
+
+        assert!(merged.starts_with(existing));
+        assert!(merged.contains("\nmirrors:\n"));
+        assert!(merged.contains(REGISTRIES_BEGIN));
+    }
+
+    #[test]
+    fn registries_yaml_upgrades_legacy_hops_file_to_managed_block() {
+        let merged = merge_registries_yaml(Some(&legacy_registries_yaml())).unwrap();
+
+        assert_eq!(merged, registries_yaml());
+        assert_eq!(
+            merged.matches(&format!("\"{}\":", REGISTRY_PULL)).count(),
+            1
+        );
+        assert_eq!(
+            merged.matches(&format!("\"{}\":", REGISTRY_PUSH)).count(),
+            1
+        );
+    }
+
+    #[test]
+    fn ports_file_appends_registry_port_without_touching_existing_lines() {
+        let existing = "# user port\n8080:80/udp\n";
+
+        let merged = ports_file_with_publish(existing);
+
+        assert_eq!(merged, "# user port\n8080:80/udp\n30500:30500\n");
+    }
+
+    #[test]
+    fn ports_file_absent_writes_only_registry_port() {
+        assert_eq!(ports_file_with_publish(""), "30500:30500\n");
+    }
+
+    #[test]
+    fn ports_file_treats_tcp_variant_as_already_present() {
+        let existing = " 30500 : 30500 / tcp  # registry\n";
+
+        assert_eq!(ports_file_with_publish(existing), existing);
+    }
+
+    #[test]
+    fn dory_enable_args_only_recreate_for_reset_path() {
+        assert_eq!(
+            dory_enable_args(false),
+            vec!["k8s", "enable", "--publish", REGISTRY_PORT_PUBLISH]
+        );
+        assert_eq!(
+            dory_enable_args(true),
+            vec![
+                "k8s",
+                "enable",
+                "--recreate",
+                "--publish",
+                REGISTRY_PORT_PUBLISH
+            ]
+        );
+    }
+
+    #[test]
+    fn start_rejects_size_flags() {
+        let size = SizeArgs {
+            cpus: Some(4),
+            memory: None,
+            disk: None,
+        };
+
+        let err = start(&size).expect_err("size flags must be rejected");
+
+        assert!(err.to_string().contains("--cpus 4"));
+        assert!(err.to_string().contains("Dory app"));
+    }
+}
diff --git a/src/commands/local/backend/mod.rs b/src/commands/local/backend/mod.rs
index 26541ec..2e2427f 100644
--- a/src/commands/local/backend/mod.rs
+++ b/src/commands/local/backend/mod.rs
@@ -5,12 +5,14 @@
 //! shaped lives outside this module and is backend-agnostic.
 
 mod colima;
+mod dory;
 mod kind;
 
-use super::{local_state_dir, run_cmd_output};
+use super::{local_state_dir, run_cmd_output, HOPS_KUBE_CONTEXT_ENV};
 use clap::Args;
 use std::error::Error;
 use std::fmt;
+use std::process::Command;
 use std::str::FromStr;
 
 /// Sizing flags for backends with a resizable VM.
@@ -61,6 +63,8 @@ pub enum Backend {
     /// docker containers as nodes; works on any docker daemon
     /// (Docker Desktop, colima, dory, CI runners)
     Kind,
+    /// k3s on dory's shared-VM engine, via the `dory` CLI
+    Dory,
 }
 
 impl Backend {
@@ -69,6 +73,7 @@ impl Backend {
         match self {
             Backend::Colima => "colima",
             Backend::Kind => "kind",
+            Backend::Dory => "dory",
         }
     }
 
@@ -77,6 +82,7 @@ impl Backend {
         match self {
             Backend::Colima => "colima",
             Backend::Kind => "kind-hops",
+            Backend::Dory => "dory",
         }
     }
 
@@ -84,6 +90,7 @@ impl Backend {
         match self {
             Backend::Colima => colima::install(),
             Backend::Kind => kind::install(),
+            Backend::Dory => dory::install(),
         }
     }
 
@@ -91,6 +98,16 @@ impl Backend {
         match self {
             Backend::Colima => colima::uninstall(),
             Backend::Kind => kind::uninstall(),
+            Backend::Dory => dory::uninstall(),
+        }
+    }
+
+    /// Whether this backend's local cluster/VM exists, running or stopped.
+    pub fn cluster_exists(self) -> bool {
+        match self {
+            Backend::Colima => colima::instance_exists(),
+            Backend::Kind => kind::cluster_exists(),
+            Backend::Dory => dory::cluster_exists(),
         }
     }
 
@@ -100,6 +117,7 @@ impl Backend {
         match self {
             Backend::Colima => colima::start(size, assume_yes),
             Backend::Kind => kind::start(size),
+            Backend::Dory => dory::start(size),
         }
     }
 
@@ -107,6 +125,7 @@ impl Backend {
         match self {
             Backend::Colima => colima::stop(),
             Backend::Kind => kind::stop(),
+            Backend::Dory => dory::stop(),
         }
     }
 
@@ -114,6 +133,7 @@ impl Backend {
         match self {
             Backend::Colima => colima::destroy(),
             Backend::Kind => kind::destroy(),
+            Backend::Dory => dory::destroy(),
         }
     }
 
@@ -121,6 +141,7 @@ impl Backend {
         match self {
             Backend::Colima => colima::reset(),
             Backend::Kind => kind::reset(),
+            Backend::Dory => dory::reset(),
         }
     }
 
@@ -128,6 +149,7 @@ impl Backend {
         match self {
             Backend::Colima => colima::resize(size),
             Backend::Kind => kind::resize(size),
+            Backend::Dory => dory::resize(size),
         }
     }
 
@@ -139,6 +161,9 @@ impl Backend {
             // containerd trust is per-name via certs.d, written in
             // wire_registry once the registry Service's ClusterIP is known.
             Backend::Kind => Ok(()),
+            // trust is static registries.yaml, written by dory::start before
+            // the cluster boots (k3s reads it at boot).
+            Backend::Dory => Ok(()),
         }
     }
 
@@ -149,6 +174,7 @@ impl Backend {
         match self {
             Backend::Colima => colima::sync_hosts_entry(cluster_ip),
             Backend::Kind => kind::wire_registry(cluster_ip),
+            Backend::Dory => dory::wire_registry(cluster_ip),
         }
     }
 }
@@ -166,8 +192,9 @@ impl FromStr for Backend {
         match s.trim() {
             "colima" => Ok(Backend::Colima),
             "kind" => Ok(Backend::Kind),
+            "dory" => Ok(Backend::Dory),
             other => Err(format!(
-                "unknown backend '{}' (expected colima or kind)",
+                "unknown backend '{}' (expected colima, kind, or dory)",
                 other
             )),
         }
@@ -176,6 +203,14 @@ impl FromStr for Backend {
 
 const BACKEND_FILE: &str = "backend";
 
+pub fn platform_default() -> Backend {
+    if cfg!(target_os = "macos") {
+        Backend::Colima
+    } else {
+        Backend::Kind
+    }
+}
+
 /// Resolve which backend to operate on: explicit flag > preference persisted
 /// by the last successful start > detection of an existing cluster (colima
 /// wins for back-compat with pre-backend installs) > platform default.
@@ -185,15 +220,48 @@ pub fn resolve(flag: Option<Backend>) -> Backend {
         persisted(),
         colima::instance_exists,
         kind::cluster_exists,
-        cfg!(target_os = "macos"),
+        dory::cluster_exists,
+        platform_default() == Backend::Colima,
     )
 }
 
+/// Resolve the backend once and activate the kube-targeting environment for
+/// child kubectl/helm processes.
+pub fn activate(flag: Option<Backend>, context: Option<&str>) -> Backend {
+    let backend = resolve(flag);
+
+    if backend == Backend::Dory {
+        dory::export_kubeconfig_env();
+    }
+
+    let explicit_context = context.filter(|ctx| !ctx.is_empty());
+    let backend_context_exists = if explicit_context.is_none() {
+        kube_context_exists(backend.kube_context())
+    } else {
+        false
+    };
+
+    match kube_context_export(backend, explicit_context, backend_context_exists) {
+        KubeContextExport::Set(ctx) => std::env::set_var(HOPS_KUBE_CONTEXT_ENV, ctx),
+        KubeContextExport::Unset { missing_context } => {
+            std::env::remove_var(HOPS_KUBE_CONTEXT_ENV);
+            log::warn!(
+                "Kubernetes context '{}' for backend '{}' was not found; using kubeconfig current-context. Pass --context to target a specific cluster.",
+                missing_context,
+                backend.name()
+            );
+        }
+    }
+
+    backend
+}
+
 fn resolve_from(
     flag: Option<Backend>,
     persisted: Option<Backend>,
     colima_detected: impl FnOnce() -> bool,
     kind_detected: impl FnOnce() -> bool,
+    dory_detected: impl FnOnce() -> bool,
     macos: bool,
 ) -> Backend {
     if let Some(backend) = flag {
@@ -208,6 +276,9 @@ fn resolve_from(
     if kind_detected() {
         return Backend::Kind;
     }
+    if dory_detected() {
+        return Backend::Dory;
+    }
     if macos {
         Backend::Colima
     } else {
@@ -265,6 +336,78 @@ pub fn wire_local_registry(backend: Backend) -> Result<(), Box<dyn Error>> {
     backend.wire_registry(&registry_cluster_ip()?)
 }
 
+pub fn should_wire_local_registry(
+    backend_flag: Option<Backend>,
+    context: Option<&str>,
+    backend: Backend,
+) -> bool {
+    if backend_flag.is_some() {
+        return true;
+    }
+
+    match context.filter(|ctx| !ctx.is_empty()) {
+        Some(ctx) => ctx == backend.kube_context(),
+        None => true,
+    }
+}
+
+pub fn wire_local_registry_for_target(
+    backend: Backend,
+    backend_flag: Option<Backend>,
+    context: Option<&str>,
+) -> Result<(), Box<dyn Error>> {
+    if should_wire_local_registry(backend_flag, context, backend) {
+        return wire_local_registry(backend);
+    }
+
+    log::warn!(
+        "registry node wiring skipped: explicit --context does not match a selected backend"
+    );
+    Ok(())
+}
+
+#[derive(Debug, PartialEq, Eq)]
+enum KubeContextExport<'a> {
+    Set(&'a str),
+    Unset { missing_context: &'static str },
+}
+
+fn kube_context_export<'a>(
+    backend: Backend,
+    explicit_context: Option<&'a str>,
+    backend_context_exists: bool,
+) -> KubeContextExport<'a> {
+    if let Some(ctx) = explicit_context {
+        return KubeContextExport::Set(ctx);
+    }
+
+    let backend_context = backend.kube_context();
+    if backend_context_exists {
+        KubeContextExport::Set(backend_context)
+    } else {
+        KubeContextExport::Unset {
+            missing_context: backend_context,
+        }
+    }
+}
+
+fn kube_context_exists(context: &str) -> bool {
+    let output = Command::new("kubectl")
+        .args(["config", "get-contexts", "-o", "name"])
+        .output();
+
+    let Ok(output) = output else {
+        return false;
+    };
+    if !output.status.success() {
+        return false;
+    }
+
+    String::from_utf8_lossy(&output.stdout)
+        .lines()
+        .any(|line| line.trim() == context)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -280,6 +423,7 @@ mod tests {
             Some(Backend::Colima),
             || true,
             no_detect,
+            no_detect,
             true,
         );
 
@@ -288,21 +432,35 @@ mod tests {
 
     #[test]
     fn persisted_beats_detection() {
-        let resolved = resolve_from(None, Some(Backend::Kind), || true, no_detect, true);
+        let resolved = resolve_from(
+            None,
+            Some(Backend::Kind),
+            || true,
+            no_detect,
+            no_detect,
+            true,
+        );
 
         assert_eq!(resolved, Backend::Kind);
     }
 
     #[test]
     fn colima_detection_beats_kind_detection() {
-        let resolved = resolve_from(None, None, || true, || true, false);
+        let resolved = resolve_from(None, None, || true, || true, || true, false);
 
         assert_eq!(resolved, Backend::Colima);
     }
 
+    #[test]
+    fn dory_detection_used_when_no_colima_or_kind() {
+        let resolved = resolve_from(None, None, no_detect, no_detect, || true, true);
+
+        assert_eq!(resolved, Backend::Dory);
+    }
+
     #[test]
     fn kind_detection_used_when_no_colima() {
-        let resolved = resolve_from(None, None, no_detect, || true, true);
+        let resolved = resolve_from(None, None, no_detect, || true, no_detect, true);
 
         assert_eq!(resolved, Backend::Kind);
     }
@@ -310,20 +468,74 @@ mod tests {
     #[test]
     fn platform_default_when_nothing_detected() {
         assert_eq!(
-            resolve_from(None, None, no_detect, no_detect, true),
+            resolve_from(None, None, no_detect, no_detect, no_detect, true),
             Backend::Colima
         );
         assert_eq!(
-            resolve_from(None, None, no_detect, no_detect, false),
+            resolve_from(None, None, no_detect, no_detect, no_detect, false),
             Backend::Kind
         );
     }
 
     #[test]
     fn backend_name_round_trips_through_from_str() {
-        for backend in [Backend::Colima, Backend::Kind] {
+        for backend in [Backend::Colima, Backend::Kind, Backend::Dory] {
             assert_eq!(backend.name().parse::<Backend>().unwrap(), backend);
         }
-        assert!("dory".parse::<Backend>().is_err());
+        assert!("podman".parse::<Backend>().is_err());
+    }
+
+    #[test]
+    fn explicit_context_is_exported_even_when_backend_context_is_absent() {
+        assert_eq!(
+            kube_context_export(Backend::Colima, Some("foreign"), false),
+            KubeContextExport::Set("foreign")
+        );
+    }
+
+    #[test]
+    fn backend_context_is_exported_only_when_present() {
+        assert_eq!(
+            kube_context_export(Backend::Kind, None, true),
+            KubeContextExport::Set("kind-hops")
+        );
+        assert_eq!(
+            kube_context_export(Backend::Kind, None, false),
+            KubeContextExport::Unset {
+                missing_context: "kind-hops"
+            }
+        );
+    }
+
+    #[test]
+    fn registry_wiring_allowed_without_explicit_context() {
+        assert!(should_wire_local_registry(None, None, Backend::Colima));
+    }
+
+    #[test]
+    fn registry_wiring_skips_foreign_explicit_context_without_backend_flag() {
+        assert!(!should_wire_local_registry(
+            None,
+            Some("kind-hops"),
+            Backend::Colima
+        ));
+    }
+
+    #[test]
+    fn registry_wiring_allowed_when_context_matches_backend() {
+        assert!(should_wire_local_registry(
+            None,
+            Some("kind-hops"),
+            Backend::Kind
+        ));
+    }
+
+    #[test]
+    fn registry_wiring_allowed_when_backend_is_explicit() {
+        assert!(should_wire_local_registry(
+            Some(Backend::Colima),
+            Some("foreign"),
+            Backend::Colima
+        ));
     }
 }
diff --git a/src/commands/local/mod.rs b/src/commands/local/mod.rs
index e59fb2a..0db9db0 100644
--- a/src/commands/local/mod.rs
+++ b/src/commands/local/mod.rs
@@ -34,22 +34,53 @@ pub const PROVIDER_INSTALL_MANAGED_BY: &str = "hops-provider-install";
 /// Env var checked by kubectl helpers to inject `--context <name>`.
 pub const HOPS_KUBE_CONTEXT_ENV: &str = "HOPS_KUBE_CONTEXT";
 
-/// Build the kubectl args prefix. Returns `["--context", ctx]` when the env var
-/// is set, or an empty vec otherwise.
-fn kubectl_context_args() -> Vec<String> {
-    match std::env::var(HOPS_KUBE_CONTEXT_ENV) {
-        Ok(ctx) if !ctx.is_empty() => vec!["--context".to_string(), ctx],
-        _ => vec![],
-    }
+fn kube_context_from_env() -> Option<String> {
+    std::env::var(HOPS_KUBE_CONTEXT_ENV)
+        .ok()
+        .filter(|ctx| !ctx.is_empty())
 }
 
 /// Prepend `--context` to a kubectl arg slice when configured.
 fn with_kube_context(args: &[&str]) -> Vec<String> {
-    let mut out = kubectl_context_args();
+    let ctx = kube_context_from_env();
+    with_kube_context_value(args, ctx.as_deref())
+}
+
+fn with_kube_context_value(args: &[&str], context: Option<&str>) -> Vec<String> {
+    let mut out = match context {
+        Some(ctx) if !ctx.is_empty() => vec!["--context".to_string(), ctx.to_string()],
+        _ => vec![],
+    };
     out.extend(args.iter().map(|s| s.to_string()));
     out
 }
 
+fn with_helm_kube_context(args: &[&str]) -> Vec<String> {
+    let ctx = kube_context_from_env();
+    with_helm_kube_context_value(args, ctx.as_deref())
+}
+
+fn with_helm_kube_context_value(args: &[&str], context: Option<&str>) -> Vec<String> {
+    let Some(ctx) = context.filter(|ctx| !ctx.is_empty()) else {
+        return args.iter().map(|s| s.to_string()).collect();
+    };
+    if args.first() == Some(&"repo") {
+        return args.iter().map(|s| s.to_string()).collect();
+    }
+
+    let mut out = Vec::with_capacity(args.len() + 2);
+    if let Some((command, rest)) = args.split_first() {
+        out.push((*command).to_string());
+        out.push("--kube-context".to_string());
+        out.push(ctx.to_string());
+        out.extend(rest.iter().map(|s| s.to_string()));
+    } else {
+        out.push("--kube-context".to_string());
+        out.push(ctx.to_string());
+    }
+    out
+}
+
 /// Build a `Command` for kubectl with `--context` injected when configured.
 pub fn kubectl_command(args: &[&str]) -> Command {
     let full = with_kube_context(args);
@@ -105,19 +136,18 @@ pub enum LocalCommands {
     /// Destroy the local cluster
     Destroy,
     /// Uninstall the local cluster backend
-    Uninstall,
+    Uninstall(uninstall::UninstallArgs),
 }
 
 pub fn run(args: &LocalArgs) -> Result<(), Box<dyn Error>> {
-    let backend = backend::resolve(args.backend);
-    // Plumb the context through the same env channel the kubectl helpers
-    // read, so every subcommand's kubectl calls target the chosen cluster.
-    // Without an explicit --context, use the backend's own context so
-    // commands work regardless of kubeconfig's current-context.
-    match &args.context {
-        Some(ctx) if !ctx.is_empty() => std::env::set_var(HOPS_KUBE_CONTEXT_ENV, ctx),
-        _ => std::env::set_var(HOPS_KUBE_CONTEXT_ENV, backend.kube_context()),
-    }
+    let explicit_context = args.context.as_deref().filter(|ctx| !ctx.is_empty());
+    let install_backend = matches!(&args.command, LocalCommands::Install)
+        .then(|| args.backend.unwrap_or_else(backend::platform_default));
+    let activation_flag = install_backend.or(args.backend);
+    let install_context = install_backend.map(backend::Backend::kube_context);
+    let activation_context = explicit_context.or(install_context);
+    let backend = backend::activate(activation_flag, activation_context);
+
     match &args.command {
         LocalCommands::Install => install::run(backend),
         LocalCommands::Reset => reset::run(backend),
@@ -131,7 +161,7 @@ pub fn run(args: &LocalArgs) -> Result<(), Box<dyn Error>> {
         LocalCommands::Listmonk(listmonk_args) => listmonk::run(listmonk_args),
         LocalCommands::Stop => stop::run(backend),
         LocalCommands::Destroy => destroy::run(backend),
-        LocalCommands::Uninstall => uninstall::run(backend),
+        LocalCommands::Uninstall(uninstall_args) => uninstall::run(backend, uninstall_args),
     }
 }
 
@@ -143,11 +173,17 @@ pub fn run_cmd(program: &str, args: &[&str]) -> Result<(), Box<dyn Error>> {
         let refs: Vec<&str> = full.iter().map(|s| s.as_str()).collect();
         return run_cmd_with_logged_args(program, &refs, &refs);
     }
+    if program == "helm" {
+        let full = with_helm_kube_context(args);
+        let refs: Vec<&str> = full.iter().map(|s| s.as_str()).collect();
+        return run_cmd_with_logged_args(program, &refs, &refs);
+    }
     run_cmd_with_logged_args(program, args, args)
 }
 
 /// Run an external command and capture stdout.
-/// For kubectl commands, automatically injects `--context` when configured.
+/// For kubectl/helm commands, automatically injects the active context when
+/// configured.
 pub fn run_cmd_output(program: &str, args: &[&str]) -> Result<String, Box<dyn Error>> {
     if program == "kubectl" {
         let full = with_kube_context(args);
@@ -159,6 +195,16 @@ pub fn run_cmd_output(program: &str, args: &[&str]) -> Result<String, Box<dyn Er
         }
         return Ok(String::from_utf8_lossy(&output.stdout).to_string());
     }
+    if program == "helm" {
+        let full = with_helm_kube_context(args);
+        log::debug!("Running: {} {}", program, full.join(" "));
+        let output = Command::new(program).args(&full).output()?;
+        if !output.status.success() {
+            let stderr = String::from_utf8_lossy(&output.stderr);
+            return Err(format!("{} exited with {}: {}", program, output.status, stderr).into());
+        }
+        return Ok(String::from_utf8_lossy(&output.stdout).to_string());
+    }
 
     log::debug!("Running: {} {}", program, args.join(" "));
     let output = Command::new(program).args(args).output()?;
@@ -269,3 +315,48 @@ pub fn kubectl_patch_merge(
     let logged_refs: Vec<&str> = full_logged.iter().map(|s| s.as_str()).collect();
     run_cmd_with_logged_args("kubectl", &args_refs, &logged_refs)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn strings(args: Vec<String>) -> Vec<String> {
+        args
+    }
+
+    #[test]
+    fn kubectl_args_prepend_context() {
+        assert_eq!(
+            strings(with_kube_context_value(&["get", "pods"], Some("kind-hops"))),
+            vec!["--context", "kind-hops", "get", "pods"]
+        );
+    }
+
+    #[test]
+    fn helm_upgrade_injects_kube_context_after_subcommand() {
+        assert_eq!(
+            strings(with_helm_kube_context_value(
+                &["upgrade", "--install", "crossplane"],
+                Some("kind-hops")
+            )),
+            vec![
+                "upgrade",
+                "--kube-context",
+                "kind-hops",
+                "--install",
+                "crossplane"
+            ]
+        );
+    }
+
+    #[test]
+    fn helm_repo_commands_skip_kube_context() {
+        assert_eq!(
+            strings(with_helm_kube_context_value(
+                &["repo", "update", "crossplane-stable"],
+                Some("kind-hops")
+            )),
+            vec!["repo", "update", "crossplane-stable"]
+        );
+    }
+}
diff --git a/src/commands/local/start.rs b/src/commands/local/start.rs
index 7acd95e..136d760 100644
--- a/src/commands/local/start.rs
+++ b/src/commands/local/start.rs
@@ -52,7 +52,9 @@ pub fn run(backend: backend::Backend, args: &StartArgs) -> Result<(), Box<dyn Er
             "https://charts.crossplane.io/stable",
         ],
     )?;
-    run_cmd("helm", &["repo", "update"])?;
+    // Update only our repo — a bare `helm repo update` fails outright when any
+    // unrelated repo in the user's helm config has gone stale.
+    run_cmd("helm", &["repo", "update", "crossplane-stable"])?;
 
     // 5. Install Crossplane
     log::info!("Installing Crossplane...");
diff --git a/src/commands/local/uninstall.rs b/src/commands/local/uninstall.rs
index 0587750..6cc4a4e 100644
--- a/src/commands/local/uninstall.rs
+++ b/src/commands/local/uninstall.rs
@@ -1,8 +1,55 @@
 use super::backend::Backend;
+use clap::Args;
 use std::error::Error;
 use std::io::{self, Write};
 
-pub fn run(backend: Backend) -> Result<(), Box<dyn Error>> {
+#[derive(Args, Debug)]
+pub struct UninstallArgs {
+    /// Uninstall the backend binary even if its local cluster still exists.
+    #[arg(long)]
+    pub force: bool,
+}
+
+pub fn run(backend: Backend, args: &UninstallArgs) -> Result<(), Box<dyn Error>> {
+    run_inner(
+        backend,
+        args.force,
+        Backend::cluster_exists,
+        Backend::uninstall,
+        super::backend::clear_persisted,
+        confirm_uninstall,
+    )
+}
+
+fn run_inner<ClusterExists, Uninstall, ClearPersisted, Confirm>(
+    backend: Backend,
+    force: bool,
+    cluster_exists: ClusterExists,
+    uninstall: Uninstall,
+    clear_persisted: ClearPersisted,
+    confirm: Confirm,
+) -> Result<(), Box<dyn Error>>
+where
+    ClusterExists: FnOnce(Backend) -> bool,
+    Uninstall: FnOnce(Backend) -> Result<(), Box<dyn Error>>,
+    ClearPersisted: FnOnce() -> Result<(), Box<dyn Error>>,
+    Confirm: FnOnce(Backend) -> Result<bool, Box<dyn Error>>,
+{
+    if !force && cluster_exists(backend) {
+        return Err("destroy the cluster first: `hops local destroy` (or pass --force to uninstall the backend binary anyway)".into());
+    }
+
+    if confirm(backend)? {
+        uninstall(backend)?;
+        clear_persisted()?;
+    } else {
+        log::info!("Uninstall cancelled");
+    }
+
+    Ok(())
+}
+
+fn confirm_uninstall(backend: Backend) -> Result<bool, Box<dyn Error>> {
     print!(
         "Uninstall {}? This will remove the binary. [y/N] ",
         backend.name()
@@ -12,12 +59,113 @@ pub fn run(backend: Backend) -> Result<(), Box<dyn Error>> {
     let mut input = String::new();
     io::stdin().read_line(&mut input)?;
 
-    if input.trim().eq_ignore_ascii_case("y") {
-        backend.uninstall()?;
-        super::backend::clear_persisted()?;
-    } else {
-        log::info!("Uninstall cancelled");
+    Ok(input.trim().eq_ignore_ascii_case("y"))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::cell::RefCell;
+    use std::rc::Rc;
+
+    #[test]
+    fn refuses_uninstall_when_cluster_exists_without_force() {
+        let events = Rc::new(RefCell::new(Vec::new()));
+        let err = run_inner(
+            Backend::Kind,
+            false,
+            {
+                let events = Rc::clone(&events);
+                move |_| {
+                    events.borrow_mut().push("cluster_exists");
+                    true
+                }
+            },
+            |_| {
+                events.borrow_mut().push("uninstall");
+                Ok(())
+            },
+            || {
+                events.borrow_mut().push("clear");
+                Ok(())
+            },
+            |_| {
+                events.borrow_mut().push("confirm");
+                Ok(true)
+            },
+        )
+        .expect_err("cluster guard should refuse uninstall");
+
+        assert!(err.to_string().contains("destroy the cluster first"));
+        assert_eq!(&*events.borrow(), &["cluster_exists"]);
     }
 
-    Ok(())
+    #[test]
+    fn force_uninstalls_without_cluster_probe_and_clears_after_success() {
+        let events = Rc::new(RefCell::new(Vec::new()));
+        run_inner(
+            Backend::Kind,
+            true,
+            {
+                let events = Rc::clone(&events);
+                move |_| {
+                    events.borrow_mut().push("cluster_exists");
+                    true
+                }
+            },
+            {
+                let events = Rc::clone(&events);
+                move |_| {
+                    events.borrow_mut().push("uninstall");
+                    Ok(())
+                }
+            },
+            {
+                let events = Rc::clone(&events);
+                move || {
+                    events.borrow_mut().push("clear");
+                    Ok(())
+                }
+            },
+            {
+                let events = Rc::clone(&events);
+                move |_| {
+                    events.borrow_mut().push("confirm");
+                    Ok(true)
+                }
+            },
+        )
+        .expect("force should allow uninstall");
+
+        assert_eq!(&*events.borrow(), &["confirm", "uninstall", "clear"]);
+    }
+
+    #[test]
+    fn failed_uninstall_does_not_clear_persisted_backend() {
+        let events = Rc::new(RefCell::new(Vec::new()));
+        let err = run_inner(
+            Backend::Kind,
+            false,
+            |_| false,
+            {
+                let events = Rc::clone(&events);
+                move |_| {
+                    events.borrow_mut().push("uninstall");
+                    Err("brew failed".into())
+                }
+            },
+            {
+                let events = Rc::clone(&events);
+                move || {
+                    events.borrow_mut().push("clear");
+                    Ok(())
+                }
+            },
+            |_| Ok(true),
+        )
+        .expect_err("uninstall failure should propagate");
+
+        assert_eq!(err.to_string(), "brew failed");
+        assert_eq!(&*events.borrow(), &["uninstall"]);
+    }
 }
diff --git a/src/commands/provider/install.rs b/src/commands/provider/install.rs
index c10a13c..c45a3e6 100644
--- a/src/commands/provider/install.rs
+++ b/src/commands/provider/install.rs
@@ -1,12 +1,11 @@
-use crate::commands::local::backend::{self, wire_local_registry};
+use crate::commands::local::backend::{self, Backend};
 use crate::commands::local::package_install::{
     docker_arch, ensure_cached_repo_checkout_at, ensure_registry, parse_repo_spec,
     resolve_repo_install_target, run_watch, sanitize_name_component, RepoInstallTarget, RepoSpec,
     REGISTRY_PULL, REGISTRY_PUSH,
 };
 use crate::commands::local::{
-    kubectl_apply_stdin, run_cmd, run_cmd_output, HOPS_KUBE_CONTEXT_ENV, MANAGED_BY_LABEL,
-    PROVIDER_INSTALL_MANAGED_BY,
+    kubectl_apply_stdin, run_cmd, run_cmd_output, MANAGED_BY_LABEL, PROVIDER_INSTALL_MANAGED_BY,
 };
 use clap::Args;
 use serde::Deserialize;
@@ -54,6 +53,10 @@ pub struct ProviderInstallArgs {
     #[arg(long)]
     pub context: Option<String>,
 
+    /// Local cluster backend whose node should be wired for local package pulls.
+    #[arg(long, value_enum)]
+    pub backend: Option<Backend>,
+
     /// Watch the project directory for changes and re-run install automatically
     #[arg(long, conflicts_with = "repo")]
     pub watch: bool,
@@ -74,9 +77,7 @@ struct PackageMetadata {
 }
 
 pub fn run(args: &ProviderInstallArgs) -> Result<(), Box<dyn Error>> {
-    if let Some(ctx) = &args.context {
-        std::env::set_var(HOPS_KUBE_CONTEXT_ENV, ctx);
-    }
+    let backend = backend::activate(args.backend, args.context.as_deref());
 
     match (args.repo.as_deref(), args.version.as_deref()) {
         (Some(repo), Some(version)) => {
@@ -87,10 +88,14 @@ pub fn run(args: &ProviderInstallArgs) -> Result<(), Box<dyn Error>> {
             args.skip_dependency_resolution,
             args.version_prefix.as_deref(),
             args.branch.as_deref(),
+            backend,
+            args.backend,
+            args.context.as_deref(),
         ),
         (None, _) => {
             let path = args.path.as_deref().unwrap_or(".");
             let prefix = args.version_prefix.clone();
+            prepare_local_registry(backend, args.backend, args.context.as_deref())?;
             run_local_path(path, args.skip_dependency_resolution, prefix.as_deref())?;
 
             if args.watch {
@@ -112,11 +117,15 @@ fn run_repo_install(
     skip_dependency_resolution: bool,
     version_prefix: Option<&str>,
     branch: Option<&str>,
+    backend: Backend,
+    backend_flag: Option<Backend>,
+    context: Option<&str>,
 ) -> Result<(), Box<dyn Error>> {
     let spec = parse_repo_spec(repo)?;
     match resolve_repo_install_target(&spec)? {
         RepoInstallTarget::SourceBuild => {
             let cache_path = ensure_cached_repo_checkout_at(&spec, branch)?;
+            prepare_local_registry(backend, backend_flag, context)?;
             run_local_path(
                 &cache_path.to_string_lossy(),
                 skip_dependency_resolution,
@@ -129,6 +138,15 @@ fn run_repo_install(
     }
 }
 
+fn prepare_local_registry(
+    backend: Backend,
+    backend_flag: Option<Backend>,
+    context: Option<&str>,
+) -> Result<(), Box<dyn Error>> {
+    ensure_registry()?;
+    backend::wire_local_registry_for_target(backend, backend_flag, context)
+}
+
 fn apply_repo_version(
     repo: &str,
     version: &str,
@@ -186,9 +204,6 @@ fn run_local_path(
     let provider_name = read_provider_name(dir)?;
     log::info!("Provider package name: {}", provider_name);
 
-    ensure_registry()?;
-    wire_local_registry(backend::resolve(None))?;
-
     // Resolve the existing upstream Provider before building so we can:
     //   1. carry the upstream package URL into the new `spec.package` (Crossplane
     //      records the URL-without-tag as the Lock Source; deps declared as
@@ -337,10 +352,10 @@ fn find_xpkg_for_provider(
         .filter_map(|e| e.ok())
         .filter(|e| {
             let p = e.path();
-            p.extension().map_or(false, |ext| ext == "xpkg")
+            p.extension().is_some_and(|ext| ext == "xpkg")
                 && p.file_name()
                     .and_then(|n| n.to_str())
-                    .map_or(false, |n| n.starts_with(&prefix))
+                    .is_some_and(|n| n.starts_with(&prefix))
         })
         .map(|e| e.path())
         .collect();
@@ -1113,6 +1128,15 @@ mod tests {
         assert!(!image.contains(REGISTRY_PULL));
     }
 
+    #[test]
+    fn local_registry_wiring_skips_foreign_context_without_backend_flag() {
+        assert!(!backend::should_wire_local_registry(
+            None,
+            Some("kind-hops"),
+            Backend::Colima
+        ));
+    }
+
     #[test]
     fn build_cluster_role_binding_yaml_targets_service_account() {
         let yaml = build_cluster_role_binding_yaml("p-cluster-admin", "p");