From 58f4ecb6366f11e47c79600ed11f20a8c76c2c0b Mon Sep 17 00:00:00 2001
From: Patrick Lee Scott <pat@patscott.io>
Date: Wed, 19 Aug 2026 14:52:29 -0500
Subject: [PATCH] fix(local)!: align GitOps lifecycle and runtime resilience

BREAKING CHANGE: remove local up, open, and stop; use local gitops cluster with --down for lifecycle management.
---
 README.md                                     |  25 +-
 bootstrap/registry/registry.yaml              |   7 +
 skills/claude/references/local-setup.md       |  38 +-
 skills/claude/references/local-workbench.md   |  16 +-
 src/commands/local/backend/kind.rs            | 325 +++++++++++++++++-
 src/commands/local/gitops.rs                  | 323 ++++++++---------
 src/commands/local/mod.rs                     |  79 ++++-
 src/commands/local/open.rs                    |  99 ------
 src/commands/local/start.rs                   |  61 +++-
 src/commands/local/stop.rs                    |   6 -
 .../local/workbench/cluster_gitops.rs         | 177 ----------
 src/commands/local/workbench/definition.rs    | 141 ++++----
 tests/local_cluster_definition.rs             |  32 +-
 13 files changed, 767 insertions(+), 562 deletions(-)
 delete mode 100644 src/commands/local/open.rs
 delete mode 100644 src/commands/local/stop.rs

diff --git a/README.md b/README.md
index 8b3d98d..f2b166e 100644
--- a/README.md
+++ b/README.md
@@ -120,8 +120,7 @@ spec:
 From that project root:
 
 ```bash
-hops local up
-hops local gitops cluster ./.gitops/local/cluster
+hops local gitops cluster ./.gitops/local/cluster.yaml
 hops local gitops environment ./.gitops/local/environment.yaml --name main
 ```
 
@@ -131,17 +130,27 @@ From another checkout of the same project:
 hops local gitops environment ./.gitops/local/environment.yaml --name feature-auth
 ```
 
-`up` validates the Cluster before starting or reusing it. `gitops cluster`
-watches shared `.gitops/local/cluster` manifests. `environment` validates the
-Environment against that Cluster, renders each deploy's `.gitops/promote`
-chart, applies the resulting local Applications to the runtime namespace, and
-watches `.gitops/local/environment.yaml` plus the referenced
-`.gitops/promote` and `.gitops/local` charts. Each application's
+`gitops cluster` validates the Cluster, starts or resumes it, bootstraps the
+local control plane, and watches the declared shared manifests.
+`environment` validates the Environment against that Cluster, turns each
+deploy's `.gitops/local` chart (or explicit `deploys[].chart`) into a local
+Application, applies it to the runtime namespace, and watches
+`.gitops/local/environment.yaml` plus those chart roots. Each application's
 `.gitops/local` chart owns its editable local workload; `.gitops/deploy` is a
 separate cloud workload chart selected by promotion outside local mode.
 The runtime name, namespace, checkout path, and Cluster binding are local state;
 they are not committed to the Cluster definition.
 
+Use the same commands for teardown:
+
+```bash
+hops local gitops environment --name feature-auth --down
+hops local gitops cluster ./.gitops/local/cluster.yaml --down
+```
+
+Deleting a watched Environment definition also purges and unregisters its
+runtime Environment.
+
 An existing kind Cluster with a different exact `mountRoot` fails with an
 explicit reset/recreate instruction and is never silently deleted. A legacy
 directory of pre-rendered Application YAMLs is still accepted by `environment`
diff --git a/bootstrap/registry/registry.yaml b/bootstrap/registry/registry.yaml
index 02cd067..98f84df 100644
--- a/bootstrap/registry/registry.yaml
+++ b/bootstrap/registry/registry.yaml
@@ -55,6 +55,11 @@ spec:
               scheme: HTTPS
             initialDelaySeconds: 2
             periodSeconds: 5
+            # A laptop resuming several local control planes can briefly take
+            # seconds to service TLS. Do not turn CPU contention into a
+            # destructive registry restart loop.
+            timeoutSeconds: 15
+            failureThreshold: 6
           livenessProbe:
             httpGet:
               path: /v2/
@@ -62,6 +67,8 @@ spec:
               scheme: HTTPS
             initialDelaySeconds: 5
             periodSeconds: 10
+            timeoutSeconds: 15
+            failureThreshold: 6
       volumes:
         - name: registry-data
           persistentVolumeClaim:
diff --git a/skills/claude/references/local-setup.md b/skills/claude/references/local-setup.md
index 66687e2..634b4ce 100644
--- a/skills/claude/references/local-setup.md
+++ b/skills/claude/references/local-setup.md
@@ -3,18 +3,17 @@
 ## Quick Start
 
 ```bash
-# 1. Start local k8s + Crossplane + providers + registry
-#    (provider selection is user-local: ~/.hops/local/providers.json)
-hops local start --cluster-provider kind --docker-provider dory --cluster-name hops
+# 1. Start/resume the declared cluster and watch shared GitOps manifests
+hops local gitops cluster ./.gitops/local/cluster.yaml
 
-# 2. Install platform packages into the CP *and* pin them in cluster gitops
+# 2. Add or update platform packages in .gitops/local/cluster when needed
 hops config install --repo hops-ops/psql-stack --version v0.9.1 \
-  --gitops ./gitops/cluster --local
+  --gitops ./.gitops/local/cluster --local
 hops config install --repo hops-ops/auth-stack --version v1.6.0 \
-  --gitops ./gitops/cluster --local
+  --gitops ./.gitops/local/cluster --local
 
-# 3. Watch/apply cluster gitops (packages + XRs). Or pass --gitops on start.
-hops local gitops cluster ./gitops/cluster
+# 3. Register this checkout as an Environment
+hops local gitops environment ./.gitops/local/environment.yaml --name main
 
 # 4. Optional cloud provider auth (writes live Secrets; use --gitops for non-secret YAML)
 hops local aws --profile hops
@@ -31,7 +30,13 @@ to `default` (scaffolded by `config install --gitops --local`). See
 ### `hops local install`
 Installs Colima via Homebrew.
 
-### `hops local start`
+### `hops local gitops cluster <cluster.yaml>`
+
+This is the normal lifecycle command. It validates the Kubernetes-shaped
+Cluster definition, invokes the local start/bootstrap pipeline, and applies +
+watches the definition's `spec.manifests.path`.
+
+The underlying `hops local start` command:
 - Starts the chosen backend (colima / kind / dory)
 - Installs **pinned** Crossplane Helm chart (`CROSSPLANE_CHART_VERSION` in `start.rs`)
 - Applies bootstrap Providers (pinned tags in `bootstrap/providers/`):
@@ -40,13 +45,9 @@ Installs Colima via Homebrew.
 - Applies ProviderConfigs named `default`, local registry, DRCs
 - Configures node trust for the in-cluster registry
 
-With **`--gitops PATH`** (e.g. `./gitops/cluster`):
-1. Writes the same helm/k8s bootstrap into the tree (`providers/`, `providerconfigs/`, `runtime/`)
-2. Runs `hops local gitops cluster PATH` (apply + watch) so day-to-day CP state is gitops-owned
-
 ```bash
-hops local start --cluster-provider kind --docker-provider dory \
-  --cluster-name hops --gitops ./gitops/cluster
+hops local gitops cluster ./.gitops/local/cluster.yaml
+hops local gitops cluster ./.gitops/local/cluster.yaml --down
 ```
 
 **Version bumps:** Renovate owns these pins (`cli/renovate.json` customManagers →
@@ -62,8 +63,11 @@ hops provider install --path /path/to/provider-helm --gitops ./gitops/cluster
 
 See [local-source-packages.md](./local-source-packages.md).
 
-### `hops local stop` / `hops local destroy` / `hops local uninstall`
-Stop, delete, or uninstall Colima respectively.
+### Cluster teardown
+
+`hops local gitops cluster <cluster.yaml> --down` stops the declared Cluster
+while preserving its data. `hops local destroy` remains the explicit,
+destructive cluster deletion command; `hops local uninstall` removes tooling.
 
 ### `hops local aws --profile <PROFILE>`
 
diff --git a/skills/claude/references/local-workbench.md b/skills/claude/references/local-workbench.md
index 8992d09..2d2c723 100644
--- a/skills/claude/references/local-workbench.md
+++ b/skills/claude/references/local-workbench.md
@@ -15,8 +15,7 @@ tree copy). You do not need to learn volume types.
 
 ```bash
 # Dory app running (engine healthy). Product Dory Kubernetes is optional.
-hops local up
-hops local gitops cluster ./.gitops/local/cluster
+hops local gitops cluster ./.gitops/local/cluster.yaml
 ```
 
 Context is typically `kind-hops`. Confirm mounts:
@@ -36,14 +35,15 @@ Stock Dory k8s (`--cluster-provider dory --docker-provider dory`) is fine for pl
 **cannot** hostPath-mount Mac paths into the node; delivery falls back to sync.
 
 ```bash
-hops local up --cluster-provider dory --docker-provider dory
+hops local gitops cluster ./.gitops/local/cluster.yaml \
+  --cluster-provider dory --docker-provider dory
 ```
 
 ## Daily loop
 
 ```bash
-# Shared CP watch (if start did not use --gitops, or after Ctrl+C)
-hops local gitops cluster ./.gitops/local/cluster
+# Start/resume the Cluster and watch its shared control-plane manifests
+hops local gitops cluster ./.gitops/local/cluster.yaml
 
 # One Environment per checkout (namespace = --name) — watches by default
 hops local gitops environment ./.gitops/local/environment.yaml --name dogfood
@@ -52,6 +52,9 @@ hops local gitops environment ./.gitops/local/environment.yaml --name dogfood
 ```
 
 Watch is the default for both gitops commands. Use `--once` for a single reconcile (CI/scripts).
+Use `environment --name <name> --down` to purge one Environment and
+`cluster <cluster.yaml> --down` to stop the control plane while preserving its
+named node volume.
 
 ## Concurrent worktrees
 
@@ -73,8 +76,7 @@ Each name maps to namespace `<name>`.
 ```bash
 cd distributed/tests/e2e-ui
 # Prefer kind-on-Dory for hostPath HMR (see One-time prerequisite)
-hops local up
-hops local gitops cluster ./.gitops/local/cluster
+hops local gitops cluster ./.gitops/local/cluster.yaml
 hops local gitops environment ./.gitops/local/environment.yaml --name dogfood
 ```
 
diff --git a/src/commands/local/backend/kind.rs b/src/commands/local/backend/kind.rs
index dc516c8..3ab6599 100644
--- a/src/commands/local/backend/kind.rs
+++ b/src/commands/local/backend/kind.rs
@@ -23,12 +23,14 @@ use crate::commands::local::package_install::{REGISTRY_PULL, REGISTRY_PUSH};
 use crate::commands::local::{command_exists, run_cmd, run_cmd_output};
 use std::collections::BTreeSet;
 use std::error::Error;
+use std::fs;
 use std::io::Write;
 use std::net::{Ipv4Addr, TcpListener};
 use std::path::{Path, PathBuf};
 use std::process::{Command, Stdio};
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::thread;
-use std::time::Duration;
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 /// Default kind cluster name (and historical hard-coded value).
 pub const DEFAULT_CLUSTER_NAME: &str = "hops";
@@ -39,6 +41,64 @@ const REGISTRY_HOST_PORT_END: u16 = 30599;
 const INOTIFY_SYSCTL_PATH: &str = "/etc/sysctl.d/99-hops-local-inotify.conf";
 const INOTIFY_MAX_USER_INSTANCES: u32 = 8192;
 const INOTIFY_MAX_USER_WATCHES: u32 = 1_048_576;
+const KIND_VOLUME_MANAGED_LABEL: &str = "dev.hops.local.managed";
+const KIND_VOLUME_CLUSTER_LABEL: &str = "dev.hops.local.kind.cluster";
+const KIND_VOLUME_NODE_LABEL: &str = "dev.hops.local.kind.node";
+static KIND_DOCKER_PROXY_COUNTER: AtomicU64 = AtomicU64::new(0);
+const KIND_DOCKER_PROXY: &str = r#"#!/bin/sh
+set -eu
+
+real_docker="${HOPS_KIND_REAL_DOCKER:?HOPS_KIND_REAL_DOCKER is required}"
+
+if [ "${1-}" != "run" ]; then
+  exec "$real_docker" "$@"
+fi
+
+previous=""
+node_name=""
+cluster_name=""
+node_role=""
+for argument in "$@"; do
+  if [ "$previous" = "--name" ]; then
+    node_name="$argument"
+  elif [ "$previous" = "--label" ]; then
+    case "$argument" in
+      io.x-k8s.kind.cluster=*) cluster_name=${argument#*=} ;;
+      io.x-k8s.kind.role=*) node_role=${argument#*=} ;;
+    esac
+  fi
+  previous="$argument"
+done
+
+case "$node_role" in
+  control-plane|worker) ;;
+  *) exec "$real_docker" "$@" ;;
+esac
+
+if [ -z "$node_name" ] || [ -z "$cluster_name" ]; then
+  echo "hops kind docker adapter: node name and cluster label are required" >&2
+  exit 1
+fi
+
+volume_name="hops-kind-${node_name}-data"
+if "$real_docker" volume inspect "$volume_name" >/dev/null 2>&1; then
+  managed=$("$real_docker" volume inspect --format '{{ index .Labels "dev.hops.local.managed" }}' "$volume_name")
+  owner=$("$real_docker" volume inspect --format '{{ index .Labels "dev.hops.local.kind.cluster" }}' "$volume_name")
+  if [ "$managed" != "true" ] || [ "$owner" != "$cluster_name" ]; then
+    echo "hops kind docker adapter: refusing non-Hops volume name collision: $volume_name" >&2
+    exit 1
+  fi
+else
+  "$real_docker" volume create \
+    --label "dev.hops.local.managed=true" \
+    --label "dev.hops.local.kind.cluster=$cluster_name" \
+    --label "dev.hops.local.kind.node=$node_name" \
+    "$volume_name" >/dev/null
+fi
+
+shift
+exec "$real_docker" run --volume "$volume_name:/var" "$@"
+"#;
 const INSTALL_INOTIFY_SYSCTL_SCRIPT: &str = r#"set -eu
 target="$1"
 expected_instances="$2"
@@ -439,6 +499,85 @@ fn kind_cmd(args: &[&str]) -> Command {
     c
 }
 
+/// A PATH-scoped Docker adapter used only while `kind create cluster` runs.
+///
+/// Stock kind deliberately passes `--volume /var`, which makes Docker create
+/// an opaque anonymous volume for every node. Docker accepts a second named
+/// mount for the same destination and selects the named mount. The adapter
+/// recognizes kind node runs from their labels and adds a deterministic,
+/// Hops-labeled volume while delegating every other Docker invocation.
+struct KindDockerProxy {
+    dir: PathBuf,
+    real_docker: PathBuf,
+}
+
+impl KindDockerProxy {
+    fn new() -> Result<Self, Box<dyn Error>> {
+        let real_docker = executable_in_path("docker")
+            .ok_or("docker executable is not available on PATH for kind")?;
+        Self::with_real_docker(real_docker)
+    }
+
+    fn with_real_docker(real_docker: PathBuf) -> Result<Self, Box<dyn Error>> {
+        let nonce = SystemTime::now().duration_since(UNIX_EPOCH)?.as_nanos();
+        let counter = KIND_DOCKER_PROXY_COUNTER.fetch_add(1, Ordering::Relaxed);
+        let dir = std::env::temp_dir().join(format!(
+            "hops-kind-docker-{}-{nonce}-{counter}",
+            std::process::id()
+        ));
+        fs::create_dir(&dir)?;
+        let proxy = dir.join("docker");
+        fs::write(&proxy, KIND_DOCKER_PROXY)?;
+        set_executable(&proxy)?;
+        Ok(Self { dir, real_docker })
+    }
+
+    fn apply(&self, command: &mut Command) -> Result<(), Box<dyn Error>> {
+        let existing = std::env::var_os("PATH").unwrap_or_default();
+        let path = std::env::join_paths(
+            std::iter::once(self.dir.clone()).chain(std::env::split_paths(&existing)),
+        )?;
+        command
+            .env("PATH", path)
+            .env("HOPS_KIND_REAL_DOCKER", &self.real_docker);
+        Ok(())
+    }
+}
+
+impl Drop for KindDockerProxy {
+    fn drop(&mut self) {
+        if let Err(error) = fs::remove_dir_all(&self.dir) {
+            log::debug!(
+                "unable to remove temporary kind docker adapter {}: {error}",
+                self.dir.display()
+            );
+        }
+    }
+}
+
+fn executable_in_path(name: &str) -> Option<PathBuf> {
+    std::env::var_os("PATH").and_then(|path| {
+        std::env::split_paths(&path)
+            .map(|entry| entry.join(name))
+            .find(|candidate| candidate.is_file())
+    })
+}
+
+#[cfg(unix)]
+fn set_executable(path: &Path) -> Result<(), Box<dyn Error>> {
+    use std::os::unix::fs::PermissionsExt;
+
+    let mut permissions = fs::metadata(path)?.permissions();
+    permissions.set_mode(0o700);
+    fs::set_permissions(path, permissions)?;
+    Ok(())
+}
+
+#[cfg(not(unix))]
+fn set_executable(_path: &Path) -> Result<(), Box<dyn Error>> {
+    Err("named kind node volumes currently require a Unix-compatible Docker CLI".into())
+}
+
 fn docker_output(args: &[&str]) -> Result<String, Box<dyn Error>> {
     let output = docker_cmd(args).output()?;
     if !output.status.success() {
@@ -530,6 +669,7 @@ pub fn destroy() -> Result<(), Box<dyn Error>> {
     if !status.success() {
         return Err(format!("kind delete cluster exited with {}", status).into());
     }
+    remove_cluster_node_data_volumes(&name)?;
     log::info!("kind cluster deleted");
     Ok(())
 }
@@ -632,6 +772,10 @@ fn create_cluster() -> Result<(), Box<dyn Error>> {
         );
     }
     let name = active_cluster_name();
+    // A missing kind node with an owned volume is residue from an interrupted
+    // create or external container cleanup. Starting a fresh kind node on old
+    // etcd/containerd state is unsupported, so recreate only Hops-owned data.
+    remove_cluster_node_data_volumes(&name)?;
     if let Some(ref m) = mount {
         log::info!(
             "Creating kind cluster '{name}' with extraMounts {} → {} (hostPath delivery)...",
@@ -644,7 +788,10 @@ fn create_cluster() -> Result<(), Box<dyn Error>> {
         );
     }
 
-    let mut child = kind_cmd(&["create", "cluster", "--name", &name, "--config", "-"])
+    let docker_proxy = KindDockerProxy::new()?;
+    let mut command = kind_cmd(&["create", "cluster", "--name", &name, "--config", "-"]);
+    docker_proxy.apply(&mut command)?;
+    let mut child = command
         .stdin(Stdio::piped())
         .stdout(Stdio::inherit())
         .stderr(Stdio::inherit())
@@ -654,6 +801,9 @@ fn create_cluster() -> Result<(), Box<dyn Error>> {
     }
     let status = child.wait()?;
     if !status.success() {
+        if let Err(error) = remove_cluster_node_data_volumes(&name) {
+            log::warn!("unable to clean named kind volumes after failed create: {error}");
+        }
         return Err(format!("kind create cluster exited with {}", status).into());
     }
 
@@ -667,6 +817,44 @@ fn create_cluster() -> Result<(), Box<dyn Error>> {
     Ok(())
 }
 
+fn remove_cluster_node_data_volumes(cluster_name: &str) -> Result<(), Box<dyn Error>> {
+    let managed_filter = format!("label={KIND_VOLUME_MANAGED_LABEL}=true");
+    let cluster_filter = format!("label={KIND_VOLUME_CLUSTER_LABEL}={cluster_name}");
+    let volumes = docker_output(&[
+        "volume",
+        "ls",
+        "--quiet",
+        "--filter",
+        &managed_filter,
+        "--filter",
+        &cluster_filter,
+    ])?;
+
+    for volume in volumes
+        .lines()
+        .map(str::trim)
+        .filter(|name| !name.is_empty())
+    {
+        let node = docker_output(&[
+            "volume",
+            "inspect",
+            "--format",
+            &format!("{{{{ index .Labels {:?} }}}}", KIND_VOLUME_NODE_LABEL),
+            volume,
+        ])?;
+        let expected = format!("hops-kind-{}-data", node.trim());
+        if volume != expected {
+            return Err(format!(
+                "refusing to remove Hops-labeled kind volume {volume}: expected {expected} from its node label"
+            )
+            .into());
+        }
+        docker_run(&["volume", "rm", volume])?;
+        log::info!("removed kind node data volume {volume}");
+    }
+    Ok(())
+}
+
 /// kind writes the Docker engine's published address into kubeconfig. Dory's
 /// engine reports `0.0.0.0`, which is reachable through its local proxy but is
 /// not present in the API server certificate. Rewrite only that Dory-specific
@@ -889,6 +1077,7 @@ fn parse_kind_version(output: &str) -> Option<(u32, u32)> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::fs;
     use std::path::Path;
 
     #[test]
@@ -1074,4 +1263,136 @@ fs.inotify.max_user_watches = 1048576\n"
         );
         assert_eq!(normalized_dory_server("https://0.0.0.0:63903", 6443), None);
     }
+
+    #[cfg(unix)]
+    #[test]
+    fn kind_docker_proxy_names_node_volume_and_delegates_other_calls() {
+        let root = test_dir("kind-docker-proxy");
+        let fake_docker = root.join("real-docker");
+        let calls = root.join("calls");
+        fs::write(
+            &fake_docker,
+            r#"#!/bin/sh
+set -eu
+printf 'CALL' >> "$HOPS_TEST_DOCKER_CALLS"
+for argument in "$@"; do
+  printf '\t%s' "$argument" >> "$HOPS_TEST_DOCKER_CALLS"
+done
+printf '\n' >> "$HOPS_TEST_DOCKER_CALLS"
+if [ "${1-}" = "volume" ] && [ "${2-}" = "inspect" ]; then
+  exit 1
+fi
+exit 0
+"#,
+        )
+        .unwrap();
+        set_executable(&fake_docker).unwrap();
+
+        let proxy = KindDockerProxy::with_real_docker(fake_docker).unwrap();
+        let mut node = Command::new("docker");
+        proxy.apply(&mut node).unwrap();
+        let node_status = node
+            .env("HOPS_TEST_DOCKER_CALLS", &calls)
+            .args([
+                "run",
+                "--name",
+                "dogfood-control-plane",
+                "--label",
+                "io.x-k8s.kind.role=control-plane",
+                "--label",
+                "io.x-k8s.kind.cluster=dogfood",
+                "--volume",
+                "/var",
+                "kindest/node:v1.36.1",
+            ])
+            .status()
+            .unwrap();
+        assert!(node_status.success());
+
+        let mut unrelated = Command::new("docker");
+        proxy.apply(&mut unrelated).unwrap();
+        let unrelated_status = unrelated
+            .env("HOPS_TEST_DOCKER_CALLS", &calls)
+            .args(["ps", "--quiet"])
+            .status()
+            .unwrap();
+        assert!(unrelated_status.success());
+
+        let calls = fs::read_to_string(&calls).unwrap();
+        assert!(calls.contains(
+            "CALL\tvolume\tcreate\t--label\tdev.hops.local.managed=true\t--label\tdev.hops.local.kind.cluster=dogfood\t--label\tdev.hops.local.kind.node=dogfood-control-plane\thops-kind-dogfood-control-plane-data"
+        ));
+        assert!(calls.contains(
+            "CALL\trun\t--volume\thops-kind-dogfood-control-plane-data:/var\t--name\tdogfood-control-plane"
+        ));
+        assert!(calls.contains("CALL\tps\t--quiet"));
+
+        drop(proxy);
+        fs::remove_dir_all(root).unwrap();
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn kind_docker_proxy_rejects_non_hops_volume_name_collision() {
+        let root = test_dir("kind-docker-proxy-collision");
+        let fake_docker = root.join("real-docker");
+        let calls = root.join("calls");
+        fs::write(
+            &fake_docker,
+            r#"#!/bin/sh
+set -eu
+printf 'CALL' >> "$HOPS_TEST_DOCKER_CALLS"
+for argument in "$@"; do
+  printf '\t%s' "$argument" >> "$HOPS_TEST_DOCKER_CALLS"
+done
+printf '\n' >> "$HOPS_TEST_DOCKER_CALLS"
+if [ "${1-}" = "volume" ] && [ "${2-}" = "inspect" ]; then
+  if [ "${3-}" = "--format" ]; then
+    printf 'false\n'
+  fi
+  exit 0
+fi
+exit 0
+"#,
+        )
+        .unwrap();
+        set_executable(&fake_docker).unwrap();
+
+        let proxy = KindDockerProxy::with_real_docker(fake_docker).unwrap();
+        let mut node = Command::new("docker");
+        proxy.apply(&mut node).unwrap();
+        let output = node
+            .env("HOPS_TEST_DOCKER_CALLS", &calls)
+            .args([
+                "run",
+                "--name",
+                "dogfood-control-plane",
+                "--label",
+                "io.x-k8s.kind.role=control-plane",
+                "--label",
+                "io.x-k8s.kind.cluster=dogfood",
+                "kindest/node:v1.36.1",
+            ])
+            .output()
+            .unwrap();
+        assert!(!output.status.success());
+        assert!(String::from_utf8_lossy(&output.stderr)
+            .contains("refusing non-Hops volume name collision"));
+        assert!(!fs::read_to_string(&calls).unwrap().contains("CALL\trun"));
+
+        drop(proxy);
+        fs::remove_dir_all(root).unwrap();
+    }
+
+    #[cfg(unix)]
+    fn test_dir(prefix: &str) -> PathBuf {
+        let nonce = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap()
+            .as_nanos();
+        let path =
+            std::env::temp_dir().join(format!("hops-{prefix}-{}-{nonce}", std::process::id()));
+        fs::create_dir(&path).unwrap();
+        path
+    }
 }
diff --git a/src/commands/local/gitops.rs b/src/commands/local/gitops.rs
index c2c9494..83d0bda 100644
--- a/src/commands/local/gitops.rs
+++ b/src/commands/local/gitops.rs
@@ -1,7 +1,7 @@
 //! `hops local gitops` — control-plane and Environment reconcile.
 //!
 //! ```text
-//! hops local gitops cluster  [PATH]   # shared CP (.gitops/local/cluster)
+//! hops local gitops cluster [cluster.yaml] # lifecycle + shared CP manifests
 //! hops local gitops environment <PATH> # Environment apps → namespace = --name
 //! ```
 //!
@@ -9,19 +9,21 @@
 
 use super::local_state_dir;
 use super::workbench::application::{
-    load_applications, resolve_delivery_host_path, Application, APPLICATION_API_VERSION,
+    load_applications, resolve_delivery_host_path, Application, ApplicationMetadata,
+    ApplicationSpec, Destination, HelmSource, Source, SyncPolicy, APPLICATION_API_VERSION,
     APPLICATION_KIND,
 };
-use super::workbench::cluster_gitops::{
-    reconcile_cluster_dir, resolve_cluster_path, should_reconcile_cluster_change,
+use super::workbench::cluster_gitops::{reconcile_cluster_dir, should_reconcile_cluster_change};
+use super::workbench::definition::{
+    load_definition, load_environment_definition, prepare_cluster, ClusterOverrides,
+    DeployDefinition, DEFAULT_DEPLOY_CHART_PATH,
 };
-use super::workbench::definition::{load_definition, load_environment_definition};
 use super::workbench::delivery::{
     attach_sync_delivery, discover_sync_targets, save_delivery_runtime, stop_delivery_runtime,
     DeliveryStrategy, NodePathProber, SystemNodeProber,
 };
 use super::workbench::reconcile::{
-    reconcile_applications, HelmRunner, ReconcileOptions, SystemHelm, SystemKubectl,
+    reconcile_applications, ReconcileOptions, SystemHelm, SystemKubectl,
 };
 use super::workbench::registry::{
     activate_workspace_cluster, load_workspace, save_workspace, WorkspaceRecord,
@@ -58,11 +60,14 @@ pub enum GitopsCommands {
 
 #[derive(Args, Debug)]
 pub struct ClusterArgs {
-    /// Path to cluster gitops directory (PSQLStack, AuthStack, packages, …).
-    /// Default: `$HOPS_LOCAL_CLUSTER`, else walk up from cwd for `.gitops/local/cluster`.
+    /// Kubernetes-shaped Cluster definition. Defaults to .gitops/local/cluster.yaml.
     #[arg(value_name = "PATH")]
     pub path: Option<PathBuf>,
 
+    /// Stop the declared Cluster instead of starting and watching it.
+    #[arg(long, default_value_t = false)]
+    pub down: bool,
+
     /// Run a single reconcile and exit (disables the default watch).
     #[arg(long, default_value_t = false)]
     pub once: bool,
@@ -82,9 +87,14 @@ pub struct ClusterArgs {
 
 #[derive(Args, Debug)]
 pub struct EnvironmentArgs {
-    /// Reusable Environment YAML, or a legacy directory of Application YAMLs.
+    /// Reusable Environment YAML, or a legacy Application directory.
+    /// Optional with --down, which resolves the registered Environment by name.
     #[arg(value_name = "PATH")]
-    pub path: PathBuf,
+    pub path: Option<PathBuf>,
+
+    /// Purge and unregister this Environment instead of reconciling it.
+    #[arg(long, default_value_t = false)]
+    pub down: bool,
 
     /// Destination namespace override (workspace isolation).
     #[arg(long, short = 'n')]
@@ -111,35 +121,52 @@ pub struct EnvironmentArgs {
     pub dry_run: bool,
 }
 
-pub fn run(args: &GitopsArgs) -> Result<(), Box<dyn Error>> {
+pub fn run_environment_command(args: &GitopsArgs) -> Result<(), Box<dyn Error>> {
     match &args.command {
-        GitopsCommands::Cluster(a) => run_cluster(a),
         GitopsCommands::Environment(a) => run_environment(a),
+        GitopsCommands::Cluster(_) => Err(
+            "internal dispatch error: Cluster must be activated before generic local dispatch"
+                .into(),
+        ),
     }
 }
 
-/// Run cluster gitops (same as `hops local gitops cluster`).
-/// Used by `hops local start --gitops` so start is not a special code path.
-pub fn run_cluster(args: &ClusterArgs) -> Result<(), Box<dyn Error>> {
-    if !args.dry_run {
-        if let Err(e) = super::run_cmd_output("kubectl", &["cluster-info"]) {
-            return Err(format!(
-                "Local control plane is not reachable ({e}).\n\
-                 Ensure the selected control plane is Ready, then run `hops local start` with matching --cluster-provider and --docker-provider values."
-            )
-            .into());
+/// Start or resume the declared control plane, then reconcile its shared
+/// manifests. The Cluster definition is the single lifecycle entry point.
+pub fn run_cluster(
+    args: &ClusterArgs,
+    overrides: ClusterOverrides<'_>,
+) -> Result<(), Box<dyn Error>> {
+    let (definition, backend) = prepare_cluster(args.path.as_deref(), overrides)?;
+
+    if args.down {
+        if definition.cluster.cluster_provider == super::backend::ClusterProvider::Kind
+            && !super::backend::kind::cluster_exists()
+        {
+            log::info!("Cluster '{}' is already down", definition.cluster.name);
+            return Ok(());
         }
+        backend.stop()?;
+        return Ok(());
     }
 
-    let cluster = resolve_cluster_path(None, args.path.as_deref()).ok_or_else(|| {
-        "no cluster gitops directory found.\n\
-         Pass a path: hops local gitops cluster ./.gitops/local/cluster\n\
-         Or set HOPS_LOCAL_CLUSTER, or create .gitops/local/cluster at the project root."
-            .to_string()
-    })?;
-    let cluster = cluster
-        .canonicalize()
-        .map_err(|e| format!("cluster path {}: {e}", cluster.display()))?;
+    if args.dry_run {
+        log::info!(
+            "Dry-run uses the declared Cluster '{}' without changing its lifecycle",
+            definition.cluster.name
+        );
+    } else {
+        super::start::run(
+            backend,
+            &super::start::StartArgs {
+                size: super::backend::SizeArgs::default(),
+                yes: false,
+                bootstrap: false,
+            },
+        )?;
+    }
+
+    let cluster = definition.cluster.manifests_path;
 
     let dry_run = args.dry_run;
     let do_once = || -> Result<(), Box<dyn Error>> {
@@ -172,17 +199,27 @@ pub fn run_cluster(args: &ClusterArgs) -> Result<(), Box<dyn Error>> {
 // ── environment ──────────────────────────────────────────────────────────────
 
 fn run_environment(args: &EnvironmentArgs) -> Result<(), Box<dyn Error>> {
-    if args.path.is_file() && yaml_kind(&args.path)?.as_deref() == Some("Environment") {
-        return run_environment_definition(args);
+    if args.down {
+        return super::down::run(&super::down::DownArgs {
+            name: args.name.clone(),
+            purge: true,
+        });
     }
-    run_application_worktree(args)
-}
 
-fn run_environment_definition(args: &EnvironmentArgs) -> Result<(), Box<dyn Error>> {
-    let source = args
+    let path = args
         .path
+        .as_deref()
+        .ok_or("Environment PATH is required unless --down is used with a registered --name")?;
+    if path.is_file() && yaml_kind(path)?.as_deref() == Some("Environment") {
+        return run_environment_definition(args, path);
+    }
+    run_application_worktree(args, path)
+}
+
+fn run_environment_definition(args: &EnvironmentArgs, path: &Path) -> Result<(), Box<dyn Error>> {
+    let source = path
         .canonicalize()
-        .map_err(|error| format!("Environment path {}: {error}", args.path.display()))?;
+        .map_err(|error| format!("Environment path {}: {error}", path.display()))?;
     let cluster_path = discover_cluster_definition(&source).ok_or_else(|| {
         format!(
             "no sibling or ancestor Cluster definition found for {}",
@@ -201,8 +238,7 @@ fn run_environment_definition(args: &EnvironmentArgs) -> Result<(), Box<dyn Erro
     let worktree_root = loaded.environment.root.clone();
     let mut chart_watch_roots = BTreeSet::new();
     for deploy in &loaded.environment.deploys {
-        chart_watch_roots.insert(deploy.promote_chart.clone());
-        chart_watch_roots.insert(deploy.application_root.join(".gitops/local"));
+        chart_watch_roots.insert(deploy.chart_path.clone());
     }
     let chart_watch_roots: Vec<_> = chart_watch_roots.into_iter().collect();
     super::backend::kind::set_active_cluster_name(&cluster.cluster.name);
@@ -225,7 +261,8 @@ fn run_environment_definition(args: &EnvironmentArgs) -> Result<(), Box<dyn Erro
             &namespace,
         )?;
         let legacy = EnvironmentArgs {
-            path: generated.clone(),
+            path: Some(generated.clone()),
+            down: false,
             namespace: Some(namespace.clone()),
             name: Some(workspace_name.clone()),
             once: true,
@@ -233,7 +270,7 @@ fn run_environment_definition(args: &EnvironmentArgs) -> Result<(), Box<dyn Erro
             debounce: args.debounce,
             dry_run: args.dry_run,
         };
-        run_application_worktree(&legacy)?;
+        run_application_worktree(&legacy, &generated)?;
         if !args.dry_run {
             persist_environment_registration(
                 &workspace_name,
@@ -257,6 +294,7 @@ fn run_environment_definition(args: &EnvironmentArgs) -> Result<(), Box<dyn Erro
         &source,
         &worktree_root,
         &chart_watch_roots,
+        &workspace_name,
         args.debounce,
         reconcile,
     )
@@ -295,17 +333,15 @@ fn render_environment_applications(
         generated,
         workspace_name,
         namespace,
-        &SystemHelm,
     )
 }
 
-fn render_environment_applications_with<H: HelmRunner>(
+fn render_environment_applications_with(
     environment_file: &Path,
     cluster_file: &Path,
     generated: &Path,
     workspace_name: &str,
     namespace: &str,
-    helm: &H,
 ) -> Result<(), Box<dyn Error>> {
     let cluster = load_definition(cluster_file)?;
     let loaded = load_environment_definition(
@@ -328,7 +364,7 @@ fn render_environment_applications_with<H: HelmRunner>(
     }
 
     let mut rendered_apps = BTreeMap::<String, Application>::new();
-    for (index, deploy) in loaded.environment.deploys.iter().enumerate() {
+    for deploy in &loaded.environment.deploys {
         let mut values = loaded.environment.values.clone();
         merge_mapping(&mut values, &deploy.values);
         values.insert(Value::String("local".into()), Value::Bool(true));
@@ -343,42 +379,30 @@ fn render_environment_applications_with<H: HelmRunner>(
             Value::String("source".into()),
             string_mapping(&[("localPath", &deploy.application_root.to_string_lossy())]),
         );
-        let values_yaml = serde_yaml::to_string(&Value::Mapping(values))?;
-        let output = helm.template(
-            &format!("{}-promote-{index}", sanitize_name(workspace_name)),
-            &deploy.promote_chart,
-            &loaded.environment.namespace,
-            &values_yaml,
-        )?;
-        for document in serde_yaml::Deserializer::from_str(&output) {
-            let value = Value::deserialize(document)?;
-            if value.is_null() {
-                continue;
-            }
-            let kind = value.get("kind").and_then(Value::as_str).unwrap_or("");
-            if kind != APPLICATION_KIND {
-                return Err(format!(
-                    "promotion chart {} emitted unsupported local kind {kind:?}; direct KRM reconciliation belongs to the Cluster controller task",
-                    deploy.promote_chart.display()
-                )
-                .into());
-            }
-            let mut application: Application = serde_yaml::from_value(value)?;
-            if application.api_version != APPLICATION_API_VERSION {
-                return Err(format!(
-                    "promotion chart {} emitted Application apiVersion {:?}; expected {APPLICATION_API_VERSION}",
-                    deploy.promote_chart.display(),
-                    application.api_version
-                )
-                .into());
-            }
-            application.spec.source.delivery_path =
-                Some(loaded.environment.root.to_string_lossy().into_owned());
-            application.spec.destination.namespace = Some(loaded.environment.namespace.clone());
-            let name = application.metadata.name.clone();
-            if rendered_apps.insert(name.clone(), application).is_some() {
-                return Err(format!("duplicate promoted Application name {name:?}").into());
-            }
+        let name = local_application_name(deploy);
+        let application = Application {
+            api_version: APPLICATION_API_VERSION.to_string(),
+            kind: APPLICATION_KIND.to_string(),
+            metadata: ApplicationMetadata {
+                name: name.clone(),
+                labels: None,
+            },
+            spec: ApplicationSpec {
+                source: Source {
+                    path: deploy.chart_path.to_string_lossy().into_owned(),
+                    delivery_path: Some(cluster.cluster.mount_root.to_string_lossy().into_owned()),
+                    helm: HelmSource {
+                        values: Some(Value::Mapping(values)),
+                    },
+                },
+                destination: Destination {
+                    namespace: Some(loaded.environment.namespace.clone()),
+                },
+                sync_policy: SyncPolicy { prune: true },
+            },
+        };
+        if rendered_apps.insert(name.clone(), application).is_some() {
+            return Err(format!("duplicate local Application name {name:?}").into());
         }
     }
     if rendered_apps.is_empty() {
@@ -395,6 +419,31 @@ fn render_environment_applications_with<H: HelmRunner>(
     Ok(())
 }
 
+fn local_application_name(deploy: &DeployDefinition) -> String {
+    let application = deploy
+        .application_root
+        .file_name()
+        .and_then(|value| value.to_str())
+        .unwrap_or("application");
+    let default_chart = deploy.application_root.join(DEFAULT_DEPLOY_CHART_PATH);
+    let raw = if deploy.chart_path == default_chart {
+        application.to_string()
+    } else {
+        let chart = deploy
+            .chart_path
+            .file_name()
+            .and_then(|value| value.to_str())
+            .unwrap_or("chart");
+        format!("{application}-{chart}")
+    };
+    let name = sanitize_name(&raw);
+    if name.is_empty() {
+        "application".to_string()
+    } else {
+        name
+    }
+}
+
 fn merge_mapping(base: &mut Mapping, overlay: &Mapping) {
     for (key, value) in overlay {
         match (base.get_mut(key), value) {
@@ -455,6 +504,7 @@ fn run_environment_watch<F>(
     environment_file: &Path,
     worktree_root: &Path,
     chart_roots: &[PathBuf],
+    workspace_name: &str,
     debounce_secs: u64,
     mut rebuild: F,
 ) -> Result<(), Box<dyn Error>>
@@ -493,7 +543,7 @@ where
         }
     }
     log::info!(
-        "Watching Environment {} and {} referenced promotion/deploy chart roots under {} (debounce {}s). Ctrl+C to stop.",
+        "Watching Environment {} and {} referenced local chart roots under {} (debounce {}s). Ctrl+C to stop.",
         environment_file.display(),
         chart_roots.len(),
         worktree_root.display(),
@@ -503,6 +553,17 @@ where
         rx.recv()
             .map_err(|_| "Environment watcher channel closed")?;
         wait_for_quiet(&rx, debounce)?;
+        if !environment_file.exists() {
+            log::info!(
+                "Environment definition {} was removed; purging Environment `{}`",
+                environment_file.display(),
+                workspace_name
+            );
+            return super::down::run(&super::down::DownArgs {
+                name: Some(workspace_name.to_string()),
+                purge: true,
+            });
+        }
         match rebuild() {
             Ok(()) => log::info!("Environment reconcile succeeded."),
             Err(error) => log::error!("Environment reconcile failed: {error}"),
@@ -514,11 +575,10 @@ fn is_environment_watch_path(path: &Path, source: &Path, chart_roots: &[PathBuf]
     path == source || chart_roots.iter().any(|root| path.starts_with(root))
 }
 
-fn run_application_worktree(args: &EnvironmentArgs) -> Result<(), Box<dyn Error>> {
-    let env_path = args
-        .path
+fn run_application_worktree(args: &EnvironmentArgs, path: &Path) -> Result<(), Box<dyn Error>> {
+    let env_path = path
         .canonicalize()
-        .map_err(|e| format!("env path {}: {e}", args.path.display()))?;
+        .map_err(|e| format!("env path {}: {e}", path.display()))?;
 
     let workspace_name = args
         .name
@@ -855,52 +915,6 @@ fn wait_for_quiet(rx: &mpsc::Receiver<()>, debounce: Duration) -> Result<(), Box
 mod tests {
     use super::*;
     use std::fs;
-    use std::sync::Mutex;
-
-    struct PromotionHelm {
-        values: Mutex<Vec<Value>>,
-    }
-
-    impl PromotionHelm {
-        fn new() -> Self {
-            Self {
-                values: Mutex::new(Vec::new()),
-            }
-        }
-    }
-
-    impl HelmRunner for PromotionHelm {
-        fn template(
-            &self,
-            _release: &str,
-            chart_path: &Path,
-            namespace: &str,
-            values_yaml: &str,
-        ) -> Result<String, Box<dyn Error>> {
-            self.values
-                .lock()
-                .unwrap()
-                .push(serde_yaml::from_str(values_yaml)?);
-            let application_root = chart_path
-                .parent()
-                .and_then(Path::parent)
-                .ok_or("promotion chart has no application root")?;
-            Ok(format!(
-                r#"apiVersion: hops.local/v1alpha1
-kind: Application
-metadata:
-  name: gateway
-spec:
-  source:
-    path: {}/.gitops/local
-  destination:
-    namespace: ignored
-"#,
-                application_root.display()
-            )
-            .replace("namespace: ignored", &format!("namespace: {namespace}")))
-        }
-    }
 
     #[test]
     fn discovers_project_root_from_git_ancestor() {
@@ -942,15 +956,8 @@ spec:
         ));
         fs::create_dir_all(&root).unwrap();
         let root = root.canonicalize().unwrap();
-        let promote = root.join("apps/gateway/.gitops/promote");
         fs::create_dir_all(root.join(".gitops/local/cluster")).unwrap();
         fs::create_dir_all(root.join("apps/gateway/.gitops/local")).unwrap();
-        fs::create_dir_all(&promote).unwrap();
-        fs::write(
-            promote.join("Chart.yaml"),
-            "apiVersion: v2\nname: gateway-promote\nversion: 0.1.0\n",
-        )
-        .unwrap();
         fs::write(
             root.join(".gitops/local/cluster.yaml"),
             r#"apiVersion: hops.local/v1alpha1
@@ -992,19 +999,27 @@ spec:
         .unwrap();
 
         let generated = root.join("generated");
-        let helm = PromotionHelm::new();
         render_environment_applications_with(
             &root.join(".gitops/local/environment.yaml"),
             &root.join(".gitops/local/cluster.yaml"),
             &generated,
             "feature-auth",
             "feature-auth-ns",
-            &helm,
         )
         .unwrap();
 
-        let values = helm.values.lock().unwrap();
-        let values = values[0].as_mapping().unwrap();
+        let applications = load_applications(&generated).unwrap();
+        assert_eq!(applications.len(), 1);
+        let application = &applications[0].1;
+        let values = application
+            .spec
+            .source
+            .helm
+            .values
+            .as_ref()
+            .unwrap()
+            .as_mapping()
+            .unwrap();
         assert_eq!(values["local"], Value::Bool(true));
         assert_eq!(values["preview"], Value::Bool(false));
         assert_eq!(values["feature"]["enabled"], Value::Bool(true));
@@ -1018,9 +1033,6 @@ spec:
             Value::String("feature-auth-ns".into())
         );
 
-        let applications = load_applications(&generated).unwrap();
-        assert_eq!(applications.len(), 1);
-        let application = &applications[0].1;
         assert_eq!(application.metadata.name, "gateway");
         assert_eq!(
             application.spec.destination.namespace.as_deref(),
@@ -1031,6 +1043,13 @@ spec:
             application.spec.source.delivery_path.as_deref(),
             Some(expected_delivery_path.as_str())
         );
+        assert_eq!(
+            application.spec.source.path,
+            root.join("apps/gateway/.gitops/local")
+                .to_string_lossy()
+                .into_owned()
+        );
+        assert!(application.spec.sync_policy.prune);
 
         fs::remove_dir_all(root).unwrap();
     }
@@ -1038,16 +1057,8 @@ spec:
     #[test]
     fn environment_watch_filters_to_definition_and_referenced_charts() {
         let source = Path::new("/project/.gitops/local/environment.yaml");
-        let chart_roots = vec![
-            PathBuf::from("/project/apps/api/.gitops/promote"),
-            PathBuf::from("/project/apps/api/.gitops/local"),
-        ];
+        let chart_roots = vec![PathBuf::from("/project/apps/api/.gitops/local")];
         assert!(is_environment_watch_path(source, source, &chart_roots));
-        assert!(is_environment_watch_path(
-            Path::new("/project/apps/api/.gitops/promote/templates/application.yaml"),
-            source,
-            &chart_roots,
-        ));
         assert!(is_environment_watch_path(
             Path::new("/project/apps/api/.gitops/local/values.yaml"),
             source,
diff --git a/src/commands/local/mod.rs b/src/commands/local/mod.rs
index 0149870..2006799 100644
--- a/src/commands/local/mod.rs
+++ b/src/commands/local/mod.rs
@@ -9,13 +9,11 @@ mod gitops;
 pub mod gitops_write;
 mod install;
 mod listmonk;
-mod open;
 pub mod package_install;
 mod reset;
 mod resize;
 mod start;
 mod status;
-mod stop;
 mod uninstall;
 pub mod workbench;
 mod zitadel;
@@ -126,7 +124,7 @@ pub struct LocalArgs {
     /// Only used with cluster-provider dory.
     ///
     /// Named `--dory-name` (not `--name`) so it never collides with workspace
-    /// `--name` on `hops local down|status|open|gitops environment`.
+    /// `--name` on `hops local down|status|gitops environment`.
     #[arg(long = "dory-name", global = true, value_name = "NAME")]
     pub dory_name: Option<String>,
 
@@ -144,8 +142,6 @@ pub enum LocalCommands {
     Reset,
     /// Start local k8s and ensure Crossplane control plane (skips helm when already healthy)
     Start(start::StartArgs),
-    /// Start or reuse the Cluster declared by .gitops/local/cluster.yaml
-    Up(workbench::definition::UpArgs),
     /// Resize the local cluster VM without destroying cluster state (colima cluster provider only)
     Resize(resize::ResizeArgs),
     /// Check what `hops local start` set up and report drift
@@ -154,8 +150,6 @@ pub enum LocalCommands {
     Down(down::DownArgs),
     /// Show local workbench workspace status and app URLs
     Status(status::StatusArgs),
-    /// Open the workspace UI URL in a browser
-    Open(open::OpenArgs),
     /// Local gitops: `cluster` (shared CP) or `environment` (app namespaces)
     Gitops(gitops::GitopsArgs),
     /// Configure crossplane-contrib provider-family-aws and AWS ProviderConfig
@@ -168,8 +162,6 @@ pub enum LocalCommands {
     Zitadel(zitadel::ZitadelArgs),
     /// Configure hops-ops/provider-listmonk and Listmonk ProviderConfig
     Listmonk(listmonk::ListmonkArgs),
-    /// Stop the local cluster
-    Stop,
     /// Destroy the local cluster
     Destroy,
     /// Uninstall local cluster-provider tools
@@ -177,10 +169,13 @@ pub enum LocalCommands {
 }
 
 pub fn run(args: &LocalArgs) -> Result<(), Box<dyn Error>> {
-    if let LocalCommands::Up(up_args) = &args.command {
-        return workbench::definition::run_up(
-            up_args,
-            workbench::definition::UpOverrides {
+    if let LocalCommands::Gitops(gitops::GitopsArgs {
+        command: gitops::GitopsCommands::Cluster(cluster),
+    }) = &args.command
+    {
+        return gitops::run_cluster(
+            cluster,
+            workbench::definition::ClusterOverrides {
                 cluster_provider: args.cluster_provider,
                 docker_provider: args.docker_provider,
                 legacy_backend: args.backend,
@@ -231,19 +226,16 @@ pub fn run(args: &LocalArgs) -> Result<(), Box<dyn Error>> {
         LocalCommands::Install => install::run(backend),
         LocalCommands::Reset => reset::run(backend),
         LocalCommands::Start(start_args) => start::run(backend, start_args),
-        LocalCommands::Up(_) => unreachable!("up dispatch returns before generic activation"),
         LocalCommands::Resize(resize_args) => resize::run(backend, resize_args),
         LocalCommands::Doctor => doctor::run(),
         LocalCommands::Down(down_args) => down::run(down_args),
         LocalCommands::Status(status_args) => status::run(status_args),
-        LocalCommands::Open(open_args) => open::run(open_args),
-        LocalCommands::Gitops(gitops_args) => gitops::run(gitops_args),
+        LocalCommands::Gitops(gitops_args) => gitops::run_environment_command(gitops_args),
         LocalCommands::Aws(aws_args) => aws::run(aws_args),
         LocalCommands::Cloudflare(cloudflare_args) => cloudflare::run(cloudflare_args),
         LocalCommands::Github(github_args) => github::run(github_args),
         LocalCommands::Zitadel(zitadel_args) => zitadel::run(zitadel_args),
         LocalCommands::Listmonk(listmonk_args) => listmonk::run(listmonk_args),
-        LocalCommands::Stop => stop::run(backend),
         LocalCommands::Destroy => destroy::run(backend),
         LocalCommands::Uninstall(uninstall_args) => uninstall::run(backend, uninstall_args),
     }
@@ -592,4 +584,57 @@ mod tests {
             other => panic!("expected Gitops, got {other:?}"),
         }
     }
+
+    #[test]
+    fn gitops_lifecycle_flags_parse_without_interim_commands() {
+        use clap::Parser;
+
+        #[derive(Parser, Debug)]
+        #[command(name = "hops-local-test")]
+        struct Cli {
+            #[command(flatten)]
+            local: LocalArgs,
+        }
+
+        let environment = Cli::try_parse_from([
+            "hops-local-test",
+            "gitops",
+            "environment",
+            "--name",
+            "feature-auth",
+            "--down",
+        ])
+        .expect("parse Environment teardown without a definition path");
+        match environment.local.command {
+            LocalCommands::Gitops(gitops::GitopsArgs {
+                command: gitops::GitopsCommands::Environment(environment),
+            }) => {
+                assert!(environment.down);
+                assert!(environment.path.is_none());
+            }
+            other => panic!("expected GitOps Environment, got {other:?}"),
+        }
+
+        let cluster = Cli::try_parse_from([
+            "hops-local-test",
+            "gitops",
+            "cluster",
+            ".gitops/local/cluster.yaml",
+            "--down",
+        ])
+        .expect("parse Cluster teardown");
+        match cluster.local.command {
+            LocalCommands::Gitops(gitops::GitopsArgs {
+                command: gitops::GitopsCommands::Cluster(cluster),
+            }) => assert!(cluster.down),
+            other => panic!("expected GitOps Cluster, got {other:?}"),
+        }
+
+        for removed in ["up", "open", "stop"] {
+            assert!(
+                Cli::try_parse_from(["hops-local-test", removed]).is_err(),
+                "interim command {removed:?} must stay removed"
+            );
+        }
+    }
 }
diff --git a/src/commands/local/open.rs b/src/commands/local/open.rs
deleted file mode 100644
index 974618d..0000000
--- a/src/commands/local/open.rs
+++ /dev/null
@@ -1,99 +0,0 @@
-//! `hops local open` — open the primary UI URL in a browser when possible.
-
-use super::workbench::net::{discover_workspace_endpoints, plan_host_access};
-use super::workbench::registry::{activate_workspace_cluster, list_workspaces, load_workspace};
-use super::{command_exists, local_state_dir, run_cmd};
-use clap::Args;
-use std::error::Error;
-
-#[derive(Args, Debug)]
-pub struct OpenArgs {
-    /// Workspace name (default: only workspace if exactly one).
-    #[arg(long)]
-    pub name: Option<String>,
-
-    /// Service to open (default: first *ui* service, else first service).
-    #[arg(long)]
-    pub service: Option<String>,
-}
-
-pub fn run(args: &OpenArgs) -> Result<(), Box<dyn Error>> {
-    let state_dir = local_state_dir()?;
-    let ws = match &args.name {
-        Some(n) => load_workspace(&state_dir, n)?
-            .ok_or_else(|| format!("Workspace `{n}` is not registered."))?,
-        None => {
-            let all = list_workspaces(&state_dir)?;
-            match all.as_slice() {
-                [only] => only.clone(),
-                [] => return Err("No workspaces registered.".into()),
-                many => {
-                    return Err(format!(
-                        "Multiple workspaces ({}); pass --name.",
-                        many.iter()
-                            .map(|w| w.name.as_str())
-                            .collect::<Vec<_>>()
-                            .join(", ")
-                    )
-                    .into())
-                }
-            }
-        }
-    };
-
-    if let Some((cluster, ctx)) = activate_workspace_cluster(&ws) {
-        log::debug!("open: bound cluster `{cluster}` (context {ctx})");
-    }
-    let services = discover_workspace_endpoints(&ws.namespace).unwrap_or_default();
-    let plan = plan_host_access(&ws.namespace, &services);
-
-    let url = pick_url(&plan.urls, args.service.as_deref()).ok_or_else(|| {
-        "No service URL available. Is the workspace up? Try hops local status.".to_string()
-    })?;
-
-    println!("Opening {url}");
-    open_browser(&url)?;
-    Ok(())
-}
-
-fn pick_url(
-    urls: &std::collections::BTreeMap<String, String>,
-    service: Option<&str>,
-) -> Option<String> {
-    if let Some(svc) = service {
-        // Accept bare name, or ns/name key.
-        if let Some(u) = urls.get(svc) {
-            return Some(u.clone());
-        }
-        for (key, url) in urls {
-            if key == svc || key.ends_with(&format!("/{svc}")) || key.contains(svc) {
-                return Some(url.clone());
-            }
-        }
-        return None;
-    }
-    // Prefer UI-ish names in the workspace namespace first.
-    for (name, url) in urls {
-        if name.contains("ui") && !name.contains("login") {
-            return Some(url.clone());
-        }
-    }
-    urls.values().next().cloned()
-}
-
-fn open_browser(url: &str) -> Result<(), Box<dyn Error>> {
-    // macOS open, Linux xdg-open; fall back to printing.
-    if cfg!(target_os = "macos") {
-        match run_cmd("open", &[url]) {
-            Ok(()) => return Ok(()),
-            Err(e) => log::warn!("open failed: {e}"),
-        }
-    } else if command_exists("xdg-open") {
-        match run_cmd("xdg-open", &[url]) {
-            Ok(()) => return Ok(()),
-            Err(e) => log::warn!("xdg-open failed: {e}"),
-        }
-    }
-    println!("Open this URL in your browser: {url}");
-    Ok(())
-}
diff --git a/src/commands/local/start.rs b/src/commands/local/start.rs
index d4ba74a..41947d6 100644
--- a/src/commands/local/start.rs
+++ b/src/commands/local/start.rs
@@ -155,17 +155,7 @@ fn bootstrap_control_plane() -> Result<(), Box<dyn Error>> {
     // still timing out (helm validate fails). Retry helm with API re-probes.
     log::info!("Installing Crossplane...");
     {
-        let helm_args = [
-            "upgrade",
-            "--install",
-            "crossplane",
-            "crossplane-stable/crossplane",
-            "-n",
-            "crossplane-system",
-            "--create-namespace",
-            "--timeout",
-            "5m",
-        ];
+        let helm_args = crossplane_helm_args();
         let mut last_err: Option<Box<dyn Error>> = None;
         for attempt in 1..=6 {
             wait_for_kubernetes()?;
@@ -238,6 +228,38 @@ fn bootstrap_control_plane() -> Result<(), Box<dyn Error>> {
     Ok(())
 }
 
+/// The local control plane is a single-node developer appliance. Kubernetes
+/// resource limits only throttle its controllers against each other and do not
+/// provide meaningful tenant isolation, so local bootstrap removes the chart's
+/// upstream requests and limits. The Dory/Colima VM remains the capacity
+/// boundary.
+fn crossplane_helm_args() -> Vec<&'static str> {
+    let mut args = vec![
+        "upgrade",
+        "--install",
+        "crossplane",
+        "crossplane-stable/crossplane",
+        "-n",
+        "crossplane-system",
+        "--create-namespace",
+        "--timeout",
+        "5m",
+    ];
+    for value in [
+        "resourcesCrossplane.limits.cpu=null",
+        "resourcesCrossplane.limits.memory=null",
+        "resourcesCrossplane.requests.cpu=null",
+        "resourcesCrossplane.requests.memory=null",
+        "resourcesRBACManager.limits.cpu=null",
+        "resourcesRBACManager.limits.memory=null",
+        "resourcesRBACManager.requests.cpu=null",
+        "resourcesRBACManager.requests.memory=null",
+    ] {
+        args.extend(["--set", value]);
+    }
+    args
+}
+
 /// In-cluster package registry + backend node/engine wiring.
 fn ensure_registry_ready(backend: backend::Backend) -> Result<(), Box<dyn Error>> {
     // Crossplane package pulls run in the pod network → Service DNS + ClusterIP.
@@ -459,4 +481,21 @@ mod tests {
             .bootstrap
         );
     }
+
+    #[test]
+    fn local_crossplane_bootstrap_removes_resource_constraints() {
+        let args = crossplane_helm_args();
+        for value in [
+            "resourcesCrossplane.limits.cpu=null",
+            "resourcesCrossplane.limits.memory=null",
+            "resourcesCrossplane.requests.cpu=null",
+            "resourcesCrossplane.requests.memory=null",
+            "resourcesRBACManager.limits.cpu=null",
+            "resourcesRBACManager.limits.memory=null",
+            "resourcesRBACManager.requests.cpu=null",
+            "resourcesRBACManager.requests.memory=null",
+        ] {
+            assert!(args.contains(&value), "missing local Helm override {value}");
+        }
+    }
 }
diff --git a/src/commands/local/stop.rs b/src/commands/local/stop.rs
deleted file mode 100644
index 4fa2bf5..0000000
--- a/src/commands/local/stop.rs
+++ /dev/null
@@ -1,6 +0,0 @@
-use super::backend::Backend;
-use std::error::Error;
-
-pub fn run(backend: Backend) -> Result<(), Box<dyn Error>> {
-    backend.stop()
-}
diff --git a/src/commands/local/workbench/cluster_gitops.rs b/src/commands/local/workbench/cluster_gitops.rs
index 209fffe..4bd4e0d 100644
--- a/src/commands/local/workbench/cluster_gitops.rs
+++ b/src/commands/local/workbench/cluster_gitops.rs
@@ -29,107 +29,6 @@ pub struct ClusterReconcileResult {
     pub errors: Vec<String>,
 }
 
-/// Resolve cluster gitops directory.
-///
-/// Order:
-/// 1. Explicit `override_path` (`--cluster`)
-/// 2. Env var `HOPS_LOCAL_CLUSTER`
-/// 3. Walk up from `env_path` looking for `.gitops/local/cluster`
-/// 4. Walk up from cwd looking for `.gitops/local/cluster`
-///
-/// The former `.gitops/cluster`, `gitops/cluster`, and `cluster` layouts
-/// remain migration fallbacks after the committed `.gitops/local/cluster`
-/// convention.
-///
-/// Returns the first existing directory. Explicit override that does not exist
-/// is left to the caller to error on canonicalize.
-pub fn resolve_cluster_path(
-    env_path: Option<&Path>,
-    override_path: Option<&Path>,
-) -> Option<PathBuf> {
-    if let Some(p) = override_path {
-        return Some(p.to_path_buf());
-    }
-    if let Ok(p) = std::env::var("HOPS_LOCAL_CLUSTER") {
-        let pb = PathBuf::from(p.trim());
-        if !p.trim().is_empty() && pb.is_dir() {
-            return Some(pb);
-        }
-    }
-    if let Some(env) = env_path {
-        if let Some(found) = discover_cluster_path(env) {
-            return Some(found);
-        }
-    }
-    if let Ok(cwd) = std::env::current_dir() {
-        return walk_up_for_cluster(&cwd);
-    }
-    None
-}
-
-/// Discover a cluster tree near an env path (or walk to meta root).
-///
-/// ```text
-/// .gitops/local/environment.yaml → sibling .gitops/local/cluster
-/// some/deep/project              → walk up → <meta>/.gitops/local/cluster
-/// <meta>/.gitops/local           → <meta>/.gitops/local/cluster
-/// ```
-pub fn discover_cluster_path(env_path: &Path) -> Option<PathBuf> {
-    let env = env_path
-        .canonicalize()
-        .unwrap_or_else(|_| env_path.to_path_buf());
-
-    // Tight layouts first (same gitops/ as envs)
-    if let Some(parent) = env.parent() {
-        let name = parent.file_name().and_then(|s| s.to_str()).unwrap_or("");
-        if name == "envs" || name == "env" {
-            if let Some(gitops) = parent.parent() {
-                let cluster = gitops.join("cluster");
-                if cluster.is_dir() {
-                    return Some(cluster);
-                }
-            }
-        }
-    }
-    if env.file_name().and_then(|s| s.to_str()) == Some("gitops") {
-        let cluster = env.join("cluster");
-        if cluster.is_dir() {
-            return Some(cluster);
-        }
-    }
-    if let Some(parent) = env.parent() {
-        let cluster = parent.join("cluster");
-        if cluster.is_dir() {
-            return Some(cluster);
-        }
-    }
-
-    // Meta-root walk: prefer the committed .gitops/local/cluster convention,
-    // then retain the old paths as migration fallbacks.
-    walk_up_for_cluster(&env)
-}
-
-/// Walk from `start` toward filesystem root for `.gitops/local/cluster` and legacy layouts.
-fn walk_up_for_cluster(start: &Path) -> Option<PathBuf> {
-    let mut cur = start.canonicalize().unwrap_or_else(|_| start.to_path_buf());
-    loop {
-        for candidate in [
-            cur.join(".gitops").join("local").join("cluster"),
-            cur.join(".gitops").join("cluster"),
-            cur.join("gitops").join("cluster"),
-            cur.join("cluster"),
-        ] {
-            if candidate.is_dir() {
-                return Some(candidate);
-            }
-        }
-        if !cur.pop() {
-            break;
-        }
-    }
-    None
-}
-
 /// Collect YAML manifests under cluster_path (recursive).
 /// Skips examples, docs, and non-manifest files.
 ///
@@ -377,82 +276,6 @@ mod tests {
         let _ = fs::remove_dir_all(&dir);
     }
 
-    #[test]
-    fn discover_from_envs_local() {
-        let dir = std::env::temp_dir().join(format!(
-            "hops-cg-{}-{}",
-            std::process::id(),
-            std::time::SystemTime::now()
-                .duration_since(std::time::UNIX_EPOCH)
-                .unwrap()
-                .as_nanos()
-        ));
-        let envs = dir.join("gitops/envs/local");
-        let cluster = dir.join("gitops/cluster");
-        fs::create_dir_all(&envs).unwrap();
-        fs::create_dir_all(&cluster).unwrap();
-        let found = discover_cluster_path(&envs).unwrap();
-        assert_eq!(
-            found.canonicalize().unwrap(),
-            cluster.canonicalize().unwrap()
-        );
-        let _ = fs::remove_dir_all(&dir);
-    }
-
-    #[test]
-    fn discover_walks_up_to_meta_root() {
-        let dir = std::env::temp_dir().join(format!(
-            "hops-cg-meta-{}-{}",
-            std::process::id(),
-            std::time::SystemTime::now()
-                .duration_since(std::time::UNIX_EPOCH)
-                .unwrap()
-                .as_nanos()
-        ));
-        // Meta-root layout: cluster at meta root, env deep under a project
-        let cluster = dir.join("gitops/cluster");
-        let deep_env = dir.join("clients/foo/gitops/envs/local");
-        fs::create_dir_all(&cluster).unwrap();
-        fs::create_dir_all(&deep_env).unwrap();
-        let found = discover_cluster_path(&deep_env).unwrap();
-        assert_eq!(
-            found.canonicalize().unwrap(),
-            cluster.canonicalize().unwrap()
-        );
-        // explicit override wins
-        let other = dir.join("other-cluster");
-        fs::create_dir_all(&other).unwrap();
-        let resolved = resolve_cluster_path(Some(&deep_env), Some(&other)).unwrap();
-        assert_eq!(resolved, other);
-        let _ = fs::remove_dir_all(&dir);
-    }
-
-    #[test]
-    fn discover_prefers_dot_gitops_cluster() {
-        let dir = std::env::temp_dir().join(format!(
-            "hops-cg-dot-meta-{}-{}",
-            std::process::id(),
-            std::time::SystemTime::now()
-                .duration_since(std::time::UNIX_EPOCH)
-                .unwrap()
-                .as_nanos()
-        ));
-        let preferred = dir.join(".gitops/local/cluster");
-        let legacy = dir.join(".gitops/cluster");
-        let environment = dir.join(".gitops/local/environment.yaml");
-        fs::create_dir_all(&preferred).unwrap();
-        fs::create_dir_all(&legacy).unwrap();
-        fs::create_dir_all(environment.parent().unwrap()).unwrap();
-        fs::write(&environment, "kind: Environment\n").unwrap();
-
-        let found = discover_cluster_path(&environment).unwrap();
-        assert_eq!(
-            found.canonicalize().unwrap(),
-            preferred.canonicalize().unwrap()
-        );
-        let _ = fs::remove_dir_all(&dir);
-    }
-
     #[test]
     fn skips_examples_and_docs() {
         let dir = std::env::temp_dir().join(format!("hops-cg-skip-{}", std::process::id()));
diff --git a/src/commands/local/workbench/definition.rs b/src/commands/local/workbench/definition.rs
index db384bc..49ce0cf 100644
--- a/src/commands/local/workbench/definition.rs
+++ b/src/commands/local/workbench/definition.rs
@@ -1,12 +1,9 @@
 //! Kubernetes-shaped Cluster and independently reusable Environment loading.
 //!
-//! This module intentionally stops at a validated, immutable handoff. The
-//! long-running controller consumes [`LoadedDefinition`] in the next rollout
-//! task; `hops local up` currently owns definition validation and named local
-//! cluster create/reuse only.
+//! The GitOps cluster command consumes the validated definition to select the
+//! backend, mount the project root, and start or resume the named cluster.
 
-use crate::commands::local::backend::{self, Backend, ClusterProvider, DockerProvider, SizeArgs};
-use clap::Args;
+use crate::commands::local::backend::{self, Backend, ClusterProvider, DockerProvider};
 use serde::de::DeserializeOwned;
 use serde::Deserialize;
 use serde_yaml::{Mapping, Value};
@@ -23,17 +20,10 @@ pub const LEGACY_DEFINITION_FILE: &str = "cluster.yaml";
 pub const DEFAULT_ENVIRONMENT_FILE: &str = ".gitops/local/environment.yaml";
 pub const CLUSTER_MANIFESTS_PATH: &str = ".gitops/local/cluster";
 pub const LEGACY_CLUSTER_MANIFESTS_PATH: &str = ".gitops/cluster";
-pub const PROMOTE_CHART_PATH: &str = ".gitops/promote";
-
-#[derive(Args, Debug, Clone)]
-pub struct UpArgs {
-    /// Cluster definition. Defaults to ./.gitops/local/cluster.yaml.
-    #[arg(short = 'f', long = "file", value_name = "PATH")]
-    pub file: Option<PathBuf>,
-}
+pub const DEFAULT_DEPLOY_CHART_PATH: &str = ".gitops/local";
 
 #[derive(Debug, Clone, Copy, Default)]
-pub struct UpOverrides<'a> {
+pub struct ClusterOverrides<'a> {
     pub cluster_provider: Option<ClusterProvider>,
     pub docker_provider: Option<DockerProvider>,
     pub legacy_backend: Option<Backend>,
@@ -84,7 +74,7 @@ pub struct EnvironmentDefinition {
 #[derive(Debug, Clone, PartialEq)]
 pub struct DeployDefinition {
     pub application_root: PathBuf,
-    pub promote_chart: PathBuf,
+    pub chart_path: PathBuf,
     pub values: Mapping,
 }
 
@@ -165,12 +155,21 @@ struct ClusterReference {
 struct DeploySpec {
     path: PathBuf,
     #[serde(default)]
+    chart: Option<PathBuf>,
+    #[serde(default)]
     values: Mapping,
 }
 
-pub fn run_up(args: &UpArgs, overrides: UpOverrides<'_>) -> Result<(), Box<dyn Error>> {
+/// Validate and activate a Cluster definition without starting or stopping it.
+///
+/// Keeping lifecycle mutation in `gitops cluster` lets `--down` use the same
+/// definition and guarantees invalid definitions fail before touching Docker.
+pub fn prepare_cluster(
+    file: Option<&Path>,
+    overrides: ClusterOverrides<'_>,
+) -> Result<(LoadedDefinition, Backend), Box<dyn Error>> {
     let cwd = std::env::current_dir()?;
-    let source = definition_path(args.file.as_deref(), &cwd);
+    let source = definition_path(file, &cwd);
 
     // All parsing, identity, provider, and filesystem validation happens
     // before process state, local state, or the cluster can be mutated.
@@ -207,7 +206,6 @@ pub fn run_up(args: &UpArgs, overrides: UpOverrides<'_>) -> Result<(), Box<dyn E
         Some(&definition.cluster.name),
         explicit_context,
     )?;
-    active_backend.start(&SizeArgs::default(), false)?;
     backend::persist_providers(
         backend::providers::ProviderPair {
             cluster: definition.cluster.cluster_provider,
@@ -217,7 +215,7 @@ pub fn run_up(args: &UpArgs, overrides: UpOverrides<'_>) -> Result<(), Box<dyn E
     )?;
 
     log::info!(
-        "Cluster '{}' ready: context={} clusterProvider={} dockerProvider={} mountRoot={} definition={}",
+        "Cluster '{}' selected: context={} clusterProvider={} dockerProvider={} mountRoot={} definition={}",
         definition.cluster.name,
         expected_context(&definition, overrides),
         definition.cluster.cluster_provider,
@@ -225,15 +223,7 @@ pub fn run_up(args: &UpArgs, overrides: UpOverrides<'_>) -> Result<(), Box<dyn E
         definition.cluster.mount_root.display(),
         definition.source.display()
     );
-    log::info!(
-        "Validated shared manifests: {}",
-        definition.cluster.manifests_path.display()
-    );
-    log::info!(
-        "Cluster definition contains no Environment inventory; register each checkout with `hops local gitops environment .gitops/local/environment.yaml --name <environment>`"
-    );
-
-    Ok(())
+    Ok((definition, active_backend))
 }
 
 pub fn definition_path(file: Option<&Path>, cwd: &Path) -> PathBuf {
@@ -345,11 +335,11 @@ pub fn load_definition(path: &Path) -> Result<LoadedDefinition, Box<dyn Error>>
     )?;
     ensure_within(&mount_root, &definition_root, "Cluster definition")?;
     let manifests_relative = &raw_cluster.spec.manifests.path;
-    if manifests_relative != Path::new(CLUSTER_MANIFESTS_PATH)
+    if !manifests_relative.ends_with(CLUSTER_MANIFESTS_PATH)
         && manifests_relative != Path::new(LEGACY_CLUSTER_MANIFESTS_PATH)
     {
         return Err(format!(
-            "Cluster.spec.manifests.path must be {CLUSTER_MANIFESTS_PATH:?} (or legacy {LEGACY_CLUSTER_MANIFESTS_PATH:?}); got {:?}",
+            "Cluster.spec.manifests.path must end with {CLUSTER_MANIFESTS_PATH:?} (or equal legacy {LEGACY_CLUSTER_MANIFESTS_PATH:?}); got {:?}",
             raw_cluster.spec.manifests.path.display().to_string()
         )
         .into());
@@ -516,23 +506,28 @@ pub fn load_environment_definition(
             &format!("Environment {name:?} deploys[].path"),
             true,
         )?;
-        if !seen_deploys.insert(application_root.clone()) {
+        let chart_relative = deploy
+            .chart
+            .as_deref()
+            .unwrap_or_else(|| Path::new(DEFAULT_DEPLOY_CHART_PATH));
+        let chart_path = resolve_bounded_path(
+            &cluster.cluster.mount_root,
+            &application_root,
+            chart_relative,
+            &format!("Environment {name:?} deploys[].chart"),
+            true,
+        )?;
+        if !seen_deploys.insert((application_root.clone(), chart_path.clone())) {
             return Err(format!(
-                "Environment {name:?} contains duplicate deploy application root {}",
-                application_root.display()
+                "Environment {name:?} contains duplicate deploy for application root {} and chart {}",
+                application_root.display(),
+                chart_path.display()
             )
             .into());
         }
-        let promote_chart = resolve_bounded_path(
-            &cluster.cluster.mount_root,
-            &application_root,
-            Path::new(PROMOTE_CHART_PATH),
-            &format!("Environment {name:?} deploy promote chart"),
-            false,
-        )?;
         deploys.push(DeployDefinition {
             application_root,
-            promote_chart,
+            chart_path,
             values: deploy.values,
         });
     }
@@ -566,7 +561,7 @@ fn parse_document<T: DeserializeOwned>(
 
 fn validate_overrides(
     definition: &LoadedDefinition,
-    overrides: UpOverrides<'_>,
+    overrides: ClusterOverrides<'_>,
 ) -> Result<(), Box<dyn Error>> {
     if overrides.legacy_backend.is_some()
         && (overrides.cluster_provider.is_some() || overrides.docker_provider.is_some())
@@ -652,7 +647,7 @@ fn validate_overrides(
     Ok(())
 }
 
-fn expected_context(definition: &LoadedDefinition, overrides: UpOverrides<'_>) -> String {
+fn expected_context(definition: &LoadedDefinition, overrides: ClusterOverrides<'_>) -> String {
     match definition.cluster.cluster_provider {
         ClusterProvider::Kind => format!("kind-{}", definition.cluster.name),
         ClusterProvider::Colima => "colima".to_string(),
@@ -812,8 +807,9 @@ mod tests {
                 uuid::Uuid::new_v4()
             ));
             fs::create_dir_all(root.join(CLUSTER_MANIFESTS_PATH)).unwrap();
-            fs::create_dir_all(root.join("apps/gateway")).unwrap();
-            fs::create_dir_all(root.join("services/api")).unwrap();
+            fs::create_dir_all(root.join("apps/gateway/.gitops/local")).unwrap();
+            fs::create_dir_all(root.join("apps/gateway/.gitops/test-users")).unwrap();
+            fs::create_dir_all(root.join("services/api/.gitops/local")).unwrap();
             let root = root.canonicalize().unwrap();
             Self { root }
         }
@@ -889,8 +885,8 @@ spec:
         assert_eq!(environment.environment.namespace, "feature-auth");
         assert_eq!(environment.environment.root, fixture.root);
         assert_eq!(
-            environment.environment.deploys[0].promote_chart,
-            fixture.root.join("apps/gateway/.gitops/promote")
+            environment.environment.deploys[0].chart_path,
+            fixture.root.join("apps/gateway/.gitops/local")
         );
     }
 
@@ -1002,6 +998,25 @@ spec:
         .contains("duplicate deploy"));
     }
 
+    #[test]
+    fn allows_distinct_charts_for_the_same_application_root() {
+        let fixture = Fixture::new();
+        let loaded = load_definition(&fixture.write(valid_yaml())).unwrap();
+        let multiple = valid_environment_yaml().replace(
+            "    - path: services/api",
+            "    - path: apps/gateway\n      chart: .gitops/test-users",
+        );
+        let environment =
+            load_environment_definition(&fixture.write_environment(&multiple), &loaded, None, None)
+                .unwrap();
+
+        assert_eq!(environment.environment.deploys.len(), 2);
+        assert_eq!(
+            environment.environment.deploys[1].chart_path,
+            fixture.root.join("apps/gateway/.gitops/test-users")
+        );
+    }
+
     #[test]
     fn rejects_non_mapping_values_and_invalid_names() {
         let fixture = Fixture::new();
@@ -1026,19 +1041,27 @@ spec:
     }
 
     #[test]
-    fn requires_explicit_hidden_cluster_manifest_path() {
+    fn requires_cluster_manifest_path_to_end_in_the_local_convention() {
         let fixture = Fixture::new();
-        for path in [
-            "gitops/cluster",
-            ".gitops/deploy",
-            "./.gitops/local/cluster",
-        ] {
+        for path in ["gitops/cluster", ".gitops/deploy"] {
             let yaml = valid_yaml().replacen(CLUSTER_MANIFESTS_PATH, path, 1);
             let error = load_definition(&fixture.write(&yaml)).unwrap_err();
-            assert!(error.to_string().contains("must be"), "{error}");
+            assert!(error.to_string().contains("must end with"), "{error}");
         }
     }
 
+    #[test]
+    fn accepts_nested_project_cluster_manifest_path() {
+        let fixture = Fixture::new();
+        let nested = "tests/e2e-ui/.gitops/local/cluster";
+        fs::create_dir_all(fixture.root.join(nested)).unwrap();
+        let yaml = valid_yaml().replacen(CLUSTER_MANIFESTS_PATH, nested, 1);
+
+        let loaded = load_definition(&fixture.write(&yaml)).unwrap();
+
+        assert_eq!(loaded.cluster.manifests_path, fixture.root.join(nested));
+    }
+
     #[test]
     fn accepts_legacy_root_definition_and_manifest_layout() {
         let fixture = Fixture::new();
@@ -1117,21 +1140,21 @@ spec:
         let loaded = load_definition(&fixture.write(valid_yaml())).unwrap();
         validate_overrides(
             &loaded,
-            UpOverrides {
+            ClusterOverrides {
                 cluster_provider: Some(ClusterProvider::Kind),
                 docker_provider: Some(DockerProvider::Dory),
                 cluster_name: Some("project-dev"),
                 context: Some("kind-project-dev"),
-                ..UpOverrides::default()
+                ..ClusterOverrides::default()
             },
         )
         .unwrap();
 
         let error = validate_overrides(
             &loaded,
-            UpOverrides {
+            ClusterOverrides {
                 docker_provider: Some(DockerProvider::Colima),
-                ..UpOverrides::default()
+                ..ClusterOverrides::default()
             },
         )
         .unwrap_err();
diff --git a/tests/local_cluster_definition.rs b/tests/local_cluster_definition.rs
index 192f52d..1ae7967 100644
--- a/tests/local_cluster_definition.rs
+++ b/tests/local_cluster_definition.rs
@@ -57,6 +57,8 @@ case "$tool" in
   docker)
     if test "$1" = "info"; then echo "27.0.0"; exit 0; fi
     if test "$1" = "ps"; then exit 0; fi
+    if test "$1" = "pull"; then exit 0; fi
+    if test "$1" = "volume"; then exit 0; fi
     if test "$1" = "inspect"; then
       case "$*" in
         *'{{json .Mounts}}'*)
@@ -78,13 +80,20 @@ case "$tool" in
           ;;
       esac
     fi
-    if test "$1" = "exec" || test "$1" = "start"; then exit 0; fi
+    if test "$1" = "exec"; then cat >/dev/null; exit 0; fi
+    if test "$1" = "start" || test "$1" = "stop"; then exit 0; fi
     exit 0
     ;;
   kubectl)
     if test "$1" = "config" && test "$2" = "get-contexts"; then
       echo kind-project-dev
+      exit 0
     fi
+    case "$*" in
+      *availableReplicas*) echo 1 ;;
+      *status.conditions*) echo True ;;
+      *'get svc registry '*'spec.clusterIP'*) echo 10.96.0.50 ;;
+    esac
     exit 0
     ;;
 esac
@@ -135,7 +144,7 @@ impl Fixture {
         let mut command = Command::new(env!("CARGO_BIN_EXE_hops-cli"));
         command
             .current_dir(&self.root)
-            .args(["local", "up"])
+            .args(["local", "gitops", "cluster", DEFINITION_PATH, "--once"])
             .env("PATH", path)
             .env("HOME", self.root.join("home"))
             .env("DOCKER_HOST", "unix:///contract-test.sock")
@@ -206,7 +215,7 @@ fn parses_cluster_only() {
     assert!(first_log.contains("kind create cluster --name project-dev --config -"));
     assert!(first_log.contains(&format!("hostPath: \"{}\"", fixture.root.display())));
     let text = output_text(&first);
-    assert!(text.contains("contains no Environment inventory"), "{text}");
+    assert!(text.contains("Cluster 'project-dev' selected"), "{text}");
     let provider_state = fs::read_to_string(fixture.root.join("home/.hops/local/providers.json"))
         .expect("successful up persists provider identity");
     assert!(provider_state.contains(r#""clusterProvider": "kind""#));
@@ -338,3 +347,20 @@ fn mount_drift_is_non_destructive() {
     assert!(!log.contains("kind delete cluster"));
     assert!(!log.contains("docker start"));
 }
+
+#[test]
+fn cluster_down_stops_the_declared_node_without_destroying_it() {
+    let fixture = Fixture::new();
+    fs::write(&fixture.cluster_exists, "existing").unwrap();
+
+    let output = fixture.command().arg("--down").output().unwrap();
+
+    assert!(output.status.success(), "{}", output_text(&output));
+    let log = fixture.log();
+    assert!(
+        log.contains("docker stop project-dev-control-plane"),
+        "{log}"
+    );
+    assert!(!log.contains("kind delete cluster"), "{log}");
+    assert!(!log.contains("docker volume rm"), "{log}");
+}